From 82c8f711159b56ce516386e5728718df9881c6d5 Mon Sep 17 00:00:00 2001
From: claudevdm <33973061+claudevdm@users.noreply.github.com>
Date: Thu, 17 Sep 2026 13:42:21 -0400
Subject: [PATCH] Addfiles wiring provider (#40150)
* AddFiles: wire the schema pre-pass behind a Wait.on gate (batch)
AddFiles gets a ninth constructor argument, SchemaEvolutionConfig
With options set, the topology becomes
paths -> ReadFooterSchema -> CollectDistinctSchemas -> CommitSchemaOnce
|
paths -> Wait.on(signal) -> ConvertToDataFile -> ... (unchanged)
so no path reaches ConvertToDataFile until the schema commit has
landed.
* IcebergAddFiles provider: schema_evolution_options, required_columns, incompatible_schema_handling, unverifiable_file_handling
Exposes the pre-pass through the iceberg_add_files SchemaTransform and
Beam YAML:
- type: WriteToIceberg (transform: IcebergAddFiles in YAML)
config:
table: db.sales
schema_evolution_options: [ALLOW_FIELD_ADDITION, ALLOW_TYPE_PROMOTION]
required_columns: [id]
incompatible_schema_handling: ROUTE_TO_ERRORS
* trigger tests
* simplify arg
* tests
* fixes
* changes
* comments, refresh only once per bundle
---
.../IO_Iceberg_Integration_Tests.json | 2 +-
CHANGES.md | 1 +
.../apache/beam/sdk/io/iceberg/AddFiles.java | 146 ++++++++++-
.../AddFilesSchemaTransformProvider.java | 129 +++++++++-
.../sdk/io/iceberg/SchemaEvolutionConfig.java | 6 +-
.../sdk/io/iceberg/SchemaEvolutionOption.java | 5 +-
.../beam/sdk/io/iceberg/AddFilesIT.java | 114 ++++++++-
.../AddFilesSchemaTransformProviderTest.java | 171 +++++++++++++
.../beam/sdk/io/iceberg/AddFilesTest.java | 240 ++++++++++++++++++
.../io/iceberg/SchemaEvolutionConfigTest.java | 11 +-
sdks/python/apache_beam/yaml/standard_io.yaml | 4 +
.../tests/iceberg_add_files_evolution.yaml | 98 +++++++
12 files changed, 906 insertions(+), 21 deletions(-)
create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
create mode 100644 sdks/python/apache_beam/yaml/tests/iceberg_add_files_evolution.yaml
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
index 89e73b29da05..e1430c74e62c 100644
--- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json
+++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
@@ -1,4 +1,4 @@
{
"comment": "Modify this file in a trivial way to cause this test suite to run.",
- "modification": 6
+ "modification": 7
}
diff --git a/CHANGES.md b/CHANGES.md
index 18351eebd1ad..2f6866e692b8 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -69,6 +69,7 @@
* BigQueryIO now supports reading BigQuery Lakehouse runtime catalog (BigLake metastore) Iceberg tables with the Storage Read API, using 4-part `project.catalog.namespace.table` identifiers (or a `TableReference` with a composite `catalog.namespace` dataset id). Previously such references were silently mis-parsed (Java) ([#39597](https://github.com/apache/beam/issues/39597)) .
* SolaceIO now supports reading and writing binary and text content data payload (Java) ([#39875](https://github.com/apache/beam/issues/39875)).
* ClickHouseIO: support writing `Decimal(P, S)` / `Decimal32/64/128/256` columns (Java) ([#39840](https://github.com/apache/beam/issues/39840)).
+* [IcebergIO] AddFiles (`IcebergAddFiles` in YAML) can evolve the table schema before registering files, with `schema_evolution_options`, `required_columns`, `incompatible_schema_handling` and `unverifiable_file_handling` (Java/YAML, batch only) ([#40144](https://github.com/apache/beam/issues/40144)).
## New Features / Improvements
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java
index daf61bf883ff..75b1a0144e22 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java
@@ -38,11 +38,13 @@
import java.util.UUID;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.beam.sdk.coders.KvCoder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
+import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.UnverifiableFileHandling;
import org.apache.beam.sdk.metrics.Counter;
import org.apache.beam.sdk.schemas.Schema;
@@ -51,13 +53,17 @@
import org.apache.beam.sdk.state.StateSpec;
import org.apache.beam.sdk.state.StateSpecs;
import org.apache.beam.sdk.state.ValueState;
+import org.apache.beam.sdk.transforms.Combine;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.transforms.GroupIntoBatches;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.Wait;
import org.apache.beam.sdk.transforms.WithKeys;
import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.GlobalWindows;
import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.values.KV;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionRowTuple;
@@ -112,8 +118,38 @@
import org.slf4j.LoggerFactory;
/**
- * A transform that takes in a stream of file paths, converts them to Iceberg {@link DataFile}s with
- * partition metadata and metrics, then commits them to an Iceberg {@link Table}.
+ * Registers existing Parquet, ORC or Avro files in an Iceberg table without rewriting them: each
+ * path becomes a {@link DataFile} with partition metadata and column stats, batched into manifests
+ * and committed as snapshots.
+ *
+ * Outputs: {@code snapshots} (one row per commit), {@code errors} (one row per file that could
+ * not be registered: {@code file}, {@code error}).
+ *
+ *
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
+ * table (add a column, relax a required column, promote a type), commits the allowed changes in one
+ * transaction, and only then registers the files. Manifest entries are immutable, so this ordering
+ * is what guarantees that every registered file carries stats for every column it has. Files whose
+ * schema needs a change that is not allowed, or that conflicts with the table or with another file,
+ * are incompatible: by default the pipeline fails before committing anything, or routes them to
+ * {@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
+ * currently requires bounded input; unbounded input with options set is rejected at construction.
+ *
+ *
{@code
+ * SchemaEvolutionConfig evolution =
+ * SchemaEvolutionConfig.builder()
+ * .setOptions(EnumSet.of(ALLOW_FIELD_ADDITION, ALLOW_TYPE_PROMOTION))
+ * .setRequiredColumns(Collections.singleton("id"))
+ * .build();
+ * paths.apply(new AddFiles(catalog, "db.sales", null, null, null, null, null, null, evolution));
+ * }
+ *
+ * Without options the table schema is never changed and files register as-is: columns the table
+ * does not have get no stats and are not readable, and a nested column the table does not know can
+ * make that file, and any scan that includes it, fail in an Iceberg reader.
*/
public class AddFiles extends PTransform, PCollectionRowTuple> {
static final String OUTPUT_TAG = "snapshots";
@@ -142,6 +178,8 @@ public class AddFiles extends PTransform, PCollectionRowTupl
private final @Nullable List partitionFields;
private final @Nullable List sortFields;
private final @Nullable Map tableProps;
+ private final SchemaEvolutionConfig evolution;
+ private CommitSchemaUnion.Committer committer = CommitSchemaUnion.DEFAULT_COMMITTER;
public AddFiles(
IcebergCatalogConfig catalogConfig,
@@ -152,6 +190,40 @@ public AddFiles(
@Nullable Map tableProps,
@Nullable Integer manifestFileSize,
@Nullable Duration intervalTrigger) {
+ this(
+ catalogConfig,
+ tableIdentifier,
+ locationPrefix,
+ partitionFields,
+ sortFields,
+ tableProps,
+ manifestFileSize,
+ intervalTrigger,
+ null);
+ }
+
+ /**
+ * @param locationPrefix when set on a partitioned table, the partition is read from the path
+ * after this prefix instead of from the file's column stats
+ * @param partitionFields partition spec applied when the table is created by this transform
+ * @param sortFields sort order applied when the table is created by this transform
+ * @param tableProps table properties applied when the table is created by this transform
+ * @param manifestFileSize data files per manifest
+ * @param intervalTrigger streaming only: how often manifests are committed
+ * @param evolution schema evolution settings; null or no options means the schema is never
+ * changed
+ */
+ public AddFiles(
+ IcebergCatalogConfig catalogConfig,
+ String tableIdentifier,
+ @Nullable String locationPrefix,
+ @Nullable List partitionFields,
+ @Nullable List sortFields,
+ @Nullable Map tableProps,
+ @Nullable Integer manifestFileSize,
+ @Nullable Duration intervalTrigger,
+ @Nullable SchemaEvolutionConfig evolution) {
+ this.evolution = evolution != null ? evolution : SchemaEvolutionConfig.disabled();
this.catalogConfig = catalogConfig;
this.tableIdentifier = tableIdentifier;
this.partitionFields = partitionFields;
@@ -166,6 +238,10 @@ public AddFiles(
@Override
public PCollectionRowTuple expand(PCollection input) {
if (input.isBounded().equals(UNBOUNDED)) {
+ Preconditions.checkArgument(
+ !evolution.isEnabled(),
+ "Schema evolution is not yet supported for unbounded input: run a batch pipeline or"
+ + " remove the schema evolution options.");
intervalTrigger = intervalTrigger != null ? intervalTrigger : DEFAULT_TRIGGER_INTERVAL;
LOG.info(
"AddFiles configured to generate a new manifest after accumulating {} files, or after {} seconds.",
@@ -182,8 +258,13 @@ public PCollectionRowTuple expand(PCollection input) {
"AddFiles configured to build partition metadata after the prefix: '{}'", locationPrefix);
}
+ PCollection paths = input;
+ if (evolution.isEnabled()) {
+ paths = gateOnSchemaCommit(input);
+ }
+
PCollectionTuple dataFiles =
- input.apply(
+ paths.apply(
"ConvertToDataFiles",
ParDo.of(
new ConvertToDataFile(
@@ -192,7 +273,8 @@ public PCollectionRowTuple expand(PCollection input) {
locationPrefix,
partitionFields,
sortFields,
- tableProps))
+ tableProps,
+ evolution))
.withOutputTags(DATA_FILES, TupleTagList.of(ERRORS)));
SchemaCoder sdfCoder;
try {
@@ -238,6 +320,47 @@ 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())
+ .apply(
+ "CommitSchemaOnce",
+ ParDo.of(
+ new CommitSchemaOnce(
+ catalogConfig,
+ tableIdentifier,
+ evolution,
+ onIncompatible,
+ creation,
+ committer)));
+ return windowed.apply("WaitForSchemaCommit", Wait.on(signal));
+ }
+
+ /** Test hook: how the schema pre-pass commits its transaction. */
+ AddFiles withSchemaCommitter(CommitSchemaUnion.Committer committer) {
+ this.committer = committer;
+ return this;
+ }
+
/**
* Reads incoming file paths, extracts Iceberg metadata, and converts them into {@link
* SerializableDataFile} objects.
@@ -272,6 +395,7 @@ static class ConvertToDataFile extends DoFn {
private transient @MonotonicNonNull BoundedAsyncTasks tasks;
private transient volatile @MonotonicNonNull Table table;
private transient @MonotonicNonNull Set warned;
+ private final AtomicBoolean refreshedThisBundle = new AtomicBoolean();
// Number of parallel threads processing incoming files
private static final int THREAD_POOL_SIZE = 10;
@@ -392,8 +516,8 @@ public void setup() {
/** Clears anything left behind if the runner reuses this instance after a failed bundle. */
@StartBundle
public void startBundle() {
-
checkStateNotNull(tasks).cancelAll();
+ refreshedThisBundle.set(false);
}
@Teardown
@@ -610,8 +734,8 @@ private Verdict verify(String filePath, FileFormat format, @Nullable ParquetMeta
/**
* The pre-pass commits the schema before paths reach this stage, so the cached table normally
- * covers every file. If not, refresh once (a commit may have landed since the table was cached)
- * and report the remaining delta. Never changes the schema.
+ * covers every file. If not, refresh once per bundle (a commit may have landed since the table
+ * was cached) and report the remaining delta. Never changes the schema.
*/
private @Nullable String uncoveredReason(org.apache.iceberg.Schema fileSchema) {
Table table = checkStateNotNull(this.table);
@@ -619,8 +743,12 @@ private Verdict verify(String filePath, FileFormat format, @Nullable ParquetMeta
if (delta.isEmpty()) {
return null;
}
- synchronized (this) {
- table.refresh();
+ // a commit can land after the table was cached; one refresh per bundle is enough to see it,
+ // and a bundle full of routed files must not load the table once per file
+ if (refreshedThisBundle.compareAndSet(false, true)) {
+ synchronized (this) {
+ table.refresh();
+ }
}
delta = SchemaDelta.classify(table, fileSchema);
if (delta.isEmpty()) {
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 1c842661dcee..fa016ddca862 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
@@ -24,8 +24,12 @@
import com.google.auto.service.AutoService;
import com.google.auto.value.AutoValue;
+import java.util.Arrays;
+import java.util.EnumSet;
+import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.beam.sdk.schemas.AutoValueSchema;
import org.apache.beam.sdk.schemas.Schema;
import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
@@ -67,7 +71,7 @@ public static Builder builder() {
@SchemaFieldDescription("Properties used to set up the Iceberg catalog.")
public abstract @Nullable Map getCatalogProperties();
- @SchemaFieldDescription("Properties passed to the Hadoop ")
+ @SchemaFieldDescription("Properties passed to the Hadoop configuration the catalog uses.")
public abstract @Nullable Map getConfigProperties();
@SchemaFieldDescription(
@@ -115,7 +119,53 @@ public static Builder builder() {
+ "For more information on sort orders, please visit https://iceberg.apache.org/spec/#sort-orders.")
public abstract @Nullable List getSortFields();
- @SchemaFieldDescription("This option specifies whether and where to output unwritable rows.")
+ @SchemaFieldDescription(
+ "Lets the transform change the table schema so that every file's columns are covered."
+ + " 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"
+ + " widened, for example int to long). Leave it empty to never change the table"
+ + " schema. When any option is set, the transform reads the footer of every Parquet"
+ + " file and commits the allowed changes before registering any file, so every"
+ + " registered file has statistics for all of its columns. A file that needs a change"
+ + " 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.")
+ public abstract @Nullable List getSchemaEvolutionOptions();
+
+ @SchemaFieldDescription(
+ "Columns that must always be present and never null, as dotted paths for nested fields"
+ + " (for example address.city). They are never made optional, whatever the options"
+ + " allow, and are created as required when the transform creates the table. A file"
+ + " that lacks one of these columns, or holds nulls in it, is sent to the error output"
+ + " (see error_handling). So is a file whose footer marks the column as optional and"
+ + " has no null-count statistics for it, unless unverifiable_file_handling is ACCEPT."
+ + " Requires schema_evolution_options.")
+ public abstract @Nullable List getRequiredColumns();
+
+ @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."
+ + " FAIL_PIPELINE (the default) fails the pipeline before any schema change is"
+ + " committed. ROUTE_TO_ERRORS commits the changes for the other files and sends the"
+ + " incompatible files to the error output; it requires error_handling.")
+ public abstract @Nullable String getIncompatibleSchemaHandling();
+
+ @SchemaFieldDescription(
+ "What happens to a file the checks cannot verify: an ORC or Avro file (the checks read"
+ + " 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"
+ + " lacks a required column, or holds nulls in it, makes reads of the table fail.")
+ public abstract @Nullable String getUnverifiableFileHandling();
+
+ @SchemaFieldDescription(
+ "Whether and where to output the files that could not be registered, as rows with the"
+ + " file path and the error. Without it those files are dropped.")
public abstract @Nullable ErrorHandling getErrorHandling();
@AutoValue.Builder
@@ -140,9 +190,81 @@ public abstract static class Builder {
public abstract Builder setErrorHandling(ErrorHandling errorHandling);
+ public abstract Builder setSchemaEvolutionOptions(List options);
+
+ public abstract Builder setRequiredColumns(List columns);
+
+ public abstract Builder setIncompatibleSchemaHandling(String handling);
+
+ public abstract Builder setUnverifiableFileHandling(String handling);
+
public abstract Configuration build();
}
+ /** Validates and converts the schema evolution settings; null when none are set. */
+ public @Nullable SchemaEvolutionConfig getSchemaEvolution() {
+ List optionNames = getSchemaEvolutionOptions();
+ List pins = getRequiredColumns();
+ String handlingName = getIncompatibleSchemaHandling();
+ String unverifiableName = getUnverifiableFileHandling();
+ boolean nothingSet =
+ (optionNames == null || optionNames.isEmpty())
+ && (pins == null || pins.isEmpty())
+ && handlingName == null
+ && unverifiableName == null;
+ 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");
+ Set options = EnumSet.noneOf(SchemaEvolutionOption.class);
+ for (String name : checkStateNotNull(optionNames)) {
+ options.add(parseEnum(SchemaEvolutionOption.class, name, "schema_evolution_options"));
+ }
+ SchemaEvolutionConfig.Builder builder = SchemaEvolutionConfig.builder().setOptions(options);
+ if (pins != null) {
+ builder = builder.setRequiredColumns(new LinkedHashSet<>(pins));
+ }
+ if (handlingName != null) {
+ SchemaEvolutionConfig.IncompatibleSchemaHandling handling =
+ parseEnum(
+ SchemaEvolutionConfig.IncompatibleSchemaHandling.class,
+ handlingName,
+ "incompatible_schema_handling");
+ // the error output exists only with error_handling; routed files would vanish otherwise
+ Preconditions.checkArgument(
+ handling != SchemaEvolutionConfig.IncompatibleSchemaHandling.ROUTE_TO_ERRORS
+ || ErrorHandling.hasOutput(getErrorHandling()),
+ "incompatible_schema_handling: ROUTE_TO_ERRORS needs error_handling to receive the"
+ + " routed files");
+ builder = builder.setIncompatibleSchemaHandling(handling);
+ }
+ if (unverifiableName != null) {
+ builder =
+ builder.setUnverifiableFileHandling(
+ parseEnum(
+ SchemaEvolutionConfig.UnverifiableFileHandling.class,
+ unverifiableName,
+ "unverifiable_file_handling"));
+ }
+ return builder.build();
+ }
+
+ private static > T parseEnum(Class type, String name, String option) {
+ for (T value : checkStateNotNull(type.getEnumConstants())) {
+ if (value.name().equalsIgnoreCase(name.trim())) {
+ return value;
+ }
+ }
+ throw new IllegalArgumentException(
+ String.format(
+ "Invalid %s value '%s'. Valid values: %s",
+ option, name, Arrays.toString(type.getEnumConstants())));
+ }
+
public IcebergCatalogConfig getIcebergCatalog() {
return IcebergCatalogConfig.builder()
.setCatalogProperties(getCatalogProperties())
@@ -186,7 +308,8 @@ public PCollectionRowTuple expand(PCollectionRowTuple input) {
configuration.getSortFields(),
configuration.getTableProperties(),
configuration.getManifestFileSize(),
- frequency != null ? Duration.standardSeconds(frequency) : null));
+ frequency != null ? Duration.standardSeconds(frequency) : null,
+ configuration.getSchemaEvolution()));
PCollectionRowTuple output = PCollectionRowTuple.of("snapshots", result.get(OUTPUT_TAG));
ErrorHandling errorHandling = configuration.getErrorHandling();
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 6b67b390ae58..fcbfc906f3c0 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
@@ -22,6 +22,7 @@
import java.util.Collections;
import java.util.EnumSet;
import java.util.Set;
+import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions;
import org.checkerframework.checker.nullness.qual.Nullable;
@@ -110,12 +111,13 @@ public boolean isPinned(String columnPath) {
public abstract UnverifiableFileHandling getUnverifiableFileHandling();
- public IncompatibleSchemaHandling incompatibleSchemaHandling(boolean bounded) {
+ /** The handling to apply: the configured one, or the default for the input's mode when unset. */
+ public IncompatibleSchemaHandling incompatibleSchemaHandlingFor(PCollection.IsBounded mode) {
IncompatibleSchemaHandling handling = getIncompatibleSchemaHandling();
if (handling != null) {
return handling;
}
- return bounded
+ return mode == PCollection.IsBounded.BOUNDED
? IncompatibleSchemaHandling.FAIL_PIPELINE
: IncompatibleSchemaHandling.ROUTE_TO_ERRORS;
}
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionOption.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionOption.java
index 62cf54db725b..ecf0ed92df2f 100644
--- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionOption.java
+++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaEvolutionOption.java
@@ -26,8 +26,9 @@
*
* The options constrain changes to columns the table already has when a window's schema commit
* starts (the whole input, in batch). A column that is new in that window takes the union of the
- * window's file schemas: its type is the widest among them and it is optional unless pinned, so two
- * files that disagree about a new column never conflict with each other, only with the table.
+ * window's file schemas: its type is the widest among them and it is optional (a pinned column is
+ * created required only when the transform creates the table), so two files that disagree about a
+ * new column never conflict with each other, only with the table.
*/
public enum SchemaEvolutionOption {
/**
diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
index 39a1ab4d427e..5afc6107c58b 100644
--- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java
@@ -24,6 +24,7 @@
import static org.apache.beam.sdk.values.TypeDescriptors.strings;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import com.google.api.services.storage.model.StorageObject;
@@ -34,7 +35,9 @@
import com.google.cloud.storage.StorageException;
import com.google.cloud.storage.StorageOptions;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
+import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -131,7 +134,9 @@ public class AddFilesIT {
"warehouse", WAREHOUSE,
"header.x-goog-user-project", PROJECT,
"rest.auth.type", "google",
- "io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO");
+ "io-impl", "org.apache.iceberg.gcp.gcs.GCSFileIO",
+ // required by vended-credentials catalogs, harmless on end-user ones
+ "header.X-Iceberg-Access-Delegation", "vended-credentials");
private Storage storage;
private PubsubClient pubsub;
private Notification notification;
@@ -497,6 +502,113 @@ record ->
checkRecordsInDestinationTable(/* alsoCheckWithBigQueryIO= */ true);
}
+ /**
+ * Schema evolution against the live catalog: a narrow pre-existing table and files that add a
+ * column. Everything must register with stats for the added column and read back.
+ */
+ @Test
+ public void testBatchParquetImportWithSchemaEvolution() throws IOException {
+ Schema narrow = Schema.builder().addInt64Field("id").addStringField("name").build();
+ catalog.createTable(destTableId, beamSchemaToIcebergSchema(narrow));
+
+ String parquetDir = format("%s/%s/", WAREHOUSE, dirName);
+ String tempDir = format("%s/%s-tmp/", WAREHOUSE, dirName);
+ writeParquet(TEST_ROWS, ROW_SCHEMA, parquetDir + "plain/", tempDir + "plain/");
+
+ GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil();
+ List writtenFilePaths =
+ Lists.newArrayList(
+ gcsUtil.listObjects(WAREHOUSE.replace("gs://", ""), dirName, null).getItems())
+ .stream()
+ .map(o -> format("gs://%s/%s", o.getBucket(), o.getName()))
+ .collect(Collectors.toList());
+ assertEquals(20, writtenFilePaths.size());
+
+ SchemaEvolutionConfig evolution =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))
+ .build();
+ Pipeline p = Pipeline.create();
+ PCollectionRowTuple tuple =
+ p.apply(Create.of(writtenFilePaths))
+ .apply(
+ new AddFiles(
+ IcebergCatalogConfig.builder().setCatalogProperties(BIGLAKE_PROPS).build(),
+ namespace + "." + destTableName,
+ null,
+ null,
+ null,
+ TABLE_PROPS,
+ null,
+ null,
+ evolution));
+ PAssert.that(tuple.get("errors")).empty();
+ p.run().waitUntilFinish();
+
+ Table destTable = catalog.loadTable(destTableId);
+ org.apache.iceberg.types.Types.NestedField age = destTable.schema().findField("age");
+ assertNotNull("column added by the pre-pass", age);
+ assertTrue(checkTableHasRegisteredParquetFiles(writtenFilePaths));
+ int nameId = destTable.schema().findField("name").fieldId();
+ for (org.apache.iceberg.FileScanTask task :
+ destTable.newScan().includeColumnStats().planFiles()) {
+ assertEquals(
+ "stats for name on " + task.file().path(),
+ Long.valueOf(0),
+ task.file().nullValueCounts().get(nameId));
+ assertEquals(
+ "stats for age on " + task.file().path(),
+ Long.valueOf(0),
+ task.file().nullValueCounts().get(age.fieldId()));
+ }
+ // every row reads back
+ List expected = new ArrayList<>();
+ Schema wide =
+ Schema.builder()
+ .addInt64Field("id")
+ .addStringField("name")
+ .addNullableInt32Field("age")
+ .build();
+ for (Row row : TEST_ROWS) {
+ expected.add(
+ Row.withSchema(wide)
+ .addValues(row.getInt64("id"), row.getString("name"), row.getInt32("age"))
+ .build());
+ }
+ Pipeline s = Pipeline.create();
+ PCollection destRows =
+ s.apply(
+ Managed.read(Managed.ICEBERG)
+ .withConfig(
+ ImmutableMap.of(
+ "table", destTableId.toString(), "catalog_properties", BIGLAKE_PROPS)))
+ .getSinglePCollection()
+ .apply(MapElements.into(strings()).via(AddFilesIT::canonicalRecord));
+ PAssert.that(destRows)
+ .containsInAnyOrder(
+ expected.stream().map(AddFilesIT::canonicalRecord).collect(Collectors.toList()));
+ s.run().waitUntilFinish();
+ }
+
+ private static void writeParquet(List rows, Schema schema, String dir, String tempDir) {
+ Pipeline q = Pipeline.create();
+ org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(schema);
+ q.apply(Create.of(rows).withRowSchema(schema))
+ .apply(
+ MapElements.into(TypeDescriptor.of(GenericRecord.class))
+ .via(AvroUtils.getRowToGenericRecordFunction(avroSchema)))
+ .setCoder(AvroCoder.of(avroSchema))
+ .apply(
+ FileIO.writeDynamic()
+ .by(record -> String.valueOf(record.get("id")))
+ .via(ParquetIO.sink(avroSchema))
+ .withNaming(name -> defaultNaming(name, ".parquet"))
+ .withTempDirectory(tempDir)
+ .to(dir)
+ .withDestinationCoder(StringUtf8Coder.of()));
+ q.run().waitUntilFinish();
+ }
+
private void checkRecordsInDestinationTable(boolean alsoCheckWithBigQueryIO) {
Pipeline s = Pipeline.create();
PCollection destRows =
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
new file mode 100644
index 000000000000..2e469ef6da71
--- /dev/null
+++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java
@@ -0,0 +1,171 @@
+/*
+ * 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.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
+import org.apache.beam.sdk.io.iceberg.AddFilesSchemaTransformProvider.Configuration;
+import org.apache.beam.sdk.schemas.transforms.providers.ErrorHandling;
+import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+@RunWith(JUnit4.class)
+public class AddFilesSchemaTransformProviderTest {
+
+ private static Configuration.Builder base() {
+ return Configuration.builder()
+ .setTable("default.t")
+ .setCatalogProperties(ImmutableMap.of("type", "hadoop", "warehouse", "/tmp/w"));
+ }
+
+ @Test
+ public void testNoEvolutionSettingsMeanEvolutionDisabled() {
+ assertNull(base().build().getSchemaEvolution());
+ assertNull(
+ base().setSchemaEvolutionOptions(Collections.emptyList()).build().getSchemaEvolution());
+ }
+
+ @Test
+ public void testOptionsParsedCaseInsensitively() {
+ SchemaEvolutionConfig config =
+ base()
+ .setSchemaEvolutionOptions(
+ Arrays.asList("allow_field_addition", " ALLOW_TYPE_PROMOTION "))
+ .build()
+ .getSchemaEvolution();
+ assertNotNull(config);
+ assertEquals(
+ EnumSet.of(
+ SchemaEvolutionOption.ALLOW_FIELD_ADDITION, SchemaEvolutionOption.ALLOW_TYPE_PROMOTION),
+ config.getOptions());
+ assertNull(config.getIncompatibleSchemaHandling());
+ }
+
+ @Test
+ public void testPinsAndHandlingParsed() {
+ SchemaEvolutionConfig config =
+ base()
+ .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_RELAXATION"))
+ .setRequiredColumns(Arrays.asList("id", "address.city"))
+ .setIncompatibleSchemaHandling("route_to_errors")
+ .setErrorHandling(ErrorHandling.builder().setOutput("errors").build())
+ .build()
+ .getSchemaEvolution();
+ assertNotNull(config);
+ assertTrue(config.isPinned("address.city"));
+ assertEquals(
+ SchemaEvolutionConfig.IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ config.getIncompatibleSchemaHandling());
+ }
+
+ @Test
+ public void testUnverifiableFileHandlingParsedAndDefaultsToReject() {
+ SchemaEvolutionConfig config =
+ base()
+ .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION"))
+ .setUnverifiableFileHandling(" accept ")
+ .build()
+ .getSchemaEvolution();
+ assertNotNull(config);
+ assertEquals(
+ SchemaEvolutionConfig.UnverifiableFileHandling.ACCEPT,
+ config.getUnverifiableFileHandling());
+ assertEquals(
+ SchemaEvolutionConfig.UnverifiableFileHandling.REJECT,
+ base()
+ .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION"))
+ .build()
+ .getSchemaEvolution()
+ .getUnverifiableFileHandling());
+ }
+
+ @Test
+ public void testInvalidOptionListsValidValues() {
+ IllegalArgumentException e =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ base()
+ .setSchemaEvolutionOptions(Arrays.asList("ALLOW_EVERYTHING"))
+ .build()
+ .getSchemaEvolution());
+ assertTrue(e.getMessage(), e.getMessage().contains("ALLOW_FIELD_ADDITION"));
+ assertTrue(e.getMessage(), e.getMessage().contains("ALLOW_EVERYTHING"));
+ }
+
+ @Test
+ public void testInvalidHandlingValuesRejected() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ base()
+ .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION"))
+ .setIncompatibleSchemaHandling("ignore")
+ .build()
+ .getSchemaEvolution());
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ base()
+ .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION"))
+ .setUnverifiableFileHandling("trust")
+ .build()
+ .getSchemaEvolution());
+ }
+
+ @Test
+ public void testRouteToErrorsWithoutErrorHandlingRejected() {
+ IllegalArgumentException e =
+ assertThrows(
+ IllegalArgumentException.class,
+ () ->
+ base()
+ .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION"))
+ .setIncompatibleSchemaHandling("ROUTE_TO_ERRORS")
+ .build()
+ .getSchemaEvolution());
+
+ assertTrue(e.getMessage(), e.getMessage().contains("error_handling"));
+ }
+
+ /** The config's "needs options" rule, reported under the YAML key the user must set. */
+ @Test
+ public void testSettingsWithoutOptionsRejectedNamingTheOptionsKey() {
+ IllegalArgumentException pins =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> base().setRequiredColumns(Arrays.asList("id")).build().getSchemaEvolution());
+ IllegalArgumentException unverifiable =
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> base().setUnverifiableFileHandling("ACCEPT").build().getSchemaEvolution());
+
+ assertTrue(pins.getMessage(), pins.getMessage().contains("schema_evolution_options"));
+ assertTrue(
+ unverifiable.getMessage(), unverifiable.getMessage().contains("schema_evolution_options"));
+ }
+}
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 85d82b845519..5e6cbdf4ea32 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
@@ -42,13 +42,16 @@
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
+import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.UnverifiableFileHandling;
import org.apache.beam.sdk.metrics.MetricNameFilter;
import org.apache.beam.sdk.metrics.MetricResult;
import org.apache.beam.sdk.metrics.MetricsFilter;
+import org.apache.beam.sdk.runners.TransformHierarchy;
import org.apache.beam.sdk.testing.ExpectedLogs;
import org.apache.beam.sdk.testing.PAssert;
import org.apache.beam.sdk.testing.TestPipeline;
@@ -56,10 +59,13 @@
import org.apache.beam.sdk.transforms.Count;
import org.apache.beam.sdk.transforms.Create;
import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.windowing.FixedWindows;
+import org.apache.beam.sdk.transforms.windowing.Window;
import org.apache.beam.sdk.values.PCollection;
import org.apache.beam.sdk.values.PCollectionRowTuple;
import org.apache.beam.sdk.values.PCollectionTuple;
import org.apache.beam.sdk.values.Row;
+import org.apache.beam.sdk.values.TimestampedValue;
import org.apache.beam.sdk.values.TupleTagList;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
@@ -67,6 +73,7 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.FileScanTask;
import org.apache.iceberg.Files;
import org.apache.iceberg.ManifestFile;
import org.apache.iceberg.Metrics;
@@ -84,6 +91,7 @@
import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.exceptions.CommitFailedException;
import org.apache.iceberg.hadoop.HadoopCatalog;
import org.apache.iceberg.io.DataWriter;
import org.apache.iceberg.io.InputFile;
@@ -97,6 +105,7 @@
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.Duration;
+import org.joda.time.Instant;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Ignore;
@@ -964,6 +973,237 @@ private Record record(int id, String name, int age) {
return GenericRecord.create(icebergSchema).copy("id", id, "name", name, "age", age);
}
+ // ---- AddFiles with schema evolution
+
+ private AddFiles addFiles(@Nullable SchemaEvolutionConfig config) {
+ return new AddFiles(
+ catalogConfig, tableId.toString(), null, null, null, null, null, null, config);
+ }
+
+ private static int countTransforms(Pipeline pipeline, String name) {
+ int[] count = {0};
+ pipeline.traverseTopologically(
+ new Pipeline.PipelineVisitor.Defaults() {
+ @Override
+ public CompositeBehavior enterCompositeTransform(TransformHierarchy.Node node) {
+ if (node.getFullName().contains(name)) {
+ count[0]++;
+ }
+ return CompositeBehavior.ENTER_TRANSFORM;
+ }
+ });
+ return count[0];
+ }
+
+ /** A file whose name column is an int: a type conflict no option allows. */
+ private String writeConflicting(String name) throws IOException {
+ Schema conflicting =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.required(2, "name", Types.IntegerType.get()),
+ Types.NestedField.required(3, "age", Types.IntegerType.get()));
+ Record record = GenericRecord.create(conflicting);
+ record.setField("id", 1);
+ record.setField("name", 5);
+ record.setField("age", 1);
+ return writeWithSchema(name, conflicting, record);
+ }
+
+ private static Map nullCountsOf(Table table, String fileName) {
+ for (FileScanTask task : table.newScan().includeColumnStats().planFiles()) {
+ if (task.file().path().toString().endsWith(fileName)) {
+ return checkStateNotNull(task.file().nullValueCounts());
+ }
+ }
+ throw new AssertionError(fileName + " is not registered");
+ }
+
+ private void assertEmailAddedAndFilesRegistered(int files) {
+ Table table = catalog.loadTable(tableId);
+ Types.NestedField email = table.schema().findField("email");
+ assertNotNull(email);
+ assertTrue(email.isOptional());
+ assertEquals(files, Iterables.size(table.newScan().planFiles()));
+ }
+
+ @Test
+ public void testEvolutionAddsColumnsBeforeRegisteringFiles() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String narrow = writeOneRecord("narrow.parquet");
+ String wide = writeWider("wide.parquet");
+
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(narrow, wide)).apply(addFiles(ADDITIONS));
+ PAssert.that(output.get("errors")).empty();
+ assertEquals(1, countTransforms(pipeline, "ReadFooterSchema"));
+
+ pipeline.run().waitUntilFinish();
+
+ assertEmailAddedAndFilesRegistered(2);
+ Table table = catalog.loadTable(tableId);
+ assertEquals(1, Iterables.size(table.snapshots()));
+ int emailId = table.schema().findField("email").fieldId();
+ assertTrue(
+ "stats for the added column", nullCountsOf(table, "wide.parquet").containsKey(emailId));
+ }
+
+ @Test
+ public void testEvolutionDisabledAddsNoPrePassTransforms() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String file = writeOneRecord("data.parquet");
+
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(file)).apply(addFiles(null));
+ PAssert.that(output.get("errors")).empty();
+ assertEquals(0, countTransforms(pipeline, "ReadFooterSchema"));
+ assertEquals(0, countTransforms(pipeline, "WaitForSchemaCommit"));
+
+ pipeline.run().waitUntilFinish();
+
+ assertEquals(1, Iterables.size(catalog.loadTable(tableId).newScan().planFiles()));
+ }
+
+ @Test
+ public void testIncompatibleSchemaFailsBatchPipelineByDefault() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String good = writeOneRecord("good.parquet");
+ String bad = writeConflicting("bad.parquet");
+
+ pipeline
+ .apply("Create Input", Create.of(good, bad))
+ .apply(addFiles(SchemaEvolutionConfig.of(SchemaEvolutionOption.values())));
+
+ Exception e = assertThrows(Exception.class, () -> pipeline.run().waitUntilFinish());
+
+ assertThat(e.getMessage(), containsString("Incompatible schemas"));
+ assertEquals(0, Iterables.size(catalog.loadTable(tableId).snapshots()));
+ }
+
+ @Test
+ public void testIncompatibleSchemaRoutedToErrorsWhenConfigured() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String good = writeOneRecord("good.parquet");
+ String bad = writeConflicting("bad.parquet");
+ SchemaEvolutionConfig route =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setIncompatibleSchemaHandling(
+ SchemaEvolutionConfig.IncompatibleSchemaHandling.ROUTE_TO_ERRORS)
+ .build();
+
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(good, bad)).apply(addFiles(route));
+ PAssert.that(output.get("errors"))
+ .satisfies(
+ rows -> {
+ Row row = Iterables.getOnlyElement(rows);
+ assertEquals(bad, row.getString("file"));
+ assertThat(row.getString("error"), containsString("does not cover the file"));
+ return null;
+ });
+
+ pipeline.run().waitUntilFinish();
+
+ Table table = catalog.loadTable(tableId);
+ assertEquals(1, Iterables.size(table.snapshots()));
+ assertEquals(1, Iterables.size(table.newScan().planFiles()));
+ }
+
+ @Test
+ public void testMissingTableIsCreatedFromTheFilesUnion() throws Exception {
+ String narrow = writeOneRecord("narrow.parquet");
+ String wide = writeWider("wide.parquet");
+
+ PCollectionRowTuple output =
+ pipeline.apply("Create Input", Create.of(narrow, wide)).apply(addFiles(ADDITIONS));
+ PAssert.that(output.get("errors")).empty();
+
+ pipeline.run().waitUntilFinish();
+
+ assertEmailAddedAndFilesRegistered(2);
+ assertTrue(
+ "created columns are optional",
+ catalog.loadTable(tableId).schema().findField("id").isOptional());
+ }
+
+ /** Streaming schema evolution comes in a follow-up: until then the front door rejects it. */
+ @Test
+ public void testUnboundedInputWithEvolutionIsRejected() {
+ pipeline.enableAbandonedNodeEnforcement(false);
+ PCollection unbounded =
+ pipeline.apply(TestStream.create(StringUtf8Coder.of()).advanceWatermarkToInfinity());
+ AddFiles streaming =
+ new AddFiles(
+ catalogConfig,
+ tableId.toString(),
+ null,
+ null,
+ null,
+ null,
+ 10,
+ Duration.standardSeconds(5),
+ ADDITIONS);
+
+ IllegalArgumentException e =
+ assertThrows(IllegalArgumentException.class, () -> unbounded.apply(streaming));
+
+ assertThat(e.getMessage(), containsString("not yet supported for unbounded input"));
+ }
+
+ /**
+ * Whatever windowing the caller applied upstream, the pre-pass rewindows into the global window:
+ * one schema commit covers the whole input and the Wait.on gate holds every file behind it.
+ */
+ @Test
+ public void testUpstreamWindowedBatchInputEvolvesAndRegisters() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String narrow = writeOneRecord("narrow.parquet");
+ String wide = writeWider("wide.parquet");
+
+ PCollectionRowTuple output =
+ pipeline
+ .apply(
+ "Create Input",
+ Create.timestamped(
+ TimestampedValue.of(narrow, new Instant(0)),
+ TimestampedValue.of(wide, new Instant(60_000))))
+ .apply("UpstreamWindow", Window.into(FixedWindows.of(Duration.standardSeconds(30))))
+ .apply(addFiles(ADDITIONS));
+ PAssert.that(output.get("errors")).empty();
+
+ pipeline.run().waitUntilFinish();
+
+ assertEmailAddedAndFilesRegistered(2);
+ }
+
+ /**
+ * 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.
+ */
+ @Test
+ public void testTransientSchemaCommitFailureIsRetried() throws Exception {
+ catalog.createTable(tableId, icebergSchema);
+ String wide = writeWider("wide.parquet");
+ AtomicInteger attempts = new AtomicInteger();
+ CommitSchemaUnion.Committer failsOnce =
+ txn -> {
+ if (attempts.incrementAndGet() == 1) {
+ throw new CommitFailedException("transient");
+ }
+ txn.commitTransaction();
+ };
+
+ PCollectionRowTuple output =
+ pipeline
+ .apply("Create Input", Create.of(wide))
+ .apply(addFiles(ADDITIONS).withSchemaCommitter(failsOnce));
+ PAssert.that(output.get("errors")).empty();
+
+ pipeline.run().waitUntilFinish();
+
+ assertEmailAddedAndFilesRegistered(1);
+ }
+
// ---- ConvertToDataFile coverage check and pinned columns
private static final SchemaEvolutionConfig ADDITIONS =
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 4ccf844538fd..c6d565cda957 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
@@ -27,6 +27,7 @@
import java.util.HashSet;
import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.UnverifiableFileHandling;
+import org.apache.beam.sdk.values.PCollection;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
@@ -105,10 +106,14 @@ public void testIncompatibleSchemaHandlingDefaultsByMode() {
.setIncompatibleSchemaHandling(IncompatibleSchemaHandling.ROUTE_TO_ERRORS)
.build();
- assertEquals(IncompatibleSchemaHandling.FAIL_PIPELINE, unset.incompatibleSchemaHandling(true));
assertEquals(
- IncompatibleSchemaHandling.ROUTE_TO_ERRORS, unset.incompatibleSchemaHandling(false));
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ unset.incompatibleSchemaHandlingFor(PCollection.IsBounded.BOUNDED));
assertEquals(
- IncompatibleSchemaHandling.ROUTE_TO_ERRORS, forced.incompatibleSchemaHandling(true));
+ IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ unset.incompatibleSchemaHandlingFor(PCollection.IsBounded.UNBOUNDED));
+ assertEquals(
+ IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ forced.incompatibleSchemaHandlingFor(PCollection.IsBounded.BOUNDED));
}
}
diff --git a/sdks/python/apache_beam/yaml/standard_io.yaml b/sdks/python/apache_beam/yaml/standard_io.yaml
index a70617ca4e6d..c41dd6f2e07f 100644
--- a/sdks/python/apache_beam/yaml/standard_io.yaml
+++ b/sdks/python/apache_beam/yaml/standard_io.yaml
@@ -657,6 +657,10 @@
partition_fields: 'partition_fields'
table_properties: 'table_properties'
sort_fields: 'sort_fields'
+ schema_evolution_options: 'schema_evolution_options'
+ required_columns: 'required_columns'
+ incompatible_schema_handling: 'incompatible_schema_handling'
+ unverifiable_file_handling: 'unverifiable_file_handling'
error_handling: 'error_handling'
underlying_provider:
type: beamJar
diff --git a/sdks/python/apache_beam/yaml/tests/iceberg_add_files_evolution.yaml b/sdks/python/apache_beam/yaml/tests/iceberg_add_files_evolution.yaml
new file mode 100644
index 000000000000..d37f872442a8
--- /dev/null
+++ b/sdks/python/apache_beam/yaml/tests/iceberg_add_files_evolution.yaml
@@ -0,0 +1,98 @@
+#
+# 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: register both; the table is created from the union of the file schemas
+ - pipeline:
+ type: chain
+ transforms:
+ - type: ReadMatchFiles
+ config:
+ file_pattern: "{TEMP_DIR}/data/*.parquet"
+ - type: MapToFields
+ config:
+ language: python
+ fields:
+ path:
+ callable: "lambda row: row.path"
+ output_type: string
+ - type: IcebergAddFiles
+ config:
+ table: "default.table"
+ catalog_properties:
+ type: "hadoop"
+ warehouse: "{TEMP_DIR}/dir"
+ schema_evolution_options: [ALLOW_FIELD_ADDITION, ALLOW_FIELD_RELAXATION]
+
+ providers:
+ - type: python
+ config: { }
+ transforms:
+ ReadMatchFiles: 'apache_beam.io.fileio.MatchFiles'
+
+ # Pipeline 4: every file is readable through the evolved schema
+ - pipeline:
+ type: chain
+ transforms:
+ - type: ReadFromIceberg
+ config:
+ table: "default.table"
+ catalog_properties:
+ type: "hadoop"
+ warehouse: "{TEMP_DIR}/dir"
+ - type: AssertEqual
+ config:
+ elements:
+ - {label: "11a", rank: 0, bool: null}
+ - {label: "37a", rank: 1, bool: null}
+ - {label: "389a", rank: 2, bool: false}
+ - {label: "3821b", rank: 3, bool: true}