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 Nov 1, 2026
Open
Open
apache/main has apache#7299, the squashed first part of this branch. Conflicting files were merged against 5dd751b, the branch commit apache#7299 was taken from, so only apache#7299's review changes and main's own changes apply on top of the branch: native UDFs are refused on native executors, and a stage that reads files and a hash exchange is refused.
…et Native Executors Spark places an exchange on both sides of a shuffled join, so when the build side turns out small a broadcast alone still shuffles the whole probe side. Comet's adaptive rules, added to Ballista's adaptive planning, hold back the probe side's stage where Spark's estimates say the build side is worth measuring, then once the build side's stage finishes either move the join into the probe side's stage with the build side broadcast to it, or release the probe side as it was. The driver sends Spark's size estimate of each stage for the decision. Prototype for apache#7391, built against Ballista's comet-native branch plus a hook for adaptive rules on a fork.
…tive Executors A pipelined query places every task before any runs, so a join whose build side turns out small still shuffles its whole probe side. The scheduler now runs the build side of each movable join that Spark's estimates say is worth measuring as a pipelined query of its own first, and where it comes back under the broadcast threshold moves the join into the probe side's stage, reading the build side from memory. The rewrite is shared with the adaptive rule for blocking mode. Prototype for apache#7408.
A partial aggregate that stops aggregating sends each row on as a partial state, which DataFusion builds with GroupsAccumulator::convert_to_state. Comet's Float64 AvgGroupsAccumulator, which runs Spark's average of a decimal as avg(UnscaledValue(x)), returned NotImplemented. Each row's state is now its value and a count of one, or a zero sum and count for a null or filtered-out row, what state() emits for such a group.
…ing, including AVG configure_skip_partial_aggregation only admitted a plan whose root is a ShuffleWriter. A Native Executors fragment has none, since Ballista writes the shuffle, so a fragment never skipped even with spark.comet.exec.aggregate.skipPartial.enabled. A fragment's session now marks it, and the check treats it as feeding a shuffle. Single-argument AVG is admitted next to COUNT, which applies to Comet on Spark too. On TPC-H q17 at SF1000 in pipelined mode this cut the query from about 277 s to 242 s, and stopped an executor from being OOM-killed (apache#7406).
… with its adaptive rules hook comet-native now carries apache/datafusion-ballista#2743, which the adaptive join rule needs, so the workspace builds against it again rather than a fork. The README describes how the scheduler handles a join whose build side turns out small.
github-actions
Bot
added
the
area:aggregation
COMET_SCHEDULER_EVENT_LOG_DIR set the scheduler config's event_log_dir, but Ballista's EventLogWriter is built only with the scheduler's rest-api feature, which default-features = false had left out, so no event log was ever written. Enable rest-api (adds ballista-api-types, ballista-history and utoipa to the lockfile).
This was referenced Nov 2, 2026
Ballista's rest-api code calls tracing, which it links only for its binaries (build-binary), so ballista-scheduler with default-features = false and rest-api did not compile.
…Comet Native Executors" This reverts commit 7e8fb40.
…otype # Conflicts: # native/native-executors/Cargo.lock # native/native-executors/Cargo.toml # native/native-executors/src/bin/comet-executor.rs # native/native-executors/src/bin/comet-scheduler.rs # native/native-executors/tests/cluster.rs
andygrove
changed the title
…r, and bound pipelined results
This was referenced Nov 6, 2026
andygrove
added a commit
to andygrove/datafusion-comet
that referenced
this pull request
Nov 10, 2026
…im what no cluster here uses Review feedback on the driver: - Planning only validates the stage graph. The files a scan reads are listed, and the subqueries in its filters run, when the query executes, so EXPLAIN no longer contacts the cluster and the per-file plans no longer stay on the plan node. - A scan that lists no files gets one empty partition, so a query over an empty table or a fully pruned partition returns no rows instead of failing in the scheduler. - The offload follows the conversion's own guards: Comet loaded, exec enabled, not streaming. - A hash exchange keyed by an expression is refused naming the exchange, since the cluster hashes columns. - Remove the Stage fields output_ordering and estimated_output_bytes, which nothing in this repository reads, and the client's handling of a scheduler that serves partitions itself. - Ask libcomet whether it was built with the feature through NativeBase.isFeatureEnabled. - Add CometNativeExecutorsPlanSuite, which checks the stage graph, the refusals and the subquery rewrite in the regular CI matrix, and compare the cluster suite with Spark's answer with Comet off. - Document the blocked driver thread on cancellation and the generic exception for cluster errors, and say which TPC-H tests need the cluster of apache#7289.
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?
Part of #7288
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 is an experimental MVP of the design: it runs all 22 TPC-H queries end to end and is off by default. A query that the cluster can't run is refused, naming the operator, rather than run wrongly. The design:
Warning
This cannot merge until Comet can depend on a Ballista release or on Ballista's
mainbranch. The cluster workspace (native/native-executors) builds on Ballista'scomet-nativebranch, which holds changes that haven't merged into Ballista yet: apache/datafusion-ballista#2682, #2721, #2722, #2723 and #2743. A Comet source release can't build against a moving branch of another repository. Before this merges, those changes must land in Ballista and be released (or be on Ballista'smain), andnative/native-executors/Cargo.tomlmust switch from the git branch to that dependency.Split out for review. Parts of this branch that don't depend on Ballista's
comet-nativebranch are in their own pull requests. They carry the corresponding changes, in some places reworded or trimmed to what is true onapache/main, so once they merge, mergingapache/mainhere will need some conflict resolution and then leaves mostly the Ballista-dependent work in this diff:CometNativeExecutorsExec, the configs and the suite)What stays here is what needs the
comet-nativebranch: the adaptive broadcast-join rule, Ballista's pipelined mode with its result caps, and the dependency switch to the branch. The sort merge and the broadcast and range-exchange handling in the cluster also stay here until they are shown to work on Ballista 55.0.0, which hasn't been tested yet.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.COMET_SCHEDULER_FLIGHT_MAX_REQUEST_BYTES,COMET_SCHEDULER_PIPELINED_RESULT_MAX_BYTES,COMET_SCHEDULER_PIPELINED_STORE_MAX_BYTES,COMET_SCHEDULER_QUERY_TIMEOUT_SECONDS). Cancelling a pipelined query that fails, times out or loses its client cancels the job.native-executorsfeature says so, rather than failing with anUnsatisfiedLinkError. The Flight client has connect and keep-alive timeouts, and a result partition that expired on the cluster says so.native_executors.mdsay to run the cluster on a trusted network only.CometNativeExecutorsSuiteis listed indev/ci/check-suites.pyas a manual suite, and the Rust CI job runs clippy ondatafusion-cometwith thenative-executorsfeature.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.
native/Cargo.locklists the feature's optional dependencies (such asarrow-flightandtonic), which a default build doesn't compile.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 passed on SF10 Parquet data, before the guardrails in item 7 were added (the suite has since gained two tests that fail at planning, and they pass), 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.