From c186f73ff345f68399444053096a21d5b5eb9587 Mon Sep 17 00:00:00 2001 From: Divjot Arora Date: Thu, 11 Jun 2026 07:48:04 +0000 Subject: [PATCH 1/6] Add new sort order for int96 timestamps --- .../parquet/column/ParquetProperties.java | 21 ++ .../apache/parquet/schema/ColumnOrder.java | 15 + .../parquet/schema/PrimitiveComparator.java | 33 ++ .../apache/parquet/schema/PrimitiveType.java | 12 +- .../schema/TestPrimitiveComparator.java | 59 +++ .../apache/parquet/ParquetReadOptions.java | 5 + .../converter/ParquetMetadataConverter.java | 27 +- .../hadoop/InternalParquetRecordWriter.java | 2 +- .../parquet/hadoop/ParquetFileWriter.java | 35 +- .../parquet/hadoop/ParquetInputFormat.java | 6 + .../parquet/hadoop/ParquetOutputFormat.java | 12 +- .../apache/parquet/hadoop/ParquetWriter.java | 12 + .../TestInt96TimestampStatistics.java | 340 ++++++++++++++++++ 13 files changed, 571 insertions(+), 8 deletions(-) create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java diff --git a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java index f29214b458..47a9585d13 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java @@ -66,6 +66,7 @@ public class ParquetProperties { public static final int DEFAULT_BLOOM_FILTER_CANDIDATES_NUMBER = 5; public static final boolean DEFAULT_STATISTICS_ENABLED = true; public static final boolean DEFAULT_SIZE_STATISTICS_ENABLED = true; + public static final boolean DEFAULT_INT96_TIMESTAMP_STATISTICS_ENABLED = false; public static final boolean DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED = true; @@ -120,6 +121,7 @@ public static WriterVersion fromString(String name) { private final int statisticsTruncateLength; private final boolean statisticsEnabled; private final boolean sizeStatisticsEnabled; + private final boolean int96TimestampStatisticsEnabled; // The expected NDV (number of distinct values) for each columns private final ColumnProperty bloomFilterNDVs; @@ -154,6 +156,7 @@ private ParquetProperties(Builder builder) { this.statisticsTruncateLength = builder.statisticsTruncateLength; this.statisticsEnabled = builder.statisticsEnabled; this.sizeStatisticsEnabled = builder.sizeStatisticsEnabled; + this.int96TimestampStatisticsEnabled = builder.int96TimestampStatisticsEnabled; this.bloomFilterNDVs = builder.bloomFilterNDVs.build(); this.bloomFilterFPPs = builder.bloomFilterFPPs.build(); this.bloomFilterEnabled = builder.bloomFilterEnabled.build(); @@ -370,6 +373,10 @@ public boolean getSizeStatisticsEnabled(ColumnDescriptor column) { return sizeStatisticsEnabled; } + public boolean getInt96TimestampStatisticsEnabled() { + return int96TimestampStatisticsEnabled; + } + @Override public String toString() { return "Parquet page size to " + getPageSizeThreshold() + '\n' @@ -406,6 +413,7 @@ public static class Builder { private int statisticsTruncateLength = DEFAULT_STATISTICS_TRUNCATE_LENGTH; private boolean statisticsEnabled = DEFAULT_STATISTICS_ENABLED; private boolean sizeStatisticsEnabled = DEFAULT_SIZE_STATISTICS_ENABLED; + private boolean int96TimestampStatisticsEnabled = DEFAULT_INT96_TIMESTAMP_STATISTICS_ENABLED; private final ColumnProperty.Builder bloomFilterNDVs; private final ColumnProperty.Builder bloomFilterFPPs; private int maxBloomFilterBytes = DEFAULT_MAX_BLOOM_FILTER_BYTES; @@ -756,6 +764,19 @@ public Builder withSizeStatisticsEnabled(String columnPath, boolean enabled) { return this; } + /** + * Sets whether min/max statistics are collected and written for INT96 columns using the + * chronological INT96_TIMESTAMP_ORDER column order (disabled by default). When enabled, INT96 + * columns are tagged with INT96_TIMESTAMP_ORDER in the file footer. + * + * @param enabled whether to collect and write INT96 timestamp statistics + * @return this builder for method chaining + */ + public Builder withInt96TimestampStatisticsEnabled(boolean enabled) { + this.int96TimestampStatisticsEnabled = enabled; + return this; + } + public ParquetProperties build() { ParquetProperties properties = new ParquetProperties(this); // we pass a constructed but uninitialized factory to ParquetProperties above as currently 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..625e5aea1c 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,18 @@ public enum ColumnOrderName { * The column order is defined by the IEEE 754 standard. */ IEEE_754_TOTAL_ORDER, + /** + * Chronological order for INT96 timestamps: values are compared by the Julian day (the last 4 + * bytes, as a little-endian signed int32), then by the nanoseconds within the day (the first 8 + * bytes, as a little-endian signed int64). Only supported for the INT96 physical type. + */ + 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 +78,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..113694f6a6 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,36 @@ 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() { + @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; + return Long.compare(bb1.getLong(0), bb2.getLong(0)); + } + + @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..6936b8e9c0 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) { @@ -651,9 +653,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/schema/TestPrimitiveComparator.java b/parquet-column/src/test/java/org/apache/parquet/schema/TestPrimitiveComparator.java index ec7425141e..65f56a6864 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; @@ -36,8 +37,12 @@ import java.math.BigInteger; import java.nio.ByteBuffer; +import java.time.LocalDateTime; import java.util.ArrayList; +import java.util.Arrays; import java.util.List; +import java.util.function.Function; +import org.apache.parquet.example.data.simple.NanoTime; import org.apache.parquet.io.api.Binary; import org.junit.Test; @@ -354,6 +359,60 @@ public void testBinaryAsSignedIntegerComparatorWithEquals() { } } + private static Binary int96(int julianDay, long nanosOfDay) { + return new NanoTime(julianDay, nanosOfDay).toBinary(); + } + + private static Binary timestampToInt96(String timestamp) { + LocalDateTime dt = LocalDateTime.parse(timestamp); + int julianDay = (int) (dt.toLocalDate().toEpochDay() + 2440588); + return new NanoTime(julianDay, dt.toLocalTime().toNanoOfDay()).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 + timestampToInt96("1968-05-23T00:00:00.000000123"), // pre-epoch but positive julian day + timestampToInt96("2020-01-01T12:00:00"), + timestampToInt96("2020-02-01T11:00:00"), // later day even though earlier time of day + timestampToInt96("2020-02-01T11:00:00.000000001"), // nanos tie-break + int96(Integer.MAX_VALUE, 86_399_999_999_999L) + }; + + // The same value in different Binary representations must compare identically; the offset + // variant guards against absolute reads not being relative to the value's start + List> representations = List.of( + b -> b, + b -> Binary.fromReusedByteArray(b.getBytes()), + b -> Binary.fromConstantByteArray(b.getBytes()), + b -> { + byte[] bytes = b.getBytes(); + byte[] padded = new byte[bytes.length + 20]; + Arrays.fill(padded, (byte) 0xAA); + System.arraycopy(bytes, 0, padded, 10, bytes.length); + return Binary.fromReusedByteArray(padded, 10, bytes.length); + }); + + for (int i = 0; i < valuesInAscendingOrder.length; ++i) { + for (int j = 0; j < valuesInAscendingOrder.length; ++j) { + for (Function fi : representations) { + for (Function fj : representations) { + Binary bi = fi.apply(valuesInAscendingOrder[i]); + Binary bj = fj.apply(valuesInAscendingOrder[j]); + assertEquals( + "comparing value " + i + " to value " + j, + Integer.signum(Integer.compare(i, j)), + Integer.signum(BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(bi, bj))); + } + } + } + } + } + @Test public void testFloat16Comparator() { Binary[] valuesInAscendingOrder = { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java b/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java index 895d0670fa..cdeadddc9c 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java @@ -25,6 +25,7 @@ import static org.apache.parquet.hadoop.ParquetInputFormat.DICTIONARY_FILTERING_ENABLED; import static org.apache.parquet.hadoop.ParquetInputFormat.HADOOP_VECTORED_IO_DEFAULT; import static org.apache.parquet.hadoop.ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED; +import static org.apache.parquet.hadoop.ParquetInputFormat.INT96_TIMESTAMP_STATISTICS_READING_ENABLED; import static org.apache.parquet.hadoop.ParquetInputFormat.OFF_HEAP_DECRYPT_BUFFER_ENABLED; import static org.apache.parquet.hadoop.ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED; import static org.apache.parquet.hadoop.ParquetInputFormat.RECORD_FILTERING_ENABLED; @@ -291,6 +292,10 @@ public Builder(ParquetConfiguration conf) { if (badRecordThresh != null) { set(BAD_RECORD_THRESHOLD_CONF_KEY, badRecordThresh); } + String readInt96TimestampStats = conf.get(INT96_TIMESTAMP_STATISTICS_READING_ENABLED); + if (readInt96TimestampStats != null) { + set(INT96_TIMESTAMP_STATISTICS_READING_ENABLED, readInt96TimestampStats); + } } public Builder useSignedStringMinMax(boolean useSignedStringMinMax) { 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..ffb44701a2 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; @@ -113,6 +114,7 @@ import org.apache.parquet.format.Uncompressed; import org.apache.parquet.format.VariantType; import org.apache.parquet.format.XxHash; +import org.apache.parquet.hadoop.ParquetInputFormat; import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.ColumnPath; @@ -146,6 +148,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 +293,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 +917,10 @@ 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; + ColumnOrderName name = type.columnOrder().getColumnOrderName(); + return name == ColumnOrderName.TYPE_DEFINED_ORDER + || name == ColumnOrderName.IEEE_754_TOTAL_ORDER + || name == ColumnOrderName.INT96_TIMESTAMP_ORDER; } /** @@ -2057,6 +2065,11 @@ 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); } childBuilder = primitiveBuilder; @@ -2103,17 +2116,25 @@ Repetition fromParquetRepetition(FieldRepetitionType repetition) { return Repetition.valueOf(repetition.name()); } - private static org.apache.parquet.schema.ColumnOrder fromParquetColumnOrder(ColumnOrder columnOrder) { + private org.apache.parquet.schema.ColumnOrder fromParquetColumnOrder(ColumnOrder columnOrder) { if (columnOrder.isSetTYPE_ORDER()) { return org.apache.parquet.schema.ColumnOrder.typeDefined(); } if (columnOrder.isSetIEEE_754_TOTAL_ORDER()) { return org.apache.parquet.schema.ColumnOrder.ieee754TotalOrder(); } + if (columnOrder.isSetINT96_TIMESTAMP_ORDER() && readInt96TimestampStatisticsEnabled()) { + return org.apache.parquet.schema.ColumnOrder.int96TimestampOrder(); + } // The column order is not yet supported by this API return org.apache.parquet.schema.ColumnOrder.undefined(); } + private boolean readInt96TimestampStatisticsEnabled() { + return options == null + || options.isEnabled(ParquetInputFormat.INT96_TIMESTAMP_STATISTICS_READING_ENABLED, true); + } + @Deprecated public void writeDataPageHeader( int uncompressedSize, diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index dd51d1ef09..facd888f76 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -89,7 +89,7 @@ public InternalParquetRecordWriter( ParquetProperties props) { this.parquetFileWriter = parquetFileWriter; this.writeSupport = Objects.requireNonNull(writeSupport, "writeSupport cannot be null"); - this.schema = schema; + this.schema = ParquetFileWriter.applyInt96TimestampOrder(schema, props); this.extraMetaData = extraMetaData; this.rowGroupSizeThreshold = rowGroupSize; this.rowGroupRecordCountThreshold = props.getRowGroupRowCountLimit(); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java index 82f4577b83..dd158ba011 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java @@ -92,8 +92,12 @@ import org.apache.parquet.io.ParquetEncodingException; import org.apache.parquet.io.PositionOutputStream; import org.apache.parquet.io.SeekableInputStream; +import org.apache.parquet.schema.ColumnOrder; +import org.apache.parquet.schema.GroupType; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType; +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; +import org.apache.parquet.schema.Type; import org.apache.parquet.schema.TypeUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -435,7 +439,7 @@ public ParquetFileWriter( throws IOException { this( file, - schema, + applyInt96TimestampOrder(schema, props), mode, rowGroupSize, maxPaddingSize, @@ -447,6 +451,35 @@ public ParquetFileWriter( props.getAllocator()); } + /** + * Returns the schema with INT96 columns tagged with the INT96_TIMESTAMP_ORDER column order if + * INT96 timestamp statistics are enabled, so that statistics are accumulated with the + * chronological comparator and the proper column order is written to the footer. + */ + static MessageType applyInt96TimestampOrder(MessageType schema, ParquetProperties props) { + if (!props.getInt96TimestampStatisticsEnabled()) { + return schema; + } + return new MessageType(schema.getName(), applyInt96TimestampOrder(schema.getFields())); + } + + private static List applyInt96TimestampOrder(List fields) { + List result = new ArrayList<>(fields.size()); + for (Type field : fields) { + if (field.isPrimitive()) { + PrimitiveType primitive = field.asPrimitiveType(); + if (primitive.getPrimitiveTypeName() == PrimitiveTypeName.INT96) { + field = primitive.withColumnOrder(ColumnOrder.int96TimestampOrder()); + } + result.add(field); + } else { + GroupType group = field.asGroupType(); + result.add(group.withNewFields(applyInt96TimestampOrder(group.getFields()))); + } + } + return result; + } + @Deprecated public ParquetFileWriter( OutputFile file, diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java index 8e05d49bd3..95f9796a4e 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java @@ -122,6 +122,12 @@ public class ParquetInputFormat extends FileInputFormat { */ public static final String STATS_FILTERING_ENABLED = "parquet.filter.stats.enabled"; + /** + * key to configure whether INT96 min/max statistics written with the INT96_TIMESTAMP_ORDER + * column order are read (enabled by default) + */ + public static final String INT96_TIMESTAMP_STATISTICS_READING_ENABLED = "parquet.int96.timestamp.statistics.read.enabled"; + /** * key to configure whether row group dictionary filtering is enabled */ diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java index 868ae634c1..ba0a6a5924 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java @@ -163,6 +163,7 @@ public static enum JobSummaryLevel { public static final String PAGE_WRITE_CHECKSUM_ENABLED = "parquet.page.write-checksum.enabled"; public static final String STATISTICS_ENABLED = "parquet.column.statistics.enabled"; public static final String SIZE_STATISTICS_ENABLED = "parquet.size.statistics.enabled"; + public static final String INT96_TIMESTAMP_STATISTICS_ENABLED = "parquet.int96.timestamp.statistics.enabled"; public static JobSummaryLevel getJobSummaryLevel(Configuration conf) { String level = conf.get(JOB_SUMMARY_LEVEL); @@ -432,6 +433,14 @@ public static boolean getStatisticsEnabled(Configuration conf, String columnPath return conf.getBoolean(STATISTICS_ENABLED, ParquetProperties.DEFAULT_STATISTICS_ENABLED); } + public static void setInt96TimestampStatisticsEnabled(JobContext jobContext, boolean enabled) { + getConfiguration(jobContext).setBoolean(INT96_TIMESTAMP_STATISTICS_ENABLED, enabled); + } + + public static boolean getInt96TimestampStatisticsEnabled(Configuration conf) { + return conf.getBoolean(INT96_TIMESTAMP_STATISTICS_ENABLED, ParquetProperties.DEFAULT_INT96_TIMESTAMP_STATISTICS_ENABLED); + } + public static void setSizeStatisticsEnabled(Configuration conf, boolean enabled) { conf.setBoolean(SIZE_STATISTICS_ENABLED, enabled); } @@ -526,7 +535,8 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp .withRowGroupRowCountLimit(getBlockRowCountLimit(conf)) .withPageRowCountLimit(getPageRowCountLimit(conf)) .withPageWriteChecksumEnabled(getPageWriteChecksumEnabled(conf)) - .withStatisticsEnabled(getStatisticsEnabled(conf)); + .withStatisticsEnabled(getStatisticsEnabled(conf)) + .withInt96TimestampStatisticsEnabled(getInt96TimestampStatisticsEnabled(conf)); new ColumnConfigParser() .withColumnConfig( ENABLE_DICTIONARY, key -> conf.getBoolean(key, false), propsBuilder::withDictionaryEncoding) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java index 8eb5f7f17b..afa06065c4 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java @@ -952,6 +952,18 @@ public SELF withStatisticsEnabled(boolean enabled) { return self(); } + /** + * Sets whether min/max statistics are collected and written for INT96 columns using the + * chronological INT96_TIMESTAMP_ORDER column order (disabled by default). + * + * @param enabled whether to collect and write INT96 timestamp statistics + * @return this builder for method chaining + */ + public SELF withInt96TimestampStatisticsEnabled(boolean enabled) { + encodingPropsBuilder.withInt96TimestampStatisticsEnabled(enabled); + return self(); + } + /** * Sets the size statistics enabled/disabled for the specified column. All column size statistics are enabled by default. * 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..8345304e80 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java @@ -0,0 +1,340 @@ +/* + * 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.assertNull; +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 org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.HadoopReadOptions; +import org.apache.parquet.example.data.Group; +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.RowGroup; +import org.apache.parquet.format.Statistics; +import org.apache.parquet.format.Type; +import org.apache.parquet.format.Util; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.ParquetFileWriter; +import org.apache.parquet.hadoop.ParquetInputFormat; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.hadoop.example.GroupWriteSupport; +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.hadoop.util.HadoopInputFile; +import org.apache.parquet.internal.column.columnindex.ColumnIndex; +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 = int96(2440000, 123L); // 1968-05-23 00:00:00.000000123 + private static final Binary SAME_DAY_EARLY = int96(2440588, 1_000L); // 1970-01-01 00:00:00.000001 + private static final Binary LATE_IN_DAY = int96(2440588, 86_399_999_999_999L); // 1970-01-01 23:59:59.999... + private static final Binary NEXT_DAY = int96(2440589, 0L); // 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 static Binary int96(int julianDay, long nanosOfDay) { + return Binary.fromConstantByteArray(ByteBuffer.allocate(12) + .order(ByteOrder.LITTLE_ENDIAN) + .putLong(nanosOfDay) + .putInt(julianDay) + .array()); + } + + private File writeFile(boolean int96StatsEnabled) throws IOException { + File file = new File(tmp.getRoot(), "int96_" + int96StatsEnabled + ".parquet"); + Configuration conf = new Configuration(); + GroupWriteSupport.setSchema(SCHEMA, conf); + SimpleGroupFactory factory = new SimpleGroupFactory(SCHEMA); + ParquetWriter writer = ExampleParquetWriter.builder(new Path(file.getAbsolutePath())) + .withConf(conf) + .withInt96TimestampStatisticsEnabled(int96StatsEnabled) + .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, Configuration conf) throws IOException { + Path path = new Path(file.getAbsolutePath()); + try (ParquetFileReader reader = ParquetFileReader.open( + HadoopInputFile.fromPath(path, conf), + HadoopReadOptions.builder(conf, path).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)); + } + + /** Rewrites the footer of src into a new file, keeping all data pages byte-identical. */ + private File rewriteFooter(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; + } + + /** + * Injects min/max statistics for the INT96 column into the raw footer while keeping TYPE_ORDER + * as the column order. This simulates a legacy writer that emitted INT96 stats before + * INT96_TIMESTAMP_ORDER existed. + */ + private static void injectInt96Stats(FileMetaData footer) { + // No need to touch column_orders: the writer already stamps TYPE_ORDER for every column when + // INT96 timestamp statistics are disabled + for (RowGroup rowGroup : footer.getRow_groups()) { + rowGroup.getColumns().stream() + .map(c -> c.getMeta_data()) + .filter(md -> md.getType() == Type.INT96) + .forEach(md -> { + Statistics stats = new Statistics(); + stats.setMin_value(EXPECTED_MIN.getBytes()); + stats.setMax_value(EXPECTED_MAX.getBytes()); + stats.setNull_count(0); + md.setStatistics(stats); + }); + } + } + + 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 testWriterOmitsInt96StatsByDefault() throws IOException { + File file = writeFile(false); + FileMetaData rawFooter = readRawFooter(file); + for (RowGroup rowGroup : rawFooter.getRow_groups()) { + for (ColumnChunk chunk : rowGroup.getColumns()) { + if (chunk.getMeta_data().getType() == Type.INT96) { + Statistics stats = chunk.getMeta_data().getStatistics(); + assertTrue(stats == null || (!stats.isSetMin_value() && !stats.isSetMax_value())); + } + } + } + + // Without the new order, the column order written for INT96 stays TYPE_ORDER + assertTrue(rawFooter.getColumn_orders().get(0).isSetTYPE_ORDER()); + assertStatsIgnored(getColumn(readFooter(file, new Configuration()), "ts")); + + // Without the new order, column index must not be written for the INT96 column. + Configuration conf = new Configuration(); + Path path = new Path(file.getAbsolutePath()); + try ( + ParquetFileReader reader = ParquetFileReader.open( + HadoopInputFile.fromPath(path, conf), HadoopReadOptions.builder(conf, path).build() + ) + ) { + assertNotNull(reader.readColumnIndex(getColumn(reader.getFooter(), "id"))); + assertNull(reader.readColumnIndex(getColumn(reader.getFooter(), "ts"))); + } + } + + @Test + public void testWriterEmitsInt96StatsAndColumnOrderWhenEnabled() throws IOException { + File file = writeFile(true); + 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()); + assertTrue(rawFooter.getColumn_orders().get(1).isSetTYPE_ORDER()); + + for (RowGroup rowGroup : rawFooter.getRow_groups()) { + for (ColumnChunk chunk : rowGroup.getColumns()) { + if (chunk.getMeta_data().getType() == Type.INT96) { + Statistics stats = chunk.getMeta_data().getStatistics(); + assertTrue(stats != null && stats.isSetMin_value()); + assertArrayEquals(EXPECTED_MIN.getBytes(), stats.getMin_value()); + assertArrayEquals(EXPECTED_MAX.getBytes(), stats.getMax_value()); + } + } + } + + // Column index should be present for both columns. + Configuration conf = new Configuration(); + Path path = new Path(file.getAbsolutePath()); + try ( + ParquetFileReader reader = ParquetFileReader.open( + HadoopInputFile.fromPath(path, conf), HadoopReadOptions.builder(conf, path).build() + ) + ) { + assertNotNull(reader.readColumnIndex(getColumn(reader.getFooter(), "id"))); + + 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(columnIndex.getMaxValues().size() - 1))); + } + } + + @Test + public void testStatsAccumulationUsesChronologicalOrder() 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(true); + ParquetMetadata footer = readFooter(file, new Configuration()); + byte[] minBytes = getColumn(footer, "ts").getStatistics().getMinBytes(); + assertFalse(Binary.fromConstantByteArray(minBytes).equals(NEXT_DAY)); + assertArrayEquals(EXPECTED_MIN.getBytes(), minBytes); + assertArrayEquals(EXPECTED_MAX.getBytes(), + getColumn(footer, "ts").getStatistics().getMaxBytes()); + } + + @Test + public void testReaderReadsStatsWrittenWithInt96TimestampOrder() throws IOException { + File file = writeFile(true); + ParquetMetadata footer = readFooter(file, new Configuration()); + + 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")); + } + + @Test + public void testReaderReadsStatsWrittenWithInt96TimestampOrderWhenDisabled() throws IOException { + File file = writeFile(true); + + Configuration conf = new Configuration(); + conf.setBoolean(ParquetInputFormat.INT96_TIMESTAMP_STATISTICS_READING_ENABLED, false); + ParquetMetadata footer = readFooter(file, conf); + + assertEquals(ColumnOrder.undefined(), + footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType().columnOrder()); + assertStatsIgnored(getColumn(footer, "ts")); + } + + @Test + public void testReaderIgnoresInt96StatsWithTypeDefinedOrder() throws IOException { + // Legacy layout: stats present, but column order is TYPE_ORDER so they are ignored. + File file = writeFile(false); + FileMetaData rawFooter = readRawFooter(file); + injectInt96Stats(rawFooter); + File legacy = rewriteFooter(file, rawFooter, "legacy.parquet"); + + // Validate the data is still intact after rewriting the footer. + try ( + ParquetReader reader = ParquetReader.builder( + new GroupReadSupport(), new Path(legacy.getAbsolutePath()) + ).build() + ) { + for (int i = 0; i < VALUES.size(); i++) { + Group group = reader.read(); + assertEquals(VALUES.get(i), group.getInt96("ts", 0)); + assertEquals(i, group.getLong("id", 0)); + } + } + + ParquetMetadata footer = readFooter(legacy, new Configuration()); + 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()); + } + + private static byte[] toArray(ByteBuffer buffer) { + byte[] bytes = new byte[buffer.remaining()]; + buffer.duplicate().get(bytes); + return bytes; + } +} From cd8062ae1e549272b0ac28a813154652d6879fc9 Mon Sep 17 00:00:00 2001 From: Divjot Arora Date: Mon, 15 Jun 2026 12:28:37 +0000 Subject: [PATCH 2/6] remove flags --- .../parquet/column/ParquetProperties.java | 21 ----- .../apache/parquet/ParquetReadOptions.java | 5 -- .../converter/ParquetMetadataConverter.java | 10 +-- .../hadoop/InternalParquetRecordWriter.java | 2 +- .../parquet/hadoop/ParquetFileWriter.java | 13 ++- .../parquet/hadoop/ParquetInputFormat.java | 6 -- .../parquet/hadoop/ParquetOutputFormat.java | 12 +-- .../apache/parquet/hadoop/ParquetWriter.java | 12 --- .../TestInt96TimestampStatistics.java | 88 ++++--------------- 9 files changed, 26 insertions(+), 143 deletions(-) diff --git a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java index 47a9585d13..f29214b458 100644 --- a/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java +++ b/parquet-column/src/main/java/org/apache/parquet/column/ParquetProperties.java @@ -66,7 +66,6 @@ public class ParquetProperties { public static final int DEFAULT_BLOOM_FILTER_CANDIDATES_NUMBER = 5; public static final boolean DEFAULT_STATISTICS_ENABLED = true; public static final boolean DEFAULT_SIZE_STATISTICS_ENABLED = true; - public static final boolean DEFAULT_INT96_TIMESTAMP_STATISTICS_ENABLED = false; public static final boolean DEFAULT_PAGE_WRITE_CHECKSUM_ENABLED = true; @@ -121,7 +120,6 @@ public static WriterVersion fromString(String name) { private final int statisticsTruncateLength; private final boolean statisticsEnabled; private final boolean sizeStatisticsEnabled; - private final boolean int96TimestampStatisticsEnabled; // The expected NDV (number of distinct values) for each columns private final ColumnProperty bloomFilterNDVs; @@ -156,7 +154,6 @@ private ParquetProperties(Builder builder) { this.statisticsTruncateLength = builder.statisticsTruncateLength; this.statisticsEnabled = builder.statisticsEnabled; this.sizeStatisticsEnabled = builder.sizeStatisticsEnabled; - this.int96TimestampStatisticsEnabled = builder.int96TimestampStatisticsEnabled; this.bloomFilterNDVs = builder.bloomFilterNDVs.build(); this.bloomFilterFPPs = builder.bloomFilterFPPs.build(); this.bloomFilterEnabled = builder.bloomFilterEnabled.build(); @@ -373,10 +370,6 @@ public boolean getSizeStatisticsEnabled(ColumnDescriptor column) { return sizeStatisticsEnabled; } - public boolean getInt96TimestampStatisticsEnabled() { - return int96TimestampStatisticsEnabled; - } - @Override public String toString() { return "Parquet page size to " + getPageSizeThreshold() + '\n' @@ -413,7 +406,6 @@ public static class Builder { private int statisticsTruncateLength = DEFAULT_STATISTICS_TRUNCATE_LENGTH; private boolean statisticsEnabled = DEFAULT_STATISTICS_ENABLED; private boolean sizeStatisticsEnabled = DEFAULT_SIZE_STATISTICS_ENABLED; - private boolean int96TimestampStatisticsEnabled = DEFAULT_INT96_TIMESTAMP_STATISTICS_ENABLED; private final ColumnProperty.Builder bloomFilterNDVs; private final ColumnProperty.Builder bloomFilterFPPs; private int maxBloomFilterBytes = DEFAULT_MAX_BLOOM_FILTER_BYTES; @@ -764,19 +756,6 @@ public Builder withSizeStatisticsEnabled(String columnPath, boolean enabled) { return this; } - /** - * Sets whether min/max statistics are collected and written for INT96 columns using the - * chronological INT96_TIMESTAMP_ORDER column order (disabled by default). When enabled, INT96 - * columns are tagged with INT96_TIMESTAMP_ORDER in the file footer. - * - * @param enabled whether to collect and write INT96 timestamp statistics - * @return this builder for method chaining - */ - public Builder withInt96TimestampStatisticsEnabled(boolean enabled) { - this.int96TimestampStatisticsEnabled = enabled; - return this; - } - public ParquetProperties build() { ParquetProperties properties = new ParquetProperties(this); // we pass a constructed but uninitialized factory to ParquetProperties above as currently diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java b/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java index cdeadddc9c..895d0670fa 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/ParquetReadOptions.java @@ -25,7 +25,6 @@ import static org.apache.parquet.hadoop.ParquetInputFormat.DICTIONARY_FILTERING_ENABLED; import static org.apache.parquet.hadoop.ParquetInputFormat.HADOOP_VECTORED_IO_DEFAULT; import static org.apache.parquet.hadoop.ParquetInputFormat.HADOOP_VECTORED_IO_ENABLED; -import static org.apache.parquet.hadoop.ParquetInputFormat.INT96_TIMESTAMP_STATISTICS_READING_ENABLED; import static org.apache.parquet.hadoop.ParquetInputFormat.OFF_HEAP_DECRYPT_BUFFER_ENABLED; import static org.apache.parquet.hadoop.ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED; import static org.apache.parquet.hadoop.ParquetInputFormat.RECORD_FILTERING_ENABLED; @@ -292,10 +291,6 @@ public Builder(ParquetConfiguration conf) { if (badRecordThresh != null) { set(BAD_RECORD_THRESHOLD_CONF_KEY, badRecordThresh); } - String readInt96TimestampStats = conf.get(INT96_TIMESTAMP_STATISTICS_READING_ENABLED); - if (readInt96TimestampStats != null) { - set(INT96_TIMESTAMP_STATISTICS_READING_ENABLED, readInt96TimestampStats); - } } public Builder useSignedStringMinMax(boolean useSignedStringMinMax) { 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 ffb44701a2..5a9ce9ab98 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 @@ -114,7 +114,6 @@ import org.apache.parquet.format.Uncompressed; import org.apache.parquet.format.VariantType; import org.apache.parquet.format.XxHash; -import org.apache.parquet.hadoop.ParquetInputFormat; import org.apache.parquet.hadoop.metadata.BlockMetaData; import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.ColumnPath; @@ -2116,25 +2115,20 @@ Repetition fromParquetRepetition(FieldRepetitionType repetition) { return Repetition.valueOf(repetition.name()); } - private org.apache.parquet.schema.ColumnOrder fromParquetColumnOrder(ColumnOrder columnOrder) { + private static org.apache.parquet.schema.ColumnOrder fromParquetColumnOrder(ColumnOrder columnOrder) { if (columnOrder.isSetTYPE_ORDER()) { return org.apache.parquet.schema.ColumnOrder.typeDefined(); } if (columnOrder.isSetIEEE_754_TOTAL_ORDER()) { return org.apache.parquet.schema.ColumnOrder.ieee754TotalOrder(); } - if (columnOrder.isSetINT96_TIMESTAMP_ORDER() && readInt96TimestampStatisticsEnabled()) { + 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(); } - private boolean readInt96TimestampStatisticsEnabled() { - return options == null - || options.isEnabled(ParquetInputFormat.INT96_TIMESTAMP_STATISTICS_READING_ENABLED, true); - } - @Deprecated public void writeDataPageHeader( int uncompressedSize, diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index facd888f76..3e8713ce5c 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -89,7 +89,7 @@ public InternalParquetRecordWriter( ParquetProperties props) { this.parquetFileWriter = parquetFileWriter; this.writeSupport = Objects.requireNonNull(writeSupport, "writeSupport cannot be null"); - this.schema = ParquetFileWriter.applyInt96TimestampOrder(schema, props); + this.schema = ParquetFileWriter.applyInt96TimestampOrder(schema); this.extraMetaData = extraMetaData; this.rowGroupSizeThreshold = rowGroupSize; this.rowGroupRecordCountThreshold = props.getRowGroupRowCountLimit(); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java index dd158ba011..67b296cc1c 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java @@ -439,7 +439,7 @@ public ParquetFileWriter( throws IOException { this( file, - applyInt96TimestampOrder(schema, props), + applyInt96TimestampOrder(schema), mode, rowGroupSize, maxPaddingSize, @@ -452,14 +452,11 @@ public ParquetFileWriter( } /** - * Returns the schema with INT96 columns tagged with the INT96_TIMESTAMP_ORDER column order if - * INT96 timestamp statistics are enabled, so that statistics are accumulated with the - * chronological comparator and the proper column order is written to the footer. + * Returns the schema with INT96 columns tagged with the INT96_TIMESTAMP_ORDER column order, so + * that statistics are accumulated with the chronological comparator and the proper column order + * is written to the footer. */ - static MessageType applyInt96TimestampOrder(MessageType schema, ParquetProperties props) { - if (!props.getInt96TimestampStatisticsEnabled()) { - return schema; - } + static MessageType applyInt96TimestampOrder(MessageType schema) { return new MessageType(schema.getName(), applyInt96TimestampOrder(schema.getFields())); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java index 95f9796a4e..8e05d49bd3 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetInputFormat.java @@ -122,12 +122,6 @@ public class ParquetInputFormat extends FileInputFormat { */ public static final String STATS_FILTERING_ENABLED = "parquet.filter.stats.enabled"; - /** - * key to configure whether INT96 min/max statistics written with the INT96_TIMESTAMP_ORDER - * column order are read (enabled by default) - */ - public static final String INT96_TIMESTAMP_STATISTICS_READING_ENABLED = "parquet.int96.timestamp.statistics.read.enabled"; - /** * key to configure whether row group dictionary filtering is enabled */ diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java index ba0a6a5924..868ae634c1 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetOutputFormat.java @@ -163,7 +163,6 @@ public static enum JobSummaryLevel { public static final String PAGE_WRITE_CHECKSUM_ENABLED = "parquet.page.write-checksum.enabled"; public static final String STATISTICS_ENABLED = "parquet.column.statistics.enabled"; public static final String SIZE_STATISTICS_ENABLED = "parquet.size.statistics.enabled"; - public static final String INT96_TIMESTAMP_STATISTICS_ENABLED = "parquet.int96.timestamp.statistics.enabled"; public static JobSummaryLevel getJobSummaryLevel(Configuration conf) { String level = conf.get(JOB_SUMMARY_LEVEL); @@ -433,14 +432,6 @@ public static boolean getStatisticsEnabled(Configuration conf, String columnPath return conf.getBoolean(STATISTICS_ENABLED, ParquetProperties.DEFAULT_STATISTICS_ENABLED); } - public static void setInt96TimestampStatisticsEnabled(JobContext jobContext, boolean enabled) { - getConfiguration(jobContext).setBoolean(INT96_TIMESTAMP_STATISTICS_ENABLED, enabled); - } - - public static boolean getInt96TimestampStatisticsEnabled(Configuration conf) { - return conf.getBoolean(INT96_TIMESTAMP_STATISTICS_ENABLED, ParquetProperties.DEFAULT_INT96_TIMESTAMP_STATISTICS_ENABLED); - } - public static void setSizeStatisticsEnabled(Configuration conf, boolean enabled) { conf.setBoolean(SIZE_STATISTICS_ENABLED, enabled); } @@ -535,8 +526,7 @@ public RecordWriter getRecordWriter(Configuration conf, Path file, Comp .withRowGroupRowCountLimit(getBlockRowCountLimit(conf)) .withPageRowCountLimit(getPageRowCountLimit(conf)) .withPageWriteChecksumEnabled(getPageWriteChecksumEnabled(conf)) - .withStatisticsEnabled(getStatisticsEnabled(conf)) - .withInt96TimestampStatisticsEnabled(getInt96TimestampStatisticsEnabled(conf)); + .withStatisticsEnabled(getStatisticsEnabled(conf)); new ColumnConfigParser() .withColumnConfig( ENABLE_DICTIONARY, key -> conf.getBoolean(key, false), propsBuilder::withDictionaryEncoding) diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java index afa06065c4..8eb5f7f17b 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetWriter.java @@ -952,18 +952,6 @@ public SELF withStatisticsEnabled(boolean enabled) { return self(); } - /** - * Sets whether min/max statistics are collected and written for INT96 columns using the - * chronological INT96_TIMESTAMP_ORDER column order (disabled by default). - * - * @param enabled whether to collect and write INT96 timestamp statistics - * @return this builder for method chaining - */ - public SELF withInt96TimestampStatisticsEnabled(boolean enabled) { - encodingPropsBuilder.withInt96TimestampStatisticsEnabled(enabled); - return self(); - } - /** * Sets the size statistics enabled/disabled for the specified column. All column size statistics are enabled by default. * 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 index 8345304e80..56389694c5 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java @@ -23,7 +23,6 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; import static org.junit.Assert.assertTrue; import java.io.ByteArrayInputStream; @@ -45,10 +44,10 @@ import org.apache.parquet.format.RowGroup; 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.ParquetInputFormat; import org.apache.parquet.hadoop.ParquetReader; import org.apache.parquet.hadoop.ParquetWriter; import org.apache.parquet.hadoop.example.ExampleParquetWriter; @@ -98,14 +97,13 @@ private static Binary int96(int julianDay, long nanosOfDay) { .array()); } - private File writeFile(boolean int96StatsEnabled) throws IOException { - File file = new File(tmp.getRoot(), "int96_" + int96StatsEnabled + ".parquet"); + private File writeFile() throws IOException { + File file = new File(tmp.getRoot(), "int96.parquet"); Configuration conf = new Configuration(); GroupWriteSupport.setSchema(SCHEMA, conf); SimpleGroupFactory factory = new SimpleGroupFactory(SCHEMA); ParquetWriter writer = ExampleParquetWriter.builder(new Path(file.getAbsolutePath())) .withConf(conf) - .withInt96TimestampStatisticsEnabled(int96StatsEnabled) .build(); try { for (int i = 0; i < VALUES.size(); i++) { @@ -159,24 +157,15 @@ private File rewriteFooter(File src, FileMetaData footer, String name) } /** - * Injects min/max statistics for the INT96 column into the raw footer while keeping TYPE_ORDER - * as the column order. This simulates a legacy writer that emitted INT96 stats before - * INT96_TIMESTAMP_ORDER existed. + * Retags the INT96 column orders in the footer as TYPE_ORDER while leaving the (already written) + * min/max statistics in place. This simulates a legacy writer that emitted INT96 stats under + * TYPE_ORDER before INT96_TIMESTAMP_ORDER existed; such stats must be ignored by the reader. */ - private static void injectInt96Stats(FileMetaData footer) { - // No need to touch column_orders: the writer already stamps TYPE_ORDER for every column when - // INT96 timestamp statistics are disabled - for (RowGroup rowGroup : footer.getRow_groups()) { - rowGroup.getColumns().stream() - .map(c -> c.getMeta_data()) - .filter(md -> md.getType() == Type.INT96) - .forEach(md -> { - Statistics stats = new Statistics(); - stats.setMin_value(EXPECTED_MIN.getBytes()); - stats.setMax_value(EXPECTED_MAX.getBytes()); - stats.setNull_count(0); - md.setStatistics(stats); - }); + private static void downgradeInt96ToTypeOrder(FileMetaData footer) { + for (org.apache.parquet.format.ColumnOrder columnOrder : footer.getColumn_orders()) { + if (columnOrder.isSetINT96_TIMESTAMP_ORDER()) { + columnOrder.setTYPE_ORDER(new TypeDefinedOrder()); + } } } @@ -199,38 +188,8 @@ private static void assertStatsUsable(ColumnChunkMetaData column) { } @Test - public void testWriterOmitsInt96StatsByDefault() throws IOException { - File file = writeFile(false); - FileMetaData rawFooter = readRawFooter(file); - for (RowGroup rowGroup : rawFooter.getRow_groups()) { - for (ColumnChunk chunk : rowGroup.getColumns()) { - if (chunk.getMeta_data().getType() == Type.INT96) { - Statistics stats = chunk.getMeta_data().getStatistics(); - assertTrue(stats == null || (!stats.isSetMin_value() && !stats.isSetMax_value())); - } - } - } - - // Without the new order, the column order written for INT96 stays TYPE_ORDER - assertTrue(rawFooter.getColumn_orders().get(0).isSetTYPE_ORDER()); - assertStatsIgnored(getColumn(readFooter(file, new Configuration()), "ts")); - - // Without the new order, column index must not be written for the INT96 column. - Configuration conf = new Configuration(); - Path path = new Path(file.getAbsolutePath()); - try ( - ParquetFileReader reader = ParquetFileReader.open( - HadoopInputFile.fromPath(path, conf), HadoopReadOptions.builder(conf, path).build() - ) - ) { - assertNotNull(reader.readColumnIndex(getColumn(reader.getFooter(), "id"))); - assertNull(reader.readColumnIndex(getColumn(reader.getFooter(), "ts"))); - } - } - - @Test - public void testWriterEmitsInt96StatsAndColumnOrderWhenEnabled() throws IOException { - File file = writeFile(true); + public void testWriterEmitsInt96StatsAndColumnOrder() throws IOException { + 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()); @@ -269,7 +228,7 @@ public void testWriterEmitsInt96StatsAndColumnOrderWhenEnabled() throws IOExcept public void testStatsAccumulationUsesChronologicalOrder() 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(true); + File file = writeFile(); ParquetMetadata footer = readFooter(file, new Configuration()); byte[] minBytes = getColumn(footer, "ts").getStatistics().getMinBytes(); assertFalse(Binary.fromConstantByteArray(minBytes).equals(NEXT_DAY)); @@ -280,7 +239,7 @@ public void testStatsAccumulationUsesChronologicalOrder() throws IOException { @Test public void testReaderReadsStatsWrittenWithInt96TimestampOrder() throws IOException { - File file = writeFile(true); + File file = writeFile(); ParquetMetadata footer = readFooter(file, new Configuration()); PrimitiveType ts = footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType(); @@ -290,25 +249,12 @@ public void testReaderReadsStatsWrittenWithInt96TimestampOrder() throws IOExcept assertStatsUsable(getColumn(footer, "ts")); } - @Test - public void testReaderReadsStatsWrittenWithInt96TimestampOrderWhenDisabled() throws IOException { - File file = writeFile(true); - - Configuration conf = new Configuration(); - conf.setBoolean(ParquetInputFormat.INT96_TIMESTAMP_STATISTICS_READING_ENABLED, false); - ParquetMetadata footer = readFooter(file, conf); - - assertEquals(ColumnOrder.undefined(), - footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType().columnOrder()); - assertStatsIgnored(getColumn(footer, "ts")); - } - @Test public void testReaderIgnoresInt96StatsWithTypeDefinedOrder() throws IOException { // Legacy layout: stats present, but column order is TYPE_ORDER so they are ignored. - File file = writeFile(false); + File file = writeFile(); FileMetaData rawFooter = readRawFooter(file); - injectInt96Stats(rawFooter); + downgradeInt96ToTypeOrder(rawFooter); File legacy = rewriteFooter(file, rawFooter, "legacy.parquet"); // Validate the data is still intact after rewriting the footer. From cc274c37df743e8fc5f644a672907baa2e76dae1 Mon Sep 17 00:00:00 2001 From: Divjot Arora Date: Mon, 15 Jun 2026 19:14:51 +0000 Subject: [PATCH 3/6] remove flags --- .../apache/parquet/schema/PrimitiveType.java | 27 +++++++++++---- .../columnindex/TestBinaryTruncator.java | 11 ++++++- .../converter/ParquetMetadataConverter.java | 7 +++- .../hadoop/InternalParquetRecordWriter.java | 2 +- .../parquet/hadoop/ParquetFileWriter.java | 32 +----------------- .../TestParquetMetadataConverter.java | 23 ++++++------- .../TestInt96TimestampStatistics.java | 33 +++++++++++-------- 7 files changed, 69 insertions(+), 66 deletions(-) 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 6936b8e9c0..51905753c9 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 @@ -580,9 +580,15 @@ public PrimitiveType( this.decimalMeta = decimalMeta; if (columnOrder == null) { - columnOrder = primitive == PrimitiveTypeName.INT96 || originalType == OriginalType.INTERVAL - ? 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 = originalType == null ? ColumnOrder.int96TimestampOrder() : ColumnOrder.undefined(); + } 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, @@ -631,10 +637,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 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..ec4bec21ab 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,16 @@ 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); + + // 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-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 5a9ce9ab98..f2f48e5974 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 @@ -2064,12 +2064,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 + // 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 { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java index 3e8713ce5c..dd51d1ef09 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java @@ -89,7 +89,7 @@ public InternalParquetRecordWriter( ParquetProperties props) { this.parquetFileWriter = parquetFileWriter; this.writeSupport = Objects.requireNonNull(writeSupport, "writeSupport cannot be null"); - this.schema = ParquetFileWriter.applyInt96TimestampOrder(schema); + this.schema = schema; this.extraMetaData = extraMetaData; this.rowGroupSizeThreshold = rowGroupSize; this.rowGroupRecordCountThreshold = props.getRowGroupRowCountLimit(); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java index 67b296cc1c..82f4577b83 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java @@ -92,12 +92,8 @@ import org.apache.parquet.io.ParquetEncodingException; import org.apache.parquet.io.PositionOutputStream; import org.apache.parquet.io.SeekableInputStream; -import org.apache.parquet.schema.ColumnOrder; -import org.apache.parquet.schema.GroupType; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType; -import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; -import org.apache.parquet.schema.Type; import org.apache.parquet.schema.TypeUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -439,7 +435,7 @@ public ParquetFileWriter( throws IOException { this( file, - applyInt96TimestampOrder(schema), + schema, mode, rowGroupSize, maxPaddingSize, @@ -451,32 +447,6 @@ public ParquetFileWriter( props.getAllocator()); } - /** - * Returns the schema with INT96 columns tagged with the INT96_TIMESTAMP_ORDER column order, so - * that statistics are accumulated with the chronological comparator and the proper column order - * is written to the footer. - */ - static MessageType applyInt96TimestampOrder(MessageType schema) { - return new MessageType(schema.getName(), applyInt96TimestampOrder(schema.getFields())); - } - - private static List applyInt96TimestampOrder(List fields) { - List result = new ArrayList<>(fields.size()); - for (Type field : fields) { - if (field.isPrimitive()) { - PrimitiveType primitive = field.asPrimitiveType(); - if (primitive.getPrimitiveTypeName() == PrimitiveTypeName.INT96) { - field = primitive.withColumnOrder(ColumnOrder.int96TimestampOrder()); - } - result.add(field); - } else { - GroupType group = field.asGroupType(); - result.add(group.withNewFields(applyInt96TimestampOrder(group.getFields()))); - } - } - return result; - } - @Deprecated public ParquetFileWriter( OutputFile file, 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..dd936bc611 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 @@ -1171,6 +1171,7 @@ public void testMissingValuesFromStats() { @Test public void testSkippedV2Stats() { + // INTERVAL has an undefined column order, so its stats are skipped. testSkippedV2Stats( Types.optional(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) .length(12) @@ -1178,10 +1179,6 @@ public void testSkippedV2Stats() { .named(""), new BigInteger("12345678"), new BigInteger("12345679")); - testSkippedV2Stats( - Types.optional(PrimitiveTypeName.INT96).named(""), - new BigInteger("-75687987"), - new BigInteger("45367657")); } private void testSkippedV2Stats(PrimitiveType type, Object min, Object max) { @@ -1383,8 +1380,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 +1394,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 +1410,7 @@ 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 @@ -1488,7 +1485,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 index 56389694c5..d642af24d7 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java @@ -189,6 +189,8 @@ private static void assertStatsUsable(ColumnChunkMetaData column) { @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 @@ -224,19 +226,6 @@ public void testWriterEmitsInt96StatsAndColumnOrder() throws IOException { } } - @Test - public void testStatsAccumulationUsesChronologicalOrder() 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(); - ParquetMetadata footer = readFooter(file, new Configuration()); - byte[] minBytes = getColumn(footer, "ts").getStatistics().getMinBytes(); - assertFalse(Binary.fromConstantByteArray(minBytes).equals(NEXT_DAY)); - assertArrayEquals(EXPECTED_MIN.getBytes(), minBytes); - assertArrayEquals(EXPECTED_MAX.getBytes(), - getColumn(footer, "ts").getStatistics().getMaxBytes()); - } - @Test public void testReaderReadsStatsWrittenWithInt96TimestampOrder() throws IOException { File file = writeFile(); @@ -255,7 +244,7 @@ public void testReaderIgnoresInt96StatsWithTypeDefinedOrder() throws IOException File file = writeFile(); FileMetaData rawFooter = readRawFooter(file); downgradeInt96ToTypeOrder(rawFooter); - File legacy = rewriteFooter(file, rawFooter, "legacy.parquet"); + File legacy = rewriteFooter(file, rawFooter, "type-defined-orders.parquet"); // Validate the data is still intact after rewriting the footer. try ( @@ -278,6 +267,22 @@ public void testReaderIgnoresInt96StatsWithTypeDefinedOrder() throws IOException assertTrue(getColumn(footer, "id").getStatistics().hasNonNullValue()); } + @Test + public void testReaderIgnoresInt96StatsWhenColumnOrdersAbsent() throws IOException { + // Legacy layout: INT96 stats present but the footer omits column_orders entirely (predating the + // order). The reader must not infer INT96_TIMESTAMP_ORDER from the construction-time default + // and must therefore ignore the stats. + File file = writeFile(); + FileMetaData rawFooter = readRawFooter(file); + rawFooter.unsetColumn_orders(); + File legacy = rewriteFooter(file, rawFooter, "no-column-orders.parquet"); + + ParquetMetadata footer = readFooter(legacy, new Configuration()); + assertEquals(ColumnOrder.undefined(), + footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType().columnOrder()); + assertStatsIgnored(getColumn(footer, "ts")); + } + private static byte[] toArray(ByteBuffer buffer) { byte[] bytes = new byte[buffer.remaining()]; buffer.duplicate().get(bytes); From 35a6036dd2d4ca333252168b3be0bea5cd6b6e59 Mon Sep 17 00:00:00 2001 From: Divjot Arora Date: Wed, 24 Jun 2026 09:09:54 +0000 Subject: [PATCH 4/6] fix javadoc --- .../java/org/apache/parquet/schema/PrimitiveComparator.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 113694f6a6..6c05711470 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 @@ -356,7 +356,7 @@ public String toString() { } }; - /* + /** * 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. * From 6ee32f5d2949f6bf2377e3cefc0176031bb0410d Mon Sep 17 00:00:00 2001 From: Divjot Arora Date: Mon, 13 Jul 2026 16:49:16 +0000 Subject: [PATCH 5/6] address comments --- .../apache/parquet/schema/ColumnOrder.java | 4 +- .../parquet/schema/PrimitiveComparator.java | 13 +- .../apache/parquet/schema/PrimitiveType.java | 5 +- .../columnindex/TestBinaryTruncator.java | 3 + .../schema/TestPrimitiveComparator.java | 62 +++----- .../converter/ParquetMetadataConverter.java | 14 +- .../TestParquetMetadataConverter.java | 52 ++++++- .../TestInt96TimestampStatistics.java | 143 ++++++------------ 8 files changed, 150 insertions(+), 146 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 625e5aea1c..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 @@ -42,9 +42,7 @@ public enum ColumnOrderName { */ IEEE_754_TOTAL_ORDER, /** - * Chronological order for INT96 timestamps: values are compared by the Julian day (the last 4 - * bytes, as a little-endian signed int32), then by the nanoseconds within the day (the first 8 - * bytes, as a little-endian signed int64). Only supported for the INT96 physical type. + * Chronological order for INT96 timestamps. */ INT96_TIMESTAMP_ORDER } 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 6c05711470..c115bb3caa 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 @@ -365,6 +365,8 @@ public String toString() { * 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) { @@ -379,7 +381,16 @@ int compareBinary(Binary b1, Binary b2) { int result = Integer.compare(bb1.getInt(8), bb2.getInt(8)); if (result != 0) return result; - return Long.compare(bb1.getLong(0), bb2.getLong(0)); + + 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 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 51905753c9..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 @@ -581,9 +581,8 @@ public PrimitiveType( if (columnOrder == null) { 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 = originalType == null ? ColumnOrder.int96TimestampOrder() : ColumnOrder.undefined(); + // INT96 is only used for deprecated timestamps and supports no other semantics. + columnOrder = ColumnOrder.int96TimestampOrder(); } else if (originalType == OriginalType.INTERVAL) { columnOrder = ColumnOrder.undefined(); } else { 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 ec4bec21ab..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,10 @@ 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); + } + @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. 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 65f56a6864..6ee8e89cc2 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 @@ -37,11 +37,8 @@ import java.math.BigInteger; import java.nio.ByteBuffer; -import java.time.LocalDateTime; import java.util.ArrayList; -import java.util.Arrays; import java.util.List; -import java.util.function.Function; import org.apache.parquet.example.data.simple.NanoTime; import org.apache.parquet.io.api.Binary; import org.junit.Test; @@ -363,12 +360,6 @@ private static Binary int96(int julianDay, long nanosOfDay) { return new NanoTime(julianDay, nanosOfDay).toBinary(); } - private static Binary timestampToInt96(String timestamp) { - LocalDateTime dt = LocalDateTime.parse(timestamp); - int julianDay = (int) (dt.toLocalDate().toEpochDay() + 2440588); - return new NanoTime(julianDay, dt.toLocalTime().toNanoOfDay()).toBinary(); - } - @Test public void testInt96TimestampComparator() { Binary[] valuesInAscendingOrder = { @@ -376,39 +367,36 @@ public void testInt96TimestampComparator() { 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 - timestampToInt96("1968-05-23T00:00:00.000000123"), // pre-epoch but positive julian day - timestampToInt96("2020-01-01T12:00:00"), - timestampToInt96("2020-02-01T11:00:00"), // later day even though earlier time of day - timestampToInt96("2020-02-01T11:00:00.000000001"), // nanos tie-break + 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) }; - // The same value in different Binary representations must compare identically; the offset - // variant guards against absolute reads not being relative to the value's start - List> representations = List.of( - b -> b, - b -> Binary.fromReusedByteArray(b.getBytes()), - b -> Binary.fromConstantByteArray(b.getBytes()), - b -> { - byte[] bytes = b.getBytes(); - byte[] padded = new byte[bytes.length + 20]; - Arrays.fill(padded, (byte) 0xAA); - System.arraycopy(bytes, 0, padded, 10, bytes.length); - return Binary.fromReusedByteArray(padded, 10, bytes.length); - }); - for (int i = 0; i < valuesInAscendingOrder.length; ++i) { for (int j = 0; j < valuesInAscendingOrder.length; ++j) { - for (Function fi : representations) { - for (Function fj : representations) { - Binary bi = fi.apply(valuesInAscendingOrder[i]); - Binary bj = fj.apply(valuesInAscendingOrder[j]); - assertEquals( - "comparing value " + i + " to value " + j, - Integer.signum(Integer.compare(i, j)), - Integer.signum(BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(bi, bj))); - } - } + 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 } } } 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 f2f48e5974..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 @@ -916,10 +916,16 @@ private static byte[] tuncateMax(BinaryTruncator truncator, int truncateLength, } private static boolean isMinMaxStatsSupported(PrimitiveType type) { - ColumnOrderName name = type.columnOrder().getColumnOrderName(); - return name == ColumnOrderName.TYPE_DEFINED_ORDER - || name == ColumnOrderName.IEEE_754_TOTAL_ORDER - || name == ColumnOrderName.INT96_TIMESTAMP_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()); + } } /** 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 dd936bc611..d8b5b8b8b1 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; @@ -1171,7 +1172,6 @@ public void testMissingValuesFromStats() { @Test public void testSkippedV2Stats() { - // INTERVAL has an undefined column order, so its stats are skipped. testSkippedV2Stats( Types.optional(PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) .length(12) @@ -1179,6 +1179,13 @@ public void testSkippedV2Stats() { .named(""), new BigInteger("12345678"), new BigInteger("12345679")); + testSkippedV2Stats( + Types.optional(PrimitiveTypeName.INT96) + .columnOrder(ColumnOrder.undefined()) + .named(""), + new NanoTime(2458850, 0L).toBinary(), + new NanoTime(2458881, 39_600_000_000_000L).toBinary() + ); } private void testSkippedV2Stats(PrimitiveType type, Object min, Object max) { @@ -1217,6 +1224,10 @@ public void testV2OnlyStats() { .named(""), new BigInteger("-6769643"), new BigInteger("9864675")); + testV2OnlyStats( + Types.optional(PrimitiveTypeName.INT96).named(""), + new NanoTime(2458850, 0L).toBinary(), // 2020-01-01 + new NanoTime(2458881, 39_600_000_000_000L).toBinary()); // 2020-02-01T11:00 } private void testV2OnlyStats(PrimitiveType type, Object min, Object max) { @@ -1255,10 +1266,12 @@ public void testV2StatsEqualMinMax() { .named(""), new BigInteger("-8752832"), new BigInteger("-8752832")); + Binary int96 = new NanoTime(2458850, 43_200_000_000_000L) + .toBinary(); // 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) { @@ -1278,6 +1291,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 Binary) { + return createStatsTyped(type, (Binary) min, (Binary) max); } fail("Not implemented"); return null; @@ -1312,6 +1327,15 @@ private static Statistics createStatsTyped(PrimitiveType type, BigInteger min return stats; } + private static Statistics createStatsTyped(PrimitiveType type, Binary min, Binary max) { + Statistics stats = Statistics.createStats(type); + stats.updateStats(max); + stats.updateStats(min); + assertEquals(min, stats.genericGetMin()); + assertEquals(max, stats.genericGetMax()); + return stats; + } + private static ParquetMetadata createParquetMetaData(Encoding dicEncoding, Encoding dataEncoding) { return createParquetMetaData(dicEncoding, dataEncoding, true); } @@ -1413,6 +1437,28 @@ public void testColumnOrders() throws IOException { assertEquals(ColumnOrder.int96TimestampOrder(), columns.get(2).getPrimitiveType().columnOrder()); } + @Test + public void testLegacyINT96ColumnOrder() throws IOException { + // Create a footer for a single INT96 column schema that does not specify a column order. + 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); + + formatMetadata.unsetColumn_orders(); + assertFalse(formatMetadata.isSetColumn_orders()); + + MessageType resultSchema = + converter.fromParquetMetadata(formatMetadata).getFileMetaData().getSchema(); + List columns = resultSchema.getColumns(); + assertEquals(1, columns.size()); + assertEquals(ColumnOrder.undefined(), columns.get(0).getPrimitiveType().columnOrder()); + } + @Test public void testOffsetIndexConversion() { for (boolean withSizeStats : new boolean[] {false, true}) { 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 index d642af24d7..cabace996d 100644 --- a/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java +++ b/parquet-hadoop/src/test/java/org/apache/parquet/statistics/TestInt96TimestampStatistics.java @@ -34,30 +34,27 @@ import java.nio.ByteOrder; import java.nio.file.Files; import java.util.List; -import org.apache.hadoop.conf.Configuration; -import org.apache.hadoop.fs.Path; -import org.apache.parquet.HadoopReadOptions; +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.RowGroup; 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.ParquetReader; import org.apache.parquet.hadoop.ParquetWriter; import org.apache.parquet.hadoop.example.ExampleParquetWriter; -import org.apache.parquet.hadoop.example.GroupReadSupport; -import org.apache.parquet.hadoop.example.GroupWriteSupport; 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.hadoop.util.HadoopInputFile; 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; @@ -77,10 +74,10 @@ public class TestInt96TimestampStatistics { // 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 = int96(2440000, 123L); // 1968-05-23 00:00:00.000000123 - private static final Binary SAME_DAY_EARLY = int96(2440588, 1_000L); // 1970-01-01 00:00:00.000001 - private static final Binary LATE_IN_DAY = int96(2440588, 86_399_999_999_999L); // 1970-01-01 23:59:59.999... - private static final Binary NEXT_DAY = int96(2440589, 0L); // 1970-01-02 00:00:00 + 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; @@ -89,21 +86,12 @@ public class TestInt96TimestampStatistics { @Rule public TemporaryFolder tmp = new TemporaryFolder(); - private static Binary int96(int julianDay, long nanosOfDay) { - return Binary.fromConstantByteArray(ByteBuffer.allocate(12) - .order(ByteOrder.LITTLE_ENDIAN) - .putLong(nanosOfDay) - .putInt(julianDay) - .array()); - } - private File writeFile() throws IOException { - File file = new File(tmp.getRoot(), "int96.parquet"); - Configuration conf = new Configuration(); - GroupWriteSupport.setSchema(SCHEMA, conf); + File file = tmp.newFile("int96.parquet"); + file.delete(); SimpleGroupFactory factory = new SimpleGroupFactory(SCHEMA); - ParquetWriter writer = ExampleParquetWriter.builder(new Path(file.getAbsolutePath())) - .withConf(conf) + ParquetWriter writer = ExampleParquetWriter.builder(new LocalOutputFile(file.toPath())) + .withType(SCHEMA) .build(); try { for (int i = 0; i < VALUES.size(); i++) { @@ -115,11 +103,9 @@ private File writeFile() throws IOException { return file; } - private static ParquetMetadata readFooter(File file, Configuration conf) throws IOException { - Path path = new Path(file.getAbsolutePath()); + private static ParquetMetadata readFooter(File file) throws IOException { try (ParquetFileReader reader = ParquetFileReader.open( - HadoopInputFile.fromPath(path, conf), - HadoopReadOptions.builder(conf, path).build())) { + new LocalInputFile(file.toPath()), ParquetReadOptions.builder().build())) { return reader.getFooter(); } } @@ -133,8 +119,11 @@ private static FileMetaData readRawFooter(File file) throws IOException { return Util.readFileMetaData(new ByteArrayInputStream(bytes, footerStart, footerLen)); } - /** Rewrites the footer of src into a new file, keeping all data pages byte-identical. */ - private File rewriteFooter(File src, FileMetaData footer, String name) + /** + * 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) @@ -156,19 +145,6 @@ private File rewriteFooter(File src, FileMetaData footer, String name) return dst; } - /** - * Retags the INT96 column orders in the footer as TYPE_ORDER while leaving the (already written) - * min/max statistics in place. This simulates a legacy writer that emitted INT96 stats under - * TYPE_ORDER before INT96_TIMESTAMP_ORDER existed; such stats must be ignored by the reader. - */ - private static void downgradeInt96ToTypeOrder(FileMetaData footer) { - for (org.apache.parquet.format.ColumnOrder columnOrder : footer.getColumn_orders()) { - if (columnOrder.isSetINT96_TIMESTAMP_ORDER()) { - columnOrder.setTYPE_ORDER(new TypeDefinedOrder()); - } - } - } - private static ColumnChunkMetaData getColumn(ParquetMetadata footer, String name) { BlockMetaData block = footer.getBlocks().get(0); return block.getColumns().stream() @@ -195,41 +171,32 @@ public void testWriterEmitsInt96StatsAndColumnOrder() throws IOException { 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()); - assertTrue(rawFooter.getColumn_orders().get(1).isSetTYPE_ORDER()); - for (RowGroup rowGroup : rawFooter.getRow_groups()) { - for (ColumnChunk chunk : rowGroup.getColumns()) { - if (chunk.getMeta_data().getType() == Type.INT96) { - Statistics stats = chunk.getMeta_data().getStatistics(); - assertTrue(stats != null && stats.isSetMin_value()); - assertArrayEquals(EXPECTED_MIN.getBytes(), stats.getMin_value()); - assertArrayEquals(EXPECTED_MAX.getBytes(), stats.getMax_value()); - } - } - } + // 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 both columns. - Configuration conf = new Configuration(); - Path path = new Path(file.getAbsolutePath()); + // Column index should be present for the INT96 column. try ( ParquetFileReader reader = ParquetFileReader.open( - HadoopInputFile.fromPath(path, conf), HadoopReadOptions.builder(conf, path).build() + new LocalInputFile(file.toPath()), ParquetReadOptions.builder().build() ) ) { - assertNotNull(reader.readColumnIndex(getColumn(reader.getFooter(), "id"))); - 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(columnIndex.getMaxValues().size() - 1))); + assertArrayEquals(EXPECTED_MAX.getBytes(), toArray(columnIndex.getMaxValues().get(0))); } } @Test public void testReaderReadsStatsWrittenWithInt96TimestampOrder() throws IOException { File file = writeFile(); - ParquetMetadata footer = readFooter(file, new Configuration()); + ParquetMetadata footer = readFooter(file); PrimitiveType ts = footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType(); assertEquals(ColumnOrder.int96TimestampOrder(), ts.columnOrder()); @@ -238,28 +205,14 @@ public void testReaderReadsStatsWrittenWithInt96TimestampOrder() throws IOExcept assertStatsUsable(getColumn(footer, "ts")); } - @Test - public void testReaderIgnoresInt96StatsWithTypeDefinedOrder() throws IOException { - // Legacy layout: stats present, but column order is TYPE_ORDER so they are ignored. + private void assertLegacyInt96StatsIgnored(String name, Consumer legacyMutation) + throws IOException { File file = writeFile(); FileMetaData rawFooter = readRawFooter(file); - downgradeInt96ToTypeOrder(rawFooter); - File legacy = rewriteFooter(file, rawFooter, "type-defined-orders.parquet"); + legacyMutation.accept(rawFooter); + File legacy = copyWithNewFooter(file, rawFooter, name + "-orders.parquet"); - // Validate the data is still intact after rewriting the footer. - try ( - ParquetReader reader = ParquetReader.builder( - new GroupReadSupport(), new Path(legacy.getAbsolutePath()) - ).build() - ) { - for (int i = 0; i < VALUES.size(); i++) { - Group group = reader.read(); - assertEquals(VALUES.get(i), group.getInt96("ts", 0)); - assertEquals(i, group.getLong("id", 0)); - } - } - - ParquetMetadata footer = readFooter(legacy, new Configuration()); + ParquetMetadata footer = readFooter(legacy); assertEquals(ColumnOrder.undefined(), footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType().columnOrder()); assertStatsIgnored(getColumn(footer, "ts")); @@ -268,19 +221,19 @@ public void testReaderIgnoresInt96StatsWithTypeDefinedOrder() throws IOException } @Test - public void testReaderIgnoresInt96StatsWhenColumnOrdersAbsent() throws IOException { - // Legacy layout: INT96 stats present but the footer omits column_orders entirely (predating the - // order). The reader must not infer INT96_TIMESTAMP_ORDER from the construction-time default - // and must therefore ignore the stats. - File file = writeFile(); - FileMetaData rawFooter = readRawFooter(file); - rawFooter.unsetColumn_orders(); - File legacy = rewriteFooter(file, rawFooter, "no-column-orders.parquet"); + 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()); + } + } + }); + } - ParquetMetadata footer = readFooter(legacy, new Configuration()); - assertEquals(ColumnOrder.undefined(), - footer.getFileMetaData().getSchema().getType("ts").asPrimitiveType().columnOrder()); - assertStatsIgnored(getColumn(footer, "ts")); + @Test + public void testReaderIgnoresInt96StatsWhenColumnOrdersAbsent() throws IOException { + assertLegacyInt96StatsIgnored("absent", FileMetaData::unsetColumn_orders); } private static byte[] toArray(ByteBuffer buffer) { From 41d1b37434087432b0d8847eefa17b454467d108 Mon Sep 17 00:00:00 2001 From: Divjot Arora Date: Wed, 15 Jul 2026 08:26:00 +0000 Subject: [PATCH 6/6] Address review comments --- .../parquet/schema/PrimitiveComparator.java | 4 +- .../schema/TestPrimitiveComparator.java | 11 ++-- .../TestParquetMetadataConverter.java | 50 ++++++++++++------- 3 files changed, 40 insertions(+), 25 deletions(-) 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 c115bb3caa..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 @@ -385,10 +385,10 @@ int compareBinary(Binary b1, Binary b2) { long nanos1 = bb1.getLong(0); long nanos2 = bb2.getLong(0); if (nanos1 < 0 || nanos1 > NANOSECONDS_PER_DAY) { - throw new IllegalArgumentException("Invalid nanos value: " + nanos1); + 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: " + nanos2); + throw new IllegalArgumentException("Invalid nanos value (must be positive and less than 1 day): " + nanos2); } return Long.compare(nanos1, nanos2); } 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 6ee8e89cc2..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 @@ -39,6 +39,7 @@ 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; @@ -392,12 +393,10 @@ public void testInt96TimestampComparatorRejectsInvalidNanos() { 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 - } + TestUtils.assertThrows( + "Expected IllegalArgumentException for nanos=" + invalidNanos, + IllegalArgumentException.class, + () -> BINARY_AS_INT96_TIMESTAMP_COMPARATOR.compare(valid, invalid)); } } 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 d8b5b8b8b1..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 @@ -1183,8 +1183,8 @@ public void testSkippedV2Stats() { Types.optional(PrimitiveTypeName.INT96) .columnOrder(ColumnOrder.undefined()) .named(""), - new NanoTime(2458850, 0L).toBinary(), - new NanoTime(2458881, 39_600_000_000_000L).toBinary() + new NanoTime(2458850, 0L), + new NanoTime(2458881, 39_600_000_000_000L) ); } @@ -1226,8 +1226,8 @@ public void testV2OnlyStats() { new BigInteger("9864675")); testV2OnlyStats( Types.optional(PrimitiveTypeName.INT96).named(""), - new NanoTime(2458850, 0L).toBinary(), // 2020-01-01 - new NanoTime(2458881, 39_600_000_000_000L).toBinary()); // 2020-02-01T11:00 + 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) { @@ -1266,8 +1266,7 @@ public void testV2StatsEqualMinMax() { .named(""), new BigInteger("-8752832"), new BigInteger("-8752832")); - Binary int96 = new NanoTime(2458850, 43_200_000_000_000L) - .toBinary(); // 2020-01-01T12:00:00 + NanoTime int96 = new NanoTime(2458850, 43_200_000_000_000L); // 2020-01-01T12:00:00 testV2StatsEqualMinMax( Types.optional(PrimitiveTypeName.INT96).named(""), int96, @@ -1291,8 +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 Binary) { - return createStatsTyped(type, (Binary) min, (Binary) max); + } else if (min instanceof NanoTime) { + return createStatsTyped(type, (NanoTime) min, (NanoTime) max); } fail("Not implemented"); return null; @@ -1327,12 +1326,14 @@ private static Statistics createStatsTyped(PrimitiveType type, BigInteger min return stats; } - private static Statistics createStatsTyped(PrimitiveType type, Binary min, Binary max) { + 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(max); - stats.updateStats(min); - assertEquals(min, stats.genericGetMin()); - assertEquals(max, stats.genericGetMax()); + stats.updateStats(maxBinary); + stats.updateStats(minBinary); + assertEquals(minBinary, stats.genericGetMin()); + assertEquals(maxBinary, stats.genericGetMax()); return stats; } @@ -1438,8 +1439,19 @@ public void testColumnOrders() throws IOException { } @Test - public void testLegacyINT96ColumnOrder() throws IOException { - // Create a footer for a single INT96 column schema that does not specify a column order. + 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;" + "}"); @@ -1449,8 +1461,12 @@ public void testLegacyINT96ColumnOrder() throws IOException { ParquetMetadataConverter converter = new ParquetMetadataConverter(); FileMetaData formatMetadata = converter.toParquetMetadata(1, metadata); - formatMetadata.unsetColumn_orders(); - assertFalse(formatMetadata.isSetColumn_orders()); + if (footerColumnOrder == null) { + formatMetadata.unsetColumn_orders(); + assertFalse(formatMetadata.isSetColumn_orders()); + } else { + formatMetadata.setColumn_orders(Collections.singletonList(footerColumnOrder)); + } MessageType resultSchema = converter.fromParquetMetadata(formatMetadata).getFileMetaData().getSchema();