rudb-exec 0.3.26

//! Turning a bound plan into the pipelines that run it.
//!
//! One match, one arm per logical operator, and nothing else. There is no physical plan and no cost
//! based choice between two ways of running the same node, which is the honest description of tier
//! 0: there is one implementation of each operator so there is nothing to choose between. The
//! physical planner that section 9.6 describes goes here, and the reason this is a separate module
//! from the operators is so that it can grow into one without any of them moving.
//!
//! The pipelines borrow the plan and the catalog for as long as they exist. A scan reads its rows
//! out of the catalog's table rather than copying them, and an expression reads its constants, its
//! function names and its types out of the plan's arena, so a plan that outlives the query it built
//! is the whole of the lifetime story here.
//!
//! # How a tree becomes a list
//!
//! The walk is the same one it always was, down from the root, and what changed is what it carries
//! back up. A node returns a [`Segment`], which is a source with the streaming operators stacked on
//! it so far, and a node that is a pipeline breaker closes the segment under it into a finished
//! [`Pipeline`] and starts a new segment over the buffer that breaker finalises into. So a plan with
//! two breakers in it comes back as three pipelines, and they are pushed onto the list in the order
//! they have to run, because a breaker's own pipeline is closed before the walk returns to whatever
//! is above it.
//!
//! A node with two inputs closes the side that has to finish first and then walks the side that uses
//! it, which is the same order the ids are handed out in and the same order the work happens in.
//!
//! # Where the measurement comes from
//!
//! Every operator this module makes is wrapped in [`Watched`] before it goes into a pipeline, and
//! the counters it reports into are registered with the [`Report`] the caller passed in. That is the
//! only place the wrapping happens, which is what makes it impossible for an operator to be left
//! out: an arm that forgets to wrap is an arm that does not compile, because the id it was handed
//! has to go somewhere.
//!
//! A breaker's counters go around two objects rather than one. The sink is the operator, and the
//! buffer the next pipeline sources from is where its rows come back out, so both are wrapped in the
//! same counters and a sort's row count is the rows it produced rather than zero.
//!
//! Neither the ids nor the pipeline numbers are worked out here. They come from [`Shape`], which is
//! one walk over the plan in `rudb-plan`, because `EXPLAIN` prints the same numbering and the same
//! decomposition without building anything, and two versions of that rule would be right on the day
//! they were written and disagree some time after. What this module does is ask which operator a
//! node is and wrap it.

use std::sync::Arc;

use rudb_catalog::{Catalog, QualifiedName};
use rudb_common::{Cancel, Memory, Result, Session, Value};
use rudb_functions::TableFunction;
use rudb_metrics::{Counters, Driver, Report};
use rudb_parquet::{Bound, Op};
use rudb_pipeline::{
    BufferId, DynSink, DynStream, Pipeline, PipelineId, Source, Watched, root, root_in_order,
};
use rudb_plan::{
    CompareOp, ConjunctionOp, Expr, ExprRef, Node, NodeRef, PipelineRef, Plan, ROOT, Shape, Slice,
    seams_of,
};
use rudb_seam::Settings;

use crate::enginenames::{
    database_size, dialects, extensions, grammar_extensions, optimizers, platform, user_agent,
    version,
};
use crate::entrynames::{columnnames, databasenames, schemanames, tablenames, viewnames};
use crate::fetch::{Fetch, TableFetch};
use crate::functionnames::functionnames;
use crate::gather::{Gather, Keep};
use crate::group::{Aggregate, Distinct};
use crate::join::{CrossProduct, Gathered, Join};
use crate::keywords::keywords;
use crate::query::Query;
use crate::register::registries;
use crate::schema::Schema;
use crate::setop::SetOp;
use crate::settingnames::settingnames;
use crate::sort::Sort;
use crate::source::{Dummy, FileScan, Scan, Series, Values};
use crate::strategies::strategies;
use crate::stream::{Filter, Limit, Project};
use crate::topn::TopN;
use crate::typenames::typenames;

/// Builds the pipelines for a plan's root, for a query nothing will stop.
///
/// Every seam is left at its default, which is what a caller with no session behind it wants and is
/// what the tests in this crate are written against.
///
/// # Errors
///
/// If the plan names a table or a column the catalog does not have, if an expression is malformed
/// in a way [`Plan::validate`] would have caught, or anything an operator's construction reports.
pub fn build<'a>(plan: &'a Plan, catalog: &'a Catalog) -> Result<Query<'a>> {
    build_with(plan, catalog, &Cancel::new(), &Memory::unlimited(), &Settings::new())
}

/// Builds the pipelines for a plan's root, stoppable through this token and held to this budget.
///
/// The token is checked once per chunk by the driver, so the query stops at the first chunk boundary
/// after the token says to. It is one check in one place rather than a decision per operator,
/// because a decision per operator is a decision somebody gets wrong when they add the twentieth
/// one. What the driver cannot see is work an operator does inside one call, and the join is the one
/// that can: its nested loop runs to the end inside a single push, and a hundred thousand left rows
/// against thirty thousand right ones is a minute with nothing looking at the token, so that loop
/// holds the token as well and checks it once per left row.
///
/// The budget is not uniform, and that is the difference between the two. A streaming operator holds
/// one chunk and gives it away again, so charging every operator would count the same megabyte once
/// per level. Only the operators that buffer without bound take a reservation, and
/// [`rudb_common::Memory`] lists which ones those are.
///
/// The measurement still happens. It goes into a report nobody reads, because the alternative is two
/// builders that drift apart, and a pair of clock readings per chunk is not a cost worth avoiding by
/// having a second one.
///
/// The seam settings are the session's with the statement's hints on top, and they are read here
/// rather than looked up later, because a choice made while the query is built is a choice `EXPLAIN`
/// can print before the query runs. An operator that sits on a seam chooses once, in its
/// constructor, and holds what it chose.
///
/// # Errors
///
/// The same as [`build`].
pub fn build_with<'a>(
    plan: &'a Plan,
    catalog: &'a Catalog,
    cancel: &Cancel,
    memory: &Memory,
    seams: &Settings,
) -> Result<Query<'a>> {
    build_measured(plan, catalog, cancel, memory, seams, &Session::new(), &Report::new())
}

/// Builds the pipelines, reporting what every operator in them did into `report`.
///
/// The report is what the caller keeps. Once the query has been run, [`Report::fill`] turns it into
/// the operator and pipeline rows of a metrics document, and that document is the same one
/// `EXPLAIN ANALYZE` prints and `--metrics` writes.
///
/// The session is what `SET` has left the settings at, and the only thing that reads it is
/// `duckdb_settings()`. It is a separate argument from the seam settings because the seams are a
/// choice an operator makes while it is built and the settings are rows in a table. [`build`] and
/// [`build_with`] pass an empty one, which reports every value as null, since a caller with no
/// database behind it has no settings to report.
///
/// # Errors
///
/// The same as [`build`].
pub fn build_measured<'a>(
    plan: &'a Plan,
    catalog: &'a Catalog,
    cancel: &Cancel,
    memory: &Memory,
    seams: &Settings,
    session: &Session,
    report: &Report,
) -> Result<Query<'a>> {
    build_measured_with_sink(
        plan,
        catalog,
        BuildUnder { cancel, memory, seams, session, report },
        None,
    )
}

/// Builds the pipelines with their root connected to a caller supplied sink.
///
/// This is the write path counterpart of [`build_measured`]. It lets an `INSERT ... SELECT`
/// consume chunks as the producing pipeline runs instead of first collecting the whole result in
/// the root queue.
///
/// # Errors
///
/// The same as [`build_measured`].
pub fn build_measured_into<'a>(
    plan: &'a Plan,
    catalog: &'a Catalog,
    cancel: &Cancel,
    memory: &Memory,
    seams: &Settings,
    session: &Session,
    sink: Arc<dyn DynSink + 'a>,
) -> Result<Query<'a>> {
    let report = Report::new();
    build_measured_with_sink(
        plan,
        catalog,
        BuildUnder { cancel, memory, seams, session, report: &report },
        Some(sink),
    )
}

struct BuildUnder<'a> {
    cancel: &'a Cancel,
    memory: &'a Memory,
    seams: &'a Settings,
    session: &'a Session,
    report: &'a Report,
}

#[derive(Clone, Copy, Default)]
struct AggregateBound {
    max_groups: Option<usize>,
    top_counts: Option<usize>,
    having_count: Option<(usize, i64)>,
}

fn build_measured_with_sink<'a>(
    plan: &'a Plan,
    catalog: &'a Catalog,
    under: BuildUnder<'_>,
    sink: Option<Arc<dyn DynSink + 'a>>,
) -> Result<Query<'a>> {
    let BuildUnder { cancel, memory, seams, session, report } = under;
    let shape = Shape::of(plan);
    for pipeline in shape.all() {
        report.pipeline(pipeline);
        for waits_for in shape.waits_for(pipeline) {
            report.depends(pipeline, *waits_for);
        }
    }
    let mut building = Building {
        plan,
        catalog,
        cancel,
        memory,
        seams,
        session,
        report,
        shape,
        done: Vec::new(),
        drivers: Vec::new(),
        pruning: Vec::new(),
        top_counts: Vec::new(),
    };
    let segment = building.node(plan.root())?;
    let schema = segment.schema.clone();
    // A query whose rows come out of a sort or a top n is already in the order somebody asked for,
    // and holding chunks back to restore the source order would only add latency to an order nobody
    // is going to look at. Everything else gets the root that puts them back, because the moment
    // several threads read the same file a plain `SELECT` would otherwise come back in a different
    // order on every run. It costs nothing to decide here and it means the scheduler never has to.
    let reader = if let Some(sink) = sink {
        building.close(segment, ROOT, sink);
        None
    } else {
        let (sink, reader) = if ordered(plan, plan.root()) {
            root(BufferId(0), None)
        } else {
            root_in_order(BufferId(0), None)
        };
        building.close(segment, ROOT, Arc::new(sink));
        Some(reader)
    };
    let Building { done, drivers, .. } = building;
    Query::new(done, drivers, reader, schema)
}

/// Whether the rows reaching the root are already in an order the plan chose.
///
/// A sort and a top n both decide one. Everything between them and the root either keeps the order
/// it was given or is not a node that can sit there, and the walk stops at the first node that is
/// neither.
fn ordered(plan: &Plan, node: NodeRef) -> bool {
    match *plan.node(node) {
        Node::Sort { .. } | Node::TopN { .. } => true,
        Node::Project { input, .. }
        | Node::Filter { input, .. }
        | Node::Limit { input, .. }
        | Node::Fetch { input, .. } => ordered(plan, input),
        _ => false,
    }
}

/// The aggregate under a TopN whose only key is its first COUNT(*) result descending.
///
/// Keeping the local prefix from every radix partition is sufficient for the global prefix: a
/// group excluded behind `k` groups in its own partition cannot enter the first `k` overall. The
/// regular TopN remains in the plan and settles the small union, so this only reduces aggregate
/// output and does not replace ordering semantics.
fn count_top_aggregate(plan: &Plan, input: NodeRef, keys: Slice) -> Option<NodeRef> {
    let [key] = plan.sort_key_list(keys) else { return None };
    if !key.descending {
        return None;
    }
    let Expr::Column(ordered) = *plan.expr(key.expr) else { return None };
    let (aggregate, output) = match *plan.node(input) {
        Node::Project { input, index, exprs, .. } => {
            if ordered.table != index {
                return None;
            }
            let projected = *plan.expr_list(exprs).get(ordered.column as usize)?;
            let Expr::Column(output) = *plan.expr(projected) else { return None };
            (input, output)
        }
        Node::Aggregate { index, .. } if ordered.table == index => (input, ordered),
        _ => return None,
    };
    let Node::Aggregate { index, groups, aggregates, .. } = *plan.node(aggregate) else {
        return None;
    };
    if output.table != index || output.column as usize != plan.expr_list(groups).len() {
        return None;
    }
    let first = *plan.expr_list(aggregates).first()?;
    let Expr::Aggregate { name, args, distinct, filter } = *plan.expr(first) else {
        return None;
    };
    let count_star = plan.string(name) == "count_star"
        && plan.expr_list(args).is_empty()
        && !distinct
        && filter.is_none();
    let distinct_count = plan.string(name) == "count"
        && plan.expr_list(args).len() == 1
        && distinct
        && filter.is_none();
    (count_star || distinct_count).then_some(aggregate)
}

/// A direct aggregate under `input` and the COUNT(*) call constrained by a simple lower bound.
///
/// The Filter remains in the pipeline and checks the predicate again. Recognizing only this narrow
/// shape therefore changes how many aggregate rows are materialized and not which rows are valid.
fn count_having_aggregate(
    plan: &Plan,
    input: NodeRef,
    predicate: ExprRef,
) -> Option<(NodeRef, usize, i64)> {
    let Node::Aggregate { index, groups, aggregates, .. } = *plan.node(input) else { return None };
    let Expr::Compare { op, left, right } = *plan.expr(predicate) else { return None };
    let Expr::Column(column) = *plan.expr(left) else { return None };
    let Expr::Constant(value) = *plan.expr(right) else { return None };
    let Value::BigInt(value) = *plan.value(value) else { return None };
    if column.table != index {
        return None;
    }
    let call = (column.column as usize).checked_sub(plan.expr_list(groups).len())?;
    let aggregate = *plan.expr_list(aggregates).get(call)?;
    let Expr::Aggregate { name, args, distinct, filter } = *plan.expr(aggregate) else {
        return None;
    };
    if plan.string(name) != "count_star"
        || !plan.expr_list(args).is_empty()
        || distinct
        || filter.is_some()
    {
        return None;
    }
    let minimum = match op {
        CompareOp::Greater => value.checked_add(1)?,
        CompareOp::GreaterOrEqual => value,
        _ => return None,
    };
    Some((input, call, minimum))
}

/// What a filter over a scan can tell that scan before it reads anything.
///
/// Both kinds of scan keep the smallest and the largest value of each of their columns: a Parquet
/// file keeps them per row group in its footer, and a table in memory keeps them per chunk in a zone
/// map. A conjunct comparing one of those columns against a constant can rule a whole unit out
/// without touching a row of it. This pulls out the conjuncts of that shape and drops everything
/// else, which is the conservative direction: a test that is not here costs time, a test that is
/// here wrongly costs rows.
///
/// Only an `AND` is walked into. Under an `OR` a conjunct being false says nothing about the row,
/// and a `NOT` is already gone by the time the plan is bound. Of the table functions only
/// `read_parquet` is worth doing this for, because a CSV has no footer to read. Only a comparison
/// against this scan's own columns counts, since a binding into some other operator's output is not
/// in this table at all.
fn bounds(plan: &Plan, input: NodeRef, predicate: ExprRef) -> Vec<(usize, Op, Bound)> {
    let index = match *plan.node(input) {
        Node::TableFunction { index, function, .. } => {
            if TableFunction::lookup(plan.string(function)) != Some(TableFunction::ReadParquet) {
                return Vec::new();
            }
            index
        }
        Node::Get { index, .. } => index,
        _ => return Vec::new(),
    };
    let mut tests = Vec::new();
    conjuncts(plan, predicate, index, &mut tests);
    tests
}

/// Every conjunct of `predicate` that reads as a bounds test, appended to `out`.
fn conjuncts(plan: &Plan, predicate: ExprRef, index: u32, out: &mut Vec<(usize, Op, Bound)>) {
    match *plan.expr(predicate) {
        Expr::Conjunction { op: ConjunctionOp::And, children } => {
            for child in plan.expr_list(children) {
                conjuncts(plan, *child, index, out);
            }
        }
        Expr::Compare { op, left, right } => {
            if let Some(test) = comparison(plan, op, left, right, index) {
                out.push(test);
            }
        }
        _ => {}
    }
}

/// One comparison read as a test on a column of the scan numbered `index`, if it is one.
///
/// Written either way round, because `5 < a` and `a > 5` say the same thing and the optimizer does
/// not normalise which side the constant sits on. The comparisons that survive a null are the four
/// orderings and equality: `<>` rules out a row group only when the group holds one distinct value,
/// which the footer does not say, and the two distinctness operators are about nulls rather than
/// about bounds.
fn comparison(
    plan: &Plan,
    op: CompareOp,
    left: ExprRef,
    right: ExprRef,
    index: u32,
) -> Option<(usize, Op, Bound)> {
    let op = match op {
        CompareOp::Equal => Op::Equal,
        CompareOp::Less => Op::Less,
        CompareOp::LessOrEqual => Op::LessOrEqual,
        CompareOp::Greater => Op::Greater,
        CompareOp::GreaterOrEqual => Op::GreaterOrEqual,
        CompareOp::NotEqual | CompareOp::DistinctFrom | CompareOp::NotDistinctFrom => return None,
    };
    let (op, binding, value) = match (plan.expr(left), plan.expr(right)) {
        (Expr::Column(binding), Expr::Constant(value)) => (op, *binding, *value),
        (Expr::Constant(value), Expr::Column(binding)) => (op.flipped(), *binding, *value),
        _ => return None,
    };
    if binding.table != index {
        return None;
    }
    Some((binding.column as usize, op, Bound::of_value(plan.value(value))?))
}

/// A pipeline being built from the bottom up.
///
/// It is not a [`Pipeline`] yet because it has no sink. What ends it is whichever node above it
/// turns out to be a pipeline breaker, or the root of the plan, and neither is known until the walk
/// gets back there.
struct Segment<'a> {
    source: Arc<dyn Source + 'a>,
    /// In the order they run, nearest the source first.
    streams: Vec<Arc<dyn DynStream + 'a>>,
    /// What the segment produces as it stands, which changes as streams are added.
    schema: Schema,
    /// The pipelines this one cannot start before.
    after: Vec<PipelineRef>,
}

impl<'a> Segment<'a> {
    /// A segment that is just its source.
    fn new(source: Arc<dyn Source + 'a>, schema: Schema) -> Self {
        Self { source, streams: Vec::new(), schema, after: Vec::new() }
    }

    /// A segment reading what a pipeline breaker finalised into.
    fn reading(source: Arc<dyn Source + 'a>, schema: Schema, after: PipelineRef) -> Self {
        Self { source, streams: Vec::new(), schema, after: vec![after] }
    }

    /// Puts a streaming operator on the end, which becomes what the segment produces.
    fn then(mut self, stream: Arc<dyn DynStream + 'a>, schema: Schema) -> Self {
        self.streams.push(stream);
        self.schema = schema;
        self
    }
}

/// What the walk down the plan carries with it.
struct Building<'a, 'b> {
    plan: &'a Plan,
    catalog: &'a Catalog,
    cancel: &'b Cancel,
    memory: &'b Memory,
    seams: &'b Settings,
    /// What `SET` has left the settings at, which only `duckdb_settings()` reads.
    session: &'b Session,
    report: &'b Report,
    shape: Shape,
    /// The pipelines closed so far, in the order they have to run.
    done: Vec<Pipeline<'a>>,
    /// One per entry of `done`, in the same order.
    drivers: Vec<Arc<Driver>>,
    /// The bounds tests the filter arm worked out for the scan it is about to walk into.
    ///
    /// A scan is built before the filter above it, because the filter needs the schema the scan
    /// produces, so by the time there is a filter to read there is already a scan that cannot be
    /// told anything. This carries the tests the other way, down the one step from a filter to its
    /// own input, and the scan arm takes them. It is empty every other time it is read, and empty
    /// means hand out every row group, which is what every scan did before pruning existed.
    pruning: Vec<(usize, Op, Bound)>,
    /// Aggregates whose parent TopN orders by COUNT descending, and its count plus offset.
    top_counts: Vec<(NodeRef, usize)>,
}

impl<'a> Building<'a, '_> {
    /// The id of the operator holding the side of this node that has to finish first.
    ///
    /// # Panics
    ///
    /// If the node has one input, which is a node whose arm below should not have called this.
    fn gathered(&self, node: NodeRef) -> u32 {
        self.shape.gathered(node).expect("a node with two inputs has a second operator")
    }

    /// Ends a segment with a sink and puts the finished pipeline on the list.
    fn close(&mut self, segment: Segment<'a>, id: PipelineRef, sink: Arc<dyn DynSink + 'a>) {
        let mut pipeline = Pipeline::new(PipelineId(id), segment.source, sink);
        for stream in segment.streams {
            pipeline = pipeline.then(stream);
        }
        for after in segment.after {
            pipeline = pipeline.after(PipelineId(after));
        }
        self.done.push(pipeline);
        self.drivers.push(self.report.driving(id));
    }

    /// The counters for one operator, registered with the report.
    ///
    /// The row records what this operator picked at each seam it sits on, which is `seams_of` on
    /// its plan node crossed with what is registered and what the statement pinned. That is the
    /// same three things `EXPLAIN` puts its reference marker from, and it is read here rather than
    /// asserted here for a reason worth writing down: this used to mark every operator as a
    /// reference implementation unconditionally, so every ClickBench run said 41 of 41 operators
    /// ran the slow path no matter what had actually run, and the fold that reported it was read as
    /// if it meant something.
    ///
    /// An operator that sits on no registered seam records nothing and stays marked as a reference,
    /// because there is one implementation of it and that one is the obvious correct one. The
    /// marker comes off by itself on the day a seam under it has something else registered and
    /// chosen, with nothing to remember to change here.
    fn watch(
        &self,
        node: NodeRef,
        id: u32,
        pipeline: u32,
        kind: &str,
        detail: Option<&str>,
    ) -> Arc<Counters> {
        let mut counters = Counters::new(id, pipeline, kind);
        if let Some(detail) = detail {
            counters = counters.detailed(detail);
        }
        for seam in seams_of(self.plan.node(node)) {
            if let Some(running) = registries().running(*seam, self.seams) {
                counters = counters.chose(seam.name(), &running.name, running.is_reference);
            }
        }
        self.report.watch(counters)
    }

    fn aggregate(
        &mut self,
        reference: NodeRef,
        input: NodeRef,
        index: u32,
        groups: Slice,
        aggregates: Slice,
        bound: AggregateBound,
    ) -> Result<Segment<'a>> {
        let below = self.node(input)?;
        let (aggregate, out) =
            Aggregate::new(self.plan, &below.schema, index, groups, aggregates, self.memory)?;
        let aggregate = aggregate.in_session(self.session);
        let aggregate = match bound.max_groups {
            Some(limit) => aggregate.limit_groups(limit),
            None => aggregate,
        };
        let aggregate = match bound.top_counts {
            Some(bound) => aggregate.top_counts(bound),
            None => aggregate,
        };
        let aggregate = match bound.having_count {
            Some((call, minimum)) => aggregate.having_count(call, minimum),
            None => aggregate,
        };
        let schema = aggregate.schema().clone();
        let id = self.shape.operator(reference);
        let pipeline = self.shape.pipeline(reference);
        let counters = self.watch(reference, id, pipeline, "Aggregate", None);
        let reading = Arc::clone(&counters);
        self.close(below, pipeline, Arc::new(Watched::new(aggregate, counters)));
        Ok(Segment::reading(Arc::new(Watched::new(out, reading)), schema, pipeline))
    }

    /// The segment a node produces, closing any pipeline that ends underneath it.
    fn node(&mut self, reference: NodeRef) -> Result<Segment<'a>> {
        let plan = self.plan;
        let memory = self.memory;
        let id = self.shape.operator(reference);
        let pipeline = self.shape.pipeline(reference);
        let segment = match *plan.node(reference) {
            Node::Get { catalog: database, schema, table, index, columns, .. } => {
                let name = QualifiedName::new(
                    plan.string(database),
                    plan.string(schema),
                    plan.string(table),
                );
                let tests = std::mem::take(&mut self.pruning);
                let scan = Scan::new(plan, self.catalog.table(&name)?, index, columns, tests)?;
                let schema = scan.schema().clone();
                let counters =
                    self.watch(reference, id, pipeline, "Scan", Some(plan.string(table)));
                Segment::new(Arc::new(Watched::new(scan, counters)), schema)
            }
            Node::Dummy => {
                let dummy = Dummy::new();
                let schema = dummy.schema().clone();
                let counters = self.watch(reference, id, pipeline, "Dummy", None);
                Segment::new(Arc::new(Watched::new(dummy, counters)), schema)
            }
            Node::Values { index, columns, rows } => {
                let values = Values::new(plan, index, columns, rows, self.session)?;
                let schema = values.schema().clone();
                let counters = self.watch(reference, id, pipeline, "Values", None);
                Segment::new(Arc::new(Watched::new(values, counters)), schema)
            }
            Node::TableFunction { index, function, args, options, settings, columns } => {
                let name = plan.string(function);
                match TableFunction::lookup(name) {
                    Some(function @ (TableFunction::ReadParquet | TableFunction::ReadCsv)) => {
                        let counters = self.watch(reference, id, pipeline, "FileScan", Some(name));
                        let tests = std::mem::take(&mut self.pruning);
                        let scan = FileScan::new(
                            plan, index, function, args, options, settings, columns, tests,
                        )?
                        .watched(counters.clone());
                        let schema = scan.schema().clone();
                        Segment::new(Arc::new(Watched::new(scan, counters)), schema)
                    }
                    Some(
                        function @ (TableFunction::RudbStrategies
                        | TableFunction::DuckdbKeywords
                        | TableFunction::DuckdbTypes
                        | TableFunction::DuckdbFunctions
                        | TableFunction::DuckdbSettings
                        | TableFunction::DuckdbDatabases
                        | TableFunction::DuckdbSchemas
                        | TableFunction::DuckdbTables
                        | TableFunction::DuckdbViews
                        | TableFunction::DuckdbColumns
                        | TableFunction::DuckdbExtensions
                        | TableFunction::DuckdbOptimizers
                        | TableFunction::DuckdbDialects
                        | TableFunction::DuckdbGrammarExtensions
                        | TableFunction::PragmaVersion
                        | TableFunction::PragmaPlatform
                        | TableFunction::PragmaUserAgent
                        | TableFunction::PragmaDatabaseSize),
                    ) => {
                        let table = match function {
                            TableFunction::DuckdbKeywords => keywords(plan, index, columns)?,
                            TableFunction::DuckdbTypes => typenames(plan, index, columns)?,
                            TableFunction::DuckdbFunctions => functionnames(plan, index, columns)?,
                            TableFunction::DuckdbSettings => {
                                settingnames(self.session, plan, index, columns)?
                            }
                            TableFunction::DuckdbDatabases => {
                                databasenames(self.catalog, plan, index, columns)?
                            }
                            TableFunction::DuckdbSchemas => {
                                schemanames(self.catalog, plan, index, columns)?
                            }
                            TableFunction::DuckdbTables => {
                                tablenames(self.catalog, plan, index, columns)?
                            }
                            TableFunction::DuckdbViews => {
                                viewnames(self.catalog, plan, index, columns)?
                            }
                            TableFunction::DuckdbColumns => {
                                columnnames(self.catalog, plan, index, columns)?
                            }
                            TableFunction::DuckdbExtensions => extensions(plan, index, columns)?,
                            TableFunction::DuckdbOptimizers => optimizers(plan, index, columns)?,
                            TableFunction::DuckdbDialects => dialects(plan, index, columns)?,
                            TableFunction::DuckdbGrammarExtensions => {
                                grammar_extensions(plan, index, columns)?
                            }
                            TableFunction::PragmaVersion => version(plan, index, columns)?,
                            TableFunction::PragmaPlatform => platform(plan, index, columns)?,
                            TableFunction::PragmaUserAgent => user_agent(plan, index, columns)?,
                            TableFunction::PragmaDatabaseSize => {
                                database_size(self.catalog, self.memory, plan, index, columns)?
                            }
                            _ => strategies(plan, index, columns)?,
                        };
                        let schema = table.schema().clone();
                        // `EXPLAIN` names the table rather than the operator, because every one of
                        // these is the same operator and a plan that said `Metadata` four times
                        // would not say which four tables it read.
                        let counters =
                            self.watch(reference, id, pipeline, "Metadata", Some(function.name()));
                        Segment::new(Arc::new(Watched::new(table, counters)), schema)
                    }
                    _ => {
                        let series = Series::new(plan, index, name, args)?;
                        let schema = series.schema().clone();
                        let counters = self.watch(reference, id, pipeline, "Series", Some(name));
                        Segment::new(Arc::new(Watched::new(series, counters)), schema)
                    }
                }
            }
            Node::Fetch { input, index, args, columns, row } => {
                let below = self.node(input)?;
                let counters = self.watch(reference, id, pipeline, "Fetch", None);
                let fetch = Fetch::new(plan, &below.schema, index, args, columns, row)?
                    .in_session(self.session)
                    .watched(counters.clone());
                let schema = fetch.schema().clone();
                below.then(Arc::new(Watched::new(fetch, counters)), schema)
            }
            Node::TableFetch { input, index, catalog, schema, table, columns, row } => {
                let below = self.node(input)?;
                let name = QualifiedName::new(
                    plan.string(catalog),
                    plan.string(schema),
                    plan.string(table),
                );
                let counters = self.watch(reference, id, pipeline, "TableFetch", None);
                let fetch = TableFetch::new(
                    plan,
                    &below.schema,
                    index,
                    self.catalog.table(&name)?,
                    columns,
                    row,
                )?
                .in_session(self.session);
                let schema = fetch.schema().clone();
                below.then(Arc::new(Watched::new(fetch, counters)), schema)
            }
            Node::Filter { input, predicate } => {
                self.pruning = bounds(plan, input, predicate);
                let below = match count_having_aggregate(plan, input, predicate) {
                    Some((aggregate, call, minimum)) => {
                        let Node::Aggregate { input: under, index, groups, aggregates } =
                            *plan.node(aggregate)
                        else {
                            unreachable!("count_having_aggregate returned another node")
                        };
                        self.aggregate(
                            aggregate,
                            under,
                            index,
                            groups,
                            aggregates,
                            AggregateBound {
                                max_groups: None,
                                top_counts: None,
                                having_count: Some((call, minimum)),
                            },
                        )?
                    }
                    None => self.node(input)?,
                };
                // Cleared whether or not the scan arm took them, because a filter over anything
                // else leaves them sitting there for whatever scan the walk reaches next.
                self.pruning = Vec::new();
                let schema = below.schema.clone();
                let filter = Filter::new(plan, reference, predicate, &schema, self.seams)?
                    .in_session(self.session);
                let counters = self.watch(reference, id, pipeline, "Filter", None);
                below.then(Arc::new(Watched::new(filter, counters)), schema)
            }
            Node::Project { input, index, exprs, names } => {
                let below = self.node(input)?;
                let project = Project::new(plan, &below.schema, index, exprs, names)?
                    .in_session(self.session);
                let schema = project.schema().clone();
                let counters = self.watch(reference, id, pipeline, "Project", None);
                below.then(Arc::new(Watched::new(project, counters)), schema)
            }
            Node::Aggregate { input, index, groups, aggregates } => {
                let top_counts = self
                    .top_counts
                    .iter()
                    .find_map(|&(aggregate, bound)| (aggregate == reference).then_some(bound));
                self.aggregate(
                    reference,
                    input,
                    index,
                    groups,
                    aggregates,
                    AggregateBound { max_groups: None, top_counts, having_count: None },
                )?
            }
            Node::Sort { input, keys } => {
                let below = self.node(input)?;
                let schema = below.schema.clone();
                let (sort, out) = Sort::new(plan, &schema, keys, memory)?;
                let sort = sort.in_session(self.session);
                let counters = self.watch(reference, id, pipeline, "Sort", None);
                let reading = Arc::clone(&counters);
                self.close(below, pipeline, Arc::new(Watched::new(sort, counters)));
                Segment::reading(Arc::new(Watched::new(out, reading)), schema, pipeline)
            }
            Node::Limit { input, count, offset } => {
                let max_groups = count
                    .and_then(|count| count.checked_add(offset))
                    .and_then(|count| usize::try_from(count).ok());
                let below = match (plan.node(input).clone(), max_groups) {
                    (
                        Node::Aggregate { input: under, index, groups, aggregates },
                        Some(max_groups),
                    ) => self.aggregate(
                        input,
                        under,
                        index,
                        groups,
                        aggregates,
                        AggregateBound {
                            max_groups: Some(max_groups),
                            top_counts: None,
                            having_count: None,
                        },
                    )?,
                    _ => self.node(input)?,
                };
                let schema = below.schema.clone();
                let limit = Limit::new(count, offset);
                let counters = self.watch(reference, id, pipeline, "Limit", None);
                below.then(Arc::new(Watched::new(limit, counters)), schema)
            }
            Node::TopN { input, keys, count, offset } => {
                if let Some(aggregate) = count_top_aggregate(plan, input, keys) {
                    let bound = count.saturating_add(offset);
                    if let Ok(bound) = usize::try_from(bound) {
                        self.top_counts.push((aggregate, bound));
                    }
                }
                let below = self.node(input)?;
                let schema = below.schema.clone();
                let (top, out) = TopN::new(plan, &schema, keys, count, offset, memory)?;
                let top = top.in_session(self.session);
                let counters = self.watch(reference, id, pipeline, "TopN", None);
                let reading = Arc::clone(&counters);
                self.close(below, pipeline, Arc::new(Watched::new(top, counters)));
                Segment::reading(Arc::new(Watched::new(out, reading)), schema, pipeline)
            }
            Node::Distinct { input, on } => {
                let below = self.node(input)?;
                let schema = below.schema.clone();
                let (distinct, out) = Distinct::new(plan, &schema, on, memory)?;
                let distinct = distinct.in_session(self.session);
                let counters = self.watch(reference, id, pipeline, "Distinct", None);
                let reading = Arc::clone(&counters);
                self.close(below, pipeline, Arc::new(Watched::new(distinct, counters)));
                Segment::reading(Arc::new(Watched::new(out, reading)), schema, pipeline)
            }
            Node::Join { left, right, kind, conditions } => {
                // The right side runs first, because no left row can be answered until every right
                // row it might match has been seen. That is the dependency edge, and it is the same
                // one the hash join builds on. The probing side is a pipeline of its own rather than
                // part of the one above it, because it ends in a sink, and it waits for the build
                // side.
                let gather_id = self.gathered(reference);
                let gathering = self.shape.pipeline(right);
                let right = self.node(right)?;
                let right_schema = right.schema.clone();
                let (gather, gathered) = Gather::new(memory);
                let kept = self.watch(reference, gather_id, gathering, "Gather", None);
                self.close(right, gathering, Arc::new(Watched::new(gather, kept)));
                let mut left = self.node(left)?;
                let side = Gathered { schema: &right_schema, rows: gathered };
                let (join, out) =
                    Join::new(plan, &left.schema, side, kind, conditions, self.cancel, memory);
                let join = join.in_session(self.session);
                let schema = join.schema().clone();
                let counters = self.watch(reference, id, pipeline, "Join", None);
                let reading = Arc::clone(&counters);
                left.after.push(gathering);
                self.close(left, pipeline, Arc::new(Watched::new(join, counters)));
                Segment::reading(Arc::new(Watched::new(out, reading)), schema, pipeline)
            }
            Node::CrossProduct { left, right } => {
                // The right side runs first and is kept as the chunks it arrived in, because it is
                // replayed once per left row. The left side streams, which is the whole point of
                // this operator: the product is produced a chunk at a time and never held, so the
                // product stays in the pipeline the left rows came from rather than starting one.
                let keep_id = self.gathered(reference);
                let aside = self.shape.pipeline(right);
                let right = self.node(right)?;
                let right_schema = right.schema.clone();
                let (keep, kept) = Keep::new(memory);
                let held = self.watch(reference, keep_id, aside, "Keep", None);
                self.close(right, aside, Arc::new(Watched::new(keep, held)));
                let mut left = self.node(left)?;
                let cross = CrossProduct::new(&left.schema, &right_schema, kept);
                let schema = cross.schema().clone();
                let counters = self.watch(reference, id, pipeline, "CrossProduct", None);
                left.after.push(aside);
                left.then(Arc::new(Watched::new(cross, counters)), schema)
            }
            Node::SetOp { left, right, kind, all, index } => {
                // The right side runs first, because nothing can be said about a left row until the
                // whole right side has been counted. That is the dependency edge, spelled out.
                let gather_id = self.gathered(reference);
                let counting = self.shape.pipeline(right);
                let right = self.node(right)?;
                let (gather, gathered) = Gather::new(memory);
                let kept = self.watch(reference, gather_id, counting, "Gather", None);
                self.close(right, counting, Arc::new(Watched::new(gather, kept)));
                let mut left = self.node(left)?;
                let (setop, out) = SetOp::new(&left.schema, gathered, kind, all, index, memory);
                let schema = setop.schema().clone();
                let counters = self.watch(reference, id, pipeline, "SetOp", None);
                let reading = Arc::clone(&counters);
                left.after.push(counting);
                self.close(left, pipeline, Arc::new(Watched::new(setop, counters)));
                Segment::reading(Arc::new(Watched::new(out, reading)), schema, pipeline)
            }
        };
        Ok(segment)
    }
}

#[cfg(test)]
mod tests {
    use rudb_common::LogicalType;
    use rudb_plan::{CompareOp, Expr, Node, Plan};

    use super::{count_having_aggregate, count_top_aggregate};

    fn plan(direction: &str) -> Plan {
        Plan::parse(&format!(
            "TopN 10 offset 0 [#2.2::BIGINT {direction} NULLS LAST]\n  \
             Project #2 [#1.0::BIGINT AS WatchID, #1.1::INTEGER AS ClientIP, #1.2::BIGINT AS c]\n    \
             Aggregate #1 groups=[#0.0::BIGINT, #0.1::INTEGER] \
             aggregates=[count_star()::BIGINT]\n      \
             Values #0 [WatchID::BIGINT, ClientIP::INTEGER] rows=[]"
        ))
        .expect("a grouped count plan")
    }

    #[test]
    fn count_descending_topn_marks_its_aggregate() {
        let plan = plan("DESC");
        let Node::TopN { input, keys, .. } = *plan.node(plan.root()) else {
            panic!("the root is a TopN")
        };
        let aggregate = count_top_aggregate(&plan, input, keys).expect("the grouped count");
        assert!(matches!(plan.node(aggregate), Node::Aggregate { .. }));
    }

    #[test]
    fn count_ascending_cannot_discard_large_counts() {
        let plan = plan("ASC");
        let Node::TopN { input, keys, .. } = *plan.node(plan.root()) else {
            panic!("the root is a TopN")
        };
        assert!(count_top_aggregate(&plan, input, keys).is_none());
    }

    #[test]
    fn distinct_count_descending_topn_marks_its_aggregate() {
        let plan = Plan::parse(
            "TopN 10 offset 0 [#1.1::BIGINT DESC NULLS LAST]\n  \
             Aggregate #1 groups=[#0.0::VARCHAR] \
             aggregates=[count(DISTINCT #0.1::BIGINT)::BIGINT]\n    \
             Values #0 [SearchPhrase::VARCHAR, UserID::BIGINT] rows=[]",
        )
        .expect("a grouped distinct count plan");
        let Node::TopN { input, keys, .. } = *plan.node(plan.root()) else {
            panic!("the root is a TopN")
        };
        let aggregate = count_top_aggregate(&plan, input, keys).expect("the distinct count");
        assert!(matches!(plan.node(aggregate), Node::Aggregate { .. }));
    }

    #[test]
    fn a_count_having_lower_bound_marks_the_count_call() {
        let plan = Plan::parse(
            "Filter (#1.2::BIGINT > 100::BIGINT)::BOOLEAN\n  \
             Aggregate #1 groups=[#0.0::BIGINT] \
             aggregates=[avg(#0.1::BIGINT)::DOUBLE, count_star()::BIGINT]\n    \
             Values #0 [key::BIGINT, value::BIGINT] rows=[]",
        )
        .expect("an aggregate with a HAVING filter");
        let Node::Filter { input, predicate } = *plan.node(plan.root()) else {
            panic!("the root is a Filter")
        };
        let (aggregate, call, minimum) =
            count_having_aggregate(&plan, input, predicate).expect("the count bound");
        assert_eq!(aggregate, input);
        assert_eq!((call, minimum), (1, 101));
    }

    #[test]
    fn an_upper_count_having_bound_cannot_drop_aggregate_output() {
        let mut plan = Plan::parse(
            "Filter (#1.1::BIGINT > 100::BIGINT)::BOOLEAN\n  \
             Aggregate #1 groups=[#0.0::BIGINT] aggregates=[count_star()::BIGINT]\n    \
             Values #0 [key::BIGINT] rows=[]",
        )
        .expect("an aggregate with a HAVING filter");
        let Node::Filter { input, predicate } = *plan.node(plan.root()) else {
            panic!("the root is a Filter")
        };
        let Expr::Compare { left, right, .. } = *plan.expr(predicate) else {
            panic!("the predicate is a comparison")
        };
        let less =
            plan.add_expr(Expr::Compare { op: CompareOp::Less, left, right }, LogicalType::Boolean);
        assert!(count_having_aggregate(&plan, input, less).is_none());
    }
}