diff --git a/core/src/main/java/org/apache/iceberg/ContentStatsStruct.java b/core/src/main/java/org/apache/iceberg/ContentStatsStruct.java index 816a0d8c255b..60831a2a63ba 100644 --- a/core/src/main/java/org/apache/iceberg/ContentStatsStruct.java +++ b/core/src/main/java/org/apache/iceberg/ContentStatsStruct.java @@ -49,7 +49,9 @@ private ContentStatsStruct(ContentStatsStruct toCopy, Set fieldIds) { } } else { for (Map.Entry> entry : toCopy.idToFieldStats.entrySet()) { - idToFieldStats.put(entry.getKey(), entry.getValue().copy()); + if (entry.getValue() != null) { + idToFieldStats.put(entry.getKey(), entry.getValue().copy()); + } } } } diff --git a/core/src/main/java/org/apache/iceberg/FieldStatsStruct.java b/core/src/main/java/org/apache/iceberg/FieldStatsStruct.java index 44c2e2009cd2..348f6204d544 100644 --- a/core/src/main/java/org/apache/iceberg/FieldStatsStruct.java +++ b/core/src/main/java/org/apache/iceberg/FieldStatsStruct.java @@ -26,6 +26,7 @@ import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.iceberg.util.ByteBuffers; +import org.apache.iceberg.util.StructLikeUtil; class FieldStatsStruct implements FieldStats, StructLike, Serializable { private final Types.StructType struct; @@ -45,7 +46,7 @@ class FieldStatsStruct implements FieldStats, StructLike, Serializable { this.struct = struct; this.posToOffset = posToOffset(struct); this.fieldId = StatsUtil.toFieldId(struct.fields().get(0).fieldId()); - this.boundType = struct.fieldType("lower_bound"); + this.boundType = struct.fieldType(StatsUtil.LOWER_BOUND_NAME); } FieldStatsStruct( @@ -69,15 +70,8 @@ class FieldStatsStruct implements FieldStats, StructLike, Serializable { private FieldStatsStruct(FieldStatsStruct toCopy) { this(toCopy.struct); - // bounds are stored using the internal representation, which is a byte array for binary types - this.lowerBound = - toCopy.lowerBound instanceof byte[] - ? copyOf((byte[]) toCopy.lowerBound) - : toCopy.lowerBound; - this.upperBound = - toCopy.upperBound instanceof byte[] - ? copyOf((byte[]) toCopy.upperBound) - : toCopy.upperBound; + this.lowerBound = copyBound(toCopy.lowerBound); + this.upperBound = copyBound(toCopy.upperBound); this.tightBounds = toCopy.tightBounds; this.valueCount = toCopy.valueCount; this.nullValueCount = toCopy.nullValueCount; @@ -234,6 +228,22 @@ private static int[] posToOffset(Types.StructType struct) { return posToOffset; } + /** + * Copies a bound stored using its internal representation. + * + *

Binary bounds are byte arrays and geo bounds are bounding box structs. Both are mutable and + * readers may reuse them across rows, so both are copied. All other bounds are immutable. + */ + private static Object copyBound(Object bound) { + if (bound instanceof byte[] bytes) { + return copyOf(bytes); + } else if (bound instanceof StructLike struct) { + return StructLikeUtil.copy(struct); + } + + return bound; + } + private static byte[] copyOf(byte[] array) { return Arrays.copyOf(array, array.length); } diff --git a/core/src/main/java/org/apache/iceberg/StatsUtil.java b/core/src/main/java/org/apache/iceberg/StatsUtil.java index 545275002c9a..327bb3e3b79c 100644 --- a/core/src/main/java/org/apache/iceberg/StatsUtil.java +++ b/core/src/main/java/org/apache/iceberg/StatsUtil.java @@ -129,7 +129,7 @@ static int statOffset(int statId) { } public static Types.NestedField contentStatsField(Types.StructType contentStats) { - return optional(146, "content_stats", contentStats); + return optional(TrackedFile.CONTENT_STATS_ID, TrackedFile.CONTENT_STATS_NAME, contentStats); } public static Types.StructType statsWriteSchema(Schema tableSchema, MetricsConfig metricsConfig) { diff --git a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java index f5ce03c7eb62..8a222ddc5404 100644 --- a/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java +++ b/core/src/main/java/org/apache/iceberg/TrackedFileStruct.java @@ -85,7 +85,7 @@ class TrackedFileStruct extends SupportsIndexProjection implements TrackedFile, super(BASE_TYPE, projection); // partition type may be null if the field was not projected, or unknown for unpartitioned // manifests - Type partType = projection.fieldType("partition"); + Type partType = projection.fieldType(TrackedFile.PARTITION_NAME); if (partType != null && partType.isStructType()) { this.partitionData = new PartitionData(partType.asStructType()); } diff --git a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java index a823454845dd..9e3e2b56ee77 100644 --- a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java +++ b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java @@ -22,6 +22,7 @@ import java.util.Collection; import java.util.Map; import java.util.Set; +import org.apache.iceberg.expressions.Binder; import org.apache.iceberg.expressions.Evaluator; import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.expressions.Expressions; @@ -32,11 +33,14 @@ import org.apache.iceberg.io.InputFile; import org.apache.iceberg.metrics.ScanMetrics; 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.ImmutableSet; 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.TypeUtil; import org.apache.iceberg.types.Types; +import org.apache.iceberg.util.ArrayUtil; import org.apache.iceberg.util.Pair; import org.apache.iceberg.util.StructProjection; @@ -63,8 +67,9 @@ private V4ManifestReader( this.scanMetrics = scanMetrics; } - static Builder builder(InputFile file, Map specsById) { - return new Builder(file, specsById); + static Builder builder( + InputFile file, Schema tableSchema, Map specsById) { + return new Builder(file, tableSchema, specsById); } /** Returns copies of the tracked files that match this reader's configured filters. */ @@ -122,7 +127,7 @@ private CloseableIterable open() { Preconditions.checkArgument( format != null, "Cannot determine format of manifest: %s", file.location()); - CloseableIterable reader = + InternalData.ReadBuilder readBuilder = InternalData.read(format, file) .project(readSchema) .setRootType(TrackedFileStruct.class) @@ -130,8 +135,19 @@ private CloseableIterable open() { .setCustomType(TrackedFile.DELETION_VECTOR.fieldId(), DeletionVectorStruct.class) .setCustomType(TrackedFile.MANIFEST_INFO.fieldId(), ManifestInfoStruct.class) .setCustomType(TrackedFile.PARTITION_ID, PartitionData.class) - .reuseContainers() - .build(); + .reuseContainers(); + + // content_stats is missing from the read schema when no stats are read + Types.NestedField statsField = readSchema.findField(TrackedFile.CONTENT_STATS_ID); + if (statsField != null) { + readBuilder.setCustomType(TrackedFile.CONTENT_STATS_ID, ContentStatsStruct.class); + // content_stats holds one stats struct per projected column + for (Types.NestedField fieldStats : statsField.type().asStructType().fields()) { + readBuilder.setCustomType(fieldStats.fieldId(), FieldStatsStruct.class); + } + } + + CloseableIterable reader = readBuilder.build(); addCloseable(reader); return reader; } @@ -153,26 +169,24 @@ private static boolean isManifest(TrackedFile trackedFile) { static class Builder { private final InputFile file; + private final Schema tableSchema; private final Types.StructType unionPartitionType; private final Map specsById; - private final Schema fullSchema; private Expression rowFilter = Expressions.alwaysTrue(); private boolean caseSensitive = true; private boolean includeAll = false; private boolean scanPlanning = false; private Collection columns = null; private Schema requestedProjection = null; + private Set statsProjectionForFieldIds = null; private ScanMetrics scanMetrics = ScanMetrics.noop(); - private Builder(InputFile file, Map specsById) { + private Builder(InputFile file, Schema tableSchema, Map specsById) { + Preconditions.checkArgument(tableSchema != null, "Invalid table schema: null"); this.file = file; + this.tableSchema = tableSchema; this.specsById = specsById; this.unionPartitionType = Partitioning.unionPartitionTypes(specsById.values()); - Schema base = TrackedFile.schema(unionPartitionType, Types.StructType.of()); - // the read schema carries row_position (via BASE_TYPE) so the reader can fill manifestPos - this.fullSchema = - TypeUtil.replaceFieldTypes( - base, ImmutableMap.of(TrackedFile.TRACKING.fieldId(), TrackingStruct.BASE_TYPE)); } /** Sets a filter; files that cannot match the expression are skipped. */ @@ -228,6 +242,25 @@ Builder project(Schema newProjection) { return this; } + /** + * Reads content stats for the given table field IDs instead of for every field. Stats for + * fields referenced by the {@link #filter(Expression) filter} are always read. + */ + Builder projectStats(int... fieldIds) { + Preconditions.checkArgument(fieldIds != null, "Invalid stats projection for field IDs: null"); + return projectStats(ArrayUtil.toIntList(fieldIds)); + } + + /** + * Reads content stats for the given table field IDs instead of for every field. Stats for + * fields referenced by the {@link #filter(Expression) filter} are always read. + */ + Builder projectStats(Iterable fieldIds) { + Preconditions.checkArgument(fieldIds != null, "Invalid stats projection for field IDs: null"); + this.statsProjectionForFieldIds = ImmutableSet.copyOf(fieldIds); + return this; + } + Builder scanMetrics(ScanMetrics newScanMetrics) { Preconditions.checkArgument(newScanMetrics != null, "Invalid scan metrics: null"); this.scanMetrics = newScanMetrics; @@ -254,6 +287,8 @@ V4ManifestReader build() { } private Schema readSchema(boolean hasPartitionFilter) { + Set requiredFieldIds = requiredStatsProjectionForFieldIds(); + Schema fullSchema = fullSchema(requiredFieldIds); if (scanPlanning) { // scan planning does not read the change-tracking fields omitted by SCAN_TYPE return TypeUtil.replaceFieldTypes( @@ -263,17 +298,65 @@ private Schema readSchema(boolean hasPartitionFilter) { if (columns != null) { Schema selected = caseSensitive ? fullSchema.select(columns) : fullSchema.caseInsensitiveSelect(columns); - return addRequiredColumns(selected, hasPartitionFilter); + return addRequiredColumns(fullSchema, selected, requiredFieldIds, hasPartitionFilter); } if (requestedProjection != null) { - return addRequiredColumns(requestedProjection, hasPartitionFilter); + return addRequiredColumns( + fullSchema, requestedProjection, requiredFieldIds, hasPartitionFilter); } return fullSchema; } - private Schema addRequiredColumns(Schema projection, boolean hasPartitionFilter) { + /** Returns the schema of everything this reader may read, including content stats. */ + private Schema fullSchema(Set requiredStatsProjectionFieldIds) { + Types.StructType contentStatsType = contentStatsType(requiredStatsProjectionFieldIds); + Schema base = TrackedFile.schema(unionPartitionType, contentStatsType); + if (contentStatsType.fields().isEmpty()) { + // schema uses the unknown type for empty stats, which cannot be paired with the stats + // struct in the manifest, so drop the field instead of reading it as unknown + base = TypeUtil.selectNot(base, ImmutableSet.of(TrackedFile.CONTENT_STATS_ID)); + } + + // the read schema carries row_position (via BASE_TYPE) so the reader can fill manifestPos + return TypeUtil.replaceFieldTypes( + base, ImmutableMap.of(TrackedFile.TRACKING.fieldId(), TrackingStruct.BASE_TYPE)); + } + + /** Returns the stats type to read, which is empty when no stats are needed. */ + private Types.StructType contentStatsType(Set requiredStatsProjectionForFieldIds) { + if (scanPlanning || statsProjectionForFieldIds != null) { + // scan planning and projectStats(fieldIds) both narrow the set of stats that are read + return StatsUtil.statsReadSchema(tableSchema, requiredStatsProjectionForFieldIds); + } + + return StatsUtil.statsReadSchema( + tableSchema, TypeUtil.indexById(tableSchema.asStruct()).keySet()); + } + + /** Returns the IDs of table fields whose stats are read regardless of the projection. */ + private Set requiredStatsProjectionForFieldIds() { + Set fieldIds = Sets.newHashSet(); + if (statsProjectionForFieldIds != null) { + fieldIds.addAll(statsProjectionForFieldIds); + } + + if (rowFilter != Expressions.alwaysTrue()) { + // stats for filter references are read so that the filter can be evaluated against them + fieldIds.addAll( + Binder.boundReferences( + tableSchema.asStruct(), ImmutableList.of(rowFilter), caseSensitive)); + } + + return fieldIds; + } + + private Schema addRequiredColumns( + Schema fullSchema, + Schema projection, + Set requiredStatsProjectionFieldIds, + boolean hasPartitionFilter) { Set projectedIds = Sets.newHashSet(TypeUtil.getProjectedIds(projection)); // fields the reader consumes internally: status for liveness filtering, row_position for @@ -293,6 +376,15 @@ private Schema addRequiredColumns(Schema projection, boolean hasPartitionFilter) projectedIds.addAll(TypeUtil.getProjectedIds(unionPartitionType)); } + // stats needed by the filter or requested by projectStats are read even when the caller's + // projection omits them + Types.StructType requiredStatsType = + StatsUtil.statsReadSchema(tableSchema, requiredStatsProjectionFieldIds); + if (!requiredStatsType.fields().isEmpty()) { + projectedIds.add(TrackedFile.CONTENT_STATS_ID); + projectedIds.addAll(TypeUtil.getProjectedIds(requiredStatsType)); + } + // project instead of select to preserve narrow struct projections from the caller return TypeUtil.project(fullSchema, projectedIds); } diff --git a/core/src/main/java/org/apache/iceberg/util/StructLikeUtil.java b/core/src/main/java/org/apache/iceberg/util/StructLikeUtil.java index 5285793a4aad..c3a728f81097 100644 --- a/core/src/main/java/org/apache/iceberg/util/StructLikeUtil.java +++ b/core/src/main/java/org/apache/iceberg/util/StructLikeUtil.java @@ -18,6 +18,7 @@ */ package org.apache.iceberg.util; +import java.io.Serializable; import org.apache.iceberg.StructLike; public class StructLikeUtil { @@ -28,7 +29,7 @@ public static StructLike copy(StructLike struct) { return StructCopy.copy(struct); } - private static class StructCopy implements StructLike { + private static class StructCopy implements StructLike, Serializable { private static StructLike copy(StructLike struct) { return struct != null ? new StructCopy(struct) : null; } diff --git a/core/src/test/java/org/apache/iceberg/TestFieldStatsStruct.java b/core/src/test/java/org/apache/iceberg/TestFieldStatsStruct.java index 936fc43c77c3..9f652ed97a13 100644 --- a/core/src/test/java/org/apache/iceberg/TestFieldStatsStruct.java +++ b/core/src/test/java/org/apache/iceberg/TestFieldStatsStruct.java @@ -317,6 +317,10 @@ public void geoSerialization(Type geoType, RoundTripSerializer roundTrippedCopy = serializer.apply(stats.copy()); + assertThat(comparator.compare(roundTrippedCopy, stats)).isEqualTo(0); } // Variant is not Serializable so this does not test Java serialization diff --git a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java index c8a5cfd61a31..e4b83c6ae0de 100644 --- a/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java +++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReader.java @@ -209,14 +209,16 @@ public void statusFiltering(FileFormat format) throws IOException { InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS).build()) { assertThat(reader) .extracting(file -> file.tracking().status()) .containsExactly(EntryStatus.ADDED, EntryStatus.EXISTING, EntryStatus.MODIFIED); } try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).includeAll().build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .includeAll() + .build()) { assertThat(reader) .extracting(file -> file.tracking().status()) .containsExactly( @@ -258,7 +260,8 @@ public void selectiveReadReturnsOnlyRequestedFields( InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); - V4ManifestReader.Builder builder = V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS); + V4ManifestReader.Builder builder = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS); configureRead.accept(builder); try (V4ManifestReader reader = builder.build()) { TrackedFile actual = Iterables.getOnlyElement(reader); @@ -316,7 +319,7 @@ public void rowFilterForcesRecordCount(FileFormat format) throws IOException { // even though the caller selected only location Schema projection = new Schema(TrackedFile.LOCATION); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .project(projection) .filter(Expressions.equal("id", 1)) .build()) { @@ -332,7 +335,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .select("location") .project(new Schema(TrackedFile.LOCATION))) .isInstanceOf(IllegalStateException.class) @@ -340,7 +343,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .project(new Schema(TrackedFile.LOCATION)) .select("location")) .isInstanceOf(IllegalStateException.class) @@ -348,7 +351,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .forScanPlanning() .select("location")) .isInstanceOf(IllegalStateException.class) @@ -356,7 +359,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .select("location") .forScanPlanning()) .isInstanceOf(IllegalStateException.class) @@ -365,7 +368,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .forScanPlanning() .project(new Schema(TrackedFile.LOCATION))) .isInstanceOf(IllegalStateException.class) @@ -373,7 +376,7 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .project(new Schema(TrackedFile.LOCATION)) .forScanPlanning()) .isInstanceOf(IllegalStateException.class) @@ -388,7 +391,9 @@ public void projectionPreservesNarrowTrackingProjection(FileFormat format) throw writeManifest(format, EMPTY_PARTITION, ImmutableList.of(FILE_WITH_FULL_TRACKING)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).select("tracking.status").build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .select("tracking.status") + .build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); assertThat(actual.status()).isEqualTo(EntryStatus.ADDED); // the narrow tracking projection is not widened to the full tracking type @@ -406,7 +411,9 @@ public void forScanPlanningOmitsChangeTrackingFields(FileFormat format) throws I writeManifest(format, EMPTY_PARTITION, ImmutableList.of(FILE_WITH_FULL_TRACKING)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).forScanPlanning().build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .forScanPlanning() + .build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); // scan-relevant tracking fields are projected assertThat(actual.status()).isEqualTo(EntryStatus.ADDED); @@ -430,7 +437,7 @@ public void defaultReadsFullTracking(FileFormat format) throws IOException { // without scanPlanning, select, or project, the reader returns the full schema for copying to // other manifests, including the change-tracking fields try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS).build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); assertThat(actual.status()).isEqualTo(EntryStatus.ADDED); assertThat(actual.snapshotId()).isEqualTo(SNAPSHOT_ID); @@ -451,7 +458,9 @@ public void projectNullReadsFullSchema(FileFormat format) throws IOException { // project(null) clears the projection and reads the full schema, like no projection at all try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).project(null).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .project(null) + .build()) { TrackedFile actual = Iterables.getOnlyElement(reader); assertThat(actual.location()).isEqualTo("s3://bucket/file.parquet"); assertThat(actual.fileFormat()).isEqualTo(FileFormat.PARQUET); @@ -469,7 +478,7 @@ public void partitionFilterForceProjectsFilterFields(FileFormat format) throws I // filter reads (spec_id, partition) or every row would be pruned Schema projection = new Schema(TrackedFile.LOCATION); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS) .project(projection) .filter(Expressions.equal("id", 1)) .build()) { @@ -488,7 +497,7 @@ public void selectWithPartitionFilterProjectsFilterFields(FileFormat format) thr // the caller selects only location; the reader must still project spec_id and partition // for the partition filter or every row would be pruned try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS) .select("location") .filter(Expressions.equal("id", 1)) .build()) { @@ -530,7 +539,7 @@ public void partitionFilterPrunesFilesAndCountsSkips(FileFormat format) throws I ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS) .filter(Expressions.equal("id", 1)) .scanMetrics(metrics) .build()) { @@ -568,7 +577,7 @@ public void rowFilterKeepsFilesWithoutStats(FileFormat format) throws IOExceptio ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .filter(Expressions.equal("id", 1)) .scanMetrics(metrics) .build()) { @@ -588,7 +597,7 @@ public void caseInsensitivePartitionFilter(FileFormat format) throws IOException // a case-insensitive filter binds the mismatched-case "ID" reference and prunes FILE_B try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS) .filter(Expressions.equal("ID", 1)) .caseSensitive(false) .build()) { @@ -598,7 +607,7 @@ public void caseInsensitivePartitionFilter(FileFormat format) throws IOException // the same filter is case-sensitive by default, so "ID" fails to bind to the "id" field assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS) .filter(Expressions.equal("ID", 1)) .build()) .isInstanceOf(ValidationException.class) @@ -633,7 +642,9 @@ public void filterMatchesFilesAcrossDisjointSpecs(FileFormat format) throws IOEx writeManifest(format, unionType, ImmutableList.of(keepById, prunedById, keptOtherSpec)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, specsById).filter(Expressions.equal("id", 1)).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, specsById) + .filter(Expressions.equal("id", 1)) + .build()) { // spec0 entries are pruned by id; the spec1 entry is not partitioned by id so it survives assertThat(reader) .extracting(TrackedFile::location) @@ -652,7 +663,7 @@ public void partialFilterStillPrunesOnCompatibleField(FileFormat format) throws InputFile manifest = writeManifest(format, ID_PARTITION_TYPE, ImmutableList.of(keep, prune)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS) .filter(Expressions.and(Expressions.equal("id", 1), Expressions.equal("data", "z"))) .build()) { assertThat(reader) @@ -672,7 +683,7 @@ public void partitionFilterKeepsFileWithUnknownSpec(FileFormat format) throws IO // the filter would prune partition id=1 under spec 0, but cannot be applied to spec 5 try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS) .filter(Expressions.equal("id", 2)) .build()) { assertThat(reader).extracting(TrackedFile::location).containsExactly(file.location()); @@ -687,7 +698,7 @@ public void partitionFilterKeepsFileWithNullSpecId(FileFormat format) throws IOE InputFile manifest = writeManifest(format, ID_PARTITION_TYPE, ImmutableList.of(file)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS) .filter(Expressions.equal("id", 2)) .build()) { assertThat(reader).extracting(TrackedFile::location).containsExactly(file.location()); @@ -706,7 +717,7 @@ public void iteratorReturnsLiveCopies(FileFormat format) throws IOException { InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS).build()) { List read = Lists.newArrayList(reader); assertThat(read) .hasSize(2) @@ -723,7 +734,8 @@ public void unknownManifestFormatThrows() throws IOException { InputFile badFile = fileIO.newInputFile(tempDir.resolve("manifest-" + System.nanoTime() + ".txt").toString()); - try (V4ManifestReader reader = V4ManifestReader.builder(badFile, UNPARTITIONED_SPECS).build()) { + try (V4ManifestReader reader = + V4ManifestReader.builder(badFile, TABLE_SCHEMA, UNPARTITIONED_SPECS).build()) { assertThatThrownBy(reader::iterator) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("Cannot determine format of manifest"); @@ -734,21 +746,48 @@ public void unknownManifestFormatThrows() throws IOException { public void invalidBuilderArguments() { InputFile manifest = fileIO.newInputFile(tempDir.resolve("manifest.avro").toString()); - assertThatThrownBy(() -> V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).filter(null)) + assertThatThrownBy( + () -> + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS).filter(null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid filter: null"); assertThatThrownBy( - () -> V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS).scanMetrics(null)) + () -> + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .scanMetrics(null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid scan metrics: null"); assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) .select((Collection) null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid columns: null"); + + assertThatThrownBy(() -> V4ManifestReader.builder(manifest, null, UNPARTITIONED_SPECS)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid table schema: null"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + public void filterOnMissingColumnFails(FileFormat format) throws IOException { + InputFile manifest = + writeManifest( + format, + EMPTY_PARTITION, + ImmutableList.of(dataFile("data-a.parquet", EMPTY_PARTITION_DATA))); + + // stats for the filter's columns are resolved against the table schema when the reader is built + assertThatThrownBy( + () -> + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .filter(Expressions.equal("missing", 34)) + .build()) + .isInstanceOf(ValidationException.class) + .hasMessageContaining("Cannot find field 'missing' in struct: %s", TABLE_SCHEMA.asStruct()); } private static TrackedFile dataFile(String location, PartitionData partition) { @@ -880,7 +919,8 @@ private InputFile writeManifest( private List read(InputFile manifest, Map specsById) throws IOException { - try (V4ManifestReader reader = V4ManifestReader.builder(manifest, specsById).build()) { + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, specsById).build()) { return Lists.newArrayList(reader); } } diff --git a/core/src/test/java/org/apache/iceberg/TestV4ManifestReaderStats.java b/core/src/test/java/org/apache/iceberg/TestV4ManifestReaderStats.java new file mode 100644 index 000000000000..d7e1c327c017 --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReaderStats.java @@ -0,0 +1,790 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.inmemory.InMemoryInputFile; +import org.apache.iceberg.inmemory.InMemoryOutputFile; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +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.apache.iceberg.variants.ShreddedObject; +import org.apache.iceberg.variants.Variant; +import org.apache.iceberg.variants.VariantMetadata; +import org.apache.iceberg.variants.VariantTestUtil; +import org.apache.iceberg.variants.Variants; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.FieldSource; + +class TestV4ManifestReaderStats { + private static final Types.StructType EMPTY_PARTITION = Types.StructType.of(); + private static final PartitionData EMPTY_PARTITION_DATA = new PartitionData(EMPTY_PARTITION); + private static final Map UNPARTITIONED_SPECS = + ImmutableMap.of(PartitionSpec.unpartitioned().specId(), PartitionSpec.unpartitioned()); + private static final List MANIFEST_FORMATS = + List.of(FileFormat.AVRO, FileFormat.PARQUET); + + private static final int ID_FIELD_ID = 1; + private static final int DATA_FIELD_ID = 2; + private static final int MEASURE_FIELD_ID = 3; + + private static final Schema TABLE_SCHEMA = + new Schema( + optional(ID_FIELD_ID, "id", Types.IntegerType.get()), + optional(DATA_FIELD_ID, "data", Types.StringType.get()), + optional(MEASURE_FIELD_ID, "measure", Types.DoubleType.get())); + private static final Types.StructType CONTENT_STATS_TYPE = + StatsUtil.statsReadSchema( + TABLE_SCHEMA, List.of(ID_FIELD_ID, DATA_FIELD_ID, MEASURE_FIELD_ID)); + private static final FieldStats ID_STATS = + new FieldStatsStruct<>( + CONTENT_STATS_TYPE.fieldType("id").asStructType(), 1, 100, true, 26L, 2L, 0L, null); + private static final FieldStats DATA_STATS = + new FieldStatsStruct<>( + CONTENT_STATS_TYPE.fieldType("data").asStructType(), "a", "z", true, 26L, 0L, 0L, 4); + private static final FieldStats MEASURE_STATS = + new FieldStatsStruct<>( + CONTENT_STATS_TYPE.fieldType("measure").asStructType(), + 1.5, + 9.5, + false, + 26L, + 1L, + 3L, + null); + + @Test + void invalidProjectStatsArguments() { + InputFile manifest = new InMemoryInputFile(new byte[0]); + + assertThatThrownBy( + () -> + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .projectStats((Iterable) null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid stats projection for field IDs: null"); + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void readContentStatsForAllFieldIds(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS).build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertThat(stats).isNotNull(); + assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS); + assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS); + assertFieldStats(stats.statsFor(MEASURE_FIELD_ID), MEASURE_STATS); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void rowFilterKeepsStatsForAllFieldIds(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // a filter narrows the stats only for scan planning, so a default read still carries the + // stats of every field even though the filter needs one of them + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .filter(Expressions.equal("id", 1)) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS); + assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS); + assertFieldStats(stats.statsFor(MEASURE_FIELD_ID), MEASURE_STATS); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void selectStatsByName(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // stats are named after the column they describe, so a caller can select one column's stats + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .select("location", "content_stats.data") + .build()) { + TrackedFile actual = Iterables.getOnlyElement(reader); + assertThat(actual.location()).isEqualTo("s3://bucket/file.parquet"); + + ContentStats stats = actual.contentStats(); + assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS); + assertThat(stats.statsFor(ID_FIELD_ID)).isNull(); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void projectStatsReadsOnlyRequestedColumns(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .projectStats(ID_FIELD_ID) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS); + assertThat(stats.statsFor(DATA_FIELD_ID)).isNull(); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void projectStatsWithoutFieldIdsOmitsStats(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // requesting no field IDs opts out of the default projection of every field + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .projectStats(List.of()) + .build()) { + assertThat(Iterables.getOnlyElement(reader).contentStats()).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void projectStatsCopiesFieldIds(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + List fieldIds = Lists.newArrayList(ID_FIELD_ID); + V4ManifestReader.Builder builder = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .projectStats(fieldIds); + + // the builder copies the field IDs, so a later change to fieldIds does not widen the projection + fieldIds.add(DATA_FIELD_ID); + + try (V4ManifestReader reader = builder.build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS); + assertThat(stats.statsFor(DATA_FIELD_ID)).isNull(); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void requestedStatsAreProjectedWhenOmittedByCaller(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // stats requested by field ID are read even though the projection omits them + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .project(new Schema(TrackedFile.LOCATION)) + .projectStats(MEASURE_FIELD_ID) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(MEASURE_FIELD_ID), MEASURE_STATS); + assertThat(stats.statsFor(ID_FIELD_ID)).isNull(); + assertThat(stats.statsFor(DATA_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterStatsAreProjectedWhenOmittedByCaller(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // the filter references data, so its stats are read even though the projection omits them + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .project(new Schema(TrackedFile.LOCATION)) + .filter(Expressions.equal("data", "m")) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS); + assertThat(stats.statsFor(ID_FIELD_ID)).isNull(); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterStatsAreProjectedForCaseInsensitiveFilter(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // the filter refers to data by a different case, which binds only when case is ignored + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .project(new Schema(TrackedFile.LOCATION)) + .caseSensitive(false) + .filter(Expressions.equal("DATA", "m")) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS); + assertThat(stats.statsFor(ID_FIELD_ID)).isNull(); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void projectStatsAndFilterStatsAreCombined(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // scan planning narrows stats to the requested fields and the fields the filter needs + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .forScanPlanning() + .projectStats(ID_FIELD_ID) + .filter(Expressions.equal("data", "m")) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS); + assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void forScanPlanningReadsOnlyFilterStats(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .forScanPlanning() + .filter(Expressions.equal("id", 1)) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS); + assertThat(stats.statsFor(DATA_FIELD_ID)).isNull(); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void forScanPlanningOmitsStatsWithoutFilter(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // scan planning without a filter has no stats to evaluate, so none are read + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .forScanPlanning() + .build()) { + assertThat(Iterables.getOnlyElement(reader).contentStats()).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void selectWithoutStatsOmitsStats(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS) + .select("location") + .build()) { + assertThat(Iterables.getOnlyElement(reader).contentStats()).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void statsAreNullForColumnsWithoutStoredStats(FileFormat format) throws IOException { + // the manifest stores stats for id alone, while the reader reads stats for every column + Types.StructType storedStatsType = + StatsUtil.statsReadSchema(TABLE_SCHEMA, List.of(ID_FIELD_ID)); + ContentStatsStruct stored = new ContentStatsStruct(storedStatsType); + stored.setStats(ID_FIELD_ID, ID_STATS); + + TrackedFile file = fileWithStats("s3://bucket/file.parquet", stored); + InputFile manifest = writeManifest(format, storedStatsType, List.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS).build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS); + assertThat(stats.statsFor(DATA_FIELD_ID)).isNull(); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + assertThat(stats.fieldStats()).doesNotContainNull().hasSize(1); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void statsAreCorrectWithContainerReuse(FileFormat format) throws IOException { + // every entry stores stats for a different column, so a reused container must not carry the + // previous entry's stats into the next one + ContentStatsStruct idStats = new ContentStatsStruct(CONTENT_STATS_TYPE); + idStats.setStats(ID_FIELD_ID, ID_STATS); + + ContentStatsStruct dataStats = new ContentStatsStruct(CONTENT_STATS_TYPE); + dataStats.setStats(DATA_FIELD_ID, DATA_STATS); + + ContentStatsStruct measureStats = new ContentStatsStruct(CONTENT_STATS_TYPE); + measureStats.setStats(MEASURE_FIELD_ID, MEASURE_STATS); + + List files = + List.of( + fileWithStats("s3://bucket/with-id-stats.parquet", idStats), + fileWithStats( + "s3://bucket/without-stats.parquet", new ContentStatsStruct(CONTENT_STATS_TYPE)), + fileWithStats("s3://bucket/with-data-stats.parquet", dataStats), + fileWithStats("s3://bucket/with-measure-stats.parquet", measureStats)); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, files); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS).build()) { + List read = Lists.newArrayList(reader); + + ContentStats withIdStats = read.get(0).contentStats(); + assertFieldStats(withIdStats.statsFor(ID_FIELD_ID), ID_STATS); + assertThat(withIdStats.statsFor(DATA_FIELD_ID)).isNull(); + assertThat(withIdStats.statsFor(MEASURE_FIELD_ID)).isNull(); + assertThat(withIdStats.fieldStats()).doesNotContainNull(); + + // containers are reused, so the second entry must not carry over the first entry's stats + ContentStats withoutStats = read.get(1).contentStats(); + assertThat(withoutStats.statsFor(ID_FIELD_ID)).isNull(); + assertThat(withoutStats.fieldStats()).isEmpty(); + + ContentStats withDataStats = read.get(2).contentStats(); + assertFieldStats(withDataStats.statsFor(DATA_FIELD_ID), DATA_STATS); + assertThat(withDataStats.statsFor(ID_FIELD_ID)).isNull(); + assertThat(withDataStats.statsFor(MEASURE_FIELD_ID)).isNull(); + assertThat(withDataStats.fieldStats()).doesNotContainNull(); + + ContentStats withMeasureStats = read.get(3).contentStats(); + assertFieldStats(withMeasureStats.statsFor(MEASURE_FIELD_ID), MEASURE_STATS); + assertThat(withMeasureStats.statsFor(ID_FIELD_ID)).isNull(); + assertThat(withMeasureStats.statsFor(DATA_FIELD_ID)).isNull(); + assertThat(withMeasureStats.fieldStats()).doesNotContainNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void readStatsForNestedFields(FileFormat format) throws IOException { + int locationFieldId = 20; + int latFieldId = 21; + int lonFieldId = 22; + int tagsFieldId = 23; + int tagFieldId = 24; + + Schema nestedSchema = + new Schema( + optional( + locationFieldId, + "location", + Types.StructType.of( + required(latFieldId, "lat", Types.DoubleType.get()), + optional(lonFieldId, "lon", Types.DoubleType.get()))), + optional( + tagsFieldId, + "tags", + Types.ListType.ofOptional(tagFieldId, Types.StringType.get()))); + Types.StructType statsType = + StatsUtil.statsReadSchema(nestedSchema, List.of(latFieldId, lonFieldId)); + FieldStats latStats = + new FieldStatsStruct<>( + statsType.fieldType("location_lat").asStructType(), 1.5, 9.5, true, 26L, 0L, 0L, null); + FieldStats lonStats = + new FieldStatsStruct<>( + statsType.fieldType("location_lon").asStructType(), + -9.5, + -1.5, + true, + 26L, + 2L, + 0L, + null); + + ContentStatsStruct stats = new ContentStatsStruct(statsType); + stats.setStats(latFieldId, latStats); + stats.setStats(lonFieldId, lonStats); + + TrackedFile file = fileWithStats("s3://bucket/file.parquet", stats); + InputFile manifest = writeManifest(format, statsType, List.of(file)); + + // the reader requests stats for every field of the table, including the struct and the list + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, nestedSchema, UNPARTITIONED_SPECS).build()) { + ContentStats actual = Iterables.getOnlyElement(reader).contentStats(); + assertThat(actual.type()).isEqualTo(statsType); + assertFieldStats(actual.statsFor(latFieldId), latStats); + assertFieldStats(actual.statsFor(lonFieldId), lonStats); + assertThat(actual.statsFor(locationFieldId)).isNull(); + assertThat(actual.statsFor(tagsFieldId)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void readVariantStats(FileFormat format) throws IOException { + int fieldId = 12; + Schema schema = new Schema(optional(fieldId, "var", Types.VariantType.get())); + Types.StructType statsType = StatsUtil.statsReadSchema(schema, List.of(fieldId)); + Types.StructType varStatsType = statsType.fieldType("var").asStructType(); + + VariantMetadata metadata = Variants.metadata("$['x']"); + Variant lowerBound = variantBound(metadata, 1); + Variant upperBound = variantBound(metadata, 10); + + ContentStatsStruct stats = new ContentStatsStruct(statsType); + stats.setStats( + fieldId, + new FieldStatsStruct<>(varStatsType, lowerBound, upperBound, false, 26L, 2L, 0L, 32)); + + TrackedFile file = fileWithStats("s3://bucket/file.parquet", stats); + InputFile manifest = writeManifest(format, statsType, List.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, schema, UNPARTITIONED_SPECS).build()) { + FieldStats actual = Iterables.getOnlyElement(reader).contentStats().statsFor(fieldId); + + assertThat(actual.type()).isEqualTo(varStatsType); + assertThat(actual.valueCount()).isEqualTo(26L); + assertThat(actual.nullValueCount()).isEqualTo(2L); + assertThat(actual.avgValueSizeInBytes()).isEqualTo(32); + assertThat(actual.tightBounds()).isFalse(); + assertThat(actual.hasNanValueCount()).isFalse(); + assertVariant(actual.lowerBound(), lowerBound); + assertVariant(actual.upperBound(), upperBound); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void variantStatsAreCorrectWithContainerReuse(FileFormat format) throws IOException { + int fieldId = 12; + Schema schema = new Schema(optional(fieldId, "var", Types.VariantType.get())); + Types.StructType statsType = StatsUtil.statsReadSchema(schema, List.of(fieldId)); + Types.StructType varStatsType = statsType.fieldType("var").asStructType(); + + VariantMetadata metadata = Variants.metadata("$['x']"); + Variant firstLower = variantBound(metadata, 1); + Variant firstUpper = variantBound(metadata, 10); + ContentStatsStruct first = new ContentStatsStruct(statsType); + first.setStats( + fieldId, + new FieldStatsStruct<>(varStatsType, firstLower, firstUpper, false, 26L, 2L, 0L, 32)); + + Variant secondLower = variantBound(metadata, 100); + Variant secondUpper = variantBound(metadata, 1000); + ContentStatsStruct second = new ContentStatsStruct(statsType); + second.setStats( + fieldId, + new FieldStatsStruct<>(varStatsType, secondLower, secondUpper, false, 45L, 1L, 0L, 64)); + + List files = + List.of( + fileWithStats("s3://bucket/first.parquet", first), + fileWithStats("s3://bucket/second.parquet", second)); + InputFile manifest = writeManifest(format, statsType, files); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, schema, UNPARTITIONED_SPECS).build()) { + List read = Lists.newArrayList(reader); + + FieldStats firstStats = read.get(0).contentStats().statsFor(fieldId); + assertThat(firstStats.type()).isEqualTo(varStatsType); + assertThat(firstStats.valueCount()).isEqualTo(26L); + assertThat(firstStats.nullValueCount()).isEqualTo(2L); + assertThat(firstStats.avgValueSizeInBytes()).isEqualTo(32); + assertThat(firstStats.tightBounds()).isFalse(); + assertThat(firstStats.hasNanValueCount()).isFalse(); + assertVariant(firstStats.lowerBound(), firstLower); + assertVariant(firstStats.upperBound(), firstUpper); + + FieldStats secondStats = read.get(1).contentStats().statsFor(fieldId); + assertThat(secondStats.type()).isEqualTo(varStatsType); + assertThat(secondStats.valueCount()).isEqualTo(45L); + assertThat(secondStats.nullValueCount()).isEqualTo(1L); + assertThat(secondStats.avgValueSizeInBytes()).isEqualTo(64); + assertThat(secondStats.tightBounds()).isFalse(); + assertThat(secondStats.hasNanValueCount()).isFalse(); + assertVariant(secondStats.lowerBound(), secondLower); + assertVariant(secondStats.upperBound(), secondUpper); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void readGeoStats(FileFormat format) throws IOException { + int geometryFieldId = 10; + int geographyFieldId = 11; + Schema geoSchema = + new Schema( + optional(geometryFieldId, "geom", Types.GeometryType.crs84()), + optional(geographyFieldId, "geog", Types.GeographyType.crs84())); + Types.StructType statsType = + StatsUtil.statsReadSchema(geoSchema, List.of(geometryFieldId, geographyFieldId)); + Types.StructType geomStatsType = statsType.fieldType("geom").asStructType(); + Types.StructType geogStatsType = statsType.fieldType("geog").asStructType(); + + ContentStatsStruct stats = new ContentStatsStruct(statsType); + stats.setStats( + geometryFieldId, + new FieldStatsStruct<>( + geomStatsType, + boundingBox(geomStatsType, StatsUtil.LOWER_BOUND_NAME, 1.0d, 2.0d, 3.0d, 4.0d), + boundingBox(geomStatsType, StatsUtil.UPPER_BOUND_NAME, 5.0d, 6.0d, 7.0d, 8.0d), + false, + 26L, + 2L, + 0L, + 32)); + stats.setStats( + geographyFieldId, + new FieldStatsStruct<>( + geogStatsType, + boundingBox(geogStatsType, StatsUtil.LOWER_BOUND_NAME, -20.0d, -10.0d, 0.0d, 1.0d), + boundingBox(geogStatsType, StatsUtil.UPPER_BOUND_NAME, 20.0d, 10.0d, 30.0d, 2.0d), + false, + 30L, + 3L, + 0L, + 48)); + + TrackedFile file = fileWithStats("s3://bucket/file.parquet", stats); + InputFile manifest = writeManifest(format, statsType, List.of(file)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, geoSchema, UNPARTITIONED_SPECS).build()) { + ContentStats actual = Iterables.getOnlyElement(reader).contentStats(); + + FieldStats geom = actual.statsFor(geometryFieldId); + assertThat(geom.type()).isEqualTo(geomStatsType); + assertThat(geom.valueCount()).isEqualTo(26L); + assertThat(geom.nullValueCount()).isEqualTo(2L); + assertThat(geom.hasNanValueCount()).isFalse(); + assertThat(geom.tightBounds()).isFalse(); + assertThat(geom.avgValueSizeInBytes()).isEqualTo(32); + assertBoundingBox(geom.lowerBound(), 1.0d, 2.0d, 3.0d, 4.0d); + assertBoundingBox(geom.upperBound(), 5.0d, 6.0d, 7.0d, 8.0d); + + FieldStats geog = actual.statsFor(geographyFieldId); + assertThat(geog.type()).isEqualTo(geogStatsType); + assertThat(geog.valueCount()).isEqualTo(30L); + assertThat(geog.nullValueCount()).isEqualTo(3L); + assertThat(geog.hasNanValueCount()).isFalse(); + assertThat(geog.tightBounds()).isFalse(); + assertThat(geog.avgValueSizeInBytes()).isEqualTo(48); + assertBoundingBox(geog.lowerBound(), -20.0d, -10.0d, 0.0d, 1.0d); + assertBoundingBox(geog.upperBound(), 20.0d, 10.0d, 30.0d, 2.0d); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void geoStatsAreCorrectWithContainerReuse(FileFormat format) throws IOException { + int geometryFieldId = 10; + Schema geoSchema = new Schema(optional(geometryFieldId, "geom", Types.GeometryType.crs84())); + Types.StructType statsType = StatsUtil.statsReadSchema(geoSchema, List.of(geometryFieldId)); + Types.StructType geomStatsType = statsType.fieldType("geom").asStructType(); + + ContentStatsStruct first = new ContentStatsStruct(statsType); + first.setStats( + geometryFieldId, + new FieldStatsStruct<>( + geomStatsType, + boundingBox(geomStatsType, StatsUtil.LOWER_BOUND_NAME, 1.0d, 2.0d, 3.0d, 4.0d), + boundingBox(geomStatsType, StatsUtil.UPPER_BOUND_NAME, 5.0d, 6.0d, 7.0d, 8.0d), + false, + 26L, + 2L, + 0L, + 32)); + + ContentStatsStruct second = new ContentStatsStruct(statsType); + second.setStats( + geometryFieldId, + new FieldStatsStruct<>( + geomStatsType, + boundingBox(geomStatsType, StatsUtil.LOWER_BOUND_NAME, 11.0d, 12.0d, 13.0d, 14.0d), + boundingBox(geomStatsType, StatsUtil.UPPER_BOUND_NAME, 15.0d, 16.0d, 17.0d, 18.0d), + false, + 45L, + 1L, + 0L, + 64)); + + List files = + List.of( + fileWithStats("s3://bucket/first.parquet", first), + fileWithStats("s3://bucket/second.parquet", second)); + InputFile manifest = writeManifest(format, statsType, files); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, geoSchema, UNPARTITIONED_SPECS).build()) { + List read = Lists.newArrayList(reader); + + FieldStats firstStats = read.get(0).contentStats().statsFor(geometryFieldId); + assertThat(firstStats.type()).isEqualTo(geomStatsType); + assertThat(firstStats.valueCount()).isEqualTo(26L); + assertThat(firstStats.nullValueCount()).isEqualTo(2L); + assertThat(firstStats.avgValueSizeInBytes()).isEqualTo(32); + assertThat(firstStats.tightBounds()).isFalse(); + assertThat(firstStats.hasNanValueCount()).isFalse(); + assertBoundingBox(firstStats.lowerBound(), 1.0d, 2.0d, 3.0d, 4.0d); + assertBoundingBox(firstStats.upperBound(), 5.0d, 6.0d, 7.0d, 8.0d); + + FieldStats secondStats = read.get(1).contentStats().statsFor(geometryFieldId); + assertThat(secondStats.type()).isEqualTo(geomStatsType); + assertThat(secondStats.valueCount()).isEqualTo(45L); + assertThat(secondStats.nullValueCount()).isEqualTo(1L); + assertThat(secondStats.avgValueSizeInBytes()).isEqualTo(64); + assertThat(secondStats.tightBounds()).isFalse(); + assertThat(secondStats.hasNanValueCount()).isFalse(); + assertBoundingBox(secondStats.lowerBound(), 11.0d, 12.0d, 13.0d, 14.0d); + assertBoundingBox(secondStats.upperBound(), 15.0d, 16.0d, 17.0d, 18.0d); + } + } + + private static void assertFieldStats(FieldStats actual, FieldStats expected) { + assertThat(actual).isNotNull(); + assertThat(actual.fieldId()).isEqualTo(expected.fieldId()); + assertThat(actual.type()).isEqualTo(expected.type()); + assertThat(actual.valueCount()).isEqualTo(expected.valueCount()); + assertThat(actual.lowerBound()).isEqualTo(expected.lowerBound()); + assertThat(actual.upperBound()).isEqualTo(expected.upperBound()); + assertThat(actual.tightBounds()).isEqualTo(expected.tightBounds()); + assertThat(actual.avgValueSizeInBytes()).isEqualTo(expected.avgValueSizeInBytes()); + + Types.StructType statsType = expected.type(); + if (statsType.field("null_value_count") != null) { + assertThat(actual.hasNullValueCount()).isTrue(); + assertThat(actual.nullValueCount()).isEqualTo(expected.nullValueCount()); + } else { + assertThat(actual.hasNullValueCount()).isFalse(); + } + + if (statsType.field("nan_value_count") != null) { + assertThat(actual.hasNanValueCount()).isTrue(); + assertThat(actual.nanValueCount()).isEqualTo(expected.nanValueCount()); + } else { + assertThat(actual.hasNanValueCount()).isFalse(); + } + } + + private static void assertBoundingBox(Object bound, double... ordinates) { + assertThat(bound).isInstanceOf(StructLike.class); + StructLike box = (StructLike) bound; + assertThat(box.size()).isEqualTo(ordinates.length); + for (int pos = 0; pos < ordinates.length; pos += 1) { + assertThat(box.get(pos, Double.class)) + .as("Bounding box ordinate at position %s", pos) + .isEqualTo(ordinates[pos]); + } + } + + /** Returns a variant bound holding the given value for the "$['x']" path. */ + private static Variant variantBound(VariantMetadata metadata, int value) { + ShreddedObject object = Variants.object(metadata); + object.put("$['x']", Variants.of(value)); + return Variant.of(metadata, object); + } + + private static void assertVariant(Object actual, Variant expected) { + assertThat(actual).isInstanceOf(Variant.class); + Variant variant = (Variant) actual; + VariantTestUtil.assertEqual(expected.metadata(), variant.metadata()); + VariantTestUtil.assertEqual(expected.value(), variant.value()); + } + + private static StructLike boundingBox( + Types.StructType statsType, String boundName, double... ordinates) { + PartitionData box = new PartitionData(statsType.fieldType(boundName).asStructType()); + for (int pos = 0; pos < ordinates.length; pos += 1) { + box.set(pos, ordinates[pos]); + } + + return box; + } + + /** Returns stats for every table column, backed by the full content stats type. */ + private static ContentStats contentStats() { + ContentStatsStruct stats = new ContentStatsStruct(CONTENT_STATS_TYPE); + stats.setStats(ID_FIELD_ID, ID_STATS); + stats.setStats(DATA_FIELD_ID, DATA_STATS); + stats.setStats(MEASURE_FIELD_ID, MEASURE_STATS); + return stats; + } + + private static TrackedFile fileWithStats(String location, ContentStats stats) { + return new TrackedFileStruct( + new TrackingStruct(EntryStatus.ADDED, 42L, null, null, null, null, null, null), + FileContent.DATA, + 4, + location, + FileFormat.PARQUET, + 100L, + 1024L, + 0, + EMPTY_PARTITION_DATA, + stats, + null, + null, + null, + null, + null, + null); + } + + private static InputFile writeManifest( + FileFormat format, Types.StructType contentStatsType, Iterable files) + throws IOException { + Schema writeSchema = TrackedFile.schema(EMPTY_PARTITION, contentStatsType); + OutputFile out = new InMemoryOutputFile("manifest." + format.name().toLowerCase(Locale.ROOT)); + try (FileAppender appender = + InternalData.write(format, out).schema(writeSchema).named("tracked_file").build()) { + for (TrackedFile file : files) { + appender.add((StructLike) file); + } + } + + return out.toInputFile(); + } +}