feat: Add metadata-only replace API to Table for REPLACE snapshot operations#3131
feat: Add metadata-only replace API to Table for REPLACE snapshot operations#3131qzyu999 wants to merge 18 commits intoapache:mainfrom
Conversation
- Fixed positional argument type mismatch for `snapshot_properties` in [_RewriteFiles](iceberg-python/pyiceberg/table/update/snapshot.py) - Added missing `Catalog` type annotations to pytest fixtures in [test_replace.py](iceberg-python/tests/table/test_replace.py) - Added strict `is not None` assertions for `table.current_snapshot()` to satisfy mypy Optional checking - Auto-formatted tests with ruff
…ass enum validation (Operation.REPLACE is valid so we can no longer use it in test_invalid_operation)
kevinjqliu
left a comment
kevinjqliu
left a comment
There was a problem hiding this comment.
Thanks for the PR.
There are a couple of things we should check (and add as test cases):
Scanning Manifests
- Every file marked for replacement was found in at least one manifest — if any are missing, abort
- Matched entries are rewritten as DELETED with the new snapshot ID
- Unmatched entries are carried over as EXISTING
- Manifests with no affected files are reused unchanged
New Manifest
- All incoming replacement files are present with status ADDED
- The new snapshot ID is applied to every entry
Manifest List
- Includes the new ADDED manifest
- Includes all rewritten manifests with DELETED entries
- Includes all unchanged manifests
- Includes any pre-existing delete manifests, passed through as-is
Invariant Check
- Records added ≤ records removed
- If the difference is due to prior soft-deletes, confirm those delete files account for it
- Records added never exceed records removed — if they do, the operation is invalid
Snapshot
- Has a unique snapshot ID
- Parent points to the previous snapshot
- Sequence number is exactly previous + 1
- Operation type is set to "replace"
- Manifest list path is correct
- Summary counts (files and records) are accurate
| """ | ||
| return UpdateStatistics(transaction=self) | ||
|
|
||
| def replace( |
There was a problem hiding this comment.
we should not expose replace as a public function as we cannot guarantee that the files_to_delete and files_to_add contains the same records.
I think we should start at _RewriteFiles
There was a problem hiding this comment.
Hi @kevinjqliu, I think that logic makes sense, as it would be dangerous for users to use these without being able to enforce the underlying expectations of the values input to these functions. Removed them in 94bd87e
| assert len(manifest_files) == 2 # One for ADDED, one for DELETED | ||
|
|
||
| # Check that sequence numbers were handled properly natively by verifying the manifest contents | ||
| entries = [] | ||
| for manifest in manifest_files: | ||
| for entry in manifest.fetch_manifest_entry(table.io, discard_deleted=False): | ||
| entries.append(entry) | ||
|
|
||
| # One entry for ADDED (new file), one for DELETED (old file) | ||
| assert len(entries) == 2 |
There was a problem hiding this comment.
we need to test a bit more on the status of each file.
https://github.com/apache/iceberg/blob/main/core/src/test/java/org/apache/iceberg/TestRewriteFiles.java is a good reference
There was a problem hiding this comment.
Hi @kevinjqliu, I've addressed this in the latest 33aaef0, where the status of each file is tested.
…k them with the new snapshot id
…ase specifications aren't met
…for how fast_append() still labels data as ManifestContent.DATA rather than ManifestContent.DELETES
|
Hi @kevinjqliu, apologies for the delay, thank you so much for taking the time to review the PR again, I understand that you are quite busy. I've addressed all your points in the latest set of tests within 33aaef0. I've thoroughly expanded the tests to integrate those requirements across a broad set of tests. There are two minor issues I noticed however:
|
| return [] | ||
|
|
||
| def _existing_manifests(self) -> list[ManifestFile]: | ||
| """To determine if there are any existing manifests.""" |
There was a problem hiding this comment.
This logic looks nearly identical to the OverwriteFiles._existing_manifests(), and probably can use some clean python oop with the different snapshot classes. So let's create a helper in the _SnapshotProducer class. Then Overwrite can add it's additional logic on top and rewrite can just use it.
|
|
||
| def _commit(self) -> UpdatesAndRequirements: | ||
| # Only produce a commit when there is something to rewrite | ||
| if self._deleted_data_files or self._added_data_files: |
There was a problem hiding this comment.
I think we can replicate the _DeleteFiles logic here by using the @cache_property on _the compute deletes function. Especially since _commit() calls self._deleted_entries() for validation and then calls the super commit to write and get delete entries.
| ) | ||
|
|
||
| def replace(self) -> _RewriteFiles: | ||
| return _RewriteFiles( |
There was a problem hiding this comment.
I'm sort of confused by the naming since we are introducing a user facing API replace but the underlying snapshot operation is a rewrite? We should rename to rewrite() for consistency? Unless I'm missing something?
There was a problem hiding this comment.
Hi @geruh, you bring up a good point, and it's something I noticed seemed off along the way. The reason why we have this discrepancy is because we're mirroring what's found in the Java code itself.
- Looking here (https://iceberg.apache.org/javadoc/latest/org/apache/iceberg/RewriteFiles.html), you can see that the "API for replacing files in a table." is called
RewriteFiles. - Looking here (https://iceberg.apache.org/javadoc/1.4.2/org/apache/iceberg/DataOperations.html#REPLACE), you can see that the
REPLACEoperation is implemented byRewriteFiles. - On a slightly unrelated note here (https://iceberg.apache.org/javadoc/1.4.2/org/apache/iceberg/DataOperations.html#OVERWRITE), the
OVERWRITEoperation itself can be implemented byReplacePartitions.
I named the Python API replace() to accurately reflect the Operation.REPLACE snapshot string it generates, while keeping the internal class named _RewriteFiles to match the Java builder logic.
That said, if you feel strongly about matching the Java API's user-facing method (rewrite()) rather than the snapshot operation (replace()), I'm happy to rename the public method to rewrite() for consistency. Let me know what you prefer!
| manifests_after = snapshot_after.manifests(table.io) | ||
| manifest_paths_after = [m.manifest_path for m in manifests_after] | ||
|
|
||
| assert delete_manifest_path in manifest_paths_after |
There was a problem hiding this comment.
We should also test to ensure that original sequence numbers are carried over from before the rewrite.
| from pyiceberg.typedef import Record | ||
|
|
||
|
|
||
| def test_replace_internally(catalog: Catalog) -> None: |
There was a problem hiding this comment.
Tests are on the right track can we add a few more like:
- test with multiple files for set matching rows
- test against partitioned table
- did we do a no-op replace to ensure we have correct state matching java impl
| added_records = sum(f.record_count for f in self._added_data_files) | ||
| deleted_records = sum(entry.data_file.record_count for entry in deleted_entries) | ||
|
|
||
| if added_records > deleted_records: |
There was a problem hiding this comment.
Where are you seeing this invariant? I mean this seems correct since the spec says rewrite must be "logically equivalent". This check could reasonable as a safety guard, but what happens when delete file rewriting is added? Then these numbers could be incorrect.
There was a problem hiding this comment.
Hi @geruh, thanks for flagging this, you're right that this is a safety guard, but it doesn't yet factor in future changes when adding delete file rewriting. Should we add something like this?
# Note: This physical record count invariant is a sanity guard for data file
# compaction to ensure no data is accidentally duplicated or invented.
# TODO: This will need to be evolved into a logical record count validation
# once PyIceberg supports rewriting delete files (Merge-on-Read).
added_records = sum(f.record_count for f in self._added_data_files)
deleted_records = sum(entry.data_file.record_count for entry in deleted_entries)
if added_records > deleted_records:
raise ValueError(f"Invalid replace: records added ({added_records}) exceeds records removed ({deleted_records})")This logical record count validation would involve something like having the _commit method to do the following, which the codebase currently cannot do:
- Identify associated Delete Files: For every
DataFileyou are deleting, you would need to find every Position Delete or Equality Delete file that points to it. - Calculate the "Subtraction": You would need to subtract those delete row counts from the physical record_count of the old files to find the Old Logical Count.
- Compare: You would then verify that Old Logical Count == New Logical Count.
The current_RewriteFilesimplementation is "blind" to deletes. It only tracks_added_data_filesand_deleted_data_files.
I believe this can be part of a full MoR implementation, something that I would love to work on after finishing these maintenance tasks.
Otherwise, I can also remove it from _RerwriteFiles and move forward, WDYT?
…erwriteFiles and _RewriteFiles
…ng with test_replace_multiple_files, test_replace_partitioned_table, and test_replace_no_op_on_non_empty_table
|
Hi @geruh, thanks for the awesome feedback, I've responded to each of your replies, PTAL. |
3 participants
Closes #3130
Rationale for this change
In a current PR (#3124, part of #1092), the proposed
replace()API accepts a PyArrow dataframe (pa.Table), forcing the table engine to physically serialize data during a metadata transaction commit. This couples execution with the catalog, diverges from Java Iceberg's nativeRewriteFilesbuilder behavior, and fails to register underOperation.REPLACE.This PR redesigns
table.replace()andtransaction.replace()to acceptIterable[DataFile]inputs. By externalizing physical data writing (e.g., compaction via Ray), the new explicit metadata-only_RewriteFilesSnapshotProducer can natively swap snapshot pointers in the manifests, perfectly inheriting ancestral sequence numbers forDELETEDentries to ensure time-travel equivalence.Are these changes tested?
Yes.
Fully exhaustive test coverage has been added to
tests/table/test_replace.py. The suite validates:len(table.history())).Operation.REPLACEtags.manifest.fetch_manifest_entry(discard_deleted=False)to natively assert thatstatus=DELETEDoverrides are accurately preserving avro sequence numbers.replace([], [])successfully short-circuits the commit loop without mutating history.Are there any user-facing changes?
Yes.
The method signature for
Table.replace()andTransaction.replace()has been updated from the original PR #3124.It no longer accepts a PyArrow DataFrame (
df: pa.Table). Instead, it now requests two arguments:files_to_delete: Iterable[DataFile]andfiles_to_add: Iterable[DataFile], following the convention seen in the Java implementation.(Please add the
changeloglabel)AI Disclosure
AI was used to help understand the code base and draft code changes. All code changes have been thoroughly reviewed, ensuring that the code changes are in line with a broader understanding of the codebase.
test_invalid_operation()intests/table/test_snapshots.pypreviously usedOperation.REPLACEas a value to test invalid operations, but with this changeOperation.REPLACEbecomes valid. In place I just put a dummy Operation._RewriteFilesinpyiceberg/table/update/snapshot.pyoverrides the_deleted_entriesand_existing_manifestsfunctions. I sought to test this thoroughly that it was done correctly. I am thinking it's possible to improve the test suite to make this more rigorous. I am open to suggestions on how that could be done.