use core::ops::ControlFlow;
use sqlparser::ast::{self, Expr, Visit, Visitor};
use crate::error::{Result, SqlError};
use crate::functions::registry::FunctionRegistry;
use crate::parser::normalize::normalize_ident;
use crate::resolver::ColumnScope;
use crate::resolver::columns::TableScope;
use crate::resolver::expr::convert_expr;
use crate::types::{AggregateExpr, SqlExpr, SqlValue};
pub fn contains_aggregate(expr: &Expr, functions: &FunctionRegistry) -> bool {
let mut detector = AggregateDetector {
functions,
found: false,
};
let _ = expr.visit(&mut detector);
detector.found
}
pub fn extract_aggregates(
expr: &Expr,
alias: &str,
functions: &FunctionRegistry,
scope: &TableScope,
) -> Result<Vec<AggregateExpr>> {
let mut extractor = AggregateExtractor {
functions,
alias,
scope,
inside_aggregate: 0,
out: Vec::new(),
error: None,
};
let _ = expr.visit(&mut extractor);
if let Some(e) = extractor.error {
return Err(e);
}
Ok(extractor.out)
}
struct AggregateDetector<'a> {
functions: &'a FunctionRegistry,
found: bool,
}
impl Visitor for AggregateDetector<'_> {
type Break = ();
fn pre_visit_expr(&mut self, expr: &Expr) -> ControlFlow<()> {
if let Expr::Function(f) = expr
&& self.functions.is_aggregate(&function_name(f))
&& f.over.is_none()
{
self.found = true;
return ControlFlow::Break(());
}
ControlFlow::Continue(())
}
}
struct AggregateExtractor<'a> {
functions: &'a FunctionRegistry,
alias: &'a str,
scope: &'a TableScope,
inside_aggregate: u32,
out: Vec<AggregateExpr>,
error: Option<SqlError>,
}
impl Visitor for AggregateExtractor<'_> {
type Break = ();
fn pre_visit_expr(&mut self, expr: &Expr) -> ControlFlow<()> {
if self.error.is_some() {
return ControlFlow::Break(());
}
if let Expr::Function(f) = expr
&& self.functions.is_aggregate(&function_name(f))
&& f.over.is_none()
{
if self.inside_aggregate > 0 {
self.error = Some(SqlError::Unsupported {
detail: format!(
"nested aggregate functions are not allowed: {}(...{}...)",
function_name(f),
function_name(f),
),
});
return ControlFlow::Break(());
}
let (args, distinct) = match function_args_and_distinct(f, self.scope) {
Ok(parsed) => parsed,
Err(e) => {
self.error = Some(e);
return ControlFlow::Break(());
}
};
let filter = match aggregate_filter(f, self.scope) {
Ok(filter) => filter,
Err(e) => {
self.error = Some(e);
return ControlFlow::Break(());
}
};
let mut arg_text = computed_arg_text(f, &args);
let (args, filter_text) = match filter {
Some((predicate, text)) => {
if arg_text.is_none() {
arg_text = Some(match args.first() {
Some(SqlExpr::Column { name, .. }) => name.clone(),
_ => "*".to_string(),
});
}
(filtered_args(args, predicate), Some(text))
}
None => (args, None),
};
self.out.push(AggregateExpr {
function: function_name(f),
args,
alias: self.alias.into(),
distinct,
grouping_col_index: None,
filter_text,
arg_text,
});
self.inside_aggregate += 1;
}
ControlFlow::Continue(())
}
fn post_visit_expr(&mut self, expr: &Expr) -> ControlFlow<()> {
if let Expr::Function(f) = expr
&& self.functions.is_aggregate(&function_name(f))
&& f.over.is_none()
&& self.inside_aggregate > 0
{
self.inside_aggregate -= 1;
}
ControlFlow::Continue(())
}
}
fn function_name(f: &ast::Function) -> String {
if f.name.0.len() > 1 {
let qualified: String = f
.name
.0
.iter()
.map(|p| match p {
ast::ObjectNamePart::Identifier(ident) => ident.value.clone(),
_ => String::new(),
})
.collect::<Vec<_>>()
.join(".");
return format!("__schema_qualified__{qualified}");
}
f.name
.0
.iter()
.map(|p| match p {
ast::ObjectNamePart::Identifier(ident) => normalize_ident(ident),
_ => String::new(),
})
.collect::<Vec<_>>()
.join(".")
}
fn function_args_and_distinct(
f: &ast::Function,
scope: &TableScope,
) -> Result<(Vec<SqlExpr>, bool)> {
let ast::FunctionArguments::List(args) = &f.args else {
return Ok((Vec::new(), false));
};
let mut parsed = Vec::with_capacity(args.args.len());
for a in &args.args {
match a {
ast::FunctionArg::Unnamed(ast::FunctionArgExpr::Expr(e)) => {
parsed.push(convert_expr(e, &ColumnScope::Relations(scope))?);
}
ast::FunctionArg::Unnamed(ast::FunctionArgExpr::Wildcard) => {
parsed.push(SqlExpr::Wildcard);
}
_ => {}
}
}
let distinct = matches!(
args.duplicate_treatment,
Some(ast::DuplicateTreatment::Distinct)
);
Ok((parsed, distinct))
}
fn aggregate_filter(f: &ast::Function, scope: &TableScope) -> Result<Option<(SqlExpr, String)>> {
let Some(predicate) = f.filter.as_deref() else {
return Ok(None);
};
Ok(Some((
convert_expr(predicate, &ColumnScope::Relations(scope))?,
format!("{predicate}").to_lowercase(),
)))
}
fn filtered_args(args: Vec<SqlExpr>, predicate: SqlExpr) -> Vec<SqlExpr> {
let args = if args.is_empty() {
vec![SqlExpr::Wildcard]
} else {
args
};
args.into_iter()
.map(|arg| {
let value = match arg {
SqlExpr::Wildcard => SqlExpr::Literal(SqlValue::Int(1)),
other => other,
};
SqlExpr::Case {
operand: None,
when_then: vec![(predicate.clone(), value)],
else_expr: None,
}
})
.collect()
}
fn computed_arg_text(f: &ast::Function, args: &[SqlExpr]) -> Option<String> {
match args.first() {
None | Some(SqlExpr::Column { .. }) | Some(SqlExpr::Wildcard) => return None,
Some(_) => {}
}
let ast::FunctionArguments::List(list) = &f.args else {
return None;
};
list.args.iter().find_map(|a| match a {
ast::FunctionArg::Unnamed(ast::FunctionArgExpr::Expr(e)) => {
Some(format!("{e}").to_lowercase())
}
_ => None,
})
}
#[cfg(test)]
mod tests {
use super::*;
use crate::parser::statement::parse_sql;
use crate::resolver::columns::ResolvedTable;
use crate::types::{CollectionInfo, EngineType};
fn first_select_projection(sql: &str) -> Vec<ast::SelectItem> {
let stmts = parse_sql(sql).unwrap();
match stmts.into_iter().next().unwrap() {
ast::Statement::Query(q) => match *q.body {
ast::SetExpr::Select(s) => s.projection,
_ => panic!(),
},
_ => panic!(),
}
}
fn first_expr(sql: &str) -> ast::Expr {
match first_select_projection(sql).into_iter().next().unwrap() {
ast::SelectItem::UnnamedExpr(e) | ast::SelectItem::ExprWithAlias { expr: e, .. } => e,
_ => panic!(),
}
}
fn functions() -> FunctionRegistry {
FunctionRegistry::new()
}
fn open_scope() -> TableScope {
let info = CollectionInfo {
name: "t".into(),
engine: EngineType::DocumentSchemaless,
columns: Vec::new(),
primary_key: None,
has_auto_tier: false,
indexes: Vec::new(),
bitemporal: false,
primary: nodedb_types::PrimaryEngine::Document,
vector_primary: None,
partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
open_schema: CollectionInfo::open_schema_for(EngineType::DocumentSchemaless),
};
TableScope::single(ResolvedTable {
name: info.name.clone(),
alias: None,
info,
})
.expect("single-relation scope")
}
#[test]
fn detect_plain_aggregate() {
assert!(contains_aggregate(
&first_expr("SELECT SUM(x) FROM t"),
&functions()
));
}
#[test]
fn detect_aggregate_inside_case() {
assert!(contains_aggregate(
&first_expr("SELECT CASE WHEN x > 0 THEN SUM(y) ELSE 0 END FROM t"),
&functions(),
));
}
#[test]
fn detect_aggregate_inside_cast() {
assert!(contains_aggregate(
&first_expr("SELECT CAST(SUM(x) AS TEXT) FROM t"),
&functions(),
));
}
#[test]
fn detect_aggregate_inside_unary_op() {
assert!(contains_aggregate(
&first_expr("SELECT -SUM(x) FROM t"),
&functions(),
));
}
#[test]
fn detect_aggregate_inside_coalesce() {
assert!(contains_aggregate(
&first_expr("SELECT COALESCE(SUM(x), 0) FROM t"),
&functions(),
));
}
#[test]
fn detect_aggregate_inside_between() {
assert!(contains_aggregate(
&first_expr("SELECT SUM(x) BETWEEN 1 AND 10 FROM t"),
&functions(),
));
}
#[test]
fn detect_aggregate_inside_in_list() {
assert!(contains_aggregate(
&first_expr("SELECT SUM(x) IN (1, 2, 3) FROM t"),
&functions(),
));
}
#[test]
fn no_aggregate_in_plain_select() {
assert!(!contains_aggregate(
&first_expr("SELECT x FROM t"),
&functions()
));
assert!(!contains_aggregate(
&first_expr("SELECT x + 1 FROM t"),
&functions()
));
assert!(!contains_aggregate(
&first_expr("SELECT upper(name) FROM t"),
&functions(),
));
}
#[test]
fn extract_plain_aggregate() {
let aggs = extract_aggregates(
&first_expr("SELECT SUM(x) FROM t"),
"total",
&functions(),
&open_scope(),
)
.unwrap();
assert_eq!(aggs.len(), 1);
assert_eq!(aggs[0].function, "sum");
assert_eq!(aggs[0].alias, "total");
}
#[test]
fn extract_aggregate_inside_cast() {
let aggs = extract_aggregates(
&first_expr("SELECT CAST(SUM(x) AS TEXT) AS n FROM t"),
"n",
&functions(),
&open_scope(),
)
.unwrap();
assert_eq!(aggs.len(), 1);
assert_eq!(aggs[0].function, "sum");
}
#[test]
fn extract_aggregate_inside_case() {
let aggs = extract_aggregates(
&first_expr("SELECT CASE WHEN x > 0 THEN SUM(y) ELSE 0 END FROM t"),
"r",
&functions(),
&open_scope(),
)
.unwrap();
assert_eq!(aggs.len(), 1);
assert_eq!(aggs[0].function, "sum");
}
#[test]
fn extract_aggregate_inside_coalesce() {
let aggs = extract_aggregates(
&first_expr("SELECT COALESCE(SUM(x), 0) FROM t"),
"r",
&functions(),
&open_scope(),
)
.unwrap();
assert_eq!(aggs.len(), 1);
assert_eq!(aggs[0].function, "sum");
}
#[test]
fn extract_two_aggregates_under_one_alias() {
let aggs = extract_aggregates(
&first_expr("SELECT SUM(x) + COUNT(y) AS total FROM t"),
"total",
&functions(),
&open_scope(),
)
.unwrap();
assert_eq!(aggs.len(), 2);
let names: Vec<&str> = aggs.iter().map(|a| a.function.as_str()).collect();
assert!(names.contains(&"sum"));
assert!(names.contains(&"count"));
}
#[test]
fn nested_aggregate_directly_inside_aggregate_rejected() {
let err = extract_aggregates(
&first_expr("SELECT SUM(AVG(x)) FROM t"),
"r",
&functions(),
&open_scope(),
)
.unwrap_err();
let msg = format!("{err:?}");
assert!(
msg.to_lowercase().contains("nested aggregate"),
"error must identify the nested-aggregate class: {msg}"
);
}
#[test]
fn nested_aggregate_through_cast_rejected() {
let err = extract_aggregates(
&first_expr("SELECT SUM(CAST(AVG(x) AS BIGINT)) FROM t"),
"r",
&functions(),
&open_scope(),
)
.unwrap_err();
assert!(
format!("{err:?}")
.to_lowercase()
.contains("nested aggregate"),
"got: {err:?}"
);
}
#[test]
fn sibling_aggregates_not_treated_as_nested() {
let aggs = extract_aggregates(
&first_expr("SELECT CAST(SUM(x) AS TEXT) || CAST(COUNT(y) AS TEXT) FROM t"),
"r",
&functions(),
&open_scope(),
)
.unwrap();
assert_eq!(aggs.len(), 2);
}
#[test]
fn extract_distinct_preserved() {
let aggs = extract_aggregates(
&first_expr("SELECT COUNT(DISTINCT x) FROM t"),
"c",
&functions(),
&open_scope(),
)
.unwrap();
assert_eq!(aggs.len(), 1);
assert!(aggs[0].distinct);
}
}