feat: add native Delta Lake scan contrib module (page/row-group pruning) - #5365
feat: add native Delta Lake scan contrib module (page/row-group pruning)#5365dwsmith1983 wants to merge 1 commit 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. |
| s.split(",").map(_.trim.toLowerCase(Locale.ROOT)).filter(_.nonEmpty).toSet | ||
| case None => Set("hdfs") | ||
| } | ||
| val unsupportedFsSchemes = scanExec.relation.location.rootPaths |
There was a problem hiding this comment.
[P2] Check selected-file schemes before claiming a shallow clone
Could we apply this filesystem gate to the selected data-file URIs, not just the table's rootPaths? A valid Delta shallow clone can have a supported file: table root while its data files still reference viewfs://review-mount/source/table/.... With the default libhdfs scheme set (hdfs only), both authority checks accept these same-authority files, and the ordinary LongType scan serializes successfully, so the contrib claims it. Native store preparation then fails with Generic URL error: Unable to recognise URL "viewfs://..." instead of leaving the scan with Spark.
At bc98657f, a local-only Spark 4.0.2 / Delta 4.0.0 probe wrote a Delta table through Hadoop's built-in viewfs mount, shallow-cloned it to a local directory, and successfully read [0, 1, 2]. Its actual scan had a file: root and viewfs: selected files. The exact-current authority helpers accepted those files, while the exact native store-preparation helper rejected their URI. This was a stock-engine/exact-helper probe, not a full Comet/JNI run. Checking the schemes of the files actually selected before claiming would preserve Spark fallback for this valid table.
There was a problem hiding this comment.
Fixed. The scheme gate now runs over the selected data-file URIs and the DV absolute paths, the same sequences the later authority gates already collect, using the exact predicate the root-paths gate had (lowercased, null tolerant, libhdfs exemption honored at the new call site). It sits ahead of the multi-store gate since an unreadable scheme is the stronger and more actionable reason, and both authority gates presume the URIs are natively resolvable. s3a is recognized by the native scheme parser so the MinIO coverage is untouched. Your probe is now a CI test: the suite mounts viewfs over a local directory, writes through it, shallow-clones to a file: root, and asserts the scan falls back with the scheme reason while answers match, including a mixed-scheme shape that pins the gate ordering end to end.
|
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
a4bd616 to
48e5d29
Compare
|
Thanks for the breadth here. Two of these were defects worth fixing in the PR, done and pushed in 48e5d29; the rest are filed as tracking issues per your suggestion. Point by point: Calendar handling: fixed in-PR. Your repro was real and it led somewhere interesting: the regular native scan has no rebase handling at all today (that is documented as a known limitation and tracked in #5010, ParquetRebaseDatetimeSuite is disabled for Comet), so there was nothing to inherit. The Delta arm now resolves calendar policy per file on the native side: the footer's writer metadata (spark version, legacyDateTime, legacyINT96, timeZone) survives into the Arrow schema the expression adapter sees per file, and it resolves date and timestamp policies the way DataSourceUtils.getRebaseSpec does. Dates rebase exactly (Spark's julianGregDiffs table ported verbatim, with calendar arithmetic before its range). Timestamps rebase exactly for UTC writer zones. Anything it cannot rebase faithfully (non UTC writer zone, missing metadata, contradictory flags) refuses ancient values with an error naming the column instead of returning shifted data; modern values always pass. Your exact 1500-01-01 and ancient INT96 cases are regression tests now, RED before the fix with your 1500-01-10 shift reproduced, plus mixed writer metadata and a temporal predicate case. The mechanism is gated to the Delta arm so the regular scan's documented behavior is unchanged, pinned by a control test. Assumed-role session policies: fixed in-PR. fs.s3a.assumed.role.policy now declines admission, global and bucket scoped, resolved exactly like its consumer (AssumedRoleCredentialProvider reads it with getTrimmed on the propagated conf and sends it in the AssumeRole request, verified against 3.3.4 and 3.4.1 bytecode; native never forwards it). The reason text names the key without echoing the policy document. Unset stays admitted, pinned. Tracking issues filed for the rest: DV file splitting #5655, compressed DV decoding #5656 (a candidate patch preserving sub bitmap containers already exists from a community contributor and can be adapted once this lands), overlapping bitmap and footer reads #5657, shared cloud compatibility helper #5658, credential scoping #5659, v2 checkpoint depth #5660, capability table #5661. Benchmark on this head (48e5d29 exactly, no extra commits; Apple M5, JDK 17, spark-3.5 profile, local FS, 120M rows, 6 files of ~490MB, zstd, medians of 5 warm runs per fresh session; results bit identical across all modes and verified against closed form expectations):
Native output_rows confirms DV application inside the scan (119,880,000 on sparse). Preparation cost measured on the real exported DVs through the PR's own decode and access plan code: decode is negligible everywhere (CRC plus roaring, up to 0.36ms per file on the 2.5MB alternating DVs); the cost center is selector expansion for alternating deletes, 61 to 93ms and ~400MB true peak per file, about 0.5s CPU across the table. Contiguous deletes cost near zero allocation since fully deleted row groups become skips. Two operational notes the alternating case surfaced. First, the default dv.maxDeletedRowsPerFile cap (1M) declines contiguous and alternating up front and falls back cleanly; the numbers show that guard picks the better path for alternating (native only ties fallback there while needing a ~1.6GB reservation per file). Second, raising the cap without sizing the off heap pool fails tasks hard at the reservation, which is the documented reserve before build behavior doing its job; 18g admits it. The reservation is about 4x the measured true peak, deliberately pessimistic. Suite batteries at this head pass 227/227 on both spark-3.5 and spark-4.0, MinIO suite live. The remaining gap from your list is live object store validation beyond MinIO, which I do not have infrastructure for here. |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed 48e5d29021dd4bec7cc8e100eb63fdb7de92026d, including the latest update. The AssumeRole session-policy P1 is addressed by the new fallback gate. Two [P2] cases remain in the calendar discussion:
- [P2] Without writer-version metadata, native policy resolution ignores the configured read mode. Spark 4.0 defaults to
CORRECTED, but the new expression rejects corrected1500-01-01DATE values and a UTC timestamp one microsecond before 1970 when the file expression is rewritten. Please preserve the effective datetime and INT96 read modes, or decline the unsupported scan before execution. - [P2] The nested-type branch rejects corrected DATE leaves when only the unrelated INT96 policy needs handling. With corrected DATE mode and legacy INT96 mode, rewriting a
STRUCT<d: DATE>column errors even for modern or null values. Please check the applicable leaf policy before rejecting the column and cover this metadata combination.
The exact-module component probes reproduced these rejections and passed the scalar/calendar controls. All 22 component tests passed, but I did not run a full Spark/Delta query. Please add Spark/native regressions for these cases and a matched benchmark for the new rebase expression, including selective predicates. The new DV benchmark does not measure this path. GitHub workflows still require authorization.
There was a problem hiding this comment.
The implementation quality here is high, and I want to say that clearly before raising concerns. Scope note so this review isn't read as broader than it is: I went through the contrib Scala, the deletion-vector decode path, the planner wiring, the CI gating, and — after @sunchao's latest review — datetime_rebase.rs, which I had not covered in my first pass.
CometScanContrib keeps core free of any compile-time reference to a contrib, and the DV → ParquetAccessPlan translation (so DV skips intersect with page-index skips rather than filtering after the read) is the right design. I checked the deletion-vector decode path in detail — the row-group boundary arithmetic, the out-of-range guard at delta_dv.rs:203, validate_cardinality covering both the inline and on-disk branches, and the DeletionVectorDescriptor.EMPTY early return requiring both cardinality == 0 and size_in_bytes == 0 — and found no wrong-results path there. The malformed-input matrix (truncation at every boundary, CRC/magic corruption, size and cardinality lies, single-bit-flip sweep) is real coverage, not a token test.
The fix record since @sunchao's earlier rounds is also strong: the column-mapping name collision is fixed by routing partitionSchema through toPhysical as well, with a rename-collision test, a collision+DV test, and a no-collision control test — that last one is the part reviewers usually don't get. The S3 long-vs-short bucket precedence fix goes further and corrects a previously-wrong assumption in the code's own comments, with the hadoop-aws 3.3.4 decompilation recorded.
The decline gate is conservative about what it checks — but the calendar-rebase path is not among those things, and I think that's the most substantive issue open on this PR.
I independently reproduced both of @sunchao's remaining [P2]s by reading the code, and want to add one observation that I think raises their severity.
resolve_spec (datetime_rebase.rs:201) returns CheckAncient when a file carries no org.apache.spark.version metadata. The comment states the reason plainly — the session read conf "is not plumbed to the native side" — so a table read under Spark 4.0's default CORRECTED mode still rejects a corrected 1500-01-01 DATE (:436) or a pre-epoch UTC timestamp (:377). And wrap_datetime_rebase (:279) rejects a whole nested column whenever its type structurally contains a rebase target, without consulting the applicable leaf policy, so STRUCT<d: DATE> fails under corrected-DATE + legacy-INT96 metadata even for modern or null values.
The part I'd flag beyond the two cases themselves: DeltaScanSupport.declineReason contains no rebase-related gate at all, so neither of these is a planning-time decline — they surface as execution-time query failures. That means this class of unsupported input bypasses the decline/fallback mechanism that is otherwise this PR's main safety property: instead of quietly falling back to Spark's reader, the user gets a failed query mid-execution. The 17 unit tests in that file don't cover either case. I'd treat "decline before execution, or preserve the effective read modes" as the requirement here rather than just fixing the two expressions.
Separately — my main architectural concern. Two findings, pointing in opposite directions at the same missing thing.
First, the concrete one. About 1,060 of DeltaScanSupport.scala's 1,742 lines (roughly 540–1600) are not Delta logic at all. They model Hadoop S3A configuration semantics: S3AUtils#lookupPassword's long-then-short bucket precedence, propagateBucketOptions, JCEKS credential-provider aliases, SSE-C, proxy, assumed-role policy, provider-class allowlisting. Every occurrence of "Delta" in that range is either the private[delta] scope or an error-message string — none of it touches the Delta protocol. It exists because Comet's native object_store client resolves S3 credentials differently from Hadoop, so the scan must detect divergence and decline rather than read under the wrong identity.
That gap is not Delta-specific, and core does not handle it today: NativeConfig.extractObjectStoreOptions forwards fs.s3a.* prefixes to native with no divergence check, so a plain Parquet table read through the native scan is exposed to the same credential mismatch, silently.
datetime_rebase.rs is the same pattern, and arguably a sharper case: 862 lines implementing Spark's hybrid-Julian calendar rebase, mirroring DataSourceUtils.datetimeRebaseSpec, wired up only by the Delta arms via rebase_from_file_metadata while the plain NativeScan keeps its documented no-rebase behavior (#5010). The isolation itself is correct — I verified false at parquet_support.rs:126,143 and planner.rs:1779, true only at delta_spark_scan.rs:252, so no non-Delta scan is affected. But it also means the gap it closes stays open for the plain path: a native Parquet read of a file written in the hybrid calendar — Spark 2.4 or earlier, Spark 3.0 for INT96 timestamps (the INT96 spec's min version is 3.1.0), or any version writing with datetimeRebaseModeInWrite=LEGACY, since a present legacyDateTime/legacyINT96 flag decides the policy regardless of writer version — still silently returns pre-1582 dates shifted by up to ten days, and correspondingly shifted ancient timestamps. That is pre-existing behavior tracked in #5010, not something this PR introduced; the point is that the code which would fix it now exists, and sits behind a Delta flag.
So this PR is paying, inside a Delta contrib, for two holes in Comet's own read path. Both belong in core, where every native scan benefits and the next format doesn't have to choose between reimplementing them or ignoring them. At minimum each deserves its own issue — the findings are valuable independently of this PR.
Second, the longer-term direction. The PR also lands a substantial amount of genuine Delta protocol logic: 2,096 lines of DV binary-format handling (unframing, z85, magic, CRC, roaring decode) and the physical/logical schema splice for column mapping. Each piece is individually justified, but the pattern is what concerns me — as Comet adds table formats, it takes on reimplementing each format's low-level semantics one at a time, and every protocol evolution becomes Comet's tracking burden. The more sustainable division is for table-read concerns to be delegated to the format's own implementation, with Comet owning execution: reading bytes, pruning, vectorized execution.
I'll be honest that this is not a settled question, and not something this PR can reasonably be asked to solve. #4366's own design docs record that it started with the delegating shape (kernel for planning, ParquetSource for bytes) and moved away from it, deliberately, because the engine-side reimplementation of DV application, synthetic columns, and column-mapping physicalisation was the bulk of its reviewable surface and where its data-corruption fixes clustered. So delegation trades one maintenance burden for another rather than eliminating it. I raise it because this PR will become the template for the next format, and the boundary is worth deciding on purpose rather than by accumulation.
Taken together: there's no settled line for what a format contrib owns versus what core owns. Delta protocol parsing landed in Comet; generic object-store and calendar-rebase correctness landed in a Delta contrib. Same missing boundary, opposite directions.
One smaller thing:
delta is in the default cargo feature set (default = ["hdfs-opendal", "delta"]), which puts roaring, crc32fast, and the DV decoder in every default build, while the adjacent contrib-delta feature documents "Default builds carry zero kernel surface". I understand the motivation (the contrib jar should work against stock Comet binaries without a custom native build) and it's a legitimate one, but it's a project-level policy call rather than something to settle by default in a PR — especially with #5411 open and explicitly asking for the opposite. Either move it out of default, or get explicit agreement on #5411 and record that agreement in the Cargo.toml comment so the two adjacent features stop stating opposite philosophies.
On sequencing: this PR leaves Comet with two Delta read paths coexisting, and of the four core-surface cleanups #5411 asks for, only the SPI reuse is done. I don't think that should block merging — #4366 is still a draft and its native read path on main is a NotImplemented stub, so refusing this PR on convergence grounds would block working code on unlanded work. But "merge then clean up" versus "settle #5411 first" is a maintainer decision that deserves to be made explicitly rather than by default.
|
Both P2s addressed in 053ec14, plus one related bug your cases flushed out. Session read modes: verified the exact getRebaseSpec semantics against the 3.5.9 and 4.0.4 sources first. The conf applies only when org.apache.spark.version is absent from the footer; with a version present, policy is purely the version comparison plus the legacy flag. The serde now resolves the effective modes the way ParquetFileFormat does (ParquetOptions over relation options and session conf, so the 3.5 EXCEPTION and 4.0 CORRECTED defaults come from the session rather than being hardcoded), carries them in two new DeltaSparkScanCommon fields, and the native resolver uses them in the version-absent arm. Your two cases are regression tests built on raw parquet-mr files with no Spark metadata, converted to Delta: corrected 1500-01-01 and the pre-1970 microsecond timestamp now read verbatim natively under CORRECTED, rebase under LEGACY, and fail loudly under EXCEPTION. One nuance mirrored deliberately: conf LEGACY resolves the writer zone from file metadata; Spark falls back to the JVM default zone, which native cannot replicate, so non UTC zones keep the refuse-ancient posture for timestamps while dates rebase fully since they are zone free. Nested types: the wrapper now checks the applicable leaf policies before rejecting, recursing through struct, list, map, and dictionary. Your STRUCT with a DATE leaf under corrected dates plus legacy INT96 stays native and correct, tested with modern and null values; a nested column with a genuinely affected leaf still declines. The bonus bug: with no pushed predicate and identical logical and physical schemas, the DataFusion opener skips the expression adapter entirely, which is exactly the shape of a metadata-free file, so even the refusal path was silently bypassed. Rebase-enabled scans now stamp a schema marker that keeps the adapter engaged, with a test pinning the ancient-value refusal on that exact shape. Rebase expression benchmark as requested (M5, release, real module code): per row the wrapper costs 0.7 to 2.2ns depending on type and policy, 0.26 to 0.29ns for the check-only path, against effectively zero unwrapped. End to end on 20M rows the full-scan cost is noise (0.225 to 0.235s across corrected, legacy modern, legacy ancient; Spark fallback 0.36 to 0.47s). The honest cost is selective predicates: the opaque wrapper disables pruning on wrapped columns, and on a sorted table a point predicate decoded 2,627,592 rows against 40,000 with pruning live, a 65x read amplification that warm cache hides but cold or remote storage will not. That is the operational tradeoff of the legacy path; correct results either way, counts identical across all modes. Batteries at this head: 231/231 on both spark-3.5 and spark-4.0, 327 native lib tests, clippy and fmt clean. |
48e5d29 to
053ec14
Compare
|
Thanks for the depth, and for the independent verification of the DV decode path. Timing note first: your review and my last push crossed. 053ec14 (pushed about an hour after your review) plumbs the effective session read modes into the scan, so the resolve_spec comment you quoted is gone: metadata-free files now follow the resolved datetimeRebaseModeInRead and int96RebaseModeInRead exactly as ParquetFileFormat resolves them (CORRECTED reads verbatim, LEGACY rebases, EXCEPTION refuses ancient), and the nested branch now checks the applicable leaf policy, so STRUCT with a DATE leaf under corrected dates plus legacy INT96 stays native. Both of your reproductions are regression tests on raw parquet-mr files with no Spark metadata. On the sharper point that refusals bypass the decline mechanism: after the modes fix the residual execution-time refusal class is much narrower, but it is not empty and I want to be precise about what remains. It fires only when ancient values actually appear in a file whose policy resolved LEGACY with a non-UTC or unrecorded writer zone, or with contradictory legacy flags. Under EXCEPTION mode an execution-time failure matches Spark's own behavior for those values, so the true fallback-bypass is the LEGACY non-UTC sliver, where Spark succeeds using zone tables native does not have. Declining at planning time requires knowing the file's writer zone, which the JVM side cannot see without reading footers; the conservative alternative is declining every scan whenever an effective mode is LEGACY, which over-declines tables that are entirely modern or UTC-written and would give up the native path for data that reads correctly. That is a genuine tradeoff rather than an oversight, and I am happy to implement the planning-time decline for LEGACY modes if you and @sunchao prefer safety-of-mechanism over coverage here; it is a small change to the existing gate. Core-path gaps: agreed on both, and they deserve to exist independently of this PR. The shared comparator extraction was already filed as #5658; I have now filed #5662 for the concrete correctness gap you named (the regular native scan forwarding S3 options with no divergence check). The calendar gap for the plain path is #5010, and the rebase module here was built so it can be lifted wholesale when core wants it; the delta-arm flag is the isolation seam, not a design commitment. On delta in default cargo features: the motivation is exactly what you inferred, letting the contrib jar work against stock Comet binaries so opting in is a Spark-side decision rather than a custom native build. I agree the philosophy conflict with the adjacent contrib-delta comment and #5411 should be settled explicitly rather than by default; I am fine either way and will move it out of default if that is the call. Same for sequencing against #4366, which is yours and sunchao's decision to make; happy to record whatever is agreed in the Cargo.toml comment and the PR description. |
053ec14 to
70c5be8
Compare
|
One correction on the default cargo feature, since it has now come up twice: this was proposed and accepted earlier in this thread (Aug 19), and sunchao clarified on Aug 27 that the adjacent contrib-delta comment means zero kernel surface, not zero Delta surface, so the two features are not in conflict. The cost is about 82 KB in the dylib and the code is unreachable without the contrib jar plus the config. I have added a cross-reference to that decision in the Cargo.toml comment and reworded the contrib-delta one to say delta-kernel explicitly (70c5be8). |
70c5be8 to
fedc36e
Compare
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed fedc36e161404d726367b67d9edb1ca020dde0ab. The calendar fixes address the two earlier examples. Three P2 cases remain:
- [P2] The timestamp check rejects every negative value. With metadata-free INT64 TIMESTAMP_MICROS and both read modes
EXCEPTION, even one microsecond before 1970 fails. Spark 3.5 and 4.0 preserve timestamps at or after 1900-01-01 UTC. Could you use Spark's unit-adjusted cutoff and cover both sides of that boundary? - [P2] Merging datetime and INT96 modes turns datetime
CORRECTEDplus INT96EXCEPTIONintoCheckAncienteven for an INT64-only file, rejecting corrected1500-01-01. Spark selects the datetime spec for INT64 MICROS/MILLIS independently of INT96. Could you preserve physical-type attribution or decline unsupported reads before execution? - [P2] The nested branch rejects metadata-free
STRUCT<d: DATE>underEXCEPTIONbefore checking values, including modern/all-null data.LEGACYfooter metadata can also overrideCORRECTEDsession modes, so declining only session-LEGACY scans misses unsupported non-UTC files. Could you support the applicable recursive/per-file behavior or arrange Spark fallback before native admission, with Spark/native tests for these cases?
These cases also mean the description's claim that modern values always pass needs narrowing. Pruning loss follows the installed policy wrapper, including modern-only LEGACY and check-only EXCEPTION files, not only values that actually need conversion. The matched benchmark answers the earlier general request. Its approximately 65x result measures decoded rows, not cold/remote bytes or latency.
I reran five diagnostic/control probes against the retained exact module. Its source and relevant dependencies are unchanged at this head. The probes reproduce the native errors and controls, not full Spark compatibility. No full Spark/Delta query or cold-storage benchmark was run for this re-review. All six current-head workflow runs still require authorization, with no executed check results reported.
fedc36e to
52efcfe
Compare
|
All three addressed in 52efcfe, with the semantics checked against the 3.5.9 sources first. Timestamp cutoff: createTimestampRebaseFuncInRead under EXCEPTION throws only below RebaseDateTime.lastSwitchJulianTs, which computes to 1900-01-01T00:00:00Z from the rebase json (the latest zone switch across all 602 zones; UTC's own is the 1582 cutover). MILLIS columns convert to micros before the check. The native check now uses that cutoff in whichever unit the column carries, dates keep lastSwitchJulianDay, and there are tests on both sides of the boundary per unit plus the Scala regressions you asked for (a metadata-free INT64 at 1969-12-31T23:59:59.999999Z reads verbatim under EXCEPTION, one before 1900 fails loudly). INT64 vs INT96 attribution: you were right that merging the two specs was wrong, and the arrow schema alone cannot fix it since the opener runs Int96Coercer before the adapter sees anything, so INT96 and INT64 MICROS both arrive as Timestamp(us, UTC). The honest source is the footer's SchemaDescriptor physical types, so our metadata reader factory now records the INT96 leaf ordinals into the file's in-memory key-value metadata when it loads the footer, and the adapter attributes each timestamp leaf by depth-first ordinal, validated against the leaf count and replacing any pre-existing key rather than trusting it. INT64 columns follow the datetime spec and INT96 columns follow the INT96 spec independently; with attribution available, files whose two legacy flags disagree now read correctly instead of refusing. Tests cover both directions at the Rust and Spark levels, including an INT96 column written through the low-level writer alongside an INT64 one, and the old mixed-flags refusal test became a four-combination round-trip against Spark's own read. Nested types: rather than declining, the wrapper now rebuilds struct, list, large list, fixed size list, map, and dictionary arrays with each leaf transformed under its own policy, preserving nulls and offsets, so STRUCT with a DATE leaf under EXCEPTION reads modern and null data natively and only an actual ancient leaf value fails. Tests cover modern and null leaves under every policy, an ancient leaf rebasing exactly under LEGACY UTC and erroring under EXCEPTION, and list of struct of date round trips, at both levels. On the narrowing: the description now states exactly what still fails at execution time instead of reading, which after these changes is a LEGACY file with a non UTC or unrecorded writer zone when a timestamp before 1900 actually appears, and an EXCEPTION file when an ancient value actually appears. It also notes that pruning is lost on every column that receives a policy wrapper, modern only LEGACY and check only EXCEPTION files included, and that the 65x figure is decoded rows rather than cold bytes or latency. Battery at this head: 235/235 on both spark-3.5 and spark-4.0, 339 native tests, clippy and fmt clean. An independent pass over the change also caught an i64 overflow in the nanosecond identity check for the LEGACY UTC path, fixed by comparing in days, with a per-unit test. |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed 52efcfe28 against ef62b463. The three earlier examples are addressed by the new cutoff, INT64/INT96 attribution, and recursive handling of modern and null nested values.
[P2] Rebase only the requested nested leaves
Following the calendar update, the struct traversal still processes every physical child before the schema adapter narrows the struct. For a converted metadata-free file containing s.d = DATE '2024-06-01' and s.ts = TIMESTAMP '1500-01-01', with both rebase read modes EXCEPTION, requesting only s.d still checks the unrequested s.ts and raises. Spark's requested nested schema excludes that timestamp.
I independently reproduced this in a component probe with real in-memory Parquet decoding and the exact calendar module. Ordinary narrowing returns the modern date, placing the wrapper beneath that narrowing raises, and reading only the requested Parquet leaf succeeds. The probe uses DataFusion's cast for narrowing. The exact Comet cast and complete adapter/opener path were source-traced, not executed as a full Spark/Delta query.
Could we restrict rebasing to requested leaves while retaining the footer's INT96 ordinal attribution, and add a Spark/native regression for this projection?
The five current-head workflows remain action_required, with no head or merge check results. No full Spark/Delta/JNI run or new benchmark was performed.
Sorry. Posted before from work account. Original message: No, but I am open to using one if you guys have one. Also open to using discord, since free slack caps message history |
52efcfe to
4a57686
Compare
|
Fixed in 4a57686. The per file policies are still computed against the physical struct so the footer's INT96 ordinal attribution stays valid, but the wrapper now derives a requested mask by pairing the physical type with the requested logical type through exactly the pairings the struct conversion narrows (struct children by name under the adapter's case rules, list elements, map entries with matching key ordering); every physical leaf outside that mask is marked corrected and never checked. Anything the conversion does not narrow (large lists, fixed size lists, dictionaries, mismatched map ordering) keeps every leaf, so the mask is a superset of what the reader materializes. Your probe is the Spark regression: a converted metadata-free file with s.d modern and s.ts ancient under EXCEPTION read modes, select s.d reads natively and matches Spark, while selecting s.ts still fails loudly. Rust unit tests cover the same at the wrapper level plus a list of struct variant and the non narrowed shapes. Batteries at this head: 236/236 on both spark-3.5 and spark-4.0, 343 native tests, clippy and fmt clean. Independent review of the change caught that the first cut recursed through list and dictionary pairings the conversion never narrows and a first-match tie-break mismatch on folded name collisions; both fixed before this push. |
|
@dwsmith1983 @ErikBPF we have a Slack channel in the ASF workspace. Could you share your email address? I can add both of you there. |
|
invite sent |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
The requested-leaf mask addresses my nested-projection P2. At 4a57686cc against ef62b463, selecting modern s.d now succeeds beside an unrequested ancient s.ts, while requesting s.ts or the whole struct still raises under EXCEPTION. The mask preserves physical leaf ordinals, and the INT96/INT64 attribution controls pass. I found no remaining verified P1/P2 in the reviewed scope.
I independently reran the 39 current calendar-module tests and four diagnostic/control probes, all 43 passed. These use real in-memory Parquet decoding and the exact calendar module, with component-level narrowing. The complete Comet adapter/JNI/Spark/Delta path, non-ASCII JVM folding and cloud reads were not executed. The added Spark regression was inspected, not run. No new benchmark was performed.
At the fresh CI snapshot, 21 checks were successful, 7 skipped and 12 still running, with no failures reported. CI and PyArrow workflows are still in progress.
Adds an optional contrib/delta-spark module that claims delta-spark DSv1 scans through CometScanContrib and runs them on Comet's shared native parquet path, including main's JVM-exact field-name folding for case-insensitive footer matching. Deletion vectors are decoded natively into per-file ParquetAccessPlans that DataFusion intersects with row-group and page-index pruning, so DV skips and page skips compose in a single scan. Scans the native path cannot serve safely (DML row-index reads, unsupported filesystem schemes, userinfo-bearing authorities, credential-provider-only auth, S3 config divergence, multi-store shapes) fall back to Spark with an explained reason. Co-authored-by: Scott Schenkein <schenksj@yahoo.com> Co-authored-by: Aditya Vaish <adivaish@microsoft.com>
4a57686 to
79b7b78
Compare
|
@andygrove could you take another look when you have a chance? Since your last review the PR was restructured along the lines you asked for: the case folding is gone in favor of #5602, the core fixes went out as #5653 (now merged), and the field id semantics are in #5654. The branch is a single commit on current main, sunchao approved at the previous head, and CI is green apart from the runs still waiting on authorization. Your changes-requested predates the restructure, so I wanted to make sure it is not holding things up by accident. |
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed 79b7b785c against 55ae4f20. The rebased Delta feature patch is unchanged after excluding Git blob headers, including the requested-leaf calendar fix. I checked the intersecting dependency and shared-scan boundaries and found no remaining P1/P2.
No product tests or benchmarks were rerun. The previous 43 calendar-component checks are historical evidence, reused only after matching the exercised source and dependency identities. At 04:52 UTC, CI had 57 successful checks, 19 running and 7 skipped, with no failures reported. This is not full current-head Spark/Delta/JNI validation.
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.