Skip to content

Commit 97c953a

Browse files
schenksjclaude
schenksj
and
claude
committed
fix(contrib-delta): emit default_row_commit_version metadata column
Delta 4.1 own-test-suite regression failures on CDC reads under row-tracking-enabled tables (especially the coordinated-commits batch backfill variant DeltaCDCScalaWithCatalogOwnedBatch2Suite) surfaced as 'Comet Internal Error: Output column count mismatch: expected 14, got 13'. The scan declared 14 output columns including the per-file `default_row_commit_version` metadata column, but the native DeltaSyntheticColumnsExec only knew about `base_row_id`, so the column was dropped on the way through and the upstream operator saw N-1. This commit: * Adds `default_row_commit_version` to JVM-side `syntheticNames` and `fixedMetadataNames` sets in CometDeltaNativeScan so it's included in `metadataColumnNamesEmitted` and the proto. * Adds the matching `META_DEFAULT_ROW_COMMIT_VERSION` constant, field schema, and emit branch in `synthetic_columns.rs`; extends `TaskMetadata` with the new field and wires it from `core_glue.rs` (the proto already carried the value via `task.default_row_commit_version`). Removes the `DeltaSyntheticColumnsExec: unknown metadata column name 'default_row_commit_version'` failure path. A second off-by-one in the same suite remains under investigation (separate column drop). Also bundled: * `CometScanWithPlanData.perPartitionFilePaths` trait method + `operators.scala` union-path collector now matches the trait instead of just `CometNativeScanExec`, so MERGE/UPDATE/DELETE flows that embed a Delta scan in a parent native tree no longer see empty `input_file_name()` -> `DELTA_FILE_TO_OVERWRITE_NOT_FOUND`. * `CometExecRDD.compute` populates `InputFileBlockHolder` whenever `partition.filePaths.nonEmpty` (not only the single-file case), matching PR apache#3932's approach. * Re-enables the two MERGE reproducers in `CometDeltaSpecialCharFilenameSuite`. * Adds `DeltaHiveTest.scala` Comet-wiring hunk to `4.1.0.diff` (the piece PR apache#3932's 4.0.0.diff had that ours was missing). * New `CometDeltaRegressionReproSuite` (one repro per root-cause cluster identified in the 4.1 regression) and a `CometDeltaCdcSuite`. * New `.github/workflows/delta_regression_test.yml` workflow that invokes `contrib/delta/dev/run-regression.sh` across Delta 3.3.2 / 4.0.0 / 4.1.0 with smoke -> full gating. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
1 parent eb44fac commit 97c953a

12 files changed

Lines changed: 640 additions & 26 deletions

File tree

‎.github/workflows/delta_regression_test.yml‎

Lines changed: 215 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,215 @@
1+
# Licensed to the Apache Software Foundation (ASF) under one
2+
# or more contributor license agreements. See the NOTICE file
3+
# distributed with this work for additional information
4+
# regarding copyright ownership. The ASF licenses this file
5+
# to you under the Apache License, Version 2.0 (the
6+
# "License"); you may not use this file except in compliance
7+
# with the License. You may obtain a copy of the License at
8+
#
9+
# http://www.apache.org/licenses/LICENSE-2.0
10+
#
11+
# Unless required by applicable law or agreed to in writing,
12+
# software distributed under the License is distributed on an
13+
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
# KIND, either express or implied. See the License for the
15+
# specific language governing permissions and limitations
16+
# under the License.
17+
18+
# Runs Delta Lake's OWN test suite against Comet (contrib-delta variant).
19+
# Mirrors the PR #3932 workflow but uses the contrib-delta build path:
20+
# Comet is installed with `-Pcontrib-delta`, the native library is built
21+
# with `--features contrib-delta`, and diffs live under
22+
# `contrib/delta/dev/diffs/`.
23+
#
24+
# The smoke job runs first as a cheap fail-fast (3 tests asserting Comet
25+
# is wired into Delta's test SparkSession). The full job runs Delta's
26+
# entire test suite for the matching Delta version with Comet enabled,
27+
# and only fires after the matching smoke cell passes.
28+
29+
name: Delta Lake Regression Tests (contrib-delta)
30+
31+
concurrency:
32+
group: ${{ github.repository }}-${{ github.head_ref || github.sha }}-${{ github.workflow }}
33+
cancel-in-progress: true
34+
35+
on:
36+
push:
37+
branches:
38+
- main
39+
paths-ignore:
40+
- "benchmarks/**"
41+
- "doc/**"
42+
- "docs/**"
43+
- "**.md"
44+
- "dev/changelog/*.md"
45+
- "native/core/benches/**"
46+
- "native/spark-expr/benches/**"
47+
- "spark/src/test/scala/org/apache/spark/sql/benchmark/**"
48+
pull_request:
49+
paths-ignore:
50+
- "benchmarks/**"
51+
- "doc/**"
52+
- "docs/**"
53+
- "**.md"
54+
- "dev/changelog/*.md"
55+
- "native/core/benches/**"
56+
- "native/spark-expr/benches/**"
57+
- "spark/src/test/scala/org/apache/spark/sql/benchmark/**"
58+
workflow_dispatch:
59+
60+
permissions:
61+
contents: read
62+
63+
env:
64+
RUST_VERSION: stable
65+
RUST_BACKTRACE: 1
66+
RUSTFLAGS: "-Clink-arg=-fuse-ld=bfd"
67+
68+
jobs:
69+
# Build libcomet ONCE with --features contrib-delta and share via artifact.
70+
# Identical to the build-native job in delta_contrib_test.yml; kept
71+
# separate to let this workflow run independently on workflow_dispatch.
72+
build-native:
73+
name: Build Native Library (contrib-delta)
74+
runs-on: ubuntu-24.04
75+
container:
76+
image: amd64/rust
77+
steps:
78+
- uses: actions/checkout@v6
79+
80+
- name: Setup Rust & Java toolchain
81+
uses: ./.github/actions/setup-builder
82+
with:
83+
rust-version: ${{ env.RUST_VERSION }}
84+
jdk-version: 17
85+
86+
- name: Restore Cargo cache
87+
uses: actions/cache/restore@v5
88+
with:
89+
path: |
90+
~/.cargo/registry
91+
~/.cargo/git
92+
native/target
93+
key: ${{ runner.os }}-cargo-ci-contrib-delta-${{ hashFiles('native/**/Cargo.lock', 'native/**/Cargo.toml', 'contrib/delta/native/**/Cargo.toml') }}-${{ hashFiles('native/**/*.rs', 'contrib/delta/native/**/*.rs') }}
94+
restore-keys: |
95+
${{ runner.os }}-cargo-ci-contrib-delta-${{ hashFiles('native/**/Cargo.lock', 'native/**/Cargo.toml', 'contrib/delta/native/**/Cargo.toml') }}-
96+
97+
- name: Build native library with contrib-delta
98+
run: |
99+
cd native && cargo build --profile ci --features contrib-delta
100+
env:
101+
RUSTFLAGS: "-Ctarget-cpu=x86-64-v3 -Clink-arg=-fuse-ld=bfd"
102+
103+
- name: Save Cargo cache
104+
uses: actions/cache/save@v5
105+
if: github.ref == 'refs/heads/main'
106+
with:
107+
path: |
108+
~/.cargo/registry
109+
~/.cargo/git
110+
native/target
111+
key: ${{ runner.os }}-cargo-ci-contrib-delta-${{ hashFiles('native/**/Cargo.lock', 'native/**/Cargo.toml', 'contrib/delta/native/**/Cargo.toml') }}-${{ hashFiles('native/**/*.rs', 'contrib/delta/native/**/*.rs') }}
112+
113+
- name: Upload native library
114+
uses: actions/upload-artifact@v7
115+
with:
116+
name: native-lib-delta-regression
117+
path: native/target/ci/libcomet.so
118+
retention-days: 1
119+
120+
# Smoke: 3 tests proving Comet is registered and active in Delta's
121+
# SparkSession. Fails the workflow before spending time on the full
122+
# suite if config-wiring has drifted.
123+
delta-smoke:
124+
needs: build-native
125+
strategy:
126+
matrix:
127+
include:
128+
- delta-version: '3.3.2'
129+
spark-short: '3.5'
130+
- delta-version: '4.0.0'
131+
spark-short: '4.0'
132+
- delta-version: '4.1.0'
133+
spark-short: '4.1'
134+
fail-fast: false
135+
name: smoke/delta-${{ matrix.delta-version }}
136+
runs-on: ubuntu-24.04
137+
container:
138+
image: amd64/rust
139+
env:
140+
SPARK_LOCAL_IP: localhost
141+
steps:
142+
- uses: actions/checkout@v6
143+
144+
- name: Setup Rust & Java toolchain
145+
uses: ./.github/actions/setup-builder
146+
with:
147+
rust-version: ${{ env.RUST_VERSION }}
148+
jdk-version: 17
149+
150+
- name: Download native library
151+
uses: actions/download-artifact@v8
152+
with:
153+
name: native-lib-delta-regression
154+
# run-regression.sh's FAST=1 path expects native/target/release/
155+
path: native/target/release/
156+
157+
- name: Run Delta smoke test with Comet
158+
run: |
159+
FAST=1 bash contrib/delta/dev/run-regression.sh ${{ matrix.delta-version }} smoke
160+
161+
- name: Upload regression log on failure
162+
if: failure()
163+
uses: actions/upload-artifact@v7
164+
with:
165+
name: smoke-log-delta-${{ matrix.delta-version }}
166+
path: target/delta-regression-logs/
167+
retention-days: 5
168+
169+
# Full Delta suite under Comet. Gated on the matching smoke cell.
170+
delta-full:
171+
needs: delta-smoke
172+
if: github.event_name == 'workflow_dispatch' || github.event_name == 'push'
173+
strategy:
174+
matrix:
175+
include:
176+
- delta-version: '3.3.2'
177+
spark-short: '3.5'
178+
- delta-version: '4.0.0'
179+
spark-short: '4.0'
180+
- delta-version: '4.1.0'
181+
spark-short: '4.1'
182+
fail-fast: false
183+
name: full/delta-${{ matrix.delta-version }}
184+
runs-on: ubuntu-24.04
185+
container:
186+
image: amd64/rust
187+
env:
188+
SPARK_LOCAL_IP: localhost
189+
steps:
190+
- uses: actions/checkout@v6
191+
192+
- name: Setup Rust & Java toolchain
193+
uses: ./.github/actions/setup-builder
194+
with:
195+
rust-version: ${{ env.RUST_VERSION }}
196+
jdk-version: 17
197+
198+
- name: Download native library
199+
uses: actions/download-artifact@v8
200+
with:
201+
name: native-lib-delta-regression
202+
path: native/target/release/
203+
204+
- name: Run Delta full suite with Comet
205+
timeout-minutes: 240
206+
run: |
207+
FAST=1 bash contrib/delta/dev/run-regression.sh ${{ matrix.delta-version }} full
208+
209+
- name: Upload regression log on failure
210+
if: failure()
211+
uses: actions/upload-artifact@v7
212+
with:
213+
name: full-log-delta-${{ matrix.delta-version }}
214+
path: target/delta-regression-logs/
215+
retention-days: 5

‎contrib/delta/dev/diffs/4.1.0.diff‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -230,3 +230,24 @@ diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/DeltaSuite.scala b/
230230
}
231231

232232
// Force the query to read files and generate metrics
233+
diff --git a/spark/src/test/scala/org/apache/spark/sql/delta/test/DeltaHiveTest.scala b/spark/src/test/scala/org/apache/spark/sql/delta/test/DeltaHiveTest.scala
234+
--- a/spark/src/test/scala/org/apache/spark/sql/delta/test/DeltaHiveTest.scala
235+
+++ b/spark/src/test/scala/org/apache/spark/sql/delta/test/DeltaHiveTest.scala
236+
@@ -43,6 +43,17 @@
237+
conf.set(SQLConf.V2_SESSION_CATALOG_IMPLEMENTATION.key, classOf[DeltaCatalog].getName)
238+
conf.set(StaticSQLConf.SPARK_SESSION_EXTENSIONS.key,
239+
classOf[DeltaSparkSessionExtension].getName)
240+
+ conf.set("spark.plugins", "org.apache.spark.CometPlugin")
241+
+ conf.set("spark.shuffle.manager",
242+
+ "org.apache.spark.sql.comet.execution.shuffle.CometShuffleManager")
243+
+ conf.set("spark.comet.enabled", "true")
244+
+ conf.set("spark.comet.exec.enabled", "true")
245+
+ conf.set("spark.comet.exec.shuffle.enabled", "true")
246+
+ conf.set("spark.comet.scan.enabled", "true")
247+
+ conf.set("spark.comet.scan.deltaNative.enabled", "true")
248+
+ conf.set("spark.comet.explainFallback.enabled", "true")
249+
+ conf.set("spark.memory.offHeap.enabled", "true")
250+
+ conf.set("spark.memory.offHeap.size", "10g")
251+
_sc = new SparkContext("local", this.getClass.getName, conf)
252+
_hiveContext = new TestHiveContext(_sc)
253+
_session = _hiveContext.sparkSession

‎contrib/delta/native/src/core_glue.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -209,6 +209,7 @@ impl PhysicalPlanner {
209209
byte_range_end: task.byte_range_end.map(|v| v as i64),
210210
modification_time_millis: task.modification_time,
211211
base_row_id: task.base_row_id,
212+
default_row_commit_version: task.default_row_commit_version,
212213
},
213214
);
214215
} else {

‎contrib/delta/native/src/synthetic_columns.rs‎

Lines changed: 27 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,6 +77,12 @@ pub const META_FILE_MODIFICATION_TIME: &str = "file_modification_time";
7777
/// `row_id = base_row_id + row_index`; the upstream Project does the addition, so this
7878
/// column carries the per-file constant.
7979
pub const META_BASE_ROW_ID: &str = "base_row_id";
80+
/// Delta's per-file `AddFile.defaultRowCommitVersion` surfaced as an attribute.
81+
/// Same role as [`META_BASE_ROW_ID`] but for the row-commit-version side: plans
82+
/// reading `_metadata.row_commit_version` on row-tracking-enabled tables before
83+
/// materialisation pick this up as the default when the parquet file doesn't
84+
/// carry a per-row version.
85+
pub const META_DEFAULT_ROW_COMMIT_VERSION: &str = "default_row_commit_version";
8086
/// Prefix for Delta's materialised row-id columns (`_row-id-col-<uuid>`). Present in
8187
/// `scan.requiredSchema` whenever row tracking is enabled but the parquet file may not
8288
/// contain the column (unmaterialised row IDs). Emit as null so the upstream Project
@@ -101,6 +107,9 @@ pub struct TaskMetadata {
101107
/// `AddFile.baseRowId`. Emitted as a per-file Int64 constant when the upstream
102108
/// asks for the `base_row_id` synthetic column.
103109
pub base_row_id: Option<i64>,
110+
/// `AddFile.defaultRowCommitVersion`. Emitted as a per-file Int64 constant when
111+
/// the upstream asks for the `default_row_commit_version` synthetic column.
112+
pub default_row_commit_version: Option<i64>,
104113
}
105114

106115
fn metadata_field(name: &str) -> Field {
@@ -121,7 +130,9 @@ fn metadata_field(name: &str) -> Field {
121130
DataType::Timestamp(TimeUnit::Microsecond, Some("UTC".into())),
122131
false,
123132
),
124-
META_BASE_ROW_ID => Field::new(name, DataType::Int64, true),
133+
META_BASE_ROW_ID | META_DEFAULT_ROW_COMMIT_VERSION => {
134+
Field::new(name, DataType::Int64, true)
135+
}
125136
_ => Field::new(name, DataType::Utf8, true),
126137
}
127138
}
@@ -572,6 +583,21 @@ impl DeltaSyntheticColumnsStream {
572583
let value = self.task_metadata.base_row_id.unwrap_or(0);
573584
Arc::new(Int64Array::from(vec![value; rows]))
574585
}
586+
META_DEFAULT_ROW_COMMIT_VERSION => {
587+
// Per-file constant from `AddFile.defaultRowCommitVersion`. Plans
588+
// reading `_metadata.row_commit_version` use this when the parquet
589+
// file doesn't materialise the column. Null when the AddFile lacks
590+
// a default (table doesn't track rows -> upstream Project emits null).
591+
match self.task_metadata.default_row_commit_version {
592+
Some(value) => {
593+
Arc::new(Int64Array::from(vec![Some(value); rows]))
594+
}
595+
None => {
596+
let nulls: Vec<Option<i64>> = vec![None; rows];
597+
Arc::new(Int64Array::from(nulls))
598+
}
599+
}
600+
}
575601
other if other.starts_with(ROW_ID_MATERIALISED_PREFIX)
576602
|| other.starts_with(ROW_COMMIT_VERSION_MATERIALISED_PREFIX) =>
577603
{

‎contrib/delta/src/main/scala/org/apache/comet/contrib/delta/CometDeltaNativeScan.scala‎

Lines changed: 15 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -881,10 +881,13 @@ object CometDeltaNativeScan extends CometOperatorSerde[CometScanExec] with Loggi
881881
"file_block_start",
882882
"file_block_length",
883883
"file_modification_time",
884-
// Delta row-tracking columns synthesised natively (base_row_id is per-file
885-
// constant from AddFile.baseRowId; the materialised columns are null when the
886-
// parquet file doesn't carry them).
887-
"base_row_id")
884+
// Delta row-tracking columns synthesised natively. Both are per-file constants
885+
// from AddFile.baseRowId / AddFile.defaultRowCommitVersion; the materialised
886+
// columns are null when the parquet file doesn't carry them. Must be kept in
887+
// sync with `fixedMetadataNames` below and the proto setters in
888+
// `buildTaskListFromAddFiles` so the native side actually emits these.
889+
"base_row_id",
890+
"default_row_commit_version")
888891
val isSynthetic = (f: StructField) => {
889892
val lc = f.name.toLowerCase(Locale.ROOT)
890893
syntheticNames.contains(lc) ||
@@ -917,7 +920,14 @@ object CometDeltaNativeScan extends CometOperatorSerde[CometScanExec] with Loggi
917920
"file_block_start",
918921
"file_block_length",
919922
"file_modification_time",
920-
"base_row_id")
923+
"base_row_id",
924+
// Delta row-tracking exposes `default_row_commit_version` as a per-file
925+
// metadata column alongside `base_row_id`. Missing this here means the
926+
// emit-name list passed to native drops the column, causing the
927+
// upstream operator to see N-1 cols where Spark expected N (e.g. CDC
928+
// reads on row-tracking-enabled tables, especially under
929+
// coordinated-commits backfill where this code path is reached).
930+
"default_row_commit_version")
921931
// The wrapped exec output is `parquet projection ++ row_index/is_row_deleted/...
922932
// ++ metadata_column_names` in the order metadata names are emitted. To make the
923933
// post-synthesis layout match scan.output WITHOUT a final reorder Project, walk

‎contrib/delta/src/main/scala/org/apache/spark/sql/comet/CometDeltaNativeScanExec.scala‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -258,6 +258,19 @@ case class CometDeltaNativeScanExec(
258258
def commonData: Array[Byte] = commonBytes
259259
def perPartitionData: Array[Array[Byte]] = planningPerPartitionBytes
260260

261+
// Surface per-partition file paths to the unified `CometExecRDD` path in
262+
// `operators.scala` so `InputFileBlockHolder` is populated when this scan
263+
// is embedded inside a larger Comet native tree (e.g. Delta MERGE's
264+
// `findTouchedFiles`). Without this, the parent operator's
265+
// `CometExecRDD.compute` sees empty filePaths -> `input_file_name()`
266+
// returns "" -> `DELTA_FILE_TO_OVERWRITE_NOT_FOUND`.
267+
override def perPartitionFilePaths: Array[Seq[String]] = {
268+
planningPerPartitionBytes.map { bytes =>
269+
OperatorOuterClass.DeltaScan.parseFrom(bytes)
270+
.getTasksList.asScala.map(_.getFilePath).toSeq
271+
}
272+
}
273+
261274
/**
262275
* Unique key for matching this scan's common/per-partition data to its operator in the native
263276
* plan. Must be distinct across multiple Delta scans in the same plan tree -- e.g. a self-join

0 commit comments

Comments
 (0)