From f032d8ade397bbe465e8d53128a2197777cd7bb7 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Thu, 1 Oct 2026 18:05:46 -0700 Subject: [PATCH 1/4] Adds an end-to-end example for migrating change data (CDC) from Delta Lake to GCP Lakehouse --- examples/java/build.gradle | 37 + .../DeltaLakeToLakehouseCdcExample.java | 639 ++++++++++++++++++ .../DeltaLakeToLakehouseCdcExampleTest.java | 257 +++++++ 3 files changed, 933 insertions(+) create mode 100644 examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java create mode 100644 examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java diff --git a/examples/java/build.gradle b/examples/java/build.gradle index 84ace728362e..6fca0ec6a76e 100644 --- a/examples/java/build.gradle +++ b/examples/java/build.gradle @@ -56,6 +56,7 @@ dependencies { implementation project(":sdks:java:io:google-cloud-platform") implementation project(":sdks:java:io:kafka") implementation project(":sdks:java:extensions:ml") + implementation project(":sdks:java:managed") implementation library.java.avro implementation library.java.bigdataoss_util implementation library.java.google_api_client @@ -111,6 +112,42 @@ dependencies { if (project.hasProperty("runnerDependency")) { runtimeOnly project(path: project.getProperty("runnerDependency")) } + + def javaVer = project.findProperty('testJavaVersion') ?: JavaVersion.current().majorVersion + if ((javaVer as int) >= 17 && !project.hasProperty('compileAndRunTestsWithJava11')) { + runtimeOnly project(":sdks:java:io:iceberg") + runtimeOnly(project(":sdks:java:io:delta")) { + exclude group: "org.apache.hadoop", module: "hadoop-aws" + } + runtimeOnly "org.yaml:snakeyaml:2.0" + } +} + +def activeJavaVer = (project.findProperty('testJavaVersion') ?: JavaVersion.current().majorVersion) as int +if (activeJavaVer >= 17 && !project.hasProperty('compileAndRunTestsWithJava11')) { + configurations.all { + if (canBeResolved) { + attributes { + attribute(TargetJvmVersion.TARGET_JVM_VERSION_ATTRIBUTE, activeJavaVer) + } + } + 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" + resolutionStrategy.force 'org.apache.avro:avro:1.12.0' + resolutionStrategy.force 'org.apache.parquet:parquet-avro:1.16.0' + resolutionStrategy.force 'org.apache.parquet:parquet-hadoop:1.16.0' + resolutionStrategy.force 'org.apache.hadoop:hadoop-common:3.3.6' + resolutionStrategy.force 'org.apache.hadoop:hadoop-client: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.hadoop:hadoop-client-api:3.3.6' + resolutionStrategy.force 'org.apache.hadoop:hadoop-client-runtime:3.3.6' + resolutionStrategy.force library.java.bigdataoss_gcs_connector + resolutionStrategy.force library.java.bigdataoss_util_hadoop + resolutionStrategy.force library.java.bigdataoss_gcsio + resolutionStrategy.force library.java.bigdataoss_util + } } /* diff --git a/examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java b/examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java new file mode 100644 index 000000000000..5b5b3a7b590e --- /dev/null +++ b/examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java @@ -0,0 +1,639 @@ +/* + * 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; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.ObjectMapper; +import java.io.BufferedReader; +import java.io.IOException; +import java.io.InputStreamReader; +import java.nio.channels.Channels; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Comparator; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.stream.Collectors; +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.io.FileSystems; +import org.apache.beam.sdk.io.fs.MatchResult; +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 (BigLake + * Metastore Iceberg REST Catalog) table using Beam's {@link Managed} I/O connectors. + * + *

Overview

+ * + *

This pipeline performs the following steps: + * + *

    + *
  1. Validates Delta Lake Change Data Feed (CDF): Inspects the input Delta Lake table's + * transaction log ({@code _delta_log/*.json}) to verify that the table property {@code + * delta.enableChangeDataFeed = true} is enabled, failing fast with an error if it is not. + *
  2. Reads Delta Lake CDC Data: 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. + *
  3. Writes CDC Data to GCP Lakehouse: Uses {@code Managed.write(Managed.ICEBERG)} in + * {@code merge-on-read} mode against the BigLake Iceberg REST catalog to apply inserts, + * updates, and deletes ordered by {@code _commit_version}. Standard BigLake REST catalog and + * Iceberg CDC sink properties are configured by default. + *
+ * + *

Prerequisites

+ * + * + * + *

Running the Example on Dataflow Runner v2

+ * + *

By default ({@code --useRunnerV2=true}), the pipeline runs on Dataflow Runner v2 ({@code + * --experiments=use_runner_v2}). Because Dataflow Runner v2 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}. + * + *

1. Reading by Commit Version Range

+ * + * 
{@code
+ * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
+ *
+ * ./gradlew :examples:java:execute \
+ *   -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \
+ *   -Pexec.args="--runner=DataflowRunner \
+ *     --project=apache-beam-testing \
+ *     --region=us-central1 \
+ *     --tempLocation=gs://apache-beam-testing-chamikara/temp \
+ *     --deltaTable=gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/ \
+ *     --lakehouseTable=apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc \
+ *     --equalityColumns=employee_id \
+ *     --startVersion=2 \
+ *     --endVersion=4"
+ * }
+ * + *

2. Reading by Timestamp Range

+ * + * 
{@code
+ * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
+ *
+ * ./gradlew :examples:java:execute \
+ *   -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \
+ *   -Pexec.args="--runner=DataflowRunner \
+ *     --project=apache-beam-testing \
+ *     --region=us-central1 \
+ *     --tempLocation=gs://apache-beam-testing-chamikara/temp \
+ *     --deltaTable=gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/ \
+ *     --lakehouseTable=apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc \
+ *     --equalityColumns=employee_id \
+ *     --startTimestamp=2026-10-01T00:00:00Z \
+ *     --endTimestamp=2026-10-02T00:00:00Z"
+ * }
+ * + *

Running the Example on Dataflow Runner v1

+ * + *

When running on Dataflow Runner v1 (the legacy worker, without {@code use_runner_v2}), + * {@code change_type_column} (and {@code change_type_map}) can be skipped 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 Dataflow Runner v1 preserves + * and passes directly to the Iceberg CDC sink. + * + *

In custom pipelines running on Dataflow Runner v1, the {@code Managed} configurations only + * need {@code _commit_version} for ordering: + * + * 

{@code
+ * // Delta Lake CDC read config on Dataflow Runner v1 (no _change_type column needed):
+ * Map 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 Runner v1 (change_type_column can be skipped):
+ * Map 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"));
+ * }
+ * + *

To run this example on Dataflow Runner v1 (which omits {@code use_runner_v2} and skips {@code + * change_type_column}), pass {@code --useRunnerV2=false}: + * + * 

{@code
+ * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
+ *
+ * ./gradlew :examples:java:execute \
+ *   -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \
+ *   -Pexec.args="--runner=DataflowRunner \
+ *     --useRunnerV2=false \
+ *     --project=apache-beam-testing \
+ *     --region=us-central1 \
+ *     --tempLocation=gs://apache-beam-testing-chamikara/temp \
+ *     --deltaTable=gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/ \
+ *     --lakehouseTable=apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc \
+ *     --equalityColumns=employee_id \
+ *     --startVersion=2 \
+ *     --endVersion=4"
+ * }
+ */ +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 BigLake Iceberg REST Catalog endpoint URI. */ + public static final String DEFAULT_BIGLAKE_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 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 BigLake table " + + "identifier '...' " + + "(e.g. 'apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc'), " + + "a 3-part identifier '..
', or a 2-part Iceberg " + + "table identifier '.
' (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 BigLake 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), runs on Dataflow Runner v2 ('use_runner_v2') and configures " + + "'change_type_column' ('_change_type') on the Iceberg CDC sink. If false, runs on " + + "Dataflow Runner v1 (legacy worker) and skips 'change_type_column', relying on the " + + "native Beam ValueKind metadata attached to each Row by the Delta Lake CDC reader.") + @Default.Boolean(true) + boolean getUseRunnerV2(); + + void setUseRunnerV2(boolean value); + + @Description("Name of the Iceberg catalog instance. Defaults to 'lakehouse'.") + @Default.String("lakehouse") + String getCatalogName(); + + void setCatalogName(String value); + + @Description( + "BigLake Iceberg REST Catalog endpoint URI. Defaults to " + + DEFAULT_BIGLAKE_CATALOG_URI + + ".") + @Default.String(DEFAULT_BIGLAKE_CATALOG_URI) + String getCatalogUri(); + + void setCatalogUri(String value); + } + + /** Parsed GCP Lakehouse table coordinates (`project`, `warehouse`, and `namespace.table`). */ + @VisibleForTesting + 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}. + */ + @VisibleForTesting + static LakehouseTableSpec resolveLakehouseTableSpec(Options options) { + String rawTable = options.getLakehouseTable(); + if (rawTable == null || rawTable.trim().isEmpty()) { + throw new IllegalArgumentException("--lakehouseTable must not be empty."); + } + List parts = Arrays.asList(rawTable.trim().split("\\.")); + @Nullable String project = options.getProject(); + @Nullable String warehouse = options.getWarehouse(); + String tableId; + + if (parts.size() == 4) { + if (project == null || project.trim().isEmpty()) { + 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 '...
', " + + "'..
', or '.
'.", + 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}. + */ + @VisibleForTesting + 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 the input Delta Lake table has {@code delta.enableChangeDataFeed = true} set in + * its latest transaction log metadata. + * + * @throws IllegalArgumentException if no Delta Lake commit logs exist at {@code deltaTablePath} + * @throws IllegalStateException if {@code delta.enableChangeDataFeed = true} is not enabled + */ + @VisibleForTesting + static void verifyChangeDataFeedEnabled(String deltaTablePath) throws IOException { + String normalizedPath = deltaTablePath.trim().replaceAll("/+$", ""); + String logPattern = normalizedPath + "/_delta_log/*.json"; + + MatchResult matchResult = FileSystems.match(logPattern); + if (matchResult.status() != MatchResult.Status.OK || matchResult.metadata().isEmpty()) { + throw new IllegalArgumentException( + String.format( + "No Delta Lake transaction log files found matching '%s'. Verify that '%s' is a " + + "valid Delta Lake table.", + logPattern, deltaTablePath)); + } + + List logFiles = new ArrayList<>(matchResult.metadata()); + logFiles.sort( + Comparator.comparing((MatchResult.Metadata m) -> m.resourceId().toString()).reversed()); + + ObjectMapper mapper = new ObjectMapper(); + for (MatchResult.Metadata metadata : logFiles) { + try (BufferedReader reader = + new BufferedReader( + new InputStreamReader( + Channels.newInputStream(FileSystems.open(metadata.resourceId())), + StandardCharsets.UTF_8))) { + @Nullable String latestCdfPropertyInFile = null; + boolean foundMetaDataInFile = false; + @Nullable String line; + while ((line = reader.readLine()) != null) { + if (!line.contains("\"metaData\"")) { + continue; + } + JsonNode root = mapper.readTree(line); + JsonNode metaDataNode = root.get("metaData"); + if (metaDataNode != null && metaDataNode.isObject()) { + foundMetaDataInFile = true; + JsonNode configNode = metaDataNode.get("configuration"); + if (configNode != null && configNode.isObject()) { + JsonNode cdfNode = configNode.get("delta.enableChangeDataFeed"); + latestCdfPropertyInFile = cdfNode != null ? cdfNode.asText() : null; + } else { + latestCdfPropertyInFile = null; + } + } + } + if (foundMetaDataInFile) { + if ("true".equalsIgnoreCase(latestCdfPropertyInFile)) { + LOG.info( + "Verified 'delta.enableChangeDataFeed = true' on Delta Lake table '{}' (from {}).", + deltaTablePath, + metadata.resourceId()); + return; + } + throw new IllegalStateException( + String.format( + "Delta Lake table '%s' does not have 'delta.enableChangeDataFeed = true' enabled " + + "(found '%s' in %s). Enable Change Data Feed on the table before running " + + "this pipeline.", + deltaTablePath, latestCdfPropertyInFile, metadata.resourceId())); + } + } + } + + throw new IllegalStateException( + String.format( + "Could not find 'metaData' action in Delta Lake commit logs under '%s' to verify " + + "'delta.enableChangeDataFeed = true'.", + logPattern)); + } + + /** Builds the {@code Managed.read(Managed.DELTA_LAKE_CDC)} configuration map. */ + @VisibleForTesting + static Map buildDeltaCdcReadConfig(Options options, String project) { + Map hadoopConfig = + 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); + + Map 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.getUseRunnerV2() + ? ImmutableList.of(CHANGE_TYPE_COLUMN, COMMIT_VERSION_COLUMN) + : ImmutableList.of(COMMIT_VERSION_COLUMN)); + readConfig.put("hadoop_config", hadoopConfig); + return readConfig; + } + + /** Builds the {@code Managed.write(Managed.ICEBERG)} CDC configuration map for GCP Lakehouse. */ + @VisibleForTesting + static Map buildLakehouseCdcWriteConfig( + Options options, LakehouseTableSpec tableSpec) { + Map catalogProps = + ImmutableMap.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 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.getUseRunnerV2()) { + 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 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) throws IOException { + validateRangeOptions(options); + LakehouseTableSpec tableSpec = resolveLakehouseTableSpec(options); + if (options.getProject() == null || options.getProject().trim().isEmpty()) { + options.setProject(tableSpec.project); + } + + // Configure Dataflow Runner v2 by default unless --useRunnerV2=false (Runner v1) is specified. + if (options.getUseRunnerV2()) { + ExperimentalOptions.addExperiment(options.as(ExperimentalOptions.class), "use_runner_v2"); + } + + // Initialize Beam FileSystems with pipeline options and verify Change Data Feed is enabled. + FileSystems.setDefaultPipelineOptions(options); + verifyChangeDataFeedEnabled(options.getDeltaTable()); + + Map readConfig = buildDeltaCdcReadConfig(options, tableSpec.project); + Map 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 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) throws IOException { + Options options = PipelineOptionsFactory.fromArgs(args).withValidation().as(Options.class); + runDeltaLakeToLakehouseCdc(options); + } +} diff --git a/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java b/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java new file mode 100644 index 000000000000..fffcdb47f7a8 --- /dev/null +++ b/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java @@ -0,0 +1,257 @@ +/* + * 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; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertThrows; + +import java.io.File; +import java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.Arrays; +import java.util.Collections; +import java.util.Map; +import org.apache.beam.examples.DeltaLakeToLakehouseCdcExample.LakehouseTableSpec; +import org.apache.beam.examples.DeltaLakeToLakehouseCdcExample.Options; +import org.apache.beam.sdk.options.PipelineOptionsFactory; +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 DeltaLakeToLakehouseCdcExample}. */ +@RunWith(JUnit4.class) +public class DeltaLakeToLakehouseCdcExampleTest { + + @Rule public final TemporaryFolder tmpFolder = new TemporaryFolder(); + + @Test + public void testVerifyChangeDataFeedEnabledPassesWhenTrueInLatestMetaData() throws IOException { + File tableDir = tmpFolder.newFolder("delta_table_enabled"); + File deltaLogDir = new File(tableDir, "_delta_log"); + deltaLogDir.mkdirs(); + + // Commit 0: CDF not enabled initially + Files.write( + new File(deltaLogDir, "00000000000000000000.json").toPath(), + Arrays.asList( + "{\"commitInfo\":{\"timestamp\":1700000000000,\"operation\":\"WRITE\"}}", + "{\"metaData\":{\"id\":\"test-id\",\"configuration\":{}}}"), + StandardCharsets.UTF_8); + + // Commit 1: ALTER TABLE SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true') + Files.write( + new File(deltaLogDir, "00000000000000000001.json").toPath(), + Arrays.asList( + "{\"commitInfo\":{\"timestamp\":1700000001000,\"operation\":\"SET TBLPROPERTIES\"}}", + "{\"metaData\":{\"id\":\"test-id\",\"configuration\":{\"delta.enableChangeDataFeed\":\"true\"}}}"), + StandardCharsets.UTF_8); + + // Commit 2: Data commit with no metaData action + Files.write( + new File(deltaLogDir, "00000000000000000002.json").toPath(), + Arrays.asList( + "{\"commitInfo\":{\"timestamp\":1700000002000,\"operation\":\"WRITE\"}}", + "{\"add\":{\"path\":\"part-00000.parquet\",\"size\":1024,\"modificationTime\":1700000002000,\"dataChange\":true}}"), + StandardCharsets.UTF_8); + + DeltaLakeToLakehouseCdcExample.verifyChangeDataFeedEnabled(tableDir.getAbsolutePath()); + } + + @Test + public void testVerifyChangeDataFeedEnabledFailsWhenFalse() throws IOException { + File tableDir = tmpFolder.newFolder("delta_table_disabled"); + File deltaLogDir = new File(tableDir, "_delta_log"); + deltaLogDir.mkdirs(); + + Files.write( + new File(deltaLogDir, "00000000000000000000.json").toPath(), + Arrays.asList( + "{\"commitInfo\":{\"timestamp\":1700000000000,\"operation\":\"WRITE\"}}", + "{\"metaData\":{\"id\":\"test-id\",\"configuration\":{\"delta.enableChangeDataFeed\":\"false\"}}}"), + StandardCharsets.UTF_8); + + IllegalStateException ex = + assertThrows( + IllegalStateException.class, + () -> + DeltaLakeToLakehouseCdcExample.verifyChangeDataFeedEnabled( + tableDir.getAbsolutePath())); + assertThat(ex.getMessage(), containsString("delta.enableChangeDataFeed = true")); + } + + @Test + public void testVerifyChangeDataFeedEnabledFailsWhenMissingFromConfiguration() + throws IOException { + File tableDir = tmpFolder.newFolder("delta_table_missing_prop"); + File deltaLogDir = new File(tableDir, "_delta_log"); + deltaLogDir.mkdirs(); + + Files.write( + new File(deltaLogDir, "00000000000000000000.json").toPath(), + Arrays.asList( + "{\"commitInfo\":{\"timestamp\":1700000000000,\"operation\":\"WRITE\"}}", + "{\"metaData\":{\"id\":\"test-id\",\"configuration\":{}}}"), + StandardCharsets.UTF_8); + + IllegalStateException ex = + assertThrows( + IllegalStateException.class, + () -> + DeltaLakeToLakehouseCdcExample.verifyChangeDataFeedEnabled( + tableDir.getAbsolutePath())); + assertThat(ex.getMessage(), containsString("delta.enableChangeDataFeed = true")); + } + + @Test + public void testVerifyChangeDataFeedEnabledFailsWhenNoLogsExist() throws IOException { + File emptyDir = tmpFolder.newFolder("not_a_delta_table"); + + IllegalArgumentException ex = + assertThrows( + IllegalArgumentException.class, + () -> + DeltaLakeToLakehouseCdcExample.verifyChangeDataFeedEnabled( + emptyDir.getAbsolutePath())); + assertThat(ex.getMessage(), containsString("No Delta Lake transaction log files found")); + } + + @Test + public void testResolveLakehouseTableSpecFourPart() { + Options options = PipelineOptionsFactory.as(Options.class); + options.setLakehouseTable( + "apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc"); + + LakehouseTableSpec spec = DeltaLakeToLakehouseCdcExample.resolveLakehouseTableSpec(options); + assertEquals("apache-beam-testing", spec.project); + assertEquals("gs://apache-beam-testing-chamikara", spec.warehouse); + assertEquals("delta_lake_test.delta_lake_to_lakehouse_cdc", spec.tableId); + } + + @Test + public void testResolveLakehouseTableSpecTwoPartWithExplicitWarehouseAndProject() { + Options options = PipelineOptionsFactory.as(Options.class); + options.setProject("apache-beam-testing"); + options.setWarehouse("gs://apache-beam-testing-chamikara"); + options.setLakehouseTable("delta_lake_test.delta_lake_to_lakehouse_cdc"); + + LakehouseTableSpec spec = DeltaLakeToLakehouseCdcExample.resolveLakehouseTableSpec(options); + assertEquals("apache-beam-testing", spec.project); + assertEquals("gs://apache-beam-testing-chamikara", spec.warehouse); + assertEquals("delta_lake_test.delta_lake_to_lakehouse_cdc", spec.tableId); + } + + @Test + public void testValidateRangeOptionsVersionAndTimestampRanges() { + Options versionOpts = PipelineOptionsFactory.as(Options.class); + versionOpts.setStartVersion(1L); + versionOpts.setEndVersion(3L); + DeltaLakeToLakehouseCdcExample.validateRangeOptions(versionOpts); + + Options tsOpts = PipelineOptionsFactory.as(Options.class); + tsOpts.setStartTimestamp("2026-10-01T00:00:00Z"); + tsOpts.setEndTimestamp("2026-10-02T00:00:00Z"); + DeltaLakeToLakehouseCdcExample.validateRangeOptions(tsOpts); + + Options missingOpts = PipelineOptionsFactory.as(Options.class); + assertThrows( + IllegalArgumentException.class, + () -> DeltaLakeToLakehouseCdcExample.validateRangeOptions(missingOpts)); + + Options mixedOpts = PipelineOptionsFactory.as(Options.class); + mixedOpts.setStartVersion(1L); + mixedOpts.setStartTimestamp("2026-10-01T00:00:00Z"); + assertThrows( + IllegalArgumentException.class, + () -> DeltaLakeToLakehouseCdcExample.validateRangeOptions(mixedOpts)); + + Options invertedOpts = PipelineOptionsFactory.as(Options.class); + invertedOpts.setStartVersion(5L); + invertedOpts.setEndVersion(2L); + assertThrows( + IllegalArgumentException.class, + () -> DeltaLakeToLakehouseCdcExample.validateRangeOptions(invertedOpts)); + } + + @Test + @SuppressWarnings("unchecked") + public void testBuildManagedConfigs() { + Options options = PipelineOptionsFactory.as(Options.class); + options.setDeltaTable("gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/"); + options.setLakehouseTable( + "apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc"); + options.setEqualityColumns("employee_id"); + options.setStartVersion(2L); + options.setEndVersion(4L); + + LakehouseTableSpec spec = DeltaLakeToLakehouseCdcExample.resolveLakehouseTableSpec(options); + Map readConfig = + DeltaLakeToLakehouseCdcExample.buildDeltaCdcReadConfig(options, spec.project); + assertEquals( + "gs://apache-beam-testing-delta-lake/delta_lake/demo_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("apache-beam-testing", hadoopConfig.get("fs.gs.project.id")); + + Map writeConfig = + DeltaLakeToLakehouseCdcExample.buildLakehouseCdcWriteConfig(options, spec); + assertEquals("delta_lake_test.delta_lake_to_lakehouse_cdc", 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://apache-beam-testing-chamikara", catalogProps.get("warehouse")); + assertEquals("apache-beam-testing", catalogProps.get("header.x-goog-user-project")); + } + + @Test + public void testBuildManagedConfigsRunnerV1SkipsChangeTypeColumn() { + Options options = PipelineOptionsFactory.as(Options.class); + options.setUseRunnerV2(false); + options.setDeltaTable("gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/"); + options.setLakehouseTable( + "apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc"); + options.setEqualityColumns("employee_id"); + options.setStartVersion(2L); + options.setEndVersion(4L); + + LakehouseTableSpec spec = DeltaLakeToLakehouseCdcExample.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")); + } +} From 5472bce2c5a4adfc21405a90c5f8c92e66557d88 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Thu, 1 Oct 2026 23:57:57 -0700 Subject: [PATCH 2/4] updates --- .../DeltaLakeToLakehouseCdcExample.java | 98 ++++++++++--------- .../DeltaLakeToLakehouseCdcExampleTest.java | 44 ++++----- 2 files changed, 72 insertions(+), 70 deletions(-) diff --git a/examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java b/examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java index 5b5b3a7b590e..5115d3bf0428 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java +++ b/examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java @@ -100,13 +100,14 @@ * GCS buckets, BigLake Metastore ({@code roles/biglake.admin}), and Dataflow. * * - *

Running the Example on Dataflow Runner v2

+ *

Running the Example on Dataflow Portable Runner (Example Default)

* - *

By default ({@code --useRunnerV2=true}), the pipeline runs on Dataflow Runner v2 ({@code - * --experiments=use_runner_v2}). Because Dataflow Runner v2 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}. + *

This example configures the Dataflow Portable Runner ({@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}. * *

1. Reading by Commit Version Range

* @@ -116,11 +117,11 @@ * ./gradlew :examples:java:execute \ * -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \ * -Pexec.args="--runner=DataflowRunner \ - * --project=apache-beam-testing \ + * --project=my-gcp-project \ * --region=us-central1 \ - * --tempLocation=gs://apache-beam-testing-chamikara/temp \ - * --deltaTable=gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/ \ - * --lakehouseTable=apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc \ + * --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" @@ -134,31 +135,33 @@ * ./gradlew :examples:java:execute \ * -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \ * -Pexec.args="--runner=DataflowRunner \ - * --project=apache-beam-testing \ + * --project=my-gcp-project \ * --region=us-central1 \ - * --tempLocation=gs://apache-beam-testing-chamikara/temp \ - * --deltaTable=gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/ \ - * --lakehouseTable=apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc \ + * --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" * } * - *

Running the Example on Dataflow Runner v1

+ *

Running the Example on Dataflow Streaming Java Runner

* - *

When running on Dataflow Runner v1 (the legacy worker, without {@code use_runner_v2}), - * {@code change_type_column} (and {@code change_type_map}) can be skipped 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 Dataflow Runner v1 preserves - * and passes directly to the Iceberg CDC sink. + *

When running on the Dataflow Streaming Java Runner (which is Dataflow's default runner + * when {@code --experiments=use_runner_v2} is not enabled), {@code change_type_column} (and {@code + * change_type_map}) can be skipped 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. * - *

In custom pipelines running on Dataflow Runner v1, the {@code Managed} configurations only - * need {@code _commit_version} for ordering: + *

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: * * 

{@code
- * // Delta Lake CDC read config on Dataflow Runner v1 (no _change_type column needed):
+ * // Delta Lake CDC read config on Dataflow Streaming Java Runner (no _change_type column needed):
  * Map readConfig = ImmutableMap.of(
  *     "table", deltaTable,
  *     "start_version", 2L,
@@ -166,7 +169,7 @@
  *     "include_metadata_columns", ImmutableList.of("_commit_version"),
  *     "hadoop_config", hadoopConfig);
  *
- * // Iceberg CDC write config on Dataflow Runner v1 (change_type_column can be skipped):
+ * // Iceberg CDC write config on Dataflow Streaming Java Runner (change_type_column can be skipped):
  * Map writeConfig = ImmutableMap.of(
  *     "table", tableId,
  *     "catalog_name", "lakehouse",
@@ -176,8 +179,9 @@
  *     "equality_columns", ImmutableList.of("employee_id"));
  * }
* - *

To run this example on Dataflow Runner v1 (which omits {@code use_runner_v2} and skips {@code - * change_type_column}), pass {@code --useRunnerV2=false}: + *

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}: * * 

{@code
  * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
@@ -185,12 +189,12 @@
  * ./gradlew :examples:java:execute \
  *   -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \
  *   -Pexec.args="--runner=DataflowRunner \
- *     --useRunnerV2=false \
- *     --project=apache-beam-testing \
+ *     --usePortableRunner=false \
+ *     --project=my-gcp-project \
  *     --region=us-central1 \
- *     --tempLocation=gs://apache-beam-testing-chamikara/temp \
- *     --deltaTable=gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/ \
- *     --lakehouseTable=apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc \
+ *     --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"
@@ -234,7 +238,7 @@ public interface Options extends GcpOptions {
     @Description(
         "Target GCP Lakehouse Iceberg table identifier. Accepts either a 4-part BigLake table "
             + "identifier '...
' " - + "(e.g. 'apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc'), " + + "(e.g. 'my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table'), " + "a 3-part identifier '..
', or a 2-part Iceberg " + "table identifier '.
' (when --warehouse is also specified).") @Validation.Required @@ -296,14 +300,15 @@ public interface Options extends GcpOptions { void setUpsert(boolean value); @Description( - "If true (default), runs on Dataflow Runner v2 ('use_runner_v2') and configures " - + "'change_type_column' ('_change_type') on the Iceberg CDC sink. If false, runs on " - + "Dataflow Runner v1 (legacy worker) and skips 'change_type_column', relying on the " - + "native Beam ValueKind metadata attached to each Row by the Delta Lake CDC reader.") + "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 getUseRunnerV2(); + boolean getUsePortableRunner(); - void setUseRunnerV2(boolean value); + void setUsePortableRunner(boolean value); @Description("Name of the Iceberg catalog instance. Defaults to 'lakehouse'.") @Default.String("lakehouse") @@ -351,9 +356,7 @@ static LakehouseTableSpec resolveLakehouseTableSpec(Options options) { String tableId; if (parts.size() == 4) { - if (project == null || project.trim().isEmpty()) { - project = parts.get(0); - } + project = parts.get(0); if (warehouse == null || warehouse.trim().isEmpty()) { warehouse = "gs://" + parts.get(1); } @@ -542,7 +545,7 @@ static Map buildDeltaCdcReadConfig(Options options, String proje } readConfig.put( "include_metadata_columns", - options.getUseRunnerV2() + options.getUsePortableRunner() ? ImmutableList.of(CHANGE_TYPE_COLUMN, COMMIT_VERSION_COLUMN) : ImmutableList.of(COMMIT_VERSION_COLUMN)); readConfig.put("hadoop_config", hadoopConfig); @@ -569,7 +572,7 @@ static Map buildLakehouseCdcWriteConfig( writeConfig.put("catalog_properties", catalogProps); writeConfig.put("mode", "merge-on-read"); writeConfig.put("sequence_number_column", COMMIT_VERSION_COLUMN); - if (options.getUseRunnerV2()) { + if (options.getUsePortableRunner()) { writeConfig.put("change_type_column", CHANGE_TYPE_COLUMN); writeConfig.put("change_type_map", DELTA_TO_ICEBERG_CHANGE_TYPES); } @@ -594,8 +597,9 @@ public static void runDeltaLakeToLakehouseCdc(Options options) throws IOExceptio options.setProject(tableSpec.project); } - // Configure Dataflow Runner v2 by default unless --useRunnerV2=false (Runner v1) is specified. - if (options.getUseRunnerV2()) { + // 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"); } diff --git a/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java b/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java index fffcdb47f7a8..4321c49e9f75 100644 --- a/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java +++ b/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java @@ -140,25 +140,25 @@ public void testVerifyChangeDataFeedEnabledFailsWhenNoLogsExist() throws IOExcep public void testResolveLakehouseTableSpecFourPart() { Options options = PipelineOptionsFactory.as(Options.class); options.setLakehouseTable( - "apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc"); + "my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table"); LakehouseTableSpec spec = DeltaLakeToLakehouseCdcExample.resolveLakehouseTableSpec(options); - assertEquals("apache-beam-testing", spec.project); - assertEquals("gs://apache-beam-testing-chamikara", spec.warehouse); - assertEquals("delta_lake_test.delta_lake_to_lakehouse_cdc", spec.tableId); + 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("apache-beam-testing"); - options.setWarehouse("gs://apache-beam-testing-chamikara"); - options.setLakehouseTable("delta_lake_test.delta_lake_to_lakehouse_cdc"); + options.setProject("my-gcp-project"); + options.setWarehouse("gs://my-lakehouse-warehouse-bucket"); + options.setLakehouseTable("my_namespace.my_iceberg_table"); LakehouseTableSpec spec = DeltaLakeToLakehouseCdcExample.resolveLakehouseTableSpec(options); - assertEquals("apache-beam-testing", spec.project); - assertEquals("gs://apache-beam-testing-chamikara", spec.warehouse); - assertEquals("delta_lake_test.delta_lake_to_lakehouse_cdc", spec.tableId); + assertEquals("my-gcp-project", spec.project); + assertEquals("gs://my-lakehouse-warehouse-bucket", spec.warehouse); + assertEquals("my_namespace.my_iceberg_table", spec.tableId); } @Test @@ -197,9 +197,9 @@ public void testValidateRangeOptionsVersionAndTimestampRanges() { @SuppressWarnings("unchecked") public void testBuildManagedConfigs() { Options options = PipelineOptionsFactory.as(Options.class); - options.setDeltaTable("gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/"); + options.setDeltaTable("gs://my-delta-lake-bucket/delta_lake/employee_data/"); options.setLakehouseTable( - "apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc"); + "my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table"); options.setEqualityColumns("employee_id"); options.setStartVersion(2L); options.setEndVersion(4L); @@ -207,37 +207,35 @@ public void testBuildManagedConfigs() { LakehouseTableSpec spec = DeltaLakeToLakehouseCdcExample.resolveLakehouseTableSpec(options); Map readConfig = DeltaLakeToLakehouseCdcExample.buildDeltaCdcReadConfig(options, spec.project); - assertEquals( - "gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/", - readConfig.get("table")); + 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("apache-beam-testing", hadoopConfig.get("fs.gs.project.id")); + assertEquals("my-gcp-project", hadoopConfig.get("fs.gs.project.id")); Map writeConfig = DeltaLakeToLakehouseCdcExample.buildLakehouseCdcWriteConfig(options, spec); - assertEquals("delta_lake_test.delta_lake_to_lakehouse_cdc", writeConfig.get("table")); + 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://apache-beam-testing-chamikara", catalogProps.get("warehouse")); - assertEquals("apache-beam-testing", catalogProps.get("header.x-goog-user-project")); + assertEquals("gs://my-lakehouse-warehouse-bucket", catalogProps.get("warehouse")); + assertEquals("my-gcp-project", catalogProps.get("header.x-goog-user-project")); } @Test - public void testBuildManagedConfigsRunnerV1SkipsChangeTypeColumn() { + public void testBuildManagedConfigsStreamingJavaRunnerSkipsChangeTypeColumn() { Options options = PipelineOptionsFactory.as(Options.class); - options.setUseRunnerV2(false); - options.setDeltaTable("gs://apache-beam-testing-delta-lake/delta_lake/demo_employee_data/"); + options.setUsePortableRunner(false); + options.setDeltaTable("gs://my-delta-lake-bucket/delta_lake/employee_data/"); options.setLakehouseTable( - "apache-beam-testing.apache-beam-testing-chamikara.delta_lake_test.delta_lake_to_lakehouse_cdc"); + "my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table"); options.setEqualityColumns("employee_id"); options.setStartVersion(2L); options.setEndVersion(4L); From 1ff670e53d4b7c0685fef418988cb3f12d7afef1 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Fri, 2 Oct 2026 00:28:26 -0700 Subject: [PATCH 3/4] updates --- examples/java/build.gradle | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/examples/java/build.gradle b/examples/java/build.gradle index 6fca0ec6a76e..74102c15dcd5 100644 --- a/examples/java/build.gradle +++ b/examples/java/build.gradle @@ -113,8 +113,9 @@ dependencies { runtimeOnly project(path: project.getProperty("runnerDependency")) } - def javaVer = project.findProperty('testJavaVersion') ?: JavaVersion.current().majorVersion - if ((javaVer as int) >= 17 && !project.hasProperty('compileAndRunTestsWithJava11')) { + def javaVer = (project.findProperty('testJavaVersion') ?: JavaVersion.current().majorVersion) as int + def isDeltaLakeCdcExecute = project.findProperty('mainClass')?.toString()?.contains('DeltaLakeToLakehouseCdcExample') + if (isDeltaLakeCdcExecute && javaVer >= 17 && !project.hasProperty('compileAndRunTestsWithJava11')) { runtimeOnly project(":sdks:java:io:iceberg") runtimeOnly(project(":sdks:java:io:delta")) { exclude group: "org.apache.hadoop", module: "hadoop-aws" @@ -124,7 +125,8 @@ dependencies { } def activeJavaVer = (project.findProperty('testJavaVersion') ?: JavaVersion.current().majorVersion) as int -if (activeJavaVer >= 17 && !project.hasProperty('compileAndRunTestsWithJava11')) { +def isDeltaLakeCdcExecute = project.findProperty('mainClass')?.toString()?.contains('DeltaLakeToLakehouseCdcExample') +if (isDeltaLakeCdcExecute && activeJavaVer >= 17 && !project.hasProperty('compileAndRunTestsWithJava11')) { configurations.all { if (canBeResolved) { attributes { From 735f36dd941ee0755f37973ccb72f3d82d289b10 Mon Sep 17 00:00:00 2001 From: Chamikara Jayalath Date: Fri, 9 Oct 2026 01:38:21 -0700 Subject: [PATCH 4/4] Address reviewer comments --- examples/java/build.gradle | 39 --- examples/java/iceberg/build.gradle | 31 ++ .../DeltaLakeToLakehouseCdcExample.java | 277 +++--------------- .../iceberg/DeltaLakeToLakehouseCdcUtils.java | 197 +++++++++++++ .../DeltaLakeToLakehouseCdcExampleTest.java | 96 ++++++ .../DeltaLakeToLakehouseCdcUtilsTest.java | 229 +++++++++++++++ .../DeltaLakeToLakehouseCdcExampleTest.java | 255 ---------------- 7 files changed, 594 insertions(+), 530 deletions(-) rename examples/java/{src/main/java/org/apache/beam/examples => iceberg/src/main/java/org/apache/beam/examples/iceberg}/DeltaLakeToLakehouseCdcExample.java (62%) create mode 100644 examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcUtils.java create mode 100644 examples/java/iceberg/src/test/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExampleTest.java create mode 100644 examples/java/iceberg/src/test/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcUtilsTest.java delete mode 100644 examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java diff --git a/examples/java/build.gradle b/examples/java/build.gradle index 74102c15dcd5..84ace728362e 100644 --- a/examples/java/build.gradle +++ b/examples/java/build.gradle @@ -56,7 +56,6 @@ dependencies { implementation project(":sdks:java:io:google-cloud-platform") implementation project(":sdks:java:io:kafka") implementation project(":sdks:java:extensions:ml") - implementation project(":sdks:java:managed") implementation library.java.avro implementation library.java.bigdataoss_util implementation library.java.google_api_client @@ -112,44 +111,6 @@ dependencies { if (project.hasProperty("runnerDependency")) { runtimeOnly project(path: project.getProperty("runnerDependency")) } - - def javaVer = (project.findProperty('testJavaVersion') ?: JavaVersion.current().majorVersion) as int - def isDeltaLakeCdcExecute = project.findProperty('mainClass')?.toString()?.contains('DeltaLakeToLakehouseCdcExample') - if (isDeltaLakeCdcExecute && javaVer >= 17 && !project.hasProperty('compileAndRunTestsWithJava11')) { - runtimeOnly project(":sdks:java:io:iceberg") - runtimeOnly(project(":sdks:java:io:delta")) { - exclude group: "org.apache.hadoop", module: "hadoop-aws" - } - runtimeOnly "org.yaml:snakeyaml:2.0" - } -} - -def activeJavaVer = (project.findProperty('testJavaVersion') ?: JavaVersion.current().majorVersion) as int -def isDeltaLakeCdcExecute = project.findProperty('mainClass')?.toString()?.contains('DeltaLakeToLakehouseCdcExample') -if (isDeltaLakeCdcExecute && activeJavaVer >= 17 && !project.hasProperty('compileAndRunTestsWithJava11')) { - configurations.all { - if (canBeResolved) { - attributes { - attribute(TargetJvmVersion.TARGET_JVM_VERSION_ATTRIBUTE, activeJavaVer) - } - } - 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" - resolutionStrategy.force 'org.apache.avro:avro:1.12.0' - resolutionStrategy.force 'org.apache.parquet:parquet-avro:1.16.0' - resolutionStrategy.force 'org.apache.parquet:parquet-hadoop:1.16.0' - resolutionStrategy.force 'org.apache.hadoop:hadoop-common:3.3.6' - resolutionStrategy.force 'org.apache.hadoop:hadoop-client: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.hadoop:hadoop-client-api:3.3.6' - resolutionStrategy.force 'org.apache.hadoop:hadoop-client-runtime:3.3.6' - resolutionStrategy.force library.java.bigdataoss_gcs_connector - resolutionStrategy.force library.java.bigdataoss_util_hadoop - resolutionStrategy.force library.java.bigdataoss_gcsio - resolutionStrategy.force library.java.bigdataoss_util - } } /* diff --git a/examples/java/iceberg/build.gradle b/examples/java/iceberg/build.gradle index 2cff15c5fb0d..7186ff9c7030 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/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java b/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExample.java similarity index 62% rename from examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java rename to examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExample.java index 5115d3bf0428..5d6827b0e040 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExample.java +++ b/examples/java/iceberg/src/main/java/org/apache/beam/examples/iceberg/DeltaLakeToLakehouseCdcExample.java @@ -15,27 +15,16 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.examples; - -import com.fasterxml.jackson.databind.JsonNode; -import com.fasterxml.jackson.databind.ObjectMapper; -import java.io.BufferedReader; -import java.io.IOException; -import java.io.InputStreamReader; -import java.nio.channels.Channels; -import java.nio.charset.StandardCharsets; -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Comparator; +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.io.FileSystems; -import org.apache.beam.sdk.io.fs.MatchResult; import org.apache.beam.sdk.managed.Managed; import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.Description; @@ -55,16 +44,16 @@ /** * 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 (BigLake - * Metastore Iceberg REST Catalog) table using Beam's {@link Managed} I/O connectors. + * `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. * *

Overview

* *

This pipeline performs the following steps: * *

    - *
  1. Validates Delta Lake Change Data Feed (CDF): Inspects the input Delta Lake table's - * transaction log ({@code _delta_log/*.json}) to verify that the table property {@code + *
  2. Validates Delta Lake Change Data Feed (CDF): 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. *
  3. Reads Delta Lake CDC Data: Uses {@code Managed.read(Managed.DELTA_LAKE_CDC)} to read * change events over either a commit version range ({@code --startVersion} / {@code @@ -72,9 +61,9 @@ * --endTimestamp}). Standard GCS Hadoop filesystem properties and Delta CDC metadata columns * are configured by default. *
  4. Writes CDC Data to GCP Lakehouse: Uses {@code Managed.write(Managed.ICEBERG)} in - * {@code merge-on-read} mode against the BigLake Iceberg REST catalog to apply inserts, - * updates, and deletes ordered by {@code _commit_version}. Standard BigLake REST catalog and - * Iceberg CDC sink properties are configured by default. + * {@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. *
* *

Prerequisites

@@ -91,13 +80,13 @@ * SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true'); * } *
  • Target GCP Lakehouse (Iceberg V2) Table: Must be an Iceberg format-version 2 table - * registered in BigLake Metastore REST Catalog. Its schema must match the source Delta Lake + * 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=} to specify the primary-key * columns used for equality deletes. *
  • GCP Authentication: Set {@code GOOGLE_APPLICATION_CREDENTIALS} to a service account * key JSON file (or configure Application Default Credentials) with permissions to access the - * GCS buckets, BigLake Metastore ({@code roles/biglake.admin}), and Dataflow. + * GCS buckets, Lakehouse runtime catalog ({@code roles/biglake.admin}), and Dataflow. * * *

    Running the Example on Dataflow Portable Runner (Example Default)

    @@ -114,8 +103,8 @@ * 
    {@code
  * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
  *
- * ./gradlew :examples:java:execute \
- *   -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \
+ * ./gradlew :examples:java:iceberg:execute \
+ *   -PmainClass=org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample \
  *   -Pexec.args="--runner=DataflowRunner \
  *     --project=my-gcp-project \
  *     --region=us-central1 \
@@ -132,8 +121,8 @@
  * 
    {@code
  * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
  *
- * ./gradlew :examples:java:execute \
- *   -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \
+ * ./gradlew :examples:java:iceberg:execute \
+ *   -PmainClass=org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample \
  *   -Pexec.args="--runner=DataflowRunner \
  *     --project=my-gcp-project \
  *     --region=us-central1 \
@@ -186,8 +175,8 @@
  * 
    {@code
  * export GOOGLE_APPLICATION_CREDENTIALS=/path/to/service-account-key.json
  *
- * ./gradlew :examples:java:execute \
- *   -PmainClass=org.apache.beam.examples.DeltaLakeToLakehouseCdcExample \
+ * ./gradlew :examples:java:iceberg:execute \
+ *   -PmainClass=org.apache.beam.examples.iceberg.DeltaLakeToLakehouseCdcExample \
  *   -Pexec.args="--runner=DataflowRunner \
  *     --usePortableRunner=false \
  *     --project=my-gcp-project \
@@ -209,8 +198,8 @@ public class DeltaLakeToLakehouseCdcExample {
 
   public static final String COMMIT_VERSION_COLUMN = "_commit_version";
 
-  /** Default BigLake Iceberg REST Catalog endpoint URI. */
-  public static final String DEFAULT_BIGLAKE_CATALOG_URI =
+  /** Default Lakehouse Iceberg REST Catalog endpoint URI. */
+  public static final String DEFAULT_LAKEHOUSE_CATALOG_URI =
       "https://biglake.googleapis.com/iceberg/v1/restcatalog";
 
   /**
@@ -236,7 +225,7 @@ public interface Options extends GcpOptions {
     void setDeltaTable(String value);
 
     @Description(
-        "Target GCP Lakehouse Iceberg table identifier. Accepts either a 4-part BigLake table "
+        "Target GCP Lakehouse Iceberg table identifier. Accepts either a 4-part Lakehouse table "
             + "identifier '...
    • ' " + "(e.g. 'my-gcp-project.my-lakehouse-warehouse-bucket.my_namespace.my_iceberg_table'), " + "a 3-part identifier '..
    ', or a 2-part Iceberg " @@ -275,7 +264,7 @@ public interface Options extends GcpOptions { void setEndTimestamp(@Nullable String value); @Description( - "GCS warehouse location for the GCP Lakehouse BigLake catalog (e.g. 'gs://my-warehouse-bucket'). " + "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(); @@ -317,214 +306,30 @@ public interface Options extends GcpOptions { void setCatalogName(String value); @Description( - "BigLake Iceberg REST Catalog endpoint URI. Defaults to " - + DEFAULT_BIGLAKE_CATALOG_URI + "Lakehouse Iceberg REST Catalog endpoint URI. Defaults to " + + DEFAULT_LAKEHOUSE_CATALOG_URI + ".") - @Default.String(DEFAULT_BIGLAKE_CATALOG_URI) + @Default.String(DEFAULT_LAKEHOUSE_CATALOG_URI) String getCatalogUri(); void setCatalogUri(String value); } - /** Parsed GCP Lakehouse table coordinates (`project`, `warehouse`, and `namespace.table`). */ - @VisibleForTesting - 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}. - */ - @VisibleForTesting - static LakehouseTableSpec resolveLakehouseTableSpec(Options options) { - String rawTable = options.getLakehouseTable(); - if (rawTable == null || rawTable.trim().isEmpty()) { - throw new IllegalArgumentException("--lakehouseTable must not be empty."); - } - List 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 '...
    ', " - + "'..
    ', or '.
    '.", - 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}. - */ - @VisibleForTesting - 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 the input Delta Lake table has {@code delta.enableChangeDataFeed = true} set in - * its latest transaction log metadata. - * - * @throws IllegalArgumentException if no Delta Lake commit logs exist at {@code deltaTablePath} - * @throws IllegalStateException if {@code delta.enableChangeDataFeed = true} is not enabled + * 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. */ - @VisibleForTesting - static void verifyChangeDataFeedEnabled(String deltaTablePath) throws IOException { - String normalizedPath = deltaTablePath.trim().replaceAll("/+$", ""); - String logPattern = normalizedPath + "/_delta_log/*.json"; - - MatchResult matchResult = FileSystems.match(logPattern); - if (matchResult.status() != MatchResult.Status.OK || matchResult.metadata().isEmpty()) { - throw new IllegalArgumentException( - String.format( - "No Delta Lake transaction log files found matching '%s'. Verify that '%s' is a " - + "valid Delta Lake table.", - logPattern, deltaTablePath)); - } - - List logFiles = new ArrayList<>(matchResult.metadata()); - logFiles.sort( - Comparator.comparing((MatchResult.Metadata m) -> m.resourceId().toString()).reversed()); - - ObjectMapper mapper = new ObjectMapper(); - for (MatchResult.Metadata metadata : logFiles) { - try (BufferedReader reader = - new BufferedReader( - new InputStreamReader( - Channels.newInputStream(FileSystems.open(metadata.resourceId())), - StandardCharsets.UTF_8))) { - @Nullable String latestCdfPropertyInFile = null; - boolean foundMetaDataInFile = false; - @Nullable String line; - while ((line = reader.readLine()) != null) { - if (!line.contains("\"metaData\"")) { - continue; - } - JsonNode root = mapper.readTree(line); - JsonNode metaDataNode = root.get("metaData"); - if (metaDataNode != null && metaDataNode.isObject()) { - foundMetaDataInFile = true; - JsonNode configNode = metaDataNode.get("configuration"); - if (configNode != null && configNode.isObject()) { - JsonNode cdfNode = configNode.get("delta.enableChangeDataFeed"); - latestCdfPropertyInFile = cdfNode != null ? cdfNode.asText() : null; - } else { - latestCdfPropertyInFile = null; - } - } - } - if (foundMetaDataInFile) { - if ("true".equalsIgnoreCase(latestCdfPropertyInFile)) { - LOG.info( - "Verified 'delta.enableChangeDataFeed = true' on Delta Lake table '{}' (from {}).", - deltaTablePath, - metadata.resourceId()); - return; - } - throw new IllegalStateException( - String.format( - "Delta Lake table '%s' does not have 'delta.enableChangeDataFeed = true' enabled " - + "(found '%s' in %s). Enable Change Data Feed on the table before running " - + "this pipeline.", - deltaTablePath, latestCdfPropertyInFile, metadata.resourceId())); - } - } - } - - throw new IllegalStateException( - String.format( - "Could not find 'metaData' action in Delta Lake commit logs under '%s' to verify " - + "'delta.enableChangeDataFeed = true'.", - logPattern)); + private static Map 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 buildDeltaCdcReadConfig(Options options, String project) { - Map hadoopConfig = - 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); - Map readConfig = new HashMap<>(); readConfig.put("table", options.getDeltaTable()); @Nullable Long startVersion = options.getStartVersion(); @@ -548,7 +353,7 @@ static Map buildDeltaCdcReadConfig(Options options, String proje options.getUsePortableRunner() ? ImmutableList.of(CHANGE_TYPE_COLUMN, COMMIT_VERSION_COLUMN) : ImmutableList.of(COMMIT_VERSION_COLUMN)); - readConfig.put("hadoop_config", hadoopConfig); + readConfig.put("hadoop_config", buildGcsHadoopConfig(project)); return readConfig; } @@ -590,9 +395,9 @@ static Map buildLakehouseCdcWriteConfig( return writeConfig; } - public static void runDeltaLakeToLakehouseCdc(Options options) throws IOException { - validateRangeOptions(options); - LakehouseTableSpec tableSpec = resolveLakehouseTableSpec(options); + 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); } @@ -603,9 +408,9 @@ public static void runDeltaLakeToLakehouseCdc(Options options) throws IOExceptio ExperimentalOptions.addExperiment(options.as(ExperimentalOptions.class), "use_runner_v2"); } - // Initialize Beam FileSystems with pipeline options and verify Change Data Feed is enabled. - FileSystems.setDefaultPipelineOptions(options); - verifyChangeDataFeedEnabled(options.getDeltaTable()); + // Verify that Change Data Feed is enabled on the source Delta Lake table. + DeltaLakeToLakehouseCdcUtils.verifyChangeDataFeedEnabled( + options.getDeltaTable(), buildGcsHadoopConfig(tableSpec.project)); Map readConfig = buildDeltaCdcReadConfig(options, tableSpec.project); Map writeConfig = buildLakehouseCdcWriteConfig(options, tableSpec); @@ -636,7 +441,7 @@ public static void runDeltaLakeToLakehouseCdc(Options options) throws IOExceptio } } - public static void main(String[] args) throws IOException { + 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 000000000000..99b3498c9fb3 --- /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 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 '...
    ', " + + "'..
    ', or '.
    '.", + 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)); + } +} diff --git a/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java b/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java deleted file mode 100644 index 4321c49e9f75..000000000000 --- a/examples/java/src/test/java/org/apache/beam/examples/DeltaLakeToLakehouseCdcExampleTest.java +++ /dev/null @@ -1,255 +0,0 @@ -/* - * 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; - -import static org.hamcrest.MatcherAssert.assertThat; -import static org.hamcrest.Matchers.containsString; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertThrows; - -import java.io.File; -import java.io.IOException; -import java.nio.charset.StandardCharsets; -import java.nio.file.Files; -import java.util.Arrays; -import java.util.Collections; -import java.util.Map; -import org.apache.beam.examples.DeltaLakeToLakehouseCdcExample.LakehouseTableSpec; -import org.apache.beam.examples.DeltaLakeToLakehouseCdcExample.Options; -import org.apache.beam.sdk.options.PipelineOptionsFactory; -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 DeltaLakeToLakehouseCdcExample}. */ -@RunWith(JUnit4.class) -public class DeltaLakeToLakehouseCdcExampleTest { - - @Rule public final TemporaryFolder tmpFolder = new TemporaryFolder(); - - @Test - public void testVerifyChangeDataFeedEnabledPassesWhenTrueInLatestMetaData() throws IOException { - File tableDir = tmpFolder.newFolder("delta_table_enabled"); - File deltaLogDir = new File(tableDir, "_delta_log"); - deltaLogDir.mkdirs(); - - // Commit 0: CDF not enabled initially - Files.write( - new File(deltaLogDir, "00000000000000000000.json").toPath(), - Arrays.asList( - "{\"commitInfo\":{\"timestamp\":1700000000000,\"operation\":\"WRITE\"}}", - "{\"metaData\":{\"id\":\"test-id\",\"configuration\":{}}}"), - StandardCharsets.UTF_8); - - // Commit 1: ALTER TABLE SET TBLPROPERTIES ('delta.enableChangeDataFeed' = 'true') - Files.write( - new File(deltaLogDir, "00000000000000000001.json").toPath(), - Arrays.asList( - "{\"commitInfo\":{\"timestamp\":1700000001000,\"operation\":\"SET TBLPROPERTIES\"}}", - "{\"metaData\":{\"id\":\"test-id\",\"configuration\":{\"delta.enableChangeDataFeed\":\"true\"}}}"), - StandardCharsets.UTF_8); - - // Commit 2: Data commit with no metaData action - Files.write( - new File(deltaLogDir, "00000000000000000002.json").toPath(), - Arrays.asList( - "{\"commitInfo\":{\"timestamp\":1700000002000,\"operation\":\"WRITE\"}}", - "{\"add\":{\"path\":\"part-00000.parquet\",\"size\":1024,\"modificationTime\":1700000002000,\"dataChange\":true}}"), - StandardCharsets.UTF_8); - - DeltaLakeToLakehouseCdcExample.verifyChangeDataFeedEnabled(tableDir.getAbsolutePath()); - } - - @Test - public void testVerifyChangeDataFeedEnabledFailsWhenFalse() throws IOException { - File tableDir = tmpFolder.newFolder("delta_table_disabled"); - File deltaLogDir = new File(tableDir, "_delta_log"); - deltaLogDir.mkdirs(); - - Files.write( - new File(deltaLogDir, "00000000000000000000.json").toPath(), - Arrays.asList( - "{\"commitInfo\":{\"timestamp\":1700000000000,\"operation\":\"WRITE\"}}", - "{\"metaData\":{\"id\":\"test-id\",\"configuration\":{\"delta.enableChangeDataFeed\":\"false\"}}}"), - StandardCharsets.UTF_8); - - IllegalStateException ex = - assertThrows( - IllegalStateException.class, - () -> - DeltaLakeToLakehouseCdcExample.verifyChangeDataFeedEnabled( - tableDir.getAbsolutePath())); - assertThat(ex.getMessage(), containsString("delta.enableChangeDataFeed = true")); - } - - @Test - public void testVerifyChangeDataFeedEnabledFailsWhenMissingFromConfiguration() - throws IOException { - File tableDir = tmpFolder.newFolder("delta_table_missing_prop"); - File deltaLogDir = new File(tableDir, "_delta_log"); - deltaLogDir.mkdirs(); - - Files.write( - new File(deltaLogDir, "00000000000000000000.json").toPath(), - Arrays.asList( - "{\"commitInfo\":{\"timestamp\":1700000000000,\"operation\":\"WRITE\"}}", - "{\"metaData\":{\"id\":\"test-id\",\"configuration\":{}}}"), - StandardCharsets.UTF_8); - - IllegalStateException ex = - assertThrows( - IllegalStateException.class, - () -> - DeltaLakeToLakehouseCdcExample.verifyChangeDataFeedEnabled( - tableDir.getAbsolutePath())); - assertThat(ex.getMessage(), containsString("delta.enableChangeDataFeed = true")); - } - - @Test - public void testVerifyChangeDataFeedEnabledFailsWhenNoLogsExist() throws IOException { - File emptyDir = tmpFolder.newFolder("not_a_delta_table"); - - IllegalArgumentException ex = - assertThrows( - IllegalArgumentException.class, - () -> - DeltaLakeToLakehouseCdcExample.verifyChangeDataFeedEnabled( - emptyDir.getAbsolutePath())); - assertThat(ex.getMessage(), containsString("No Delta Lake transaction log files 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 = DeltaLakeToLakehouseCdcExample.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 = DeltaLakeToLakehouseCdcExample.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); - DeltaLakeToLakehouseCdcExample.validateRangeOptions(versionOpts); - - Options tsOpts = PipelineOptionsFactory.as(Options.class); - tsOpts.setStartTimestamp("2026-10-01T00:00:00Z"); - tsOpts.setEndTimestamp("2026-10-02T00:00:00Z"); - DeltaLakeToLakehouseCdcExample.validateRangeOptions(tsOpts); - - Options missingOpts = PipelineOptionsFactory.as(Options.class); - assertThrows( - IllegalArgumentException.class, - () -> DeltaLakeToLakehouseCdcExample.validateRangeOptions(missingOpts)); - - Options mixedOpts = PipelineOptionsFactory.as(Options.class); - mixedOpts.setStartVersion(1L); - mixedOpts.setStartTimestamp("2026-10-01T00:00:00Z"); - assertThrows( - IllegalArgumentException.class, - () -> DeltaLakeToLakehouseCdcExample.validateRangeOptions(mixedOpts)); - - Options invertedOpts = PipelineOptionsFactory.as(Options.class); - invertedOpts.setStartVersion(5L); - invertedOpts.setEndVersion(2L); - assertThrows( - IllegalArgumentException.class, - () -> DeltaLakeToLakehouseCdcExample.validateRangeOptions(invertedOpts)); - } - - @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 = DeltaLakeToLakehouseCdcExample.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 = DeltaLakeToLakehouseCdcExample.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")); - } -}