diff --git a/core/src/main/java/org/apache/iceberg/FilePlanner.java b/core/src/main/java/org/apache/iceberg/FilePlanner.java new file mode 100644 index 000000000000..78c2206b4793 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/FilePlanner.java @@ -0,0 +1,207 @@ +/* + * 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.io.IOException; +import java.io.UncheckedIOException; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.expressions.ResidualEvaluator; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.metrics.ScanMetrics; +import org.apache.iceberg.metrics.ScanMetricsUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +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.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.util.ParallelIterable; +import org.apache.iceberg.util.ThreadPools; + +/** + * Plans {@link FileScanTask}s from a V4 root manifest. + * + *

Emits a task for each live {@code DATA} entry and expands {@code DATA_MANIFEST} entries into + * their leaf manifests. A data file's colocated deletion vector is attached to its task as a {@link + * DeleteFile}. + */ +class FilePlanner { + private static final DeleteFile[] NO_DELETES = new DeleteFile[0]; + + private final FileIO io; + private final ManifestFile root; + private final Schema tableSchema; + private final Map specsById; + private final Map taskContextsBySpec = Maps.newHashMap(); + + private String tableLocation = null; + private Expression dataFilter = Expressions.alwaysTrue(); + private boolean ignoreResiduals = false; + private boolean caseSensitive = true; + private ScanMetrics scanMetrics = ScanMetrics.noop(); + private ExecutorService executorService = ThreadPools.getWorkerPool(); + + FilePlanner( + FileIO io, ManifestFile root, Schema tableSchema, Map specsById) { + Preconditions.checkArgument(io != null, "Invalid file IO: null"); + Preconditions.checkArgument(root != null, "Invalid root manifest: null"); + Preconditions.checkArgument(tableSchema != null, "Invalid table schema: null"); + Preconditions.checkArgument(specsById != null, "Invalid specs by ID: null"); + this.io = io; + this.root = root; + this.tableSchema = tableSchema; + this.specsById = ImmutableMap.copyOf(specsById); + } + + FilePlanner tableLocation(String newTableLocation) { + Preconditions.checkArgument(newTableLocation != null, "Invalid table location: null"); + this.tableLocation = newTableLocation; + return this; + } + + FilePlanner filterData(Expression expr) { + Preconditions.checkArgument(expr != null, "Invalid filter: null"); + this.dataFilter = expr; + return this; + } + + FilePlanner ignoreResiduals() { + this.ignoreResiduals = true; + return this; + } + + FilePlanner caseSensitive(boolean newCaseSensitive) { + this.caseSensitive = newCaseSensitive; + return this; + } + + FilePlanner scanMetrics(ScanMetrics newScanMetrics) { + Preconditions.checkArgument(newScanMetrics != null, "Invalid scan metrics: null"); + this.scanMetrics = newScanMetrics; + return this; + } + + FilePlanner planWith(ExecutorService newExecutorService) { + this.executorService = newExecutorService; + return this; + } + + CloseableIterable planFiles() { + List rootDataFiles = Lists.newArrayList(); + List leafManifests = Lists.newArrayList(); + + try (CloseableIterable rootEntries = reader(root)) { + for (TrackedFile entry : rootEntries) { + switch (entry.contentType()) { + case DATA: + rootDataFiles.add(TrackedFileAdapters.asDataFile(entry, specsById)); + break; + case DATA_MANIFEST: + leafManifests.add(TrackedFileAdapters.asManifestFile(entry)); + break; + case DELETE_MANIFEST: + throw new UnsupportedOperationException("v3 and earlier deletes are not yet supported"); + default: + throw new UnsupportedOperationException( + "Unsupported file type in root manifest: " + entry.contentType()); + } + } + } catch (IOException e) { + throw new UncheckedIOException("Failed to close root manifest: " + root.path(), e); + } + + List> leafPlanTasks = Lists.newArrayList(); + for (ManifestFile leaf : leafManifests) { + leafPlanTasks.add(reader(leaf)); + } + + CloseableIterable leafFiles = + executorService != null && leafManifests.size() > 1 + ? new ParallelIterable<>(leafPlanTasks, executorService) + : CloseableIterable.concat(leafPlanTasks); + + CloseableIterable leafDataFiles = + CloseableIterable.transform( + leafFiles, file -> TrackedFileAdapters.asDataFile(file, specsById)); + + CloseableIterable dataFiles = + CloseableIterable.concat( + ImmutableList.of(CloseableIterable.of(rootDataFiles), leafDataFiles)); + + return CloseableIterable.transform(dataFiles, this::createTask); + } + + private CloseableIterable reader(ManifestFile manifest) { + V4ManifestReader.Builder builder = + V4ManifestReader.builder(manifest, io, tableSchema, specsById) + .forScanPlanning() + .filter(dataFilter) + .caseSensitive(caseSensitive) + .scanMetrics(scanMetrics); + if (tableLocation != null) { + builder.tableLocation(tableLocation); + } + + return builder.build(); + } + + private FileScanTask createTask(DataFile dataFile) { + TaskContext context = + taskContextsBySpec.computeIfAbsent(dataFile.specId(), this::newTaskContext); + + DeleteFile[] deletes = NO_DELETES; + if (dataFile.deletionVector() != null) { + TrackedFile tracked = TrackedFileAdapters.asTrackedFile(dataFile); + DeleteFile dv = TrackedFileAdapters.asDVDeleteFile(tracked, specsById); + scanMetrics.dvs().increment(); + deletes = new DeleteFile[] {dv}; + } + + ScanMetricsUtil.fileTask(scanMetrics, dataFile, deletes); + + return new BaseFileScanTask( + dataFile, deletes, context.schemaAsString, context.specAsString, context.residuals); + } + + private TaskContext newTaskContext(int specId) { + PartitionSpec spec = specsById.get(specId); + Expression filter = ignoreResiduals ? Expressions.alwaysTrue() : dataFilter; + return new TaskContext( + SchemaParser.toJson(spec.schema()), + PartitionSpecParser.toJson(spec), + ResidualEvaluator.of(spec, filter, caseSensitive)); + } + + /** Per-spec task inputs computed once and shared across all files of a spec. */ + private static class TaskContext { + private final String schemaAsString; + private final String specAsString; + private final ResidualEvaluator residuals; + + private TaskContext(String schemaAsString, String specAsString, ResidualEvaluator residuals) { + this.schemaAsString = schemaAsString; + this.specAsString = specAsString; + this.residuals = residuals; + } + } +} diff --git a/core/src/main/java/org/apache/iceberg/RootManifestFile.java b/core/src/main/java/org/apache/iceberg/RootManifestFile.java new file mode 100644 index 000000000000..5c7bf8bbc9b6 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/RootManifestFile.java @@ -0,0 +1,148 @@ +/* + * 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.Arrays; +import java.util.List; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.util.ByteBuffers; + +/** A {@link ManifestFile} for the root manifest of an adaptive metadata tree. */ +class RootManifestFile implements ManifestFile { + private static final int FORMAT_VERSION = 4; + + private final String location; + private final long length; + private final long snapshotId; + private final long sequenceNumber; + private final Long firstRowId; + private final byte[] keyMetadata; + + RootManifestFile( + String location, + long length, + long snapshotId, + long sequenceNumber, + Long firstRowId, + ByteBuffer keyMetadata) { + Preconditions.checkArgument(location != null, "Invalid location: null"); + this.location = location; + this.length = length; + this.snapshotId = snapshotId; + this.sequenceNumber = sequenceNumber; + this.firstRowId = firstRowId; + this.keyMetadata = ByteBuffers.toByteArray(keyMetadata); + } + + @Override + public String path() { + return location; + } + + @Override + public long length() { + return length; + } + + @Override + public int partitionSpecId() { + throw new UnsupportedOperationException("Root manifest has no partition spec"); + } + + @Override + public ManifestContent content() { + return ManifestContent.DATA; + } + + @Override + public long sequenceNumber() { + return sequenceNumber; + } + + @Override + public long minSequenceNumber() { + throw new UnsupportedOperationException("Root manifest has no minimum sequence number"); + } + + @Override + public Long snapshotId() { + return snapshotId; + } + + @Override + public Integer addedFilesCount() { + throw new UnsupportedOperationException("Root manifest has no added files count"); + } + + @Override + public Long addedRowsCount() { + throw new UnsupportedOperationException("Root manifest has no added rows count"); + } + + @Override + public Integer existingFilesCount() { + throw new UnsupportedOperationException("Root manifest has no existing files count"); + } + + @Override + public Long existingRowsCount() { + throw new UnsupportedOperationException("Root manifest has no existing rows count"); + } + + @Override + public Integer deletedFilesCount() { + throw new UnsupportedOperationException("Root manifest has no deleted files count"); + } + + @Override + public Long deletedRowsCount() { + throw new UnsupportedOperationException("Root manifest has no deleted rows count"); + } + + @Override + public List partitions() { + throw new UnsupportedOperationException("Root manifest has no partition summaries"); + } + + @Override + public ByteBuffer keyMetadata() { + return keyMetadata == null ? null : ByteBuffer.wrap(keyMetadata); + } + + @Override + public Long firstRowId() { + return firstRowId; + } + + @Override + public int formatVersion() { + return FORMAT_VERSION; + } + + @Override + public ManifestFile copy() { + ByteBuffer keyMetadataCopy = + keyMetadata == null + ? null + : ByteBuffer.wrap(Arrays.copyOf(keyMetadata, keyMetadata.length)); + return new RootManifestFile( + location, length, snapshotId, sequenceNumber, firstRowId, keyMetadataCopy); + } +} diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java index 5acb03b9089c..7936b2ef31b9 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java @@ -38,6 +38,10 @@ static DataFile asDataFile(TrackedFile file, Map specsBy return new TrackedDataFile(file, resolveSpecId(file, specsById)); } + static TrackedFile asTrackedFile(DataFile file) { + return ((TrackedDataFile) file).file(); + } + static DeleteFile asDVDeleteFile(TrackedFile file, Map specsById) { Preconditions.checkArgument( file.contentType() == FileContent.DATA, diff --git a/core/src/test/java/org/apache/iceberg/TestFilePlanner.java b/core/src/test/java/org/apache/iceberg/TestFilePlanner.java new file mode 100644 index 000000000000..2edeb58dc10c --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestFilePlanner.java @@ -0,0 +1,758 @@ +/* + * 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.V4TestHelpers.ADDED_TRACKING; +import static org.apache.iceberg.V4TestHelpers.FILE_SIZE_IN_BYTES; +import static org.apache.iceberg.V4TestHelpers.RECORD_COUNT; +import static org.apache.iceberg.V4TestHelpers.SNAPSHOT_ID; +import static org.apache.iceberg.V4TestHelpers.dataFile; +import static org.apache.iceberg.V4TestHelpers.deletionVector; +import static org.apache.iceberg.V4TestHelpers.manifestRef; +import static org.apache.iceberg.V4TestHelpers.manifestRefWithStats; +import static org.apache.iceberg.V4TestHelpers.writeTrackedFiles; +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.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.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.FieldSource; + +class TestFilePlanner { + private static final long SEQUENCE_NUMBER = 7L; + private static final long FIRST_ROW_ID = 1000L; + private static final String TABLE_LOCATION = "s3://bucket/db/table"; + private static final DeletionVector DV = + deletionVector("s3://bucket/db/table/dv.puffin", 100L, 50L, 5L); + + private static final String FILE_A = "s3://bucket/db/table/file-a.parquet"; + private static final String FILE_B = "s3://bucket/db/table/file-b.parquet"; + private static final String FILE_C = "s3://bucket/db/table/file-c.parquet"; + private static final String FILE_D = "s3://bucket/db/table/file-d.parquet"; + private static final String PARTITIONED_FILE = "s3://bucket/db/table/partitioned.parquet"; + private static final String UNPARTITIONED_FILE = "s3://bucket/db/table/unpartitioned.parquet"; + + private static final Schema TABLE_SCHEMA = + new Schema( + optional(1, "id", Types.IntegerType.get()), optional(2, "data", Types.StringType.get())); + private static final int UNPARTITIONED_SPEC_ID = PartitionSpec.unpartitioned().specId(); + 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 UNPARTITIONED_SPECS = + ImmutableMap.of(UNPARTITIONED_SPEC_ID, PartitionSpec.unpartitioned()); + + 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 MIXED_SPECS = + ImmutableMap.of( + UNPARTITIONED_WITH_SCHEMA.specId(), UNPARTITIONED_WITH_SCHEMA, + BY_ID_SPEC.specId(), BY_ID_SPEC); + + private static final List 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 { + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(unpartitionedDataFile(FILE_A), unpartitionedDataFile(FILE_B))); + + List tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .hasSize(2) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder(FILE_A, FILE_B); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void rootWithOnlyLeafManifests(FileFormat format) throws IOException { + ManifestFile leafA = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(unpartitionedDataFile(FILE_A))); + ManifestFile leafB = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(unpartitionedDataFile(FILE_B))); + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataManifest(leafA.path()), dataManifest(leafB.path()))); + + List tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .hasSize(2) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder(FILE_A, FILE_B); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void rootWithMixedDataAndLeafManifests(FileFormat format) throws IOException { + ManifestFile leaf1 = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(unpartitionedDataFile(FILE_C))); + ManifestFile leaf2 = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(unpartitionedDataFile(FILE_D))); + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + unpartitionedDataFile(FILE_A), + dataManifest(leaf1.path()), + dataManifest(leaf2.path()))); + + List tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .hasSize(3) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder(FILE_A, FILE_C, FILE_D); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterSkipsAndMatchesRootDataFiles(FileFormat format) throws IOException { + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats(FILE_A, idStats(0, 99)), + dataFileWithStats(FILE_B, idStats(100, 199)))); + + List tasks = + plan(root, UNPARTITIONED_SPECS, planner -> planner.filterData(Expressions.equal("id", 50))); + + assertThat(tasks) + .extracting(task -> task.file().location()) + .as("only the root data file whose stats match the filter survives") + .containsExactlyInAnyOrder(FILE_A); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterSkipsAndMatchesRootDataFilesAndManifests(FileFormat format) throws IOException { + ManifestFile matchedLeaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats(FILE_C, idStats(0, 99)), + dataFileWithStats(FILE_D, idStats(100, 199)))); + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats(FILE_A, idStats(0, 99)), + dataFileWithStats(FILE_B, idStats(100, 199)), + dataManifestWithStats(matchedLeaf.path(), idStats(0, 199)), + dataManifestWithStats( + "s3://bucket/db/table/pruned-leaf.parquet", idStats(100, 199)))); + + List tasks = + plan(root, UNPARTITIONED_SPECS, planner -> planner.filterData(Expressions.equal("id", 50))); + + assertThat(tasks) + .extracting(task -> task.file().location()) + .as( + "the matched root data file and the file in the matched leaf survive; the pruned data file and leaf do not") + .containsExactlyInAnyOrder(FILE_A, FILE_C); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterSkipsAndMatchesLeafDataFiles(FileFormat format) throws IOException { + ManifestFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats(FILE_C, idStats(0, 99)), + dataFileWithStats(FILE_D, idStats(100, 199)))); + ManifestFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(dataManifest(leaf.path(), 2))); + + List 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(FILE_C); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterHonorsCaseSensitivity(FileFormat format) throws IOException { + ManifestFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats(FILE_C, idStats(0, 99)), + dataFileWithStats(FILE_D, idStats(100, 199)))); + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats(FILE_A, idStats(0, 99)), + dataFileWithStats(FILE_B, idStats(100, 199)), + dataManifestWithStats(leaf.path(), idStats(0, 199)))); + + List 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\" in both the root and leaf readers") + .containsExactlyInAnyOrder(FILE_A, FILE_C); + + 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 taskAttachesColocatedDV(FileFormat format) throws IOException { + TrackedFile fileWithDV = + unpartitionedDataFileWithDV("s3://bucket/db/table/with-dv.parquet", DV); + TrackedFile fileWithoutDV = unpartitionedDataFile(FILE_A); + ManifestFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(fileWithDV, fileWithoutDV)); + + List tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .hasSize(2) + .satisfiesExactlyInAnyOrder( + task -> { + assertThat(task.file().location()).isEqualTo("s3://bucket/db/table/with-dv.parquet"); + assertThat(task.deletes()) + .as("the colocated DV is attached as the file's only delete") + .hasSize(1) + .allSatisfy(delete -> assertThat(delete.location()).isEqualTo(DV.location())); + }, + task -> { + assertThat(task.file().location()).isEqualTo(FILE_A); + assertThat(task.deletes()) + .as("a file without a colocated DV has no deletes") + .isEmpty(); + }); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void taskExposesFileSchemaAndSpec(FileFormat format) throws IOException { + Types.StructType unionType = Partitioning.unionPartitionTypes(MIXED_SPECS.values()); + ManifestFile root = + writeManifest( + format, + unionType, + ImmutableList.of( + dataFile( + PARTITIONED_FILE, BY_ID_SPEC.specId(), idPartition(unionType, 1), null, null), + dataFile( + UNPARTITIONED_FILE, + UNPARTITIONED_WITH_SCHEMA.specId(), + idPartition(unionType, null), + null, + null))); + + List tasks = plan(root, MIXED_SPECS); + + assertThat(tasks) + .hasSize(2) + .satisfiesExactlyInAnyOrder( + task -> { + assertThat(task.file().location()).isEqualTo(PARTITIONED_FILE); + assertThat(task.spec()) + .as("the task carries the full spec its file was written with") + .isEqualTo(BY_ID_SPEC); + assertThat(task.schema().asStruct()).isEqualTo(TABLE_SCHEMA.asStruct()); + }, + task -> { + assertThat(task.file().location()).isEqualTo(UNPARTITIONED_FILE); + assertThat(task.spec()) + .as("the task carries the full spec its file was written with") + .isEqualTo(UNPARTITIONED_WITH_SCHEMA); + assertThat(task.schema().asStruct()).isEqualTo(TABLE_SCHEMA.asStruct()); + }); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void residualForMixedPartitionedAndUnpartitioned(FileFormat format) throws IOException { + Types.StructType unionType = Partitioning.unionPartitionTypes(MIXED_SPECS.values()); + ManifestFile root = + writeManifest( + format, + unionType, + ImmutableList.of( + dataFile( + PARTITIONED_FILE, BY_ID_SPEC.specId(), idPartition(unionType, 1), null, null), + dataFile( + UNPARTITIONED_FILE, + UNPARTITIONED_WITH_SCHEMA.specId(), + idPartition(unionType, null), + null, + null))); + + Expression filter = Expressions.and(Expressions.equal("id", 1), Expressions.equal("data", "x")); + List tasks = plan(root, MIXED_SPECS, planner -> planner.filterData(filter)); + + assertThat(tasks) + .extracting(task -> task.file().location(), task -> task.residual().toString()) + .as( + "the id predicate drops out of the partitioned file's residual but not the unpartitioned") + .containsExactlyInAnyOrder( + tuple(PARTITIONED_FILE, Expressions.equal("data", "x").toString()), + tuple(UNPARTITIONED_FILE, filter.toString())); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void ignoreResidualsProducesAlwaysTrueResidual(FileFormat format) throws IOException { + Types.StructType unionType = Partitioning.unionPartitionTypes(MIXED_SPECS.values()); + ManifestFile root = + writeManifest( + format, + unionType, + ImmutableList.of( + dataFile( + PARTITIONED_FILE, BY_ID_SPEC.specId(), idPartition(unionType, 1), null, null), + dataFile( + UNPARTITIONED_FILE, + UNPARTITIONED_WITH_SCHEMA.specId(), + idPartition(unionType, null), + null, + null))); + + Expression filter = Expressions.and(Expressions.equal("id", 1), Expressions.equal("data", "x")); + List tasks = + plan(root, MIXED_SPECS, planner -> planner.filterData(filter).ignoreResiduals()); + + assertThat(tasks) + .as("ignoreResiduals forces an alwaysTrue residual for every file regardless of spec") + .hasSize(2) + .allSatisfy(task -> assertThat(task.residual()).isEqualTo(Expressions.alwaysTrue())); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void tableLocationForwardedToRootAndLeafReaders(FileFormat format) throws IOException { + ManifestFile leaf = + writeManifest( + format, EMPTY_PARTITION, ImmutableList.of(unpartitionedDataFile("leaf-data.parquet"))); + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + unpartitionedDataFile("root-data.parquet"), dataManifest(leaf.path()))); + + List tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .extracting(task -> task.file().location()) + .as("relative paths resolve against the table location in both the root and leaf readers") + .containsExactlyInAnyOrder( + "s3://bucket/db/table/root-data.parquet", "s3://bucket/db/table/leaf-data.parquet"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void closingWithoutIteratingOpensOnlyRoot(FileFormat format) throws IOException { + ManifestFile leaf = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(unpartitionedDataFile(FILE_A))); + ManifestFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(dataManifest(leaf.path()))); + + RecordingFileIO recordingIO = new RecordingFileIO(fileIO); + FilePlanner planner = + new FilePlanner(recordingIO, root, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .tableLocation(TABLE_LOCATION); + planner.planFiles().close(); + + assertThat(recordingIO.opened(root.path())).as("the root is read eagerly").isTrue(); + assertThat(recordingIO.opened(leaf.path())) + .as("leaf readers open lazily, so closing without iterating leaves them unopened") + .isFalse(); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void leafOpenedOnlyWhenPrecedingLeafExhausted(FileFormat format) throws IOException { + ManifestFile leaf1 = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(unpartitionedDataFile(FILE_A), unpartitionedDataFile(FILE_B))); + ManifestFile leaf2 = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(unpartitionedDataFile(FILE_C))); + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataManifest(leaf1.path()), dataManifest(leaf2.path()))); + + RecordingFileIO recordingIO = new RecordingFileIO(fileIO); + FilePlanner planner = + new FilePlanner(recordingIO, root, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .tableLocation(TABLE_LOCATION) + .planWith(null); + + try (CloseableIterable plan = planner.planFiles()) { + CloseableIterator tasks = plan.iterator(); + assertThat(tasks.next().file().location()).isEqualTo(FILE_A); + assertThat(recordingIO.opened(leaf1.path())) + .as("leaf1 opens to produce the first task") + .isTrue(); + assertThat(recordingIO.opened(leaf2.path())) + .as("leaf2 stays closed until leaf1 is exhausted") + .isFalse(); + + assertThat(tasks.next().file().location()).isEqualTo(FILE_B); + assertThat(recordingIO.opened(leaf2.path())) + .as("reading the last file in leaf1 does not open leaf2") + .isFalse(); + + assertThat(tasks.next().file().location()).isEqualTo(FILE_C); + assertThat(recordingIO.opened(leaf2.path())).as("leaf2 opens once leaf1 is drained").isTrue(); + assertThat(tasks).isExhausted(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void deleteManifestInRootUnsupported(FileFormat format) throws IOException { + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(deleteManifest("s3://bucket/db/table/deletes.avro"))); + + FilePlanner planner = + new FilePlanner(fileIO, root, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .tableLocation(TABLE_LOCATION); + assertThatThrownBy(() -> Lists.newArrayList(planner.planFiles())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("v3 and earlier deletes are not yet supported"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void v3ManifestLeafUnsupported(FileFormat format) throws IOException { + String leafLocation = "s3://bucket/db/table/v3-leaf." + format.name().toLowerCase(Locale.ROOT); + TrackedFile v3Leaf = + new TrackedFileStruct( + ADDED_TRACKING, + FileContent.DATA_MANIFEST, + 3, // format version the v4 reader rejects + leafLocation, + FileFormat.fromFileName(leafLocation), + RECORD_COUNT, + FILE_SIZE_IN_BYTES, + null, // specId + null, // partition + null, // contentStats + null, // sortOrderId + null, // deletionVector + dataManifestInfo(1), + null, // keyMetadata + null, // splitOffsets + null); // equalityIds + ManifestFile root = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(v3Leaf)); + + FilePlanner planner = + new FilePlanner(fileIO, root, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .tableLocation(TABLE_LOCATION); + assertThatThrownBy(() -> Lists.newArrayList(planner.planFiles())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Cannot read manifest with format version 3: only 4 is supported"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void scanMetricsForComplexFilteredPlan(FileFormat format) throws IOException { + ManifestFile matchedLeaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats(FILE_C, idStats(0, 99)), + dataFileWithStats(FILE_D, idStats(100, 199)))); + ManifestFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStats(FILE_A, idStats(0, 99), DV), + dataFileWithStats(FILE_B, idStats(100, 199)), + dataManifestWithStats(matchedLeaf.path(), idStats(0, 199)), + dataManifestWithStats( + "s3://bucket/db/table/pruned-leaf.parquet", idStats(100, 199)))); + + ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); + List tasks = + plan( + root, + UNPARTITIONED_SPECS, + planner -> planner.filterData(Expressions.equal("id", 50)).scanMetrics(metrics)); + + assertThat(tasks) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder(FILE_A, FILE_C); + assertThat(metrics.resultDataFiles().value()).isEqualTo(2L); + assertThat(metrics.totalFileSizeInBytes().value()).isEqualTo(2 * FILE_SIZE_IN_BYTES); + assertThat(metrics.scannedDataManifests().value()) + .as("the root and the matched leaf are scanned") + .isEqualTo(2L); + assertThat(metrics.skippedDataManifests().value()) + .as("the pruned leaf ref is skipped") + .isEqualTo(1L); + assertThat(metrics.skippedDataFiles().value()) + .as("the pruned root and leaf data files are skipped") + .isEqualTo(2L); + assertThat(metrics.resultDeleteFiles().value()) + .as("only the colocated DV contributes a delete file") + .isEqualTo(1L); + assertThat(metrics.totalDeleteFileSizeInBytes().value()).isEqualTo(DV.sizeInBytes()); + assertThat(metrics.dvs().value()).isEqualTo(1L); + assertThat(metrics.indexedDeleteFiles().value()) + .as("colocated DVs are not indexed delete files") + .isEqualTo(0L); + } + + private List plan(ManifestFile root, Map specsById) + throws IOException { + return plan(root, specsById, UnaryOperator.identity()); + } + + private List plan( + ManifestFile root, + Map specsById, + UnaryOperator configure) + throws IOException { + FilePlanner planner = + configure.apply( + new FilePlanner(fileIO, root, TABLE_SCHEMA, specsById).tableLocation(TABLE_LOCATION)); + try (CloseableIterable tasks = planner.planFiles()) { + return Lists.newArrayList(tasks); + } + } + + private static TrackedFile unpartitionedDataFile(String location) { + return dataFile(location, UNPARTITIONED_SPEC_ID, EMPTY_PARTITION_DATA); + } + + private static TrackedFile unpartitionedDataFileWithDV(String location, DeletionVector dv) { + return dataFile(location, UNPARTITIONED_SPEC_ID, EMPTY_PARTITION_DATA, null, dv); + } + + private static TrackedFile dataFileWithStats(String location, ContentStats stats) { + return dataFileWithStats(location, stats, null); + } + + private static TrackedFile dataFileWithStats( + String location, ContentStats stats, DeletionVector dv) { + return dataFile(location, UNPARTITIONED_SPEC_ID, EMPTY_PARTITION_DATA, stats, dv); + } + + private static TrackedFile dataManifest(String location) { + return dataManifest(location, 1); + } + + private static TrackedFile dataManifest(String location, int addedFiles) { + return manifestRef(FileContent.DATA_MANIFEST, location, dataManifestInfo(addedFiles)); + } + + private static TrackedFile dataManifestWithStats(String location, ContentStats stats) { + return manifestRefWithStats(FileContent.DATA_MANIFEST, location, stats, dataManifestInfo(1)); + } + + private static ContentStats idStats(int lower, int upper) { + ContentStatsStruct stats = new ContentStatsStruct(STATS_TYPE); + stats.setStats( + TABLE_SCHEMA.findField("id").fieldId(), + new FieldStatsStruct<>( + STATS_TYPE.fieldType("id").asStructType(), + lower, + upper, + true, // tightBounds + RECORD_COUNT, + 0, // nullValueCount + 0, // nanValueCount + null)); // avgValueSize + return stats; + } + + private static ManifestInfo dataManifestInfo(int addedFiles) { + return ManifestInfoStruct.builder() + .addedFilesCount(addedFiles) + .existingFilesCount(0) + .deletedFilesCount(0) + .replacedFilesCount(0) + .modifiedFilesCount(0) + .addedRowsCount(addedFiles * RECORD_COUNT) + .existingRowsCount(0) + .deletedRowsCount(0) + .replacedRowsCount(0) + .modifiedRowsCount(0) + .minSequenceNumber(0L) + .build(); + } + + private static TrackedFile deleteManifest(String location) { + return manifestRef(FileContent.DELETE_MANIFEST, location, null); + } + + private static PartitionData idPartition(Types.StructType unionType, Integer id) { + PartitionData partition = new PartitionData(unionType); + partition.set(0, id); + return partition; + } + + private ManifestFile writeManifest( + FileFormat format, Types.StructType partitionType, Iterable files) + throws IOException { + OutputFile out = + fileIO.newOutputFile( + TABLE_LOCATION + + "/metadata/manifest-" + + System.nanoTime() + + "." + + format.name().toLowerCase(Locale.ROOT)); + InputFile written = + writeTrackedFiles(out, format, partitionType, STATS_TYPE, files).toInputFile(); + return new RootManifestFile( + written.location(), + written.getLength(), + SNAPSHOT_ID, + SEQUENCE_NUMBER, + FIRST_ROW_ID, + /* keyMetadata= */ null); + } + + private static class RecordingFileIO implements FileIO { + private final FileIO delegate; + private final Set openedPaths = Sets.newHashSet(); + + private RecordingFileIO(FileIO delegate) { + this.delegate = delegate; + } + + private boolean opened(String location) { + return openedPaths.contains(location); + } + + @Override + public InputFile newInputFile(String path) { + return new RecordingInputFile(delegate.newInputFile(path), openedPaths); + } + + @Override + public OutputFile newOutputFile(String path) { + return delegate.newOutputFile(path); + } + + @Override + public void deleteFile(String path) { + delegate.deleteFile(path); + } + } + + private static class RecordingInputFile implements InputFile { + private final InputFile delegate; + private final Set openedPaths; + + private RecordingInputFile(InputFile delegate, Set openedPaths) { + this.delegate = delegate; + this.openedPaths = openedPaths; + } + + @Override + public long getLength() { + return delegate.getLength(); + } + + @Override + public SeekableInputStream newStream() { + openedPaths.add(delegate.location()); + return delegate.newStream(); + } + + @Override + public String location() { + return delegate.location(); + } + + @Override + public boolean exists() { + return delegate.exists(); + } + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestRootManifestFile.java b/core/src/test/java/org/apache/iceberg/TestRootManifestFile.java new file mode 100644 index 000000000000..b69e6a21db2b --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestRootManifestFile.java @@ -0,0 +1,110 @@ +/* + * 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.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.nio.ByteBuffer; +import org.junit.jupiter.api.Test; + +class TestRootManifestFile { + 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 long LENGTH = 1024L; + private static final String LOCATION = "s3://bucket/db/table/metadata/root.parquet"; + + @Test + void fieldAccess() { + ByteBuffer keyMetadata = ByteBuffer.wrap(new byte[] {9, 8, 7}); + RootManifestFile root = + new RootManifestFile( + LOCATION, LENGTH, SNAPSHOT_ID, SEQUENCE_NUMBER, FIRST_ROW_ID, keyMetadata); + + assertThat(root.path()).isEqualTo(LOCATION); + assertThat(root.length()).isEqualTo(LENGTH); + assertThat(root.snapshotId()).isEqualTo(SNAPSHOT_ID); + assertThat(root.sequenceNumber()).isEqualTo(SEQUENCE_NUMBER); + assertThat(root.firstRowId()).isEqualTo(FIRST_ROW_ID); + assertThat(root.keyMetadata()).isEqualTo(keyMetadata); + } + + @Test + void defaultedFields() { + RootManifestFile root = + new RootManifestFile(LOCATION, LENGTH, SNAPSHOT_ID, SEQUENCE_NUMBER, FIRST_ROW_ID, null); + + assertThat(root.content()).isEqualTo(ManifestContent.DATA); + assertThat(root.formatVersion()).isEqualTo(4); + } + + @Test + void copy() { + ByteBuffer keyMetadata = ByteBuffer.wrap(new byte[] {9, 8, 7}); + RootManifestFile original = + new RootManifestFile( + LOCATION, LENGTH, SNAPSHOT_ID, SEQUENCE_NUMBER, FIRST_ROW_ID, keyMetadata); + ManifestFile copy = original.copy(); + + assertThat(copy.path()).isEqualTo(LOCATION); + assertThat(copy.length()).isEqualTo(LENGTH); + assertThat(copy.snapshotId()).isEqualTo(SNAPSHOT_ID); + assertThat(copy.sequenceNumber()).isEqualTo(SEQUENCE_NUMBER); + assertThat(copy.firstRowId()).isEqualTo(FIRST_ROW_ID); + assertThat(copy.keyMetadata()).isEqualTo(keyMetadata); + assertThat(copy.keyMetadata().array()) + .as("copy should not share the key metadata backing array") + .isNotSameAs(original.keyMetadata().array()); + } + + @Test + void unsupportedAccessorsThrow() { + RootManifestFile root = + new RootManifestFile(LOCATION, LENGTH, SNAPSHOT_ID, SEQUENCE_NUMBER, FIRST_ROW_ID, null); + + assertThatThrownBy(root::partitionSpecId) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no partition spec"); + assertThatThrownBy(root::minSequenceNumber) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no minimum sequence number"); + assertThatThrownBy(root::addedFilesCount) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no added files count"); + assertThatThrownBy(root::addedRowsCount) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no added rows count"); + assertThatThrownBy(root::existingFilesCount) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no existing files count"); + assertThatThrownBy(root::existingRowsCount) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no existing rows count"); + assertThatThrownBy(root::deletedFilesCount) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no deleted files count"); + assertThatThrownBy(root::deletedRowsCount) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no deleted rows count"); + assertThatThrownBy(root::partitions) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Root manifest has no partition summaries"); + } +} diff --git a/core/src/test/java/org/apache/iceberg/V4TestHelpers.java b/core/src/test/java/org/apache/iceberg/V4TestHelpers.java index 1b922b576ca8..6d79bcef377a 100644 --- a/core/src/test/java/org/apache/iceberg/V4TestHelpers.java +++ b/core/src/test/java/org/apache/iceberg/V4TestHelpers.java @@ -43,6 +43,15 @@ static TrackedFile dataFile(String location, Integer specId, PartitionData parti return dataFile(ADDED_TRACKING, location, specId, partition, null, null); } + static TrackedFile dataFile( + String location, + Integer specId, + PartitionData partition, + ContentStats stats, + DeletionVector dv) { + return dataFile(ADDED_TRACKING, location, specId, partition, stats, dv); + } + static TrackedFile dataFileWithStats(String location, ContentStats stats) { return dataFile(ADDED_TRACKING, location, null, null, stats, null); } @@ -165,8 +174,18 @@ static OutputFile writeTrackedFiles( Types.StructType statsType, Iterable files) throws IOException { - Schema writeSchema = TrackedFile.schema(partitionType, statsType); OutputFile out = io.newOutputFile(format.addExtension("manifest." + System.nanoTime())); + return writeTrackedFiles(out, format, partitionType, statsType, files); + } + + static OutputFile writeTrackedFiles( + OutputFile out, + FileFormat format, + Types.StructType partitionType, + Types.StructType statsType, + Iterable files) + throws IOException { + Schema writeSchema = TrackedFile.schema(partitionType, statsType); try (FileAppender appender = InternalData.write(format, out).schema(writeSchema).named("tracked_file").build()) { for (TrackedFile file : files) {