mbutrovich opened a new pull request, #4982:
URL: https://github.com/apache/datafusion-comet/pull/4982
## Which issue does this PR close?
Relates to #4944.
## Rationale for this change
`CometIcebergNativeScanExec` serializes an Iceberg scan's shared data into a
single `IcebergScanCommon` protobuf message that is broadcast to executors. On
a large merge-on-read pipeline this was reported to crash on the driver:
```
java.lang.NegativeArraySizeException: -954238687
at
org.apache.comet.shaded.protobuf.AbstractMessageLite.toByteArray(AbstractMessageLite.java:46)
at
org.apache.comet.serde.operator.CometIcebergNativeScan$.serializePartitions(CometIcebergNativeScan.scala:1025)
```
`AbstractMessageLite.toByteArray` allocates `new byte[getSerializedSize()]`,
and `getSerializedSize()` returns `int`. The reported value implies the common
message serialized to ~3.11 GiB (`-954238687 + 2^32 = 3340728609`), so the size
wrapped negative and the array allocation threw. protobuf cannot represent a
single message larger than 2 GiB.
A likely contributor is `delete_files_pool`. Delete files were pooled by
whole set: each `DeleteFileList` held full copies of every `IcebergDeleteFile`
in a `FileScanTask`'s delete set. Under Iceberg's default partition delete
granularity, one position-delete file applies to every data file in the
partition with a compatible sequence number (`DeleteFileIndex.forDataFile`).
Interleaving inserts and deletes staggers data-file sequence numbers, so each
task sees a different subset of the same delete files. Set-level pooling then
re-serialized a shared delete file once per distinct set, which grows
quadratically with the number of delete commits. Iceberg's own
`DeleteFileIndex` stores each `DeleteFile` once and references it many times;
the serde did not preserve that sharing.
This PR removes that duplication. It should reduce the common message size
substantially on merge-on-read tables with shared delete files, which is
expected to resolve the overflow, though the original reporter's workload has
not yet been confirmed against this change.
## What changes are included in this PR?
Normalize the wire format to intern each delete file once and reference it
by index, mirroring Iceberg's in-memory model:
- **proto** (`operator.proto`): add `repeated IcebergDeleteFile
delete_file_pool` to `IcebergScanCommon`; change `DeleteFileList` to hold
`repeated uint32 delete_file_indices` into that flat pool.
- **Scala serde** (`CometIcebergNativeScan.scala`): intern each delete file
into the flat pool (keyed by path, its identity), then dedup per-task sets as
lists of pool indices. The set-level pool is retained so tasks sharing a delete
set still share one `DeleteFileList`.
- **Rust decode** (`planner.rs`): build the flat `delete_file_pool` once,
then resolve each `DeleteFileList`'s indices against it. An out-of-range index
is a hard error rather than a silent drop, consistent with the existing
invalid-content-type and invalid-`schema_idx` handling.
Error-handling semantics are unchanged: delete extraction and serialization
failures at execution time remain fatal; the Rust decode fails loudly on a bad
index.
## How are these changes tested?
New regression test in `CometIcebergNativeSuite`: `"delete file pool does
not duplicate shared delete files (serde size)"`. It builds a merge-on-read
table with default partition delete granularity, interleaves 5 insert/delete
rounds to stagger sequence numbers, then parses the serialized
`IcebergScanCommon` and asserts each delete-file path appears exactly once in
the raw message bytes (a format-agnostic check). It also guards against a
vacuous pass by requiring that delete files are actually shared across tasks
(`totalReferences > distinctPaths.size`).
This test fails on the pre-change serde (a shared delete file appears 2-5
times in the common message for the 5-round setup: 5 distinct files, 15
references) and passes after the fix. `checkSparkAnswer` in the same test
confirms the deletes are still applied correctly through the new index-based
decode path. The existing positional-delete, equality-delete, dropped-column
equality-delete, `bytes_scanned`, and compaction delete tests in the suite
exercise the decode path end to end.
--
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]
