feat: prototype Comet Native Executors built on DataFusion Ballista

.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/release_process.md

@@ -202,7 +202,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`).
 

native/Cargo.toml

@@ -20,7 +20,8 @@ default-members = ["core", "spark-expr", "common", "proto", "jni-bridge", "shuff
 members = ["core", "spark-expr", "common", "proto", "jni-bridge", "shuffle", "operators", "comet-udf-sdk", "comet-test-udfs"]
 # Crates under ../contrib are intentionally NOT workspace members. Core pulls them in as
 # optional path dependencies when their corresponding contrib feature is enabled.
-exclude = ["../contrib"]
+# native-executors is its own workspace, so that this one never depends on Ballista.
+exclude = ["../contrib", "native-executors"]
 resolver = "2"
 
 [workspace.package]

native/core/Cargo.toml

@@ -54,6 +54,9 @@ async-trait = { workspace = true }
 log = "0.4"
 log4rs = "1.4.0"
 prost = "0.14.3"
+# The driver's client of a Comet Native Executor cluster; see the native-executors feature.
+arrow-flight = { version = "59.2.0", optional = true }
+tonic = { version = "0.14", optional = true }
 jni = { workspace = true }
 rand = { workspace = true }
 num = { workspace = true }
@@ -132,6 +135,9 @@ python-udf = ["dep:pyo3", "dep:libc"]
 contrib-lance = ["dep:comet-contrib-lance"]
 hdfs-opendal = ["opendal/services-hdfs", "object_store_opendal", "hdfs-sys"]
 jemalloc = ["tikv-jemallocator", "tikv-jemalloc-ctl"]
+# The JNI client a Spark driver uses to run queries on Comet Native Executors (issue #7288). Off
+# by default while the feature is experimental, so the standard build links no gRPC client.
+native-executors = ["dep:arrow-flight", "dep:tonic"]
 # Native Delta Lake scan support for the JVM-planned path (contrib/delta-spark).
 # In the default set so trying the contrib needs only the jar and the config,
 # not a custom native build. It is inert at runtime unless the contrib jar is

native/core/src/execution/jni_api.rs

@@ -1067,11 +1067,27 @@ fn prepare_datafusion_session_context(
     spark_plan: &Operator,
 ) -> CometResult<SessionContext> {
     let paths = local_dirs.into_iter().map(PathBuf::from).collect();
-    let mut rt_config = RuntimeEnvBuilder::new()
-        .with_disk_manager_builder(plan_disk_manager(paths, max_temp_directory_size))
-        .with_object_store_registry(Arc::new(CometObjectStoreRegistry::default()));
-    rt_config = rt_config.with_memory_pool(memory_pool);
+    let rt_config = comet_runtime_env(memory_pool)
+        .with_disk_manager_builder(plan_disk_manager(paths, max_temp_directory_size));
+    comet_session_context(rt_config, batch_size, task_cpus, spark_config, spark_plan)
+}
+
+/// The runtime of a Comet session that reserves memory from `memory_pool`. It has no disk
+/// manager of its own yet.
+pub(crate) fn comet_runtime_env(memory_pool: Arc<dyn MemoryPool>) -> RuntimeEnvBuilder {
+    RuntimeEnvBuilder::new()
+        .with_object_store_registry(Arc::new(CometObjectStoreRegistry::default()))
+        .with_memory_pool(memory_pool)
+}
 
+/// Configures a DataFusion session for `spark_plan` over the runtime `rt_config` builds.
+pub(crate) fn comet_session_context(
+    rt_config: RuntimeEnvBuilder,
+    batch_size: usize,
+    task_cpus: usize,
+    spark_config: &HashMap<String, String>,
+    spark_plan: &Operator,
+) -> CometResult<SessionContext> {
     let mut session_config = SessionConfig::new()
         .with_target_partitions(task_cpus)
         // This DataFusion context is within the scope of an executing Spark Task. We want to set
@@ -1216,7 +1232,7 @@ fn prepare_output(
 /// # Safety
 /// Each address must point to a writable `FFI_ArrowArray` / `FFI_ArrowSchema` the caller owns.
 /// Whatever the structs hold is overwritten without being released.
-unsafe fn export_batch(
+pub(crate) unsafe fn export_batch(
     array_addrs: &[i64],
     schema_addrs: &[i64],
     output_batch: RecordBatch,

native/core/src/execution/mod.rs

@@ -24,6 +24,7 @@ pub mod expressions;
 pub mod jni_api;
 pub(crate) mod merge_as_partial;
 pub(crate) mod metrics;
+pub mod native_executors;
 pub mod operators;
 pub(crate) mod planner;
 #[cfg(feature = "python-udf")]