andygrove opened a new pull request, #7299:
URL: https://github.com/apache/datafusion-comet/pull/7299
## Which issue does this PR close?
Part of #7288.
## Rationale for this change
#7288 designs 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. The draft
prototype, #7289, runs TPC-H q1 that way end to end.
This PR lands the native side of that prototype: everything needed to run a
Comet stage graph on a cluster, without connecting it to Spark yet. Users see
no change, and the main Comet build gains no dependency. The Spark driver side
follows in #7289, which will be rebased onto main once this merges.
## What changes are included in this PR?
One commit per part; it reviews best commit by commit.
1. **Wire protocol.** `native_executors.proto` defines `QueryRequest`: a
stage graph of Comet native blocks, connected by the exchanges between them,
with each partition's plan for stages that read files. The proto crate exports
a fingerprint of the `.proto` files, so a cluster can refuse a query from
another Comet build.
2. **Native blocks outside Spark.** `CometFragmentExec` runs a stage's block
for each of its partitions. A new planner seam,
`PhysicalPlanner::with_native_inputs`, feeds the block's input leaves from
other DataFusion plans instead of from 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
(apache/datafusion-ballista#2679). Nothing in the main build calls any of this
yet.
3. **`native/native-executors`.** A new Cargo workspace, with its own
lockfile, that builds two binaries on the Ballista 55 crates:
- `comet-scheduler` wraps Ballista's scheduler in an Arrow Flight
service. `GetFlightInfo` takes a `QueryRequest`, checks the client's Comet
version and plan format, plans each block with Comet's planner, builds a
Ballista physical plan, runs it, and returns one endpoint per result partition.
Each endpoint carries an opaque ticket for `DoGet` on the executor that holds
the partition.
- `comet-executor` is Ballista's executor with Comet's codec. It links no
libjvm.
Exchanges map onto Ballista's planner like this:
| Exchange | Ballista plan
|
| --------------------------- |
---------------------------------------------------------------------------------------
|
| Hash | `RepartitionExec(Hash)`, partitioned with
DataFusion's hash (apache/datafusion-ballista#2681) |
| Single partition, broadcast | `CoalescePartitionsExec`
|
| Range | `CoalescePartitionsExec`, so the sort
runs in one task (apache/datafusion-ballista#2680) |
4. **Tests through Flight.** `tests/cluster.rs` starts the two binaries and
runs queries the way a driver would.
5. **CI.** A new `linux-test-native-executors` job checks formatting, runs
clippy and runs the workspace's tests, since the existing Rust jobs don't cover
a separate workspace. It builds its dependencies without a cache, in parallel
with `linux-test-rust`.
The workflow comment and the README describe what to do when Comet moves to
a DataFusion major that Ballista hasn't released yet: the upgrade can turn the
job off until Ballista catches up. Other Ballista changes this will want are
apache/datafusion-ballista#2678 (AQE for submitted physical plans) and the
issues linked above.
## How are these changes tested?
- **Stage graphs on DataFusion alone:** `tests/assembly.rs` (5 tests) runs
them with no Ballista. A `RangeScan` leaf stage feeds hash, range and
single-partition exchanges, and a partition-count mismatch is rejected.
- **The binaries through Flight:** `tests/cluster.rs` (3 tests) starts
`comet-scheduler` and `comet-executor` and runs queries through the scheduler's
Flight service, reading every result partition with `DoGet`. It covers a hash
shuffle and a range exchange, and checks that a query from another Comet build
is refused. It runs in the new CI job.
- **End to end from Spark:** the same binaries pass the Spark suite in
#7289, which runs scans, aggregates, joins and TPC-H q1 on SF1 data from a
Spark driver.
- **Existing tests:** the core planner unit tests pass, and clippy is clean
for the main workspace and the new one.
--
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]
