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 35ef0ec9d2..6f8341b24a 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,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) { 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 9d22d25312..360caaa5f3 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 @@ -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_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 81f2781cde..1a4566e21b 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 T convert(PrimitiveTypeNameConverter 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( 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 8d85f3b84f..d91a401111 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 @@ -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 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 ec7425141e..b9600dc9bd 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 @@ 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,8 @@ import java.nio.ByteBuffer; import java.util.ArrayList; import java.util.List; +import org.apache.parquet.TestUtils; +import org.apache.parquet.example.data.simple.NanoTime; import org.apache.parquet.io.api.Binary; import org.junit.Test; @@ -354,6 +357,49 @@ 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); + TestUtils.assertThrows( + "Expected IllegalArgumentException for nanos=" + invalidNanos, + IllegalArgumentException.class, + () -> BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(valid, invalid)); + } + } + @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 50c2e344e2..2bd21fe758 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 @@ -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 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(); } 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 8d778f7b91..2b0706fa2c 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 @@ -96,6 +96,7 @@ 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; @@ -1179,9 +1180,12 @@ public void testSkippedV2Stats() { 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) { @@ -1220,6 +1224,10 @@ public void testV2OnlyStats() { .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) { @@ -1258,10 +1266,11 @@ public void testV2StatsEqualMinMax() { .named(""), new BigInteger("-8752832"), new BigInteger("-8752832")); + NanoTime int96 = new NanoTime(2458850, 43_200_000_000_000L); // 2020-01-01T12:00:00 testV2StatsEqualMinMax( Types.optional(PrimitiveTypeName.INT96).named(""), - new BigInteger("81032984"), - new BigInteger("81032984")); + int96, + int96); } private void testV2StatsEqualMinMax(PrimitiveType type, Object min, Object max) { @@ -1281,6 +1290,8 @@ private static Statistics createStats(PrimitiveType type, T min, T max) { 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; @@ -1315,6 +1326,17 @@ private static Statistics createStatsTyped(PrimitiveType type, BigInteger min 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); + assertEquals(minBinary, stats.genericGetMin()); + assertEquals(maxBinary, stats.genericGetMax()); + return stats; + } + private static ParquetMetadata createParquetMetaData(Encoding dicEncoding, Encoding dataEncoding) { return createParquetMetaData(dicEncoding, dataEncoding, true); } @@ -1383,8 +1405,7 @@ public void testColumnOrders() throws IOException { + " 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 + " }" + " }" + " }" @@ -1398,9 +1419,10 @@ public void testColumnOrders() throws IOException { List columnOrders = formatMetadata.getColumn_orders(); assertEquals(3, columnOrders.size()); - for (org.apache.parquet.format.ColumnOrder columnOrder : columnOrders) { - assertTrue(columnOrder.isSetTYPE_ORDER()); - } + // binary_col and key get TYPE_ORDER, the INT96 array_element gets INT96_TIMESTAMP_ORDER. + assertTrue(columnOrders.get(0).isSetTYPE_ORDER()); + assertTrue(columnOrders.get(1).isSetTYPE_ORDER()); + assertTrue(columnOrders.get(2).isSetINT96_TIMESTAMP_ORDER()); // Simulate that thrift got a union type that is not in the generated code // (when the file contains a not-yet-supported column order) @@ -1413,7 +1435,44 @@ public void testColumnOrders() throws IOException { assertEquals( ColumnOrder.typeDefined(), columns.get(0).getPrimitiveType().columnOrder()); assertEquals(ColumnOrder.undefined(), columns.get(1).getPrimitiveType().columnOrder()); - assertEquals(ColumnOrder.undefined(), columns.get(2).getPrimitiveType().columnOrder()); + assertEquals(ColumnOrder.int96TimestampOrder(), columns.get(2).getPrimitiveType().columnOrder()); + } + + @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(), null); + ParquetMetadata metadata = new ParquetMetadata(fileMetaData, new ArrayList()); + ParquetMetadataConverter converter = new ParquetMetadataConverter(); + FileMetaData formatMetadata = converter.toParquetMetadata(1, metadata); + + if (footerColumnOrder == null) { + formatMetadata.unsetColumn_orders(); + assertFalse(formatMetadata.isSetColumn_orders()); + } else { + formatMetadata.setColumn_orders(Collections.singletonList(footerColumnOrder)); + } + + MessageType resultSchema = + converter.fromParquetMetadata(formatMetadata).getFileMetaData().getSchema(); + List columns = resultSchema.getColumns(); + assertEquals(1, columns.size()); + assertEquals(ColumnOrder.undefined(), columns.get(0).getPrimitiveType().columnOrder()); } @Test @@ -1488,7 +1547,11 @@ public void testColumnIndexConversion() { assertNull( "Should ignore unsupported types", 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)); assertNull( "Should ignore unsupported types", ParquetMetadataConverter.fromParquetColumnIndex( 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 0000000000..cabace996d --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java @@ -0,0 +1,244 @@ +/* + * 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.junit.Assert.assertArrayEquals; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertTrue; + +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.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.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +/** + * 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 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; + + @Rule + public TemporaryFolder tmp = new TemporaryFolder(); + + private File writeFile() throws IOException { + File file = tmp.newFile("int96.parquet"); + file.delete(); + SimpleGroupFactory factory = new SimpleGroupFactory(SCHEMA); + ParquetWriter 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 = new File(tmp.getRoot(), name); + 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) { + assertTrue(column.getStatistics() == null || !column.getStatistics().hasNonNullValue()); + } + + private static void assertStatsUsable(ColumnChunkMetaData column) { + assertTrue(column.getStatistics() != null && column.getStatistics().hasNonNullValue()); + assertArrayEquals(EXPECTED_MIN.getBytes(), column.getStatistics().getMinBytes()); + assertArrayEquals(EXPECTED_MAX.getBytes(), column.getStatistics().getMaxBytes()); + } + + @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 + assertTrue(rawFooter.getColumn_orders().get(0).isSetINT96_TIMESTAMP_ORDER()); + + // Statistics should be present for the INT96 column. + ColumnChunk tsChunk = rawFooter.getRow_groups().get(0).getColumns().get(0); + assertEquals(Type.INT96, tsChunk.getMeta_data().getType()); + Statistics stats = tsChunk.getMeta_data().getStatistics(); + assertTrue(stats != null); + assertArrayEquals(EXPECTED_MIN.getBytes(), stats.getMin_value()); + assertArrayEquals(EXPECTED_MAX.getBytes(), stats.getMax_value()); + + // 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")); + assertNotNull(columnIndex); + assertArrayEquals(EXPECTED_MIN.getBytes(), toArray(columnIndex.getMinValues().get(0))); + assertArrayEquals(EXPECTED_MAX.getBytes(), toArray(columnIndex.getMaxValues().get(0))); + } + } + + @Test + public void testReaderReadsStatsWrittenWithInt96TimestampOrder() throws IOException { + File file = writeFile(); + ParquetMetadata footer = readFooter(file); + + PrimitiveType ts = footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType(); + assertEquals(ColumnOrder.int96TimestampOrder(), ts.columnOrder()); + assertEquals("BINARY_AS_INT96_TIMESTAMP_COMPARATOR", ts.comparator().toString()); + + assertStatsUsable(getColumn(footer, "ts")); + } + + private void assertLegacyInt96StatsIgnored(String name, Consumer 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); + assertEquals(ColumnOrder.undefined(), + footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType().columnOrder()); + assertStatsIgnored(getColumn(footer, "ts")); + // The non-INT96 sibling column is unaffected. + assertTrue(getColumn(footer, "id").getStatistics().hasNonNullValue()); + } + + @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; + } +}