This is an automated email from the ASF dual-hosted git repository.
ahmedabu98 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new aea45221b65 Managed Iceberg AddFiles (#40827)
aea45221b65 is described below
commit aea45221b65a0c2e6fef2bf5c8f0f6f3b0753dd1
Author: Ahmed Abualsaud <[email protected]>
AuthorDate: Tue Oct 27 16:57:50 2026 -0700
Managed Iceberg AddFiles (#40827)
---
CHANGES.md | 2 +
.../model/pipeline/v1/external_transforms.proto | 2 +
.../org/apache/beam/sdk/io/iceberg/AddFiles.java | 31 +++--
.../iceberg/AddFilesSchemaTransformProvider.java | 21 ++-
.../iceberg/IcebergSchemaTransformTranslation.java | 15 ++
.../org/apache/beam/sdk/io/iceberg/AddFilesIT.java | 153 ++++++++++-----------
.../AddFilesSchemaTransformProviderTest.java | 55 ++++++++
.../IcebergSchemaTransformTranslationTest.java | 68 +++++++++
.../java/org/apache/beam/sdk/managed/Managed.java | 7 +
sdks/python/apache_beam/transforms/external.py | 1 +
sdks/python/apache_beam/transforms/managed.py | 5 +-
sdks/python/apache_beam/yaml/standard_io.yaml | 5 +
12 files changed, 275 insertions(+), 90 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index c4cf7e11f13..1fef9780020 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -68,6 +68,7 @@
* (Java) FirestoreIO: added
`RpcQosOptions.Builder.withRampupThrottlingDisabled()`, mirroring the existing
DatastoreIO option, to opt out of the 500/50/5 write ramp-up
([#27170](https://github.com/apache/beam/issues/27170)).
* Added support for attributing BigQuery API quota and billing to a specific
GCP project (quota project): `quota_project_id` parameter in `ReadFromBigQuery`
or `--quota_project_id` pipeline option, honored by the Python BigQuery IO
transforms (`WriteToBigQuery` with `FILE_LOADS` or `STREAMING_INSERTS`; not yet
`STORAGE_WRITE_API`, which is Java-backed), `--bigQueryQuotaProjectId` pipeline
option (Java), and `bigqueryio.WithQuotaProject` read/query option (Go)
([#37431](https://github.com [...]
* (Java/Python/Go) DebeziumIO: `readAsJson()` can now include `op` and the
remaining Debezium `source` fields (e.g. MySQL `file`, `pos`, `row`, `ts_ms`;
PostgreSQL `lsn`, `txId`, `ts_ms`) in `metadata` via `withFullMetadata(true)` /
`full_metadata=True` / `debeziumio.FullMetadata()`. Off by default
([#40701](https://github.com/apache/beam/issues/40701)).
+* [IcebergIO] AddFiles is now available through the Managed API with
`Managed.write(Managed.ICEBERG_ADD_FILES)` (Java/Python)
([#40827](https://github.com/apache/beam/issues/40827)).
## New Features / Improvements
@@ -100,6 +101,7 @@
* (Python) `--dataflow_endpoint` now defaults to `''` instead of
`https://dataflow.googleapis.com`. When unset, the Google Cloud client library
resolves the Dataflow endpoint and honors `GOOGLE_CLOUD_UNIVERSE_DOMAIN`, so
Dataflow jobs can be submitted in non-default universes. Code that reads
`GoogleCloudOptions.dataflow_endpoint` should handle an empty value
([#40569](https://github.com/apache/beam/issues/40569)).
* (Python) The `xgboost` extras now install `xgboost>=2.0.0,<4` and
`datatable>=1.1.0,<2`, moving both the lower and upper bounds for both. The
previous minimum `xgboost` and `datatable` versions (`1.6.0` and `1.0.0`
respectively) are no longer allowed
([#40694](https://github.com/apache/beam/issues/40694)).
* (Java) [IcebergIO] AddFiles (`IcebergAddFiles` in YAML) now carries the
destination table alongside each file through the registration chain, so that
one transform can register files into many tables. The coders of its internal
`PCollection`s change, which makes this an update incompatible change for
streaming pipelines that use it
([#40817](https://github.com/apache/beam/pull/40817)).
+* (YAML) `IcebergAddFiles` will be automatically replaced with the compatible
managed transform. For older Beam versions, streaming update compatibility can
be maintained by specifying the pipeline option `update_compatibility_version`
([#40827](https://github.com/apache/beam/issues/40827)).
## Deprecations
diff --git
a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto
b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto
index debacc245d6..d08ef8fe615 100644
---
a/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto
+++
b/model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/external_transforms.proto
@@ -109,6 +109,8 @@ message ManagedTransforms {
"beam:schematransform:org.apache.beam:delta_lake_read:v1"];
DELTA_LAKE_CDC_READ = 14 [(org.apache.beam.model.pipeline.v1.beam_urn) =
"beam:schematransform:org.apache.beam:delta_lake_cdc_read:v1"];
+ ICEBERG_ADD_FILES = 15 [(org.apache.beam.model.pipeline.v1.beam_urn) =
+ "beam:schematransform:iceberg_add_files:v1"];
}
}
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java
index c97a20c6221..06bbee362a6 100644
---
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java
@@ -125,11 +125,26 @@ import org.slf4j.LoggerFactory;
* path becomes a {@link DataFile} with partition metadata and column stats,
batched into manifests
* and committed as snapshots.
*
+ * <p>{@link AddFiles} is offered as a Managed transform. Its input is a
{@link PCollection} of
+ * {@link Row}s with a single string field, the file's path, and its options
are listed in {@link
+ * AddFilesSchemaTransformProvider.Configuration}:
+ *
+ * <pre>{@code
+ * Map<String, Object> config = Map.of(
+ * "table", "db.sales",
+ * "catalog_properties", Map.of("type", "rest", "uri", catalogUri),
+ * "error_handling", Map.of("output", "errors"));
+ *
+ * PCollectionRowTuple result =
+ *
paths.apply(Managed.write(Managed.ICEBERG_ADD_FILES).withConfig(config));
+ * }</pre>
+ *
* <p>Outputs: {@code snapshots} (one row per commit), {@code errors} (one row
per file that could
* not be registered: {@code file}, {@code error}), and {@code dry_run_report}
when a dry run is
- * configured.
+ * configured. Through Managed, error rows are only output when {@code
error_handling} is set, under
+ * the name it gives.
*
- * <p><b>Schema evolution.</b> With a {@link SchemaEvolutionConfig} whose
options are set, a
+ * <p><b>Schema evolution.</b> With schema evolution options set ({@link
SchemaEvolutionConfig}), a
* pre-pass reads every Parquet footer, classifies the change each distinct
file schema needs on the
* table (add a column, relax a required column, promote a type), commits the
allowed changes in one
* transaction, and only then registers the files. Manifest entries are
immutable, so this ordering
@@ -145,12 +160,12 @@ import org.slf4j.LoggerFactory;
* currently requires bounded input; unbounded input with options set is
rejected at construction.
*
* <pre>{@code
- * SchemaEvolutionConfig evolution =
- * SchemaEvolutionConfig.builder()
- * .setOptions(EnumSet.of(ALLOW_FIELD_ADDITION, ALLOW_TYPE_PROMOTION))
- * .setRequiredColumns(Collections.singleton("id"))
- * .build();
- * paths.apply(new AddFiles(catalog, "db.sales", null, null, null, null, null,
null, evolution));
+ * Map<String, Object> config = Map.of(
+ * "table", "db.sales",
+ * "catalog_properties", catalogProperties,
+ * "schema_evolution_options", List.of("ALLOW_FIELD_ADDITION",
"ALLOW_TYPE_PROMOTION"),
+ * "required_columns", List.of("id"));
+ * paths.apply(Managed.write(Managed.ICEBERG_ADD_FILES).withConfig(config));
* }</pre>
*
* <p>Without options the table schema is never changed and files register
as-is: columns the table
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java
index c5cc758c555..dced27a3aa2 100644
---
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java
@@ -21,6 +21,7 @@ import static
org.apache.beam.sdk.io.iceberg.AddFiles.ERROR_TAG;
import static org.apache.beam.sdk.io.iceberg.AddFiles.OUTPUT_TAG;
import static
org.apache.beam.sdk.io.iceberg.AddFilesSchemaTransformProvider.Configuration;
import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+import static org.apache.beam.sdk.util.construction.BeamUrns.getUrn;
import com.google.auto.service.AutoService;
import com.google.auto.value.AutoValue;
@@ -30,8 +31,11 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.beam.model.pipeline.v1.ExternalTransforms;
import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.NoSuchSchemaException;
import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.schemas.SchemaRegistry;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
import org.apache.beam.sdk.schemas.annotations.SchemaFieldDescription;
import org.apache.beam.sdk.schemas.transforms.SchemaTransform;
@@ -41,6 +45,7 @@ import
org.apache.beam.sdk.schemas.transforms.providers.ErrorHandling;
import org.apache.beam.sdk.transforms.Filter;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
import org.apache.beam.sdk.values.TypeDescriptors;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -55,7 +60,7 @@ public class AddFilesSchemaTransformProvider extends
TypedSchemaTransformProvide
@Override
public String identifier() {
- return "beam:schematransform:iceberg_add_files:v1";
+ return getUrn(ExternalTransforms.ManagedTransforms.Urns.ICEBERG_ADD_FILES);
}
@DefaultSchema(AutoValueSchema.class)
@@ -302,6 +307,20 @@ public class AddFilesSchemaTransformProvider extends
TypedSchemaTransformProvide
this.configuration = configuration;
}
+ Row getConfigurationRow() {
+ try {
+ // To stay consistent with our SchemaTransform configuration naming
conventions,
+ // we sort lexicographically and convert field names to snake_case
+ return SchemaRegistry.createDefault()
+ .getToRowFunction(Configuration.class)
+ .apply(configuration)
+ .sorted()
+ .toSnakeCase();
+ } catch (NoSuchSchemaException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
@Override
public PCollectionRowTuple expand(PCollectionRowTuple input) {
Schema inputSchema = input.getSinglePCollection().getSchema();
diff --git
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslation.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslation.java
index 801464010c6..15beb99a7a2 100644
---
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslation.java
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslation.java
@@ -23,6 +23,7 @@ import static
org.apache.beam.sdk.schemas.transforms.SchemaTransformTranslation.
import com.google.auto.service.AutoService;
import java.util.Map;
+import
org.apache.beam.sdk.io.iceberg.AddFilesSchemaTransformProvider.AddFilesSchemaTransform;
import
org.apache.beam.sdk.io.iceberg.IcebergCdcReadSchemaTransformProvider.IcebergCdcReadSchemaTransform;
import org.apache.beam.sdk.schemas.transforms.SchemaTransformProvider;
import org.apache.beam.sdk.transforms.PTransform;
@@ -87,6 +88,19 @@ public class IcebergSchemaTransformTranslation {
}
}
+ static class AddFilesSchemaTransformTranslator
+ extends SchemaTransformPayloadTranslator<AddFilesSchemaTransform> {
+ @Override
+ public SchemaTransformProvider provider() {
+ return new AddFilesSchemaTransformProvider();
+ }
+
+ @Override
+ public Row toConfigRow(AddFilesSchemaTransform transform) {
+ return transform.getConfigurationRow();
+ }
+ }
+
@AutoService(TransformPayloadTranslatorRegistrar.class)
public static class WriteRegistrar implements
TransformPayloadTranslatorRegistrar {
@Override
@@ -97,6 +111,7 @@ public class IcebergSchemaTransformTranslation {
getTransformPayloadTranslators() {
return ImmutableMap.<Class<? extends PTransform>,
TransformPayloadTranslator>builder()
.put(IcebergWriteSchemaTransform.class, new
IcebergWriteSchemaTransformTranslator())
+ .put(AddFilesSchemaTransform.class, new
AddFilesSchemaTransformTranslator())
.build();
}
}
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
index 7595483d58a..9449e929a93 100644
---
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
@@ -21,6 +21,7 @@ import static java.lang.String.format;
import static org.apache.beam.sdk.io.FileIO.Write.defaultNaming;
import static
org.apache.beam.sdk.io.iceberg.IcebergUtils.beamSchemaToIcebergSchema;
import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+import static org.apache.beam.sdk.values.TypeDescriptors.rows;
import static org.apache.beam.sdk.values.TypeDescriptors.strings;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.startsWith;
@@ -43,7 +44,6 @@ import com.google.cloud.storage.StorageOptions;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
-import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -96,7 +96,6 @@ import org.apache.iceberg.rest.RESTCatalog;
import org.apache.iceberg.util.Pair;
import org.awaitility.Awaitility;
import org.checkerframework.checker.nullness.qual.Nullable;
-import org.joda.time.Duration;
import org.junit.After;
import org.junit.Before;
import org.junit.Rule;
@@ -133,6 +132,7 @@ public class AddFilesIT {
.addStringField("name")
.addStringField("kind")
.build();
+ private static final Schema PATH_SCHEMA =
Schema.builder().addStringField("path").build();
private static final Map<String, String> LAKEHOUSE_PROPS =
LakehouseTestCatalog.catalogProperties();
private Storage storage;
@@ -477,17 +477,15 @@ public class AddFilesIT {
List<String> writtenFilePaths = writeParquetFiles();
Pipeline p = Pipeline.create();
PCollectionRowTuple tuple =
- p.apply(Create.of(writtenFilePaths))
- .apply(
- new AddFiles(
-
IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(),
- destTableId.toString(),
- null,
- PARTITION_FIELDS,
- null,
- TABLE_PROPS,
- null,
- null));
+ addFiles(
+ p.apply(Create.of(writtenFilePaths)),
+ ImmutableMap.of(
+ "table",
+ destTableId.toString(),
+ "partition_fields",
+ PARTITION_FIELDS,
+ "table_properties",
+ TABLE_PROPS));
PAssert.that(tuple.get("errors")).empty();
p.run().waitUntilFinish();
@@ -582,19 +580,14 @@ public class AddFilesIT {
}
// run batch AddFiles
+ Map<String, Object> config = new HashMap<>();
+ config.put("table", namespace + "." + destTableName);
+ config.put("table_properties", tableProps);
+ if (!isUIT) {
+ config.put("partition_fields", PARTITION_FIELDS);
+ }
Pipeline p = Pipeline.create();
- PCollectionRowTuple tuple =
- p.apply(Create.of(writtenFilePaths))
- .apply(
- new AddFiles(
-
IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(),
- namespace + "." + destTableName,
- null,
- isUIT ? null : PARTITION_FIELDS,
- null,
- tableProps,
- null,
- null));
+ PCollectionRowTuple tuple = addFiles(p.apply(Create.of(writtenFilePaths)),
config);
PAssert.that(tuple.get("errors")).empty();
p.run().waitUntilFinish();
@@ -631,24 +624,17 @@ public class AddFilesIT {
.collect(Collectors.toList());
assertEquals(20, writtenFilePaths.size());
- SchemaEvolutionConfig evolution =
- SchemaEvolutionConfig.builder()
- .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))
- .build();
Pipeline p = Pipeline.create();
PCollectionRowTuple tuple =
- p.apply(Create.of(writtenFilePaths))
- .apply(
- new AddFiles(
-
IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(),
- namespace + "." + destTableName,
- null,
- null,
- null,
- TABLE_PROPS,
- null,
- null,
- evolution));
+ addFiles(
+ p.apply(Create.of(writtenFilePaths)),
+ ImmutableMap.of(
+ "table",
+ namespace + "." + destTableName,
+ "table_properties",
+ TABLE_PROPS,
+ "schema_evolution_options",
+ Arrays.asList("ALLOW_FIELD_ADDITION")));
PAssert.that(tuple.get("errors")).empty();
p.run().waitUntilFinish();
@@ -751,17 +737,10 @@ public class AddFilesIT {
Pipeline p = Pipeline.create();
PCollectionRowTuple tuple =
- p.apply(Create.of(inputs))
- .apply(
- new AddFiles(
-
IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(),
- namespace + "." + destTableName,
- null,
- null,
- null,
- TABLE_PROPS,
- null,
- null));
+ addFiles(
+ p.apply(Create.of(inputs)),
+ ImmutableMap.of(
+ "table", namespace + "." + destTableName, "table_properties",
TABLE_PROPS));
PAssert.that(tuple.get("errors"))
.satisfies(
rows -> {
@@ -810,24 +789,19 @@ public class AddFilesIT {
}
assertEquals(20, files.size());
- SchemaEvolutionConfig evolution =
- SchemaEvolutionConfig.builder()
- .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))
- .build();
Pipeline p = Pipeline.create();
PCollectionRowTuple tuple =
- p.apply(Create.of(files))
- .apply(
- new AddFiles(
-
IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(),
- namespace + "." + destTableName,
- parquetDir,
- null,
- null,
- TABLE_PROPS,
- null,
- null,
- evolution));
+ addFiles(
+ p.apply(Create.of(files)),
+ ImmutableMap.of(
+ "table",
+ namespace + "." + destTableName,
+ "location_prefix",
+ parquetDir,
+ "table_properties",
+ TABLE_PROPS,
+ "schema_evolution_options",
+ Arrays.asList("ALLOW_FIELD_ADDITION")));
PAssert.that(tuple.get("errors")).empty();
p.run().waitUntilFinish();
@@ -969,7 +943,7 @@ public class AddFilesIT {
options.setBlockOnRun(false);
Pipeline p = Pipeline.create(options);
- PCollectionRowTuple tuple =
+ PCollection<String> paths =
p.apply(PubsubIO.readStrings().fromTopic(notificationsTopic))
.apply(JsonToRow.withSchema(NOTIFICATION_SCHEMA))
.apply(Filter.by(row ->
objectNameFilter.apply(row.getString("name"))))
@@ -978,17 +952,21 @@ public class AddFilesIT {
.via(
row ->
format("gs://%s/%s", row.getString("bucket"),
row.getString("name"))))
- .apply(Deduplicate.values())
- .apply(
- new AddFiles(
-
IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(),
- namespace + "." + destTableName,
- null,
- PARTITION_FIELDS,
- null,
- TABLE_PROPS,
- 10,
- Duration.standardSeconds(10)));
+ .apply(Deduplicate.values());
+ PCollectionRowTuple tuple =
+ addFiles(
+ paths,
+ ImmutableMap.of(
+ "table",
+ namespace + "." + destTableName,
+ "partition_fields",
+ PARTITION_FIELDS,
+ "table_properties",
+ TABLE_PROPS,
+ "manifest_file_size",
+ 10,
+ "triggering_frequency_seconds",
+ 10));
PAssert.that(tuple.get("errors")).empty();
PipelineResult result = p.run();
@@ -997,4 +975,19 @@ public class AddFilesIT {
Thread.sleep(10_000);
return result;
}
+
+ /** Registers the paths through the Managed API, with error rows output
under "errors". */
+ private static PCollectionRowTuple addFiles(
+ PCollection<String> paths, Map<String, Object> config) {
+ Map<String, Object> managedConfig = new HashMap<>(config);
+ managedConfig.put("catalog_properties", LAKEHOUSE_PROPS);
+ managedConfig.put("error_handling", ImmutableMap.of("output", "errors"));
+ return paths
+ .apply(
+ "ToRows",
+ MapElements.into(rows())
+ .via(path ->
Row.withSchema(PATH_SCHEMA).addValue(path).build()))
+ .setRowSchema(PATH_SCHEMA)
+
.apply(Managed.write(Managed.ICEBERG_ADD_FILES).withConfig(managedConfig));
+ }
}
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
index 3b58af20154..463ee24ab55 100644
---
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
@@ -17,24 +17,50 @@
*/
package org.apache.beam.sdk.io.iceberg;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
+import java.io.IOException;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumSet;
+import java.util.List;
import
org.apache.beam.sdk.io.iceberg.AddFilesSchemaTransformProvider.Configuration;
+import org.apache.beam.sdk.managed.Managed;
+import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.transforms.providers.ErrorHandling;
+import org.apache.beam.sdk.testing.PAssert;
+import org.apache.beam.sdk.testing.TestPipeline;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PCollectionRowTuple;
+import org.apache.beam.sdk.values.Row;
+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.apache.iceberg.DataFile;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.data.IcebergGenerics;
+import org.apache.iceberg.data.Record;
+import org.hamcrest.Matchers;
+import org.junit.ClassRule;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@RunWith(JUnit4.class)
public class AddFilesSchemaTransformProviderTest {
+ @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new
TemporaryFolder();
+
+ @Rule
+ public transient TestDataWarehouse warehouse = new
TestDataWarehouse(TEMPORARY_FOLDER, "default");
+
+ @Rule public transient TestPipeline pipeline = TestPipeline.create();
private static Configuration.Builder base() {
return Configuration.builder()
@@ -182,4 +208,33 @@ public class AddFilesSchemaTransformProviderTest {
assertTrue(
unverifiable.getMessage(),
unverifiable.getMessage().contains("schema_evolution_options"));
}
+
+ @Test
+ public void testAddFilesUsingManagedTransform() throws IOException {
+ TableIdentifier tableId = TableIdentifier.of("default",
"managed_add_files");
+ DataFile file =
+ warehouse.writeRecords("data.parquet", TestFixtures.SCHEMA,
TestFixtures.FILE1SNAPSHOT1);
+
+ Schema inputSchema = Schema.builder().addStringField("file").build();
+ PCollectionRowTuple output =
+ pipeline
+
.apply(Create.of(Row.withSchema(inputSchema).addValue(file.location()).build()))
+ .setRowSchema(inputSchema)
+ .apply(
+ Managed.write(Managed.ICEBERG_ADD_FILES)
+ .withConfig(
+ ImmutableMap.<String, Object>builder()
+ .put("table", tableId.toString())
+ .put(
+ "catalog_properties",
+ ImmutableMap.of("type", "hadoop", "warehouse",
warehouse.location))
+ .put("error_handling", ImmutableMap.of("output",
"errors"))
+ .build()));
+ PAssert.that(output.get("errors")).empty();
+ pipeline.run().waitUntilFinish();
+
+ Table table = warehouse.loadTable(tableId);
+ List<Record> records =
ImmutableList.copyOf(IcebergGenerics.read(table).build());
+ assertThat(records,
Matchers.containsInAnyOrder(TestFixtures.FILE1SNAPSHOT1.toArray()));
+ }
}
diff --git
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslationTest.java
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslationTest.java
index 3c07c97b4a5..5a49eb7bb7f 100644
---
a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslationTest.java
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/IcebergSchemaTransformTranslationTest.java
@@ -34,8 +34,11 @@ import
org.apache.beam.model.pipeline.v1.ExternalTransforms.SchemaTransformPaylo
import org.apache.beam.model.pipeline.v1.RunnerApi;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.coders.RowCoder;
+import
org.apache.beam.sdk.io.iceberg.AddFilesSchemaTransformProvider.AddFilesSchemaTransform;
import
org.apache.beam.sdk.io.iceberg.IcebergCdcReadSchemaTransformProvider.IcebergCdcReadSchemaTransform;
+import
org.apache.beam.sdk.io.iceberg.IcebergSchemaTransformTranslation.AddFilesSchemaTransformTranslator;
import
org.apache.beam.sdk.io.iceberg.IcebergSchemaTransformTranslation.IcebergCdcReadSchemaTransformTranslator;
+import org.apache.beam.sdk.managed.Managed;
import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.SchemaTranslation;
@@ -74,6 +77,8 @@ public class IcebergSchemaTransformTranslationTest {
new IcebergReadSchemaTransformProvider();
static final IcebergCdcReadSchemaTransformProvider READ_CDC_PROVIDER =
new IcebergCdcReadSchemaTransformProvider();
+ static final AddFilesSchemaTransformProvider ADD_FILES_PROVIDER =
+ new AddFilesSchemaTransformProvider();
private static final Map<String, String> CATALOG_PROPERTIES =
ImmutableMap.<String, String>builder()
@@ -133,6 +138,15 @@ public class IcebergSchemaTransformTranslationTest {
.withFieldValue("watermark_column", "event_micros")
.build();
+ private static final Row ADD_FILES_CONFIG_ROW =
+ Row.withSchema(ADD_FILES_PROVIDER.configurationSchema())
+ .withFieldValue("table", "test_table_identifier")
+ .withFieldValue("catalog_properties", CATALOG_PROPERTIES)
+ .withFieldValue("partition_fields", Collections.singletonList("id"))
+ .withFieldValue("manifest_file_size", 100)
+ .withFieldValue("error_handling.output", "errors")
+ .build();
+
@Test
public void testReCreateWriteTransformFromRow() {
IcebergWriteSchemaTransform writeTransform =
@@ -374,6 +388,60 @@ public class IcebergSchemaTransformTranslationTest {
assertEquals(transformConfigRow,
readTransformFromSpec.getConfigurationRow());
}
+ @Test
+ public void testAddFilesTransformProtoTranslation()
+ throws InvalidProtocolBufferException, IOException {
+ // First build a pipeline that adds files through the Managed API
+ Pipeline p = Pipeline.create();
+ Schema inputSchema = Schema.builder().addStringField("file").build();
+
p.apply(Create.of(Row.withSchema(inputSchema).addValue("data.parquet").build()))
+ .setRowSchema(inputSchema)
+ .apply(
+ Managed.write(Managed.ICEBERG_ADD_FILES)
+ .withConfig(
+ ImmutableMap.<String, Object>builder()
+ .put("table", "test_table_identifier")
+ .put("catalog_properties", CATALOG_PROPERTIES)
+ .put("partition_fields",
Collections.singletonList("id"))
+ .put("manifest_file_size", 100)
+ .put("error_handling", ImmutableMap.of("output",
"errors"))
+ .build()));
+
+ // Then translate the pipeline to a proto and extract
AddFilesSchemaTransform proto
+ RunnerApi.Pipeline pipelineProto = PipelineTranslation.toProto(p);
+ List<RunnerApi.PTransform> addFilesTransformProto =
+ pipelineProto.getComponents().getTransformsMap().values().stream()
+ .filter(
+ tr -> {
+ RunnerApi.FunctionSpec spec = tr.getSpec();
+ try {
+ return
spec.getUrn().equals(BeamUrns.getUrn(SCHEMA_TRANSFORM))
+ && SchemaTransformPayload.parseFrom(spec.getPayload())
+ .getIdentifier()
+ .equals(ADD_FILES_PROVIDER.identifier());
+ } catch (InvalidProtocolBufferException e) {
+ throw new RuntimeException(e);
+ }
+ })
+ .collect(Collectors.toList());
+ assertEquals(1, addFilesTransformProto.size());
+ RunnerApi.FunctionSpec spec = addFilesTransformProto.get(0).getSpec();
+
+ // Check that the proto contains correct values
+ SchemaTransformPayload payload =
SchemaTransformPayload.parseFrom(spec.getPayload());
+ Schema schemaFromSpec =
SchemaTranslation.schemaFromProto(payload.getConfigurationSchema());
+ assertEquals(ADD_FILES_PROVIDER.configurationSchema(), schemaFromSpec);
+ Row rowFromSpec =
RowCoder.of(schemaFromSpec).decode(payload.getConfigurationRow().newInput());
+ assertEquals(ADD_FILES_CONFIG_ROW, rowFromSpec);
+
+ // Use the information in the proto to recreate the AddFilesSchemaTransform
+ AddFilesSchemaTransformTranslator translator = new
AddFilesSchemaTransformTranslator();
+ AddFilesSchemaTransform transformFromSpec =
+ translator.fromConfigRow(rowFromSpec, PipelineOptionsFactory.create());
+
+ assertEquals(ADD_FILES_CONFIG_ROW,
transformFromSpec.getConfigurationRow());
+ }
+
private static Record record(long id, String data, long eventMicros) {
return TestFixtures.createRecord(
CDC_TRANSLATION_SCHEMA,
diff --git
a/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java
b/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java
index 27c647478e1..8ba003c9495 100644
--- a/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java
+++ b/sdks/java/managed/src/main/java/org/apache/beam/sdk/managed/Managed.java
@@ -101,6 +101,7 @@ public class Managed {
public static final String POSTGRES = "postgres";
public static final String MYSQL = "mysql";
public static final String SQL_SERVER = "sqlserver";
+ public static final String ICEBERG_ADD_FILES = "iceberg_add_files";
// Supported SchemaTransforms
public static final Map<String, String> READ_TRANSFORMS =
@@ -124,6 +125,9 @@ public class Managed {
.put(POSTGRES,
getUrn(ExternalTransforms.ManagedTransforms.Urns.POSTGRES_WRITE))
.put(MYSQL,
getUrn(ExternalTransforms.ManagedTransforms.Urns.MYSQL_WRITE))
.put(SQL_SERVER,
getUrn(ExternalTransforms.ManagedTransforms.Urns.SQL_SERVER_WRITE))
+ .put(
+ ICEBERG_ADD_FILES,
+
getUrn(ExternalTransforms.ManagedTransforms.Urns.ICEBERG_ADD_FILES))
.build();
/**
@@ -168,6 +172,9 @@ public class Managed {
*
href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/io/kafka/KafkaIO.html";>KafkaIO</a>
* <li>{@link Managed#BIGQUERY} : Write to GCP BigQuery tables <a
*
href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIO.html";>BigQueryIO</a>
+ * <li>{@link Managed#ICEBERG_ADD_FILES} : Register existing data files in
Apache Iceberg tables
+ * using <a
+ *
href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/io/iceberg/AddFiles.html";>AddFiles</a>
* </ul>
*/
public static ManagedTransform write(String sink) {
diff --git a/sdks/python/apache_beam/transforms/external.py
b/sdks/python/apache_beam/transforms/external.py
index 33926281074..5c5dbcd58fd 100644
--- a/sdks/python/apache_beam/transforms/external.py
+++ b/sdks/python/apache_beam/transforms/external.py
@@ -88,6 +88,7 @@ MANAGED_TRANSFORM_URN_TO_JAR_TARGET_MAPPING = {
ManagedTransforms.Urns.SQL_SERVER_WRITE.urn:
_GCP_EXPANSION_SERVICE_JAR_TARGET, # pylint: disable=line-too-long
ManagedTransforms.Urns.DELTA_LAKE_READ.urn:
_IO_EXPANSION_SERVICE_JAR_TARGET,
ManagedTransforms.Urns.DELTA_LAKE_CDC_READ.urn:
_IO_EXPANSION_SERVICE_JAR_TARGET,
+ ManagedTransforms.Urns.ICEBERG_ADD_FILES.urn:
_IO_EXPANSION_SERVICE_JAR_TARGET, # pylint: disable=line-too-long
}
diff --git a/sdks/python/apache_beam/transforms/managed.py
b/sdks/python/apache_beam/transforms/managed.py
index c252b10de5d..aeea50de56a 100644
--- a/sdks/python/apache_beam/transforms/managed.py
+++ b/sdks/python/apache_beam/transforms/managed.py
@@ -90,6 +90,7 @@ MYSQL = "mysql"
SQL_SERVER = "sqlserver"
DELTA = "delta"
DELTA_CDC = "delta_cdc"
+ICEBERG_ADD_FILES = "iceberg_add_files"
__all__ = [
"ICEBERG",
@@ -97,6 +98,7 @@ __all__ = [
"BIGQUERY",
"DELTA",
"DELTA_CDC",
+ "ICEBERG_ADD_FILES",
"Read",
"Write",
]
@@ -165,7 +167,8 @@ class Write(PTransform):
BIGQUERY: ManagedTransforms.Urns.BIGQUERY_WRITE.urn,
POSTGRES: ManagedTransforms.Urns.POSTGRES_WRITE.urn,
MYSQL: ManagedTransforms.Urns.MYSQL_WRITE.urn,
- SQL_SERVER: ManagedTransforms.Urns.SQL_SERVER_WRITE.urn
+ SQL_SERVER: ManagedTransforms.Urns.SQL_SERVER_WRITE.urn,
+ ICEBERG_ADD_FILES: ManagedTransforms.Urns.ICEBERG_ADD_FILES.urn,
}
def __init__(
diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml
b/sdks/python/apache_beam/yaml/standard_io.yaml
index 9dc3eff38b1..a53bedd6b30 100644
--- a/sdks/python/apache_beam/yaml/standard_io.yaml
+++ b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -713,6 +713,11 @@
'IcebergAddFiles': 'beam:schematransform:iceberg_add_files:v1'
config:
gradle_target: 'sdks:java:io:expansion-service:shadowJar'
+ managed_replacement:
+ # Following transforms may be replaced with equivalent managed
transforms,
+ # if the pipelines 'updateCompatibilityBeamVersion' match the
provided
+ # version.
+ 'IcebergAddFiles': '2.78.0'
#Datadog
- type: renaming
