rupg-func 0.1.84

//! The text input and output of the OID alias types, such as `regclass` and `regprocedure`, and the functions `to_regclass` and the others, which give a null for a name that does not exist.
//!
//! A port of `src/backend/utils/adt/regproc.c` and of the lookups of `namespace.c` that it calls. A schema, a relation or a type is a built-in object or an object of the catalog of the session. The other kinds have only the built-in objects.
//!
//! The input separates two kinds of errors, as the `escontext` of PostgreSQL does. A soft error, such as a name that does not exist, is an error of the input function and a null of `to_regclass`. A hard error, such as a syntax error in a type name, is an error of both.

use std::borrow::Cow;

use rupg_analyze::Env;
use rupg_catalog::Catalog;
use rupg_common::{Error, Result, SqlState};
use rupg_pgcatalog::builtin::{self, Named, NamedRow, OperatorRow, ProcRow};
use rupg_types::{self as types, Record, RegInput, RegKind, Value, oid, qualified_name_list};

use crate::text::quote_identifier;
use crate::{Call, Kernel, Session, bad_value, type_error};

/// The OID of `pg_catalog`.
const PG_CATALOG: u32 = 11;
/// `PG_UTF8`, the encoding of the database.
const UTF8: i32 = 6;
/// The name of the encoding of the database, as `GetDatabaseEncodingName` gives it.
const UTF8_NAME: &str = "UTF8";
/// `FUNC_MAX_ARGS`.
const FUNC_MAX_ARGS: usize = 100;

/// The result of the input: the outer error is a hard error, and the inner error is a soft error.
type Soft<T> = Result<Result<T>>;

/// The session as the analyzer reads it, for the type names in the input of `regtype` and `regprocedure` and for the query of a view.
pub(crate) struct SessionEnv<'a>(pub(crate) &'a dyn Session);

impl Env for SessionEnv<'_> {
    fn search_path(&self) -> Vec<u32> {
        self.0.schemas().iter().filter_map(|name| namespace_oid(name, self.0)).collect()
    }

    fn namespace(&self, name: &str) -> Option<u32> {
        namespace_oid(name, self.0)
    }

    fn database(&self) -> String {
        self.0.database().to_string()
    }

    fn input(&self, ty: u32, text: &str, typmod: i32) -> Result<Value> {
        crate::io::input(ty, text, typmod, self.0)
    }

    fn catalog(&self) -> Option<&Catalog> {
        self.0.catalog()
    }

    fn temp_number(&self) -> Option<i32> {
        self.0.temp_number()
    }
}

/// `myTempNamespace`: the OID of the temporary schema of the session, or 0 when it has none yet.
pub(crate) fn temp_namespace(session: &dyn Session) -> u32 {
    session
        .catalog()
        .map_or(0, |catalog| rupg_analyze::temp_namespace(&SessionEnv(session), catalog))
}

/// `isTempOrTempToastNamespace`: true for the temporary schema of the session and its TOAST schema.
pub(crate) fn is_own_temp(oid: u32, session: &dyn Session) -> bool {
    let Some(n) = session.temp_number() else { return false };
    namespace_name(oid, session)
        .is_some_and(|name| name == format!("pg_temp_{n}") || name == format!("pg_toast_temp_{n}"))
}

/// The OID of the schema with this name. `pg_temp` names the temporary schema of the session.
fn namespace_oid(name: &str, session: &dyn Session) -> Option<u32> {
    if name == "pg_temp" {
        return Some(temp_namespace(session)).filter(|&oid| oid != 0);
    }
    builtin::named(Named::Namespace)
        .iter()
        .find(|row| row.name == name && !schema_dropped(row.oid, session))
        .map(|row| row.oid)
        .or_else(|| session.catalog()?.schema_by_name(name).map(|s| s.oid))
}

/// The name of the schema with this OID.
pub(crate) fn namespace_name(oid: u32, session: &dyn Session) -> Option<&str> {
    builtin::named_by_oid(Named::Namespace, oid)
        .filter(|row| !schema_dropped(row.oid, session))
        .map(|row| row.name)
        .or_else(|| session.catalog()?.schema(oid).map(|s| s.name.as_str()))
}

/// True when `DROP SCHEMA` removed the built-in schema with the OID.
fn schema_dropped(oid: u32, session: &dyn Session) -> bool {
    session.catalog().is_some_and(|catalog| catalog.schema_dropped(oid))
}

/// The schemas that an unqualified name of a function, an operator or another object that is not a relation or a type searches: `pg_catalog` first when `search_path` does not name it, then the schemas of `search_path` that exist. The lookups of `namespace.c` skip the temporary schema.
pub(crate) fn path(session: &dyn Session) -> Vec<u32> {
    let mut path = temp_path(session);
    let temp = temp_namespace(session);
    path.retain(|&ns| ns != temp);
    path
}

/// `StatisticsObjIsVisible`: true when the first statistics object with the name in [`path`] is this object, or `None` for an OID that is not a statistics object.
pub(crate) fn stat_ext_visible(oid: u32, session: &dyn Session) -> Option<bool> {
    let catalog = session.catalog()?;
    let stat = catalog.stat_ext(oid)?;
    let first = path(session)
        .into_iter()
        .find(|&ns| ns == stat.namespace || catalog.stat_ext_by_name(ns, &stat.name).is_some());
    Some(first == Some(stat.namespace))
}

/// `activeSearchPath`: the schemas that an unqualified name of a relation or a type searches. The temporary schema of the session comes first, then `pg_catalog`, when `search_path` does not name them.
pub(crate) fn temp_path(session: &dyn Session) -> Vec<u32> {
    let mut path: Vec<u32> =
        session.schemas().iter().filter_map(|n| namespace_oid(n, session)).collect();
    if !path.contains(&PG_CATALOG) {
        path.insert(0, PG_CATALOG);
    }
    let temp = temp_namespace(session);
    if temp != 0 && !path.contains(&temp) {
        path.insert(0, temp);
    }
    path
}

/// `quote_qualified_identifier`: the name with its schema, or the name alone when `schema` is `None`.
pub(crate) fn qualified(schema: Option<&str>, name: &str) -> String {
    match schema {
        Some(schema) => format!("{}.{}", quote_identifier(schema), quote_identifier(name)),
        None => quote_identifier(name),
    }
}

/// `NameListToString`: the names with a dot between them, without quotes.
fn name_list(names: &[Cow<'_, str>]) -> String {
    names.join(".")
}

/// The error of a soft error with this SQLSTATE and message.
fn soft<T>(state: SqlState, message: impl Into<String>) -> Soft<T> {
    Ok(Err(Error::new(state, message)))
}

/// `DeconstructQualifiedName`: the schema and the name of a list of names.
///
/// # Errors
///
/// `0A000` for a database that is not the current database, and `42601` for more than three names.
fn deconstruct<'n>(
    names: &'n [Cow<'_, str>],
    session: &dyn Session,
) -> Result<(Option<&'n str>, &'n str)> {
    match names {
        [name] => Ok((None, name)),
        [schema, name] => Ok((Some(schema), name)),
        [database, schema, name] => {
            if database != session.database() {
                return Err(Error::new(
                    SqlState::FEATURE_NOT_SUPPORTED,
                    format!("cross-database references are not implemented: {}", name_list(names)),
                ));
            }
            Ok((Some(schema), name))
        }
        _ => Err(Error::new(
            SqlState::SYNTAX_ERROR,
            format!("improper qualified name (too many dotted names): {}", name_list(names)),
        )),
    }
}

/// The schemas that a name searches: the schema that the name gives, or none when it does not exist, as the lookups with `missing_ok` give. A name without a schema searches the path.
fn spaces(schema: Option<&str>, session: &dyn Session) -> Vec<u32> {
    match schema {
        Some(schema) => namespace_oid(schema, session).into_iter().collect(),
        None => path(session),
    }
}

/// The object of a kind with a name in the first schema of `spaces` that has one. An operator class or an operator family must have the access method `method`, which is 0 for the other kinds. A collation of the encoding of the database comes before a collation of any encoding in the same schema, as `lookup_collation` finds it.
fn find(kind: Named, spaces: &[u32], name: &str, method: u32) -> Option<&'static NamedRow> {
    let rows = builtin::named(kind);
    spaces.iter().find_map(|&ns| {
        let here = |encoding: i32| {
            rows.iter().find(|r| {
                r.namespace == ns && r.name == name && r.method == method && r.encoding == encoding
            })
        };
        if kind == Named::Collation { here(UTF8).or_else(|| here(-1)) } else { here(-1) }
    })
}

/// The checks of `namespace.c` for the kinds that have only built-in objects: true when the unqualified name finds the object.
pub(crate) fn visible(kind: Named, row: &NamedRow, session: &dyn Session) -> bool {
    find(kind, &path(session), row.name, row.method).is_some_and(|found| found.oid == row.oid)
}

/// The name, the schema and the access method of an operator class or an operator family, built in or of the catalog.
pub(crate) fn method_row(
    kind: Named,
    oid: u32,
    session: &dyn Session,
) -> Option<(String, u32, u32)> {
    if let Some(row) = builtin::named_by_oid(kind, oid) {
        return Some((row.name.to_string(), row.namespace, row.method));
    }
    let catalog = session.catalog()?;
    match kind {
        Named::Opclass => catalog.opclass(oid).map(|c| (c.name.clone(), c.namespace, c.method)),
        Named::Opfamily => catalog.opfamily(oid).map(|f| (f.name.clone(), f.namespace, f.method)),
        _ => None,
    }
}

/// `OpclassIsVisible` and `OpfamilyIsVisible` for an operator class or an operator family, built in or of the catalog: true when the first schema of the search path with an object of the name and the access method has this one.
pub(crate) fn method_visible(kind: Named, oid: u32, session: &dyn Session) -> bool {
    let Some((name, _, method)) = method_row(kind, oid, session) else { return false };
    let catalog = session.catalog();
    let first = path(session).into_iter().find_map(|ns| {
        find(kind, &[ns], &name, method).map(|row| row.oid).or_else(|| match kind {
            Named::Opclass => catalog?
                .opclasses()
                .find(|c| c.namespace == ns && c.name == name && c.method == method)
                .map(|c| c.oid),
            _ => catalog?
                .opfamilies()
                .find(|f| f.namespace == ns && f.name == name && f.method == method)
                .map(|f| f.oid),
        })
    });
    first == Some(oid)
}

/// The name and the schema of a collation or an object of text search, built in or of the catalog.
pub(crate) fn catalog_name(kind: Named, oid: u32, session: &dyn Session) -> Option<(String, u32)> {
    if let Some(row) = builtin::named_by_oid(kind, oid) {
        return Some((row.name.to_string(), row.namespace));
    }
    let catalog = session.catalog()?;
    let pair = |name: &str, namespace: u32| (name.to_string(), namespace);
    match kind {
        Named::Template => catalog.ts_template(oid).map(|t| pair(&t.name, t.namespace)),
        Named::Parser => catalog.ts_parser(oid).map(|p| pair(&p.name, p.namespace)),
        Named::Dictionary => catalog.ts_dictionary(oid).map(|d| pair(&d.name, d.namespace)),
        Named::Config => catalog.ts_config(oid).map(|c| pair(&c.name, c.namespace)),
        Named::Collation => catalog.collation(oid).map(|c| pair(&c.name, c.namespace)),
        _ => None,
    }
}

/// The OID of the collation or the object of text search with the name in the first schema of `spaces` that has one, built in or of the catalog.
fn catalog_find(kind: Named, spaces: &[u32], name: &str, session: &dyn Session) -> Option<u32> {
    let catalog = session.catalog();
    spaces.iter().find_map(|&ns| {
        find(kind, &[ns], name, 0).map(|row| row.oid).or_else(|| match kind {
            Named::Template => catalog?.ts_template_by_name(ns, name).map(|t| t.oid),
            Named::Parser => catalog?.ts_parser_by_name(ns, name).map(|p| p.oid),
            Named::Dictionary => catalog?.ts_dictionary_by_name(ns, name).map(|d| d.oid),
            Named::Config => catalog?.ts_config_by_name(ns, name).map(|c| c.oid),
            Named::Collation => catalog?.collation_by_name(ns, name).map(|c| c.oid),
            _ => None,
        })
    })
}

/// `check_default_text_search_config`: the value that `default_text_search_config` keeps for the configuration with the name in `text`. The value is the name with its schema, so a later change of `search_path` does not change the configuration. `None` when no configuration has the name.
///
/// # Errors
///
/// The errors of the input of `regconfig` that are not soft.
pub fn ts_config_setting(text: &str, session: &dyn Session) -> Result<Option<String>> {
    let Ok(oid) = input(RegKind::Config, text, session)? else { return Ok(None) };
    let Some((name, namespace)) = catalog_name(Named::Config, oid, session) else {
        return Ok(None);
    };
    Ok(Some(qualified(namespace_name(namespace, session), &name)))
}

/// `get_ts_parser_oid` of the name in a text, as `ts_parse` and `ts_token_type` find a parser by name.
///
/// # Errors
///
/// `42602` for a name that is not valid, `3F000` for a schema that does not exist, and `42704` for a parser that does not exist.
pub(crate) fn ts_parser_oid(text: &str, session: &dyn Session) -> Result<u32> {
    let names = qualified_name_list(text).map_err(type_error)?;
    let (schema, name) = deconstruct(&names, session)?;
    if let Some(schema) = schema
        && namespace_oid(schema, session).is_none()
    {
        return Err(Error::new(
            SqlState::UNDEFINED_SCHEMA,
            format!("schema \"{schema}\" does not exist"),
        ));
    }
    catalog_find(Named::Parser, &spaces(schema, session), name, session).ok_or_else(|| {
        Error::new(
            SqlState::UNDEFINED_OBJECT,
            format!("text search parser \"{}\" does not exist", name_list(&names)),
        )
    })
}

/// `CollationIsVisible`, `TSParserIsVisible`, `TSDictionaryIsVisible`, `TSTemplateIsVisible` and `TSConfigIsVisible`: true when the unqualified name finds the object. `None` when no object has the OID.
pub(crate) fn catalog_visible(kind: Named, oid: u32, session: &dyn Session) -> Option<bool> {
    let (name, _) = catalog_name(kind, oid, session)?;
    Some(catalog_find(kind, &path(session), &name, session) == Some(oid))
}

/// The text of `regcollationout`, `regconfigout` and `regdictionaryout`: the name, with the schema when the name is not visible.
pub(crate) fn catalog_text(kind: Named, oid: u32, session: &dyn Session) -> Option<String> {
    let (name, namespace) = catalog_name(kind, oid, session)?;
    let schema = if catalog_find(kind, &path(session), &name, session) == Some(oid) {
        None
    } else {
        namespace_name(namespace, session)
    };
    Some(qualified(schema, &name))
}

/// The OID of the relation with the name in the first schema of `spaces` that has one.
fn find_class(spaces: &[u32], name: &str, session: &dyn Session) -> Option<u32> {
    spaces.iter().find_map(|&ns| {
        find(Named::Class, &[ns], name, 0)
            .map(|row| row.oid)
            .or_else(|| session.catalog()?.relation_by_name(ns, name).map(|r| r.oid))
    })
}

/// The OID of the type with the name in the first schema of `spaces` that has one.
fn find_type(spaces: &[u32], name: &str, session: &dyn Session) -> Option<u32> {
    spaces.iter().find_map(|&ns| {
        builtin::type_by_name(ns, name)
            .map(|row| row.oid)
            .or_else(|| session.catalog()?.type_by_name(ns, name).map(|t| t.oid))
    })
}

/// The name and the schema of the relation with this OID.
pub(crate) fn class_name(oid: u32, session: &dyn Session) -> Option<(&str, u32)> {
    builtin::named_by_oid(Named::Class, oid)
        .map(|row| (row.name, row.namespace))
        .or_else(|| session.catalog()?.relation(oid).map(|r| (r.name.as_str(), r.namespace)))
}

/// The name and the schema of the type with this OID.
pub(crate) fn type_name(oid: u32, session: &dyn Session) -> Option<(&str, u32)> {
    builtin::type_by_oid(oid)
        .map(|row| (row.name, row.namespace))
        .or_else(|| session.catalog()?.type_by_oid(oid).map(|t| (t.name.as_str(), t.namespace)))
        .or_else(|| row_type_relation(oid, session).map(|r| (r.name.as_str(), r.namespace)))
}

/// The relation of the user whose row type is `ty`, which has the name and the schema of the relation.
fn row_type_relation(ty: u32, session: &dyn Session) -> Option<&rupg_catalog::Relation> {
    session.catalog()?.relations().find(|r| ty != 0 && r.row_type == ty)
}

/// `RelationIsVisible`, or `None` when no relation has the OID.
pub(crate) fn class_visible(oid: u32, session: &dyn Session) -> Option<bool> {
    let (name, _) = class_name(oid, session)?;
    Some(find_class(&temp_path(session), name, session) == Some(oid))
}

/// `TypeIsVisible`, or `None` when no type has the OID.
pub(crate) fn type_visible(oid: u32, session: &dyn Session) -> Option<bool> {
    let (name, _) = type_name(oid, session)?;
    Some(find_type(&temp_path(session), name, session) == Some(oid))
}

/// `FuncnameGetCandidates`: the functions with the name in the schemas, in the order of the schemas. A function with the same argument types as a function in a schema before it is hidden.
fn functions(spaces: &[u32], name: &str, nargs: Option<usize>) -> Vec<&'static ProcRow> {
    let mut found: Vec<&'static ProcRow> = Vec::new();
    for &ns in spaces {
        for proc in builtin::procs_named(name) {
            if proc.namespace == ns
                && nargs.is_none_or(|n| proc.argtypes.len() == n)
                && !found.iter().any(|f| f.argtypes == proc.argtypes)
            {
                found.push(proc);
            }
        }
    }
    found
}

/// `OpernameGetCandidates`: the operators with the name in the schemas, in the order of the schemas. An operator with the same argument types as an operator in a schema before it is hidden.
fn operators(spaces: &[u32], name: &str) -> Vec<&'static OperatorRow> {
    let mut found: Vec<&'static OperatorRow> = Vec::new();
    for &ns in spaces {
        for op in builtin::operators_named(name) {
            if op.namespace == ns && !found.iter().any(|f| (f.left, f.right) == (op.left, op.right))
            {
                found.push(op);
            }
        }
    }
    found
}

/// `FunctionIsVisible`.
pub(crate) fn function_visible(proc: &ProcRow, session: &dyn Session) -> bool {
    functions(&path(session), proc.name, Some(proc.argtypes.len()))
        .iter()
        .find(|f| f.argtypes == proc.argtypes)
        .is_some_and(|f| f.oid == proc.oid)
}

/// `OperatorIsVisible`.
pub(crate) fn operator_visible(op: &OperatorRow, session: &dyn Session) -> bool {
    operators(&path(session), op.name)
        .iter()
        .find(|f| (f.left, f.right) == (op.left, op.right))
        .is_some_and(|f| f.oid == op.oid)
}

/// `F_ARRAY_SUBSCRIPT_HANDLER`, the `typsubscript` of a true array type.
const ARRAY_SUBSCRIPT_HANDLER: u32 = 6179;
/// The bytes of the header of a `varlena` value, which a typmod of a string type counts.
const VARHDRSZ: i32 = 4;

/// `format_type_extended` with `FORMAT_TYPE_ALLOW_INVALID`: `-` for OID 0, or the name of a type, with its schema when the name alone does not find it, and with its typmod. A typmod of `None` is the form without `FORMAT_TYPE_TYPEMOD_GIVEN`.
///
/// # Errors
///
/// `0A000` for a typmod of `interval`, which the engine cannot show yet.
pub(crate) fn type_text(ty: u32, typmod: Option<i32>, session: &dyn Session) -> Result<String> {
    if ty == 0 {
        return Ok("-".to_string());
    }
    format_type(ty, typmod, false, session)
}

/// `format_type_be_qualified` with `FORMAT_TYPE_INVALID_AS_NULL`: the name of a type with its schema, also when the name is visible, or `None` when no type has the OID. A type with a special name of SQL, such as `integer`, has no schema.
pub(crate) fn type_text_qualified(ty: u32, session: &dyn Session) -> Option<String> {
    type_name(ty, session)?;
    format_type(ty, None, true, session).ok().filter(|text| !text.starts_with("???"))
}

/// `format_type_extended`: the name of a type, with its schema when `qualify` is true or when the name alone does not find it.
fn format_type(
    ty: u32,
    typmod: Option<i32>,
    qualify: bool,
    session: &dyn Session,
) -> Result<String> {
    // A true array type shows the name of its element type with `[]`, and a plain array type such as `oidvector` shows its own name.
    let element = match builtin::type_by_oid(ty) {
        Some(row) if row.elem != 0 && row.subscript == ARRAY_SUBSCRIPT_HANDLER => {
            (row.storage != b'p').then_some(row.elem)
        }
        Some(_) => None,
        None => match session.catalog().and_then(|c| c.type_by_oid(ty)) {
            Some(user) => (user.element != 0).then_some(user.element),
            None if row_type_relation(ty, session).is_some() => None,
            None => return Ok("???".to_string()),
        },
    };
    match element {
        Some(element) if type_name(element, session).is_none() => Ok("???[]".to_string()),
        Some(element) => Ok(format!("{}[]", base_text(element, typmod, qualify, session)?)),
        None => base_text(ty, typmod, qualify, session),
    }
}

/// The part of `format_type_extended` after the array test: the name of a type that is not a true array type.
fn base_text(ty: u32, typmod: Option<i32>, qualify: bool, session: &dyn Session) -> Result<String> {
    let given = typmod.is_some();
    let with = typmod.filter(|m| *m >= 0);
    let special = match (ty, with) {
        (oid::BIT, Some(m)) => Some(format!("bit({m})")),
        (oid::BIT, None) if !given => Some("bit".to_string()),
        (oid::BOOL, _) => Some("boolean".to_string()),
        (oid::BPCHAR, Some(m)) => Some(format!("character{}", length(m))),
        (oid::BPCHAR, None) if !given => Some("character".to_string()),
        (oid::FLOAT4, _) => Some("real".to_string()),
        (oid::FLOAT8, _) => Some("double precision".to_string()),
        (oid::INT2, _) => Some("smallint".to_string()),
        (oid::INT4, _) => Some("integer".to_string()),
        (oid::INT8, _) => Some("bigint".to_string()),
        (oid::NUMERIC, Some(m)) if m >= VARHDRSZ => {
            let bits = m - VARHDRSZ;
            let scale = ((bits & 0x7ff) ^ 1024) - 1024;
            Some(format!("numeric({},{scale})", (bits >> 16) & 0xffff))
        }
        (oid::NUMERIC, _) => Some("numeric".to_string()),
        (oid::INTERVAL, Some(m)) => Some(format!("interval{}", interval_typmod(m)?)),
        (oid::INTERVAL, None) => Some("interval".to_string()),
        (oid::TIME, m) => Some(format!("time{} without time zone", precision(m))),
        (oid::TIMETZ, m) => Some(format!("time{} with time zone", precision(m))),
        (oid::TIMESTAMP, m) => Some(format!("timestamp{} without time zone", precision(m))),
        (oid::TIMESTAMPTZ, m) => Some(format!("timestamp{} with time zone", precision(m))),
        (oid::VARBIT, m) => Some(format!("bit varying{}", precision(m))),
        (oid::VARCHAR, m) => Some(format!("character varying{}", m.map_or(String::new(), length))),
        (oid::JSON, _) => Some("json".to_string()),
        _ => None,
    };
    if let Some(text) = special {
        return Ok(text);
    }
    let Some((name, namespace)) = type_name(ty, session) else { return Ok("???".to_string()) };
    let schema = if !qualify && type_visible(ty, session) == Some(true) {
        None
    } else {
        namespace_name(namespace, session)
    };
    let text = qualified(schema, name);
    // The types with a `typmodout` function all have a special form, so the default form shows the typmod as a number.
    Ok(match with {
        Some(m) => format!("{text}({m})"),
        None => text,
    })
}

/// The typmod text of `anytime_typmodout` and `bittypmodout`.
fn precision(typmod: Option<i32>) -> String {
    typmod.map_or(String::new(), |m| format!("({m})"))
}

/// The typmod text of `intervaltypmodout`: the fields of the range, then the precision when the typmod has one.
fn interval_typmod(typmod: i32) -> Result<String> {
    const YEAR: i32 = 1 << 2;
    const MONTH: i32 = 1 << 1;
    const DAY: i32 = 1 << 3;
    const HOUR: i32 = 1 << 10;
    const MINUTE: i32 = 1 << 11;
    const SECOND: i32 = 1 << 12;
    let (precision, range) = types::typmod::interval_precision_range(typmod).unwrap_or_default();
    let fields = match range {
        YEAR => " year",
        MONTH => " month",
        DAY => " day",
        HOUR => " hour",
        MINUTE => " minute",
        SECOND => " second",
        r if r == YEAR | MONTH => " year to month",
        r if r == DAY | HOUR => " day to hour",
        r if r == DAY | HOUR | MINUTE => " day to minute",
        r if r == DAY | HOUR | MINUTE | SECOND => " day to second",
        r if r == HOUR | MINUTE => " hour to minute",
        r if r == HOUR | MINUTE | SECOND => " hour to second",
        r if r == MINUTE | SECOND => " minute to second",
        types::typmod::INTERVAL_FULL_RANGE => "",
        _ => {
            return Err(Error::new(
                SqlState::INTERNAL_ERROR,
                format!("invalid INTERVAL typmod: 0x{typmod:x}"),
            ));
        }
    };
    Ok(if precision == types::typmod::INTERVAL_FULL_PRECISION {
        fields.to_string()
    } else {
        format!("{fields}({precision})")
    })
}

/// The typmod text of `bpchartypmodout` and `varchartypmodout`.
fn length(typmod: i32) -> String {
    if typmod > VARHDRSZ { format!("({})", typmod - VARHDRSZ) } else { String::new() }
}

/// The text output of an OID alias type: the name of the object, with its schema when the name alone does not find it, the text of OID 0 for the kind, or the number for an OID that no object has.
pub(crate) fn output(kind: RegKind, oid: u32, session: &dyn Session, out: &mut Vec<u8>) {
    let text = if oid == 0 { None } else { name_of(kind, oid, session) };
    match text {
        Some(text) => out.extend_from_slice(text.as_bytes()),
        None => types::reg_out_oid(kind, oid, out),
    }
}

/// The text of the object with this OID, or `None` when no object has it.
pub(crate) fn name_of(kind: RegKind, oid: u32, session: &dyn Session) -> Option<String> {
    match kind {
        RegKind::Class => {
            let (name, namespace) = class_name(oid, session)?;
            let schema = if class_visible(oid, session)? {
                None
            } else {
                namespace_name(namespace, session)
            };
            Some(qualified(schema, name))
        }
        RegKind::Collation => catalog_text(Named::Collation, oid, session),
        RegKind::Config => catalog_text(Named::Config, oid, session),
        RegKind::Dictionary => catalog_text(Named::Dictionary, oid, session),
        RegKind::Namespace => namespace_name(oid, session).map(quote_identifier),
        RegKind::Role => rupg_pgcatalog::roles::current().name(oid).map(quote_identifier),
        RegKind::Database => match builtin::named_by_oid(Named::Database, oid) {
            Some(row) => Some(quote_identifier(row.name)),
            None => {
                session.databases().iter().find(|d| d.oid == oid).map(|d| quote_identifier(&d.name))
            }
        },
        RegKind::Type => {
            type_name(oid, session)?;
            Some(type_text(oid, None, session).ok()?)
        }
        RegKind::Proc => {
            let proc = builtin::proc_by_oid(oid)?;
            let only =
                matches!(functions(&path(session), proc.name, None)[..], [f] if f.oid == oid);
            let schema = if only { None } else { namespace_name(proc.namespace, session) };
            Some(qualified(schema, proc.name))
        }
        RegKind::Procedure => {
            let proc = builtin::proc_by_oid(oid)?;
            let schema = if function_visible(proc, session) {
                None
            } else {
                namespace_name(proc.namespace, session)
            };
            let args: Vec<_> = proc
                .argtypes
                .iter()
                .map(|&ty| type_text(ty, None, session).unwrap_or_else(|_| "???".to_string()))
                .collect();
            Some(format!("{}({})", qualified(schema, proc.name), args.join(",")))
        }
        RegKind::Oper => {
            let op = builtin::operator_by_oid(oid)?;
            let only = matches!(operators(&path(session), op.name)[..], [f] if f.oid == oid);
            match namespace_name(op.namespace, session) {
                Some(schema) if !only => Some(format!("{}.{}", quote_identifier(schema), op.name)),
                _ => Some(op.name.to_string()),
            }
        }
        RegKind::Operator => {
            let op = builtin::operator_by_oid(oid)?;
            let mut text = String::new();
            if !operator_visible(op, session)
                && let Some(schema) = namespace_name(op.namespace, session)
            {
                text.push_str(&quote_identifier(schema));
                text.push('.');
            }
            let arg =
                |ty: u32| if ty == 0 { Cow::Borrowed("NONE") } else { types::format_type(ty) };
            Some(format!("{text}{}({},{})", op.name, arg(op.left), arg(op.right)))
        }
    }
}

/// The text input of an OID alias type: a number, `-` for OID 0, or the name of an object.
///
/// # Errors
///
/// The outer error is an error for any caller, such as a name with too many dots or a syntax error in a type name. The inner error is an error of the input function that `to_regclass` and the other functions give as a null, such as a name that does not exist.
pub(crate) fn input(kind: RegKind, text: &str, session: &dyn Session) -> Soft<u32> {
    match types::reg_in(kind, text) {
        Ok(RegInput::Oid(oid)) => return Ok(Ok(oid)),
        Ok(RegInput::Name(_)) => {}
        Err(e) => return Ok(Err(type_error(e))),
    }
    match kind {
        RegKind::Type => {
            let env = SessionEnv(session);
            Ok(rupg_analyze::parse_type(text, &env)?.map(|(ty, _)| ty))
        }
        RegKind::Procedure => procedure_in(text, session),
        RegKind::Operator => operator_in(text, session),
        _ => {
            let names = match qualified_name_list(text) {
                Ok(names) => names,
                Err(e) => return Ok(Err(type_error(e))),
            };
            name_in(kind, text, &names, session)
        }
    }
}

/// The input of a kind whose name is a list of names.
fn name_in(kind: RegKind, text: &str, names: &[Cow<'_, str>], session: &dyn Session) -> Soft<u32> {
    let list = name_list(names);
    let simple = |named: &[(&str, u32)], state: SqlState, what: &str| match names {
        [name] => match named.iter().find(|n| n.0 == name) {
            Some(n) => Ok(Ok(n.1)),
            None => soft(state, format!("{what} \"{name}\" does not exist")),
        },
        _ => soft(SqlState::INVALID_NAME, "invalid name syntax"),
    };
    let rows = |named: Named| -> Vec<(&str, u32)> {
        builtin::named(named).iter().map(|r| (r.name, r.oid)).collect()
    };
    let mut schemas = rows(Named::Namespace);
    schemas.retain(|&(_, oid)| !schema_dropped(oid, session));
    if let Some(catalog) = session.catalog() {
        schemas.extend(catalog.schemas().map(|s| (s.name.as_str(), s.oid)));
    }
    let catalog_in = |named: Named, state: SqlState, message: String| {
        let (schema, name) = deconstruct(names, session)?;
        match catalog_find(named, &spaces(schema, session), name, session) {
            Some(oid) => Ok(Ok(oid)),
            None => soft(state, message),
        }
    };
    match kind {
        RegKind::Class => class_in(names, session, false),
        RegKind::Namespace => simple(&schemas, SqlState::UNDEFINED_SCHEMA, "schema"),
        RegKind::Role => {
            let roles = rupg_pgcatalog::roles::current();
            simple(&roles.all().collect::<Vec<_>>(), SqlState::UNDEFINED_OBJECT, "role")
        }
        RegKind::Database => {
            let created = session.databases();
            let mut databases = rows(Named::Database);
            databases.extend(created.iter().map(|d| (d.name.as_str(), d.oid)));
            simple(&databases, SqlState::UNDEFINED_OBJECT, "database")
        }
        RegKind::Collation => catalog_in(
            Named::Collation,
            SqlState::UNDEFINED_OBJECT,
            format!("collation \"{list}\" for encoding \"{UTF8_NAME}\" does not exist"),
        ),
        RegKind::Config => catalog_in(
            Named::Config,
            SqlState::UNDEFINED_OBJECT,
            format!("text search configuration \"{list}\" does not exist"),
        ),
        RegKind::Dictionary => catalog_in(
            Named::Dictionary,
            SqlState::UNDEFINED_OBJECT,
            format!("text search dictionary \"{list}\" does not exist"),
        ),
        RegKind::Proc => {
            let (schema, name) = deconstruct(names, session)?;
            match functions(&spaces(schema, session), name, None)[..] {
                [] => soft(
                    SqlState::UNDEFINED_FUNCTION,
                    format!("function \"{text}\" does not exist"),
                ),
                [proc] => Ok(Ok(proc.oid)),
                _ => soft(
                    SqlState::AMBIGUOUS_FUNCTION,
                    format!("more than one function named \"{text}\""),
                ),
            }
        }
        RegKind::Oper => {
            let (schema, name) = deconstruct(names, session)?;
            match operators(&spaces(schema, session), name)[..] {
                [] => {
                    soft(SqlState::UNDEFINED_FUNCTION, format!("operator does not exist: {text}"))
                }
                [op] => Ok(Ok(op.oid)),
                _ => soft(
                    SqlState::AMBIGUOUS_FUNCTION,
                    format!("more than one operator named {text}"),
                ),
            }
        }
        RegKind::Type | RegKind::Procedure | RegKind::Operator => {
            Err(Error::internal("a type name read as a list of names"))
        }
    }
}

/// `makeRangeVarFromNameList` and `RangeVarGetRelid`: the relation of a list of names. With `strict`, as the cast from `text` gives, a relation or a schema that does not exist is a hard error with the message of `RangeVarGetRelid`.
fn class_in(names: &[Cow<'_, str>], session: &dyn Session, strict: bool) -> Soft<u32> {
    let (schema, name) = match names {
        [name] => (None, name),
        [schema, name] => (Some(schema), name),
        [database, schema, name] => {
            if database != session.database() {
                return Err(Error::new(
                    SqlState::FEATURE_NOT_SUPPORTED,
                    format!(
                        "cross-database references are not implemented: \"{database}.{schema}.{name}\""
                    ),
                ));
            }
            (Some(schema), name)
        }
        _ => {
            return Err(Error::new(
                SqlState::SYNTAX_ERROR,
                format!("improper relation name (too many dotted names): {}", name_list(names)),
            ));
        }
    };
    let spaces = match schema {
        Some(schema) => match namespace_oid(schema, session) {
            Some(ns) => vec![ns],
            None if strict => {
                return Err(Error::new(
                    SqlState::UNDEFINED_SCHEMA,
                    format!("schema \"{schema}\" does not exist"),
                ));
            }
            None => Vec::new(),
        },
        None => temp_path(session),
    };
    if let Some(oid) = find_class(&spaces, name, session) {
        return Ok(Ok(oid));
    }
    let message = if strict {
        match schema {
            Some(schema) => format!("relation \"{schema}.{name}\" does not exist"),
            None => format!("relation \"{name}\" does not exist"),
        }
    } else {
        format!("relation \"{}\" does not exist", name_list(names))
    };
    let error = Error::new(SqlState::UNDEFINED_TABLE, message);
    if strict { Err(error) } else { Ok(Err(error)) }
}

/// The bytes that `scanner_isspace` takes as white space.
fn is_space(b: u8) -> bool {
    matches!(b, b' ' | b'\t' | b'\n' | b'\r' | 0x0b | 0x0c)
}

/// The error `22P02` of `parseNameAndArgTypes`.
fn bad_syntax<T>(message: &str) -> Soft<T> {
    soft(SqlState::INVALID_TEXT_REPRESENTATION, message)
}

/// The name and the argument types of a function or an operator.
type NameAndArgs<'a> = (Vec<Cow<'a, str>>, Vec<u32>);

/// `parseNameAndArgTypes`: the name and the argument types of a string such as `abs(integer)`. With `allow_none`, `NONE` in any case is the missing argument of a prefix operator, which is 0.
fn name_and_args<'a>(
    text: &'a str,
    allow_none: bool,
    session: &dyn Session,
) -> Soft<NameAndArgs<'a>> {
    let b = text.as_bytes();
    let mut in_quote = false;
    let mut open = None;
    for (i, &c) in b.iter().enumerate() {
        if c == b'"' {
            in_quote = !in_quote;
        } else if c == b'(' && !in_quote {
            open = Some(i);
            break;
        }
    }
    let Some(open) = open else { return bad_syntax("expected a left parenthesis") };
    let names = match qualified_name_list(&text[..open]) {
        Ok(names) => names,
        Err(e) => return Ok(Err(type_error(e))),
    };
    let rest = &text[open + 1..];
    let rb = rest.as_bytes();
    // The backward scan of the C code stops at the first byte after the parenthesis, so that byte is the end when all the bytes after it are white space.
    let close = (1..rb.len()).rev().find(|&i| !is_space(rb[i])).unwrap_or(0);
    if rb.get(close) != Some(&b')') {
        return bad_syntax("expected a right parenthesis");
    }
    let args = &rest[..close];
    let ab = args.as_bytes();
    let env = SessionEnv(session);
    let mut types = Vec::new();
    let mut at = 0;
    let mut had_comma = false;
    loop {
        while at < ab.len() && is_space(ab[at]) {
            at += 1;
        }
        if at == ab.len() {
            if had_comma {
                return bad_syntax("expected a type name");
            }
            break;
        }
        let start = at;
        let mut in_quote = false;
        let mut depth = 0i32;
        while at < ab.len() {
            match ab[at] {
                b'"' => in_quote = !in_quote,
                b',' if !in_quote && depth == 0 => break,
                b'(' | b'[' if !in_quote => depth += 1,
                b')' | b']' if !in_quote => depth -= 1,
                _ => {}
            }
            at += 1;
        }
        if in_quote || depth != 0 {
            return bad_syntax("improper type name");
        }
        let name = args[start..at].trim_end_matches(|c: char| u8::try_from(c).is_ok_and(is_space));
        had_comma = at < ab.len();
        if had_comma {
            at += 1;
        }
        let ty = if allow_none && name.eq_ignore_ascii_case("none") {
            0
        } else {
            match rupg_analyze::parse_type(name, &env)? {
                Ok((ty, _)) => ty,
                Err(e) => return Ok(Err(e)),
            }
        };
        if types.len() >= FUNC_MAX_ARGS {
            return soft(SqlState::TOO_MANY_ARGUMENTS, "too many arguments");
        }
        types.push(ty);
    }
    Ok(Ok((names, types)))
}

/// `regprocedurein`.
fn procedure_in(text: &str, session: &dyn Session) -> Soft<u32> {
    let (names, args) = match name_and_args(text, false, session)? {
        Ok(found) => found,
        Err(e) => return Ok(Err(e)),
    };
    let (schema, name) = deconstruct(&names, session)?;
    match functions(&spaces(schema, session), name, Some(args.len()))
        .iter()
        .find(|f| f.argtypes == args.as_slice())
    {
        Some(proc) => Ok(Ok(proc.oid)),
        None => soft(SqlState::UNDEFINED_FUNCTION, format!("function \"{text}\" does not exist")),
    }
}

/// `regoperatorin`.
fn operator_in(text: &str, session: &dyn Session) -> Soft<u32> {
    let (names, args) = match name_and_args(text, true, session)? {
        Ok(found) => found,
        Err(e) => return Ok(Err(e)),
    };
    let [left, right] = args[..] else {
        let error = if args.len() == 1 {
            Error::new(SqlState::UNDEFINED_PARAMETER, "missing argument")
                .with_hint("Use NONE to denote the missing argument of a unary operator.")
        } else {
            Error::new(SqlState::TOO_MANY_ARGUMENTS, "too many arguments")
                .with_hint("Provide two argument types for operator.")
        };
        return Ok(Err(error));
    };
    let (schema, name) = deconstruct(&names, session)?;
    let found = spaces(schema, session).into_iter().find_map(|ns| {
        builtin::operators_named(name)
            .find(|op| op.namespace == ns && op.left == left && op.right == right)
    });
    match found {
        Some(op) => Ok(Ok(op.oid)),
        None => soft(SqlState::UNDEFINED_FUNCTION, format!("operator does not exist: {text}")),
    }
}

/// The text argument of a function.
pub(crate) fn text_arg(args: &[Value]) -> Result<&str> {
    args.first().and_then(Value::as_str).ok_or_else(bad_value)
}

/// `to_regclass` and the other functions: the input of the type, or null for a soft error.
fn to_reg(kind: RegKind, call: &Call<'_>, args: &[Value]) -> Result<Value> {
    Ok(input(kind, text_arg(args)?, call.session)?.map_or(Value::Null, Value::Oid))
}

/// `to_regtypemod`: the typmod of a type name, or null for a type that does not exist.
fn to_regtypemod(call: &Call<'_>, args: &[Value]) -> Result<Value> {
    let env = SessionEnv(call.session);
    let found = rupg_analyze::parse_type(text_arg(args)?, &env)?;
    Ok(found.map_or(Value::Null, |(_, typmod)| Value::Int4(typmod)))
}

/// `pg_input_validate`: the soft error of the input of the first argument as the type that the second argument names, or `None` for a good value. A type name that is not good and a hard error of the input are errors. For a domain, the value must also pass the check constraints.
fn input_error(call: &Call<'_>, args: &[Value]) -> Result<Option<Error>> {
    let text = text_arg(args)?;
    let name = args.get(1).and_then(Value::as_str).ok_or_else(bad_value)?;
    let env = SessionEnv(call.session);
    let (ty, typmod) = rupg_analyze::parse_type(name, &env)??;
    let value = match crate::io::input_soft(ty, text, typmod, call.session)? {
        Ok(value) => value,
        Err(error) => return Ok(Some(error)),
    };
    let element = builtin::type_by_oid(ty).map_or(0, |row| row.elem);
    if builtin::domain_by_oid(ty).is_some() || builtin::domain_by_oid(element).is_some() {
        return call.session.input_checks(ty, &value);
    }
    Ok(None)
}

/// `pg_input_is_valid`: true when the text is a good value of the type.
fn input_is_valid(call: &Call<'_>, args: &[Value]) -> Result<Value> {
    Ok(Value::Bool(input_error(call, args)?.is_none()))
}

/// `pg_input_error_info`: the message, the detail, the hint and the SQLSTATE of the soft error of the input, or a row of nulls for a good value.
fn input_error_info(call: &Call<'_>, args: &[Value]) -> Result<Value> {
    let values = match input_error(call, args)? {
        None => vec![Value::Null; 4],
        Some(error) => {
            let text = |s: Option<&str>| s.map_or(Value::Null, Value::text);
            vec![
                Value::text(error.message()),
                text(error.detail()),
                text(error.hint()),
                Value::text(error.state().as_str()),
            ]
        }
    };
    Ok(Value::Record(Box::new(Record::new(vec![oid::TEXT; 4], values))))
}

/// The relation of a list of names, as `RangeVarGetRelid` finds it for `text_regclass`.
///
/// # Errors
///
/// `42P01` for a relation that does not exist, `3F000` for a schema that does not exist, and the errors of a list of names that is not correct.
pub(crate) fn relation_oid(names: &[Cow<'_, str>], session: &dyn Session) -> Result<u32> {
    class_in(names, session, true)?
}

/// `text_regclass`, the cast from `text` to `regclass`. A name that does not exist is an error, as `RangeVarGetRelid` gives it.
fn text_regclass(call: &Call<'_>, args: &[Value]) -> Result<Value> {
    let names = qualified_name_list(text_arg(args)?).map_err(type_error)?;
    Ok(Value::Oid(relation_oid(&names, call.session)?))
}

/// The kernel of a function of this module by its `prosrc`.
pub(crate) fn by_src(src: &str) -> Option<Kernel> {
    let kernel: Kernel = match src {
        "to_regclass" => |c, a| to_reg(RegKind::Class, c, a),
        "to_regcollation" => |c, a| to_reg(RegKind::Collation, c, a),
        "to_regdatabase" => |c, a| to_reg(RegKind::Database, c, a),
        "to_regnamespace" => |c, a| to_reg(RegKind::Namespace, c, a),
        "to_regoper" => |c, a| to_reg(RegKind::Oper, c, a),
        "to_regoperator" => |c, a| to_reg(RegKind::Operator, c, a),
        "to_regproc" => |c, a| to_reg(RegKind::Proc, c, a),
        "to_regprocedure" => |c, a| to_reg(RegKind::Procedure, c, a),
        "to_regrole" => |c, a| to_reg(RegKind::Role, c, a),
        "to_regtype" => |c, a| to_reg(RegKind::Type, c, a),
        "to_regtypemod" => to_regtypemod,
        "pg_input_is_valid" => input_is_valid,
        "pg_input_error_info" => input_error_info,
        "text_regclass" => text_regclass,
        _ => return None,
    };
    Some(kernel)
}