This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new d3756036475 Addfiles wiring provider (#40150)
d3756036475 is described below
commit d37560364756bface6fcffad8575e6c47cb91d5e
Author: claudevdm <[email protected]>
AuthorDate: Thu Sep 17 13:42:21 2026 -0400
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 +
.../org/apache/beam/sdk/io/iceberg/AddFiles.java | 146 ++++++++++++-
.../iceberg/AddFilesSchemaTransformProvider.java | 129 ++++++++++-
.../beam/sdk/io/iceberg/SchemaEvolutionConfig.java | 6 +-
.../beam/sdk/io/iceberg/SchemaEvolutionOption.java | 5 +-
.../org/apache/beam/sdk/io/iceberg/AddFilesIT.java | 114 +++++++++-
.../AddFilesSchemaTransformProviderTest.java | 171 +++++++++++++++
.../apache/beam/sdk/io/iceberg/AddFilesTest.java | 240 +++++++++++++++++++++
.../sdk/io/iceberg/SchemaEvolutionConfigTest.java | 11 +-
sdks/python/apache_beam/yaml/standard_io.yaml | 4 +
.../yaml/tests/iceberg_add_files_evolution.yaml | 98 +++++++++
12 files changed, 906 insertions(+), 21 deletions(-)
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json
b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
index 89e73b29da0..e1430c74e62 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 aecd194e1f2..47da5f81fb0 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -99,6 +99,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 daf61bf883f..75b1a0144e2 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.Set;
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.schemas.SchemaRegistry;
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.Logger;
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.
+ *
+ * <p>Outputs: {@code snapshots} (one row per commit), {@code errors} (one row
per file that could
+ * not be registered: {@code file}, {@code error}).
+ *
+ * <p><b>Schema evolution.</b> With a {@link SchemaEvolutionConfig} whose
options are set, a
+ * pre-pass reads every Parquet footer, classifies the change each distinct
file schema needs on the
+ * 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.
+ *
+ * <pre>{@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));
+ * }</pre>
+ *
+ * <p>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<PCollection<String>,
PCollectionRowTuple> {
static final String OUTPUT_TAG = "snapshots";
@@ -142,6 +178,8 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
private final @Nullable List<String> partitionFields;
private final @Nullable List<String> sortFields;
private final @Nullable Map<String, String> tableProps;
+ private final SchemaEvolutionConfig evolution;
+ private CommitSchemaUnion.Committer committer =
CommitSchemaUnion.DEFAULT_COMMITTER;
public AddFiles(
IcebergCatalogConfig catalogConfig,
@@ -152,6 +190,40 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
@Nullable Map<String, String> 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<String> partitionFields,
+ @Nullable List<String> sortFields,
+ @Nullable Map<String, String> 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 class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
@Override
public PCollectionRowTuple expand(PCollection<String> 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 class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
"AddFiles configured to build partition metadata after the prefix:
'{}'", locationPrefix);
}
+ PCollection<String> paths = input;
+ if (evolution.isEnabled()) {
+ paths = gateOnSchemaCommit(input);
+ }
+
PCollectionTuple dataFiles =
- input.apply(
+ paths.apply(
"ConvertToDataFiles",
ParDo.of(
new ConvertToDataFile(
@@ -192,7 +273,8 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
locationPrefix,
partitionFields,
sortFields,
- tableProps))
+ tableProps,
+ evolution))
.withOutputTags(DATA_FILES, TupleTagList.of(ERRORS)));
SchemaCoder<SerializableDataFile> sdfCoder;
try {
@@ -238,6 +320,47 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
OUTPUT_TAG, snapshots, ERROR_TAG,
dataFiles.get(ERRORS).setRowSchema(ERROR_SCHEMA));
}
+ /**
+ * Holds every path until the schema commit has landed. The pre-pass window
is the unit of commit,
+ * and for bounded input that unit is the whole input: paths are rewindowed
into the global window
+ * first, so one commit covers everything and the Wait.on signal lines up
with the main input
+ * whatever windowing the caller applied upstream.
+ */
+ private PCollection<String> gateOnSchemaCommit(PCollection<String> input) {
+ PCollection<String> windowed =
+ input.apply("PrePassGlobalWindow", Window.into(new GlobalWindows()));
+ CommitSchemaUnion.TableCreation creation =
+ new CommitSchemaUnion.TableCreation(partitionFields, sortFields,
tableProps);
+ // unset handling follows the mode (fail the job in batch, route in
streaming); the pre-pass
+ // only runs on bounded input, see expand
+ IncompatibleSchemaHandling onIncompatible =
+ evolution.incompatibleSchemaHandlingFor(input.isBounded());
+ PCollection<Long> signal =
+ windowed
+ .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema()))
+ .setCoder(CollectDistinctSchemas.groupCoder())
+ .apply(
+ "CollectDistinctSchemas",
+ Combine.globally(new
CollectDistinctSchemas()).withoutDefaults())
+ .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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
private transient @MonotonicNonNull BoundedAsyncTasks<ProcessResult> tasks;
private transient volatile @MonotonicNonNull Table table;
private transient @MonotonicNonNull Set<String> 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 class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
/** 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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
/**
* 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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
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 1c842661dce..fa016ddca86 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 static
org.apache.beam.sdk.util.Preconditions.checkStateNotNull;
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 class AddFilesSchemaTransformProvider extends
TypedSchemaTransformProvide
@SchemaFieldDescription("Properties used to set up the Iceberg catalog.")
public abstract @Nullable Map<String, String> getCatalogProperties();
- @SchemaFieldDescription("Properties passed to the Hadoop ")
+ @SchemaFieldDescription("Properties passed to the Hadoop configuration the
catalog uses.")
public abstract @Nullable Map<String, String> getConfigProperties();
@SchemaFieldDescription(
@@ -115,7 +119,53 @@ public class AddFilesSchemaTransformProvider extends
TypedSchemaTransformProvide
+ "For more information on sort orders, please visit
https://iceberg.apache.org/spec/#sort-orders.";)
public abstract @Nullable List<String> 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<String> 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<String> 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 class AddFilesSchemaTransformProvider extends
TypedSchemaTransformProvide
public abstract Builder setErrorHandling(ErrorHandling errorHandling);
+ public abstract Builder setSchemaEvolutionOptions(List<String> options);
+
+ public abstract Builder setRequiredColumns(List<String> 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<String> optionNames = getSchemaEvolutionOptions();
+ List<String> 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<SchemaEvolutionOption> 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 extends Enum<T>> T parseEnum(Class<T> 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 class AddFilesSchemaTransformProvider extends
TypedSchemaTransformProvide
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 6b67b390ae5..fcbfc906f3c 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.io.Serializable;
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 abstract class SchemaEvolutionConfig implements
Serializable {
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 62cf54db725..ecf0ed92df2 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 @@ package org.apache.beam.sdk.io.iceberg;
*
* <p>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 39a1ab4d427..5afc6107c58 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.util.Preconditions.checkStateNotNull;
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.Storage;
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 @@ public class AddFilesIT {
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<String> 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<Row> 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<String> 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<Row> 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.<String, GenericRecord>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<Row> 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 00000000000..2e469ef6da7
--- /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 85d82b84551..5e6cbdf4ea3 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.EnumSet;
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.testing.TestStream;
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.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists;
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.catalog.TableIdentifier;
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.iceberg.util.SerializableFunction;
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 @@ public class AddFilesTest {
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<Integer, Long> 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<String> 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 4ccf844538f..c6d565cda95 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.EnumSet;
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 class SchemaEvolutionConfigTest {
.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 a70617ca4e6..c41dd6f2e07 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 00000000000..d37f872442a
--- /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}
