This is an automated email from the ASF dual-hosted git repository.
chamikaramj 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 0c47ade8d29 Adds an end-to-end example for migrating change data
(CDC)... (#40387)
0c47ade8d29 is described below
commit 0c47ade8d29d63a17b2343e2f2fb189c26490f9d
Author: Chamikara Jayalath <[email protected]>
AuthorDate: Mon Oct 12 14:17:23 2026 -0700
Adds an end-to-end example for migrating change data (CDC)... (#40387)
* Adds an end-to-end example for migrating change data (CDC) from Delta
Lake to GCP Lakehouse
* updates
* updates
* Address reviewer comments
---
examples/java/iceberg/build.gradle | 31 ++
.../iceberg/DeltaLakeToLakehouseCdcExample.java | 448 +++++++++++++++++++++
.../iceberg/DeltaLakeToLakehouseCdcUtils.java | 197 +++++++++
.../DeltaLakeToLakehouseCdcExampleTest.java | 96 +++++
.../iceberg/DeltaLakeToLakehouseCdcUtilsTest.java | 229 +++++++++++
5 files changed, 1001 insertions(+)
diff --git a/examples/java/iceberg/build.gradle
b/examples/java/iceberg/build.gradle
index 2cff15c5fb0..7186ff9c703 100644
--- a/examples/java/iceberg/build.gradle
+++ b/examples/java/iceberg/build.gradle
@@ -39,17 +39,33 @@ dependencies {
implementation
enforcedPlatform(library.java.google_cloud_platform_libraries_bom)
runtimeOnly project(":sdks:java:io:iceberg")
runtimeOnly project(":sdks:java:io:iceberg:bqms")
+ runtimeOnly(project(":sdks:java:io:delta")) {
+ exclude group: "org.apache.hadoop", module: "hadoop-aws"
+ }
implementation project(path: ":sdks:java:core", configuration: "shadow")
implementation project(":sdks:java:extensions:google-cloud-platform-core")
implementation project(":sdks:java:io:google-cloud-platform")
implementation project(":sdks:java:managed")
+ implementation library.java.delta_kernel_api
+ implementation library.java.delta_kernel_defaults
implementation library.java.google_auth_library_oauth2_http
+ implementation library.java.hadoop_common
implementation library.java.joda_time
+ implementation library.java.slf4j_api
runtimeOnly project(path: ":runners:direct-java", configuration: "shadow")
implementation library.java.vendored_guava_32_1_2_jre
runtimeOnly library.java.hadoop_client
runtimeOnly library.java.bigdataoss_gcs_connector
+ testImplementation library.java.hamcrest
+ testImplementation library.java.junit
+ // Provides DeltaWriteTestUtils, which writes Delta Lake tables for tests.
+ testImplementation(project(path: ":sdks:java:io:delta", configuration:
"testRuntimeMigration")) {
+ exclude group: "org.apache.hadoop", module: "hadoop-aws"
+ exclude group: "org.testcontainers"
+ exclude group: "org.slf4j", module: "slf4j-simple"
+ }
+
// Add dependencies for the PreCommit configurations
// For each runner a project level dependency on the examples project.
for (String runner : preCommitRunners) {
@@ -65,4 +81,19 @@ dependencies {
configurations.all {
// iceberg-core needs avro:1.12.0
resolutionStrategy.force 'org.apache.avro:avro:1.12.0'
+ // The Delta Lake source reads tables on GCS through the GCS connector,
whose version enforced by
+ // the GCP libraries BOM conflicts with the latest Hadoop (see
sdks/java/io/delta/build.gradle). So
+ // use Hadoop 3.3 together with a compatible Parquet version.
+ resolutionStrategy.force 'org.apache.hadoop:hadoop-client:3.3.6'
+ resolutionStrategy.force 'org.apache.hadoop:hadoop-client-api:3.3.6'
+ resolutionStrategy.force 'org.apache.hadoop:hadoop-client-runtime:3.3.6'
+ resolutionStrategy.force 'org.apache.hadoop:hadoop-common:3.3.6'
+ resolutionStrategy.force 'org.apache.hadoop:hadoop-hdfs:3.3.6'
+ resolutionStrategy.force 'org.apache.hadoop:hadoop-hdfs-client:3.3.6'
+ resolutionStrategy.force 'org.apache.parquet:parquet-avro:1.16.0'
+ resolutionStrategy.force 'org.apache.parquet:parquet-hadoop:1.16.0'
+ // Exclude conflicting SLF4J bindings pulled in by the Delta Lake and Hadoop
dependencies.
+ exclude group: "org.apache.logging.log4j", module: "log4j-slf4j2-impl"
+ exclude group: "org.apache.logging.log4j", module: "log4j-slf4j-impl"
+ exclude group: "org.slf4j", module: "slf4j-reload4j"
}
diff --git
a/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExample.java
b/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExample.java
new file mode 100644
index 00000000000..5d6827b0e04
--- /dev/null
+++
b/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExample.java
@@ -0,0 +1,448 @@
+/*
+ * 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 java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import
org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcUtils.LakehouseTableSpec;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.PipelineResult;
+import org.apache.beam.sdk.extensions.gcp.options.GcpOptions;
+import org.apache.beam.sdk.managed.Managed;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.Description;
+import org.apache.beam.sdk.options.ExperimentalOptions;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
+import org.apache.beam.sdk.options.Validation;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.Row;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Splitter;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableList;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * An Apache Beam Java example pipeline that reads Change Data Feed (CDC)
records from a Delta Lake
+ * table on Google Cloud Storage (GCS) and applies those row-level changes
(`INSERT`,
+ * `UPDATE_BEFORE`, `UPDATE_AFTER`, `DELETE`) to a Google Cloud Platform (GCP)
Lakehouse table (via
+ * the Lakehouse runtime catalog's Iceberg REST API) using Beam's {@link
Managed} I/O connectors.
+ *
+ * <h2>Overview</h2>
+ *
+ * <p>This pipeline performs the following steps:
+ *
+ * <ol>
+ * <li><b>Validates Delta Lake Change Data Feed (CDF):</b> Uses the Delta
Kernel API to read the
+ * latest snapshot of the input Delta Lake table and verify that the
table property {@code
+ * delta.enableChangeDataFeed = true} is enabled, failing fast with an
error if it is not.
+ * <li><b>Reads Delta Lake CDC Data:</b> Uses {@code
Managed.read(Managed.DELTA_LAKE_CDC)} to read
+ * change events over either a commit version range ({@code
--startVersion} / {@code
+ * --endVersion}) or an ISO-8601 timestamp range ({@code
--startTimestamp} / {@code
+ * --endTimestamp}). Standard GCS Hadoop filesystem properties and Delta
CDC metadata columns
+ * are configured by default.
+ * <li><b>Writes CDC Data to GCP Lakehouse:</b> Uses {@code
Managed.write(Managed.ICEBERG)} in
+ * {@code merge-on-read} mode against the Lakehouse Iceberg REST catalog
to apply inserts,
+ * updates, and deletes ordered by {@code _commit_version}. Standard
Lakehouse REST catalog
+ * and Iceberg CDC sink properties are configured by default.
+ * </ol>
+ *
+ * <h2>Prerequisites</h2>
+ *
+ * <ul>
+ * <li><b>Java 17+:</b> Both the Delta Lake Kernel API and Iceberg require
Java 17 or later. If
+ * using SDKMAN, switch to Java 17 before running:
+ * <pre>{@code
+ * sdk use java 17.0.15-tem
+ * }</pre>
+ * <li><b>Source Delta Lake Table on GCS:</b> Must have Change Data Feed
enabled:
+ * <pre>{@code
+ * ALTER TABLE delta.`gs://my-bucket/path/to/delta_table`
+ * SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true');
+ * }</pre>
+ * <li><b>Target GCP Lakehouse (Iceberg V2) Table:</b> Must be an Iceberg
format-version 2 table
+ * registered in the Lakehouse runtime catalog. Its schema must match
the source Delta Lake
+ * table's columns. If the Iceberg table does not declare identifier
(primary-key) fields in
+ * its table schema, pass {@code --equalityColumns=<col1,col2>} to
specify the primary-key
+ * columns used for equality deletes.
+ * <li><b>GCP Authentication:</b> Set {@code GOOGLE_APPLICATION_CREDENTIALS}
to a service account
+ * key JSON file (or configure Application Default Credentials) with
permissions to access the
+ * GCS buckets, Lakehouse runtime catalog ({@code roles/biglake.admin}),
and Dataflow.
+ * </ul>
+ *
+ * <h2>Running the Example on Dataflow Portable Runner (Example Default)</h2>
+ *
+ * <p>This example configures the <b>Dataflow Portable Runner</b> ({@code
+ * --experiments=use_runner_v2}) by default ({@code
--usePortableRunner=true}). Because the Dataflow
+ * Portable Runner does not yet propagate an element's native Beam {@code
ValueKind} metadata across
+ * the FnAPI boundary, the source includes {@code _change_type} in {@code
include_metadata_columns}
+ * and the Iceberg CDC sink is configured with {@code change_type_column:
"_change_type"} and {@code
+ * change_type_map}.
+ *
+ * <h3>1. Reading by Commit Version Range</h3>
+ *
+ * <pre>{@code
+ * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
+ *
+ * ./gradlew :examples:java:iceberg:execute \
+ *
-PmainClass=org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample \
+ * -Pexec.args="--runner=DataflowRunner \
+ * --project=my-gcp-project \
+ * --region=us-central1 \
+ * --tempLocation=gs://my-temp-bucket/temp \
+ * --deltaTable=gs://my-delta-lake-bucket/delta_lake/employee_data/ \
+ *
--lakehouseTable=my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table
\
+ * --equalityColumns=employee_id \
+ * --startVersion=2 \
+ * --endVersion=4"
+ * }</pre>
+ *
+ * <h3>2. Reading by Timestamp Range</h3>
+ *
+ * <pre>{@code
+ * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
+ *
+ * ./gradlew :examples:java:iceberg:execute \
+ *
-PmainClass=org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample \
+ * -Pexec.args="--runner=DataflowRunner \
+ * --project=my-gcp-project \
+ * --region=us-central1 \
+ * --tempLocation=gs://my-temp-bucket/temp \
+ * --deltaTable=gs://my-delta-lake-bucket/delta_lake/employee_data/ \
+ *
--lakehouseTable=my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table
\
+ * --equalityColumns=employee_id \
+ * --startTimestamp=2026-10-01T00:00:00Z \
+ * --endTimestamp=2026-10-02T00:00:00Z"
+ * }</pre>
+ *
+ * <h2>Running the Example on Dataflow Streaming Java Runner</h2>
+ *
+ * <p>When running on the <b>Dataflow Streaming Java Runner</b> (which is
Dataflow's default runner
+ * when {@code --experiments=use_runner_v2} is not enabled), {@code
change_type_column} (and {@code
+ * change_type_map}) <b>can be skipped</b> on the Iceberg CDC sink, and {@code
_change_type} does
+ * not need to be included in {@code include_metadata_columns} on the Delta
Lake CDC source. The
+ * Delta Lake CDC reader ({@code Managed.DELTA_LAKE_CDC}) automatically sets
each emitted {@link
+ * Row}'s native Beam {@code ValueKind} ({@code INSERT}, {@code DELETE},
{@code UPDATE_BEFORE},
+ * {@code UPDATE_AFTER}), which the Dataflow Streaming Java Runner preserves
and passes directly to
+ * the Iceberg CDC sink.
+ *
+ * <p>In your own pipelines running on the Dataflow Streaming Java Runner, you
do not need to set
+ * any runner-version flag (since it is Dataflow's default when {@code
use_runner_v2} is not added),
+ * and the {@code Managed} configurations only need {@code _commit_version}
for ordering:
+ *
+ * <pre>{@code
+ * // Delta Lake CDC read config on Dataflow Streaming Java Runner (no
_change_type column needed):
+ * Map<String, Object> readConfig = ImmutableMap.of(
+ * "table", deltaTable,
+ * "start_version", 2L,
+ * "end_version", 4L,
+ * "include_metadata_columns", ImmutableList.of("_commit_version"),
+ * "hadoop_config", hadoopConfig);
+ *
+ * // Iceberg CDC write config on Dataflow Streaming Java Runner
(change_type_column can be skipped):
+ * Map<String, Object> writeConfig = ImmutableMap.of(
+ * "table", tableId,
+ * "catalog_name", "lakehouse",
+ * "catalog_properties", catalogProps,
+ * "mode", "merge-on-read",
+ * "sequence_number_column", "_commit_version",
+ * "equality_columns", ImmutableList.of("employee_id"));
+ * }</pre>
+ *
+ * <p>Because this example enables the Dataflow Portable Runner by default
({@code
+ * --usePortableRunner=true}), you can override it to run on the Dataflow
Streaming Java Runner (and
+ * skip {@code change_type_column}) by passing {@code
--usePortableRunner=false}:
+ *
+ * <pre>{@code
+ * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
+ *
+ * ./gradlew :examples:java:iceberg:execute \
+ *
-PmainClass=org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample \
+ * -Pexec.args="--runner=DataflowRunner \
+ * --usePortableRunner=false \
+ * --project=my-gcp-project \
+ * --region=us-central1 \
+ * --tempLocation=gs://my-temp-bucket/temp \
+ * --deltaTable=gs://my-delta-lake-bucket/delta_lake/employee_data/ \
+ *
--lakehouseTable=my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table
\
+ * --equalityColumns=employee_id \
+ * --startVersion=2 \
+ * --endVersion=4"
+ * }</pre>
+ */
+public class DeltaLakeToLakehouseCdcExample {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DeltaLakeToLakehouseCdcExample.class);
+
+ /** Delta Lake CDC metadata column names produced by {@code
Managed.DELTA_LAKE_CDC}. */
+ public static final String CHANGE_TYPE_COLUMN = "_change_type";
+
+ public static final String COMMIT_VERSION_COLUMN = "_commit_version";
+
+ /** Default Lakehouse Iceberg REST Catalog endpoint URI. */
+ public static final String DEFAULT_LAKEHOUSE_CATALOG_URI =
+ "https://biglake.googleapis.com/iceberg/v1/restcatalog";;
+
+ /**
+ * Mapping from Delta Lake Change Data Feed {@code _change_type} values to
the canonical change
+ * types expected by the Iceberg CDC sink.
+ */
+ public static final Map<String, String> DELTA_TO_ICEBERG_CHANGE_TYPES =
+ ImmutableMap.of(
+ "insert", "INSERT",
+ "delete", "DELETE",
+ "update_preimage", "UPDATE_BEFORE",
+ "update_postimage", "UPDATE_AFTER");
+
+ /** Pipeline options for {@link DeltaLakeToLakehouseCdcExample}. */
+ public interface Options extends GcpOptions {
+
+ @Description(
+ "GCS path of the source Delta Lake table repository to read CDC data
from "
+ + "(e.g. gs://my-bucket/delta_lake/my_table/).")
+ @Validation.Required
+ String getDeltaTable();
+
+ void setDeltaTable(String value);
+
+ @Description(
+ "Target GCP Lakehouse Iceberg table identifier. Accepts either a
4-part Lakehouse table "
+ + "identifier '<project>.<warehouse_bucket>.<namespace>.<table>' "
+ + "(e.g.
'my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table'), "
+ + "a 3-part identifier '<warehouse_bucket>.<namespace>.<table>',
or a 2-part Iceberg "
+ + "table identifier '<namespace>.<table>' (when --warehouse is
also specified).")
+ @Validation.Required
+ String getLakehouseTable();
+
+ void setLakehouseTable(String value);
+
+ @Description(
+ "Starting Delta Lake commit version (inclusive) to read changes from. "
+ + "Either --startVersion or --startTimestamp must be provided.")
+ @Nullable Long getStartVersion();
+
+ void setStartVersion(@Nullable Long value);
+
+ @Description(
+ "Ending Delta Lake commit version (inclusive) to read changes up to.
Optional; defaults "
+ + "to the latest commit version if omitted.")
+ @Nullable Long getEndVersion();
+
+ void setEndVersion(@Nullable Long value);
+
+ @Description(
+ "Starting timestamp in ISO-8601 format (e.g. '2026-10-01T00:00:00Z')
to read Delta Lake "
+ + "changes from. Either --startVersion or --startTimestamp must be
provided.")
+ @Nullable String getStartTimestamp();
+
+ void setStartTimestamp(@Nullable String value);
+
+ @Description(
+ "Ending timestamp in ISO-8601 format (e.g. '2026-10-01T23:59:59Z') to
read Delta Lake "
+ + "changes up to. Optional.")
+ @Nullable String getEndTimestamp();
+
+ void setEndTimestamp(@Nullable String value);
+
+ @Description(
+ "GCS warehouse location for the GCP Lakehouse catalog (e.g.
'gs://my-warehouse-bucket'). "
+ + "Optional when --lakehouseTable is specified as a 3-part or
4-part identifier containing "
+ + "the warehouse bucket name.")
+ @Nullable String getWarehouse();
+
+ void setWarehouse(@Nullable String value);
+
+ @Description(
+ "Comma-separated list of primary-key (equality-delete) column names
that uniquely identify "
+ + "rows in the target Lakehouse Iceberg table (e.g.
'employee_id'). Required if the "
+ + "target Iceberg table does not declare identifier-field-ids in
its schema or if the "
+ + "table does not exist yet.")
+ @Nullable String getEqualityColumns();
+
+ void setEqualityColumns(@Nullable String value);
+
+ @Description(
+ "If true, applies changes in upsert mode (UPDATE_BEFORE records are
dropped and "
+ + "INSERT/UPDATE_AFTER records are applied as upserts). Defaults
to false.")
+ @Default.Boolean(false)
+ boolean getUpsert();
+
+ void setUpsert(boolean value);
+
+ @Description(
+ "If true (default for this example), runs on the Dataflow Portable
Runner ('use_runner_v2') "
+ + "and configures 'change_type_column' ('_change_type') on the
Iceberg CDC sink. "
+ + "Set to false to override and run on the Dataflow Streaming Java
Runner, which skips "
+ + "'change_type_column' and relies on the native Beam ValueKind
metadata attached to "
+ + "each Row by the Delta Lake CDC reader.")
+ @Default.Boolean(true)
+ boolean getUsePortableRunner();
+
+ void setUsePortableRunner(boolean value);
+
+ @Description("Name of the Iceberg catalog instance. Defaults to
'lakehouse'.")
+ @Default.String("lakehouse")
+ String getCatalogName();
+
+ void setCatalogName(String value);
+
+ @Description(
+ "Lakehouse Iceberg REST Catalog endpoint URI. Defaults to "
+ + DEFAULT_LAKEHOUSE_CATALOG_URI
+ + ".")
+ @Default.String(DEFAULT_LAKEHOUSE_CATALOG_URI)
+ String getCatalogUri();
+
+ void setCatalogUri(String value);
+ }
+
+ /**
+ * Builds the Hadoop configuration used to access the Delta Lake table on
GCS, both by the Delta
+ * Lake CDC source and when verifying that Change Data Feed is enabled on
the table.
+ */
+ private static Map<String, String> buildGcsHadoopConfig(String project) {
+ return ImmutableMap.of(
+ "fs.gs.impl", "com.google.cloud.hadoop.fs.gcs.GoogleHadoopFileSystem",
+ "fs.AbstractFileSystem.gs.impl",
"com.google.cloud.hadoop.fs.gcs.GoogleHadoopFS",
+ "fs.gs.auth.type", "APPLICATION_DEFAULT",
+ "fs.gs.project.id", project);
+ }
+
+ /** Builds the {@code Managed.read(Managed.DELTA_LAKE_CDC)} configuration
map. */
+ @VisibleForTesting
+ static Map<String, Object> buildDeltaCdcReadConfig(Options options, String
project) {
+ Map<String, Object> readConfig = new HashMap<>();
+ readConfig.put("table", options.getDeltaTable());
+ @Nullable Long startVersion = options.getStartVersion();
+ if (startVersion != null) {
+ readConfig.put("start_version", startVersion);
+ }
+ @Nullable Long endVersion = options.getEndVersion();
+ if (endVersion != null) {
+ readConfig.put("end_version", endVersion);
+ }
+ @Nullable String startTimestamp = options.getStartTimestamp();
+ if (startTimestamp != null && !startTimestamp.trim().isEmpty()) {
+ readConfig.put("start_timestamp", startTimestamp.trim());
+ }
+ @Nullable String endTimestamp = options.getEndTimestamp();
+ if (endTimestamp != null && !endTimestamp.trim().isEmpty()) {
+ readConfig.put("end_timestamp", endTimestamp.trim());
+ }
+ readConfig.put(
+ "include_metadata_columns",
+ options.getUsePortableRunner()
+ ? ImmutableList.of(CHANGE_TYPE_COLUMN, COMMIT_VERSION_COLUMN)
+ : ImmutableList.of(COMMIT_VERSION_COLUMN));
+ readConfig.put("hadoop_config", buildGcsHadoopConfig(project));
+ return readConfig;
+ }
+
+ /** Builds the {@code Managed.write(Managed.ICEBERG)} CDC configuration map
for GCP Lakehouse. */
+ @VisibleForTesting
+ static Map<String, Object> buildLakehouseCdcWriteConfig(
+ Options options, LakehouseTableSpec tableSpec) {
+ Map<String, String> catalogProps =
+ ImmutableMap.<String, String>builder()
+ .put("type", "rest")
+ .put("uri", options.getCatalogUri())
+ .put("warehouse", tableSpec.warehouse)
+ .put("header.x-goog-user-project", tableSpec.project)
+ .put("io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO")
+ .put("rest.auth.type",
"org.apache.iceberg.gcp.auth.GoogleAuthManager")
+ .build();
+
+ Map<String, Object> writeConfig = new HashMap<>();
+ writeConfig.put("table", tableSpec.tableId);
+ writeConfig.put("catalog_name", options.getCatalogName());
+ writeConfig.put("catalog_properties", catalogProps);
+ writeConfig.put("mode", "merge-on-read");
+ writeConfig.put("sequence_number_column", COMMIT_VERSION_COLUMN);
+ if (options.getUsePortableRunner()) {
+ writeConfig.put("change_type_column", CHANGE_TYPE_COLUMN);
+ writeConfig.put("change_type_map", DELTA_TO_ICEBERG_CHANGE_TYPES);
+ }
+ writeConfig.put("upsert", options.getUpsert());
+
+ @Nullable String equalityColsStr = options.getEqualityColumns();
+ if (equalityColsStr != null && !equalityColsStr.trim().isEmpty()) {
+ List<String> equalityCols =
+
Splitter.on(',').trimResults().omitEmptyStrings().splitToList(equalityColsStr).stream()
+ .collect(Collectors.toList());
+ if (!equalityCols.isEmpty()) {
+ writeConfig.put("equality_columns", equalityCols);
+ }
+ }
+ return writeConfig;
+ }
+
+ public static void runDeltaLakeToLakehouseCdc(Options options) {
+ DeltaLakeToLakehouseCdcUtils.validateRangeOptions(options);
+ LakehouseTableSpec tableSpec =
DeltaLakeToLakehouseCdcUtils.resolveLakehouseTableSpec(options);
+ if (options.getProject() == null || options.getProject().trim().isEmpty())
{
+ options.setProject(tableSpec.project);
+ }
+
+ // Configure Dataflow Portable Runner by default unless
--usePortableRunner=false (Dataflow
+ // Streaming Java Runner) is specified.
+ if (options.getUsePortableRunner()) {
+ ExperimentalOptions.addExperiment(options.as(ExperimentalOptions.class),
"use_runner_v2");
+ }
+
+ // Verify that Change Data Feed is enabled on the source Delta Lake table.
+ DeltaLakeToLakehouseCdcUtils.verifyChangeDataFeedEnabled(
+ options.getDeltaTable(), buildGcsHadoopConfig(tableSpec.project));
+
+ Map<String, Object> readConfig = buildDeltaCdcReadConfig(options,
tableSpec.project);
+ Map<String, Object> writeConfig = buildLakehouseCdcWriteConfig(options,
tableSpec);
+
+ LOG.info(
+ "Starting Delta Lake to GCP Lakehouse CDC pipeline: source='{}',
target='{}' (warehouse='{}', project='{}')",
+ options.getDeltaTable(),
+ tableSpec.tableId,
+ tableSpec.warehouse,
+ tableSpec.project);
+
+ Pipeline p = Pipeline.create(options);
+
+ PCollection<Row> changes =
+ p.apply(
+ "ReadDeltaLakeChangeFeed",
+ Managed.read(Managed.DELTA_LAKE_CDC).withConfig(readConfig))
+ .getSinglePCollection();
+
+ changes.apply(
+ "ApplyChangesToLakehouse",
Managed.write(Managed.ICEBERG).withConfig(writeConfig));
+
+ PipelineResult.State state = p.run().waitUntilFinish();
+ if (state != PipelineResult.State.DONE) {
+ throw new IllegalStateException(
+ String.format(
+ "Delta Lake to Lakehouse CDC pipeline failed with terminal
state: %s", state));
+ }
+ }
+
+ public static void main(String[] args) {
+ Options options =
PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class);
+ runDeltaLakeToLakehouseCdc(options);
+ }
+}
diff --git
a/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcUtils.java
b/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcUtils.java
new file mode 100644
index 00000000000..99b3498c9fb
--- /dev/null
+++
b/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcUtils.java
@@ -0,0 +1,197 @@
+/*
+ * 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 io.delta.kernel.Snapshot;
+import io.delta.kernel.Table;
+import io.delta.kernel.defaults.engine.DefaultEngine;
+import io.delta.kernel.engine.Engine;
+import io.delta.kernel.exceptions.TableNotFoundException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample.Options;
+import org.apache.hadoop.conf.Configuration;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Helpers for {@link DeltaLakeToLakehouseCdcExample} that parse the target
Lakehouse table
+ * identifier and validate the pipeline options and the source Delta Lake
table.
+ */
+final class DeltaLakeToLakehouseCdcUtils {
+
+ private static final Logger LOG =
LoggerFactory.getLogger(DeltaLakeToLakehouseCdcUtils.class);
+
+ /** Delta Lake table property that enables the Change Data Feed of a table.
*/
+ static final String ENABLE_CHANGE_DATA_FEED_PROPERTY =
"delta.enableChangeDataFeed";
+
+ private DeltaLakeToLakehouseCdcUtils() {}
+
+ /** Parsed GCP Lakehouse table coordinates (`project`, `warehouse`, and
`namespace.table`). */
+ static final class LakehouseTableSpec {
+ final String project;
+ final String warehouse;
+ final String tableId;
+
+ LakehouseTableSpec(String project, String warehouse, String tableId) {
+ this.project = project;
+ this.warehouse = warehouse;
+ this.tableId = tableId;
+ }
+ }
+
+ /**
+ * Resolves the GCP project, GCS warehouse URI, and 2-part Iceberg {@code
namespace.table}
+ * identifier from {@link Options}.
+ */
+ static LakehouseTableSpec resolveLakehouseTableSpec(Options options) {
+ String rawTable = options.getLakehouseTable();
+ if (rawTable == null || rawTable.trim().isEmpty()) {
+ throw new IllegalArgumentException("--lakehouseTable must not be
empty.");
+ }
+ List<String> parts = Arrays.asList(rawTable.trim().split("\\."));
+ @Nullable String project = options.getProject();
+ @Nullable String warehouse = options.getWarehouse();
+ String tableId;
+
+ if (parts.size() == 4) {
+ project = parts.get(0);
+ if (warehouse == null || warehouse.trim().isEmpty()) {
+ warehouse = "gs://" + parts.get(1);
+ }
+ tableId = parts.get(2) + "." + parts.get(3);
+ } else if (parts.size() == 3) {
+ if (warehouse == null || warehouse.trim().isEmpty()) {
+ warehouse = "gs://" + parts.get(0);
+ }
+ tableId = parts.get(1) + "." + parts.get(2);
+ } else if (parts.size() == 2) {
+ tableId = rawTable.trim();
+ } else {
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid --lakehouseTable '%s'. Expected
'<project>.<warehouse_bucket>.<namespace>.<table>', "
+ + "'<warehouse_bucket>.<namespace>.<table>', or
'<namespace>.<table>'.",
+ 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}.
+ *
+ * <p>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<String,
String> 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 00000000000..a6cd955eb67
--- /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<String, Object> 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<String, String> hadoopConfig = (Map<String, String>)
readConfig.get("hadoop_config");
+ assertEquals("my-gcp-project", hadoopConfig.get("fs.gs.project.id"));
+
+ Map<String, Object> 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<String, String> catalogProps = (Map<String, String>)
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<String, Object> readConfig =
+ DeltaLakeToLakehouseCdcExample.buildDeltaCdcReadConfig(options,
spec.project);
+ assertEquals(
+ Collections.singletonList("_commit_version"),
readConfig.get("include_metadata_columns"));
+
+ Map<String, Object> 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 00000000000..02b3042e7c1
--- /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<String, String> 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<String, String>
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));
+ }
+}
