rudb-exec 0.3.36

//! The five catalog tables: what is attached, what is in it, and what somebody created there.
//!
//! `duckdb_databases()`, `duckdb_schemas()`, `duckdb_tables()`, `duckdb_views()` and
//! `duckdb_columns()`, the metadata tables whose rows come out of the catalog rather than out of a
//! list this binary was compiled with.
//! `rudb_functions::entrycatalog` has their columns, because the binder resolves the call and needs
//! the columns before there is a catalog in reach, and the rows are here because this is where one is.
//!
//! `duckdb_columns()` lists a view's columns as well as a table's. What it reads is the list the
//! binder wrote down the last time the view was bound, which is a cache that goes stale, and that is
//! upstream's design rather than a shortcut taken here. See `rudb_catalog::View` for the measurement.
//!
//! `duckdb_views()` is the fifth and it reports the statement written back out, which the binder
//! wrote down at `CREATE VIEW` using `rudb_parse::deparse`. The pin reports a deparse there too
//! rather than the text somebody typed, which is measured, so the column is a comparison like any
//! other and not a place where the two are allowed to differ.
//!
//! Every table is walked in catalog order, which is the order things were attached and created in.
//! That is not sorted and it is not reproduced from the pin either, whose order is its own catalog's.
//! A client that wants an order writes one, which is what the two corpus records that read these
//! tables already do.
//!
//! Most of the columns are a fact about a database rudb does not have yet. `path` is null because
//! every database here is in memory, `readonly`, `encrypted` and `cipher` say so, and `options` is
//! empty because `ATTACH` takes none. The day there is a file behind a database these columns say
//! something.
//!
//! `internal` is read off the database an entry is in rather than being a constant. A session has
//! three of them, `memory` to create in and `system` and `temp` that the engine owns, so the column
//! separates what somebody wrote from what rudb shipped with, which is the difference a client asks
//! about when it lists what is in a database. See `rudb_catalog::system` for what is in `system`.
//!
//! `sql`, `parent_schema` and `parent_schema_oid` are null on every schema row. Upstream's are null
//! too on everything it returns from a fresh session, because a schema created by `CREATE SCHEMA`
//! has no stored text and nothing nests schemas.

use rudb_catalog::{Catalog, Database, Schema, Table};
use rudb_common::{LogicalType, Result, Value};
use rudb_functions::{
    DUCKDB, canonical, column_fields, database_fields, numeric_facts, schema_fields,
    show_database_fields, show_expanded_fields, show_table_fields, table_fields, type_oid,
    view_fields,
};
use rudb_parse::quoted;
use rudb_plan::{Plan, Slice};

use crate::metadata::{Metadata, text};

/// Every attached database, in the columns the plan asked for.
///
/// # Errors
///
/// If the plan asks for a column this table does not have.
pub(crate) fn databasenames(
    catalog: &Catalog,
    plan: &Plan,
    index: u32,
    columns: Slice,
) -> Result<Metadata> {
    let mut rows = Vec::with_capacity(catalog.databases().len());
    for database in catalog.databases() {
        rows.push(vec![
            text(database.name()),
            Value::BigInt(database.oid()),
            Value::Null,
            Value::Null,
            empty(),
            Value::Boolean(database.internal()),
            text(DUCKDB),
            Value::Boolean(false),
            Value::Boolean(false),
            Value::Null,
            empty(),
        ]);
    }
    Metadata::new("duckdb_databases", &database_fields(), &rows, plan, index, columns)
}

/// Every schema in every attached database, in the columns the plan asked for.
///
/// # Errors
///
/// If the plan asks for a column this table does not have.
pub(crate) fn schemanames(
    catalog: &Catalog,
    plan: &Plan,
    index: u32,
    columns: Slice,
) -> Result<Metadata> {
    let mut rows = Vec::new();
    for database in catalog.databases() {
        for schema in database.schemas() {
            rows.push(vec![
                Value::BigInt(schema.oid()),
                text(database.name()),
                Value::BigInt(database.oid()),
                text(schema.name()),
                Value::Null,
                empty(),
                // True on every schema upstream returns from a fresh session, including `memory.main`
                // which is the one rudb has. A schema somebody made with `CREATE SCHEMA` is false
                // there, and the day rudb can tell the two apart this stops being a constant.
                Value::Boolean(true),
                Value::Null,
                Value::Null,
                Value::Null,
            ]);
        }
    }
    Metadata::new("duckdb_schemas", &schema_fields(), &rows, plan, index, columns)
}

/// Every base table in the catalog, in the columns the plan asked for.
///
/// # Errors
///
/// If the plan asks for a column this table does not have.
pub(crate) fn tablenames(
    catalog: &Catalog,
    plan: &Plan,
    index: u32,
    columns: Slice,
) -> Result<Metadata> {
    let mut rows = Vec::new();
    for (database, schema, table) in entries(catalog) {
        let count = i64::try_from(table.columns().len()).unwrap_or(i64::MAX);
        rows.push(vec![
            text(database.name()),
            Value::BigInt(database.oid()),
            text(schema.name()),
            Value::BigInt(schema.oid()),
            text(&table.name().table),
            Value::BigInt(table.oid()),
            Value::Null,
            empty(),
            Value::Boolean(database.internal()),
            // Neither of these can be true yet. `CREATE TEMP TABLE` is not a statement rudb takes
            // and a primary key is not a constraint it stores, so both are a constant rather than a
            // fact read off the entry, and both stop being one the day the DDL grows the clause.
            Value::Boolean(false),
            Value::Boolean(false),
            Value::BigInt(i64::try_from(table.rows().len()).unwrap_or(i64::MAX)),
            Value::BigInt(count),
            Value::BigInt(0),
            Value::BigInt(0),
            text(&create_table(table)),
        ]);
    }
    Metadata::new("duckdb_tables", &table_fields(), &rows, plan, index, columns)
}

/// Every view in the catalog, in the columns the plan asked for.
///
/// `sql` is the statement written back out, which the binder wrote down at creation. `column_count`
/// is the length of the column cache on the entry, so it is as stale as the cache is, which is what
/// upstream reports too.
///
/// `is_bound` says whether that cache holds anything, and `column_count` is null when it does not.
/// A view somebody wrote is bound at `CREATE VIEW` and cannot get into the catalog without one, so
/// the two of them only say no for a view the engine ships with, which goes in unbound and is bound
/// at the first read. That is the pin's answer as well, measured on a fresh session and again after
/// reading one of them.
///
/// `internal` and `temporary` are both read off the database the view is in. Upstream sets both on
/// everything in `system`, which reads oddly for `temporary` on a view that is not in `temp`, and it
/// is what the pin prints for all 47 of them.
///
/// # Errors
///
/// If the plan asks for a column this table does not have.
pub(crate) fn viewnames(
    catalog: &Catalog,
    plan: &Plan,
    index: u32,
    columns: Slice,
) -> Result<Metadata> {
    let mut rows = Vec::new();
    for database in catalog.databases() {
        for schema in database.schemas() {
            for view in schema.views() {
                let columns = view.columns();
                rows.push(vec![
                    text(database.name()),
                    Value::BigInt(database.oid()),
                    text(schema.name()),
                    Value::BigInt(schema.oid()),
                    text(&view.name().table),
                    Value::BigInt(view.oid()),
                    Value::Null,
                    empty(),
                    Value::Boolean(database.internal()),
                    Value::Boolean(database.internal()),
                    if columns.is_empty() {
                        Value::Null
                    } else {
                        Value::BigInt(i64::try_from(columns.len()).unwrap_or(i64::MAX))
                    },
                    text(view.statement()),
                    Value::Boolean(!columns.is_empty()),
                ]);
            }
        }
    }
    Metadata::new("duckdb_views", &view_fields(), &rows, plan, index, columns)
}

/// Every column of every base table, in the columns the plan asked for.
///
/// # Errors
///
/// If the plan asks for a column this table does not have.
pub(crate) fn columnnames(
    catalog: &Catalog,
    plan: &Plan,
    index: u32,
    columns: Slice,
) -> Result<Metadata> {
    let mut rows = Vec::new();
    for database in catalog.databases() {
        for schema in database.schemas() {
            for table in schema.tables() {
                for (at, column) in table.columns().iter().enumerate() {
                    rows.push(column_row(
                        database,
                        schema,
                        &table.name().table,
                        table.oid(),
                        at,
                        &column.name,
                        &column.ty,
                        !column.not_null,
                    ));
                }
            }
            for view in schema.views() {
                for (at, field) in view.columns().iter().enumerate() {
                    // Nullable on every column of every view the pin returns, including one that
                    // reads a `NOT NULL` column straight through, so it is a constant here and not a
                    // fact carried over from the table underneath.
                    rows.push(column_row(
                        database,
                        schema,
                        &view.name().table,
                        view.oid(),
                        at,
                        &field.name,
                        &field.ty,
                        true,
                    ));
                }
            }
        }
    }
    Metadata::new("duckdb_columns", &column_fields(), &rows, plan, index, columns)
}

/// One row of `duckdb_columns()`, which is the same twenty one columns for a table and for a view.
#[expect(clippy::too_many_arguments, reason = "a row of a twenty one column table")]
fn column_row(
    database: &Database,
    schema: &Schema,
    table: &str,
    oid: i64,
    at: usize,
    name: &str,
    ty: &LogicalType,
    nullable: bool,
) -> Vec<Value> {
    let (precision, radix, scale) = numeric_facts(ty);
    vec![
        text(database.name()),
        Value::BigInt(database.oid()),
        text(schema.name()),
        Value::BigInt(schema.oid()),
        text(table),
        Value::BigInt(oid),
        text(name),
        // One based, which is the pin's answer and not the position in the vector.
        Value::Integer(i32::try_from(at + 1).unwrap_or(i32::MAX)),
        Value::Null,
        Value::Boolean(database.internal()),
        Value::Null,
        Value::Boolean(nullable),
        text(&ty.to_string()),
        type_oid(&canonical(ty)).map_or(Value::Null, Value::BigInt),
        // Null even on a VARCHAR the DDL gave a length, because the pin reports null there too:
        // DuckDB parses the length modifier and then drops it, so by the time a column is in a
        // catalog there is no length left to report.
        Value::Null,
        precision.map_or(Value::Null, Value::Integer),
        radix.map_or(Value::Null, Value::Integer),
        scale.map_or(Value::Null, Value::Integer),
        empty(),
        Value::Boolean(false),
        Value::Null,
    ]
}

/// The name of everything an unqualified name can reach, which is what `PRAGMA show_tables` is.
///
/// Sorted by name, and views sit among the tables rather than after them, which is the pin's answer
/// and the only sensible one for a list whose whole purpose is to say what a name will find.
///
/// The search path is the default schema of the default database, plus `temp` on the pin. rudb has
/// neither `CREATE SCHEMA` nor a temporary table yet, so the path is one schema and this walks it
/// directly. The day either lands this is the function that has to grow a real search path rather
/// than a lookup of one name.
///
/// # Errors
///
/// If the plan asks for a column this table does not have.
pub(crate) fn showtables(
    catalog: &Catalog,
    plan: &Plan,
    index: u32,
    columns: Slice,
) -> Result<Metadata> {
    let mut names = Vec::new();
    for database in catalog.databases() {
        if !database.name().eq_ignore_ascii_case(catalog.default_catalog()) {
            continue;
        }
        for schema in database.schemas() {
            if !schema.name().eq_ignore_ascii_case(catalog.default_schema()) {
                continue;
            }
            names.extend(schema.tables().iter().map(|table| table.name().table.clone()));
            names.extend(schema.views().iter().map(|view| view.name().table.clone()));
        }
    }
    names.sort();
    let rows: Vec<Vec<Value>> = names.iter().map(|name| vec![text(name)]).collect();
    Metadata::new("pragma_show_tables", &show_table_fields(), &rows, plan, index, columns)
}

/// The name of everything that is attached, which is what `PRAGMA show_databases` is.
///
/// The databases the engine owns are left out, because the pin leaves its own out too and the
/// question the statement asks is which databases a client can write a name into.
///
/// # Errors
///
/// If the plan asks for a column this table does not have.
pub(crate) fn showdatabases(
    catalog: &Catalog,
    plan: &Plan,
    index: u32,
    columns: Slice,
) -> Result<Metadata> {
    let mut names: Vec<&str> = catalog
        .databases()
        .iter()
        .filter(|database| !database.internal())
        .map(Database::name)
        .collect();
    names.sort_unstable();
    let rows: Vec<Vec<Value>> = names.iter().map(|name| vec![text(name)]).collect();
    Metadata::new("pragma_show_databases", &show_database_fields(), &rows, plan, index, columns)
}

/// Every table and view anywhere with its columns beside it, which is `PRAGMA show_tables_expanded`.
///
/// Sorted by database, then schema, then name, and the engine's own databases are left out the way
/// `PRAGMA show_databases` leaves them out. The two list columns are the one place in these tables
/// where a table's shape is reported without a join, and the types are written the way the type
/// prints rather than the way it was declared.
///
/// `temporary` is false on every row, because `CREATE TEMPORARY TABLE` is not a statement rudb takes
/// yet and the pin only puts true there for an entry in its `temp` database. The day that statement
/// lands this column is read off the database the entry is in rather than being a constant.
///
/// # Errors
///
/// If the plan asks for a column this table does not have.
pub(crate) fn showtablesexpanded(
    catalog: &Catalog,
    plan: &Plan,
    index: u32,
    columns: Slice,
) -> Result<Metadata> {
    let mut rows = Vec::new();
    for database in catalog.databases() {
        if database.internal() {
            continue;
        }
        for schema in database.schemas() {
            for table in schema.tables() {
                rows.push(expanded_row(database, schema, &table.name().table, table.columns()));
            }
            for view in schema.views() {
                rows.push(expanded_row(database, schema, &view.name().table, &view.columns()));
            }
        }
    }
    // On the three name columns, which are the three the pin orders by. They are varchars on every
    // row, so the key is those three strings and nothing in it can be null.
    rows.sort_by_key(|row| {
        let at = |index: usize| match &row[index] {
            Value::Varchar(name) => name.clone(),
            _ => String::new(),
        };
        (at(0), at(1), at(2))
    });
    Metadata::new(
        "pragma_show_tables_expanded",
        &show_expanded_fields(),
        &rows,
        plan,
        index,
        columns,
    )
}

/// One row of `PRAGMA show_tables_expanded`, for a table or for a view.
fn expanded_row(
    database: &Database,
    schema: &Schema,
    name: &str,
    columns: &[rudb_common::Field],
) -> Vec<Value> {
    vec![
        text(database.name()),
        text(schema.name()),
        text(name),
        names_of(columns.iter().map(|field| field.name.clone())),
        names_of(columns.iter().map(|field| field.ty.to_string())),
        Value::Boolean(false),
    ]
}

/// A `VARCHAR[]` of whatever was handed in, which is how both list columns are built.
fn names_of(values: impl Iterator<Item = String>) -> Value {
    Value::List { element: LogicalType::Varchar, values: values.map(Value::Varchar).collect() }
}

/// Every base table in the catalog with the database and schema it is in.
fn entries(catalog: &Catalog) -> impl Iterator<Item = (&Database, &Schema, &Table)> {
    catalog.databases().iter().flat_map(|database| {
        database.schemas().iter().flat_map(move |schema| {
            schema.tables().iter().map(move |table| (database, schema, table))
        })
    })
}

/// The `CREATE TABLE` a table would be made by, which is what `duckdb_tables()` reports as `sql`.
///
/// Written back out rather than stored. The pin does the same, which is measured: a table created
/// with odd spacing and lower case type names comes back normalised, so the column is a deparse of
/// the entry and not the text somebody typed. Identifiers go through [`rudb_parse::quoted`], which
/// is the rule the binder already uses for a generated column name.
fn create_table(table: &Table) -> String {
    let columns: Vec<String> = table
        .columns()
        .iter()
        .map(|column| {
            let null = if column.not_null { " NOT NULL" } else { "" };
            format!("{} {}{null}", quoted(&column.name), column.ty)
        })
        .collect();
    format!("CREATE TABLE {}({});", quoted(&table.name().table), columns.join(", "))
}

/// An empty `MAP(VARCHAR, VARCHAR)`, which is what `tags` and `options` are on every row.
fn empty() -> Value {
    Value::map(LogicalType::Varchar, LogicalType::Varchar, Vec::new())
}