rudb-opt 0.3.62

//! How many rows a node produces, guessed.
//!
//! `spec/09-optimizer.md` section 9.3 puts cardinality estimation under join ordering, and join
//! ordering is not here yet. This arrives before it because two smaller things need it first: the
//! build side flag on [`Node::Join`], which is a choice between two numbers, and `EXPLAIN`, which
//! has to print something next to each operator. Both of those want the same function and neither
//! of them wants a search.
//!
//! Almost everything here is a guess and the type says so. [`rows`] returns `None` rather than a
//! default, because a caller that has to decide between two sides can only do that when it has two
//! numbers, and a made up number that looks like a measurement is how an optimizer talks itself
//! into the wrong plan. The one rule the whole module follows is that a node whose input is unknown
//! is unknown: uncertainty travels up rather than being rounded away at the first operator that has
//! a formula.
//!
//! [`rows_stat`] is the same walk with the class kept, and it is the one that says which numbers
//! are not guesses. A scan is the catalog's count and is exact, a Parquet read is the sum of the
//! footers the binder read and is exact too, a cross product of two counted sides is arithmetic on
//! counted numbers and is exact, a `LIMIT` over an unknown input is a real ceiling, and everything
//! above the first filter, group by or equijoin is an estimate from a constant. Both functions walk
//! the same tree and the numbers they give back are the same numbers, so a caller that only
//! compares two sides can go on using [`rows`] and ignore all of this.
//!
//! The constants are the textbook ones, which is to say they are DuckDB's, which is to say they
//! are Selinger's. They are wrong for any particular query and they are wrong in a direction that
//! does not depend on the query, which is the property that makes them usable: two sides of a join
//! estimated the same wrong way still compare correctly most of the time, and comparing is all the
//! first caller does. Nothing here should be read as a row count. It is a way of ordering two
//! plans.
//!
//! What is missing is every part of estimation that needs data rather than shape. There are no
//! column histograms, no distinct counts, no correlation between predicates, and no sample. A
//! filter on a primary key and a filter on a boolean get the same selectivity here. That is the
//! part `spec/09-optimizer.md` section 9.3 actually specifies and it needs the facts that
//! M3's storage layer collects, so it waits for them.

use std::collections::BTreeMap;
use std::sync::Arc;

use rudb_common::bounds::{Spread, Test, Zones};
use rudb_common::stat::{Class, Direction, Provenance, Stat, Use};
use rudb_plan::{
    ColumnBinding, CompareOp, ConjunctionOp, Expr, ExprRef, JoinKind, Node, NodeRef, Plan,
    SetOpKind, Slice,
};

use crate::{bounds, walk};

/// What one conjunct of a filter is assumed to keep.
///
/// A fifth, which is DuckDB's default for a predicate it cannot reason about and has been the
/// textbook guess since System R. It is too generous for an equality on a key and far too harsh
/// for `WHERE x > 0` on a column of counts, and it is applied per conjunct, so three conditions
/// anded together take a table to one row in a hundred and twenty five. That compounding is the
/// part most likely to be wrong, and it is kept because the alternative is to treat a query with
/// three conditions as though it were as selective as a query with one.
const KEPT_BY_A_CONDITION: f64 = 0.2;

/// What a group by is assumed to collapse its input to.
///
/// A tenth. Grouping is the operator where shape alone says the least: `GROUP BY user_id` over a
/// log table is close to one row in one, and `GROUP BY country` over the same table is a few
/// hundred rows out of any number. Without distinct counts there is nothing to tell them apart, so
/// this is a middle that is wrong for both rather than a guess that favours one.
const KEPT_BY_A_GROUP_BY: f64 = 0.1;

/// What one column of one table is known by, which is the whole key space of [`Facts`].
///
/// One key type and one `get` rather than a reader per kind of number, because the rule about what
/// a missing answer means has to be in one place. Two accessors are two chances to write `0` where
/// `Unknown` belongs.
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
pub enum Key<'a> {
    /// How many rows that table holds.
    Rows {
        /// The database it is attached in.
        catalog: &'a str,
        /// The schema it is in.
        schema: &'a str,
        /// The table.
        table: &'a str,
    },
    /// How many distinct values that column of that table holds.
    Distinct {
        /// The database it is attached in.
        catalog: &'a str,
        /// The schema it is in.
        schema: &'a str,
        /// The table.
        table: &'a str,
        /// The column, by name, because the position a column has in a scan is whatever column
        /// pruning left and the name is not moved by anything.
        column: &'a str,
    },
}

/// Everything counted that the optimizer was handed, by table and by column.
///
/// Called facts rather than statistics because that is what these are. Every number in here was
/// counted rather than sampled or guessed, the estimates are what this module makes out of them,
/// and the two want different names or a reader has to work out which one a variable holds.
///
/// A side table rather than a field on [`Node::Get`], and a plain count rather than a handle on the
/// catalog. Both of those are so that a plan stays a value: the optimizer's own tests build plans
/// out of text with no database anywhere near them, `Plan::parse` of a printed plan gives back the
/// plan it was printed from, and neither of those survives a node that carries a number only a
/// live catalog could have filled in.
///
/// # Reading it never waits
///
/// [`Facts::get`] is a lookup in a map this value owns, so there is no lock to take, nothing to
/// fault in and nothing to wait for. That is the guarantee `spec/stats/04-in-memory.md` asks for and
/// it is a property of the shape rather than of the code: a set of facts is built once, is never
/// written to again, and is handed to a statement behind an [`std::sync::Arc`]. A key nobody filled
/// in answers [`Stat::Unknown`], which every caller already has to handle, rather than going and
/// finding out and making the planner wait while it does.
///
/// # One generation per plan
///
/// [`Facts::generation`] says which version of the catalog these were read from, and a plan is
/// planned from exactly one of them. Two statements that see the same number are looking at the
/// same catalog, so the second reuses what the first built instead of walking every table and every
/// column again. That walk is what this used to cost per statement whether the query touched those
/// tables or not.
///
/// Files are not in here. What a Parquet call produces is counted by the binder, which is the only
/// thing in the chain holding the file open, and it rides on the plan against the table index of
/// the call. See [`Plan::measured`] and [`Plan::distinct_measured`].
///
/// The distinct counts are here beside the row counts and not somewhere else, because the two are
/// asked together: `join` divides one by the other and a divisor that arrived by a different road
/// than the dividend is a divisor nobody can keep in step.
///
/// Empty is the ordinary state for anything that is not a real query, and an empty one makes every
/// scan unknown rather than making every scan zero. A scan of a table nobody measured and a scan of
/// an empty table are not the same thing, and an optimizer that confuses them will happily build a
/// hash table from the side it thinks has no rows.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct Facts {
    tables: BTreeMap<(String, String, String), u64>,
    columns: BTreeMap<(String, String, String, String), u64>,
    generation: u64,
}

impl Facts {
    /// Nothing known about anything, and read from no catalog.
    #[must_use]
    pub fn new() -> Self {
        Self::default()
    }

    /// Nothing known about anything yet, to be filled in from that version of the catalog.
    #[must_use]
    pub fn at(generation: u64) -> Self {
        Self { generation, ..Self::default() }
    }

    /// Which version of the catalog these were read from, or zero for a set built by hand.
    ///
    /// Zero never matches a real catalog, whose own count starts at one, so a set built in a test
    /// can never be mistaken for one that is still current.
    #[must_use]
    pub const fn generation(&self) -> u64 {
        self.generation
    }

    /// What is known about one key, without waiting for anything.
    ///
    /// A number nobody recorded comes back [`Stat::Unknown`] rather than as a zero or as a guess,
    /// which is the distinction `spec/stats/02-the-catalogue.md` section 2.1.1 exists to keep. A
    /// scan of a table nobody measured and a scan of an empty table are not the same thing, and an
    /// optimizer that confuses them will build a hash table from the side it thinks has no rows.
    ///
    /// What comes back is [`Class::Exact`], because these are counted rather than estimated, and
    /// the provenance says which count it is so that a reader of `EXPLAIN` can tell the two apart.
    /// A distinct count says [`Provenance::Dictionary`] because that is where it comes from: the
    /// only tables that answer one are the ones holding a dictionary for the column.
    #[must_use]
    pub fn get(&self, key: &Key<'_>) -> Stat<u64> {
        let (found, provenance) = match *key {
            Key::Rows { catalog, schema, table } => {
                (self.rows_in(catalog, schema, table), Provenance::RowCount)
            }
            Key::Distinct { catalog, schema, table, column } => {
                (self.distinct_in(catalog, schema, table, column), Provenance::Dictionary)
            }
        };
        found.map_or(Stat::Unknown, |value| Stat::exact(value, provenance))
    }

    /// Record what one table held.
    pub fn record(&mut self, catalog: &str, schema: &str, table: &str, rows: u64) {
        self.tables.insert((catalog.to_owned(), schema.to_owned(), table.to_owned()), rows);
    }

    /// What that table held, where anybody said.
    ///
    /// Private, because [`Facts::get`] is the one reader and the rule about what a missing answer
    /// means belongs in one place. This is the map lookup under it.
    fn rows_in(&self, catalog: &str, schema: &str, table: &str) -> Option<u64> {
        self.tables.get(&(catalog.to_owned(), schema.to_owned(), table.to_owned())).copied()
    }

    /// Record how many distinct values one column of one table holds.
    ///
    /// By name and not by position, because the position a column has in a scan is whatever is left
    /// after column pruning moved it and the name is not moved by anything.
    pub fn record_distinct(
        &mut self,
        catalog: &str,
        schema: &str,
        table: &str,
        column: &str,
        distinct: u64,
    ) {
        let key = (catalog.to_owned(), schema.to_owned(), table.to_owned(), column.to_owned());
        self.columns.insert(key, distinct);
    }

    /// How many distinct values that column holds, where anybody counted.
    ///
    /// Private for the same reason [`Facts::rows_in`] is.
    fn distinct_in(&self, catalog: &str, schema: &str, table: &str, column: &str) -> Option<u64> {
        let key = (catalog.to_owned(), schema.to_owned(), table.to_owned(), column.to_owned());
        self.columns.get(&key).copied()
    }

    /// Whether anything at all was recorded.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.tables.is_empty() && self.columns.is_empty()
    }
}

/// The class of a number that came out of one of the constants above.
const GUESSED: Class = Class::Estimated;

/// Where a number that came out of one of the constants above says it came from.
///
/// [`Provenance::Default`] and not [`Provenance::Propagation`], because the guess is the constant
/// and the propagation only carried it. The point of printing the provenance in `EXPLAIN` is to
/// find the place where nobody had a number, and this is that place.
const FROM_A_CONSTANT: Provenance = Provenance::Default;

/// The class of a number that is a proven ceiling with nothing under it.
///
/// A `LIMIT 10` over an unknown input produces somewhere between no rows and ten, so the value is
/// certain from above and the relative error can be the whole of it, which is a bound of one. That
/// is the weakest certificate there is and it is still worth telling apart from a guess: a guess
/// can be exceeded and this cannot.
const CEILING: Class = Class::Certified { bound: 1.0, direction: Direction::AtMost };

/// What a row count in a plan is read for, in the vocabulary of `spec/stats/05-every-query.md`
/// section 5.1.1.
///
/// One constant rather than a `.decide()` written out at each call site, because that section asks
/// `EXPLAIN` to print which use happened, and a word printed in one file about a call made in
/// another is a word that goes stale the first time somebody changes the call.
///
/// It is [`Use::Decide`] for every cardinality the optimizer reads today and that is not a gap.
/// The three passes that read one, join order in `crate::order`, build side in `crate::sides` and
/// semi lowering in `crate::semi`, each choose between plans that produce the same rows, which is
/// exactly what Decide means. The worst a wrong number does there is a slow query. The day a pass
/// reads a cardinality to license a rewrite instead, that read declares [`Use::Enable`], the class
/// rule refuses everything but an exact count, and the statistics section says an enable happened.
pub const CARDINALITY: Use = Use::Decide;

/// How many rows this node is guessed to produce, where a guess can be made at all.
///
/// `None` means nothing downstream of here should pretend to know, which is the answer for a scan
/// of a table nobody measured, for a table function nobody measured either, and for anything above
/// either of those.
///
/// The same answer as [`rows_stat`] read for the Decide use of `spec/stats/05-every-query.md`
/// section 5.1.1, which is the only use a cardinality is ever put to: it chooses between two plans
/// that produce the same rows, so every class is allowed through and a caller that gets `None` has
/// to fall back to a documented default rather than to a number. A caller that wants to answer a
/// query from this, or to license a rewrite with it, has to call [`rows_stat`] and ask with
/// [`Stat::answer`] or [`Stat::enable`], and both of those will refuse almost everything this
/// module produces. That is the point.
#[must_use]
pub fn rows(plan: &Plan, node: NodeRef, stats: &Facts) -> Option<u64> {
    rows_stat(plan, node, stats).read(CARDINALITY).copied()
}

/// How many rows this node produces, and how much of that is knowledge.
///
/// `Unknown` means nothing downstream of here should pretend to know, which is the answer for a
/// scan of a table nobody measured, for every table function, and for anything above either of
/// those. A `Known` carries the class of `spec/stats/04-in-memory.md` section 4.1, and the classes
/// combine up the tree the way that document's section 4.7 asks: a number derived from an exact one
/// and a guess is a guess, and the degradation is never rounded away.
///
/// This walks the subtree once per call and does not cache. A caller that wants the whole plan
/// annotated will walk it top down and ask for each node, which is quadratic in the depth, and a
/// plan deep enough for that to matter is a plan with other problems. The cache goes in when join
/// ordering arrives and asks the same question about the same subtree a thousand times.
#[must_use]
pub fn rows_stat(plan: &Plan, node: NodeRef, stats: &Facts) -> Stat<u64> {
    let of = |child: NodeRef| rows_stat(plan, child, stats);
    match *plan.node(node) {
        // One row with no columns, which is what a `SELECT` with no `FROM` is bound against.
        Node::Dummy => Stat::exact(1, Provenance::RowCount),
        // The catalog counted these rather than estimating them, so the count is the count. That
        // is the one exact number a plan starts from today and it is why the histogram does not
        // read all unknown: a scan knows, and everything above it stops knowing.
        Node::Get { catalog, schema, table, .. } => stats.get(&Key::Rows {
            catalog: plan.string(catalog),
            schema: plan.string(schema),
            table: plan.string(table),
        }),
        // Counted rather than guessed. A literal row list is the one place in a plan where the
        // number of rows is written down.
        Node::Values { rows: list, .. } => u64::try_from(plan.row_list(list).len())
            .map_or(Stat::Unknown, |rows| Stat::exact(rows, Provenance::RowCount)),
        // Whatever the binder measured, which for a Parquet read is the sum of the footers and is
        // exact, and for every other table function is unknown. The answer comes from the binder
        // because the binder is the only thing in the chain with the file open, and it is read from
        // the plan rather than worked out here because a function nobody taught this about has to
        // stay unknown. Guessing on behalf of all of them is the failure mode this module exists to
        // avoid.
        Node::TableFunction { index, .. } => plan.measured(index),
        // Once per row of whatever is on its left, and nothing here knows how many rows that is or
        // how many the call gives back for each of them.
        Node::LateralFunction { .. } => Stat::Unknown,
        Node::Filter { input, predicate } => {
            let (kept, from) = kept(plan, input, predicate, stats);
            // The bounds are a ceiling over the guess and not a new thing to guess about. A store
            // that keeps a minimum and a maximum per part can say which parts this filter rules
            // out, and the rows in the parts that survive is a number no answer to this filter can
            // exceed. So the guess runs exactly as it always did and the ceiling is applied to it,
            // which is the one composition that cannot be worse than the guess alone: the ceiling
            // is never below the truth, so it replaces the guess only where the guess was above
            // the truth and it lands no further from it than the guess was.
            //
            // Taking the ceiling as the estimate instead, or taking a fifth of it, both measured
            // better on the fourteen predicates in the pull request that added this and neither is
            // safe in general. A fifth of a ceiling that is already tight is a number below the
            // truth, which is the direction that picks the wrong build side, and a ceiling read as
            // an estimate is above the guess on any predicate the bounds barely narrow.
            match surviving(plan, input, predicate) {
                // Provably none. Every part is ruled out by bounds that cannot be wrong in this
                // direction, so this is a fact and not an estimate, and it is the one answer here
                // that is allowed below the one row floor `guess` puts in.
                Some(0) => Stat::exact(0, Provenance::ZoneMap),
                Some(ceiling) => capped(guess_from(of(input), kept, from), ceiling),
                None => guess_from(of(input), kept, from),
            }
        }
        // A projection changes the width and not the height, and a sort changes neither.
        // A fetch reads a column of each row it is handed, so it is as tall as its input too.
        Node::Project { input, .. }
        | Node::Window { input, .. }
        | Node::Sort { input, .. }
        | Node::Fetch { input, .. }
        | Node::TableFetch { input, .. } => of(input),
        Node::Aggregate { input, groups, .. } => {
            // An aggregate with no group keys produces exactly one row, over an empty input as
            // much as over a billion, which is the one case here that is a fact rather than a
            // guess. `empty_result_pullup` stops at this node for the same reason.
            if plan.expr_list(groups).is_empty() {
                return Stat::exact(1, Provenance::RowCount);
            }
            guess(of(input), KEPT_BY_A_GROUP_BY)
        }
        // The same shape as a group by on those columns, because that is what it is.
        Node::Distinct { input, .. } => guess(of(input), KEPT_BY_A_GROUP_BY),
        Node::Limit { input, count, offset } => {
            let input = of(input);
            match count {
                // `OFFSET` with no `LIMIT` takes rows away and cannot add any, and taking a known
                // number of rows off a counted one leaves a counted one.
                None => input.map(|n| n.saturating_sub(offset)),
                // A limit is a ceiling even when the input is unknown, which is the one place in
                // this module where an unknown input still gives an answer. It is an upper bound
                // rather than an estimate, and for the callers here that is the useful direction:
                // a side that cannot produce more than ten rows is the small side whatever feeds
                // it. An input that was counted keeps its class, because the smaller of two known
                // numbers is known.
                Some(count) => match input {
                    Stat::Unknown => {
                        Stat::Known { value: count, class: CEILING, provenance: FROM_A_CONSTANT }
                    }
                    known => known.map(|n| n.saturating_sub(offset).min(count)),
                },
            }
        }
        Node::TopN { input, count, offset, .. } => match of(input) {
            Stat::Unknown => {
                Stat::Known { value: count, class: CEILING, provenance: FROM_A_CONSTANT }
            }
            known => known.map(|n| n.saturating_sub(offset).min(count)),
        },
        Node::Join { left, right, kind, conditions, .. } => join(
            of(left),
            of(right),
            kind,
            plan.expr_list(conditions).len(),
            keyspace(plan, conditions, stats),
        ),
        // The right cardinality is a function of each left row until decorrelation, so treating it
        // as one independently measured input would be a made-up estimate.
        Node::DependentJoin { .. } => Stat::Unknown,
        // Two counted sides multiply to a counted answer. Nothing is guessed here at all.
        Node::CrossProduct { left, right } => of(left).zip(of(right), u64::saturating_mul),
        // Every set operation is bounded above by both sides together, and `UNION ALL` reaches it.
        // The deduplicating ones and `EXCEPT` are somewhere below it and nothing here knows where,
        // so the bound is what they get, and the bound is what their class says they got.
        // Holding rows does not change how many there are, so a materialisation is as tall as the
        // query that reads it and the definition it holds is counted where it is read.
        Node::MaterializedCte { body, .. } => of(body),
        // What a read produces is what the definition produced, and the definition is above this
        // node rather than under it, which is the one place in a plan where that is true. A walk
        // that only sees the subtree cannot reach it, so this says so rather than guessing. Giving
        // a real answer takes the count being recorded when the definition is walked, which is
        // worth doing when something asks a question this would change the answer to.
        Node::CteScan { .. } => Stat::Unknown,
        Node::SetOp { left, right, kind, all, .. } => {
            let total = of(left).zip(of(right), u64::saturating_add);
            match (kind, all) {
                // `UNION ALL` emits both sides and reaches the bound, so two counted sides give a
                // counted answer.
                (SetOpKind::Union, true) => total,
                _ => ceiling(total),
            }
        }
    }
}

/// One of the constant guesses applied to a child's count.
///
/// `Unknown` in, `Unknown` out, which is the rule the whole module follows. Otherwise the child's
/// class combines with [`GUESSED`], so an exact scan under a filter is an estimate and stays one all
/// the way up.
fn guess(input: Stat<u64>, kept: f64) -> Stat<u64> {
    guess_from(input, kept, FROM_A_CONSTANT)
}

/// [`guess`] where the fraction did not come from a constant.
///
/// The class is [`GUESSED`] either way. A fraction worked out from a distinct count is still a
/// guess, because it assumes the rows are spread evenly over the values and nothing here has
/// checked that, and a column where they are not is exactly the column somebody wants to find.
/// What changes is the provenance, so that `EXPLAIN` distinguishes a node where a number was read
/// from one where nobody had a number at all, which is the whole reason the field is printed.
fn guess_from(input: Stat<u64>, kept: f64, from: Provenance) -> Stat<u64> {
    match input {
        Stat::Unknown => Stat::Unknown,
        Stat::Known { value, class, .. } => Stat::Known {
            value: scale(value, kept).max(1),
            class: class.combine(GUESSED),
            provenance: from,
        },
    }
}

/// A guess held under a number the answer provably cannot exceed.
///
/// The ceiling comes from the bounds a store keeps per part of itself, so it is a real fact about
/// this filter over this file, and the guess is a constant that knows nothing about either. Where
/// the guess is already under the ceiling it is left alone, because a ceiling says nothing about
/// how far under it the answer sits and overwriting an estimate with a bound would be trading a
/// number for a worse one. Where the ceiling bites, it is the answer and it says so: the value is
/// certain from above and unknown from below, which is what [`CEILING`] means, and it names the
/// zone map rather than the constant it replaced.
///
/// That is the whole reason this is a minimum rather than a new base to take a fraction of. A
/// ceiling is never below the truth, so a minimum of it and the guess is never further from the
/// truth than the guess was. This cannot make an estimate worse, ever, and nothing else that reads
/// these bounds has that property.
fn capped(guessed: Stat<u64>, ceiling: u64) -> Stat<u64> {
    match guessed {
        // Not `Unknown` any more. A filter over a table nobody counted still cannot produce more
        // rows than the parts the bounds leave hold, and that is the same kind of answer a `LIMIT`
        // over an unknown input gives.
        Stat::Unknown => {
            Stat::Known { value: ceiling, class: CEILING, provenance: Provenance::ZoneMap }
        }
        Stat::Known { value, .. } if ceiling < value => {
            Stat::Known { value: ceiling, class: CEILING, provenance: Provenance::ZoneMap }
        }
        known => known,
    }
}

/// The same number, said as a ceiling rather than as a count.
fn ceiling(stat: Stat<u64>) -> Stat<u64> {
    match stat {
        Stat::Unknown => Stat::Unknown,
        Stat::Known { value, class, provenance } => {
            Stat::Known { value, class: class.combine(CEILING), provenance }
        }
    }
}

/// What fraction of its input a filter is assumed to keep, and where that fraction came from.
///
/// The product over the conditions, with each condition answered by the best supplier that can
/// answer it and by [`KEPT_BY_A_CONDITION`] where none can.
///
/// An equality against a constant on a column somebody counted keeps one value out of however many
/// the column holds. That is the uniformity assumption and is the oldest textbook rule there is. One
/// over the count is not always smaller than the constant and is not meant to be: a column of three
/// values gives a third, which is above the fifth the constant guessed, and that is the direction
/// the constant was wrong in for `o_orderstatus`. The rule is to use the number where there is one,
/// not to make the answer smaller.
///
/// A range against a constant is interpolated between the bounds the file already states, per part,
/// by [`spread`]. `l_shipdate <= '1998-09-02'` keeps ninety eight percent of TPC-H's lineitem and
/// the constant called it twenty, and the two ends of the column were sitting in the footer the
/// whole time saying 1992 and 1998.
///
/// The conditions the count did not answer go to the bounds together rather than one at a time, so
/// that two of them on one column are intersected into the interval they name instead of multiplied
/// into a wider one. What is still multiplied is one condition against the next, and both numbers
/// are the column's own taken where the column is read, so a filter above a join reads the base
/// table's facts and applies them to an input something else has already cut down. That is the
/// standard reading and it is why this stays [`GUESSED`]: it assumes the conditions and whatever
/// happened underneath are independent, which is the assumption every estimator makes and the one
/// that fails first.
fn kept(plan: &Plan, input: NodeRef, predicate: ExprRef, stats: &Facts) -> (f64, Provenance) {
    let mut fraction = 1.0;
    let mut counted = 0;
    let mut pending = Vec::new();
    for conjunct in conjuncts(plan, predicate) {
        match values(plan, conjunct, stats) {
            Some(values) => {
                fraction /= widened(values);
                counted += 1;
            }
            None => pending.push(conjunct),
        }
    }
    let interpolated = spread(plan, input, &pending).map_or(0, |spread| {
        fraction *= spread.fraction;
        spread.read
    });
    // Every condition nobody could answer is still worth the constant, and there is one of them per
    // condition rather than one for the lot, because the fifth is a guess about one condition.
    let guessed = pending.len().saturating_sub(interpolated);
    for _ in 0..guessed {
        fraction *= KEPT_BY_A_CONDITION;
    }
    // A fraction two different suppliers contributed to came from the arithmetic over them rather
    // than from either, which is what `Propagation` is for. A reader chasing a bad estimate wants to
    // know which supplier to go and look at without reading the predicate back.
    let from = match (counted, interpolated, guessed) {
        (0, 0, _) => FROM_A_CONSTANT,
        (_, 0, 0) => Provenance::Sketch,
        (0, _, 0) => Provenance::ZoneMap,
        _ => Provenance::Propagation,
    };
    (fraction, from)
}

/// What fraction of a scan's rows these conditions are expected to keep, interpolated between bounds.
///
/// The same reading of the plan [`surviving`] makes, and the same refusals, for the same reasons:
/// straight on a scan whose store kept bounds, positions turned into names before the store is
/// asked, and a name the store does not have gives up rather than guessing.
///
/// What is different is which way the answer can be wrong. [`surviving`] proves parts hold nothing
/// and its answer is a ceiling. This one assumes the values inside a part are spread evenly between
/// its two ends, which is a guess that can land either side of the truth, so it multiplies into the
/// guess rather than capping it.
///
/// Every condition at once, because two of them on one column are one interval and not two, and a
/// store handed them separately has no way to know they belong together. [`Spread::read`] comes back
/// with how many of them the store could read, which is what the caller charges its constant for the
/// rest by.
fn spread(plan: &Plan, input: NodeRef, conjuncts: &[ExprRef]) -> Option<Spread> {
    let (zones, tests) = asked(plan, input, conjuncts)?;
    zones.spread(&tests)
}

/// The conditions a predicate is made of, capped.
///
/// A top level `AND` is the only thing that splits, which is the same split filter pushdown makes
/// and for the same reason: those are the parts that each have to hold. An `OR` is one condition
/// however many branches it has, and something inside a function call is not reached, because
/// `f(a AND b)` is one predicate about whatever `f` does.
///
/// A conjunct with the same value for every row is not here. The selectivity constant is a guess
/// about a predicate over data, and a condition that does not read the data keeps every row or none
/// of them rather than a fifth of them. Most of those are folded away before this ever sees them,
/// and the one that survives is the fold that was abandoned so that the error still comes from
/// running the query.
///
/// Capped at eight so that a query written by a generator does not compound its way to a factor of
/// a million. Past a handful of conditions the product has stopped meaning anything anyway, and the
/// cap is where it stops pretending to.
fn conjuncts(plan: &Plan, predicate: ExprRef) -> Vec<ExprRef> {
    let parts = match *plan.expr(predicate) {
        Expr::Conjunction { op: ConjunctionOp::And, children } => plan.expr_list(children).to_vec(),
        _ => vec![predicate],
    };
    parts.into_iter().filter(|&part| !walk::constant(plan, part)).take(8).collect()
}

/// How many values an equality against a constant picks one of, where anybody counted them.
///
/// Only `=`. The other comparisons are about order rather than about one value, `<>` picks all but
/// one and is the complement of this rather than this, and the two distinctness operators are about
/// nulls, where a distinct count says nothing because the Parquet footer does not count the null as
/// a value.
///
/// `None` rather than a fallback to the row count, which is the fallback [`distinct`] makes for the
/// join arithmetic. One over the rows is one row, so a filter that took it would call every equality
/// on an uncounted column a single row lookup, and `c_mktsegment = 'BUILDING'` would come out at one
/// row instead of thirty thousand. The join can fall back because the row count put through its
/// arithmetic gives the containment assumption back exactly. This has no such identity and has to
/// refuse.
fn values(plan: &Plan, conjunct: ExprRef, stats: &Facts) -> Option<u64> {
    let Expr::Compare { op: CompareOp::Equal, left, right } = *plan.expr(conjunct) else {
        return None;
    };
    let binding = match (plan.expr(left), plan.expr(right)) {
        (&Expr::Column(binding), _) if walk::constant(plan, right) => binding,
        (_, &Expr::Column(binding)) if walk::constant(plan, left) => binding,
        _ => return None,
    };
    // A column with no values in it is an empty column or a column of nothing but nulls, and
    // neither is something to divide by.
    stated(plan, binding, stats).filter(|&values| values > 0)
}

/// How many rows sit in the parts of `input` that `predicate` cannot rule out.
///
/// A ceiling and not an estimate, which is why the caller takes the smaller of this and its guess
/// rather than multiplying the two together.
fn surviving(plan: &Plan, input: NodeRef, predicate: ExprRef) -> Option<u64> {
    let (zones, tests) = asked(plan, input, &[predicate])?;
    zones.surviving(&tests)
}

/// The store under `input` and the tests `predicates` read as, named the way that store names them.
///
/// `None` unless the filter sits straight on a scan whose store kept bounds and at least one
/// conjunct reads as a test. Straight on, with no projection in between, because a projection
/// renames columns and the name is what the store is asked by, and following one through would be a
/// second place that has to agree with the first about what a column is called.
///
/// A conjunct that is not a test is not a refusal. Dropping it leaves parts in that a full reading
/// would have ruled out, which is the safe direction for both callers: it leaves [`surviving`] with
/// a looser ceiling and [`spread`] with a larger fraction, and the guess above either still applies.
/// What is a refusal is a test naming a column the store does not have, which means this plan and
/// this store disagree about what is being read, and a number worked out from that disagreement
/// would rule out parts holding rows the query wants.
///
/// The position is turned into a name and the name is given to the store, rather than the position
/// being handed over directly. Column pruning moves a scan's positions and moves nothing else, so a
/// position is about the plan and the store numbers its columns the way the file does.
fn asked<'a>(
    plan: &'a Plan,
    input: NodeRef,
    predicates: &[ExprRef],
) -> Option<(&'a Arc<dyn Zones>, Vec<Test>)> {
    let index = bounds::scanned(plan, input)?;
    let zones = plan.zones(index)?;
    let names = match *plan.node(input) {
        Node::Get { columns, .. } | Node::TableFunction { columns, .. } => plan.field_list(columns),
        _ => return None,
    };
    let mut tests = Vec::new();
    for predicate in predicates {
        for (position, op, value) in bounds::of(plan, input, *predicate) {
            let name = &names.get(position)?.name;
            tests.push(Test { column: zones.column(name)?, op, value });
        }
    }
    (!tests.is_empty()).then_some((zones, tests))
}

/// How many distinct values the column a binding names holds, where anybody counted.
///
/// A binding names the operator that produces the column and the position it has there, so this
/// finds the operator and asks what the column at that position is called. Only a scan has an
/// answer, and a projection is followed through to one.
///
/// A column nobody counted falls back to the table's rows. See [`Missing::Rows`] for why, and
/// [`stated`] for the caller that cannot take that answer.
fn distinct(plan: &Plan, binding: ColumnBinding, stats: &Facts) -> Option<u64> {
    follow(plan, binding, stats, Missing::Rows, 16)
}

/// [`distinct`] restricted to columns somebody actually counted.
fn stated(plan: &Plan, binding: ColumnBinding, stats: &Facts) -> Option<u64> {
    follow(plan, binding, stats, Missing::Nothing, 16)
}

/// What to answer at a scan for a column nobody counted.
#[derive(Clone, Copy, PartialEq, Eq)]
enum Missing {
    /// The table's rows. A column cannot hold more distinct values than the table has rows, so this
    /// is a bound rather than a guess, and put through the arithmetic in [`join`] it gives back the
    /// containment assumption exactly. That is why the join can fall back rather than refusing: a
    /// refusal on one side would throw away a stated count on the other, and that is most of TPC-H,
    /// where the low cardinality column is stated and the key it joins against is not.
    Rows,
    /// Nothing. For a caller with no such identity, where the row count would come out as a claim
    /// rather than as a fallback.
    Nothing,
}

/// `distinct` with the budget it spends going through projections.
///
/// A projection is followed when the expression at the position is nothing but a reference to a
/// column underneath, because a projection that carries a column through unchanged carries its
/// distinct values through unchanged as well. Every query written against a view goes through one
/// of those, so without this the counts would be read by almost nothing.
///
/// A column that came out of an expression, an aggregate or a set operation stops the search. What
/// a function or a group by did to the number of distinct values under it is not something this
/// knows, and the count of the input is a wrong answer rather than an approximate one. A cast is
/// not followed either: a widening one keeps the distinct values and a narrowing one can merge
/// them, and telling those apart is more than this needs.
///
/// The budget is against a plan that is malformed rather than against one that is deep. A
/// projection over a projection over a projection is ordinary and sixteen of them is not, and the
/// arena invariant means a well formed plan cannot cycle here anyway.
fn follow(
    plan: &Plan,
    binding: ColumnBinding,
    stats: &Facts,
    missing: Missing,
    depth: u32,
) -> Option<u64> {
    let depth = depth.checked_sub(1)?;
    let position = binding.column as usize;
    let rows = missing == Missing::Rows;
    for at in 0..u32::try_from(plan.node_count()).unwrap_or(u32::MAX) {
        match *plan.node(at) {
            Node::Get { catalog, schema, table, index, columns, .. } if index == binding.table => {
                let name = &plan.field_list(columns).get(position)?.name;
                let catalog = plan.string(catalog);
                let schema = plan.string(schema);
                let table = plan.string(table);
                let distinct = stats.get(&Key::Distinct { catalog, schema, table, column: name });
                if let Stat::Known { value, .. } = distinct {
                    return Some(value);
                }
                if !rows {
                    return None;
                }
                return match stats.get(&Key::Rows { catalog, schema, table }) {
                    Stat::Known { value, .. } => Some(value),
                    Stat::Unknown => None,
                };
            }
            Node::TableFunction { index, columns, .. } if index == binding.table => {
                let name = &plan.field_list(columns).get(position)?.name;
                if let Some(distinct) = plan.distinct_measured(index, name) {
                    return Some(distinct);
                }
                return if rows { plan.measured(index).read(CARDINALITY).copied() } else { None };
            }
            Node::Project { index, exprs, .. } if index == binding.table => {
                let &carried = plan.expr_list(exprs).get(position)?;
                let &Expr::Column(carried) = plan.expr(carried) else {
                    return None;
                };
                return follow(plan, carried, stats, missing, depth);
            }
            _ => {}
        }
    }
    None
}

/// How many pairs of values the conditions of a join can match on, where every one is understood.
///
/// The product over the conditions of the larger of the two sides' distinct counts, which is the
/// standard reading of an equijoin: the two columns draw from a shared set of values, the larger
/// count is how big that set is, and the values are assumed to be spread evenly over it. `None`
/// unless every condition is an equality between two base columns that both have a count, because a
/// condition nobody understood could be the one doing all the work and a divisor that left it out
/// would claim more rows than the join can produce.
fn keyspace(plan: &Plan, conditions: Slice, stats: &Facts) -> Option<u64> {
    keyspace_of(plan, plan.expr_list(conditions), stats)
}

/// `keyspace` over conditions the caller is holding rather than over a slice of the arena.
///
/// Join ordering wants this. It is deciding which pair to join next and the conditions that would
/// apply at that pair are the ones it has just worked out are testable there, which is a list it
/// built and not a list any node in the arena holds.
#[must_use]
pub fn keyspace_of(plan: &Plan, conditions: &[ExprRef], stats: &Facts) -> Option<u64> {
    if conditions.is_empty() {
        return None;
    }
    let mut product: u64 = 1;
    for &condition in conditions {
        let Expr::Compare { op: CompareOp::Equal | CompareOp::NotDistinctFrom, left, right } =
            *plan.expr(condition)
        else {
            return None;
        };
        let (&Expr::Column(left), &Expr::Column(right)) = (plan.expr(left), plan.expr(right))
        else {
            return None;
        };
        let pair = distinct(plan, left, stats)?.max(distinct(plan, right, stats)?);
        product = product.checked_mul(pair)?;
    }
    // A column with no distinct values at all is an empty column or a column of nothing but nulls,
    // and neither is something to divide by.
    (product > 0).then_some(product)
}

/// How many rows an equijoin of two sides of these sizes over that many key values produces.
///
/// The containment assumption: every row of the smaller side finds a match, so an equijoin produces
/// about as many rows as its larger side. It is the standard guess and it is right whenever one side
/// of the condition is a key, which is most joins anybody writes and none of the joins that hurt.
///
/// Where the key has a distinct count, the join can also be counted directly. Each side spreads its
/// rows over the same set of key values, so a value gets `left / keys` rows from one side and
/// `right / keys` from the other, and the pairs come to `left * right / keys`. On a key that is a
/// key the two agree: a thousand rows joined to a million on a column with a million values is a
/// million rows either way. On a column with twenty five values in it they do not agree at all, and
/// the second one is right. TPC-H q5 joins a hundred and fifty thousand customers to ten thousand
/// suppliers on a nation, and the containment assumption calls that a hundred and fifty thousand
/// rows when it is sixty million.
///
/// The larger of the two is taken rather than the second one outright, so this can only ever raise
/// an estimate above what shape alone said. A distinct count larger than the rows on the smaller
/// side is the case where it would lower one, and that happens when the count came from a table
/// that a filter underneath has already cut down, which makes the count stale rather than the join
/// small.
///
/// Public because the join ordering pass scores an order with it. The two have to be one function
/// rather than two that agree today, since an order chosen by one arithmetic and kept by another is
/// an order nobody can reason about.
#[must_use]
pub fn matched(left: u64, right: u64, keys: Option<u64>) -> u64 {
    let counted = keys.map_or(0, |keys| left.saturating_mul(right) / keys);
    left.max(right).max(counted)
}

/// The join kinds, each of which is a different question.
fn join(
    left: Stat<u64>,
    right: Stat<u64>,
    kind: JoinKind,
    conditions: usize,
    keys: Option<u64>,
) -> Stat<u64> {
    match kind {
        // Left rows, filtered by whether a match exists. Never more than the left side, and the
        // right side's size does not enter into it.
        JoinKind::Semi => guess(left, KEPT_BY_A_CONDITION),
        JoinKind::Anti => guess(left, 1.0 - KEPT_BY_A_CONDITION),
        // At most one right row each, by definition, which makes this a fact about the node rather
        // than a guess about the data.
        JoinKind::Single | JoinKind::Mark => left,
        // The nth with the nth, so the shorter side decides, and it decides exactly.
        JoinKind::Positional => left.zip(right, u64::min),
        _ => {
            let (
                Stat::Known { value: left, class: left_class, provenance: left_from },
                Stat::Known { value: right, class: right_class, provenance: right_from },
            ) = (left, right)
            else {
                return Stat::Unknown;
            };
            let both = left_class.combine(right_class);
            // Two sides that came from different places make a number that came from the
            // arithmetic, which is what a reader chasing this node needs to be told.
            let from = if left_from == right_from { left_from } else { Provenance::Propagation };
            // A join with no condition is a cross product wearing a different node.
            if conditions == 0 {
                return Stat::Known {
                    value: left.saturating_mul(right),
                    class: both,
                    provenance: from,
                };
            }
            let matched = matched(left, right, keys);
            let value = match kind {
                // An outer join emits every row of the preserved side whether it matched or not,
                // so the estimate cannot fall below that side.
                JoinKind::Left => matched.max(left),
                JoinKind::Right => matched.max(right),
                JoinKind::Full => matched.max(left).max(right),
                _ => matched,
            };
            // The containment assumption is the guess, so this is one however exact both sides
            // were. Two counted tables joined on a column nobody has a distinct count for is the
            // single most common way a plan goes wrong, and a class saying exact here would hide
            // exactly that.
            Stat::Known { value, class: both.combine(GUESSED), provenance: FROM_A_CONSTANT }
        }
    }
}

/// A count as a divisor, at least one so that nothing divides by zero or grows.
#[expect(
    clippy::cast_precision_loss,
    reason = "a count past two to the fifty third is not a count anybody measured"
)]
fn widened(count: u64) -> f64 {
    (count as f64).max(1.0)
}

/// A row count times a fraction, without letting the float arithmetic invent anything.
#[expect(
    clippy::cast_precision_loss,
    clippy::cast_possible_truncation,
    clippy::cast_sign_loss,
    reason = "an estimate going through f64 is the point, and the result is clamped"
)]
fn scale(rows: u64, by: f64) -> u64 {
    let scaled = rows as f64 * by;
    if scaled.is_finite() && scaled >= 0.0 { scaled.min(u64::MAX as f64) as u64 } else { 0 }
}

#[cfg(test)]
mod tests {
    use std::sync::{Arc, Mutex};

    use rudb_common::bounds::{Op, Spread, Test, Zones};
    use rudb_common::stat::{Class, Direction, Provenance, Stat};
    use rudb_plan::Plan;

    use super::{Facts, Key, rows, rows_stat};

    /// A one column scan of the named table, which is what most of these sit on.
    fn scan(table: &str, index: u32) -> String {
        format!("Get memory.main.{table} AS {table} #{index} [a::INTEGER]\n")
    }

    /// The tables named here, sized as given, and nothing else measured.
    fn facts(tables: &[(&str, u64)]) -> Facts {
        let mut stats = Facts::new();
        for (table, count) in tables {
            stats.record("memory", "main", table, *count);
        }
        stats
    }

    /// The estimate for the root of a plan written as text, against the given table sizes.
    fn estimate(text: &str, tables: &[(&str, u64)]) -> Option<u64> {
        let stats = facts(tables);
        let plan =
            Plan::parse(text).unwrap_or_else(|error| panic!("{text} did not parse: {error}"));
        rows(&plan, plan.root(), &stats)
    }

    /// The same estimate with the class still attached.
    fn stat(text: &str, tables: &[(&str, u64)]) -> Stat<u64> {
        let stats = facts(tables);
        let plan =
            Plan::parse(text).unwrap_or_else(|error| panic!("{text} did not parse: {error}"));
        rows_stat(&plan, plan.root(), &stats)
    }

    /// The estimate against table sizes and distinct counts, the counts named table then column.
    fn counted(text: &str, tables: &[(&str, u64)], columns: &[(&str, &str, u64)]) -> Option<u64> {
        let mut stats = facts(tables);
        for (table, column, distinct) in columns {
            stats.record_distinct("memory", "main", table, column, *distinct);
        }
        let plan =
            Plan::parse(text).unwrap_or_else(|error| panic!("{text} did not parse: {error}"));
        rows(&plan, plan.root(), &stats)
    }

    /// The same estimate with the class and the provenance still attached.
    fn counted_stat(
        text: &str,
        tables: &[(&str, u64)],
        columns: &[(&str, &str, u64)],
    ) -> Stat<u64> {
        let mut stats = facts(tables);
        for (table, column, distinct) in columns {
            stats.record_distinct("memory", "main", table, column, *distinct);
        }
        let plan =
            Plan::parse(text).unwrap_or_else(|error| panic!("{text} did not parse: {error}"));
        rows_stat(&plan, plan.root(), &stats)
    }

    /// A filter of the given predicate over a two column scan of `t`.
    fn filtered(predicate: &str) -> String {
        format!("Filter {predicate}\n  Get memory.main.t AS t #0 [a::INTEGER, b::INTEGER]\n")
    }

    /// A join of two one column scans on their one column, which is what the counted tests sit on.
    fn joined(left: &str, right: &str) -> String {
        format!(
            "Join INNER on=[(#0.0::INTEGER = #1.0::INTEGER)::BOOLEAN]\n  {}  {}",
            scan(left, 0),
            scan(right, 1)
        )
    }

    /// The guess this module has always made, spelled out.
    const GUESSED: Class = Class::Estimated;

    #[test]
    fn a_scan_is_what_the_catalog_said_and_nothing_when_nobody_said() {
        let text = scan("t", 0);
        assert_eq!(estimate(&text, &[("t", 5000)]), Some(5000));
        // Not zero. A table nobody measured and an empty table are different, and an optimizer
        // that confuses them builds its hash table from the wrong side.
        assert_eq!(estimate(&text, &[]), None);
        assert_eq!(estimate(&text, &[("t", 0)]), Some(0));
    }

    #[test]
    fn not_knowing_travels_up_rather_than_being_rounded_away() {
        // The filter has a formula and the thing under it does not, so the filter has no answer
        // either. This is the property the whole module rests on.
        let text = format!("Filter (#0.0::INTEGER > 1::INTEGER)::BOOLEAN\n  {}", scan("t", 0));
        assert_eq!(estimate(&text, &[]), None);
        assert!(estimate(&text, &[("t", 1000)]).is_some());
    }

    #[test]
    fn an_ungrouped_aggregate_is_one_row_whatever_is_under_it() {
        // The one answer here that is a fact rather than a guess, and it holds with no facts
        // at all, which is what makes it worth special casing.
        let text = format!("Aggregate #1 groups=[] aggregates=[]\n  {}", scan("t", 0));
        assert_eq!(estimate(&text, &[]), Some(1));
        assert_eq!(estimate(&text, &[("t", 9_000_000)]), Some(1));
    }

    #[test]
    fn a_group_by_collapses_its_input_and_a_scan_under_it_still_decides_whether_it_can() {
        let text = format!("Aggregate #1 groups=[#0.0::INTEGER] aggregates=[]\n  {}", scan("t", 0));
        assert_eq!(estimate(&text, &[("t", 1000)]), Some(100));
        assert_eq!(estimate(&text, &[]), None);
    }

    #[test]
    fn a_limit_is_a_ceiling_even_over_an_input_nobody_measured() {
        // The only place an unknown input still produces a number. It is an upper bound rather
        // than an estimate, and a side that cannot produce more than ten rows is the small side of
        // a join whatever feeds it.
        let text = format!("Limit 10 offset 0\n  {}", scan("t", 0));
        assert_eq!(estimate(&text, &[]), Some(10));
        assert_eq!(estimate(&text, &[("t", 3)]), Some(3));
        assert_eq!(estimate(&text, &[("t", 3_000_000)]), Some(10));
    }

    #[test]
    fn an_offset_with_no_limit_takes_rows_away_and_cannot_add_any() {
        let text = format!("Limit ALL offset 5\n  {}", scan("t", 0));
        assert_eq!(estimate(&text, &[("t", 12)]), Some(7));
        assert_eq!(estimate(&text, &[("t", 2)]), Some(0));
        // No ceiling to fall back on here, so an unmeasured input stays unmeasured.
        assert_eq!(estimate(&text, &[]), None);
    }

    #[test]
    fn a_filter_never_estimates_a_relation_away_entirely() {
        // Six conditions at a fifth each is a factor of fifteen thousand, and a plan that believes
        // a subtree produces no rows is a plan that stops reading it. Pruning a subtree is
        // `empty_result_pullup`'s job and it does it from a proof rather than from a guess.
        let and = "(#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND (#0.0::INTEGER > 2::INTEGER)::BOOLEAN \
                   AND (#0.0::INTEGER > 3::INTEGER)::BOOLEAN AND \
                   (#0.0::INTEGER > 4::INTEGER)::BOOLEAN AND (#0.0::INTEGER > 5::INTEGER)::BOOLEAN \
                   AND (#0.0::INTEGER > 6::INTEGER)::BOOLEAN";
        let text = format!("Filter ({and})::BOOLEAN\n  {}", scan("t", 0));
        assert_eq!(estimate(&text, &[("t", 10)]), Some(1));
        // And each part counts, so six of them cut harder than one of them.
        let one = format!("Filter (#0.0::INTEGER > 1::INTEGER)::BOOLEAN\n  {}", scan("t", 0));
        assert_eq!(estimate(&one, &[("t", 1_000_000)]), Some(200_000));
    }

    #[test]
    fn a_condition_that_reads_no_column_is_not_counted_as_a_condition() {
        // A fifth is a guess about a predicate over data. One that does not read the data keeps
        // every row or none of them, and taking a fifth for it is taking a fifth for nothing.
        let both = format!(
            "Filter ((#0.0::INTEGER > 1::INTEGER)::BOOLEAN AND (1::INTEGER > 2::INTEGER)::BOOLEAN)::BOOLEAN\n  {}",
            scan("t", 0)
        );
        assert_eq!(estimate(&both, &[("t", 1_000_000)]), Some(200_000));
        let alone = format!("Filter (1::INTEGER > 2::INTEGER)::BOOLEAN\n  {}", scan("t", 0));
        assert_eq!(estimate(&alone, &[("t", 1_000_000)]), Some(1_000_000));
    }

    #[test]
    fn an_inner_join_comes_out_the_size_of_its_larger_side() {
        // The containment assumption. Ten thousand against fifty thousand is fifty thousand.
        let text = format!(
            "Join INNER on=[(#0.0::INTEGER = #1.0::INTEGER)::BOOLEAN]\n  {}  {}",
            scan("small", 0),
            scan("big", 1)
        );
        assert_eq!(estimate(&text, &[("small", 10_000), ("big", 50_000)]), Some(50_000));
        // And one side missing is the whole join missing, since the assumption is about both.
        assert_eq!(estimate(&text, &[("small", 10_000)]), None);
    }

    #[test]
    fn a_join_with_no_condition_is_the_product_and_says_so() {
        let text = format!("Join INNER on=[]\n  {}  {}", scan("small", 0), scan("big", 1));
        assert_eq!(estimate(&text, &[("small", 1000), ("big", 1000)]), Some(1_000_000));
    }

    #[test]
    fn an_outer_join_never_estimates_below_the_side_it_preserves() {
        // A left join that keeps every left row cannot produce fewer than that, however small the
        // right side is, and the containment assumption on its own would say otherwise.
        let text = format!(
            "Join LEFT on=[(#0.0::INTEGER = #1.0::INTEGER)::BOOLEAN]\n  {}  {}",
            scan("big", 0),
            scan("small", 1)
        );
        assert_eq!(estimate(&text, &[("big", 50_000), ("small", 10)]), Some(50_000));
    }

    #[test]
    fn a_semi_join_is_bounded_by_its_left_side_and_ignores_the_right() {
        // A semi join emits left rows, once each. However large the right side is, it cannot make
        // more of them, and the nested loop that runs it today should still know which side is
        // which.
        let text = format!(
            "Join SEMI on=[(#0.0::INTEGER = #1.0::INTEGER)::BOOLEAN]\n  {}  {}",
            scan("small", 0),
            scan("big", 1)
        );
        let estimated =
            estimate(&text, &[("small", 1000), ("big", 9_000_000)]).expect("both sides known");
        assert!(estimated <= 1000, "a semi join produced {estimated} out of 1000 left rows");
    }

    #[test]
    fn a_cross_product_of_two_enormous_sides_saturates_rather_than_wrapping() {
        // The number is meaningless and the point is that it is enormous rather than that it is
        // small, which is what a wrap would turn it into.
        let text = format!("CrossProduct\n  {}  {}", scan("a", 0), scan("b", 1));
        assert_eq!(estimate(&text, &[("a", u64::MAX), ("b", 2)]), Some(u64::MAX));
    }

    #[test]
    fn a_union_all_is_both_sides_and_so_is_the_bound_on_the_rest_of_them() {
        let text = format!("SetOp UNION ALL #2\n  {}  {}", scan("a", 0), scan("b", 1));
        assert_eq!(estimate(&text, &[("a", 30), ("b", 12)]), Some(42));
    }

    #[test]
    fn a_count_that_came_from_the_catalog_says_it_is_exact() {
        // The one number in here that was counted rather than guessed, and the only reason the
        // histogram does not read all unknown at this point in the series.
        assert_eq!(stat(&scan("t", 0), &[("t", 5000)]).class(), Some(Class::Exact));
        assert_eq!(stat(&scan("t", 0), &[]).class(), None);
    }

    #[test]
    fn every_number_here_says_where_it_came_from() {
        // The scan is the catalog's count and says so. The filter over it is the constant and says
        // that, which is the word somebody searches an `EXPLAIN` for when a plan went wrong,
        // because it means nobody had a number at that node at all.
        assert_eq!(stat(&scan("t", 0), &[("t", 5000)]).provenance(), Some(Provenance::RowCount));
        let text = format!("Filter (#0.0::INTEGER > 1::INTEGER)::BOOLEAN\n  {}", scan("t", 0));
        assert_eq!(stat(&text, &[("t", 1000)]).provenance(), Some(Provenance::Default));
    }

    #[test]
    fn a_cardinality_is_for_deciding_and_answers_nothing() {
        // A filtered count is a guess, so the build side chooser is welcome to it and nothing that
        // changes an answer is. This is the rule of section 5.1.1 read off one node, and it is the
        // one a later pass is most likely to break in good faith.
        let text = format!("Filter (#0.0::INTEGER > 1::INTEGER)::BOOLEAN\n  {}", scan("t", 0));
        let guessed = stat(&text, &[("t", 1000)]);
        assert_eq!(guessed.decide(), Some(&200));
        assert_eq!(guessed.answer(), None);
        assert_eq!(guessed.enable(), None);
        // A scan is the one node in here that could answer, and it still only does so because the
        // catalog counted rather than because the walk was clever.
        let counted = stat(&scan("t", 0), &[("t", 5000)]);
        assert_eq!(counted.answer(), Some(&5000));
        assert_eq!(counted.enable(), Some(&5000));
        // Nobody measured the table, so every use gets nothing rather than a zero.
        let nothing = stat(&scan("t", 0), &[]);
        assert_eq!(nothing.decide(), None);
        assert_eq!(nothing.answer(), None);
        assert_eq!(nothing.enable(), None);
    }

    #[test]
    fn one_guess_anywhere_under_a_node_makes_the_node_a_guess() {
        // Exact combined with a guess is the guess. A filter over a counted table is not a counted
        // number any more, and reading the class back as exact is what would make somebody fold a
        // constant on it later.
        let text = format!("Filter (#0.0::INTEGER > 1::INTEGER)::BOOLEAN\n  {}", scan("t", 0));
        assert_eq!(stat(&text, &[("t", 1000)]).class(), Some(GUESSED));
        // And the guess stays the same guess however many of them stack up, since they all come
        // from the same constant.
        let twice = format!(
            "Aggregate #1 groups=[#0.0::INTEGER] aggregates=[]\n  Filter (#0.0::INTEGER > \
             1::INTEGER)::BOOLEAN\n    {}",
            scan("t", 0)
        );
        assert_eq!(stat(&twice, &[("t", 1000)]).class(), Some(GUESSED));
    }

    #[test]
    fn an_ungrouped_aggregate_is_exact_because_one_row_is_a_fact() {
        let text = format!("Aggregate #1 groups=[] aggregates=[]\n  {}", scan("t", 0));
        assert_eq!(stat(&text, &[]).class(), Some(Class::Exact));
    }

    #[test]
    fn a_limit_over_an_unmeasured_input_is_certified_rather_than_estimated() {
        // Ten is not a guess about what the scan produces, it is the most this node can emit, so a
        // caller asking whether the number can be exceeded gets the right answer.
        let text = format!("Limit 10 offset 0\n  {}", scan("t", 0));
        assert_eq!(
            stat(&text, &[]).class(),
            Some(Class::Certified { bound: 1.0, direction: Direction::AtMost })
        );
        // Over a counted input the count wins and the answer is a fact again.
        assert_eq!(stat(&text, &[("t", 3)]).class(), Some(Class::Exact));
    }

    #[test]
    fn a_join_with_no_condition_is_a_product_and_the_product_is_exact() {
        // Every row against every row is arithmetic rather than an assumption. The equijoin next to
        // it is the assumption, and the two should not read the same.
        let product = format!("Join INNER on=[]\n  {}  {}", scan("a", 0), scan("b", 1));
        assert_eq!(stat(&product, &[("a", 1000), ("b", 1000)]).class(), Some(Class::Exact));
        let equi = format!(
            "Join INNER on=[(#0.0::INTEGER = #1.0::INTEGER)::BOOLEAN]\n  {}  {}",
            scan("a", 0),
            scan("b", 1)
        );
        assert_eq!(stat(&equi, &[("a", 1000), ("b", 1000)]).class(), Some(GUESSED));
    }

    #[test]
    fn a_table_function_nobody_measured_is_unknown_and_stays_unknown_over_it() {
        let text = "TableFunction range args=[] #0 [a::BIGINT]\n";
        assert_eq!(stat(text, &[]), Stat::Unknown);
    }

    #[test]
    fn a_table_function_the_binder_measured_is_as_tall_as_the_binder_said() {
        // What a Parquet read looks like from here. The number is the footer's and this module has
        // no opinion about it beyond passing it on with the class it arrived with.
        let text = "TableFunction read_parquet args=[] #0 [a::BIGINT]\n";
        let mut plan = Plan::parse(text).expect("a table function");
        plan.measure(0, Stat::exact(6_001_215, Provenance::RowCount));
        assert_eq!(
            rows_stat(&plan, plan.root(), &Facts::new()),
            Stat::exact(6_001_215, Provenance::RowCount)
        );
    }

    #[test]
    fn a_table_function_is_measured_against_its_index_and_not_against_its_name() {
        // Two reads of two different files in one statement. The index is what tells them apart,
        // and it is the one identifier a rewrite cannot move without rewriting every expression
        // above it, which is why the measurement is filed under it.
        let text = concat!(
            "Join INNER on=[]\n",
            "  TableFunction read_parquet args=[] #0 [a::BIGINT]\n",
            "  TableFunction read_parquet args=[] #1 [b::BIGINT]\n"
        );
        let mut plan = Plan::parse(text).expect("two table functions");
        plan.measure(0, Stat::exact(3, Provenance::RowCount));
        plan.measure(1, Stat::exact(5, Provenance::RowCount));
        assert_eq!(
            rows_stat(&plan, plan.root(), &Facts::new()),
            Stat::exact(15, Provenance::RowCount)
        );
    }

    #[test]
    fn a_table_function_nobody_measured_is_unknown_even_beside_one_that_was() {
        // A CSV read says nothing about its own length, so it stays unknown while the Parquet read
        // next to it is counted, and the join over the two is unknown because one side is.
        let text = concat!(
            "Join INNER on=[]\n",
            "  TableFunction read_parquet args=[] #0 [a::BIGINT]\n",
            "  TableFunction read_csv args=[] #1 [b::BIGINT]\n"
        );
        let mut plan = Plan::parse(text).expect("two table functions");
        plan.measure(0, Stat::exact(3, Provenance::RowCount));
        assert_eq!(rows_stat(&plan, plan.root(), &Facts::new()), Stat::Unknown);
    }

    #[test]
    fn a_guess_over_a_measured_file_is_a_guess_with_a_number_under_it() {
        // The point of measuring the file. Before it the filter had nothing to multiply and came
        // out unknown, and the two constants in this module were dead code on every query over a
        // Parquet file, which is every ClickBench query.
        let text = concat!(
            "Filter (#0.0::BIGINT > 5::BIGINT)::BOOLEAN\n",
            "  TableFunction read_parquet args=[] #0 [a::BIGINT]\n"
        );
        let mut plan = Plan::parse(text).expect("a filter over a table function");
        assert_eq!(rows_stat(&plan, plan.root(), &Facts::new()), Stat::Unknown);
        plan.measure(0, Stat::exact(1000, Provenance::RowCount));
        let over = rows_stat(&plan, plan.root(), &Facts::new());
        assert_eq!(over.value().copied(), Some(200));
        assert_eq!(over.class(), Some(Class::Estimated));
        assert_eq!(over.provenance(), Some(Provenance::Default));
    }

    #[test]
    fn a_lateral_function_is_unknown_however_well_the_file_beside_it_is_measured() {
        // It runs once per row of its input and nothing here knows how many rows it gives back for
        // each of them, so a measurement of some other table says nothing about this one.
        let text = concat!(
            "LateralFunction range args=[#0.0::INTEGER] #1 [a::BIGINT]\n",
            "  Get memory.main.t AS t #0 [a::INTEGER]\n"
        );
        let mut plan = Plan::parse(text).expect("a lateral function");
        plan.measure(1, Stat::exact(4096, Provenance::RowCount));
        assert_eq!(rows_stat(&plan, plan.root(), &Facts::new()), Stat::Unknown);
    }

    #[test]
    fn facts_that_nobody_filled_in_say_so() {
        let mut stats = Facts::new();
        assert!(stats.is_empty());
        stats.record("memory", "main", "t", 7);
        assert!(!stats.is_empty());
        assert_eq!(
            stats.get(&Key::Rows { catalog: "memory", schema: "main", table: "t" }),
            Stat::exact(7, Provenance::RowCount)
        );
        // The three names are one key. A table of the same name in another schema is another table,
        // and what comes back for it is unknown rather than a zero, because a table nobody measured
        // and an empty table are not the same thing.
        assert_eq!(
            stats.get(&Key::Rows { catalog: "memory", schema: "other", table: "t" }),
            Stat::Unknown
        );
    }

    #[test]
    fn a_distinct_count_says_where_it_came_from_and_a_missing_one_says_nothing() {
        let mut stats = Facts::new();
        stats.record_distinct("memory", "main", "t", "a", 25);
        let key = |column| Key::Distinct { catalog: "memory", schema: "main", table: "t", column };
        assert_eq!(stats.get(&key("a")), Stat::exact(25, Provenance::Dictionary));
        assert_eq!(stats.get(&key("b")), Stat::Unknown, "a column nobody counted");
    }

    #[test]
    fn a_set_built_by_hand_can_never_be_mistaken_for_a_catalog_that_is_current() {
        // A catalog counts from one, so a set nobody read one for answers a generation no catalog
        // ever has. That is what stops a set assembled in a test from being reused as though it
        // were the live one.
        assert_eq!(Facts::new().generation(), 0);
        assert_eq!(Facts::at(12).generation(), 12);
    }

    #[test]
    fn a_join_on_a_column_with_few_values_in_it_produces_more_rows_than_its_larger_side() {
        // TPC-H q5 written small: customers joined to suppliers on a nation key with twenty five
        // values in it. The containment assumption calls this a hundred and fifty thousand rows
        // and the real answer is sixty million, which is the estimate that made the join ordering
        // pass pick the worst order it could find.
        let text = joined("customer", "supplier");
        let tables = &[("customer", 150_000), ("supplier", 10_000)];
        assert_eq!(counted(&text, tables, &[]), Some(150_000));
        let counts = &[("customer", "a", 25), ("supplier", "a", 25)];
        assert_eq!(counted(&text, tables, counts), Some(60_000_000));
    }

    #[test]
    fn a_join_on_a_key_is_the_containment_assumption_and_a_count_does_not_change_it() {
        // Every order belongs to one customer, so the join is as tall as the orders however the
        // number is arrived at. The two readings agree here, which is why the containment
        // assumption survived as long as it did.
        let text = joined("customer", "orders");
        let tables = &[("customer", 150_000), ("orders", 1_500_000)];
        assert_eq!(counted(&text, tables, &[]), Some(1_500_000));
        let counts = &[("customer", "a", 150_000), ("orders", "a", 150_000)];
        assert_eq!(counted(&text, tables, counts), Some(1_500_000));
    }

    #[test]
    fn a_count_larger_than_the_rows_on_the_smaller_side_does_not_shrink_the_estimate() {
        // A count that big means the table it was taken on has been cut down by something
        // underneath since anybody counted, which makes the count stale rather than the join
        // small. The larger of the two readings is taken so a stale count cannot lower anything.
        let text = joined("small", "big");
        let tables = &[("small", 10), ("big", 1_000)];
        let counts = &[("small", "a", 1_000_000), ("big", "a", 1_000_000)];
        assert_eq!(counted(&text, tables, counts), Some(1_000));
    }

    #[test]
    fn two_conditions_match_on_the_pairs_of_values_and_not_on_either_column() {
        // Ten values on one column and ten on the other is a hundred pairs, and the rows spread
        // over the pairs rather than over either column alone.
        let text = concat!(
            "Join INNER on=[(#0.0::INTEGER = #1.0::INTEGER)::BOOLEAN, ",
            "(#0.1::INTEGER = #1.1::INTEGER)::BOOLEAN]\n",
            "  Get memory.main.l AS l #0 [a::INTEGER, b::INTEGER]\n",
            "  Get memory.main.r AS r #1 [a::INTEGER, b::INTEGER]\n"
        );
        let tables = &[("l", 1_000_000), ("r", 1_000_000)];
        let counts = &[("l", "a", 10), ("l", "b", 10), ("r", "a", 10), ("r", "b", 10)];
        assert_eq!(counted(text, tables, counts), Some(10_000_000_000));
    }

    #[test]
    fn a_condition_that_is_not_an_equality_between_two_columns_leaves_the_counts_unread() {
        // A range condition could be the one doing all the work, and a divisor that left it out
        // would claim more rows than the join can produce. Nothing is divided by at all here.
        let text = concat!(
            "Join INNER on=[(#0.0::INTEGER < #1.0::INTEGER)::BOOLEAN]\n",
            "  Get memory.main.l AS l #0 [a::INTEGER]\n",
            "  Get memory.main.r AS r #1 [a::INTEGER]\n"
        );
        let tables = &[("l", 150_000), ("r", 10_000)];
        let counts = &[("l", "a", 25), ("r", "a", 25)];
        assert_eq!(counted(text, tables, counts), Some(150_000));
    }

    #[test]
    fn a_projection_that_carries_a_column_through_carries_its_count_through_as_well() {
        // Every query written against a view goes through one of these, so without this the
        // counts would be read by almost nothing.
        let text = concat!(
            "Join INNER on=[(#1.0::INTEGER = #3.0::INTEGER)::BOOLEAN]\n",
            "  Project #1 [#0.0::INTEGER AS a]\n",
            "    Get memory.main.customer AS customer #0 [a::INTEGER]\n",
            "  Project #3 [#2.0::INTEGER AS a]\n",
            "    Get memory.main.supplier AS supplier #2 [a::INTEGER]\n"
        );
        let tables = &[("customer", 150_000), ("supplier", 10_000)];
        let counts = &[("customer", "a", 25), ("supplier", "a", 25)];
        assert_eq!(counted(text, tables, counts), Some(60_000_000));
    }

    #[test]
    fn a_projection_that_computes_something_is_where_the_count_stops() {
        // What the addition did to the number of distinct values under it is not something this
        // knows, and the count of the input is a wrong answer rather than an approximate one.
        let text = concat!(
            "Join INNER on=[(#1.0::INTEGER = #3.0::INTEGER)::BOOLEAN]\n",
            "  Project #1 [\"+\"(#0.0::INTEGER, 1::INTEGER)::INTEGER AS a]\n",
            "    Get memory.main.customer AS customer #0 [a::INTEGER]\n",
            "  Project #3 [#2.0::INTEGER AS a]\n",
            "    Get memory.main.supplier AS supplier #2 [a::INTEGER]\n"
        );
        let tables = &[("customer", 150_000), ("supplier", 10_000)];
        let counts = &[("customer", "a", 25), ("supplier", "a", 25)];
        assert_eq!(counted(text, tables, counts), Some(150_000));
    }

    #[test]
    fn a_column_with_no_distinct_values_at_all_is_not_divided_by() {
        // An empty column or a column of nothing but nulls. Neither is something to divide by, and
        // the estimate falls back to the shape it used before there were any counts.
        let text = joined("l", "r");
        let tables = &[("l", 1_000), ("r", 1_000)];
        let counts = &[("l", "a", 0), ("r", "a", 0)];
        assert_eq!(counted(&text, tables, counts), Some(1_000));
    }

    #[test]
    fn an_equality_against_a_constant_keeps_one_value_out_of_the_count() {
        // TPC-H o_clerk written small. A fifth of three hundred thousand orders is sixty thousand
        // and the answer is fifteen hundred, which is the error this rule exists to remove.
        let text = filtered("(#0.0::INTEGER = 3::INTEGER)::BOOLEAN");
        let tables = &[("t", 300_000)];
        assert_eq!(counted(&text, tables, &[]), Some(60_000));
        assert_eq!(counted(&text, tables, &[("t", "a", 200)]), Some(1_500));
    }

    #[test]
    fn a_count_that_says_more_rows_than_the_constant_did_is_still_the_count() {
        // Three values is a third, which is above the fifth the constant guessed. The rule is to
        // use the number where there is one rather than to make the answer smaller, and this is
        // the direction the constant was wrong in for o_orderstatus.
        let text = filtered("(#0.0::INTEGER = 3::INTEGER)::BOOLEAN");
        assert_eq!(counted(&text, &[("t", 900_000)], &[("t", "a", 3)]), Some(300_000));
    }

    #[test]
    fn two_counted_equalities_divide_by_both_counts() {
        // The independence assumption, which is the same one the constant made when it multiplied
        // two fifths together, with the counts standing in for the fifths.
        let text = filtered(
            "((#0.0::INTEGER = 3::INTEGER)::BOOLEAN AND (#0.1::INTEGER = 4::INTEGER)::BOOLEAN)::BOOLEAN",
        );
        let tables = &[("t", 200_000)];
        let counts = &[("t", "a", 25), ("t", "b", 40)];
        assert_eq!(counted(&text, tables, counts), Some(200));
    }

    #[test]
    fn a_column_nobody_counted_keeps_the_constant_rather_than_becoming_one_row() {
        // One over the rows is one row, so a fallback to the row count here would call every
        // equality on an uncounted column a single row lookup. The count on the other column is
        // still read, so one uncounted condition does not throw away the one next to it.
        let text = filtered(
            "((#0.0::INTEGER = 3::INTEGER)::BOOLEAN AND (#0.1::INTEGER = 4::INTEGER)::BOOLEAN)::BOOLEAN",
        );
        let tables = &[("t", 1_000_000)];
        assert_eq!(counted(&text, tables, &[]), Some(40_000));
        assert_eq!(counted(&text, tables, &[("t", "a", 50)]), Some(4_000));
    }

    #[test]
    fn only_an_equality_against_a_constant_reads_the_count() {
        // A range is about order rather than about one value, and two columns comparing to each
        // other pick a value neither count describes. Both keep the constant with the count sitting
        // right there unread.
        let tables = &[("t", 1_000_000)];
        let counts = &[("t", "a", 50), ("t", "b", 50)];
        let above = filtered("(#0.0::INTEGER > 3::INTEGER)::BOOLEAN");
        assert_eq!(counted(&above, tables, counts), Some(200_000));
        let other = filtered("(#0.0::INTEGER <> 3::INTEGER)::BOOLEAN");
        assert_eq!(counted(&other, tables, counts), Some(200_000));
        let columns = filtered("(#0.0::INTEGER = #0.1::INTEGER)::BOOLEAN");
        assert_eq!(counted(&columns, tables, counts), Some(200_000));
    }

    #[test]
    fn an_equality_reads_the_count_whichever_side_the_constant_is_on() {
        let text = filtered("(3::INTEGER = #0.0::INTEGER)::BOOLEAN");
        assert_eq!(counted(&text, &[("t", 100_000)], &[("t", "a", 50)]), Some(2_000));
    }

    #[test]
    fn a_projection_carries_a_count_up_to_a_filter_as_well() {
        // The same walk the join arithmetic makes, which matters here because every query written
        // against a view puts a projection between the filter and the scan that counted.
        let text = concat!(
            "Filter (#1.0::INTEGER = 3::INTEGER)::BOOLEAN\n",
            "  Project #1 [#0.0::INTEGER AS a]\n",
            "    Get memory.main.t AS t #0 [a::INTEGER]\n"
        );
        assert_eq!(counted(text, &[("t", 100_000)], &[("t", "a", 50)]), Some(2_000));
    }

    #[test]
    fn a_count_bigger_than_the_table_still_leaves_a_row() {
        // The floor the whole module has always had. A plan that believes a subtree produces no
        // rows is a plan that stops reading it, and a stale count is not a proof of anything.
        let text = filtered("(#0.0::INTEGER = 3::INTEGER)::BOOLEAN");
        assert_eq!(counted(&text, &[("t", 10)], &[("t", "a", 1_000_000)]), Some(1));
    }

    #[test]
    fn where_a_filter_got_its_fraction_from_is_printed() {
        // Three answers, so that somebody reading EXPLAIN and chasing a bad estimate can tell
        // which of the three this was without reading the predicate back.
        let tables = &[("t", 1_000_000)];
        let one = filtered("(#0.0::INTEGER = 3::INTEGER)::BOOLEAN");
        assert_eq!(
            counted_stat(&one, tables, &[("t", "a", 50)]).provenance(),
            Some(Provenance::Sketch)
        );
        assert_eq!(counted_stat(&one, tables, &[]).provenance(), Some(Provenance::Default));
        let both = filtered(
            "((#0.0::INTEGER = 3::INTEGER)::BOOLEAN AND (#0.1::INTEGER > 4::INTEGER)::BOOLEAN)::BOOLEAN",
        );
        assert_eq!(
            counted_stat(&both, tables, &[("t", "a", 50)]).provenance(),
            Some(Provenance::Propagation)
        );
    }

    #[test]
    fn a_counted_filter_is_still_a_guess() {
        // The count is a fact about the column and the fraction taken from it is not a fact about
        // the query. It assumes the values are spread evenly and that the filter is independent of
        // whatever happened underneath, which is the assumption that fails first.
        let text = filtered("(#0.0::INTEGER = 3::INTEGER)::BOOLEAN");
        let stat = counted_stat(&text, &[("t", 1_000_000)], &[("t", "a", 50)]);
        assert_eq!(stat.decide(), Some(&20_000));
        assert_eq!(stat.enable(), None);
        assert_eq!(stat.answer(), None);
    }

    /// A store of two columns, `a` then `b`, answering with a count fixed when it is built.
    ///
    /// The columns are deliberately in the other order from the scan above, because that is the
    /// case the name mapping exists for. A plan numbers a scan's columns by where they sit in what
    /// the scan produces, column pruning moves that, and the file's own order never moves. Handing
    /// the plan's position straight to the store would test the wrong column.
    #[derive(Debug)]
    struct Stub {
        /// What [`Zones::surviving`] answers, whatever it is asked.
        surviving: Option<u64>,
        /// What [`Zones::spread`] answers, whatever it is asked.
        spread: Option<f64>,
        /// Every test it was asked, so a test can check which column the estimator named.
        asked: Mutex<Vec<Test>>,
    }

    impl Stub {
        /// A store that answers the ceiling and refuses to interpolate, which is most of these.
        fn new(surviving: Option<u64>) -> Arc<Self> {
            Arc::new(Self { surviving, spread: None, asked: Mutex::new(Vec::new()) })
        }

        /// A store that interpolates and states no ceiling, which is the range case.
        fn spreading(spread: f64) -> Arc<Self> {
            Arc::new(Self { surviving: None, spread: Some(spread), asked: Mutex::new(Vec::new()) })
        }
    }

    impl Zones for Stub {
        fn column(&self, name: &str) -> Option<usize> {
            match name {
                "b" => Some(0),
                "a" => Some(1),
                _ => None,
            }
        }

        fn surviving(&self, tests: &[Test]) -> Option<u64> {
            self.asked.lock().expect("no test panics while holding this").extend_from_slice(tests);
            self.surviving
        }

        fn spread(&self, tests: &[Test]) -> Option<Spread> {
            self.asked.lock().expect("no test panics while holding this").extend_from_slice(tests);
            self.spread.map(|fraction| Spread { fraction, read: tests.len() })
        }
    }

    /// A two column scan, whose columns the stub above numbers the other way round.
    fn bounded_scan() -> String {
        "Get memory.main.t AS t #0 [a::INTEGER, b::INTEGER]\n".to_string()
    }

    /// The estimate for a plan whose table zero is the given store.
    fn zoned(text: &str, rows: u64, zones: &Arc<Stub>) -> Stat<u64> {
        let stats = facts(&[("t", rows)]);
        let mut plan =
            Plan::parse(text).unwrap_or_else(|error| panic!("{text} did not parse: {error}"));
        plan.set_zones(0, Arc::clone(zones) as Arc<dyn Zones>);
        rows_stat(&plan, plan.root(), &stats)
    }

    #[test]
    fn a_filter_the_bounds_rule_out_entirely_is_exactly_no_rows() {
        // The one answer bounds give that is a fact rather than a guess. No row group holds a
        // value the constant falls inside, so there is no row to produce, and saying so exactly
        // lets everything above it plan against zero instead of against a fifth of the file.
        let text = format!("Filter (#0.0::INTEGER > 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let zones = Stub::new(Some(0));
        assert_eq!(zoned(&text, 10_000_000, &zones), Stat::exact(0, Provenance::ZoneMap));
    }

    #[test]
    fn a_ceiling_below_the_guess_replaces_it_and_says_it_is_a_ceiling() {
        // On the ten million row smoke file `id < 1000` guesses two million and the bounds leave
        // 122,880, against a truth of a thousand. The number the planner gets is the smaller one
        // and it is marked as a bound rather than as an estimate, because that is what it is: the
        // answer is somewhere between no rows and this, and no guess is involved in the ceiling.
        let text = format!("Filter (#0.0::INTEGER < 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let zones = Stub::new(Some(100_000));
        let capped = zoned(&text, 10_000_000, &zones);
        assert_eq!(capped.value(), Some(&100_000));
        assert_eq!(
            capped.class(),
            Some(Class::Certified { bound: 1.0, direction: Direction::AtMost })
        );
        assert_eq!(capped.provenance(), Some(Provenance::ZoneMap));
        // A ceiling of one row is allowed, unlike the guess, which floors at one for a different
        // reason: the guess floors because a relation estimated away is a subtree nobody reads,
        // and a ceiling of one is a fact that happens to be one.
        let tiny = Stub::new(Some(1));
        assert_eq!(zoned(&text, 10_000_000, &tiny).value(), Some(&1));
    }

    #[test]
    fn a_ceiling_above_the_guess_leaves_the_guess_alone() {
        // This is what makes the whole thing safe. A bound says nothing about how far under it the
        // answer sits, so replacing a guess that is already below it would be trading a number for
        // a worse one. `k = 42` on the smoke file is the case: every row group holds a 42 in its
        // range, the bounds rule nothing out, and the estimate stays the two million it was.
        let text = format!("Filter (#0.0::INTEGER < 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let wide = Stub::new(Some(10_000_000));
        assert_eq!(
            zoned(&text, 10_000_000, &wide),
            Stat::estimated(2_000_000, Provenance::Default)
        );
    }

    #[test]
    fn a_store_that_cannot_answer_leaves_the_estimate_exactly_as_it_was() {
        let text = format!("Filter (#0.0::INTEGER < 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let quiet = Stub::new(None);
        assert_eq!(zoned(&text, 1_000_000, &quiet), stat(&text, &[("t", 1_000_000)]));
        assert_eq!(zoned(&text, 1_000_000, &quiet), Stat::estimated(200_000, Provenance::Default));
    }

    #[test]
    fn a_filter_that_reads_as_no_test_at_all_does_not_ask_the_store() {
        // Asking with no tests would come back with the whole file, which is the right number and
        // the wrong provenance: nothing was ruled out by any bound, so nothing should claim to
        // have been. A comparison of two columns is the case, since bounds on one say nothing
        // about the other's value in the same row.
        let text = format!("Filter (#0.0::INTEGER = #0.1::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let zones = Stub::new(Some(7));
        assert_eq!(zoned(&text, 1_000_000, &zones), Stat::estimated(200_000, Provenance::Default));
        assert!(zones.asked.lock().expect("not poisoned").is_empty(), "it was never asked");
    }

    #[test]
    fn a_range_the_store_can_interpolate_takes_its_fraction_rather_than_the_constant() {
        // The number that matters most in this file. Every range in TPC-H took a fifth whatever it
        // asked for, and `l_shipdate <= '1998-09-02'` keeps ninety eight percent of lineitem while
        // the two ends of the column sit in the footer saying 1992 and 1998.
        let text = format!("Filter (#0.0::INTEGER < 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let zones = Stub::spreading(0.98);
        assert_eq!(zoned(&text, 1_000_000, &zones), Stat::estimated(980_000, Provenance::ZoneMap));
    }

    #[test]
    fn a_fraction_above_the_constant_is_taken_as_readily_as_one_below_it() {
        // The rule is to use the number where there is one and not to make the answer smaller. All
        // sixteen ranges measured on TPC-H were underestimates, so the fifth was too small far more
        // often than it was too large, and a rule that only ever cut would have fixed none of them.
        let text = format!("Filter (#0.0::INTEGER < 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let narrow = Stub::spreading(0.01);
        assert_eq!(zoned(&text, 1_000_000, &narrow), Stat::estimated(10_000, Provenance::ZoneMap));
    }

    #[test]
    fn a_fraction_of_nothing_is_still_a_row_and_is_still_a_guess() {
        // The floor the guess has always had. A relation estimated away is a subtree nobody reads,
        // and an interpolated zero is an assumption about how values are spread rather than a fact
        // that the file holds none, so it does not get the exactness the ceiling of zero gets.
        let text = format!("Filter (#0.0::INTEGER < 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let empty = Stub::spreading(0.0);
        assert_eq!(zoned(&text, 1_000_000, &empty), Stat::estimated(1, Provenance::ZoneMap));
    }

    #[test]
    fn a_counted_condition_and_an_interpolated_one_report_the_arithmetic_over_both() {
        // Two suppliers answering two conditions of one filter. Neither name is the truth about
        // where the fraction came from, so the reader gets told it was worked out rather than read.
        let text = format!(
            "Filter ((#0.0::INTEGER = 5::INTEGER)::BOOLEAN AND (#0.1::INTEGER < 9::INTEGER)::BOOLEAN)::BOOLEAN\n  {}",
            bounded_scan()
        );
        let mut stats = facts(&[("t", 1_000_000)]);
        stats.record_distinct("memory", "main", "t", "a", 10);
        let mut plan = Plan::parse(&text).expect("parses");
        let zones = Stub::spreading(0.5);
        plan.set_zones(0, Arc::clone(&zones) as Arc<dyn Zones>);
        let stat = rows_stat(&plan, plan.root(), &stats);
        // A tenth for the equality from the count, a half for the range from the bounds.
        assert_eq!(stat, Stat::estimated(50_000, Provenance::Propagation));
    }

    #[test]
    fn a_condition_the_store_could_not_read_still_costs_the_constant() {
        // Two conditions, one of which is a comparison of two columns and reads as no test at all.
        // The store answers for the one it was handed and says so, and the other is charged the
        // fifth on its own. Charging nothing for it would say a condition nobody can estimate keeps
        // every row, and charging the fifth twice would guess at a condition that was answered.
        let text = format!(
            "Filter ((#0.0::INTEGER < 9::INTEGER)::BOOLEAN AND (#0.0::INTEGER = #0.1::INTEGER)::BOOLEAN)::BOOLEAN\n  {}",
            bounded_scan()
        );
        // A half from the store and a fifth for the other one, which is a tenth of a million.
        let zones = Stub::spreading(0.5);
        assert_eq!(
            zoned(&text, 1_000_000, &zones),
            Stat::estimated(100_000, Provenance::Propagation)
        );
    }

    #[test]
    fn the_count_answers_an_equality_before_the_bounds_are_asked_to_interpolate_it() {
        // One value out of a range is not a fraction a range knows, so the store refuses equality
        // anyway, but the order is worth pinning: the count is the better number and goes first.
        let text = format!("Filter (#0.0::INTEGER = 5::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let mut stats = facts(&[("t", 1_000_000)]);
        stats.record_distinct("memory", "main", "t", "a", 8);
        let mut plan = Plan::parse(&text).expect("parses");
        plan.set_zones(0, Stub::spreading(0.5) as Arc<dyn Zones>);
        assert_eq!(
            rows_stat(&plan, plan.root(), &stats),
            Stat::estimated(125_000, Provenance::Sketch)
        );
    }

    #[test]
    fn the_column_the_store_is_asked_about_is_the_one_the_plan_named_and_not_the_position() {
        // The bug this mapping exists to stop. The plan's column zero is `a` and the store's
        // column zero is `b`, so a test handed straight through would rule out row groups on the
        // wrong column's bounds. That drops rows the query wanted, which is a wrong answer and not
        // a slow one.
        //
        // The store is asked twice for this filter, once for the ceiling and once for the
        // fraction, and both have to name the same column. Checking every ask rather than the
        // first means adding a third question later cannot quietly skip the mapping.
        let text = format!("Filter (#0.0::INTEGER < 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        let zones = Stub::new(Some(100));
        zoned(&text, 1_000_000, &zones);
        let asked = zones.asked.lock().expect("not poisoned");
        assert!(!asked.is_empty(), "it was asked");
        for test in asked.iter() {
            assert_eq!(test.column, 1, "`a` is the store's column one");
            assert_eq!(test.op, Op::Less);
        }
    }

    #[test]
    fn a_table_with_no_store_recorded_is_estimated_the_way_it_always_was() {
        // Which is every table today except a `read_parquet` of one file, so this is the path
        // almost every query still takes and it has to be untouched.
        let text = format!("Filter (#0.0::INTEGER < 9::INTEGER)::BOOLEAN\n  {}", bounded_scan());
        assert_eq!(stat(&text, &[("t", 1_000_000)]), Stat::estimated(200_000, Provenance::Default));
    }
}