[VL][Delta] Guard DV DML row-index scans - #12215
Conversation
|
Run Gluten Clickhouse CI on x86 |
1 similar comment
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
1 similar comment
|
Run Gluten Clickhouse CI on x86 |
f124a30 to
6d70f4b
Compare
|
Run Gluten Clickhouse CI on x86 |
6d70f4b to
d83aac4
Compare
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
This PR adds a safety guard for Delta Lake DML (DELETE/UPDATE/MERGE) on tables with deletion vectors by detecting the special DV “row-index” target scan shape and preventing that scan (and its immediate filter/project) from being offloaded to Gluten/Velox unless explicitly enabled via native flags. This is intended to prevent incorrect DV DML results until native row-index scan support is complete.
Changes:
- Tag Delta DV DML row-index target scans during planning, and use those tags to drive offload/fallback decisions.
- Force Spark fallback for the tagged DV DML row-index scans unless both native-write and native-DML-row-index flags are enabled; keep the parent filter/project on Spark to avoid extra row↔columnar transitions.
- Add Delta 3.3 (Spark 3.5) and Delta 4.0 (Spark 4.0) unit coverage to assert fallback plan shape and repeated-DELETE DV correctness.
Reviewed changes
Copilot reviewed 6 out of 6 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-delta/src/main/scala/org/apache/gluten/extension/OffloadDeltaScan.scala | Adds DV DML row-index scan fallback gating and propagates the DML-scan tag to the transformer when offloaded. |
| gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaPostTransformRules.scala | Adds a post-transform rule to keep the filter/project above a DV DML fallback scan on Spark; expands DV row-index column name handling. |
| gluten-delta/src/main/scala/org/apache/gluten/extension/DeltaDeletionVectorDmlUtils.scala | New utility to detect/tag DV DML row-index scan shapes and provide shared Delta-scan detection helpers. |
| backends-velox/src-delta40/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala | Adds Spark 4 / Delta 4.0 tests asserting DV DML row-index scan fallback and repeated-DELETE DV correctness. |
| backends-velox/src-delta33/test/scala/org/apache/spark/sql/delta/DeltaDeletionVectorHandoffSuite.scala | Adds Spark 3.5 / Delta 3.3 tests asserting DV DML row-index scan fallback and repeated-DELETE DV correctness. |
| backends-velox/src-delta/main/scala/org/apache/gluten/component/VeloxDeltaComponent.scala | Injects the pre-transform tagging rule for DV DML row-index scans in the Velox Delta component. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| private def containsDmlRowIndexFallbackScan(plan: SparkPlan): Boolean = { | ||
| plan.exists { | ||
| case scan: FileSourceScanExec => | ||
| FallbackTags | ||
| .getOption(scan) | ||
| .exists(_.reason().contains("fallback Delta DV DML row-index scan")) | ||
| case _ => false | ||
| } | ||
| } |
| val scanColumnNames = (scan.output.map(_.name) ++ scan.requiredSchema.fieldNames).toSet | ||
| scanColumnNames.exists(columnNames.contains) || columnNames.exists(scan.treeString.contains) |
| private def referencesRowIndexColumn(expr: Expression): Boolean = { | ||
| val expressionText = expr.toString() | ||
| expr.references.exists(attr => deletionVectorRowIndexColumnNames.contains(attr.name)) || | ||
| deletionVectorRowIndexColumnNames.exists(expressionText.contains) | ||
| } |
| private def referencesFilePathColumn(expr: Expression): Boolean = { | ||
| val expressionText = expr.toString() | ||
| expr.references.exists(attr => filePathColumnNames.contains(attr.name)) || | ||
| filePathColumnNames.exists(expressionText.contains) | ||
| } |
d83aac4 to
fcc657c
Compare
|
Run Gluten Clickhouse CI on x86 |
Native DELETE/UPDATE/MERGE deletion-vector support deliberately keeps the target row-index scan on Spark until native row-index execution is proven for the required Delta table shapes. Detect those DML row-index scan shapes and keep the small scan subtree (and its parent filter/project) off the native path, so the offload path stays correctness-first while the write side is built out. - add DeltaDeletionVectorDmlUtils to detect and tag Delta DV DML row-index scans (row-index / file-path / bitmap-aggregator references) - gate OffloadDeltaScan to fall back on DML row-index scans unless native Delta write and native DML row-index scan are both enabled - add keepDmlRowIndexFallbackSubtreeOnSpark post-transform rule to keep the filter/project above a fallen-back DML scan on Spark, avoiding row<->columnar transitions right before Delta's JVM bitmap path - cover the fallback shape with DeltaDeletionVectorHandoffSuite cases for Spark 3.5 / Delta 3.3 and Spark 4.0 / Delta 4.0
fcc657c to
1bc128b
Compare
|
Run Gluten Clickhouse CI on x86 |
|
cc: @felipepessoto |
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
The DV bitmap invalid-row-index quarantine covered the native crash this PR's guard prevents by keeping the DV DML row-index target scan on Spark. Drop both patterns so the gate reports that crash class again. (cherry picked from commit 06e772e) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Thanks @felipepessoto cherry-picked now. |
The tagging rule can only reach a target scan that has Delta's BitmapAggregator as an ancestor of the same plan tree. Under AQE, columnar rules see one query stage at a time, so a shuffle join between the target scan and that aggregate -- the usual MERGE shape -- leaves the scan in a stage of its own, untagged, and it offloads natively. Delta builds every DML target relation over a TahoeBatchFileIndex carrying the command name, which survives stage splits and arbitrary join placement, so recognize the scan by that as well. The row-index and file-path shape check still gates the decision, keeping DML that rewrites whole files eligible for native execution. Cover MERGE in the handoff suites over both join shapes, since the broadcast shape alone cannot exercise the stage split. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Run Gluten Clickhouse CI on x86 |
Keeping the DV DML row-index scan on Spark makes these two tests read their data files through Spark's parquet reader, which resolves columns by parquet field id under column mapping id mode. Gluten's native Delta write does not emit those field ids, so the read fails; the native reader resolves by name and never noticed. That write gap is pre-existing and independent of this PR -- the baseline already carries DeltaColumnMappingSuite "id and name mode should write field_id in parquet schema" and DeletionVectorsSuite "DELETE with DVs with column mapping mode=id" for it. Baseline these two alongside them; all four come out together when the writer emits field ids. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI on x86 |
|
|
||
| case class OffloadDeltaScan() extends OffloadSingleNode { | ||
| case class OffloadDeltaScan( | ||
| enableNativeDeletionVectorDmlRowIndexScanKey: String) |
There was a problem hiding this comment.
nit: Can we just pass a boolean flag?
There was a problem hiding this comment.
Done in b55362e now a boolean. The rule is built per query from ColumnarRuleCall's live conf, so session-level toggling still works.
|
Thanks for the PR. I went through #12377 and was trying to understand why row ID 9223372036854775807 was written / read. Have you guys got any clue? cc @felipepessoto |
|
@zhztheplayer I couldn't reproduce it locally — including the real That said, I don't think So it looks like garbage reaching the aggregator's input column, not a real row index. Since it |
Per review: OffloadDeltaScan took the config key as a string and re-read the session conf per scan. The transform builder constructs the rule per query from ColumnarRuleCall's live conf, so a plain boolean carries the same session-level semantics without the lookup. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Run Gluten Clickhouse CI on x86 |
|
@zhztheplayer I tried to repro the issue several times and I couldn't. Maybe something else have fixed it. @malinjawi the current CI failures are not related to your change: |
Same VeloxUserError decimal-cast gap as their dozens of baselined siblings (e.g. the identical streaming MERGE case differing only in followAnsiEnabled); these parameterizations flap with shard content and fork parallelism and were missing from the machine-seeded baseline. Unrelated to this PR's DV DML scan changes. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
|
Run Gluten Clickhouse CI on x86 |
Agree with @felipepessoto everything we found also says garbage in the row-index column, not a real row ID. One upstream data point: velox#9943 documents the row-index vector getting only outputRows().size() slots written, with the tail serving recycled buffer contents as valid non-null BIGINTs same huge positive/negative value shapes as our two signatures. Not yet confirmed live in our pinned Velox, but the mechanism and the plan-shape-dependent intermittency line up. Should pair well with the diagnostics in #12783. |
Thanks @felipepessoto I'd added the same 4 entries on this branch (f8b4b5a) to unblock CI before seeing #12779/#12780. Same lines, so they'll merge cleanly whichever lands first; #12779 is clearly the right attribution rather than my commit message's flakiness guess. |
|
This is the most recent CI failure I found, from Aug 14: https://github.com/apache/gluten/actions/runs/31773203848/job/94686948558
|
|
Answering the question about row ID The chain, and how each step was checked. The scan emits a batch whose row-index child is empty. With Velox's operator output validation turned on: Downstream, that empty child gets wrapped in a dictionary of size 1, so the aggregator's input is a dictionary addressing row 0 of a base that has no rows: Reading it reads past the end of a zero-length buffer, so the value is whatever heap memory is there. That is why it differs on every occurrence. This now reproduces deterministically, which it did not before: with Why it looked so rare. The bounds check only rejects negatives and values above That means a garbage row index is usually added to the deletion vector rather than raising. Pointer-magnitude values address rows that do not exist; a The fix does not belong in the aggregator; it is the first reader of a column that was already malformed when the scan produced it. #12783 has the diagnostic that produced the output above. |
|
@malinjawi @felipepessoto Thank you for the contexts and investigation. If there is something else who caused the abnormal row ID, should we fix the bug first and hold this PR off? cc @malinjawi |
|
Following up on which stage builds the dictionary over an empty base: it is not Delta, and not the aggregator. It is an upstream Velox bug in the scan.
outputRows_.clear();
if (hasDeletion_) {
fillOutputRowsFromMutation(numValues);
numValues = outputRows_.size();
}
...
resultRowVector->unsafeResize(numValues); // parent <- numValues
setRowNumberField(currentRowNumber_, outputRows(), ..., childField); // child <- outputRows().size()and bool useOutputRows() const { return scanSpec_->hasFilter() || hasDeletion(); }So a scan with a filter but no deletion gets the empty The Delta DV write path hits all three conditions at once: it projects only
Upstream:
With the fix applied to the Velox revision we pin, the deterministic reproduction goes from 9 validation failures and a failed test to 0 and a pass, and the full Delta DV suite run with Two side notes that may be useful:
|
@zhztheplayer With @felipepessoto velox#18536 the root cause is now found and fixed at the right layer, so I don't think holding this PR buys anything as the guard is what keeps the affected path off by default until that fix is actually in our pin and proven. I would suggest we merge this (safe default + gate coverage), pull velox#18536 in the next bump, then a small follow-up flips Huge thanks @felipepessoto for the fix. |
|
@malinjawi Thanks. I am more or less inclined to hold this off unless such issue still exist after the fixes are landed. This PR can be a bit costly to disable a particular type of scan. It would be simpler to add an option to enable / disable the entire DV scan feature if you think it's feasible. I assume we are now missing that option in the codebase. |
@malinjawi the tests are not failing anymore, fixed by #12051. Now we have a different problem, that PR also fixed other 4 tests that we need to remove known issues list: #12815 |
|
facebookincubator/velox#18536 is merged Does anyone know in which version of daily Velox it will be included? 2026-08-20? |
What changes are proposed in this pull request?
Delta DELETE/UPDATE/MERGE with persistent deletion vectors runs a special target scan that returns each matched row's file path and row index. Native execution of that target scan is still experimental, so this PR keeps the scan and only its contiguous scan-adjacent Project/Filter chain on Spark by default. Joins, exchanges, BitmapAggregator, and the rest of the DML plan remain eligible for native execution.
The native target scan is available only when this experimental flag is enabled:
spark.gluten.sql.columnar.backend.velox.delta.enableNativeDmlRowIndexScanIt defaults to
false.spark.gluten.sql.columnar.backend.velox.delta.enableNativeWriteindependently controls command/write offload and is not a prerequisite for target-scan offload.Main pieces:
DeltaDeletionVectorDmlUtilsdetects and tags the structured BitmapAggregator/file-path/row-index scan shape.OffloadDeltaScanapplies the guarded fallback and copies the tag when the experimental native path is explicitly enabled.keepDmlRowIndexFallbackSubtreeOnSparklimits fallback propagation to the contiguous Project/Filter chain directly above the scan.VeloxDeltaConfigregisters and documents the experimental configuration._metadata.row_indexpredicates.This follows the JVM/native DV scan handoff in #12269 and works with the native BitmapAggregator merged in #12214. The DELETE implementation stack continues in #12216 and the benchmark harness is in #12217.
Related to, but not a root fix for, #12377: this PR prevents the experimental native DML target scan by default; it does not change native BitmapAggregator behavior.
Issue: #11901
How was this patch tested?
With JDK 17:
test-compilefor Spark 3.3, 3.4, 3.5, 4.0, and 4.1 with the Velox and Delta profilestest-compilefor Spark 3.5 and 4.0 after the latest CI amendmentgit diff --checkThe latest slow-lane failures were test-only. CI's final executed plans contained the intended Spark Project/Filter/ColumnarToRow/FileSourceScan fallback chain, but the assertions required a planning-time
TreeNodeTagthat does not reliably survive AQE plan copies. Final executed-plan assertions now identify the fallback from the same structured Delta file-path/row-index scan shape, while the production fallback decision continues to require the planning tag. Fresh native-runtime CI is running on commitde8e42218.Was this patch authored or co-authored using generative AI tooling?
Generated-by: IBM BOB