lance 14.0.1

// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The Lance Authors

//! Arrow ordering compatibility fixtures.
//!
//! Lance defines no comparison rules of its own. Scalar indices, column
//! statistics and `ORDER BY` all inherit the total order implemented by Arrow
//! Rust, and by DataFusion on top of it. These tests pin that order with
//! hand-written expectations so that an upgrade which changes it fails here,
//! naming the type, the inputs and both orderings, instead of silently
//! reordering btree pages, zone map extrema or query results.
//!
//! # Baseline
//!
//! Established against `arrow-array`, `arrow-ord` and `arrow-row` 58.4.0 and
//! `datafusion` 54.1.0 on 2026-10-07. The expected orders were written from
//! the IEEE 754 `totalOrder` predicate for floats, two's-complement order for
//! integers, decimals and temporal natives, byte-wise lexicographic order for
//! strings and binary, and `false < true` for booleans. None of them were
//! generated by running Arrow. Special floats are spelled as bit patterns and
//! compared by bit pattern, never by `==`.
//!
//! # Production paths covered
//!
//! | API under test                          | Lance path that relies on it                                   |
//! |-----------------------------------------|----------------------------------------------------------------|
//! | `arrow_ord::sort::sort_to_indices`      | `RecordBatchExt::sort_by_column`; DataFusion `SortExec` on one column, which is how `Scanner::order_by` and btree, bitmap, label-list and JSON index training sort |
//! | `arrow_ord::sort::lexsort_to_indices`   | DataFusion `SortExec` on several columns (`Scanner::order_by`, FTS and vector result ordering) |
//! | `arrow_ord::ord::make_comparator`       | `BTreeLookup::pages_between` and the btree equality fallback   |
//! | `ArrowNativeTypeOp::compare`            | the btree's inlined primitive comparator; DataFusion's single-column merge cursor |
//! | `arrow_row::RowConverter`               | `HashJoiner` keys, `FirstByPk` keys, DataFusion's TopK (`SortExec` with a fetch) and its row-based merge cursor for non-primitive types |
//! | `ArrowScalar` (`arrow_row` encoding)    | `StatisticsAccumulator` extrema (zone map, bloom filter)       |
//! | `StatisticsAccumulator` min/max         | zone map and bloom filter statistics                            |
//! | `OrderableScalarValue` / `ScalarValue`  | btree query bounds; zone map pruning                            |
//! | `SortExec`, with and without a fetch    | `Scanner::order_by`, with `limit` taking the TopK path         |
//! | `SortPreservingMergeExec`               | btree training over several inputs                             |
//!
//! Sorting and aggregate statistics are deliberately distinct contracts: a sort
//! places negative NaNs below `-inf` and positive NaNs above `+inf`, while
//! `StatisticsAccumulator` leaves NaN of either sign out of `min`/`max` and
//! reports it through `nan_count`. Both are pinned.
//!
//! # When a test here fails after a dependency upgrade
//!
//! Do not refresh the expectations to make the build green. The baseline is
//! the contract; a change needs a review that answers, in the PR description:
//!
//! 1. Is it an upstream contract change, an upstream bug fix, or a change in
//!    how Lance calls the API? Cite the arrow-rs or DataFusion release note or
//!    pull request.
//! 2. Which Lance behaviours depend on it? Use the table above.
//! 3. Do persisted structures need compatibility handling? Btree page order,
//!    zone map and bloom filter extrema, and sorted index files written by the
//!    previous version are read by the upgraded one.
//!
//! Only once those are answered should the expectations change, together with
//! regression cases for any newly discovered edge case. The reverse also
//! applies: a passing run shows the covered cases are unchanged, not that every
//! input orders identically.
//!
//! # Deferred: fragment column statistics
//!
//! Fragment-level column statistics and the pruning that reads them are not
//! implemented yet. When they are, add to these fixtures:
//!
//! - identical query results with pruning enabled and disabled,
//! - boundary cases for NaN, signed zero, nulls and strings,
//! - statistics written by the previous supported release and read by the
//!   upgraded one, as checked-in `test_data` fixtures.

use std::cmp::Ordering;
use std::sync::Arc;

use arrow::datatypes::i256;
use arrow_array::cast::AsArray;
use arrow_array::types::{
    ArrowPrimitiveType, Date32Type, Decimal128Type, Decimal256Type, Float16Type, Float32Type,
    Float64Type, Int64Type, TimestampNanosecondType, UInt64Type,
};
use arrow_array::{
    Array, ArrayRef, ArrowNativeTypeOp, BinaryArray, BooleanArray, Date32Array, Decimal128Array,
    Decimal256Array, Float16Array, Float32Array, Float64Array, Int64Array, LargeStringArray,
    RecordBatch, StringArray, TimestampNanosecondArray, UInt64Array,
};
use arrow_ord::ord::make_comparator;
use arrow_ord::sort::{SortColumn, lexsort_to_indices, sort_to_indices};
use arrow_row::{RowConverter, SortField};
use arrow_schema::{DataType, Field, Schema, SortOptions};
use arrow_select::concat::concat_batches;
use datafusion::common::ScalarValue;
use datafusion::physical_expr::{PhysicalSortExpr, expressions::Column};
use datafusion::physical_plan::sorts::sort::SortExec;
use datafusion::physical_plan::sorts::sort_preserving_merge::SortPreservingMergeExec;
use datafusion::physical_plan::stream::RecordBatchStreamAdapter;
use datafusion::physical_plan::union::UnionExec;
use datafusion::physical_plan::{ExecutionPlan, SendableRecordBatchStream};
use futures::TryStreamExt;
use half::f16;
use lance_arrow_scalar::ArrowScalar;
use lance_arrow_stats::StatisticsAccumulator;
use lance_datafusion::exec::{LanceExecutionOptions, OneShotExec, execute_plan};
use lance_index::scalar::btree::OrderableScalarValue;
use rstest::rstest;

/// The sort used to train every ordered scalar index (btree, bitmap, label
/// list, JSON) and the value of `SortOptions::default()`.
const ASC_NULLS_FIRST: SortOptions = SortOptions {
    descending: false,
    nulls_first: true,
};
const ASC_NULLS_LAST: SortOptions = SortOptions {
    descending: false,
    nulls_first: false,
};
const DESC_NULLS_FIRST: SortOptions = SortOptions {
    descending: true,
    nulls_first: true,
};
const DESC_NULLS_LAST: SortOptions = SortOptions {
    descending: true,
    nulls_first: false,
};

fn describe(opts: SortOptions) -> String {
    format!(
        "{} nulls {}",
        if opts.descending { "desc" } else { "asc" },
        if opts.nulls_first { "first" } else { "last" }
    )
}

/// One typed array plus the hand-written ascending order of its non-null
/// values.
///
/// The input array is the ordered values followed by `null_count` nulls, then
/// shuffled with a fixed permutation, so a test that reads the input back in
/// the baseline order has to apply the ordering under test.
struct Fixture {
    name: &'static str,
    /// Labels of the non-null values in expected ascending order. Adjacent
    /// entries with the same identity are ties.
    labels: Vec<&'static str>,
    /// Bit-exact identity of each ordered value: little-endian bits for
    /// numbers, the bytes for strings and binary.
    identities: Vec<Vec<u8>>,
    /// Whether each ordered value is a NaN, which statistics leave out of
    /// `min`/`max`.
    is_nan: Vec<bool>,
    null_count: usize,
    input: ArrayRef,
    /// `ordered_pos[i]` is the position of input row `i` in the ordered list;
    /// nulls get positions at or past `labels.len()`.
    ordered_pos: Vec<usize>,
    /// Equality group of each ordered position, ascending; nulls share the
    /// last group.
    group_of_pos: Vec<usize>,
    /// Bit-exact identity of row `i` of an array of this fixture's type.
    identity_of: fn(&dyn Array, usize) -> Vec<u8>,
    /// `ArrowNativeTypeOp::compare` over two non-null rows, for primitive
    /// types only.
    native_cmp: Option<NativeCmp>,
}

/// Compares two non-null rows of a primitive array by native value.
type NativeCmp = fn(&dyn Array, usize, usize) -> Ordering;

/// A fixed shuffle that does not depend on any Arrow code. Every input length
/// gets the same permutation on every run.
fn permutation(len: usize) -> Vec<usize> {
    let mut perm: Vec<usize> = (0..len).collect();
    for i in (1..len).rev() {
        let j = (i * 7 + 3) % (i + 1);
        perm.swap(i, j);
    }
    perm
}

#[allow(clippy::too_many_arguments)]
fn build<T: Clone>(
    name: &'static str,
    ordered: Vec<(&'static str, T)>,
    null_count: usize,
    identity: fn(&T) -> Vec<u8>,
    is_nan: fn(&T) -> bool,
    to_array: fn(Vec<Option<T>>) -> ArrayRef,
    identity_of: fn(&dyn Array, usize) -> Vec<u8>,
    native_cmp: Option<NativeCmp>,
) -> Fixture {
    let labels: Vec<&'static str> = ordered.iter().map(|(label, _)| *label).collect();
    let identities: Vec<Vec<u8>> = ordered.iter().map(|(_, v)| identity(v)).collect();
    let nan_flags: Vec<bool> = ordered.iter().map(|(_, v)| is_nan(v)).collect();

    let mut group_of_pos = Vec::with_capacity(labels.len() + null_count);
    let mut group = 0;
    for (pos, id) in identities.iter().enumerate() {
        if pos > 0 && identities[pos - 1] != *id {
            group += 1;
        }
        group_of_pos.push(group);
    }
    let null_group = if labels.is_empty() { 0 } else { group + 1 };
    group_of_pos.extend(std::iter::repeat_n(null_group, null_count));

    let mut in_order: Vec<Option<T>> = ordered.into_iter().map(|(_, v)| Some(v)).collect();
    in_order.extend(std::iter::repeat_n(None, null_count));
    let ordered_pos = permutation(in_order.len());
    let shuffled: Vec<Option<T>> = ordered_pos
        .iter()
        .map(|&pos| in_order[pos].clone())
        .collect();

    Fixture {
        name,
        labels,
        identities,
        is_nan: nan_flags,
        null_count,
        input: to_array(shuffled),
        ordered_pos,
        group_of_pos,
        identity_of,
        native_cmp,
    }
}

impl Fixture {
    fn len(&self) -> usize {
        self.input.len()
    }

    fn is_null_row(&self, row: usize) -> bool {
        self.ordered_pos[row] >= self.labels.len()
    }

    fn label(&self, row: usize) -> &'static str {
        if self.is_null_row(row) {
            "null"
        } else {
            self.labels[self.ordered_pos[row]]
        }
    }

    fn group(&self, row: usize) -> usize {
        self.group_of_pos[self.ordered_pos[row]]
    }

    fn non_null_rows(&self) -> Vec<usize> {
        (0..self.len())
            .filter(|&row| !self.is_null_row(row))
            .collect()
    }

    /// Input rows in the baseline order for `opts`, with ties in input order.
    fn baseline_rows(&self, opts: SortOptions) -> Vec<usize> {
        let mut values: Vec<usize> = self.non_null_rows();
        values.sort_by_key(|&row| self.ordered_pos[row]);
        if opts.descending {
            values.reverse();
        }
        let nulls: Vec<usize> = (0..self.len())
            .filter(|&row| self.is_null_row(row))
            .collect();
        if opts.nulls_first {
            nulls.into_iter().chain(values).collect()
        } else {
            values.into_iter().chain(nulls).collect()
        }
    }

    /// Checks that `rows`, an ordering of every input row, agrees with the
    /// baseline up to the order within a tie group.
    fn check_order(&self, path: &str, opts: SortOptions, rows: &[usize]) -> Result<(), String> {
        let expected = self.baseline_rows(opts);
        let expected_groups: Vec<usize> = expected.iter().map(|&row| self.group(row)).collect();
        let actual_groups: Vec<usize> = rows.iter().map(|&row| self.group(row)).collect();
        if expected_groups == actual_groups {
            return Ok(());
        }
        Err(format!(
            "fixture {} via {} ({}):\n  expected [{}]\n  actual   [{}]",
            self.name,
            path,
            describe(opts),
            expected
                .iter()
                .map(|&row| self.label(row))
                .collect::<Vec<_>>()
                .join(", "),
            rows.iter()
                .map(|&row| self.label(row))
                .collect::<Vec<_>>()
                .join(", "),
        ))
    }

    /// Rank of an input row under ascending, nulls-first order: null is the
    /// lowest rank, and tied values share a rank.
    fn rank(&self, row: usize) -> usize {
        if self.is_null_row(row) {
            0
        } else {
            self.group(row) + 1
        }
    }

    /// Baseline comparison of two input rows under ascending, nulls-first
    /// order: null sorts below every value and equals null.
    fn expected_cmp(&self, left: usize, right: usize) -> Ordering {
        self.rank(left).cmp(&self.rank(right))
    }

    /// Checks `cmp` against the baseline for every ordered pair of `rows`,
    /// reporting the first disagreement.
    fn check_pairwise(
        &self,
        path: &str,
        rows: &[usize],
        cmp: impl Fn(usize, usize) -> Ordering,
    ) -> Result<(), String> {
        for &left in rows {
            for &right in rows {
                let expected = self.expected_cmp(left, right);
                let actual = cmp(left, right);
                if expected != actual {
                    return Err(format!(
                        "fixture {} via {}: cmp({}, {}) expected {:?}, got {:?}",
                        self.name,
                        path,
                        self.label(left),
                        self.label(right),
                        expected,
                        actual,
                    ));
                }
            }
        }
        Ok(())
    }

    fn all_rows(&self) -> Vec<usize> {
        (0..self.len()).collect()
    }

    fn value_batch(&self, rows: &[usize]) -> RecordBatch {
        let indices = UInt64Array::from_iter_values(rows.iter().map(|&row| row as u64));
        let values = arrow_select::take::take(self.input.as_ref(), &indices, None).unwrap();
        RecordBatch::try_new(
            Arc::new(Schema::new(vec![
                Field::new("value", self.input.data_type().clone(), true),
                Field::new("row", DataType::UInt64, false),
            ])),
            vec![values, Arc::new(indices)],
        )
        .unwrap()
    }
}

fn native_cmp<T: ArrowPrimitiveType>(array: &dyn Array, left: usize, right: usize) -> Ordering {
    let array = array.as_primitive::<T>();
    array.value(left).compare(array.value(right))
}

fn no_nan<T>(_: &T) -> bool {
    false
}

fn float64() -> Fixture {
    build(
        "Float64",
        vec![
            ("-NaN(all payload bits)", 0xFFFF_FFFF_FFFF_FFFF_u64),
            ("-NaN(quiet)", 0xFFF8_0000_0000_0000),
            ("-inf", 0xFFF0_0000_0000_0000),
            ("-MAX", 0xFFEF_FFFF_FFFF_FFFF),
            ("-1.5", 0xBFF8_0000_0000_0000),
            ("-MIN_POSITIVE", 0x8010_0000_0000_0000),
            ("-smallest subnormal", 0x8000_0000_0000_0001),
            ("-0.0", 0x8000_0000_0000_0000),
            ("+0.0", 0x0000_0000_0000_0000),
            ("+smallest subnormal", 0x0000_0000_0000_0001),
            ("+MIN_POSITIVE", 0x0010_0000_0000_0000),
            ("1.5", 0x3FF8_0000_0000_0000),
            ("1.5 (duplicate)", 0x3FF8_0000_0000_0000),
            ("+MAX", 0x7FEF_FFFF_FFFF_FFFF),
            ("+inf", 0x7FF0_0000_0000_0000),
            ("+NaN(signaling)", 0x7FF0_0000_0000_0001),
            ("+NaN(quiet)", 0x7FF8_0000_0000_0000),
            ("+NaN(quiet, duplicate)", 0x7FF8_0000_0000_0000),
            ("+NaN(quiet, payload 1)", 0x7FF8_0000_0000_0001),
            ("+NaN(all payload bits)", 0x7FFF_FFFF_FFFF_FFFF),
        ],
        2,
        |bits| bits.to_le_bytes().to_vec(),
        |bits| f64::from_bits(*bits).is_nan(),
        |values| {
            Arc::new(Float64Array::from_iter(
                values.into_iter().map(|bits| bits.map(f64::from_bits)),
            ))
        },
        |array, row| {
            array
                .as_primitive::<Float64Type>()
                .value(row)
                .to_bits()
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<Float64Type>),
    )
}

fn float32() -> Fixture {
    build(
        "Float32",
        vec![
            ("-NaN(all payload bits)", 0xFFFF_FFFF_u32),
            ("-NaN(quiet)", 0xFFC0_0000),
            ("-inf", 0xFF80_0000),
            ("-MAX", 0xFF7F_FFFF),
            ("-1.5", 0xBFC0_0000),
            ("-MIN_POSITIVE", 0x8080_0000),
            ("-smallest subnormal", 0x8000_0001),
            ("-0.0", 0x8000_0000),
            ("+0.0", 0x0000_0000),
            ("+smallest subnormal", 0x0000_0001),
            ("+MIN_POSITIVE", 0x0080_0000),
            ("1.5", 0x3FC0_0000),
            ("1.5 (duplicate)", 0x3FC0_0000),
            ("+MAX", 0x7F7F_FFFF),
            ("+inf", 0x7F80_0000),
            ("+NaN(signaling)", 0x7F80_0001),
            ("+NaN(quiet)", 0x7FC0_0000),
            ("+NaN(quiet, duplicate)", 0x7FC0_0000),
            ("+NaN(quiet, payload 1)", 0x7FC0_0001),
            ("+NaN(all payload bits)", 0x7FFF_FFFF),
        ],
        2,
        |bits| bits.to_le_bytes().to_vec(),
        |bits| f32::from_bits(*bits).is_nan(),
        |values| {
            Arc::new(Float32Array::from_iter(
                values.into_iter().map(|bits| bits.map(f32::from_bits)),
            ))
        },
        |array, row| {
            array
                .as_primitive::<Float32Type>()
                .value(row)
                .to_bits()
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<Float32Type>),
    )
}

fn float16() -> Fixture {
    build(
        "Float16",
        vec![
            ("-NaN(all payload bits)", 0xFFFF_u16),
            ("-NaN(quiet)", 0xFE00),
            ("-inf", 0xFC00),
            ("-MAX", 0xFBFF),
            ("-1.5", 0xBE00),
            ("-MIN_POSITIVE", 0x8400),
            ("-smallest subnormal", 0x8001),
            ("-0.0", 0x8000),
            ("+0.0", 0x0000),
            ("+smallest subnormal", 0x0001),
            ("+MIN_POSITIVE", 0x0400),
            ("1.5", 0x3E00),
            ("1.5 (duplicate)", 0x3E00),
            ("+MAX", 0x7BFF),
            ("+inf", 0x7C00),
            ("+NaN(signaling)", 0x7C01),
            ("+NaN(quiet)", 0x7E00),
            ("+NaN(quiet, duplicate)", 0x7E00),
            ("+NaN(quiet, payload 1)", 0x7E01),
            ("+NaN(all payload bits)", 0x7FFF),
        ],
        2,
        |bits| bits.to_le_bytes().to_vec(),
        |bits| f16::from_bits(*bits).is_nan(),
        |values| {
            Arc::new(Float16Array::from_iter(
                values.into_iter().map(|bits| bits.map(f16::from_bits)),
            ))
        },
        |array, row| {
            array
                .as_primitive::<Float16Type>()
                .value(row)
                .to_bits()
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<Float16Type>),
    )
}

/// Byte-wise order: shorter prefixes first, ASCII upper case before lower
/// case, and multi-byte UTF-8 after every ASCII byte. The precomposed `é`
/// (C3 A9) and the decomposed `e` + U+0301 (65 CC 81) are distinct values far
/// apart from each other.
fn string_values() -> Vec<(&'static str, &'static str)> {
    vec![
        ("\"\"", ""),
        ("\"A\"", "A"),
        ("\"AB\"", "AB"),
        ("\"Ab\"", "Ab"),
        ("\"a\"", "a"),
        ("\"ab\"", "ab"),
        ("\"ab\" (duplicate)", "ab"),
        ("\"abc\"", "abc"),
        ("\"b\"", "b"),
        ("\"e\\u{301}\" (decomposed é)", "e\u{301}"),
        ("\"z\"", "z"),
        ("\"\\u{e9}\" (precomposed é)", "\u{e9}"),
        ("\"\\u{1F600}\" (emoji)", "\u{1F600}"),
    ]
}

fn utf8() -> Fixture {
    build(
        "Utf8",
        string_values(),
        2,
        |s| s.as_bytes().to_vec(),
        no_nan,
        |values| Arc::new(StringArray::from(values)),
        |array, row| array.as_string::<i32>().value(row).as_bytes().to_vec(),
        None,
    )
}

fn large_utf8() -> Fixture {
    build(
        "LargeUtf8",
        string_values(),
        2,
        |s| s.as_bytes().to_vec(),
        no_nan,
        |values| Arc::new(LargeStringArray::from(values)),
        |array, row| array.as_string::<i64>().value(row).as_bytes().to_vec(),
        None,
    )
}

fn binary() -> Fixture {
    build(
        "Binary",
        vec![
            ("[]", vec![]),
            ("[00]", vec![0x00]),
            ("[00 00]", vec![0x00, 0x00]),
            ("[00 01]", vec![0x00, 0x01]),
            ("[01]", vec![0x01]),
            ("[7F]", vec![0x7F]),
            ("[80]", vec![0x80]),
            ("[80] (duplicate)", vec![0x80]),
            ("[FF]", vec![0xFF]),
            ("[FF 00]", vec![0xFF, 0x00]),
        ],
        1,
        |bytes| bytes.clone(),
        no_nan,
        |values| Arc::new(BinaryArray::from_iter(values)),
        |array, row| array.as_binary::<i32>().value(row).to_vec(),
        None,
    )
}

fn int64() -> Fixture {
    build(
        "Int64",
        vec![
            ("MIN", i64::MIN),
            ("-1", -1),
            ("0", 0),
            ("0 (duplicate)", 0),
            ("1", 1),
            ("MAX", i64::MAX),
        ],
        1,
        |v| v.to_le_bytes().to_vec(),
        no_nan,
        |values| Arc::new(Int64Array::from(values)),
        |array, row| {
            array
                .as_primitive::<Int64Type>()
                .value(row)
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<Int64Type>),
    )
}

/// `2^63` has the sign bit set; a signed misread would place it below zero.
fn uint64() -> Fixture {
    build(
        "UInt64",
        vec![
            ("0", 0_u64),
            ("1", 1),
            ("2^63", 0x8000_0000_0000_0000),
            ("MAX", u64::MAX),
        ],
        1,
        |v| v.to_le_bytes().to_vec(),
        no_nan,
        |values| Arc::new(UInt64Array::from(values)),
        |array, row| {
            array
                .as_primitive::<UInt64Type>()
                .value(row)
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<UInt64Type>),
    )
}

fn date32() -> Fixture {
    build(
        "Date32",
        vec![
            ("MIN", i32::MIN),
            ("-1", -1),
            ("0 (epoch)", 0),
            ("MAX", i32::MAX),
        ],
        1,
        |v| v.to_le_bytes().to_vec(),
        no_nan,
        |values| Arc::new(Date32Array::from(values)),
        |array, row| {
            array
                .as_primitive::<Date32Type>()
                .value(row)
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<Date32Type>),
    )
}

fn timestamp_ns_utc() -> Fixture {
    build(
        "Timestamp(ns, UTC)",
        vec![
            ("MIN", i64::MIN),
            ("-1", -1),
            ("0 (epoch)", 0),
            ("MAX", i64::MAX),
        ],
        1,
        |v| v.to_le_bytes().to_vec(),
        no_nan,
        |values| Arc::new(TimestampNanosecondArray::from(values).with_timezone("UTC")),
        |array, row| {
            array
                .as_primitive::<TimestampNanosecondType>()
                .value(row)
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<TimestampNanosecondType>),
    )
}

/// The largest magnitude representable at precision 38: thirty-eight nines.
const DECIMAL38_MAX: i128 = 99_999_999_999_999_999_999_999_999_999_999_999_999;

fn decimal128() -> Fixture {
    build(
        "Decimal128(38, 10)",
        vec![
            ("-(10^38 - 1)", -DECIMAL38_MAX),
            ("-1", -1),
            ("0", 0),
            ("1", 1),
            ("10^38 - 1", DECIMAL38_MAX),
        ],
        1,
        |v| v.to_le_bytes().to_vec(),
        no_nan,
        |values| {
            Arc::new(
                Decimal128Array::from(values)
                    .with_precision_and_scale(38, 10)
                    .unwrap(),
            )
        },
        |array, row| {
            array
                .as_primitive::<Decimal128Type>()
                .value(row)
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<Decimal128Type>),
    )
}

fn decimal256() -> Fixture {
    build(
        "Decimal256(76, 10)",
        vec![
            ("-(10^38 - 1)", i256::from_i128(-DECIMAL38_MAX)),
            ("-1", i256::from_i128(-1)),
            ("0", i256::ZERO),
            ("1", i256::ONE),
            ("10^38 - 1", i256::from_i128(DECIMAL38_MAX)),
        ],
        1,
        |v| v.to_le_bytes().to_vec(),
        no_nan,
        |values| {
            Arc::new(
                Decimal256Array::from(values)
                    .with_precision_and_scale(76, 10)
                    .unwrap(),
            )
        },
        |array, row| {
            array
                .as_primitive::<Decimal256Type>()
                .value(row)
                .to_le_bytes()
                .to_vec()
        },
        Some(native_cmp::<Decimal256Type>),
    )
}

fn boolean() -> Fixture {
    build(
        "Boolean",
        vec![
            ("false", false),
            ("false (duplicate)", false),
            ("true", true),
        ],
        1,
        |v| vec![u8::from(*v)],
        no_nan,
        |values| Arc::new(BooleanArray::from(values)),
        |array, row| vec![u8::from(array.as_boolean().value(row))],
        None,
    )
}

/// Every fixture, so that each path below covers the same set of types. A
/// type added here is covered by every test without touching them.
fn fixtures() -> Vec<Fixture> {
    vec![
        float64(),
        float32(),
        float16(),
        utf8(),
        large_utf8(),
        binary(),
        int64(),
        uint64(),
        date32(),
        timestamp_ns_utc(),
        decimal128(),
        decimal256(),
        boolean(),
    ]
}

const ALL_OPTIONS: [SortOptions; 4] = [
    ASC_NULLS_FIRST,
    ASC_NULLS_LAST,
    DESC_NULLS_FIRST,
    DESC_NULLS_LAST,
];

/// Collects every disagreement in a test so one run reports all affected
/// types and options, not only the first.
#[derive(Default)]
struct Failures(Vec<String>);

impl Failures {
    fn check(&mut self, result: Result<(), String>) {
        if let Err(message) = result {
            self.0.push(message);
        }
    }

    fn finish(self) {
        assert!(
            self.0.is_empty(),
            "{} ordering disagreement(s):\n{}",
            self.0.len(),
            self.0.join("\n")
        );
    }
}

#[test]
fn default_sort_options_are_ascending_nulls_first() {
    // `RecordBatchExt::sort_by_column(.., None)`, the btree flat index, the
    // hilbert sort and `Dataset::delta` all rely on this default.
    assert_eq!(SortOptions::default(), ASC_NULLS_FIRST);
}

#[test]
fn sort_to_indices_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        for opts in ALL_OPTIONS {
            let indices = sort_to_indices(fixture.input.as_ref(), Some(opts), None).unwrap();
            let rows: Vec<usize> = indices.values().iter().map(|&i| i as usize).collect();
            failures.check(fixture.check_order("arrow_ord::sort::sort_to_indices", opts, &rows));
        }
    }
    failures.finish();
}

#[test]
fn lexsort_to_indices_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        for opts in ALL_OPTIONS {
            // A second column keeps arrow on the multi-column comparator path
            // that `SortExec` uses for `ORDER BY a, b`.
            let row_ids: ArrayRef =
                Arc::new(UInt64Array::from_iter_values(0..fixture.len() as u64));
            let indices = lexsort_to_indices(
                &[
                    SortColumn {
                        values: Arc::clone(&fixture.input),
                        options: Some(opts),
                    },
                    SortColumn {
                        values: row_ids,
                        options: None,
                    },
                ],
                None,
            )
            .unwrap();
            let rows: Vec<usize> = indices.values().iter().map(|&i| i as usize).collect();
            failures.check(fixture.check_order("arrow_ord::sort::lexsort_to_indices", opts, &rows));
        }
    }
    failures.finish();
}

#[test]
fn make_comparator_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        let cmp = make_comparator(
            fixture.input.as_ref(),
            fixture.input.as_ref(),
            ASC_NULLS_FIRST,
        )
        .unwrap();
        failures.check(fixture.check_pairwise(
            "arrow_ord::ord::make_comparator",
            &fixture.all_rows(),
            cmp,
        ));
    }
    failures.finish();
}

#[test]
fn native_compare_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        // Only primitive fixtures have a native comparator.
        let Some(cmp) = fixture.native_cmp else {
            continue;
        };
        failures.check(fixture.check_pairwise(
            "ArrowNativeTypeOp::compare",
            &fixture.non_null_rows(),
            |left, right| cmp(fixture.input.as_ref(), left, right),
        ));
    }
    failures.finish();
}

/// `RowConverter` with the caller's `SortOptions` is what the hash joiner,
/// the mem-WAL `FirstByPk` operator and DataFusion's TopK and row cursors
/// compare with, so every option combination is pinned, not only the default.
#[test]
fn row_converter_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        for opts in ALL_OPTIONS {
            let converter = RowConverter::new(vec![SortField::new_with_options(
                fixture.input.data_type().clone(),
                opts,
            )])
            .unwrap();
            let rows = converter
                .convert_columns(&[Arc::clone(&fixture.input)])
                .unwrap();
            let mut order = fixture.all_rows();
            order.sort_by(|&left, &right| rows.row(left).cmp(&rows.row(right)));
            failures.check(fixture.check_order("arrow_row::RowConverter", opts, &order));
        }
    }
    failures.finish();
}

#[test]
fn arrow_scalar_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        let scalars: Vec<ArrowScalar> = (0..fixture.len())
            .map(|row| ArrowScalar::try_new(&fixture.input, row).unwrap())
            .collect();
        failures.check(fixture.check_pairwise(
            "ArrowScalar::cmp",
            &fixture.all_rows(),
            |left, right| scalars[left].cmp(&scalars[right]),
        ));
        // The row encoding also decides key equality (hash joins, merge
        // insert): `-0.0 != +0.0`, equal NaN bits are equal, null equals null.
        failures.check(fixture.check_pairwise(
            "ArrowScalar::eq",
            &fixture.all_rows(),
            |left, right| {
                if scalars[left] == scalars[right] {
                    Ordering::Equal
                } else {
                    fixture.expected_cmp(left, right).then(Ordering::Less)
                }
            },
        ));
    }
    failures.finish();
}

#[test]
fn scalar_value_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        let scalars: Vec<ScalarValue> = (0..fixture.len())
            .map(|row| ScalarValue::try_from_array(fixture.input.as_ref(), row).unwrap())
            .collect();
        // The btree compares query bounds with its own `Ord` wrapper.
        let orderable: Vec<OrderableScalarValue> = scalars
            .iter()
            .map(|scalar| OrderableScalarValue(scalar.clone()))
            .collect();
        failures.check(fixture.check_pairwise(
            "OrderableScalarValue::cmp",
            &fixture.all_rows(),
            |left, right| orderable[left].cmp(&orderable[right]),
        ));
        // The zone map compares query values to zone extrema with DataFusion's
        // `PartialOrd`.
        failures.check(fixture.check_pairwise(
            "ScalarValue::partial_cmp",
            &fixture.all_rows(),
            |left, right| {
                scalars[left]
                    .partial_cmp(&scalars[right])
                    .unwrap_or_else(|| {
                        panic!(
                            "fixture {}: ScalarValue::partial_cmp({}, {}) returned None",
                            fixture.name,
                            fixture.label(left),
                            fixture.label(right)
                        )
                    })
            },
        ));
    }
    failures.finish();
}

#[test]
fn statistics_extrema_match_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        let mut whole = StatisticsAccumulator::new(fixture.input.data_type());
        whole.update(&fixture.input).unwrap();

        // Merging two halves goes through `ArrowScalar` ordering and must agree.
        let split = fixture.len() / 2;
        let mut merged = StatisticsAccumulator::new(fixture.input.data_type());
        merged.update(&fixture.input.slice(0, split)).unwrap();
        let mut rest = StatisticsAccumulator::new(fixture.input.data_type());
        rest.update(&fixture.input.slice(split, fixture.len() - split))
            .unwrap();
        merged.merge(&rest).unwrap();

        let min_pos = fixture.is_nan.iter().position(|nan| !nan);
        let max_pos = fixture.is_nan.iter().rposition(|nan| !nan);
        let nan_count = fixture.is_nan.iter().filter(|nan| **nan).count() as u64;
        let is_float = matches!(
            fixture.input.data_type(),
            DataType::Float16 | DataType::Float32 | DataType::Float64
        );

        for (path, stats) in [
            ("StatisticsAccumulator::update", whole.finish()),
            ("StatisticsAccumulator::merge", merged.finish()),
        ] {
            if stats.null_count != fixture.null_count as u64 {
                failures.check(Err(format!(
                    "fixture {} via {path}: null_count expected {}, got {}",
                    fixture.name, fixture.null_count, stats.null_count
                )));
            }
            if stats.nan_count != is_float.then_some(nan_count) {
                failures.check(Err(format!(
                    "fixture {} via {path}: nan_count expected {:?}, got {:?}",
                    fixture.name,
                    is_float.then_some(nan_count),
                    stats.nan_count
                )));
            }
            for (which, expected_pos, actual) in
                [("min", min_pos, &stats.min), ("max", max_pos, &stats.max)]
            {
                let expected_label = expected_pos.map(|pos| fixture.labels[pos]);
                let actual_identity = actual
                    .as_ref()
                    .map(|scalar| (fixture.identity_of)(scalar.as_array().as_ref(), 0));
                let expected_identity = expected_pos.map(|pos| fixture.identities[pos].clone());
                if actual_identity != expected_identity {
                    failures.check(Err(format!(
                        "fixture {} via {path}: {which} expected {:?} ({:?}), got {:?} ({:?})",
                        fixture.name,
                        expected_label,
                        expected_identity,
                        actual.as_ref().map(|scalar| scalar.to_string()),
                        actual_identity,
                    )));
                }
            }
        }
    }
    failures.finish();
}

/// Signed zeros are the one pair that `==` calls equal and the total order
/// does not. The mixed fixtures above always have the infinities as extrema,
/// so only an input made of zeros shows whether statistics still tell them
/// apart: the baseline is `min = -0.0`, `max = +0.0`, by bit pattern, in
/// either input order and whether the two arrive in one batch or in two
/// accumulators that are merged.
#[rstest]
#[case::f64_neg_first(Arc::new(Float64Array::from(vec![
    f64::from_bits(0x8000_0000_0000_0000),
    f64::from_bits(0),
])))]
#[case::f64_pos_first(Arc::new(Float64Array::from(vec![
    f64::from_bits(0),
    f64::from_bits(0x8000_0000_0000_0000),
])))]
#[case::f32_neg_first(Arc::new(Float32Array::from(vec![
    f32::from_bits(0x8000_0000),
    f32::from_bits(0),
])))]
#[case::f32_pos_first(Arc::new(Float32Array::from(vec![
    f32::from_bits(0),
    f32::from_bits(0x8000_0000),
])))]
#[case::f16_neg_first(Arc::new(Float16Array::from(vec![
    f16::from_bits(0x8000),
    f16::from_bits(0),
])))]
#[case::f16_pos_first(Arc::new(Float16Array::from(vec![
    f16::from_bits(0),
    f16::from_bits(0x8000),
])))]
fn statistics_keep_signed_zeros_apart(#[case] input: ArrayRef) {
    let (identity_of, width) = match input.data_type() {
        DataType::Float64 => (float64().identity_of, 8),
        DataType::Float32 => (float32().identity_of, 4),
        DataType::Float16 => (float16().identity_of, 2),
        other => panic!("unexpected type {other}"),
    };
    // IEEE zeros: every bit clear is `+0.0`; the sign bit alone is `-0.0`.
    // Identities are little-endian, so the sign lands in the last byte.
    let positive_zero = vec![0u8; width];
    let mut negative_zero = vec![0u8; width];
    negative_zero[width - 1] = 0x80;

    let mut whole = StatisticsAccumulator::new(input.data_type());
    whole.update(&input).unwrap();

    // Each value in its own accumulator, merged in input order, goes through
    // `ArrowScalar` ordering rather than the per-batch scan.
    let mut merged = StatisticsAccumulator::new(input.data_type());
    merged.update(&input.slice(0, 1)).unwrap();
    let mut second = StatisticsAccumulator::new(input.data_type());
    second.update(&input.slice(1, 1)).unwrap();
    merged.merge(&second).unwrap();

    for (path, stats) in [
        ("StatisticsAccumulator::update", whole.finish()),
        ("StatisticsAccumulator::merge", merged.finish()),
    ] {
        let min = stats
            .min
            .as_ref()
            .map(|s| identity_of(s.as_array().as_ref(), 0));
        let max = stats
            .max
            .as_ref()
            .map(|s| identity_of(s.as_array().as_ref(), 0));
        assert!(
            min.as_ref() == Some(&negative_zero),
            "{} via {path}: min expected -0.0 ({negative_zero:?}), got {:?} ({min:?})",
            input.data_type(),
            stats.min.as_ref().map(|s| s.to_string()),
        );
        assert!(
            max.as_ref() == Some(&positive_zero),
            "{} via {path}: max expected +0.0 ({positive_zero:?}), got {:?} ({max:?})",
            input.data_type(),
            stats.max.as_ref().map(|s| s.to_string()),
        );
    }
}

fn one_shot(batches: Vec<RecordBatch>, schema: Arc<Schema>) -> Arc<dyn ExecutionPlan> {
    let stream =
        RecordBatchStreamAdapter::new(schema, futures::stream::iter(batches.into_iter().map(Ok)));
    Arc::new(OneShotExec::new(
        Box::pin(stream) as SendableRecordBatchStream
    ))
}

fn sort_expr(opts: SortOptions) -> PhysicalSortExpr {
    PhysicalSortExpr {
        expr: Arc::new(Column::new("value", 0)),
        options: opts,
    }
}

async fn output_rows(plan: Arc<dyn ExecutionPlan>) -> Vec<usize> {
    let stream = execute_plan(plan, LanceExecutionOptions::default()).unwrap();
    let schema = stream.schema();
    let batches: Vec<RecordBatch> = stream.try_collect().await.unwrap();
    let batch = concat_batches(&schema, &batches).unwrap();
    batch
        .column(1)
        .as_primitive::<UInt64Type>()
        .values()
        .iter()
        .map(|&row| row as usize)
        .collect()
}

/// `Scanner::order_by` builds exactly this plan: a `SortExec` with the
/// caller's `SortOptions` over an unsorted input. With a `limit` it becomes a
/// `SortExec` with a fetch, which DataFusion runs as a TopK over the row
/// encoding instead of the sort kernel; a fetch of the full length keeps the
/// whole order visible.
#[tokio::test]
async fn sort_exec_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        for opts in ALL_OPTIONS {
            let batch = fixture.value_batch(&fixture.all_rows());
            let input = one_shot(vec![batch.clone()], batch.schema());
            let plan = Arc::new(SortExec::new([sort_expr(opts)].into(), input));
            let rows = output_rows(plan).await;
            failures.check(fixture.check_order("datafusion SortExec", opts, &rows));

            let input = one_shot(vec![batch.clone()], batch.schema());
            let plan = Arc::new(
                SortExec::new([sort_expr(opts)].into(), input).with_fetch(Some(fixture.len())),
            );
            let rows = output_rows(plan).await;
            failures.check(fixture.check_order(
                "datafusion SortExec with fetch (TopK)",
                opts,
                &rows,
            ));
        }
    }
    failures.finish();
}

/// Btree training merges several already-sorted inputs with a
/// `SortPreservingMergeExec` under a `UnionExec`, which compares through
/// DataFusion's merge cursors rather than through a sort kernel. The inputs
/// here are sorted by the baseline, not by arrow, so the merge order is only
/// right if the cursors agree with the baseline.
#[tokio::test]
async fn sort_preserving_merge_matches_baseline() {
    let mut failures = Failures::default();
    for fixture in fixtures() {
        let baseline = fixture.baseline_rows(ASC_NULLS_FIRST);
        let partitions = 3;
        let schema = fixture.value_batch(&[]).schema();
        let inputs: Vec<Arc<dyn ExecutionPlan>> = (0..partitions)
            .map(|partition| {
                let rows: Vec<usize> = baseline
                    .iter()
                    .copied()
                    .skip(partition)
                    .step_by(partitions)
                    .collect();
                one_shot(vec![fixture.value_batch(&rows)], Arc::clone(&schema))
            })
            .collect();
        let union = UnionExec::try_new(inputs).unwrap();
        let plan = Arc::new(SortPreservingMergeExec::new(
            [sort_expr(ASC_NULLS_FIRST)].into(),
            union,
        ));
        let rows = output_rows(plan).await;
        failures.check(fixture.check_order(
            "datafusion SortPreservingMergeExec",
            ASC_NULLS_FIRST,
            &rows,
        ));
    }
    failures.finish();
}