From 86b74f85a2b97ad902091bb4dd730f12bd0b46ae Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 09:34:18 -0400 Subject: [PATCH 1/8] 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. --- .../apache/beam/sdk/io/iceberg/AddFiles.java | 126 ++++++++- .../beam/sdk/io/iceberg/AddFilesIT.java | 114 +++++++- .../beam/sdk/io/iceberg/AddFilesTest.java | 257 ++++++++++++++++++ 3 files changed, 492 insertions(+), 5 deletions(-) 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..25880e847e76 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 @@ -51,13 +51,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 +116,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(List.of(ALLOW_FIELD_ADDITION, ALLOW_TYPE_PROMOTION))
+ *         .setRequiredColumns(List.of("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 the table unreadable through Iceberg's reader. */ public class AddFiles extends PTransform, PCollectionRowTuple> { static final String OUTPUT_TAG = "snapshots"; @@ -142,6 +176,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 +188,41 @@ 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, and the width of the + * schema pre-pass window + * @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 +237,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 +257,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 +272,8 @@ public PCollectionRowTuple expand(PCollection input) { locationPrefix, partitionFields, sortFields, - tableProps)) + tableProps, + evolution)) .withOutputTags(DATA_FILES, TupleTagList.of(ERRORS))); SchemaCoder sdfCoder; try { @@ -238,6 +319,43 @@ 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); + 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, + evolution.incompatibleSchemaHandling(true), + 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. 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/AddFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java index 85d82b845519..b6be74879501 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 @@ -43,12 +43,14 @@ import java.util.List; import java.util.Map; 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 +58,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; @@ -97,6 +102,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 +970,257 @@ 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]; + } + + @Test + public void testEvolutionAddsColumnsBeforeRegisteringFiles() throws Exception { + catalog.createTable(tableId, icebergSchema); + String narrow = root + "narrow.parquet"; + DataWriter writer = createWriter(narrow); + writer.write(record(1, "a", 1)); + writer.close(); + String wide = writeWider("wide.parquet"); + + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(narrow, wide)) + .apply(addFiles(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))); + PAssert.that(output.get("errors")).empty(); + assertEquals(1, countTransforms(pipeline, "ReadFooterSchema")); + pipeline.run().waitUntilFinish(); + + Table table = catalog.loadTable(tableId); + Types.NestedField email = table.schema().findField("email"); + assertNotNull(email); + assertTrue(email.isOptional()); + assertEquals(1, Iterables.size(table.snapshots())); + // the wide file's stats cover the new column + boolean sawEmailStats = false; + for (org.apache.iceberg.FileScanTask task : table.newScan().includeColumnStats().planFiles()) { + if (task.file().path().toString().endsWith("wide.parquet")) { + sawEmailStats = task.file().nullValueCounts().containsKey(email.fieldId()); + } + } + assertTrue("stats for the added column", sawEmailStats); + } + + @Test + public void testEvolutionDisabledKeepsTodaysGraph() throws Exception { + catalog.createTable(tableId, icebergSchema); + String file = root + "data.parquet"; + DataWriter writer = createWriter(file); + writer.write(record(1, "a", 1)); + writer.close(); + pipeline.apply("Create Input", Create.of(file)).apply(addFiles(null)); + assertEquals(0, countTransforms(pipeline, "ReadFooterSchema")); + assertEquals(0, countTransforms(pipeline, "WaitForSchemaCommit")); + pipeline.run().waitUntilFinish(); + } + + @Test + public void testIncompatibleSchemaFailsBatchPipelineByDefault() throws Exception { + catalog.createTable(tableId, icebergSchema); + 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); + String bad = writeWithSchema("bad.parquet", conflicting, record); + String good = root + "good.parquet"; + DataWriter writer = createWriter(good); + writer.write(record(1, "a", 1)); + writer.close(); + + 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); + 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); + String bad = writeWithSchema("bad.parquet", conflicting, record); + String good = root + "good.parquet"; + DataWriter writer = createWriter(good); + writer.write(record(1, "a", 1)); + writer.close(); + + 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 = root + "narrow.parquet"; + DataWriter writer = createWriter(narrow); + writer.write(record(1, "a", 1)); + writer.close(); + String wide = writeWider("wide.parquet"); + + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(narrow, wide)) + .apply(addFiles(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))); + PAssert.that(output.get("errors")).empty(); + pipeline.run().waitUntilFinish(); + + Table table = catalog.loadTable(tableId); + assertNotNull(table.schema().findField("email")); + assertTrue("created columns are optional", table.schema().findField("id").isOptional()); + assertEquals(2, Iterables.size(table.newScan().planFiles())); + } + + private PCollection stream(String... files) { + TestStream.Builder builder = TestStream.create(StringUtf8Coder.of()); + for (String file : files) { + // far in the past: only the forward stamp keeps these on time + builder = builder.addElements(TimestampedValue.of(file, new Instant(0))); + builder = builder.advanceProcessingTime(Duration.standardSeconds(1)); + } + return pipeline.apply(builder.advanceWatermarkToInfinity()); + } + + private AddFiles streamingAddFiles(SchemaEvolutionConfig config) { + return new AddFiles( + catalogConfig, + tableId.toString(), + null, + null, + null, + null, + 10, + Duration.standardSeconds(5), + config); + } + + /** Streaming schema evolution comes in a follow-up: until then the front door rejects it. */ + @Test + public void testUnboundedInputWithEvolutionIsRejected() { + pipeline.enableAbandonedNodeEnforcement(false); + PCollection paths = stream("unused.parquet"); + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> + paths.apply( + streamingAddFiles( + SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)))); + 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 = root + "narrow.parquet"; + DataWriter writer = createWriter(narrow); + writer.write(record(1, "a", 1)); + writer.close(); + 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(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))); + PAssert.that(output.get("errors")).empty(); + pipeline.run().waitUntilFinish(); + + Table table = catalog.loadTable(tableId); + assertNotNull(table.schema().findField("email")); + assertEquals(2, Iterables.size(table.newScan().planFiles())); + } + + /** + * Async results are emitted in FinishBundle with their own timestamps. Emitting a finished result + * during a later element's ProcessElement would attribute it to that element and trip the + * runner's output-timestamp skew check when timestamps differ. + */ + @Test + public void testTransientSchemaCommitFailureIsRetried() throws Exception { + catalog.createTable(tableId, icebergSchema); + String wide = writeWider("wide.parquet"); + java.util.concurrent.atomic.AtomicInteger attempts = + new java.util.concurrent.atomic.AtomicInteger(); + CommitSchemaUnion.Committer flaky = + txn -> { + if (attempts.incrementAndGet() == 1) { + throw new org.apache.iceberg.exceptions.CommitFailedException("transient"); + } + txn.commitTransaction(); + }; + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(wide)) + .apply( + addFiles(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .withSchemaCommitter(flaky)); + PAssert.that(output.get("errors")).empty(); + pipeline.run().waitUntilFinish(); + assertNotNull(catalog.loadTable(tableId).schema().findField("email")); + } + // ---- ConvertToDataFile coverage check and pinned columns private static final SchemaEvolutionConfig ADDITIONS = From b59ba8a4f78d67775c23983afaf0d68f7ded1ceb Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 09:35:39 -0400 Subject: [PATCH 2/8] 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 --- .../AddFilesSchemaTransformProvider.java | 111 ++++++++++++- .../AddFilesSchemaTransformProviderTest.java | 155 ++++++++++++++++++ sdks/python/apache_beam/yaml/standard_io.yaml | 4 + .../tests/iceberg_add_files_evolution.yaml | 98 +++++++++++ 4 files changed, 367 insertions(+), 1 deletion(-) 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/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..b0610d6d55ea 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,11 @@ import com.google.auto.service.AutoService; import com.google.auto.value.AutoValue; +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; @@ -115,6 +118,46 @@ 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( + "Schema changes the transform may make so that every file's columns are covered by the" + + " table schema, from: ALLOW_FIELD_ADDITION (add file columns absent from the table," + + " as optional), ALLOW_FIELD_RELAXATION (make a required table column optional when" + + " a file may contain nulls in it or lacks it), ALLOW_TYPE_PROMOTION (widen a column" + + " type, e.g. int to long). Empty or absent: the table schema is never changed." + + " With any option set, every Parquet file's footer is read and the allowed changes" + + " are committed before files are registered, so stats cover every file. A file" + + " needing a change that is not allowed is incompatible: see" + + " incompatible_schema_handling. Only Parquet files can be checked: with options" + + " set, ORC and Avro files are routed to the error output unless" + + " unverifiable_file_handling is ACCEPT. Requires a batch pipeline; streaming with" + + " schema evolution is not yet supported.") + public abstract @Nullable List getSchemaEvolutionOptions(); + + @SchemaFieldDescription( + "Columns (dotted paths for nested fields) that must stay required whatever the options" + + " say. A file that lacks such a column or holds nulls in it is routed to the error" + + " output; so is one whose footer has no null-count statistics for it, unless" + + " unverifiable_file_handling is ACCEPT. Only meaningful with" + + " schema_evolution_options.") + public abstract @Nullable List getRequiredColumns(); + + @SchemaFieldDescription( + "What to do when a file's schema is incompatible with the table (needs a change that" + + " is not allowed, or conflicts with the table or another file): FAIL_PIPELINE" + + " fails the pipeline before any schema change is committed; ROUTE_TO_ERRORS skips" + + " the schema and routes its files to the error output. Default: FAIL_PIPELINE.") + public abstract @Nullable String getIncompatibleSchemaHandling(); + + @SchemaFieldDescription( + "What to do with a file the per-file checks cannot verify: an ORC or Avro file (the" + + " checks read Parquet footers), 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 routes it to the error output; ACCEPT registers it unchecked, counted" + + " and logged. A file that fails a check is always routed. An accepted file that" + + " lacks a required column or holds nulls in it breaks reads of the table, not" + + " registration. Default: REJECT.") + public abstract @Nullable String getUnverifiableFileHandling(); + @SchemaFieldDescription("This option specifies whether and where to output unwritable rows.") public abstract @Nullable ErrorHandling getErrorHandling(); @@ -140,9 +183,74 @@ 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; + } + 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) { + builder = + builder.setIncompatibleSchemaHandling( + parseEnum( + SchemaEvolutionConfig.IncompatibleSchemaHandling.class, + handlingName, + "incompatible_schema_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, java.util.Arrays.toString(type.getEnumConstants()))); + } + public IcebergCatalogConfig getIcebergCatalog() { return IcebergCatalogConfig.builder() .setCatalogProperties(getCatalogProperties()) @@ -186,7 +294,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/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..0f92fa7f816f --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java @@ -0,0 +1,155 @@ +/* + * 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.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 testNoEvolutionSettingsGiveNull() { + 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") + .build() + .getSchemaEvolution(); + assertNotNull(config); + assertTrue(config.isPinned("address.city")); + assertEquals( + SchemaEvolutionConfig.IncompatibleSchemaHandling.ROUTE_TO_ERRORS, + config.getIncompatibleSchemaHandling()); + } + + @Test + public void testUnverifiableFileHandlingParsed() { + 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 testUnverifiableFileHandlingWithoutOptionsRejected() { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> base().setUnverifiableFileHandling("ACCEPT").build().getSchemaEvolution()); + assertTrue(e.getMessage(), e.getMessage().contains("schema_evolution_options")); + assertThrows( + IllegalArgumentException.class, + () -> + base() + .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION")) + .setUnverifiableFileHandling("trust") + .build() + .getSchemaEvolution()); + } + + @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 testInvalidHandlingRejected() { + assertThrows( + IllegalArgumentException.class, + () -> + base() + .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION")) + .setIncompatibleSchemaHandling("ignore") + .build() + .getSchemaEvolution()); + } + + @Test + public void testPinsWithoutOptionsRejected() { + IllegalArgumentException e = + assertThrows( + IllegalArgumentException.class, + () -> base().setRequiredColumns(Arrays.asList("id")).build().getSchemaEvolution()); + assertTrue(e.getMessage(), e.getMessage().contains("schema_evolution_options")); + } +} 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} From d143d0171dac0924eebe81e03990c984b9c06d95 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 07:32:21 -0400 Subject: [PATCH 3/8] trigger tests --- .github/trigger_files/IO_Iceberg_Integration_Tests.json | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 } From ee2734f32940383d78e655dc2309165304afb188 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 07:50:44 -0400 Subject: [PATCH 4/8] simplify arg --- .../java/org/apache/beam/sdk/io/iceberg/AddFiles.java | 7 ++++++- .../beam/sdk/io/iceberg/SchemaEvolutionConfig.java | 6 ++++-- .../sdk/io/iceberg/SchemaEvolutionConfigTest.java | 11 ++++++++--- 3 files changed, 18 insertions(+), 6 deletions(-) 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 25880e847e76..9e62115578e5 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 @@ -43,6 +43,7 @@ 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; @@ -330,6 +331,10 @@ private PCollection gateOnSchemaCommit(PCollection input) { 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())) @@ -344,7 +349,7 @@ private PCollection gateOnSchemaCommit(PCollection input) { catalogConfig, tableIdentifier, evolution, - evolution.incompatibleSchemaHandling(true), + onIncompatible, creation, committer))); return windowed.apply("WaitForSchemaCommit", Wait.on(signal)); 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/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)); } } From 2450570a35a9a9da9d91d4b154b59f5d0acc18a4 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 08:01:32 -0400 Subject: [PATCH 5/8] tests --- .../AddFilesSchemaTransformProviderTest.java | 45 ++-- .../beam/sdk/io/iceberg/AddFilesTest.java | 209 ++++++++---------- 2 files changed, 118 insertions(+), 136 deletions(-) diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java index 0f92fa7f816f..f5c4fd9dade9 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java @@ -42,7 +42,7 @@ private static Configuration.Builder base() { } @Test - public void testNoEvolutionSettingsGiveNull() { + public void testNoEvolutionSettingsMeanEvolutionDisabled() { assertNull(base().build().getSchemaEvolution()); assertNull( base().setSchemaEvolutionOptions(Collections.emptyList()).build().getSchemaEvolution()); @@ -81,7 +81,7 @@ public void testPinsAndHandlingParsed() { } @Test - public void testUnverifiableFileHandlingParsed() { + public void testUnverifiableFileHandlingParsedAndDefaultsToReject() { SchemaEvolutionConfig config = base() .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION")) @@ -101,23 +101,6 @@ public void testUnverifiableFileHandlingParsed() { .getUnverifiableFileHandling()); } - @Test - public void testUnverifiableFileHandlingWithoutOptionsRejected() { - IllegalArgumentException e = - assertThrows( - IllegalArgumentException.class, - () -> base().setUnverifiableFileHandling("ACCEPT").build().getSchemaEvolution()); - assertTrue(e.getMessage(), e.getMessage().contains("schema_evolution_options")); - assertThrows( - IllegalArgumentException.class, - () -> - base() - .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION")) - .setUnverifiableFileHandling("trust") - .build() - .getSchemaEvolution()); - } - @Test public void testInvalidOptionListsValidValues() { IllegalArgumentException e = @@ -133,7 +116,7 @@ public void testInvalidOptionListsValidValues() { } @Test - public void testInvalidHandlingRejected() { + public void testInvalidHandlingValuesRejected() { assertThrows( IllegalArgumentException.class, () -> @@ -142,14 +125,30 @@ public void testInvalidHandlingRejected() { .setIncompatibleSchemaHandling("ignore") .build() .getSchemaEvolution()); + assertThrows( + IllegalArgumentException.class, + () -> + base() + .setSchemaEvolutionOptions(Arrays.asList("ALLOW_FIELD_ADDITION")) + .setUnverifiableFileHandling("trust") + .build() + .getSchemaEvolution()); } + /** The config's "needs options" rule, reported under the YAML key the user must set. */ @Test - public void testPinsWithoutOptionsRejected() { - IllegalArgumentException e = + public void testSettingsWithoutOptionsRejectedNamingTheOptionsKey() { + IllegalArgumentException pins = assertThrows( IllegalArgumentException.class, () -> base().setRequiredColumns(Arrays.asList("id")).build().getSchemaEvolution()); - assertTrue(e.getMessage(), e.getMessage().contains("schema_evolution_options")); + 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 b6be74879501..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,6 +42,7 @@ 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; @@ -72,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; @@ -89,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; @@ -992,73 +995,86 @@ public CompositeBehavior enterCompositeTransform(TransformHierarchy.Node node) { 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 = root + "narrow.parquet"; - DataWriter writer = createWriter(narrow); - writer.write(record(1, "a", 1)); - writer.close(); + String narrow = writeOneRecord("narrow.parquet"); String wide = writeWider("wide.parquet"); PCollectionRowTuple output = - pipeline - .apply("Create Input", Create.of(narrow, wide)) - .apply(addFiles(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))); + 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); - Types.NestedField email = table.schema().findField("email"); - assertNotNull(email); - assertTrue(email.isOptional()); assertEquals(1, Iterables.size(table.snapshots())); - // the wide file's stats cover the new column - boolean sawEmailStats = false; - for (org.apache.iceberg.FileScanTask task : table.newScan().includeColumnStats().planFiles()) { - if (task.file().path().toString().endsWith("wide.parquet")) { - sawEmailStats = task.file().nullValueCounts().containsKey(email.fieldId()); - } - } - assertTrue("stats for the added column", sawEmailStats); + int emailId = table.schema().findField("email").fieldId(); + assertTrue( + "stats for the added column", nullCountsOf(table, "wide.parquet").containsKey(emailId)); } @Test - public void testEvolutionDisabledKeepsTodaysGraph() throws Exception { + public void testEvolutionDisabledAddsNoPrePassTransforms() throws Exception { catalog.createTable(tableId, icebergSchema); - String file = root + "data.parquet"; - DataWriter writer = createWriter(file); - writer.write(record(1, "a", 1)); - writer.close(); - pipeline.apply("Create Input", Create.of(file)).apply(addFiles(null)); + 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); - 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); - String bad = writeWithSchema("bad.parquet", conflicting, record); - String good = root + "good.parquet"; - DataWriter writer = createWriter(good); - writer.write(record(1, "a", 1)); - writer.close(); + 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())); } @@ -1066,27 +1082,15 @@ public void testIncompatibleSchemaFailsBatchPipelineByDefault() throws Exception @Test public void testIncompatibleSchemaRoutedToErrorsWhenConfigured() throws Exception { catalog.createTable(tableId, icebergSchema); - 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); - String bad = writeWithSchema("bad.parquet", conflicting, record); - String good = root + "good.parquet"; - DataWriter writer = createWriter(good); - writer.write(record(1, "a", 1)); - writer.close(); - + 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")) @@ -1097,7 +1101,9 @@ public void testIncompatibleSchemaRoutedToErrorsWhenConfigured() throws Exceptio 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())); @@ -1105,60 +1111,42 @@ public void testIncompatibleSchemaRoutedToErrorsWhenConfigured() throws Exceptio @Test public void testMissingTableIsCreatedFromTheFilesUnion() throws Exception { - String narrow = root + "narrow.parquet"; - DataWriter writer = createWriter(narrow); - writer.write(record(1, "a", 1)); - writer.close(); + String narrow = writeOneRecord("narrow.parquet"); String wide = writeWider("wide.parquet"); PCollectionRowTuple output = - pipeline - .apply("Create Input", Create.of(narrow, wide)) - .apply(addFiles(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))); + pipeline.apply("Create Input", Create.of(narrow, wide)).apply(addFiles(ADDITIONS)); PAssert.that(output.get("errors")).empty(); - pipeline.run().waitUntilFinish(); - Table table = catalog.loadTable(tableId); - assertNotNull(table.schema().findField("email")); - assertTrue("created columns are optional", table.schema().findField("id").isOptional()); - assertEquals(2, Iterables.size(table.newScan().planFiles())); - } - - private PCollection stream(String... files) { - TestStream.Builder builder = TestStream.create(StringUtf8Coder.of()); - for (String file : files) { - // far in the past: only the forward stamp keeps these on time - builder = builder.addElements(TimestampedValue.of(file, new Instant(0))); - builder = builder.advanceProcessingTime(Duration.standardSeconds(1)); - } - return pipeline.apply(builder.advanceWatermarkToInfinity()); - } + pipeline.run().waitUntilFinish(); - private AddFiles streamingAddFiles(SchemaEvolutionConfig config) { - return new AddFiles( - catalogConfig, - tableId.toString(), - null, - null, - null, - null, - 10, - Duration.standardSeconds(5), - config); + 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 paths = stream("unused.parquet"); + 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, - () -> - paths.apply( - streamingAddFiles( - SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)))); + assertThrows(IllegalArgumentException.class, () -> unbounded.apply(streaming)); + assertThat(e.getMessage(), containsString("not yet supported for unbounded input")); } @@ -1169,10 +1157,7 @@ public void testUnboundedInputWithEvolutionIsRejected() { @Test public void testUpstreamWindowedBatchInputEvolvesAndRegisters() throws Exception { catalog.createTable(tableId, icebergSchema); - String narrow = root + "narrow.parquet"; - DataWriter writer = createWriter(narrow); - writer.write(record(1, "a", 1)); - writer.close(); + String narrow = writeOneRecord("narrow.parquet"); String wide = writeWider("wide.parquet"); PCollectionRowTuple output = @@ -1183,42 +1168,40 @@ public void testUpstreamWindowedBatchInputEvolvesAndRegisters() throws Exception TimestampedValue.of(narrow, new Instant(0)), TimestampedValue.of(wide, new Instant(60_000)))) .apply("UpstreamWindow", Window.into(FixedWindows.of(Duration.standardSeconds(30)))) - .apply(addFiles(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))); + .apply(addFiles(ADDITIONS)); PAssert.that(output.get("errors")).empty(); + pipeline.run().waitUntilFinish(); - Table table = catalog.loadTable(tableId); - assertNotNull(table.schema().findField("email")); - assertEquals(2, Iterables.size(table.newScan().planFiles())); + assertEmailAddedAndFilesRegistered(2); } /** - * Async results are emitted in FinishBundle with their own timestamps. Emitting a finished result - * during a later element's ProcessElement would attribute it to that element and trip the - * runner's output-timestamp skew check when timestamps differ. + * 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"); - java.util.concurrent.atomic.AtomicInteger attempts = - new java.util.concurrent.atomic.AtomicInteger(); - CommitSchemaUnion.Committer flaky = + AtomicInteger attempts = new AtomicInteger(); + CommitSchemaUnion.Committer failsOnce = txn -> { if (attempts.incrementAndGet() == 1) { - throw new org.apache.iceberg.exceptions.CommitFailedException("transient"); + throw new CommitFailedException("transient"); } txn.commitTransaction(); }; + PCollectionRowTuple output = pipeline .apply("Create Input", Create.of(wide)) - .apply( - addFiles(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) - .withSchemaCommitter(flaky)); + .apply(addFiles(ADDITIONS).withSchemaCommitter(failsOnce)); PAssert.that(output.get("errors")).empty(); + pipeline.run().waitUntilFinish(); - assertNotNull(catalog.loadTable(tableId).schema().findField("email")); + + assertEmailAddedAndFilesRegistered(1); } // ---- ConvertToDataFile coverage check and pinned columns From a3cdea9bb02ba99d73be59bfdcfad9489dbf4514 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 08:16:43 -0400 Subject: [PATCH 6/8] fixes --- CHANGES.md | 1 + .../apache/beam/sdk/io/iceberg/AddFiles.java | 11 ++-- .../AddFilesSchemaTransformProvider.java | 50 +++++++++++-------- .../sdk/io/iceberg/SchemaEvolutionOption.java | 5 +- .../AddFilesSchemaTransformProviderTest.java | 17 +++++++ 5 files changed, 56 insertions(+), 28 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index aecd194e1f22..c055d22e5e96 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -65,6 +65,7 @@ ## I/Os * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). +* [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). ## 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 9e62115578e5..31d77f8d5a76 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 @@ -140,15 +140,15 @@ * 

{@code
  * SchemaEvolutionConfig evolution =
  *     SchemaEvolutionConfig.builder()
- *         .setOptions(List.of(ALLOW_FIELD_ADDITION, ALLOW_TYPE_PROMOTION))
- *         .setRequiredColumns(List.of("id"))
+ *         .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 + *

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 the table unreadable through Iceberg's reader. + * make that file, and any scan that includes it, fail in Iceberg's reader. */ public class AddFiles extends PTransform, PCollectionRowTuple> { static final String OUTPUT_TAG = "snapshots"; @@ -208,8 +208,7 @@ public AddFiles( * @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, and the width of the - * schema pre-pass window + * @param intervalTrigger streaming only: how often manifests are committed * @param evolution schema evolution settings; null or no options means the schema is never * changed */ 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 b0610d6d55ea..183c19c62df7 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,6 +24,7 @@ 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; @@ -125,19 +126,20 @@ public static Builder builder() { + " a file may contain nulls in it or lacks it), ALLOW_TYPE_PROMOTION (widen a column" + " type, e.g. int to long). Empty or absent: the table schema is never changed." + " With any option set, every Parquet file's footer is read and the allowed changes" - + " are committed before files are registered, so stats cover every file. A file" - + " needing a change that is not allowed is incompatible: see" - + " incompatible_schema_handling. Only Parquet files can be checked: with options" - + " set, ORC and Avro files are routed to the error output unless" - + " unverifiable_file_handling is ACCEPT. Requires a batch pipeline; streaming with" - + " schema evolution is not yet supported.") + + " are committed before files are registered, so every registered file carries stats" + + " for every column it has. A file needing a change that is not allowed is" + + " incompatible: see incompatible_schema_handling. Only Parquet files can be checked:" + + " with options set, ORC and Avro files are routed to the error output unless" + + " unverifiable_file_handling is ACCEPT. Routed files reach the error output only" + + " when error_handling is set. Requires a batch pipeline; streaming with schema" + + " evolution is not yet supported.") public abstract @Nullable List getSchemaEvolutionOptions(); @SchemaFieldDescription( "Columns (dotted paths for nested fields) that must stay required whatever the options" + " say. A file that lacks such a column or holds nulls in it is routed to the error" - + " output; so is one whose footer has no null-count statistics for it, unless" - + " unverifiable_file_handling is ACCEPT. Only meaningful with" + + " output (see error_handling); so is one whose footer has no null-count statistics" + + " for it, unless unverifiable_file_handling is ACCEPT. Only meaningful with" + " schema_evolution_options.") public abstract @Nullable List getRequiredColumns(); @@ -145,17 +147,18 @@ public static Builder builder() { "What to do when a file's schema is incompatible with the table (needs a change that" + " is not allowed, or conflicts with the table or another file): FAIL_PIPELINE" + " fails the pipeline before any schema change is committed; ROUTE_TO_ERRORS skips" - + " the schema and routes its files to the error output. Default: FAIL_PIPELINE.") + + " the schema and routes its files to the error output, so it requires" + + " error_handling. Default: FAIL_PIPELINE.") public abstract @Nullable String getIncompatibleSchemaHandling(); @SchemaFieldDescription( "What to do with a file the per-file checks cannot verify: an ORC or Avro file (the" + " checks read Parquet footers), 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 routes it to the error output; ACCEPT registers it unchecked, counted" - + " and logged. A file that fails a check is always routed. An accepted file that" - + " lacks a required column or holds nulls in it breaks reads of the table, not" - + " registration. Default: REJECT.") + + " map). REJECT routes it to the error output (see error_handling); ACCEPT registers" + + " it unchecked, counted and logged. A file that fails a check is always routed. An" + + " accepted file that lacks a required column or holds nulls in it breaks reads of" + + " the table, not registration. Default: REJECT.") public abstract @Nullable String getUnverifiableFileHandling(); @SchemaFieldDescription("This option specifies whether and where to output unwritable rows.") @@ -208,6 +211,7 @@ public abstract static class Builder { 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" @@ -221,12 +225,18 @@ public abstract static class Builder { builder = builder.setRequiredColumns(new LinkedHashSet<>(pins)); } if (handlingName != null) { - builder = - builder.setIncompatibleSchemaHandling( - parseEnum( - SchemaEvolutionConfig.IncompatibleSchemaHandling.class, - handlingName, - "incompatible_schema_handling")); + 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 = @@ -248,7 +258,7 @@ private static > T parseEnum(Class type, String name, Strin throw new IllegalArgumentException( String.format( "Invalid %s value '%s'. Valid values: %s", - option, name, java.util.Arrays.toString(type.getEnumConstants()))); + option, name, Arrays.toString(type.getEnumConstants()))); } public IcebergCatalogConfig getIcebergCatalog() { 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/AddFilesSchemaTransformProviderTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java index f5c4fd9dade9..2e469ef6da71 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesSchemaTransformProviderTest.java @@ -27,6 +27,7 @@ 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; @@ -71,6 +72,7 @@ public void testPinsAndHandlingParsed() { .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); @@ -135,6 +137,21 @@ public void testInvalidHandlingValuesRejected() { .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() { From cc923e9ccf6e085ed51b2d5c705b2b00b49a7d04 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 08:18:35 -0400 Subject: [PATCH 7/8] changes --- CHANGES.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGES.md b/CHANGES.md index c055d22e5e96..47da5f81fb02 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -65,7 +65,6 @@ ## I/Os * Support for X source added (Java/Python) ([#X](https://github.com/apache/beam/issues/X)). -* [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). ## New Features / Improvements @@ -100,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 From 35adfc14e19373e9671c4d1444d866da3c5fdb24 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 17 Sep 2026 12:26:36 -0400 Subject: [PATCH 8/8] comments, refresh only once per bundle --- .../apache/beam/sdk/io/iceberg/AddFiles.java | 18 +++-- .../AddFilesSchemaTransformProvider.java | 68 ++++++++++--------- 2 files changed, 48 insertions(+), 38 deletions(-) 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 31d77f8d5a76..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,6 +38,7 @@ 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; @@ -148,7 +149,7 @@ * *

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 Iceberg's reader. + * 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"; @@ -394,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; @@ -514,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 @@ -732,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); @@ -741,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 183c19c62df7..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 @@ -71,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( @@ -120,48 +120,52 @@ public static Builder builder() { public abstract @Nullable List getSortFields(); @SchemaFieldDescription( - "Schema changes the transform may make so that every file's columns are covered by the" - + " table schema, from: ALLOW_FIELD_ADDITION (add file columns absent from the table," - + " as optional), ALLOW_FIELD_RELAXATION (make a required table column optional when" - + " a file may contain nulls in it or lacks it), ALLOW_TYPE_PROMOTION (widen a column" - + " type, e.g. int to long). Empty or absent: the table schema is never changed." - + " With any option set, every Parquet file's footer is read and the allowed changes" - + " are committed before files are registered, so every registered file carries stats" - + " for every column it has. A file needing a change that is not allowed is" - + " incompatible: see incompatible_schema_handling. Only Parquet files can be checked:" - + " with options set, ORC and Avro files are routed to the error output unless" - + " unverifiable_file_handling is ACCEPT. Routed files reach the error output only" - + " when error_handling is set. Requires a batch pipeline; streaming with schema" - + " evolution is not yet supported.") + "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 (dotted paths for nested fields) that must stay required whatever the options" - + " say. A file that lacks such a column or holds nulls in it is routed to the error" - + " output (see error_handling); so is one whose footer has no null-count statistics" - + " for it, unless unverifiable_file_handling is ACCEPT. Only meaningful with" - + " schema_evolution_options.") + "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 to do when a file's schema is incompatible with the table (needs a change that" - + " is not allowed, or conflicts with the table or another file): FAIL_PIPELINE" - + " fails the pipeline before any schema change is committed; ROUTE_TO_ERRORS skips" - + " the schema and routes its files to the error output, so it requires" - + " error_handling. Default: FAIL_PIPELINE.") + "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 to do with a file the per-file checks cannot verify: an ORC or Avro file (the" - + " checks read Parquet footers), 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 routes it to the error output (see error_handling); ACCEPT registers" - + " it unchecked, counted and logged. A file that fails a check is always routed. An" - + " accepted file that lacks a required column or holds nulls in it breaks reads of" - + " the table, not registration. Default: REJECT.") + "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("This option specifies whether and where to output unwritable rows.") + @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