Managed Iceberg AddFiles

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
 

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"];
   }
 }
 

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

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();

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<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 static class WriteRegistrar implements TransformPayloadTranslatorRegistra
         getTransformPayloadTranslators() {
       return ImmutableMap.<Class<? extends PTransform>, TransformPayloadTranslator>builder()
           .put(IcebergWriteSchemaTransform.class, new IcebergWriteSchemaTransformTranslator())
+          .put(AddFilesSchemaTransform.class, new AddFilesSchemaTransformTranslator())
           .build();
     }
   }