From f5aa76ca22e3f056eda517f93b31bf0735c1fbed Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 09:24:00 -0400 Subject: [PATCH] AddFiles: create the table from the file schemas when it does not exist Before this change the pre-pass required an existing table. AddFiles without evolution creates a missing table from whichever file happens to reach ConvertToDataFile first; with the pre-pass we can do better, since every schema of the window is known before anything is created. Create path (CommitSchemaUnion.create), taken when loadTable throws NoSuchTableException: - The window's schemas are folded into one union on a scratch create transaction that is never committed (the same staging loop as the evolve path, so a schema conflicting with another is reported or skipped per the handling, and the create is abandoned under FAIL_PIPELINE before anything exists). - The real table is built directly from createdSchema(union, config): every column optional at every level - one lucky file's declared or proven required-ness must not become a table constraint that the next file violates - except pinned paths and their ancestors, which are created required. Creation is the schema-authoring moment; evolution never tightens columns afterwards. A pin that no file schema carries fails the create under FAIL_PIPELINE (later windows could only add the column optional, so the pin would stay inert forever) and warns under ROUTE_TO_ERRORS. The created table is born with a single schema version. - Creation is not gated on any particular evolution option: the options guard an existing table's schema, and there is none to guard yet. - Partition spec, sort order and table properties come from TableCreation (the AddFiles constructor arguments), resolved against the union, so a partition or sort column carried by any schema of the window works; the name mapping is written in the same transaction. - Create race: two workers (or a concurrent non-evolution AddFiles) may create the table at once. AlreadyExistsException joins CommitFailedException in the retry, and the next attempt takes the evolve path against the table the other party created. - Empty window against a missing table returns NO_TABLE (-1) and creates nothing. --- .../beam/sdk/io/iceberg/CommitSchemaOnce.java | 89 ++++ .../sdk/io/iceberg/CommitSchemaUnion.java | 464 ++++++++++++++---- .../sdk/io/iceberg/CommitSchemaOnceTest.java | 157 ++++++ .../sdk/io/iceberg/CommitSchemaUnionTest.java | 392 +++++++++++++++ 4 files changed, 994 insertions(+), 108 deletions(-) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java new file mode 100644 index 000000000000..6fd5171c2eae --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnce.java @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.metrics.Metrics.counter; + +import java.util.List; +import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling; +import org.apache.beam.sdk.metrics.Counter; +import org.apache.beam.sdk.transforms.DoFn; +import org.apache.iceberg.catalog.Catalog; +import org.apache.iceberg.catalog.TableIdentifier; +import org.checkerframework.checker.nullness.qual.MonotonicNonNull; + +/** + * Commits one schema union per window (the combine output is one element per window) and emits the + * resulting schema id as the signal for the Wait.on gate ahead of file registration. + */ +class CommitSchemaOnce extends DoFn, Long> { + static final String COMMITS_COUNTER = "numSchemaCommits"; + private static final Counter numSchemaCommits = counter(CommitSchemaOnce.class, COMMITS_COUNTER); + + private final IcebergCatalogConfig catalogConfig; + private final String identifier; + private final SchemaEvolutionConfig config; + private final IncompatibleSchemaHandling handling; + private final CommitSchemaUnion.TableCreation creation; + private final CommitSchemaUnion.Committer committer; + private transient @MonotonicNonNull Catalog catalog; + + CommitSchemaOnce( + IcebergCatalogConfig catalogConfig, + String identifier, + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling, + CommitSchemaUnion.TableCreation creation) { + this( + catalogConfig, identifier, config, handling, creation, CommitSchemaUnion.DEFAULT_COMMITTER); + } + + CommitSchemaOnce( + IcebergCatalogConfig catalogConfig, + String identifier, + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling, + CommitSchemaUnion.TableCreation creation, + CommitSchemaUnion.Committer committer) { + this.catalogConfig = catalogConfig; + this.identifier = identifier; + this.config = config; + this.handling = handling; + this.creation = creation; + this.committer = committer; + } + + @ProcessElement + public void process( + @Element List schemas, OutputReceiver out) { + if (catalog == null) { + catalog = catalogConfig.catalog(); + } + TableIdentifier tableId = IcebergUtils.parseTableIdentifier(identifier); + // The committer runs only when a transaction is actually committed, so wrapping it counts + // real commits and skips no-op windows. + CommitSchemaUnion.Committer counting = + txn -> { + committer.commit(txn); + numSchemaCommits.inc(); + }; + long schemaId = + CommitSchemaUnion.commit(catalog, tableId, schemas, config, handling, creation, counting); + out.output(schemaId); + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java index 8ffc4780f6e3..170d5c3d4150 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java @@ -21,7 +21,10 @@ import java.io.Serializable; import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; import java.util.List; +import java.util.Map; import java.util.Set; import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling; import org.apache.beam.sdk.util.BackOff; @@ -36,7 +39,9 @@ import org.apache.iceberg.UpdateSchema; import org.apache.iceberg.catalog.Catalog; import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.exceptions.AlreadyExistsException; import org.apache.iceberg.exceptions.CommitFailedException; +import org.apache.iceberg.exceptions.NoSuchTableException; import org.apache.iceberg.exceptions.ValidationException; import org.apache.iceberg.mapping.NameMapping; import org.apache.iceberg.types.Type; @@ -48,12 +53,17 @@ import org.slf4j.LoggerFactory; /** - * Applies the distinct file schemas of a window to the table in one transaction: fresh load, - * classify each schema most common first, fold the allowed unions (plus explicit relaxations for - * required columns absent from files) on a scratch transaction, stage the folded result as one - * schema update, repair the name mapping, commit once. The fold keeps per-schema blame for - * cross-schema conflicts while the table gains a single schema version per window; the scratch - * transaction is never committed. Nothing is committed when nothing changes. + * Applies the distinct file schemas of a window to the table in one commit, in phases named by the + * methods of this class: {@code classify} each schema against a fresh load of the table (most + * common first), {@code fold} the accepted ones into a single union on scratch transactions that + * are never committed, {@code replay} the folded result onto the real transaction as one schema + * update, repair the name mapping, commit once. The fold keeps per-schema blame for cross-schema + * conflicts while the table gains a single schema version per window. Nothing is committed when + * nothing changes. + * + *

When the table does not exist, {@code create} builds it instead: {@code foldForCreate} + * computes the same union, and the table is born from it directly with pinned columns and their + * ancestors required. * *

Incompatible schemas either fail the whole call before any commit ({@link * IncompatibleSchemaHandling#FAIL_PIPELINE}) or are skipped so their files reach the error output @@ -64,6 +74,25 @@ final class CommitSchemaUnion { static final int MAX_ATTEMPTS = 5; + /** Returned when the table does not exist and there is no schema to create it from. */ + static final long NO_TABLE = -1L; + + /** How to create the table when it does not exist: from the union of the window's schemas. */ + static final class TableCreation implements Serializable { + final @Nullable List partitionFields; + final @Nullable List sortFields; + final @Nullable Map properties; + + TableCreation( + @Nullable List partitionFields, + @Nullable List sortFields, + @Nullable Map properties) { + this.partitionFields = partitionFields; + this.sortFields = sortFields; + this.properties = properties; + } + } + /** Injectable so tests can exercise the commit retry path. */ interface Committer extends Serializable { void commit(Transaction txn); @@ -78,6 +107,22 @@ static final class IncompatibleSchemaException extends IllegalStateException { } } + private static final class Accepted { + final Schema schema; + final String json; + final long files; + + /** Null on the create path: the seed table is empty, so there is nothing to relax. */ + final @Nullable SchemaDelta delta; + + Accepted(Schema schema, String json, long files, @Nullable SchemaDelta delta) { + this.schema = schema; + this.json = json; + this.files = files; + this.delta = delta; + } + } + private static final class Incompatible { final String schemaJson; final long files; @@ -111,7 +156,8 @@ private static String truncate(String json) { private CommitSchemaUnion() {} /** - * Applies the schemas and returns the table's schema id after the call. + * Applies the schemas and returns the table's schema id after the call, or {@link #NO_TABLE} when + * the table is missing and there is no schema to create it from. * * @param schemas the window's distinct schema groups, most common first */ @@ -121,6 +167,7 @@ static long commit( List schemas, SchemaEvolutionConfig config, IncompatibleSchemaHandling handling, + TableCreation creation, Committer committer) { // The catalog is already under contention when a retry fires; back off (jittered by // FluentBackoff) instead of piling on. Iceberg's own metadata retries (commit.retry.*) @@ -133,8 +180,9 @@ static long commit( .backoff(); for (int attempt = 1; ; attempt++) { try { - return commitOnce(catalog, tableId, schemas, config, handling, committer); - } catch (CommitFailedException e) { + return commitOnce(catalog, tableId, schemas, config, handling, creation, committer); + } catch (CommitFailedException | AlreadyExistsException e) { + // a concurrent commit, or a create race: the next attempt loads the fresh state try { if (!BackOffUtils.next(Sleeper.DEFAULT, backoff)) { throw e; @@ -159,78 +207,31 @@ private static long commitOnce( List schemas, SchemaEvolutionConfig config, IncompatibleSchemaHandling handling, + TableCreation creation, Committer committer) { - Table table = catalog.loadTable(tableId); + Table table; + try { + table = catalog.loadTable(tableId); + } catch (NoSuchTableException e) { + return create(catalog, tableId, schemas, config, handling, creation, committer); + } // Every transaction below must share this snapshot: classification, the fold and the replay // all reason about the same table state (newTransactionOn enforces it). Schema base = table.schema(); - List incompatible = new ArrayList<>(); - List accepted = new ArrayList<>(); - for (CollectDistinctSchemas.SchemaGroup group : schemas) { - Schema fileSchema = - FileSchemas.markRequired( - SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns()); - SchemaDelta delta = SchemaDelta.classify(table, fileSchema); - if (delta.isEmpty()) { - continue; - } - if (!delta.allowedBy(config)) { - incompatible.add( - new Incompatible( - group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config))); - continue; - } - accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta)); - } - Transaction scratch = stageAll(table, base, tableId, accepted, incompatible); - boolean folded = !accepted.isEmpty(); - if (folded) { - relaxNewRequiredFields(scratch, base); - } + List incompatible = new ArrayList<>(); + List accepted = classify(table, schemas, config, incompatible); + Schema merged = fold(table, base, tableId, accepted, incompatible); Transaction txn = newTransactionOn(table, base, tableId); - if (folded) { - Schema merged = scratch.table().schema(); - // One union replays the fold's net effect (additions, promotions, relaxations) so the - // table gains a single schema version instead of one per folded schema. - txn.updateSchema().unionByNameWith(merged).commit(); - // toString of the args runs only on failure - Schema foldResult = TypeUtil.assignIncreasingFreshIds(merged); - Schema replayResult = TypeUtil.assignIncreasingFreshIds(txn.table().schema()); - checkState( - replayResult.sameSchema(foldResult), - "replaying the folded schema union for %s diverged from the fold; fold: %s replay: %s", - tableId, - foldResult, - replayResult); + if (merged != null) { + replay(txn, merged, tableId); } - boolean staged = folded; + boolean staged = merged != null; staged |= stageNameMapping(txn); if (!incompatible.isEmpty()) { - long files = 0; - for (Incompatible item : incompatible) { - files += item.files; - } - if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) { - throw new IncompatibleSchemaException( - "Incompatible schemas for " - + tableId - + " (" - + incompatible.size() - + " schema(s), " - + files - + " file(s)); no schema change was committed:\n " - + joinLines(incompatible)); - } - LOG.warn( - "Skipping {} incompatible schema(s) ({} file(s)) for {}; their files will be routed to" - + " the error output:\n {}", - incompatible.size(), - files, - tableId, - joinLines(incompatible)); + reportIncompatible(tableId, incompatible, handling, "no schema change was committed"); } if (!staged) { @@ -255,57 +256,302 @@ private static long commitOnce( return table.schema().schemaId(); } - private static final class Accepted { - final Schema schema; - final String json; - final long files; - final SchemaDelta delta; - - Accepted(Schema schema, String json, long files, SchemaDelta delta) { - this.schema = schema; - this.json = json; - this.files = files; - this.delta = delta; + /** + * Sorts the window's schemas into the ones the table must change for (accepted) and the ones it + * must not ({@code incompatible}, with the reason); schemas the table already covers drop out. + */ + private static List classify( + Table table, + List schemas, + SchemaEvolutionConfig config, + List incompatible) { + List accepted = new ArrayList<>(); + for (CollectDistinctSchemas.SchemaGroup group : schemas) { + Schema fileSchema = + FileSchemas.markRequired( + SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns()); + SchemaDelta delta = SchemaDelta.classify(table, fileSchema); + if (delta.isEmpty()) { + continue; + } + if (!delta.allowedBy(config)) { + incompatible.add( + new Incompatible( + group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config))); + continue; + } + accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta)); } + return accepted; } /** - * Folds one union per accepted schema into a scratch transaction the caller must never commit; - * its intermediate schema versions exist only in memory. A schema can conflict with another - * schema's additions, which only surfaces while staging and poisons the transaction, so on a - * conflict the offender moves to {@code incompatible} and the transaction is rebuilt without it. + * Unions the accepted schemas into the table schema on scratch transactions that are never + * committed, relaxing every field the window adds; a schema that conflicts with another only + * surfaces here, moves to {@code incompatible} and the fold restarts without it. Returns the + * folded schema, or null when nothing needs to change. */ - private static Transaction stageAll( + private static @Nullable Schema fold( Table table, Schema base, TableIdentifier tableId, List accepted, List incompatible) { while (true) { - Transaction txn = newTransactionOn(table, base, tableId); - Accepted failed = null; - for (Accepted item : accepted) { - // Both caught types carry staging conflicts: ValidationException from Schema - // construction at apply ("multiple fields for name"), IllegalArgumentException from - // SchemaUpdate preconditions ("Cannot change column type"). - try { - stage(txn, item); - } catch (ValidationException | IllegalArgumentException e) { - failed = item; - incompatible.add( - new Incompatible( - item.json, - item.files, - "conflicts with another file schema in the same window: " - + AddFiles.errorMessage(e))); - break; - } + Transaction scratch = newTransactionOn(table, base, tableId); + Accepted failed = stageAll(scratch, accepted, incompatible); + if (failed != null) { + accepted.remove(failed); + continue; + } + if (accepted.isEmpty()) { + return null; } + relaxNewRequiredFields(scratch, base); + return scratch.table().schema(); + } + } + + /** + * One union replays the fold's net effect (additions, promotions, relaxations) so the table gains + * a single schema version instead of one per folded schema. The checkState is a pure bug + * detector: concurrent changes are caught earlier, by newTransactionOn. + */ + private static void replay(Transaction txn, Schema merged, TableIdentifier tableId) { + txn.updateSchema().unionByNameWith(merged).commit(); + // toString of the args runs only on failure + Schema foldResult = TypeUtil.assignIncreasingFreshIds(merged); + Schema replayResult = TypeUtil.assignIncreasingFreshIds(txn.table().schema()); + checkState( + replayResult.sameSchema(foldResult), + "replaying the folded schema union for %s diverged from the fold; fold: %s replay: %s", + tableId, + foldResult, + replayResult); + } + + /** + * Creates the table from the union of the window's schemas, with every column optional at every + * level so that one lucky file cannot impose required columns on the table - except pinned + * columns and their ancestors, which are created required. + */ + private static long create( + Catalog catalog, + TableIdentifier tableId, + List schemas, + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling, + TableCreation creation, + Committer committer) { + if (schemas.isEmpty()) { + LOG.info("Table {} does not exist and no file schema was read; not creating it", tableId); + return NO_TABLE; + } + List incompatible = new ArrayList<>(); + Schema merged = foldForCreate(catalog, tableId, schemas, incompatible); + if (!incompatible.isEmpty()) { + reportIncompatible(tableId, incompatible, handling, "no table was created"); + } + // The real table is built from the folded result directly. + Schema created = createdSchema(merged, config); + reportUnenforceablePins(tableId, created, config, handling); + Map properties = + creation.properties == null ? new HashMap<>() : new HashMap<>(creation.properties); + Transaction txn = + catalog + .buildTable(tableId, created) + .withPartitionSpec(PartitionUtils.toPartitionSpec(creation.partitionFields, created)) + .withSortOrder(SortOrderUtils.toSortOrder(creation.sortFields, created)) + .withProperties(properties) + .createTransaction(); + stageNameMapping(txn); + committer.commit(txn); + Table table = catalog.loadTable(tableId); + LOG.info( + "Created table {} from {} file schema(s), schema id {}", + tableId, + schemas.size() - incompatible.size(), + table.schema().schemaId()); + return table.schema().schemaId(); + } + + /** + * Unions the window's schemas into one on scratch create transactions that are never committed, + * seeded by the most common schema; conflicts move to {@code incompatible} and the fold restarts + * without the offender. + */ + private static Schema foldForCreate( + Catalog catalog, + TableIdentifier tableId, + List schemas, + List incompatible) { + Schema seed = SchemaParser.fromJson(schemas.get(0).getSchemaJson()); + List rest = new ArrayList<>(); + for (CollectDistinctSchemas.SchemaGroup group : schemas.subList(1, schemas.size())) { + Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson()); + rest.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), null)); + } + while (true) { + Transaction scratch = catalog.buildTable(tableId, seed).createTransaction(); + Accepted failed = stageAll(scratch, rest, incompatible); if (failed == null) { - return txn; + return scratch.table().schema(); + } + rest.remove(failed); + } + } + + /** + * A pin the created schema did not end up enforcing - the column appears in no file schema, or + * the configured spelling resolves to a field the pin walk did not reach (a short container + * spelling like a.b for a.element.b, or a path inside a map key) - would stay inert forever, + * since later windows only add columns optional: a config error under FAIL_PIPELINE, a warning + * under ROUTE_TO_ERRORS (streaming may see the column later). + */ + private static void reportUnenforceablePins( + TableIdentifier tableId, + Schema created, + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling) { + List unenforceable = new ArrayList<>(); + for (String pin : config.getRequiredColumns()) { + Types.NestedField field = created.findField(pin); + if (field == null || field.isOptional()) { + unenforceable.add(pin); + } + } + if (unenforceable.isEmpty()) { + return; + } + Collections.sort(unenforceable); + if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) { + throw new IncompatibleSchemaException( + "Pinned column(s) " + + unenforceable + + " appear in none of the file schemas creating " + + tableId + + ", or their spelling does not match the column path; the created table cannot" + + " make them required"); + } + LOG.warn( + "Pinned column(s) {} appear in none of the file schemas creating {}, or their spelling" + + " does not match the column path; the created table cannot make them required", + unenforceable, + tableId); + } + + /** + * The created schema: every field optional at every level, list elements and map values included, + * except pinned paths and their ancestors, which stay required so the schema advertises the + * guarantee the per-file pin check enforces (a null ancestor nulls the pinned leaf). Map key + * subtrees keep their declared shape (keys are required by definition; pins inside them are not + * honored). Nothing depends on a created table's schema yet, so this is the schema-authoring + * moment; evolution never tightens columns afterwards. + */ + static Schema createdSchema(Schema merged, SchemaEvolutionConfig config) { + Pins pins = new Pins(config.getRequiredColumns()); + List fields = new ArrayList<>(); + for (Types.NestedField field : merged.asStruct().fields()) { + fields.add(createdField(field, field.name(), pins)); + } + return new Schema(fields); + } + + private static Types.NestedField createdField(Types.NestedField field, String path, Pins pins) { + boolean required = pins.isPinned(path) || pins.pinnedColumnBeneath(path) != null; + return Types.NestedField.from(field) + .ofType(createdType(field.type(), path, pins)) + .isOptional(!required) + .build(); + } + + private static Type createdType(Type type, String path, Pins pins) { + if (type.isStructType()) { + List fields = new ArrayList<>(); + for (Types.NestedField field : type.asStructType().fields()) { + fields.add(createdField(field, path + "." + field.name(), pins)); + } + return Types.StructType.of(fields); + } + if (type.isListType()) { + Types.ListType list = type.asListType(); + String elementPath = path + ".element"; + Type elementType = createdType(list.elementType(), elementPath, pins); + boolean required = + pins.isPinned(elementPath) || pins.pinnedColumnBeneath(elementPath) != null; + return required + ? Types.ListType.ofRequired(list.elementId(), elementType) + : Types.ListType.ofOptional(list.elementId(), elementType); + } + if (type.isMapType()) { + Types.MapType map = type.asMapType(); + String valuePath = path + ".value"; + Type valueType = createdType(map.valueType(), valuePath, pins); + boolean required = pins.isPinned(valuePath) || pins.pinnedColumnBeneath(valuePath) != null; + return required + ? Types.MapType.ofRequired(map.keyId(), map.valueId(), map.keyType(), valueType) + : Types.MapType.ofOptional(map.keyId(), map.valueId(), map.keyType(), valueType); + } + return type; + } + + private static void reportIncompatible( + TableIdentifier tableId, + List incompatible, + IncompatibleSchemaHandling handling, + String consequence) { + long files = 0; + for (Incompatible item : incompatible) { + files += item.files; + } + if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) { + throw new IncompatibleSchemaException( + "Incompatible schemas for " + + tableId + + " (" + + incompatible.size() + + " schema(s), " + + files + + " file(s)); " + + consequence + + ":\n " + + joinLines(incompatible)); + } + LOG.warn( + "Skipping {} incompatible schema(s) ({} file(s)) for {}; their files will be routed to" + + " the error output:\n {}", + incompatible.size(), + files, + tableId, + joinLines(incompatible)); + } + + /** + * Stages one union per accepted schema onto {@code txn}: a scratch transaction on the evolve path + * (its per-schema versions stay in memory; only the folded result is ever committed), the create + * transaction on the create path. A schema can conflict with another schema's additions, which + * only surfaces while staging and poisons the transaction, so on a conflict the offender is + * returned for the caller to drop and retry with a fresh transaction. + */ + private static @Nullable Accepted stageAll( + Transaction txn, List accepted, List incompatible) { + for (Accepted item : accepted) { + // Both caught types carry staging conflicts: ValidationException from Schema + // construction at apply ("multiple fields for name"), IllegalArgumentException from + // SchemaUpdate preconditions ("Cannot change column type"). + try { + stage(txn, item); + } catch (ValidationException | IllegalArgumentException e) { + incompatible.add( + new Incompatible( + item.json, + item.files, + "conflicts with another file schema in the same window: " + + AddFiles.errorMessage(e))); + return item; } - accepted.remove(failed); } + return null; } /** @@ -325,8 +571,10 @@ private static Transaction newTransactionOn(Table table, Schema base, TableIdent private static void stage(Transaction txn, Accepted item) { UpdateSchema update = txn.updateSchema().unionByNameWith(item.schema); - for (String path : item.delta.absentRequiredPaths()) { - update = update.makeColumnOptional(path); + if (item.delta != null) { + for (String path : item.delta.absentRequiredPaths()) { + update = update.makeColumnOptional(path); + } } update.commit(); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java new file mode 100644 index 000000000000..e3d6889b1616 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaOnceTest.java @@ -0,0 +1,157 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; + +import java.util.Arrays; +import java.util.List; +import org.apache.beam.sdk.PipelineResult; +import org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling; +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.testing.PAssert; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.ParDo; +import org.apache.beam.sdk.values.PCollection; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.Schema; +import org.apache.iceberg.SchemaParser; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.types.Types; +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.rules.TestName; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class CommitSchemaOnceTest { + @ClassRule public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder(); + + @Rule + public transient TestDataWarehouse warehouse = new TestDataWarehouse(TEMPORARY_FOLDER, "default"); + + @Rule public TestName testName = new TestName(); + @Rule public final TestPipeline pipeline = TestPipeline.create(); + + private static final Schema TABLE = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "name", Types.StringType.get())); + + @Test + public void testCommitsTheWindowsSchemasAndEmitsTheSchemaId() { + TableIdentifier tableId = TableIdentifier.of("default", testName.getMethodName()); + warehouse.createTable(tableId, TABLE); + IcebergCatalogConfig catalogConfig = + IcebergCatalogConfig.builder() + .setCatalogProperties( + ImmutableMap.of("type", "hadoop", "warehouse", warehouse.location)) + .build(); + Schema file = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "email", Types.StringType.get())); + List schemas = + Arrays.asList( + CollectDistinctSchemas.SchemaGroup.of( + SchemaParser.toJson(FileSchemas.canonical(file)), 3L, Arrays.asList())); + + PCollection schemaIds = + pipeline + .apply( + Create.of(Arrays.asList(schemas)).withCoder(CollectDistinctSchemas.outputCoder())) + .apply( + ParDo.of( + new CommitSchemaOnce( + catalogConfig, + "default." + testName.getMethodName(), + SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + IncompatibleSchemaHandling.FAIL_PIPELINE, + new CommitSchemaUnion.TableCreation(null, null, null)))); + PAssert.that(schemaIds).containsInAnyOrder(1L); + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); + assertEquals(1, commitsCounted(result)); + + Table table = warehouse.loadTable(tableId); + assertNotNull(table.schema().findField("email")); + } + + /** A window the table already covers commits nothing and does not count as a commit. */ + @Test + public void testNoOpWindowDoesNotCountACommit() { + TableIdentifier tableId = TableIdentifier.of("default", testName.getMethodName()); + warehouse.createTable(tableId, TABLE); + Table table = warehouse.loadTable(tableId); + table + .updateProperties() + .set( + TableProperties.DEFAULT_NAME_MAPPING, NameMappingUtils.regenerate(table.schema(), null)) + .commit(); + IcebergCatalogConfig catalogConfig = + IcebergCatalogConfig.builder() + .setCatalogProperties( + ImmutableMap.of("type", "hadoop", "warehouse", warehouse.location)) + .build(); + Schema covered = new Schema(required(1, "id", Types.LongType.get())); + List schemas = + Arrays.asList( + CollectDistinctSchemas.SchemaGroup.of( + SchemaParser.toJson(FileSchemas.canonical(covered)), 2L, Arrays.asList())); + + pipeline + .apply(Create.of(Arrays.asList(schemas)).withCoder(CollectDistinctSchemas.outputCoder())) + .apply( + ParDo.of( + new CommitSchemaOnce( + catalogConfig, + "default." + testName.getMethodName(), + SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + IncompatibleSchemaHandling.FAIL_PIPELINE, + new CommitSchemaUnion.TableCreation(null, null, null)))); + PipelineResult result = pipeline.run(); + result.waitUntilFinish(); + assertEquals(0, commitsCounted(result)); + } + + private static long commitsCounted(PipelineResult result) { + long total = 0; + for (MetricResult counter : + result + .metrics() + .queryMetrics( + MetricsFilter.builder() + .addNameFilter( + MetricNameFilter.named( + CommitSchemaOnce.class, CommitSchemaOnce.COMMITS_COUNTER)) + .build()) + .getCounters()) { + total += counter.getAttempted(); + } + return total; + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java index b0323064d38e..1a45231da816 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java @@ -29,6 +29,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.EnumSet; +import java.util.HashSet; import java.util.List; import java.util.concurrent.atomic.AtomicInteger; import org.apache.beam.sdk.io.iceberg.CommitSchemaUnion.Committer; @@ -44,6 +45,7 @@ import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.hadoop.HadoopCatalog; import org.apache.iceberg.mapping.NameMapping; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; import org.junit.Before; import org.junit.ClassRule; @@ -75,6 +77,9 @@ public class CommitSchemaUnionTest { private static final SchemaEvolutionConfig ADDITION_ONLY = SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION); + private static final CommitSchemaUnion.TableCreation NO_CREATION = + new CommitSchemaUnion.TableCreation(null, null, null); + private HadoopCatalog catalog; private TableIdentifier tableId; @@ -108,6 +113,7 @@ private long commit( Arrays.asList(schemas), config, handling, + NO_CREATION, CommitSchemaUnion.DEFAULT_COMMITTER); } @@ -115,6 +121,13 @@ private Table load() { return catalog.loadTable(tableId); } + /** Full-schema comparison, field ids normalized; string form so a failure shows the diff. */ + private static void assertSameSchema(Schema expected, Schema actual) { + assertEquals( + TypeUtil.assignIncreasingFreshIds(expected).asStruct().toString(), + TypeUtil.assignIncreasingFreshIds(actual).asStruct().toString()); + } + private static String metadataLocation(Table table) { return ((BaseTable) table).operations().current().metadataFileLocation(); } @@ -539,6 +552,7 @@ public void testFinalSchemaIsOrderIndependent() { Arrays.asList(files(b, 2), files(a, 1)), ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, CommitSchemaUnion.DEFAULT_COMMITTER); assertTrue(first.sameSchema(catalog.loadTable(other).schema())); } @@ -730,6 +744,7 @@ public void testCommitFailedOnceIsRetriedAgainstFreshState() { Arrays.asList(files(file, 1)), ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, flakyThenExternalChange); Table table = load(); assertEquals(2, attempts.get()); @@ -759,10 +774,386 @@ public void testPersistentCommitFailurePropagates() { Arrays.asList(files(file, 1)), ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, alwaysFails)); assertEquals(CommitSchemaUnion.MAX_ATTEMPTS, attempts.get()); } + // ---- create path + + private TableIdentifier missing() { + return TableIdentifier.of("default", testName.getMethodName() + "_new"); + } + + private long commitTo( + TableIdentifier id, + SchemaEvolutionConfig config, + IncompatibleSchemaHandling handling, + CommitSchemaUnion.TableCreation creation, + CollectDistinctSchemas.SchemaGroup... schemas) { + return CommitSchemaUnion.commit( + catalog, + id, + Arrays.asList(schemas), + config, + handling, + creation, + CommitSchemaUnion.DEFAULT_COMMITTER); + } + + @Test + public void testMissingTableIsCreatedFromTheUnion() { + TableIdentifier id = missing(); + Schema seed = + new Schema( + required(1, "id", Types.LongType.get()), + required(2, "region", Types.StringType.get()), + optional(3, "email", Types.StringType.get())); + Schema other = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "extra", Types.LongType.get())); + CommitSchemaUnion.TableCreation creation = + new CommitSchemaUnion.TableCreation( + Arrays.asList("region"), null, java.util.Collections.singletonMap("k", "v")); + long schemaId = + commitTo( + id, + ALL, + IncompatibleSchemaHandling.FAIL_PIPELINE, + creation, + files(seed, 5), + files(other, 1)); + Table table = catalog.loadTable(id); + assertEquals(table.schema().schemaId(), schemaId); + // canonical (sorted) seed columns first, the union's addition last + assertSameSchema( + new Schema( + optional(1, "email", Types.StringType.get()), + optional(2, "id", Types.LongType.get()), + optional(3, "region", Types.StringType.get()), + optional(4, "extra", Types.LongType.get())), + table.schema()); + assertEquals("region", table.spec().fields().get(0).name()); + assertEquals("v", table.properties().get("k")); + assertNotNull(table.properties().get(TableProperties.DEFAULT_NAME_MAPPING)); + assertEquals("born with one schema version", 1, table.schemas().size()); + } + + @Test + public void testPartitionFieldFromNonSeedSchemaResolves() { + TableIdentifier id = missing(); + Schema seed = + new Schema( + required(1, "id", Types.LongType.get()), required(2, "region", Types.StringType.get())); + Schema other = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "extra", Types.StringType.get())); + CommitSchemaUnion.TableCreation creation = + new CommitSchemaUnion.TableCreation(Arrays.asList("extra"), null, null); + commitTo( + id, + ALL, + IncompatibleSchemaHandling.FAIL_PIPELINE, + creation, + files(seed, 5), + files(other, 1)); + Table table = catalog.loadTable(id); + assertEquals("extra", table.spec().fields().get(0).name()); + } + + @Test + public void testPinnedColumnsAndAncestorsAreRequiredOnCreate() { + TableIdentifier id = missing(); + SchemaEvolutionConfig pinned = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.allOf(SchemaEvolutionOption.class)) + .setRequiredColumns(new HashSet<>(Arrays.asList("id", "address.city"))) + .build(); + Schema seed = + new Schema( + required(1, "id", Types.LongType.get()), + required(2, "region", Types.StringType.get()), + optional( + 3, + "address", + Types.StructType.of( + optional(4, "city", Types.StringType.get()), + optional(5, "zip", Types.IntegerType.get())))); + commitTo(id, pinned, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION, files(seed, 1)); + assertSameSchema( + new Schema( + required( + 1, + "address", + Types.StructType.of( + required(2, "city", Types.StringType.get()), + optional(3, "zip", Types.IntegerType.get()))), + required(4, "id", Types.LongType.get()), + optional(5, "region", Types.StringType.get())), + catalog.loadTable(id).schema()); + } + + /** A pin no file schema carries cannot shape the table: fail loudly under FAIL_PIPELINE. */ + @Test + public void testUnmatchedPinFailsCreationUnderFailPipeline() { + TableIdentifier id = missing(); + SchemaEvolutionConfig pinned = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.allOf(SchemaEvolutionOption.class)) + .setRequiredColumns(Collections.singleton("email")) + .build(); + Schema seed = new Schema(required(1, "id", Types.LongType.get())); + IncompatibleSchemaException e = + assertThrows( + IncompatibleSchemaException.class, + () -> + commitTo( + id, + pinned, + IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, + files(seed, 1))); + assertTrue(e.getMessage(), e.getMessage().contains("email")); + assertFalse(catalog.tableExists(id)); + } + + /** + * Iceberg resolves the short container spelling a.b for a.element.b, but the pin walk matches + * segments, so such a pin would silently shape only the ancestors; it is rejected instead. + */ + @Test + public void testShortSpellingPinFailsCreationUnderFailPipeline() { + TableIdentifier id = missing(); + SchemaEvolutionConfig pinned = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.allOf(SchemaEvolutionOption.class)) + .setRequiredColumns(Collections.singleton("l.q")) + .build(); + Schema seed = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "l", + Types.ListType.ofOptional( + 3, Types.StructType.of(optional(4, "q", Types.IntegerType.get()))))); + IncompatibleSchemaException e = + assertThrows( + IncompatibleSchemaException.class, + () -> + commitTo( + id, + pinned, + IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, + files(seed, 1))); + assertTrue(e.getMessage(), e.getMessage().contains("l.q")); + assertFalse(catalog.tableExists(id)); + } + + @Test + public void testUnmatchedPinWarnsAndCreatesUnderRouteToErrors() { + TableIdentifier id = missing(); + SchemaEvolutionConfig pinned = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.allOf(SchemaEvolutionOption.class)) + .setRequiredColumns(Collections.singleton("email")) + .build(); + Schema seed = new Schema(required(1, "id", Types.LongType.get())); + commitTo(id, pinned, IncompatibleSchemaHandling.ROUTE_TO_ERRORS, NO_CREATION, files(seed, 1)); + assertSameSchema( + new Schema(optional(1, "id", Types.LongType.get())), catalog.loadTable(id).schema()); + } + + // ---- createdSchema (direct) + + @Test + public void testCreatedSchemaPinsHoldAtEveryLevel() { + Schema merged = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "l", + Types.ListType.ofOptional( + 3, Types.StructType.of(optional(4, "q", Types.IntegerType.get()))))); + SchemaEvolutionConfig pinned = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.allOf(SchemaEvolutionOption.class)) + .setRequiredColumns(Collections.singleton("l.element.q")) + .build(); + Schema created = CommitSchemaUnion.createdSchema(merged, pinned); + assertSameSchema( + new Schema( + optional(1, "id", Types.LongType.get()), + required( + 2, + "l", + Types.ListType.ofRequired( + 3, Types.StructType.of(required(4, "q", Types.IntegerType.get()))))), + created); + } + + @Test + public void testCreatedSchemaEveryLevelOptionalExceptMapKeys() { + Schema schema = + new Schema( + required(1, "id", Types.LongType.get()), + required( + 2, + "s", + Types.StructType.of( + required(3, "a", Types.IntegerType.get()), + required( + 4, + "items", + Types.ListType.ofRequired( + 5, Types.StructType.of(required(6, "qty", Types.IntegerType.get())))))), + required( + 7, + "attrs", + Types.MapType.ofRequired( + 8, + 9, + Types.StructType.of(required(10, "k", Types.StringType.get())), + Types.StructType.of(required(11, "v", Types.IntegerType.get()))))); + assertSameSchema( + new Schema( + optional(1, "id", Types.LongType.get()), + optional( + 2, + "s", + Types.StructType.of( + optional(3, "a", Types.IntegerType.get()), + optional( + 4, + "items", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "qty", Types.IntegerType.get())))))), + optional( + 7, + "attrs", + Types.MapType.ofOptional( + 8, + 9, + Types.StructType.of(required(10, "k", Types.StringType.get())), + Types.StructType.of(optional(11, "v", Types.IntegerType.get()))))), + CommitSchemaUnion.createdSchema(schema, ALL)); + } + + /** Options guard an existing table's schema; with no table there is nothing to guard. */ + @Test + public void testCreationIsNotGatedOnAnyParticularOption() { + TableIdentifier id = missing(); + Schema seed = + new Schema( + required(1, "id", Types.LongType.get()), required(2, "region", Types.StringType.get())); + commitTo( + id, + SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, + files(seed, 1)); + assertSameSchema( + new Schema( + optional(1, "id", Types.LongType.get()), optional(2, "region", Types.StringType.get())), + catalog.loadTable(id).schema()); + } + + @Test + public void testMissingTableWithoutSchemasIsNotCreated() { + TableIdentifier id = missing(); + long result = + CommitSchemaUnion.commit( + catalog, + id, + new ArrayList<>(), + ALL, + IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, + CommitSchemaUnion.DEFAULT_COMMITTER); + assertEquals(CommitSchemaUnion.NO_TABLE, result); + assertFalse(catalog.tableExists(id)); + } + + @Test + public void testConflictOnCreateFailsWithoutCreating() { + TableIdentifier id = missing(); + Schema asString = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "code", Types.StringType.get())); + Schema asLong = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "code", Types.LongType.get())); + assertThrows( + IncompatibleSchemaException.class, + () -> + commitTo( + id, + ALL, + IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, + files(asLong, 3), + files(asString, 1))); + assertFalse(catalog.tableExists(id)); + } + + @Test + public void testConflictOnCreateRoutesLoserAndCreates() { + TableIdentifier id = missing(); + Schema asString = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "code", Types.StringType.get())); + Schema asLong = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "code", Types.LongType.get())); + commitTo( + id, + ALL, + IncompatibleSchemaHandling.ROUTE_TO_ERRORS, + NO_CREATION, + files(asLong, 3), + files(asString, 1)); + assertSameSchema( + new Schema( + optional(1, "code", Types.LongType.get()), optional(2, "id", Types.LongType.get())), + catalog.loadTable(id).schema()); + } + + @Test + public void testCreateRaceFallsBackToEvolvingTheExistingTable() { + TableIdentifier id = missing(); + AtomicInteger attempts = new AtomicInteger(); + Committer raced = + txn -> { + if (attempts.incrementAndGet() == 1) { + // someone else creates the table first + warehouse.createTable(id, TABLE); + throw new org.apache.iceberg.exceptions.AlreadyExistsException("raced"); + } + txn.commitTransaction(); + }; + Schema file = + new Schema( + required(1, "id", Types.LongType.get()), + required(2, "region", Types.StringType.get()), + optional(3, "email", Types.StringType.get())); + CommitSchemaUnion.commit( + catalog, + id, + Arrays.asList(files(file, 1)), + ALL, + IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, + raced); + Table table = catalog.loadTable(id); + assertEquals(2, attempts.get()); + assertNotNull(table.schema().findField("email")); + assertTrue( + "evolved, not recreated: name from TABLE is still there", + table.schema().findField("name") != null); + } + @Test public void testEmptyInputCommitsNothing() { seedNameMapping(); @@ -774,6 +1165,7 @@ public void testEmptyInputCommitsNothing() { none, ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, CommitSchemaUnion.DEFAULT_COMMITTER); assertEquals(before, metadataLocation(load())); }