feat(optimizer): support forced nested-loop joins

Author: KKould

docs/features.md

@@ -57,16 +57,15 @@ ORDER BY user_id;
 ```
 
 For ORM queries, enable both `orm` and `spill`, then call `force_spill()` before
-`group_by()` or `distinct()` so the option is passed when the aggregate plan is
-built:
+building the projection and aggregate plan:
 
 ```rust,ignore
 let grouped = database.bind(|ctx| {
     ctx.from::<Order>()?
+        .force_spill()?
         .project_tuple(|e| {
             Ok(vec![e.column(Order::user_id())?, e.count_all()?])
         })?
-        .force_spill()?
         .group_by(|e| e.column(Order::user_id()))?
         .order_by(Order::user_id())?
         .finish()
@@ -78,6 +77,33 @@ an external sort on the group keys and executes a streaming aggregate. The
 hint is intended for grouped aggregates or `DISTINCT`; an aggregate without
 group keys already uses constant-size accumulator state.
 
+### Nested-loop Join Hint
+
+Use `FORCE_NEST_LOOP_JOIN` to select the nested-loop implementation for joins
+in the current `SELECT` query block:
+
+```sql
+SELECT /*+ FORCE_NEST_LOOP_JOIN */ orders.id, users.name
+FROM orders
+JOIN users ON orders.user_id = users.id;
+```
+
+ORM queries can select the same implementation before adding joins:
+
+```rust,ignore
+let joined = database.bind(|ctx| {
+    ctx.from::<Order>()?
+        .force_nested_loop()
+        .inner_join::<User, _>(|e| {
+            e.column(Order::user_id())?.eq(e.column(User::id())?)
+        })?
+        .finish()
+})?;
+```
+
+It avoids Hash Join's build-side tuple materialization, but may perform
+substantially more work on large inputs.
+
 ### User-Defined Function: `features = ["macros"]`
 ```rust
 scala_function!(TestFunction::test(LogicalType::Integer, LogicalType::Integer) -> LogicalType::Integer => |v1: DataValue, v2: DataValue| {

src/binder/aggregate.rs

@@ -48,15 +48,14 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
         children: LogicalPlan,
         agg_calls: Vec<ScalarExpression>,
         groupby_exprs: Vec<ScalarExpression>,
-        force_spill: bool,
     ) -> Result<LogicalPlan, DatabaseError> {
         self.context.step(QueryBindStep::Agg);
         Ok(AggregateOperator::build(
             children,
             agg_calls,
             groupby_exprs,
             false,
-            force_spill,
+            self.force_spill,
         ))
     }
 

src/binder/distinct.rs

@@ -27,7 +27,6 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
         &mut self,
         children: LogicalPlan,
         select_list: Vec<ScalarExpression>,
-        force_spill: bool,
     ) -> Result<LogicalPlan, DatabaseError> {
         self.context.step(QueryBindStep::Distinct);
 
@@ -36,7 +35,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
             vec![],
             select_list,
             true,
-            force_spill,
+            self.force_spill,
         ))
     }
 

src/binder/mod.rs

@@ -561,6 +561,8 @@ impl<'a, T: Transaction> BinderContext<'a, T> {
 pub struct Binder<'a, 'parent, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> {
     pub(crate) context: BinderContext<'a, T>,
     pub(crate) args: &'a A,
+    pub(crate) force_spill: bool,
+    pub(crate) force_nested_loop: bool,
     with_pk: Option<TableName>,
     pub(crate) parent: Option<&'parent BinderContext<'a, T>>,
 }
@@ -574,6 +576,8 @@ impl<'a, 'parent, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<
         Binder {
             context,
             args,
+            force_spill: false,
+            force_nested_loop: false,
             with_pk: None,
             parent,
         }

src/binder/parser.rs

@@ -1364,6 +1364,7 @@ where
                     self.binder.bind_table_ref_sql(from, self.arena)?,
                     JoinCondition::None,
                     JoinType::Cross,
+                    self.binder.force_nested_loop,
                 )
             }
             plan
@@ -1424,7 +1425,6 @@ where
         group_by: &GroupByExpr,
         having: Option<&Expr>,
         orderby: Option<&[OrderByExpr]>,
-        force_spill: bool,
     ) -> Result<BindPlanAggregated<'s, 'a, 'b, 'arena, T, A>, DatabaseError> {
         let group_by = with_query_bind_step!(self.binder, QueryBindStep::Agg, {
             match group_by {
@@ -1453,22 +1453,16 @@ where
                 })
             })
             .transpose()?;
-        self.aggregate(
-            group_by,
-            having,
-            orderby,
-            force_spill,
-            |binder, arena, orderby| {
-                let OrderByExpr { expr, options, .. } = orderby;
-                with_query_bind_step!(binder, QueryBindStep::Sort, {
-                    SortField::new(
-                        binder.bind_expr(expr, arena)?,
-                        options.asc.is_none_or(|asc| asc),
-                        options.nulls_first.unwrap_or(false),
-                    )
-                })
-            },
-        )
+        self.aggregate(group_by, having, orderby, |binder, arena, orderby| {
+            let OrderByExpr { expr, options, .. } = orderby;
+            with_query_bind_step!(binder, QueryBindStep::Sort, {
+                SortField::new(
+                    binder.bind_expr(expr, arena)?,
+                    options.asc.is_none_or(|asc| asc),
+                    options.nulls_first.unwrap_or(false),
+                )
+            })
+        })
     }
 }
 
@@ -1480,9 +1474,8 @@ where
     pub(crate) fn distinct_sql(
         self,
         distinct: Option<&Distinct>,
-        force_spill: bool,
     ) -> Result<BindPlanDistinct<'s, 'a, 'b, 'arena, T, A>, DatabaseError> {
-        self.distinct(matches!(distinct, Some(Distinct::Distinct)), force_spill)
+        self.distinct(matches!(distinct, Some(Distinct::Distinct)))
     }
 }
 
@@ -2624,27 +2617,40 @@ impl<'a, 'parent, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<
                 "QUALIFY is not supported".to_string(),
             ));
         }
-        let force_spill = optimizer_hint
-            .as_ref()
-            .is_some_and(|hint| hint.text.trim().eq_ignore_ascii_case("FORCE_AGG_SPILL"));
+        let has_hint = |expected: &str| {
+            optimizer_hint.as_ref().is_some_and(|hint| {
+                hint.text
+                    .split(|char: char| char.is_ascii_whitespace() || char == ',')
+                    .any(|hint| hint.eq_ignore_ascii_case(expected))
+            })
+        };
+        let force_spill = has_hint("FORCE_AGG_SPILL");
+        let force_nested_loop = has_hint("FORCE_NEST_LOOP_JOIN");
         if force_spill && !cfg!(feature = "spill") {
             return Err(DatabaseError::UnsupportedStmt(
                 "FORCE_AGG_SPILL requires the `spill` feature".to_string(),
             ));
         }
-        Ok(self
-            .build_plan(arena)
-            .from_sql(from)?
-            .select_list_from_sql(projection)?
-            .where_sql(selection.as_ref())?
-            .aggregate_sql(group_by, having.as_ref(), orderby, force_spill)?
-            .having()?
-            .window()?
-            .distinct_sql(distinct.as_ref(), force_spill)?
-            .order_by()?
-            .project()?
-            .select_into_sql(into.as_ref())?
-            .finish())
+        let previous_options = (self.force_spill, self.force_nested_loop);
+        self.force_spill = force_spill;
+        self.force_nested_loop = force_nested_loop;
+        let result = (|| {
+            Ok(self
+                .build_plan(arena)
+                .from_sql(from)?
+                .select_list_from_sql(projection)?
+                .where_sql(selection.as_ref())?
+                .aggregate_sql(group_by, having.as_ref(), orderby)?
+                .having()?
+                .window()?
+                .distinct_sql(distinct.as_ref())?
+                .order_by()?
+                .project()?
+                .select_into_sql(into.as_ref())?
+                .finish())
+        })();
+        (self.force_spill, self.force_nested_loop) = previous_options;
+        result
     }
 
     /// FIXME: temp values need to register BindContext.bind_table
@@ -3154,6 +3160,54 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn force_nest_loop_join_marks_join_operator() -> Result<(), DatabaseError> {
+        let tables = build_t1_table()?;
+        let plan =
+            tables.plan("select /*+ FORCE_NEST_LOOP_JOIN */ c1, c3 from t1 join t2 on c1 = c3")?;
+        let join = plan
+            .childrens
+            .iter()
+            .find_map(|plan| match &plan.operator {
+                Operator::Join(operator) => Some(operator),
+                _ => None,
+            })
+            .expect("query should contain a join");
+
+        assert!(join.force_nested_loop);
+        Ok(())
+    }
+
+    #[cfg(feature = "spill")]
+    #[test]
+    fn optimizer_hints_can_be_combined() -> Result<(), DatabaseError> {
+        let tables = build_t1_table()?;
+        let plan = tables.plan(
+            "select /*+ FORCE_AGG_SPILL, FORCE_NEST_LOOP_JOIN */ c1, count(c3) \
+             from t1 join t2 on c1 = c3 group by c1",
+        )?;
+        let aggregate_plan = plan
+            .childrens
+            .iter()
+            .find(|plan| matches!(plan.operator, Operator::Aggregate(_)))
+            .expect("query should contain an aggregate");
+        let Operator::Aggregate(aggregate) = &aggregate_plan.operator else {
+            unreachable!()
+        };
+        let join = aggregate_plan
+            .childrens
+            .iter()
+            .find_map(|plan| match &plan.operator {
+                Operator::Join(operator) => Some(operator),
+                _ => None,
+            })
+            .expect("aggregate input should contain a join");
+
+        assert!(aggregate.force_spill);
+        assert!(join.force_nested_loop);
+        Ok(())
+    }
+
     #[cfg(feature = "copy")]
     #[test]
     fn test_copy_file_format_options() -> Result<(), DatabaseError> {