[GH-3609]: Add new sort order for int96 timestamps

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

@@ -41,11 +41,16 @@ public enum ColumnOrderName {
      * The column order is defined by the IEEE 754 standard.
      */
     IEEE_754_TOTAL_ORDER,
+    /**
+     * Chronological order for INT96 timestamps.
+     */
+    INT96_TIMESTAMP_ORDER
   }
 
   private static final ColumnOrder UNDEFINED_COLUMN_ORDER = new ColumnOrder(ColumnOrderName.UNDEFINED);
   private static final ColumnOrder TYPE_DEFINED_COLUMN_ORDER = new ColumnOrder(ColumnOrderName.TYPE_DEFINED_ORDER);
   private static final ColumnOrder IEEE_754_TOTAL_ORDER = new ColumnOrder(ColumnOrderName.IEEE_754_TOTAL_ORDER);
+  private static final ColumnOrder INT96_TIMESTAMP_COLUMN_ORDER = new ColumnOrder(ColumnOrderName.INT96_TIMESTAMP_ORDER);
 
   /**
    * @return a {@link ColumnOrder} instance representing an undefined order
@@ -71,6 +76,14 @@ public static ColumnOrder ieee754TotalOrder() {
     return IEEE_754_TOTAL_ORDER;
   }
 
+  /**
+   * @return a {@link ColumnOrder} instance representing the chronological order of INT96 timestamps
+   * @see ColumnOrderName#INT96_TIMESTAMP_ORDER
+   */
+  public static ColumnOrder int96TimestampOrder() {
+    return INT96_TIMESTAMP_COLUMN_ORDER;
+  }
+
   private final ColumnOrderName columnOrderName;
 
   private ColumnOrder(ColumnOrderName columnOrderName) {

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

@@ -20,6 +20,7 @@
 
 import java.io.Serializable;
 import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
 import java.util.Comparator;
 import org.apache.parquet.io.api.Binary;
 
@@ -354,4 +355,47 @@ public String toString() {
           return "BINARY_AS_FLOAT16_IEEE_754_TOTAL_ORDER_COMPARATOR";
         }
       };
+
+  /**
+   * Comparator for two timestamps encoded as INT96 (12-byte little-endian) binary.
+   * Layout: first 8 bytes = nanoseconds within the day, last 4 bytes = Julian day.
+   *
+   * Two-level comparison, matching the INT96 timestamp sort order:
+   * 1. Compare the last 4 bytes (Julian day) as a signed little-endian int32.
+   * 2. If equal, compare the first 8 bytes (nanos) as a signed little-endian int64.
+   */
+  static final PrimitiveComparator<Binary> BINARY_AS_INT96_TIMESTAMP_COMPARATOR = new BinaryComparator() {
+    private static final long NANOSECONDS_PER_DAY = 86_400_000_000_000L;
+
+    @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(ByteOrder.LITTLE_ENDIAN);
+      bb2.order(ByteOrder.LITTLE_ENDIAN);
+
+      int result = Integer.compare(bb1.getInt(8), bb2.getInt(8));
+      if (result != 0) return result;
+
+      long nanos1 = bb1.getLong(0);
+      long nanos2 = bb2.getLong(0);
+      if (nanos1 < 0 || nanos1 > NANOSECONDS_PER_DAY) {
+        throw new IllegalArgumentException("Invalid nanos value: " + nanos1);
+      }
+      if (nanos2 < 0 || nanos2 > NANOSECONDS_PER_DAY) {
+        throw new IllegalArgumentException("Invalid nanos value: " + nanos2);
+      }
+      return Long.compare(nanos1, nanos2);
+    }
+
+    @Override
+    public String toString() {
+      return "BINARY_AS_INT96_TIMESTAMP_COMPARATOR";
+    }
+  };
 }

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

@@ -385,7 +385,9 @@ public <T, E extends Exception> T convert(PrimitiveTypeNameConverter<T, E> conve
 
       @Override
       PrimitiveComparator<?> comparator(LogicalTypeAnnotation logicalType, ColumnOrder columnOrder) {
-        return PrimitiveComparator.BINARY_AS_SIGNED_INTEGER_COMPARATOR;
+        return columnOrder != null && columnOrder.getColumnOrderName() == ColumnOrderName.INT96_TIMESTAMP_ORDER
+            ? PrimitiveComparator.BINARY_AS_INT96_TIMESTAMP_COMPARATOR
+            : PrimitiveComparator.BINARY_AS_SIGNED_INTEGER_COMPARATOR;
       }
     },
     FIXED_LEN_BYTE_ARRAY("getBinary", Binary.class) {
@@ -578,9 +580,14 @@ public PrimitiveType(
     this.decimalMeta = decimalMeta;
 
     if (columnOrder == null) {
-      columnOrder = primitive == PrimitiveTypeName.INT96 || originalType == OriginalType.INTERVAL
-          ? ColumnOrder.undefined()
-          : ColumnOrder.typeDefined();
+      if (primitive == PrimitiveTypeName.INT96) {
+        // INT96 is only used for deprecated timestamps and supports no other semantics.
+        columnOrder = ColumnOrder.int96TimestampOrder();
+      } else if (originalType == OriginalType.INTERVAL) {
+        columnOrder = ColumnOrder.undefined();
+      } else {
+        columnOrder = ColumnOrder.typeDefined();
+      }
     } else if (columnOrder.getColumnOrderName() == ColumnOrderName.IEEE_754_TOTAL_ORDER) {
       Preconditions.checkArgument(
           primitive == PrimitiveTypeName.FLOAT || primitive == PrimitiveTypeName.DOUBLE,
@@ -629,10 +636,17 @@ public PrimitiveType(
     }
 
     if (columnOrder == null) {
-      columnOrder = primitive == PrimitiveTypeName.INT96
-              || logicalTypeAnnotation instanceof LogicalTypeAnnotation.IntervalLogicalTypeAnnotation
-          ? ColumnOrder.undefined()
-          : ColumnOrder.typeDefined();
+      if (primitive == PrimitiveTypeName.INT96) {
+        // A plain INT96 is the legacy timestamp encoding; default it to the chronological order.
+        // An annotated INT96 carries other semantics, so leave its order undefined.
+        columnOrder = logicalTypeAnnotation == null
+            ? ColumnOrder.int96TimestampOrder()
+            : ColumnOrder.undefined();
+      } else if (logicalTypeAnnotation instanceof LogicalTypeAnnotation.IntervalLogicalTypeAnnotation) {
+        columnOrder = ColumnOrder.undefined();
+      } else {
+        columnOrder = ColumnOrder.typeDefined();
+      }
     } else if (columnOrder.getColumnOrderName() == ColumnOrderName.IEEE_754_TOTAL_ORDER) {
       Preconditions.checkArgument(
           primitive == PrimitiveTypeName.FLOAT
@@ -651,9 +665,15 @@ public PrimitiveType(
   private ColumnOrder requireValidColumnOrder(ColumnOrder columnOrder) {
     if (primitive == PrimitiveTypeName.INT96) {
       Preconditions.checkArgument(
-          columnOrder.getColumnOrderName() == ColumnOrderName.UNDEFINED,
+          columnOrder.getColumnOrderName() == ColumnOrderName.UNDEFINED
+              || columnOrder.getColumnOrderName() == ColumnOrderName.INT96_TIMESTAMP_ORDER,
           "The column order %s is not supported by INT96",
           columnOrder);
+    } else {
+      Preconditions.checkArgument(
+          columnOrder.getColumnOrderName() != ColumnOrderName.INT96_TIMESTAMP_ORDER,
+          "The column order %s is only supported by INT96",
+          columnOrder);
     }
     if (getLogicalTypeAnnotation() != null) {
       Preconditions.checkArgument(

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

@@ -93,7 +93,19 @@ 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);
+  }
+
+  @Test
+  public void testInt96() {
+    // INT96 has a fixed 12-byte width and a chronological comparator (so it is excluded from the
+    // variable-length checks above, like FLOAT16). Its truncator is a no-op: verify it returns the
+    // value unchanged regardless of the requested length.
+    BinaryTruncator int96Truncator = BinaryTruncator.getTruncator(
+      Types.required(INT96).named("test_int96"));
+    Binary int96Value = Binary.fromConstantByteArray(
+      new byte[] {0, 0, 0, 0, 0, 0, 0, 0, 1, 2, 3, 4});
+    assertSame(int96Value, int96Truncator.truncateMin(int96Value, 4));
+    assertSame(int96Value, int96Truncator.truncateMax(int96Value, 4));
   }
 
   @Test

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

@@ -20,6 +20,7 @@
 
 import static org.apache.parquet.schema.PrimitiveComparator.BINARY_AS_FLOAT16_COMPARATOR;
 import static org.apache.parquet.schema.PrimitiveComparator.BINARY_AS_FLOAT16_IEEE_754_TOTAL_ORDER_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;
@@ -38,6 +39,7 @@
 import java.nio.ByteBuffer;
 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;
 
@@ -354,6 +356,51 @@ public void testBinaryAsSignedIntegerComparatorWithEquals() {
     }
   }
 
+  private static Binary int96(int julianDay, long nanosOfDay) {
+    return new NanoTime(julianDay, nanosOfDay).toBinary();
+  }
+
+  @Test
+  public void testInt96TimestampComparator() {
+    Binary[] valuesInAscendingOrder = {
+      int96(Integer.MIN_VALUE, 0), // most negative julian day
+      int96(-1, 86_399_999_999_999L), // negative julian days sort before day 0
+      int96(0, 0), // start of the julian period
+      int96(0, 86_399_999_999_999L), // same day, later time of day
+      int96(2440000, 123L), // 1968-05-23T00:00:00.000000123, pre-epoch but positive julian day
+      int96(2458850, 43_200_000_000_000L), // 2020-01-01T12:00:00
+      int96(2458881, 39_600_000_000_000L), // 2020-02-01T11:00:00, later day even though earlier time of day
+      int96(2458881, 39_600_000_000_001L), // 2020-02-01T11:00:00.000000001, nanos tie-break
+      int96(Integer.MAX_VALUE, 86_399_999_999_999L)
+    };
+
+    for (int i = 0; i < valuesInAscendingOrder.length; ++i) {
+      for (int j = 0; j < valuesInAscendingOrder.length; ++j) {
+        assertEquals(
+            "comparing value " + i + " to value " + j,
+            Integer.signum(Integer.compare(i, j)),
+            Integer.signum(BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(
+                valuesInAscendingOrder[i], valuesInAscendingOrder[j])));
+      }
+    }
+  }
+
+  @Test
+  public void testInt96TimestampComparatorRejectsInvalidNanos() {
+    // Same Julian day so the comparator reaches the nanos validation instead of
+    // returning early on the day comparison.
+    Binary valid = int96(0, 0);
+    for (long invalidNanos : new long[] {-1L, Long.MIN_VALUE, 86_400_000_000_001L, Long.MAX_VALUE}) {
+      Binary invalid = int96(0, invalidNanos);
+      try {
+        BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(valid, invalid);
+        fail("Expected IllegalArgumentException for nanos=" + invalidNanos);
+      } catch (IllegalArgumentException e) {
+        // expected
+      }
+    }
+  }
+
   @Test
   public void testFloat16Comparator() {
     Binary[] valuesInAscendingOrder = {

parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java

@@ -88,6 +88,7 @@
 import org.apache.parquet.format.GeometryType;
 import org.apache.parquet.format.GeospatialStatistics;
 import org.apache.parquet.format.IEEE754TotalOrder;
+import org.apache.parquet.format.Int96TimestampOrder;
 import org.apache.parquet.format.IntType;
 import org.apache.parquet.format.KeyValue;
 import org.apache.parquet.format.LogicalType;
@@ -146,6 +147,7 @@ public class ParquetMetadataConverter {
 
   private static final TypeDefinedOrder TYPE_DEFINED_ORDER = new TypeDefinedOrder();
   private static final IEEE754TotalOrder IEEE_754_TOTAL_ORDER = new IEEE754TotalOrder();
+  private static final Int96TimestampOrder INT96_TIMESTAMP_ORDER = new Int96TimestampOrder();
   public static final MetadataFilter NO_FILTER = new NoFilter();
   public static final MetadataFilter SKIP_ROW_GROUPS = new SkipMetadataFilter();
   public static final long MAX_STATS_SIZE = 4096; // limit stats to 4k
@@ -290,6 +292,9 @@ private List<ColumnOrder> getColumnOrders(MessageType schema) {
         case IEEE_754_TOTAL_ORDER:
           columnOrder.setIEEE_754_TOTAL_ORDER(IEEE_754_TOTAL_ORDER);
           break;
+        case INT96_TIMESTAMP_ORDER:
+          columnOrder.setINT96_TIMESTAMP_ORDER(INT96_TIMESTAMP_ORDER);
+          break;
         case UNDEFINED:
           // Use TypeDefinedOrder if some types (e.g. INT96) have undefined column orders.
           columnOrder.setTYPE_ORDER(TYPE_DEFINED_ORDER);
@@ -911,8 +916,16 @@ private static byte[] tuncateMax(BinaryTruncator truncator, int truncateLength,
   }
 
   private static boolean isMinMaxStatsSupported(PrimitiveType type) {
-    return type.columnOrder().getColumnOrderName() == ColumnOrderName.TYPE_DEFINED_ORDER
-        || type.columnOrder().getColumnOrderName() == ColumnOrderName.IEEE_754_TOTAL_ORDER;
+    switch (type.columnOrder().getColumnOrderName()) {
+      case TYPE_DEFINED_ORDER:
+      case IEEE_754_TOTAL_ORDER:
+      case INT96_TIMESTAMP_ORDER:
+        return true;
+      case UNDEFINED:
+        return false;
+      default:
+        throw new IllegalArgumentException("Unknown column order: " + type.columnOrder());
+    }
   }
 
   /**
@@ -2057,7 +2070,17 @@ private void buildChildren(
                   || schemaElement.converted_type == ConvertedType.INTERVAL)) {
             columnOrder = org.apache.parquet.schema.ColumnOrder.undefined();
           }
+          // INT96_TIMESTAMP_ORDER is only valid for INT96 columns, ignore it anywhere else.
+          if (columnOrder.getColumnOrderName() == ColumnOrderName.INT96_TIMESTAMP_ORDER
+              && schemaElement.type != Type.INT96) {
+            columnOrder = org.apache.parquet.schema.ColumnOrder.undefined();
+          }
           primitiveBuilder.columnOrder(columnOrder);
+        } else if (schemaElement.type == Type.INT96) {
+          // A footer without column orders predates INT96_TIMESTAMP_ORDER, so an INT96 column here
+          // must not inherit the (chronological) construction-time default: its stats, if any, were
+          // written under the legacy order and must be ignored.
+          primitiveBuilder.columnOrder(org.apache.parquet.schema.ColumnOrder.undefined());
         }
         childBuilder = primitiveBuilder;
       } else {
@@ -2110,6 +2133,9 @@ private static org.apache.parquet.schema.ColumnOrder fromParquetColumnOrder(Colu
     if (columnOrder.isSetIEEE_754_TOTAL_ORDER()) {
       return org.apache.parquet.schema.ColumnOrder.ieee754TotalOrder();
     }
+    if (columnOrder.isSetINT96_TIMESTAMP_ORDER()) {
+      return org.apache.parquet.schema.ColumnOrder.int96TimestampOrder();
+    }
     // The column order is not yet supported by this API
     return org.apache.parquet.schema.ColumnOrder.undefined();
   }