Add NaN counter to Metrics and implement in Parquet writers

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

@@ -0,0 +1,91 @@
+/*
+ * 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 java.nio.ByteBuffer;
+
+/**
+ * Iceberg internally tracked field level metrics.
+ */
+public class FieldMetrics {
+  private final int id;
+  private final long valueCount;
+  private final long nullValueCount;
+  private final long nanValueCount;
+  private final ByteBuffer lowerBound;
+  private final ByteBuffer upperBound;
+
+  public FieldMetrics(int id,
+                      long valueCount,
+                      long nullValueCount,
+                      long nanValueCount,
+                      ByteBuffer lowerBound,
+                      ByteBuffer upperBound) {
+    this.id = id;
+    this.valueCount = valueCount;
+    this.nullValueCount = nullValueCount;
+    this.nanValueCount = nanValueCount;
+    this.lowerBound = lowerBound;
+    this.upperBound = upperBound;
+  }
+
+  /**
+   * Returns the id of the field that the metrics within this class are associated with.
+   */
+  public int id() {
+    return id;
+  }
+
+  /**
+   * Returns the number of all values, including nulls, NaN and repeated, for the given field.
+   */
+  public long valueCount() {
+    return valueCount;
+  }
+
+  /**
+   * Returns the number of null values for this field.
+   */
+  public long nullValueCount() {
+    return nullValueCount;
+  }
+
+  /**
+   * Returns the number of NaN values for this field. Will only be non-0 if this field is a double or float field.
+   */
+  public long nanValueCount() {
+    return nanValueCount;
+  }
+
+  /**
+   * Returns the lower bound value of this field.
+   */
+  public ByteBuffer lowerBound() {
+    return lowerBound;
+  }
+
+  /**
+   * Returns the upper bound value of this field.
+   */
+  public ByteBuffer upperBound() {
+    return upperBound;
+  }
+}

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

@@ -37,12 +37,17 @@ public class Metrics implements Serializable {
   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, ByteBuffer> lowerBounds = null;
   private Map<Integer, ByteBuffer> upperBounds = null;
 
   public Metrics() {
   }
 
+  /**
+   * @deprecated will be removed in 0.12.0; use {@link #Metrics(Long, Map, Map, Map, Map)} instead.
+   */
+  @Deprecated
   public Metrics(Long rowCount,
                  Map<Integer, Long> columnSizes,
                  Map<Integer, Long> valueCounts,
@@ -53,6 +58,22 @@ public Metrics(Long rowCount,
     this.nullValueCounts = nullValueCounts;
   }
 
+  public Metrics(Long rowCount,
+                 Map<Integer, Long> columnSizes,
+                 Map<Integer, Long> valueCounts,
+                 Map<Integer, Long> nullValueCounts,
+                 Map<Integer, Long> nanValueCounts) {
+    this.rowCount = rowCount;
+    this.columnSizes = columnSizes;
+    this.valueCounts = valueCounts;
+    this.nullValueCounts = nullValueCounts;
+    this.nanValueCounts = nanValueCounts;
+  }
+
+  /**
+   * @deprecated will be removed in 0.12.0; use {@link #Metrics(Long, Map, Map, Map, Map, Map, Map)} instead.
+   */
+  @Deprecated
   public Metrics(Long rowCount,
                  Map<Integer, Long> columnSizes,
                  Map<Integer, Long> valueCounts,
@@ -67,6 +88,22 @@ public Metrics(Long rowCount,
     this.upperBounds = upperBounds;
   }
 
+  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) {
+    this.rowCount = rowCount;
+    this.columnSizes = columnSizes;
+    this.valueCounts = valueCounts;
+    this.nullValueCounts = nullValueCounts;
+    this.nanValueCounts = nanValueCounts;
+    this.lowerBounds = lowerBounds;
+    this.upperBounds = upperBounds;
+  }
+
   /**
    * Get the number of records (rows) in file.
    *
@@ -103,6 +140,15 @@ public Map<Integer, Long> nullValueCounts() {
     return nullValueCounts;
   }
 
+  /**
+   * Get the number of NaN values for all float and double fields in a file.
+   *
+   * @return a Map of fieldId to the number of NaN counts
+   */
+  public Map<Integer, Long> nanValueCounts() {
+    return nanValueCounts;
+  }
+
   /**
    * Get the non-null lower bound values for all fields in a file.
    *
@@ -136,6 +182,7 @@ private void writeObject(ObjectOutputStream out) throws IOException {
     out.writeObject(columnSizes);
     out.writeObject(valueCounts);
     out.writeObject(nullValueCounts);
+    out.writeObject(nanValueCounts);
 
     writeByteBufferMap(out, lowerBounds);
     writeByteBufferMap(out, upperBounds);
@@ -169,6 +216,7 @@ private void readObject(ObjectInputStream in) throws IOException, ClassNotFoundE
     columnSizes = (Map<Integer, Long>) in.readObject();
     valueCounts = (Map<Integer, Long>) in.readObject();
     nullValueCounts = (Map<Integer, Long>) in.readObject();
+    nanValueCounts = (Map<Integer, Long>) in.readObject();
 
     lowerBounds = readByteBufferMap(in);
     upperBounds = readByteBufferMap(in);