You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "claudevdm (via GitHub)" <gi...@apache.org> on 2026/09/08 19:35:04 UTC
[PR] AddFiles: SchemaDelta classifies what a file schema needs from the table [beam]
claudevdm opened a new pull request, #40062:
URL: https://github.com/apache/beam/pull/40062
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".
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: SchemaDelta classifies what a file schema needs from the table [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: SchemaDelta classifies what a file schema needs from the table [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm merged PR #40062:
URL: https://github.com/apache/beam/pull/40062
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: SchemaDelta classifies what a file schema needs from the table [beam]
Posted by "claudevdm (via GitHub)" <gi...@apache.org>.
claudevdm commented on code in PR #40062:
URL: https://github.com/apache/beam/pull/40062#discussion_r3983877045
##########
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:
Iceberg accepts a literal a.b column through unionByNameWith. But icebergs addColumn rejects dots as "ambiguous name". So upstream laready treats creating them as user error that unionByNameWith can bypass.
Empty names are rejected by iceberg at the top level. Empty nested names are silently added but I think that is not a valid use case and missing validation in iceberg?
Case collisions are legal in iceberg file format, but it breaks reader implementations. Users can work around this with a column alias.
So I think these are protecting the user against configuring something that might blow up without their knowledge.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: SchemaDelta classifies what a file schema needs from the table [beam]
Posted by "kennknowles (via GitHub)" <gi...@apache.org>.
kennknowles commented on PR #40062:
URL: https://github.com/apache/beam/pull/40062#issuecomment-5637635416
R: @chamikaramj
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: SchemaDelta classifies what a file schema needs from the table [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40062:
URL: https://github.com/apache/beam/pull/40062#issuecomment-5637650509
Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment `assign set of reviewers`
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: SchemaDelta classifies what a file schema needs from the table [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40062:
URL: https://github.com/apache/beam/pull/40062#issuecomment-5624834439
Assigning reviewers:
R: @kennknowles for label java.
Note: If you would like to opt out of this review, comment `assign to next reviewer`.
Available commands:
- `stop reviewer notifications` - opt out of the automated review tooling
- `remind me after tests pass` - tag the comment author after tests pass
- `waiting on author` - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)
The PR bot will only process comments in the main thread (not review comments).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AddFiles: SchemaDelta classifies what a file schema needs from the table [beam]
Posted by "chamikaramj (via GitHub)" <gi...@apache.org>.
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org