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 000000000000..1c94bd57c114 --- /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 { + private final Map replacementsById; + + ReplaceTypeById(Map replacementsById) { + this.replacementsById = replacementsById; + } + + @Override + public Type schema(Schema schema, Type structResult) { + return structResult; + } + + @Override + public Type struct(Types.StructType struct, List fieldResults) { + List fields = struct.fields(); + List 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 a3bee3e3d860..8e39ae7a43bc 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 static Schema selectNot(Schema schema, Set fieldIds) { 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 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 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 b7da4b3108e6..d540d239614e 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.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 void testIndexStatsNames() { .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 c708d39f523e..66a681f88486 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 static StructType partitionType(Table table) { "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 specs) { + return buildPartitionProjectionType("union partition", specs, allFieldIds(specs)); + } + /** * Checks if any of the specs in a table is partitioned. * @@ -347,6 +358,14 @@ private static boolean compatibleTransforms(Transform t1, Transform || t2.equals(Transforms.alwaysNull()); } + // collects IDs of all partition fields used across specs + private static Set allFieldIds(Collection 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 allActiveFieldIds(Schema schema, Collection 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 9aaeae48d688..1a457fb048e2 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.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. + * + *

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 @@ static Types.StructType schemaWithContentStats( 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 @@ static Types.StructType schemaWithContentStats( 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 3ed764ff17d9..f5ce03c7eb62 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 8ae4b7e4ce88..283e0a51ffb3 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 @@ /** 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 000000000000..a823454845dd --- /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 { + 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> partitionFilters; + + private V4ManifestReader( + InputFile file, + Schema readSchema, + Map> 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 specsById) { + return new Builder(file, specsById); + } + + /** Returns copies of the tracked files that match this reader's configured filters. */ + @Override + public CloseableIterator iterator() { + CloseableIterable 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 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 open() { + FileFormat format = FileFormat.fromFileName(file.location()); + Preconditions.checkArgument( + format != null, "Cannot determine format of manifest: %s", file.location()); + + CloseableIterable 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 specsById; + private final Schema fullSchema; + private Expression rowFilter = Expressions.alwaysTrue(); + private boolean caseSensitive = true; + private boolean includeAll = false; + private boolean scanPlanning = false; + private Collection columns = null; + private Schema requestedProjection = null; + private ScanMetrics scanMetrics = ScanMetrics.noop(); + + private Builder(InputFile file, Map 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) 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 newColumns) { + Preconditions.checkArgument(newColumns != null, "Invalid columns: null"); + Preconditions.checkState( + !scanPlanning, "Cannot use select(Collection) with forScanPlanning()"); + Preconditions.checkState( + requestedProjection == null, + "Cannot use select(Collection) 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)"); + 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> 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 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 eaabf5d3ed4d..4c98d66c10f7 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 void testPartitionTypeIgnoreInactiveFields() { 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 5840a1cd5f3d..9a32243cc12f 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 fields = type.fields(); assertThat(fields) @@ -65,8 +65,8 @@ public void schemaWithContentStatsFieldOrder() { } @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 fields = type.fields(); assertThat(fields) @@ -76,8 +76,8 @@ public void schemaWithContentStatsFieldIds() { } @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 void schemaWithContentStatsUsesProvidedType() { } @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 void schemaWithContentStatsReflectsInput() { 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 void schemaWithContentStatsReflectsInput() { } @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 37b8beb112a2..96422c126411 100644 --- a/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java +++ b/core/src/test/java/org/apache/iceberg/TestTrackedFileAdapters.java @@ -110,10 +110,10 @@ void testDataFileAdapterDelegation() { FORMAT_VERSION_V4, DATA_FILE_LOCATION, FileFormat.PARQUET, - PARTITION, 100L, 1024L, PARTITIONED_SPEC_ID, + PARTITION, CONTENT_STATS, 3, null, @@ -186,10 +186,10 @@ void testEqualityDeleteFileAdapterDelegation() { FORMAT_VERSION_V4, "s3://bucket/eq-delete.avro", FileFormat.AVRO, - PARTITION, 50L, 512L, PARTITIONED_SPEC_ID, + PARTITION, CONTENT_STATS, 5, null, @@ -271,10 +271,10 @@ void testDVDeleteFileAdapterDelegation() { FORMAT_VERSION_V4, DATA_FILE_LOCATION, FileFormat.PARQUET, - PARTITION, 100L, 1024L, PARTITIONED_SPEC_ID, + PARTITION, null, null, dv, @@ -368,12 +368,12 @@ void testNullTrackingReturnsNullTrackingFields() { 0, null, null, - null, 0L, 0L, null, null, null, + null, deletionVector(), null, null, @@ -422,7 +422,6 @@ void testUnknownSpecIdThrows() { 0, null, null, - null, 0L, 0L, 99, @@ -432,6 +431,7 @@ void testUnknownSpecIdThrows() { null, null, null, + null, null); assertThatThrownBy(() -> TrackedFileAdapters.asDataFile(file, ImmutableMap.of())) @@ -448,7 +448,6 @@ void testSpecIdMismatchThrows() { 0, null, null, - null, 0L, 0L, PARTITIONED_SPEC_ID, @@ -458,6 +457,7 @@ void testSpecIdMismatchThrows() { null, null, null, + null, null); int mismatchedSpecId = PARTITIONED_SPEC_ID + 1; PartitionSpec mismatched = @@ -503,7 +503,6 @@ private static TrackedFileStruct dummyTrackedFile(FileContent contentType) { FORMAT_VERSION_V4, DATA_FILE_LOCATION, FileFormat.PARQUET, - null, 1L, 1L, null, @@ -513,6 +512,7 @@ private static TrackedFileStruct dummyTrackedFile(FileContent contentType) { 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 d888c6a7dc68..14265c91f692 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 @@ 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 FIELDS = - TrackedFile.schemaWithContentStats(PARTITION_TYPE, Types.StructType.of()).fields(); + private static final List 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 @@ void fieldAccess() { 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 @@ void getByPosition() { 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 @@ void copy() { 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 @@ void copyWithStats() { 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 @@ void copyWithoutStats() { 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 @@ void projectedStructLike() { @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 @@ void serializationRoundTrip(RoundTripSerializer serializer) t 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 @@ void serializationRoundTrip(RoundTripSerializer serializer) t } 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 000000000000..c8a5cfd61a31 --- /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 ID_PARTITIONING_SPECS = + ImmutableMap.of(ID_PARTITIONING.specId(), ID_PARTITIONING); + private static final Map UNPARTITIONED_SPECS = + ImmutableMap.of(PartitionSpec.unpartitioned().specId(), PartitionSpec.unpartitioned()); + + private static final List 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 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 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 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 configureRead) throws IOException { + List 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 selectiveReadModes() { + Map> 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)"); + + assertThatThrownBy( + () -> + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + .project(new Schema(TrackedFile.LOCATION)) + .select("location")) + .isInstanceOf(IllegalStateException.class) + .hasMessage("Cannot use select(Collection) with project(Schema)"); + + assertThatThrownBy( + () -> + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + .forScanPlanning() + .select("location")) + .isInstanceOf(IllegalStateException.class) + .hasMessage("Cannot use select(Collection) with forScanPlanning()"); + + assertThatThrownBy( + () -> + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + .select("location") + .forScanPlanning()) + .isInstanceOf(IllegalStateException.class) + .hasMessage( + "Cannot use forScanPlanning() with select(Collection) 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) 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 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 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 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) 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 files) + throws IOException { + Schema writeSchema = TrackedFile.schema(partitionType, Types.StructType.of()); + OutputFile out = new InMemoryOutputFile("manifest." + format.name().toLowerCase(Locale.ROOT)); + try (FileAppender appender = + InternalData.write(format, out).schema(writeSchema).named("tracked_file").build()) { + for (TrackedFile file : files) { + appender.add((StructLike) file); + } + } + + return out.toInputFile(); + } + + private List read(InputFile manifest, Map specsById) + throws IOException { + try (V4ManifestReader reader = V4ManifestReader.builder(manifest, specsById).build()) { + return Lists.newArrayList(reader); + } + } +}