AddFiles: Add dry_run

.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": 7
+    "modification": 8
 }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java

@@ -42,9 +42,9 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.coders.RowCoder;
 import org.apache.beam.sdk.coders.VarIntCoder;
 import org.apache.beam.sdk.coders.VarLongCoder;
-import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
 import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.UnverifiableFileHandling;
 import org.apache.beam.sdk.metrics.Counter;
 import org.apache.beam.sdk.schemas.Schema;
@@ -54,6 +54,7 @@
 import org.apache.beam.sdk.state.StateSpecs;
 import org.apache.beam.sdk.state.ValueState;
 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.GroupIntoBatches;
 import org.apache.beam.sdk.transforms.PTransform;
@@ -123,7 +124,8 @@
  * and committed as snapshots.
  *
  * <p>Outputs: {@code snapshots} (one row per commit), {@code errors} (one row per file that could
- * not be registered: {@code file}, {@code error}).
+ * not be registered: {@code file}, {@code error}), and {@code dry_run_report} when a dry run is
+ * configured.
  *
  * <p><b>Schema evolution.</b> With a {@link SchemaEvolutionConfig} whose options are set, a
  * pre-pass reads every Parquet footer, classifies the change each distinct file schema needs on the
@@ -154,6 +156,10 @@
 public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTuple> {
   static final String OUTPUT_TAG = "snapshots";
   static final String ERROR_TAG = "errors";
+
+  /** Only present with {@link SchemaEvolutionConfig#getDryRun()}. */
+  static final String DRY_RUN_TAG = "dry_run_report";
+
   private static final Duration DEFAULT_TRIGGER_INTERVAL = Duration.standardMinutes(10);
   private static final Counter numManifestFilesAdded =
       counter(AddFiles.class, "numManifestFilesAdded");
@@ -260,7 +266,20 @@ public PCollectionRowTuple expand(PCollection<String> input) {
 
     PCollection<String> paths = input;
     if (evolution.isEnabled()) {
-      paths = gateOnSchemaCommit(input);
+      // 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);
+      }
+      PCollection<Long> committed = commitSchema(schemas, settings);
+      paths = windowed.apply("WaitForSchemaCommit", Wait.on(committed));
     }
 
     PCollectionTuple dataFiles =
@@ -320,39 +339,49 @@ public PCollectionRowTuple expand(PCollection<String> input) {
         OUTPUT_TAG, snapshots, ERROR_TAG, dataFiles.get(ERRORS).setRowSchema(ERROR_SCHEMA));
   }
 
-  /**
-   * Holds every path until the schema commit has landed. The pre-pass window is the unit of commit,
-   * and for bounded input that unit is the whole input: paths are rewindowed into the global window
-   * first, so one commit covers everything and the Wait.on signal lines up with the main input
-   * whatever windowing the caller applied upstream.
-   */
-  private PCollection<String> gateOnSchemaCommit(PCollection<String> input) {
-    PCollection<String> windowed =
-        input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
-    CommitSchemaUnion.TableCreation creation =
-        new CommitSchemaUnion.TableCreation(partitionFields, sortFields, tableProps);
-    // unset handling follows the mode (fail the job in batch, route in streaming); the pre-pass
-    // only runs on bounded input, see expand
-    IncompatibleSchemaHandling onIncompatible =
-        evolution.incompatibleSchemaHandlingFor(input.isBounded());
-    PCollection<Long> signal =
-        windowed
-            .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema()))
-            .setCoder(CollectDistinctSchemas.groupCoder())
-            .apply(
-                "CollectDistinctSchemas",
-                Combine.globally(new CollectDistinctSchemas()).withoutDefaults())
+  private PCollection<List<CollectDistinctSchemas.SchemaGroup>> distinctSchemas(
+      PCollection<String> windowed) {
+    return windowed
+        .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema(evolution)))
+        .setCoder(CollectDistinctSchemas.groupCoder())
+        .apply(
+            "CollectDistinctSchemas",
+            Combine.globally(new CollectDistinctSchemas()).withoutDefaults());
+  }
+
+  /** Commits the plan for the window's schemas once; the signal releases the gated paths. */
+  private PCollection<Long> commitSchema(
+      PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas,
+      CommitSchemaUnion.Settings settings) {
+    return schemas.apply(
+        "CommitSchemaOnce",
+        ParDo.of(new CommitSchemaOnce(catalogConfig, tableIdentifier, settings, committer)));
+  }
+
+  /** Reports the plan for the window's schemas instead; nothing is committed or registered. */
+  private PCollectionRowTuple report(
+      PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas,
+      CommitSchemaUnion.Settings settings) {
+    PCollection<Row> report =
+        schemas
             .apply(
-                "CommitSchemaOnce",
-                ParDo.of(
-                    new CommitSchemaOnce(
-                        catalogConfig,
-                        tableIdentifier,
-                        evolution,
-                        onIncompatible,
-                        creation,
-                        committer)));
-    return windowed.apply("WaitForSchemaCommit", Wait.on(signal));
+                "DryRunReport",
+                ParDo.of(new DryRunReport(catalogConfig, tableIdentifier, settings)))
+            .setRowSchema(DryRunReport.REPORT_SCHEMA);
+
+    PCollection<Row> emptySnapshots =
+        schemas
+            .getPipeline()
+            .apply("NoSnapshots", Create.empty(RowCoder.of(SnapshotInfo.getSchema())))
+            .setRowSchema(SnapshotInfo.getSchema());
+    PCollection<Row> emptyErrors =
+        schemas
+            .getPipeline()
+            .apply("NoErrors", Create.empty(RowCoder.of(ERROR_SCHEMA)))
+            .setRowSchema(ERROR_SCHEMA);
+    return PCollectionRowTuple.of(OUTPUT_TAG, emptySnapshots)
+        .and(ERROR_TAG, emptyErrors)
+        .and(DRY_RUN_TAG, report);
   }
 
   /** Test hook: how the schema pre-pass commits its transaction. */

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java

@@ -120,7 +120,8 @@ public static Builder builder() {
     public abstract @Nullable List<String> getSortFields();
 
     @SchemaFieldDescription(
-        "Lets the transform change the table schema so that every file's columns are covered."
+        "Lets the transform change the table schema so that the table has a column for every"
+            + " column the files have."
             + " Values: ALLOW_FIELD_ADDITION (columns a file has and the table lacks are added, as"
             + " optional), ALLOW_FIELD_RELAXATION (a required table column becomes optional when a"
             + " file lacks it or may hold nulls in it), ALLOW_TYPE_PROMOTION (a column type is"
@@ -145,6 +146,19 @@ public static Builder builder() {
             + " Requires schema_evolution_options.")
     public abstract @Nullable List<String> getRequiredColumns();
 
+    @SchemaFieldDescription(
+        "When true, nothing is committed or registered: the transform reads the files' schemas"
+            + " and emits a dry_run_report output with one row that describes what a real run"
+            + " would do. Its allowed field is true when every file schema can be merged and the"
+            + " configuration raises no problem; otherwise its reason field says what a real run"
+            + " would do about it (fail, or route the files to the error output). Its schemas"
+            + " field lists each distinct file schema with the changes a real run would make for"
+            + " it and, when it cannot be merged, why. The output only exists when this is set;"
+            + " consume it as input: <this transform's name>.dry_run_report. Against a missing"
+            + " table, a REST catalog needs table-create permission even though no table is"
+            + " created.")
+    public abstract @Nullable Boolean getDryRun();
+
     @SchemaFieldDescription(
         "What happens when a file's schema cannot be made to fit the table: it needs a change"
             + " that is not allowed, or it conflicts with the table or with another file."
@@ -158,8 +172,9 @@ public static Builder builder() {
             + " Parquet footers only), or a Parquet file with no null-count statistics for a"
             + " required column (statistics disabled by the writer, or a column under a list or"
             + " map). REJECT (the default) sends the file to the error output (see"
-            + " error_handling). ACCEPT registers it without checks, counted and logged. A file"
-            + " that fails a check is always sent to the error output. An accepted file that"
+            + " error_handling). ACCEPT registers it without the checks; such files are counted"
+            + " and logged. A file that fails a check is always sent to the error output. An"
+            + " accepted file that"
             + " lacks a required column, or holds nulls in it, makes reads of the table fail.")
     public abstract @Nullable String getUnverifiableFileHandling();
 
@@ -198,6 +213,8 @@ public abstract static class Builder {
 
       public abstract Builder setUnverifiableFileHandling(String handling);
 
+      public abstract Builder setDryRun(Boolean dryRun);
+
       public abstract Configuration build();
     }
 
@@ -207,19 +224,21 @@ public abstract static class Builder {
       List<String> pins = getRequiredColumns();
       String handlingName = getIncompatibleSchemaHandling();
       String unverifiableName = getUnverifiableFileHandling();
+      boolean dryRun = Boolean.TRUE.equals(getDryRun());
       boolean nothingSet =
           (optionNames == null || optionNames.isEmpty())
               && (pins == null || pins.isEmpty())
               && handlingName == null
-              && unverifiableName == null;
+              && unverifiableName == null
+              && !dryRun;
       if (nothingSet) {
         return null;
       }
       // SchemaEvolutionConfig.build() checks this too; this copy names the YAML keys
       Preconditions.checkArgument(
           optionNames != null && !optionNames.isEmpty(),
-          "required_columns, incompatible_schema_handling and unverifiable_file_handling need at"
-              + " least one schema_evolution_options entry");
+          "required_columns, incompatible_schema_handling, unverifiable_file_handling and"
+              + " dry_run need at least one schema_evolution_options entry");
       Set<SchemaEvolutionOption> options = EnumSet.noneOf(SchemaEvolutionOption.class);
       for (String name : checkStateNotNull(optionNames)) {
         options.add(parseEnum(SchemaEvolutionOption.class, name, "schema_evolution_options"));
@@ -228,6 +247,7 @@ public abstract static class Builder {
       if (pins != null) {
         builder = builder.setRequiredColumns(new LinkedHashSet<>(pins));
       }
+      builder = builder.setDryRun(dryRun);
       if (handlingName != null) {
         SchemaEvolutionConfig.IncompatibleSchemaHandling handling =
             parseEnum(
@@ -316,6 +336,9 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
       if (errorHandling != null) {
         output = output.and(errorHandling.getOutput(), result.get(ERROR_TAG));
       }
+      if (Boolean.TRUE.equals(configuration.getDryRun())) {
+        output = output.and(AddFiles.DRY_RUN_TAG, result.get(AddFiles.DRY_RUN_TAG));
+      }
       return output;
     }
   }

sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java

@@ -20,7 +20,6 @@
 import static org.apache.beam.sdk.metrics.Metrics.counter;
 
 import java.util.List;
-import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
 import org.apache.beam.sdk.metrics.Counter;
 import org.apache.beam.sdk.transforms.DoFn;
 import org.apache.iceberg.catalog.Catalog;
@@ -37,34 +36,23 @@ class CommitSchemaOnce extends DoFn<List<CollectDistinctSchemas.SchemaGroup>, Lo
 
   private final IcebergCatalogConfig catalogConfig;
   private final String identifier;
-  private final SchemaEvolutionConfig config;
-  private final IncompatibleSchemaHandling handling;
-  private final CommitSchemaUnion.TableCreation creation;
+  private final CommitSchemaUnion.Settings settings;
   private final CommitSchemaUnion.Committer committer;
   private transient @MonotonicNonNull Catalog catalog;
 
   CommitSchemaOnce(
-      IcebergCatalogConfig catalogConfig,
-      String identifier,
-      SchemaEvolutionConfig config,
-      IncompatibleSchemaHandling handling,
-      CommitSchemaUnion.TableCreation creation) {
-    this(
-        catalogConfig, identifier, config, handling, creation, CommitSchemaUnion.DEFAULT_COMMITTER);
+      IcebergCatalogConfig catalogConfig, String identifier, CommitSchemaUnion.Settings settings) {
+    this(catalogConfig, identifier, settings, CommitSchemaUnion.DEFAULT_COMMITTER);
   }
 
   CommitSchemaOnce(
       IcebergCatalogConfig catalogConfig,
       String identifier,
-      SchemaEvolutionConfig config,
-      IncompatibleSchemaHandling handling,
-      CommitSchemaUnion.TableCreation creation,
+      CommitSchemaUnion.Settings settings,
       CommitSchemaUnion.Committer committer) {
     this.catalogConfig = catalogConfig;
     this.identifier = identifier;
-    this.config = config;
-    this.handling = handling;
-    this.creation = creation;
+    this.settings = settings;
     this.committer = committer;
   }
 
@@ -82,8 +70,7 @@ public void process(
           committer.commit(txn);
           numSchemaCommits.inc();
         };
-    long schemaId =
-        CommitSchemaUnion.commit(catalog, tableId, schemas, config, handling, creation, counting);
+    long schemaId = CommitSchemaUnion.commit(catalog, tableId, schemas, settings, counting);
     out.output(schemaId);
   }
 }