claudevdm · Oct 27, 2026 · Oct 27, 2026 · Oct 27, 2026 · Oct 27, 2026
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
@@ -1,4 +1,4 @@
 {
     "comment": "Modify this file in a trivial way to cause this test suite to run.",
-    "modification": 3
+    "modification": 4
 }
diff --git 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)).
+* (Java) IcebergIO: `AddFiles` can register files in several tables in one transform, naming each file's table from its path. Set `path_pattern` (`withPathPattern` in Java) to a glob with `{name}` placeholders, such as `gs://lake/export/{table}/**`, and make `table` a template over what it captures, such as `default.{table}`. The schema pre-pass then runs per table, so a table whose files conflict is refused and its files reported while every other table is still registered. Requires `schema_evolution_options` and `error_handling`; not supported with `dry_run`. Error rows gain a `table` field only when `path_pattern` is set, so an existing pipeline's error output is unchanged ([#40849](https://github.com/apache/beam/issues/40849)).
 
 ## New Features / Improvements
 

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
@@ -56,10 +56,12 @@
 import org.apache.beam.sdk.transforms.Combine;
 import org.apache.beam.sdk.transforms.Create;
 import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.Flatten;
 import org.apache.beam.sdk.transforms.GroupIntoBatches;
 import org.apache.beam.sdk.transforms.PTransform;
 import org.apache.beam.sdk.transforms.ParDo;
 import org.apache.beam.sdk.transforms.Values;
+import org.apache.beam.sdk.transforms.View;
 import org.apache.beam.sdk.transforms.Wait;
 import org.apache.beam.sdk.transforms.WithKeys;
 import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
@@ -68,8 +70,10 @@
 import org.apache.beam.sdk.transforms.windowing.Window;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
 import org.apache.beam.sdk.values.PCollectionRowTuple;
 import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PCollectionView;
 import org.apache.beam.sdk.values.Row;
 import org.apache.beam.sdk.values.TupleTag;
 import org.apache.beam.sdk.values.TupleTagList;
@@ -179,10 +183,24 @@ public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTupl
   private static final int DEFAULT_MAX_MANIFESTS_PER_SNAPSHOT = 100;
   static final Schema ERROR_SCHEMA =
       Schema.builder().addStringField("file").addStringField("error").build();
+
+  /**
+   * Error rows from the dynamic path, which names the table the file was headed for; empty when
+   * none could be determined. Single-table mode keeps {@link #ERROR_SCHEMA} as it was, so adding
+   * dynamic destinations does not change the shape of an existing pipeline's error output.
+   */
+  static final Schema DYNAMIC_ERROR_SCHEMA =
+      Schema.builder()
+          .addStringField("file")
+          .addStringField("error")
+          .addStringField("table")
+          .build();
+
   private static final long MANIFEST_PREFIX = UUID.randomUUID().getMostSignificantBits();
   private final IcebergCatalogConfig catalogConfig;
   private final String tableIdentifier;
   private @Nullable Duration intervalTrigger;
+  private @Nullable String pathPattern;
   private final int manifestFileSize;
   private final @Nullable String locationPrefix;
   private final @Nullable List<String> partitionFields;
@@ -246,6 +264,20 @@ public AddFiles(
     this.locationPrefix = locationPrefix;
   }
 
+  /**
+   * Registers each file in the table its own path names. The pattern is a glob with {@code {name}}
+   * placeholders, and {@code table} becomes a template over what it captures.
+   *
+   * <p>Both checks run here rather than per file: a mistyped pattern or a template naming something
+   * the pattern never captures is one exception at submit time, not one error row per file.
+   */
+  public AddFiles withPathPattern(String pathPattern) {
+    PathPattern compiled = PathPattern.compile(pathPattern);
+    compiled.checkRenderable(tableIdentifier);
+    this.pathPattern = pathPattern;
+    return this;
+  }
+
   @Override
   public PCollectionRowTuple expand(PCollection<String> input) {
     if (input.isBounded().equals(UNBOUNDED)) {
@@ -269,35 +301,77 @@ public PCollectionRowTuple expand(PCollection<String> input) {
           "AddFiles configured to build partition metadata after the prefix: '{}'", locationPrefix);
     }
 
-    PCollection<String> paths = input;
-    if (evolution.isEnabled()) {
-      // one commit per window, and for bounded input the window is the whole input
-      PCollection<String> windowed =
-          input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
-      PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas = distinctSchemas(windowed);
-      CommitSchemaUnion.Settings settings =
-          new CommitSchemaUnion.Settings(
-              evolution,
-              evolution.incompatibleSchemaHandlingFor(input.isBounded()),
-              new CommitSchemaUnion.NewTableSettings(partitionFields, sortFields, tableProps));
-      if (evolution.getDryRun()) {
-        return report(schemas, settings);
+    CommitSchemaUnion.Settings settings =
+        new CommitSchemaUnion.Settings(
+            evolution,
+            evolution.incompatibleSchemaHandlingFor(input.isBounded()),
+            new CommitSchemaUnion.NewTableSettings(partitionFields, sortFields, tableProps));
+
+    // every stage that can reject a file appends here; one Flatten publishes them together
+    List<PCollection<Row>> errorSources = new ArrayList<>();
+
+    PCollection<KV<String, String>> keyedPaths;
+
+    if (pathPattern != null) {
+      checkState(
+          evolution.isEnabled(),
+          "A path pattern needs schema evolution: the pre-pass is what decides each table.");
+      checkState(!evolution.getDryRun(), "A dry run of a path pattern is not supported yet.");
+
+      PCollectionTuple routed =
+          input.apply(
+              "RouteFiles",
+              ParDo.of(new RouteFiles(checkStateNotNull(pathPattern), tableIdentifier))
+                  .withOutputTags(RouteFiles.ROUTED, TupleTagList.of(RouteFiles.ERRORS)));
+      errorSources.add(routed.get(RouteFiles.ERRORS).setRowSchema(DYNAMIC_ERROR_SCHEMA));
+
+      PCollection<KV<String, String>> windowed =
+          routed
+              .get(RouteFiles.ROUTED)
+              .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()))
+              .apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
+
+      PCollectionView<Map<String, SchemaCommitVerdict>> verdicts =
+          verdictView(distinctSchemasPerTable(windowed), settings);
+
+      PCollectionTuple gated =
+          windowed.apply(
+              "TableGate",
+              ParDo.of(new TableGate(verdicts))
+                  .withSideInputs(verdicts)
+                  .withOutputTags(TableGate.RELEASED, TupleTagList.of(TableGate.ERRORS)));
+      errorSources.add(gated.get(TableGate.ERRORS).setRowSchema(DYNAMIC_ERROR_SCHEMA));
+      keyedPaths =
+          gated
+              .get(TableGate.RELEASED)
+              .setCoder(KvCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
+    } else {
+      PCollection<String> paths = input;
+      if (evolution.isEnabled()) {
+        // one commit per window, and for bounded input the window is the whole input
+        PCollection<String> windowed =
+            input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
+        PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas = distinctSchemas(windowed);
+        if (evolution.getDryRun()) {
+          return report(schemas, settings);
+        }
+        PCollection<SchemaCommitVerdict> committed = commitSchema(schemas, settings);
+        paths = windowed.apply("WaitForSchemaCommit", Wait.on(committed));
       }
-      PCollection<Long> committed = commitSchema(schemas, settings);
-      paths = windowed.apply("WaitForSchemaCommit", Wait.on(committed));
-    }
 
-    PCollection<KV<String, String>> keyedPaths =
-        paths.apply(
-            "KeyByTable",
-            WithKeys.<String, String>of(tableIdentifier).withKeyType(TypeDescriptors.strings()));
+      keyedPaths =
+          paths.apply(
+              "KeyByTable",
+              WithKeys.<String, String>of(tableIdentifier).withKeyType(TypeDescriptors.strings()));
+    }
 
     PCollectionTuple dataFiles =
         keyedPaths.apply(
             "ConvertToDataFiles",
             ParDo.of(
                     new ConvertToDataFile(
                         catalogConfig,
+                        pathPattern != null,
                         locationPrefix,
                         partitionFields,
                         sortFields,
@@ -358,8 +432,15 @@ public void process(
             .apply("CommitManifests", ParDo.of(new CommitManifestFilesDoFn(catalogConfig)))
             .setRowSchema(SnapshotInfo.getSchema());
 
-    return PCollectionRowTuple.of(
-        OUTPUT_TAG, snapshots, ERROR_TAG, dataFiles.get(ERRORS).setRowSchema(ERROR_SCHEMA));
+    errorSources.add(
+        dataFiles
+            .get(ERRORS)
+            .setRowSchema(pathPattern == null ? ERROR_SCHEMA : DYNAMIC_ERROR_SCHEMA));
+
+    PCollection<Row> errors =
+        PCollectionList.of(errorSources).apply("FlattenErrors", Flatten.pCollections());
+
+    return PCollectionRowTuple.of(OUTPUT_TAG, snapshots, ERROR_TAG, errors);
   }
 
   /** The footer read is keyed, so the single-table path supplies its one table as the key. */
@@ -384,8 +465,33 @@ private PCollection<List<CollectDistinctSchemas.SchemaGroup>> distinctSchemas(
             Combine.globally(new CollectDistinctSchemas()).withoutDefaults());
   }
 
+  /** One schema union per table: the same CombineFn, lifted to a per-key combine. */
+  private PCollection<KV<String, List<CollectDistinctSchemas.SchemaGroup>>> distinctSchemasPerTable(
+      PCollection<KV<String, String>> routed) {
+    return fileSchemas(routed)
+        .apply("CollectDistinctSchemasPerTable", Combine.perKey(new CollectDistinctSchemas()));
+  }
+
+  /** The verdicts, as a map the gate can look a table up in. */
+  private PCollectionView<Map<String, SchemaCommitVerdict>> verdictView(
+      PCollection<KV<String, List<CollectDistinctSchemas.SchemaGroup>>> schemas,
+      CommitSchemaUnion.Settings settings) {
+    SchemaCoder<SchemaCommitVerdict> verdictCoder;
+    try {
+      verdictCoder = SchemaRegistry.createDefault().getSchemaCoder(SchemaCommitVerdict.class);
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+    return schemas
+        .apply(
+            "CommitSchemaPerTable",
+            ParDo.of(new CommitSchemaPerTable(catalogConfig, settings, committer)))
+        .setCoder(KvCoder.of(StringUtf8Coder.of(), verdictCoder))
+        .apply("VerdictsByTable", View.asMap());
+  }
+
   /** Commits the plan for the window's schemas once; the signal releases the gated paths. */
-  private PCollection<Long> commitSchema(
+  private PCollection<SchemaCommitVerdict> commitSchema(
       PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas,
       CommitSchemaUnion.Settings settings) {
     return schemas.apply(
@@ -449,6 +555,7 @@ AddFiles withSchemaCommitter(CommitSchemaUnion.Committer committer) {
   static class ConvertToDataFile
       extends DoFn<KV<String, String>, KV<String, SerializableDataFile>> {
     private final IcebergCatalogConfig catalogConfig;
+    private final boolean dynamic;
     public static final TupleTag<Row> ERRORS = new TupleTag<>();
     public static final TupleTag<KV<String, SerializableDataFile>> DATA_FILES = new TupleTag<>();
     private final @Nullable String prefix;
@@ -484,12 +591,14 @@ private TableState stateFor(String tableName) {
 
     public ConvertToDataFile(
         IcebergCatalogConfig catalogConfig,
+        boolean dynamic,
         @Nullable String prefix,
         @Nullable List<String> partitionFields,
         @Nullable List<String> sortFields,
         @Nullable Map<String, String> tableProps) {
       this(
           catalogConfig,
+          dynamic,
           prefix,
           partitionFields,
           sortFields,
@@ -499,12 +608,14 @@ public ConvertToDataFile(
 
     public ConvertToDataFile(
         IcebergCatalogConfig catalogConfig,
+        boolean dynamic,
         @Nullable String prefix,
         @Nullable List<String> partitionFields,
         @Nullable List<String> sortFields,
         @Nullable Map<String, String> tableProps,
         SchemaEvolutionConfig evolution) {
       this.catalogConfig = catalogConfig;
+      this.dynamic = dynamic;
       this.prefix = prefix;
       this.partitionFields = partitionFields;
       this.sortFields = sortFields;
@@ -1020,21 +1131,21 @@ private boolean firstTime(String tableName, String key) {
       return checkStateNotNull(warned).add(tableName + ":" + key);
     }
 
-    private static ProcessResult errorResult(
+    private ProcessResult errorResult(
         String tableName,
         String filePath,
         String message,
         Instant timestamp,
         BoundedWindow window,
         PaneInfo pane) {
-      return new ProcessResult(
-          tableName,
-          null,
-          Row.withSchema(ERROR_SCHEMA).addValues(filePath, message).build(),
-          null,
-          timestamp,
-          window,
-          pane);
+      Row errorRow;
+      if (dynamic) {
+        errorRow =
+            Row.withSchema(DYNAMIC_ERROR_SCHEMA).addValues(filePath, message, tableName).build();
+      } else {
+        errorRow = Row.withSchema(ERROR_SCHEMA).addValues(filePath, message).build();
+      }
+      return new ProcessResult(tableName, null, errorRow, null, timestamp, window, pane);
     }
 
     static <W, T> T transformValue(Transform<W, T> transform, Type type, ByteBuffer bytes) {