GH-3242: Emit and Read min/max statistics for int96 timestamp columns

parquet-column/src/main/java/org/apache/parquet/ValidInt96Stats.java

@@ -0,0 +1,78 @@
+/*
+ * 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;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.parquet.VersionParser.ParsedVersion;
+import org.apache.parquet.VersionParser.VersionParseException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Not all parquet writers populate the int96 statistics correctly. For example: arrow-rs
+ * https://github.com/apache/arrow-rs/blob/3ed9aedabc9e5a90170e43ff818f24a29eafb35b/parquet/src/file/statistics.rs#L212-L215
+ * This class is used to detect whether a file was written with a version that has correct int96 statistics.
+ */
+public class ValidInt96Stats {
+  private static final AtomicBoolean alreadyLogged = new AtomicBoolean(false);
+
+  private static final Logger LOG = LoggerFactory.getLogger(ValidInt96Stats.class);
+
+  private static final SemanticVersion MINIMUM_PARQUET_MR_VERSION = new SemanticVersion(1, 16, 0);
+
+  /**
+   * Decides if the statistics from a file created by createdBy (the created_by field from parquet format)
+   * should be trusted for INT96 columns.
+   *
+   * @param createdBy  the created-by string from a file footer
+   * @return true if the statistics are valid and can be trusted, false otherwise
+   */
+  public static boolean hasValidInt96Stats(String createdBy) {
+    if (Strings.isNullOrEmpty(createdBy)) {
+      warnOnce("Cannot verify INT96 statistics because created_by is null or empty");
+      return false;
+    }
+
+    try {
+      ParsedVersion version = VersionParser.parse(createdBy);
+      if ("parquet-mr".equals(version.application)) {
+        return version.hasSemanticVersion()
+            && version.getSemanticVersion().compareTo(MINIMUM_PARQUET_MR_VERSION) >= 0;
+      }
+      if ("parquet-mr compatible Photon".equals(version.application)) {
+        return true;
+      }
+    } catch (RuntimeException | VersionParseException e) {
+      warnParseErrorOnce(createdBy, e);
+    }
+    return false;
+  }
+
+  private static void warnParseErrorOnce(String createdBy, Throwable e) {
+    if (!alreadyLogged.getAndSet(true)) {
+      LOG.warn("Cannot verify INT96 statistics because created_by could not be parsed: " + createdBy, e);
+    }
+  }
+
+  private static void warnOnce(String message) {
+    if (!alreadyLogged.getAndSet(true)) {
+      LOG.warn(message);
+    }
+  }
+}

parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveComparator.java

@@ -206,6 +206,38 @@ public String toString() {
         }
       };
 
+  /*
+   * This comparator is for comparing two timestamps represented as int96 binary.
+   * It is a two level comparison.
+   * Days (last 4 bytes compared as unsigned little endian int32),
+   * Nanoseconds (first 8 bytes compared as unsigned little endian int64)
+   */
+  static final PrimitiveComparator<Binary> BINARY_AS_INT96_TIMESTAMP_COMPARATOR = new BinaryComparator() {
+    @Override
+    int compareBinary(Binary b1, Binary b2) {
+      if (b1.length() != 12 || b2.length() != 12) {
+        throw new IllegalArgumentException(
+            "INT96 binary length must be 12, got " + b1.length() + " and " + b2.length());
+      }
+      ByteBuffer bb1 = b1.toByteBuffer().slice();
+      ByteBuffer bb2 = b2.toByteBuffer().slice();
+      bb1.order(java.nio.ByteOrder.LITTLE_ENDIAN);
+      bb2.order(java.nio.ByteOrder.LITTLE_ENDIAN);
+      int jd1 = bb1.getInt(8);
+      int jd2 = bb2.getInt(8);
+      if (jd1 != jd2) return Integer.compareUnsigned(jd1, jd2) < 0 ? -1 : 1;
+      long s1 = bb1.getLong(0);
+      long s2 = bb2.getLong(0);
+      if (s1 != s2) return Long.compareUnsigned(s1, s2) < 0 ? -1 : 1;
+      return 0;
+    }
+
+    @Override
+    public String toString() {
+      return "BINARY_AS_INT96_TIMESTAMP_COMPARATOR";
+    }
+  };
+
   /*
    * This comparator is for comparing two signed decimal values represented in twos-complement binary. In case of the
    * binary length of one value is shorter than the other it will be padded by the corresponding prefix (0xFF for

parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java

@@ -35,7 +35,6 @@
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.io.api.PrimitiveConverter;
 import org.apache.parquet.io.api.RecordConsumer;
-import org.apache.parquet.schema.ColumnOrder.ColumnOrderName;
 import org.apache.parquet.schema.LogicalTypeAnnotation.UUIDLogicalTypeAnnotation;
 
 /**
@@ -381,7 +380,7 @@ public <T, E extends Exception> T convert(PrimitiveTypeNameConverter<T, E> conve
 
       @Override
       PrimitiveComparator<?> comparator(LogicalTypeAnnotation logicalType) {
-        return PrimitiveComparator.BINARY_AS_SIGNED_INTEGER_COMPARATOR;
+        return PrimitiveComparator.BINARY_AS_INT96_TIMESTAMP_COMPARATOR;
       }
     },
     FIXED_LEN_BYTE_ARRAY("getBinary", Binary.class) {
@@ -566,9 +565,7 @@ public PrimitiveType(
     this.decimalMeta = decimalMeta;
 
     if (columnOrder == null) {
-      columnOrder = primitive == PrimitiveTypeName.INT96 || originalType == OriginalType.INTERVAL
-          ? ColumnOrder.undefined()
-          : ColumnOrder.typeDefined();
+      columnOrder = originalType == OriginalType.INTERVAL ? ColumnOrder.undefined() : ColumnOrder.typeDefined();
     }
     this.columnOrder = requireValidColumnOrder(columnOrder);
   }
@@ -611,21 +608,14 @@ public PrimitiveType(
     }
 
     if (columnOrder == null) {
-      columnOrder = primitive == PrimitiveTypeName.INT96
-              || logicalTypeAnnotation instanceof LogicalTypeAnnotation.IntervalLogicalTypeAnnotation
+      columnOrder = logicalTypeAnnotation instanceof LogicalTypeAnnotation.IntervalLogicalTypeAnnotation
           ? ColumnOrder.undefined()
           : ColumnOrder.typeDefined();
     }
     this.columnOrder = requireValidColumnOrder(columnOrder);
   }
 
   private ColumnOrder requireValidColumnOrder(ColumnOrder columnOrder) {
-    if (primitive == PrimitiveTypeName.INT96) {
-      Preconditions.checkArgument(
-          columnOrder.getColumnOrderName() == ColumnOrderName.UNDEFINED,
-          "The column order %s is not supported by INT96",
-          columnOrder);
-    }
     if (getLogicalTypeAnnotation() != null) {
       Preconditions.checkArgument(
           getLogicalTypeAnnotation().isValidColumnOrder(columnOrder),

parquet-column/src/main/java/org/apache/parquet/schema/Types.java

@@ -414,8 +414,8 @@ public THIS scale(int scale) {
     /**
      * Adds the column order for the primitive type.
      * <p>
-     * In case of not set the default column order is {@link ColumnOrderName#TYPE_DEFINED_ORDER} except the type
-     * {@link PrimitiveTypeName#INT96} and the types annotated by {@link OriginalType#INTERVAL} where the default column
+     * In case of not set the default column order is {@link ColumnOrderName#TYPE_DEFINED_ORDER} except the types
+     * annotated by {@link OriginalType#INTERVAL} where the default column
      * order is {@link ColumnOrderName#UNDEFINED}.
      *
      * @param columnOrder the column order for the primitive type

parquet-column/src/test/java/org/apache/parquet/ValidInt96StatsTest.java

@@ -0,0 +1,71 @@
+/*
+ * 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;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+import org.junit.Test;
+
+public class ValidInt96StatsTest {
+
+  @Test
+  public void testNullAndEmpty() {
+    assertFalse(ValidInt96Stats.hasValidInt96Stats(null));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats(""));
+  }
+
+  @Test
+  public void testParquetMrValid() {
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.16.0"));
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.16.1"));
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 2.0.0"));
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.16.0 (build abcd)"));
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.16.1-SNAPSHOT"));
+  }
+
+  @Test
+  public void testParquetMrInvalid() {
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.15.0"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.15.1"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.15.1-SNAPSHOT"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.16.0-SNAPSHOT"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.12.3"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.14.0"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.12.3 (build abcd)"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.12.3-SNAPSHOT"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.12.3rc1"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("parquet-mr version 1.12.3rc1-SNAPSHOT"));
+  }
+
+  @Test
+  public void testParquetMrCompatiblePhotonValid() {
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr compatible Photon version 1.0.0"));
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr compatible Photon version 1.0.0 (build abcd)"));
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr compatible Photon version 1.0.0-SNAPSHOT"));
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr compatible Photon version 1.0.0rc1"));
+    assertTrue(ValidInt96Stats.hasValidInt96Stats("parquet-mr compatible Photon version 1.0.0rc1-SNAPSHOT"));
+  }
+
+  @Test
+  public void testInvalidApplications() {
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("arrow-rs version 0.1.0"));
+    assertFalse(ValidInt96Stats.hasValidInt96Stats("impala version 1.6.0"));
+  }
+}

parquet-column/src/test/java/org/apache/parquet/internal/column/columnindex/TestBinaryTruncator.java

@@ -32,6 +32,8 @@
 import static org.junit.Assert.assertSame;
 import static org.junit.Assert.assertTrue;
 
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.nio.charset.CharacterCodingException;
 import java.nio.charset.CharsetDecoder;
 import java.nio.charset.CodingErrorAction;
@@ -93,7 +95,7 @@ public void testContractNonStringTypes() {
     testTruncator(
         Types.required(FIXED_LEN_BYTE_ARRAY).length(12).as(INTERVAL).named("test_fixed_interval"), false);
     testTruncator(Types.required(BINARY).as(DECIMAL).precision(10).scale(2).named("test_binary_decimal"), false);
-    testTruncator(Types.required(INT96).named("test_int96"), false);
+    testInt96Truncator(Types.required(INT96).named("test_int96"), false);
   }
 
   @Test
@@ -157,6 +159,21 @@ public void testContractStringTypes() {
     testTruncator(Types.required(FIXED_LEN_BYTE_ARRAY).length(5).named("test_fixed"), true);
   }
 
+  private Binary createInt96Value(long nanoseconds, int julianDay) {
+    return Binary.fromConstantByteArray(ByteBuffer.allocate(12)
+        .order(ByteOrder.LITTLE_ENDIAN)
+        .putLong(nanoseconds)
+        .putInt(julianDay)
+        .array());
+  }
+
+  private void testInt96Truncator(PrimitiveType type, boolean strict) {
+    BinaryTruncator truncator = BinaryTruncator.getTruncator(type);
+    Comparator<Binary> comparator = type.comparator();
+    checkContract(truncator, comparator, createInt96Value(0, 2458849), strict, strict);
+    checkContract(truncator, comparator, createInt96Value(100, 128849), strict, strict);
+  }
+
   private void testTruncator(PrimitiveType type, boolean strict) {
     BinaryTruncator truncator = BinaryTruncator.getTruncator(type);
     Comparator<Binary> comparator = type.comparator();

parquet-column/src/test/java/org/apache/parquet/schema/TestPrimitiveComparator.java

@@ -19,6 +19,7 @@
 package org.apache.parquet.schema;
 
 import static org.apache.parquet.schema.PrimitiveComparator.BINARY_AS_FLOAT16_COMPARATOR;
+import static org.apache.parquet.schema.PrimitiveComparator.BINARY_AS_INT96_TIMESTAMP_COMPARATOR;
 import static org.apache.parquet.schema.PrimitiveComparator.BINARY_AS_SIGNED_INTEGER_COMPARATOR;
 import static org.apache.parquet.schema.PrimitiveComparator.BOOLEAN_COMPARATOR;
 import static org.apache.parquet.schema.PrimitiveComparator.DOUBLE_COMPARATOR;
@@ -33,8 +34,10 @@
 
 import java.math.BigInteger;
 import java.nio.ByteBuffer;
+import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.List;
+import org.apache.parquet.example.data.simple.NanoTime;
 import org.apache.parquet.io.api.Binary;
 import org.junit.Test;
 
@@ -297,6 +300,84 @@ public void testBinaryAsSignedIntegerComparatorWithEquals() {
     }
   }
 
+  private Binary timestampToInt96(String timestamp) {
+    LocalDateTime dt = LocalDateTime.parse(timestamp);
+    long julianDay = dt.toLocalDate().toEpochDay() + 2440588; // Convert to Julian Day
+    long nanos = dt.toLocalTime().toNanoOfDay();
+    return new NanoTime((int) julianDay, nanos).toBinary();
+  }
+
+  @Test
+  public void testInt96Comparator() {
+    Binary[] valuesInAscendingOrder = {
+      timestampToInt96("2020-01-01T00:00:00.000"),
+      timestampToInt96("2020-01-01T10:00:00.000"),
+      timestampToInt96("2020-02-29T23:59:59.999"),
+      timestampToInt96("2020-12-31T23:59:59.999"),
+      timestampToInt96("2021-01-01T00:00:00.000"),
+      timestampToInt96("2023-06-15T12:30:45.500"),
+      timestampToInt96("2024-02-29T15:45:30.750"),
+      timestampToInt96("2024-12-25T07:00:00.000"),
+      timestampToInt96("2025-01-01T00:00:00.000"),
+      timestampToInt96("2025-07-04T20:00:00.000"),
+      timestampToInt96("2025-07-04T20:50:00.000"),
+      timestampToInt96("2025-12-31T23:59:59.999")
+    };
+
+    java.util.function.Function<Binary, Binary>[] perturb = new java.util.function.Function[] {
+      (java.util.function.Function<Binary, Binary>) b -> b,
+      (java.util.function.Function<Binary, Binary>) b -> Binary.fromReusedByteArray(b.getBytes()),
+      (java.util.function.Function<Binary, Binary>) b -> Binary.fromConstantByteArray(b.getBytes()),
+      (java.util.function.Function<Binary, Binary>) b -> {
+        byte[] originalBytes = b.getBytes();
+        byte[] paddedBuffer = new byte[originalBytes.length + 20];
+        int offset = 10;
+        for (int i = 0; i < paddedBuffer.length; i++) {
+          paddedBuffer[i] = (byte) (0xAA + (i % 5));
+        }
+        System.arraycopy(originalBytes, 0, paddedBuffer, offset, originalBytes.length);
+        return Binary.fromReusedByteArray(paddedBuffer, offset, originalBytes.length);
+      }
+    };
+
+    for (int i = 0; i < valuesInAscendingOrder.length; ++i) {
+      for (int j = 0; j < valuesInAscendingOrder.length; ++j) {
+        Binary bi = valuesInAscendingOrder[i];
+        Binary bj = valuesInAscendingOrder[j];
+        for (java.util.function.Function<Binary, Binary> fi : perturb) {
+          for (java.util.function.Function<Binary, Binary> fj : perturb) {
+            Binary perturbedBi = fi.apply(bi);
+            Binary perturbedBj = fj.apply(bj);
+            assertEquals(
+                Integer.compare(i, j),
+                BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(perturbedBi, perturbedBj));
+          }
+        }
+      }
+    }
+  }
+
+  @Test
+  public void testInt96ComparatorRejectsInvalidLengths() {
+    Binary valid = timestampToInt96("2020-01-01T00:00:00.000");
+    int[] invalidLengths = {0, 4, 8, 11, 13, 16};
+    for (int len : invalidLengths) {
+      Binary wrongLength = Binary.fromConstantByteArray(new byte[len]);
+      try {
+        BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(wrongLength, valid);
+        fail("Expected IllegalArgumentException when left operand has length " + len);
+      } catch (IllegalArgumentException expected) {
+        // ok
+      }
+      try {
+        BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(valid, wrongLength);
+        fail("Expected IllegalArgumentException when right operand has length " + len);
+      } catch (IllegalArgumentException expected) {
+        // ok
+      }
+    }
+  }
+
   @Test
   public void testFloat16Comparator() {
     Binary[] valuesInAscendingOrder = {