Core: Read content stats from v4 Manifest

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,10 @@ 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) {
+    Preconditions.checkArgument(tableSchema != null, "Invalid table schema: null");
+    return new Builder(file, tableSchema, specsById);
   }
 
   /** Returns copies of the tracked files that match this reader's configured filters. */
@@ -122,16 +128,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 when it is not projected and unknown when no stats are read
+    Types.NestedField statsField = readSchema.findField(TrackedFile.CONTENT_STATS_ID);
+    if (statsField != null && statsField.type().isStructType()) {
+      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 +170,23 @@ 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 Collection<Integer> statsFieldIds = null;
     private ScanMetrics scanMetrics = ScanMetrics.noop();
 
-    private Builder(InputFile file, Map<Integer, PartitionSpec> specsById) {
+    private Builder(InputFile file, Schema tableSchema, Map<Integer, PartitionSpec> specsById) {
       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. */
@@ -193,7 +207,10 @@ Builder includeAll() {
       return this;
     }
 
-    /** Configures the reader to select the minimal fields needed for scan planning. */
+    /**
+     * Configures the reader to select the minimal fields needed for scan planning. Stats are
+     * narrowed to the fields that the {@link #filter(Expression) filter} needs.
+     */
     Builder forScanPlanning() {
       Preconditions.checkState(
           columns == null && requestedProjection == null,
@@ -228,6 +245,22 @@ Builder project(Schema newProjection) {
       return this;
     }
 
+    /** Reads content stats for the given table field IDs instead of for every field. */
+    Builder projectStats(int... fieldIds) {
+      Preconditions.checkArgument(fieldIds != null, "Invalid stats 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(Collection<Integer> fieldIds) {
+      Preconditions.checkArgument(fieldIds != null, "Invalid stats field IDs: null");
+      this.statsFieldIds = fieldIds;
+      return this;
+    }
+
     Builder scanMetrics(ScanMetrics newScanMetrics) {
       Preconditions.checkArgument(newScanMetrics != null, "Invalid scan metrics: null");
       this.scanMetrics = newScanMetrics;
@@ -254,6 +287,7 @@ V4ManifestReader build() {
     }
 
     private Schema readSchema(boolean hasPartitionFilter) {
+      Schema fullSchema = fullSchema();
       if (scanPlanning) {
         // scan planning does not read the change-tracking fields omitted by SCAN_TYPE
         return TypeUtil.replaceFieldTypes(
@@ -263,17 +297,62 @@ 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, hasPartitionFilter);
       }
 
       if (requestedProjection != null) {
-        return addRequiredColumns(requestedProjection, hasPartitionFilter);
+        return addRequiredColumns(fullSchema, requestedProjection, 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() {
+      Types.StructType contentStatsType = contentStatsType();
+      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> fieldIds = requiredStatsFieldIds();
+      if (!scanPlanning && statsFieldIds == null) {
+        // projecting all stats carries every column so entries can be copied to a new manifest;
+        // scan planning and projectStats(fieldIds) both narrow the set of stats that are read
+        fieldIds.addAll(TypeUtil.indexById(tableSchema.asStruct()).keySet());
+      }
+
+      return StatsUtil.statsReadSchema(tableSchema, fieldIds);
+    }
+
+    /** Returns the IDs of table fields whose stats are read regardless of the projection. */
+    private Set<Integer> requiredStatsFieldIds() {
+      Set<Integer> fieldIds = Sets.newHashSet();
+      if (statsFieldIds != null) {
+        fieldIds.addAll(statsFieldIds);
+      }
+
+      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, boolean hasPartitionFilter) {
       Set<Integer> projectedIds = Sets.newHashSet(TypeUtil.getProjectedIds(projection));
 
       // fields the reader consumes internally: status for liveness filtering, row_position for
@@ -293,6 +372,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, requiredStatsFieldIds());
+      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);
     }