chamikaramj commented on code in PR #40062: URL: https://github.com/apache/beam/pull/40062#discussion_r3982033337
########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { Review Comment: Seems a bit brittel to depend on the structure of the string here ? Should we move it to a class ? ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { + if (pin.startsWith(path + ".")) { + return pin; + } + } + return null; + } + + String disallowedReason(SchemaEvolutionConfig config) { + @Nullable String pin = kind == Kind.FIELD_RELAXATION ? forbiddingPin(config) : null; + if (pin != null) { + if (pin.equals(path)) { + return description + " (pinned as required)"; + } + return description + " (ancestor of pinned column " + pin + ")"; + } + return description + " (needs " + kind.option + ")"; + } + } + + private final List<Change> changes; + + private SchemaDelta(List<Change> changes) { + this.changes = Collections.unmodifiableList(changes); + } + + static SchemaDelta classify(Table table, Schema fileSchema) { Review Comment: Probably add a small docstring here. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { Review Comment: config.getRequiredColumns() returns a set that has a non deterministic order. If this returns multiple required columns different calls could return different pins here introducing flaky behavior for tests and potentially production failures. Consider making this deterministic (for example, order lexicographically). ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { + if (pin.startsWith(path + ".")) { + return pin; + } + } + return null; + } + + String disallowedReason(SchemaEvolutionConfig config) { + @Nullable String pin = kind == Kind.FIELD_RELAXATION ? forbiddingPin(config) : null; + if (pin != null) { + if (pin.equals(path)) { + return description + " (pinned as required)"; + } + return description + " (ancestor of pinned column " + pin + ")"; + } + return description + " (needs " + kind.option + ")"; + } + } + + private final List<Change> changes; + + private SchemaDelta(List<Change> changes) { + this.changes = Collections.unmodifiableList(changes); + } + + static SchemaDelta classify(Table table, Schema fileSchema) { + Schema before = table.schema(); + if (before.sameSchema(fileSchema)) { + return new SchemaDelta(Collections.emptyList()); + } + + List<Change> nameConflicts = new ArrayList<>(); + findInvalidNames(fileSchema.asStruct(), "", nameConflicts); + findCaseCollisions(before.asStruct(), fileSchema.asStruct(), "", nameConflicts); + if (!nameConflicts.isEmpty()) { + return new SchemaDelta(nameConflicts); + } + + List<Change> absent = new ArrayList<>(); + findAbsentRequired(before.asStruct(), fileSchema.asStruct(), "", absent); + Schema merged; + try { + // The absent-path relaxations are applied here too, so anything Iceberg refuses (an + // identifier field, say) is classified as this file's conflict instead of surfacing + // mid-transaction under a cross-schema message. + UpdateSchema update = table.updateSchema().unionByNameWith(fileSchema); + for (Change change : absent) { + update = update.makeColumnOptional(change.path); + } + merged = update.apply(); + } catch (ValidationException | IllegalArgumentException e) { + // SchemaUpdate reports type conflicts through both exception types + return conflict(e.getClass().getSimpleName() + ": " + AddFiles.errorMessage(e)); + } + Map<String, Change> absentByPath = new HashMap<>(); + for (Change change : absent) { + absentByPath.put(change.path, change); + } + return diff(before, merged, absentByPath); + } + + /** + * File column names no table can absorb, checked at every level including structs the table does + * not have yet. A literal dot is a conflict because Iceberg's name APIs, pins, aliases and + * ignores all treat the dot as a path separator, and a colliding struct in a later window would + * make the whole table unresolvable by name; rejected whether or not it collides today. An empty + * name would otherwise be added as a real column (the union only rejects it at the top level). + * Two file columns at one level differing only in case would be added as two columns, after which + * Iceberg cannot build the lower-case name index. + */ + private static void findInvalidNames( + Types.StructType struct, String prefix, List<Change> 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 Change(Kind.CONFLICT, rawPath, "empty column name" + at)); + } else if (field.name().contains(".")) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "column name " + + quoteIfDotted(field.name()) + + " contains '.', which Iceberg treats as a path separator; rename the column" + + " at its source")); + } + @Nullable String seen = + seenByLowerCase.put(field.name().toLowerCase(Locale.ROOT), field.name()); + if (seen != null) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "columns " + + prefix + + quoteIfDotted(seen) + + " and " + + prefix + + quoteIfDotted(field.name()) + + " differ only in case; rename one or map it with a column alias")); + } + findInvalidNamesInType(field.type(), rawPath, changes); + } + } + + private static void findInvalidNamesInType(Type type, String rawPath, List<Change> changes) { + if (type.isStructType()) { + findInvalidNames(type.asStructType(), rawPath + ".", changes); + } else if (type.isListType()) { + findInvalidNamesInType(type.asListType().elementType(), rawPath + ".element", changes); + } else if (type.isMapType()) { + findInvalidNamesInType(type.asMapType().valueType(), rawPath + ".value", changes); + } + } + + /** + * A file column whose name matches a table column at the same level only case-insensitively would + * be added as a separate column, after which Iceberg cannot build the lower-case name index and + * every case-insensitive reader of the table fails. + */ + private static void findCaseCollisions( + Types.StructType tableStruct, + Types.StructType fileStruct, + String prefix, + List<Change> changes) { + for (Types.NestedField fileField : fileStruct.fields()) { + String rawPath = prefix + fileField.name(); + Types.NestedField exact = tableStruct.field(fileField.name()); + if (exact == null) { + for (Types.NestedField tableField : tableStruct.fields()) { + if (tableField.name().equalsIgnoreCase(fileField.name())) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "column " + + prefix + + quoteIfDotted(fileField.name()) + + " differs only in case from table column " + + quoteIfDotted(tableField.name()) + + "; rename it or map it with a column alias")); + break; + } + } + continue; + } + findCaseCollisionsInType(exact.type(), fileField.type(), rawPath, changes); + } + } + + private static void findCaseCollisionsInType( Review Comment: BTW there are many util methods here that are private and hence untestable via separate unit tests. Consider making these package private and adding unit tests. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { + if (pin.startsWith(path + ".")) { + return pin; + } + } + return null; + } + + String disallowedReason(SchemaEvolutionConfig config) { + @Nullable String pin = kind == Kind.FIELD_RELAXATION ? forbiddingPin(config) : null; + if (pin != null) { + if (pin.equals(path)) { + return description + " (pinned as required)"; + } + return description + " (ancestor of pinned column " + pin + ")"; + } + return description + " (needs " + kind.option + ")"; + } + } + + private final List<Change> changes; + + private SchemaDelta(List<Change> changes) { + this.changes = Collections.unmodifiableList(changes); + } + + static SchemaDelta classify(Table table, Schema fileSchema) { + Schema before = table.schema(); + if (before.sameSchema(fileSchema)) { + return new SchemaDelta(Collections.emptyList()); + } + + List<Change> nameConflicts = new ArrayList<>(); + findInvalidNames(fileSchema.asStruct(), "", nameConflicts); + findCaseCollisions(before.asStruct(), fileSchema.asStruct(), "", nameConflicts); + if (!nameConflicts.isEmpty()) { + return new SchemaDelta(nameConflicts); + } + + List<Change> absent = new ArrayList<>(); + findAbsentRequired(before.asStruct(), fileSchema.asStruct(), "", absent); + Schema merged; + try { + // The absent-path relaxations are applied here too, so anything Iceberg refuses (an + // identifier field, say) is classified as this file's conflict instead of surfacing + // mid-transaction under a cross-schema message. + UpdateSchema update = table.updateSchema().unionByNameWith(fileSchema); + for (Change change : absent) { + update = update.makeColumnOptional(change.path); + } + merged = update.apply(); + } catch (ValidationException | IllegalArgumentException e) { + // SchemaUpdate reports type conflicts through both exception types + return conflict(e.getClass().getSimpleName() + ": " + AddFiles.errorMessage(e)); + } + Map<String, Change> absentByPath = new HashMap<>(); + for (Change change : absent) { + absentByPath.put(change.path, change); + } + return diff(before, merged, absentByPath); + } + + /** + * File column names no table can absorb, checked at every level including structs the table does + * not have yet. A literal dot is a conflict because Iceberg's name APIs, pins, aliases and + * ignores all treat the dot as a path separator, and a colliding struct in a later window would + * make the whole table unresolvable by name; rejected whether or not it collides today. An empty + * name would otherwise be added as a real column (the union only rejects it at the top level). + * Two file columns at one level differing only in case would be added as two columns, after which + * Iceberg cannot build the lower-case name index. + */ + private static void findInvalidNames( + Types.StructType struct, String prefix, List<Change> 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 Change(Kind.CONFLICT, rawPath, "empty column name" + at)); + } else if (field.name().contains(".")) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "column name " + + quoteIfDotted(field.name()) + + " contains '.', which Iceberg treats as a path separator; rename the column" + + " at its source")); + } + @Nullable String seen = + seenByLowerCase.put(field.name().toLowerCase(Locale.ROOT), field.name()); + if (seen != null) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "columns " + + prefix + + quoteIfDotted(seen) + + " and " + + prefix + + quoteIfDotted(field.name()) + + " differ only in case; rename one or map it with a column alias")); + } + findInvalidNamesInType(field.type(), rawPath, changes); + } + } + + private static void findInvalidNamesInType(Type type, String rawPath, List<Change> changes) { + if (type.isStructType()) { + findInvalidNames(type.asStructType(), rawPath + ".", changes); + } else if (type.isListType()) { + findInvalidNamesInType(type.asListType().elementType(), rawPath + ".element", changes); + } else if (type.isMapType()) { + findInvalidNamesInType(type.asMapType().valueType(), rawPath + ".value", changes); Review Comment: Check map keys as well ? ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A Review Comment: Ideally we do not return different things from the same method. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { + if (pin.startsWith(path + ".")) { + return pin; + } + } + return null; + } + + String disallowedReason(SchemaEvolutionConfig config) { + @Nullable String pin = kind == Kind.FIELD_RELAXATION ? forbiddingPin(config) : null; + if (pin != null) { + if (pin.equals(path)) { + return description + " (pinned as required)"; + } + return description + " (ancestor of pinned column " + pin + ")"; + } + return description + " (needs " + kind.option + ")"; + } + } + + private final List<Change> changes; + + private SchemaDelta(List<Change> changes) { + this.changes = Collections.unmodifiableList(changes); + } + + static SchemaDelta classify(Table table, Schema fileSchema) { + Schema before = table.schema(); + if (before.sameSchema(fileSchema)) { + return new SchemaDelta(Collections.emptyList()); + } + + List<Change> nameConflicts = new ArrayList<>(); + findInvalidNames(fileSchema.asStruct(), "", nameConflicts); + findCaseCollisions(before.asStruct(), fileSchema.asStruct(), "", nameConflicts); + if (!nameConflicts.isEmpty()) { + return new SchemaDelta(nameConflicts); + } + + List<Change> absent = new ArrayList<>(); + findAbsentRequired(before.asStruct(), fileSchema.asStruct(), "", absent); + Schema merged; + try { + // The absent-path relaxations are applied here too, so anything Iceberg refuses (an + // identifier field, say) is classified as this file's conflict instead of surfacing + // mid-transaction under a cross-schema message. + UpdateSchema update = table.updateSchema().unionByNameWith(fileSchema); + for (Change change : absent) { + update = update.makeColumnOptional(change.path); + } + merged = update.apply(); + } catch (ValidationException | IllegalArgumentException e) { + // SchemaUpdate reports type conflicts through both exception types + return conflict(e.getClass().getSimpleName() + ": " + AddFiles.errorMessage(e)); + } + Map<String, Change> absentByPath = new HashMap<>(); + for (Change change : absent) { + absentByPath.put(change.path, change); + } + return diff(before, merged, absentByPath); + } + + /** + * File column names no table can absorb, checked at every level including structs the table does Review Comment: To clarify are these conflicts for Iceberg itself or just for out solution. I think if we make column names more strict than Iceberg, some customers might ran into issues (we ran into similar issues before with BigQuery). ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed Review Comment: Should this be in it's own file ? This class seems to be very long so it's better if we can move some of the inner classes (and may be some util methods) to separate files. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { Review Comment: Could you make the class name more descriptive and move to it's own file please. Also add a docstring. ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { + if (pin.startsWith(path + ".")) { + return pin; + } + } + return null; + } + + String disallowedReason(SchemaEvolutionConfig config) { + @Nullable String pin = kind == Kind.FIELD_RELAXATION ? forbiddingPin(config) : null; + if (pin != null) { + if (pin.equals(path)) { + return description + " (pinned as required)"; + } + return description + " (ancestor of pinned column " + pin + ")"; + } + return description + " (needs " + kind.option + ")"; + } + } + + private final List<Change> changes; + + private SchemaDelta(List<Change> changes) { + this.changes = Collections.unmodifiableList(changes); + } + + static SchemaDelta classify(Table table, Schema fileSchema) { + Schema before = table.schema(); + if (before.sameSchema(fileSchema)) { + return new SchemaDelta(Collections.emptyList()); + } + + List<Change> nameConflicts = new ArrayList<>(); + findInvalidNames(fileSchema.asStruct(), "", nameConflicts); + findCaseCollisions(before.asStruct(), fileSchema.asStruct(), "", nameConflicts); + if (!nameConflicts.isEmpty()) { + return new SchemaDelta(nameConflicts); + } + + List<Change> absent = new ArrayList<>(); + findAbsentRequired(before.asStruct(), fileSchema.asStruct(), "", absent); + Schema merged; + try { + // The absent-path relaxations are applied here too, so anything Iceberg refuses (an + // identifier field, say) is classified as this file's conflict instead of surfacing + // mid-transaction under a cross-schema message. + UpdateSchema update = table.updateSchema().unionByNameWith(fileSchema); + for (Change change : absent) { + update = update.makeColumnOptional(change.path); + } + merged = update.apply(); + } catch (ValidationException | IllegalArgumentException e) { + // SchemaUpdate reports type conflicts through both exception types + return conflict(e.getClass().getSimpleName() + ": " + AddFiles.errorMessage(e)); + } + Map<String, Change> absentByPath = new HashMap<>(); + for (Change change : absent) { + absentByPath.put(change.path, change); + } + return diff(before, merged, absentByPath); + } + + /** + * File column names no table can absorb, checked at every level including structs the table does + * not have yet. A literal dot is a conflict because Iceberg's name APIs, pins, aliases and + * ignores all treat the dot as a path separator, and a colliding struct in a later window would + * make the whole table unresolvable by name; rejected whether or not it collides today. An empty + * name would otherwise be added as a real column (the union only rejects it at the top level). + * Two file columns at one level differing only in case would be added as two columns, after which + * Iceberg cannot build the lower-case name index. + */ + private static void findInvalidNames( + Types.StructType struct, String prefix, List<Change> 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 Change(Kind.CONFLICT, rawPath, "empty column name" + at)); + } else if (field.name().contains(".")) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "column name " + + quoteIfDotted(field.name()) + + " contains '.', which Iceberg treats as a path separator; rename the column" + + " at its source")); + } + @Nullable String seen = + seenByLowerCase.put(field.name().toLowerCase(Locale.ROOT), field.name()); + if (seen != null) { + changes.add( + new Change( + Kind.CONFLICT, + rawPath, + "columns " + + prefix + + quoteIfDotted(seen) + + " and " + + prefix + + quoteIfDotted(field.name()) + + " differ only in case; rename one or map it with a column alias")); + } + findInvalidNamesInType(field.type(), rawPath, changes); + } + } + + private static void findInvalidNamesInType(Type type, String rawPath, List<Change> changes) { + if (type.isStructType()) { + findInvalidNames(type.asStructType(), rawPath + ".", changes); + } else if (type.isListType()) { + findInvalidNamesInType(type.asListType().elementType(), rawPath + ".element", changes); + } else if (type.isMapType()) { + findInvalidNamesInType(type.asMapType().valueType(), rawPath + ".value", changes); + } + } + + /** + * A file column whose name matches a table column at the same level only case-insensitively would Review Comment: Seems like we are documenting the reason for adding this method here not what this method does ? ########## sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java: ########## @@ -0,0 +1,610 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.EnumSet; +import java.util.HashMap; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.UpdateSchema; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.types.Type; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * What {@code unionByNameWith(fileSchema)} would change on a table, without changing it. Computed + * by diffing the union result against the table schema by field id: existing fields keep their ids + * and additions get fresh ones, so the diff is exact and independent of column order. + * + * <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 { + + enum Kind { + FIELD_ADDITION(SchemaEvolutionOption.ALLOW_FIELD_ADDITION), + FIELD_RELAXATION(SchemaEvolutionOption.ALLOW_FIELD_RELAXATION), + TYPE_PROMOTION(SchemaEvolutionOption.ALLOW_TYPE_PROMOTION), + /** The union is impossible (for example string vs int); never allowed. */ + CONFLICT(null); + + final @Nullable SchemaEvolutionOption option; + + Kind(@Nullable SchemaEvolutionOption option) { + this.option = option; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + return option != null && config.allows(option); + } + } + + private static final class Change { + final Kind kind; + + /** Unquoted column path for the config lookup; empty for conflicts without a field. */ + final String path; + + final String description; + + /** A relaxation because the column is absent from the file, not declared optional. */ + final boolean absent; + + Change(Kind kind, String path, String description) { + this(kind, path, description, false); + } + + Change(Kind kind, String path, String description, boolean absent) { + this.kind = kind; + this.path = path; + this.description = description; + this.absent = absent; + } + + boolean allowedBy(SchemaEvolutionConfig config) { + if (kind == Kind.FIELD_RELAXATION && forbiddingPin(config) != null) { + return false; + } + return kind.allowedBy(config); + } + + /** + * The pin that forbids relaxing this path: the path itself, or a pinned column beneath it. A + * null ancestor nulls the pinned leaf, so relaxing the ancestor only manufactures files that + * fail the pin check at registration. + */ + private @Nullable String forbiddingPin(SchemaEvolutionConfig config) { + if (config.isPinned(path)) { + return path; + } + for (String pin : config.getRequiredColumns()) { + if (pin.startsWith(path + ".")) { + return pin; + } + } + return null; + } + + String disallowedReason(SchemaEvolutionConfig config) { + @Nullable String pin = kind == Kind.FIELD_RELAXATION ? forbiddingPin(config) : null; + if (pin != null) { + if (pin.equals(path)) { + return description + " (pinned as required)"; + } + return description + " (ancestor of pinned column " + pin + ")"; + } + return description + " (needs " + kind.option + ")"; Review Comment: Kind.CONFLICT, kind.option is null, so adjust the reason for that ? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
