diff --git a/CHANGES.md b/CHANGES.md index c4cf7e11f133..1fef9780020f 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/apache/beam/issues/37431)). * (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 debacc245d60..d08ef8fe6158 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 c97a20c6221b..06bbee362a65 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 @@ * path becomes a {@link DataFile} with partition metadata and column stats, batched into manifests * and committed as snapshots. * + *

{@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}: + * + * 

{@code
+ * Map 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));
+ * }
+ * *

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

Schema evolution. With a {@link SchemaEvolutionConfig} whose options are set, a + *

Schema evolution. 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 @@ * currently requires bounded input; unbounded input with options set is rejected at construction. * * 

{@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 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));
  * }
* *

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 c5cc758c555b..dced27a3aa2b 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.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.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.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 AddFilesSchemaTransform from(Configuration configuration) { @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 AddFilesSchemaTransform(Configuration configuration) { 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 801464010c69..15beb99a7a2b 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 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 Row toConfigRow(IcebergWriteSchemaTransform transform) { } } + static class AddFilesSchemaTransformTranslator + extends SchemaTransformPayloadTranslator { + @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 static class WriteRegistrar implements TransformPayloadTranslatorRegistra getTransformPayloadTranslators() { return ImmutableMap., 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 7595483d58ac..9449e929a935 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 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 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.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 LAKEHOUSE_PROPS = LakehouseTestCatalog.catalogProperties(); private Storage storage; @@ -477,17 +477,15 @@ public void testBatchParquetImportToTableInAdditionalLocation() throws IOExcepti List 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 @@ record -> } // run batch AddFiles + Map 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 void testBatchParquetImportWithSchemaEvolution() throws IOException { .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 void testBatchImportOfParquetFilesWithoutExtension() throws IOException { 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 void testBatchImportOfRequiredPartitionColumnFromThePath() throws IOExcep } 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 @@ private PipelineResult startAddFilesListener( options.setBlockOnRun(false); Pipeline p = Pipeline.create(options); - PCollectionRowTuple tuple = + PCollection 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 @@ private PipelineResult startAddFilesListener( .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 @@ private PipelineResult startAddFilesListener( Thread.sleep(10_000); return result; } + + /** Registers the paths through the Managed API, with error rows output under "errors". */ + private static PCollectionRowTuple addFiles( + PCollection paths, Map config) { + Map 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 3b58af201543..463ee24ab55f 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 void testSettingsWithoutOptionsRejectedNamingTheOptionsKey() { 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.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 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 3c07c97b4a52..5a49eb7bb7f1 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.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 CATALOG_PROPERTIES = ImmutableMap.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 void testCdcReadTransformProtoTranslation() 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.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 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 27c647478e17..8ba003c94957 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 READ_TRANSFORMS = @@ -124,6 +125,9 @@ DELTA_LAKE_CDC, getUrn(ExternalTransforms.ManagedTransforms.Urns.DELTA_LAKE_CDC_ .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 static ManagedTransform read(String source) { * href="https://beam.apache.org/releases/javadoc/current/org/apache/beam/sdk/io/kafka/KafkaIO.html">KafkaIO *

  • {@link Managed#BIGQUERY} : Write to GCP BigQuery tables BigQueryIO + *
  • {@link Managed#ICEBERG_ADD_FILES} : Register existing data files in Apache Iceberg tables + * using AddFiles * */ 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 33926281074d..5c5dbcd58fdf 100644 --- a/sdks/python/apache_beam/transforms/external.py +++ b/sdks/python/apache_beam/transforms/external.py @@ -88,6 +88,7 @@ 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 c252b10de5db..aeea50de56a9 100644 --- a/sdks/python/apache_beam/transforms/managed.py +++ b/sdks/python/apache_beam/transforms/managed.py @@ -90,6 +90,7 @@ SQL_SERVER = "sqlserver" DELTA = "delta" DELTA_CDC = "delta_cdc" +ICEBERG_ADD_FILES = "iceberg_add_files" __all__ = [ "ICEBERG", @@ -97,6 +98,7 @@ "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 9dc3eff38b1e..a53bedd6b300 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