You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by cv...@apache.org on 2026/09/23 19:15:57 UTC
(beam) branch master updated: AddFiles: Add dry_run (#40167)
This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 7a20b37023f AddFiles: Add dry_run (#40167)
7a20b37023f is described below
commit 7a20b37023f92c4d14b07d26d934ce0df5094f46
Author: claudevdm <33...@users.noreply.github.com>
AuthorDate: Wed Sep 23 15:15:47 2026 -0400
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<STRING>: the SchemaDelta descriptions on schema
rows; "create <optional|required> <name> <type>" 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 +-
.../org/apache/beam/sdk/io/iceberg/AddFiles.java | 119 ++--
.../iceberg/AddFilesSchemaTransformProvider.java | 41 +-
.../beam/sdk/io/iceberg/CommitSchemaOnce.java | 25 +-
.../beam/sdk/io/iceberg/CommitSchemaUnion.java | 671 +++++++--------------
.../apache/beam/sdk/io/iceberg/DryRunReport.java | 434 +++++++++++++
.../beam/sdk/io/iceberg/ReadFooterSchema.java | 33 +-
.../apache/beam/sdk/io/iceberg/SchemaDelta.java | 9 +-
.../beam/sdk/io/iceberg/SchemaEvolutionConfig.java | 34 +-
.../org/apache/beam/sdk/io/iceberg/SchemaPlan.java | 618 +++++++++++++++++++
.../AddFilesSchemaTransformProviderTest.java | 14 +
.../apache/beam/sdk/io/iceberg/AddFilesTest.java | 347 ++++++++++-
.../beam/sdk/io/iceberg/CommitSchemaOnceTest.java | 14 +-
.../beam/sdk/io/iceberg/CommitSchemaUnionTest.java | 345 +++++------
.../beam/sdk/io/iceberg/ReadFooterSchemaTest.java | 2 +-
.../sdk/io/iceberg/SchemaEvolutionConfigTest.java | 10 +
.../apache/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(-)
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
index e1430c74e62..6d9121eaaa4 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 75b1a0144e2..4093b2fb6e5 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.concurrent.atomic.AtomicBoolean;
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.StateSpec;
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 @@ import org.slf4j.LoggerFactory;
* and committed as snapshots.
*
* <p>Outputs: {@code snapshots} (one row per commit), {@code errors} (one row per file that could
- * not be registered: {@code file}, {@code error}).
+ * not be registered: {@code file}, {@code error}), and {@code dry_run_report} when a dry run is
+ * configured.
*
* <p><b>Schema evolution.</b> With a {@link SchemaEvolutionConfig} whose options are set, a
* pre-pass reads every Parquet footer, classifies the change each distinct file schema needs on the
@@ -135,7 +137,9 @@ import org.slf4j.LoggerFactory;
* {@code errors} (see {@link SchemaEvolutionConfig.IncompatibleSchemaHandling}). The per-file
* checks read Parquet footers: an ORC or Avro file, or a pinned column the footer has no null count
* for, cannot be verified and goes to {@code errors} unless {@link
- * SchemaEvolutionConfig.UnverifiableFileHandling#ACCEPT} registers it on trust. Schema evolution
+ * SchemaEvolutionConfig.UnverifiableFileHandling#ACCEPT} registers it on trust. When the table does
+ * not exist, the pre-pass creates it from the union of the file schemas; if no readable Parquet
+ * schema can seed it, nothing is created and every file goes to {@code errors}. Schema evolution
* currently requires bounded input; unbounded input with options set is rejected at construction.
*
* <pre>{@code
@@ -154,6 +158,10 @@ import org.slf4j.LoggerFactory;
public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTuple> {
static final String OUTPUT_TAG = "snapshots";
static final String ERROR_TAG = "errors";
+
+ /** Only present with {@link SchemaEvolutionConfig#getDryRun()}. */
+ static final String DRY_RUN_TAG = "dry_run_report";
+
private static final Duration DEFAULT_TRIGGER_INTERVAL = Duration.standardMinutes(10);
private static final Counter numManifestFilesAdded =
counter(AddFiles.class, "numManifestFilesAdded");
@@ -260,7 +268,20 @@ public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTupl
PCollection<String> paths = input;
if (evolution.isEnabled()) {
- paths = gateOnSchemaCommit(input);
+ // one commit per window, and for bounded input the window is the whole input
+ PCollection<String> windowed =
+ input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
+ PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas = distinctSchemas(windowed);
+ CommitSchemaUnion.Settings settings =
+ new CommitSchemaUnion.Settings(
+ evolution,
+ evolution.incompatibleSchemaHandlingFor(input.isBounded()),
+ new CommitSchemaUnion.NewTableSettings(partitionFields, sortFields, tableProps));
+ if (evolution.getDryRun()) {
+ return report(schemas, settings);
+ }
+ PCollection<Long> committed = commitSchema(schemas, settings);
+ paths = windowed.apply("WaitForSchemaCommit", Wait.on(committed));
}
PCollectionTuple dataFiles =
@@ -320,39 +341,49 @@ public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTupl
OUTPUT_TAG, snapshots, ERROR_TAG, dataFiles.get(ERRORS).setRowSchema(ERROR_SCHEMA));
}
- /**
- * Holds every path until the schema commit has landed. The pre-pass window is the unit of commit,
- * and for bounded input that unit is the whole input: paths are rewindowed into the global window
- * first, so one commit covers everything and the Wait.on signal lines up with the main input
- * whatever windowing the caller applied upstream.
- */
- private PCollection<String> gateOnSchemaCommit(PCollection<String> input) {
- PCollection<String> windowed =
- input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
- CommitSchemaUnion.TableCreation creation =
- new CommitSchemaUnion.TableCreation(partitionFields, sortFields, tableProps);
- // unset handling follows the mode (fail the job in batch, route in streaming); the pre-pass
- // only runs on bounded input, see expand
- IncompatibleSchemaHandling onIncompatible =
- evolution.incompatibleSchemaHandlingFor(input.isBounded());
- PCollection<Long> signal =
- windowed
- .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema()))
- .setCoder(CollectDistinctSchemas.groupCoder())
- .apply(
- "CollectDistinctSchemas",
- Combine.globally(new CollectDistinctSchemas()).withoutDefaults())
+ private PCollection<List<CollectDistinctSchemas.SchemaGroup>> distinctSchemas(
+ PCollection<String> windowed) {
+ return windowed
+ .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema(evolution)))
+ .setCoder(CollectDistinctSchemas.groupCoder())
+ .apply(
+ "CollectDistinctSchemas",
+ Combine.globally(new CollectDistinctSchemas()).withoutDefaults());
+ }
+
+ /** Commits the plan for the window's schemas once; the signal releases the gated paths. */
+ private PCollection<Long> commitSchema(
+ PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas,
+ CommitSchemaUnion.Settings settings) {
+ return schemas.apply(
+ "CommitSchemaOnce",
+ ParDo.of(new CommitSchemaOnce(catalogConfig, tableIdentifier, settings, committer)));
+ }
+
+ /** Reports the plan for the window's schemas instead; nothing is committed or registered. */
+ private PCollectionRowTuple report(
+ PCollection<List<CollectDistinctSchemas.SchemaGroup>> schemas,
+ CommitSchemaUnion.Settings settings) {
+ PCollection<Row> report =
+ schemas
.apply(
- "CommitSchemaOnce",
- ParDo.of(
- new CommitSchemaOnce(
- catalogConfig,
- tableIdentifier,
- evolution,
- onIncompatible,
- creation,
- committer)));
- return windowed.apply("WaitForSchemaCommit", Wait.on(signal));
+ "DryRunReport",
+ ParDo.of(new DryRunReport(catalogConfig, tableIdentifier, settings)))
+ .setRowSchema(DryRunReport.REPORT_SCHEMA);
+
+ PCollection<Row> emptySnapshots =
+ schemas
+ .getPipeline()
+ .apply("NoSnapshots", Create.empty(RowCoder.of(SnapshotInfo.getSchema())))
+ .setRowSchema(SnapshotInfo.getSchema());
+ PCollection<Row> emptyErrors =
+ schemas
+ .getPipeline()
+ .apply("NoErrors", Create.empty(RowCoder.of(ERROR_SCHEMA)))
+ .setRowSchema(ERROR_SCHEMA);
+ return PCollectionRowTuple.of(OUTPUT_TAG, emptySnapshots)
+ .and(ERROR_TAG, emptyErrors)
+ .and(DRY_RUN_TAG, report);
}
/** Test hook: how the schema pre-pass commits its transaction. */
@@ -394,6 +425,7 @@ public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTupl
private final SchemaEvolutionConfig evolution;
private transient @MonotonicNonNull BoundedAsyncTasks<ProcessResult> tasks;
private transient volatile @MonotonicNonNull Table table;
+ private transient volatile boolean tableMissing;
private transient @MonotonicNonNull Set<String> warned;
private final AtomicBoolean refreshedThisBundle = new AtomicBoolean();
@@ -441,6 +473,9 @@ public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTupl
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 @@ public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTupl
// ---- 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 @@ public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTupl
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 @@ public class AddFiles extends PTransform<PCollection<String>, PCollectionRowTupl
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 fa016ddca86..c5cc758c555 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 class AddFilesSchemaTransformProvider extends TypedSchemaTransformProvide
public abstract @Nullable List<String> getSortFields();
@SchemaFieldDescription(
- "Lets the transform change the table schema so that every file's columns are covered."
+ "Lets the transform change the table schema so that the table has a column for every"
+ + " column the files have."
+ " Values: ALLOW_FIELD_ADDITION (columns a file has and the table lacks are added, as"
+ " optional), ALLOW_FIELD_RELAXATION (a required table column becomes optional when a"
+ " file lacks it or may hold nulls in it), ALLOW_TYPE_PROMOTION (a column type is"
@@ -131,8 +132,10 @@ public class AddFilesSchemaTransformProvider extends TypedSchemaTransformProvide
+ " that is not allowed is incompatible; see incompatible_schema_handling. Only"
+ " Parquet files can be checked: ORC and Avro files are sent to the error output"
+ " unless unverifiable_file_handling is ACCEPT. Files sent to the error output are"
- + " dropped unless error_handling is set. Batch pipelines only; streaming pipelines"
- + " cannot use schema evolution yet.")
+ + " dropped unless error_handling is set. If the table does not exist it is created"
+ + " from the union of the Parquet schemas; with none, every file goes to the error"
+ + " output. Batch pipelines only; streaming pipelines cannot use schema evolution"
+ + " yet.")
public abstract @Nullable List<String> getSchemaEvolutionOptions();
@SchemaFieldDescription(
@@ -145,6 +148,19 @@ public class AddFilesSchemaTransformProvider extends TypedSchemaTransformProvide
+ " Requires schema_evolution_options.")
public abstract @Nullable List<String> getRequiredColumns();
+ @SchemaFieldDescription(
+ "When true, nothing is committed or registered: the transform reads the files' schemas"
+ + " and emits a `dry_run_report` output with one row that describes what a real run"
+ + " would do. Its `allowed` field is true when every file schema can be merged and the"
+ + " configuration raises no problem; otherwise its `reason` field says what a real run"
+ + " would do about it (fail, or route the files to the error output). Its `schemas`"
+ + " field lists each distinct file schema with the changes a real run would make for"
+ + " it and, when it cannot be merged, why. The output only exists when this is set;"
+ + " consume it as input: `<this transform's name>.dry_run_report`. Against a missing"
+ + " table, a REST catalog needs table-create permission even though no table is"
+ + " created.")
+ public abstract @Nullable Boolean getDryRun();
+
@SchemaFieldDescription(
"What happens when a file's schema cannot be made to fit the table: it needs a change"
+ " that is not allowed, or it conflicts with the table or with another file."
@@ -158,8 +174,9 @@ public class AddFilesSchemaTransformProvider extends TypedSchemaTransformProvide
+ " 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 class AddFilesSchemaTransformProvider extends TypedSchemaTransformProvide
public abstract Builder setUnverifiableFileHandling(String handling);
+ public abstract Builder setDryRun(Boolean dryRun);
+
public abstract Configuration build();
}
@@ -207,19 +226,21 @@ public class AddFilesSchemaTransformProvider extends TypedSchemaTransformProvide
List<String> pins = getRequiredColumns();
String handlingName = getIncompatibleSchemaHandling();
String unverifiableName = getUnverifiableFileHandling();
+ boolean dryRun = Boolean.TRUE.equals(getDryRun());
boolean nothingSet =
(optionNames == null || optionNames.isEmpty())
&& (pins == null || pins.isEmpty())
&& handlingName == null
- && unverifiableName == null;
+ && unverifiableName == null
+ && !dryRun;
if (nothingSet) {
return null;
}
// SchemaEvolutionConfig.build() checks this too; this copy names the YAML keys
Preconditions.checkArgument(
optionNames != null && !optionNames.isEmpty(),
- "required_columns, incompatible_schema_handling and unverifiable_file_handling need at"
- + " least one schema_evolution_options entry");
+ "required_columns, incompatible_schema_handling, unverifiable_file_handling and"
+ + " dry_run need at least one schema_evolution_options entry");
Set<SchemaEvolutionOption> options = EnumSet.noneOf(SchemaEvolutionOption.class);
for (String name : checkStateNotNull(optionNames)) {
options.add(parseEnum(SchemaEvolutionOption.class, name, "schema_evolution_options"));
@@ -228,6 +249,7 @@ public class AddFilesSchemaTransformProvider extends TypedSchemaTransformProvide
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 class AddFilesSchemaTransformProvider extends TypedSchemaTransformProvide
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 6fd5171c2ea..768fb564de9 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 @@ package org.apache.beam.sdk.io.iceberg;
import static org.apache.beam.sdk.metrics.Metrics.counter;
import java.util.List;
-import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.iceberg.catalog.Catalog;
@@ -37,34 +36,23 @@ class CommitSchemaOnce extends DoFn<List<CollectDistinctSchemas.SchemaGroup>, Lo
private final IcebergCatalogConfig catalogConfig;
private final String identifier;
- private final SchemaEvolutionConfig config;
- private final IncompatibleSchemaHandling handling;
- private final CommitSchemaUnion.TableCreation creation;
+ private final CommitSchemaUnion.Settings settings;
private final CommitSchemaUnion.Committer committer;
private transient @MonotonicNonNull Catalog catalog;
CommitSchemaOnce(
- IcebergCatalogConfig catalogConfig,
- String identifier,
- SchemaEvolutionConfig config,
- IncompatibleSchemaHandling handling,
- CommitSchemaUnion.TableCreation creation) {
- this(
- catalogConfig, identifier, config, handling, creation, CommitSchemaUnion.DEFAULT_COMMITTER);
+ IcebergCatalogConfig catalogConfig, String identifier, CommitSchemaUnion.Settings settings) {
+ this(catalogConfig, identifier, settings, CommitSchemaUnion.DEFAULT_COMMITTER);
}
CommitSchemaOnce(
IcebergCatalogConfig catalogConfig,
String identifier,
- SchemaEvolutionConfig config,
- IncompatibleSchemaHandling handling,
- CommitSchemaUnion.TableCreation creation,
+ CommitSchemaUnion.Settings settings,
CommitSchemaUnion.Committer committer) {
this.catalogConfig = catalogConfig;
this.identifier = identifier;
- this.config = config;
- this.handling = handling;
- this.creation = creation;
+ this.settings = settings;
this.committer = committer;
}
@@ -82,8 +70,7 @@ class CommitSchemaOnce extends DoFn<List<CollectDistinctSchemas.SchemaGroup>, Lo
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 1e145a122c8..09374979878 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.Collections;
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.
*
- * <p>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.
+ * <p>When the table does not exist, {@code create} builds it from the planned union instead, with
+ * pinned columns and their ancestors required.
*
* <p>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 @@ import org.slf4j.LoggerFactory;
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<String> partitionFields;
final @Nullable List<String> sortFields;
final @Nullable Map<String, String> properties;
- TableCreation(
+ NewTableSettings(
@Nullable List<String> partitionFields,
@Nullable List<String> sortFields,
@Nullable Map<String, String> properties) {
@@ -93,6 +101,25 @@ final class CommitSchemaUnion {
}
}
+ /** 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 @@ final class CommitSchemaUnion {
}
}
- 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 @@ final class CommitSchemaUnion {
Catalog catalog,
TableIdentifier tableId,
List<CollectDistinctSchemas.SchemaGroup> 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<CollectDistinctSchemas.SchemaGroup> 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> {
+ 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> T withRetry(
+ Catalog catalog, TableIdentifier tableId, Sleeper sleeper, Attempt<T> 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<String, String> 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<CollectDistinctSchemas.SchemaGroup> 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<CollectDistinctSchemas.SchemaGroup> 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> incompatible = new ArrayList<>();
- List<Accepted> 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<Accepted> classify(
- Table table,
- List<CollectDistinctSchemas.SchemaGroup> schemas,
- SchemaEvolutionConfig config,
- List<Incompatible> incompatible) {
- List<Accepted> 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> accepted,
- List<Incompatible> 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 @@ final class CommitSchemaUnion {
* append after them.
*/
private static long create(
+ SchemaPlan.Creation plan,
Catalog catalog,
TableIdentifier tableId,
- List<CollectDistinctSchemas.SchemaGroup> 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> 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<String, String> 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<String, String> 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<CollectDistinctSchemas.SchemaGroup> schemas,
- List<Incompatible> 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<Accepted> valid = new ArrayList<>();
- for (CollectDistinctSchemas.SchemaGroup group : schemas) {
- Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson());
- List<SchemaChange> 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<Accepted> 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<String> 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<Types.NestedField> 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<Types.NestedField> 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> incompatible,
+ List<SchemaPlan.IncompatibleSchema> 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 @@ final class CommitSchemaUnion {
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> accepted, List<Incompatible> 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<SchemaChange> 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<SchemaChange> changes) {
- List<String> 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<String> 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<String> 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<String> newRequiredPaths(Schema before, Schema after) {
- Set<Integer> beforeIds = TypeUtil.indexById(before.asStruct()).keySet();
- List<String> paths = new ArrayList<>();
- collectNewRequired(after.asStruct(), "", beforeIds, paths);
- return paths;
- }
-
- private static void collectNewRequired(
- Type.NestedType type, String prefix, Set<Integer> beforeIds, List<String> 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<Incompatible> items) {
+ private static String joinLines(List<SchemaPlan.IncompatibleSchema> items) {
List<String> 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 00000000000..ae069598e0a
--- /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.
+ *
+ * <p>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.
+ *
+ * <p>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<List<CollectDistinctSchemas.SchemaGroup>, 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<String> changes;
+ final boolean allowed;
+ final String reason;
+
+ SchemaEntry(
+ String key,
+ String schema,
+ long files,
+ List<String> 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<CollectDistinctSchemas.SchemaGroup> readable = new ArrayList<>();
+ long unreadableFiles;
+ long uncheckedFiles;
+
+ static Input of(List<CollectDistinctSchemas.SchemaGroup> 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<SchemaEntry> 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<CollectDistinctSchemas.SchemaGroup> schemas, OutputReceiver<Row> 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<SchemaEntry> 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<String> configProblems = new ArrayList<>(plan.configProblems);
+ if (creationProblem != null) {
+ configProblems.add(creationProblem);
+ }
+ List<String> 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<Row> 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<String> 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<String> 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<String> configProblems,
+ @Nullable String creationProblem,
+ boolean noTable) {
+ IncompatibleSchemaHandling handling = settings.handling;
+ List<String> 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<SchemaEntry> entries, List<String> 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<String> cut(List<String> changes) {
+ if (changes.size() <= MAX_RENDERED_CHANGES) {
+ return changes;
+ }
+ List<String> 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<String> createdColumns(org.apache.iceberg.Schema created) {
+ List<String> 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 e559feab351..389f03490fe 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<String, CollectDistinctSchemas.SchemaGroup>
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<ReadResult> 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 @@ class ReadFooterSchema extends DoFn<String, CollectDistinctSchemas.SchemaGroup>
}
}
- private static Callable<ReadResult> createReadTask(
+ private Callable<ReadResult> 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 @@ class ReadFooterSchema extends DoFn<String, CollectDistinctSchemas.SchemaGroup>
"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 bacdac2e7cf..743558e68f3 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 @@ final class SchemaDelta {
* (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 fcbfc906f3c..da6fd33ac9d 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 @@ import org.checkerframework.checker.nullness.qual.Nullable;
* 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.
+ *
+ * <p><b>Dry run.</b> 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 abstract class SchemaEvolutionConfig implements Serializable {
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 abstract class SchemaEvolutionConfig implements Serializable {
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 class SchemaEvolutionConfig implements Serializable {
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 abstract class SchemaEvolutionConfig implements Serializable {
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 00000000000..e9e72199c82
--- /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<SchemaToMerge> schemasToMerge;
+ final List<IncompatibleSchema> 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<String> configProblems;
+
+ private SchemaPlan(Verdicts verdicts, @Nullable Schema newSchema, List<String> 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<String> 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<String> 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<String> 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<String> 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<SchemaToMerge> toMerge = new ArrayList<>();
+ final List<IncompatibleSchema> 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<CollectDistinctSchemas.SchemaGroup> 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<CollectDistinctSchemas.SchemaGroup> 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<CollectDistinctSchemas.SchemaGroup> 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<String> 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<String> configProblems) {
+ List<String> 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<CollectDistinctSchemas.SchemaGroup> 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<CollectDistinctSchemas.SchemaGroup> schemas) {
+ Verdicts verdicts = new Verdicts();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson());
+ List<SchemaChange> 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<SchemaToMerge> 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<String> unenforceablePins(Schema created, SchemaEvolutionConfig config) {
+ List<String> 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<Types.NestedField> 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<Types.NestedField> 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<SchemaToMerge> 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<SchemaChange> 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<SchemaChange> changes) {
+ List<String> 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<String> 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<String> newRequiredPaths(Schema before, Schema after) {
+ Set<Integer> beforeIds = TypeUtil.indexById(before.asStruct()).keySet();
+ List<String> paths = new ArrayList<>();
+ collectNewRequired(after.asStruct(), "", beforeIds, paths);
+ return paths;
+ }
+
+ private static void collectNewRequired(
+ Type.NestedType type, String prefix, Set<Integer> beforeIds, List<String> 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<String, String> 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 2e469ef6da7..3b58af20154 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 class AddFilesSchemaTransformProviderTest {
.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 5e6cbdf4ea3..b9fe94f27a5 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.MatcherAssert.assertThat;
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.File;
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.Immuta
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 class AddFilesTest {
"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<Row> schemas = schemas(report);
+ assertEquals(3, schemas.size());
+ boolean sawAddition = false;
+ boolean sawConflict = false;
+ for (Row schema : schemas) {
+ Collection<String> 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<Row> rows) {
+ return Iterables.getOnlyElement(rows);
+ }
+
+ private static Collection<Row> 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<Row> 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 class AddFilesTest {
/**
* 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 class AddFilesTest {
.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<Long> 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 class AddFilesTest {
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 @@ public class AddFilesTest {
}
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<Long> 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 e3d6889b161..07a0adeb6cf 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 class CommitSchemaOnceTest {
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 class CommitSchemaOnceTest {
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 7df6af49920..0c5f379a110 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 @@ package org.apache.beam.sdk.io.iceberg;
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 java.util.concurrent.atomic.AtomicInteger;
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 @@ public class CommitSchemaUnionTest {
catalog,
tableId,
Arrays.asList(schemas),
- config,
- handling,
- NO_CREATION,
+ settings(config, handling, NO_CREATION),
CommitSchemaUnion.DEFAULT_COMMITTER);
}
@@ -260,99 +272,6 @@ public class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 @@ public class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
@Test
public void testEmptyInputCommitsNothing() {
- seedNameMapping();
String before = metadataLocation(load());
List<CollectDistinctSchemas.SchemaGroup> 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 5cac095201d..4f17bc73840 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 @@ public class ReadFooterSchemaTest {
private PCollection<CollectDistinctSchemas.SchemaGroup> 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 c6d565cda95..c76c06d9b7c 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 class SchemaEvolutionConfigTest {
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 00000000000..12d34d773b8
--- /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<CollectDistinctSchemas.SchemaGroup> 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 a8691317d3c..b6050738651 100644
--- a/sdks/python/apache_beam/yaml/standard_io.yaml
+++ b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -660,6 +660,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 00000000000..75b7b1a16dd
--- /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}