dramaticlly commented on code in PR #16958: URL: https://github.com/apache/iceberg/pull/16958#discussion_r3647457321
########## core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java: ########## @@ -0,0 +1,926 @@ +/* + * 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.List; +import java.util.Locale; +import java.util.Map; +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.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.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.FieldSource; + +public 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 int SORT_ORDER_ID = 1; + private static final String DV_LOCATION = "s3://bucket/dv.puffin"; + private static final long DV_OFFSET = 100L; + private static final long DV_SIZE_IN_BYTES = 50L; + private static final long DV_CARDINALITY = 5L; + + private static final Schema TABLE_SCHEMA = + new Schema( + optional(1, "id", Types.IntegerType.get()), optional(2, "data", Types.StringType.get())); + private static final PartitionSpec SPEC = + PartitionSpec.builderFor(TABLE_SCHEMA).identity("id").build(); + private static final Types.StructType PARTITION_TYPE = SPEC.partitionType(); + private static final Types.StructType EMPTY_PARTITION = Types.StructType.of(); + private static final PartitionData EMPTY_PARTITION_DATA = new PartitionData(EMPTY_PARTITION); + private static final Map<Integer, PartitionSpec> PARTITIONED_SPECS = + ImmutableMap.of(SPEC.specId(), SPEC); + private static final Map<Integer, PartitionSpec> UNPARTITIONED_SPECS = + ImmutableMap.of(PartitionSpec.unpartitioned().specId(), PartitionSpec.unpartitioned()); + + private static final List<FileFormat> FORMATS = + ImmutableList.of(FileFormat.AVRO, FileFormat.PARQUET); + + // row_position is appended after the tracking schema fields by the reader + private static final int MANIFEST_POS_ORDINAL = Tracking.schema().fields().size(); + + @TempDir private Path tempDir; + + private final FileIO fileIO = new TestTables.LocalFileIO(); + + @ParameterizedTest + @FieldSource("FORMATS") + public void testRoundTrip(FileFormat format) throws IOException { + DeletionVector dv = deletionVector(DV_LOCATION, DV_OFFSET, DV_SIZE_IN_BYTES, DV_CARDINALITY); + + TrackedFile file = + new TrackedFileStruct( + addedTracking(), + FileContent.DATA, + FORMAT_VERSION_V4, + "s3://bucket/data/file.parquet", + FileFormat.PARQUET, + RECORD_COUNT, + FILE_SIZE_IN_BYTES, + SPEC.specId(), + partition(7), + null, + SORT_ORDER_ID, + dv, + null, + ByteBuffer.wrap(new byte[] {1, 2, 3}), + ImmutableList.of(50L, 100L), + null); + + InputFile manifest = writeManifest(format, PARTITION_TYPE, ImmutableList.of(file)); + + List<TrackedFile> read = read(manifest, PARTITIONED_SPECS); + assertThat(read).hasSize(1); + TrackedFile actual = read.get(0); + + // the reader fills row_position (manifestPos) and manifestLocation, which the written file + // does not have; mirror them on the expected file before comparing + TrackingStruct expectedTracking = (TrackingStruct) ((TrackedFileStruct) file).tracking(); + expectedTracking.set(MANIFEST_POS_ORDINAL, 0L); + expectedTracking.setManifestLocation(manifest.location()); + + Types.StructType readType = + TypeUtil.replaceFieldTypes( + TrackedFile.schema(PARTITION_TYPE, Types.StructType.of()), + ImmutableMap.of(TrackedFile.TRACKING.fieldId(), TrackingStruct.BASE_TYPE)) + .asStruct(); + assertThat(Comparators.forType(readType).compare((StructLike) file, (StructLike) actual)) + .isEqualTo(0); + } + + @ParameterizedTest + @FieldSource("FORMATS") + public void testEqualityDeleteRoundTrip(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 = read(manifest, UNPARTITIONED_SPECS).get(0); + assertThat(actual.contentType()).isEqualTo(FileContent.EQUALITY_DELETES); + assertThat(actual.equalityIds()).containsExactly(1, 2); + } + + @ParameterizedTest + @FieldSource("FORMATS") + public void testStatusFiltering(FileFormat format) throws IOException { + List<TrackedFile> files = + ImmutableList.of( + fileWithStatus(EntryStatus.ADDED, "s3://bucket/added.parquet"), + fileWithStatus(EntryStatus.EXISTING, "s3://bucket/existing.parquet"), + fileWithStatus(EntryStatus.MODIFIED, "s3://bucket/modified.parquet"), + fileWithStatus(EntryStatus.DELETED, "s3://bucket/deleted.parquet"), + fileWithStatus(EntryStatus.REPLACED, "s3://bucket/replaced.parquet")); + + InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) { + assertThat(reader) + .extracting(file -> file.tracking().status()) + .containsExactly(EntryStatus.ADDED, EntryStatus.EXISTING, EntryStatus.MODIFIED); + } + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).includeAll().build()) { + assertThat(reader) + .extracting(file -> file.tracking().status()) + .containsExactly( + EntryStatus.ADDED, + EntryStatus.EXISTING, + EntryStatus.MODIFIED, + EntryStatus.DELETED, + EntryStatus.REPLACED); + } + } + + @ParameterizedTest + @FieldSource("FORMATS") + public void testManifestLocationAndPosition(FileFormat format) throws IOException { + List<TrackedFile> files = + ImmutableList.of( + dataFile("s3://bucket/a.parquet", EMPTY_PARTITION_DATA), + dataFile("s3://bucket/b.parquet", EMPTY_PARTITION_DATA), + dataFile("s3://bucket/c.parquet", EMPTY_PARTITION_DATA)); + + InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); + + List<TrackedFile> read = read(manifest, UNPARTITIONED_SPECS); + assertThat(read) + .allSatisfy( + file -> assertThat(file.tracking().manifestLocation()).isEqualTo(manifest.location())); + assertThat(read).extracting(file -> file.tracking().manifestPos()).containsExactly(0L, 1L, 2L); + } + + @ParameterizedTest + @FieldSource("FORMATS") + public void testProjectionRestrictsFields(FileFormat format) throws IOException { + TrackedFile file = + new TrackedFileStruct( + addedTracking(), + FileContent.DATA, + FORMAT_VERSION_V4, + "s3://bucket/file.parquet", + FileFormat.PARQUET, + RECORD_COUNT, + FILE_SIZE_IN_BYTES, + 0, + EMPTY_PARTITION_DATA, + null, + SORT_ORDER_ID, + null, + null, + null, + null, + null); Review Comment: ```suggestion TrackedFile file = dataFile("s3://bucket/file.parquet", EMPTY_PARTITION_DATA); ``` can probably simplify as projected location is part of dataFile helper ########## core/src/main/java/org/apache/iceberg/V4ManifestReader.java: ########## @@ -0,0 +1,304 @@ +/* + * 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.Collection; +import java.util.Map; +import java.util.Set; +import org.apache.iceberg.expressions.Evaluator; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.Projections; +import org.apache.iceberg.io.CloseableGroup; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.CloseableIterator; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.metrics.ScanMetrics; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.types.TypeUtil; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.Pair; +import org.apache.iceberg.util.StructProjection; + +/** Reader that reads a v4+ manifest file as {@link TrackedFile}s. */ +class V4ManifestReader extends CloseableGroup implements CloseableIterable<TrackedFile> { + private final InputFile file; + private final Schema readSchema; + private final boolean includeAll; + private final ScanMetrics scanMetrics; + + // partition filters keyed by spec ID; empty when no partition filter applies + private final Map<Integer, Pair<Evaluator, StructProjection>> partitionFilters; + + private V4ManifestReader( + InputFile file, + Schema readSchema, + Map<Integer, Pair<Evaluator, StructProjection>> partitionFilters, + boolean includeAll, + ScanMetrics scanMetrics) { + this.file = file; + this.readSchema = readSchema; + this.partitionFilters = partitionFilters; + this.includeAll = includeAll; + this.scanMetrics = scanMetrics; + } + + static Builder builder(InputFile file, Map<Integer, PartitionSpec> specsById) { + return new Builder(file, specsById); + } + + /** Returns copies of the tracked files that match this reader's configured filters. */ + @Override + public CloseableIterator<TrackedFile> iterator() { + CloseableIterable<TrackedFile> entries = CloseableIterable.transform(open(), this::prepare); + if (!partitionFilters.isEmpty()) { + // manifests have no partition, so the partition filter cannot apply to them + entries = + CloseableIterable.filter(entries, entry -> isManifest(entry) || matchesPartition(entry)); + } + + if (!includeAll) { + entries = CloseableIterable.filter(entries, entry -> entry.tracking().isLive()); + } + + return CloseableIterable.transform(entries, TrackedFile::copy).iterator(); + } + + private boolean matchesPartition(TrackedFile trackedFile) { + Integer specId = trackedFile.specId(); + if (specId == null) { + // a file without a spec is not partitioned and may match the filter + return true; + } + + Pair<Evaluator, StructProjection> partitionFilter = partitionFilters.get(specId); + if (partitionFilter == null) { + // the row filter does not project to a partition filter for this spec + return true; + } + + Evaluator evaluator = partitionFilter.first(); + StructProjection projection = partitionFilter.second(); + boolean matches = evaluator.eval(projection.wrap(trackedFile.partition())); + if (!matches) { + incrementSkipCount(trackedFile.contentType()); + } + + return matches; + } + + private void incrementSkipCount(FileContent content) { + switch (content) { + case DATA: + scanMetrics.skippedDataFiles().increment(); + break; + case EQUALITY_DELETES: + scanMetrics.skippedDeleteFiles().increment(); + break; + case DATA_MANIFEST: + scanMetrics.skippedDataManifests().increment(); + break; + case DELETE_MANIFEST: + scanMetrics.skippedDeleteManifests().increment(); + break; Review Comment: non-blocking: is there a way to wire this skip count to the actual filter? Currently looks like a dead code due to short circuit logic in `entry -> isManifest(entry) || matchesPartition(entry)` and incrementSkipCount is only called within matchesPartition. ########## core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java: ########## @@ -0,0 +1,926 @@ +/* + * 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.List; +import java.util.Locale; +import java.util.Map; +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.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.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.FieldSource; + +public 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 int SORT_ORDER_ID = 1; + private static final String DV_LOCATION = "s3://bucket/dv.puffin"; + private static final long DV_OFFSET = 100L; + private static final long DV_SIZE_IN_BYTES = 50L; + private static final long DV_CARDINALITY = 5L; + + private static final Schema TABLE_SCHEMA = + new Schema( + optional(1, "id", Types.IntegerType.get()), optional(2, "data", Types.StringType.get())); + private static final PartitionSpec SPEC = + PartitionSpec.builderFor(TABLE_SCHEMA).identity("id").build(); + private static final Types.StructType PARTITION_TYPE = SPEC.partitionType(); + private static final Types.StructType EMPTY_PARTITION = Types.StructType.of(); + private static final PartitionData EMPTY_PARTITION_DATA = new PartitionData(EMPTY_PARTITION); + private static final Map<Integer, PartitionSpec> PARTITIONED_SPECS = + ImmutableMap.of(SPEC.specId(), SPEC); + private static final Map<Integer, PartitionSpec> UNPARTITIONED_SPECS = + ImmutableMap.of(PartitionSpec.unpartitioned().specId(), PartitionSpec.unpartitioned()); + + private static final List<FileFormat> FORMATS = + ImmutableList.of(FileFormat.AVRO, FileFormat.PARQUET); + + // row_position is appended after the tracking schema fields by the reader + private static final int MANIFEST_POS_ORDINAL = Tracking.schema().fields().size(); + + @TempDir private Path tempDir; + + private final FileIO fileIO = new TestTables.LocalFileIO(); + + @ParameterizedTest + @FieldSource("FORMATS") + public void testRoundTrip(FileFormat format) throws IOException { + DeletionVector dv = deletionVector(DV_LOCATION, DV_OFFSET, DV_SIZE_IN_BYTES, DV_CARDINALITY); + + TrackedFile file = + new TrackedFileStruct( + addedTracking(), + FileContent.DATA, + FORMAT_VERSION_V4, + "s3://bucket/data/file.parquet", + FileFormat.PARQUET, + RECORD_COUNT, + FILE_SIZE_IN_BYTES, + SPEC.specId(), + partition(7), + null, + SORT_ORDER_ID, + dv, + null, + ByteBuffer.wrap(new byte[] {1, 2, 3}), + ImmutableList.of(50L, 100L), + null); + + InputFile manifest = writeManifest(format, PARTITION_TYPE, ImmutableList.of(file)); + + List<TrackedFile> read = read(manifest, PARTITIONED_SPECS); + assertThat(read).hasSize(1); + TrackedFile actual = read.get(0); + + // the reader fills row_position (manifestPos) and manifestLocation, which the written file + // does not have; mirror them on the expected file before comparing + TrackingStruct expectedTracking = (TrackingStruct) ((TrackedFileStruct) file).tracking(); + expectedTracking.set(MANIFEST_POS_ORDINAL, 0L); + expectedTracking.setManifestLocation(manifest.location()); + + Types.StructType readType = + TypeUtil.replaceFieldTypes( + TrackedFile.schema(PARTITION_TYPE, Types.StructType.of()), + ImmutableMap.of(TrackedFile.TRACKING.fieldId(), TrackingStruct.BASE_TYPE)) + .asStruct(); + assertThat(Comparators.forType(readType).compare((StructLike) file, (StructLike) actual)) + .isEqualTo(0); + } + + @ParameterizedTest + @FieldSource("FORMATS") + public void testEqualityDeleteRoundTrip(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)); Review Comment: nit: want to check if this warrant a quick helper for `equalityDelete(location, equalityIds)` as it can be used in both `testEqualityDeleteRoundTrip` and `testPartitionFilterCountsSkippedDeleteFiles`. Or maybe combine the 2 tests ? -- 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]
