From 09837507bfde04ee6cade14ca7fd62af68bd3dc0 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 09:22:34 -0400 Subject: [PATCH 1/2] 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. --- .../sdk/io/iceberg/CommitSchemaUnion.java | 363 +++++++++ .../sdk/io/iceberg/CommitSchemaUnionTest.java | 762 ++++++++++++++++++ 2 files changed, 1125 insertions(+) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java 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..634e363c3858 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnion.java @@ -0,0 +1,363 @@ +/* + * 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.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.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 " + schemaJson + ": " + reason; + } + } + + 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) { + for (int attempt = 1; ; attempt++) { + try { + return commitOnce(catalog, tableId, schemas, config, handling, committer); + } catch (CommitFailedException e) { + if (attempt >= MAX_ATTEMPTS) { + 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(); + checkState( + TypeUtil.assignIncreasingFreshIds(txn.table().schema()) + .sameSchema(TypeUtil.assignIncreasingFreshIds(merged)), + "replaying the folded schema union for %s diverged from the fold", + tableId); + } + 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) { + 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..7f44f75aff4f --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java @@ -0,0 +1,762 @@ +/* + * 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())); + } + + @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())); + } +} From b2d8222bfa986cc1611eafc60b1ece3d471f010f Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 14 Sep 2026 11:56:20 -0400 Subject: [PATCH 2/2] comments --- .../sdk/io/iceberg/CommitSchemaUnion.java | 51 ++++++++++++++++--- .../sdk/io/iceberg/CommitSchemaUnionTest.java | 18 +++++++ 2 files changed, 63 insertions(+), 6 deletions(-) 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 634e363c3858..8ffc4780f6e3 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 @@ -24,6 +24,10 @@ 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; @@ -39,6 +43,7 @@ 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; @@ -86,10 +91,23 @@ private static final class Incompatible { @Override public String toString() { - return files + " file(s) with schema " + schemaJson + ": " + reason; + 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() {} /** @@ -104,11 +122,25 @@ static long commit( 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) { - if (attempt >= MAX_ATTEMPTS) { + try { + if (!BackOffUtils.next(Sleeper.DEFAULT, backoff)) { + throw e; + } + } catch (InterruptedException interrupted) { + Thread.currentThread().interrupt(); throw e; } LOG.info( @@ -163,11 +195,15 @@ private static long commitOnce( // 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( - TypeUtil.assignIncreasingFreshIds(txn.table().schema()) - .sameSchema(TypeUtil.assignIncreasingFreshIds(merged)), - "replaying the folded schema union for %s diverged from the fold", - tableId); + 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); @@ -249,6 +285,9 @@ private static Transaction stageAll( 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) { 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 7f44f75aff4f..b0323064d38e 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 @@ -572,6 +572,24 @@ public void testFailPipelineCommitsNothingWhenAnySchemaIsIncompatible() { 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 =