viirya opened a new pull request, #6071:
URL: https://github.com/apache/datafusion-comet/pull/6071
## Which issue does this PR close?
Closes #1204.
## Rationale for this change
Comet currently repeats native plan construction for tasks executing the same
native block. For workloads with many short tasks and wide expression trees,
this adds CPU and allocation overhead.
This PR allows tasks within the same executor, Spark stage attempt, and
native
block to execute the same DataFusion physical plan using their partition IDs.
It requires no changes to DataFusion.
## What changes are included in this PR?
- Add an executor-local cache for decoded protobuf plan definitions.
- Add a stage-scoped registry for sharing actual DataFusion physical
operators.
Task-specific inputs are bound through TaskContext.
- Support eligible JVM-input pipelines containing projection, filter,
COUNT/SUM/AVG/MIN/MAX aggregates, partitioned hash joins, and full sorts.
- Isolate execution by stage attempt and partition, and attribute metrics by
partition. Retries, speculative attempts, and duplicate partition claims
use
private native plans.
- Fall back to private native plans for unsupported blocks, including native
file scans, shuffle operators, Top-K, and positive sort offsets.
- Use weak registry references so physical plans and their metrics can be
released when the last task releases its reference. Idle gaps may therefore
cause rebuilding.
- Keep both configuration options internal and disabled by default.
Initial shared-plan construction currently holds the registry mutex, which
can block lookups for unrelated plans.
## How are these changes tested?
Correctness and lifecycle validation:
- 22 native shared-plan tests and 11 decoded-plan cache tests.
- 155 tests across CometExecSuite and CometExecIteratorLifecycleSuite.
- Spark SQL `dev/local-ci.sh spark sql_core-1` with both options enabled:
12,861 passed, 0 failed.
- Coverage includes actual operator identity, concurrent partitions, output
equivalence with private plans, retry isolation, partition metrics,
aggregate modes, DISTINCT aggregates, and plan/metrics reclamation.
- Rust formatting, Clippy, JVM formatting, and license checks passed during
development.
The PR branch retains the implementation and correctness tests byte-for-byte
from the validated development branch. After extracting this branch, diff,
Rust formatting, and CI suite-registration checks were rerun; the full test
suites were not rerun.
Performance measurements used Spark 4.1.3, local[4], 1,024 partitions, and
32 rows per partition on an Apple M4 Max. Timing measurements used two JVM
runs, alternating disabled/enabled modes, with four warmups and eight timed
iterations per mode and case in each run.
| Workload | Task CPU | Process CPU | Rust allocated bytes | Query elapsed
time |
|---|---:|---:|---:|---:|
| 16-column projection/filter | -39.7% | -24.9% | -56.1% | -4.0% |
| 64-column projection/filter | -55.8% | -46.7% | -65.6% | -12.6% |
Allocation was measured separately with diagnostic instrumentation and
represents cumulative requested Rust allocation bytes, not peak memory.
Both decoded-plan caching and physical-plan sharing were enabled, so these
results do not isolate their individual contributions.
Wide aggregates did not show a meaningful elapsed-time improvement, and the
measurements did not establish a reduction in peak process memory. These
results describe a local short-task workload, not multi-executor production
performance.
--
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]
