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

> 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.

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

> 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.

[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(())
}