Core, OpenAPI: Preserve avg value sizes in REST JSON

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

@@ -22,10 +22,12 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.util.JsonUtil;
 
@@ -43,6 +45,8 @@ public class ContentFileParser {
   private static final String NAN_VALUE_COUNTS = "nan-value-counts";
   private static final String LOWER_BOUNDS = "lower-bounds";
   private static final String UPPER_BOUNDS = "upper-bounds";
+  private static final String CONTENT_STATS = "content-stats";
+  private static final String AVG_VALUE_SIZE_IN_BYTES = "avg-value-size-in-bytes";
   private static final String KEY_METADATA = "key-metadata";
   private static final String SPLIT_OFFSETS = "split-offsets";
   private static final String EQUALITY_IDS = "equality-ids";
@@ -240,6 +244,18 @@ private static void metricsToJson(ContentFile<?> contentFile, JsonGenerator gene
       generator.writeFieldName(UPPER_BOUNDS);
       SingleValueParser.toJson(DataFile.UPPER_BOUNDS.type(), contentFile.upperBounds(), generator);
     }
+
+    if (contentFile.avgValueSizes() != null) {
+      generator.writeFieldName(CONTENT_STATS);
+      generator.writeStartObject();
+      for (Map.Entry<Integer, Integer> entry : contentFile.avgValueSizes().entrySet()) {
+        generator.writeObjectFieldStart(String.valueOf(entry.getKey()));
+        generator.writeNumberField(AVG_VALUE_SIZE_IN_BYTES, entry.getValue());
+        generator.writeEndObject();
+      }
+
+      generator.writeEndObject();
+    }
   }
 
   private static Metrics metricsFromJson(JsonNode jsonNode) {
@@ -289,14 +305,54 @@ private static Metrics metricsFromJson(JsonNode jsonNode) {
               SingleValueParser.fromJson(DataFile.UPPER_BOUNDS.type(), jsonNode.get(UPPER_BOUNDS));
     }
 
+    Map<Integer, Integer> avgValueSizes = null;
+    if (jsonNode.hasNonNull(CONTENT_STATS)) {
+      avgValueSizes = avgValueSizesFromJson(jsonNode.get(CONTENT_STATS));
+    }
+
     return new Metrics(
         recordCount,
         columnSizes,
         valueCounts,
         nullValueCounts,
         nanValueCounts,
         lowerBounds,
-        upperBounds);
+        upperBounds,
+        avgValueSizes,
+        null /* originalTypes */);
+  }
+
+  private static Map<Integer, Integer> avgValueSizesFromJson(JsonNode contentStats) {
+    Preconditions.checkArgument(
+        contentStats.isObject(),
+        "Invalid JSON node for content stats: non-object (%s)",
+        contentStats);
+
+    Map<Integer, Integer> avgValueSizes = Maps.newHashMap();
+    Iterator<String> fieldIds = contentStats.fieldNames();
+    while (fieldIds.hasNext()) {
+      String fieldId = fieldIds.next();
+      JsonNode fieldStats = contentStats.get(fieldId);
+      Preconditions.checkArgument(
+          fieldStats != null && fieldStats.isObject(),
+          "Invalid JSON node for field statistics: non-object (%s)",
+          fieldStats);
+      Integer avgValueSize = JsonUtil.getIntOrNull(AVG_VALUE_SIZE_IN_BYTES, fieldStats);
+      if (avgValueSize != null) {
+        avgValueSizes.put(parseFieldId(fieldId), avgValueSize);
+      }
+    }
+
+    return avgValueSizes.isEmpty() ? null : avgValueSizes;
+  }
+
+  private static int parseFieldId(String fieldId) {
+    try {
+      return Integer.parseInt(fieldId);
+    } catch (NumberFormatException e) {
+      throw new IllegalArgumentException(
+          String.format("Invalid field ID for content stats: %s", fieldId), e);
+    }
   }
 
   private static void partitionToJson(

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

@@ -72,13 +72,40 @@ public void testNanCountsOnlyWritesNanValueCounts() throws Exception {
     // ensure nan counts are present and null counts are not emitted
     assertThat(jsonStr).contains("\"nan-value-counts\"");
     assertThat(jsonStr).doesNotContain("\"null-value-counts\"");
+    assertThat(jsonStr).doesNotContain("\"content-stats\"");
     JsonNode jsonNode = JsonUtil.mapper().readTree(jsonStr);
     ContentFile<?> deserialized =
         ContentFileParser.fromJson(jsonNode, Map.of(TestBase.SPEC.specId(), spec));
     assertThat(deserialized).isInstanceOf(DataFile.class);
     assertContentFileEquals(dataFile, deserialized, spec);
   }
 
+  @Test
+  void nullContentStatsIsAbsent() throws Exception {
+    String jsonStr =
+        "{\"spec-id\":0,\"content\":\"data\",\"file-path\":\"/path/to/data.parquet\","
+            + "\"file-format\":\"parquet\",\"partition\":[],\"file-size-in-bytes\":10,"
+            + "\"record-count\":1,\"content-stats\":null}";
+    JsonNode jsonNode = JsonUtil.mapper().readTree(jsonStr);
+    ContentFile<?> contentFile =
+        ContentFileParser.fromJson(jsonNode, Map.of(0, PartitionSpec.unpartitioned()));
+    assertThat(contentFile).isInstanceOf(DataFile.class);
+    assertThat(contentFile.avgValueSizes()).isNull();
+  }
+
+  @Test
+  void invalidContentStatsFieldId() throws Exception {
+    String jsonStr =
+        "{\"spec-id\":0,\"content\":\"data\",\"file-path\":\"/path/to/data.parquet\","
+            + "\"file-format\":\"parquet\",\"partition\":[],\"file-size-in-bytes\":10,"
+            + "\"record-count\":1,\"content-stats\":{\"abc\":{\"avg-value-size-in-bytes\":8}}}";
+    JsonNode jsonNode = JsonUtil.mapper().readTree(jsonStr);
+    assertThatThrownBy(
+            () -> ContentFileParser.fromJson(jsonNode, Map.of(0, PartitionSpec.unpartitioned())))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid field ID for content stats: abc");
+  }
+
   @ParameterizedTest
   @MethodSource("provideSpecAndDataFile")
   public void testDataFile(PartitionSpec spec, DataFile dataFile, String expectedJson)
@@ -357,6 +384,7 @@ private static String dataFileJsonWithAllOptional(PartitionSpec spec) {
           + "\"nan-value-counts\":{\"keys\":[3,4],\"values\":[0,0]},"
           + "\"lower-bounds\":{\"keys\":[3,4],\"values\":[\"01000000\",\"02000000\"]},"
           + "\"upper-bounds\":{\"keys\":[3,4],\"values\":[\"05000000\",\"0A000000\"]},"
+          + "\"content-stats\":{\"3\":{\"avg-value-size-in-bytes\":8},\"4\":{\"avg-value-size-in-bytes\":16}},"
           + "\"key-metadata\":\"00000000000000000000000000000000\","
           + "\"split-offsets\":[128,256],\"sort-order-id\":1}";
     } else {
@@ -368,6 +396,7 @@ private static String dataFileJsonWithAllOptional(PartitionSpec spec) {
           + "\"nan-value-counts\":{\"keys\":[3,4],\"values\":[0,0]},"
           + "\"lower-bounds\":{\"keys\":[3,4],\"values\":[\"01000000\",\"02000000\"]},"
           + "\"upper-bounds\":{\"keys\":[3,4],\"values\":[\"05000000\",\"0A000000\"]},"
+          + "\"content-stats\":{\"3\":{\"avg-value-size-in-bytes\":8},\"4\":{\"avg-value-size-in-bytes\":16}},"
           + "\"key-metadata\":\"00000000000000000000000000000000\","
           + "\"split-offsets\":[128,256],\"sort-order-id\":1}";
     }
@@ -393,8 +422,9 @@ private static DataFile dataFileWithAllOptional(PartitionSpec spec) {
                         3,
                         Conversions.toByteBuffer(Types.IntegerType.get(), 5),
                         4,
-                        Conversions.toByteBuffer(Types.IntegerType.get(), 10)) // upperbounds
-                    ))
+                        Conversions.toByteBuffer(Types.IntegerType.get(), 10)), // upper bounds
+                    ImmutableMap.of(3, 8, 4, 16), // avg value sizes
+                    null /* originalTypes */))
             .withFileSizeInBytes(350)
             .withSplitOffsets(Arrays.asList(128L, 256L))
             .withEncryptionKeyMetadata(ByteBuffer.wrap(new byte[16]))
@@ -533,8 +563,9 @@ private static DeleteFile deleteFileWithAllOptional(PartitionSpec spec) {
                 3,
                 Conversions.toByteBuffer(Types.IntegerType.get(), 5),
                 4,
-                Conversions.toByteBuffer(Types.IntegerType.get(), 10)) // upperbounds
-            );
+                Conversions.toByteBuffer(Types.IntegerType.get(), 10)), // upper bounds
+            ImmutableMap.of(3, 8, 4, 16), // avg value sizes
+            null /* originalTypes */);
 
     return new GenericDeleteFile(
         spec.specId(),
@@ -573,6 +604,7 @@ private static String deleteFileJsonWithAllOptional(PartitionSpec spec) {
           + "\"nan-value-counts\":{\"keys\":[3,4],\"values\":[0,0]},"
           + "\"lower-bounds\":{\"keys\":[3,4],\"values\":[\"01000000\",\"02000000\"]},"
           + "\"upper-bounds\":{\"keys\":[3,4],\"values\":[\"05000000\",\"0A000000\"]},"
+          + "\"content-stats\":{\"3\":{\"avg-value-size-in-bytes\":8},\"4\":{\"avg-value-size-in-bytes\":16}},"
           + "\"key-metadata\":\"00000000000000000000000000000000\","
           + "\"split-offsets\":[128],\"equality-ids\":[3],\"sort-order-id\":1}";
     } else {
@@ -584,6 +616,7 @@ private static String deleteFileJsonWithAllOptional(PartitionSpec spec) {
           + "\"nan-value-counts\":{\"keys\":[3,4],\"values\":[0,0]},"
           + "\"lower-bounds\":{\"keys\":[3,4],\"values\":[\"01000000\",\"02000000\"]},"
           + "\"upper-bounds\":{\"keys\":[3,4],\"values\":[\"05000000\",\"0A000000\"]},"
+          + "\"content-stats\":{\"3\":{\"avg-value-size-in-bytes\":8},\"4\":{\"avg-value-size-in-bytes\":16}},"
           + "\"key-metadata\":\"00000000000000000000000000000000\","
           + "\"split-offsets\":[128],\"equality-ids\":[3],\"sort-order-id\":1}";
     }
@@ -607,6 +640,7 @@ static void assertContentFileEquals(
     assertThat(actual.nanValueCounts()).isEqualTo(expected.nanValueCounts());
     assertThat(actual.lowerBounds()).isEqualTo(expected.lowerBounds());
     assertThat(actual.upperBounds()).isEqualTo(expected.upperBounds());
+    assertThat(actual.avgValueSizes()).isEqualTo(expected.avgValueSizes());
     assertThat(actual.keyMetadata()).isEqualTo(expected.keyMetadata());
     assertThat(actual.splitOffsets()).isEqualTo(expected.splitOffsets());
     assertThat(actual.equalityFieldIds()).isEqualTo(expected.equalityFieldIds());

open-api/rest-catalog-open-api.py

@@ -1034,6 +1034,14 @@ class CountMap(BaseModel):
     )
 
 
+class FieldStatistics(BaseModel):
+    avg_value_size_in_bytes: int | None = Field(
+        None,
+        alias='avg-value-size-in-bytes',
+        description='Avg value size in memory (uncompressed) in bytes over non-null values to estimate memory consumption',
+    )
+
+
 class PrimitiveTypeValue(
     RootModel[
         BooleanTypeValue
@@ -1101,6 +1109,11 @@ class ContentFile(BaseModel):
         None, alias='split-offsets', description='List of splittable offsets'
     )
     sort_order_id: int | None = Field(None, alias='sort-order-id')
+    content_stats: dict[str, FieldStatistics] | None = Field(
+        None,
+        alias='content-stats',
+        description='Container struct for per-field metrics structs',
+    )
 
 
 class PositionDeleteFile(ContentFile):

open-api/rest-catalog-open-api.yaml

@@ -5123,6 +5123,20 @@ components:
           "values": [ 100, 200 ]
         }
 
+    FieldStatistics:
+      type: object
+      properties:
+        avg-value-size-in-bytes:
+          type: integer
+          format: int32
+          description:
+            Avg value size in memory (uncompressed) in bytes over non-null
+            values to estimate memory consumption
+      example:
+        {
+          "avg-value-size-in-bytes": 8
+        }
+
     ValueMap:
       type: object
       properties:
@@ -5222,6 +5236,11 @@ components:
           description: "List of splittable offsets"
         sort-order-id:
           type: integer
+        content-stats:
+          type: object
+          additionalProperties:
+            $ref: '#/components/schemas/FieldStatistics'
+          description: Container struct for per-field metrics structs
 
     DataFile:
       allOf: