diff --git a/core/src/main/java/org/apache/iceberg/AllPuffinFilesTable.java b/core/src/main/java/org/apache/iceberg/AllPuffinFilesTable.java new file mode 100644 index 000000000000..87e8779e66a2 --- /dev/null +++ b/core/src/main/java/org/apache/iceberg/AllPuffinFilesTable.java @@ -0,0 +1,906 @@ +/* + * 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.util.List; +import java.util.Map; +import java.util.Set; +import java.util.function.IntPredicate; +import org.apache.iceberg.exceptions.RuntimeIOException; +import org.apache.iceberg.expressions.Binder; +import org.apache.iceberg.expressions.Bound; +import org.apache.iceberg.expressions.BoundReference; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.ExpressionVisitors; +import org.apache.iceberg.expressions.ExpressionVisitors.BoundExpressionVisitor; +import org.apache.iceberg.expressions.Literal; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileIO; +import org.apache.iceberg.puffin.StandardBlobTypes; +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.relocated.com.google.common.collect.Sets; +import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ContentFileUtil; +import org.apache.iceberg.util.StructProjection; + +/** + * A {@link Table} implementation that exposes Puffin files associated with all snapshots currently + * tracked by a table. + * + *

Each row represents a Puffin file referenced by one snapshot through one metadata source. The + * same physical Puffin file may appear in multiple rows when it is referenced by multiple snapshots + * or through multiple sources. + * + *

Deletion vector files are reconstructed from live delete-manifest entries for each reference + * snapshot. Statistics files come from the statistics files registered in the current table + * metadata. Statistics registration can change independently of snapshots, so this table does not + * reconstruct historical statistics registration state. + */ +public class AllPuffinFilesTable extends BaseMetadataTable { + + private static final String SOURCE_STATISTICS = "statistics"; + private static final String SOURCE_DELETION_VECTOR = "deletion_vector"; + + private static final int MIN_FORMAT_VERSION_DELETION_VECTORS = 3; + + private static final List DV_REFERENCED_BLOB_TYPES = + ImmutableList.of(StandardBlobTypes.DV_V1); + + private static final List DV_REFERENCED_FIELD_IDS = + ImmutableList.of(MetadataColumns.ROW_POSITION.fieldId()); + + private static final List DV_SCAN_COLUMNS = + ImmutableList.of( + DataFile.FILE_PATH.name(), + DataFile.FILE_FORMAT.name(), + DataFile.FILE_SIZE.name(), + DataFile.CONTENT_OFFSET.name(), + DataFile.CONTENT_SIZE.name()); + + private static final Types.NestedField REFERENCE_SNAPSHOT_ID = + Types.NestedField.required( + 1, + "reference_snapshot_id", + Types.LongType.get(), + "ID of the snapshot that references the Puffin file"); + + private static final Types.NestedField SOURCE = + Types.NestedField.required( + 3, + "source", + Types.StringType.get(), + "Metadata source that associates the Puffin file with the reference snapshot"); + + private static final Schema ALL_PUFFIN_FILES_SCHEMA = + new Schema( + REFERENCE_SNAPSHOT_ID, + Types.NestedField.required( + 2, + "file_path", + Types.StringType.get(), + "Fully qualified location of the Puffin file"), + SOURCE, + Types.NestedField.required( + 4, + "file_size_in_bytes", + Types.LongType.get(), + "Total size of the Puffin file in bytes"), + Types.NestedField.required( + 5, + "referenced_blob_count", + Types.IntegerType.get(), + "Number of referenced Puffin blobs represented by this row"), + Types.NestedField.required( + 6, + "referenced_blob_types", + Types.ListType.ofRequired(7, Types.StringType.get()), + "Distinct blob types across the referenced blobs"), + Types.NestedField.required( + 8, + "referenced_fields", + Types.ListType.ofRequired( + 9, + Types.StructType.of( + Types.NestedField.required( + 10, + "field_id", + Types.IntegerType.get(), + "Field ID referenced by at least one blob"), + Types.NestedField.optional( + 11, + "current_field_name", + Types.StringType.get(), + "Name currently assigned to the field ID; null if no longer present"))), + "Distinct fields referenced across the blobs")); + + AllPuffinFilesTable(Table table) { + this(table, table.name() + ".all_puffin_files"); + } + + AllPuffinFilesTable(Table table, String name) { + super(table, name); + } + + @Override + public TableScan newScan() { + return new AllPuffinFilesTableScan(table(), schema()); + } + + @Override + public Schema schema() { + return ALL_PUFFIN_FILES_SCHEMA; + } + + @Override + MetadataTableType metadataTableType() { + return MetadataTableType.ALL_PUFFIN_FILES; + } + + private static Map> statisticsRowsBySnapshotId( + List statisticsFiles, Schema tableSchema) { + Map> rowsBySnapshotId = Maps.newHashMap(); + + for (StatisticsFile statisticsFile : statisticsFiles) { + long referenceSnapshotId = statisticsFile.snapshotId(); + StaticDataTask.Row row = + puffinFileToRow(tableSchema, PuffinFileRow.fromStatisticsFile(statisticsFile)); + + rowsBySnapshotId + .computeIfAbsent(referenceSnapshotId, ignored -> Lists.newArrayList()) + .add(row); + } + + return rowsBySnapshotId; + } + + private static boolean mayHaveDeleteFiles(Snapshot snapshot) { + Map summary = snapshot.summary(); + if (summary == null) { + return true; + } + + String totalDeleteFiles = summary.get(SnapshotSummary.TOTAL_DELETE_FILES_PROP); + try { + return totalDeleteFiles == null || Long.parseLong(totalDeleteFiles) > 0L; + } catch (NumberFormatException ignored) { + return true; + } + } + + private static List scanDeletionVectorPuffinFiles( + FileIO io, + Map specsById, + long referenceSnapshotId, + ManifestListFile manifestList) { + Map dvFiles = Maps.newLinkedHashMap(); + + try (CloseableIterable manifests = readManifestList(io, manifestList)) { + for (ManifestFile manifest : manifests) { + if (manifest.content() == ManifestContent.DELETES) { + accumulateDeletionVectorsFromManifest( + io, specsById, referenceSnapshotId, manifest, dvFiles); + } + } + + } catch (IOException e) { + throw new RuntimeIOException( + e, + "Failed to read manifest list for snapshot %s: %s", + referenceSnapshotId, + manifestList.location()); + } + + return dvFiles.values().stream() + .map(DeletionVectorFileAccumulator::toPuffinFileRow) + .collect(ImmutableList.toImmutableList()); + } + + private static CloseableIterable readManifestList( + FileIO io, ManifestListFile manifestList) { + return InternalData.read(FileFormat.AVRO, io.newInputFile(manifestList)) + .setRootType(GenericManifestFile.class) + .setCustomType( + ManifestFile.PARTITION_SUMMARIES_ELEMENT_ID, GenericPartitionFieldSummary.class) + .project(ManifestFile.schema()) + .build(); + } + + private static void accumulateDeletionVectorsFromManifest( + FileIO io, + Map specsById, + long referenceSnapshotId, + ManifestFile deleteManifest, + Map dvFiles) { + try (ManifestReader reader = + ManifestFiles.readDeleteManifest(deleteManifest, io, specsById) + .select(DV_SCAN_COLUMNS); + CloseableIterable> entries = reader.liveEntries()) { + + for (ManifestEntry entry : entries) { + DeleteFile deleteFile = entry.file(); + if (!ContentFileUtil.isDV(deleteFile)) { + continue; + } + + String path = deleteFile.location(); + Long contentOffset = deleteFile.contentOffset(); + Long contentSizeInBytes = deleteFile.contentSizeInBytes(); + + Preconditions.checkState( + contentOffset != null, + "Missing content offset for deletion vector: snapshot=%s, manifest=%s, path=%s", + referenceSnapshotId, + deleteManifest.path(), + path); + Preconditions.checkState( + contentSizeInBytes != null, + "Missing content size for deletion vector: snapshot=%s, manifest=%s, path=%s", + referenceSnapshotId, + deleteManifest.path(), + path); + + long fileSizeInBytes = deleteFile.fileSizeInBytes(); + + dvFiles + .computeIfAbsent( + path, + ignored -> + new DeletionVectorFileAccumulator(referenceSnapshotId, path, fileSizeInBytes)) + .add(fileSizeInBytes, contentOffset, contentSizeInBytes); + } + + } catch (IOException e) { + throw new RuntimeIOException( + e, + "Failed to read delete manifest for snapshot %s: %s", + referenceSnapshotId, + deleteManifest.path()); + } + } + + private static StaticDataTask.Row puffinFileToRow(Schema tableSchema, PuffinFileRow puffinFile) { + ImmutableList.Builder referencedFields = + ImmutableList.builderWithExpectedSize(puffinFile.referencedFieldIds().size()); + + for (Integer fieldId : puffinFile.referencedFieldIds()) { + referencedFields.add(StaticDataTask.Row.of(fieldId, currentFieldName(tableSchema, fieldId))); + } + + return StaticDataTask.Row.of( + puffinFile.referenceSnapshotId(), + puffinFile.path(), + puffinFile.source(), + puffinFile.fileSizeInBytes(), + puffinFile.referencedBlobCount(), + puffinFile.referencedBlobTypes(), + referencedFields.build()); + } + + private static String currentFieldName(Schema tableSchema, int fieldId) { + if (fieldId == MetadataColumns.ROW_POSITION.fieldId()) { + return MetadataColumns.ROW_POSITION.name(); + } + + return tableSchema.findColumnName(fieldId); + } + + private static long estimatedDvTaskSize(Snapshot snapshot) { + long deleteFileCount = totalDeleteFileCount(snapshot); + + if (deleteFileCount <= 1_000L) { + return 4L * 1024 * 1024; + } else if (deleteFileCount <= 10_000L) { + return 8L * 1024 * 1024; + } else if (deleteFileCount <= 50_000L) { + return 16L * 1024 * 1024; + } else { + return 32L * 1024 * 1024; + } + } + + private static long totalDeleteFileCount(Snapshot snapshot) { + Map summary = snapshot.summary(); + if (summary == null) { + return 1L; + } + + String value = summary.get(SnapshotSummary.TOTAL_DELETE_FILES_PROP); + if (value == null) { + return 1L; + } + + try { + return Math.max(1L, Long.parseLong(value)); + } catch (NumberFormatException ignored) { + return 1L; + } + } + + private static class AllPuffinFilesTableScan extends BaseAllMetadataTableScan { + + AllPuffinFilesTableScan(Table table, Schema schema) { + super(table, schema, MetadataTableType.ALL_PUFFIN_FILES); + } + + private AllPuffinFilesTableScan(Table table, Schema schema, TableScanContext context) { + super(table, schema, MetadataTableType.ALL_PUFFIN_FILES, context); + } + + @Override + protected TableScan newRefinedScan(Table table, Schema schema, TableScanContext context) { + return new AllPuffinFilesTableScan(table, schema, context); + } + + @Override + protected CloseableIterable doPlanFiles() { + BaseTable baseTable = (BaseTable) table(); + TableMetadata metadata = baseTable.operations().current(); + Schema tableSchema = metadata.schema(); + Schema projectedSchema = schema(); + + Expression planningFilter = filter(); + Expression residual = residualFilter(); + SnapshotReferenceEvaluator taskEvaluator = + new SnapshotReferenceEvaluator( + planningFilter, + ALL_PUFFIN_FILES_SCHEMA.asStruct(), + REFERENCE_SNAPSHOT_ID.fieldId(), + SOURCE.fieldId(), + isCaseSensitive()); + + boolean mayScanStatistics = taskEvaluator.mayMatchSource(SOURCE_STATISTICS); + boolean mayScanDeletionVectors = + metadata.formatVersion() >= MIN_FORMAT_VERSION_DELETION_VECTORS + && taskEvaluator.mayMatchSource(SOURCE_DELETION_VECTOR); + + Map> statisticsRowsBySnapshotId = + mayScanStatistics + ? AllPuffinFilesTable.statisticsRowsBySnapshotId( + metadata.statisticsFiles(), tableSchema) + : ImmutableMap.of(); + Map specsById = + mayScanDeletionVectors ? ImmutableMap.copyOf(metadata.specsById()) : ImmutableMap.of(); + + String location = + metadata.metadataFileLocation() != null + ? metadata.metadataFileLocation() + : table().location(); + ImmutableList.Builder tasks = ImmutableList.builder(); + + for (Snapshot snapshot : metadata.snapshots()) { + addStatisticsTask( + tasks, + baseTable, + projectedSchema, + location, + snapshot, + mayScanStatistics, + taskEvaluator, + statisticsRowsBySnapshotId); + + addDeletionVectorTask( + tasks, + baseTable, + tableSchema, + projectedSchema, + specsById, + residual, + snapshot, + mayScanDeletionVectors, + taskEvaluator); + } + return CloseableIterable.withNoopClose(tasks.build()); + } + + private static void addStatisticsTask( + ImmutableList.Builder tasks, + BaseTable baseTable, + Schema projectedSchema, + String location, + Snapshot snapshot, + boolean mayScanStatistics, + SnapshotReferenceEvaluator taskEvaluator, + Map> statisticsRowsBySnapshotId) { + + if (!mayScanStatistics || !taskEvaluator.mayMatch(snapshot, SOURCE_STATISTICS)) { + return; + } + + List statisticsRows = + statisticsRowsBySnapshotId.getOrDefault(snapshot.snapshotId(), ImmutableList.of()); + + if (!statisticsRows.isEmpty()) { + tasks.add( + StaticDataTask.of( + baseTable.io().newInputFile(location), + ALL_PUFFIN_FILES_SCHEMA, + projectedSchema, + statisticsRows, + row -> row)); + } + } + + private static void addDeletionVectorTask( + ImmutableList.Builder tasks, + BaseTable baseTable, + Schema tableSchema, + Schema projectedSchema, + Map specsById, + Expression residual, + Snapshot snapshot, + boolean mayScanDeletionVectors, + SnapshotReferenceEvaluator taskEvaluator) { + + String manifestListLocation = snapshot.manifestListLocation(); + boolean shouldScanDeletionVectors = + mayScanDeletionVectors + && manifestListLocation != null + && mayHaveDeleteFiles(snapshot) + && taskEvaluator.mayMatch(snapshot, SOURCE_DELETION_VECTOR); + + if (!shouldScanDeletionVectors) { + return; + } + + tasks.add( + new DeletionVectorPuffinFilesTask( + baseTable.io(), + tableSchema, + projectedSchema, + specsById, + residual, + snapshot.snapshotId(), + new BaseManifestListFile(manifestListLocation, snapshot.keyId()), + estimatedDvTaskSize(snapshot))); + } + } + + /** + * Conservatively evaluates a metadata-table filter using the snapshot ID and Puffin metadata + * source values that are known during task planning. + * + *

The evaluator makes exact decisions only for predicates over supported fields whose values + * are known in the current evaluation context. Predicates over unknown values or unsupported + * fields return {@link MatchResult#MIGHT_MATCH}. The three-state result is important for NOT: + * negating an unknown predicate is still unknown and must not prune a task. + */ + private static class SnapshotReferenceEvaluator { + + private final int referenceSnapshotIdFieldId; + private final int sourceFieldId; + private final Expression boundExpression; + + private SnapshotReferenceEvaluator( + Expression expression, + Types.StructType structType, + int referenceSnapshotIdFieldId, + int sourceFieldId, + boolean caseSensitive) { + this.referenceSnapshotIdFieldId = referenceSnapshotIdFieldId; + this.sourceFieldId = sourceFieldId; + this.boundExpression = Binder.bind(structType, expression, caseSensitive); + } + + private boolean mayMatch(Snapshot snapshot, String source) { + return new SnapshotEvalVisitor(snapshot.snapshotId(), source).eval() + != MatchResult.CANNOT_MATCH; + } + + private boolean mayMatchSource(String source) { + return new SnapshotEvalVisitor(source).eval() != MatchResult.CANNOT_MATCH; + } + + private enum MatchResult { + CANNOT_MATCH, + MIGHT_MATCH, + MUST_MATCH; + + private static MatchResult from(boolean matches) { + return matches ? MUST_MATCH : CANNOT_MATCH; + } + + private MatchResult negate() { + return switch (this) { + case CANNOT_MATCH -> MUST_MATCH; + case MUST_MATCH -> CANNOT_MATCH; + case MIGHT_MATCH -> MIGHT_MATCH; + }; + } + + private static MatchResult and(MatchResult left, MatchResult right) { + if (left == CANNOT_MATCH || right == CANNOT_MATCH) { + return CANNOT_MATCH; + } else if (left == MUST_MATCH && right == MUST_MATCH) { + return MUST_MATCH; + } + + return MIGHT_MATCH; + } + + private static MatchResult or(MatchResult left, MatchResult right) { + if (left == MUST_MATCH || right == MUST_MATCH) { + return MUST_MATCH; + } else if (left == CANNOT_MATCH && right == CANNOT_MATCH) { + return CANNOT_MATCH; + } + + return MIGHT_MATCH; + } + } + + private class SnapshotEvalVisitor extends BoundExpressionVisitor { + + private final boolean snapshotIdKnown; + private final long snapshotId; + private final String source; + + private SnapshotEvalVisitor(long snapshotId, String source) { + this.snapshotIdKnown = true; + this.snapshotId = snapshotId; + this.source = source; + } + + private SnapshotEvalVisitor(String source) { + this.snapshotIdKnown = false; + this.snapshotId = 0L; + this.source = source; + } + + private MatchResult eval() { + return ExpressionVisitors.visit(boundExpression, this); + } + + @Override + public MatchResult alwaysTrue() { + return MatchResult.MUST_MATCH; + } + + @Override + public MatchResult alwaysFalse() { + return MatchResult.CANNOT_MATCH; + } + + @Override + public MatchResult not(MatchResult result) { + return result.negate(); + } + + @Override + public MatchResult and(MatchResult leftResult, MatchResult rightResult) { + return MatchResult.and(leftResult, rightResult); + } + + @Override + public MatchResult or(MatchResult leftResult, MatchResult rightResult) { + return MatchResult.or(leftResult, rightResult); + } + + @Override + public MatchResult isNull(BoundReference reference) { + return isSupportedReference(reference) ? MatchResult.CANNOT_MATCH : MatchResult.MIGHT_MATCH; + } + + @Override + public MatchResult notNull(BoundReference reference) { + return isSupportedReference(reference) ? MatchResult.MUST_MATCH : MatchResult.MIGHT_MATCH; + } + + @Override + public MatchResult isNaN(BoundReference reference) { + return MatchResult.MIGHT_MATCH; + } + + @Override + public MatchResult notNaN(BoundReference reference) { + return MatchResult.MIGHT_MATCH; + } + + @Override + public MatchResult lt(BoundReference reference, Literal literal) { + return evaluateComparison(reference, literal, compareResult -> compareResult < 0); + } + + @Override + public MatchResult ltEq(BoundReference reference, Literal literal) { + return evaluateComparison(reference, literal, compareResult -> compareResult <= 0); + } + + @Override + public MatchResult gt(BoundReference reference, Literal literal) { + return evaluateComparison(reference, literal, compareResult -> compareResult > 0); + } + + @Override + public MatchResult gtEq(BoundReference reference, Literal literal) { + return evaluateComparison(reference, literal, compareResult -> compareResult >= 0); + } + + @Override + public MatchResult eq(BoundReference reference, Literal literal) { + return evaluateComparison(reference, literal, compareResult -> compareResult == 0); + } + + @Override + public MatchResult notEq(BoundReference reference, Literal literal) { + return eq(reference, literal).negate(); + } + + @Override + public MatchResult in(BoundReference reference, Set literalSet) { + if (isReferenceSnapshotId(reference)) { + if (!snapshotIdKnown) { + return MatchResult.MIGHT_MATCH; + } + + return MatchResult.from(literalSet.contains(snapshotId)); + } else if (isSource(reference)) { + return MatchResult.from(literalSet.contains(source)); + } + + return MatchResult.MIGHT_MATCH; + } + + @Override + public MatchResult notIn(BoundReference reference, Set literalSet) { + return in(reference, literalSet).negate(); + } + + @Override + public MatchResult startsWith(BoundReference reference, Literal literal) { + if (!isSource(reference)) { + return MatchResult.MIGHT_MATCH; + } + + Literal stringLiteral = literal.to(Types.StringType.get()); + Preconditions.checkState( + stringLiteral != null, "Cannot convert literal to string for source: %s", literal); + return MatchResult.from(source.startsWith(stringLiteral.value().toString())); + } + + @Override + public MatchResult notStartsWith(BoundReference reference, Literal literal) { + return startsWith(reference, literal).negate(); + } + + @Override + public MatchResult handleNonReference(Bound term) { + return MatchResult.MIGHT_MATCH; + } + + private MatchResult evaluateComparison( + BoundReference reference, Literal literal, IntPredicate matches) { + int compareResult; + if (isReferenceSnapshotId(reference)) { + if (!snapshotIdKnown) { + return MatchResult.MIGHT_MATCH; + } + + Literal longLiteral = literal.to(Types.LongType.get()); + Preconditions.checkState( + longLiteral != null, + "Cannot convert literal to long for reference_snapshot_id: %s", + literal); + compareResult = longLiteral.comparator().compare(snapshotId, longLiteral.value()); + + } else if (isSource(reference)) { + Literal stringLiteral = literal.to(Types.StringType.get()); + Preconditions.checkState( + stringLiteral != null, "Cannot convert literal to string for source: %s", literal); + compareResult = stringLiteral.comparator().compare(source, stringLiteral.value()); + + } else { + return MatchResult.MIGHT_MATCH; + } + + return MatchResult.from(matches.test(compareResult)); + } + + private boolean isSupportedReference(BoundReference reference) { + return isReferenceSnapshotId(reference) || isSource(reference); + } + + private boolean isReferenceSnapshotId(BoundReference reference) { + return reference.fieldId() == referenceSnapshotIdFieldId; + } + + private boolean isSource(BoundReference reference) { + return reference.fieldId() == sourceFieldId; + } + } + } + + private static class DeletionVectorPuffinFilesTask implements DataTask { + + private final FileIO io; + private final Schema tableSchema; + private final Schema projectedSchema; + private final Map specsById; + private final Expression residual; + private final long referenceSnapshotId; + private final ManifestListFile manifestList; + private final long estimatedTaskSize; + + private DataFile lazyDataFile = null; + + private DeletionVectorPuffinFilesTask( + FileIO io, + Schema tableSchema, + Schema projectedSchema, + Map specsById, + Expression residual, + long referenceSnapshotId, + ManifestListFile manifestList, + long estimatedTaskSize) { + this.io = io; + this.tableSchema = tableSchema; + this.projectedSchema = projectedSchema; + this.specsById = specsById; + this.residual = residual; + this.referenceSnapshotId = referenceSnapshotId; + this.manifestList = manifestList; + this.estimatedTaskSize = estimatedTaskSize; + } + + @Override + public List deletes() { + return ImmutableList.of(); + } + + @Override + public CloseableIterable rows() { + List puffinFiles = + scanDeletionVectorPuffinFiles(io, specsById, referenceSnapshotId, manifestList); + + CloseableIterable rows = + CloseableIterable.transform( + CloseableIterable.withNoopClose(puffinFiles), + puffinFile -> puffinFileToRow(tableSchema, puffinFile)); + + StructProjection projection = + StructProjection.create(ALL_PUFFIN_FILES_SCHEMA, projectedSchema); + + return CloseableIterable.transform(rows, projection::wrap); + } + + @Override + public DataFile file() { + if (lazyDataFile == null) { + // The actual result row count is unknown until the manifests are scanned. + this.lazyDataFile = + DataFiles.builder(PartitionSpec.unpartitioned()) + .withInputFile(io.newInputFile(manifestList)) + .withFormat(FileFormat.AVRO) + .withRecordCount(1L) + .build(); + } + + return lazyDataFile; + } + + @Override + public PartitionSpec spec() { + return PartitionSpec.unpartitioned(); + } + + @Override + public long start() { + return 0L; + } + + @Override + public long length() { + return estimatedTaskSize; + } + + @Override + public Expression residual() { + return residual; + } + + @Override + public Iterable split(long splitSize) { + return ImmutableList.of(this); + } + + @Override + public Schema schema() { + return projectedSchema; + } + } + + private record PuffinFileRow( + long referenceSnapshotId, + String path, + String source, + long fileSizeInBytes, + int referencedBlobCount, + List referencedBlobTypes, + List referencedFieldIds) { + + private static PuffinFileRow fromStatisticsFile(StatisticsFile statisticsFile) { + List blobs = statisticsFile.blobMetadata(); + + List referencedBlobTypes = + blobs.stream() + .map(BlobMetadata::type) + .distinct() + .collect(ImmutableList.toImmutableList()); + + List referencedFieldIds = + blobs.stream() + .flatMap(blob -> blob.fields().stream()) + .distinct() + .collect(ImmutableList.toImmutableList()); + + return new PuffinFileRow( + statisticsFile.snapshotId(), + statisticsFile.path(), + SOURCE_STATISTICS, + statisticsFile.fileSizeInBytes(), + blobs.size(), + referencedBlobTypes, + referencedFieldIds); + } + } + + private static class DeletionVectorFileAccumulator { + + private final long referenceSnapshotId; + private final String path; + private final long fileSizeInBytes; + private final Set referencedBlobs = Sets.newLinkedHashSet(); + + private DeletionVectorFileAccumulator( + long referenceSnapshotId, String path, long fileSizeInBytes) { + this.referenceSnapshotId = referenceSnapshotId; + this.path = path; + this.fileSizeInBytes = fileSizeInBytes; + } + + private void add(long entryFileSizeInBytes, long contentOffset, long contentSizeInBytes) { + Preconditions.checkState( + fileSizeInBytes == entryFileSizeInBytes, + "Deletion vectors in the same Puffin file have different file sizes: " + + "path=%s, referenceSnapshotId=%s, expected=%s, actual=%s", + path, + referenceSnapshotId, + fileSizeInBytes, + entryFileSizeInBytes); + referencedBlobs.add(new DeletionVectorBlobKey(contentOffset, contentSizeInBytes)); + } + + private PuffinFileRow toPuffinFileRow() { + return new PuffinFileRow( + referenceSnapshotId, + path, + SOURCE_DELETION_VECTOR, + fileSizeInBytes, + referencedBlobs.size(), + DV_REFERENCED_BLOB_TYPES, + DV_REFERENCED_FIELD_IDS); + } + } + + private record DeletionVectorBlobKey(long contentOffset, long contentSizeInBytes) {} +} diff --git a/core/src/main/java/org/apache/iceberg/MetadataTableType.java b/core/src/main/java/org/apache/iceberg/MetadataTableType.java index 733a11d900f1..f7994faa4d4a 100644 --- a/core/src/main/java/org/apache/iceberg/MetadataTableType.java +++ b/core/src/main/java/org/apache/iceberg/MetadataTableType.java @@ -36,7 +36,8 @@ public enum MetadataTableType { ALL_FILES, ALL_MANIFESTS, ALL_ENTRIES, - POSITION_DELETES; + POSITION_DELETES, + ALL_PUFFIN_FILES; public static MetadataTableType from(String name) { try { diff --git a/core/src/main/java/org/apache/iceberg/MetadataTableUtils.java b/core/src/main/java/org/apache/iceberg/MetadataTableUtils.java index adb0f18ba1ad..10290bbe2bd1 100644 --- a/core/src/main/java/org/apache/iceberg/MetadataTableUtils.java +++ b/core/src/main/java/org/apache/iceberg/MetadataTableUtils.java @@ -88,6 +88,8 @@ private static Table createMetadataTableInstance( return new AllEntriesTable(baseTable, metadataTableName); case POSITION_DELETES: return new PositionDeletesTable(baseTable, metadataTableName); + case ALL_PUFFIN_FILES: + return new AllPuffinFilesTable(baseTable, metadataTableName); default: throw new NoSuchTableException( "Unknown metadata table type: %s for %s", type, metadataTableName); diff --git a/core/src/test/java/org/apache/iceberg/TestAllPuffinFilesTable.java b/core/src/test/java/org/apache/iceberg/TestAllPuffinFilesTable.java new file mode 100644 index 000000000000..383f46457cda --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestAllPuffinFilesTable.java @@ -0,0 +1,863 @@ +/* + * 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 static org.assertj.core.api.Assumptions.assumeThat; + +import java.io.IOException; +import java.util.List; +import java.util.Set; +import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.ExpressionUtil; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.puffin.StandardBlobTypes; +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.Sets; +import org.apache.iceberg.types.Types; +import org.junit.jupiter.api.TestTemplate; +import org.junit.jupiter.api.extension.ExtendWith; + +@ExtendWith(ParameterizedTestExtension.class) +public class TestAllPuffinFilesTable extends TestBase { + + private static final String SOURCE_STATISTICS = "statistics"; + private static final String SOURCE_DELETION_VECTOR = "deletion_vector"; + + @TestTemplate + public void testSchema() { + Schema expected = + new Schema( + Types.NestedField.required( + 1, + "reference_snapshot_id", + Types.LongType.get(), + "ID of the snapshot that references the Puffin file"), + Types.NestedField.required( + 2, + "file_path", + Types.StringType.get(), + "Fully qualified location of the Puffin file"), + Types.NestedField.required( + 3, + "source", + Types.StringType.get(), + "Metadata source that associates the Puffin file with the reference snapshot"), + Types.NestedField.required( + 4, + "file_size_in_bytes", + Types.LongType.get(), + "Total size of the Puffin file in bytes"), + Types.NestedField.required( + 5, + "referenced_blob_count", + Types.IntegerType.get(), + "Number of referenced Puffin blobs represented by this row"), + Types.NestedField.required( + 6, + "referenced_blob_types", + Types.ListType.ofRequired(7, Types.StringType.get()), + "Distinct blob types across the referenced blobs"), + Types.NestedField.required( + 8, + "referenced_fields", + Types.ListType.ofRequired( + 9, + Types.StructType.of( + Types.NestedField.required( + 10, + "field_id", + Types.IntegerType.get(), + "Field ID referenced by at least one blob"), + Types.NestedField.optional( + 11, + "current_field_name", + Types.StringType.get(), + "Name currently assigned to the field ID; null if no longer present"))), + "Distinct fields referenced across the blobs")); + + assertThat(metadataTable().schema().sameSchema(expected)).isTrue(); + } + + @TestTemplate + public void testEmptyTable() throws IOException { + assertThat(rawTaskRows(metadataTable().newScan())).isEmpty(); + } + + @TestTemplate + public void testStatisticsPuffinFile() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + Snapshot snapshot = table.currentSnapshot(); + int idFieldId = fieldId("id"); + int dataFieldId = fieldId("data"); + + StatisticsFile statisticsFile = + statisticsFile( + snapshot, + "/path/to/statistics.puffin", + 100L, + blob(snapshot, "test-blob-v1", ImmutableList.of(idFieldId, dataFieldId))); + + table.updateStatistics().setStatistics(statisticsFile).commit(); + + assertThat(rawTaskRows(metadataTable().newScan())) + .containsExactly( + new PuffinRow( + snapshot.snapshotId(), + statisticsFile.path(), + SOURCE_STATISTICS, + statisticsFile.fileSizeInBytes(), + 1, + ImmutableList.of("test-blob-v1"), + ImmutableList.of( + new ReferencedField(idFieldId, "id"), + new ReferencedField(dataFieldId, "data")))); + } + + @TestTemplate + public void testStatisticsBlobAggregation() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + Snapshot snapshot = table.currentSnapshot(); + int idFieldId = fieldId("id"); + int dataFieldId = fieldId("data"); + + StatisticsFile statisticsFile = + statisticsFile( + snapshot, + "/path/to/aggregated-statistics.puffin", + 200L, + blob(snapshot, "type-a", ImmutableList.of(idFieldId, dataFieldId)), + blob(snapshot, "type-a", ImmutableList.of(dataFieldId)), + blob(snapshot, "type-b", ImmutableList.of(idFieldId))); + + table.updateStatistics().setStatistics(statisticsFile).commit(); + + PuffinRow row = onlyRow(metadataTable().newScan()); + assertThat(row.referencedBlobCount()).isEqualTo(3); + assertThat(row.referencedBlobTypes()).containsExactly("type-a", "type-b"); + assertThat(row.referencedFields()) + .containsExactly( + new ReferencedField(idFieldId, "id"), new ReferencedField(dataFieldId, "data")); + } + + @TestTemplate + public void testStatisticsReferenceSnapshotComesFromStatisticsFile() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + Snapshot blobSnapshot = table.currentSnapshot(); + + table.newFastAppend().appendFile(FILE_B).commit(); + Snapshot referenceSnapshot = table.currentSnapshot(); + + StatisticsFile statisticsFile = + statisticsFile( + referenceSnapshot, + "/path/to/cross-snapshot-statistics.puffin", + 100L, + blob(blobSnapshot, "test-type", ImmutableList.of(fieldId("id")))); + + table.updateStatistics().setStatistics(statisticsFile).commit(); + + PuffinRow row = onlyRow(metadataTable().newScan()); + assertThat(row.referenceSnapshotId()).isEqualTo(referenceSnapshot.snapshotId()); + } + + @TestTemplate + public void testStatisticsFileReplacementAndRemoval() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + Snapshot snapshot = table.currentSnapshot(); + int idFieldId = fieldId("id"); + int dataFieldId = fieldId("data"); + + StatisticsFile oldStatistics = + statisticsFile( + snapshot, + "/path/to/old-statistics.puffin", + 100L, + blob(snapshot, "old-type", ImmutableList.of(idFieldId))); + + StatisticsFile newStatistics = + statisticsFile( + snapshot, + "/path/to/new-statistics.puffin", + 200L, + blob(snapshot, "new-type", ImmutableList.of(dataFieldId))); + + table.updateStatistics().setStatistics(oldStatistics).commit(); + table.updateStatistics().setStatistics(newStatistics).commit(); + + assertThat(rawTaskRows(metadataTable().newScan())) + .containsExactly( + new PuffinRow( + snapshot.snapshotId(), + newStatistics.path(), + SOURCE_STATISTICS, + newStatistics.fileSizeInBytes(), + 1, + ImmutableList.of("new-type"), + ImmutableList.of(new ReferencedField(dataFieldId, "data")))); + + table.updateStatistics().removeStatistics(snapshot.snapshotId()).commit(); + + assertThat(rawTaskRows(metadataTable().newScan())).isEmpty(); + } + + @TestTemplate + public void testCurrentFieldNameAfterRename() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + Snapshot snapshot = table.currentSnapshot(); + int dataFieldId = fieldId("data"); + + table + .updateStatistics() + .setStatistics( + statisticsFile( + snapshot, + "/path/to/rename-statistics.puffin", + 100L, + blob(snapshot, "test-type", ImmutableList.of(dataFieldId)))) + .commit(); + + table.updateSchema().renameColumn("data", "renamed_data").commit(); + + assertThat(table.schema().findColumnName(dataFieldId)).isEqualTo("renamed_data"); + + PuffinRow row = onlyRow(metadataTable().newScan()); + assertThat(row.referencedFields()) + .containsExactly(new ReferencedField(dataFieldId, "renamed_data")); + } + + @TestTemplate + public void testCurrentFieldNameAfterDelete() throws IOException { + table.newFastAppend().appendFile(FILE_A).commit(); + Snapshot snapshot = table.currentSnapshot(); + int idFieldId = fieldId("id"); + + table + .updateStatistics() + .setStatistics( + statisticsFile( + snapshot, + "/path/to/deleted-field-statistics.puffin", + 100L, + blob(snapshot, "test-type", ImmutableList.of(idFieldId)))) + .commit(); + + table.updateSchema().deleteColumn("id").commit(); + + assertThat(table.schema().findColumnName(idFieldId)).isNull(); + + PuffinRow row = onlyRow(metadataTable().newScan()); + assertThat(row.referencedFields()).containsExactly(new ReferencedField(idFieldId, null)); + } + + @TestTemplate + public void testDeletionVectorPuffinFile() throws IOException { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).commit(); + table.newRowDelta().addDeletes(FILE_A_DV).commit(); + Snapshot deletionVectorSnapshot = table.currentSnapshot(); + + assertThat(rawTaskRows(metadataTable().newScan())) + .containsExactly( + new PuffinRow( + deletionVectorSnapshot.snapshotId(), + FILE_A_DV.location(), + SOURCE_DELETION_VECTOR, + FILE_A_DV.fileSizeInBytes(), + 1, + ImmutableList.of(StandardBlobTypes.DV_V1), + ImmutableList.of( + new ReferencedField( + MetadataColumns.ROW_POSITION.fieldId(), + MetadataColumns.ROW_POSITION.name())))); + } + + @TestTemplate + public void testMultipleDeletionVectorsInSamePuffinFile() throws IOException { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit(); + + String sharedPuffinPath = "/path/to/shared-deletion-vectors.puffin"; + DeleteFile dvA = deletionVector(FILE_A, sharedPuffinPath, 100L, 4L, 10L); + DeleteFile dvB = deletionVector(FILE_B, sharedPuffinPath, 100L, 20L, 12L); + + table.newRowDelta().addDeletes(dvA).addDeletes(dvB).commit(); + Snapshot deletionVectorSnapshot = table.currentSnapshot(); + + assertThat(rawTaskRows(metadataTable().newScan())) + .containsExactly( + new PuffinRow( + deletionVectorSnapshot.snapshotId(), + sharedPuffinPath, + SOURCE_DELETION_VECTOR, + 100L, + 2, + ImmutableList.of(StandardBlobTypes.DV_V1), + ImmutableList.of( + new ReferencedField( + MetadataColumns.ROW_POSITION.fieldId(), + MetadataColumns.ROW_POSITION.name())))); + } + + @TestTemplate + public void testDuplicateDeletionVectorBlobIsCountedOnce() throws IOException { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit(); + + String sharedPuffinPath = "/path/to/duplicate-deletion-vector.puffin"; + DeleteFile dvA = deletionVector(FILE_A, sharedPuffinPath, 100L, 4L, 10L); + DeleteFile dvB = deletionVector(FILE_B, sharedPuffinPath, 100L, 4L, 10L); + + table.newRowDelta().addDeletes(dvA).addDeletes(dvB).commit(); + + PuffinRow row = onlyRow(metadataTable().newScan()); + assertThat(row.referencedBlobCount()).isEqualTo(1); + } + + @TestTemplate + public void testDeletionVectorsInSamePuffinFileMustHaveConsistentFileSize() { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).appendFile(FILE_B).commit(); + + String sharedPuffinPath = "/path/to/inconsistent-deletion-vectors.puffin"; + DeleteFile dvA = deletionVector(FILE_A, sharedPuffinPath, 100L, 4L, 10L); + DeleteFile dvB = deletionVector(FILE_B, sharedPuffinPath, 200L, 20L, 12L); + + table.newRowDelta().addDeletes(dvA).addDeletes(dvB).commit(); + + assertThatThrownBy( + () -> + rawTaskRows( + metadataTable() + .newScan() + .filter(Expressions.equal("source", SOURCE_DELETION_VECTOR)))) + .isInstanceOf(IllegalStateException.class) + .hasMessageContaining("different file sizes") + .hasMessageContaining(sharedPuffinPath); + } + + @TestTemplate + public void testEqualityDeletesAreNotPuffinFiles() throws IOException { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).commit(); + + DeleteFile equalityDelete = + FileMetadata.deleteFileBuilder(SPEC) + .ofEqualityDeletes(fieldId("id")) + .withPath("/path/to/equality-deletes.parquet") + .withFileSizeInBytes(10L) + .withPartition(FILE_A.partition()) + .withRecordCount(1L) + .build(); + + table.newRowDelta().addDeletes(equalityDelete).commit(); + + assertThat( + rawTaskRows( + metadataTable() + .newScan() + .filter(Expressions.equal("source", SOURCE_DELETION_VECTOR)))) + .isEmpty(); + } + + @TestTemplate + public void testDeletionVectorRepeatedAcrossSnapshots() throws IOException { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).commit(); + + table.newRowDelta().addDeletes(FILE_A_DV).commit(); + long deletionVectorSnapshotId = table.currentSnapshot().snapshotId(); + + table.newFastAppend().appendFile(FILE_B).commit(); + long laterSnapshotId = table.currentSnapshot().snapshotId(); + + List deletionVectorRows = deletionVectorRows(); + + assertThat(deletionVectorRows).hasSize(2); + assertThat(deletionVectorRows) + .extracting(PuffinRow::referenceSnapshotId) + .containsExactlyInAnyOrder(deletionVectorSnapshotId, laterSnapshotId); + assertThat(deletionVectorRows) + .extracting(PuffinRow::filePath) + .containsOnly(FILE_A_DV.location()); + } + + @TestTemplate + public void testDeletionVectorRemovedFromLaterSnapshot() throws IOException { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).commit(); + + table.newRowDelta().addDeletes(FILE_A_DV).commit(); + long addedSnapshotId = table.currentSnapshot().snapshotId(); + + table.newFastAppend().appendFile(FILE_B).commit(); + long carriedSnapshotId = table.currentSnapshot().snapshotId(); + + table.newRowDelta().removeDeletes(FILE_A_DV).commit(); + long removedSnapshotId = table.currentSnapshot().snapshotId(); + + assertThat(deletionVectorRows()) + .extracting(PuffinRow::referenceSnapshotId) + .containsExactlyInAnyOrder(addedSnapshotId, carriedSnapshotId) + .doesNotContain(removedSnapshotId); + } + + @TestTemplate + public void testExpiredSnapshotIsNotReported() throws IOException { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).commit(); + + table.newRowDelta().addDeletes(FILE_A_DV).commit(); + long expiredSnapshotId = table.currentSnapshot().snapshotId(); + + table.newFastAppend().appendFile(FILE_B).commit(); + long currentSnapshotId = table.currentSnapshot().snapshotId(); + + table + .expireSnapshots() + .expireSnapshotId(expiredSnapshotId) + .cleanupLevel(ExpireSnapshots.CleanupLevel.NONE) + .commit(); + + assertThat(deletionVectorRows()) + .extracting(PuffinRow::referenceSnapshotId) + .containsExactly(currentSnapshotId); + } + + @TestTemplate + public void testSourcePredicatePruning() throws IOException { + assumeDeletionVectorsSupported(); + PreparedTable prepared = prepareStatisticsAndDeletionVector(); + + assertPlannedTaskFormats(Expressions.equal("source", SOURCE_STATISTICS), FileFormat.METADATA); + assertPlannedTaskFormats(Expressions.notEqual("source", SOURCE_STATISTICS), FileFormat.AVRO); + assertPlannedTaskFormats(Expressions.in("source", SOURCE_STATISTICS), FileFormat.METADATA); + assertPlannedTaskFormats(Expressions.notIn("source", SOURCE_STATISTICS), FileFormat.AVRO); + assertPlannedTaskFormats(Expressions.startsWith("source", "stat"), FileFormat.METADATA); + assertPlannedTaskFormats(Expressions.notStartsWith("source", "stat"), FileFormat.AVRO); + assertPlannedTaskFormats(Expressions.isNull("source")); + assertPlannedTaskFormats(Expressions.notNull("source"), FileFormat.METADATA, FileFormat.AVRO); + + FileScanTask deletionVectorTask = + onlyTaskWithFormat( + plannedTasks( + metadataTable() + .newScan() + .filter(Expressions.equal("source", SOURCE_DELETION_VECTOR))), + FileFormat.AVRO); + assertThat(deletionVectorTask.file().location()) + .isEqualTo(prepared.deletionVectorManifestList()); + } + + @TestTemplate + public void testSourcePruningUsesCaseInsensitiveBinding() throws IOException { + assumeDeletionVectorsSupported(); + prepareStatisticsAndDeletionVector(); + + List tasks = + plannedTasks( + metadataTable() + .newScan() + .caseSensitive(false) + .filter(Expressions.equal("SOURCE", SOURCE_STATISTICS))); + + assertThat(tasks).extracting(task -> task.file().format()).containsExactly(FileFormat.METADATA); + } + + @TestTemplate + public void testReferenceSnapshotIdPredicatePruning() throws IOException { + assumeDeletionVectorsSupported(); + PreparedTable prepared = prepareStatisticsAndDeletionVector(); + + assertPlannedTaskFormats( + Expressions.equal("reference_snapshot_id", prepared.statisticsSnapshotId()), + FileFormat.METADATA); + assertPlannedTaskFormats( + Expressions.notEqual("reference_snapshot_id", prepared.statisticsSnapshotId()), + FileFormat.AVRO); + assertPlannedTaskFormats( + Expressions.in("reference_snapshot_id", prepared.statisticsSnapshotId()), + FileFormat.METADATA); + assertPlannedTaskFormats( + Expressions.notIn("reference_snapshot_id", prepared.statisticsSnapshotId()), + FileFormat.AVRO); + assertPlannedTaskFormats(Expressions.isNull("reference_snapshot_id")); + assertPlannedTaskFormats( + Expressions.notNull("reference_snapshot_id"), FileFormat.METADATA, FileFormat.AVRO); + } + + @TestTemplate + public void testReferenceSnapshotIdComparisonPruning() throws IOException { + assumeDeletionVectorsSupported(); + PreparedTable prepared = prepareStatisticsAndDeletionVector(); + + long lowerSnapshotId = + Math.min(prepared.statisticsSnapshotId(), prepared.deletionVectorSnapshotId()); + long higherSnapshotId = + Math.max(prepared.statisticsSnapshotId(), prepared.deletionVectorSnapshotId()); + + assertPlannedTaskFormats( + Expressions.lessThan("reference_snapshot_id", higherSnapshotId), + formatForSnapshot(prepared, lowerSnapshotId)); + assertPlannedTaskFormats( + Expressions.greaterThanOrEqual("reference_snapshot_id", higherSnapshotId), + formatForSnapshot(prepared, higherSnapshotId)); + assertPlannedTaskFormats( + Expressions.lessThanOrEqual("reference_snapshot_id", lowerSnapshotId), + formatForSnapshot(prepared, lowerSnapshotId)); + assertPlannedTaskFormats( + Expressions.greaterThan("reference_snapshot_id", lowerSnapshotId), + formatForSnapshot(prepared, higherSnapshotId)); + } + + @TestTemplate + public void testBooleanPredicatePruning() throws IOException { + assumeDeletionVectorsSupported(); + PreparedTable prepared = prepareStatisticsAndDeletionVector(); + + Expression unknown = Expressions.equal("file_path", "/unknown/path.puffin"); + + assertPlannedTaskFormats( + Expressions.and(unknown, Expressions.equal("source", SOURCE_STATISTICS)), + FileFormat.METADATA); + assertPlannedTaskFormats(Expressions.and(unknown, Expressions.equal("source", "unknown"))); + assertPlannedTaskFormats( + Expressions.or(unknown, Expressions.equal("source", SOURCE_STATISTICS)), + FileFormat.METADATA, + FileFormat.AVRO); + assertPlannedTaskFormats(Expressions.not(unknown), FileFormat.METADATA, FileFormat.AVRO); + assertPlannedTaskFormats( + Expressions.not(Expressions.equal("source", SOURCE_STATISTICS)), FileFormat.AVRO); + assertPlannedTaskFormats( + Expressions.and( + Expressions.equal("source", SOURCE_STATISTICS), + Expressions.equal("reference_snapshot_id", prepared.statisticsSnapshotId())), + FileFormat.METADATA); + assertPlannedTaskFormats( + Expressions.or( + Expressions.equal("source", SOURCE_STATISTICS), + Expressions.equal("reference_snapshot_id", prepared.deletionVectorSnapshotId())), + FileFormat.METADATA, + FileFormat.AVRO); + } + + @TestTemplate + public void testDeletionVectorTaskPreservesUnknownPredicateAsResidual() throws IOException { + assumeDeletionVectorsSupported(); + prepareStatisticsAndDeletionVector(); + + Expression filter = Expressions.greaterThan("referenced_blob_count", 100); + List tasks = plannedTasks(metadataTable().newScan().filter(filter)); + FileScanTask deletionVectorTask = onlyTaskWithFormat(tasks, FileFormat.AVRO); + + assertThat( + ExpressionUtil.equivalent( + filter, deletionVectorTask.residual(), metadataTable().schema().asStruct(), true)) + .isTrue(); + } + + @TestTemplate + public void testProjection() throws IOException { + assumeDeletionVectorsSupported(); + PreparedTable prepared = prepareStatisticsAndDeletionVector(); + + Table metadataTable = metadataTable(); + TableScan scan = metadataTable.newScan().select("reference_snapshot_id", "source"); + Schema expectedSchema = metadataTable.schema().select("reference_snapshot_id", "source"); + + assertThat(scan.schema().sameSchema(expectedSchema)) + .as("Projected scan schema should match the selected metadata-table schema") + .isTrue(); + + List projectedRows = Lists.newArrayList(); + try (CloseableIterable tasks = scan.planFiles()) { + for (FileScanTask task : tasks) { + try (CloseableIterable taskRows = task.asDataTask().rows()) { + for (StructLike row : taskRows) { + assertThat(row.size()).isEqualTo(2); + projectedRows.add( + new SnapshotSourceRow( + row.get(0, Long.class), row.get(1, CharSequence.class).toString())); + } + } + } + } + + assertThat(projectedRows) + .containsExactlyInAnyOrder( + new SnapshotSourceRow(prepared.statisticsSnapshotId(), SOURCE_STATISTICS), + new SnapshotSourceRow(prepared.deletionVectorSnapshotId(), SOURCE_DELETION_VECTOR)); + } + + @TestTemplate + public void testReferencedFieldsProjection() throws IOException { + assumeDeletionVectorsSupported(); + prepareStatisticsAndDeletionVector(); + + Table metadataTable = metadataTable(); + TableScan scan = metadataTable.newScan().select("referenced_fields"); + Schema expectedSchema = metadataTable.schema().select("referenced_fields"); + + assertThat(scan.schema().sameSchema(expectedSchema)).isTrue(); + + Set referencedFields = Sets.newHashSet(); + try (CloseableIterable tasks = scan.planFiles()) { + for (FileScanTask task : tasks) { + try (CloseableIterable taskRows = task.asDataTask().rows()) { + for (StructLike row : taskRows) { + assertThat(row.size()).isEqualTo(1); + + List fields = row.get(0, List.class); + for (Object fieldObject : fields) { + StructLike field = (StructLike) fieldObject; + assertThat(field.size()).isEqualTo(2); + + CharSequence currentFieldName = field.get(1, CharSequence.class); + referencedFields.add( + new ReferencedField( + field.get(0, Integer.class), + currentFieldName != null ? currentFieldName.toString() : null)); + } + } + } + } + } + + assertThat(referencedFields) + .containsExactlyInAnyOrder( + new ReferencedField(fieldId("id"), "id"), + new ReferencedField( + MetadataColumns.ROW_POSITION.fieldId(), MetadataColumns.ROW_POSITION.name())); + } + + @TestTemplate + public void testDeletionVectorTaskHonorsIgnoreResiduals() throws IOException { + assumeDeletionVectorsSupported(); + + table.newFastAppend().appendFile(FILE_A).commit(); + table.newRowDelta().addDeletes(FILE_A_DV).commit(); + + Expression filter = Expressions.greaterThan("referenced_blob_count", 0); + + FileScanTask task = onlyTask(plannedTasks(metadataTable().newScan().filter(filter))); + assertThat( + ExpressionUtil.equivalent( + filter, task.residual(), metadataTable().schema().asStruct(), true)) + .isTrue(); + + FileScanTask taskWithoutResidual = + onlyTask(plannedTasks(metadataTable().newScan().filter(filter).ignoreResiduals())); + + assertThat(taskWithoutResidual.residual()).isEqualTo(Expressions.alwaysTrue()); + } + + private FileScanTask onlyTask(List tasks) { + assertThat(tasks).hasSize(1); + return tasks.get(0); + } + + private Table metadataTable() { + return new AllPuffinFilesTable(table); + } + + private int fieldId(String name) { + Types.NestedField field = table.schema().findField(name); + assertThat(field).as("Field %s should exist in the current table schema", name).isNotNull(); + return field.fieldId(); + } + + private void assumeDeletionVectorsSupported() { + assumeThat(formatVersion) + .as("Deletion vectors require format version 3 or later") + .isGreaterThanOrEqualTo(3); + } + + private PreparedTable prepareStatisticsAndDeletionVector() { + table.newFastAppend().appendFile(FILE_A).commit(); + Snapshot statisticsSnapshot = table.currentSnapshot(); + + table + .updateStatistics() + .setStatistics( + statisticsFile( + statisticsSnapshot, + "/path/to/source-pruning-statistics.puffin", + 100L, + blob(statisticsSnapshot, "test-type", ImmutableList.of(fieldId("id"))))) + .commit(); + + table.newRowDelta().addDeletes(FILE_A_DV).commit(); + Snapshot deletionVectorSnapshot = table.currentSnapshot(); + + return new PreparedTable( + statisticsSnapshot.snapshotId(), + deletionVectorSnapshot.snapshotId(), + deletionVectorSnapshot.manifestListLocation()); + } + + private List deletionVectorRows() throws IOException { + return rawTaskRows( + metadataTable().newScan().filter(Expressions.equal("source", SOURCE_DELETION_VECTOR))); + } + + private void assertPlannedTaskFormats(Expression filter, FileFormat... expectedFormats) + throws IOException { + assertThat(plannedTasks(metadataTable().newScan().filter(filter))) + .extracting(task -> task.file().format()) + .containsExactlyInAnyOrder(expectedFormats); + } + + private FileFormat formatForSnapshot(PreparedTable prepared, long snapshotId) { + if (snapshotId == prepared.statisticsSnapshotId()) { + return FileFormat.METADATA; + } + + assertThat(snapshotId).isEqualTo(prepared.deletionVectorSnapshotId()); + return FileFormat.AVRO; + } + + private FileScanTask onlyTaskWithFormat(List tasks, FileFormat format) { + List matchingTasks = + tasks.stream() + .filter(task -> task.file().format() == format) + .collect(ImmutableList.toImmutableList()); + + assertThat(matchingTasks).hasSize(1); + return matchingTasks.get(0); + } + + private static StatisticsFile statisticsFile( + Snapshot snapshot, String path, long fileSizeInBytes, BlobMetadata... blobs) { + return new GenericStatisticsFile( + snapshot.snapshotId(), path, fileSizeInBytes, 10L, ImmutableList.copyOf(blobs)); + } + + private static BlobMetadata blob(Snapshot snapshot, String type, List fields) { + return new GenericBlobMetadata( + type, snapshot.snapshotId(), snapshot.sequenceNumber(), fields, ImmutableMap.of()); + } + + private static DeleteFile deletionVector( + DataFile referencedDataFile, + String puffinPath, + long fileSizeInBytes, + long contentOffset, + long contentSizeInBytes) { + return FileMetadata.deleteFileBuilder(SPEC) + .ofPositionDeletes() + .withPath(puffinPath) + .withFileSizeInBytes(fileSizeInBytes) + .withPartition(referencedDataFile.partition()) + .withRecordCount(1L) + .withReferencedDataFile(referencedDataFile.location()) + .withContentOffset(contentOffset) + .withContentSizeInBytes(contentSizeInBytes) + .build(); + } + + private PuffinRow onlyRow(TableScan scan) throws IOException { + List rows = rawTaskRows(scan); + assertThat(rows).hasSize(1); + return rows.get(0); + } + + private List plannedTasks(TableScan scan) throws IOException { + try (CloseableIterable tasks = scan.planFiles()) { + return Lists.newArrayList(tasks); + } + } + + /** Returns raw data-task rows without evaluating task residuals. */ + private List rawTaskRows(TableScan scan) throws IOException { + List results = Lists.newArrayList(); + + try (CloseableIterable tasks = scan.planFiles()) { + for (FileScanTask task : tasks) { + try (CloseableIterable taskRows = task.asDataTask().rows()) { + for (StructLike row : taskRows) { + results.add(copyRow(row)); + } + } + } + } + + return results; + } + + private static PuffinRow copyRow(StructLike row) { + CharSequence filePath = row.get(1, CharSequence.class); + CharSequence source = row.get(2, CharSequence.class); + + List rawBlobTypes = row.get(5, List.class); + List blobTypes = + rawBlobTypes.stream().map(Object::toString).collect(ImmutableList.toImmutableList()); + + List rawReferencedFields = row.get(6, List.class); + List referencedFields = + rawReferencedFields.stream() + .map(StructLike.class::cast) + .map( + field -> { + CharSequence currentName = field.get(1, CharSequence.class); + return new ReferencedField( + field.get(0, Integer.class), + currentName != null ? currentName.toString() : null); + }) + .collect(ImmutableList.toImmutableList()); + + return new PuffinRow( + row.get(0, Long.class), + filePath.toString(), + source.toString(), + row.get(3, Long.class), + row.get(4, Integer.class), + blobTypes, + referencedFields); + } + + private record PreparedTable( + long statisticsSnapshotId, + long deletionVectorSnapshotId, + String deletionVectorManifestList) {} + + private record SnapshotSourceRow(long referenceSnapshotId, String source) {} + + private record PuffinRow( + long referenceSnapshotId, + String filePath, + String source, + long fileSizeInBytes, + int referencedBlobCount, + List referencedBlobTypes, + List referencedFields) {} + + private record ReferencedField(int fieldId, String currentFieldName) {} +} diff --git a/core/src/test/java/org/apache/iceberg/hadoop/TestTableSerialization.java b/core/src/test/java/org/apache/iceberg/hadoop/TestTableSerialization.java index ece9b24af3d1..aff5e353edc4 100644 --- a/core/src/test/java/org/apache/iceberg/hadoop/TestTableSerialization.java +++ b/core/src/test/java/org/apache/iceberg/hadoop/TestTableSerialization.java @@ -35,6 +35,7 @@ import java.util.Set; import org.apache.iceberg.BaseTable; import org.apache.iceberg.FileScanTask; +import org.apache.iceberg.GenericStatisticsFile; import org.apache.iceberg.HasTableOperations; import org.apache.iceberg.MetadataTableType; import org.apache.iceberg.MetadataTableUtils; @@ -42,6 +43,7 @@ import org.apache.iceberg.PositionDeletesTable; import org.apache.iceberg.ScanTask; import org.apache.iceberg.SerializableTable; +import org.apache.iceberg.Snapshot; import org.apache.iceberg.StaticTableOperations; import org.apache.iceberg.Table; import org.apache.iceberg.TableProperties; @@ -49,6 +51,7 @@ import org.apache.iceberg.TestHelpers; import org.apache.iceberg.Transaction; import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.Types; @@ -183,6 +186,18 @@ public void testSerializableMetadataTablesPlanning(boolean fromSerialized) throw table.newAppend().appendFile(FILE_B).commit(); table.newRowDelta().addDeletes(FILE_B_DELETES).commit(); + Snapshot currentSnapshot = table.currentSnapshot(); + table + .updateStatistics() + .setStatistics( + new GenericStatisticsFile( + currentSnapshot.snapshotId(), + "/path/to/new-statistics.puffin", + 100L, + 10L, + ImmutableList.of())) + .commit(); + for (MetadataTableType type : MetadataTableType.values()) { // Collect the deserialized data Set deserializedFiles = getFiles(deserializeFromBytes(serialized.get(type))); diff --git a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java index 117bf7fac24b..74488e3e4d2d 100644 --- a/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java +++ b/core/src/test/java/org/apache/iceberg/rest/TestRESTScanPlanning.java @@ -382,6 +382,8 @@ public void metadataTablesWithRemotePlanning(MetadataTableType type) { if (type.equals(MetadataTableType.POSITION_DELETES)) { // Position deletes table only uses batch scan assertThat(metadataTableInstance.newBatchScan().planFiles()).isNotEmpty(); + } else if (type.equals(MetadataTableType.ALL_PUFFIN_FILES)) { + assertThat(metadataTableInstance.newScan().planFiles()).isEmpty(); } else { assertThat(metadataTableInstance.newScan().planFiles()).isNotEmpty(); }