Conversation
u70b3
changed the title
e6803c1 to
d177d34
Compare
u70b3
changed the title
88eb905 to
4f03d35
Compare
dcd3474 to
db7680c
Compare
d2ff512 to
42080dd
Compare
d627d46 to
ef4f848
Compare
|
Hi @mbutrovich, could you please take a look at this core COW rewrite primitive when you have time? It reuses the scan planning/read path (including delete-file application), clears the row predicate only after candidate selection so each chosen file is rewritten in full, and returns data-file sets for a later overwrite-style commit. I have rebased it onto current |
ef4f848 to
53cc129
Compare
|
Hi @CTTY @blackmwk, could you help review this core COW rewrite primitive when you have time? It's been waiting on review for a bit, so I'd like to widen the reviewer pool — any feedback is welcome. Quick context to make the review easier:
Thanks! |
eb2ecda to
70e01a9
Compare
511e840 to
79fef94
Compare
65dbb54 to
e1615dd
Compare
e0f940d to
a9f0b9d
Compare
a9f0b9d to
4de5df1
Compare
4de5df1 to
3176587
Compare
3176587 to
0105360
Compare
|
Rebased onto current
Verified locally: |
|
Hi @laskoviymishka @kevinjqliu @Fokko — could one of you take a look at this PR when you have a chance? Quick context to make the review easier:
It has been waiting on review since July; happy to split it further or address any feedback. |
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
Really nice groundwork here — reusing the concurrent manifest scan by generalizing plan_files into plan_data_files over a mapping closure is a clean way to get COW planning basically for free, and the schema-evolution test (writing replacements in the snapshot schema rather than the evolved current schema) shows the tricky case was actually thought about.
I'd hold it before merging, though — this is the primitive the overwrite/row-delta commit adapters are going to build on, so I'd like the correctness and the public contract nailed down before that stacks up.
The one that blocks for me is the batch-rewrite contract: a rewriter returning {changed: false, output: None} silently drops those rows, and if a later batch in the same file flips the file to changed, the dropped rows vanish from the replacement with no error. I'd make the orchestrator derive changed || output.is_none() itself rather than trusting every rewriter to get the flag right, and add a regression test on that exact case.
A few more I'd want to settle in this PR before the follow-ups lean on it:
- files whose rows are entirely removed by delete files land in
unchanged_data_filesinstead ofremoved— as written they can never be compacted away - the replacement writer always uses
DefaultLocationGenerator, so object-storage-layout tables get flat-layout outliers - the prefix buffer can hold an entire source file in memory on the no-change path; at minimum the "bounded by rows before the first change" comment should say so
output_rowscounts rows from unchanged files, so it can't be read as rows actually writtenCowRewriteFileis public but unconstructable outside the crate, andrewrite_batchbeing sync will be a breaking change to revisit once it's stabilized — worth deciding now
None of the API-shape ones are hard blockers on their own, but they're the kind of thing that's painful to walk back once it's released. Once the data-loss path and the two correctness items are handled I'm happy to take another pass and approve.
| result.stats.input_rows += batch.num_rows() as u64; | ||
|
|
||
| let rewrite = rewriter.rewrite_batch(batch)?; | ||
| if rewrite.changed { |
There was a problem hiding this comment.
I think there's a data-loss path here. file_changed only tracks rewrite.changed, and we only ever look at rewrite.output through the if let Some(output) below — so a batch that comes back {changed: false, output: None} silently drops its rows and leaves file_changed untouched.
The trap is that a later batch for the same file can still flip file_changed to true. Say batch 1 returns {changed: false, output: None} (rows 1–2 meant to be dropped) and batch 2 returns {changed: true, output: Some([3,4])}: we write a replacement holding only [3,4], move the original [1,2,3,4] to removed, and rows 1–2 vanish with no error. The docstring on CowBatchRewrite says to set changed whenever output differs including None, but correctness shouldn't hinge on every rewriter getting that exactly right.
I'd make the orchestrator self-enforcing — derive the effective flag and use it everywhere we currently read rewrite.changed:
let changed = rewrite.changed || rewrite.output.is_none();Alternatively, reject {changed: false, output: None} up front with a PreconditionFailed. Either's fine, but I'd want one of them before this lands. Worth a regression test on exactly this case too — nothing in the suite exercises it today, so it won't surface on its own. wdyt?
There was a problem hiding this comment.
Fixed in 66c74f2 — the orchestrator now derives the effective flag itself (let changed = rewrite.changed || rewrite.output.is_none();) and uses it everywhere rewrite.changed was read, so {changed: false, output: None} can no longer leave dropped rows in a file marked unchanged. Added two regression tests: cow_rewrite_none_output_implies_change (every batch silently dropped → file removed with no replacement) and cow_rewrite_silent_drop_before_change_loses_no_rows (your exact scenario: batch 1 {changed: false, output: None}, batch 2 {changed: true, output: Some} → replacement holds only the second batch's rows).
| pub added_data_files: Vec<DataFile>, | ||
| /// Candidate files that were read and left unchanged. | ||
| /// | ||
| /// This includes files whose visible rows were all removed by delete |
There was a problem hiding this comment.
This one I'd push back on. When a file's visible rows are all removed by delete files the reader yields zero batches, the loop never runs, and the file lands in unchanged_data_files — carrying its delete files with it, forever.
The Java side (RewriteDataFiles) drops the file in that case: the data-file-plus-delete-file pair gets compacted away rather than pinned in the table. As written, any DELETE/UPDATE adapter built on this primitive can never compact an all-deleted file, so snapshot size and delete-file read I/O only grow.
I'd treat a scan task that has delete files and produces zero rows as changed-with-no-replacement (the same terminal state as "rewriter dropped every batch"). If we do want to defer it, I'd at least frame it in the struct doc as a known gap the commit adapter has to handle, rather than as correct behavior. wdyt?
There was a problem hiding this comment.
Agreed — fixed in 66c74f2. A candidate that has delete files and reads as zero rows is now treated as changed-with-no-replacement and lands in removed_data_files, matching RewriteDataFiles, so the commit adapter can drop it together with the delete files that reference it. The unchanged_data_files doc now calls this out explicitly. One caveat to flag: no end-to-end test for this path yet, because the crate has no way to commit delete files in a fixture today — that is exactly the row-delta work in #2185/#2203. I would rather cover it with a real fixture once that lands than hand-roll delete manifests here.
| write_schema: SchemaRef, | ||
| partition_key: Option<PartitionKey>, | ||
| ) -> Result<Box<dyn crate::writer::IcebergWriter>> { | ||
| let location_generator = DefaultLocationGenerator::new(table.metadata())?; |
There was a problem hiding this comment.
We always construct a DefaultLocationGenerator here, which ignores write.object-storage.enabled. On a table using the object-storage layout the replacement files land in the flat write.data.path layout instead of the hash-entropy dirs, and if write.object-storage.path differs from write.data.path they can end up under a different prefix entirely.
Not corruption, but it makes COW output an inconsistent outlier and breaks the S3 prefix-sharding those tables opt into. I'd branch the same way the other write paths do — read write.object-storage.enabled and pick ObjectStorageLocationGenerator or DefaultLocationGenerator accordingly.
There was a problem hiding this comment.
Good catch, with a wrinkle: this repo did not actually have write.object-storage.enabled — only write.object-storage.path and write.object-storage.partitioned-paths existed, and no write path was branching on the object-storage layout at all. In 66c74f2 I added the missing property (Java semantics, default false) and build_replacement_writer now picks ObjectStorageLocationGenerator vs DefaultLocationGenerator accordingly. Covered by cow_replacement_writer_honors_object_storage_layout, which asserts the hash-entropy directory shape of the output path.
| // emit a replacement file for a source file that turns out to be | ||
| // unchanged. Once a changed batch is observed the buffered prefix is | ||
| // flushed to the writer and all subsequent batches stream straight | ||
| // through, so the in-memory footprint is bounded by the rows that |
There was a problem hiding this comment.
The comment says the footprint is bounded by the rows that precede the first change — but for an unchanged file that bound is the whole file. Every batch gets pushed to prefix and nothing drains until we drop it in the else branch, so a no-op KeepAll over a compaction-sized file buffers the entire decoded Arrow file in memory. A DELETE whose rows sit near the end of each file approaches the same bound.
Since we process files sequentially the peak is one file at a time, but that's still potentially several GB. At minimum I'd fix the comment to state the real worst case. Longer term a with_max_prefix_bytes escape hatch that falls back to just writing the replacement once the buffer crosses a threshold would cap it — happy to leave that for a follow-up as long as the bound is documented here.
There was a problem hiding this comment.
Comment rewritten in 66c74f2 to state the real bound: an unchanged file — or a first change at the very end — buffers the entire decoded source file, peak one file at a time since files are processed sequentially, but potentially several GB for a compaction-sized file. It also names the size-capped fallback that starts writing the replacement past a threshold as follow-up work.
| } | ||
|
|
||
| if let Some(output) = rewrite.output { | ||
| result.stats.output_rows += output.num_rows() as u64; |
There was a problem hiding this comment.
output_rows gets bumped for every Some(output) batch, including prefix batches on files that end up unchanged — so a KeepAll over a 3-row file reports output_rows == 3 while rewritten_files == 0 and nothing is written. cow_rewrite_keep_all_produces_no_changes actually asserts this, so it's currently baked in as intended.
The problem is a caller can't read this as "rows persisted to replacement files" — anyone cross-checking against added_data_files row counts sees a phantom mismatch. I'd either only accumulate it when file_changed, or rename to something like rewriter_emitted_rows and add a separate written_rows. wdyt?
There was a problem hiding this comment.
Went with the first option in 66c74f2: output_rows now only accumulates for files that end up changed, i.e. rows actually written to replacement files, and the doc states it always matches the row counts of added_data_files so cross-checking works. The two tests that baked in the old semantics (keep_all, delete_no_matching_rows) now assert output_rows == 0 on the unchanged path.
| /// Rewrites record batches for copy-on-write operations. | ||
| pub trait CowBatchRewriter: Send + Sync { | ||
| /// Rewrites a record batch and reports whether it changed. | ||
| fn rewrite_batch(&self, batch: RecordBatch) -> Result<CowBatchRewrite>; |
There was a problem hiding this comment.
rewrite_batch is synchronous, so a rewriter that needs async I/O — catalog enrichment, cross-table dedup, an audit lookup — has to block the runtime thread or shell out to spawn_blocking. That rules out a fair chunk of what people will want to plug in here.
Making it async later is a breaking change once this is stabilized, so I'd rather decide now. If object safety is the reason it's sync (I see cow_batch_rewriter_is_object_safe), that's a fair constraint — but then I'd document in the trait doc that blocking work isn't supported, so nobody learns it the hard way. wdyt?
There was a problem hiding this comment.
Keeping it sync — object safety for Arc<dyn CowBatchRewriter> is indeed the constraint — and the contract is now documented on the trait in 66c74f2: rewrite_batch runs on the async runtime thread driving the read/write pipeline, must not block, and async I/O such as catalog enrichment is not supported. If a real async use case shows up we can revisit with boxed futures before the API is stabilized.
| use crate::Result; | ||
|
|
||
| /// Result of rewriting a single record batch. | ||
| pub struct CowBatchRewrite { |
There was a problem hiding this comment.
Could we derive Debug on CowBatchRewrite? It's a public return type but not printable, so callers can't dbg! it or fold it into error context, and it shows up as a gap in public-api.txt. Both fields are already Debug, so it's a one-line derive.
There was a problem hiding this comment.
Done in 66c74f2 — one-line derive as you said, and public-api.txt regenerated so the gap shows up there too.
| })? | ||
| .as_ref() | ||
| .clone(); | ||
| spec.partition_type(schema).map_err(|err| { |
There was a problem hiding this comment.
This calls partition_type purely for the early-error side effect and drops the StructType, then PartitionKey::new below likely computes it again. It reads like dead code — a future reader could easily delete it and quietly remove the validation.
If PartitionKey::new already validates, I'd drop this pre-call; if it doesn't, a one-line comment on why we're binding here would save the next person the double-take.
There was a problem hiding this comment.
It is the second case — PartitionKey::new does not bind or validate, it just stores spec/schema/data; the first place the binding is checked is to_path, which the writer calls much later with a far less clear failure. Kept the pre-call and added a comment in 66c74f2 saying exactly that, so it no longer reads as dead code.
| .await?, | ||
| ); | ||
| } | ||
| let writer = writer.as_mut().expect("writer just built"); |
There was a problem hiding this comment.
This .expect("writer just built") is sound today given the is_none() check right above, but a future reorder would turn it into a runtime panic mid-stream rather than a compile error. I'd either use unreachable! to signal it's an invariant, or restructure as if let Some(w) = writer.as_mut() after the init block so the compiler enforces it.
There was a problem hiding this comment.
Fixed in 66c74f2 — restructured as let Some(writer) = writer.as_mut() else { unreachable!("writer initialized above") }; right after the init block, so the invariant is stated in code rather than asserted by message.
| // Planning already cleared the row predicate (see | ||
| // `ManifestEntryContext::into_cow_rewrite_file`), so this task | ||
| // reads every row of the source file. | ||
| let tasks = Box::pin(futures::stream::iter(vec![Ok(file.scan_task.clone())])) |
There was a problem hiding this comment.
Small one, non-blocker — file.scan_task.clone() here and file.old_data_file.clone() at the end of the loop are both avoidable since file isn't used again after the if/else. Destructuring at the top (let CowRewriteFile { old_data_file, scan_task } = file;) lets you move both instead, and FileScanTask / DataFile aren't cheap to clone (paths, projection vecs, stats maps) when it's once per candidate. Just while we're here.
There was a problem hiding this comment.
Done in 66c74f2 — the loop now destructures CowRewriteFile { old_data_file, scan_task } up front and moves both, so neither clone survives.
|
Thanks for the thorough pass — all eleven comments are addressed in 66c74f2, just pushed. The short version:
Full test suite (1768 tests), clippy and fmt are clean, and |
Add a core copy-on-write rewrite primitive that plans candidate data files, applies a caller-provided RecordBatch rewriter, writes replacement data files, and returns removed/added/unchanged file lists for future overwrite-style commit actions. - Plan candidates through the existing scan path so manifest pruning and delete-file application stay consistent with normal reads; the row predicate is cleared only for the full-file rewrite input. - Stream rewritten batches to the replacement writer: buffer only the prefix before the first changed batch, then open the writer lazily; unchanged files open no writer, fully-deleted files produce no replacement. - Bind the replacement writer and partition key to the planned snapshot's schema so rewrites survive schema evolution. - Build the replacement parquet writer via ParquetWriterBuilder::from_table_properties so replacement files honor the table's write.parquet.* properties, carry the table's encryption manager so encrypted tables are not downgraded to plaintext, and refuse to run when write.format.default is not parquet. - Keep the planner internal; CowRewriteBuilder is the only public entry and CowRewriteFile fields sit behind accessors. Cover delete-style, update-style, no-op, full-file delete, no-match delete, delete-file planning, writer edge cases, schema evolution, multi-source replacement path uniqueness, and partition-value preservation (including the null partition) end-to-end. Part of apache#2269. Co-Authored-By: Claude <noreply@anthropic.com>
- derive the effective change flag as `changed || output.is_none()` so a
rewriter returning {changed: false, output: None} cannot silently drop
rows while leaving the file marked unchanged; add regression tests
- treat a candidate whose visible rows are all removed by delete files as
removed with no replacement, so it can be compacted away with its
delete files instead of landing in unchanged_data_files forever
- honor write.object-storage.enabled when building replacement writers
(add the missing table property) so object-storage-layout tables keep
hash-entropy paths
- count output_rows only for rows actually written to replacement files
- derive Debug for CowBatchRewrite
- document the sync rewriter contract, the read-only CowRewriteFile
surface, and the prefix buffer's whole-file worst case
- drop avoidable FileScanTask/DataFile clones and replace the writer
expect() with a let-else invariant
66c74f2 to
7e949ec
Compare
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
Really close now, almost everything from last round landed, and this is down to one blocking path.
The batch-rewrite data-loss contract still isn't fully closed. Deriving changed || output.is_none() fixed the case I hit last round (a None output before any kept rows), but the prefix is still only flushed inside the if let Some(output) arm, so the mirror ordering — a prefix of kept {Some, changed: false} batches followed by a {None, changed: true} batch — skips the flush, exits with writer == None, and the "fully deleted" branch drops the buffered rows while still removing the file. With batch_size=2 and rows [1, 3, 2, 4], rows 1 and 3 vanish. I'd flush the prefix and build the writer at the point file_changed first flips true regardless of that batch's output, and add the keep-first/drop-second regression test. While that path is open I'd also swap the unreachable! in the writer branch for an Err — we don't want a panic compiled into a library path.
Everything else I asked for is in:
changed || output.is_none()is derived by the orchestrator, with the drop-first regression test- files fully removed by delete files now land in
removed_data_filesso they can be compacted away - the replacement writer honors
write.object-storage.enabledinstead of always using the default layout, with a test - the prefix's worst-case memory footprint is documented, with the size-capped fallback called out as follow-up
output_rowsno longer counts rows from files that turned out unchanged
Two smaller things worth folding in while we're here: the module doc calls equality deletes "redundant" after rewrite, which could read as safe-to-drop to a commit adapter — they still apply to other files by sequence number, so I'd say only position deletes and deletion vectors exclusively referencing a removed file should be dropped. And the fully_removed_by_deletes path has no test that drives it through real delete files rather than the batch rewriter.
Fix the prefix-drop path and the panic and I'm happy to approve — the rest is close.
| result.stats.changed_batches += 1; | ||
| } | ||
|
|
||
| if let Some(output) = rewrite.output { |
There was a problem hiding this comment.
This is the same data-loss shape I flagged last round, just from the other side. The changed || output.is_none() derivation landed and handles the drop-first/keep-later case, but the prefix only ever gets flushed inside this if let Some(output) arm — so when a file accumulates a prefix of {output: Some, changed: false} batches and then flips to changed via a later {output: None, changed: true} batch, we skip this whole block, exit the loop with writer == None, and the post-loop "fully deleted" branch drops the buffered prefix. The kept rows vanish and the file still moves to removed_data_files.
Concrete trigger with batch_size=2 and rows [1, 3, 2, 4]: batch one is {Some([1,3]), changed: false} → prefix, batch two is {None, changed: true} → the file flips changed but nothing flushes. Rows 1 and 3 are lost, and output_rows still reports 2, which also breaks its documented invariant.
I'd flush the prefix and build the writer at the edge where file_changed first becomes true, regardless of whether that batch carried output — then writer == None reliably means "nothing was ever written" and the fully-deleted branch is correct. A keep-first/drop-second regression test (the mirror of cow_rewrite_silent_drop_before_change_loses_no_rows) would lock it. Same condition as last round: I'd want this resolved before we merge.
| ); | ||
| } | ||
| let Some(writer) = writer.as_mut() else { | ||
| unreachable!("writer initialized above"); |
There was a problem hiding this comment.
unreachable! compiles a panic into a library path, which is the one thing we don't want here — even a "can't happen" invariant should return Err rather than abort the caller's process.
If you take the prefix-flush fix from the data-loss comment above and build the writer eagerly when the file first flips to changed, this branch disappears on its own. If it stays, I'd make it writer.as_mut().ok_or_else(|| Error::new(ErrorKind::Unexpected, "COW writer missing after file marked changed"))?.
| // the same as a rewriter that dropped every batch — changed with | ||
| // no replacement — so the file and its delete files can be | ||
| // compacted away instead of being pinned in the table forever. | ||
| let fully_removed_by_deletes = |
There was a problem hiding this comment.
The fully_removed_by_deletes classification is exactly what I asked for last round and reads correctly — this is the path that lets a stale data+delete pair actually get compacted away.
What's missing is a test that drives it through real delete files: every current test reaches the removed-with-no-replacement state via the batch rewriter, not via the delete-file reader returning zero rows. I'd add one that writes a data file plus a position delete covering all its rows, runs a KeepAll rewriter, and asserts the file lands in removed_data_files with added_data_files empty. Without it, a reader change that emits one empty batch instead of zero batches would silently send the file back to unchanged_data_files and re-pin the pair forever.
| //! lists must also account for delete files that reference removed files — | ||
| //! for example deletion vectors whose referenced data file is being removed, | ||
| //! and position deletes scoped to it; equality deletes remain valid but | ||
| //! become redundant once their target rows are rewritten. |
There was a problem hiding this comment.
I'd tighten this line — "become redundant" reads like a signal that a commit adapter can drop these equality delete files, and that's not safe. An equality delete applies by sequence number, so the same file can still apply to other data files that weren't part of this rewrite; dropping it would resurrect deleted rows elsewhere.
Since this module doc is the contract the overwrite/row-delta adapters will read, I'd say it explicitly: only position deletes and deletion vectors that exclusively reference a removed file should be dropped, and equality deletes must be left in place.
laskoviymishka
left a comment
laskoviymishka
left a comment
There was a problem hiding this comment.
Thanks for the update, the round-1 work is all still solid (the derived changed || output.is_none(), fully-removed files routing into removed_data_files, the object-storage layout, the output_rows accounting). But I have to be honest that this pass didn't move the two things I blocked on last round.
The prefix-drop path is unchanged: writer init and the prefix flush still live inside the if let Some(output) arm, so the mirror ordering still drops rows — batch_size=2, [1, 3, 2, 4], a kept {Some, changed:false} batch followed by {None, changed:true}, and rows 1 and 3 vanish while the file is still removed. I'd flush and build the writer the moment file_changed flips regardless of that batch's output, and add the keep-first/drop-second regression test — the existing one only covers the opposite ordering where the prefix is empty. And the unreachable! in that same branch is still there; I'd swap it for an Err so we don't compile a panic into a library path.
The two smaller round-2 asks are also still open: the module doc still calls equality deletes "redundant" (they keep applying by sequence number, so that reads as safe-to-drop to a commit adapter), and fully_removed_by_deletes still has no test that drives it through real delete files.
What's left before I can approve:
- flush the prefix + build the writer when
file_changedfirst flips, with the keep-first/drop-second test - swap the
unreachable!for anErr - reword the equality-delete "redundant" doc (and the matching
unchanged_data_filesfield doc) - a real-delete-file test for
fully_removed_by_deletes
One new thing, non-blocking: writing replacements in the snapshot schema diverges from what RewriteDataFiles does with the current schema — worth a doc line, but I still think the snapshot-schema choice itself is right. The CowRewriteFile stabilization question from round 1 is technically still open too, though the plan.rs doc now explains the deferral, so I'm fine leaving it.
The data-loss path is really the only thing standing between this and an approve — it's a small hoist. Fix that and the panic and I'll take another pass.
| //! lists must also account for delete files that reference removed files — | ||
| //! for example deletion vectors whose referenced data file is being removed, | ||
| //! and position deletes scoped to it; equality deletes remain valid but | ||
| //! become redundant once their target rows are rewritten. |
There was a problem hiding this comment.
Still reads as safe-to-drop here — "redundant" is the wording I flagged last round. Equality deletes are scoped by sequence number, so they keep applying to every un-rewritten file with a low-enough sequence number, not just the rows we rewrote; a commit adapter that reads "redundant" as droppable would resurface deleted rows elsewhere.
I'd scope the drop to position deletes and DVs that exclusively reference a removed file, and say equality deletes must be retained. The same thing bites the unchanged_data_files field doc just above (around line 113) — "drop them together with the delete files that reference them" should make explicit that equality deletes are not included.
| // replacement files must be written with this schema so that batches | ||
| // remain compatible when the table's current schema has evolved past | ||
| // the snapshot the source files belong to. | ||
| let write_schema = scan_task.schema_ref(); |
There was a problem hiding this comment.
New thought this pass, and not a blocker — to be clear this isn't walking back the snapshot-schema behavior I liked in round 1, which is still correct for keeping batches readable.
Writing replacements in the planned snapshot's schema does diverge from RewriteDataFiles / Spark DML, which write replacements in the table's current schema (projecting nulls for columns added since). So after a COW rewrite on an evolved table our replacements stay on the old schema and need a second rewrite to promote. Mixed-schema file sets are legal, so it's not a correctness fault, but it's a parity gap worth a doc line here noting the choice — and maybe a follow-up for an opt-in promote-to-current. wdyt?
| result.stats.changed_batches += 1; | ||
| } | ||
|
|
||
| if let Some(output) = rewrite.output { |
There was a problem hiding this comment.
This is the same prefix-drop path from last round, and the structure is unchanged — writer init and the prefix.drain(..) flush still live entirely inside this if let Some(output) arm.
So the mirror ordering still loses rows: with batch_size=2 and [1, 3, 2, 4], a {Some([1,3]), changed:false} batch buffers into prefix, then {None, changed:true} flips file_changed but skips this arm, so the writer is never built and the prefix is never drained. At EOF the file goes into removed_data_files with no replacement — rows 1 and 3 are gone and output_rows is inflated by the buffered count.
I'd hoist the writer-init + prefix flush to fire the moment changed first flips true, regardless of this batch's output. And cow_rewrite_silent_drop_before_change_loses_no_rows still only covers the opposite ordering (first batch None, so the prefix is empty and the bug stays invisible) — the keep-first/drop-second regression test from last round isn't here yet. I'd add it asserting the kept batch survives in the replacement and the source is removed.
| ); | ||
| } | ||
| let Some(writer) = writer.as_mut() else { | ||
| unreachable!("writer initialized above"); |
There was a problem hiding this comment.
This unreachable! is still here from last round — it compiles a panic! into the library path, and the prefix-drop fix above makes the branch easier to actually reach.
I'd return an error instead: return Err(Error::new(ErrorKind::Unexpected, "COW rewrite writer unexpectedly uninitialized after a change was detected")).
| // the same as a rewriter that dropped every batch — changed with | ||
| // no replacement — so the file and its delete files can be | ||
| // compacted away instead of being pinned in the table forever. | ||
| let fully_removed_by_deletes = |
There was a problem hiding this comment.
This branch still has no test that drives it through real delete files — every zero-input-row case in the suite goes through the batch rewriter (SilentFullDrop / DeleteEvenIds), and cow_planner_preserves_delete_files only checks that planning keeps the delete files, it never runs rewrite().
I'd add an integration test that writes a data file plus a position-delete covering all its rows, then asserts the file lands in removed_data_files with no replacement and unchanged_data_files empty.
2 participants
Summary
Core Design
write.parquet.*properties and encryption settings; tables whosewrite.format.defaultis not parquet are rejected rather than silently written as parquet.Related Work Map
flowchart LR Read["#2367 · #2414 (closed) · #2532 (closed)<br/>delete-read improvements"] --> Scan["existing scan/read path<br/>visible rows"] Scan --> ThisPR["#2752<br/>COW rewrite primitive<br/>plan + rewrite + write files"] ThisPR --> Files["removed DataFiles<br/>added DataFiles<br/>stats"] Files --> Overwrite["#2185<br/>OverwriteAction<br/>preferred COW commit path"] Files --> RowDelta["#2203<br/>RowDelta / MoR path"] Validator["#2590 (closed)<br/>SnapshotValidator sketch"] -. future commit validation .-> Overwrite Validator -. future commit validation .-> RowDeltaNotes
Test Plan
make check-public-api)