rdblue commented on code in PR #17541: URL: https://github.com/apache/iceberg/pull/17541#discussion_r4169653084
########## core/src/test/java/org/apache/iceberg/TestFilePlanner.java: ########## @@ -0,0 +1,780 @@ +/* + * 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 static org.assertj.core.api.Assertions.tuple; + +import java.io.IOException; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.Set; +import java.util.function.UnaryOperator; +import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.inmemory.InMemoryFileIO; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.CloseableIterator; +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.io.SeekableInputStream; +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.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.LocationUtil; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.FieldSource; + +class TestFilePlanner { + private static final long SNAPSHOT_ID = 42L; + private static final long SEQUENCE_NUMBER = 7L; + private static final long FIRST_ROW_ID = 1000L; + private static final int WRITER_FORMAT_VERSION = 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 String DV_LOCATION = "s3://bucket/db/table/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 EMPTY_PARTITION = Types.StructType.of(); + private static final PartitionData EMPTY_PARTITION_DATA = new PartitionData(EMPTY_PARTITION); + private static final Map<Integer, PartitionSpec> UNPARTITIONED_SPECS = + ImmutableMap.of(PartitionSpec.unpartitioned().specId(), PartitionSpec.unpartitioned()); + + // a table that evolved from unpartitioned (spec 0) to identity(id) (spec 1) + private static final PartitionSpec UNPARTITIONED_WITH_SCHEMA = + PartitionSpec.builderFor(TABLE_SCHEMA).withSpecId(0).build(); + private static final PartitionSpec BY_ID_SPEC = + PartitionSpec.builderFor(TABLE_SCHEMA).withSpecId(1).identity("id").build(); + private static final Map<Integer, PartitionSpec> MIXED_SPECS = + ImmutableMap.of( + UNPARTITIONED_WITH_SCHEMA.specId(), UNPARTITIONED_WITH_SCHEMA, + BY_ID_SPEC.specId(), BY_ID_SPEC); + + 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 final InMemoryFileIO fileIO = new InMemoryFileIO(); + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void rootWithOnlyDataFiles(FileFormat format) throws IOException { + InputFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFile("a.parquet", EMPTY_PARTITION_DATA), + dataFile("b.parquet", EMPTY_PARTITION_DATA))); + + List<FileScanTask> tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .hasSize(2) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder(resolved("a.parquet"), resolved("b.parquet")); + assertThat(tasks).allSatisfy(task -> assertThat(task.deletes()).isEmpty()); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void rootWithOnlyLeafManifests(FileFormat format) throws IOException { + InputFile leafA = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf-a.parquet", EMPTY_PARTITION_DATA))); + InputFile leafB = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf-b.parquet", EMPTY_PARTITION_DATA))); + InputFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataManifest(leafA.location()), dataManifest(leafB.location()))); + + List<FileScanTask> tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .hasSize(2) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder(resolved("leaf-a.parquet"), resolved("leaf-b.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void rootWithMixedDataAndLeafManifests(FileFormat format) throws IOException { + InputFile leaf1 = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf-file-1.parquet", EMPTY_PARTITION_DATA))); + InputFile leaf2 = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf-file-2.parquet", EMPTY_PARTITION_DATA))); + InputFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFile("root-data-file.parquet", EMPTY_PARTITION_DATA), + dataManifest(leaf1.location()), + dataManifest(leaf2.location()))); + + List<FileScanTask> tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .hasSize(3) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder( + resolved("root-data-file.parquet"), + resolved("leaf-file-1.parquet"), + resolved("leaf-file-2.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterSkipsAndMatchesRootDataFilesAndManifests(FileFormat format) throws IOException { + InputFile root = filterableRoot(format); + + List<FileScanTask> tasks = + plan(root, UNPARTITIONED_SPECS, planner -> planner.filterData(Expressions.equal("id", 50))); + + assertThat(tasks) + .extracting(task -> task.file().location()) + .as( + "matched root data file and the file in the matched leaf survive; pruned data file and leaf do not") + .containsExactlyInAnyOrder(resolved("root-keep.parquet"), resolved("leaf-file.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterSkipsAndMatchesLeafDataFiles(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats("leaf-keep.parquet", idStats(0, 99)), + dataFileWithStats("leaf-prune.parquet", idStats(100, 199)))); + InputFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(dataManifest(leaf.location(), 2))); + + List<FileScanTask> tasks = + plan(root, UNPARTITIONED_SPECS, planner -> planner.filterData(Expressions.equal("id", 50))); + + assertThat(Iterables.getOnlyElement(tasks).file().location()) + .as("only the leaf file whose stats match the filter survives") + .isEqualTo(resolved("leaf-keep.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterHonorsCaseSensitivity(FileFormat format) throws IOException { + InputFile root = filterableRoot(format); + + List<FileScanTask> tasks = + plan( + root, + UNPARTITIONED_SPECS, + planner -> planner.filterData(Expressions.equal("ID", 50)).caseSensitive(false)); + assertThat(tasks) + .extracting(task -> task.file().location()) + .as("a case-insensitive filter resolves \"ID\" to \"id\" everywhere the filter is applied") + .containsExactlyInAnyOrder(resolved("root-keep.parquet"), resolved("leaf-file.parquet")); + + assertThatThrownBy( + () -> + plan( + root, + UNPARTITIONED_SPECS, + planner -> planner.filterData(Expressions.equal("ID", 50)))) + .as("the default case-sensitive filter cannot resolve \"ID\"") + .isInstanceOf(ValidationException.class) + .hasMessageContaining("Cannot find field 'ID'"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void colocatedDvAttachedAsPositionDeleteFile(FileFormat format) throws IOException { Review Comment: This is validating the conversion from data file to task. I'd like to see a prefix in the test name to signal this and group other tests of task conversion together. Style: in Iceberg, acronyms are left capitalized except in specific cases. I think I've pointed out before that "DV" should always be "DV" in method or variable names, never "Dv". -- 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]
