feat(sql): implement subquery-to-join rewriting for IN/NOT IN

Replace the subquery.rs stub with a complete implementation that
rewrites WHERE-clause subqueries into semi/anti joins, allowing the
existing hash-join executor to handle them without a separate subquery
engine.

Supported patterns:
- `WHERE col IN (SELECT col2 FROM tbl)` → semi-join
- `WHERE col NOT IN (SELECT col2 FROM tbl)` → anti-join
- `WHERE col > (SELECT AGG(...) FROM tbl)` → materialized scalar subquery

Both select and join planners now call extract_subqueries() to strip
subquery predicates from the WHERE clause before filter conversion, then
wrap the base plan with the extracted join nodes. Semi and Anti join
operators are also wired into the join operator mapping.

Author: farhan-syah

nodedb-sql/src/planner/join.rs

@@ -13,7 +13,7 @@ use crate::types::*;
 pub fn plan_join_from_select(
     select: &Select,
     scope: &TableScope,
-    _catalog: &dyn SqlCatalog,
+    catalog: &dyn SqlCatalog,
     functions: &FunctionRegistry,
 ) -> Result<Option<SqlPlan>> {
     let from = &select.from[0];
@@ -83,14 +83,33 @@ pub fn plan_join_from_select(
         };
     }
 
-    // Apply WHERE as filters on the join plan.
-    // Apply projection.
+    // Extract subqueries from WHERE and wrap as additional joins.
+    let (subquery_joins, effective_where) = if let Some(expr) = &select.selection {
+        let extraction = super::subquery::extract_subqueries(expr, catalog, functions)?;
+        (extraction.joins, extraction.remaining_where)
+    } else {
+        (Vec::new(), None)
+    };
+
+    // Apply remaining WHERE as filters on the join plan.
     let _projection = super::select::convert_projection(&select.projection)?;
-    let _filters = match &select.selection {
+    let _filters = match &effective_where {
         Some(expr) => super::select::convert_where_to_filters(expr)?,
         None => Vec::new(),
     };
 
+    // Wrap with subquery joins (semi/anti/cross) if any.
+    for sq in subquery_joins {
+        current_plan = SqlPlan::Join {
+            left: Box::new(current_plan),
+            right: Box::new(sq.inner_plan),
+            on: vec![(sq.outer_column, sq.inner_column)],
+            join_type: sq.join_type,
+            condition: None,
+            limit: 10000,
+        };
+    }
+
     // Check if there's aggregation on the join.
     let group_by_non_empty = match &select.group_by {
         ast::GroupByExpr::All(_) => true,
@@ -124,15 +143,15 @@ type JoinSpec = (JoinType, Vec<(String, String)>, Option<SqlExpr>);
 /// Extract join type, equi-join keys, and non-equi condition.
 fn extract_join_spec(op: &ast::JoinOperator) -> Result<JoinSpec> {
     match op {
-        ast::JoinOperator::Inner(constraint) => {
+        ast::JoinOperator::Inner(constraint) | ast::JoinOperator::Join(constraint) => {
             let (keys, cond) = extract_join_constraint(constraint)?;
             Ok((JoinType::Inner, keys, cond))
         }
-        ast::JoinOperator::LeftOuter(constraint) => {
+        ast::JoinOperator::Left(constraint) | ast::JoinOperator::LeftOuter(constraint) => {
             let (keys, cond) = extract_join_constraint(constraint)?;
             Ok((JoinType::Left, keys, cond))
         }
-        ast::JoinOperator::RightOuter(constraint) => {
+        ast::JoinOperator::Right(constraint) | ast::JoinOperator::RightOuter(constraint) => {
             let (keys, cond) = extract_join_constraint(constraint)?;
             Ok((JoinType::Right, keys, cond))
         }
@@ -144,6 +163,14 @@ fn extract_join_spec(op: &ast::JoinOperator) -> Result<JoinSpec> {
             let (keys, cond) = extract_join_constraint(constraint)?;
             Ok((JoinType::Cross, keys, cond))
         }
+        ast::JoinOperator::Semi(constraint) | ast::JoinOperator::LeftSemi(constraint) => {
+            let (keys, cond) = extract_join_constraint(constraint)?;
+            Ok((JoinType::Semi, keys, cond))
+        }
+        ast::JoinOperator::Anti(constraint) | ast::JoinOperator::LeftAnti(constraint) => {
+            let (keys, cond) = extract_join_constraint(constraint)?;
+            Ok((JoinType::Anti, keys, cond))
+        }
         _ => Err(SqlError::Unsupported {
             detail: format!("join type: {op:?}"),
         }),

nodedb-sql/src/planner/select.rs

@@ -93,8 +93,16 @@ fn plan_select(
         detail: "multi-table FROM without JOIN".into(),
     })?;
 
-    // 4. Convert WHERE filters.
-    let filters = match &select.selection {
+    // 4. Extract subqueries from WHERE and rewrite as semi/anti joins.
+    let (subquery_joins, effective_where) = if let Some(expr) = &select.selection {
+        let extraction = super::subquery::extract_subqueries(expr, catalog, functions)?;
+        (extraction.joins, extraction.remaining_where)
+    } else {
+        (Vec::new(), None)
+    };
+
+    // 5. Convert remaining WHERE filters.
+    let filters = match &effective_where {
         Some(expr) => {
             // Check for search-triggering functions in WHERE.
             if let Some(plan) = try_extract_where_search(expr, table, functions)? {
@@ -105,19 +113,32 @@ fn plan_select(
         None => Vec::new(),
     };
 
-    // 5. Check for GROUP BY / aggregation.
+    // 6. Check for GROUP BY / aggregation.
     if has_aggregation(select, functions) {
-        return super::aggregate::plan_aggregate(select, table, &filters, &scope, functions);
+        let mut plan =
+            super::aggregate::plan_aggregate(select, table, &filters, &scope, functions)?;
+        // Wrap with subquery joins if any.
+        for sq in subquery_joins {
+            plan = SqlPlan::Join {
+                left: Box::new(plan),
+                right: Box::new(sq.inner_plan),
+                on: vec![(sq.outer_column, sq.inner_column)],
+                join_type: sq.join_type,
+                condition: None,
+                limit: 10000,
+            };
+        }
+        return Ok(plan);
     }
 
-    // 6. Convert projection.
+    // 7. Convert projection.
     let projection = convert_projection(&select.projection)?;
 
-    // 7. Convert window functions (SELECT with OVER).
+    // 8. Convert window functions (SELECT with OVER).
     let window_functions = super::window::extract_window_functions(&select.projection, functions)?;
 
-    // 8. Build base scan plan.
-    Ok(SqlPlan::Scan {
+    // 9. Build base scan plan.
+    let mut plan = SqlPlan::Scan {
         collection: table.name.clone(),
         engine: table.info.engine,
         filters,
@@ -127,7 +148,21 @@ fn plan_select(
         offset: 0,
         distinct: select.distinct.is_some(),
         window_functions,
-    })
+    };
+
+    // 10. Wrap with subquery joins (semi/anti) if any.
+    for sq in subquery_joins {
+        plan = SqlPlan::Join {
+            left: Box::new(plan),
+            right: Box::new(sq.inner_plan),
+            on: vec![(sq.outer_column, sq.inner_column)],
+            join_type: sq.join_type,
+            condition: None,
+            limit: 10000,
+        };
+    }
+
+    Ok(plan)
 }
 
 /// Check if a SELECT has aggregation (GROUP BY or aggregate functions in projection).