This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 2bd256532b5 AddFiles: create the table from the file schemas when it
does not exist (#40124)
2bd256532b5 is described below
commit 2bd256532b55f82196d3524c56ac1e2a42647da1
Author: claudevdm <[email protected]>
AuthorDate: Wed Sep 16 12:01:40 2026 -0400
AddFiles: create the table from the file schemas when it does not exist
(#40124)
* 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.
* comments
* test
---
.../beam/sdk/io/iceberg/CommitSchemaOnce.java | 89 ++++
.../beam/sdk/io/iceberg/CommitSchemaUnion.java | 513 +++++++++++++++-----
.../java/org/apache/beam/sdk/io/iceberg/Pins.java | 5 +
.../beam/sdk/io/iceberg/CommitSchemaOnceTest.java | 157 +++++++
.../beam/sdk/io/iceberg/CommitSchemaUnionTest.java | 516 +++++++++++++++++++++
5 files changed, 1169 insertions(+), 111 deletions(-)
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 00000000000..6fd5171c2ea
--- /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<List<CollectDistinctSchemas.SchemaGroup>,
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<CollectDistinctSchemas.SchemaGroup> schemas,
OutputReceiver<Long> 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 8ffc4780f6e..1e145a122c8 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 static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Pr
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.Transaction;
import org.apache.iceberg.UpdateSchema;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.CommitFailedException;
+import org.apache.iceberg.exceptions.NoSuchTableException;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.mapping.NameMapping;
import org.apache.iceberg.types.Type;
@@ -48,12 +53,17 @@ import org.slf4j.Logger;
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.
+ *
+ * <p>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.
*
* <p>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<String> partitionFields;
+ final @Nullable List<String> sortFields;
+ final @Nullable Map<String, String> properties;
+
+ TableCreation(
+ @Nullable List<String> partitionFields,
+ @Nullable List<String> sortFields,
+ @Nullable Map<String, String> 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 @@ final class CommitSchemaUnion {
}
}
+ 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 @@ final class CommitSchemaUnion {
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 @@ final class CommitSchemaUnion {
List<CollectDistinctSchemas.SchemaGroup> 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 @@ final class CommitSchemaUnion {
.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 @@ final class CommitSchemaUnion {
List<CollectDistinctSchemas.SchemaGroup> 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> incompatible = new ArrayList<>();
- List<Accepted> 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> incompatible = new ArrayList<>();
+ List<Accepted> 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) {
@@ -241,7 +242,7 @@ final class CommitSchemaUnion {
return table.schema().schemaId();
}
committer.commit(txn);
- table.refresh();
+ long schemaId = txn.table().schema().schemaId();
long acceptedFiles = 0;
for (Accepted item : accepted) {
acceptedFiles += item.files;
@@ -251,63 +252,351 @@ final class CommitSchemaUnion {
tableId,
accepted.size(),
acceptedFiles,
- table.schema().schemaId());
- return table.schema().schemaId();
+ schemaId);
+ return 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<Accepted> classify(
+ Table table,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ List<Incompatible> incompatible) {
+ List<Accepted> 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> accepted,
List<Incompatible> 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. Columns come out
in the read side's
+ * canonical order (sorted by name at every level), not in any file's
declared order; later unions
+ * append after them.
+ */
+ private static long create(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> 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> incompatible = new ArrayList<>();
+ @Nullable Schema merged = foldForCreate(catalog, tableId, schemas,
incompatible);
+ if (!incompatible.isEmpty()) {
+ reportIncompatible(tableId, incompatible, handling, "no table was
created");
+ }
+ if (merged == null) {
+ LOG.info("Table {} does not exist and no file schema can seed it; not
creating it", tableId);
+ return NO_TABLE;
+ }
+ // The real table is built from the folded result directly.
+ Schema created = createdSchema(merged, config);
+ reportUnenforceablePins(tableId, created, config, handling);
+ Map<String, String> 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);
+ long schemaId = txn.table().schema().schemaId();
+ LOG.info(
+ "Created table {} from {} file schema(s), schema id {}",
+ tableId,
+ schemas.size() - incompatible.size(),
+ schemaId);
+ return schemaId;
+ }
+
+ /**
+ * Unions the window's schemas into one on scratch create transactions that
are never committed,
+ * seeded by the most common schema; conflicts move to {@code incompatible}
and the fold restarts
+ * without the offender.
+ */
+ private static @Nullable Schema foldForCreate(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ List<Incompatible> incompatible) {
+ // The evolve path refuses names no table can absorb (dotted, empty,
differing only in case
+ // within one file) as conflicts in classify; a table must not be born
with them either.
+ List<Accepted> valid = new ArrayList<>();
+ for (CollectDistinctSchemas.SchemaGroup group : schemas) {
+ Schema fileSchema = SchemaParser.fromJson(group.getSchemaJson());
+ List<SchemaChange> invalidNames = new ArrayList<>();
+ ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "",
invalidNames);
+ if (!invalidNames.isEmpty()) {
+ incompatible.add(
+ new Incompatible(
+ group.getSchemaJson(),
+ group.getFiles(),
+ "file schema has column names no table can hold: " +
describe(invalidNames)));
+ continue;
}
+ valid.add(new Accepted(fileSchema, group.getSchemaJson(),
group.getFiles(), null));
+ }
+ if (valid.isEmpty()) {
+ return null;
+ }
+ Schema seed = valid.get(0).schema;
+ List<Accepted> rest = new ArrayList<>(valid.subList(1, valid.size()));
+ while (true) {
+ Transaction scratch = catalog.buildTable(tableId,
seed).createTransaction();
+ Accepted failed = stageAll(scratch, rest, incompatible);
if (failed == null) {
- return txn;
+ return scratch.table().schema();
}
- accepted.remove(failed);
+ 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<String> unenforceable = new ArrayList<>();
+ for (String pin : config.getRequiredColumns()) {
+ Types.NestedField field = created.findField(pin);
+ if (field == null || field.isOptional()) {
+ unenforceable.add(pin);
+ }
+ }
+ if (unenforceable.isEmpty()) {
+ return;
+ }
+ Collections.sort(unenforceable);
+ if (handling == IncompatibleSchemaHandling.FAIL_PIPELINE) {
+ throw new IncompatibleSchemaException(
+ "Pinned column(s) "
+ + unenforceable
+ + " appear in none of the file schemas creating "
+ + tableId
+ + ", or their spelling does not match the column path; the
created table cannot"
+ + " make them required");
+ }
+ LOG.warn(
+ "Pinned column(s) {} appear in none of the file schemas creating {},
or their spelling"
+ + " does not match the column path; the created table cannot make
them required",
+ unenforceable,
+ tableId);
+ }
+
+ /**
+ * The created schema: every field optional at every level, list elements
and map values included,
+ * except pinned paths and their ancestors, which stay required so the
schema advertises the
+ * guarantee the per-file pin check enforces (a null ancestor nulls the
pinned leaf). Map key
+ * subtrees keep their declared shape (keys are required by definition; pins
inside them are not
+ * honored). Nothing depends on a created table's schema yet, so this is the
schema-authoring
+ * moment; evolution never tightens columns afterwards. Column order is the
union's, which is the
+ * canonical (name-sorted) order of the file schemas.
+ */
+ static Schema createdSchema(Schema merged, SchemaEvolutionConfig config) {
+ Pins pins = new Pins(config.getRequiredColumns());
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : merged.asStruct().fields()) {
+ fields.add(createdField(field, field.name(), pins));
+ }
+ return new Schema(fields);
+ }
+
+ private static Types.NestedField createdField(Types.NestedField field,
String path, Pins pins) {
+ boolean required = pins.isPinnedOrAncestorOfPin(path);
+ return Types.NestedField.from(field)
+ .ofType(createdType(field.type(), path, pins))
+ .isOptional(!required)
+ .build();
+ }
+
+ private static Type createdType(Type type, String path, Pins pins) {
+ if (type.isStructType()) {
+ List<Types.NestedField> fields = new ArrayList<>();
+ for (Types.NestedField field : type.asStructType().fields()) {
+ fields.add(createdField(field, path + "." + field.name(), pins));
+ }
+ return Types.StructType.of(fields);
+ }
+ if (type.isListType()) {
+ Types.ListType list = type.asListType();
+ String elementPath = path + ".element";
+ Type elementType = createdType(list.elementType(), elementPath, pins);
+ boolean required = pins.isPinnedOrAncestorOfPin(elementPath);
+ return required
+ ? Types.ListType.ofRequired(list.elementId(), elementType)
+ : Types.ListType.ofOptional(list.elementId(), elementType);
+ }
+ if (type.isMapType()) {
+ Types.MapType map = type.asMapType();
+ String valuePath = path + ".value";
+ Type valueType = createdType(map.valueType(), valuePath, pins);
+ boolean required = pins.isPinnedOrAncestorOfPin(valuePath);
+ return required
+ ? Types.MapType.ofRequired(map.keyId(), map.valueId(),
map.keyType(), valueType)
+ : Types.MapType.ofOptional(map.keyId(), map.valueId(),
map.keyType(), valueType);
+ }
+ return type;
+ }
+
+ private static void reportIncompatible(
+ TableIdentifier tableId,
+ List<Incompatible> 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> accepted, List<Incompatible>
incompatible) {
+ for (Accepted item : accepted) {
+ // classify checked each schema against the base table only; a column
that differs only in
+ // case from one an EARLIER schema of the window added would union as a
second column.
+ List<SchemaChange> collisions = new ArrayList<>();
+ ColumnNameChecks.findCaseCollisions(
+ txn.table().schema().asStruct(), item.schema.asStruct(), "",
collisions);
+ if (!collisions.isEmpty()) {
+ incompatible.add(
+ new Incompatible(
+ item.json,
+ item.files,
+ "conflicts with another file schema in the same window: " +
describe(collisions)));
+ return item;
+ }
+ // Both caught types carry staging conflicts: ValidationException from
Schema
+ // construction at apply ("multiple fields for name"),
IllegalArgumentException from
+ // SchemaUpdate preconditions ("Cannot change column type").
+ try {
+ stage(txn, item);
+ } catch (ValidationException | IllegalArgumentException e) {
+ incompatible.add(
+ new Incompatible(
+ item.json,
+ item.files,
+ "conflicts with another file schema in the same window: "
+ + AddFiles.errorMessage(e)));
+ return item;
+ }
+ }
+ return null;
+ }
+
+ private static String describe(List<SchemaChange> changes) {
+ List<String> descriptions = new ArrayList<>();
+ for (SchemaChange change : changes) {
+ descriptions.add(change.description);
+ }
+ return String.join("; ", descriptions);
+ }
+
/**
* Iceberg refreshes the table on every {@code newTransaction()}, so a
concurrent schema commit
* can slip between two transactions here. Any drift from the snapshot the
window classified
@@ -325,8 +614,10 @@ final class CommitSchemaUnion {
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/main/java/org/apache/beam/sdk/io/iceberg/Pins.java
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java
index b516bfeefed..2ed865b9b87 100644
---
a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java
@@ -43,6 +43,11 @@ final class Pins {
return dotted.contains(dottedPath);
}
+ /** Whether {@code dottedPath} is pinned or has a pin somewhere below it. */
+ boolean isPinnedOrAncestorOfPin(String dottedPath) {
+ return isPinned(dottedPath) || pinnedColumnBeneath(dottedPath) != null;
+ }
+
/**
* Returns the pinned column strictly below {@code dottedPath} (the
lexicographically first when
* several are), or null when there is none. Columns below a pin, or beside
it, have none.
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 00000000000..e3d6889b161
--- /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<CollectDistinctSchemas.SchemaGroup> schemas =
+ Arrays.asList(
+ CollectDistinctSchemas.SchemaGroup.of(
+ SchemaParser.toJson(FileSchemas.canonical(file)), 3L,
Arrays.asList()));
+
+ PCollection<Long> 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<CollectDistinctSchemas.SchemaGroup> 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<Long> 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 b0323064d38..7df6af49920 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.ArrayList;
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.catalog.TableIdentifier;
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 @@ public class CommitSchemaUnionTest {
Arrays.asList(schemas),
config,
handling,
+ NO_CREATION,
CommitSchemaUnion.DEFAULT_COMMITTER);
}
@@ -115,6 +121,13 @@ public class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
Arrays.asList(files(file, 1)),
ALL,
IncompatibleSchemaHandling.FAIL_PIPELINE,
+ NO_CREATION,
flakyThenExternalChange);
Table table = load();
assertEquals(2, attempts.get());
@@ -759,10 +774,386 @@ public class CommitSchemaUnionTest {
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,7 +1165,132 @@ public class CommitSchemaUnionTest {
none,
ALL,
IncompatibleSchemaHandling.FAIL_PIPELINE,
+ NO_CREATION,
CommitSchemaUnion.DEFAULT_COMMITTER);
assertEquals(before, metadataLocation(load()));
}
+
+ /** The create path refuses the same column names classify refuses on the
evolve path. */
+ @Test
+ public void testInvalidColumnNamesCannotSeedACreatedTable() {
+ TableIdentifier id = missing();
+ Schema dotted = new Schema(optional(1, "a.b", Types.LongType.get()));
+ Schema caseDuplicate =
+ new Schema(
+ optional(1, "Id", Types.LongType.get()), optional(2, "id",
Types.LongType.get()));
+ IncompatibleSchemaException e =
+ assertThrows(
+ IncompatibleSchemaException.class,
+ () ->
+ commitTo(
+ id,
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ NO_CREATION,
+ files(dotted, 1)));
+ assertTrue(e.getMessage(), e.getMessage().contains("contains '.'"));
+ assertFalse(catalog.tableExists(id));
+
+ // nothing valid to seed from: routed, and no table
+ long none =
+ commitTo(
+ id,
+ ALL,
+ IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ NO_CREATION,
+ files(dotted, 1),
+ files(caseDuplicate, 2));
+ assertEquals(CommitSchemaUnion.NO_TABLE, none);
+ assertFalse(catalog.tableExists(id));
+
+ // the valid schema alone seeds the table
+ Schema valid = new Schema(required(1, "id", Types.LongType.get()));
+ commitTo(
+ id,
+ ALL,
+ IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ NO_CREATION,
+ files(dotted, 1),
+ files(caseDuplicate, 2),
+ files(valid, 1));
+ assertSameSchema(
+ new Schema(optional(1, "id", Types.LongType.get())),
catalog.loadTable(id).schema());
+ }
+
+ private static final Schema WITH_EMAIL_UPPER =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(2, "name", Types.StringType.get()),
+ optional(3, "score", Types.FloatType.get()),
+ required(4, "region", Types.StringType.get()),
+ optional(5, "Email", Types.StringType.get()));
+ private static final Schema WITH_EMAIL_LOWER =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(2, "name", Types.StringType.get()),
+ optional(3, "score", Types.FloatType.get()),
+ required(4, "region", Types.StringType.get()),
+ optional(5, "email", Types.StringType.get()));
+
+ /**
+ * Each schema is fine against the base table; against each other they
differ only in case, which
+ * the fold must refuse like classify refuses it against the table. The
later schema loses.
+ */
+ @Test
+ public void testCaseCollidingSchemasInOneWindowKeepTheFirst() {
+ IncompatibleSchemaException e =
+ assertThrows(
+ IncompatibleSchemaException.class,
+ () ->
+ commit(
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ files(WITH_EMAIL_UPPER, 3),
+ files(WITH_EMAIL_LOWER, 2)));
+ assertTrue(e.getMessage(), e.getMessage().contains("differs only in
case"));
+ assertTrue(e.getMessage(), e.getMessage().contains("another file schema in
the same window"));
+ assertSameSchema(TABLE, load().schema());
+
+ commit(
+ ALL,
+ IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ files(WITH_EMAIL_UPPER, 3),
+ files(WITH_EMAIL_LOWER, 2));
+ assertSameSchema(WITH_EMAIL_UPPER, load().schema());
+ }
+
+ @Test
+ public void testCaseCollidingSchemasOnCreateKeepTheFirst() {
+ TableIdentifier id = missing();
+ IncompatibleSchemaException e =
+ assertThrows(
+ IncompatibleSchemaException.class,
+ () ->
+ commitTo(
+ id,
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ NO_CREATION,
+ files(WITH_EMAIL_UPPER, 3),
+ files(WITH_EMAIL_LOWER, 2)));
+ assertTrue(e.getMessage(), e.getMessage().contains("differs only in
case"));
+ assertFalse(catalog.tableExists(id));
+
+ commitTo(
+ id,
+ ALL,
+ IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ NO_CREATION,
+ files(WITH_EMAIL_UPPER, 3),
+ files(WITH_EMAIL_LOWER, 2));
+ // created columns: all optional, canonical (name-sorted, upper case
first) order
+ assertSameSchema(
+ new Schema(
+ optional(1, "Email", Types.StringType.get()),
+ optional(2, "id", Types.LongType.get()),
+ optional(3, "name", Types.StringType.get()),
+ optional(4, "region", Types.StringType.get()),
+ optional(5, "score", Types.FloatType.get())),
+ catalog.loadTable(id).schema());
+ }
}
