//! The statements that define objects: `CREATE TABLE`, `CREATE INDEX`, `CREATE SCHEMA`, `CREATE VIEW`, `CREATE RULE`, `CREATE TRIGGER`, `CREATE DOMAIN`, `CREATE FUNCTION`, `COMMENT`, `DROP`, `ALTER TABLE`, `ALTER SCHEMA ... OWNER TO`, `CREATE TEXT SEARCH`, `ALTER TEXT SEARCH CONFIGURATION`, `CREATE OPERATOR` and `TRUNCATE`. They run as `transformCreateStmt`, `DefineRelation`, `DefineIndex`, `CreateSchemaCommand`, `DefineView`, `DefineRule`, `CreateTrigger`, `DefineDomain`, `CreateFunction`, `CommentObject`, `ExecDropStmt`, `AlterTable`, `AlterSchemaOwner`, the functions of `tsearchcmds.c`, `DefineOperator` and `ExecuteTruncate` run them, and they change a [`Catalog`]. `CREATE FUNCTION` makes a function in SQL. [`lock_table`] runs the checks of `LOCK TABLE` and takes its locks. It finds its relations by name as these statements do.
mod aggregate;
mod alter;
mod alter_column;
mod alter_constraint;
mod alter_function;
mod alter_inherit;
mod alter_type;
mod attmap;
mod cluster;
mod comment;
mod copy;
mod create_table_as;
mod domain;
mod drop;
mod drop_constraint;
mod drop_function;
mod enum_label;
mod enumtype;
mod fkey;
mod fkey_partition;
mod function;
mod generated;
mod grant;
mod identity;
mod index;
mod lock;
mod matview;
mod opclass;
mod operator;
mod owned;
mod partition;
mod partition_alter;
mod partition_attach;
mod partition_index;
mod rangetype;
mod reindex;
mod relation_options;
mod reloptions;
mod rename;
mod row_type;
mod rule;
mod schema_elements;
mod sequence;
mod set_schema;
mod statistics;
mod table;
#[cfg(test)]
mod tests;
mod trigger;
mod truncate;
mod tsearch;
mod type_change;
mod using_index;
mod vacuum;
mod validate;
mod view;
pub use copy::{
CopyRelation, CopyTarget, attnums, copy_filter, copy_insert, copy_relation, copy_target,
rewrite_insert,
};
pub(crate) use drop::quote_identifier;
pub use function::{
Body, REGRESS_SYMBOLS, Script, SqlBody, find_library, function_body, function_script,
function_sqlbody,
};
pub use grant::acl_roles;
pub(crate) use lock::relkind;
pub use lock::{Allowed, Take, lock_table};
pub use owned::{acl_dependents, owned_objects};
pub use reindex::{ReindexOptions, reindex_options};
pub use reloptions::tablespace_options;
pub use rule::{RuleTree, rule_tree};
pub use schema_elements::SchemaElements;
pub(crate) use table::check_attribute_type;
pub use trigger::trigger_when;
pub use vacuum::vacuum_relations;
use std::cell::RefCell;
use rupg_catalog::{Catalog, ObjRef, PG_CLASS, PG_NAMESPACE, PG_PROC, PG_TYPE, RelKind};
use rupg_common::{Error, Result, SqlState};
use rupg_pgcatalog::builtin::{self, ClassRow, Named, NamedRow, Owned};
use rupg_sql::nodes::{CreateSchemaStmt, GrantTargetType, Node, ObjectType, RangeVar};
use rupg_types::{Value, oid};
use crate::coerce::AtOpt;
use crate::expr::{Expr, ExprKind};
use crate::typename::place;
use crate::{Analyzer, Env, PG_CATALOG_NAMESPACE, Params};
/// The OID of the schema `pg_toast`, where a user cannot create a relation.
const PG_TOAST_NAMESPACE: u32 = 99;
/// `BOOTSTRAP_SUPERUSERID`, the owner of the temporary schemas.
const BOOTSTRAP_SUPERUSER: u32 = 10;
/// A message of a statement that defines an object. The client gets the messages before the result.
#[derive(Clone, Debug, PartialEq)]
pub enum Message {
/// A `WARNING`, such as a precision that is too large.
Warning(Error),
/// A `NOTICE`, such as an object that `IF NOT EXISTS` skips.
Notice(Error),
}
/// The result of [`define`].
#[derive(Debug)]
pub struct Defined {
/// The messages in order. The client gets them also when the statement fails.
pub messages: Vec<Message>,
/// The result of the statement. After an error the caller discards the changes of the catalog, but the OIDs that the statement used stay used, as in PostgreSQL.
pub result: Result<()>,
/// The tables that the statement read with a sequential scan, once for each scan, as `pgstat_count_heap_scan` counts them: the build of an index, the check of new constraints by `ALTER TABLE` and the check of a new foreign key. The scans count also when the statement fails.
pub scans: Vec<u32>,
/// The tables of `TRUNCATE`, whose rows the caller removes after the checks pass.
pub truncated: Vec<u32>,
/// The expressions of `USING` of `ALTER COLUMN ... TYPE`, which give the new values of the columns when the caller writes the rows of the table again.
pub transforms: Vec<Transform>,
/// The defaults that the new columns of `ADD COLUMN` had when the subcommand ran, for each new column whose default a later subcommand of the same statement changed. The caller writes the rows of the table again with these defaults.
pub fills: Vec<Fill>,
/// The table of `CREATE TABLE AS` or the materialized view of `CREATE MATERIALIZED VIEW` or `REFRESH MATERIALIZED VIEW`, which the caller fills with the rows of the query, or `None`. It is `None` also for `WITH NO DATA` and for a table that exists with `IF NOT EXISTS`.
pub filled: Option<u32>,
/// The views whose query reads an object that the statement renamed. The catalog keeps the text of a view, so the caller writes the text again from the query that the old names give.
pub views: Vec<u32>,
/// The partitions of `ATTACH PARTITION` whose rows the caller checks against their new partition constraints after the statement, as `QueuePartitionConstraintValidation` queues them.
pub checks: Vec<PartitionCheck>,
/// The partitions of `DETACH PARTITION` whose keys a foreign key refers to through the parent. Before the statement the caller checks that no row of a referencing table has a key of the partition, as `ATDetachCheckNoForeignKeyRefs` checks it, with the catalog before the statement.
pub detached: Vec<u32>,
/// The roles of `DROP OWNED`. The caller removes the grants of membership that these roles made, as `shdepDropOwned` drops the rows of `pg_auth_members` with the role as the grantor.
pub grantors: Vec<u32>,
/// The elements of `CREATE SCHEMA`, which the caller runs after it keeps the new schema, or `None`.
pub schema: Option<SchemaElements>,
/// The check of `PreventInTransactionBlock` that the statement needs after it found its relations, such as `REINDEX` of a partitioned table, or `None`. The caller runs the check before it keeps the changes, and after an error the messages of the statement go.
pub top_level: Option<TopLevel>,
/// The new index of `CREATE INDEX CONCURRENTLY` on a table that is not temporary, or `None`. The first transaction of `DefineIndex` commits the index before the build, so after a failure of the build the caller keeps the index with `indisvalid` and `indisready` false.
pub concurrent: Option<u32>,
}
/// A statement that cannot run inside a transaction block or in a function, as `PreventInTransactionBlock` checks it.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct TopLevel {
/// The name of the statement in the error, such as `REINDEX TABLE`.
pub statement: &'static str,
/// The context of the error, such as the partitioned table of `REINDEX`.
pub context: String,
}
/// A partition whose rows must satisfy its partition constraint, one entry of `tab->partition_constraint`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct PartitionCheck {
/// The OID of the partition. The caller checks the rows of each of its leaf partitions.
pub table: u32,
/// True for the default partition, whose constraint is smaller after a new partition, with the error of `validateDefaultPartitionConstraint` for a row that violates it.
pub default: bool,
}
/// The expression of `USING` of `ALTER COLUMN ... TYPE` for one column, as `ATPrepAlterColumnType` keeps it in `tab->newvals`.
#[derive(Clone, Debug)]
pub struct Transform {
/// The OID of the table.
pub table: u32,
/// The name of the column.
pub column: String,
/// The expression before its transform. It refers to the columns of the table by their names, and it gives the old values of the columns.
pub expr: Node,
}
/// The default of a new column of `ALTER TABLE ... ADD COLUMN`, as `ATExecAddColumn` keeps it in `tab->newvals`, or the new expression of a generated column of `SET EXPRESSION`. The rows of the table get this default, also when a later subcommand of the statement sets or drops the default of the column.
#[derive(Clone, Debug)]
pub struct Fill {
/// The OID of the table.
pub table: u32,
/// The number of the column.
pub column: i16,
/// The text of the default expression, or `None` for a column with no default, which gets a null.
pub default: Option<String>,
}
/// True for a statement that [`define`] runs.
pub fn is_definition(stmt: &Node) -> bool {
matches!(
stmt,
Node::CreateStmt(_)
| Node::CreateTableAsStmt(_)
| Node::IndexStmt(_)
| Node::CreateSchemaStmt(_)
| Node::CreateDomainStmt(_)
| Node::CompositeTypeStmt(_)
| Node::CreateEnumStmt(_)
| Node::AlterEnumStmt(_)
| Node::CreateRangeStmt(_)
| Node::ViewStmt(_)
| Node::RuleStmt(_)
| Node::CreateTrigStmt(_)
| Node::CreateFunctionStmt(_)
| Node::AlterFunctionStmt(_)
| Node::CommentStmt(_)
| Node::CreateSeqStmt(_)
| Node::AlterSeqStmt(_)
| Node::AlterTSConfigurationStmt(_)
| Node::AlterOperatorStmt(_)
| Node::CreateOpFamilyStmt(_)
| Node::CreateOpClassStmt(_)
| Node::AlterOpFamilyStmt(_)
| Node::TruncateStmt(_)
| Node::RefreshMatViewStmt(_)
| Node::CreateStatsStmt(_)
| Node::AlterStatsStmt(_)
| Node::DropOwnedStmt(_)
| Node::ReassignOwnedStmt(_)
| Node::AlterDefaultPrivilegesStmt(_)
) || matches!(stmt, Node::DefineStmt(define) if tsearch::is_ts_kind(define.kind) || matches!(define.kind, ObjectType::OBJECT_OPERATOR | ObjectType::OBJECT_AGGREGATE))
|| matches!(stmt, Node::DropStmt(drop) if is_drop_kind(drop.removeType))
|| matches!(stmt, Node::AlterTableStmt(alter) if is_alter_kind(alter.objtype) || alter.objtype == ObjectType::OBJECT_TYPE)
|| matches!(stmt, Node::AlterOwnerStmt(alter) if matches!(alter.objectType, ObjectType::OBJECT_SCHEMA | ObjectType::OBJECT_OPERATOR | ObjectType::OBJECT_STATISTIC_EXT) || is_opcf_kind(alter.objectType) || type_change::is_type_kind(alter.objectType) || alter_function::is_function_kind(alter.objectType))
|| matches!(stmt, Node::RenameStmt(rename) if rename::is_rename_kind(rename.renameType))
|| matches!(stmt, Node::AlterObjectSchemaStmt(alter) if is_alter_kind(alter.objectType) || matches!(alter.objectType, ObjectType::OBJECT_OPERATOR | ObjectType::OBJECT_STATISTIC_EXT) || is_opcf_kind(alter.objectType) || type_change::is_type_kind(alter.objectType) || alter_function::is_function_kind(alter.objectType))
|| matches!(stmt, Node::SelectStmt(select) if select.intoClause.is_some())
|| matches!(stmt, Node::GrantStmt(grant) if grant.targtype != GrantTargetType::ACL_TARGET_DEFAULTS)
}
/// True for an operator class and an operator family.
fn is_opcf_kind(kind: ObjectType) -> bool {
matches!(kind, ObjectType::OBJECT_OPCLASS | ObjectType::OBJECT_OPFAMILY)
}
/// True for the kinds of relations of `ALTER TABLE` and of the other statements that change a relation. `ALTER TYPE` of the attributes of a composite type is an `AlterTableStmt` too.
fn is_alter_kind(kind: ObjectType) -> bool {
matches!(
kind,
ObjectType::OBJECT_TABLE
| ObjectType::OBJECT_VIEW
| ObjectType::OBJECT_INDEX
| ObjectType::OBJECT_SEQUENCE
| ObjectType::OBJECT_MATVIEW
| ObjectType::OBJECT_FOREIGN_TABLE
)
}
/// True for the kinds of objects that `DROP` can remove.
fn is_drop_kind(kind: ObjectType) -> bool {
matches!(
kind,
ObjectType::OBJECT_TABLE
| ObjectType::OBJECT_VIEW
| ObjectType::OBJECT_MATVIEW
| ObjectType::OBJECT_INDEX
| ObjectType::OBJECT_SEQUENCE
| ObjectType::OBJECT_SCHEMA
| ObjectType::OBJECT_TYPE
| ObjectType::OBJECT_DOMAIN
| ObjectType::OBJECT_RULE
| ObjectType::OBJECT_TRIGGER
| ObjectType::OBJECT_FUNCTION
| ObjectType::OBJECT_PROCEDURE
| ObjectType::OBJECT_ROUTINE
| ObjectType::OBJECT_AGGREGATE
| ObjectType::OBJECT_TSPARSER
| ObjectType::OBJECT_TSDICTIONARY
| ObjectType::OBJECT_TSTEMPLATE
| ObjectType::OBJECT_TSCONFIGURATION
| ObjectType::OBJECT_OPERATOR
| ObjectType::OBJECT_OPCLASS
| ObjectType::OBJECT_OPFAMILY
| ObjectType::OBJECT_STATISTIC_EXT
)
}
/// Runs a statement that defines an object, for the role `user`, and adds the object to the catalog. The statement is the `stmt` of a `RawStmt`, and `text` is its text, which the catalog keeps for a view and for a function.
pub fn define(stmt: &Node, text: &str, env: &dyn Env, catalog: &mut Catalog, user: u32) -> Defined {
let schemas = SchemaEnv::new(env, catalog);
let mut definer = Definer {
an: Analyzer::new(&schemas, &Params::default()),
catalog,
user,
messages: Vec::new(),
scans: Vec::new(),
truncated: Vec::new(),
transforms: Vec::new(),
added: Vec::new(),
views: Vec::new(),
checks: Vec::new(),
detached: Vec::new(),
grantors: Vec::new(),
schema: None,
top_level: None,
concurrent: None,
};
let mut filled = None;
let result = match stmt {
Node::CreateStmt(create) => definer.create_table(create),
Node::CreateTableAsStmt(ctas) => {
definer.create_table_as(ctas, text).map(|table| filled = table)
}
Node::SelectStmt(select) => definer.select_into(select).map(|table| filled = table),
Node::RefreshMatViewStmt(stmt) => {
definer.refresh_matview(stmt).map(|matview| filled = matview)
}
Node::IndexStmt(index) => definer.create_index(index, false),
Node::CreateSchemaStmt(schema) => definer.create_schema(schema, text),
Node::ViewStmt(view) => definer.create_view(view, text),
Node::RuleStmt(rule) => definer.create_rule(rule, text),
Node::CreateTrigStmt(trigger) => definer.create_trigger(trigger),
Node::CreateDomainStmt(domain) => definer.create_domain(domain).map(|_| ()),
Node::CompositeTypeStmt(stmt) => definer.create_composite_type(stmt),
Node::CreateEnumStmt(stmt) => definer.create_enum(stmt),
Node::AlterEnumStmt(stmt) => definer.alter_enum(stmt),
Node::CreateRangeStmt(stmt) => definer.create_range(stmt),
Node::CreateFunctionStmt(function) => definer.create_function(function, text).map(|_| ()),
Node::AlterFunctionStmt(alter) => definer.alter_function(alter),
Node::CommentStmt(comment) => definer.comment(comment),
Node::DropStmt(drop) => definer.drop(drop),
Node::AlterTableStmt(alter) => definer.alter_table(alter),
Node::AlterOwnerStmt(alter) => definer.alter_owner(alter),
Node::RenameStmt(rename) => definer.rename(rename),
Node::AlterObjectSchemaStmt(alter) if type_change::is_type_kind(alter.objectType) => {
definer.alter_type_schema(alter)
}
Node::AlterObjectSchemaStmt(alter)
if alter_function::is_function_kind(alter.objectType) =>
{
definer.alter_function_schema(alter)
}
Node::AlterObjectSchemaStmt(alter) if alter.objectType == ObjectType::OBJECT_OPERATOR => {
definer.alter_operator_schema(alter)
}
Node::AlterObjectSchemaStmt(alter) if is_opcf_kind(alter.objectType) => {
definer.alter_opcf_schema(alter)
}
Node::AlterObjectSchemaStmt(alter)
if alter.objectType == ObjectType::OBJECT_STATISTIC_EXT =>
{
definer.alter_stats_schema(alter)
}
Node::AlterObjectSchemaStmt(alter) => definer.alter_relation_schema(alter),
Node::CreateStatsStmt(create) => definer.create_statistics(create),
Node::AlterStatsStmt(alter) => definer.alter_statistics(alter),
Node::CreateSeqStmt(create) => definer.create_sequence(create),
Node::AlterSeqStmt(alter) => definer.alter_sequence(alter),
Node::DefineStmt(define) if tsearch::is_ts_kind(define.kind) => definer.define_ts(define),
Node::DefineStmt(define) if define.kind == ObjectType::OBJECT_OPERATOR => {
definer.create_operator(define)
}
Node::DefineStmt(define) if define.kind == ObjectType::OBJECT_AGGREGATE => {
definer.create_aggregate(define)
}
Node::AlterOperatorStmt(alter) => definer.alter_operator(alter),
Node::CreateOpFamilyStmt(create) => definer.create_opfamily(create),
Node::CreateOpClassStmt(create) => definer.create_opclass(create),
Node::AlterOpFamilyStmt(alter) => definer.alter_opfamily(alter),
Node::AlterTSConfigurationStmt(alter) => definer.alter_ts_config(alter),
Node::TruncateStmt(truncate) => definer.truncate(truncate),
Node::GrantStmt(grant) => definer.grant(grant),
Node::DropOwnedStmt(stmt) => definer.drop_owned(stmt),
Node::AlterDefaultPrivilegesStmt(stmt) => definer.alter_default_privileges(stmt),
Node::ReassignOwnedStmt(stmt) => definer.reassign_owned(stmt),
Node::ReindexStmt(stmt) => definer.reindex(stmt),
_ => {
Err(Error::new(SqlState::FEATURE_NOT_SUPPORTED, "this statement is not supported yet"))
}
};
definer.flush();
let catalog = &*definer.catalog;
let fills = definer
.added
.into_iter()
.filter(|fill| {
let column = catalog.relation(fill.table).and_then(|r| r.column(fill.column));
let live = column.is_some_and(|c| !c.dropped);
// A stored generated column computes its values again from the expression that it has now.
let generated = column.is_some_and(|c| c.generated != 0);
let now = catalog.default_of(fill.table, fill.column).map(|d| &d.expr);
live && (generated || now != fill.default.as_ref())
})
.collect();
Defined {
messages: definer.messages,
result,
scans: definer.scans,
truncated: definer.truncated,
transforms: definer.transforms,
fills,
filled,
views: definer.views,
checks: definer.checks,
detached: definer.detached,
grantors: definer.grantors,
schema: definer.schema,
top_level: definer.top_level,
concurrent: definer.concurrent,
}
}
/// The session with the schemas of the catalog, which the analyzer finds by name.
struct SchemaEnv<'e> {
env: &'e dyn Env,
schemas: Vec<(String, u32)>,
/// The relations that the statement added before the last [`Env::command_counter_increment`], which the session does not have: the schema, the name and the OID.
relations: RefCell<Vec<(u32, String, u32)>>,
}
impl<'e> SchemaEnv<'e> {
fn new(env: &'e dyn Env, catalog: &Catalog) -> SchemaEnv<'e> {
let schemas = catalog.schemas().map(|s| (s.name.clone(), s.oid)).collect();
SchemaEnv { env, schemas, relations: RefCell::new(Vec::new()) }
}
/// The OID of a relation that the statement added, for the input `text` of `regclass`, as `regclassin` finds it after `CommandCounterIncrement`.
fn new_class(&self, text: &str) -> Option<u32> {
let names = rupg_types::qualified_name_list(text).ok()?;
let (spaces, name) = match names.as_slice() {
[name] => (self.search_path(), name),
[schema, name] => (vec![self.namespace(schema)?], name),
_ => return None,
};
let relations = self.relations.borrow();
spaces.iter().find_map(|&ns| {
relations.iter().find(|(n, rel, _)| *n == ns && rel == name).map(|(_, _, oid)| *oid)
})
}
}
impl Env for SchemaEnv<'_> {
fn search_path(&self) -> Vec<u32> {
self.env.search_path()
}
fn namespace(&self, name: &str) -> Option<u32> {
self.schemas
.iter()
.find(|(n, _)| n == name)
.map(|(_, oid)| *oid)
.or_else(|| self.env.namespace(name))
}
fn database(&self) -> String {
self.env.database()
}
fn input(&self, ty: u32, text: &str, typmod: i32) -> Result<Value> {
match self.env.input(ty, text, typmod) {
Err(error) if ty == oid::REGCLASS && error.state() == SqlState::UNDEFINED_TABLE => {
self.new_class(text).map(Value::Oid).ok_or(error)
}
value => value,
}
}
fn catalog(&self) -> Option<&Catalog> {
self.env.catalog()
}
fn allow_system_table_mods(&self) -> bool {
self.env.allow_system_table_mods()
}
fn check_function_bodies(&self) -> bool {
self.env.check_function_bodies()
}
fn sequence_row(&self, file: u32) -> Option<rupg_catalog::SequenceData> {
self.env.sequence_row(file)
}
fn deparse(&self, text: &str, relation: u32) -> Result<String> {
self.env.deparse(text, relation)
}
fn temp_number(&self) -> Option<i32> {
self.env.temp_number()
}
fn check_function(&self, function: &rupg_catalog::Function) -> Result<()> {
self.env.check_function(function)
}
fn function_setting(
&self,
set: &rupg_sql::nodes::VariableSetStmt,
) -> Result<(String, Option<String>)> {
self.env.function_setting(set)
}
fn push_function_settings(&self, config: &[String]) -> Result<()> {
self.env.push_function_settings(config)
}
fn pop_function_settings(&self) {
self.env.pop_function_settings();
}
fn command_counter_increment(&self, catalog: &Catalog) {
let session = self.env.catalog();
*self.relations.borrow_mut() = catalog
.relations()
.filter(|r| session.is_none_or(|s| s.relation(r.oid).is_none()))
.map(|r| (r.namespace, r.name.clone(), r.oid))
.collect();
self.env.command_counter_increment(catalog);
}
fn evaluate(&self, expr: &Expr) -> Result<Value> {
self.env.evaluate(expr)
}
fn compare(&self, ty: u32, a: &Value, b: &Value) -> Result<std::cmp::Ordering> {
self.env.compare(ty, a, b)
}
fn deparse_constant(&self, text: &str) -> Result<String> {
self.env.deparse_constant(text)
}
}
/// A relation that a name gives.
#[derive(Clone, Copy, Debug)]
enum Found {
/// A relation of the catalog, with its OID.
User(u32),
/// A relation of the built-in catalog.
Builtin(&'static ClassRow),
}
/// The state of one statement that defines an object.
pub(crate) struct Definer<'a, 'c> {
an: Analyzer<'a>,
catalog: &'c mut Catalog,
user: u32,
messages: Vec<Message>,
/// The tables of [`Defined::scans`].
scans: Vec<u32>,
/// The tables of [`Defined::truncated`].
truncated: Vec<u32>,
/// The expressions of [`Defined::transforms`].
transforms: Vec<Transform>,
/// The defaults of the new columns, from which [`Defined::fills`] keeps the changed ones.
added: Vec<Fill>,
/// The views of [`Defined::views`].
views: Vec<u32>,
/// The partitions of [`Defined::checks`].
checks: Vec<PartitionCheck>,
/// The partitions of [`Defined::detached`].
detached: Vec<u32>,
/// The roles of [`Defined::grantors`].
grantors: Vec<u32>,
/// The elements of [`Defined::schema`].
schema: Option<SchemaElements>,
/// The check of [`Defined::top_level`].
top_level: Option<TopLevel>,
/// The index of [`Defined::concurrent`].
concurrent: Option<u32>,
}
/// An error for a part of a statement that rupg does not support yet.
fn not_yet(what: &str, at: Option<usize>) -> Error {
Error::new(SqlState::FEATURE_NOT_SUPPORTED, format!("{what} is not supported yet")).at_opt(at)
}
/// The name of a relation as the query wrote it, with the schema and the database.
fn qualified(rv: &RangeVar) -> String {
[rv.catalogname.as_deref(), rv.schemaname.as_deref(), rv.relname.as_deref()]
.into_iter()
.flatten()
.collect::<Vec<_>>()
.join(".")
}
/// The name of the relation of a `RangeVar`.
fn relname(rv: &RangeVar) -> &str {
rv.relname.as_deref().unwrap_or_default()
}
/// `contain_mutable_functions_after_planning`: true when the expression calls a function that is not immutable. The planner inlines `textanycat` and `anytextcat`, so a call of them is mutable only when the output function of the type of the other argument is.
fn is_mutable(catalog: &Catalog, expr: &Expr) -> bool {
let mutable_proc = |oid: u32| match builtin::proc_by_oid(oid) {
Some(proc) => proc.volatile != b'i',
None => catalog.function(oid).is_none_or(|f| f.volatile != b'i'),
};
expr.find(0, &mut |e, _| {
let mutable = match &e.kind {
// `textanycat` and `anytextcat`: `$1 || $2::text` and `$1::text || $2`.
ExprKind::Func(f) if matches!(f.oid, 2003 | 2004) => {
let other = f.args.get(if f.oid == 2003 { 1 } else { 0 });
other.is_none_or(|arg| {
mutable_proc(builtin::type_by_oid(arg.ty).map_or(0, |t| t.output))
})
}
ExprKind::Func(f) => mutable_proc(f.oid),
ExprKind::NullIf { equal, .. } | ExprKind::Distinct { equal, .. } => {
mutable_proc(*equal)
}
ExprKind::ScalarArrayOp { func, .. } => mutable_proc(*func),
ExprKind::CoerceViaIo(arg, _) => {
let output = builtin::type_by_oid(arg.ty).map_or(0, |t| t.output);
let input = builtin::type_by_oid(e.ty).map_or(0, |t| t.input);
mutable_proc(output) || mutable_proc(input)
}
ExprKind::SqlValue(_) => true,
_ => false,
};
mutable.then_some(())
})
.is_some()
}
/// The attribute numbers of the columns of the relation that the expressions read, in the order of their first use, without duplicates.
fn column_numbers(exprs: &[&Expr]) -> Vec<i16> {
let mut numbers = Vec::new();
for expr in exprs {
expr.find(0, &mut |e, depth| {
if let ExprKind::Var(var) = e.kind
&& var.levels_up == depth
&& !numbers.contains(&var.attnum)
{
numbers.push(var.attnum);
}
None::<()>
});
}
numbers
}
/// `recordDependencyOnSingleRelExpr`: the objects that the expressions refer to. These are the columns of the relation, the functions that the expressions call, the relations and types of `regclass` and `regtype` constants, and the fields of composite types that the expressions select. The catalog drops the references to pinned objects. A `regrole` or `regdatabase` constant is an error, because a role and a database are shared objects.
fn references(catalog: &Catalog, relation: u32, exprs: &[&Expr]) -> Result<Vec<ObjRef>> {
shared_constants(exprs)?;
let mut refs: Vec<ObjRef> =
column_numbers(exprs).into_iter().map(|n| ObjRef::column(relation, n)).collect();
for expr in exprs {
expr.find(0, &mut |e, _| {
let found = match (&e.kind, e.ty) {
(ExprKind::Const(Value::Oid(oid)), oid::REGCLASS) => {
Some(ObjRef::new(PG_CLASS, *oid))
}
(ExprKind::Const(Value::Oid(oid)), oid::REGTYPE) => {
Some(ObjRef::new(PG_TYPE, *oid))
}
(ExprKind::Func(f), _) => Some(ObjRef::new(PG_PROC, f.oid)),
_ => field_reference(catalog, e),
};
for found in found.into_iter().chain(type_reference(e)) {
if !refs.contains(&found) {
refs.push(found);
}
}
None::<()>
});
}
Ok(refs)
}
/// The error of `find_expr_references_walker` for a `regrole` or `regdatabase` constant in the expressions, because a role and a database are shared objects.
fn shared_constants(exprs: &[&Expr]) -> Result<()> {
for expr in exprs {
let shared = expr.find(0, &mut |e, _| match (&e.kind, e.ty) {
(ExprKind::Const(Value::Oid(_)), oid::REGROLE) => Some("regrole"),
(ExprKind::Const(Value::Oid(_)), oid::REGDATABASE) => Some("regdatabase"),
_ => None,
});
if let Some(name) = shared {
return Err(Error::new(
SqlState::FEATURE_NOT_SUPPORTED,
format!("constant of the type {name} cannot be used here"),
));
}
}
Ok(())
}
/// The column that `find_expr_references_walker` records for a `FieldSelect` of a composite type of the catalog, also through a domain: the column of the relation of the type that has the field.
pub(crate) fn field_reference(catalog: &Catalog, expr: &Expr) -> Option<ObjRef> {
let ExprKind::FieldSelect(arg, field) = &expr.kind else { return None };
let rel = row_type::row_relation(catalog, arg.ty)?;
let (at, _) = rel.columns.iter().enumerate().filter(|(_, c)| !c.dropped).nth(*field)?;
Some(ObjRef::column(rel.oid, i16::try_from(at + 1).ok()?))
}
/// The type that `find_expr_references_walker` records for a constant, a cast or the `Param` of a column of a subquery, which is the type of its result. The catalog drops the references to the built-in types, so only a type such as a domain of `initdb` stays.
pub(crate) fn type_reference(expr: &Expr) -> Option<ObjRef> {
match expr.kind {
ExprKind::Const(_)
| ExprKind::Relabel(..)
| ExprKind::CoerceViaIo(..)
| ExprKind::ArrayCoerce { .. }
| ExprKind::CoerceToDomain(..)
| ExprKind::SubColumn(_) => Some(ObjRef::new(PG_TYPE, expr.ty)),
_ => None,
}
}
impl Definer<'_, '_> {
/// Moves the warnings and the notices of the analyzer to the messages.
fn flush(&mut self) {
self.messages.extend(self.an.notices.drain(..).map(|error| {
if error.is_notice() { Message::Notice(error) } else { Message::Warning(error) }
}));
}
/// Adds a `NOTICE` after the warnings that came before it.
fn notice(&mut self, state: SqlState, message: String) {
self.flush();
self.messages.push(Message::Notice(Error::new(state, message)));
}
/// Adds a `WARNING` after the messages that came before it.
fn warning(&mut self, state: SqlState, message: String) {
self.flush();
self.messages.push(Message::Warning(Error::new(state, message)));
}
/// True when the role of the statement is a superuser.
fn is_superuser(&self) -> bool {
rupg_pgcatalog::roles::current().superuser(self.user)
}
/// The built-in schema with the name, unless `DROP SCHEMA` removed it.
fn builtin_schema(&self, name: &str) -> Option<&'static NamedRow> {
builtin::named(Named::Namespace)
.iter()
.find(|n| n.name == name && !self.catalog.schema_dropped(n.oid))
}
/// The owner of a schema, or `None` for a schema that does not exist.
fn schema_owner(&self, namespace: u32) -> Option<u32> {
match self.catalog.schema(namespace) {
Some(schema) => Some(schema.owner),
None if self.catalog.schema_dropped(namespace) => None,
None => builtin::owned_by_oid(Owned::Namespace, namespace).map(|n| n.owner),
}
}
/// The name of a schema.
fn schema_name(&self, namespace: u32) -> String {
match self.catalog.schema(namespace) {
Some(schema) => schema.name.clone(),
None => builtin::named_by_oid(Named::Namespace, namespace)
.filter(|n| !self.catalog.schema_dropped(n.oid))
.map_or_else(String::new, |n| n.name.to_string()),
}
}
/// `object_aclcheck` of `CREATE` on a schema for a role: a superuser, the owner of the schema and a role that `nspacl` grants `CREATE` can create objects in it.
fn may_create_in(&self, namespace: u32, role: u32) -> bool {
rupg_pgcatalog::roles::current().superuser(role)
|| self.schema_owner(namespace) == Some(role)
|| self.schema_create_allowed(namespace, role)
}
/// The check of `CREATE` on a schema for the role of the statement.
fn check_create_in(&self, namespace: u32, at: Option<usize>) -> Result<()> {
if self.may_create_in(namespace, self.user) {
return Ok(());
}
Err(Error::new(
SqlState::INSUFFICIENT_PRIVILEGE,
format!("permission denied for schema {}", self.schema_name(namespace)),
)
.at_opt(at))
}
/// `RangeVarGetAndCheckCreationNamespace`: the schema where a new relation goes. A temporary relation and a relation in `pg_temp` go to the temporary schema of the session, and `RangeVarAdjustRelationPersistence` checks the persistence of the relation against its schema.
fn creation_namespace(&mut self, rv: &RangeVar) -> Result<u32> {
let at = place(rv.location);
if let Some(catalog) = rv.catalogname.as_deref()
&& catalog != self.an.env.database()
{
return Err(Error::new(
SqlState::FEATURE_NOT_SUPPORTED,
format!("cross-database references are not implemented: \"{}\"", qualified(rv)),
)
.at_opt(at));
}
let unlogged = rv.relpersistence == b'u';
// `RangeVarAdjustRelationPersistence`: an unlogged relation cannot go to a temporary schema.
let only_temporary = || {
Error::new(
SqlState::INVALID_TABLE_DEFINITION,
"only temporary relations may be created in temporary schemas",
)
.at_opt(at)
};
let temp = rv.relpersistence == b't';
if rv.schemaname.as_deref() == Some("pg_temp") || (temp && rv.schemaname.is_none()) {
let namespace = self.temp_namespace()?;
return if unlogged { Err(only_temporary()) } else { Ok(namespace) };
}
let namespace = match rv.schemaname.as_deref() {
Some(schema) => self.an.schema(schema).ok_or_else(|| {
Error::new(
SqlState::UNDEFINED_SCHEMA,
format!("schema \"{schema}\" does not exist"),
)
.at_opt(at)
})?,
None => self.an.env.search_path().first().copied().ok_or_else(|| {
Error::new(SqlState::UNDEFINED_SCHEMA, "no schema has been selected to create in")
.at_opt(at)
})?,
};
if unlogged && self.catalog.is_temp_schema(namespace) {
return Err(only_temporary());
}
let own = namespace == self.an.temp || self.is_own_toast_temp(namespace);
if !own && self.catalog.is_temp_schema(namespace) {
return Err(Error::new(
SqlState::INVALID_TABLE_DEFINITION,
"cannot create relations in temporary schemas of other sessions",
));
}
if temp && !own {
return Err(Error::new(
SqlState::INVALID_TABLE_DEFINITION,
"cannot create temporary relation in non-temporary schema",
)
.at_opt(at));
}
if !own {
self.check_create_in(namespace, at)?;
}
Ok(namespace)
}
/// True for the TOAST schema `pg_toast_temp_N` of the temporary schema of the session.
fn is_own_toast_temp(&self, namespace: u32) -> bool {
let Some(n) = self.an.env.temp_number() else { return false };
self.catalog.schema(namespace).is_some_and(|s| s.name == format!("pg_toast_temp_{n}"))
}
/// `AccessTempTableNamespace` and `InitTempTableNamespace`: the temporary schema `pg_temp_N` of the session. The first temporary object of the session makes it and its TOAST schema `pg_toast_temp_N`, which the bootstrap superuser owns.
fn temp_namespace(&mut self) -> Result<u32> {
if self.an.temp != 0 {
return Ok(self.an.temp);
}
let Some(n) = self.an.env.temp_number() else {
return Err(not_yet("a temporary object in a session with no server", None));
};
let name = format!("pg_temp_{n}");
let oid = match self.catalog.schema_by_name(&name) {
Some(schema) => schema.oid,
None => {
let oid = self.catalog.create_schema(&name, BOOTSTRAP_SUPERUSER)?;
self.catalog.create_schema(&format!("pg_toast_temp_{n}"), BOOTSTRAP_SUPERUSER)?;
oid
}
};
self.an.temp = oid;
self.an.path.retain(|&ns| ns != oid);
if !self.an.temp_path.contains(&oid) {
self.an.temp_path.insert(0, oid);
}
Ok(oid)
}
/// A relation with this name in the schema, of the catalog or built in.
fn relation_in(&self, namespace: u32, name: &str) -> Option<Found> {
if let Some(rel) = self.catalog.relation_by_name(namespace, name) {
return Some(Found::User(rel.oid));
}
builtin::named(Named::Class)
.iter()
.find(|r| r.namespace == namespace && r.name == name)
.and_then(|r| builtin::class_by_oid(r.oid))
.map(Found::Builtin)
}
/// `RangeVarGetRelid`: the relation that a name gives. The errors have no position, as in PostgreSQL.
fn lookup_relation(&self, rv: &RangeVar) -> Result<Found> {
let name = relname(rv);
if let Some(catalog) = rv.catalogname.as_deref()
&& catalog != self.an.env.database()
{
return Err(Error::new(
SqlState::FEATURE_NOT_SUPPORTED,
format!("cross-database references are not implemented: \"{}\"", qualified(rv)),
));
}
let found = match rv.schemaname.as_deref() {
Some(schema) => {
let namespace = self.an.schema(schema).ok_or_else(|| {
Error::new(
SqlState::UNDEFINED_SCHEMA,
format!("schema \"{schema}\" does not exist"),
)
})?;
self.relation_in(namespace, name)
}
None => self.an.temp_path.iter().find_map(|&ns| self.relation_in(ns, name)),
};
found.ok_or_else(|| {
let shown = match rv.schemaname.as_deref() {
Some(schema) => format!("{schema}.{name}"),
None => name.to_string(),
};
Error::new(SqlState::UNDEFINED_TABLE, format!("relation \"{shown}\" does not exist"))
})
}
/// `IsSystemRelation`: true for a relation of the catalog and for a TOAST table or its index.
fn is_system(&self, found: Found) -> bool {
match found {
Found::Builtin(row) => row.oid < rupg_catalog::FIRST_UNPINNED_OID,
Found::User(oid) => {
self.catalog.relation(oid).is_some_and(|r| r.namespace == PG_TOAST_NAMESPACE)
}
}
}
/// The kind of a relation as a `relkind` code.
fn kind_of(&self, found: Found) -> u8 {
match found {
Found::User(oid) => match self.catalog.relation(oid).map(|r| r.kind) {
Some(RelKind::Index) => b'i',
Some(RelKind::Sequence) => b'S',
Some(RelKind::Toast) => b't',
Some(RelKind::View) => b'v',
Some(RelKind::Composite) => b'c',
Some(RelKind::Partitioned) => b'p',
Some(RelKind::PartitionedIndex) => b'I',
Some(RelKind::MatView) => b'm',
_ => b'r',
},
Found::Builtin(row) => row.kind,
}
}
/// `CreateSchemaCommand`. The elements of the statement go to [`Defined::schema`] with their parts of `text`, and the caller runs them.
fn create_schema(&mut self, stmt: &CreateSchemaStmt, text: &str) -> Result<()> {
let owner = match &stmt.authrole {
None => self.user,
Some(role) => self.role_oid(role)?,
};
let roles = rupg_pgcatalog::roles::current();
let owner_name = roles.name(owner);
let name = match stmt.schemaname.as_deref() {
Some(name) => name.to_string(),
None => owner_name.unwrap_or_default().to_string(),
};
if !self.is_superuser() {
return Err(Error::new(
SqlState::INSUFFICIENT_PRIVILEGE,
format!("permission denied for database {}", self.an.env.database()),
));
}
if name.starts_with("pg_") {
return Err(Error::new(
SqlState::RESERVED_NAME,
format!("unacceptable schema name \"{name}\""),
)
.with_detail("The prefix \"pg_\" is reserved for system schemas."));
}
let exists =
self.catalog.schema_by_name(&name).is_some() || self.builtin_schema(&name).is_some();
if exists && stmt.if_not_exists {
self.notice(
SqlState::DUPLICATE_SCHEMA,
format!("schema \"{name}\" already exists, skipping"),
);
return Ok(());
}
if exists {
return Err(Error::new(
SqlState::DUPLICATE_SCHEMA,
format!("schema \"{name}\" already exists"),
));
}
let schema = self.catalog.create_schema(&name, owner)?;
self.apply_default_acl(ObjRef::new(PG_NAMESPACE, schema))?;
if !stmt.schemaElts.is_empty() {
let elements: Vec<Node> = stmt.schemaElts.iter().flatten().cloned().collect();
let elements = schema_elements::schema_elements(&elements, &name, text)?;
self.schema = Some(SchemaElements { owner, schema: name, elements });
}
Ok(())
}
}
/// The detail of `errdetail_relkind_not_supported`.
fn relkind_detail(kind: u8) -> &'static str {
match kind {
b'i' => "This operation is not supported for indexes.",
b'I' => "This operation is not supported for partitioned indexes.",
b'p' => "This operation is not supported for partitioned tables.",
b'S' => "This operation is not supported for sequences.",
b'v' => "This operation is not supported for views.",
b'm' => "This operation is not supported for materialized views.",
b'c' => "This operation is not supported for composite types.",
b'f' => "This operation is not supported for foreign tables.",
b't' => "This operation is not supported for TOAST tables.",
_ => "This operation is not supported for tables.",
}
}
/// The OID of the schema `pg_catalog` or `pg_toast`, where only the bootstrap can create relations.
fn is_system_namespace(namespace: u32) -> bool {
namespace == PG_CATALOG_NAMESPACE || namespace == PG_TOAST_NAMESPACE
}