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 35ed6c70d80 Add YAML tests for writing to Iceberg CDC sink (#40542)
35ed6c70d80 is described below
commit 35ed6c70d80008f5560f3baeff7c45e192830487
Author: Ahmed Abualsaud <[email protected]>
AuthorDate: Wed Oct 14 14:26:06 2026 -0700
Add YAML tests for writing to Iceberg CDC sink (#40542)
---
.../beam_PostCommit_Yaml_Xlang_Direct.json | 2 +-
.../sdk/io/iceberg/cdc/sink/OrderedCommitFn.java | 2 -
.../beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java | 9 +-
.../sdk/io/iceberg/cdc/sink/CommitDeltasTest.java | 3 +-
sdks/python/apache_beam/coders/row_coder_test.py | 19 +++
sdks/python/apache_beam/typehints/schemas.py | 23 ++-
.../e2e/delta_lake_cdc_to_iceberg.yaml | 94 +++++++++++++
.../extended_tests/e2e/iceberg_cdc_to_iceberg.yaml | 156 +++++++++++++++++++++
sdks/python/apache_beam/yaml/integration_tests.py | 34 +++++
sdks/python/apache_beam/yaml/standard_io.yaml | 1 +
10 files changed, 332 insertions(+), 11 deletions(-)
diff --git a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json
b/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json
index 2b9b913580a..a975cd1cd10 100644
--- a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json
+++ b/.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
}
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java
index 970910a9120..212e43c76df 100644
---
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/OrderedCommitFn.java
@@ -157,7 +157,6 @@ class OrderedCommitFn extends DoFn<KV<String,
WindowedCommit>, KV<String, Snapsh
this.pendingSpec = StateSpecs.bag(CommitDeltas.windowedCommitCoder());
}
- @RequiresStableInput
@ProcessElement
public void process(
@Element KV<String, WindowedCommit> element,
@@ -222,7 +221,6 @@ class OrderedCommitFn extends DoFn<KV<String,
WindowedCommit>, KV<String, Snapsh
+ "continuation.");
}
- @RequiresStableInput
@OnTimer("commit")
public void onCommit(
OnTimerContext c,
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java
index a9ca1db6bc2..c6ab40108ca 100644
---
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/cdc/sink/WriteCdcRows.java
@@ -81,10 +81,11 @@ import org.slf4j.LoggerFactory;
*
* <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}.
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java
index 38b38ca1b4d..892f1985af5 100644
---
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/cdc/sink/CommitDeltasTest.java
@@ -870,8 +870,7 @@ public class CommitDeltasTest {
* 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() {
diff --git a/sdks/python/apache_beam/coders/row_coder_test.py
b/sdks/python/apache_beam/coders/row_coder_test.py
index bd1307ff992..9bc6b23356b 100644
--- a/sdks/python/apache_beam/coders/row_coder_test.py
+++ b/sdks/python/apache_beam/coders/row_coder_test.py
@@ -282,6 +282,25 @@ class RowCoderTest(unittest.TestCase):
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")
diff --git a/sdks/python/apache_beam/typehints/schemas.py
b/sdks/python/apache_beam/typehints/schemas.py
index 71bd540c522..9caccc762f7 100644
--- a/sdks/python/apache_beam/typehints/schemas.py
+++ b/sdks/python/apache_beam/typehints/schemas.py
@@ -71,6 +71,7 @@ any backwards-compatibility guarantee.
# pytype: skip-file
# ruff: noqa: UP006
+import dataclasses
import datetime
import decimal
import logging
@@ -666,7 +667,16 @@ class SchemaTranslation(object):
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 @@ class SchemaTranslation(object):
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__
diff --git
a/sdks/python/apache_beam/yaml/extended_tests/e2e/delta_lake_cdc_to_iceberg.yaml
b/sdks/python/apache_beam/yaml/extended_tests/e2e/delta_lake_cdc_to_iceberg.yaml
new file mode 100644
index 00000000000..642ef4c1ecc
--- /dev/null
+++
b/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}
diff --git
a/sdks/python/apache_beam/yaml/extended_tests/e2e/iceberg_cdc_to_iceberg.yaml
b/sdks/python/apache_beam/yaml/extended_tests/e2e/iceberg_cdc_to_iceberg.yaml
new file mode 100644
index 00000000000..1e916788bbb
--- /dev/null
+++
b/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}
diff --git a/sdks/python/apache_beam/yaml/integration_tests.py
b/sdks/python/apache_beam/yaml/integration_tests.py
index becfaa79303..107d55df19b 100644
--- a/sdks/python/apache_beam/yaml/integration_tests.py
+++ b/sdks/python/apache_beam/yaml/integration_tests.py
@@ -1279,6 +1279,40 @@ def temp_delta_cdc_table():
yield temp_dir
[email protected]
+def temp_delta_cdc_table_with_updates():
+ try:
+ from deltalake import DeltaTable
+ from deltalake import write_deltalake
+ except ImportError as exn:
+ raise unittest.SkipTest('deltalake is not installed') from exn
+
+ # The CDC sink needs a non-null key.
+ schema = pa.schema([
+ pa.field("id", pa.int64(), nullable=False),
+ pa.field("item", pa.string()),
+ pa.field("qty", pa.int64()),
+ ])
+
+ def orders(*rows):
+ return pa.Table.from_pylist([dict(zip(schema.names, row)) for row in rows],
+ schema=schema)
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ # Version 0: three new orders.
+ write_deltalake(
+ temp_dir,
+ orders((1, "apple", 3), (2, "pear", 5), (3, "plum", 2)),
+ configuration={"delta.enableChangeDataFeed": "true"})
+ # Versions 1 to 3: update order 1, delete order 2, add order 4.
+ table = DeltaTable(temp_dir)
+ table.update(updates={"qty": "4"}, predicate="id = 1")
+ table.delete("id = 2")
+ write_deltalake(temp_dir, orders((4, "kiwi", 1)), mode="append")
+
+ yield temp_dir
+
+
def replace_recursive(spec, vars):
"""Recursively replaces string placeholders in a spec with values from vars.
diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml
b/sdks/python/apache_beam/yaml/standard_io.yaml
index 5176907d2fc..9dc3eff38b1 100644
--- a/sdks/python/apache_beam/yaml/standard_io.yaml
+++ b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -671,6 +671,7 @@
filter: 'filter'
from_snapshot: 'from_snapshot'
from_timestamp: 'from_timestamp'
+ include_metadata_columns: 'include_metadata_columns'
keep: 'keep'
poll_interval_seconds: 'poll_interval_seconds'
starting_strategy: 'starting_strategy'
