Core: guard MetricsUtil.nullValueCounts for columns without null_value_count

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

@@ -501,7 +501,10 @@ static Map<Integer, Long> nullValueCounts(ContentStats stats) {
 
     Map<Integer, Long> result = Maps.newHashMap();
     for (FieldStats<?> fs : stats.fieldStats()) {
-      if (fs != null) {
+      // null_value_count is only stored for optional (nullable) columns; skip columns that do not
+      // track it so the primitive nullValueCount() accessor is never called on an absent value.
+      if (fs != null
+          && StatsUtil.tracksStat(fs.type(), fs.fieldId(), StatsUtil.NULL_VALUE_COUNT_OFFSET)) {
         result.put(fs.fieldId(), fs.nullValueCount());
       }
     }

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

@@ -125,6 +125,13 @@ static int statOffset(int statId) {
     return statId % NUM_RESERVED_FIELD_STATS_IDS;
   }
 
+  /**
+   * Returns whether the field stats struct for the given field ID tracks the metric at the offset.
+   */
+  static boolean tracksStat(Types.StructType fieldStatsType, int fieldId, int statOffset) {
+    return fieldStatsType.field(toBaseId(fieldId) + statOffset) != null;
+  }
+
   public static Types.NestedField contentStatsField(Types.StructType contentStats) {
     return optional(146, "content_stats", contentStats);
   }

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

@@ -0,0 +1,85 @@
+/*
+ * 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.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.util.List;
+import java.util.Map;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+public class TestMetricsUtil {
+  private static final Schema SCHEMA =
+      new Schema(
+          required(1, "req", Types.LongType.get()),
+          optional(2, "opt", Types.LongType.get()),
+          optional(3, "dbl", Types.DoubleType.get()));
+  private static final Types.StructType STATS = StatsUtil.statsReadSchema(SCHEMA, List.of(1, 2, 3));
+
+  private static Types.StructType statsType(String name) {
+    return STATS.field(name).type().asStructType();
+  }
+
+  @Test
+  public void testValueCounts() {
+    ContentStatsStruct stats = new ContentStatsStruct(STATS);
+    stats.setStats(1, new FieldStatsStruct<>(statsType("req"), 1L, 5L, false, 10L, 0L, 0L, null));
+    stats.setStats(2, new FieldStatsStruct<>(statsType("opt"), 1L, 5L, false, 20L, 0L, 0L, null));
+    stats.setStats(3, new FieldStatsStruct<>(statsType("dbl"), 1.0, 5.0, false, 30L, 0L, 0L, null));
+
+    // value_count is tracked for every column
+    assertThat(MetricsUtil.valueCounts(stats))
+        .containsOnly(Map.entry(1, 10L), Map.entry(2, 20L), Map.entry(3, 30L));
+  }
+
+  @Test
+  public void testNullValueCountsSkipsColumnsThatDoNotTrackIt() {
+    ContentStatsStruct stats = new ContentStatsStruct(STATS);
+    // A required column has no null_value_count field; a deserialized struct leaves it null, so
+    // nullValueCounts must not call the primitive accessor on it.
+    stats.setStats(1, new FieldStatsStruct<Long>(statsType("req")));
+    stats.setStats(2, new FieldStatsStruct<>(statsType("opt"), 1L, 5L, false, 20L, 3L, 0L, null));
+    stats.setStats(3, new FieldStatsStruct<>(statsType("dbl"), 1.0, 5.0, false, 30L, 7L, 0L, null));
+
+    // required column 1 is skipped; the optional columns are kept
+    assertThat(MetricsUtil.nullValueCounts(stats)).containsOnly(Map.entry(2, 3L), Map.entry(3, 7L));
+  }
+
+  @Test
+  public void testNanValueCountsOnlyForFloatingColumns() {
+    ContentStatsStruct stats = new ContentStatsStruct(STATS);
+    stats.setStats(2, new FieldStatsStruct<>(statsType("opt"), 1L, 5L, false, 20L, 0L, 0L, null));
+    stats.setStats(3, new FieldStatsStruct<>(statsType("dbl"), 1.0, 5.0, false, 30L, 0L, 4L, null));
+
+    // nan_value_count is tracked only for floating-point columns
+    assertThat(MetricsUtil.nanValueCounts(stats)).containsOnly(Map.entry(3, 4L));
+  }
+
+  @Test
+  public void testNullContentStatsReturnsNull() {
+    assertThat(MetricsUtil.valueCounts(null)).isNull();
+    assertThat(MetricsUtil.nullValueCounts(null)).isNull();
+    assertThat(MetricsUtil.nanValueCounts(null)).isNull();
+    assertThat(MetricsUtil.lowerBounds(null)).isNull();
+    assertThat(MetricsUtil.upperBounds(null)).isNull();
+  }
+}

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

@@ -592,4 +592,23 @@ private static void assertSameStructure(Types.StructType expected, Types.StructT
       }
     }
   }
+
+  @Test
+  public void testTracksStat() {
+    Schema schema =
+        new Schema(
+            Types.NestedField.required(1, "req", Types.LongType.get()),
+            Types.NestedField.optional(2, "opt", Types.LongType.get()));
+    Types.StructType stats = StatsUtil.statsReadSchema(schema, List.of(1, 2));
+    Types.StructType reqStats = stats.field("req").type().asStructType();
+    Types.StructType optStats = stats.field("opt").type().asStructType();
+
+    // value_count is tracked for every column
+    assertThat(StatsUtil.tracksStat(reqStats, 1, StatsUtil.VALUE_COUNT_OFFSET)).isTrue();
+    assertThat(StatsUtil.tracksStat(optStats, 2, StatsUtil.VALUE_COUNT_OFFSET)).isTrue();
+
+    // null_value_count is tracked only for optional (nullable) columns
+    assertThat(StatsUtil.tracksStat(reqStats, 1, StatsUtil.NULL_VALUE_COUNT_OFFSET)).isFalse();
+    assertThat(StatsUtil.tracksStat(optStats, 2, StatsUtil.NULL_VALUE_COUNT_OFFSET)).isTrue();
+  }
 }