From 125e65f18667a56757c818e93f1c43725474ce73 Mon Sep 17 00:00:00 2001 From: claudevdm <33973061+claudevdm@users.noreply.github.com> Date: Wed, 23 Sep 2026 15:15:47 -0400 Subject: [PATCH] AddFiles: Add dry_run (#40167) * AddFiles: dry_run reports what the schema pre-pass would do without committing or registering SchemaEvolutionConfig.setDryRun(true) (provider key dry_run) turns AddFiles into a report-only transform: the read side runs as usual (footers, distinct schemas), then DryRunReport emits one Row per distinct schema plus one summary row on a new output, dry_run_report. Why: before enabling evolution on a large import, a user wants to know what the options would do to the table and which files would be refused, without touching anything. Running the real transform under FAIL_PIPELINE answers only "would it fail", and only for the first failure. The verdicts come from CommitSchemaUnion.plan, the one step that decides everything a commit does before writing anything. A Plan says which distinct file schemas are merged (SchemaToMerge), which are refused and why (IncompatibleSchema), and the schema the table ends with. It is an EvolutionPlan against an existing table (base schema snapshot, name-mapping repair) or a CreationPlan when the table is missing (partition spec and sort order resolved against the union, or the problem that blocks creation). commitOnce dispatches to evolve or create, which fail or warn under the handling mode and then write; the dry run turns the same plan into rows, so a check added to the plan reaches both and the report cannot drift from the commit. A schema that is fine against the table but conflicts with another schema of the input is therefore reported with the blame a real run assigns. Real-run changes that come with planning first: partition or sort fields that do not fit the union fail with a message naming them, before any catalog write and under either handling (the per-file fallback creation throws the same error, so it could never be routed); problems are reported before the transaction is opened; every transaction is checked against the one base snapshot; planning retries like committing; a window without schemas plans nothing. Settings (config, the handling resolved for the mode, NewTableSettings) replaces the three loose arguments both DoFns carried. Report rows (REPORT_SCHEMA), told apart by row_type: row_type schema | create | unreadable | unchecked | summary schema_key short murmur3 key of the schema JSON on schema rows schema the canonical schema JSON; on the create row, the union the table would be created with num_files files the row covers changes ARRAY: the SchemaDelta descriptions on schema rows; "create " per column on the create row (pins shown required); the totals line and table-level changes on the summary allowed whether a real run would accept it reason why not, else ""; the consequence on the summary would_create_table false whenever a real run would not create the table, including when it would fail first Provider: the dry_run_report output only exists when dry_run is set, so existing YAML pipelines that enumerate outputs are unaffected. * improve retry * comments * add untracked file * move tests * track file * comments --- .../IO_Iceberg_Integration_Tests.json | 2 +- .../apache/beam/sdk/io/iceberg/AddFiles.java | 119 +++- .../AddFilesSchemaTransformProvider.java | 41 +- .../beam/sdk/io/iceberg/CommitSchemaOnce.java | 25 +- .../sdk/io/iceberg/CommitSchemaUnion.java | 671 ++++++------------ .../beam/sdk/io/iceberg/DryRunReport.java | 434 +++++++++++ .../beam/sdk/io/iceberg/ReadFooterSchema.java | 33 +- .../beam/sdk/io/iceberg/SchemaDelta.java | 9 +- .../sdk/io/iceberg/SchemaEvolutionConfig.java | 34 +- .../beam/sdk/io/iceberg/SchemaPlan.java | 618 ++++++++++++++++ .../AddFilesSchemaTransformProviderTest.java | 14 + .../beam/sdk/io/iceberg/AddFilesTest.java | 347 ++++++++- .../sdk/io/iceberg/CommitSchemaOnceTest.java | 14 +- .../sdk/io/iceberg/CommitSchemaUnionTest.java | 345 ++++----- .../sdk/io/iceberg/ReadFooterSchemaTest.java | 2 +- .../io/iceberg/SchemaEvolutionConfigTest.java | 10 + .../beam/sdk/io/iceberg/SchemaPlanTest.java | 325 +++++++++ sdks/python/apache_beam/yaml/standard_io.yaml | 1 + .../yaml/tests/iceberg_add_files_dry_run.yaml | 185 +++++ 19 files changed, 2475 insertions(+), 754 deletions(-) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/DryRunReport.java create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaPlanTest.java create mode 100644 sdks/python/apache_beam/yaml/tests/iceberg_add_files_dry_run.yaml diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json index e1430c74e62c..6d9121eaaa4b 100644 --- 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": 7 + "modification": 8 } 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 75b1a0144e22..4093b2fb6e5d 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 @@ -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. * *

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

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

{@code
@@ -154,6 +158,10 @@
 public class AddFiles extends PTransform, 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 input) {
 
     PCollection paths = input;
     if (evolution.isEnabled()) {
-      paths = gateOnSchemaCommit(input);
+      // one commit per window, and for bounded input the window is the whole input
+      PCollection windowed =
+          input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
+      PCollection> 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 committed = commitSchema(schemas, settings);
+      paths = windowed.apply("WaitForSchemaCommit", Wait.on(committed));
     }
 
     PCollectionTuple dataFiles =
@@ -320,39 +341,49 @@ public PCollectionRowTuple expand(PCollection 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 gateOnSchemaCommit(PCollection input) {
-    PCollection 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 signal =
-        windowed
-            .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema()))
-            .setCoder(CollectDistinctSchemas.groupCoder())
-            .apply(
-                "CollectDistinctSchemas",
-                Combine.globally(new CollectDistinctSchemas()).withoutDefaults())
+  private PCollection> distinctSchemas(
+      PCollection 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 commitSchema(
+      PCollection> 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> schemas,
+      CommitSchemaUnion.Settings settings) {
+    PCollection 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 emptySnapshots =
+        schemas
+            .getPipeline()
+            .apply("NoSnapshots", Create.empty(RowCoder.of(SnapshotInfo.getSchema())))
+            .setRowSchema(SnapshotInfo.getSchema());
+    PCollection 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 {
     private final SchemaEvolutionConfig evolution;
     private transient @MonotonicNonNull BoundedAsyncTasks tasks;
     private transient volatile @MonotonicNonNull Table table;
+    private transient volatile boolean tableMissing;
     private transient @MonotonicNonNull Set 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,6 +633,10 @@ private Callable 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) {
@@ -605,6 +644,10 @@ private Callable createProcessTask(
                 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);
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 fa016ddca862..c5cc758c555b 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
@@ -120,7 +120,8 @@ public static Builder builder() {
     public abstract @Nullable List 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 getSchemaEvolutionOptions();
 
     @SchemaFieldDescription(
@@ -145,6 +148,19 @@ public static Builder builder() {
             + " Requires schema_evolution_options.")
     public abstract @Nullable List 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: `.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 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 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;
     }
   }
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java
index 6fd5171c2eae..768fb564de9b 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java
+++ b/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, 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);
   }
 }
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java
index 1e145a122c81..093749798780 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java
@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.io.iceberg;
 
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
 import java.io.Serializable;
@@ -25,45 +26,39 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
 import org.apache.beam.sdk.util.BackOff;
 import org.apache.beam.sdk.util.BackOffUtils;
 import org.apache.beam.sdk.util.FluentBackoff;
 import org.apache.beam.sdk.util.Sleeper;
 import org.apache.iceberg.Schema;
-import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.Transaction;
-import org.apache.iceberg.UpdateSchema;
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.exceptions.AlreadyExistsException;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.apache.iceberg.exceptions.NoSuchTableException;
-import org.apache.iceberg.exceptions.ValidationException;
 import org.apache.iceberg.mapping.NameMapping;
-import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.TypeUtil;
-import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PropertyUtil;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.joda.time.Duration;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * Applies the distinct file schemas of a window to the table in one commit, in phases named by the
- * methods of this class: {@code classify} each schema against a fresh load of the table (most
- * common first), {@code fold} the accepted ones into a single union on scratch transactions that
- * are never committed, {@code replay} the folded result onto the real transaction as one schema
- * update, repair the name mapping, commit once. The fold keeps per-schema blame for cross-schema
- * conflicts while the table gains a single schema version per window. Nothing is committed when
- * nothing changes.
+ * Applies the distinct file schemas of a window to the table in one commit: computes the {@link
+ * SchemaPlan} against a fresh load of the table, {@code replay}s the planned union onto the real
+ * transaction as one schema update, repairs the name mapping, commits once. The table gains a
+ * single schema version per window; nothing is committed when nothing changes. A dry run reports
+ * the same plan instead of committing it.
  *
- * 

When the table does not exist, {@code create} builds it instead: {@code foldForCreate} - * computes the same union, and the table is born from it directly with pinned columns and their - * ancestors required. + *

When the table does not exist, {@code create} builds it from the planned union instead, with + * pinned columns and their ancestors required. * *

Incompatible schemas either fail the whole call before any commit ({@link * IncompatibleSchemaHandling#FAIL_PIPELINE}) or are skipped so their files reach the error output @@ -72,18 +67,31 @@ final class CommitSchemaUnion { private static final Logger LOG = LoggerFactory.getLogger(CommitSchemaUnion.class); - static final int MAX_ATTEMPTS = 5; + /** + * How long a commit or a plan keeps retrying, unless the table's own {@code commit.retry.*} + * properties say otherwise. More patient than Iceberg's defaults because Iceberg cannot retry + * these commits itself: a schema update inside a transaction cannot be re-applied after a + * concurrent commit of any kind, a data append included, so every such commit lands here. + */ + static final Duration DEFAULT_RETRY_MIN_WAIT = Duration.millis(250); + + static final Duration DEFAULT_RETRY_MAX_WAIT = Duration.standardSeconds(10); + static final Duration DEFAULT_RETRY_TOTAL_TIMEOUT = Duration.standardMinutes(1); + + /** Tells the runner a worker waiting out a busy catalog is throttled, not busy. */ + private static final Counter throttledMillis = + Metrics.counter(Metrics.THROTTLE_TIME_NAMESPACE, Metrics.THROTTLE_TIME_COUNTER_NAME); /** Returned when the table does not exist and there is no schema to create it from. */ static final long NO_TABLE = -1L; - /** How to create the table when it does not exist: from the union of the window's schemas. */ - static final class TableCreation implements Serializable { + /** What a table is created with when it does not exist yet; unused on an existing table. */ + static final class NewTableSettings implements Serializable { final @Nullable List partitionFields; final @Nullable List sortFields; final @Nullable Map properties; - TableCreation( + NewTableSettings( @Nullable List partitionFields, @Nullable List sortFields, @Nullable Map properties) { @@ -93,6 +101,25 @@ static final class TableCreation implements Serializable { } } + /** How a schema commit behaves; the same for the commit and for a dry run of it. */ + static final class Settings implements Serializable { + final SchemaEvolutionConfig config; + + /** Resolved for the pipeline's mode; the config's own value may be unset. */ + final IncompatibleSchemaHandling handling; + + final NewTableSettings newTable; + + Settings( + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling, + NewTableSettings newTable) { + this.config = config; + this.handling = handling; + this.newTable = newTable; + } + } + /** Injectable so tests can exercise the commit retry path. */ interface Committer extends Serializable { void commit(Transaction txn); @@ -107,52 +134,6 @@ static final class IncompatibleSchemaException extends IllegalStateException { } } - private static final class Accepted { - final Schema schema; - final String json; - final long files; - - /** Null on the create path: the seed table is empty, so there is nothing to relax. */ - final @Nullable SchemaDelta delta; - - Accepted(Schema schema, String json, long files, @Nullable SchemaDelta delta) { - this.schema = schema; - this.json = json; - this.files = files; - this.delta = delta; - } - } - - private static final class Incompatible { - final String schemaJson; - final long files; - final String reason; - - Incompatible(String schemaJson, long files, String reason) { - this.schemaJson = schemaJson; - this.files = files; - this.reason = reason; - } - - @Override - public String toString() { - return files + " file(s) with schema " + truncate(schemaJson) + ": " + reason; - } - } - - /** Canonical JSON of a wide schema runs to hundreds of KB; the reason is what matters. */ - private static final int MAX_SCHEMA_JSON_CHARS = 1024; - - private static String truncate(String json) { - if (json.length() <= MAX_SCHEMA_JSON_CHARS) { - return json; - } - return json.substring(0, MAX_SCHEMA_JSON_CHARS) - + "... (" - + (json.length() - MAX_SCHEMA_JSON_CHARS) - + " chars truncated)"; - } - private CommitSchemaUnion() {} /** @@ -165,157 +146,169 @@ static long commit( Catalog catalog, TableIdentifier tableId, List schemas, - SchemaEvolutionConfig config, - IncompatibleSchemaHandling handling, - TableCreation creation, + Settings settings, Committer committer) { - // The catalog is already under contention when a retry fires; back off (jittered by - // FluentBackoff) instead of piling on. Iceberg's own metadata retries (commit.retry.*) - // sit below this loop. - BackOff backoff = - FluentBackoff.DEFAULT - .withMaxRetries(MAX_ATTEMPTS - 1) - .withInitialBackoff(Duration.millis(100)) - .withMaxBackoff(Duration.standardSeconds(2)) - .backoff(); + return commit(catalog, tableId, schemas, settings, committer, Sleeper.DEFAULT); + } + + /** Test hook: the sleeper keeps the retry tests from waiting out the backoff. */ + static long commit( + Catalog catalog, + TableIdentifier tableId, + List schemas, + Settings settings, + Committer committer, + Sleeper sleeper) { + return withRetry( + catalog, + tableId, + sleeper, + table -> commitOnce(catalog, tableId, table, schemas, settings, committer)); + } + + /** One attempt at a plan or a commit, against the table as it is now; null when missing. */ + private interface Attempt { + T run(@Nullable Table table); + } + + /** + * Runs an attempt again, against a fresh load of the table, when a concurrent commit or a create + * race invalidates the state it started from. Gives up when the backoff time is spent. + */ + private static T withRetry( + Catalog catalog, TableIdentifier tableId, Sleeper sleeper, Attempt once) { + @Nullable BackOff backoff = null; for (int attempt = 1; ; attempt++) { + @Nullable Table table; + try { + table = catalog.loadTable(tableId); + } catch (NoSuchTableException e) { + table = null; + } try { - return commitOnce(catalog, tableId, schemas, config, handling, creation, committer); + return once.run(table); } catch (CommitFailedException | AlreadyExistsException e) { - // a concurrent commit, or a create race: the next attempt loads the fresh state + if (backoff == null) { + backoff = retryBackoff(table).backoff(); + } + LOG.info( + "Schema pre-pass attempt {} for {} failed: {}", + attempt, + tableId, + AddFiles.errorMessage(e)); try { - if (!BackOffUtils.next(Sleeper.DEFAULT, backoff)) { + if (!BackOffUtils.next(sleeper, backoff)) { throw e; } } catch (InterruptedException interrupted) { Thread.currentThread().interrupt(); throw e; } - LOG.info( - "Schema commit attempt {}/{} for {} failed; reloading and rebuilding", - attempt, - MAX_ATTEMPTS, - tableId, - e); } } } + /** + * Jittered exponential backoff, bounded by time; a retry count only applies when the table sets + * one. Values the backoff would reject are clamped, so a bad table property cannot replace the + * commit failure with a configuration error. + */ + private static FluentBackoff retryBackoff(@Nullable Table table) { + Map properties = table == null ? Collections.emptyMap() : table.properties(); + long minWaitMillis = + PropertyUtil.propertyAsLong( + properties, + TableProperties.COMMIT_MIN_RETRY_WAIT_MS, + DEFAULT_RETRY_MIN_WAIT.getMillis()); + long maxWaitMillis = + PropertyUtil.propertyAsLong( + properties, + TableProperties.COMMIT_MAX_RETRY_WAIT_MS, + DEFAULT_RETRY_MAX_WAIT.getMillis()); + long totalMillis = + PropertyUtil.propertyAsLong( + properties, + TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, + DEFAULT_RETRY_TOTAL_TIMEOUT.getMillis()); + int maxRetries = + PropertyUtil.propertyAsInt( + properties, TableProperties.COMMIT_NUM_RETRIES, Integer.MAX_VALUE); + return FluentBackoff.DEFAULT + .withExponent(2.0) + .withInitialBackoff(Duration.millis(Math.max(1, minWaitMillis))) + .withMaxBackoff(Duration.millis(Math.max(1, maxWaitMillis))) + .withMaxCumulativeBackoff(Duration.millis(Math.max(1, totalMillis))) + .withMaxRetries(Math.max(0, maxRetries)) + .withThrottledTimeCounter(throttledMillis); + } + + /** The plan for the window's schemas, reloading the table when a concurrent change moves it. */ + static SchemaPlan plan( + Catalog catalog, + TableIdentifier tableId, + List schemas, + Settings settings) { + return withRetry( + catalog, + tableId, + Sleeper.DEFAULT, + table -> SchemaPlan.compute(catalog, tableId, table, schemas, settings)); + } + private static long commitOnce( Catalog catalog, TableIdentifier tableId, + @Nullable Table table, List schemas, - SchemaEvolutionConfig config, - IncompatibleSchemaHandling handling, - TableCreation creation, + Settings settings, Committer committer) { - Table table; - try { - table = catalog.loadTable(tableId); - } catch (NoSuchTableException e) { - return create(catalog, tableId, schemas, config, handling, creation, committer); + SchemaPlan plan = SchemaPlan.compute(catalog, tableId, table, schemas, settings); + if (plan instanceof SchemaPlan.Creation) { + return create((SchemaPlan.Creation) plan, catalog, tableId, settings, committer); } - // Every transaction below must share this snapshot: classification, the fold and the replay - // all reason about the same table state (newTransactionOn enforces it). - Schema base = table.schema(); + return evolve((SchemaPlan.Evolution) plan, tableId, settings.handling, committer); + } - List incompatible = new ArrayList<>(); - List accepted = classify(table, schemas, config, incompatible); - Schema merged = fold(table, base, tableId, accepted, incompatible); + /** Replays the planned union onto the table and repairs the name mapping, in one commit. */ + private static long evolve( + SchemaPlan.Evolution plan, + TableIdentifier tableId, + IncompatibleSchemaHandling handling, + Committer committer) { + failOrWarnOnIncompatibleSchemas( + tableId, plan.incompatibleSchemas, handling, "no schema change was committed"); + failOrWarnOnConfigProblems(tableId, plan.configProblems, handling); - Transaction txn = newTransactionOn(table, base, tableId); - if (merged != null) { - replay(txn, merged, tableId); + Transaction txn = plan.newTransaction(tableId); + if (plan.newSchema != null) { + replay(txn, plan.newSchema, tableId); } - boolean staged = merged != null; - staged |= stageNameMapping(txn); - - if (!incompatible.isEmpty()) { - reportIncompatible(tableId, incompatible, handling, "no schema change was committed"); + if (plan.repairsNameMapping) { + stageNameMapping(txn); } - - if (!staged) { - LOG.info( - "Table {} already covers all {} file schema(s); nothing to commit", - tableId, - schemas.size()); - return table.schema().schemaId(); + if (plan.newSchema == null && !plan.repairsNameMapping) { + LOG.info("Table {} already covers every file schema; nothing to commit", tableId); + return plan.base.schemaId(); } committer.commit(txn); long schemaId = txn.table().schema().schemaId(); - long acceptedFiles = 0; - for (Accepted item : accepted) { - acceptedFiles += item.files; + long mergedFiles = 0; + for (SchemaPlan.SchemaToMerge item : plan.schemasToMerge) { + mergedFiles += item.files; } LOG.info( "Committed schema union for {}: {} schema(s) covering {} file(s), now at schema id {}", tableId, - accepted.size(), - acceptedFiles, + plan.schemasToMerge.size(), + mergedFiles, schemaId); return schemaId; } - /** - * Sorts the window's schemas into the ones the table must change for (accepted) and the ones it - * must not ({@code incompatible}, with the reason); schemas the table already covers drop out. - */ - private static List classify( - Table table, - List schemas, - SchemaEvolutionConfig config, - List incompatible) { - List accepted = new ArrayList<>(); - for (CollectDistinctSchemas.SchemaGroup group : schemas) { - Schema fileSchema = - FileSchemas.markRequired( - SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns()); - SchemaDelta delta = SchemaDelta.classify(table, fileSchema); - if (delta.isEmpty()) { - continue; - } - if (!delta.allowedBy(config)) { - incompatible.add( - new Incompatible( - group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config))); - continue; - } - accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta)); - } - return accepted; - } - - /** - * Unions the accepted schemas into the table schema on scratch transactions that are never - * committed, relaxing every field the window adds; a schema that conflicts with another only - * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the - * folded schema, or null when nothing needs to change. - */ - private static @Nullable Schema fold( - Table table, - Schema base, - TableIdentifier tableId, - List accepted, - List incompatible) { - while (true) { - Transaction scratch = newTransactionOn(table, base, tableId); - Accepted failed = stageAll(scratch, accepted, incompatible); - if (failed != null) { - accepted.remove(failed); - continue; - } - if (accepted.isEmpty()) { - return null; - } - relaxNewRequiredFields(scratch, base); - return scratch.table().schema(); - } - } - /** * One union replays the fold's net effect (additions, promotions, relaxations) so the table gains * a single schema version instead of one per folded schema. The checkState is a pure bug - * detector: concurrent changes are caught earlier, by newTransactionOn. + * detector: concurrent changes are caught earlier, when the plan opens its transaction. */ private static void replay(Transaction txn, Schema merged, TableIdentifier tableId) { txn.updateSchema().unionByNameWith(merged).commit(); @@ -338,192 +331,59 @@ private static void replay(Transaction txn, Schema merged, TableIdentifier table * append after them. */ private static long create( + SchemaPlan.Creation plan, Catalog catalog, TableIdentifier tableId, - List schemas, - SchemaEvolutionConfig config, - IncompatibleSchemaHandling handling, - TableCreation creation, + Settings settings, Committer committer) { - if (schemas.isEmpty()) { + if (plan.schemasToMerge.isEmpty() && plan.incompatibleSchemas.isEmpty()) { LOG.info("Table {} does not exist and no file schema was read; not creating it", tableId); return NO_TABLE; } - List incompatible = new ArrayList<>(); - @Nullable Schema merged = foldForCreate(catalog, tableId, schemas, incompatible); - if (!incompatible.isEmpty()) { - reportIncompatible(tableId, incompatible, handling, "no table was created"); - } - if (merged == null) { + failOrWarnOnIncompatibleSchemas( + tableId, plan.incompatibleSchemas, settings.handling, "no table was created"); + if (plan.newSchema == null) { LOG.info("Table {} does not exist and no file schema can seed it; not creating it", tableId); return NO_TABLE; } - // The real table is built from the folded result directly. - Schema created = createdSchema(merged, config); - reportUnenforceablePins(tableId, created, config, handling); - Map properties = - creation.properties == null ? new HashMap<>() : new HashMap<>(creation.properties); + if (plan.problem != null) { + throw new IllegalStateException(plan.problem); + } + failOrWarnOnConfigProblems(tableId, plan.configProblems, settings.handling); + Map properties = new HashMap<>(); + if (settings.newTable.properties != null) { + properties.putAll(settings.newTable.properties); + } Transaction txn = catalog - .buildTable(tableId, created) - .withPartitionSpec(PartitionUtils.toPartitionSpec(creation.partitionFields, created)) - .withSortOrder(SortOrderUtils.toSortOrder(creation.sortFields, created)) + .buildTable(tableId, plan.newSchema) + .withPartitionSpec(checkStateNotNull(plan.spec)) + .withSortOrder(checkStateNotNull(plan.sortOrder)) .withProperties(properties) .createTransaction(); - stageNameMapping(txn); + if (SchemaPlan.needsNameMapping(txn.table().properties(), txn.table().schema())) { + stageNameMapping(txn); + } committer.commit(txn); long schemaId = txn.table().schema().schemaId(); LOG.info( "Created table {} from {} file schema(s), schema id {}", tableId, - schemas.size() - incompatible.size(), + plan.schemasToMerge.size(), schemaId); return schemaId; } - /** - * Unions the window's schemas into one on scratch create transactions that are never committed, - * seeded by the most common schema; conflicts move to {@code incompatible} and the fold restarts - * without the offender. - */ - private static @Nullable Schema foldForCreate( - Catalog catalog, - TableIdentifier tableId, - List schemas, - List incompatible) { - // The evolve path refuses names no table can absorb (dotted, empty, differing only in case - // within one file) as conflicts in classify; a table must not be born with them either. - List valid = new ArrayList<>(); - for (CollectDistinctSchemas.SchemaGroup group : schemas) { - Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson()); - List invalidNames = new ArrayList<>(); - ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "", invalidNames); - if (!invalidNames.isEmpty()) { - incompatible.add( - new Incompatible( - group.getSchemaJson(), - group.getFiles(), - "file schema has column names no table can hold: " + describe(invalidNames))); - continue; - } - valid.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), null)); - } - if (valid.isEmpty()) { - return null; - } - Schema seed = valid.get(0).schema; - List rest = new ArrayList<>(valid.subList(1, valid.size())); - while (true) { - Transaction scratch = catalog.buildTable(tableId, seed).createTransaction(); - Accepted failed = stageAll(scratch, rest, incompatible); - if (failed == null) { - return scratch.table().schema(); - } - rest.remove(failed); - } - } - - /** - * A pin the created schema did not end up enforcing - the column appears in no file schema, or - * the configured spelling resolves to a field the pin walk did not reach (a short container - * spelling like a.b for a.element.b, or a path inside a map key) - would stay inert forever, - * since later windows only add columns optional: a config error under FAIL_PIPELINE, a warning - * under ROUTE_TO_ERRORS (streaming may see the column later). - */ - private static void reportUnenforceablePins( - TableIdentifier tableId, - Schema created, - SchemaEvolutionConfig config, - IncompatibleSchemaHandling handling) { - List unenforceable = new ArrayList<>(); - for (String pin : config.getRequiredColumns()) { - Types.NestedField field = created.findField(pin); - if (field == null || field.isOptional()) { - unenforceable.add(pin); - } - } - if (unenforceable.isEmpty()) { - return; - } - Collections.sort(unenforceable); - if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) { - throw new IncompatibleSchemaException( - "Pinned column(s) " - + unenforceable - + " appear in none of the file schemas creating " - + tableId - + ", or their spelling does not match the column path; the created table cannot" - + " make them required"); - } - LOG.warn( - "Pinned column(s) {} appear in none of the file schemas creating {}, or their spelling" - + " does not match the column path; the created table cannot make them required", - unenforceable, - tableId); - } - - /** - * The created schema: every field optional at every level, list elements and map values included, - * except pinned paths and their ancestors, which stay required so the schema advertises the - * guarantee the per-file pin check enforces (a null ancestor nulls the pinned leaf). Map key - * subtrees keep their declared shape (keys are required by definition; pins inside them are not - * honored). Nothing depends on a created table's schema yet, so this is the schema-authoring - * moment; evolution never tightens columns afterwards. Column order is the union's, which is the - * canonical (name-sorted) order of the file schemas. - */ - static Schema createdSchema(Schema merged, SchemaEvolutionConfig config) { - Pins pins = new Pins(config.getRequiredColumns()); - List fields = new ArrayList<>(); - for (Types.NestedField field : merged.asStruct().fields()) { - fields.add(createdField(field, field.name(), pins)); - } - return new Schema(fields); - } - - private static Types.NestedField createdField(Types.NestedField field, String path, Pins pins) { - boolean required = pins.isPinnedOrAncestorOfPin(path); - return Types.NestedField.from(field) - .ofType(createdType(field.type(), path, pins)) - .isOptional(!required) - .build(); - } - - private static Type createdType(Type type, String path, Pins pins) { - if (type.isStructType()) { - List fields = new ArrayList<>(); - for (Types.NestedField field : type.asStructType().fields()) { - fields.add(createdField(field, path + "." + field.name(), pins)); - } - return Types.StructType.of(fields); - } - if (type.isListType()) { - Types.ListType list = type.asListType(); - String elementPath = path + ".element"; - Type elementType = createdType(list.elementType(), elementPath, pins); - boolean required = pins.isPinnedOrAncestorOfPin(elementPath); - return required - ? Types.ListType.ofRequired(list.elementId(), elementType) - : Types.ListType.ofOptional(list.elementId(), elementType); - } - if (type.isMapType()) { - Types.MapType map = type.asMapType(); - String valuePath = path + ".value"; - Type valueType = createdType(map.valueType(), valuePath, pins); - boolean required = pins.isPinnedOrAncestorOfPin(valuePath); - return required - ? Types.MapType.ofRequired(map.keyId(), map.valueId(), map.keyType(), valueType) - : Types.MapType.ofOptional(map.keyId(), map.valueId(), map.keyType(), valueType); - } - return type; - } - - private static void reportIncompatible( + private static void failOrWarnOnIncompatibleSchemas( TableIdentifier tableId, - List incompatible, + List incompatible, IncompatibleSchemaHandling handling, String consequence) { + if (incompatible.isEmpty()) { + return; + } long files = 0; - for (Incompatible item : incompatible) { + for (SchemaPlan.IncompatibleSchema item : incompatible) { files += item.files; } if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) { @@ -548,144 +408,33 @@ private static void reportIncompatible( joinLines(incompatible)); } - /** - * Stages one union per accepted schema onto {@code txn}: a scratch transaction on the evolve path - * (its per-schema versions stay in memory; only the folded result is ever committed), the create - * transaction on the create path. A schema can conflict with another schema's additions, which - * only surfaces while staging and poisons the transaction, so on a conflict the offender is - * returned for the caller to drop and retry with a fresh transaction. - */ - private static @Nullable Accepted stageAll( - Transaction txn, List accepted, List incompatible) { - for (Accepted item : accepted) { - // classify checked each schema against the base table only; a column that differs only in - // case from one an EARLIER schema of the window added would union as a second column. - List collisions = new ArrayList<>(); - ColumnNameChecks.findCaseCollisions( - txn.table().schema().asStruct(), item.schema.asStruct(), "", collisions); - if (!collisions.isEmpty()) { - incompatible.add( - new Incompatible( - item.json, - item.files, - "conflicts with another file schema in the same window: " + describe(collisions))); - return item; - } - // Both caught types carry staging conflicts: ValidationException from Schema - // construction at apply ("multiple fields for name"), IllegalArgumentException from - // SchemaUpdate preconditions ("Cannot change column type"). - try { - stage(txn, item); - } catch (ValidationException | IllegalArgumentException e) { - incompatible.add( - new Incompatible( - item.json, - item.files, - "conflicts with another file schema in the same window: " - + AddFiles.errorMessage(e))); - return item; - } - } - return null; - } - - private static String describe(List changes) { - List descriptions = new ArrayList<>(); - for (SchemaChange change : changes) { - descriptions.add(change.description); - } - return String.join("; ", descriptions); - } - - /** - * Iceberg refreshes the table on every {@code newTransaction()}, so a concurrent schema commit - * can slip between two transactions here. Any drift from the snapshot the window classified - * against is thrown as {@link CommitFailedException} so the commit-level retry reloads and - * rebuilds, leaving the replay checkState as a pure bug detector. - */ - private static Transaction newTransactionOn(Table table, Schema base, TableIdentifier tableId) { - Transaction txn = table.newTransaction(); - if (!txn.table().schema().sameSchema(base)) { - throw new CommitFailedException( - "concurrent schema change on %s while staging the schema union", tableId); - } - return txn; - } - - private static void stage(Transaction txn, Accepted item) { - UpdateSchema update = txn.updateSchema().unionByNameWith(item.schema); - if (item.delta != null) { - for (String path : item.delta.absentRequiredPaths()) { - update = update.makeColumnOptional(path); - } - } - update.commit(); - } - - /** - * New columns are optional at every level. The union adds top-level columns optional but keeps - * the file's optionality below them, so one file's luck would otherwise impose required fields on - * everyone. Pins do not shape new columns: they keep existing required columns from being relaxed - * (SchemaDelta) and gate files at registration. - */ - private static void relaxNewRequiredFields(Transaction txn, Schema before) { - List toRelax = newRequiredPaths(before, txn.table().schema()); - if (toRelax.isEmpty()) { + /** A configuration problem fails the run under FAIL_PIPELINE and warns under ROUTE_TO_ERRORS. */ + private static void failOrWarnOnConfigProblems( + TableIdentifier tableId, List problems, IncompatibleSchemaHandling handling) { + if (problems.isEmpty()) { return; } - UpdateSchema update = txn.updateSchema(); - for (String path : toRelax) { - update = update.makeColumnOptional(path); + if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) { + throw new IncompatibleSchemaException(String.join("; ", problems)); } - update.commit(); - } - - /** - * Paths of required fields that {@code after} has and {@code before} lacks, in schema order; - * includes fields under lists and maps (a required list element or map value counts). Map key - * subtrees are skipped: keys are required by definition and relaxing inside a struct key would - * change key identity. - */ - static List newRequiredPaths(Schema before, Schema after) { - Set beforeIds = TypeUtil.indexById(before.asStruct()).keySet(); - List paths = new ArrayList<>(); - collectNewRequired(after.asStruct(), "", beforeIds, paths); - return paths; - } - - private static void collectNewRequired( - Type.NestedType type, String prefix, Set beforeIds, List paths) { - for (Types.NestedField field : type.fields()) { - if (type.isMapType() && field.fieldId() == type.asMapType().keyId()) { - continue; - } - String path = prefix + field.name(); - if (!beforeIds.contains(field.fieldId()) && field.isRequired()) { - paths.add(path); - } - if (field.type().isNestedType()) { - collectNewRequired(field.type().asNestedType(), path + ".", beforeIds, paths); - } + for (String problem : problems) { + LOG.warn("Configuration problem on {}: {}", tableId, problem); } } - /** Regenerates the name mapping when absent, malformed or not covering the staged schema. */ - private static boolean stageNameMapping(Transaction txn) { + /** Regenerates the name mapping property for the transaction's schema. */ + private static void stageNameMapping(Transaction txn) { Schema schema = txn.table().schema(); @Nullable NameMapping existing = NameMappingUtils.parseOrNull( txn.table().properties().get(TableProperties.DEFAULT_NAME_MAPPING)); - if (existing != null && NameMappingUtils.covers(existing, schema.asStruct())) { - return false; - } String regenerated = NameMappingUtils.regenerate(schema, existing); txn.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, regenerated).commit(); - return true; } - private static String joinLines(List items) { + private static String joinLines(List items) { List lines = new ArrayList<>(); - for (Incompatible item : items) { + for (SchemaPlan.IncompatibleSchema item : items) { lines.add(item.toString()); } return String.join("\n ", lines); diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/DryRunReport.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/DryRunReport.java new file mode 100644 index 000000000000..ae069598e0a6 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/DryRunReport.java @@ -0,0 +1,434 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.metrics.Metrics.counter; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Locale; +import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.schemas.Schema; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.beam.sdk.values.Row; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.hash.Hashing; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.types.Type; +import org.checkerframework.checker.nullness.qual.MonotonicNonNull; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * Dry run of the schema pre-pass: computes the {@link CommitSchemaUnion#plan} a real commit would, + * on scratch transactions only, and reports it without committing or registering anything. Planning + * the real fold rather than classifying each schema alone is what makes a schema that is fine + * against the table but conflicts with another schema of the input come out with the blame a real + * run would assign. + * + *

One row per window. {@code allowed} is true when every file schema can be merged and the + * configuration raises no problem; otherwise {@code reason} says what a real run would do about it. + * {@code schemas} holds one entry per distinct file schema with the changes a real run would make + * on an existing table ({@code schema_key} is a short hash to group by) and, when the schema cannot + * be merged, why. {@code created_table} is the table a real run would create from the union of the + * allowed schemas, absent when the table exists or no schema can seed it; {@code + * would_create_table} is false whenever a real run would not create it, including when it would + * fail first. When the table does not exist and no file schema can create it, nothing is allowed: a + * real run sends every file to the error output. {@code table_changes} lists what a real run + * changes without a schema change, such as regenerating the name mapping. + * + *

The file counts and the counters count checked Parquet files by whether their schema is + * allowed. Unreadable files always go to the error output in a real run; unchecked (ORC, Avro) + * files count separately even when {@code unchecked_registered} says ACCEPT registers them. Pin + * evidence is per file (the footer's null counts), so a pin violation or an unproven pin is not + * predicted here. + */ +class DryRunReport extends DoFn, Row> { + private static final Logger LOG = LoggerFactory.getLogger(DryRunReport.class); + + static final String FILES_ALLOWED_COUNTER = "numDryRunFilesAllowed"; + static final String FILES_INCOMPATIBLE_COUNTER = "numDryRunFilesIncompatible"; + static final String FILES_UNREADABLE_COUNTER = "numDryRunFilesUnreadable"; + static final String FILES_UNCHECKED_COUNTER = "numDryRunFilesUnchecked"; + static final String CONFIG_PROBLEMS_COUNTER = "numDryRunConfigProblems"; + + private static final Counter numFilesAllowed = counter(DryRunReport.class, FILES_ALLOWED_COUNTER); + private static final Counter numFilesIncompatible = + counter(DryRunReport.class, FILES_INCOMPATIBLE_COUNTER); + private static final Counter numFilesUnreadable = + counter(DryRunReport.class, FILES_UNREADABLE_COUNTER); + private static final Counter numFilesUnchecked = + counter(DryRunReport.class, FILES_UNCHECKED_COUNTER); + private static final Counter numConfigProblems = + counter(DryRunReport.class, CONFIG_PROBLEMS_COUNTER); + + static final Schema SCHEMA_ENTRY = + Schema.builder() + .addStringField("schema_key") + .addStringField("schema") + .addInt64Field("num_files") + .addArrayField("changes", Schema.FieldType.STRING) + .addBooleanField("allowed") + .addStringField("reason") + .build(); + + static final Schema CREATED_TABLE = + Schema.builder() + .addStringField("schema") + .addArrayField("columns", Schema.FieldType.STRING) + .build(); + + static final Schema REPORT_SCHEMA = + Schema.builder() + .addBooleanField("allowed") + .addStringField("reason") + .addBooleanField("would_create_table") + .addInt64Field("files_allowed") + .addInt64Field("files_incompatible") + .addInt64Field("files_unreadable") + .addInt64Field("files_unchecked") + .addBooleanField("unchecked_registered") + .addArrayField("table_changes", Schema.FieldType.STRING) + .addArrayField("config_problems", Schema.FieldType.STRING) + .addNullableRowField("created_table", CREATED_TABLE) + .addArrayField("schemas", Schema.FieldType.row(SCHEMA_ENTRY)) + .build(); + + static final String NAME_MAPPING_CHANGE = + "regenerate the name mapping property to cover the schema"; + + static final String NO_TABLE_REASON = + "the table does not exist and no file schema can create it; a real run sends every file to" + + " the error output"; + + /** Wide inputs would otherwise put every column of every schema into one log entry. */ + private static final int MAX_RENDERED_ENTRIES = 50; + + private static final int MAX_RENDERED_CHANGES = 10; + + private final IcebergCatalogConfig catalogConfig; + private final String identifier; + private final CommitSchemaUnion.Settings settings; + private transient @MonotonicNonNull Catalog catalog; + + DryRunReport( + IcebergCatalogConfig catalogConfig, String identifier, CommitSchemaUnion.Settings settings) { + this.catalogConfig = catalogConfig; + this.identifier = identifier; + this.settings = settings; + } + + /** One {@code schemas} entry before it is a Row: the totals and the rendered log read these. */ + private static final class SchemaEntry { + final String key; + final String schema; + final long files; + final List changes; + final boolean allowed; + final String reason; + + SchemaEntry( + String key, + String schema, + long files, + List changes, + boolean allowed, + String reason) { + this.key = key; + this.schema = schema; + this.files = files; + this.changes = changes; + this.allowed = allowed; + this.reason = reason; + } + + Row toRow() { + return Row.withSchema(SCHEMA_ENTRY) + .withFieldValue("schema_key", key) + .withFieldValue("schema", schema) + .withFieldValue("num_files", files) + .withFieldValue("changes", changes) + .withFieldValue("allowed", allowed) + .withFieldValue("reason", reason) + .build(); + } + } + + /** The window's schema groups: the ones the plan sees, and the files that contribute none. */ + private static final class Input { + final List readable = new ArrayList<>(); + long unreadableFiles; + long uncheckedFiles; + + static Input of(List schemas) { + Input input = new Input(); + for (CollectDistinctSchemas.SchemaGroup group : schemas) { + if (group.getSchemaJson().equals(ReadFooterSchema.UNREADABLE_KEY)) { + input.unreadableFiles += group.getFiles(); + } else if (group.getSchemaJson().equals(ReadFooterSchema.UNCHECKED_FORMAT_KEY)) { + input.uncheckedFiles += group.getFiles(); + } else { + input.readable.add(group); + } + } + return input; + } + } + + private static final class Totals { + int allowedSchemas; + long allowedFiles; + int incompatibleSchemas; + long incompatibleFiles; + + static Totals of(List entries) { + Totals totals = new Totals(); + for (SchemaEntry entry : entries) { + if (entry.allowed) { + totals.allowedSchemas++; + totals.allowedFiles += entry.files; + } else { + totals.incompatibleSchemas++; + totals.incompatibleFiles += entry.files; + } + } + return totals; + } + } + + @ProcessElement + public void process( + @Element List schemas, OutputReceiver out) { + if (catalog == null) { + catalog = catalogConfig.catalog(); + } + TableIdentifier tableId = IcebergUtils.parseTableIdentifier(identifier); + Input input = Input.of(schemas); + SchemaPlan plan = CommitSchemaUnion.plan(catalog, tableId, input.readable, settings); + + List entries = new ArrayList<>(); + for (CollectDistinctSchemas.SchemaGroup group : input.readable) { + entries.add(schemaEntry(group, plan)); + } + Totals totals = Totals.of(entries); + + SchemaPlan.@Nullable Creation creation = null; + if (plan instanceof SchemaPlan.Creation) { + creation = (SchemaPlan.Creation) plan; + } + @Nullable String creationProblem = creation == null ? null : creation.problem; + long files = + totals.allowedFiles + + totals.incompatibleFiles + + input.unreadableFiles + + input.uncheckedFiles; + boolean noTable = creation != null && creation.newSchema == null && files > 0; + boolean allowed = + totals.incompatibleSchemas == 0 + && plan.configProblems.isEmpty() + && creationProblem == null + && !noTable; + boolean wouldFail = + creationProblem != null + || (settings.handling == IncompatibleSchemaHandling.FAIL_PIPELINE + && (totals.incompatibleSchemas > 0 || !plan.configProblems.isEmpty())); + boolean wouldCreateTable = creation != null && creation.canCreate() && !wouldFail; + String consequence = consequence(totals, plan.configProblems, creationProblem, noTable); + + List configProblems = new ArrayList<>(plan.configProblems); + if (creationProblem != null) { + configProblems.add(creationProblem); + } + List tableChanges = new ArrayList<>(); + if (plan instanceof SchemaPlan.Evolution && ((SchemaPlan.Evolution) plan).repairsNameMapping) { + tableChanges.add(NAME_MAPPING_CHANGE); + } + boolean uncheckedRegistered = + settings.config.getUnverifiableFileHandling() + == SchemaEvolutionConfig.UnverifiableFileHandling.ACCEPT + && !noTable; + List entryRows = new ArrayList<>(); + for (SchemaEntry entry : entries) { + entryRows.add(entry.toRow()); + } + + Row.FieldValueBuilder report = + Row.withSchema(REPORT_SCHEMA) + .withFieldValue("allowed", allowed) + .withFieldValue("reason", consequence) + .withFieldValue("would_create_table", wouldCreateTable) + .withFieldValue("files_allowed", totals.allowedFiles) + .withFieldValue("files_incompatible", totals.incompatibleFiles) + .withFieldValue("files_unreadable", input.unreadableFiles) + .withFieldValue("files_unchecked", input.uncheckedFiles) + .withFieldValue("unchecked_registered", uncheckedRegistered) + .withFieldValue("table_changes", tableChanges) + .withFieldValue("config_problems", configProblems) + .withFieldValue("schemas", entryRows); + List createdColumns = Collections.emptyList(); + if (creation != null && creation.newSchema != null) { + createdColumns = createdColumns(creation.newSchema); + report = + report.withFieldValue( + "created_table", + Row.withSchema(CREATED_TABLE) + .withFieldValue("schema", SchemaParser.toJson(creation.newSchema)) + .withFieldValue("columns", createdColumns) + .build()); + } + out.output(report.build()); + + numFilesAllowed.inc(totals.allowedFiles); + numFilesIncompatible.inc(totals.incompatibleFiles); + numFilesUnreadable.inc(input.unreadableFiles); + numFilesUnchecked.inc(input.uncheckedFiles); + numConfigProblems.inc(configProblems.size()); + LOG.info( + "Dry run for {}{}: {}{}{}\n{}", + identifier, + wouldCreateTable ? " (table would be created)" : "", + summary(input, totals), + consequence.isEmpty() ? "" : "; " + consequence, + tableChanges.isEmpty() ? "" : "; " + String.join("; ", tableChanges), + render(entries, createdColumns)); + } + + private SchemaEntry schemaEntry(CollectDistinctSchemas.SchemaGroup group, SchemaPlan plan) { + String json = group.getSchemaJson(); + @Nullable String incompatibleReason = plan.incompatibleReason(json); + // a created table's columns are reported once, as created_table: no delta exists then + List changes = Collections.emptyList(); + SchemaPlan.@Nullable SchemaToMerge toMerge = plan.toMerge(json); + if (toMerge != null && toMerge.delta != null) { + changes = toMerge.delta.descriptions(); + } + return new SchemaEntry( + key(json), + json, + group.getFiles(), + changes, + incompatibleReason == null, + incompatibleReason == null ? "" : incompatibleReason); + } + + private static String summary(Input input, Totals totals) { + return String.format( + "%d distinct schemas; %d allowed covering %d files; %d incompatible covering %d files;" + + " %d files unreadable; %d files unchecked (ORC or Avro)", + input.readable.size(), + totals.allowedSchemas, + totals.allowedFiles, + totals.incompatibleSchemas, + totals.incompatibleFiles, + input.unreadableFiles, + input.uncheckedFiles); + } + + private String consequence( + Totals totals, + List configProblems, + @Nullable String creationProblem, + boolean noTable) { + IncompatibleSchemaHandling handling = settings.handling; + List parts = new ArrayList<>(); + if (creationProblem != null) { + parts.add("a real run would fail to create the table: " + creationProblem); + } + if (noTable) { + parts.add(NO_TABLE_REASON); + } + if (totals.incompatibleSchemas > 0) { + parts.add( + handling == IncompatibleSchemaHandling.FAIL_PIPELINE + ? "a real run would fail before committing (" + handling + ")" + : "a real run would route " + + totals.incompatibleFiles + + " files to errors (" + + handling + + ")"); + } + if (!configProblems.isEmpty()) { + parts.add( + (handling == IncompatibleSchemaHandling.FAIL_PIPELINE + ? "a real run would fail on the configuration: " + : "a real run would warn on the configuration: ") + + String.join("; ", configProblems)); + } + return String.join("; ", parts); + } + + private static String render(List entries, List createdColumns) { + StringBuilder rendered = new StringBuilder(); + for (SchemaEntry entry : entries.subList(0, Math.min(entries.size(), MAX_RENDERED_ENTRIES))) { + rendered.append( + String.format( + " %-7s %8d allowed=%-5s %s %s%n", + entry.key, entry.files, entry.allowed, cut(entry.changes), entry.reason)); + } + if (entries.size() > MAX_RENDERED_ENTRIES) { + rendered + .append(" ... and ") + .append(entries.size() - MAX_RENDERED_ENTRIES) + .append(" more schemas\n"); + } + if (!createdColumns.isEmpty()) { + rendered.append(" created table: ").append(cut(createdColumns)).append('\n'); + } + return rendered.toString(); + } + + private static List cut(List changes) { + if (changes.size() <= MAX_RENDERED_CHANGES) { + return changes; + } + List shown = new ArrayList<>(changes.subList(0, MAX_RENDERED_CHANGES)); + shown.add("... and " + (changes.size() - MAX_RENDERED_CHANGES) + " more"); + return shown; + } + + /** Top-level columns of the table a real run would create; nested detail is in the schema. */ + private static List createdColumns(org.apache.iceberg.Schema created) { + List columns = new ArrayList<>(); + for (org.apache.iceberg.types.Types.NestedField field : created.columns()) { + columns.add( + "create " + + (field.isOptional() ? "optional " : "required ") + + field.name() + + " " + + typeLabel(field.type())); + } + return columns; + } + + /** Nested types print field ids, which a creation reassigns, so only their kind is named. */ + private static String typeLabel(Type type) { + if (type.isPrimitiveType()) { + return type.toString(); + } + return type.typeId().name().toLowerCase(Locale.ROOT); + } + + static String key(String schemaJson) { + return "s" + + Hashing.murmur3_32_fixed().hashUnencodedChars(schemaJson).toString().substring(0, 6); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java index e559feab3512..389f03490fe4 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java @@ -52,15 +52,25 @@ class ReadFooterSchema extends DoFn private static final Counter numFooterReadErrors = counter(ReadFooterSchema.class, FOOTER_READ_ERRORS_COUNTER); + /** + * Emitted in a dry run for files that contribute no schema. The key travels in the group's schema + * JSON field, which the combine groups by, so the files of one kind add up without a second path. + */ + static final String UNREADABLE_KEY = "unread"; + + static final String UNCHECKED_FORMAT_KEY = "unchecked"; + + private final SchemaEvolutionConfig config; private final int threadPoolSize; private final int maxInFlightTasks; private transient @MonotonicNonNull BoundedAsyncTasks tasks; - ReadFooterSchema() { - this(DEFAULT_THREAD_POOL_SIZE, DEFAULT_MAX_IN_FLIGHT_TASKS); + ReadFooterSchema(SchemaEvolutionConfig config) { + this(config, DEFAULT_THREAD_POOL_SIZE, DEFAULT_MAX_IN_FLIGHT_TASKS); } - ReadFooterSchema(int threadPoolSize, int maxInFlightTasks) { + ReadFooterSchema(SchemaEvolutionConfig config, int threadPoolSize, int maxInFlightTasks) { + this.config = config; this.threadPoolSize = threadPoolSize; this.maxInFlightTasks = maxInFlightTasks; } @@ -148,17 +158,18 @@ private static void count(ReadResult result) { } } - private static Callable createReadTask( + private Callable createReadTask( String filePath, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) { return () -> { FileFormat format; try { format = AddFiles.inferFormat(filePath); } catch (AddFiles.UnknownFormatException e) { - return new ReadResult(null, false, timestamp, window, paneInfo); + return new ReadResult(dryRunMarker(UNREADABLE_KEY), false, timestamp, window, paneInfo); } if (!format.equals(FileFormat.PARQUET)) { - return new ReadResult(null, false, timestamp, window, paneInfo); + return new ReadResult( + dryRunMarker(UNCHECKED_FORMAT_KEY), false, timestamp, window, paneInfo); } try { ParquetMetadata footer = ParquetFooters.read(filePath); @@ -168,8 +179,16 @@ private static Callable createReadTask( "Could not read the footer of {}; the file will not contribute to schema inference: {}", filePath, AddFiles.errorMessage(e)); - return new ReadResult(null, true, timestamp, window, paneInfo); + return new ReadResult(dryRunMarker(UNREADABLE_KEY), true, timestamp, window, paneInfo); } }; } + + /** A dry run counts files that contribute no schema; a real run handles them at registration. */ + private CollectDistinctSchemas.@Nullable SchemaGroup dryRunMarker(String key) { + if (!config.getDryRun()) { + return null; + } + return CollectDistinctSchemas.SchemaGroup.of(key, 1, Collections.emptyList()); + } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index bacdac2e7cfe..743558e68f3c 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -60,7 +60,14 @@ private SchemaDelta(List changes) { * (dotted, empty, case-colliding) come back as conflicts without attempting the union. */ static SchemaDelta classify(Table table, Schema fileSchema) { - Schema before = table.schema(); + return classify(table, table.schema(), fileSchema); + } + + /** + * Classifies against {@code before}, the caller's snapshot of the table schema. The union itself + * is applied to the table's current metadata, which the caller keeps equal to the snapshot. + */ + static SchemaDelta classify(Table table, Schema before, Schema fileSchema) { if (before.sameSchema(fileSchema)) { return new SchemaDelta(Collections.emptyList()); } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java index fcbfc906f3c0..da6fd33ac9d1 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java @@ -55,7 +55,23 @@ * conflicts with the table or with another file's schema. {@link IncompatibleSchemaHandling} * decides whether that fails the pipeline before any schema commit (the batch default) or skips the * schema so its files reach the error output (the streaming default). Files whose footer cannot be - * read or converted always go to the error output and never fail the pipeline. + * read or converted always go to the error output and never fail the pipeline. 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 the error output. + * + *

Dry run. Reports what a real run would do on the {@code dry_run_report} output, one row + * per window; nothing is committed or registered. {@code allowed} is true when every file schema + * can be merged and the configuration raises no problem; otherwise {@code reason} says what a real + * run would do about it. {@code schemas} holds one entry per distinct file schema with the changes + * a real run would make and, when the schema cannot be merged, the option or conflict to fix; + * {@code created_table} shows the table a real run would create. The report is a PCollection like + * any other, so attach a sink to keep it; it is also logged at INFO and the file counts are + * published as counters ({@code numDryRunFilesAllowed}, {@code numDryRunFilesIncompatible}, {@code + * numDryRunFilesUnreadable}, {@code numDryRunFilesUnchecked}, {@code numDryRunConfigProblems}). + * Adjust the settings, rerun until the report is allowed, then run for real with an error output + * attached. Against a missing table the dry run computes the union through the catalog's + * create-transaction API, which a REST catalog serves as a stage-create request: the credentials + * need table-create permission even though no table is created. */ @AutoValue public abstract class SchemaEvolutionConfig implements Serializable { @@ -104,6 +120,12 @@ public boolean isPinned(String columnPath) { return getRequiredColumns().contains(columnPath); } + /** + * Report what the pre-pass would do on the {@code dry_run_report} output, one row per window; + * commit and register nothing. + */ + public abstract boolean getDryRun(); + /** * Unset resolves by mode: {@code FAIL_PIPELINE} in batch, {@code ROUTE_TO_ERRORS} in streaming. */ @@ -144,7 +166,8 @@ public static Builder builder() { return new AutoValue_SchemaEvolutionConfig.Builder() .setOptions(Collections.emptySet()) .setRequiredColumns(Collections.emptySet()) - .setUnverifiableFileHandling(UnverifiableFileHandling.REJECT); + .setUnverifiableFileHandling(UnverifiableFileHandling.REJECT) + .setDryRun(false); } @AutoValue.Builder @@ -158,6 +181,8 @@ public abstract Builder setIncompatibleSchemaHandling( public abstract Builder setUnverifiableFileHandling(UnverifiableFileHandling handling); + public abstract Builder setDryRun(boolean dryRun); + abstract SchemaEvolutionConfig autoBuild(); /** Any setting without an option would silently do nothing, so they are rejected. */ @@ -172,10 +197,11 @@ public SchemaEvolutionConfig build() { Preconditions.checkArgument( config.isEnabled() || (config.getRequiredColumns().isEmpty() + && !config.getDryRun() && config.getIncompatibleSchemaHandling() == null && config.getUnverifiableFileHandling() == UnverifiableFileHandling.REJECT), - "required columns, incompatible schema handling and unverifiable file handling need at" - + " least one schema evolution option"); + "required columns, dry run, incompatible schema handling and unverifiable file" + + " handling need at least one schema evolution option"); return config; } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java new file mode 100644 index 000000000000..e9e72199c82b --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java @@ -0,0 +1,618 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.SortOrder; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.Transaction; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.mapping.NameMapping; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What one schema commit would do, as data: which distinct file schemas it merges into the table, + * which it refuses and why, and the schema the table ends up with. {@link #compute} works it out on + * scratch transactions that are never committed: {@code classify} each schema against the table + * (most common first), then {@code fold} the accepted ones into a single union, which keeps + * per-schema blame for cross-schema conflicts. {@link CommitSchemaUnion} executes the plan and the + * dry run reports it, so a check added here reaches both and the two cannot drift. + */ +abstract class SchemaPlan { + final List schemasToMerge; + final List incompatibleSchemas; + + /** The union to replay, or the schema to create the table with; null when nothing changes. */ + final @Nullable Schema newSchema; + + /** Problems the configuration raises against the planned schema, as messages. */ + final List configProblems; + + private SchemaPlan(Verdicts verdicts, @Nullable Schema newSchema, List configProblems) { + this.schemasToMerge = verdicts.toMerge; + this.incompatibleSchemas = verdicts.incompatible; + this.newSchema = newSchema; + this.configProblems = configProblems; + } + + /** + * The schema's entry when the table must change for it; null when the schema is incompatible or + * the table already covers it. + */ + @Nullable SchemaToMerge toMerge(String schemaJson) { + for (SchemaToMerge item : schemasToMerge) { + if (item.json.equals(schemaJson)) { + return item; + } + } + return null; + } + + /** Why the schema is incompatible; null when it is compatible. */ + @Nullable String incompatibleReason(String schemaJson) { + for (IncompatibleSchema item : incompatibleSchemas) { + if (item.schemaJson.equals(schemaJson)) { + return item.reason; + } + } + return null; + } + + /** The plan against an existing table. */ + static final class Evolution extends SchemaPlan { + final Table table; + + /** The table schema every decision was made against. */ + final Schema base; + + /** Whether the commit regenerates the name mapping property, schema change or not. */ + final boolean repairsNameMapping; + + private Evolution( + Table table, + Schema base, + Verdicts verdicts, + @Nullable Schema newSchema, + boolean repairsNameMapping, + List configProblems) { + super(verdicts, newSchema, configProblems); + this.table = table; + this.base = base; + this.repairsNameMapping = repairsNameMapping; + } + + /** A transaction on the table, checked against the snapshot this plan reasoned about. */ + Transaction newTransaction(TableIdentifier tableId) { + return transactionOn(table, base, tableId); + } + } + + /** The plan when the table does not exist; {@code newSchema} is what it would be created with. */ + static final class Creation extends SchemaPlan { + /** Set together with {@link #sortOrder} unless {@link #problem} is. */ + final @Nullable PartitionSpec spec; + + final @Nullable SortOrder sortOrder; + + /** Why the table cannot be created as configured; fails a real run under either handling. */ + final @Nullable String problem; + + private Creation( + Verdicts verdicts, + @Nullable Schema newSchema, + @Nullable PartitionSpec spec, + @Nullable SortOrder sortOrder, + @Nullable String problem, + List configProblems) { + super(verdicts, newSchema, configProblems); + this.spec = spec; + this.sortOrder = sortOrder; + this.problem = problem; + } + + /** No file schema is left to create the table from. */ + static Creation nothingToCreate(Verdicts verdicts) { + return new Creation(verdicts, null, null, null, null, new ArrayList<>()); + } + + static Creation of( + Verdicts verdicts, + Schema created, + PartitionSpec spec, + SortOrder sortOrder, + List configProblems) { + return new Creation(verdicts, created, spec, sortOrder, null, configProblems); + } + + /** The configured partition or sort fields do not fit the schema. */ + static Creation blocked( + Verdicts verdicts, Schema created, String problem, List configProblems) { + return new Creation(verdicts, created, null, null, problem, configProblems); + } + + /** There is a schema to create from, and the configured partition and sort fields fit it. */ + boolean canCreate() { + return newSchema != null && problem == null; + } + } + + /** One distinct file schema the table must change for, and may under the options. */ + static final class SchemaToMerge { + final Schema schema; + final String json; + final long files; + + /** Null on the create path: the seed table is empty, so there is nothing to relax. */ + final @Nullable SchemaDelta delta; + + SchemaToMerge(Schema schema, String json, long files, @Nullable SchemaDelta delta) { + this.schema = schema; + this.json = json; + this.files = files; + this.delta = delta; + } + } + + /** One distinct file schema the commit refuses, with the reason. */ + static final class IncompatibleSchema { + final String schemaJson; + final long files; + final String reason; + + IncompatibleSchema(String schemaJson, long files, String reason) { + this.schemaJson = schemaJson; + this.files = files; + this.reason = reason; + } + + @Override + public String toString() { + return files + " file(s) with schema " + truncate(schemaJson) + ": " + reason; + } + } + + /** Canonical JSON of a wide schema runs to hundreds of KB; the reason is what matters. */ + private static final int MAX_SCHEMA_JSON_CHARS = 1024; + + private static String truncate(String json) { + if (json.length() <= MAX_SCHEMA_JSON_CHARS) { + return json; + } + return json.substring(0, MAX_SCHEMA_JSON_CHARS) + + "... (" + + (json.length() - MAX_SCHEMA_JSON_CHARS) + + " chars truncated)"; + } + + /** Where each distinct file schema of the window stands while the plan is worked out. */ + private static final class Verdicts { + final List toMerge = new ArrayList<>(); + final List incompatible = new ArrayList<>(); + + void refuse(CollectDistinctSchemas.SchemaGroup group, String reason) { + incompatible.add(new IncompatibleSchema(group.getSchemaJson(), group.getFiles(), reason)); + } + + void refuse(Conflict conflict) { + toMerge.remove(conflict.schema); + incompatible.add( + new IncompatibleSchema(conflict.schema.json, conflict.schema.files, conflict.reason)); + } + } + + /** + * A schema that fits the table on its own but not the union of the window: a column another + * schema of the window adds has the same name with a different type, or a name differing only in + * case. Classification checks each schema against the table alone, so this surfaces only while + * the union is staged. + */ + private static final class Conflict { + final SchemaToMerge schema; + final String reason; + + Conflict(SchemaToMerge schema, String reason) { + this.schema = schema; + this.reason = reason; + } + } + + /** + * The plan for the window's schemas against the table as loaded ({@code null} when it does not + * exist). Nothing is committed; on a missing table the fold goes through the catalog's create + * transaction, which a REST catalog serves as a stage-create request. + * + * @param schemas the window's distinct schema groups, most common first + */ + static SchemaPlan compute( + Catalog catalog, + TableIdentifier tableId, + @Nullable Table table, + List schemas, + CommitSchemaUnion.Settings settings) { + if (table == null) { + return planCreation(catalog, tableId, schemas, settings); + } + return planEvolution(table, tableId, schemas, settings.config); + } + + private static Evolution planEvolution( + Table table, + TableIdentifier tableId, + List schemas, + SchemaEvolutionConfig config) { + Schema base = table.schema(); + if (schemas.isEmpty()) { + // the pipeline never commits for a window without schemas; a direct caller gets the same + return new Evolution(table, base, new Verdicts(), null, false, new ArrayList<>()); + } + Verdicts verdicts = classify(table, base, schemas, config); + @Nullable Schema newSchema = fold(table, base, tableId, verdicts); + Schema afterwards = newSchema != null ? newSchema : base; + boolean repairsNameMapping = needsNameMapping(table.properties(), afterwards); + return new Evolution(table, base, verdicts, newSchema, repairsNameMapping, new ArrayList<>()); + } + + private static Creation planCreation( + Catalog catalog, + TableIdentifier tableId, + List schemas, + CommitSchemaUnion.Settings settings) { + Verdicts verdicts = classifyForCreate(schemas); + @Nullable Schema union = foldForCreate(catalog, tableId, verdicts); + if (union == null) { + return Creation.nothingToCreate(verdicts); + } + Schema created = createdSchema(union, settings.config); + List configProblems = new ArrayList<>(); + addPinProblems(tableId, created, settings.config, configProblems); + CommitSchemaUnion.NewTableSettings newTable = settings.newTable; + try { + PartitionSpec spec = PartitionUtils.toPartitionSpec(newTable.partitionFields, created); + SortOrder sortOrder = SortOrderUtils.toSortOrder(newTable.sortFields, created); + return Creation.of(verdicts, created, spec, sortOrder, configProblems); + } catch (IllegalArgumentException | ValidationException e) { + String problem = + "Table " + + tableId + + " cannot be created with partition fields " + + newTable.partitionFields + + " and sort fields " + + newTable.sortFields + + " on the union of the file schemas: " + + AddFiles.errorMessage(e); + return Creation.blocked(verdicts, created, problem, configProblems); + } + } + + private static void addPinProblems( + TableIdentifier tableId, + Schema created, + SchemaEvolutionConfig config, + List configProblems) { + List unenforceable = unenforceablePins(created, config); + if (!unenforceable.isEmpty()) { + configProblems.add( + "Pinned column(s) " + + unenforceable + + " appear in none of the file schemas creating " + + tableId + + ", or their spelling does not match the column path; the created table cannot" + + " make them required"); + } + } + + /** + * Sorts the window's schemas into the ones the table must change for ({@code toMerge}) and the + * ones it must not ({@code incompatible}, with the reason); schemas the table already covers drop + * out. + */ + private static Verdicts classify( + Table table, + Schema base, + List schemas, + SchemaEvolutionConfig config) { + Verdicts verdicts = new Verdicts(); + for (CollectDistinctSchemas.SchemaGroup group : schemas) { + Schema fileSchema = + FileSchemas.markRequired( + SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns()); + SchemaDelta delta = SchemaDelta.classify(table, base, fileSchema); + if (delta.isEmpty()) { + continue; + } + if (!delta.allowedBy(config)) { + verdicts.refuse(group, delta.disallowedReason(config)); + continue; + } + verdicts.toMerge.add( + new SchemaToMerge(fileSchema, group.getSchemaJson(), group.getFiles(), delta)); + } + return verdicts; + } + + /** + * Unions the accepted schemas into the table schema on scratch transactions that are never + * committed, relaxing every field the window adds; a schema that conflicts with another only + * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the + * folded schema, or null when nothing needs to change. + */ + private static @Nullable Schema fold( + Table table, Schema base, TableIdentifier tableId, Verdicts verdicts) { + while (!verdicts.toMerge.isEmpty()) { + Transaction scratch = transactionOn(table, base, tableId); + @Nullable Conflict conflict = stageAll(scratch, verdicts.toMerge); + if (conflict == null) { + relaxNewRequiredFields(scratch, base); + return scratch.table().schema(); + } + verdicts.refuse(conflict); + } + return null; + } + + /** + * The evolve path refuses names no table can absorb (dotted, empty, differing only in case within + * one file) as conflicts in classify; a table must not be born with them either. + */ + private static Verdicts classifyForCreate(List schemas) { + Verdicts verdicts = new Verdicts(); + for (CollectDistinctSchemas.SchemaGroup group : schemas) { + Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson()); + List invalidNames = new ArrayList<>(); + ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "", invalidNames); + if (!invalidNames.isEmpty()) { + verdicts.refuse( + group, "file schema has column names no table can hold: " + describe(invalidNames)); + continue; + } + verdicts.toMerge.add( + new SchemaToMerge(fileSchema, group.getSchemaJson(), group.getFiles(), null)); + } + return verdicts; + } + + /** + * Unions the accepted schemas into one on scratch create transactions that are never committed, + * seeded by the most common schema; a schema that conflicts with the others moves to {@code + * incompatible}, leaves {@code toMerge}, and the fold restarts without it. Returns the union, or + * null when no schema is left to create from. A REST catalog serves each create transaction as a + * stage-create request, so the caller needs table-create permission even in a dry run. + */ + private static @Nullable Schema foldForCreate( + Catalog catalog, TableIdentifier tableId, Verdicts verdicts) { + List toMerge = verdicts.toMerge; + while (!toMerge.isEmpty()) { + Schema seed = toMerge.get(0).schema; + Transaction scratch = catalog.buildTable(tableId, seed).createTransaction(); + @Nullable Conflict conflict = stageAll(scratch, toMerge.subList(1, toMerge.size())); + if (conflict == null) { + return scratch.table().schema(); + } + verdicts.refuse(conflict); + } + return null; + } + + /** + * Pins the created schema did not end up enforcing: the column appears in no file schema, or the + * configured spelling resolves to a field the pin walk did not reach (a short container spelling + * like a.b for a.element.b, or a path inside a map key). Such a pin would stay inert forever, + * since later windows only add columns optional, so the plan reports it as a configuration + * problem. + */ + static List unenforceablePins(Schema created, SchemaEvolutionConfig config) { + List unenforceable = new ArrayList<>(); + for (String pin : config.getRequiredColumns()) { + Types.NestedField field = created.findField(pin); + if (field == null || field.isOptional()) { + unenforceable.add(pin); + } + } + Collections.sort(unenforceable); + return unenforceable; + } + + /** + * The created schema: every field optional at every level, list elements and map values included, + * except pinned paths and their ancestors, which stay required so the schema advertises the + * guarantee the per-file pin check enforces (a null ancestor nulls the pinned leaf). Map key + * subtrees keep their declared shape (keys are required by definition; pins inside them are not + * honored). Nothing depends on a created table's schema yet, so this is the schema-authoring + * moment; evolution never tightens columns afterwards. Column order is the union's, which is the + * canonical (name-sorted) order of the file schemas. + */ + static Schema createdSchema(Schema merged, SchemaEvolutionConfig config) { + Pins pins = new Pins(config.getRequiredColumns()); + List fields = new ArrayList<>(); + for (Types.NestedField field : merged.asStruct().fields()) { + fields.add(createdField(field, field.name(), pins)); + } + return new Schema(fields); + } + + private static Types.NestedField createdField(Types.NestedField field, String path, Pins pins) { + boolean required = pins.isPinnedOrAncestorOfPin(path); + return Types.NestedField.from(field) + .ofType(createdType(field.type(), path, pins)) + .isOptional(!required) + .build(); + } + + private static Type createdType(Type type, String path, Pins pins) { + if (type.isStructType()) { + List fields = new ArrayList<>(); + for (Types.NestedField field : type.asStructType().fields()) { + fields.add(createdField(field, path + "." + field.name(), pins)); + } + return Types.StructType.of(fields); + } + if (type.isListType()) { + Types.ListType list = type.asListType(); + String elementPath = path + ".element"; + Type elementType = createdType(list.elementType(), elementPath, pins); + boolean required = pins.isPinnedOrAncestorOfPin(elementPath); + return required + ? Types.ListType.ofRequired(list.elementId(), elementType) + : Types.ListType.ofOptional(list.elementId(), elementType); + } + if (type.isMapType()) { + Types.MapType map = type.asMapType(); + String valuePath = path + ".value"; + Type valueType = createdType(map.valueType(), valuePath, pins); + boolean required = pins.isPinnedOrAncestorOfPin(valuePath); + return required + ? Types.MapType.ofRequired(map.keyId(), map.valueId(), map.keyType(), valueType) + : Types.MapType.ofOptional(map.keyId(), map.valueId(), map.keyType(), valueType); + } + return type; + } + + /** + * Stages one union per accepted schema onto {@code txn}: a scratch transaction on the evolve path + * (its per-schema versions stay in memory; only the folded result is ever committed), the create + * transaction on the create path. A schema can conflict with another schema's additions, which + * only surfaces while staging and poisons the transaction, so on a conflict the offender is + * returned for the caller to drop and retry with a fresh transaction. + */ + private static @Nullable Conflict stageAll(Transaction txn, List toMerge) { + for (SchemaToMerge item : toMerge) { + // classify checked each schema against the base table only; a column that differs only in + // case from one an EARLIER schema of the window added would union as a second column. + List collisions = new ArrayList<>(); + ColumnNameChecks.findCaseCollisions( + txn.table().schema().asStruct(), item.schema.asStruct(), "", collisions); + if (!collisions.isEmpty()) { + return new Conflict( + item, "conflicts with another file schema in the same window: " + describe(collisions)); + } + // Both caught types carry staging conflicts: ValidationException from Schema + // construction at apply ("multiple fields for name"), IllegalArgumentException from + // SchemaUpdate preconditions ("Cannot change column type"). + try { + stage(txn, item); + } catch (ValidationException | IllegalArgumentException e) { + return new Conflict( + item, + "conflicts with another file schema in the same window: " + AddFiles.errorMessage(e)); + } + } + return null; + } + + private static String describe(List changes) { + List descriptions = new ArrayList<>(); + for (SchemaChange change : changes) { + descriptions.add(change.description); + } + return String.join("; ", descriptions); + } + + /** + * Iceberg refreshes the table on every {@code newTransaction()}, so a concurrent schema commit + * can slip between two transactions here. Any drift from the snapshot the window classified + * against is thrown as {@link CommitFailedException} so the retry in {@link CommitSchemaUnion} + * reloads and rebuilds, leaving the replay checkState as a pure bug detector. + */ + static Transaction transactionOn(Table table, Schema base, TableIdentifier tableId) { + Transaction txn = table.newTransaction(); + if (!txn.table().schema().sameSchema(base)) { + throw new CommitFailedException( + "concurrent schema change on %s while staging the schema union", tableId); + } + return txn; + } + + private static void stage(Transaction txn, SchemaToMerge item) { + UpdateSchema update = txn.updateSchema().unionByNameWith(item.schema); + if (item.delta != null) { + for (String path : item.delta.absentRequiredPaths()) { + update = update.makeColumnOptional(path); + } + } + update.commit(); + } + + /** + * New columns are optional at every level. The union adds top-level columns optional but keeps + * the file's optionality below them, so one file's luck would otherwise impose required fields on + * everyone. Pins do not shape new columns: they keep existing required columns from being relaxed + * (SchemaDelta) and gate files at registration. + */ + private static void relaxNewRequiredFields(Transaction txn, Schema before) { + List toRelax = newRequiredPaths(before, txn.table().schema()); + if (toRelax.isEmpty()) { + return; + } + UpdateSchema update = txn.updateSchema(); + for (String path : toRelax) { + update = update.makeColumnOptional(path); + } + update.commit(); + } + + /** + * Paths of required fields that {@code after} has and {@code before} lacks, in schema order; + * includes fields under lists and maps (a required list element or map value counts). Map key + * subtrees are skipped: keys are required by definition and relaxing inside a struct key would + * change key identity. + */ + static List newRequiredPaths(Schema before, Schema after) { + Set beforeIds = TypeUtil.indexById(before.asStruct()).keySet(); + List paths = new ArrayList<>(); + collectNewRequired(after.asStruct(), "", beforeIds, paths); + return paths; + } + + private static void collectNewRequired( + Type.NestedType type, String prefix, Set beforeIds, List paths) { + for (Types.NestedField field : type.fields()) { + if (type.isMapType() && field.fieldId() == type.asMapType().keyId()) { + continue; + } + String path = prefix + field.name(); + if (!beforeIds.contains(field.fieldId()) && field.isRequired()) { + paths.add(path); + } + if (field.type().isNestedType()) { + collectNewRequired(field.type().asNestedType(), path + ".", beforeIds, paths); + } + } + } + + /** The table's name mapping property is absent, malformed or does not cover {@code schema}. */ + static boolean needsNameMapping(Map tableProperties, Schema schema) { + @Nullable NameMapping existing = + NameMappingUtils.parseOrNull(tableProperties.get(TableProperties.DEFAULT_NAME_MAPPING)); + return existing == null || !NameMappingUtils.covers(existing, schema.asStruct()); + } +} 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 2e469ef6da71..3b58af201543 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 @@ -137,6 +137,20 @@ public void testInvalidHandlingValuesRejected() { .getSchemaEvolution()); } + @Test + public void testDryRunParsed() { + SchemaEvolutionConfig config = + base() + .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION")) + .setDryRun(true) + .build() + .getSchemaEvolution(); + assertNotNull(config); + assertTrue(config.getDryRun()); + assertThrows( + IllegalArgumentException.class, () -> base().setDryRun(true).build().getSchemaEvolution()); + } + @Test public void testRouteToErrorsWithoutErrorHandlingRejected() { IllegalArgumentException e = diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java index 5e6cbdf4ea32..b9fe94f27a5c 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java @@ -26,7 +26,9 @@ import static org.hamcrest.Matchers.containsString; import static org.hamcrest.Matchers.hasEntry; import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; @@ -34,8 +36,10 @@ import java.io.IOException; import java.nio.ByteBuffer; import java.nio.CharBuffer; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; import java.util.Collections; import java.util.Comparator; import java.util.EnumSet; @@ -71,6 +75,7 @@ import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.BaseTable; import org.apache.iceberg.DataFile; import org.apache.iceberg.FileFormat; import org.apache.iceberg.FileScanTask; @@ -1047,6 +1052,315 @@ public void testEvolutionAddsColumnsBeforeRegisteringFiles() throws Exception { "stats for the added column", nullCountsOf(table, "wide.parquet").containsKey(emailId)); } + @Test + public void testDryRunReportsWithoutCommittingOrRegistering() throws Exception { + catalog.createTable(tableId, icebergSchema); + String before = metadataLocation(); + String covered = writeOneRecord("covered.parquet"); + String wide = writeWider("wide.parquet"); + String conflict = writeConflicting("conflict.parquet"); + + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(covered, wide, conflict)).apply(addFiles(DRY_RUN)); + PAssert.that(output.get("errors")).empty(); + PAssert.that(output.get("snapshots")).empty(); + PAssert.that(output.get(AddFiles.DRY_RUN_TAG)) + .satisfies( + rows -> { + Row report = report(rows); + Collection schemas = schemas(report); + assertEquals(3, schemas.size()); + boolean sawAddition = false; + boolean sawConflict = false; + for (Row schema : schemas) { + Collection changes = schema.getArray("changes"); + if (changes.contains("add optional email string")) { + sawAddition = schema.getBoolean("allowed"); + } + if (!schema.getBoolean("allowed")) { + sawConflict = schema.getString("reason").contains("conflicts"); + } + } + assertTrue(sawAddition); + assertTrue(sawConflict); + assertFalse(report.getBoolean("allowed")); + assertEquals(Long.valueOf(2), report.getInt64("files_allowed")); + assertEquals(Long.valueOf(1), report.getInt64("files_incompatible")); + assertThat(report.getString("reason"), containsString("would fail")); + return null; + }); + assertEquals(0, countTransforms(pipeline, "ConvertToDataFiles")); + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); + Table table = catalog.loadTable(tableId); + assertEquals(0, Iterables.size(table.snapshots())); + assertEquals(2, counted(result, DryRunReport.class, DryRunReport.FILES_ALLOWED_COUNTER)); + assertEquals(1, counted(result, DryRunReport.class, DryRunReport.FILES_INCOMPATIBLE_COUNTER)); + assertEquals(0, counted(result, DryRunReport.class, DryRunReport.FILES_UNREADABLE_COUNTER)); + assertEquals(0, counted(result, DryRunReport.class, DryRunReport.CONFIG_PROBLEMS_COUNTER)); + assertEquals(before, metadataLocation()); + } + + private String metadataLocation() { + return ((BaseTable) catalog.loadTable(tableId)).operations().current().metadataFileLocation(); + } + + private static Row report(Iterable rows) { + return Iterables.getOnlyElement(rows); + } + + private static Collection schemas(Row report) { + return checkStateNotNull(report.getArray("schemas")); + } + + @Test + public void testDryRunAgainstMissingTableReportsCreation() throws Exception { + String wide = writeWider("wide.parquet"); + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(wide)).apply(addFiles(DRY_RUN)); + PAssert.that(output.get(AddFiles.DRY_RUN_TAG)) + .satisfies( + rows -> { + Row report = report(rows); + assertTrue(report.getBoolean("allowed")); + assertTrue(report.getBoolean("would_create_table")); + Row created = checkStateNotNull(report.getRow("created_table")); + assertThat( + created.getArray("columns").toString(), + containsString("create optional email string")); + assertThat(created.getString("schema"), containsString("\"name\":\"email\"")); + for (Row schema : schemas(report)) { + assertTrue( + "the created table is described once", schema.getArray("changes").isEmpty()); + } + return null; + }); + pipeline.run().waitUntilFinish(); + assertFalse(catalog.tableExists(tableId)); + } + + /** A real run under FAIL_PIPELINE throws before creating the table, and the report says so. */ + @Test + public void testDryRunAgainstMissingTableDoesNotCreateWhenARealRunWouldFail() throws Exception { + String wide = writeWider("wide.parquet"); + String conflict = writeConflicting("conflict.parquet"); + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(wide, conflict)).apply(addFiles(DRY_RUN)); + PAssert.that(output.get(AddFiles.DRY_RUN_TAG)) + .satisfies( + rows -> { + Row report = report(rows); + assertFalse(report.getBoolean("allowed")); + assertFalse(report.getBoolean("would_create_table")); + assertNotNull("the union still describes the table", report.getRow("created_table")); + assertThat(report.getString("reason"), containsString("would fail")); + return null; + }); + pipeline.run().waitUntilFinish(); + assertFalse(catalog.tableExists(tableId)); + } + + /** A table-level change without a schema change is reported as such. */ + @Test + public void testDryRunReportsNameMappingRepair() throws Exception { + catalog.createTable(tableId, icebergSchema); + String before = metadataLocation(); + String covered = writeOneRecord("covered.parquet"); + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(covered)).apply(addFiles(DRY_RUN)); + PAssert.that(output.get(AddFiles.DRY_RUN_TAG)) + .satisfies( + rows -> { + Row report = report(rows); + assertTrue(report.getBoolean("allowed")); + assertEquals( + Arrays.asList(DryRunReport.NAME_MAPPING_CHANGE), + new ArrayList<>(checkStateNotNull(report.getArray("table_changes")))); + return null; + }); + pipeline.run().waitUntilFinish(); + assertEquals(before, metadataLocation()); + } + + /** Creation settings are part of the plan: a partition field the union lacks is reported. */ + @Test + public void testDryRunReportsCreationBlockedByPartitionFields() throws Exception { + String wide = writeWider("wide.parquet"); + AddFiles partitionedByMissing = + new AddFiles( + catalogConfig, + tableId.toString(), + null, + Arrays.asList("missing"), + null, + null, + null, + null, + DRY_RUN); + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(wide)).apply(partitionedByMissing); + PAssert.that(output.get(AddFiles.DRY_RUN_TAG)) + .satisfies( + rows -> { + Row report = report(rows); + assertFalse(report.getBoolean("allowed")); + assertFalse(report.getBoolean("would_create_table")); + assertThat(report.getString("reason"), containsString("would fail to create")); + assertThat(report.getString("reason"), containsString("partition fields [missing]")); + assertEquals(1, checkStateNotNull(report.getArray("config_problems")).size()); + return null; + }); + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); + assertEquals(1, counted(result, DryRunReport.class, DryRunReport.CONFIG_PROBLEMS_COUNTER)); + assertFalse(catalog.tableExists(tableId)); + } + + /** + * The dry run reuses the real fold on scratch transactions: a schema compatible with the table + * but incompatible with another schema of the input is reported, as a real run would. + */ + @Test + public void testDryRunSurfacesCrossSchemaConflicts() throws Exception { + catalog.createTable(tableId, icebergSchema); + Schema emailInt = + new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.required(2, "name", Types.StringType.get()), + Types.NestedField.required(3, "age", Types.IntegerType.get()), + Types.NestedField.optional(4, "email", Types.IntegerType.get())); + String a = writeWider("email_string.parquet"); + Record intRecord = GenericRecord.create(emailInt); + intRecord.setField("id", 1); + intRecord.setField("name", "a"); + intRecord.setField("age", 1); + intRecord.setField("email", 7); + String b = writeWithSchema("email_int.parquet", emailInt, intRecord); + + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(a, b)).apply(addFiles(DRY_RUN)); + PAssert.that(output.get(AddFiles.DRY_RUN_TAG)) + .satisfies( + rows -> { + Collection schemas = schemas(report(rows)); + assertEquals(2, schemas.size()); + int allowed = 0; + for (Row schema : schemas) { + if (schema.getBoolean("allowed")) { + allowed++; + } + } + assertEquals("one of the two schemas loses the fold", 1, allowed); + return null; + }); + pipeline.run().waitUntilFinish(); + assertNull(catalog.loadTable(tableId).schema().findField("email")); + } + + /** With evolution on, the pre-pass is the only creator; registration never falls back to one. */ + @Test + public void testMissingTableIsNotCreatedAtRegistrationWithEvolution() throws Exception { + File avro = temp.newFile("data.avro"); + File garbage = temp.newFile("garbage.parquet"); + java.nio.file.Files.write(garbage.toPath(), "not parquet".getBytes(StandardCharsets.UTF_8)); + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(avro.getAbsolutePath(), garbage.getAbsolutePath())) + .apply(addFiles(ADDITIONS)); + PAssert.that(output.get("snapshots")).empty(); + PAssert.that(output.get("errors")) + .satisfies( + rows -> { + int count = 0; + for (Row row : rows) { + count++; + assertThat( + row.getString("error"), + containsString(AddFiles.ConvertToDataFile.MISSING_TABLE_ERROR)); + } + assertEquals(2, count); + return null; + }); + pipeline.run().waitUntilFinish(); + assertFalse(catalog.tableExists(tableId)); + } + + /** Nothing can create the table, so nothing is allowed, whatever ACCEPT would register. */ + @Test + public void testDryRunAgainstMissingTableWithNoUsableSchema() throws Exception { + File avro = temp.newFile("data.avro"); + SchemaEvolutionConfig config = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .setUnverifiableFileHandling(UnverifiableFileHandling.ACCEPT) + .setDryRun(true) + .build(); + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(avro.getAbsolutePath())).apply(addFiles(config)); + PAssert.that(output.get(AddFiles.DRY_RUN_TAG)) + .satisfies( + rows -> { + Row report = report(rows); + assertFalse(report.getBoolean("allowed")); + assertFalse(report.getBoolean("would_create_table")); + assertFalse(report.getBoolean("unchecked_registered")); + assertEquals(Long.valueOf(1), report.getInt64("files_unchecked")); + assertThat(report.getString("reason"), containsString(DryRunReport.NO_TABLE_REASON)); + return null; + }); + pipeline.run().waitUntilFinish(); + assertFalse(catalog.tableExists(tableId)); + } + + /** Files that contribute no schema are counted instead of vanishing. */ + @Test + public void testDryRunReportsUnreadableAndNonParquetFiles() throws Exception { + dryRunWithUnreadableAndAvro(UnverifiableFileHandling.REJECT, false); + } + + @Test + public void testDryRunReportsNonParquetFilesAsRegisteredWhenAccepted() throws Exception { + dryRunWithUnreadableAndAvro(UnverifiableFileHandling.ACCEPT, true); + } + + private void dryRunWithUnreadableAndAvro( + UnverifiableFileHandling handling, boolean uncheckedRegistered) throws Exception { + catalog.createTable(tableId, icebergSchema); + String good = writeOneRecord("good.parquet"); + File garbage = temp.newFile("garbage.parquet"); + java.nio.file.Files.write(garbage.toPath(), "not parquet".getBytes(StandardCharsets.UTF_8)); + File avro = temp.newFile("data.avro"); + + SchemaEvolutionConfig config = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .setUnverifiableFileHandling(handling) + .setDryRun(true) + .build(); + PCollectionRowTuple output = + pipeline + .apply( + "Create Input", Create.of(good, garbage.getAbsolutePath(), avro.getAbsolutePath())) + .apply(addFiles(config)); + PAssert.that(output.get(AddFiles.DRY_RUN_TAG)) + .satisfies( + rows -> { + Row report = report(rows); + assertTrue(report.getBoolean("allowed")); + assertEquals(Long.valueOf(1), report.getInt64("files_allowed")); + assertEquals(Long.valueOf(1), report.getInt64("files_unreadable")); + assertEquals(Long.valueOf(1), report.getInt64("files_unchecked")); + assertEquals(uncheckedRegistered, report.getBoolean("unchecked_registered")); + return null; + }); + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); + assertEquals(1, counted(result, DryRunReport.class, DryRunReport.FILES_ALLOWED_COUNTER)); + assertEquals(0, counted(result, DryRunReport.class, DryRunReport.FILES_INCOMPATIBLE_COUNTER)); + assertEquals(1, counted(result, DryRunReport.class, DryRunReport.FILES_UNREADABLE_COUNTER)); + assertEquals(1, counted(result, DryRunReport.class, DryRunReport.FILES_UNCHECKED_COUNTER)); + } + @Test public void testEvolutionDisabledAddsNoPrePassTransforms() throws Exception { catalog.createTable(tableId, icebergSchema); @@ -1178,7 +1492,8 @@ public void testUpstreamWindowedBatchInputEvolvesAndRegisters() throws Exception /** * The schema commit retries a CommitFailedException (another writer got in first). The committer - * is serialized with the DoFn, so only the table shows the retry happened. + * is serialized with the DoFn, so the retry shows in the table and in the time the backoff + * reports to the runner as throttled. */ @Test public void testTransientSchemaCommitFailureIsRetried() throws Exception { @@ -1199,9 +1514,25 @@ public void testTransientSchemaCommitFailureIsRetried() throws Exception { .apply(addFiles(ADDITIONS).withSchemaCommitter(failsOnce)); PAssert.that(output.get("errors")).empty(); - pipeline.run().waitUntilFinish(); + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); assertEmailAddedAndFilesRegistered(1); + long throttledMillis = 0; + for (MetricResult metric : + result + .metrics() + .queryMetrics( + MetricsFilter.builder() + .addNameFilter( + MetricNameFilter.named( + org.apache.beam.sdk.metrics.Metrics.THROTTLE_TIME_NAMESPACE, + org.apache.beam.sdk.metrics.Metrics.THROTTLE_TIME_COUNTER_NAME)) + .build()) + .getCounters()) { + throttledMillis += metric.getAttempted(); + } + assertTrue("one backoff wait was reported: " + throttledMillis, throttledMillis > 0); } // ---- ConvertToDataFile coverage check and pinned columns @@ -1209,6 +1540,12 @@ public void testTransientSchemaCommitFailureIsRetried() throws Exception { private static final SchemaEvolutionConfig ADDITIONS = SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION); + private static final SchemaEvolutionConfig DRY_RUN = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .setDryRun(true) + .build(); + private PCollectionTuple convert(SchemaEvolutionConfig config, String... files) { PCollectionTuple out = pipeline @@ -1590,13 +1927,17 @@ private static SchemaEvolutionConfig accepting(SchemaEvolutionConfig base) { } private static long counted(PipelineResult result, String counter) { + return counted(result, AddFiles.class, counter); + } + + private static long counted(PipelineResult result, Class namespace, String counter) { long total = 0; for (MetricResult metric : result .metrics() .queryMetrics( MetricsFilter.builder() - .addNameFilter(MetricNameFilter.named(AddFiles.class, counter)) + .addNameFilter(MetricNameFilter.named(namespace, counter)) .build()) .getCounters()) { total += metric.getAttempted(); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java index e3d6889b1616..07a0adeb6cfb 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java @@ -89,9 +89,10 @@ public void testCommitsTheWindowsSchemasAndEmitsTheSchemaId() { new CommitSchemaOnce( catalogConfig, "default." + testName.getMethodName(), - SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), - IncompatibleSchemaHandling.FAIL_PIPELINE, - new CommitSchemaUnion.TableCreation(null, null, null)))); + new CommitSchemaUnion.Settings( + SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + IncompatibleSchemaHandling.FAIL_PIPELINE, + new CommitSchemaUnion.NewTableSettings(null, null, null))))); PAssert.that(schemaIds).containsInAnyOrder(1L); PipelineResult result = pipeline.run(); result.waitUntilFinish(); @@ -130,9 +131,10 @@ public void testNoOpWindowDoesNotCountACommit() { new CommitSchemaOnce( catalogConfig, "default." + testName.getMethodName(), - SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), - IncompatibleSchemaHandling.FAIL_PIPELINE, - new CommitSchemaUnion.TableCreation(null, null, null)))); + new CommitSchemaUnion.Settings( + SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + IncompatibleSchemaHandling.FAIL_PIPELINE, + new CommitSchemaUnion.NewTableSettings(null, null, null))))); PipelineResult result = pipeline.run(); result.waitUntilFinish(); assertEquals(0, commitsCounted(result)); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java index 7df6af49920c..0c5f379a1105 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java @@ -19,9 +19,12 @@ import static org.apache.iceberg.types.Types.NestedField.optional; import static org.apache.iceberg.types.Types.NestedField.required; +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.containsString; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; import static org.junit.Assert.assertTrue; @@ -35,6 +38,7 @@ import org.apache.beam.sdk.io.iceberg.CommitSchemaUnion.Committer; import org.apache.beam.sdk.io.iceberg.CommitSchemaUnion.IncompatibleSchemaException; import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling; +import org.apache.beam.sdk.util.FastNanoClockAndSleeper; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.BaseTable; import org.apache.iceberg.Schema; @@ -65,6 +69,9 @@ public class CommitSchemaUnionTest { @Rule public TestName testName = new TestName(); + /** Advances a fake clock instead of sleeping out the commit backoff. */ + @Rule public FastNanoClockAndSleeper fastClock = new FastNanoClockAndSleeper(); + private static final Schema TABLE = new Schema( required(1, "id", Types.LongType.get()), @@ -77,8 +84,15 @@ public class CommitSchemaUnionTest { private static final SchemaEvolutionConfig ADDITION_ONLY = SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION); - private static final CommitSchemaUnion.TableCreation NO_CREATION = - new CommitSchemaUnion.TableCreation(null, null, null); + private static final CommitSchemaUnion.NewTableSettings NO_CREATION = + new CommitSchemaUnion.NewTableSettings(null, null, null); + + private static CommitSchemaUnion.Settings settings( + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling, + CommitSchemaUnion.NewTableSettings newTable) { + return new CommitSchemaUnion.Settings(config, handling, newTable); + } private HadoopCatalog catalog; private TableIdentifier tableId; @@ -111,9 +125,7 @@ private long commit( catalog, tableId, Arrays.asList(schemas), - config, - handling, - NO_CREATION, + settings(config, handling, NO_CREATION), CommitSchemaUnion.DEFAULT_COMMITTER); } @@ -260,99 +272,6 @@ public void testFieldsInsideAddedContainersAreOptional() { assertTrue(schema.findField("attrs.value.v").isOptional()); } - // ---- newRequiredPaths (direct) - - @Test - public void testNewRequiredPathsAtEveryLevelExceptMapKeys() { - Schema before = new Schema(required(1, "id", Types.LongType.get())); - Schema after = - new Schema( - required(1, "id", Types.LongType.get()), - optional( - 2, - "s", - Types.StructType.of( - required(3, "a", Types.IntegerType.get()), - optional(4, "b", Types.IntegerType.get()))), - optional( - 5, - "items", - Types.ListType.ofRequired( - 6, Types.StructType.of(required(7, "qty", Types.IntegerType.get())))), - optional( - 8, - "attrs", - Types.MapType.ofRequired( - 9, - 10, - Types.StructType.of(required(11, "k", Types.StringType.get())), - Types.StructType.of(required(12, "v", Types.IntegerType.get()))))); - assertEquals( - Arrays.asList("s.a", "items.element", "items.element.qty", "attrs.value", "attrs.value.v"), - CommitSchemaUnion.newRequiredPaths(before, after)); - } - - /** Names containing element/key/value are not containers; regression for a substring check. */ - @Test - public void testNewRequiredPathsContainerLikeNamesAreNotContainers() { - Schema before = new Schema(required(1, "id", Types.LongType.get())); - Schema after = - new Schema( - required(1, "id", Types.LongType.get()), - optional( - 2, - "stats", - Types.StructType.of( - required(3, "keyword", Types.StringType.get()), - required(4, "value_sum", Types.LongType.get()), - required(5, "element", Types.StringType.get())))); - assertEquals( - Arrays.asList("stats.keyword", "stats.value_sum", "stats.element"), - CommitSchemaUnion.newRequiredPaths(before, after)); - } - - @Test - public void testNewRequiredPathsInNestedContainers() { - Schema before = new Schema(required(1, "id", Types.LongType.get())); - Schema after = - new Schema( - required(1, "id", Types.LongType.get()), - optional( - 2, - "ll", - Types.ListType.ofRequired( - 3, Types.ListType.ofRequired(4, Types.IntegerType.get()))), - optional( - 5, - "lm", - Types.ListType.ofRequired( - 6, - Types.MapType.ofRequired( - 7, 8, Types.StringType.get(), Types.IntegerType.get())))); - assertEquals( - Arrays.asList("ll.element", "ll.element.element", "lm.element", "lm.element.value"), - CommitSchemaUnion.newRequiredPaths(before, after)); - } - - /** Growing an existing struct: only the field with a new id is a candidate. */ - @Test - public void testNewRequiredPathsIgnoreExistingFields() { - Schema before = - new Schema( - required(1, "id", Types.LongType.get()), - optional(2, "s", Types.StructType.of(required(3, "old", Types.IntegerType.get())))); - Schema after = - new Schema( - required(1, "id", Types.LongType.get()), - optional( - 2, - "s", - Types.StructType.of( - required(3, "old", Types.IntegerType.get()), - required(4, "fresh", Types.IntegerType.get())))); - assertEquals(Arrays.asList("s.fresh"), CommitSchemaUnion.newRequiredPaths(before, after)); - } - /** A declared-optional column every file proved null-free does not relax the table. */ @Test public void testNullFreeColumnIsNotRelaxed() { @@ -550,9 +469,7 @@ public void testFinalSchemaIsOrderIndependent() { catalog, other, Arrays.asList(files(b, 2), files(a, 1)), - ALL, - IncompatibleSchemaHandling.FAIL_PIPELINE, - NO_CREATION, + settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION), CommitSchemaUnion.DEFAULT_COMMITTER); assertTrue(first.sameSchema(catalog.loadTable(other).schema())); } @@ -742,18 +659,21 @@ public void testCommitFailedOnceIsRetriedAgainstFreshState() { catalog, tableId, Arrays.asList(files(file, 1)), - ALL, - IncompatibleSchemaHandling.FAIL_PIPELINE, - NO_CREATION, - flakyThenExternalChange); + settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION), + flakyThenExternalChange, + fastClock); Table table = load(); assertEquals(2, attempts.get()); assertNotNull(table.schema().findField("external")); assertNotNull(table.schema().findField("email")); } - @Test - public void testPersistentCommitFailurePropagates() { + private long millisSleptSince(long startNanos) { + return (fastClock.nanoTime() - startNanos) / 1_000_000; + } + + /** A commit that always loses; returns how many times it was attempted. */ + private int attemptsUntilGivingUp() { AtomicInteger attempts = new AtomicInteger(); Committer alwaysFails = txn -> { @@ -772,11 +692,54 @@ public void testPersistentCommitFailurePropagates() { catalog, tableId, Arrays.asList(files(file, 1)), - ALL, - IncompatibleSchemaHandling.FAIL_PIPELINE, - NO_CREATION, - alwaysFails)); - assertEquals(CommitSchemaUnion.MAX_ATTEMPTS, attempts.get()); + settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION), + alwaysFails, + fastClock)); + return attempts.get(); + } + + /** The budget is time, not a count: the waits add up to the timeout, whatever the jitter. */ + @Test + public void testPersistentCommitFailureGivesUpWhenTheRetryTimeIsSpent() { + long start = fastClock.nanoTime(); + int attempts = attemptsUntilGivingUp(); + assertEquals( + CommitSchemaUnion.DEFAULT_RETRY_TOTAL_TIMEOUT.getMillis(), millisSleptSince(start)); + assertTrue("retried: " + attempts, attempts > 1); + assertNull("nothing was committed", load().schema().findField("email")); + } + + @Test + public void testTableRetryPropertiesSetTheBudget() { + load() + .updateProperties() + .set(TableProperties.COMMIT_MIN_RETRY_WAIT_MS, "10") + .set(TableProperties.COMMIT_MAX_RETRY_WAIT_MS, "40") + .set(TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, "200") + .commit(); + long start = fastClock.nanoTime(); + attemptsUntilGivingUp(); + assertEquals(200, millisSleptSince(start)); + } + + /** A bad table property must not replace the commit failure with a configuration error. */ + @Test + public void testUnusableTableRetryPropertiesAreClamped() { + load() + .updateProperties() + .set(TableProperties.COMMIT_MIN_RETRY_WAIT_MS, "0") + .set(TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, "-5") + .set(TableProperties.COMMIT_NUM_RETRIES, "-1") + .commit(); + long start = fastClock.nanoTime(); + assertEquals("no retries", 1, attemptsUntilGivingUp()); + assertEquals(0, millisSleptSince(start)); + } + + @Test + public void testTableRetryCountCapsTheAttempts() { + load().updateProperties().set(TableProperties.COMMIT_NUM_RETRIES, "2").commit(); + assertEquals(3, attemptsUntilGivingUp()); } // ---- create path @@ -789,15 +752,13 @@ private long commitTo( TableIdentifier id, SchemaEvolutionConfig config, IncompatibleSchemaHandling handling, - CommitSchemaUnion.TableCreation creation, + CommitSchemaUnion.NewTableSettings creation, CollectDistinctSchemas.SchemaGroup... schemas) { return CommitSchemaUnion.commit( catalog, id, Arrays.asList(schemas), - config, - handling, - creation, + settings(config, handling, creation), CommitSchemaUnion.DEFAULT_COMMITTER); } @@ -812,8 +773,8 @@ public void testMissingTableIsCreatedFromTheUnion() { Schema other = new Schema( required(1, "id", Types.LongType.get()), optional(2, "extra", Types.LongType.get())); - CommitSchemaUnion.TableCreation creation = - new CommitSchemaUnion.TableCreation( + CommitSchemaUnion.NewTableSettings creation = + new CommitSchemaUnion.NewTableSettings( Arrays.asList("region"), null, java.util.Collections.singletonMap("k", "v")); long schemaId = commitTo( @@ -848,8 +809,8 @@ public void testPartitionFieldFromNonSeedSchemaResolves() { Schema other = new Schema( required(1, "id", Types.LongType.get()), optional(2, "extra", Types.StringType.get())); - CommitSchemaUnion.TableCreation creation = - new CommitSchemaUnion.TableCreation(Arrays.asList("extra"), null, null); + CommitSchemaUnion.NewTableSettings creation = + new CommitSchemaUnion.NewTableSettings(Arrays.asList("extra"), null, null); commitTo( id, ALL, @@ -965,82 +926,6 @@ public void testUnmatchedPinWarnsAndCreatesUnderRouteToErrors() { new Schema(optional(1, "id", Types.LongType.get())), catalog.loadTable(id).schema()); } - // ---- createdSchema (direct) - - @Test - public void testCreatedSchemaPinsHoldAtEveryLevel() { - Schema merged = - new Schema( - required(1, "id", Types.LongType.get()), - optional( - 2, - "l", - Types.ListType.ofOptional( - 3, Types.StructType.of(optional(4, "q", Types.IntegerType.get()))))); - SchemaEvolutionConfig pinned = - SchemaEvolutionConfig.builder() - .setOptions(EnumSet.allOf(SchemaEvolutionOption.class)) - .setRequiredColumns(Collections.singleton("l.element.q")) - .build(); - Schema created = CommitSchemaUnion.createdSchema(merged, pinned); - assertSameSchema( - new Schema( - optional(1, "id", Types.LongType.get()), - required( - 2, - "l", - Types.ListType.ofRequired( - 3, Types.StructType.of(required(4, "q", Types.IntegerType.get()))))), - created); - } - - @Test - public void testCreatedSchemaEveryLevelOptionalExceptMapKeys() { - Schema schema = - new Schema( - required(1, "id", Types.LongType.get()), - required( - 2, - "s", - Types.StructType.of( - required(3, "a", Types.IntegerType.get()), - required( - 4, - "items", - Types.ListType.ofRequired( - 5, Types.StructType.of(required(6, "qty", Types.IntegerType.get())))))), - required( - 7, - "attrs", - Types.MapType.ofRequired( - 8, - 9, - Types.StructType.of(required(10, "k", Types.StringType.get())), - Types.StructType.of(required(11, "v", Types.IntegerType.get()))))); - assertSameSchema( - new Schema( - optional(1, "id", Types.LongType.get()), - optional( - 2, - "s", - Types.StructType.of( - optional(3, "a", Types.IntegerType.get()), - optional( - 4, - "items", - Types.ListType.ofOptional( - 5, Types.StructType.of(optional(6, "qty", Types.IntegerType.get())))))), - optional( - 7, - "attrs", - Types.MapType.ofOptional( - 8, - 9, - Types.StructType.of(required(10, "k", Types.StringType.get())), - Types.StructType.of(optional(11, "v", Types.IntegerType.get()))))), - CommitSchemaUnion.createdSchema(schema, ALL)); - } - /** Options guard an existing table's schema; with no table there is nothing to guard. */ @Test public void testCreationIsNotGatedOnAnyParticularOption() { @@ -1068,9 +953,7 @@ public void testMissingTableWithoutSchemasIsNotCreated() { catalog, id, new ArrayList<>(), - ALL, - IncompatibleSchemaHandling.FAIL_PIPELINE, - NO_CREATION, + settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION), CommitSchemaUnion.DEFAULT_COMMITTER); assertEquals(CommitSchemaUnion.NO_TABLE, result); assertFalse(catalog.tableExists(id)); @@ -1098,6 +981,58 @@ public void testConflictOnCreateFailsWithoutCreating() { assertFalse(catalog.tableExists(id)); } + /** The fallback creation at registration throws the same error, so no handling can route it. */ + @Test + public void testPartitionFieldAbsentFromUnionFailsCreationUnderEitherHandling() { + TableIdentifier id = missing(); + Schema seed = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "region", Types.StringType.get())); + CommitSchemaUnion.NewTableSettings creation = + new CommitSchemaUnion.NewTableSettings(Arrays.asList("missing"), null, null); + for (IncompatibleSchemaHandling handling : IncompatibleSchemaHandling.values()) { + IllegalStateException e = + assertThrows( + IllegalStateException.class, + () -> commitTo(id, ALL, handling, creation, files(seed, 2))); + assertEquals( + "not an IncompatibleSchemaException, which the handling could downgrade", + IllegalStateException.class, + e.getClass()); + assertThat( + e.getMessage(), containsString("cannot be created with partition fields [missing]")); + } + assertFalse(catalog.tableExists(id)); + } + + @Test + public void testPartitionFieldOnlyInASkippedSchemaFailsCreation() { + TableIdentifier id = missing(); + Schema seed = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "code", Types.StringType.get())); + Schema loser = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "code", Types.LongType.get()), + optional(3, "region", Types.StringType.get())); + CommitSchemaUnion.NewTableSettings creation = + new CommitSchemaUnion.NewTableSettings(Arrays.asList("region"), null, null); + IllegalStateException e = + assertThrows( + IllegalStateException.class, + () -> + commitTo( + id, + ALL, + IncompatibleSchemaHandling.ROUTE_TO_ERRORS, + creation, + files(seed, 3), + files(loser, 1))); + assertThat(e.getMessage(), containsString("region")); + assertFalse(catalog.tableExists(id)); + } + @Test public void testConflictOnCreateRoutesLoserAndCreates() { TableIdentifier id = missing(); @@ -1142,10 +1077,9 @@ public void testCreateRaceFallsBackToEvolvingTheExistingTable() { catalog, id, Arrays.asList(files(file, 1)), - ALL, - IncompatibleSchemaHandling.FAIL_PIPELINE, - NO_CREATION, - raced); + settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION), + raced, + fastClock); Table table = catalog.loadTable(id); assertEquals(2, attempts.get()); assertNotNull(table.schema().findField("email")); @@ -1156,16 +1090,13 @@ public void testCreateRaceFallsBackToEvolvingTheExistingTable() { @Test public void testEmptyInputCommitsNothing() { - seedNameMapping(); String before = metadataLocation(load()); List none = new ArrayList<>(); CommitSchemaUnion.commit( catalog, tableId, none, - ALL, - IncompatibleSchemaHandling.FAIL_PIPELINE, - NO_CREATION, + settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION), CommitSchemaUnion.DEFAULT_COMMITTER); assertEquals(before, metadataLocation(load())); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java index 5cac095201db..4f17bc738400 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java @@ -253,7 +253,7 @@ private static long counter(PipelineResult result, String name) { private PCollection run(String... paths) { return pipeline .apply(Create.of(Arrays.asList(paths))) - .apply(ParDo.of(new ReadFooterSchema())) + .apply(ParDo.of(new ReadFooterSchema(SchemaEvolutionConfig.disabled()))) .setCoder(CollectDistinctSchemas.groupCoder()); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java index c6d565cda957..c76c06d9b7c7 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java @@ -116,4 +116,14 @@ public void testIncompatibleSchemaHandlingDefaultsByMode() { IncompatibleSchemaHandling.ROUTE_TO_ERRORS, forced.incompatibleSchemaHandlingFor(PCollection.IsBounded.BOUNDED)); } + + /** A dry run of nothing is a config error, not a silent real run. */ + @Test + public void testDryRunAloneIsRejectedAtBuild() { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> SchemaEvolutionConfig.builder().setDryRun(true).build()); + assertTrue(e.getMessage(), e.getMessage().contains("dry run")); + } } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaPlanTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaPlanTest.java new file mode 100644 index 000000000000..12d34d773b84 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaPlanTest.java @@ -0,0 +1,325 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumSet; +import java.util.List; +import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.hadoop.HadoopCatalog; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.junit.Before; +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.rules.TestName; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Planning without committing; what a plan does once committed is CommitSchemaUnionTest's. */ +@RunWith(JUnit4.class) +public class SchemaPlanTest { + @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + + @Rule + public transient TestDataWarehouse warehouse = new TestDataWarehouse(TEMPORARY_FOLDER, "default"); + + @Rule public TestName testName = new TestName(); + + private static final Schema TABLE = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "name", Types.StringType.get()), + optional(3, "score", Types.FloatType.get()), + required(4, "region", Types.StringType.get())); + + private static final SchemaEvolutionConfig ALL = + SchemaEvolutionConfig.of(SchemaEvolutionOption.values()); + + private static final CommitSchemaUnion.Settings SETTINGS = + new CommitSchemaUnion.Settings( + ALL, + IncompatibleSchemaHandling.FAIL_PIPELINE, + new CommitSchemaUnion.NewTableSettings(null, null, null)); + + private HadoopCatalog catalog; + private TableIdentifier tableId; + + @Before + public void setUp() { + catalog = new HadoopCatalog(new Configuration(), warehouse.location); + tableId = TableIdentifier.of("default", testName.getMethodName()); + warehouse.createTable(tableId, TABLE); + } + + private static String json(Schema schema) { + return SchemaParser.toJson(FileSchemas.canonical(schema)); + } + + private static CollectDistinctSchemas.SchemaGroup files(Schema schema, long count) { + return CollectDistinctSchemas.SchemaGroup.of(json(schema), count, Collections.emptyList()); + } + + private Table load() { + return catalog.loadTable(tableId); + } + + /** Full-schema comparison, field ids normalized; string form so a failure shows the diff. */ + private static void assertSameSchema(Schema expected, Schema actual) { + assertEquals( + TypeUtil.assignIncreasingFreshIds(expected).asStruct().toString(), + TypeUtil.assignIncreasingFreshIds(actual).asStruct().toString()); + } + + // ---- plans + + /** The mapping repair is a plan decision, made before anything is committed. */ + @Test + public void testPlanReportsTheNameMappingRepair() { + Table table = load(); + assertFalse(table.properties().containsKey(TableProperties.DEFAULT_NAME_MAPPING)); + List covered = Arrays.asList(files(TABLE, 1)); + SchemaPlan.Evolution plan = + (SchemaPlan.Evolution) SchemaPlan.compute(catalog, tableId, table, covered, SETTINGS); + assertNull(plan.newSchema); + assertTrue(plan.repairsNameMapping); + + table + .updateProperties() + .set( + TableProperties.DEFAULT_NAME_MAPPING, NameMappingUtils.regenerate(table.schema(), null)) + .commit(); + plan = (SchemaPlan.Evolution) SchemaPlan.compute(catalog, tableId, load(), covered, SETTINGS); + assertFalse(plan.repairsNameMapping); + } + + @Test + public void testPlanForCreationListsAcceptedSchemasAndCreationSettings() { + TableIdentifier missing = TableIdentifier.of("default", testName.getMethodName() + "_new"); + Schema seed = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "region", Types.StringType.get())); + Schema other = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "extra", Types.LongType.get())); + CommitSchemaUnion.NewTableSettings creation = + new CommitSchemaUnion.NewTableSettings(Arrays.asList("region"), Arrays.asList("id"), null); + SchemaPlan.Creation plan = + (SchemaPlan.Creation) + SchemaPlan.compute( + catalog, + missing, + null, + Arrays.asList(files(seed, 5), files(other, 1)), + new CommitSchemaUnion.Settings( + ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, creation)); + assertTrue(plan.canCreate()); + assertEquals(2, plan.schemasToMerge.size()); + assertEquals(5, checkStateNotNull(plan.toMerge(json(seed))).files); + assertEquals(1, checkStateNotNull(plan.toMerge(json(other))).files); + assertEquals("region", checkStateNotNull(plan.spec).fields().get(0).name()); + assertEquals(1, checkStateNotNull(plan.sortOrder).fields().size()); + assertFalse(catalog.tableExists(missing)); + } + + // ---- newRequiredPaths (direct) + + @Test + public void testNewRequiredPathsAtEveryLevelExceptMapKeys() { + Schema before = new Schema(required(1, "id", Types.LongType.get())); + Schema after = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "s", + Types.StructType.of( + required(3, "a", Types.IntegerType.get()), + optional(4, "b", Types.IntegerType.get()))), + optional( + 5, + "items", + Types.ListType.ofRequired( + 6, Types.StructType.of(required(7, "qty", Types.IntegerType.get())))), + optional( + 8, + "attrs", + Types.MapType.ofRequired( + 9, + 10, + Types.StructType.of(required(11, "k", Types.StringType.get())), + Types.StructType.of(required(12, "v", Types.IntegerType.get()))))); + assertEquals( + Arrays.asList("s.a", "items.element", "items.element.qty", "attrs.value", "attrs.value.v"), + SchemaPlan.newRequiredPaths(before, after)); + } + + /** Names containing element/key/value are not containers; regression for a substring check. */ + @Test + public void testNewRequiredPathsContainerLikeNamesAreNotContainers() { + Schema before = new Schema(required(1, "id", Types.LongType.get())); + Schema after = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "stats", + Types.StructType.of( + required(3, "keyword", Types.StringType.get()), + required(4, "value_sum", Types.LongType.get()), + required(5, "element", Types.StringType.get())))); + assertEquals( + Arrays.asList("stats.keyword", "stats.value_sum", "stats.element"), + SchemaPlan.newRequiredPaths(before, after)); + } + + @Test + public void testNewRequiredPathsInNestedContainers() { + Schema before = new Schema(required(1, "id", Types.LongType.get())); + Schema after = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "ll", + Types.ListType.ofRequired( + 3, Types.ListType.ofRequired(4, Types.IntegerType.get()))), + optional( + 5, + "lm", + Types.ListType.ofRequired( + 6, + Types.MapType.ofRequired( + 7, 8, Types.StringType.get(), Types.IntegerType.get())))); + assertEquals( + Arrays.asList("ll.element", "ll.element.element", "lm.element", "lm.element.value"), + SchemaPlan.newRequiredPaths(before, after)); + } + + /** Growing an existing struct: only the field with a new id is a candidate. */ + @Test + public void testNewRequiredPathsIgnoreExistingFields() { + Schema before = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "s", Types.StructType.of(required(3, "old", Types.IntegerType.get())))); + Schema after = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "s", + Types.StructType.of( + required(3, "old", Types.IntegerType.get()), + required(4, "fresh", Types.IntegerType.get())))); + assertEquals(Arrays.asList("s.fresh"), SchemaPlan.newRequiredPaths(before, after)); + } + + // ---- createdSchema (direct) + + @Test + public void testCreatedSchemaPinsHoldAtEveryLevel() { + Schema merged = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "l", + Types.ListType.ofOptional( + 3, Types.StructType.of(optional(4, "q", Types.IntegerType.get()))))); + SchemaEvolutionConfig pinned = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.allOf(SchemaEvolutionOption.class)) + .setRequiredColumns(Collections.singleton("l.element.q")) + .build(); + Schema created = SchemaPlan.createdSchema(merged, pinned); + assertSameSchema( + new Schema( + optional(1, "id", Types.LongType.get()), + required( + 2, + "l", + Types.ListType.ofRequired( + 3, Types.StructType.of(required(4, "q", Types.IntegerType.get()))))), + created); + } + + @Test + public void testCreatedSchemaEveryLevelOptionalExceptMapKeys() { + Schema schema = + new Schema( + required(1, "id", Types.LongType.get()), + required( + 2, + "s", + Types.StructType.of( + required(3, "a", Types.IntegerType.get()), + required( + 4, + "items", + Types.ListType.ofRequired( + 5, Types.StructType.of(required(6, "qty", Types.IntegerType.get())))))), + required( + 7, + "attrs", + Types.MapType.ofRequired( + 8, + 9, + Types.StructType.of(required(10, "k", Types.StringType.get())), + Types.StructType.of(required(11, "v", Types.IntegerType.get()))))); + assertSameSchema( + new Schema( + optional(1, "id", Types.LongType.get()), + optional( + 2, + "s", + Types.StructType.of( + optional(3, "a", Types.IntegerType.get()), + optional( + 4, + "items", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "qty", Types.IntegerType.get())))))), + optional( + 7, + "attrs", + Types.MapType.ofOptional( + 8, + 9, + Types.StructType.of(required(10, "k", Types.StringType.get())), + Types.StructType.of(optional(11, "v", Types.IntegerType.get()))))), + SchemaPlan.createdSchema(schema, ALL)); + } +} diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml index c41dd6f2e07f..2d2c2af6da13 100644 --- a/sdks/python/apache_beam/yaml/standard_io.yaml +++ b/sdks/python/apache_beam/yaml/standard_io.yaml @@ -661,6 +661,7 @@ required_columns: 'required_columns' incompatible_schema_handling: 'incompatible_schema_handling' unverifiable_file_handling: 'unverifiable_file_handling' + dry_run: 'dry_run' error_handling: 'error_handling' underlying_provider: type: beamJar diff --git a/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dry_run.yaml b/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dry_run.yaml new file mode 100644 index 000000000000..75b7b1a16ddc --- /dev/null +++ b/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dry_run.yaml @@ -0,0 +1,185 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +fixtures: + - name: TEMP_DIR + type: "tempfile.TemporaryDirectory" + + +pipelines: + # Pipeline 1: files with two columns + - pipeline: + type: chain + transforms: + - type: Create + config: + elements: + - {label: "11a", rank: 0} + - {label: "37a", rank: 1} + - type: WriteToParquet + config: + path: "{TEMP_DIR}/data/narrow" + file_name_suffix: ".parquet" + num_shards: 1 + + # Pipeline 2: files with an extra column + - pipeline: + type: chain + transforms: + - type: Create + config: + elements: + - {label: "389a", rank: 2, bool: false} + - {label: "3821b", rank: 3, bool: true} + - type: WriteToParquet + config: + path: "{TEMP_DIR}/data/wide" + file_name_suffix: ".parquet" + num_shards: 1 + + # Pipeline 3: a dry run against a table that does not exist yet; the report is written as JSON + # from the dry_run_report output, the way a user keeps it + - pipeline: + transforms: + - type: ReadMatchFiles + name: Match + config: + file_pattern: "{TEMP_DIR}/data/*.parquet" + - type: MapToFields + name: Paths + input: Match + config: + language: python + fields: + path: + callable: "lambda row: row.path" + output_type: string + - type: IcebergAddFiles + name: DryRun + input: Paths + config: + table: "default.table" + catalog_properties: + type: "hadoop" + warehouse: "{TEMP_DIR}/dir" + schema_evolution_options: [ALLOW_FIELD_ADDITION] + dry_run: true + - type: WriteToJson + name: WriteReport + input: DryRun.dry_run_report + config: + path: "{TEMP_DIR}/report.json" + # The full report, as a real run against this input would decide it. The report is one + # row with nested fields; the two assertions below read it the way a user would: the + # verdict, the file counts and the table a real run would create (from the union, every + # column optional) come from the top level, the per-schema entries from an Explode. + - type: MapToFields + name: Verdict + input: DryRun.dry_run_report + config: + language: python + fields: + allowed: allowed + reason: reason + would_create_table: would_create_table + files_allowed: files_allowed + files_incompatible: files_incompatible + files_unreadable: files_unreadable + files_unchecked: files_unchecked + unchecked_registered: unchecked_registered + table_changes: table_changes + config_problems: config_problems + created_schema: created_table.schema + created_columns: created_table.columns + - type: AssertEqual + name: AssertVerdict + input: Verdict + config: + elements: + - allowed: true + reason: "" + would_create_table: true + files_allowed: 2 + files_incompatible: 0 + files_unreadable: 0 + files_unchecked: 0 + unchecked_registered: false + table_changes: [] + config_problems: [] + created_schema: '{{"type":"struct","schema-id":0,"fields":[{{"id":1,"name":"bool","required":false,"type":"boolean"}},{{"id":2,"name":"label","required":false,"type":"string"}},{{"id":3,"name":"rank","required":false,"type":"long"}}]}}' + created_columns: ["create optional bool boolean", "create optional label string", "create optional rank long"] + # one entry per distinct schema (canonical JSON), most common first + - type: Explode + name: Schemas + input: DryRun.dry_run_report + config: + fields: [schemas] + - type: MapToFields + name: SchemaEntries + input: Schemas + config: + language: python + fields: + schema_key: schemas.schema_key + schema: schemas.schema + num_files: schemas.num_files + changes: schemas.changes + allowed: schemas.allowed + reason: schemas.reason + - type: AssertEqual + name: AssertSchemas + input: SchemaEntries + config: + elements: + - schema_key: "s27e62d" + schema: '{{"type":"struct","schema-id":0,"fields":[{{"id":1,"name":"bool","required":true,"type":"boolean"}},{{"id":2,"name":"label","required":true,"type":"string"}},{{"id":3,"name":"rank","required":true,"type":"long"}}]}}' + num_files: 1 + changes: [] + allowed: true + reason: "" + - schema_key: "s48334d" + schema: '{{"type":"struct","schema-id":0,"fields":[{{"id":1,"name":"label","required":true,"type":"string"}},{{"id":2,"name":"rank","required":true,"type":"long"}}]}}' + num_files: 1 + changes: [] + allowed: true + reason: "" + + providers: + - type: python + config: { } + transforms: + ReadMatchFiles: 'apache_beam.io.fileio.MatchFiles' + + # Pipeline 4: the report was written (nested fields come back as strings through ReadFromJson, + # so only the scalar columns are compared here; the full row is asserted above) + - pipeline: + type: chain + transforms: + - type: ReadFromJson + config: + path: "{TEMP_DIR}/report.json*" + - type: MapToFields + config: + fields: + allowed: allowed + would_create_table: would_create_table + files_allowed: files_allowed + files_incompatible: files_incompatible + - type: AssertEqual + config: + elements: + - {allowed: true, would_create_table: true, files_allowed: 2, files_incompatible: 0}