This is an automated email from the ASF dual-hosted git repository.
rdblue pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 7461b50476 Core: Add V4 file planner (#17541)
7461b50476 is described below
commit 7461b5047638fe07285a175a3e113e73fc5e19de
Author: Anoop Johnson <[email protected]>
AuthorDate: Thu Oct 22 08:36:03 2026 -0700
Core: Add V4 file planner (#17541)
Plan FileScanTasks from a V4 root manifest, routing DATA entries to tasks
and
expanding DATA_MANIFEST entries into their leaf manifests through
V4ManifestReader. Attach each data file's colocated deletion vector as a
DeleteFile. Add RootManifestFile so the root reads through the same reader
as
the leaves, and TrackedFileAdapters.asDVDeleteFile(DataFile).
---
.../main/java/org/apache/iceberg/FilePlanner.java | 207 +++++++
.../java/org/apache/iceberg/RootManifestFile.java | 148 +++++
.../org/apache/iceberg/TrackedFileAdapters.java | 4 +
.../java/org/apache/iceberg/TestFilePlanner.java | 660 +++++++++++++++++++++
.../org/apache/iceberg/TestRootManifestFile.java | 110 ++++
.../java/org/apache/iceberg/V4TestHelpers.java | 21 +-
6 files changed, 1149 insertions(+), 1 deletion(-)
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 0000000000..78c2206b47
--- /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.
+ *
+ * <p>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<Integer, PartitionSpec> specsById;
+ private final Map<Integer, TaskContext> 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<Integer,
PartitionSpec> 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<FileScanTask> planFiles() {
+ List<DataFile> rootDataFiles = Lists.newArrayList();
+ List<ManifestFile> leafManifests = Lists.newArrayList();
+
+ try (CloseableIterable<TrackedFile> 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<CloseableIterable<TrackedFile>> leafPlanTasks = Lists.newArrayList();
+ for (ManifestFile leaf : leafManifests) {
+ leafPlanTasks.add(reader(leaf));
+ }
+
+ CloseableIterable<TrackedFile> leafFiles =
+ executorService != null && leafManifests.size() > 1
+ ? new ParallelIterable<>(leafPlanTasks, executorService)
+ : CloseableIterable.concat(leafPlanTasks);
+
+ CloseableIterable<DataFile> leafDataFiles =
+ CloseableIterable.transform(
+ leafFiles, file -> TrackedFileAdapters.asDataFile(file,
specsById));
+
+ CloseableIterable<DataFile> dataFiles =
+ CloseableIterable.concat(
+ ImmutableList.of(CloseableIterable.of(rootDataFiles),
leafDataFiles));
+
+ return CloseableIterable.transform(dataFiles, this::createTask);
+ }
+
+ private CloseableIterable<TrackedFile> 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 0000000000..5c7bf8bbc9
--- /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<PartitionFieldSummary> 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 cd3c573968..d1d074c116 100644
--- a/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java
+++ b/core/src/main/java/org/apache/iceberg/TrackedFileAdapters.java
@@ -42,6 +42,10 @@ class TrackedFileAdapters {
return new TrackedDataFile(file, resolveSpecId(file, specsById));
}
+ static TrackedFile asTrackedFile(DataFile file) {
+ return ((TrackedDataFile) file).file();
+ }
+
static DeleteFile asDVDeleteFile(TrackedFile file, Map<Integer,
PartitionSpec> 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 0000000000..dd2e7d3ed1
--- /dev/null
+++ b/core/src/test/java/org/apache/iceberg/TestFilePlanner.java
@@ -0,0 +1,660 @@
+/*
+ * 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.FILE_SIZE_IN_BYTES;
+import static org.apache.iceberg.V4TestHelpers.FORMAT_VERSION_V4;
+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.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.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.types.Types;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.FieldSource;
+import org.mockito.Mockito;
+
+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<Integer, PartitionSpec> 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<Integer, PartitionSpec> MIXED_SPECS =
+ ImmutableMap.of(
+ UNPARTITIONED_WITH_SCHEMA.specId(), UNPARTITIONED_WITH_SCHEMA,
+ BY_ID_SPEC.specId(), BY_ID_SPEC);
+
+ private static final List<FileFormat> MANIFEST_FORMATS =
+ ImmutableList.of(FileFormat.AVRO, FileFormat.PARQUET);
+
+ private static final MetricsConfig METRICS_CONFIG =
+ MetricsConfig.from(ImmutableMap.of(), TABLE_SCHEMA, null);
+ private static final Types.StructType STATS_TYPE =
+ StatsUtil.statsWriteSchema(TABLE_SCHEMA, METRICS_CONFIG);
+
+ private final InMemoryFileIO fileIO = new InMemoryFileIO();
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ void rootWithOnlyDataFiles(FileFormat format) throws IOException {
+ ManifestFile root =
+ writeManifest(
+ format,
+ EMPTY_PARTITION,
+ ImmutableList.of(unpartitionedDataFile(FILE_A),
unpartitionedDataFile(FILE_B)));
+
+ List<FileScanTask> 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<FileScanTask> 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<FileScanTask> 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<FileScanTask> 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<FileScanTask> 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<FileScanTask> tasks =
+ plan(root, UNPARTITIONED_SPECS, planner ->
planner.filterData(Expressions.equal("id", 50)));
+
+ assertThat(Iterables.getOnlyElement(tasks).file().location())
+ .as("only the leaf file whose stats match the filter survives")
+ .isEqualTo(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<FileScanTask> tasks =
+ plan(
+ root,
+ UNPARTITIONED_SPECS,
+ planner -> planner.filterData(Expressions.equal("ID",
50)).caseSensitive(false));
+ assertThat(tasks)
+ .extracting(task -> task.file().location())
+ .as("a case-insensitive filter resolves \"ID\" to \"id\" 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<FileScanTask> 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)),
+ dataFile(
+ UNPARTITIONED_FILE,
+ UNPARTITIONED_WITH_SCHEMA.specId(),
+ idPartition(unionType, null))));
+
+ List<FileScanTask> 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)),
+ dataFile(
+ UNPARTITIONED_FILE,
+ UNPARTITIONED_WITH_SCHEMA.specId(),
+ idPartition(unionType, null))));
+
+ Expression filter = Expressions.and(Expressions.equal("id", 1),
Expressions.equal("data", "x"));
+ List<FileScanTask> 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)),
+ dataFile(
+ UNPARTITIONED_FILE,
+ UNPARTITIONED_WITH_SCHEMA.specId(),
+ idPartition(unionType, null))));
+
+ Expression filter = Expressions.and(Expressions.equal("id", 1),
Expressions.equal("data", "x"));
+ List<FileScanTask> 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<FileScanTask> tasks =
+ plan(root, UNPARTITIONED_SPECS, planner ->
planner.tableLocation(TABLE_LOCATION));
+
+ assertThat(tasks)
+ .extracting(task -> task.file().location())
+ .as("relative paths resolve against the table location in both the
root and leaf readers")
+ .containsExactlyInAnyOrder(
+ TABLE_LOCATION + "/root-data.parquet", TABLE_LOCATION +
"/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())));
+
+ FileIO io = Mockito.spy(fileIO);
+ new FilePlanner(io, root, TABLE_SCHEMA,
UNPARTITIONED_SPECS).planFiles().close();
+
+ Mockito.verify(io).newInputFile(root.path());
+ Mockito.verify(io, Mockito.never()).newInputFile(leaf.path());
+ }
+
+ @ParameterizedTest
+ @FieldSource("MANIFEST_FORMATS")
+ void leafFilesOpenedLazily(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())));
+
+ FileIO io = Mockito.spy(fileIO);
+ FilePlanner planner =
+ new FilePlanner(io, root, TABLE_SCHEMA,
UNPARTITIONED_SPECS).planWith(null);
+
+ try (CloseableIterable<FileScanTask> plan = planner.planFiles()) {
+ CloseableIterator<FileScanTask> tasks = plan.iterator();
+
+ assertThat(tasks.next().file().location()).isEqualTo(FILE_A);
+ Mockito.verify(io).newInputFile(leaf1.path());
+ Mockito.verify(io, Mockito.never()).newInputFile(leaf2.path());
+
+ assertThat(tasks.next().file().location()).isEqualTo(FILE_B);
+ Mockito.verify(io, Mockito.never()).newInputFile(leaf2.path());
+
+ assertThat(tasks.next().file().location()).isEqualTo(FILE_C);
+ Mockito.verify(io).newInputFile(leaf2.path());
+ 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);
+ 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 =
+ manifestRef(FileContent.DATA_MANIFEST, leafLocation,
dataManifestInfo(1, 0));
+ ManifestFile root = writeManifest(format, EMPTY_PARTITION,
ImmutableList.of(v3Leaf));
+
+ FilePlanner planner = new FilePlanner(fileIO, root, TABLE_SCHEMA,
UNPARTITIONED_SPECS);
+ assertThatThrownBy(() -> Lists.newArrayList(planner.planFiles()))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Cannot read manifest with format version 0: 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<FileScanTask> 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<FileScanTask> plan(ManifestFile root, Map<Integer,
PartitionSpec> specsById)
+ throws IOException {
+ return plan(root, specsById, UnaryOperator.identity());
+ }
+
+ private List<FileScanTask> plan(
+ ManifestFile root,
+ Map<Integer, PartitionSpec> specsById,
+ UnaryOperator<FilePlanner> configure)
+ throws IOException {
+ FilePlanner planner = configure.apply(new FilePlanner(fileIO, root,
TABLE_SCHEMA, specsById));
+ try (CloseableIterable<FileScanTask> 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
+ 0L)); // totalBytes
+ return stats;
+ }
+
+ private static ManifestInfo dataManifestInfo(int addedFiles) {
+ return dataManifestInfo(addedFiles, FORMAT_VERSION_V4);
+ }
+
+ private static ManifestInfo dataManifestInfo(int addedFiles, int
formatVersion) {
+ 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)
+ .formatVersion(formatVersion)
+ .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<TrackedFile>
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);
+ }
+}
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 0000000000..b69e6a21db
--- /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 2e803be312..b8cf5164d4 100644
--- a/core/src/test/java/org/apache/iceberg/V4TestHelpers.java
+++ b/core/src/test/java/org/apache/iceberg/V4TestHelpers.java
@@ -42,6 +42,15 @@ class V4TestHelpers {
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);
}
@@ -141,8 +150,18 @@ class V4TestHelpers {
Types.StructType statsType,
Iterable<TrackedFile> 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<TrackedFile> files)
+ throws IOException {
+ Schema writeSchema = TrackedFile.schema(partitionType, statsType);
try (FileAppender<StructLike> appender =
InternalData.write(format,
out).schema(writeSchema).named("tracked_file").build()) {
for (TrackedFile file : files) {
