Core: Fix float and double metrics for Parquet and ORC (#2464)

This replaces the metrics from Parquet and ORC with metrics that are accumulated by Iceberg writers to ensure that the metrics do not include NaN values.

Author: yyanyy

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

@@ -0,0 +1,66 @@
+/*
+ * 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;
+
+/**
+ * Iceberg internally tracked field level metrics, used by Parquet and ORC writers only.
+ * <p>
+ * Parquet/ORC keeps track of most metrics in file statistics, and only NaN counter is actually tracked by writers.
+ * This wrapper ensures that metrics not being updated by those writers will not be incorrectly used, by throwing
+ * exceptions when they are accessed.
+ */
+public class DoubleFieldMetrics extends FieldMetrics<Double> {
+
+  private DoubleFieldMetrics(int id, long valueCount, long nanValueCount, Double lowerBound, Double upperBound) {
+    super(id, valueCount, 0L, nanValueCount, lowerBound, upperBound);
+  }
+
+  public static class Builder {
+    private final int id;
+    private long valueCount = 0;
+    private long nanValueCount = 0;
+    private double lowerBound = Double.POSITIVE_INFINITY;
+    private double upperBound = Double.NEGATIVE_INFINITY;
+
+    public Builder(int id) {
+      this.id = id;
+    }
+
+    public void addValue(double value) {
+      this.valueCount++;
+      if (Double.isNaN(value)) {
+        this.nanValueCount++;
+      } else {
+        if (Double.compare(value, lowerBound) < 0) {
+          this.lowerBound = value;
+        }
+        if (Double.compare(value, upperBound) > 0) {
+          this.upperBound = value;
+        }
+      }
+    }
+
+    public DoubleFieldMetrics build() {
+      boolean hasBound = valueCount - nanValueCount > 0;
+      return new DoubleFieldMetrics(id, valueCount, nanValueCount,
+          hasBound ? lowerBound : null, hasBound ? upperBound : null);
+    }
+  }
+}

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

@@ -20,25 +20,23 @@
 package org.apache.iceberg;
 
 
-import java.nio.ByteBuffer;
-
 /**
  * Iceberg internally tracked field level metrics.
  */
-public class FieldMetrics {
+public class FieldMetrics<T> {
   private final int id;
   private final long valueCount;
   private final long nullValueCount;
   private final long nanValueCount;
-  private final ByteBuffer lowerBound;
-  private final ByteBuffer upperBound;
+  private final T lowerBound;
+  private final T upperBound;
 
   public FieldMetrics(int id,
                       long valueCount,
                       long nullValueCount,
                       long nanValueCount,
-                      ByteBuffer lowerBound,
-                      ByteBuffer upperBound) {
+                      T lowerBound,
+                      T upperBound) {
     this.id = id;
     this.valueCount = valueCount;
     this.nullValueCount = nullValueCount;
@@ -78,14 +76,21 @@ public long nanValueCount() {
   /**
    * Returns the lower bound value of this field.
    */
-  public ByteBuffer lowerBound() {
+  public T lowerBound() {
     return lowerBound;
   }
 
   /**
    * Returns the upper bound value of this field.
    */
-  public ByteBuffer upperBound() {
+  public T upperBound() {
     return upperBound;
   }
+
+  /**
+   * 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/FloatFieldMetrics.java

@@ -19,44 +19,52 @@
 
 package org.apache.iceberg;
 
-import java.nio.ByteBuffer;
-
 /**
  * Iceberg internally tracked field level metrics, used by Parquet and ORC writers only.
  * <p>
  * Parquet/ORC keeps track of most metrics in file statistics, and only NaN counter is actually tracked by writers.
  * This wrapper ensures that metrics not being updated by those writers will not be incorrectly used, by throwing
  * exceptions when they are accessed.
  */
-public class FloatFieldMetrics extends FieldMetrics {
-
-  /**
-   * Constructor for creating a FieldMetrics with only NaN counter.
-   * @param id field id being tracked by the writer
-   * @param nanValueCount number of NaN values, will only be non-0 for double or float field.
-   */
-  public FloatFieldMetrics(int id,
-                           long nanValueCount) {
-    super(id, 0L, 0L, nanValueCount, null, null);
-  }
+public class FloatFieldMetrics extends FieldMetrics<Float> {
 
-  @Override
-  public long valueCount() {
-    throw new IllegalStateException("Shouldn't access this method, as this metric is tracked in file statistics. ");
+  private FloatFieldMetrics(int id, long valueCount, long nanValueCount, Float lowerBound, Float upperBound) {
+    super(id, valueCount, 0L, nanValueCount, lowerBound, upperBound);
   }
 
-  @Override
-  public long nullValueCount() {
-    throw new IllegalStateException("Shouldn't access this method, as this metric is tracked in file statistics. ");
+  public Builder builderFor(int id) {
+    return new Builder(id);
   }
 
-  @Override
-  public ByteBuffer lowerBound() {
-    throw new IllegalStateException("Shouldn't access this method, as this metric is tracked in file statistics. ");
-  }
+  public static class Builder {
+    private final int id;
+    private long valueCount = 0;
+    private long nanValueCount = 0;
+    private float lowerBound = Float.POSITIVE_INFINITY;
+    private float upperBound = Float.NEGATIVE_INFINITY;
+
+    public Builder(int id) {
+      this.id = id;
+    }
+
+    public void addValue(float value) {
+      this.valueCount++;
+      if (Float.isNaN(value)) {
+        this.nanValueCount++;
+      } else {
+        if (Float.compare(value, lowerBound) < 0) {
+          this.lowerBound = value;
+        }
+        if (Float.compare(value, upperBound) > 0) {
+          this.upperBound = value;
+        }
+      }
+    }
 
-  @Override
-  public ByteBuffer upperBound() {
-    throw new IllegalStateException("Shouldn't access this method, as this metric is tracked in file statistics. ");
+    public FloatFieldMetrics build() {
+      boolean hasBound = valueCount - nanValueCount > 0;
+      return new FloatFieldMetrics(id, valueCount, nanValueCount,
+          hasBound ? lowerBound : null, hasBound ? upperBound : null);
+    }
   }
 }

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

@@ -34,7 +34,7 @@ private MetricsUtil() {
    * Construct mapping relationship between column id to NaN value counts from input metrics and metrics config.
    */
   public static Map<Integer, Long> createNanValueCounts(
-      Stream<FieldMetrics> fieldMetrics, MetricsConfig metricsConfig, Schema inputSchema) {
+      Stream<FieldMetrics<?>> fieldMetrics, MetricsConfig metricsConfig, Schema inputSchema) {
     Preconditions.checkNotNull(metricsConfig, "metricsConfig is required");
 
     if (fieldMetrics == null || inputSchema == null) {

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

@@ -274,7 +274,7 @@ public void testMetricsForNestedStructFields() throws IOException {
     assertBounds(6, BinaryType.get(),
         ByteBuffer.wrap("A".getBytes()), ByteBuffer.wrap("A".getBytes()), metrics);
     assertCounts(7, 1L, 0L, 1L, metrics);
-    assertBounds(7, DoubleType.get(), Double.NaN, Double.NaN, metrics);
+    assertBounds(7, DoubleType.get(), null, null, metrics);
   }
 
   private Record buildNestedTestRecord() {
@@ -354,9 +354,9 @@ public void testMetricsForNaNColumns() throws IOException {
     Assert.assertEquals(2L, (long) metrics.recordCount());
     assertCounts(1, 2L, 0L, 2L, metrics);
     assertCounts(2, 2L, 0L, 2L, metrics);
-    // below: current behavior; will be null once NaN is excluded from upper/lower bound
-    assertBounds(1, FloatType.get(), Float.NaN, Float.NaN, metrics);
-    assertBounds(2, DoubleType.get(), Double.NaN, Double.NaN, metrics);
+
+    assertBounds(1, FloatType.get(), null, null, metrics);
+    assertBounds(2, DoubleType.get(), null, null, metrics);
   }
 
   @Test
@@ -367,15 +367,8 @@ public void testColumnBoundsWithNaNValueAtFront() throws IOException {
     assertCounts(1, 3L, 0L, 1L, metrics);
     assertCounts(2, 3L, 0L, 1L, metrics);
 
-    // below: current behavior; will be non-NaN values once NaN is excluded from upper/lower bound. ORC and Parquet's
-    // behaviors differ due to their implementation of comparison being different.
-    if (fileFormat() == FileFormat.ORC) {
-      assertBounds(1, FloatType.get(), Float.NaN, Float.NaN, metrics);
-      assertBounds(2, DoubleType.get(), Double.NaN, Double.NaN, metrics);
-    } else {
-      assertBounds(1, FloatType.get(), 1.2F, Float.NaN, metrics);
-      assertBounds(2, DoubleType.get(), 3.4D, Double.NaN, metrics);
-    }
+    assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
+    assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
   }
 
   @Test
@@ -386,15 +379,8 @@ public void testColumnBoundsWithNaNValueInMiddle() throws IOException {
     assertCounts(1, 3L, 0L, 1L, metrics);
     assertCounts(2, 3L, 0L, 1L, metrics);
 
-    // below: current behavior; will be non-NaN values once NaN is excluded from upper/lower bound. ORC and Parquet's
-    // behaviors differ due to their implementation of comparison being different.
-    if (fileFormat() == FileFormat.ORC) {
-      assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
-      assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
-    } else {
-      assertBounds(1, FloatType.get(), 1.2F, Float.NaN, metrics);
-      assertBounds(2, DoubleType.get(), 3.4D, Double.NaN, metrics);
-    }
+    assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
+    assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
   }
 
   @Test
@@ -405,15 +391,8 @@ public void testColumnBoundsWithNaNValueAtEnd() throws IOException {
     assertCounts(1, 3L, 0L, 1L, metrics);
     assertCounts(2, 3L, 0L, 1L, metrics);
 
-    // below: current behavior; will be non-NaN values once NaN is excluded from upper/lower bound. ORC and Parquet's
-    // behaviors differ due to their implementation of comparison being different.
-    if (fileFormat() == FileFormat.ORC) {
-      assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
-      assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
-    } else {
-      assertBounds(1, FloatType.get(), 1.2F, Float.NaN, metrics);
-      assertBounds(2, DoubleType.get(), 3.4D, Double.NaN, metrics);
-    }
+    assertBounds(1, FloatType.get(), 1.2F, 5.6F, metrics);
+    assertBounds(2, DoubleType.get(), 3.4D, 7.8D, metrics);
   }
 
   @Test
@@ -506,7 +485,7 @@ public void testMetricsForNestedStructFieldsWithMultipleRowGroup() throws IOExce
     assertBounds(6, BinaryType.get(),
         ByteBuffer.wrap("A".getBytes()), ByteBuffer.wrap("A".getBytes()), metrics);
     assertCounts(7, 201L, 0L, 201L, metrics);
-    assertBounds(7, DoubleType.get(), Double.NaN, Double.NaN, metrics);
+    assertBounds(7, DoubleType.get(), null, null, metrics);
   }
 
   @Test
@@ -567,7 +546,7 @@ public void testFullMetricsMode() throws IOException {
     assertBounds(6, Types.BinaryType.get(),
         ByteBuffer.wrap("A".getBytes()), ByteBuffer.wrap("A".getBytes()), metrics);
     assertCounts(7, 1L, 0L, 1L, metrics);
-    assertBounds(7, Types.DoubleType.get(), Double.NaN, Double.NaN, metrics);
+    assertBounds(7, Types.DoubleType.get(), null, null, metrics);
   }
 
   @Test

data/src/main/java/org/apache/iceberg/data/orc/GenericOrcWriter.java

@@ -129,7 +129,7 @@ public void write(Record value, VectorizedRowBatch output) {
   }
 
   @Override
-  public Stream<FieldMetrics> metrics() {
+  public Stream<FieldMetrics<?>> metrics() {
     return writer.metrics();
   }
 
@@ -160,7 +160,7 @@ public void nonNullWrite(int rowId, Record data, ColumnVector output) {
     }
 
     @Override
-    public Stream<FieldMetrics> metrics() {
+    public Stream<FieldMetrics<?>> metrics() {
       return writers.stream().flatMap(OrcValueWriter::metrics);
     }
   }