Native Delta Lake Reads Ship in DataFusion Comet 1.2.0. Here's How to Run Them

In August, I wrote about what the native Delta read path does, and in October, about the 58 days it took to merge. Both promised a follow-up once you could run it. Comet 1.2.0 is out, and it’s the first release that can read a Delta table natively, with a small caveat.

That’s why this post is mostly a guide. Part of the feature is in the jar you already download from Maven, and the rest you build yourself. It’s a small build, a couple of minutes in Docker, and after it, I’ll walk through running a Delta table with deletion vectors on Spark 4.1 and reading the plan to see whether the scan actually went native.

Two and a Half Years of Asking

The request is older than any of the code. Chao Sun opened #174, “Explore integration with Delta Lake,” in March 2024, with the idea of building on delta-rs once Comet had a fully native parquet reader. A year later, Andy Grove pointed out that Comet’s move to DataFusion’s DataSourceExec should make that much easier. A user in October 2025 asked whether anyone was working on it, since their company ran Spark on Delta heavily and was weighing a move to plain parquet or Iceberg just to get Comet’s speedup, and the answer was that nobody was.

Then it moved quickly. Scott Schenkein picked it up in April 2026 with a delta-kernel-rs reader (#3932), Aditya Vaish’s #4669 revived an older idea (#3035) that Comet’s own parquet scan could do the reading, and #5365 merged in October. The design changed along the way. Instead of a Rust Delta implementation on the native side, delta-spark does all the Delta planning and Comet reads the parquet files it picked, which the August post explains.

What’s in the Release and What You Build

The reader has two halves, and 1.2.0 handles them differently.

The native half, which decodes deletion vectors and hands Delta’s parquet files to the same DataFusion scan Comet uses for plain parquet, is compiled into every comet-spark 1.2.0 jar on Maven Central. It sits behind a cargo feature that’s on by default, so you don’t need a custom native build. It also does nothing unless the other half is present and you turn it on.

The JVM half is the contrib jar, comet-contrib-delta-spark. It recognizes the Delta scans delta-spark has already planned, decides which ones the native reader can take, and hands the rest to Spark. That jar isn’t published. How contrib jars get built, versioned and released is still an open question for the project (#5882). Until that’s settled, you build it from the release tag.

The contrib is plain Scala that compiles against the released comet-spark jar, though, so a JDK and Maven are all the build needs. You also skip the trap from August, where my own native build on Ubuntu 24.04 refused to load on an older fleet image (GLIBC_2.38 not found) and Comet quietly fell back to Spark. The release jar’s native libraries are built on Ubuntu 20.04, so they load almost anywhere.

Supported pairings in 1.2.0:

For Spark 4.2, the 1.2.0 build uses Delta 4.3.1, which predates Spark 4.2 support (Delta 4.4.0 added that in August) and can’t run on it at all: Delta’s own CREATE TABLE fails before Comet is involved. The contrib itself is fine, and it comes out byte for byte the same whichever Delta it compiles against, so you don’t need to override the Delta version. To try it anyway:

• build with -Pspark-4.2 and spark4.2 in the Dockerfile’s COPY line
• use the spark4.2 names for both Comet jars (the contrib and comet-spark)
• pair them with delta-spark and delta-storage 4.4.1
• run on apache/spark:4.2.0-scala2.13-java17-python3-ubuntu

The demo below then reads through the native scan on Spark 4.2.0 and matches Spark’s reader. That’s one query on my laptop, and nobody has run the contrib’s test suites on this combination, so treat it as unofficial until #7455 moves the build to Delta 4.4.

I’ll use Spark 4.1 with Delta 4.3.1 throughout. For 3.5 or 4.0, change the Maven profile and the jar names to match.

Build the Contrib Jar

You don’t need a JDK or Maven installed, only Docker. Put this in an empty directory as Dockerfile:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

# Builds comet-contrib-delta-spark for Spark 4.1 / Scala 2.13.
FROM maven:3.9-eclipse-temurin-17 AS build
ARG COMET_REF=1.2.0
RUN git clone --depth 1 --branch ${COMET_REF} https://github.com/apache/datafusion-comet.git /comet
WORKDIR /comet
# Explicit goals instead of `package`: any lifecycle run resolves the module's
# test-scoped comet-spark test-jar (even with -Dmaven.test.skip), and that jar
# is not published to Maven Central.
RUN ./mvnw -B -Pspark-4.1,delta -pl contrib/delta-spark \
    resources:resources scala:compile jar:jar

FROM scratch
COPY --from=build /comet/contrib/delta-spark/target/comet-contrib-delta-spark4.1_2.13-*.jar /

Then build it, and the jar lands in out/:

1

docker build --output type=local,dest=out .

It took 84 seconds on my laptop, most of it Maven downloading Spark and Delta. The only Comet artifact it needs is the released comet-spark jar, which it pulls from Maven Central, so the contrib and the jar you’ll run with come from the same release. Keep those two versions in sync: a contrib jar from one Comet version next to comet-spark from another is not something anyone tests.

The comment in the middle explains the one odd line. I started with the obvious mvn package -Dmaven.test.skip=true, and it fails, because Maven resolves every dependency scope the build lifecycle might need before it skips anything, and the contrib’s tests depend on a comet-spark test jar that isn’t published. Calling the three goals directly only resolves what compiling needs. For Spark 3.5 or 4.0, change -Pspark-4.1 and the jar name in the last line.

Run It

You need four jars in one directory: the contrib you just built, comet-spark, and Delta’s two.

1
2
3
4
5
6
7
8

mkdir -p jars && cp out/*.jar jars/
M=https://repo1.maven.org/maven2
curl -fsSLo jars/comet-spark-spark4.1_2.13-1.2.0.jar \
  $M/org/apache/datafusion/comet-spark-spark4.1_2.13/1.2.0/comet-spark-spark4.1_2.13-1.2.0.jar
curl -fsSLo jars/delta-spark_2.13-4.3.1.jar \
  $M/io/delta/delta-spark_2.13/4.3.1/delta-spark_2.13-4.3.1.jar
curl -fsSLo jars/delta-storage-4.3.1.jar \
  $M/io/delta/delta-storage/4.3.1/delta-storage-4.3.1.jar

The demo builds a million-row Delta table with deletion vectors turned on, deletes every tenth row, and runs a filtered aggregate twice: once through the native scan, then again with the Delta scan switched off so Spark’s own reader does the work. It prints both plans and checks that the answers agree. Save it as delta_demo.py:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

import time

from pyspark.sql import SparkSession

TABLE = "delta.`/tmp/delta/sales`"

spark = SparkSession.builder.appName("comet-delta-demo").getOrCreate()

# 1,000,000 rows, deletion vectors on
spark.sql(f"""
    CREATE TABLE {TABLE} USING delta
    TBLPROPERTIES ('delta.enableDeletionVectors' = 'true')
    AS SELECT
        id,
        element_at(array('books', 'games', 'garden', 'music', 'toys'),
                   CAST(id % 5 AS INT) + 1) AS category,
        CAST(id % 1000 AS DOUBLE) * 1.5 AS amount
    FROM range(1000000)
""")

# Soft-delete every tenth row: Delta writes deletion vectors, not new Parquet files
spark.sql(f"DELETE FROM {TABLE} WHERE id % 10 = 0")
metrics = spark.sql(f"DESCRIBE HISTORY {TABLE} LIMIT 1").first().operationMetrics
print("deletion vectors added:", metrics["numDeletionVectorsAdded"],
      "| rows deleted:", metrics["numDeletedRows"])

QUERY = f"""
    SELECT category, count(*) AS orders, round(sum(amount), 2) AS revenue
    FROM {TABLE}
    WHERE amount > 1000
    GROUP BY category
    ORDER BY category
"""


def run(label):
    df = spark.sql(QUERY)
    start = time.perf_counter()
    rows = df.collect()
    print(f"\n=== {label}: {time.perf_counter() - start:.2f}s")
    for r in rows:
        print(r.category, r.orders, r.revenue)
    df.explain()
    return rows


native = run("native Delta scan")
spark.conf.set("spark.comet.scan.delta.enabled", "false")
fallback = run("Spark Delta reader")
assert native == fallback, "results differ"
print("results match")

And run it in the official Spark 4.1.3 image, from the directory holding jars/ and the script:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22

CP=/jars/comet-spark-spark4.1_2.13-1.2.0.jar
CP=$CP:/jars/comet-contrib-delta-spark4.1_2.13-1.2.0.jar
CP=$CP:/jars/delta-spark_2.13-4.3.1.jar
CP=$CP:/jars/delta-storage-4.3.1.jar

docker run --rm --memory=4g --cpus=4 \
  -v "$PWD/jars:/jars:ro" \
  -v "$PWD/delta_demo.py:/opt/delta_demo.py:ro" \
  apache/spark:4.1.3-scala2.13-java17-python3-ubuntu \
  /opt/spark/bin/spark-submit \
    --master 'local[4]' \
    --conf spark.log.level=WARN \
    --conf spark.driver.extraClassPath=$CP \
    --conf spark.executor.extraClassPath=$CP \
    --conf spark.sql.extensions=io.delta.sql.DeltaSparkSessionExtension \
    --conf spark.sql.catalog.spark_catalog=org.apache.spark.sql.delta.catalog.DeltaCatalog \
    --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=2g \
    --conf spark.comet.scan.delta.enabled=true \
    /opt/delta_demo.py

The first two --conf lines after the log level put all four jars on one classpath, the next two are Delta’s usual setup, the next four are Comet’s standard ones from its installation guide, and the last is the switch for the Delta reader. On a cluster, the same settings go in spark-defaults.conf, and the jars need to exist at those paths on every executor.

The delete wrote four deletion vectors, one per data file, since local[4] wrote the table as four files:

deletion vectors added: 4 | rows deleted: 100000

Both runs return the same five rows and the script ends with results match. It also prints a time for each run. Ignore it: at a million rows on a laptop, the first query pays for JVM warm-up and whichever reader goes first looks slow. If you want numbers, the August post has them at 1 TB.

Reading the Plan

The plan is the only reliable way to know which reader ran, because a declined scan doesn’t fail. It runs on Spark, returns correct results, and you’d never know from the output.

Here’s the bottom of the native plan, trimmed (the scan’s column lists run long):

+- CometHashAggregate [category#2030, amount#2031], [Partial], [category#2030], [partial_count(1), partial_sum(amount#2031)]
   +- CometProject [category#2030, amount#2031], [category#2030, amount#2031]
      +- CometFilter [...], (((__delta_internal_is_row_deleted#3309 = 0) AND isnotnull(amount#2031)) AND (amount#2031 > 1000.0))
         +- CometProject [...]
            +- CometDeltaNativeScan parquet [category#2030, amount#2031, __delta_internal_is_row_deleted#3309, ...]

CometDeltaNativeScan is the node to look for. Everything above it starts with Comet, so the filter, the aggregate and the shuffle all run natively too.

The filter on __delta_internal_is_row_deleted = 0 looks like the place deleted rows get removed, and on Spark’s reader, it is. Delta’s planner adds that column and the filter above the scan. The native reader doesn’t need them: it applies the deletion vector as a row selection inside the scan, so deleted rows never leave the parquet reader, and the column arrives as a constant 0. The filter stays in the plan because delta-spark put it there, and on this path, it has nothing to remove.

The same part of the plan with spark.comet.scan.delta.enabled=false:

+- *(1) HashAggregate(keys=[category#3452], functions=[partial_count(1), partial_sum(amount#3453)])
   +- *(1) Project [category#3452, amount#3453]
      +- *(1) Filter (((__delta_internal_is_row_deleted#3596 = 0) AND isnotnull(amount#3453)) AND (amount#3453 > 1000.0))
         +- *(1) Project [...]
            +- *(1) ColumnarToRow
               +- FileScan parquet [...] Format: Parquet, Location: PreparedDeltaFileIndex(1 paths)[file:/tmp/delta/sales], ...

That’s what a fallback looks like: FileScan parquet over a PreparedDeltaFileIndex, with a ColumnarToRow above it. The rest of the plan went with it. The filter, the project, both aggregates and the first shuffle all stayed in Spark. Only the sort and the shuffle feeding it made it back to Comet, and that shuffle is the JVM columnar one (CometColumnarExchange), not the native shuffle the first run used. So losing the scan cost most of the plan, not just the read.

If you see FileScan where you expected the native scan, add --conf spark.comet.explain.fallback.enabled=true and the driver log will say why Comet declined it. The common reasons are a table feature the reader doesn’t support yet, a file whose deletion vector is larger than spark.comet.scan.delta.dv.maxDeletedRowsPerFile (a million rows by default), or the classpath problem below.

Two Things That Will Trip You Up

Don’t load Delta with --packages. It’s the natural way to add Delta to a Spark session, and with Comet, it puts delta-spark in a different classloader from the contrib jar. In August, that meant a NoClassDefFoundError in the middle of query planning. In 1.2.0, it’s quieter, which I’d argue is worse. The contrib catches the error, declines the scan, and Spark reads the table, so the query succeeds on the slow path. The only trace is a warning in the driver log, once per scan:

WARN CometScanContrib: Contrib scan handler org.apache.comet.contrib.delta.DeltaScanContrib failed with java.lang.NoClassDefFoundError, indicating it was built against a different version of Comet's internals than is on the classpath now; declining it and continuing with Comet's built-in handling
java.lang.NoClassDefFoundError: org/apache/spark/sql/delta/DeltaParquetFileFormat

The message blames a Comet version mismatch, but the real cause is the line under it: the contrib can’t see Delta’s classes. Put all four jars on extraClassPath as above and it goes away.

The flag alone does nothing, and so does the jar alone. You need the contrib jar on the classpath and spark.comet.scan.delta.enabled=true. The default is false on purpose, so adding the jar to a shared cluster image doesn’t change anyone’s queries until they opt in.

Since the Merge

Two Delta changes merged after the 1.2 branch was cut on October 23, so they’ll be in 1.3.0. #6932 makes the Delta scan’s plan-data key end with its plan id, as the core and Iceberg keys already do, so two scans of one table that differ only in which files they read can’t be confused for each other. I couldn’t find a query that hits it today, so it closes a gap rather than fixing a failure anyone has seen. Erik Bogado’s #6945 keeps deletion vector bitmaps compressed while decoding them. The old decoder expanded each deleted row, so a contiguous million-row deletion became 131,208 bytes where 230 would do.

The rest of the follow-up work is on file as issues, most of them from the review: sharing more of the admission checks and scan builder with the core parquet path (#5944, #5945), decoding a deletion vector once across the splits of a file (#5655), a capability table for the docs (#5661), and closing the remaining predicate pushdown gaps (#3957). The one that matters most for anyone trying this is #5882, since publishing the jar would make most of this post unnecessary.

Also in 1.2.0

Delta took most of my time this cycle, but a few other fixes of mine are in the release.

• Join output order (#6810). Spark drops a sort above a hash join when the join keeps its streamed side’s order. Comet’s native join didn’t always keep it, so a sorted query could return unsorted rows. Andy fixed the main case in #6835; mine makes Comet fall back for a NOT IN over an ordered input. A third, #7126, does the same for outer joins whose order Comet can’t check, and missed the branch cut by a few hours, so it lands in 1.3.0. This came out of the audit that kept Comet’s in-memory cache opt-in for 1.2.0.
• Lost rows in TPC-DS q64 (#6270). Two scans of the same table with different dynamic partition pruning filters could look identical to adaptive query execution, which then reused one exchange for both, and a branch of the query lost rows.
• Memory retries (#6310, #6403). A race in Spark’s execution memory pool can fail an allocation with key not found: <taskAttemptId> after the task’s entry is dropped. Comet now retries instead of failing the task, and Liang-Chi Hsieh folded the two retries into one helper in #6813. Chao’s fix for Spark itself is apache/spark#58747.

Links

• Apache DataFusion Comet 1.2.0 release post
• Comet 1.2.0 changelog
• Comet Delta Lake guide
• contrib/delta-spark README
• apache/datafusion-comet#5365, the PR
• #5882, publishing the contrib jar
• Comet Falls Back to Spark on Every Delta Table. I Wrote the PR That Fixes It
• 58 Days to Merge: Getting Native Delta Lake Into DataFusion Comet