icydb-core 0.259.6

//! Module: db::query::plan::semantics::group_having
//! Responsibility: grouped cursor policy and HAVING streaming eligibility.
//! Does not own: grouped executor runtime or generic predicate normalization outside HAVING.
//! Boundary: keeps HAVING-specific grouped semantics isolated within planning.

use crate::db::{
    cursor::CursorPlanError,
    query::plan::{
        GroupPlan,
        expr::{BinaryOp, Expr, truth_condition_binary_compare_op},
    },
};
use icydb_diagnostic_code::DiagnosticDecodeReason;

///
/// GroupedCursorPolicyViolation
///
/// Canonical grouped cursor-policy violations shared by planner and executor
/// boundaries so grouped continuation rules are not reimplemented per layer.
///

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub(in crate::db) enum GroupedCursorPolicyViolation {
    ContinuationRequiresLimit,
    GlobalDistinctContinuationUnsupported,
}

impl GroupedCursorPolicyViolation {
    /// Convert one grouped cursor-policy violation into the cursor-plan error
    /// surface used by continuation validation.
    #[must_use]
    pub(in crate::db) const fn into_cursor_plan_error(self) -> CursorPlanError {
        // A decoded cursor can still violate request policy without indicating
        // an internal planner/executor contract failure.
        let reason = match self {
            Self::ContinuationRequiresLimit => {
                DiagnosticDecodeReason::CursorGroupedContinuationRequiresLimit
            }
            Self::GlobalDistinctContinuationUnsupported => {
                DiagnosticDecodeReason::CursorGlobalDistinctContinuationUnsupported
            }
        };
        CursorPlanError::invalid_continuation_cursor_payload(reason)
    }
}

/// Return grouped cursor-policy violations for one grouped plan shape.
#[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, .. } => {
                // Streaming supports comparisons joined by AND. Other binary
                // expressions remain executable through non-streaming routes.
                truth_condition_binary_compare_op(*op).is_some() || matches!(op, BinaryOp::And)
            }
            #[cfg(test)]
            Expr::Alias { .. } => true,
        })
    })
}

// Exhaustive cache-retention coverage; new owned fields require accounting.
crate::retained::retained_copy!(GroupedCursorPolicyViolation);

///
/// TESTS
///

#[cfg(test)]
mod tests {
    use super::{GroupedCursorPolicyViolation, grouped_having_streaming_compatible};
    use crate::{
        db::query::plan::expr::{BinaryOp, CaseWhenArm, Expr},
        value::Value,
    };

    #[test]
    fn grouped_cursor_policies_preserve_request_rejection_taxonomy() {
        use crate::{
            db::QueryError,
            error::{ErrorClass, ErrorOrigin},
        };
        use icydb_diagnostic_code::{DiagnosticDecodeReason, DiagnosticFactTag, ErrorCode};

        for (policy, reason) in [
            (
                GroupedCursorPolicyViolation::ContinuationRequiresLimit,
                DiagnosticDecodeReason::CursorGroupedContinuationRequiresLimit,
            ),
            (
                GroupedCursorPolicyViolation::GlobalDistinctContinuationUnsupported,
                DiagnosticDecodeReason::CursorGlobalDistinctContinuationUnsupported,
            ),
        ] {
            let internal = policy.into_cursor_plan_error().into_internal_error();
            assert_eq!(internal.class(), ErrorClass::Unsupported);
            assert_eq!(internal.origin(), ErrorOrigin::Cursor);
            let query = QueryError::from_cursor_plan_error(policy.into_cursor_plan_error());
            assert_eq!(
                query.diagnostic().error_code(),
                ErrorCode::QUERY_INVALID_CONTINUATION_CURSOR
            );
            assert_eq!(
                query.diagnostic_facts(),
                vec![(DiagnosticFactTag::DecodeReason, reason.raw())]
            );
        }
    }

    #[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)
        );
        // Even a branch that row evaluation would skip still affects route eligibility.
        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)
        );
    }
}