andygrove opened a new pull request, #5780:
URL: https://github.com/apache/datafusion-comet/pull/5780
## 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), but that can
only move the roll by one 1000-row step, not halve the file count.
## 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 that the two writers can still roll one 1000-row step apart
because they compare
different size estimates against the target.
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.
Also run locally: `CometIcebergWriteActionSuite` (56), plus
`CometIcebergRewriteActionSuite`,
`CometIcebergWriteDetectionSuite`, `IcebergWriteProtoTranslationSuite` and
`CometIcebergEncryptionSuite` (77 together), and the `iceberg_write` Rust
tests (35). All pass.
I also A/B'd the pacing cost in a release build — 2.4M rows of int + string
through the writer
stack, paced versus whole batches: 71.9 ms / 70.6 ms / 67.7 ms / 69.3 ms
alternating, i.e. within
run-to-run noise, and the same single output file either way.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
