parthchandra opened a new issue, #7267:
URL: https://github.com/apache/datafusion-comet/issues/7267
## Summary
Arrow `Utf8` and `Binary` arrays store all values of one array in a single
buffer
indexed by 32-bit (i32) offsets. One array (one column of one batch) can
therefore
hold at most 2^31-1 bytes of values.
Several native operators in Comet build one array out of many input rows:
they
merge small batches into bigger ones, or concatenate a whole buffered input
into a
single batch. Some batch sources (broadcast, table cache) also build batches
that
are larger than, and unaffected by, `spark.comet.batchSize`. When a column
has large values (for example serialized blobs of a
few hundred KB to a few MB per row), those merged arrays cross 2 GiB and the
query
fails with:
```
CometNativeException: Offset overflow error: <n> // n > 2147483647
```
No single value needs to be large. A Parquet `BYTE_ARRAY` value is capped at
2^31-1, so the overflow is always the sum of many values in one merged array.
Spark does not fail on the same data. Its vectorized reader has the same ~2
GiB
per-column-per-batch limit (`WritableColumnVector`, "Cannot reserve
additional
contiguous bytes in the vectorized reader"), but its batches never cross a
row group
and nothing downstream re-merges them.
## Example
A table with a binary column averaging ~0.5-1 MB per row:
```sql
SELECT blob_col, other_col FROM table_a WHERE length(blob_col) > 0
```
Comet plans `ProjectionExec -> FilterExec -> <native scan>`. Each scan batch
holds
at most min(batch size, rows in one row group), because the Parquet reader
builds one
reader per row group; with typical row groups that is far below 2 GiB.
`FilterExec` then merges them
back up to 8192 rows, about 4-8 GB of `blob_col`, and the merge fails.
Setting
`spark.comet.batchSize=1` still fails, because `FilterExec` ignores that
setting
(sub-issue 1).
## Root causes
There are three separate problems.
1. **Batch size not honored by an operator.** `FilterExec` is the only native
operator Comet builds with a hardcoded batch size (DataFusion's default,
8192)
instead of the session value. So `spark.comet.batchSize` can't be used as
a
workaround.
2. **Batch size not honored by a batch source.** The broadcast exchange
merges the
whole broadcast side into one batch (and fails on the JVM side at about 1
GiB per
column), and the in-memory table cache uses Spark's cache batch size
(default
10000). Both reach native code unchanged.
3. **Row-count-only limits and whole-input concatenation.** Even with the
right batch
size, several code paths decide how many rows go into one output array by
row
count only, or concatenate an entire buffered input:
- arrow's `BatchCoalescer` (used by `FilterExec` and the native shuffle
writer)
has no byte limit.
- `interleave` / `concat` in the shuffle writer, TopK, joins and window
operators
have no retry when they hit the offset limit.
- DataFusion's sort merge is the exception: it retries with fewer rows on
`OffsetOverflowError` (`sorts/builder.rs`, `retry_interleave`).
## Affected operators
Validated against main at 64518088a1c. Line references are for DataFusion
55.2.0
and arrow 59.3.1.
| Operator | Where it overflows | Risk | Sub-issue |
|---|---|---|---|
| `FilterExec` | Coalescer target hardcoded to 8192 rows (`filter.rs:80`) |
High, any filter over large values | 1 |
| Native shuffle writer | `interleave_record_batch` and `BatchCoalescer`, no
overflow retry; blocks over i32::MAX bytes rejected | Medium | 2 |
| `WindowAggExec` | `concat_batches` over the whole task input
(`window_agg_exec.rs:539`); used for `percent_rank`, `cume_dist`, `ntile` and
any aggregate over an UNBOUNDED FOLLOWING frame, including `sum(x) OVER
(PARTITION BY k)` | High | 3 |
| `BatchCoalescer` (arrow) / `LimitedBatchCoalescer` (DataFusion) | No byte
limit | Root of 1 and 2 at default batch size | 4 |
| TopK (`SortExec` with fetch) | `emit` interleaves all K rows at once
(`topk/mod.rs:1015`) | Medium, large LIMIT | 5 |
| `HashJoinExec`, `NestedLoopJoinExec` | Build side concatenated into one
batch (`hash_join/exec.rs:2365`, `nested_loop_join.rs:968`) | Medium, large
build side | 6 |
| `HashJoinExec` (build side is the outer/semi side) | Unmatched build rows
handed downstream as one batch (`hash_join/stream.rs:1038-1054`) | Low | 6 |
| `BoundedWindowAggExec` | Buffer concatenated on every batch while output
is held back (`bounded_window_agg_exec.rs:706`); long RANGE tie groups, large
FOLLOWING or lag/lead offsets | Low to medium | 7 |
| Any of the above | Values stored as i32-offset arrays | Long-term
alternative | 8 |
| Broadcast exchange | All broadcast batches merged into one
(`Utils.scala:379`, `coalesceBroadcastBatches`); Arrow Java fails at ~1 GiB per
column with an uncaught exception | Medium, large broadcast side | 9 |
| In-memory table cache | Batch size from
`spark.sql.inMemoryColumnarStorage.batchSize` on the row and AQE write paths
(`ArrowCachedBatchSerializer.scala:664`, `CometCacheBuildRule.scala:63`) |
Medium, cached tables | 10 |
| `MergeRowsExec`, `AggregateExec` emit | Up to ~2x input rows
(`merge_rows.rs:663`); all groups in one batch before slicing
(`aggregate_hash_table/common.rs:241-249`) | Low | 11 |
| `FilterExec` / `LimitedBatchCoalescer` memory | Buffered copies and the
`concat` output are not tracked by the memory pool | Medium (memory accounting)
| 12 |
Known and tracked elsewhere: multi-column GROUP BY keys over 2 GiB fail with
a
clear "offset overflow" error (`multi_group_by/bytes.rs:204`); a single
string/binary GROUP BY key over 2 GiB panics instead (`binary_map.rs:470`).
See
#4718 and PR #4791 (open, experimental).
Other offset overflow issues outside this epic: #7247 (a large scalar string
expanded to the batch length) and #7254 (a nested dictionary-encoded binary
column
flattened to `Utf8` per batch).
## Audit: what already honors the batch size
`spark.comet.batchSize` reaches native code as the session batch size, and
every
plan runs with a task context built from that session, so operators that
read the
batch size at execute time see it. Comet runs no DataFusion physical
optimizer
rules that could reset it.
Honors it: the native Parquet, Iceberg and CSV scans (one row group at most
per
batch; Iceberg deletes and transforms only shrink batches), range, `SortExec`
(including spill re-merge, which retries on overflow), `AggregateExec`
output,
`HashJoinExec` probe output, `NestedLoopJoinExec` output,
`SortMergeJoinExec` (can
reach about 2x the batch size), explode, the native shuffle writer, JVM
inputs
(Spark-to-columnar, local table scan; except the AQE cache build, sub-issue
10) and
JVM columnar shuffle.
Pass-through only: projections, limits, expand (one batch per projection,
each the
size of its input), sample, rank limit, dynamic filters and shuffle read
(one block
per batch, as written).
Not yet checked: Celeborn shuffle read framing, Python UDF (`mapInBatch`)
output,
sort-merge join spill read-back, the Iceberg Java fallback reader.
## Sub-issues
- [ ] 1. Pass the session batch size to `FilterExec`
- [ ] 2. Native shuffle writer: retry with fewer rows on offset overflow
- [ ] 3. `WindowAggExec` concatenates the whole task input into one batch
- [ ] 4. Byte limit in arrow `BatchCoalescer` and DataFusion
`LimitedBatchCoalescer` (upstream)
- [ ] 5. TopK `emit` interleaves all K rows into one batch (upstream)
- [ ] 6. Hash join and nested loop join build side concatenated into one
batch (upstream)
- [ ] 7. `BoundedWindowAggExec` concatenates the unfinished partition buffer
- [ ] 8. Read large string/binary columns as view types
- [ ] 9. Broadcast exchange merges every batch into one
- [ ] 10. In-memory table cache ignores `spark.comet.batchSize`
- [ ] 11. `MergeRowsExec` and aggregate emit build batches larger than the
batch size
- [ ] 12. Filter merge buffers are not tracked by the memory pool
Suggested order: 1, then 2, 4, 9 and 10 in parallel, then 3, then 5-7, 11
and 12 as
upstream allows (12 pairs with 4). 8 is a separate feature.
## Testing approach
A real 2 GiB reproduction is too heavy for CI. Each sub-issue should instead:
- test the mechanism directly, for example that the operator honors a small
batch
size or byte limit, using a few MB of data and a small configured limit;
and
- where an end-to-end check is possible, compare results with Spark at a
small
`spark.comet.batchSize` over a column with values of a few hundred KB.
Prefer SQL-file tests. Use Scala or Rust tests only when the property being
tested
(batch row counts, byte sizes) can't be seen from SQL.
## Workaround today
For the filter case, `spark.comet.exec.filter.enabled=false` moves filters
back to
Spark. Lowering `spark.comet.batchSize` has no effect on `FilterExec` until
sub-issue 1 lands. After that, lowering it (for ~1 MB rows, ~1024 or less)
avoids
the filter and shuffle cases, but not the whole-input cases (3, 6, 7) or the
batch
sources that ignore it (9, 10).
--
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]
