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..d14995e5cdc9 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecks.java @@ -0,0 +1,154 @@ +/* + * 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.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 SchemaChange(SchemaChange.Kind.CONFLICT, rawPath, "empty column name" + at)); + } else if (field.name().contains(".")) { + changes.add( + new SchemaChange( + SchemaChange.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 SchemaChange( + SchemaChange.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().keyType(), rawPath + ".key", changes); + 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 SchemaChange( + SchemaChange.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().keyType(), + fileType.asMapType().keyType(), + rawPath + ".key", + changes); + 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/Pins.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/Pins.java new file mode 100644 index 000000000000..b516bfeefed2 --- /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.Collections; +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.dotted = new ArrayList<>(requiredColumns); + Collections.sort(dotted); + this.segments = new ArrayList<>(); + for (String column : dotted) { + segments.add(Arrays.asList(column.split("\\.", -1))); + } + } + + /** Whether {@code dottedPath} itself is pinned. */ + boolean isPinned(String dottedPath) { + return dotted.contains(dottedPath); + } + + /** + * 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 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)) { + return dotted.get(i); + } + } + return null; + } +} 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..5e1583e2c2d6 --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaChange.java @@ -0,0 +1,95 @@ +/* + * 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.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)"; + } + @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 new file mode 100644 index 000000000000..bacdac2e7cfe --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -0,0 +1,439 @@ +/* + * 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.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 { + + private final List changes; + + 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)) { + return new SchemaDelta(Collections.emptyList()); + } + + List nameConflicts = new ArrayList<>(); + ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "", nameConflicts); + ColumnNameChecks.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 (SchemaChange 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 (SchemaChange change : absent) { + absentByPath.put(change.path, change); + } + return diff(before, merged, absentByPath); + } + + /** + * 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 SchemaChange( + SchemaChange.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 (SchemaChange change : changes) { + if (change.absent) { + paths.add(change.path); + } + } + return paths; + } + + private static SchemaDelta conflict(String message) { + List changes = new ArrayList<>(); + changes.add(new SchemaChange(SchemaChange.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 SchemaChange( + SchemaChange.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 SchemaChange(SchemaChange.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 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 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 SchemaChange( + SchemaChange.Kind.CONFLICT, + rawPath, + "default changed on " + path + " (not supported)")); + } + if (oldField.isRequired() && newField.isOptional()) { + @Nullable SchemaChange absent = absentRemaining.remove(rawPath); + changes.add( + absent != null + ? absent + : new SchemaChange( + SchemaChange.Kind.FIELD_RELAXATION, rawPath, "relax " + path + " to optional")); + } else if (oldField.isOptional() && newField.isRequired()) { + changes.add( + new SchemaChange( + SchemaChange.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 SchemaChange( + SchemaChange.Kind.TYPE_PROMOTION, + rawPath, + "promote " + path + " " + oldField.type() + " to " + newField.type())); + } else { + changes.add( + new SchemaChange( + SchemaChange.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 SchemaChange( + SchemaChange.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; + } + + 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(SchemaChange.Kind.class); + for (SchemaChange change : changes) { + kinds.add(change.kind); + } + return kinds; + } + + List descriptions() { + List descriptions = new ArrayList<>(); + for (SchemaChange change : changes) { + descriptions.add(change.description); + } + return Collections.unmodifiableList(descriptions); + } + + @Nullable String conflict() { + for (SchemaChange change : changes) { + if (change.kind == SchemaChange.Kind.CONFLICT) { + return change.description; + } + } + return null; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + Pins pins = new Pins(config.getRequiredColumns()); + for (SchemaChange change : changes) { + if (!change.allowedBy(config, pins)) { + return false; + } + } + return true; + } + + /** Why {@link #allowedBy} is false; empty when it is true. */ + String disallowedReason(SchemaEvolutionConfig config) { + List conflicts = new ArrayList<>(); + for (SchemaChange change : changes) { + if (change.kind == SchemaChange.Kind.CONFLICT) { + conflicts.add(change.description); + } + } + 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 (SchemaChange change : changes) { + if (!change.allowedBy(config, pins)) { + disallowed.add(change.disallowedReason(pins)); + } + } + 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/ColumnNameChecksTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java new file mode 100644 index 000000000000..6892eaf98b5e --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ColumnNameChecksTest.java @@ -0,0 +1,258 @@ +/* + * 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.SchemaChange.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 (SchemaChange 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)); + } + + /** 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")); + } +} 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)); + } +} 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..d3d4172f0f95 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java @@ -0,0 +1,582 @@ +/* + * 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.SchemaChange.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"))); + // 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 + + @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: one classify wiring test per check; exhaustive + // shapes are covered directly on the walks below + + @Test + public void testDottedColumnNameIsConflict() { + Schema file = new Schema(optional(1, "address.zip", Types.IntegerType.get())); + SchemaDelta delta = classify(file); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + String reason = delta.disallowedReason(ALL); + assertTrue(reason, reason.contains("path separator")); + } + + /** The union rejects an empty name only at the top level; nested ones would be added. */ + @Test + public void testEmptyColumnNameIsConflict() { + Schema file = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "", Types.StringType.get())); + SchemaDelta delta = classify(file); + assertEquals(delta.toString(), EnumSet.of(Kind.CONFLICT), delta.kinds()); + String reason = delta.disallowedReason(ALL); + assertTrue(reason, reason.contains("empty column name")); + } + + @Test + public void testCaseOnlyDifferenceFromTableIsConflict() { + Schema file = + new Schema( + optional(1, "NAME", Types.StringType.get()), required(2, "id", Types.LongType.get())); + SchemaDelta delta = classify(file); + 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")); + } + + /** Two new columns differing only in case would break the lower-case index between them. */ + @Test + public void testFileInternalCaseCollisionIsConflict() { + Schema file = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "email", Types.StringType.get()), + optional(3, "EMAIL", Types.StringType.get())); + SchemaDelta delta = classify(file); + 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")); + } + + // ---- 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()); + + 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. */ + @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()); + } +}