[GLUTEN][VL] Use authoritative TahoeFileIndex table root for Delta deletion vectors - #12612
Conversation
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
This PR improves native Delta Lake deletion-vector (DV) materialization correctness and planning efficiency by sourcing the Delta table root from Delta’s authoritative TahoeFileIndex.path rather than inferring it from data-file paths.
Changes:
- Read the Delta table root from
relation.locationwhen it is aTahoeFileIndex, and thread that root into DV normalization. - Update
DeltaDeletionVectorScanInfo.normalize/extractacross Delta profiles (2.3 / 2.4 / 3.3 / 4.0) to take an explicittablePath: Path, and remove the heuristic table-path resolution (partition-walk +_delta_logexists probe + manual unescape). - Harden tests/benchmarks to validate the supplied table root is actually used, and add a partitioned-table DV integration assertion.
Reviewed changes
Copilot reviewed 9 out of 9 changed files in this pull request and generated no comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala | Uses TahoeFileIndex.path as the authoritative table root and passes it into DV normalization only for Tahoe-backed Delta scans. |
| gluten-delta/src-delta40/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala | Switches DV normalization/extraction to require an explicit tablePath and removes heuristic root resolution helpers. |
| gluten-delta/src-delta33/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala | Same explicit-root API and heuristic-removal as delta40 profile. |
| gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala | Updates API signature to accept tablePath (no-op behavior remains for pre-3.3). |
| gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala | Updates API signature to accept tablePath (no-op behavior remains for pre-3.3). |
| gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala | Adds an integration test ensuring partitioned-table DELETE produces UUID-based DVs and validates native scan presence where applicable. |
| backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala | Updates DV tests for new extract signature and adds a test proving the supplied table root is used for DV materialization. |
| backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala | Same as delta40 test updates/hardening. |
| backends-velox/src-delta33/test/scala/org/apache/spark/sql/execution/benchmark/DeltaPlanningBenchmark.scala | Updates benchmark to use the explicit table-root API for normalize(). |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
c5795ce to
10988b1
Compare
|
Run Gluten Clickhouse CI on x86 |
|
The failing PTAL when you get a chance, thanks! |
|
The Both failing jobs are environment/native flakes and just need a re-run. PTAL, thanks! |
10988b1 to
a9f74ff
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
49689ce to
1bebe9d
Compare
|
Run Gluten Clickhouse CI on x86 |
|
@jackylee-ch / @zhztheplayer would you mind to take a look at this one. It is the final step to complet the DV related fixes I have worked on recently. PTAL |
There was a problem hiding this comment.
LGTM @iemejia
Much better to take the root from Delta than guess it; this is what PreprocessTableWithDVs does anyway.
Also, worth saying in the description that this is a real bug fix, not cleanup — details inline on the baseline change. The perf win is also bigger than the benchmark suggests: the removed exists("_delta_log") ran once per FilePartition unconditionally, even on tables with no DVs.
Could you rebase please?
Side question: readRawDvBytes does skipBytes(offset) where Delta's own store does reader.seek(offset). Any reason?
cc: @zhztheplayer @zhouyuan Any other thoughts?
| org.apache.spark.sql.delta.CloneTableScalaDeletionVectorSuite#Cloning table with persistent DVs and absolute parquet paths | ||
| org.apache.spark.sql.delta.CloneTableScalaDeletionVectorSuite#Shallow clone round-trip with DVs |
There was a problem hiding this comment.
The 4 DV-clone tests in CloneTableScalaDeletionVectorSuite split exactly as your fix predicts. The 2 that failed write a "u" DV after a shallow clone DV relative to the clone root, data files absolute into the upstream table, so the old walk-up stops at the upstream _delta_log. The 2 that passed had makePathsAbsolute rewrite their DVs to absolute "p" first. Worth putting in the description as it's a real bug fix, not cleanup.
Not blocking: these only run on delta40, since delta_spark_ut.yml defaults to spark-4.1. The caller is shared so I don't think there's real risk, but a shallow-clone-then-DELETE read in DeltaSuite would cover 3.5 too if you think it's worth it.
There was a problem hiding this comment.
Thanks for the detailed breakdown — it matches exactly what I see. I've reframed the PR description around this being a correctness fix: a shallow-clone DELETE writes a clone-root-relative "u" DV while the data files point absolute into the upstream table, so the old walk-up stopped at the upstream _delta_log and resolved the wrong root; the 2 tests that passed only did so because makePathsAbsolute had already rewritten their DVs to absolute "p" paths. Using TahoeFileIndex.path fixes all four, which is why they can drop from the baseline.
On the not-blocking suggestion: I agree a shallow-clone-then-DELETE read in DeltaSuite would extend this to 3.5 (the caller is shared, so the risk is low, but it would pin the DeltaScanTransformer Tahoe arm directly). I'm happy to add it — replied on the DeltaScanTransformer thread too.
There was a problem hiding this comment.
Follow-up: added the shallow-clone-then-DELETE read as a DeltaSuite test in ffa69ed, so the 3.5 gap is now covered directly (not just via the delta40 clone shards). Details on the DeltaScanTransformer thread.
| case tahoe: TahoeFileIndex => | ||
| val tableRootPath = tahoe.path |
There was a problem hiding this comment.
Nothing pins this arm. DeltaDeletionVectorScanInfoSuite:140 is the only root-discriminating test and it calls normalize directly, bypassing DeltaScanTransformer. The integration test uses a table where both candidate roots coincide. A wrong root here is caught only by the delta40-only clone shards.
There was a problem hiding this comment.
Agreed that this arm is under-pinned: the only root-discriminating unit test calls normalize() directly (bypassing DeltaScanTransformer), and the integration test uses a table where both candidate roots coincide, so today only the delta40-only clone shards would catch a wrong root selected here. The shallow-clone-then-DELETE read you suggested on the known-failures thread is the right way to pin this on 3.5 too — happy to add it in a follow-up commit to this PR. Let me know if you'd prefer it in-PR before merge.
There was a problem hiding this comment.
Added in ffa69ed. New DeltaSuite test "deletion vector on shallow-cloned table" (min Spark 3.4): it DELETEs on a SHALLOW CLONE, whose data files point absolute into the source while the DV is written clone-root-relative ("u"). The old walk-up would resolve the source root and fail to find the DV here; sourcing the root from TahoeFileIndex.path makes it pass. It asserts the DeltaScanTransformer arm is used (on 3.5+) and checks results, so this arm is now pinned on 3.5. Compiles clean on Scala 2.12/Spark 3.5 and Scala 2.13/Spark 4.0; will confirm the runtime assertion once CI runs.
| // Partitioned so data files live under partition subdirs (region=.../...). The DV path is | ||
| // resolved from the table root (TahoeFileIndex.path) regardless of partition nesting; this | ||
| // guards the removal of the old partition-count-based table-path walk-up. |
There was a problem hiding this comment.
I think this comment claims more than the test delivers. getPartitionSchema is relation.partitionSchema, so partitionBy("region") gives partitionColumnCount = 1 the old loop goes <root>/region=a → <root>, finds _delta_log, and returns on the first probe. This test would be green on main too.
No issue with keeping it, but could you reword? The test that actually discriminates is DeltaDeletionVectorScanInfoSuite:140, since it points the PartitionedFile at an unrelated directory worth naming that one instead.
There was a problem hiding this comment.
Good catch — you're right. With partitionColumnCount = 1 the old walk-up hits _delta_log on the first probe, so this test would be green on main too. Reworded in 4e6b4ad to describe what it actually covers (end-to-end DV read over a partitioned table) and to point at DeltaDeletionVectorScanInfoSuite ("normalize materializes DV read options using the supplied table path") as the test that actually discriminates the root, since it points the PartitionedFile at an unrelated directory.
| DeltaDeletionVectorScanInfo.normalize( | ||
| partitionColumnCount = 0, | ||
| partitionFiles = partitionedFiles) | ||
| _ => DeltaDeletionVectorScanInfo.normalize(partitionedFiles, new Path(path)) |
There was a problem hiding this comment.
nit: scaladoc at L33-35 and L88 still talks about resolving/caching the table path. Would you mind updating?
There was a problem hiding this comment.
Done in 4e6b4ad — dropped the stale "resolving/caching table path" wording at L33-35 and L88. The normalize path no longer resolves or caches the root now that it's passed in explicitly.
…ization DeltaScanTransformer already knows the Delta table root via relation.location (a TahoeFileIndex, which PreparedDeltaFileIndex also extends). Thread that path into DeltaDeletionVectorScanInfo.normalize so it no longer re-derives the root from a file path via _delta_log existence probing -- one FileSystem.exists() (an HTTP HEAD on object stores) per partition. normalize gains an optional tablePath parameter; when absent it falls back to the previous resolveTablePath heuristic, so non-TahoeFileIndex locations (e.g. DeltaParquetFileFormat scans without a Tahoe index) and the public single-file extract entry point are unchanged. Add a test to DeltaDeletionVectorScanInfoSuite (delta33 and delta40) asserting the supplied-path and derived-path branches materialize an identical DV payload.
…blePath fallback Now that DeltaScanTransformer passes the table root from TahoeFileIndex.path, make it the single source of truth for DV materialization and remove the previous file-path-derivation fallback. - DeltaDeletionVectorScanInfo.normalize takes a required tablePath: Path (no Option, no fallback); partitionColumnCount is dropped since only the walk-up used it. - Delete resolveTablePath / isDeltaTablePath / unescapePathName and their per-partition _delta_log FileSystem.exists() probing. - DeltaScanTransformer materializes DVs only when relation.location is a TahoeFileIndex (which also covers PreparedDeltaFileIndex); other locations carry no Delta DV metadata and keep the generic split. - Update the public single-file extract(spark, file, tablePath), the delta23/24 stubs, the benchmark, and the suites accordingly. Net ~160 fewer lines. The Hadoop-conf caching and raw on-disk DV byte reading optimizations are retained.
…kipBytes Match Delta's own HadoopFileSystemDVStore.read, which opens an FSDataInputStream and seeks to the entry offset. seek is a positioned reposition (a ranged read on object stores) rather than a read-and-discard, and it avoids DataInputStream.skipBytes's best-effort semantics: an under-skip would silently misalign the read and fail the CRC check in readRangeFromStream. FSDataInputStream is a DataInputStream, so it is passed straight through.
- DeltaSuite partitioned-DV test: reword the comment to describe the end-to-end coverage it actually provides and point at DeltaDeletionVectorScanInfoSuite as the test that discriminates the table root; the old wording overclaimed (the removed walk-up would still find _delta_log on the first probe here). - DeltaPlanningBenchmark: drop stale scaladoc about resolving/caching the table path, which no longer happens now that the root is passed in.
1bebe9d to
4e6b4ad
Compare
|
Thanks for the review, @malinjawi! Rebased onto latest On your side question about I've also updated the PR description to frame this as the real bug fix it is (shallow-clone DV root mis-resolution), per your suggestion. Validated locally: |
|
Run Gluten Clickhouse CI on x86 |
Adds an integration test in DeltaSuite that DELETEs on a shallow clone.
The clone's data files point absolute into the source table while its DV
is written clone-root-relative ("u"), so the removed data-file walk-up
would resolve the source root and fail to find the DV. This directly
pins the DeltaScanTransformer TahoeFileIndex.path arm on Spark 3.5, which
the delta40-only CloneTableScalaDeletionVectorSuite shards did not cover.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 10 out of 10 changed files in this pull request and generated no new comments.
Suppressed comments (4)
backends-velox/src-delta40/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala:162
- Using
.geton the result offind(...)will throw a genericNoSuchElementExceptionif a DV isn’t produced, which makes failures harder to diagnose. Prefer failing with an explicit assertion/message (e.g.,getOrElse(fail(...))) so test output explains why DV production was missing.
val dataFile = DeltaLog
.forTable(spark, tablePath)
.update()
.allFiles
.collect()
.find(_.deletionVector != null)
.get
assert(dataFile.deletionVector.storageType == "u")
backends-velox/src-delta33/test/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfoSuite.scala:163
- Same issue as the delta40 suite:
.getcan fail with an unhelpfulNoSuchElementException. Switch to an explicit failure message so it’s clear whether DELETE didn’t generate a DV or the snapshot scan didn’t find it.
val dataFile = DeltaLog
.forTable(spark, tablePath)
.update()
.allFiles
.collect()
.find(_.deletionVector != null)
.get
assert(dataFile.deletionVector.storageType == "u")
gluten-delta/src-delta23/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala:31
tablePathis intentionally unused in the Delta 2.3 shim, but leaving it unused can trigger unused-parameter warnings depending on build flags/tools. Consider marking it as unused (e.g.,@scala.annotation.unused tablePath: Path) or renaming it to_tablePathto make the intent explicit.
def normalize(
partitionFiles: Seq[PartitionedFile],
tablePath: Path)
: Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
gluten-delta/src-delta24/main/scala/org/apache/gluten/delta/DeltaDeletionVectorScanInfo.scala:31
- Same as delta23 shim:
tablePathis unused by design here, but it’s worth annotating/renaming to avoid unused-parameter warnings and to document intent.
def normalize(
partitionFiles: Seq[PartitionedFile],
tablePath: Path)
: Option[(Seq[JMap[String, Object]], Seq[DeltaFileReadOptions])] = None
|
Run Gluten Clickhouse CI on x86 |
What changes are proposed in this pull request?
(Note:
apache/glutenhas nomasterbranch, so this PR targets the default branchmain.)This is a follow-up to #12390 that makes native Delta deletion-vector (DV) materialization both correct and cheaper by sourcing the Delta table root from the authoritative place instead of inferring it.
Background
Delta DV descriptors reference the on-disk bitmap using a table-root-relative UUID path. To read those bitmaps during planning,
DeltaDeletionVectorScanInfo.normalizeneeds the Delta table root. Previously the root was inferred from each data file path via a heuristicresolveTablePath:_delta_logdirectory,_delta_logat each level,%-encoded path characters along the way.This is fragile (it depends on file layout and partition depth, and can resolve the wrong root for non-trivial paths) and it performs at least one
FileSystem.exists("_delta_log")call pernormalizeinvocation.This is a correctness fix, not just cleanup
The walk-up resolves the wrong root whenever a data file does not live under the table that owns its DV. The clearest case is a shallow clone: a shallow-clone
DELETEwrites a clone-root-relative"u"(UUID) DV, while the cloned data files still point absolute into the upstream table. The old walk-up starts from the data file's directory (in the upstream table) and stops at the first_delta_logit finds — the upstream table's — so it resolves the wrong root and DV materialization fails.This is exactly the split observed in
CloneTableScalaDeletionVectorSuite: the 2 tests that failed on the old code write a"u"DV after a shallow clone; the 2 that passed only did so becausemakePathsAbsolutehad already rewritten their DVs to absolute"p"paths, so the resolved root did not matter. Using the authoritativeTahoeFileIndex.pathfixes all four, which is why this PR can drop them from the delta known-failures baseline.This change
DeltaScanTransformer.getSplitInfosFromPartitionsnow reads the table root directly fromrelation.locationwhen it is aTahoeFileIndex(which also coversPreparedDeltaFileIndexand the other Tahoe subclasses used for time travel, path-based reads, DML, and CDC).TahoeFileIndex.pathis the authoritative Delta table root — this is the same source Delta's ownPreprocessTableWithDVsuses.DeltaDeletionVectorScanInfo.normalize(partitionFiles, tablePath)andextract(...)across all Delta profiles (2.3 / 2.4 / 3.3 / 4.0).resolveTablePath,isDeltaTablePath(the_delta_logfilesystem probe), and the manualunescapePathNamehelper are removed.readRawDvBytesnow seeks to the DV entry withFSDataInputStream.seek(offset)instead of wrapping the stream and callingDataInputStream.skipBytes(offset), matching Delta's ownHadoopFileSystemDVStore.read.seekis a positioned reposition (a ranged read on object stores) rather than a read-and-discard, and it avoidsskipBytes's best-effort semantics where an under-skip would silently misalign the read and fail the CRC check inreadRangeFromStream.DeltaPlanningBenchmarkis updated to the explicit table-root API.Why this is an improvement
exists("_delta_log")probe ran once perFilePartitionunconditionally — even on tables with no DVs at all. On local storage this is a small, consistent win; on remote object stores (S3/ABFS/HDFS) anexists()is a network round-trip per split, so the saving is proportionally larger there — consistent with the remote-storage motivation of the parent tracking issue [VL] Optimize Delta Lake Deletion Vector processing on remote storage #12399.Test hardening
The DV unit tests were strengthened so they actually prove the supplied root is used: the synthetic
PartitionedFilenow points at an unrelated directory while the real table root is supplied separately, and the tests require a table-root-relative UUID (storageType == "u") DV — seeDeltaDeletionVectorScanInfoSuite("normalize materializes DV read options using the supplied table path"), which is the test that actually discriminates the root. The native partitioned-table integration test asserts that the DELETE really produced an on-disk"u"DV and then validates query results, giving end-to-end coverage of the DV read path over a partitioned table.How was this patch tested?
./dev/format-scala-code.sh(JDK 17), andscalastyleclean.-amreactor build), and Velox Delta 4.0 / Spark 4.0 / Scala 2.13, including backend test sources.DeltaDeletionVectorScanInfoSuiteon Delta 3.3 and Delta 4.0: 4/4 tests pass on each.CloneTableScalaDeletionVectorSuiteDV tests now pass and are dropped from the delta known-failures baseline.git diff --checkclean.DeltaPlanningBenchmark(normalize, 100 DV files x 10k rows, 200 timed iters, alternating fresh JVMs) comparing currentmainvsmain+ this change: median 61.40 µs/file vs 61.91 µs/file (~0.8% faster, ~51 µs saved per 100-file call) on local storage, with every run of the change faster than every baseline run. Larger gains are expected on remote filesystems where the removedexists()probe is a network round-trip.Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenCode github-copilot/gpt-5.6-sol
Generated-by: OpenCode github-copilot/claude-opus-4.8