(beam) branch master updated: test for column default values (#39739)

[ahmedabualsaud]

This is an automated email from the ASF dual-hosted git repository.

ahmedabu98 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/master by this push:
     new fde5698dc99 test for column default values (#39739)
fde5698dc99 is described below

commit fde5698dc99100f48d5df03dddf2aab8e35bfe14
Author: Ahmed Abualsaud <65791736+ahmedab...@users.noreply.github.com>
AuthorDate: Mon Aug 17 16:56:20 2026 -0400

    test for column default values (#39739)
---
 .../cdc/IncrementalChangelogSourceTest.java        | 86 ++++++++++++++++++++++
 1 file changed, 86 insertions(+)

diff --git 
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/IncrementalChangelogSourceTest.java
 
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/IncrementalChangelogSourceTest.java
index 14e7208dad0..27a65a0be8f 100644
--- 
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/IncrementalChangelogSourceTest.java
+++ 
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/IncrementalChangelogSourceTest.java
@@ -58,6 +58,7 @@ import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.data.Record;
+import org.apache.iceberg.expressions.Literal;
 import org.apache.iceberg.types.Types;
 import org.joda.time.Instant;
 import org.junit.ClassRule;
@@ -122,6 +123,76 @@ public class IncrementalChangelogSourceTest {
     pipeline.run().waitUntilFinish();
   }
 
+  @Test
+  public void readsInitialDefaultFromOldFile() throws Exception {
+    TableIdentifier tableId = tableId();
+    Table table = warehouse.createTable(tableId, CDC_SCHEMA, null, 
tablePropertiesV3());
+    commitAppend(table, "before-schema-evolution.parquet", records(1L, "one"));
+
+    table
+        .updateSchema()
+        .addColumn(
+            "category", Types.StringType.get(), "Record category", 
Literal.of("default_category"))
+        .commit();
+    table.refresh();
+
+    IcebergScanConfig scanConfig =
+        baseConfigBuilder(table, tableId)
+            .setToSnapshot(table.currentSnapshot().snapshotId())
+            .build();
+    Schema outputSchema = 
IcebergUtils.icebergSchemaToBeamSchema(table.schema());
+
+    PCollection<Row> rows = pipeline.apply(new 
IncrementalChangelogSource(scanConfig));
+
+    assertEquals(outputSchema, rows.getSchema());
+    PAssert.that(rows)
+        .containsInAnyOrder(
+            Row.withSchema(outputSchema).addValues(1L, "one", 
"default_category").build());
+
+    pipeline.run().waitUntilFinish();
+  }
+
+  @Test
+  public void 
overwriteUsesInitialDefaultForOldFileAndExplicitValueForNewFile() throws 
Exception {
+    TableIdentifier tableId = tableId();
+    Table table = warehouse.createTable(tableId, CDC_SCHEMA, null, 
tablePropertiesV3());
+    DataFile oldFile =
+        commitAppend(table, "before-schema-evolution.parquet", records(1L, 
"before"));
+
+    table
+        .updateSchema()
+        .addColumn(
+            "category", Types.StringType.get(), "Record category", 
Literal.of("default_category"))
+        .commit();
+    table.refresh();
+
+    Record replacement =
+        TestFixtures.createRecord(
+            table.schema(),
+            ImmutableMap.of(
+                "id", 1L,
+                "data", "after",
+                "category", "explicit_category"));
+    commitOverwrite(table, "after-schema-evolution.parquet", oldFile, 
replacement);
+
+    IcebergScanConfig scanConfig =
+        baseConfigBuilder(table, tableId)
+            .setFromSnapshotInclusive(table.currentSnapshot().snapshotId())
+            .setToSnapshot(table.currentSnapshot().snapshotId())
+            .build();
+
+    PCollection<String> changes =
+        pipeline
+            .apply(new IncrementalChangelogSource(scanConfig))
+            .apply("Format Defaulted Changes", ParDo.of(new 
FormatDefaultedChange()));
+
+    PAssert.that(changes)
+        .containsInAnyOrder(
+            "UPDATE_BEFORE:1:before:default_category", 
"UPDATE_AFTER:1:after:explicit_category");
+
+    pipeline.run().waitUntilFinish();
+  }
+
   @Test
   public void metadataColumnsAreAppendedToProjectedRecord() throws Exception {
     TableIdentifier tableId = tableId();
@@ -511,4 +582,19 @@ public class IncrementalChangelogSourceTest {
           valueKind.name() + ":" + row.getInt64("id") + ":" + 
row.getString("data"));
     }
   }
+
+  private static final class FormatDefaultedChange extends DoFn<Row, String> {
+    @ProcessElement
+    public void process(
+        @Element Row row, ValueKind valueKind, OutputReceiver<String> 
outputReceiver) {
+      outputReceiver.output(
+          valueKind.name()
+              + ":"
+              + row.getInt64("id")
+              + ":"
+              + row.getString("data")
+              + ":"
+              + row.getString("category"));
+    }
+  }
 }