use crate::catalog::DriftReport;
use crate::catalog::error::CatalogError;
use crate::catalog::filter::CatalogFilter;
use crate::catalog::rows::Row;
use crate::identifier::QualifiedName;
use crate::ir::function::{
ArgMode, Function, FunctionArg, FunctionLanguage, NormalizedArgTypes, ParallelSafety,
ReturnType, SecurityMode, TableColumn, Volatility,
};
use crate::ir::procedure::Procedure;
use crate::parse::error::SourceLocation;
use super::ident_required;
#[allow(clippy::too_many_lines)] pub(super) fn build_functions_and_procedures(
rows: &[Row],
filter: &CatalogFilter,
drift: &mut DriftReport,
) -> Result<(Vec<Function>, Vec<Procedure>), CatalogError> {
use crate::catalog::CatalogQuery as Q;
let mut functions: Vec<Function> = Vec::new();
let mut procedures: Vec<Procedure> = Vec::new();
let catalog_location = SourceLocation::new(std::path::PathBuf::from("<catalog>"), 1, 1);
for r in rows {
let schema_name = r.get_text(Q::Functions, "schema_name")?;
let name = r.get_text(Q::Functions, "name")?;
let kind = r.get_text(Q::Functions, "kind")?;
let language_str = r.get_text(Q::Functions, "language")?;
let arg_full = r.get_text(Q::Functions, "arg_full")?;
let return_type_str = r
.get_opt_text(Q::Functions, "return_type")?
.unwrap_or_default();
let full_def = r.get_text(Q::Functions, "full_def")?;
let comment = r.get_opt_text(Q::Functions, "comment")?;
let security_definer = r.get_bool(Q::Functions, "security_definer")?;
let qname = QualifiedName::new(ident_required(&schema_name)?, ident_required(&name)?);
if !filter.allows(&qname) {
continue;
}
let language = match language_str.to_lowercase().as_str() {
"sql" => FunctionLanguage::Sql,
"plpgsql" => FunctionLanguage::PlPgSql,
other => {
drift
.unmanaged_language_routines
.push((qname, other.to_string()));
continue;
}
};
let security = if security_definer {
SecurityMode::Definer
} else {
SecurityMode::Invoker
};
let args = parse_arg_full(&arg_full, &qname, &catalog_location)?;
let body_text = extract_body_from_functiondef(&full_def, &qname)?;
let (body, body_dependencies, commits_in_body) =
crate::parse::builder::plpgsql::parse_routine_body(
&body_text,
language,
&qname,
&catalog_location,
)
.map_err(|e| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"catalog body parse for {qname}: {e}"
)))
})?;
let owner_str = r.get_text(Q::Functions, "owner")?;
let owner_ident =
crate::identifier::Identifier::from_unquoted(&owner_str).map_err(|e| {
CatalogError::BadColumnType {
query: Q::Functions,
column: "owner".to_string(),
message: format!("invalid owner {owner_str:?}: {e}"),
}
})?;
let acl_strings = r.get_text_array(Q::Functions, "acl")?;
let raw_grants = crate::catalog::grants::decode_aclitem_array(&acl_strings)?;
let grants = crate::catalog::grants::strip_owner_self_grants(raw_grants, &owner_ident);
let owner = Some(owner_ident);
match kind.as_str() {
"f" => {
let volatility_str = r.get_text(Q::Functions, "volatility")?;
let strict = r.get_bool(Q::Functions, "strict")?;
let parallel_str = r.get_text(Q::Functions, "parallel")?;
let leakproof = r.get_bool(Q::Functions, "leakproof")?;
let cost_str = r.get_opt_text(Q::Functions, "cost")?;
let rows_str = r.get_opt_text(Q::Functions, "rows")?;
let volatility = match volatility_str.as_str() {
"i" => Volatility::Immutable,
"s" => Volatility::Stable,
_ => Volatility::Volatile,
};
let parallel = match parallel_str.as_str() {
"s" => ParallelSafety::Safe,
"r" => ParallelSafety::Restricted,
_ => ParallelSafety::Unsafe,
};
let cost: Option<f32> = cost_str.as_deref().and_then(|s| s.parse::<f32>().ok());
let rows: Option<f32> = rows_str.as_deref().and_then(|s| s.parse::<f32>().ok());
let return_type = parse_return_type_from_string(&return_type_str, &qname)?;
let arg_types_normalized = NormalizedArgTypes::from_args(&args);
functions.push(Function {
qname,
args,
arg_types_normalized,
return_type,
language,
body,
body_dependencies,
volatility,
strict,
security,
parallel,
leakproof,
cost,
rows,
comment,
owner,
grants,
});
}
"p" => {
procedures.push(Procedure {
qname,
args,
language,
body,
body_dependencies,
security,
commits_in_body,
comment,
owner,
grants,
});
}
other => {
return Err(CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(
format!("unexpected prokind {other:?} in functions query"),
)));
}
}
}
Ok((functions, procedures))
}
fn parse_arg_full(
arg_full: &str,
qname: &QualifiedName,
location: &SourceLocation,
) -> Result<Vec<FunctionArg>, CatalogError> {
use pg_query::NodeEnum;
use pg_query::protobuf::FunctionParameterMode;
if arg_full.trim().is_empty() {
return Ok(vec![]);
}
let wrapper = format!(
"CREATE FUNCTION pgevolve_temp({arg_full}) RETURNS void LANGUAGE sql AS $$ SELECT NULL $$;"
);
let parsed = pg_query::parse(&wrapper).map_err(|e| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"catalog arg parse for {qname} ({arg_full:?}): {e}"
)))
})?;
let stmt = parsed
.protobuf
.stmts
.into_iter()
.next()
.and_then(|r| r.stmt)
.and_then(|n| n.node)
.ok_or_else(|| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"catalog arg parse for {qname}: no statement"
)))
})?;
let NodeEnum::CreateFunctionStmt(stmt) = stmt else {
return Err(CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(
format!("catalog arg parse for {qname}: unexpected stmt kind"),
)));
};
let mut args: Vec<FunctionArg> = Vec::new();
for param_node in &stmt.parameters {
let Some(NodeEnum::FunctionParameter(p)) = param_node.node.as_ref() else {
continue;
};
let raw_mode =
FunctionParameterMode::try_from(p.mode).unwrap_or(FunctionParameterMode::Undefined);
let mode = match raw_mode {
FunctionParameterMode::FuncParamIn
| FunctionParameterMode::FuncParamDefault
| FunctionParameterMode::Undefined => ArgMode::In,
FunctionParameterMode::FuncParamOut | FunctionParameterMode::FuncParamTable => {
ArgMode::Out
}
FunctionParameterMode::FuncParamInout => ArgMode::InOut,
FunctionParameterMode::FuncParamVariadic => ArgMode::Variadic,
};
let Some(tn) = p.arg_type.as_ref() else {
continue;
};
let ty =
crate::parse::builder::shared::type_name_to_column_type(tn, location).map_err(|e| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"catalog arg type for {qname}: {e}"
)))
})?;
let arg_name = if p.name.is_empty() {
None
} else {
Some(ident_required(&p.name)?)
};
let default = p
.defexpr
.as_ref()
.and_then(|defexpr| defexpr.node.as_ref())
.and_then(|node_enum| {
crate::parse::normalize_expr::from_pg_node(node_enum, Some(&ty), location).ok()
});
args.push(FunctionArg {
name: arg_name,
mode,
ty,
default,
});
}
Ok(args)
}
fn extract_body_from_functiondef(
full_def: &str,
qname: &QualifiedName,
) -> Result<String, CatalogError> {
let upper = full_def.to_ascii_uppercase();
if let Some(as_pos) = find_as_dollar_quote_pos(&upper) {
let rest = &full_def[as_pos..];
let tag_end = rest[1..].find('$').ok_or_else(|| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"catalog body for {qname}: malformed dollar-quote open"
)))
})?;
let tag = &rest[1..=tag_end]; let open = format!("${tag}$");
let body_start = as_pos + open.len();
let body_text_after_open = &full_def[body_start..];
let close_pos = body_text_after_open.find(open.as_str()).ok_or_else(|| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"catalog body for {qname}: closing dollar-quote {open:?} not found"
)))
})?;
let body = &body_text_after_open[..close_pos];
return Ok(body.to_string());
}
if let Some(begin_pos) = upper.find("BEGIN ATOMIC") {
let body_start = begin_pos; let tail = &full_def[body_start..];
let tail_upper = tail.to_ascii_uppercase();
let end_pos = tail_upper.rfind("END").ok_or_else(|| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"catalog body for {qname}: BEGIN ATOMIC without END"
)))
})?;
let body = &tail[..end_pos + "END".len()];
return Ok(body.to_string());
}
Err(CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(
format!("catalog body for {qname}: could not extract body from pg_get_functiondef"),
)))
}
fn find_as_dollar_quote_pos(upper: &str) -> Option<usize> {
let bytes = upper.as_bytes();
for i in 0..bytes.len().saturating_sub(3) {
if bytes[i] == b'A' && bytes[i + 1] == b'S' {
if i > 0 && !bytes[i - 1].is_ascii_whitespace() {
continue;
}
let mut j = i + 2;
while j < bytes.len() && bytes[j].is_ascii_whitespace() {
j += 1;
}
if j < bytes.len() && bytes[j] == b'$' {
return Some(j);
}
}
}
None
}
fn parse_return_type_from_string(
s: &str,
qname: &QualifiedName,
) -> Result<ReturnType, CatalogError> {
let trimmed = s.trim();
let lower = trimmed.to_ascii_lowercase();
if lower == "trigger" {
return Ok(ReturnType::Trigger);
}
if lower == "event_trigger" {
return Ok(ReturnType::EventTrigger);
}
if lower == "void" {
return Ok(ReturnType::Void);
}
if let Some(inner) = lower.strip_prefix("setof ") {
let ty = parse_return_scalar_type(inner, trimmed, qname)?;
return Ok(ReturnType::SetOf { ty });
}
if lower.starts_with("table(") && lower.ends_with(')') {
let inner = &trimmed[6..trimmed.len() - 1]; let columns = parse_table_return_columns(inner, qname)?;
return Ok(ReturnType::Table { columns });
}
let ty = parse_return_scalar_type(&lower, trimmed, qname)?;
Ok(ReturnType::Scalar { ty })
}
fn parse_return_scalar_type(
lower: &str,
original: &str,
_qname: &QualifiedName,
) -> Result<crate::ir::column_type::ColumnType, CatalogError> {
crate::ir::column_type::ColumnType::parse_from_pg_type_string(lower)
.or_else(|_| crate::ir::column_type::ColumnType::parse_from_pg_type_string(original))
.map_err(|e| {
CatalogError::Ir(crate::ir::IrError::InvalidColumnType(format!(
"return type {original:?}: {e}"
)))
})
}
fn parse_table_return_columns(
inner: &str,
qname: &QualifiedName,
) -> Result<Vec<TableColumn>, CatalogError> {
use pg_query::protobuf::FunctionParameterMode;
let wrapper = format!(
"CREATE FUNCTION pgevolve_temp() RETURNS TABLE({inner}) LANGUAGE sql AS $$ SELECT NULL $$;"
);
let parsed = pg_query::parse(&wrapper).map_err(|e| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"RETURNS TABLE parse for {qname} ({inner:?}): {e}"
)))
})?;
let stmt = parsed
.protobuf
.stmts
.into_iter()
.next()
.and_then(|r| r.stmt)
.and_then(|n| n.node)
.ok_or_else(|| {
CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(format!(
"RETURNS TABLE parse for {qname}: no stmt"
)))
})?;
let pg_query::NodeEnum::CreateFunctionStmt(stmt) = stmt else {
return Err(CatalogError::Ir(crate::ir::IrError::InvalidIdentifier(
format!("RETURNS TABLE parse for {qname}: unexpected stmt kind"),
)));
};
let catalog_location = SourceLocation::new(std::path::PathBuf::from("<catalog>"), 1, 1);
let mut columns: Vec<TableColumn> = Vec::new();
for param_node in &stmt.parameters {
let Some(pg_query::NodeEnum::FunctionParameter(p)) = param_node.node.as_ref() else {
continue;
};
let raw_mode =
FunctionParameterMode::try_from(p.mode).unwrap_or(FunctionParameterMode::Undefined);
if raw_mode != FunctionParameterMode::FuncParamTable {
continue;
}
let Some(tn) = p.arg_type.as_ref() else {
continue;
};
let ty = crate::parse::builder::shared::type_name_to_column_type(tn, &catalog_location)
.map_err(|e| {
CatalogError::Ir(crate::ir::IrError::InvalidColumnType(format!(
"RETURNS TABLE col type for {qname}: {e}"
)))
})?;
let nm = ident_required(&p.name)?;
columns.push(TableColumn { name: nm, ty });
}
Ok(columns)
}