diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json b/.github/trigger_files/IO_Iceberg_Integration_Tests.json index b73af5e61a43..7ab7bcd9a9c6 100644 --- a/.github/trigger_files/IO_Iceberg_Integration_Tests.json +++ b/.github/trigger_files/IO_Iceberg_Integration_Tests.json @@ -1,4 +1,4 @@ { "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 1 + "modification": 2 } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java index c72479aa39b3..00cf77af1dca 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/AddFiles.java @@ -211,7 +211,8 @@ public AddFiles( /** * @param locationPrefix when set on a partitioned table, the partition is read from the path - * after this prefix instead of from the file's column stats + * after this prefix instead of from the file's column stats. A table partitioned by {@code + * dt} then accepts files under {@code dt=2024-01-15/} that have no {@code dt} column * @param partitionFields partition spec applied when the table is created by this transform * @param sortFields sort order applied when the table is created by this transform * @param tableProps table properties applied when the table is created by this transform @@ -343,7 +344,7 @@ public PCollectionRowTuple expand(PCollection input) { private PCollection> distinctSchemas( PCollection windowed) { return windowed - .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema(evolution))) + .apply("ReadFooterSchema", ParDo.of(new ReadFooterSchema(evolution, locationPrefix))) .setCoder(CollectDistinctSchemas.groupCoder()) .apply( "CollectDistinctSchemas", @@ -811,11 +812,26 @@ private Verdict verify(String filePath, FileFormat format, @Nullable ParquetMeta } catch (Exception e) { return Verdict.error(UNREADABLE_SCHEMA_ERROR + errorMessage(e)); } - @Nullable String uncovered = uncoveredReason(fileSchema); + Set identityPartitionColumnsFromPath = identityPartitionColumnsFromPath(filePath); + @Nullable String uncovered = uncoveredReason(fileSchema, identityPartitionColumnsFromPath); if (uncovered != null) { return Verdict.error(uncovered); } - return checkPins(filePath, fileSchema, parquetFooter); + return checkPins(filePath, fileSchema, parquetFooter, identityPartitionColumnsFromPath); + } + + /** + * Identity partition columns this file may lack because its path gives them a value: {@code dt} + * for a file under {@code dt=2024-01-15/} when the table is partitioned by {@code dt}. Empty + * without a prefix. + */ + private Set identityPartitionColumnsFromPath(String filePath) { + Table table = checkStateNotNull(this.table); + if (Strings.isNullOrEmpty(prefix) || table.spec().isUnpartitioned()) { + return Collections.emptySet(); + } + return PartitionPaths.identityColumnsFilledByKeys( + table.spec(), PartitionPaths.partitionKeysInPath(filePath, prefix)); } /** @@ -823,9 +839,10 @@ private Verdict verify(String filePath, FileFormat format, @Nullable ParquetMeta * covers every file. If not, refresh once per bundle (a commit may have landed since the table * was cached) and report the remaining delta. Never changes the schema. */ - private @Nullable String uncoveredReason(org.apache.iceberg.Schema fileSchema) { + private @Nullable String uncoveredReason( + org.apache.iceberg.Schema fileSchema, Set identityPartitionColumnsFromPath) { Table table = checkStateNotNull(this.table); - SchemaDelta delta = SchemaDelta.classify(table, fileSchema); + SchemaDelta delta = SchemaDelta.classify(table, fileSchema, identityPartitionColumnsFromPath); if (delta.isEmpty()) { return null; } @@ -836,7 +853,7 @@ private Verdict verify(String filePath, FileFormat format, @Nullable ParquetMeta table.refresh(); } } - delta = SchemaDelta.classify(table, fileSchema); + delta = SchemaDelta.classify(table, fileSchema, identityPartitionColumnsFromPath); if (delta.isEmpty()) { return null; } @@ -857,14 +874,23 @@ private Verdict verify(String filePath, FileFormat format, @Nullable ParquetMeta * goes on, so a pin the footer does count nulls for still fails the file. */ private Verdict checkPins( - String filePath, org.apache.iceberg.Schema fileSchema, ParquetMetadata footer) { + String filePath, + org.apache.iceberg.Schema fileSchema, + ParquetMetadata footer, + Set identityPartitionColumnsFromPath) { Table table = checkStateNotNull(this.table); List unproven = new ArrayList<>(); for (String pinned : evolution.getRequiredColumns()) { - if (table.schema().findField(pinned) == null) { + org.apache.iceberg.types.Types.@Nullable NestedField tableField = + table.schema().findField(pinned); + if (tableField == null) { continue; } if (fileSchema.findField(pinned) == null) { + if (identityPartitionColumnsFromPath.contains(tableField.fieldId())) { + // the path gives it a value, e.g. dt=2024-01-15, so it is never null + continue; + } return Verdict.error(PINNED_COLUMN_ERROR + pinned + " is absent from the file"); } @Nullable Long nulls = FileSchemas.nullCount(footer, fileSchema, pinned); @@ -969,10 +995,7 @@ private String getPartitionFromFilePath(String filePath) { if (checkStateNotNull(table).spec().isUnpartitioned()) { return ""; } - String partitionPath = filePath.substring(checkStateNotNull(prefix).length()); - int lastSlashIndex = partitionPath.lastIndexOf('/'); - - return lastSlashIndex > 0 ? partitionPath.substring(0, lastSlashIndex) : ""; + return PartitionPaths.relative(filePath, checkStateNotNull(prefix)); } /** diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java index 5c36967cfeba..8e11bd1ed3ec 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemas.java @@ -22,6 +22,7 @@ import java.io.InputStream; import java.io.OutputStream; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Map; import java.util.Objects; @@ -47,9 +48,10 @@ /** * One output entry per distinct schema: its file count and the columns EVERY file carrying it * proved free of nulls; one file with a null in "name" forces "name" to relax, however many clean - * files sit next to it. Entries come out most common first (ties broken by the JSON text) because - * the commit side applies schemas in that order and the most common schema should win a conflict. - * Schemas are compared as strings, so inputs must already be canonical. + * files sit next to it. Partition keys in path are kept the same way: {@code dt} stays only if + * every file sits under a {@code dt=...} folder. Entries come out most common first (ties broken by + * the JSON text) because the commit side applies schemas in that order and the most common schema + * should win a conflict. Schemas are compared as strings, so inputs must already be canonical. */ class CollectDistinctSchemas extends Combine.CombineFn< @@ -61,10 +63,12 @@ class CollectDistinctSchemas static final class Group { long files; TreeSet nullFreeColumns; + TreeSet partitionKeysInPath; - Group(long files, TreeSet nullFreeColumns) { + Group(long files, TreeSet nullFreeColumns, TreeSet partitionKeysInPath) { this.files = files; this.nullFreeColumns = nullFreeColumns; + this.partitionKeysInPath = partitionKeysInPath; } @Override @@ -73,12 +77,14 @@ public boolean equals(@Nullable Object other) { return false; } Group that = (Group) other; - return files == that.files && nullFreeColumns.equals(that.nullFreeColumns); + return files == that.files + && nullFreeColumns.equals(that.nullFreeColumns) + && partitionKeysInPath.equals(that.partitionKeysInPath); } @Override public int hashCode() { - return Objects.hash(files, nullFreeColumns); + return Objects.hash(files, nullFreeColumns, partitionKeysInPath); } } @@ -92,7 +98,20 @@ abstract static class SchemaGroup { private static @MonotonicNonNull SchemaCoder coder; static SchemaGroup of(String schemaJson, long files, List nullFreeColumns) { - return new AutoValue_CollectDistinctSchemas_SchemaGroup(schemaJson, files, nullFreeColumns); + return of(schemaJson, files, nullFreeColumns, Collections.emptyList()); + } + + static SchemaGroup of( + String schemaJson, + long files, + List nullFreeColumns, + List partitionKeysInPath) { + return new AutoValue_CollectDistinctSchemas_SchemaGroup( + schemaJson, files, nullFreeColumns, partitionKeysInPath); + } + + SchemaGroup withPartitionKeysInPath(List partitionKeysInPath) { + return of(getSchemaJson(), getFiles(), getNullFreeColumns(), partitionKeysInPath); } static SchemaCoder getCoder() { @@ -115,11 +134,21 @@ static SchemaCoder getCoder() { @SchemaFieldNumber("2") abstract List getNullFreeColumns(); + /** + * Folder keys every file of the group has a value for, e.g. {@code [dt]} when every file sits + * under a {@code dt=...} folder. A table partitioned by {@code dt} then needs no {@code dt} + * column in these files. + */ + @SchemaFieldNumber("3") + abstract List getPartitionKeysInPath(); + @Override public final String toString() { return getFiles() + " file(s), null-free in " + getNullFreeColumns() + + ", partition keys in path " + + getPartitionKeysInPath() + ", schema " + getSchemaJson(); } @@ -132,7 +161,12 @@ public Map createAccumulator() { @Override public Map addInput(Map accumulator, SchemaGroup file) { - add(accumulator, file.getSchemaJson(), file.getFiles(), file.getNullFreeColumns()); + add( + accumulator, + file.getSchemaJson(), + file.getFiles(), + file.getNullFreeColumns(), + file.getPartitionKeysInPath()); return accumulator; } @@ -141,7 +175,8 @@ public Map mergeAccumulators(Iterable> accumul Map merged = createAccumulator(); for (Map accumulator : accumulators) { for (Map.Entry entry : accumulator.entrySet()) { - add(merged, entry.getKey(), entry.getValue().files, entry.getValue().nullFreeColumns); + Group group = entry.getValue(); + add(merged, entry.getKey(), group.files, group.nullFreeColumns, group.partitionKeysInPath); } } return merged; @@ -155,7 +190,8 @@ public List extractOutput(Map accumulator) { SchemaGroup.of( entry.getKey(), entry.getValue().files, - new ArrayList<>(entry.getValue().nullFreeColumns))); + new ArrayList<>(entry.getValue().nullFreeColumns), + new ArrayList<>(entry.getValue().partitionKeysInPath))); } schemas.sort( (a, b) -> { @@ -200,12 +236,15 @@ private GroupCoder() {} public void encode(Group value, OutputStream out) throws IOException { VarLongCoder.of().encode(value.files, out); COLUMNS_CODER.encode(new ArrayList<>(value.nullFreeColumns), out); + COLUMNS_CODER.encode(new ArrayList<>(value.partitionKeysInPath), out); } @Override public Group decode(InputStream in) throws IOException { long files = VarLongCoder.of().decode(in); - return new Group(files, new TreeSet<>(COLUMNS_CODER.decode(in))); + TreeSet nullFreeColumns = new TreeSet<>(COLUMNS_CODER.decode(in)); + TreeSet partitionKeysInPath = new TreeSet<>(COLUMNS_CODER.decode(in)); + return new Group(files, nullFreeColumns, partitionKeysInPath); } } @@ -213,23 +252,34 @@ private static void add( Map accumulator, String schemaJson, long files, - Iterable nullFreeColumns) { + Iterable nullFreeColumns, + Iterable partitionKeysInPath) { Group existing = accumulator.get(schemaJson); if (existing == null) { - TreeSet copy = new TreeSet<>(); - for (String column : nullFreeColumns) { - copy.add(column); - } - accumulator.put(schemaJson, new Group(files, copy)); + accumulator.put( + schemaJson, new Group(files, copy(nullFreeColumns), copy(partitionKeysInPath))); return; } existing.files += files; - TreeSet stillNullFree = new TreeSet<>(); - for (String column : nullFreeColumns) { - if (existing.nullFreeColumns.contains(column)) { - stillNullFree.add(column); + existing.nullFreeColumns = intersect(existing.nullFreeColumns, nullFreeColumns); + existing.partitionKeysInPath = intersect(existing.partitionKeysInPath, partitionKeysInPath); + } + + private static TreeSet copy(Iterable values) { + TreeSet copy = new TreeSet<>(); + for (String value : values) { + copy.add(value); + } + return copy; + } + + private static TreeSet intersect(TreeSet kept, Iterable values) { + TreeSet stillKept = new TreeSet<>(); + for (String value : values) { + if (kept.contains(value)) { + stillKept.add(value); } } - existing.nullFreeColumns = stillNullFree; + return stillKept; } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PartitionPaths.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PartitionPaths.java new file mode 100644 index 000000000000..76dcf612f20b --- /dev/null +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/PartitionPaths.java @@ -0,0 +1,91 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; +import java.util.TreeSet; +import org.apache.iceberg.PartitionField; +import org.apache.iceberg.PartitionSpec; +import org.checkerframework.checker.nullness.qual.Nullable; + +/** + * Partition values in a file's path. With prefix {@code gs://lake/orders/}, the file {@code + * gs://lake/orders/dt=2024-01-15/f.parquet} has {@code dt = 2024-01-15}. If the table is + * partitioned by {@code dt} as is (not by a transform such as {@code bucket}), readers take {@code + * dt} from that value, so the file needs no {@code dt} column. + */ +final class PartitionPaths { + /** + * Hive writes {@code dt=__HIVE_DEFAULT_PARTITION__} when dt is null; Iceberg reads it as null. + */ + static final String HIVE_NULL = "__HIVE_DEFAULT_PARTITION__"; + + private PartitionPaths() {} + + /** + * With prefix {@code gs://lake/orders/}: {@code gs://lake/orders/dt=1/region=EU/f.parquet} gives + * {@code dt=1/region=EU}, and {@code gs://lake/orders/f.parquet} gives "". + */ + static String relative(String filePath, String prefix) { + String partitionPath = filePath.substring(prefix.length()); + int lastSlashIndex = partitionPath.lastIndexOf('/'); + return lastSlashIndex > 0 ? partitionPath.substring(0, lastSlashIndex) : ""; + } + + /** + * The partition keys the file's folders give a value: {@code dt=1/region=EU/f.parquet} gives + * {@code [dt, region]}, {@code dt=__HIVE_DEFAULT_PARTITION__/region=EU/f.parquet} gives {@code + * [region]}, and {@code 2024/f.parquet} gives none. Also none without a prefix or for a file + * outside it. Needs no table, unlike Iceberg's parser, so the pre-pass can use it. + */ + static Set partitionKeysInPath(String filePath, @Nullable String prefix) { + Set keys = new TreeSet<>(); + if (prefix == null || prefix.isEmpty() || !filePath.startsWith(prefix)) { + return keys; + } + String relative = relative(filePath, prefix); + if (relative.isEmpty()) { + return keys; + } + for (String folder : relative.split("/", -1)) { + String[] parts = folder.split("=", 2); + if (parts.length == 2 && !parts[0].isEmpty() && !HIVE_NULL.equals(parts[1])) { + keys.add(parts[0]); + } + } + return keys; + } + + /** + * Ids of the columns that identity partitions named by these keys fill: {@code dt} for key {@code + * dt} in a table partitioned by {@code dt}. Other transforms fill nothing: a file under {@code + * id_bucket=3/} in a table partitioned by {@code bucket(id, 4)} is in bucket 3, which does not + * say what its {@code id} is. + */ + static Set identityColumnsFilledByKeys(PartitionSpec spec, Collection keys) { + Set ids = new HashSet<>(); + for (PartitionField field : spec.fields()) { + if (field.transform().isIdentity() && keys.contains(field.name())) { + ids.add(field.sourceId()); + } + } + return ids; + } +} diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java index f111ed4b8554..37310b8d7f71 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchema.java @@ -20,7 +20,9 @@ import static org.apache.beam.sdk.metrics.Metrics.counter; import static org.apache.beam.sdk.util.Preconditions.checkStateNotNull; +import java.util.ArrayList; import java.util.Collections; +import java.util.List; import java.util.concurrent.Callable; import org.apache.beam.sdk.metrics.Counter; import org.apache.beam.sdk.transforms.DoFn; @@ -61,16 +63,31 @@ class ReadFooterSchema extends DoFn static final String UNCHECKED_FORMAT_KEY = "unchecked"; private final SchemaEvolutionConfig config; + private final @Nullable String locationPrefix; private final int threadPoolSize; private final int maxInFlightTasks; private transient @MonotonicNonNull BoundedAsyncTasks tasks; ReadFooterSchema(SchemaEvolutionConfig config) { - this(config, DEFAULT_THREAD_POOL_SIZE, DEFAULT_MAX_IN_FLIGHT_TASKS); + this(config, null); + } + + /** {@code locationPrefix}: where the {@code key=value} folders start, e.g. gs://lake/orders/. */ + ReadFooterSchema(SchemaEvolutionConfig config, @Nullable String locationPrefix) { + this(config, locationPrefix, DEFAULT_THREAD_POOL_SIZE, DEFAULT_MAX_IN_FLIGHT_TASKS); } ReadFooterSchema(SchemaEvolutionConfig config, int threadPoolSize, int maxInFlightTasks) { + this(config, null, threadPoolSize, maxInFlightTasks); + } + + ReadFooterSchema( + SchemaEvolutionConfig config, + @Nullable String locationPrefix, + int threadPoolSize, + int maxInFlightTasks) { this.config = config; + this.locationPrefix = locationPrefix; this.threadPoolSize = threadPoolSize; this.maxInFlightTasks = maxInFlightTasks; } @@ -186,7 +203,11 @@ private Callable createReadTask( } } try { - return new ReadResult(FileSchemas.schemaGroup(footer), false, timestamp, window, paneInfo); + List partitionKeysInPath = + new ArrayList<>(PartitionPaths.partitionKeysInPath(filePath, locationPrefix)); + CollectDistinctSchemas.SchemaGroup group = + FileSchemas.schemaGroup(footer).withPartitionKeysInPath(partitionKeysInPath); + return new ReadResult(group, false, timestamp, window, paneInfo); } catch (Exception e) { return unconvertibleSchema(filePath, e, timestamp, window, paneInfo); } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java index 743558e68f3c..cd599b746804 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaDelta.java @@ -60,7 +60,12 @@ private SchemaDelta(List changes) { * (dotted, empty, case-colliding) come back as conflicts without attempting the union. */ static SchemaDelta classify(Table table, Schema fileSchema) { - return classify(table, table.schema(), fileSchema); + return classify(table, table.schema(), fileSchema, Collections.emptySet()); + } + + /** As {@link #classify(Table, Schema, Schema, Set)}, against the table's current schema. */ + static SchemaDelta classify(Table table, Schema fileSchema, Set filledOutsideFile) { + return classify(table, table.schema(), fileSchema, filledOutsideFile); } /** @@ -68,6 +73,17 @@ static SchemaDelta classify(Table table, Schema fileSchema) { * is applied to the table's current metadata, which the caller keeps equal to the snapshot. */ static SchemaDelta classify(Table table, Schema before, Schema fileSchema) { + return classify(table, before, fileSchema, Collections.emptySet()); + } + + /** + * {@code filledOutsideFile}: ids of required table columns whose values come from somewhere other + * than the file and are never null, so the file may lack them without the column becoming + * optional. For example {@code dt} for a file under {@code dt=2024-01-15/} in a table partitioned + * by {@code dt}. + */ + static SchemaDelta classify( + Table table, Schema before, Schema fileSchema, Set filledOutsideFile) { if (before.sameSchema(fileSchema)) { return new SchemaDelta(Collections.emptyList()); } @@ -81,7 +97,7 @@ static SchemaDelta classify(Table table, Schema before, Schema fileSchema) { } List absent = new ArrayList<>(); - findAbsentRequired(before.asStruct(), fileSchema.asStruct(), "", absent); + findAbsentRequired(before.asStruct(), fileSchema.asStruct(), "", filledOutsideFile, absent); Schema merged; try { // The absent-path relaxations are applied here too, so anything Iceberg refuses (an @@ -114,12 +130,13 @@ private static void findAbsentRequired( Types.StructType tableStruct, Types.StructType fileStruct, String prefix, + Set filledOutsideFile, List changes) { for (Types.NestedField field : tableStruct.fields()) { String rawPath = prefix + field.name(); Types.NestedField fileField = fileStruct.field(field.name()); if (fileField == null) { - if (field.isRequired()) { + if (field.isRequired() && !filledOutsideFile.contains(field.fieldId())) { changes.add( new SchemaChange( SchemaChange.Kind.FIELD_RELAXATION, @@ -132,25 +149,36 @@ private static void findAbsentRequired( } continue; } - findAbsentRequiredInType(field.type(), fileField.type(), rawPath, changes); + findAbsentRequiredInType(field.type(), fileField.type(), rawPath, filledOutsideFile, changes); } } private static void findAbsentRequiredInType( - Type tableType, Type fileType, String rawPath, List changes) { + Type tableType, + Type fileType, + String rawPath, + Set filledOutsideFile, + List changes) { if (tableType.isStructType() && fileType.isStructType()) { - findAbsentRequired(tableType.asStructType(), fileType.asStructType(), rawPath + ".", changes); + findAbsentRequired( + tableType.asStructType(), + fileType.asStructType(), + rawPath + ".", + filledOutsideFile, + changes); } else if (tableType.isListType() && fileType.isListType()) { findAbsentRequiredInType( tableType.asListType().elementType(), fileType.asListType().elementType(), rawPath + ".element", + filledOutsideFile, changes); } else if (tableType.isMapType() && fileType.isMapType()) { findAbsentRequiredInType( tableType.asMapType().valueType(), fileType.asMapType().valueType(), rawPath + ".value", + filledOutsideFile, changes); } } diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java index e9e72199c82b..0c298e79542e 100644 --- a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java +++ b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/SchemaPlan.java @@ -346,7 +346,10 @@ private static Verdicts classify( Schema fileSchema = FileSchemas.markRequired( SchemaParser.fromJson(group.getSchemaJson()), group.getNullFreeColumns()); - SchemaDelta delta = SchemaDelta.classify(table, base, fileSchema); + Set identityPartitionColumnsFromPath = + PartitionPaths.identityColumnsFilledByKeys(table.spec(), group.getPartitionKeysInPath()); + SchemaDelta delta = + SchemaDelta.classify(table, base, fileSchema, identityPartitionColumnsFromPath); if (delta.isEmpty()) { continue; } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java index 276ea4b0ee90..7595483d58ac 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesIT.java @@ -778,6 +778,104 @@ public void testBatchImportOfParquetFilesWithoutExtension() throws IOException { checkRecordsInDestinationTable(/* alsoCheckWithBigQueryIO= */ true); } + /** + * Files under dt=2024-01-15/ and dt=2024-01-16/ with no dt column, into a table with a required + * dt: dt stays required, and Beam and BigQuery read it from the folder names. + */ + @Test + public void testBatchImportOfRequiredPartitionColumnFromThePath() throws IOException { + org.apache.iceberg.Schema withDt = + new org.apache.iceberg.Schema( + org.apache.iceberg.types.Types.NestedField.required( + 1, "id", org.apache.iceberg.types.Types.LongType.get()), + org.apache.iceberg.types.Types.NestedField.required( + 2, "name", org.apache.iceberg.types.Types.StringType.get()), + org.apache.iceberg.types.Types.NestedField.required( + 3, "age", org.apache.iceberg.types.Types.IntegerType.get()), + org.apache.iceberg.types.Types.NestedField.required( + 4, "dt", org.apache.iceberg.types.Types.StringType.get())); + catalog.createTable( + destTableId, withDt, PartitionSpec.builderFor(withDt).identity("dt").build()); + + String parquetDir = format("gs://%s/%s/", DATA_BUCKET, dirName); + String tempDir = format("gs://%s/%s-tmp/", DATA_BUCKET, dirName); + writeParquet( + TEST_ROWS.subList(0, 10), ROW_SCHEMA, parquetDir + "dt=2024-01-15/", tempDir + "a/"); + writeParquet( + TEST_ROWS.subList(10, 20), ROW_SCHEMA, parquetDir + "dt=2024-01-16/", tempDir + "b/"); + GcsUtil gcsUtil = TestPipeline.testingPipelineOptions().as(GcsOptions.class).getGcsUtil(); + List files = new ArrayList<>(); + for (StorageObject object : gcsUtil.listObjects(DATA_BUCKET, dirName + "/", null).getItems()) { + files.add(format("gs://%s/%s", object.getBucket(), object.getName())); + } + assertEquals(20, files.size()); + + SchemaEvolutionConfig evolution = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .build(); + Pipeline p = Pipeline.create(); + PCollectionRowTuple tuple = + p.apply(Create.of(files)) + .apply( + new AddFiles( + IcebergCatalogConfig.builder().setCatalogProperties(LAKEHOUSE_PROPS).build(), + namespace + "." + destTableName, + parquetDir, + null, + null, + TABLE_PROPS, + null, + null, + evolution)); + PAssert.that(tuple.get("errors")).empty(); + p.run().waitUntilFinish(); + + Table destTable = catalog.loadTable(destTableId); + assertTrue("dt stays required", destTable.schema().findField("dt").isRequired()); + + List expected = new ArrayList<>(); + for (Row row : TEST_ROWS) { + String dt = checkStateNotNull(row.getInt64("id")) < 10 ? "2024-01-15" : "2024-01-16"; + expected.add(canonicalRecord(row) + "|" + dt); + } + Pipeline s = Pipeline.create(); + PCollection icebergRows = + s.apply( + Managed.read(Managed.ICEBERG) + .withConfig( + ImmutableMap.of( + "table", + destTableId.toString(), + "catalog_properties", + LAKEHOUSE_PROPS))) + .getSinglePCollection() + .apply( + "canonicalize iceberg rows", MapElements.into(strings()).via(AddFilesIT::withDt)); + PAssert.that(icebergRows).containsInAnyOrder(expected); + PCollection bqRows = + s.apply( + "read with BigQueryIO", + Managed.read(Managed.BIGQUERY) + .withConfig( + ImmutableMap.of( + "table", + format( + "%s.%s.%s.%s", + PROJECT, + LakehouseTestCatalog.CATALOG_ID, + destTableId.namespace(), + destTableId.name())))) + .getSinglePCollection() + .apply("canonicalize bq rows", MapElements.into(strings()).via(AddFilesIT::withDt)); + PAssert.that(bqRows).containsInAnyOrder(expected); + s.run().waitUntilFinish(); + } + + private static String withDt(Row row) { + return canonicalRecord(row) + "|" + String.valueOf((Object) row.getValue("dt")); + } + private static void writeParquet(List rows, Schema schema, String dir, String tempDir) { Pipeline q = Pipeline.create(); org.apache.avro.Schema avroSchema = AvroUtils.toAvroSchema(schema); diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java index 578d18c539d2..47aa8290f724 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/AddFilesTest.java @@ -96,6 +96,7 @@ import org.apache.iceberg.TableProperties; import org.apache.iceberg.catalog.TableIdentifier; import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.IcebergGenerics; import org.apache.iceberg.data.Record; import org.apache.iceberg.data.parquet.GenericParquetReaders; import org.apache.iceberg.data.parquet.GenericParquetWriter; @@ -1086,6 +1087,112 @@ private Record record(int id, String name, int age) { return GenericRecord.create(icebergSchema).copy("id", id, "name", name, "age", age); } + // ---- partition column read from the path (dt=.../) + + private static final Schema WITH_DT = + new Schema( + Types.NestedField.required(1, "id", Types.IntegerType.get()), + Types.NestedField.required(2, "name", Types.StringType.get()), + Types.NestedField.required(3, "age", Types.IntegerType.get()), + Types.NestedField.required(4, "dt", Types.StringType.get())); + + private void createTablePartitionedByDt() { + catalog.createTable(tableId, WITH_DT, PartitionSpec.builderFor(WITH_DT).identity("dt").build()); + } + + /** Writes a file without a dt column to root/folder/, e.g. root/dt=2024-01-15/. */ + private String writeUnder(String folder, int id) throws IOException { + new File(root + folder).mkdirs(); + String file = root + folder + "/data" + id + ".parquet"; + DataWriter writer = createWriter(file); + writer.write(record(id, "n" + id, id)); + writer.close(); + return file; + } + + private AddFiles addFilesUnderRoot(SchemaEvolutionConfig config) { + return new AddFiles( + catalogConfig, tableId.toString(), root, null, null, null, null, null, config); + } + + /** Files under dt=2024-01-15/ and dt=2024-01-16/ with no dt column; dt reads back from them. */ + @Test + public void testRequiredPartitionColumnFromThePathNeedsNoRelaxation() throws Exception { + createTablePartitionedByDt(); + String first = writeUnder("dt=2024-01-15", 1); + String second = writeUnder("dt=2024-01-16", 2); + + PCollectionRowTuple output = + pipeline + .apply("Create Input", Create.of(first, second)) + .apply(addFilesUnderRoot(ADDITIONS)); + PAssert.that(output.get("errors")).empty(); + + pipeline.run().waitUntilFinish(); + + Table table = catalog.loadTable(tableId); + assertTrue("dt stays required", table.schema().findField("dt").isRequired()); + List rows = new ArrayList<>(); + try (CloseableIterable records = IcebergGenerics.read(table).build()) { + for (Record record : records) { + rows.add(record.getField("id") + " " + record.getField("dt")); + } + } + Collections.sort(rows); + assertEquals(Arrays.asList("1 2024-01-15", "2 2024-01-16"), rows); + } + + @Test + public void testPinnedPartitionColumnFromThePathIsSatisfied() throws Exception { + createTablePartitionedByDt(); + String file = writeUnder("dt=2024-01-15", 1); + SchemaEvolutionConfig pinned = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .setRequiredColumns(Collections.singleton("dt")) + .build(); + + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(file)).apply(addFilesUnderRoot(pinned)); + PAssert.that(output.get("errors")).empty(); + + pipeline.run().waitUntilFinish(); + + assertEquals(1, Iterables.size(catalog.loadTable(tableId).newScan().planFiles())); + } + + /** A file directly under the prefix has no dt=... folder, so it alone is refused. */ + @Test + public void testFileWithoutAPathValueForARequiredPartitionColumnIsRefused() throws Exception { + createTablePartitionedByDt(); + String inFolder = writeUnder("dt=2024-01-15", 1); + String stray = root + "stray.parquet"; + DataWriter writer = createWriter(stray); + writer.write(record(2, "n2", 2)); + writer.close(); + SchemaEvolutionConfig routed = + SchemaEvolutionConfig.builder() + .setOptions(EnumSet.of(SchemaEvolutionOption.ALLOW_FIELD_ADDITION)) + .setIncompatibleSchemaHandling( + SchemaEvolutionConfig.IncompatibleSchemaHandling.ROUTE_TO_ERRORS) + .build(); + + PCollectionRowTuple output = + pipeline.apply("Create Input", Create.of(inFolder, stray)).apply(addFilesUnderRoot(routed)); + PAssert.that(output.get("errors")) + .satisfies( + rows -> { + assertEquals(stray, Iterables.getOnlyElement(rows).getString("file")); + return null; + }); + + pipeline.run().waitUntilFinish(); + + Table table = catalog.loadTable(tableId); + assertTrue("dt stays required", table.schema().findField("dt").isRequired()); + assertEquals(1, Iterables.size(table.newScan().planFiles())); + } + // ---- AddFiles with schema evolution private AddFiles addFiles(@Nullable SchemaEvolutionConfig config) { diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemasTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemasTest.java index 7b4a22b2618e..76f650758f65 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemasTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CollectDistinctSchemasTest.java @@ -130,6 +130,29 @@ public void testNullFreeColumnsIntersectAcrossMergedAccumulators() { out); } + /** One file under dt=.../region=EU/, one under dt=.../ only: the group keeps [dt]. */ + @Test + public void testPartitionKeysInPathIntersect() { + List out = + combine( + withPartitionKeysInPath(ID_NAME, Arrays.asList("dt", "region")), + withPartitionKeysInPath(ID_NAME, Arrays.asList("dt")), + withPartitionKeysInPath(NAME_ID, NONE)); + assertEquals( + Arrays.asList( + SchemaGroup.of(ID_NAME, 2, NONE, Arrays.asList("dt")), + SchemaGroup.of(NAME_ID, 1, NONE, NONE)), + out); + } + + @Test + public void testAccumulatorCoderRoundTripWithPartitionKeysInPath() throws Exception { + Coder> coder = fn.getAccumulatorCoder(null, null); + Map accumulator = + fn.addInput(fn.createAccumulator(), withPartitionKeysInPath(ID_NAME, Arrays.asList("dt"))); + CoderProperties.coderDecodeEncodeEqual(coder, accumulator); + } + @Test public void testEmptyInput() { assertEquals(Arrays.asList(), combine()); @@ -182,6 +205,10 @@ private static SchemaGroup group(String json, long files, List proven) { return SchemaGroup.of(json, files, proven); } + private static SchemaGroup withPartitionKeysInPath(String json, List keys) { + return SchemaGroup.of(json, 1, NONE, keys); + } + private static String json(Schema schema) { return SchemaParser.toJson(schema); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java index 0c5f379a1105..ce226f37f9f0 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/CommitSchemaUnionTest.java @@ -41,6 +41,7 @@ import org.apache.beam.sdk.util.FastNanoClockAndSleeper; import org.apache.hadoop.conf.Configuration; import org.apache.iceberg.BaseTable; +import org.apache.iceberg.PartitionSpec; import org.apache.iceberg.Schema; import org.apache.iceberg.SchemaParser; import org.apache.iceberg.Table; @@ -117,6 +118,24 @@ private static CollectDistinctSchemas.SchemaGroup files( return CollectDistinctSchemas.SchemaGroup.of(json(schema), count, nullFreeColumns); } + /** {@code count} files, each under a folder per key, e.g. dt=2024-01-15/ for key dt. */ + private static CollectDistinctSchemas.SchemaGroup filesUnder( + Schema schema, long count, String... partitionKeysInPath) { + return CollectDistinctSchemas.SchemaGroup.of( + json(schema), count, Collections.emptyList(), Arrays.asList(partitionKeysInPath)); + } + + private static final Schema PARTITIONED = + new Schema( + required(1, "id", Types.LongType.get()), required(2, "dt", Types.StringType.get())); + + private TableIdentifier partitionedByDt() { + TableIdentifier id = missing(); + catalog.createTable( + id, PARTITIONED, PartitionSpec.builderFor(PARTITIONED).identity("dt").build()); + return id; + } + private long commit( SchemaEvolutionConfig config, IncompatibleSchemaHandling handling, @@ -744,6 +763,44 @@ public void testTableRetryCountCapsTheAttempts() { // ---- create path + /** Files under dt=.../ folders, with no dt column: required dt stays required. */ + @Test + public void testRequiredPartitionColumnSuppliedByThePathIsNotRelaxed() { + TableIdentifier id = partitionedByDt(); + Schema file = new Schema(required(1, "id", Types.LongType.get())); + int before = catalog.loadTable(id).schema().schemaId(); + + long schemaId = + commitTo( + id, + ADDITION_ONLY, + IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, + filesUnder(file, 3, "dt")); + + assertEquals(before, schemaId); + assertTrue(catalog.loadTable(id).schema().findField("dt").isRequired()); + } + + /** Files with no dt=... folder would read dt as null, so dt must still become optional. */ + @Test + public void testRequiredPartitionColumnWithoutAPathValueStillNeedsRelaxation() { + TableIdentifier id = partitionedByDt(); + Schema file = new Schema(required(1, "id", Types.LongType.get())); + + assertThrows( + IncompatibleSchemaException.class, + () -> + commitTo( + id, + ADDITION_ONLY, + IncompatibleSchemaHandling.FAIL_PIPELINE, + NO_CREATION, + filesUnder(file, 3))); + commitTo(id, ALL, IncompatibleSchemaHandling.FAIL_PIPELINE, NO_CREATION, filesUnder(file, 3)); + assertTrue(catalog.loadTable(id).schema().findField("dt").isOptional()); + } + private TableIdentifier missing() { return TableIdentifier.of("default", testName.getMethodName() + "_new"); } diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PartitionPathsTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PartitionPathsTest.java new file mode 100644 index 000000000000..e5883ee9fbe8 --- /dev/null +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/PartitionPathsTest.java @@ -0,0 +1,101 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.beam.sdk.io.iceberg; + +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.junit.Assert.assertEquals; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.types.Types; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public class PartitionPathsTest { + private static final String PREFIX = "gs://lake/orders/"; + + private static final Schema SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), + required(2, "dt", Types.StringType.get()), + required(3, "region", Types.StringType.get())); + + @Test + public void testRelativeIsTheFoldersAfterThePrefix() { + assertEquals( + "dt=2024-01-15/region=EU", + PartitionPaths.relative(PREFIX + "dt=2024-01-15/region=EU/f.parquet", PREFIX)); + assertEquals("", PartitionPaths.relative(PREFIX + "f.parquet", PREFIX)); + } + + @Test + public void testPartitionKeysInPath() { + assertEquals( + set("dt", "region"), + PartitionPaths.partitionKeysInPath(PREFIX + "dt=2024-01-15/region=EU/f.parquet", PREFIX)); + assertEquals( + "a value may hold '='", + set("dt"), + PartitionPaths.partitionKeysInPath(PREFIX + "dt=a=b/f.parquet", PREFIX)); + assertEquals( + "plain folders and empty keys give nothing", + set("dt"), + PartitionPaths.partitionKeysInPath(PREFIX + "2024/=x/dt=1/f.parquet", PREFIX)); + } + + @Test + public void testHiveNullGivesNoValue() { + assertEquals( + set("region"), + PartitionPaths.partitionKeysInPath( + PREFIX + "dt=__HIVE_DEFAULT_PARTITION__/region=EU/f.parquet", PREFIX)); + } + + @Test + public void testNoKeysWithoutAPrefixOrOutsideIt() { + assertEquals(set(), PartitionPaths.partitionKeysInPath(PREFIX + "dt=1/f.parquet", null)); + assertEquals(set(), PartitionPaths.partitionKeysInPath(PREFIX + "dt=1/f.parquet", "")); + assertEquals(set(), PartitionPaths.partitionKeysInPath("gs://other/dt=1/f.parquet", PREFIX)); + assertEquals(set(), PartitionPaths.partitionKeysInPath(PREFIX + "f.parquet", PREFIX)); + } + + @Test + public void testIdentityColumnsFilledByKeysIgnoresOtherTransforms() { + PartitionSpec spec = + PartitionSpec.builderFor(SCHEMA).identity("dt").bucket("region", 4).build(); + assertEquals( + set(2), + PartitionPaths.identityColumnsFilledByKeys( + spec, Arrays.asList("dt", "region_bucket", "unknown"))); + assertEquals(set(), PartitionPaths.identityColumnsFilledByKeys(spec, Collections.emptyList())); + assertEquals( + set(), + PartitionPaths.identityColumnsFilledByKeys( + PartitionSpec.unpartitioned(), Arrays.asList("dt"))); + } + + @SafeVarargs + private static HashSet set(T... values) { + return new HashSet<>(Arrays.asList(values)); + } +} diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java index a6c178a520f2..ea5a6e768685 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/ReadFooterSchemaTest.java @@ -28,6 +28,7 @@ import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; +import java.util.Collections; import java.util.List; import org.apache.avro.SchemaBuilder; import org.apache.beam.sdk.PipelineResult; @@ -183,6 +184,43 @@ public void testExtensionlessNonParquetEmitsNothing() throws IOException { pipeline.run(); } + @Test + public void testRecordsPartitionKeysInPath() throws IOException { + String prefix = temporaryFolder.getRoot().getAbsolutePath() + "/"; + new File(prefix + "dt=2024-01-15/region=EU").mkdirs(); + String file = + writeParquet( + "dt=2024-01-15/region=EU/f.parquet", FLAT_SCHEMA, record(FLAT_SCHEMA, "id", 1)); + + assertPartitionKeysInPath(runWithPrefix(prefix, file), Arrays.asList("dt", "region")); + + pipeline.run(); + } + + @Test + public void testHiveNullFolderGivesNoPartitionKey() throws IOException { + String prefix = temporaryFolder.getRoot().getAbsolutePath() + "/"; + new File(prefix + "dt=__HIVE_DEFAULT_PARTITION__").mkdirs(); + String file = + writeParquet( + "dt=__HIVE_DEFAULT_PARTITION__/f.parquet", FLAT_SCHEMA, record(FLAT_SCHEMA, "id", 1)); + + assertPartitionKeysInPath(runWithPrefix(prefix, file), Collections.emptyList()); + + pipeline.run(); + } + + @Test + public void testNoPartitionKeysWithoutAPrefix() throws IOException { + new File(temporaryFolder.getRoot(), "dt=2024-01-15").mkdirs(); + String file = + writeParquet("dt=2024-01-15/f.parquet", FLAT_SCHEMA, record(FLAT_SCHEMA, "id", 1)); + + assertPartitionKeysInPath(run(file), Collections.emptyList()); + + pipeline.run(); + } + /** * The schema is emitted exactly as declared; which columns proved null-free travels beside it. */ @@ -309,6 +347,26 @@ private PCollection run(String... paths) { .setCoder(CollectDistinctSchemas.groupCoder()); } + private PCollection runWithPrefix( + String prefix, String... paths) { + return pipeline + .apply(Create.of(Arrays.asList(paths))) + .apply(ParDo.of(new ReadFooterSchema(SchemaEvolutionConfig.disabled(), prefix))) + .setCoder(CollectDistinctSchemas.groupCoder()); + } + + private static void assertPartitionKeysInPath( + PCollection out, List expected) { + PAssert.that(out) + .satisfies( + groups -> { + for (CollectDistinctSchemas.SchemaGroup group : groups) { + assertEquals(expected, group.getPartitionKeysInPath()); + } + return null; + }); + } + /** Asserts the emitted declared schemas equal the canonical forms of {@code expected}. */ private static void assertSchemas( PCollection out, Schema... expected) { diff --git a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java index d3d4172f0f95..0832f618ff08 100644 --- a/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java +++ b/sdks/java/io/iceberg/src/test/java/org/apache/beam/sdk/io/iceberg/SchemaDeltaTest.java @@ -27,6 +27,7 @@ import static org.junit.Assert.assertTrue; import java.util.Arrays; +import java.util.Collections; import java.util.EnumSet; import java.util.HashSet; import org.apache.beam.sdk.io.iceberg.SchemaChange.Kind; @@ -128,6 +129,17 @@ public void testAbsentRequiredColumnIsRelaxation() { delta.disallowedReason(pinned("id"))); } + /** As for a file under id=7/ in a table partitioned by id: required id is not made optional. */ + @Test + public void testAbsentRequiredColumnSuppliedByThePathIsNoChange() { + Schema file = new Schema(optional(1, "name", Types.StringType.get())); + assertEquals(Arrays.asList("id"), classify(file).absentRequiredPaths()); + Table created = checkStateNotNull(table); + int id = created.schema().findField("id").fieldId(); + + assertTrue(SchemaDelta.classify(created, file, Collections.singleton(id)).isEmpty()); + } + @Test public void testAbsentNestedRequiredChildIsRelaxation() { Schema file =