oxilite-core 0.9.0

//! Statistics used by the join-order planner.
//!
//! Stats are refreshed explicitly (`optimize()`, or after a bulk load) rather than on every
//! write: on D1 every index entry written is billed, and a per-write counter would be a hot
//! row. Stale stats only degrade plan quality, never correctness.
//!
// @lat: [[architecture#Query planner#Statistics]]

use crate::encoding::rdf_type_id;
use crate::error::Result;
use crate::sql::{col, expect_len, Capabilities, Request, Response, Statement};
use std::collections::{BTreeSet, HashMap};

/// Per-predicate statistics.
#[derive(Debug, Clone, Copy, PartialEq)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct PredicateStats {
    pub triples: f64,
    pub distinct_subjects: f64,
    pub distinct_objects: f64,
}

/// Planner statistics loaded from `stats_pred` / `stats_class`.
#[derive(Debug, Clone, Default)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct Stats {
    /// `true` when `optimize()` has been run at least once.
    pub available: bool,
    pub total: f64,
    pub predicates: HashMap<i64, PredicateStats>,
    /// Instance count per `rdf:type` class.
    pub classes: HashMap<i64, f64>,
    /// Triples per frequent (predicate, object) pair of low-cardinality predicates: values
    /// much more common than their predicate's average (skew the average would hide).
    pub pairs: HashMap<(i64, i64), f64>,
    pub graph_index: bool,
    /// The full-text index exists (`oxl:textMatch` compiles to FTS5).
    pub text_index: bool,
    /// Transitive properties (from `tbox_closure`, any scope), for query-time reasoning.
    pub transitive: BTreeSet<i64>,
    /// Graphs with a closure scope of their own (see `registry::ontology_axioms`).
    pub schema_scopes: BTreeSet<i64>,
    /// The versioning level and history of the store (see `version`).
    pub version: crate::version::VersionState,
    /// Vector index definitions of `<oxilite:vectors>` (loaded on backends with vectors).
    #[cfg_attr(feature = "serde", serde(skip))]
    pub vector_indexes: Vec<crate::vector::VectorIndex>,
    /// Why descriptions in `<oxilite:vectors>` are not valid definitions.
    #[cfg_attr(feature = "serde", serde(skip))]
    pub vector_problems: Vec<String>,
    /// Fingerprints of the built vector tables, by lower-case index name.
    #[cfg_attr(feature = "serde", serde(skip))]
    pub vector_built: std::collections::BTreeMap<String, String>,
}

fn id_col(caps: &Capabilities, c: &str) -> String {
    if caps.int64_as_text {
        format!("CAST({c} AS TEXT)")
    } else {
        c.into()
    }
}

impl Stats {
    /// Statements that load statistics and store settings.
    pub fn load_request(caps: &Capabilities) -> Request {
        let mut r = Request::read(vec![
            Statement::new("SELECT key, value FROM oxilite_meta"),
            Statement::new(format!(
                "SELECT {}, triples, distinct_s, distinct_o FROM stats_pred",
                id_col(caps, "p")
            )),
            Statement::new(format!(
                "SELECT {}, instances FROM stats_class",
                id_col(caps, "o")
            )),
            crate::reason::transitive_statement(|c| id_col(caps, c)),
            Statement::new(format!(
                "SELECT {}, {}, n FROM stats_po",
                id_col(caps, "p"),
                id_col(caps, "o")
            )),
            crate::registry::scopes_statement(|c| id_col(caps, c)),
        ]);
        // Only a backend with vector functions reads the definitions, so the load request on
        // every other backend is what it always was.
        if caps.vectors {
            r.statements
                .push(crate::vector::definitions_statement(|c| id_col(caps, c)));
        }
        r
    }

    pub fn from_response(response: &Response) -> Result<Self> {
        if response.len() != 7 {
            expect_len(response, 6)?;
        }
        let mut stats = Self::default();
        for row in &response[0].rows {
            let key = col(row, 0)?.as_str().unwrap_or_default();
            let value = col(row, 1)?.clone().into_string().unwrap_or_default();
            if let Some(name) = key.strip_prefix("vector:") {
                stats.vector_built.insert(name.to_owned(), value);
                continue;
            }
            match key {
                "graph_index" => stats.graph_index = value == "1",
                "text_index" => stats.text_index = value == "1",
                k @ ("versioning" | "history" | "stamp_column" | "stamp_index" | "as_of_index") => {
                    stats.version.absorb(k, &value)
                }
                "total" => {
                    stats.total = value.parse().unwrap_or(0.0);
                    stats.available = true;
                }
                _ => {}
            }
        }
        for row in &response[1].rows {
            let (Some(p), Some(t), Some(ds), Some(d_o)) = (
                col(row, 0)?.as_i64(),
                col(row, 1)?.as_f64(),
                col(row, 2)?.as_f64(),
                col(row, 3)?.as_f64(),
            ) else {
                continue;
            };
            stats.predicates.insert(
                p,
                PredicateStats {
                    triples: t,
                    distinct_subjects: ds.max(1.0),
                    distinct_objects: d_o.max(1.0),
                },
            );
        }
        for row in &response[2].rows {
            if let (Some(o), Some(n)) = (col(row, 0)?.as_i64(), col(row, 1)?.as_f64()) {
                stats.classes.insert(o, n);
            }
        }
        for row in &response[3].rows {
            if let Some(p) = col(row, 0)?.as_i64() {
                stats.transitive.insert(p);
            }
        }
        for row in &response[5].rows {
            if let Some(g) = col(row, 0)?.as_i64() {
                stats.schema_scopes.insert(g);
            }
        }
        for row in &response[4].rows {
            if let (Some(p), Some(o), Some(n)) = (
                col(row, 0)?.as_i64(),
                col(row, 1)?.as_i64(),
                col(row, 2)?.as_f64(),
            ) {
                stats.pairs.insert((p, o), n);
            }
        }
        if let Some(defs) = response.get(6) {
            let (indexes, problems) = crate::vector::definitions_from_rows(&defs.rows);
            stats.vector_indexes = indexes;
            stats.vector_problems = problems;
        }
        Ok(stats)
    }

    /// Statements recomputing statistics, the schema closure and the shape index (run by
    /// `optimize()`).
    pub fn refresh_request() -> Request {
        let mut r = Self::refresh_statements();
        r.extend(crate::reason::closure_statements());
        r.extend(crate::shapes::refresh_statements());
        Request::atomic(r)
    }

    fn refresh_statements() -> Vec<Statement> {
        vec![
            "DELETE FROM stats_pred".into(),
            "INSERT INTO stats_pred(p, triples, distinct_s, distinct_o) \
             SELECT p, COUNT(*), COUNT(DISTINCT s), COUNT(DISTINCT o) FROM quads GROUP BY p"
                .into(),
            "DELETE FROM stats_class".into(),
            Statement::new(format!(
                "INSERT INTO stats_class(o, instances) SELECT o, COUNT(*) FROM quads WHERE p = {} GROUP BY o",
                rdf_type_id()
            )),
            "INSERT OR REPLACE INTO oxilite_meta(key, value) SELECT 'total', CAST(COUNT(*) AS TEXT) FROM quads"
                .into(),
            "DELETE FROM stats_po".into(),
            Statement::new(format!(
                "INSERT INTO stats_po(p, o, n) SELECT q.p, q.o, COUNT(*) FROM quads q JOIN stats_pred sp ON sp.p = q.p \
                 WHERE sp.distinct_o <= 1024 AND q.p <> {} GROUP BY q.p, q.o, sp.triples, sp.distinct_o \
                 HAVING COUNT(*) >= 4.0 * sp.triples / sp.distinct_o ORDER BY COUNT(*) DESC LIMIT 2000",
                rdf_type_id()
            )),
        ]
    }
}