This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 8b3370c310d AddFiles: SchemaDelta classifies what a file schema needs
from the table (#40062)
8b3370c310d is described below
commit 8b3370c310d55f65488e7ec1983f3e0b4c5a5751
Author: claudevdm <[email protected]>
AuthorDate: Fri Sep 11 12:58:00 2026 -0400
AddFiles: SchemaDelta classifies what a file schema needs from the table
(#40062)
* 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".
* move pins to class
* pins
* docstring
* Fix docstring
* tests
* split file
* split
* map keys
* track test
---
.../beam/sdk/io/iceberg/ColumnNameChecks.java | 154 ++++++
.../java/org/apache/beam/sdk/io/iceberg/Pins.java | 60 +++
.../apache/beam/sdk/io/iceberg/SchemaChange.java | 95 ++++
.../apache/beam/sdk/io/iceberg/SchemaDelta.java | 439 ++++++++++++++++
.../beam/sdk/io/iceberg/ColumnNameChecksTest.java | 258 +++++++++
.../beam/sdk/io/iceberg/SchemaChangeTest.java | 77 +++
.../beam/sdk/io/iceberg/SchemaDeltaTest.java | 582 +++++++++++++++++++++
7 files changed, 1665 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
new file mode 100644
index 00000000000..d14995e5cdc
--- /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<SchemaChange> changes) {
+ Map<String, String> 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<SchemaChange> 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<SchemaChange> 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<SchemaChange>
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 00000000000..b516bfeefed
--- /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<List<String>> segments;
+ private final List<String> dotted;
+
+ Pins(Collection<String> 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<String> path = Arrays.asList(dottedPath.split("\\.", -1));
+ for (int i = 0; i < segments.size(); i++) {
+ List<String> 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 00000000000..5e1583e2c2d
--- /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 00000000000..bacdac2e7cf
--- /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.
+ *
+ * <p>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<SchemaChange> changes;
+
+ private SchemaDelta(List<SchemaChange> 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<SchemaChange> nameConflicts = new ArrayList<>();
+ ColumnNameChecks.findInvalidNames(fileSchema.asStruct(), "",
nameConflicts);
+ ColumnNameChecks.findCaseCollisions(
+ before.asStruct(), fileSchema.asStruct(), "", nameConflicts);
+ if (!nameConflicts.isEmpty()) {
+ return new SchemaDelta(nameConflicts);
+ }
+
+ List<SchemaChange> 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<String, SchemaChange> 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<SchemaChange> 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<SchemaChange>
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<String> absentRequiredPaths() {
+ List<String> paths = new ArrayList<>();
+ for (SchemaChange change : changes) {
+ if (change.absent) {
+ paths.add(change.path);
+ }
+ }
+ return paths;
+ }
+
+ private static SchemaDelta conflict(String message) {
+ List<SchemaChange> 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<String, SchemaChange> absentByPath) {
+ Map<String, SchemaChange> absentRemaining = new HashMap<>(absentByPath);
+ Map<Integer, Types.NestedField> beforeById =
TypeUtil.indexById(before.asStruct());
+ Map<Integer, Types.NestedField> afterById =
TypeUtil.indexById(after.asStruct());
+ Map<Integer, Integer> parentById = TypeUtil.indexParents(after.asStruct());
+ Map<Integer, String> rawPathById =
TypeUtil.indexNameById(after.asStruct());
+ Map<Integer, String> pathById =
+ TypeUtil.indexQuotedNameById(after.asStruct(),
SchemaDelta::quoteIfDotted);
+
+ List<Integer> idsByPath = new ArrayList<>(afterById.keySet());
+ idsByPath.sort(
+ (a, b) ->
+
checkStateNotNull(rawPathById.get(a)).compareTo(checkStateNotNull(rawPathById.get(b))));
+
+ List<SchemaChange> 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<Integer, String> beforePathById =
+ TypeUtil.indexQuotedNameById(before.asStruct(),
SchemaDelta::quoteIfDotted);
+ List<String> 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<String, SchemaChange> absentRemaining,
+ List<SchemaChange> 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<Types.NestedField> 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<Integer, Integer> parentById, Map<Integer,
Types.NestedField> 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<SchemaChange.Kind> kinds() {
+ Set<SchemaChange.Kind> kinds = EnumSet.noneOf(SchemaChange.Kind.class);
+ for (SchemaChange change : changes) {
+ kinds.add(change.kind);
+ }
+ return kinds;
+ }
+
+ List<String> descriptions() {
+ List<String> 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<String> 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<String> 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 00000000000..6892eaf98b5
--- /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<String> invalidNames(Types.StructType fileStruct) {
+ List<SchemaChange> changes = new ArrayList<>();
+ ColumnNameChecks.findInvalidNames(fileStruct, "", changes);
+ return conflictDescriptions(changes);
+ }
+
+ private static List<String> caseCollisions(
+ Types.StructType tableStruct, Types.StructType fileStruct) {
+ List<SchemaChange> changes = new ArrayList<>();
+ ColumnNameChecks.findCaseCollisions(tableStruct, fileStruct, "", changes);
+ return conflictDescriptions(changes);
+ }
+
+ private static List<String> conflictDescriptions(List<SchemaChange> changes)
{
+ List<String> 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<String> 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<String> 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<String> 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<String> 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 00000000000..b9a6f0f93fd
--- /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 00000000000..d3d4172f0f9
--- /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<lat: optional double, lon:
optional double>"),
+ 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<x: optional int> 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<string, int>", "add optional
tags list<string>"),
+ 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<c: optional int>", "add optional
`a.b` string"),
+ delta.descriptions());
+ }
+}
