feat: add native Delta Lake scan contrib module (page/row-group pruning) - #5365
dwsmith1983 wants to merge 27 commits into
Conversation
|
Update: pushed two follow-up commits extending the scan's pruning and object-store behavior.
|
888e4a7 to
7fd81aa
Compare
|
HI @andygrove, Can you review this as it adds Delta functionality? |
|
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 In addition, would it also make sense to land this in smaller pieces, for easier review and iterating? For example:
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. |
|
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. |
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
|
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. |
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? |
|
Thanks @dwsmith1983. This makes sense to me! #4952 has just been merged. Could you rebase this PR and adapt to it? Thanks! |
|
@sunchao 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
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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.
| let (dv_url, dv_store_path) = prepare_object_store_with_configs( | ||
| Arc::clone(&runtime_env), | ||
| dv_path.clone(), | ||
| object_store_options, | ||
| )?; |
There was a problem hiding this comment.
[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.
There was a problem hiding this comment.
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.
|
Thanks @dwsmith1983. On the design topics you flagged, I’d prefer using |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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.
|
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). |
|
Hi @schenksj , I think your series implements Delta native scan based on the 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. |
|
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. |
|
Reposting the two remaining P2 findings here for visibility. Both remain present at [P2] Check selected-file schemes before claiming a shallow cloneThe filesystem gate checks only the table's 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 allocationConstruction admission and the initial reader clone are now covered. However, DataFusion 54.1 subsequently calls 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. |
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. |
|
@schenksj Let’s see how it goes. For now, I see the 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 |
|
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
left a comment
sunchao
left a comment
There was a problem hiding this comment.
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.
| override lazy val outputPartitioning: Partitioning = | ||
| UnknownPartitioning(perPartitionData.length) |
There was a problem hiding this comment.
[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.
There was a problem hiding this comment.
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.
|
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. |
ec2ad9b to
92ae71b
Compare
… one vectorised minimum
Done: both prerequisite installs in
Done: the MinIO job no longer uses
Measured on a release build, 1M-row microsecond UTC timestamp batches, median of 20 runs, before against after:
The first cut used a validity-aware scalar loop and lost on null-heavy batches, so the landed version takes one vectorised minimum only on null-free batches and returns the input untouched when nothing predates the cutover; batches with nulls or ancient values keep the previous per-value arms, so the cost there is the extra minimum pass. Modern null-free batches, the common case for a metadata-free file under EXCEPTION mode, are 2.3x to 3.5x faster; the mixed-age case pays about 30% more. Pruning is unchanged by this, as you noted.
Filed, one per request: #5943 (stable injection key), #5944 (shared admission checks), #5945 (shared builder for the DV path), #5946 (explicit planning boundary), #5947 (envelope and provider contract tests), #5948 (preparation I/O through the scan's instrumented reader), #5949 (delegate schema preparation to |
|
CI on 6039c30 failed in two places. |
…scan # Conflicts: # dev/ci/check-ci-config.py
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked 68dd8e3f. Both P2 build issues are fixed in source: the prerequisite installs now activate delta to produce the current Spark test JAR, and the MinIO job uses sudo for package installation. The CI configuration checker passes locally.
The null-free minimum check preserves Spark's calendar cutoffs, and nullable checks retain masked-null behavior. The earlier Rust job passed 1,563 tests, including the rebasing cases. I verified that its rebasing source and dependency lock match this head. Current-head CI and CodeQL remain action_required with zero jobs, and that earlier Delta integration job was skipped, so this does not establish current-head Spark/JNI coverage.
The reported microbenchmarks show the modern-batch benefit alongside slower null-heavy and mixed-ancient cases. I have not independently rerun those timings. The optimization still does not restore pruning on wrapped columns.
The shared-scan work is now explicitly tracked in #5943, #5944, #5945, #5946, #5947, #5948, and #5949. These remain separate implementation work, including preparation-I/O accounting. I found no remaining P1/P2 issue in this update. Approving with the validation limits above.
…scan # Conflicts: # .github/workflows/README.md # dev/ci/check-ci-config.py # dev/ci/compute-changes.py # docs/source/contributor-guide/ci.md
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked dcacee43 against the previously reviewed 68dd8e3f. This update merges mainline changes. The Delta implementation, tests, dependency lock and both prior P2 build fixes are unchanged.
The merge preserves Delta's merge-queue gate and run-delta-tests opt-in while adopting mainline's nightly routing for other suites. The CI configuration checker passes locally. Another 1,092 path/event cases confirm that Delta routing matches the previous head and all other routing matches the new base. The merge result's full tree matches this head.
No new or remaining P1/P2 finding. Approving with the same runtime-validation limits: current-head CI and CodeQL remain action_required with zero jobs. These local checks do not establish Spark/JNI or MinIO integration coverage. The previously linked shared-scan follow-ups remain separate work.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked b3cd0710 against dcacee43 and base 8c229a70. All 59 contribution files are byte-identical to the previous head. The 39-file increment exactly matches the mainline increment. Both prior P2 build fixes remain intact.
The local CI configuration checker and 1,300 path/event routing cases pass. Delta routing remains queue-gated or enabled by run-delta-tests, and the merge result's full tree matches the head. No new or remaining P1/P2 finding. The seven previously linked shared-scan follow-ups remain separate work.
Preserving the existing approval. Current-head CI and CodeQL remain action_required with zero jobs. No new native build, Spark/JNI, MinIO integration run, or benchmark was executed.
…scan # Conflicts: # native/core/src/execution/operators/dynamic_filter/tests.rs
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked d21f5d4d against b3cd0710 and base 36146a87. The update merges main and separates Maven cache restore/save with a main-only save guard. The merge result's full tree matches the head. Both prior P2 workflow fixes remain intact.
One new [P2] integration compile issue: the merged plan-data API requires a CometExecRDD fingerprint argument and a prepared-common provider interface, but the Delta call and DeltaPlanDataInjector still use their previous signatures. The inline comment covers both sites. Isolated probes reproduce both failures on Scala 2.12.18 and 2.13.16, with passing prior-signature and adapted controls. The local CI configuration checker passes.
The seven tracked shared-scan follow-ups remain open and unchanged. Current-head CI and CodeQL remain action_required with zero jobs. Only labeling succeeded. No full product build, Spark/JNI or MinIO integration run, or benchmark was executed. This revision needs the API integration fix before approval.
| Map(sourceKey -> commonData), | ||
| Map(sourceKey -> perPartitionData), | ||
| serializedPlan, | ||
| perPartitionData.length, |
There was a problem hiding this comment.
Correctness
[P2] Adapt both Delta integration sites to the merged plan-data API
The main merge brings in #5615, but this constructor call still uses its previous argument list. CometExecRDD now requires planFingerprint: Long after serializedPlan, so nativeMetrics here is matched to numOutputCols: Int and compilation fails. There is a second stale site in DeltaPlanDataInjector: it still overrides inject(..., Array[Byte], ...) and does not implement the new prepareCommon/Prepared contract.
Source-extracted probes reproduce both failures with Scala 2.12.18 and 2.13.16. The prior signatures and adapted controls compile. These are isolated probes with type stubs, not a full module build. Could you pass PlanDataInjector.planFingerprint(serializedPlan) here, adapt the provider to prepare the common protobuf once, and compile the Delta contrib for both Scala lines? The runtime opt-in does not avoid compiling these sources under -Pdelta.
There was a problem hiding this comment.
Could you pass
PlanDataInjector.planFingerprint(serializedPlan)here, adapt the provider to prepare the common protobuf once, and compile the Delta contrib for both Scala lines?
Done in adc17c2. The scan passes the fingerprint, and DeltaPlanDataInjector now parses the common DeltaSparkScan once in prepareCommon and only merges the partition's file list in inject, matching the built-in injectors. DeltaPlanDataInjectorSuite pins the contract, including reuse and stale replacement through core's prepareShared. The contrib test-compiles under -Pspark-3.5,delta (Scala 2.12) and -Pspark-4.0,delta (Scala 2.13), and the three Delta suites pass locally. The follow-up commit drops redundant s interpolators in CometDeltaNativeScanSuite that the semantic scalafix check flags.
CometExecRDD now carries the driver-computed plan fingerprint, and PlanDataInjector splits parsing the partition-invariant common from injecting a partition. The Delta scan passes the fingerprint, and DeltaPlanDataInjector parses the common DeltaSparkScan once in prepareCommon and merges only the file list in inject. DeltaPlanDataInjectorSuite pins the contract, including reuse and stale replacement through core's shared memo.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked a6fe101f against d21f5d4d and base 58ab5f61. Both sites in the previous P2 are fixed: the scan passes the plan fingerprint, and DeltaPlanDataInjector uses the prepared-common API. The new suite covers common/partition assembly, child identity, memo reuse and stale-common replacement. The remaining test edits remove redundant string interpolators.
I reran 16 source-extracted compilation cases on Scala 2.12.18 and 2.13.16, including the original failures and controls. The old constructor/provider cases still fail, while the updated cases and prior-signature/adapted controls pass. These are source-extracted probes with type stubs, not a full Delta module build. The CI configuration checker also passes. No new or remaining verified P1/P2 findings.
All seven tracked shared-scan follow-ups remain open. The new tests cover part of #5947, but do not complete its envelope and provider test scope. The author reports passing local test-compiles and Delta suites. At September 17, 11:37 UTC, CI and CodeQL still require approval and have no jobs. Only labeling has passed. I did not run a full product build, Spark/JNI or MinIO integration tests, or a benchmark.
andygrove
left a comment
andygrove
left a comment
There was a problem hiding this comment.
Second round from me. @sunchao has the open threads well covered, so I focused on what changed since my September 12 comments and on running this locally.
Everything I raised last time looks handled, and I checked each one: the LinkageError containment now reaches discovery in ContribServices.loadFrom, the inline DV payload is checked against its descriptor size, row-group offsets use checked_add, fs.s3a.connection.ssl.enabled and the STS endpoint keys are modelled in the claim gate, the -tests.jar execution moved inside the delta profile, the duplicate delta.version is gone, the README link points at latest/, and the change filter picked up !**.md. Thanks for working through all of that.
What I ran on a6fe101f:
| Check | Result |
|---|---|
cargo check --features delta --all-targets |
clean |
./mvnw -Pspark-3.5,delta package -DskipTests |
clean |
| Delta contrib suites, Spark 3.5 | 256 passed, 0 failed, 3 canceled (MinIO, no local Docker) |
datetime_rebase Rust tests |
42 passed |
So today's plan-data adaptation compiles and the suite is green locally.
The thing I would most like sorted before this merges is CI. POLICY["delta"] in compute-changes.py is queue-only, so the contrib suites have never actually executed in CI on this PR. The last full run was 5b22bd5b on September 12 and "Delta Contrib Tests" was skipped there, and five commits have landed since, including the workflow restructure itself, the spark/pom.xml test-jar move, and today's plan-data API fix. The PR that introduces a workflow is the one that should prove the workflow works, otherwise the first real execution happens in the merge queue where a failure blocks everyone. Could you add the run-delta-tests label so the 3.5, 4.0 and 4.1 legs and the MinIO job all get a run? The branch needs a merge too, it currently conflicts with main in native/core/src/execution/planner.rs.
I left one finding inline on the rebase pass-through, plus two smaller ones on CI cost and the config table.
Three of @sunchao's threads also look resolved to me but are still open, so they may be worth closing to keep the remaining list honest: the Unicode case-folding P1 is main's name_fold behaviour now that #5602 landed, unsupportedSelectedSchemeReason is applied to the selected data-file URIs at DeltaScanSupport.scala:311, and objectStoreRejectedPathReason covers the real-path gate.
| array: &PrimitiveArray<T>, | ||
| cutover: i64, | ||
| ) -> bool { | ||
| array.null_count() == 0 && arrow::compute::min(array).is_none_or(|min| min >= cutover) |
There was a problem hiding this comment.
The null_count() == 0 guard here undoes the pass-through this function exists for. arrow::compute::min already skips nulls, so the guard is redundant, and with it any column holding a single null takes the try_unary path and allocates a fresh values buffer on every batch, even when every valid value is modern. I proved it with a probe asserting Arc::ptr_eq on a nullable all-modern batch, which fails on this head under all three non-Corrected policies.
This is reachable by default rather than exotic. On Spark 3.4 and 3.5 datetimeRebaseModeInRead defaults to EXCEPTION, which maps to CheckAncient for any file with no Spark writer metadata, so every requested date and timestamp leaf of a Delta table written by delta-rs, Trino, Flink or DuckDB gets the wrapper. Spark 4.x defaults to CORRECTED, so the exposure is 3.4 and 3.5, and it is Delta-arm only since rebase_from_file_metadata gates it.
Measured on an 8192-row TimestampMicrosecond batch under CheckAncient, release build, 20k iterations:
| variant | null-free | one null |
|---|---|---|
| this head | 0.18 ns/row | 0.71 ns/row, plus a fresh 64 KB buffer per batch |
min alone |
0.14 | 0.78, since null-aware min is the slow part |
| hybrid below | 0.16 | 0.43, same Arc returned |
So dropping the guard on its own is not the answer. Keeping the vectorised min only when the array is null-free, and falling back to the validity-aware loop this commit replaced, gets the best of both:
match array.nulls() {
None => arrow::compute::min(array).is_none_or(|min| min >= cutover),
Some(nulls) => array
.values()
.iter()
.zip(nulls.iter())
.all(|(&v, valid)| !valid || v >= cutover),
}All 43 tests in the module pass with that, plus the probe. The same shape applies to the date copy at line 913.
One more thing. 6039c304 also weakened the test that would have caught this: modern_batches_pass_through_without_a_new_buffer used to loop for input in [&dates, &masked] asserting Arc::ptr_eq on both, and it now pulls masked out and only checks out.null_count() == 1. Could that assertion go back?
There was a problem hiding this comment.
Keeping the vectorised
minonly when the array is null-free, and falling back to the validity-aware loop this commit replaced, gets the best of both ... Could that assertion go back?
Done in f0351b9. all_modern takes the hybrid for both the timestamp and the date arm, and the pass-through test again asserts Arc::ptr_eq for the masked dates, plus a masked timestamp batch with an ancient value under the null slot, under all three policies. The test fails on the old guard and passes now; the module's 41 tests and clippy are clean.
| ".mvn/**", | ||
| "mvnw", | ||
| ], | ||
| "delta": [ |
There was a problem hiding this comment.
This filter matches native/**/src/** and spark/src/main/**, so it adds four jobs to essentially every queued PR, each doing a native build plus a Maven install plus a suite run. My local Spark 3.5 leg alone was 8m6s of test time on top of the build. Could you put the measured runner-minutes in the description? If it lands where I think it does, it may be worth trimming to one Spark profile in the queue with the other two on the label or nightly, which is how the Iceberg and Spark SQL tiers are already split.
There was a problem hiding this comment.
Could you put the measured runner-minutes in the description? If it lands where I think it does, it may be worth trimming to one Spark profile in the queue with the other two on the label or nightly
Trimmed in e320d96 to the Iceberg shape: Spark 3.5 runs in the queue, 4.0 and 4.1 run nightly, and run-delta-tests opts a pull request into all three. The reusable workflow takes the profile as an input, and the MinIO, feature-off and dev-script jobs run once, on the 3.5 call. Measured minutes have to come from the labelled run, since the suites have not executed in CI on this PR yet; your 8m6s of test time for the 3.5 leg matches what I see locally for the three suites, and I will put the runner minutes in the description once that run exists.
|
|
||
| ## Configuration | ||
|
|
||
| <!--BEGIN:CONFIG_TABLE[delta]--> |
There was a problem hiding this comment.
This page carries CONFIG_TABLE markers, but DeltaSparkConfigProvider's own comment says the doc build cannot see the provider with the current module layout, so nothing fills them. The table is hand-maintained while looking generated, which is how it will drift. The guard in DeltaScanContribSuite covers only COMET_DELTA_NATIVE_ENABLED's key and doc string, not maxDeletedRowsPerFile and not the default values, and it cancels rather than fails when it cannot find the file. Could the guard cover every entry in DeltaScanConf.all including defaults, and fail instead of cancel on a missing file? Or drop the markers so nobody reads the table as generated.
There was a problem hiding this comment.
Could the guard cover every entry in
DeltaScanConf.allincluding defaults, and fail instead of cancel on a missing file? Or drop the markers so nobody reads the table as generated.
Both, in d707d8d. The markers are gone and a comment says the table is hand-maintained. The guard renders every DeltaScanConf entry the way GenerateDocs would, key, doc and default, asserts each row verbatim, flags rows that match no entry, and fails when the page cannot be found. It fails on a changed default and passes on the current page.
…scan # Conflicts: # native/core/src/execution/planner.rs
The null-count guard sent any column holding a single null through try_unary, allocating a fresh buffer per batch even when every valid value is modern. Null-free arrays keep the vectorised minimum; arrays with nulls take one validity-aware pass, and the pass-through test asserts the same Arc comes back for masked dates and timestamps under every policy.
…ilure The dedicated runner job exists only to run CometDeltaS3Suite, so a cancelled run and a real one looked the same. The job sets the switch, and the suite then fails instead of cancelling when Docker is unavailable or MinIO does not start.
…markers Nothing fills the CONFIG_TABLE markers on the Delta page, so the table is hand-maintained. The guard now renders every DeltaScanConf entry with its default, flags stale rows, and fails rather than cancels when the page cannot be found.
|
Merged main; the |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked d707d8d1 against a6fe101f at base 10f0fdd8. The nullable rebase pass-through, required S3 setup, and config-table guard changes address the existing comments. The earlier plan-fingerprint and prepared-common API fixes remain intact after the main merge.
I reran 17 old/new lifecycle scenarios with real ScalaTest 3.2.16 and simulated Docker, MinIO and Spark components. Required mode now aborts on unavailable Docker, MinIO startup failure or bucket-creation failure. Default mode still cancels, and healthy/failing-test controls behave as expected. The CI configuration checker passes. I source-reviewed the native array-identity assertions and documentation guard but did not execute those product tests, a full build, real MinIO/Spark/JNI integration, or a benchmark.
No new or remaining verified P1/P2 findings. The existing CI runner-minute request still awaits measurements, and the seven tracked shared-scan follow-ups remain open. At September 18, 02:03 UTC, Comet CI and CodeQL require approval with zero jobs. Only labeling has passed.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked 94e0be8c against d707d8d1 at base f51f5831. All 60 authored file deltas are unchanged. All 38 intervening file changes match the main-branch update. The inherited duplicate-field-ID check reaches Delta through the shared schema checker and conservatively falls back. Delta admission, deletion-vector planning, per-file rebasing and the prior fixes remain intact. Arrow/Parquet 59.3.0 and DataFusion 55.1.0 versions and checksums are unchanged.
The CI configuration checker and both diff checks pass. Prior lifecycle and compiler-API evidence remains applicable by exact source identity. I did not rerun those probes or execute native tests, a full product build, real MinIO/Spark/JNI integration, or benchmarks on this revision.
No new or remaining verified P1/P2 findings. My existing approval stands. The CI runner-minute request still awaits measurements, and the seven tracked shared-scan follow-ups remain open. At September 18, 14:20 UTC, Comet CI and CodeQL require approval with zero jobs. Only labeling has passed.
The single Delta job ran three profiles on every queued change that touched native or spark sources. Spark 3.5 now runs in the queue and 4.0 and 4.1 run nightly, with run-delta-tests opting a pull request into all three, the same split the Iceberg tiers use. The reusable workflow takes the profile as an input, and the MinIO, feature-off and dev script jobs run once, on the 3.5 call.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed 3f68cc9e against 94e0be8c, at base 67168ca1. Delta CI now runs Spark 3.5 in the merge queue, Spark 4.0/4.1 nightly, and all three with run-delta-tests. The MinIO, feature-off and dev-script jobs run once through the 3.5 caller. The required-results aggregator includes all three callers, and the MinIO job retains required S3 setup with JDK 17 after the setup-java v6 update.
The routing scenarios and CI-configuration check pass. Native/JVM implementation, regression tests and dependency lock are unchanged, so the prior fixes and qualified lifecycle/API evidence remain applicable. No new or remaining verified P1/P2 findings. The runner-minute request still awaits measurements, as the author confirms, and the seven shared-scan follow-ups remain open.
At September 18, 16:17 UTC, CodeQL passed and Comet CI was still running. Logs confirm the reviewed merge and all 60 changed files, but all three Delta suites were skipped on this run without run-delta-tests. I did not rerun lifecycle/compiler probes, native/product tests, real MinIO/Spark/JNI integration or benchmarks. Maintained Spark 3.4/4.1 source coverage remains unavailable.
9 participants
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
-PdeltaMaven profile builds a separatecomet-contrib-deltajar that is never bundled intocomet-spark, andspark.comet.scan.delta.enableddefaults to false. Thedeltacargo 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 adjacentcontrib-deltafeature 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:
${...}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.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):
fs.s3a.assumed.role.policy) decline outright since Hadoop sends them in the AssumeRole request and native does not.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.delta_dv.rs(deletion vector decode with a full malformed input matrix, and access plan construction),delta_spark_scan.rsplanner arm,datetime_rebase.rs, proto messages for the Delta scan envelope, S3 object store helper.build_parquet_scan_plan/prepare_scan_store_and_filesextraction in the planner,object_store_url_key/prepare_object_store_with_config_hash,buildNativeScanCommonextraction,reportScanInputMetrics,hasScanInputwidening, contrib LinkageError containment.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?
--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.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.
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.maxDeletedRowsPerFilecap (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.