optimizer.rs: refactor to use new data structures and remove unnecessary stuff

We don't need `push_predicates()` because that never REALLY was a predicate
pushdown optimization -- it just pushed WHERE clause condition expressions
into the correct SourceOperator nodes in the tree.

Now that we don't have a SourceOperator tree anymore and we keep the conditions
in the WHERE clause instead, we don't need to "push" anything anymore. Leaves
room for ACTUAL predicate pushdown optimizations later :)

We also don't need any weird bitmask stuff anymore, and perhaps we never did,
to determine where conditions should be evaluated.

core/translate/optimizer.rs

@@ -5,9 +5,8 @@ use sqlite3_parser::ast;
 use crate::{schema::Index, Result};
 
 use super::plan::{
-    get_table_ref_bitmask_for_ast_expr, get_table_ref_bitmask_for_operator, DeletePlan, Direction,
-    IterationDirection, Plan, Search, SelectPlan, SourceOperator, TableReference,
-    TableReferenceType,
+    DeletePlan, Direction, IterationDirection, JoinAwareConditionExpr, Operation, Plan, Search,
+    SelectPlan, TableReference,
 };
 
 pub fn optimize_plan(plan: &mut Plan) -> Result<()> {
@@ -23,33 +22,22 @@ pub fn optimize_plan(plan: &mut Plan) -> Result<()> {
  * but having them separate makes them easier to understand
  */
 fn optimize_select_plan(plan: &mut SelectPlan) -> Result<()> {
-    optimize_subqueries(&mut plan.source)?;
+    optimize_subqueries(plan)?;
     rewrite_exprs_select(plan)?;
     if let ConstantConditionEliminationResult::ImpossibleCondition =
-        eliminate_constants(&mut plan.source, &mut plan.where_clause)?
+        eliminate_constant_conditions(&mut plan.where_clause)?
     {
         plan.contains_constant_false_condition = true;
         return Ok(());
     }
 
-    push_predicates(
-        &mut plan.source,
-        &mut plan.where_clause,
-        &plan.referenced_tables,
-    )?;
-
     use_indexes(
-        &mut plan.source,
-        &plan.referenced_tables,
+        &mut plan.table_references,
         &plan.available_indexes,
+        &mut plan.where_clause,
     )?;
 
-    eliminate_unnecessary_orderby(
-        &mut plan.source,
-        &mut plan.order_by,
-        &plan.referenced_tables,
-        &plan.available_indexes,
-    )?;
+    eliminate_unnecessary_orderby(plan)?;
 
     Ok(())
 }
@@ -57,83 +45,67 @@ fn optimize_select_plan(plan: &mut SelectPlan) -> Result<()> {
 fn optimize_delete_plan(plan: &mut DeletePlan) -> Result<()> {
     rewrite_exprs_delete(plan)?;
     if let ConstantConditionEliminationResult::ImpossibleCondition =
-        eliminate_constants(&mut plan.source, &mut plan.where_clause)?
+        eliminate_constant_conditions(&mut plan.where_clause)?
     {
         plan.contains_constant_false_condition = true;
         return Ok(());
     }
 
     use_indexes(
-        &mut plan.source,
-        &plan.referenced_tables,
+        &mut plan.table_references,
         &plan.available_indexes,
+        &mut plan.where_clause,
     )?;
 
     Ok(())
 }
 
-fn optimize_subqueries(operator: &mut SourceOperator) -> Result<()> {
-    match operator {
-        SourceOperator::Subquery { plan, .. } => {
+fn optimize_subqueries(plan: &mut SelectPlan) -> Result<()> {
+    for table in plan.table_references.iter_mut() {
+        if let Operation::Subquery { plan, .. } = &mut table.op {
             optimize_select_plan(&mut *plan)?;
-            Ok(())
         }
-        SourceOperator::Join { left, right, .. } => {
-            optimize_subqueries(left)?;
-            optimize_subqueries(right)?;
-            Ok(())
-        }
-        _ => Ok(()),
     }
+
+    Ok(())
 }
 
-fn _operator_is_already_ordered_by(
-    operator: &mut SourceOperator,
+fn query_is_already_ordered_by(
+    table_references: &[TableReference],
     key: &mut ast::Expr,
-    referenced_tables: &[TableReference],
     available_indexes: &Vec<Rc<Index>>,
 ) -> Result<bool> {
-    match operator {
-        SourceOperator::Scan {
-            table_reference, ..
-        } => Ok(key.is_rowid_alias_of(table_reference.table_index)),
-        SourceOperator::Search {
-            table_reference,
-            search,
-            ..
-        } => match search {
-            Search::RowidEq { .. } => Ok(key.is_rowid_alias_of(table_reference.table_index)),
-            Search::RowidSearch { .. } => Ok(key.is_rowid_alias_of(table_reference.table_index)),
+    let first_table = table_references.first();
+    if first_table.is_none() {
+        return Ok(false);
+    }
+    let table_reference = first_table.unwrap();
+    match &table_reference.op {
+        Operation::Scan { .. } => Ok(key.is_rowid_alias_of(0)),
+        Operation::Search(search) => match search {
+            Search::RowidEq { .. } => Ok(key.is_rowid_alias_of(0)),
+            Search::RowidSearch { .. } => Ok(key.is_rowid_alias_of(0)),
             Search::IndexSearch { index, .. } => {
-                let index_idx = key.check_index_scan(
-                    table_reference.table_index,
-                    referenced_tables,
-                    available_indexes,
-                )?;
+                let index_idx = key.check_index_scan(0, &table_reference, available_indexes)?;
                 let index_is_the_same = index_idx
                     .map(|i| Rc::ptr_eq(&available_indexes[i], index))
                     .unwrap_or(false);
                 Ok(index_is_the_same)
             }
         },
-        SourceOperator::Join { left, .. } => {
-            _operator_is_already_ordered_by(left, key, referenced_tables, available_indexes)
-        }
         _ => Ok(false),
     }
 }
 
-fn eliminate_unnecessary_orderby(
-    operator: &mut SourceOperator,
-    order_by: &mut Option<Vec<(ast::Expr, Direction)>>,
-    referenced_tables: &[TableReference],
-    available_indexes: &Vec<Rc<Index>>,
-) -> Result<()> {
-    if order_by.is_none() {
+fn eliminate_unnecessary_orderby(plan: &mut SelectPlan) -> Result<()> {
+    if plan.order_by.is_none() {
+        return Ok(());
+    }
+    if plan.table_references.len() == 0 {
         return Ok(());
     }
 
-    let o = order_by.as_mut().unwrap();
+    let o = plan.order_by.as_mut().unwrap();
 
     if o.len() != 1 {
         // TODO: handle multiple order by keys
@@ -143,76 +115,55 @@ fn eliminate_unnecessary_orderby(
     let (key, direction) = o.first_mut().unwrap();
 
     let already_ordered =
-        _operator_is_already_ordered_by(operator, key, referenced_tables, available_indexes)?;
+        query_is_already_ordered_by(&plan.table_references, key, &plan.available_indexes)?;
 
     if already_ordered {
-        push_scan_direction(operator, direction);
-        *order_by = None;
+        push_scan_direction(&mut plan.table_references[0], direction);
+        plan.order_by = None;
     }
 
     Ok(())
 }
 
 /**
- * Use indexes where possible
+ * Use indexes where possible.
+ * Right now we make decisions about using indexes ONLY based on condition expressions, not e.g. ORDER BY or others.
+ * This is just because we are WIP.
+ *
+ * When this function is called, condition expressions from both the actual WHERE clause and the JOIN clauses are in the where_clause vector.
+ * If we find a condition that can be used to index scan, we pop it off from the where_clause vector and put it into a Search operation.
+ * We put it there simply because it makes it a bit easier to track during translation.
  */
 fn use_indexes(
-    operator: &mut SourceOperator,
-    referenced_tables: &[TableReference],
-    available_indexes: &[Rc<Index>],
+    table_references: &mut [TableReference],
+    available_indexes: &Vec<Rc<Index>>,
+    where_clause: &mut Vec<JoinAwareConditionExpr>,
 ) -> Result<()> {
-    match operator {
-        SourceOperator::Subquery { .. } => Ok(()),
-        SourceOperator::Search { .. } => Ok(()),
-        SourceOperator::Scan {
-            table_reference,
-            predicates: filter,
-            id,
-            ..
-        } => {
-            if filter.is_none() {
-                return Ok(());
-            }
+    if where_clause.is_empty() {
+        return Ok(());
+    }
 
-            let fs = filter.as_mut().unwrap();
-            for i in 0..fs.len() {
-                let f = fs[i].take_ownership();
-                let table_index = referenced_tables
-                    .iter()
-                    .position(|t| t.table_identifier == table_reference.table_identifier)
-                    .unwrap();
-                match try_extract_index_search_expression(
-                    f,
+    'outer: for (table_index, table_reference) in table_references.iter_mut().enumerate() {
+        if let Operation::Scan { .. } = &mut table_reference.op {
+            let mut i = 0;
+            while i < where_clause.len() {
+                let cond = where_clause.get_mut(i).unwrap();
+                if let Some(index_search) = try_extract_index_search_expression(
+                    cond,
                     table_index,
-                    referenced_tables,
+                    &table_reference,
                     available_indexes,
                 )? {
-                    Either::Left(non_index_using_expr) => {
-                        fs[i] = non_index_using_expr;
-                    }
-                    Either::Right(index_search) => {
-                        fs.remove(i);
-                        *operator = SourceOperator::Search {
-                            id: *id,
-                            table_reference: table_reference.clone(),
-                            predicates: Some(fs.clone()),
-                            search: index_search,
-                        };
-
-                        return Ok(());
-                    }
+                    where_clause.remove(i);
+                    table_reference.op = Operation::Search(index_search);
+                    continue 'outer;
                 }
+                i += 1;
             }
-
-            Ok(())
         }
-        SourceOperator::Join { left, right, .. } => {
-            use_indexes(left, referenced_tables, available_indexes)?;
-            use_indexes(right, referenced_tables, available_indexes)?;
-            Ok(())
-        }
-        SourceOperator::Nothing { .. } => Ok(()),
     }
+
+    Ok(())
 }
 
 #[derive(Debug, PartialEq, Clone)]
@@ -221,377 +172,38 @@ enum ConstantConditionEliminationResult {
     ImpossibleCondition,
 }
 
-// removes predicates that are always true
-// returns a ConstantEliminationResult indicating whether any predicates are always false
-fn eliminate_constants(
-    operator: &mut SourceOperator,
-    where_clause: &mut Option<Vec<ast::Expr>>,
+/// Removes predicates that are always true.
+/// Returns a ConstantEliminationResult indicating whether any predicates are always false.
+/// This is used to determine whether the query can be aborted early.
+fn eliminate_constant_conditions(
+    where_clause: &mut Vec<JoinAwareConditionExpr>,
 ) -> Result<ConstantConditionEliminationResult> {
-    if let Some(predicates) = where_clause {
-        let mut i = 0;
-        while i < predicates.len() {
-            let predicate = &predicates[i];
-            if predicate.is_always_true()? {
-                // true predicates can be removed since they don't affect the result
-                predicates.remove(i);
-            } else if predicate.is_always_false()? {
-                // any false predicate in a list of conjuncts (AND-ed predicates) will make the whole list false
-                predicates.truncate(0);
-                return Ok(ConstantConditionEliminationResult::ImpossibleCondition);
-            } else {
+    let mut i = 0;
+    while i < where_clause.len() {
+        let predicate = &where_clause[i];
+        if predicate.expr.is_always_true()? {
+            // true predicates can be removed since they don't affect the result
+            where_clause.remove(i);
+        } else if predicate.expr.is_always_false()? {
+            // any false predicate in a list of conjuncts (AND-ed predicates) will make the whole list false,
+            // except an outer join condition, because that just results in NULLs, not skipping the whole loop
+            if predicate.from_outer_join {
                 i += 1;
-            }
-        }
-    }
-    match operator {
-        SourceOperator::Subquery { .. } => Ok(ConstantConditionEliminationResult::Continue),
-        SourceOperator::Join {
-            left,
-            right,
-            predicates,
-            outer,
-            ..
-        } => {
-            if eliminate_constants(left, where_clause)?
-                == ConstantConditionEliminationResult::ImpossibleCondition
-            {
-                return Ok(ConstantConditionEliminationResult::ImpossibleCondition);
-            }
-            if eliminate_constants(right, where_clause)?
-                == ConstantConditionEliminationResult::ImpossibleCondition
-                && !*outer
-            {
-                return Ok(ConstantConditionEliminationResult::ImpossibleCondition);
-            }
-
-            if predicates.is_none() {
-                return Ok(ConstantConditionEliminationResult::Continue);
-            }
-
-            let predicates = predicates.as_mut().unwrap();
-
-            let mut i = 0;
-            while i < predicates.len() {
-                let predicate = &mut predicates[i];
-                if predicate.is_always_true()? {
-                    predicates.remove(i);
-                } else if predicate.is_always_false()? {
-                    if !*outer {
-                        predicates.truncate(0);
-                        return Ok(ConstantConditionEliminationResult::ImpossibleCondition);
-                    }
-                    // in an outer join, we can't skip rows, so just replace all constant false predicates with 0
-                    // so we don't later have to evaluate anything more complex or special-case the identifiers true and false
-                    // which are just aliases for 1 and 0
-                    *predicate = ast::Expr::Literal(ast::Literal::Numeric("0".to_string()));
-                    i += 1;
-                } else {
-                    i += 1;
-                }
-            }
-
-            Ok(ConstantConditionEliminationResult::Continue)
-        }
-        SourceOperator::Scan { predicates, .. } => {
-            if let Some(ps) = predicates {
-                let mut i = 0;
-                while i < ps.len() {
-                    let predicate = &ps[i];
-                    if predicate.is_always_true()? {
-                        // true predicates can be removed since they don't affect the result
-                        ps.remove(i);
-                    } else if predicate.is_always_false()? {
-                        // any false predicate in a list of conjuncts (AND-ed predicates) will make the whole list false
-                        ps.truncate(0);
-                        return Ok(ConstantConditionEliminationResult::ImpossibleCondition);
-                    } else {
-                        i += 1;
-                    }
-                }
-
-                if ps.is_empty() {
-                    *predicates = None;
-                }
-            }
-            Ok(ConstantConditionEliminationResult::Continue)
-        }
-        SourceOperator::Search { predicates, .. } => {
-            if let Some(predicates) = predicates {
-                let mut i = 0;
-                while i < predicates.len() {
-                    let predicate = &predicates[i];
-                    if predicate.is_always_true()? {
-                        // true predicates can be removed since they don't affect the result
-                        predicates.remove(i);
-                    } else if predicate.is_always_false()? {
-                        // any false predicate in a list of conjuncts (AND-ed predicates) will make the whole list false
-                        predicates.truncate(0);
-                        return Ok(ConstantConditionEliminationResult::ImpossibleCondition);
-                    } else {
-                        i += 1;
-                    }
-                }
-            }
-
-            Ok(ConstantConditionEliminationResult::Continue)
-        }
-        SourceOperator::Nothing { .. } => Ok(ConstantConditionEliminationResult::Continue),
-    }
-}
-
-/**
-  Recursively pushes predicates down the tree, as far as possible.
-  Where a predicate is pushed determines at which loop level it will be evaluated.
-  For example, in SELECT * FROM t1 JOIN t2 JOIN t3 WHERE t1.a = t2.a AND t2.b = t3.b AND t1.c = 1
-  the predicate t1.c = 1 can be pushed to t1 and will be evaluated in the first (outermost) loop,
-  the predicate t1.a = t2.a can be pushed to t2 and will be evaluated in the second loop
-  while t2.b = t3.b will be evaluated in the third loop.
-*/
-fn push_predicates(
-    operator: &mut SourceOperator,
-    where_clause: &mut Option<Vec<ast::Expr>>,
-    referenced_tables: &Vec<TableReference>,
-) -> Result<()> {
-    // First try to push down any predicates from the WHERE clause
-    if let Some(predicates) = where_clause {
-        let mut i = 0;
-        while i < predicates.len() {
-            // Take ownership of predicate to try pushing it down
-            let predicate = predicates[i].take_ownership();
-            // If predicate was successfully pushed (None returned), remove it from WHERE
-            let Some(predicate) = push_predicate(operator, predicate, referenced_tables)? else {
-                predicates.remove(i);
                 continue;
-            };
-            predicates[i] = predicate;
+            }
+            where_clause.truncate(0);
+            return Ok(ConstantConditionEliminationResult::ImpossibleCondition);
+        } else {
             i += 1;
         }
-        // Clean up empty WHERE clause
-        if predicates.is_empty() {
-            *where_clause = None;
-        }
     }
 
-    match operator {
-        SourceOperator::Subquery { .. } => Ok(()),
-        SourceOperator::Join {
-            left,
-            right,
-            predicates,
-            outer,
-            ..
-        } => {
-            // Recursively push predicates down both sides of join
-            push_predicates(left, where_clause, referenced_tables)?;
-            push_predicates(right, where_clause, referenced_tables)?;
-
-            if predicates.is_none() {
-                return Ok(());
-            }
-
-            let predicates = predicates.as_mut().unwrap();
-
-            let mut i = 0;
-            while i < predicates.len() {
-                let predicate_owned = predicates[i].take_ownership();
-
-                // For a join like SELECT * FROM left INNER JOIN right ON left.id = right.id AND left.name = 'foo'
-                // the predicate 'left.name = 'foo' can already be evaluated in the outer loop (left side of join)
-                // because the row can immediately be skipped if left.name != 'foo'.
-                // But for a LEFT JOIN, we can't do this since we need to ensure that all rows from the left table are included,
-                // even if there are no matching rows from the right table. This is why we can't push LEFT JOIN predicates to the left side.
-                let push_result = if *outer {
-                    Some(predicate_owned)
-                } else {
-                    push_predicate(left, predicate_owned, referenced_tables)?
-                };
-
-                // Try pushing to left side first (see comment above for reasoning)
-                let Some(predicate) = push_result else {
-                    predicates.remove(i);
-                    continue;
-                };
-
-                // Then try right side
-                let Some(predicate) = push_predicate(right, predicate, referenced_tables)? else {
-                    predicates.remove(i);
-                    continue;
-                };
-
-                // If neither side could take it, keep in join predicates (not sure if this actually happens in practice)
-                // this is effectively the same as pushing to the right side, so maybe it could be removed and assert here
-                // that we don't reach this code
-                predicates[i] = predicate;
-                i += 1;
-            }
-
-            Ok(())
-        }
-        // Base cases - nowhere else to push to
-        SourceOperator::Scan { .. } => Ok(()),
-        SourceOperator::Search { .. } => Ok(()),
-        SourceOperator::Nothing { .. } => Ok(()),
-    }
+    Ok(ConstantConditionEliminationResult::Continue)
 }
 
-/**
-  Push a single predicate down the tree, as far as possible.
-  Returns Ok(None) if the predicate was pushed, otherwise returns itself as Ok(Some(predicate))
-*/
-fn push_predicate(
-    operator: &mut SourceOperator,
-    predicate: ast::Expr,
-    referenced_tables: &Vec<TableReference>,
-) -> Result<Option<ast::Expr>> {
-    match operator {
-        SourceOperator::Subquery {
-            predicates,
-            table_reference,
-            ..
-        } => {
-            // **TODO**: we are currently just evaluating the predicate after the subquery yields,
-            // and not trying to do anythign more sophisticated.
-            // E.g. literally: SELECT * FROM (SELECT * FROM t1) sub WHERE sub.col = 'foo'
-            //
-            // It is possible, and not overly difficult, to determine that we can also push the
-            // predicate into the subquery coroutine itself before it yields. The above query would
-            // effectively become: SELECT * FROM (SELECT * FROM t1 WHERE col = 'foo') sub
-            //
-            // This matters more in cases where the subquery builds some kind of sorter/index in memory
-            // (or on disk) and in those cases pushing the predicate down to the coroutine will make the
-            // subquery produce less intermediate data. In cases where no intermediate data structures are
-            // built, it doesn't matter.
-            //
-            // Moreover, in many cases the subquery can even be completely eliminated, e.g. the above original
-            // query would become: SELECT * FROM t1 WHERE col = 'foo' without the subquery.
-            // **END TODO**
-
-            // Find position of this subquery in referenced_tables array
-            let subquery_index = referenced_tables
-                .iter()
-                .position(|t| {
-                    t.table_identifier == table_reference.table_identifier
-                        && matches!(t.reference_type, TableReferenceType::Subquery { .. })
-                })
-                .unwrap();
-
-            // Get bitmask showing which tables this predicate references
-            let predicate_bitmask =
-                get_table_ref_bitmask_for_ast_expr(referenced_tables, &predicate)?;
-
-            // Each table has a bit position based on join order from left to right
-            // e.g. in SELECT * FROM t1 JOIN t2 JOIN t3
-            // t1 is position 0 (001), t2 is position 1 (010), t3 is position 2 (100)
-            // To push a predicate to a given table, it can only reference that table and tables to its left
-            // Example: For table t2 at position 1 (bit 010):
-            // - Can push: 011 (t2 + t1), 001 (just t1), 010 (just t2)
-            // - Can't push: 110 (t2 + t3)
-            let next_table_on_the_right_in_join_bitmask = 1 << (subquery_index + 1);
-            if predicate_bitmask >= next_table_on_the_right_in_join_bitmask {
-                return Ok(Some(predicate));
-            }
-
-            if predicates.is_none() {
-                predicates.replace(vec![predicate]);
-            } else {
-                predicates.as_mut().unwrap().push(predicate);
-            }
-
-            Ok(None)
-        }
-        SourceOperator::Scan {
-            predicates,
-            table_reference,
-            ..
-        } => {
-            // Find position of this table in referenced_tables array
-            let table_index = referenced_tables
-                .iter()
-                .position(|t| {
-                    t.table_identifier == table_reference.table_identifier
-                        && t.reference_type == TableReferenceType::BTreeTable
-                })
-                .unwrap();
-
-            // Get bitmask showing which tables this predicate references
-            let predicate_bitmask =
-                get_table_ref_bitmask_for_ast_expr(referenced_tables, &predicate)?;
-
-            // Each table has a bit position based on join order from left to right
-            // e.g. in SELECT * FROM t1 JOIN t2 JOIN t3
-            // t1 is position 0 (001), t2 is position 1 (010), t3 is position 2 (100)
-            // To push a predicate to a given table, it can only reference that table and tables to its left
-            // Example: For table t2 at position 1 (bit 010):
-            // - Can push: 011 (t2 + t1), 001 (just t1), 010 (just t2)
-            // - Can't push: 110 (t2 + t3)
-            let next_table_on_the_right_in_join_bitmask = 1 << (table_index + 1);
-            if predicate_bitmask >= next_table_on_the_right_in_join_bitmask {
-                return Ok(Some(predicate));
-            }
-
-            // Add predicate to this table's filters
-            if predicates.is_none() {
-                predicates.replace(vec![predicate]);
-            } else {
-                predicates.as_mut().unwrap().push(predicate);
-            }
-
-            Ok(None)
-        }
-        // Search nodes don't exist yet at this point; Scans are transformed to Search in use_indexes()
-        SourceOperator::Search { .. } => unreachable!(),
-        SourceOperator::Join {
-            left,
-            right,
-            predicates: join_on_preds,
-            outer,
-            ..
-        } => {
-            // Try pushing to left side first
-            let push_result_left = push_predicate(left, predicate, referenced_tables)?;
-            if push_result_left.is_none() {
-                return Ok(None);
-            }
-            // Then try right side
-            let push_result_right =
-                push_predicate(right, push_result_left.unwrap(), referenced_tables)?;
-            if push_result_right.is_none() {
-                return Ok(None);
-            }
-
-            // For LEFT JOIN, predicates must stay at join level
-            if *outer {
-                return Ok(Some(push_result_right.unwrap()));
-            }
-
-            let pred = push_result_right.unwrap();
-
-            // Get bitmasks for tables referenced in predicate and both sides of join
-            let table_refs_bitmask = get_table_ref_bitmask_for_ast_expr(referenced_tables, &pred)?;
-            let left_bitmask = get_table_ref_bitmask_for_operator(referenced_tables, left)?;
-            let right_bitmask = get_table_ref_bitmask_for_operator(referenced_tables, right)?;
-
-            // If predicate doesn't reference tables from both sides, it can't be a join condition
-            if table_refs_bitmask & left_bitmask == 0 || table_refs_bitmask & right_bitmask == 0 {
-                return Ok(Some(pred));
-            }
-
-            // Add as join predicate since it references both sides
-            if join_on_preds.is_none() {
-                join_on_preds.replace(vec![pred]);
-            } else {
-                join_on_preds.as_mut().unwrap().push(pred);
-            }
-
-            Ok(None)
-        }
-        SourceOperator::Nothing { .. } => Ok(Some(predicate)),
-    }
-}
-
-fn push_scan_direction(operator: &mut SourceOperator, direction: &Direction) {
-    match operator {
-        SourceOperator::Scan { iter_dir, .. } => {
+fn push_scan_direction(table: &mut TableReference, direction: &Direction) {
+    match &mut table.op {
+        Operation::Scan { iter_dir, .. } => {
             if iter_dir.is_none() {
                 match direction {
                     Direction::Ascending => *iter_dir = Some(IterationDirection::Forwards),
@@ -599,22 +211,19 @@ fn push_scan_direction(operator: &mut SourceOperator, direction: &Direction) {
                 }
             }
         }
-        _ => todo!(),
+        _ => {}
     }
 }
 
 fn rewrite_exprs_select(plan: &mut SelectPlan) -> Result<()> {
-    rewrite_source_operator_exprs(&mut plan.source)?;
     for rc in plan.result_columns.iter_mut() {
         rewrite_expr(&mut rc.expr)?;
     }
     for agg in plan.aggregates.iter_mut() {
         rewrite_expr(&mut agg.original_expr)?;
     }
-    if let Some(predicates) = &mut plan.where_clause {
-        for expr in predicates {
-            rewrite_expr(expr)?;
-        }
+    for cond in plan.where_clause.iter_mut() {
+        rewrite_expr(&mut cond.expr)?;
     }
     if let Some(group_by) = &mut plan.group_by {
         for expr in group_by.exprs.iter_mut() {
@@ -631,57 +240,12 @@ fn rewrite_exprs_select(plan: &mut SelectPlan) -> Result<()> {
 }
 
 fn rewrite_exprs_delete(plan: &mut DeletePlan) -> Result<()> {
-    rewrite_source_operator_exprs(&mut plan.source)?;
-    if let Some(predicates) = &mut plan.where_clause {
-        for expr in predicates {
-            rewrite_expr(expr)?;
-        }
+    for cond in plan.where_clause.iter_mut() {
+        rewrite_expr(&mut cond.expr)?;
     }
-
     Ok(())
 }
 
-fn rewrite_source_operator_exprs(operator: &mut SourceOperator) -> Result<()> {
-    match operator {
-        SourceOperator::Join {
-            left,
-            right,
-            predicates,
-            ..
-        } => {
-            rewrite_source_operator_exprs(left)?;
-            rewrite_source_operator_exprs(right)?;
-
-            if let Some(predicates) = predicates {
-                for expr in predicates.iter_mut() {
-                    rewrite_expr(expr)?;
-                }
-            }
-
-            Ok(())
-        }
-        SourceOperator::Scan { predicates, .. } | SourceOperator::Search { predicates, .. } => {
-            if let Some(predicates) = predicates {
-                for expr in predicates.iter_mut() {
-                    rewrite_expr(expr)?;
-                }
-            }
-
-            Ok(())
-        }
-        SourceOperator::Subquery { predicates, .. } => {
-            if let Some(predicates) = predicates {
-                for expr in predicates.iter_mut() {
-                    rewrite_expr(expr)?;
-                }
-            }
-
-            Ok(())
-        }
-        SourceOperator::Nothing { .. } => Ok(()),
-    }
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum ConstantPredicate {
     AlwaysTrue,
@@ -709,7 +273,7 @@ pub trait Optimizable {
     fn check_index_scan(
         &mut self,
         table_index: usize,
-        referenced_tables: &[TableReference],
+        table_reference: &TableReference,
         available_indexes: &[Rc<Index>],
     ) -> Result<Option<usize>>;
 }
@@ -728,7 +292,7 @@ impl Optimizable for ast::Expr {
     fn check_index_scan(
         &mut self,
         table_index: usize,
-        referenced_tables: &[TableReference],
+        table_reference: &TableReference,
         available_indexes: &[Rc<Index>],
     ) -> Result<Option<usize>> {
         match self {
@@ -737,9 +301,8 @@ impl Optimizable for ast::Expr {
                     return Ok(None);
                 }
                 for (idx, index) in available_indexes.iter().enumerate() {
-                    let table_ref = &referenced_tables[*table];
-                    if index.table_name == table_ref.table.get_name() {
-                        let column = table_ref.table.get_column_at(*column);
+                    if index.table_name == table_reference.table.get_name() {
+                        let column = table_reference.table.get_column_at(*column);
                         if index.columns.first().unwrap().name == column.name {
                             return Ok(Some(idx));
                         }
@@ -766,12 +329,12 @@ impl Optimizable for ast::Expr {
                     return Ok(None);
                 }
                 let lhs_index =
-                    lhs.check_index_scan(table_index, referenced_tables, available_indexes)?;
+                    lhs.check_index_scan(table_index, &table_reference, available_indexes)?;
                 if lhs_index.is_some() {
                     return Ok(lhs_index);
                 }
                 let rhs_index =
-                    rhs.check_index_scan(table_index, referenced_tables, available_indexes)?;
+                    rhs.check_index_scan(table_index, &table_reference, available_indexes)?;
                 if rhs_index.is_some() {
                     // swap lhs and rhs
                     let swapped_operator = match *op {
@@ -911,31 +474,41 @@ impl Optimizable for ast::Expr {
     }
 }
 
-pub enum Either<T, U> {
-    Left(T),
-    Right(U),
-}
-
 pub fn try_extract_index_search_expression(
-    expr: ast::Expr,
+    cond: &mut JoinAwareConditionExpr,
     table_index: usize,
-    referenced_tables: &[TableReference],
+    table_reference: &TableReference,
     available_indexes: &[Rc<Index>],
-) -> Result<Either<ast::Expr, Search>> {
-    match expr {
-        ast::Expr::Binary(mut lhs, operator, mut rhs) => {
+) -> Result<Option<Search>> {
+    if cond.eval_at_loop != table_index {
+        return Ok(None);
+    }
+    match &mut cond.expr {
+        ast::Expr::Binary(lhs, operator, rhs) => {
             if lhs.is_rowid_alias_of(table_index) {
                 match operator {
                     ast::Operator::Equals => {
-                        return Ok(Either::Right(Search::RowidEq { cmp_expr: *rhs }));
+                        let rhs_owned = rhs.take_ownership();
+                        return Ok(Some(Search::RowidEq {
+                            cmp_expr: JoinAwareConditionExpr {
+                                expr: rhs_owned,
+                                from_outer_join: cond.from_outer_join,
+                                eval_at_loop: cond.eval_at_loop,
+                            },
+                        }));
                     }
                     ast::Operator::Greater
                     | ast::Operator::GreaterEquals
                     | ast::Operator::Less
                     | ast::Operator::LessEquals => {
-                        return Ok(Either::Right(Search::RowidSearch {
-                            cmp_op: operator,
-                            cmp_expr: *rhs,
+                        let rhs_owned = rhs.take_ownership();
+                        return Ok(Some(Search::RowidSearch {
+                            cmp_op: *operator,
+                            cmp_expr: JoinAwareConditionExpr {
+                                expr: rhs_owned,
+                                from_outer_join: cond.from_outer_join,
+                                eval_at_loop: cond.eval_at_loop,
+                            },
                         }));
                     }
                     _ => {}
@@ -945,15 +518,27 @@ pub fn try_extract_index_search_expression(
             if rhs.is_rowid_alias_of(table_index) {
                 match operator {
                     ast::Operator::Equals => {
-                        return Ok(Either::Right(Search::RowidEq { cmp_expr: *lhs }));
+                        let lhs_owned = lhs.take_ownership();
+                        return Ok(Some(Search::RowidEq {
+                            cmp_expr: JoinAwareConditionExpr {
+                                expr: lhs_owned,
+                                from_outer_join: cond.from_outer_join,
+                                eval_at_loop: cond.eval_at_loop,
+                            },
+                        }));
                     }
                     ast::Operator::Greater
                     | ast::Operator::GreaterEquals
                     | ast::Operator::Less
                     | ast::Operator::LessEquals => {
-                        return Ok(Either::Right(Search::RowidSearch {
-                            cmp_op: operator,
-                            cmp_expr: *lhs,
+                        let lhs_owned = lhs.take_ownership();
+                        return Ok(Some(Search::RowidSearch {
+                            cmp_op: *operator,
+                            cmp_expr: JoinAwareConditionExpr {
+                                expr: lhs_owned,
+                                from_outer_join: cond.from_outer_join,
+                                eval_at_loop: cond.eval_at_loop,
+                            },
                         }));
                     }
                     _ => {}
@@ -961,7 +546,7 @@ pub fn try_extract_index_search_expression(
             }
 
             if let Some(index_index) =
-                lhs.check_index_scan(table_index, referenced_tables, available_indexes)?
+                lhs.check_index_scan(table_index, &table_reference, available_indexes)?
             {
                 match operator {
                     ast::Operator::Equals
@@ -969,10 +554,15 @@ pub fn try_extract_index_search_expression(
                     | ast::Operator::GreaterEquals
                     | ast::Operator::Less
                     | ast::Operator::LessEquals => {
-                        return Ok(Either::Right(Search::IndexSearch {
+                        let rhs_owned = rhs.take_ownership();
+                        return Ok(Some(Search::IndexSearch {
                             index: available_indexes[index_index].clone(),
-                            cmp_op: operator,
-                            cmp_expr: *rhs,
+                            cmp_op: *operator,
+                            cmp_expr: JoinAwareConditionExpr {
+                                expr: rhs_owned,
+                                from_outer_join: cond.from_outer_join,
+                                eval_at_loop: cond.eval_at_loop,
+                            },
                         }));
                     }
                     _ => {}
@@ -980,7 +570,7 @@ pub fn try_extract_index_search_expression(
             }
 
             if let Some(index_index) =
-                rhs.check_index_scan(table_index, referenced_tables, available_indexes)?
+                rhs.check_index_scan(table_index, &table_reference, available_indexes)?
             {
                 match operator {
                     ast::Operator::Equals
@@ -988,19 +578,24 @@ pub fn try_extract_index_search_expression(
                     | ast::Operator::GreaterEquals
                     | ast::Operator::Less
                     | ast::Operator::LessEquals => {
-                        return Ok(Either::Right(Search::IndexSearch {
+                        let lhs_owned = lhs.take_ownership();
+                        return Ok(Some(Search::IndexSearch {
                             index: available_indexes[index_index].clone(),
-                            cmp_op: operator,
-                            cmp_expr: *lhs,
+                            cmp_op: *operator,
+                            cmp_expr: JoinAwareConditionExpr {
+                                expr: lhs_owned,
+                                from_outer_join: cond.from_outer_join,
+                                eval_at_loop: cond.eval_at_loop,
+                            },
                         }));
                     }
                     _ => {}
                 }
             }
 
-            Ok(Either::Left(ast::Expr::Binary(lhs, operator, rhs)))
+            Ok(None)
         }
-        _ => Ok(Either::Left(expr)),
+        _ => Ok(None),
     }
 }