use crate::db::{
cursor::CursorPlanError,
query::plan::{
GroupPlan,
expr::{BinaryOp, Expr, truth_condition_binary_compare_op},
},
};
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum GroupedCursorPolicyViolation {
ContinuationRequiresLimit,
GlobalDistinctContinuationUnsupported,
}
impl GroupedCursorPolicyViolation {
#[must_use]
pub(in crate::db) const fn into_cursor_plan_error(self) -> CursorPlanError {
let _ = self;
CursorPlanError::continuation_cursor_invariant()
}
}
#[must_use]
pub(in crate::db) fn grouped_cursor_policy_violation(
grouped: &GroupPlan,
cursor_present: bool,
) -> Option<GroupedCursorPolicyViolation> {
if !cursor_present {
return None;
}
if grouped
.scalar
.page
.as_ref()
.and_then(|page| page.limit)
.is_none()
{
return Some(GroupedCursorPolicyViolation::ContinuationRequiresLimit);
}
if grouped.is_global_distinct_aggregate_without_group_keys() {
return Some(GroupedCursorPolicyViolation::GlobalDistinctContinuationUnsupported);
}
None
}
pub(in crate::db::query::plan::semantics) fn grouped_having_streaming_compatible<E>(
having_expr: Option<&Expr>,
observe: &mut impl FnMut(u64) -> Result<(), E>,
) -> Result<bool, E> {
let Some(expr) = having_expr else {
return Ok(true);
};
expr.try_all_tree_expr(&mut |node| {
observe(1)?;
Ok(match node {
Expr::Field(_) | Expr::FieldPath(_) | Expr::Literal(_) | Expr::Aggregate(_) => true,
Expr::FunctionCall { .. } | Expr::Unary { .. } | Expr::Case { .. } => true,
Expr::Binary { op, .. } => {
truth_condition_binary_compare_op(*op).is_some() || matches!(op, BinaryOp::And)
}
#[cfg(test)]
Expr::Alias { .. } => true,
})
})
}
crate::retained::retained_copy!(GroupedCursorPolicyViolation);
#[cfg(test)]
mod tests {
use super::grouped_having_streaming_compatible;
use crate::{
db::query::plan::expr::{BinaryOp, CaseWhenArm, Expr},
value::Value,
};
#[test]
fn streaming_having_preserves_binary_operator_policy_and_short_circuiting() {
for (op, expected) in [
(BinaryOp::Eq, true),
(BinaryOp::Ne, true),
(BinaryOp::Lt, true),
(BinaryOp::Lte, true),
(BinaryOp::Gt, true),
(BinaryOp::Gte, true),
(BinaryOp::And, true),
(BinaryOp::Or, false),
(BinaryOp::Add, false),
(BinaryOp::Sub, false),
(BinaryOp::Mul, false),
(BinaryOp::Div, false),
] {
let expr = Expr::Binary {
op,
left: Box::new(Expr::Literal(Value::Bool(true))),
right: Box::new(Expr::Literal(Value::Bool(false))),
};
let mut visits = 0;
let result = grouped_having_streaming_compatible(Some(&expr), &mut |steps| {
visits += steps;
Ok::<(), ()>(())
})
.unwrap();
assert_eq!(result, expected, "{op:?}");
assert_eq!(visits, if expected { 3 } else { 1 });
}
}
#[test]
fn streaming_having_checks_nested_branches_and_propagates_observer_failure() {
let expr = Expr::Case {
when_then_arms: vec![CaseWhenArm::new(
Expr::Literal(Value::Bool(true)),
Expr::Literal(Value::Bool(true)),
)],
else_expr: Box::new(Expr::Binary {
op: BinaryOp::Or,
left: Box::new(Expr::Literal(Value::Bool(true))),
right: Box::new(Expr::Literal(Value::Bool(false))),
}),
};
assert_eq!(
grouped_having_streaming_compatible(Some(&expr), &mut |_| Ok::<(), ()>(())),
Ok(false)
);
let mut visits = 0;
assert_eq!(
grouped_having_streaming_compatible(Some(&expr), &mut |_| {
visits += 1;
if visits == 3 { Err(()) } else { Ok(()) }
}),
Err(())
);
assert_eq!(visits, 3);
assert_eq!(
grouped_having_streaming_compatible::<()>(None, &mut |_| Err(())),
Ok(true)
);
}
}