use super::builtin_function_dispatch_name;
use crate::{
ast::RelationPersistence,
catalog::domain::StoredDomain,
plan::{QueryPlan, UnifiedPlan},
};
use crate::{expr::EngineHook, routines::RoutineResolution, SQLError};
use std::collections::BTreeSet;
pub trait QueryEffectCatalog: RoutineResolution + EngineHook {
fn registered_runtime_function_may_mutate_engine(&self, name: &str) -> bool;
fn domain_by_oid(&self, oid: u32) -> Option<StoredDomain>;
fn sequence_persistence(&self, name: &str) -> Result<Option<RelationPersistence>, String>;
fn table_persistence(&self, name: &str) -> Result<Option<RelationPersistence>, String>;
fn view_plan(&self, name: &str) -> Result<Option<QueryPlan>, SQLError>;
fn lookup_prepared(&self, name: &str) -> Option<UnifiedPlan>;
}
mod domains;
pub struct QueryEffectContext<'a> {
pub catalog: &'a dyn QueryEffectCatalog,
pub optimizer_effects: fn(&QueryPlan) -> bool,
pub graph_effects: fn(&str) -> Result<bool, SQLError>,
}
pub mod read_only;
mod routines;
use routines::plpgsql_function_may_mutate_engine;
#[derive(Clone, Copy)]
struct MutabilityClassification {
include_session_mutations: bool,
procedural_state_requires_transaction: bool,
}
impl MutabilityClassification {
const DATABASE_WRITES: Self = Self {
include_session_mutations: false,
procedural_state_requires_transaction: false,
};
const ENGINE_MUTATIONS: Self = Self {
include_session_mutations: true,
procedural_state_requires_transaction: false,
};
const STATEMENT_TRANSACTION: Self = Self {
include_session_mutations: true,
procedural_state_requires_transaction: true,
};
}
pub fn query_may_mutate_engine(
context: &QueryEffectContext<'_>,
query: &crate::plan::QueryPlan,
) -> Result<bool, SQLError> {
query_may_mutate_engine_inner(
context,
query,
&mut BTreeSet::new(),
&mut BTreeSet::new(),
MutabilityClassification::ENGINE_MUTATIONS,
)
}
pub fn query_requires_statement_transaction(
context: &QueryEffectContext<'_>,
query: &crate::plan::QueryPlan,
) -> Result<bool, SQLError> {
query_may_mutate_engine_inner(
context,
query,
&mut BTreeSet::new(),
&mut BTreeSet::new(),
MutabilityClassification::STATEMENT_TRANSACTION,
)
}
pub fn query_may_write_database(
context: &QueryEffectContext<'_>,
query: &crate::plan::QueryPlan,
) -> Result<bool, SQLError> {
query_may_mutate_engine_inner(
context,
query,
&mut BTreeSet::new(),
&mut BTreeSet::new(),
MutabilityClassification::DATABASE_WRITES,
)
}
pub fn command_payload_may_write_database(
context: &QueryEffectContext<'_>,
command: &crate::plan::CommandPlan,
) -> Result<bool, SQLError> {
let mut plan = crate::plan::UnifiedPlan::Command(Box::new(command.clone()));
let mut writes = false;
let mut classification_error = None;
plan.rewrite_scalar_expressions(&mut |expression| {
if classification_error.is_some() {
return;
}
match scalar_node_may_mutate_engine(
context,
expression,
&mut BTreeSet::new(),
&mut BTreeSet::new(),
MutabilityClassification::DATABASE_WRITES,
) {
Ok(value) => writes |= value,
Err(error) => classification_error = Some(error),
}
});
if let Some(error) = classification_error {
return Err(error);
}
if writes {
return Ok(true);
}
match command {
crate::plan::CommandPlan::Insert(plan) => {
if ctes_may_write_database(context, &plan.ctes)? {
return Ok(true);
}
if let Some(source) = plan.source.as_deref() {
if query_may_write_database(context, source)? {
return Ok(true);
}
}
queries_may_write_database(context, &plan.subqueries)
}
crate::plan::CommandPlan::Update(plan) => {
if ctes_may_write_database(context, &plan.ctes)? {
return Ok(true);
}
if let Some(source) = plan.source.as_deref() {
if source_may_mutate_engine(
context,
source,
&mut BTreeSet::new(),
&mut BTreeSet::new(),
MutabilityClassification::DATABASE_WRITES,
)? {
return Ok(true);
}
}
queries_may_write_database(context, &plan.subqueries)
}
crate::plan::CommandPlan::Delete(plan) => {
if ctes_may_write_database(context, &plan.ctes)? {
return Ok(true);
}
if let Some(source) = plan.source.as_deref() {
if source_may_mutate_engine(
context,
source,
&mut BTreeSet::new(),
&mut BTreeSet::new(),
MutabilityClassification::DATABASE_WRITES,
)? {
return Ok(true);
}
}
queries_may_write_database(context, &plan.subqueries)
}
crate::plan::CommandPlan::Merge(plan) => {
if ctes_may_write_database(context, &plan.ctes)? {
return Ok(true);
}
if source_may_mutate_engine(
context,
&plan.source,
&mut BTreeSet::new(),
&mut BTreeSet::new(),
MutabilityClassification::DATABASE_WRITES,
)? {
return Ok(true);
}
queries_may_write_database(context, &plan.subqueries)
}
_ => Ok(false),
}
}
fn command_may_write_database(
context: &QueryEffectContext<'_>,
command: &crate::plan::CommandPlan,
) -> Result<bool, SQLError> {
read_only::forbidden_command(
context,
&crate::plan::UnifiedPlan::Command(Box::new(command.clone())),
)
.map(|forbidden| forbidden.is_some())
}
fn ctes_may_write_database(
context: &QueryEffectContext<'_>,
ctes: &[crate::plan::CtePlan],
) -> Result<bool, SQLError> {
for cte in ctes {
let writes = match &cte.body {
crate::plan::CtePlanBody::Query(query) => query_may_write_database(context, query)?,
crate::plan::CtePlanBody::Command(command) => {
command_may_write_database(context, command)?
}
};
if writes {
return Ok(true);
}
}
Ok(false)
}
fn queries_may_write_database<'a>(
context: &QueryEffectContext<'_>,
queries: impl IntoIterator<Item = &'a crate::plan::QueryPlan>,
) -> Result<bool, SQLError> {
for query in queries {
if query_may_write_database(context, query)? {
return Ok(true);
}
}
Ok(false)
}
fn query_may_mutate_engine_inner(
context: &QueryEffectContext<'_>,
query: &crate::plan::QueryPlan,
visiting_views: &mut BTreeSet<String>,
visiting_routines: &mut BTreeSet<String>,
classification: MutabilityClassification,
) -> Result<bool, SQLError> {
if query_source_may_mutate_engine(
context,
query,
visiting_views,
visiting_routines,
classification,
)? {
return Ok(true);
}
let mut plan = crate::plan::UnifiedPlan::Query(Box::new(query.clone()));
let mut mutates = (context.optimizer_effects)(query);
let mut classification_error = None;
plan.rewrite_scalar_expressions(&mut |expression| {
if classification_error.is_some() {
return;
}
match scalar_node_may_mutate_engine(
context,
expression,
visiting_views,
visiting_routines,
classification,
) {
Ok(value) => mutates |= value,
Err(error) => classification_error = Some(error),
}
});
classification_error.map_or(Ok(mutates), Err)
}
fn scalar_node_may_mutate_engine(
context: &QueryEffectContext<'_>,
expression: &crate::ScalarExpr,
visiting_views: &mut BTreeSet<String>,
visiting_routines: &mut BTreeSet<String>,
classification: MutabilityClassification,
) -> Result<bool, SQLError> {
match expression {
crate::ScalarExpr::Func {
name,
binding,
args,
..
} => function_may_mutate_engine(
context,
name,
binding.as_ref(),
args,
visiting_views,
visiting_routines,
classification,
),
crate::ScalarExpr::Cast { ty, .. } => {
crate::expr::EngineHook::resolve_type_name(context.catalog, ty)
.ok()
.flatten()
.map_or(Ok(false), |ty| {
domains::domain_cast_may_mutate(
context,
&ty,
visiting_views,
visiting_routines,
classification,
)
})
}
_ => Ok(false),
}
}
fn function_may_mutate_engine(
context: &QueryEffectContext<'_>,
name: &str,
binding: Option<&crate::ast::FunctionBinding>,
args: &[crate::ScalarExpr],
visiting_views: &mut BTreeSet<String>,
visiting_routines: &mut BTreeSet<String>,
classification: MutabilityClassification,
) -> Result<bool, SQLError> {
let identity = name.to_ascii_lowercase();
let dispatch_name = builtin_function_dispatch_name(&identity);
let cypher_mutates = if dispatch_name == "cypher" {
match args.get(1) {
Some(crate::ScalarExpr::Literal(uqa_core::Value::Str(query))) => {
(context.graph_effects)(query)?
}
_ => classification.include_session_mutations,
}
} else {
false
};
let mutates_database_directly = cypher_mutates
|| matches!(
dispatch_name.as_str(),
"create_analyzer"
| "drop_analyzer"
| "set_table_analyzer"
| "graph_create"
| "graph_drop"
| "create_graph"
| "drop_graph"
| "create_vlabel"
| "create_elabel"
| "drop_label"
| "alter_graph"
| "deep_learn"
| "bayesian_match"
| "bayesian_match_with_prior"
)
|| context
.catalog
.registered_runtime_function_may_mutate_engine(&identity);
let temporary_sequence_call = matches!(dispatch_name.as_str(), "nextval" | "setval")
&& sequence_function_targets_temporary(context, args)?;
let builtin_requires_mutating_execution =
matches!(
dispatch_name.as_str(),
"random" | "setseed" | "pg_notify" | "fts_match" | "multi_field_match"
) || (matches!(dispatch_name.as_str(), "nextval" | "setval") && !temporary_sequence_call);
let sql_routine_mutates = classification.include_session_mutations
&& sql_routine_may_mutate_engine(
context,
&identity,
binding,
visiting_views,
visiting_routines,
classification,
)?;
Ok(mutates_database_directly
|| (classification.include_session_mutations
&& (builtin_requires_mutating_execution || sql_routine_mutates)))
}
fn sequence_function_targets_temporary(
context: &QueryEffectContext<'_>,
args: &[crate::ScalarExpr],
) -> Result<bool, SQLError> {
fn literal_reference(expression: &crate::ScalarExpr) -> Option<&str> {
match expression {
crate::ScalarExpr::Literal(uqa_core::Value::Str(reference)) => Some(reference),
crate::ScalarExpr::Cast { expr, .. } => literal_reference(expr),
_ => None,
}
}
let Some(reference) = args.first().and_then(literal_reference) else {
return Ok(false);
};
Ok(context
.catalog
.sequence_persistence(reference)
.map_err(|error| {
SQLError::Internal(format!(
"resolve sequence `{reference}` while classifying query mutability: {error}"
))
})?
.is_some_and(|persistence| persistence == crate::ast::RelationPersistence::Temporary))
}
fn sql_routine_may_mutate_engine(
context: &QueryEffectContext<'_>,
name: &str,
binding: Option<&crate::ast::FunctionBinding>,
visiting_views: &mut BTreeSet<String>,
visiting_routines: &mut BTreeSet<String>,
classification: MutabilityClassification,
) -> Result<bool, SQLError> {
if binding.is_some_and(|binding| binding.builtin) {
return Ok(false);
}
let overloads = match binding {
Some(binding) => context
.catalog
.lookup_bound_sql_functions_by_binding(binding),
None => context
.catalog
.lookup_visible_sql_functions_for_analysis(name)?,
};
let Some(overloads) = overloads else {
return Ok(false);
};
for function in overloads {
if function.def.is_procedure
|| binding.is_some_and(|binding| {
crate::routines::routine_signature_types(&function.def) != binding.argument_types
})
{
continue;
}
let signature = crate::routines::routine_signature_types(&function.def);
let key = format!("{}({})", function.def.name, signature.join(","));
if !visiting_routines.insert(key.clone()) {
continue;
}
let result = (|| {
if domains::routine_coercions_may_mutate(
context,
&function.def,
visiting_views,
visiting_routines,
classification,
)? {
return Ok(true);
}
match &function.compiled {
crate::routines::CompiledFunctionBody::SQL(plans) => (|| {
let mut mutates = false;
for plan in plans {
match plan {
crate::plan::UnifiedPlan::Query(query) => {
if query_may_mutate_engine_inner(
context,
query,
visiting_views,
visiting_routines,
classification,
)? {
mutates = true;
break;
}
}
crate::plan::UnifiedPlan::Command(_) => {
mutates = true;
break;
}
}
}
Ok(mutates)
})(),
crate::routines::CompiledFunctionBody::PLpgSQL(function) => {
plpgsql_function_may_mutate_engine(
context,
function,
visiting_views,
visiting_routines,
classification,
)
}
}
})();
visiting_routines.remove(&key);
if result? {
return Ok(true);
}
}
Ok(false)
}
fn query_source_may_mutate_engine(
context: &QueryEffectContext<'_>,
query: &crate::plan::QueryPlan,
visiting_views: &mut BTreeSet<String>,
visiting_routines: &mut BTreeSet<String>,
classification: MutabilityClassification,
) -> Result<bool, SQLError> {
for cte in &query.ctes {
let mutates = match &cte.body {
crate::plan::CtePlanBody::Query(query) => query_source_may_mutate_engine(
context,
query,
visiting_views,
visiting_routines,
classification,
)?,
crate::plan::CtePlanBody::Command(command) => {
classification.include_session_mutations
|| classification.procedural_state_requires_transaction
|| command_may_write_database(context, command)?
}
};
if mutates {
return Ok(true);
}
}
match &query.root {
crate::plan::RelationalPlan::QueryBlock(block) => {
if let Some(source) = block.from.as_ref() {
if source_may_mutate_engine(
context,
source,
visiting_views,
visiting_routines,
classification,
)? {
return Ok(true);
}
}
for subquery in &block.subqueries {
if query_source_may_mutate_engine(
context,
subquery,
visiting_views,
visiting_routines,
classification,
)? {
return Ok(true);
}
}
}
crate::plan::RelationalPlan::SetOp {
left,
right,
subqueries,
..
} => {
if query_source_may_mutate_engine(
context,
left,
visiting_views,
visiting_routines,
classification,
)? || query_source_may_mutate_engine(
context,
right,
visiting_views,
visiting_routines,
classification,
)? {
return Ok(true);
}
for subquery in subqueries {
if query_source_may_mutate_engine(
context,
subquery,
visiting_views,
visiting_routines,
classification,
)? {
return Ok(true);
}
}
}
crate::plan::RelationalPlan::Values { subqueries, .. } => {
for subquery in subqueries {
if query_source_may_mutate_engine(
context,
subquery,
visiting_views,
visiting_routines,
classification,
)? {
return Ok(true);
}
}
}
}
Ok(false)
}
fn source_may_mutate_engine(
context: &QueryEffectContext<'_>,
source: &crate::plan::SourcePlan,
visiting_views: &mut BTreeSet<String>,
visiting_routines: &mut BTreeSet<String>,
classification: MutabilityClassification,
) -> Result<bool, SQLError> {
match source {
crate::plan::SourcePlan::Function {
name,
binding,
args,
..
} => function_may_mutate_engine(
context,
name,
binding.as_ref(),
args,
visiting_views,
visiting_routines,
classification,
),
crate::plan::SourcePlan::FunctionGroup { functions, .. } => {
for function in functions {
if function_may_mutate_engine(
context,
&function.name,
function.binding.as_ref(),
&function.args,
visiting_views,
visiting_routines,
classification,
)? {
return Ok(true);
}
}
Ok(false)
}
crate::plan::SourcePlan::Join { left, right, .. } => Ok(source_may_mutate_engine(
context,
left,
visiting_views,
visiting_routines,
classification,
)? || source_may_mutate_engine(
context,
right,
visiting_views,
visiting_routines,
classification,
)?),
crate::plan::SourcePlan::Subquery { body, .. } => query_source_may_mutate_engine(
context,
body,
visiting_views,
visiting_routines,
classification,
),
crate::plan::SourcePlan::Table { name, .. } => {
let key = name.to_ascii_lowercase();
if !visiting_views.insert(key.clone()) {
return Ok(false);
}
let result = match context.catalog.view_plan(name) {
Ok(Some(view)) => query_may_mutate_engine_inner(
context,
&view,
visiting_views,
visiting_routines,
classification,
),
Ok(None) => Ok(false),
Err(error) => Err(error),
};
visiting_views.remove(&key);
result
}
crate::plan::SourcePlan::Values { .. } => Ok(false),
}
}
pub fn is_transaction_control(plan: &crate::plan::UnifiedPlan) -> bool {
matches!(
plan,
crate::plan::UnifiedPlan::Command(command)
if matches!(command.as_ref(), crate::plan::CommandPlan::Transaction(_))
)
}
pub mod transaction_blocks;