From dd87ab021b26b404efce6e41c5daf4d5b860d069 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 26 Aug 2026 09:20:29 -0400 Subject: [PATCH 01/10] AddFiles: SchemaDelta classifies what a file schema needs from the table The classifier behind the options. Iceberg's unionByNameWith has no knobs: it adds, relaxes and promotes in one go, or throws. To honour ALLOW_FIELD_ADDITION / ALLOW_FIELD_RELAXATION / ALLOW_TYPE_PROMOTION separately, SchemaDelta.classify(table, fileSchema) applies the union on a throwaway UpdateSchema (apply(), never commit()), diffs the result against the current table schema, and labels every change: - FIELD_ADDITION: a field id present only after the union. - FIELD_RELAXATION: required before, optional after; or a required table column with no counterpart in the file at all (see below). - TYPE_PROMOTION: same id, wider primitive type after. - CONFLICT: anything else. The union throwing (ValidationException or IllegalArgumentException, e.g. int column vs string file column, a dotted or empty file column name), a field removed by the union (cannot happen with unionByName but is refused rather than trusted), a struct where a primitive was, a doc string or default changing, a promotion Iceberg would not allow (TypeUtil.isPromotionAllowed guard, so a bad union result is never staged as a "promotion"). The diff is keyed by field id and walks fields attribute by attribute (name, optionality, type kind, doc, defaults), so an attribute the union silently changes is reported rather than committed unnoticed. Changes are listed in a deterministic order (unquoted path) with quoted names in messages so a reviewer can find them in the schema. Absence rule: a required table column that the file lacks is a FIELD_RELAXATION, not a pass. Registering such a file would put nulls in a required column for every reader; the fix is to relax the column explicitly (the commit side stages makeColumnOptional for exactly the paths absentRequiredPaths() reports). The walk descends through structs whose parent is present, through list elements and map values (paths use "element" and "value", which makeColumnOptional accepts); an absent struct is itself the relaxation, its children are not listed separately; map keys are required by definition and skipped. Pins: Change.allowedBy(config) refuses a relaxation of a pinned path even when ALLOW_FIELD_RELAXATION is set, and disallowedReason(config) names it, so "id is pinned" shows up as the reason rather than a generic "relaxation not allowed". --- .../beam/sdk/io/iceberg/SchemaDelta.java | 610 ++++++++++++++++++ .../beam/sdk/io/iceberg/SchemaDeltaTest.java | 603 +++++++++++++++++ 2 files changed, 1213 insertions(+) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java new file mode 100644 index 000000000000..0d59254d9a54 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +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; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + *

The union ignores table columns absent from the file, but every row of such a file reads null + * in them, so a required column absent from the file is also a relaxation. The commit side stages + * those explicitly via {@link #absentRequiredPaths()}. + */ +final class SchemaDelta { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { + if (pin.startsWith(path + ".")) { + return pin; + } + } + return null; + } + + String disallowedReason(SchemaEvolutionConfig config) { + @Nullable String pin = kind == Kind.FIELD_RELAXATION ? forbiddingPin(config) : null; + if (pin != null) { + if (pin.equals(path)) { + return description + " (pinned as required)"; + } + return description + " (ancestor of pinned column " + pin + ")"; + } + return description + " (needs " + kind.option + ")"; + } + } + + private final List changes; + + private SchemaDelta(List changes) { + this.changes = Collections.unmodifiableList(changes); + } + + static SchemaDelta classify(Table table, Schema fileSchema) { + Schema before = table.schema(); + if (before.sameSchema(fileSchema)) { + return new SchemaDelta(Collections.emptyList()); + } + + List nameConflicts = new ArrayList<>(); + findInvalidNames(fileSchema.asStruct(), "", nameConflicts); + findCaseCollisions(before.asStruct(), fileSchema.asStruct(), "", nameConflicts); + if (!nameConflicts.isEmpty()) { + return new SchemaDelta(nameConflicts); + } + + List absent = new ArrayList<>(); + findAbsentRequired(before.asStruct(), fileSchema.asStruct(), "", absent); + Schema merged; + try { + // The absent-path relaxations are applied here too, so anything Iceberg refuses (an + // identifier field, say) is classified as this file's conflict instead of surfacing + // mid-transaction under a cross-schema message. + UpdateSchema update = table.updateSchema().unionByNameWith(fileSchema); + for (Change change : absent) { + update = update.makeColumnOptional(change.path); + } + merged = update.apply(); + } catch (ValidationException | IllegalArgumentException e) { + // SchemaUpdate reports type conflicts through both exception types + return conflict(e.getClass().getSimpleName() + ": " + AddFiles.errorMessage(e)); + } + Map absentByPath = new HashMap<>(); + for (Change change : absent) { + absentByPath.put(change.path, change); + } + return diff(before, merged, absentByPath); + } + + /** + * File column names no table can absorb, checked at every level including structs the table does + * not have yet. A literal dot is a conflict because Iceberg's name APIs, pins, aliases and + * ignores all treat the dot as a path separator, and a colliding struct in a later window would + * make the whole table unresolvable by name; rejected whether or not it collides today. An empty + * name would otherwise be added as a real column (the union only rejects it at the top level). + * Two file columns at one level differing only in case would be added as two columns, after which + * Iceberg cannot build the lower-case name index. + */ + private static void findInvalidNames( + Types.StructType struct, String prefix, List changes) { + Map seenByLowerCase = new HashMap<>(); + for (Types.NestedField field : struct.fields()) { + String rawPath = prefix + field.name(); + if (field.name().isEmpty()) { + String at = prefix.isEmpty() ? "" : " under " + prefix.substring(0, prefix.length() - 1); + changes.add(new Change(Kind.CONFLICT, rawPath, "empty column name" + at)); + } else if (field.name().contains(".")) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "column name " + + quoteIfDotted(field.name()) + + " contains '.', which Iceberg treats as a path separator; rename the column" + + " at its source")); + } + @Nullable String seen = + seenByLowerCase.put(field.name().toLowerCase(Locale.ROOT), field.name()); + if (seen != null) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "columns " + + prefix + + quoteIfDotted(seen) + + " and " + + prefix + + quoteIfDotted(field.name()) + + " differ only in case; rename one or map it with a column alias")); + } + findInvalidNamesInType(field.type(), rawPath, changes); + } + } + + private static void findInvalidNamesInType(Type type, String rawPath, List changes) { + if (type.isStructType()) { + findInvalidNames(type.asStructType(), rawPath + ".", changes); + } else if (type.isListType()) { + findInvalidNamesInType(type.asListType().elementType(), rawPath + ".element", changes); + } else if (type.isMapType()) { + findInvalidNamesInType(type.asMapType().valueType(), rawPath + ".value", changes); + } + } + + /** + * A file column whose name matches a table column at the same level only case-insensitively would + * be added as a separate column, after which Iceberg cannot build the lower-case name index and + * every case-insensitive reader of the table fails. + */ + private static void findCaseCollisions( + Types.StructType tableStruct, + Types.StructType fileStruct, + String prefix, + List changes) { + for (Types.NestedField fileField : fileStruct.fields()) { + String rawPath = prefix + fileField.name(); + Types.NestedField exact = tableStruct.field(fileField.name()); + if (exact == null) { + for (Types.NestedField tableField : tableStruct.fields()) { + if (tableField.name().equalsIgnoreCase(fileField.name())) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "column " + + prefix + + quoteIfDotted(fileField.name()) + + " differs only in case from table column " + + quoteIfDotted(tableField.name()) + + "; rename it or map it with a column alias")); + break; + } + } + continue; + } + findCaseCollisionsInType(exact.type(), fileField.type(), rawPath, changes); + } + } + + private static void findCaseCollisionsInType( + Type tableType, Type fileType, String rawPath, List changes) { + if (tableType.isStructType() && fileType.isStructType()) { + findCaseCollisions(tableType.asStructType(), fileType.asStructType(), rawPath + ".", changes); + } else if (tableType.isListType() && fileType.isListType()) { + findCaseCollisionsInType( + tableType.asListType().elementType(), + fileType.asListType().elementType(), + rawPath + ".element", + changes); + } else if (tableType.isMapType() && fileType.isMapType()) { + findCaseCollisionsInType( + tableType.asMapType().valueType(), + fileType.asMapType().valueType(), + rawPath + ".value", + changes); + } + } + + /** + * Required table columns with no counterpart in the file, by name per level. Children are checked + * only when their parent is present; an absent struct is the relaxation itself. Descends through + * list elements and map values (paths use {@code element} and {@code value}, which + * makeColumnOptional accepts); map keys are required by definition. FileSchemas tightening stops + * at lists and maps for a different reason (ambiguous null counts); the two are independent. + */ + private static void findAbsentRequired( + Types.StructType tableStruct, + Types.StructType fileStruct, + String prefix, + List changes) { + for (Types.NestedField field : tableStruct.fields()) { + String rawPath = prefix + field.name(); + Types.NestedField fileField = fileStruct.field(field.name()); + if (fileField == null) { + if (field.isRequired()) { + changes.add( + new Change( + Kind.FIELD_RELAXATION, + rawPath, + "relax " + + prefix + + quoteIfDotted(field.name()) + + " to optional (absent from file)", + true)); + } + continue; + } + findAbsentRequiredInType(field.type(), fileField.type(), rawPath, changes); + } + } + + private static void findAbsentRequiredInType( + Type tableType, Type fileType, String rawPath, List changes) { + if (tableType.isStructType() && fileType.isStructType()) { + findAbsentRequired(tableType.asStructType(), fileType.asStructType(), rawPath + ".", changes); + } else if (tableType.isListType() && fileType.isListType()) { + findAbsentRequiredInType( + tableType.asListType().elementType(), + fileType.asListType().elementType(), + rawPath + ".element", + changes); + } else if (tableType.isMapType() && fileType.isMapType()) { + findAbsentRequiredInType( + tableType.asMapType().valueType(), + fileType.asMapType().valueType(), + rawPath + ".value", + changes); + } + } + + /** Paths of required table columns absent from the file; the union alone does not relax them. */ + List absentRequiredPaths() { + List paths = new ArrayList<>(); + for (Change change : changes) { + if (change.absent) { + paths.add(change.path); + } + } + return paths; + } + + private static SchemaDelta conflict(String message) { + List changes = new ArrayList<>(); + changes.add(new Change(Kind.CONFLICT, "", message)); + return new SchemaDelta(changes); + } + + /** + * Changes from {@code before} to {@code after}, ordered by field path. Fields are matched by id; + * paths only appear in messages (quoted when a name contains a dot). Anything a union by name + * cannot produce is reported as a conflict so it is never applied unclassified. + */ + static SchemaDelta diff(Schema before, Schema after) { + return diff(before, after, Collections.emptyMap()); + } + + /** + * {@code absentByPath}: classify's absent-column relaxations, emitted here in path order where + * the diff sees the required-to-optional flip that classify itself staged. + */ + private static SchemaDelta diff(Schema before, Schema after, Map absentByPath) { + Map absentRemaining = new HashMap<>(absentByPath); + Map beforeById = TypeUtil.indexById(before.asStruct()); + Map afterById = TypeUtil.indexById(after.asStruct()); + Map parentById = TypeUtil.indexParents(after.asStruct()); + Map rawPathById = TypeUtil.indexNameById(after.asStruct()); + Map pathById = + TypeUtil.indexQuotedNameById(after.asStruct(), SchemaDelta::quoteIfDotted); + + List idsByPath = new ArrayList<>(afterById.keySet()); + idsByPath.sort( + (a, b) -> + checkStateNotNull(rawPathById.get(a)).compareTo(checkStateNotNull(rawPathById.get(b)))); + + List changes = new ArrayList<>(); + for (Integer id : idsByPath) { + String path = checkStateNotNull(pathById.get(id)); + String rawPath = checkStateNotNull(rawPathById.get(id)); + Types.NestedField newField = checkStateNotNull(afterById.get(id)); + Types.NestedField oldField = beforeById.get(id); + if (oldField == null) { + if (!hasAddedAncestor(id, parentById, beforeById)) { + changes.add( + new Change( + Kind.FIELD_ADDITION, + rawPath, + "add " + optionality(newField) + " " + path + " " + describe(newField.type()))); + } + continue; + } + compareField(path, rawPath, oldField, newField, absentRemaining, changes); + } + checkState( + absentRemaining.isEmpty(), + "absent-column relaxations did not surface in the diff: %s", + absentRemaining.keySet()); + + Map beforePathById = + TypeUtil.indexQuotedNameById(before.asStruct(), SchemaDelta::quoteIfDotted); + List removed = new ArrayList<>(); + for (Integer id : beforeById.keySet()) { + if (!afterById.containsKey(id)) { + removed.add(checkStateNotNull(beforePathById.get(id))); + } + } + Collections.sort(removed); + for (String path : removed) { + changes.add(new Change(Kind.CONFLICT, "", "field removed: " + path)); + } + return new SchemaDelta(changes); + } + + /** + * Attribute by attribute: name, doc and defaults must be equal; required to optional is the + * relaxation; primitive types must be equal or a promotion; nested types must stay the same kind, + * their children are compared on their own ids. + */ + private static void compareField( + String path, + String rawPath, + Types.NestedField oldField, + Types.NestedField newField, + Map absentRemaining, + List changes) { + if (!oldField.name().equals(newField.name())) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "renamed " + path + " from " + oldField.name() + " to " + newField.name())); + } + if (!Objects.equals(oldField.doc(), newField.doc())) { + // benign but unsupported: schema evolution has no option for doc updates + changes.add( + new Change(Kind.CONFLICT, rawPath, "doc changed on " + path + " (not supported)")); + } + if (!Objects.equals(oldField.initialDefault(), newField.initialDefault()) + || !Objects.equals(oldField.writeDefault(), newField.writeDefault())) { + changes.add( + new Change(Kind.CONFLICT, rawPath, "default changed on " + path + " (not supported)")); + } + if (oldField.isRequired() && newField.isOptional()) { + @Nullable Change absent = absentRemaining.remove(rawPath); + changes.add( + absent != null + ? absent + : new Change(Kind.FIELD_RELAXATION, rawPath, "relax " + path + " to optional")); + } else if (oldField.isOptional() && newField.isRequired()) { + changes.add(new Change(Kind.CONFLICT, rawPath, "optionality tightened on " + path)); + } + boolean oldPrimitive = oldField.type().isPrimitiveType(); + boolean newPrimitive = newField.type().isPrimitiveType(); + if (oldPrimitive && newPrimitive) { + if (oldField.type().equals(newField.type())) { + return; + } + if (TypeUtil.isPromotionAllowed(oldField.type(), newField.type().asPrimitiveType())) { + changes.add( + new Change( + Kind.TYPE_PROMOTION, + rawPath, + "promote " + path + " " + oldField.type() + " to " + newField.type())); + } else { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "type changed on " + + path + + " from " + + oldField.type() + + " to " + + newField.type() + + " (not a promotion)")); + } + } else if (oldPrimitive != newPrimitive + || oldField.type().typeId() != newField.type().typeId()) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "type changed on " + + path + + " from " + + describe(oldField.type()) + + " to " + + describe(newField.type()))); + } + } + + /** Renders a type without field ids: file-side ids are positional and would only mislead. */ + private static String describe(Type type) { + if (type.isStructType()) { + StringBuilder rendered = new StringBuilder("struct<"); + List fields = type.asStructType().fields(); + for (int i = 0; i < fields.size(); i++) { + Types.NestedField field = fields.get(i); + if (i > 0) { + rendered.append(", "); + } + rendered + .append(quoteIfDotted(field.name())) + .append(": ") + .append(optionality(field)) + .append(" ") + .append(describe(field.type())); + } + return rendered.append(">").toString(); + } + if (type.isListType()) { + return "list<" + describe(type.asListType().elementType()) + ">"; + } + if (type.isMapType()) { + Types.MapType map = type.asMapType(); + return "map<" + describe(map.keyType()) + ", " + describe(map.valueType()) + ">"; + } + return type.toString(); + } + + /** A field added inside a newly added struct is reported once, as part of its ancestor. */ + private static boolean hasAddedAncestor( + int id, Map parentById, Map beforeById) { + Integer parent = parentById.get(id); + while (parent != null) { + if (!beforeById.containsKey(parent)) { + return true; + } + parent = parentById.get(parent); + } + return false; + } + + private static String quoteIfDotted(String name) { + if (name.contains(".")) { + return "`" + name + "`"; + } + return name; + } + + private static String optionality(Types.NestedField field) { + return field.isOptional() ? "optional" : "required"; + } + + boolean isEmpty() { + return changes.isEmpty(); + } + + Set kinds() { + Set kinds = EnumSet.noneOf(Kind.class); + for (Change change : changes) { + kinds.add(change.kind); + } + return kinds; + } + + List descriptions() { + List descriptions = new ArrayList<>(); + for (Change change : changes) { + descriptions.add(change.description); + } + return Collections.unmodifiableList(descriptions); + } + + @Nullable String conflict() { + for (Change change : changes) { + if (change.kind == Kind.CONFLICT) { + return change.description; + } + } + return null; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + for (Change change : changes) { + if (!change.allowedBy(config)) { + return false; + } + } + return true; + } + + /** Why {@link #allowedBy} is false; empty when it is true. */ + String disallowedReason(SchemaEvolutionConfig config) { + List conflicts = new ArrayList<>(); + for (Change change : changes) { + if (change.kind == Kind.CONFLICT) { + conflicts.add(change.description); + } + } + if (!conflicts.isEmpty()) { + return "file schema conflicts with the table schema: " + String.join("; ", conflicts); + } + List disallowed = new ArrayList<>(); + for (Change change : changes) { + if (!change.allowedBy(config)) { + disallowed.add(change.disallowedReason(config)); + } + } + if (disallowed.isEmpty()) { + return ""; + } + return "file schema needs changes that are not allowed: " + String.join("; ", disallowed); + } + + @Override + public String toString() { + return "SchemaDelta" + descriptions(); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java new file mode 100644 index 000000000000..26c759fb8ebc --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java @@ -0,0 +1,603 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +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.assertNull; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.EnumSet; +import java.util.HashSet; +import org.apache.beam.sdk.io.iceberg.SchemaDelta.Kind; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.catalog.TableIdentifier; +import org.apache.iceberg.types.Types; +import org.junit.ClassRule; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; +import org.junit.rules.TestName; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class SchemaDeltaTest { + @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()), + optional( + 4, + "address", + Types.StructType.of( + required(5, "city", Types.StringType.get()), + optional(6, "zip", Types.IntegerType.get()))), + optional(7, "tags", Types.ListType.ofOptional(8, Types.StringType.get())), + optional(9, "amount", Types.DecimalType.of(9, 2))); + + private static final SchemaEvolutionConfig ALL = + SchemaEvolutionConfig.of(SchemaEvolutionOption.values()); + + private Table table; + private Schema tableCreatedWith; + + private SchemaDelta classify(Schema fileSchema) { + return classify(TABLE, fileSchema); + } + + private SchemaDelta classify(Schema tableSchema, Schema fileSchema) { + if (table == null) { + table = + warehouse.createTable( + TableIdentifier.of("default", testName.getMethodName()), tableSchema); + tableCreatedWith = tableSchema; + } else if (tableCreatedWith != null) { + assertTrue( + "classify already created the table with a different schema", + tableCreatedWith.sameSchema(tableSchema)); + } + return SchemaDelta.classify(table, fileSchema); + } + + private static SchemaEvolutionConfig pinned(String... columns) { + return SchemaEvolutionConfig.builder() + .setOptions(EnumSet.allOf(SchemaEvolutionOption.class)) + .setRequiredColumns(new HashSet<>(Arrays.asList(columns))) + .build(); + } + + // ---- empty deltas + + @Test + public void testIdenticalSchemaIsEmpty() { + assertTrue(classify(TABLE).isEmpty()); + assertTrue(classify(TABLE).allowedBy(SchemaEvolutionConfig.disabled())); + // The catalog renumbered TABLE's nested ids, so the calls above walk the full diff; only a + // schema with the table's own ids takes the sameSchema fast path. + Table created = checkStateNotNull(table); + assertTrue(SchemaDelta.classify(created, created.schema()).isEmpty()); + } + + // ---- required columns absent from the file + + @Test + public void testAbsentRequiredColumnIsRelaxation() { + Schema file = new Schema(optional(1, "name", Types.StringType.get())); + SchemaDelta delta = classify(file); + assertEquals(EnumSet.of(Kind.FIELD_RELAXATION), delta.kinds()); + assertEquals(Arrays.asList("relax id to optional (absent from file)"), delta.descriptions()); + assertEquals(Arrays.asList("id"), delta.absentRequiredPaths()); + assertTrue( + delta.allowedBy(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION))); + assertFalse( + delta.allowedBy(SchemaEvolutionConfig.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION))); + assertFalse(delta.allowedBy(pinned("id"))); + assertEquals( + "file schema needs changes that are not allowed: " + + "relax id to optional (absent from file) (pinned as required)", + delta.disallowedReason(pinned("id"))); + } + + @Test + public void testAbsentNestedRequiredChildIsRelaxation() { + Schema file = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, "address", Types.StructType.of(optional(3, "zip", Types.IntegerType.get())))); + SchemaDelta delta = classify(file); + assertEquals( + Arrays.asList("relax address.city to optional (absent from file)"), delta.descriptions()); + assertEquals(Arrays.asList("address.city"), delta.absentRequiredPaths()); + } + + @Test + public void testAbsentRequiredStructIsOneRelaxation() { + Schema tableSchema = + new Schema( + required(1, "id", Types.LongType.get()), + required( + 2, "address", Types.StructType.of(required(3, "city", Types.StringType.get())))); + Schema file = new Schema(required(1, "id", Types.LongType.get())); + SchemaDelta delta = classify(tableSchema, file); + assertEquals( + Arrays.asList("relax address to optional (absent from file)"), delta.descriptions()); + } + + /** Pins, absent-path reporting and makeColumnOptional share the element/value path spelling. */ + @Test + public void testAbsentRequiredUnderListAndMapIsRelaxation() { + Schema tableSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "items", + Types.ListType.ofOptional( + 3, Types.StructType.of(required(4, "sku", Types.StringType.get())))), + optional( + 5, + "attrs", + Types.MapType.ofOptional( + 6, + 7, + Types.StringType.get(), + Types.StructType.of(required(8, "v", Types.StringType.get()))))); + Schema file = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "items", + Types.ListType.ofOptional( + 3, Types.StructType.of(optional(4, "qty", Types.IntegerType.get())))), + optional( + 5, + "attrs", + Types.MapType.ofOptional( + 6, + 7, + Types.StringType.get(), + Types.StructType.of(optional(8, "w", Types.StringType.get()))))); + SchemaDelta delta = classify(tableSchema, file); + assertEquals( + Arrays.asList( + "relax attrs.value.v to optional (absent from file)", + "add optional attrs.value.w string", + "add optional items.element.qty int", + "relax items.element.sku to optional (absent from file)"), + delta.descriptions()); + assertEquals(Arrays.asList("attrs.value.v", "items.element.sku"), delta.absentRequiredPaths()); + assertFalse(delta.allowedBy(pinned("items.element.sku"))); + assertEquals( + "file schema needs changes that are not allowed: " + + "relax items.element.sku to optional (absent from file) (pinned as required)", + delta.disallowedReason(pinned("items.element.sku"))); + } + + @Test + public void testMultipleAbsentRequiredColumns() { + Schema tableSchema = + new Schema( + required(1, "id", Types.LongType.get()), + required(2, "region", Types.StringType.get()), + optional(3, "name", Types.StringType.get())); + Schema file = new Schema(optional(1, "name", Types.StringType.get())); + SchemaDelta delta = classify(tableSchema, file); + assertEquals( + Arrays.asList( + "relax id to optional (absent from file)", + "relax region to optional (absent from file)"), + delta.descriptions()); + assertEquals(Arrays.asList("id", "region"), delta.absentRequiredPaths()); + } + + /** + * Pins classify's own staging of makeColumnOptional inside the try: Iceberg's identifier-field + * refusal must come out classified as this file's conflict, not thrown mid-transaction. + */ + @Test + public void testAbsentRequiredIdentifierFieldIsConflict() { + Schema tableSchema = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "name", Types.StringType.get())); + table = + warehouse.createTable(TableIdentifier.of("default", testName.getMethodName()), tableSchema); + table.updateSchema().setIdentifierFields("id").commit(); + Schema file = new Schema(optional(1, "name", Types.StringType.get())); + SchemaDelta delta = SchemaDelta.classify(table, file); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + assertNotNull(delta.conflict()); + } + + // ---- additions + + /** A required new column is still added optional (addColumn always adds optional). */ + @Test + public void testTopLevelAddition() { + Schema file = + new Schema( + required(1, "email", Types.StringType.get()), required(2, "id", Types.LongType.get())); + SchemaDelta delta = classify(file); + assertEquals(EnumSet.of(Kind.FIELD_ADDITION), delta.kinds()); + assertEquals(Arrays.asList("add optional email string"), delta.descriptions()); + // classify's apply() never commits: the table is untouched. + assertNull(checkStateNotNull(table).schema().findField("email")); + } + + @Test + public void testNestedAddition() { + Schema file = + new Schema( + required(3, "id", Types.LongType.get()), + optional( + 1, + "address", + Types.StructType.of( + required(4, "city", Types.StringType.get()), + optional(2, "country", Types.StringType.get())))); + SchemaDelta delta = classify(file); + assertEquals(EnumSet.of(Kind.FIELD_ADDITION), delta.kinds()); + assertEquals(Arrays.asList("add optional address.country string"), delta.descriptions()); + } + + @Test + public void testAddedStructIsReportedOnce() { + Schema file = + new Schema( + required(9, "id", Types.LongType.get()), + optional( + 1, + "geo", + Types.StructType.of( + optional(2, "lat", Types.DoubleType.get()), + optional(3, "lon", Types.DoubleType.get())))); + SchemaDelta delta = classify(file); + assertEquals( + Arrays.asList("add optional geo struct"), + delta.descriptions()); + } + + // ---- relaxations and pins + + @Test + public void testRelaxations() { + Schema file = + new Schema( + optional(1, "id", Types.LongType.get()), + optional( + 2, "address", Types.StructType.of(optional(3, "city", Types.StringType.get())))); + SchemaDelta delta = classify(file); + assertEquals(EnumSet.of(Kind.FIELD_RELAXATION), delta.kinds()); + assertEquals( + Arrays.asList("relax address.city to optional", "relax id to optional"), + delta.descriptions()); + assertTrue(delta.allowedBy(ALL)); + assertFalse(delta.allowedBy(pinned("address.city"))); + assertEquals( + "file schema needs changes that are not allowed: " + + "relax address.city to optional (pinned as required)", + delta.disallowedReason(pinned("address.city"))); + } + + @Test + public void testRelaxingTheAncestorOfAPinnedColumnIsRefused() { + Schema tableSchema = + new Schema( + required(1, "id", Types.LongType.get()), + required( + 2, "address", Types.StructType.of(required(3, "city", Types.StringType.get())))); + Schema file = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, "address", Types.StructType.of(required(3, "city", Types.StringType.get())))); + SchemaDelta delta = classify(tableSchema, file); + assertFalse(delta.allowedBy(pinned("address.city"))); + assertEquals( + "file schema needs changes that are not allowed: " + + "relax address to optional (ancestor of pinned column address.city)", + delta.disallowedReason(pinned("address.city"))); + } + + // ---- promotion + + @Test + public void testPromotions() { + Schema file = + new Schema( + required(9, "id", Types.LongType.get()), + optional(1, "score", Types.DoubleType.get()), + optional(2, "amount", Types.DecimalType.of(18, 2)), + optional( + 3, + "address", + Types.StructType.of( + required(5, "city", Types.StringType.get()), + optional(4, "zip", Types.LongType.get())))); + SchemaDelta delta = classify(file); + assertEquals(EnumSet.of(Kind.TYPE_PROMOTION), delta.kinds()); + assertEquals( + Arrays.asList( + "promote address.zip int to long", + "promote amount decimal(9, 2) to decimal(18, 2)", + "promote score float to double"), + delta.descriptions()); + } + + // ---- combined and gating + + @Test + public void testCombinedDeltaReportsEveryKind() { + Schema file = + new Schema( + optional(1, "id", Types.LongType.get()), + optional(2, "score", Types.DoubleType.get()), + optional(3, "email", Types.StringType.get())); + SchemaDelta delta = classify(file); + assertEquals( + EnumSet.of(Kind.FIELD_ADDITION, Kind.FIELD_RELAXATION, Kind.TYPE_PROMOTION), delta.kinds()); + assertNull(delta.conflict()); + assertTrue(delta.allowedBy(ALL)); + assertFalse( + delta.allowedBy( + SchemaEvolutionConfig.of( + SchemaEvolutionOption.ALLOW_FIELD_ADDITION, + SchemaEvolutionOption.ALLOW_TYPE_PROMOTION))); + assertEquals( + "file schema needs changes that are not allowed: " + + "relax id to optional (needs ALLOW_FIELD_RELAXATION)", + delta.disallowedReason( + SchemaEvolutionConfig.of( + SchemaEvolutionOption.ALLOW_FIELD_ADDITION, + SchemaEvolutionOption.ALLOW_TYPE_PROMOTION))); + assertEquals("", delta.disallowedReason(ALL)); + } + + // ---- names the table cannot absorb + + @Test + public void testDottedColumnNamesAreConflicts() { + // colliding with an existing struct, top-level with no collision, and nested + Schema colliding = new Schema(optional(1, "address.zip", Types.IntegerType.get())); + SchemaDelta delta = classify(colliding); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + String reason = delta.disallowedReason(ALL); + assertTrue(reason, reason.contains("path separator")); + + Schema flat = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "x.y", Types.StringType.get())); + assertEquals(EnumSet.of(Kind.CONFLICT), classify(flat).kinds()); + + Schema nested = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "a", Types.StructType.of(optional(3, "b.c", Types.StringType.get())))); + SchemaDelta nestedDelta = classify(nested); + assertEquals(nestedDelta.toString(), EnumSet.of(Kind.CONFLICT), nestedDelta.kinds()); + String nestedReason = nestedDelta.disallowedReason(ALL); + assertTrue(nestedReason, nestedReason.contains("`b.c`")); + } + + /** The union rejects an empty name only at the top level; nested ones would be added. */ + @Test + public void testEmptyColumnNamesAreConflicts() { + Schema file = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "", Types.StringType.get()), + optional( + 3, + "address", + Types.StructType.of( + optional(4, "city", Types.StringType.get()), + optional(5, "", Types.StringType.get())))); + SchemaDelta delta = classify(file); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + assertEquals(2, delta.descriptions().size()); + String reason = delta.disallowedReason(ALL); + assertTrue(reason, reason.contains("empty column name under address")); + } + + @Test + public void testCaseOnlyDifferenceFromTableIsConflict() { + Schema flat = + new Schema( + optional(1, "NAME", Types.StringType.get()), required(2, "id", Types.LongType.get())); + SchemaDelta delta = classify(flat); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + String reason = delta.disallowedReason(ALL); + assertTrue(reason, reason.contains("differs only in case from table column name")); + + Schema nested = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, "address", Types.StructType.of(optional(3, "CITY", Types.StringType.get())))); + SchemaDelta nestedDelta = classify(nested); + assertEquals(nestedDelta.toString(), EnumSet.of(Kind.CONFLICT), nestedDelta.kinds()); + String nestedReason = nestedDelta.disallowedReason(ALL); + assertTrue( + nestedReason, + nestedReason.contains("address.CITY differs only in case from table column city")); + } + + /** Two new columns differing only in case would break the lower-case index between them. */ + @Test + public void testFileInternalCaseCollisionIsConflict() { + Schema flat = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "email", Types.StringType.get()), + optional(3, "EMAIL", Types.StringType.get())); + SchemaDelta delta = classify(flat); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + String reason = delta.disallowedReason(ALL); + assertTrue(reason, reason.contains("email and EMAIL differ only in case")); + + Schema insideNewStruct = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "geo", + Types.StructType.of( + optional(3, "lat", Types.DoubleType.get()), + optional(4, "LAT", Types.DoubleType.get())))); + SchemaDelta nested = classify(insideNewStruct); + assertEquals(nested.toString(), EnumSet.of(Kind.CONFLICT), nested.kinds()); + String nestedReason = nested.disallowedReason(ALL); + assertTrue(nestedReason, nestedReason.contains("geo.lat and geo.LAT differ only in case")); + } + + // ---- Iceberg behaviors classify depends on; after a version bump failure, start here + + /** The union never tightens: a reordered, subset, stricter-optionality file changes nothing. */ + @Test + public void testReorderedSubsetTighterFileIsCovered() { + Schema file = + new Schema( + required(1, "name", Types.StringType.get()), required(2, "id", Types.LongType.get())); + assertTrue(classify(file).isEmpty()); + } + + /** + * Iceberg 1.11's union ignores a file primitive that promotes to the table's type + * (UnionByNameVisitor.isIgnorableTypeUpdate): readers widen narrower files on read. + */ + @Test + public void testNarrowerFileTypeIsCovered() { + Schema file = new Schema(required(1, "id", Types.IntegerType.get())); + assertTrue(classify(file).isEmpty()); + } + + /** The union throws for an impossible type change; classify catches and classifies it. */ + @Test + public void testTypeMismatchIsConflict() { + Schema file = new Schema(optional(1, "name", Types.IntegerType.get())); + SchemaDelta delta = classify(file); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + assertNotNull(delta.conflict()); + assertFalse(delta.allowedBy(ALL)); + String reason = delta.disallowedReason(ALL); + assertTrue(reason, reason.startsWith("file schema conflicts with the table schema: ")); + } + + /** + * Unreachable through AddFiles today (Parquet conversion never sets docs), pinned as deliberate: + * the union would silently rewrite the table's doc, so a doc-bearing file schema must conflict. + */ + @Test + public void testDocBearingFileSchemaIsConflict() { + Schema file = new Schema(required(1, "id", Types.LongType.get(), "the id")); + SchemaDelta delta = classify(file); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + String reason = delta.disallowedReason(ALL); + assertTrue(reason, reason.contains("doc changed on id")); + } + + // ---- diff sanity checks, driven directly + + /** + * No classify input reaches these branches (a union never removes, tightens, renames, narrows or + * edits defaults; it throws first), but the "never applied unclassified" contract says diff must + * flag them if Iceberg ever changes. + */ + @Test + public void testDiffFlagsChangesTheUnionCannotProduce() { + Schema id = new Schema(required(1, "id", Types.LongType.get())); + + Schema withStruct = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "s", + Types.StructType.of( + optional(3, "x", Types.IntegerType.get()), + optional(4, "y", Types.IntegerType.get())))); + SchemaDelta removed = SchemaDelta.diff(withStruct, id); + assertFalse(removed.allowedBy(ALL)); + assertEquals( + "file schema conflicts with the table schema: " + + "field removed: s; field removed: s.x; field removed: s.y", + removed.disallowedReason(ALL)); + + Schema optionalA = + new Schema( + optional(1, "a", Types.StructType.of(optional(2, "b", Types.IntegerType.get())))); + Schema requiredA = + new Schema( + required(1, "a", Types.StructType.of(optional(2, "b", Types.IntegerType.get())))); + assertEquals( + Arrays.asList("optionality tightened on a"), + SchemaDelta.diff(optionalA, requiredA).descriptions()); + + Schema renamed = new Schema(required(1, "id2", Types.LongType.get())); + assertEquals( + Arrays.asList("renamed id2 from id to id2"), SchemaDelta.diff(id, renamed).descriptions()); + + Schema narrowed = new Schema(required(1, "id", Types.IntegerType.get())); + SchemaDelta narrowing = SchemaDelta.diff(id, narrowed); + assertEquals( + Arrays.asList("type changed on id from long to int (not a promotion)"), + narrowing.descriptions()); + assertEquals(EnumSet.of(Kind.CONFLICT), narrowing.kinds()); + + Schema defaulted = + new Schema( + Types.NestedField.optional("id") + .withId(1) + .ofType(Types.LongType.get()) + .withWriteDefault(org.apache.iceberg.expressions.Literal.of(7L)) + .build()); + assertEquals( + EnumSet.of(Kind.CONFLICT, Kind.FIELD_RELAXATION), SchemaDelta.diff(id, defaulted).kinds()); + } + + /** Iceberg rejects a schema where a dotted name equals a nested path, so only quoting matters. */ + @Test + public void testDottedNameIsQuotedAndDoesNotSwallowSiblings() { + Schema before = new Schema(required(1, "id", Types.LongType.get())); + Schema after = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "a.b", Types.StringType.get()), + optional(3, "a", Types.StructType.of(optional(4, "c", Types.IntegerType.get())))); + SchemaDelta delta = SchemaDelta.diff(before, after); + assertEquals( + Arrays.asList("add optional a struct", "add optional `a.b` string"), + delta.descriptions()); + } +} From 2da6b33fd5ff5391fb7246e39b1a37c9408eaaad Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 17:34:55 -0400 Subject: [PATCH 02/10] move pins to class --- .../org/apache/beam/sdk/io/iceberg/Pins.java | 60 +++++++++++++++++++ .../beam/sdk/io/iceberg/SchemaDelta.java | 34 ++++------- 2 files changed, 70 insertions(+), 24 deletions(-) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java new file mode 100644 index 000000000000..c31a10ccd3a8 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java @@ -0,0 +1,60 @@ +/* + * 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 java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * The pinned columns of a {@link SchemaEvolutionConfig}, as path segments. + */ +final class Pins { + private final List> segments; + private final List dotted; + + Pins(Collection requiredColumns) { + this.segments = new ArrayList<>(); + this.dotted = new ArrayList<>(); + for (String column : requiredColumns) { + segments.add(Arrays.asList(column.split("\\.", -1))); + dotted.add(column); + } + } + + /** + * Returns the pinned column that forbids relaxing {@code dottedPath} to optional, or null + * if none does. + * + *

A pin forbids relaxing the pinned column itself, and also any struct above it. + * + *

Relaxing a column below a pinned one is allowed. + */ + @Nullable String forbiddingRelaxationOf(String dottedPath) { + List path = Arrays.asList(dottedPath.split("\\.", -1)); + for (int i = 0; i < segments.size(); i++) { + List pin = segments.get(i); + if (pin.size() >= path.size() && pin.subList(0, path.size()).equals(path)) { + return dotted.get(i); + } + } + return null; + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index 0d59254d9a54..98c73e44a54f 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -89,32 +89,16 @@ private static final class Change { this.absent = absent; } - boolean allowedBy(SchemaEvolutionConfig config) { - if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + boolean allowedBy(SchemaEvolutionConfig config, Pins pins) { + if (kind == Kind.FIELD_RELAXATION && pins.forbiddingRelaxationOf(path) != null) { return false; } return kind.allowedBy(config); } - /** - * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A - * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that - * fail the pin check at registration. - */ - private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { - if (config.isPinned(path)) { - return path; - } - for (String pin : config.getRequiredColumns()) { - if (pin.startsWith(path + ".")) { - return pin; - } - } - return null; - } - - String disallowedReason(SchemaEvolutionConfig config) { - @Nullable String pin = kind == Kind.FIELD_RELAXATION ? forbiddingPin(config) : null; + String disallowedReason(Pins pins) { + @Nullable String pin = + kind == Kind.FIELD_RELAXATION ? pins.forbiddingRelaxationOf(path) : null; if (pin != null) { if (pin.equals(path)) { return description + " (pinned as required)"; @@ -572,8 +556,9 @@ List descriptions() { } boolean allowedBy(SchemaEvolutionConfig config) { + Pins pins = new Pins(config.getRequiredColumns()); for (Change change : changes) { - if (!change.allowedBy(config)) { + if (!change.allowedBy(config, pins)) { return false; } } @@ -591,10 +576,11 @@ String disallowedReason(SchemaEvolutionConfig config) { if (!conflicts.isEmpty()) { return "file schema conflicts with the table schema: " + String.join("; ", conflicts); } + Pins pins = new Pins(config.getRequiredColumns()); List disallowed = new ArrayList<>(); for (Change change : changes) { - if (!change.allowedBy(config)) { - disallowed.add(change.disallowedReason(config)); + if (!change.allowedBy(config, pins)) { + disallowed.add(change.disallowedReason(pins)); } } if (disallowed.isEmpty()) { From 3ca2a6e8ff4379a3cd731d57617b85a44c448847 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 17:43:28 -0400 Subject: [PATCH 03/10] pins --- .../org/apache/beam/sdk/io/iceberg/Pins.java | 28 +++++++++---------- .../beam/sdk/io/iceberg/SchemaDelta.java | 15 ++++++---- .../beam/sdk/io/iceberg/SchemaDeltaTest.java | 6 ++++ 3 files changed, 29 insertions(+), 20 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java index c31a10ccd3a8..b516bfeefed2 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java @@ -20,38 +20,38 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; +import java.util.Collections; import java.util.List; import org.checkerframework.checker.nullness.qual.Nullable; -/** - * The pinned columns of a {@link SchemaEvolutionConfig}, as path segments. - */ +/** The pinned columns of a {@link SchemaEvolutionConfig}, as path segments. */ final class Pins { private final List> segments; private final List dotted; Pins(Collection requiredColumns) { + this.dotted = new ArrayList<>(requiredColumns); + Collections.sort(dotted); this.segments = new ArrayList<>(); - this.dotted = new ArrayList<>(); - for (String column : requiredColumns) { + for (String column : dotted) { segments.add(Arrays.asList(column.split("\\.", -1))); - dotted.add(column); } } + /** Whether {@code dottedPath} itself is pinned. */ + boolean isPinned(String dottedPath) { + return dotted.contains(dottedPath); + } + /** - * Returns the pinned column that forbids relaxing {@code dottedPath} to optional, or null - * if none does. - * - *

A pin forbids relaxing the pinned column itself, and also any struct above it. - * - *

Relaxing a column below a pinned one is allowed. + * Returns the pinned column strictly below {@code dottedPath} (the lexicographically first when + * several are), or null when there is none. Columns below a pin, or beside it, have none. */ - @Nullable String forbiddingRelaxationOf(String dottedPath) { + @Nullable String pinnedColumnBeneath(String dottedPath) { List path = Arrays.asList(dottedPath.split("\\.", -1)); for (int i = 0; i < segments.size(); i++) { List pin = segments.get(i); - if (pin.size() >= path.size() && pin.subList(0, path.size()).equals(path)) { + if (pin.size() > path.size() && pin.subList(0, path.size()).equals(path)) { return dotted.get(i); } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index 98c73e44a54f..7872b177fe1c 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -90,20 +90,23 @@ private static final class Change { } boolean allowedBy(SchemaEvolutionConfig config, Pins pins) { - if (kind == Kind.FIELD_RELAXATION && pins.forbiddingRelaxationOf(path) != null) { + // A pin also forbids relaxing the structs above it: a null ancestor nulls the pinned leaf. + if (kind == Kind.FIELD_RELAXATION + && (pins.isPinned(path) || pins.pinnedColumnBeneath(path) != null)) { return false; } return kind.allowedBy(config); } String disallowedReason(Pins pins) { - @Nullable String pin = - kind == Kind.FIELD_RELAXATION ? pins.forbiddingRelaxationOf(path) : null; - if (pin != null) { - if (pin.equals(path)) { + if (kind == Kind.FIELD_RELAXATION) { + if (pins.isPinned(path)) { return description + " (pinned as required)"; } - return description + " (ancestor of pinned column " + pin + ")"; + @Nullable String pin = pins.pinnedColumnBeneath(path); + if (pin != null) { + return description + " (ancestor of pinned column " + pin + ")"; + } } return description + " (needs " + kind.option + ")"; } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java index 26c759fb8ebc..833c3df72af8 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java @@ -328,6 +328,12 @@ public void testRelaxingTheAncestorOfAPinnedColumnIsRefused() { "file schema needs changes that are not allowed: " + "relax address to optional (ancestor of pinned column address.city)", delta.disallowedReason(pinned("address.city"))); + // When several pins forbid the same relaxation, the lexicographically first is reported, + // independent of the set's iteration order. + assertEquals( + "file schema needs changes that are not allowed: " + + "relax address to optional (ancestor of pinned column address.city)", + delta.disallowedReason(pinned("address.zip", "address.city"))); } // ---- promotion From e2680ef249805f055e60d063d22627e79e9055f5 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 17:44:45 -0400 Subject: [PATCH 04/10] docstring --- .../java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index 7872b177fe1c..452a50586cd4 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -118,6 +118,11 @@ private SchemaDelta(List changes) { this.changes = Collections.unmodifiableList(changes); } + /** + * What registering a file with {@code fileSchema} would need from the table, as changes ordered + * by column path; the table itself is never modified. File column names no table can absorb + * (dotted, empty, case-colliding) come back as conflicts without attempting the union. + */ static SchemaDelta classify(Table table, Schema fileSchema) { Schema before = table.schema(); if (before.sameSchema(fileSchema)) { From 8689e203fcd20b45a64ac1cd931e53dc47790f37 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 18:01:54 -0400 Subject: [PATCH 05/10] Fix docstring --- .../beam/sdk/io/iceberg/SchemaDelta.java | 22 ++++++++++--------- 1 file changed, 12 insertions(+), 10 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index 452a50586cd4..abec6689d954 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -160,13 +160,14 @@ static SchemaDelta classify(Table table, Schema fileSchema) { } /** - * File column names no table can absorb, checked at every level including structs the table does - * not have yet. A literal dot is a conflict because Iceberg's name APIs, pins, aliases and - * ignores all treat the dot as a path separator, and a colliding struct in a later window would - * make the whole table unresolvable by name; rejected whether or not it collides today. An empty - * name would otherwise be added as a real column (the union only rejects it at the top level). - * Two file columns at one level differing only in case would be added as two columns, after which - * Iceberg cannot build the lower-case name index. + * Adds a conflict for every file column name no table can absorb, at every level including + * structs the table does not have yet: names containing a literal dot, empty names, and pairs of + * names at one level differing only in case. A dot is a conflict because Iceberg's name APIs, + * pins, aliases and ignores all treat it as a path separator, and a colliding struct in a later + * window would make the whole table unresolvable by name; rejected whether or not it collides + * today. An empty name would otherwise be added as a real column (the union only rejects it at + * the top level). A case-only pair would be added as two columns, after which Iceberg cannot + * build the lower-case name index. */ private static void findInvalidNames( Types.StructType struct, String prefix, List changes) { @@ -216,9 +217,10 @@ private static void findInvalidNamesInType(Type type, String rawPath, List Date: Thu, 10 Sep 2026 18:16:07 -0400 Subject: [PATCH 06/10] tests --- .../beam/sdk/io/iceberg/SchemaDelta.java | 10 +- .../beam/sdk/io/iceberg/SchemaDeltaTest.java | 266 ++++++++++++++---- 2 files changed, 214 insertions(+), 62 deletions(-) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index abec6689d954..b950a5c5641c 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -29,6 +29,7 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; import org.apache.iceberg.Schema; import org.apache.iceberg.Table; import org.apache.iceberg.UpdateSchema; @@ -67,7 +68,7 @@ boolean allowedBy(SchemaEvolutionConfig config) { } } - private static final class Change { + static final class Change { final Kind kind; /** Unquoted column path for the config lookup; empty for conflicts without a field. */ @@ -169,8 +170,8 @@ static SchemaDelta classify(Table table, Schema fileSchema) { * the top level). A case-only pair would be added as two columns, after which Iceberg cannot * build the lower-case name index. */ - private static void findInvalidNames( - Types.StructType struct, String prefix, List changes) { + @VisibleForTesting + static void findInvalidNames(Types.StructType struct, String prefix, List changes) { Map seenByLowerCase = new HashMap<>(); for (Types.NestedField field : struct.fields()) { String rawPath = prefix + field.name(); @@ -222,7 +223,8 @@ private static void findInvalidNamesInType(Type type, String rawPath, List invalidNames(Types.StructType fileStruct) { + List changes = new ArrayList<>(); + SchemaDelta.findInvalidNames(fileStruct, "", changes); + return conflictDescriptions(changes); + } + + private static List caseCollisions( + Types.StructType tableStruct, Types.StructType fileStruct) { + List changes = new ArrayList<>(); + SchemaDelta.findCaseCollisions(tableStruct, fileStruct, "", changes); + return conflictDescriptions(changes); + } + + private static List conflictDescriptions(List changes) { + List descriptions = new ArrayList<>(); + for (SchemaDelta.Change change : changes) { + assertEquals(Kind.CONFLICT, change.kind); + descriptions.add(change.description); + } + return descriptions; + } + + @Test + public void testFindInvalidNamesFlagsDottedNamesAtEveryLevel() { + Types.StructType file = + Types.StructType.of( + optional(1, "a.b", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "c.d", Types.IntegerType.get()))), optional( - 2, - "geo", - Types.StructType.of( - optional(3, "lat", Types.DoubleType.get()), - optional(4, "LAT", Types.DoubleType.get())))); - SchemaDelta nested = classify(insideNewStruct); - assertEquals(nested.toString(), EnumSet.of(Kind.CONFLICT), nested.kinds()); - String nestedReason = nested.disallowedReason(ALL); - assertTrue(nestedReason, nestedReason.contains("geo.lat and geo.LAT differ only in case")); + 4, + "l", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "e.f", Types.StringType.get())))), + optional( + 7, + "m", + Types.MapType.ofOptional( + 8, + 9, + Types.StringType.get(), + Types.StructType.of(optional(10, "g.h", Types.StringType.get()))))); + List conflicts = invalidNames(file); + assertEquals(conflicts.toString(), 4, conflicts.size()); + for (String name : Arrays.asList("`a.b`", "`c.d`", "`e.f`", "`g.h`")) { + assertTrue(conflicts.toString(), conflicts.toString().contains(name)); + } + } + + @Test + public void testFindInvalidNamesFlagsEmptyNamesAtEveryLevel() { + Types.StructType file = + Types.StructType.of( + optional(1, "", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "", Types.IntegerType.get()))), + optional( + 4, + "l", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "", Types.StringType.get())))), + optional( + 7, + "m", + Types.MapType.ofOptional( + 8, + 9, + Types.StringType.get(), + Types.StructType.of(optional(10, "", Types.StringType.get()))))); + assertEquals( + Arrays.asList( + "empty column name", + "empty column name under s", + "empty column name under l.element", + "empty column name under m.value"), + invalidNames(file)); + } + + @Test + public void testFindInvalidNamesFlagsCaseDuplicatesPerLevel() { + Types.StructType file = + Types.StructType.of( + optional(1, "email", Types.StringType.get()), + optional(2, "EMAIL", Types.StringType.get()), + optional( + 3, + "l", + Types.ListType.ofOptional( + 4, + Types.StructType.of( + optional(5, "lat", Types.DoubleType.get()), + optional(6, "LAT", Types.DoubleType.get()))))); + List conflicts = invalidNames(file); + assertEquals(conflicts.toString(), 2, conflicts.size()); + assertTrue( + conflicts.toString(), conflicts.get(0).contains("email and EMAIL differ only in case")); + assertTrue( + conflicts.toString(), + conflicts.get(1).contains("l.element.lat and l.element.LAT differ only in case")); + } + + /** The duplicate rule is per level: the same name at different levels is fine. */ + @Test + public void testFindInvalidNamesAcceptsCleanSchemas() { + Types.StructType file = + Types.StructType.of( + optional(1, "name", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "NAME", Types.StringType.get())))); + assertEquals(Collections.emptyList(), invalidNames(file)); + } + + @Test + public void testFindCaseCollisionsAtEveryLevel() { + Types.StructType table = + Types.StructType.of( + optional(1, "name", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "city", Types.StringType.get()))), + optional( + 4, + "l", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "sku", Types.StringType.get())))), + optional( + 7, + "m", + Types.MapType.ofOptional( + 8, + 9, + Types.StringType.get(), + Types.StructType.of(optional(10, "v", Types.StringType.get()))))); + Types.StructType file = + Types.StructType.of( + optional(1, "NAME", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "CITY", Types.StringType.get()))), + optional( + 4, + "l", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "SKU", Types.StringType.get())))), + optional( + 7, + "m", + Types.MapType.ofOptional( + 8, + 9, + Types.StringType.get(), + Types.StructType.of(optional(10, "V", Types.StringType.get()))))); + assertEquals( + Arrays.asList( + "column NAME differs only in case from table column name;" + + " rename it or map it with a column alias", + "column s.CITY differs only in case from table column city;" + + " rename it or map it with a column alias", + "column l.element.SKU differs only in case from table column sku;" + + " rename it or map it with a column alias", + "column m.value.V differs only in case from table column v;" + + " rename it or map it with a column alias"), + caseCollisions(table, file)); + } + + @Test + public void testFindCaseCollisionsPassesExactNewAndKindMismatchedNames() { + Types.StructType table = + Types.StructType.of( + optional(1, "name", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "x", Types.IntegerType.get())))); + Types.StructType file = + Types.StructType.of( + optional(1, "name", Types.StringType.get()), + optional(2, "email", Types.StringType.get()), + optional(3, "s", Types.StringType.get())); + assertEquals(Collections.emptyList(), caseCollisions(table, file)); } // ---- Iceberg behaviors classify depends on; after a version bump failure, start here @@ -590,6 +719,27 @@ public void testDiffFlagsChangesTheUnionCannotProduce() { .build()); assertEquals( EnumSet.of(Kind.CONFLICT, Kind.FIELD_RELAXATION), SchemaDelta.diff(id, defaulted).kinds()); + + Schema structOfX = + new Schema( + optional(1, "s", Types.StructType.of(optional(2, "x", Types.IntegerType.get())))); + Schema primitiveS = new Schema(optional(1, "s", Types.StringType.get())); + assertEquals( + Arrays.asList( + "type changed on s from struct to string", "field removed: s.x"), + SchemaDelta.diff(structOfX, primitiveS).descriptions()); + + Schema withContainers = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "attrs", + Types.MapType.ofOptional(3, 4, Types.StringType.get(), Types.IntegerType.get())), + optional(5, "tags", Types.ListType.ofOptional(6, Types.StringType.get()))); + assertEquals( + Arrays.asList("add optional attrs map", "add optional tags list"), + SchemaDelta.diff(id, withContainers).descriptions()); } /** Iceberg rejects a schema where a dotted name equals a nested path, so only quoting matters. */ From 013eaa83fdcb5f6e0214b8ee43c7356c16a76072 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 18:26:35 -0400 Subject: [PATCH 07/10] split file --- .../beam/sdk/io/iceberg/ColumnNameChecks.java | 148 +++++++++++++ .../beam/sdk/io/iceberg/SchemaDelta.java | 122 +--------- .../sdk/io/iceberg/ColumnNameChecksTest.java | 208 ++++++++++++++++++ .../beam/sdk/io/iceberg/SchemaDeltaTest.java | 177 --------------- 4 files changed, 360 insertions(+), 295 deletions(-) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java new file mode 100644 index 000000000000..4dd57fd4f6c8 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java @@ -0,0 +1,148 @@ +/* + * 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 java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import org.apache.beam.sdk.io.iceberg.SchemaDelta.Change; +import org.apache.beam.sdk.io.iceberg.SchemaDelta.Kind; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Rejections of file column names no table can absorb, run by {@link SchemaDelta#classify} before + * the union is attempted so the conflict is attributed to the offending file, with a message naming + * the column. + */ +final class ColumnNameChecks { + private ColumnNameChecks() {} + + /** + * Adds a conflict for every file column name no table can absorb, at every level including + * structs the table does not have yet: names containing a literal dot, empty names, and pairs of + * names at one level differing only in case. A dot is a conflict because Iceberg's name APIs, + * pins, aliases and ignores all treat it as a path separator, and a colliding struct in a later + * window would make the whole table unresolvable by name; rejected whether or not it collides + * today. An empty name would otherwise be added as a real column (the union only rejects it at + * the top level). A case-only pair would be added as two columns, after which Iceberg cannot + * build the lower-case name index. + */ + static void findInvalidNames(Types.StructType struct, String prefix, List changes) { + Map seenByLowerCase = new HashMap<>(); + for (Types.NestedField field : struct.fields()) { + String rawPath = prefix + field.name(); + if (field.name().isEmpty()) { + String at = prefix.isEmpty() ? "" : " under " + prefix.substring(0, prefix.length() - 1); + changes.add(new Change(Kind.CONFLICT, rawPath, "empty column name" + at)); + } else if (field.name().contains(".")) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "column name " + + SchemaDelta.quoteIfDotted(field.name()) + + " contains '.', which Iceberg treats as a path separator; rename the column" + + " at its source")); + } + @Nullable String seen = + seenByLowerCase.put(field.name().toLowerCase(Locale.ROOT), field.name()); + if (seen != null) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "columns " + + prefix + + SchemaDelta.quoteIfDotted(seen) + + " and " + + prefix + + SchemaDelta.quoteIfDotted(field.name()) + + " differ only in case; rename one or map it with a column alias")); + } + findInvalidNamesInType(field.type(), rawPath, changes); + } + } + + private static void findInvalidNamesInType(Type type, String rawPath, List changes) { + if (type.isStructType()) { + findInvalidNames(type.asStructType(), rawPath + ".", changes); + } else if (type.isListType()) { + findInvalidNamesInType(type.asListType().elementType(), rawPath + ".element", changes); + } else if (type.isMapType()) { + findInvalidNamesInType(type.asMapType().valueType(), rawPath + ".value", changes); + } + } + + /** + * Adds a conflict for every file column whose name matches a table column at the same level only + * case-insensitively; exact matches and genuinely new names pass. Such a column would be added as + * a separate column, after which Iceberg cannot build the lower-case name index and every + * case-insensitive reader of the table fails. + */ + static void findCaseCollisions( + Types.StructType tableStruct, + Types.StructType fileStruct, + String prefix, + List changes) { + for (Types.NestedField fileField : fileStruct.fields()) { + String rawPath = prefix + fileField.name(); + Types.NestedField exact = tableStruct.field(fileField.name()); + if (exact == null) { + for (Types.NestedField tableField : tableStruct.fields()) { + if (tableField.name().equalsIgnoreCase(fileField.name())) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "column " + + prefix + + SchemaDelta.quoteIfDotted(fileField.name()) + + " differs only in case from table column " + + SchemaDelta.quoteIfDotted(tableField.name()) + + "; rename it or map it with a column alias")); + break; + } + } + continue; + } + findCaseCollisionsInType(exact.type(), fileField.type(), rawPath, changes); + } + } + + private static void findCaseCollisionsInType( + Type tableType, Type fileType, String rawPath, List changes) { + if (tableType.isStructType() && fileType.isStructType()) { + findCaseCollisions(tableType.asStructType(), fileType.asStructType(), rawPath + ".", changes); + } else if (tableType.isListType() && fileType.isListType()) { + findCaseCollisionsInType( + tableType.asListType().elementType(), + fileType.asListType().elementType(), + rawPath + ".element", + changes); + } else if (tableType.isMapType() && fileType.isMapType()) { + findCaseCollisionsInType( + tableType.asMapType().valueType(), + fileType.asMapType().valueType(), + rawPath + ".value", + changes); + } + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index b950a5c5641c..d2e4d4819df4 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -25,11 +25,9 @@ import java.util.EnumSet; import java.util.HashMap; import java.util.List; -import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Set; -import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.annotations.VisibleForTesting; import org.apache.iceberg.Schema; import org.apache.iceberg.Table; import org.apache.iceberg.UpdateSchema; @@ -131,8 +129,9 @@ static SchemaDelta classify(Table table, Schema fileSchema) { } List nameConflicts = new ArrayList<>(); - findInvalidNames(fileSchema.asStruct(), "", nameConflicts); - findCaseCollisions(before.asStruct(), fileSchema.asStruct(), "", nameConflicts); + ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "", nameConflicts); + ColumnNameChecks.findCaseCollisions( + before.asStruct(), fileSchema.asStruct(), "", nameConflicts); if (!nameConflicts.isEmpty()) { return new SchemaDelta(nameConflicts); } @@ -160,119 +159,6 @@ static SchemaDelta classify(Table table, Schema fileSchema) { return diff(before, merged, absentByPath); } - /** - * Adds a conflict for every file column name no table can absorb, at every level including - * structs the table does not have yet: names containing a literal dot, empty names, and pairs of - * names at one level differing only in case. A dot is a conflict because Iceberg's name APIs, - * pins, aliases and ignores all treat it as a path separator, and a colliding struct in a later - * window would make the whole table unresolvable by name; rejected whether or not it collides - * today. An empty name would otherwise be added as a real column (the union only rejects it at - * the top level). A case-only pair would be added as two columns, after which Iceberg cannot - * build the lower-case name index. - */ - @VisibleForTesting - static void findInvalidNames(Types.StructType struct, String prefix, List changes) { - Map seenByLowerCase = new HashMap<>(); - for (Types.NestedField field : struct.fields()) { - String rawPath = prefix + field.name(); - if (field.name().isEmpty()) { - String at = prefix.isEmpty() ? "" : " under " + prefix.substring(0, prefix.length() - 1); - changes.add(new Change(Kind.CONFLICT, rawPath, "empty column name" + at)); - } else if (field.name().contains(".")) { - changes.add( - new Change( - Kind.CONFLICT, - rawPath, - "column name " - + quoteIfDotted(field.name()) - + " contains '.', which Iceberg treats as a path separator; rename the column" - + " at its source")); - } - @Nullable String seen = - seenByLowerCase.put(field.name().toLowerCase(Locale.ROOT), field.name()); - if (seen != null) { - changes.add( - new Change( - Kind.CONFLICT, - rawPath, - "columns " - + prefix - + quoteIfDotted(seen) - + " and " - + prefix - + quoteIfDotted(field.name()) - + " differ only in case; rename one or map it with a column alias")); - } - findInvalidNamesInType(field.type(), rawPath, changes); - } - } - - private static void findInvalidNamesInType(Type type, String rawPath, List changes) { - if (type.isStructType()) { - findInvalidNames(type.asStructType(), rawPath + ".", changes); - } else if (type.isListType()) { - findInvalidNamesInType(type.asListType().elementType(), rawPath + ".element", changes); - } else if (type.isMapType()) { - findInvalidNamesInType(type.asMapType().valueType(), rawPath + ".value", changes); - } - } - - /** - * Adds a conflict for every file column whose name matches a table column at the same level only - * case-insensitively; exact matches and genuinely new names pass. Such a column would be added as - * a separate column, after which Iceberg cannot build the lower-case name index and every - * case-insensitive reader of the table fails. - */ - @VisibleForTesting - static void findCaseCollisions( - Types.StructType tableStruct, - Types.StructType fileStruct, - String prefix, - List changes) { - for (Types.NestedField fileField : fileStruct.fields()) { - String rawPath = prefix + fileField.name(); - Types.NestedField exact = tableStruct.field(fileField.name()); - if (exact == null) { - for (Types.NestedField tableField : tableStruct.fields()) { - if (tableField.name().equalsIgnoreCase(fileField.name())) { - changes.add( - new Change( - Kind.CONFLICT, - rawPath, - "column " - + prefix - + quoteIfDotted(fileField.name()) - + " differs only in case from table column " - + quoteIfDotted(tableField.name()) - + "; rename it or map it with a column alias")); - break; - } - } - continue; - } - findCaseCollisionsInType(exact.type(), fileField.type(), rawPath, changes); - } - } - - private static void findCaseCollisionsInType( - Type tableType, Type fileType, String rawPath, List changes) { - if (tableType.isStructType() && fileType.isStructType()) { - findCaseCollisions(tableType.asStructType(), fileType.asStructType(), rawPath + ".", changes); - } else if (tableType.isListType() && fileType.isListType()) { - findCaseCollisionsInType( - tableType.asListType().elementType(), - fileType.asListType().elementType(), - rawPath + ".element", - changes); - } else if (tableType.isMapType() && fileType.isMapType()) { - findCaseCollisionsInType( - tableType.asMapType().valueType(), - fileType.asMapType().valueType(), - rawPath + ".value", - changes); - } - } - /** * Required table columns with no counterpart in the file, by name per level. Children are checked * only when their parent is present; an absent struct is the relaxation itself. Descends through @@ -527,7 +413,7 @@ private static boolean hasAddedAncestor( return false; } - private static String quoteIfDotted(String name) { + static String quoteIfDotted(String name) { if (name.contains(".")) { return "`" + name + "`"; } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java new file mode 100644 index 000000000000..9064e8e79f3f --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java @@ -0,0 +1,208 @@ +/* + * 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.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import org.apache.beam.sdk.io.iceberg.SchemaDelta.Kind; +import org.apache.iceberg.types.Types; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class ColumnNameChecksTest { + + private static List invalidNames(Types.StructType fileStruct) { + List changes = new ArrayList<>(); + ColumnNameChecks.findInvalidNames(fileStruct, "", changes); + return conflictDescriptions(changes); + } + + private static List caseCollisions( + Types.StructType tableStruct, Types.StructType fileStruct) { + List changes = new ArrayList<>(); + ColumnNameChecks.findCaseCollisions(tableStruct, fileStruct, "", changes); + return conflictDescriptions(changes); + } + + private static List conflictDescriptions(List changes) { + List descriptions = new ArrayList<>(); + for (SchemaDelta.Change change : changes) { + assertEquals(Kind.CONFLICT, change.kind); + descriptions.add(change.description); + } + return descriptions; + } + + @Test + public void testFindInvalidNamesFlagsDottedNamesAtEveryLevel() { + Types.StructType file = + Types.StructType.of( + optional(1, "a.b", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "c.d", Types.IntegerType.get()))), + optional( + 4, + "l", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "e.f", Types.StringType.get())))), + optional( + 7, + "m", + Types.MapType.ofOptional( + 8, + 9, + Types.StringType.get(), + Types.StructType.of(optional(10, "g.h", Types.StringType.get()))))); + List conflicts = invalidNames(file); + assertEquals(conflicts.toString(), 4, conflicts.size()); + for (String name : Arrays.asList("`a.b`", "`c.d`", "`e.f`", "`g.h`")) { + assertTrue(conflicts.toString(), conflicts.toString().contains(name)); + } + } + + @Test + public void testFindInvalidNamesFlagsEmptyNamesAtEveryLevel() { + Types.StructType file = + Types.StructType.of( + optional(1, "", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "", Types.IntegerType.get()))), + optional( + 4, + "l", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "", Types.StringType.get())))), + optional( + 7, + "m", + Types.MapType.ofOptional( + 8, + 9, + Types.StringType.get(), + Types.StructType.of(optional(10, "", Types.StringType.get()))))); + assertEquals( + Arrays.asList( + "empty column name", + "empty column name under s", + "empty column name under l.element", + "empty column name under m.value"), + invalidNames(file)); + } + + @Test + public void testFindInvalidNamesFlagsCaseDuplicatesPerLevel() { + Types.StructType file = + Types.StructType.of( + optional(1, "email", Types.StringType.get()), + optional(2, "EMAIL", Types.StringType.get()), + optional( + 3, + "l", + Types.ListType.ofOptional( + 4, + Types.StructType.of( + optional(5, "lat", Types.DoubleType.get()), + optional(6, "LAT", Types.DoubleType.get()))))); + List conflicts = invalidNames(file); + assertEquals(conflicts.toString(), 2, conflicts.size()); + assertTrue( + conflicts.toString(), conflicts.get(0).contains("email and EMAIL differ only in case")); + assertTrue( + conflicts.toString(), + conflicts.get(1).contains("l.element.lat and l.element.LAT differ only in case")); + } + + /** The duplicate rule is per level: the same name at different levels is fine. */ + @Test + public void testFindInvalidNamesAcceptsCleanSchemas() { + Types.StructType file = + Types.StructType.of( + optional(1, "name", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "NAME", Types.StringType.get())))); + assertEquals(Collections.emptyList(), invalidNames(file)); + } + + @Test + public void testFindCaseCollisionsAtEveryLevel() { + Types.StructType table = + Types.StructType.of( + optional(1, "name", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "city", Types.StringType.get()))), + optional( + 4, + "l", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "sku", Types.StringType.get())))), + optional( + 7, + "m", + Types.MapType.ofOptional( + 8, + 9, + Types.StringType.get(), + Types.StructType.of(optional(10, "v", Types.StringType.get()))))); + Types.StructType file = + Types.StructType.of( + optional(1, "NAME", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "CITY", Types.StringType.get()))), + optional( + 4, + "l", + Types.ListType.ofOptional( + 5, Types.StructType.of(optional(6, "SKU", Types.StringType.get())))), + optional( + 7, + "m", + Types.MapType.ofOptional( + 8, + 9, + Types.StringType.get(), + Types.StructType.of(optional(10, "V", Types.StringType.get()))))); + assertEquals( + Arrays.asList( + "column NAME differs only in case from table column name;" + + " rename it or map it with a column alias", + "column s.CITY differs only in case from table column city;" + + " rename it or map it with a column alias", + "column l.element.SKU differs only in case from table column sku;" + + " rename it or map it with a column alias", + "column m.value.V differs only in case from table column v;" + + " rename it or map it with a column alias"), + caseCollisions(table, file)); + } + + @Test + public void testFindCaseCollisionsPassesExactNewAndKindMismatchedNames() { + Types.StructType table = + Types.StructType.of( + optional(1, "name", Types.StringType.get()), + optional(2, "s", Types.StructType.of(optional(3, "x", Types.IntegerType.get())))); + Types.StructType file = + Types.StructType.of( + optional(1, "name", Types.StringType.get()), + optional(2, "email", Types.StringType.get()), + optional(3, "s", Types.StringType.get())); + assertEquals(Collections.emptyList(), caseCollisions(table, file)); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java index 37f996125442..6ae7704e29e4 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java @@ -26,12 +26,9 @@ import static org.junit.Assert.assertNull; 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.HashSet; -import java.util.List; import org.apache.beam.sdk.io.iceberg.SchemaDelta.Kind; import org.apache.iceberg.Schema; import org.apache.iceberg.Table; @@ -442,180 +439,6 @@ public void testFileInternalCaseCollisionIsConflict() { assertTrue(reason, reason.contains("email and EMAIL differ only in case")); } - // ---- the name walks, driven directly (pure, no table) - - private static List invalidNames(Types.StructType fileStruct) { - List changes = new ArrayList<>(); - SchemaDelta.findInvalidNames(fileStruct, "", changes); - return conflictDescriptions(changes); - } - - private static List caseCollisions( - Types.StructType tableStruct, Types.StructType fileStruct) { - List changes = new ArrayList<>(); - SchemaDelta.findCaseCollisions(tableStruct, fileStruct, "", changes); - return conflictDescriptions(changes); - } - - private static List conflictDescriptions(List changes) { - List descriptions = new ArrayList<>(); - for (SchemaDelta.Change change : changes) { - assertEquals(Kind.CONFLICT, change.kind); - descriptions.add(change.description); - } - return descriptions; - } - - @Test - public void testFindInvalidNamesFlagsDottedNamesAtEveryLevel() { - Types.StructType file = - Types.StructType.of( - optional(1, "a.b", Types.StringType.get()), - optional(2, "s", Types.StructType.of(optional(3, "c.d", Types.IntegerType.get()))), - optional( - 4, - "l", - Types.ListType.ofOptional( - 5, Types.StructType.of(optional(6, "e.f", Types.StringType.get())))), - optional( - 7, - "m", - Types.MapType.ofOptional( - 8, - 9, - Types.StringType.get(), - Types.StructType.of(optional(10, "g.h", Types.StringType.get()))))); - List conflicts = invalidNames(file); - assertEquals(conflicts.toString(), 4, conflicts.size()); - for (String name : Arrays.asList("`a.b`", "`c.d`", "`e.f`", "`g.h`")) { - assertTrue(conflicts.toString(), conflicts.toString().contains(name)); - } - } - - @Test - public void testFindInvalidNamesFlagsEmptyNamesAtEveryLevel() { - Types.StructType file = - Types.StructType.of( - optional(1, "", Types.StringType.get()), - optional(2, "s", Types.StructType.of(optional(3, "", Types.IntegerType.get()))), - optional( - 4, - "l", - Types.ListType.ofOptional( - 5, Types.StructType.of(optional(6, "", Types.StringType.get())))), - optional( - 7, - "m", - Types.MapType.ofOptional( - 8, - 9, - Types.StringType.get(), - Types.StructType.of(optional(10, "", Types.StringType.get()))))); - assertEquals( - Arrays.asList( - "empty column name", - "empty column name under s", - "empty column name under l.element", - "empty column name under m.value"), - invalidNames(file)); - } - - @Test - public void testFindInvalidNamesFlagsCaseDuplicatesPerLevel() { - Types.StructType file = - Types.StructType.of( - optional(1, "email", Types.StringType.get()), - optional(2, "EMAIL", Types.StringType.get()), - optional( - 3, - "l", - Types.ListType.ofOptional( - 4, - Types.StructType.of( - optional(5, "lat", Types.DoubleType.get()), - optional(6, "LAT", Types.DoubleType.get()))))); - List conflicts = invalidNames(file); - assertEquals(conflicts.toString(), 2, conflicts.size()); - assertTrue( - conflicts.toString(), conflicts.get(0).contains("email and EMAIL differ only in case")); - assertTrue( - conflicts.toString(), - conflicts.get(1).contains("l.element.lat and l.element.LAT differ only in case")); - } - - /** The duplicate rule is per level: the same name at different levels is fine. */ - @Test - public void testFindInvalidNamesAcceptsCleanSchemas() { - Types.StructType file = - Types.StructType.of( - optional(1, "name", Types.StringType.get()), - optional(2, "s", Types.StructType.of(optional(3, "NAME", Types.StringType.get())))); - assertEquals(Collections.emptyList(), invalidNames(file)); - } - - @Test - public void testFindCaseCollisionsAtEveryLevel() { - Types.StructType table = - Types.StructType.of( - optional(1, "name", Types.StringType.get()), - optional(2, "s", Types.StructType.of(optional(3, "city", Types.StringType.get()))), - optional( - 4, - "l", - Types.ListType.ofOptional( - 5, Types.StructType.of(optional(6, "sku", Types.StringType.get())))), - optional( - 7, - "m", - Types.MapType.ofOptional( - 8, - 9, - Types.StringType.get(), - Types.StructType.of(optional(10, "v", Types.StringType.get()))))); - Types.StructType file = - Types.StructType.of( - optional(1, "NAME", Types.StringType.get()), - optional(2, "s", Types.StructType.of(optional(3, "CITY", Types.StringType.get()))), - optional( - 4, - "l", - Types.ListType.ofOptional( - 5, Types.StructType.of(optional(6, "SKU", Types.StringType.get())))), - optional( - 7, - "m", - Types.MapType.ofOptional( - 8, - 9, - Types.StringType.get(), - Types.StructType.of(optional(10, "V", Types.StringType.get()))))); - assertEquals( - Arrays.asList( - "column NAME differs only in case from table column name;" - + " rename it or map it with a column alias", - "column s.CITY differs only in case from table column city;" - + " rename it or map it with a column alias", - "column l.element.SKU differs only in case from table column sku;" - + " rename it or map it with a column alias", - "column m.value.V differs only in case from table column v;" - + " rename it or map it with a column alias"), - caseCollisions(table, file)); - } - - @Test - public void testFindCaseCollisionsPassesExactNewAndKindMismatchedNames() { - Types.StructType table = - Types.StructType.of( - optional(1, "name", Types.StringType.get()), - optional(2, "s", Types.StructType.of(optional(3, "x", Types.IntegerType.get())))); - Types.StructType file = - Types.StructType.of( - optional(1, "name", Types.StringType.get()), - optional(2, "email", Types.StringType.get()), - optional(3, "s", Types.StringType.get())); - assertEquals(Collections.emptyList(), caseCollisions(table, file)); - } - // ---- Iceberg behaviors classify depends on; after a version bump failure, start here /** The union never tightens: a reordered, subset, stricter-optionality file changes nothing. */ From 6c690028a42dc08792a520988e271fb8d6d2c3d3 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 18:30:38 -0400 Subject: [PATCH 08/10] split --- .../beam/sdk/io/iceberg/ColumnNameChecks.java | 26 +-- .../beam/sdk/io/iceberg/SchemaChange.java | 91 ++++++++++ .../beam/sdk/io/iceberg/SchemaDelta.java | 161 ++++++------------ .../sdk/io/iceberg/ColumnNameChecksTest.java | 10 +- .../beam/sdk/io/iceberg/SchemaDeltaTest.java | 2 +- 5 files changed, 163 insertions(+), 127 deletions(-) create mode 100644 sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java index 4dd57fd4f6c8..c1e6aaf8c5a9 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java @@ -21,8 +21,6 @@ import java.util.List; import java.util.Locale; import java.util.Map; -import org.apache.beam.sdk.io.iceberg.SchemaDelta.Change; -import org.apache.beam.sdk.io.iceberg.SchemaDelta.Kind; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.checkerframework.checker.nullness.qual.Nullable; @@ -45,17 +43,18 @@ private ColumnNameChecks() {} * the top level). A case-only pair would be added as two columns, after which Iceberg cannot * build the lower-case name index. */ - static void findInvalidNames(Types.StructType struct, String prefix, List changes) { + static void findInvalidNames(Types.StructType struct, String prefix, List changes) { Map seenByLowerCase = new HashMap<>(); for (Types.NestedField field : struct.fields()) { String rawPath = prefix + field.name(); if (field.name().isEmpty()) { String at = prefix.isEmpty() ? "" : " under " + prefix.substring(0, prefix.length() - 1); - changes.add(new Change(Kind.CONFLICT, rawPath, "empty column name" + at)); + changes.add( + new SchemaChange(SchemaChange.Kind.CONFLICT, rawPath, "empty column name" + at)); } else if (field.name().contains(".")) { changes.add( - new Change( - Kind.CONFLICT, + new SchemaChange( + SchemaChange.Kind.CONFLICT, rawPath, "column name " + SchemaDelta.quoteIfDotted(field.name()) @@ -66,8 +65,8 @@ static void findInvalidNames(Types.StructType struct, String prefix, List changes) { + private static void findInvalidNamesInType( + Type type, String rawPath, List changes) { if (type.isStructType()) { findInvalidNames(type.asStructType(), rawPath + ".", changes); } else if (type.isListType()) { @@ -101,7 +101,7 @@ static void findCaseCollisions( Types.StructType tableStruct, Types.StructType fileStruct, String prefix, - List changes) { + List changes) { for (Types.NestedField fileField : fileStruct.fields()) { String rawPath = prefix + fileField.name(); Types.NestedField exact = tableStruct.field(fileField.name()); @@ -109,8 +109,8 @@ static void findCaseCollisions( for (Types.NestedField tableField : tableStruct.fields()) { if (tableField.name().equalsIgnoreCase(fileField.name())) { changes.add( - new Change( - Kind.CONFLICT, + new SchemaChange( + SchemaChange.Kind.CONFLICT, rawPath, "column " + prefix @@ -128,7 +128,7 @@ static void findCaseCollisions( } private static void findCaseCollisionsInType( - Type tableType, Type fileType, String rawPath, List changes) { + Type tableType, Type fileType, String rawPath, List changes) { if (tableType.isStructType() && fileType.isStructType()) { findCaseCollisions(tableType.asStructType(), fileType.asStructType(), rawPath + ".", changes); } else if (tableType.isListType() && fileType.isListType()) { diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java new file mode 100644 index 000000000000..cbcff264edaa --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java @@ -0,0 +1,91 @@ +/* + * 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 org.checkerframework.checker.nullness.qual.Nullable; + +/** + * One change that registering a file schema would make on the table, classified by the {@link + * SchemaEvolutionOption} it needs. Produced by {@link SchemaDelta#classify} (and, for name + * conflicts, {@link ColumnNameChecks}); {@link SchemaDelta} decides whether a file's set of changes + * is allowed by a {@link SchemaEvolutionConfig}. + */ +final class SchemaChange { + + /** The option a change needs to be allowed. */ + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + SchemaChange(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + SchemaChange(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config, Pins pins) { + // A pin also forbids relaxing the structs above it: a null ancestor nulls the pinned leaf. + if (kind == Kind.FIELD_RELAXATION + && (pins.isPinned(path) || pins.pinnedColumnBeneath(path) != null)) { + return false; + } + return kind.allowedBy(config); + } + + String disallowedReason(Pins pins) { + if (kind == Kind.FIELD_RELAXATION) { + if (pins.isPinned(path)) { + return description + " (pinned as required)"; + } + @Nullable String pin = pins.pinnedColumnBeneath(path); + if (pin != null) { + return description + " (ancestor of pinned column " + pin + ")"; + } + } + return description + " (needs " + kind.option + ")"; + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index d2e4d4819df4..bacdac2e7cfe 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -48,72 +48,9 @@ */ final class SchemaDelta { - enum Kind { - FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), - FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), - TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), - /** The union is impossible (for example string vs int); never allowed. */ - CONFLICT(null); + private final List changes; - final @Nullable SchemaEvolutionOption option; - - Kind(@Nullable SchemaEvolutionOption option) { - this.option = option; - } - - boolean allowedBy(SchemaEvolutionConfig config) { - return option != null && config.allows(option); - } - } - - static final class Change { - final Kind kind; - - /** Unquoted column path for the config lookup; empty for conflicts without a field. */ - final String path; - - final String description; - - /** A relaxation because the column is absent from the file, not declared optional. */ - final boolean absent; - - Change(Kind kind, String path, String description) { - this(kind, path, description, false); - } - - Change(Kind kind, String path, String description, boolean absent) { - this.kind = kind; - this.path = path; - this.description = description; - this.absent = absent; - } - - boolean allowedBy(SchemaEvolutionConfig config, Pins pins) { - // A pin also forbids relaxing the structs above it: a null ancestor nulls the pinned leaf. - if (kind == Kind.FIELD_RELAXATION - && (pins.isPinned(path) || pins.pinnedColumnBeneath(path) != null)) { - return false; - } - return kind.allowedBy(config); - } - - String disallowedReason(Pins pins) { - if (kind == Kind.FIELD_RELAXATION) { - if (pins.isPinned(path)) { - return description + " (pinned as required)"; - } - @Nullable String pin = pins.pinnedColumnBeneath(path); - if (pin != null) { - return description + " (ancestor of pinned column " + pin + ")"; - } - } - return description + " (needs " + kind.option + ")"; - } - } - - private final List changes; - - private SchemaDelta(List changes) { + private SchemaDelta(List changes) { this.changes = Collections.unmodifiableList(changes); } @@ -128,7 +65,7 @@ static SchemaDelta classify(Table table, Schema fileSchema) { return new SchemaDelta(Collections.emptyList()); } - List nameConflicts = new ArrayList<>(); + List nameConflicts = new ArrayList<>(); ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "", nameConflicts); ColumnNameChecks.findCaseCollisions( before.asStruct(), fileSchema.asStruct(), "", nameConflicts); @@ -136,7 +73,7 @@ static SchemaDelta classify(Table table, Schema fileSchema) { return new SchemaDelta(nameConflicts); } - List absent = new ArrayList<>(); + List absent = new ArrayList<>(); findAbsentRequired(before.asStruct(), fileSchema.asStruct(), "", absent); Schema merged; try { @@ -144,7 +81,7 @@ static SchemaDelta classify(Table table, Schema fileSchema) { // identifier field, say) is classified as this file's conflict instead of surfacing // mid-transaction under a cross-schema message. UpdateSchema update = table.updateSchema().unionByNameWith(fileSchema); - for (Change change : absent) { + for (SchemaChange change : absent) { update = update.makeColumnOptional(change.path); } merged = update.apply(); @@ -152,8 +89,8 @@ static SchemaDelta classify(Table table, Schema fileSchema) { // SchemaUpdate reports type conflicts through both exception types return conflict(e.getClass().getSimpleName() + ": " + AddFiles.errorMessage(e)); } - Map absentByPath = new HashMap<>(); - for (Change change : absent) { + Map absentByPath = new HashMap<>(); + for (SchemaChange change : absent) { absentByPath.put(change.path, change); } return diff(before, merged, absentByPath); @@ -170,15 +107,15 @@ private static void findAbsentRequired( Types.StructType tableStruct, Types.StructType fileStruct, String prefix, - List changes) { + List changes) { for (Types.NestedField field : tableStruct.fields()) { String rawPath = prefix + field.name(); Types.NestedField fileField = fileStruct.field(field.name()); if (fileField == null) { if (field.isRequired()) { changes.add( - new Change( - Kind.FIELD_RELAXATION, + new SchemaChange( + SchemaChange.Kind.FIELD_RELAXATION, rawPath, "relax " + prefix @@ -193,7 +130,7 @@ private static void findAbsentRequired( } private static void findAbsentRequiredInType( - Type tableType, Type fileType, String rawPath, List changes) { + Type tableType, Type fileType, String rawPath, List changes) { if (tableType.isStructType() && fileType.isStructType()) { findAbsentRequired(tableType.asStructType(), fileType.asStructType(), rawPath + ".", changes); } else if (tableType.isListType() && fileType.isListType()) { @@ -214,7 +151,7 @@ private static void findAbsentRequiredInType( /** Paths of required table columns absent from the file; the union alone does not relax them. */ List absentRequiredPaths() { List paths = new ArrayList<>(); - for (Change change : changes) { + for (SchemaChange change : changes) { if (change.absent) { paths.add(change.path); } @@ -223,8 +160,8 @@ List absentRequiredPaths() { } private static SchemaDelta conflict(String message) { - List changes = new ArrayList<>(); - changes.add(new Change(Kind.CONFLICT, "", message)); + List changes = new ArrayList<>(); + changes.add(new SchemaChange(SchemaChange.Kind.CONFLICT, "", message)); return new SchemaDelta(changes); } @@ -241,8 +178,9 @@ static SchemaDelta diff(Schema before, Schema after) { * {@code absentByPath}: classify's absent-column relaxations, emitted here in path order where * the diff sees the required-to-optional flip that classify itself staged. */ - private static SchemaDelta diff(Schema before, Schema after, Map absentByPath) { - Map absentRemaining = new HashMap<>(absentByPath); + private static SchemaDelta diff( + Schema before, Schema after, Map absentByPath) { + Map absentRemaining = new HashMap<>(absentByPath); Map beforeById = TypeUtil.indexById(before.asStruct()); Map afterById = TypeUtil.indexById(after.asStruct()); Map parentById = TypeUtil.indexParents(after.asStruct()); @@ -255,7 +193,7 @@ private static SchemaDelta diff(Schema before, Schema after, Map (a, b) -> checkStateNotNull(rawPathById.get(a)).compareTo(checkStateNotNull(rawPathById.get(b)))); - List changes = new ArrayList<>(); + List changes = new ArrayList<>(); for (Integer id : idsByPath) { String path = checkStateNotNull(pathById.get(id)); String rawPath = checkStateNotNull(rawPathById.get(id)); @@ -264,8 +202,8 @@ private static SchemaDelta diff(Schema before, Schema after, Map if (oldField == null) { if (!hasAddedAncestor(id, parentById, beforeById)) { changes.add( - new Change( - Kind.FIELD_ADDITION, + new SchemaChange( + SchemaChange.Kind.FIELD_ADDITION, rawPath, "add " + optionality(newField) + " " + path + " " + describe(newField.type()))); } @@ -288,7 +226,7 @@ private static SchemaDelta diff(Schema before, Schema after, Map } Collections.sort(removed); for (String path : removed) { - changes.add(new Change(Kind.CONFLICT, "", "field removed: " + path)); + changes.add(new SchemaChange(SchemaChange.Kind.CONFLICT, "", "field removed: " + path)); } return new SchemaDelta(changes); } @@ -303,33 +241,40 @@ private static void compareField( String rawPath, Types.NestedField oldField, Types.NestedField newField, - Map absentRemaining, - List changes) { + Map absentRemaining, + List changes) { if (!oldField.name().equals(newField.name())) { changes.add( - new Change( - Kind.CONFLICT, + new SchemaChange( + SchemaChange.Kind.CONFLICT, rawPath, "renamed " + path + " from " + oldField.name() + " to " + newField.name())); } if (!Objects.equals(oldField.doc(), newField.doc())) { // benign but unsupported: schema evolution has no option for doc updates changes.add( - new Change(Kind.CONFLICT, rawPath, "doc changed on " + path + " (not supported)")); + new SchemaChange( + SchemaChange.Kind.CONFLICT, rawPath, "doc changed on " + path + " (not supported)")); } if (!Objects.equals(oldField.initialDefault(), newField.initialDefault()) || !Objects.equals(oldField.writeDefault(), newField.writeDefault())) { changes.add( - new Change(Kind.CONFLICT, rawPath, "default changed on " + path + " (not supported)")); + new SchemaChange( + SchemaChange.Kind.CONFLICT, + rawPath, + "default changed on " + path + " (not supported)")); } if (oldField.isRequired() && newField.isOptional()) { - @Nullable Change absent = absentRemaining.remove(rawPath); + @Nullable SchemaChange absent = absentRemaining.remove(rawPath); changes.add( absent != null ? absent - : new Change(Kind.FIELD_RELAXATION, rawPath, "relax " + path + " to optional")); + : new SchemaChange( + SchemaChange.Kind.FIELD_RELAXATION, rawPath, "relax " + path + " to optional")); } else if (oldField.isOptional() && newField.isRequired()) { - changes.add(new Change(Kind.CONFLICT, rawPath, "optionality tightened on " + path)); + changes.add( + new SchemaChange( + SchemaChange.Kind.CONFLICT, rawPath, "optionality tightened on " + path)); } boolean oldPrimitive = oldField.type().isPrimitiveType(); boolean newPrimitive = newField.type().isPrimitiveType(); @@ -339,14 +284,14 @@ private static void compareField( } if (TypeUtil.isPromotionAllowed(oldField.type(), newField.type().asPrimitiveType())) { changes.add( - new Change( - Kind.TYPE_PROMOTION, + new SchemaChange( + SchemaChange.Kind.TYPE_PROMOTION, rawPath, "promote " + path + " " + oldField.type() + " to " + newField.type())); } else { changes.add( - new Change( - Kind.CONFLICT, + new SchemaChange( + SchemaChange.Kind.CONFLICT, rawPath, "type changed on " + path @@ -359,8 +304,8 @@ private static void compareField( } else if (oldPrimitive != newPrimitive || oldField.type().typeId() != newField.type().typeId()) { changes.add( - new Change( - Kind.CONFLICT, + new SchemaChange( + SchemaChange.Kind.CONFLICT, rawPath, "type changed on " + path @@ -428,9 +373,9 @@ boolean isEmpty() { return changes.isEmpty(); } - Set kinds() { - Set kinds = EnumSet.noneOf(Kind.class); - for (Change change : changes) { + Set kinds() { + Set kinds = EnumSet.noneOf(SchemaChange.Kind.class); + for (SchemaChange change : changes) { kinds.add(change.kind); } return kinds; @@ -438,15 +383,15 @@ Set kinds() { List descriptions() { List descriptions = new ArrayList<>(); - for (Change change : changes) { + for (SchemaChange change : changes) { descriptions.add(change.description); } return Collections.unmodifiableList(descriptions); } @Nullable String conflict() { - for (Change change : changes) { - if (change.kind == Kind.CONFLICT) { + for (SchemaChange change : changes) { + if (change.kind == SchemaChange.Kind.CONFLICT) { return change.description; } } @@ -455,7 +400,7 @@ List descriptions() { boolean allowedBy(SchemaEvolutionConfig config) { Pins pins = new Pins(config.getRequiredColumns()); - for (Change change : changes) { + for (SchemaChange change : changes) { if (!change.allowedBy(config, pins)) { return false; } @@ -466,8 +411,8 @@ boolean allowedBy(SchemaEvolutionConfig config) { /** Why {@link #allowedBy} is false; empty when it is true. */ String disallowedReason(SchemaEvolutionConfig config) { List conflicts = new ArrayList<>(); - for (Change change : changes) { - if (change.kind == Kind.CONFLICT) { + for (SchemaChange change : changes) { + if (change.kind == SchemaChange.Kind.CONFLICT) { conflicts.add(change.description); } } @@ -476,7 +421,7 @@ String disallowedReason(SchemaEvolutionConfig config) { } Pins pins = new Pins(config.getRequiredColumns()); List disallowed = new ArrayList<>(); - for (Change change : changes) { + for (SchemaChange change : changes) { if (!change.allowedBy(config, pins)) { disallowed.add(change.disallowedReason(pins)); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java index 9064e8e79f3f..3f48359cde50 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java @@ -25,7 +25,7 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; -import org.apache.beam.sdk.io.iceberg.SchemaDelta.Kind; +import org.apache.beam.sdk.io.iceberg.SchemaChange.Kind; import org.apache.iceberg.types.Types; import org.junit.Test; import org.junit.runner.RunWith; @@ -35,21 +35,21 @@ public class ColumnNameChecksTest { private static List invalidNames(Types.StructType fileStruct) { - List changes = new ArrayList<>(); + List changes = new ArrayList<>(); ColumnNameChecks.findInvalidNames(fileStruct, "", changes); return conflictDescriptions(changes); } private static List caseCollisions( Types.StructType tableStruct, Types.StructType fileStruct) { - List changes = new ArrayList<>(); + List changes = new ArrayList<>(); ColumnNameChecks.findCaseCollisions(tableStruct, fileStruct, "", changes); return conflictDescriptions(changes); } - private static List conflictDescriptions(List changes) { + private static List conflictDescriptions(List changes) { List descriptions = new ArrayList<>(); - for (SchemaDelta.Change change : changes) { + for (SchemaChange change : changes) { assertEquals(Kind.CONFLICT, change.kind); descriptions.add(change.description); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java index 6ae7704e29e4..d3d4172f0f95 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java @@ -29,7 +29,7 @@ import java.util.Arrays; import java.util.EnumSet; import java.util.HashSet; -import org.apache.beam.sdk.io.iceberg.SchemaDelta.Kind; +import org.apache.beam.sdk.io.iceberg.SchemaChange.Kind; import org.apache.iceberg.Schema; import org.apache.iceberg.Table; import org.apache.iceberg.catalog.TableIdentifier; From 2e1f11d4ae4fbae9feeb84bf04a307243c673844 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 18:36:16 -0400 Subject: [PATCH 09/10] map keys --- .../beam/sdk/io/iceberg/ColumnNameChecks.java | 6 +++ .../beam/sdk/io/iceberg/SchemaChange.java | 4 ++ .../sdk/io/iceberg/ColumnNameChecksTest.java | 50 +++++++++++++++++++ 3 files changed, 60 insertions(+) diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java index c1e6aaf8c5a9..d14995e5cdc9 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java @@ -87,6 +87,7 @@ private static void findInvalidNamesInType( } else if (type.isListType()) { findInvalidNamesInType(type.asListType().elementType(), rawPath + ".element", changes); } else if (type.isMapType()) { + findInvalidNamesInType(type.asMapType().keyType(), rawPath + ".key", changes); findInvalidNamesInType(type.asMapType().valueType(), rawPath + ".value", changes); } } @@ -138,6 +139,11 @@ private static void findCaseCollisionsInType( rawPath + ".element", changes); } else if (tableType.isMapType() && fileType.isMapType()) { + findCaseCollisionsInType( + tableType.asMapType().keyType(), + fileType.asMapType().keyType(), + rawPath + ".key", + changes); findCaseCollisionsInType( tableType.asMapType().valueType(), fileType.asMapType().valueType(), diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java index cbcff264edaa..5e1583e2c2d6 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java @@ -77,6 +77,10 @@ boolean allowedBy(SchemaEvolutionConfig config, Pins pins) { } String disallowedReason(Pins pins) { + if (kind == Kind.CONFLICT) { + // A conflict needs no option and its description stands alone. + return description; + } if (kind == Kind.FIELD_RELAXATION) { if (pins.isPinned(path)) { return description + " (pinned as required)"; diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java index 3f48359cde50..6892eaf98b5e 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java @@ -205,4 +205,54 @@ public void testFindCaseCollisionsPassesExactNewAndKindMismatchedNames() { optional(3, "s", Types.StringType.get())); assertEquals(Collections.emptyList(), caseCollisions(table, file)); } + + /** Map keys can be structs, and their field names are checked like any other level. */ + @Test + public void testInvalidNamesInsideStructMapKeysAreConflicts() { + Types.StructType file = + Types.StructType.of( + optional( + 1, + "m", + Types.MapType.ofOptional( + 2, + 3, + Types.StructType.of( + optional(4, "a.b", Types.StringType.get()), + optional(5, "", Types.StringType.get())), + Types.StringType.get()))); + List conflicts = invalidNames(file); + assertEquals(conflicts.toString(), 2, conflicts.size()); + assertTrue(conflicts.toString(), conflicts.get(0).contains("`a.b`")); + assertEquals("empty column name under m.key", conflicts.get(1)); + } + + @Test + public void testCaseCollisionInsideStructMapKeyIsConflict() { + Types.StructType table = + Types.StructType.of( + optional( + 1, + "m", + Types.MapType.ofOptional( + 2, + 3, + Types.StructType.of(optional(4, "k", Types.StringType.get())), + Types.StringType.get()))); + Types.StructType file = + Types.StructType.of( + optional( + 1, + "m", + Types.MapType.ofOptional( + 2, + 3, + Types.StructType.of(optional(4, "K", Types.StringType.get())), + Types.StringType.get()))); + List conflicts = caseCollisions(table, file); + assertEquals(conflicts.toString(), 1, conflicts.size()); + assertTrue( + conflicts.toString(), + conflicts.get(0).contains("m.key.K differs only in case from table column k")); + } } From f2edbbf0ea544bcebacebac8fba4df1864c3536f Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 10 Sep 2026 18:38:30 -0400 Subject: [PATCH 10/10] track test --- .../beam/sdk/io/iceberg/SchemaChangeTest.java | 77 +++++++++++++++++++ 1 file changed, 77 insertions(+) create mode 100644 sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaChangeTest.java diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaChangeTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaChangeTest.java new file mode 100644 index 000000000000..b9a6f0f93fd0 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaChangeTest.java @@ -0,0 +1,77 @@ +/* + * 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.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +import java.util.Arrays; +import java.util.Collections; +import org.apache.beam.sdk.io.iceberg.SchemaChange.Kind; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class SchemaChangeTest { + + private static final Pins NO_PINS = new Pins(Collections.emptyList()); + private static final Pins CITY_PINNED = new Pins(Arrays.asList("address.city")); + + @Test + public void testDisallowedReasonPerKind() { + // A conflict's description stands alone: no option unblocks it, so no "(needs ...)" suffix. + assertEquals( + "type changed on name", + new SchemaChange(Kind.CONFLICT, "", "type changed on name").disallowedReason(NO_PINS)); + assertEquals( + "add optional email string (needs ALLOW_FIELD_ADDITION)", + new SchemaChange(Kind.FIELD_ADDITION, "email", "add optional email string") + .disallowedReason(NO_PINS)); + assertEquals( + "relax name to optional (needs ALLOW_FIELD_RELAXATION)", + new SchemaChange(Kind.FIELD_RELAXATION, "name", "relax name to optional") + .disallowedReason(CITY_PINNED)); + assertEquals( + "relax address.city to optional (pinned as required)", + new SchemaChange(Kind.FIELD_RELAXATION, "address.city", "relax address.city to optional") + .disallowedReason(CITY_PINNED)); + assertEquals( + "relax address to optional (ancestor of pinned column address.city)", + new SchemaChange(Kind.FIELD_RELAXATION, "address", "relax address to optional") + .disallowedReason(CITY_PINNED)); + } + + @Test + public void testAllowedBy() { + SchemaEvolutionConfig all = SchemaEvolutionConfig.of(SchemaEvolutionOption.values()); + assertFalse(new SchemaChange(Kind.CONFLICT, "", "boom").allowedBy(all, NO_PINS)); + assertTrue(new SchemaChange(Kind.FIELD_ADDITION, "email", "add").allowedBy(all, NO_PINS)); + assertFalse( + new SchemaChange(Kind.FIELD_ADDITION, "email", "add") + .allowedBy(SchemaEvolutionConfig.disabled(), NO_PINS)); + assertFalse( + new SchemaChange(Kind.FIELD_RELAXATION, "address.city", "relax") + .allowedBy(all, CITY_PINNED)); + assertFalse( + new SchemaChange(Kind.FIELD_RELAXATION, "address", "relax").allowedBy(all, CITY_PINNED)); + assertTrue( + new SchemaChange(Kind.FIELD_RELAXATION, "name", "relax").allowedBy(all, CITY_PINNED)); + } +}