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 8f02ad3868d AddFiles: CommitSchemaUnion (#40104)
8f02ad3868d is described below
commit 8f02ad3868d2c5498bcc7d147a1ae318a861679d
Author: claudevdm <[email protected]>
AuthorDate: Mon Sep 14 13:48:09 2026 -0400
AddFiles: CommitSchemaUnion (#40104)
* AddFiles: CommitSchemaUnion applies a window's file schemas to the table
in one transaction
The commit side of the pre-pass, for an existing table (creation comes
next). commit(catalog, tableId, schemas, config, handling, committer)
takes the distinct canonical file schemas of one window, most common
first, and returns the table's schema id afterwards.
One attempt, in order:
1. Fresh load of the table; the classification below must run against
the current schema, never a cached one.
2. SchemaDelta.classify for each schema. Empty deltas are skipped;
deltas the config does not allow go to the incompatible list with
their reason; the rest are accepted.
3. Staging, most common first, on one Transaction: per accepted schema
one unionByNameWith plus an explicit makeColumnOptional for every
path the delta reported as an absent required column. Two accepted
schemas can still conflict with each other (file A says score is
long, file B says string; each is fine against the table alone).
That only surfaces while staging and poisons the transaction, so the
offender is moved to the incompatible list ("conflicts with another
file schema in the same window") and the transaction is rebuilt
without it. Most common first means the majority wins such a tie.
4. relaxNewRequiredFields: the union adds top-level columns as optional
but keeps the file's optionality for fields inside a struct it just
added, so one file's luck would impose required nested columns on
every future file. Every field new in this transaction is made
optional at every level, except pinned paths and fields under lists
and maps (which the union already handles). Found by fuzzing;
testFieldsInsideAnAddedStructAreOptional covers it.
5. Name mapping repair: if the table's schema.name-mapping.default is
absent, malformed or does not cover the staged schema, it is
regenerated (NameMappingUtils.regenerate, preserving custom names by
field id). Zero-copy files carry no field ids, so a column missing
from the mapping is unreadable in every registered file; a missing
mapping alone is reason enough to commit even when the schema did
not change.
6. Incompatible schemas are reported: under FAIL_PIPELINE an
IncompatibleSchemaException listing every one (schema, file count,
reason) is thrown before anything is committed; under
ROUTE_TO_ERRORS they are logged and the rest proceeds (their files
fail the per-file coverage check later, with the same reason).
7. Single commit through the injectable Committer, then refresh.
Nothing is committed when nothing was staged.
Retry: CommitFailedException (a concurrent writer moved the table
between load and commit) restarts the whole attempt from step 1, up to
MAX_ATTEMPTS (5); classification is redone against the new state rather
than replayed, since the concurrent change may have made a delta empty
or incompatible. A persistent failure propagates.
Worked example: table {id required long, name required string}; window
schemas A x5 {id, name optional, score long}, B x1 {id, score string}
under all options. A: relax name, add score. B: absent name
(relaxation), score conflicts with A's addition while staging, so B is
incompatible. FAIL_PIPELINE: exception, table untouched. ROUTE_TO_ERRORS:
one commit with name optional and score optional long; B's file will be
routed to errors at registration.
The tests use a local HadoopCatalog and check metadata versions to
prove "nothing committed" claims, and inject a Committer that fails
once to exercise the retry path.
* comments
---
.../beam/sdk/io/iceberg/CommitSchemaUnion.java | 402 +++++++++++
.../beam/sdk/io/iceberg/CommitSchemaUnionTest.java | 780 +++++++++++++++++++++
2 files changed, 1182 insertions(+)
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
new file mode 100644
index 00000000000..8ffc4780f6e
--- /dev/null
+++
b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java
@@ -0,0 +1,402 @@
+/*
+ * 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.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import
org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
+import org.apache.beam.sdk.util.BackOff;
+import org.apache.beam.sdk.util.BackOffUtils;
+import org.apache.beam.sdk.util.FluentBackoff;
+import org.apache.beam.sdk.util.Sleeper;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SchemaParser;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.TableProperties;
+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.CommitFailedException;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Duration;
+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.
+ *
+ * <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
+ * at registration ({@link IncompatibleSchemaHandling#ROUTE_TO_ERRORS}).
+ */
+final class CommitSchemaUnion {
+ private static final Logger LOG =
LoggerFactory.getLogger(CommitSchemaUnion.class);
+
+ static final int MAX_ATTEMPTS = 5;
+
+ /** Injectable so tests can exercise the commit retry path. */
+ interface Committer extends Serializable {
+ void commit(Transaction txn);
+ }
+
+ static final Committer DEFAULT_COMMITTER = Transaction::commitTransaction;
+
+ /** Thrown under FAIL_PIPELINE; the message lists every incompatible schema.
*/
+ static final class IncompatibleSchemaException extends IllegalStateException
{
+ IncompatibleSchemaException(String message) {
+ super(message);
+ }
+ }
+
+ private static final class Incompatible {
+ final String schemaJson;
+ final long files;
+ final String reason;
+
+ Incompatible(String schemaJson, long files, String reason) {
+ this.schemaJson = schemaJson;
+ this.files = files;
+ this.reason = reason;
+ }
+
+ @Override
+ public String toString() {
+ return files + " file(s) with schema " + truncate(schemaJson) + ": " +
reason;
+ }
+ }
+
+ /** Canonical JSON of a wide schema runs to hundreds of KB; the reason is
what matters. */
+ private static final int MAX_SCHEMA_JSON_CHARS = 1024;
+
+ private static String truncate(String json) {
+ if (json.length() <= MAX_SCHEMA_JSON_CHARS) {
+ return json;
+ }
+ return json.substring(0, MAX_SCHEMA_JSON_CHARS)
+ + "... ("
+ + (json.length() - MAX_SCHEMA_JSON_CHARS)
+ + " chars truncated)";
+ }
+
+ private CommitSchemaUnion() {}
+
+ /**
+ * Applies the schemas and returns the table's schema id after the call.
+ *
+ * @param schemas the window's distinct schema groups, most common first
+ */
+ static long commit(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ 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.*)
+ // sit below this loop.
+ BackOff backoff =
+ FluentBackoff.DEFAULT
+ .withMaxRetries(MAX_ATTEMPTS - 1)
+ .withInitialBackoff(Duration.millis(100))
+ .withMaxBackoff(Duration.standardSeconds(2))
+ .backoff();
+ for (int attempt = 1; ; attempt++) {
+ try {
+ return commitOnce(catalog, tableId, schemas, config, handling,
committer);
+ } catch (CommitFailedException e) {
+ try {
+ if (!BackOffUtils.next(Sleeper.DEFAULT, backoff)) {
+ throw e;
+ }
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ throw e;
+ }
+ LOG.info(
+ "Schema commit attempt {}/{} for {} failed; reloading and
rebuilding",
+ attempt,
+ MAX_ATTEMPTS,
+ tableId,
+ e);
+ }
+ }
+ }
+
+ private static long commitOnce(
+ Catalog catalog,
+ TableIdentifier tableId,
+ List<CollectDistinctSchemas.SchemaGroup> schemas,
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ Committer committer) {
+ Table table = catalog.loadTable(tableId);
+ // 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);
+ }
+
+ 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);
+ }
+ boolean staged = folded;
+ 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));
+ }
+
+ if (!staged) {
+ LOG.info(
+ "Table {} already covers all {} file schema(s); nothing to commit",
+ tableId,
+ schemas.size());
+ return table.schema().schemaId();
+ }
+ committer.commit(txn);
+ table.refresh();
+ long acceptedFiles = 0;
+ for (Accepted item : accepted) {
+ acceptedFiles += item.files;
+ }
+ LOG.info(
+ "Committed schema union for {}: {} schema(s) covering {} file(s), now
at schema id {}",
+ tableId,
+ accepted.size(),
+ acceptedFiles,
+ table.schema().schemaId());
+ 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;
+ }
+ }
+
+ /**
+ * 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.
+ */
+ private static Transaction stageAll(
+ 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;
+ }
+ }
+ if (failed == null) {
+ return txn;
+ }
+ accepted.remove(failed);
+ }
+ }
+
+ /**
+ * 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
+ * against is thrown as {@link CommitFailedException} so the commit-level
retry reloads and
+ * rebuilds, leaving the replay checkState as a pure bug detector.
+ */
+ private static Transaction newTransactionOn(Table table, Schema base,
TableIdentifier tableId) {
+ Transaction txn = table.newTransaction();
+ if (!txn.table().schema().sameSchema(base)) {
+ throw new CommitFailedException(
+ "concurrent schema change on %s while staging the schema union",
tableId);
+ }
+ return txn;
+ }
+
+ 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);
+ }
+ update.commit();
+ }
+
+ /**
+ * New columns are optional at every level. The union adds top-level columns
optional but keeps
+ * the file's optionality below them, so one file's luck would otherwise
impose required fields on
+ * everyone. Pins do not shape new columns: they keep existing required
columns from being relaxed
+ * (SchemaDelta) and gate files at registration.
+ */
+ private static void relaxNewRequiredFields(Transaction txn, Schema before) {
+ List<String> toRelax = newRequiredPaths(before, txn.table().schema());
+ if (toRelax.isEmpty()) {
+ return;
+ }
+ UpdateSchema update = txn.updateSchema();
+ for (String path : toRelax) {
+ update = update.makeColumnOptional(path);
+ }
+ update.commit();
+ }
+
+ /**
+ * Paths of required fields that {@code after} has and {@code before} lacks,
in schema order;
+ * includes fields under lists and maps (a required list element or map
value counts). Map key
+ * subtrees are skipped: keys are required by definition and relaxing inside
a struct key would
+ * change key identity.
+ */
+ static List<String> newRequiredPaths(Schema before, Schema after) {
+ Set<Integer> beforeIds = TypeUtil.indexById(before.asStruct()).keySet();
+ List<String> paths = new ArrayList<>();
+ collectNewRequired(after.asStruct(), "", beforeIds, paths);
+ return paths;
+ }
+
+ private static void collectNewRequired(
+ Type.NestedType type, String prefix, Set<Integer> beforeIds,
List<String> paths) {
+ for (Types.NestedField field : type.fields()) {
+ if (type.isMapType() && field.fieldId() == type.asMapType().keyId()) {
+ continue;
+ }
+ String path = prefix + field.name();
+ if (!beforeIds.contains(field.fieldId()) && field.isRequired()) {
+ paths.add(path);
+ }
+ if (field.type().isNestedType()) {
+ collectNewRequired(field.type().asNestedType(), path + ".", beforeIds,
paths);
+ }
+ }
+ }
+
+ /** Regenerates the name mapping when absent, malformed or not covering the
staged schema. */
+ private static boolean stageNameMapping(Transaction txn) {
+ Schema schema = txn.table().schema();
+ @Nullable NameMapping existing =
+ NameMappingUtils.parseOrNull(
+
txn.table().properties().get(TableProperties.DEFAULT_NAME_MAPPING));
+ if (existing != null && NameMappingUtils.covers(existing,
schema.asStruct())) {
+ return false;
+ }
+ String regenerated = NameMappingUtils.regenerate(schema, existing);
+ txn.updateProperties().set(TableProperties.DEFAULT_NAME_MAPPING,
regenerated).commit();
+ return true;
+ }
+
+ private static String joinLines(List<Incompatible> items) {
+ List<String> lines = new ArrayList<>();
+ for (Incompatible item : items) {
+ lines.add(item.toString());
+ }
+ return String.join("\n ", lines);
+ }
+}
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
new file mode 100644
index 00000000000..b0323064d38
--- /dev/null
+++
b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java
@@ -0,0 +1,780 @@
+/*
+ * 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.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.EnumSet;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.beam.sdk.io.iceberg.CommitSchemaUnion.Committer;
+import
org.apache.beam.sdk.io.iceberg.CommitSchemaUnion.IncompatibleSchemaException;
+import
org.apache.beam.sdk.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHandling;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.BaseTable;
+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.exceptions.CommitFailedException;
+import org.apache.iceberg.hadoop.HadoopCatalog;
+import org.apache.iceberg.mapping.NameMapping;
+import org.apache.iceberg.types.Types;
+import org.junit.Before;
+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 CommitSchemaUnionTest {
+ @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();
+
+ private static final Schema TABLE =
+ 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()));
+
+ private static final SchemaEvolutionConfig ALL =
+ SchemaEvolutionConfig.of(SchemaEvolutionOption.values());
+ private static final SchemaEvolutionConfig ADDITION_ONLY =
+ SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION);
+
+ private HadoopCatalog catalog;
+ private TableIdentifier tableId;
+
+ @Before
+ public void setUp() {
+ catalog = new HadoopCatalog(new Configuration(), warehouse.location);
+ tableId = TableIdentifier.of("default", testName.getMethodName());
+ warehouse.createTable(tableId, TABLE);
+ }
+
+ private static String json(Schema schema) {
+ return SchemaParser.toJson(FileSchemas.canonical(schema));
+ }
+
+ private static CollectDistinctSchemas.SchemaGroup files(Schema schema, long
count) {
+ return files(schema, count, Collections.emptyList());
+ }
+
+ private static CollectDistinctSchemas.SchemaGroup files(
+ Schema schema, long count, List<String> nullFreeColumns) {
+ return CollectDistinctSchemas.SchemaGroup.of(json(schema), count,
nullFreeColumns);
+ }
+
+ private long commit(
+ SchemaEvolutionConfig config,
+ IncompatibleSchemaHandling handling,
+ CollectDistinctSchemas.SchemaGroup... schemas) {
+ return CommitSchemaUnion.commit(
+ catalog,
+ tableId,
+ Arrays.asList(schemas),
+ config,
+ handling,
+ CommitSchemaUnion.DEFAULT_COMMITTER);
+ }
+
+ private Table load() {
+ return catalog.loadTable(tableId);
+ }
+
+ private static String metadataLocation(Table table) {
+ return ((BaseTable) table).operations().current().metadataFileLocation();
+ }
+
+ /** The vN in .../metadata/vN.metadata.json: one commit bumps it by exactly
one. */
+ private static int metadataVersion(Table table) {
+ String location = metadataLocation(table);
+ String file = location.substring(location.lastIndexOf('/') + 2);
+ return Integer.parseInt(file.substring(0, file.indexOf('.')));
+ }
+
+ /** A healthy mapping, so no-op tests are not turned into commits by the
mapping repair. */
+ private void seedNameMapping() {
+ Table table = load();
+ table
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING,
NameMappingUtils.regenerate(table.schema(), null))
+ .commit();
+ }
+
+ // ---- no-op
+
+ @Test
+ public void testCoveredSchemasCommitNothing() {
+ seedNameMapping();
+ String before = metadataLocation(load());
+ Schema covered =
+ new Schema(
+ required(1, "id", Types.IntegerType.get()),
+ required(2, "region", Types.StringType.get()));
+ long schemaId = commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE,
files(covered, 3));
+ Table table = load();
+ assertEquals(table.schema().schemaId(), schemaId);
+ assertEquals("no metadata written", before, metadataLocation(table));
+ }
+
+ // ---- changes applied
+
+ @Test
+ public void testAdditionPromotionAndRelaxationInOneCommit() {
+ Schema file =
+ new Schema(
+ optional(1, "id", Types.LongType.get()),
+ optional(2, "score", Types.DoubleType.get()),
+ optional(3, "email", Types.StringType.get()),
+ required(4, "region", Types.StringType.get()));
+ long schemaId = commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE,
files(file, 2));
+ Table table = load();
+ assertEquals(table.schema().schemaId(), schemaId);
+ assertTrue(table.schema().findField("id").isOptional());
+ assertEquals(Types.DoubleType.get(),
table.schema().findField("score").type());
+ assertNotNull(table.schema().findField("email"));
+ assertTrue(table.schema().findField("email").isOptional());
+ }
+
+ @Test
+ public void testAbsentRequiredColumnIsRelaxedExplicitly() {
+ Schema file = new Schema(required(1, "id", Types.LongType.get()));
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(file, 1));
+ assertTrue(load().schema().findField("region").isOptional());
+ }
+
+ @Test
+ public void testFieldsInsideAnAddedStructAreOptional() {
+ Schema file =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ required(
+ 3,
+ "address",
+ Types.StructType.of(
+ required(4, "city", Types.StringType.get()),
+ required(5, "zip", Types.IntegerType.get()))));
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(file, 1));
+ Table table = load();
+ assertTrue(table.schema().findField("address").isOptional());
+ assertTrue(table.schema().findField("address.city").isOptional());
+ assertTrue(table.schema().findField("address.zip").isOptional());
+ }
+
+ /**
+ * Pins do not shape new columns: a pinned field arriving inside an added
struct is relaxed like
+ * any other. Pins keep existing required columns from being relaxed and
gate files at
+ * registration.
+ */
+ @Test
+ public void testPinnedFieldInsideAnAddedStructIsStillAddedOptional() {
+ SchemaEvolutionConfig pinned =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setRequiredColumns(Collections.singleton("address.city"))
+ .build();
+ Schema file =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ required(
+ 3, "address", Types.StructType.of(required(4, "city",
Types.StringType.get()))));
+ commit(pinned, IncompatibleSchemaHandling.FAIL_PIPELINE, files(file, 1));
+ assertTrue(load().schema().findField("address.city").isOptional());
+ assertTrue(load().schema().findField("address").isOptional());
+ }
+
+ @Test
+ public void testFieldsInsideAddedContainersAreOptional() {
+ Schema file =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(
+ 3,
+ "items",
+ Types.ListType.ofRequired(
+ 4, Types.StructType.of(required(5, "qty",
Types.IntegerType.get())))),
+ optional(
+ 6,
+ "attrs",
+ Types.MapType.ofRequired(
+ 7,
+ 8,
+ Types.StringType.get(),
+ Types.StructType.of(required(9, "v",
Types.IntegerType.get())))));
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(file, 1));
+ Schema schema = load().schema();
+ assertTrue(schema.findField("items.element").isOptional());
+ assertTrue(schema.findField("items.element.qty").isOptional());
+ assertTrue(schema.findField("attrs.value").isOptional());
+ assertTrue(schema.findField("attrs.value.v").isOptional());
+ }
+
+ // ---- newRequiredPaths (direct)
+
+ @Test
+ public void testNewRequiredPathsAtEveryLevelExceptMapKeys() {
+ Schema before = new Schema(required(1, "id", Types.LongType.get()));
+ Schema after =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "s",
+ Types.StructType.of(
+ required(3, "a", Types.IntegerType.get()),
+ optional(4, "b", Types.IntegerType.get()))),
+ optional(
+ 5,
+ "items",
+ Types.ListType.ofRequired(
+ 6, Types.StructType.of(required(7, "qty",
Types.IntegerType.get())))),
+ optional(
+ 8,
+ "attrs",
+ Types.MapType.ofRequired(
+ 9,
+ 10,
+ Types.StructType.of(required(11, "k",
Types.StringType.get())),
+ Types.StructType.of(required(12, "v",
Types.IntegerType.get())))));
+ assertEquals(
+ Arrays.asList("s.a", "items.element", "items.element.qty",
"attrs.value", "attrs.value.v"),
+ CommitSchemaUnion.newRequiredPaths(before, after));
+ }
+
+ /** Names containing element/key/value are not containers; regression for a
substring check. */
+ @Test
+ public void testNewRequiredPathsContainerLikeNamesAreNotContainers() {
+ Schema before = new Schema(required(1, "id", Types.LongType.get()));
+ Schema after =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "stats",
+ Types.StructType.of(
+ required(3, "keyword", Types.StringType.get()),
+ required(4, "value_sum", Types.LongType.get()),
+ required(5, "element", Types.StringType.get()))));
+ assertEquals(
+ Arrays.asList("stats.keyword", "stats.value_sum", "stats.element"),
+ CommitSchemaUnion.newRequiredPaths(before, after));
+ }
+
+ @Test
+ public void testNewRequiredPathsInNestedContainers() {
+ Schema before = new Schema(required(1, "id", Types.LongType.get()));
+ Schema after =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "ll",
+ Types.ListType.ofRequired(
+ 3, Types.ListType.ofRequired(4, Types.IntegerType.get()))),
+ optional(
+ 5,
+ "lm",
+ Types.ListType.ofRequired(
+ 6,
+ Types.MapType.ofRequired(
+ 7, 8, Types.StringType.get(),
Types.IntegerType.get()))));
+ assertEquals(
+ Arrays.asList("ll.element", "ll.element.element", "lm.element",
"lm.element.value"),
+ CommitSchemaUnion.newRequiredPaths(before, after));
+ }
+
+ /** Growing an existing struct: only the field with a new id is a candidate.
*/
+ @Test
+ public void testNewRequiredPathsIgnoreExistingFields() {
+ Schema before =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(2, "s", Types.StructType.of(required(3, "old",
Types.IntegerType.get()))));
+ Schema after =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ optional(
+ 2,
+ "s",
+ Types.StructType.of(
+ required(3, "old", Types.IntegerType.get()),
+ required(4, "fresh", Types.IntegerType.get()))));
+ assertEquals(Arrays.asList("s.fresh"),
CommitSchemaUnion.newRequiredPaths(before, after));
+ }
+
+ /** A declared-optional column every file proved null-free does not relax
the table. */
+ @Test
+ public void testNullFreeColumnIsNotRelaxed() {
+ seedNameMapping();
+ Schema file =
+ new Schema(
+ required(1, "id", Types.LongType.get()), optional(2, "region",
Types.StringType.get()));
+ int before = metadataVersion(load());
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(file, 3,
Arrays.asList("region")));
+ assertTrue(load().schema().findField("region").isRequired());
+ assertEquals(before, metadataVersion(load()));
+ }
+
+ @Test
+ public void testColumnWithoutNullFreeEvidenceStillRelaxes() {
+ Schema file =
+ new Schema(
+ required(1, "id", Types.LongType.get()), optional(2, "region",
Types.StringType.get()));
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(file, 3));
+ assertTrue(load().schema().findField("region").isOptional());
+ }
+
+ @Test
+ public void testMultipleSchemasMergeIntoOneSchemaVersion() {
+ Schema a =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "a", Types.StringType.get()),
+ optional(4, "shared", Types.StringType.get()));
+ Schema b =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "b", Types.LongType.get()),
+ optional(4, "shared", Types.StringType.get()));
+ int before = metadataVersion(load());
+ int schemasBefore = load().schemas().size();
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(a, 5),
files(b, 1));
+ Table table = load();
+ assertNotNull(table.schema().findField("a"));
+ assertNotNull(table.schema().findField("b"));
+ assertNotNull(table.schema().findField("shared"));
+ assertEquals("one metadata commit", before + 1, metadataVersion(table));
+ assertEquals(
+ "one schema version for the whole window", schemasBefore + 1,
table.schemas().size());
+ }
+
+ /**
+ * Columns born in the window negotiate their type among the window's files;
the options guard
+ * only columns the table already had. The winner is the widest type
regardless of file counts.
+ */
+ @Test
+ public void
testOverlappingNewColumnsNegotiateTheWidestTypeInOneSchemaVersion() {
+ Schema a =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.IntegerType.get()));
+ Schema b =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.LongType.get()));
+ int schemasBefore = load().schemas().size();
+ commit(ADDITION_ONLY, IncompatibleSchemaHandling.FAIL_PIPELINE, files(a,
5), files(b, 1));
+ Table table = load();
+ assertEquals(Types.LongType.get(),
table.schema().findField("code").type());
+ assertEquals(schemasBefore + 1, table.schemas().size());
+ }
+
+ /** The narrower schema folding second is ignorable for the union, not a
conflict. */
+ @Test
+ public void testOverlappingNewColumnsWidestFirstOrder() {
+ Schema wide =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.LongType.get()));
+ Schema narrow =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.IntegerType.get()));
+ int schemasBefore = load().schemas().size();
+ commit(
+ ADDITION_ONLY, IncompatibleSchemaHandling.FAIL_PIPELINE, files(wide,
5), files(narrow, 1));
+ Table table = load();
+ assertEquals(Types.LongType.get(),
table.schema().findField("code").type());
+ assertEquals(schemasBefore + 1, table.schemas().size());
+ }
+
+ @Test
+ public void testThreeOverlappingSchemasAllAccepted() {
+ Schema narrow =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.IntegerType.get()));
+ Schema wide =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.LongType.get()));
+ Schema narrowExtra =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.IntegerType.get()),
+ optional(4, "extra", Types.StringType.get()));
+ commit(
+ ADDITION_ONLY,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ files(narrow, 3),
+ files(wide, 2),
+ files(narrowExtra, 1));
+ Table table = load();
+ assertEquals(Types.LongType.get(),
table.schema().findField("code").type());
+ assertNotNull(table.schema().findField("extra"));
+ }
+
+ /**
+ * Pins do not shape new columns: a pinned new top-level column is added
optional like any other.
+ * The pin is enforced per file at registration.
+ */
+ @Test
+ public void testPinnedNewTopLevelColumnIsAddedOptional() {
+ SchemaEvolutionConfig pinned =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setRequiredColumns(Collections.singleton("email"))
+ .build();
+ Schema file =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ required(3, "email", Types.StringType.get()));
+ commit(pinned, IncompatibleSchemaHandling.FAIL_PIPELINE, files(file, 1));
+ assertTrue(load().schema().findField("email").isOptional());
+ }
+
+ /**
+ * Because pinned new columns are added optional, a sibling schema lacking
the pinned column is
+ * not a schema-level problem: both schemas land, and the pin is enforced
per file instead.
+ */
+ @Test
+ public void testSiblingLackingPinnedNestedColumnIsStillCompatible() {
+ SchemaEvolutionConfig pinned =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setRequiredColumns(Collections.singleton("address.city"))
+ .build();
+ Schema withCity =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(
+ 3, "address", Types.StructType.of(required(4, "city",
Types.StringType.get()))));
+ Schema withZipOnly =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(
+ 3, "address", Types.StructType.of(optional(4, "zip",
Types.IntegerType.get()))));
+ int schemasBefore = load().schemas().size();
+ commit(
+ pinned,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ files(withCity, 2),
+ files(withZipOnly, 1));
+ Table table = load();
+ assertTrue(table.schema().findField("address.city").isOptional());
+ assertTrue(table.schema().findField("address.zip").isOptional());
+ assertEquals(schemasBefore + 1, table.schemas().size());
+ }
+
+ @Test
+ public void testFinalSchemaIsOrderIndependent() {
+ Schema a =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "score", Types.DoubleType.get()));
+ Schema b =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "extra", Types.StringType.get()));
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(a, 1),
files(b, 2));
+ Schema first = load().schema();
+
+ TableIdentifier other = TableIdentifier.of("default",
testName.getMethodName() + "_2");
+ warehouse.createTable(other, TABLE);
+ CommitSchemaUnion.commit(
+ catalog,
+ other,
+ Arrays.asList(files(b, 2), files(a, 1)),
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ CommitSchemaUnion.DEFAULT_COMMITTER);
+ assertTrue(first.sameSchema(catalog.loadTable(other).schema()));
+ }
+
+ // ---- incompatible schemas
+
+ @Test
+ public void testFailPipelineCommitsNothingWhenAnySchemaIsIncompatible() {
+ String before = metadataLocation(load());
+ Schema good =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "email", Types.StringType.get()));
+ Schema needsPromotion =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "score", Types.DoubleType.get()));
+ IncompatibleSchemaException e =
+ assertThrows(
+ IncompatibleSchemaException.class,
+ () ->
+ commit(
+ ADDITION_ONLY,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ files(good, 9),
+ files(needsPromotion, 1)));
+ assertTrue(e.getMessage(), e.getMessage().contains("1 schema(s), 1
file(s)"));
+ assertTrue(e.getMessage(), e.getMessage().contains("promote score float to
double"));
+ assertEquals(before, metadataLocation(load()));
+ }
+
+ /** A wide schema's JSON is truncated in the message; the reason stays
whole. */
+ @Test
+ public void testIncompatibleMessageTruncatesWideSchemaJson() {
+ List<Types.NestedField> fields = new ArrayList<>();
+ // id as string conflicts with the table's long: incompatible under every
option
+ fields.add(required(1, "id", Types.StringType.get()));
+ for (int i = 2; i <= 60; i++) {
+ fields.add(optional(i, "very_long_column_name_number_" + i,
Types.StringType.get()));
+ }
+ Schema wide = new Schema(fields);
+ IncompatibleSchemaException e =
+ assertThrows(
+ IncompatibleSchemaException.class,
+ () -> commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE,
files(wide, 1)));
+ assertTrue(e.getMessage(), e.getMessage().contains("chars truncated)"));
+ assertFalse(e.getMessage(),
e.getMessage().contains("very_long_column_name_number_60"));
+ }
+
+ @Test
+ public void testRouteToErrorsSkipsIncompatibleAndCommitsTheRest() {
+ Schema good =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "email", Types.StringType.get()));
+ Schema needsPromotion =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "score", Types.DoubleType.get()));
+ commit(
+ ADDITION_ONLY,
+ IncompatibleSchemaHandling.ROUTE_TO_ERRORS,
+ files(good, 9),
+ files(needsPromotion, 1));
+ Table table = load();
+ assertNotNull(table.schema().findField("email"));
+ assertEquals(Types.FloatType.get(),
table.schema().findField("score").type());
+ }
+
+ @Test
+ public void testPinnedRelaxationIsIncompatible() {
+ SchemaEvolutionConfig pinned =
+ SchemaEvolutionConfig.builder()
+ .setOptions(EnumSet.allOf(SchemaEvolutionOption.class))
+ .setRequiredColumns(Collections.singleton("region"))
+ .build();
+ Schema file = new Schema(required(1, "id", Types.LongType.get()));
+ IncompatibleSchemaException e =
+ assertThrows(
+ IncompatibleSchemaException.class,
+ () -> commit(pinned, IncompatibleSchemaHandling.FAIL_PIPELINE,
files(file, 4)));
+ assertTrue(e.getMessage(), e.getMessage().contains("pinned as required"));
+ assertTrue(load().schema().findField("region").isRequired());
+ }
+
+ @Test
+ public void testMostCommonSchemaWinsAConflictBetweenFiles() {
+ Schema asString =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.StringType.get()));
+ Schema asLong =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.LongType.get()));
+ commit(ALL, IncompatibleSchemaHandling.ROUTE_TO_ERRORS, files(asLong, 7),
files(asString, 2));
+ assertEquals(Types.LongType.get(),
load().schema().findField("code").type());
+ }
+
+ @Test
+ public void testConflictBetweenFilesFailsPipelineWithoutCommit() {
+ String before = metadataLocation(load());
+ Schema asString =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.StringType.get()));
+ Schema asLong =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "code", Types.LongType.get()));
+ IncompatibleSchemaException e =
+ assertThrows(
+ IncompatibleSchemaException.class,
+ () ->
+ commit(
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ files(asLong, 7),
+ files(asString, 2)));
+ assertTrue(e.getMessage(), e.getMessage().contains("conflicts with another
file schema"));
+ assertEquals(before, metadataLocation(load()));
+ }
+
+ // ---- name mapping
+
+ @Test
+ public void testStaleNameMappingIsRegeneratedForTheNewSchema() {
+ Table table = load();
+ table
+ .updateProperties()
+ .set(
+ TableProperties.DEFAULT_NAME_MAPPING,
NameMappingUtils.regenerate(table.schema(), null))
+ .commit();
+ Schema file =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "email", Types.StringType.get()));
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(file, 1));
+ table = load();
+ NameMapping mapping =
+
NameMappingUtils.parseOrNull(table.properties().get(TableProperties.DEFAULT_NAME_MAPPING));
+ assertNotNull(mapping);
+ assertTrue(NameMappingUtils.covers(mapping, table.schema().asStruct()));
+ assertNotNull(mapping.find("email"));
+ }
+
+ @Test
+ public void testMissingNameMappingIsAddedEvenWithoutSchemaChanges() {
+
assertFalse(load().properties().containsKey(TableProperties.DEFAULT_NAME_MAPPING));
+ commit(ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, files(TABLE, 1));
+ Table table = load();
+ NameMapping mapping =
+
NameMappingUtils.parseOrNull(table.properties().get(TableProperties.DEFAULT_NAME_MAPPING));
+ assertNotNull(mapping);
+ assertTrue(NameMappingUtils.covers(mapping, table.schema().asStruct()));
+ }
+
+ // ---- retry
+
+ @Test
+ public void testCommitFailedOnceIsRetriedAgainstFreshState() {
+ AtomicInteger attempts = new AtomicInteger();
+ Committer flakyThenExternalChange =
+ txn -> {
+ if (attempts.incrementAndGet() == 1) {
+ // someone else adds a column between our load and commit
+ load().updateSchema().addColumn("external",
Types.StringType.get()).commit();
+ throw new CommitFailedException("simulated concurrent commit");
+ }
+ 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,
+ tableId,
+ Arrays.asList(files(file, 1)),
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ flakyThenExternalChange);
+ Table table = load();
+ assertEquals(2, attempts.get());
+ assertNotNull(table.schema().findField("external"));
+ assertNotNull(table.schema().findField("email"));
+ }
+
+ @Test
+ public void testPersistentCommitFailurePropagates() {
+ AtomicInteger attempts = new AtomicInteger();
+ Committer alwaysFails =
+ txn -> {
+ attempts.incrementAndGet();
+ throw new CommitFailedException("always");
+ };
+ Schema file =
+ new Schema(
+ required(1, "id", Types.LongType.get()),
+ required(2, "region", Types.StringType.get()),
+ optional(3, "email", Types.StringType.get()));
+ assertThrows(
+ CommitFailedException.class,
+ () ->
+ CommitSchemaUnion.commit(
+ catalog,
+ tableId,
+ Arrays.asList(files(file, 1)),
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ alwaysFails));
+ assertEquals(CommitSchemaUnion.MAX_ATTEMPTS, attempts.get());
+ }
+
+ @Test
+ public void testEmptyInputCommitsNothing() {
+ seedNameMapping();
+ String before = metadataLocation(load());
+ List<CollectDistinctSchemas.SchemaGroup> none = new ArrayList<>();
+ CommitSchemaUnion.commit(
+ catalog,
+ tableId,
+ none,
+ ALL,
+ IncompatibleSchemaHandling.FAIL_PIPELINE,
+ CommitSchemaUnion.DEFAULT_COMMITTER);
+ assertEquals(before, metadataLocation(load()));
+ }
+}
