fix(semantics): stop reading a lambda parameter as an ungrouped column

The grouping check walked into lambda bodies and reported a lambda's own
parameter, such as e in list_filter(ARRAY_AGG(x), lambda e: e <> 'null'),
as a column that must be grouped. A parameter is bound only inside its
lambda's body, so it is now exempt there, while outer columns in the body
and a column of the same name outside the lambda are still checked.

Author: kvlonge

crates/sqlquack-semantics/src/duckdb/profile/tests/test_statement_findings.rs

@@ -523,6 +523,84 @@ fn given_from_items_that_read_each_other_when_resolving_names_then_each_reads_on
     }
 }
 
+#[test]
+fn given_a_lambda_in_a_grouped_select_when_checking_grouping_then_only_outer_columns_must_be_grouped()
+ {
+    // Each verdict is DuckDB 2.0.0's.
+    let test_cases = [
+        QueryFindingsTestCase {
+            description: "a lambda parameter, which is no column",
+            sql: "SELECT o.customer_id, len(list_filter(ARRAY_AGG(o.status), lambda e: e <> \
+                  'null')) AS c FROM orders AS o GROUP BY o.customer_id",
+            expected_findings: &[],
+        },
+        QueryFindingsTestCase {
+            description: "a lambda parameter cast over a JSON list of an aggregate",
+            sql: "SELECT o.customer_id, len(list_filter(CAST(to_json(ARRAY_AGG(o.status)) AS \
+                  JSON[]), lambda e: CAST(e AS VARCHAR) <> 'null')) AS c FROM orders AS o GROUP BY \
+                  o.customer_id",
+            expected_findings: &[],
+        },
+        QueryFindingsTestCase {
+            description: "a grouped outer column inside the lambda body",
+            sql: "SELECT o.customer_id, len(list_filter(ARRAY_AGG(o.status), lambda e: e <> \
+                  CAST(o.customer_id AS VARCHAR))) AS c FROM orders AS o GROUP BY o.customer_id",
+            expected_findings: &[],
+        },
+        QueryFindingsTestCase {
+            description: "an ungrouped outer column inside the lambda body",
+            sql: "SELECT o.customer_id, len(list_filter(ARRAY_AGG(o.status), lambda e: e <> \
+                  o.status)) AS c FROM orders AS o GROUP BY o.customer_id",
+            expected_findings: &[(
+                "E230",
+                "Column must be grouped or used in an aggregate function",
+                76,
+            )],
+        },
+        QueryFindingsTestCase {
+            description: "a parameter's name read outside its lambda is a column",
+            sql: "SELECT o.customer_id, concat(list_filter(ARRAY_AGG(o.status), lambda status: \
+                  status <> 'x')[1], status) AS c FROM orders AS o GROUP BY o.customer_id",
+            expected_findings: &[(
+                "E230",
+                "Column must be grouped or used in an aggregate function",
+                96,
+            )],
+        },
+        QueryFindingsTestCase {
+            description: "a lambda field the FROM item of that name also has is that item's column",
+            sql: "SELECT list_transform([{'amount': 99}], lambda o: o.amount) AS v FROM orders AS o GROUP BY o.status",
+            expected_findings: &[(
+                "E230",
+                "Column must be grouped or used in an aggregate function",
+                52,
+            )],
+        },
+        QueryFindingsTestCase {
+            description: "a lambda field only the parameter has",
+            sql: "SELECT list_transform([{'m': 99}], lambda o: o.m) AS v FROM orders AS o GROUP BY o.status",
+            expected_findings: &[],
+        },
+        QueryFindingsTestCase {
+            description: "a JSON arrow is no lambda, so its column must be grouped",
+            sql: "SELECT o.items -> o.status AS v FROM orders AS o GROUP BY o.status",
+            expected_findings: &[(
+                "E230",
+                "Column must be grouped or used in an aggregate function",
+                9,
+            )],
+        },
+    ];
+    for test_case in test_cases {
+        assert_eq!(
+            duckdb_parser_findings(test_case.sql),
+            owned_findings(test_case.expected_findings),
+            "{}",
+            test_case.description
+        );
+    }
+}
+
 #[test]
 fn given_duckdbs_parser_reads_window_calls_when_checking_arity_then_every_count_duckdb_binds_is_known()
  {

crates/sqlquack-semantics/src/lineage/_helpers/checks/scope_rules.rs

@@ -1008,9 +1008,10 @@ fn grouping(
     }
     let mut findings: Vec<NameFinding> = Vec::new();
     for (expression, named) in read {
-        let mut pending: Vec<&Expression> = vec![unalias(expression)];
+        let mut lambdas: Vec<(usize, Vec<String>)> = vec![(0, Vec::new())];
+        let mut pending: Vec<(&Expression, usize)> = vec![(unalias(expression), 0)];
         let mut expanded: HashSet<String> = HashSet::new();
-        while let Some(current) = pending.pop() {
+        while let Some((current, frame)) = pending.pop() {
             if query_start(current)
                 || is_aggregate(current)
                 || matches!(current, Expression::WithinGroup(group) if is_aggregate(&group.this))
@@ -1021,8 +1022,21 @@ fn grouping(
             {
                 continue;
             }
+            if let Expression::Lambda(lambda) = current
+                && (lambda.colon || Engine::of(level.context) != Engine::DuckDB)
+            {
+                let names: Vec<String> = lambda
+                    .parameters
+                    .iter()
+                    .map(|name| level.key(name))
+                    .collect();
+                lambdas.push((frame, names));
+                pending.push((&lambda.body, lambdas.len() - 1));
+                continue;
+            }
             if let Expression::Column(column) = current {
-                if grouped(&column_key(column, level))
+                if lambda_parameter(column, (&lambdas, frame), level)
+                    || grouped(&column_key(column, level))
                     || keyword_function(column, level.scope, level.context)
                 {
                     continue;
@@ -1031,7 +1045,7 @@ fn grouping(
                     let name: String = level.key(&column.name);
                     if let Some(alias) = aliases.get(&name) {
                         if expanded.insert(name) {
-                            pending.push(alias);
+                            pending.push((alias, frame));
                         }
                         continue;
                     }
@@ -1040,23 +1054,62 @@ fn grouping(
                 break;
             }
             if let Expression::WindowFunction(window) = current {
-                pending.extend(window.this.children());
+                pending.extend(
+                    window
+                        .this
+                        .children()
+                        .into_iter()
+                        .map(|child| (child, frame)),
+                );
                 pending.extend(
                     current
                         .children()
                         .into_iter()
-                        .filter(|child| !std::ptr::eq(*child, &window.this)),
+                        .filter(|child| !std::ptr::eq(*child, &window.this))
+                        .map(|child| (child, frame)),
                 );
             } else {
                 let mut children: Vec<&Expression> = current.children();
                 children.reverse();
-                pending.extend(children);
+                pending.extend(children.into_iter().map(|child| (child, frame)));
             }
         }
     }
     findings
 }
 
+/// Whether `column` or its qualifier names an enclosing lambda's parameter no FROM item shadows.
+fn lambda_parameter(
+    column: &Column,
+    (lambdas, mut frame): (&[(usize, Vec<String>)], usize),
+    level: RuleLevel<'_, '_>,
+) -> bool {
+    let name: String = level.key(column.table.as_ref().unwrap_or(&column.name));
+    let field: String = level.key(&column.name);
+    if column.table.is_some()
+        && level.scope.sources.iter().any(|source| {
+            source
+                .key
+                .as_ref()
+                .is_some_and(|key| level.context.profile.same_identifier(key, &name))
+                && offers(source, &field, level)
+        })
+    {
+        return false;
+    }
+    while frame != 0 {
+        let (parent, names) = &lambdas[frame];
+        if names
+            .iter()
+            .any(|parameter| level.context.profile.same_identifier(parameter, &name))
+        {
+            return true;
+        }
+        frame = *parent;
+    }
+    false
+}
+
 /// Snowflake's DISTINCT with ORDER BY a column it does not select.
 fn distinct_order(
     (node, select): (&Expression, &Select),