fix: roll native Iceberg data files on iceberg-java's 1000-row grid

[andygrove]

Which issue does this PR close?

Closes #5690.

Rationale for this change

iceberg-java's RollingFileWriter re-checks the current file's size against
write.target-file-size-bytes once every 1000 rows (ROWS_DIVISOR), counted per open file.
iceberg-rust's RollingFileWriter re-checks once per write call, which for the native writer
means once per input batch. Two consequences:

• A task whose rows all arrive in one batch never rolls at all. That is what
  TestSparkDataWrite.testUnpartitionedCreateWithTargetFileSizeViaTableProperties sees: local[2]
  gives two tasks of 2000 rows, each a single batch, so the native path commits 2 files of 2000
  rows where iceberg-java commits 4 of 1000. Same for the partitioned case, where each partition's
  2000 rows arrive as one part.
• Where it does roll, the roll point depends on how Spark batched the rows rather than on the row
  count of the file, so the file layout is not reproducible from the table properties.

So the diagnosis in the issue is a bit different from what I guessed there: it is the check
cadence, not the size estimate. The size estimates do differ (parquet-rs's flushed bytes plus
its estimate of the open row group, versus parquet-mr's position plus buffered size), and nothing
bounds how far apart those two independent estimates cross the target. But a differing estimate
does not explain a halved file count on its own; the check cadence does.

What changes are included in this PR?

IcebergWriteExec now hands the iceberg-rust writer rows in 1000-row units rather than whole
batches, so the wrapped RollingFileWriter re-checks the target size on exactly the boundaries
iceberg-java checks on.

• Units are paced per destination file (RowPacer), not per input batch. Rows left over from a
  batch wait for the rows that complete their unit; that is what makes the roll point independent
  of the batch shape. Without it, a coalesce(1) insert of 4000 rows arrives as five 800-row
  batches and rolls into five 800-row files — I hit exactly that while testing.
• Pacing is per partition for the partitioned writers, so a partition's file is measured against
  its own row count. The clustered writer closes a partition's file when the next key arrives, so
  its leftovers are written out before the switch and the next partition starts a fresh grid.
• Leftovers are flushed at close. iceberg-java's writer does the same, so a trailing short file
  appears on both paths.
• The units are cut zero-copy, except when a float or double sits under a list or map. There,
  iceberg-rust's NaN-count visitor reaches children through list_array.values() /
  map_array.entries(), which ignore a slice's offset window, so a sliced batch's NaNs would be
  counted once per unit — and a nan_value_count that reaches record_count makes Iceberg's
  metrics evaluator prune the file from ordinary comparisons. Those ranges go through take.
  This is the reason materialize_run gathered unconditionally; it is now RowSlicer, which makes
  the same decision once per task from the schema and lets the clustered partition-run split take
  the zero-copy path when the schema allows it.
• iceberg-writes.md: file rolling moves out of the "cadence differs" bullet. What remains
  documented is the shared grid only — each writer rolls on a 1000-row boundary of its own file —
  with an explicit note that nothing bounds the distance between the two writers' roll points,
  since they compare different, independent size estimates against the target using different
  threshold comparisons. The page now says not to rely on file-layout parity between the writers.

What this does not change: the size estimate itself. The two TestRewriteDataFilesAction failures
in the issue depend on the rewritten files landing at the same sizes iceberg-java produced, so they
should improve with the cadence fixed but I cannot promise they pass — I could not run Iceberg's
gradle suites in my environment.

How are these changes tested?

CometIcebergWriteActionSuite — the existing target-file-size test now makes the JVM writer's own
assertion: 4000 rows in one task with a 1-byte target commit as four files of exactly 1000 rows.
That is testUnpartitionedCreateWithTargetFileSizeViaTableProperties's assertion, and it fails on
main (2 files) and failed again at 5 files of 800 before the pacing was added.

New Rust tests in iceberg_write.rs:

• The 1000-row grid holds inside one batch, across batches that are not a multiple of 1000
  (800-row batches), and for the trailing remainder.
• Per-partition pacing for both the fanout and clustered writers, including a clustered write whose
  leftovers must flush before the key changes.
• A file under the target still does not roll, even though it is written in units.
• NaN counts stay exact when a list<double> batch is cut into units — this one fails with 9
  instead of 3 if the gather is dropped, so it pins the reason for it.
• RowSlicer's slice-vs-gather decision per data type, and RowPacer's unit shapes and
  row-order preservation.
• The gate: batches pass through whole below the target, and when it is crossed mid-stream the
  first paced block completes the block the whole batches left open ([800, 800, 400, 1000, 1000]
  for five 800-row batches that cross during the third).

Also run locally: CometIcebergWriteActionSuite (56), plus CometIcebergRewriteActionSuite,
CometIcebergWriteDetectionSuite, IcebergWriteProtoTranslationSuite and
CometIcebergEncryptionSuite (77 together), and the iceberg_write Rust tests (35). All pass.

Performance

Pacing costs one writer call per 1000 rows instead of one per batch, and both ArrowWriter::write
and in_progress_size() do per-leaf-column work, so the extra calls cost more the wider the table.
Release-build A/B, unpartitioned, zstd(3), ~4M cells per case, alternating runs:

columns	whole batches	gated (reverted, see below)	pacing (this PR)
2	74.2 / 64.0 ms	64.6 / 63.6 ms	67.2 / 67.1 ms
20	103.4 / 101.3 ms	101.2 / 99.7 ms	105.4 / 103.6 ms
100	115.4 / 112.9 ms	112.7 / 114.1 ms	127.0 / 123.6 ms

So pacing costs ~4% at 20 columns and ~10-15% at 100 columns (this data is cheap to encode, which
is the pessimistic direction — higher-entropy columns dilute it).

A second commit tried to take that back for the common case, by skipping pacing until the rows
handed to a writer could have reached the target, bounded by their Arrow in-memory footprint. That
commit is reverted: as @unikdahal worked out on the PR, get_array_memory_size() measures
Arrow-side allocation while the rolling decision uses parquet's encoded-size estimate, which
includes dictionary and data-page state, and there is no invariant making the former an upper
bound on the latter. If parquet reached the target while pacing was still off, the writer could
roll, block_rows would still be counting the previous file, and every later block would be
aligned from the wrong boundary — a silently wrong grid rather than a late roll.

The suggested alternative, deciding from real writer state, is not available against the pinned
iceberg-rust: RollingFileWriter::current_written_size and current_row_num are private, and
Comet only reaches the writer through DataFileWriter. So this PR paces unconditionally and pays
the cost above. The gate can come back if iceberg-rust exposes the writer's own size.

Output size shifts by well under 1% in both directions, from page boundaries landing differently.
Two things get cheaper: fanout with many partitions makes fewer writer calls than before, since
sub-1000-row parts accumulate instead of being written per batch, and the clustered partition-run
split no longer gathers with take unconditionally.

[sunchao]

Correctness

Prior state and resulting behavior

The previous native path let iceberg-rust check file size only once per incoming batch. A single 4,000-row batch could therefore stay in one file even with a tiny target. This change introduces RowPacer, which emits complete 1,000-row units and holds at most 999 logical rows per destination until another batch or close supplies the remainder. Unpartitioned writes use one pacer. Fanout uses one per partition. Clustered writes flush the previous partition before changing keys.

I checked the actual rolling writer in the Iceberg versions pinned by the Spark profiles: 1.5.2, 1.8.1, 1.10.0 and 1.11.0. Java writes a row and checks after each 1,000th row. The locked Rust dependency checks before writing the next unit. For matching threshold decisions this gives the same non-empty file boundaries, including exact multiples, empty input and final partial units. Rust lazily opens files, so it need not create the empty trailing file that Java subsequently removes. The partition spec is fixed for a task, and the clustered path preserves the underlying writer's rejection of a previously closed partition.

The inline P2 concerns the documentation guarantee: a shared check grid does not establish that JVM/native roll points differ by fewer than 1,000 rows or at most one step. The two estimators remain independent and retain different >=/> comparisons. I found no separate verified row-loss, row-order or partition-assignment defect in the changed control flow.

Spark compatibility and validation limits

I compared task commit/abort semantics with the maintained Spark 3.5 and 4.0 branches. This patch leaves the native manifest/commit interface and existing failure handling unchanged. The required maintained Spark 3.4, 4.1 and 4.2 branches were unavailable. Those versions are not source-qualified here. The NaN safeguard is justified by the locked Iceberg visitor reading entire list/map child arrays: materializing the selected rows avoids counting children outside a new slice. The schema decision uses the decorated Iceberg target types, including floats nested through structs, lists and maps.

An isolated test of the unmodified RowPacer control flow passed 11,662 cases / 262,500,232 rows, checking full-unit sizes, ordering, no missing or duplicated rows, pending-row bounds and repeated flush. It used RecordBatch/Arrow-operation test doubles. It did not execute Arrow, Parquet, iceberg-rust, JNI or Spark. The added native and Scala tests were inspected, but the author's 35 Rust tests, 56 write-action tests and 77 other suite tests were not rerun. This head conflicts with main. The fresh check at September 8, 20:41 UTC still found no CI checks, statuses or workflow runs for this head. The newer partition-location and void-spec manifest fixes on main overlap this writer and still require conflict resolution and integration validation.

Performance

The schema decision is made once per task. For eligible schemas, complete units share Arrow buffers. Partial units may gather and concatenate, and nested list/map floating-point columns gather to preserve NaN counts. Fanout also retains a pacer and up to 999 logical rows for each partition. push constructs all complete units for a batch before the write loop consumes them, so the gathered path temporarily retains those copies as well. The row bound is not a byte-memory bound.

The author's 2.4-million-row int/string A/B numbers are close, but cover the flat path and were not reproduced here. Please include a focused real-writer benchmark for nested floating-point lists/maps and non-aligned input batches, including partitioned writes, before extending that performance claim to the new copying paths. The isolated pacing check above is not a performance benchmark, and I am not claiming a measured regression.

Design

Pacing after partition splitting is the right boundary: splitting each input batch independently would still make 800-row batches produce different files. Keeping a partition's grid across batches and flushing its remainder before a clustered switch handles that case without changing the upstream Iceberg writer API. The implementation also avoids changing size estimation or forcing extra Parquet flushes, which keeps the functional scope understandable. The documentation should describe that scope precisely: consistent sampling cadence does not guarantee identical file sizes or counts across the Java and Rust encoders.

Abstraction & complexity

RowPacer owns the small pending-row state, while RowSlicer owns the schema-dependent materialization decision. That separation earns its complexity because clustered partition splitting and rolling-unit splitting need the same NaN-safe slicing rule. The three enum variants make their different lifetimes explicit. The additional fanout map is needed to retain remainders and partition keys through close. I found no further abstraction issue requiring a change in this patch.

[sunchao]

Correctness

[P2] Remove the claimed 1,000-row bound on JVM/native roll-point drift

The new pacer puts size checks on a 1,000-row grid. It does not bound the difference between the two writers' roll points. The locked Rust writer uses flushed bytes plus in_progress_size(), while iceberg-java uses file position plus buffered size. They also retain different > and >= threshold comparisons. These are independent estimates, and nothing here limits their threshold crossings to adjacent grid points. This table promises less than 1,000 rows for any target, while the paragraph below also describes a full 1,000-row difference. Please document the shared sampling grid without a numeric bound on cross-writer row counts, and remove the matching one-step-only claim from the PR rationale. Otherwise users may rely on file-layout parity that this change does not establish.

[andygrove]

Fixed in 7a4a173. You're right that the grid does not bound the distance between the two writers' roll points, and the table was promising something this change does not establish.

The table row now reads:

write.target-file-size-bytes | any value (the two writers can choose different roll points; see accepted divergences)

and the divergences bullet documents the shared grid without a numeric cross-writer bound:

... so each writer rolls only on a 1000-row boundary of its own file.
The shared grid is all that is shared. What each writer compares against the target differs — flushed bytes plus parquet-rs's estimate of the open row group, versus parquet-mr's file position plus its buffered size — and the two use different threshold comparisons. These are independent size estimates, so nothing bounds how far apart the two writers' roll points are: they may cross the target several grid steps apart, and the resulting files can differ in row count by an arbitrary number of 1000-row blocks. Do not rely on file-layout parity between the two writers; rely only on each file rolling on its own 1000-row boundary.

I also took the one-step claim out of the PR rationale in both places it appeared — the diagnosis paragraph and the docs-changes bullet.

Worth noting these two reviews turned out to be the same mistake in two places. The "one 1000-row step" framing is exactly what made the Arrow-size gate in @unikdahal's thread look acceptable to me: I reasoned that violating its bound cost a roll up to 1000 rows late, which sounded like it fell inside a divergence I had already documented as bounded. It was not bounded, and the gate's real failure mode was a misaligned grid rather than a late roll. That commit is reverted in the same push.

[unikdahal]

Thanks for the detailed work here @andygrove . The pacing approach and added coverage look solid overall.

I found one correctness concern in the new pacing gate after tracing the pinned iceberg-rust/parquet-rs sizing path; left the details inline.

[unikdahal]

I dug a bit further into the pinned iceberg-rust/parquet-rs code here, and I don't think this bound is safe.

get_array_memory_size() measures Arrow-side allocation, whereas the rolling decision uses Parquet's encoded-size estimate, which also includes things like dictionary/data-page state. I couldn't find an invariant guaranteeing the Arrow size is always >= that estimate.

If Parquet reaches the target while pacing is still false, the rolling writer can roll before the next batch is written. block_rows would then still be tracking the previous file, so subsequent 1000-row blocks can be aligned from the wrong file boundary.

I think this needs to avoid using the Arrow-memory estimate as a correctness gate, either pace unconditionally, or make the decision from actual writer state.

[andygrove]

You're right, and I've reverted the gate in 7a4a173 — it was the newest commit, so edc5597e3 is gone in its entirety and pacing is unconditional again.

Your reasoning is the part I got wrong. I was treating get_array_memory_size() as an upper bound on what parquet would have written, on the grounds that it is capacity-based and parquet encodes and compresses. But those are different quantities measured on different sides — the rolling decision uses parquet's encoded-size estimate, which carries dictionary and data-page state that has no Arrow counterpart — and "usually larger" is not an invariant. I could not find one either.

The consequence is also worse than I had written down. I had told myself the penalty for violating the bound was a roll up to 1000 rows late, which sounded like it sat inside the divergence the two size estimates already allow. That is wrong: as you say, if the writer rolls while pacing is still false, block_rows is still counting the previous file, so every later block is aligned from the wrong boundary. That is a silently wrong grid for the rest of the task, not a late roll — and the grid is the entire point of the change.

On your second option, deciding from actual writer state: I looked and it is not reachable against the pinned iceberg-rust. RollingFileWriter::should_roll consults current_written_size(), but both that and current_row_num() are private, and Comet only reaches the writer through DataFileWriter, so there is no way to ask it what it has actually written. So unconditional pacing is the available answer and this PR takes the cost: ~4% at 20 columns and ~10-15% at 100 columns on the A/B in the description, which I have updated along with the reverted-commit rationale. The gate can come back if iceberg-rust exposes the writer's own size.

Thanks for chasing this into the pinned source rather than taking the commit message's word for the bound.

[sunchao]

Correctness

Re-reviewed edc5597e32d20f073c597a1710614fdaaa6bb2e8 against b56268d149f19e8a13cf35ec62a9145988f6f5e5, focusing on the changes since the previous review.

The previous P2 documentation finding remains. The table still promises less than 1,000 rows of cross-writer drift, and the revised paragraph and PR rationale retain the one-step claim. That finding still needs correction in both the documentation and rationale.

[P2] I independently confirmed the gate issue already raised by @unikdahal, so I am not adding another inline. A concrete counterexample uses an ordinary distinct-Int32 column with default dictionary/page settings and an 80,000-byte target. Two compact 8,192-row batches remain below that Arrow-memory threshold, but Parquet's pending estimate is 65,536 dictionary bytes plus 30,720 index-estimate bytes, totaling 96,256. On the third batch, pacing starts with block_rows=384. The writer rolls before the first 616-row completion unit, closing a 16,384-row file and putting that completion into a new file. This loses the per-file grid. A focused diagnostic executed the current gate and exact RLE estimate formula with explicit Arrow/writer stubs and confirmed that source-predicted transition while preserving all rows. It is not actual Arrow/Parquet/native/JNI/Spark execution or a performance measurement.

The maintained Spark 3.5/4.0 lifecycle sources, relevant callers and dependency lockfile match the prior review. The update does not change commit, abort, fallback or NaN-slicing interfaces. The adjusted NaN test now checks counts separately in all three rolled files, and the two new gate tests cover pass-through and phase alignment but not the estimator-bound failure above. No unchanged prior probe was rerun. At September 8, 21:44 UTC, the current head remains conflicted, with no CI checks, statuses or workflows. This is missing execution evidence, not a passing CI result. Required maintained Spark 3.4/4.1/4.2 sources remain unavailable.

Performance

The new author-reported unpartitioned A/B results quantify the motivation: forced pacing costs more on wide schemas, while gated runs approach whole-batch timings. These measurements were not reproduced, and they do not qualify the nested floating-point or partitioned copying paths requested previously. The gate can avoid extra write calls, slices and concatenations below its threshold, but that optimization needs a valid bound on the rolling predicate. Correctness cannot depend on Arrow allocation being larger than a pending encoded-size estimate.

Design

Carrying block_rows across the switch is sound only if no earlier whole-batch write could reach the rolling threshold. The distinct-Int32 path falsifies that assumption. Once a roll occurs before the completion unit, the pacer has no notification with which to reset the new file's phase. The gate should use a proven bound or writer state, with the always-paced path retained until that invariant is established. The main-side integration diff is byte-identical to the previous review and introduces no additional interaction in this update.

Abstraction & complexity

PacingPolicy reasonably keeps the slicer and target together. The new monotonic pacing flag and phase counter make the transition explicit, but handed_bytes is not the file-size state its correctness argument requires. I found no additional abstraction issue beyond fixing that state invariant.

[sunchao]

Correctness

Re-reviewed a7592cfcedc910b1f007e431b0833e04c30b90c9 against 424c31aa79d13fddf743ffa29bae3c6f146e6c5e, including the increment from the previously reviewed edc5597e and the overlapping rebase changes.

Both prior P2 findings are addressed. The documentation finding is fixed in the table, divergence explanation, and PR rationale. The text now distinguishes the shared 1,000-row grid from the independent Java/Rust size estimates, with no numerical bound on cross-writer drift. The Arrow-size gate concern is fixed by removing the gate, byte estimate, and phase-switch state. The resulting RowPacer, RowSlicer, and type predicates are byte-identical to the original always-paced implementation. No whole-batch bypass remains to let a file roll before a partial grid unit completes.

The rebase preserves main's partition-location and void-spec manifest handling. It also carries clustered_write_err through all three paced write sites: full units, a previous partition's remainder, and the final close-time remainder. This matters for short A/B/A input, whose closed-partition rejection now occurs during the final flush. The error still escapes before manifest production and driver commit. Existing native and Scala short-input tests exercise that integration path. I rechecked the maintained Spark 3.5/4.0 write lifecycle sources and the unchanged Comet commit/abort callers. No new or remaining P1/P2 issue was verified.

The completed Linux lint, Scala syntactic lint, and Spark 4.1/JDK 17 build jobs checked out merge f2222454, whose parents are this base/head and whose tree equals the reviewed head. The build explicitly skipped tests. The description still lists a former gate test, but that removed test is not current-head coverage. At September 9, 17:25 UTC, the fresh check found 17 successful checks, 7 skipped, 6 running and 1 queued, with no reported failure. Runtime CI remains in progress at the review refresh, so this is not a claim that native or JVM runtime validation has passed. No local native/JNI/Spark test or benchmark was run, and no prior standalone probe was rerun. Maintained Spark 3.4/4.1/4.2 sources remain unavailable.

Performance

Reverting the gate restores the known cost of one writer call per 1,000 rows, including repeated per-column writer and size-estimation work. The updated author-reported A/B results explicitly accept about 4% overhead at 20 columns and 10–15% at 100 columns on the tested cheap-to-encode input. That is a stated correctness tradeoff, not evidence that the pacing change is faster for every schema. I did not reproduce those measurements. They still do not qualify nested floating-point or partitioned copying paths.

The retained schema-based slicing decision avoids copying ordinary complete units. Remainders and nested list/map floating-point slices retain the materialization costs already discussed. Removing get_array_memory_size() also removes the gate's extra allocation-size traversal. I found no additional performance issue introduced by this revision.

Design

Unconditional pacing removes the unsupported relationship between Arrow allocation size and Parquet's pending encoded-size estimate. The restored invariant is local: every nonterminal unit handed to a destination writer has 1,000 rows. It no longer requires observing private upstream rolling state or recovering a phase after an unseen roll. Passing the clustered splitter into close is justified because deferred rows can produce the same partition-specific error there as during ordinary writes. The rebase is now integrated with the authoritative base, rather than leaving the earlier conflicts unresolved.

Abstraction & complexity

Removing PacingPolicy, handed_bytes, block_rows, and the monotonic gate flag simplifies the state machine. RowPacer again owns pending rows and their count, while RowSlicer owns the existing NaN-safe materialization decision. Error translation stays in the shared helper and uses the actual key for each deferred write. No further abstraction change is needed for this update.

[andygrove]

Merged. Thanks @unikdahal and @sunchao