viirya commented on PR #6883: URL: https://github.com/apache/datafusion-comet/pull/6883#issuecomment-6512662636
Thanks @comphead. andygrove's 2026-10-27 points are answered above and in 227e263ec, and this round is in 28d5702dd, after merging main (the `jni_api.rs` conflict was only the new `with_output_stream` in the planner construction, now in `ExecutionContext::planner`). **The root `CollectLimitExec`.** You're right. The `DIVIDE_BY_ZERO` case was real at 64f7257d6: with one child partition, `CollectLimitExec.doExecute` skips the per-partition `take`, so `limit(1).toLocalIterator()` returned `25` where Spark fails. 227e263ec already made `CometLocalResultExec.doExecute` run the block's N partitions, which fixed that. But it also left admitting the limit with no benefit, since `executeCollect` goes through `executeTake`, which is ordinary anyway. So a root limit is no longer admitted: `CometLocalRule` converts the plan once, and the `CollectLimitExec` case is gone from `isLocal`. The limit tests now assert that it isn't admitted and that it matches Spark, and your `toLocalIterator` example is among them. The docs are updated. **Partition order as the only mode.** Done. `LocalQuery::new(plan, context, max_running, read_ahead)`, and the `execute_stream` branch is removed. I dropped the tests that only exercised `RepartitionExec` / `GlobalLimitExec` behind it (round-robin, hash, nested exchanges, pending-exchange cancel, global limit), and moved the rest (empty results, startup errors, a missing runtime, drop and first-error teardown, independent cancellation) to partition order. **`QueryHandle` duplicating `BatchProducer`.** I took the first option. The JNI pull now polls `LocalQueryStream` on its own thread, so the producer task, the std channel, the semaphore and the `completed` flag are gone, and a batch crosses one channel and one task before the pull. Runtime shutdown is still an error rather than EOF, through the per-partition `ended` flag that `PartitionOrder` already checks. Two details differ from a plain `block_on` with a timeout: - The bounded wait parks the thread (`park_timeout`, with a waker that unparks it) rather than using a Tokio timer. In the multi-thread runtime, timers are driven by workers, so when every worker is waiting in `acquireMemory`, a `tokio::time` timeout wouldn't fire and the task couldn't check for interruption. - `cancel` still drops the stream on its own thread. It unparks a waiting reader first, so it waits only for a poll the reader is already in, and `PartitionOrder::poll_next` only receives from channels and starts partitions. `Partition::spawn` stays separate from `BatchProducer`, because its backpressure is the read-ahead byte bound shared across partitions rather than a fixed per-stream channel capacity. -- 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]
