andygrove opened a new pull request, #7289:
URL: https://github.com/apache/datafusion-comet/pull/7289
## Which issue does this PR close?
Part of #7288.
## Rationale for this change
#7288 proposes Comet Native Executors: an optional deployment mode where a
Spark driver plans each query and a cluster of native processes built on Apache
DataFusion Ballista runs it, with no JVM in the data plane. This draft is the
prototype that the issue's first sub-task calls for. It runs simple queries end
to end, to check the design before building it out:
- The driver and the cluster talk only through Comet protobuf over Arrow
Flight. The driver never links Ballista.
- The cluster is a separate Cargo workspace on the Ballista 55 crates. The
scheduler wraps Ballista's scheduler and uses Comet's planner to check each
block.
- Executors rebuild each block with Comet's planner, with no JVM.
It is not meant to merge as is.
## What changes are included in this PR?
One commit per part:
1. **Wire protocol.** `native_executors.proto` defines `QueryRequest`, a
stage graph of Comet native blocks connected by the exchanges between them, and
`ResultEndpoints`. The proto crate exports a fingerprint of the `.proto` files,
so a cluster can refuse a driver from another build.
2. **Native blocks outside Spark.** `CometFragmentExec` runs a stage's block
for each partition. Its input leaves are fed by other DataFusion plans through
a new planner seam, `PhysicalPlanner::with_native_inputs`, instead of by the
JVM. A partition descriptor child (an in-memory source with one row per
partition) carries each partition's index and, for a stage that reads files,
its own copy of the block. That works around Ballista only slicing a fixed list
of leaf types per task.
3. **`native/native-executors`.** A new Cargo workspace with:
- `comet-scheduler`: an Arrow Flight service wrapping Ballista's
scheduler. It checks the version, plans each block with Comet's planner, builds
a Ballista plan (hash exchanges become `RepartitionExec(Hash)`,
single-partition exchanges become `CoalescePartitionsExec`), runs it, and
returns one endpoint per result partition.
- `comet-executor`: Ballista's executor with Comet's codec. It links no
libjvm.
4. **Driver.**
- A JNI Flight client in libcomet, behind a `native-executors` Cargo
feature that is off by default.
- Two internal configs: `spark.comet.nativeExecutors.enabled` and
`spark.comet.nativeExecutors.scheduler.url`.
- `CometNativeExecutorsExec`, a leaf that `CometRule` substitutes for the
whole converted plan, with AQE on or off.
- A stage-graph builder that refuses, naming the operator, any query that
can't run entirely natively.
**What works:** Parquet scans, filters, projections, aggregates and shuffled
joins, across hash and single-partition exchanges.
**Not yet:** sorts (range exchanges), broadcast joins, subqueries and DPP,
writes, cancellation, metrics in the Spark UI, and Spark-compatible hash
partitioning.
**Default builds are unaffected.** The configs default to off, the Cargo
feature is off, and Ballista exists only in the separate workspace, which CI
doesn't build yet.
## How are these changes tested?
- **Rust:** `native/native-executors/tests/assembly.rs` (4 tests) runs
assembled stage graphs on DataFusion alone. A `RangeScan` leaf stage feeds hash
and single-partition exchanges, and a partition-count mismatch is rejected.
- **Scala:** `CometNativeExecutorsSuite` (7 tests) starts a real
`comet-scheduler` and `comet-executor`, runs queries through Spark with the
feature on, and compares the rows with Spark's. It also asserts that the plan
ran on the cluster. The scheduler log confirms each query ran as a Ballista
job. Queries covered:
- scan, filter and project
- an aggregate across a hash shuffle, with AQE on and off
- a global aggregate
- sort-merge and hash joins
- a sort the cluster can't run, which fails naming the operator
The suite is skipped unless `COMET_NATIVE_EXECUTORS_BIN_DIR` is set, so CI
doesn't run it yet. To run it:
```shell
cargo build --manifest-path native/Cargo.toml -p datafusion-comet
--features native-executors
CARGO_TARGET_DIR=native/target cargo build --manifest-path
native/native-executors/Cargo.toml
COMET_NATIVE_EXECUTORS_BIN_DIR=native/target/debug ./mvnw test -Dtest=none
\
-Dsuites=org.apache.comet.exec.CometNativeExecutorsSuite
```
- **Existing tests:** the core planner unit tests pass (115), and clippy is
clean for the main workspace, for core with the new feature, and for the new
workspace.
--
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]
