AddFiles: SchemaDelta classifies what a file schema needs from the table

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);
+    }
+  }
+}

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;
+  }
+}

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 + ")";
+  }
+}