feat: Native Delta Lake scan via delta-kernel-rs

[schenksj]

Summary

Part of #174

Adds native Delta Lake read support to Comet using delta-kernel-rs for log replay, matching all optimizations in the existing Iceberg native scan path. Delta tables (spark.sql("SELECT ... FROM delta.\/path`")) now execute through CometDeltaNativeScanExec→ protobufDeltaScan→ Rust planner → Comet's tunedParquetSource`, preserving every Comet Parquet-read optimization (parallel I/O, range merging, page-index filtering, schema adapter for Spark semantics).

Also adds a Delta regression suite mirroring the existing Iceberg one (clone upstream Delta, apply a Comet diff, run Delta's own tests with Comet enabled) — that suite immediately surfaced two latent Comet bugs, both fixed here.

Design

Architecture

Driver (Scala)                          Executor (Rust)
─────────────────                       ─────────────────
CometScanRule                           OpStruct::DeltaScan match arm
  └─ detect DeltaParquetFileFormat        └─ deserialize DeltaScanCommon
  └─ stripDeltaDvWrappers                 └─ apply column mapping to data_schema
  └─ nativeDeltaScan validation           └─ rewrite filters (ColumnMappingFilterRewriter)
                                          └─ build PartitionedFiles from tasks
CometExecRule                             └─ split by DV presence
  └─ CometDeltaNativeScan.convert()       └─ init_datasource_exec (ParquetSource)
     └─ JNI: Native.planDeltaScan()       └─ wrap with DeltaDvFilterExec (if DVs)
        └─ delta-kernel-rs log replay
        └─ DV materialization
        └─ column mapping extraction
     └─ partition pruning (static)
     └─ serialize DeltaScanCommon proto

CometDeltaNativeScanExec
  └─ doPrepare() (DPP subqueries)
  └─ serializedPartitionData (lazy)
     └─ apply DPP filters
     └─ per-file split-mode serialization
  └─ DeltaPlanDataInjector (LRU-cached)

Key design decisions

1. Kernel on driver, ParquetSource on executors — delta-kernel-rs handles log replay + file enumeration once on the driver via JNI. Data reads go through Comet's existing ParquetSource (not kernel's ArrowReader), inheriting all Comet optimizations.
2. Arrow version isolation — kernel pins arrow-57 / object_store-0.12 internally; Comet uses arrow-58 / object_store-0.13. Only plain Rust types (String, HashMap, Vec<u64>) cross the boundary. Both arrow versions coexist in the dep tree without conflict.
3. Detection by class name — DeltaReflection uses string-based class name matching (no compile-time dep on spark-delta), same pattern as Iceberg's SparkBatchQueryScan detection.
4. DV handling via plan-tree rewrite — Delta's PreprocessTableWithDVs Catalyst strategy injects synthetic __delta_internal_is_row_deleted columns. stripDeltaDvWrappers undoes this at scan-rule time, and CometDeltaDvConfigRule disables the incompatible useMetadataRowIndex strategy automatically.

Capabilities

Phases implemented

Phase	Feature	Status
0	Dependency spike (delta_kernel + object_store + roaring)	✅
1	Read-only happy path (unpartitioned, no DV, no column mapping)	✅
2	Predicate pushdown (Catalyst → kernel predicate translation, stats-based file pruning)	✅
3	Deletion vectors (inline + on-disk, materialized on driver, applied via DeltaDvFilterExec)	✅
4	Column mapping (mode=id and mode=name, schema evolution with missing columns)	✅
5	Split-mode serialization, per-file parallelism, partition pruning	✅
5b	Dynamic Partition Pruning (DPP via doPrepare + deferred task filtering)	✅
6	Reader-feature gate (unsupported features → tagged fallback, not silent wrong results)	✅

Supported Delta features

• Partitioned and unpartitioned tables
• Schema evolution (mergeSchema, missing columns → null)
• Time travel (VERSION AS OF, TIMESTAMP AS OF)
• Column mapping modes: none, id, name (including rename after write)
• Deletion vectors (inline bitmaps + on-disk UUID files)
• Stats-based file pruning via kernel predicates
• Data filter pushdown into ParquetSource
• Dynamic partition pruning through joins
• Multi-column partitioning with typed columns (int, long, date, string, etc.)
• Complex types (array, map, struct, deeply nested)
• Cloud storage (S3/S3A, Azure ABFSS, GCS, local filesystem)
• Protocol feature gating (rowTracking, typeWidening → graceful fallback)

Configuration

Config	Default	Description
spark.comet.scan.deltaNative.enabled	false	Enable native Delta scan
spark.comet.scan.deltaNative.dataFileConcurrencyLimit	1	Concurrent file reads per task (2-8 suggested)
spark.comet.scan.deltaNative.fallbackOnUnsupportedFeature	true	Fallback to Spark on unsupported reader features

Iceberg parity

Every optimization in Comet's Iceberg path has a Delta equivalent:

#	Feature	Iceberg	Delta
1	Split-mode serialization	lazy val + IcebergPlanDataInjector	lazy val + DeltaPlanDataInjector
2	DPP support	doPrepare() + SubqueryAdaptiveBroadcastExec	doPrepare() + applyDppFilters()
3	LRU cache in PlanDataInjector	16-entry synchronized LinkedHashMap	identical
4	ImmutableSQLMetric	prevents accumulator merge overwrites	identical
5	Planning metrics	Iceberg V2 custom metrics	total_files, dv_files
6	Runtime metrics	output_rows, num_splits	output_rows, num_splits
7	doExecuteColumnar()	explicit CometExecRDD	identical
8	convertBlock()	preserves @transient fields	identical
9	Filesystem scheme validation	9 schemes	same 9 schemes
10	Schema adapter	SparkPhysicalExprAdapterFactory	same adapter
11	Delete handling	iceberg-rust ArrowReader MOR	DeltaDvFilterExec per-batch masking
12	Config gating	COMET_ICEBERG_NATIVE_ENABLED	COMET_DELTA_NATIVE_ENABLED
13	Feature fallback	format version check	kernel unsupported_features gate
14	Cloud credentials	Hadoop→Iceberg key mapping	Hadoop→kernel dual-key lookup

Intentional differences (by design, not gaps):

• Rust execution — Iceberg uses dedicated IcebergScanExec with iceberg-rust ArrowReader; Delta reuses init_datasource_exec → Comet's ParquetSource (gets parallel I/O and range merging for free).
• Proto dedup pools — Iceberg has 8 deduplication pools for repeated schemas/partitions; Delta tasks are simpler and don't need pools.
• Scan rule validation depth — Iceberg validates 11+ conditions via reflection; Delta delegates most to kernel's built-in validation.

New files

File	Purpose
native/core/src/delta/mod.rs	Module root, quarantine documentation
native/core/src/delta/scan.rs	plan_delta_scan_with_predicate() — kernel log replay
native/core/src/delta/engine.rs	DeltaStorageConfig + create_engine() (S3/Azure/local)
native/core/src/delta/jni.rs	Java_org_apache_comet_Native_planDeltaScan JNI entry
native/core/src/delta/predicate.rs	Catalyst → kernel predicate translator
native/core/src/delta/error.rs	DeltaError enum
native/core/src/execution/operators/delta_dv_filter.rs	DeltaDvFilterExec — per-batch DV row masking
spark/.../CometDeltaNativeScanExec.scala	Split-mode exec with DPP, metrics, lazy serialization
spark/.../CometDeltaNativeScan.scala	Serde: JNI call, partition pruning, proto construction
spark/.../DeltaReflection.scala	Class-name detection, table root/version extraction
spark/.../CometDeltaDvConfigRule	Auto-configures useMetadataRowIndex=false

Delta regression suite

Clones Delta Lake at a pinned tag, applies a Comet diff, and runs Delta's own test suite with Comet enabled — mirroring dev/diffs/iceberg/. Catches compatibility regressions at the plan-rewrite and execution layers that CometDeltaNativeSuite alone can't, because Delta's own tests cover a far broader range of scenarios (time travel, DML, CDC, streaming, etc.).

Matrix: Delta 2.4.0 (Spark 3.4), 3.3.2 (Spark 3.5), and 4.0.0 (Spark 4.0) on Java 17.

What was added

• dev/diffs/delta/{2.4.0,3.3.2,4.0.0}.diff — version-specific patches wiring Comet into Delta's test SparkSession
• .github/actions/setup-delta-builder/ — reusable composite action (clone + apply diff)
• .github/workflows/delta_regression_test.yml — CI matrix across the three combos
• dev/run-delta-regression.sh — single-command end-to-end local runner
• CometSmokeTest.scala (added via the diff) — asserts the Comet plugin is registered AND that Comet operators appear in a Delta query's executed plan; runs first in CI as a fail-fast guard against silent config drift

Bugs surfaced and fixed

1. URI parsing of input file paths (c97b60e). CometScanRule.nativeDeltaScan passed raw file paths to new java.net.URI(f), which threw URISyntaxException on paths with unescaped %, spaces, or other characters invalid in a raw URI. Delta's test framework inserts test%file%prefix- into filenames and tripped it, but the same code path would break for production users with % or spaces in their S3 object keys. Fixed by parsing through org.apache.hadoop.fs.Path, which handles URI escaping correctly.
2. Time-travel self-join incorrectly merged (29361d6). Two CometDeltaNativeScanExec instances reading the same Delta path at different snapshot versions were treated as equal by Spark's ReuseExchangeAndSubquery rule, so v1's exchange was replaced by ReusedExchange of v0's. A full-outer join on key between the two versions then read v0's file list on both sides, dropping unmatched rows. findAllPlanData's .toMap had the same collision. Fixed by deriving a per-scan sourceKey from the DeltaScanCommon proto (includes snapshot_version) and using it as the map key + including it in equals/hashCode, mirroring the pattern CometNativeScanExec already uses.

Running locally

dev/run-delta-regression.sh 3.3.2            # smoke test (~90s)
dev/run-delta-regression.sh 3.3.2 DeltaTimeTravelSuite
dev/run-delta-regression.sh 3.3.2 full

Test plan

• CometDeltaNativeSuite (26) — core reads, projections, filters, partitioning, schema evolution, time travel, complex types, primitive coverage
• CometDeltaColumnMappingSuite (5) — column mapping (name/id), deletion vectors, DV + column mapping, column mapping + schema evolution
• CometDeltaAdvancedSuite (11) — joins, aggregations, unions, window functions, DPP, DPP file pruning, planning metrics, filesystem scheme validation
• CometFuzzDeltaSuite — property-based testing with random schemas
• DeltaReadFromS3Suite — MinIO-based S3 integration tests
• All 82 Comet-side Delta tests passing (Spark 3.5)
• Delta regression suite — Comet's smoke test + DeltaTimeTravelSuite pass end-to-end on Spark 3.5 / Delta 3.3.2 (24/24) and Spark 3.4 / Delta 2.4.0 (23/23); Spark 4.0 covered by CI

Follow-up: TPC-DS plan stability golden files

This PR adds a SCAN_NATIVE_DELTA_COMPAT scan implementation constant and the infrastructure to support it, but does not include the TPC-DS plan stability golden files (q*.native_delta_compat/ under spark/src/test/resources/tpcds-plan-stability/). Generating them produces ~810 files (135 queries × 6 profile roots) which would drown this PR, so they'll land as a separate follow-up. Procedure: create the TPC-DS dataset as Delta tables via benchmarks/tpc/create-delta-tables.py, run CometPlanStabilitySuite with COMET_NATIVE_SCAN_IMPL=native_delta_compat to emit plans, then commit the fixture files.

🤖 Generated with Claude Code

[andygrove]

Thanks @schenksj. Could you fix the linter issues (see contributor guide for instructions).

Thanks for acknowledging that this was written by AI. This is a very large PR for a significant new feature. Adding support for Delta Lake certainly has value, but we need to consider who is going to maintain this code going forward. I am concerned that if we merge this and then there are changes in the delta-lake-rs dependency in the future then it could cause an extra maintenance burden on the existing maintainers, who are more focused on Iceberg support and have been contributing to Iceberg as well.

Could you tell me more about the motivation for this work? Do you have any suggestions for how this could be maintained in the future?

[schenksj]

Thanks for acknowledging that this was written by AI. This is a very large PR for a significant new feature. Adding support for Delta Lake certainly has value, but we need to consider who is going to maintain this code going forward. I am concerned that if we merge this and then there are changes in the delta-lake-rs dependency in the future then it could cause an extra maintenance burden on the existing maintainers, who are more focused on Iceberg support and have been contributing to Iceberg as well.

Could you tell me more about the motivation for this work? Do you have any suggestions for how this could be maintained in the future?

---

Hi Andy,

First, thanks for the quick response. I appreciate it. On the AI side, I think its better to use best tools available and be honest about our processes so that we can mature our practices and focus as an industry. To address your questions...

The motivation on my side is that my day-job employer is a significant user of Delta, and I find the current state and future direction of Delta Uniform, particularly its openness, a bit unclear. It is important for us to preserve vendor flexibility within our Spark stacks, and having a viable accelerator outside of Databricks is a key part of that. This work is a step in that direction.

From a maintainability perspective, I have a couple of thoughts. The design of this PR intentionally minimizes direct reliance on delta-rs by using the kernel only for scan planning, not execution. It also has fairly extensive test cases to detect regressions, but as you know that has its own limitations. As long as Comet continues to directly support Parquet, this approach should remain relatively stable over time.

That said, there is an opportunity to move toward a more pluggable architecture. For example, a third-party library, such as a Delta or Hudi provider, could implement a native scan planning interface exposed by Comet. This would allow dependencies and integrations to be fully externalized and would shift the maintenance burden to the plugin owner.

Longer term, I would like to see IndexTables and Comet become compatible to help accelerate joins and such on plain spark. Achieving that would likely require a more robust plugin model that supports not just scan planning, but also FFI-based columnar streaming. That is a more involved effort and likely a ways out, given the current state of my codebase.

Love your thoughts, and of course no hard feelings if this doesn't align with where you want to focus your product.

---

[andygrove]

First, thanks for the quick response. I appreciate it. On the AI side, I think its better to use best tools available and be honest about our processes so that we can mature our practices and focus as an industry. To address your questions...

Agreed. I use AI extensively. The main challenge for this project that the contribution velocity exceeds review capacity.

The motivation on my side is that my day-job employer is a significant user of Delta, and I find the current state and future direction of Delta Uniform, particularly its openness, a bit unclear. It is important for us to preserve vendor flexibility within our Spark stacks, and having a viable accelerator outside of Databricks is a key part of that. This work is a step in that direction.

Adding Delta Lake support makes Comet appealing to a wider audience, which hopefully leads to more contributors/maintainers over time.

From a maintainability perspective, I have a couple of thoughts. The design of this PR intentionally minimizes direct reliance on delta-rs by using the kernel only for scan planning, not execution. It also has fairly extensive test cases to detect regressions, but as you know that has its own limitations. As long as Comet continues to directly support Parquet, this approach should remain relatively stable over time.

Makes sense.

That said, there is an opportunity to move toward a more pluggable architecture. For example, a third-party library, such as a Delta or Hudi provider, could implement a native scan planning interface exposed by Comet. This would allow dependencies and integrations to be fully externalized and would shift the maintenance burden to the plugin owner.

Interesting idea. We tried something like this in the past with the Java implementation of Iceberg. It led to some challenges with circular dependencies. It would be worth creating an issue to discuss.

Longer term, I would like to see IndexTables and Comet become compatible to help accelerate joins and such on plain spark. Achieving that would likely require a more robust plugin model that supports not just scan planning, but also FFI-based columnar streaming. That is a more involved effort and likely a ways out, given the current state of my codebase.

Love your thoughts, and of course no hard feelings if this doesn't align with where you want to focus your product.

Oh, it's definitely not my product. Let's see what other maintainers have to say. Adding Delta Lake support would be great for Comet's futures. My concern is just over maintenance going forward. However, the feature is marked as experimental and disabled by default, so the feature could always be removed in the future if we get into a situation where the code is no longer maintained and causing issues.

[mbutrovich]

This is awesome @schenksj! Thank you!

At 6,500 lines, I'd like to take some time to review this one in stages. Without looking too closely at it yet, the first questions that come to mind that I want to look at first:

1. Does Delta have a Spark test suite we can run, similar to what we do with Iceberg's Java library to have confidence in the implementation's correctness?
2. Are there significant downstream dependencies we pick up with this change? We're already struggling a bit with iceberg-rust being locked to specific DataFusion and Arrow-rs versions. Does this bring the same challenge, and would Comet be limited from upgrading until all dependencies are in sync?

Like @andygrove, I am mostly concerned about the maintenance burden. Though perhaps I am more concerned about future Comet changes than I am about maintaining this new delta code. I am imagining future major Comet changes like rewriting our rules to run later to be compatible with AQE improvements in Spark 4.0+, and this delta integration becomes something we have to update or leave behind. I don't think any of this should be disqualifying from a merge, but it's another reason I want to sit with the PR for a bit. I'd like to try to imagine ways we could be possibly boxed in by this code.

Thank you again for the contribution! I am looking forward to digging into it this week.

[schenksj]

This is awesome @schenksj! Thank you!

At 6,500 lines, I'd like to take some time to review this one in stages. Without looking too closely at it yet, the first questions that come to mind that I want to look at first:

1. Does Delta have a Spark test suite we can run, similar to what we do with Iceberg's Java library to have confidence in the implementation's correctness?
2. Are there significant downstream dependencies we pick up with this change? We're already struggling a bit with iceberg-rust being locked to specific DataFusion and Arrow-rs versions. Does this bring the same challenge, and would Comet be limited from upgrading until all dependencies are in sync?

Like @andygrove, I am mostly concerned about the maintenance burden. Though perhaps I am more concerned about future Comet changes than I am about maintaining this new delta code. I am imagining future major Comet changes like rewriting our rules to run later to be compatible with AQE improvements in Spark 4.0+, and this delta integration becomes something we have to update or leave behind. I don't think any of this should be disqualifying from a merge, but it's another reason I want to sit with the PR for a bit. I'd like to try to imagine ways we could be possibly boxed in by this code.

Thank you again for the contribution! I am looking forward to digging into it this week.

Happy to answer any questions you have. Fortunately, I think most of the actual code is test cases.

1. Will look into the whether there is a delta spark acceptance suite like the Iceberg one you mentioned. I did find DAT, but i'm not sure if its actively maintained. I may need to ask my Databricks contacts.
2. Dependency creep... This is using semver-based crate identity, so both versions of arrow co-exist. The tradeoff is binary size but not symbol conflicts (thank you rust). Checking what the link-time optimizer actually drags across. I suspect it will be very small since most of arrow isn't actually used anywhere in this new code.

Downstream Dependencies Added by This PR

Direct Dependencies (Cargo.toml)

Crate	Version	Purpose
delta_kernel	0.19	Delta log replay (scan planning only)
object_store	0.12 (renamed object_store_kernel)	Required by kernel's engine (S3/Azure)
roaring	0.10	Deletion vector bitmap decoding
thiserror	workspace	Error type derive (already in workspace, added to core's deps)

Transitive: Second Versions of Existing Crates (16)

Kernel pins arrow-57 / parquet-57 / object_store-0.12 internally. These coexist alongside Comet's arrow-58 / parquet-58 / object_store-0.13. No types cross the boundary.

arrow, arrow-arith, arrow-array, arrow-buffer, arrow-cast, arrow-csv, arrow-data, arrow-ipc, arrow-json, arrow-ord, arrow-row, arrow-schema, arrow-select, arrow-string,
parquet, object_store

Truly New Crates (10)

Crate	Pulled by	Purpose
delta_kernel	direct	Core dependency
delta_kernel_derive	delta_kernel	Proc macros for kernel
roaring	direct	Bitmap codec for deletion vectors
crc / crc-catalog	delta_kernel	Checksum validation for checkpoints
lz4_flex	delta_kernel	Log entry compression
z85	delta_kernel	Deletion vector encoding
document-features	delta_kernel	Doc generation
litrs	delta_kernel_derive	Literal parsing for proc macros
rustls-pemfile	kernel's rustls engine	TLS certificate loading
crossterm / crossterm_winapi	delta_kernel	Transitive dev dependency

Java/Scala Dependencies

None added to production. Delta-spark is test-scope only (unchanged from before this PR).

[hntd187]

Hi @schenksj, DAT is still actively maintained, we use it in delta-rs to do correctness testing. Feel free to reach out if you'd like to see something specific there. I admit it hasn't been updated in awhile, but we are still actively maintaining it.

[schenksj]

Hi @schenksj, DAT is still actively maintained, we use it in delta-rs to do correctness testing. Feel free to reach out if you'd like to see something specific there. I admit it hasn't been updated in awhile, but we are still actively maintaining it.

Thanks @hntd187 and @mbutrovich , please check out this commit. It models the existing iceberg tests: 3e4b6a0

It seems to work, and naturally found one bug which is fixed in 29361d6

[schenksj]

There are a few extended test failures I need to look into in the regression suite. Putting in draft until done

[andygrove]

There are a few extended test failures I need to look into in the regression suite. Putting in draft until done

Thanks @schenksj. Having these tests running in CI will give us much greater confidence in maintaining this code.

[schenksj]

Full regression re-baseline

A fresh full-regression run (Delta 3.3.2, ~24h with the in-flight macOS host this time) just finished:

• Total: 13,612 tests run
• Passed: 13,555 (was 13,437 at the prior baseline)
• Failed: 57 (was 139)
• Canceled: 5, Ignored: 3,683

Net 82 tests cleared since the previous full run, via:

Commit	Cluster cleared
da19481a	DV-prefix override revert (CheckpointsSuite + entire DeletionVectorsSuite, ~30 tests)
bb35abf0	File-splitting infra (latent — no direct count, lays groundwork for future kernel-path large-file scans)
b1fce4f8	Synthetic-column gate (DeltaParquetFileFormatSuite, 9 tests)
7d8c3b0b	AQE logical-link preservation on Comet exchange wrappers (broad fix — touched IdentityColumnIngestionScalaSuite alone for 18, plus knock-on across MERGE / CDF / streaming-watermark plans)

Remaining 57 fails — top clusters

Cluster	Count	Suites
PredicatePushdown partitions.size === 2 (DV-bypass, partition splitting)	15	DeletionVectorsWithPredicatePushdownSuite
huge table: read/delete… 2B rows with DVs	4	DeletionVectorsSuite (variants)
query with predicates should skip partitions (numFiles metric on Comet exec)	2	DeltaSuite + ColumnMapping variants
optimization not supported - join …	4	OptimizeMetadataOnlyDeltaQuery (Id + Name CM variants)
create table with NOT NULL - check violation through file writing	3	DeltaDDLSuite (Id + Name + Hive variants)
update-metrics / delete-metrics / merge-metrics	~6	UpdateMetricsSuite / DeleteMetricsSuite
streaming progress / numInputRows == 0 (basic, streaming, SC-8810, SC-11561, recreate the reservoir…, initial snapshot ends at base index of next version, allow to change schema…, skip change commits, can delete old files of a snapshot without update, can consume new data without update)	~16	DeltaSourceSuite / DeltaSourceLargeLogSuite
partitioned writing and batch reading (CM variants)	3	DeltaSinkSuite (+ Id/Name CM variants)
column mapping batch scan should detect physical name changes	1	DeltaColumnMappingSuite
explicit id matching	1	DeltaColumnMappingSuite
use TIMESTAMP_NTZ in a partition column	1	DeltaTimestampNTZSuite
Validate that links to docs in DeltaErrors are correct	1	DeltaErrorsSuite (HTTP 301 — environmental)
Misc	balance	various

Next step

Three highest-yield clusters for the next stretch:

1. Streaming progress (~16 tests) — likely a single fix in how CometDeltaNativeScanExec populates numOutputRows / numInputRows. Worth one targeted look.
2. PredicatePushdown DV-bypass (15) — already-traced cluster; needs the DV-fallback to either go through the kernel path (so splitTasks from bb35abf0 activates) or a separate split fix in the fallback path.
3. DDL "NOT NULL violation through file writing" (3) + UpdateMetrics/DeleteMetrics/MergeMetrics (~6) — likely a single scan-metric fix where the test reads numFiles off c.wrapped but the metric lives on the Comet exec.

Going to start with (1) since streaming has the largest single-cluster yield and the fix is most likely localised.

Co-Authored-By: Claude Opus 4.7 (1M context) noreply@anthropic.com