rudb-exec 0.4.3

//! Sorting.
//!
//! A pipeline breaker: nothing can be emitted until the last row has arrived, because the last row
//! of the input can be the first row of the output. It reads everything, sorts, and then hands the
//! result out a chunk at a time.
//!
//! The sort is Rust's `sort_by`, which is stable. Stability is not something SQL promises and it is
//! kept anyway, because `ORDER BY a` over rows that tie on `a` producing a different order on two
//! runs of the same query is the kind of difference that makes a compatibility diff useless.
//!
//! # Keeping it stable on more than one thread
//!
//! A stable sort is stable in the order the rows were given to it, and on several threads that is
//! the order the threads happened to finish. So every row carries where it arrived from, which is
//! the morsel it came out of and its place in that morsel, and two rows that tie on every key are
//! separated by that instead. Reading it lexicographically is exactly the order one thread would
//! have produced, because one thread takes the morsels in the order they were cut and reads each
//! one through from the start, so a parallel sort answers what a serial sort answers rather than
//! answering something SQL also allows.
//!
//! A filter between the scan and the sort does not break that. The place a row carries is its place
//! among the rows that reached the sort rather than its row number in the file, and the rows that
//! reach the sort from one morsel still reach it in order.
//!
//! # Where the payload is while the sort runs
//!
//! Not in the sort. The chunks are kept as they arrived and a row is a chunk and a row in it, so
//! what the comparator moves is the keys and two pairs of numbers rather than a copy of every
//! column. The columns are moved once at the end, by [`gathered`], which hands each column's
//! pieces to an [`Assembly`] and lets it do the interleave as a typed copy per physical layout.
//!
//! This used to hold a `Vec<Value>` of the whole row per row. Sorting lineitem at SF1 on three
//! keys cost 197 billion instructions that way, of which most were the allocator: sixteen columns
//! a row over six million rows is around a hundred million `Value`s and thirty million of them
//! were strings. The same sort is 54 billion now, and the system time it spends asking the
//! operating system for memory went from 45.7 seconds to 2.8. See #1210.
//!
//! # And where the key is
//!
//! In the row, as bytes, whenever the key list allows it. [`crate::normal`] writes every key of a
//! row into one fixed width buffer in an encoding whose byte order is the sort order, with the
//! direction and the null placement already folded in, so the comparator is a byte compare and the
//! row is one flat thing rather than a pointer to a heap allocation per row. The `Vec<Value>` path
//! is still here and still handles everything, because a string key has no fixed width and a float
//! key does not order the way its bytes do. Which one a sort takes is decided once, from the types
//! of the key expressions, and [`Keyed`] is the two of them.
//!
//! # The shape a sink has
//!
//! [`Sort`] is a [`Sink`], so the rows arrive through `sink`, one instance's rows are handed over
//! through `combine`, and `finalize` does the sort once after every instance has combined. On one
//! thread that is the same work in the same order as reading the input in a loop would be. On
//! several it is the shape that makes the sort possible at all, and having it now is why F4 changes
//! no operator.
//!
//! The finished chunks go into a [`Buffered`], which is a separate source rather than something
//! `finalize` hands back, for the reason [`Sink::finalize`] gives.
//!
//! # Giving the input back while the output is being built
//!
//! A sort holds the rows that arrived and the rows it is handing out at the same time, and for a
//! moment near the end it holds both in full. That moment is what a big sort dies at: SF10 lineitem
//! is around ten gigabytes of payload, so two copies of it is twenty, and the limit on a machine
//! with 24 GiB lands at 19.1.
//!
//! It does not have to hold both. [`gathered`] lays one column at a time, so the input's copy of a
//! column is finished with the moment that column has been laid, and the input is taken apart into
//! its columns up front so that each one can be dropped exactly then. What the sort holds is
//! therefore one payload and one column of headroom rather than two payloads, whichever column it
//! is on, and the charge against the memory limit comes down as the columns go.
//!
//! The rows themselves go before any of that. All [`gathered`] wants from them is where each input
//! row lands, which is four bytes a row, against the forty eight a row that carries a normalized
//! key and an arrival. So the order is turned into that and the rows are dropped, which at SF10 is
//! another three gigabytes that is not held while the assembly runs.

use std::cmp::Ordering;
use std::sync::Mutex;

use rudb_common::{Error, LogicalType, Memory, Reservation, Result, Session, Value};
use rudb_pipeline::{Lease, Progress, Sink};
use rudb_plan::{Plan, Slice, SortKey};
use rudb_vector::{Assembly, Chunk, VECTOR_SIZE, Vector};

use crate::buffer::Buffered;
use crate::normal::{self, Normal};
use crate::prepared::{Prepared, Scratch};
use crate::rows;
use crate::schema::Schema;

/// One row on its way through a sort: the values of its keys, where it arrived, and where it is.
///
/// The payload is not here. A row is a [`Source`] into the chunks the sink kept, and the columns
/// are moved once at the end by [`gathered`] rather than carried through the sort as a boxed value
/// a field. That is the second half of what the note on #63 asks for, and [`Normalized`] is the
/// first half: this arm is what a key list with a string, a float or too many bytes in it falls
/// back to.
pub(crate) type Sortable = (Vec<Value>, Arrival, Source);

/// Where a row arrived: the morsel it came from and its place among the rows of that morsel.
///
/// Sixteen bytes beside a `Vec` header and whatever it points at, which is why it is carried per
/// row rather than reconstructed. What it buys is that the answer does not depend on how many
/// threads ran.
pub(crate) type Arrival = (u64, u64);

/// Where a row is: the chunk the sink kept it in and its row in that chunk.
pub(crate) type Source = (u32, u32);

/// One row on its way through a sort with its keys written as bytes rather than held as values.
///
/// The same three fields as [`Sortable`] with the first one flattened. Nothing here points at
/// anything: the key is a fixed array in the row, so the whole vector is one allocation and a
/// comparison is a byte compare over that array. See [`crate::normal`] for what is in it and which
/// key lists can have one.
pub(crate) type Normalized = (Normal, Arrival, Source);

/// What one row costs beside the values of its keys, which [`rows::footprint`] already counts.
const BESIDE: u64 = (size_of::<Sortable>() - size_of::<Vec<Value>>()) as u64;

/// What one row of the normalized path costs, which is the whole of it since nothing is borrowed.
const NORMALIZED: u64 = size_of::<Normalized>() as u64;

/// An ordering over the input.
#[derive(Debug)]
pub(crate) struct Sort {
    keys: Vec<SortKey>,
    /// How wide each key writes into a normalized key, when the key list has one.
    ///
    /// `None` is the `Value` path, and it is what a string key, a float key or a key list that does
    /// not fit gets. See [`crate::normal::layout`].
    widths: Option<Vec<usize>>,
    /// The key expressions, evaluated against the input's schema.
    exprs: Prepared,
    /// The input's types, which are also the output's, since a sort changes no column.
    types: Vec<LogicalType>,
    memory: Memory,
    /// Every instance's rows and the chunks they point into, waiting for the sort.
    gathered: Mutex<Combined>,
    /// What those rows are charged, taken from the instances that gathered them and given back
    /// once the sorted chunks have been charged instead.
    charged: Mutex<Vec<Reservation>>,
    /// What the sorted chunks are charged, held for as long as they are readable.
    held: Mutex<Reservation>,
    out: Buffered,
}

/// Every instance's rows after they have been handed over, in one lock rather than two.
///
/// The two halves are read and written together and a row is an index into the chunks beside it,
/// so a pair of locks would be two that always have to be taken in the same order and nothing
/// would ever hold one of them alone.
#[derive(Debug, Default)]
struct Combined {
    /// The chunks as they arrived, which hold the payload of every row.
    chunks: Vec<Chunk>,
    /// One entry a row, pointing into `chunks`.
    rows: Keyed,
}

/// The rows of a sort, with their keys held whichever way this key list allows.
///
/// Two arms and not two operators, because everything either arm does differently is in this file
/// and everything else about a sort is the same: the same chunks, the same arrivals, the same
/// assembly at the end. Which arm a sort takes is decided once in [`Sort::new`], off the types of
/// the key expressions, so an instance never has to ask and the two can never be mixed.
#[derive(Debug)]
enum Keyed {
    /// Keys as bytes, which is the fast path and covers the fixed width types.
    Normal(Vec<Normalized>),
    /// Keys as values, which handles every type including the ones with no fixed width.
    Valued(Vec<Sortable>),
}

impl Default for Keyed {
    fn default() -> Self {
        Self::Valued(Vec::new())
    }
}

impl Keyed {
    /// An empty set of rows of the same arm as this one.
    fn empty(&self) -> Self {
        match self {
            Self::Normal(_) => Self::Normal(Vec::new()),
            Self::Valued(_) => Self::Valued(Vec::new()),
        }
    }

    /// Takes another instance's rows, moving every row's chunk along by `base`.
    ///
    /// The chunk an instance's row points at is its chunk among that instance's, so it moves along
    /// by however many chunks are already here.
    fn absorb(&mut self, other: Self, base: u32) -> Result<()> {
        match (self, other) {
            (Self::Normal(into), Self::Normal(from)) => {
                into.extend(from.into_iter().map(|(key, arrival, (chunk, row))| {
                    (key, arrival, (chunk.saturating_add(base), row))
                }));
                Ok(())
            }
            (Self::Valued(into), Self::Valued(from)) => {
                into.extend(from.into_iter().map(|(key, arrival, (chunk, row))| {
                    (key, arrival, (chunk.saturating_add(base), row))
                }));
                Ok(())
            }
            _ => Err(Error::internal("two instances of one sort holding their keys differently")),
        }
    }

    /// Puts the rows in order, keys first and where they arrived settling a tie.
    ///
    /// Unstable on the normalized arm and stable on the other, which is the same order either way:
    /// the arrival is unique per row and it is the last thing compared, so no two rows are ever
    /// equal and there is nothing for stability to decide.
    fn sort(&mut self, keys: &[SortKey]) -> Result<()> {
        match self {
            Self::Normal(rows) => {
                rows.sort_unstable_by(|left, right| {
                    left.0.cmp(&right.0).then_with(|| left.1.cmp(&right.1))
                });
                Ok(())
            }
            Self::Valued(rows) => {
                let mut failure: Option<Error> = None;
                rows.sort_by(|left, right| settled(keys, left, right, &mut failure));
                failure.map_or(Ok(()), Err)
            }
        }
    }

    /// Where each row sits, in the order the sort put them.
    fn sources(&self) -> Box<dyn ExactSizeIterator<Item = Source> + '_> {
        match self {
            Self::Normal(rows) => Box::new(rows.iter().map(|row| row.2)),
            Self::Valued(rows) => Box::new(rows.iter().map(|row| row.2)),
        }
    }

    /// How many rows there are.
    fn len(&self) -> usize {
        match self {
            Self::Normal(rows) => rows.len(),
            Self::Valued(rows) => rows.len(),
        }
    }

    /// What these rows were charged when they were taken in, so that dropping them can give it back.
    ///
    /// Counted again rather than carried, because the valued arm's rows are not all the same size
    /// and a running total would have to be threaded through the instance handover as a fourth
    /// thing that has to stay in step with the other three. One walk over the rows at the end of a
    /// sort is nothing beside the sort.
    fn footprint(&self) -> u64 {
        match self {
            Self::Normal(rows) => {
                u64::try_from(rows.len()).unwrap_or(u64::MAX).saturating_mul(NORMALIZED)
            }
            Self::Valued(rows) => rows.iter().map(|row| rows::footprint(&row.0) + BESIDE).sum(),
        }
    }
}

/// What one instance of a sort gathers before it combines.
#[derive(Debug)]
pub(crate) struct Gathered {
    held: Combined,
    scratch: Scratch,
    charged: Reservation,
    /// The morsel this instance is reading and how many of its rows have arrived.
    place: Place,
}

/// How far through a morsel an instance is, which is the second half of an [`Arrival`].
#[derive(Debug, Default)]
pub(crate) struct Place {
    pub(crate) morsel: u64,
    pub(crate) at: u64,
}

impl Place {
    /// Where the row at `row` of the chunk that starts here arrived.
    pub(crate) fn of(&self, row: usize) -> Arrival {
        (self.morsel, self.at.saturating_add(row as u64))
    }

    /// Moves past a chunk of `rows` rows of the same morsel.
    pub(crate) fn past(&mut self, rows: usize) {
        self.at = self.at.saturating_add(rows as u64);
    }

    /// Starts a new morsel.
    pub(crate) fn start(&mut self, morsel: u64) {
        self.morsel = morsel;
        self.at = 0;
    }
}

impl Sort {
    /// Applies the session semantics to the prepared sort keys.
    #[must_use]
    pub(crate) fn in_session(mut self, session: &Session) -> Self {
        self.exprs = self.exprs.in_session(session);
        self
    }

    /// # Errors
    ///
    /// If a sort key does not resolve against the input's schema, which is a failure of the plan
    /// and is found here rather than on the first chunk.
    pub(crate) fn new(
        plan: &Plan,
        input: &Schema,
        keys: Slice,
        memory: &Memory,
    ) -> Result<(Self, Buffered)> {
        let keys = plan.sort_key_list(keys).to_vec();
        let exprs: Vec<_> = keys.iter().map(|key| key.expr).collect();
        let types: Vec<_> = exprs.iter().map(|&expr| plan.expr_type(expr).clone()).collect();
        let out = Buffered::new();
        let widths = normal::layout(&types);
        let rows =
            if widths.is_some() { Keyed::Normal(Vec::new()) } else { Keyed::Valued(Vec::new()) };
        let sort = Self {
            widths,
            exprs: Prepared::new(plan, &exprs, input)?,
            keys,
            types: input.types(),
            memory: memory.clone(),
            gathered: Mutex::new(Combined { chunks: Vec::new(), rows }),
            charged: Mutex::new(Vec::new()),
            held: Mutex::new(memory.reservation()),
            out: out.clone(),
        };
        Ok((sort, out))
    }
}

impl Sink for Sort {
    type Local = Gathered;

    fn local(&self) -> Gathered {
        let rows = if self.widths.is_some() {
            Keyed::Normal(Vec::new())
        } else {
            Keyed::Valued(Vec::new())
        };
        Gathered {
            held: Combined { chunks: Vec::new(), rows },
            scratch: self.exprs.scratch(),
            charged: self.memory.reservation(),
            place: Place::default(),
        }
    }

    fn at(&self, morsel: &rudb_pipeline::Morsel, local: &mut Gathered) -> Result<()> {
        local.place.start(morsel.index());
        Ok(())
    }

    fn sink(&self, chunk: &Chunk, local: &mut Gathered) -> Result<Progress> {
        if chunk.is_empty() {
            return Ok(Progress::More);
        }
        let mut keys = Vec::with_capacity(self.keys.len());
        self.exprs.evaluate(chunk, &mut local.scratch, &mut keys)?;
        let at = u32::try_from(local.held.chunks.len()).map_err(|_| too_many())?;
        let mut taken = u64::try_from(chunk.footprint()).unwrap_or(u64::MAX);
        // row at a time: the keys, and nothing else. The payload is not read here at all, which is
        // the point of `Sortable`.
        match (&self.widths, &mut local.held.rows) {
            (Some(widths), Keyed::Normal(rows)) => {
                for row in 0..chunk.len() {
                    let mut key: Normal = [0; normal::WIDTH];
                    let mut written = 0;
                    for (at, column) in keys.iter().enumerate() {
                        let wide = *widths.get(at).ok_or_else(mismatched)?;
                        let value = column.try_value_at(row)?;
                        normal::write(&mut key, written, wide, &value, self.keys[at])?;
                        written += wide;
                    }
                    taken += NORMALIZED;
                    let row = u32::try_from(row).map_err(|_| too_many())?;
                    rows.push((key, local.place.of(row as usize), (at, row)));
                }
            }
            (None, Keyed::Valued(rows)) => {
                for row in 0..chunk.len() {
                    let key: Vec<Value> = keys
                        .iter()
                        .map(|column| column.try_value_at(row))
                        .collect::<Result<_>>()?;
                    taken += rows::footprint(&key) + BESIDE;
                    let row = u32::try_from(row).map_err(|_| too_many())?;
                    rows.push((key, local.place.of(row as usize), (at, row)));
                }
            }
            _ => return Err(mismatched()),
        }
        local.held.chunks.push(chunk.clone());
        local.place.past(chunk.len());
        local.charged.grow(taken)?;
        Ok(Progress::More)
    }

    fn combine(&self, local: Gathered) -> Result<()> {
        let mut gathered = self.gathered.lock().map_err(poisoned)?;
        // Appended rather than merged, because the sort has not happened yet. The order the
        // instances combine in does not decide anything, since every row carries where it arrived
        // and the comparison falls back to that when the keys tie.
        let base = u32::try_from(gathered.chunks.len()).map_err(|_| too_many())?;
        gathered.rows.absorb(local.held.rows, base)?;
        gathered.chunks.extend(local.held.chunks);
        self.charged.lock().map_err(poisoned)?.push(local.charged);
        Ok(())
    }

    fn finalize(&self, _threads: &Lease<'_>) -> Result<()> {
        let Combined { chunks, mut rows } = {
            let mut gathered = self.gathered.lock().map_err(poisoned)?;
            let empty = Combined { chunks: Vec::new(), rows: gathered.rows.empty() };
            std::mem::replace(&mut *gathered, empty)
        };
        rows.sort(&self.keys)?;
        // What the instances took, moved out so that it can be given back a column at a time rather
        // than all at once when this returns. Dropping what is left of it is what releases the rest.
        let mut charged = std::mem::take(&mut *self.charged.lock().map_err(poisoned)?);
        let total = rows.len();
        if u32::try_from(total).is_err() {
            return Err(too_many());
        }
        let at = places(&chunks, &rows)?;
        // The rows have said everything they had to say. Holding them through the assembly is
        // holding a key and an arrival a row for the sake of a number that is already in `at`.
        let taken = rows.footprint();
        drop(rows);
        give(&mut charged, taken);
        let mut held = self.held.lock().map_err(poisoned)?;
        let out = gathered(&self.types, chunks, &at, total, &mut held, &mut charged)?;
        self.out.fill(out)?;
        Ok(())
    }
}

/// Where each row that arrived lands, kept the way an [`Assembly`] wants to be handed it.
///
/// One run of positions a chunk, indexed by the row's place in that chunk. Every input row reaches
/// the sort and every one of them is somewhere in the order, so every position is written and the
/// zero this starts from is never read.
fn places(chunks: &[Chunk], rows: &Keyed) -> Result<Vec<Vec<u32>>> {
    let mut at: Vec<Vec<u32>> = chunks.iter().map(|chunk| vec![0; chunk.len()]).collect();
    for (rank, (chunk, row)) in rows.sources().enumerate() {
        let Some(place) = at.get_mut(chunk as usize).and_then(|run| run.get_mut(row as usize))
        else {
            return Err(Error::internal("a sorted row pointing outside the chunks it came from"));
        };
        *place = rank as u32;
    }
    Ok(at)
}

/// Gives `bytes` back across the reservations the instances handed over.
///
/// From the last one backwards, which is arbitrary and is fine: they all charge the same budget and
/// what matters is only that the total comes down. A reservation that has nothing left is dropped,
/// which is the same as shrinking it to nothing and is one fewer to walk next time.
fn give(charged: &mut Vec<Reservation>, mut bytes: u64) {
    while bytes > 0 {
        let Some(last) = charged.last_mut() else { return };
        let held = last.bytes();
        if held > bytes {
            last.shrink(bytes);
            return;
        }
        bytes -= held;
        charged.pop();
    }
}

/// The sorted rows as chunks, with every column moved once.
///
/// This is where the sort stops being row shaped. The order is a permutation of the rows that
/// arrived, so what each column needs is for its values to be written out in that order, and an
/// [`Assembly`] is exactly that: the chunks that arrived are placed into it, each row landing at
/// the position the sort gave it, and the interleave is one typed copy per physical layout rather
/// than a `Value` a field. A string moves as sixteen bytes of view over an arena its bytes were
/// copied into once.
///
/// One column at a time, because the assembly for a column holds a second copy of that column and
/// holding one of them at a time is a column of headroom rather than a table of it. The finished
/// column is then cut into chunk sized windows, which for a page is a window and no copy.
///
/// The chunks come in by value and are taken apart into their columns before anything is laid, so
/// that the input's copy of a column can be dropped the moment it has been laid and the charge
/// against the memory limit can come down with it. Held as chunks there is nowhere to put the
/// column that is finished with, and the sort ends up holding the whole input and the whole output
/// at once, which is what made a big one die at the allocator rather than get slower.
///
/// # Errors
///
/// If a column has no layout an assembly can lay, or if the chunks pass the limit the database was
/// opened with.
fn gathered(
    types: &[LogicalType],
    chunks: Vec<Chunk>,
    at: &[Vec<u32>],
    rows: usize,
    held: &mut Reservation,
    charged: &mut Vec<Reservation>,
) -> Result<Vec<Chunk>> {
    if rows == 0 {
        return Ok(Vec::new());
    }
    // Transposed: the pieces of one column of every chunk, so that a column is one thing to drop.
    let mut pieces: Vec<Vec<Vector>> = vec![Vec::with_capacity(chunks.len()); types.len()];
    for chunk in chunks {
        for (position, column) in chunk.into_columns().into_iter().enumerate() {
            let Some(into) = pieces.get_mut(position) else {
                return Err(Error::internal("a sorted chunk wider than the schema it came from"));
            };
            into.push(column);
        }
    }
    let blocks = rows.div_ceil(VECTOR_SIZE);
    let mut columns: Vec<Vec<Vector>> = vec![Vec::with_capacity(types.len()); blocks];
    for (position, ty) in types.iter().enumerate() {
        let mut assembly = Assembly::new(ty.clone(), rows)?;
        let laid = pieces.get_mut(position).map(std::mem::take).unwrap_or_default();
        for (piece, places) in laid.iter().zip(at) {
            assembly.place(places, piece)?;
        }
        let given = laid.iter().map(Vector::footprint).sum::<usize>();
        drop(laid);
        give(charged, u64::try_from(given).unwrap_or(u64::MAX));
        let whole = assembly.finish()?.into_pages();
        for (block, into) in columns.iter_mut().enumerate() {
            let start = block * VECTOR_SIZE;
            into.push(whole.slice(start, (rows - start).min(VECTOR_SIZE))?);
        }
    }
    let mut built = Vec::with_capacity(blocks);
    for (block, columns) in columns.into_iter().enumerate() {
        let start = block * VECTOR_SIZE;
        let chunk = Chunk::with_rows(columns, (rows - start).min(VECTOR_SIZE))?;
        held.grow(u64::try_from(chunk.footprint()).unwrap_or(u64::MAX))?;
        built.push(chunk);
    }
    Ok(built)
}

/// More rows or more chunks than a sort addresses.
///
/// A row is found by a chunk and a row in it, both counted in a `u32`, and it lands at a position
/// an [`Assembly`] also counts in a `u32`. Four billion rows is a sort of something like a hundred
/// gigabytes, which is past where this operator should be asked anyway, and saying so is better
/// than an index that wrapped and an answer in the wrong order.
fn too_many() -> Error {
    Error::internal("a sort of more than 4294967295 rows")
}

/// A sort whose two halves disagree about how its keys are held.
///
/// Which way they are held is decided once, in [`Sort::new`], and every instance is built from that
/// decision, so the only way here is a bug in this file. It is an error rather than a fallback
/// because falling back means one instance's rows sorted one way and another's the other, which is
/// not an order.
fn mismatched() -> Error {
    Error::internal("a sort holding its keys two ways at once")
}

/// Where two rows sit relative to each other, with a tie on every key settled by where they arrived.
///
/// This is what makes the answer the same however many threads ran. [`compare`] on its own leaves
/// tied rows to the stability of the sort, which is the order they were handed over, and that is the
/// order the threads finished in.
pub(crate) fn settled(
    keys: &[SortKey],
    left: &Sortable,
    right: &Sortable,
    failure: &mut Option<Error>,
) -> Ordering {
    match compare(keys, &left.0, &right.0, failure) {
        Ordering::Equal => left.1.cmp(&right.1),
        ordering => ordering,
    }
}

/// Where two rows of keys sit relative to each other, under the whole key list in priority order.
///
/// The first key that separates them decides, and rows that agree on every key are equal, which is
/// where the stability of the sort does the rest.
///
/// A comparison that fails is reported as equal and remembered in `failure`, because `sort_by` wants
/// a total order and has nowhere to put an error. The order that comes out of a run that failed is
/// not an order anybody looks at, since the caller returns the error instead of the rows.
pub(crate) fn compare(
    keys: &[SortKey],
    left: &[Value],
    right: &[Value],
    failure: &mut Option<Error>,
) -> Ordering {
    for (at, key) in keys.iter().enumerate() {
        let ordering = match rank(&left[at], &right[at], *key) {
            Ok(ordering) => ordering,
            Err(error) => {
                failure.get_or_insert(error);
                Ordering::Equal
            }
        };
        if ordering != Ordering::Equal {
            return ordering;
        }
    }
    Ordering::Equal
}

/// Where two values sit relative to each other under one sort key.
///
/// The direction and the null placement are independent, which is the detail worth being careful
/// about. `ORDER BY x DESC NULLS LAST` is not `ORDER BY x NULLS FIRST` reversed: reversing the
/// whole comparison would move the nulls too, and DuckDB's answer keeps them where the query put
/// them. So the direction is applied to the comparison of two values and never to the rule that
/// places a null.
pub(crate) fn rank(left: &Value, right: &Value, key: SortKey) -> Result<Ordering> {
    match (left.is_null(), right.is_null()) {
        (true, true) => Ok(Ordering::Equal),
        (true, false) => Ok(if key.nulls_first { Ordering::Less } else { Ordering::Greater }),
        (false, true) => Ok(if key.nulls_first { Ordering::Greater } else { Ordering::Less }),
        (false, false) => {
            let ordering = rudb_kernels::order(left, right)?;
            Ok(if key.descending { ordering.reverse() } else { ordering })
        }
    }
}

fn poisoned<T>(_: T) -> Error {
    Error::internal("a thread panicked while holding the rows a sort is gathering")
}