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
@@ -135,7 +137,9 @@
  * {@code errors} (see {@link SchemaEvolutionConfig.IncompatibleSchemaHandling}). The per-file
  * checks read Parquet footers: an ORC or Avro file, or a pinned column the footer has no null count
  * for, cannot be verified and goes to {@code errors} unless {@link
- * SchemaEvolutionConfig.UnverifiableFileHandling#ACCEPT} registers it on trust. Schema evolution
+ * SchemaEvolutionConfig.UnverifiableFileHandling#ACCEPT} registers it on trust. When the table does
+ * not exist, the pre-pass creates it from the union of the file schemas; if no readable Parquet
+ * schema can seed it, nothing is created and every file goes to {@code errors}. Schema evolution
  * currently requires bounded input; unbounded input with options set is rejected at construction.
  *
  * <pre>{@code
@@ -154,6 +158,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 +268,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 +341,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. */
@@ -394,6 +425,7 @@ static class ConvertToDataFile extends DoFn<String, SerializableDataFile> {
     private final SchemaEvolutionConfig evolution;
     private transient @MonotonicNonNull BoundedAsyncTasks<ProcessResult> tasks;
     private transient volatile @MonotonicNonNull Table table;
+    private transient volatile boolean tableMissing;
     private transient @MonotonicNonNull Set<String> warned;
     private final AtomicBoolean refreshedThisBundle = new AtomicBoolean();
 
@@ -441,6 +473,9 @@ public ConvertToDataFile(
     static final String UNREADABLE_SCHEMA_ERROR = "Could not read the file's schema: ";
     static final String UNCOVERED_ERROR = "Table schema does not cover the file after refresh: ";
     static final String PINNED_COLUMN_ERROR = "Pinned required column ";
+    static final String MISSING_TABLE_ERROR =
+        "Table does not exist and the schema pre-pass could not create it (no readable Parquet"
+            + " schema seeded it, or every file schema was refused): ";
     static final String UNCHECKED_FORMAT_ERROR =
         "Schema evolution is enabled but coverage and pin checks support only Parquet;"
             + " refusing to register an unchecked file of format ";
@@ -598,13 +633,21 @@ private Callable<ProcessResult> createProcessTask(
         // ---- Infrastructure phase. Failures propagate so the runner retries the bundle;
         // per-file error rows here would silently drop in-flight files on a transient blip.
         // Only conditions that are properties of the file go to the error output.
+        if (tableMissing) {
+          return errorResult(
+              filePath, MISSING_TABLE_ERROR + identifier, timestamp, window, paneInfo);
+        }
         if (table == null) {
           synchronized (this) {
             if (table == null) {
               try {
                 table = getOrCreateTable(filePath, format);
               } catch (FileNotFoundException e) {
                 return errorResult(filePath, errorMessage(e), timestamp, window, paneInfo);
+              } catch (NoSuchTableException e) {
+                tableMissing = true;
+                return errorResult(
+                    filePath, MISSING_TABLE_ERROR + identifier, timestamp, window, paneInfo);
               }
             }
           }
@@ -835,6 +878,10 @@ private Table getOrCreateTable(String filePath, FileFormat format) throws IOExce
       try {
         return catalogConfig.catalog().loadTable(tableId);
       } catch (NoSuchTableException e) {
+        if (evolution.isEnabled()) {
+          // the pre-pass is the only creator then, and it has already declined
+          throw e;
+        }
         try {
           org.apache.iceberg.Schema schema = getSchema(filePath, format);
           PartitionSpec spec = PartitionUtils.toPartitionSpec(partitionFields, schema);

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"
@@ -131,8 +132,10 @@ public static Builder builder() {
             + " that is not allowed is incompatible; see incompatible_schema_handling. Only"
             + " Parquet files can be checked: ORC and Avro files are sent to the error output"
             + " unless unverifiable_file_handling is ACCEPT. Files sent to the error output are"
-            + " dropped unless error_handling is set. Batch pipelines only; streaming pipelines"
-            + " cannot use schema evolution yet.")
+            + " dropped unless error_handling is set. If the table does not exist it is created"
+            + " from the union of the Parquet schemas; with none, every file goes to the error"
+            + " output. Batch pipelines only; streaming pipelines cannot use schema evolution"
+            + " yet.")
     public abstract @Nullable List<String> getSchemaEvolutionOptions();
 
     @SchemaFieldDescription(
@@ -145,6 +148,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 +174,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 +215,8 @@ public abstract static class Builder {
 
       public abstract Builder setUnverifiableFileHandling(String handling);
 
+      public abstract Builder setDryRun(Boolean dryRun);
+
       public abstract Configuration build();
     }
 
@@ -207,19 +226,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 +249,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 +338,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);
   }
 }