feat: add comet-scheduler and comet-executor for Comet Native Executors

.github/dependabot.yml

@@ -53,6 +53,42 @@ updates:
         update-types:
           - "minor"
           - "patch"
+  - package-ecosystem: cargo
+    directory: "/native/native-executors"
+    schedule:
+      interval: weekly
+    target-branch: main
+    labels: [dependencies]
+    ignore:
+      # major version bumps of datafusion*, arrow*, parquet are handled manually
+      - dependency-name: "arrow*"
+        update-types: ["version-update:semver-major"]
+      - dependency-name: "parquet"
+        update-types: ["version-update:semver-major"]
+      - dependency-name: "datafusion*"
+        update-types: ["version-update:semver-major"]
+    groups:
+      proto:
+        applies-to: version-updates
+        patterns:
+          - "prost*"
+          - "pbjson*"
+      # Catch-all: group only minor/patch into a single PR,
+      # excluding deps we want always separate (and excluding arrow/parquet which have their own group)
+      all-other-cargo-deps:
+        applies-to: version-updates
+        patterns:
+          - "*"
+        exclude-patterns:
+          - "arrow*"
+          - "parquet"
+          - "object_store"
+          - "sqlparser"
+          - "prost*"
+          - "pbjson*"
+        update-types:
+          - "minor"
+          - "patch"
   - package-ecosystem: "github-actions"
     directory: "/"
     schedule:

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

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")]