[GLUTEN][VL] Defer Delta deletion vector reads to executors - #12836
[GLUTEN][VL] Defer Delta deletion vector reads to executors#12836malinjawi wants to merge 7 commits into
Conversation
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI |
|
Run Gluten Clickhouse CI on x86 |
1 similar comment
|
Run Gluten Clickhouse CI on x86 |
bb950e3 to
90aea76
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI |
90aea76 to
62cd99c
Compare
malinjawi
changed the title
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI |
62cd99c to
85b52e4
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI |
85b52e4 to
436f32a
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI |
1 similar comment
|
Run Gluten Clickhouse CI |
436f32a to
372f498
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI |
|
Are there any circumstances that could benefit from driver loading? If not, the legacy implementation should be removed. |
|
@marin-ma I checked the remaining cases and do not see a normal workload that benefits from loading on-disk DVs on the driver. The only theoretical differences are earlier failure for a missing/corrupt sidecar and avoiding a repeated DV request on a retried or speculative task; neither justifies preserving the serial planning path. The standard Delta fallback already reads DVs on executors. I will remove the rollback config and the legacy on-disk driver branch, keep inline DVs eager because their bytes are already in Delta metadata, and retain |
|
Run Gluten Clickhouse CI on x86 |
1 similar comment
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
Warning
Copilot couldn't run its full agentic review because it didn't start before the timeout. Make sure your repository has a runner available, or add a copilot-code-review.yml file specifying one with the runs-on attribute. See the docs for more details.
Defers Delta on-disk deletion vector (DV) payload reads from driver planning to executor-side split serialization to reduce driver bottlenecks and speed up job start.
Changes:
- Adds a serializable DV payload abstraction to support deferred executor materialization with per-task memoization.
- Wires DV read metrics (and descriptor-prep metrics) into native Delta scans and propagates them to executors.
- Updates/extends Delta DV handoff and deferred-read tests, plus CI known-failure baseline maintenance.
Reviewed changes
Copilot reviewed 16 out of 16 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
| gluten-substrait/src/main/scala/org/apache/gluten/execution/FileSourceScanExecTransformer.scala | Adds hook for format-specific scan metrics to be included in the native scan metrics map |
| gluten-substrait/src/main/java/org/apache/gluten/substrait/rel/DeltaLocalFilesNode.java | Introduces DeletionVectorPayload and updates DeltaFileReadOptions to materialize payload on demand |
| gluten-delta/src/main/scala/org/apache/gluten/execution/DeltaScanTransformer.scala | Registers new DV metrics and passes DV read metrics into DV normalization |
| gluten-delta/src/main/scala/org/apache/gluten/delta/DeletionVectorReadMetrics.scala | Implements deferred task-time accumulator registration for DV payload read metrics |
| gluten-delta/src/main/java/org/apache/gluten/delta/TaskAccumulatorRegistry.java | Adds helper to register deserialized accumulators once TaskContext is available |
| gluten-delta/src-delta33/main/scala/.../DeltaDeletionVectorScanInfo.scala | Implements deferred on-disk DV payload source + memoized executor materialization (Delta 3.3) |
| gluten-delta/src-delta40/main/scala/.../DeltaDeletionVectorScanInfo.scala | Same as above for Delta 4.0 |
| backends-velox/src-delta*/test/... | Adds/updates tests validating executor materialization, memoization, and metric propagation |
| .github/workflows/util/delta-spark-ut/known-failures.txt | Removes fixed baseline failures per CI policy |
💡 Add a code-review agent skill or configure MCP servers for context-aware, tailored reviews. Learn more in the docs.
| // Executor-side metrics only (excludes driverMetricsAlias). | ||
| @transient private lazy val executorSideScanMetrics: Map[String, SQLMetric] = | ||
| BackendsApiManager.getMetricsApiInstance | ||
| .genFileSourceScanTransformerMetrics(sparkContext) | ||
| .filter(m => !driverMetricsAlias.contains(m._1)) | ||
| .filter(m => !driverMetricsAlias.contains(m._1)) ++ additionalScanMetrics |
There was a problem hiding this comment.
Updated. The map is now named nativeScanMetrics, and the comment states that format-specific metrics may be updated on the driver or executors while driver-only aliases are excluded.
| private final byte[] payload; | ||
|
|
||
| public SerializedDeletionVectorPayload(byte[] payload) { | ||
| this.payload = payload == null ? new byte[0] : payload; |
There was a problem hiding this comment.
Updated. SerializedDeletionVectorPayload now clones the constructor input, with a regression test covering mutation of the caller-owned array. The on-disk deferred path is unaffected.
| public byte[] serializedDeletionVector() { | ||
| return serializedDeletionVector; | ||
| return deletionVectorPayload.materialize(); | ||
| } |
There was a problem hiding this comment.
Updated the method documentation to state that on-disk payload materialization may perform blocking filesystem I/O, is intended for executor-side split-to-protobuf conversion, and returns bytes that must not be modified. The existing method name is retained for API compatibility.
| metrics("dvDescriptorCount") | ||
| .add(deltaReadOptions.count(_.hasDeletionVector()).toLong) |
There was a problem hiding this comment.
I checked this path and kept the count where it is. It is a local O(files-per-partition) traversal on the driver; protobuf construction occurs later on the executor. Folding the count into normalization would widen the shared Delta-version API for negligible cost. We can revisit it if profiling shows this traversal is material.
| /** Avoid double registration when Spark deserializes this object after installing TaskContext. */ | ||
| private def readObject(input: ObjectInputStream): Unit = { | ||
| input.defaultReadObject() | ||
| registeredInTask = TaskContext.get() != null |
There was a problem hiding this comment.
I verified the deserialization order and kept the logic unchanged, with a clearer comment. defaultReadObject() deserializes the nested SQLMetrics first, and Spark AccumulatorV2.readObject registers each one when TaskContext exists. The wrapper then records that state to avoid registering them twice. If no context exists, registerForCurrentTask() performs the deferred registration. Resetting the flag to false would append duplicate accumulators to TaskMetrics; the end-to-end tests also assert exactly one reported read attempt.
3457718 to
c565daa
Compare
|
Run Gluten Clickhouse CI on x86 |
marin-ma
left a comment
marin-ma
left a comment
There was a problem hiding this comment.
Some minor comments. Thanks.
| deletionVectorPayload: DeletionVectorPayload) { | ||
| def serializedDeletionVector: Array[Byte] = deletionVectorPayload.materialize() | ||
|
|
||
| def isPayloadMaterialized: Boolean = deletionVectorPayload.isMaterialized() |
There was a problem hiding this comment.
I don't see this function being invoked elsewhere. Can you remove it? Ditto for src-delta40
There was a problem hiding this comment.
Removed isPayloadMaterialized from both the Delta 3.3 and 4.0 implementations. Thanks.
| } | ||
|
|
||
| /** A payload source for inline DVs whose bytes are already present in Delta metadata. */ | ||
| public static final class SerializedDeletionVectorPayload implements DeletionVectorPayload { |
There was a problem hiding this comment.
Would it be better to rename to InMemoryDeletionVectorPayload?
There was a problem hiding this comment.
Agreed. Renamed it to InMemoryDeletionVectorPayload and updated all references and tests.
c565daa to
6e95beb
Compare
|
Run Gluten Clickhouse CI on x86 |
|
Run Gluten Clickhouse CI |
|
Run Gluten Clickhouse CI on x86 |
marin-ma
left a comment
marin-ma
left a comment
There was a problem hiding this comment.
LGTM. Thanks!
cc: @zhztheplayer Do you have any comments?
|
Some delta UT failed. @malinjawi PTAL. |
Yeah, thanks @marin-ma. I checked the failed shards: they report zero new regressions. The failures are caused by 32 stale known-failure entries that now pass after #12821. #12907 removes exactly those entries. Once it merges, I’ll sync main and rerun the Delta CI. |
|
Run Gluten Clickhouse CI on x86 |
What changes are proposed in this pull request?
Native Delta scans currently load every on-disk deletion vector while the driver builds file splits. With 2,461 S3-backed DVs, this kept executors idle for about two minutes before the first Spark job.
This PR passes each DV's absolute path, offset, size, and Hadoop configuration with the Spark partition, then loads the payload during executor-side split serialization.
TaskContextbecomes available.The production DV data-path change is confined to the JVM handoff. The PR also adds metrics, tests, documentation, and required CI baseline maintenance.
spark.gluten.sql.columnar.filescan=falseremains the broad fallback that disables native file scan. Direct Velox range reads are handled separately in #12867.How was this patch tested?
git diff --check.Targeted DV reads
A Spark 3.5.4 validation build retained a temporary legacy switch solely to compare both paths from the same build. PR A used executor deferral with native range reading disabled; Legacy used driver loading with native range reading disabled. Native file scan and metadata row index were enabled in both arms. The final PR does not publish the legacy driver-loading switch.
Each arm ran three times in balanced order against the same snapshot and resources.
count(*)sumThe combined physical value is the median of each run's
sum + grouped_sumtime. SQL-start-to-first-job delay fell from 114.7 s to 0.64 s forsum, and from 111.2 s to 0.38 s for groupedsum.The snapshot contained 2,461 DV-bearing files and 6,479,887,870 visible rows after applying DV cardinality 720,032,919. Every arm returned identical results.
Full Delta ingestion
A separate balanced three-round test processed all 24 TPC-DS SF2500 tables. Only
store_salescarried DVs; native writing, output settings, inputs, and resources were fixed.store_salesAll runs completed 24 of 24 tables with matching schemas, partitioning, Delta metrics, row statistics, and corresponding output row/file counts.
The claims are limited to these Spark 3.5.4 workloads: 12.55x for the two targeted physical DV scans and 15.52% for full ingestion. They do not cover a full query suite, DV creation, native writing, or Spark 4 performance.
Was this patch authored or co-authored using generative AI tooling?
Generated-by: IBM BOB