Add YAML tests for writing to Iceberg CDC sink

.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json

@@ -1,4 +1,4 @@
 {
   "comment": "Modify this file in a trivial way to cause this test suite to run",
-  "revision": 23
+  "revision": 1
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java

@@ -157,7 +157,6 @@ interface Clock extends LongSupplier, Serializable {}
     this.pendingSpec = StateSpecs.bag(CommitDeltas.windowedCommitCoder());
   }
 
-  @RequiresStableInput
   @ProcessElement
   public void process(
       @Element KV<String, WindowedCommit> element,
@@ -222,7 +221,6 @@ private void checkRecoveredTokenNotBatchEnd(String dest, long recoveredMs) {
             + "continuation.");
   }
 
-  @RequiresStableInput
   @OnTimer("commit")
   public void onCommit(
       OnTimerContext c,

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java

@@ -81,10 +81,11 @@
  *
  * <p>The sink commits one new snapshot to the destination table per commit window, in ascending
  * window order, with an idempotency token written to each snapshot's summary. A retried or
- * restarted commit finds the token and skips already-committed windows, making the commit
- * effectively-once on runners that honor {@code @RequiresStableInput}. Records whose grouped pane
- * fires late (watermark is past their commit window's end) are diverted to the DLQ output,
- * accessible with {@link IcebergWriteResult#getDeadLetterRows()}.
+ * restarted commit finds the token and skips already-committed windows, so each window commits
+ * once. The commit reads straight from a {@code GroupByKey}, so on runners that persist shuffle
+ * output, like Dataflow, a retried commit sees the same files. Records whose grouped pane fires
+ * late (watermark is past their commit window's end) are diverted to the DLQ output, accessible
+ * with {@link IcebergWriteResult#getDeadLetterRows()}.
  *
  * <p>Each dead-lettered record nests the data row under {@code record}, beside {@code change_type},
  * {@code sequence_number}, and {@code destination}.

sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java

@@ -856,8 +856,7 @@ public void rowDeltaUsedWhenDeleteFilesPresentAppendWhenNot() {
    * An injected commit failure must carry operator-triage context and leave NO snapshot (strict
    * ascending order halts W1 behind the failing W0); a second run re-feeds the same inputs and
    * commits both windows, proving the failure left the files intact. (The DirectRunner fails fast
-   * rather than retrying {@code @RequiresStableInput} bundles, so the in-run retry is an IT
-   * concern.)
+   * rather than retrying bundles, so the in-run retry is an IT concern.)
    */
   @Test
   public void commitFailureCarriesTriageContextAndRetryRunCommitsInOrder() {

sdks/python/apache_beam/coders/row_coder_test.py

@@ -282,6 +282,25 @@ def test_row_coder_negative_varint(self):
     for test_case in test_cases:
       self.assertEqual(test_case, coder.decode(coder.encode(test_case)))
 
+  def test_row_coder_underscore_field_name(self):
+    # NamedTuple rejects a leading underscore, as in the CDC metadata column
+    # _change_type.
+    schema = schema_pb2.Schema(
+        id="underscore_field_name",
+        fields=[
+            schema_pb2.Field(
+                name="id",
+                type=schema_pb2.FieldType(atomic_type=schema_pb2.INT64)),
+            schema_pb2.Field(
+                name="_change_type",
+                type=schema_pb2.FieldType(atomic_type=schema_pb2.STRING)),
+        ])
+    coder = RowCoder(schema)
+    row = named_tuple_from_schema(schema)(1, "INSERT")
+    self.assertEqual(row, coder.decode(coder.encode(row)))
+    self.assertEqual(row, pickler.loads(pickler.dumps(row)))
+    self.assertEqual({"id": 1, "_change_type": "INSERT"}, row._asdict())
+
   @unittest.skip(
       "https://github.com/apache/beam/issues/19696 - Overflow behavior in "
       "VarIntCoder is currently inconsistent")

sdks/python/apache_beam/typehints/schemas.py

@@ -71,6 +71,7 @@
 # pytype: skip-file
 
 # ruff: noqa: UP006
+import dataclasses
 import datetime
 import decimal
 import logging
@@ -666,7 +667,16 @@ def named_tuple_from_schema(self, schema: schema_pb2.Schema) -> type:
     if schema.id in self.schema_registry.by_id:
       user_type = self.schema_registry.by_id[schema.id][0]
     else:
-      user_type = NamedTuple(type_name, subfields)
+      if any(name.startswith('_') for name, _ in subfields):
+        # NamedTuple rejects field names with a leading underscore, like the
+        # CDC metadata column _change_type.
+        user_type = dataclasses.make_dataclass(
+            type_name,
+            subfields,
+            namespace={'_asdict': _dataclass_asdict},
+            frozen=True)
+      else:
+        user_type = NamedTuple(type_name, subfields)
 
       # Define a reduce function, otherwise these types can't be pickled
       # (See BEAM-9574)
@@ -683,9 +693,18 @@ def named_tuple_from_schema(self, schema: schema_pb2.Schema) -> type:
     return user_type
 
 
+def _dataclass_asdict(self):
+  # Matches NamedTuple._asdict, so code written for NamedTuple rows still works.
+  return {f.name: getattr(self, f.name) for f in dataclasses.fields(self)}
+
+
 def _named_tuple_reduce_method(serialized_schema):
   def __reduce__(self):
-    return _hydrate_namedtuple_instance, (serialized_schema, tuple(self))
+    if dataclasses.is_dataclass(self):
+      values = tuple(getattr(self, f.name) for f in dataclasses.fields(self))
+    else:
+      values = tuple(self)
+    return _hydrate_namedtuple_instance, (serialized_schema, values)
 
   return __reduce__
 

sdks/python/apache_beam/yaml/extended_tests/e2e/delta_lake_cdc_to_iceberg.yaml

@@ -0,0 +1,94 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+fixtures:
+  - name: DELTA_TABLE
+    type: "apache_beam.yaml.integration_tests.temp_delta_cdc_table_with_updates"
+  - name: TEMP_DIR
+    type: "tempfile.TemporaryDirectory"
+
+pipelines:
+  # Applies the Delta Lake table's changes in two batch syncs: version 0 (three
+  # new orders), then versions 1 to 3 (an update, a delete and a new order). The
+  # second sync changes rows that the first one committed.
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromDeltaCDC
+          config:
+            table: "{DELTA_TABLE}"
+            start_version: 0
+            end_version: 0
+            include_metadata_columns: ["_change_type", "_commit_version"]
+        - type: WriteToIceberg
+          config:
+            table: "shop.orders"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+            mode: "merge-on-read"
+            equality_columns: ["id"]
+            change_type_column: "_change_type"
+            change_type_map:
+              insert: "INSERT"
+              update_preimage: "UPDATE_BEFORE"
+              update_postimage: "UPDATE_AFTER"
+              delete: "DELETE"
+            sequence_number_column: "_commit_version"
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromDeltaCDC
+          config:
+            table: "{DELTA_TABLE}"
+            start_version: 1
+            include_metadata_columns: ["_change_type", "_commit_version"]
+        - type: WriteToIceberg
+          config:
+            table: "shop.orders"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+            mode: "merge-on-read"
+            equality_columns: ["id"]
+            change_type_column: "_change_type"
+            change_type_map:
+              insert: "INSERT"
+              update_preimage: "UPDATE_BEFORE"
+              update_postimage: "UPDATE_AFTER"
+              delete: "DELETE"
+            sequence_number_column: "_commit_version"
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromIceberg
+          config:
+            table: "shop.orders"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+        - type: AssertEqual
+          config:
+            elements:
+              - {id: 1, item: "apple", qty: 4}
+              - {id: 3, item: "plum", qty: 2}
+              - {id: 4, item: "kiwi", qty: 1}

sdks/python/apache_beam/yaml/extended_tests/e2e/iceberg_cdc_to_iceberg.yaml

@@ -0,0 +1,156 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+fixtures:
+  - name: TEMP_DIR
+    type: "tempfile.TemporaryDirectory"
+
+pipelines:
+  # Gives shop.orders two snapshots: three new orders, then an update, a delete
+  # and a new order. Each row carries its change type (op) and sequence number
+  # (seq).
+  - pipeline:
+      type: chain
+      transforms:
+        - type: Create
+          config:
+            elements:
+              - {id: 1, item: "apple", qty: 3, op: "INSERT", seq: 1}
+              - {id: 2, item: "pear", qty: 5, op: "INSERT", seq: 1}
+              - {id: 3, item: "plum", qty: 2, op: "INSERT", seq: 1}
+        - type: WriteToIceberg
+          config:
+            table: "shop.orders"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+            mode: "merge-on-read"
+            equality_columns: ["id"]
+            change_type_column: "op"
+            sequence_number_column: "seq"
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: Create
+          config:
+            elements:
+              - {id: 1, item: "apple", qty: 3, op: "UPDATE_BEFORE", seq: 2}
+              - {id: 1, item: "apple", qty: 4, op: "UPDATE_AFTER", seq: 2}
+              - {id: 2, item: "pear", qty: 5, op: "DELETE", seq: 2}
+              - {id: 4, item: "kiwi", qty: 1, op: "INSERT", seq: 2}
+        - type: WriteToIceberg
+          config:
+            table: "shop.orders"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+            mode: "merge-on-read"
+            equality_columns: ["id"]
+            change_type_column: "op"
+            sequence_number_column: "seq"
+
+  # Batch: replays every change in shop.orders into shop.orders_mirror.
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromIcebergCDC
+          config:
+            table: "shop.orders"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+            include_metadata_columns:
+              ["_change_type", "_commit_snapshot_sequence_number"]
+        - type: WriteToIceberg
+          config:
+            table: "shop.orders_mirror"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+            mode: "merge-on-read"
+            equality_columns: ["id"]
+            change_type_column: "_change_type"
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromIceberg
+          config:
+            table: "shop.orders_mirror"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+        - type: AssertEqual
+          config:
+            elements:
+              - {id: 1, item: "apple", qty: 4}
+              - {id: 3, item: "plum", qty: 2}
+              - {id: 4, item: "kiwi", qty: 1}
+
+  # Streaming: replays the same changes into shop.orders_streaming_mirror. The
+  # far-future to_timestamp stops the stream at the latest snapshot. One-second
+  # commit windows put the two snapshots in separate commits, so the second
+  # commit updates and deletes rows that the first one wrote.
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromIcebergCDC
+          config:
+            table: "shop.orders"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+            streaming: true
+            starting_strategy: "earliest"
+            to_timestamp: 4102444800000
+            include_metadata_columns:
+              ["_change_type", "_commit_snapshot_sequence_number"]
+        - type: WriteToIceberg
+          config:
+            table: "shop.orders_streaming_mirror"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+            mode: "merge-on-read"
+            equality_columns: ["id"]
+            change_type_column: "_change_type"
+            triggering_frequency_seconds: 1
+
+  - pipeline:
+      type: chain
+      transforms:
+        - type: ReadFromIceberg
+          config:
+            table: "shop.orders_streaming_mirror"
+            catalog_name: "hadoop_catalog"
+            catalog_properties:
+              type: "hadoop"
+              warehouse: "{TEMP_DIR}"
+        - type: AssertEqual
+          config:
+            elements:
+              - {id: 1, item: "apple", qty: 4}
+              - {id: 3, item: "plum", qty: 2}
+              - {id: 4, item: "kiwi", qty: 1}