polyc-query 2026.9.6

//! The closed function surface for projected Query sessions (POLY-371).
//!
//! A caller-written statement reaches a session whose function registries
//! hold only the names in [`ALLOWED`]. Everything else fails planning —
//! refused names by decision, unclassified names by construction — and the
//! refusal reaches the caller as the collapsed invalid-query error.
//!
//! ## The admission rule
//!
//! A function is [`FunctionClass::Allowed`] only when both hold:
//!
//! 1. Its output for one row is at most a constant times the bytes of its
//!    inputs for that row. No argument may select an output size, a
//!    repetition count, a padding length, or a replacement that multiplies
//!    matches.
//! 2. Its CPU cost is at most near-linear in its input size.
//!
//! [`REFUSED`] lists every name — canonical and alias — the rule refuses,
//! with the reason. The enumeration test walks the engine's own default
//! registries at the pinned version and fails on any name in neither
//! table, so an upgrade forces a decision rather than silently admitting
//! a new function.
//!
//! ## Residual structure
//!
//! Some planner-desugared forms construct their implementation directly
//! rather than resolving a name in the registry — `OVERLAY` syntax, array
//! literals, field access. The forms that map to input-shaped
//! implementations stay working; the one that maps to a refused function
//! (`Expr::Overlay`) is refused by the statement gate, where the scan sees
//! the AST node the planner would consume.

use datafusion::execution::SessionStateDefaults;
use datafusion::logical_expr::{AggregateUDF, ScalarUDF, WindowUDF};
use std::collections::BTreeSet;
use std::sync::{Arc, OnceLock};

/// Which registry a function's name lives in — the classification axis
/// the census enumerates.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum FunctionKind {
    /// A scalar function from `datafusion-functions`.
    Scalar,
    /// A scalar function from `datafusion-functions-nested` (array, map,
    /// struct, and JSON-adjacent helpers).
    Nested,
    /// An aggregate function.
    Aggregate,
    /// A window function.
    Window,
    /// A lambda-taking higher-order function.
    HigherOrder,
    /// A table function (`FROM generate_series(...)`, `FROM range(...)`).
    Table,
    /// A JSON function registered by `polyc-query-json`.
    Json,
}

/// One allowed canonical name with the reason it qualifies.
pub(crate) struct AllowedFunction {
    /// The canonical name the engine reports for the implementation.
    pub(crate) name: &'static str,
    /// Which registry the name belongs to.
    pub(crate) kind: FunctionKind,
    /// Why the admission rule admits it — review data the enumeration
    /// test keeps honest, not a runtime input.
    #[allow(dead_code)]
    pub(crate) reason: &'static str,
}

/// One refused name group — a canonical name and every alias it answers
/// to — with the reason the admission rule refuses it.
#[allow(dead_code)]
pub(crate) struct RefusedFunction {
    /// Every spelling that must fail: canonical name plus all aliases.
    pub(crate) names: &'static [&'static str],
    /// Which registry the name belongs to.
    pub(crate) kind: FunctionKind,
    /// Why the admission rule refuses it.
    pub(crate) reason: &'static str,
}

/// The allowed list: every function the projected session registers.
///
/// Aliases need no row of their own — registering an implementation
/// registers every alias it declares, and the enumeration test accounts
/// for them.
pub(crate) static ALLOWED: &[AllowedFunction] = &[
    // Scalars whose output is a fixed-size value or a constant-factor
    // function of the input bytes.
    AllowedFunction {
        name: "abs",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "acos",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "acosh",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "arrow_cast",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "arrow_field",
        kind: FunctionKind::Scalar,
        reason: "fixed-size type metadata",
    },
    AllowedFunction {
        name: "arrow_metadata",
        kind: FunctionKind::Scalar,
        reason: "fixed-size metadata",
    },
    AllowedFunction {
        name: "arrow_try_cast",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "arrow_typeof",
        kind: FunctionKind::Scalar,
        reason: "fixed-size type name",
    },
    AllowedFunction {
        name: "ascii",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "asin",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "asinh",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "atan",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "atan2",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "atanh",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "bit_length",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "btrim",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "cast_to_type",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "cbrt",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "ceil",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "character_length",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "chr",
        kind: FunctionKind::Scalar,
        reason: "single-codepoint output",
    },
    AllowedFunction {
        name: "coalesce",
        kind: FunctionKind::Scalar,
        reason: "returns one argument",
    },
    AllowedFunction {
        name: "concat",
        kind: FunctionKind::Scalar,
        reason: "output is the sum of its inputs; bounded by the emit-reference caps",
    },
    AllowedFunction {
        name: "concat_ws",
        kind: FunctionKind::Scalar,
        reason: "output is the sum of its inputs; bounded by the emit-reference caps",
    },
    AllowedFunction {
        name: "contains",
        kind: FunctionKind::Scalar,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "cos",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "cosh",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "cot",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "current_date",
        kind: FunctionKind::Scalar,
        reason: "fixed-size result",
    },
    AllowedFunction {
        name: "current_time",
        kind: FunctionKind::Scalar,
        reason: "fixed-size result",
    },
    AllowedFunction {
        name: "date_bin",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "date_part",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "date_trunc",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "decode",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "degrees",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "digest",
        kind: FunctionKind::Scalar,
        reason: "fixed-size hash",
    },
    AllowedFunction {
        name: "encode",
        kind: FunctionKind::Scalar,
        reason: "output bounded by a constant factor of input",
    },
    AllowedFunction {
        name: "ends_with",
        kind: FunctionKind::Scalar,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "exp",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "factorial",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "find_in_set",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "floor",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "from_unixtime",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "gcd",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "get_field",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "greatest",
        kind: FunctionKind::Scalar,
        reason: "returns one argument",
    },
    AllowedFunction {
        name: "initcap",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "isnan",
        kind: FunctionKind::Scalar,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "iszero",
        kind: FunctionKind::Scalar,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "lcm",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "least",
        kind: FunctionKind::Scalar,
        reason: "returns one argument",
    },
    AllowedFunction {
        name: "left",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "ln",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "log",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "log10",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "log2",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "lower",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "ltrim",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "make_date",
        kind: FunctionKind::Scalar,
        reason: "fixed-size date",
    },
    AllowedFunction {
        name: "make_time",
        kind: FunctionKind::Scalar,
        reason: "fixed-size time",
    },
    AllowedFunction {
        name: "md5",
        kind: FunctionKind::Scalar,
        reason: "fixed-size hash",
    },
    AllowedFunction {
        name: "named_struct",
        kind: FunctionKind::Scalar,
        reason: "output is the sum of its inputs",
    },
    AllowedFunction {
        name: "nanvl",
        kind: FunctionKind::Scalar,
        reason: "returns one argument",
    },
    AllowedFunction {
        name: "now",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "nullif",
        kind: FunctionKind::Scalar,
        reason: "returns one argument",
    },
    AllowedFunction {
        name: "nvl",
        kind: FunctionKind::Scalar,
        reason: "returns one argument",
    },
    AllowedFunction {
        name: "nvl2",
        kind: FunctionKind::Scalar,
        reason: "returns one argument",
    },
    AllowedFunction {
        name: "octet_length",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "pi",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "power",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "radians",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "random",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "regexp_count",
        kind: FunctionKind::Scalar,
        reason: "numeric result; linear-time matcher",
    },
    AllowedFunction {
        name: "regexp_instr",
        kind: FunctionKind::Scalar,
        reason: "numeric result; linear-time matcher",
    },
    AllowedFunction {
        name: "regexp_like",
        kind: FunctionKind::Scalar,
        reason: "boolean result; linear-time matcher",
    },
    AllowedFunction {
        name: "regexp_match",
        kind: FunctionKind::Scalar,
        reason: "matched groups bounded by input; linear-time matcher",
    },
    AllowedFunction {
        name: "reverse",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "right",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "round",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "rtrim",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "sha224",
        kind: FunctionKind::Scalar,
        reason: "fixed-size hash",
    },
    AllowedFunction {
        name: "sha256",
        kind: FunctionKind::Scalar,
        reason: "fixed-size hash",
    },
    AllowedFunction {
        name: "sha384",
        kind: FunctionKind::Scalar,
        reason: "fixed-size hash",
    },
    AllowedFunction {
        name: "sha512",
        kind: FunctionKind::Scalar,
        reason: "fixed-size hash",
    },
    AllowedFunction {
        name: "signum",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "sin",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "sinh",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "split_part",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "sqrt",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "starts_with",
        kind: FunctionKind::Scalar,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "strpos",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "struct",
        kind: FunctionKind::Scalar,
        reason: "output is the sum of its inputs",
    },
    AllowedFunction {
        name: "substr",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "substr_index",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "tan",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "tanh",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "to_char",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input plus format literal",
    },
    AllowedFunction {
        name: "to_date",
        kind: FunctionKind::Scalar,
        reason: "fixed-size date",
    },
    AllowedFunction {
        name: "to_hex",
        kind: FunctionKind::Scalar,
        reason: "output bounded by a constant factor of input",
    },
    AllowedFunction {
        name: "to_local_time",
        kind: FunctionKind::Scalar,
        reason: "fixed-size time",
    },
    AllowedFunction {
        name: "to_time",
        kind: FunctionKind::Scalar,
        reason: "fixed-size time",
    },
    AllowedFunction {
        name: "to_timestamp",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "to_timestamp_micros",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "to_timestamp_millis",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "to_timestamp_nanos",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "to_timestamp_seconds",
        kind: FunctionKind::Scalar,
        reason: "fixed-size timestamp",
    },
    AllowedFunction {
        name: "to_unixtime",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "translate",
        kind: FunctionKind::Scalar,
        reason: "one-to-one character map; output bounded by input",
    },
    AllowedFunction {
        name: "trunc",
        kind: FunctionKind::Scalar,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "try_cast_to_type",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "union_extract",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "union_tag",
        kind: FunctionKind::Scalar,
        reason: "fixed-size tag",
    },
    AllowedFunction {
        name: "upper",
        kind: FunctionKind::Scalar,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "uuid",
        kind: FunctionKind::Scalar,
        reason: "fixed-size identifier",
    },
    AllowedFunction {
        name: "version",
        kind: FunctionKind::Scalar,
        reason: "fixed-size string",
    },
    // Nested functions: every admitted one is input-shaped — it
    // transforms, selects, or combines elements already in its inputs.
    AllowedFunction {
        name: "array_add",
        kind: FunctionKind::Nested,
        reason: "element-wise result bounded by inputs",
    },
    AllowedFunction {
        name: "array_any_value",
        kind: FunctionKind::Nested,
        reason: "returns one element",
    },
    AllowedFunction {
        name: "array_append",
        kind: FunctionKind::Nested,
        reason: "output is the sum of its inputs",
    },
    AllowedFunction {
        name: "array_avg",
        kind: FunctionKind::Nested,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "array_compact",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_concat",
        kind: FunctionKind::Nested,
        reason: "output is the sum of its inputs",
    },
    AllowedFunction {
        name: "array_dims",
        kind: FunctionKind::Nested,
        reason: "fixed-size dimensions",
    },
    AllowedFunction {
        name: "array_distance",
        kind: FunctionKind::Nested,
        reason: "numeric result; linear in input length",
    },
    AllowedFunction {
        name: "array_distinct",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_element",
        kind: FunctionKind::Nested,
        reason: "returns one element",
    },
    AllowedFunction {
        name: "array_except",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_has",
        kind: FunctionKind::Nested,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "array_has_all",
        kind: FunctionKind::Nested,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "array_has_any",
        kind: FunctionKind::Nested,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "array_intersect",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_length",
        kind: FunctionKind::Nested,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "array_max",
        kind: FunctionKind::Nested,
        reason: "returns one element",
    },
    AllowedFunction {
        name: "array_min",
        kind: FunctionKind::Nested,
        reason: "returns one element",
    },
    AllowedFunction {
        name: "array_ndims",
        kind: FunctionKind::Nested,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "array_normalize",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_pop_back",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_pop_front",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_position",
        kind: FunctionKind::Nested,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "array_positions",
        kind: FunctionKind::Nested,
        reason: "output bounded by input length",
    },
    AllowedFunction {
        name: "array_prepend",
        kind: FunctionKind::Nested,
        reason: "output is the sum of its inputs",
    },
    AllowedFunction {
        name: "array_product",
        kind: FunctionKind::Nested,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "array_remove",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_remove_all",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_remove_n",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_replace",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_replace_all",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_replace_n",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_reverse",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_scale",
        kind: FunctionKind::Nested,
        reason: "element-wise scale; output bounded by input",
    },
    AllowedFunction {
        name: "array_slice",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_sort",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_subtract",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "array_sum",
        kind: FunctionKind::Nested,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "array_to_string",
        kind: FunctionKind::Nested,
        reason: "output bounded by input plus separator",
    },
    AllowedFunction {
        name: "array_union",
        kind: FunctionKind::Nested,
        reason: "output bounded by the sum of its inputs",
    },
    AllowedFunction {
        name: "arrays_zip",
        kind: FunctionKind::Nested,
        reason: "output bounded by the sum of its inputs",
    },
    AllowedFunction {
        name: "cardinality",
        kind: FunctionKind::Nested,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "cosine_distance",
        kind: FunctionKind::Nested,
        reason: "numeric result; linear in input length",
    },
    AllowedFunction {
        name: "empty",
        kind: FunctionKind::Nested,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "flatten",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "inner_product",
        kind: FunctionKind::Nested,
        reason: "numeric result; linear in input length",
    },
    AllowedFunction {
        name: "make_array",
        kind: FunctionKind::Nested,
        reason: "output is the sum of its inputs",
    },
    AllowedFunction {
        name: "map",
        kind: FunctionKind::Nested,
        reason: "output is the sum of its inputs",
    },
    AllowedFunction {
        name: "map_entries",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "map_extract",
        kind: FunctionKind::Nested,
        reason: "returns one value",
    },
    AllowedFunction {
        name: "map_keys",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "map_values",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "string_to_array",
        kind: FunctionKind::Nested,
        reason: "output bounded by input",
    },
    // Aggregates whose state is metered by the memory pool — they
    // accumulate input rows, never a caller-chosen size.
    AllowedFunction {
        name: "any_value",
        kind: FunctionKind::Aggregate,
        reason: "pool-metered state",
    },
    AllowedFunction {
        name: "approx_distinct",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size sketch state, pool-metered",
    },
    AllowedFunction {
        name: "approx_median",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size digest state, pool-metered",
    },
    AllowedFunction {
        name: "array_agg",
        kind: FunctionKind::Aggregate,
        reason: "pool-metered accumulation",
    },
    AllowedFunction {
        name: "avg",
        kind: FunctionKind::Aggregate,
        reason: "pool-metered accumulation",
    },
    AllowedFunction {
        name: "bit_and",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "bit_or",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "bit_xor",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "bool_and",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "bool_or",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "corr",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "count",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "covar_pop",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "covar_samp",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "first_value",
        kind: FunctionKind::Aggregate,
        reason: "returns one element",
    },
    AllowedFunction {
        name: "grouping",
        kind: FunctionKind::Aggregate,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "last_value",
        kind: FunctionKind::Aggregate,
        reason: "returns one element",
    },
    AllowedFunction {
        name: "max",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "median",
        kind: FunctionKind::Aggregate,
        reason: "pool-metered accumulation",
    },
    AllowedFunction {
        name: "min",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "nth_value",
        kind: FunctionKind::Aggregate,
        reason: "returns one element; the offset is an index, not a size",
    },
    AllowedFunction {
        name: "percentile_cont",
        kind: FunctionKind::Aggregate,
        reason: "pool-metered accumulation",
    },
    AllowedFunction {
        name: "regr_avgx",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "regr_avgy",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "regr_count",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "regr_intercept",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "regr_r2",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "regr_slope",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "regr_sxx",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "regr_sxy",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "regr_syy",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "stddev",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "stddev_pop",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "string_agg",
        kind: FunctionKind::Aggregate,
        reason: "pool-metered accumulation",
    },
    AllowedFunction {
        name: "sum",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "var",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    AllowedFunction {
        name: "var_pop",
        kind: FunctionKind::Aggregate,
        reason: "fixed-size state",
    },
    // Window functions all run inside the pool-metered window operator;
    // `ntile`, `lag`, `lead`, and `nth_value` take offsets and counts,
    // never byte sizes.
    AllowedFunction {
        name: "cume_dist",
        kind: FunctionKind::Window,
        reason: "pool-metered window state",
    },
    AllowedFunction {
        name: "dense_rank",
        kind: FunctionKind::Window,
        reason: "pool-metered window state",
    },
    AllowedFunction {
        name: "first_value",
        kind: FunctionKind::Window,
        reason: "returns one element",
    },
    AllowedFunction {
        name: "lag",
        kind: FunctionKind::Window,
        reason: "offset is an index, not a size",
    },
    AllowedFunction {
        name: "last_value",
        kind: FunctionKind::Window,
        reason: "returns one element",
    },
    AllowedFunction {
        name: "lead",
        kind: FunctionKind::Window,
        reason: "offset is an index, not a size",
    },
    AllowedFunction {
        name: "nth_value",
        kind: FunctionKind::Window,
        reason: "offset is an index, not a size",
    },
    AllowedFunction {
        name: "ntile",
        kind: FunctionKind::Window,
        reason: "bucket count shapes row labels, not bytes",
    },
    AllowedFunction {
        name: "percent_rank",
        kind: FunctionKind::Window,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "rank",
        kind: FunctionKind::Window,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "row_number",
        kind: FunctionKind::Window,
        reason: "numeric result",
    },
    // The JSON surface `polyc-query-json` registers.
    AllowedFunction {
        name: "json_as_text",
        kind: FunctionKind::Json,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "json_contains",
        kind: FunctionKind::Json,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "json_from_scalar",
        kind: FunctionKind::Json,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "json_get",
        kind: FunctionKind::Json,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "json_get_array",
        kind: FunctionKind::Json,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "json_get_bool",
        kind: FunctionKind::Json,
        reason: "boolean result",
    },
    AllowedFunction {
        name: "json_get_float",
        kind: FunctionKind::Json,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "json_get_int",
        kind: FunctionKind::Json,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "json_get_json",
        kind: FunctionKind::Json,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "json_get_str",
        kind: FunctionKind::Json,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "json_length",
        kind: FunctionKind::Json,
        reason: "numeric result",
    },
    AllowedFunction {
        name: "json_object_keys",
        kind: FunctionKind::Json,
        reason: "output bounded by input",
    },
    AllowedFunction {
        name: "json_union_to_text",
        kind: FunctionKind::Json,
        reason: "output bounded by input",
    },
];

/// The refused list: every name the admission rule refuses, with every
/// alias spelled out so the per-alias refusal test can iterate them.
/// Read by the enumeration and refusal tests; outside them it is the
/// documented refusal record the module's claim cites.
#[allow(dead_code)]
pub(crate) static REFUSED: &[RefusedFunction] = &[
    RefusedFunction {
        names: &["repeat"],
        kind: FunctionKind::Scalar,
        reason: "the count argument chooses the output length — one call can ask for ~2 GiB",
    },
    RefusedFunction {
        names: &["lpad"],
        kind: FunctionKind::Scalar,
        reason: "the length argument chooses the padded output size",
    },
    RefusedFunction {
        names: &["rpad"],
        kind: FunctionKind::Scalar,
        reason: "the length argument chooses the padded output size",
    },
    RefusedFunction {
        names: &["replace"],
        kind: FunctionKind::Scalar,
        reason: "the replacement argument can multiply every match in the input",
    },
    RefusedFunction {
        names: &["regexp_replace"],
        kind: FunctionKind::Scalar,
        reason: "the replacement argument can multiply every match in the input",
    },
    RefusedFunction {
        names: &["overlay"],
        kind: FunctionKind::Scalar,
        reason: "the position arguments choose the covered span and output length",
    },
    RefusedFunction {
        names: &["levenshtein"],
        kind: FunctionKind::Scalar,
        reason: "quadratic CPU in the input size inside one poll",
    },
    RefusedFunction {
        names: &["file_row_index", "input_file_name", "with_metadata"],
        kind: FunctionKind::Scalar,
        reason: "scan internals, not table data — and a file name can leak an object key",
    },
    RefusedFunction {
        names: &["array_repeat", "list_repeat"],
        kind: FunctionKind::Nested,
        reason: "the count argument chooses the output size",
    },
    RefusedFunction {
        names: &["array_resize", "list_resize"],
        kind: FunctionKind::Nested,
        reason: "the size argument chooses the output size",
    },
    RefusedFunction {
        names: &["generate_series"],
        kind: FunctionKind::Nested,
        reason: "the arguments choose how many elements the array holds",
    },
    RefusedFunction {
        names: &["range"],
        kind: FunctionKind::Nested,
        reason: "the arguments choose how many elements the array holds",
    },
    RefusedFunction {
        names: &["approx_percentile_cont"],
        kind: FunctionKind::Aggregate,
        reason: "the size argument chooses the digest's eager allocation",
    },
    RefusedFunction {
        names: &["approx_percentile_cont_with_weight"],
        kind: FunctionKind::Aggregate,
        reason: "the size argument chooses the digest's eager allocation",
    },
    RefusedFunction {
        names: &[
            "array_any_match",
            "any_match",
            "list_any_match",
            "array_filter",
            "list_filter",
            "array_first",
            "list_first",
            "array_transform",
            "list_transform",
        ],
        kind: FunctionKind::HigherOrder,
        reason: "a lambda evaluates per element and escapes the per-expression bound",
    },
    RefusedFunction {
        names: &["generate_series"],
        kind: FunctionKind::Table,
        reason: "the arguments choose the row count",
    },
    RefusedFunction {
        names: &["range"],
        kind: FunctionKind::Table,
        reason: "the arguments choose the row count",
    },
];

/// The canonical names the composed session registers as scalar UDFs —
/// the allowed scalar and nested rows.
fn allowed_scalar_names() -> impl Iterator<Item = &'static str> {
    ALLOWED
        .iter()
        .filter(|row| matches!(row.kind, FunctionKind::Scalar | FunctionKind::Nested))
        .map(|row| row.name)
}

/// The scalar UDFs to register: the engine's own defaults filtered to
/// [`ALLOWED`]'s scalar and nested canonical names.
pub(crate) fn closed_scalar_functions() -> Vec<Arc<ScalarUDF>> {
    let names: BTreeSet<&str> = allowed_scalar_names().collect();
    SessionStateDefaults::default_scalar_functions()
        .into_iter()
        .filter(|function| names.contains(function.name()))
        .collect()
}

/// The aggregate UDFs to register, filtered the same way.
pub(crate) fn closed_aggregate_functions() -> Vec<Arc<AggregateUDF>> {
    let names: BTreeSet<&str> = ALLOWED
        .iter()
        .filter(|row| row.kind == FunctionKind::Aggregate)
        .map(|row| row.name)
        .collect();
    SessionStateDefaults::default_aggregate_functions()
        .into_iter()
        .filter(|function| names.contains(function.name()))
        .collect()
}

/// The window UDFs to register, filtered the same way.
pub(crate) fn closed_window_functions() -> Vec<Arc<WindowUDF>> {
    let names: BTreeSet<&str> = ALLOWED
        .iter()
        .filter(|row| row.kind == FunctionKind::Window)
        .map(|row| row.name)
        .collect();
    SessionStateDefaults::default_window_functions()
        .into_iter()
        .filter(|function| names.contains(function.name()))
        .collect()
}

/// Every name — canonical or alias — the closed surface answers to,
/// including the JSON surface. The statement gate refuses a function call
/// whose resolved name is absent here, so the refusal carries a reason
/// instead of surfacing as an engine lookup error.
pub(crate) fn allowed_names() -> &'static BTreeSet<String> {
    static NAMES: OnceLock<BTreeSet<String>> = OnceLock::new();
    NAMES.get_or_init(|| {
        let mut names = BTreeSet::new();
        for function in closed_scalar_functions() {
            names.insert(function.name().to_owned());
            names.extend(function.aliases().iter().map(ToString::to_string));
        }
        for function in closed_aggregate_functions() {
            names.insert(function.name().to_owned());
            names.extend(function.aliases().iter().map(ToString::to_string));
        }
        for function in closed_window_functions() {
            names.insert(function.name().to_owned());
            names.extend(function.aliases().iter().map(ToString::to_string));
        }
        for row in ALLOWED {
            if row.kind == FunctionKind::Json {
                names.insert(row.name.to_owned());
            }
        }
        names
    })
}

/// Every refused spelling, for the enumeration and refusal tests.
#[allow(dead_code)]
pub(crate) fn refused_names() -> &'static BTreeSet<&'static str> {
    static NAMES: OnceLock<BTreeSet<&'static str>> = OnceLock::new();
    NAMES.get_or_init(|| {
        REFUSED
            .iter()
            .flat_map(|row| row.names.iter().copied())
            .collect()
    })
}

#[cfg(test)]
mod tests {
    #![allow(clippy::pedantic, clippy::nursery, missing_docs)]

    use super::*;

    /// Every name the engine's default feature set offers — canonical or
    /// alias, scalar, nested, higher-order, aggregate, window, or table —
    /// is classified by exactly one of the two tables. A `DataFusion`
    /// upgrade that adds a function fails here with the unclassified names
    /// listed, forcing the admission decision this module exists for.
    #[test]
    fn every_default_name_is_classified() {
        let allowed = allowed_names();
        let refused = refused_names();
        let mut unclassified: Vec<String> = Vec::new();
        let mut both: Vec<String> = Vec::new();
        let mut classify = |name: &str| {
            let in_allowed = allowed.contains(name);
            let in_refused = refused.contains(name);
            match (in_allowed, in_refused) {
                (true, true) => both.push(name.to_owned()),
                (false, false) => unclassified.push(name.to_owned()),
                _ => {}
            }
        };
        for function in SessionStateDefaults::default_scalar_functions() {
            classify(function.name());
            for alias in function.aliases() {
                classify(alias);
            }
        }
        for function in SessionStateDefaults::default_higher_order_functions() {
            classify(function.name());
            for alias in function.aliases() {
                classify(alias);
            }
        }
        for function in SessionStateDefaults::default_aggregate_functions() {
            classify(function.name());
            for alias in function.aliases() {
                classify(alias);
            }
        }
        for function in SessionStateDefaults::default_window_functions() {
            classify(function.name());
            for alias in function.aliases() {
                classify(alias);
            }
        }
        for function in SessionStateDefaults::default_table_functions() {
            classify(function.name());
        }
        assert!(
            both.is_empty() && unclassified.is_empty(),
            "classified in both tables: {both:?}; classified in neither: {unclassified:?}"
        );
    }

    /// A name that exists in neither table is not in [`allowed_names`] —
    /// the surface fails closed on anything new.
    #[test]
    fn unclassified_names_stay_unregistered() {
        assert!(!allowed_names().contains("a_function_nobody_listed"));
        assert!(!refused_names().contains("a_function_nobody_listed"));
    }

    /// Every allowed canonical name actually resolves to an engine
    /// implementation — a typo in [`ALLOWED`] cannot shrink the registered
    /// set silently.
    #[test]
    fn every_allowed_canonical_resolves() {
        let scalars: BTreeSet<String> = closed_scalar_functions()
            .iter()
            .map(|function| function.name().to_owned())
            .collect();
        let aggregates: BTreeSet<String> = closed_aggregate_functions()
            .iter()
            .map(|function| function.name().to_owned())
            .collect();
        let windows: BTreeSet<String> = closed_window_functions()
            .iter()
            .map(|function| function.name().to_owned())
            .collect();
        for row in ALLOWED {
            let resolved = match row.kind {
                FunctionKind::Scalar | FunctionKind::Nested => scalars.contains(row.name),
                FunctionKind::Aggregate => aggregates.contains(row.name),
                FunctionKind::Window => windows.contains(row.name),
                // The JSON surface registers separately and is checked by
                // the registry-equality test on the composed session.
                FunctionKind::Json
                // Higher-order and table names are never allowed rows.
                | FunctionKind::HigherOrder
                | FunctionKind::Table => true,
            };
            assert!(
                resolved,
                "allowed name {:?} resolved to no implementation",
                row.name
            );
        }
    }

    /// The registration helpers produce exactly the allowed canonical
    /// names — no default slips through the filter and no allowed name is
    /// dropped.
    #[test]
    fn closed_lists_match_the_allowed_table() {
        let scalar_names: BTreeSet<String> = closed_scalar_functions()
            .iter()
            .map(|function| function.name().to_owned())
            .collect();
        let expected_scalars: BTreeSet<String> = ALLOWED
            .iter()
            .filter(|row| matches!(row.kind, FunctionKind::Scalar | FunctionKind::Nested))
            .map(|row| row.name.to_owned())
            .collect();
        assert_eq!(scalar_names, expected_scalars);
    }
}