Parquet: Track avg_value_size_in_bytes for Geo types

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

@@ -29,6 +29,7 @@ public class FieldMetrics<T> {
   private final T lowerBound;
   private final T upperBound;
   private final Type originalType;
+  private final Integer avgValueSizeInBytes;
 
   public FieldMetrics(int id, long valueCount, long nullValueCount) {
     this(id, valueCount, nullValueCount, -1L, null, null, null);
@@ -65,13 +66,26 @@ public FieldMetrics(
       T lowerBound,
       T upperBound,
       Type originalType) {
+    this(id, valueCount, nullValueCount, nanValueCount, lowerBound, upperBound, originalType, null);
+  }
+
+  public FieldMetrics(
+      int id,
+      long valueCount,
+      long nullValueCount,
+      long nanValueCount,
+      T lowerBound,
+      T upperBound,
+      Type originalType,
+      Integer avgValueSizeInBytes) {
     this.id = id;
     this.valueCount = valueCount;
     this.nullValueCount = nullValueCount;
     this.nanValueCount = nanValueCount;
     this.lowerBound = lowerBound;
     this.upperBound = upperBound;
     this.originalType = originalType;
+    this.avgValueSizeInBytes = avgValueSizeInBytes;
   }
 
   /** Returns the id of the field that the metrics within this class are associated with. */
@@ -112,6 +126,11 @@ public Type originalType() {
     return originalType;
   }
 
+  /** Returns the average size in bytes over non-null values, or null if it is not known. */
+  public Integer avgValueSizeInBytes() {
+    return avgValueSizeInBytes;
+  }
+
   /** Returns if the metrics has bounds (i.e. there is at least non-null value for this field) */
   public boolean hasBounds() {
     return upperBound != null;

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

@@ -98,7 +98,7 @@ void fromFieldMetrics(FieldMetrics<T> fieldMetrics) {
     this.valueCount = fieldMetrics.valueCount();
     this.nullValueCount = fieldMetrics.nullValueCount() < 0 ? null : fieldMetrics.nullValueCount();
     this.nanValueCount = fieldMetrics.nanValueCount() < 0 ? null : fieldMetrics.nanValueCount();
-    this.avgValueSize = null;
+    this.avgValueSize = fieldMetrics.avgValueSizeInBytes();
   }
 
   private boolean isBinary() {

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

@@ -0,0 +1,51 @@
+/*
+ * 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 org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+
+/** Field-level metrics for tracking the average size of variable-length values. */
+public class ValueSizeFieldMetrics extends FieldMetrics<Object> {
+
+  private ValueSizeFieldMetrics(int id, long valueCount, Integer avgValueSizeInBytes) {
+    super(id, valueCount, 0L, -1L, null, null, null, avgValueSizeInBytes);
+  }
+
+  public static class Builder {
+    private final int id;
+    private long valueCount = 0;
+    private long totalValueSizeInBytes = 0;
+
+    public Builder(int id) {
+      this.id = id;
+    }
+
+    public void addValueSize(int sizeInBytes) {
+      Preconditions.checkArgument(sizeInBytes >= 0, "Invalid value size: %s", sizeInBytes);
+      this.valueCount += 1;
+      this.totalValueSizeInBytes += sizeInBytes;
+    }
+
+    public ValueSizeFieldMetrics build() {
+      Integer avgValueSizeInBytes =
+          valueCount > 0 ? Math.toIntExact(totalValueSizeInBytes / valueCount) : null;
+      return new ValueSizeFieldMetrics(id, valueCount, avgValueSizeInBytes);
+    }
+  }
+}

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

@@ -150,14 +150,14 @@ public void testFromFieldMetricsWrongField() {
   public void testFromFieldMetricsString() {
     FieldStatsStruct<String> stats = new FieldStatsStruct<>(STRING_STATS);
 
-    stats.fromFieldMetrics(new FieldMetrics<>(100, 28, 2, "a", "z"));
+    stats.fromFieldMetrics(new FieldMetrics<>(100, 28, 2, -1, "a", "z", null, 4));
 
     assertThat(stats.lowerBound()).isEqualTo("a");
     assertThat(stats.upperBound()).isEqualTo("z");
     assertThat(stats.tightBounds()).isFalse();
     assertThat(stats.valueCount()).isEqualTo(28L);
     assertThat(stats.nullValueCount()).isEqualTo(2L);
-    assertThat(stats.avgValueSizeInBytes()).isNull(); // unknown
+    assertThat(stats.avgValueSizeInBytes()).isEqualTo(4);
   }
 
   @Test

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

@@ -0,0 +1,59 @@
+/*
+ * 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 static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import org.junit.jupiter.api.Test;
+
+class TestValueSizeFieldMetrics {
+
+  @Test
+  void averageValueSize() {
+    ValueSizeFieldMetrics.Builder builder = new ValueSizeFieldMetrics.Builder(2);
+    builder.addValueSize(21);
+    builder.addValueSize(42);
+
+    FieldMetrics<?> metrics = builder.build();
+
+    assertThat(metrics.id()).isEqualTo(2);
+    assertThat(metrics.valueCount()).isEqualTo(2);
+    assertThat(metrics.nullValueCount()).isZero();
+    assertThat(metrics.nanValueCount()).isEqualTo(-1);
+    assertThat(metrics.avgValueSizeInBytes()).isEqualTo(31);
+  }
+
+  @Test
+  void noValues() {
+    FieldMetrics<?> metrics = new ValueSizeFieldMetrics.Builder(2).build();
+
+    assertThat(metrics.valueCount()).isZero();
+    assertThat(metrics.avgValueSizeInBytes()).isNull();
+  }
+
+  @Test
+  void rejectsNegativeValueSize() {
+    ValueSizeFieldMetrics.Builder builder = new ValueSizeFieldMetrics.Builder(2);
+
+    assertThatThrownBy(() -> builder.addValueSize(-1))
+        .isInstanceOf(IllegalArgumentException.class)
+        .hasMessage("Invalid value size: -1");
+  }
+}

parquet/src/main/java/org/apache/iceberg/data/parquet/BaseParquetWriter.java

@@ -269,14 +269,14 @@ public Optional<ParquetValueWriter<?>> visit(
     public Optional<ParquetValueWriter<?>> visit(
         LogicalTypeAnnotation.GeometryLogicalTypeAnnotation geometryType) {
       // geometry values are pure WKB stored in a BINARY column
-      return Optional.of(ParquetValueWriters.byteBuffers(desc));
+      return Optional.of(ParquetValueWriters.geospatial(desc));
     }
 
     @Override
     public Optional<ParquetValueWriter<?>> visit(
         LogicalTypeAnnotation.GeographyLogicalTypeAnnotation geographyType) {
       // geography values are pure WKB stored in a BINARY column
-      return Optional.of(ParquetValueWriters.byteBuffers(desc));
+      return Optional.of(ParquetValueWriters.geospatial(desc));
     }
 
     @Override

parquet/src/main/java/org/apache/iceberg/parquet/ParquetMetrics.java

@@ -264,7 +264,11 @@ private <T> FieldMetrics<T> metricsFromFieldMetrics(
             fieldMetrics.id(),
             fieldMetrics.valueCount(),
             fieldMetrics.nullValueCount(),
-            fieldMetrics.nanValueCount());
+            fieldMetrics.nanValueCount(),
+            null,
+            null,
+            null,
+            fieldMetrics.avgValueSizeInBytes());
       } else {
         T lowerBound = truncateLowerBound(icebergType, fieldMetrics.lowerBound(), truncateLength);
         T upperBound = truncateUpperBound(icebergType, fieldMetrics.upperBound(), truncateLength);
@@ -275,7 +279,8 @@ private <T> FieldMetrics<T> metricsFromFieldMetrics(
             fieldMetrics.nanValueCount(),
             lowerBound,
             upperBound,
-            icebergType);
+            icebergType,
+            fieldMetrics.avgValueSizeInBytes());
       }
     }
 

parquet/src/main/java/org/apache/iceberg/parquet/ParquetValueWriters.java

@@ -37,6 +37,7 @@
 import org.apache.iceberg.FieldMetrics;
 import org.apache.iceberg.FloatFieldMetrics;
 import org.apache.iceberg.StructLike;
+import org.apache.iceberg.ValueSizeFieldMetrics;
 import org.apache.iceberg.deletes.PositionDelete;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -120,6 +121,10 @@ public static PrimitiveWriter<ByteBuffer> byteBuffers(ColumnDescriptor desc) {
     return new BytesWriter(desc);
   }
 
+  public static PrimitiveWriter<ByteBuffer> geospatial(ColumnDescriptor desc) {
+    return new GeospatialWriter(desc);
+  }
+
   public static PrimitiveWriter<ByteBuffer> fixedBuffers(ColumnDescriptor desc) {
     return new FixedBufferWriter(desc);
   }
@@ -336,6 +341,27 @@ public void write(int repetitionLevel, ByteBuffer buffer) {
     }
   }
 
+  private static class GeospatialWriter extends PrimitiveWriter<ByteBuffer> {
+    private final ValueSizeFieldMetrics.Builder metricsBuilder;
+
+    private GeospatialWriter(ColumnDescriptor desc) {
+      super(desc);
+      this.metricsBuilder =
+          new ValueSizeFieldMetrics.Builder(desc.getPrimitiveType().getId().intValue());
+    }
+
+    @Override
+    public void write(int repetitionLevel, ByteBuffer buffer) {
+      metricsBuilder.addValueSize(buffer.remaining());
+      column.writeBinary(repetitionLevel, Binary.fromReusedByteBuffer(buffer));
+    }
+
+    @Override
+    public Stream<FieldMetrics<?>> metrics() {
+      return Stream.of(metricsBuilder.build());
+    }
+  }
+
   private static class FixedBufferWriter extends PrimitiveWriter<ByteBuffer> {
     private final int length;
 
@@ -457,7 +483,8 @@ public Stream<FieldMetrics<?>> metrics() {
                   metrics.nanValueCount(),
                   metrics.lowerBound(),
                   metrics.upperBound(),
-                  metrics.originalType()));
+                  metrics.originalType(),
+                  metrics.avgValueSizeInBytes()));
         } else {
           throw new IllegalStateException(
               String.format(

parquet/src/test/java/org/apache/iceberg/parquet/TestParquetValueWriters.java

@@ -0,0 +1,63 @@
+/*
+ * 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.parquet;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.nio.ByteBuffer;
+import org.apache.iceberg.FieldMetrics;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.ColumnWriteStore;
+import org.apache.parquet.column.ColumnWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+import org.junit.jupiter.api.Test;
+
+class TestParquetValueWriters {
+
+  @Test
+  void geospatialValueSizeMetricsExcludeNulls() {
+    Schema schema = new Schema(optional(2, "geom", Types.GeometryType.crs84()));
+    MessageType parquetSchema = ParquetSchemaUtil.convert(schema, "table");
+    Type parquetType = parquetSchema.getType("geom");
+    ColumnDescriptor desc = parquetSchema.getColumnDescription(new String[] {"geom"});
+    ParquetValueWriter<ByteBuffer> writer =
+        ParquetValueWriters.option(
+            parquetType,
+            parquetSchema.getMaxDefinitionLevel(new String[] {"geom"}),
+            ParquetValueWriters.geospatial(desc));
+
+    ColumnWriteStore columnStore = mock(ColumnWriteStore.class);
+    when(columnStore.getColumnWriter(desc)).thenReturn(mock(ColumnWriter.class));
+    writer.setColumnStore(columnStore);
+    writer.write(0, ByteBuffer.allocate(21));
+    writer.write(0, ByteBuffer.allocate(42));
+    writer.write(0, null);
+
+    FieldMetrics<?> metrics = writer.metrics().findFirst().orElseThrow();
+    assertThat(metrics.valueCount()).isEqualTo(3);
+    assertThat(metrics.nullValueCount()).isEqualTo(1);
+    assertThat(metrics.avgValueSizeInBytes()).isEqualTo(31);
+  }
+}