linkrs-query 0.1.0

//! Cost-based subquery decorrelation optimizer.

use crate::optimizer::analysis::BatchPlanAnalysis;
use crate::optimizer::cost::SelectivityEstimator;
use crate::optimizer::cost_based::row_estimates::estimate_node_output_rows_corrected;
use crate::optimizer::stats::feedback::cardinality::CardinalityFeedbackManager;
use crate::optimizer::stats::feedback::decision::DecorrelationAdvice;
use crate::optimizer::stats::StatsView;
use crate::planning::plan::core::nodes::base::plan_node_traits::SingleInputNode;
use crate::planning::plan::core::nodes::PlanNodeEnum;
use crate::planning::plan::core::nodes::{PatternApplyNode, SemiJoinNode};
use linkrs_core::types::operators::BinaryOperator;
use linkrs_core::Expression;

use super::types::{KeepReason, UnnestDecision, UnnestReason, MAX_BOUNDED_SUBQUERY_ROWS};

/// Subquery decorrelation optimizer
///
/// Based on batch plan analysis and statistical information, a decision is
/// made as to whether to convert PatternApply to SemiJoin/AntiJoin.
#[derive(Debug, Clone)]
pub struct SubqueryUnnestingOptimizer {
    /// The maximum number of estimated rows allowed for a subquery
    max_subquery_rows: u64,
    /// The maximum allowable complexity of the expression
    max_complexity: u32,
}

impl Default for SubqueryUnnestingOptimizer {
    fn default() -> Self {
        Self::new()
    }
}

impl SubqueryUnnestingOptimizer {
    /// Create a new optimizer.
    pub fn new() -> Self {
        Self {
            max_subquery_rows: 1000,
            max_complexity: 50,
        }
    }

    /// Set a threshold for the maximum number of rows in subqueries
    pub fn with_max_rows(mut self, max_rows: u64) -> Self {
        self.max_subquery_rows = max_rows;
        self
    }

    /// Set a threshold for the maximum complexity.
    pub fn with_max_complexity(mut self, max_complexity: u32) -> Self {
        self.max_complexity = max_complexity;
        self
    }

    /// Determine whether decorrelation should be performed.
    ///
    /// # Parameters
    /// `pattern_apply`: The PatternApply node
    /// `analysis`: The batch plan analysis result
    /// `stats`: The space-scoped statistics for the query
    /// `selectivity`: The selectivity estimator used for filter row estimates
    /// `cardinality`: The learned per-shape cardinality corrections (may be
    ///   consulted to correct the row estimates of both inputs)
    /// `advice`: Measured execution feedback of both decorrelation paths
    ///
    /// # Decision
    /// 1. Empirical feedback: when past executions of the same shape have
    ///    measured one path clearly faster, follow the measurement.
    /// 2. Otherwise compare corrected cost estimates.
    pub fn should_unnest(
        &self,
        pattern_apply: &PatternApplyNode,
        analysis: &BatchPlanAnalysis,
        stats: &StatsView,
        selectivity: &SelectivityEstimator,
        cardinality: &CardinalityFeedbackManager,
        advice: &DecorrelationAdvice,
    ) -> UnnestDecision {
        // 1. Check determinism from batch analysis
        if !analysis.expression_summary.is_fully_deterministic {
            return UnnestDecision::KeepPatternApply {
                reason: KeepReason::NonDeterministic,
            };
        }

        // 2. Check complexity from batch analysis
        if analysis.expression_summary.total_complexity > self.max_complexity {
            return UnnestDecision::KeepPatternApply {
                reason: KeepReason::ComplexCondition,
            };
        }

        // 3. Checking subqueries for simplicity (shape). The static
        // constant-limit cap is waived here: the CBO row estimate in step 4
        // bounds the subquery size with statistics instead.
        if !Self::is_simple_subquery_shape_with_row_bound(pattern_apply.right_input(), i64::MAX) {
            return UnnestDecision::KeepPatternApply {
                reason: KeepReason::TooComplex,
            };
        }

        // 3b. Aggregated subqueries are not equi semi joins; keep the apply.
        if Self::contains_aggregation(pattern_apply.right_input()) {
            return UnnestDecision::KeepPatternApply {
                reason: KeepReason::AggregateSubquery,
            };
        }

        // 3c. Empirical feedback: measured execution evidence overrides the
        // static cost comparison once enough runs have been observed.
        if advice.prefer_unnest && advice.confidence >= 0.7 {
            return UnnestDecision::ShouldUnnest {
                reason: UnnestReason::Empirical,
                original_cost: 0.0,
                unnested_cost: 0.0,
            };
        }
        if advice.prefer_keep && advice.confidence >= 0.7 {
            return UnnestDecision::KeepPatternApply {
                reason: KeepReason::EmpiricalKeep,
            };
        }

        // 4. Checking the number of estimated rows for subqueries
        let estimated_rows = self.estimate_subquery_rows(
            pattern_apply.right_input(),
            stats,
            selectivity,
            cardinality,
        );
        if estimated_rows > self.max_subquery_rows {
            return UnnestDecision::KeepPatternApply {
                reason: KeepReason::TooManyRows,
            };
        }

        // 5. Comparison of costs
        let left_rows = estimate_node_output_rows_corrected(
            pattern_apply.left_input(),
            stats,
            selectivity,
            cardinality,
        )
        .max(1) as f64;
        let original_cost =
            self.estimate_pattern_apply_cost(left_rows, estimated_rows, advice.apply_cost_per_row);
        let unnested_cost = self.estimate_hash_join_cost(left_rows, estimated_rows);

        if unnested_cost < original_cost {
            UnnestDecision::ShouldUnnest {
                reason: UnnestReason::CostBased,
                original_cost,
                unnested_cost,
            }
        } else {
            UnnestDecision::ShouldUnnest {
                reason: UnnestReason::SimpleSubquery,
                original_cost,
                unnested_cost,
            }
        }
    }

    /// Check whether the subquery has a decorrelatable shape: a scan of a
    /// single table (vertex/edge/index) — optionally a simple two-table
    /// inner-join tree — wrapped in equality / disjunctive-equality filters
    /// and projections, capped by a constant `Limit` within
    /// [`MAX_BOUNDED_SUBQUERY_ROWS`].
    ///
    /// Shared with the heuristic decorrelation rule (stat-free shape gate).
    pub fn is_simple_subquery_shape(node: &PlanNodeEnum) -> bool {
        Self::is_simple_subquery_shape_with_row_bound(node, MAX_BOUNDED_SUBQUERY_ROWS)
    }

    /// Shape gate with a configurable constant row bound.
    ///
    /// The stat-free heuristic rule passes [`MAX_BOUNDED_SUBQUERY_ROWS`];
    /// the cost-based decision (`should_unnest`) passes `i64::MAX` and lets
    /// the CBO row estimate (step 4 of the decision) bound the subquery
    /// size instead of the static limit cap.
    pub fn is_simple_subquery_shape_with_row_bound(node: &PlanNodeEnum, max_rows: i64) -> bool {
        match node {
            // Single table scans
            PlanNodeEnum::ScanVertices(_)
            | PlanNodeEnum::ScanEdges(_)
            | PlanNodeEnum::IndexScan(_) => true,

            // Constant-limit cap keeps the subquery bounded without
            // statistics; non-constant limits (parameter/expression based)
            // are rejected.
            PlanNodeEnum::Limit(n) => {
                let count = n.count();
                (0..=max_rows).contains(&count)
                    && Self::is_simple_subquery_shape_with_row_bound(
                        SingleInputNode::input(n),
                        max_rows,
                    )
            }

            // Simple filtration (equalities or disjunctions thereof).
            PlanNodeEnum::Filter(n) => {
                let condition = n.condition();
                if let Some(expr_meta) = condition.expression() {
                    if !Self::is_simple_equality_condition(expr_meta.inner()) {
                        return false;
                    }
                }
                Self::is_simple_subquery_shape_with_row_bound(SingleInputNode::input(n), max_rows)
            }

            // Simple projection
            PlanNodeEnum::Project(n) => {
                Self::is_simple_subquery_shape_with_row_bound(SingleInputNode::input(n), max_rows)
            }

            // Simple two-table inner join tree: both sides must be simple
            // shapes themselves; deeper join trees are accepted by
            // recursion only while every level stays a two-input inner join
            // over simple inputs.
            PlanNodeEnum::InnerJoin(n) => {
                Self::is_simple_subquery_shape_with_row_bound(n.left_input(), max_rows)
                    && Self::is_simple_subquery_shape_with_row_bound(n.right_input(), max_rows)
            }

            // Not supported in other cases
            _ => false,
        }
    }

    /// Check whether the subquery has a Mark-Join decorrelatable shape.
    ///
    /// The Mark-Join path keeps the subquery plan as the join's right side
    /// and moves the correlated residual into the join condition, so the
    /// right side only needs to be a single-table scan wrapped in arbitrary
    /// (subquery-local) filters and projections, optionally capped by a
    /// constant `Limit`. Unlike [`Self::is_simple_subquery_shape`] the inner
    /// filter conditions may be non-equality: they stay as residual filters
    /// on the materialized right side rather than becoming equi keys.
    pub fn is_mark_join_shape(node: &PlanNodeEnum) -> bool {
        match node {
            // Single table scans
            PlanNodeEnum::ScanVertices(_)
            | PlanNodeEnum::ScanEdges(_)
            | PlanNodeEnum::IndexScan(_) => true,

            // Constant-limit cap keeps the subquery bounded.
            PlanNodeEnum::Limit(n) => {
                let count = n.count();
                (0..=MAX_BOUNDED_SUBQUERY_ROWS).contains(&count)
                    && Self::is_mark_join_shape(SingleInputNode::input(n))
            }

            // Any deterministic subquery-local filter is fine.
            PlanNodeEnum::Filter(n) => Self::is_mark_join_shape(SingleInputNode::input(n)),

            // Simple projection
            PlanNodeEnum::Project(n) => Self::is_mark_join_shape(SingleInputNode::input(n)),

            // Not supported in other cases
            _ => false,
        }
    }

    /// Check whether the subquery input is provably bounded: a plain leaf
    /// scan or a constant `Limit` wrapper with a literal count within
    /// [`MAX_BOUNDED_SUBQUERY_ROWS`].
    pub fn has_bounded_input(node: &PlanNodeEnum) -> bool {
        match node {
            PlanNodeEnum::ScanVertices(_)
            | PlanNodeEnum::ScanEdges(_)
            | PlanNodeEnum::IndexScan(_) => true,
            PlanNodeEnum::Limit(n) => {
                let count = n.count();
                (0..=MAX_BOUNDED_SUBQUERY_ROWS).contains(&count)
            }
            _ => false,
        }
    }

    /// Whether the subquery contains an aggregation anywhere in its shape.
    ///
    /// Shared with the heuristic decorrelation rule (stat-free shape gate).
    pub fn contains_aggregation(node: &PlanNodeEnum) -> bool {
        match node {
            PlanNodeEnum::Aggregate(_) | PlanNodeEnum::Window(_) => true,
            _ => node.children().into_iter().any(Self::contains_aggregation),
        }
    }

    /// Check whether the condition is a simple equality comparison, an AND
    /// chain of such comparisons, or a disjunction (OR) of them — i.e. the
    /// correlated condition is expressible as a DNF of equality conjuncts.
    fn is_simple_equality_condition(expr: &Expression) -> bool {
        match expr {
            Expression::Binary { op, left, right } => match op {
                BinaryOperator::Equal => {
                    Self::is_simple_expression(left.as_ref())
                        && Self::is_simple_expression(right.as_ref())
                }
                BinaryOperator::And | BinaryOperator::Or => {
                    Self::is_simple_equality_condition(left.as_ref())
                        && Self::is_simple_equality_condition(right.as_ref())
                }
                _ => false,
            },
            _ => false,
        }
    }

    /// Check whether the expression is simple (consisting of literals,
    /// variables, or properties).
    fn is_simple_expression(expr: &Expression) -> bool {
        matches!(
            expr,
            Expression::Literal(_) | Expression::Variable(_) | Expression::Property { .. }
        )
    }

    /// Estimating the number of rows returned by a subquery
    ///
    /// Delegates to the shared cost-based estimator: leaf scans use the tag
    /// statistics, filters apply `SelectivityEstimator::estimate_from_expression`
    /// (EWMA-corrected), limits cap the input, and learned per-shape
    /// cardinality corrections are applied.
    fn estimate_subquery_rows(
        &self,
        node: &PlanNodeEnum,
        stats: &StatsView,
        selectivity: &SelectivityEstimator,
        cardinality: &CardinalityFeedbackManager,
    ) -> u64 {
        estimate_node_output_rows_corrected(node, stats, selectivity, cardinality)
    }

    /// Estimate the cost of applying the PatternApply method
    ///
    /// `measured_per_row_cost` replaces the fixed subquery-startup coefficient
    /// when execution feedback has measured the nested-loop path.
    fn estimate_pattern_apply_cost(
        &self,
        left_rows: f64,
        subquery_rows: u64,
        measured_per_row_cost: Option<f64>,
    ) -> f64 {
        // Nested loops that execute the subquery once per left row; the
        // default coefficient models subquery startup + execution, a measured
        // value (us per row) is used when feedback is available.
        match measured_per_row_cost {
            Some(cost) => left_rows * cost * 0.001,
            None => left_rows * (subquery_rows as f64 * 0.1),
        }
    }

    /// Estimating the cost of a HashJoin operation
    fn estimate_hash_join_cost(&self, left_rows: f64, subquery_rows: u64) -> f64 {
        // Simplified estimation: hash connections
        let right_rows = subquery_rows as f64;

        // Cost of building the hash table + cost of probing
        let build_cost = right_rows;
        let probe_cost = left_rows * 0.5; // Hash detection is fast.

        build_cost + probe_cost
    }

    /// Perform the de-association transformation.
    ///
    /// # Parameters
    /// - `pattern_apply`: PatternApply node
    ///
    /// # Returns
    /// The transformed SemiJoin (EXISTS) / AntiJoin (NOT EXISTS) node
    pub fn unnest(
        &self,
        pattern_apply: PatternApplyNode,
    ) -> Result<PlanNodeEnum, crate::planning::planner::PlannerError> {
        Self::build_semi_join_from_pattern_apply(pattern_apply)
    }

    /// Shared transformation: PatternApply → SemiJoin / AntiJoin.
    ///
    /// PatternApply has semi-join semantics: it keeps the left rows that
    /// have (EXISTS) / do not have (NOT EXISTS) a matching right row. The
    /// old InnerJoin conversion was unsound for both cases (duplicated left
    /// rows and leaked right columns).
    ///
    /// The keys are a direct passthrough: `PatternApplyNode` already carries
    /// per-side key expressions (`hash_keys` for the left/outer layout,
    /// `probe_keys` for the right/subquery layout), matching the
    /// `SemiJoinNode` key convention.
    pub fn build_semi_join_from_pattern_apply(
        pattern_apply: PatternApplyNode,
    ) -> Result<PlanNodeEnum, crate::planning::planner::PlannerError> {
        let left_input = pattern_apply.left_input().clone();
        let right_input = pattern_apply.right_input().clone();

        let join_node = if pattern_apply.is_anti_predicate() {
            SemiJoinNode::new_anti(
                left_input,
                right_input,
                pattern_apply.hash_keys().to_vec(),
                pattern_apply.probe_keys().to_vec(),
            )?
        } else {
            SemiJoinNode::new_semi(
                left_input,
                right_input,
                pattern_apply.hash_keys().to_vec(),
                pattern_apply.probe_keys().to_vec(),
            )?
        };

        Ok(PlanNodeEnum::SemiJoin(join_node))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::optimizer::analysis::BatchPlanAnalyzer;
    use crate::optimizer::stats::StatisticsManager;
    use crate::planning::plan::core::nodes::access::graph_scan_node::ScanVerticesNode;
    use crate::planning::plan::core::nodes::operation::filter_node::FilterNode;
    use crate::planning::plan::core::nodes::operation::sort_node::LimitNode;
    use crate::planning::plan::core::nodes::PlanNodeEnum;
    use linkrs_core::types::expr::expression_context::ExpressionAnalysisContext;
    use linkrs_core::types::expr::ExpressionMeta;
    use linkrs_core::types::ContextualExpression;
    use std::sync::Arc;

    fn test_scan() -> PlanNodeEnum {
        let mut scan = ScanVerticesNode::new(1, "test");
        scan.set_tag("person");
        PlanNodeEnum::ScanVertices(scan)
    }

    fn test_selectivity() -> SelectivityEstimator {
        SelectivityEstimator::new(std::sync::Arc::new(StatisticsManager::new()))
    }

    #[test]
    fn test_optimizer_creation() {
        let optimizer = SubqueryUnnestingOptimizer::new();
        assert_eq!(optimizer.max_subquery_rows, 1000);
        assert_eq!(optimizer.max_complexity, 50);
    }

    #[test]
    fn test_optimizer_with_config() {
        let _optimizer = SubqueryUnnestingOptimizer::new()
            .with_max_rows(500)
            .with_max_complexity(30);
    }

    #[test]
    fn test_simple_expression_check() {
        let literal = Expression::Literal(linkrs_core::Value::Int(42));
        assert!(SubqueryUnnestingOptimizer::is_simple_expression(&literal));

        let variable = Expression::Variable("n".to_string());
        assert!(SubqueryUnnestingOptimizer::is_simple_expression(&variable));

        let property = Expression::Property {
            object: Box::new(Expression::Variable("n".to_string())),
            property: "name".to_string(),
        };
        assert!(SubqueryUnnestingOptimizer::is_simple_expression(&property));

        let binary = Expression::Binary {
            left: Box::new(Expression::Literal(linkrs_core::Value::Int(1))),
            op: linkrs_core::types::operators::BinaryOperator::Add,
            right: Box::new(Expression::Literal(linkrs_core::Value::Int(2))),
        };
        assert!(!SubqueryUnnestingOptimizer::is_simple_expression(&binary));
    }

    #[test]
    fn test_shape_accepts_scans_and_limits() {
        assert!(SubqueryUnnestingOptimizer::is_simple_subquery_shape(
            &test_scan()
        ));

        let limit = LimitNode::new(test_scan(), 0, 10).expect("limit should build");
        assert!(SubqueryUnnestingOptimizer::is_simple_subquery_shape(
            &PlanNodeEnum::Limit(limit)
        ));
        assert!(SubqueryUnnestingOptimizer::has_bounded_input(
            &PlanNodeEnum::Limit(LimitNode::new(test_scan(), 0, 10).expect("limit should build"))
        ));

        let large = LimitNode::new(test_scan(), 0, 5000).expect("limit should build");
        assert!(!SubqueryUnnestingOptimizer::has_bounded_input(
            &PlanNodeEnum::Limit(large)
        ));
    }

    #[test]
    fn test_shape_rejects_aggregation() {
        use crate::planning::plan::core::nodes::AggregateNode;
        let aggregate =
            AggregateNode::new(test_scan(), vec![], vec![]).expect("aggregate should build");
        assert!(SubqueryUnnestingOptimizer::contains_aggregation(
            &PlanNodeEnum::Aggregate(aggregate)
        ));
    }

    #[test]
    fn test_shape_accepts_disjunctive_equality_filter() {
        use linkrs_core::types::expr::expression_context::ExpressionAnalysisContext;
        use linkrs_core::types::ContextualExpression;
        let condition = {
            let ctx = Arc::new(ExpressionAnalysisContext::new());
            let prop = |name: &str| Expression::Property {
                object: Box::new(Expression::Variable("n".to_string())),
                property: name.to_string(),
            };
            let expr = Expression::Binary {
                left: Box::new(Expression::Binary {
                    left: Box::new(prop("city")),
                    op: BinaryOperator::Equal,
                    right: Box::new(Expression::Literal(linkrs_core::Value::Int(1))),
                }),
                op: BinaryOperator::Or,
                right: Box::new(Expression::Binary {
                    left: Box::new(prop("age")),
                    op: BinaryOperator::Equal,
                    right: Box::new(Expression::Literal(linkrs_core::Value::Int(2))),
                }),
            };
            let id = ctx.register_expression(linkrs_core::types::expr::ExpressionMeta::new(expr));
            ContextualExpression::new(id, ctx)
        };
        // OR of equalities is accepted by the relaxed gate.
        let or_filter = PlanNodeEnum::Filter(
            FilterNode::new(test_scan(), condition.clone()).expect("filter should build"),
        );
        assert!(SubqueryUnnestingOptimizer::is_simple_subquery_shape(
            &or_filter
        ));

        // OR containing a non-equality comparison is still rejected.
        let range_condition = {
            let ctx = Arc::new(ExpressionAnalysisContext::new());
            let expr = Expression::Binary {
                left: Box::new(Expression::Property {
                    object: Box::new(Expression::Variable("n".to_string())),
                    property: "age".to_string(),
                }),
                op: BinaryOperator::GreaterThan,
                right: Box::new(Expression::Literal(linkrs_core::Value::Int(30))),
            };
            let id = ctx.register_expression(linkrs_core::types::expr::ExpressionMeta::new(expr));
            ContextualExpression::new(id, ctx)
        };
        let range_filter = PlanNodeEnum::Filter(
            FilterNode::new(test_scan(), range_condition).expect("filter should build"),
        );
        assert!(!SubqueryUnnestingOptimizer::is_simple_subquery_shape(
            &range_filter
        ));
    }

    #[test]
    fn test_shape_accepts_two_table_inner_join() {
        use crate::planning::plan::core::nodes::join::join_node::InnerJoinNode;
        use linkrs_core::types::expr::expression_context::ExpressionAnalysisContext;
        use linkrs_core::types::ContextualExpression;
        let key = || {
            let ctx = Arc::new(ExpressionAnalysisContext::new());
            let expr = Expression::Binary {
                left: Box::new(Expression::Property {
                    object: Box::new(Expression::Variable("a".to_string())),
                    property: "id".to_string(),
                }),
                op: BinaryOperator::Equal,
                right: Box::new(Expression::Property {
                    object: Box::new(Expression::Variable("b".to_string())),
                    property: "id".to_string(),
                }),
            };
            let id = ctx.register_expression(linkrs_core::types::expr::ExpressionMeta::new(expr));
            ContextualExpression::new(id, ctx)
        };
        let join = PlanNodeEnum::InnerJoin(
            InnerJoinNode::new(test_scan(), test_scan(), vec![key()], vec![key()])
                .expect("inner join should build"),
        );
        assert!(SubqueryUnnestingOptimizer::is_simple_subquery_shape(&join));

        // A join over a non-simple side (aggregation) stays rejected.
        use crate::planning::plan::core::nodes::AggregateNode;
        let agg_side = PlanNodeEnum::Aggregate(
            AggregateNode::new(test_scan(), vec![], vec![]).expect("aggregate should build"),
        );
        let bad_join = PlanNodeEnum::InnerJoin(
            InnerJoinNode::new(agg_side, test_scan(), vec![key()], vec![key()])
                .expect("inner join should build"),
        );
        assert!(!SubqueryUnnestingOptimizer::is_simple_subquery_shape(
            &bad_join
        ));
    }

    #[test]
    fn test_row_bound_is_configurable_for_cbo() {
        // A large constant limit exceeds the stat-free heuristic bound but is
        // admitted by the CBO path (the row estimate bounds it instead).
        let large = PlanNodeEnum::Limit(
            crate::planning::plan::core::nodes::operation::sort_node::LimitNode::new(
                test_scan(),
                0,
                5000,
            )
            .expect("limit should build"),
        );
        assert!(!SubqueryUnnestingOptimizer::is_simple_subquery_shape(
            &large
        ));
        assert!(
            SubqueryUnnestingOptimizer::is_simple_subquery_shape_with_row_bound(&large, i64::MAX)
        );
    }

    #[test]
    fn test_simple_subquery_unnests() {
        let scan = test_scan();
        let condition = {
            let ctx = Arc::new(ExpressionAnalysisContext::new());
            let expr = Expression::Binary {
                left: Box::new(Expression::Property {
                    object: Box::new(Expression::Variable("n".to_string())),
                    property: "age".to_string(),
                }),
                op: BinaryOperator::Equal,
                right: Box::new(Expression::Literal(linkrs_core::Value::Int(18))),
            };
            let id = ctx.register_expression(ExpressionMeta::new(expr));
            ContextualExpression::new(id, ctx)
        };
        let filter = PlanNodeEnum::Filter(FilterNode::new(scan, condition).expect("filter"));

        let pattern_apply = PatternApplyNode::new(test_scan(), filter, vec![], vec![], false)
            .expect("pattern apply should build");

        let analysis =
            BatchPlanAnalyzer::new().analyze(&PlanNodeEnum::PatternApply(pattern_apply.clone()));
        let selectivity = test_selectivity();
        let stats_manager = StatisticsManager::new();
        let stats = StatsView::new(&stats_manager, Some("test"));
        let decision = SubqueryUnnestingOptimizer::new().should_unnest(
            &pattern_apply,
            &analysis,
            &stats,
            &selectivity,
            &CardinalityFeedbackManager::new(),
            &DecorrelationAdvice::default(),
        );
        assert!(
            matches!(decision, UnnestDecision::ShouldUnnest { .. }),
            "a deterministic single-scan equality-filtered subquery must unnest"
        );
    }

    #[test]
    fn test_unnest_produces_semi_join_with_split_keys() {
        let ctx = Arc::new(ExpressionAnalysisContext::new());
        let hash_key_expr = Expression::Property {
            object: Box::new(Expression::Variable("t".to_string())),
            property: "city".to_string(),
        };
        let probe_key_expr = Expression::Property {
            object: Box::new(Expression::Variable("p".to_string())),
            property: "city".to_string(),
        };
        let hash_key_id = ctx.register_expression(ExpressionMeta::new(hash_key_expr));
        let hash_key = ContextualExpression::new(hash_key_id, ctx.clone());
        let probe_key_id = ctx.register_expression(ExpressionMeta::new(probe_key_expr));
        let probe_key = ContextualExpression::new(probe_key_id, ctx);

        let pattern_apply = PatternApplyNode::new(
            test_scan(),
            test_scan(),
            vec![hash_key],
            vec![probe_key],
            false,
        )
        .expect("pattern apply should build");

        let transformed = SubqueryUnnestingOptimizer::new()
            .unnest(pattern_apply)
            .expect("unnest should succeed");
        match &transformed {
            PlanNodeEnum::SemiJoin(join) => {
                assert!(!join.is_anti());
                assert_eq!(join.hash_keys().len(), 1);
                assert_eq!(join.probe_keys().len(), 1);
                let hash_meta = join.hash_keys()[0].expression().expect("hash key meta");
                let probe_meta = join.probe_keys()[0].expression().expect("probe key meta");
                assert_eq!(
                    hash_meta.inner(),
                    &Expression::Property {
                        object: Box::new(Expression::Variable("t".to_string())),
                        property: "city".to_string(),
                    }
                );
                assert_eq!(
                    probe_meta.inner(),
                    &Expression::Property {
                        object: Box::new(Expression::Variable("p".to_string())),
                        property: "city".to_string(),
                    }
                );
            }
            _ => panic!("expected SemiJoin, got {:?}", transformed.name()),
        }
    }

    #[test]
    fn test_unnest_anti_produces_anti_join() {
        let pattern_apply = PatternApplyNode::new(test_scan(), test_scan(), vec![], vec![], true)
            .expect("pattern apply should build");

        let transformed = SubqueryUnnestingOptimizer::new()
            .unnest(pattern_apply)
            .expect("unnest should succeed");
        match &transformed {
            PlanNodeEnum::SemiJoin(join) => assert!(join.is_anti()),
            _ => panic!("expected AntiJoin, got {:?}", transformed.name()),
        }
    }
}