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
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Follow-up review of 1d3557cf7eef. These five findings concern outstanding issues in the PR as a whole: comparison with the previous PR head, ef46cb8dd458, confirms that all five predate the latest force-push. They are not attributed to the latest update.
Validation: 43 targeted Delta Rust tests and the feature-disabled test passed. The inline comments distinguish source tracing, isolated native-reader/compiler/path probes, and Spark plan-shape checks. The Spark shape checks used a cached Spark 4.0.2-based build with Delta 4.0.0, not the target profile. Full current-head Spark/JNI validation could not complete because Maven dependency resolution stalled.
| // directly: its unapply arity differs across Spark versions and this module ships no | ||
| // version shim. | ||
| case e: InSubqueryExec => isAdaptivePlaceholder(e.plan) | ||
| case _: ScalarSubquery => true |
There was a problem hiding this comment.
[P2] Restore the execution partition count after scalar-subquery resolution
Could we preserve the planning-time guard while exposing the actual partition count during execution? For a partition filter such as p = (SELECT max(p) FROM thresholds), the expression remains a ScalarSubquery after its result is available, so this branch keeps outputPartitioning at zero.
A fused native parent reads that zero in buildNativeContext, and NativeExecContext's validation rejects nonempty scan partition data with All per-partition arrays must have length 0. This also affects contexts without broadcast inputs.
A separate Spark/Delta probe confirmed that the scalar expression remains in the partition filter after collect(), with an evaluable result and matching rows. The native-context failure is source-traced, not a full current-head JNI reproduction. Please add a regression that requires the native Delta scan beneath a native parent; the existing scalar-filter test permits fallback.
There was a problem hiding this comment.
Fixed. The guard now probes resolution instead of presence: eval on a resolved scalar subquery is a pure cached read (verified across the 3.5 through 4.2 jars), so once prepare resolves it the getter returns the real partition count, and only the unresolved case reports UnknownPartitioning(0). Your fused-parent shape reproduced the per-partition length failure locally and is now a regression that requires the native scan beneath a native aggregate.
| val internalFields = requiredSchema.fields.toSeq | ||
| .filter(f => internalColumnNames.contains(f.name)) | ||
| .map(f => StructField(s"$deltaConstFieldPrefix${f.name}", f.dataType, f.nullable)) |
There was a problem hiding this comment.
[P2] Make DV synthetic field names unique against the physical schemas
Could we choose collision-free synthetic names, or decline conflicting scans? A legal user column named _comet_delta___delta_internal_is_row_deleted collides with the partition constant generated here. DataFusion substitutes partition constants by column name, so the real data projection is replaced by the bookkeeping value 0.
A separate Spark/Delta probe accepted that TINYINT column with value 7, created a real DV through DELETE, and showed the expected internal-column suffix in the scan schema. A standalone probe against the locked public DataFusion 54.1.0 reader then read real Parquet values (1,7),(2,7) as (1,0),(2,0) with this colliding constant; a distinct-name control preserved 7. These are separate plan-shape and reader probes, not a full current-head Spark/JNI run.
Please cover the collision in the native Delta differential suite, including the physical data and partition namespaces when allocating the synthetic fields.
There was a problem hiding this comment.
Fixed. The internal bookkeeping constants are now allocated collision-free against the physical data and partition namespaces, and while in there we gave the file-metadata constants the same protection in the shared core builder, so plain reads are covered too. Your colliding-column shape reproduced the 7 to 0 corruption locally and is covered in the differential suite end to end plus at the unit level.
| val buckets = uris.flatMap(s3Bucket).distinct | ||
| if (buckets.isEmpty) { | ||
| return None |
There was a problem hiding this comment.
[P2] Decline Hadoop-only GCS authentication until it can be forwarded
Could we also gate GCS authentication compatibility? A scan with local/S3 data files and an absolute gs://private-bucket/... DV can pass the scheme and authority gates while relying only on Hadoop's fs.gs.auth.service.account.json.keyfile for GCS access. This function and the provider-class gate consider only S3 authorities.
Although conversion forwards fs.gs.*, the native resolver's GCS/default branch calls parse_url(&url) without those options. With no alternative ADC or metadata-service credentials, Spark can read the sidecar but the claimed native scan fails authentication instead of falling back.
The core GCS options omission predates this PR; the finding here is the Delta admission/fallback gap exposing that limitation. This is a source-traced case; no cloud credentials or GCS requests were used. A conservative fallback would be sufficient until native GCS option translation is supported.
There was a problem hiding this comment.
Added the conservative gate: any scan whose data files or DVs touch gs:// declines when a fs.gs.auth.* key is set, since the native resolver forwards no GCS options. ADC-only configs still claim because both engines resolve credentials the same way there. Tests cover your mixed local-data plus gs DV sidecar shape and the scheme scoping.
There was a problem hiding this comment.
Extended the gate to the connector's legacy prefix: google.cloud.auth.* keys now decline the same way fs.gs.auth.* does, still scoped to gs paths and naming keys only.
There was a problem hiding this comment.
Stopped enumerating prefixes: the gate now declines any fs.gs.* or google.cloud.* key involving auth, which covers the deprecated service.account.auth forms you found and whatever other spellings the connector's prefix machinery accepts. ADC-only still claims. Happy to hear if you see a shape this misses.
| builder.setInlineData( | ||
| org.apache.comet.shaded.protobuf.ByteString.copyFrom(desc.inlineData)) |
There was a problem hiding this comment.
[P2] Support the unshaded classpath during reactor compilation
Could we make the contrib's compile/shading arrangement work before core's package phase as well? A clean root ./mvnw -Pspark-4.0,delta compile or test sees unshaded spark/target/classes. The org.apache.comet.shaded.protobuf namespace is created only by core's package-phase shade execution, so this hard-coded reference cannot resolve in that reactor path. The separate core-install/contrib-test CI sequence masks it.
An isolated Scala compilation against Java types generated from the current protocol reproduced object shaded is not a member of package org.apache.comet; the unshaded ByteString control compiled. The full reactor attempt was blocked earlier in dependency resolution, so this is compiler/classpath evidence rather than a completed reactor reproduction.
Simply changing the import would invert the problem for packaged-core consumers; the module needs a consistent arrangement for both classpaths.
There was a problem hiding this comment.
Fixed for both classpaths. The contrib no longer names the protobuf package at all: a small helper in comet-spark takes the builder and raw bytes and sets the field internally. You were right that just changing the import would invert the problem, and it's worse than that: a helper returning ByteString also breaks the packaged side, because the shade plugin rewrites bytecode descriptors but not Scala pickled signatures. Keeping the type inside the helper's body sidesteps both. Your clean root compile repro now passes, and the packaged path plus full suite stay green.
1d3557c to
b033153
Compare
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed b033153da230 against 1d3557cf7eef, including the full PR diff. Two existing P2 discussions remain:
- [P2] S3 credential references (existing item 3, credential thread): a real local MinIO probe reads 10 rows with stock Spark and a native plain-credential control. The asserted native scan with
${review.access}/${review.secret}still forwards those literals and fails with403 InvalidAccessKeyId. - [P2] GCS Hadoop-only authentication (existing thread): the
fs.gs.auth.*case now declines, but the runtime connectorhadoop3-2.2.26also acceptsgoogle.cloud.auth.service.account.json.keyfile. Its actual credential factory loads a generated local fixture without creating anyfs.gs.auth.*entries; current admission returnsNoneand native extraction omits the key. The modern-prefix control declines. This was a local credential/admission probe with zero network requests, not a live GCS scan.
The mapping, scalar-count, synthetic-name, and protobuf-classpath fixes checked out. On chao-reviews-1, 143 Delta contrib tests (including both MinIO tests), 43 native Delta tests, and the feature-off test passed; root test and packaged-build paths passed. No additional verified P1/P2 finding; no duplicate inline comments.
| val multiChildLeak = plan.exists { | ||
| case u if isPositionalUnion(u) => false | ||
| case n if n.children.size >= 2 => |
There was a problem hiding this comment.
[P1] Treat write sinks as consumers of the reader's row index
Could we make the unused-row-index proof account for unary write sinks? This is a reader eligibility issue: Spark's existing writer simply persists the values returned by the reader.
For example, with the Delta contrib/native scan enabled and the native Parquet writer disabled, this workload reads the source row position into an ordinary output column:
val root = java.nio.file.Files.createTempDirectory("comet-row-index")
val src = root.resolve("src").toString
val dst = root.resolve("dst").toString
spark.range(32).coalesce(1).write.format("delta")
.option("delta.enableDeletionVectors", "true").save(src)
spark.sql(s"DELETE FROM delta.`$src` WHERE id IN (1, 7, 13)").collect()
spark.read.format("delta").load(src)
.selectExpr("id", "_metadata.row_index AS ri")
.write.parquet(dst)
spark.read.parquet(dst).where("id = 31").show() // expected: (31, 31)
spark.read.parquet(dst).selectExpr("sum(ri)").show() // expected: 475The real physical write plan is DataWritingCommandExec -> WriteFilesExec -> Project -> ... -> Delta scan. Both write nodes have one child, empty output, and no expression reference to ri; they consume their child's rows positionally. The two Projects propagate the row-index dependency, but neither nonProjectConsumer, the root-output escapes check, nor this multi-child-only guard sees the write consuming it. The exact current rowIndexUnusedAbove returns true for the Parquet write, while the equivalent SELECT correctly returns false.
That permits the DV reader to supply the synthetic Long zero. On the same actual Delta fixture, the current native reader retained all 29 correct surviving IDs but returned ri=0 for every row: 28 incorrect row-index values, sum 0, and (31,0) rather than (31,31). The unchanged writer would persist those incorrect input values.
Validation: the exact liveness method was exercised on real cached Spark 4.0.2-based/Delta 4.0.0 write plans with AQE off/on, including an independent default-codegen check. Separately, the production native DV attachment and Parquet reader were exercised on the actual file and DV, with the source-verified scan layout. The remaining admission path was source-traced; a full current-profile Spark/Comet JNI write was not executed.
Please treat write sinks as consumers of tainted input attributes, or conservatively decline this shape, and add a regression checking both the eligibility decision and saved ri values. The liveness logic already exists at 1d3557cf; this is not introduced by the latest b033153d update.
There was a problem hiding this comment.
Fixed. The liveness walk now treats any node that drops the column from its own output as a consumer unless it's a Project, whose alias tracking already decides that case, so one-child write sinks decline. Your workload reproduced the ri=0 corruption here and is now the regression: the write declines and the readback shows (31,31) with sum 475. One note on coverage: an end-to-end DSv2 write test turned out to be blocked by delta-spark itself (its DV planning requires a pinned TahoeLogFileIndex under DSv2 writes), so DSv2 sinks are covered by the structural rule and a test pins that upstream limitation so a future delta upgrade surfaces it.
| requiredSchema = toPhysical(scanExec, scanExec.requiredSchema), | ||
| dataSchema = toPhysical(scanExec, relation.dataSchema), |
There was a problem hiding this comment.
[P1] Guard Unicode case-insensitive Delta reads before using the shared reader
With column mapping disabled and spark.sql.caseSensitive=false, this call sends the logical names unchanged to the shared native Parquet reader. Its ASCII-only matching does not preserve Spark's handling of non-ASCII case variants in physical column names.
A normal converted Delta table can expose this:
import org.apache.spark.sql.functions.{col, lit}
spark.conf.set("spark.sql.caseSensitive", false)
val root = java.nio.file.Files.createTempDirectory("comet-unicode")
val path = root.resolve("data").toString
val table = "comet_unicode_" + java.util.UUID.randomUUID().toString.replace("-", "")
spark.range(1, 2).select(col("id"), lit(71).as("É"))
.coalesce(1).write.parquet(path)
spark.range(2, 3).select(col("id"), lit(72).as("é"))
.coalesce(1).write.mode("append").parquet(path)
spark.sql(s"CREATE TABLE $table (id BIGINT, `É` INT) USING PARQUET LOCATION '$path'")
spark.sql(s"CONVERT TO DELTA $table NO STATISTICS")
spark.read.format("delta").load(path).selectExpr("id", "`É`").show()Stock Spark/Delta returns [(1,71),(2,72)]. The current native reader on these exact files returns [(1,71),(2,NULL)]. Adding É IS NOT NULL or É > 70 loses the second row. A covering Spark filter cannot recover the stored value after the scan has replaced it with NULL.
This is an ordinary CONVERT transaction with protocol (1,2), mapping none, no defaults or DVs, and two local files. The declared required/data schemas are both [id, É], projection is [0,1], and field-ID matching is off. The case difference appears only in the second file's footer. Current admission has no Unicode-name exclusion, and the new Delta handler installs the same SparkPhysicalExprAdapterFactory exercised by the probe.
Validation: the standard conversion command and reference reads ran on cached Spark 4.0.2-based/Delta 4.0.0. Current native-reader execution was compared against those results: nine mismatches across É/é, Σ/σ, and Б/б for plain reads and the two predicates; all 15 ASCII/case-sensitive controls matched. The complete current claim path was source-traced, not executed through a full current-profile Spark/JNI pipeline.
Could we fix the shared name matching or conservatively fall back for affected Delta schemas, with an asserted native/fallback regression for this conversion history? The matcher predates the PR, but this PR newly routes these previously Spark-read Delta tables through it. The same exposure exists at 1d3557cf, so it is not introduced by the latest update.
There was a problem hiding this comment.
Added the conservative fallback: under caseSensitive=false, any data-schema name the reader would match against footers (logical names, or physical names under column mapping) containing a non-ASCII character declines. Partition names are exempt since their constants bind by exact name and never touch footer matching. Your CONVERT repro reproduced the (2,NULL) here and now declines with answers matching Spark; ASCII and caseSensitive=true controls still claim natively.
There was a problem hiding this comment.
Took your suggestion to fix the shared matcher instead of gating. Native case-insensitive matching now mirrors Java's equalsIgnoreCase (per char, simple mappings, including the U+0130 special case where simple and full mappings diverge), applied across the read path. The schema gate is gone since it couldn't see footer-side names anyway. Your Kelvin repro reproduced the (2,NULL) here and now reads correctly natively, as does the earlier accented pair.
There was a problem hiding this comment.
You were right about which API this is. Spark keys footers by toLowerCase(Locale.ROOT), not equalsIgnoreCase, so that's what the matcher does now, and your three shapes flip accordingly. To stop chasing these one at a time I diffed the entire codepoint space against JDK 17: 95 codepoints differ purely from Unicode version skew between the JDK and the Rust tables. Those are pinned with a checked-in fixture and full-sweep tests on both sides, plus contextual sigma cases, which turned out to matter since the newer Rust tables treat some of those codepoints as cased and that flips the final sigma decision. If your harness still finds a disagreement I'd genuinely like to see it.
There was a problem hiding this comment.
[P1] Current-head reader reproduction of the Unicode mismatch
Follow-up on 4ca3207af385: the current native library reads a real Parquet value 42 as follows with case sensitivity disabled. The reference column is Spark 4.1.3's actual ParquetReadSupport.clipParquetSchema on JDK21:
| Physical / requested name | Spark footer lookup | Native value |
|---|---|---|
A1Σ / a1σ |
Missing field | 42 |
A1Σ / a1ς |
Match | NULL |
| U+A7C0 / U+A7C1 | Match | NULL |
The inverse sigma case therefore reads a field Spark considers missing, in addition to the previously reported lost values. A covering IS NOT NULL filter retains or drops the wrong row. Latin and case-sensitive controls pass.
Could the matcher account for the executing JVM's Unicode version and contextual sigma behavior? This is native Parquet execution plus Spark footer-method execution, not a full Spark/Delta/JNI query. The unchanged JVM parity suite also fails its full sweep on JDK21 (2 passed, 1 failed; only the test repository locator was stubbed).
There was a problem hiding this comment.
Rebuilt this so it can't be JDK-dependent anymore: the driver now ships the running JVM's own case tables (about 15 KB, only when case insensitive) and native runs Java's algorithm over them, so whatever JDK executes is the JDK that's matched. Your A1Σ pair reproduced here in both wrong directions and both flip correctly now. Interesting find along the way: Java's final sigma is word-boundary based, not the Unicode rule. The parity suite is self-validating now and passes on 17, 21, and 25 locally, including the full-sweep test that failed for you on 21.
There was a problem hiding this comment.
The dash plus ZWJ shape came from hand-rolled bridge rules approximating the word-break automaton, so I stopped approximating: the classifier now probes the running JDK's own BreakIterator per codepoint and both sides run the full word-boundary rule set over that data. Your shape matches Java now, and the sweeps grew to triples of exotic characters plus a million-string fuzz, zero mismatches on 17, 21, and 25.
b033153 to
0bc592b
Compare
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked 0bc592b529f2 (base 6fa652b43ec3) with fresh Spark 4.0.4 / Delta 4.0.1 artifacts.
The write-sink P1 now passes: planned writes, forced AQE, and a later-UNION-branch control fall back and persist the correct row indexes.
Three existing discussions still have uncovered cases:
- [P1] Unicode names: with
spark.sql.caseSensitive=false, an ordinaryCONVERT ... NO STATISTICStable (mappingnone, no DVs) declared with ASCIIKorkcan contain a file whose physical column name isK(U+212A). The guard admits it; native reads return(2,NULL)instead of(2,72), andIS NOT NULL/> 70lose that row. Stock Spark and the ASCII/case-sensitive controls agree; the originalÉ/écase now falls back. - [P2] S3 references: credential-value references now fall back, but
fs.s3a.aws.credentials.provider=${review.provider}—expanding toSimpleAWSCredentialsProvider—and its short bucket override are admitted using the expanded value and forwarded raw. Stock MinIO reads and literal-provider native controls return 10 rows; native execution with the reference fails withUnsupported credential provider: ${review.provider}. - [P2] GCS authentication: connector 2.2.26 also accepts
fs.gs.service.account.auth.keyfile/.emailand the correspondinggoogle.cloud.service.account.auth.*keys. Both deprecated forms load a generated PKCS12 credential but bypass the new prefixes, including with normal Hadoop defaults. Modern-key controls decline. The native GCS resolver still ignores these Hadoop options, so access relying on them is not preserved when no usable ADC is available.
Unicode and S3 were exercised through the current Spark/JNI path with asserted native/fallback controls and verified JAR/library origins. GCS validation was the actual connector credential factory and admission method plus source tracing; no GCS request or full cloud scan was performed.
0bc592b to
21c08f9
Compare
21c08f9 to
7a0f2a5
Compare
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Rechecked 7a0f2a59a575625094a9b07729b84c9c30bf0582 after the rebase.
- [P1] Unicode footer matching: Freshly compiled current matcher controls still disagree with actual Spark 4.0.4 footer clipping, including
I/ı, contextual sigma, and dotted-I expansion. Accent, Kelvin, ASCII and case-sensitive controls now agree. - [P2] S3 references: Bucket-dependent substitutions still pass admission when the actual Hadoop provider resolves bucket credentials but native-forwarded options retain global credentials. Ordinary substitution controls pass.
The reported write-sink/UNION and deprecated GCS-alias mechanisms are addressed in the current source and pass fresh real-plan and connector-credential controls.
These are component checks with verified reused inputs, not a complete current build or JNI/query/cloud run. Current persisted-write/AQE behavior and end-to-end read/authentication outcomes remain unverified; the disk reserve blocked rebuilding the complete runtime.
The core insight is a good one, and it is the thing that makes this PR much more attractive than the two it supersedes: delta-spark has already done log replay, snapshot resolution, and partition pruning by the time Five things.
-default = ["hdfs-opendal"]
+default = ["hdfs-opendal", "delta"]The comment argues this is inert without the contrib jar and the config, which I believe. But it still means every default build of What is the argument for defaulting
Deletion-vector bitmaps are an on-disk format with a CRC, produced by delta-spark and by other Delta writers. This is the highest-risk file in the PR: a parsing bug is either a wrong answer (rows that should be deleted come back) or a panic on malformed input. Two questions. Is the decoder validated against DV files produced by delta-spark itself, or only against ones this code round-trips? And what happens on a truncated, CRC-mismatched, or otherwise malformed DV file: a clean error, or a panic or out-of-bounds read? Delta tables are read from object storage where partial reads happen, so the malformed path matters. Reflection surface
CI cost A new Benchmark numbers The description starts a "Local benchmark (20M rows, ..." sentence that appears to be cut off in what I read. Whatever the numbers are, they are the justification for the whole PR and should be prominent: Comet versus Spark on a Delta table with and without deletion vectors, and ideally with and without page-index pruning so the DV-composes-with-page-skips claim is demonstrated. |
|
@andygrove, a few factual clarifications against
These are source/discussion clarifications; I did not rerun the tests or benchmarks for this follow-up. |
7a0f2a5 to
ac86ef6
Compare
|
@andygrove The default feature choice came out of earlier review discussion, but you're right it belonged in the description. It's there now along with the measured cost, about 82 KB on a roughly 380 MB debug dylib, and it's unreachable without the contrib jar plus the config. The DV decoder concern was fair, so there's now a malformed-input matrix in the native tests: truncation at every byte offset, CRC and magic corruption, length and cardinality lies, and bit-flip sweeps. Everything errors cleanly, no panics. On version skew, Chao was right that linkage errors used to propagate. The claim hook now contains LinkageError as a logged decline, so a mismatched contrib jar falls back to Spark instead of breaking planning. There's a test that fakes a NoSuchMethodError contrib to pin that. Also cleaned up the description itself, a few garbled words and stale pre-convergence module names, which is probably what made the benchmark section look cut off. CI numbers are in there too: the contrib jobs are path gated and the suite runs 4 to 5 minutes per Spark profile. |
ac86ef6 to
4ca3207
Compare
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
[P1] Re-reviewed 4ca3207a; the existing Unicode field-matching issue still has two source-confirmed cases. The new matcher pins JDK17 mappings, but the supported JDK21 runtime maps U+A7C0 to U+A7C1 while native preserves U+A7C0. There is no JVM-version admission guard. Even on JDK17, physical A1Σ / requested a1ς diverge: Java's word-boundary-based lowercasing yields a1ς, whereas native yields a1σ.
For nullable, case-insensitive mapping=none columns, these misses reach missing-column NULL substitution and can lose rows under filters. These consequences were traced through current source and dependencies, not reproduced in a fresh Spark/JNI query. The previous S3 bucket-reference case is addressed in source; no query/cloud tests were rerun, and the current workflows remain action_required.
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Reviewed 4ca3207af385 with five independent specialist scopes. Two new findings are inline; additional reproductions are in the existing Unicode and S3 credential discussions.
Validation: the default native build and 165 focused Rust tests (56 Delta, 108 Parquet, one feature-off; tests with HDFS disabled) passed. The unchanged JVM parity suite on JDK21 passed two tests and failed one in component validation. Full Spark/Delta/JNI validation was blocked by Maven dependency access; no live cloud validation was performed.
| Seq.empty, | ||
| None, | ||
| Seq.empty, | ||
| perPartitionFilePaths = perPartitionFilePaths) |
There was a problem hiding this comment.
[P2] Report Delta scan input to Spark task metrics
Could we register reportScanInputMetrics for this RDD and include contrib scans in the fused reporting decision? A fresh, uncached Delta-only read returns this bare CometExecRDD, unlike the existing native scans, so no callback copies SQL scan metrics into task input counters. Fused/native-shuffle execution also skips it because hasScanInput recognizes only CometNativeScanExec.
Task/stage/event-log input bytes and records consequently remain zero even though SQL scan metrics are populated; results are unaffected. Please cover standalone and fused reads with a task-listener regression. Independently source-traced and challenged; a full Spark/Delta/JNI reproduction was blocked by Maven dependency access.
There was a problem hiding this comment.
Wired up. The RDD registers the input-metrics callback now and the fused path recognizes contrib scans through the shared trait rather than the concrete class, so future contribs inherit it. Task-listener regressions cover the standalone and fused reads, both were zero before the change.
| if (CometParquetUtils.encryptionEnabled(hadoopConf)) { | ||
| return Some("Native Delta scan does not support encrypted parquet") |
There was a problem hiding this comment.
[P2] Decline SSE-C scans until the customer key reaches S3
With the Delta contrib enabled, an existing S3 Delta table readable through Hadoop using fs.s3a.encryption.algorithm=SSE-C plus fs.s3a.encryption.key passes this check: it only detects Parquet modular encryption. The later S3 gates also accept it. NativeConfig forwards these settings, but native extract_s3_config_options drops them, leaving GET and HEAD without their required customer encryption headers. The new Delta admission therefore exposes those SSE-C objects to native read failures; the base retained Spark's reader. Could we conservatively decline SSE-C until the native client receives the key?
Fresh isolated checks cover canonical/deprecated global and bucket settings: unchanged storage helpers accept them, Hadoop 3.3.4 request factories attach the key, and the unchanged native translation with locked object_store 0.13.2 produces requests missing all three headers. An explicit SSE-C builder control supplies them. Full admission and the read-failure consequence were source/API-traced; no live S3 or full Spark/Delta/JNI read was run.
There was a problem hiding this comment.
Added the decline: any s3 scan whose effective encryption algorithm resolves to SSE-C falls back, covering the canonical, deprecated, and per-bucket spellings (Hadoop registers the deprecated ones as process-wide aliases, which one test learned the hard way). SSE-S3 and KMS stay claimable since decryption is server-side on read. The customer key never appears in the reason.
There was a problem hiding this comment.
You were right, encryption settings resolve through lookupPassword and the long form wins, my earlier read of that was wrong. The gate now resolves every key that way (for plain options that direction can only over-decline, which is the safe side) and the algorithm check flipped to an allowlist: unset, AES256, SSE-KMS, DSSE-KMS pass, everything else falls back. Your long-form shape declines now.
4ca3207 to
fb4858a
Compare
sunchao
left a comment
sunchao
left a comment
There was a problem hiding this comment.
Re-reviewed fb4858a0. Two P2 residuals remain in the existing discussions:
- [P2] Unicode name matching: the earlier concrete mapping controls are addressed, but a sigma followed by
-, U+200D, and another letter still differs from Java's word-boundary behavior. The resulting wrong column match or NULL is a residual of this family. - [P2] SSE-C fallback: literal global/short-bucket cases are addressed, but the long bucket algorithm alias still bypasses the gate on the declared Spark 3.5.9/Hadoop 3.3.4 source route. Hadoop selects SSE-C while the native request omits the customer-key headers.
The two separate findings are attached inline. These conclusions are from source review; no runtime reproduction or CI pass is claimed.
| declined | ||
| } else { | ||
| effectiveEncryptionAlgorithm(hadoopConf, bucket).collect { | ||
| case (key, value) if value.equalsIgnoreCase(SseCustomerKeyAlgorithm) => |
There was a problem hiding this comment.
[P2] Decline Hadoop CSE-KMS before claiming the native scan
This gate admits fs.s3a.encryption.algorithm=CSE-KMS, but the native S3 client has no client-side decryption layer. On the declared Spark 3.5.9/Hadoop 3.3.4 route, Hadoop's default S3 client factory selects an encryption client with KMS materials and range-read decryption. An existing, uniformly CSE-KMS-encrypted Delta table with valid credentials/key permissions, no S3Guard, and ordinary primitive NoMapping/no-DV data is therefore readable through Spark; enabling this contribution sends ciphertext to native Parquet and fails the scan. SSE-KMS is transparent server-side encryption and is not this case. Please retain Spark fallback for CSE-KMS until native decryption is supported. The clean requested BASE has no bundled scan provider and leaves this Delta scan with Spark; this finding predates the latest increment.
There was a problem hiding this comment.
Covered by the allowlist change: CSE-KMS, CSE-CUSTOM, and any unknown algorithm decline, server-side ones stay claimable. Two adjacent things got picked up in the same pass: the algorithm can live in a JCEKS keystore so it resolves through the provider machinery now, and there's a proxy gate since native has no proxy support at all.
| && unmatched_id_logical_names | ||
| .iter() | ||
| .any(|name| name.eq_ignore_ascii_case(field.name())) | ||
| .any(|name| names_equal_ignore_case_java(name, field.name(), case_tables)) |
There was a problem hiding this comment.
[P2] Respect case sensitivity when shielding unmatched field IDs
For an ordinary case-sensitive V1 Parquet scan with field-ID reading enabled, request nullable BIGINT fields Κ (U+039A, ID 1) and κ (U+03BA, no ID), and read a file containing only κ with ID 2 and value 7. Spark null-fills the missing ID 1 field but reads the ID-less field by its exact name. This unconditional case-folded check instead matches physical κ to unmatched logical Κ and replaces the real column with a fake name; the final nullable projection yields (NULL, NULL) instead of (NULL, 7). File ID 2 satisfies the file-ID presence guard, and the names are distinct with case sensitivity enabled. Requested BASE's ASCII comparison does not equate this Greek pair. Please avoid hiding a physical field needed by a legitimate exact-name match. This regression exists at the previous reviewed HEAD too, so it is new to the discussion, not newly introduced by the latest increment.
There was a problem hiding this comment.
Fixed, the shield compares exactly under case sensitivity and only fires for fields the name pass didn't claim, matching clipParquetGroupFields. Auditing the neighboring comparisons turned up four more divergences from Spark's semantics (a stray ID-less column could win an exact-name lookup for an ID-matched field, the fake name could collide with a real column, and two duplicate-handling mismatches), all fixed with tests against real files.
One thing from that which was found: validateIcebergFileScanTasks keys on a hardcoded scheme list including gcs, oss, wasb, and wasbs, which the native url parser doesn't recognize, so those Iceberg tables get claimed and then fail at execution. Left untouched here since it's the Iceberg path, but it's the same class of drift this PR kept hitting.
Adds an optional contrib/delta-spark module that claims delta-spark DSv1 scans through CometScanContrib and runs them on Comet's native parquet reader. Deletion vectors are decoded 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, 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>
fb4858a to
a293d19
Compare
6 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-spark/new Maven module behind the-Pdeltaprofile, shipped as its own optional jar: claim/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.CometScanContribServiceLoader hook;CometNativeScan.convertbody extracted into the reusablebuildNativeScanCommon. The hook contains contrib failures (includingLinkageErrorfrom version skew) as a logged decline, so a mismatched contrib jar falls back to Spark instead of failing planning.deltacargo feature:DeltaSparkScanmessages dispatched through the genericContribScanenvelope into the shared parquet scan builder;delta_dv.rsfor DV blob unframing (CRC verified), roaring decode (portable + native magic), and access-plan construction. Malformed DV input (truncation at any byte, CRC/magic corruption, length or cardinality lies, arbitrary bit flips) is covered by a no-panic test matrix; every case errors cleanly.deltais in the default cargo feature set so the optional contrib jar works against stock Comet binaries without a custom native build (discussed and agreed in review). The cost is ~82 KB on a ~380 MB debug dylib and the code is unreachable unless the contrib jar is on the classpath andspark.comet.scan.delta.enabled=true; default builds without the jar carry no runtime behavior change.input_file_name().How are these changes tested?
page_index_rows_pruned > 0,row_groups_pruned_statistics > 0), not benchmark notes.dev/run-delta-regression.sh): DeletionVectorsSuite 29/29, TimeTravel/ColumnMapping/DeleteSQL/UpdateSQL 197/197, MergeIntoSQLSuite 664/665 (the one failure is a scan-telemetry count assertion, a plan-shape artifact; data assertions pass).dev/ci/compute-changes.py(they run only when Delta, shared Spark/common, or native code changes); the contrib suite takes about 4-5 minutes per Spark profile on a laptop.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