GH-3679: Allow TimestampType to annotate FLBA(12)

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;
 
@@ -278,6 +279,34 @@ public String toString() {
     }
   };
 
+  /**
+   * Comparator for timestamps encoded as 12-byte little-endian two's-complement values.
+   */
+  static final PrimitiveComparator<Binary> BINARY_AS_SIGNED_TIMESTAMP_COMPARATOR = new BinaryComparator() {
+    @Override
+    int compareBinary(Binary b1, Binary b2) {
+      if (b1.length() != 12 || b2.length() != 12) {
+        throw new IllegalArgumentException(
+            "Timestamp binary length must be 12 bytes, got " + b1.length() + " and " + b2.length());
+      }
+      ByteBuffer bb1 = b1.toByteBuffer().slice();
+      bb1.order(ByteOrder.LITTLE_ENDIAN);
+      ByteBuffer bb2 = b2.toByteBuffer().slice();
+      bb2.order(ByteOrder.LITTLE_ENDIAN);
+      // Signed comparison of the high 4 bytes followed by unsigned comparison of the low 8 bytes.
+      int hiResult = Integer.compare(bb1.getInt(8), bb2.getInt(8));
+      if (hiResult != 0) {
+        return hiResult;
+      }
+      return Long.compareUnsigned(bb1.getLong(0), bb2.getLong(0));
+    }
+
+    @Override
+    public String toString() {
+      return "BINARY_AS_SIGNED_TIMESTAMP_COMPARATOR";
+    }
+  };
+
   /**
    * This comparator is for comparing two float16 values represented in 2 bytes binary.
    */

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

@@ -29,6 +29,7 @@
 import java.math.BigInteger;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.time.DateTimeException;
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.time.format.DateTimeFormatter;
@@ -249,7 +250,7 @@ public String stringify(long value) {
       return toFormattedString(getInstant(value));
     }
 
-    private String toFormattedString(Instant instant) {
+    final String toFormattedString(Instant instant) {
       return formatter.format(instant);
     }
 
@@ -266,6 +267,41 @@ Instant getInstant(long value) {
     }
   }
 
+  /**
+   * Stringifier implementation for timestamps that handles both int64 and FLBA(12) carriers.
+   * Values outside of Instant's supported range render as raw integers rather than human-readable
+   * timestamps.
+   */
+  private abstract static class TimestampStringifier extends DateStringifier {
+    private TimestampStringifier(String name, String format) {
+      super(name, format);
+    }
+
+    @Override
+    public String stringify(Binary value) {
+      if (value == null) {
+        return BINARY_NULL;
+      }
+      byte[] littleEndian = value.getBytesUnsafe();
+      byte[] bigEndian = new byte[littleEndian.length];
+      for (int i = 0; i < littleEndian.length; ++i) {
+        bigEndian[i] = littleEndian[littleEndian.length - 1 - i];
+      }
+      try {
+        BigInteger units = new BigInteger(bigEndian);
+        try {
+          return toFormattedString(getInstant(units));
+        } catch (ArithmeticException | DateTimeException e) {
+          return units.toString();
+        }
+      } catch (NumberFormatException e) {
+        return BINARY_INVALID;
+      }
+    }
+
+    abstract Instant getInstant(BigInteger value);
+  }
+
   static final PrimitiveStringifier DATE_STRINGIFIER = new DateStringifier("DATE_STRINGIFIER", "yyyy-MM-dd") {
     @Override
     Instant getInstant(int value) {
@@ -274,56 +310,96 @@ Instant getInstant(int value) {
     ;
   };
 
+  // Converts a count of time units since epoch, held as a BigInteger (the 96-bit FLBA(12) carrier),
+  // into an Instant. The 96-bit count does not fit in a long, but the whole-second count does, so
+  // divide first and narrow after.
+  private static Instant instantFromUnits(BigInteger units, TimeUnit unit) {
+    BigInteger[] secondsAndUnits = units.divideAndRemainder(BigInteger.valueOf(unit.convert(1, SECONDS)));
+    long seconds = secondsAndUnits[0].longValueExact();
+    long nanos = secondsAndUnits[1].longValue() * unit.toNanos(1);
+    return Instant.ofEpochSecond(seconds, nanos);
+  }
+
   static final PrimitiveStringifier TIMESTAMP_MILLIS_STRINGIFIER =
-      new DateStringifier("TIMESTAMP_MILLIS_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSS") {
+      new TimestampStringifier("TIMESTAMP_MILLIS_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSS") {
         @Override
         Instant getInstant(long value) {
           return Instant.ofEpochMilli(value);
         }
+
+        @Override
+        Instant getInstant(BigInteger value) {
+          return instantFromUnits(value, MILLISECONDS);
+        }
       };
 
   static final PrimitiveStringifier TIMESTAMP_MICROS_STRINGIFIER =
-      new DateStringifier("TIMESTAMP_MICROS_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSSSS") {
+      new TimestampStringifier("TIMESTAMP_MICROS_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSSSS") {
         @Override
         Instant getInstant(long value) {
           return Instant.ofEpochSecond(
               MICROSECONDS.toSeconds(value), MICROSECONDS.toNanos(value % SECONDS.toMicros(1)));
         }
+
+        @Override
+        Instant getInstant(BigInteger value) {
+          return instantFromUnits(value, MICROSECONDS);
+        }
       };
 
   static final PrimitiveStringifier TIMESTAMP_NANOS_STRINGIFIER =
-      new DateStringifier("TIMESTAMP_NANOS_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSSSSSSS") {
+      new TimestampStringifier("TIMESTAMP_NANOS_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSSSSSSS") {
         @Override
         Instant getInstant(long value) {
           return Instant.ofEpochSecond(
               NANOSECONDS.toSeconds(value), NANOSECONDS.toNanos(value % SECONDS.toNanos(1)));
         }
+
+        @Override
+        Instant getInstant(BigInteger value) {
+          return instantFromUnits(value, NANOSECONDS);
+        }
       };
 
   static final PrimitiveStringifier TIMESTAMP_MILLIS_UTC_STRINGIFIER =
-      new DateStringifier("TIMESTAMP_MILLIS_UTC_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSZ") {
+      new TimestampStringifier("TIMESTAMP_MILLIS_UTC_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSZ") {
         @Override
         Instant getInstant(long value) {
           return Instant.ofEpochMilli(value);
         }
+
+        @Override
+        Instant getInstant(BigInteger value) {
+          return instantFromUnits(value, MILLISECONDS);
+        }
       };
 
   static final PrimitiveStringifier TIMESTAMP_MICROS_UTC_STRINGIFIER =
-      new DateStringifier("TIMESTAMP_MICROS_UTC_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSSSSZ") {
+      new TimestampStringifier("TIMESTAMP_MICROS_UTC_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSSSSZ") {
         @Override
         Instant getInstant(long value) {
           return Instant.ofEpochSecond(
               MICROSECONDS.toSeconds(value), MICROSECONDS.toNanos(value % SECONDS.toMicros(1)));
         }
+
+        @Override
+        Instant getInstant(BigInteger value) {
+          return instantFromUnits(value, MICROSECONDS);
+        }
       };
 
   static final PrimitiveStringifier TIMESTAMP_NANOS_UTC_STRINGIFIER =
-      new DateStringifier("TIMESTAMP_NANOS_UTC_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSSSSSSSZ") {
+      new TimestampStringifier("TIMESTAMP_NANOS_UTC_STRINGIFIER", "yyyy-MM-dd'T'HH:mm:ss.SSSSSSSSSZ") {
         @Override
         Instant getInstant(long value) {
           return Instant.ofEpochSecond(
               NANOSECONDS.toSeconds(value), NANOSECONDS.toNanos(value % SECONDS.toNanos(1)));
         }
+
+        @Override
+        Instant getInstant(BigInteger value) {
+          return instantFromUnits(value, NANOSECONDS);
+        }
       };
 
   private abstract static class TimeStringifier extends PrimitiveStringifier {

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

@@ -444,6 +444,12 @@ public Optional<PrimitiveComparator> visit(
                     : of(PrimitiveComparator.BINARY_AS_FLOAT16_COMPARATOR);
               }
 
+              @Override
+              public Optional<PrimitiveComparator> visit(
+                  LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {
+                return of(PrimitiveComparator.BINARY_AS_SIGNED_TIMESTAMP_COMPARATOR);
+              }
+
               @Override
               public Optional<PrimitiveComparator> visit(
                   LogicalTypeAnnotation.UnknownLogicalTypeAnnotation unknownLogicalTypeAnnotation) {

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

@@ -563,7 +563,13 @@ public Optional<Boolean> visit(
               @Override
               public Optional<Boolean> visit(
                   LogicalTypeAnnotation.TimestampLogicalTypeAnnotation timestampLogicalType) {
-                return checkInt64PrimitiveType(timestampLogicalType);
+                Preconditions.checkState(
+                    primitiveType == PrimitiveTypeName.INT64
+                        || (primitiveType == PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
+                            && length == 12),
+                    "%s can only annotate INT64 or FIXED_LEN_BYTE_ARRAY(12)",
+                    timestampLogicalType);
+                return Optional.of(true);
               }
 
               @Override

parquet-column/src/test/java/org/apache/parquet/column/statistics/TestStatistics.java

@@ -22,6 +22,8 @@
 import static java.lang.Float.floatToIntBits;
 import static org.apache.parquet.bytes.BytesUtils.intToBytes;
 import static org.apache.parquet.bytes.BytesUtils.longToBytes;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit.NANOS;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.timestampType;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BOOLEAN;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.DOUBLE;
@@ -940,6 +942,32 @@ public void testNoopStatistics() {
         .hasMessage("isSmallerThan is not supported by org.apache.parquet.column.statistics.NoopStatistics");
   }
 
+  @Test
+  public void testFlba12TimestampStats() {
+    PrimitiveType type = Types.required(FIXED_LEN_BYTE_ARRAY)
+        .length(12)
+        .as(timestampType(true, NANOS))
+        .named("ts");
+    BinaryStatistics stats = (BinaryStatistics) Statistics.createStats(type);
+
+    // -1 (all 0xFF) — just before epoch, negative value
+    byte[] negOne = new byte[12];
+    for (int i = 0; i < 12; i++) negOne[i] = (byte) 0xFF;
+    // +1 (0x01 followed by zeros) — just after epoch, positive value
+    byte[] posOne = new byte[12];
+    posOne[0] = 1;
+    // epoch (all 0x00)
+    byte[] epoch = new byte[12];
+
+    stats.updateStats(Binary.fromConstantByteArray(posOne));
+    stats.updateStats(Binary.fromConstantByteArray(negOne));
+    stats.updateStats(Binary.fromConstantByteArray(epoch));
+
+    // min must be the negative value (-1), max must be the positive value (+1)
+    assertThat(stats.genericGetMin()).isEqualTo(Binary.fromConstantByteArray(negOne));
+    assertThat(stats.genericGetMax()).isEqualTo(Binary.fromConstantByteArray(posOne));
+  }
+
   @Test
   public void testBinaryIsSmallerThanNoOverflowForLargeValues() {
     BinaryStatistics stats = new BinaryStatistics();

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

@@ -36,6 +36,8 @@
 import java.util.Comparator;
 import java.util.Random;
 import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit;
 import org.apache.parquet.schema.PrimitiveStringifier;
 import org.apache.parquet.schema.PrimitiveType;
 import org.apache.parquet.schema.Types;
@@ -91,6 +93,17 @@ public void testContractNonStringTypes() {
     testTruncator(Types.required(INT96).named("test_int96"), false);
   }
 
+  @Test
+  public void testFlba12Timestamp() {
+    BinaryTruncator truncator = BinaryTruncator.getTruncator(Types.required(FIXED_LEN_BYTE_ARRAY)
+        .length(12)
+        .as(LogicalTypeAnnotation.timestampType(true, TimeUnit.NANOS))
+        .named("test_fixed_timestamp"));
+    Binary value = Binary.fromConstantByteArray(new byte[] {0, 0, 0, 0, 0, 0, 0, 0, 1, 2, 3, 4});
+    assertThat(truncator.truncateMin(value, 4)).isSameAs(value);
+    assertThat(truncator.truncateMax(value, 4)).isSameAs(value);
+  }
+
   @Test
   public void testStringTruncate() {
     BinaryTruncator truncator =

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

@@ -37,12 +37,15 @@
 import static org.apache.parquet.filter2.predicate.FilterApi.or;
 import static org.apache.parquet.filter2.predicate.FilterApi.userDefined;
 import static org.apache.parquet.filter2.predicate.LogicalInverter.invert;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.TimeUnit.NANOS;
+import static org.apache.parquet.schema.LogicalTypeAnnotation.timestampType;
 import static org.apache.parquet.schema.OriginalType.DECIMAL;
 import static org.apache.parquet.schema.OriginalType.UINT_8;
 import static org.apache.parquet.schema.OriginalType.UTF8;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BINARY;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.BOOLEAN;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.DOUBLE;
+import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.FLOAT;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT32;
 import static org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName.INT64;
@@ -2013,6 +2016,50 @@ public void testBuildPreservesValidColumnIndex() {
     assertCorrectNullCounts(ci6, 0, 1);
   }
 
+  @Test
+  public void testBuildFlba12Timestamp() {
+    // FLBA(12) TIMESTAMP column with negative (pre-1970) and positive (post-1970) values.
+    // The LE signed comparator must order negatives before zero before positives.
+    PrimitiveType type = Types.required(FIXED_LEN_BYTE_ARRAY)
+        .length(12)
+        .as(timestampType(true, NANOS))
+        .named("ts");
+    ColumnIndexBuilder builder = ColumnIndexBuilder.getBuilder(type, Integer.MAX_VALUE);
+    assertThat(builder).isInstanceOf(BinaryColumnIndexBuilder.class);
+
+    // -1 (all 0xFF) encodes a value just before epoch — smallest in the test
+    byte[] negOne = new byte[12];
+    for (int i = 0; i < 12; i++) negOne[i] = (byte) 0xFF;
+    // 0 (all 0x00) — epoch
+    byte[] epoch = new byte[12];
+    // +1 (0x01 followed by zeros) — just after epoch
+    byte[] posOne = new byte[12];
+    posOne[0] = 1;
+
+    StatsBuilder sb = new StatsBuilder();
+    // Page 0: only negative values, min=-1 max=-1
+    builder.add(sb.stats(type, Binary.fromConstantByteArray(negOne)));
+    // Page 1: straddles epoch, min=-1 max=+1
+    builder.add(sb.stats(type, Binary.fromConstantByteArray(negOne), Binary.fromConstantByteArray(posOne)));
+    // Page 2: only positive values, min=epoch max=+1
+    builder.add(sb.stats(type, Binary.fromConstantByteArray(epoch), Binary.fromConstantByteArray(posOne)));
+
+    ColumnIndex columnIndex = builder.build();
+    assertThat(columnIndex).isNotNull();
+    // min of page 0 must equal -1
+    assertCorrectValues(
+        columnIndex.getMinValues(),
+        Binary.fromConstantByteArray(negOne),
+        Binary.fromConstantByteArray(negOne),
+        Binary.fromConstantByteArray(epoch));
+    // max of page 0 must equal -1 (all values are negative)
+    assertCorrectValues(
+        columnIndex.getMaxValues(),
+        Binary.fromConstantByteArray(negOne),
+        Binary.fromConstantByteArray(posOne),
+        Binary.fromConstantByteArray(posOne));
+  }
+
   @Test
   public void testBuildWithoutNullCountsIsNotRejected() {
     PrimitiveType type = Types.required(INT32).named("test_col");