viirya opened a new pull request, #6883: URL: https://github.com/apache/datafusion-comet/pull/6883
## Which issue does this PR close? Part of #6499. ## Rationale for this change This is the first part of the split proposed in #6499 for local execution: running a whole query as one DataFusion graph when the application has a local master. It adds the execution lifecycle, the JNI bridge and result delivery, admitting only native blocks whose leaf is a range, so the mechanism can be reviewed before admission grows. It follows the direction agreed in #6499: the graph is built from the operator tree `CometExecRule` produces and lowered with `PhysicalPlanner::create_plan`, and the session and memory pool come from `create_execution_context`. ## What changes are included in this PR? - `spark.comet.exec.local.enabled` (internal, default false). `CometRule` offers the converted plan to `CometLocalRule`, which admits it only as one native block with no JVM input whose only leaf is a `CometRangeExec`, with deterministic expressions and no subquery or JVM callback. A root `CollectLimitExec` is kept. These plans have no exchange, so AQE does not wrap them and they are admitted with AQE enabled; there is no Spark version gate. - The result task creates one graph for all partitions. `create_execution_context` provides the session and memory pool, so the pool acquires memory through the task's `CometTaskMemoryManager` and spills to Spark's local directories. Each partition's operators are created by the planner for that partition, so range slices and partition-dependent state match ordinary Comet execution, and `RangeScan`'s `mayOverflow` keeps overflowing ranges on Spark. - `native/local` runs the partitions concurrently and delivers them in partition order, so results keep Spark's row order and a limit returns the same rows as Spark. Closing a query drops its streams on the calling thread, as `BatchProducer::stop` does. - `CometLocalExec` reports `UnknownPartitioning(1)`, so a query over a cached local result still gets its required exchange. `CometLocalResultExec` hands `collect` and `take` results to the driver within the JVM and enforces `spark.driver.maxResultSize` while copying. - `ExecutionContext::planner` now builds the per-partition planner for both `execute_plan` and local queries. ## How are these changes tested? - `native/local` tests cover partition-order delivery, early failure from a later partition, cancellation while every Tokio worker is busy, the one-batch handoff and runtime shutdown. - `CometLocalExecutionSuite` (17 tests, passing on Spark 3.4, 3.5, 4.0, 4.1 and 4.2) compares results and order with Spark, including ranges near the long boundaries and ranges Spark overflows, and covers declined expressions, AQE on and off, limits, a query over a cached local result, the result size limit, job and task cancellation and native errors. After each test it checks that no native query, imported Arrow memory or result slot is left. This pull request and its description were written by Isaac. -- 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]
