Support pruning on INT96 timestamp columns via `SortOrder::INT96_TIMESTAMP` / `ColumnOrder::INT96_TIMESTAMP_ORDER`

Start with has_untrusted_min_max_order and the linked code in datafusion/datasource-parquet/src/metadata.rs, row_group_filter.rs, and page_filter.rs. Trace how ColumnOrder and INT96 statistics are handled, then use the supplied int96_pruning.zip SQL and EXPLAIN ANALYZE example to verify pruning. Done means supported INT96_TIMESTAMP ordering enables correct file, row-group, and page pruning without treating untrusted bounds as valid.

Is your feature request related to a problem or challenge?

The Parquet specification historically left the sort order of INT96 undefined, so DataFusion doesn't trust min/max statistics of INT96 columns and therefore cannot prune row groups, pages, or files using predicates on those columns.

However, Parquet files written by Spark, Hive, Impala, and Databricks Photon commonly store timestamps as the (deprecated) INT96 physical type and show no signs of changing this behavior. 👎 This means any query that filters timestamps with such a file must scan every row group

The Parquet specification recently added a new ColumnOrder variant, INT96_TIMESTAMP_ORDER, to explicitly define the order. See

• apache/parquet-format#584

arrow-rs / parquet 60.0.0 implemented this in

• https://github.com/apache/arrow-rs/pull/10106

However, to avoid introducing any new correctness bugs, when upgrading to arrow/parquet 60.0.0 in https://github.com/apache/datafusion/pull/25335, I changed has_untrusted_min_max_order to keep treating INT96 columns as untrusted (see the review comment at https://github.com/apache/datafusion/pull/25335/changes#r4018522067 and the code)

However, this means that DataFusion will still not prune files with INT96 , even those that could

• File-level statistics: https://github.com/apache/datafusion/blob/5a502dffce2f1881479e8c1a5b68a4a5108e7934/datafusion/datasource-parquet/src/metadata.rs#L555-L585
• Row group pruning: https://github.com/apache/datafusion/blob/5a502dffce2f1881479e8c1a5b68a4a5108e7934/datafusion/datasource-parquet/src/row_group_filter.rs#L518-L531
• Page pruning: https://github.com/apache/datafusion/blob/5a502dffce2f1881479e8c1a5b68a4a5108e7934/datafusion/datasource-parquet/src/page_filter.rs#L760-L764

Here is an example file written with int96 timestamps: int96_pruning.zip

It has two row groups

> select row_group_id, stats_min, stats_max, type from parquet_metadata('parquet_rs_int96_10k.parquet') where "path_in_schema" = '"ts"';
+--------------+-----------------------------+-----------------------------+-------+
| row_group_id | stats_min                   | stats_max                   | type  |
+--------------+-----------------------------+-----------------------------+-------+
| 0            | [0, 0, 2458850]             | [2235197440, 9485, 2458853] | INT96 |
| 1            | [2105655296, 9499, 2458853] | [45885440, 18985, 2458856]  | INT96 |
+--------------+-----------------------------+-----------------------------+-------+
2 row(s) fetched.
Elapsed 0.001 seconds.

Int96 stores nanos-of-day in the low two words and the Julian day in the high word, so decoded those bounds are:

And you would expect that this query could prune one of those row groups:

select ts from 'parquet_rs_int96_10k.parquet' where ts > TIMESTAMP '2020-01-07T22:34:00';

However, it doesn't as you can see from this explain:

> explain analyze  select ts from 'parquet_rs_int96_10k.parquet' where ts > TIMESTAMP '2020-01-07T22:34:00';
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| plan_type         | plan                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                 |
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
| Plan with Metrics | FilterExec: ts@0 > 1578436440000000000, metrics=[output_rows=5, elapsed_compute=43.47µs, output_bytes=64.0 KB, output_batches=1, selectivity=0.05% (5/10.00 K)]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
|                   |   RepartitionExec: partitioning=RoundRobinBatch(16), input_partitions=1, metrics=[output_rows=10.00 K, elapsed_compute=2.63µs, output_bytes=78.1 KB, output_batches=2, spill_count=0, spilled_bytes=0.0 B, spilled_rows=0, fetch_time=1.84ms, repartition_time=1ns, send_time=7.51µs]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
|                   |     DataSourceExec: file_groups={1 group: [[private/tmp/int96_pruning/parquet_rs_int96_10k.parquet]]}, projection=[ts], file_type=parquet, predicate=ts@1 > 1578436440000000000, pruning_predicate=ts_null_count@1 != row_count@2 AND ts_max@0 > 1578436440000000000, required_guarantees=[], metrics=[output_rows=10.00 K, elapsed_compute=2.00µs, output_bytes=78.1 KB, output_batches=2, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=2 total → 2 matched, row_groups_pruned_bloom_filter=2 total → 2 matched, page_index_pages_pruned=2 total → 2 matched, page_index_rows_pruned=10.00 K total → 10.00 K matched, limit_pruned_row_groups=0 total → 0 matched, batches_split=0, bytes_processed=131.2 KB, bytes_scanned=72.7 KB, file_open_errors=0, file_scan_errors=0, files_opened=1, files_processed=1, num_predicate_creation_errors=0, page_index_load_skipped=1, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, row_groups_pruned_dynamic_filter=0, predicate_cache_inner_records=0, predicate_cache_records=0, bloom_filter_eval_time=585ns, metadata_load_time=6.42µs, page_index_eval_time=14.54µs, row_pushdown_eval_time=3ns, statistics_eval_time=13.71µs, time_elapsed_opening=112.25µs, time_elapsed_processing=373.83µs, time_elapsed_scanning_total=1.71ms, time_elapsed_scanning_until_data=1.17ms, output_rows_skew=0%, scan_efficiency_ratio=55.38% (74.43 K/134.4 K)] |
|                   |                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      |
+-------------------+----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.012 seconds.

Note the row_groups_pruned_statistics=2 total → 2 matched term that says we couldn't prune any row groups. Here is how I made the file:

[package]
name = "int96gen"
version = "0.1.0"
edition = "2021"

[dependencies]
parquet = { version = "60.0.0", default-features = false, features = ["snap"] }

//! Writes `parquet_rs_int96_10k.parquet`: 10,000 rows, one row per minute from
//! 2020-01-01T00:00:00Z, in exactly two row groups of 5,000.
//!
//! parquet-rs 60 implements parquet-format #584, so the `ts` column gets real
//! INT96 min/max statistics and the footer advertises
//! `ColumnOrder::INT96_TIMESTAMP_ORDER` for it.

use std::fs::File;
use std::sync::Arc;

use parquet::basic::Compression;
use parquet::data_type::{Int32Type, Int64Type, Int96, Int96Type};
use parquet::file::properties::{EnabledStatistics, WriterProperties};
use parquet::file::writer::SerializedFileWriter;
use parquet::schema::parser::parse_message_type;

const JULIAN_DAY_OF_EPOCH: i64 = 2_440_588;
const NANOS_IN_DAY: i64 = 86_400 * 1_000_000_000;
/// 2020-01-01T00:00:00Z in seconds since the unix epoch.
const START_SECS: i64 = 1_577_836_800;
const N_ROWS: i64 = 10_000;
const ROWS_PER_GROUP: i64 = 5_000;

/// Encode nanoseconds-since-epoch as an INT96: nanos-of-day in the low two
/// words, Julian day in the high word.
fn int96_from_nanos(nanos_since_epoch: i64) -> Int96 {
    let day = nanos_since_epoch.div_euclid(NANOS_IN_DAY);
    let nanos_of_day = nanos_since_epoch.rem_euclid(NANOS_IN_DAY);
    let mut v = Int96::new();
    v.set_data(
        (nanos_of_day & 0xFFFF_FFFF) as u32,
        ((nanos_of_day >> 32) & 0xFFFF_FFFF) as u32,
        (day + JULIAN_DAY_OF_EPOCH) as u32,
    );
    v
}

fn main() -> Result<(), Box<dyn std::error::Error>> {
    let path = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "parquet_rs_int96_10k.parquet".to_string());

    let schema = Arc::new(parse_message_type(
        "message schema {
             REQUIRED INT64 id;
             REQUIRED INT96 ts;
             REQUIRED INT32 val;
         }",
    )?);

    let props = Arc::new(
        WriterProperties::builder()
            .set_compression(Compression::SNAPPY)
            .set_statistics_enabled(EnabledStatistics::Chunk)
            .build(),
    );

    let mut writer = SerializedFileWriter::new(File::create(&path)?, schema, props)?;

    // One explicit `next_row_group` call per row group, so the split is exact.
    for group in 0..(N_ROWS / ROWS_PER_GROUP) {
        let lo = group * ROWS_PER_GROUP;
        let hi = lo + ROWS_PER_GROUP;

        let ids: Vec<i64> = (lo..hi).collect();
        let ts: Vec<Int96> = (lo..hi)
            .map(|i| int96_from_nanos((START_SECS + i * 60) * 1_000_000_000))
            .collect();
        let vals: Vec<i32> = (lo..hi).map(|i| (i % 97) as i32).collect();

        let mut rg = writer.next_row_group()?;

        let mut c = rg.next_column()?.expect("id column");
        c.typed::<Int64Type>().write_batch(&ids, None, None)?;
        c.close()?;

        let mut c = rg.next_column()?.expect("ts column");
        c.typed::<Int96Type>().write_batch(&ts, None, None)?;
        c.close()?;

        let mut c = rg.next_column()?.expect("val column");
        c.typed::<Int32Type>().write_batch(&vals, None, None)?;
        c.close()?;

        rg.close()?;
    }
    writer.close()?;

    println!("wrote {path}");
    Ok(())
}

Describe the solution you'd like

Trust INT96 min/max statistics when column actually has ColumnOrder::INT96_TIMESTAMP_ORDER, and use them for file, row group, and page pruning.

Describe alternatives you've considered

Trusting the bounds whenever column.sort_order() == SortOrder::INT96_TIMESTAMP, without consulting the footer. This would be wrong: ColumnDescriptor::sort_order() is derived from the physical type alone, so every INT96 column reports INT96_TIMESTAMP, including legacy files whose statistics were computed with signed byte-wise comparison (the original motivation for https://github.com/apache/arrow-rs/issues/7686). The footer's column_orders entry is the only trustworthy signal, which mirrors how the existing byte-array path only trusts UNSIGNED when column_orders confirms it.

Additional context

• Upgrade PR where the conservative check was added: https://github.com/apache/datafusion/pull/25335
• arrow-rs implementation: https://github.com/apache/arrow-rs/pull/10106
• Parquet format change: https://github.com/apache/parquet-format/pull/584
• Prior arrow-rs discussion of incorrect INT96 statistics: https://github.com/apache/arrow-rs/issues/7686