You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@parquet.apache.org by ga...@apache.org on 2026/10/12 02:40:12 UTC
(parquet-java) branch master updated: GH-3609: Add new sort order for int96 timestamps (#3610)
This is an automated email from the ASF dual-hosted git repository.
wgtmac pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/parquet-java.git
The following commit(s) were added to refs/heads/master by this push:
new 878a5e550 GH-3609: Add new sort order for int96 timestamps (#3610)
878a5e550 is described below
commit 878a5e550d1372533da388d1fbbe9564ac92a384
Author: Divjot Arora <di...@databricks.com>
AuthorDate: Mon Oct 12 04:40:07 2026 +0200
GH-3609: Add new sort order for int96 timestamps (#3610)
---
.../org/apache/parquet/schema/ColumnOrder.java | 14 ++
.../apache/parquet/schema/MessageTypeParser.java | 3 +
.../apache/parquet/schema/PrimitiveComparator.java | 45 ++++
.../org/apache/parquet/schema/PrimitiveType.java | 27 ++-
.../column/columnindex/TestBinaryTruncator.java | 13 +-
.../parquet/schema/TestPrimitiveComparator.java | 43 ++++
.../format/converter/ParquetMetadataConverter.java | 30 ++-
.../dictionarylevel/DictionaryFilterTest.java | 19 +-
.../converter/TestParquetMetadataConverter.java | 85 +++++--
.../apache/parquet/statistics/RandomValues.java | 5 +-
.../statistics/TestInt96TimestampStatistics.java | 243 +++++++++++++++++++++
11 files changed, 492 insertions(+), 35 deletions(-)
diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/ColumnOrder.java b/parquet-column/src/main/java/org/apache/parquet/schema/ColumnOrder.java
index 35ef0ec9d..4edd1c1f1 100644
--- a/parquet-column/src/main/java/org/apache/parquet/schema/ColumnOrder.java
+++ b/parquet-column/src/main/java/org/apache/parquet/schema/ColumnOrder.java
@@ -41,11 +41,17 @@ public class ColumnOrder {
* 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 +77,14 @@ public class ColumnOrder {
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) {
diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java b/parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java
index 6ccdc4ca7..de0178ac3 100644
--- a/parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java
+++ b/parquet-column/src/main/java/org/apache/parquet/schema/MessageTypeParser.java
@@ -257,6 +257,9 @@ public class MessageTypeParser {
if (t.equalsIgnoreCase(ColumnOrder.ColumnOrderName.IEEE_754_TOTAL_ORDER.name())) {
return ColumnOrder.ieee754TotalOrder();
}
+ if (t.equalsIgnoreCase(ColumnOrder.ColumnOrderName.INT96_TIMESTAMP_ORDER.name())) {
+ return ColumnOrder.int96TimestampOrder();
+ }
return ColumnOrder.undefined();
}
diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveComparator.java b/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveComparator.java
index 8ef548a99..1b7a6ecba 100644
--- a/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveComparator.java
+++ b/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveComparator.java
@@ -383,4 +383,49 @@ public abstract class PrimitiveComparator<T> implements Comparator<T>, Serializa
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 (must be positive and less than 1 day): " + nanos1);
+ }
+ if (nanos2 < 0 || nanos2 >= NANOSECONDS_PER_DAY) {
+ throw new IllegalArgumentException(
+ "Invalid nanos value (must be positive and less than 1 day): " + nanos2);
+ }
+ return Long.compare(nanos1, nanos2);
+ }
+
+ @Override
+ public String toString() {
+ return "BINARY_AS_INT96_TIMESTAMP_COMPARATOR";
+ }
+ };
}
diff --git a/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java b/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java
index c359c9f41..f94ede0c9 100644
--- a/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java
+++ b/parquet-column/src/main/java/org/apache/parquet/schema/PrimitiveType.java
@@ -385,7 +385,9 @@ public final class PrimitiveType extends Type {
@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) {
@@ -653,15 +655,20 @@ public final class PrimitiveType extends Type {
}
/**
- * The column order used when none is specified explicitly. INT96 and INTERVAL have no defined
- * ordering, so they default to undefined. Floating-point types default to IEEE 754 total order so
- * that NaN values and the sign of zero are ordered deterministically and nan_count statistics can
- * be written; this is skipped when the logical type annotation does not accept IEEE 754 total
+ * The column order used when none is specified explicitly. A plain INT96 is the deprecated
+ * timestamp encoding and defaults to INT96 timestamp order so its min/max are chronological and
+ * usable for pruning; an annotated INT96 carries other semantics and, like INTERVAL, has no
+ * defined ordering and defaults to undefined. Floating-point types default to IEEE 754 total order
+ * so that NaN values and the sign of zero are ordered deterministically and nan_count statistics
+ * can be written; this is skipped when the logical type annotation does not accept IEEE 754 total
* order (e.g. an unknown annotation), leaving the type constructible with the type-defined order.
*/
private static ColumnOrder defaultColumnOrder(
PrimitiveTypeName primitive, OriginalType originalType, LogicalTypeAnnotation logicalTypeAnnotation) {
- if (primitive == PrimitiveTypeName.INT96 || originalType == OriginalType.INTERVAL) {
+ if (primitive == PrimitiveTypeName.INT96) {
+ return logicalTypeAnnotation == null ? ColumnOrder.int96TimestampOrder() : ColumnOrder.undefined();
+ }
+ if (originalType == OriginalType.INTERVAL) {
return ColumnOrder.undefined();
}
boolean isFloatingType = primitive == PrimitiveTypeName.FLOAT
@@ -676,9 +683,15 @@ public final class PrimitiveType extends Type {
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(
diff --git a/parquet-column/src/test/java/org/apache/parquet/internal/column/columnindex/TestBinaryTruncator.java b/parquet-column/src/test/java/org/apache/parquet/internal/column/columnindex/TestBinaryTruncator.java
index 716c01770..0993fa07b 100644
--- a/parquet-column/src/test/java/org/apache/parquet/internal/column/columnindex/TestBinaryTruncator.java
+++ b/parquet-column/src/test/java/org/apache/parquet/internal/column/columnindex/TestBinaryTruncator.java
@@ -90,7 +90,18 @@ public class TestBinaryTruncator {
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});
+ assertThat(int96Truncator.truncateMin(int96Value, 4)).isSameAs(int96Value);
+ assertThat(int96Truncator.truncateMax(int96Value, 4)).isSameAs(int96Value);
}
@Test
diff --git a/parquet-column/src/test/java/org/apache/parquet/schema/TestPrimitiveComparator.java b/parquet-column/src/test/java/org/apache/parquet/schema/TestPrimitiveComparator.java
index c07fdb810..96d2ed4e9 100644
--- a/parquet-column/src/test/java/org/apache/parquet/schema/TestPrimitiveComparator.java
+++ b/parquet-column/src/test/java/org/apache/parquet/schema/TestPrimitiveComparator.java
@@ -20,6 +20,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_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.BINARY_AS_SIGNED_TIMESTAMP_COMPARATOR;
import static org.apache.parquet.schema.PrimitiveComparator.BOOLEAN_COMPARATOR;
@@ -40,6 +41,7 @@ import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
+import org.apache.parquet.example.data.simple.NanoTime;
import org.apache.parquet.io.api.Binary;
import org.junit.jupiter.api.Test;
@@ -354,6 +356,47 @@ public class TestPrimitiveComparator {
}
}
+ 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) {
+ assertThat(Integer.signum(BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(
+ valuesInAscendingOrder[i], valuesInAscendingOrder[j])))
+ .as("comparing value " + i + " to value " + j)
+ .isEqualTo(Integer.signum(Integer.compare(i, 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_000L, Long.MAX_VALUE}) {
+ Binary invalid = int96(0, invalidNanos);
+ assertThatThrownBy(() -> BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(valid, invalid))
+ .as("Expected IllegalArgumentException for nanos=" + invalidNanos)
+ .isInstanceOf(IllegalArgumentException.class);
+ }
+ }
+
@Test
public void testFloat16Comparator() {
Binary[] valuesInAscendingOrder = {
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java
index 4252523fa..bbbe1e78e 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java
@@ -90,6 +90,7 @@ import org.apache.parquet.format.GeographyType;
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;
@@ -148,6 +149,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
@@ -292,6 +294,9 @@ public class ParquetMetadataConverter {
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);
@@ -918,8 +923,16 @@ public class ParquetMetadataConverter {
}
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());
+ }
}
/**
@@ -2114,6 +2127,11 @@ public class ParquetMetadataConverter {
|| 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.FLOAT
|| schemaElement.type == Type.DOUBLE
@@ -2122,6 +2140,11 @@ public class ParquetMetadataConverter {
// here must not inherit the (IEEE 754 total order) construction-time default: its stats, if
// any, were written under the legacy type-defined order and must be read under it.
primitiveBuilder.columnOrder(org.apache.parquet.schema.ColumnOrder.typeDefined());
+ } 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());
}
// Gracefully handle unsupported logical type combinations on the read path.
primitiveBuilder.ignoreUnsupportedLogicalAnnotations();
@@ -2179,6 +2202,9 @@ public class ParquetMetadataConverter {
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();
}
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java b/parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java
index d50306caf..5020f77b7 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/filter2/dictionarylevel/DictionaryFilterTest.java
@@ -70,6 +70,7 @@ import org.apache.parquet.column.ParquetProperties.WriterVersion;
import org.apache.parquet.column.page.DictionaryPage;
import org.apache.parquet.column.page.DictionaryPageReadStore;
import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.NanoTime;
import org.apache.parquet.example.data.simple.SimpleGroupFactory;
import org.apache.parquet.filter2.predicate.FilterPredicate;
import org.apache.parquet.filter2.predicate.LogicalInverseRewriter;
@@ -157,15 +158,15 @@ public class DictionaryFilterTest {
toBinary("9999999999999999999999999999999999999999", 17)
};
private static final Binary[] INT96_VALUES = new Binary[] {
- toBinary("-9999999999999999999999999999", 12),
- toBinary("-9999999999999999999999999998", 12),
- toBinary("-1234567890", 12),
- toBinary("-1", 12),
- toBinary("-0", 12),
- toBinary("1", 12),
- toBinary("1234567890", 12),
- toBinary("-9999999999999999999999999998", 12),
- toBinary("9999999999999999999999999999", 12)
+ new NanoTime(2440587, 86_399_999_999_999L).toBinary(),
+ new NanoTime(2440588, 0L).toBinary(),
+ new NanoTime(2440588, 1L).toBinary(),
+ new NanoTime(2440588, 1_234_567_890L).toBinary(),
+ new NanoTime(2440588, 43_200_000_000_000L).toBinary(),
+ new NanoTime(2440588, 86_399_999_999_999L).toBinary(),
+ new NanoTime(2440589, 0L).toBinary(),
+ new NanoTime(2440589, 1L).toBinary(),
+ new NanoTime(2440589, 1_234_567_890L).toBinary()
};
private static Binary toBinary(String decimalWithoutScale, int byteCount) {
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java b/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java
index 46ffccf7f..a47166ae9 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/format/converter/TestParquetMetadataConverter.java
@@ -93,6 +93,7 @@ import org.apache.parquet.crypto.FileDecryptionProperties;
import org.apache.parquet.crypto.InternalFileDecryptor;
import org.apache.parquet.example.Paper;
import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.NanoTime;
import org.apache.parquet.example.data.simple.SimpleGroup;
import org.apache.parquet.format.BoundingBox;
import org.apache.parquet.format.ColumnChunk;
@@ -1321,9 +1322,11 @@ public class TestParquetMetadataConverter {
new BigInteger("12345678"),
new BigInteger("12345679"));
testSkippedV2Stats(
- Types.optional(PrimitiveTypeName.INT96).named(""),
- new BigInteger("-75687987"),
- new BigInteger("45367657"));
+ Types.optional(PrimitiveTypeName.INT96)
+ .columnOrder(ColumnOrder.undefined())
+ .named(""),
+ new NanoTime(2458850, 0L),
+ new NanoTime(2458881, 39_600_000_000_000L));
}
private void testSkippedV2Stats(PrimitiveType type, Object min, Object max) {
@@ -1362,6 +1365,10 @@ public class TestParquetMetadataConverter {
.named(""),
new BigInteger("-6769643"),
new BigInteger("9864675"));
+ testV2OnlyStats(
+ Types.optional(PrimitiveTypeName.INT96).named(""),
+ new NanoTime(2458850, 0L), // 2020-01-01
+ new NanoTime(2458881, 39_600_000_000_000L)); // 2020-02-01T11:00
}
private void testV2OnlyStats(PrimitiveType type, Object min, Object max) {
@@ -1400,10 +1407,8 @@ public class TestParquetMetadataConverter {
.named(""),
new BigInteger("-8752832"),
new BigInteger("-8752832"));
- testV2StatsEqualMinMax(
- Types.optional(PrimitiveTypeName.INT96).named(""),
- new BigInteger("81032984"),
- new BigInteger("81032984"));
+ NanoTime int96 = new NanoTime(2458850, 43_200_000_000_000L); // 2020-01-01T12:00:00
+ testV2StatsEqualMinMax(Types.optional(PrimitiveTypeName.INT96).named(""), int96, int96);
}
private void testV2StatsEqualMinMax(PrimitiveType type, Object min, Object max) {
@@ -1423,6 +1428,8 @@ public class TestParquetMetadataConverter {
return createStatsTyped(type, (Long) min, (Long) max);
} else if (c == BigInteger.class) {
return createStatsTyped(type, (BigInteger) min, (BigInteger) max);
+ } else if (min instanceof NanoTime) {
+ return createStatsTyped(type, (NanoTime) min, (NanoTime) max);
}
fail("Not implemented");
return null;
@@ -1457,6 +1464,17 @@ public class TestParquetMetadataConverter {
return stats;
}
+ private static Statistics<?> createStatsTyped(PrimitiveType type, NanoTime min, NanoTime max) {
+ Binary minBinary = min.toBinary();
+ Binary maxBinary = max.toBinary();
+ Statistics<?> stats = Statistics.createStats(type);
+ stats.updateStats(maxBinary);
+ stats.updateStats(minBinary);
+ assertThat(stats.genericGetMin()).isEqualTo(minBinary);
+ assertThat(stats.genericGetMax()).isEqualTo(maxBinary);
+ return stats;
+ }
+
private static ParquetMetadata createParquetMetaData(Encoding dicEncoding, Encoding dataEncoding) {
return createParquetMetaData(dicEncoding, dataEncoding, true);
}
@@ -1525,8 +1543,7 @@ public class TestParquetMetadataConverter {
+ " required binary key (UTF8);" // Key to be hacked to have unknown column order -> undefined
+ " optional group list_col (LIST) {"
+ " repeated group list {"
- + " optional int96 array_element;" // INT96 element with type defined column order ->
- // undefined
+ + " optional int96 array_element;" // plain INT96 element -> INT96_TIMESTAMP_ORDER
+ " }"
+ " }"
+ " }"
@@ -1540,9 +1557,10 @@ public class TestParquetMetadataConverter {
List<org.apache.parquet.format.ColumnOrder> columnOrders = formatMetadata.getColumn_orders();
assertThat(columnOrders).hasSize(3);
- for (org.apache.parquet.format.ColumnOrder columnOrder : columnOrders) {
- assertThat(columnOrder.isSetTYPE_ORDER()).isTrue();
- }
+ // binary_col and key get TYPE_ORDER, the INT96 array_element gets INT96_TIMESTAMP_ORDER.
+ assertThat(columnOrders.get(0).isSetTYPE_ORDER()).isTrue();
+ assertThat(columnOrders.get(1).isSetTYPE_ORDER()).isTrue();
+ assertThat(columnOrders.get(2).isSetINT96_TIMESTAMP_ORDER()).isTrue();
// Simulate that thrift got a union type that is not in the generated code
// (when the file contains a not-yet-supported column order)
@@ -1554,7 +1572,42 @@ public class TestParquetMetadataConverter {
assertThat(columns).hasSize(3);
assertThat(columns.get(0).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.typeDefined());
assertThat(columns.get(1).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.undefined());
- assertThat(columns.get(2).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.undefined());
+ assertThat(columns.get(2).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.int96TimestampOrder());
+ }
+
+ @Test
+ public void testUndefinedINT96ColumnOrder() throws IOException {
+ assertInt96ColumnOrderIgnored(null);
+ }
+
+ @Test
+ public void testTypeDefinedOrderINT96ColumnOrder() throws IOException {
+ org.apache.parquet.format.ColumnOrder typeOrder = new org.apache.parquet.format.ColumnOrder();
+ typeOrder.setTYPE_ORDER(new org.apache.parquet.format.TypeDefinedOrder());
+ assertInt96ColumnOrderIgnored(typeOrder);
+ }
+
+ private void assertInt96ColumnOrderIgnored(org.apache.parquet.format.ColumnOrder footerColumnOrder)
+ throws IOException {
+ MessageType schema = parseMessageType("message test {" + " optional int96 int96_col;" + "}");
+ org.apache.parquet.hadoop.metadata.FileMetaData fileMetaData =
+ new org.apache.parquet.hadoop.metadata.FileMetaData(schema, new HashMap<String, String>(), null);
+ ParquetMetadata metadata = new ParquetMetadata(fileMetaData, new ArrayList<BlockMetaData>());
+ ParquetMetadataConverter converter = new ParquetMetadataConverter();
+ FileMetaData formatMetadata = converter.toParquetMetadata(1, metadata);
+
+ if (footerColumnOrder == null) {
+ formatMetadata.unsetColumn_orders();
+ assertThat(formatMetadata.isSetColumn_orders()).isFalse();
+ } else {
+ formatMetadata.setColumn_orders(Collections.singletonList(footerColumnOrder));
+ }
+
+ MessageType resultSchema =
+ converter.fromParquetMetadata(formatMetadata).getFileMetaData().getSchema();
+ List<ColumnDescriptor> columns = resultSchema.getColumns();
+ assertThat(columns).hasSize(1);
+ assertThat(columns.get(0).getPrimitiveType().columnOrder()).isEqualTo(ColumnOrder.undefined());
}
@Test
@@ -1627,7 +1680,11 @@ public class TestParquetMetadataConverter {
.as("Should handle null column index")
.isNull();
assertThat(ParquetMetadataConverter.toParquetColumnIndex(
- Types.required(PrimitiveTypeName.INT96).named("test_int96"), columnIndex))
+ Types.required(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
+ .length(12)
+ .as(OriginalType.INTERVAL)
+ .named("test_interval"),
+ columnIndex))
.as("Should ignore unsupported types")
.isNull();
assertThat(ParquetMetadataConverter.fromParquetColumnIndex(
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/statistics/RandomValues.java b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/RandomValues.java
index f2cf26d86..1b49503bb 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/statistics/RandomValues.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/RandomValues.java
@@ -26,7 +26,7 @@ import java.util.Iterator;
import java.util.List;
import java.util.Random;
import java.util.function.Supplier;
-import org.apache.parquet.FixedBinaryTestUtils;
+import org.apache.parquet.example.data.simple.NanoTime;
import org.apache.parquet.io.api.Binary;
public class RandomValues {
@@ -214,6 +214,7 @@ public class RandomValues {
}
public static class Int96Generator extends RandomBinaryBase<BigInteger> {
+ private static final long NANOSECONDS_PER_DAY = 86_400_000_000_000L;
private final RandomRange<BigInteger> randomRange = new RandomRange<BigInteger>(randomInt96(), randomInt96());
private final BigInteger minimum = randomRange.minimum();
private final BigInteger maximum = randomRange.maximum();
@@ -232,7 +233,7 @@ public class RandomValues {
@Override
public Binary nextBinaryValue() {
- return FixedBinaryTestUtils.getFixedBinary(INT_96_LENGTH, nextValue());
+ return new NanoTime(randomInt(), Math.floorMod(randomLong(), NANOSECONDS_PER_DAY)).toBinary();
}
}
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java
new file mode 100644
index 000000000..0ae291ed8
--- /dev/null
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java
@@ -0,0 +1,243 @@
+/*
+ * 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.statistics;
+
+import static org.apache.parquet.schema.MessageTypeParser.parseMessageType;
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
+import java.util.function.Consumer;
+import org.apache.parquet.ParquetReadOptions;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.NanoTime;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.format.ColumnChunk;
+import org.apache.parquet.format.FileMetaData;
+import org.apache.parquet.format.Statistics;
+import org.apache.parquet.format.Type;
+import org.apache.parquet.format.TypeDefinedOrder;
+import org.apache.parquet.format.Util;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.ParquetFileWriter;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.internal.column.columnindex.ColumnIndex;
+import org.apache.parquet.io.LocalInputFile;
+import org.apache.parquet.io.LocalOutputFile;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.ColumnOrder;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+/**
+ * Tests for INT96 timestamp statistics support (INT96_TIMESTAMP_ORDER).
+ */
+public class TestInt96TimestampStatistics {
+
+ private static final MessageType SCHEMA =
+ parseMessageType("message test { required int96 ts; required int64 id; } ");
+
+ // Chronologically: EARLY < SAME_DAY_EARLY < LATE_IN_DAY < NEXT_DAY.
+ // Byte-wise lexicographic comparison would order these incorrectly (nanos bytes come first),
+ // so these values detect a reader/writer using the wrong order.
+ private static final Binary EARLY = new NanoTime(2440000, 123L).toBinary(); // 1968-05-23 00:00:00.000000123
+ private static final Binary SAME_DAY_EARLY = new NanoTime(2440588, 1_000L).toBinary(); // 1970-01-01 00:00:00.000001
+ private static final Binary LATE_IN_DAY =
+ new NanoTime(2440588, 86_399_999_999_999L).toBinary(); // 1970-01-01 23:59:59.999...
+ private static final Binary NEXT_DAY = new NanoTime(2440589, 0L).toBinary(); // 1970-01-02 00:00:00
+
+ private static final List<Binary> VALUES = List.of(LATE_IN_DAY, NEXT_DAY, EARLY, SAME_DAY_EARLY);
+ private static final Binary EXPECTED_MIN = EARLY;
+ private static final Binary EXPECTED_MAX = NEXT_DAY;
+
+ @TempDir
+ private Path tempDir;
+
+ private File writeFile() throws IOException {
+ File file = tempDir.resolve("int96.parquet").toFile();
+ SimpleGroupFactory factory = new SimpleGroupFactory(SCHEMA);
+ ParquetWriter<Group> writer = ExampleParquetWriter.builder(new LocalOutputFile(file.toPath()))
+ .withType(SCHEMA)
+ .build();
+ try {
+ for (int i = 0; i < VALUES.size(); i++) {
+ writer.write(factory.newGroup().append("ts", VALUES.get(i)).append("id", (long) i));
+ }
+ } finally {
+ writer.close();
+ }
+ return file;
+ }
+
+ private static ParquetMetadata readFooter(File file) throws IOException {
+ try (ParquetFileReader reader = ParquetFileReader.open(
+ new LocalInputFile(file.toPath()), ParquetReadOptions.builder().build())) {
+ return reader.getFooter();
+ }
+ }
+
+ private static FileMetaData readRawFooter(File file) throws IOException {
+ byte[] bytes = Files.readAllBytes(file.toPath());
+ int footerLen = ByteBuffer.wrap(bytes, bytes.length - 8, 4)
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .getInt();
+ int footerStart = bytes.length - 8 - footerLen;
+ return Util.readFileMetaData(new ByteArrayInputStream(bytes, footerStart, footerLen));
+ }
+
+ /**
+ * Copies the contents of src into a new file, but replaces the existing footer with the
+ * given footer.
+ */
+ private File copyWithNewFooter(File src, FileMetaData footer, String name) throws IOException {
+ byte[] bytes = Files.readAllBytes(src.toPath());
+ int footerLen = ByteBuffer.wrap(bytes, bytes.length - 8, 4)
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .getInt();
+ int footerStart = bytes.length - 8 - footerLen;
+ File dst = tempDir.resolve(name).toFile();
+ try (FileOutputStream out = new FileOutputStream(dst)) {
+ out.write(bytes, 0, footerStart);
+ ByteArrayOutputStream serialized = new ByteArrayOutputStream();
+ Util.writeFileMetaData(footer, serialized);
+ out.write(serialized.toByteArray());
+ out.write(ByteBuffer.allocate(4)
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .putInt(serialized.size())
+ .array());
+ out.write(ParquetFileWriter.MAGIC);
+ }
+ return dst;
+ }
+
+ private static ColumnChunkMetaData getColumn(ParquetMetadata footer, String name) {
+ BlockMetaData block = footer.getBlocks().get(0);
+ return block.getColumns().stream()
+ .filter(c -> c.getPath().toDotString().equals(name))
+ .findFirst()
+ .orElseThrow(() -> new IllegalArgumentException("no column " + name));
+ }
+
+ private static void assertStatsIgnored(ColumnChunkMetaData column) {
+ var stats = column.getStatistics();
+ if (stats != null) assertThat(stats.hasNonNullValue()).isFalse();
+ }
+
+ private static void assertStatsUsable(ColumnChunkMetaData column) {
+ var stats = column.getStatistics();
+ assertThat(stats).isNotNull();
+ assertThat(stats.hasNonNullValue()).isTrue();
+ assertThat(stats.getMinBytes()).containsExactly(EXPECTED_MIN.getBytes());
+ assertThat(stats.getMaxBytes()).containsExactly(EXPECTED_MAX.getBytes());
+ }
+
+ @Test
+ public void testWriterEmitsInt96StatsAndColumnOrder() throws IOException {
+ // Values are written in non-chronological order; the writer must still produce the
+ // chronological min/max, not first/last or byte-wise extremes.
+ File file = writeFile();
+ FileMetaData rawFooter = readRawFooter(file);
+ // schema[0] is the message root; column_orders are indexed by leaf order: ts=0, id=1
+ assertThat(rawFooter.getColumn_orders().get(0).isSetINT96_TIMESTAMP_ORDER())
+ .isTrue();
+
+ // Statistics should be present for the INT96 column.
+ ColumnChunk tsChunk = rawFooter.getRow_groups().get(0).getColumns().get(0);
+ assertThat(tsChunk.getMeta_data().getType()).isEqualTo(Type.INT96);
+ Statistics stats = tsChunk.getMeta_data().getStatistics();
+ assertThat(stats).isNotNull();
+ assertThat(stats.getMin_value()).containsExactly(EXPECTED_MIN.getBytes());
+ assertThat(stats.getMax_value()).containsExactly(EXPECTED_MAX.getBytes());
+
+ // Column index should be present for the INT96 column.
+ try (ParquetFileReader reader = ParquetFileReader.open(
+ new LocalInputFile(file.toPath()), ParquetReadOptions.builder().build())) {
+ ColumnIndex columnIndex = reader.readColumnIndex(getColumn(reader.getFooter(), "ts"));
+ assertThat(columnIndex).isNotNull();
+ assertThat(toArray(columnIndex.getMinValues().get(0))).containsExactly(EXPECTED_MIN.getBytes());
+ assertThat(toArray(columnIndex.getMaxValues().get(0))).containsExactly(EXPECTED_MAX.getBytes());
+ }
+ }
+
+ @Test
+ public void testReaderReadsStatsWrittenWithInt96TimestampOrder() throws IOException {
+ File file = writeFile();
+ ParquetMetadata footer = readFooter(file);
+
+ PrimitiveType ts = footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType();
+ assertThat(ts.columnOrder()).isEqualTo(ColumnOrder.int96TimestampOrder());
+ assertThat(ts.comparator()).asString().isEqualTo("BINARY_AS_INT96_TIMESTAMP_COMPARATOR");
+
+ assertStatsUsable(getColumn(footer, "ts"));
+ }
+
+ private void assertLegacyInt96StatsIgnored(String name, Consumer<FileMetaData> legacyMutation) throws IOException {
+ File file = writeFile();
+ FileMetaData rawFooter = readRawFooter(file);
+ legacyMutation.accept(rawFooter);
+ File legacy = copyWithNewFooter(file, rawFooter, name + "-orders.parquet");
+
+ ParquetMetadata footer = readFooter(legacy);
+ assertThat(footer.getFileMetaData()
+ .getSchema()
+ .getType("ts")
+ .asPrimitiveType()
+ .columnOrder())
+ .isEqualTo(ColumnOrder.undefined());
+ assertStatsIgnored(getColumn(footer, "ts"));
+ // The non-INT96 sibling column is unaffected.
+ assertThat(getColumn(footer, "id").getStatistics().hasNonNullValue()).isTrue();
+ }
+
+ @Test
+ public void testReaderIgnoresInt96StatsWithTypeDefinedOrder() throws IOException {
+ assertLegacyInt96StatsIgnored("type-defined", footer -> {
+ for (org.apache.parquet.format.ColumnOrder columnOrder : footer.getColumn_orders()) {
+ if (columnOrder.isSetINT96_TIMESTAMP_ORDER()) {
+ columnOrder.setTYPE_ORDER(new TypeDefinedOrder());
+ }
+ }
+ });
+ }
+
+ @Test
+ public void testReaderIgnoresInt96StatsWhenColumnOrdersAbsent() throws IOException {
+ assertLegacyInt96StatsIgnored("absent", FileMetaData::unsetColumn_orders);
+ }
+
+ private static byte[] toArray(ByteBuffer buffer) {
+ byte[] bytes = new byte[buffer.remaining()];
+ buffer.duplicate().get(bytes);
+ return bytes;
+ }
+}