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 53 commits into
Conversation
|
Update: pushed two follow-up commits extending the scan's pruning and object-store behavior.
|
…ng + in-scan DVs) Adds contrib/delta behind a -Pdelta profile: delta-spark keeps all planning (log replay, snapshot, partition pruning); Comet claims the DSv1 scan via a new CometScanRuleExtension SPI and reads data through the same native DataFusion parquet path as CometNativeScanExec, inheriting row-group stats pruning, page-index pruning, and filter pushdown. Deletion vectors are decoded natively into per-file ParquetAccessPlans that intersect with page-index pruning, so deleted rows are skipped in-scan; DV blob and footer fetches run concurrently and footers go through the scan's shared FileMetadataCache (no extra metadata round-trips for DV files). Scalar-subquery data filters are resolved at execution time and pushed to the native reader — a capability stock Spark 3.x lacks entirely. Column mapping name mode, DPP, time travel, checkpoints, schema evolution, and INT96 covered. The `delta` native feature ships in the default set: runtime stays double-gated (contrib jar via ServiceLoader + conf), so it is inert for non-Delta users; roaring is the only net-new default dependency. CI runs the contrib suites on Spark 3.5/4.0/4.1, byte-compiles the dev scripts on Python 3.11-3.14, and keeps the feature-off error path tested. Verified: 39-test differential suite green on Spark 3.5/4.0/4.1; Delta's own suites with Comet injected fully green, 1156/1156 (DeletionVectors, TimeTravel, ColumnMapping, DeleteSQL, UpdateSQL, MergeIntoSQL — one test-only harness patch maps the Comet scan node to its originalPlan for Delta's ScanReportHelper). Local bench (20M rows, release): 1.35x vs stock at 9.4% of bytes on literal bounds; 3.35x on subquery bounds (stock scans 100%, contrib 5%). Supersedes apache#4366 (delta-kernel-rs contrib) and apache#4669 (plain-table native scan), deliberately building on both. Co-authored-by: Scott Schenkein <schenksj@yahoo.com> Co-authored-by: Aditya Vaish <adivaish@microsoft.com>
888e4a7 to
7fd81aa
Compare
|
HI @andygrove, Can you review this as it adds Delta functionality? |
Two deficiencies surfaced by 'make release PROFILES="-Pspark-3.5,delta"' (the path a vendor uses to package the contrib), neither visible in CI: - CometScanRuleExtension scaladoc violated spotless line-wrapping (mvn spotless:apply; CI's contrib job builds deps with -Dspotless.check.skip=true so it never checked this file). - BanDuplicateClasses fired for five comet-common exception classes: the comet-spark shaded jar bundles comet-common, so inside a single reactor the contrib sees both artifacts (the dependency-reduced pom only shields repository consumers), and shade's ASM pass renumbers some constant pools so ignoreWhenIdentical cannot collapse them. Ignore the org.apache.comet.* overlap for that pair explicitly. Verified: ./mvnw install -Prelease -DskipTests -Pspark-3.5,delta now completes with no skip flags.
isDeltaScan used classOf[DeltaParquetFileFormat], which resolves the Delta class on the FIRST V1 scan the extension inspects. With the contrib jar deployed but delta-spark not on the classpath, that raises NoClassDefFoundError inside CometScanRule and takes down every parquet scan in the session - the exact opposite of the module's inert-by- default contract (found live: a parquet-only benchmark arm with the contrib jar staged died on its first query). Compare the class NAME instead: no Delta type is touched until the name matches, and a match proves delta-spark is present (the instance exists), so every Delta reference past this gate stays safe. Exact string equality preserves the previous exact-class semantics. Verified: compiled bytecode of isDeltaScan carries only a string constant (javap: getName + ldc + String.equals - no Delta constant- pool entry); compile, spotless and scalastyle green.
|
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! |
Reconciles this PR's JVM-planned Delta contrib with the contrib infrastructure that landed in apache#4952 (kernel-path build gate + inert wiring): - The JVM-planned scan now rides the generic ContribScan envelope (type_url comet.contrib.delta_spark.DeltaSparkScan) instead of its own oneof slot; delta_scan = 119 is removed and reserved. The native handler moves to planner/delta_spark_scan.rs, a sibling of the kernel path's delta_scan.rs, dispatched by type_url. - Our proto messages are renamed DeltaSpark* so both contribs' message sets coexist in operator.proto without collision. - DeltaPlanDataInjector keys on CONTRIB_SCAN and disambiguates by type_url (the injector registry already supports multiple injectors per kind). - The module relocates to contrib/delta-spark/ so apache#4952's -Pcontrib-delta add-source of contrib/delta/src no longer overlaps our sources; our conf object is renamed DeltaScanConf since both contribs share the org.apache.comet.contrib.delta package. - CometNativeScan keeps the buildNativeScanCommon refactor and grafts in main's Variant-schema substitution; CometScanRule carries both claim hooks (CometScanContrib and CometScanRuleExtensions). - verify-contrib-delta-gate.sh needs no changes: its assertions target kernel-path identifiers and return zero hits on this tree.
|
@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.
| val filtersAboveScan = plan.collect { | ||
| case f: org.apache.spark.sql.execution.FilterExec if f.find(_ eq scanExec).isDefined => f | ||
| } | ||
| filtersAboveScan.lastOption | ||
| .map { f => | ||
| splitConjunctivePredicates(f.condition) | ||
| .filter(_.references.subsetOf(scanExec.outputSet)) |
There was a problem hiding this comment.
[P1] Keep scalar-subquery pushdown above non-commuting operators
Could we restrict this walk to plan shapes where moving the predicate into the scan is safe? The reference check does not protect against LIMIT or TopN. With a Delta table t containing IDs 0, 1, and 2, SELECT id FROM (SELECT id FROM t ORDER BY id LIMIT 1) q WHERE id > (SELECT max(id) FROM range(1)) must return no rows. A stock Spark 4.0.3 / Delta 4.0.0 probe produces Filter -> TakeOrderedAndProject -> FileSourceScanExec, with empty scan dataFilters, and this guard accepts the outer predicate. Moving id > 0 below TopN returns row 1 instead. The execution-time path then installs that predicate in the native Parquet scan, so enabling row-filter pushdown can change the result. Please keep the covering filter and only harvest across operators for which pushdown is valid. This query would make a useful regression test.
There was a problem hiding this comment.
Fixed. The harvest now walks the plan from the covering filter down to the scan and only accepts Project and Filter nodes in between, anything else stops it, plus a deterministic check on the conjuncts. Your exact query shape is a regression test now: empty result, scan still native, and an assertion that nothing was pushed. The covering filter is never removed, so this only prevents evaluating the predicate too early.
There was a problem hiding this comment.
[P1] Also stop at nondeterministic projections
The LIMIT/TopN case is fixed at 7e09e04f, but spineToScan still crosses every ProjectExec. A deterministic conjunct does not commute with a nondeterministic projection. For a single-file Delta table t containing IDs 0–4:
SELECT id
FROM (SELECT id, monotonically_increasing_id() AS seq FROM t) q
WHERE id > (SELECT max(id) FROM range(1)) AND seq = 1An isolated probe using the unchanged current-head harvesting method on a real Spark 4.0.2 / Delta 4.0.0 physical plan harvests id > scalar-subquery; applying that resolved predicate immediately above the scan changes the result from ID 1 to ID 2. With spark.comet.parquet.rowFilterPushdown.enabled=true, the native path can perform that same early filtering. Keeping the covering filter does not restore the sequence values assigned to the surviving rows. This was an exact-method/physical-plan probe, not a full Comet/JNI run.
Could we require deterministic projection expressions before crossing a ProjectExec, as Spark's predicate-pushdown rule does, and add this regression?
There was a problem hiding this comment.
Fixed. The spine walk now requires every projection expression to be deterministic before crossing a ProjectExec, mirroring PushPredicateThroughNonJoin's guard, and intermediate filters get the same check on their condition as belt and braces (near unreachable today since the harvest picks the lowest qualifying filter, but it makes the invariant explicit). Your exact query is a regression test with rowFilterPushdown enabled: result is ID 1, the scan stays native, and the test asserts the predicate was not harvested. One deliberate non-change: a nondeterministic conjunct in the same FilterExec still lets its deterministic siblings push, matching FileSourceStrategy.
| let (object_store_url, mut files) = | ||
| planner.prepare_scan_store_and_files(common, &spark_partition)?; |
There was a problem hiding this comment.
[P1] Preserve the object-store authority of every Delta data file
Could we decline mixed-store scans before claiming them, or preserve each file's actual store identity? prepare_scan_store_and_files chooses the first file's store, while get_partitioned_files reduces every URL to its path. A shallow clone from bucket A into bucket B followed by an append is a valid Delta table containing absolute source files and new target files. The size-based file packing can put both in one task, which then requests B's key from A. That normally produces NoSuchKey, but it can read the wrong data if the same key exists in A. This assumption was already present in the generic helper, but the new Delta reader makes it reachable through ordinary shallow-clone behavior. Please cover a cross-bucket clone plus append and preserve Spark's file order if the scan is split internally.
There was a problem hiding this comment.
Went with decline rather than per-file routing. The claim gate now compares the full lowercased URI authority (userinfo included, so abfss containers count as distinct) across all selected files, with a native check as backstop, and file order is never changed on the native side. One residual worth flagging: the shared object-store cache key in core drops userinfo, so a DV sidecar on a different container of the same account could still resolve through the wrong store. That's in the shared store layer and predates this PR, so I left it as a core follow-up rather than changing cache semantics here.
There was a problem hiding this comment.
[P1] Decline the residual cross-container DV case before claiming the scan
The mixed-data-file guard fixes the original case. For the residual you identified, could we add a Delta-side decline guard until store identity is fixed? A shallow clone from abfss://source@account.dfs.core.windows.net/... to abfss://clone@account.dfs.core.windows.net/..., followed by DELETE, can keep all data files in source while putting the new DV in clone. Both current authority gates accept it because they inspect only data files.
At 7e09e04f, an offline probe using the exact cache/prepare functions, new resolver, and real Azure store builder confirms that both URLs map to the same cache/registry identity: the DV resolves to MicrosoftAzure { container: source }, and its handle is pointer-identical to the data store. The DV read therefore requests the clone-relative path from the wrong container. Normally that fails with a missing object; a matching, well-formed object could instead supply the wrong deletion bitmap.
The helper predates this PR, but the new single-scan data-plus-DV path makes this reachable. Declining sidecars whose full authority differs but collides in the current store identity would contain it without a broad cache refactor. Fixing only the local resolved_stores key is insufficient because the shared cache and DataFusion registry also collapse the container. No live Azure request was used in the probe.
There was a problem hiding this comment.
Added, both sides. At claim time the JVM declines when any two URIs the native side resolves stores for (data files and DV sidecars) share the native store-identity key but differ in userinfo, which is exactly your source/clone container case. The JVM key is a deliberate conservative superset of the native one (it also lowercases host and port and always rewrites s3a to s3), so it can only over-decline, never miss a collision native would hit. The native side got a matching check at store-resolution time, which sees DV paths too; it's deliberately not an extension of the data-file authority check so the legitimate cross-bucket DV shape (data in A, sidecar in B) still claims, which the MinIO suite now proves live. Remaining honest gaps, all narrow: the JVM compares decoded authorities while native parses the url-encoded form, so a userinfo differing only by percent-encoding would pass the gate and hit the native error (not constructible from real container names); the native compare uses the username without the password component; and the native check is per partition while the JVM gate is whole-scan.
| if row > cursor { | ||
| selectors.push(RowSelector::select((row - cursor) as usize)); | ||
| } | ||
| // Merge runs of consecutive deleted rows into one skip. | ||
| match selectors.last_mut() { | ||
| Some(last) if last.skip => last.row_count += 1, | ||
| _ => selectors.push(RowSelector::skip(1)), |
There was a problem hiding this comment.
[P1] Bound memory used by expanded DV row selections
Could we account for and bound these allocations, preferably creating the access plan when its file is opened? An alternating deleted/retained bitmap creates one non-coalescing RowSelector per row. A bounded probe using these unchanged functions and the exact Roaring 0.11.4 / Parquet 58.4.0 types expanded an 8,224-byte bitmap for 65,536 rows into 65,536 selectors, retaining 1,048,600 bytes and peaking at 2,097,176 bytes. An 8-million-row group therefore needs roughly 128 MiB for the retained selectors alone. attach_access_plans().buffered(8).try_collect() bounds fetch concurrency, but retains all completed file plans before the scan starts. None of this allocation is reserved against the execution memory pool. A multi-file test with distinct alternating-row DVs would exercise the executor-OOM case without relying on malformed input.
There was a problem hiding this comment.
Two layers now. At planning, spark.comet.scan.deltaNative.dv.maxDeletedRowsPerFile (default 1M) declines pathological DVs before they reach native; the bound is cardinality-based so it's deliberately pessimistic, a contiguous delete declines the same as an alternating one, but run structure isn't knowable without decoding bitmaps on the driver, and the knob makes it recoverable. Natively, the exact selector bytes (derived from size_of::<RowSelector>()) are reserved against the memory pool, released with the plan, with multi-file accumulation and release-on-failure covered by tests. I looked at building the access plan at file-open time as you suggested, but DataFusion 54.1's ParquetOpener reads the plan out of the file extensions in create_initial_plan with no hook to compute it lazily, so that needs an upstream API first.
There was a problem hiding this comment.
[P2] Reserve construction memory and account for the reader's copy
The cardinality cap and retained-plan reservations improve the original issue, but two allocations still bypass the pool at 7e09e04f. build_access_plan finishes before try_grow, and Parquet's RowSelection::from(Vec) allocates a second vector during construction. Later, DataFusion 54.1's create_initial_plan deep-clones the attached access plan while the original remains in the file extensions, without an additional DV reservation.
A counting-allocator probe using the unchanged build_access_plan/total_selectors functions and the locked dependencies tested 2,000,000 rows with exactly 1,000,000 alternating deletions, which the default cap admits. It measured 65,554,457 bytes of peak construction allocation before a 1-byte pool rejected the reservation. With a 32,000,000-byte reservation accepted, the opener-equivalent clone allocated another 32,000,025 bytes while the pool stayed at 32,000,000. These are allocator-requested bytes, not RSS or a reproduced executor OOM; reservation release on drop works correctly.
Could we pre-reserve a conservative construction bound and then shrink it, and either transfer ownership or account for the active reader's copy? The current check can reject after memory pressure has already occurred. I would treat this as a remaining P2 under the new cap, rather than the original unbounded P1.
There was a problem hiding this comment.
Reworked as you suggested. Before build_access_plan runs (row-group count is known from the footer at that point), we reserve a construction bound of 3 x (2*cardinality + row_groups) x size_of::(); the factor is derived, not guessed: with r selectors retained from finished groups and c in the current group, r + c <= the per-file bound, and the peak is r + 2c (source Vec at doubling capacity) + c (the RowSelection::from copy) <= 3(r + c). After the build we resize down to 2 x the exact retained bytes, the second factor covering create_initial_plan's deep clone alongside the original in the file extensions; since retained <= bound this always shrinks. A rejection now happens before any large allocation, with a distinct construction-phase error, and a test pins that ordering (pool sized to the steady state, below the construction bound). Two things to be upfront about: the bound is worst-case-alternating, so a large contiguous delete now reserves far more than it retains (your 1M example would reserve ~96MB to keep ~48 bytes) and rejection is an execution-time error rather than a planning fallback; the exact selector count is computable in O(runs) from the already-decoded bitmap, so making the reservation exact is a clean follow-up if you'd like it. The DV blob and decoded bitmap themselves (~250KB at the default cap) remain outside the pool.
There was a problem hiding this comment.
[P2] Also account for the reader's combined-selection allocation
Construction admission and the initial reader clone are now covered at bc98657f. One later allocation is still outside the reservation: DataFusion 54.1's build_stream calls ParquetAccessPlan::prepare -> into_overall_row_selection, which collects another RowSelection while the attached original and the consumed clone's backing vector remain live. The resize at delta_dv.rs:504-507 has already reduced the reservation to twice the retained selector bytes by then.
I verified this using the unchanged current attachment code and the real locked DataFusion/Parquet conversion. For one 2,000,000-row group with exactly 1,000,000 alternating deletions (permitted by the default cap):
- Construction reserves 96,000,048 bytes, then attachment reduces the reservation to 64,000,000 bytes.
- The attached selectors plus reader-normalization allocations peak at 97,554,457 bytes and retain 65,554,432 bytes afterward, while the reservation remains 64,000,000.
These are allocator-requested bytes, not RSS or a reproduced executor OOM; the dependency conversion was invoked directly, with its production call path and allocation lifetimes checked in source, rather than running a complete Comet scan. Reservation release still works. Could we account for normalization and vector capacity before it allocates, or transfer ownership to avoid the extra buffer? Simply changing the steady-state factor to 3 would still fall below this measured peak. This is a remaining part of the existing memory P2, not a separate finding.
| 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.
| CometNativeScan.buildNativeScanCommon( | ||
| source = scanExec.simpleStringWithNodeId(), | ||
| output = scanExec.output, | ||
| requiredSchema = toPhysical(scanExec, scanExec.requiredSchema), | ||
| dataSchema = toPhysical(scanExec, relation.dataSchema), |
There was a problem hiding this comment.
[P2] Restore logical nested names in the scan output
Could we separate the physical read schema from the logical output schema and restore nested field names before parent expressions run? toPhysical rewrites nested names as well as top-level names. The shared native builder returns that physical schema directly, and the native ToJson implementation takes JSON keys from the actual Arrow struct fields. For a name-mapped Delta column s STRUCT<a: BIGINT>, enabling spark.comet.expression.StructsToJson.allowIncompatible=true makes SELECT to_json(s) reach this path. Spark returns {"a":7}, but the native expression receives the physical col-... name. A stock Delta 4 probe confirms the logical and physical schemas differ this way. Please cover a mapped nested struct with a native expression that observes field names, rather than only positional field access.
There was a problem hiding this comment.
For now this declines column mapping over nested structs rather than restoring the names. Restoring needs a logical-schema field in the proto plus a native rename step, since the existing adapter remap is top-level only and field-id based, and we strip field ids to support pre-upgrade files. The decline does cost the ordinal-access case that worked before, which I converted into an explicit fallback test rather than leaving it implicit. Happy to do the restore-names design as the follow-up that lifts the gate if that ordering works for you.
| let deleted = deserialize_dv_bitmap(&data) | ||
| .map_err(|e| GeneralError(format!("Invalid deletion vector for {file_path}: {e}")))?; |
There was a problem hiding this comment.
[P2] Validate the decoded DV cardinality
Could we compare deleted.len() with dv.cardinality before applying the bitmap? The descriptor carries the expected count, but the native reader never checks it. A valid-CRC bitmap containing only row 1, paired with a descriptor declaring two deletions, is accepted by the current decoder and access-plan builder and selects 9 of 10 rows. Delta's JVM reader rejects this mismatch in StoredBitmap.validateCardinality. CRC and row-range validation do not catch a stale but otherwise well-formed bitmap. Please add a mismatched-cardinality case alongside the existing corruption checks.
There was a problem hiding this comment.
Fixed as suggested: decoded cardinality is validated against the descriptor right after decode, mirroring StoredBitmap.validateCardinality, with negative cardinality rejected alongside the existing negative-size check. The mismatch test sits next to the framing/CRC cases.
| let dv = match dv { | ||
| Some(dv) => dv, | ||
| None => return Ok(file), | ||
| }; |
There was a problem hiding this comment.
[P2] Treat an empty DV as a pass-through file
Could we handle the canonical zero-cardinality descriptor before attempting bitmap decoding? Delta's DeletionVectorDescriptor.EMPTY has inline storage, an empty payload, size 0, and cardinality 0. I checked a committed Delta 4 table containing that exact descriptor: the actual Spark FilePartition carries it and the normal Delta reader returns all rows. Here it becomes Some(empty_bytes) and deserialize_dv_bitmap returns Deletion vector bitmap too short for magic number. Returning the unchanged file for the empty-DV case would match Spark. Please add this to the existing attach_access_plans test.
There was a problem hiding this comment.
Fixed: the canonical zero-cardinality, zero-size descriptor now short-circuits to a pass-through file before any bitmap read or footer fetch. Added to the existing attach_access_plans test as a fourth file asserting no access plan, preserved order, and no footer fetched.
| val firstFileUri = scanHelper.selectedPartitions | ||
| .flatMap(_.files.headOption) | ||
| .headOption | ||
| .map(_.getPath.toUri) |
There was a problem hiding this comment.
[P2] Include configuration for external DV providers
Could we derive execution-time object-store options from every selected data-file and external-DV authority, or decline combinations that cannot be represented? The common configuration is extracted using only this first data file's scheme, while extractDvDescriptor permits an absolute DV URI on another provider. For an S3 data file with an ABFS sidecar, NativeConfig forwards fs.s3a.* but omits the Hadoop Azure account-key/OAuth settings. The native DV loader then opens the ABFS URI using that same S3-only map, so a table readable by Spark can fail authentication. This is separate from mixed-bucket data-file routing and occurs with a single data file. Same-scheme S3 bucket overrides are already forwarded. A cross-provider sidecar test would cover the missing case.
There was a problem hiding this comment.
Went with derive rather than decline: options are now unioned over the data-file authority, the table root, and every on-disk DV authority, deduped per authority. The extracted prefixes are scheme-disjoint so the union can't collide. Cross-provider coverage is a unit test (S3 data plus ABFS sidecar, asserting both prefixes appear and that an S3-only set gets no azure keys), since a same-scheme MinIO scenario wouldn't prove anything, option extraction is scheme-global.
|
Thanks @dwsmith1983. On the design topics you flagged, I’d prefer using |
check_same_object_store_authority compared raw url[BeforeHost..AfterPort] slices with only the scheme lowercased. The url crate does not lowercase hosts for opaque (non-"special") schemes like s3a/abfss/hdfs, so the same bucket recorded with mixed host casing (s3a://Bucket-A vs s3a://bucket-a) passed the JVM-side gate (which does lowercase) but hard-errored here instead of matching it. Build the authority from an explicitly lowercased scheme and host plus Option<u16> port, mirroring DeltaScanSupport.uriAuthority.
The Delta scan's object-store options were extracted from only the first selected data file's URI, so a DV sidecar living on a different provider or bucket than the data files (e.g. S3 data + ABFS DV) never received its own credentials. Union extractObjectStoreOptions over the data-file authority, the table root, and every distinct on-disk DV authority instead; the extracted keys are scheme-disjoint prefixes so the merge cannot leak one provider's credentials into another's map. Adds DeltaScanSupport.selectedDvDescriptors, a planning-time helper that deserializes each selected file's DV descriptor once and normalizes it to an absolute path, shared with the upcoming DV cardinality gate.
Factor the DV-authority URI assembly out of CometDeltaNativeScan.convert into a standalone storeUris(descriptors, tableRootPath, firstFileUri) helper, and unit-test it directly with hand-built DeletionVectorDescriptor fixtures (absolute, UUID-relative, and inline storage types). A Spark-session test using local file:// paths can't reach this logic, since extractObjectStoreOptions returns an empty map for that scheme.
… pool An alternating deleted/retained bitmap expands into one non-coalescing RowSelector per row with no upper bound and no memory-pool accounting, risking executor OOM on a large DV. Adds native accounting (MemoryReservation sized from size_of::<RowSelector>(), never hardcoded, attached alongside the ParquetAccessPlan and released on drop) plus a JVM claim-time decline gate (spark.comet.scan.deltaNative.dv.maxDeletedRowsPerFile, default 1000000) built on selectedDvDescriptors to reject oversized DVs before native execution, pessimistically but soundly (selectors <= 2*cardinality + #row-groups).
Adds CometDeltaS3Suite (contrib/delta-spark), mixing CometDeltaTestBase
with the spark module's CometS3TestBase, covering the two scenarios
that need real multi-bucket object storage:
(a) shallow clone bucket A -> B + append -> declines with the
multi-store fallback reason, checkSparkAnswer still matches.
(b) clone A -> B + DELETE on the clone (DV sidecar lands in B, data
files stay absolute in A) -> reads correct rows natively through
a cold bucket-B object store. Deliberately does not trip the
multi-store gate (data authorities = {A} only).
A per-bucket-credentials scenario is intentionally omitted:
extractObjectStoreOptions is scheme-global, so it would add nothing
beyond the existing cross-scheme unit test in DeltaScanContribSuite.
Unlike ParquetReadFromS3Suite, which CI never discovers (the PR
workflows enumerate suites by name and simply omit it), the contrib
module's CI job runs a blanket `mvn test` with no such allowlist, so
this suite IS discovered. Each test opens with
`assume(dockerAvailable, ...)`, which ScalaTest reports as CANCELED
rather than failed when no Docker daemon is reachable.
contrib/delta-spark/pom.xml gains test-scope testcontainers:minio and
awssdk:s3, matching the spark module's existing GAVs for
CometS3TestBase.
The clone-then-DELETE scenario asserted only end-to-end behavior
(native scan + row count), so a change in Delta's shallow-clone or
DELETE-on-DV semantics could silently stop exercising the cold
cross-bucket object-store path while the test kept passing.
Before the read, inspect the clone's live Delta snapshot
(DeltaLog.forTable(...).update().allFiles) and assert structurally:
- at least one AddFile resolving into bucket A (data still absolute
into the source, untouched by the clone), and
- at least one non-inline, non-empty deletionVector descriptor
resolving into bucket B, using the same copyWithAbsolutePath +
absolutePath resolution DeltaScanSupport.selectedDvDescriptors and
CometDeltaNativeScan.storeUris use in production.
Both assertions run inside the same assume(dockerAvailable) guard as
the rest of the test.
The JVM decline gate and the native defense-in-depth check both keyed object-store authority on scheme/host/port alone, which drops URI userinfo and mishandles reg-names URI#getHost cannot parse: - abfss://containerA@account vs abfss://containerB@account collapsed to one authority, so cross-container shallow clones were not declined. - URI#getHost (and getUserInfo/getPort) return null for the whole authority on underscore reg-names (gs://my_bucket), so the JVM gate passed scans the native side then hard-errored on. Both sides now key on the URI's raw authority (JVM: getAuthority; native: scheme/username/host/port), keeping their equivalence classes in sync so the JVM gate always declines before the native check would ever error. Reword the native comment claiming a later data file can live under a different store -- impossible now that authority is verified up front. Also dedup CometDeltaNativeScan.storeUris by (scheme, authority) instead of by full URI, so N deletion-vector files on one external store contribute one representative URI to the object-store options merge instead of N.
Add a two-file variant of the DV access-plan reservation tests: reserved bytes sum across files while the returned Vec is alive, drop releases everything, and a pool sized to fit only the larger file rejects the batch cleanly with no leaked reservation from a file that transiently succeeded first. Also drop redundant `s"..."` interpolation prefixes flagged by scalafix on literal segments with no interpolated values.
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.
| val nonProjectConsumer = plan.exists { | ||
| case _: ProjectExec => false | ||
| case n if n ne scanExec => | ||
| n.expressions.exists(_.references.exists(r => tainted.contains(r.exprId))) |
There was a problem hiding this comment.
[P2] Track live row-index values across Union output remapping
Could we propagate row-index dependencies through UnionExec's positional outputs, or conservatively decline this shape? The current analysis only propagates through ProjectExec aliases. UnionExec.expressions is empty and its output uses the first child's expression IDs, so a live row-index alias in the second branch disappears from both checks here. With native Delta scans enabled and AQE disabled, this valid query over DV-backed tables reaches the gap:
SELECT id, _metadata.row_index AS ri FROM delta.`t1`
UNION ALL
SELECT id, _metadata.row_index AS ri FROM delta.`t2`I ran the unchanged rowIndexUnusedAbove method on actual Spark 4.0.2 / Delta 4.0.0 plans. Each table had one file containing IDs 0–4, with ID 2 deleted from the first and ID 3 from the second. The first scan correctly returns false, but the second returns true; its other admission inputs pass. The DV serializer then supplies 0L for that supposedly unused row index. Modeling exactly that substitution changes the second branch's row indexes to zero and SUM(ri) over the union from 15 to 8. This is an exact-method/physical-plan probe, not a full Comet/JNI execution. Please add second-branch selection and aggregate-over-union regression coverage.
There was a problem hiding this comment.
Took the positional-propagation option, since a blanket decline would reject every DV-backed UNION ALL (Delta appends the row-index column to requiredSchema on all DV reads). The taint fixed point now maps child.output(i) to union.output(i) for every branch, covering UnionExec and CometUnionExec; a CometUnionExec whose frozen output ever diverges in arity from its re-parented children declines outright rather than zipping silently. There's also a generic safety net now: any node with two or more children that isn't a positional union, where a tainted child attribute neither appears in the output by exprId nor in the node's expressions, declines, so this gap class can't silently recur (joins pass since they carry child exprIds; semi/anti joins trip only on eliminated-side taint, where declining is correct). Your exact query plus second-branch selection and aggregate-over-union are regression tests asserting values, SUM(ri) is 15, and an anti-regression proves DV unions without _metadata still claim both branches natively. One residual documented in the code: ReusedExchangeExec has the same positional shape but is unreachable at claim time because ReuseExchangeAndSubquery runs after the columnar rules.
|
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). |
…n it is unreachable
…ntrib/delta-spark
…er deletion vectors
…stores A Delta shallow clone across containers on one storage account can leave data files in one container and a later DELETE's deletion vector in another. Both authorities collapse onto the same cached ObjectStoreUrl (userinfo is dropped when computing the store key), so the DV read would silently request the clone-relative path through the wrong container's store handle. Add a collision check inside resolve_store, the one place in the scan that resolves both data and DV URLs, tracking the userinfo seen per resolved store and erroring on a mismatch. Distinct-bucket DV placement (e.g. MinIO/S3 cross-bucket) is unaffected since it resolves to distinct stores in the first place.
…ss plans build_access_plan finished before the pool reservation ran, and parquet's RowSelection::from(Vec) allocates a second vector while building each row group's selection, so peak construction allocation exceeded the reserved byte count. DataFusion 54.1's create_initial_plan also deep-clones the attached access plan while the original stays alive in the file's extensions, with no reservation covering the second copy. Pre-reserve a conservative bound (3x the worst-case 2*cardinality + num_row_groups selector count, derived from a measured peak allocation) before build_access_plan runs, then shrink the reservation to 2x the plan's actual retained selectors afterward to cover both the original and DataFusion's clone. A rejection now happens before any large Vec is allocated instead of after.
…mment The steady-state resize comment substituted construction_bytes for two different quantities: construction_bytes is already 3*S (S being the selector-bound term), so "2*construction_bytes < 3*construction_bytes" worked out to 6S < 9S, not the actual invariant. Name the intermediate S (selector_bound_bytes) inside construction_bound_bytes and restate the call site's comment as the true chain: retained <= S, so 2*retained <= 2*S < 3*S = construction_bytes, the amount already reserved.
storeIdentityKey's scaladoc claimed it matches the native url_key exactly and that host-case-differing URIs build one native store; url_key never lowercases, so those URIs actually get two native stores. Reword to state the true relationship: storeIdentityKey is a deliberate superset of the native key, so the collision gate can only over-decline, never miss a real collision. CONSTRUCTION_PEAK_FACTOR's derivation summed 1x + 2x + 1x = 4x instead of supporting the declared factor of 3. Replace it with the joint accounting over retained (r) and current-group (c) selectors, r + c <= S, giving a peak of r + 3c <= 3S.
|
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.
|
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. |
4 participants
Which issue does this PR close?
Part of #174 (Explore integration with Delta Lake). It does not close #174, that issue also tracks writes, CDF, and broader integration; this PR delivers the native read path.
Supersedes two earlier efforts, and deliberately builds on both (both given co-authored by since ideas were learned and borrowed):
Rationale for this change
Comet currently falls back to Spark's reader for all Delta tables (
isFileFormatSupportedrequires exactParquetFileFormat, andDeltaParquetFileFormatis a subclass). That forfeits native execution and all of Comet's parquet pruning on one of the most common table formats.Key observation: delta-spark has already done log replay, snapshot resolution, time travel, and partition pruning by the time
CometScanRulesees theFileSourceScanExec. So no Delta planning is needed on the native side at all, the scan can route through the exact same DataFusionParquetSourcepath asCometNativeScanExec, inheriting row-group stats pruning, page-index pruning (#5142), and filter pushdown (#4722) for free. The only genuinely Delta-specific native code is deletion-vector decoding: DV bitmaps are decoded into per-fileParquetAccessPlans, which DataFusion intersects with page-index pruning, so deleted rows are skipped in-scan and DV skips compose with page skips.Local benchmark (20M rows, selective predicate): 1.44x faster than stock Spark 9.4% of bytes read; DV tables at time parity with in-scan DV application.
What changes are included in this PR?
contrib/delta/new Maven module behind a-Pdeltaprofile: scan rule, decline gates, serde,CometDeltaNativeScanExec(split-mode partition serialization, DPP via derived scan helper), ServiceLoader registrations, differential test suites, Delta own-suite regression harness, benchmark script.CometScanRuleExtensionSPI + ServiceLoader hook at the top oftransformV1Scan;CometNativeScan.convert body extracted into reusablebuildNativeScanCommon`.deltacargo feature:DeltaScanproto + planner arm delegati the shared parquet scan builder;delta_dv.rsfor DV blob unframing (CRC verified), roaring decode (portable + native magic), and access-plan construction.input_file_name().How are these changes tested?
page_index_rows_pruned > 0,row_groups_pruned_statistics > 0), not benchmark notes.useMetadataRowIndexmodes.1 TB benchmark (real S3, this branch)
Independent run on TPC-DS-derived
store_salesat 1 TB (2.75 B rows), hilbert-clustered, on S3 (ap-southeast-1). The same physical parquet files are read through three paths, raw parquet (recursive glob), the Delta table, and an Iceberg table registered over the identical files viaadd_files, so the table-format scan path is isolated on byte-identical data. Spark 3.5.6 standalone, 252 executor cores (Graviton m7g), this branch at7fd81aa9built with-Pspark-3.5,delta. Four selective query families x 20 queries each;rows_scannedsummed from executed-plan scan metrics; warm = median of 3 runs. Control = same session with Comet disabled and the vectorized reader off (parquet-mr row reader, which prunes pages honestly).Fraction of rows decoded (Comet-native / control), and Comet bytes read per query:
Warm query times, Comet-native vs the row-reader control:
Takeaways:
ParquetSourcepath, inherit row-group + page-index pruning) holds at 1 TB on real S3.CometIcebergNativeScanExecreports post-filter-pushdown rows inoutput_rows, so its row fractions are not comparable to the other arms,bytes_scannedis the honest cross-arm metric (Iceberg reads ~2-3x the bytes of the parquet/Delta arms here).Deployment note for anyone staging this on a standalone cluster: core discovers the contrib via
ServiceLoaderon its own classloader, and the contrib links Delta types from that same loader — so when comet-spark ridesspark.{driver,executor}.extraClassPath(required forCometShuffleManager), the contrib jar and delta-spark/delta-storage must be real files on that same classpath;--packagesjars land in Spark's child loader where neither lookup can see them.Co-authored-by: Scott Schenkein schenksj@yahoo.com
Co-authored-by: Aditya Vaish adivaish@microsoft.com