[WIP]Core, Parquet: Carry avg value sizes for v4 content stats

api/src/main/java/org/apache/iceberg/ContentFile.java

@@ -93,6 +93,14 @@ default String location() {
   /** Returns if collected, map from column ID to its NaN value count, null otherwise. */
   Map<Integer, Long> nanValueCounts();
 
+  /**
+   * Returns if collected, map from column ID to its average non-null value size in bytes, null
+   * otherwise.
+   */
+  default Map<Integer, Integer> avgValueSizes() {
+    return null;
+  }
+
   /** Returns if collected, map from column ID to value lower bounds, null otherwise. */
   Map<Integer, ByteBuffer> lowerBounds();
 
@@ -187,7 +195,7 @@ default Long firstRowId() {
    * to copy data without stats when collecting files.
    *
    * @return a copy of this data file, without lower bounds, upper bounds, value counts, null value
-   *     counts, or nan value counts
+   *     counts, nan value counts, or average value sizes
    */
   F copyWithoutStats();
 
@@ -198,7 +206,7 @@ default Long firstRowId() {
    *
    * @param requestedColumnIds column IDs for which to keep stats.
    * @return a copy of data file, with lower bounds, upper bounds, value counts, null value counts,
-   *     and nan value counts for only specific columns.
+   *     nan value counts, and average value sizes for only specific columns.
    */
   default F copyWithStats(Set<Integer> requestedColumnIds) {
     throw new UnsupportedOperationException(
@@ -211,8 +219,8 @@ default F copyWithStats(Set<Integer> requestedColumnIds) {
    *
    * @param withStats Will copy this file without file stats if set to <code>false</code>.
    * @return a copy of this data file. If <code>withStats</code> is set to <code>false</code> the
-   *     file will not contain lower bounds, upper bounds, value counts, null value counts, or nan
-   *     value counts
+   *     file will not contain lower bounds, upper bounds, value counts, null value counts, nan
+   *     value counts, or average value sizes
    */
   default F copy(boolean withStats) {
     return withStats ? copy() : copyWithoutStats();

api/src/main/java/org/apache/iceberg/Metrics.java

@@ -21,6 +21,7 @@
 import java.io.IOException;
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
+import java.io.OptionalDataException;
 import java.io.Serializable;
 import java.nio.ByteBuffer;
 import java.util.Map;
@@ -31,11 +32,14 @@
 /** Iceberg file format metrics. */
 public class Metrics implements Serializable {
 
+  private static final long serialVersionUID = 5337054944254511702L;
+
   private Long rowCount = null;
   private Map<Integer, Long> columnSizes = null;
   private Map<Integer, Long> valueCounts = null;
   private Map<Integer, Long> nullValueCounts = null;
   private Map<Integer, Long> nanValueCounts = null;
+  private Map<Integer, Integer> avgValueSizes = null;
   private Map<Integer, ByteBuffer> lowerBounds = null;
   private Map<Integer, ByteBuffer> upperBounds = null;
   // this is not serialized with all the other fields
@@ -53,7 +57,16 @@ public Metrics(
       Map<Integer, Long> valueCounts,
       Map<Integer, Long> nullValueCounts,
       Map<Integer, Long> nanValueCounts) {
-    this(rowCount, columnSizes, valueCounts, nullValueCounts, nanValueCounts, null, null, null);
+    this(
+        rowCount,
+        columnSizes,
+        valueCounts,
+        nullValueCounts,
+        nanValueCounts,
+        null,
+        null,
+        null,
+        null);
   }
 
   public Metrics(
@@ -72,6 +85,7 @@ public Metrics(
         nanValueCounts,
         lowerBounds,
         upperBounds,
+        null,
         null);
   }
 
@@ -84,11 +98,34 @@ public Metrics(
       Map<Integer, ByteBuffer> lowerBounds,
       Map<Integer, ByteBuffer> upperBounds,
       Map<Integer, Type> originalTypes) {
+    this(
+        rowCount,
+        columnSizes,
+        valueCounts,
+        nullValueCounts,
+        nanValueCounts,
+        lowerBounds,
+        upperBounds,
+        originalTypes,
+        null);
+  }
+
+  public Metrics(
+      Long rowCount,
+      Map<Integer, Long> columnSizes,
+      Map<Integer, Long> valueCounts,
+      Map<Integer, Long> nullValueCounts,
+      Map<Integer, Long> nanValueCounts,
+      Map<Integer, ByteBuffer> lowerBounds,
+      Map<Integer, ByteBuffer> upperBounds,
+      Map<Integer, Type> originalTypes,
+      Map<Integer, Integer> avgValueSizes) {
     this.rowCount = rowCount;
     this.columnSizes = columnSizes;
     this.valueCounts = valueCounts;
     this.nullValueCounts = nullValueCounts;
     this.nanValueCounts = nanValueCounts;
+    this.avgValueSizes = avgValueSizes;
     this.lowerBounds = lowerBounds;
     this.upperBounds = upperBounds;
     this.originalTypes = originalTypes;
@@ -139,6 +176,15 @@ public Map<Integer, Long> nanValueCounts() {
     return nanValueCounts;
   }
 
+  /**
+   * Get the average non-null value size in bytes for all fields where it was collected.
+   *
+   * @return a Map of fieldId to average value size in bytes
+   */
+  public Map<Integer, Integer> avgValueSizes() {
+    return avgValueSizes;
+  }
+
   /**
    * Get the non-null lower bound values for all fields in a file.
    *
@@ -186,6 +232,7 @@ private void writeObject(ObjectOutputStream out) throws IOException {
 
     writeByteBufferMap(out, lowerBounds);
     writeByteBufferMap(out, upperBounds);
+    out.writeObject(avgValueSizes);
   }
 
   private static void writeByteBufferMap(
@@ -222,6 +269,15 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
 
     lowerBounds = readByteBufferMap(in);
     upperBounds = readByteBufferMap(in);
+    try {
+      avgValueSizes = (Map<Integer, Integer>) in.readObject();
+    } catch (OptionalDataException e) {
+      if (!e.eof) {
+        throw e;
+      }
+
+      avgValueSizes = null;
+    }
   }
 
   @SuppressWarnings("DangerousJavaDeserialization")

api/src/test/java/org/apache/iceberg/TestMetricsSerialization.java

@@ -26,6 +26,7 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.nio.ByteBuffer;
+import java.util.Base64;
 import java.util.Map;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
@@ -35,6 +36,14 @@
 
 public class TestMetricsSerialization {
 
+  private static final String OLD_SERIALIZED_METRICS =
+      "rO0ABXNyABpvcmcuYXBhY2hlLmljZWJlcmcuTWV0cmljc0oRBzHjCEpWAwAITAALY29sdW1uU2l6"
+          + "ZXN0AA9MamF2YS91dGlsL01hcDtMAAtsb3dlckJvdW5kc3EAfgABTAAObmFuVmFsdWVDb3VudHNx"
+          + "AH4AAUwAD251bGxWYWx1ZUNvdW50c3EAfgABTAANb3JpZ2luYWxUeXBlc3EAfgABTAAIcm93Q291"
+          + "bnR0ABBMamF2YS9sYW5nL0xvbmc7TAALdXBwZXJCb3VuZHNxAH4AAUwAC3ZhbHVlQ291bnRzcQB+"
+          + "AAF4cHNyAA5qYXZhLmxhbmcuTG9uZzuL5JDMjyPfAgABSgAFdmFsdWV4cgAQamF2YS5sYW5nLk51"
+          + "bWJlcoaslR0LlOCLAgAAeHAAAAAAAAAAB3BwcHB3CP//////////eA==";
+
   @Test
   public void testSerialization() throws IOException, ClassNotFoundException {
     Metrics original = generateMetrics();
@@ -55,6 +64,14 @@ public void testSerializationWithNulls() throws IOException, ClassNotFoundExcept
     assertEquals(original, result);
   }
 
+  @Test
+  void oldSerializationCompatibility() throws IOException, ClassNotFoundException {
+    Metrics metrics = deserialize(Base64.getDecoder().decode(OLD_SERIALIZED_METRICS));
+
+    assertThat(metrics.recordCount()).isEqualTo(7L);
+    assertThat(metrics.avgValueSizes()).isNull();
+  }
+
   private static byte[] serialize(Metrics metrics) throws IOException {
     try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
       ObjectOutputStream objectOutputStream = new ObjectOutputStream(byteArrayOutputStream);
@@ -93,7 +110,9 @@ private static Metrics generateMetrics() {
 
     Map<Integer, Type> originalTypes =
         ImmutableMap.of(1, Types.IntegerType.get(), 2, Types.IntegerType.get());
-    return new Metrics(0L, longMap1, longMap2, longMap3, null, byteMap1, byteMap2, originalTypes);
+    Map<Integer, Integer> avgValueSizes = ImmutableMap.of(9, 10);
+    return new Metrics(
+        0L, longMap1, longMap2, longMap3, null, byteMap1, byteMap2, originalTypes, avgValueSizes);
   }
 
   private static Metrics generateMetricsWithNulls() {
@@ -106,14 +125,19 @@ private static Metrics generateMetricsWithNulls() {
     byteMap.put(4, null);
 
     Map<Integer, Type> originalTypes = ImmutableMap.of(4, Types.IntegerType.get());
-    return new Metrics(null, null, longMap, longMap, null, null, byteMap, originalTypes);
+    Map<Integer, Integer> avgValueSizes = Maps.newHashMap();
+    avgValueSizes.put(null, 1);
+    avgValueSizes.put(2, null);
+    return new Metrics(
+        null, null, longMap, longMap, null, null, byteMap, originalTypes, avgValueSizes);
   }
 
   private static void assertEquals(Metrics expected, Metrics actual) {
     assertThat(actual.recordCount()).isEqualTo(expected.recordCount());
     assertThat(actual.columnSizes()).isEqualTo(expected.columnSizes());
     assertThat(actual.valueCounts()).isEqualTo(expected.valueCounts());
     assertThat(actual.nullValueCounts()).isEqualTo(expected.nullValueCounts());
+    assertThat(actual.avgValueSizes()).isEqualTo(expected.avgValueSizes());
 
     assertEquals(expected.lowerBounds(), actual.lowerBounds());
     assertEquals(expected.upperBounds(), actual.upperBounds());

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

@@ -68,6 +68,7 @@ abstract class BaseFile<F> extends SupportsIndexProjection
   private Map<Integer, Long> valueCounts = null;
   private Map<Integer, Long> nullValueCounts = null;
   private Map<Integer, Long> nanValueCounts = null;
+  private Map<Integer, Integer> avgValueSizes = null;
   private Map<Integer, ByteBuffer> lowerBounds = null;
   private Map<Integer, ByteBuffer> upperBounds = null;
   private long[] splitOffsets = null;
@@ -146,6 +147,7 @@ abstract class BaseFile<F> extends SupportsIndexProjection
       Map<Integer, Long> valueCounts,
       Map<Integer, Long> nullValueCounts,
       Map<Integer, Long> nanValueCounts,
+      Map<Integer, Integer> avgValueSizes,
       Map<Integer, ByteBuffer> lowerBounds,
       Map<Integer, ByteBuffer> upperBounds,
       List<Long> splitOffsets,
@@ -178,6 +180,7 @@ abstract class BaseFile<F> extends SupportsIndexProjection
     this.valueCounts = valueCounts;
     this.nullValueCounts = nullValueCounts;
     this.nanValueCounts = nanValueCounts;
+    this.avgValueSizes = avgValueSizes;
     this.lowerBounds = SerializableByteBufferMap.wrap(lowerBounds);
     this.upperBounds = SerializableByteBufferMap.wrap(upperBounds);
     this.splitOffsets = ArrayUtil.toLongArray(splitOffsets);
@@ -215,13 +218,15 @@ abstract class BaseFile<F> extends SupportsIndexProjection
       this.valueCounts = copyMap(toCopy.valueCounts, requestedColumnIds);
       this.nullValueCounts = copyMap(toCopy.nullValueCounts, requestedColumnIds);
       this.nanValueCounts = copyMap(toCopy.nanValueCounts, requestedColumnIds);
+      this.avgValueSizes = copyMap(toCopy.avgValueSizes, requestedColumnIds);
       this.lowerBounds = copyByteBufferMap(toCopy.lowerBounds, requestedColumnIds);
       this.upperBounds = copyByteBufferMap(toCopy.upperBounds, requestedColumnIds);
     } else {
       this.columnSizes = null;
       this.valueCounts = null;
       this.nullValueCounts = null;
       this.nanValueCounts = null;
+      this.avgValueSizes = null;
       this.lowerBounds = null;
       this.upperBounds = null;
     }
@@ -511,6 +516,11 @@ public Map<Integer, Long> nanValueCounts() {
     return toReadableMap(nanValueCounts);
   }
 
+  @Override
+  public Map<Integer, Integer> avgValueSizes() {
+    return toReadableMap(avgValueSizes);
+  }
+
   @Override
   public Map<Integer, ByteBuffer> lowerBounds() {
     return toReadableByteBufferMap(lowerBounds);
@@ -648,6 +658,7 @@ public String toString() {
         .add("value_counts", valueCounts)
         .add("null_value_counts", nullValueCounts)
         .add("nan_value_counts", nanValueCounts)
+        .add("avg_value_sizes", avgValueSizes)
         .add("lower_bounds", lowerBounds)
         .add("upper_bounds", upperBounds)
         .add("key_metadata", keyMetadata == null ? "null" : "(redacted)")

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

@@ -35,6 +35,7 @@ private enum Kind {
     VALUE_COUNT,
     NULL_VALUE_COUNT,
     NAN_VALUE_COUNT,
+    AVG_VALUE_SIZE,
     LOWER_BOUND,
     UPPER_BOUND
   }
@@ -54,6 +55,11 @@ static <V> Map<Integer, V> nanValueCounts(ContentStats stats) {
     return viewOrNull(stats, Kind.NAN_VALUE_COUNT);
   }
 
+  /** Per-column average non-null value sizes, or null if no column tracks the average size. */
+  static <V> Map<Integer, V> avgValueSizes(ContentStats stats) {
+    return viewOrNull(stats, Kind.AVG_VALUE_SIZE);
+  }
+
   /** Per-column lower bounds, or null if no column tracks a lower bound. */
   static <V> Map<Integer, V> lowerBounds(ContentStats stats) {
     return viewOrNull(stats, Kind.LOWER_BOUND);
@@ -142,6 +148,7 @@ private static boolean isKnown(FieldStats<?> fieldStats, Kind kind) {
       case VALUE_COUNT -> fieldStats.hasValueCount();
       case NULL_VALUE_COUNT -> fieldStats.hasNullValueCount();
       case NAN_VALUE_COUNT -> fieldStats.hasNanValueCount();
+      case AVG_VALUE_SIZE -> fieldStats.avgValueSizeInBytes() != null;
       case LOWER_BOUND -> fieldStats.lowerBound() != null;
       case UPPER_BOUND -> fieldStats.upperBound() != null;
     };
@@ -156,6 +163,7 @@ private static <V> V statValue(FieldStats<?> fieldStats, Kind kind) {
           fieldStats.hasNullValueCount() ? (V) Long.valueOf(fieldStats.nullValueCount()) : null;
       case NAN_VALUE_COUNT ->
           fieldStats.hasNanValueCount() ? (V) Long.valueOf(fieldStats.nanValueCount()) : null;
+      case AVG_VALUE_SIZE -> (V) fieldStats.avgValueSizeInBytes();
       case LOWER_BOUND ->
           (V) bound(fieldStats, fieldStats.lowerBound(), StatsUtil.LOWER_BOUND_NAME);
       case UPPER_BOUND ->

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

@@ -150,6 +150,7 @@ public static class Builder {
     private Map<Integer, Long> valueCounts = null;
     private Map<Integer, Long> nullValueCounts = null;
     private Map<Integer, Long> nanValueCounts = null;
+    private Map<Integer, Integer> avgValueSizes = null;
     private Map<Integer, ByteBuffer> lowerBounds = null;
     private Map<Integer, ByteBuffer> upperBounds = null;
     private Map<Integer, Type> originalTypes = null;
@@ -177,6 +178,7 @@ public void clear() {
       this.valueCounts = null;
       this.nullValueCounts = null;
       this.nanValueCounts = null;
+      this.avgValueSizes = null;
       this.lowerBounds = null;
       this.upperBounds = null;
       this.splitOffsets = null;
@@ -198,6 +200,7 @@ public Builder copy(DataFile toCopy) {
       this.valueCounts = toCopy.valueCounts();
       this.nullValueCounts = toCopy.nullValueCounts();
       this.nanValueCounts = toCopy.nanValueCounts();
+      this.avgValueSizes = toCopy.avgValueSizes();
       this.lowerBounds = toCopy.lowerBounds();
       this.upperBounds = toCopy.upperBounds();
       this.keyMetadata =
@@ -290,6 +293,7 @@ public Builder withMetrics(Metrics metrics) {
       this.valueCounts = metrics.valueCounts();
       this.nullValueCounts = metrics.nullValueCounts();
       this.nanValueCounts = metrics.nanValueCounts();
+      this.avgValueSizes = metrics.avgValueSizes();
       this.lowerBounds = metrics.lowerBounds();
       this.upperBounds = metrics.upperBounds();
       this.originalTypes = metrics.originalTypes();
@@ -354,7 +358,8 @@ public DataFile build() {
               nanValueCounts,
               lowerBounds,
               upperBounds,
-              originalTypes),
+              originalTypes,
+              avgValueSizes),
           keyMetadata,
           splitOffsets,
           sortOrderId,

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

@@ -232,6 +232,11 @@ public Map<Integer, Long> nanValueCounts() {
       return wrapped.nanValueCounts();
     }
 
+    @Override
+    public Map<Integer, Integer> avgValueSizes() {
+      return wrapped.avgValueSizes();
+    }
+
     @Override
     public Map<Integer, ByteBuffer> lowerBounds() {
       return wrapped.lowerBounds();

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

@@ -63,7 +63,7 @@ interface FieldStats<T> {
 
   /**
    * The avg value size in memory (uncompressed) in bytes for variable-length types (string, binary,
-   * variant)
+   * variant, geometry, geography)
    */
   Integer avgValueSizeInBytes();