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` 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..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 @@ -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 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, "_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 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", + "_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 4e4a6fd252fb..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 @@ -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,16 @@ import org.apache.spark.sql.Dataset; import org.apache.spark.sql.Encoders; import org.apache.spark.sql.Row; +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; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; @@ -97,6 +106,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 +117,13 @@ public class CreateChangelogViewProcedure extends BaseProcedure { COMPUTE_UPDATES_PARAM, IDENTIFIER_COLUMNS_PARAM, NET_CHANGES, + SCD_TYPE2_PARAM, }; + 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( new StructField[] { @@ -148,6 +164,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 +177,13 @@ 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,71 @@ 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); + + // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts + String changeTypeCol = MetadataColumns.CHANGE_TYPE.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(); + for (Snapshot snapshot : table.snapshots()) { + snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); + } + + // 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, snapshotTimestamp.apply(withoutUpdateBefore.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..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 @@ -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 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, "_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 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", + "_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 73c94f778510..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 @@ -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.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; @@ -46,6 +51,10 @@ 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; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; @@ -103,6 +112,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 +123,13 @@ public class CreateChangelogViewProcedure extends BaseProcedure { COMPUTE_UPDATES_PARAM, IDENTIFIER_COLUMNS_PARAM, NET_CHANGES, + SCD_TYPE2_PARAM, }; + 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( new StructField[] { @@ -154,6 +170,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 +183,13 @@ 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,71 @@ 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); + + // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts + String changeTypeCol = MetadataColumns.CHANGE_TYPE.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(); + for (Snapshot snapshot : table.snapshots()) { + snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); + } + + // 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, snapshotTimestamp.apply(withoutUpdateBefore.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..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 @@ -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 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, "_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 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", + "_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 73c94f778510..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 @@ -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.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; @@ -46,6 +51,10 @@ 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; import org.apache.spark.sql.types.DataTypes; import org.apache.spark.sql.types.Metadata; import org.apache.spark.sql.types.StructField; @@ -103,6 +112,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 +123,13 @@ public class CreateChangelogViewProcedure extends BaseProcedure { COMPUTE_UPDATES_PARAM, IDENTIFIER_COLUMNS_PARAM, NET_CHANGES, + SCD_TYPE2_PARAM, }; + 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( new StructField[] { @@ -154,6 +170,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 +183,13 @@ 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,71 @@ 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); + + // Step 2: filter out UPDATE_BEFORE rows — they are intermediate artifacts + String changeTypeCol = MetadataColumns.CHANGE_TYPE.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(); + for (Snapshot snapshot : table.snapshots()) { + snapshotTimestamps.put(snapshot.snapshotId(), snapshot.timestampMillis()); + } + + // 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, snapshotTimestamp.apply(withoutUpdateBefore.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);