This is an automated email from the ASF dual-hosted git repository.
danielcweeks pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 1e325bd6ce Core: Add v4 manifest reader (#16958)
1e325bd6ce is described below
commit 1e325bd6cec4cb19536a5d2577723a3570c43983
Author: Anoop Johnson <[email protected]>
AuthorDate: Thu Jul 30 07:30:54 2026 -0700
Core: Add v4 manifest reader (#16958)
* Core: Add v4 manifest reader
---
.../org/apache/iceberg/types/ReplaceTypeById.java | 107 +++
.../java/org/apache/iceberg/types/TypeUtil.java | 18 +
.../org/apache/iceberg/types/TestTypeUtil.java | 130 +++
.../main/java/org/apache/iceberg/Partitioning.java | 19 +
.../main/java/org/apache/iceberg/TrackedFile.java | 24 +-
.../java/org/apache/iceberg/TrackedFileStruct.java | 12 +-
.../java/org/apache/iceberg/TrackingStruct.java | 14 +-
.../java/org/apache/iceberg/V4ManifestReader.java | 300 +++++++
.../java/org/apache/iceberg/TestPartitioning.java | 21 +
.../java/org/apache/iceberg/TestTrackedFile.java | 37 +-
.../apache/iceberg/TestTrackedFileAdapters.java | 14 +-
.../org/apache/iceberg/TestTrackedFileStruct.java | 26 +-
.../org/apache/iceberg/TestV4ManifestReader.java | 887 +++++++++++++++++++++
13 files changed, 1564 insertions(+), 45 deletions(-)
diff --git a/api/src/main/java/org/apache/iceberg/types/ReplaceTypeById.java
b/api/src/main/java/org/apache/iceberg/types/ReplaceTypeById.java
new file mode 100644
index 0000000000..1c94bd57c1
--- /dev/null
+++ b/api/src/main/java/org/apache/iceberg/types/ReplaceTypeById.java
@@ -0,0 +1,107 @@
+/*
+ * 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.iceberg.types;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+class ReplaceTypeById extends TypeUtil.SchemaVisitor<Type> {
+ private final Map<Integer, Type> replacementsById;
+
+ ReplaceTypeById(Map<Integer, Type> replacementsById) {
+ this.replacementsById = replacementsById;
+ }
+
+ @Override
+ public Type schema(Schema schema, Type structResult) {
+ return structResult;
+ }
+
+ @Override
+ public Type struct(Types.StructType struct, List<Type> fieldResults) {
+ List<Types.NestedField> fields = struct.fields();
+ List<Types.NestedField> newFields =
Lists.newArrayListWithExpectedSize(fields.size());
+ boolean hasChanged = false;
+
+ for (int i = 0; i < fields.size(); i += 1) {
+ Type fieldReplacement = fieldResults.get(i);
+ Types.NestedField field = fields.get(i);
+ if (field.type() != fieldReplacement) {
+ hasChanged = true;
+
newFields.add(Types.NestedField.from(field).ofType(fieldReplacement).build());
+ } else {
+ newFields.add(field);
+ }
+ }
+
+ if (hasChanged) {
+ return Types.StructType.of(newFields);
+ }
+
+ return struct;
+ }
+
+ @Override
+ public Type field(Types.NestedField field, Type fieldResult) {
+ return replacementsById.getOrDefault(field.fieldId(), fieldResult);
+ }
+
+ @Override
+ public Type list(Types.ListType list, Type elementResult) {
+ Type elementReplacement = replacementsById.getOrDefault(list.elementId(),
elementResult);
+ if (list.elementType() != elementReplacement) {
+ if (list.isElementRequired()) {
+ return Types.ListType.ofRequired(list.elementId(), elementReplacement);
+ } else {
+ return Types.ListType.ofOptional(list.elementId(), elementReplacement);
+ }
+ }
+
+ return list;
+ }
+
+ @Override
+ public Type map(Types.MapType map, Type keyResult, Type valueResult) {
+ Type keyReplacement = replacementsById.getOrDefault(map.keyId(),
keyResult);
+ Type valueReplacement = replacementsById.getOrDefault(map.valueId(),
valueResult);
+ if (map.keyType() != keyReplacement || map.valueType() !=
valueReplacement) {
+ if (map.isValueRequired()) {
+ return Types.MapType.ofRequired(
+ map.keyId(), map.valueId(), keyReplacement, valueReplacement);
+ } else {
+ return Types.MapType.ofOptional(
+ map.keyId(), map.valueId(), keyReplacement, valueReplacement);
+ }
+ }
+
+ return map;
+ }
+
+ @Override
+ public Type variant(Types.VariantType variant) {
+ return variant;
+ }
+
+ @Override
+ public Type primitive(Type.PrimitiveType primitive) {
+ return primitive;
+ }
+}
diff --git a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java
b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java
index a3bee3e3d8..8e39ae7a43 100644
--- a/api/src/main/java/org/apache/iceberg/types/TypeUtil.java
+++ b/api/src/main/java/org/apache/iceberg/types/TypeUtil.java
@@ -157,6 +157,24 @@ public class TypeUtil {
return project(schema, projectedIds);
}
+ /**
+ * Returns a copy of the schema with the type of each field in {@code
replacementsById} replaced
+ * by its mapped type. Fields not in the map are unchanged.
+ *
+ * @param schema a schema
+ * @param replacementsById a map from field ID to the type that should
replace the field's type
+ * @return a schema with the replaced field types
+ */
+ public static Schema replaceFieldTypes(Schema schema, Map<Integer, Type>
replacementsById) {
+ Types.StructType struct = visit(schema, new
ReplaceTypeById(replacementsById)).asStructType();
+ if (struct.equals(schema.asStruct())) {
+ return schema;
+ }
+
+ return new Schema(
+ schema.schemaId(), struct.fields(), schema.getAliases(),
schema.identifierFieldIds());
+ }
+
public static Schema join(Schema left, Schema right) {
List<Types.NestedField> joinedColumns = Lists.newArrayList(left.columns());
for (Types.NestedField rightColumn : right.columns()) {
diff --git a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
index b7da4b3108..d540d23961 100644
--- a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
+++ b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java
@@ -29,6 +29,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;
import org.apache.iceberg.Schema;
import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.relocated.com.google.common.collect.Sets;
import org.apache.iceberg.types.Types.IntegerType;
@@ -1037,4 +1038,133 @@ public class TestTypeUtil {
.containsEntry(24, "addresses_value") // the leaf takes precedence
.hasSize(22);
}
+
+ @Test
+ public void testReplaceFieldTypes() {
+ Types.StructType replacement = Types.StructType.of(required(10, "x",
IntegerType.get()));
+ Schema schema =
+ new Schema(
+ required(1, "id", IntegerType.get()),
+ required(2, "s", Types.StructType.of(required(3, "a",
Types.LongType.get()))));
+
+ Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(2,
replacement));
+
+ assertThat(result.findField(1).type()).isEqualTo(IntegerType.get());
+ assertThat(result.findField(2).type()).isEqualTo(replacement);
+ }
+
+ @Test
+ public void testReplaceFieldTypesPrimitive() {
+ Schema schema =
+ new Schema(required(1, "id", IntegerType.get()), required(2, "count",
IntegerType.get()));
+
+ Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(2,
Types.LongType.get()));
+
+ assertThat(result.findField(1).type()).isEqualTo(IntegerType.get());
+ assertThat(result.findField(2).type()).isEqualTo(Types.LongType.get());
+ }
+
+ @Test
+ public void testReplaceFieldTypesListElement() {
+ Schema schema =
+ new Schema(required(1, "list", Types.ListType.ofRequired(2,
Types.StructType.of())));
+
+ Schema result =
+ TypeUtil.replaceFieldTypes(
+ schema, ImmutableMap.of(2, Types.StructType.of(required(3, "x",
IntegerType.get()))));
+
+ Types.ListType list = (Types.ListType) result.findField(1).type();
+
assertThat(list.elementType().asStructType().field(3).name()).isEqualTo("x");
+ assertThat(list.isElementRequired()).isTrue();
+ }
+
+ @Test
+ public void testReplaceFieldTypesMapKeyAndValue() {
+ Schema schema =
+ new Schema(
+ required(1, "m", Types.MapType.ofRequired(2, 3, IntegerType.get(),
IntegerType.get())));
+
+ Schema result =
+ TypeUtil.replaceFieldTypes(
+ schema, ImmutableMap.of(2, Types.LongType.get(), 3,
Types.StringType.get()));
+
+ Types.MapType map = (Types.MapType) result.findField(1).type();
+ assertThat(map.keyType()).isEqualTo(Types.LongType.get());
+ assertThat(map.valueType()).isEqualTo(Types.StringType.get());
+ assertThat(map.isValueRequired()).isTrue();
+ }
+
+ @Test
+ public void testReplaceFieldTypesPreservesDefaultsOnCompatibleType() {
+ Types.NestedField field =
+ Types.NestedField.optional("count")
+ .withId(1)
+ .ofType(IntegerType.get())
+ .withInitialDefault(Literal.of(5))
+ .withWriteDefault(Literal.of(7))
+ .build();
+ Schema schema = new Schema(field);
+
+ Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(1,
Types.LongType.get()));
+
+ Types.NestedField replaced = result.findField(1);
+ assertThat(replaced.type()).isEqualTo(Types.LongType.get());
+ assertThat(replaced.initialDefault()).isEqualTo(5L);
+ assertThat(replaced.writeDefault()).isEqualTo(7L);
+ }
+
+ @Test
+ public void testReplaceFieldTypesRejectsDefaultIncompatibleWithType() {
+ Types.NestedField field =
+ Types.NestedField.optional("count")
+ .withId(1)
+ .ofType(IntegerType.get())
+ .withInitialDefault(Literal.of(5))
+ .build();
+ Schema schema = new Schema(field);
+
+ assertThatThrownBy(
+ () -> TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(1,
Types.StringType.get())))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot cast default value to string");
+ }
+
+ @Test
+ public void testReplaceFieldTypesPreservesIdentifierField() {
+ Schema schema =
+ new Schema(
+ Lists.newArrayList(
+ required(1, "id", IntegerType.get()), required(2, "data",
Types.StringType.get())),
+ Sets.newHashSet(1));
+
+ Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(1,
Types.LongType.get()));
+
+ assertThat(result.findField(1).type()).isEqualTo(Types.LongType.get());
+ assertThat(result.identifierFieldIds()).containsExactly(1);
+ }
+
+ @Test
+ public void testReplaceFieldTypesRejectsNonPrimitiveIdentifierField() {
+ Schema schema =
+ new Schema(
+ Lists.newArrayList(
+ required(1, "id", IntegerType.get()), required(2, "data",
Types.StringType.get())),
+ Sets.newHashSet(1));
+
+ // an identifier field must be primitive, so replacing it with a struct is
rejected
+ assertThatThrownBy(
+ () ->
+ TypeUtil.replaceFieldTypes(
+ schema,
+ ImmutableMap.of(1, Types.StructType.of(required(3, "x",
IntegerType.get())))))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("not a primitive type field");
+ }
+
+ @Test
+ public void testReplaceFieldTypesNoMatchReturnsSameSchema() {
+ Schema schema = new Schema(required(1, "id", IntegerType.get()));
+ Schema result = TypeUtil.replaceFieldTypes(schema, ImmutableMap.of(99,
Types.LongType.get()));
+ assertThat(result).isSameAs(schema);
+ }
}
diff --git a/core/src/main/java/org/apache/iceberg/Partitioning.java
b/core/src/main/java/org/apache/iceberg/Partitioning.java
index c708d39f52..66a681f884 100644
--- a/core/src/main/java/org/apache/iceberg/Partitioning.java
+++ b/core/src/main/java/org/apache/iceberg/Partitioning.java
@@ -243,6 +243,17 @@ public class Partitioning {
"table partition", specs, allActiveFieldIds(table.schema(), specs));
}
+ /**
+ * Builds a unified partition type containing all partition fields from the
given specs, including
+ * fields whose source columns are no longer present in the table schema.
+ *
+ * @param specs the partition specs to unify
+ * @return the constructed unified partition type
+ */
+ static StructType unionPartitionTypes(Collection<PartitionSpec> specs) {
+ return buildPartitionProjectionType("union partition", specs,
allFieldIds(specs));
+ }
+
/**
* Checks if any of the specs in a table is partitioned.
*
@@ -347,6 +358,14 @@ public class Partitioning {
|| t2.equals(Transforms.alwaysNull());
}
+ // collects IDs of all partition fields used across specs
+ private static Set<Integer> allFieldIds(Collection<PartitionSpec> specs) {
+ return FluentIterable.from(specs)
+ .transformAndConcat(PartitionSpec::fields)
+ .transform(PartitionField::fieldId)
+ .toSet();
+ }
+
// collects IDs of all partition field used across specs that are in the
current schema
private static Set<Integer> allActiveFieldIds(Schema schema,
Collection<PartitionSpec> specs) {
return FluentIterable.from(specs)
diff --git a/core/src/main/java/org/apache/iceberg/TrackedFile.java
b/core/src/main/java/org/apache/iceberg/TrackedFile.java
index 9aaeae48d6..1a457fb048 100644
--- a/core/src/main/java/org/apache/iceberg/TrackedFile.java
+++ b/core/src/main/java/org/apache/iceberg/TrackedFile.java
@@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.List;
import java.util.Set;
+import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
/** A file tracked by a manifest. */
@@ -95,9 +96,14 @@ interface TrackedFile {
Types.ListType.ofRequired(136, Types.IntegerType.get()),
"Field ids used to determine row equality in equality delete files");
- static Types.StructType schemaWithContentStats(
- Types.StructType partitionType, Types.StructType contentStatsType) {
- return Types.StructType.of(
+ /**
+ * Returns the schema for the given partition and content stats types.
+ *
+ * <p>The partition and content stats fields use {@link Types.UnknownType}
when their types have
+ * no fields, so that they are not stored in manifest files.
+ */
+ static Schema schema(Types.StructType partitionType, Types.StructType
contentStatsType) {
+ return new Schema(
TRACKING,
CONTENT_TYPE,
FORMAT_VERSION,
@@ -106,9 +112,13 @@ interface TrackedFile {
RECORD_COUNT,
FILE_SIZE_IN_BYTES,
SPEC_ID,
- Types.NestedField.optional(PARTITION_ID, PARTITION_NAME,
partitionType, PARTITION_DOC),
Types.NestedField.optional(
- CONTENT_STATS_ID, CONTENT_STATS_NAME, contentStatsType,
CONTENT_STATS_DOC),
+ PARTITION_ID, PARTITION_NAME, typeOrUnknown(partitionType),
PARTITION_DOC),
+ Types.NestedField.optional(
+ CONTENT_STATS_ID,
+ CONTENT_STATS_NAME,
+ typeOrUnknown(contentStatsType),
+ CONTENT_STATS_DOC),
SORT_ORDER_ID,
DELETION_VECTOR,
MANIFEST_INFO,
@@ -117,6 +127,10 @@ interface TrackedFile {
EQUALITY_IDS);
}
+ private static Type typeOrUnknown(Types.StructType structType) {
+ return structType.fields().isEmpty() ? Types.UnknownType.get() :
structType;
+ }
+
/** Returns the tracking information for this entry. */
Tracking tracking();
diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java
b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java
index 3ed764ff17..f5ce03c7eb 100644
--- a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java
+++ b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java
@@ -83,10 +83,11 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
/** Used by internal readers to instantiate this class with a projection
schema. */
TrackedFileStruct(Types.StructType projection) {
super(BASE_TYPE, projection);
- // partition type may be null if the field was not projected
+ // partition type may be null if the field was not projected, or unknown
for unpartitioned
+ // manifests
Type partType = projection.fieldType("partition");
- if (partType != null) {
- this.partitionData = new
PartitionData(partType.asNestedType().asStructType());
+ if (partType != null && partType.isStructType()) {
+ this.partitionData = new PartitionData(partType.asStructType());
}
}
@@ -101,10 +102,10 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
int formatVersion,
String location,
FileFormat fileFormat,
- PartitionData partition,
long recordCount,
long fileSizeInBytes,
Integer specId,
+ PartitionData partition,
ContentStats contentStats,
Integer sortOrderId,
DeletionVector deletionVector,
@@ -120,9 +121,8 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
this.fileFormat = fileFormat;
this.recordCount = recordCount;
this.fileSizeInBytes = fileSizeInBytes;
- this.partitionData = partition;
-
this.specId = specId;
+ this.partitionData = partition;
this.contentStats = contentStats;
this.sortOrderId = sortOrderId;
this.deletionVector = deletionVector;
diff --git a/core/src/main/java/org/apache/iceberg/TrackingStruct.java
b/core/src/main/java/org/apache/iceberg/TrackingStruct.java
index 8ae4b7e4ce..283e0a51ff 100644
--- a/core/src/main/java/org/apache/iceberg/TrackingStruct.java
+++ b/core/src/main/java/org/apache/iceberg/TrackingStruct.java
@@ -30,7 +30,8 @@ import org.apache.iceberg.util.ByteBuffers;
/** Mutable {@link StructLike} implementation of {@link Tracking}. */
class TrackingStruct extends SupportsIndexProjection implements Tracking,
Serializable {
- private static final Types.StructType BASE_TYPE =
+ // Package-private only for read projection.
+ static final Types.StructType BASE_TYPE =
Types.StructType.of(
Tracking.STATUS,
Tracking.SNAPSHOT_ID,
@@ -42,6 +43,17 @@ class TrackingStruct extends SupportsIndexProjection
implements Tracking, Serial
Tracking.REPLACED_POSITIONS,
MetadataColumns.ROW_POSITION);
+ // tracking fields read on the scan path; row_position backs manifestPos.
+ // Package-private only for read projection.
+ static final Types.StructType SCAN_TYPE =
+ Types.StructType.of(
+ Tracking.STATUS,
+ Tracking.SNAPSHOT_ID,
+ Tracking.SEQUENCE_NUMBER,
+ Tracking.FILE_SEQUENCE_NUMBER,
+ Tracking.FIRST_ROW_ID,
+ MetadataColumns.ROW_POSITION);
+
private EntryStatus status = null;
private Long snapshotId = null;
private Long dataSequenceNumber = null;
diff --git a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
new file mode 100644
index 0000000000..a823454845
--- /dev/null
+++ b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java
@@ -0,0 +1,300 @@
+/*
+ * 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.iceberg;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.expressions.Projections;
+import org.apache.iceberg.io.CloseableGroup;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.CloseableIterator;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.metrics.ScanMetrics;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.util.StructProjection;
+
+/** Reader that reads a v4+ manifest file as {@link TrackedFile}s. */
+class V4ManifestReader extends CloseableGroup implements
CloseableIterable<TrackedFile> {
+ private final InputFile file;
+ private final Schema readSchema;
+ private final boolean includeAll;
+ private final ScanMetrics scanMetrics;
+
+ // partition filters keyed by spec ID; empty when no partition filter applies
+ private final Map<Integer, Pair<Evaluator, StructProjection>>
partitionFilters;
+
+ private V4ManifestReader(
+ InputFile file,
+ Schema readSchema,
+ Map<Integer, Pair<Evaluator, StructProjection>> partitionFilters,
+ boolean includeAll,
+ ScanMetrics scanMetrics) {
+ this.file = file;
+ this.readSchema = readSchema;
+ this.partitionFilters = partitionFilters;
+ this.includeAll = includeAll;
+ this.scanMetrics = scanMetrics;
+ }
+
+ static Builder builder(InputFile file, Map<Integer, PartitionSpec>
specsById) {
+ return new Builder(file, specsById);
+ }
+
+ /** Returns copies of the tracked files that match this reader's configured
filters. */
+ @Override
+ public CloseableIterator<TrackedFile> iterator() {
+ CloseableIterable<TrackedFile> entries =
CloseableIterable.transform(open(), this::prepare);
+ if (!partitionFilters.isEmpty()) {
+ // manifests have no partition, so the partition filter cannot apply to
them
+ entries =
+ CloseableIterable.filter(entries, entry -> isManifest(entry) ||
matchesPartition(entry));
+ }
+
+ if (!includeAll) {
+ entries = CloseableIterable.filter(entries, entry ->
entry.tracking().isLive());
+ }
+
+ return CloseableIterable.transform(entries, TrackedFile::copy).iterator();
+ }
+
+ private boolean matchesPartition(TrackedFile trackedFile) {
+ Integer specId = trackedFile.specId();
+ if (specId == null) {
+ // a file without a spec is not partitioned and may match the filter
+ return true;
+ }
+
+ Pair<Evaluator, StructProjection> partitionFilter =
partitionFilters.get(specId);
+ if (partitionFilter == null) {
+ // the row filter does not project to a partition filter for this spec
+ return true;
+ }
+
+ Evaluator evaluator = partitionFilter.first();
+ StructProjection projection = partitionFilter.second();
+ boolean matches = evaluator.eval(projection.wrap(trackedFile.partition()));
+ if (!matches) {
+ incrementSkipCount(trackedFile.contentType());
+ }
+
+ return matches;
+ }
+
+ private void incrementSkipCount(FileContent content) {
+ switch (content) {
+ case DATA -> scanMetrics.skippedDataFiles().increment();
+ case EQUALITY_DELETES -> scanMetrics.skippedDeleteFiles().increment();
+ case DATA_MANIFEST -> scanMetrics.skippedDataManifests().increment();
+ case DELETE_MANIFEST -> scanMetrics.skippedDeleteManifests().increment();
+ default -> throw new UnsupportedOperationException("Unsupported content
type: " + content);
+ }
+ }
+
+ private CloseableIterable<TrackedFile> open() {
+ FileFormat format = FileFormat.fromFileName(file.location());
+ Preconditions.checkArgument(
+ format != null, "Cannot determine format of manifest: %s",
file.location());
+
+ CloseableIterable<TrackedFile> reader =
+ InternalData.read(format, file)
+ .project(readSchema)
+ .setRootType(TrackedFileStruct.class)
+ .setCustomType(TrackedFile.TRACKING.fieldId(),
TrackingStruct.class)
+ .setCustomType(TrackedFile.DELETION_VECTOR.fieldId(),
DeletionVectorStruct.class)
+ .setCustomType(TrackedFile.MANIFEST_INFO.fieldId(),
ManifestInfoStruct.class)
+ .setCustomType(TrackedFile.PARTITION_ID, PartitionData.class)
+ .reuseContainers()
+ .build();
+ addCloseable(reader);
+ return reader;
+ }
+
+ private TrackedFile prepare(TrackedFile trackedFile) {
+ Tracking tracking = trackedFile.tracking();
+ // manifestLocation is not stored in the manifest; the reader fills it in
+ if (tracking instanceof TrackingStruct) {
+ ((TrackingStruct) tracking).setManifestLocation(file.location());
+ }
+
+ return trackedFile;
+ }
+
+ private static boolean isManifest(TrackedFile trackedFile) {
+ FileContent content = trackedFile.contentType();
+ return content == FileContent.DATA_MANIFEST || content ==
FileContent.DELETE_MANIFEST;
+ }
+
+ static class Builder {
+ private final InputFile file;
+ private final Types.StructType unionPartitionType;
+ private final Map<Integer, PartitionSpec> specsById;
+ private final Schema fullSchema;
+ private Expression rowFilter = Expressions.alwaysTrue();
+ private boolean caseSensitive = true;
+ private boolean includeAll = false;
+ private boolean scanPlanning = false;
+ private Collection<String> columns = null;
+ private Schema requestedProjection = null;
+ private ScanMetrics scanMetrics = ScanMetrics.noop();
+
+ private Builder(InputFile file, Map<Integer, PartitionSpec> specsById) {
+ this.file = file;
+ this.specsById = specsById;
+ this.unionPartitionType =
Partitioning.unionPartitionTypes(specsById.values());
+ Schema base = TrackedFile.schema(unionPartitionType,
Types.StructType.of());
+ // the read schema carries row_position (via BASE_TYPE) so the reader
can fill manifestPos
+ this.fullSchema =
+ TypeUtil.replaceFieldTypes(
+ base, ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.BASE_TYPE));
+ }
+
+ /** Sets a filter; files that cannot match the expression are skipped. */
+ Builder filter(Expression expr) {
+ Preconditions.checkArgument(expr != null, "Invalid filter: null");
+ this.rowFilter = expr;
+ return this;
+ }
+
+ Builder caseSensitive(boolean isCaseSensitive) {
+ this.caseSensitive = isCaseSensitive;
+ return this;
+ }
+
+ /** Returns all entries without filtering by {@link Tracking#isLive()
liveness}. */
+ Builder includeAll() {
+ this.includeAll = true;
+ return this;
+ }
+
+ /** Configures the reader to select the minimal fields needed for scan
planning. */
+ Builder forScanPlanning() {
+ Preconditions.checkState(
+ columns == null && requestedProjection == null,
+ "Cannot use forScanPlanning() with select(Collection<String>) or
project(Schema)");
+ this.scanPlanning = true;
+ return this;
+ }
+
+ /** Selects columns to read by name; fields needed by the reader are
always read. */
+ Builder select(String... newColumns) {
+ return select(Arrays.asList(newColumns));
+ }
+
+ /** Selects columns to read by name; fields needed by the reader are
always read. */
+ Builder select(Collection<String> newColumns) {
+ Preconditions.checkArgument(newColumns != null, "Invalid columns: null");
+ Preconditions.checkState(
+ !scanPlanning, "Cannot use select(Collection<String>) with
forScanPlanning()");
+ Preconditions.checkState(
+ requestedProjection == null,
+ "Cannot use select(Collection<String>) with project(Schema)");
+ this.columns = newColumns;
+ return this;
+ }
+
+ /** Sets the exact schema to read; used in place of {@link
#select(Collection)}. */
+ Builder project(Schema newProjection) {
+ Preconditions.checkState(!scanPlanning, "Cannot use project(Schema) with
forScanPlanning()");
+ Preconditions.checkState(
+ columns == null, "Cannot use project(Schema) with
select(Collection<String>)");
+ this.requestedProjection = newProjection;
+ return this;
+ }
+
+ Builder scanMetrics(ScanMetrics newScanMetrics) {
+ Preconditions.checkArgument(newScanMetrics != null, "Invalid scan
metrics: null");
+ this.scanMetrics = newScanMetrics;
+ return this;
+ }
+
+ V4ManifestReader build() {
+ Map<Integer, Pair<Evaluator, StructProjection>> partitionFilters =
Maps.newHashMap();
+ if (rowFilter != Expressions.alwaysTrue() &&
!unionPartitionType.fields().isEmpty()) {
+ for (PartitionSpec spec : specsById.values()) {
+ Expression partFilter = Projections.inclusive(spec,
caseSensitive).project(rowFilter);
+ if (partFilter != Expressions.alwaysTrue()) {
+ Evaluator evaluator = new Evaluator(spec.partitionType(),
partFilter, caseSensitive);
+ StructProjection projection =
+ StructProjection.create(unionPartitionType,
spec.partitionType());
+ partitionFilters.put(spec.specId(), Pair.of(evaluator,
projection));
+ }
+ }
+ }
+
+ boolean hasPartitionFilter = !partitionFilters.isEmpty();
+ return new V4ManifestReader(
+ file, readSchema(hasPartitionFilter), partitionFilters, includeAll,
scanMetrics);
+ }
+
+ private Schema readSchema(boolean hasPartitionFilter) {
+ if (scanPlanning) {
+ // scan planning does not read the change-tracking fields omitted by
SCAN_TYPE
+ return TypeUtil.replaceFieldTypes(
+ fullSchema, ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.SCAN_TYPE));
+ }
+
+ if (columns != null) {
+ Schema selected =
+ caseSensitive ? fullSchema.select(columns) :
fullSchema.caseInsensitiveSelect(columns);
+ return addRequiredColumns(selected, hasPartitionFilter);
+ }
+
+ if (requestedProjection != null) {
+ return addRequiredColumns(requestedProjection, hasPartitionFilter);
+ }
+
+ return fullSchema;
+ }
+
+ private Schema addRequiredColumns(Schema projection, boolean
hasPartitionFilter) {
+ Set<Integer> projectedIds =
Sets.newHashSet(TypeUtil.getProjectedIds(projection));
+
+ // fields the reader consumes internally: status for liveness filtering,
row_position for
+ // manifestPos, and content type to distinguish entry kinds
+ projectedIds.add(Tracking.STATUS.fieldId());
+ projectedIds.add(MetadataColumns.ROW_POSITION.fieldId());
+ projectedIds.add(TrackedFile.CONTENT_TYPE.fieldId());
+ if (rowFilter != Expressions.alwaysTrue()) {
+ // record_count is read when evaluating a filter against file metrics
+ projectedIds.add(TrackedFile.RECORD_COUNT.fieldId());
+ }
+
+ // add the partition tuple only when it is needed to evaluate a
partition filter
+ if (hasPartitionFilter) {
+ projectedIds.add(TrackedFile.SPEC_ID.fieldId());
+ projectedIds.add(TrackedFile.PARTITION_ID);
+ projectedIds.addAll(TypeUtil.getProjectedIds(unionPartitionType));
+ }
+
+ // project instead of select to preserve narrow struct projections from
the caller
+ return TypeUtil.project(fullSchema, projectedIds);
+ }
+ }
+}
diff --git a/core/src/test/java/org/apache/iceberg/TestPartitioning.java
b/core/src/test/java/org/apache/iceberg/TestPartitioning.java
index eaabf5d3ed..4c98d66c10 100644
--- a/core/src/test/java/org/apache/iceberg/TestPartitioning.java
+++ b/core/src/test/java/org/apache/iceberg/TestPartitioning.java
@@ -211,6 +211,27 @@ public class TestPartitioning {
assertThat(actualType).isEqualTo(StructType.of());
}
+ @Test
+ public void testUnionPartitionTypesRetainsDroppedSourceFields() {
+ TestTables.TestTable table =
+ TestTables.create(
+ tableDir, "test", SCHEMA, BY_DATA_CATEGORY_BUCKET_SPEC,
V2_FORMAT_VERSION);
+
+ table.updateSpec().removeField("category_bucket").commit();
+ table.updateSchema().deleteColumn("category").commit();
+ table.updateSpec().removeField("data").commit();
+ table.updateSchema().deleteColumn("data").commit();
+
+ // fields with dropped source columns are retained to preserve partition
tuple equality;
+ // bucket keeps its fixed result type while the identity result type falls
back to unknown
+ StructType actualType =
Partitioning.unionPartitionTypes(table.specs().values());
+ assertThat(actualType)
+ .isEqualTo(
+ StructType.of(
+ NestedField.optional(1000, "data", Types.UnknownType.get()),
+ NestedField.optional(1001, "category_bucket",
Types.IntegerType.get())));
+ }
+
@Test
public void testPartitionTypeWithDroppedSourceColumn() {
TestTables.TestTable table =
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFile.java
b/core/src/test/java/org/apache/iceberg/TestTrackedFile.java
index 5840a1cd5f..9a32243cc1 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackedFile.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackedFile.java
@@ -39,8 +39,8 @@ public class TestTrackedFile {
PartitionSpec.builderFor(TABLE_SCHEMA).identity("id").build().partitionType();
@Test
- public void schemaWithContentStatsFieldOrder() {
- Types.StructType type = TrackedFile.schemaWithContentStats(PARTITION_TYPE,
CONTENT_STATS_TYPE);
+ public void schemaFieldOrder() {
+ Types.StructType type = TrackedFile.schema(PARTITION_TYPE,
CONTENT_STATS_TYPE).asStruct();
List<Types.NestedField> fields = type.fields();
assertThat(fields)
@@ -65,8 +65,8 @@ public class TestTrackedFile {
}
@Test
- public void schemaWithContentStatsFieldIds() {
- Types.StructType type = TrackedFile.schemaWithContentStats(PARTITION_TYPE,
CONTENT_STATS_TYPE);
+ public void schemaFieldIds() {
+ Types.StructType type = TrackedFile.schema(PARTITION_TYPE,
CONTENT_STATS_TYPE).asStruct();
List<Types.NestedField> fields = type.fields();
assertThat(fields)
@@ -76,8 +76,8 @@ public class TestTrackedFile {
}
@Test
- public void schemaWithContentStatsUsesProvidedType() {
- Types.StructType type = TrackedFile.schemaWithContentStats(PARTITION_TYPE,
CONTENT_STATS_TYPE);
+ public void schemaUsesProvidedType() {
+ Types.StructType type = TrackedFile.schema(PARTITION_TYPE,
CONTENT_STATS_TYPE).asStruct();
Types.NestedField contentStatsField =
type.field(TrackedFile.CONTENT_STATS_ID);
Types.NestedField partitionField = type.field(TrackedFile.PARTITION_ID);
@@ -86,7 +86,7 @@ public class TestTrackedFile {
}
@Test
- public void schemaWithContentStatsReflectsInput() {
+ public void schemaReflectsInput() {
Schema smallSchema = new Schema(optional(1, "id",
Types.IntegerType.get()));
Schema largeSchema =
new Schema(
@@ -97,8 +97,8 @@ public class TestTrackedFile {
Types.StructType smallStats = StatsUtil.statsReadSchema(smallSchema,
ImmutableList.of(1));
Types.StructType largeStats = StatsUtil.statsReadSchema(largeSchema,
ImmutableList.of(1, 3));
- Types.StructType smallType =
TrackedFile.schemaWithContentStats(PARTITION_TYPE, smallStats);
- Types.StructType largeType =
TrackedFile.schemaWithContentStats(PARTITION_TYPE, largeStats);
+ Types.StructType smallType = TrackedFile.schema(PARTITION_TYPE,
smallStats).asStruct();
+ Types.StructType largeType = TrackedFile.schema(PARTITION_TYPE,
largeStats).asStruct();
Types.StructType smallResult =
smallType.field(TrackedFile.CONTENT_STATS_ID).type().asStructType();
@@ -110,12 +110,27 @@ public class TestTrackedFile {
}
@Test
- public void schemaWithContentStatsPartitionIsOptional() {
- Types.StructType type = TrackedFile.schemaWithContentStats(PARTITION_TYPE,
CONTENT_STATS_TYPE);
+ public void schemaPartitionIsOptional() {
+ Types.StructType type = TrackedFile.schema(PARTITION_TYPE,
CONTENT_STATS_TYPE).asStruct();
Types.NestedField partitionField = type.field(TrackedFile.PARTITION_ID);
assertThat(partitionField.isOptional()).isTrue();
assertThat(partitionField.name()).isEqualTo(TrackedFile.PARTITION_NAME);
assertThat(partitionField.doc()).isEqualTo(TrackedFile.PARTITION_DOC);
}
+
+ @Test
+ public void schemaUsesUnknownForEmptyStructs() {
+ Types.StructType type =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).asStruct();
+
+
assertThat(type.field(TrackedFile.PARTITION_ID).type()).isEqualTo(Types.UnknownType.get());
+
assertThat(type.field(TrackedFile.CONTENT_STATS_ID).type()).isEqualTo(Types.UnknownType.get());
+
+ Types.StructType partitionedType =
+ TrackedFile.schema(PARTITION_TYPE, Types.StructType.of()).asStruct();
+
assertThat(partitionedType.field(TrackedFile.PARTITION_ID).type()).isEqualTo(PARTITION_TYPE);
+ assertThat(partitionedType.field(TrackedFile.CONTENT_STATS_ID).type())
+ .isEqualTo(Types.UnknownType.get());
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
index 912f36d860..1992390506 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java
@@ -103,10 +103,10 @@ class TestTrackedFileAdapters {
FORMAT_VERSION_V4,
DATA_FILE_LOCATION,
FileFormat.PARQUET,
- PARTITION,
100L,
1024L,
PARTITIONED_SPEC_ID,
+ PARTITION,
CONTENT_STATS,
3,
null,
@@ -179,10 +179,10 @@ class TestTrackedFileAdapters {
FORMAT_VERSION_V4,
"s3://bucket/eq-delete.avro",
FileFormat.AVRO,
- PARTITION,
50L,
512L,
PARTITIONED_SPEC_ID,
+ PARTITION,
CONTENT_STATS,
5,
null,
@@ -264,10 +264,10 @@ class TestTrackedFileAdapters {
FORMAT_VERSION_V4,
DATA_FILE_LOCATION,
FileFormat.PARQUET,
- PARTITION,
100L,
1024L,
PARTITIONED_SPEC_ID,
+ PARTITION,
null,
null,
dv,
@@ -361,12 +361,12 @@ class TestTrackedFileAdapters {
0,
null,
null,
- null,
0L,
0L,
null,
null,
null,
+ null,
deletionVector(),
null,
null,
@@ -415,7 +415,6 @@ class TestTrackedFileAdapters {
0,
null,
null,
- null,
0L,
0L,
99,
@@ -425,6 +424,7 @@ class TestTrackedFileAdapters {
null,
null,
null,
+ null,
null);
assertThatThrownBy(() -> TrackedFileAdapters.asDataFile(file,
ImmutableMap.of()))
@@ -441,7 +441,6 @@ class TestTrackedFileAdapters {
0,
null,
null,
- null,
0L,
0L,
PARTITIONED_SPEC_ID,
@@ -451,6 +450,7 @@ class TestTrackedFileAdapters {
null,
null,
null,
+ null,
null);
int mismatchedSpecId = PARTITIONED_SPEC_ID + 1;
PartitionSpec mismatched =
@@ -496,7 +496,6 @@ class TestTrackedFileAdapters {
FORMAT_VERSION_V4,
DATA_FILE_LOCATION,
FileFormat.PARQUET,
- null,
1L,
1L,
null,
@@ -506,6 +505,7 @@ class TestTrackedFileAdapters {
null,
null,
null,
+ null,
null);
}
diff --git a/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java
b/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java
index d888c6a7dc..14265c91f6 100644
--- a/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java
+++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileStruct.java
@@ -33,13 +33,9 @@ import org.mockito.Mockito;
class TestTrackedFileStruct {
private static final int FORMAT_VERSION_V4 = 4;
- private static final Types.StructType PARTITION_TYPE =
- Types.StructType.of(
- Types.NestedField.optional(1000, "id_bucket",
Types.IntegerType.get()),
- Types.NestedField.optional(1001, "category",
Types.StringType.get()));
- private static final List<Types.NestedField> FIELDS =
- TrackedFile.schemaWithContentStats(PARTITION_TYPE,
Types.StructType.of()).fields();
+ private static final List<Types.NestedField> DEFAULT_FIELDS =
+ TrackedFile.schema(Types.StructType.of(),
Types.StructType.of()).asStruct().fields();
private static final Tracking TRACKING = Mockito.mock(Tracking.class);
private static final Tracking TRACKING_COPY = Mockito.mock(Tracking.class);
@@ -73,10 +69,10 @@ class TestTrackedFileStruct {
FORMAT_VERSION_V4,
"s3://bucket/data/00000-0-file.parquet",
FileFormat.PARQUET,
- PARTITION,
50L,
512L,
1,
+ PARTITION,
CONTENT_STATS,
5,
DELETION_VECTOR,
@@ -150,10 +146,10 @@ class TestTrackedFileStruct {
FORMAT_VERSION_V4,
"s3://bucket/data/00000-0-file.parquet",
FileFormat.PARQUET,
- PARTITION,
50L,
512L,
1,
+ PARTITION,
CONTENT_STATS,
5,
DELETION_VECTOR,
@@ -191,10 +187,10 @@ class TestTrackedFileStruct {
FORMAT_VERSION_V4,
"s3://bucket/data/00000-0-file.parquet",
FileFormat.PARQUET,
- PARTITION,
50L,
512L,
1,
+ PARTITION,
CONTENT_STATS,
5,
DELETION_VECTOR,
@@ -240,10 +236,10 @@ class TestTrackedFileStruct {
FORMAT_VERSION_V4,
"s3://bucket/data/00000-0-file.parquet",
FileFormat.PARQUET,
- PARTITION,
50L,
512L,
1,
+ PARTITION,
stats,
5,
DELETION_VECTOR,
@@ -287,10 +283,10 @@ class TestTrackedFileStruct {
FORMAT_VERSION_V4,
"s3://bucket/data/00000-0-file.parquet",
FileFormat.PARQUET,
- PARTITION,
50L,
512L,
1,
+ PARTITION,
stats,
5,
DELETION_VECTOR,
@@ -347,7 +343,7 @@ class TestTrackedFileStruct {
@Test
void structLikeSize() {
TrackedFileStruct file = new TrackedFileStruct();
- assertThat(file.size()).isEqualTo(FIELDS.size());
+ assertThat(file.size()).isEqualTo(DEFAULT_FIELDS.size());
}
@ParameterizedTest
@@ -360,10 +356,10 @@ class TestTrackedFileStruct {
FORMAT_VERSION_V4,
"s3://bucket/data/file.parquet",
FileFormat.PARQUET,
- null, // PartitionData has its own serialization tests
100L,
1024L,
7,
+ null, // PartitionData has its own serialization tests
null,
1,
null, // DeletionVector has its own serialization tests
@@ -392,8 +388,8 @@ class TestTrackedFileStruct {
}
private static int pos(String fieldName) {
- for (int i = 0; i < FIELDS.size(); i += 1) {
- if (FIELDS.get(i).name().equals(fieldName)) {
+ for (int i = 0; i < DEFAULT_FIELDS.size(); i += 1) {
+ if (DEFAULT_FIELDS.get(i).name().equals(fieldName)) {
return i;
}
}
diff --git a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
new file mode 100644
index 0000000000..c8a5cfd61a
--- /dev/null
+++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java
@@ -0,0 +1,887 @@
+/*
+ * 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.iceberg;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.file.Path;
+import java.util.Collection;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.stream.Stream;
+import org.apache.iceberg.exceptions.ValidationException;
+import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.FileIO;
+import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.metrics.DefaultMetricsContext;
+import org.apache.iceberg.metrics.ScanMetrics;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.transforms.Transforms;
+import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Named;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.FieldSource;
+import org.junit.jupiter.params.provider.MethodSource;
+
+class TestV4ManifestReader {
+ private static final long SNAPSHOT_ID = 42L;
+ private static final int FORMAT_VERSION_V4 = 4;
+ private static final long RECORD_COUNT = 100L;
+ private static final long FILE_SIZE_IN_BYTES = 1024L;
+ private static final DeletionVector DV =
+ DeletionVectorStruct.builder()
+ .location("s3://bucket/dv.puffin")
+ .offset(100L)
+ .sizeInBytes(50L)
+ .cardinality(5L)
+ .build();
+
+ private static final Schema TABLE_SCHEMA =
+ new Schema(
+ optional(1, "id", Types.IntegerType.get()), optional(2, "data",
Types.StringType.get()));
+ private static final PartitionSpec ID_PARTITIONING =
+ PartitionSpec.builderFor(TABLE_SCHEMA).identity("id").build();
+ private static final Types.StructType ID_PARTITION_TYPE =
ID_PARTITIONING.partitionType();
+ private static final Types.StructType EMPTY_PARTITION =
Types.StructType.of();
+ private static final PartitionData EMPTY_PARTITION_DATA = new
PartitionData(EMPTY_PARTITION);
+ private static final Map<Integer, PartitionSpec> ID_PARTITIONING_SPECS =
+ ImmutableMap.of(ID_PARTITIONING.specId(), ID_PARTITIONING);
+ private static final Map<Integer, PartitionSpec> UNPARTITIONED_SPECS =
+ ImmutableMap.of(PartitionSpec.unpartitioned().specId(),
PartitionSpec.unpartitioned());
+
+ private static final List<FileFormat> MANIFEST_FORMATS =
+ ImmutableList.of(FileFormat.AVRO, FileFormat.PARQUET);
+
+ // a data file whose tracking carries every inheritable and change-tracking
value set
+ private static final TrackedFile FILE_WITH_FULL_TRACKING =
+ new TrackedFileStruct(
+ new TrackingStruct(
+ EntryStatus.ADDED,
+ SNAPSHOT_ID,
+ 5L, // data sequence number
+ 6L, // file sequence number
+ 7L, // dv snapshot id
+ 8L, // first row id
+ new byte[] {1, 2}, // deleted positions
+ new byte[] {3, 4}), // replaced positions
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ "s3://bucket/file.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ 0,
+ EMPTY_PARTITION_DATA,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null);
+
+ // shared data files: FILE_A is in partition id=1, FILE_B in partition id=2
+ private static final TrackedFile FILE_A = dataFile("data-a.parquet",
partition(1));
+ private static final TrackedFile FILE_B = dataFile("data-b.parquet",
partition(2));
+ private static final TrackedFile EQ_DELETES_A =
deleteFile("eq-deletes-a.parquet", partition(1));
+ private static final TrackedFile EQ_DELETES_B =
deleteFile("eq-deletes-b.parquet", partition(2));
+ private static final TrackedFile DATA_MANIFEST_REF =
+ manifestRef(FileContent.DATA_MANIFEST, "data-leaf.parquet");
+ private static final TrackedFile DELETE_MANIFEST_REF =
+ manifestRef(FileContent.DELETE_MANIFEST, "delete-leaf.parquet");
+
+ @TempDir private Path tempDir;
+
+ private final FileIO fileIO = new TestTables.LocalFileIO();
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void readsWrittenFile(FileFormat format) throws IOException {
+ TrackedFile file =
+ new TrackedFileStruct(
+ addedTracking(),
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ "s3://bucket/data/file.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ ID_PARTITIONING.specId(),
+ partition(7),
+ null,
+ 1, // sort order id
+ DV,
+ null,
+ ByteBuffer.wrap(new byte[] {1, 2, 3}),
+ ImmutableList.of(50L, 100L),
+ null);
+
+ InputFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(file));
+
+ TrackedFile actual = Iterables.getOnlyElement(read(manifest,
ID_PARTITIONING_SPECS));
+
+ // compare with tracking reduced to status: the reader fills
status-independent tracking
+ // fields (row position, sequence numbers via inheritance) that the
written file does not have
+ Types.StructType comparisonType =
+ TypeUtil.replaceFieldTypes(
+ TrackedFile.schema(ID_PARTITION_TYPE, Types.StructType.of()),
+ ImmutableMap.of(
+ TrackedFile.TRACKING.fieldId(),
Types.StructType.of(Tracking.STATUS)))
+ .asStruct();
+ assertThat((StructLike) actual)
+ .usingComparator(Comparators.forType(comparisonType))
+ .isEqualTo(file);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void equalityDeleteRoundTrip(FileFormat format) throws IOException {
+ TrackedFile delete =
+ new TrackedFileStruct(
+ addedTracking(),
+ FileContent.EQUALITY_DELETES,
+ FORMAT_VERSION_V4,
+ "s3://bucket/eq-delete.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ 0,
+ EMPTY_PARTITION_DATA,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ ImmutableList.of(1, 2));
+
+ InputFile manifest = writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(delete));
+
+ TrackedFile actual = Iterables.getOnlyElement(read(manifest,
UNPARTITIONED_SPECS));
+ assertThat(actual.contentType()).isEqualTo(FileContent.EQUALITY_DELETES);
+ assertThat(actual.equalityIds()).containsExactly(1, 2);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void statusFiltering(FileFormat format) throws IOException {
+ List<TrackedFile> files =
+ ImmutableList.of(
+ fileWithStatus(EntryStatus.ADDED, "s3://bucket/added.parquet"),
+ fileWithStatus(EntryStatus.EXISTING,
"s3://bucket/existing.parquet"),
+ fileWithStatus(EntryStatus.MODIFIED,
"s3://bucket/modified.parquet"),
+ fileWithStatus(EntryStatus.DELETED, "s3://bucket/deleted.parquet"),
+ fileWithStatus(EntryStatus.REPLACED,
"s3://bucket/replaced.parquet"));
+
+ InputFile manifest = writeManifest(format, EMPTY_PARTITION, files);
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) {
+ assertThat(reader)
+ .extracting(file -> file.tracking().status())
+ .containsExactly(EntryStatus.ADDED, EntryStatus.EXISTING,
EntryStatus.MODIFIED);
+ }
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).includeAll().build()) {
+ assertThat(reader)
+ .extracting(file -> file.tracking().status())
+ .containsExactly(
+ EntryStatus.ADDED,
+ EntryStatus.EXISTING,
+ EntryStatus.MODIFIED,
+ EntryStatus.DELETED,
+ EntryStatus.REPLACED);
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void manifestLocationAndPosition(FileFormat format) throws
IOException {
+ List<TrackedFile> files =
+ ImmutableList.of(
+ dataFile("s3://bucket/a.parquet", EMPTY_PARTITION_DATA),
+ dataFile("s3://bucket/b.parquet", EMPTY_PARTITION_DATA),
+ dataFile("s3://bucket/c.parquet", EMPTY_PARTITION_DATA));
+
+ InputFile manifest = writeManifest(format, EMPTY_PARTITION, files);
+
+ List<TrackedFile> read = read(manifest, UNPARTITIONED_SPECS);
+ assertThat(read)
+ .allSatisfy(
+ file ->
assertThat(file.tracking().manifestLocation()).isEqualTo(manifest.location()));
+ assertThat(read).extracting(file ->
file.tracking().manifestPos()).containsExactly(0L, 1L, 2L);
+ }
+
+ @ParameterizedTest(name = "{0} / {1}")
+ @MethodSource("selectiveReadModes")
+ public void selectiveReadReturnsOnlyRequestedFields(
+ FileFormat format, Consumer<V4ManifestReader.Builder> configureRead)
throws IOException {
+ List<TrackedFile> files =
+ ImmutableList.of(
+ dataFile("s3://bucket/live.parquet", EMPTY_PARTITION_DATA),
+ fileWithStatus(EntryStatus.DELETED, "s3://bucket/deleted.parquet"),
+ fileWithStatus(EntryStatus.REPLACED,
"s3://bucket/replaced.parquet"));
+
+ InputFile manifest = writeManifest(format, EMPTY_PARTITION, files);
+
+ V4ManifestReader.Builder builder = V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS);
+ configureRead.accept(builder);
+ try (V4ManifestReader reader = builder.build()) {
+ TrackedFile actual = Iterables.getOnlyElement(reader);
+
+ // the requested field is read
+ assertThat(actual.location()).isEqualTo("s3://bucket/live.parquet");
+
+ // the reader always projects the fields it consumes internally, even
though the caller
+ // selected only location: content type and status (liveness filtering
keeps only the live
+ // entry), and manifest position (from row_position)
+ assertThat(actual.contentType()).isEqualTo(FileContent.DATA);
+ assertThat(actual.tracking().status()).isEqualTo(EntryStatus.ADDED);
+ assertThat(actual.tracking().manifestPos()).isEqualTo(0L);
+
+ // every field the caller did not request and the reader does not
require is omitted;
+ // content stats in particular (the largest projection) is not read
+ assertThat(actual.contentStats()).isNull();
+ assertThat(actual.fileFormat()).isNull();
+ assertThat(actual.recordCount()).isEqualTo(-1L);
+ assertThat(actual.fileSizeInBytes()).isEqualTo(-1L);
+ assertThat(actual.specId()).isNull();
+ assertThat(actual.partition()).isNull();
+ assertThat(actual.sortOrderId()).isNull();
+ assertThat(actual.deletionVector()).isNull();
+ assertThat(actual.keyMetadata()).isNull();
+ assertThat(actual.splitOffsets()).isNull();
+ assertThat(actual.equalityIds()).isNull();
+ }
+ }
+
+ private static Stream<Arguments> selectiveReadModes() {
+ Map<String, Consumer<V4ManifestReader.Builder>> modes =
+ ImmutableMap.of(
+ "project",
+ builder -> builder.project(new Schema(TrackedFile.LOCATION)),
+ "select",
+ builder -> builder.select("location"),
+ "case-insensitive select",
+ builder -> builder.select("LOCATION").caseSensitive(false));
+ return MANIFEST_FORMATS.stream()
+ .flatMap(
+ format ->
+ modes.entrySet().stream()
+ .map(mode -> Arguments.of(format, Named.of(mode.getKey(),
mode.getValue()))));
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void rowFilterForcesRecordCount(FileFormat format) throws IOException
{
+ TrackedFile file = dataFile("s3://bucket/file.parquet",
EMPTY_PARTITION_DATA);
+
+ InputFile manifest = writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(file));
+
+ // record_count is read when evaluating a row filter against file metrics,
so it is projected
+ // even though the caller selected only location
+ Schema projection = new Schema(TrackedFile.LOCATION);
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .project(projection)
+ .filter(Expressions.equal("id", 1))
+ .build()) {
+ TrackedFile actual = Iterables.getOnlyElement(reader);
+ assertThat(actual.location()).isEqualTo(file.location());
+ assertThat(actual.recordCount()).isEqualTo(RECORD_COUNT);
+ }
+ }
+
+ @Test
+ public void projectionModesAreMutuallyExclusive() {
+ InputFile manifest =
fileIO.newInputFile(tempDir.resolve("manifest.avro").toString());
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .select("location")
+ .project(new Schema(TrackedFile.LOCATION)))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Cannot use project(Schema) with
select(Collection<String>)");
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .project(new Schema(TrackedFile.LOCATION))
+ .select("location"))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Cannot use select(Collection<String>) with
project(Schema)");
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .forScanPlanning()
+ .select("location"))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Cannot use select(Collection<String>) with
forScanPlanning()");
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .select("location")
+ .forScanPlanning())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Cannot use forScanPlanning() with select(Collection<String>) or
project(Schema)");
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .forScanPlanning()
+ .project(new Schema(TrackedFile.LOCATION)))
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage("Cannot use project(Schema) with forScanPlanning()");
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .project(new Schema(TrackedFile.LOCATION))
+ .forScanPlanning())
+ .isInstanceOf(IllegalStateException.class)
+ .hasMessage(
+ "Cannot use forScanPlanning() with select(Collection<String>) or
project(Schema)");
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void projectionPreservesNarrowTrackingProjection(FileFormat format)
throws IOException {
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(FILE_WITH_FULL_TRACKING));
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).select("tracking.status").build()) {
+ Tracking actual = Iterables.getOnlyElement(reader).tracking();
+ assertThat(actual.status()).isEqualTo(EntryStatus.ADDED);
+ // the narrow tracking projection is not widened to the full tracking
type
+ assertThat(actual.snapshotId()).isNull();
+ assertThat(actual.dvSnapshotId()).isNull();
+ assertThat(actual.deletedPositions()).isNull();
+ assertThat(actual.replacedPositions()).isNull();
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void forScanPlanningOmitsChangeTrackingFields(FileFormat format)
throws IOException {
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(FILE_WITH_FULL_TRACKING));
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).forScanPlanning().build()) {
+ Tracking actual = Iterables.getOnlyElement(reader).tracking();
+ // scan-relevant tracking fields are projected
+ assertThat(actual.status()).isEqualTo(EntryStatus.ADDED);
+ assertThat(actual.snapshotId()).isEqualTo(SNAPSHOT_ID);
+ assertThat(actual.dataSequenceNumber()).isEqualTo(5L);
+ assertThat(actual.fileSequenceNumber()).isEqualTo(6L);
+ assertThat(actual.firstRowId()).isEqualTo(8L);
+ // change-tracking fields are omitted from the scan projection
+ assertThat(actual.dvSnapshotId()).isNull();
+ assertThat(actual.deletedPositions()).isNull();
+ assertThat(actual.replacedPositions()).isNull();
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void defaultReadsFullTracking(FileFormat format) throws IOException {
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(FILE_WITH_FULL_TRACKING));
+
+ // without scanPlanning, select, or project, the reader returns the full
schema for copying to
+ // other manifests, including the change-tracking fields
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) {
+ Tracking actual = Iterables.getOnlyElement(reader).tracking();
+ assertThat(actual.status()).isEqualTo(EntryStatus.ADDED);
+ assertThat(actual.snapshotId()).isEqualTo(SNAPSHOT_ID);
+ assertThat(actual.dataSequenceNumber()).isEqualTo(5L);
+ assertThat(actual.fileSequenceNumber()).isEqualTo(6L);
+ assertThat(actual.firstRowId()).isEqualTo(8L);
+ assertThat(actual.dvSnapshotId()).isEqualTo(7L);
+ assertThat(actual.deletedPositions()).isEqualTo(ByteBuffer.wrap(new
byte[] {1, 2}));
+ assertThat(actual.replacedPositions()).isEqualTo(ByteBuffer.wrap(new
byte[] {3, 4}));
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void projectNullReadsFullSchema(FileFormat format) throws IOException
{
+ InputFile manifest =
+ writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(FILE_WITH_FULL_TRACKING));
+
+ // project(null) clears the projection and reads the full schema, like no
projection at all
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).project(null).build()) {
+ TrackedFile actual = Iterables.getOnlyElement(reader);
+ assertThat(actual.location()).isEqualTo("s3://bucket/file.parquet");
+ assertThat(actual.fileFormat()).isEqualTo(FileFormat.PARQUET);
+ assertThat(actual.recordCount()).isEqualTo(RECORD_COUNT);
+ assertThat(actual.fileSizeInBytes()).isEqualTo(FILE_SIZE_IN_BYTES);
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void partitionFilterForceProjectsFilterFields(FileFormat format)
throws IOException {
+ InputFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(FILE_A, FILE_B));
+
+ // the caller projects only location; the reader must still project the
fields the partition
+ // filter reads (spec_id, partition) or every row would be pruned
+ Schema projection = new Schema(TrackedFile.LOCATION);
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS)
+ .project(projection)
+ .filter(Expressions.equal("id", 1))
+ .build()) {
+ TrackedFile actual = Iterables.getOnlyElement(reader);
+ assertThat(actual.location()).isEqualTo(FILE_A.location());
+ assertThat(actual.specId()).isEqualTo(ID_PARTITIONING.specId());
+ assertThat(actual.partition().get(0, Integer.class)).isEqualTo(1);
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void selectWithPartitionFilterProjectsFilterFields(FileFormat format)
throws IOException {
+ InputFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(FILE_A, FILE_B));
+
+ // the caller selects only location; the reader must still project spec_id
and partition
+ // for the partition filter or every row would be pruned
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS)
+ .select("location")
+ .filter(Expressions.equal("id", 1))
+ .build()) {
+ TrackedFile actual = Iterables.getOnlyElement(reader);
+ assertThat(actual.location()).isEqualTo(FILE_A.location());
+ assertThat(actual.specId()).isEqualTo(ID_PARTITIONING.specId());
+ assertThat(actual.partition().get(0, Integer.class)).isEqualTo(1);
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void unpartitionedProducesNullPartitionValue(FileFormat format)
throws IOException {
+ TrackedFile file = dataFile("s3://bucket/file.parquet",
EMPTY_PARTITION_DATA);
+
+ InputFile manifest = writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(file));
+
+ TrackedFile actual = Iterables.getOnlyElement(read(manifest,
UNPARTITIONED_SPECS));
+ // unpartitioned manifests omit the partition field, which is read as null
+ assertThat(actual.partition()).isNull();
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void partitionFilterPrunesFilesAndCountsSkips(FileFormat format)
throws IOException {
+ // FILE_A and EQ_DELETES_A match the filter; FILE_B and EQ_DELETES_B are
pruned; manifest
+ // references have no partition and are always kept
+ InputFile manifest =
+ writeManifest(
+ format,
+ ID_PARTITION_TYPE,
+ ImmutableList.of(
+ FILE_A,
+ FILE_B,
+ EQ_DELETES_A,
+ EQ_DELETES_B,
+ DATA_MANIFEST_REF,
+ DELETE_MANIFEST_REF));
+
+ ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext());
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS)
+ .filter(Expressions.equal("id", 1))
+ .scanMetrics(metrics)
+ .build()) {
+ assertThat(reader)
+ .extracting(TrackedFile::location)
+ .containsExactlyInAnyOrder(
+ FILE_A.location(),
+ EQ_DELETES_A.location(),
+ DATA_MANIFEST_REF.location(),
+ DELETE_MANIFEST_REF.location());
+ }
+
+ assertThat(metrics.skippedDataFiles().value())
+ .as("one data file is pruned by the partition filter")
+ .isEqualTo(1L);
+ assertThat(metrics.skippedDeleteFiles().value())
+ .as("one delete file is pruned by the partition filter")
+ .isEqualTo(1L);
+ assertThat(metrics.skippedDataManifests().value())
+ .as("manifests have no partition and are not pruned")
+ .isEqualTo(0L);
+ assertThat(metrics.skippedDeleteManifests().value())
+ .as("manifests have no partition and are not pruned")
+ .isEqualTo(0L);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void rowFilterKeepsFilesWithoutStats(FileFormat format) throws
IOException {
+ // with no content stats to evaluate, the row filter cannot prune any file
+ TrackedFile file1 = dataFile("s3://bucket/a.parquet",
EMPTY_PARTITION_DATA);
+ TrackedFile file2 = dataFile("s3://bucket/b.parquet",
EMPTY_PARTITION_DATA);
+
+ InputFile manifest = writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(file1, file2));
+
+ ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext());
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .filter(Expressions.equal("id", 1))
+ .scanMetrics(metrics)
+ .build()) {
+ assertThat(reader)
+ .extracting(TrackedFile::location)
+ .containsExactly(file1.location(), file2.location());
+ }
+
+ assertThat(metrics.skippedDataFiles().value()).isEqualTo(0L);
+ assertThat(metrics.skippedDeleteFiles().value()).isEqualTo(0L);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void caseInsensitivePartitionFilter(FileFormat format) throws
IOException {
+ InputFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(FILE_A, FILE_B));
+
+ // a case-insensitive filter binds the mismatched-case "ID" reference and
prunes FILE_B
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS)
+ .filter(Expressions.equal("ID", 1))
+ .caseSensitive(false)
+ .build()) {
+
assertThat(reader).extracting(TrackedFile::location).containsExactly(FILE_A.location());
+ }
+
+ // the same filter is case-sensitive by default, so "ID" fails to bind to
the "id" field
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS)
+ .filter(Expressions.equal("ID", 1))
+ .build())
+ .isInstanceOf(ValidationException.class)
+ .hasMessageContaining("Cannot find field 'ID'");
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void filterMatchesFilesAcrossDisjointSpecs(FileFormat format) throws
IOException {
+ PartitionSpec spec0 =
+ PartitionSpec.builderFor(TABLE_SCHEMA)
+ .withSpecId(0)
+ .add(1, 1000, "id", Transforms.identity())
+ .build();
+ PartitionSpec spec1 =
+ PartitionSpec.builderFor(TABLE_SCHEMA)
+ .withSpecId(1)
+ .add(2, 1001, "data", Transforms.identity())
+ .build();
+ Map<Integer, PartitionSpec> specsById =
+ ImmutableMap.of(spec0.specId(), spec0, spec1.specId(), spec1);
+ Types.StructType unionType =
Partitioning.unionPartitionTypes(specsById.values());
+
+ TrackedFile keepById =
+ dataFile("spec0-id1.parquet", spec0.specId(),
unionPartition(unionType, 1, null));
+ TrackedFile prunedById =
+ dataFile("spec0-id2.parquet", spec0.specId(),
unionPartition(unionType, 2, null));
+ TrackedFile keptOtherSpec =
+ dataFile("spec1-data.parquet", spec1.specId(),
unionPartition(unionType, null, "x"));
+
+ InputFile manifest =
+ writeManifest(format, unionType, ImmutableList.of(keepById,
prunedById, keptOtherSpec));
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest,
specsById).filter(Expressions.equal("id", 1)).build()) {
+ // spec0 entries are pruned by id; the spec1 entry is not partitioned by
id so it survives
+ assertThat(reader)
+ .extracting(TrackedFile::location)
+ .containsExactlyInAnyOrder(keepById.location(),
keptOtherSpec.location());
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void partialFilterStillPrunesOnCompatibleField(FileFormat format)
throws IOException {
+ // the spec partitions on id only; a filter of id = 1 AND data = 'z'
should still prune by id
+ // even though data is not a partition source
+ TrackedFile keep = dataFile("id1.parquet", partition(1));
+ TrackedFile prune = dataFile("id2.parquet", partition(2));
+
+ InputFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(keep, prune));
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS)
+ .filter(Expressions.and(Expressions.equal("id", 1),
Expressions.equal("data", "z")))
+ .build()) {
+ assertThat(reader)
+ .extracting(TrackedFile::location)
+ .as("the id predicate prunes even though data is not a partition
field")
+ .containsExactly(keep.location());
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void partitionFilterKeepsFileWithUnknownSpec(FileFormat format)
throws IOException {
+ // spec ID 5 is not in ID_PARTITIONING_SPECS, so no partition filter
applies to this file
+ TrackedFile file = dataFile("orphan.parquet", 5, partition(1));
+
+ InputFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(file));
+
+ // the filter would prune partition id=1 under spec 0, but cannot be
applied to spec 5
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS)
+ .filter(Expressions.equal("id", 2))
+ .build()) {
+
assertThat(reader).extracting(TrackedFile::location).containsExactly(file.location());
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void partitionFilterKeepsFileWithNullSpecId(FileFormat format) throws
IOException {
+ TrackedFile file = dataFile("no-spec.parquet", null, null);
+
+ InputFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(file));
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS)
+ .filter(Expressions.equal("id", 2))
+ .build()) {
+
assertThat(reader).extracting(TrackedFile::location).containsExactly(file.location());
+ }
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void iteratorReturnsLiveCopies(FileFormat format) throws IOException {
+ TrackedFile added1 = dataFile("s3://bucket/added-1.parquet",
EMPTY_PARTITION_DATA);
+ TrackedFile added2 = dataFile("s3://bucket/added-2.parquet",
EMPTY_PARTITION_DATA);
+ List<TrackedFile> files =
+ ImmutableList.of(
+ added1, added2, fileWithStatus(EntryStatus.DELETED,
"s3://bucket/deleted.parquet"));
+
+ InputFile manifest = writeManifest(format, EMPTY_PARTITION, files);
+
+ try (V4ManifestReader reader =
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) {
+ List<TrackedFile> read = Lists.newArrayList(reader);
+ assertThat(read)
+ .hasSize(2)
+ .extracting(TrackedFile::location)
+ .containsExactly(added1.location(), added2.location());
+ assertThat(read.get(0))
+ .as("iterator() should copy each entry rather than yield one reused
container")
+ .isNotSameAs(read.get(1));
+ }
+ }
+
+ @Test
+ public void unknownManifestFormatThrows() throws IOException {
+ InputFile badFile =
+ fileIO.newInputFile(tempDir.resolve("manifest-" + System.nanoTime() +
".txt").toString());
+
+ try (V4ManifestReader reader = V4ManifestReader.builder(badFile,
UNPARTITIONED_SPECS).build()) {
+ assertThatThrownBy(reader::iterator)
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessageContaining("Cannot determine format of manifest");
+ }
+ }
+
+ @Test
+ public void invalidBuilderArguments() {
+ InputFile manifest =
fileIO.newInputFile(tempDir.resolve("manifest.avro").toString());
+
+ assertThatThrownBy(() -> V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).filter(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid filter: null");
+
+ assertThatThrownBy(
+ () -> V4ManifestReader.builder(manifest,
UNPARTITIONED_SPECS).scanMetrics(null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid scan metrics: null");
+
+ assertThatThrownBy(
+ () ->
+ V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS)
+ .select((Collection<String>) null))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Invalid columns: null");
+ }
+
+ private static TrackedFile dataFile(String location, PartitionData
partition) {
+ return dataFile(location, 0, partition);
+ }
+
+ private static TrackedFile dataFile(String location, Integer specId,
PartitionData partition) {
+ return new TrackedFileStruct(
+ addedTracking(),
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ location,
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ specId,
+ partition,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private static TrackedFile deleteFile(String location, PartitionData
partition) {
+ return new TrackedFileStruct(
+ addedTracking(),
+ FileContent.EQUALITY_DELETES,
+ FORMAT_VERSION_V4,
+ location,
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ 0,
+ partition,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ ImmutableList.of(1));
+ }
+
+ private static TrackedFile manifestRef(FileContent content, String location)
{
+ ManifestInfo info = new ManifestInfoStruct(1, 0, 0, 0, 1L, 0L, 0L, 0L, 1L,
null, null);
+ return new TrackedFileStruct(
+ addedTracking(),
+ content,
+ FORMAT_VERSION_V4,
+ location,
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ null, // spec_id: a manifest reference has no spec
+ null, // partition: a manifest reference has no partition tuple
+ null, // content_stats
+ null, // sort_order_id
+ null, // deletion_vector
+ info,
+ null, // key_metadata
+ null, // split_offsets
+ null); // equality_ids
+ }
+
+ private static TrackedFile fileWithStatus(EntryStatus status, String
location) {
+ Tracking tracking =
+ new TrackingStruct(
+ status,
+ SNAPSHOT_ID,
+ 3L, // data sequence number
+ 3L, // file sequence number
+ null, // dv snapshot id
+ null, // first row id
+ null, // deleted positions
+ null); // replaced positions
+ return new TrackedFileStruct(
+ tracking,
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ location,
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ 0,
+ EMPTY_PARTITION_DATA,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null);
+ }
+
+ private static Tracking addedTracking() {
+ return new TrackingStruct(EntryStatus.ADDED, SNAPSHOT_ID, null, null,
null, null, null, null);
+ }
+
+ private static PartitionData partition(int id) {
+ PartitionData partition = new PartitionData(ID_PARTITION_TYPE);
+ partition.set(0, id);
+ return partition;
+ }
+
+ private static PartitionData unionPartition(Types.StructType unionType,
Integer id, String data) {
+ PartitionData partition = new PartitionData(unionType);
+ partition.set(0, id);
+ partition.set(1, data);
+ return partition;
+ }
+
+ private InputFile writeManifest(
+ FileFormat format, Types.StructType partitionType, Iterable<TrackedFile>
files)
+ throws IOException {
+ Schema writeSchema = TrackedFile.schema(partitionType,
Types.StructType.of());
+ OutputFile out = new InMemoryOutputFile("manifest." +
format.name().toLowerCase(Locale.ROOT));
+ try (FileAppender<StructLike> appender =
+ InternalData.write(format,
out).schema(writeSchema).named("tracked_file").build()) {
+ for (TrackedFile file : files) {
+ appender.add((StructLike) file);
+ }
+ }
+
+ return out.toInputFile();
+ }
+
+ private List<TrackedFile> read(InputFile manifest, Map<Integer,
PartitionSpec> specsById)
+ throws IOException {
+ try (V4ManifestReader reader = V4ManifestReader.builder(manifest,
specsById).build()) {
+ return Lists.newArrayList(reader);
+ }
+ }
+}
