Spark: Add scd_type2 option to create_changelog_view procedure

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<string>       | 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', '<last processed 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`

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;
@@ -691,4 +692,174 @@ 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<Object[]> returns =
+        sql(
+            "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)",
+            catalogName, tableName);
+
+    String viewName = (String) returns.get(0)[0];
+    List<Object[]> 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<Object[]> returns =
+        sql(
+            "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)",
+            catalogName, tableName);
+
+    String viewName = (String) returns.get(0)[0];
+    List<Object[]> currentRows =
+        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);
+    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<Object[]> returns =
+        sql(
+            "CALL %s.system.create_changelog_view(table => '%s', scd_type2 => true)",
+            catalogName, tableName);
+
+    String viewName = (String) returns.get(0)[0];
+    List<Object[]> 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<Object[]> 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",
+            "_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");
+  }
 }