Outputs: {@code snapshots} (one row per commit), {@code errors} (one row per file that could - * not be registered: {@code file}, {@code error}). + * not be registered: {@code file}, {@code error}), and {@code dry_run_report} when a dry run is + * configured. * *
Schema evolution. With a {@link SchemaEvolutionConfig} whose options are set, a * pre-pass reads every Parquet footer, classifies the change each distinct file schema needs on the @@ -135,7 +137,9 @@ * {@code errors} (see {@link SchemaEvolutionConfig.IncompatibleSchemaHandling}). The per-file * checks read Parquet footers: an ORC or Avro file, or a pinned column the footer has no null count * for, cannot be verified and goes to {@code errors} unless {@link - * SchemaEvolutionConfig.UnverifiableFileHandling#ACCEPT} registers it on trust. Schema evolution + * SchemaEvolutionConfig.UnverifiableFileHandling#ACCEPT} registers it on trust. When the table does + * not exist, the pre-pass creates it from the union of the file schemas; if no readable Parquet + * schema can seed it, nothing is created and every file goes to {@code errors}. Schema evolution * currently requires bounded input; unbounded input with options set is rejected at construction. * *
{@code
@@ -154,6 +158,10 @@
public class AddFiles extends PTransform, PCollectionRowTuple> {
static final String OUTPUT_TAG = "snapshots";
static final String ERROR_TAG = "errors";
+
+ /** Only present with {@link SchemaEvolutionConfig#getDryRun()}. */
+ static final String DRY_RUN_TAG = "dry_run_report";
+
private static final Duration DEFAULT_TRIGGER_INTERVAL = Duration.standardMinutes(10);
private static final Counter numManifestFilesAdded =
counter(AddFiles.class, "numManifestFilesAdded");
@@ -260,7 +268,20 @@ public PCollectionRowTuple expand(PCollection input) {
PCollection paths = input;
if (evolution.isEnabled()) {
- paths = gateOnSchemaCommit(input);
+ // one commit per window, and for bounded input the window is the whole input
+ PCollection windowed =
+ input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
+ PCollection> schemas = distinctSchemas(windowed);
+ CommitSchemaUnion.Settings settings =
+ new CommitSchemaUnion.Settings(
+ evolution,
+ evolution.incompatibleSchemaHandlingFor(input.isBounded()),
+ new CommitSchemaUnion.NewTableSettings(partitionFields, sortFields, tableProps));
+ if (evolution.getDryRun()) {
+ return report(schemas, settings);
+ }
+ PCollection committed = commitSchema(schemas, settings);
+ paths = windowed.apply("WaitForSchemaCommit", Wait.on(committed));
}
PCollectionTuple dataFiles =
@@ -320,39 +341,49 @@ public PCollectionRowTuple expand(PCollection input) {
OUTPUT_TAG, snapshots, ERROR_TAG, dataFiles.get(ERRORS).setRowSchema(ERROR_SCHEMA));
}
- /**
- * Holds every path until the schema commit has landed. The pre-pass window is the unit of commit,
- * and for bounded input that unit is the whole input: paths are rewindowed into the global window
- * first, so one commit covers everything and the Wait.on signal lines up with the main input
- * whatever windowing the caller applied upstream.
- */
- private PCollection gateOnSchemaCommit(PCollection input) {
- PCollection windowed =
- input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
- CommitSchemaUnion.TableCreation creation =
- new CommitSchemaUnion.TableCreation(partitionFields, sortFields, tableProps);
- // unset handling follows the mode (fail the job in batch, route in streaming); the pre-pass
- // only runs on bounded input, see expand
- IncompatibleSchemaHandling onIncompatible =
- evolution.incompatibleSchemaHandlingFor(input.isBounded());
- PCollection signal =
- windowed
- .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema()))
- .setCoder(CollectDistinctSchemas.groupCoder())
- .apply(
- "CollectDistinctSchemas",
- Combine.globally(new CollectDistinctSchemas()).withoutDefaults())
+ private PCollection> distinctSchemas(
+ PCollection windowed) {
+ return windowed
+ .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema(evolution)))
+ .setCoder(CollectDistinctSchemas.groupCoder())
+ .apply(
+ "CollectDistinctSchemas",
+ Combine.globally(new CollectDistinctSchemas()).withoutDefaults());
+ }
+
+ /** Commits the plan for the window's schemas once; the signal releases the gated paths. */
+ private PCollection commitSchema(
+ PCollection> schemas,
+ CommitSchemaUnion.Settings settings) {
+ return schemas.apply(
+ "CommitSchemaOnce",
+ ParDo.of(new CommitSchemaOnce(catalogConfig, tableIdentifier, settings, committer)));
+ }
+
+ /** Reports the plan for the window's schemas instead; nothing is committed or registered. */
+ private PCollectionRowTuple report(
+ PCollection> schemas,
+ CommitSchemaUnion.Settings settings) {
+ PCollection report =
+ schemas
.apply(
- "CommitSchemaOnce",
- ParDo.of(
- new CommitSchemaOnce(
- catalogConfig,
- tableIdentifier,
- evolution,
- onIncompatible,
- creation,
- committer)));
- return windowed.apply("WaitForSchemaCommit", Wait.on(signal));
+ "DryRunReport",
+ ParDo.of(new DryRunReport(catalogConfig, tableIdentifier, settings)))
+ .setRowSchema(DryRunReport.REPORT_SCHEMA);
+
+ PCollection emptySnapshots =
+ schemas
+ .getPipeline()
+ .apply("NoSnapshots", Create.empty(RowCoder.of(SnapshotInfo.getSchema())))
+ .setRowSchema(SnapshotInfo.getSchema());
+ PCollection emptyErrors =
+ schemas
+ .getPipeline()
+ .apply("NoErrors", Create.empty(RowCoder.of(ERROR_SCHEMA)))
+ .setRowSchema(ERROR_SCHEMA);
+ return PCollectionRowTuple.of(OUTPUT_TAG, emptySnapshots)
+ .and(ERROR_TAG, emptyErrors)
+ .and(DRY_RUN_TAG, report);
}
/** Test hook: how the schema pre-pass commits its transaction. */
@@ -394,6 +425,7 @@ static class ConvertToDataFile extends DoFn {
private final SchemaEvolutionConfig evolution;
private transient @MonotonicNonNull BoundedAsyncTasks tasks;
private transient volatile @MonotonicNonNull Table table;
+ private transient volatile boolean tableMissing;
private transient @MonotonicNonNull Set warned;
private final AtomicBoolean refreshedThisBundle = new AtomicBoolean();
@@ -441,6 +473,9 @@ public ConvertToDataFile(
static final String UNREADABLE_SCHEMA_ERROR = "Could not read the file's schema: ";
static final String UNCOVERED_ERROR = "Table schema does not cover the file after refresh: ";
static final String PINNED_COLUMN_ERROR = "Pinned required column ";
+ static final String MISSING_TABLE_ERROR =
+ "Table does not exist and the schema pre-pass could not create it (no readable Parquet"
+ + " schema seeded it, or every file schema was refused): ";
static final String UNCHECKED_FORMAT_ERROR =
"Schema evolution is enabled but coverage and pin checks support only Parquet;"
+ " refusing to register an unchecked file of format ";
@@ -598,6 +633,10 @@ private Callable createProcessTask(
// ---- Infrastructure phase. Failures propagate so the runner retries the bundle;
// per-file error rows here would silently drop in-flight files on a transient blip.
// Only conditions that are properties of the file go to the error output.
+ if (tableMissing) {
+ return errorResult(
+ filePath, MISSING_TABLE_ERROR + identifier, timestamp, window, paneInfo);
+ }
if (table == null) {
synchronized (this) {
if (table == null) {
@@ -605,6 +644,10 @@ private Callable createProcessTask(
table = getOrCreateTable(filePath, format);
} catch (FileNotFoundException e) {
return errorResult(filePath, errorMessage(e), timestamp, window, paneInfo);
+ } catch (NoSuchTableException e) {
+ tableMissing = true;
+ return errorResult(
+ filePath, MISSING_TABLE_ERROR + identifier, timestamp, window, paneInfo);
}
}
}
@@ -835,6 +878,10 @@ private Table getOrCreateTable(String filePath, FileFormat format) throws IOExce
try {
return catalogConfig.catalog().loadTable(tableId);
} catch (NoSuchTableException e) {
+ if (evolution.isEnabled()) {
+ // the pre-pass is the only creator then, and it has already declined
+ throw e;
+ }
try {
org.apache.iceberg.Schema schema = getSchema(filePath, format);
PartitionSpec spec = PartitionUtils.toPartitionSpec(partitionFields, schema);
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java
index fa016ddca862..c5cc758c555b 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProvider.java
@@ -120,7 +120,8 @@ public static Builder builder() {
public abstract @Nullable List getSortFields();
@SchemaFieldDescription(
- "Lets the transform change the table schema so that every file's columns are covered."
+ "Lets the transform change the table schema so that the table has a column for every"
+ + " column the files have."
+ " Values: ALLOW_FIELD_ADDITION (columns a file has and the table lacks are added, as"
+ " optional), ALLOW_FIELD_RELAXATION (a required table column becomes optional when a"
+ " file lacks it or may hold nulls in it), ALLOW_TYPE_PROMOTION (a column type is"
@@ -131,8 +132,10 @@ public static Builder builder() {
+ " that is not allowed is incompatible; see incompatible_schema_handling. Only"
+ " Parquet files can be checked: ORC and Avro files are sent to the error output"
+ " unless unverifiable_file_handling is ACCEPT. Files sent to the error output are"
- + " dropped unless error_handling is set. Batch pipelines only; streaming pipelines"
- + " cannot use schema evolution yet.")
+ + " dropped unless error_handling is set. If the table does not exist it is created"
+ + " from the union of the Parquet schemas; with none, every file goes to the error"
+ + " output. Batch pipelines only; streaming pipelines cannot use schema evolution"
+ + " yet.")
public abstract @Nullable List getSchemaEvolutionOptions();
@SchemaFieldDescription(
@@ -145,6 +148,19 @@ public static Builder builder() {
+ " Requires schema_evolution_options.")
public abstract @Nullable List getRequiredColumns();
+ @SchemaFieldDescription(
+ "When true, nothing is committed or registered: the transform reads the files' schemas"
+ + " and emits a `dry_run_report` output with one row that describes what a real run"
+ + " would do. Its `allowed` field is true when every file schema can be merged and the"
+ + " configuration raises no problem; otherwise its `reason` field says what a real run"
+ + " would do about it (fail, or route the files to the error output). Its `schemas`"
+ + " field lists each distinct file schema with the changes a real run would make for"
+ + " it and, when it cannot be merged, why. The output only exists when this is set;"
+ + " consume it as input: `.dry_run_report`. Against a missing"
+ + " table, a REST catalog needs table-create permission even though no table is"
+ + " created.")
+ public abstract @Nullable Boolean getDryRun();
+
@SchemaFieldDescription(
"What happens when a file's schema cannot be made to fit the table: it needs a change"
+ " that is not allowed, or it conflicts with the table or with another file."
@@ -158,8 +174,9 @@ public static Builder builder() {
+ " Parquet footers only), or a Parquet file with no null-count statistics for a"
+ " required column (statistics disabled by the writer, or a column under a list or"
+ " map). REJECT (the default) sends the file to the error output (see"
- + " error_handling). ACCEPT registers it without checks, counted and logged. A file"
- + " that fails a check is always sent to the error output. An accepted file that"
+ + " error_handling). ACCEPT registers it without the checks; such files are counted"
+ + " and logged. A file that fails a check is always sent to the error output. An"
+ + " accepted file that"
+ " lacks a required column, or holds nulls in it, makes reads of the table fail.")
public abstract @Nullable String getUnverifiableFileHandling();
@@ -198,6 +215,8 @@ public abstract static class Builder {
public abstract Builder setUnverifiableFileHandling(String handling);
+ public abstract Builder setDryRun(Boolean dryRun);
+
public abstract Configuration build();
}
@@ -207,19 +226,21 @@ public abstract static class Builder {
List pins = getRequiredColumns();
String handlingName = getIncompatibleSchemaHandling();
String unverifiableName = getUnverifiableFileHandling();
+ boolean dryRun = Boolean.TRUE.equals(getDryRun());
boolean nothingSet =
(optionNames == null || optionNames.isEmpty())
&& (pins == null || pins.isEmpty())
&& handlingName == null
- && unverifiableName == null;
+ && unverifiableName == null
+ && !dryRun;
if (nothingSet) {
return null;
}
// SchemaEvolutionConfig.build() checks this too; this copy names the YAML keys
Preconditions.checkArgument(
optionNames != null && !optionNames.isEmpty(),
- "required_columns, incompatible_schema_handling and unverifiable_file_handling need at"
- + " least one schema_evolution_options entry");
+ "required_columns, incompatible_schema_handling, unverifiable_file_handling and"
+ + " dry_run need at least one schema_evolution_options entry");
Set options = EnumSet.noneOf(SchemaEvolutionOption.class);
for (String name : checkStateNotNull(optionNames)) {
options.add(parseEnum(SchemaEvolutionOption.class, name, "schema_evolution_options"));
@@ -228,6 +249,7 @@ public abstract static class Builder {
if (pins != null) {
builder = builder.setRequiredColumns(new LinkedHashSet<>(pins));
}
+ builder = builder.setDryRun(dryRun);
if (handlingName != null) {
SchemaEvolutionConfig.IncompatibleSchemaHandling handling =
parseEnum(
@@ -316,6 +338,9 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
if (errorHandling != null) {
output = output.and(errorHandling.getOutput(), result.get(ERROR_TAG));
}
+ if (Boolean.TRUE.equals(configuration.getDryRun())) {
+ output = output.and(AddFiles.DRY_RUN_TAG, result.get(AddFiles.DRY_RUN_TAG));
+ }
return output;
}
}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java
index 6fd5171c2eae..768fb564de9b 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java
@@ -20,7 +20,6 @@
import static org.apache.beam.sdk.metrics.Metrics.counter;
import java.util.List;
-import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.iceberg.catalog.Catalog;
@@ -37,34 +36,23 @@ class CommitSchemaOnce extends DoFn, Lo
private final IcebergCatalogConfig catalogConfig;
private final String identifier;
- private final SchemaEvolutionConfig config;
- private final IncompatibleSchemaHandling handling;
- private final CommitSchemaUnion.TableCreation creation;
+ private final CommitSchemaUnion.Settings settings;
private final CommitSchemaUnion.Committer committer;
private transient @MonotonicNonNull Catalog catalog;
CommitSchemaOnce(
- IcebergCatalogConfig catalogConfig,
- String identifier,
- SchemaEvolutionConfig config,
- IncompatibleSchemaHandling handling,
- CommitSchemaUnion.TableCreation creation) {
- this(
- catalogConfig, identifier, config, handling, creation, CommitSchemaUnion.DEFAULT_COMMITTER);
+ IcebergCatalogConfig catalogConfig, String identifier, CommitSchemaUnion.Settings settings) {
+ this(catalogConfig, identifier, settings, CommitSchemaUnion.DEFAULT_COMMITTER);
}
CommitSchemaOnce(
IcebergCatalogConfig catalogConfig,
String identifier,
- SchemaEvolutionConfig config,
- IncompatibleSchemaHandling handling,
- CommitSchemaUnion.TableCreation creation,
+ CommitSchemaUnion.Settings settings,
CommitSchemaUnion.Committer committer) {
this.catalogConfig = catalogConfig;
this.identifier = identifier;
- this.config = config;
- this.handling = handling;
- this.creation = creation;
+ this.settings = settings;
this.committer = committer;
}
@@ -82,8 +70,7 @@ public void process(
committer.commit(txn);
numSchemaCommits.inc();
};
- long schemaId =
- CommitSchemaUnion.commit(catalog, tableId, schemas, config, handling, creation, counting);
+ long schemaId = CommitSchemaUnion.commit(catalog, tableId, schemas, settings, counting);
out.output(schemaId);
}
}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java
index 1e145a122c81..093749798780 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java
@@ -17,6 +17,7 @@
*/
package org.apache.beam.sdk.io.iceberg;
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
import java.io.Serializable;
@@ -25,45 +26,39 @@
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Metrics;
import org.apache.beam.sdk.util.BackOff;
import org.apache.beam.sdk.util.BackOffUtils;
import org.apache.beam.sdk.util.FluentBackoff;
import org.apache.beam.sdk.util.Sleeper;
import org.apache.iceberg.Schema;
-import org.apache.iceberg.SchemaParser;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.Transaction;
-import org.apache.iceberg.UpdateSchema;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.exceptions.NoSuchTableException;
-import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.mapping.NameMapping;
-import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
-import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.PropertyUtil;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.Duration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
- * Applies the distinct file schemas of a window to the table in one commit, in phases named by the
- * methods of this class: {@code classify} each schema against a fresh load of the table (most
- * common first), {@code fold} the accepted ones into a single union on scratch transactions that
- * are never committed, {@code replay} the folded result onto the real transaction as one schema
- * update, repair the name mapping, commit once. The fold keeps per-schema blame for cross-schema
- * conflicts while the table gains a single schema version per window. Nothing is committed when
- * nothing changes.
+ * Applies the distinct file schemas of a window to the table in one commit: computes the {@link
+ * SchemaPlan} against a fresh load of the table, {@code replay}s the planned union onto the real
+ * transaction as one schema update, repairs the name mapping, commits once. The table gains a
+ * single schema version per window; nothing is committed when nothing changes. A dry run reports
+ * the same plan instead of committing it.
*
- * When the table does not exist, {@code create} builds it instead: {@code foldForCreate}
- * computes the same union, and the table is born from it directly with pinned columns and their
- * ancestors required.
+ *
When the table does not exist, {@code create} builds it from the planned union instead, with
+ * pinned columns and their ancestors required.
*
*
Incompatible schemas either fail the whole call before any commit ({@link
* IncompatibleSchemaHandling#FAIL_PIPELINE}) or are skipped so their files reach the error output
@@ -72,18 +67,31 @@
final class CommitSchemaUnion {
private static final Logger LOG = LoggerFactory.getLogger(CommitSchemaUnion.class);
- static final int MAX_ATTEMPTS = 5;
+ /**
+ * How long a commit or a plan keeps retrying, unless the table's own {@code commit.retry.*}
+ * properties say otherwise. More patient than Iceberg's defaults because Iceberg cannot retry
+ * these commits itself: a schema update inside a transaction cannot be re-applied after a
+ * concurrent commit of any kind, a data append included, so every such commit lands here.
+ */
+ static final Duration DEFAULT_RETRY_MIN_WAIT = Duration.millis(250);
+
+ static final Duration DEFAULT_RETRY_MAX_WAIT = Duration.standardSeconds(10);
+ static final Duration DEFAULT_RETRY_TOTAL_TIMEOUT = Duration.standardMinutes(1);
+
+ /** Tells the runner a worker waiting out a busy catalog is throttled, not busy. */
+ private static final Counter throttledMillis =
+ Metrics.counter(Metrics.THROTTLE_TIME_NAMESPACE, Metrics.THROTTLE_TIME_COUNTER_NAME);
/** Returned when the table does not exist and there is no schema to create it from. */
static final long NO_TABLE = -1L;
- /** How to create the table when it does not exist: from the union of the window's schemas. */
- static final class TableCreation implements Serializable {
+ /** What a table is created with when it does not exist yet; unused on an existing table. */
+ static final class NewTableSettings implements Serializable {
final @Nullable List partitionFields;
final @Nullable List sortFields;
final @Nullable Map properties;
- TableCreation(
+ NewTableSettings(
@Nullable List partitionFields,
@Nullable List sortFields,
@Nullable Map properties) {
@@ -93,6 +101,25 @@ static final class TableCreation implements Serializable {
}
}
+ /** How a schema commit behaves; the same for the commit and for a dry run of it. */
+ static final class Settings implements Serializable {
+ final SchemaEvolutionConfig config;
+
+ /** Resolved for the pipeline's mode; the config's own value may be unset. */
+ final IncompatibleSchemaHandling handling;
+
+ final NewTableSettings newTable;
+
+ Settings(
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ NewTableSettings newTable) {
+ this.config = config;
+ this.handling = handling;
+ this.newTable = newTable;
+ }
+ }
+
/** Injectable so tests can exercise the commit retry path. */
interface Committer extends Serializable {
void commit(Transaction txn);
@@ -107,52 +134,6 @@ static final class IncompatibleSchemaException extends IllegalStateException {
}
}
- private static final class Accepted {
- final Schema schema;
- final String json;
- final long files;
-
- /** Null on the create path: the seed table is empty, so there is nothing to relax. */
- final @Nullable SchemaDelta delta;
-
- Accepted(Schema schema, String json, long files, @Nullable SchemaDelta delta) {
- this.schema = schema;
- this.json = json;
- this.files = files;
- this.delta = delta;
- }
- }
-
- private static final class Incompatible {
- final String schemaJson;
- final long files;
- final String reason;
-
- Incompatible(String schemaJson, long files, String reason) {
- this.schemaJson = schemaJson;
- this.files = files;
- this.reason = reason;
- }
-
- @Override
- public String toString() {
- return files + " file(s) with schema " + truncate(schemaJson) + ": " + reason;
- }
- }
-
- /** Canonical JSON of a wide schema runs to hundreds of KB; the reason is what matters. */
- private static final int MAX_SCHEMA_JSON_CHARS = 1024;
-
- private static String truncate(String json) {
- if (json.length() <= MAX_SCHEMA_JSON_CHARS) {
- return json;
- }
- return json.substring(0, MAX_SCHEMA_JSON_CHARS)
- + "... ("
- + (json.length() - MAX_SCHEMA_JSON_CHARS)
- + " chars truncated)";
- }
-
private CommitSchemaUnion() {}
/**
@@ -165,157 +146,169 @@ static long commit(
Catalog catalog,
TableIdentifier tableId,
List schemas,
- SchemaEvolutionConfig config,
- IncompatibleSchemaHandling handling,
- TableCreation creation,
+ Settings settings,
Committer committer) {
- // The catalog is already under contention when a retry fires; back off (jittered by
- // FluentBackoff) instead of piling on. Iceberg's own metadata retries (commit.retry.*)
- // sit below this loop.
- BackOff backoff =
- FluentBackoff.DEFAULT
- .withMaxRetries(MAX_ATTEMPTS - 1)
- .withInitialBackoff(Duration.millis(100))
- .withMaxBackoff(Duration.standardSeconds(2))
- .backoff();
+ return commit(catalog, tableId, schemas, settings, committer, Sleeper.DEFAULT);
+ }
+
+ /** Test hook: the sleeper keeps the retry tests from waiting out the backoff. */
+ static long commit(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List schemas,
+ Settings settings,
+ Committer committer,
+ Sleeper sleeper) {
+ return withRetry(
+ catalog,
+ tableId,
+ sleeper,
+ table -> commitOnce(catalog, tableId, table, schemas, settings, committer));
+ }
+
+ /** One attempt at a plan or a commit, against the table as it is now; null when missing. */
+ private interface Attempt {
+ T run(@Nullable Table table);
+ }
+
+ /**
+ * Runs an attempt again, against a fresh load of the table, when a concurrent commit or a create
+ * race invalidates the state it started from. Gives up when the backoff time is spent.
+ */
+ private static T withRetry(
+ Catalog catalog, TableIdentifier tableId, Sleeper sleeper, Attempt once) {
+ @Nullable BackOff backoff = null;
for (int attempt = 1; ; attempt++) {
+ @Nullable Table table;
+ try {
+ table = catalog.loadTable(tableId);
+ } catch (NoSuchTableException e) {
+ table = null;
+ }
try {
- return commitOnce(catalog, tableId, schemas, config, handling, creation, committer);
+ return once.run(table);
} catch (CommitFailedException | AlreadyExistsException e) {
- // a concurrent commit, or a create race: the next attempt loads the fresh state
+ if (backoff == null) {
+ backoff = retryBackoff(table).backoff();
+ }
+ LOG.info(
+ "Schema pre-pass attempt {} for {} failed: {}",
+ attempt,
+ tableId,
+ AddFiles.errorMessage(e));
try {
- if (!BackOffUtils.next(Sleeper.DEFAULT, backoff)) {
+ if (!BackOffUtils.next(sleeper, backoff)) {
throw e;
}
} catch (InterruptedException interrupted) {
Thread.currentThread().interrupt();
throw e;
}
- LOG.info(
- "Schema commit attempt {}/{} for {} failed; reloading and rebuilding",
- attempt,
- MAX_ATTEMPTS,
- tableId,
- e);
}
}
}
+ /**
+ * Jittered exponential backoff, bounded by time; a retry count only applies when the table sets
+ * one. Values the backoff would reject are clamped, so a bad table property cannot replace the
+ * commit failure with a configuration error.
+ */
+ private static FluentBackoff retryBackoff(@Nullable Table table) {
+ Map properties = table == null ? Collections.emptyMap() : table.properties();
+ long minWaitMillis =
+ PropertyUtil.propertyAsLong(
+ properties,
+ TableProperties.COMMIT_MIN_RETRY_WAIT_MS,
+ DEFAULT_RETRY_MIN_WAIT.getMillis());
+ long maxWaitMillis =
+ PropertyUtil.propertyAsLong(
+ properties,
+ TableProperties.COMMIT_MAX_RETRY_WAIT_MS,
+ DEFAULT_RETRY_MAX_WAIT.getMillis());
+ long totalMillis =
+ PropertyUtil.propertyAsLong(
+ properties,
+ TableProperties.COMMIT_TOTAL_RETRY_TIME_MS,
+ DEFAULT_RETRY_TOTAL_TIMEOUT.getMillis());
+ int maxRetries =
+ PropertyUtil.propertyAsInt(
+ properties, TableProperties.COMMIT_NUM_RETRIES, Integer.MAX_VALUE);
+ return FluentBackoff.DEFAULT
+ .withExponent(2.0)
+ .withInitialBackoff(Duration.millis(Math.max(1, minWaitMillis)))
+ .withMaxBackoff(Duration.millis(Math.max(1, maxWaitMillis)))
+ .withMaxCumulativeBackoff(Duration.millis(Math.max(1, totalMillis)))
+ .withMaxRetries(Math.max(0, maxRetries))
+ .withThrottledTimeCounter(throttledMillis);
+ }
+
+ /** The plan for the window's schemas, reloading the table when a concurrent change moves it. */
+ static SchemaPlan plan(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List schemas,
+ Settings settings) {
+ return withRetry(
+ catalog,
+ tableId,
+ Sleeper.DEFAULT,
+ table -> SchemaPlan.compute(catalog, tableId, table, schemas, settings));
+ }
+
private static long commitOnce(
Catalog catalog,
TableIdentifier tableId,
+ @Nullable Table table,
List schemas,
- SchemaEvolutionConfig config,
- IncompatibleSchemaHandling handling,
- TableCreation creation,
+ Settings settings,
Committer committer) {
- Table table;
- try {
- table = catalog.loadTable(tableId);
- } catch (NoSuchTableException e) {
- return create(catalog, tableId, schemas, config, handling, creation, committer);
+ SchemaPlan plan = SchemaPlan.compute(catalog, tableId, table, schemas, settings);
+ if (plan instanceof SchemaPlan.Creation) {
+ return create((SchemaPlan.Creation) plan, catalog, tableId, settings, committer);
}
- // Every transaction below must share this snapshot: classification, the fold and the replay
- // all reason about the same table state (newTransactionOn enforces it).
- Schema base = table.schema();
+ return evolve((SchemaPlan.Evolution) plan, tableId, settings.handling, committer);
+ }
- List incompatible = new ArrayList<>();
- List accepted = classify(table, schemas, config, incompatible);
- Schema merged = fold(table, base, tableId, accepted, incompatible);
+ /** Replays the planned union onto the table and repairs the name mapping, in one commit. */
+ private static long evolve(
+ SchemaPlan.Evolution plan,
+ TableIdentifier tableId,
+ IncompatibleSchemaHandling handling,
+ Committer committer) {
+ failOrWarnOnIncompatibleSchemas(
+ tableId, plan.incompatibleSchemas, handling, "no schema change was committed");
+ failOrWarnOnConfigProblems(tableId, plan.configProblems, handling);
- Transaction txn = newTransactionOn(table, base, tableId);
- if (merged != null) {
- replay(txn, merged, tableId);
+ Transaction txn = plan.newTransaction(tableId);
+ if (plan.newSchema != null) {
+ replay(txn, plan.newSchema, tableId);
}
- boolean staged = merged != null;
- staged |= stageNameMapping(txn);
-
- if (!incompatible.isEmpty()) {
- reportIncompatible(tableId, incompatible, handling, "no schema change was committed");
+ if (plan.repairsNameMapping) {
+ stageNameMapping(txn);
}
-
- if (!staged) {
- LOG.info(
- "Table {} already covers all {} file schema(s); nothing to commit",
- tableId,
- schemas.size());
- return table.schema().schemaId();
+ if (plan.newSchema == null && !plan.repairsNameMapping) {
+ LOG.info("Table {} already covers every file schema; nothing to commit", tableId);
+ return plan.base.schemaId();
}
committer.commit(txn);
long schemaId = txn.table().schema().schemaId();
- long acceptedFiles = 0;
- for (Accepted item : accepted) {
- acceptedFiles += item.files;
+ long mergedFiles = 0;
+ for (SchemaPlan.SchemaToMerge item : plan.schemasToMerge) {
+ mergedFiles += item.files;
}
LOG.info(
"Committed schema union for {}: {} schema(s) covering {} file(s), now at schema id {}",
tableId,
- accepted.size(),
- acceptedFiles,
+ plan.schemasToMerge.size(),
+ mergedFiles,
schemaId);
return schemaId;
}
- /**
- * Sorts the window's schemas into the ones the table must change for (accepted) and the ones it
- * must not ({@code incompatible}, with the reason); schemas the table already covers drop out.
- */
- private static List classify(
- Table table,
- List schemas,
- SchemaEvolutionConfig config,
- List incompatible) {
- List accepted = new ArrayList<>();
- for (CollectDistinctSchemas.SchemaGroup group : schemas) {
- Schema fileSchema =
- FileSchemas.markRequired(
- SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns());
- SchemaDelta delta = SchemaDelta.classify(table, fileSchema);
- if (delta.isEmpty()) {
- continue;
- }
- if (!delta.allowedBy(config)) {
- incompatible.add(
- new Incompatible(
- group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config)));
- continue;
- }
- accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta));
- }
- return accepted;
- }
-
- /**
- * Unions the accepted schemas into the table schema on scratch transactions that are never
- * committed, relaxing every field the window adds; a schema that conflicts with another only
- * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the
- * folded schema, or null when nothing needs to change.
- */
- private static @Nullable Schema fold(
- Table table,
- Schema base,
- TableIdentifier tableId,
- List accepted,
- List incompatible) {
- while (true) {
- Transaction scratch = newTransactionOn(table, base, tableId);
- Accepted failed = stageAll(scratch, accepted, incompatible);
- if (failed != null) {
- accepted.remove(failed);
- continue;
- }
- if (accepted.isEmpty()) {
- return null;
- }
- relaxNewRequiredFields(scratch, base);
- return scratch.table().schema();
- }
- }
-
/**
* One union replays the fold's net effect (additions, promotions, relaxations) so the table gains
* a single schema version instead of one per folded schema. The checkState is a pure bug
- * detector: concurrent changes are caught earlier, by newTransactionOn.
+ * detector: concurrent changes are caught earlier, when the plan opens its transaction.
*/
private static void replay(Transaction txn, Schema merged, TableIdentifier tableId) {
txn.updateSchema().unionByNameWith(merged).commit();
@@ -338,192 +331,59 @@ private static void replay(Transaction txn, Schema merged, TableIdentifier table
* append after them.
*/
private static long create(
+ SchemaPlan.Creation plan,
Catalog catalog,
TableIdentifier tableId,
- List schemas,
- SchemaEvolutionConfig config,
- IncompatibleSchemaHandling handling,
- TableCreation creation,
+ Settings settings,
Committer committer) {
- if (schemas.isEmpty()) {
+ if (plan.schemasToMerge.isEmpty() && plan.incompatibleSchemas.isEmpty()) {
LOG.info("Table {} does not exist and no file schema was read; not creating it", tableId);
return NO_TABLE;
}
- List incompatible = new ArrayList<>();
- @Nullable Schema merged = foldForCreate(catalog, tableId, schemas, incompatible);
- if (!incompatible.isEmpty()) {
- reportIncompatible(tableId, incompatible, handling, "no table was created");
- }
- if (merged == null) {
+ failOrWarnOnIncompatibleSchemas(
+ tableId, plan.incompatibleSchemas, settings.handling, "no table was created");
+ if (plan.newSchema == null) {
LOG.info("Table {} does not exist and no file schema can seed it; not creating it", tableId);
return NO_TABLE;
}
- // The real table is built from the folded result directly.
- Schema created = createdSchema(merged, config);
- reportUnenforceablePins(tableId, created, config, handling);
- Map properties =
- creation.properties == null ? new HashMap<>() : new HashMap<>(creation.properties);
+ if (plan.problem != null) {
+ throw new IllegalStateException(plan.problem);
+ }
+ failOrWarnOnConfigProblems(tableId, plan.configProblems, settings.handling);
+ Map properties = new HashMap<>();
+ if (settings.newTable.properties != null) {
+ properties.putAll(settings.newTable.properties);
+ }
Transaction txn =
catalog
- .buildTable(tableId, created)
- .withPartitionSpec(PartitionUtils.toPartitionSpec(creation.partitionFields, created))
- .withSortOrder(SortOrderUtils.toSortOrder(creation.sortFields, created))
+ .buildTable(tableId, plan.newSchema)
+ .withPartitionSpec(checkStateNotNull(plan.spec))
+ .withSortOrder(checkStateNotNull(plan.sortOrder))
.withProperties(properties)
.createTransaction();
- stageNameMapping(txn);
+ if (SchemaPlan.needsNameMapping(txn.table().properties(), txn.table().schema())) {
+ stageNameMapping(txn);
+ }
committer.commit(txn);
long schemaId = txn.table().schema().schemaId();
LOG.info(
"Created table {} from {} file schema(s), schema id {}",
tableId,
- schemas.size() - incompatible.size(),
+ plan.schemasToMerge.size(),
schemaId);
return schemaId;
}
- /**
- * Unions the window's schemas into one on scratch create transactions that are never committed,
- * seeded by the most common schema; conflicts move to {@code incompatible} and the fold restarts
- * without the offender.
- */
- private static @Nullable Schema foldForCreate(
- Catalog catalog,
- TableIdentifier tableId,
- List schemas,
- List incompatible) {
- // The evolve path refuses names no table can absorb (dotted, empty, differing only in case
- // within one file) as conflicts in classify; a table must not be born with them either.
- List valid = new ArrayList<>();
- for (CollectDistinctSchemas.SchemaGroup group : schemas) {
- Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson());
- List invalidNames = new ArrayList<>();
- ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "", invalidNames);
- if (!invalidNames.isEmpty()) {
- incompatible.add(
- new Incompatible(
- group.getSchemaJson(),
- group.getFiles(),
- "file schema has column names no table can hold: " + describe(invalidNames)));
- continue;
- }
- valid.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), null));
- }
- if (valid.isEmpty()) {
- return null;
- }
- Schema seed = valid.get(0).schema;
- List rest = new ArrayList<>(valid.subList(1, valid.size()));
- while (true) {
- Transaction scratch = catalog.buildTable(tableId, seed).createTransaction();
- Accepted failed = stageAll(scratch, rest, incompatible);
- if (failed == null) {
- return scratch.table().schema();
- }
- rest.remove(failed);
- }
- }
-
- /**
- * A pin the created schema did not end up enforcing - the column appears in no file schema, or
- * the configured spelling resolves to a field the pin walk did not reach (a short container
- * spelling like a.b for a.element.b, or a path inside a map key) - would stay inert forever,
- * since later windows only add columns optional: a config error under FAIL_PIPELINE, a warning
- * under ROUTE_TO_ERRORS (streaming may see the column later).
- */
- private static void reportUnenforceablePins(
- TableIdentifier tableId,
- Schema created,
- SchemaEvolutionConfig config,
- IncompatibleSchemaHandling handling) {
- List unenforceable = new ArrayList<>();
- for (String pin : config.getRequiredColumns()) {
- Types.NestedField field = created.findField(pin);
- if (field == null || field.isOptional()) {
- unenforceable.add(pin);
- }
- }
- if (unenforceable.isEmpty()) {
- return;
- }
- Collections.sort(unenforceable);
- if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) {
- throw new IncompatibleSchemaException(
- "Pinned column(s) "
- + unenforceable
- + " appear in none of the file schemas creating "
- + tableId
- + ", or their spelling does not match the column path; the created table cannot"
- + " make them required");
- }
- LOG.warn(
- "Pinned column(s) {} appear in none of the file schemas creating {}, or their spelling"
- + " does not match the column path; the created table cannot make them required",
- unenforceable,
- tableId);
- }
-
- /**
- * The created schema: every field optional at every level, list elements and map values included,
- * except pinned paths and their ancestors, which stay required so the schema advertises the
- * guarantee the per-file pin check enforces (a null ancestor nulls the pinned leaf). Map key
- * subtrees keep their declared shape (keys are required by definition; pins inside them are not
- * honored). Nothing depends on a created table's schema yet, so this is the schema-authoring
- * moment; evolution never tightens columns afterwards. Column order is the union's, which is the
- * canonical (name-sorted) order of the file schemas.
- */
- static Schema createdSchema(Schema merged, SchemaEvolutionConfig config) {
- Pins pins = new Pins(config.getRequiredColumns());
- List fields = new ArrayList<>();
- for (Types.NestedField field : merged.asStruct().fields()) {
- fields.add(createdField(field, field.name(), pins));
- }
- return new Schema(fields);
- }
-
- private static Types.NestedField createdField(Types.NestedField field, String path, Pins pins) {
- boolean required = pins.isPinnedOrAncestorOfPin(path);
- return Types.NestedField.from(field)
- .ofType(createdType(field.type(), path, pins))
- .isOptional(!required)
- .build();
- }
-
- private static Type createdType(Type type, String path, Pins pins) {
- if (type.isStructType()) {
- List fields = new ArrayList<>();
- for (Types.NestedField field : type.asStructType().fields()) {
- fields.add(createdField(field, path + "." + field.name(), pins));
- }
- return Types.StructType.of(fields);
- }
- if (type.isListType()) {
- Types.ListType list = type.asListType();
- String elementPath = path + ".element";
- Type elementType = createdType(list.elementType(), elementPath, pins);
- boolean required = pins.isPinnedOrAncestorOfPin(elementPath);
- return required
- ? Types.ListType.ofRequired(list.elementId(), elementType)
- : Types.ListType.ofOptional(list.elementId(), elementType);
- }
- if (type.isMapType()) {
- Types.MapType map = type.asMapType();
- String valuePath = path + ".value";
- Type valueType = createdType(map.valueType(), valuePath, pins);
- boolean required = pins.isPinnedOrAncestorOfPin(valuePath);
- return required
- ? Types.MapType.ofRequired(map.keyId(), map.valueId(), map.keyType(), valueType)
- : Types.MapType.ofOptional(map.keyId(), map.valueId(), map.keyType(), valueType);
- }
- return type;
- }
-
- private static void reportIncompatible(
+ private static void failOrWarnOnIncompatibleSchemas(
TableIdentifier tableId,
- List incompatible,
+ List incompatible,
IncompatibleSchemaHandling handling,
String consequence) {
+ if (incompatible.isEmpty()) {
+ return;
+ }
long files = 0;
- for (Incompatible item : incompatible) {
+ for (SchemaPlan.IncompatibleSchema item : incompatible) {
files += item.files;
}
if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) {
@@ -548,144 +408,33 @@ private static void reportIncompatible(
joinLines(incompatible));
}
- /**
- * Stages one union per accepted schema onto {@code txn}: a scratch transaction on the evolve path
- * (its per-schema versions stay in memory; only the folded result is ever committed), the create
- * transaction on the create path. A schema can conflict with another schema's additions, which
- * only surfaces while staging and poisons the transaction, so on a conflict the offender is
- * returned for the caller to drop and retry with a fresh transaction.
- */
- private static @Nullable Accepted stageAll(
- Transaction txn, List accepted, List incompatible) {
- for (Accepted item : accepted) {
- // classify checked each schema against the base table only; a column that differs only in
- // case from one an EARLIER schema of the window added would union as a second column.
- List collisions = new ArrayList<>();
- ColumnNameChecks.findCaseCollisions(
- txn.table().schema().asStruct(), item.schema.asStruct(), "", collisions);
- if (!collisions.isEmpty()) {
- incompatible.add(
- new Incompatible(
- item.json,
- item.files,
- "conflicts with another file schema in the same window: " + describe(collisions)));
- return item;
- }
- // Both caught types carry staging conflicts: ValidationException from Schema
- // construction at apply ("multiple fields for name"), IllegalArgumentException from
- // SchemaUpdate preconditions ("Cannot change column type").
- try {
- stage(txn, item);
- } catch (ValidationException | IllegalArgumentException e) {
- incompatible.add(
- new Incompatible(
- item.json,
- item.files,
- "conflicts with another file schema in the same window: "
- + AddFiles.errorMessage(e)));
- return item;
- }
- }
- return null;
- }
-
- private static String describe(List changes) {
- List descriptions = new ArrayList<>();
- for (SchemaChange change : changes) {
- descriptions.add(change.description);
- }
- return String.join("; ", descriptions);
- }
-
- /**
- * Iceberg refreshes the table on every {@code newTransaction()}, so a concurrent schema commit
- * can slip between two transactions here. Any drift from the snapshot the window classified
- * against is thrown as {@link CommitFailedException} so the commit-level retry reloads and
- * rebuilds, leaving the replay checkState as a pure bug detector.
- */
- private static Transaction newTransactionOn(Table table, Schema base, TableIdentifier tableId) {
- Transaction txn = table.newTransaction();
- if (!txn.table().schema().sameSchema(base)) {
- throw new CommitFailedException(
- "concurrent schema change on %s while staging the schema union", tableId);
- }
- return txn;
- }
-
- private static void stage(Transaction txn, Accepted item) {
- UpdateSchema update = txn.updateSchema().unionByNameWith(item.schema);
- if (item.delta != null) {
- for (String path : item.delta.absentRequiredPaths()) {
- update = update.makeColumnOptional(path);
- }
- }
- update.commit();
- }
-
- /**
- * New columns are optional at every level. The union adds top-level columns optional but keeps
- * the file's optionality below them, so one file's luck would otherwise impose required fields on
- * everyone. Pins do not shape new columns: they keep existing required columns from being relaxed
- * (SchemaDelta) and gate files at registration.
- */
- private static void relaxNewRequiredFields(Transaction txn, Schema before) {
- List toRelax = newRequiredPaths(before, txn.table().schema());
- if (toRelax.isEmpty()) {
+ /** A configuration problem fails the run under FAIL_PIPELINE and warns under ROUTE_TO_ERRORS. */
+ private static void failOrWarnOnConfigProblems(
+ TableIdentifier tableId, List problems, IncompatibleSchemaHandling handling) {
+ if (problems.isEmpty()) {
return;
}
- UpdateSchema update = txn.updateSchema();
- for (String path : toRelax) {
- update = update.makeColumnOptional(path);
+ if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) {
+ throw new IncompatibleSchemaException(String.join("; ", problems));
}
- update.commit();
- }
-
- /**
- * Paths of required fields that {@code after} has and {@code before} lacks, in schema order;
- * includes fields under lists and maps (a required list element or map value counts). Map key
- * subtrees are skipped: keys are required by definition and relaxing inside a struct key would
- * change key identity.
- */
- static List newRequiredPaths(Schema before, Schema after) {
- Set beforeIds = TypeUtil.indexById(before.asStruct()).keySet();
- List paths = new ArrayList<>();
- collectNewRequired(after.asStruct(), "", beforeIds, paths);
- return paths;
- }
-
- private static void collectNewRequired(
- Type.NestedType type, String prefix, Set beforeIds, List paths) {
- for (Types.NestedField field : type.fields()) {
- if (type.isMapType() && field.fieldId() == type.asMapType().keyId()) {
- continue;
- }
- String path = prefix + field.name();
- if (!beforeIds.contains(field.fieldId()) && field.isRequired()) {
- paths.add(path);
- }
- if (field.type().isNestedType()) {
- collectNewRequired(field.type().asNestedType(), path + ".", beforeIds, paths);
- }
+ for (String problem : problems) {
+ LOG.warn("Configuration problem on {}: {}", tableId, problem);
}
}
- /** Regenerates the name mapping when absent, malformed or not covering the staged schema. */
- private static boolean stageNameMapping(Transaction txn) {
+ /** Regenerates the name mapping property for the transaction's schema. */
+ private static void stageNameMapping(Transaction txn) {
Schema schema = txn.table().schema();
@Nullable NameMapping existing =
NameMappingUtils.parseOrNull(
txn.table().properties().get(TableProperties.DEFAULT_NAME_MAPPING));
- if (existing != null && NameMappingUtils.covers(existing, schema.asStruct())) {
- return false;
- }
String regenerated = NameMappingUtils.regenerate(schema, existing);
txn.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING, regenerated).commit();
- return true;
}
- private static String joinLines(List items) {
+ private static String joinLines(List items) {
List lines = new ArrayList<>();
- for (Incompatible item : items) {
+ for (SchemaPlan.IncompatibleSchema item : items) {
lines.add(item.toString());
}
return String.join("\n ", lines);
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/DryRunReport.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/DryRunReport.java
new file mode 100644
index 000000000000..ae069598e0a6
--- /dev/null
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/DryRunReport.java
@@ -0,0 +1,434 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg;
+
+import static org.apache.beam.sdk.metrics.Metrics.counter;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.schemas.Schema;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.values.Row;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.hash.Hashing;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.types.Type;
+import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Dry run of the schema pre-pass: computes the {@link CommitSchemaUnion#plan} a real commit would,
+ * on scratch transactions only, and reports it without committing or registering anything. Planning
+ * the real fold rather than classifying each schema alone is what makes a schema that is fine
+ * against the table but conflicts with another schema of the input come out with the blame a real
+ * run would assign.
+ *
+ * One row per window. {@code allowed} is true when every file schema can be merged and the
+ * configuration raises no problem; otherwise {@code reason} says what a real run would do about it.
+ * {@code schemas} holds one entry per distinct file schema with the changes a real run would make
+ * on an existing table ({@code schema_key} is a short hash to group by) and, when the schema cannot
+ * be merged, why. {@code created_table} is the table a real run would create from the union of the
+ * allowed schemas, absent when the table exists or no schema can seed it; {@code
+ * would_create_table} is false whenever a real run would not create it, including when it would
+ * fail first. When the table does not exist and no file schema can create it, nothing is allowed: a
+ * real run sends every file to the error output. {@code table_changes} lists what a real run
+ * changes without a schema change, such as regenerating the name mapping.
+ *
+ *
The file counts and the counters count checked Parquet files by whether their schema is
+ * allowed. Unreadable files always go to the error output in a real run; unchecked (ORC, Avro)
+ * files count separately even when {@code unchecked_registered} says ACCEPT registers them. Pin
+ * evidence is per file (the footer's null counts), so a pin violation or an unproven pin is not
+ * predicted here.
+ */
+class DryRunReport extends DoFn, Row> {
+ private static final Logger LOG = LoggerFactory.getLogger(DryRunReport.class);
+
+ static final String FILES_ALLOWED_COUNTER = "numDryRunFilesAllowed";
+ static final String FILES_INCOMPATIBLE_COUNTER = "numDryRunFilesIncompatible";
+ static final String FILES_UNREADABLE_COUNTER = "numDryRunFilesUnreadable";
+ static final String FILES_UNCHECKED_COUNTER = "numDryRunFilesUnchecked";
+ static final String CONFIG_PROBLEMS_COUNTER = "numDryRunConfigProblems";
+
+ private static final Counter numFilesAllowed = counter(DryRunReport.class, FILES_ALLOWED_COUNTER);
+ private static final Counter numFilesIncompatible =
+ counter(DryRunReport.class, FILES_INCOMPATIBLE_COUNTER);
+ private static final Counter numFilesUnreadable =
+ counter(DryRunReport.class, FILES_UNREADABLE_COUNTER);
+ private static final Counter numFilesUnchecked =
+ counter(DryRunReport.class, FILES_UNCHECKED_COUNTER);
+ private static final Counter numConfigProblems =
+ counter(DryRunReport.class, CONFIG_PROBLEMS_COUNTER);
+
+ static final Schema SCHEMA_ENTRY =
+ Schema.builder()
+ .addStringField("schema_key")
+ .addStringField("schema")
+ .addInt64Field("num_files")
+ .addArrayField("changes", Schema.FieldType.STRING)
+ .addBooleanField("allowed")
+ .addStringField("reason")
+ .build();
+
+ static final Schema CREATED_TABLE =
+ Schema.builder()
+ .addStringField("schema")
+ .addArrayField("columns", Schema.FieldType.STRING)
+ .build();
+
+ static final Schema REPORT_SCHEMA =
+ Schema.builder()
+ .addBooleanField("allowed")
+ .addStringField("reason")
+ .addBooleanField("would_create_table")
+ .addInt64Field("files_allowed")
+ .addInt64Field("files_incompatible")
+ .addInt64Field("files_unreadable")
+ .addInt64Field("files_unchecked")
+ .addBooleanField("unchecked_registered")
+ .addArrayField("table_changes", Schema.FieldType.STRING)
+ .addArrayField("config_problems", Schema.FieldType.STRING)
+ .addNullableRowField("created_table", CREATED_TABLE)
+ .addArrayField("schemas", Schema.FieldType.row(SCHEMA_ENTRY))
+ .build();
+
+ static final String NAME_MAPPING_CHANGE =
+ "regenerate the name mapping property to cover the schema";
+
+ static final String NO_TABLE_REASON =
+ "the table does not exist and no file schema can create it; a real run sends every file to"
+ + " the error output";
+
+ /** Wide inputs would otherwise put every column of every schema into one log entry. */
+ private static final int MAX_RENDERED_ENTRIES = 50;
+
+ private static final int MAX_RENDERED_CHANGES = 10;
+
+ private final IcebergCatalogConfig catalogConfig;
+ private final String identifier;
+ private final CommitSchemaUnion.Settings settings;
+ private transient @MonotonicNonNull Catalog catalog;
+
+ DryRunReport(
+ IcebergCatalogConfig catalogConfig, String identifier, CommitSchemaUnion.Settings settings) {
+ this.catalogConfig = catalogConfig;
+ this.identifier = identifier;
+ this.settings = settings;
+ }
+
+ /** One {@code schemas} entry before it is a Row: the totals and the rendered log read these. */
+ private static final class SchemaEntry {
+ final String key;
+ final String schema;
+ final long files;
+ final List changes;
+ final boolean allowed;
+ final String reason;
+
+ SchemaEntry(
+ String key,
+ String schema,
+ long files,
+ List changes,
+ boolean allowed,
+ String reason) {
+ this.key = key;
+ this.schema = schema;
+ this.files = files;
+ this.changes = changes;
+ this.allowed = allowed;
+ this.reason = reason;
+ }
+
+ Row toRow() {
+ return Row.withSchema(SCHEMA_ENTRY)
+ .withFieldValue("schema_key", key)
+ .withFieldValue("schema", schema)
+ .withFieldValue("num_files", files)
+ .withFieldValue("changes", changes)
+ .withFieldValue("allowed", allowed)
+ .withFieldValue("reason", reason)
+ .build();
+ }
+ }
+
+ /** The window's schema groups: the ones the plan sees, and the files that contribute none. */
+ private static final class Input {
+ final List readable = new ArrayList<>();
+ long unreadableFiles;
+ long uncheckedFiles;
+
+ static Input of(List schemas) {
+ Input input = new Input();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ if (group.getSchemaJson().equals(ReadFooterSchema.UNREADABLE_KEY)) {
+ input.unreadableFiles += group.getFiles();
+ } else if (group.getSchemaJson().equals(ReadFooterSchema.UNCHECKED_FORMAT_KEY)) {
+ input.uncheckedFiles += group.getFiles();
+ } else {
+ input.readable.add(group);
+ }
+ }
+ return input;
+ }
+ }
+
+ private static final class Totals {
+ int allowedSchemas;
+ long allowedFiles;
+ int incompatibleSchemas;
+ long incompatibleFiles;
+
+ static Totals of(List entries) {
+ Totals totals = new Totals();
+ for (SchemaEntry entry : entries) {
+ if (entry.allowed) {
+ totals.allowedSchemas++;
+ totals.allowedFiles += entry.files;
+ } else {
+ totals.incompatibleSchemas++;
+ totals.incompatibleFiles += entry.files;
+ }
+ }
+ return totals;
+ }
+ }
+
+ @ProcessElement
+ public void process(
+ @Element List schemas, OutputReceiver out) {
+ if (catalog == null) {
+ catalog = catalogConfig.catalog();
+ }
+ TableIdentifier tableId = IcebergUtils.parseTableIdentifier(identifier);
+ Input input = Input.of(schemas);
+ SchemaPlan plan = CommitSchemaUnion.plan(catalog, tableId, input.readable, settings);
+
+ List entries = new ArrayList<>();
+ for (CollectDistinctSchemas.SchemaGroup group : input.readable) {
+ entries.add(schemaEntry(group, plan));
+ }
+ Totals totals = Totals.of(entries);
+
+ SchemaPlan.@Nullable Creation creation = null;
+ if (plan instanceof SchemaPlan.Creation) {
+ creation = (SchemaPlan.Creation) plan;
+ }
+ @Nullable String creationProblem = creation == null ? null : creation.problem;
+ long files =
+ totals.allowedFiles
+ + totals.incompatibleFiles
+ + input.unreadableFiles
+ + input.uncheckedFiles;
+ boolean noTable = creation != null && creation.newSchema == null && files > 0;
+ boolean allowed =
+ totals.incompatibleSchemas == 0
+ && plan.configProblems.isEmpty()
+ && creationProblem == null
+ && !noTable;
+ boolean wouldFail =
+ creationProblem != null
+ || (settings.handling == IncompatibleSchemaHandling.FAIL_PIPELINE
+ && (totals.incompatibleSchemas > 0 || !plan.configProblems.isEmpty()));
+ boolean wouldCreateTable = creation != null && creation.canCreate() && !wouldFail;
+ String consequence = consequence(totals, plan.configProblems, creationProblem, noTable);
+
+ List configProblems = new ArrayList<>(plan.configProblems);
+ if (creationProblem != null) {
+ configProblems.add(creationProblem);
+ }
+ List tableChanges = new ArrayList<>();
+ if (plan instanceof SchemaPlan.Evolution && ((SchemaPlan.Evolution) plan).repairsNameMapping) {
+ tableChanges.add(NAME_MAPPING_CHANGE);
+ }
+ boolean uncheckedRegistered =
+ settings.config.getUnverifiableFileHandling()
+ == SchemaEvolutionConfig.UnverifiableFileHandling.ACCEPT
+ && !noTable;
+ List entryRows = new ArrayList<>();
+ for (SchemaEntry entry : entries) {
+ entryRows.add(entry.toRow());
+ }
+
+ Row.FieldValueBuilder report =
+ Row.withSchema(REPORT_SCHEMA)
+ .withFieldValue("allowed", allowed)
+ .withFieldValue("reason", consequence)
+ .withFieldValue("would_create_table", wouldCreateTable)
+ .withFieldValue("files_allowed", totals.allowedFiles)
+ .withFieldValue("files_incompatible", totals.incompatibleFiles)
+ .withFieldValue("files_unreadable", input.unreadableFiles)
+ .withFieldValue("files_unchecked", input.uncheckedFiles)
+ .withFieldValue("unchecked_registered", uncheckedRegistered)
+ .withFieldValue("table_changes", tableChanges)
+ .withFieldValue("config_problems", configProblems)
+ .withFieldValue("schemas", entryRows);
+ List createdColumns = Collections.emptyList();
+ if (creation != null && creation.newSchema != null) {
+ createdColumns = createdColumns(creation.newSchema);
+ report =
+ report.withFieldValue(
+ "created_table",
+ Row.withSchema(CREATED_TABLE)
+ .withFieldValue("schema", SchemaParser.toJson(creation.newSchema))
+ .withFieldValue("columns", createdColumns)
+ .build());
+ }
+ out.output(report.build());
+
+ numFilesAllowed.inc(totals.allowedFiles);
+ numFilesIncompatible.inc(totals.incompatibleFiles);
+ numFilesUnreadable.inc(input.unreadableFiles);
+ numFilesUnchecked.inc(input.uncheckedFiles);
+ numConfigProblems.inc(configProblems.size());
+ LOG.info(
+ "Dry run for {}{}: {}{}{}\n{}",
+ identifier,
+ wouldCreateTable ? " (table would be created)" : "",
+ summary(input, totals),
+ consequence.isEmpty() ? "" : "; " + consequence,
+ tableChanges.isEmpty() ? "" : "; " + String.join("; ", tableChanges),
+ render(entries, createdColumns));
+ }
+
+ private SchemaEntry schemaEntry(CollectDistinctSchemas.SchemaGroup group, SchemaPlan plan) {
+ String json = group.getSchemaJson();
+ @Nullable String incompatibleReason = plan.incompatibleReason(json);
+ // a created table's columns are reported once, as created_table: no delta exists then
+ List changes = Collections.emptyList();
+ SchemaPlan.@Nullable SchemaToMerge toMerge = plan.toMerge(json);
+ if (toMerge != null && toMerge.delta != null) {
+ changes = toMerge.delta.descriptions();
+ }
+ return new SchemaEntry(
+ key(json),
+ json,
+ group.getFiles(),
+ changes,
+ incompatibleReason == null,
+ incompatibleReason == null ? "" : incompatibleReason);
+ }
+
+ private static String summary(Input input, Totals totals) {
+ return String.format(
+ "%d distinct schemas; %d allowed covering %d files; %d incompatible covering %d files;"
+ + " %d files unreadable; %d files unchecked (ORC or Avro)",
+ input.readable.size(),
+ totals.allowedSchemas,
+ totals.allowedFiles,
+ totals.incompatibleSchemas,
+ totals.incompatibleFiles,
+ input.unreadableFiles,
+ input.uncheckedFiles);
+ }
+
+ private String consequence(
+ Totals totals,
+ List configProblems,
+ @Nullable String creationProblem,
+ boolean noTable) {
+ IncompatibleSchemaHandling handling = settings.handling;
+ List parts = new ArrayList<>();
+ if (creationProblem != null) {
+ parts.add("a real run would fail to create the table: " + creationProblem);
+ }
+ if (noTable) {
+ parts.add(NO_TABLE_REASON);
+ }
+ if (totals.incompatibleSchemas > 0) {
+ parts.add(
+ handling == IncompatibleSchemaHandling.FAIL_PIPELINE
+ ? "a real run would fail before committing (" + handling + ")"
+ : "a real run would route "
+ + totals.incompatibleFiles
+ + " files to errors ("
+ + handling
+ + ")");
+ }
+ if (!configProblems.isEmpty()) {
+ parts.add(
+ (handling == IncompatibleSchemaHandling.FAIL_PIPELINE
+ ? "a real run would fail on the configuration: "
+ : "a real run would warn on the configuration: ")
+ + String.join("; ", configProblems));
+ }
+ return String.join("; ", parts);
+ }
+
+ private static String render(List entries, List createdColumns) {
+ StringBuilder rendered = new StringBuilder();
+ for (SchemaEntry entry : entries.subList(0, Math.min(entries.size(), MAX_RENDERED_ENTRIES))) {
+ rendered.append(
+ String.format(
+ " %-7s %8d allowed=%-5s %s %s%n",
+ entry.key, entry.files, entry.allowed, cut(entry.changes), entry.reason));
+ }
+ if (entries.size() > MAX_RENDERED_ENTRIES) {
+ rendered
+ .append(" ... and ")
+ .append(entries.size() - MAX_RENDERED_ENTRIES)
+ .append(" more schemas\n");
+ }
+ if (!createdColumns.isEmpty()) {
+ rendered.append(" created table: ").append(cut(createdColumns)).append('\n');
+ }
+ return rendered.toString();
+ }
+
+ private static List cut(List changes) {
+ if (changes.size() <= MAX_RENDERED_CHANGES) {
+ return changes;
+ }
+ List shown = new ArrayList<>(changes.subList(0, MAX_RENDERED_CHANGES));
+ shown.add("... and " + (changes.size() - MAX_RENDERED_CHANGES) + " more");
+ return shown;
+ }
+
+ /** Top-level columns of the table a real run would create; nested detail is in the schema. */
+ private static List createdColumns(org.apache.iceberg.Schema created) {
+ List columns = new ArrayList<>();
+ for (org.apache.iceberg.types.Types.NestedField field : created.columns()) {
+ columns.add(
+ "create "
+ + (field.isOptional() ? "optional " : "required ")
+ + field.name()
+ + " "
+ + typeLabel(field.type()));
+ }
+ return columns;
+ }
+
+ /** Nested types print field ids, which a creation reassigns, so only their kind is named. */
+ private static String typeLabel(Type type) {
+ if (type.isPrimitiveType()) {
+ return type.toString();
+ }
+ return type.typeId().name().toLowerCase(Locale.ROOT);
+ }
+
+ static String key(String schemaJson) {
+ return "s"
+ + Hashing.murmur3_32_fixed().hashUnencodedChars(schemaJson).toString().substring(0, 6);
+ }
+}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java
index e559feab3512..389f03490fe4 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java
@@ -52,15 +52,25 @@ class ReadFooterSchema extends DoFn
private static final Counter numFooterReadErrors =
counter(ReadFooterSchema.class, FOOTER_READ_ERRORS_COUNTER);
+ /**
+ * Emitted in a dry run for files that contribute no schema. The key travels in the group's schema
+ * JSON field, which the combine groups by, so the files of one kind add up without a second path.
+ */
+ static final String UNREADABLE_KEY = "unread";
+
+ static final String UNCHECKED_FORMAT_KEY = "unchecked";
+
+ private final SchemaEvolutionConfig config;
private final int threadPoolSize;
private final int maxInFlightTasks;
private transient @MonotonicNonNull BoundedAsyncTasks tasks;
- ReadFooterSchema() {
- this(DEFAULT_THREAD_POOL_SIZE, DEFAULT_MAX_IN_FLIGHT_TASKS);
+ ReadFooterSchema(SchemaEvolutionConfig config) {
+ this(config, DEFAULT_THREAD_POOL_SIZE, DEFAULT_MAX_IN_FLIGHT_TASKS);
}
- ReadFooterSchema(int threadPoolSize, int maxInFlightTasks) {
+ ReadFooterSchema(SchemaEvolutionConfig config, int threadPoolSize, int maxInFlightTasks) {
+ this.config = config;
this.threadPoolSize = threadPoolSize;
this.maxInFlightTasks = maxInFlightTasks;
}
@@ -148,17 +158,18 @@ private static void count(ReadResult result) {
}
}
- private static Callable createReadTask(
+ private Callable createReadTask(
String filePath, Instant timestamp, BoundedWindow window, PaneInfo paneInfo) {
return () -> {
FileFormat format;
try {
format = AddFiles.inferFormat(filePath);
} catch (AddFiles.UnknownFormatException e) {
- return new ReadResult(null, false, timestamp, window, paneInfo);
+ return new ReadResult(dryRunMarker(UNREADABLE_KEY), false, timestamp, window, paneInfo);
}
if (!format.equals(FileFormat.PARQUET)) {
- return new ReadResult(null, false, timestamp, window, paneInfo);
+ return new ReadResult(
+ dryRunMarker(UNCHECKED_FORMAT_KEY), false, timestamp, window, paneInfo);
}
try {
ParquetMetadata footer = ParquetFooters.read(filePath);
@@ -168,8 +179,16 @@ private static Callable createReadTask(
"Could not read the footer of {}; the file will not contribute to schema inference: {}",
filePath,
AddFiles.errorMessage(e));
- return new ReadResult(null, true, timestamp, window, paneInfo);
+ return new ReadResult(dryRunMarker(UNREADABLE_KEY), true, timestamp, window, paneInfo);
}
};
}
+
+ /** A dry run counts files that contribute no schema; a real run handles them at registration. */
+ private CollectDistinctSchemas.@Nullable SchemaGroup dryRunMarker(String key) {
+ if (!config.getDryRun()) {
+ return null;
+ }
+ return CollectDistinctSchemas.SchemaGroup.of(key, 1, Collections.emptyList());
+ }
}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java
index bacdac2e7cfe..743558e68f3c 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java
@@ -60,7 +60,14 @@ private SchemaDelta(List changes) {
* (dotted, empty, case-colliding) come back as conflicts without attempting the union.
*/
static SchemaDelta classify(Table table, Schema fileSchema) {
- Schema before = table.schema();
+ return classify(table, table.schema(), fileSchema);
+ }
+
+ /**
+ * Classifies against {@code before}, the caller's snapshot of the table schema. The union itself
+ * is applied to the table's current metadata, which the caller keeps equal to the snapshot.
+ */
+ static SchemaDelta classify(Table table, Schema before, Schema fileSchema) {
if (before.sameSchema(fileSchema)) {
return new SchemaDelta(Collections.emptyList());
}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java
index fcbfc906f3c0..da6fd33ac9d1 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfig.java
@@ -55,7 +55,23 @@
* conflicts with the table or with another file's schema. {@link IncompatibleSchemaHandling}
* decides whether that fails the pipeline before any schema commit (the batch default) or skips the
* schema so its files reach the error output (the streaming default). Files whose footer cannot be
- * read or converted always go to the error output and never fail the pipeline.
+ * read or converted always go to the error output and never fail the pipeline. When the table does
+ * not exist, the pre-pass creates it from the union of the file schemas; if no readable Parquet
+ * schema can seed it, nothing is created and every file goes to the error output.
+ *
+ * Dry run. Reports what a real run would do on the {@code dry_run_report} output, one row
+ * per window; nothing is committed or registered. {@code allowed} is true when every file schema
+ * can be merged and the configuration raises no problem; otherwise {@code reason} says what a real
+ * run would do about it. {@code schemas} holds one entry per distinct file schema with the changes
+ * a real run would make and, when the schema cannot be merged, the option or conflict to fix;
+ * {@code created_table} shows the table a real run would create. The report is a PCollection like
+ * any other, so attach a sink to keep it; it is also logged at INFO and the file counts are
+ * published as counters ({@code numDryRunFilesAllowed}, {@code numDryRunFilesIncompatible}, {@code
+ * numDryRunFilesUnreadable}, {@code numDryRunFilesUnchecked}, {@code numDryRunConfigProblems}).
+ * Adjust the settings, rerun until the report is allowed, then run for real with an error output
+ * attached. Against a missing table the dry run computes the union through the catalog's
+ * create-transaction API, which a REST catalog serves as a stage-create request: the credentials
+ * need table-create permission even though no table is created.
*/
@AutoValue
public abstract class SchemaEvolutionConfig implements Serializable {
@@ -104,6 +120,12 @@ public boolean isPinned(String columnPath) {
return getRequiredColumns().contains(columnPath);
}
+ /**
+ * Report what the pre-pass would do on the {@code dry_run_report} output, one row per window;
+ * commit and register nothing.
+ */
+ public abstract boolean getDryRun();
+
/**
* Unset resolves by mode: {@code FAIL_PIPELINE} in batch, {@code ROUTE_TO_ERRORS} in streaming.
*/
@@ -144,7 +166,8 @@ public static Builder builder() {
return new AutoValue_SchemaEvolutionConfig.Builder()
.setOptions(Collections.emptySet())
.setRequiredColumns(Collections.emptySet())
- .setUnverifiableFileHandling(UnverifiableFileHandling.REJECT);
+ .setUnverifiableFileHandling(UnverifiableFileHandling.REJECT)
+ .setDryRun(false);
}
@AutoValue.Builder
@@ -158,6 +181,8 @@ public abstract Builder setIncompatibleSchemaHandling(
public abstract Builder setUnverifiableFileHandling(UnverifiableFileHandling handling);
+ public abstract Builder setDryRun(boolean dryRun);
+
abstract SchemaEvolutionConfig autoBuild();
/** Any setting without an option would silently do nothing, so they are rejected. */
@@ -172,10 +197,11 @@ public SchemaEvolutionConfig build() {
Preconditions.checkArgument(
config.isEnabled()
|| (config.getRequiredColumns().isEmpty()
+ && !config.getDryRun()
&& config.getIncompatibleSchemaHandling() == null
&& config.getUnverifiableFileHandling() == UnverifiableFileHandling.REJECT),
- "required columns, incompatible schema handling and unverifiable file handling need at"
- + " least one schema evolution option");
+ "required columns, dry run, incompatible schema handling and unverifiable file"
+ + " handling need at least one schema evolution option");
return config;
}
}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java
new file mode 100644
index 000000000000..e9e72199c82b
--- /dev/null
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java
@@ -0,0 +1,618 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.Transaction;
+import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/**
+ * What one schema commit would do, as data: which distinct file schemas it merges into the table,
+ * which it refuses and why, and the schema the table ends up with. {@link #compute} works it out on
+ * scratch transactions that are never committed: {@code classify} each schema against the table
+ * (most common first), then {@code fold} the accepted ones into a single union, which keeps
+ * per-schema blame for cross-schema conflicts. {@link CommitSchemaUnion} executes the plan and the
+ * dry run reports it, so a check added here reaches both and the two cannot drift.
+ */
+abstract class SchemaPlan {
+ final List schemasToMerge;
+ final List incompatibleSchemas;
+
+ /** The union to replay, or the schema to create the table with; null when nothing changes. */
+ final @Nullable Schema newSchema;
+
+ /** Problems the configuration raises against the planned schema, as messages. */
+ final List configProblems;
+
+ private SchemaPlan(Verdicts verdicts, @Nullable Schema newSchema, List configProblems) {
+ this.schemasToMerge = verdicts.toMerge;
+ this.incompatibleSchemas = verdicts.incompatible;
+ this.newSchema = newSchema;
+ this.configProblems = configProblems;
+ }
+
+ /**
+ * The schema's entry when the table must change for it; null when the schema is incompatible or
+ * the table already covers it.
+ */
+ @Nullable SchemaToMerge toMerge(String schemaJson) {
+ for (SchemaToMerge item : schemasToMerge) {
+ if (item.json.equals(schemaJson)) {
+ return item;
+ }
+ }
+ return null;
+ }
+
+ /** Why the schema is incompatible; null when it is compatible. */
+ @Nullable String incompatibleReason(String schemaJson) {
+ for (IncompatibleSchema item : incompatibleSchemas) {
+ if (item.schemaJson.equals(schemaJson)) {
+ return item.reason;
+ }
+ }
+ return null;
+ }
+
+ /** The plan against an existing table. */
+ static final class Evolution extends SchemaPlan {
+ final Table table;
+
+ /** The table schema every decision was made against. */
+ final Schema base;
+
+ /** Whether the commit regenerates the name mapping property, schema change or not. */
+ final boolean repairsNameMapping;
+
+ private Evolution(
+ Table table,
+ Schema base,
+ Verdicts verdicts,
+ @Nullable Schema newSchema,
+ boolean repairsNameMapping,
+ List configProblems) {
+ super(verdicts, newSchema, configProblems);
+ this.table = table;
+ this.base = base;
+ this.repairsNameMapping = repairsNameMapping;
+ }
+
+ /** A transaction on the table, checked against the snapshot this plan reasoned about. */
+ Transaction newTransaction(TableIdentifier tableId) {
+ return transactionOn(table, base, tableId);
+ }
+ }
+
+ /** The plan when the table does not exist; {@code newSchema} is what it would be created with. */
+ static final class Creation extends SchemaPlan {
+ /** Set together with {@link #sortOrder} unless {@link #problem} is. */
+ final @Nullable PartitionSpec spec;
+
+ final @Nullable SortOrder sortOrder;
+
+ /** Why the table cannot be created as configured; fails a real run under either handling. */
+ final @Nullable String problem;
+
+ private Creation(
+ Verdicts verdicts,
+ @Nullable Schema newSchema,
+ @Nullable PartitionSpec spec,
+ @Nullable SortOrder sortOrder,
+ @Nullable String problem,
+ List configProblems) {
+ super(verdicts, newSchema, configProblems);
+ this.spec = spec;
+ this.sortOrder = sortOrder;
+ this.problem = problem;
+ }
+
+ /** No file schema is left to create the table from. */
+ static Creation nothingToCreate(Verdicts verdicts) {
+ return new Creation(verdicts, null, null, null, null, new ArrayList<>());
+ }
+
+ static Creation of(
+ Verdicts verdicts,
+ Schema created,
+ PartitionSpec spec,
+ SortOrder sortOrder,
+ List configProblems) {
+ return new Creation(verdicts, created, spec, sortOrder, null, configProblems);
+ }
+
+ /** The configured partition or sort fields do not fit the schema. */
+ static Creation blocked(
+ Verdicts verdicts, Schema created, String problem, List configProblems) {
+ return new Creation(verdicts, created, null, null, problem, configProblems);
+ }
+
+ /** There is a schema to create from, and the configured partition and sort fields fit it. */
+ boolean canCreate() {
+ return newSchema != null && problem == null;
+ }
+ }
+
+ /** One distinct file schema the table must change for, and may under the options. */
+ static final class SchemaToMerge {
+ final Schema schema;
+ final String json;
+ final long files;
+
+ /** Null on the create path: the seed table is empty, so there is nothing to relax. */
+ final @Nullable SchemaDelta delta;
+
+ SchemaToMerge(Schema schema, String json, long files, @Nullable SchemaDelta delta) {
+ this.schema = schema;
+ this.json = json;
+ this.files = files;
+ this.delta = delta;
+ }
+ }
+
+ /** One distinct file schema the commit refuses, with the reason. */
+ static final class IncompatibleSchema {
+ final String schemaJson;
+ final long files;
+ final String reason;
+
+ IncompatibleSchema(String schemaJson, long files, String reason) {
+ this.schemaJson = schemaJson;
+ this.files = files;
+ this.reason = reason;
+ }
+
+ @Override
+ public String toString() {
+ return files + " file(s) with schema " + truncate(schemaJson) + ": " + reason;
+ }
+ }
+
+ /** Canonical JSON of a wide schema runs to hundreds of KB; the reason is what matters. */
+ private static final int MAX_SCHEMA_JSON_CHARS = 1024;
+
+ private static String truncate(String json) {
+ if (json.length() <= MAX_SCHEMA_JSON_CHARS) {
+ return json;
+ }
+ return json.substring(0, MAX_SCHEMA_JSON_CHARS)
+ + "... ("
+ + (json.length() - MAX_SCHEMA_JSON_CHARS)
+ + " chars truncated)";
+ }
+
+ /** Where each distinct file schema of the window stands while the plan is worked out. */
+ private static final class Verdicts {
+ final List toMerge = new ArrayList<>();
+ final List incompatible = new ArrayList<>();
+
+ void refuse(CollectDistinctSchemas.SchemaGroup group, String reason) {
+ incompatible.add(new IncompatibleSchema(group.getSchemaJson(), group.getFiles(), reason));
+ }
+
+ void refuse(Conflict conflict) {
+ toMerge.remove(conflict.schema);
+ incompatible.add(
+ new IncompatibleSchema(conflict.schema.json, conflict.schema.files, conflict.reason));
+ }
+ }
+
+ /**
+ * A schema that fits the table on its own but not the union of the window: a column another
+ * schema of the window adds has the same name with a different type, or a name differing only in
+ * case. Classification checks each schema against the table alone, so this surfaces only while
+ * the union is staged.
+ */
+ private static final class Conflict {
+ final SchemaToMerge schema;
+ final String reason;
+
+ Conflict(SchemaToMerge schema, String reason) {
+ this.schema = schema;
+ this.reason = reason;
+ }
+ }
+
+ /**
+ * The plan for the window's schemas against the table as loaded ({@code null} when it does not
+ * exist). Nothing is committed; on a missing table the fold goes through the catalog's create
+ * transaction, which a REST catalog serves as a stage-create request.
+ *
+ * @param schemas the window's distinct schema groups, most common first
+ */
+ static SchemaPlan compute(
+ Catalog catalog,
+ TableIdentifier tableId,
+ @Nullable Table table,
+ List schemas,
+ CommitSchemaUnion.Settings settings) {
+ if (table == null) {
+ return planCreation(catalog, tableId, schemas, settings);
+ }
+ return planEvolution(table, tableId, schemas, settings.config);
+ }
+
+ private static Evolution planEvolution(
+ Table table,
+ TableIdentifier tableId,
+ List schemas,
+ SchemaEvolutionConfig config) {
+ Schema base = table.schema();
+ if (schemas.isEmpty()) {
+ // the pipeline never commits for a window without schemas; a direct caller gets the same
+ return new Evolution(table, base, new Verdicts(), null, false, new ArrayList<>());
+ }
+ Verdicts verdicts = classify(table, base, schemas, config);
+ @Nullable Schema newSchema = fold(table, base, tableId, verdicts);
+ Schema afterwards = newSchema != null ? newSchema : base;
+ boolean repairsNameMapping = needsNameMapping(table.properties(), afterwards);
+ return new Evolution(table, base, verdicts, newSchema, repairsNameMapping, new ArrayList<>());
+ }
+
+ private static Creation planCreation(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List schemas,
+ CommitSchemaUnion.Settings settings) {
+ Verdicts verdicts = classifyForCreate(schemas);
+ @Nullable Schema union = foldForCreate(catalog, tableId, verdicts);
+ if (union == null) {
+ return Creation.nothingToCreate(verdicts);
+ }
+ Schema created = createdSchema(union, settings.config);
+ List configProblems = new ArrayList<>();
+ addPinProblems(tableId, created, settings.config, configProblems);
+ CommitSchemaUnion.NewTableSettings newTable = settings.newTable;
+ try {
+ PartitionSpec spec = PartitionUtils.toPartitionSpec(newTable.partitionFields, created);
+ SortOrder sortOrder = SortOrderUtils.toSortOrder(newTable.sortFields, created);
+ return Creation.of(verdicts, created, spec, sortOrder, configProblems);
+ } catch (IllegalArgumentException | ValidationException e) {
+ String problem =
+ "Table "
+ + tableId
+ + " cannot be created with partition fields "
+ + newTable.partitionFields
+ + " and sort fields "
+ + newTable.sortFields
+ + " on the union of the file schemas: "
+ + AddFiles.errorMessage(e);
+ return Creation.blocked(verdicts, created, problem, configProblems);
+ }
+ }
+
+ private static void addPinProblems(
+ TableIdentifier tableId,
+ Schema created,
+ SchemaEvolutionConfig config,
+ List configProblems) {
+ List unenforceable = unenforceablePins(created, config);
+ if (!unenforceable.isEmpty()) {
+ configProblems.add(
+ "Pinned column(s) "
+ + unenforceable
+ + " appear in none of the file schemas creating "
+ + tableId
+ + ", or their spelling does not match the column path; the created table cannot"
+ + " make them required");
+ }
+ }
+
+ /**
+ * Sorts the window's schemas into the ones the table must change for ({@code toMerge}) and the
+ * ones it must not ({@code incompatible}, with the reason); schemas the table already covers drop
+ * out.
+ */
+ private static Verdicts classify(
+ Table table,
+ Schema base,
+ List schemas,
+ SchemaEvolutionConfig config) {
+ Verdicts verdicts = new Verdicts();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ Schema fileSchema =
+ FileSchemas.markRequired(
+ SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns());
+ SchemaDelta delta = SchemaDelta.classify(table, base, fileSchema);
+ if (delta.isEmpty()) {
+ continue;
+ }
+ if (!delta.allowedBy(config)) {
+ verdicts.refuse(group, delta.disallowedReason(config));
+ continue;
+ }
+ verdicts.toMerge.add(
+ new SchemaToMerge(fileSchema, group.getSchemaJson(), group.getFiles(), delta));
+ }
+ return verdicts;
+ }
+
+ /**
+ * Unions the accepted schemas into the table schema on scratch transactions that are never
+ * committed, relaxing every field the window adds; a schema that conflicts with another only
+ * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the
+ * folded schema, or null when nothing needs to change.
+ */
+ private static @Nullable Schema fold(
+ Table table, Schema base, TableIdentifier tableId, Verdicts verdicts) {
+ while (!verdicts.toMerge.isEmpty()) {
+ Transaction scratch = transactionOn(table, base, tableId);
+ @Nullable Conflict conflict = stageAll(scratch, verdicts.toMerge);
+ if (conflict == null) {
+ relaxNewRequiredFields(scratch, base);
+ return scratch.table().schema();
+ }
+ verdicts.refuse(conflict);
+ }
+ return null;
+ }
+
+ /**
+ * The evolve path refuses names no table can absorb (dotted, empty, differing only in case within
+ * one file) as conflicts in classify; a table must not be born with them either.
+ */
+ private static Verdicts classifyForCreate(List schemas) {
+ Verdicts verdicts = new Verdicts();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson());
+ List invalidNames = new ArrayList<>();
+ ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "", invalidNames);
+ if (!invalidNames.isEmpty()) {
+ verdicts.refuse(
+ group, "file schema has column names no table can hold: " + describe(invalidNames));
+ continue;
+ }
+ verdicts.toMerge.add(
+ new SchemaToMerge(fileSchema, group.getSchemaJson(), group.getFiles(), null));
+ }
+ return verdicts;
+ }
+
+ /**
+ * Unions the accepted schemas into one on scratch create transactions that are never committed,
+ * seeded by the most common schema; a schema that conflicts with the others moves to {@code
+ * incompatible}, leaves {@code toMerge}, and the fold restarts without it. Returns the union, or
+ * null when no schema is left to create from. A REST catalog serves each create transaction as a
+ * stage-create request, so the caller needs table-create permission even in a dry run.
+ */
+ private static @Nullable Schema foldForCreate(
+ Catalog catalog, TableIdentifier tableId, Verdicts verdicts) {
+ List toMerge = verdicts.toMerge;
+ while (!toMerge.isEmpty()) {
+ Schema seed = toMerge.get(0).schema;
+ Transaction scratch = catalog.buildTable(tableId, seed).createTransaction();
+ @Nullable Conflict conflict = stageAll(scratch, toMerge.subList(1, toMerge.size()));
+ if (conflict == null) {
+ return scratch.table().schema();
+ }
+ verdicts.refuse(conflict);
+ }
+ return null;
+ }
+
+ /**
+ * Pins the created schema did not end up enforcing: the column appears in no file schema, or the
+ * configured spelling resolves to a field the pin walk did not reach (a short container spelling
+ * like a.b for a.element.b, or a path inside a map key). Such a pin would stay inert forever,
+ * since later windows only add columns optional, so the plan reports it as a configuration
+ * problem.
+ */
+ static List unenforceablePins(Schema created, SchemaEvolutionConfig config) {
+ List unenforceable = new ArrayList<>();
+ for (String pin : config.getRequiredColumns()) {
+ Types.NestedField field = created.findField(pin);
+ if (field == null || field.isOptional()) {
+ unenforceable.add(pin);
+ }
+ }
+ Collections.sort(unenforceable);
+ return unenforceable;
+ }
+
+ /**
+ * The created schema: every field optional at every level, list elements and map values included,
+ * except pinned paths and their ancestors, which stay required so the schema advertises the
+ * guarantee the per-file pin check enforces (a null ancestor nulls the pinned leaf). Map key
+ * subtrees keep their declared shape (keys are required by definition; pins inside them are not
+ * honored). Nothing depends on a created table's schema yet, so this is the schema-authoring
+ * moment; evolution never tightens columns afterwards. Column order is the union's, which is the
+ * canonical (name-sorted) order of the file schemas.
+ */
+ static Schema createdSchema(Schema merged, SchemaEvolutionConfig config) {
+ Pins pins = new Pins(config.getRequiredColumns());
+ List fields = new ArrayList<>();
+ for (Types.NestedField field : merged.asStruct().fields()) {
+ fields.add(createdField(field, field.name(), pins));
+ }
+ return new Schema(fields);
+ }
+
+ private static Types.NestedField createdField(Types.NestedField field, String path, Pins pins) {
+ boolean required = pins.isPinnedOrAncestorOfPin(path);
+ return Types.NestedField.from(field)
+ .ofType(createdType(field.type(), path, pins))
+ .isOptional(!required)
+ .build();
+ }
+
+ private static Type createdType(Type type, String path, Pins pins) {
+ if (type.isStructType()) {
+ List fields = new ArrayList<>();
+ for (Types.NestedField field : type.asStructType().fields()) {
+ fields.add(createdField(field, path + "." + field.name(), pins));
+ }
+ return Types.StructType.of(fields);
+ }
+ if (type.isListType()) {
+ Types.ListType list = type.asListType();
+ String elementPath = path + ".element";
+ Type elementType = createdType(list.elementType(), elementPath, pins);
+ boolean required = pins.isPinnedOrAncestorOfPin(elementPath);
+ return required
+ ? Types.ListType.ofRequired(list.elementId(), elementType)
+ : Types.ListType.ofOptional(list.elementId(), elementType);
+ }
+ if (type.isMapType()) {
+ Types.MapType map = type.asMapType();
+ String valuePath = path + ".value";
+ Type valueType = createdType(map.valueType(), valuePath, pins);
+ boolean required = pins.isPinnedOrAncestorOfPin(valuePath);
+ return required
+ ? Types.MapType.ofRequired(map.keyId(), map.valueId(), map.keyType(), valueType)
+ : Types.MapType.ofOptional(map.keyId(), map.valueId(), map.keyType(), valueType);
+ }
+ return type;
+ }
+
+ /**
+ * Stages one union per accepted schema onto {@code txn}: a scratch transaction on the evolve path
+ * (its per-schema versions stay in memory; only the folded result is ever committed), the create
+ * transaction on the create path. A schema can conflict with another schema's additions, which
+ * only surfaces while staging and poisons the transaction, so on a conflict the offender is
+ * returned for the caller to drop and retry with a fresh transaction.
+ */
+ private static @Nullable Conflict stageAll(Transaction txn, List toMerge) {
+ for (SchemaToMerge item : toMerge) {
+ // classify checked each schema against the base table only; a column that differs only in
+ // case from one an EARLIER schema of the window added would union as a second column.
+ List collisions = new ArrayList<>();
+ ColumnNameChecks.findCaseCollisions(
+ txn.table().schema().asStruct(), item.schema.asStruct(), "", collisions);
+ if (!collisions.isEmpty()) {
+ return new Conflict(
+ item, "conflicts with another file schema in the same window: " + describe(collisions));
+ }
+ // Both caught types carry staging conflicts: ValidationException from Schema
+ // construction at apply ("multiple fields for name"), IllegalArgumentException from
+ // SchemaUpdate preconditions ("Cannot change column type").
+ try {
+ stage(txn, item);
+ } catch (ValidationException | IllegalArgumentException e) {
+ return new Conflict(
+ item,
+ "conflicts with another file schema in the same window: " + AddFiles.errorMessage(e));
+ }
+ }
+ return null;
+ }
+
+ private static String describe(List changes) {
+ List descriptions = new ArrayList<>();
+ for (SchemaChange change : changes) {
+ descriptions.add(change.description);
+ }
+ return String.join("; ", descriptions);
+ }
+
+ /**
+ * Iceberg refreshes the table on every {@code newTransaction()}, so a concurrent schema commit
+ * can slip between two transactions here. Any drift from the snapshot the window classified
+ * against is thrown as {@link CommitFailedException} so the retry in {@link CommitSchemaUnion}
+ * reloads and rebuilds, leaving the replay checkState as a pure bug detector.
+ */
+ static Transaction transactionOn(Table table, Schema base, TableIdentifier tableId) {
+ Transaction txn = table.newTransaction();
+ if (!txn.table().schema().sameSchema(base)) {
+ throw new CommitFailedException(
+ "concurrent schema change on %s while staging the schema union", tableId);
+ }
+ return txn;
+ }
+
+ private static void stage(Transaction txn, SchemaToMerge item) {
+ UpdateSchema update = txn.updateSchema().unionByNameWith(item.schema);
+ if (item.delta != null) {
+ for (String path : item.delta.absentRequiredPaths()) {
+ update = update.makeColumnOptional(path);
+ }
+ }
+ update.commit();
+ }
+
+ /**
+ * New columns are optional at every level. The union adds top-level columns optional but keeps
+ * the file's optionality below them, so one file's luck would otherwise impose required fields on
+ * everyone. Pins do not shape new columns: they keep existing required columns from being relaxed
+ * (SchemaDelta) and gate files at registration.
+ */
+ private static void relaxNewRequiredFields(Transaction txn, Schema before) {
+ List toRelax = newRequiredPaths(before, txn.table().schema());
+ if (toRelax.isEmpty()) {
+ return;
+ }
+ UpdateSchema update = txn.updateSchema();
+ for (String path : toRelax) {
+ update = update.makeColumnOptional(path);
+ }
+ update.commit();
+ }
+
+ /**
+ * Paths of required fields that {@code after} has and {@code before} lacks, in schema order;
+ * includes fields under lists and maps (a required list element or map value counts). Map key
+ * subtrees are skipped: keys are required by definition and relaxing inside a struct key would
+ * change key identity.
+ */
+ static List newRequiredPaths(Schema before, Schema after) {
+ Set beforeIds = TypeUtil.indexById(before.asStruct()).keySet();
+ List paths = new ArrayList<>();
+ collectNewRequired(after.asStruct(), "", beforeIds, paths);
+ return paths;
+ }
+
+ private static void collectNewRequired(
+ Type.NestedType type, String prefix, Set beforeIds, List paths) {
+ for (Types.NestedField field : type.fields()) {
+ if (type.isMapType() && field.fieldId() == type.asMapType().keyId()) {
+ continue;
+ }
+ String path = prefix + field.name();
+ if (!beforeIds.contains(field.fieldId()) && field.isRequired()) {
+ paths.add(path);
+ }
+ if (field.type().isNestedType()) {
+ collectNewRequired(field.type().asNestedType(), path + ".", beforeIds, paths);
+ }
+ }
+ }
+
+ /** The table's name mapping property is absent, malformed or does not cover {@code schema}. */
+ static boolean needsNameMapping(Map tableProperties, Schema schema) {
+ @Nullable NameMapping existing =
+ NameMappingUtils.parseOrNull(tableProperties.get(TableProperties.DEFAULT_NAME_MAPPING));
+ return existing == null || !NameMappingUtils.covers(existing, schema.asStruct());
+ }
+}
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
index 2e469ef6da71..3b58af201543 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
@@ -137,6 +137,20 @@ public void testInvalidHandlingValuesRejected() {
.getSchemaEvolution());
}
+ @Test
+ public void testDryRunParsed() {
+ SchemaEvolutionConfig config =
+ base()
+ .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION"))
+ .setDryRun(true)
+ .build()
+ .getSchemaEvolution();
+ assertNotNull(config);
+ assertTrue(config.getDryRun());
+ assertThrows(
+ IllegalArgumentException.class, () -> base().setDryRun(true).build().getSchemaEvolution());
+ }
+
@Test
public void testRouteToErrorsWithoutErrorHandlingRejected() {
IllegalArgumentException e =
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java
index 5e6cbdf4ea32..b9fe94f27a5c 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java
@@ -26,7 +26,9 @@
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.hasEntry;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
@@ -34,8 +36,10 @@
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.CharBuffer;
+import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.EnumSet;
@@ -71,6 +75,7 @@
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
import org.apache.iceberg.FileScanTask;
@@ -1047,6 +1052,315 @@ public void testEvolutionAddsColumnsBeforeRegisteringFiles() throws Exception {
"stats for the added column", nullCountsOf(table, "wide.parquet").containsKey(emailId));
}
+ @Test
+ public void testDryRunReportsWithoutCommittingOrRegistering() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String before = metadataLocation();
+ String covered = writeOneRecord("covered.parquet");
+ String wide = writeWider("wide.parquet");
+ String conflict = writeConflicting("conflict.parquet");
+
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(covered, wide, conflict)).apply(addFiles(DRY_RUN));
+ PAssert.that(output.get("errors")).empty();
+ PAssert.that(output.get("snapshots")).empty();
+ PAssert.that(output.get(AddFiles.DRY_RUN_TAG))
+ .satisfies(
+ rows -> {
+ Row report = report(rows);
+ Collection schemas = schemas(report);
+ assertEquals(3, schemas.size());
+ boolean sawAddition = false;
+ boolean sawConflict = false;
+ for (Row schema : schemas) {
+ Collection changes = schema.getArray("changes");
+ if (changes.contains("add optional email string")) {
+ sawAddition = schema.getBoolean("allowed");
+ }
+ if (!schema.getBoolean("allowed")) {
+ sawConflict = schema.getString("reason").contains("conflicts");
+ }
+ }
+ assertTrue(sawAddition);
+ assertTrue(sawConflict);
+ assertFalse(report.getBoolean("allowed"));
+ assertEquals(Long.valueOf(2), report.getInt64("files_allowed"));
+ assertEquals(Long.valueOf(1), report.getInt64("files_incompatible"));
+ assertThat(report.getString("reason"), containsString("would fail"));
+ return null;
+ });
+ assertEquals(0, countTransforms(pipeline, "ConvertToDataFiles"));
+ PipelineResult result = pipeline.run();
+ result.waitUntilFinish();
+ Table table = catalog.loadTable(tableId);
+ assertEquals(0, Iterables.size(table.snapshots()));
+ assertEquals(2, counted(result, DryRunReport.class, DryRunReport.FILES_ALLOWED_COUNTER));
+ assertEquals(1, counted(result, DryRunReport.class, DryRunReport.FILES_INCOMPATIBLE_COUNTER));
+ assertEquals(0, counted(result, DryRunReport.class, DryRunReport.FILES_UNREADABLE_COUNTER));
+ assertEquals(0, counted(result, DryRunReport.class, DryRunReport.CONFIG_PROBLEMS_COUNTER));
+ assertEquals(before, metadataLocation());
+ }
+
+ private String metadataLocation() {
+ return ((BaseTable) catalog.loadTable(tableId)).operations().current().metadataFileLocation();
+ }
+
+ private static Row report(Iterable rows) {
+ return Iterables.getOnlyElement(rows);
+ }
+
+ private static Collection schemas(Row report) {
+ return checkStateNotNull(report.getArray("schemas"));
+ }
+
+ @Test
+ public void testDryRunAgainstMissingTableReportsCreation() throws Exception {
+ String wide = writeWider("wide.parquet");
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(wide)).apply(addFiles(DRY_RUN));
+ PAssert.that(output.get(AddFiles.DRY_RUN_TAG))
+ .satisfies(
+ rows -> {
+ Row report = report(rows);
+ assertTrue(report.getBoolean("allowed"));
+ assertTrue(report.getBoolean("would_create_table"));
+ Row created = checkStateNotNull(report.getRow("created_table"));
+ assertThat(
+ created.getArray("columns").toString(),
+ containsString("create optional email string"));
+ assertThat(created.getString("schema"), containsString("\"name\":\"email\""));
+ for (Row schema : schemas(report)) {
+ assertTrue(
+ "the created table is described once", schema.getArray("changes").isEmpty());
+ }
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ assertFalse(catalog.tableExists(tableId));
+ }
+
+ /** A real run under FAIL_PIPELINE throws before creating the table, and the report says so. */
+ @Test
+ public void testDryRunAgainstMissingTableDoesNotCreateWhenARealRunWouldFail() throws Exception {
+ String wide = writeWider("wide.parquet");
+ String conflict = writeConflicting("conflict.parquet");
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(wide, conflict)).apply(addFiles(DRY_RUN));
+ PAssert.that(output.get(AddFiles.DRY_RUN_TAG))
+ .satisfies(
+ rows -> {
+ Row report = report(rows);
+ assertFalse(report.getBoolean("allowed"));
+ assertFalse(report.getBoolean("would_create_table"));
+ assertNotNull("the union still describes the table", report.getRow("created_table"));
+ assertThat(report.getString("reason"), containsString("would fail"));
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ assertFalse(catalog.tableExists(tableId));
+ }
+
+ /** A table-level change without a schema change is reported as such. */
+ @Test
+ public void testDryRunReportsNameMappingRepair() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String before = metadataLocation();
+ String covered = writeOneRecord("covered.parquet");
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(covered)).apply(addFiles(DRY_RUN));
+ PAssert.that(output.get(AddFiles.DRY_RUN_TAG))
+ .satisfies(
+ rows -> {
+ Row report = report(rows);
+ assertTrue(report.getBoolean("allowed"));
+ assertEquals(
+ Arrays.asList(DryRunReport.NAME_MAPPING_CHANGE),
+ new ArrayList<>(checkStateNotNull(report.getArray("table_changes"))));
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ assertEquals(before, metadataLocation());
+ }
+
+ /** Creation settings are part of the plan: a partition field the union lacks is reported. */
+ @Test
+ public void testDryRunReportsCreationBlockedByPartitionFields() throws Exception {
+ String wide = writeWider("wide.parquet");
+ AddFiles partitionedByMissing =
+ new AddFiles(
+ catalogConfig,
+ tableId.toString(),
+ null,
+ Arrays.asList("missing"),
+ null,
+ null,
+ null,
+ null,
+ DRY_RUN);
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(wide)).apply(partitionedByMissing);
+ PAssert.that(output.get(AddFiles.DRY_RUN_TAG))
+ .satisfies(
+ rows -> {
+ Row report = report(rows);
+ assertFalse(report.getBoolean("allowed"));
+ assertFalse(report.getBoolean("would_create_table"));
+ assertThat(report.getString("reason"), containsString("would fail to create"));
+ assertThat(report.getString("reason"), containsString("partition fields [missing]"));
+ assertEquals(1, checkStateNotNull(report.getArray("config_problems")).size());
+ return null;
+ });
+ PipelineResult result = pipeline.run();
+ result.waitUntilFinish();
+ assertEquals(1, counted(result, DryRunReport.class, DryRunReport.CONFIG_PROBLEMS_COUNTER));
+ assertFalse(catalog.tableExists(tableId));
+ }
+
+ /**
+ * The dry run reuses the real fold on scratch transactions: a schema compatible with the table
+ * but incompatible with another schema of the input is reported, as a real run would.
+ */
+ @Test
+ public void testDryRunSurfacesCrossSchemaConflicts() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ Schema emailInt =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "name", Types.StringType.get()),
+ Types.NestedField.required(3, "age", Types.IntegerType.get()),
+ Types.NestedField.optional(4, "email", Types.IntegerType.get()));
+ String a = writeWider("email_string.parquet");
+ Record intRecord = GenericRecord.create(emailInt);
+ intRecord.setField("id", 1);
+ intRecord.setField("name", "a");
+ intRecord.setField("age", 1);
+ intRecord.setField("email", 7);
+ String b = writeWithSchema("email_int.parquet", emailInt, intRecord);
+
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(a, b)).apply(addFiles(DRY_RUN));
+ PAssert.that(output.get(AddFiles.DRY_RUN_TAG))
+ .satisfies(
+ rows -> {
+ Collection schemas = schemas(report(rows));
+ assertEquals(2, schemas.size());
+ int allowed = 0;
+ for (Row schema : schemas) {
+ if (schema.getBoolean("allowed")) {
+ allowed++;
+ }
+ }
+ assertEquals("one of the two schemas loses the fold", 1, allowed);
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ assertNull(catalog.loadTable(tableId).schema().findField("email"));
+ }
+
+ /** With evolution on, the pre-pass is the only creator; registration never falls back to one. */
+ @Test
+ public void testMissingTableIsNotCreatedAtRegistrationWithEvolution() throws Exception {
+ File avro = temp.newFile("data.avro");
+ File garbage = temp.newFile("garbage.parquet");
+ java.nio.file.Files.write(garbage.toPath(), "not parquet".getBytes(StandardCharsets.UTF_8));
+ PCollectionRowTuple output =
+ pipeline
+ .apply("Create Input", Create.of(avro.getAbsolutePath(), garbage.getAbsolutePath()))
+ .apply(addFiles(ADDITIONS));
+ PAssert.that(output.get("snapshots")).empty();
+ PAssert.that(output.get("errors"))
+ .satisfies(
+ rows -> {
+ int count = 0;
+ for (Row row : rows) {
+ count++;
+ assertThat(
+ row.getString("error"),
+ containsString(AddFiles.ConvertToDataFile.MISSING_TABLE_ERROR));
+ }
+ assertEquals(2, count);
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ assertFalse(catalog.tableExists(tableId));
+ }
+
+ /** Nothing can create the table, so nothing is allowed, whatever ACCEPT would register. */
+ @Test
+ public void testDryRunAgainstMissingTableWithNoUsableSchema() throws Exception {
+ File avro = temp.newFile("data.avro");
+ SchemaEvolutionConfig config =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))
+ .setUnverifiableFileHandling(UnverifiableFileHandling.ACCEPT)
+ .setDryRun(true)
+ .build();
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(avro.getAbsolutePath())).apply(addFiles(config));
+ PAssert.that(output.get(AddFiles.DRY_RUN_TAG))
+ .satisfies(
+ rows -> {
+ Row report = report(rows);
+ assertFalse(report.getBoolean("allowed"));
+ assertFalse(report.getBoolean("would_create_table"));
+ assertFalse(report.getBoolean("unchecked_registered"));
+ assertEquals(Long.valueOf(1), report.getInt64("files_unchecked"));
+ assertThat(report.getString("reason"), containsString(DryRunReport.NO_TABLE_REASON));
+ return null;
+ });
+ pipeline.run().waitUntilFinish();
+ assertFalse(catalog.tableExists(tableId));
+ }
+
+ /** Files that contribute no schema are counted instead of vanishing. */
+ @Test
+ public void testDryRunReportsUnreadableAndNonParquetFiles() throws Exception {
+ dryRunWithUnreadableAndAvro(UnverifiableFileHandling.REJECT, false);
+ }
+
+ @Test
+ public void testDryRunReportsNonParquetFilesAsRegisteredWhenAccepted() throws Exception {
+ dryRunWithUnreadableAndAvro(UnverifiableFileHandling.ACCEPT, true);
+ }
+
+ private void dryRunWithUnreadableAndAvro(
+ UnverifiableFileHandling handling, boolean uncheckedRegistered) throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String good = writeOneRecord("good.parquet");
+ File garbage = temp.newFile("garbage.parquet");
+ java.nio.file.Files.write(garbage.toPath(), "not parquet".getBytes(StandardCharsets.UTF_8));
+ File avro = temp.newFile("data.avro");
+
+ SchemaEvolutionConfig config =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))
+ .setUnverifiableFileHandling(handling)
+ .setDryRun(true)
+ .build();
+ PCollectionRowTuple output =
+ pipeline
+ .apply(
+ "Create Input", Create.of(good, garbage.getAbsolutePath(), avro.getAbsolutePath()))
+ .apply(addFiles(config));
+ PAssert.that(output.get(AddFiles.DRY_RUN_TAG))
+ .satisfies(
+ rows -> {
+ Row report = report(rows);
+ assertTrue(report.getBoolean("allowed"));
+ assertEquals(Long.valueOf(1), report.getInt64("files_allowed"));
+ assertEquals(Long.valueOf(1), report.getInt64("files_unreadable"));
+ assertEquals(Long.valueOf(1), report.getInt64("files_unchecked"));
+ assertEquals(uncheckedRegistered, report.getBoolean("unchecked_registered"));
+ return null;
+ });
+ PipelineResult result = pipeline.run();
+ result.waitUntilFinish();
+ assertEquals(1, counted(result, DryRunReport.class, DryRunReport.FILES_ALLOWED_COUNTER));
+ assertEquals(0, counted(result, DryRunReport.class, DryRunReport.FILES_INCOMPATIBLE_COUNTER));
+ assertEquals(1, counted(result, DryRunReport.class, DryRunReport.FILES_UNREADABLE_COUNTER));
+ assertEquals(1, counted(result, DryRunReport.class, DryRunReport.FILES_UNCHECKED_COUNTER));
+ }
+
@Test
public void testEvolutionDisabledAddsNoPrePassTransforms() throws Exception {
catalog.createTable(tableId, icebergSchema);
@@ -1178,7 +1492,8 @@ public void testUpstreamWindowedBatchInputEvolvesAndRegisters() throws Exception
/**
* The schema commit retries a CommitFailedException (another writer got in first). The committer
- * is serialized with the DoFn, so only the table shows the retry happened.
+ * is serialized with the DoFn, so the retry shows in the table and in the time the backoff
+ * reports to the runner as throttled.
*/
@Test
public void testTransientSchemaCommitFailureIsRetried() throws Exception {
@@ -1199,9 +1514,25 @@ public void testTransientSchemaCommitFailureIsRetried() throws Exception {
.apply(addFiles(ADDITIONS).withSchemaCommitter(failsOnce));
PAssert.that(output.get("errors")).empty();
- pipeline.run().waitUntilFinish();
+ PipelineResult result = pipeline.run();
+ result.waitUntilFinish();
assertEmailAddedAndFilesRegistered(1);
+ long throttledMillis = 0;
+ for (MetricResult metric :
+ result
+ .metrics()
+ .queryMetrics(
+ MetricsFilter.builder()
+ .addNameFilter(
+ MetricNameFilter.named(
+ org.apache.beam.sdk.metrics.Metrics.THROTTLE_TIME_NAMESPACE,
+ org.apache.beam.sdk.metrics.Metrics.THROTTLE_TIME_COUNTER_NAME))
+ .build())
+ .getCounters()) {
+ throttledMillis += metric.getAttempted();
+ }
+ assertTrue("one backoff wait was reported: " + throttledMillis, throttledMillis > 0);
}
// ---- ConvertToDataFile coverage check and pinned columns
@@ -1209,6 +1540,12 @@ public void testTransientSchemaCommitFailureIsRetried() throws Exception {
private static final SchemaEvolutionConfig ADDITIONS =
SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION);
+ private static final SchemaEvolutionConfig DRY_RUN =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))
+ .setDryRun(true)
+ .build();
+
private PCollectionTuple convert(SchemaEvolutionConfig config, String... files) {
PCollectionTuple out =
pipeline
@@ -1590,13 +1927,17 @@ private static SchemaEvolutionConfig accepting(SchemaEvolutionConfig base) {
}
private static long counted(PipelineResult result, String counter) {
+ return counted(result, AddFiles.class, counter);
+ }
+
+ private static long counted(PipelineResult result, Class namespace, String counter) {
long total = 0;
for (MetricResult metric :
result
.metrics()
.queryMetrics(
MetricsFilter.builder()
- .addNameFilter(MetricNameFilter.named(AddFiles.class, counter))
+ .addNameFilter(MetricNameFilter.named(namespace, counter))
.build())
.getCounters()) {
total += metric.getAttempted();
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java
index e3d6889b1616..07a0adeb6cfb 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java
@@ -89,9 +89,10 @@ public void testCommitsTheWindowsSchemasAndEmitsTheSchemaId() {
new CommitSchemaOnce(
catalogConfig,
"default." + testName.getMethodName(),
- SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION),
- IncompatibleSchemaHandling.FAIL_PIPELINE,
- new CommitSchemaUnion.TableCreation(null, null, null))));
+ new CommitSchemaUnion.Settings(
+ SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION),
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ new CommitSchemaUnion.NewTableSettings(null, null, null)))));
PAssert.that(schemaIds).containsInAnyOrder(1L);
PipelineResult result = pipeline.run();
result.waitUntilFinish();
@@ -130,9 +131,10 @@ public void testNoOpWindowDoesNotCountACommit() {
new CommitSchemaOnce(
catalogConfig,
"default." + testName.getMethodName(),
- SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION),
- IncompatibleSchemaHandling.FAIL_PIPELINE,
- new CommitSchemaUnion.TableCreation(null, null, null))));
+ new CommitSchemaUnion.Settings(
+ SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION),
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ new CommitSchemaUnion.NewTableSettings(null, null, null)))));
PipelineResult result = pipeline.run();
result.waitUntilFinish();
assertEquals(0, commitsCounted(result));
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java
index 7df6af49920c..0c5f379a1105 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java
@@ -19,9 +19,12 @@
import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsString;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertThrows;
import static org.junit.Assert.assertTrue;
@@ -35,6 +38,7 @@
import org.apache.beam.sdk.io.iceberg.CommitSchemaUnion.Committer;
import org.apache.beam.sdk.io.iceberg.CommitSchemaUnion.IncompatibleSchemaException;
import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
+import org.apache.beam.sdk.util.FastNanoClockAndSleeper;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.BaseTable;
import org.apache.iceberg.Schema;
@@ -65,6 +69,9 @@ public class CommitSchemaUnionTest {
@Rule public TestName testName = new TestName();
+ /** Advances a fake clock instead of sleeping out the commit backoff. */
+ @Rule public FastNanoClockAndSleeper fastClock = new FastNanoClockAndSleeper();
+
private static final Schema TABLE =
new Schema(
required(1, "id", Types.LongType.get()),
@@ -77,8 +84,15 @@ public class CommitSchemaUnionTest {
private static final SchemaEvolutionConfig ADDITION_ONLY =
SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION);
- private static final CommitSchemaUnion.TableCreation NO_CREATION =
- new CommitSchemaUnion.TableCreation(null, null, null);
+ private static final CommitSchemaUnion.NewTableSettings NO_CREATION =
+ new CommitSchemaUnion.NewTableSettings(null, null, null);
+
+ private static CommitSchemaUnion.Settings settings(
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ CommitSchemaUnion.NewTableSettings newTable) {
+ return new CommitSchemaUnion.Settings(config, handling, newTable);
+ }
private HadoopCatalog catalog;
private TableIdentifier tableId;
@@ -111,9 +125,7 @@ private long commit(
catalog,
tableId,
Arrays.asList(schemas),
- config,
- handling,
- NO_CREATION,
+ settings(config, handling, NO_CREATION),
CommitSchemaUnion.DEFAULT_COMMITTER);
}
@@ -260,99 +272,6 @@ public void testFieldsInsideAddedContainersAreOptional() {
assertTrue(schema.findField("attrs.value.v").isOptional());
}
- // ---- newRequiredPaths (direct)
-
- @Test
- public void testNewRequiredPathsAtEveryLevelExceptMapKeys() {
- Schema before = new Schema(required(1, "id", Types.LongType.get()));
- Schema after =
- new Schema(
- required(1, "id", Types.LongType.get()),
- optional(
- 2,
- "s",
- Types.StructType.of(
- required(3, "a", Types.IntegerType.get()),
- optional(4, "b", Types.IntegerType.get()))),
- optional(
- 5,
- "items",
- Types.ListType.ofRequired(
- 6, Types.StructType.of(required(7, "qty", Types.IntegerType.get())))),
- optional(
- 8,
- "attrs",
- Types.MapType.ofRequired(
- 9,
- 10,
- Types.StructType.of(required(11, "k", Types.StringType.get())),
- Types.StructType.of(required(12, "v", Types.IntegerType.get())))));
- assertEquals(
- Arrays.asList("s.a", "items.element", "items.element.qty", "attrs.value", "attrs.value.v"),
- CommitSchemaUnion.newRequiredPaths(before, after));
- }
-
- /** Names containing element/key/value are not containers; regression for a substring check. */
- @Test
- public void testNewRequiredPathsContainerLikeNamesAreNotContainers() {
- Schema before = new Schema(required(1, "id", Types.LongType.get()));
- Schema after =
- new Schema(
- required(1, "id", Types.LongType.get()),
- optional(
- 2,
- "stats",
- Types.StructType.of(
- required(3, "keyword", Types.StringType.get()),
- required(4, "value_sum", Types.LongType.get()),
- required(5, "element", Types.StringType.get()))));
- assertEquals(
- Arrays.asList("stats.keyword", "stats.value_sum", "stats.element"),
- CommitSchemaUnion.newRequiredPaths(before, after));
- }
-
- @Test
- public void testNewRequiredPathsInNestedContainers() {
- Schema before = new Schema(required(1, "id", Types.LongType.get()));
- Schema after =
- new Schema(
- required(1, "id", Types.LongType.get()),
- optional(
- 2,
- "ll",
- Types.ListType.ofRequired(
- 3, Types.ListType.ofRequired(4, Types.IntegerType.get()))),
- optional(
- 5,
- "lm",
- Types.ListType.ofRequired(
- 6,
- Types.MapType.ofRequired(
- 7, 8, Types.StringType.get(), Types.IntegerType.get()))));
- assertEquals(
- Arrays.asList("ll.element", "ll.element.element", "lm.element", "lm.element.value"),
- CommitSchemaUnion.newRequiredPaths(before, after));
- }
-
- /** Growing an existing struct: only the field with a new id is a candidate. */
- @Test
- public void testNewRequiredPathsIgnoreExistingFields() {
- Schema before =
- new Schema(
- required(1, "id", Types.LongType.get()),
- optional(2, "s", Types.StructType.of(required(3, "old", Types.IntegerType.get()))));
- Schema after =
- new Schema(
- required(1, "id", Types.LongType.get()),
- optional(
- 2,
- "s",
- Types.StructType.of(
- required(3, "old", Types.IntegerType.get()),
- required(4, "fresh", Types.IntegerType.get()))));
- assertEquals(Arrays.asList("s.fresh"), CommitSchemaUnion.newRequiredPaths(before, after));
- }
-
/** A declared-optional column every file proved null-free does not relax the table. */
@Test
public void testNullFreeColumnIsNotRelaxed() {
@@ -550,9 +469,7 @@ public void testFinalSchemaIsOrderIndependent() {
catalog,
other,
Arrays.asList(files(b, 2), files(a, 1)),
- ALL,
- IncompatibleSchemaHandling.FAIL_PIPELINE,
- NO_CREATION,
+ settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION),
CommitSchemaUnion.DEFAULT_COMMITTER);
assertTrue(first.sameSchema(catalog.loadTable(other).schema()));
}
@@ -742,18 +659,21 @@ public void testCommitFailedOnceIsRetriedAgainstFreshState() {
catalog,
tableId,
Arrays.asList(files(file, 1)),
- ALL,
- IncompatibleSchemaHandling.FAIL_PIPELINE,
- NO_CREATION,
- flakyThenExternalChange);
+ settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION),
+ flakyThenExternalChange,
+ fastClock);
Table table = load();
assertEquals(2, attempts.get());
assertNotNull(table.schema().findField("external"));
assertNotNull(table.schema().findField("email"));
}
- @Test
- public void testPersistentCommitFailurePropagates() {
+ private long millisSleptSince(long startNanos) {
+ return (fastClock.nanoTime() - startNanos) / 1_000_000;
+ }
+
+ /** A commit that always loses; returns how many times it was attempted. */
+ private int attemptsUntilGivingUp() {
AtomicInteger attempts = new AtomicInteger();
Committer alwaysFails =
txn -> {
@@ -772,11 +692,54 @@ public void testPersistentCommitFailurePropagates() {
catalog,
tableId,
Arrays.asList(files(file, 1)),
- ALL,
- IncompatibleSchemaHandling.FAIL_PIPELINE,
- NO_CREATION,
- alwaysFails));
- assertEquals(CommitSchemaUnion.MAX_ATTEMPTS, attempts.get());
+ settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION),
+ alwaysFails,
+ fastClock));
+ return attempts.get();
+ }
+
+ /** The budget is time, not a count: the waits add up to the timeout, whatever the jitter. */
+ @Test
+ public void testPersistentCommitFailureGivesUpWhenTheRetryTimeIsSpent() {
+ long start = fastClock.nanoTime();
+ int attempts = attemptsUntilGivingUp();
+ assertEquals(
+ CommitSchemaUnion.DEFAULT_RETRY_TOTAL_TIMEOUT.getMillis(), millisSleptSince(start));
+ assertTrue("retried: " + attempts, attempts > 1);
+ assertNull("nothing was committed", load().schema().findField("email"));
+ }
+
+ @Test
+ public void testTableRetryPropertiesSetTheBudget() {
+ load()
+ .updateProperties()
+ .set(TableProperties.COMMIT_MIN_RETRY_WAIT_MS, "10")
+ .set(TableProperties.COMMIT_MAX_RETRY_WAIT_MS, "40")
+ .set(TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, "200")
+ .commit();
+ long start = fastClock.nanoTime();
+ attemptsUntilGivingUp();
+ assertEquals(200, millisSleptSince(start));
+ }
+
+ /** A bad table property must not replace the commit failure with a configuration error. */
+ @Test
+ public void testUnusableTableRetryPropertiesAreClamped() {
+ load()
+ .updateProperties()
+ .set(TableProperties.COMMIT_MIN_RETRY_WAIT_MS, "0")
+ .set(TableProperties.COMMIT_TOTAL_RETRY_TIME_MS, "-5")
+ .set(TableProperties.COMMIT_NUM_RETRIES, "-1")
+ .commit();
+ long start = fastClock.nanoTime();
+ assertEquals("no retries", 1, attemptsUntilGivingUp());
+ assertEquals(0, millisSleptSince(start));
+ }
+
+ @Test
+ public void testTableRetryCountCapsTheAttempts() {
+ load().updateProperties().set(TableProperties.COMMIT_NUM_RETRIES, "2").commit();
+ assertEquals(3, attemptsUntilGivingUp());
}
// ---- create path
@@ -789,15 +752,13 @@ private long commitTo(
TableIdentifier id,
SchemaEvolutionConfig config,
IncompatibleSchemaHandling handling,
- CommitSchemaUnion.TableCreation creation,
+ CommitSchemaUnion.NewTableSettings creation,
CollectDistinctSchemas.SchemaGroup... schemas) {
return CommitSchemaUnion.commit(
catalog,
id,
Arrays.asList(schemas),
- config,
- handling,
- creation,
+ settings(config, handling, creation),
CommitSchemaUnion.DEFAULT_COMMITTER);
}
@@ -812,8 +773,8 @@ public void testMissingTableIsCreatedFromTheUnion() {
Schema other =
new Schema(
required(1, "id", Types.LongType.get()), optional(2, "extra", Types.LongType.get()));
- CommitSchemaUnion.TableCreation creation =
- new CommitSchemaUnion.TableCreation(
+ CommitSchemaUnion.NewTableSettings creation =
+ new CommitSchemaUnion.NewTableSettings(
Arrays.asList("region"), null, java.util.Collections.singletonMap("k", "v"));
long schemaId =
commitTo(
@@ -848,8 +809,8 @@ public void testPartitionFieldFromNonSeedSchemaResolves() {
Schema other =
new Schema(
required(1, "id", Types.LongType.get()), optional(2, "extra", Types.StringType.get()));
- CommitSchemaUnion.TableCreation creation =
- new CommitSchemaUnion.TableCreation(Arrays.asList("extra"), null, null);
+ CommitSchemaUnion.NewTableSettings creation =
+ new CommitSchemaUnion.NewTableSettings(Arrays.asList("extra"), null, null);
commitTo(
id,
ALL,
@@ -965,82 +926,6 @@ public void testUnmatchedPinWarnsAndCreatesUnderRouteToErrors() {
new Schema(optional(1, "id", Types.LongType.get())), catalog.loadTable(id).schema());
}
- // ---- createdSchema (direct)
-
- @Test
- public void testCreatedSchemaPinsHoldAtEveryLevel() {
- Schema merged =
- new Schema(
- required(1, "id", Types.LongType.get()),
- optional(
- 2,
- "l",
- Types.ListType.ofOptional(
- 3, Types.StructType.of(optional(4, "q", Types.IntegerType.get())))));
- SchemaEvolutionConfig pinned =
- SchemaEvolutionConfig.builder()
- .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
- .setRequiredColumns(Collections.singleton("l.element.q"))
- .build();
- Schema created = CommitSchemaUnion.createdSchema(merged, pinned);
- assertSameSchema(
- new Schema(
- optional(1, "id", Types.LongType.get()),
- required(
- 2,
- "l",
- Types.ListType.ofRequired(
- 3, Types.StructType.of(required(4, "q", Types.IntegerType.get()))))),
- created);
- }
-
- @Test
- public void testCreatedSchemaEveryLevelOptionalExceptMapKeys() {
- Schema schema =
- new Schema(
- required(1, "id", Types.LongType.get()),
- required(
- 2,
- "s",
- Types.StructType.of(
- required(3, "a", Types.IntegerType.get()),
- required(
- 4,
- "items",
- Types.ListType.ofRequired(
- 5, Types.StructType.of(required(6, "qty", Types.IntegerType.get())))))),
- required(
- 7,
- "attrs",
- Types.MapType.ofRequired(
- 8,
- 9,
- Types.StructType.of(required(10, "k", Types.StringType.get())),
- Types.StructType.of(required(11, "v", Types.IntegerType.get())))));
- assertSameSchema(
- new Schema(
- optional(1, "id", Types.LongType.get()),
- optional(
- 2,
- "s",
- Types.StructType.of(
- optional(3, "a", Types.IntegerType.get()),
- optional(
- 4,
- "items",
- Types.ListType.ofOptional(
- 5, Types.StructType.of(optional(6, "qty", Types.IntegerType.get())))))),
- optional(
- 7,
- "attrs",
- Types.MapType.ofOptional(
- 8,
- 9,
- Types.StructType.of(required(10, "k", Types.StringType.get())),
- Types.StructType.of(optional(11, "v", Types.IntegerType.get()))))),
- CommitSchemaUnion.createdSchema(schema, ALL));
- }
-
/** Options guard an existing table's schema; with no table there is nothing to guard. */
@Test
public void testCreationIsNotGatedOnAnyParticularOption() {
@@ -1068,9 +953,7 @@ public void testMissingTableWithoutSchemasIsNotCreated() {
catalog,
id,
new ArrayList<>(),
- ALL,
- IncompatibleSchemaHandling.FAIL_PIPELINE,
- NO_CREATION,
+ settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION),
CommitSchemaUnion.DEFAULT_COMMITTER);
assertEquals(CommitSchemaUnion.NO_TABLE, result);
assertFalse(catalog.tableExists(id));
@@ -1098,6 +981,58 @@ public void testConflictOnCreateFailsWithoutCreating() {
assertFalse(catalog.tableExists(id));
}
+ /** The fallback creation at registration throws the same error, so no handling can route it. */
+ @Test
+ public void testPartitionFieldAbsentFromUnionFailsCreationUnderEitherHandling() {
+ TableIdentifier id = missing();
+ Schema seed =
+ new Schema(
+ required(1, "id", Types.LongType.get()), optional(2, "region", Types.StringType.get()));
+ CommitSchemaUnion.NewTableSettings creation =
+ new CommitSchemaUnion.NewTableSettings(Arrays.asList("missing"), null, null);
+ for (IncompatibleSchemaHandling handling : IncompatibleSchemaHandling.values()) {
+ IllegalStateException e =
+ assertThrows(
+ IllegalStateException.class,
+ () -> commitTo(id, ALL, handling, creation, files(seed, 2)));
+ assertEquals(
+ "not an IncompatibleSchemaException, which the handling could downgrade",
+ IllegalStateException.class,
+ e.getClass());
+ assertThat(
+ e.getMessage(), containsString("cannot be created with partition fields [missing]"));
+ }
+ assertFalse(catalog.tableExists(id));
+ }
+
+ @Test
+ public void testPartitionFieldOnlyInASkippedSchemaFailsCreation() {
+ TableIdentifier id = missing();
+ Schema seed =
+ new Schema(
+ required(1, "id", Types.LongType.get()), optional(2, "code", Types.StringType.get()));
+ Schema loser =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(2, "code", Types.LongType.get()),
+ optional(3, "region", Types.StringType.get()));
+ CommitSchemaUnion.NewTableSettings creation =
+ new CommitSchemaUnion.NewTableSettings(Arrays.asList("region"), null, null);
+ IllegalStateException e =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ commitTo(
+ id,
+ ALL,
+ IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ creation,
+ files(seed, 3),
+ files(loser, 1)));
+ assertThat(e.getMessage(), containsString("region"));
+ assertFalse(catalog.tableExists(id));
+ }
+
@Test
public void testConflictOnCreateRoutesLoserAndCreates() {
TableIdentifier id = missing();
@@ -1142,10 +1077,9 @@ public void testCreateRaceFallsBackToEvolvingTheExistingTable() {
catalog,
id,
Arrays.asList(files(file, 1)),
- ALL,
- IncompatibleSchemaHandling.FAIL_PIPELINE,
- NO_CREATION,
- raced);
+ settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION),
+ raced,
+ fastClock);
Table table = catalog.loadTable(id);
assertEquals(2, attempts.get());
assertNotNull(table.schema().findField("email"));
@@ -1156,16 +1090,13 @@ public void testCreateRaceFallsBackToEvolvingTheExistingTable() {
@Test
public void testEmptyInputCommitsNothing() {
- seedNameMapping();
String before = metadataLocation(load());
List none = new ArrayList<>();
CommitSchemaUnion.commit(
catalog,
tableId,
none,
- ALL,
- IncompatibleSchemaHandling.FAIL_PIPELINE,
- NO_CREATION,
+ settings(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION),
CommitSchemaUnion.DEFAULT_COMMITTER);
assertEquals(before, metadataLocation(load()));
}
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java
index 5cac095201db..4f17bc738400 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java
@@ -253,7 +253,7 @@ private static long counter(PipelineResult result, String name) {
private PCollection run(String... paths) {
return pipeline
.apply(Create.of(Arrays.asList(paths)))
- .apply(ParDo.of(new ReadFooterSchema()))
+ .apply(ParDo.of(new ReadFooterSchema(SchemaEvolutionConfig.disabled())))
.setCoder(CollectDistinctSchemas.groupCoder());
}
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java
index c6d565cda957..c76c06d9b7c7 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionConfigTest.java
@@ -116,4 +116,14 @@ public void testIncompatibleSchemaHandlingDefaultsByMode() {
IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
forced.incompatibleSchemaHandlingFor(PCollection.IsBounded.BOUNDED));
}
+
+ /** A dry run of nothing is a config error, not a silent real run. */
+ @Test
+ public void testDryRunAloneIsRejectedAtBuild() {
+ IllegalArgumentException e =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> SchemaEvolutionConfig.builder().setDryRun(true).build());
+ assertTrue(e.getMessage(), e.getMessage().contains("dry run"));
+ }
}
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaPlanTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaPlanTest.java
new file mode 100644
index 000000000000..12d34d773b84
--- /dev/null
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaPlanTest.java
@@ -0,0 +1,325 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.iceberg;
+
+import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.junit.Before;
+import org.junit.ClassRule;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.rules.TestName;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Planning without committing; what a plan does once committed is CommitSchemaUnionTest's. */
+@RunWith(JUnit4.class)
+public class SchemaPlanTest {
+ @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
+
+ @Rule
+ public transient TestDataWarehouse warehouse = new TestDataWarehouse(TEMPORARY_FOLDER, "default");
+
+ @Rule public TestName testName = new TestName();
+
+ private static final Schema TABLE =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(2, "name", Types.StringType.get()),
+ optional(3, "score", Types.FloatType.get()),
+ required(4, "region", Types.StringType.get()));
+
+ private static final SchemaEvolutionConfig ALL =
+ SchemaEvolutionConfig.of(SchemaEvolutionOption.values());
+
+ private static final CommitSchemaUnion.Settings SETTINGS =
+ new CommitSchemaUnion.Settings(
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ new CommitSchemaUnion.NewTableSettings(null, null, null));
+
+ private HadoopCatalog catalog;
+ private TableIdentifier tableId;
+
+ @Before
+ public void setUp() {
+ catalog = new HadoopCatalog(new Configuration(), warehouse.location);
+ tableId = TableIdentifier.of("default", testName.getMethodName());
+ warehouse.createTable(tableId, TABLE);
+ }
+
+ private static String json(Schema schema) {
+ return SchemaParser.toJson(FileSchemas.canonical(schema));
+ }
+
+ private static CollectDistinctSchemas.SchemaGroup files(Schema schema, long count) {
+ return CollectDistinctSchemas.SchemaGroup.of(json(schema), count, Collections.emptyList());
+ }
+
+ private Table load() {
+ return catalog.loadTable(tableId);
+ }
+
+ /** Full-schema comparison, field ids normalized; string form so a failure shows the diff. */
+ private static void assertSameSchema(Schema expected, Schema actual) {
+ assertEquals(
+ TypeUtil.assignIncreasingFreshIds(expected).asStruct().toString(),
+ TypeUtil.assignIncreasingFreshIds(actual).asStruct().toString());
+ }
+
+ // ---- plans
+
+ /** The mapping repair is a plan decision, made before anything is committed. */
+ @Test
+ public void testPlanReportsTheNameMappingRepair() {
+ Table table = load();
+ assertFalse(table.properties().containsKey(TableProperties.DEFAULT_NAME_MAPPING));
+ List covered = Arrays.asList(files(TABLE, 1));
+ SchemaPlan.Evolution plan =
+ (SchemaPlan.Evolution) SchemaPlan.compute(catalog, tableId, table, covered, SETTINGS);
+ assertNull(plan.newSchema);
+ assertTrue(plan.repairsNameMapping);
+
+ table
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING, NameMappingUtils.regenerate(table.schema(), null))
+ .commit();
+ plan = (SchemaPlan.Evolution) SchemaPlan.compute(catalog, tableId, load(), covered, SETTINGS);
+ assertFalse(plan.repairsNameMapping);
+ }
+
+ @Test
+ public void testPlanForCreationListsAcceptedSchemasAndCreationSettings() {
+ TableIdentifier missing = TableIdentifier.of("default", testName.getMethodName() + "_new");
+ Schema seed =
+ new Schema(
+ required(1, "id", Types.LongType.get()), optional(2, "region", Types.StringType.get()));
+ Schema other =
+ new Schema(
+ required(1, "id", Types.LongType.get()), optional(2, "extra", Types.LongType.get()));
+ CommitSchemaUnion.NewTableSettings creation =
+ new CommitSchemaUnion.NewTableSettings(Arrays.asList("region"), Arrays.asList("id"), null);
+ SchemaPlan.Creation plan =
+ (SchemaPlan.Creation)
+ SchemaPlan.compute(
+ catalog,
+ missing,
+ null,
+ Arrays.asList(files(seed, 5), files(other, 1)),
+ new CommitSchemaUnion.Settings(
+ ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, creation));
+ assertTrue(plan.canCreate());
+ assertEquals(2, plan.schemasToMerge.size());
+ assertEquals(5, checkStateNotNull(plan.toMerge(json(seed))).files);
+ assertEquals(1, checkStateNotNull(plan.toMerge(json(other))).files);
+ assertEquals("region", checkStateNotNull(plan.spec).fields().get(0).name());
+ assertEquals(1, checkStateNotNull(plan.sortOrder).fields().size());
+ assertFalse(catalog.tableExists(missing));
+ }
+
+ // ---- newRequiredPaths (direct)
+
+ @Test
+ public void testNewRequiredPathsAtEveryLevelExceptMapKeys() {
+ Schema before = new Schema(required(1, "id", Types.LongType.get()));
+ Schema after =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "s",
+ Types.StructType.of(
+ required(3, "a", Types.IntegerType.get()),
+ optional(4, "b", Types.IntegerType.get()))),
+ optional(
+ 5,
+ "items",
+ Types.ListType.ofRequired(
+ 6, Types.StructType.of(required(7, "qty", Types.IntegerType.get())))),
+ optional(
+ 8,
+ "attrs",
+ Types.MapType.ofRequired(
+ 9,
+ 10,
+ Types.StructType.of(required(11, "k", Types.StringType.get())),
+ Types.StructType.of(required(12, "v", Types.IntegerType.get())))));
+ assertEquals(
+ Arrays.asList("s.a", "items.element", "items.element.qty", "attrs.value", "attrs.value.v"),
+ SchemaPlan.newRequiredPaths(before, after));
+ }
+
+ /** Names containing element/key/value are not containers; regression for a substring check. */
+ @Test
+ public void testNewRequiredPathsContainerLikeNamesAreNotContainers() {
+ Schema before = new Schema(required(1, "id", Types.LongType.get()));
+ Schema after =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "stats",
+ Types.StructType.of(
+ required(3, "keyword", Types.StringType.get()),
+ required(4, "value_sum", Types.LongType.get()),
+ required(5, "element", Types.StringType.get()))));
+ assertEquals(
+ Arrays.asList("stats.keyword", "stats.value_sum", "stats.element"),
+ SchemaPlan.newRequiredPaths(before, after));
+ }
+
+ @Test
+ public void testNewRequiredPathsInNestedContainers() {
+ Schema before = new Schema(required(1, "id", Types.LongType.get()));
+ Schema after =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "ll",
+ Types.ListType.ofRequired(
+ 3, Types.ListType.ofRequired(4, Types.IntegerType.get()))),
+ optional(
+ 5,
+ "lm",
+ Types.ListType.ofRequired(
+ 6,
+ Types.MapType.ofRequired(
+ 7, 8, Types.StringType.get(), Types.IntegerType.get()))));
+ assertEquals(
+ Arrays.asList("ll.element", "ll.element.element", "lm.element", "lm.element.value"),
+ SchemaPlan.newRequiredPaths(before, after));
+ }
+
+ /** Growing an existing struct: only the field with a new id is a candidate. */
+ @Test
+ public void testNewRequiredPathsIgnoreExistingFields() {
+ Schema before =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(2, "s", Types.StructType.of(required(3, "old", Types.IntegerType.get()))));
+ Schema after =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "s",
+ Types.StructType.of(
+ required(3, "old", Types.IntegerType.get()),
+ required(4, "fresh", Types.IntegerType.get()))));
+ assertEquals(Arrays.asList("s.fresh"), SchemaPlan.newRequiredPaths(before, after));
+ }
+
+ // ---- createdSchema (direct)
+
+ @Test
+ public void testCreatedSchemaPinsHoldAtEveryLevel() {
+ Schema merged =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "l",
+ Types.ListType.ofOptional(
+ 3, Types.StructType.of(optional(4, "q", Types.IntegerType.get())))));
+ SchemaEvolutionConfig pinned =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setRequiredColumns(Collections.singleton("l.element.q"))
+ .build();
+ Schema created = SchemaPlan.createdSchema(merged, pinned);
+ assertSameSchema(
+ new Schema(
+ optional(1, "id", Types.LongType.get()),
+ required(
+ 2,
+ "l",
+ Types.ListType.ofRequired(
+ 3, Types.StructType.of(required(4, "q", Types.IntegerType.get()))))),
+ created);
+ }
+
+ @Test
+ public void testCreatedSchemaEveryLevelOptionalExceptMapKeys() {
+ Schema schema =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(
+ 2,
+ "s",
+ Types.StructType.of(
+ required(3, "a", Types.IntegerType.get()),
+ required(
+ 4,
+ "items",
+ Types.ListType.ofRequired(
+ 5, Types.StructType.of(required(6, "qty", Types.IntegerType.get())))))),
+ required(
+ 7,
+ "attrs",
+ Types.MapType.ofRequired(
+ 8,
+ 9,
+ Types.StructType.of(required(10, "k", Types.StringType.get())),
+ Types.StructType.of(required(11, "v", Types.IntegerType.get())))));
+ assertSameSchema(
+ new Schema(
+ optional(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "s",
+ Types.StructType.of(
+ optional(3, "a", Types.IntegerType.get()),
+ optional(
+ 4,
+ "items",
+ Types.ListType.ofOptional(
+ 5, Types.StructType.of(optional(6, "qty", Types.IntegerType.get())))))),
+ optional(
+ 7,
+ "attrs",
+ Types.MapType.ofOptional(
+ 8,
+ 9,
+ Types.StructType.of(required(10, "k", Types.StringType.get())),
+ Types.StructType.of(optional(11, "v", Types.IntegerType.get()))))),
+ SchemaPlan.createdSchema(schema, ALL));
+ }
+}
diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml
index c41dd6f2e07f..2d2c2af6da13 100644
--- a/sdks/python/apache_beam/yaml/standard_io.yaml
+++ b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -661,6 +661,7 @@
required_columns: 'required_columns'
incompatible_schema_handling: 'incompatible_schema_handling'
unverifiable_file_handling: 'unverifiable_file_handling'
+ dry_run: 'dry_run'
error_handling: 'error_handling'
underlying_provider:
type: beamJar
diff --git a/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dry_run.yaml b/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dry_run.yaml
new file mode 100644
index 000000000000..75b7b1a16ddc
--- /dev/null
+++ b/sdks/python/apache_beam/yaml/tests/iceberg_add_files_dry_run.yaml
@@ -0,0 +1,185 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+fixtures:
+ - name: TEMP_DIR
+ type: "tempfile.TemporaryDirectory"
+
+
+pipelines:
+ # Pipeline 1: files with two columns
+ - pipeline:
+ type: chain
+ transforms:
+ - type: Create
+ config:
+ elements:
+ - {label: "11a", rank: 0}
+ - {label: "37a", rank: 1}
+ - type: WriteToParquet
+ config:
+ path: "{TEMP_DIR}/data/narrow"
+ file_name_suffix: ".parquet"
+ num_shards: 1
+
+ # Pipeline 2: files with an extra column
+ - pipeline:
+ type: chain
+ transforms:
+ - type: Create
+ config:
+ elements:
+ - {label: "389a", rank: 2, bool: false}
+ - {label: "3821b", rank: 3, bool: true}
+ - type: WriteToParquet
+ config:
+ path: "{TEMP_DIR}/data/wide"
+ file_name_suffix: ".parquet"
+ num_shards: 1
+
+ # Pipeline 3: a dry run against a table that does not exist yet; the report is written as JSON
+ # from the dry_run_report output, the way a user keeps it
+ - pipeline:
+ transforms:
+ - type: ReadMatchFiles
+ name: Match
+ config:
+ file_pattern: "{TEMP_DIR}/data/*.parquet"
+ - type: MapToFields
+ name: Paths
+ input: Match
+ config:
+ language: python
+ fields:
+ path:
+ callable: "lambda row: row.path"
+ output_type: string
+ - type: IcebergAddFiles
+ name: DryRun
+ input: Paths
+ config:
+ table: "default.table"
+ catalog_properties:
+ type: "hadoop"
+ warehouse: "{TEMP_DIR}/dir"
+ schema_evolution_options: [ALLOW_FIELD_ADDITION]
+ dry_run: true
+ - type: WriteToJson
+ name: WriteReport
+ input: DryRun.dry_run_report
+ config:
+ path: "{TEMP_DIR}/report.json"
+ # The full report, as a real run against this input would decide it. The report is one
+ # row with nested fields; the two assertions below read it the way a user would: the
+ # verdict, the file counts and the table a real run would create (from the union, every
+ # column optional) come from the top level, the per-schema entries from an Explode.
+ - type: MapToFields
+ name: Verdict
+ input: DryRun.dry_run_report
+ config:
+ language: python
+ fields:
+ allowed: allowed
+ reason: reason
+ would_create_table: would_create_table
+ files_allowed: files_allowed
+ files_incompatible: files_incompatible
+ files_unreadable: files_unreadable
+ files_unchecked: files_unchecked
+ unchecked_registered: unchecked_registered
+ table_changes: table_changes
+ config_problems: config_problems
+ created_schema: created_table.schema
+ created_columns: created_table.columns
+ - type: AssertEqual
+ name: AssertVerdict
+ input: Verdict
+ config:
+ elements:
+ - allowed: true
+ reason: ""
+ would_create_table: true
+ files_allowed: 2
+ files_incompatible: 0
+ files_unreadable: 0
+ files_unchecked: 0
+ unchecked_registered: false
+ table_changes: []
+ config_problems: []
+ created_schema: '{{"type":"struct","schema-id":0,"fields":[{{"id":1,"name":"bool","required":false,"type":"boolean"}},{{"id":2,"name":"label","required":false,"type":"string"}},{{"id":3,"name":"rank","required":false,"type":"long"}}]}}'
+ created_columns: ["create optional bool boolean", "create optional label string", "create optional rank long"]
+ # one entry per distinct schema (canonical JSON), most common first
+ - type: Explode
+ name: Schemas
+ input: DryRun.dry_run_report
+ config:
+ fields: [schemas]
+ - type: MapToFields
+ name: SchemaEntries
+ input: Schemas
+ config:
+ language: python
+ fields:
+ schema_key: schemas.schema_key
+ schema: schemas.schema
+ num_files: schemas.num_files
+ changes: schemas.changes
+ allowed: schemas.allowed
+ reason: schemas.reason
+ - type: AssertEqual
+ name: AssertSchemas
+ input: SchemaEntries
+ config:
+ elements:
+ - schema_key: "s27e62d"
+ schema: '{{"type":"struct","schema-id":0,"fields":[{{"id":1,"name":"bool","required":true,"type":"boolean"}},{{"id":2,"name":"label","required":true,"type":"string"}},{{"id":3,"name":"rank","required":true,"type":"long"}}]}}'
+ num_files: 1
+ changes: []
+ allowed: true
+ reason: ""
+ - schema_key: "s48334d"
+ schema: '{{"type":"struct","schema-id":0,"fields":[{{"id":1,"name":"label","required":true,"type":"string"}},{{"id":2,"name":"rank","required":true,"type":"long"}}]}}'
+ num_files: 1
+ changes: []
+ allowed: true
+ reason: ""
+
+ providers:
+ - type: python
+ config: { }
+ transforms:
+ ReadMatchFiles: 'apache_beam.io.fileio.MatchFiles'
+
+ # Pipeline 4: the report was written (nested fields come back as strings through ReadFromJson,
+ # so only the scalar columns are compared here; the full row is asserted above)
+ - pipeline:
+ type: chain
+ transforms:
+ - type: ReadFromJson
+ config:
+ path: "{TEMP_DIR}/report.json*"
+ - type: MapToFields
+ config:
+ fields:
+ allowed: allowed
+ would_create_table: would_create_table
+ files_allowed: files_allowed
+ files_incompatible: files_incompatible
+ - type: AssertEqual
+ config:
+ elements:
+ - {allowed: true, would_create_table: true, files_allowed: 2, files_incompatible: 0}