parthchandra opened a new issue, #7278:
URL: https://github.com/apache/datafusion-comet/issues/7278
Part of #7267. Low priority.
## Problem
Two more places build one output batch that can hold more rows than
`spark.comet.batchSize`:
1. **`MergeRowsExec`** (Comet, used for MERGE INTO). For each input batch it
projects
the output of every matching instruction
(`native/core/src/execution/operators/merge_rows.rs:528-531`)
and concatenates them (`:663`). A split instruction emits two rows per
input row
(`MergeInstructionExec` doc, `:57`), so one output batch can hold up to
twice the
rows of one input batch. There is no cap. In bytes the growth may be much
less
than 2x, since the delete row of a split likely nulls the data columns
(not
verified).
2. **Grouped `AggregateExec` emit** (DataFusion). Comet's grouped aggregates
run on
DataFusion's newer aggregate streams (`enable_migration_aggregate`
defaults to
true, `datafusion-common config.rs:890`, and Comet doesn't change it).
Those build
every group into one batch (`aggregate_hash_table/common.rs:241-249`,
`next_output_batch_inner`), then slice it to the batch size
(`MaterializedAggregateOutput::next_batch`, `:484-493`). The spill path
does the
same (`take_state_batch`, `:315-317`). If the group keys or accumulator
state
(for example `max(string)` or `collect_list`) hold more than 2 GiB of
string/binary data, that one batch overflows.
## Proposed fix
- `MergeRowsExec`: emit the concatenated output in chunks of at most
`spark.comet.batchSize` rows, or emit each instruction group as its own
batch
instead of concatenating.
- Aggregate: upstream DataFusion change to build the output in batch-size
chunks
instead of materializing all groups first. Related to #4718 (aggregation
offset
overflow) and PR #4791 (open, experimental), which tries large types for
aggregated values.
## Tests
- `MergeRowsExec`: Rust test with a split instruction and a small batch size,
checking output batches stay within the batch size.
- SQL-file test: MERGE INTO with matched-update and not-matched-insert
clauses over a
binary column, compared with Spark.
## Notes
- Both are bounded (2x the rows of one input batch for MergeRows; total
group state
for aggregates), so they are less likely to hit 2 GiB than sub-issues 3, 6
and 9.
- Related, tracked in the epic's "known" list: a single string/binary GROUP
BY key
over 2 GiB panics in `binary_map.rs:470` instead of returning an error.
--
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]
