feat: add native Delta Lake scan contrib module (page/row-group pruning)

[dwsmith1983]

Which issue does this PR close?

Part of #174 (Explore integration with Delta Lake). It does not close #174, that issue also tracks writes, CDF, and broader integration; this PR delivers the native read path.

Supersedes two earlier efforts, and deliberately builds on both (both given co-authored by since ideas were learned and borrowed):

• [Tracking] feat(contrib): Native Delta Lake scan via delta-kernel-rs (Iceberg-style contrib) #4366 (@schenksj) — the delta-kernel-rs-based contrib. This PR keeps its contrib module shape, build gating, decline-gate philosophy, and test scenario catalog, but replaces the kernel data-read path: kernel's parquet reader has no page-index pruning, and its planning duplicates work delta-spark has already done by the time the DSv1 scan reaches Comet.
• feat: support native Comet scan of plain Delta Lake tables #4669 (@adityavaish) — the minimal plain-table native scan. This PR takes the same core insight (reuse Comet's existing native parquet scan; no kernel needed) and extends it to deletion vectors, column mapping, DPP, and the full Spark matrix.

Rationale for this change

Comet currently falls back to Spark's reader for all Delta tables (isFileFormatSupported requires exact ParquetFileFormat, and DeltaParquetFileFormat is a subclass). That forfeits native execution and all of Comet's parquet pruning on one of the most common table formats.

Key observation: delta-spark has already done log replay, snapshot resolution, time travel, and partition pruning by the time CometScanRule sees the FileSourceScanExec. So no Delta planning is needed on the native side at all, the scan can route through the exact same DataFusion ParquetSource path as CometNativeScanExec, inheriting row-group stats pruning, page-index pruning (#5142), and filter pushdown (#4722) for free. The only genuinely Delta-specific native code is deletion-vector decoding: DV bitmaps are decoded into per-file ParquetAccessPlans, which DataFusion intersects with page-index pruning, so deleted rows are skipped in-scan and DV skips compose with page skips.

Local benchmark (20M rows, selective predicate): 1.44x faster than stock Spark 9.4% of bytes read; DV tables at time parity with in-scan DV application.

What changes are included in this PR?

• contrib/delta/ new Maven module behind a -Pdelta profile: scan rule, decline gates, serde, CometDeltaNativeScanExec (split-mode partition serialization, DPP via derived scan helper), ServiceLoader registrations, differential test suites, Delta own-suite regression harness, benchmark script.
• Core (Delta-free, generic) new CometScanRuleExtension SPI + ServiceLoader hook at the top of transformV1Scan; CometNativeScan.convert body extracted into reusable buildNativeScanCommon`.
• Native delta cargo feature: DeltaScan proto + planner arm delegati the shared parquet scan builder; delta_dv.rs for DV blob unframing (CRC verified), roaring decode (portable + native magic), and access-plan construction.
• Supported plain/partitioned tables, deletion vectors (inline + on-disk), column mapping name mode, time travel, checkpoints, OPTIMIZE'd tables, schema evolution + defaults, DPP, INT96, special-character paths.
• Declines (fall back to Spark, tagged for EXPLAIN) CDC reads, column mapping id mode, row-index-consuming plans, unknown reader features, generated columns, encryption, input_file_name().
• Spark matrix: 3.5 / 4.0 / 4.1 (Delta 3.3.2 / 4.0.1 / 4.3.1 — Delta 4.1.0/4.2.0 are binary-incompatible with Spark 4.1.3). Spark 3.4 declines (Delta 2.4 needs shims); spark-4.2 profile is dormant until Delta ships support.

How are these changes tested?

• 36-test differential suite (comet-on, comet-off) covering the DV, column mapping, DPP, schema evolution, and decline-gate matrices green on Spark 3.5/4.0/4.1 cells. Pruning is asserted as hard metrics checks (page_index_rows_pruned > 0, row_groups_pruned_statistics > 0), not benchmark notes.
• Delta's own test suites run with Comet injected (`dev/run-delta-regression.s DeletionVectorsSuite 29/29, TimeTravel/ColumnMapping/DeleteSQL/UpdateSQL 197/197, MergeIntoSQLSuite 664/665 (the one failure is a scan-telemetry count assertio plan-shape artifact; data assertions pass).
• Rust unit tests for DV unframing/decode/access-plan edge cases and the feature-off error path.
• A DML repro suite proving DELETE writes DVs (not rewrites) under the claimed scan, in both useMetadataRowIndex modes.

1 TB benchmark (real S3, this branch)

Independent run on TPC-DS-derived store_sales at 1 TB (2.75 B rows), hilbert-clustered, on S3 (ap-southeast-1). The same physical parquet files are read through three paths, raw parquet (recursive glob), the Delta table, and an Iceberg table registered over the identical files via add_files, so the table-format scan path is isolated on byte-identical data. Spark 3.5.6 standalone, 252 executor cores (Graviton m7g), this branch at 7fd81aa9 built with -Pspark-3.5,delta. Four selective query families x 20 queries each; rows_scanned summed from executed-plan scan metrics; warm = median of 3 runs. Control = same session with Comet disabled and the vectorized reader off (parquet-mr row reader, which prunes pages honestly).

Fraction of rows decoded (Comet-native / control), and Comet bytes read per query:

family	parquet-direct	Delta (this PR)	Iceberg	GB read (pq / delta / ice)
3-dim boxes	0.00593 / 0.00593	0.00593 / 0.00593	0.00091 / 0.09750	0.15 / 0.15 / 0.43
item range	0.10740 / 0.10740	0.10740 / 0.10740	0.02005 / 0.41614	1.63 / 1.63 / 3.17
price range	0.10828 / 0.10828	0.10828 / 0.10828	0.01949 / 0.47716	1.51 / 1.51 / 3.23
store range	0.13856 / 0.13856	0.13856 / 0.13856	0.05048 / 0.43516	1.43 / 1.43 / 2.70

Warm query times, Comet-native vs the row-reader control:

family	parquet-direct	Delta (this PR)	Iceberg
3-dim boxes	1.52 s / 5.51 s	0.95 s / 2.82 s	0.85 s / 1.39 s
item range	1.33 s / 6.62 s	1.48 s / 6.40 s	2.34 s / 4.21 s
price range	1.34 s / 6.78 s	1.52 s / 6.86 s	2.97 s / 4.44 s
store range	1.40 s / 6.81 s	1.47 s / 6.23 s	2.42 s / 4.20 s

Takeaways:

• The native Delta scan adds zero read overhead: rows decoded and bytes read are identical to the raw-parquet native scan on every family, the PR's core design claim (route through the same ParquetSource path, inherit row-group + page-index pruning) holds at 1 TB on real S3.
• No Iceberg regression: the Iceberg-native arm reproduces a prior Comet 1.0.0 run of the same suite to the fourth decimal (boxes 0.00091 vs 0.00091), across different builds and cluster shapes.
• Comparability note: CometIcebergNativeScanExec reports post-filter-pushdown rows in output_rows, so its row fractions are not comparable to the other arms, bytes_scanned is the honest cross-arm metric (Iceberg reads ~2-3x the bytes of the parquet/Delta arms here).
• Honest framing: this workload (highly selective literal predicates, scan-dominated plans) is the least favorable case for Comet, a modern page-honest vectorized reader already prunes these queries to the floor, and against one we estimate rough parity to ~1.3x. The differentiated wins are the capability gaps: scalar-subquery data filters pushed to the reader at execution time (stock Spark 3.x scans 100% there; see the local bench in the description), in-scan DV application, and native exec on compute-heavy plans.

Deployment note for anyone staging this on a standalone cluster: core discovers the contrib via ServiceLoader on its own classloader, and the contrib links Delta types from that same loader — so when comet-spark rides spark.{driver,executor}.extraClassPath (required for CometShuffleManager), the contrib jar and delta-spark/delta-storage must be real files on that same classpath; --packages jars land in Spark's child loader where neither lookup can see them.

---

Co-authored-by: Scott Schenkein schenksj@yahoo.com
Co-authored-by: Aditya Vaish adivaish@microsoft.com

[dwsmith1983]

Update: pushed two follow-up commits extending the scan's pruning and object-store behavior.

perf: fetch Delta deletion vectors and footers concurrently DV blob and footer reads were sequential: two serial round-trips per DV'd file before the scan could start, which scales badly on object stores. They now fetch with a bounded fan-out of 8, preserving file order and fail-fast error semantics. Covered by a new end-to-end unit test (inline DVs, on-disk DVs, pass-through files, exact row selections, output ordering).

feat: push resolved scalar-subquery filters into the native Delta scan predicates like id >= (SELECT max(ts) FROM checkpoint) previously contributed nothing to the native scan: subquery results don't exist at planning, so the scan
decoded the full table and Spark's covering FilterExec did all the filtering. They are now resolved at execution time and appended as pushed filters, so row-group and page-index pruning fire the same as for literal bounds. Three version-specific traps handled:

1. Spark 3.x strips subquery predicates from a scan's dataFilters (FileSourceStrategy); Spark 4.x keeps them. The contrib harvests them from the covering FilterExec at claim time and dedups, so both behaviors converge.
2. The DV plan shape interposes nodes between the filter and the scan, so the harvest matches the nearest filter above the scan, guarded by references scan output.
3. MergeScalarSubqueries fuses multiple scalar subqueries into one struct-returning subquery accessed via GetStructField; that subtree is folded to a literal before serialization.

[dwsmith1983]

HI @andygrove,

Can you review this as it adds Delta functionality?

[sunchao]

Hi @dwsmith1983 Thanks for putting this together! We are also actively looking at Delta support for Comet, and it'd be great if we can collaborate on this effort!

Since #4952 is already approved and close to landing, what do you think about using it as the shared foundation for this work? Ideally, the same contrib infrastructure could support both JVM-planned Delta scans and the Rust Kernel-based approach, with this PR providing the JVM-planned path. We have related work in progress, so it would be good to converge on one implementation.

In addition, would it also make sense to land this in smaller pieces, for easier review and iterating? For example:

• Basic native Delta reads, including time travel and fallback for unsupported features
• Column mapping and schema evolution
• Deletion vectors
• Row tracking
• Change Data Feed

Starting to support this in Spark 4 & Delta 4 would be a useful first milestone. Curious how you see the relationship between the two PRs and whether that direction makes sense to you. Thanks.

[dwsmith1983]

Hi @sunchao,

On #4952 as the foundation: we already share more than it might look like. This PR builds on part 1 of that same breakup (#4700's CometScanWithPlanData / PlanDataInjector SPI) and keeps #4366's contrib shape, decline-gate philosophy, and test catalog, with co-authored-by credit to both earlier efforts. The remaining overlap is contrib infrastructure, and I'm glad to reconcile it once #4952 lands: adopt its contrib-delta profile and feature naming, the per-Spark delta.version matrix, the verify-gate script, and unify the proto slot (this PR is at 119, #4952 at 118). For the claim hook I'd suggest the generic CometScanRuleExtension SPI from this PR, since it keeps core free of Delta-specific code and the kernel path can register through it the same way.

I do see the two read paths as different layers rather than one thing to converge on. By the time CometScanRule sees the scan, delta-spark has already done log replay, time travel, and partition pruning, so this path reuses Comet's existing native parquet scan and gets row-group pruning, page-index pruning, and filter pushdown for free. DVs become ParquetAccessPlans that DataFusion intersects with page-index pruning, so DV skips and page skips compose in one scan. As far as I know no vectorized Delta reader does all of that today, including kernel's, which has no page-index pruning. I'd want convergence to keep this as the default read path, with the kernel path covering what JVM planning can't reach (DSv2, non-Spark frontends, likely CDF and row tracking).

On splitting: I'd push back on slicing by feature, for two reasons. First, the features aren't independent. Several decline gates only exist because DVs, column mapping, and Delta's own suites ran together. For example, Delta's findTouchedFiles scan looks like a plain read, and if a basic-reads slice claims it, DELETE silently rewrites files instead of writing DVs. Second, the proof is holistic: this branch runs Delta's own suites at 1156/1156 and the contrib suites at 39/39 on Spark 3.5, 4.0, and 4.1. Feature slices would decline most tables and couldn't run that meaningfully. What I can do is split along review surfaces instead: core SPI additions, native DV decode with its unit tests, the contrib module and read path, and the regression harness and CI, keeping the read path itself (DVs, column mapping, gates) as one reviewable unit. If it lands whole, Comet ships the only vectorized Delta reader with complete skipping.

The Spark 4 milestone is already met, the suites are green on 4.0 and 4.1 today. Row tracking and CDF are out of scope here and seem like a natural place for the kernel work to lead. Happy to set up a chat with you and @schenksj to work out the details.

[sunchao]

Thanks @dwsmith1983 Your proposed split by review surface sounds reasonable. I agree that the reader, its safety gates, and the essential DML/fallback tests should stay together. Thanks also for being open to aligning with #4952 once it lands. We can leave row tracking and CDF for later discussions rather than expand this PR’s scope. The main additional point I’d like us to settle is keeping experimental Delta support explicitly opt-in.

[dwsmith1983]

Thanks @dwsmith1983 Your proposed split by review surface sounds reasonable. I agree that the reader, its safety gates, and the essential DML/fallback tests should stay together. Thanks also for being open to aligning with #4952 once it lands. We can leave row tracking and CDF for later discussions rather than expand this PR’s scope. The main additional point I’d like us to settle is keeping experimental Delta support explicitly opt-in.

@sunchao

Yeah, agreed on explicit opt-in. It's mostly already set up that way. All the Delta code lives in a separate comet-contrib-delta jar that never gets bundled into comet-spark, so a stock Comet install has no Delta surface at all. If we publish that jar with releases, trying it out is just --packages and a conf, nobody has to build from source. Right now the conf defaults to on when the jar is present though, so I'll flip spark.comet.scan.delta.enabled to default false to make the opt-in explicit.

The one spot where I'd differ from #4952's gate is the native binary. The Delta bits in libcomet are tiny (DV decoding plus a hand-off to the existing parquet scan, no delta-kernel dependency) and can't be reached without the jar and the conf. I'd rather keep them in the default build than make people compile their own native binary to try an experimental feature. Sound reasonable?