Core: Read content stats from v4 Manifest

core/src/main/java/org/apache/iceberg/ContentStatsStruct.java

@@ -49,7 +49,9 @@ private ContentStatsStruct(ContentStatsStruct toCopy, Set<Integer> fieldIds) {
       }
     } else {
       for (Map.Entry<Integer, FieldStats<?>> entry : toCopy.idToFieldStats.entrySet()) {
-        idToFieldStats.put(entry.getKey(), entry.getValue().copy());
+        if (entry.getValue() != null) {
+          idToFieldStats.put(entry.getKey(), entry.getValue().copy());
+        }
       }
     }
   }

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<T> implements FieldStats<T>, StructLike, Serializable {
   private final Types.StructType struct;
@@ -45,7 +46,7 @@ class FieldStatsStruct<T> implements FieldStats<T>, 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<T> implements FieldStats<T>, StructLike, Serializable {
 
   private FieldStatsStruct(FieldStatsStruct<T> 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.
+   *
+   * <p>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);
   }

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) {

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());
     }

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<Integer, PartitionSpec> specsById) {
-    return new Builder(file, specsById);
+  static Builder builder(
+      InputFile file, Schema tableSchema, Map<Integer, PartitionSpec> specsById) {
+    return new Builder(file, tableSchema, specsById);
   }
 
   /** Returns copies of the tracked files that match this reader's configured filters. */
@@ -122,16 +127,27 @@ private CloseableIterable<TrackedFile> open() {
     Preconditions.checkArgument(
         format != null, "Cannot determine format of manifest: %s", file.location());
 
-    CloseableIterable<TrackedFile> reader =
+    InternalData.ReadBuilder readBuilder =
         InternalData.read(format, file)
             .project(readSchema)
             .setRootType(TrackedFileStruct.class)
             .setCustomType(TrackedFile.TRACKING.fieldId(), TrackingStruct.class)
             .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<TrackedFile> 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<Integer, PartitionSpec> specsById;
-    private final Schema fullSchema;
     private Expression rowFilter = Expressions.alwaysTrue();
     private boolean caseSensitive = true;
     private boolean includeAll = false;
     private boolean scanPlanning = false;
     private Collection<String> columns = null;
     private Schema requestedProjection = null;
+    private Set<Integer> statsProjectionForFieldIds = null;
     private ScanMetrics scanMetrics = ScanMetrics.noop();
 
-    private Builder(InputFile file, Map<Integer, PartitionSpec> specsById) {
+    private Builder(InputFile file, Schema tableSchema, Map<Integer, PartitionSpec> 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<Integer> 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<Integer> 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<Integer> 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<Integer> 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<Integer> requiredStatsProjectionForFieldIds() {
+      Set<Integer> 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<Integer> requiredStatsProjectionFieldIds,
+        boolean hasPartitionFilter) {
       Set<Integer> 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);
     }

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;
     }

core/src/test/java/org/apache/iceberg/TestFieldStatsStruct.java

@@ -317,6 +317,10 @@ public void geoSerialization(Type geoType, RoundTripSerializer<FieldStatsStruct<
     assertThat(copy.fieldId()).isEqualTo(stats.fieldId());
     assertThat(copy.type()).isEqualTo(stats.type());
     assertThat(comparator.compare(copy, stats)).isEqualTo(0);
+
+    // readers hold copies, so a copied bounding box must survive the same round trip
+    FieldStatsStruct<?> roundTrippedCopy = serializer.apply(stats.copy());
+    assertThat(comparator.compare(roundTrippedCopy, stats)).isEqualTo(0);
   }
 
   // Variant is not Serializable so this does not test Java serialization