'.",
+ rawTable));
+ }
+
+ if (project == null || project.trim().isEmpty()) {
+ throw new IllegalArgumentException(
+ "GCP project must be specified via --project or as the first component of a 4-part "
+ + "--lakehouseTable identifier.");
+ }
+ if (warehouse == null || warehouse.trim().isEmpty()) {
+ throw new IllegalArgumentException(
+ "Lakehouse warehouse bucket must be specified via --warehouse or as part of a 3-part / "
+ + "4-part --lakehouseTable identifier.");
+ }
+ String normalizedWarehouse =
+ warehouse.startsWith("gs://") ? warehouse : "gs://" + warehouse.trim();
+
+ return new LakehouseTableSpec(project.trim(), normalizedWarehouse, tableId);
+ }
+
+ /**
+ * Validates that either a commit version range or a timestamp range (and not a mix of both) is
+ * configured on {@link Options}.
+ */
+ static void validateRangeOptions(Options options) {
+ @Nullable Long startVersion = options.getStartVersion();
+ @Nullable Long endVersion = options.getEndVersion();
+ @Nullable String startTimestamp = options.getStartTimestamp();
+ @Nullable String endTimestamp = options.getEndTimestamp();
+
+ boolean hasStartVersion = startVersion != null;
+ boolean hasEndVersion = endVersion != null;
+ boolean hasStartTimestamp = startTimestamp != null && !startTimestamp.trim().isEmpty();
+ boolean hasEndTimestamp = endTimestamp != null && !endTimestamp.trim().isEmpty();
+
+ if (!hasStartVersion && !hasStartTimestamp) {
+ throw new IllegalArgumentException(
+ "Either --startVersion or --startTimestamp must be provided to read Delta Lake CDC data.");
+ }
+ if (hasStartVersion && hasStartTimestamp) {
+ throw new IllegalArgumentException(
+ "Cannot set both --startVersion and --startTimestamp; specify either a version range or "
+ + "a timestamp range.");
+ }
+ if (hasEndVersion && hasEndTimestamp) {
+ throw new IllegalArgumentException(
+ "Cannot set both --endVersion and --endTimestamp; specify either a version range or "
+ + "a timestamp range.");
+ }
+ if (hasStartVersion && hasEndTimestamp) {
+ throw new IllegalArgumentException(
+ "Cannot mix --startVersion with --endTimestamp; use --endVersion instead.");
+ }
+ if (hasStartTimestamp && hasEndVersion) {
+ throw new IllegalArgumentException(
+ "Cannot mix --startTimestamp with --endVersion; use --endTimestamp instead.");
+ }
+ if (startVersion != null && endVersion != null && startVersion > endVersion) {
+ throw new IllegalArgumentException(
+ String.format(
+ "--startVersion (%d) must be less than or equal to --endVersion (%d).",
+ startVersion, endVersion));
+ }
+ }
+
+ /**
+ * Verifies that Change Data Feed is enabled on the input Delta Lake table, i.e. that the latest
+ * snapshot of the table has the table property {@code delta.enableChangeDataFeed = true}.
+ *
+ * The table is read using the Delta Kernel API with the given Hadoop configuration (for
+ * example, the GCS connector settings that the Delta Lake CDC source uses to access the table).
+ *
+ * @throws IllegalArgumentException if there is no Delta Lake table at {@code deltaTablePath}
+ * @throws IllegalStateException if {@code delta.enableChangeDataFeed = true} is not set
+ */
+ static void verifyChangeDataFeedEnabled(String deltaTablePath, Map hadoopConfig) {
+ Configuration conf = new Configuration();
+ hadoopConfig.forEach(conf::set);
+ Engine engine = DefaultEngine.create(conf);
+
+ Snapshot snapshot;
+ try {
+ snapshot = Table.forPath(engine, deltaTablePath).getLatestSnapshot(engine);
+ } catch (TableNotFoundException e) {
+ throw new IllegalArgumentException(
+ String.format(
+ "No Delta Lake table found at '%s'. Verify that it is a valid Delta Lake table.",
+ deltaTablePath),
+ e);
+ }
+
+ @Nullable String changeDataFeed =
+ snapshot.getTableProperties().get(ENABLE_CHANGE_DATA_FEED_PROPERTY);
+ if (!"true".equalsIgnoreCase(changeDataFeed)) {
+ throw new IllegalStateException(
+ String.format(
+ "Delta Lake table '%s' does not have 'delta.enableChangeDataFeed = true' enabled "
+ + "(found '%s' at version %d). Enable Change Data Feed on the table before "
+ + "running this pipeline.",
+ deltaTablePath, changeDataFeed, snapshot.getVersion()));
+ }
+ LOG.info(
+ "Verified 'delta.enableChangeDataFeed = true' on Delta Lake table '{}' (version {}).",
+ deltaTablePath,
+ snapshot.getVersion());
+ }
+}
diff --git a/examples/java/iceberg/src/test/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExampleTest.java b/examples/java/iceberg/src/test/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExampleTest.java
new file mode 100644
index 000000000000..a6cd955eb675
--- /dev/null
+++ b/examples/java/iceberg/src/test/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExampleTest.java
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+package org.apache.beam.examples.iceberg;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample.Options;
+import org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcUtils.LakehouseTableSpec;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link DeltaLakeToLakehouseCdcExample}. */
+@RunWith(JUnit4.class)
+public class DeltaLakeToLakehouseCdcExampleTest {
+
+ @Test
+ @SuppressWarnings("unchecked")
+ public void testBuildManagedConfigs() {
+ Options options = PipelineOptionsFactory.as(Options.class);
+ options.setDeltaTable("gs://my-delta-lake-bucket/delta_lake/employee_data/");
+ options.setLakehouseTable(
+ "my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table");
+ options.setEqualityColumns("employee_id");
+ options.setStartVersion(2L);
+ options.setEndVersion(4L);
+
+ LakehouseTableSpec spec = DeltaLakeToLakehouseCdcUtils.resolveLakehouseTableSpec(options);
+ Map readConfig =
+ DeltaLakeToLakehouseCdcExample.buildDeltaCdcReadConfig(options, spec.project);
+ assertEquals("gs://my-delta-lake-bucket/delta_lake/employee_data/", readConfig.get("table"));
+ assertEquals(2L, readConfig.get("start_version"));
+ assertEquals(4L, readConfig.get("end_version"));
+ assertEquals(
+ Arrays.asList("_change_type", "_commit_version"),
+ readConfig.get("include_metadata_columns"));
+ Map hadoopConfig = (Map) readConfig.get("hadoop_config");
+ assertEquals("my-gcp-project", hadoopConfig.get("fs.gs.project.id"));
+
+ Map writeConfig =
+ DeltaLakeToLakehouseCdcExample.buildLakehouseCdcWriteConfig(options, spec);
+ assertEquals("my_namespace.my_iceberg_table", writeConfig.get("table"));
+ assertEquals("merge-on-read", writeConfig.get("mode"));
+ assertEquals("_commit_version", writeConfig.get("sequence_number_column"));
+ assertEquals("_change_type", writeConfig.get("change_type_column"));
+ assertEquals(Arrays.asList("employee_id"), writeConfig.get("equality_columns"));
+ Map catalogProps = (Map) writeConfig.get("catalog_properties");
+ assertEquals("rest", catalogProps.get("type"));
+ assertEquals("gs://my-lakehouse-warehouse-bucket", catalogProps.get("warehouse"));
+ assertEquals("my-gcp-project", catalogProps.get("header.x-goog-user-project"));
+ }
+
+ @Test
+ public void testBuildManagedConfigsStreamingJavaRunnerSkipsChangeTypeColumn() {
+ Options options = PipelineOptionsFactory.as(Options.class);
+ options.setUsePortableRunner(false);
+ options.setDeltaTable("gs://my-delta-lake-bucket/delta_lake/employee_data/");
+ options.setLakehouseTable(
+ "my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table");
+ options.setEqualityColumns("employee_id");
+ options.setStartVersion(2L);
+ options.setEndVersion(4L);
+
+ LakehouseTableSpec spec = DeltaLakeToLakehouseCdcUtils.resolveLakehouseTableSpec(options);
+ Map readConfig =
+ DeltaLakeToLakehouseCdcExample.buildDeltaCdcReadConfig(options, spec.project);
+ assertEquals(
+ Collections.singletonList("_commit_version"), readConfig.get("include_metadata_columns"));
+
+ Map writeConfig =
+ DeltaLakeToLakehouseCdcExample.buildLakehouseCdcWriteConfig(options, spec);
+ assertEquals("_commit_version", writeConfig.get("sequence_number_column"));
+ assertFalse(writeConfig.containsKey("change_type_column"));
+ assertFalse(writeConfig.containsKey("change_type_map"));
+ }
+}
diff --git a/examples/java/iceberg/src/test/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcUtilsTest.java b/examples/java/iceberg/src/test/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcUtilsTest.java
new file mode 100644
index 000000000000..02b3042e7c14
--- /dev/null
+++ b/examples/java/iceberg/src/test/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcUtilsTest.java
@@ -0,0 +1,229 @@
+/*
+ * 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.
+ */
+package org.apache.beam.examples.iceberg;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+import io.delta.kernel.Operation;
+import io.delta.kernel.Table;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.utils.CloseableIterable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Map;
+import org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample.Options;
+import org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcUtils.LakehouseTableSpec;
+import org.apache.beam.sdk.io.delta.DeltaWriteTestUtils;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Unit tests for {@link DeltaLakeToLakehouseCdcUtils}. */
+@RunWith(JUnit4.class)
+public class DeltaLakeToLakehouseCdcUtilsTest {
+
+ private static final String ENGINE_INFO = "DeltaLakeToLakehouseCdcUtilsTest";
+
+ private static final String ENABLE_CDF_PROPERTY = "delta.enableChangeDataFeed";
+
+ private static final StructType DELTA_SCHEMA =
+ new StructType().add("employee_id", LongType.LONG).add("name", StringType.STRING);
+
+ private static final Schema BEAM_SCHEMA =
+ Schema.builder().addInt64Field("employee_id").addStringField("name").build();
+
+ @Rule public final TemporaryFolder tmpFolder = new TemporaryFolder();
+
+ private final Engine engine = DefaultEngine.create(new Configuration());
+
+ private static Row employee(long id, String name) {
+ return Row.withSchema(BEAM_SCHEMA).addValues(id, name).build();
+ }
+
+ /**
+ * Appends rows to a Delta Lake table as commit {@code version} using {@link DeltaWriteTestUtils}.
+ * Version 0 creates the table with Change Data Feed enabled.
+ */
+ private void appendRows(String tablePath, long version, Row... rows) throws Exception {
+ DeltaWriteTestUtils.writeAppendCommit(
+ engine, tablePath, version, System.currentTimeMillis(), DELTA_SCHEMA, Arrays.asList(rows));
+ }
+
+ /** Creates an empty Delta Lake table (commit version 0) with the given table properties. */
+ private String createDeltaTable(Map tableProperties) throws IOException {
+ String tablePath = tmpFolder.newFolder().getAbsolutePath();
+ Table.forPath(engine, tablePath)
+ .createTransactionBuilder(engine, ENGINE_INFO, Operation.CREATE_TABLE)
+ .withSchema(engine, DELTA_SCHEMA)
+ .withTableProperties(engine, tableProperties)
+ .build(engine)
+ .commit(engine, CloseableIterable.emptyIterable());
+ return tablePath;
+ }
+
+ /**
+ * Commits a new table version that updates the given table properties, like {@code ALTER TABLE
+ * ... SET TBLPROPERTIES}.
+ */
+ private void setTableProperties(String tablePath, Map tableProperties) {
+ Table.forPath(engine, tablePath)
+ .createTransactionBuilder(engine, ENGINE_INFO, Operation.MANUAL_UPDATE)
+ .withTableProperties(engine, tableProperties)
+ .build(engine)
+ .commit(engine, CloseableIterable.emptyIterable());
+ }
+
+ /** Runs the Change Data Feed check on a local table, which needs no Hadoop configuration. */
+ private static void verifyChangeDataFeedEnabled(String tablePath) {
+ DeltaLakeToLakehouseCdcUtils.verifyChangeDataFeedEnabled(tablePath, Collections.emptyMap());
+ }
+
+ @Test
+ public void testVerifyChangeDataFeedEnabledPassesWhenEnabled() throws Exception {
+ String tablePath = tmpFolder.newFolder().getAbsolutePath();
+ appendRows(tablePath, 0L, employee(1L, "Alice"), employee(2L, "Bob"));
+ appendRows(tablePath, 1L, employee(3L, "Carol"));
+
+ verifyChangeDataFeedEnabled(tablePath);
+ }
+
+ @Test
+ public void testVerifyChangeDataFeedEnabledPassesWhenEnabledInLatestSnapshot() throws Exception {
+ // Version 0: the table is created without Change Data Feed.
+ String tablePath = createDeltaTable(Collections.emptyMap());
+ // Version 1: ALTER TABLE ... SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true')
+ setTableProperties(tablePath, ImmutableMap.of(ENABLE_CDF_PROPERTY, "true"));
+ // Version 2: a data write, which doesn't change the table metadata.
+ appendRows(tablePath, 2L, employee(1L, "Alice"));
+
+ verifyChangeDataFeedEnabled(tablePath);
+ }
+
+ @Test
+ public void testVerifyChangeDataFeedEnabledFailsWhenFalse() throws Exception {
+ String tablePath = createDeltaTable(ImmutableMap.of(ENABLE_CDF_PROPERTY, "false"));
+
+ IllegalStateException ex =
+ assertThrows(IllegalStateException.class, () -> verifyChangeDataFeedEnabled(tablePath));
+ assertThat(ex.getMessage(), containsString("delta.enableChangeDataFeed = true"));
+ }
+
+ @Test
+ public void testVerifyChangeDataFeedEnabledFailsWhenMissingFromTableProperties()
+ throws Exception {
+ String tablePath = createDeltaTable(Collections.emptyMap());
+
+ IllegalStateException ex =
+ assertThrows(IllegalStateException.class, () -> verifyChangeDataFeedEnabled(tablePath));
+ assertThat(ex.getMessage(), containsString("delta.enableChangeDataFeed = true"));
+ }
+
+ @Test
+ public void testVerifyChangeDataFeedEnabledFailsWhenDisabledInLatestSnapshot() throws Exception {
+ // Version 0: the table is created with Change Data Feed enabled.
+ String tablePath = tmpFolder.newFolder().getAbsolutePath();
+ appendRows(tablePath, 0L, employee(1L, "Alice"));
+ // Version 1: ALTER TABLE ... SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'false')
+ setTableProperties(tablePath, ImmutableMap.of(ENABLE_CDF_PROPERTY, "false"));
+
+ IllegalStateException ex =
+ assertThrows(IllegalStateException.class, () -> verifyChangeDataFeedEnabled(tablePath));
+ assertThat(ex.getMessage(), containsString("delta.enableChangeDataFeed = true"));
+ }
+
+ @Test
+ public void testVerifyChangeDataFeedEnabledFailsWhenNotADeltaTable() throws Exception {
+ String emptyDir = tmpFolder.newFolder("not_a_delta_table").getAbsolutePath();
+
+ IllegalArgumentException ex =
+ assertThrows(IllegalArgumentException.class, () -> verifyChangeDataFeedEnabled(emptyDir));
+ assertThat(ex.getMessage(), containsString("No Delta Lake table found"));
+ }
+
+ @Test
+ public void testResolveLakehouseTableSpecFourPart() {
+ Options options = PipelineOptionsFactory.as(Options.class);
+ options.setLakehouseTable(
+ "my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table");
+
+ LakehouseTableSpec spec = DeltaLakeToLakehouseCdcUtils.resolveLakehouseTableSpec(options);
+ assertEquals("my-gcp-project", spec.project);
+ assertEquals("gs://my-lakehouse-warehouse-bucket", spec.warehouse);
+ assertEquals("my_namespace.my_iceberg_table", spec.tableId);
+ }
+
+ @Test
+ public void testResolveLakehouseTableSpecTwoPartWithExplicitWarehouseAndProject() {
+ Options options = PipelineOptionsFactory.as(Options.class);
+ options.setProject("my-gcp-project");
+ options.setWarehouse("gs://my-lakehouse-warehouse-bucket");
+ options.setLakehouseTable("my_namespace.my_iceberg_table");
+
+ LakehouseTableSpec spec = DeltaLakeToLakehouseCdcUtils.resolveLakehouseTableSpec(options);
+ assertEquals("my-gcp-project", spec.project);
+ assertEquals("gs://my-lakehouse-warehouse-bucket", spec.warehouse);
+ assertEquals("my_namespace.my_iceberg_table", spec.tableId);
+ }
+
+ @Test
+ public void testValidateRangeOptionsVersionAndTimestampRanges() {
+ Options versionOpts = PipelineOptionsFactory.as(Options.class);
+ versionOpts.setStartVersion(1L);
+ versionOpts.setEndVersion(3L);
+ DeltaLakeToLakehouseCdcUtils.validateRangeOptions(versionOpts);
+
+ Options tsOpts = PipelineOptionsFactory.as(Options.class);
+ tsOpts.setStartTimestamp("2026-10-01T00:00:00Z");
+ tsOpts.setEndTimestamp("2026-10-02T00:00:00Z");
+ DeltaLakeToLakehouseCdcUtils.validateRangeOptions(tsOpts);
+
+ Options missingOpts = PipelineOptionsFactory.as(Options.class);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> DeltaLakeToLakehouseCdcUtils.validateRangeOptions(missingOpts));
+
+ Options mixedOpts = PipelineOptionsFactory.as(Options.class);
+ mixedOpts.setStartVersion(1L);
+ mixedOpts.setStartTimestamp("2026-10-01T00:00:00Z");
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> DeltaLakeToLakehouseCdcUtils.validateRangeOptions(mixedOpts));
+
+ Options invertedOpts = PipelineOptionsFactory.as(Options.class);
+ invertedOpts.setStartVersion(5L);
+ invertedOpts.setEndVersion(2L);
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> DeltaLakeToLakehouseCdcUtils.validateRangeOptions(invertedOpts));
+ }
+}