From ec292a05ea6413a176728767f1302a4827515f72 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Thu, 12 Mar 2026 13:50:29 +0800 Subject: [PATCH 1/5] Spark: Add scd_type2 option to create_changelog_view procedure Introduces a new `scd_type2 => boolean` parameter to the `create_changelog_view` Spark procedure that natively produces a Slowly Changing Dimensions Type-2 history view from an Iceberg changelog table. When enabled, the procedure adds three new columns to the view: - `_valid_from` (TIMESTAMP): commit timestamp when this version became active - `_valid_to` (TIMESTAMP, nullable): commit timestamp when superseded; NULL if still active - `_is_current` (BOOLEAN): true when _valid_to IS NULL AND _change_type != 'DELETE' UPDATE_BEFORE rows are filtered out as they are intermediate artifacts. DELETE rows are kept with `_is_current = false` to track hard deletes. Validation: scd_type2 requires identifier_columns and is incompatible with net_changes. --- .../TestCreateChangelogViewProcedure.java | 159 ++++++++++++++++++ .../CreateChangelogViewProcedure.java | 80 ++++++++- .../TestCreateChangelogViewProcedure.java | 159 ++++++++++++++++++ .../CreateChangelogViewProcedure.java | 80 ++++++++- .../TestCreateChangelogViewProcedure.java | 159 ++++++++++++++++++ .../CreateChangelogViewProcedure.java | 80 ++++++++- 6 files changed, 714 insertions(+), 3 deletions(-) diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 3734d24e969b..1af0e6a9c2c9 100644 --- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.sql.Timestamp; import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; @@ -30,6 +31,7 @@ import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.spark.SparkReadOptions; +import org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure; import org.apache.spark.sql.types.StructField; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.TestTemplate; @@ -691,4 +693,161 @@ public void testUpdateWithInComparableType() { .hasMessageContaining( "Identifier field is required as table contains unorderable columns: [data]"); } + + @TestTemplate + public void testScdType2BasicInsertUpdateDelete() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + Snapshot snap1 = table.currentSnapshot(); + + sql("INSERT OVERWRITE %s VALUES (1, 'b')", tableName); + table.refresh(); + Snapshot snap2 = table.currentSnapshot(); + + sql("DELETE FROM %s WHERE id = 1", tableName); + table.refresh(); + Snapshot snap3 = table.currentSnapshot(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List rows = sql("SELECT * FROM %s ORDER BY _change_ordinal", viewName); + + assertThat(rows).hasSize(3); + + Timestamp ts1 = new Timestamp(snap1.timestampMillis()); + Timestamp ts2 = new Timestamp(snap2.timestampMillis()); + Timestamp ts3 = new Timestamp(snap3.timestampMillis()); + + // INSERT row: valid from snap1 to snap2 + assertThat(rows.get(0)) + .containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); + // UPDATE_AFTER row: valid from snap2 to snap3 + assertThat(rows.get(1)) + .containsExactly(1, "b", UPDATE_AFTER, 1, snap2.snapshotId(), ts2, ts3, false); + // DELETE row: valid from snap3, _valid_to is NULL, _is_current = false + assertThat(rows.get(2)) + .containsExactly(1, "b", DELETE, 2, snap3.snapshotId(), ts3, null, false); + } + + @TestTemplate + public void testScdType2CurrentRows() { + // Use a table partitioned by id so INSERT OVERWRITE only affects id=1's partition + sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName); + sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName); + sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + + sql("INSERT INTO %s VALUES (2, 'b')", tableName); + table.refresh(); + + // Update only id=1 (partition-level overwrite, id=2 is untouched) + sql("INSERT OVERWRITE %s VALUES (1, 'c')", tableName); + table.refresh(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List currentRows = + sql( + "SELECT id FROM %s WHERE %s = true ORDER BY id", + viewName, CreateChangelogViewProcedure.IS_CURRENT_COL); + + // id=1's UPDATE_AFTER row should be current; id=2's INSERT row should also be current + assertThat(currentRows).hasSize(2); + assertThat(currentRows.get(0)[0]).isEqualTo(1); + assertThat(currentRows.get(1)[0]).isEqualTo(2); + } + + @TestTemplate + public void testScdType2HardDelete() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + + sql("DELETE FROM %s WHERE id = 1", tableName); + table.refresh(); + Snapshot snap2 = table.currentSnapshot(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List rows = sql("SELECT * FROM %s ORDER BY _change_ordinal", viewName); + + assertThat(rows).hasSize(2); + + // DELETE row: _valid_to IS NULL, _is_current = false + Object[] deleteRow = rows.get(1); + assertThat(deleteRow[2]).isEqualTo(DELETE); + assertThat(deleteRow[6]).isNull(); // _valid_to + assertThat(deleteRow[7]).isEqualTo(false); // _is_current + } + + @TestTemplate + public void testScdType2RequiresIdentifierColumns() { + createTableWithTwoColumns(); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("SCD Type-2 requires identifier columns to be set"); + } + + @TestTemplate + public void testScdType2IncompatibleWithNetChanges() { + createTableWithIdentifierField(); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', identifier_columns => array('id'), scd_type2 => true, net_changes => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Cannot use net_changes with scd_type2"); + } + + @TestTemplate + public void testScdType2OutputSchema() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + var df = spark.sql("SELECT * FROM " + viewName); + var fieldNames = + Arrays.stream(df.schema().fields()).map(StructField::name).collect(Collectors.toList()); + + assertThat(fieldNames) + .containsExactly( + "id", + "data", + "_change_type", + "_change_ordinal", + "_commit_snapshot_id", + CreateChangelogViewProcedure.VALID_FROM_COL, + CreateChangelogViewProcedure.VALID_TO_COL, + CreateChangelogViewProcedure.IS_CURRENT_COL); + } } diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index 4e4a6fd252fb..c5735b8db24e 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -18,17 +18,21 @@ */ package org.apache.iceberg.spark.procedures; +import java.sql.Timestamp; import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Set; import java.util.function.Predicate; import java.util.stream.Collectors; +import org.apache.iceberg.ChangelogOperation; import org.apache.iceberg.MetadataColumns; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.spark.ChangelogIterator; import org.apache.iceberg.spark.source.SparkChangelogTable; @@ -37,11 +41,15 @@ import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.OrderUtils; import org.apache.spark.sql.connector.catalog.Identifier; import org.apache.spark.sql.connector.catalog.TableCatalog; import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter; +import org.apache.spark.sql.expressions.Window; +import org.apache.spark.sql.expressions.WindowSpec; +import org.apache.spark.sql.functions; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; @@ -97,6 +105,8 @@ public class CreateChangelogViewProcedure extends BaseProcedure { ProcedureParameter.optional("identifier_columns", STRING_ARRAY); private static final ProcedureParameter NET_CHANGES = ProcedureParameter.optional("net_changes", DataTypes.BooleanType); + private static final ProcedureParameter SCD_TYPE2_PARAM = + ProcedureParameter.optional("scd_type2", DataTypes.BooleanType); private static final ProcedureParameter[] PARAMETERS = new ProcedureParameter[] { @@ -106,8 +116,13 @@ public class CreateChangelogViewProcedure extends BaseProcedure { COMPUTE_UPDATES_PARAM, IDENTIFIER_COLUMNS_PARAM, NET_CHANGES, + SCD_TYPE2_PARAM, }; + public static final String VALID_FROM_COL = "_valid_from"; + public static final String VALID_TO_COL = "_valid_to"; + public static final String IS_CURRENT_COL = "_is_current"; + private static final StructType OUTPUT_TYPE = new StructType( new StructField[] { @@ -148,6 +163,7 @@ public InternalRow[] call(InternalRow args) { Dataset df = loadRows(changelogTableIdent, options(input)); boolean netChanges = input.asBoolean(NET_CHANGES, false); + boolean scdType2 = input.asBoolean(SCD_TYPE2_PARAM, false); String[] identifierColumns = identifierColumns(input, tableIdent); Set unorderableColumnNames = Arrays.stream(df.schema().fields()) @@ -160,7 +176,14 @@ public InternalRow[] call(InternalRow args) { "Identifier field is required as table contains unorderable columns: %s", unorderableColumnNames); - if (shouldComputeUpdateImages(input)) { + if (scdType2) { + Preconditions.checkArgument(!netChanges, "Cannot use net_changes with scd_type2"); + Preconditions.checkArgument( + identifierColumns.length > 0, + "SCD Type-2 requires identifier columns to be set"); + Table table = loadSparkTable(tableIdent).table(); + df = computeScdType2(identifierColumns, df, table); + } else if (shouldComputeUpdateImages(input)) { Preconditions.checkArgument(!netChanges, "Not support net changes with update images"); df = computeUpdateImages(identifierColumns, df); } else { @@ -189,6 +212,61 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2( + String[] identifierColumns, Dataset df, Table table) { + // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) + df = computeUpdateImages(identifierColumns, df); + + // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts + String changeTypeCol = MetadataColumns.CHANGE_TYPE.name(); + df = df.filter(df.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); + + // Step 3: build a Map from table metadata + Map snapshotTimestamps = Maps.newHashMap(); + for (Snapshot snapshot : table.snapshots()) { + snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); + } + + // Step 4: resolve _valid_from via a UDF + SparkSession spark = spark(); + spark + .udf() + .register( + "__iceberg_snapshot_timestamp", + (Long snapshotId) -> { + Long tsMs = snapshotTimestamps.get(snapshotId); + return tsMs != null ? new Timestamp(tsMs) : null; + }, + DataTypes.TimestampType); + + String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); + Dataset dfWithValidFrom = + df.withColumn( + VALID_FROM_COL, + functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); + + // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY _change_ordinal) + String changeOrdinalCol = MetadataColumns.CHANGE_ORDINAL.name(); + Column[] partitionCols = + Arrays.stream(identifierColumns) + .map(c -> dfWithValidFrom.col(delimitedName(c))) + .toArray(Column[]::new); + WindowSpec window = + Window.partitionBy(partitionCols) + .orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); + Dataset dfWithValidTo = + dfWithValidFrom.withColumn( + VALID_TO_COL, functions.lead(dfWithValidFrom.col(VALID_FROM_COL), 1).over(window)); + + // Step 6: compute _is_current = (_valid_to IS NULL AND _change_type != 'DELETE') + return dfWithValidTo.withColumn( + IS_CURRENT_COL, + dfWithValidTo + .col(VALID_TO_COL) + .isNull() + .and(dfWithValidTo.col(changeTypeCol).notEqual(ChangelogOperation.DELETE.name()))); + } + private boolean shouldComputeUpdateImages(ProcedureInput input) { // If the identifier columns are set, we compute pre/post update images by default. boolean defaultValue = input.isProvided(IDENTIFIER_COLUMNS_PARAM); diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 3734d24e969b..1af0e6a9c2c9 100644 --- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.sql.Timestamp; import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; @@ -30,6 +31,7 @@ import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.spark.SparkReadOptions; +import org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure; import org.apache.spark.sql.types.StructField; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.TestTemplate; @@ -691,4 +693,161 @@ public void testUpdateWithInComparableType() { .hasMessageContaining( "Identifier field is required as table contains unorderable columns: [data]"); } + + @TestTemplate + public void testScdType2BasicInsertUpdateDelete() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + Snapshot snap1 = table.currentSnapshot(); + + sql("INSERT OVERWRITE %s VALUES (1, 'b')", tableName); + table.refresh(); + Snapshot snap2 = table.currentSnapshot(); + + sql("DELETE FROM %s WHERE id = 1", tableName); + table.refresh(); + Snapshot snap3 = table.currentSnapshot(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List rows = sql("SELECT * FROM %s ORDER BY _change_ordinal", viewName); + + assertThat(rows).hasSize(3); + + Timestamp ts1 = new Timestamp(snap1.timestampMillis()); + Timestamp ts2 = new Timestamp(snap2.timestampMillis()); + Timestamp ts3 = new Timestamp(snap3.timestampMillis()); + + // INSERT row: valid from snap1 to snap2 + assertThat(rows.get(0)) + .containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); + // UPDATE_AFTER row: valid from snap2 to snap3 + assertThat(rows.get(1)) + .containsExactly(1, "b", UPDATE_AFTER, 1, snap2.snapshotId(), ts2, ts3, false); + // DELETE row: valid from snap3, _valid_to is NULL, _is_current = false + assertThat(rows.get(2)) + .containsExactly(1, "b", DELETE, 2, snap3.snapshotId(), ts3, null, false); + } + + @TestTemplate + public void testScdType2CurrentRows() { + // Use a table partitioned by id so INSERT OVERWRITE only affects id=1's partition + sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName); + sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName); + sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + + sql("INSERT INTO %s VALUES (2, 'b')", tableName); + table.refresh(); + + // Update only id=1 (partition-level overwrite, id=2 is untouched) + sql("INSERT OVERWRITE %s VALUES (1, 'c')", tableName); + table.refresh(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List currentRows = + sql( + "SELECT id FROM %s WHERE %s = true ORDER BY id", + viewName, CreateChangelogViewProcedure.IS_CURRENT_COL); + + // id=1's UPDATE_AFTER row should be current; id=2's INSERT row should also be current + assertThat(currentRows).hasSize(2); + assertThat(currentRows.get(0)[0]).isEqualTo(1); + assertThat(currentRows.get(1)[0]).isEqualTo(2); + } + + @TestTemplate + public void testScdType2HardDelete() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + + sql("DELETE FROM %s WHERE id = 1", tableName); + table.refresh(); + Snapshot snap2 = table.currentSnapshot(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List rows = sql("SELECT * FROM %s ORDER BY _change_ordinal", viewName); + + assertThat(rows).hasSize(2); + + // DELETE row: _valid_to IS NULL, _is_current = false + Object[] deleteRow = rows.get(1); + assertThat(deleteRow[2]).isEqualTo(DELETE); + assertThat(deleteRow[6]).isNull(); // _valid_to + assertThat(deleteRow[7]).isEqualTo(false); // _is_current + } + + @TestTemplate + public void testScdType2RequiresIdentifierColumns() { + createTableWithTwoColumns(); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("SCD Type-2 requires identifier columns to be set"); + } + + @TestTemplate + public void testScdType2IncompatibleWithNetChanges() { + createTableWithIdentifierField(); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', identifier_columns => array('id'), scd_type2 => true, net_changes => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Cannot use net_changes with scd_type2"); + } + + @TestTemplate + public void testScdType2OutputSchema() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + var df = spark.sql("SELECT * FROM " + viewName); + var fieldNames = + Arrays.stream(df.schema().fields()).map(StructField::name).collect(Collectors.toList()); + + assertThat(fieldNames) + .containsExactly( + "id", + "data", + "_change_type", + "_change_ordinal", + "_commit_snapshot_id", + CreateChangelogViewProcedure.VALID_FROM_COL, + CreateChangelogViewProcedure.VALID_TO_COL, + CreateChangelogViewProcedure.IS_CURRENT_COL); + } } diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index 73c94f778510..c9627c3916a0 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -18,6 +18,7 @@ */ package org.apache.iceberg.spark.procedures; +import java.sql.Timestamp; import java.util.Arrays; import java.util.Iterator; import java.util.List; @@ -25,11 +26,14 @@ import java.util.Set; import java.util.function.Predicate; import java.util.stream.Collectors; +import org.apache.iceberg.ChangelogOperation; import org.apache.iceberg.MetadataColumns; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.spark.ChangelogIterator; import org.apache.iceberg.spark.source.SparkChangelogTable; @@ -39,6 +43,7 @@ import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.OrderUtils; import org.apache.spark.sql.connector.catalog.Identifier; @@ -46,6 +51,9 @@ import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure; import org.apache.spark.sql.connector.catalog.procedures.ProcedureParameter; import org.apache.spark.sql.connector.read.Scan; +import org.apache.spark.sql.expressions.Window; +import org.apache.spark.sql.expressions.WindowSpec; +import org.apache.spark.sql.functions; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; @@ -103,6 +111,8 @@ public class CreateChangelogViewProcedure extends BaseProcedure { optionalInParameter("identifier_columns", STRING_ARRAY); private static final ProcedureParameter NET_CHANGES = optionalInParameter("net_changes", DataTypes.BooleanType); + private static final ProcedureParameter SCD_TYPE2_PARAM = + optionalInParameter("scd_type2", DataTypes.BooleanType); private static final ProcedureParameter[] PARAMETERS = new ProcedureParameter[] { @@ -112,8 +122,13 @@ public class CreateChangelogViewProcedure extends BaseProcedure { COMPUTE_UPDATES_PARAM, IDENTIFIER_COLUMNS_PARAM, NET_CHANGES, + SCD_TYPE2_PARAM, }; + public static final String VALID_FROM_COL = "_valid_from"; + public static final String VALID_TO_COL = "_valid_to"; + public static final String IS_CURRENT_COL = "_is_current"; + private static final StructType OUTPUT_TYPE = new StructType( new StructField[] { @@ -154,6 +169,7 @@ public Iterator call(InternalRow args) { Dataset df = loadRows(changelogTableIdent, options(input)); boolean netChanges = input.asBoolean(NET_CHANGES, false); + boolean scdType2 = input.asBoolean(SCD_TYPE2_PARAM, false); String[] identifierColumns = identifierColumns(input, tableIdent); Set unorderableColumnNames = Arrays.stream(df.schema().fields()) @@ -166,7 +182,14 @@ public Iterator call(InternalRow args) { "Identifier field is required as table contains unorderable columns: %s", unorderableColumnNames); - if (shouldComputeUpdateImages(input)) { + if (scdType2) { + Preconditions.checkArgument(!netChanges, "Cannot use net_changes with scd_type2"); + Preconditions.checkArgument( + identifierColumns.length > 0, + "SCD Type-2 requires identifier columns to be set"); + Table table = loadSparkTable(tableIdent).table(); + df = computeScdType2(identifierColumns, df, table); + } else if (shouldComputeUpdateImages(input)) { Preconditions.checkArgument(!netChanges, "Not support net changes with update images"); df = computeUpdateImages(identifierColumns, df); } else { @@ -195,6 +218,61 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2( + String[] identifierColumns, Dataset df, Table table) { + // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) + df = computeUpdateImages(identifierColumns, df); + + // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts + String changeTypeCol = MetadataColumns.CHANGE_TYPE.name(); + df = df.filter(df.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); + + // Step 3: build a Map from table metadata + Map snapshotTimestamps = Maps.newHashMap(); + for (Snapshot snapshot : table.snapshots()) { + snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); + } + + // Step 4: resolve _valid_from via a UDF + SparkSession spark = spark(); + spark + .udf() + .register( + "__iceberg_snapshot_timestamp", + (Long snapshotId) -> { + Long tsMs = snapshotTimestamps.get(snapshotId); + return tsMs != null ? new Timestamp(tsMs) : null; + }, + DataTypes.TimestampType); + + String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); + Dataset dfWithValidFrom = + df.withColumn( + VALID_FROM_COL, + functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); + + // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY _change_ordinal) + String changeOrdinalCol = MetadataColumns.CHANGE_ORDINAL.name(); + Column[] partitionCols = + Arrays.stream(identifierColumns) + .map(c -> dfWithValidFrom.col(delimitedName(c))) + .toArray(Column[]::new); + WindowSpec window = + Window.partitionBy(partitionCols) + .orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); + Dataset dfWithValidTo = + dfWithValidFrom.withColumn( + VALID_TO_COL, functions.lead(dfWithValidFrom.col(VALID_FROM_COL), 1).over(window)); + + // Step 6: compute _is_current = (_valid_to IS NULL AND _change_type != 'DELETE') + return dfWithValidTo.withColumn( + IS_CURRENT_COL, + dfWithValidTo + .col(VALID_TO_COL) + .isNull() + .and(dfWithValidTo.col(changeTypeCol).notEqual(ChangelogOperation.DELETE.name()))); + } + private boolean shouldComputeUpdateImages(ProcedureInput input) { // If the identifier columns are set, we compute pre/post update images by default. boolean defaultValue = input.isProvided(IDENTIFIER_COLUMNS_PARAM); diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 3734d24e969b..1af0e6a9c2c9 100644 --- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -21,6 +21,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.sql.Timestamp; import java.util.Arrays; import java.util.List; import java.util.stream.Collectors; @@ -30,6 +31,7 @@ import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.spark.SparkReadOptions; +import org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure; import org.apache.spark.sql.types.StructField; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.TestTemplate; @@ -691,4 +693,161 @@ public void testUpdateWithInComparableType() { .hasMessageContaining( "Identifier field is required as table contains unorderable columns: [data]"); } + + @TestTemplate + public void testScdType2BasicInsertUpdateDelete() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + Snapshot snap1 = table.currentSnapshot(); + + sql("INSERT OVERWRITE %s VALUES (1, 'b')", tableName); + table.refresh(); + Snapshot snap2 = table.currentSnapshot(); + + sql("DELETE FROM %s WHERE id = 1", tableName); + table.refresh(); + Snapshot snap3 = table.currentSnapshot(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List rows = sql("SELECT * FROM %s ORDER BY _change_ordinal", viewName); + + assertThat(rows).hasSize(3); + + Timestamp ts1 = new Timestamp(snap1.timestampMillis()); + Timestamp ts2 = new Timestamp(snap2.timestampMillis()); + Timestamp ts3 = new Timestamp(snap3.timestampMillis()); + + // INSERT row: valid from snap1 to snap2 + assertThat(rows.get(0)) + .containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); + // UPDATE_AFTER row: valid from snap2 to snap3 + assertThat(rows.get(1)) + .containsExactly(1, "b", UPDATE_AFTER, 1, snap2.snapshotId(), ts2, ts3, false); + // DELETE row: valid from snap3, _valid_to is NULL, _is_current = false + assertThat(rows.get(2)) + .containsExactly(1, "b", DELETE, 2, snap3.snapshotId(), ts3, null, false); + } + + @TestTemplate + public void testScdType2CurrentRows() { + // Use a table partitioned by id so INSERT OVERWRITE only affects id=1's partition + sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName); + sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName); + sql("ALTER TABLE %s ADD PARTITION FIELD id", tableName); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + + sql("INSERT INTO %s VALUES (2, 'b')", tableName); + table.refresh(); + + // Update only id=1 (partition-level overwrite, id=2 is untouched) + sql("INSERT OVERWRITE %s VALUES (1, 'c')", tableName); + table.refresh(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List currentRows = + sql( + "SELECT id FROM %s WHERE %s = true ORDER BY id", + viewName, CreateChangelogViewProcedure.IS_CURRENT_COL); + + // id=1's UPDATE_AFTER row should be current; id=2's INSERT row should also be current + assertThat(currentRows).hasSize(2); + assertThat(currentRows.get(0)[0]).isEqualTo(1); + assertThat(currentRows.get(1)[0]).isEqualTo(2); + } + + @TestTemplate + public void testScdType2HardDelete() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + Table table = validationCatalog.loadTable(tableIdent); + + sql("DELETE FROM %s WHERE id = 1", tableName); + table.refresh(); + Snapshot snap2 = table.currentSnapshot(); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + List rows = sql("SELECT * FROM %s ORDER BY _change_ordinal", viewName); + + assertThat(rows).hasSize(2); + + // DELETE row: _valid_to IS NULL, _is_current = false + Object[] deleteRow = rows.get(1); + assertThat(deleteRow[2]).isEqualTo(DELETE); + assertThat(deleteRow[6]).isNull(); // _valid_to + assertThat(deleteRow[7]).isEqualTo(false); // _is_current + } + + @TestTemplate + public void testScdType2RequiresIdentifierColumns() { + createTableWithTwoColumns(); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("SCD Type-2 requires identifier columns to be set"); + } + + @TestTemplate + public void testScdType2IncompatibleWithNetChanges() { + createTableWithIdentifierField(); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', identifier_columns => array('id'), scd_type2 => true, net_changes => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Cannot use net_changes with scd_type2"); + } + + @TestTemplate + public void testScdType2OutputSchema() { + createTableWithIdentifierField(); + + sql("INSERT INTO %s VALUES (1, 'a')", tableName); + + List returns = + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName); + + String viewName = (String) returns.get(0)[0]; + var df = spark.sql("SELECT * FROM " + viewName); + var fieldNames = + Arrays.stream(df.schema().fields()).map(StructField::name).collect(Collectors.toList()); + + assertThat(fieldNames) + .containsExactly( + "id", + "data", + "_change_type", + "_change_ordinal", + "_commit_snapshot_id", + CreateChangelogViewProcedure.VALID_FROM_COL, + CreateChangelogViewProcedure.VALID_TO_COL, + CreateChangelogViewProcedure.IS_CURRENT_COL); + } } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index 73c94f778510..c9627c3916a0 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -18,6 +18,7 @@ */ package org.apache.iceberg.spark.procedures; +import java.sql.Timestamp; import java.util.Arrays; import java.util.Iterator; import java.util.List; @@ -25,11 +26,14 @@ import java.util.Set; import java.util.function.Predicate; import java.util.stream.Collectors; +import org.apache.iceberg.ChangelogOperation; import org.apache.iceberg.MetadataColumns; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.spark.ChangelogIterator; import org.apache.iceberg.spark.source.SparkChangelogTable; @@ -39,6 +43,7 @@ import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.OrderUtils; import org.apache.spark.sql.connector.catalog.Identifier; @@ -46,6 +51,9 @@ import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure; import org.apache.spark.sql.connector.catalog.procedures.ProcedureParameter; import org.apache.spark.sql.connector.read.Scan; +import org.apache.spark.sql.expressions.Window; +import org.apache.spark.sql.expressions.WindowSpec; +import org.apache.spark.sql.functions; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; @@ -103,6 +111,8 @@ public class CreateChangelogViewProcedure extends BaseProcedure { optionalInParameter("identifier_columns", STRING_ARRAY); private static final ProcedureParameter NET_CHANGES = optionalInParameter("net_changes", DataTypes.BooleanType); + private static final ProcedureParameter SCD_TYPE2_PARAM = + optionalInParameter("scd_type2", DataTypes.BooleanType); private static final ProcedureParameter[] PARAMETERS = new ProcedureParameter[] { @@ -112,8 +122,13 @@ public class CreateChangelogViewProcedure extends BaseProcedure { COMPUTE_UPDATES_PARAM, IDENTIFIER_COLUMNS_PARAM, NET_CHANGES, + SCD_TYPE2_PARAM, }; + public static final String VALID_FROM_COL = "_valid_from"; + public static final String VALID_TO_COL = "_valid_to"; + public static final String IS_CURRENT_COL = "_is_current"; + private static final StructType OUTPUT_TYPE = new StructType( new StructField[] { @@ -154,6 +169,7 @@ public Iterator call(InternalRow args) { Dataset df = loadRows(changelogTableIdent, options(input)); boolean netChanges = input.asBoolean(NET_CHANGES, false); + boolean scdType2 = input.asBoolean(SCD_TYPE2_PARAM, false); String[] identifierColumns = identifierColumns(input, tableIdent); Set unorderableColumnNames = Arrays.stream(df.schema().fields()) @@ -166,7 +182,14 @@ public Iterator call(InternalRow args) { "Identifier field is required as table contains unorderable columns: %s", unorderableColumnNames); - if (shouldComputeUpdateImages(input)) { + if (scdType2) { + Preconditions.checkArgument(!netChanges, "Cannot use net_changes with scd_type2"); + Preconditions.checkArgument( + identifierColumns.length > 0, + "SCD Type-2 requires identifier columns to be set"); + Table table = loadSparkTable(tableIdent).table(); + df = computeScdType2(identifierColumns, df, table); + } else if (shouldComputeUpdateImages(input)) { Preconditions.checkArgument(!netChanges, "Not support net changes with update images"); df = computeUpdateImages(identifierColumns, df); } else { @@ -195,6 +218,61 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2( + String[] identifierColumns, Dataset df, Table table) { + // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) + df = computeUpdateImages(identifierColumns, df); + + // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts + String changeTypeCol = MetadataColumns.CHANGE_TYPE.name(); + df = df.filter(df.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); + + // Step 3: build a Map from table metadata + Map snapshotTimestamps = Maps.newHashMap(); + for (Snapshot snapshot : table.snapshots()) { + snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); + } + + // Step 4: resolve _valid_from via a UDF + SparkSession spark = spark(); + spark + .udf() + .register( + "__iceberg_snapshot_timestamp", + (Long snapshotId) -> { + Long tsMs = snapshotTimestamps.get(snapshotId); + return tsMs != null ? new Timestamp(tsMs) : null; + }, + DataTypes.TimestampType); + + String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); + Dataset dfWithValidFrom = + df.withColumn( + VALID_FROM_COL, + functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); + + // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY _change_ordinal) + String changeOrdinalCol = MetadataColumns.CHANGE_ORDINAL.name(); + Column[] partitionCols = + Arrays.stream(identifierColumns) + .map(c -> dfWithValidFrom.col(delimitedName(c))) + .toArray(Column[]::new); + WindowSpec window = + Window.partitionBy(partitionCols) + .orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); + Dataset dfWithValidTo = + dfWithValidFrom.withColumn( + VALID_TO_COL, functions.lead(dfWithValidFrom.col(VALID_FROM_COL), 1).over(window)); + + // Step 6: compute _is_current = (_valid_to IS NULL AND _change_type != 'DELETE') + return dfWithValidTo.withColumn( + IS_CURRENT_COL, + dfWithValidTo + .col(VALID_TO_COL) + .isNull() + .and(dfWithValidTo.col(changeTypeCol).notEqual(ChangelogOperation.DELETE.name()))); + } + private boolean shouldComputeUpdateImages(ProcedureInput input) { // If the identifier columns are set, we compute pre/post update images by default. boolean defaultValue = input.isProvided(IDENTIFIER_COLUMNS_PARAM); From 1967f4f3c735b3d153feb6f696093556523ccfe7 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Thu, 12 Mar 2026 14:10:15 +0800 Subject: [PATCH 2/5] Apply Spotless formatting fixes --- .../extensions/TestCreateChangelogViewProcedure.java | 3 +-- .../procedures/CreateChangelogViewProcedure.java | 12 +++++------- .../extensions/TestCreateChangelogViewProcedure.java | 3 +-- .../procedures/CreateChangelogViewProcedure.java | 12 +++++------- .../extensions/TestCreateChangelogViewProcedure.java | 3 +-- .../procedures/CreateChangelogViewProcedure.java | 12 +++++------- 6 files changed, 18 insertions(+), 27 deletions(-) diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 1af0e6a9c2c9..438e098c8e3b 100644 --- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -725,8 +725,7 @@ public void testScdType2BasicInsertUpdateDelete() { Timestamp ts3 = new Timestamp(snap3.timestampMillis()); // INSERT row: valid from snap1 to snap2 - assertThat(rows.get(0)) - .containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); + assertThat(rows.get(0)).containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); // UPDATE_AFTER row: valid from snap2 to snap3 assertThat(rows.get(1)) .containsExactly(1, "b", UPDATE_AFTER, 1, snap2.snapshotId(), ts2, ts3, false); diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index c5735b8db24e..b5f002f33e15 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -179,8 +179,7 @@ public InternalRow[] call(InternalRow args) { if (scdType2) { Preconditions.checkArgument(!netChanges, "Cannot use net_changes with scd_type2"); Preconditions.checkArgument( - identifierColumns.length > 0, - "SCD Type-2 requires identifier columns to be set"); + identifierColumns.length > 0, "SCD Type-2 requires identifier columns to be set"); Table table = loadSparkTable(tableIdent).table(); df = computeScdType2(identifierColumns, df, table); } else if (shouldComputeUpdateImages(input)) { @@ -212,8 +211,7 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2( - String[] identifierColumns, Dataset df, Table table) { + private Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) df = computeUpdateImages(identifierColumns, df); @@ -245,15 +243,15 @@ private Dataset computeScdType2( VALID_FROM_COL, functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); - // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY _change_ordinal) + // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY + // _change_ordinal) String changeOrdinalCol = MetadataColumns.CHANGE_ORDINAL.name(); Column[] partitionCols = Arrays.stream(identifierColumns) .map(c -> dfWithValidFrom.col(delimitedName(c))) .toArray(Column[]::new); WindowSpec window = - Window.partitionBy(partitionCols) - .orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); + Window.partitionBy(partitionCols).orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); Dataset dfWithValidTo = dfWithValidFrom.withColumn( VALID_TO_COL, functions.lead(dfWithValidFrom.col(VALID_FROM_COL), 1).over(window)); diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 1af0e6a9c2c9..438e098c8e3b 100644 --- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -725,8 +725,7 @@ public void testScdType2BasicInsertUpdateDelete() { Timestamp ts3 = new Timestamp(snap3.timestampMillis()); // INSERT row: valid from snap1 to snap2 - assertThat(rows.get(0)) - .containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); + assertThat(rows.get(0)).containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); // UPDATE_AFTER row: valid from snap2 to snap3 assertThat(rows.get(1)) .containsExactly(1, "b", UPDATE_AFTER, 1, snap2.snapshotId(), ts2, ts3, false); diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index c9627c3916a0..101186e5831c 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -185,8 +185,7 @@ public Iterator call(InternalRow args) { if (scdType2) { Preconditions.checkArgument(!netChanges, "Cannot use net_changes with scd_type2"); Preconditions.checkArgument( - identifierColumns.length > 0, - "SCD Type-2 requires identifier columns to be set"); + identifierColumns.length > 0, "SCD Type-2 requires identifier columns to be set"); Table table = loadSparkTable(tableIdent).table(); df = computeScdType2(identifierColumns, df, table); } else if (shouldComputeUpdateImages(input)) { @@ -218,8 +217,7 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2( - String[] identifierColumns, Dataset df, Table table) { + private Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) df = computeUpdateImages(identifierColumns, df); @@ -251,15 +249,15 @@ private Dataset computeScdType2( VALID_FROM_COL, functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); - // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY _change_ordinal) + // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY + // _change_ordinal) String changeOrdinalCol = MetadataColumns.CHANGE_ORDINAL.name(); Column[] partitionCols = Arrays.stream(identifierColumns) .map(c -> dfWithValidFrom.col(delimitedName(c))) .toArray(Column[]::new); WindowSpec window = - Window.partitionBy(partitionCols) - .orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); + Window.partitionBy(partitionCols).orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); Dataset dfWithValidTo = dfWithValidFrom.withColumn( VALID_TO_COL, functions.lead(dfWithValidFrom.col(VALID_FROM_COL), 1).over(window)); diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 1af0e6a9c2c9..438e098c8e3b 100644 --- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -725,8 +725,7 @@ public void testScdType2BasicInsertUpdateDelete() { Timestamp ts3 = new Timestamp(snap3.timestampMillis()); // INSERT row: valid from snap1 to snap2 - assertThat(rows.get(0)) - .containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); + assertThat(rows.get(0)).containsExactly(1, "a", INSERT, 0, snap1.snapshotId(), ts1, ts2, false); // UPDATE_AFTER row: valid from snap2 to snap3 assertThat(rows.get(1)) .containsExactly(1, "b", UPDATE_AFTER, 1, snap2.snapshotId(), ts2, ts3, false); diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index c9627c3916a0..101186e5831c 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -185,8 +185,7 @@ public Iterator call(InternalRow args) { if (scdType2) { Preconditions.checkArgument(!netChanges, "Cannot use net_changes with scd_type2"); Preconditions.checkArgument( - identifierColumns.length > 0, - "SCD Type-2 requires identifier columns to be set"); + identifierColumns.length > 0, "SCD Type-2 requires identifier columns to be set"); Table table = loadSparkTable(tableIdent).table(); df = computeScdType2(identifierColumns, df, table); } else if (shouldComputeUpdateImages(input)) { @@ -218,8 +217,7 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2( - String[] identifierColumns, Dataset df, Table table) { + private Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) df = computeUpdateImages(identifierColumns, df); @@ -251,15 +249,15 @@ private Dataset computeScdType2( VALID_FROM_COL, functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); - // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY _change_ordinal) + // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY + // _change_ordinal) String changeOrdinalCol = MetadataColumns.CHANGE_ORDINAL.name(); Column[] partitionCols = Arrays.stream(identifierColumns) .map(c -> dfWithValidFrom.col(delimitedName(c))) .toArray(Column[]::new); WindowSpec window = - Window.partitionBy(partitionCols) - .orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); + Window.partitionBy(partitionCols).orderBy(dfWithValidFrom.col(changeOrdinalCol).asc()); Dataset dfWithValidTo = dfWithValidFrom.withColumn( VALID_TO_COL, functions.lead(dfWithValidFrom.col(VALID_FROM_COL), 1).over(window)); From b30f31cb58a480d7cc96676188099b5b6083fa3f Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Thu, 12 Mar 2026 14:38:32 +0800 Subject: [PATCH 3/5] Fix Checkstyle ParameterAssignment violations in computeScdType2 --- .../procedures/CreateChangelogViewProcedure.java | 11 +++++++---- .../procedures/CreateChangelogViewProcedure.java | 11 +++++++---- .../procedures/CreateChangelogViewProcedure.java | 11 +++++++---- 3 files changed, 21 insertions(+), 12 deletions(-) diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index b5f002f33e15..10336b80fdb6 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -213,11 +213,13 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) - df = computeUpdateImages(identifierColumns, df); + Dataset withUpdates = computeUpdateImages(identifierColumns, df); // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts String changeTypeCol = MetadataColumns.CHANGE_TYPE.name(); - df = df.filter(df.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); + Dataset withoutUpdateBefore = + withUpdates.filter( + withUpdates.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); // Step 3: build a Map from table metadata Map snapshotTimestamps = Maps.newHashMap(); @@ -239,9 +241,10 @@ private Dataset computeScdType2(String[] identifierColumns, Dataset df String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); Dataset dfWithValidFrom = - df.withColumn( + withoutUpdateBefore.withColumn( VALID_FROM_COL, - functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); + functions.callUDF( + "__iceberg_snapshot_timestamp", withoutUpdateBefore.col(snapshotIdCol))); // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY // _change_ordinal) diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index 101186e5831c..abee78d5e695 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -219,11 +219,13 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) - df = computeUpdateImages(identifierColumns, df); + Dataset withUpdates = computeUpdateImages(identifierColumns, df); // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts String changeTypeCol = MetadataColumns.CHANGE_TYPE.name(); - df = df.filter(df.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); + Dataset withoutUpdateBefore = + withUpdates.filter( + withUpdates.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); // Step 3: build a Map from table metadata Map snapshotTimestamps = Maps.newHashMap(); @@ -245,9 +247,10 @@ private Dataset computeScdType2(String[] identifierColumns, Dataset df String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); Dataset dfWithValidFrom = - df.withColumn( + withoutUpdateBefore.withColumn( VALID_FROM_COL, - functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); + functions.callUDF( + "__iceberg_snapshot_timestamp", withoutUpdateBefore.col(snapshotIdCol))); // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY // _change_ordinal) diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index 101186e5831c..abee78d5e695 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -219,11 +219,13 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) - df = computeUpdateImages(identifierColumns, df); + Dataset withUpdates = computeUpdateImages(identifierColumns, df); // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts String changeTypeCol = MetadataColumns.CHANGE_TYPE.name(); - df = df.filter(df.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); + Dataset withoutUpdateBefore = + withUpdates.filter( + withUpdates.col(changeTypeCol).notEqual(ChangelogOperation.UPDATE_BEFORE.name())); // Step 3: build a Map from table metadata Map snapshotTimestamps = Maps.newHashMap(); @@ -245,9 +247,10 @@ private Dataset computeScdType2(String[] identifierColumns, Dataset df String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); Dataset dfWithValidFrom = - df.withColumn( + withoutUpdateBefore.withColumn( VALID_FROM_COL, - functions.callUDF("__iceberg_snapshot_timestamp", df.col(snapshotIdCol))); + functions.callUDF( + "__iceberg_snapshot_timestamp", withoutUpdateBefore.col(snapshotIdCol))); // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY // _change_ordinal) From 40f83f48c54d5a737583b70bce4af66cd96d4df4 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Wed, 26 Aug 2026 10:39:46 +0900 Subject: [PATCH 4/5] Docs: Document the scd_type2 option of create_changelog_view Adds the argument to the usage table, describes the three columns the SCD Type-2 view adds and why UPDATE_BEFORE rows are dropped while DELETE rows are kept, and documents how to build a long-lived SCD Type-2 table. The view only spans the changes the changelog scan can still see, so its history is bounded by snapshot retention. For history that must outlive snapshot expiration, the procedure derives each batch of versions and the user's own table stores them; the new section shows that loop with an explicit start snapshot and a MERGE that closes the previously open version. --- docs/docs/spark-procedures.md | 61 +++++++++++++++++++++++++++++++++++ 1 file changed, 61 insertions(+) diff --git a/docs/docs/spark-procedures.md b/docs/docs/spark-procedures.md index ced96eb9cd6a..43431776d905 100644 --- a/docs/docs/spark-procedures.md +++ b/docs/docs/spark-procedures.md @@ -841,6 +841,7 @@ Creates a view that contains the changes from a given table. | `net_changes` | | boolean | Whether to output net changes (see below for more information). Defaults to false. It must be false when `compute_updates` is true. | | `compute_updates` | | boolean | Whether to compute pre/post update images (see below for more information). Defaults to true if `identifer_columns` are provided; otherwise, defaults to false. | | `identifier_columns` | | array | The list of identifier columns to compute updates. If the argument `compute_updates` is set to true and `identifier_columns` are not provided, the table’s current identifier fields will be used. | +| `scd_type2` | | boolean | Whether to produce a Slowly Changing Dimensions Type-2 view (see below for more information). Defaults to false. Requires identifier columns, forces `compute_updates` to true, and must not be combined with `net_changes`. | Here is a list of commonly used Spark read options: @@ -963,6 +964,66 @@ as an `UPDATE_AFTER` image, resulting in the following pre/post update images: | 3 | Robert | UPDATE_BEFORE| | 3 | Dan | UPDATE_AFTER | +#### SCD Type-2 View + +With `scd_type2 => true` the procedure turns the changelog into a Slowly Changing Dimensions Type-2 history, so that +each row describes the period during which a version of a record was in effect. Three columns are added to the view: + +- `_valid_from`: the commit timestamp at which this version became active +- `_valid_to`: the commit timestamp at which it was superseded, or `NULL` while it is still in effect +- `_is_current`: `true` when the version is still in effect and was not deleted + +`UPDATE_BEFORE` rows are dropped, since the pre-update state of a row is an intermediate artifact rather than a version +of the record. `DELETE` rows are kept with `_is_current = false`, so that hard deletes remain visible in the history. + +```sql +CALL spark_catalog.system.create_changelog_view( + table => 'db.products', + changelog_view => 'products_history', + identifier_columns => array('product_id'), + scd_type2 => true +); +``` + +| product_id | name | _change_type | _valid_from | _valid_to | _is_current | +|------------|------------|--------------|---------------------|---------------------|-------------| +| 1 | Widget | INSERT | 2024-01-01 00:00:00 | 2024-02-01 00:00:00 | false | +| 1 | Widget Pro | UPDATE_AFTER | 2024-02-01 00:00:00 | NULL | true | + +#### Building a Long-Lived SCD Type-2 Table + +The view spans only the changes the changelog scan can still see, so its history is bounded by snapshot retention — +`history.expire.max-snapshot-age-ms` defaults to 5 days, and expiring snapshots drops the corresponding changes from +the view. For history that must outlive snapshot expiration, keep an SCD Type-2 table of your own and extend it +incrementally, using the procedure to derive each batch of versions. Run it with an explicit start snapshot, close the +version that was open in the target table, and append the new ones: + +```sql +CALL spark_catalog.system.create_changelog_view( + table => 'db.products', + changelog_view => 'products_scd2_batch', + identifier_columns => array('product_id'), + scd_type2 => true, + options => map('start-snapshot-id', '') +); + +MERGE INTO db.products_history t +USING ( + SELECT product_id, MIN(_valid_from) AS next_valid_from + FROM products_scd2_batch + GROUP BY product_id +) s +ON t.product_id = s.product_id AND t._is_current +WHEN MATCHED THEN UPDATE SET t._valid_to = s.next_valid_from, t._is_current = false; + +INSERT INTO db.products_history +SELECT product_id, name, _change_type, _valid_from, _valid_to, _is_current +FROM products_scd2_batch; +``` + +Record the end snapshot of each run and pass it as the next run's `start-snapshot-id`, so that consecutive batches +neither overlap nor leave gaps. + ## Table Statistics ### `compute_table_stats` From 3948b6741dbb85aa5c7eabe633f93f5a166c8a81 Mon Sep 17 00:00:00 2001 From: Noritaka Sekiyama Date: Sun, 6 Sep 2026 05:05:26 +0900 Subject: [PATCH 5/5] Guard generated SCD Type 2 column names and stop registering a session-wide UDF --- .../TestCreateChangelogViewProcedure.java | 27 +++++++++--- .../CreateChangelogViewProcedure.java | 43 +++++++++++-------- .../TestCreateChangelogViewProcedure.java | 27 +++++++++--- .../CreateChangelogViewProcedure.java | 43 +++++++++++-------- .../TestCreateChangelogViewProcedure.java | 27 +++++++++--- .../CreateChangelogViewProcedure.java | 43 +++++++++++-------- 6 files changed, 138 insertions(+), 72 deletions(-) diff --git a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 438e098c8e3b..14faf474161d 100644 --- a/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v3.5/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -31,7 +31,6 @@ import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.spark.SparkReadOptions; -import org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure; import org.apache.spark.sql.types.StructField; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.TestTemplate; @@ -758,9 +757,7 @@ public void testScdType2CurrentRows() { String viewName = (String) returns.get(0)[0]; List currentRows = - sql( - "SELECT id FROM %s WHERE %s = true ORDER BY id", - viewName, CreateChangelogViewProcedure.IS_CURRENT_COL); + sql("SELECT id FROM %s WHERE %s = true ORDER BY id", viewName, "_is_current"); // id=1's UPDATE_AFTER row should be current; id=2's INSERT row should also be current assertThat(currentRows).hasSize(2); @@ -845,8 +842,24 @@ public void testScdType2OutputSchema() { "_change_type", "_change_ordinal", "_commit_snapshot_id", - CreateChangelogViewProcedure.VALID_FROM_COL, - CreateChangelogViewProcedure.VALID_TO_COL, - CreateChangelogViewProcedure.IS_CURRENT_COL); + "_valid_from", + "_valid_to", + "_is_current"); + } + + @TestTemplate + public void testScdType2RejectsCollidingColumnName() { + sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName); + sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName); + sql("ALTER TABLE %s ADD COLUMN _valid_from timestamp", tableName); + sql("INSERT INTO %s VALUES (1, 'a', null)", tableName); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table already has a column named _valid_from"); } } diff --git a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index 10336b80fdb6..c916e1a1c61b 100644 --- a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -41,12 +41,13 @@ import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; -import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.api.java.UDF1; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.OrderUtils; import org.apache.spark.sql.connector.catalog.Identifier; import org.apache.spark.sql.connector.catalog.TableCatalog; import org.apache.spark.sql.connector.iceberg.catalog.ProcedureParameter; +import org.apache.spark.sql.expressions.UserDefinedFunction; import org.apache.spark.sql.expressions.Window; import org.apache.spark.sql.expressions.WindowSpec; import org.apache.spark.sql.functions; @@ -119,9 +120,9 @@ public class CreateChangelogViewProcedure extends BaseProcedure { SCD_TYPE2_PARAM, }; - public static final String VALID_FROM_COL = "_valid_from"; - public static final String VALID_TO_COL = "_valid_to"; - public static final String IS_CURRENT_COL = "_is_current"; + private static final String VALID_FROM_COL = "_valid_from"; + private static final String VALID_TO_COL = "_valid_to"; + private static final String IS_CURRENT_COL = "_is_current"; private static final StructType OUTPUT_TYPE = new StructType( @@ -212,6 +213,15 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { + // withColumn replaces a column of the same name, so a table already carrying one of the + // generated names would lose it from the view without any warning + for (String generated : new String[] {VALID_FROM_COL, VALID_TO_COL, IS_CURRENT_COL}) { + Preconditions.checkArgument( + !Arrays.asList(df.columns()).contains(generated), + "Cannot generate an SCD Type 2 view: table already has a column named %s", + generated); + } + // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) Dataset withUpdates = computeUpdateImages(identifierColumns, df); @@ -227,24 +237,23 @@ private Dataset computeScdType2(String[] identifierColumns, Dataset df snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); } - // Step 4: resolve _valid_from via a UDF - SparkSession spark = spark(); - spark - .udf() - .register( - "__iceberg_snapshot_timestamp", - (Long snapshotId) -> { - Long tsMs = snapshotTimestamps.get(snapshotId); - return tsMs != null ? new Timestamp(tsMs) : null; - }, + // Step 4: resolve _valid_from via a UDF. It is applied directly rather than registered by name, + // so concurrent calls of this procedure cannot overwrite each other's snapshot timestamps and + // no + // function of that name is left behind in the session. + UserDefinedFunction snapshotTimestamp = + functions.udf( + (UDF1) + snapshotId -> { + Long tsMs = snapshotTimestamps.get(snapshotId); + return tsMs != null ? new Timestamp(tsMs) : null; + }, DataTypes.TimestampType); String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); Dataset dfWithValidFrom = withoutUpdateBefore.withColumn( - VALID_FROM_COL, - functions.callUDF( - "__iceberg_snapshot_timestamp", withoutUpdateBefore.col(snapshotIdCol))); + VALID_FROM_COL, snapshotTimestamp.apply(withoutUpdateBefore.col(snapshotIdCol))); // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY // _change_ordinal) diff --git a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 438e098c8e3b..14faf474161d 100644 --- a/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -31,7 +31,6 @@ import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.spark.SparkReadOptions; -import org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure; import org.apache.spark.sql.types.StructField; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.TestTemplate; @@ -758,9 +757,7 @@ public void testScdType2CurrentRows() { String viewName = (String) returns.get(0)[0]; List currentRows = - sql( - "SELECT id FROM %s WHERE %s = true ORDER BY id", - viewName, CreateChangelogViewProcedure.IS_CURRENT_COL); + sql("SELECT id FROM %s WHERE %s = true ORDER BY id", viewName, "_is_current"); // id=1's UPDATE_AFTER row should be current; id=2's INSERT row should also be current assertThat(currentRows).hasSize(2); @@ -845,8 +842,24 @@ public void testScdType2OutputSchema() { "_change_type", "_change_ordinal", "_commit_snapshot_id", - CreateChangelogViewProcedure.VALID_FROM_COL, - CreateChangelogViewProcedure.VALID_TO_COL, - CreateChangelogViewProcedure.IS_CURRENT_COL); + "_valid_from", + "_valid_to", + "_is_current"); + } + + @TestTemplate + public void testScdType2RejectsCollidingColumnName() { + sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName); + sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName); + sql("ALTER TABLE %s ADD COLUMN _valid_from timestamp", tableName); + sql("INSERT INTO %s VALUES (1, 'a', null)", tableName); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table already has a column named _valid_from"); } } diff --git a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index abee78d5e695..3d22899e44d9 100644 --- a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -43,7 +43,7 @@ import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; -import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.api.java.UDF1; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.OrderUtils; import org.apache.spark.sql.connector.catalog.Identifier; @@ -51,6 +51,7 @@ import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure; import org.apache.spark.sql.connector.catalog.procedures.ProcedureParameter; import org.apache.spark.sql.connector.read.Scan; +import org.apache.spark.sql.expressions.UserDefinedFunction; import org.apache.spark.sql.expressions.Window; import org.apache.spark.sql.expressions.WindowSpec; import org.apache.spark.sql.functions; @@ -125,9 +126,9 @@ public class CreateChangelogViewProcedure extends BaseProcedure { SCD_TYPE2_PARAM, }; - public static final String VALID_FROM_COL = "_valid_from"; - public static final String VALID_TO_COL = "_valid_to"; - public static final String IS_CURRENT_COL = "_is_current"; + private static final String VALID_FROM_COL = "_valid_from"; + private static final String VALID_TO_COL = "_valid_to"; + private static final String IS_CURRENT_COL = "_is_current"; private static final StructType OUTPUT_TYPE = new StructType( @@ -218,6 +219,15 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { + // withColumn replaces a column of the same name, so a table already carrying one of the + // generated names would lose it from the view without any warning + for (String generated : new String[] {VALID_FROM_COL, VALID_TO_COL, IS_CURRENT_COL}) { + Preconditions.checkArgument( + !Arrays.asList(df.columns()).contains(generated), + "Cannot generate an SCD Type 2 view: table already has a column named %s", + generated); + } + // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) Dataset withUpdates = computeUpdateImages(identifierColumns, df); @@ -233,24 +243,23 @@ private Dataset computeScdType2(String[] identifierColumns, Dataset df snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); } - // Step 4: resolve _valid_from via a UDF - SparkSession spark = spark(); - spark - .udf() - .register( - "__iceberg_snapshot_timestamp", - (Long snapshotId) -> { - Long tsMs = snapshotTimestamps.get(snapshotId); - return tsMs != null ? new Timestamp(tsMs) : null; - }, + // Step 4: resolve _valid_from via a UDF. It is applied directly rather than registered by name, + // so concurrent calls of this procedure cannot overwrite each other's snapshot timestamps and + // no + // function of that name is left behind in the session. + UserDefinedFunction snapshotTimestamp = + functions.udf( + (UDF1) + snapshotId -> { + Long tsMs = snapshotTimestamps.get(snapshotId); + return tsMs != null ? new Timestamp(tsMs) : null; + }, DataTypes.TimestampType); String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); Dataset dfWithValidFrom = withoutUpdateBefore.withColumn( - VALID_FROM_COL, - functions.callUDF( - "__iceberg_snapshot_timestamp", withoutUpdateBefore.col(snapshotIdCol))); + VALID_FROM_COL, snapshotTimestamp.apply(withoutUpdateBefore.col(snapshotIdCol))); // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY // _change_ordinal) diff --git a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java index 438e098c8e3b..14faf474161d 100644 --- a/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java +++ b/spark/v4.1/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestCreateChangelogViewProcedure.java @@ -31,7 +31,6 @@ import org.apache.iceberg.Table; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.spark.SparkReadOptions; -import org.apache.iceberg.spark.procedures.CreateChangelogViewProcedure; import org.apache.spark.sql.types.StructField; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.TestTemplate; @@ -758,9 +757,7 @@ public void testScdType2CurrentRows() { String viewName = (String) returns.get(0)[0]; List currentRows = - sql( - "SELECT id FROM %s WHERE %s = true ORDER BY id", - viewName, CreateChangelogViewProcedure.IS_CURRENT_COL); + sql("SELECT id FROM %s WHERE %s = true ORDER BY id", viewName, "_is_current"); // id=1's UPDATE_AFTER row should be current; id=2's INSERT row should also be current assertThat(currentRows).hasSize(2); @@ -845,8 +842,24 @@ public void testScdType2OutputSchema() { "_change_type", "_change_ordinal", "_commit_snapshot_id", - CreateChangelogViewProcedure.VALID_FROM_COL, - CreateChangelogViewProcedure.VALID_TO_COL, - CreateChangelogViewProcedure.IS_CURRENT_COL); + "_valid_from", + "_valid_to", + "_is_current"); + } + + @TestTemplate + public void testScdType2RejectsCollidingColumnName() { + sql("CREATE TABLE %s (id INT NOT NULL, data STRING) USING iceberg", tableName); + sql("ALTER TABLE %s SET IDENTIFIER FIELDS id", tableName); + sql("ALTER TABLE %s ADD COLUMN _valid_from timestamp", tableName); + sql("INSERT INTO %s VALUES (1, 'a', null)", tableName); + + assertThatThrownBy( + () -> + sql( + "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)", + catalogName, tableName)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("table already has a column named _valid_from"); } } diff --git a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java index abee78d5e695..3d22899e44d9 100644 --- a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java +++ b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/procedures/CreateChangelogViewProcedure.java @@ -43,7 +43,7 @@ import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; -import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.api.java.UDF1; import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.OrderUtils; import org.apache.spark.sql.connector.catalog.Identifier; @@ -51,6 +51,7 @@ import org.apache.spark.sql.connector.catalog.procedures.BoundProcedure; import org.apache.spark.sql.connector.catalog.procedures.ProcedureParameter; import org.apache.spark.sql.connector.read.Scan; +import org.apache.spark.sql.expressions.UserDefinedFunction; import org.apache.spark.sql.expressions.Window; import org.apache.spark.sql.expressions.WindowSpec; import org.apache.spark.sql.functions; @@ -125,9 +126,9 @@ public class CreateChangelogViewProcedure extends BaseProcedure { SCD_TYPE2_PARAM, }; - public static final String VALID_FROM_COL = "_valid_from"; - public static final String VALID_TO_COL = "_valid_to"; - public static final String IS_CURRENT_COL = "_is_current"; + private static final String VALID_FROM_COL = "_valid_from"; + private static final String VALID_TO_COL = "_valid_to"; + private static final String IS_CURRENT_COL = "_is_current"; private static final StructType OUTPUT_TYPE = new StructType( @@ -218,6 +219,15 @@ private Dataset computeUpdateImages(String[] identifierColumns, Dataset computeScdType2(String[] identifierColumns, Dataset df, Table table) { + // withColumn replaces a column of the same name, so a table already carrying one of the + // generated names would lose it from the view without any warning + for (String generated : new String[] {VALID_FROM_COL, VALID_TO_COL, IS_CURRENT_COL}) { + Preconditions.checkArgument( + !Arrays.asList(df.columns()).contains(generated), + "Cannot generate an SCD Type 2 view: table already has a column named %s", + generated); + } + // Step 1: compute update images (UPDATE_BEFORE / UPDATE_AFTER) Dataset withUpdates = computeUpdateImages(identifierColumns, df); @@ -233,24 +243,23 @@ private Dataset computeScdType2(String[] identifierColumns, Dataset df snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); } - // Step 4: resolve _valid_from via a UDF - SparkSession spark = spark(); - spark - .udf() - .register( - "__iceberg_snapshot_timestamp", - (Long snapshotId) -> { - Long tsMs = snapshotTimestamps.get(snapshotId); - return tsMs != null ? new Timestamp(tsMs) : null; - }, + // Step 4: resolve _valid_from via a UDF. It is applied directly rather than registered by name, + // so concurrent calls of this procedure cannot overwrite each other's snapshot timestamps and + // no + // function of that name is left behind in the session. + UserDefinedFunction snapshotTimestamp = + functions.udf( + (UDF1) + snapshotId -> { + Long tsMs = snapshotTimestamps.get(snapshotId); + return tsMs != null ? new Timestamp(tsMs) : null; + }, DataTypes.TimestampType); String snapshotIdCol = MetadataColumns.COMMIT_SNAPSHOT_ID.name(); Dataset dfWithValidFrom = withoutUpdateBefore.withColumn( - VALID_FROM_COL, - functions.callUDF( - "__iceberg_snapshot_timestamp", withoutUpdateBefore.col(snapshotIdCol))); + VALID_FROM_COL, snapshotTimestamp.apply(withoutUpdateBefore.col(snapshotIdCol))); // Step 5: compute _valid_to = LEAD(_valid_from) OVER (PARTITION BY id_cols ORDER BY // _change_ordinal)