diff --git a/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json b/.github/trigger_files/beam_PostCommit_Yaml_Xlang_Direct.json index 2b9b913580a2..a975cd1cd104 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 69bf48aa431d..35d94e4e1444 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 @@ interface Clock extends LongSupplier, Serializable {} this.pendingSpec = StateSpecs.bag(CommitDeltas.windowedCommitCoder()); } - @RequiresStableInput @ProcessElement public void process( @Element KV element, @@ -222,7 +221,6 @@ private void checkRecoveredTokenNotBatchEnd(String dest, long recoveredMs) { + "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 e32ad2f8d5ab..44ee39a46ba1 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 @@ * *

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()}. * *

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 95dd70de3282..76bfd1575e56 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 @@ -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() { diff --git a/sdks/python/apache_beam/coders/row_coder_test.py b/sdks/python/apache_beam/coders/row_coder_test.py index bd1307ff992e..9bc6b23356b8 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 @@ 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") diff --git a/sdks/python/apache_beam/typehints/schemas.py b/sdks/python/apache_beam/typehints/schemas.py index 71bd540c522a..9caccc762f71 100644 --- a/sdks/python/apache_beam/typehints/schemas.py +++ b/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__ 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 000000000000..642ef4c1eccc --- /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 000000000000..1e916788bbb2 --- /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 becfaa793032..107d55df19b0 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 +@contextlib.contextmanager +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 5176907d2fc9..9dc3eff38b1e 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'