PARQUET-2249: Add IEEE-754 total order and nan count for floating types

parquet-column/src/main/java/org/apache/parquet/column/statistics/BinaryStatistics.java

@@ -28,8 +28,8 @@ public class BinaryStatistics extends Statistics<Binary> {
   private static final PrimitiveType DEFAULT_FAKE_TYPE =
       Types.optional(PrimitiveType.PrimitiveTypeName.BINARY).named("fake_binary_type");
 
-  private Binary max;
-  private Binary min;
+  protected Binary max;
+  protected Binary min;
 
   /**
    * @deprecated will be removed in 2.0.0. Use {@link Statistics#createStats(org.apache.parquet.schema.Type)} instead
@@ -43,7 +43,7 @@ public BinaryStatistics() {
     super(type);
   }
 
-  private BinaryStatistics(BinaryStatistics other) {
+  protected BinaryStatistics(BinaryStatistics other) {
     super(other.type());
     if (other.hasNonNullValue()) {
       initializeStats(other.min, other.max);
@@ -57,10 +57,12 @@ public void updateStats(Binary value) {
       min = value.copy();
       max = value.copy();
       this.markAsNotEmpty();
-    } else if (comparator().compare(min, value) > 0) {
-      min = value.copy();
-    } else if (comparator().compare(max, value) < 0) {
-      max = value.copy();
+    } else {
+      if (comparator().compare(min, value) > 0) {
+        min = value.copy();
+      } else if (comparator().compare(max, value) < 0) {
+        max = value.copy();
+      }
     }
   }
 

parquet-column/src/main/java/org/apache/parquet/column/statistics/DoubleStatistics.java

@@ -28,8 +28,8 @@ public class DoubleStatistics extends Statistics<Double> {
   private static final PrimitiveType DEFAULT_FAKE_TYPE =
       Types.optional(PrimitiveType.PrimitiveTypeName.DOUBLE).named("fake_double_type");
 
-  private double max;
-  private double min;
+  protected double max;
+  protected double min;
 
   /**
    * @deprecated will be removed in 2.0.0. Use {@link Statistics#createStats(org.apache.parquet.schema.Type)} instead
@@ -41,22 +41,28 @@ public DoubleStatistics() {
 
   DoubleStatistics(PrimitiveType type) {
     super(type);
+    incrementNanCount(0);
   }
 
-  private DoubleStatistics(DoubleStatistics other) {
+  protected DoubleStatistics(DoubleStatistics other) {
     super(other.type());
     if (other.hasNonNullValue()) {
       initializeStats(other.min, other.max);
     }
     setNumNulls(other.getNumNulls());
+    incrementNanCount(other.getNanCount());
   }
 
   @Override
   public void updateStats(double value) {
+    if (Double.isNaN(value)) {
+      incrementNanCount();
+      return;
+    }
     if (!this.hasNonNullValue()) {
-      initializeStats(value, value);
+      initializeStats(normalizeMinValue(value), normalizeMaxValue(value));
     } else {
-      updateStats(value, value);
+      updateStats(normalizeMinValue(value), normalizeMaxValue(value));
     }
   }
 
@@ -79,12 +85,12 @@ public void setMinMaxFromBytes(byte[] minBytes, byte[] maxBytes) {
 
   @Override
   public byte[] getMaxBytes() {
-    return BytesUtils.longToBytes(Double.doubleToLongBits(max));
+    return BytesUtils.longToBytes(Double.doubleToRawLongBits(max));
   }
 
   @Override
   public byte[] getMinBytes() {
-    return BytesUtils.longToBytes(Double.doubleToLongBits(min));
+    return BytesUtils.longToBytes(Double.doubleToRawLongBits(min));
   }
 
   @Override
@@ -98,6 +104,19 @@ public boolean isSmallerThan(long size) {
   }
 
   public void updateStats(double min_value, double max_value) {
+    if (Double.isNaN(min_value) && Double.isNaN(max_value)) {
+      return;
+    }
+    if (Double.isNaN(min_value)) {
+      min_value = max_value;
+    }
+    if (Double.isNaN(max_value)) {
+      max_value = min_value;
+    }
+
+    min_value = normalizeMinValue(min_value);
+    max_value = normalizeMaxValue(max_value);
+
     if (comparator().compare(min, min_value) > 0) {
       min = min_value;
     }
@@ -144,6 +163,14 @@ public void setMinMax(double min, double max) {
     this.markAsNotEmpty();
   }
 
+  private static double normalizeMinValue(double value) {
+    return value == 0.0 ? -0.0 : value;
+  }
+
+  private static double normalizeMaxValue(double value) {
+    return value == 0.0 ? 0.0 : value;
+  }
+
   @Override
   public DoubleStatistics copy() {
     return new DoubleStatistics(this);

parquet-column/src/main/java/org/apache/parquet/column/statistics/Float16Statistics.java

@@ -0,0 +1,92 @@
+/*
+ * 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.parquet.column.statistics;
+
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.Float16;
+import org.apache.parquet.schema.PrimitiveType;
+
+public class Float16Statistics extends BinaryStatistics {
+
+  Float16Statistics(PrimitiveType type) {
+    super(type);
+    incrementNanCount(0);
+  }
+
+  private Float16Statistics(Float16Statistics other) {
+    super(other);
+    incrementNanCount(other.getNanCount());
+  }
+
+  @Override
+  public void updateStats(Binary value) {
+    if (Float16.isNaN(value.get2BytesLittleEndian())) {
+      incrementNanCount();
+      return;
+    }
+    if (!this.hasNonNullValue()) {
+      initializeStats(normalizeMinValue(value), normalizeMaxValue(value));
+    } else {
+      updateStats(normalizeMinValue(value), normalizeMaxValue(value));
+    }
+  }
+
+  @Override
+  @Deprecated
+  public void updateStats(Binary min_value, Binary max_value) {
+    boolean minIsNaN = Float16.isNaN(min_value.get2BytesLittleEndian());
+    boolean maxIsNaN = Float16.isNaN(max_value.get2BytesLittleEndian());
+    if (minIsNaN && maxIsNaN) {
+      return;
+    }
+    if (minIsNaN) {
+      min_value = max_value;
+    }
+    if (maxIsNaN) {
+      max_value = min_value;
+    }
+
+    min_value = normalizeMinValue(min_value);
+    max_value = normalizeMaxValue(max_value);
+
+    if (comparator().compare(min, min_value) > 0) {
+      min = min_value.copy();
+    }
+    if (comparator().compare(max, max_value) < 0) {
+      max = max_value.copy();
+    }
+  }
+
+  @Override
+  public Float16Statistics copy() {
+    return new Float16Statistics(this);
+  }
+
+  private Binary normalizeMinValue(Binary value) {
+    return isZero(value) ? Float16.NEGATIVE_ZERO_LITTLE_ENDIAN : value;
+  }
+
+  private Binary normalizeMaxValue(Binary value) {
+    return isZero(value) ? Float16.POSITIVE_ZERO_LITTLE_ENDIAN : value;
+  }
+
+  private static boolean isZero(Binary value) {
+    return (value.get2BytesLittleEndian() & 0x7fff) == 0;
+  }
+}

parquet-column/src/main/java/org/apache/parquet/column/statistics/FloatStatistics.java

@@ -28,8 +28,8 @@ public class FloatStatistics extends Statistics<Float> {
   private static final PrimitiveType DEFAULT_FAKE_TYPE =
       Types.optional(PrimitiveType.PrimitiveTypeName.FLOAT).named("fake_float_type");
 
-  private float max;
-  private float min;
+  protected float max;
+  protected float min;
 
   /**
    * @deprecated will be removed in 2.0.0. Use {@link Statistics#createStats(org.apache.parquet.schema.Type)} instead
@@ -42,22 +42,28 @@ public FloatStatistics() {
 
   FloatStatistics(PrimitiveType type) {
     super(type);
+    incrementNanCount(0);
   }
 
-  private FloatStatistics(FloatStatistics other) {
+  protected FloatStatistics(FloatStatistics other) {
     super(other.type());
     if (other.hasNonNullValue()) {
       initializeStats(other.min, other.max);
     }
     setNumNulls(other.getNumNulls());
+    incrementNanCount(other.getNanCount());
   }
 
   @Override
   public void updateStats(float value) {
+    if (Float.isNaN(value)) {
+      incrementNanCount();
+      return;
+    }
     if (!this.hasNonNullValue()) {
-      initializeStats(value, value);
+      initializeStats(normalizeMinValue(value), normalizeMaxValue(value));
     } else {
-      updateStats(value, value);
+      updateStats(normalizeMinValue(value), normalizeMaxValue(value));
     }
   }
 
@@ -80,12 +86,12 @@ public void setMinMaxFromBytes(byte[] minBytes, byte[] maxBytes) {
 
   @Override
   public byte[] getMaxBytes() {
-    return BytesUtils.intToBytes(Float.floatToIntBits(max));
+    return BytesUtils.intToBytes(Float.floatToRawIntBits(max));
   }
 
   @Override
   public byte[] getMinBytes() {
-    return BytesUtils.intToBytes(Float.floatToIntBits(min));
+    return BytesUtils.intToBytes(Float.floatToRawIntBits(min));
   }
 
   @Override
@@ -99,6 +105,19 @@ public boolean isSmallerThan(long size) {
   }
 
   public void updateStats(float min_value, float max_value) {
+    if (Float.isNaN(min_value) && Float.isNaN(max_value)) {
+      return;
+    }
+    if (Float.isNaN(min_value)) {
+      min_value = max_value;
+    }
+    if (Float.isNaN(max_value)) {
+      max_value = min_value;
+    }
+
+    min_value = normalizeMinValue(min_value);
+    max_value = normalizeMaxValue(max_value);
+
     if (comparator().compare(min, min_value) > 0) {
       min = min_value;
     }
@@ -145,6 +164,14 @@ public void setMinMax(float min, float max) {
     this.markAsNotEmpty();
   }
 
+  private static float normalizeMinValue(float value) {
+    return value == 0.0f ? -0.0f : value;
+  }
+
+  private static float normalizeMaxValue(float value) {
+    return value == 0.0f ? 0.0f : value;
+  }
+
   @Override
   public FloatStatistics copy() {
     return new FloatStatistics(this);

parquet-column/src/main/java/org/apache/parquet/column/statistics/IEEE754DoubleStatistics.java

@@ -0,0 +1,72 @@
+/*
+ * 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.parquet.column.statistics;
+
+import org.apache.parquet.schema.PrimitiveType;
+
+public class IEEE754DoubleStatistics extends DoubleStatistics {
+
+  IEEE754DoubleStatistics(PrimitiveType type) {
+    super(type);
+  }
+
+  private IEEE754DoubleStatistics(IEEE754DoubleStatistics other) {
+    super(other);
+  }
+
+  @Override
+  public void updateStats(double value) {
+    if (Double.isNaN(value)) {
+      incrementNanCount();
+    }
+    if (!this.hasNonNullValue()) {
+      initializeStats(value, value);
+    } else {
+      updateStats(value, value);
+    }
+  }
+
+  @Override
+  public void updateStats(double min_value, double max_value) {
+    boolean minValueIsNaN = Double.isNaN(min_value);
+    boolean minIsNaN = Double.isNaN(min);
+    if (minValueIsNaN) {
+      if (minIsNaN && comparator().compare(min, min_value) > 0) {
+        min = min_value;
+      }
+    } else if (minIsNaN || comparator().compare(min, min_value) > 0) {
+      min = min_value;
+    }
+
+    boolean maxValueIsNaN = Double.isNaN(max_value);
+    boolean maxIsNaN = Double.isNaN(max);
+    if (maxValueIsNaN) {
+      if (maxIsNaN && comparator().compare(max, max_value) < 0) {
+        max = max_value;
+      }
+    } else if (maxIsNaN || comparator().compare(max, max_value) < 0) {
+      max = max_value;
+    }
+  }
+
+  @Override
+  public IEEE754DoubleStatistics copy() {
+    return new IEEE754DoubleStatistics(this);
+  }
+}