feat: prototype Comet Native Executors

.github/workflows/pr_build_linux.yml

@@ -106,6 +106,7 @@ jobs:
         run: |
           rustup component add rustfmt
           cd native && cargo fmt --all -- --check
+          cargo fmt --manifest-path native-executors/Cargo.toml --all -- --check
 
   # Fast syntactic-only scalafix check. Parses sources without compiling, so it
   # surfaces version-independent style issues (e.g. RedundantSyntax) in seconds,
@@ -476,6 +477,36 @@ jobs:
       - name: Rust test steps
         uses: ./.github/actions/rust-test
 
+  # native/native-executors, the Comet Native Executors cluster (issue #7288),
+  # is a Cargo workspace of its own so that the main build never depends on
+  # Ballista. linux-test-rust doesn't build it, so this job does, and the lint
+  # job checks its formatting. When Comet moves to a DataFusion major that
+  # Ballista hasn't released yet, the upgrade can turn this job off until
+  # Ballista catches up.
+  linux-test-native-executors:
+    needs: lint
+    name: ubuntu-latest/native-executors
+    if: ${{ !inputs.cache-refresh-only && inputs.profiles != 'nightly' }}
+    runs-on: ubuntu-24.04
+    container:
+      image: amd64/rust
+    steps:
+      - uses: actions/checkout@v7
+
+      - name: Setup Rust & Java toolchain
+        uses: ./.github/actions/setup-builder
+        with:
+          rust-version: ${{ env.RUST_VERSION }}
+          jdk-version: 17
+
+      - name: Check Cargo clippy
+        run: |
+          cargo clippy --manifest-path native/native-executors/Cargo.toml --color=never \
+            --all-targets -- -D warnings
+
+      - name: Run Cargo test
+        run: cargo test --manifest-path native/native-executors/Cargo.toml --color=never
+
   linux-test:
     # `lint` is already upstream via build-native; it is listed here so this
     # job can read its `profile-matrix` output.

dev/ci/check-suites.py

@@ -42,6 +42,7 @@ def file_to_class_name(path: Path) -> str | None:
         "org.apache.comet.CometIcebergSchedulerFailureSuite", # manual, multiple hosts + SIGKILL
         "org.apache.comet.cloud.s3.CometS3CredentialBridgeSuite", # manual test suite
         "org.apache.comet.cloud.s3.HadoopS3ACredentialProviderAdapterBridgeSuite", # manual test suite
+        "org.apache.comet.exec.CometNativeExecutorsSuite", # manual, needs comet-scheduler and comet-executor binaries
         "org.apache.comet.shuffle.CelebornReflectionCompatibilitySuite", # dedicated version matrix
         "org.apache.spark.sql.comet.CometPlanStabilitySuite", # abstract
         "org.apache.spark.sql.comet.ParquetDatetimeRebaseSuite", # abstract

dev/release/rat_exclude_files.txt

@@ -6,6 +6,7 @@
 .claude/skills
 CLAUDE.md
 native/Cargo.lock
+native/native-executors/Cargo.lock
 dev/copyright/scala-header.txt
 dev/diffs/iceberg/*.diff
 dev/release/requirements.txt

docs/source/contributor-guide/index.md

@@ -51,6 +51,7 @@ Floating-Point Semantics <floating_point>
 JVM Shuffle <jvm_shuffle>
 Native Shuffle <native_shuffle>
 Memory Management <memory_management>
+Comet Native Executors <native_executors>
 Iceberg Writes <iceberg-writes>
 ANSI Error Propagation <sql_error_propagation>
 S3 Credential Provider Design <s3-credential-provider-design>

docs/source/contributor-guide/native_executors.md

@@ -0,0 +1,165 @@
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements.  See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership.  The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License.  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied.  See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+# Comet Native Executors
+
+**Experimental.** Comet Native Executors run whole Spark queries without a JVM in the data plane. A
+Spark driver plans each query with Catalyst and Comet, then submits the plan as a stage graph of
+Comet native blocks to a cluster of one `comet-scheduler` and any number of `comet-executor`
+processes, built on [Apache DataFusion Ballista](https://github.com/apache/datafusion-ballista).
+The design is in [issue #7288](https://github.com/apache/datafusion-comet/issues/7288), and
+`native/native-executors/README.md` describes the cluster, how to build it and how to test it.
+
+## Current limitation: whole queries only
+
+The driver offloads a query only if every operator and expression in it can run natively
+(`CometNativeExecutors` in `CometNativeExecutorsExec.scala`). Otherwise it refuses the query,
+naming the operator that couldn't run and Comet's fallback reasons for it. That includes plans that
+call back into the JVM, such as JVM scalar UDFs and Arrow Python UDFs, because the cluster has no
+JVM.
+
+Data already crosses between Spark and the cluster in two places:
+
+- **Scalar subqueries.** Spark runs each scalar subquery, and the driver writes its result into the
+  plan as a literal before submitting the query. Data flows from Spark to the cluster, but only one
+  value.
+- **Query results.** The driver reads the final stage's output over Arrow Flight, one endpoint per
+  result partition (`ResultEndpoint`). Data flows from the cluster to Spark, but only at the root
+  of the plan.
+
+## Possible future research: hybrid Spark and native execution
+
+A query could run partly on the cluster and partly in Spark, so that a query with one unsupported
+operator or UDF can still run most of its work natively. The approaches below generalize the two
+crossings above. They differ mainly in which way data crosses the boundary. Moving data from the
+cluster into Spark is much easier than moving data from Spark into the cluster.
+
+### A native subtree as a leaf of the Spark plan
+
+The driver could offload the largest native subtrees under an operator that must run in Spark,
+rather than requiring the whole plan to be native. Spark would see each offloaded subtree as a leaf
+with one partition per result endpoint, and each Spark task would read its partition from the
+cluster over Flight. The batches arrive as Arrow, so Comet's columnar operators above the leaf can
+consume them without converting them to rows.
+
+This reuses the result path that already works. What remains:
+
+- Serving partitioned results in pipelined mode. Today the scheduler coalesces a pipelined job's
+  results to one partition.
+- Reporting the cluster's output statistics to Spark's adaptive query execution.
+- Choosing where the native region should end.
+
+It suits the common shape where the scans, joins and partial aggregates are native, and an
+unsupported operator or UDF sits near the top of the plan.
+
+### Small Spark outputs as inputs to native stages
+
+Spark could compute a whole relation that fits in a broadcast, such as the build side of a
+broadcast join, rather than a single value. The driver would encode the relation as Arrow IPC and
+send it with the plan or upload it to the scheduler, where a native leaf reads it. This only
+works for small data, but it costs little, because it uses the same round trip through the driver
+that scalar subqueries already make.
+
+### A shuffle both engines can read and write
+
+Mixing stages freely needs a shuffle that Spark tasks can write and cluster tasks can read, and the
+other way round. Some options:
+
+- **A remote shuffle service.** Comet already integrates with Apache Celeborn. A Celeborn shuffle
+  reader and writer for Ballista would let either engine run any stage.
+- **An object store or shared filesystem.** Comet's native shuffle already writes Arrow IPC. Spark
+  stages would write it to a location the cluster can read, and the scheduler would register those
+  files as a finished stage's output.
+- **Pushing batches over Flight.** Spark tasks would send batches to executors with `DoPut`. This is
+  simpler to deploy, but it ties the lifetimes of Spark tasks and cluster tasks together and makes
+  retrying a failed task difficult.
+
+The hard part is scheduling rather than the data format. Spark's DAG scheduler and Ballista's
+scheduler would have to agree on stage dependencies, failure handling and adaptive replanning.
+Realistically, Spark stays the coordinator, and the cluster runs individual stages on its behalf.
+
+### Running JVM expressions on native executors
+
+Instead of splitting the plan, `comet-executor` could embed a JVM to evaluate the expressions and
+UDFs that have no native implementation, reusing Comet's existing bridges for calling back into
+the JVM. The plan would stay fully native and scheduling would not change. The costs are that every
+executor needs Spark's classpath and enough session state to evaluate the expressions, and that the
+plan must carry those expressions in serialized Spark form.
+
+### Speaking Spark's protocols
+
+Another approach is for `comet-executor` to speak Spark's own protocols, so that Spark tasks and
+native tasks exchange data the way Spark executors do. A Spark executor uses two protocols, and
+they differ in how practical they are to implement.
+
+The control plane runs over Spark's Netty RPC. Its messages, such as `RegisterExecutor`,
+`LaunchTask`, `StatusUpdate` and heartbeats, are Scala case classes in Java serialization. A
+`LaunchTask` carries a serialized `ShuffleMapTask` or `ResultTask`, whose RDD chain and closures
+are JVM code, and the formats change between Spark versions.
+
+The data plane is Spark's block transfer protocol in `network-common` (`OpenBlocks`,
+`FetchShuffleBlocks`, `StreamHandle`, `ChunkFetchRequest` and so on). Its messages have a compact
+binary encoding rather than Java serialization, the protocol has been stable for many releases, and
+the external shuffle service speaks it too. Shuffle data sits in Spark's sort-based layout of a data
+file and an index file per map task. Comet's native shuffle already writes that layout: the data
+comes from Rust and `CometNativeShuffleWriter` commits the index through Spark's
+`IndexShuffleBlockResolver`.
+
+That suggests three options:
+
+- **A fully native Spark executor.** Not practical. Running a Spark task means deserializing Java
+  objects and running JVM closures. Even a task that only wraps a Comet plan would need its plan
+  extracted from a Java-serialized object graph, separately for each Spark version.
+- **A native data plane, with Spark coordinating.** `comet-executor` writes shuffle output in
+  Spark's layout, serves those blocks over the block transfer protocol, and fetches blocks from
+  Spark executors and the external shuffle service. When a native stage finishes, the driver
+  registers its map outputs with Spark's `MapOutputTrackerMaster`, and downstream Spark stages read
+  them as they would any other shuffle. In the other direction, the driver gives a native stage the
+  block locations of the Spark shuffle it reads. This provides a shuffle both engines can read and
+  write without a separate shuffle service. The work includes Spark's authentication and encryption
+  (SASL, or `spark.network.crypto`), which a first version could require to be off, mapping fetch
+  failures onto Spark's `FetchFailed` handling and stage retries, and possibly push-based shuffle
+  later. A native block fetch client would also benefit Comet on Spark, by moving shuffle reads out
+  of the JVM.
+- **An embedded JVM for the control plane.** `comet-executor` hosts a small JVM that runs a cut-down
+  `CoarseGrainedExecutorBackend`. It registers with the driver, deserializes tasks, passes each
+  task's Comet plan to native code, and reports map statuses and accumulators. This works, but the
+  result is a Spark executor with a small heap whose tasks run natively: Comet on Spark with a more
+  native data plane. Spark schedules every task one stage at a time, so the cluster's pipelined
+  execution and lower per-task overhead are lost. An embedded JVM is more useful as an evaluator for
+  expressions that have no native implementation, as described above, where Spark's protocols are
+  not involved.
+
+The data-plane option combines with either way of placing native work. Native executors could join
+a Spark application as executors, which needs the embedded JVM for their control plane, with
+stage-level scheduling sending fully native stages to them through a `ResourceProfile` that asks for
+a resource only native executors offer. Or the driver could keep offloading whole native subgraphs
+to the cluster, which keeps its own scheduling and pipelining, and use Spark-compatible shuffle as
+the boundary between Spark stages and native stages. The second keeps more of what the cluster
+provides.
+
+A first experiment ([#7384](https://github.com/apache/datafusion-comet/issues/7384)) is a Rust client and server for `OpenBlocks` and `FetchShuffleBlocks`, tested
+against a Spark executor and the external shuffle service by reading a shuffle that Comet's native
+shuffle wrote. That shows whether the data plane works independently of any scheduling decision.
+
+### When splitting pays off
+
+Whichever approach is used, the driver should decide whether to split a query based on cost. If
+most of a query's work has to run in Spark, running all of it with Comet on Spark is likely to beat
+sending part of it to the cluster and reading the results back over the network.

docs/source/contributor-guide/release_process.md

@@ -203,7 +203,9 @@ Create a PR against the main branch to prepare for developing the next release:
 
 - Update the Rust crate version to `0.14.0` in `native/Cargo.toml` and in each `contrib/*/native/Cargo.toml`.
   The contrib crates sit outside the `native/` workspace, so they do not inherit its version. Then run
-  `cargo update --workspace` in `native/` and in each contrib crate that has its own `Cargo.lock`.
+  `cargo update --workspace` in `native/`, in `native/native-executors/` (a workspace with its own `Cargo.lock`,
+  which records the versions of the `native/` crates it uses) and in each contrib crate that has its own
+  `Cargo.lock`.
 - Update the Maven version to `0.14.0-SNAPSHOT` in the same set of files listed above (the `pom.xml` files,
   the Spark test diffs under `dev/diffs`, and the Iceberg test diffs under `dev/diffs/iceberg`).