-
Notifications
You must be signed in to change notification settings - Fork 581
Fix DataFile spec ID and reject unknown record fields #3954
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Changes from all commits
7d6ae63
c5c2886
b57b2e7
1b3a628
7c5d9ff
e99c006
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change | ||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
|
|
@@ -471,9 +471,14 @@ def data_file_with_partition(partition_type: StructType, format_version: TableVe | |||||||||||||||||||||
|
|
||||||||||||||||||||||
| class DataFile(Record): | ||||||||||||||||||||||
| @classmethod | ||||||||||||||||||||||
| def from_args(cls, _table_format_version: TableVersion = DEFAULT_READ_VERSION, **arguments: Any) -> DataFile: | ||||||||||||||||||||||
| def from_args( | ||||||||||||||||||||||
| cls, _table_format_version: TableVersion = DEFAULT_READ_VERSION, *, spec_id: int | None = None, **arguments: Any | ||||||||||||||||||||||
| ) -> DataFile: | ||||||||||||||||||||||
| struct = DATA_FILE_TYPE[_table_format_version] | ||||||||||||||||||||||
| return super()._bind(struct, **arguments) | ||||||||||||||||||||||
| data_file = super()._bind(struct, **arguments) | ||||||||||||||||||||||
|
Comment on lines
+475
to
+478
|
||||||||||||||||||||||
| if spec_id is not None: | ||||||||||||||||||||||
| data_file.spec_id = spec_id | ||||||||||||||||||||||
| return data_file | ||||||||||||||||||||||
|
Comment on lines
+479
to
+481
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this is using DataFile's own setter iceberg-python/pyiceberg/manifest.py Lines 551 to 560 in 562d3af
|
||||||||||||||||||||||
|
|
||||||||||||||||||||||
| @property | ||||||||||||||||||||||
| def content(self) -> DataFileContent: | ||||||||||||||||||||||
|
|
||||||||||||||||||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -179,6 +179,9 @@ class Record(StructProtocol): | |
|
|
||
| @classmethod | ||
| def _bind(cls, struct: StructType, **arguments: Any) -> Self: | ||
| field_names = {field.name for field in struct.fields} | ||
|
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. including this fix as part of the PR. This is the footgun that was silently dropping fields. We now check for unknown fields. This caught |
||
| if unknown_fields := arguments.keys() - field_names: | ||
| raise TypeError(f"Unexpected {cls.__name__} fields: {', '.join(sorted(unknown_fields))}") | ||
|
Comment on lines
+182
to
+184
|
||
| return cls(*[arguments[field.name] if field.name in arguments else field.initial_default for field in struct.fields]) | ||
|
|
||
| def __init__(self, *data: Any) -> None: | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -266,8 +266,7 @@ def test_write_v2_referenced_data_file_with_fastavro() -> None: | |
|
|
||
| @pytest.mark.parametrize("format_version", [1, 2]) | ||
| def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: TableVersion) -> None: | ||
| data_file_dict = { | ||
| "content": DataFileContent.DATA, | ||
| common_data_file_args = { | ||
| "file_path": "s3://some-path/some-file.parquet", | ||
| "file_format": FileFormat.PARQUET, | ||
| "partition": Record(), | ||
|
|
@@ -281,16 +280,16 @@ def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: Ta | |
| "upper_bounds": {1: b"zzzzzzzzzzzzzzzz"}, | ||
| "key_metadata": b"\xde\xad\xbe\xef", | ||
| "split_offsets": [4, 133697593], | ||
| "equality_ids": [], | ||
|
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| "sort_order_id": 4, | ||
| "spec_id": 3, | ||
| } | ||
| data_file_v2 = DataFile.from_args(**data_file_dict) # type: ignore | ||
| data_file = DataFile.from_args(content=DataFileContent.DATA, **common_data_file_args) # type: ignore | ||
|
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| assert data_file.spec_id == 3 | ||
|
|
||
| entry = ManifestEntry.from_args( | ||
| status=ManifestEntryStatus.ADDED, | ||
| snapshot_id=8638475580105682862, | ||
| data_file=data_file_v2, | ||
| data_file=data_file, | ||
| ) | ||
|
|
||
| with TemporaryDirectory() as tmpdir: | ||
|
|
@@ -322,7 +321,11 @@ def test_write_manifest_entry_with_fastavro_read_with_iceberg(format_version: Ta | |
| avro_entry = next(it) | ||
|
|
||
| if format_version == 1: | ||
| data_file_v1 = DataFile.from_args(**data_file_dict, _table_format_version=format_version) | ||
| data_file_v1 = DataFile.from_args( | ||
| _table_format_version=format_version, | ||
| block_size_in_bytes=DEFAULT_BLOCK_SIZE, | ||
|
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| **common_data_file_args, # type: ignore | ||
| ) | ||
|
|
||
| assert avro_entry == ManifestEntry.from_args( | ||
| status=1, | ||
|
|
||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -90,7 +90,6 @@ def test_write_sample_manifest(table_test_all_types: Table, compression: AvroCom | |
| test_schema = table_test_all_types.schema() | ||
| test_spec = table_test_all_types.spec() | ||
| wrapped_data_file_v2_debug = DataFile.from_args( | ||
| format_version=2, | ||
|
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. this is not a valid field, the typedef.py change caught it 😄 |
||
| content=entry.data_file.content, | ||
| file_path=entry.data_file.file_path, | ||
| file_format=entry.data_file.file_format, | ||
|
|
@@ -112,7 +111,7 @@ def test_write_sample_manifest(table_test_all_types: Table, compression: AvroCom | |
| wrapped_entry_v2 = copy(entry) | ||
| wrapped_entry_v2.data_file = wrapped_data_file_v2_debug | ||
| wrapped_entry_v2_dict = todict(wrapped_entry_v2, [field.name for field in test_spec.fields]) | ||
| for field in ("first_row_id", "content_offset", "content_size_in_bytes"): | ||
| for field in ("first_row_id", "content_offset", "content_size_in_bytes", "spec_id"): | ||
|
Contributor
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more.
|
||
| del wrapped_entry_v2_dict["data_file"][field] | ||
|
|
||
| with TemporaryDirectory() as tmpdir: | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
explicitly add
spec_idto the function.its called in
iceberg-python/pyiceberg/io/pyarrow.py
Line 2771 in 308768d
iceberg-python/pyiceberg/io/pyarrow.py
Line 2909 in 308768d
I think this is better. Otherwise caller has to set it after constructing DataFile. For example