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

[dwsmith1983]

Which issue does this PR close?

Part of #174. This PR does not close it: the delta-kernel contrib and the convergence discussion in #5411 are tracked there as well.

Rationale for this change

Adds an optional contrib module that plans Delta Lake table scans on the JVM and executes them natively, including deletion vector application inside the native scan. delta-spark has already done log replay, snapshot resolution, and partition pruning by the time CometScanRule sees the FileSourceScanExec, so there is no Delta planning to do natively: the scan reuses the existing ParquetSource path and gets row group pruning, page index pruning, and filter pushdown for free, with deletion vectors composed into the ParquetAccessPlan so DV skips and page skips intersect rather than filtering after the read.

The module is explicit opt in: the -Pdelta Maven profile builds a separate comet-contrib-delta jar that is never bundled into comet-spark, and spark.comet.scan.delta.enabled defaults to false. The delta cargo feature (DV decoding plus the planner hand-off, no delta-kernel dependency, about 82 KB of dylib) stays in the default native build so trying the contrib needs only the jar and the config, not a custom native binary; this was agreed in review and is recorded in the Cargo.toml comment. The adjacent contrib-delta feature is unrelated: it gates the delta-kernel integration and default builds carry no kernel surface.

Restructured after review

Core changes that previously traveled with this PR now live elsewhere:

• JVM-exact case folding for Parquet field names: superseded by feat: support unicode case sensitive field names for reading parquet #5602 on main. This branch's own case table implementation is gone and the Delta scan inherits main's folding through the shared parquet path.
• Object store ${...} expansion, constant metadata field uniquification, and dead JNI removal: fix: expand object store option references, uniquify constant metadata names, drop dead parquet JNI #5653, now merged. The first two are prerequisites of this module and this branch is rebased on top of them.
• Duplicate field and field id matching semantics: fix: match Spark's duplicate field and field id semantics in parquet field lookup #5654, re-derived on top of feat: support unicode case sensitive field names for reading parquet #5602's folding.

Two core-generic capabilities remain in this PR because the native read path does not have them yet and the Delta scan needs them for correctness; both are candidates to lift into core, tracked in #5662 (S3 configuration divergence for the regular native scan) and #5010 (calendar rebasing for the regular native scan):

• S3 configuration divergence gating: Comet's native object store client resolves S3 configuration differently from Hadoop's S3AFileSystem in several ways (bucket precedence in lookupPassword, JCEKS credential aliases, clear text fallback, assumed role session policies, provider class semantics). DeltaScanSupport models each consumer's real resolution, verified against hadoop-aws 3.3.4 and 3.4.1 bytecode, and declines to Spark whenever native would read under a different identity or endpoint. Assumed role session policies (fs.s3a.assumed.role.policy) decline outright since Hadoop sends them in the AssumeRole request and native does not.
• Per file calendar rebasing: the regular native scan ignores legacy calendar metadata (Datetime rebase: track the documented scan limitation, and spark.comet.exceptionOnDatetimeRebase is dead code #5010). The Delta arm resolves date and timestamp rebase policy per file from the parquet writer metadata, mirroring Spark's DataSourceUtils.getRebaseSpec, with the effective session read modes carried in the scan for files without Spark metadata and INT64 and INT96 timestamp columns each attributed to their own spec from the footer's physical types. Dates rebase exactly (Spark's Julian to Gregorian table), UTC writer timestamps rebase exactly, nested struct, list, and map leaves are handled recursively with only the requested leaves checked (an unrequested ancient sibling never blocks a projection), and EXCEPTION mode uses Spark's cutoffs (1582-10-15 for dates, 1900-01-01T00:00:00Z for timestamps). Only two inputs still fail at execution time instead of reading: a LEGACY policy file with a non UTC or unrecorded writer zone when a timestamp before 1900-01-01Z actually appears, and an EXCEPTION policy file (or one whose two legacy flags disagree without physical type attribution) when an ancient value actually appears. Everything else reads natively with Spark's values. Pruning is lost on every column that receives a policy wrapper, including modern only LEGACY files and check only EXCEPTION files, not just values that need conversion. This is gated to the Delta arm so the regular scan's documented behavior is unchanged.

What changes are included in this PR?

• contrib/delta-spark: DeltaScanSupport (scan eligibility, S3 divergence gating, DV descriptor extraction), CometDeltaNativeScan serde, service registration via the contrib scan SPI, documentation.
• Native: delta_dv.rs (deletion vector decode with a full malformed input matrix, and access plan construction), delta_spark_scan.rs planner arm, datetime_rebase.rs, proto messages for the Delta scan envelope, S3 object store helper.
• Shared refactors the module needs: build_parquet_scan_plan/prepare_scan_store_and_files extraction in the planner, object_store_url_key/prepare_object_store_with_config_hash, buildNativeScanCommon extraction, reportScanInputMetrics, hasScanInput widening, contrib LinkageError containment.
• CI: a dedicated delta contrib workflow running the suite on Spark 3.5 and 4.0.

Follow-up work from review is tracked in #5655 (DV file splitting), #5656 (compressed DV decoding), #5657 (overlapping bitmap and footer reads), #5658 (shared cloud compatibility helper), #5659 (credential scoping), #5660 (v2 checkpoint coverage), #5661 (capability table), and #5662.

How are these changes tested?

• The contrib suite (CometDeltaNativeScanSuite, CometDeltaS3Suite against MinIO, CometDeltaDmlReproSuite, DeltaScanContribSuite) passes on both the Spark 3.5 and 4.0 profiles: 236 tests each at the current head, MinIO suite live.
• Native tests pass under --features delta (343 in the core crate), including the DV malformed input matrix (truncation at every boundary, CRC and magic corruption, size and cardinality lies, bit flip sweeps), the calendar rebase unit tests against Spark's own anchors, and end to end scan pins for per file metadata resolution; clippy and fmt clean.
• Regressions from review are pinned: legacy written ancient dates and INT96 timestamps, metadata-free files under each read mode, nested columns with mixed policies, assumed role session policies, column mapping name collisions with and without DVs, and S3 bucket precedence.

Benchmarks at the current head

Apple M5, JDK 17, Spark 3.5 profile, local filesystem, 120M rows in 6 files of about 490 MB (zstd), full table aggregate touching every surviving row, medians of 5 warm runs per fresh session. Results are bit identical across all modes and verified against closed form expectations.

deletion pattern	deleted	stock Spark	Comet fallback	native Delta scan
none	0	5.31s	5.25s	1.13s
sparse (0.1 percent scattered)	120K	7.76s	5.35s	1.52s
contiguous (20 percent)	24M	5.76s	2.77s	1.22s
alternating (50 percent)	60M	4.76s	2.53s	2.51s

DV decoding is negligible in every pattern; the cost center is selector expansion for alternating deletes (61 to 93 ms and about 400 MB peak per file). The default spark.comet.scan.delta.dv.maxDeletedRowsPerFile cap (1M) declines the contiguous and alternating tables up front and falls back cleanly, which the numbers show is the better path for alternating; raising the cap without sizing the off heap pool fails tasks at the reservation by design.

The calendar rebase wrapper costs 0.7 to 2.2 ns per row and is noise at scan level, but it is opaque to pruning: a selective predicate on a rebased column decoded 65x more rows than with pruning live on a sorted table. That is the tradeoff of the legacy path and only applies to files that need rebasing.

An independent run on public data (NYC taxi with a DV delete) is in the PR discussion and confirmed exact DV row removal with timing parity.

[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?

[sunchao]

Thanks @dwsmith1983. This makes sense to me! #4952 has just been merged. Could you rebase this PR and adapt to it? Thanks!

[dwsmith1983]

@sunchao
Merged main to pick up #4952 and reconciled the two Delta efforts as discussed. The JVM-planned scan now rides the generic ContribScan envelope with its own type_url (comet.contrib.delta_spark.DeltaSparkScan), so the dedicated oneof slot is gone (removed and reserved). The native handler is now a sibling of the kernel path's handler, dispatched by type_url, and the module moved to contrib/delta-spark so it no longer overlaps contrib/delta's source root. Our proto messages are renamed DeltaSpark* so both message sets coexist, and nothing from #4952 was reverted or modified; verify-contrib-delta-gate.sh passes unchanged. Both contribs' suites are green side by side (contrib 40/40, CometScanContribSuite and the injector suites 29/29, native 172/172).

A few things I deliberately left for discussion rather than deciding unilaterally: unifying the two claim hooks in CometScanRule (CometScanContrib vs the CometScanRuleExtension SPI), conf naming (spark.comet.scan.delta.* vs spark.comet.scan.deltaNative.*), and Maven packaging (the -Pcontrib-delta add-source vs this module's separate jar, which is what keeps the opt-in story build-free).

[sunchao]

Thanks for reconciling this with #4952. The generic envelope, separate source roots, and explicit runtime opt-in look like useful progress. I reviewed 9b393c15 and left eight concrete correctness and compatibility comments. The main concerns are unsafe scalar-subquery pushdown, mixed-authority file routing, and unbounded deletion-vector row-selection memory.

I checked these against Spark/Delta source and used bounded stock Spark 4.0.3 / Delta 4.0.0 and isolated Rust probes. I have not built this PR's full JNI library or run cloud-backed end-to-end tests. The Delta CI suites are green on Spark 3.5, 4.0, and 4.1. I am leaving the already-acknowledged claim-hook, naming, and packaging choices for the existing design discussion.

[sunchao]

[P2] Avoid constructing a cold S3 store inside the DV runtime

Could we resolve the required stores before entering attach_access_plans, or make their initialization async-safe? The caller enters get_runtime().block_on(...), but an uncached S3 sidecar reaches this synchronous helper and then objectstore/s3.rs calls get_runtime().block_on(build_credential_provider(...)) again. Tokio rejects that nested Handle::block_on with a panic. A fresh executor reading a shallow clone whose data is in bucket A and whose new DV is in bucket B reaches a cold cache entry. Same-bucket tests hide the problem because the data store was created before the outer block_on. Explicit endpoint/region or static Hadoop credentials do not avoid the inner credential-provider call. Please add a test with distinct data-file and DV buckets.

[dwsmith1983]

Fixed by pre-resolution: all stores (data and DV authorities) are resolved on the JNI thread before entering the runtime, and attach_access_plans no longer takes an options map or imports the store builder at all, so the async path structurally can't construct one. Your distinct data/DV bucket scenario is encoded in a new MinIO suite (CometDeltaS3Suite), but heads up that it's docker-gated and hasn't run against a live daemon yet, the contrib CI job has no docker socket so those tests cancel. First live signal needs a Docker environment.

[sunchao]

Thanks @dwsmith1983. On the design topics you flagged, I’d prefer using CometScanContrib as the shared interface and agreeing on consistent configuration naming. The separate optional JAR sounds reasonable if it lets users try the feature without rebuilding Comet. We can discuss the packaging details separately.

[sunchao]

Rechecked 7e09e04f with five independent review scopes. One additional P2 is inline; I also followed up in the existing threads on the remaining scalar-pushdown, Azure DV store, and DV-memory issues. Verification used exact-source Spark/Delta physical-plan probes and locked-dependency Rust probes, not a full Comet/JNI or live cloud run.

[schenksj]

Hey @sunchao / @parthchandra — are you looking to move this work over to @dwsmith1983’s series?

We’re 2 PRs into the 10-PR series now that the contrib modules have been merged. The series fully implements all of the Delta protocols, with all 10k+ Delta test cases passing.

I’m fine either way. I won’t have a huge amount of time to work on this over the next couple of months, so it could go faster with a different attendant, but the PRs are ready to roll.

You can see the series here: https://github.com/schenksj/datafusion-comet/pulls (PRs #5–13).

[sunchao]

Hi @schenksj , I think your series implements Delta native scan based on the delta-kernel-rs while the PR here uses the JVM based delta-spark for planning, so they are different while both are based on the same contrib groundwork.

I think your series is pretty valuable and should be continued to push forward. At some point we should compare feature coverage and performance between the two.

[dwsmith1983]

CI notes for this push: the S3 test base built its client without a region, which aborted CometDeltaS3Suite in CI's empty AWS environment before any test ran; fixed, and since GitHub mounts the Docker socket into job containers the MinIO scenarios will actually execute in CI now. They've been run live locally with all AWS env vars unset (first executions ever, both pass on Spark 3.5 and 4.1; that surfaced a missing spark-hadoop-cloud test dependency, also fixed). Heads up that inside the job container the MinIO endpoint may resolve as unreachable sibling-container networking; the suite now fails soft to canceled rather than aborting the build, and logs the resolved endpoint so the first CI run tells us whether a testcontainers host override is needed. The Spark 4.0 cell wasn't rerun locally, so CI is its first pass over these changes. The Rust 1.98 clippy fix I'd pushed got dropped in favor of #5400 from main during rebase.

[sunchao]

Reposting the two remaining P2 findings here for visibility. Both remain present at 95125623; these are the existing findings, not additional issues.

[P2] Check selected-file schemes before claiming a shallow clone

The filesystem gate checks only the table's rootPaths. A valid Delta shallow clone can have a supported file: root while its selected data files still reference viewfs:. With the default libhdfs configuration (hdfs only), both authority checks accept those files and the contrib claims the scan. Native store preparation then fails with Unable to recognise URL "viewfs://..." instead of falling back to Spark.

This was verified with a real Spark 4.0.2 / Delta 4.0.0 shallow clone that Spark successfully reads, plus the exact native store-preparation helper. Please apply the supported-scheme check to the selected data-file URIs before claiming the scan.

Code · Existing discussion and reproduction details

[P2] Account for the DV reader's combined-selection allocation

Construction admission and the initial reader clone are now covered. However, DataFusion 54.1 subsequently calls into_overall_row_selection, which allocates another selector buffer while the attached original and the consumed clone's backing vector remain live. The reservation has already been reduced to twice the retained selector bytes.

With the default-permitted 1,000,000 alternating deletions across 2,000,000 rows, the current attachment reserves 64,000,000 bytes, but the attached selectors plus reader-normalization allocations peak at 97,554,457 bytes and retain 65,554,432 bytes afterward. Please account for normalization and vector capacity, or avoid the additional allocation through ownership transfer. Simply changing the factor to 3 would still fall below this measured peak.

This was reproduced using the unchanged attachment code and the real locked dependency conversion. These are allocator-requested bytes, not RSS or a reproduced executor OOM. Both findings were checked with focused probes and source tracing, not a full Comet/JNI integration run.

Code · Existing discussion and reproduction details

[parthchandra]

Hey @sunchao / @parthchandra — are you looking to move this work over to @dwsmith1983’s series?

We’re 2 PRs into the 10-PR series now that the contrib modules have been merged. The series fully implements all of the Delta protocols, with all 10k+ Delta test cases passing.

I’m fine either way. I won’t have a huge amount of time to work on this over the next couple of months, so it could go faster with a different attendant, but the PRs are ready to roll.

You can see the series here: https://github.com/schenksj/datafusion-comet/pulls (PRs #5–13).

I see value in both (even though it is extra work to maintain both paths) and in principle agree with @sunchao. Ideally, we want to converge these two. Logged an issue based on an AI generated convergence path - #5411

[schenksj]

Hey @sunchao / @parthchandra — are you looking to move this work over to @dwsmith1983’s series?
We’re 2 PRs into the 10-PR series now that the contrib modules have been merged. The series fully implements all of the Delta protocols, with all 10k+ Delta test cases passing.
I’m fine either way. I won’t have a huge amount of time to work on this over the next couple of months, so it could go faster with a different attendant, but the PRs are ready to roll.
You can see the series here: https://github.com/schenksj/datafusion-comet/pulls (PRs #5–13).

I see value in both (even though it is extra work to maintain both paths) and in principle agree with @sunchao. Ideally, we want to converge these two. Logged an issue based on an AI generated convergence path - #5411

Thanks guys. I'm concerned that having 2 will create a lot of confusion when it comes to support.. Even enabling and disabling various scan features is too much to understand for most of the expert data engineers I work with every day.

I'm happy to move forward initially in parallel, though like I mentioned before my time to work with this is going to be pretty sparse for the next couple of months.

[sunchao]

@schenksj Let’s see how it goes. For now, I see the delta-spark-based implementation as the most practical approach: it builds on mature Delta planning while allowing Comet to reuse its optimized native Parquet reader. Longer term, I’m also excited about delta-kernel-rs as a shared foundation for native Delta integrations, and I've also heard that the Delta community is also converging on the Rust implementation.

In terms of your concern, I think we should aim to keep the user-facing configuration simple, perhaps with one flag to enable Delta scans and another to opt into an experimental Rust-kernel-backed path. Ideally, both approaches would share as much integration and testing infrastructure as possible.

Really appreciate all your work on this! We’re planning to move quickly with the current delta-spark integration and evaluate it against some very large-scale production workloads. We also plan to evaluate the delta-kernel-rs-based approach in the future, and I’d love to collaborate on your series and take on some work to move the Rust-based reader forward.

[dwsmith1983]

On the macOS scans failure: pulled the hs_err from the run artifact. The crashing thread is a native thread (not a Java thread) that was exiting: the stack is pthread_start into pthread_exit into pthread TSD cleanup, then a jump through a corrupted destructor slot whose value is ASCII string bytes, at 119s elapsed, immediately after ParquetReadFromFakeHadoopFsSuite, the only suite in the group that exercises the libhdfs bridge and its JNI-attached native threads. The Delta code in this PR is structurally unreachable in those suites (native side is dispatch-gated on an operator those plans never emit, and the contrib jar is not on that build's classpath), and the Linux scans group passed on the same commit. My guess is a teardown race in the libhdfs bridge or a runner flake rather than anything this PR executes; the falsifying experiment would be rebuilding the dylib without the delta feature and re-running, since the same crash would exonerate it by construction. Could someone re-run the job? Happy to file the hs_err as an issue either way.

[sunchao]

Thanks for addressing the earlier findings. I think the larger file-selection refactor, shared admission/schema cleanup, packaging changes, and broader deployment coverage can be tracked in follow-up PRs. I'd keep the remaining [P1] Azure safety guard, [P2] S3-authentication and AQE lifecycle fixes, and their focused regressions in this PR.

Could we replace spark.comet.scan.deltaNative.enabled with spark.comet.scan.delta.enabled consistently across both Delta contributions, keeping the default false? Please update the config definitions, tests, documentation, and dev scripts together, and use the spark.comet.scan.delta.* prefix for related settings. The intent is one consistent configuration namespace, not another enable flag.

This rename does not depend on changing the separate-JAR packaging. Broader reader-selection behavior can be discussed separately.

[sunchao]

[P2] Avoid executing adaptive pruning while inspecting partitioning

This getter forces perPartitionData, which calls InSubqueryExec.updateResult(). During AQE, that subquery can still be a non-executable adaptive broadcast placeholder.

A reduced Spark 4.0.2 / Delta 4.0.0 planning harness reproduced this through Spark's normal AQE validation: a DPP join in one UNION ALL branch and a coalescible shuffle in another caused validation to inspect this partitioning before the custom DPP rewrite. It then failed with CometSubqueryAdaptiveBroadcastExec ... does not support the execute() code path. Other operators remained on Spark, and no native Comet reader executed.

Could we return UnknownPartitioning(0) while adaptive placeholders remain and make this a non-lazy def, so the temporary value is not cached? A regression with a query-time dimension filter would help. The current DPP test filters the dimension before writing it, so it does not require dynamic pruning.

[dwsmith1983]

Fixed as described: outputPartitioning is now a plain def returning UnknownPartitioning(0) while any runtime filter still holds an adaptive broadcast placeholder, so AQE validation never forces perPartitionData. Rewrote the DPP test to filter at query time and added your UNION ALL shape as a regression. That shape didn't reproduce the crash pre-fix on my Spark 3.5.9 / Delta 3.3.2 profile, so it likely needs your Spark 4.0.2 harness, but the guard matches your analysis.

[dwsmith1983]

Agreed on keeping it simple. The conf is now spark.comet.scan.delta.enabled (plus spark.comet.scan.delta.dv.maxDeletedRowsPerFile), so there's one flag to enable Delta scans, and the kernel path can add its own experimental key later. Docs updated. Fixes for the three open threads are pushed as well.

[sunchao]

Summary

• Prior state and problem: Delta reads used Spark’s reader and could not use this JVM-planned native scan path.
• Design approach: Retain delta-spark’s snapshot and file planning, then execute eligible scans through Comet’s shared Parquet reader.
• Correctness / compatibility analysis: Found one new P2: transition reversion restores stale AQE pruning filters and fails an otherwise valid query. Compared relevant Spark 3.5.9, 4.0.4 and 4.1.3 semantics. Previously reported P1/P2 concerns are addressed.
• Key design decisions: The separate contrib jar and default-disabled configuration preserve explicit activation. Shared builders avoid duplicating the Parquet reader.
• Implementation sketch: Serialize common settings and partition files through DeltaSparkScan, combine deletion-vector selections with Parquet pruning, reserve selector memory and apply per-file calendar policies.
• Behavioral changes worth calling out: Relative to branch-1.1 at e9efd9f764ee0a59b7898ff028d6985d4a7a28e1, intended changes include opt-in Delta scanning, default native decoder support and contrib linkage-error containment. Ordinary Parquet rebasing remains unchanged. Each split still decodes its full-file deletion vector, and calendar wrappers restrict pruning on affected columns. No benchmark was run.
• Suggested improvements: Implement sparkFallback for the Delta scan using its live pruning filters and add a Delta AQE/transition-reversion regression. This fits the existing fallback contract without another abstraction.

Reviewed the full 64-file diff from 46d91ef02d80f2a42cd89034699550f45fa13394 to 05ccf045c82cbb5fc3629088cace24e746c61def, including prerequisite and shared changes. Reused verified unchanged evidence and checked integration with the newer base. Considered existing non-Copilot discussions. GitHub confirms the requested head remains open and non-draft.

Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-memory-pr and review-comet-ffi-pr.

Exact-head CI: the label check passed. Comet CI and CodeQL report action_required. There is no product-test verdict for this head.

Validation: 194 distinct native tests passed across 196 executions, covering calendar rebasing, Delta, Parquet, S3 configuration and the feature-disabled planner. CI configuration, suite registration, script syntax and diff checks passed. A bounded Spark 3.5.9/Delta 3.3.2 probe reproduced the finding and passed the corrected control.

Limits: native tests used --no-default-features, enabling delta where applicable. Maven bootstrap failed because repo.maven.apache.org could not resolve. Full Comet/Delta/JNI integration, Spark SQL and Iceberg suites, MinIO/cloud integration, default-feature/HDFS builds and the complete build gate were not run. The reproduction exercised source-extracted fallback methods on real Spark/Delta plans. The checkout remains clean, and no GitHub state was changed.

[sunchao]

[P2] [P2] Preserve live pruning filters when restoring the Spark scan

With native Delta scanning, AQE and dynamic partition pruning enabled, a broadcast join fails if transition reversion restores this scan. For example, enable spark.comet.exec.transitionRevert.enabled=true, set spark.comet.exec.transitionRevert.maxTransitions=0, and disable native projection with spark.comet.exec.project.enabled=false. This rewrite updates runtimeFilters, but leaves originalPlan holding its original SubqueryAdaptiveBroadcastExec. The class inherits CometExec.sparkFallback, which restores that frozen plan after AQE’s pruning rewrite has already run. Spark then throws SubqueryAdaptiveBroadcastExec does not support the execute() code path instead of returning the matching rows.

A source-extracted probe on a real partitioned Delta scan reproduced the exception. Stock Spark and a control restoring the live filters both returned 200 rows. Could this scan override sparkFallback, following CometNativeScanExec, to preserve the current partition filters and logical link? Please add a Delta regression covering AQE pruning together with transition reversion.

Evidence: Spark 3.5.9 / Delta 3.3.2 reproduction: write 400 fact rows with fact_id=i, fact_key=i%10 and fact_str='f'+i, partitioned by fact_key, plus a Parquet dimension containing dim_id=dim_key=0..9 and dim_str='d'+i. Execute SELECT f.fact_id, f.fact_str, d.dim_str FROM f JOIN d ON f.fact_key=d.dim_key WHERE d.dim_id < 5. Capture the initial and final AQE fact scans. Apply the exact current withDynamicPruningFilters and inherited sparkFallback implementations in reduced scan classes. Executing the restored scan through ColumnarToRowExec throws the unsupported-execution exception. Copying live partition filters into the original scan returns 200 rows, matching the stock query. Source routing confirms RevertNativeForTransitionHeavyStages.revertToSpark calls this inherited fallback after the AQE pruning rules. Official Spark 3.5.9, 4.0.4 and 4.1.3 sources all define the restored placeholder as unexecutable. Probe and output: /tmp/5365-review-05cc-oct7/FallbackProbe.scala and /tmp/5365-review-05cc-oct7/fallback-run.log. This is a method-level reproduction using real Spark/Delta plans, not a full Comet/JNI run.

[dwsmith1983]

Could this scan override sparkFallback, following CometNativeScanExec, to preserve the current partition filters and logical link?

Done in 2fe0af9. CometDeltaNativeScanExec.sparkFallback now restores originalPlan with runtimeFilters as its partition filters, which are the same filters AQE rewrote on the live scan, and carries over the logical link. Data filters stay as the original scan had them: the live list also holds subquery predicates taken from the parent Filter, which Spark left out of its own scan.

CometDeltaNativeScanSuite has your query as a regression, with DPP on and transition reversion forced, run with AQE both on and off. The AQE run failed with the SubqueryAdaptiveBroadcastExec error before the change. It now matches Spark and checks that the restored scan is a FileSourceScanExec whose pruning subquery is an executable broadcast, not the adaptive placeholder.

[sunchao]

Summary

• Prior state and problem: JVM-planned Delta reads used Spark’s reader and could not use Comet’s native Parquet pruning.
• Design approach: Add an optional Delta contrib that retains delta-spark’s snapshot and file planning, then executes eligible scans through Comet’s shared Parquet reader.
• Correctness: No introduced P1/P2 issues found within this review. Checked deletion-vector application, split ownership, column mapping, metadata liveness, subqueries and calendar rebasing. The latest sparkFallback override preserves executable AQE pruning filters and the original logical link. Previously reported P1/P2 mechanisms are addressed in the current source.
• Compatibility analysis: Compared relevant Spark 3.5.9, 4.0.4 and 4.1.3 sources and supported Delta APIs, including requested-type rebasing and adaptive broadcast placeholders. The contrib supports Spark 3.5/Delta 3.3.x, Spark 4.0/Delta 4.0.x and Spark 4.1/Delta 4.3.x. Spark 3.4 and 4.2 remain outside its supported matrix.
• Key design decisions: Activation requires the separate contrib jar and default-disabled spark.comet.scan.delta.enabled. Admission guards retain Spark fallback for unsupported shapes. Deletion-vector selectors use native memory reservations, with one consumer registration per preparation call.
• Implementation sketch: Service discovery connects the Scala scan and DeltaSparkScan envelope to the native dispatcher. Plan-data injection supplies partition files. Native preparation intersects deletion-vector selections with Parquet pruning and applies per-file calendar policies through the schema adapter.
• Performance: Split-aware selector construction, shared footer caching, per-partition store reuse and modern-value buffer reuse avoid identifiable repeated work. Each split still decodes its full-file deletion vector and uses conservative memory admission. Calendar wrappers inhibit pruning on affected columns. No new P1/P2 performance regression was established, and no benchmark was run.
• Design: Keeping Delta snapshot semantics on the JVM and Parquet execution in the existing native reader gives the implementation a clear boundary. The shared planner extraction preserves ordinary scan behavior, including the current base’s floating-point filter handling.
• Abstraction & complexity: The envelope, PlanDataInjector and ParquetAccessPlan reuse existing interfaces appropriately. Reservation lifetimes follow the retained file plans. The fallback correction fits the existing override contract without another abstraction. No new P1/P2 complexity issue was established.
• Behavioral changes worth calling out: Compared with branch-1.1 at e9efd9f764ee0a59b7898ff028d6985d4a7a28e1, intended additions include opt-in native Delta reads, default native decoder support and contrib linkage-error containment. Ordinary Parquet rebasing remains unchanged. Unsupported ancient writer-timezone cases raise the documented native error.
• Suggested improvements: No additional code change meeting the requested P1/P2 reporting bar. Exact-head integration validation remains outstanding as detailed below.

Reviewed the full 64-file diff from b56349697b786ff2ad1c1bcf6ecf45b809af5f30 to d69e78ef2a33334739f4f9ea1319faca6e4b2e77, including prerequisite and shared scope. Verified that 61 contribution patches match the earlier reviewed revision, reused verified evidence, and examined the changed fallback, regression tests and planner integration. Considered existing non-Copilot reviews, issue comments, inline comments and threads. The PR remains open and non-draft at the requested head.

Routed skills: review-comet-pr, review-comet-expression-pr, review-comet-memory-pr and review-comet-ffi-pr.

Exact-head CI: labeling passed. Comet CI and CodeQL report action_required. There is no product-test verdict for this head.

Validation: 216 distinct native tests passed across focused rebasing, Delta, Parquet and object-store test groups. The feature-disabled planner variant also passed. CI configuration, suite registration, script syntax and diff checks passed. A freshly compiled Spark 3.5.9/Delta 3.3.2 probe using the current fallback methods returned the expected 200 rows and preserved the logical link and original data filters. Its stale-filter control reproduced the previous unsupported-execution exception.

Validation limits: Native tests used --locked --offline --no-default-features, enabling delta where applicable. Maven bootstrap failed with UnknownHostException: repo.maven.apache.org. Full Comet/Delta/JNI integration, Spark SQL and Iceberg suites, MinIO/cloud integration, default-feature/HDFS builds and the complete build gate were not run locally. The fallback probe exercised extracted methods on real Spark/Delta plans, not full Comet/JNI execution. The project checkout remains clean, and no GitHub state was changed.

[dwsmith1983]

@andygrove @sunchao @comphead hi all. Do you know when we target getting this merged?

There are some follow on tasks I am hoping to work on as well.

[andygrove]

Thanks for your patience @dwsmith1983. I am reviewing this one now.

[andygrove]

Thanks for your patience with this one @dwsmith1983. This pass covered everything the PR changes outside contrib/delta-spark, on this head merged with main at 0691bd39f7. With the contrib off, the NativeScan arm in planner.rs, the buildNativeScanCommon and populateScanConfFlags split in CometNativeScan.scala, the object store helpers in parquet_support.rs, the schema adapter and the reader factory build the same plan and read the same values as before. The rebase wrapper and the INT96 stamp only run when rebase_from_file_metadata is set, and only the Delta arm sets it. The LinkageError handling in contrib discovery only makes a broken contrib jar less harmful. The default native library now carries the deletion vector decoder, but nothing on the regular scan path reaches it.

I also ran the merge locally. The native core tests (863), clippy, the --no-default-features build, the Spark 3.5 strict warnings compile and the CI config checks all pass, and so do the core suites that cover the shared scan code. The Delta suites passed twice in a row on Spark 3.5, 275 tests in about five and a half minutes each. Only the MinIO suite was cancelled, because I don't have Docker here. The docs mark the feature as experimental and opt-in, which is how I'd want it to land. It's a great base to build on.

I have one request about CI, inline. Before this goes into the queue I'll set up the run-delta-tests label and run it here, since the MinIO job and the current tiering have never run in CI.

[andygrove]

This makes delta_3_5 a required queue job for most PRs. Replaying this filter over the last 40 merges on main, it would have run for 30 of them, the same set as spark_4_1. Nothing in the PR tier builds with -Pdelta, though. A core change that breaks the contrib passes its own PR run and first fails in the queue, where it gets evicted and the PRs behind it are rebuilt. The contrib calls a lot of core internals (QueryPlanSerde, CometNativeScan, CometScanRule, CometExecRDD, NativeConfig and more), so I'd expect this to come up. The Rust test you added for the S3 keys covers one case of it, not the general one.

I suggested keeping one Delta profile in the queue back in September, but I'd like to revisit that while the contrib is experimental. Could delta_3_5 move to the nightly tier with 4.0 and 4.1, still behind run-delta-tests? And could the Spark 3.5 strict warnings compile in pr_build_linux.yml add the delta profile, so a change that breaks the contrib build fails on its own PR? I tried ./mvnw test-compile -Pspark-3.5,delta -Pstrict-warnings -DskipTests on the merge with main, and the contrib compiles cleanly, including the new unused-params lint from #6846.

[dwsmith1983]

Could delta_3_5 move to the nightly tier with 4.0 and 4.1, still behind run-delta-tests? And could the Spark 3.5 strict warnings compile in pr_build_linux.yml add the delta profile, so a change that breaks the contrib build fails on its own PR?

Both in 3062796. delta_3_5 is nightly now, still behind run-delta-tests, and the strict warnings job runs -Pspark-3.5,delta -Pstrict-warnings. Since nothing in the queue runs the contrib any more, contrib/delta-spark/** also routes to the Linux PR build, so a change under the contrib alone gets that compile too. The cost is that a contrib-only PR runs the whole Linux pipeline; a narrower output that gates only the strict job would avoid that if you'd rather. The workflows README, ci.md and check-ci-config.py follow the new tiering, and ci.md lists run-delta-tests among the labels to reach for.

[andygrove]

Nit, optional: the release build doesn't pass -Pdelta yet (#5882), so there is no published comet-contrib-delta-spark jar to put on the classpath. Could this line say the jar has to be built from source for now and point at the README's build steps? The Delta Lake <delta> entry in index.rst also sets its own title, so the sidebar won't show "(experimental)" the way the page heading does.

[dwsmith1983]

Could this line say the jar has to be built from source for now and point at the README's build steps?

Done in 28d0ce7. It says no release publishes the jar yet and links the README's build section, and the toctree entry reads "Delta Lake (experimental)".

[andygrove]

I added the run-delta-tests label, but #6059 has since landed and this now conflicts with main in pom.xml and CometNativeScan.scala, so the Delta jobs didn't start. Once you merge main, the next run will include them and I'll approve it. The CometNativeScan one needs a decision on your side, since buildNativeScanCommon and populateScanConfFlags still take firstFileUri while convert now uses extractScanObjectStoreOptions over every file.

[andygrove]

@dwsmith1983 could you fix conflicts?

[dwsmith1983]

The CometNativeScan one needs a decision on your side, since buildNativeScanCommon and populateScanConfFlags still take firstFileUri while convert now uses extractScanObjectStoreOptions over every file.

Merged main in e2d6deb. buildNativeScanCommon and populateScanConfFlags now take the resolved object store options instead of firstFileUri, so each caller decides. CometNativeScan.convert does what main does: extractScanObjectStoreOptions over every selected file, and a fallback with the reason on both the scan and the wrapped scan. The Delta scan runs the same check over its data files. Delta's existing gate already declines data files and deletion vectors whose authority carries userinfo, which covers every ABFS path, so the only Azure authority that can reach the Delta option union is the table root. The union's comment now says so, since #6059 made the ABFS keys global. The Delta suites pass on Spark 3.5 with strict warnings, the MinIO ones included, and so do the object store and ABFS suites on 3.5 and 4.1.

[andygrove]

Thanks @dwsmith1983 - I added to merge queue 🎉

[dwsmith1983]

The Delta S3 job failure (https://github.com/apache/datafusion-comet/actions/runs/38233957666/job/114789734253) is the MinIO image, not this PR: Docker Hub no longer serves minio/minio, which CometS3TestBase pulls (#6884). #6887 moves the test base to another image and fixes two region gaps the S3 suites hit once the image pulls. That needs to merge first, and I'll merge main into this branch after it does.