use alloc::borrow::Cow;
use alloc::boxed::Box;
use alloc::collections::BTreeSet;
use alloc::format;
use alloc::string::{String, ToString};
use alloc::vec::Vec;
use spg_sql::ast::{Expr, SelectItem, SelectStatement};
use spg_storage::{ColumnSchema, DataType, Row, Value};
use crate::eval::{self, EvalContext, EvalError};
use crate::join::AggRows;
impl crate::Engine {
pub(crate) fn expand_aggregate_wildcard(
&self,
stmt: &SelectStatement,
) -> Option<SelectStatement> {
use spg_sql::ast::SelectItem;
if !stmt
.items
.iter()
.any(|i| matches!(i, SelectItem::Wildcard | SelectItem::QualifiedWildcard(_)))
{
return None;
}
if !uses_aggregate(stmt) {
return None;
}
let from = stmt.from.as_ref()?;
if !from.joins.is_empty()
|| from.primary.unnest_expr.is_some()
|| from.primary.lateral_subquery.is_some()
|| from.primary.generate_series_args.is_some()
|| from.primary.table_fn_call.is_some()
|| from.primary.json_table.is_some()
|| from.primary.jsonb_each_text_arg.is_some()
{
return None;
}
let table = self.active_catalog().get(&from.primary.name)?;
let alias = from
.primary
.alias
.clone()
.unwrap_or_else(|| from.primary.name.clone());
let mut items: Vec<SelectItem> = Vec::with_capacity(stmt.items.len());
for item in &stmt.items {
match item {
SelectItem::Wildcard => {
for c in &table.schema().columns {
items.push(SelectItem::Expr {
expr: Expr::Column(spg_sql::ast::ColumnName {
qualifier: None,
name: c.name.clone(),
}),
alias: None,
});
}
}
SelectItem::QualifiedWildcard(q) => {
if !q.eq_ignore_ascii_case(&alias) {
return None; }
for c in &table.schema().columns {
items.push(SelectItem::Expr {
expr: Expr::Column(spg_sql::ast::ColumnName {
qualifier: None,
name: c.name.clone(),
}),
alias: None,
});
}
}
other => items.push(other.clone()),
}
}
let mut out = stmt.clone();
out.items = items;
Some(out)
}
}
pub fn uses_aggregate(stmt: &SelectStatement) -> bool {
if stmt.group_by.is_some() || stmt.having.is_some() {
return true;
}
for item in &stmt.items {
if let SelectItem::Expr { expr, .. } = item
&& contains_aggregate(expr)
{
return true;
}
}
for o in &stmt.order_by {
if contains_aggregate(&o.expr) {
return true;
}
}
if let Some(h) = &stmt.having
&& contains_aggregate(h)
{
return true;
}
false
}
pub fn contains_aggregate(e: &Expr) -> bool {
match e {
Expr::FunctionCall { name, args } => {
is_aggregate_name(name) || args.iter().any(contains_aggregate)
}
Expr::NamedArg { expr, .. } => contains_aggregate(expr),
Expr::Variadic(expr) => contains_aggregate(expr),
Expr::AggregateOrdered { .. } => true,
Expr::Binary { lhs, rhs, .. } => contains_aggregate(lhs) || contains_aggregate(rhs),
Expr::Unary { expr, .. }
| Expr::Cast { expr, .. }
| Expr::IsNull { expr, .. }
| Expr::BoolTest { expr, .. }
| Expr::FieldAccess { base: expr, .. } => contains_aggregate(expr),
Expr::Like { expr, pattern, .. } => contains_aggregate(expr) || contains_aggregate(pattern),
Expr::Extract { source, .. } => contains_aggregate(source),
Expr::ScalarSubquery(_)
| Expr::Exists { .. }
| Expr::InSubquery { .. }
| Expr::RowInSubquery { .. }
| Expr::RowCmpSubquery { .. }
| Expr::WindowFunction { .. }
| Expr::Literal(_)
| Expr::Placeholder(_)
| Expr::Column(_) => false,
Expr::Array(items) => items.iter().any(contains_aggregate),
Expr::ArraySubscript { target, index } => {
contains_aggregate(target) || contains_aggregate(index)
}
Expr::ArraySlice { target, lo, hi } => {
contains_aggregate(target)
|| lo.as_deref().is_some_and(contains_aggregate)
|| hi.as_deref().is_some_and(contains_aggregate)
}
Expr::AnyAll { expr, array, .. } => contains_aggregate(expr) || contains_aggregate(array),
Expr::InList { expr, list, .. } => {
contains_aggregate(expr) || list.iter().any(contains_aggregate)
}
Expr::Case {
operand,
branches,
else_branch,
} => {
operand.as_deref().is_some_and(contains_aggregate)
|| branches
.iter()
.any(|(w, t)| contains_aggregate(w) || contains_aggregate(t))
|| else_branch.as_deref().is_some_and(contains_aggregate)
}
}
}
pub fn is_aggregate_name(name: &str) -> bool {
matches!(
name.to_ascii_lowercase().as_str(),
"count"
| "count_star"
| "sum"
| "min"
| "max"
| "avg"
| "string_agg"
| "array_agg"
| "any_value"
| "range_agg"
| "range_intersect_agg"
| "group_concat"
| "xmlagg"
| "bool_and"
| "bool_or"
| "every"
| "stddev" | "stddev_samp" | "stddev_pop"
| "variance" | "var_samp" | "var_pop"
| "bit_and" | "bit_or" | "bit_xor"
| "percentile_cont" | "percentile_disc" | "mode"
| "rank" | "dense_rank" | "percent_rank" | "cume_dist"
| "covar_pop" | "covar_samp" | "corr"
| "regr_count" | "regr_avgx" | "regr_avgy" | "regr_slope"
| "regr_intercept" | "regr_r2" | "regr_sxx" | "regr_syy" | "regr_sxy"
| "json_agg" | "jsonb_agg" | "json_object_agg" | "jsonb_object_agg"
| "json_agg_strict" | "jsonb_agg_strict"
| "json_object_agg_strict" | "jsonb_object_agg_strict"
| "json_object_agg_unique" | "jsonb_object_agg_unique"
| "json_object_agg_unique_strict" | "jsonb_object_agg_unique_strict"
| "json_arrayagg" | "json_objectagg"
)
}
fn is_regression_name(name: &str) -> bool {
matches!(
name,
"covar_pop"
| "covar_samp"
| "corr"
| "regr_count"
| "regr_avgx"
| "regr_avgy"
| "regr_slope"
| "regr_intercept"
| "regr_r2"
| "regr_sxx"
| "regr_syy"
| "regr_sxy"
)
}
fn agg_uses_second_arg(name: &str) -> bool {
name == "group_concat"
|| name == "string_agg"
|| name.starts_with("json_object_agg")
|| name.starts_with("jsonb_object_agg")
|| name == "jsonb_object_agg"
|| name == "json_objectagg"
|| is_regression_name(name)
}
pub fn is_ordered_set_name(name: &str) -> bool {
["percentile_cont", "percentile_disc", "mode"]
.iter()
.any(|k| name.eq_ignore_ascii_case(k))
}
pub fn is_hypothetical_set_name(name: &str) -> bool {
["rank", "dense_rank", "percent_rank", "cume_dist"]
.iter()
.any(|k| name.eq_ignore_ascii_case(k))
}
pub fn is_within_group_name(name: &str) -> bool {
is_ordered_set_name(name) || is_hypothetical_set_name(name)
}
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub(crate) enum AggKind {
CountStar,
Count,
Sum,
Avg,
Min,
Max,
AnyValue,
RangeAgg,
RangeIntersectAgg,
StringAgg,
ArrayAgg,
BoolAnd,
BoolOr,
StddevFamily,
BitAnd,
BitOr,
BitXor,
WithinGroup,
Regression,
JsonAgg,
JsonObjectAgg,
}
pub(crate) fn canonical_agg_name(name: &str) -> &str {
if name.eq_ignore_ascii_case("every") {
"bool_and"
} else {
name
}
}
pub(crate) fn classify_agg_name(name: &str) -> AggKind {
match name {
"count_star" => AggKind::CountStar,
"count" => AggKind::Count,
"sum" => AggKind::Sum,
"avg" => AggKind::Avg,
"min" => AggKind::Min,
"max" => AggKind::Max,
"any_value" => AggKind::AnyValue,
"range_agg" => AggKind::RangeAgg,
"range_intersect_agg" => AggKind::RangeIntersectAgg,
"string_agg" | "group_concat" | "xmlagg" => AggKind::StringAgg,
"array_agg" => AggKind::ArrayAgg,
"bool_and" => AggKind::BoolAnd,
"bool_or" => AggKind::BoolOr,
"stddev" | "stddev_samp" | "stddev_pop" | "variance" | "var_samp" | "var_pop" => {
AggKind::StddevFamily
}
"bit_and" => AggKind::BitAnd,
"bit_or" => AggKind::BitOr,
"bit_xor" => AggKind::BitXor,
"json_agg" | "jsonb_agg" | "json_arrayagg" | "json_agg_strict" | "jsonb_agg_strict" => {
AggKind::JsonAgg
}
"json_object_agg"
| "jsonb_object_agg"
| "json_objectagg"
| "json_object_agg_strict"
| "jsonb_object_agg_strict"
| "json_object_agg_unique"
| "jsonb_object_agg_unique"
| "json_object_agg_unique_strict"
| "jsonb_object_agg_unique_strict" => AggKind::JsonObjectAgg,
n if is_within_group_name(n) => AggKind::WithinGroup,
n if is_regression_name(n) => AggKind::Regression,
other => panic!("classify_agg_name: unknown aggregate {other}"),
}
}
#[allow(clippy::struct_excessive_bools)]
#[derive(Debug, Default, Clone)]
pub(crate) struct AggState {
num: NumAcc,
extreme: Option<Value<'static>>,
items: Vec<Value<'static>>,
item_seps: Vec<Option<String>>,
seen: BTreeSet<String>,
seen_int: Option<BTreeSet<i64>>,
item_keys: Vec<Value<'static>>,
separator: Option<String>,
bool_acc: Option<bool>,
sum_sq: f64,
stddev_saw_float: bool,
stddev_sum: Option<spg_storage::bignum::BigNumeric>,
stddev_sum_sq: Option<spg_storage::bignum::BigNumeric>,
stddev_i_sum: i128,
stddev_i_sum_sq: i128,
stddev_i_spent: bool,
bit_acc: Option<i64>,
bit_wide: bool,
within_group_rows: usize,
reg_n: i64,
reg_sx: f64,
reg_sy: f64,
reg_sxx: f64,
reg_syy: f64,
reg_sxy: f64,
aux_items: Vec<Value<'static>>,
first_best: Option<(Vec<Value<'static>>, Value<'static>)>,
}
#[derive(Debug, Clone)]
struct AggSpec {
name: String, arg: Option<Expr>,
arg2: Option<Expr>,
distinct: bool,
order_by: Vec<spg_sql::ast::OrderBy>,
filter: Option<Expr>,
direct_arg: Option<Expr>,
direct_args_extra: Vec<Expr>,
first_ordered: bool,
kind: AggKind,
enum_labels: Option<Vec<String>>,
arg_collation: Option<alloc::string::String>,
order_enum_labels: Vec<Option<Vec<String>>>,
}
#[derive(Debug)]
pub struct AggResult {
pub columns: Vec<ColumnSchema>,
pub rows: Vec<Row<'static>>,
pub deferred: Vec<(usize, Expr)>,
pub synth_rows: Vec<Row<'static>>,
pub synth_schema: Vec<ColumnSchema>,
}
#[allow(clippy::too_many_lines)]
pub type CorrelatedEval<'a> =
&'a dyn Fn(&Expr, &Row<'static>, &EvalContext<'_>) -> Result<Value<'static>, EvalError>;
struct Projection {
columns: Vec<ColumnSchema>,
out_rows: Vec<Row<'static>>,
kept_synth: Vec<Row<'static>>,
deferred: Vec<(usize, Expr)>,
order_rewritten: Vec<Expr>,
deferred_project: Option<DeferredProject>,
}
struct DeferredProject {
items_rewritten: Vec<Option<Expr>>,
items_compiled: Vec<Option<eval::CompiledExpr>>,
}
fn try_pure_count_star_short_circuit(
stmt: &SelectStatement,
rows: AggRows<'_>,
schema_cols: &[ColumnSchema],
table_alias: Option<&str>,
) -> Option<AggResult> {
if stmt.distinct
|| stmt.limit_with_ties
|| stmt.group_by.is_some()
|| stmt.having.is_some()
|| !stmt.order_by.is_empty()
{
return None;
}
if stmt.items.len() != 1 {
return None;
}
let SelectItem::Expr { expr, alias } = &stmt.items[0] else {
return None;
};
let Expr::FunctionCall { name, args } = expr else {
return None;
};
if !name.eq_ignore_ascii_case("count") && !name.eq_ignore_ascii_case("count_star") {
return None;
}
let count_star_shape = match args.as_slice() {
[] if name.eq_ignore_ascii_case("count_star") => true,
[Expr::Literal(lit)] => !matches!(lit, spg_sql::ast::Literal::Null),
[Expr::Column(c)] => {
if let Some(q) = c.qualifier.as_deref()
&& let Some(alias) = table_alias
&& !q.eq_ignore_ascii_case(alias)
{
return None;
}
schema_cols
.iter()
.find(|s| s.name.eq_ignore_ascii_case(&c.name))
.is_some_and(|s| !s.nullable)
}
_ => return None,
};
if !count_star_shape {
return None;
}
let col_name = alias.clone().unwrap_or_else(|| "count".to_string());
let count = i64::try_from(rows.len()).unwrap_or(i64::MAX);
Some(AggResult {
columns: alloc::vec![ColumnSchema::new(col_name, DataType::BigInt, false)],
rows: alloc::vec![Row::new(alloc::vec![Value::BigInt(count)])],
deferred: Vec::new(),
synth_rows: Vec::new(),
synth_schema: Vec::new(),
})
}
fn resolve_group_by_aliases(
keys: Vec<Expr>,
stmt: &SelectStatement,
schema_cols: &[ColumnSchema],
) -> Result<Vec<Expr>, EvalError> {
let mut out = Vec::with_capacity(keys.len());
for key in keys {
let Expr::Column(c) = &key else {
out.push(key);
continue;
};
if c.qualifier.is_some()
|| schema_cols
.iter()
.any(|sc| sc.name.eq_ignore_ascii_case(&c.name))
{
out.push(key);
continue;
}
let target = stmt.items.iter().find_map(|it| match it {
SelectItem::Expr {
expr,
alias: Some(a),
} if a.eq_ignore_ascii_case(&c.name) => Some(expr),
_ => None,
});
match target {
Some(e) if contains_aggregate(e) => {
return Err(EvalError::TypeMismatch {
detail: alloc::string::String::from(
"aggregate functions are not allowed in GROUP BY",
),
});
}
Some(e) => out.push(e.clone()),
None => out.push(key),
}
}
Ok(out)
}
pub(crate) fn run(
stmt: &SelectStatement,
rows: AggRows<'_>,
schema_cols: &[ColumnSchema],
table_alias: Option<&str>,
correlated_eval: Option<CorrelatedEval<'_>>,
runner: Option<&dyn crate::ParallelRunner>,
catalog: Option<&spg_storage::Catalog>,
engine: Option<&crate::Engine>,
) -> Result<AggResult, EvalError> {
let __spg_row_count = rows.len();
crate::injection_point!("aggregate_spill_trigger", &__spg_row_count);
if let Some(short) = try_pure_count_star_short_circuit(stmt, rows, schema_cols, table_alias) {
return Ok(short);
}
let group_exprs: Vec<Expr> = stmt.group_by.clone().unwrap_or_default();
let group_exprs = resolve_group_by_aliases(group_exprs, stmt, schema_cols)?;
let group_keys_all_resolve = group_exprs.iter().all(|g| match g {
Expr::Column(c) => {
c.qualifier.is_some()
|| schema_cols
.iter()
.any(|sc| sc.name.eq_ignore_ascii_case(&c.name))
}
_ => true,
});
let licensed = qualifiers_grouped_by_primary_key(stmt, &group_exprs, schema_cols, catalog);
let fd_on_primary_key = !licensed.is_empty();
if group_keys_all_resolve && !engine.is_some_and(|e| e.backslash_escapes) {
let offender = stmt
.items
.iter()
.find_map(|it| match it {
SelectItem::Expr { expr, .. } => {
first_ungrouped_column(expr, &group_exprs, schema_cols, &licensed)
}
_ => None,
})
.or_else(|| {
stmt.order_by.iter().find_map(|o| {
first_ungrouped_column(&o.expr, &group_exprs, schema_cols, &licensed)
})
})
.or_else(|| {
stmt.having
.as_ref()
.and_then(|h| first_ungrouped_column(h, &group_exprs, schema_cols, &licensed))
});
if let Some(c) = offender {
let qual = c
.qualifier
.as_deref()
.or(table_alias)
.or_else(|| stmt.from.as_ref().map(|f| f.primary.name.as_str()))
.unwrap_or("");
return Err(EvalError::TypeMismatch {
detail: alloc::format!(
"column \"{qual}.{}\" must appear in the GROUP BY clause or be used in an aggregate function",
c.name
),
});
}
}
let mysql_loose = engine.is_some_and(|e| e.backslash_escapes);
let loose_stmt;
let stmt = if (mysql_loose || fd_on_primary_key) && !group_exprs.is_empty() {
let claim: Option<&[alloc::string::String]> =
if mysql_loose { None } else { Some(&licensed) };
let mut s = stmt.clone();
for item in &mut s.items {
if let SelectItem::Expr { expr, .. } = item {
let taken = core::mem::replace(expr, Expr::Literal(spg_sql::ast::Literal::Null));
*expr = wrap_loose_group_columns(taken, &group_exprs, schema_cols, claim);
}
}
for o in &mut s.order_by {
let taken = core::mem::replace(&mut o.expr, Expr::Literal(spg_sql::ast::Literal::Null));
o.expr = wrap_loose_group_columns(taken, &group_exprs, schema_cols, claim);
}
if let Some(h) = s.having.take() {
s.having = Some(wrap_loose_group_columns(
h,
&group_exprs,
schema_cols,
claim,
));
}
loose_stmt = s;
&loose_stmt
} else {
stmt
};
let mut agg_specs: Vec<AggSpec> = Vec::new();
for item in &stmt.items {
if let SelectItem::Expr { expr, .. } = item {
collect_aggregates(expr, &mut agg_specs);
}
}
for o in &stmt.order_by {
collect_aggregates(&o.expr, &mut agg_specs);
}
if let Some(h) = &stmt.having {
collect_aggregates(h, &mut agg_specs);
}
validate_agg_arities(stmt, &agg_specs)?;
validate_within_group(&agg_specs, schema_cols, stmt.group_by.as_deref())?;
for spec in &mut agg_specs {
if matches!(spec.kind, AggKind::Min | AggKind::Max)
&& let Some(Expr::Column(c)) = &spec.arg
{
spec.arg_collation = schema_cols
.iter()
.find(|sc| sc.name.eq_ignore_ascii_case(&c.name))
.and_then(|sc| sc.collation_name.clone())
.filter(|n| crate::collate::is_supported(n));
}
}
if let Some(cat) = catalog
&& !cat.enum_types().is_empty()
{
for spec in &mut agg_specs {
if (matches!(spec.kind, AggKind::Min | AggKind::Max) || spec.distinct)
&& let Some(arg) = &spec.arg
{
spec.enum_labels = crate::eval::expr_enum_labels(arg, schema_cols, catalog)
.map(<[String]>::to_vec);
}
if !spec.order_by.is_empty() {
spec.order_enum_labels = spec
.order_by
.iter()
.map(|o| {
crate::eval::expr_enum_labels(&o.expr, schema_cols, catalog)
.map(<[String]>::to_vec)
})
.collect();
}
}
}
let order = accumulate_groups(
rows,
&group_exprs,
&agg_specs,
schema_cols,
table_alias,
correlated_eval,
runner,
catalog,
engine,
)?;
let synth_schema = build_synth_schema(
rows,
&group_exprs,
&agg_specs,
schema_cols,
table_alias,
catalog,
engine,
)?;
let synth_rows = finalize_synth_rows(
&order,
&agg_specs,
&synth_schema,
rows,
schema_cols,
table_alias,
catalog,
engine,
runner,
)?;
let any_srf_item = stmt.items.iter().any(|i| match i {
SelectItem::Expr { expr, .. } => crate::select::top_level_srf_kind(expr).is_some(),
_ => false,
});
let defer_projection = !stmt.order_by.is_empty()
&& !stmt.distinct
&& !stmt.limit_with_ties
&& !any_srf_item
&& stmt.limit_literal().is_some_and(|l| {
let off = stmt.offset_literal().unwrap_or(0) as usize;
let k = (l as usize).saturating_add(off);
k > 0 && k < synth_rows.len()
});
let Projection {
columns,
mut out_rows,
mut kept_synth,
deferred,
order_rewritten,
deferred_project,
} = project_groups(
synth_rows,
stmt,
&group_exprs,
&agg_specs,
&synth_schema,
correlated_eval,
defer_projection,
catalog,
engine.is_some_and(|e| e.backslash_escapes),
)?;
let keep_n: Option<usize> =
if !stmt.order_by.is_empty() && !stmt.distinct && !stmt.limit_with_ties {
stmt.limit_literal().map(|l| {
let off = stmt.offset_literal().unwrap_or(0) as usize;
(l as usize).saturating_add(off)
})
} else {
None
};
if !stmt.order_by.is_empty() {
let (sorted_synth, sorted_out) = sort_synth_by_order_by(
&synth_schema,
&columns,
&stmt.order_by,
&order_rewritten,
kept_synth,
out_rows,
correlated_eval,
keep_n,
catalog,
engine.is_some_and(|e| e.backslash_escapes),
)?;
kept_synth = sorted_synth;
out_rows = sorted_out;
}
if let Some(DeferredProject {
items_rewritten,
items_compiled,
}) = deferred_project
{
let mut synth_ctx = EvalContext::new(&synth_schema, None);
if let Some(cat) = catalog {
synth_ctx = synth_ctx.with_catalog(cat);
}
let mut stack: Vec<Value<'static>> = Vec::new();
for (idx, srow) in kept_synth.iter().enumerate() {
let mut values: Vec<Value<'static>> = Vec::with_capacity(columns.len());
for (i, rewritten) in items_rewritten.iter().enumerate() {
let Some(rewritten) = rewritten else { continue };
if deferred.iter().any(|(c, _)| *c == i) {
values.push(Value::Null);
continue;
}
values.push(if let Some(cc) = &items_compiled[i] {
eval::eval_compiled(cc, srow, &synth_ctx, &mut stack)?
} else {
match correlated_eval {
Some(f) if crate::expr_has_subquery(rewritten) => {
f(rewritten, srow, &synth_ctx)?
}
_ => eval::eval_expr(rewritten, srow, &synth_ctx)?,
}
});
}
out_rows[idx] = Row::new(values);
}
}
if stmt.distinct {
out_rows = crate::select::dedup_rows(out_rows, engine.is_some_and(|e| e.backslash_escapes));
}
let (synth_rows_out, synth_schema_out) = if deferred.is_empty() {
(Vec::new(), Vec::new())
} else {
(kept_synth, synth_schema.clone())
};
Ok(AggResult {
columns,
rows: out_rows,
deferred,
synth_rows: synth_rows_out,
synth_schema: synth_schema_out,
})
}
fn ordered_set_arg_type_name(e: &Expr, columns: &[ColumnSchema]) -> String {
if matches!(
e,
Expr::Literal(spg_sql::ast::Literal::String(_))
| Expr::Literal(spg_sql::ast::Literal::Null)
) {
return String::from("unknown");
}
match e {
Expr::Cast { .. } | Expr::Column(_) | Expr::Literal(_) => {
crate::describe::describe_expr(e, columns).map_or_else(
|| String::from("unknown"),
|s| crate::conversions::pg_type_name_for_error(s.ty),
)
}
_ => String::from("unknown"),
}
}
fn ordered_set_signature_error(name: &str, spec: &AggSpec, columns: &[ColumnSchema]) -> EvalError {
let mut parts: Vec<String> = Vec::new();
if let Some(d) = &spec.direct_arg {
parts.push(ordered_set_arg_type_name(d, columns));
}
for d in &spec.direct_args_extra {
parts.push(ordered_set_arg_type_name(d, columns));
}
for o in &spec.order_by {
parts.push(ordered_set_arg_type_name(&o.expr, columns));
}
EvalError::TypeMismatch {
detail: format!("function {name}({}) does not exist", parts.join(", ")),
}
}
fn validate_within_group(
agg_specs: &[AggSpec],
columns: &[ColumnSchema],
group_by: Option<&[Expr]>,
) -> Result<(), EvalError> {
fn first_ungrouped(e: &Expr, group_by: Option<&[Expr]>) -> Option<String> {
let mut found: Option<String> = None;
let mut subs: Vec<&SelectStatement> = Vec::new();
crate::visit_expr_columns_and_subqueries(
e,
&mut |c| {
if found.is_some() {
return;
}
let grouped = group_by.is_some_and(|gs| {
gs.iter().any(|g| match g {
Expr::Column(gc) => gc.name.eq_ignore_ascii_case(&c.name),
_ => false,
})
});
if !grouped && !c.name.is_empty() {
found = Some(match &c.qualifier {
Some(q) => format!("{q}.{}", c.name),
None => c.name.clone(),
});
}
},
&mut |s| subs.push(s),
);
found
}
for spec in agg_specs {
if !is_within_group_name(&spec.name) {
continue;
}
for d in spec.direct_arg.iter().chain(spec.direct_args_extra.iter()) {
if let Some(col) = first_ungrouped(d, group_by) {
return Err(EvalError::TypeMismatch {
detail: format!(
"column \"{col}\" must appear in the GROUP BY clause or be used in an aggregate function"
),
});
}
}
}
for spec in agg_specs {
if is_within_group_name(&spec.name) {
if spec.order_by.is_empty() {
if spec.direct_arg.is_none() && is_hypothetical_set_name(&spec.name) {
return Err(EvalError::TypeMismatch {
detail: format!("window function {} requires an OVER clause", spec.name),
});
}
return Err(EvalError::TypeMismatch {
detail: format!("{}() requires WITHIN GROUP (ORDER BY …)", spec.name),
});
}
if spec.name != "mode" && spec.direct_arg.is_none() {
return Err(EvalError::TypeMismatch {
detail: format!("{}() requires a direct argument", spec.name),
});
}
if spec.name == "mode" && spec.direct_arg.is_some() {
return Err(ordered_set_signature_error(&spec.name, spec, columns));
}
let hypothetical = matches!(
spec.name.as_str(),
"rank" | "dense_rank" | "percent_rank" | "cume_dist"
);
if hypothetical {
if 1 + spec.direct_args_extra.len() != spec.order_by.len() {
return Err(ordered_set_signature_error(&spec.name, spec, columns));
}
} else if spec.order_by.len() > 1 || !spec.direct_args_extra.is_empty() {
return Err(ordered_set_signature_error(&spec.name, spec, columns));
}
if spec.name == "percentile_cont"
&& let Some(o) = spec.order_by.first()
&& matches!(o.expr, Expr::Cast { .. } | Expr::Column(_))
&& let Some(sch) = crate::describe::describe_expr(&o.expr, columns)
&& !matches!(
sch.ty,
spg_storage::DataType::SmallInt
| spg_storage::DataType::Int
| spg_storage::DataType::BigInt
| spg_storage::DataType::Float
| spg_storage::DataType::Real
| spg_storage::DataType::Numeric { .. }
| spg_storage::DataType::Interval
)
{
return Err(ordered_set_signature_error(&spec.name, spec, columns));
}
}
}
Ok(())
}
#[derive(Debug, Default, Clone)]
struct NumAcc {
sum_int: i64,
sum_float: f64,
use_float: bool,
float_not_real: bool,
sum_num_scaled: i128,
sum_num_kind: spg_storage::NumericKind,
sum_num_scale: u16,
sum_big: SumBig,
use_numeric: bool,
sum_iv_months: i64,
sum_iv_days: i64,
sum_iv_micros: i128,
use_interval: bool,
sum_money: i128,
use_money: bool,
count: i64,
}
#[allow(clippy::too_many_lines, clippy::type_complexity)]
#[derive(Default, Clone)]
struct FusedAcc {
num: NumAcc,
extreme: Option<Value<'static>>,
extreme_max: bool,
extreme_mysql: bool,
extreme_coll: Option<alloc::string::String>,
items: Vec<Value<'static>>,
item_keys: Vec<Value<'static>>,
}
fn fused_accs(ops: &[FusedOp], mysql: bool) -> Vec<FusedAcc> {
ops.iter()
.map(|op| {
let mut a = FusedAcc::default();
if let FusedOp::Extreme { max, coll, .. } | FusedOp::ExtremeExpr { max, coll, .. } = op
{
a.extreme_max = *max;
a.extreme_mysql = mysql;
a.extreme_coll = coll.clone();
}
a
})
.collect()
}
enum FusedOp {
CountCol(usize),
AccCol(usize),
Extreme {
pos: usize,
max: bool,
coll: Option<alloc::string::String>,
},
CountExpr(usize),
AccExpr(usize),
ExtremeExpr {
spec: usize,
max: bool,
coll: Option<alloc::string::String>,
},
Collect {
spec: usize,
string_kind: bool,
},
}
fn fused_layout(
agg_specs: &[AggSpec],
arg_pos: &[Option<usize>],
arg_compiled: &[Option<eval::CompiledExpr>],
order_pos: &[Vec<Option<usize>>],
arg2_literal_val: &[Option<Value<'static>>],
) -> Option<(Vec<Option<usize>>, Vec<FusedOp>)> {
if agg_specs.is_empty() {
return None;
}
let has_arg = |i: usize| arg_pos[i].is_some() || arg_compiled[i].is_some();
let collectible = |i: usize, s: &AggSpec| -> bool {
!s.distinct
&& s.filter.is_none()
&& !s.first_ordered
&& arg_pos[i].is_some()
&& s.order_by
.iter()
.enumerate()
.all(|(k, _)| order_pos[i].get(k).copied().flatten().is_some())
&& match s.name.as_str() {
"string_agg" => matches!(&arg2_literal_val[i], Some(Value::Text(_))),
"array_agg" => s.arg2.is_none() && s.enum_labels.is_none(),
_ => false,
}
};
let eligible = agg_specs.iter().enumerate().all(|(i, s)| {
collectible(i, s)
|| (s.filter.is_none()
&& s.arg2.is_none()
&& s.order_by.is_empty()
&& !s.distinct
&& !s.first_ordered
&& match s.name.as_str() {
"count_star" => s.arg.is_none(),
"count" | "sum" | "avg" => has_arg(i),
"min" | "max" => has_arg(i) && s.enum_labels.is_none(),
_ => false,
})
});
if !eligible {
return None;
}
let mut unique_ops: Vec<FusedOp> = Vec::new();
let same_arg = |j: usize, i: usize| agg_specs[j].arg == agg_specs[i].arg;
let spec_src: Vec<Option<usize>> = agg_specs
.iter()
.enumerate()
.map(|(i, s)| match s.name.as_str() {
"count_star" => None,
"string_agg" | "array_agg" => {
unique_ops.push(FusedOp::Collect {
spec: i,
string_kind: s.name.as_str() == "string_agg",
});
Some(unique_ops.len() - 1)
}
"min" | "max" => {
let max = s.name.as_str() == "max";
let slot = if let Some(p) = arg_pos[i] {
unique_ops
.iter()
.position(|o| {
matches!(o, FusedOp::Extreme { pos, max: m, coll }
if *pos == p && *m == max && *coll == s.arg_collation)
})
.unwrap_or_else(|| {
unique_ops.push(FusedOp::Extreme {
pos: p,
max,
coll: s.arg_collation.clone(),
});
unique_ops.len() - 1
})
} else {
unique_ops
.iter()
.position(|o| {
matches!(o, FusedOp::ExtremeExpr { spec, max: m, coll }
if same_arg(*spec, i) && *m == max && *coll == s.arg_collation)
})
.unwrap_or_else(|| {
unique_ops.push(FusedOp::ExtremeExpr {
spec: i,
max,
coll: s.arg_collation.clone(),
});
unique_ops.len() - 1
})
};
Some(slot)
}
"count" => {
let slot = if let Some(p) = arg_pos[i] {
unique_ops
.iter()
.position(|o| matches!(o, FusedOp::CountCol(q) if *q == p))
.unwrap_or_else(|| {
unique_ops.push(FusedOp::CountCol(p));
unique_ops.len() - 1
})
} else {
unique_ops
.iter()
.position(|o| matches!(o, FusedOp::CountExpr(j) if same_arg(*j, i)))
.unwrap_or_else(|| {
unique_ops.push(FusedOp::CountExpr(i));
unique_ops.len() - 1
})
};
Some(slot)
}
_ => {
let slot = if let Some(p) = arg_pos[i] {
unique_ops
.iter()
.position(|o| matches!(o, FusedOp::AccCol(q) if *q == p))
.unwrap_or_else(|| {
unique_ops.push(FusedOp::AccCol(p));
unique_ops.len() - 1
})
} else {
unique_ops
.iter()
.position(|o| matches!(o, FusedOp::AccExpr(j) if same_arg(*j, i)))
.unwrap_or_else(|| {
unique_ops.push(FusedOp::AccExpr(i));
unique_ops.len() - 1
})
};
Some(slot)
}
})
.collect();
Some((spec_src, unique_ops))
}
fn merge_fused(a: &mut FusedAcc, b: &mut FusedAcc) {
if let Some(be) = &b.extreme {
let take = match &a.extreme {
None => true,
Some(ae) => {
let ord = extreme_cmp_in(None, a.extreme_coll.as_deref(), be, ae, a.extreme_mysql);
if a.extreme_max {
ord == core::cmp::Ordering::Greater
} else {
ord == core::cmp::Ordering::Less
}
}
};
if take {
a.extreme = Some(be.clone());
}
}
a.num.count += b.num.count;
a.num.sum_int += b.num.sum_int;
a.num.sum_float += b.num.sum_float;
a.num.use_float |= b.num.use_float;
a.num.float_not_real |= b.num.float_not_real;
if b.num.use_numeric {
if let Some(bb) = &b.num.sum_big {
sum_add_bignum(
&mut a.num.sum_num_scaled,
&mut a.num.sum_num_scale,
&mut a.num.sum_big,
bb,
);
}
sum_add_exact(
&mut a.num.sum_num_scaled,
&mut a.num.sum_num_scale,
&mut a.num.sum_big,
b.num.sum_num_scaled,
b.num.sum_num_scale,
);
a.num.sum_num_kind = fold_sum_kind(a.num.sum_num_kind, b.num.sum_num_kind);
a.num.use_numeric = true;
}
a.num.sum_iv_months += b.num.sum_iv_months;
a.num.sum_iv_days += b.num.sum_iv_days;
a.num.sum_iv_micros += b.num.sum_iv_micros;
a.num.use_interval |= b.num.use_interval;
a.num.sum_money += b.num.sum_money;
a.num.use_money |= b.num.use_money;
a.items.extend(core::mem::take(&mut b.items));
a.item_keys.extend(core::mem::take(&mut b.item_keys));
}
fn collect_cell(
a: &mut FusedAcc,
row: &crate::join::RowRef<'_>,
pos: usize,
key_pos: &[Option<usize>],
string_kind: bool,
) -> Result<(), EvalError> {
let v = row.get(pos).unwrap_or(&Value::Null);
if string_kind {
if matches!(v, Value::Null) {
return Ok(());
}
let Some(item) = render_string_agg_item(v) else {
return Err(EvalError::TypeMismatch {
detail: format!(
"string_agg requires text value, got {}",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
});
};
a.items.push(item);
} else {
a.items.push(v.clone().into_owned());
}
a.num.count += 1;
for kp in key_pos {
let kv = row
.get(kp.expect("layout-gated bound key"))
.cloned()
.map(Value::into_owned)
.unwrap_or(Value::Null);
a.item_keys.push(kv);
}
Ok(())
}
fn render_string_agg_item(v: &Value<'_>) -> Option<Value<'static>> {
match v {
Value::Text(s) => Some(Value::text(s.clone())),
Value::BpChar(s) => Some(Value::text(s.trim_end_matches(' ').to_string())),
Value::Xml(s) => Some(Value::text(s.to_string())),
Value::Int(n) => Some(Value::text(n.to_string())),
Value::BigInt(n) => Some(Value::text(n.to_string())),
Value::SmallInt(n) => Some(Value::text(n.to_string())),
Value::Float(f) => Some(Value::text(f.to_string())),
Value::Bool(b) => Some(Value::text(if *b { "1" } else { "0" })),
_ => None,
}
}
fn fill_states_from_fused(
states: &mut [AggState],
spec_src: &[Option<usize>],
accs: &mut [FusedAcc],
group_rows: i64,
arg2_literal_val: &[Option<Value<'static>>],
) {
for (i, src) in spec_src.iter().enumerate() {
let state = &mut states[i];
match src {
None => state.num.count = group_rows,
Some(slot) => {
{
let a = &mut accs[*slot];
if !a.items.is_empty() {
state.items = core::mem::take(&mut a.items);
state.item_keys = core::mem::take(&mut a.item_keys);
}
}
if let Some(Value::Text(sep)) = &arg2_literal_val[i] {
state.separator = Some(sep.to_string());
}
let a = &accs[*slot];
state.num.count = a.num.count;
state.num.sum_int = a.num.sum_int;
state.num.sum_float = a.num.sum_float;
state.num.use_float = a.num.use_float;
state.num.float_not_real = a.num.float_not_real;
state.num.sum_num_scaled = a.num.sum_num_scaled;
state.num.sum_num_kind = a.num.sum_num_kind;
state.num.sum_num_scale = a.num.sum_num_scale;
state.num.sum_big = a.num.sum_big.clone();
state.num.use_numeric = a.num.use_numeric;
state.num.sum_iv_months = a.num.sum_iv_months;
state.num.sum_iv_days = a.num.sum_iv_days;
state.num.sum_iv_micros = a.num.sum_iv_micros;
state.num.use_interval = a.num.use_interval;
state.num.sum_money = a.num.sum_money;
state.num.use_money = a.num.use_money;
if a.extreme.is_some() {
state.extreme = a.extreme.clone();
}
}
}
}
}
type SumBig = Option<alloc::boxed::Box<spg_storage::bignum::BigNumeric>>;
fn sum_add_exact(
scaled: &mut i128,
scale: &mut u16,
big: &mut SumBig,
add_scaled: i128,
add_scale: u16,
) {
use spg_storage::bignum::BigNumeric;
if let Some(b) = big {
**b = b.add(&BigNumeric::from_i128(add_scaled, add_scale));
return;
}
match crate::numeric::numeric_add_checked(*scaled, *scale, add_scaled, add_scale) {
Some((s, sc)) => {
*scaled = s;
*scale = sc;
}
None => {
*big = Some(alloc::boxed::Box::new(
BigNumeric::from_i128(*scaled, *scale)
.add(&BigNumeric::from_i128(add_scaled, add_scale)),
));
*scaled = 0;
*scale = 0;
}
}
}
fn sum_add_bignum(
scaled: &mut i128,
scale: &mut u16,
big: &mut SumBig,
b_in: &spg_storage::bignum::BigNumeric,
) {
use spg_storage::bignum::BigNumeric;
let cur = match big.take() {
Some(b) => *b,
None => {
let c = BigNumeric::from_i128(*scaled, *scale);
*scaled = 0;
*scale = 0;
c
}
};
*big = Some(alloc::boxed::Box::new(cur.add(b_in)));
}
#[inline]
fn fused_extreme_cell(a: &mut FusedAcc, v: &Value<'_>, max: bool) -> Result<(), EvalError> {
if matches!(v, Value::Null) {
return Ok(());
}
if !a.extreme_mysql && min_max_unsupported_type(v) {
return Err(EvalError::TypeMismatch {
detail: format!(
"function {}({}) does not exist",
if max { "max" } else { "min" },
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
});
}
let take = match &a.extreme {
None => true,
Some(prev) => {
let ord = extreme_cmp_in(None, a.extreme_coll.as_deref(), v, prev, a.extreme_mysql);
if max {
ord == core::cmp::Ordering::Greater
} else {
ord == core::cmp::Ordering::Less
}
}
};
if take {
a.extreme = Some(v.clone().into_owned());
}
Ok(())
}
fn min_max_unsupported_type(v: &Value<'_>) -> bool {
matches!(
v.data_type(),
Some(
spg_storage::DataType::Bool
| spg_storage::DataType::Uuid
| spg_storage::DataType::Macaddr
| spg_storage::DataType::Macaddr8
| spg_storage::DataType::Json
| spg_storage::DataType::Jsonb
| spg_storage::DataType::Bit(_)
| spg_storage::DataType::BitVarying(_)
| spg_storage::DataType::Xml
| spg_storage::DataType::TsVector
| spg_storage::DataType::TsQuery
| spg_storage::DataType::Xid
)
)
}
fn acc_cell(a: &mut NumAcc, v: &Value<'_>) -> Result<(), EvalError> {
match v {
Value::Null => {}
Value::SmallInt(n) => {
a.sum_int += i64::from(*n);
a.count += 1;
}
Value::Int(n) => {
a.sum_int += i64::from(*n);
a.count += 1;
}
Value::BigInt(n) => {
sum_add_exact(
&mut a.sum_num_scaled,
&mut a.sum_num_scale,
&mut a.sum_big,
i128::from(*n),
0,
);
a.use_numeric = true;
a.count += 1;
}
Value::Float(x) => {
a.sum_float += *x;
a.use_float = true;
a.float_not_real = true;
a.count += 1;
}
Value::Real(x) => {
a.sum_float += f64::from(*x);
a.use_float = true;
a.count += 1;
}
Value::Numeric {
scaled,
scale,
kind,
} => {
sum_add_exact(
&mut a.sum_num_scaled,
&mut a.sum_num_scale,
&mut a.sum_big,
*scaled,
*scale,
);
a.sum_num_kind = fold_sum_kind(a.sum_num_kind, *kind);
a.use_numeric = true;
a.count += 1;
}
Value::NumericBig(b) => {
sum_add_bignum(
&mut a.sum_num_scaled,
&mut a.sum_num_scale,
&mut a.sum_big,
b,
);
a.use_numeric = true;
a.count += 1;
}
Value::Interval {
months,
days,
micros,
} => {
a.sum_iv_months += i64::from(*months);
a.sum_iv_days += i64::from(*days);
a.sum_iv_micros += i128::from(*micros);
a.use_interval = true;
a.count += 1;
}
Value::Money(c) => {
a.sum_money += i128::from(*c);
a.use_money = true;
a.count += 1;
}
other => {
return Err(EvalError::TypeMismatch {
detail: format!(
"sum/avg need numeric, got {}",
crate::conversions::pg_type_name_for_error_opt(other.data_type())
),
});
}
}
Ok(())
}
fn with_catalog<'a>(
ctx: EvalContext<'a>,
catalog: Option<&'a spg_storage::Catalog>,
engine: Option<&'a crate::Engine>,
) -> EvalContext<'a> {
let ctx = match catalog {
Some(c) => ctx.with_catalog(c),
None => ctx,
};
match engine {
Some(e) => ctx.with_engine(e),
None => ctx,
}
}
fn accumulate_groups(
rows: AggRows<'_>,
group_exprs: &[Expr],
agg_specs: &[AggSpec],
schema_cols: &[ColumnSchema],
table_alias: Option<&str>,
correlated_eval: Option<CorrelatedEval<'_>>,
runner: Option<&dyn crate::ParallelRunner>,
catalog: Option<&spg_storage::Catalog>,
engine: Option<&crate::Engine>,
) -> Result<Vec<(Vec<Value<'static>>, Vec<AggState>)>, EvalError> {
let ctx = with_catalog(EvalContext::new(schema_cols, table_alias), catalog, engine);
let mut order: Vec<(Vec<Value<'static>>, Vec<AggState>)> = Vec::new();
let mut groups: hashbrown::HashMap<String, usize> = hashbrown::HashMap::new();
let mut groups_text: hashbrown::HashMap<String, usize> = hashbrown::HashMap::new();
let mut groups_int: hashbrown::HashMap<i64, usize> = hashbrown::HashMap::new();
let mut null_group_idx: Option<usize> = None;
if rows.is_empty() && group_exprs.is_empty() {
let init: Vec<AggState> = (0..agg_specs.len()).map(|_| AggState::default()).collect();
order.push((Vec::new(), init));
}
let col_pos = |e: &Expr| -> Option<usize> {
if let Expr::Column(c) = e {
eval::compile_column_pos(c, &ctx)
} else {
None
}
};
let group_pos: Vec<Option<usize>> = group_exprs.iter().map(col_pos).collect();
let all_groups_bound = group_pos.iter().all(Option::is_some);
let single_text_group_col: bool = !ctx.mysql_dialect
&& group_pos.len() == 1
&& group_pos[0].is_some_and(|p| {
schema_cols
.get(p)
.is_some_and(|c| matches!(c.ty, spg_storage::DataType::Text))
});
let single_int_group_col: bool = group_pos.len() == 1
&& group_pos[0].is_some_and(|p| {
schema_cols.get(p).is_some_and(|c| {
matches!(
c.ty,
spg_storage::DataType::SmallInt
| spg_storage::DataType::Int
| spg_storage::DataType::BigInt
)
})
});
let arg_pos: Vec<Option<usize>> = agg_specs
.iter()
.map(|spec| spec.arg.as_ref().and_then(|e| col_pos(e)))
.collect();
let is_binary_key_col = |p: Option<usize>| -> bool {
p.and_then(|i| schema_cols.get(i))
.is_some_and(|c| matches!(c.collation, spg_storage::Collation::Binary))
};
let mysql_fold_groups: bool = ctx.mysql_dialect
&& !group_pos.iter().any(|&p| is_binary_key_col(p))
&& !group_exprs
.iter()
.any(|e| crate::eval::is_binary_coerced(e));
let distinct_fold: Vec<bool> = agg_specs
.iter()
.enumerate()
.map(|(i, spec)| {
ctx.mysql_dialect
&& !is_binary_key_col(arg_pos[i])
&& !spec
.arg
.as_ref()
.is_some_and(|e| crate::eval::is_binary_coerced(e))
})
.collect();
let dedicated_max_loop: bool = single_text_group_col
&& agg_specs.len() == 1
&& matches!(agg_specs[0].kind, AggKind::Max)
&& agg_specs[0].filter.is_none()
&& agg_specs[0].arg2.is_none()
&& agg_specs[0].order_by.is_empty()
&& !agg_specs[0].distinct
&& !agg_specs[0].first_ordered
&& arg_pos[0].is_some();
let arg_compiled: Vec<Option<eval::CompiledExpr>> = agg_specs
.iter()
.enumerate()
.map(|(i, spec)| match (&arg_pos[i], &spec.arg) {
(Some(_), _) => None,
(None, Some(e)) if eval::fully_compilable(e) => Some(eval::compile_expr(e, &ctx)),
_ => None,
})
.collect();
let mut arg_unique_idx: Vec<usize> = Vec::new();
let mut arg_slot: Vec<Option<usize>> = Vec::with_capacity(agg_specs.len());
arg_slot.resize(agg_specs.len(), None);
for (i, spec) in agg_specs.iter().enumerate() {
if arg_pos[i].is_some() || arg_compiled[i].is_none() {
continue;
}
let src = spec.arg.as_ref().expect("arg_compiled => spec.arg is Some");
let pos = arg_unique_idx
.iter()
.position(|&j| agg_specs[j].arg.as_ref().is_some_and(|other| other == src));
arg_slot[i] = Some(match pos {
Some(p) => p,
None => {
arg_unique_idx.push(i);
arg_unique_idx.len() - 1
}
});
}
let mut row_eval_cache: Vec<Option<Value>> = Vec::with_capacity(arg_unique_idx.len());
row_eval_cache.resize(arg_unique_idx.len(), None);
let order_pos: Vec<Vec<Option<usize>>> = agg_specs
.iter()
.map(|spec| spec.order_by.iter().map(|o| col_pos(&o.expr)).collect())
.collect();
let arg2_literal_val: Vec<Option<Value<'static>>> = agg_specs
.iter()
.map(|s| match &s.arg2 {
Some(Expr::Literal(l)) => Some(eval::literal_to_value(l)),
_ => None,
})
.collect();
let needs_mat = agg_specs.iter().enumerate().any(|(i, s)| {
s.filter.is_some()
|| (s.arg.is_some() && arg_pos[i].is_none() && arg_compiled[i].is_none())
|| (s.arg2.is_some() && arg2_literal_val[i].is_none())
|| order_pos[i].iter().any(Option::is_none)
});
let ci_positions: Vec<usize> = group_exprs
.iter()
.enumerate()
.filter(|(_, g)| {
matches!(
eval::column_collation(g, &ctx),
Some(spg_storage::Collation::CaseInsensitive)
)
})
.map(|(i, _)| i)
.collect();
let mut keybuf_s = String::new();
let mut eval_stack: Vec<Value<'_>> = Vec::new();
let mut dkeybuf = String::new();
let mut refs: Vec<&Value> = Vec::with_capacity(group_pos.len());
let any_agg_subquery = correlated_eval.is_some()
&& agg_specs.iter().any(|s| {
s.filter
.as_ref()
.is_some_and(|e| crate::expr_has_subquery(e))
|| s.arg.as_ref().is_some_and(|e| crate::expr_has_subquery(e))
|| s.arg2.as_ref().is_some_and(|e| crate::expr_has_subquery(e))
|| s.order_by.iter().any(|o| crate::expr_has_subquery(&o.expr))
});
let eval_arg =
|e: &Expr, r: &Row<'static>, c: &EvalContext<'_>| -> Result<Value<'static>, EvalError> {
match correlated_eval {
Some(f) if any_agg_subquery && crate::expr_has_subquery(e) => f(e, r, c),
_ => eval::eval_expr(e, r, c),
}
};
let single_anon_group = group_exprs.is_empty() && !rows.is_empty();
if single_anon_group {
let init: Vec<AggState> = (0..agg_specs.len()).map(|_| AggState::default()).collect();
order.clear();
order.push((Vec::new(), init));
}
if single_anon_group
&& agg_specs.len() == 1
&& agg_specs[0].name == "count_star"
&& agg_specs[0].filter.is_none()
&& agg_specs[0].arg.is_none()
&& agg_specs[0].arg2.is_none()
&& agg_specs[0].order_by.is_empty()
&& !agg_specs[0].distinct
{
let state = &mut order[0].1[0];
state.num.count = rows.len() as i64;
return Ok(order);
}
if single_anon_group
&& let Some((spec_src, unique_ops)) = fused_layout(
agg_specs,
&arg_pos,
&arg_compiled,
&order_pos,
&arg2_literal_val,
)
{
let mut accs: Vec<FusedAcc> = fused_accs(&unique_ops, ctx.mysql_dialect);
let fused_scan = |range: core::ops::Range<usize>,
accs: &mut Vec<FusedAcc>,
fctx: &EvalContext<'_>|
-> Result<(), EvalError> {
let mut stack: Vec<Value<'_>> = Vec::new();
for row in rows.range(range.start, range.end).iter() {
for (si, op) in unique_ops.iter().enumerate() {
match op {
FusedOp::CountCol(p) => {
if !matches!(row.get(*p), Some(Value::Null) | None) {
accs[si].num.count += 1;
}
}
FusedOp::AccCol(p) => {
{
let a = &mut accs[si];
acc_cell(&mut a.num, row.get(*p).unwrap_or(&Value::Null))
}?;
}
FusedOp::Extreme { pos, max, .. } => {
fused_extreme_cell(
&mut accs[si],
row.get(*pos).unwrap_or(&Value::Null),
*max,
)?;
}
FusedOp::CountExpr(sp) => {
let c = arg_compiled[*sp].as_ref().expect("gated compiled");
let v = eval::eval_compiled_ref(c, row, fctx, &mut stack)?;
if !matches!(v, Value::Null) {
accs[si].num.count += 1;
}
}
FusedOp::AccExpr(sp) => {
let c = arg_compiled[*sp].as_ref().expect("gated compiled");
let v = eval::eval_compiled_ref(c, row, fctx, &mut stack)?;
acc_cell(&mut accs[si].num, &v)?;
}
FusedOp::ExtremeExpr { spec, max, .. } => {
let c = arg_compiled[*spec].as_ref().expect("gated compiled");
let v = eval::eval_compiled_ref(c, row, fctx, &mut stack)?;
fused_extreme_cell(&mut accs[si], &v, *max)?;
}
FusedOp::Collect { spec, string_kind } => {
collect_cell(
&mut accs[si],
&row,
arg_pos[*spec].expect("gated bound"),
&order_pos[*spec],
*string_kind,
)?;
}
}
}
}
Ok(())
};
if !unique_ops.is_empty() {
let par = runner.filter(|_| rows.len() >= crate::PARALLEL_MIN_ROWS);
if let Some(r) = par {
crate::PARALLEL_AGG_FIRED.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
let n_shards = (rows.len() / crate::PARALLEL_MIN_ROWS).clamp(2, 8);
let chunk = rows.len().div_ceil(n_shards);
type ShardOut = Result<Vec<FusedAcc>, EvalError>;
let ops = &unique_ops;
let mysql_for_accs = ctx.mysql_dialect;
let outer_style = ctx.render_style;
let results = r.run_shards(n_shards, &|i| {
let lo = i * chunk;
let hi = ((i + 1) * chunk).min(rows.len());
let mut local: Vec<FusedAcc> = fused_accs(ops, mysql_for_accs);
let mut sctx = EvalContext::new(schema_cols, table_alias);
sctx.mysql_dialect = mysql_for_accs;
sctx.render_style = outer_style;
let sctx = match catalog {
Some(c) => sctx.with_catalog(c),
None => sctx,
};
let out: ShardOut = fused_scan(lo..hi, &mut local, &sctx).map(|()| local);
alloc::boxed::Box::new(out)
});
for boxed in results {
let shard = boxed
.downcast::<ShardOut>()
.expect("runner echoes the closure's box");
let mut shard_accs = (*shard)?;
for (si, b) in shard_accs.iter_mut().enumerate() {
merge_fused(&mut accs[si], b);
}
}
} else {
fused_scan(0..rows.len(), &mut accs, &ctx)?;
}
}
fill_states_from_fused(
&mut order[0].1,
&spec_src,
&mut accs,
rows.len() as i64,
&arg2_literal_val,
);
return Ok(order);
}
if single_int_group_col
&& group_exprs.len() == 1
&& rows.len() >= crate::PARALLEL_MIN_ROWS
&& let Some(r) = runner
&& let Some((spec_src, unique_ops)) = fused_layout(
agg_specs,
&arg_pos,
&arg_compiled,
&order_pos,
&arg2_literal_val,
)
&& !unique_ops.is_empty()
{
crate::PARALLEL_AGG_FIRED.fetch_add(1, core::sync::atomic::Ordering::Relaxed);
let gp = group_pos[0].expect("single_int_group_col implies bound");
struct ShardMap {
keys: Vec<(i64, Value<'static>)>,
slots: hashbrown::HashMap<i64, Vec<FusedAcc>>,
null_slot: Option<Vec<FusedAcc>>,
null_rows: i64,
key_rows: hashbrown::HashMap<i64, i64>,
}
type ShardOut = Result<ShardMap, Option<EvalError>>;
let n_shards = (rows.len() / crate::PARALLEL_MIN_ROWS).clamp(2, 8);
let chunk = rows.len().div_ceil(n_shards);
let ops = &unique_ops;
let mysql_for_accs = ctx.mysql_dialect;
let outer_style = ctx.render_style;
let results = r.run_shards(n_shards, &|si| {
let lo = si * chunk;
let hi = ((si + 1) * chunk).min(rows.len());
let mut m = ShardMap {
keys: Vec::new(),
slots: hashbrown::HashMap::new(),
null_slot: None,
null_rows: 0,
key_rows: hashbrown::HashMap::new(),
};
let out: ShardOut = (|| {
let mut stack: Vec<Value<'_>> = Vec::new();
let mut sctx = EvalContext::new(schema_cols, table_alias);
sctx.mysql_dialect = mysql_for_accs;
sctx.render_style = outer_style;
let sctx = match catalog {
Some(c) => sctx.with_catalog(c),
None => sctx,
};
for row in rows.range(lo, hi).iter() {
let v = row.get(gp).unwrap_or(&Value::Null);
let key: Option<i64> = match v {
Value::SmallInt(n) => Some(i64::from(*n)),
Value::Int(n) => Some(i64::from(*n)),
Value::BigInt(n) => Some(*n),
Value::Null => None,
_ => return Err(None), };
let slots = match key {
Some(k) => {
*m.key_rows.entry(k).or_insert(0) += 1;
m.slots.entry(k).or_insert_with(|| {
m.keys.push((k, v.clone().into_owned()));
fused_accs(ops, mysql_for_accs)
})
}
None => {
m.null_rows += 1;
m.null_slot
.get_or_insert_with(|| fused_accs(ops, mysql_for_accs))
}
};
for (oi, op) in ops.iter().enumerate() {
match op {
FusedOp::CountCol(p) => {
if !matches!(row.get(*p), Some(Value::Null) | None) {
slots[oi].num.count += 1;
}
}
FusedOp::AccCol(p) => {
{
let a = &mut slots[oi];
acc_cell(&mut a.num, row.get(*p).unwrap_or(&Value::Null))
}
.map_err(Some)?;
}
FusedOp::Extreme { pos, max, .. } => {
fused_extreme_cell(
&mut slots[oi],
row.get(*pos).unwrap_or(&Value::Null),
*max,
)
.map_err(Some)?;
}
FusedOp::CountExpr(sp) => {
let c = arg_compiled[*sp].as_ref().expect("gated compiled");
let v = eval::eval_compiled_ref(c, row, &sctx, &mut stack)
.map_err(Some)?;
if !matches!(v, Value::Null) {
slots[oi].num.count += 1;
}
}
FusedOp::AccExpr(sp) => {
let c = arg_compiled[*sp].as_ref().expect("gated compiled");
let v = eval::eval_compiled_ref(c, row, &sctx, &mut stack)
.map_err(Some)?;
acc_cell(&mut slots[oi].num, &v).map_err(Some)?;
}
FusedOp::ExtremeExpr { spec, max, .. } => {
let c = arg_compiled[*spec].as_ref().expect("gated compiled");
let v = eval::eval_compiled_ref(c, row, &sctx, &mut stack)
.map_err(Some)?;
fused_extreme_cell(&mut slots[oi], &v, *max).map_err(Some)?;
}
FusedOp::Collect { spec, string_kind } => {
collect_cell(
&mut slots[oi],
&row,
arg_pos[*spec].expect("gated bound"),
&order_pos[*spec],
*string_kind,
)
.map_err(Some)?;
}
}
}
}
Ok(m)
})();
alloc::boxed::Box::new(out)
});
let mut merged_keys: Vec<(i64, Value<'static>)> = Vec::new();
let mut merged: hashbrown::HashMap<i64, (Vec<FusedAcc>, i64)> = hashbrown::HashMap::new();
let mut merged_null: Option<(Vec<FusedAcc>, i64)> = None;
let mut fallback = false;
let mut shard_err: Option<EvalError> = None;
for boxed in results {
let shard = boxed
.downcast::<ShardOut>()
.expect("runner echoes the closure's box");
match *shard {
Ok(mut m) => {
for (k, kv) in m.keys {
let mut accs = m.slots.remove(&k).expect("keyed slot");
let rows_k = m.key_rows[&k];
match merged.get_mut(&k) {
Some((dst, cnt)) => {
for (i, b) in accs.iter_mut().enumerate() {
merge_fused(&mut dst[i], b);
}
*cnt += rows_k;
}
None => {
merged_keys.push((k, kv));
merged.insert(k, (accs, rows_k));
}
}
}
if let Some(mut nb) = m.null_slot.take() {
match &mut merged_null {
Some((dst, cnt)) => {
for (i, b) in nb.iter_mut().enumerate() {
merge_fused(&mut dst[i], b);
}
*cnt += m.null_rows;
}
None => merged_null = Some((nb, m.null_rows)),
}
}
}
Err(None) => fallback = true,
Err(Some(e)) => shard_err = Some(e),
}
}
if let Some(e) = shard_err {
return Err(e);
}
if !fallback {
for (k, kv) in merged_keys {
let (mut accs, group_rows) = merged.remove(&k).expect("key recorded");
let mut states: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
fill_states_from_fused(
&mut states,
&spec_src,
&mut accs,
group_rows,
&arg2_literal_val,
);
order.push((alloc::vec![kv], states));
}
if let Some((mut accs, group_rows)) = merged_null {
let mut states: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
fill_states_from_fused(
&mut states,
&spec_src,
&mut accs,
group_rows,
&arg2_literal_val,
);
order.push((alloc::vec![Value::Null], states));
}
return Ok(order);
}
}
if single_anon_group
&& agg_specs.len() == 1
&& agg_specs[0].name == "count"
&& agg_specs[0].filter.is_none()
&& agg_specs[0].arg2.is_none()
&& agg_specs[0].order_by.is_empty()
&& !agg_specs[0].distinct
&& arg_pos[0].is_some()
{
let p = arg_pos[0].unwrap();
let mut count: i64 = 0;
for row in rows.iter() {
if !matches!(row.get(p), Some(Value::Null) | None) {
count += 1;
}
}
let state = &mut order[0].1[0];
state.num.count = count;
return Ok(order);
}
if single_anon_group
&& agg_specs.len() == 1
&& agg_specs[0].filter.is_none()
&& agg_specs[0].arg2.is_none()
&& agg_specs[0].order_by.is_empty()
&& !agg_specs[0].distinct
&& (agg_specs[0].name == "sum" || agg_specs[0].name == "avg")
&& (arg_pos[0].is_some() || arg_compiled[0].is_some())
{
let arg_pos0 = arg_pos[0];
let arg_c0 = &arg_compiled[0];
let mut na = NumAcc::default();
if let Some(p) = arg_pos0 {
for row in rows.iter() {
let v_ref = row.get(p).unwrap_or(&Value::Null);
acc_cell(&mut na, v_ref)?;
}
} else if let Some(p) = arg_c0.as_ref().and_then(|c| c.as_single_column_length()) {
for row in rows.iter() {
let Some(v_ref) = row.get(p) else {
continue;
};
let n = match v_ref {
Value::Null => continue,
Value::Text(s) => {
if s.is_ascii() {
s.len() as i64
} else {
s.chars().count() as i64
}
}
other => {
return Err(EvalError::TypeMismatch {
detail: format!(
"length() needs text, got {}",
crate::conversions::pg_type_name_for_error_opt(other.data_type())
),
});
}
};
na.sum_int += n;
na.count += 1;
}
} else {
let c = arg_c0.as_ref().unwrap();
for row in rows.iter() {
let v = eval::eval_compiled_ref(c, row, &ctx, &mut eval_stack)?;
acc_cell(&mut na, &v)?;
}
}
let state = &mut order[0].1[0];
state.num = na;
return Ok(order);
}
if dedicated_max_loop && !single_anon_group {
let gpos = group_pos[0].expect("dedicated_max_loop gates on Some");
let apos = arg_pos[0].expect("dedicated_max_loop gates on Some");
for row in rows.iter() {
let kv = row.get(gpos).unwrap_or(&Value::Null);
let idx = match kv {
Value::Text(s) => match groups_text.get(s.as_ref()) {
Some(&i) => i,
None => {
let i = order.len();
order.push((
alloc::vec![Value::text(s.clone())],
alloc::vec![AggState::default()],
));
groups_text.insert(s.to_string(), i);
i
}
},
Value::Null => match null_group_idx {
Some(i) => i,
None => {
let i = order.len();
order.push((alloc::vec![Value::Null], alloc::vec![AggState::default()]));
null_group_idx = Some(i);
i
}
},
_ => {
refs.clear();
refs.push(kv);
encode_key_refs_into_in(&refs, &mut keybuf_s, mysql_fold_groups);
match groups.get(keybuf_s.as_str()) {
Some(&i) => i,
None => {
let i = order.len();
order.push((
alloc::vec![kv.clone().into_owned()],
alloc::vec![AggState::default()],
));
groups.insert(keybuf_s.clone(), i);
i
}
}
}
};
let av = row.get(apos).unwrap_or(&Value::Null);
if !matches!(av, Value::Null) {
let st = &mut order[idx].1[0];
let upd = match &st.extreme {
None => true,
Some(prev) => {
extreme_cmp_in(
agg_specs[0].enum_labels.as_deref(),
agg_specs[0].arg_collation.as_deref(),
av,
prev,
ctx.mysql_dialect,
) == core::cmp::Ordering::Greater
}
};
if upd {
st.extreme = Some(av.clone().into_owned());
}
}
}
return Ok(order);
}
for row in rows.iter() {
for slot in row_eval_cache.iter_mut() {
*slot = None;
}
if single_anon_group {
let entry = &mut order[0];
let mat: Option<Cow<'_, Row>> = if needs_mat { Some(row.as_row()) } else { None };
for (i, spec) in agg_specs.iter().enumerate() {
if let Some(f) = &spec.filter
&& !matches!(
eval_arg(f, mat.as_deref().expect("needs_mat for FILTER"), &ctx)?,
Value::Bool(true)
)
{
continue;
}
let arg_owned: Value;
let arg_ref: &Value = match (&arg_pos[i], arg_slot[i], &spec.arg) {
(Some(p), _, _) => {
crate::bump_counter!(AGG_PER_ROW_FAST_POS);
row.get(*p).unwrap_or(&Value::Null)
}
(None, None, None) => {
crate::bump_counter!(AGG_PER_ROW_COUNT_STAR_SENTINEL);
arg_owned = Value::Bool(true);
&arg_owned
}
(None, Some(s), _) => {
if row_eval_cache[s].is_none() {
crate::bump_counter!(AGG_PER_ROW_COMPILED_MISS);
let c = arg_compiled[arg_unique_idx[s]]
.as_ref()
.expect("arg_unique_idx points at a compiled spec");
let v = eval::eval_compiled_ref(c, row, &ctx, &mut eval_stack)?;
row_eval_cache[s] = Some(v);
} else {
crate::bump_counter!(AGG_PER_ROW_COMPILED_HIT);
}
row_eval_cache[s].as_ref().expect("just filled above")
}
(None, None, Some(e)) => {
crate::bump_counter!(AGG_PER_ROW_EVAL_FALLBACK);
arg_owned = eval_arg(
e,
mat.as_deref().expect("needs_mat for non-bound arg"),
&ctx,
)?;
&arg_owned
}
};
let arg2_val = match (&spec.arg2, &arg2_literal_val[i]) {
(None, _) => None,
(Some(_), Some(lit)) => {
crate::bump_counter!(DISTA_LITERAL_ARG2_CACHE_FIRE);
Some(lit.clone())
}
(Some(e), None) => Some(eval_arg(
e,
mat.as_deref().expect("needs_mat for arg2"),
&ctx,
)?),
};
let order_keys: Option<Vec<Value<'static>>> = if spec.order_by.is_empty() {
None
} else {
crate::bump_counter!(AGGREGATE_ARRAY_AGG_ORDER_BY_FIRE);
let mut keys: Vec<Value<'static>> = Vec::with_capacity(spec.order_by.len());
for (k, o) in spec.order_by.iter().enumerate() {
let v: Value<'static> = if let Some(p) = order_pos[i][k] {
row.get(p)
.cloned()
.map(Value::into_owned)
.unwrap_or(Value::Null)
} else {
eval_arg(
&o.expr,
mat.as_deref().expect("needs_mat for ORDER key"),
&ctx,
)?
};
keys.push(v);
}
Some(keys)
};
if spec.first_ordered {
if let Some(keys) = order_keys {
let st = &mut entry.1[i];
let better = match &st.first_best {
None => true,
Some((bk, _)) => {
cmp_order_keys(
&spec.order_by,
&spec.order_enum_labels,
&keys,
bk,
ctx.mysql_dialect,
) == core::cmp::Ordering::Less
}
};
if better {
st.first_best = Some((keys, arg_ref.clone().into_owned()));
}
}
continue;
}
if spec.distinct {
if let Value::Text(s) = arg_ref {
if distinct_fold[i] {
let k = spg_storage::mysql_compare_fold(s);
if entry.1[i].seen.contains(k.as_str()) {
continue;
}
entry.1[i].seen.insert(k);
} else {
if entry.1[i].seen.contains(s.as_ref()) {
continue;
}
entry.1[i].seen.insert(s.to_string());
}
} else if let Value::BigInt(n) = arg_ref {
let set = entry.1[i].seen_int.get_or_insert_with(BTreeSet::new);
if !set.insert(*n) {
continue;
}
} else if let Value::Int(n) = arg_ref {
let set = entry.1[i].seen_int.get_or_insert_with(BTreeSet::new);
if !set.insert(i64::from(*n)) {
continue;
}
} else {
encode_key_refs_into_in(
core::slice::from_ref(&arg_ref),
&mut dkeybuf,
distinct_fold[i],
);
if entry.1[i].seen.contains(dkeybuf.as_str()) {
continue;
}
entry.1[i].seen.insert(dkeybuf.clone());
}
}
match spec.kind {
AggKind::Max => {
if !matches!(arg_ref, Value::Null) {
if !ctx.mysql_dialect && min_max_unsupported_type(arg_ref) {
return Err(EvalError::TypeMismatch {
detail: format!(
"function max({}) does not exist",
crate::conversions::pg_type_name_for_error_opt(
arg_ref.data_type()
)
),
});
}
let st = &mut entry.1[i];
let upd = match &st.extreme {
None => true,
Some(prev) => {
extreme_cmp_in(
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
arg_ref,
prev,
ctx.mysql_dialect,
) == core::cmp::Ordering::Greater
}
};
if upd {
st.extreme = Some(arg_ref.clone().into_owned());
}
}
}
AggKind::Min => {
if !matches!(arg_ref, Value::Null) {
if !ctx.mysql_dialect && min_max_unsupported_type(arg_ref) {
return Err(EvalError::TypeMismatch {
detail: format!(
"function min({}) does not exist",
crate::conversions::pg_type_name_for_error_opt(
arg_ref.data_type()
)
),
});
}
let st = &mut entry.1[i];
let upd = match &st.extreme {
None => true,
Some(prev) => {
extreme_cmp_in(
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
arg_ref,
prev,
ctx.mysql_dialect,
) == core::cmp::Ordering::Less
}
};
if upd {
st.extreme = Some(arg_ref.clone().into_owned());
}
}
}
AggKind::AnyValue => {
if !matches!(arg_ref, Value::Null) {
let st = &mut entry.1[i];
if st.extreme.is_none() {
st.extreme = Some(arg_ref.clone().into_owned());
}
}
}
AggKind::CountStar => {
entry.1[i].num.count += 1;
}
AggKind::Count => {
if !matches!(arg_ref, Value::Null) {
entry.1[i].num.count += 1;
}
}
AggKind::BoolOr => match arg_ref {
Value::Bool(b) => {
let st = &mut entry.1[i];
st.bool_acc = Some(st.bool_acc.unwrap_or(false) || *b);
}
Value::Null => {}
_ => update_state(
&mut entry.1[i],
spec.kind,
&spec.name,
arg_ref,
arg2_val.as_ref(),
order_keys,
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
ctx.mysql_dialect,
)?,
},
AggKind::BoolAnd => match arg_ref {
Value::Bool(b) => {
let st = &mut entry.1[i];
st.bool_acc = Some(st.bool_acc.unwrap_or(true) && *b);
}
Value::Null => {}
_ => update_state(
&mut entry.1[i],
spec.kind,
&spec.name,
arg_ref,
arg2_val.as_ref(),
order_keys,
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
ctx.mysql_dialect,
)?,
},
_ => {
update_state(
&mut entry.1[i],
spec.kind,
&spec.name,
arg_ref,
arg2_val.as_ref(),
order_keys,
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
ctx.mysql_dialect,
)?;
}
}
}
continue;
}
if all_groups_bound && ci_positions.is_empty() {
let idx = if single_text_group_col {
let v = row.get(group_pos[0].unwrap()).unwrap_or(&Value::Null);
match v {
Value::Text(s) => match groups_text.get(s.as_ref()) {
Some(&i) => i,
None => {
let i = order.len();
let init: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
order.push((alloc::vec![Value::text(s.clone())], init));
groups_text.insert(s.to_string(), i);
i
}
},
Value::Null => match null_group_idx {
Some(i) => i,
None => {
let i = order.len();
let init: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
order.push((alloc::vec![Value::Null], init));
null_group_idx = Some(i);
i
}
},
_ => {
refs.clear();
refs.push(v);
encode_key_refs_into_in(&refs, &mut keybuf_s, mysql_fold_groups);
match groups.get(keybuf_s.as_str()) {
Some(&i) => i,
None => {
let i = order.len();
let init: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
order.push((alloc::vec![v.clone().into_owned()], init));
groups.insert(keybuf_s.clone(), i);
i
}
}
}
}
} else if single_int_group_col {
let v = row.get(group_pos[0].unwrap()).unwrap_or(&Value::Null);
let key: Option<i64> = match v {
Value::SmallInt(n) => Some(i64::from(*n)),
Value::Int(n) => Some(i64::from(*n)),
Value::BigInt(n) => Some(*n),
_ => None,
};
match (key, v) {
(Some(k), _) => match groups_int.get(&k) {
Some(&i) => i,
None => {
let i = order.len();
let init: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
order.push((alloc::vec![v.clone().into_owned()], init));
groups_int.insert(k, i);
i
}
},
(None, Value::Null) => match null_group_idx {
Some(i) => i,
None => {
let i = order.len();
let init: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
order.push((alloc::vec![Value::Null], init));
null_group_idx = Some(i);
i
}
},
(None, _) => {
refs.clear();
refs.push(v);
encode_key_refs_into_in(&refs, &mut keybuf_s, mysql_fold_groups);
match groups.get(keybuf_s.as_str()) {
Some(&i) => i,
None => {
let i = order.len();
let init: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
order.push((alloc::vec![v.clone().into_owned()], init));
groups.insert(keybuf_s.clone(), i);
i
}
}
}
}
} else {
refs.clear();
refs.extend(
group_pos
.iter()
.map(|p| row.get(p.unwrap()).unwrap_or(&Value::Null)),
);
encode_key_refs_into_in(&refs, &mut keybuf_s, mysql_fold_groups);
match groups.get(keybuf_s.as_str()) {
Some(&i) => i,
None => {
let i = order.len();
let init: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
let owned: Vec<Value<'static>> =
refs.iter().map(|v| (*v).clone().into_owned()).collect();
order.push((owned, init));
groups.insert(keybuf_s.clone(), i);
i
}
}
};
let entry = &mut order[idx];
let mat: Option<Cow<'_, Row>> = if needs_mat { Some(row.as_row()) } else { None };
for (i, spec) in agg_specs.iter().enumerate() {
if let Some(f) = &spec.filter
&& !matches!(
eval_arg(f, mat.as_deref().expect("needs_mat for FILTER"), &ctx)?,
Value::Bool(true)
)
{
continue;
}
let arg_owned: Value;
let arg_ref: &Value = match (&arg_pos[i], arg_slot[i], &spec.arg) {
(Some(p), _, _) => {
crate::bump_counter!(AGG_PER_ROW_FAST_POS);
row.get(*p).unwrap_or(&Value::Null)
}
(None, None, None) => {
crate::bump_counter!(AGG_PER_ROW_COUNT_STAR_SENTINEL);
arg_owned = Value::Bool(true);
&arg_owned
}
(None, Some(s), _) => {
if row_eval_cache[s].is_none() {
crate::bump_counter!(AGG_PER_ROW_COMPILED_MISS);
let c = arg_compiled[arg_unique_idx[s]]
.as_ref()
.expect("arg_unique_idx points at a compiled spec");
let v = eval::eval_compiled_ref(c, row, &ctx, &mut eval_stack)?;
row_eval_cache[s] = Some(v);
} else {
crate::bump_counter!(AGG_PER_ROW_COMPILED_HIT);
}
row_eval_cache[s].as_ref().expect("just filled above")
}
(None, None, Some(e)) => {
crate::bump_counter!(AGG_PER_ROW_EVAL_FALLBACK);
arg_owned = eval_arg(
e,
mat.as_deref().expect("needs_mat for non-bound arg"),
&ctx,
)?;
&arg_owned
}
};
let arg2_val = match (&spec.arg2, &arg2_literal_val[i]) {
(None, _) => None,
(Some(_), Some(lit)) => {
crate::bump_counter!(DISTA_LITERAL_ARG2_CACHE_FIRE);
Some(lit.clone())
}
(Some(e), None) => Some(eval_arg(
e,
mat.as_deref().expect("needs_mat for arg2"),
&ctx,
)?),
};
let order_keys: Option<Vec<Value<'static>>> = if spec.order_by.is_empty() {
None
} else {
crate::bump_counter!(AGGREGATE_ARRAY_AGG_ORDER_BY_FIRE);
let mut keys: Vec<Value<'static>> = Vec::with_capacity(spec.order_by.len());
for (k, o) in spec.order_by.iter().enumerate() {
keys.push(match order_pos[i][k] {
Some(p) => row
.get(p)
.cloned()
.map(Value::into_owned)
.unwrap_or(Value::Null),
None => eval_arg(
&o.expr,
mat.as_deref().expect("needs_mat for non-bound ORDER key"),
&ctx,
)?,
});
}
Some(keys)
};
if spec.first_ordered {
if let Some(keys) = order_keys {
let st = &mut entry.1[i];
let better = match &st.first_best {
None => true,
Some((bk, _)) => {
cmp_order_keys(
&spec.order_by,
&spec.order_enum_labels,
&keys,
bk,
ctx.mysql_dialect,
) == core::cmp::Ordering::Less
}
};
if better {
st.first_best = Some((keys, arg_ref.clone().into_owned()));
}
}
continue;
}
if spec.distinct {
if let Value::Text(s) = arg_ref {
if entry.1[i].seen.contains(s.as_ref()) {
continue;
}
entry.1[i].seen.insert(s.to_string());
} else if let Value::BigInt(n) = arg_ref {
let set = entry.1[i].seen_int.get_or_insert_with(BTreeSet::new);
if !set.insert(*n) {
continue;
}
} else if let Value::Int(n) = arg_ref {
let set = entry.1[i].seen_int.get_or_insert_with(BTreeSet::new);
if !set.insert(i64::from(*n)) {
continue;
}
} else {
encode_key_refs_into_in(
core::slice::from_ref(&arg_ref),
&mut dkeybuf,
distinct_fold[i],
);
if entry.1[i].seen.contains(dkeybuf.as_str()) {
continue;
}
entry.1[i].seen.insert(dkeybuf.clone());
}
}
match spec.kind {
AggKind::Max => {
if !matches!(arg_ref, Value::Null) {
if !ctx.mysql_dialect && min_max_unsupported_type(arg_ref) {
return Err(EvalError::TypeMismatch {
detail: format!(
"function max({}) does not exist",
crate::conversions::pg_type_name_for_error_opt(
arg_ref.data_type()
)
),
});
}
let st = &mut entry.1[i];
let upd = match &st.extreme {
None => true,
Some(prev) => {
extreme_cmp_in(
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
arg_ref,
prev,
ctx.mysql_dialect,
) == core::cmp::Ordering::Greater
}
};
if upd {
st.extreme = Some(arg_ref.clone().into_owned());
}
}
}
AggKind::Min => {
if !matches!(arg_ref, Value::Null) {
if !ctx.mysql_dialect && min_max_unsupported_type(arg_ref) {
return Err(EvalError::TypeMismatch {
detail: format!(
"function min({}) does not exist",
crate::conversions::pg_type_name_for_error_opt(
arg_ref.data_type()
)
),
});
}
let st = &mut entry.1[i];
let upd = match &st.extreme {
None => true,
Some(prev) => {
extreme_cmp_in(
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
arg_ref,
prev,
ctx.mysql_dialect,
) == core::cmp::Ordering::Less
}
};
if upd {
st.extreme = Some(arg_ref.clone().into_owned());
}
}
}
AggKind::AnyValue => {
if !matches!(arg_ref, Value::Null) {
let st = &mut entry.1[i];
if st.extreme.is_none() {
st.extreme = Some(arg_ref.clone().into_owned());
}
}
}
AggKind::CountStar => {
entry.1[i].num.count += 1;
}
AggKind::Count => {
if !matches!(arg_ref, Value::Null) {
entry.1[i].num.count += 1;
}
}
AggKind::BoolOr => match arg_ref {
Value::Bool(b) => {
let st = &mut entry.1[i];
st.bool_acc = Some(st.bool_acc.unwrap_or(false) || *b);
}
Value::Null => {}
_ => update_state(
&mut entry.1[i],
spec.kind,
&spec.name,
arg_ref,
arg2_val.as_ref(),
order_keys,
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
ctx.mysql_dialect,
)?,
},
AggKind::BoolAnd => match arg_ref {
Value::Bool(b) => {
let st = &mut entry.1[i];
st.bool_acc = Some(st.bool_acc.unwrap_or(true) && *b);
}
Value::Null => {}
_ => update_state(
&mut entry.1[i],
spec.kind,
&spec.name,
arg_ref,
arg2_val.as_ref(),
order_keys,
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
ctx.mysql_dialect,
)?,
},
_ => {
update_state(
&mut entry.1[i],
spec.kind,
&spec.name,
arg_ref,
arg2_val.as_ref(),
order_keys,
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
ctx.mysql_dialect,
)?;
}
}
}
continue;
}
let row_materialised = row.as_row();
let row: &Row<'static> = &row_materialised;
let group_vals: Vec<Value<'static>> = group_exprs
.iter()
.map(|g| eval::eval_expr(g, row, &ctx))
.collect::<Result<_, _>>()?;
let key = if ci_positions.is_empty() {
encode_key(&group_vals)
} else {
let mut key_vals = group_vals.clone();
for &i in &ci_positions {
if let Value::Text(s) = &key_vals[i] {
key_vals[i] = Value::text(if ctx.mysql_dialect {
spg_storage::mysql_compare_fold(s)
} else {
s.to_ascii_lowercase()
});
}
}
encode_key(&key_vals)
};
let idx = match groups.get(key.as_str()) {
Some(&i) => i,
None => {
let i = order.len();
let init: Vec<AggState> =
(0..agg_specs.len()).map(|_| AggState::default()).collect();
order.push((group_vals.clone(), init));
groups.insert(key, i);
i
}
};
let entry = &mut order[idx];
for (i, spec) in agg_specs.iter().enumerate() {
if let Some(f) = &spec.filter
&& !matches!(eval_arg(f, row, &ctx)?, Value::Bool(true))
{
continue;
}
let arg_val = match &spec.arg {
None => Value::Bool(true), Some(e) => eval_arg(e, row, &ctx)?,
};
let arg2_val = match &spec.arg2 {
None => None,
Some(e) => Some(eval_arg(e, row, &ctx)?),
};
let order_keys: Option<Vec<Value<'static>>> = if spec.order_by.is_empty() {
None
} else {
let mut keys: Vec<Value<'static>> = Vec::with_capacity(spec.order_by.len());
for o in &spec.order_by {
keys.push(eval_arg(&o.expr, row, &ctx)?);
}
Some(keys)
};
if spec.first_ordered {
if let Some(keys) = order_keys {
let st = &mut entry.1[i];
let better = match &st.first_best {
None => true,
Some((bk, _)) => {
cmp_order_keys(
&spec.order_by,
&spec.order_enum_labels,
&keys,
bk,
ctx.mysql_dialect,
) == core::cmp::Ordering::Less
}
};
if better {
st.first_best = Some((keys, arg_val.clone().into_owned()));
}
}
continue;
}
if spec.distinct {
let inserted = match &arg_val {
Value::Text(s) => entry.1[i].seen.insert(s.to_string()),
Value::BigInt(n) => entry.1[i]
.seen_int
.get_or_insert_with(BTreeSet::new)
.insert(*n),
Value::Int(n) => entry.1[i]
.seen_int
.get_or_insert_with(BTreeSet::new)
.insert(i64::from(*n)),
_ => {
let key = encode_key(core::slice::from_ref(&arg_val));
entry.1[i].seen.insert(key)
}
};
if !inserted {
continue;
}
}
update_state(
&mut entry.1[i],
spec.kind,
&spec.name,
&arg_val,
arg2_val.as_ref(),
order_keys,
spec.enum_labels.as_deref(),
spec.arg_collation.as_deref(),
ctx.mysql_dialect,
)?;
}
}
Ok(order)
}
fn build_synth_schema(
rows: AggRows<'_>,
group_exprs: &[Expr],
agg_specs: &[AggSpec],
schema_cols: &[ColumnSchema],
table_alias: Option<&str>,
catalog: Option<&spg_storage::Catalog>,
engine: Option<&crate::Engine>,
) -> Result<Vec<ColumnSchema>, EvalError> {
let ctx = with_catalog(EvalContext::new(schema_cols, table_alias), catalog, engine);
let group_types: Vec<DataType> = if rows.is_empty() {
group_exprs.iter().map(|_| DataType::Text).collect()
} else {
let probe = rows.get(0).expect("non-empty checked above");
let probe_row = probe.as_row();
let probe: &Row<'static> = &probe_row;
group_exprs
.iter()
.map(|g| {
eval::eval_expr(g, probe, &ctx).map(|v| v.data_type().unwrap_or(DataType::Text))
})
.collect::<Result<_, _>>()?
};
let agg_types: Vec<DataType> = agg_specs
.iter()
.map(|spec| infer_agg_type(spec, schema_cols))
.collect();
let mut synth_schema: Vec<ColumnSchema> = Vec::new();
for (i, ty) in group_types.iter().enumerate() {
let mut col = ColumnSchema::new(format!("__grp_{i}"), *ty, true);
if let Some(Expr::Column(c)) = group_exprs.get(i) {
let src = schema_cols.iter().find(|sc| sc.name == c.name);
col.user_enum_type = src.and_then(|sc| sc.user_enum_type.clone());
col.collation_name = src.and_then(|sc| sc.collation_name.clone());
}
synth_schema.push(col);
}
for (i, ty) in agg_types.iter().enumerate() {
synth_schema.push(ColumnSchema::new(format!("__agg_{i}"), *ty, true));
}
Ok(synth_schema)
}
fn cmp_order_keys(
order_by: &[spg_sql::ast::OrderBy],
order_enum_labels: &[Option<Vec<String>>],
a: &[Value<'static>],
b: &[Value<'static>],
mysql: bool,
) -> core::cmp::Ordering {
for (k, o) in order_by.iter().enumerate() {
if let Some(Some(labels)) = order_enum_labels.get(k)
&& !matches!(&a[k], Value::Null)
&& !matches!(&b[k], Value::Null)
&& let Some(ord) = crate::eval::enum_ord_cmp(labels, &a[k], &b[k])
{
let ord = if o.desc { ord.reverse() } else { ord };
if ord != core::cmp::Ordering::Equal {
return ord;
}
continue;
}
let fold = mysql && !crate::eval::is_binary_coerced(&o.expr);
let cmp = crate::order_by_value_cmp_in(o.desc, o.nulls_first, &a[k], &b[k], fold);
if cmp != core::cmp::Ordering::Equal {
return cmp;
}
}
core::cmp::Ordering::Equal
}
#[allow(clippy::too_many_arguments)]
fn finalize_synth_rows(
order: &[(Vec<Value<'static>>, Vec<AggState>)],
agg_specs: &[AggSpec],
synth_schema: &[ColumnSchema],
rows: AggRows<'_>,
schema_cols: &[ColumnSchema],
table_alias: Option<&str>,
catalog: Option<&spg_storage::Catalog>,
engine: Option<&crate::Engine>,
runner: Option<&dyn crate::ParallelRunner>,
) -> Result<Vec<Row<'static>>, EvalError> {
let ctx = with_catalog(EvalContext::new(schema_cols, table_alias), catalog, engine);
let collections_only = agg_specs.iter().all(|s| {
matches!(
classify_agg_name(&s.name),
AggKind::StringAgg | AggKind::ArrayAgg | AggKind::JsonAgg
) && !s.first_ordered
&& !is_within_group_name(&s.name)
});
if collections_only
&& order.len() >= 16
&& let Some(r) = runner
{
let group_len_probe = order.first().map(|(g, _)| g.len()).unwrap_or(0);
let _ = group_len_probe;
let n_shards = (order.len() / 8).clamp(2, 8);
let chunk = order.len().div_ceil(n_shards);
type ShardOut = Result<Vec<Row<'static>>, EvalError>;
let mysql = ctx.mysql_dialect;
let style = ctx.render_style;
let results = r.run_shards(n_shards, &|si| {
let lo = si * chunk;
let hi = ((si + 1) * chunk).min(order.len());
let mut sctx = EvalContext::new(schema_cols, table_alias);
sctx.mysql_dialect = mysql;
sctx.render_style = style;
let run = || -> ShardOut {
let mut out: Vec<Row<'static>> = Vec::with_capacity(hi - lo);
for (gvals, states) in &order[lo..hi] {
out.push(finalize_one_group(
gvals,
states,
agg_specs,
synth_schema,
&sctx,
)?);
}
Ok(out)
};
alloc::boxed::Box::new(run())
});
let mut synth_rows: Vec<Row<'static>> = Vec::with_capacity(order.len());
for boxed in results {
let shard = boxed
.downcast::<ShardOut>()
.expect("runner echoes the closure's box");
synth_rows.extend((*shard)?);
}
return Ok(synth_rows);
}
let direct_arg_vals: Vec<Option<Value>> = agg_specs
.iter()
.map(|spec| match (&spec.direct_arg, rows.first().as_ref()) {
(Some(e), Some(r)) => eval::eval_expr(e, &r.as_row(), &ctx).map(Some),
_ => Ok(None),
})
.collect::<Result<_, _>>()?;
let direct_extra_vals: Vec<Vec<Value>> = agg_specs
.iter()
.map(|spec| match rows.first().as_ref() {
Some(r) if !spec.direct_args_extra.is_empty() => spec
.direct_args_extra
.iter()
.map(|e| eval::eval_expr(e, &r.as_row(), &ctx))
.collect(),
_ => Ok(Vec::new()),
})
.collect::<Result<_, _>>()?;
let mut synth_rows: Vec<Row<'static>> = Vec::new();
for (gvals, states) in order {
let mut values: Vec<Value<'static>> = Vec::with_capacity(synth_schema.len());
let group_len = gvals.len();
values.extend(gvals.iter().cloned());
for (i, st) in states.iter().enumerate() {
if agg_specs[i].first_ordered {
values.push(
st.first_best
.as_ref()
.map_or(Value::Null, |(_, v)| v.clone()),
);
continue;
}
let st_sorted;
let kw = agg_specs[i].order_by.len();
let st_final: &AggState = if kw > 0 && st.item_keys.len() == st.items.len() * kw {
let mut idx: Vec<usize> = (0..st.items.len()).collect();
let ob = &agg_specs[i].order_by;
idx.sort_by(|&x, &y| {
cmp_order_keys(
ob,
&agg_specs[i].order_enum_labels,
&st.item_keys[x * kw..(x + 1) * kw],
&st.item_keys[y * kw..(y + 1) * kw],
ctx.mysql_dialect,
)
});
let mut sorted = st.clone();
let mut new_items: Vec<Value<'static>> = Vec::with_capacity(idx.len());
for &j in &idx {
new_items.push(core::mem::replace(&mut sorted.items[j], Value::Null));
}
if sorted.item_seps.len() == sorted.items.len() {
let mut new_seps: Vec<Option<String>> = Vec::with_capacity(idx.len());
for &j in &idx {
new_seps.push(core::mem::take(&mut sorted.item_seps[j]));
}
sorted.item_seps = new_seps;
}
sorted.items = new_items;
st_sorted = sorted;
&st_sorted
} else if agg_specs[i].distinct && st.items.len() > 1 {
let labels = agg_specs[i].enum_labels.as_deref();
let mut sorted = st.clone();
sorted.item_seps.clear();
sorted.items.sort_by(|a, b| {
if let Some(labels) = labels
&& !matches!(a, Value::Null)
&& !matches!(b, Value::Null)
&& let Some(ord) = crate::eval::enum_ord_cmp(labels, a, b)
{
return ord;
}
crate::order_by_value_cmp_in(false, Some(false), a, b, ctx.mysql_dialect)
});
st_sorted = sorted;
&st_sorted
} else {
st
};
let v = if is_within_group_name(&agg_specs[i].name) {
finalize_ordered_set(
&agg_specs[i].name,
st_final,
direct_arg_vals[i].as_ref(),
&direct_extra_vals[i],
&agg_specs[i].order_by,
ctx.mysql_dialect,
)?
} else {
finalize(&agg_specs[i].name, st_final, ctx.mysql_dialect)
};
let v = match (v, synth_schema.get(group_len + i).map(|c| c.ty)) {
(Value::TimestampArray(items), Some(DataType::TimestamptzArray)) => {
Value::TimestamptzArray(items)
}
(v, _) => v,
};
values.push(v);
}
synth_rows.push(Row::new(values));
}
Ok(synth_rows)
}
fn finalize_one_group(
gvals: &[Value<'static>],
states: &[AggState],
agg_specs: &[AggSpec],
synth_schema: &[ColumnSchema],
ctx: &EvalContext<'_>,
) -> Result<Row<'static>, EvalError> {
let group_len = gvals.len();
let mut values: Vec<Value<'static>> = Vec::with_capacity(synth_schema.len());
values.extend(gvals.iter().cloned());
for (i, st) in states.iter().enumerate() {
let st_sorted;
let kw = agg_specs[i].order_by.len();
let st_final: &AggState = if kw > 0 && st.item_keys.len() == st.items.len() * kw {
let mut idx: Vec<usize> = (0..st.items.len()).collect();
let ob = &agg_specs[i].order_by;
idx.sort_by(|&x, &y| {
cmp_order_keys(
ob,
&agg_specs[i].order_enum_labels,
&st.item_keys[x * kw..(x + 1) * kw],
&st.item_keys[y * kw..(y + 1) * kw],
ctx.mysql_dialect,
)
});
let mut sorted = st.clone();
let mut new_items: Vec<Value<'static>> = Vec::with_capacity(idx.len());
for &j in &idx {
new_items.push(core::mem::replace(&mut sorted.items[j], Value::Null));
}
if sorted.item_seps.len() == sorted.items.len() {
let mut new_seps: Vec<Option<String>> = Vec::with_capacity(idx.len());
for &j in &idx {
new_seps.push(core::mem::take(&mut sorted.item_seps[j]));
}
sorted.item_seps = new_seps;
}
sorted.items = new_items;
st_sorted = sorted;
&st_sorted
} else if agg_specs[i].distinct && st.items.len() > 1 {
let labels = agg_specs[i].enum_labels.as_deref();
let mut sorted = st.clone();
sorted.item_seps.clear();
sorted.items.sort_by(|a, b| {
if let Some(labels) = labels
&& !matches!(a, Value::Null)
&& !matches!(b, Value::Null)
&& let Some(ord) = crate::eval::enum_ord_cmp(labels, a, b)
{
return ord;
}
crate::order_by_value_cmp_in(false, Some(false), a, b, ctx.mysql_dialect)
});
st_sorted = sorted;
&st_sorted
} else {
st
};
let v = finalize(&agg_specs[i].name, st_final, ctx.mysql_dialect);
let v = match (v, synth_schema.get(group_len + i).map(|c| c.ty)) {
(Value::TimestampArray(items), Some(DataType::TimestamptzArray)) => {
Value::TimestamptzArray(items)
}
(v, _) => v,
};
values.push(v);
}
Ok(Row::new(values))
}
#[allow(clippy::too_many_lines)]
fn project_groups(
synth_rows: Vec<Row<'static>>,
stmt: &SelectStatement,
group_exprs: &[Expr],
agg_specs: &[AggSpec],
synth_schema: &[ColumnSchema],
correlated_eval: Option<CorrelatedEval<'_>>,
defer_projection: bool,
catalog: Option<&spg_storage::Catalog>,
mysql: bool,
) -> Result<Projection, EvalError> {
let columns: Vec<ColumnSchema> = stmt
.items
.iter()
.map(|item| match item {
SelectItem::Wildcard | SelectItem::QualifiedWildcard(_) => {
Err(EvalError::TypeMismatch {
detail: "SELECT * with aggregates is not supported".into(),
})
}
SelectItem::Expr { expr, alias } => {
let rewritten = rewrite_expr(expr, group_exprs, agg_specs);
let name = alias
.clone()
.unwrap_or_else(|| crate::select::default_output_name(expr, mysql));
Ok(ColumnSchema::new(
name,
agg_or_group_type(&rewritten, synth_schema),
true,
))
}
})
.collect::<Result<_, _>>()?;
let mut synth_ctx = EvalContext::new(synth_schema, None);
if let Some(cat) = catalog {
synth_ctx = synth_ctx.with_catalog(cat);
}
let having_aliases: Vec<(String, Expr)> = if mysql {
stmt.items
.iter()
.filter_map(|it| match it {
SelectItem::Expr {
expr,
alias: Some(a),
} if !matches!(expr, Expr::Column(c)
if c.qualifier.is_none() && c.name.eq_ignore_ascii_case(a)) =>
{
Some((a.clone(), expr.clone()))
}
_ => None,
})
.collect()
} else {
Vec::new()
};
let having_rewritten = stmt.having.as_ref().map(|h| {
let h = if having_aliases.is_empty() {
h.clone()
} else {
substitute_having_aliases(h.clone(), &having_aliases)
};
rewrite_expr(&h, group_exprs, agg_specs)
});
let items_rewritten: alloc::vec::Vec<Option<Expr>> = stmt
.items
.iter()
.map(|item| match item {
SelectItem::Expr { expr, .. } => Some(rewrite_expr(expr, group_exprs, agg_specs)),
SelectItem::Wildcard | SelectItem::QualifiedWildcard(_) => None,
})
.collect();
let order_rewritten: Vec<Expr> = stmt
.order_by
.iter()
.map(|o| rewrite_expr(&o.expr, group_exprs, agg_specs))
.collect();
let defer_enabled = correlated_eval.is_some()
&& !stmt.distinct
&& !having_rewritten
.as_ref()
.is_some_and(crate::expr_has_subquery)
&& !order_rewritten.iter().any(crate::expr_has_subquery);
let deferred: Vec<(usize, Expr)> = if defer_enabled {
items_rewritten
.iter()
.enumerate()
.filter_map(|(i, r)| {
r.as_ref()
.filter(|e| crate::expr_has_subquery(e))
.map(|e| (i, e.clone()))
})
.collect()
} else {
Vec::new()
};
let having_compiled = having_rewritten
.as_ref()
.filter(|h| eval::fully_compilable(h))
.map(|h| eval::compile_expr(h, &synth_ctx));
let items_compiled: Vec<Option<eval::CompiledExpr>> = items_rewritten
.iter()
.enumerate()
.map(|(i, r)| {
r.as_ref()
.filter(|e| !deferred.iter().any(|(c, _)| *c == i) && eval::fully_compilable(e))
.map(|e| eval::compile_expr(e, &synth_ctx))
})
.collect();
let srf_items: Vec<bool> = items_rewritten
.iter()
.map(|r| {
r.as_ref()
.is_some_and(|e| crate::select::top_level_srf_kind(e).is_some())
})
.collect();
let any_srf = srf_items.iter().any(|b| *b);
let mut kept_synth: Vec<Row<'static>> = Vec::new();
let mut out_rows: Vec<Row<'static>> = Vec::new();
let mut stack: Vec<Value<'static>> = Vec::new();
for srow in synth_rows {
if let Some(hc) = &having_compiled {
let cond = eval::eval_compiled(hc, &srow, &synth_ctx, &mut stack)?;
if !crate::eval::predicate_is_true(&cond, "HAVING", synth_ctx.mysql_dialect)? {
continue;
}
} else if let Some(h) = &having_rewritten {
let cond = match correlated_eval {
Some(f) if crate::expr_has_subquery(h) => f(h, &srow, &synth_ctx)?,
_ => eval::eval_expr(h, &srow, &synth_ctx)?,
};
if !crate::eval::predicate_is_true(&cond, "HAVING", synth_ctx.mysql_dialect)? {
continue;
}
}
if defer_projection {
kept_synth.push(srow);
out_rows.push(Row::new(Vec::new()));
continue;
}
let mut values: Vec<Value<'static>> = Vec::with_capacity(columns.len());
for (i, rewritten) in items_rewritten.iter().enumerate() {
let Some(rewritten) = rewritten else { continue };
if deferred.iter().any(|(c, _)| *c == i) {
values.push(Value::Null);
continue;
}
if srf_items[i] {
values.push(Value::Null);
continue;
}
values.push(if let Some(cc) = &items_compiled[i] {
eval::eval_compiled(cc, &srow, &synth_ctx, &mut stack)?
} else {
match correlated_eval {
Some(f) if crate::expr_has_subquery(rewritten) => {
f(rewritten, &srow, &synth_ctx)?
}
_ => eval::eval_expr(rewritten, &srow, &synth_ctx)?,
}
});
}
if any_srf {
let mut lists: Vec<Vec<Value<'static>>> = Vec::with_capacity(items_rewritten.len());
for (i, rewritten) in items_rewritten.iter().enumerate() {
match (srf_items[i], rewritten) {
(true, Some(r)) => {
lists.push(
crate::select::top_level_srf_output(r, &srow, &synth_ctx).map_err(
|e| match e {
crate::EngineError::Eval(ev) => ev,
other => EvalError::TypeMismatch {
detail: alloc::format!("{other}"),
},
},
)?,
);
}
_ => lists.push(Vec::new()),
}
}
let n = lists.iter().map(Vec::len).max().unwrap_or(0);
for k in 0..n {
let mut vals = values.clone();
for (i, list) in lists.iter().enumerate() {
if srf_items[i]
&& let Some(slot) = vals.get_mut(i)
{
*slot = list.get(k).cloned().unwrap_or(Value::Null);
}
}
kept_synth.push(srow.clone());
out_rows.push(Row::new(vals));
}
continue;
}
kept_synth.push(srow);
out_rows.push(Row::new(values));
}
let deferred_project_state = if defer_projection {
Some(DeferredProject {
items_rewritten,
items_compiled,
})
} else {
None
};
Ok(Projection {
columns,
out_rows,
kept_synth,
deferred,
order_rewritten,
deferred_project: deferred_project_state,
})
}
fn sort_synth_by_order_by(
synth_schema: &[ColumnSchema],
out_columns: &[ColumnSchema],
order_by: &[spg_sql::ast::OrderBy],
order_rewritten: &[Expr],
mut kept_synth: Vec<Row<'static>>,
mut out_rows: Vec<Row<'static>>,
correlated_eval: Option<CorrelatedEval<'_>>,
keep_n: Option<usize>,
catalog: Option<&spg_storage::Catalog>,
mysql: bool,
) -> Result<(Vec<Row<'static>>, Vec<Row<'static>>), EvalError> {
let mut synth_ctx = EvalContext::new(synth_schema, None);
if let Some(cat) = catalog {
synth_ctx = synth_ctx.with_catalog(cat);
}
let key_enum_labels: Vec<Option<&[String]>> = order_rewritten
.iter()
.map(|e| crate::eval::expr_enum_labels(e, synth_schema, catalog))
.collect();
let key_colls: Vec<Option<alloc::string::String>> = order_rewritten
.iter()
.map(|e| {
let spg_sql::ast::Expr::Column(c) = e else {
return None;
};
let pos = crate::eval::find_column_pos(c, &synth_ctx)?;
let name = synth_schema.get(pos)?.collation_name.clone()?;
crate::collate::is_supported(&name).then_some(name)
})
.collect();
let keys_meta: Vec<(bool, Option<bool>)> =
order_by.iter().map(|o| (o.desc, o.nulls_first)).collect();
let order_compiled: Vec<Option<eval::CompiledExpr>> = order_rewritten
.iter()
.map(|e| {
Some(e)
.filter(|e| eval::fully_compilable(e))
.map(|e| eval::compile_expr(e, &synth_ctx))
})
.collect();
let out_key_idx: Vec<Option<usize>> = order_rewritten
.iter()
.map(|e| {
let spg_sql::ast::Expr::Column(c) = e else {
return None;
};
if c.qualifier.is_some() || crate::eval::find_column_pos(c, &synth_ctx).is_some() {
return None;
}
out_columns
.iter()
.position(|oc| oc.name.eq_ignore_ascii_case(&c.name))
})
.collect();
let mut keystack: Vec<Value<'static>> = Vec::new();
let mut tagged: Vec<(Vec<Value<'static>>, Row, Row)> = Vec::with_capacity(kept_synth.len());
for (s, o) in kept_synth.into_iter().zip(out_rows) {
let mut keys = Vec::with_capacity(order_rewritten.len());
for (i, (e, oc)) in order_rewritten.iter().zip(&order_compiled).enumerate() {
if let Some(oi) = out_key_idx[i] {
keys.push(o.values.get(oi).cloned().unwrap_or(Value::Null));
continue;
}
keys.push(if let Some(oc) = oc {
eval::eval_compiled(oc, &s, &synth_ctx, &mut keystack)?
} else {
match correlated_eval {
Some(f) if crate::expr_has_subquery(e) => f(e, &s, &synth_ctx)?,
_ => eval::eval_expr(e, &s, &synth_ctx)?,
}
});
}
tagged.push((keys, s, o));
}
let cmp = |a: &(Vec<Value<'static>>, Row, Row), b: &(Vec<Value<'static>>, Row, Row)| {
use core::cmp::Ordering;
for (i, (ka, kb)) in a.0.iter().zip(b.0.iter()).enumerate() {
let (desc, nf) = keys_meta[i];
if let Some(Some(labels)) = key_enum_labels.get(i)
&& !matches!(ka, Value::Null)
&& !matches!(kb, Value::Null)
&& let Some(ord) = crate::eval::enum_ord_cmp(labels, ka, kb)
{
let ord = if desc { ord.reverse() } else { ord };
if ord != Ordering::Equal {
return ord;
}
continue;
}
let c = crate::orderby::order_by_value_cmp_coll(
desc,
nf,
ka,
kb,
mysql,
key_colls.get(i).and_then(|c| c.as_deref()),
);
if c != Ordering::Equal {
return c;
}
}
Ordering::Equal
};
match keep_n {
Some(k) if k < tagged.len() && k > 0 => {
let pivot = k - 1;
tagged.select_nth_unstable_by(pivot, cmp);
tagged[..k].sort_by(cmp);
tagged.truncate(k);
}
_ => {
tagged.sort_by(cmp);
}
}
kept_synth = Vec::with_capacity(tagged.len());
out_rows = Vec::with_capacity(tagged.len());
for (_, s, o) in tagged {
kept_synth.push(s);
out_rows.push(o);
}
Ok((kept_synth, out_rows))
}
fn validate_agg_arities(stmt: &SelectStatement, _specs: &[AggSpec]) -> Result<(), EvalError> {
fn walk(e: &Expr) -> Result<(), EvalError> {
if let Expr::FunctionCall { name, args } = e {
let lower = name.to_ascii_lowercase();
let expected: Option<usize> = match lower.as_str() {
"count_star" => Some(0),
"count" | "sum" | "avg" | "min" | "max" | "array_agg"
| "any_value" | "range_agg" | "range_intersect_agg"
| "bool_and" | "bool_or" | "every"
| "stddev" | "stddev_samp" | "stddev_pop"
| "variance" | "var_samp" | "var_pop"
| "bit_and" | "bit_or" | "bit_xor"
| "json_agg" | "jsonb_agg" | "xmlagg"
| "json_arrayagg" | "json_agg_strict" | "jsonb_agg_strict" => Some(1),
"group_concat" => None,
"string_agg"
| "covar_pop" | "covar_samp" | "corr"
| "regr_count" | "regr_avgx" | "regr_avgy" | "regr_slope"
| "regr_intercept" | "regr_r2" | "regr_sxx" | "regr_syy" | "regr_sxy"
| "json_object_agg" | "jsonb_object_agg"
| "json_objectagg"
| "json_object_agg_strict" | "jsonb_object_agg_strict"
| "json_object_agg_unique" | "jsonb_object_agg_unique"
| "json_object_agg_unique_strict" | "jsonb_object_agg_unique_strict" => Some(2),
_ => None,
};
if let Some(want) = expected
&& args.len() != want
{
return Err(EvalError::TypeMismatch {
detail: alloc::format!("{lower}() takes {want} arg(s), got {}", args.len()),
});
}
for a in args {
walk(a)?;
}
} else if let Expr::Binary { lhs, rhs, .. } = e {
walk(lhs)?;
walk(rhs)?;
} else if let Expr::Unary { expr, .. }
| Expr::Cast { expr, .. }
| Expr::IsNull { expr, .. }
| Expr::BoolTest { expr, .. } = e
{
walk(expr)?;
}
Ok(())
}
for item in &stmt.items {
if let SelectItem::Expr { expr, .. } = item {
walk(expr)?;
}
}
for o in &stmt.order_by {
walk(&o.expr)?;
}
if let Some(h) = &stmt.having {
walk(h)?;
}
Ok(())
}
fn first_ordered_array_agg(e: &Expr) -> Option<(&Expr, &[spg_sql::ast::OrderBy], Option<&Expr>)> {
let Expr::ArraySubscript { target, index } = e else {
return None;
};
if !matches!(
index.as_ref(),
Expr::Literal(spg_sql::ast::Literal::Integer(1))
) {
return None;
}
let Expr::AggregateOrdered {
call,
order_by,
distinct,
filter,
} = target.as_ref()
else {
return None;
};
if *distinct || order_by.is_empty() {
return None;
}
let Expr::FunctionCall { name, args } = call.as_ref() else {
return None;
};
if !name.eq_ignore_ascii_case("array_agg") || args.len() != 1 {
return None;
}
Some((&args[0], order_by, filter.as_deref()))
}
fn stddev_exact_pair(
st: &AggState,
) -> Option<(
spg_storage::bignum::BigNumeric,
spg_storage::bignum::BigNumeric,
)> {
use spg_storage::bignum::BigNumeric as BN;
let fast =
(!st.stddev_i_spent && (st.stddev_i_sum != 0 || st.stddev_i_sum_sq != 0)).then(|| {
(
BN::from_i128(st.stddev_i_sum, 0),
BN::from_i128(st.stddev_i_sum_sq, 0),
)
});
match (st.stddev_sum.as_ref(), st.stddev_sum_sq.as_ref(), fast) {
(Some(s), Some(sq), Some((fs, fsq))) => Some((s.add(&fs), sq.add(&fsq))),
(Some(s), Some(sq), None) => Some((s.clone(), sq.clone())),
(None, None, Some(pair)) => Some(pair),
_ => None,
}
}
fn spend_stddev_i128(st: &mut AggState) {
if st.stddev_i_spent {
return;
}
st.stddev_i_spent = true;
if st.stddev_i_sum == 0 && st.stddev_i_sum_sq == 0 {
return;
}
use spg_storage::bignum::BigNumeric as BN;
let sum = BN::from_i128(st.stddev_i_sum, 0);
let sum_sq = BN::from_i128(st.stddev_i_sum_sq, 0);
st.stddev_sum = Some(st.stddev_sum.as_ref().map_or(sum.clone(), |s| s.add(&sum)));
st.stddev_sum_sq = Some(
st.stddev_sum_sq
.as_ref()
.map_or(sum_sq.clone(), |s| s.add(&sum_sq)),
);
}
fn collect_aggregates(e: &Expr, out: &mut Vec<AggSpec>) {
match e {
Expr::NamedArg { expr, .. } => collect_aggregates(expr, out),
Expr::Variadic(expr) => collect_aggregates(expr, out),
Expr::AggregateOrdered {
call,
order_by,
distinct,
filter,
} => {
if let Expr::FunctionCall { name, args } = call.as_ref() {
let lower = name.to_ascii_lowercase();
if is_aggregate_name(&lower) {
let canonical = if lower == "every" {
"bool_and".to_string()
} else {
lower
};
let ordered_set = is_within_group_name(&canonical);
let (arg, direct_arg, direct_args_extra) = if ordered_set {
(
order_by.first().map(|o| o.expr.clone()),
args.first().cloned(),
args.iter().skip(1).cloned().collect(),
)
} else {
(args.first().cloned(), None, Vec::new())
};
let spec = AggSpec {
kind: classify_agg_name(&canonical),
enum_labels: None,
arg_collation: None,
order_enum_labels: Vec::new(),
name: canonical.clone(),
arg,
arg2: if agg_uses_second_arg(&canonical) {
args.get(1).cloned()
} else {
None
},
distinct: *distinct,
order_by: order_by.clone(),
filter: filter.as_deref().cloned(),
direct_arg,
direct_args_extra,
first_ordered: false,
};
if !out.iter().any(|s| {
s.name == spec.name
&& s.arg == spec.arg
&& s.arg2 == spec.arg2
&& s.distinct == spec.distinct
&& s.order_by == spec.order_by
&& s.filter == spec.filter
&& s.direct_arg == spec.direct_arg
&& s.direct_args_extra == spec.direct_args_extra
&& s.first_ordered == spec.first_ordered
}) {
out.push(spec);
}
return;
}
}
collect_aggregates(call, out);
for o in order_by {
collect_aggregates(&o.expr, out);
}
}
Expr::FunctionCall { name, args } => {
let lower = name.to_ascii_lowercase();
if is_aggregate_name(&lower) {
let arg = if lower == "count_star" {
None
} else {
args.first().cloned()
};
let arg2 = if agg_uses_second_arg(&lower) {
args.get(1).cloned()
} else {
None
};
let canonical = if lower == "every" {
"bool_and".to_string()
} else {
lower
};
let spec = AggSpec {
kind: classify_agg_name(&canonical),
enum_labels: None,
arg_collation: None,
order_enum_labels: Vec::new(),
name: canonical,
arg: arg.clone(),
arg2: arg2.clone(),
distinct: false,
order_by: Vec::new(),
filter: None,
direct_arg: None,
direct_args_extra: Vec::new(),
first_ordered: false,
};
if !out.iter().any(|s| {
s.name == spec.name
&& s.arg == spec.arg
&& s.arg2 == spec.arg2
&& !s.distinct
&& s.order_by == spec.order_by
&& s.filter.is_none()
&& !s.first_ordered
}) {
out.push(spec);
}
} else {
for a in args {
collect_aggregates(a, out);
}
}
}
Expr::Binary { lhs, rhs, .. } => {
collect_aggregates(lhs, out);
collect_aggregates(rhs, out);
}
Expr::Unary { expr, .. }
| Expr::Cast { expr, .. }
| Expr::IsNull { expr, .. }
| Expr::BoolTest { expr, .. }
| Expr::FieldAccess { base: expr, .. } => {
collect_aggregates(expr, out);
}
Expr::Like { expr, pattern, .. } => {
collect_aggregates(expr, out);
collect_aggregates(pattern, out);
}
Expr::InList { expr, list, .. } => {
collect_aggregates(expr, out);
for item in list {
collect_aggregates(item, out);
}
}
Expr::Extract { source, .. } => collect_aggregates(source, out),
Expr::ScalarSubquery(_)
| Expr::Exists { .. }
| Expr::InSubquery { .. }
| Expr::RowInSubquery { .. }
| Expr::RowCmpSubquery { .. }
| Expr::WindowFunction { .. }
| Expr::Literal(_)
| Expr::Placeholder(_)
| Expr::Column(_) => {}
Expr::Array(items) => {
for elem in items {
collect_aggregates(elem, out);
}
}
Expr::ArraySubscript { target, index } => {
if let Some((arg, order_by, filter)) = first_ordered_array_agg(e) {
let spec = AggSpec {
kind: AggKind::ArrayAgg,
enum_labels: None,
arg_collation: None,
order_enum_labels: Vec::new(),
name: "array_agg".to_string(),
arg: Some(arg.clone()),
arg2: None,
distinct: false,
order_by: order_by.to_vec(),
filter: filter.cloned(),
direct_arg: None,
direct_args_extra: Vec::new(),
first_ordered: true,
};
if !out.iter().any(|s| {
s.name == spec.name
&& s.arg == spec.arg
&& s.order_by == spec.order_by
&& s.filter == spec.filter
&& s.first_ordered
}) {
out.push(spec);
}
return;
}
collect_aggregates(target, out);
collect_aggregates(index, out);
}
Expr::ArraySlice { target, lo, hi } => {
collect_aggregates(target, out);
if let Some(l) = lo {
collect_aggregates(l, out);
}
if let Some(h) = hi {
collect_aggregates(h, out);
}
}
Expr::AnyAll { expr, array, .. } => {
collect_aggregates(expr, out);
collect_aggregates(array, out);
}
Expr::Case {
operand,
branches,
else_branch,
} => {
if let Some(o) = operand {
collect_aggregates(o, out);
}
for (w, t) in branches {
collect_aggregates(w, out);
collect_aggregates(t, out);
}
if let Some(e) = else_branch {
collect_aggregates(e, out);
}
}
}
}
pub(crate) fn update_state(
st: &mut AggState,
kind: AggKind,
name: &str,
v: &Value<'_>,
arg2: Option<&Value<'_>>,
order_keys: Option<Vec<Value<'static>>>,
enum_labels: Option<&[String]>,
arg_collation: Option<&str>,
mysql: bool,
) -> Result<(), EvalError> {
let is_null = matches!(v, Value::Null);
match kind {
AggKind::CountStar => st.num.count += 1,
AggKind::Count => {
if !is_null {
st.num.count += 1;
}
}
AggKind::Sum | AggKind::Avg => {
acc_cell(&mut st.num, v)?;
}
AggKind::Min => {
if is_null {
return Ok(());
}
if !mysql && min_max_unsupported_type(v) {
return Err(EvalError::TypeMismatch {
detail: format!(
"function min({}) does not exist",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
});
}
match &st.extreme {
None => st.extreme = Some(v.clone().into_owned()),
Some(cur) => {
if extreme_cmp_in(enum_labels, arg_collation, v, cur, mysql)
== core::cmp::Ordering::Less
{
st.extreme = Some(v.clone().into_owned());
}
}
}
}
AggKind::AnyValue => {
if is_null {
return Ok(());
}
if st.extreme.is_none() {
st.extreme = Some(v.clone().into_owned());
}
}
AggKind::RangeAgg => {
if is_null {
return Ok(());
}
let Value::Range {
kind,
lower,
upper,
lower_inc,
upper_inc,
empty,
} = v
else {
return Err(EvalError::TypeMismatch {
detail: format!(
"range_agg requires a range value, got {}",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
});
};
if st.extreme.is_none() {
st.extreme = Some(Value::Multirange {
kind: *kind,
ranges: alloc::vec::Vec::new(),
});
}
if !empty && let Some(Value::Multirange { ranges, .. }) = &mut st.extreme {
ranges.push(spg_storage::RangeSpan {
lower: lower.clone(),
upper: upper.clone(),
lower_inc: *lower_inc,
upper_inc: *upper_inc,
empty: false,
});
}
}
AggKind::RangeIntersectAgg => {
if is_null {
return Ok(());
}
if !matches!(v, Value::Range { .. }) {
return Err(EvalError::TypeMismatch {
detail: format!(
"range_intersect_agg requires a range value, got {}",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
});
}
match &st.extreme {
None => st.extreme = Some(v.clone().into_owned()),
Some(prev) => {
st.extreme = Some(range_intersect(prev, &v.clone().into_owned()));
}
}
}
AggKind::Max => {
if is_null {
return Ok(());
}
if !mysql && min_max_unsupported_type(v) {
return Err(EvalError::TypeMismatch {
detail: format!(
"function max({}) does not exist",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
});
}
match &st.extreme {
None => st.extreme = Some(v.clone().into_owned()),
Some(cur) => {
if extreme_cmp_in(enum_labels, arg_collation, v, cur, mysql)
== core::cmp::Ordering::Greater
{
st.extreme = Some(v.clone().into_owned());
}
}
}
}
AggKind::StringAgg => {
let has_arg2 = arg2.is_some();
if let Some(sep) = arg2
&& let Value::Text(s) = sep
{
st.separator = Some(s.to_string());
}
if is_null {
return Ok(());
}
let rendered = render_string_agg_item(v);
if let Some(item) = rendered {
st.items.push(item);
if has_arg2 {
st.item_seps.push(match arg2 {
Some(Value::Text(sp)) => Some(sp.to_string()),
_ => None,
});
}
if let Some(k) = order_keys {
st.item_keys.extend(k);
}
st.num.count += 1;
} else {
return Err(EvalError::TypeMismatch {
detail: format!(
"string_agg requires text value, got {}",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
});
}
}
AggKind::ArrayAgg => {
st.items.push(v.clone().into_owned());
if let Some(k) = order_keys {
st.item_keys.extend(k);
}
st.num.count += 1;
}
AggKind::BoolAnd => {
if is_null {
return Ok(());
}
let b = match v {
Value::Bool(b) => *b,
other => {
return Err(EvalError::TypeMismatch {
detail: format!(
"bool_and requires bool, got {}",
crate::conversions::pg_type_name_for_error_opt(other.data_type())
),
});
}
};
st.bool_acc = Some(st.bool_acc.map_or(b, |acc| acc && b));
}
AggKind::BoolOr => {
if is_null {
return Ok(());
}
let b = match v {
Value::Bool(b) => *b,
other => {
return Err(EvalError::TypeMismatch {
detail: format!(
"bool_or requires bool, got {}",
crate::conversions::pg_type_name_for_error_opt(other.data_type())
),
});
}
};
st.bool_acc = Some(st.bool_acc.map_or(b, |acc| acc || b));
}
AggKind::StddevFamily => {
if is_null {
return Ok(());
}
if !st.stddev_saw_float {
let as_int = match v {
Value::SmallInt(n) => Some(i128::from(*n)),
Value::Int(n) => Some(i128::from(*n)),
Value::BigInt(n) => Some(i128::from(*n)),
_ => None,
};
let folded = if st.stddev_i_spent {
None
} else if let Some(x) = as_int {
match (
st.stddev_i_sum.checked_add(x),
x.checked_mul(x)
.and_then(|xx| st.stddev_i_sum_sq.checked_add(xx)),
) {
(Some(s), Some(sq)) => {
st.stddev_i_sum = s;
st.stddev_i_sum_sq = sq;
Some(())
}
_ => None,
}
} else {
None
};
if folded.is_none() {
spend_stddev_i128(st);
match crate::eval::binop::value_to_bignum(v) {
Some(b) => {
let sq = b.mul(&b);
st.stddev_sum = Some(
st.stddev_sum
.as_ref()
.map_or_else(|| b.clone(), |s| s.add(&b)),
);
st.stddev_sum_sq = Some(
st.stddev_sum_sq
.as_ref()
.map_or_else(|| sq.clone(), |s| s.add(&sq)),
);
}
None => st.stddev_saw_float = true,
}
}
}
let Some(x) = agg_value_to_f64(v) else {
return Err(EvalError::TypeMismatch {
detail: format!(
"{name} needs numeric, got {}",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
});
};
st.num.count += 1;
st.num.sum_float += x;
st.sum_sq += x * x;
}
AggKind::BitAnd | AggKind::BitOr | AggKind::BitXor => {
if is_null {
return Ok(());
}
let n = match v {
Value::Int(n) => i64::from(*n),
Value::SmallInt(n) => i64::from(*n),
Value::BigInt(n) => *n,
other => {
return Err(EvalError::TypeMismatch {
detail: format!(
"{name} needs integer, got {}",
crate::conversions::pg_type_name_for_error_opt(other.data_type())
),
});
}
};
if matches!(v, Value::BigInt(_)) {
st.bit_wide = true;
}
st.bit_acc = Some(match (st.bit_acc, kind) {
(None, _) => n,
(Some(acc), AggKind::BitAnd) => acc & n,
(Some(acc), AggKind::BitOr) => acc | n,
(Some(acc), _) => acc ^ n, });
}
AggKind::WithinGroup => {
st.within_group_rows += 1;
if is_null {
return Ok(());
}
st.items.push(v.clone().into_owned());
if let Some(k) = order_keys {
st.item_keys.extend(k);
}
st.num.count += 1;
}
AggKind::Regression => {
let (Some(y), Some(x)) = (agg_value_to_f64(v), arg2.and_then(agg_value_to_f64)) else {
return Ok(()); };
st.reg_n += 1;
let new_n = st.reg_n as f64;
let new_sx = st.reg_sx + x;
let new_sy = st.reg_sy + y;
if st.reg_n > 1 {
let n_prev = new_n - 1.0;
let tmp_x = x * new_n - new_sx;
let tmp_y = y * new_n - new_sy;
let scale = 1.0 / (n_prev * new_n);
st.reg_sxx += tmp_x * tmp_x * scale;
st.reg_syy += tmp_y * tmp_y * scale;
st.reg_sxy += tmp_x * tmp_y * scale;
}
st.reg_sx = new_sx;
st.reg_sy = new_sy;
}
AggKind::JsonAgg => {
if is_null && name.ends_with("_strict") {
return Ok(());
}
st.items.push(v.clone().into_owned());
if let Some(k) = order_keys {
st.item_keys.extend(k);
}
st.num.count += 1;
}
AggKind::JsonObjectAgg => {
if is_null {
return Ok(());
}
let val = arg2.cloned().map(Value::into_owned).unwrap_or(Value::Null);
if matches!(val, Value::Null) && name.contains("_strict") {
return Ok(());
}
if name.contains("_unique") {
let kt = match v {
Value::Text(s) | Value::Json(s) => s.to_string(),
other => crate::json::value_to_json_text(other),
};
let dup = st.items.iter().any(|k| match k {
Value::Text(s) | Value::Json(s) => *s == kt,
other => crate::json::value_to_json_text(other) == kt,
});
if dup {
return Err(EvalError::TypeMismatch {
detail: alloc::format!("duplicate JSON object key value: {kt:?}"),
});
}
}
st.items.push(v.clone().into_owned());
st.aux_items.push(val);
st.num.count += 1;
}
}
Ok(())
}
#[allow(clippy::cast_precision_loss, clippy::cast_possible_truncation)]
pub(crate) fn finalize(name: &str, st: &AggState, mysql: bool) -> Value<'static> {
match name {
"count" | "count_star" => Value::BigInt(st.num.count),
"sum" => {
if st.num.count == 0 {
Value::Null
} else if st.num.use_interval {
Value::Interval {
months: st.num.sum_iv_months as i32,
days: st.num.sum_iv_days as i32,
micros: st.num.sum_iv_micros as i64,
}
} else if st.num.use_money {
Value::Money(st.num.sum_money as i64)
} else if st.num.use_numeric {
if st.num.sum_num_kind != spg_storage::NumericKind::Finite {
Value::numeric_special(st.num.sum_num_kind)
} else if let Some(big) = &st.num.sum_big {
let tot = big.add(&spg_storage::bignum::BigNumeric::from_i128(
i128::from(st.num.sum_int),
0,
));
crate::eval::binop::bignum_to_value(tot)
} else {
let (scaled, scale) = crate::numeric::numeric_add(
st.num.sum_num_scaled,
st.num.sum_num_scale,
i128::from(st.num.sum_int),
0,
);
Value::Numeric {
scaled,
scale,
kind: spg_storage::NumericKind::Finite,
}
}
} else if st.num.use_float {
let total = st.num.sum_float + (st.num.sum_int as f64);
if st.num.float_not_real {
Value::Float(total)
} else {
#[allow(clippy::cast_possible_truncation)]
Value::Real(total as f32)
}
} else {
Value::BigInt(st.num.sum_int)
}
}
"avg" => {
if st.num.count == 0 {
Value::Null
} else if st.num.use_interval {
let n = i128::from(st.num.count);
let day_us = 86_400_000_000i128;
let months = i128::from(st.num.sum_iv_months);
let days = i128::from(st.num.sum_iv_days);
let month_out = months / n;
let mrem_days_total = (months % n) * 30; let days_from_month = mrem_days_total / n;
let mrem_frac_us = (mrem_days_total % n) * day_us / n;
let day_out = days / n;
let drem_us = (days % n) * day_us / n;
let micros = st.num.sum_iv_micros / n + mrem_frac_us + drem_us;
Value::Interval {
months: month_out as i32,
days: (day_out + days_from_month) as i32,
micros: micros as i64,
}
} else if st.num.use_money {
let n = i128::from(st.num.count);
let q =
(st.num.sum_money * 2 + if st.num.sum_money >= 0 { n } else { -n }) / (2 * n);
Value::Money(q as i64)
} else if st.num.use_numeric {
if st.num.sum_num_kind != spg_storage::NumericKind::Finite {
Value::numeric_special(st.num.sum_num_kind)
} else if let Some(big) = &st.num.sum_big {
use spg_storage::bignum::BigNumeric;
let sum_tot = big.add(&BigNumeric::from_i128(i128::from(st.num.sum_int), 0));
let cnt = BigNumeric::from_i128(i128::from(st.num.count), 0);
let rscale = crate::numeric::division_display_scale_big(&sum_tot, &cnt);
match sum_tot.div(&cnt, rscale) {
Some(q) => crate::eval::binop::bignum_to_value(q),
None => Value::Null,
}
} else {
let (sum_scaled, sum_scale) = crate::numeric::numeric_add(
st.num.sum_num_scaled,
st.num.sum_num_scale,
i128::from(st.num.sum_int),
0,
);
let (scaled, scale) = crate::numeric::numeric_avg(
sum_scaled,
sum_scale,
i128::from(st.num.count),
);
Value::Numeric {
scaled,
scale,
kind: spg_storage::NumericKind::Finite,
}
}
} else if st.num.use_float {
Value::Float((st.num.sum_float + (st.num.sum_int as f64)) / (st.num.count as f64))
} else {
let (scaled, scale) = crate::numeric::numeric_avg(
i128::from(st.num.sum_int),
0,
i128::from(st.num.count),
);
Value::Numeric {
scaled,
scale,
kind: spg_storage::NumericKind::Finite,
}
}
}
"min" | "max" | "any_value" => st.extreme.clone().unwrap_or(Value::Null),
"range_agg" => match st.extreme.clone() {
Some(Value::Multirange { kind, ranges }) => Value::Multirange {
kind,
ranges: crate::eval::binop::normalize_multirange_spans(kind, &ranges),
},
other => other.unwrap_or(Value::Null),
},
"range_intersect_agg" => st.extreme.clone().unwrap_or(Value::Null),
"string_agg" | "group_concat" | "xmlagg" => {
if st.items.is_empty() {
return Value::Null;
}
let sep = st.separator.clone().unwrap_or_else(|| {
if name == "group_concat" {
",".into()
} else {
String::new()
}
});
let per_row: Option<&[Option<String>]> =
if !st.item_seps.is_empty() && st.item_seps.len() == st.items.len() {
Some(&st.item_seps)
} else {
None
};
let mut out = String::new();
for (i, item) in st.items.iter().enumerate() {
if i > 0 {
match per_row {
Some(seps) => {
if let Some(sp) = &seps[i] {
out.push_str(sp);
}
}
None => out.push_str(&sep),
}
}
match item {
Value::Text(s) => out.push_str(s),
Value::Int(n) => out.push_str(&n.to_string()),
Value::BigInt(n) => out.push_str(&n.to_string()),
Value::SmallInt(n) => out.push_str(&n.to_string()),
Value::Float(f) => out.push_str(&f.to_string()),
Value::Bool(b) => {
out.push_str(if *b { "1" } else { "0" });
}
_ => {}
}
}
Value::text(out)
}
"array_agg" => {
if st.items.is_empty() {
return Value::Null;
}
crate::eval::values::build_array_from_values(&st.items)
}
"bool_and" | "bool_or" => st.bool_acc.map_or(Value::Null, Value::Bool),
"variance" | "var_samp" | "var_pop" | "stddev" | "stddev_samp" | "stddev_pop" => {
let n = st.num.count;
if n == 0 {
return Value::Null;
}
let nf = n as f64;
let pop = name.ends_with("_pop") || (mysql && (name == "stddev" || name == "variance"));
if !pop && n < 2 {
return Value::Null;
}
if !st.stddev_saw_float {
if let Some((sum, sum_sq)) = stddev_exact_pair(st) {
let (sum, sum_sq) = (&sum, &sum_sq);
use spg_storage::bignum::BigNumeric as BN;
let nb = BN::from_i128(i128::from(n), 0);
let numerator = nb.mul(sum_sq).sub(&sum.mul(sum));
let divisor = if pop {
nb.mul(&nb)
} else {
nb.mul(&BN::from_i128(i128::from(n - 1), 0))
};
if numerator.is_zero() || numerator.parts().0 {
return Value::Numeric {
scaled: 0,
scale: 0,
kind: spg_storage::NumericKind::Finite,
};
}
let rscale = crate::numeric::division_display_scale_big(&numerator, &divisor);
if let Some(var) = numerator.div(&divisor, rscale) {
let out = if name.starts_with("stddev") {
var.sqrt(crate::numeric::sqrt_display_scale_big(&var))
} else {
Some(var)
};
if let Some(o) = out {
return crate::eval::binop::bignum_to_value(o);
}
}
}
}
let numerator = (nf * st.sum_sq - st.num.sum_float * st.num.sum_float).max(0.0);
let divisor = if pop { nf * nf } else { nf * (nf - 1.0) };
let var = numerator / divisor;
let result = if name.starts_with("stddev") {
crate::eval::f64_sqrt(var)
} else {
var
};
Value::Float(result)
}
"bit_and" | "bit_or" | "bit_xor" => st.bit_acc.map_or(Value::Null, |acc| {
if st.bit_wide {
Value::BigInt(acc)
} else {
Value::Int(acc as i32)
}
}),
"regr_count" => Value::BigInt(st.reg_n),
"covar_pop" | "covar_samp" | "corr" | "regr_avgx" | "regr_avgy" | "regr_slope"
| "regr_intercept" | "regr_r2" | "regr_sxx" | "regr_syy" | "regr_sxy" => {
let n = st.reg_n;
if n == 0 {
return Value::Null;
}
let nf = n as f64;
let sxx = st.reg_sxx;
let syy = st.reg_syy;
let sxy = st.reg_sxy;
let avgx = st.reg_sx / nf;
let avgy = st.reg_sy / nf;
let out = match name {
"regr_avgx" => Some(avgx),
"regr_avgy" => Some(avgy),
"regr_sxx" => Some(sxx),
"regr_syy" => Some(syy),
"regr_sxy" => Some(sxy),
"covar_pop" => Some(sxy / nf),
"covar_samp" => (n >= 2).then(|| sxy / (nf - 1.0)),
"regr_slope" => (sxx != 0.0).then(|| sxy / sxx),
"regr_intercept" => (sxx != 0.0).then(|| avgy - (sxy / sxx) * avgx),
"corr" => {
let d = sxx * syy;
(d > 0.0).then(|| sxy / crate::eval::f64_sqrt(d))
}
"regr_r2" => {
if sxx == 0.0 {
None
} else if syy == 0.0 {
Some(1.0)
} else {
Some((sxy * sxy) / (sxx * syy))
}
}
_ => None,
};
out.map_or(Value::Null, Value::Float)
}
"json_agg" | "jsonb_agg" | "json_arrayagg" | "json_agg_strict" | "jsonb_agg_strict" => {
if st.items.is_empty() {
return Value::Null;
}
let mut out = String::from("[");
for (i, item) in st.items.iter().enumerate() {
if i > 0 {
out.push_str(", ");
}
out.push_str(&crate::json::value_to_json_text(item));
}
out.push(']');
let result = Value::json(out);
if name.starts_with("jsonb_agg") {
crate::json::canonicalize_value(result)
} else {
result
}
}
"json_object_agg"
| "jsonb_object_agg"
| "json_objectagg"
| "json_object_agg_strict"
| "jsonb_object_agg_strict"
| "json_object_agg_unique"
| "jsonb_object_agg_unique"
| "json_object_agg_unique_strict"
| "jsonb_object_agg_unique_strict" => {
if st.items.is_empty() {
return Value::Null;
}
let key_text = |key: &Value| -> String {
match key {
Value::Text(s) | Value::Json(s) => s.to_string(),
other => crate::json::value_to_json_text(other),
}
};
let dedup = name.starts_with("jsonb_object_agg");
let mut pairs: Vec<(String, usize)> = Vec::with_capacity(st.items.len());
for (i, key) in st.items.iter().enumerate() {
let kt = key_text(key);
if dedup {
if let Some(slot) = pairs.iter_mut().find(|(k, _)| *k == kt) {
slot.1 = i;
continue;
}
}
pairs.push((kt, i));
}
let mut out = String::from("{ ");
for (n, (kt, i)) in pairs.iter().enumerate() {
if n > 0 {
out.push_str(", ");
}
out.push_str(&crate::json::value_to_json_text(&Value::text(kt.clone())));
out.push_str(" : ");
let val = st.aux_items.get(*i).unwrap_or(&Value::Null);
out.push_str(&crate::json::value_to_json_text(val));
}
out.push_str(" }");
let result = Value::json(out);
if dedup {
crate::json::canonicalize_value(result)
} else {
result
}
}
_ => unreachable!(),
}
}
fn agg_value_to_f64(v: &Value) -> Option<f64> {
match v {
Value::Int(n) => Some(f64::from(*n)),
Value::SmallInt(n) => Some(f64::from(*n)),
Value::BigInt(n) => Some(*n as f64),
Value::Float(x) => Some(*x),
Value::Real(x) => Some(f64::from(*x)),
Value::Numeric { scaled, scale, .. } => Some(numeric_to_f64(*scaled, *scale)),
_ => None,
}
}
fn percentile_fraction_array(v: Option<&Value>) -> Option<Vec<Option<f64>>> {
match v? {
Value::FloatArray(a) => Some(a.clone()),
Value::NumericArray(a) => Some(
a.iter()
.map(|x| x.map(|(scaled, scale)| numeric_to_f64(scaled, scale)))
.collect(),
),
Value::IntArray(a) => Some(a.iter().map(|x| x.map(f64::from)).collect()),
Value::TextArray(a) => Some(
a.iter()
.map(|x| x.as_deref().and_then(|s| s.parse::<f64>().ok()))
.collect(),
),
_ => None,
}
}
fn values_to_array(picked: &[Value<'_>]) -> Value<'static> {
let owned: alloc::vec::Vec<Value<'static>> =
picked.iter().map(|v| v.clone().into_owned()).collect();
crate::eval::values::build_array_from_values(&owned)
}
#[allow(clippy::cast_precision_loss)]
fn numeric_to_f64(scaled: i128, scale: u16) -> f64 {
(scaled as f64) / (10i128.pow(u32::from(scale)) as f64)
}
#[allow(
clippy::cast_precision_loss,
clippy::cast_possible_truncation,
clippy::cast_sign_loss,
clippy::too_many_lines
)]
fn finalize_ordered_set(
name: &str,
st: &AggState,
direct: Option<&Value>,
direct_extra: &[Value<'static>],
order_by: &[spg_sql::ast::OrderBy],
mysql: bool,
) -> Result<Value<'static>, EvalError> {
let fraction = direct;
let check_fraction = |f: f64| -> Result<f64, EvalError> {
if !(0.0..=1.0).contains(&f) || f.is_nan() {
return Err(EvalError::TypeMismatch {
detail: format!("percentile value {f} is not between 0 and 1"),
});
}
Ok(f)
};
let scalar_fraction: Option<Result<f64, EvalError>> =
if matches!(name, "percentile_cont" | "percentile_disc") {
match fraction {
None | Some(Value::Null) => return Ok(Value::Null),
Some(v) => match percentile_fraction_array(Some(v)) {
Some(fracs) => {
for f in fracs.iter().flatten() {
check_fraction(*f)?;
}
None
}
None => Some(
agg_value_to_f64(v)
.ok_or_else(|| EvalError::TypeMismatch {
detail: format!(
"percentile fraction must be numeric, got {}",
crate::conversions::pg_type_name_for_error_opt(v.data_type())
),
})
.and_then(check_fraction),
),
},
}
} else {
None
};
let items = &st.items;
if items.is_empty() {
return Ok(match name {
"rank" | "dense_rank" => Value::BigInt(1),
"percent_rank" => Value::Float(0.0),
"cume_dist" => Value::Float(1.0),
_ => Value::Null,
});
}
let n = items.len();
Ok(match name {
"rank" | "dense_rank" | "percent_rank" | "cume_dist" => {
let Some(h) = fraction else {
return Ok(Value::Null);
};
let kw = order_by.len();
let multi = kw > 1 && st.item_keys.len() == items.len() * kw;
let hv: Vec<Value<'static>> = core::iter::once(h.clone().into_owned())
.chain(direct_extra.iter().cloned())
.collect();
let (desc, nulls_first) = order_by
.first()
.map_or((false, None), |o| (o.desc, o.nulls_first));
let cmp_i = |i: usize| -> core::cmp::Ordering {
if multi {
cmp_order_keys(
order_by,
&[],
&st.item_keys[i * kw..(i + 1) * kw],
&hv,
mysql,
)
} else {
crate::order_by_value_cmp_in(desc, nulls_first, &items[i], h, mysql)
}
};
let mut before: Vec<usize> = Vec::new(); let mut before_or_eq = 0usize; for i in 0..n {
match cmp_i(i) {
core::cmp::Ordering::Less => {
before.push(i);
before_or_eq += 1;
}
core::cmp::Ordering::Equal => before_or_eq += 1,
core::cmp::Ordering::Greater => {}
}
}
let nn = st.within_group_rows.max(n) as f64;
match name {
"rank" => Value::BigInt((before.len() + 1) as i64),
"dense_rank" => {
let tuple_cmp = |&x: &usize, &y: &usize| -> core::cmp::Ordering {
if multi {
cmp_order_keys(
order_by,
&[],
&st.item_keys[x * kw..(x + 1) * kw],
&st.item_keys[y * kw..(y + 1) * kw],
mysql,
)
} else {
value_cmp(&items[x], &items[y])
}
};
let mut sorted = before.clone();
sorted.sort_by(tuple_cmp);
let mut distinct = 0usize;
for (k, &i) in sorted.iter().enumerate() {
if k == 0 || tuple_cmp(&sorted[k - 1], &i) != core::cmp::Ordering::Equal {
distinct += 1;
}
}
Value::BigInt((distinct + 1) as i64)
}
"percent_rank" => Value::Float(before.len() as f64 / nn),
"cume_dist" => Value::Float((before_or_eq as f64 + 1.0) / (nn + 1.0)),
_ => unreachable!(),
}
}
"mode" => {
let (mut best_i, mut best_cnt) = (0usize, 1usize);
let (mut run_i, mut run_cnt) = (0usize, 1usize);
for i in 1..n {
if value_cmp(&items[i], &items[run_i]) == core::cmp::Ordering::Equal {
run_cnt += 1;
} else {
run_i = i;
run_cnt = 1;
}
if run_cnt > best_cnt {
best_cnt = run_cnt;
best_i = run_i;
}
}
items[best_i].clone()
}
"percentile_disc" => {
let idx_at = |f: f64| -> usize {
if f <= 0.0 {
0
} else {
(crate::eval::f64_ceil(f * n as f64) as usize)
.saturating_sub(1)
.min(n - 1)
}
};
if let Some(fracs) = percentile_fraction_array(fraction) {
let picked: Vec<Value> = fracs
.iter()
.map(|f| f.map_or(Value::Null, |f| items[idx_at(f)].clone()))
.collect();
return Ok(values_to_array(&picked));
}
let f = scalar_fraction.transpose()?.unwrap_or(0.0);
items[idx_at(f)].clone()
}
"percentile_cont" => {
if items.iter().all(|v| matches!(v, Value::Interval { .. })) {
let iv = |i: usize| -> (f64, f64, f64) {
match &items[i] {
Value::Interval {
months,
days,
micros,
} => (f64::from(*months), f64::from(*days), *micros as f64),
_ => unreachable!(),
}
};
let at = |f: f64| -> Value<'static> {
if n == 1 {
return items[0].clone();
}
let rank = f * (n as f64 - 1.0);
let lo = crate::eval::f64_floor(rank) as usize;
let hi = crate::eval::f64_ceil(rank) as usize;
let frac = rank - lo as f64;
let (lm, ld, lu) = iv(lo);
let (hm, hd, hu) = iv(hi);
let dm = (hm - lm) * frac;
let m_i = dm as i64; let rem_days = (dm - m_i as f64) * 30.0 + (hd - ld) * frac;
let d_i = rem_days as i64;
let us = (rem_days - d_i as f64) * 86_400_000_000.0 + (hu - lu) * frac;
Value::Interval {
months: (lm as i64 + m_i) as i32,
days: (ld as i64 + d_i) as i32,
micros: lu as i64 + libm::round(us) as i64,
}
};
if let Some(fracs) = percentile_fraction_array(fraction) {
let picked: Vec<Value> =
fracs.iter().map(|f| f.map_or(Value::Null, at)).collect();
return Ok(values_to_array(&picked));
}
let f = scalar_fraction.transpose()?.unwrap_or(0.0);
return Ok(at(f));
}
let Some(nums) = items
.iter()
.map(agg_value_to_f64)
.collect::<Option<Vec<f64>>>()
else {
return Ok(Value::Null); };
let at = |f: f64| -> f64 {
if n == 1 {
return nums[0];
}
let rank = f * (n as f64 - 1.0);
let lo = crate::eval::f64_floor(rank) as usize;
let hi = crate::eval::f64_ceil(rank) as usize;
let frac = rank - lo as f64;
nums[lo] + (nums[hi] - nums[lo]) * frac
};
if let Some(fracs) = percentile_fraction_array(fraction) {
return Ok(Value::FloatArray(fracs.iter().map(|f| f.map(at)).collect()));
}
let f = scalar_fraction.transpose()?.unwrap_or(0.0);
Value::Float(at(f))
}
_ => unreachable!(),
})
}
fn infer_agg_type(spec: &AggSpec, schema_cols: &[ColumnSchema]) -> DataType {
let arg_ty = spec
.arg
.as_ref()
.and_then(|a| crate::describe::describe_expr(a, schema_cols))
.map(|shape| shape.ty);
if spec.first_ordered {
return arg_ty.unwrap_or(DataType::Text);
}
match spec.name.as_str() {
"count" | "count_star" => DataType::BigInt,
"sum" => match arg_ty {
Some(DataType::Float) => DataType::Float,
Some(DataType::BigInt) => DataType::Numeric {
precision: 0,
scale: 0,
},
_ => DataType::BigInt,
},
"avg" => match arg_ty {
Some(DataType::Float) => DataType::Float,
_ => DataType::Numeric {
precision: 0,
scale: 0,
},
},
"string_agg" | "group_concat" | "xmlagg" => DataType::Text,
"array_agg" => match arg_ty {
Some(DataType::Int | DataType::SmallInt) => DataType::IntArray,
Some(DataType::BigInt) => DataType::BigIntArray,
Some(DataType::Bool) => DataType::BoolArray,
Some(DataType::Date) => DataType::DateArray,
Some(DataType::Timestamp) => DataType::TimestampArray,
Some(DataType::Timestamptz) => DataType::TimestamptzArray,
Some(DataType::Uuid) => DataType::UuidArray,
Some(DataType::Float) => DataType::FloatArray,
Some(DataType::Numeric { .. }) => DataType::NumericArray,
Some(DataType::Bytes) => DataType::BytesArray,
_ => DataType::TextArray,
},
"bool_and" | "bool_or" => DataType::Bool,
"stddev" | "stddev_samp" | "stddev_pop" | "variance" | "var_samp" | "var_pop" => {
DataType::Numeric {
precision: 0,
scale: 0,
}
}
"percentile_cont" | "covar_pop" | "covar_samp" | "corr" | "regr_avgx" | "regr_avgy"
| "regr_slope" | "regr_intercept" | "regr_r2" | "regr_sxx" | "regr_syy" | "regr_sxy" => {
DataType::Float
}
"bit_and" | "bit_or" | "bit_xor" => match arg_ty {
Some(DataType::SmallInt) => DataType::SmallInt,
Some(DataType::BigInt) => DataType::BigInt,
_ => DataType::Int,
},
"regr_count" | "rank" | "dense_rank" => DataType::BigInt,
"percent_rank" | "cume_dist" => DataType::Float,
"json_agg" | "jsonb_agg" | "json_object_agg" | "jsonb_object_agg" | "json_arrayagg"
| "json_objectagg" => DataType::Json,
_ => arg_ty.unwrap_or(DataType::Text),
}
}
fn agg_or_group_type(e: &Expr, synth: &[ColumnSchema]) -> DataType {
if let Expr::Column(c) = e
&& let Some(s) = synth.iter().find(|s| s.name == c.name)
{
return s.ty;
}
crate::describe::describe_expr(e, synth)
.map(|shape| shape.ty)
.unwrap_or(DataType::Text)
}
fn first_ungrouped_column<'a>(
e: &'a Expr,
group_exprs: &[Expr],
columns: &[ColumnSchema],
licensed: &[alloc::string::String],
) -> Option<&'a spg_sql::ast::ColumnName> {
if group_exprs.iter().any(|g| g == e) {
return None;
}
let rec = |x: &'a Expr| first_ungrouped_column(x, group_exprs, columns, licensed);
match e {
Expr::Column(c) => {
(column_ref_is_input(c, columns) && !column_is_key_determined(c, licensed)).then_some(c)
}
Expr::FunctionCall { name, .. } if is_aggregate_name(&name.to_ascii_lowercase()) => None,
Expr::AggregateOrdered { .. } => None,
Expr::ScalarSubquery(_) | Expr::Exists { .. } | Expr::InSubquery { .. } => None,
Expr::FunctionCall { args, .. } => args.iter().find_map(rec),
Expr::Binary { lhs, rhs, .. } => rec(lhs).or_else(|| rec(rhs)),
Expr::Unary { expr, .. }
| Expr::Cast { expr, .. }
| Expr::IsNull { expr, .. }
| Expr::BoolTest { expr, .. } => rec(expr),
Expr::Like { expr, pattern, .. } => rec(expr).or_else(|| rec(pattern)),
Expr::InList { expr, list, .. } => rec(expr).or_else(|| list.iter().find_map(rec)),
Expr::Case {
operand,
branches,
else_branch,
} => operand
.as_deref()
.and_then(rec)
.or_else(|| branches.iter().find_map(|(w, t)| rec(w).or_else(|| rec(t))))
.or_else(|| else_branch.as_deref().and_then(rec)),
_ => None,
}
}
fn column_ref_is_input(c: &spg_sql::ast::ColumnName, columns: &[ColumnSchema]) -> bool {
if let Some(q) = &c.qualifier {
let composite = alloc::format!("{q}.{}", c.name);
if columns
.iter()
.any(|col| col.name.eq_ignore_ascii_case(&composite))
{
return true;
}
}
columns
.iter()
.any(|col| col.name.eq_ignore_ascii_case(&c.name))
}
fn qualifiers_grouped_by_primary_key(
stmt: &SelectStatement,
group_exprs: &[Expr],
columns: &[ColumnSchema],
catalog: Option<&spg_storage::Catalog>,
) -> Vec<alloc::string::String> {
let (Some(from), Some(cat)) = (stmt.from.as_ref(), catalog) else {
return Vec::new();
};
let mut out = Vec::new();
let refs = core::iter::once(&from.primary).chain(from.joins.iter().map(|j| &j.table));
let single = from.joins.is_empty();
for tr in refs {
if tr.unnest_expr.is_some() {
continue;
}
let Some(table) = cat.get(&tr.name) else {
continue;
};
let schema = table.schema();
let Some(pk) = schema
.uniqueness_constraints
.iter()
.find(|u| u.is_primary_key && !u.columns.is_empty())
else {
continue;
};
let qual = tr.alias.as_deref().unwrap_or(tr.name.as_str());
let all_keys_grouped = pk.columns.iter().all(|&pos| {
let Some(name) = schema.columns.get(pos).map(|c| &c.name) else {
return false;
};
group_exprs.iter().any(|g| match g {
Expr::Column(c) if c.name.eq_ignore_ascii_case(name) => {
let belongs = match &c.qualifier {
Some(q) => q.eq_ignore_ascii_case(qual),
None => single,
};
belongs && column_ref_is_input(c, columns)
}
_ => false,
})
});
if all_keys_grouped {
out.push(alloc::string::String::from(qual));
if single {
out.push(alloc::string::String::new());
}
}
}
out
}
fn column_is_key_determined(
c: &spg_sql::ast::ColumnName,
licensed: &[alloc::string::String],
) -> bool {
let q = c.qualifier.as_deref().unwrap_or("");
licensed.iter().any(|l| l.eq_ignore_ascii_case(q))
}
fn wrap_loose_group_columns(
e: Expr,
group_exprs: &[Expr],
columns: &[ColumnSchema],
licensed: Option<&[alloc::string::String]>,
) -> Expr {
if group_exprs.iter().any(|g| *g == e) {
return e;
}
let wrap = |x: Expr| wrap_loose_group_columns(x, group_exprs, columns, licensed);
match e {
Expr::Column(c) => {
let claimed = column_ref_is_input(&c, columns)
&& licensed.is_none_or(|l| column_is_key_determined(&c, l));
if claimed {
Expr::FunctionCall {
name: String::from("any_value"),
args: alloc::vec![Expr::Column(c)],
}
} else {
Expr::Column(c)
}
}
Expr::FunctionCall { name, args } if is_aggregate_name(&name.to_ascii_lowercase()) => {
Expr::FunctionCall { name, args }
}
Expr::AggregateOrdered { .. } => e,
Expr::FunctionCall { name, args } => Expr::FunctionCall {
name,
args: args.into_iter().map(wrap).collect(),
},
Expr::Binary { op, lhs, rhs } => Expr::Binary {
op,
lhs: Box::new(wrap(*lhs)),
rhs: Box::new(wrap(*rhs)),
},
Expr::Unary { op, expr } => Expr::Unary {
op,
expr: Box::new(wrap(*expr)),
},
Expr::Cast { expr, target } => Expr::Cast {
expr: Box::new(wrap(*expr)),
target,
},
Expr::IsNull { expr, negated } => Expr::IsNull {
expr: Box::new(wrap(*expr)),
negated,
},
Expr::BoolTest {
expr,
value,
negated,
} => Expr::BoolTest {
expr: Box::new(wrap(*expr)),
value,
negated,
},
Expr::Like {
expr,
pattern,
negated,
case_insensitive,
} => Expr::Like {
expr: Box::new(wrap(*expr)),
pattern: Box::new(wrap(*pattern)),
negated,
case_insensitive,
},
Expr::InList {
expr,
list,
negated,
} => Expr::InList {
expr: Box::new(wrap(*expr)),
list: list.into_iter().map(wrap).collect(),
negated,
},
Expr::Case {
operand,
branches,
else_branch,
} => Expr::Case {
operand: operand.map(|o| Box::new(wrap(*o))),
branches: branches
.into_iter()
.map(|(w, t)| (wrap(w), wrap(t)))
.collect(),
else_branch: else_branch.map(|b| Box::new(wrap(*b))),
},
other => other,
}
}
fn substitute_having_aliases(e: Expr, aliases: &[(String, Expr)]) -> Expr {
use spg_sql::ast::ColumnName;
let sub = |x: Expr| substitute_having_aliases(x, aliases);
match e {
Expr::Column(ColumnName {
qualifier: None,
name,
}) => aliases
.iter()
.find(|(a, _)| a.eq_ignore_ascii_case(&name))
.map_or_else(
|| {
Expr::Column(ColumnName {
qualifier: None,
name,
})
},
|(_, expr)| expr.clone(),
),
Expr::Binary { op, lhs, rhs } => Expr::Binary {
op,
lhs: Box::new(sub(*lhs)),
rhs: Box::new(sub(*rhs)),
},
Expr::Unary { op, expr } => Expr::Unary {
op,
expr: Box::new(sub(*expr)),
},
Expr::FunctionCall { name, args } => Expr::FunctionCall {
name,
args: args.into_iter().map(sub).collect(),
},
Expr::IsNull { expr, negated } => Expr::IsNull {
expr: Box::new(sub(*expr)),
negated,
},
Expr::BoolTest {
expr,
value,
negated,
} => Expr::BoolTest {
expr: Box::new(sub(*expr)),
value,
negated,
},
Expr::Like {
expr,
pattern,
negated,
case_insensitive,
} => Expr::Like {
expr: Box::new(sub(*expr)),
pattern: Box::new(sub(*pattern)),
negated,
case_insensitive,
},
Expr::InList {
expr,
list,
negated,
} => Expr::InList {
expr: Box::new(sub(*expr)),
list: list.into_iter().map(sub).collect(),
negated,
},
Expr::Case {
operand,
branches,
else_branch,
} => Expr::Case {
operand: operand.map(|o| Box::new(sub(*o))),
branches: branches
.into_iter()
.map(|(w, t)| (sub(w), sub(t)))
.collect(),
else_branch: else_branch.map(|b| Box::new(sub(*b))),
},
Expr::Cast { expr, target } => Expr::Cast {
expr: Box::new(sub(*expr)),
target,
},
other => other,
}
}
fn rewrite_expr(e: &Expr, group_exprs: &[Expr], aggs: &[AggSpec]) -> Expr {
if let Some((arg, order_by, filter)) = first_ordered_array_agg(e) {
let arg_owned = Some(arg.clone());
let filter_owned = filter.cloned();
for (i, spec) in aggs.iter().enumerate() {
if spec.first_ordered
&& spec.name == "array_agg"
&& spec.arg == arg_owned
&& spec.order_by == *order_by
&& spec.filter == filter_owned
{
return Expr::Column(spg_sql::ast::ColumnName {
qualifier: None,
name: format!("__agg_{i}"),
});
}
}
}
if let Expr::AggregateOrdered {
call,
order_by,
distinct,
filter,
} = e
&& let Expr::FunctionCall { name, args } = call.as_ref()
{
let lower = name.to_ascii_lowercase();
if is_aggregate_name(&lower) {
let canonical: &str = if lower == "every" { "bool_and" } else { &lower };
let (arg, direct_arg) = if is_within_group_name(canonical) {
(
order_by.first().map(|o| o.expr.clone()),
args.first().cloned(),
)
} else {
(args.first().cloned(), None)
};
let arg2 = if agg_uses_second_arg(canonical) {
args.get(1).cloned()
} else {
None
};
let filter_owned = filter.as_deref().cloned();
for (i, spec) in aggs.iter().enumerate() {
if spec.name == canonical
&& spec.arg == arg
&& spec.arg2 == arg2
&& spec.distinct == *distinct
&& spec.order_by == *order_by
&& spec.filter == filter_owned
&& spec.direct_arg == direct_arg
{
return Expr::Column(spg_sql::ast::ColumnName {
qualifier: None,
name: format!("__agg_{i}"),
});
}
}
}
}
if let Expr::FunctionCall { name, args } = e {
let lower = name.to_ascii_lowercase();
if is_aggregate_name(&lower) {
let arg = if lower == "count_star" {
None
} else {
args.first().cloned()
};
let arg2 = if agg_uses_second_arg(&lower) {
args.get(1).cloned()
} else {
None
};
let canonical: &str = if lower == "every" {
"bool_and"
} else {
lower.as_str()
};
for (i, spec) in aggs.iter().enumerate() {
if spec.name == canonical
&& spec.arg == arg
&& spec.arg2 == arg2
&& !spec.distinct
&& spec.order_by.is_empty()
{
return Expr::Column(spg_sql::ast::ColumnName {
qualifier: None,
name: format!("__agg_{i}"),
});
}
}
}
}
for (i, g) in group_exprs.iter().enumerate() {
if g == e {
return Expr::Column(spg_sql::ast::ColumnName {
qualifier: None,
name: format!("__grp_{i}"),
});
}
}
match e {
Expr::NamedArg { name, expr } => Expr::NamedArg {
name: name.clone(),
expr: alloc::boxed::Box::new(rewrite_expr(expr, group_exprs, aggs)),
},
Expr::Variadic(expr) => Expr::Variadic(alloc::boxed::Box::new(rewrite_expr(
expr,
group_exprs,
aggs,
))),
Expr::AggregateOrdered {
call,
order_by,
distinct,
filter,
} => Expr::AggregateOrdered {
call: Box::new(rewrite_expr(call, group_exprs, aggs)),
distinct: *distinct,
order_by: order_by
.iter()
.map(|o| spg_sql::ast::OrderBy {
expr: rewrite_expr(&o.expr, group_exprs, aggs),
desc: o.desc,
nulls_first: o.nulls_first,
collation: o.collation.clone(),
})
.collect(),
filter: filter.clone(),
},
Expr::Binary { lhs, op, rhs } => Expr::Binary {
lhs: Box::new(rewrite_expr(lhs, group_exprs, aggs)),
op: *op,
rhs: Box::new(rewrite_expr(rhs, group_exprs, aggs)),
},
Expr::Unary { op, expr } => Expr::Unary {
op: *op,
expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
},
Expr::Cast { expr, target } => Expr::Cast {
expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
target: target.clone(),
},
Expr::FieldAccess { base, field } => Expr::FieldAccess {
base: Box::new(rewrite_expr(base, group_exprs, aggs)),
field: field.clone(),
},
Expr::IsNull { expr, negated } => Expr::IsNull {
expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
negated: *negated,
},
Expr::BoolTest {
expr,
value,
negated,
} => Expr::BoolTest {
expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
value: *value,
negated: *negated,
},
Expr::FunctionCall { name, args } => Expr::FunctionCall {
name: name.clone(),
args: args
.iter()
.map(|a| rewrite_expr(a, group_exprs, aggs))
.collect(),
},
Expr::Like {
expr,
pattern,
negated,
case_insensitive,
} => Expr::Like {
expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
pattern: Box::new(rewrite_expr(pattern, group_exprs, aggs)),
negated: *negated,
case_insensitive: *case_insensitive,
},
Expr::Extract { field, source } => Expr::Extract {
field: field.clone(),
source: Box::new(rewrite_expr(source, group_exprs, aggs)),
},
Expr::ScalarSubquery(s) => {
Expr::ScalarSubquery(Box::new(rewrite_group_keys_in_select(s, group_exprs)))
}
Expr::Exists { subquery, negated } => Expr::Exists {
subquery: Box::new(rewrite_group_keys_in_select(subquery, group_exprs)),
negated: *negated,
},
Expr::InSubquery {
expr,
subquery,
negated,
} => Expr::InSubquery {
expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
subquery: Box::new(rewrite_group_keys_in_select(subquery, group_exprs)),
negated: *negated,
},
Expr::RowInSubquery {
row,
subquery,
negated,
} => Expr::RowInSubquery {
row: row
.iter()
.map(|el| rewrite_expr(el, group_exprs, aggs))
.collect(),
subquery: Box::new(rewrite_group_keys_in_select(subquery, group_exprs)),
negated: *negated,
},
Expr::RowCmpSubquery { row, op, subquery } => Expr::RowCmpSubquery {
row: row
.iter()
.map(|el| rewrite_expr(el, group_exprs, aggs))
.collect(),
op: *op,
subquery: Box::new(rewrite_group_keys_in_select(subquery, group_exprs)),
},
Expr::WindowFunction { .. } | Expr::Literal(_) | Expr::Placeholder(_) | Expr::Column(_) => {
e.clone()
}
Expr::Array(items) => Expr::Array(
items
.iter()
.map(|elem| rewrite_expr(elem, group_exprs, aggs))
.collect(),
),
Expr::ArraySubscript { target, index } => Expr::ArraySubscript {
target: Box::new(rewrite_expr(target, group_exprs, aggs)),
index: Box::new(rewrite_expr(index, group_exprs, aggs)),
},
Expr::ArraySlice { target, lo, hi } => Expr::ArraySlice {
target: Box::new(rewrite_expr(target, group_exprs, aggs)),
lo: lo
.as_ref()
.map(|b| Box::new(rewrite_expr(b, group_exprs, aggs))),
hi: hi
.as_ref()
.map(|b| Box::new(rewrite_expr(b, group_exprs, aggs))),
},
Expr::AnyAll {
expr,
op,
array,
is_any,
} => Expr::AnyAll {
expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
op: *op,
array: Box::new(rewrite_expr(array, group_exprs, aggs)),
is_any: *is_any,
},
Expr::InList {
expr,
list,
negated,
} => Expr::InList {
expr: Box::new(rewrite_expr(expr, group_exprs, aggs)),
list: list
.iter()
.map(|item| rewrite_expr(item, group_exprs, aggs))
.collect(),
negated: *negated,
},
Expr::Case {
operand,
branches,
else_branch,
} => Expr::Case {
operand: operand
.as_deref()
.map(|o| Box::new(rewrite_expr(o, group_exprs, aggs))),
branches: branches
.iter()
.map(|(w, t)| {
(
rewrite_expr(w, group_exprs, aggs),
rewrite_expr(t, group_exprs, aggs),
)
})
.collect(),
else_branch: else_branch
.as_deref()
.map(|e| Box::new(rewrite_expr(e, group_exprs, aggs))),
},
}
}
fn rewrite_group_keys_in_select(
s: &spg_sql::ast::SelectStatement,
group_exprs: &[Expr],
) -> spg_sql::ast::SelectStatement {
let mut out = s.clone();
let _ = crate::walk_select_exprs_mut(&mut out, &mut |e| {
*e = rewrite_expr(e, group_exprs, &[]);
Ok(())
});
out
}
fn encode_one(out: &mut String, v: &Value) {
encode_one_in(out, v, false);
}
fn encode_one_in(out: &mut String, v: &Value, mysql: bool) {
use core::fmt::Write;
if mysql {
if let Value::Text(s) | Value::Json(s) = v {
let _ = write!(out, "S{}|", spg_storage::mysql_compare_fold(s));
return;
}
if let Value::BpChar(s) = v {
let folded = spg_storage::mysql_ci_fold(s.trim_end_matches(' '));
let _ = write!(out, "S{folded}|");
return;
}
}
encode_one_raw(out, v);
}
fn encode_one_raw(out: &mut String, v: &Value) {
use core::fmt::Write;
match v {
Value::Null => out.push_str("N|"),
Value::SmallInt(n) => {
let _ = write!(out, "s{n}|");
}
Value::Int(n) => {
let _ = write!(out, "I{n}|");
}
Value::BigInt(n) => {
let _ = write!(out, "B{n}|");
}
Value::Float(x) => {
let x = if *x == 0.0 { 0.0 } else { *x };
let _ = write!(out, "F{x}|");
}
Value::Real(x) => {
let x = if *x == 0.0 { 0.0 } else { *x };
let _ = write!(out, "R{x}|");
}
Value::Bool(b) => {
out.push(if *b { 'T' } else { 'f' });
out.push('|');
}
Value::Text(s) => {
out.push('S');
out.push_str(s);
out.push('|');
}
Value::BpChar(s) => {
out.push('S');
out.push_str(s.trim_end_matches(' '));
out.push('|');
}
Value::Vector(v) => {
out.push('V');
for x in v.iter() {
out.push_str(&x.to_string());
out.push(',');
}
out.push('|');
}
Value::Sq8Vector(q) => {
out.push('Q');
out.push_str(&q.min.to_string());
out.push('@');
out.push_str(&q.max.to_string());
out.push(':');
for b in &q.bytes {
out.push_str(&b.to_string());
out.push(',');
}
out.push('|');
}
Value::HalfVector(h) => {
out.push('H');
for b in &h.bytes {
out.push_str(&b.to_string());
out.push(',');
}
out.push('|');
}
Value::Numeric { scaled, scale, .. } => {
let (mut s, mut sc) = (*scaled, *scale);
while sc > 0 && s % 10 == 0 {
s /= 10;
sc -= 1;
}
out.push('D');
out.push_str(&s.to_string());
out.push('@');
out.push_str(&sc.to_string());
out.push('|');
}
Value::Date(d) => {
out.push('d');
out.push_str(&d.to_string());
out.push('|');
}
Value::Timestamp(t) => {
out.push('t');
out.push_str(&t.to_string());
out.push('|');
}
Value::Interval {
months,
days,
micros,
} => {
out.push('i');
out.push_str(&months.to_string());
out.push('m');
out.push_str(&days.to_string());
out.push('d');
out.push_str(µs.to_string());
out.push('|');
}
Value::Json(s) => {
out.push('j');
out.push_str(s);
out.push('|');
}
_ => {
out.push('?');
out.push_str(&format!("{v:?}"));
out.push('|');
}
}
}
pub(crate) fn encode_key_refs(vals: &[&Value]) -> String {
let mut out = String::new();
for v in vals {
encode_one(&mut out, v);
}
out
}
pub(crate) fn push_canonical_key(out: &mut String, v: &Value) {
use core::fmt::Write;
match v {
Value::SmallInt(n) => {
let _ = write!(out, "n{n}|");
}
Value::Int(n) => {
let _ = write!(out, "n{n}|");
}
Value::BigInt(n) => {
let _ = write!(out, "n{n}|");
}
Value::Float(f) if *f == 0.0 => out.push_str("n0|"),
Value::Float(f) => {
let _ = write!(out, "n{f}|");
}
Value::Numeric { .. } => {
let t = crate::eval::value_to_text(v);
let t = if t.contains('.') {
t.trim_end_matches('0').trim_end_matches('.')
} else {
t.as_str()
};
let _ = write!(out, "n{t}|");
}
_ => encode_one_into(out, v),
}
}
pub(crate) fn encode_canonical_key(vals: &[Value<'_>]) -> String {
let mut out = String::new();
for v in vals {
push_canonical_key(&mut out, v);
}
out
}
pub(crate) fn encode_one_into(out: &mut String, v: &Value) {
encode_one_raw(out, v);
}
pub(crate) fn encode_key_refs_into(vals: &[&Value], out: &mut String) {
encode_key_refs_into_in(vals, out, false);
}
pub(crate) fn encode_key_refs_into_in(vals: &[&Value], out: &mut String, mysql: bool) {
out.clear();
for v in vals {
encode_one_in(out, v, mysql);
}
}
pub(crate) fn encode_key(vals: &[Value<'static>]) -> String {
let mut out = String::new();
for v in vals {
encode_one(&mut out, v);
}
out
}
#[allow(clippy::cast_precision_loss)]
fn range_intersect(a: &Value<'static>, b: &Value<'static>) -> Value<'static> {
let (
Value::Range {
kind,
lower: la,
upper: ua,
lower_inc: lia,
upper_inc: uia,
empty: ea,
},
Value::Range {
lower: lb,
upper: ub,
lower_inc: lib_,
upper_inc: uib,
empty: eb,
..
},
) = (a, b)
else {
return Value::Null;
};
let kind = *kind;
let empty_range = Value::Range {
kind,
lower: None,
upper: None,
lower_inc: false,
upper_inc: false,
empty: true,
};
if *ea || *eb {
return empty_range;
}
let (lower, lower_inc) = match (la, lb) {
(None, None) => (None, false),
(Some(x), None) => (Some(x.clone()), *lia),
(None, Some(y)) => (Some(y.clone()), *lib_),
(Some(x), Some(y)) => match value_cmp(x, y) {
core::cmp::Ordering::Greater => (Some(x.clone()), *lia),
core::cmp::Ordering::Less => (Some(y.clone()), *lib_),
core::cmp::Ordering::Equal => (Some(x.clone()), *lia && *lib_),
},
};
let (upper, upper_inc) = match (ua, ub) {
(None, None) => (None, false),
(Some(x), None) => (Some(x.clone()), *uia),
(None, Some(y)) => (Some(y.clone()), *uib),
(Some(x), Some(y)) => match value_cmp(x, y) {
core::cmp::Ordering::Less => (Some(x.clone()), *uia),
core::cmp::Ordering::Greater => (Some(y.clone()), *uib),
core::cmp::Ordering::Equal => (Some(x.clone()), *uia && *uib),
},
};
if let (Some(lo), Some(up)) = (&lower, &upper) {
match value_cmp(lo, up) {
core::cmp::Ordering::Greater => return empty_range,
core::cmp::Ordering::Equal if !(lower_inc && upper_inc) => {
return empty_range;
}
_ => {}
}
}
Value::Range {
kind,
lower,
upper,
lower_inc,
upper_inc,
empty: false,
}
}
fn fold_sum_kind(
acc: spg_storage::NumericKind,
incoming: spg_storage::NumericKind,
) -> spg_storage::NumericKind {
use spg_storage::NumericKind as NK;
match (acc, incoming) {
(NK::NaN, _) | (_, NK::NaN) => NK::NaN,
(NK::Finite, k) | (k, NK::Finite) => k,
(a, b) if a == b => a,
_ => NK::NaN,
}
}
fn extreme_cmp(
enum_labels: Option<&[String]>,
a: &Value,
b: &Value,
mysql: bool,
) -> core::cmp::Ordering {
extreme_cmp_in(enum_labels, None, a, b, mysql)
}
fn extreme_cmp_in(
enum_labels: Option<&[String]>,
collation: Option<&str>,
a: &Value,
b: &Value,
mysql: bool,
) -> core::cmp::Ordering {
if let Some(labels) = enum_labels
&& let Some(ord) = crate::eval::enum_ord_cmp(labels, a, b)
{
return ord;
}
if let (Value::Text(x), Value::Text(y), Some(c)) = (a, b, collation)
&& let Some(ord) = crate::collate::compare(c, x, y)
{
return ord;
}
if mysql {
if let (Value::Text(x), Value::Text(y)) | (Value::BpChar(x), Value::BpChar(y)) = (a, b) {
return spg_storage::mysql_compare_fold(x).cmp(&spg_storage::mysql_compare_fold(y));
}
}
value_cmp(a, b)
}
fn value_cmp(a: &Value, b: &Value) -> core::cmp::Ordering {
use core::cmp::Ordering;
match (a, b) {
(Value::Null, Value::Null) => Ordering::Equal,
(Value::Null, _) => Ordering::Greater,
(_, Value::Null) => Ordering::Less,
_ => crate::orderby::value_cmp(a, b),
}
}
pub static DISTA_LITERAL_ARG2_CACHE_FIRE: core::sync::atomic::AtomicU64 =
core::sync::atomic::AtomicU64::new(0);
pub static AGGREGATE_ARRAY_AGG_ORDER_BY_FIRE: core::sync::atomic::AtomicU64 =
core::sync::atomic::AtomicU64::new(0);
pub static AGG_PER_ROW_FAST_POS: core::sync::atomic::AtomicU64 =
core::sync::atomic::AtomicU64::new(0);
pub static AGG_PER_ROW_COMPILED_HIT: core::sync::atomic::AtomicU64 =
core::sync::atomic::AtomicU64::new(0);
pub static AGG_PER_ROW_COMPILED_MISS: core::sync::atomic::AtomicU64 =
core::sync::atomic::AtomicU64::new(0);
pub static AGG_PER_ROW_EVAL_FALLBACK: core::sync::atomic::AtomicU64 =
core::sync::atomic::AtomicU64::new(0);
pub static AGG_PER_ROW_COUNT_STAR_SENTINEL: core::sync::atomic::AtomicU64 =
core::sync::atomic::AtomicU64::new(0);
#[cfg(test)]
mod value_cmp_mixed_numeric_tests {
use super::value_cmp;
use core::cmp::Ordering;
use spg_storage::Value;
fn num(scaled: i128, scale: u16) -> Value<'static> {
Value::Numeric {
scaled,
scale,
kind: spg_storage::NumericKind::Finite,
}
}
#[test]
fn numeric_vs_integer_and_float() {
assert_eq!(value_cmp(&num(250, 2), &Value::Int(5)), Ordering::Less);
assert_eq!(value_cmp(&Value::Int(5), &num(250, 2)), Ordering::Greater);
assert_eq!(
value_cmp(&num(1000, 0), &Value::SmallInt(9)),
Ordering::Greater
);
assert_eq!(value_cmp(&num(20, 1), &Value::BigInt(2)), Ordering::Equal);
assert_eq!(value_cmp(&Value::BigInt(2), &num(20, 1)), Ordering::Equal);
assert_eq!(value_cmp(&num(35, 1), &Value::Float(3.5)), Ordering::Equal);
assert_eq!(
value_cmp(&num(35, 1), &Value::Float(3.0)),
Ordering::Greater
);
assert_eq!(value_cmp(&Value::Float(1.0), &num(25, 1)), Ordering::Less);
}
#[test]
fn every_aggregate_name_classifies() {
const NAMES: &[&str] = &[
"count",
"count_star",
"sum",
"min",
"max",
"avg",
"any_value",
"range_agg",
"range_intersect_agg",
"string_agg",
"group_concat",
"xmlagg",
"array_agg",
"bool_and",
"bool_or",
"every",
"stddev",
"stddev_samp",
"stddev_pop",
"variance",
"var_samp",
"var_pop",
"bit_and",
"bit_or",
"bit_xor",
"json_agg",
"jsonb_agg",
"json_object_agg",
"jsonb_object_agg",
];
for n in NAMES {
assert!(
super::is_aggregate_name(n),
"{n} should be an aggregate name"
);
let _ = super::classify_agg_name(super::canonical_agg_name(n));
}
for n in NAMES {
assert!(
super::is_aggregate_name(&n.to_ascii_uppercase()),
"{n} should be case-insensitive"
);
}
}
}