use super::context::{
PortalExecutionContext, SessionPortalCommandDeclaration, SessionPortalDeclaration,
};
use crate::{
query::{
binding::analyze_query_plan_schema,
locking::{lock_query_relations, query_has_row_locks, validate_query_row_locks},
statement::directional_support::query_plan_backward_scan_support,
},
statement::context::session::StatementPortals,
BackwardScanSupport,
};
use uqa_sql::{
plan::{CommandPlan, QueryPlan, UnifiedPlan},
routines::resolution::RoutineOverloadContext,
semantics::portals::{
cannot_open_command_cursor, command_scroll_returns_nulls, validate_query_options,
PortalDeclarationContext,
},
SQLError, SQLParam, SQLResult,
};
pub fn declare_session_portal<S: Clone + Send + Sync + 'static>(
inputs: &PortalExecutionContext<'_, S>,
params: &[SQLParam],
name: &str,
binary: bool,
scroll: Option<bool>,
hold: bool,
query: &QueryPlan,
) -> Result<SQLResult, SQLError> {
prepare_session_portal(
inputs,
params,
name,
binary,
scroll,
hold,
query,
PortalDeclarationContext::Sql,
)?;
Ok(SQLResult::empty())
}
pub fn open_plpgsql_session_portal<S: Clone + Send + Sync + 'static>(
inputs: &PortalExecutionContext<'_, S>,
params: &[SQLParam],
name: &str,
scroll: Option<bool>,
plan: &UnifiedPlan,
) -> Result<(), SQLError> {
match plan {
UnifiedPlan::Query(query) => prepare_session_portal(
inputs,
params,
name,
false,
scroll,
false,
query,
PortalDeclarationContext::PLpgSQL,
),
UnifiedPlan::Command(command) => {
open_plpgsql_command_portal(inputs, params, name, scroll, command)
}
}
}
fn open_plpgsql_command_portal<S: Clone + Send + Sync + 'static>(
inputs: &PortalExecutionContext<'_, S>,
params: &[SQLParam],
name: &str,
scroll: Option<bool>,
command: &CommandPlan,
) -> Result<(), SQLError> {
let schema = match command {
CommandPlan::Insert(_)
| CommandPlan::Update(_)
| CommandPlan::Delete(_)
| CommandPlan::Merge(_) => cursor_command_returning_schema(inputs, command, params)?,
CommandPlan::Call { name, args } => analyze_call_result_schema(inputs, name, args, params)?,
CommandPlan::ShowVariable { name } => {
inputs.session.show_variable(name)?;
Some(crate::RowSchema::with_types(
vec![name.clone()],
vec![Some(uqa_sql::ColumnType::Text)],
))
}
CommandPlan::Explain { body, format, .. } => {
validate_explain_cursor_body(inputs, params, body)?;
let result = (inputs.explain)(body, false, format.as_deref(), None)?;
Some(crate::RowSchema::with_types(
result.columns,
result.column_types,
))
}
_ => None,
}
.ok_or_else(|| cannot_open_command_cursor(command))?;
let null_returning_values = command_scroll_returns_nulls(command, scroll)?;
inputs
.state
.open_pending_command_session_portal(SessionPortalCommandDeclaration {
metadata: uqa_sql::catalog::session::CursorMetadata {
name: name.to_string(),
source_sql: inputs.source_sql.map(Into::into),
is_holdable: false,
is_binary: false,
is_scrollable: scroll.unwrap_or(false),
created_at_micros: inputs.created_at_micros,
},
command: Box::new(command.clone()),
params: params.to_vec(),
columns: schema.columns().to_vec(),
column_types: schema.column_types().to_vec(),
null_returning_values,
})
}
fn validate_explain_cursor_body<S: Clone + Send + Sync + 'static>(
inputs: &PortalExecutionContext<'_, S>,
params: &[SQLParam],
body: &UnifiedPlan,
) -> Result<(), SQLError> {
match body {
UnifiedPlan::Query(query) => {
lock_query_relations(inputs.queries.row_lock_context(), query)?;
let ctes = inputs.queries.statement_scope(None);
analyze_query_plan_schema(inputs.routines, query, params, &ctes, None)?;
Ok(())
}
UnifiedPlan::Command(command) => {
let _ = cursor_command_returning_schema(inputs, command, params)?;
Ok(())
}
}
}
pub fn ensure_plpgsql_session_portal_available(
state: &dyn StatementPortals,
name: &str,
) -> Result<(), SQLError> {
state
.ensure_session_portal_available(name)
.map_err(|error| {
if error.sqlstate() == Some("42P03") {
SQLError::Routine {
sqlstate: "42P03".into(),
message: format!("cursor \"{name}\" already in use"),
}
} else {
error
}
})
}
#[expect(
clippy::too_many_arguments,
reason = "keeps SQL and PL/pgSQL portal contracts explicit"
)]
fn prepare_session_portal<S: Clone + Send + Sync + 'static>(
inputs: &PortalExecutionContext<'_, S>,
params: &[SQLParam],
name: &str,
binary: bool,
scroll: Option<bool>,
hold: bool,
query: &QueryPlan,
context: PortalDeclarationContext,
) -> Result<(), SQLError> {
if context == PortalDeclarationContext::Sql && !hold && !inputs.state.in_transaction_block() {
return Err(SQLError::Routine {
sqlstate: "25P01".into(),
message: "DECLARE CURSOR can only be used in transaction blocks".into(),
});
}
if context == PortalDeclarationContext::PLpgSQL {
ensure_plpgsql_session_portal_available(inputs.state, name)?;
} else {
inputs.state.ensure_session_portal_available(name)?;
}
let has_row_locks = query_has_row_locks(query);
validate_query_options(context, has_row_locks, hold, scroll)?;
lock_query_relations(inputs.queries.row_lock_context(), query)?;
let ctes = inputs.queries.statement_scope(None);
let schema = analyze_query_plan_schema(inputs.routines, query, params, &ctes, None)?;
validate_query_row_locks(inputs.queries.row_lock_context(), query, params)?;
inputs
.state
.open_pending_session_portal(SessionPortalDeclaration {
metadata: uqa_sql::catalog::session::CursorMetadata {
name: name.to_string(),
source_sql: inputs.source_sql.map(Into::into),
is_holdable: hold,
is_binary: binary,
is_scrollable: query_scrollable(
inputs.routines,
inputs.queries.query_context().source.volatility,
query,
params,
&ctes,
scroll,
)?,
created_at_micros: inputs.created_at_micros,
},
query: query.clone(),
params: params.to_vec(),
columns: schema.columns().to_vec(),
column_types: schema.column_types().to_vec(),
})?;
Ok(())
}
fn query_scrollable<S: Clone>(
routines: &dyn uqa_sql::routines::RoutineResolution,
volatility: &dyn uqa_sql::semantics::volatility::VolatilityCatalog,
query: &QueryPlan,
params: &[SQLParam],
ctes: &crate::query::CteScope<S>,
requested: Option<bool>,
) -> Result<bool, SQLError> {
if let Some(requested) = requested {
return Ok(requested);
}
Ok(!query_has_row_locks(query)
&& query_plan_backward_scan_support(routines, volatility, query, params, ctes)?
== BackwardScanSupport::Native)
}
#[cfg(test)]
mod tests;
fn cursor_command_returning_schema<S: Clone + 'static>(
inputs: &PortalExecutionContext<'_, S>,
command: &CommandPlan,
params: &[SQLParam],
) -> Result<Option<crate::RowSchema>, SQLError> {
crate::mutation::entry::cursor_command_returning_schema(
&inputs.returning.returning_execution_context(),
inputs.returning.returning_analysis_context(),
inputs.command_scopes,
command,
params,
)
}
fn analyze_call_result_schema<S: Clone + 'static>(
inputs: &PortalExecutionContext<'_, S>,
name: &str,
arguments: &[uqa_sql::plan::ExpressionPlan],
params: &[uqa_sql::SQLParam],
) -> Result<Option<uqa_sql::RowSchema>, SQLError> {
let analysis = uqa_sql::routines::call::ProcedureCallAnalysis::new(arguments)?;
let scope = inputs.queries.statement_scope(None);
analysis.result_schema(
name,
&RoutineOverloadContext {
catalog: inputs.overloads,
},
inputs.types,
&mut |argument| {
crate::query::binding::bind_expression_plan_type(
inputs.routines,
argument,
params,
&scope,
)
},
)
}