[andygrove]

Which issue does this PR close?

Part of #7288.

The native side of this prototype is up for review on its own in #7299. Until #7299 merges, this branch also holds its commits, the first five below. Once it merges, I'll rebase this PR onto main so it holds only the driver side.

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 and TPC-H q1 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?

The first five commits are #7299, which describes them: the wire protocol, Comet native blocks as DataFusion operators, the comet-scheduler and comet-executor binaries, their tests through Flight, and a CI job for the new workspace. On top of them, one commit per part:

1. Driver.
   • A JNI Flight client in libcomet, behind a native-executors Cargo feature that is off by default. It fills in the query's Comet version and plan format, submits it to the scheduler, and returns the result partitions' endpoints to the JVM as a ResultEndpoints message.
   • 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. Each partition of its RDD reads one result partition from the executor that holds it.
   • A stage-graph builder that turns each native block into a stage and each native shuffle into an input. It declares each stage's output as Comet's shuffle writer declares its input, and refuses, naming the operator, any query that can't run entirely natively.
2. Sorts. A range exchange is sent with its sort keys. Ballista can't plan a range shuffle for a submitted physical plan yet (Static planner drops a RepartitionExec with range partitioning instead of planning a range shuffle datafusion-ballista#2680), so the scheduler reads the exchange as one partition and the sort runs in a single task. That is correct for a global sort, just serial. It is what TPC-H q1 needed.
3. CI. CometNativeExecutorsSuite is listed in dev/ci/check-suites.py as a manual suite.

What works: Parquet scans, filters, projections, aggregates, shuffled joins and sorts, across hash, range and single-partition exchanges. That includes TPC-H q1. On debug builds at SF1, q1 runs as three stages (scan and partial aggregate, final aggregate, sort) in about 4 seconds.

Not yet: broadcast joins, subqueries and DPP, writes, cancellation, metrics in the Spark UI, Spark-compatible hash partitioning, and parallel sorts.

Ballista enhancements this needs, filed upstream:

• Run submitted physical plans through the adaptive planner datafusion-ballista#2678: AQE for submitted physical plans.
• Let custom ExecutionPlan nodes restrict themselves to a task's partitions datafusion-ballista#2679: custom nodes restrict themselves to a task's partitions. This would replace the descriptor child.
• Static planner drops a RepartitionExec with range partitioning instead of planning a range shuffle datafusion-ballista#2680: a range shuffle for Partitioning::Range. This would allow parallel sorts.
• Let hash shuffles assign rows to partitions with a client's own hash function datafusion-ballista#2681: hash shuffles with a client's own hash function.

Default builds are unaffected. The configs default to off, the Cargo feature is off, and Ballista exists only in the separate workspace.

How are these changes tested?

• Native side: feat: add comet-scheduler and comet-executor for Comet Native Executors #7299's tests. tests/assembly.rs runs stage graphs on DataFusion alone, and tests/cluster.rs runs queries on the real binaries through Flight, in CI.

• Scala: CometNativeExecutorsSuite (9 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. 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
  • TPC-H q1, with and without its order by, checking row order too
  • a broadcast join the cluster can't run, which fails naming the operator

  The q1 tests read lineitem from SPARK_TPCH_DATA when it is set, and otherwise generated data with the same schema. Both pass on SF1 data from tpchgen-cli -s 1 -T lineitem -f parquet.

  The suite is manual: it skips itself unless COMET_NATIVE_EXECUTORS_BIN_DIR names the binaries. To run it:

  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: clippy is clean for the main workspace, for core with the new feature, and for the new workspace.