diff --git a/.github/actions/rust-test/action.yaml b/.github/actions/rust-test/action.yaml
index ab5af0b68ca..6a4d430edba 100644
--- a/.github/actions/rust-test/action.yaml
+++ b/.github/actions/rust-test/action.yaml
@@ -45,6 +45,13 @@ runs:
cd native
cargo clippy --color=never -p datafusion-comet --all-targets --features python-udf -- -D warnings
+ # The JNI client for Comet Native Executors is off by default, so nothing above compiles it.
+ - name: Check native executors feature
+ shell: bash
+ run: |
+ cd native
+ cargo clippy --color=never -p datafusion-comet --all-targets --features native-executors -- -D warnings
+
- name: Check compilation
shell: bash
run: |
@@ -106,6 +113,13 @@ runs:
export LD_LIBRARY_PATH=${JAVA_HOME}/lib/server:${LD_LIBRARY_PATH}
RUST_BACKTRACE=1 cargo nextest run -p datafusion-comet --lib --features python-udf python_udf
+ - name: Run native executors client tests
+ shell: bash
+ run: |
+ cd native
+ export LD_LIBRARY_PATH=${JAVA_HOME}/lib/server:${LD_LIBRARY_PATH}
+ RUST_BACKTRACE=1 cargo nextest run -p datafusion-comet --lib --features native-executors native_executors
+
# The steps above lint and test the accounting allocator over the system allocator. Lint and
# test it over jemalloc too, since that is a different allocator backend and nothing else in CI
# builds it.
diff --git a/dev/ci/check-suites.py b/dev/ci/check-suites.py
index 68bf972c5cf..5ced2ba6971 100644
--- a/dev/ci/check-suites.py
+++ b/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
diff --git a/docs/source/_static/images/comet-native-executors.svg b/docs/source/_static/images/comet-native-executors.svg
new file mode 100644
index 00000000000..7b51ab13041
--- /dev/null
+++ b/docs/source/_static/images/comet-native-executors.svg
@@ -0,0 +1,91 @@
+
diff --git a/docs/source/contributor-guide/index.md b/docs/source/contributor-guide/index.md
index 1be327be39e..41a4701426e 100644
--- a/docs/source/contributor-guide/index.md
+++ b/docs/source/contributor-guide/index.md
@@ -51,6 +51,7 @@ Floating-Point Semantics
JVM Shuffle
Native Shuffle
Memory Management
+Comet Native Executors
Iceberg Writes
ANSI Error Propagation
S3 Credential Provider Design
diff --git a/docs/source/contributor-guide/native_executors.md b/docs/source/contributor-guide/native_executors.md
new file mode 100644
index 00000000000..fce25af4f68
--- /dev/null
+++ b/docs/source/contributor-guide/native_executors.md
@@ -0,0 +1,231 @@
+
+
+# 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 graph of stages
+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). Each stage is a
+_block_: a run of Comet operators that Comet serializes as one Rust plan, with the exchanges between
+blocks as the edges of the graph. The design is in
+[issue #7288](https://github.com/apache/datafusion-comet/issues/7288), and
+[`native/native-executors/README.md`](https://github.com/apache/datafusion-comet/blob/main/native/native-executors/README.md)
+describes the cluster, how to build it and how to test it.
+
+This page is for contributors. It explains how to run a query on a cluster, lists what does not work
+yet, and sketches ways to run part of a query in Spark.
+
+
+
+## Running it
+
+1. Build libcomet with the `native-executors` Cargo feature. It is off by default, so the standard
+ build links no gRPC client. A libcomet without it fails the first query with an error that names
+ the feature. For example, `make COMET_FEATURES=native-executors` builds libcomet and the Comet
+ jar that bundles it. The Spark executors that read the results need this libcomet too.
+2. Build and start the cluster. The `comet-scheduler` and `comet-executor` binaries are their own
+ Cargo workspace, so the main Comet build does not depend on Ballista. The cluster and the driver
+ must come from the same Comet build, which the scheduler checks. See
+ [`native/native-executors/README.md`](https://github.com/apache/datafusion-comet/blob/main/native/native-executors/README.md)
+ for how to build the binaries and for the environment variables that configure them.
+3. Point Spark at the scheduler's Flight service and turn the feature on:
+
+ ```shell
+ --conf spark.comet.nativeExecutors.enabled=true \
+ --conf spark.comet.nativeExecutors.scheduler.url=http://scheduler-host:50060
+ ```
+
+ Both configs are internal while the feature is experimental. Comet itself must be enabled as
+ usual.
+
+4. Check that the query ran on the cluster: its physical plan has a `CometNativeExecutorsExec` leaf
+ where the Comet plan would otherwise be. A query that cannot run entirely in the Comet pipeline
+ and on the cluster fails instead, with a message that names the operator and Comet's reasons
+ for it.
+
+`CometNativeExecutorsSuite` runs queries against a cluster it starts from the binaries in
+`COMET_NATIVE_EXECUTORS_BIN_DIR`. Its header comment gives the commands.
+
+## Current limitations
+
+Comet Native Executors are a first version. The sub-issues of
+[#7288](https://github.com/apache/datafusion-comet/issues/7288) track what is missing:
+
+- Whole queries only. See [below](#whole-queries-only).
+- `LIMIT` and `show()`: [#7365](https://github.com/apache/datafusion-comet/issues/7365).
+- Global sorts by floats or nested types sort in a single task rather than in parallel:
+ [#7366](https://github.com/apache/datafusion-comet/issues/7366).
+- Hash partitioning is DataFusion's, not Spark-compatible:
+ [#7368](https://github.com/apache/datafusion-comet/issues/7368).
+- A small broadcast side runs in every task that reads it:
+ [#7369](https://github.com/apache/datafusion-comet/issues/7369).
+- Dynamic partition pruning: [#7370](https://github.com/apache/datafusion-comet/issues/7370). A scan
+ with a dynamic pruning filter is refused.
+- Writes: [#7371](https://github.com/apache/datafusion-comet/issues/7371). Commands and writes stay
+ in Spark.
+- Cancellation: [#7372](https://github.com/apache/datafusion-comet/issues/7372). A killed Spark job
+ leaves its query running on the cluster.
+- Spark UI metrics: [#7373](https://github.com/apache/datafusion-comet/issues/7373).
+- Security. The scheduler and the executors have no authentication and no TLS, and a query names
+ the files and object stores the executors read. Run a cluster only on a trusted network.
+- Result expiry. A query's results stay on the cluster for
+ `COMET_SCHEDULER_RESULT_RETENTION_SECONDS` after it ends (five minutes by default), whether or
+ not Spark has read them. A Spark task that reads a partition after that fails with a message that
+ names the setting.
+
+### Whole queries only
+
+The driver offloads a query only if every operator and expression in it runs in the Comet pipeline
+(`CometNativeExecutors` in `CometNativeExecutorsExec.scala`). Otherwise it refuses the query, naming
+the operator that could not 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. It
+also includes a scan that stays in Spark, such as one whose rows Comet converts to Arrow batches,
+because the cluster cannot read from a 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.** Each Spark task reads one result partition from the executor that holds it over
+ Arrow Flight, one endpoint per partition (`ResultEndpoint`). Data flows from the cluster to Spark,
+ but only at the root of the plan.
+
+## Possible future research: running part of a query in Spark
+
+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 on the cluster. 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 subtree that runs on the cluster as a leaf of the Spark plan
+
+The driver could offload the largest subtrees of Comet operators under an operator that must run in
+Spark, rather than requiring the whole plan to run in the Comet pipeline. 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 offloaded subtree should end.
+
+It suits the common shape where the scans, joins and partial aggregates run in the Comet pipeline,
+and an unsupported operator or UDF sits near the top of the plan.
+
+### Small Spark outputs as inputs to stages on the cluster
+
+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 stage 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 the executors
+
+Instead of splitting the plan, `comet-executor` could embed a JVM to evaluate the expressions and
+UDFs that have no Rust implementation, reusing Comet's existing bridges for calling back into the
+JVM. The whole plan would still run on the cluster 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
+tasks on the cluster 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 Spark executor without a JVM.** 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 data plane in Rust, 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 stage on the cluster 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 stage on
+ the cluster 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 block fetch client in Rust 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 the Rust side, and reports map statuses and accumulators. This works, but the
+ result is a Spark executor with a small heap whose tasks run in Rust: Comet on Spark with a data
+ plane that lives more in Rust. 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 Rust implementation, as described above, where Spark's protocols are
+ not involved.
+
+The data-plane option combines with either way of placing the work. The cluster's executors could
+join a Spark application as executors, which needs the embedded JVM for their control plane, with
+stage-level scheduling sending the stages that run entirely in the Comet pipeline to them through a
+`ResourceProfile` that asks for a resource only they offer. Or the driver could keep offloading
+whole subgraphs of stages to the cluster, which keeps its own scheduling and pipelining, and use a
+Spark-compatible shuffle as the boundary between Spark stages and stages on the cluster. 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.
diff --git a/native/Cargo.lock b/native/Cargo.lock
index eb93db1252b..20c9d19861b 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -347,6 +347,27 @@ dependencies = [
"num-traits",
]
+[[package]]
+name = "arrow-flight"
+version = "59.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b543210778ac15d5f7a10153dc39e099f1eee756363589a842029e6c5c76e423"
+dependencies = [
+ "arrow-array",
+ "arrow-buffer",
+ "arrow-cast",
+ "arrow-data",
+ "arrow-ipc",
+ "arrow-schema",
+ "base64 0.23.1",
+ "bytes",
+ "futures",
+ "prost",
+ "prost-types",
+ "tonic",
+ "tonic-prost",
+]
+
[[package]]
name = "arrow-ipc"
version = "59.3.2"
@@ -1033,6 +1054,49 @@ dependencies = [
"tracing",
]
+[[package]]
+name = "axum"
+version = "0.8.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90"
+dependencies = [
+ "axum-core",
+ "bytes",
+ "futures-util",
+ "http",
+ "http-body",
+ "http-body-util",
+ "itoa",
+ "matchit",
+ "memchr",
+ "mime",
+ "percent-encoding",
+ "pin-project-lite",
+ "serde_core",
+ "sync_wrapper",
+ "tower",
+ "tower-layer",
+ "tower-service",
+]
+
+[[package]]
+name = "axum-core"
+version = "0.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1"
+dependencies = [
+ "bytes",
+ "futures-core",
+ "http",
+ "http-body",
+ "http-body-util",
+ "mime",
+ "pin-project-lite",
+ "sync_wrapper",
+ "tower-layer",
+ "tower-service",
+]
+
[[package]]
name = "backon"
version = "1.6.0"
@@ -1979,6 +2043,7 @@ name = "datafusion-comet"
version = "1.3.0"
dependencies = [
"arrow",
+ "arrow-flight",
"assertables",
"async-trait",
"aws-config",
@@ -2044,6 +2109,7 @@ dependencies = [
"tikv-jemalloc-ctl",
"tikv-jemallocator",
"tokio",
+ "tonic",
"typetag",
"url",
"uuid",
@@ -3465,6 +3531,12 @@ version = "1.10.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87"
+[[package]]
+name = "httpdate"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9"
+
[[package]]
name = "humantime"
version = "2.4.0"
@@ -3494,6 +3566,7 @@ dependencies = [
"http",
"http-body",
"httparse",
+ "httpdate",
"itoa",
"pin-project-lite",
"smallvec",
@@ -3517,6 +3590,19 @@ dependencies = [
"tower-service",
]
+[[package]]
+name = "hyper-timeout"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0"
+dependencies = [
+ "hyper",
+ "hyper-util",
+ "pin-project-lite",
+ "tokio",
+ "tower-service",
+]
+
[[package]]
name = "hyper-util"
version = "0.1.21"
@@ -4241,6 +4327,12 @@ dependencies = [
"twox-hash",
]
+[[package]]
+name = "matchit"
+version = "0.8.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3"
+
[[package]]
name = "md-5"
version = "0.10.6"
@@ -4295,6 +4387,12 @@ dependencies = [
"libmimalloc-sys",
]
+[[package]]
+name = "mime"
+version = "0.3.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a"
+
[[package]]
name = "miniz_oxide"
version = "0.8.9"
@@ -6883,6 +6981,47 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "tonic"
+version = "0.14.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd1fa9b1a51188f3266a823208078315811269d06d272c3a61be0b9a194194ff"
+dependencies = [
+ "async-trait",
+ "axum",
+ "base64 0.22.1",
+ "bytes",
+ "h2",
+ "http",
+ "http-body",
+ "http-body-util",
+ "hyper",
+ "hyper-timeout",
+ "hyper-util",
+ "percent-encoding",
+ "pin-project",
+ "socket2",
+ "sync_wrapper",
+ "tokio",
+ "tokio-stream",
+ "tokio-util",
+ "tower",
+ "tower-layer",
+ "tower-service",
+ "tracing",
+]
+
+[[package]]
+name = "tonic-prost"
+version = "0.14.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "f63c616ffaa648f7c5dd5657e97d9a285fcb44ff3b70da761d436f5a13ff7be6"
+dependencies = [
+ "bytes",
+ "prost",
+ "tonic",
+]
+
[[package]]
name = "tower"
version = "0.5.3"
@@ -6891,11 +7030,15 @@ checksum = "ebe5ef63511595f1344e2d5cfa636d973292adc0eec1f0ad45fae9f0851ab1d4"
dependencies = [
"futures-core",
"futures-util",
+ "indexmap 2.14.2",
"pin-project-lite",
+ "slab",
"sync_wrapper",
"tokio",
+ "tokio-util",
"tower-layer",
"tower-service",
+ "tracing",
]
[[package]]
diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml
index 4ba78cdfa43..cdff06aee94 100644
--- a/native/core/Cargo.toml
+++ b/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 }
@@ -133,6 +136,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
diff --git a/native/core/src/execution/jni_api.rs b/native/core/src/execution/jni_api.rs
index a01789dead3..91ad8a8ef68 100644
--- a/native/core/src/execution/jni_api.rs
+++ b/native/core/src/execution/jni_api.rs
@@ -1058,6 +1058,9 @@ fn configure_skip_partial_aggregation(config: &mut SessionConfig, plan: &Operato
}
}
+/// Set in the configuration of a Comet Native Executors fragment's session.
+pub(crate) const NATIVE_EXECUTORS_FRAGMENT: &str = "spark.comet.nativeExecutors.fragment";
+
/// Configure DataFusion session context.
fn prepare_datafusion_session_context(
batch_size: usize,
@@ -1152,9 +1155,25 @@ pub(crate) fn comet_session_context(
crate::execution::c_udf::policy::NativeUdfPolicy::from_spark_config(spark_config),
));
+ // A Comet Native Executors fragment has no ShuffleWriter root, but its output goes to the
+ // cluster's shuffle all the same, so it is checked as if it had one.
+ let fragment_root;
+ let gate_plan = if spark_config
+ .get(NATIVE_EXECUTORS_FRAGMENT)
+ .is_some_and(|value| value == "true")
+ {
+ fragment_root = Operator {
+ op_struct: Some(OpStruct::ShuffleWriter(Default::default())),
+ children: vec![spark_plan.clone()],
+ ..Default::default()
+ };
+ &fragment_root
+ } else {
+ spark_plan
+ };
configure_skip_partial_aggregation(
&mut session_config,
- spark_plan,
+ gate_plan,
spark_config.get_bool(COMET_EXEC_AGGREGATE_SKIP_PARTIAL_ENABLED),
);
@@ -1234,7 +1253,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,
diff --git a/native/core/src/execution/native_executors/client.rs b/native/core/src/execution/native_executors/client.rs
new file mode 100644
index 00000000000..7d3843acef3
--- /dev/null
+++ b/native/core/src/execution/native_executors/client.rs
@@ -0,0 +1,271 @@
+// 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.
+
+//! The JNI client a Spark driver uses to run queries on a Comet Native Executor cluster: the
+//! native methods of `org.apache.comet.NativeExecutorsClient`.
+//!
+//! The driver submits a query to the scheduler's Arrow Flight service, which answers with one
+//! endpoint per result partition, and then reads each partition from its executor with `DoGet`.
+
+use std::time::Duration;
+
+use arrow_flight::decode::FlightRecordBatchStream;
+use arrow_flight::error::FlightError;
+use arrow_flight::flight_service_client::FlightServiceClient;
+use arrow_flight::{FlightClient, FlightDescriptor, Ticket};
+use datafusion_comet_proto::spark_native_executors::{
+ QueryRequest, ResultEndpoint, ResultEndpoints,
+};
+use datafusion_comet_proto::PROTO_FINGERPRINT;
+use futures::StreamExt;
+use jni::objects::{JByteArray, JClass, JLongArray, JString, ReleaseMode};
+use jni::sys::{jbyteArray, jlong};
+use jni::EnvUnowned;
+use prost::Message;
+use tonic::transport::{Channel, Endpoint};
+use tonic::Code;
+
+use super::COMET_VERSION;
+use crate::errors::{try_unwrap_or_throw, CometError, CometResult};
+use crate::execution::jni_api::{export_batch, get_runtime};
+
+fn client_error(context: &str, error: impl std::fmt::Display) -> CometError {
+ CometError::Internal(format!("Comet Native Executors: {context}: {error}"))
+}
+
+/// How long to wait for a connection to a scheduler or an executor.
+const CONNECT_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// How often to probe a connection that carries no traffic, at the TCP level and at the HTTP/2
+/// level. A query can run for a long time before the scheduler answers its submission, so a
+/// request has no overall timeout. These probes are what notice a scheduler or an executor that
+/// died or became unreachable without closing the connection.
+const KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(30);
+
+/// How long to wait for an answer to an HTTP/2 probe before giving up on the connection.
+const KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(20);
+
+/// The error for a failed read of a result partition. A partition the cluster has deleted gets a
+/// message that names the setting to change. The scheduler says that a partition it stores has
+/// expired, and an executor fails to open the file of a partition it has deleted.
+fn read_error(context: &str, error: FlightError) -> CometError {
+ let expired = match &error {
+ FlightError::Tonic(status) => match status.code() {
+ Code::NotFound => status.message().contains("expired"),
+ Code::Internal => {
+ let message = status.message();
+ message.contains("Failed to open partition file") && message.contains("NotFound")
+ }
+ _ => false,
+ },
+ _ => false,
+ };
+ if expired {
+ client_error(
+ context,
+ format!(
+ "the result partition expired on the cluster; raise \
+ COMET_SCHEDULER_RESULT_RETENTION_SECONDS ({error})"
+ ),
+ )
+ } else {
+ client_error(context, error)
+ }
+}
+
+/// Connects to the Flight service at `location`, a Flight location such as
+/// `grpc+tcp://host:port` or an `http://` URL.
+async fn connect(location: &str) -> CometResult {
+ let uri = match location.strip_prefix("grpc+tcp://") {
+ Some(address) => format!("http://{address}"),
+ None => location.to_string(),
+ };
+ let channel: Channel = Endpoint::from_shared(uri)
+ .map_err(|e| client_error(&format!("invalid location {location}"), e))?
+ .connect_timeout(CONNECT_TIMEOUT)
+ .tcp_keepalive(Some(KEEP_ALIVE_INTERVAL))
+ .http2_keep_alive_interval(KEEP_ALIVE_INTERVAL)
+ .keep_alive_timeout(KEEP_ALIVE_TIMEOUT)
+ .keep_alive_while_idle(true)
+ .connect()
+ .await
+ .map_err(|e| client_error(&format!("cannot connect to {location}"), e))?;
+ // Query plans and result batches can be larger than gRPC's default 4 MiB message limit.
+ let inner = FlightServiceClient::new(channel)
+ .max_decoding_message_size(usize::MAX)
+ .max_encoding_message_size(usize::MAX);
+ Ok(FlightClient::new_from_inner(inner))
+}
+
+/// Submits `query`, a serialized `QueryRequest`, to the scheduler at `scheduler_url` and waits
+/// for it to finish. Returns where its result partitions are.
+async fn submit(scheduler_url: &str, mut query: Vec) -> CometResult {
+ // Stamp the query with this build's version and plan format. A field that appears again
+ // replaces the first, so appending them spares decoding and encoding the whole query.
+ QueryRequest {
+ comet_version: COMET_VERSION.to_string(),
+ proto_fingerprint: PROTO_FINGERPRINT.to_string(),
+ ..Default::default()
+ }
+ .encode(&mut query)
+ .map_err(|e| client_error("bad query", e))?;
+ let mut client = connect(scheduler_url).await?;
+ let info = client
+ .get_flight_info(FlightDescriptor::new_cmd(query))
+ .await
+ .map_err(|e| client_error("the query failed", e))?;
+ let endpoints = info
+ .endpoint
+ .into_iter()
+ .map(|endpoint| {
+ // A partition without a location is served by the scheduler itself.
+ let location = endpoint.location.first().map_or_else(
+ || scheduler_url.to_string(),
+ |location| location.uri.clone(),
+ );
+ let ticket = endpoint
+ .ticket
+ .map(|ticket| ticket.ticket.to_vec())
+ .ok_or_else(|| client_error("bad result", "a partition has no ticket"))?;
+ Ok(ResultEndpoint { location, ticket })
+ })
+ .collect::>>()?;
+ Ok(ResultEndpoints { endpoints })
+}
+
+/// Runs a query: `query` is a serialized `QueryRequest`, which this fills in with the driver's
+/// Comet version and plan format. Returns a serialized `ResultEndpoints`.
+#[no_mangle]
+pub extern "system" fn Java_org_apache_comet_NativeExecutorsClient_submit(
+ e: EnvUnowned,
+ _class: JClass,
+ scheduler_url: JString,
+ query: JByteArray,
+) -> jbyteArray {
+ try_unwrap_or_throw(&e, |env| {
+ let scheduler_url = scheduler_url
+ .try_to_string(env)
+ .map_err(|e| client_error("bad scheduler URL", e))?;
+ let query = env.convert_byte_array(query)?;
+ let endpoints = get_runtime().block_on(submit(&scheduler_url, query))?;
+ Ok(env
+ .byte_array_from_slice(&endpoints.encode_to_vec())?
+ .into_raw())
+ })
+}
+
+/// Starts reading a result partition from the executor at `location`. Returns a handle for
+/// `nextBatch` and `closePartition`.
+#[no_mangle]
+pub extern "system" fn Java_org_apache_comet_NativeExecutorsClient_openPartition(
+ e: EnvUnowned,
+ _class: JClass,
+ location: JString,
+ ticket: JByteArray,
+) -> jlong {
+ try_unwrap_or_throw(&e, |env| {
+ let location = location
+ .try_to_string(env)
+ .map_err(|e| client_error("bad location", e))?;
+ let ticket = env.convert_byte_array(ticket)?;
+ let stream = get_runtime().block_on(async {
+ connect(&location)
+ .await?
+ .do_get(Ticket::new(ticket))
+ .await
+ .map_err(|e| read_error(&format!("cannot read a partition from {location}"), e))
+ })?;
+ Ok(Box::into_raw(Box::new(stream)) as jlong)
+ })
+}
+
+/// Moves the partition's next batch into the Arrow C Data Interface structs at `array_addrs`
+/// and `schema_addrs`, one pair per column. Returns its row count, or -1 at the end.
+#[no_mangle]
+pub extern "system" fn Java_org_apache_comet_NativeExecutorsClient_nextBatch(
+ e: EnvUnowned,
+ _class: JClass,
+ handle: jlong,
+ array_addrs: JLongArray,
+ schema_addrs: JLongArray,
+) -> jlong {
+ try_unwrap_or_throw(&e, |env| {
+ // SAFETY: `handle` comes from `openPartition` and hasn't been closed.
+ let stream = unsafe { &mut *(handle as *mut FlightRecordBatchStream) };
+ let Some(batch) = get_runtime().block_on(stream.next()) else {
+ return Ok(-1);
+ };
+ let batch = batch.map_err(|e| read_error("cannot read a partition", e))?;
+ let array_addrs = unsafe { array_addrs.get_elements(env, ReleaseMode::NoCopyBack)? };
+ let schema_addrs = unsafe { schema_addrs.get_elements(env, ReleaseMode::NoCopyBack)? };
+ // SAFETY: the JVM allocated one array and one schema struct per column.
+ unsafe { export_batch(&array_addrs, &schema_addrs, batch, false) }
+ })
+}
+
+/// Stops reading a partition and releases its handle.
+#[no_mangle]
+pub extern "system" fn Java_org_apache_comet_NativeExecutorsClient_closePartition(
+ e: EnvUnowned,
+ _class: JClass,
+ handle: jlong,
+) {
+ try_unwrap_or_throw(&e, |_env| {
+ if handle != 0 {
+ // SAFETY: `handle` comes from `openPartition`, and the JVM closes it once.
+ drop(unsafe { Box::from_raw(handle as *mut FlightRecordBatchStream) });
+ }
+ Ok(())
+ })
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use tonic::Status;
+
+ fn message(status: Status) -> String {
+ read_error("cannot read a partition", FlightError::from(status)).to_string()
+ }
+
+ #[test]
+ fn an_expired_result_names_the_setting_to_raise() {
+ for status in [
+ Status::not_found("The results of job abc have expired"),
+ Status::internal(
+ "Ballista Error: General(\"Failed to open partition file at \\\"p\\\": \
+ Os { code: 2, kind: NotFound, message: \\\"No such file or directory\\\" }\")",
+ ),
+ ] {
+ assert!(
+ message(status).contains("COMET_SCHEDULER_RESULT_RETENTION_SECONDS"),
+ "no hint"
+ );
+ }
+ }
+
+ #[test]
+ fn other_failures_do_not_blame_expiry() {
+ for status in [
+ Status::not_found("Job abc has no result partition 7"),
+ Status::internal("Ballista Error: something else"),
+ Status::unavailable("connection reset"),
+ ] {
+ assert!(!message(status).contains("expired"));
+ }
+ }
+}
diff --git a/native/core/src/execution/native_executors/fragment.rs b/native/core/src/execution/native_executors/fragment.rs
index 4e4aec790e0..3e542265e4d 100644
--- a/native/core/src/execution/native_executors/fragment.rs
+++ b/native/core/src/execution/native_executors/fragment.rs
@@ -22,8 +22,9 @@ use std::sync::Arc;
use arrow::array::{Array, AsArray, BinaryArray, RecordBatch, RecordBatchOptions, UInt32Array};
use arrow::datatypes::{DataType, Field, Schema, SchemaRef, UInt32Type};
use datafusion::common::tree_node::TreeNodeRecursion;
-use datafusion::common::{internal_datafusion_err, DataFusionError, Result};
+use datafusion::common::{internal_datafusion_err, plan_datafusion_err, DataFusionError, Result};
use datafusion::datasource::memory::MemorySourceConfig;
+use datafusion::datasource::source::DataSourceExec;
use datafusion::execution::disk_manager::DiskManager;
use datafusion::execution::memory_pool::{MemoryPool, UnboundedMemoryPool};
use datafusion::execution::TaskContext;
@@ -41,7 +42,9 @@ use datafusion_comet_proto::spark_operator::{Operator, SparkStructField};
use futures::stream::BoxStream;
use futures::{StreamExt, TryStreamExt};
-use crate::execution::jni_api::{comet_runtime_env, comet_session_context};
+use crate::execution::jni_api::{
+ comet_runtime_env, comet_session_context, NATIVE_EXECUTORS_FRAGMENT,
+};
use crate::execution::planner::{
align_shuffle_writer_input, convert_spark_types_to_arrow_schema, PhysicalPlanner,
};
@@ -52,6 +55,14 @@ use crate::execution::spark_config::{
const DEFAULT_BATCH_SIZE: usize = 8192;
+/// The most partitions a stage or an exchange may have. A request names the counts, and the
+/// scheduler allocates for each partition, so a request can't ask for more than this. Spark's
+/// own default is 200 shuffle partitions, and large queries use a few thousand.
+pub const MAX_PARTITIONS: usize = 1 << 16;
+
+/// The largest `spark.comet.batchSize` a request may set. Each operator sizes its buffers by it.
+const MAX_BATCH_SIZE: usize = 1 << 20;
+
fn descriptor_schema() -> SchemaRef {
Arc::new(Schema::new(vec![
Field::new("partition", DataType::UInt32, false),
@@ -71,6 +82,11 @@ pub fn partition_descriptor(
num_partitions: usize,
plans: &[Vec],
) -> Result> {
+ if num_partitions == 0 || num_partitions > MAX_PARTITIONS {
+ return Err(plan_datafusion_err!(
+ "A stage must have between 1 and {MAX_PARTITIONS} partitions, not {num_partitions}"
+ ));
+ }
let schema = descriptor_schema();
let partitions = (0..num_partitions)
.map(|index| {
@@ -88,6 +104,29 @@ pub fn partition_descriptor(
Ok(MemorySourceConfig::try_new_exec(&partitions, schema, None)?)
}
+/// The partitions' own copies of the block in a fragment's partition `descriptor`, one per
+/// partition, or none when its partitions all run the fragment's block.
+pub fn partition_plans(descriptor: &Arc) -> Result>> {
+ let memory = descriptor
+ .downcast_ref::()
+ .and_then(|source| source.data_source().downcast_ref::())
+ .ok_or_else(|| internal_datafusion_err!("Not a fragment's partition descriptor"))?;
+ let mut plans = Vec::new();
+ for partition in memory.partitions() {
+ let [batch] = partition.as_slice() else {
+ return Err(internal_datafusion_err!(
+ "A fragment's partition descriptor must have one row per partition"
+ ));
+ };
+ let column = batch.column(1).as_binary::();
+ if column.is_null(0) {
+ return Ok(vec![]);
+ }
+ plans.push(column.value(0).to_vec());
+ }
+ Ok(plans)
+}
+
/// Runs a Comet native block for each partition of a stage.
///
/// The first child is the partition descriptor (see [`partition_descriptor`]), and each other
@@ -126,15 +165,6 @@ impl CometFragmentExec {
));
}
let partitions = descriptor.output_partitioning().partition_count();
- for (index, (input, broadcast)) in inputs.iter().zip(&fragment.broadcast).enumerate() {
- let expected = if *broadcast { 1 } else { partitions };
- let actual = input.output_partitioning().partition_count();
- if actual != expected {
- return Err(internal_datafusion_err!(
- "CometFragmentExec input {index} has {actual} partitions, expected {expected}"
- ));
- }
- }
let props = Arc::new(PlanProperties::new(
EquivalenceProperties::new(fragment_schema(&fragment.output_schema)),
Partitioning::UnknownPartitioning(partitions),
@@ -151,13 +181,41 @@ impl CometFragmentExec {
pub fn fragment(&self) -> &Arc {
&self.fragment
}
+
+ /// Checks that each input's partitions line up with the descriptor's. This is checked when the
+ /// plan runs rather than when it is built, since an optimizer can rebuild the node with an
+ /// input whose exchange it has taken away and is about to put back.
+ fn check_inputs(&self) -> Result<()> {
+ let partitions = self.props.output_partitioning().partition_count();
+ if self.children.len() != self.fragment.broadcast.len() + 1 {
+ return Err(internal_datafusion_err!(
+ "CometFragmentExec has {} inputs but {} broadcast flags",
+ self.children.len() - 1,
+ self.fragment.broadcast.len()
+ ));
+ }
+ for (index, (input, broadcast)) in self.children[1..]
+ .iter()
+ .zip(&self.fragment.broadcast)
+ .enumerate()
+ {
+ let expected = if *broadcast { 1 } else { partitions };
+ let actual = input.output_partitioning().partition_count();
+ if actual != expected {
+ return Err(internal_datafusion_err!(
+ "CometFragmentExec input {index} has {actual} partitions, expected {expected}"
+ ));
+ }
+ }
+ Ok(())
+ }
}
/// The schema of a fragment's output: the types Spark declared for it, every column nullable.
/// The cast to those types keeps a column nullable wherever the block's output is, even where
/// Spark knows it has no nulls, such as a join key a filter has checked, so a fragment can't
/// promise more. Nothing between stages relies on a column having no nulls.
-fn fragment_schema(output_schema: &[SparkStructField]) -> SchemaRef {
+pub fn fragment_schema(output_schema: &[SparkStructField]) -> SchemaRef {
let declared = convert_spark_types_to_arrow_schema(output_schema);
let fields: Vec = declared
.fields()
@@ -193,10 +251,13 @@ impl ExecutionPlan for CometFragmentExec {
/// The descriptor's and the partitioned inputs' partitions line up with this plan's, while a
/// broadcast input is read whole by every partition, so a distributed engine must not split
- /// it between tasks.
+ /// it between tasks. A stage with one partition reads every input as one partition, which
+ /// keeps an optimizer such as `EnsureRequirements` from removing the coalesce that gathers
+ /// it.
fn input_distribution_requirements(&self) -> InputDistributionRequirements {
+ let single = self.props.output_partitioning().partition_count() == 1;
let inputs = self.fragment.broadcast.iter().map(|broadcast| {
- if *broadcast {
+ if *broadcast || single {
Distribution::SinglePartition
} else {
Distribution::UnspecifiedDistribution
@@ -237,6 +298,7 @@ impl ExecutionPlan for CometFragmentExec {
partition: usize,
context: Arc,
) -> Result {
+ self.check_inputs()?;
let descriptor = Arc::clone(&self.children[0]);
let inputs = self.children[1..]
.iter()
@@ -350,14 +412,16 @@ fn fragment_session(
) -> Result> {
let mut runtime = comet_runtime_env(memory_pool);
runtime.disk_manager = disk_manager;
- let session = comet_session_context(
- runtime,
- conf.get_usize(COMET_BATCH_SIZE, DEFAULT_BATCH_SIZE),
- 1,
- conf,
- spark_plan,
- )
- .map_err(|e| DataFusionError::External(Box::new(e)))?;
+ let mut conf = conf.clone();
+ conf.insert(NATIVE_EXECUTORS_FRAGMENT.to_string(), "true".to_string());
+ let batch_size = conf.get_usize(COMET_BATCH_SIZE, DEFAULT_BATCH_SIZE);
+ if batch_size == 0 || batch_size > MAX_BATCH_SIZE {
+ return Err(plan_datafusion_err!(
+ "{COMET_BATCH_SIZE} must be between 1 and {MAX_BATCH_SIZE}, not {batch_size}"
+ ));
+ }
+ let session = comet_session_context(runtime, batch_size, 1, &conf, spark_plan)
+ .map_err(|e| DataFusionError::External(Box::new(e)))?;
Ok(Arc::new(session))
}
diff --git a/native/core/src/execution/native_executors/mod.rs b/native/core/src/execution/native_executors/mod.rs
index fc0b9d6b2f4..7e079be7298 100644
--- a/native/core/src/execution/native_executors/mod.rs
+++ b/native/core/src/execution/native_executors/mod.rs
@@ -24,9 +24,14 @@
//! partition of a stage, reading its inputs from other DataFusion plans, such as a distributed
//! engine's shuffle readers, instead of from the JVM.
+#[cfg(feature = "native-executors")]
+mod client;
mod fragment;
-pub use fragment::{block_schema, partition_descriptor, CometFragmentExec};
+pub use fragment::{
+ block_schema, fragment_schema, partition_descriptor, partition_plans, CometFragmentExec,
+ MAX_PARTITIONS,
+};
/// The version of this Comet build. A Comet Native Executor cluster requires a driver's to match
/// its own, together with [`datafusion_comet_proto::PROTO_FINGERPRINT`].
diff --git a/native/native-executors/Cargo.lock b/native/native-executors/Cargo.lock
index 47a671f95bd..c20f0300da1 100644
--- a/native/native-executors/Cargo.lock
+++ b/native/native-executors/Cargo.lock
@@ -999,8 +999,7 @@ dependencies = [
[[package]]
name = "ballista-api-types"
version = "55.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b97d9eea12c759eda5f828b4f0eba2b42784d96a4e7858afe5624a9522507841"
+source = "git+https://github.com/apache/datafusion-ballista?branch=comet-native#0f092ee494d47d611f6e83f3660e60a4e4038d8c"
dependencies = [
"serde",
"utoipa",
@@ -1009,8 +1008,7 @@ dependencies = [
[[package]]
name = "ballista-core"
version = "55.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5a090511fa9b3661dbf17444ecbbf5efcd11793e140ddd7fe2fb85f2de88f561"
+source = "git+https://github.com/apache/datafusion-ballista?branch=comet-native#0f092ee494d47d611f6e83f3660e60a4e4038d8c"
dependencies = [
"arrow-flight",
"async-trait",
@@ -1040,8 +1038,7 @@ dependencies = [
[[package]]
name = "ballista-executor"
version = "55.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "403cdac6a4e822e6025348f4f7e43765ba0ca176ca1f225888a9eeb90b9c66ce"
+source = "git+https://github.com/apache/datafusion-ballista?branch=comet-native#0f092ee494d47d611f6e83f3660e60a4e4038d8c"
dependencies = [
"arrow",
"arrow-flight",
@@ -1070,8 +1067,7 @@ dependencies = [
[[package]]
name = "ballista-history"
version = "55.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "926ed4783d3b4d24fc28e899c7aef2f0127a511238320eadfc06c65ba2772bf8"
+source = "git+https://github.com/apache/datafusion-ballista?branch=comet-native#0f092ee494d47d611f6e83f3660e60a4e4038d8c"
dependencies = [
"ballista-api-types",
"log",
@@ -1083,8 +1079,7 @@ dependencies = [
[[package]]
name = "ballista-scheduler"
version = "55.0.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "845bb1c2c763ae2b50ecfd9debdfe8e092d13a4fefe84c20e9262bb191a1f1e7"
+source = "git+https://github.com/apache/datafusion-ballista?branch=comet-native#0f092ee494d47d611f6e83f3660e60a4e4038d8c"
dependencies = [
"arrow-flight",
"async-trait",
@@ -4110,9 +4105,9 @@ dependencies = [
[[package]]
name = "lru"
-version = "0.18.5"
+version = "0.19.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ef9ac18847474e638e3702b76c65d4eb93428471a74778ef0f1be711717f89b5"
+checksum = "4a9f71e3591c7ec44d3e563e87bf14ddc2bada8519dca046e5b0e560cd004e2d"
dependencies = [
"hashbrown 0.17.1",
]
@@ -5105,8 +5100,8 @@ version = "0.14.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042"
dependencies = [
- "heck 0.5.0",
- "itertools 0.14.0",
+ "heck 0.4.1",
+ "itertools 0.13.0",
"log",
"multimap",
"petgraph",
@@ -5127,7 +5122,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
dependencies = [
"anyhow",
- "itertools 0.14.0",
+ "itertools 0.13.0",
"proc-macro2",
"quote",
"syn 2.0.119",
diff --git a/native/native-executors/Cargo.toml b/native/native-executors/Cargo.toml
index 388c06189aa..c298f5c9f08 100644
--- a/native/native-executors/Cargo.toml
+++ b/native/native-executors/Cargo.toml
@@ -19,8 +19,10 @@
# DataFusion Ballista. See https://github.com/apache/datafusion-comet/issues/7288.
#
# This is its own workspace, with its own lockfile, so that the main Comet build never depends
-# on Ballista. It builds Comet's native crates by path together with the Ballista release on the
-# same DataFusion major version.
+# on Ballista. It builds Comet's native crates by path together with Ballista's `comet-native`
+# branch, which is Ballista's main branch with the enhancements Comet Native Executors need that
+# haven't merged yet (apache/datafusion-ballista#2682, #2721, #2722, #2723 and #2743), on the same
+# DataFusion major version.
[workspace]
[package]
@@ -35,11 +37,11 @@ publish = false
[dependencies]
arrow-flight = "59.2.0"
-ballista-core = "55.0.0"
-ballista-executor = { version = "55.0.0", default-features = false, features = ["arrow-ipc-optimizations"] }
-# `rest-api` compiles the scheduler's event-log writer (COMET_SCHEDULER_EVENT_LOG_DIR) and its REST
-# API. In Ballista 55.0.0 it logs through `tracing` without enabling it, so that is enabled here.
-ballista-scheduler = { version = "55.0.0", default-features = false, features = ["rest-api", "tracing"] }
+ballista-core = { git = "https://github.com/apache/datafusion-ballista", branch = "comet-native" }
+ballista-executor = { git = "https://github.com/apache/datafusion-ballista", branch = "comet-native", default-features = false, features = ["arrow-ipc-optimizations"] }
+# `rest-api` compiles the scheduler's event-log writer (COMET_SCHEDULER_EVENT_LOG_DIR) and its REST API.
+# Its code calls `tracing`, which Ballista links only for its binaries (`build-binary`), so it is added here.
+ballista-scheduler = { git = "https://github.com/apache/datafusion-ballista", branch = "comet-native", default-features = false, features = ["rest-api", "tracing"] }
datafusion = { version = "55.2.0", default-features = false }
# Comet's core crate, under the name of its library. Without the default hdfs-opendal feature,
# which links libjvm through libhdfs.
diff --git a/native/native-executors/README.md b/native/native-executors/README.md
index a271ba863f9..16496ed89d1 100644
--- a/native/native-executors/README.md
+++ b/native/native-executors/README.md
@@ -25,9 +25,14 @@ Spark driver plans each query with Catalyst and Comet, then submits it to a clus
[Apache DataFusion Ballista](https://github.com/apache/datafusion-ballista). The design is in
[issue #7288](https://github.com/apache/datafusion-comet/issues/7288).
-This directory holds the cluster. The Spark driver side is in progress in
-[#7289](https://github.com/apache/datafusion-comet/pull/7289). It is its own Cargo workspace, so the
-main Comet build never depends on Ballista.
+This directory holds the cluster, which is its own Cargo workspace, so the main Comet build never
+depends on Ballista. The other half is in Comet's Spark plugin: with
+`spark.comet.nativeExecutors.enabled` set, the driver submits each query it can offload to the
+scheduler's Flight service (`spark.comet.nativeExecutors.scheduler.url`) and reads the results back.
+The two must come from the same Comet build, which the scheduler checks.
+
+**Security.** The scheduler and the executors have no authentication and no TLS, and a query names
+the files and object stores the executors read. Run a cluster only on a trusted network.
## The scheduler's Flight service
@@ -39,34 +44,78 @@ A client runs a query through the scheduler's Arrow Flight service:
from another Comet build and plans each block with Comet's planner to check it. That builds no
object store for the block's files, so the scheduler uses no storage credentials. It then
builds a Ballista physical plan (`src/assembly.rs`), runs it on the Ballista scheduler it
- wraps, and answers with one endpoint per result partition.
+ wraps, and answers with one endpoint per result partition. It refuses a request with more
+ partitions or a larger batch size than a cluster plausibly needs, a stage whose block reads a
+ different number of inputs than the stage declares, and a block that writes files.
2. `DoGet` with each endpoint's ticket, at the endpoint's location: the executor that holds the
partition. A query's results stay there for `COMET_SCHEDULER_RESULT_RETENTION_SECONDS` after
the query ends (five minutes by default), whether or not a client has read them, and each
endpoint's expiration time says until when.
+Ballista runs a query in one of two modes, set for every query with
+`COMET_SCHEDULER_BALLISTA_CONFIG` (`ballista.execution.mode`, blocking by default). In pipelined
+mode the executors stream a result once and cannot hold it, so the scheduler reads each query's
+result into its own memory and serves it from its own Flight service. Its endpoints have no
+location: read them from the scheduler. The scheduler buffers at most
+`COMET_SCHEDULER_PIPELINED_RESULT_MAX_BYTES` for a query and
+`COMET_SCHEDULER_PIPELINED_STORE_MAX_BYTES` for all queries, and fails a query whose result is
+larger than the first. When a new result does not fit in the second, the scheduler drops the oldest
+results, which a client then finds expired. With a retention of `0` that is the only thing that
+drops them. Run a query with a large result in blocking mode.
+
+The scheduler cancels a job whose client disconnects, whose result it cannot read, or that runs
+longer than `COMET_SCHEDULER_QUERY_TIMEOUT_SECONDS`.
+
A stage that several other stages read runs once for each of them, since Ballista's planner
doesn't share a stage between consumers yet.
+Spark plans a global sort as a range exchange followed by a sort of each partition, and its
+driver samples the exchange's input for the ranges' bounds. A cluster doesn't range-partition:
+the stage reads the producer's partitions as they are, sorts each, and merges the sorted
+partitions in the order the driver sends with the stage, as DataFusion plans a global sort. A
+stage whose order isn't over columns of types Arrow orders as Spark does, such as floats, reads
+the exchange as one partition and sorts it in a single task.
+
Each stage runs on the executors as a `CometFragmentExec`
(`native/core/src/execution/native_executors`), which plans the stage's block with Comet's planner
for each partition. Ballista moves the data between stages:
-| Exchange | Ballista plan |
-| --------------------------- | --------------------------------------------------------------------------------------------------------------------------------- |
-| Hash | `RepartitionExec(Hash)`: Ballista's hash shuffle, with DataFusion's hash |
-| Single partition, broadcast | `CoalescePartitionsExec` |
-| Range | `CoalescePartitionsExec`, so a sort runs in one task ([ballista#2680](https://github.com/apache/datafusion-ballista/issues/2680)) |
+| Exchange | Ballista plan |
+| --------------------------- | -------------------------------------------------------------------------------------------- |
+| Hash | `RepartitionExec(Hash)`: Ballista's hash shuffle, with DataFusion's hash |
+| Single partition, broadcast | `CoalescePartitionsExec` |
+| Range | None: the stage sorts the producer's partitions, and a `SortPreservingMergeExec` merges them |
A stage that fails to plan is refused, not run wrongly or loaded with code it names:
-- A stage can't call into the JVM: scalar subqueries and JVM UDFs fail to plan, and a scan whose
- object store needs a JVM credential provider fails when it runs.
+- A stage can't call into the JVM: JVM UDFs and a scalar subquery the driver has not replaced with
+ its value fail to plan, and a scan whose object store needs a JVM credential provider fails when
+ it runs.
+- A stage can't write files: a Parquet writer, an Iceberg write or a shuffle writer in a block is
+ refused. The cluster moves data between stages itself.
- A stage can't load a native UDF library. These processes serve every client, and a plan names
the library path, so a client would choose what the processes load.
- A stage that reads files can't also read a hash exchange. Spark places the rows of a bucketed
scan with Murmur3, so a join with a side shuffled by DataFusion's hash would return wrong rows.
+## Joins whose build side turns out small
+
+Spark plans a join's sides from its estimates, and its estimate of a filtered relation ignores the
+filter, so it shuffles both sides of a join whose build side is far smaller than it thinks. The
+scheduler moves such a join into the stage that produces its probe side, with the build side
+broadcast to it, so the probe side's exchange carries the join's output rather than all of its
+input (`src/adaptive.rs`). It picks the joins Spark's estimates say are worth measuring: an inner
+hash join built on the right, with column keys and no condition, whose build side Spark estimates
+within Ballista's maximum estimate multiple of its broadcast threshold and whose probe side many
+times larger.
+
+- In blocking mode, a rule in Ballista's adaptive planner holds the probe side's stage back until
+ the build side's stage finishes, then moves the join if the build side measured under
+ `ballista.optimizer.broadcast_join_threshold_bytes`, or releases the probe side's stage as it
+ was.
+- In pipelined mode every task is placed before any runs, so a build side cannot be measured first
+ and the join stays as Spark planned it.
+
## Building and running
```shell
@@ -77,7 +126,16 @@ native/target/debug/comet-executor &
Sharing `native/target` with the main workspace saves building the common dependencies twice. Both
binaries read their settings from environment variables, listed at the top of
-`src/bin/comet-scheduler.rs` and `src/bin/comet-executor.rs`.
+`src/bin/comet-scheduler.rs` and `src/bin/comet-executor.rs`. The scheduler's include:
+
+| Variable | Meaning |
+| -------------------------------------------- | ------------------------------------------------------------------------------ |
+| `COMET_SCHEDULER_BALLISTA_CONFIG` | Ballista settings for every query, such as `ballista.execution.mode=pipelined` |
+| `COMET_SCHEDULER_RESULT_RETENTION_SECONDS` | How long a query's results stay readable after it ends (default 300) |
+| `COMET_SCHEDULER_FLIGHT_MAX_REQUEST_BYTES` | The largest query the Flight service accepts (default 512 MiB) |
+| `COMET_SCHEDULER_QUERY_TIMEOUT_SECONDS` | The longest a query may take (default 0: no limit) |
+| `COMET_SCHEDULER_PIPELINED_RESULT_MAX_BYTES` | The most Arrow data buffered for one pipelined result (default 1 GiB) |
+| `COMET_SCHEDULER_PIPELINED_STORE_MAX_BYTES` | The most Arrow data kept for all pipelined results together (default 4 GiB) |
## Testing
@@ -90,8 +148,16 @@ binaries and runs queries through the scheduler's Flight service, as a driver wo
## Keeping it building
-This workspace builds Comet's native crates by path against the Ballista release on the same
-DataFusion major version, and CI builds and tests it in the `linux-test-native-executors` job. When
-Comet moves to a DataFusion major that Ballista hasn't released yet, the upgrade can turn that job
-off until Ballista catches up. Dependabot updates this workspace's lockfile, which is separate from
-`native/Cargo.lock`, so the two can resolve a crate to different versions between updates.
+This workspace builds Comet's native crates by path against Ballista's
+[`comet-native`](https://github.com/apache/datafusion-ballista/tree/comet-native) branch: Ballista's
+main branch with the enhancements Comet Native Executors need that haven't merged yet
+([#2682](https://github.com/apache/datafusion-ballista/pull/2682),
+[#2721](https://github.com/apache/datafusion-ballista/pull/2721),
+[#2722](https://github.com/apache/datafusion-ballista/pull/2722),
+[#2723](https://github.com/apache/datafusion-ballista/pull/2723) and
+[#2743](https://github.com/apache/datafusion-ballista/pull/2743)), on the same DataFusion major
+version. `Cargo.lock` pins the branch's commit. CI builds and tests the workspace in the
+`linux-test-native-executors` job. When Comet moves to a DataFusion major that Ballista hasn't
+caught up with, the upgrade can turn that job off until it does. Dependabot updates this
+workspace's lockfile, which is separate from `native/Cargo.lock`, so the two can resolve a crate to
+different versions between updates.
diff --git a/native/native-executors/src/adaptive.rs b/native/native-executors/src/adaptive.rs
new file mode 100644
index 00000000000..e837c3cf624
--- /dev/null
+++ b/native/native-executors/src/adaptive.rs
@@ -0,0 +1,517 @@
+// 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.
+
+//! Adaptive join planning for Comet fragments.
+//!
+//! A stage's joins are inside its Comet block, so Ballista's adaptive join rules see the stage
+//! as a leaf, and Spark has already placed an exchange on each side of every shuffled join. When
+//! a join's build side turns out small, broadcasting it is not enough: the probe side would still
+//! be shuffled whole to reach the join. These rules move the join into the stage that produces
+//! the probe side instead, with the build side broadcast to it, so the probe side's exchange
+//! carries the join's output rather than all of its input.
+//!
+//! That only pays off if the probe side's stage hasn't run yet when the build side's size is
+//! known. [`HoldProbeStages`] holds it back before anything runs, where Spark's estimates say the
+//! measurement could change the plan, and [`BroadcastIntoProbe`] decides once the build side's
+//! stage finishes: it moves the join where the build side measured small, and otherwise releases
+//! the probe side's stage as it was. Holding serialises the two stages, so like Ballista's own
+//! build-side staging it uses Ballista's broadcast threshold, minimum probe ratio and maximum
+//! estimate multiple.
+
+use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::Arc;
+
+use ballista_core::config::BallistaConfig;
+use ballista_scheduler::state::adaptive::{AdaptiveRule, AdaptiveRules, ExchangeExec};
+use comet::execution::native_executors::{
+ partition_descriptor, partition_plans, CometFragmentExec,
+};
+use datafusion::common::tree_node::{Transformed, TreeNode};
+use datafusion::common::{internal_datafusion_err, Result};
+use datafusion::config::ConfigOptions;
+use datafusion::physical_optimizer::PhysicalOptimizerRule;
+use datafusion::physical_plan::statistics::StatisticsContext;
+use datafusion::physical_plan::{
+ ChildrenPropertiesMode, ExecutionPlan, Partitioning, ReplaceChildrenOptions, StatisticsArgs,
+};
+use datafusion_comet_proto::spark_expression::expr::ExprStruct;
+use datafusion_comet_proto::spark_native_executors::Fragment;
+use datafusion_comet_proto::spark_operator::operator::OpStruct;
+use datafusion_comet_proto::spark_operator::{BuildSide, JoinType, Operator, ShuffleScan};
+use log::info;
+use prost::Message;
+
+/// The rules Comet adds to Ballista's adaptive planning.
+#[derive(Debug, Default)]
+pub struct CometAdaptiveRules;
+
+impl AdaptiveRules for CometAdaptiveRules {
+ fn preparation_rules(&self, _plan_ids: Arc) -> Vec {
+ vec![Arc::new(HoldProbeStages)]
+ }
+
+ fn replan_rules(&self, plan_ids: Arc) -> Vec {
+ vec![Arc::new(BroadcastIntoProbe { plan_ids })]
+ }
+}
+
+/// A join in a stage's block that can run in its probe side's stage instead.
+///
+/// It is an inner hash join built on the right, with no condition and with keys that are
+/// columns, whose inputs are both input leaves of the block. Moved below the probe side's
+/// exchange its output keeps the probe side's columns first, so the exchange's keys still name
+/// the same columns.
+struct MovableJoin {
+ /// The child indexes from the block's root to the join.
+ path: Vec,
+ /// The stage inputs that the join's probe (left) and build (right) leaves read.
+ probe_input: usize,
+ build_input: usize,
+}
+
+/// The joins of `block` that [`MovableJoin`] describes.
+fn movable_joins(block: &Operator) -> Vec {
+ fn walk(
+ op: &Operator,
+ path: &mut Vec,
+ leaves: &mut usize,
+ found: &mut Vec,
+ ) {
+ if is_input_leaf(op) {
+ *leaves += 1;
+ return;
+ }
+ if let Some(OpStruct::HashJoin(join)) = &op.op_struct {
+ let movable = join.join_type == JoinType::Inner as i32
+ && join.build_side == BuildSide::BuildRight as i32
+ && join.condition.is_none()
+ && !join.null_aware_anti_join
+ && join
+ .left_join_keys
+ .iter()
+ .chain(&join.right_join_keys)
+ .all(|key| matches!(key.expr_struct, Some(ExprStruct::Bound(_))))
+ && op.children.len() == 2
+ && op.children.iter().all(is_input_leaf);
+ if movable {
+ found.push(MovableJoin {
+ path: path.clone(),
+ probe_input: *leaves,
+ build_input: *leaves + 1,
+ });
+ }
+ }
+ for (index, child) in op.children.iter().enumerate() {
+ path.push(index);
+ walk(child, path, leaves, found);
+ path.pop();
+ }
+ }
+ let mut found = vec![];
+ walk(block, &mut vec![], &mut 0, &mut found);
+ found
+}
+
+/// Whether `op` reads one of its stage's inputs. Comet's planner feeds them to these leaves in
+/// the order a depth-first walk meets them.
+fn is_input_leaf(op: &Operator) -> bool {
+ matches!(
+ op.op_struct,
+ Some(OpStruct::Scan(_)) | Some(OpStruct::ShuffleScan(_))
+ )
+}
+
+fn leaf_fields(op: &Operator) -> &[datafusion_comet_proto::spark_expression::DataType] {
+ match &op.op_struct {
+ Some(OpStruct::Scan(scan)) => &scan.fields,
+ Some(OpStruct::ShuffleScan(scan)) => &scan.fields,
+ _ => &[],
+ }
+}
+
+fn node_at<'a>(block: &'a Operator, path: &[usize]) -> &'a Operator {
+ path.iter().fold(block, |op, index| &op.children[*index])
+}
+
+/// A stage's block, or `None` for a stage that reads files, whose partitions each run their own
+/// copy of it.
+fn block_of(fragment: &CometFragmentExec) -> Result