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 93de05a2e6ac..bb3aa1a556bb 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,16 @@ private static int[] posToOffset(Types.StructType struct) { return posToOffset; } + 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/V4ManifestReader.java b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java index dca4d67fa3e7..41393cf05281 100644 --- a/core/src/main/java/org/apache/iceberg/V4ManifestReader.java +++ b/core/src/main/java/org/apache/iceberg/V4ManifestReader.java @@ -22,6 +22,9 @@ import java.util.Collection; import java.util.Map; import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.Stream; +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 +35,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.LocationUtil; import org.apache.iceberg.util.Pair; import org.apache.iceberg.util.StructProjection; @@ -68,8 +74,11 @@ private V4ManifestReader( } static Builder builder( - InputFile file, Map specsById, String tableLocation) { - return new Builder(file, specsById, tableLocation); + InputFile file, + Schema tableSchema, + Map specsById, + String tableLocation) { + return new Builder(file, tableSchema, specsById, tableLocation); } /** Returns copies of the tracked files that match this reader's configured filters. */ @@ -127,7 +136,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) @@ -135,8 +144,18 @@ 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); + for (Types.NestedField fieldStats : statsField.type().asStructType().fields()) { + readBuilder.setCustomType(fieldStats.fieldId(), FieldStatsStruct.class); + } + } + + CloseableIterable reader = readBuilder.build(); addCloseable(reader); return reader; } @@ -174,9 +193,9 @@ 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 final String tableLocation; private Expression rowFilter = Expressions.alwaysTrue(); private boolean caseSensitive = true; @@ -184,19 +203,22 @@ static class Builder { private boolean scanPlanning = false; private Collection columns = null; private Schema requestedProjection = null; + private Set fieldIdsWithRequestedStats = null; + private MetricsConfig metricsConfig = null; private ScanMetrics scanMetrics = ScanMetrics.noop(); - private Builder(InputFile file, Map specsById, String tableLocation) { + private Builder( + InputFile file, + Schema tableSchema, + Map specsById, + String tableLocation) { + Preconditions.checkArgument(tableSchema != null, "Invalid table schema: null"); Preconditions.checkArgument(tableLocation != null, "Invalid table location: null"); this.file = file; + this.tableSchema = tableSchema; this.specsById = specsById; this.tableLocation = LocationUtil.stripTrailingSlash(tableLocation); 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. */ @@ -252,6 +274,39 @@ 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. + * + *

Passing no field IDs reads only the stats that the filter needs. + */ + Builder projectStats(int... fieldIds) { + Preconditions.checkArgument(fieldIds != null, "Invalid 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. + * + *

Passing an empty iterable reads only the stats that the filter needs. + */ + Builder projectStats(Iterable fieldIds) { + Preconditions.checkArgument(fieldIds != null, "Invalid field IDs: null"); + this.fieldIdsWithRequestedStats = ImmutableSet.copyOf(fieldIds); + return this; + } + + /** + * Sets the metrics config that determines which stats the manifest holds. Defaults to the + * config produced by the table's default metrics properties. + */ + Builder metricsConfig(MetricsConfig newMetricsConfig) { + Preconditions.checkArgument(newMetricsConfig != null, "Invalid metrics config: null"); + this.metricsConfig = newMetricsConfig; + return this; + } + Builder scanMetrics(ScanMetrics newScanMetrics) { Preconditions.checkArgument(newScanMetrics != null, "Invalid scan metrics: null"); this.scanMetrics = newScanMetrics; @@ -283,26 +338,128 @@ V4ManifestReader build() { } private Schema readSchema(boolean hasPartitionFilter) { + Types.StructType requiredStatsType = + StatsUtil.statsReadSchema( + tableSchema, + fieldIdsWithRequiredStats( + tableSchema, fieldIdsWithRequestedStats, rowFilter, caseSensitive)); + Schema fullSchema = fullSchema(contentStatsType(requiredStatsType)); if (scanPlanning) { // scan planning does not read the change-tracking fields omitted by SCAN_TYPE return TypeUtil.replaceFieldTypes( fullSchema, ImmutableMap.of(TrackedFile.TRACKING.fieldId(), TrackingStruct.SCAN_TYPE)); } + Types.StructType partitionType = hasPartitionFilter ? unionPartitionType : null; if (columns != null) { Schema selected = caseSensitive ? fullSchema.select(columns) : fullSchema.caseInsensitiveSelect(columns); - return addRequiredColumns(selected, hasPartitionFilter); + return addRequiredColumns( + fullSchema, selected, requiredStatsType, partitionType, rowFilter); } if (requestedProjection != null) { - return addRequiredColumns(requestedProjection, hasPartitionFilter); + return addRequiredColumns( + fullSchema, requestedProjection, requiredStatsType, partitionType, rowFilter); } return fullSchema; } - private Schema addRequiredColumns(Schema projection, boolean hasPartitionFilter) { + /** Returns the schema of everything this reader may read, including content stats. */ + private Schema fullSchema(Types.StructType contentStatsType) { + Schema base = TrackedFile.schema(unionPartitionType, contentStatsType); + if (contentStatsType.fields().isEmpty()) { + // the schema uses the unknown type for empty stats, but readers fail to pair the stats + // struct stored in the manifest with unknown, so drop the field instead of projecting it + 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. + * + *

Stats for every field the manifest holds are read unless the caller narrows them with + * {@link #forScanPlanning()}, {@link #projectStats(Iterable)}, or a {@link #project(Schema) + * projection} that carries its own stats, because copying entries into a new manifest needs all + * of them. A {@link #filter(Expression) filter} therefore never narrows the stats that are + * read; it only widens a set the caller has already narrowed. + */ + private Types.StructType contentStatsType(Types.StructType requiredStatsType) { + if (scanPlanning || fieldIdsWithRequestedStats != null) { + return requiredStatsType; + } + + if (requestedProjection != null) { + return union(requiredStatsType, projectedStatsType(requestedProjection)); + } + + return StatsUtil.statsWriteSchema(tableSchema, metricsConfig()); + } + + private MetricsConfig metricsConfig() { + if (metricsConfig == null) { + this.metricsConfig = MetricsConfig.from(ImmutableMap.of(), tableSchema, null); + } + + return metricsConfig; + } + + /** Returns the requested field IDs, plus the field IDs referenced by the row filter. */ + private static Set fieldIdsWithRequiredStats( + Schema tableSchema, + Set fieldIdsWithRequestedStats, + Expression rowFilter, + boolean caseSensitive) { + Set fieldIds = Sets.newHashSet(); + if (fieldIdsWithRequestedStats != null) { + fieldIds.addAll(fieldIdsWithRequestedStats); + } + + if (rowFilter != Expressions.alwaysTrue()) { + fieldIds.addAll( + Binder.boundReferences( + tableSchema.asStruct(), ImmutableList.of(rowFilter), caseSensitive)); + } + + return fieldIds; + } + + /** Returns the fields of both stats structs */ + private static Types.StructType union( + Types.StructType statsType, Types.StructType otherStatsType) { + if (otherStatsType.fields().isEmpty()) { + return statsType; + } + + return Types.StructType.of( + Stream.concat( + statsType.fields().stream(), + otherStatsType.fields().stream() + .filter(field -> statsType.field(field.fieldId()) == null)) + .collect(Collectors.toList())); + } + + /** Returns the content stats struct carried by the projection, which may be empty. */ + private static Types.StructType projectedStatsType(Schema projection) { + Types.NestedField statsField = projection.findField(TrackedFile.CONTENT_STATS_ID); + if (statsField == null || !statsField.type().isStructType()) { + return Types.StructType.of(); + } + + return statsField.type().asStructType(); + } + + private static Schema addRequiredColumns( + Schema fullSchema, + Schema projection, + Types.StructType requiredStatsType, + Types.StructType partitionType, + Expression rowFilter) { Set projectedIds = Sets.newHashSet(TypeUtil.getProjectedIds(projection)); // fields the reader consumes internally: status for liveness filtering, row_position for @@ -315,13 +472,14 @@ private Schema addRequiredColumns(Schema projection, boolean hasPartitionFilter) projectedIds.add(TrackedFile.RECORD_COUNT.fieldId()); } - // add the partition tuple only when it is needed to evaluate a partition filter - if (hasPartitionFilter) { + if (partitionType != null) { projectedIds.add(TrackedFile.SPEC_ID.fieldId()); projectedIds.add(TrackedFile.PARTITION_ID); - projectedIds.addAll(TypeUtil.getProjectedIds(unionPartitionType)); + projectedIds.addAll(TypeUtil.getProjectedIds(partitionType)); } + 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 847d05ba41e2..8603240ca033 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 file.tracking().status()) .containsExactly(EntryStatus.ADDED, EntryStatus.EXISTING, EntryStatus.MODIFIED); } try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .includeAll() .build()) { assertThat(reader) @@ -258,7 +259,7 @@ public void selectiveReadReturnsOnlyRequestedFields( InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); V4ManifestReader.Builder builder = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION); + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION); configureRead.accept(builder); try (V4ManifestReader reader = builder.build()) { TrackedFile actual = Iterables.getOnlyElement(reader); @@ -316,7 +317,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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .project(projection) .filter(Expressions.equal("id", 1)) .build()) { @@ -332,7 +333,8 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .select("location") .project(new Schema(TrackedFile.LOCATION))) .isInstanceOf(IllegalStateException.class) @@ -340,7 +342,8 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .project(new Schema(TrackedFile.LOCATION)) .select("location")) .isInstanceOf(IllegalStateException.class) @@ -348,7 +351,8 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .forScanPlanning() .select("location")) .isInstanceOf(IllegalStateException.class) @@ -356,7 +360,8 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .select("location") .forScanPlanning()) .isInstanceOf(IllegalStateException.class) @@ -365,7 +370,8 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .forScanPlanning() .project(new Schema(TrackedFile.LOCATION))) .isInstanceOf(IllegalStateException.class) @@ -373,7 +379,8 @@ public void projectionModesAreMutuallyExclusive() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .project(new Schema(TrackedFile.LOCATION)) .forScanPlanning()) .isInstanceOf(IllegalStateException.class) @@ -388,7 +395,7 @@ public void projectionPreservesNarrowTrackingProjection(FileFormat format) throw writeManifest(format, EMPTY_PARTITION, ImmutableList.of(FILE_WITH_FULL_TRACKING)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .select("tracking.status") .build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); @@ -408,7 +415,7 @@ 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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .forScanPlanning() .build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); @@ -434,7 +441,8 @@ 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, TABLE_LOCATION).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .build()) { Tracking actual = Iterables.getOnlyElement(reader).tracking(); assertThat(actual.status()).isEqualTo(EntryStatus.ADDED); assertThat(actual.snapshotId()).isEqualTo(SNAPSHOT_ID); @@ -455,7 +463,7 @@ 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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .project(null) .build()) { TrackedFile actual = Iterables.getOnlyElement(reader); @@ -475,7 +483,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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS, TABLE_LOCATION) .project(projection) .filter(Expressions.equal("id", 1)) .build()) { @@ -494,7 +502,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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS, TABLE_LOCATION) .select("location") .filter(Expressions.equal("id", 1)) .build()) { @@ -536,7 +544,7 @@ public void partitionFilterPrunesFilesAndCountsSkips(FileFormat format) throws I ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, ID_PARTITIONING_SPECS, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("id", 1)) .scanMetrics(metrics) .build()) { @@ -574,7 +582,7 @@ public void rowFilterKeepsFilesWithoutStats(FileFormat format) throws IOExceptio ScanMetrics metrics = ScanMetrics.of(new DefaultMetricsContext()); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .filter(Expressions.equal("id", 1)) .scanMetrics(metrics) .build()) { @@ -594,7 +602,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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("ID", 1)) .caseSensitive(false) .build()) { @@ -604,7 +612,8 @@ 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, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("ID", 1)) .build()) .isInstanceOf(ValidationException.class) @@ -639,7 +648,7 @@ public void filterMatchesFilesAcrossDisjointSpecs(FileFormat format) throws IOEx writeManifest(format, unionType, ImmutableList.of(keepById, prunedById, keptOtherSpec)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, specsById, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, specsById, TABLE_LOCATION) .filter(Expressions.equal("id", 1)) .build()) { // spec0 entries are pruned by id; the spec1 entry is not partitioned by id so it survives @@ -660,7 +669,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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.and(Expressions.equal("id", 1), Expressions.equal("data", "z"))) .build()) { assertThat(reader) @@ -680,7 +689,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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("id", 2)) .build()) { assertThat(reader).extracting(TrackedFile::location).containsExactly(resolved(file)); @@ -695,7 +704,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, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, ID_PARTITIONING_SPECS, TABLE_LOCATION) .filter(Expressions.equal("id", 2)) .build()) { assertThat(reader).extracting(TrackedFile::location).containsExactly(resolved(file)); @@ -714,7 +723,8 @@ public void iteratorReturnsLiveCopies(FileFormat format) throws IOException { InputFile manifest = writeManifest(format, EMPTY_PARTITION, files); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .build()) { List read = Lists.newArrayList(reader); assertThat(read) .hasSize(2) @@ -732,7 +742,8 @@ public void unknownManifestFormatThrows() throws IOException { fileIO.newInputFile(tempDir.resolve("manifest-" + System.nanoTime() + ".txt").toString()); try (V4ManifestReader reader = - V4ManifestReader.builder(badFile, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { + V4ManifestReader.builder(badFile, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .build()) { assertThatThrownBy(reader::iterator) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("Cannot determine format of manifest"); @@ -755,7 +766,8 @@ public void resolvesRelativeDeletionVectorLocation(FileFormat format) throws IOE InputFile manifest = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(file)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .build()) { TrackedFile actual = Iterables.getOnlyElement(reader); assertThat(actual.location()).isEqualTo(resolved(file)); assertThat(actual.deletionVector().location()) @@ -787,7 +799,8 @@ public void mixedAbsoluteAndRelativePathWithDataFileAndDV(FileFormat format) thr ImmutableList.of(absoluteFileRelativeDv, relativeFileAbsoluteDv)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .build()) { List actual = Lists.newArrayList(reader); // absolute locations pass through unchanged; relative ones resolve against the table location assertThat(actual.get(0).location()).isEqualTo("s3://other/abs.parquet"); @@ -808,7 +821,7 @@ public void resolutionSkippedWhenLocationNotProjected(FileFormat format) throws // location is not projected, so there is nothing to resolve even though it is relative try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .select("tracking.status") .build()) { TrackedFile actual = Iterables.getOnlyElement(reader); @@ -822,28 +835,37 @@ public void invalidBuilderArguments() { assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .filter(null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid filter: null"); assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .scanMetrics(null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid scan metrics: null"); assertThatThrownBy( () -> - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION) + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) .select((Collection) null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid columns: null"); - assertThatThrownBy(() -> V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, null)) + assertThatThrownBy( + () -> V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, null)) .isInstanceOf(IllegalArgumentException.class) .hasMessage("Invalid table location: null"); + + assertThatThrownBy( + () -> V4ManifestReader.builder(manifest, null, UNPARTITIONED_SPECS, TABLE_LOCATION)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid table schema: null"); } // the location a relative fixture resolves to once read against TABLE_LOCATION @@ -857,7 +879,8 @@ private void verifyLocationAfterWriteReadRoundTrip( InputFile manifest = writeManifest(format, EMPTY_PARTITION, ImmutableList.of(file)); try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, UNPARTITIONED_SPECS, TABLE_LOCATION).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .build()) { TrackedFile actual = Iterables.getOnlyElement(reader); assertThat(actual.location()).isEqualTo(expectedLocation); } @@ -872,6 +895,26 @@ private static DeletionVector dv(String location) { .build(); } + @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, TABLE_LOCATION) + .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) { return dataFile(location, 0, partition); } @@ -1022,7 +1065,7 @@ private InputFile writeManifest( private List read(InputFile manifest, Map specsById) throws IOException { try (V4ManifestReader reader = - V4ManifestReader.builder(manifest, specsById, TABLE_LOCATION).build()) { + V4ManifestReader.builder(manifest, TABLE_SCHEMA, specsById, TABLE_LOCATION).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..5dda9c71e0db --- /dev/null +++ b/core/src/test/java/org/apache/iceberg/TestV4ManifestReaderStats.java @@ -0,0 +1,1033 @@ +/* + * 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 String TABLE_LOCATION = "s3://bucket/db/table"; + + 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.statsWriteSchema( + TABLE_SCHEMA, + MetricsConfig.from( + ImmutableMap.of( + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "id", + "full", + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "data", + "full", + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "measure", + "full"), + TABLE_SCHEMA, + null)); + 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, TABLE_LOCATION) + .projectStats((Iterable) null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid field IDs: null"); + + assertThatThrownBy( + () -> + V4ManifestReader.builder( + manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .projectStats((int[]) null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Invalid 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, TABLE_LOCATION) + .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 statsAreReadWithMetricsConfig(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + MetricsConfig metricsConfig = + MetricsConfig.from( + ImmutableMap.of( + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "data", + "counts", + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "measure", + "none"), + TABLE_SCHEMA, + null); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .metricsConfig(metricsConfig) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(ID_FIELD_ID), ID_STATS); + + FieldStats dataStats = stats.statsFor(DATA_FIELD_ID); + assertThat(dataStats).isNotNull(); + assertThat(dataStats.lowerBound()).isNull(); + assertThat(dataStats.upperBound()).isNull(); + assertThat(dataStats.valueCount()).isEqualTo(DATA_STATS.valueCount()); + assertThat(dataStats.nullValueCount()).isEqualTo(DATA_STATS.nullValueCount()); + assertThat(dataStats.avgValueSizeInBytes()).isEqualTo(DATA_STATS.avgValueSizeInBytes()); + + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void schemaProjectionStatsAreReadWhenOmittedByMetricsConfig(FileFormat format) + throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // the projection is based on the manifest schema, so it determines the stats that are read even + // though the metrics config would not produce stats for measure + Schema projection = + new Schema(TrackedFile.LOCATION, StatsUtil.contentStatsField(CONTENT_STATS_TYPE)); + MetricsConfig metricsConfig = + MetricsConfig.from( + ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "measure", "none"), + TABLE_SCHEMA, + null); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .project(projection) + .metricsConfig(metricsConfig) + .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 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, TABLE_LOCATION) + .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)); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .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, TABLE_LOCATION) + .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, TABLE_LOCATION) + .projectStats(List.of()) + .build()) { + assertThat(Iterables.getOnlyElement(reader).contentStats()).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void projectStatsWithoutFieldIdsStillReadsFilterStats(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 every field's stats, but not out of what the filter needs + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .projectStats(List.of()) + .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 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, TABLE_LOCATION) + .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 requestedStatsAreProjectedWhenOmittedBySchemaProjection(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, TABLE_LOCATION) + .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 filterStatsAreProjectedWhenOmittedBySchemaProjection(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // filter references data, so its stats are read even though the schema projection omits them + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .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 singleStatsFieldIsReadForSchemaProjection(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // a projection may narrow stats to a single field rather than to whole stats structs + Types.NestedField dataStats = CONTENT_STATS_TYPE.field("data"); + Types.StructType lowerBoundOnly = + Types.StructType.of( + optional( + dataStats.fieldId(), + dataStats.name(), + Types.StructType.of( + dataStats.type().asStructType().field(StatsUtil.LOWER_BOUND_NAME)))); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .project(new Schema(TrackedFile.LOCATION, StatsUtil.contentStatsField(lowerBoundOnly))) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + FieldStats actual = stats.statsFor(DATA_FIELD_ID); + assertThat(actual).isNotNull(); + assertThat(actual.type()).isEqualTo(lowerBoundOnly.fieldType("data").asStructType()); + assertThat(actual.lowerBound()).isEqualTo(DATA_STATS.lowerBound()); + assertThat(actual.upperBound()).isNull(); + assertThat(actual.hasValueCount()).isFalse(); + assertThat(stats.statsFor(ID_FIELD_ID)).isNull(); + assertThat(stats.statsFor(MEASURE_FIELD_ID)).isNull(); + } + } + + @ParameterizedTest + @FieldSource("MANIFEST_FORMATS") + void filterStatsAreProjectedWhenNarrowedBySchemaProjection(FileFormat format) throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // the projection carries stats for data, but without the bounds that the filter is evaluated + // against, so the stats that are read for data must be widened back to the full stats struct + Types.StructType narrowStatsType = + StatsUtil.statsWriteSchema( + TABLE_SCHEMA, + MetricsConfig.from( + ImmutableMap.of( + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "id", + "none", + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "data", + "counts", + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "measure", + "none"), + TABLE_SCHEMA, + null)); + assertThat(narrowStatsType.fieldType("data").asStructType().field(StatsUtil.LOWER_BOUND_NAME)) + .isNull(); + + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .project(new Schema(TrackedFile.LOCATION, StatsUtil.contentStatsField(narrowStatsType))) + .filter(Expressions.equal("data", "m")) + .build()) { + ContentStats stats = Iterables.getOnlyElement(reader).contentStats(); + assertFieldStats(stats.statsFor(DATA_FIELD_ID), DATA_STATS); + + // stats omitted by the projection are not read because the filter does not need them + 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, TABLE_LOCATION) + .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, TABLE_LOCATION) + .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 projectStatsAndFilterStatsAreCombinedWithoutScanPlanning(FileFormat format) + throws IOException { + TrackedFile file = fileWithStats("s3://bucket/file.parquet", contentStats()); + InputFile manifest = writeManifest(format, CONTENT_STATS_TYPE, List.of(file)); + + // projectStats narrows stats to the requested fields and the fields the filter needs, which + // does not depend on scan planning + try (V4ManifestReader reader = + V4ManifestReader.builder(manifest, TABLE_SCHEMA, UNPARTITIONED_SPECS, TABLE_LOCATION) + .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, TABLE_LOCATION) + .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, TABLE_LOCATION) + .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, TABLE_LOCATION) + .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.statsWriteSchema( + TABLE_SCHEMA, + MetricsConfig.from( + ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "id", "full"), + TABLE_SCHEMA, + null)); + 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, TABLE_LOCATION) + .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 { + // the reader decodes entries into a reused container and copies each one before returning it, + // so every copy must hold the stats of its own entry rather than alias the container that the + // next entry overwrites; giving every entry stats for a different column makes a carried over + // value visible + 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, TABLE_LOCATION) + .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(); + + 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.statsWriteSchema( + nestedSchema, + MetricsConfig.from( + ImmutableMap.of( + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "location.lat", + "full", + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "location.lon", + "full"), + nestedSchema, + null)); + 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, TABLE_LOCATION) + .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.statsWriteSchema( + schema, + MetricsConfig.from( + ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "var", "full"), + schema, + null)); + 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, TABLE_LOCATION).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.statsWriteSchema( + schema, + MetricsConfig.from( + ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "var", "full"), + schema, + null)); + 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, TABLE_LOCATION).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.statsWriteSchema( + geoSchema, + MetricsConfig.from( + ImmutableMap.of( + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "geom", + "full", + TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "geog", + "full"), + geoSchema, + null)); + 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, TABLE_LOCATION) + .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.statsWriteSchema( + geoSchema, + MetricsConfig.from( + ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "geom", "full"), + geoSchema, + null)); + Types.StructType geomStatsType = statsType.fieldType("geom").asStructType(); + + // geo bounds are bounding box structs that the reader reuses across entries, so a copy that + // keeps a reference to one reports the last entry's box for every entry + 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, TABLE_LOCATION) + .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(); + } +}