rudb-exec 0.4.30

//! The sink that does nothing but hold its input.
//!
//! Some operators need one whole side of their input in hand before they can start, and the side
//! they need is not the side they are a sink for. A set operation reads the right side, counts it,
//! and then decides one left row at a time. A join builds from one side and probes with the other.
//! In push terms that is two pipelines and a dependency between them, and the pipeline that is
//! depended on ends in a sink that keeps its rows and does nothing else.
//!
//! This is that sink, and there are two of it. [`Gather`] takes its input apart into rows, which
//! come out through a [`Rows`] handle, and [`Keep`] holds the chunks as they were given to it and
//! fills a [`Buffered`]. Which one an operator wants is decided by what it does with the side: one
//! that answers a row at a time wants rows, and one that replays the side as it stands wants the
//! chunks it was handed.

use std::sync::{Arc, Mutex};

use rudb_common::{Error, Memory, Reservation, Result, Value};
use rudb_pipeline::{Lease, Progress, Sink};
use rudb_vector::Chunk;

use crate::buffer::Buffered;
use crate::rows;
use crate::sideways::{self, Sideways};

/// Rows somebody gathered, readable once the pipeline that filled them has finished.
///
/// Cloning one gives another handle on the same rows.
#[derive(Debug, Default, Clone)]
pub(crate) struct Rows {
    held: Arc<Mutex<Vec<Vec<Value>>>>,
}

impl Rows {
    /// Take the rows out, leaving nothing behind.
    ///
    /// Taking rather than borrowing, because the one caller is a `finalize` that is about to turn
    /// them into something else and holding two copies of a side of a join is the thing the memory
    /// budget exists to stop.
    ///
    /// # Errors
    ///
    /// [`ErrorCode::Internal`](rudb_common::ErrorCode::Internal) if a thread panicked while holding
    /// them.
    pub(crate) fn take(&self) -> Result<Vec<Vec<Value>>> {
        Ok(std::mem::take(&mut *self.held.lock().map_err(poisoned)?))
    }
}

/// A sink that keeps every row it is given.
#[derive(Debug)]
pub(crate) struct Gather {
    memory: Memory,
    out: Rows,
    /// What the gathered rows are charged, kept until this sink is dropped, which is when the
    /// operator that depended on them is done with them.
    charged: Mutex<Vec<Reservation>>,
}

/// What one instance of a gather is holding.
#[derive(Debug)]
pub(crate) struct Gathering {
    rows: Vec<Vec<Value>>,
    charged: Reservation,
    /// What the vector holding the rows has been charged for, so that its doubling is charged once
    /// per growth rather than once per chunk. See [`rows::capacity`].
    counted: u64,
}

impl Gather {
    /// A gather and the handle its rows come out of.
    pub(crate) fn new(memory: &Memory) -> (Self, Rows) {
        let out = Rows::default();
        let gather =
            Self { memory: memory.clone(), out: out.clone(), charged: Mutex::new(Vec::new()) };
        (gather, out)
    }
}

impl Sink for Gather {
    type Local = Gathering;

    fn local(&self) -> Gathering {
        Gathering { rows: Vec::new(), charged: self.memory.reservation(), counted: 0 }
    }

    /// Not yet, because the rows come back out in the order they went in.
    ///
    /// A join reads this side row by row and a set operation walks it, so the order two instances
    /// combined in is the order of part of the answer. Nothing here is wrong with more than one
    /// instance, it is just that the answer would come out in an order that depends on which thread
    /// read which morsel. What unblocks it is either a hash join, which does not care what order it
    /// built in, or a gather that keeps the morsel each row came from the way the root does.
    fn parallel(&self) -> bool {
        false
    }

    fn sink(&self, chunk: &Chunk, local: &mut Gathering) -> Result<Progress> {
        take(chunk, local)?;
        Ok(Progress::More)
    }

    fn combine(&self, local: Gathering) -> Result<()> {
        self.out.held.lock().map_err(poisoned)?.extend(local.rows);
        self.charged.lock().map_err(poisoned)?.push(local.charged);
        Ok(())
    }

    fn finalize(&self, _threads: &Lease<'_>) -> Result<()> {
        Ok(())
    }
}

/// Reads one chunk into rows and charges what they took.
///
/// The charge happens once per chunk rather than once per row, so a query passes its limit by up to
/// a chunk of rows before it is told. That is the same granularity the cancellation check runs at
/// and for the same reason: a thousand rows is a bounded overshoot and a check per row is a branch
/// in the row loop.
///
/// Shared with the set operation, which gathers its left side the same way and then does something
/// with it, because two copies of this loop would be two places to get the charging wrong.
///
/// # Errors
///
/// [`rudb_common::ErrorCode::OutOfMemory`] when the rows pass the limit.
pub(crate) fn take(chunk: &Chunk, local: &mut Gathering) -> Result<()> {
    let mut taken = 0;
    // row at a time: this is the `Vec<Vec<Value>>` layout `rows` documents, and section 7.4's row
    // prefix with a payload beside it replaces it everywhere at once.
    for row in 0..chunk.len() {
        let values: Vec<Value> = chunk.row(row).collect();
        taken += rows::heap(&values);
        local.rows.push(values);
    }
    local.charged.grow(taken)?;
    let slots = u64::try_from(local.rows.capacity() * size_of::<Vec<Value>>()).unwrap_or(u64::MAX);
    rows::capacity(slots, &mut local.counted, &mut local.charged)
}

/// A sink that keeps every chunk it is given, as the chunk it was given.
///
/// The other sink in this file takes its input apart into rows, which is what an operator that
/// decides one row at a time needs. An operator that replays its side as it stands does not: a cross
/// product pairs one left row with a whole right chunk, so taking those chunks apart and building
/// them again would be a copy of the whole side for nothing. Same edge, same shape, different thing
/// kept.
///
/// A join takes this one too, for a reason that is nearly the opposite and comes out the same. It
/// reads its gathered side by position rather than in order, once to build its table and then once
/// per match, so what it wants is columns it can gather out of. Rows would be an allocation apiece
/// on the way in and a transpose back into columns on the way out. See [`Build`](crate::side::Build).
///
/// What it fills is a [`Buffered`], because that is already the source that reads finished chunks
/// back out and there is no reason for a second one.
///
/// A join also asks it what its key column turned out to hold. See [`Sideways`], and see
/// [`Keep::fill`] for what that costs.
#[derive(Debug)]
pub(crate) struct Keep<'a> {
    memory: Memory,
    chunks: Mutex<Vec<Chunk>>,
    /// What the kept chunks are charged, until they are finished and the charge goes with them to
    /// the [`Buffered`] that holds them from then on.
    charged: Mutex<Vec<Reservation>>,
    out: Buffered,
    /// The runtime filter of the join this side is about to be looked up by, where the operator on
    /// the other side of the dependency edge is a join that can use one. Empty for everything else,
    /// including every cross product, which is the other operator that keeps its side this way.
    sideways: Option<Arc<Sideways<'a>>>,
    /// Whether the order the chunks arrive in is part of what the operator above reads out of them.
    ///
    /// It decides whether the pipeline that fills this may run on more than one thread, because the
    /// order chunks are kept in is the order the threads finished in and one thread is the only way
    /// to make that the order they were produced in.
    ordered: bool,
}

/// What one instance of a keep is holding.
#[derive(Debug)]
pub(crate) struct Kept {
    chunks: Vec<Chunk>,
    charged: Reservation,
}

impl<'a> Keep<'a> {
    /// A keep and the source its chunks come out of, for an operator that replays them in order.
    pub(crate) fn new(memory: &Memory) -> (Self, Buffered) {
        Self::watching(memory, None, true)
    }

    /// The same, with a join's runtime filter to fill on the way past, and with the order the
    /// chunks arrive in said either way.
    pub(crate) fn watching(
        memory: &Memory,
        sideways: Option<Arc<Sideways<'a>>>,
        ordered: bool,
    ) -> (Self, Buffered) {
        let out = Buffered::new();
        let keep = Self {
            memory: memory.clone(),
            chunks: Mutex::new(Vec::new()),
            charged: Mutex::new(Vec::new()),
            out: out.clone(),
            sideways,
            ordered,
        };
        (keep, out)
    }

    /// Fills the runtime filter from the chunks this side turned out to hold.
    ///
    /// One evaluation of the key expression and one hash per chunk, over the side that is going
    /// into the hash table. That is the same expression the table itself is built on and so it is
    /// evaluated twice, once here and once there, which is a pass over the smaller side of the join
    /// to save reading part of the larger one. The alternative is building the table here instead of
    /// at the first probe, which is a better answer and a larger change than this one.
    ///
    /// Here rather than in [`Sink::sink`] because the filter is sized from the exact row count, and
    /// the exact row count is what a side has only once it is finished.
    fn fill(&self, chunks: &[Chunk]) -> Result<()> {
        let Some(sideways) = self.sideways.as_ref() else { return Ok(()) };
        let Some(keyed) = sideways.keyed() else { return Ok(()) };
        sideways.found(sideways::found_for(keyed, sideways.exact(), chunks, sideways.is_wanted())?);
        Ok(())
    }
}

impl Sink for Keep<'_> {
    type Local = Kept;

    fn local(&self) -> Kept {
        Kept { chunks: Vec::new(), charged: self.memory.reservation() }
    }

    /// Whichever the caller said, because the two operators that keep a side this way want opposite
    /// things from it.
    ///
    /// A cross product pairs each left row with these chunks one after another and a positional
    /// join pairs row `n` with row `n`, so for those the order the chunks arrive in is the answer
    /// and one thread is what keeps it the order they were produced in. A hash join reads them by
    /// position through a table it built itself, so the order decides which of two equal rows comes
    /// out first and nothing else, which is a thing SQL does not promise and which the driving side
    /// of the same join has been deciding on several threads all along.
    ///
    /// The difference this makes is the whole build side of a hash join, because a sink that refuses
    /// to run twice pins the pipeline it ends, and that pipeline is a scan and a filter over the
    /// larger part of a fact table. On TPC-H at SF1 it was 60 ms of a 65 ms query, on one core of
    /// ten, doing work the same scan does in 6 ms when nothing is asking it for a hash table.
    fn parallel(&self) -> bool {
        !self.ordered
    }

    fn sink(&self, chunk: &Chunk, local: &mut Kept) -> Result<Progress> {
        // An empty chunk is dropped rather than kept, because an operator replaying this side would
        // pair every one of its rows with it and produce nothing each time.
        if !chunk.is_empty() {
            local.charged.grow(u64::try_from(chunk.footprint()).unwrap_or(u64::MAX))?;
            local.chunks.push(chunk.clone());
        }
        Ok(Progress::More)
    }

    fn combine(&self, local: Kept) -> Result<()> {
        self.chunks.lock().map_err(poisoned)?.extend(local.chunks);
        self.charged.lock().map_err(poisoned)?.push(local.charged);
        Ok(())
    }

    fn finalize(&self, _threads: &Lease<'_>) -> Result<()> {
        let chunks = std::mem::take(&mut *self.chunks.lock().map_err(poisoned)?);
        // Before the chunks are handed on, because handing them on is what lets the pipeline that
        // depends on this one start, and the scan in that pipeline reads the filter as it starts.
        self.fill(&chunks)?;
        self.out.charge(std::mem::take(&mut *self.charged.lock().map_err(poisoned)?))?;
        self.out.fill(chunks)
    }
}

/// A fresh instance's state, for an operator that gathers one of its sides itself.
pub(crate) fn gathering(memory: &Memory) -> Gathering {
    Gathering { rows: Vec::new(), charged: memory.reservation(), counted: 0 }
}

/// What an instance gathered, and what it was charged for it.
pub(crate) fn into_parts(local: Gathering) -> (Vec<Vec<Value>>, Reservation) {
    (local.rows, local.charged)
}

fn poisoned<T>(_: T) -> Error {
    Error::internal("a thread panicked while holding the rows an operator gathered")
}

#[cfg(test)]
mod tests {
    use rudb_common::{LogicalType, Memory, Value};
    use rudb_vector::{Data, Vector};

    use super::{Chunk, Gather, Keep, Sink};

    fn chunk(values: &[i32]) -> Chunk {
        let column = Vector::flat(LogicalType::Integer, Data::Int32(values.to_vec().into()))
            .expect("integers are an i32 layout");
        Chunk::new(vec![column]).expect("one column is one length")
    }

    fn first(rows: &[Vec<Value>]) -> Vec<Value> {
        rows.iter().map(|row| row[0].clone()).collect()
    }

    #[test]
    fn what_goes_in_comes_out_in_the_order_it_was_combined() {
        let memory = Memory::unlimited();
        let (gather, rows) = Gather::new(&memory);

        let mut local = gather.local();
        gather.sink(&chunk(&[1, 2]), &mut local).expect("two rows");
        gather.sink(&chunk(&[3]), &mut local).expect("one more");
        gather.combine(local).expect("the one instance");
        gather.finalize(&rudb_pipeline::Lease::alone()).expect("nothing to do");

        assert_eq!(
            first(&rows.take().expect("readable")),
            [Value::Integer(1), Value::Integer(2), Value::Integer(3)]
        );
    }

    /// Every instance's rows end up in the one list. What order the instances combine in is what
    /// decides the order of the list, which is why the operator that reads it either does not care
    /// or has to sort it, and a set operation does not care.
    #[test]
    fn two_instances_both_end_up_in_the_one_list() {
        let memory = Memory::unlimited();
        let (gather, rows) = Gather::new(&memory);

        let mut left = gather.local();
        let mut right = gather.local();
        gather.sink(&chunk(&[1]), &mut left).expect("one row");
        gather.sink(&chunk(&[2]), &mut right).expect("one row");
        gather.combine(left).expect("the first instance");
        gather.combine(right).expect("the second instance");

        assert_eq!(first(&rows.take().expect("readable")), [Value::Integer(1), Value::Integer(2)]);
    }

    /// Taking leaves nothing behind, because the caller is about to turn the rows into something
    /// else and two copies of a side of a join is what the memory budget exists to stop.
    #[test]
    fn taking_the_rows_empties_them() {
        let memory = Memory::unlimited();
        let (gather, rows) = Gather::new(&memory);
        gather.combine(gather.local()).expect("an instance that saw nothing");

        assert!(rows.take().expect("readable").is_empty());
        assert!(rows.take().expect("readable").is_empty());
    }

    /// The other sink in here. What goes in comes back as the chunks it went in as, and an empty
    /// one is not one of them, because an operator replaying this side would pair every row it has
    /// with it and produce nothing each time.
    #[test]
    fn a_keep_holds_the_chunks_it_was_given_and_drops_the_empty_ones() {
        let memory = Memory::unlimited();
        let (keep, out) = Keep::new(&memory);

        let mut local = keep.local();
        keep.sink(&chunk(&[1, 2]), &mut local).expect("two rows");
        keep.sink(&Chunk::empty(&[LogicalType::Integer]), &mut local).expect("no rows");
        keep.sink(&chunk(&[3]), &mut local).expect("one row");
        keep.combine(local).expect("the one instance");
        keep.finalize(&rudb_pipeline::Lease::alone()).expect("the chunks");

        assert_eq!(out.len().expect("readable"), 2);
        assert_eq!(out.at(0).expect("readable").expect("the first").len(), 2);
        assert_eq!(out.at(1).expect("readable").expect("the second").len(), 1);
    }

    /// What a keep charged stays charged after it finishes, for as long as the chunks can be read,
    /// and goes with them when the one reader that turns them into something else takes them.
    #[test]
    fn the_charge_for_kept_chunks_goes_with_them() {
        let memory = Memory::unlimited();
        let (keep, out) = Keep::new(&memory);

        let mut local = keep.local();
        keep.sink(&chunk(&[1, 2, 3]), &mut local).expect("three rows");
        keep.combine(local).expect("the one instance");
        keep.finalize(&rudb_pipeline::Lease::alone()).expect("the chunks");
        assert!(memory.used() > 0, "the chunks are still charged once they are finished");

        let (chunks, charged) = out.take().expect("readable");
        assert_eq!(chunks.len(), 1);
        assert!(memory.used() > 0, "and until the caller lets them go");
        drop((chunks, charged));
        assert_eq!(memory.used(), 0);
        assert_eq!(out.len().expect("readable"), 0, "nothing is left for another handle");
    }

    /// The pipeline asks its sink whether it may run twice, and a keep answers with what the
    /// operator above it does with the order. The one that replays the chunks in order says no, and
    /// saying no pins a whole scan to one thread, so the two answers are worth pinning down here
    /// rather than only in the builder that chooses between them.
    #[test]
    fn a_keep_runs_on_one_thread_only_where_the_order_it_keeps_is_the_answer() {
        let memory = Memory::unlimited();

        let (replayed, _) = Keep::new(&memory);
        assert!(!replayed.parallel(), "a cross product pairs a row with these chunks in order");

        let (looked_up, _) = Keep::watching(&memory, None, false);
        assert!(looked_up.parallel(), "a hash join reads these through a table of its own");
    }
}