diff --git a/core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java b/core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java index 027a3d2298f0..05f016999a49 100644 --- a/core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java +++ b/core/src/main/java/org/apache/iceberg/BaseDistributedDataScan.java @@ -35,6 +35,7 @@ import org.apache.iceberg.expressions.Projections; import org.apache.iceberg.expressions.ResidualEvaluator; import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.InputFile; import org.apache.iceberg.metrics.ScanMetricsUtil; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.relocated.com.google.common.collect.Maps; @@ -144,6 +145,10 @@ protected PlanningMode deletePlanningMode() { @Override protected CloseableIterable doPlanFiles() { + if (TableUtil.formatVersion(table()) >= TableMetadata.MIN_FORMAT_VERSION_PARQUET_MANIFESTS) { + return doPlanFilesV4(); + } + Snapshot snapshot = snapshot(); List deleteManifests = findMatchingDeleteManifests(snapshot); @@ -187,6 +192,26 @@ protected CloseableIterable doPlanFiles() { } } + private CloseableIterable doPlanFilesV4() { + Snapshot snapshot = snapshot(); + InputFile rootManifest = table().io().newInputFile(snapshot.manifestListLocation()); + ScanTaskPlanner.Builder planner = + ScanTaskPlanner.builder(table().io(), rootManifest, specs(), table().location()) + .filterData(filter()) + .caseSensitive(isCaseSensitive()) + .scanMetrics(scanMetrics()); + + if (shouldIgnoreResiduals()) { + planner = planner.ignoreResiduals(); + } + + if (shouldPlanWithExecutor()) { + planner = planner.planWith(planExecutor()); + } + + return CloseableIterable.transform(planner.build().planFiles(), task -> (ScanTask) task); + } + @Override public CloseableIterable> planTasks() { return TableScanUtil.planTaskGroups( diff --git a/core/src/main/java/org/apache/iceberg/BaseSnapshot.java b/core/src/main/java/org/apache/iceberg/BaseSnapshot.java index 826b9624c0e6..99860a83e9a5 100644 --- a/core/src/main/java/org/apache/iceberg/BaseSnapshot.java +++ b/core/src/main/java/org/apache/iceberg/BaseSnapshot.java @@ -171,6 +171,14 @@ private void cacheManifests(FileIO fileIO) { throw new IllegalArgumentException("Cannot cache changes: FileIO is null"); } + if (allManifests == null + && manifestListLocation != null + && FileFormat.fromFileName(manifestListLocation) == FileFormat.PARQUET) { + // A v4 flat-tree root manifest stores data entries inline rather than referencing leaf + // manifests, so it exposes no ManifestFiles. Scans plan directly from the root manifest. + this.allManifests = ImmutableList.of(); + } + if (allManifests == null && v1ManifestLocations != null) { // if we have a collection of manifest locations, then we need to load them here allManifests = diff --git a/core/src/main/java/org/apache/iceberg/DataTableScan.java b/core/src/main/java/org/apache/iceberg/DataTableScan.java index dc130c8064fc..4c695f98655a 100644 --- a/core/src/main/java/org/apache/iceberg/DataTableScan.java +++ b/core/src/main/java/org/apache/iceberg/DataTableScan.java @@ -21,6 +21,7 @@ import java.util.List; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.io.InputFile; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; public class DataTableScan extends BaseTableScan { @@ -62,6 +63,34 @@ protected TableScan newRefinedScan(Table table, Schema schema, TableScanContext @Override public CloseableIterable doPlanFiles() { + if (TableUtil.formatVersion(table()) >= TableMetadata.MIN_FORMAT_VERSION_PARQUET_MANIFESTS) { + return doPlanFilesV4(); + } + + return doPlanFilesV3(); + } + + private CloseableIterable doPlanFilesV4() { + Snapshot snapshot = snapshot(); + InputFile rootManifest = table().io().newInputFile(snapshot.manifestListLocation()); + ScanTaskPlanner.Builder planner = + ScanTaskPlanner.builder(table().io(), rootManifest, specs(), table().location()) + .filterData(filter()) + .caseSensitive(isCaseSensitive()) + .scanMetrics(scanMetrics()); + + if (shouldIgnoreResiduals()) { + planner = planner.ignoreResiduals(); + } + + if (shouldPlanWithExecutor()) { + planner = planner.planWith(planExecutor()); + } + + return planner.build().planFiles(); + } + + private CloseableIterable doPlanFilesV3() { Snapshot snapshot = snapshot(); FileIO io = table().io(); diff --git a/core/src/main/java/org/apache/iceberg/ScanTaskPlanner.java b/core/src/main/java/org/apache/iceberg/ScanTaskPlanner.java new file mode 100644 index 000000000000..5ba0845f3e45 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/ScanTaskPlanner.java @@ -0,0 +1,320 @@ +/* + * 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.CloseableGroup; +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.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; + +/** + * 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 entry's colocated deletion vector is attached to its task as a + * {@link DeleteFile}. + * + *

Emitted tasks do not yet carry inherited tracking: leaf and root {@code DATA} entries are not + * assigned the data/file sequence numbers, snapshot id, and first-row-id they inherit from their + * parent, so {@code data_sequence_number}, {@code file_sequence_number}, and row-lineage columns + * ({@code _row_id}, {@code _last_updated_sequence_number}) read null for added files. Inheritance + * must be applied before this planner is wired into a scan or the delete-manifest matching path, + * since delete scoping compares a data file's sequence number against the delete's. + */ +class ScanTaskPlanner { + private static final int FORMAT_VERSION = 4; + private static final DeleteFile[] NO_DELETES = new DeleteFile[0]; + + private final FileIO io; + private final InputFile rootManifest; + private final Map specsById; + private final String tableLocation; + private final Expression dataFilter; + private final boolean ignoreResiduals; + private final boolean caseSensitive; + private final ScanMetrics scanMetrics; + private final ExecutorService executorService; + private final Map taskContextsBySpec = Maps.newConcurrentMap(); + + private ScanTaskPlanner( + FileIO io, + InputFile rootManifest, + Map specsById, + String tableLocation, + Expression dataFilter, + boolean ignoreResiduals, + boolean caseSensitive, + ScanMetrics scanMetrics, + ExecutorService executorService) { + this.io = io; + this.rootManifest = rootManifest; + this.specsById = specsById; + this.tableLocation = tableLocation; + this.dataFilter = dataFilter; + this.ignoreResiduals = ignoreResiduals; + this.caseSensitive = caseSensitive; + this.scanMetrics = scanMetrics; + this.executorService = executorService; + } + + static Builder builder( + FileIO io, + InputFile rootManifest, + Map specsById, + String tableLocation) { + return new Builder(io, rootManifest, specsById, tableLocation); + } + + CloseableIterable planFiles() { + List dataFiles = Lists.newArrayList(); + List leafManifests = Lists.newArrayList(); + + // root is drained into these lists. Leaf references must be buffered so they can be fanned + // out; direct DATA entries are buffered too, bounded by how many data files a tree keeps + // directly in the root. Leaf tasks stay lazy: createLeafTasks opens no reader until iterated. + scanMetrics.scannedDataManifests().increment(); + try (CloseableIterable rootEntries = open(rootManifest)) { + for (TrackedFile entry : rootEntries) { + switch (entry.contentType()) { + case DATA: + dataFiles.add(entry); + break; + case DATA_MANIFEST: + leafManifests.add(entry); + break; + default: + // delete content appears only on upgraded trees + throw new UnsupportedOperationException( + "Cannot plan content type in root manifest: " + entry.contentType()); + } + } + } catch (IOException e) { + throw new UncheckedIOException( + "Failed to close root manifest: " + rootManifest.location(), e); + } + + // root DATA tasks are already in hand, so emit them directly; only leaf expansion, which reads + // each leaf manifest, is worth handing to the parallel backend + CloseableIterable rootTasks = + CloseableIterable.transform(CloseableIterable.withNoopClose(dataFiles), this::createTask); + + List> leafTasks = Lists.newArrayList(); + for (TrackedFile leaf : leafManifests) { + leafTasks.add(createLeafTasks(leaf)); + } + + CloseableIterable expandedLeafTasks = + executorService != null + ? new ParallelIterable<>(leafTasks, executorService) + : CloseableIterable.concat(leafTasks); + + return CloseableIterable.concat(ImmutableList.of(rootTasks, expandedLeafTasks)); + } + + private CloseableIterable createLeafTasks(TrackedFile leaf) { + // an upgraded tree can reference a legacy-format leaf + if (leaf.formatVersion() != FORMAT_VERSION) { + throw new UnsupportedOperationException( + "Cannot expand leaf manifest with format version " + + leaf.formatVersion() + + ": " + + leaf.location()); + } + + if (leaf.manifestInfo() != null && leaf.manifestInfo().dv() != null) { + throw new UnsupportedOperationException( + "Cannot apply manifest deletion vector for leaf manifest: " + leaf.location()); + } + + if (leaf.keyMetadata() != null) { + throw new UnsupportedOperationException( + "Cannot read encrypted leaf manifest: " + leaf.location()); + } + + return new LeafTasks(leaf); + } + + /** A leaf's tasks, with each iteration owning and counting its own reader. */ + private class LeafTasks extends CloseableGroup implements CloseableIterable { + private final TrackedFile leaf; + + private LeafTasks(TrackedFile leaf) { + this.leaf = leaf; + } + + @Override + public CloseableIterator iterator() { + scanMetrics.scannedDataManifests().increment(); + // pass the known leaf size so the reader sizes the read instead of stat-ing the file + CloseableIterable entries = + open(io.newInputFile(leaf.location(), leaf.fileSizeInBytes())); + CloseableIterable tasks = + CloseableIterable.transform(entries, ScanTaskPlanner.this::createTaskFromDataFileEntry); + addCloseable(tasks); + return tasks.iterator(); + } + } + + private FileScanTask createTaskFromDataFileEntry(TrackedFile entry) { + // the tree is at most two levels, so a leaf holds only DATA entries + if (entry.contentType() == FileContent.DATA_MANIFEST) { + throw new IllegalArgumentException( + "Cannot expand a nested manifest in a leaf manifest: " + entry.location()); + } else if (entry.contentType() != FileContent.DATA) { + throw new UnsupportedOperationException( + "Cannot plan content type in leaf manifest: " + entry.contentType()); + } + + return createTask(entry); + } + + private CloseableIterable open(InputFile manifest) { + return V4ManifestReader.builder(manifest, specsById, tableLocation) + .forScanPlanning() + .filter(dataFilter) + .caseSensitive(caseSensitive) + .scanMetrics(scanMetrics) + .build(); + } + + private FileScanTask createTask(TrackedFile trackedFile) { + DataFile dataFile = TrackedFileAdapters.asDataFile(trackedFile, specsById); + TaskContext context = + taskContextsBySpec.computeIfAbsent(dataFile.specId(), this::newTaskContext); + + DeleteFile[] deletes; + if (trackedFile.deletionVector() != null) { + deletes = new DeleteFile[] {TrackedFileAdapters.asDVDeleteFile(trackedFile, specsById)}; + } else { + deletes = NO_DELETES; + } + + 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; + } + } + + static class Builder { + private final FileIO io; + private final InputFile rootManifest; + private final Map specsById; + private final String tableLocation; + private Expression dataFilter = Expressions.alwaysTrue(); + private boolean ignoreResiduals = false; + private boolean caseSensitive = true; + private ScanMetrics scanMetrics = ScanMetrics.noop(); + private ExecutorService executorService = null; + + private Builder( + FileIO io, + InputFile rootManifest, + Map specsById, + String tableLocation) { + Preconditions.checkArgument(io != null, "Invalid file IO: null"); + Preconditions.checkArgument(rootManifest != null, "Invalid root manifest: null"); + Preconditions.checkArgument(specsById != null, "Invalid specs by ID: null"); + Preconditions.checkArgument(tableLocation != null, "Invalid table location: null"); + this.io = io; + this.rootManifest = rootManifest; + this.specsById = ImmutableMap.copyOf(specsById); + this.tableLocation = tableLocation; + } + + /** Narrows the filter used for partition pruning and residual evaluation. */ + Builder filterData(Expression expr) { + Preconditions.checkArgument(expr != null, "Invalid filter: null"); + this.dataFilter = Expressions.and(dataFilter, expr); + return this; + } + + Builder ignoreResiduals() { + this.ignoreResiduals = true; + return this; + } + + Builder caseSensitive(boolean newCaseSensitive) { + this.caseSensitive = newCaseSensitive; + return this; + } + + Builder scanMetrics(ScanMetrics newScanMetrics) { + Preconditions.checkArgument(newScanMetrics != null, "Invalid scan metrics: null"); + this.scanMetrics = newScanMetrics; + return this; + } + + Builder planWith(ExecutorService newExecutorService) { + this.executorService = newExecutorService; + return this; + } + + ScanTaskPlanner build() { + return new ScanTaskPlanner( + io, + rootManifest, + specsById, + tableLocation, + dataFilter, + ignoreResiduals, + caseSensitive, + scanMetrics, + executorService); + } + } +} diff --git a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java index d97c63b61608..6c865c121fda 100644 --- a/core/src/main/java/org/apache/iceberg/SnapshotProducer.java +++ b/core/src/main/java/org/apache/iceberg/SnapshotProducer.java @@ -55,6 +55,9 @@ import org.apache.iceberg.exceptions.CommitStateUnknownException; import org.apache.iceberg.exceptions.RuntimeIOException; import org.apache.iceberg.exceptions.ValidationException; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; import org.apache.iceberg.metrics.CommitMetrics; import org.apache.iceberg.metrics.CommitMetricsResult; @@ -69,6 +72,7 @@ import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.relocated.com.google.common.math.IntMath; +import org.apache.iceberg.types.Types; import org.apache.iceberg.util.Exceptions; import org.apache.iceberg.util.PropertyUtil; import org.apache.iceberg.util.SnapshotUtil; @@ -301,6 +305,22 @@ public Snapshot apply() { List manifests = apply(base, parentSnapshot); + ManifestFile[] manifestFiles = new ManifestFile[manifests.size()]; + Tasks.range(manifestFiles.length) + .stopOnFailure() + .throwFailureWhenFinished() + .executeWith(workerPool()) + .run(index -> manifestFiles[index] = manifestsWithMetadata.get(manifests.get(index))); + + if (base.formatVersion() >= TableMetadata.MIN_FORMAT_VERSION_PARQUET_MANIFESTS) { + return applyV4(manifestFiles, sequenceNumber, parentSnapshotId, parentSnapshot); + } + + return applyV3(manifestFiles, sequenceNumber, parentSnapshotId); + } + + private Snapshot applyV3( + ManifestFile[] manifestFiles, long sequenceNumber, Long parentSnapshotId) { OutputFile manifestList = manifestListPath(); ManifestListWriter writer = @@ -316,15 +336,6 @@ public Snapshot apply() { try (writer) { // keep track of the manifest lists created manifestLists.add(manifestList.location()); - - ManifestFile[] manifestFiles = new ManifestFile[manifests.size()]; - - Tasks.range(manifestFiles.length) - .stopOnFailure() - .throwFailureWhenFinished() - .executeWith(workerPool()) - .run(index -> manifestFiles[index] = manifestsWithMetadata.get(manifests.get(index))); - writer.addAll(Arrays.asList(manifestFiles)); } catch (IOException e) { throw new RuntimeIOException(e, "Failed to write manifest list file"); @@ -337,10 +348,25 @@ public Snapshot apply() { assignedRows = writer.nextRowId() - base.nextRowId(); } - Map summary = summary(); - String operation = operation(); + validateReplace(); - if (summary != null && DataOperations.REPLACE.equals(operation)) { + return new BaseSnapshot( + sequenceNumber, + snapshotId(), + parentSnapshotId, + System.currentTimeMillis(), + operation(), + summary(base), + base.currentSchemaId(), + manifestList.location(), + nextRowId, + assignedRows, + writer.toManifestListFile().encryptionKeyID()); + } + + private void validateReplace() { + Map summary = summary(); + if (summary != null && DataOperations.REPLACE.equals(operation())) { long addedRecords = PropertyUtil.propertyAsLong(summary, SnapshotSummary.ADDED_RECORDS_PROP, 0L); long replacedRecords = @@ -353,6 +379,52 @@ public Snapshot apply() { addedRecords, replacedRecords); } + } + + /** + * Writes a v4 flat-tree root manifest. + * + *

The data files from the freshly written leaf manifests are inlined into the root manifest as + * {@code DATA} entries (the leaf manifests are not referenced), alongside data entries carried + * forward from the parent snapshot's root manifest. + */ + private Snapshot applyV4( + ManifestFile[] manifestFiles, + long sequenceNumber, + Long parentSnapshotId, + Snapshot parentSnapshot) { + List entries = Lists.newArrayList(); + + for (ManifestFile manifest : manifestFiles) { + Preconditions.checkArgument( + manifest.content() == ManifestContent.DATA, + "Cannot write delete manifest to v4 flat-tree root manifest: %s", + manifest.path()); + + try (ManifestReader reader = + ManifestFiles.read(manifest, ops.io(), base.specsById())) { + for (DataFile file : reader) { + entries.add(dataFileToTrackedFile(file, sequenceNumber)); + } + } catch (IOException e) { + throw new RuntimeIOException(e, "Failed to read manifest: %s", manifest.path()); + } + } + + entries.addAll(readParentDataEntries(parentSnapshot)); + + OutputFile rootManifest = rootManifestPath(); + writeRootManifest(rootManifest, entries); + manifestLists.add(rootManifest.location()); + + long addedRows = 0L; + for (ManifestFile manifest : manifestFiles) { + if (manifest.addedRowsCount() != null) { + addedRows += manifest.addedRowsCount(); + } + } + + validateReplace(); return new BaseSnapshot( sequenceNumber, @@ -362,10 +434,92 @@ public Snapshot apply() { operation(), summary(base), base.currentSchemaId(), - manifestList.location(), - nextRowId, - assignedRows, - writer.toManifestListFile().encryptionKeyID()); + rootManifest.location(), + base.nextRowId(), + addedRows, + null); + } + + private TrackedFile dataFileToTrackedFile(DataFile file, long sequenceNumber) { + Tracking tracking = + new TrackingStruct( + EntryStatus.ADDED, + snapshotId(), + sequenceNumber, + sequenceNumber, + null, + null, + null, + null); + return new TrackedFileStruct( + tracking, + FileContent.DATA, + base.formatVersion(), + file.location(), + file.format(), + file.recordCount(), + file.fileSizeInBytes(), + file.specId(), + coercePartition(file.partition()), + null, + file.sortOrderId(), + null, + null, + file.keyMetadata(), + file.splitOffsets(), + file.equalityFieldIds()); + } + + private PartitionData coercePartition(StructLike partition) { + if (partition instanceof PartitionData) { + return ((PartitionData) partition).copy(); + } + + return null; + } + + /** Reads DATA entries carried forward from the parent snapshot's flat-tree root manifest. */ + private List readParentDataEntries(Snapshot parentSnapshot) { + List dataEntries = Lists.newArrayList(); + if (parentSnapshot == null || parentSnapshot.manifestListLocation() == null) { + return dataEntries; + } + + String location = parentSnapshot.manifestListLocation(); + if (FileFormat.fromFileName(location) != FileFormat.PARQUET) { + return dataEntries; + } + + InputFile input = ops.io().newInputFile(location); + try (CloseableIterable parentEntries = + V4ManifestReader.builder(input, base.specsById(), base.location()).build()) { + for (TrackedFile entry : parentEntries) { + if (entry.contentType() == FileContent.DATA && entry.tracking().isLive()) { + dataEntries.add(entry.copy()); + } + } + } catch (IOException e) { + throw new RuntimeIOException(e, "Failed to read parent root manifest: %s", location); + } + + return dataEntries; + } + + private void writeRootManifest(OutputFile output, List entries) { + Schema schema = TrackedFile.schema(base.spec().partitionType(), Types.StructType.of()); + try (FileAppender appender = + InternalData.write(FileFormat.PARQUET, output) + .schema(schema) + .named("tracked_file") + .meta("format-version", "4") + .overwrite() + .build()) { + for (TrackedFile entry : entries) { + appender.add((StructLike) entry); + } + } catch (IOException e) { + throw new RuntimeIOException(e, "Failed to write root manifest: %s", output.location()); + } } private void runValidations(Snapshot parentSnapshot) { @@ -621,6 +775,19 @@ protected OutputFile manifestListPath() { commitUUID)))); } + protected OutputFile rootManifestPath() { + return ops.io() + .newOutputFile( + ops.metadataFileLocation( + FileFormat.PARQUET.addExtension( + String.format( + Locale.ROOT, + "root-%d-%d-%s", + snapshotId(), + attempt.incrementAndGet(), + commitUUID)))); + } + protected EncryptedOutputFile newManifestOutputFile() { String manifestFileLocation = ops.metadataFileLocation( diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java index 154abb9f7478..ac9b8e7e5b9a 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java @@ -18,6 +18,7 @@ */ package org.apache.iceberg; +import java.io.Serializable; import java.nio.ByteBuffer; import java.util.List; import java.util.Map; @@ -55,7 +56,7 @@ static DeleteFile asEqualityDeleteFile(TrackedFile file, Map> - implements ContentFile { + implements ContentFile, Serializable { private final TrackedFile file; private final PartitionSpec spec; diff --git a/core/src/test/java/org/apache/iceberg/TestScanTaskPlanner.java b/core/src/test/java/org/apache/iceberg/TestScanTaskPlanner.java new file mode 100644 index 000000000000..9dabea65f065 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestScanTaskPlanner.java @@ -0,0 +1,847 @@ +/* + * 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.nio.ByteBuffer; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.function.UnaryOperator; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.inmemory.InMemoryFileIO; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileAppender; +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.Types; +import org.apache.iceberg.util.LocationUtil; +import org.apache.iceberg.util.ThreadPools; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.FieldSource; + +class TestScanTaskPlanner { + private static final long SNAPSHOT_ID = 42L; + 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 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 PARTITIONED_SPECS = + ImmutableMap.of(SPEC.specId(), SPEC); + private static final Map UNPARTITIONED_SPECS = + ImmutableMap.of(PartitionSpec.unpartitioned().specId(), PartitionSpec.unpartitioned()); + + private static final List MANIFEST_FORMATS = + ImmutableList.of(FileFormat.AVRO, FileFormat.PARQUET); + + private final InMemoryFileIO fileIO = new InMemoryFileIO(); + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void rootWithDirectDataEntries(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 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 rootWithDataManifestExpandsLeaf(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFile("leaf-a.parquet", EMPTY_PARTITION_DATA), + dataFile("leaf-b.parquet", EMPTY_PARTITION_DATA))); + InputFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(dataManifest(leaf.location()))); + + List 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 mixedRootDataAndLeafManifest(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf-data-file.parquet", EMPTY_PARTITION_DATA))); + InputFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFile("root-data-file.parquet", EMPTY_PARTITION_DATA), + dataManifest(leaf.location()))); + + List tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .hasSize(2) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder( + resolved("root-data-file.parquet"), resolved("leaf-data-file.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void deletionVectorAttachedToTask(FileFormat format) throws IOException { + TrackedFile fileWithDv = + dataFile( + "with-dv.parquet", + EMPTY_PARTITION_DATA, + deletionVector(DV_LOCATION, DV_OFFSET, DV_SIZE_IN_BYTES, DV_CARDINALITY)); + InputFile root = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(fileWithDv)); + + List tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks).hasSize(1); + assertThat(tasks.get(0).deletes()) + .hasSize(1) + .allSatisfy( + delete -> { + assertThat(delete.content()).isEqualTo(FileContent.POSITION_DELETES); + assertThat(delete.referencedDataFile()).isEqualTo(resolved("with-dv.parquet")); + assertThat(delete.recordCount()).isEqualTo(DV_CARDINALITY); + assertThat(delete.contentOffset()).isEqualTo(DV_OFFSET); + assertThat(delete.contentSizeInBytes()).isEqualTo(DV_SIZE_IN_BYTES); + }); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void onlyLiveEntriesArePlanned(FileFormat format) throws IOException { + // EXISTING/ADDED/MODIFIED are live; DELETED/REPLACED are not + InputFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFileWithStatus(EntryStatus.ADDED, "added.parquet"), + dataFileWithStatus(EntryStatus.EXISTING, "existing.parquet"), + dataFileWithStatus(EntryStatus.MODIFIED, "modified.parquet"), + dataFileWithStatus(EntryStatus.DELETED, "deleted.parquet"), + dataFileWithStatus(EntryStatus.REPLACED, "replaced.parquet"))); + + List tasks = plan(root, UNPARTITIONED_SPECS); + + assertThat(tasks) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder( + resolved("added.parquet"), resolved("existing.parquet"), resolved("modified.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void partitionFilterPrunesFiles(FileFormat format) throws IOException { + InputFile root = + writeManifest( + format, + PARTITION_TYPE, + ImmutableList.of( + dataFile("keep.parquet", partition(1)), dataFile("prune.parquet", partition(2)))); + + List tasks = + plan(root, PARTITIONED_SPECS, expander -> expander.filterData(Expressions.equal("id", 1))); + + assertThat(tasks) + .hasSize(1) + .extracting(task -> task.file().location()) + .containsExactly(resolved("keep.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterDataAccumulatesWithAnd(FileFormat format) throws IOException { + InputFile root = + writeManifest( + format, + PARTITION_TYPE, + ImmutableList.of( + dataFile("id1.parquet", partition(1)), + dataFile("id2.parquet", partition(2)), + dataFile("id3.parquet", partition(3)))); + + // two filters AND together: id >= 2 AND id <= 2 keeps only id2. A last-filter-wins bug would + // apply just id <= 2 and also keep id1; dropping the second filter would also keep id3. + List tasks = + plan( + root, + PARTITIONED_SPECS, + expander -> + expander + .filterData(Expressions.greaterThanOrEqual("id", 2)) + .filterData(Expressions.lessThanOrEqual("id", 2))); + + assertThat(tasks) + .extracting(task -> task.file().location()) + .containsExactly(resolved("id2.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void residualAttachedFromFilter(FileFormat format) throws IOException { + InputFile root = + writeManifest( + format, PARTITION_TYPE, ImmutableList.of(dataFile("keep.parquet", partition(1)))); + + List withResidual = + plan( + root, + PARTITIONED_SPECS, + expander -> expander.filterData(Expressions.equal("data", "x"))); + // the identity partition on id leaves the data predicate as a residual + assertThat(withResidual.get(0).residual()) + .hasToString(Expressions.equal("data", "x").toString()); + + List ignored = + plan( + root, + PARTITIONED_SPECS, + expander -> expander.filterData(Expressions.equal("data", "x")).ignoreResiduals()); + assertThat(ignored.get(0).residual()).isEqualTo(Expressions.alwaysTrue()); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void sameSpecFilesShareContext(FileFormat format) throws IOException { + InputFile root = + writeManifest( + format, + PARTITION_TYPE, + ImmutableList.of( + dataFile("a.parquet", partition(1)), dataFile("b.parquet", partition(1)))); + + List tasks = + plan( + root, + PARTITIONED_SPECS, + expander -> expander.filterData(Expressions.equal("data", "x"))); + + // both files share spec 0, so both tasks must carry the same schema, spec, and residual + assertThat(tasks) + .hasSize(2) + .allSatisfy( + task -> { + assertThat(task.schema().asStruct()).isEqualTo(TABLE_SCHEMA.asStruct()); + assertThat(task.spec()).isEqualTo(SPEC); + assertThat(task.residual()).hasToString(Expressions.equal("data", "x").toString()); + }); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void buildingAndClosingWithoutIteratingDoesNotScanLeaves(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf.parquet", EMPTY_PARTITION_DATA))); + InputFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(dataManifest(leaf.location()))); + + ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); + ScanTaskPlanner planner = + ScanTaskPlanner.builder(fileIO, root, UNPARTITIONED_SPECS, TABLE_LOCATION) + .scanMetrics(metrics) + .build(); + // the root is read eagerly to route its entries; leaf readers open lazily, so closing the plan + // without iterating scans only the root, not the leaf + planner.planFiles().close(); + + assertThat(metrics.scannedDataManifests().value()).isEqualTo(1L); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void caseInsensitiveFilterPrunesFiles(FileFormat format) throws IOException { + InputFile root = + writeManifest( + format, + PARTITION_TYPE, + ImmutableList.of( + dataFile("keep.parquet", partition(1)), dataFile("prune.parquet", partition(2)))); + + List tasks = + plan( + root, + PARTITIONED_SPECS, + expander -> expander.caseSensitive(false).filterData(Expressions.equal("ID", 1))); + + assertThat(tasks) + .as("a case-insensitive filter resolves the upper-case column and prunes") + .hasSize(1) + .extracting(task -> task.file().location()) + .containsExactly(resolved("keep.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void parallelPlanningMatchesSequential(FileFormat format) throws IOException { + // a DV and a residual-bearing filter in the tree so parity actually exercises createTask's DV + // branch and per-spec residual keying on worker threads, not just locations + TrackedFile withDv = + dataFile( + "leaf1.parquet", + partition(1), + deletionVector(DV_LOCATION, DV_OFFSET, DV_SIZE_IN_BYTES, DV_CARDINALITY)); + InputFile leaf1 = writeManifest(format, PARTITION_TYPE, ImmutableList.of(withDv)); + InputFile leaf2 = + writeManifest( + format, PARTITION_TYPE, ImmutableList.of(dataFile("leaf2.parquet", partition(1)))); + InputFile root = + writeManifest( + format, + PARTITION_TYPE, + ImmutableList.of(dataManifest(leaf1.location()), dataManifest(leaf2.location()))); + + List sequential = + plan( + root, + PARTITIONED_SPECS, + expander -> expander.filterData(Expressions.equal("data", "x"))); + + ExecutorService pool = ThreadPools.newFixedThreadPool("test-scan-task-planner", 2); + try { + List parallel = + plan( + root, + PARTITIONED_SPECS, + expander -> expander.filterData(Expressions.equal("data", "x")).planWith(pool)); + assertThat(parallel) + .extracting( + task -> task.file().location(), + task -> task.residual().toString(), + task -> task.deletes().size()) + .containsExactlyInAnyOrderElementsOf( + Lists.transform( + sequential, + task -> + tuple( + task.file().location(), + task.residual().toString(), + task.deletes().size()))); + } finally { + pool.shutdownNow(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void deleteContentInRootIsUnsupported(FileFormat format) throws IOException { + InputFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(deleteManifest("deletes.avro"))); + + // delete content is only produced by upgraded trees; that path is not yet implemented + ScanTaskPlanner expander = + ScanTaskPlanner.builder(fileIO, root, UNPARTITIONED_SPECS, TABLE_LOCATION).build(); + assertThatThrownBy(() -> Lists.newArrayList(expander.planFiles())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot plan content type in root manifest: DELETE_MANIFEST"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void deleteContentInLeafIsUnsupported(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of( + dataFile("leaf-data.parquet", EMPTY_PARTITION_DATA), + deleteManifest("leaf-deletes.avro"))); + InputFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(dataManifest(leaf.location()))); + + ScanTaskPlanner expander = + ScanTaskPlanner.builder(fileIO, root, UNPARTITIONED_SPECS, TABLE_LOCATION).build(); + assertThatThrownBy(() -> Lists.newArrayList(expander.planFiles())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot plan content type in leaf manifest: DELETE_MANIFEST"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void nestedDataManifestInLeafIsRejected(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, EMPTY_PARTITION, ImmutableList.of(dataManifest("nested-leaf.parquet"))); + InputFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(dataManifest(leaf.location()))); + + // a nested manifest is structurally impossible in a two-level tree, not merely unsupported + ScanTaskPlanner expander = + ScanTaskPlanner.builder(fileIO, root, UNPARTITIONED_SPECS, TABLE_LOCATION).build(); + assertThatThrownBy(() -> Lists.newArrayList(expander.planFiles())) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage( + "Cannot expand a nested manifest in a leaf manifest: " + + resolved("nested-leaf.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void leafWithManifestDeletionVectorIsUnsupported(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf.parquet", EMPTY_PARTITION_DATA))); + InputFile root = + writeManifest( + format, EMPTY_PARTITION, ImmutableList.of(dataManifestWithDv(leaf.location()))); + + // applying the manifest-level DV is not built yet; expanding the leaf would resurface deleted + // entries, so the planner rejects rather than silently corrupting + ScanTaskPlanner expander = + ScanTaskPlanner.builder(fileIO, root, UNPARTITIONED_SPECS, TABLE_LOCATION).build(); + assertThatThrownBy(() -> Lists.newArrayList(expander.planFiles())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot apply manifest deletion vector for leaf manifest: " + leaf.location()); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void nonV4LeafManifestIsUnsupported(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf.parquet", EMPTY_PARTITION_DATA))); + // an upgraded tree can reference a v3-format leaf; per-leaf reader dispatch is not built yet + InputFile root = + writeManifest(format, EMPTY_PARTITION, ImmutableList.of(dataManifest(leaf.location(), 3))); + + ScanTaskPlanner expander = + ScanTaskPlanner.builder(fileIO, root, UNPARTITIONED_SPECS, TABLE_LOCATION).build(); + assertThatThrownBy(() -> Lists.newArrayList(expander.planFiles())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot expand leaf manifest with format version 3: " + leaf.location()); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void encryptedLeafManifestIsUnsupported(FileFormat format) throws IOException { + InputFile leaf = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf.parquet", EMPTY_PARTITION_DATA))); + TrackedFile encryptedLeaf = + encryptedDataManifest(leaf.location(), ByteBuffer.wrap(new byte[] {1, 2, 3})); + InputFile root = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(encryptedLeaf)); + + // reading the leaf as a plain file would hand ciphertext to the reader; the decrypting path + // is not built yet, so the planner rejects rather than silently corrupting + ScanTaskPlanner expander = + ScanTaskPlanner.builder(fileIO, root, UNPARTITIONED_SPECS, TABLE_LOCATION).build(); + assertThatThrownBy(() -> Lists.newArrayList(expander.planFiles())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Cannot read encrypted leaf manifest: " + leaf.location()); + } + + @Test + void emptyRootYieldsNoTasks() throws IOException { + // an empty root has no plannable entries regardless of manifest format; AVRO is used because + // the Parquet writer does not materialize a file when no records are appended + InputFile root = writeManifest(FileFormat.AVRO, EMPTY_PARTITION, ImmutableList.of()); + + assertThat(plan(root, UNPARTITIONED_SPECS)).isEmpty(); + } + + @Test + void planFilesAcrossMixedManifestFormats() throws IOException { + // each manifest's format is derived from its own location, so a tree can mix formats; a + // Parquet root can point at both an Avro and a Parquet leaf + InputFile avroLeaf = + writeManifest( + FileFormat.AVRO, + EMPTY_PARTITION, + ImmutableList.of(dataFile("data-from-avro-leaf.parquet", EMPTY_PARTITION_DATA))); + InputFile parquetLeaf = + writeManifest( + FileFormat.PARQUET, + EMPTY_PARTITION, + ImmutableList.of(dataFile("data-from-parquet-leaf.parquet", EMPTY_PARTITION_DATA))); + InputFile root = + writeManifest( + FileFormat.PARQUET, + EMPTY_PARTITION, + ImmutableList.of( + dataManifest(avroLeaf.location()), dataManifest(parquetLeaf.location()))); + + assertThat(plan(root, UNPARTITIONED_SPECS)) + .extracting(task -> task.file().location()) + .containsExactlyInAnyOrder( + resolved("data-from-avro-leaf.parquet"), resolved("data-from-parquet-leaf.parquet")); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void planFilesAcrossMultipleSpecs(FileFormat format) throws IOException { + PartitionSpec spec0 = + PartitionSpec.builderFor(TABLE_SCHEMA) + .withSpecId(0) + .add(1, 1000, "id", Transforms.identity()) + .build(); + PartitionSpec spec1 = + PartitionSpec.builderFor(TABLE_SCHEMA) + .withSpecId(1) + .add(2, 1001, "data", Transforms.identity()) + .build(); + Map specsById = + ImmutableMap.of(spec0.specId(), spec0, spec1.specId(), spec1); + Types.StructType unionType = Partitioning.unionPartitionTypes(specsById.values()); + + TrackedFile spec0Keep = + dataFile("spec0-keep.parquet", spec0.specId(), unionPartition(unionType, 1, null), null); + TrackedFile spec0Prune = + dataFile("spec0-prune.parquet", spec0.specId(), unionPartition(unionType, 2, null), null); + TrackedFile spec1File = + dataFile("spec1.parquet", spec1.specId(), unionPartition(unionType, null, "x"), null); + InputFile root = + writeManifest(format, unionType, ImmutableList.of(spec0Keep, spec0Prune, spec1File)); + + // id = 1 prunes the spec0 file partitioned on id=2; spec1 is not partitioned by id, so its + // residual keeps the predicate. A per-spec residual mis-keying would surface here. + List tasks = + plan(root, specsById, expander -> expander.filterData(Expressions.equal("id", 1))); + + assertThat(tasks) + .extracting( + task -> task.file().location(), + task -> task.spec().specId(), + task -> task.residual().toString()) + .containsExactlyInAnyOrder( + tuple(resolved("spec0-keep.parquet"), 0, Expressions.alwaysTrue().toString()), + tuple(resolved("spec1.parquet"), 1, Expressions.equal("id", 1).toString())); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void planFilesReportsScanMetrics(FileFormat format) throws IOException { + TrackedFile fileWithDv = + dataFile( + "with-dv.parquet", + EMPTY_PARTITION_DATA, + deletionVector(DV_LOCATION, DV_OFFSET, DV_SIZE_IN_BYTES, DV_CARDINALITY)); + InputFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("plain.parquet", EMPTY_PARTITION_DATA), fileWithDv)); + + ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); + List tasks = + plan(root, UNPARTITIONED_SPECS, expander -> expander.scanMetrics(metrics)); + + assertThat(tasks).hasSize(2); + assertThat(metrics.scannedDataManifests().value()) + .as("the root is scanned; there are no leaves") + .isEqualTo(1L); + assertThat(metrics.resultDataFiles().value()).isEqualTo(2L); + assertThat(metrics.totalFileSizeInBytes().value()).isEqualTo(2 * FILE_SIZE_IN_BYTES); + assertThat(metrics.resultDeleteFiles().value()) + .as("only the file with a colocated DV contributes a delete file") + .isEqualTo(1L); + assertThat(metrics.totalDeleteFileSizeInBytes().value()) + .as("the DV delete contributes its size") + .isEqualTo(DV_SIZE_IN_BYTES); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void scannedManifestsCountsRootAndLeaves(FileFormat format) throws IOException { + InputFile leaf1 = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf1.parquet", EMPTY_PARTITION_DATA))); + InputFile leaf2 = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("leaf2.parquet", EMPTY_PARTITION_DATA))); + InputFile root = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataManifest(leaf1.location()), dataManifest(leaf2.location()))); + + ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); + plan(root, UNPARTITIONED_SPECS, expander -> expander.scanMetrics(metrics)); + + assertThat(metrics.scannedDataManifests().value()) + .as("the root and both leaves are each counted as a scanned manifest") + .isEqualTo(3L); + } + + private List plan(InputFile root, Map specsById) + throws IOException { + return plan(root, specsById, UnaryOperator.identity()); + } + + private List plan( + InputFile root, + Map specsById, + UnaryOperator configure) + throws IOException { + ScanTaskPlanner expander = + configure.apply(ScanTaskPlanner.builder(fileIO, root, specsById, TABLE_LOCATION)).build(); + try (CloseableIterable tasks = expander.planFiles()) { + return Lists.newArrayList(tasks); + } + } + + private static TrackedFile dataFile(String location, PartitionData partition) { + return dataFile(location, specId(partition), partition, null); + } + + private static TrackedFile dataFile(String location, PartitionData partition, DeletionVector dv) { + return dataFile(location, specId(partition), partition, dv); + } + + private static TrackedFile dataFile( + String location, Integer specId, PartitionData partition, DeletionVector dv) { + return trackedFile(addedTracking(), FileContent.DATA, location, specId, partition, dv, null); + } + + private static TrackedFile dataFileWithStatus(EntryStatus status, String location) { + return trackedFile( + tracking(status), + FileContent.DATA, + location, + specId(EMPTY_PARTITION_DATA), + EMPTY_PARTITION_DATA, + null, + null); + } + + private static TrackedFile dataManifest(String location) { + return trackedFile( + addedTracking(), FileContent.DATA_MANIFEST, location, null, null, null, null); + } + + private static TrackedFile dataManifest(String location, int formatVersion) { + return new TrackedFileStruct( + /* tracking= */ addedTracking(), + /* contentType= */ FileContent.DATA_MANIFEST, + /* formatVersion= */ formatVersion, + /* location= */ location, + /* fileFormat= */ FileFormat.PARQUET, + /* recordCount= */ RECORD_COUNT, + /* fileSizeInBytes= */ FILE_SIZE_IN_BYTES, + /* specId= */ null, + /* partition= */ null, + /* contentStats= */ null, + /* sortOrderId= */ null, + /* deletionVector= */ null, + /* manifestInfo= */ null, + /* keyMetadata= */ null, + /* splitOffsets= */ null, + /* equalityIds= */ null); + } + + private static TrackedFile dataManifestWithDv(String location) { + ManifestInfo manifestInfo = + ManifestInfoStruct.builder() + .addedFilesCount(1) + .existingFilesCount(0) + .deletedFilesCount(0) + .replacedFilesCount(0) + .addedRowsCount(RECORD_COUNT) + .existingRowsCount(0) + .deletedRowsCount(0) + .replacedRowsCount(0) + .minSequenceNumber(0L) + .dv(ByteBuffer.wrap(new byte[] {1, 2, 3})) + .dvCardinality(1L) + .build(); + return trackedFile( + addedTracking(), FileContent.DATA_MANIFEST, location, null, null, null, manifestInfo); + } + + private static TrackedFile encryptedDataManifest(String location, ByteBuffer keyMetadata) { + return new TrackedFileStruct( + /* tracking= */ addedTracking(), + /* contentType= */ FileContent.DATA_MANIFEST, + /* formatVersion= */ WRITER_FORMAT_VERSION, + /* location= */ location, + /* fileFormat= */ FileFormat.PARQUET, + /* recordCount= */ RECORD_COUNT, + /* fileSizeInBytes= */ FILE_SIZE_IN_BYTES, + /* specId= */ null, + /* partition= */ null, + /* contentStats= */ null, + /* sortOrderId= */ null, + /* deletionVector= */ null, + /* manifestInfo= */ null, + /* keyMetadata= */ keyMetadata, + /* splitOffsets= */ null, + /* equalityIds= */ null); + } + + private static TrackedFile deleteManifest(String location) { + return trackedFile( + addedTracking(), FileContent.DELETE_MANIFEST, location, null, null, null, null); + } + + private static TrackedFile trackedFile( + TrackingStruct tracking, + FileContent contentType, + String location, + Integer specId, + PartitionData partition, + DeletionVector dv, + ManifestInfo manifestInfo) { + return new TrackedFileStruct( + /* tracking= */ tracking, + /* contentType= */ contentType, + /* formatVersion= */ WRITER_FORMAT_VERSION, + /* location= */ location, + /* fileFormat= */ FileFormat.PARQUET, + /* recordCount= */ RECORD_COUNT, + /* fileSizeInBytes= */ FILE_SIZE_IN_BYTES, + /* specId= */ specId, + /* partition= */ partition, + /* contentStats= */ null, + /* sortOrderId= */ null, + /* deletionVector= */ dv, + /* manifestInfo= */ manifestInfo, + /* keyMetadata= */ null, + /* splitOffsets= */ null, + /* equalityIds= */ null); + } + + private static TrackingStruct addedTracking() { + return tracking(EntryStatus.ADDED); + } + + private static TrackingStruct tracking(EntryStatus status) { + return new TrackingStruct( + /* status= */ status, + /* snapshotId= */ SNAPSHOT_ID, + /* dataSequenceNumber= */ null, + /* fileSequenceNumber= */ null, + /* dvSnapshotId= */ null, + /* firstRowId= */ null, + /* deletedPositions= */ null, + /* replacedPositions= */ null); + } + + private static Integer specId(PartitionData partition) { + boolean unpartitioned = partition.size() == 0; + return unpartitioned ? PartitionSpec.unpartitioned().specId() : SPEC.specId(); + } + + private static PartitionData partition(int id) { + PartitionData partition = new PartitionData(PARTITION_TYPE); + partition.set(0, id); + return partition; + } + + // the location a relative fixture resolves to once read against TABLE_LOCATION + private static String resolved(String location) { + return LocationUtil.resolveLocation(TABLE_LOCATION, location); + } + + private static PartitionData unionPartition(Types.StructType unionType, Integer id, String data) { + PartitionData partition = new PartitionData(unionType); + partition.set(0, id); + partition.set(1, data); + return partition; + } + + private static DeletionVector deletionVector( + String location, long offset, long sizeInBytes, long cardinality) { + return DeletionVectorStruct.builder() + .location(location) + .offset(offset) + .sizeInBytes(sizeInBytes) + .cardinality(cardinality) + .build(); + } + + private InputFile writeManifest( + FileFormat format, Types.StructType partitionType, Iterable files) + throws IOException { + Schema writeSchema = TrackedFile.schema(partitionType, Types.StructType.of()); + // write under the table location so a leaf's resolved reference round-trips to the file on disk + OutputFile out = + fileIO.newOutputFile( + TABLE_LOCATION + + "/metadata/manifest-" + + System.nanoTime() + + "." + + format.name().toLowerCase(Locale.ROOT)); + try (FileAppender appender = + InternalData.write(format, out).schema(writeSchema).named("tracked_file").build()) { + for (TrackedFile file : files) { + appender.add((StructLike) file); + } + } + + return out.toInputFile(); + } +} diff --git a/core/src/test/java/org/apache/iceberg/TestV4FlatTreeReadWrite.java b/core/src/test/java/org/apache/iceberg/TestV4FlatTreeReadWrite.java new file mode 100644 index 000000000000..52585172c6be --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestV4FlatTreeReadWrite.java @@ -0,0 +1,106 @@ +/* + * 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.required; +import static org.assertj.core.api.Assertions.assertThat; + +import java.io.File; +import java.io.IOException; +import java.util.List; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +/** + * End-to-end round trip for the v4 flat-tree prototype: a fast append writes a Parquet root + * manifest with inlined DATA entries, and a scan plans directly from it via {@link + * ScanTaskPlanner}. + */ +public class TestV4FlatTreeReadWrite { + private static final Schema SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), required(2, "data", Types.StringType.get())); + + private static final DataFile FILE_A = + DataFiles.builder(PartitionSpec.unpartitioned()) + .withPath("file:/tmp/v4/data-a.parquet") + .withFileSizeInBytes(100) + .withRecordCount(3) + .withFormat(FileFormat.PARQUET) + .build(); + + private static final DataFile FILE_B = + DataFiles.builder(PartitionSpec.unpartitioned()) + .withPath("file:/tmp/v4/data-b.parquet") + .withFileSizeInBytes(120) + .withRecordCount(1) + .withFormat(FileFormat.PARQUET) + .build(); + + @TempDir private File tableDir; + + @Test + public void testFastAppendWritesRootManifest() throws IOException { + Table table = TestTables.create(tableDir, "v4_root", SCHEMA, PartitionSpec.unpartitioned(), 4); + + table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit(); + + Snapshot snapshot = table.currentSnapshot(); + assertThat(snapshot.manifestListLocation()) + .as("v4 snapshot points at a Parquet root manifest") + .endsWith(".parquet") + .doesNotContain("snap-"); + } + + @Test + public void testScanPlansInlinedDataEntries() throws IOException { + Table table = TestTables.create(tableDir, "v4_scan", SCHEMA, PartitionSpec.unpartitioned(), 4); + + table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit(); + + assertThat(scanLocations(table)) + .containsExactlyInAnyOrder(FILE_A.location(), FILE_B.location()); + } + + @Test + public void testSecondAppendCarriesForwardParentEntries() throws IOException { + Table table = TestTables.create(tableDir, "v4_carry", SCHEMA, PartitionSpec.unpartitioned(), 4); + + table.newFastAppend().appendFile(FILE_A).commit(); + table.newFastAppend().appendFile(FILE_B).commit(); + + assertThat(scanLocations(table)) + .as("second append carries forward the first append's data entries") + .containsExactlyInAnyOrder(FILE_A.location(), FILE_B.location()); + } + + private static List scanLocations(Table table) throws IOException { + List locations = Lists.newArrayList(); + try (CloseableIterable tasks = table.newScan().planFiles()) { + for (FileScanTask task : tasks) { + locations.add(task.file().location()); + } + } + + return locations; + } +} diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestV4ReadEndToEnd.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestV4ReadEndToEnd.java new file mode 100644 index 000000000000..0cf0905ee740 --- /dev/null +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestV4ReadEndToEnd.java @@ -0,0 +1,105 @@ +/* + * 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.spark.source; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.util.List; +import org.apache.iceberg.ParameterizedTestExtension; +import org.apache.iceberg.Parameters; +import org.apache.iceberg.Snapshot; +import org.apache.iceberg.Table; +import org.apache.iceberg.spark.SparkCatalogConfig; +import org.apache.iceberg.spark.TestBaseWithCatalog; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; + +/** + * End-to-end tests for v4 table reads and writes using the Adaptive Metadata Tree format. + * + *

A v4 commit writes a Parquet root manifest of {@code TrackedFile} entries; scans plan directly + * from it via the native planner. These tests verify the full pipeline: Spark INSERT -> v4 root + * manifest write -> native scan planning -> Spark SELECT. + */ +@ExtendWith(ParameterizedTestExtension.class) +public class TestV4ReadEndToEnd extends TestBaseWithCatalog { + + @Parameters(name = "catalogName = {0}, implementation = {1}, config = {2}") + protected static Object[][] parameters() { + return new Object[][] { + { + SparkCatalogConfig.HADOOP.catalogName(), + SparkCatalogConfig.HADOOP.implementation(), + SparkCatalogConfig.HADOOP.properties() + } + }; + } + + @AfterEach + public void dropTable() { + sql("DROP TABLE IF EXISTS %s", tableName); + } + + @TestTemplate + public void testV4DataQuery() { + sql( + "CREATE TABLE %s (id bigint, data string) USING iceberg " + + "TBLPROPERTIES ('format-version' = '4')", + tableName); + + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c')", tableName); + + List rows = sql("SELECT * FROM %s ORDER BY id", tableName); + assertThat(rows).hasSize(3); + assertThat(rows.get(0)).isEqualTo(row(1L, "a")); + assertThat(rows.get(1)).isEqualTo(row(2L, "b")); + assertThat(rows.get(2)).isEqualTo(row(3L, "c")); + } + + @TestTemplate + public void testV4RootManifestFormat() { + sql( + "CREATE TABLE %s (id bigint, data string) USING iceberg " + + "TBLPROPERTIES ('format-version' = '4')", + tableName); + + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c')", tableName); + + Table table = validationCatalog.loadTable(tableIdent); + Snapshot snapshot = table.currentSnapshot(); + assertThat(snapshot.manifestListLocation()).endsWith(".parquet").doesNotContain("snap-"); + } + + @TestTemplate + public void testV4MultiSnapshot() { + sql( + "CREATE TABLE %s (id bigint, data string) USING iceberg " + + "TBLPROPERTIES ('format-version' = '4')", + tableName); + + sql("INSERT INTO %s VALUES (1, 'a'), (2, 'b'), (3, 'c')", tableName); + sql("INSERT INTO %s VALUES (4, 'd')", tableName); + + List rows = sql("SELECT * FROM %s ORDER BY id", tableName); + assertThat(rows).hasSize(4); + assertThat(rows.get(0)).isEqualTo(row(1L, "a")); + assertThat(rows.get(3)).isEqualTo(row(4L, "d")); + } +}