comphead commented on issue #7146: URL: https://github.com/apache/datafusion-comet/issues/7146#issuecomment-6322451705
### Workaround until this lands: run grouped aggregates on DataFusion's legacy stream Set both of these, at launch with `--conf` or per session with `SET`: ``` spark.comet.exec.respectDataFusionConfigs=true spark.comet.datafusion.execution.enable_migration_aggregate=false ``` **Why it avoids the first failure.** For a `Partial` aggregate over input sorted on every grouping key, DataFusion 55.2 uses `OrderedPartialAggregateStream`. It reserves memory for the whole batch it just aggregated before it emits the completed groups. For fully ordered input it returns the error when the pool refuses (`resize_or_take_state_batch` in `ordered_partial_stream.rs`). That is the `Additional allocation failed for OrderedPartialAggregateStream[0]` in the description, and the upstream report is apache/datafusion#26575. The legacy `GroupedHashAggregateStream` emits the completed groups first and only then resizes its reservation, ignoring a refusal (`emit` in `grouped_hash_stream.rs`). Comet forwards `spark.comet.datafusion.*` to the native session when `respectDataFusionConfigs` is on, so the flag reaches DataFusion. With it set, memory errors name `GroupedHashAggregateStream[...]` instead of `OrderedPartialAggregateStream[...]`. **What was checked.** A standalone DataFusion 55.2.0 reproduction with a `GreedyMemoryPool`, so Comet's pools were not involved: a `Partial` `min` over a string column, 65,536 distinct groups with 100-character keys, and a spillable consumer holding all but 256 KiB of a 64 MiB pool. | Input and pool | default (`true`) | `false` | |---|---|---| | sorted on all keys, 256 KiB left | fails: `Failed to allocate additional 2.7 MB for OrderedPartialAggregateStream[0]` | succeeds | | sorted on all keys, pool completely full | fails | succeeds | | sorted on all keys, pool has room | succeeds | succeeds | | unsorted (hash partial), 256 KiB left | succeeds | succeeds | Under pressure the legacy run returned the same 65,536 rows as the run with room (same checksum), still sorted on the keys. **Limits.** - It only removes this failure. The retry failures in `ExternalSorterMerge[0]` come from the sort (apache/datafusion#24739) and the OOM-killed executors are separate. It has not been run on the 87-key job. - It is session-wide. Every grouped aggregate Comet runs natively moves to the legacy stream, including hash partial and final aggregates. Performance and final-aggregate spilling were not measured. #6880 shows the legacy spill path failing under a tight shared pool for a shape that already runs on it. - The legacy stream does not fail when the pool is full, it stops being accounted. The reproduction also completed with the pool completely full. For fully ordered input the unreserved state is about one batch of groups (the log showed a 12 MB request), and it releases its reservation at emit while the emitted batch is still alive (apache/datafusion#26165). - It is temporary. `enable_migration_aggregate` is documented as going away, and apache/datafusion#25902 proposes removing it after 56.0.0. `respectDataFusionConfigs` is documented as a development and testing option. -- 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]
