chamikaramj commented on code in PR #40062: URL: https://github.com/apache/beam/pull/40062#discussion_r3991485002
########## 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: Not supporting these cases sounds good to me. Thanks! -- 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]
