Core: present v4 ContentStats as legacy stat maps via ContentStatsBackedMap

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

@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg;
+
+import java.nio.ByteBuffer;
+import java.util.AbstractMap;
+import java.util.Map;
+import java.util.Set;
+import org.apache.iceberg.relocated.com.google.common.collect.Sets;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Type;
+
+/**
+ * A lazy, read-only {@link Map} view of one stat across the columns of a {@link ContentStats},
+ * keyed by field ID, mirroring the per-column stat maps on {@link ContentFile}.
+ */
+class ContentStatsBackedMap<V> extends AbstractMap<Integer, V> {
+  private enum Kind {
+    VALUE_COUNT,
+    NULL_VALUE_COUNT,
+    NAN_VALUE_COUNT,
+    LOWER_BOUND,
+    UPPER_BOUND
+  }
+
+  /** Per-column value counts, or {@code null} if no column tracks the value count. */
+  static <V> Map<Integer, V> valueCounts(ContentStats stats) {
+    return viewOrNull(stats, Kind.VALUE_COUNT);
+  }
+
+  /** Per-column null value counts, or {@code null} if no column tracks the null value count. */
+  static <V> Map<Integer, V> nullValueCounts(ContentStats stats) {
+    return viewOrNull(stats, Kind.NULL_VALUE_COUNT);
+  }
+
+  /** Per-column NaN value counts, or {@code null} if no column tracks the NaN value count. */
+  static <V> Map<Integer, V> nanValueCounts(ContentStats stats) {
+    return viewOrNull(stats, Kind.NAN_VALUE_COUNT);
+  }
+
+  /** Per-column lower bounds, or {@code null} if no column tracks a lower bound. */
+  static <V> Map<Integer, V> lowerBounds(ContentStats stats) {
+    return viewOrNull(stats, Kind.LOWER_BOUND);
+  }
+
+  /** Per-column upper bounds, or {@code null} if no column tracks an upper bound. */
+  static <V> Map<Integer, V> upperBounds(ContentStats stats) {
+    return viewOrNull(stats, Kind.UPPER_BOUND);
+  }
+
+  private static <V> Map<Integer, V> viewOrNull(ContentStats stats, Kind kind) {
+    return isEmpty(stats, kind) ? null : new ContentStatsBackedMap<>(stats, kind);
+  }
+
+  private final ContentStats stats;
+  private final Kind kind;
+  private Set<Entry<Integer, V>> materialized;
+
+  private ContentStatsBackedMap(ContentStats stats, Kind kind) {
+    this.stats = stats;
+    this.kind = kind;
+  }
+
+  @Override
+  public V get(Object key) {
+    if (!(key instanceof Integer)) {
+      return null;
+    }
+
+    FieldStats<?> fieldStats = stats.statsFor((Integer) key);
+    if (fieldStats == null) {
+      return null;
+    }
+
+    return statValue(fieldStats, kind);
+  }
+
+  @Override
+  public boolean containsKey(Object key) {
+    return get(key) != null;
+  }
+
+  @Override
+  public boolean isEmpty() {
+    // avoid AbstractMap's default, which materializes entrySet() just to answer emptiness
+    return isEmpty(stats, kind);
+  }
+
+  @Override
+  public Set<Entry<Integer, V>> entrySet() {
+    if (materialized == null) {
+      Set<Entry<Integer, V>> entries = Sets.newLinkedHashSet();
+      for (FieldStats<?> fieldStats : stats.fieldStats()) {
+        if (fieldStats != null) {
+          V value = statValue(fieldStats, kind);
+          if (value != null) {
+            entries.add(new SimpleImmutableEntry<>(fieldStats.fieldId(), value));
+          }
+        }
+      }
+
+      this.materialized = entries;
+    }
+
+    return materialized;
+  }
+
+  /** Returns whether no column contributes an entry for the metric. */
+  private static boolean isEmpty(ContentStats stats, Kind kind) {
+    for (FieldStats<?> fieldStats : stats.fieldStats()) {
+      if (fieldStats != null && isKnown(fieldStats, kind)) {
+        return false;
+      }
+    }
+
+    return true;
+  }
+
+  // Whether statValue would return a non-null value, without allocating a boxed count or decoding a
+  // bound. Must mirror statValue's null-ness.
+  private static boolean isKnown(FieldStats<?> fieldStats, Kind kind) {
+    switch (kind) {
+      case VALUE_COUNT:
+        return fieldStats.hasValueCount();
+      case NULL_VALUE_COUNT:
+        return fieldStats.hasNullValueCount();
+      case NAN_VALUE_COUNT:
+        return fieldStats.hasNanValueCount();
+      case LOWER_BOUND:
+        return fieldStats.lowerBound() != null;
+      case UPPER_BOUND:
+        return fieldStats.upperBound() != null;
+      default:
+        throw new IllegalArgumentException("Unknown content stats kind: " + kind);
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private static <V> V statValue(FieldStats<?> fieldStats, Kind kind) {
+    switch (kind) {
+      case VALUE_COUNT:
+        return fieldStats.hasValueCount() ? (V) Long.valueOf(fieldStats.valueCount()) : null;
+      case NULL_VALUE_COUNT:
+        return fieldStats.hasNullValueCount()
+            ? (V) Long.valueOf(fieldStats.nullValueCount())
+            : null;
+      case NAN_VALUE_COUNT:
+        return fieldStats.hasNanValueCount() ? (V) Long.valueOf(fieldStats.nanValueCount()) : null;
+      case LOWER_BOUND:
+        return (V) bound(fieldStats, fieldStats.lowerBound(), StatsUtil.LOWER_BOUND_NAME);
+      case UPPER_BOUND:
+        return (V) bound(fieldStats, fieldStats.upperBound(), StatsUtil.UPPER_BOUND_NAME);
+      default:
+        throw new IllegalArgumentException("Unknown content stats kind: " + kind);
+    }
+  }
+
+  private static ByteBuffer bound(FieldStats<?> fieldStats, Object bound, String boundFieldName) {
+    Type boundType = fieldStats.type().fieldType(boundFieldName);
+    // toByteBuffer returns null for a null bound
+    return boundType == null ? null : Conversions.toByteBuffer(boundType, bound);
+  }
+}

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

@@ -43,13 +43,22 @@ interface FieldStats<T> {
    */
   boolean tightBounds();
 
-  /** The total value count, including null and NaN */
+  /** Whether a value count is tracked for this field. */
+  boolean hasValueCount();
+
+  /** The total value count, including null and NaN, defined only when {@link #hasValueCount()}. */
   long valueCount();
 
-  /** The total null value count */
+  /** Whether a null value count is tracked for this field. */
+  boolean hasNullValueCount();
+
+  /** The total null value count, defined only when {@link #hasNullValueCount()}. */
   long nullValueCount();
 
-  /** The total NaN value count */
+  /** Whether a NaN value count is tracked for this field. */
+  boolean hasNanValueCount();
+
+  /** The total NaN value count, defined only when {@link #hasNanValueCount()}. */
   long nanValueCount();
 
   /**

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

@@ -143,18 +143,38 @@ public boolean tightBounds() {
     return tightBounds;
   }
 
+  @Override
+  public boolean hasValueCount() {
+    return valueCount != null;
+  }
+
   @Override
   public long valueCount() {
+    Preconditions.checkState(hasValueCount(), "Field %s does not track a value count", fieldId);
     return valueCount;
   }
 
+  @Override
+  public boolean hasNullValueCount() {
+    return nullValueCount != null;
+  }
+
   @Override
   public long nullValueCount() {
+    Preconditions.checkState(
+        hasNullValueCount(), "Field %s does not track a null value count", fieldId);
     return nullValueCount;
   }
 
+  @Override
+  public boolean hasNanValueCount() {
+    return nanValueCount != null;
+  }
+
   @Override
   public long nanValueCount() {
+    Preconditions.checkState(
+        hasNanValueCount(), "Field %s does not track a NaN value count", fieldId);
     return nanValueCount;
   }
 

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

@@ -20,8 +20,6 @@
 
 import static org.apache.iceberg.types.Types.NestedField.optional;
 
-import java.nio.ByteBuffer;
-import java.util.Collections;
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
@@ -478,88 +476,4 @@ public <T> void set(int pos, T value) {
       throw new UnsupportedOperationException("StructWithReadableMetrics is read only");
     }
   }
-
-  static Map<Integer, Long> valueCounts(ContentStats stats) {
-    if (stats == null) {
-      return null;
-    }
-
-    Map<Integer, Long> result = Maps.newHashMap();
-    for (FieldStats<?> fs : stats.fieldStats()) {
-      if (fs != null) {
-        result.put(fs.fieldId(), fs.valueCount());
-      }
-    }
-
-    return result.isEmpty() ? null : Collections.unmodifiableMap(result);
-  }
-
-  static Map<Integer, Long> nullValueCounts(ContentStats stats) {
-    if (stats == null) {
-      return null;
-    }
-
-    Map<Integer, Long> result = Maps.newHashMap();
-    for (FieldStats<?> fs : stats.fieldStats()) {
-      if (fs != null) {
-        result.put(fs.fieldId(), fs.nullValueCount());
-      }
-    }
-
-    return result.isEmpty() ? null : Collections.unmodifiableMap(result);
-  }
-
-  static Map<Integer, Long> nanValueCounts(ContentStats stats) {
-    if (stats == null) {
-      return null;
-    }
-
-    Map<Integer, Long> result = Maps.newHashMap();
-    for (FieldStats<?> fs : stats.fieldStats()) {
-      if (fs != null) {
-        Type boundType = fs.type().fieldType("lower_bound");
-        if (boundType.typeId() == Type.TypeID.FLOAT || boundType.typeId() == Type.TypeID.DOUBLE) {
-          result.put(fs.fieldId(), fs.nanValueCount());
-        }
-      }
-    }
-
-    return result.isEmpty() ? null : Collections.unmodifiableMap(result);
-  }
-
-  static Map<Integer, ByteBuffer> lowerBounds(ContentStats stats) {
-    if (stats == null) {
-      return null;
-    }
-
-    Map<Integer, ByteBuffer> result = Maps.newHashMap();
-    for (FieldStats<?> fs : stats.fieldStats()) {
-      if (fs != null) {
-        Type boundType = fs.type().fieldType("lower_bound");
-        if (fs.lowerBound() != null && boundType != null) {
-          result.put(fs.fieldId(), Conversions.toByteBuffer(boundType, fs.lowerBound()));
-        }
-      }
-    }
-
-    return result.isEmpty() ? null : Collections.unmodifiableMap(result);
-  }
-
-  static Map<Integer, ByteBuffer> upperBounds(ContentStats stats) {
-    if (stats == null) {
-      return null;
-    }
-
-    Map<Integer, ByteBuffer> result = Maps.newHashMap();
-    for (FieldStats<?> fs : stats.fieldStats()) {
-      if (fs != null) {
-        Type boundType = fs.type().fieldType("upper_bound");
-        if (fs.upperBound() != null && boundType != null) {
-          result.put(fs.fieldId(), Conversions.toByteBuffer(boundType, fs.upperBound()));
-        }
-      }
-    }
-
-    return result.isEmpty() ? null : Collections.unmodifiableMap(result);
-  }
 }

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

@@ -55,6 +55,11 @@ private StatsUtil() {}
   static final int NAN_VALUE_COUNT_OFFSET = 6;
   static final int AVG_VALUE_SIZE_OFFSET = 7;
 
+  // Bound field names, shared with the read-side lookups in ContentStatsBackedMap so a rename or
+  // typo cannot silently drift between the schema and the reader.
+  static final String LOWER_BOUND_NAME = "lower_bound";
+  static final String UPPER_BOUND_NAME = "upper_bound";
+
   // Offsets used within geo_lower struct
   private static final int GEO_LOWER_X_OFFSET = 10;
   private static final int GEO_LOWER_Y_OFFSET = 11;
@@ -249,12 +254,12 @@ private static Types.StructType geoUpperBound(int baseId) {
 
   private static Types.NestedField lowerBoundField(Type type, int baseId) {
     Type boundType = isGeoType(type) ? geoLowerBound(baseId) : type;
-    return optional(baseId + LOWER_BOUND_OFFSET, "lower_bound", boundType);
+    return optional(baseId + LOWER_BOUND_OFFSET, LOWER_BOUND_NAME, boundType);
   }
 
   private static Types.NestedField upperBoundField(Type type, int baseId) {
     Type boundType = isGeoType(type) ? geoUpperBound(baseId) : type;
-    return optional(baseId + UPPER_BOUND_OFFSET, "upper_bound", boundType);
+    return optional(baseId + UPPER_BOUND_OFFSET, UPPER_BOUND_NAME, boundType);
   }
 
   @VisibleForTesting