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 000000000000..8ffc4780f6e3 --- /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. + * + *

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 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 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 = new ArrayList<>(); + List accepted = new ArrayList<>(); + for (CollectDistinctSchemas.SchemaGroup group : schemas) { + Schema fileSchema = + FileSchemas.markRequired( + SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns()); + SchemaDelta delta = SchemaDelta.classify(table, fileSchema); + if (delta.isEmpty()) { + continue; + } + if (!delta.allowedBy(config)) { + incompatible.add( + new Incompatible( + group.getSchemaJson(), group.getFiles(), delta.disallowedReason(config))); + continue; + } + accepted.add(new Accepted(fileSchema, group.getSchemaJson(), group.getFiles(), delta)); + } + + Transaction scratch = stageAll(table, base, tableId, accepted, incompatible); + boolean folded = !accepted.isEmpty(); + if (folded) { + relaxNewRequiredFields(scratch, base); + } + + 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, + List incompatible) { + while (true) { + Transaction txn = newTransactionOn(table, base, tableId); + Accepted failed = null; + for (Accepted item : accepted) { + // Both caught types carry staging conflicts: ValidationException from Schema + // construction at apply ("multiple fields for name"), IllegalArgumentException from + // SchemaUpdate preconditions ("Cannot change column type"). + try { + stage(txn, item); + } catch (ValidationException | IllegalArgumentException e) { + failed = item; + incompatible.add( + new Incompatible( + item.json, + item.files, + "conflicts with another file schema in the same window: " + + AddFiles.errorMessage(e))); + break; + } + } + 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 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 newRequiredPaths(Schema before, Schema after) { + Set beforeIds = TypeUtil.indexById(before.asStruct()).keySet(); + List paths = new ArrayList<>(); + collectNewRequired(after.asStruct(), "", beforeIds, paths); + return paths; + } + + private static void collectNewRequired( + Type.NestedType type, String prefix, Set beforeIds, List 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 items) { + List 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 000000000000..b0323064d38e --- /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 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 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 none = new ArrayList<>(); + CommitSchemaUnion.commit( + catalog, + tableId, + none, + ALL, + IncompatibleSchemaHandling.FAIL_PIPELINE, + CommitSchemaUnion.DEFAULT_COMMITTER); + assertEquals(before, metadataLocation(load())); + } +}