Repository navigation
Conversation
github-actions
Bot
added
enhancement
This was referenced Oct 28, 2026
andygrove
force-pushed
the
native-executors-prototype
branch
3 times, most recently
from
October 28, 2026 17:05
ef46676 to
eca9510
Compare
A Spark driver submits a query to a Comet Native Executor cluster as a QueryRequest: a stage graph of Comet native blocks, connected by the exchanges between them, with each stage's Spark output schema and, for stages that read files, each partition's plan. A Fragment carries a stage from the scheduler to its executors. The proto crate also exports a fingerprint of the .proto files, so that a cluster can refuse a query from a client that speaks another plan format. See apache#7288.
CometFragmentExec runs a stage's native block for each of its partitions, with the block's input leaves fed by other DataFusion plans, such as a distributed engine's shuffle readers, instead of by the JVM. PhysicalPlanner::with_native_inputs is the seam: an input leaf (Scan or ShuffleScan) takes the next native input, cast to the types Spark declared for it, and an expression that calls into the JVM fails to plan. Each partition's output is cast to the stage's declared Spark schema before it leaves the stage, as a Spark task's shuffle writer casts its input, since a hash exchange places rows by the types of their values too. The block runs in a Comet session over the host's memory pool and disk manager. A partition descriptor child, an in-memory source with one row per partition, carries each partition's index in its stage and, for a stage that reads files, the partition's own copy of the block. An engine that splits a stage between tasks slices it with the stage's other leaves. block_schema plans a block with empty inputs, so that a scheduler can check a block before running it. See apache#7288.
…ista A new Cargo workspace, native/native-executors, builds the two processes of a Comet Native Executor cluster on the Ballista 55 crates. It is a workspace of its own, so the main Comet build never depends on Ballista. comet-scheduler wraps Ballista's scheduler in an Arrow Flight service. GetFlightInfo takes a QueryRequest, checks that the client runs the same Comet build, plans each block with Comet's planner, turns the stage graph into a Ballista physical plan, runs it, and returns one endpoint per result partition with an opaque ticket for the executor's DoGet. Hash exchanges become RepartitionExec(Hash), and single-partition and broadcast exchanges become CoalescePartitionsExec, which Ballista's planner turns into stage boundaries. A stage that reads exchanges has as many partitions as they deliver. Ballista can't plan a range shuffle for a physical plan yet (apache/datafusion-ballista#2680), so a range exchange is read as one partition, and the stage that sorts it runs in a single task. comet-executor is Ballista's executor with Comet's codec, which carries each CometFragmentExec's Fragment through Ballista's stage and task plans. It needs no JVM. Both binaries take their settings from the environment, with Ballista's defaults otherwise. See apache#7288.
tests/cluster.rs starts the two binaries, submits stage graphs to the scheduler's Flight service as a driver would, and reads every result partition from the executors with DoGet. It covers a hash shuffle, a range exchange, and a query from another Comet build, which the scheduler refuses.
native/native-executors is a Cargo workspace of its own, so the Rust jobs don't cover it. The new linux-test-native-executors job runs clippy and the workspace's tests, including the ones that start a scheduler and an executor, and the lint job checks its formatting.
With spark.comet.nativeExecutors.enabled, CometRule replaces the plan Comet would execute with a CometNativeExecutorsExec leaf, which runs the whole query on the cluster at spark.comet.nativeExecutors.scheduler.url. This works with AQE on, where Comet's query-stage-prep rule sees the whole plan, and off. The leaf is columnar, so Spark Connect's execute-and-submitJob path and collect() both read it through local tasks on the driver, each of which fetches one result partition. The stage graph builder turns each native block into a stage and each native shuffle into an input, and refuses, naming the operator, any query that cannot run entirely natively, including a block whose plan calls back into the JVM. Operators that only turn the result into JVM objects, as Dataset.rdd adds, stay on the driver. The driver's Flight client is a JNI library in libcomet behind the native-executors Cargo feature, off by default. CometNativeExecutorsSuite starts a scheduler and an executor and checks queries against Spark; it runs when COMET_NATIVE_EXECUTORS_BIN_DIR names the binaries. See apache#7288.
The driver now serializes a range exchange as a RangePartition with its sort keys and no bounds. Until Ballista can plan a range shuffle for a physical plan (apache/datafusion-ballista#2680), the scheduler reads a range exchange as one partition, so the stage that sorts it runs in a single task. That is correct for a global sort, if serial. CometNativeExecutorsSuite runs TPC-H q1, with and without its order by, on the lineitem table under SPARK_TPCH_DATA when it is set, and on generated data with the same schema otherwise. The query that the cluster must refuse is now a broadcast join. See apache#7288.
It runs only against comet-scheduler and comet-executor binaries named by COMET_NATIVE_EXECUTORS_BIN_DIR, so the PR build matrix would only skip it.
…warnings withSQLConf returns Unit on Spark 3.x, so onCluster takes and returns Unit. NativeExecutorsClient's JNI declarations join Native's under the strict-warnings unused-params filter.
andygrove
force-pushed
the
native-executors-prototype
branch
from
October 31, 2026 12:33
eca9510 to
9cf6478
Compare
The branch is Ballista's main branch with the enhancements Comet Native Executors need that haven't merged yet: apache/datafusion-ballista#2682, apache#2721, apache#2722 and apache#2723. Cargo.lock pins its commit.
… moves Ballista runs EnsureRequirements on each stage, which takes away the coalesce above a stage's input and puts one back only where the parent requires a single partition. It rebuilds the node in between, so CometFragmentExec now checks that its inputs line up with its partitions when it runs rather than when it is built, and a stage with one partition requires each input as one partition.
…e Executors The stage graph builder now takes a broadcast exchange as a broadcast input and resolves a reused exchange to the stage it reuses. A query's scalar subqueries are expressions of CometNativeExecutorsExec, so Spark runs them before the query, and their results replace them in the plan as literals before it is submitted. CometTakeOrderedAndProjectExec becomes a stage that reads its child as one partition and runs Comet's final top-K plan.
With SPARK_TPCH_DATA set, CometNativeExecutorsSuite runs each TPC-H query on the cluster and compares its rows, in order when the query sorts, with Spark's. TPCH_QUERIES picks a subset. The suite also covers a broadcast join and a scalar subquery, and the query the cluster must refuse is now a sort that stays in Spark.
…t points A range exchange arrives without bounds, which Spark's driver would find by sampling the exchange's input. The scheduler now does the same while it builds the plan: it runs a job of its own over the stage that feeds the exchange, in which a SampleExec keeps a reservoir sample of the sort keys of each partition, reads the sample back, and cuts it into ranges of about the same number of rows. The exchange becomes a range RepartitionExec with those split points, which Ballista plans as a range shuffle, so the sort that reads it runs in as many tasks. A range exchange whose sort keys aren't all columns, or whose sample yields no split points, is still read as one partition. The workspace now pins Ballista's comet-native branch at the commit that keeps a submitted range repartition as a stage boundary.
…ures as such A range shuffle compares raw values, while Comet's sort normalizes float keys first, so a NaN with its sign bit set would land in the lowest range but sort last. A range exchange now range-partitions only by keys whose types Arrow orders as Spark does, and reads any other as one partition. A failed sampling job now reaches the driver with its own status rather than as an invalid query, the sampler no longer wraps its plan in a coalesce the adaptive planner removes, and a subquery's result, already in Catalyst's internal form, becomes a Literal directly.
This was referenced Oct 31, 2026
This was referenced Oct 31, 2026
Comet Native Executors: pass scalar subquery results to the cluster without rewriting the plan
#7374
Open
The scheduler has one build_plan, which takes the sampler, and a range exchange that can't be range-partitioned falls back through one helper. It checks that a stage's exchanges agree on their partition count before the plan reaches Ballista. It reads a sampling job's partitions with Ballista's client, a few at a time, and a SampleExec travels as a Sample message. split_points sorts the sample by the row format alone, and the reservoir allocates only when a batch changes it. The driver stamps a query with its version by appending the fields rather than decoding and encoding it, rewrites a stage's partition plans only when its plan holds a subquery, and walks protobuf messages with one helper. Comments link the follow-on issues under apache#7288.
COMET_SCHEDULER_BALLISTA_CONFIG sets Ballista settings, such as ballista.execution.mode=pipelined, for every query. A pipelined job's final stage streams each result partition to one reader, once, and can't finish until it is read, while Spark may read a result partition more than once. So the scheduler reads a pipelined query's results itself, gathered into one partition, and serves them from its own Flight service for as long as a blocking query's results are kept. The driver reads an endpoint without a location from the scheduler. A pipelined job can't range-partition, so the scheduler doesn't sample range exchanges in that mode. The workspace now builds on comet-native at 8ab06362a, which merges Ballista's main branch with its pipelined execution mode.
withSQLConf returns Unit on Spark 3.x, so Spark's expected rows come out through a var.
Executors report task status to the host the scheduler advertises, which is localhost by default. On Kubernetes they need the scheduler's Service name, which COMET_SCHEDULER_EXTERNAL_HOST sets.
… limits A task of a stage that reads files carries each of its partitions' copy of the block, which at TPC-H SF1000 can exceed Ballista's 16 MiB gRPC limit, and the scheduler then loses the executor it can't send the task to. COMET_GRPC_MAX_MESSAGE_SIZE sets the limit for both binaries.
A command's plan, once it has run, is a CommandResultExec, which explaining it, as tpcbench does for q15's CREATE VIEW, plans again. The cluster has nothing to run for it, so Comet now leaves it in Spark, as it does the commands themselves.
COMET_EXECUTOR_MEMORY_POOL_SIZE sets the bytes the executor's tasks share. Ballista's default, 70% of the container's limit, leaves too little for the memory outside the pool at TPC-H SF1000, where an executor was OOM-killed.
andygrove
changed the title
… Executors 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 bounds. The scheduler did the same, with a sampling job that ran every stage below the exchange once more, which at TPC-H SF1000 more than doubled q4 and q5. The driver now sends a stage that reads a range exchange with the order its block sorts by. The cluster reads the producer's partitions as they are, sorts each with the stage's block, and merges them with a SortPreservingMergeExec, as DataFusion plans a global sort. That needs no sampling, works in pipelined mode too, and leaves only the merge in one task. A stage whose order isn't over columns of types Arrow and Spark order alike still reads the exchange as one partition. SampleExec, the sampling job and their tests are gone.
…ture hybrid execution research
…met Native Executors
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Closes #7288
The native side of this prototype is up for review on its own in #7299. Until #7299 merges, this branch also holds its commits, the first five below. Once it merges, I'll rebase this PR onto main so it holds only the driver side.
Rationale for this change
#7288 proposes Comet Native Executors: an optional deployment mode where a Spark driver plans each query and a cluster of native processes built on Apache DataFusion Ballista runs it, with no JVM in the data plane. This draft is the prototype that the issue's first sub-task calls for. It runs all 22 TPC-H queries end to end, to check the design before building it out:
It is not meant to merge as is.
What changes are included in this PR?
The first five commits are #7299, which describes them: the wire protocol, Comet native blocks as DataFusion operators, the
comet-schedulerandcomet-executorbinaries, their tests through Flight, and a CI job for the new workspace. On top of them:native-executorsCargo feature that is off by default. It fills in the query's Comet version and plan format, submits it to the scheduler, and returns the result partitions' endpoints to the JVM as aResultEndpointsmessage.spark.comet.nativeExecutors.enabledandspark.comet.nativeExecutors.scheduler.url.CometNativeExecutorsExec, a leaf thatCometRulesubstitutes for the whole converted plan, with AQE on or off. Each partition of its RDD reads one result partition from the executor that holds it.Stage.output_ordering), and the cluster reads the producer's partitions as they are, sorts each with the stage's block, and merges them with aSortPreservingMergeExec. That needs no sampling, works in pipelined mode too, and leaves only the merge in one task. A stage whose order isn't over columns of types Arrow and Spark order alike, such as floats, reads the exchange as one partition and sorts it in a single task (Comet Native Executors: range-partition sorts on float, nested and expression keys #7366).CometNativeExecutorsExec, so Spark runs them before the query, and their results replace them in the plan as literals before it is submitted.CometTakeOrderedAndProjectExecbecomes a stage that reads its child as one partition and runs Comet's final top-K plan.comet-nativebranch. The workspace builds oncomet-nativein apache/datafusion-ballista, andCargo.lockpins its commit. The branch is Ballista's main branch, including its experimental pipelined execution mode (feat: Add experimental pipelined execution mode (all tasks placed up front, streaming exchange) datafusion-ballista#2657), with the enhancements this needs that haven't merged yet:Partitioning::Range. Sorts no longer use it.CometFragmentExec. Ballista runsEnsureRequirementson each stage. That takes away the coalesce above a stage's input and puts one back only where the parent requires a single partition, and it rebuilds the node in between.CometFragmentExecnow checks that its inputs line up with its partitions when it runs rather than when it is built, and a stage with one partition requires each input as one partition.COMET_SCHEDULER_BALLISTA_CONFIGsets Ballista settings for every query, such asballista.execution.mode=pipelined, in which a job places all its tasks up front and streams batches between stages instead of writing shuffle files.COMET_SCHEDULER_EXTERNAL_HOSTsets the host executors report task status to,COMET_GRPC_MAX_MESSAGE_SIZEraises Ballista's 16 MiB gRPC limit, which a task carrying the plans of many scan partitions outgrows at SF1000, andCOMET_EXECUTOR_MEMORY_POOL_SIZEsets the executor's memory pool.CommandResultExec), which explaining a command such asCREATE VIEWplans again, stays in Spark.CometNativeExecutorsSuiteis listed indev/ci/check-suites.pyas a manual suite.What works: Parquet scans, filters, projections, aggregates, shuffled and broadcast joins, sorts, top-K and scalar subqueries, across hash, range, single-partition and broadcast exchanges, in Ballista's blocking and pipelined modes. That covers all 22 TPC-H queries, which return the same rows as Spark at SF10 in both modes.
Not yet, filed as sub-issues of #7288:
LIMITandshow(), and a per-partition top-K before a top-K's exchange: Comet Native Executors: run LIMIT and top-K queries with a per-partition limit before the exchange #7365Ballista follow-ons: apache/datafusion-ballista#2727 (a stage that several stages read runs once for each), apache/datafusion-ballista#2728 (partition coalescing for stages Comet runs), and, not yet filed, a memory budget that a pipelined job's tasks on an executor share, instead of a pool each (see the memory note under the benchmark).
Default builds are unaffected. The configs default to off, the Cargo feature is off, and Ballista exists only in the separate workspace.
How are these changes tested?
Native side: feat: add comet-scheduler and comet-executor for Comet Native Executors #7299's tests.
tests/assembly.rsruns stage graphs on DataFusion alone, andtests/cluster.rsruns queries on the real binaries through Flight, in CI. New tests check that a range exchange is sorted per partition and merged in order, through Flight too, and that one sorted by a float is read as one partition.Scala:
CometNativeExecutorsSuitestarts a realcomet-schedulerandcomet-executor, runs queries through Spark with the feature on, and compares the rows with Spark's. It also asserts that the plan ran on the cluster. Queries covered:order by, checking row order tooSPARK_TPCH_DATAnames the tables, checking row order for those that sort.TPCH_QUERIESpicks a subset. Expected rows come from Spark with Comet off.All 34 tests pass on SF10 Parquet data, with release builds, both with Ballista's default blocking mode and with
COMET_SCHEDULER_BALLISTA_CONFIG=ballista.execution.mode=pipelined. In the pipelined run, each of the 14 global sorts ran as sorts of the producer's partitions and a merge, with no range shuffle.The suite is manual: it skips itself unless
COMET_NATIVE_EXECUTORS_BIN_DIRnames the binaries. To run it:cargo build --release --manifest-path native/Cargo.toml -p datafusion-comet --features native-executors CARGO_TARGET_DIR=native/target cargo build --release --manifest-path native/native-executors/Cargo.toml ./mvnw install -Prelease -DskipTests COMET_NATIVE_EXECUTORS_BIN_DIR=native/target/release SPARK_TPCH_DATA=/path/to/tpch/sf10 \ ./mvnw test -Prelease -Dtest=none -Dsuites=org.apache.comet.exec.CometNativeExecutorsSuiteExisting tests: clippy is clean for core with the new feature and for the new workspace, and the new workspace's tests pass.
Benchmark: TPC-H SF1000 on EKS
All 22 queries ran on an Amazon EKS cluster, in Ballista's blocking and pipelined modes, and every Comet on Ballista result had the expected number of rows. In pipelined mode the total was 462.2 s, 22% less than in blocking mode (593.6 s), and 1.02x the time of Comet 1.1.0-rc2 on Spark. Ballista's own
tpchrunner on the same cluster took 281.1 s in pipelined mode and 483.9 s in blocking mode, so Comet on Ballista took 1.64x and 1.23x as long.Cluster. Kubernetes on AWS in
us-west-2, on a mix ofr6i.16xlargeandr6i.24xlargenodes with EBS only. There are 32comet-executorpods of 8 vCPU and 64 GiB, each with a dedicated 1000 GiB gp3 volume for its work directory, and onecomet-schedulerpod. The driver is a pod of 8 CPUs and 40 GiB running Spark 4.1.3 inlocal[8], submitting each query to the scheduler. TPC-H SF1000 Parquet (ZSTD, Hive-partitioned) is on S3 in the same region, and the executors read it from there.Setup. The executors' memory pool is 21.3 GiB, a third of the pod (
COMET_EXECUTOR_MEMORY_POOL_SIZE). Comet uses native shuffle, and Spark's other settings are the benchmark harness's defaults, such as 512 shuffle partitions. Each query ran twice, fromspark.sql()to the end ofcollect(), with no warm-up, so the first iteration is cold. The times are the mean of the two. No executor restarted in either run. The Comet on Ballista columns run this branch, which builds on Ballista'scomet-nativebranch pinned at8ab06362: Ballistamainat0ef7399eplus 21 commits, including the four unmerged pull requests above.Results. Times in seconds, one run of each. The Ballista columns are Ballista
mainat0ef7399e(DataFusion 55.2.0) with its owntpchrunner, on the same cluster shape: 32 executors of 8 vCPU and 64 GiB. The Comet on Spark column is from Ballista's benchmarking page: Comet 1.1.0-rc2 on Spark 4.1.3, with executors of 8 vCPU, 64 GiB and 10 GiB overhead plus 32 GiB off-heap. It is from another run. The Ballista runner uses the SQLBench-H phrasing of the queries and Spark runs spark-perf's, which write some dates and Q15 differently, so small differences are within noise.q11 returns no rows at SF1000, since the query file's threshold doesn't scale with the data.
Notes.