stevenzwu commented on code in PR #18109:
URL: https://github.com/apache/iceberg/pull/18109#discussion_r4022567946
##########
core/src/main/java/org/apache/iceberg/TrackedFileStruct.java:
##########
@@ -85,7 +85,7 @@ class TrackedFileStruct extends SupportsIndexProjection
implements TrackedFile,
super(BASE_TYPE, projection);
// partition type may be null if the field was not projected, or unknown
for unpartitioned
// manifests
- Type partType = projection.fieldType(TrackedFile.PARTITION_NAME);
+ Type partType = projection.fieldType("partition");
Review Comment:
This reverts to a literal; `TrackedFile.PARTITION_NAME` is still defined and
used a few lines above (`BASE_TYPE`, line 49). Suggest keeping the constant
reference for consistency.
##########
core/src/test/java/org/apache/iceberg/V4TestComparators.java:
##########
@@ -0,0 +1,195 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
+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.Comparators;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+
+/**
+ * Comparators for manifest interfaces that compare the values returned by API
methods.
+ *
+ * <p>Comparators for methods that return an object are null tolerant, as are
the comparators for
+ * the interfaces themselves.
+ */
+class V4TestComparators {
+ private V4TestComparators() {}
+
+ public static Comparator<TrackedFile> trackedFileStatusOnly(Types.StructType
partitionType) {
+ return trackedFile(STATUS_ONLY_TRACKING, partitionType);
+ }
+
+ public static Comparator<TrackedFile> trackedFile(Types.StructType
partitionType) {
+ return trackedFile(TRACKING, partitionType);
+ }
+
+ private static Comparator<TrackedFile> trackedFile(
+ Comparator<Tracking> trackingComparator, Types.StructType partitionType)
{
+ Comparator<StructLike> partitionComparator =
+ Comparator.nullsFirst(Comparators.forType(partitionType));
+
+ return Comparator.nullsFirst(
+ Comparator.comparing(TrackedFile::tracking, trackingComparator)
+ .thenComparing(TrackedFile::contentType, natural())
+ .thenComparingInt(TrackedFile::formatVersion)
+ .thenComparing(TrackedFile::location, natural())
+ .thenComparing(TrackedFile::fileFormat, natural())
+ .thenComparingLong(TrackedFile::recordCount)
+ .thenComparingLong(TrackedFile::fileSizeInBytes)
+ .thenComparing(TrackedFile::specId, natural())
+ .thenComparing(TrackedFile::partition, partitionComparator)
+ .thenComparing(TrackedFile::contentStats, CONTENT_STATS)
+ .thenComparing(TrackedFile::sortOrderId, natural())
+ .thenComparing(TrackedFile::deletionVector, DELETION_VECTOR)
+ .thenComparing(TrackedFile::manifestInfo, MANIFEST_INFO)
+ .thenComparing(TrackedFile::keyMetadata, BYTES)
+ .thenComparing(TrackedFile::splitOffsets, SPLIT_OFFSETS)
+ .thenComparing(TrackedFile::equalityIds, EQ_IDS));
+ }
+
+ // convenience method for a null-safe natural order comparator
+ private static <T extends Comparable<T>> Comparator<T> natural() {
+ return Comparator.nullsFirst(Comparator.naturalOrder());
+ }
+
+ private static final Comparator<ByteBuffer> BYTES =
+ Comparator.nullsFirst(Comparators.unsignedBytes());
+ private static final Comparator<List<Long>> SPLIT_OFFSETS =
+
Comparator.nullsFirst(Comparators.forType(TrackedFile.SPLIT_OFFSETS.type().asListType()));
+ private static final Comparator<List<Integer>> EQ_IDS =
+
Comparator.nullsFirst(Comparators.forType(TrackedFile.EQUALITY_IDS.type().asListType()));
+
+ static final Comparator<Tracking> TRACKING =
+ Comparator.nullsFirst(
+ Comparator.comparing(Tracking::status, natural())
+ .thenComparing(Tracking::snapshotId, natural())
+ .thenComparing(Tracking::dataSequenceNumber, natural())
+ .thenComparing(Tracking::fileSequenceNumber, natural())
+ .thenComparing(Tracking::dvSnapshotId, natural())
+ .thenComparing(Tracking::firstRowId, natural())
+ .thenComparing(Tracking::deletedPositions, BYTES)
+ .thenComparing(Tracking::replacedPositions, BYTES)
+ .thenComparing(Tracking::manifestLocation, natural())
+ .thenComparingLong(Tracking::manifestPos));
+
+ // compare only status, ignoring inherited fields and fields set during a
write
+ static final Comparator<Tracking> STATUS_ONLY_TRACKING =
+ Comparator.nullsFirst(Comparator.comparing(Tracking::status, natural()));
+
+ private static final Comparator<DeletionVector> DELETION_VECTOR =
+ Comparator.nullsFirst(
+ Comparator.comparing(DeletionVector::location, natural())
+ .thenComparingLong(DeletionVector::offset)
+ .thenComparingLong(DeletionVector::sizeInBytes)
+ .thenComparingLong(DeletionVector::cardinality)
+ .thenComparing(DeletionVector::keyMetadata, BYTES));
+
+ private static final Comparator<ManifestInfo> MANIFEST_INFO =
+ Comparator.nullsFirst(
+ Comparator.comparingInt(ManifestInfo::addedFilesCount)
+ .thenComparingInt(ManifestInfo::existingFilesCount)
+ .thenComparingInt(ManifestInfo::deletedFilesCount)
+ .thenComparingInt(ManifestInfo::replacedFilesCount)
+ .thenComparingLong(ManifestInfo::addedRowsCount)
+ .thenComparingLong(ManifestInfo::existingRowsCount)
+ .thenComparingLong(ManifestInfo::deletedRowsCount)
+ .thenComparingLong(ManifestInfo::replacedRowsCount)
+ .thenComparingLong(ManifestInfo::minSequenceNumber)
+ .thenComparing(ManifestInfo::dv, BYTES)
+ .thenComparing(ManifestInfo::dvCardinality, natural()));
+
+ private static final Comparator<ContentStats> CONTENT_STATS =
+ Comparator.nullsFirst(new ContentStatsComparator());
+
+ private static class ContentStatsComparator implements
Comparator<ContentStats> {
+ @Override
+ public int compare(ContentStats left, ContentStats right) {
+ Map<Integer, FieldStats<?>> leftStats = statsById(left);
+ Map<Integer, FieldStats<?>> rightStats = statsById(right);
+
+ Set<Integer> fieldIds =
+ Sets.newTreeSet(Iterables.concat(leftStats.keySet(),
rightStats.keySet()));
+ for (Integer fieldId : fieldIds) {
+ FieldStats<?> leftField = leftStats.get(fieldId);
+ FieldStats<?> rightField = rightStats.get(fieldId);
+ Comparator<FieldStats<?>> fieldComparator =
fieldStatsComparator(leftField);
Review Comment:
`fieldStatsComparator` derives the primitive type from `leftField` only.
When a `fieldId` is present only in `rightStats` (union above), `leftField` is
null and `type(stats).type()` NPEs. Maybe derive the type from whichever side
is non-null? or we can return -1 (null first) if the `leftField` is null?
##########
core/src/main/java/org/apache/iceberg/types/RestoreColumns.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.schema.SchemaWithPartnerVisitor;
+
+public class RestoreColumns extends SchemaWithPartnerVisitor<Type, Type> {
+ /**
+ * Restores columns from a base schema in a projection that may not include
them.
+ *
+ * @param base a base schema, from which the projection was produced
+ * @param projection a projection of the base schema
+ * @param fieldIds a set of field IDs to add to the projection if they are
missing
+ * @return an updated projection with the fields restored
+ */
+ public static Schema restore(Schema base, Schema projection, Set<Integer>
fieldIds) {
+ return SchemaWithPartnerVisitor.visit(
+ base, projection.asStruct(), new RestoreColumns(fieldIds),
FieldIdAccessors.get())
+ .asStructType()
+ .asSchema();
+ }
+
+ private final Set<Integer> restoredFields;
+
+ private RestoreColumns(Set<Integer> fieldsToRestore) {
+ this.restoredFields = fieldsToRestore;
+ }
+
+ @Override
+ public Type schema(Schema schema, Type partner, Type structResult) {
+ return structResult;
+ }
+
+ @Override
+ public Type struct(Types.StructType struct, Type partner, List<Type>
fieldResults) {
+ List<Types.NestedField> fields = struct.fields();
+
+ boolean hasFields = false;
+ List<Types.NestedField> newFields = Lists.newArrayList();
+ for (int i = 0; i < fields.size(); i += 1) {
+ Type newType = fieldResults.get(i);
+ if (newType != null) {
+ hasFields = true;
+
newFields.add(Types.NestedField.from(fields.get(i)).ofType(newType).build());
+ }
+ }
+
+ if (hasFields) {
Review Comment:
nit: instead of keeping the `hasFields` boolean flag, maybe we can just
check `if (!newFields.isEmpty())`?
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -44,486 +45,700 @@
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.apache.iceberg.util.LocationUtil;
import org.junit.jupiter.api.Named;
import org.junit.jupiter.api.Test;
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;
+import org.mockito.Mockito;
class TestV4ManifestReader {
+ private static final ManifestFile UNREAD_MANIFEST_FILE =
Mockito.mock(ManifestFile.class);
+
+ static {
+ Mockito.when(UNREAD_MANIFEST_FILE.path())
+ .thenReturn(FileFormat.PARQUET.addExtension("manifest"));
+ Mockito.when(UNREAD_MANIFEST_FILE.formatVersion()).thenReturn(4);
+
Mockito.when(UNREAD_MANIFEST_FILE.content()).thenReturn(ManifestContent.DATA);
+ }
+
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 String TABLE_LOCATION = "s3://bucket/db/table";
private static final DeletionVector DV = dv("s3://bucket/dv.puffin");
+ private static final Tracking ADDED_TRACKING =
TrackingBuilder.added(SNAPSHOT_ID).build();
+
+ private static final ManifestInfo MANIFEST_INFO =
+ new ManifestInfoStruct(49, 51, 0, 0, 4_900L, 5_100L, 0L, 0L, 1L, null,
null);
+
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 Schema LOCATION_ONLY_SCHEMA = new
Schema(TrackedFile.LOCATION);
+
+ private static final PartitionSpec ID_PARTITIONED =
+
PartitionSpec.builderFor(TABLE_SCHEMA).withSpecId(1).identity("id").build();
+ private static final Types.StructType ID_PARTITIONED_TYPE =
ID_PARTITIONED.partitionType();
private static final Map<Integer, PartitionSpec> ID_PARTITIONING_SPECS =
- ImmutableMap.of(ID_PARTITIONING.specId(), ID_PARTITIONING);
+ ImmutableMap.of(ID_PARTITIONED.specId(), ID_PARTITIONED);
+
+ private static final Types.StructType UNPARTITIONED_TYPE =
Types.StructType.of();
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);
+ private static final MetricsConfig METRICS_CONFIG =
+ MetricsConfig.from(ImmutableMap.of(), TABLE_SCHEMA, null);
+ private static final Types.StructType STATS_TYPE =
+ StatsUtil.statsWriteSchema(TABLE_SCHEMA, METRICS_CONFIG);
+ private static final Types.StructType ID_ONLY_STATS_TYPE =
+ Types.StructType.of(STATS_TYPE.field("id"));
+ private static final Types.StructType DATA_ONLY_STATS_TYPE =
+ Types.StructType.of(STATS_TYPE.field("data"));
+
+ private static final FieldStatsStruct<Integer> ID_STATS =
+ new FieldStatsStruct<>(
+ STATS_TYPE.fieldType("id").asStructType(), 0, 99, true,
RECORD_COUNT, 0, 0, null);
+ private static final FieldStatsStruct<String> DATA_STATS =
+ new FieldStatsStruct<>(
+ STATS_TYPE.fieldType("data").asStructType(), "a", "z", false,
RECORD_COUNT, 20, 0, null);
+ private static final ContentStatsStruct CONTENT_STATS = new
ContentStatsStruct(STATS_TYPE);
+
+ static {
+ CONTENT_STATS.setStats(1, ID_STATS);
+ CONTENT_STATS.setStats(2, DATA_STATS);
+ }
- // 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.
Locations are stored
- // relative to the table location (the default), so the reader resolves them
against the table
- 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 Comparator<TrackedFile> FILE_COMPARATOR =
+ V4TestComparators.trackedFileStatusOnly(ID_PARTITIONED_TYPE);
+ private static final Comparator<TrackedFile> UNPARTITIONED_FILE_COMPARATOR =
+ V4TestComparators.trackedFileStatusOnly(UNPARTITIONED_TYPE);
+
+ // shared data files: FILE_A is in partition id=1, FILE_B in partition id=2
+ private static final TrackedFile UNPARTITIONED_FILE =
+ unpartitionedFileWithoutStats("s3://bucket/table/unpartitioned.parquet");
+ private static final TrackedFile FILE_A =
+
idPartitionedDataFileWithoutStats("s3://bucket/table/id=1/file-a.parquet",
idPartition(1));
+ private static final TrackedFile FILE_B =
+
idPartitionedDataFileWithoutStats("s3://bucket/table/id=2/file-b.parquet",
idPartition(2));
+ private static final TrackedFile EQ_DELETES_A =
+ idPartitionedDeleteFileWithoutStats(
+ "s3://bucket/table/id=1/eq-deletes-a.parquet", idPartition(1));
+ private static final TrackedFile EQ_DELETES_B =
+ idPartitionedDeleteFileWithoutStats(
+ "s3://bucket/table/id=2/eq-deletes-b.parquet", idPartition(2));
private static final TrackedFile DATA_MANIFEST_REF =
- manifestRef(FileContent.DATA_MANIFEST, "data-leaf.parquet");
+ manifestRef(FileContent.DATA_MANIFEST,
"s3://bucket/table/data-leaf.parquet");
private static final TrackedFile DELETE_MANIFEST_REF =
- manifestRef(FileContent.DELETE_MANIFEST, "delete-leaf.parquet");
+ manifestRef(FileContent.DELETE_MANIFEST,
"s3://bucket/table/delete-leaf.parquet");
+
+ private static final List<FileFormat> MANIFEST_FORMATS =
+ ImmutableList.of(FileFormat.AVRO, FileFormat.PARQUET);
- private final InMemoryFileIO io = new InMemoryFileIO();
+ private static final InMemoryFileIO IO = new InMemoryFileIO();
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
- public void readsWrittenFile(FileFormat format) throws IOException {
+ public void readDataFile(FileFormat format) throws IOException {
TrackedFile file =
new TrackedFileStruct(
- addedTracking(),
+ ADDED_TRACKING,
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
+ ID_PARTITIONED.specId(),
+ idPartition(7),
+ CONTENT_STATS,
+ SortOrder.unsorted().orderId(),
DV,
- null,
- ByteBuffer.wrap(new byte[] {1, 2, 3}),
+ null, // manifest info
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
ImmutableList.of(50L, 100L),
- null);
+ null); // equality field IDs
- ManifestFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(file));
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
- TrackedFile actual = Iterables.getOnlyElement(read(manifest,
ID_PARTITIONING_SPECS));
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ TrackedFile actual = readOne(builder);
- // 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);
+ assertThat(actual).usingComparator(FILE_COMPARATOR).isEqualTo(file);
}
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
- public void equalityDeleteRoundTrip(FileFormat format) throws IOException {
+ public void readForScanPlanningDoesNotCopyStats(FileFormat format) throws
IOException {
+ TrackedFile file =
+ new TrackedFileStruct(
+ ADDED_TRACKING,
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ "s3://bucket/data/file.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ ID_PARTITIONED.specId(),
+ idPartition(7),
+ CONTENT_STATS,
+ SortOrder.unsorted().orderId(),
+ DV,
+ null, // manifest info
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
+ ImmutableList.of(50L, 100L),
+ null); // equality field IDs
+
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .forScanPlanning()
+ .metricsConfig(METRICS_CONFIG);
+ TrackedFile actual = readOne(builder);
+
+
assertThat(actual).usingComparator(FILE_COMPARATOR).isEqualTo(file.copyWithoutStats());
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void readForScanPlanningCopiesRequestedStats(FileFormat format)
throws IOException {
+ int idFieldId = TABLE_SCHEMA.findField("id").fieldId();
+ TrackedFile file =
+ new TrackedFileStruct(
+ ADDED_TRACKING,
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ "s3://bucket/data/file.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ ID_PARTITIONED.specId(),
+ idPartition(7),
+ CONTENT_STATS,
+ SortOrder.unsorted().orderId(),
+ DV,
+ null, // manifest info
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
+ ImmutableList.of(50L, 100L),
+ null); // equality field IDs
+
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .forScanPlanning()
+ .metricsConfig(METRICS_CONFIG)
+ .projectStats(idFieldId);
+ TrackedFile actual = readOne(builder);
+
+ assertThat(actual)
+ .usingComparator(FILE_COMPARATOR)
+ .isEqualTo(file.copyWithStats(Set.of(idFieldId)));
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void readEqualityDelete(FileFormat format) throws IOException {
TrackedFile delete =
new TrackedFileStruct(
- addedTracking(),
+ ADDED_TRACKING,
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,
+ ID_PARTITIONED.specId(),
+ idPartition(7),
+ CONTENT_STATS,
+ SortOrder.unsorted().orderId(),
+ null, // dv
+ null, // manifest info
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
+ null, // split offsets
ImmutableList.of(1, 2));
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(delete));
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, delete);
- TrackedFile actual = Iterables.getOnlyElement(read(manifest,
UNPARTITIONED_SPECS));
- assertThat(actual.contentType()).isEqualTo(FileContent.EQUALITY_DELETES);
- assertThat(actual.equalityIds()).containsExactly(1, 2);
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ TrackedFile actual = readOne(builder);
+
+ assertThat(actual).usingComparator(FILE_COMPARATOR).isEqualTo(delete);
}
@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"));
-
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION, files);
-
- try (V4ManifestReader reader =
- V4ManifestReader.builder(manifest, io, UNPARTITIONED_SPECS,
TABLE_LOCATION).build()) {
- assertThat(reader)
- .extracting(file -> file.tracking().status())
- .containsExactly(EntryStatus.ADDED, EntryStatus.EXISTING,
EntryStatus.MODIFIED);
- }
+ public void readManifestFile(FileFormat format) throws IOException {
+ TrackedFile manifestRef =
+ new TrackedFileStruct(
+ ADDED_TRACKING,
+ FileContent.DATA_MANIFEST,
+ FORMAT_VERSION_V4,
+ "s3://bucket/leaf-manifest.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ null, // spec id
+ null, // partition
+ CONTENT_STATS,
+ null, // sort order id
+ null, // dv
+ MANIFEST_INFO,
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
+ null, // split offsets
+ ImmutableList.of(1, 2));
- try (V4ManifestReader reader =
- V4ManifestReader.builder(manifest, io, UNPARTITIONED_SPECS,
TABLE_LOCATION)
- .includeAll()
- .build()) {
- assertThat(reader)
- .extracting(file -> file.tracking().status())
- .containsExactly(
- EntryStatus.ADDED,
- EntryStatus.EXISTING,
- EntryStatus.MODIFIED,
- EntryStatus.DELETED,
- EntryStatus.REPLACED);
- }
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE,
manifestRef);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ TrackedFile actual = readOne(builder);
+
+ assertThat(actual).usingComparator(FILE_COMPARATOR).isEqualTo(manifestRef);
}
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
- public void manifestLocationAndPosition(FileFormat format) throws
IOException {
+ public void statusFilter(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));
+ unpartitionedFileWithStatus(EntryStatus.ADDED,
"s3://bucket/added.parquet"),
+ unpartitionedFileWithStatus(EntryStatus.MODIFIED,
"s3://bucket/modified.parquet"),
+ unpartitionedFileWithStatus(EntryStatus.DELETED,
"s3://bucket/deleted.parquet"),
+ unpartitionedFileWithStatus(EntryStatus.EXISTING,
"s3://bucket/existing.parquet"),
+ unpartitionedFileWithStatus(EntryStatus.REPLACED,
"s3://bucket/replaced.parquet"));
+
+ ManifestFile manifest = writeManifest(format, UNPARTITIONED_TYPE, files);
+
+ List<TrackedFile> liveFiles =
+ read(
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .metricsConfig(METRICS_CONFIG));
+ assertThat(liveFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactly(files.get(0), files.get(1), files.get(3));
+
+ List<TrackedFile> allFiles =
+ read(
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .metricsConfig(METRICS_CONFIG)
+ .includeAll());
+ assertThat(allFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactlyElementsOf(files);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceManifestLocationAndPosition(FileFormat format) throws
IOException {
+ List<TrackedFile> files =
+ ImmutableList.of(FILE_A, FILE_B, DATA_MANIFEST_REF,
DELETE_MANIFEST_REF);
+
+ ManifestFile manifest = writeManifest(format, UNPARTITIONED_TYPE, files);
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION, files);
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ List<TrackedFile> read = read(builder);
- List<TrackedFile> read = read(manifest, UNPARTITIONED_SPECS);
assertThat(read)
.allSatisfy(
file ->
assertThat(file.tracking().manifestLocation()).isEqualTo(manifest.path()));
- assertThat(read).extracting(file ->
file.tracking().manifestPos()).containsExactly(0L, 1L, 2L);
+ assertThat(read)
+ .extracting(file -> file.tracking().manifestPos())
+ .containsExactly(0L, 1L, 2L, 3L);
}
- @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"));
+ @Test
+ public void projectionFullByDefault() {
+ Types.StructType readSchema =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG)
+ .filter(Expressions.equal("id", 5)) // does not cause stats to be
filtered
+ .build()
+ .readSchema()
+ .asStruct();
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION, files);
+ Types.StructType expected =
+ TypeUtil.replaceFieldTypes(
+ TrackedFile.schema(ID_PARTITIONED_TYPE, STATS_TYPE),
+ ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.BASE_TYPE))
+ .asStruct();
- V4ManifestReader.Builder builder =
- V4ManifestReader.builder(manifest, io, UNPARTITIONED_SPECS,
TABLE_LOCATION);
- 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();
- }
+ assertThat(readSchema)
+ .as("No projection configuration should project the full table
manifest schema")
+ .isEqualTo(expected);
}
- 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()))));
+ @Test
+ public void projectionDependsOnMetricsConfig() {
+ MetricsConfig metricsWithoutID =
+ MetricsConfig.from(
+ ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"id", "none"),
+ TABLE_SCHEMA,
+ null);
+
+ Types.StructType readSchema =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(metricsWithoutID)
+ .build()
+ .readSchema()
+ .asStruct();
+
+ Types.StructType expected =
+ TypeUtil.replaceFieldTypes(
+ TrackedFile.schema(ID_PARTITIONED_TYPE, DATA_ONLY_STATS_TYPE),
+ ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.BASE_TYPE))
+ .asStruct();
+
+ assertThat(readSchema)
+ .as("Scan planning configuration should automatically prune tracking
and stats")
Review Comment:
This test isn't using `forScanPlanning()` — the description looks copied
from the scan-planning cases below. Something like "metrics config should omit
stats for columns configured as none"?
##########
core/src/main/java/org/apache/iceberg/types/RestoreColumns.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.schema.SchemaWithPartnerVisitor;
+
+public class RestoreColumns extends SchemaWithPartnerVisitor<Type, Type> {
+ /**
+ * Restores columns from a base schema in a projection that may not include
them.
+ *
+ * @param base a base schema, from which the projection was produced
+ * @param projection a projection of the base schema
+ * @param fieldIds a set of field IDs to add to the projection if they are
missing
+ * @return an updated projection with the fields restored
+ */
+ public static Schema restore(Schema base, Schema projection, Set<Integer>
fieldIds) {
+ return SchemaWithPartnerVisitor.visit(
+ base, projection.asStruct(), new RestoreColumns(fieldIds),
FieldIdAccessors.get())
+ .asStructType()
+ .asSchema();
+ }
+
+ private final Set<Integer> restoredFields;
+
+ private RestoreColumns(Set<Integer> fieldsToRestore) {
+ this.restoredFields = fieldsToRestore;
+ }
+
+ @Override
+ public Type schema(Schema schema, Type partner, Type structResult) {
+ return structResult;
+ }
+
+ @Override
+ public Type struct(Types.StructType struct, Type partner, List<Type>
fieldResults) {
+ List<Types.NestedField> fields = struct.fields();
+
+ boolean hasFields = false;
+ List<Types.NestedField> newFields = Lists.newArrayList();
+ for (int i = 0; i < fields.size(); i += 1) {
+ Type newType = fieldResults.get(i);
+ if (newType != null) {
+ hasFields = true;
+
newFields.add(Types.NestedField.from(fields.get(i)).ofType(newType).build());
+ }
+ }
+
+ if (hasFields) {
+ return Types.StructType.of(newFields);
+ }
+
+ return partner;
+ }
+
+ @Override
+ public Type field(Types.NestedField field, Type projection, Type
fieldResult) {
+ if (restoredFields.contains(field.fieldId())) {
+ return field.type();
+ }
+
+ if (fieldResult != null) {
+ return fieldResult;
+ }
+
+ return projection;
+ }
+
+ @Override
+ public Type list(Types.ListType list, Type projection, Type elementResult) {
+ if (restoredFields.contains(list.elementId())) {
+ // replace the original element type with the original, by returning the
original list
Review Comment:
> replace the original element type with the original
this comment reads weird. drop the first `original` word?
##########
core/src/main/java/org/apache/iceberg/types/RestoreColumns.java:
##########
@@ -0,0 +1,170 @@
+/*
+ * 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.Set;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.schema.SchemaWithPartnerVisitor;
+
+public class RestoreColumns extends SchemaWithPartnerVisitor<Type, Type> {
+ /**
+ * Restores columns from a base schema in a projection that may not include
them.
+ *
+ * @param base a base schema, from which the projection was produced
+ * @param projection a projection of the base schema
+ * @param fieldIds a set of field IDs to add to the projection if they are
missing
+ * @return an updated projection with the fields restored
+ */
+ public static Schema restore(Schema base, Schema projection, Set<Integer>
fieldIds) {
Review Comment:
nit: This is a public entry point — worth `Preconditions.checkArgument` for
null `base`, `projection`, and `fieldIds`.
##########
core/src/main/java/org/apache/iceberg/TrackedFile.java:
##########
@@ -131,6 +127,24 @@ private static Type typeOrUnknown(Types.StructType
structType) {
return structType.fields().isEmpty() ? Types.UnknownType.get() :
structType;
}
+ /**
+ * 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(fields(partitionType, contentStatsType));
+ }
+
+ static Schema readSchema(Types.StructType partitionType, Types.StructType
contentStatsType) {
+ List<Types.NestedField> nonEmptyFields =
+ fields(partitionType, contentStatsType).stream()
+ .filter(field -> field.type().typeId() != Type.TypeID.UNKNOWN)
Review Comment:
just to confirm that we only have `UNKNOWN` type for top-level fields,
right? more specifically partitionType and contentType.
##########
core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java:
##########
@@ -44,486 +45,700 @@
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.apache.iceberg.util.LocationUtil;
import org.junit.jupiter.api.Named;
import org.junit.jupiter.api.Test;
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;
+import org.mockito.Mockito;
class TestV4ManifestReader {
+ private static final ManifestFile UNREAD_MANIFEST_FILE =
Mockito.mock(ManifestFile.class);
+
+ static {
+ Mockito.when(UNREAD_MANIFEST_FILE.path())
+ .thenReturn(FileFormat.PARQUET.addExtension("manifest"));
+ Mockito.when(UNREAD_MANIFEST_FILE.formatVersion()).thenReturn(4);
+
Mockito.when(UNREAD_MANIFEST_FILE.content()).thenReturn(ManifestContent.DATA);
+ }
+
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 String TABLE_LOCATION = "s3://bucket/db/table";
private static final DeletionVector DV = dv("s3://bucket/dv.puffin");
+ private static final Tracking ADDED_TRACKING =
TrackingBuilder.added(SNAPSHOT_ID).build();
+
+ private static final ManifestInfo MANIFEST_INFO =
+ new ManifestInfoStruct(49, 51, 0, 0, 4_900L, 5_100L, 0L, 0L, 1L, null,
null);
+
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 Schema LOCATION_ONLY_SCHEMA = new
Schema(TrackedFile.LOCATION);
+
+ private static final PartitionSpec ID_PARTITIONED =
+
PartitionSpec.builderFor(TABLE_SCHEMA).withSpecId(1).identity("id").build();
+ private static final Types.StructType ID_PARTITIONED_TYPE =
ID_PARTITIONED.partitionType();
private static final Map<Integer, PartitionSpec> ID_PARTITIONING_SPECS =
- ImmutableMap.of(ID_PARTITIONING.specId(), ID_PARTITIONING);
+ ImmutableMap.of(ID_PARTITIONED.specId(), ID_PARTITIONED);
+
+ private static final Types.StructType UNPARTITIONED_TYPE =
Types.StructType.of();
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);
+ private static final MetricsConfig METRICS_CONFIG =
+ MetricsConfig.from(ImmutableMap.of(), TABLE_SCHEMA, null);
+ private static final Types.StructType STATS_TYPE =
+ StatsUtil.statsWriteSchema(TABLE_SCHEMA, METRICS_CONFIG);
+ private static final Types.StructType ID_ONLY_STATS_TYPE =
+ Types.StructType.of(STATS_TYPE.field("id"));
+ private static final Types.StructType DATA_ONLY_STATS_TYPE =
+ Types.StructType.of(STATS_TYPE.field("data"));
+
+ private static final FieldStatsStruct<Integer> ID_STATS =
+ new FieldStatsStruct<>(
+ STATS_TYPE.fieldType("id").asStructType(), 0, 99, true,
RECORD_COUNT, 0, 0, null);
+ private static final FieldStatsStruct<String> DATA_STATS =
+ new FieldStatsStruct<>(
+ STATS_TYPE.fieldType("data").asStructType(), "a", "z", false,
RECORD_COUNT, 20, 0, null);
+ private static final ContentStatsStruct CONTENT_STATS = new
ContentStatsStruct(STATS_TYPE);
+
+ static {
+ CONTENT_STATS.setStats(1, ID_STATS);
+ CONTENT_STATS.setStats(2, DATA_STATS);
+ }
- // 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.
Locations are stored
- // relative to the table location (the default), so the reader resolves them
against the table
- 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 Comparator<TrackedFile> FILE_COMPARATOR =
+ V4TestComparators.trackedFileStatusOnly(ID_PARTITIONED_TYPE);
+ private static final Comparator<TrackedFile> UNPARTITIONED_FILE_COMPARATOR =
+ V4TestComparators.trackedFileStatusOnly(UNPARTITIONED_TYPE);
+
+ // shared data files: FILE_A is in partition id=1, FILE_B in partition id=2
+ private static final TrackedFile UNPARTITIONED_FILE =
+ unpartitionedFileWithoutStats("s3://bucket/table/unpartitioned.parquet");
+ private static final TrackedFile FILE_A =
+
idPartitionedDataFileWithoutStats("s3://bucket/table/id=1/file-a.parquet",
idPartition(1));
+ private static final TrackedFile FILE_B =
+
idPartitionedDataFileWithoutStats("s3://bucket/table/id=2/file-b.parquet",
idPartition(2));
+ private static final TrackedFile EQ_DELETES_A =
+ idPartitionedDeleteFileWithoutStats(
+ "s3://bucket/table/id=1/eq-deletes-a.parquet", idPartition(1));
+ private static final TrackedFile EQ_DELETES_B =
+ idPartitionedDeleteFileWithoutStats(
+ "s3://bucket/table/id=2/eq-deletes-b.parquet", idPartition(2));
private static final TrackedFile DATA_MANIFEST_REF =
- manifestRef(FileContent.DATA_MANIFEST, "data-leaf.parquet");
+ manifestRef(FileContent.DATA_MANIFEST,
"s3://bucket/table/data-leaf.parquet");
private static final TrackedFile DELETE_MANIFEST_REF =
- manifestRef(FileContent.DELETE_MANIFEST, "delete-leaf.parquet");
+ manifestRef(FileContent.DELETE_MANIFEST,
"s3://bucket/table/delete-leaf.parquet");
+
+ private static final List<FileFormat> MANIFEST_FORMATS =
+ ImmutableList.of(FileFormat.AVRO, FileFormat.PARQUET);
- private final InMemoryFileIO io = new InMemoryFileIO();
+ private static final InMemoryFileIO IO = new InMemoryFileIO();
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
- public void readsWrittenFile(FileFormat format) throws IOException {
+ public void readDataFile(FileFormat format) throws IOException {
TrackedFile file =
new TrackedFileStruct(
- addedTracking(),
+ ADDED_TRACKING,
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
+ ID_PARTITIONED.specId(),
+ idPartition(7),
+ CONTENT_STATS,
+ SortOrder.unsorted().orderId(),
DV,
- null,
- ByteBuffer.wrap(new byte[] {1, 2, 3}),
+ null, // manifest info
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
ImmutableList.of(50L, 100L),
- null);
+ null); // equality field IDs
- ManifestFile manifest = writeManifest(format, ID_PARTITION_TYPE,
ImmutableList.of(file));
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
- TrackedFile actual = Iterables.getOnlyElement(read(manifest,
ID_PARTITIONING_SPECS));
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ TrackedFile actual = readOne(builder);
- // 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);
+ assertThat(actual).usingComparator(FILE_COMPARATOR).isEqualTo(file);
}
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
- public void equalityDeleteRoundTrip(FileFormat format) throws IOException {
+ public void readForScanPlanningDoesNotCopyStats(FileFormat format) throws
IOException {
+ TrackedFile file =
+ new TrackedFileStruct(
+ ADDED_TRACKING,
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ "s3://bucket/data/file.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ ID_PARTITIONED.specId(),
+ idPartition(7),
+ CONTENT_STATS,
+ SortOrder.unsorted().orderId(),
+ DV,
+ null, // manifest info
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
+ ImmutableList.of(50L, 100L),
+ null); // equality field IDs
+
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .forScanPlanning()
+ .metricsConfig(METRICS_CONFIG);
+ TrackedFile actual = readOne(builder);
+
+
assertThat(actual).usingComparator(FILE_COMPARATOR).isEqualTo(file.copyWithoutStats());
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void readForScanPlanningCopiesRequestedStats(FileFormat format)
throws IOException {
+ int idFieldId = TABLE_SCHEMA.findField("id").fieldId();
+ TrackedFile file =
+ new TrackedFileStruct(
+ ADDED_TRACKING,
+ FileContent.DATA,
+ FORMAT_VERSION_V4,
+ "s3://bucket/data/file.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ ID_PARTITIONED.specId(),
+ idPartition(7),
+ CONTENT_STATS,
+ SortOrder.unsorted().orderId(),
+ DV,
+ null, // manifest info
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
+ ImmutableList.of(50L, 100L),
+ null); // equality field IDs
+
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, file);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .forScanPlanning()
+ .metricsConfig(METRICS_CONFIG)
+ .projectStats(idFieldId);
+ TrackedFile actual = readOne(builder);
+
+ assertThat(actual)
+ .usingComparator(FILE_COMPARATOR)
+ .isEqualTo(file.copyWithStats(Set.of(idFieldId)));
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void readEqualityDelete(FileFormat format) throws IOException {
TrackedFile delete =
new TrackedFileStruct(
- addedTracking(),
+ ADDED_TRACKING,
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,
+ ID_PARTITIONED.specId(),
+ idPartition(7),
+ CONTENT_STATS,
+ SortOrder.unsorted().orderId(),
+ null, // dv
+ null, // manifest info
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
+ null, // split offsets
ImmutableList.of(1, 2));
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(delete));
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE, delete);
- TrackedFile actual = Iterables.getOnlyElement(read(manifest,
UNPARTITIONED_SPECS));
- assertThat(actual.contentType()).isEqualTo(FileContent.EQUALITY_DELETES);
- assertThat(actual.equalityIds()).containsExactly(1, 2);
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ TrackedFile actual = readOne(builder);
+
+ assertThat(actual).usingComparator(FILE_COMPARATOR).isEqualTo(delete);
}
@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"));
-
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION, files);
-
- try (V4ManifestReader reader =
- V4ManifestReader.builder(manifest, io, UNPARTITIONED_SPECS,
TABLE_LOCATION).build()) {
- assertThat(reader)
- .extracting(file -> file.tracking().status())
- .containsExactly(EntryStatus.ADDED, EntryStatus.EXISTING,
EntryStatus.MODIFIED);
- }
+ public void readManifestFile(FileFormat format) throws IOException {
+ TrackedFile manifestRef =
+ new TrackedFileStruct(
+ ADDED_TRACKING,
+ FileContent.DATA_MANIFEST,
+ FORMAT_VERSION_V4,
+ "s3://bucket/leaf-manifest.parquet",
+ FileFormat.PARQUET,
+ RECORD_COUNT,
+ FILE_SIZE_IN_BYTES,
+ null, // spec id
+ null, // partition
+ CONTENT_STATS,
+ null, // sort order id
+ null, // dv
+ MANIFEST_INFO,
+ ByteBuffer.wrap(new byte[] {1, 2, 3}), // key metadata
+ null, // split offsets
+ ImmutableList.of(1, 2));
- try (V4ManifestReader reader =
- V4ManifestReader.builder(manifest, io, UNPARTITIONED_SPECS,
TABLE_LOCATION)
- .includeAll()
- .build()) {
- assertThat(reader)
- .extracting(file -> file.tracking().status())
- .containsExactly(
- EntryStatus.ADDED,
- EntryStatus.EXISTING,
- EntryStatus.MODIFIED,
- EntryStatus.DELETED,
- EntryStatus.REPLACED);
- }
+ ManifestFile manifest = writeManifest(format, ID_PARTITIONED_TYPE,
manifestRef);
+
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ TrackedFile actual = readOne(builder);
+
+ assertThat(actual).usingComparator(FILE_COMPARATOR).isEqualTo(manifestRef);
}
@ParameterizedTest
@FieldSource("MANIFEST_FORMATS")
- public void manifestLocationAndPosition(FileFormat format) throws
IOException {
+ public void statusFilter(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));
+ unpartitionedFileWithStatus(EntryStatus.ADDED,
"s3://bucket/added.parquet"),
+ unpartitionedFileWithStatus(EntryStatus.MODIFIED,
"s3://bucket/modified.parquet"),
+ unpartitionedFileWithStatus(EntryStatus.DELETED,
"s3://bucket/deleted.parquet"),
+ unpartitionedFileWithStatus(EntryStatus.EXISTING,
"s3://bucket/existing.parquet"),
+ unpartitionedFileWithStatus(EntryStatus.REPLACED,
"s3://bucket/replaced.parquet"));
+
+ ManifestFile manifest = writeManifest(format, UNPARTITIONED_TYPE, files);
+
+ List<TrackedFile> liveFiles =
+ read(
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .metricsConfig(METRICS_CONFIG));
+ assertThat(liveFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactly(files.get(0), files.get(1), files.get(3));
+
+ List<TrackedFile> allFiles =
+ read(
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
UNPARTITIONED_SPECS)
+ .metricsConfig(METRICS_CONFIG)
+ .includeAll());
+ assertThat(allFiles)
+ .usingComparatorForType(FILE_COMPARATOR, TrackedFile.class)
+ .containsExactlyElementsOf(files);
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ public void inheritanceManifestLocationAndPosition(FileFormat format) throws
IOException {
+ List<TrackedFile> files =
+ ImmutableList.of(FILE_A, FILE_B, DATA_MANIFEST_REF,
DELETE_MANIFEST_REF);
+
+ ManifestFile manifest = writeManifest(format, UNPARTITIONED_TYPE, files);
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION, files);
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(manifest, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
+ List<TrackedFile> read = read(builder);
- List<TrackedFile> read = read(manifest, UNPARTITIONED_SPECS);
assertThat(read)
.allSatisfy(
file ->
assertThat(file.tracking().manifestLocation()).isEqualTo(manifest.path()));
- assertThat(read).extracting(file ->
file.tracking().manifestPos()).containsExactly(0L, 1L, 2L);
+ assertThat(read)
+ .extracting(file -> file.tracking().manifestPos())
+ .containsExactly(0L, 1L, 2L, 3L);
}
- @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"));
+ @Test
+ public void projectionFullByDefault() {
+ Types.StructType readSchema =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG)
+ .filter(Expressions.equal("id", 5)) // does not cause stats to be
filtered
+ .build()
+ .readSchema()
+ .asStruct();
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION, files);
+ Types.StructType expected =
+ TypeUtil.replaceFieldTypes(
+ TrackedFile.schema(ID_PARTITIONED_TYPE, STATS_TYPE),
+ ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.BASE_TYPE))
+ .asStruct();
- V4ManifestReader.Builder builder =
- V4ManifestReader.builder(manifest, io, UNPARTITIONED_SPECS,
TABLE_LOCATION);
- 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();
- }
+ assertThat(readSchema)
+ .as("No projection configuration should project the full table
manifest schema")
+ .isEqualTo(expected);
}
- 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()))));
+ @Test
+ public void projectionDependsOnMetricsConfig() {
+ MetricsConfig metricsWithoutID =
+ MetricsConfig.from(
+ ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"id", "none"),
+ TABLE_SCHEMA,
+ null);
+
+ Types.StructType readSchema =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(metricsWithoutID)
+ .build()
+ .readSchema()
+ .asStruct();
+
+ Types.StructType expected =
+ TypeUtil.replaceFieldTypes(
+ TrackedFile.schema(ID_PARTITIONED_TYPE, DATA_ONLY_STATS_TYPE),
+ ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.BASE_TYPE))
+ .asStruct();
+
+ assertThat(readSchema)
+ .as("Scan planning configuration should automatically prune tracking
and stats")
+ .isEqualTo(expected);
}
+ @Test
+ public void projectionForScanPlanning() {
+ Types.StructType readSchema =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .forScanPlanning()
+ .filter(Expressions.equal("id", 5))
+ .metricsConfig(METRICS_CONFIG)
+ .build()
+ .readSchema()
+ .asStruct();
+
+ Types.StructType expected =
+ TypeUtil.replaceFieldTypes(
+ TrackedFile.schema(ID_PARTITIONED_TYPE, ID_ONLY_STATS_TYPE),
+ ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.SCAN_TYPE))
+ .asStruct();
+
+ assertThat(readSchema)
+ .as("Scan planning configuration should automatically prune tracking
and stats")
+ .isEqualTo(expected);
+ }
+
+ @Test
+ public void projectionForScanPlanningOverridesMetricsConfig() {
+ MetricsConfig metricsWithoutID =
+ MetricsConfig.from(
+ ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"id", "none"),
+ TABLE_SCHEMA,
+ null);
+
+ Types.StructType readSchema =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .forScanPlanning()
+ .filter(Expressions.equal("id", 5))
+ .metricsConfig(metricsWithoutID)
+ .build()
+ .readSchema()
+ .asStruct();
+
+ Types.StructType expected =
+ TypeUtil.replaceFieldTypes(
+ TrackedFile.schema(ID_PARTITIONED_TYPE, ID_ONLY_STATS_TYPE),
+ ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.SCAN_TYPE))
+ .asStruct();
+
+ assertThat(readSchema)
+ .as("Scan planning configuration should automatically prune tracking
and stats")
+ .isEqualTo(expected);
+ }
+
+ @Test
+ public void projectionForScanPlanningIncludesRequestedStatsMetricsConfig() {
+ MetricsConfig metricsWithoutID =
+ MetricsConfig.from(
+ ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"id", "none"),
+ TABLE_SCHEMA,
+ null);
+
+ Types.StructType readSchema =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .forScanPlanning()
+ .filter(Expressions.equal("id", 5))
+ .projectStats(TABLE_SCHEMA.findField("data").fieldId())
+ .metricsConfig(metricsWithoutID)
+ .build()
+ .readSchema()
+ .asStruct();
+
+ Types.StructType expected =
+ TypeUtil.replaceFieldTypes(
+ TrackedFile.schema(ID_PARTITIONED_TYPE, STATS_TYPE),
+ ImmutableMap.of(TrackedFile.TRACKING.fieldId(),
TrackingStruct.SCAN_TYPE))
+ .asStruct();
+
+ assertThat(readSchema)
+ .as("Scan planning configuration should automatically prune tracking
and stats")
+ .isEqualTo(expected);
+ }
+
+ private static final List<Named<Consumer<V4ManifestReader.Builder>>>
PROJECTION_CASES =
+ ImmutableList.of(
+ Named.of("select", builder -> builder.select("location")),
+ Named.of(
+ "case-insensitive select",
+ builder -> builder.select("LOCATION").caseSensitive(false)),
+ Named.of("project", builder ->
builder.project(LOCATION_ONLY_SCHEMA)));
+
@ParameterizedTest
- @FieldSource("MANIFEST_FORMATS")
- public void rowFilterForcesRecordCount(FileFormat format) throws IOException
{
- TrackedFile file = dataFile("s3://bucket/file.parquet",
EMPTY_PARTITION_DATA);
+ @FieldSource("PROJECTION_CASES")
+ public void projectionCustomization(Consumer<V4ManifestReader.Builder>
config) {
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .metricsConfig(METRICS_CONFIG);
- ManifestFile manifest = writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(file));
+ config.accept(builder);
- // 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, io, UNPARTITIONED_SPECS,
TABLE_LOCATION)
- .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);
- }
+ Types.StructType readSchema = builder.build().readSchema().asStruct();
+
+ assertThat(readSchema.field(TrackedFile.LOCATION.fieldId()))
+ .as("Projected field 'location' should be present")
+ .isEqualTo(TrackedFile.LOCATION);
+
+ assertThat(readSchema.field(TrackedFile.KEY_METADATA.fieldId()))
+ .as("Unselected, non-required field should be omitted")
+ .isNull();
+
+ assertThat(readSchema.field(TrackedFile.RECORD_COUNT.fieldId()))
+ .as("Required field 'record_count' should be present")
+ .isEqualTo(TrackedFile.RECORD_COUNT);
+
+ assertThat(readSchema.field(TrackedFile.PARTITION_ID))
+ .as("Partition is not automatically projected")
+ .isNull();
+
+ assertThat(readSchema.field(TrackedFile.CONTENT_STATS_ID))
+ .as("Content stats are not automatically projected")
+ .isNull();
+ }
+
+ @ParameterizedTest
+ @FieldSource("PROJECTION_CASES")
+ public void
projectionCustomizationWithFilter(Consumer<V4ManifestReader.Builder> config) {
+ V4ManifestReader.Builder builder =
+ V4ManifestReader.builder(UNREAD_MANIFEST_FILE, IO, TABLE_SCHEMA,
ID_PARTITIONING_SPECS)
+ .filter(Expressions.equal("id", 5))
+ .metricsConfig(METRICS_CONFIG);
+
+ config.accept(builder);
+
+ Types.StructType readSchema = builder.build().readSchema().asStruct();
+
+ assertThat(readSchema.field(TrackedFile.LOCATION.fieldId()))
+ .as("Projected field 'location' should be present")
+ .isEqualTo(TrackedFile.LOCATION);
+
+ assertThat(readSchema.field(TrackedFile.KEY_METADATA.fieldId()))
+ .as("Unselected, non-required field should be omitted")
+ .isNull();
+
+ assertThat(readSchema.field(TrackedFile.RECORD_COUNT.fieldId()))
+ .as("Required field 'record_count' should be present")
+ .isEqualTo(TrackedFile.RECORD_COUNT);
+
+ assertThat(readSchema.field(TrackedFile.PARTITION_ID).type())
+ .as("Partition is projected for filtering")
+ .isEqualTo(ID_PARTITIONED_TYPE);
+
+ assertThat(readSchema.field(TrackedFile.CONTENT_STATS_ID).type())
+ .as("Content stats are projected for filtering")
+ .isEqualTo(ID_ONLY_STATS_TYPE);
}
@Test
public void projectionModesAreMutuallyExclusive() {
Review Comment:
This covers select/project/`forScanPlanning()` pairing, but `projectStats`
vs `select`/`project` is also rejected (`Cannot use projectStats with select` /
`... with project`) and isn't tested here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
