This is an automated email from the ASF dual-hosted git repository.
claudevdm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 2c21f02ce61 AddFiles: Parse partition keys in filepath for requirement
check (#40691)
2c21f02ce61 is described below
commit 2c21f02ce6154dc01093433af682ddc78c1f7e49
Author: claudevdm <[email protected]>
AuthorDate: Fri Oct 23 17:24:23 2026 -0400
AddFiles: Parse partition keys in filepath for requirement check (#40691)
* initial
* revert-doc
* trigger tests
* comments
---
.../IO_Iceberg_Integration_Tests.json | 2 +-
.../org/apache/beam/sdk/io/iceberg/AddFiles.java | 49 +++++++---
.../sdk/io/iceberg/CollectDistinctSchemas.java | 94 +++++++++++++-----
.../apache/beam/sdk/io/iceberg/PartitionPaths.java | 91 ++++++++++++++++++
.../beam/sdk/io/iceberg/ReadFooterSchema.java | 25 ++++-
.../apache/beam/sdk/io/iceberg/SchemaDelta.java | 40 ++++++--
.../org/apache/beam/sdk/io/iceberg/SchemaPlan.java | 5 +-
.../org/apache/beam/sdk/io/iceberg/AddFilesIT.java | 98 +++++++++++++++++++
.../apache/beam/sdk/io/iceberg/AddFilesTest.java | 107 +++++++++++++++++++++
.../sdk/io/iceberg/CollectDistinctSchemasTest.java | 27 ++++++
.../beam/sdk/io/iceberg/CommitSchemaUnionTest.java | 57 +++++++++++
.../beam/sdk/io/iceberg/PartitionPathsTest.java | 101 +++++++++++++++++++
.../beam/sdk/io/iceberg/ReadFooterSchemaTest.java | 58 +++++++++++
.../beam/sdk/io/iceberg/SchemaDeltaTest.java | 12 +++
14 files changed, 721 insertions(+), 45 deletions(-)
diff --git a/.github/trigger_files/IO_Iceberg_Integration_Tests.json
b/.github/trigger_files/IO_Iceberg_Integration_Tests.json
index b73af5e61a4..7ab7bcd9a9c 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 c72479aa39b..00cf77af1dc 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 class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
/**
* @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 class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
private PCollection<List<CollectDistinctSchemas.SchemaGroup>>
distinctSchemas(
PCollection<String> 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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
} catch (Exception e) {
return Verdict.error(UNREADABLE_SCHEMA_ERROR + errorMessage(e));
}
- @Nullable String uncovered = uncoveredReason(fileSchema);
+ Set<Integer> 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<Integer> 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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
* 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<Integer>
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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
table.refresh();
}
}
- delta = SchemaDelta.classify(table, fileSchema);
+ delta = SchemaDelta.classify(table, fileSchema,
identityPartitionColumnsFromPath);
if (delta.isEmpty()) {
return null;
}
@@ -857,14 +874,23 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
* 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<Integer> identityPartitionColumnsFromPath) {
Table table = checkStateNotNull(this.table);
List<String> 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 @@ public class AddFiles extends
PTransform<PCollection<String>, PCollectionRowTupl
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 5c36967cfeb..8e11bd1ed3e 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.IOException;
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 @@ import org.checkerframework.checker.nullness.qual.Nullable;
/**
* 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<String> nullFreeColumns;
+ TreeSet<String> partitionKeysInPath;
- Group(long files, TreeSet<String> nullFreeColumns) {
+ Group(long files, TreeSet<String> nullFreeColumns, TreeSet<String>
partitionKeysInPath) {
this.files = files;
this.nullFreeColumns = nullFreeColumns;
+ this.partitionKeysInPath = partitionKeysInPath;
}
@Override
@@ -73,12 +77,14 @@ class CollectDistinctSchemas
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 @@ class CollectDistinctSchemas
private static @MonotonicNonNull SchemaCoder<SchemaGroup> coder;
static SchemaGroup of(String schemaJson, long files, List<String>
nullFreeColumns) {
- return new AutoValue_CollectDistinctSchemas_SchemaGroup(schemaJson,
files, nullFreeColumns);
+ return of(schemaJson, files, nullFreeColumns, Collections.emptyList());
+ }
+
+ static SchemaGroup of(
+ String schemaJson,
+ long files,
+ List<String> nullFreeColumns,
+ List<String> partitionKeysInPath) {
+ return new AutoValue_CollectDistinctSchemas_SchemaGroup(
+ schemaJson, files, nullFreeColumns, partitionKeysInPath);
+ }
+
+ SchemaGroup withPartitionKeysInPath(List<String> partitionKeysInPath) {
+ return of(getSchemaJson(), getFiles(), getNullFreeColumns(),
partitionKeysInPath);
}
static SchemaCoder<SchemaGroup> getCoder() {
@@ -115,11 +134,21 @@ class CollectDistinctSchemas
@SchemaFieldNumber("2")
abstract List<String> 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<String> 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 @@ class CollectDistinctSchemas
@Override
public Map<String, Group> addInput(Map<String, Group> 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 @@ class CollectDistinctSchemas
Map<String, Group> merged = createAccumulator();
for (Map<String, Group> accumulator : accumulators) {
for (Map.Entry<String, Group> 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 @@ class CollectDistinctSchemas
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 @@ class CollectDistinctSchemas
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<String> nullFreeColumns = new
TreeSet<>(COLUMNS_CODER.decode(in));
+ TreeSet<String> partitionKeysInPath = new
TreeSet<>(COLUMNS_CODER.decode(in));
+ return new Group(files, nullFreeColumns, partitionKeysInPath);
}
}
@@ -213,23 +252,34 @@ class CollectDistinctSchemas
Map<String, Group> accumulator,
String schemaJson,
long files,
- Iterable<String> nullFreeColumns) {
+ Iterable<String> nullFreeColumns,
+ Iterable<String> partitionKeysInPath) {
Group existing = accumulator.get(schemaJson);
if (existing == null) {
- TreeSet<String> 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<String> 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<String> copy(Iterable<String> values) {
+ TreeSet<String> copy = new TreeSet<>();
+ for (String value : values) {
+ copy.add(value);
+ }
+ return copy;
+ }
+
+ private static TreeSet<String> intersect(TreeSet<String> kept,
Iterable<String> values) {
+ TreeSet<String> 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 00000000000..76dcf612f20
--- /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<String> partitionKeysInPath(String filePath, @Nullable String
prefix) {
+ Set<String> 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<Integer> identityColumnsFilledByKeys(PartitionSpec spec,
Collection<String> keys) {
+ Set<Integer> 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 f111ed4b855..37310b8d7f7 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 @@ package org.apache.beam.sdk.io.iceberg;
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<String,
CollectDistinctSchemas.SchemaGroup>
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<ReadResult> 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 @@ class ReadFooterSchema extends DoFn<String,
CollectDistinctSchemas.SchemaGroup>
}
}
try {
- return new ReadResult(FileSchemas.schemaGroup(footer), false,
timestamp, window, paneInfo);
+ List<String> 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 743558e68f3..cd599b74680 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 @@ final class SchemaDelta {
* (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<Integer>
filledOutsideFile) {
+ return classify(table, table.schema(), fileSchema, filledOutsideFile);
}
/**
@@ -68,6 +73,17 @@ final class SchemaDelta {
* 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<Integer>
filledOutsideFile) {
if (before.sameSchema(fileSchema)) {
return new SchemaDelta(Collections.emptyList());
}
@@ -81,7 +97,7 @@ final class SchemaDelta {
}
List<SchemaChange> 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 @@ final class SchemaDelta {
Types.StructType tableStruct,
Types.StructType fileStruct,
String prefix,
+ Set<Integer> filledOutsideFile,
List<SchemaChange> 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 @@ final class SchemaDelta {
}
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<SchemaChange>
changes) {
+ Type tableType,
+ Type fileType,
+ String rawPath,
+ Set<Integer> filledOutsideFile,
+ List<SchemaChange> 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 e9e72199c82..0c298e79542 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 @@ abstract class SchemaPlan {
Schema fileSchema =
FileSchemas.markRequired(
SchemaParser.fromJson(group.getSchemaJson()),
group.getNullFreeColumns());
- SchemaDelta delta = SchemaDelta.classify(table, base, fileSchema);
+ Set<Integer> 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 276ea4b0ee9..7595483d58a 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 class AddFilesIT {
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<String> 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<String> 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<String> 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<String> 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<Row> 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 578d18c539d..47aa8290f72 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.Table;
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 @@ public class AddFilesTest {
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<Record> 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<String> rows = new ArrayList<>();
+ try (CloseableIterable<Record> 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<Record> 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 7b4a22b2618..76f650758f6 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 class CollectDistinctSchemasTest {
out);
}
+ /** One file under dt=.../region=EU/, one under dt=.../ only: the group
keeps [dt]. */
+ @Test
+ public void testPartitionKeysInPathIntersect() {
+ List<SchemaGroup> 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<Map<String, Group>> coder = fn.getAccumulatorCoder(null, null);
+ Map<String, Group> 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 @@ public class CollectDistinctSchemasTest {
return SchemaGroup.of(json, files, proven);
}
+ private static SchemaGroup withPartitionKeysInPath(String json, List<String>
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 0c5f379a110..ce226f37f9f 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.io.iceberg.SchemaEvolutionConfig.IncompatibleSchemaHa
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 @@ public class CommitSchemaUnionTest {
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 class CommitSchemaUnionTest {
// ---- 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 00000000000..e5883ee9fbe
--- /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 <T> HashSet<T> 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 a6c178a520f..ea5a6e76868 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.charset.StandardCharsets;
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 class ReadFooterSchemaTest {
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 @@ public class ReadFooterSchemaTest {
.setCoder(CollectDistinctSchemas.groupCoder());
}
+ private PCollection<CollectDistinctSchemas.SchemaGroup> 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<CollectDistinctSchemas.SchemaGroup> out, List<String>
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<CollectDistinctSchemas.SchemaGroup> 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 d3d4172f0f9..0832f618ff0 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.assertNull;
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 class SchemaDeltaTest {
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 =
