Native Spark execution: Apache Gluten and DataFusion Comet compared

• What problem do Gluten and Comet solve?
• How are the two engines put together?
  • Comet: a protobuf plan for DataFusion
  • Gluten: a Substrait plan for Velox
• Under the hood: following one query through each engine
  • Rewrite: whole operators, or nothing
  • Scan: Parquet straight into columns
  • Shuffle: columnar bytes on disk
  • Transitions: where rows and columns meet
  • Memory: both live off the heap
• Hands-on: one query, three engines
  • Start the stack and fetch both engines
  • Generate the data
  • The query
  • Three launch commands
  • Fifteen identical rows
  • Three plans
• Real-world scenarios and production failures
  • The job that ran on stock Spark while claiming acceleration
  • Gluten on Java 17: the error that points at Velox
  • The version warning that is only a warning
  • One Python UDF, two very different plans
  • Architecture and platform: where each one can run at all
• Optimisation, tuning knobs and best practices
  • Choosing between them
  • Settings that matter
  • Best practices
• Summary and key takeaways
• Frequently asked questions
• References
• Trademarks

TL;DR

• Gluten and Comet are both Spark plugins that rewrite the physical plan after Catalyst and run the supported operators in native code on columnar batches. Neither changes the DataFrame or SQL you write, and on the query in this post all three runs return identical rows.
• They hand the plan to different engines. Comet serialises it to protobuf for Apache DataFusion, written in Rust. Gluten converts it to Substrait for Velox, a C++ library, or for a ClickHouse-derived backend.
• They fall back differently. When a Python UDF sits mid-plan, Gluten converts rows back to columns above it and keeps the aggregates and shuffles native. Comet leaves the aggregates and the shuffle between them on the JVM.
• They fail differently. Without off-heap memory, Comet logs one WARN and silently runs stock Spark. Gluten refuses to start the SparkContext.
• Read the physical plan to prove either one is on. Comet* operators, or Gluten’s *Transformer operators under ^(n) stage markers, are the only reliable evidence.

What problem do Gluten and Comet solve?

Spark’s SQL engine runs on the JVM. Whole-stage code generation, added in Spark 2.0, fuses a chain of operators into one generated Java method. That removed most of the per-row virtual calls of the old iterator model. But the result is still JVM bytecode walking rows, one at a time, through code the JIT may or may not vectorise.

Analytical queries spend most of their CPU in a handful of operators:

• decoding Parquet
• evaluating filters and projections
• hashing for joins and aggregations
• serialising rows for the shuffle

Those are exactly the operations that native code does well on columnar batches: tight loops over contiguous arrays of one type, which the compiler can turn into SIMD instructions, with no garbage collector involved.

Two projects set out to put native engines under Spark without asking anyone to port a query:

The constraint both accept: neither may change your results or your API. Each is a plan rewrite, not a new engine you port to. So each falls back to Spark rather than failing when it meets something it cannot do. How each one falls back is the most important difference between them.

How are the two engines put together?

Both hook into the same place: after Catalyst has produced a physical plan, and before Spark executes it. Both replace supported operators with native ones and leave the rest alone. What differs is what crosses into native code and how.

Comet: a protobuf plan for DataFusion

 Driver (JVM)                                  Executor
 +----------------------------------+
 | DataFrame / SQL                  |
 |        |                         |
 |   Catalyst -> physical plan      |
 |        |                         |
 |   CometExecRule                  |
 |   (replace supported operators)  |
 |        |                         |
 |   QueryPlanSerde -> protobuf     |
 +--------|-------------------------+
          |  task carries the protobuf plan
          v
                       +-------------------------------------+
                       | CometNativeExec (JVM)               |
                       |   Native.createPlan(protobuf)  JNI  |
                       |        |                            |
                       |   libcomet.so (Rust)                |
                       |   PhysicalPlanner -> DataFusion     |
                       |        |                            |
                       |   Arrow batches back via Arrow FFI  |
                       +-------------------------------------+

Walk it top down:

1. CometPlugin is named in spark.plugins. Its driver half sets spark.sql.extensions to org.apache.comet.CometSparkSessionExtensions, and logs that it did.
2. CometExecRule, installed by that extension, walks the physical plan and swaps each supported operator for its Comet* equivalent.
3. QueryPlanSerde serialises each native block of operators into a protobuf message. The message travels to executors inside the task, like any other part of a Spark plan.
4. On the executor, CometNativeExec passes the protobuf to Native.createPlan over JNI. On the Rust side, PhysicalPlanner rebuilds it as a DataFusion execution plan.
5. Results return to the JVM as Arrow batches through the Arrow C Data Interface (FFI), so the columns are handed over by pointer and not copied.

The native library ships inside the jar, one per platform:

130312592  org/apache/comet/linux/amd64/libcomet.so
122080992  org/apache/comet/linux/aarch64/libcomet.so

Gluten: a Substrait plan for Velox

 Driver (JVM)                                  Executor
 +----------------------------------+
 | DataFrame / SQL                  |
 |        |                         |
 |   Catalyst -> physical plan      |
 |        |                         |
 |   Gluten rules: offload, then    |
 |   validate each *Transformer     |
 |   (failed validation = fallback, |
 |    logged by GlutenFallbackReporter)
 |        |                         |
 |   WholeStageTransformer ^(n)     |
 |   -> one Substrait plan per stage|
 +--------|-------------------------+
          |  task carries Substrait bytes
          v
                       +-------------------------------------+
                       | libgluten.so (JNI bridge)           |
                       |        |                            |
                       |   Substrait -> Velox plan           |
                       |   libvelox.so executes              |
                       |        |                            |
                       |   Arrow-compatible ColumnarBatch    |
                       +-------------------------------------+

The shape is the same, but three details differ from Comet.

Validation happens in native code. Gluten does more than check an allow-list on the JVM side. Each candidate *Transformer builds its Substrait fragment and asks the native backend whether it can execute it. A rejection falls back that node alone, and it is logged with the reason, as shown later.

Stages are Substrait plans. Gluten groups consecutive native operators under a WholeStageTransformer, its counterpart of Spark’s WholeStageCodegenExec. The plan string marks each one ^(n), the way Spark marks code-generated stages *(n). Each ^(n) stage is one Substrait plan, serialised to protobuf bytes and sent to the executor.

The backend is pluggable. The released jar is the Velox backend. It carries two native libraries, libgluten.so as the JNI bridge and libvelox.so as the engine:

 33356520  linux/amd64/libgluten.so
260293336  linux/amd64/libvelox.so

The other backend ports ClickHouse, based on version 23.1, as a library named libch.so. It has no release binary; you build it from source. Everything below uses the Velox backend.

Under the hood: following one query through each engine

This section follows one query through the phases where the two designs differ: the plan rewrite, the scan, the shuffle, the row-to-column transitions, and memory.

Rewrite: whole operators, or nothing

Both rewrites are per operator. Spark hands over a tree like this one:

Sort
+- Exchange (range)
   +- HashAggregate (final)
      +- Exchange (hash)
         +- HashAggregate (partial)
            +- BroadcastHashJoin
               :- Filter
               :  +- FileScan parquet
               +- BroadcastExchange
                  +- Filter
                     +- FileScan parquet

Each engine asks the same question of every node: can I run this, with these expressions and these types? It keeps the nodes where the answer is yes. The difference is in what happens at the boundary between a yes and a no, covered under “Transitions” below.

Scan: Parquet straight into columns

Stock Spark 3.5 already has a vectorised Parquet reader. It decodes into Spark’s ColumnarBatch, and then a ColumnarToRow node converts to rows immediately, because every operator above it is row-based. That conversion is visible in the stock plan below, directly above each FileScan.

Comet’s CometNativeScan decodes Parquet in Rust, straight into Arrow arrays. Gluten’s scan transformer, printed as FileFileSourceScanExecTransformer, hands the files to Velox’s Parquet reader. Gluten also prints the filters it pushed into the native reader as NativeFilters, separately from Spark’s own PushedFilters:

NativeFilters: [(amount#3 > 100.00),isnotnull(status#2),(status#2 = delivered),
                isnotnull(customer_id#1L),isnotnull(amount#3)]

In both engines, what leaves the scan stays columnar all the way up, as long as the operators above it are native.

Shuffle: columnar bytes on disk

Both engines replace Spark’s shuffle manager. Spark’s sort-based shuffle serialises rows. A columnar pipeline would have to convert to rows just to write the shuffle, and back to columns to read it.

Gluten’s shuffle adds two things to the plan. First, a ProjectExecTransformer computes the partition key as an explicit column, hash(country, segment, 42) AS hash_partition_key. That is the same Murmur3 hash and seed 42 that Spark’s HashPartitioning uses. Second, a VeloxResizeBatches node sits below each ColumnarExchange. It appends small batches to earlier ones, or splits large ones. Its three arguments in the plan, 1024, 2147483647, 10485760, are the minimum and maximum output batch size in rows and the preferred batch size in bytes (10 MiB).

Both still use Spark’s shuffle service, block manager and map output tracker. Only the bytes differ.

Transitions: where rows and columns meet

This is the design difference that matters most in production.

When a native pipeline meets an operator it cannot run, it must convert to rows: Comet inserts CometColumnarToRow and Gluten inserts VeloxColumnarToRow. The question is what happens above the unsupported operator.

Gluten converts back. It inserts RowToVeloxColumnar above the fallen-back node, and the operators above it can be native again:

  native     ^(3) FlushableHashAggregateTransformer (partial)
  native     ^(3) ProjectExecTransformer
                 InputIteratorTransformer
  R2C            RowToVeloxColumnar            <- rows back to columns
  JVM            BatchEvalPython               <- the one fallback
  C2R            VeloxColumnarToRow            <- columns to rows
  native     ^(2) ProjectExecTransformer ... BroadcastHashJoinExecTransformer

Comet does not. Above a JVM operator, Comet 1.0.0 leaves that operator’s consumers on the JVM too. For aggregates, the reason is specific: Comet’s partial-aggregate buffers are not all compatible with Spark’s, so a native final aggregate refuses to merge buffers that a JVM partial aggregate produced. With JVM operators on both sides, the shuffle between them reverts to Spark’s as well. The hands-on section shows both plans in full.

Gluten’s policy has limits of its own. It counts transitions, and it can decide that a heavily fragmented stage is better left entirely on the JVM. spark.gluten.sql.columnar.wholeStage.fallback.threshold and spark.gluten.sql.columnar.query.fallback.threshold control this; both default to -1, which means off.

Memory: both live off the heap

Both engines allocate native buffers outside the Java heap, and both account for them against Spark’s off-heap pool. That is why both require spark.memory.offHeap.enabled=true and a spark.memory.offHeap.size. What differs is how they react when it is missing, covered under “Production failures”.

Gluten also sets spark.executor.memoryOverhead itself if you have not:

INFO VeloxListenerApi: Memory overhead is not set. Setting it to 966367641
automatically. Gluten doesn't follow Spark's calculation on default value of
this option because the actual required memory overhead will depend on off-heap
usage than on on-heap usage.

On a cluster manager, that changes the container size you are billed for, so set it explicitly.

Hands-on: one query, three engines

Everything below ran on Spark 3.5.9, Scala 2.12 and Java 17, in the datalake-playground stack, with DataFusion Comet 1.0.0 and Apache Gluten 1.7.0 (Velox backend). Both engines publish Spark 3.3 to 4.1 builds under their own coordinates. Comet’s artifacts on Maven Central are comet-spark-spark3.5_2.12, comet-spark-spark4.1_2.13 and so on. Gluten’s tarballs are apache-gluten-<version>-bin-spark-<minor>.tar.gz on the Apache download site.

Start the stack and fetch both engines

If you have not set up the stack before, the setup post walks through it. Start it with:

git clone https://github.com/rangareddy/datalake-playground.git
cd datalake-playground
SPARK_VERSION=3.5.9 sh run_datalake.sh start
sh run_datalake.sh status

Download both engines inside the spark-master container. Comet is a single jar on Maven Central. Gluten is a tarball on the Apache archive, and its Velox bundle jar is inside it:

docker exec spark-master bash -c '
  mkdir -p /tmp/accel && cd /tmp/accel &&
  curl -sfLO https://repo1.maven.org/maven2/org/apache/datafusion/comet-spark-spark3.5_2.12/1.0.0/comet-spark-spark3.5_2.12-1.0.0.jar &&
  curl -sfLO https://archive.apache.org/dist/gluten/1.7.0/apache-gluten-1.7.0-bin-spark-3.5.tar.gz &&
  tar xzf apache-gluten-1.7.0-bin-spark-3.5.tar.gz &&
  ls -la *.jar apache-gluten-1.7.0-bin-spark-3.5/*.jar'

The two jars are 87,851,995 and 183,177,421 bytes. Gluten’s release ships linux_amd64 only. On an arm64 host, the Gluten half of this demo needs a source build.

Generate the data

Two tables, written as Parquet. Note the moduli. The status cycle has period 4, so a country derived from a period-4 expression would make status = 'delivered' select exactly one country, and the demo would quietly measure nothing. Five countries and three segments are coprime with four, which keeps the filter and the grouping independent.

The stack’s core-site.xml sets fs.defaultFS to s3a://warehouse/, so a bare path like /tmp/accel_demo would land in MinIO. The scripts use explicit file:// paths to keep the demo on local disk.

Save this as gen_data.py:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder.appName("accel-demo-data")
         .config("spark.ui.enabled", "false")
         .config("spark.eventLog.enabled", "false")
         .getOrCreate())

customers = (spark.range(0, 200_000)
             .select(
                 F.col("id").alias("customer_id"),
                 F.element_at(
                     F.array(F.lit("IN"), F.lit("US"), F.lit("DE"), F.lit("SG"), F.lit("BR")),
                     (F.col("id") % 5 + 1).cast("int")).alias("country"),
                 F.element_at(
                     F.array(F.lit("consumer"), F.lit("business"), F.lit("enterprise")),
                     (F.col("id") % 3 + 1).cast("int")).alias("segment"),
                 F.date_add(F.lit("2023-01-01").cast("date"),
                            (F.col("id") % 900).cast("int")).alias("signup_date"),
             ))
customers.write.mode("overwrite").parquet("file:///tmp/accel_demo/customers")

orders = (spark.range(0, 20_000_000)
          .select(
              F.col("id").alias("order_id"),
              (F.col("id") % 200_000).alias("customer_id"),
              F.element_at(
                  F.array(F.lit("placed"), F.lit("shipped"), F.lit("delivered"), F.lit("cancelled")),
                  (F.col("id") % 4 + 1).cast("int")).alias("status"),
              (F.col("id") % 997 + 1).cast("decimal(10,2)").alias("amount"),
              (F.col("id") % 7 + 1).cast("int").alias("quantity"),
              F.to_timestamp(F.date_add(F.lit("2025-01-01").cast("date"),
                                        (F.col("id") % 365).cast("int"))).alias("order_ts"),
          ))
orders.write.mode("overwrite").parquet("file:///tmp/accel_demo/orders")

print("customers:", spark.read.parquet("file:///tmp/accel_demo/customers").count())
print("orders   :", spark.read.parquet("file:///tmp/accel_demo/orders").count())
spark.stop()

Run it on stock Spark:

docker cp gen_data.py spark-master:/tmp/accel/
docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[4]" \
  --driver-memory 3g --conf spark.eventLog.enabled=false --conf spark.ui.enabled=false \
  /tmp/accel/gen_data.py'

customers: 200000
orders   : 20000000

That is 169 MB of Parquet for orders and 872 KB for customers.

The query

Save this as revenue_by_segment.py. It exercises scan, filter, broadcast join, two-phase aggregation, two exchanges and a sort:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = (SparkSession.builder.appName("accel-demo-query")
         .config("spark.ui.enabled", "false")
         .config("spark.eventLog.enabled", "false")
         .config("spark.sql.shuffle.partitions", "8")
         .getOrCreate())

orders = spark.read.parquet("file:///tmp/accel_demo/orders")
customers = spark.read.parquet("file:///tmp/accel_demo/customers")

revenue = (orders
           .where((F.col("status") == "delivered") & (F.col("amount") > 100))
           .join(customers, "customer_id")
           .groupBy("country", "segment")
           .agg(F.sum("amount").alias("revenue"),
                F.count("*").alias("order_count"),
                F.avg("quantity").alias("avg_quantity"))
           .orderBy(F.col("revenue").desc()))

rows = revenue.collect()
print("=== RESULT ===")
for r in rows:
    print(f"{r['country']:<4} {r['segment']:<12} revenue={r['revenue']:>14} "
          f"orders={r['order_count']:>9} avg_qty={r['avg_quantity']:.3f}")
print("=== PHYSICAL PLAN ===")
print(revenue._jdf.queryExecution().executedPlan().toString())
spark.stop()

The script prints executedPlan after collect(), not explain() before it. With adaptive execution on, only the executed plan shows the final stages, and the engines’ rewrites apply to those.

Three launch commands

Copy the script in once:

docker cp revenue_by_segment.py spark-master:/tmp/accel/

Stock Spark:

docker exec spark-master bash -c '$SPARK_HOME/bin/spark-submit --master "local[4]" \
  --driver-memory 3g --conf spark.eventLog.enabled=false --conf spark.ui.enabled=false \
  /tmp/accel/revenue_by_segment.py'

Comet needs the jar, the plugin, its shuffle manager and off-heap memory:

docker exec spark-master bash -c 'COMET=/tmp/accel/comet-spark-spark3.5_2.12-1.0.0.jar
$SPARK_HOME/bin/spark-submit --master "local[4]" \
  --driver-memory 3g --conf spark.eventLog.enabled=false --conf spark.ui.enabled=false \
  --jars $COMET \
  --conf spark.driver.extraClassPath=$COMET \
  --conf spark.executor.extraClassPath=$COMET \
  --conf spark.plugins=org.apache.spark.CometPlugin \
  --conf spark.shuffle.manager=org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=3g \
  /tmp/accel/revenue_by_segment.py'

Gluten needs the same four things under its own names, plus a Netty flag on Java 17:

docker exec spark-master bash -c 'GLUTEN=/tmp/accel/apache-gluten-1.7.0-bin-spark-3.5/gluten-velox-bundle-spark3.5_2.12-linux_amd64-1.7.0.jar
$SPARK_HOME/bin/spark-submit --master "local[4]" \
  --driver-memory 3g --conf spark.eventLog.enabled=false --conf spark.ui.enabled=false \
  --jars $GLUTEN \
  --conf spark.driver.extraClassPath=$GLUTEN \
  --conf spark.executor.extraClassPath=$GLUTEN \
  --conf spark.plugins=org.apache.gluten.GlutenPlugin \
  --conf spark.shuffle.manager=org.apache.spark.shuffle.sort.ColumnarShuffleManager \
  --conf spark.memory.offHeap.enabled=true \
  --conf spark.memory.offHeap.size=3g \
  --conf "spark.driver.extraJavaOptions=-Divy.cache.dir=/tmp -Divy.home=/tmp -Dcom.amazonaws.services.s3.enableV4=true -Dio.netty.tryReflectionSetAccessible=true" \
  --conf "spark.executor.extraJavaOptions=-Dcom.amazonaws.services.s3.enableV4=true -Dio.netty.tryReflectionSetAccessible=true" \
  /tmp/accel/revenue_by_segment.py'

The two extraJavaOptions values repeat what the image’s spark-defaults.conf already sets and add -Dio.netty.tryReflectionSetAccessible=true. A --conf replaces a default outright rather than appending to it. So passing only the Netty flag would silently drop the image’s own options.

Fifteen identical rows

All three runs return the same fifteen groups. These are the first five:

SG   consumer     revenue=  164676437.00 orders=   299958 avg_qty=4.000
IN   business     revenue=  164676249.00 orders=   299958 avg_qty=4.000
BR   enterprise   revenue=  164675895.00 orders=   299958 avg_qty=4.000
US   consumer     revenue=  164675610.00 orders=   299957 avg_qty=4.000
DE   enterprise   revenue=  164674970.00 orders=   299956 avg_qty=4.000

Three plans

Stock Spark 3.5.9. Plan IDs, full filter lists and file locations are trimmed in all three plans:

*(4) Sort [revenue#41 DESC NULLS LAST], true, 0
+- AQEShuffleRead coalesced
   +- ShuffleQueryStage 2
      +- Exchange rangepartitioning(revenue#41 DESC NULLS LAST, 8)
         +- *(3) HashAggregate(keys=[country#13, segment#14], ...)
            +- AQEShuffleRead coalesced
               +- ShuffleQueryStage 1
                  +- Exchange hashpartitioning(country#13, segment#14, 8)
                     +- *(2) HashAggregate(keys=[...], ...partial_sum...)
                        +- *(2) Project [amount#3, quantity#4, country#13, ...]
                           +- *(2) BroadcastHashJoin [customer_id#1L], ...
                              :- *(2) Project [customer_id#1L, amount#3, ...]
                              :  +- *(2) Filter (... (status#2 = delivered) ...)
                              :     +- *(2) ColumnarToRow
                              :        +- FileScan parquet [...]
                              +- BroadcastQueryStage 0
                                 +- BroadcastExchange HashedRelation...
                                    +- *(1) Filter isnotnull(customer_id#12L)
                                       +- *(1) ColumnarToRow
                                          +- FileScan parquet [...]

Comet:

*(1) CometColumnarToRow
+- CometSort [...], [revenue#41 DESC NULLS LAST]
   +- AQEShuffleRead coalesced
      +- ShuffleQueryStage 2
         +- CometExchange rangepartitioning(...), CometNativeShuffle
            +- CometHashAggregate [...], [Final], [country#13, segment#14], ...
               +- AQEShuffleRead coalesced
                  +- ShuffleQueryStage 1
                     +- CometExchange hashpartitioning(...), CometNativeShuffle
                        +- CometHashAggregate [...], [Partial], ...
                           +- CometProject [amount#3, quantity#4, ...]
                              +- CometBroadcastHashJoin [customer_id#1L], ...
                                 :- CometProject [customer_id#1L, amount#3, ...]
                                 :  +- CometFilter [...], (... = delivered ...)
                                 :     +- CometNativeScan parquet [...]
                                 +- BroadcastQueryStage 0
                                    +- CometBroadcastExchange [...]
                                       +- CometFilter [...], isnotnull(...)
                                          +- CometNativeScan parquet [...]

Gluten, with indentation compressed to two spaces per level so the operator names fit:

VeloxColumnarToRow
+- ^(4) SortExecTransformer [revenue#41 DESC NULLS LAST], true, 0
 +- ^(4) InputIteratorTransformer[...]
  +- AQEShuffleRead coalesced
   +- ShuffleQueryStage 2
    +- ColumnarExchange rangepartitioning(...), [shuffle_writer_type=hash]
     +- VeloxResizeBatches 1024, 2147483647, 10485760
      +- ^(3) HashAggregateTransformer(keys=[country#13, segment#14], ...)
       +- ^(3) InputIteratorTransformer[...]
        +- AQEShuffleRead coalesced
         +- ShuffleQueryStage 1
          +- ColumnarExchange hashpartitioning(country#13, segment#14, 8)
           +- VeloxResizeBatches 1024, 2147483647, 10485760
            +- ^(2) ProjectExecTransformer [hash(country, segment, 42) ...]
             +- ^(2) FlushableHashAggregateTransformer(... partial_sum ...)
              +- ^(2) ProjectExecTransformer [amount#3, quantity#4, ...]
               +- ^(2) BroadcastHashJoinExecTransformer [customer_id#1L], ...
                :- ^(2) ProjectExecTransformer [customer_id#1L, ...]
                :  +- ^(2) FilterExecTransformer (... = delivered ...)
                :     +- ^(2) FileFileSourceScanExecTransformer parquet
                +- ^(2) InputIteratorTransformer[customer_id#12L, ...]
                   +- BroadcastQueryStage 0
                      +- ColumnarBroadcastExchange HashedRelationBroadcastMode
                         +- ^(1) FilterExecTransformer isnotnull(...)
                            +- ^(1) FileFileSourceScanExecTransformer parquet

Operator for operator:

The last row is the point. In the stock plan, the conversion to rows sits directly above the scans, so everything from the filter upward is row-at-a-time. In both accelerated plans, that conversion has moved to the very top. The whole pipeline underneath stays columnar, and the only rows left are the fifteen going back to the driver.

Two Gluten-specific nodes need explaining. InputIteratorTransformer is the bridge that gives a ^(n) stage its input from outside itself, here a shuffle or a broadcast. On the native side it becomes a ValueStreamNode, which is why a Velox error names a value-stream plan node. FlushableHashAggregateTransformer is the partial aggregation. Gluten’s source describes it as a hash aggregation that “emits pre-aggregated data which allows duplications on grouping keys”. It may flush what it holds in memory before reading all its input. That is safe only because a shuffle and a final aggregate follow.

Real-world scenarios and production failures

The job that ran on stock Spark while claiming acceleration

Drop the two spark.memory.offHeap.* settings and launch each engine again.

Comet:

WARN CometDriverPlugin: Comet plugin is disabled because Spark is not running in off-heap mode.

Then the job runs to completion on stock Spark, exits 0 and returns correct results. The plan contains no Comet operator. One WARN among thousands of INFO lines is the only signal. A benchmark run this way compares stock Spark with stock Spark and reports that Comet “made no difference”.

Gluten:

ERROR SparkContext: Error initializing SparkContext.
org.apache.gluten.exception.GlutenException: Must set 'spark.memory.offHeap.enabled' to true and set 'spark.memory.offHeap.size' to be greater than 1MB

No SparkContext, no job, exit code 1. Gluten turns a misconfiguration into an outage, and Comet turns it into a silent regression. Which you prefer depends on whether your scheduler pages someone on a failed job, but the Comet behaviour needs a plan assertion in CI to catch it.

Comet has an escape hatch Gluten does not: spark.comet.exec.onHeap.enabled=true runs Comet without off-heap memory. Comet’s configuration guide describes it as the mode Spark’s own SQL tests need.

Gluten on Java 17: the error that points at Velox

Launch Gluten without -Dio.netty.tryReflectionSetAccessible=true and the SparkContext starts normally. The job then fails on its first scan with a stack trace that begins in Velox’s C++ driver loop:

E20260930 ... Exceptions.h:87] Line: .../velox/exec/Driver.cpp:690,
  Function:operator(), Expression: Operator::getOutput failed for
  [operator: TableScan, plan node ID: value-stream:0]: Error during calling
  Java code from native code: ... java.lang.UnsupportedOperationException:
  sun.misc.Unsafe or java.nio.DirectByteBuffer.<init>(long, int) not available

The native frame is a distraction. The cause is the last line. On Java 9 and later, Netty, which Arrow’s Java memory code uses, cannot reach DirectByteBuffer’s private constructor unless reflection is allowed. Gluten’s developer guide names the flag, citing SPARK-29924 and ARROW-6206. Comet 1.0.0 needed no such flag on the same image.

The version warning that is only a warning

Every Gluten run on this stack logged:

WARN SparkShimProvider: Spark runtime version 3.5.9 is not matched with Gluten's fully tested version 3.5.5

The build banner explains it: the 1.7.0 Spark 3.5 bundle was built against Spark 3.5.5, and Gluten names one fully tested patch release for each Spark minor line. Every query in this post returned correct results on 3.5.9. But “not fully tested” means exactly that. In a regulated pipeline, either pin Spark to the tested patch, or run your own result comparison on each Spark upgrade before turning Gluten on for that job.

One Python UDF, two very different plans

Add a Python UDF in the middle of the pipeline. Save this as risk_bands.py:

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.types import StringType

spark = (SparkSession.builder.appName("accel-demo-fallback")
         .config("spark.ui.enabled", "false")
         .config("spark.eventLog.enabled", "false")
         .config("spark.sql.shuffle.partitions", "8")
         .getOrCreate())

orders = spark.read.parquet("file:///tmp/accel_demo/orders")
customers = spark.read.parquet("file:///tmp/accel_demo/customers")


@F.udf(returnType=StringType())
def risk_band(country, amount):
    if country in ("IN", "BR") and amount > 800:
        return "review"
    return "clear"


flagged = (orders.where(F.col("status") == "delivered")
           .join(customers, "customer_id")
           .withColumn("band", risk_band(F.col("country"), F.col("amount")))
           .groupBy("country", "band").count()
           .orderBy("country", "band"))

for r in flagged.collect():
    print(r["country"], r["band"], r["count"])
print("=== PHYSICAL PLAN ===")
print(flagged._jdf.queryExecution().executedPlan().toString())
spark.stop()

Run it with the same three launch commands, replacing the script name. Add --conf spark.comet.explain.fallback.enabled=true to the Comet run. All three return the same seven rows: IN and BR each get a clear and a review row, and the other three countries only clear. They start:

BR clear 802409
BR review 197591
DE clear 1000000

Gluten reports the fallback without being asked:

WARN GlutenFallbackReporter: Validation failed for plan: BatchEvalPython[QueryId=0], due to:
 - Validation failed with exception from: EvalPythonExecTransformer, reason: Not supported python udf: risk_band(country#13, amount#3)#31.

Comet reports it once the flag is on:

WARN CometExecRule: Comet cannot execute some parts of this plan natively (set spark.comet.explain.fallback.enabled=false to disable this logging):
CometSort
+- CometSinkPlaceHolder
   +- CometColumnarExchange
      +-  HashAggregate [COMET: Comet aggregate that merges intermediate buffers requires a Comet child aggregate when the intermediate buffer formats are incompatible with Spark. Incompatible aggregate function(s): count]
         +- CometSinkPlaceHolder
            +- CometColumnarExchange
               +- HashAggregate
                  +- Project
                     +-  BatchEvalPython [COMET: BatchEvalPython is not supported]
                        +- CometProject
                           +- CometBroadcastHashJoin
INFO CometExecRule: Reverting Comet columnar shuffle to Spark shuffle between HashAggregateExec and HashAggregateExec (no Comet operator on either side to consume columnar output)

Now the plans. Comet, from the UDF upward:

*(4) CometColumnarToRow
+- CometSort [...]
   +- CometColumnarExchange rangepartitioning(...), CometColumnarShuffle
      +- *(3) HashAggregate(keys=[country#13, band#32], functions=[count(1)])
         +- Exchange hashpartitioning(country#13, band#32, 8)
            +- *(2) HashAggregate(keys=[...], functions=[partial_count(1)])
               +- *(2) Project [country#13, pythonUDF0#58 AS band#32]
                  +- BatchEvalPython [risk_band(country#13, amount#3)#31]
                     +- *(1) CometColumnarToRow
                        +- CometProject [amount#3, country#13]
                           +- CometBroadcastHashJoin [customer_id#1L], ...

Gluten, from the UDF upward:

VeloxColumnarToRow
+- ^(5) SortExecTransformer [...]
   +- ColumnarExchange rangepartitioning(...)
      +- ^(4) HashAggregateTransformer(keys=[country#13, band#32], ...)
         +- ColumnarExchange hashpartitioning(country#13, band#32, 8)
            +- ^(3) FlushableHashAggregateTransformer(... partial_count(1))
               +- ^(3) ProjectExecTransformer [country#13, pythonUDF0#58 ...]
                  +- ^(3) InputIteratorTransformer[...]
                     +- RowToVeloxColumnar
                        +- BatchEvalPython [risk_band(country#13, amount#3)#31]
                           +- VeloxColumnarToRow
                              +- ^(2) ProjectExecTransformer [amount#3, ...]
                                 +- ^(2) BroadcastHashJoinExec...

Count the JVM operators above the UDF:

Gluten pays for two conversions, columns to rows and back, around the one node it cannot run. Comet pays for one conversion and gives up everything up to the next exchange.

I tried to close that gap in Comet. spark.comet.sparkToColumnar.enabled=true with BatchEvalPython added to spark.comet.sparkToColumnar.supportedOperatorList produced exactly the same plan. The option’s default list is Range,InMemoryTableScan,RDDScan,OneRowRelation, leaf-style operators whose output Comet converts to Arrow. It is not a general “resume native above any JVM node” switch.

Either way, the highest-value change is to remove the UDF. F.when((F.col("country").isin("IN", "BR")) & (F.col("amount") > 800), "review") .otherwise("clear") expresses the same logic as a built-in expression. With that one change, both engines ran the whole plan natively, with no JVM operator left, and returned the same seven rows.

Architecture and platform: where each one can run at all

The last row matters for platform teams. Comet can come from your Maven mirror like any other dependency. Gluten needs a step that downloads a release, verifies the .sha512 and places the jar. That is what the demo above does.

Optimisation, tuning knobs and best practices

Choosing between them

Neither table answers the question that decides the choice: how much of your query time is in operators each engine supports. That is a property of your queries, and the plan is where it shows.

Settings that matter

Both engines:

Comet:

Gluten:

Two of those defaults change behaviour you may rely on. forceShuffledHashJoin=true means a join that stock Spark would sort-merge, spilling gracefully, becomes a hash join that builds an in-memory table from one side. Gluten also falls back under ANSI mode by default. That matters on Spark 4, where ANSI mode is on by default, but it is outside this post’s 3.5.9 runs.

Best practices

• Assert on the plan in CI. For Comet, check for Comet operator names. For Gluten, check for Transformer operators under ^( stage markers. Both engines return correct results when they are off, so results alone prove nothing.
• Commit off-heap settings with the plugin settings. With Comet, forgetting them is invisible. With Gluten, it is an outage.
• Read the fallback reasons before tuning. Gluten logs them by default. Comet needs spark.comet.explain.fallback.enabled=true.
• Remove UDFs in the middle of pipelines first. A built-in expression beats any setting, in both engines.
• Pin the Spark patch Gluten was tested against, or run your own result comparison when you upgrade Spark.
• Measure on your own cluster and data. The mechanism is real: fewer conversions and native operators. How much it saves depends on your query mix, and a laptop container measures nothing useful.

Summary and key takeaways

Gluten and Comet solve the same problem in the same place. Both rewrite Spark’s physical plan after Catalyst, run the supported operators natively on Arrow-format batches, and keep Spark’s scheduler, shuffle service and APIs. On a scan, join, aggregate and sort query, both moved the conversion to rows from the bottom of the plan to the very top, and both returned the stock answer.

They differ where production is hardest:

• Fallback: Gluten isolates an unsupported operator between two conversions. Comet 1.0.0 lets it take the aggregates above it back to the JVM.
• Misconfiguration: Gluten fails to start. Comet silently becomes stock Spark.
• Distribution: Comet is a Maven dependency with amd64 and arm64 natives. Gluten is a downloaded bundle for amd64, built against one Spark patch release.

The mental model worth keeping: neither engine makes Spark faster. Each replaces specific operators with native ones. Everything it cannot replace stays as slow as it was, plus the cost of the conversions around it. So the value you get depends on your query shape, not on the plugin, and the physical plan shows it.

Frequently asked questions

Do I need to rewrite queries for Gluten or Comet? No. Both rewrite the physical plan after Catalyst, so DataFrame and SQL code is unchanged. Every query in this post returned the same rows on stock Spark, Comet and Gluten.

How do I prove Gluten is running? Look for *Transformer operators under ^(n) stage markers, and VeloxColumnarToRow at the top of the executed plan. At startup Gluten also logs its build banner, with Gluten Version, the Spark version it was built against and the Velox revision.

How do I prove Comet is running? Look for Comet-prefixed operators in the executed plan. At startup it logs Comet native library version ... initialized once the native library has actually loaded.

Why did Comet make no difference? Most often because off-heap memory was not enabled. Comet logs Comet plugin is disabled because Spark is not running in off-heap mode and steps aside.

Why does Gluten fail with DirectByteBuffer.<init>(long, int) not available? Java 9 and later block the reflective access that Netty needs. Add -Dio.netty.tryReflectionSetAccessible=true to both driver and executor extraJavaOptions, appended to whatever those options already hold.

Does either engine run Python UDFs natively? Not the plain Python UDF in this post. Both fell back on BatchEvalPython. Gluten kept everything around it native; Comet did not.

Can I get Gluten from Maven Central with --packages? No. Gluten publishes release tarballs on the Apache download site; the Velox bundle jar is inside them.

What is Gluten’s ClickHouse backend? A port of ClickHouse, based on 23.1, built as libch.so and loaded by Gluten the same way it loads Velox. It has no release binary and must be built from source.

References

• Apache DataFusion Comet documentation for installation, supported versions and the configuration reference
• Comet plugin overview for QueryPlanSerde, Native.createPlan and the Arrow FFI hand-back
• Apache Gluten README for the Substrait design and the fallback mechanism
• Gluten with Velox for the plugin, shuffle manager and off-heap settings
• Gluten configuration for every spark.gluten.* key and its default
• New to Gluten for the Java 11+ Netty flag
• WholeStageTransformer, which builds one Substrait plan per ^(n) stage
• Gluten with ClickHouse for the libch.so backend
• Velox and Substrait for the engine and the plan format Gluten uses
• Apache Arrow columnar format for the in-memory layout both engines use across JNI
• Catalyst optimizer internals for the plan stages both engines rewrite after
• Spark memory management for the off-heap pool both engines draw on

Trademarks

Apache Spark, Apache Gluten, Apache DataFusion, Apache DataFusion Comet, Apache Arrow, Apache Parquet and Apache are either registered trademarks or trademarks of The Apache Software Foundation in the United States and other countries. ClickHouse is a registered trademark of ClickHouse, Inc.