Function check_subquery_expr

datafusion/expr/src/logical_plan/invariants.rs:157–266

Do necessary check on subquery expressions and fail the invalid plan 1) Check whether the outer plan is in the allowed outer plans list to use subquery expressions, the allowed while list: [Projection, Filter, Window, Aggregate, Join]. 2) Check whether the inner plan is in the allowed inner plans list to use correlated(outer) expressions. 3) Check and validate unsupported cases to use the correlat

(
    outer_plan: &LogicalPlan,
    inner_plan: &LogicalPlan,
    expr: &Expr,
)

///    For example, we do not want to support to use correlated expressions as the Join conditions in the subquery plan when the Join
///    is a Full Out Join
pub fn check_subquery_expr(
    outer_plan: &LogicalPlan,
    inner_plan: &LogicalPlan,
    expr: &Expr,
) -> Result<()> {
    assert_subqueries_are_valid(inner_plan)?;
    if let Expr::ScalarSubquery(subquery) = expr {
        // Scalar subquery should only return one column
        if subquery.subquery.schema().fields().len() > 1 {
            return plan_err!(
                "Scalar subquery should only return one column, but found {}: {}",
                subquery.subquery.schema().fields().len(),
                subquery.subquery.schema().field_names().join(", ")
            );
        }
        // Correlated scalar subquery must be aggregated to return at most one row
        if !subquery.outer_ref_columns.is_empty() {
            match strip_inner_query(inner_plan) {
                LogicalPlan::Aggregate(agg) => {
                    check_aggregation_in_scalar_subquery(inner_plan, agg)
                }
                LogicalPlan::Filter(Filter { input, .. })
                    if matches!(input.as_ref(), LogicalPlan::Aggregate(_)) =>
                {
                    if let LogicalPlan::Aggregate(agg) = input.as_ref() {
                        check_aggregation_in_scalar_subquery(inner_plan, agg)
                    } else {
                        Ok(())
                    }
                }
                _ => {
                    if inner_plan
                        .max_rows()
                        .filter(|max_row| *max_row <= 1)
                        .is_some()
                    {
                        Ok(())
                    } else {
                        plan_err!(
                            "Correlated scalar subquery must be aggregated to return at most one row"
                        )
                    }
                }
            }?;
            match outer_plan {
                LogicalPlan::Projection(_) | LogicalPlan::Filter(_) => Ok(()),
                LogicalPlan::Aggregate(Aggregate {
                    group_expr,
                    aggr_expr,
                    ..
                }) => {
                    if group_expr.contains(expr) && !aggr_expr.contains(expr) {
                        // TODO revisit this validation logic
                        plan_err!(
                            "Correlated scalar subquery in the GROUP BY clause must \
                            also be in the aggregate expressions"
                        )
                    } else {