viirya commented on PR #6883: URL: https://github.com/apache/datafusion-comet/pull/6883#issuecomment-6510808116
Thanks @andygrove. Addressed in 227e263ec. **Read-ahead memory is not in the pool.** That's right, and I kept it that way on purpose. A reservation for held batches can't be handed back when an operator later needs that memory, so reserving them would fail queries (non-spillable consumers in particular) that ordinary execution runs fine. Spark's own collect also keeps task results outside the execution pool, on the driver. The bound is `spark.driver.maxResultSize`, the most a collect may bring back anyway, and past it each running partition holds at most two batches. As for the JVM copying the same rows, a delivered batch is closed right after its rows are copied, so only one batch overlaps with the copy. To size the difference, I measured peak RSS for a 5M-row, 8-long-column collect (about 360 MiB of `UnsafeRow`) with a fixed 4g heap: | | run 1 | run 2 | |---|---|---| | Spark | 5.07 G | 5.05 G | | Comet | 5.13 G | 5.13 G | | Local execution | 5.36 G | 5.38 G | That's roughly 240 MB more native memory. With a 3g heap, Spark failed with an OOM and Comet with `TaskResultLost`, while local execution finished (4.27 G / 4.34 G), because it doesn't encode and decode the whole result on the driver. I added this to the accounting gaps in `memory_management.md` and explained it in `local-execution.md`. **Non-spillable consumers sharing one task's pool.** The total is bounded the same way as with ordinary execution, since all the tasks' consumers already share the executor's pool. What changes is the per-partition cap: one partition's merge, for example, can take more than one task's share and leave less for the others. That's intended. The existing many-partitions sort test in a small pool exercises the merge path. `local-execution.md` now spells this out. **The result was a single partition for every consumer.** Good catch. `CometLocalResultExec` now executes as the ordinary native block and reports the block's partitioning and ordering. Only `collect` runs the whole query as one graph. Cache, `df.rdd`, `foreachPartition`, `localCheckpoint`, and PySpark's `collect` / `toLocalIterator` (which go through `execute()`) now see the block's N partitions. The cache test now checks that the cached relation has the ordinary partition count, and a new test checks `rdd`, `toRdd`, `outputPartitioning`, per-partition sizes, and `localCheckpoint`. **`ffi.md`.** It now describes the local path: `NativeLocal.nextBatch` exports through the same `prepare_output` / `move_to_spark`, returns pending while no batch is ready, and `LocalQueryIterator` closes imported batches the same 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]
