merge join OK

src/algebra/op.rs

@@ -17,12 +17,13 @@ mod from;
 mod group;
 mod insert;
 mod limit;
+mod merge;
 mod nested_loop;
 mod scan;
 mod select;
+mod sort;
 mod tagged;
 mod values;
-mod sort;
 
 use crate::data::expr::Expr;
 use crate::data::tuple::{DataKind, OwnTuple, Tuple};
@@ -35,12 +36,13 @@ pub(crate) use from::*;
 pub(crate) use group::*;
 pub(crate) use insert::*;
 pub(crate) use limit::*;
+pub(crate) use merge::*;
 pub(crate) use nested_loop::*;
 pub(crate) use scan::*;
 pub(crate) use select::*;
+pub(crate) use sort::*;
 pub(crate) use tagged::*;
 pub(crate) use values::*;
-pub(crate) use sort::*;
 
 #[derive(thiserror::Error, Debug)]
 pub(crate) enum QueryError {

src/algebra/op/assoc.rs

@@ -88,7 +88,10 @@ impl<'b> RelationalAlgebra for AssocOp<'b> {
     fn binding_map(&self) -> Result<BindingMap> {
         let mut binding_map = self.source.binding_map()?;
         let mvi = binding_map.val_size;
-        let sub_map = binding_map.inner_map.entry(self.binding.clone()).or_default();
+        let sub_map = binding_map
+            .inner_map
+            .entry(self.binding.clone())
+            .or_default();
         for (i, info) in self.assoc_infos.iter().enumerate() {
             for (j, col) in info.vals.iter().enumerate() {
                 sub_map.insert(

src/algebra/op/cartesian.rs

@@ -1,8 +1,6 @@
 use crate::algebra::op::RelationalAlgebra;
 use crate::algebra::parser::RaBox;
-use crate::data::tuple_set::{
-    shift_merge_binding_map, BindingMap, TupleSet,
-};
+use crate::data::tuple_set::{shift_merge_binding_map, BindingMap, TupleSet};
 use crate::ddl::reify::TableInfo;
 use anyhow::Result;
 use std::collections::BTreeSet;

src/algebra/op/insert.rs

@@ -103,7 +103,7 @@ impl<'a> RelationalAlgebra for Insertion<'a> {
         Ok(BindingMap {
             inner_map: BTreeMap::from([(self.binding.clone(), inner)]),
             key_size: 1,
-            val_size: 1
+            val_size: 1,
         })
     }
 

src/algebra/op/merge.rs

@@ -0,0 +1,399 @@
+use crate::algebra::op::{drop_temp_table, RelationalAlgebra};
+use crate::algebra::parser::{assert_rule, build_relational_expr, AlgebraParseError, RaBox};
+use crate::context::TempDbContext;
+use crate::data::expr::{Expr, StaticExpr};
+use crate::data::op::NAME_OP_EQ;
+use crate::data::tuple::{DataKind, OwnTuple, Tuple};
+use crate::data::tuple_set::{merge_binding_maps, BindingMap, BindingMapEvalContext, TupleSet};
+use crate::ddl::reify::{DdlContext, TableInfo};
+use crate::parser::{Pairs, Rule};
+use anyhow::Result;
+use cozorocks::IteratorPtr;
+use std::cmp::Ordering;
+use std::collections::BTreeSet;
+use std::sync::atomic::AtomicU32;
+use std::sync::atomic::Ordering::SeqCst;
+
+pub(crate) const NAME_INNER_JOIN: &str = "Join";
+pub(crate) const NAME_LEFT_JOIN: &str = "LeftJoin";
+pub(crate) const NAME_RIGHT_JOIN: &str = "RightJoin";
+pub(crate) const NAME_OUTER_JOIN: &str = "OuterJoin";
+
+pub(crate) struct MergeJoin<'a> {
+    ctx: &'a TempDbContext<'a>,
+    pub(crate) left: RaBox<'a>,
+    pub(crate) right: RaBox<'a>,
+    pub(crate) join_keys: Vec<(StaticExpr, StaticExpr)>,
+    pub(crate) left_outer: bool,
+    pub(crate) right_outer: bool,
+    left_temp_id: AtomicU32,
+    right_temp_id: AtomicU32,
+}
+
+#[derive(thiserror::Error, Debug)]
+pub(crate) enum JoinError {
+    #[error("Invalid join condition {0:?}")]
+    JoinCondition(StaticExpr),
+    #[error("Join condition {0:?} must contain variables {1:?}")]
+    NoBoundVariable(StaticExpr, BTreeSet<String>),
+    #[error("Join condition {0:?} must not contain variables {1:?}")]
+    WrongBoundVariable(StaticExpr, BTreeSet<String>),
+}
+
+impl<'a> MergeJoin<'a> {
+    pub(crate) fn build(
+        ctx: &'a TempDbContext<'a>,
+        prev: Option<RaBox<'a>>,
+        mut args: Pairs,
+        kind: &str,
+    ) -> Result<Self> {
+        let not_enough_args = || AlgebraParseError::NotEnoughArguments(kind.to_string());
+        let left = match prev {
+            Some(v) => v,
+            None => build_relational_expr(ctx, args.next().ok_or_else(not_enough_args)?)?,
+        };
+        let right = build_relational_expr(ctx, args.next().ok_or_else(not_enough_args)?)?;
+
+        let left_bindings = left.bindings()?;
+        let right_bindings = right.bindings()?;
+        if !left_bindings.is_disjoint(&right_bindings) {
+            return Err(AlgebraParseError::DuplicateBinding(
+                left_bindings
+                    .intersection(&right_bindings)
+                    .next()
+                    .unwrap()
+                    .to_string(),
+            )
+            .into());
+        }
+        let mut join_keys: Vec<(StaticExpr, StaticExpr)> = vec![];
+        for (i, arg) in args.enumerate() {
+            let pair = arg.into_inner().next().unwrap();
+            assert_rule(&pair, Rule::expr, kind, i + 2)?;
+            let expr = Expr::try_from(pair)?;
+            match expr {
+                Expr::Apply(op, args) if op.name() == NAME_OP_EQ => {
+                    let mut args = args.into_iter();
+                    let left_condition = args.next().unwrap().into_static();
+                    let right_condition = args.next().unwrap().into_static();
+                    let left_variables = left_condition.all_variables();
+                    let right_variables = right_condition.all_variables();
+                    if left_variables.is_disjoint(&left_bindings) {
+                        return Err(
+                            JoinError::NoBoundVariable(left_condition, left_bindings).into()
+                        );
+                    }
+                    if right_variables.is_disjoint(&right_bindings) {
+                        return Err(
+                            JoinError::NoBoundVariable(right_condition, right_bindings).into()
+                        );
+                    }
+                    if !left_variables.is_disjoint(&right_bindings) {
+                        return Err(
+                            JoinError::WrongBoundVariable(left_condition, right_bindings).into(),
+                        );
+                    }
+                    if !right_variables.is_disjoint(&left_bindings) {
+                        return Err(
+                            JoinError::WrongBoundVariable(right_condition, left_bindings).into(),
+                        );
+                    }
+                    join_keys.push((left_condition, right_condition))
+                }
+                ex => return Err(JoinError::JoinCondition(ex.into_static()).into()),
+            }
+        }
+
+        Ok(Self {
+            ctx,
+            left,
+            right,
+            join_keys,
+            left_outer: matches!(kind, NAME_LEFT_JOIN | NAME_OUTER_JOIN),
+            right_outer: matches!(kind, NAME_RIGHT_JOIN | NAME_OUTER_JOIN),
+            left_temp_id: Default::default(),
+            right_temp_id: Default::default(),
+        })
+    }
+    fn materialize(
+        &self,
+        temp_table_id: u32,
+        keys: Vec<StaticExpr>,
+        source: &RaBox<'a>,
+    ) -> Result<()> {
+        let source_map = source.binding_map()?;
+        let binding_ctx = BindingMapEvalContext {
+            map: &source_map,
+            parent: self.ctx,
+        };
+        let sort_exprs = keys
+            .iter()
+            .map(|ex| ex.clone().partial_eval(&binding_ctx))
+            .collect::<Result<Vec<_>>>()?;
+        let mut insertion_key = OwnTuple::with_prefix(temp_table_id);
+        let mut insertion_val = OwnTuple::with_data_prefix(DataKind::Data);
+        for (i, tset) in source.iter()?.enumerate() {
+            insertion_key.truncate_all();
+            insertion_val.truncate_all();
+            let tset = tset?;
+            for expr in &sort_exprs {
+                let val = expr.row_eval(&tset)?;
+                insertion_key.push_value(&val);
+            }
+            insertion_key.push_int(i as i64);
+            tset.encode_as_tuple(&mut insertion_val);
+            self.ctx
+                .sess
+                .temp
+                .put(&self.ctx.sess.w_opts_temp, &insertion_key, &insertion_val)?;
+        }
+        Ok(())
+    }
+}
+
+impl<'a> Drop for MergeJoin<'a> {
+    fn drop(&mut self) {
+        drop_temp_table(self.ctx, self.left_temp_id.load(SeqCst));
+        drop_temp_table(self.ctx, self.right_temp_id.load(SeqCst));
+    }
+}
+
+impl<'b> RelationalAlgebra for MergeJoin<'b> {
+    fn name(&self) -> &str {
+        match (self.left_outer, self.right_outer) {
+            (false, false) => NAME_INNER_JOIN,
+            (true, false) => NAME_LEFT_JOIN,
+            (false, true) => NAME_RIGHT_JOIN,
+            (true, true) => NAME_OUTER_JOIN,
+        }
+    }
+
+    fn bindings(&self) -> Result<BTreeSet<String>> {
+        let mut left = self.left.bindings()?;
+        let right = self.right.bindings()?;
+        left.extend(right);
+        Ok(left)
+    }
+
+    fn binding_map(&self) -> Result<BindingMap> {
+        let left = self.left.binding_map()?;
+        let right = self.right.binding_map()?;
+        Ok(merge_binding_maps([left, right].into_iter()))
+    }
+
+    fn iter<'a>(&'a self) -> Result<Box<dyn Iterator<Item = Result<TupleSet>> + 'a>> {
+        let mut left_temp_id = self.left_temp_id.load(SeqCst);
+        if left_temp_id == 0 {
+            left_temp_id = self.ctx.gen_table_id()?.id;
+            self.left_temp_id.store(left_temp_id, SeqCst);
+            self.materialize(
+                left_temp_id,
+                self.join_keys
+                    .iter()
+                    .map(|(l, _)| l.clone())
+                    .collect::<Vec<_>>(),
+                &self.left,
+            )?;
+        }
+
+        let mut right_temp_id = self.right_temp_id.load(SeqCst);
+        if right_temp_id == 0 {
+            right_temp_id = self.ctx.gen_table_id()?.id;
+            self.left_temp_id.store(right_temp_id, SeqCst);
+            self.materialize(
+                right_temp_id,
+                self.join_keys
+                    .iter()
+                    .map(|(_, r)| r.clone())
+                    .collect::<Vec<_>>(),
+                &self.right,
+            )?;
+        }
+        Ok(Box::new(MergeJoinIterator {
+            left_tid: left_temp_id,
+            right_tid: right_temp_id,
+            left_tset_size: self.left.binding_map()?.kv_size(),
+            right_tset_size: self.right.binding_map()?.kv_size(),
+            left_outer: self.left_outer,
+            right_outer: self.right_outer,
+            key_len: self.join_keys.len(),
+            left_it: self.ctx.sess.temp.iterator(&self.ctx.sess.r_opts_temp),
+            right_it: self.ctx.sess.temp.iterator(&self.ctx.sess.r_opts_temp),
+            last_op: MergeJoinIteratorLastOp::NotStarted,
+            scratch: OwnTuple::with_null_prefix(),
+        }))
+    }
+
+    fn identity(&self) -> Option<TableInfo> {
+        None
+    }
+}
+
+#[derive(Copy, Clone, Debug, Eq, PartialEq)]
+enum MergeJoinIteratorLastOp {
+    NotStarted,
+    LeftOutput,
+    RightOutput,
+    NestedLoopOutput,
+    Done,
+}
+
+struct MergeJoinIterator {
+    left_tid: u32,
+    right_tid: u32,
+    left_tset_size: (usize, usize),
+    right_tset_size: (usize, usize),
+    left_outer: bool,
+    right_outer: bool,
+    key_len: usize,
+    left_it: IteratorPtr,
+    right_it: IteratorPtr,
+    last_op: MergeJoinIteratorLastOp,
+    scratch: OwnTuple,
+}
+
+impl MergeJoinIterator {
+    fn next_inner(&mut self) -> Result<Option<TupleSet>> {
+        loop {
+            match self.last_op {
+                MergeJoinIteratorLastOp::NotStarted => {
+                    self.scratch.overwrite_prefix(self.left_tid);
+                    self.left_it.seek(&self.scratch);
+                    self.scratch.overwrite_prefix(self.right_tid);
+                    self.right_it.seek(&self.scratch);
+                }
+                MergeJoinIteratorLastOp::Done => return Ok(None),
+                MergeJoinIteratorLastOp::LeftOutput => {
+                    self.left_it.next();
+                }
+                MergeJoinIteratorLastOp::RightOutput => {
+                    self.right_it.next();
+                }
+                MergeJoinIteratorLastOp::NestedLoopOutput => {
+                    self.right_it.next();
+                    if let Some(right_key) = self.right_it.key() {
+                        let right_key = Tuple::new(right_key);
+                        let left_key = self.left_it.key().unwrap();
+                        let left_key = Tuple::new(left_key);
+                        match left_key.key_part_truncate_cmp(&right_key, self.key_len) {
+                            Ordering::Less => {
+                                let old_key = left_key.to_owned();
+                                self.left_it.next();
+                                match self.left_it.key() {
+                                    None => {
+                                        self.last_op = MergeJoinIteratorLastOp::RightOutput;
+                                        continue;
+                                    }
+                                    Some(left_key) => {
+                                        let left_key = Tuple::new(left_key);
+                                        match old_key.key_part_truncate_cmp(&left_key, self.key_len)
+                                        {
+                                            Ordering::Less => {
+                                                self.scratch.truncate_all();
+                                                self.scratch.overwrite_prefix(self.right_tid);
+                                                for val in old_key.iter() {
+                                                    self.scratch.push_value(&val?)
+                                                }
+                                                self.right_it.seek(&self.scratch);
+                                            }
+                                            Ordering::Equal => {}
+                                            Ordering::Greater => unreachable!(),
+                                        }
+                                    }
+                                }
+                            }
+                            Ordering::Equal => {}
+                            Ordering::Greater => unreachable!(),
+                        }
+                    } else {
+                        self.last_op = MergeJoinIteratorLastOp::LeftOutput;
+                        continue;
+                    }
+                }
+            }
+            match (self.left_it.pair(), self.right_it.pair()) {
+                (None, None) => {
+                    self.last_op = MergeJoinIteratorLastOp::Done;
+                    return Ok(None);
+                }
+                (None, Some((_rk, rv))) => {
+                    return if self.right_outer {
+                        self.last_op = MergeJoinIteratorLastOp::RightOutput;
+                        let rv = Tuple::new(rv);
+                        let mut l_tset = TupleSet::padded_tset(self.left_tset_size);
+                        let r_tset = TupleSet::decode_from_tuple(&rv)?;
+                        l_tset.merge(r_tset);
+                        Ok(Some(l_tset))
+                    } else {
+                        self.last_op = MergeJoinIteratorLastOp::Done;
+                        Ok(None)
+                    }
+                }
+                (Some((_lk, lv)), None) => {
+                    return if self.left_outer {
+                        self.last_op = MergeJoinIteratorLastOp::LeftOutput;
+                        let lv = Tuple::new(lv);
+                        let mut l_tset = TupleSet::decode_from_tuple(&lv)?;
+                        let r_tset = TupleSet::padded_tset(self.right_tset_size);
+                        l_tset.merge(r_tset);
+                        Ok(Some(l_tset))
+                    } else {
+                        self.last_op = MergeJoinIteratorLastOp::Done;
+                        Ok(None)
+                    }
+                }
+                (Some((lk, lv)), Some((rk, rv))) => {
+                    let lk = Tuple::new(lk);
+                    let rk = Tuple::new(rk);
+                    match lk.key_part_truncate_cmp(&rk, self.key_len) {
+                        Ordering::Less => {
+                            self.last_op = MergeJoinIteratorLastOp::LeftOutput;
+                            if self.left_outer {
+                                let lv = Tuple::new(lv);
+                                let mut l_tset = TupleSet::decode_from_tuple(&lv)?;
+                                let r_tset = TupleSet::padded_tset(self.right_tset_size);
+                                l_tset.merge(r_tset);
+                                return Ok(Some(l_tset));
+                            } else {
+                                continue;
+                            }
+                        }
+                        Ordering::Greater => {
+                            self.last_op = MergeJoinIteratorLastOp::RightOutput;
+                            if self.right_outer {
+                                let rv = Tuple::new(rv);
+                                let mut l_tset = TupleSet::padded_tset(self.left_tset_size);
+                                let r_tset = TupleSet::decode_from_tuple(&rv)?;
+                                l_tset.merge(r_tset);
+                                return Ok(Some(l_tset));
+                            } else {
+                                continue;
+                            }
+                        }
+                        Ordering::Equal => {
+                            self.last_op = MergeJoinIteratorLastOp::NestedLoopOutput;
+                            let lv = Tuple::new(lv);
+                            let mut l_tset = TupleSet::decode_from_tuple(&lv)?;
+                            let rv = Tuple::new(rv);
+                            let r_tset = TupleSet::decode_from_tuple(&rv)?;
+                            l_tset.merge(r_tset);
+                            return Ok(Some(l_tset));
+                        }
+                    }
+                }
+            }
+        }
+    }
+}
+
+impl Iterator for MergeJoinIterator {
+    type Item = Result<TupleSet>;
+
+    fn next(&mut self) -> Option<Self::Item> {
+        match self.next_inner() {
+            Ok(Some(v)) => Some(Ok(v)),
+            Ok(None) => None,
+            Err(e) => Some(Err(e)),
+        }
+    }
+}

src/algebra/op/nested_loop.rs

@@ -10,9 +10,7 @@ use crate::data::tuple_set::{
 use crate::ddl::reify::TableInfo;
 use crate::runtime::options::{default_read_options, default_write_options};
 use anyhow::Result;
-use cozorocks::{
-    DbPtr, PrefixIterator, ReadOptionsPtr, TransactionPtr, WriteOptionsPtr,
-};
+use cozorocks::{DbPtr, PrefixIterator, ReadOptionsPtr, TransactionPtr, WriteOptionsPtr};
 use std::collections::{BTreeMap, BTreeSet};
 
 pub(crate) const NAME_NESTED_LOOP_LEFT: &str = "NestedLoop";
@@ -44,7 +42,7 @@ fn nested_binding_map(
     let right_map = BindingMap {
         inner_map: BTreeMap::from([(right_binding.to_string(), right)]),
         key_size: 1,
-        val_size: 1
+        val_size: 1,
     };
     shift_merge_binding_map(&mut binding_map, right_map);
     Ok(binding_map)

src/algebra/op/select.rs

@@ -121,7 +121,7 @@ impl<'b> RelationalAlgebra for SelectOp<'b> {
         Ok(BindingMap {
             inner_map: BTreeMap::from([(self.binding.clone(), extract_map)]),
             key_size: 1,
-            val_size: 1
+            val_size: 1,
         })
     }
 

src/algebra/op/sort.rs

@@ -10,7 +10,6 @@ use crate::parser::{Pairs, Rule};
 use crate::runtime::options::default_read_options;
 use anyhow::Result;
 use log::error;
-use std::cell::RefCell;
 use std::cmp::Reverse;
 use std::collections::BTreeSet;
 use std::sync::atomic::{AtomicU32, Ordering};
@@ -66,7 +65,6 @@ impl<'a> SortOp<'a> {
     }
     fn sort_data(&self) -> Result<()> {
         let temp_table_id = self.temp_table_id.load(Ordering::SeqCst);
-        dbg!(temp_table_id);
         assert!(temp_table_id > MIN_TABLE_ID_BOUND);
         let source_map = self.source.binding_map()?;
         let binding_ctx = BindingMapEvalContext {
@@ -105,22 +103,24 @@ impl<'a> SortOp<'a> {
     }
 }
 
+pub(crate) fn drop_temp_table(ctx: &TempDbContext, id: u32) {
+    if id > MIN_TABLE_ID_BOUND {
+        let start_key = OwnTuple::with_prefix(id);
+        let mut end_key = OwnTuple::with_prefix(id);
+        end_key.seal_with_sentinel();
+        if let Err(e) = ctx
+            .sess
+            .temp
+            .del_range(&ctx.sess.w_opts_temp, start_key, end_key)
+        {
+            error!("Undefine temp table failed: {:?}", e)
+        }
+    }
+}
+
 impl<'a> Drop for SortOp<'a> {
     fn drop(&mut self) {
-        let id = self.temp_table_id.load(Ordering::SeqCst);
-        if id > MIN_TABLE_ID_BOUND {
-            let start_key = OwnTuple::with_prefix(id);
-            let mut end_key = OwnTuple::with_prefix(id);
-            end_key.seal_with_sentinel();
-            if let Err(e) =
-                self.ctx
-                    .sess
-                    .temp
-                    .del_range(&self.ctx.sess.w_opts_temp, start_key, end_key)
-            {
-                error!("Undefine temp table failed: {:?}", e)
-            }
-        }
+        drop_temp_table(self.ctx, self.temp_table_id.load(Ordering::SeqCst));
     }
 }
 

src/algebra/op/tagged.rs

@@ -273,7 +273,7 @@ impl<'b> RelationalAlgebra for TaggedInsertion<'b> {
                 ]),
             )]),
             key_size: 1,
-            val_size: 1
+            val_size: 1,
         })
     }
 

src/algebra/parser.rs

@@ -7,7 +7,6 @@ use crate::ddl::reify::TableInfo;
 use crate::parser::{Pair, Rule};
 use anyhow::Result;
 use std::collections::BTreeSet;
-use std::error::Error;
 use std::fmt::{Debug, Formatter};
 
 #[derive(thiserror::Error, Debug)]
@@ -72,6 +71,7 @@ pub(crate) enum RaBox<'a> {
     Cartesian(Box<CartesianJoin<'a>>),
     NestedLoopLeft(Box<NestedLoopLeft<'a>>),
     SortOp(Box<SortOp<'a>>),
+    MergeJoin(Box<MergeJoin<'a>>),
 }
 
 impl<'a> RaBox<'a> {
@@ -88,6 +88,7 @@ impl<'a> RaBox<'a> {
             RaBox::Cartesian(inner) => vec![&inner.left, &inner.right],
             RaBox::NestedLoopLeft(inner) => vec![&inner.left],
             RaBox::SortOp(inner) => vec![&inner.source],
+            RaBox::MergeJoin(inner) => vec![&inner.left, &inner.right],
         }
     }
 }
@@ -116,6 +117,7 @@ impl<'b> RelationalAlgebra for RaBox<'b> {
             RaBox::Cartesian(inner) => inner.name(),
             RaBox::NestedLoopLeft(inner) => inner.name(),
             RaBox::SortOp(inner) => inner.name(),
+            RaBox::MergeJoin(inner) => inner.name(),
         }
     }
 
@@ -132,6 +134,7 @@ impl<'b> RelationalAlgebra for RaBox<'b> {
             RaBox::Cartesian(inner) => inner.bindings(),
             RaBox::NestedLoopLeft(inner) => inner.bindings(),
             RaBox::SortOp(inner) => inner.bindings(),
+            RaBox::MergeJoin(inner) => inner.bindings(),
         }
     }
 
@@ -148,6 +151,7 @@ impl<'b> RelationalAlgebra for RaBox<'b> {
             RaBox::Cartesian(inner) => inner.binding_map(),
             RaBox::NestedLoopLeft(inner) => inner.binding_map(),
             RaBox::SortOp(inner) => inner.binding_map(),
+            RaBox::MergeJoin(inner) => inner.binding_map(),
         }
     }
 
@@ -164,6 +168,7 @@ impl<'b> RelationalAlgebra for RaBox<'b> {
             RaBox::Cartesian(inner) => inner.iter(),
             RaBox::NestedLoopLeft(inner) => inner.iter(),
             RaBox::SortOp(inner) => inner.iter(),
+            RaBox::MergeJoin(inner) => inner.iter(),
         }
     }
 
@@ -180,15 +185,24 @@ impl<'b> RelationalAlgebra for RaBox<'b> {
             RaBox::Cartesian(inner) => inner.identity(),
             RaBox::NestedLoopLeft(inner) => inner.identity(),
             RaBox::SortOp(inner) => inner.identity(),
+            RaBox::MergeJoin(inner) => inner.identity(),
         }
     }
 }
 
-pub(crate) fn build_relational_expr<'a>(ctx: &'a TempDbContext, pair: Pair) -> Result<RaBox<'a>> {
+pub(crate) fn build_relational_expr<'a>(
+    ctx: &'a TempDbContext,
+    mut pair: Pair,
+) -> Result<RaBox<'a>> {
+    if pair.as_rule() == Rule::ra_arg {
+        pair = pair.into_inner().next().unwrap();
+    }
+    assert_rule(&pair, Rule::ra_expr, pair.as_str(), 0)?;
     let mut built: Option<RaBox> = None;
     for pair in pair.into_inner() {
         let mut pairs = pair.into_inner();
-        match pairs.next().unwrap().as_str() {
+        let pair = pairs.next().unwrap();
+        match pair.as_str() {
             NAME_INSERTION => {
                 built = Some(RaBox::Insertion(Box::new(Insertion::build(
                     ctx, built, pairs, false,
@@ -227,18 +241,20 @@ pub(crate) fn build_relational_expr<'a>(ctx: &'a TempDbContext, pair: Pair) -> R
                     ctx, built, pairs,
                 )?)))
             }
-            NAME_TAKE => {
+            n @ (NAME_TAKE | NAME_SKIP) => {
                 built = Some(RaBox::LimitOp(Box::new(LimitOp::build(
-                    ctx, built, pairs, NAME_TAKE,
+                    ctx, built, pairs, n,
                 )?)))
             }
-            NAME_SKIP => {
-                built = Some(RaBox::LimitOp(Box::new(LimitOp::build(
-                    ctx, built, pairs, NAME_SKIP,
+            NAME_SORT => built = Some(RaBox::SortOp(Box::new(SortOp::build(ctx, built, pairs)?))),
+            n @ (NAME_INNER_JOIN | NAME_LEFT_JOIN | NAME_RIGHT_JOIN | NAME_OUTER_JOIN) => {
+                built = Some(RaBox::MergeJoin(Box::new(MergeJoin::build(
+                    ctx, built, pairs, n,
                 )?)))
             }
-            NAME_SORT => built = Some(RaBox::SortOp(Box::new(SortOp::build(ctx, built, pairs)?))),
-            _ => unimplemented!(),
+            name => {
+                unimplemented!("{}", name)
+            }
         }
     }
     Ok(built.unwrap())
@@ -363,6 +379,37 @@ pub(crate) mod tests {
         }
         let duration_join_back = start.elapsed();
         let start = Instant::now();
+        {
+            let ctx = sess.temp_ctx(true);
+            let s = r#"
+            OuterJoin(
+              From(l:Job),
+              From(r:Job),
+              l.min_salary == r.max_salary
+            ).Select({
+                l_id: l.id,
+                r_id: r.id,
+                l_min: l.min_salary,
+                // l_max: l.max_salary,
+                // r_min: r.min_salary,
+                r_max: r.max_salary,
+                l_title: l.title,
+                r_title: r.title,
+            })
+            "#;
+            let ra = build_relational_expr(
+                &ctx,
+                CozoParser::parse(Rule::ra_expr_all, s)
+                    .unwrap()
+                    .into_iter()
+                    .next()
+                    .unwrap(),
+            )?;
+            dbg!(&ra);
+            dbg!(ra.get_values()?);
+        }
+        let duration_outer_join = start.elapsed();
+        let start = Instant::now();
         let mut r_opts = default_read_options();
         r_opts.set_total_order_seek(true);
         r_opts.set_prefix_same_as_start(false);
@@ -384,6 +431,7 @@ pub(crate) mod tests {
             duration_scan,
             duration_join,
             duration_join_back,
+            duration_outer_join,
             duration_list,
             n
         );

src/data/expr.rs

@@ -7,7 +7,7 @@ use crate::data::value::{StaticValue, Value};
 use crate::parser::{CozoParser, Rule};
 use anyhow::Result;
 use pest::Parser;
-use std::collections::BTreeMap;
+use std::collections::{BTreeMap, BTreeSet};
 use std::fmt::{Debug, Formatter};
 use std::result;
 use std::sync::Arc;
@@ -73,6 +73,62 @@ impl<'a> Expr<'a> {
             _ => None,
         }
     }
+    pub(crate) fn all_variables(&self) -> BTreeSet<String> {
+        let mut ret = BTreeSet::new();
+        fn collect(ex: &Expr, accum: &mut BTreeSet<String>) {
+            match ex {
+                Expr::Const(_) => {}
+                Expr::List(ls) => {
+                    for el in ls {
+                        collect(el, accum);
+                    }
+                }
+                Expr::Dict(d) => {
+                    for el in d.values() {
+                        collect(el, accum);
+                    }
+                }
+                Expr::Variable(v) => {
+                    accum.insert(v.clone());
+                }
+                Expr::TupleSetIdx(_) => {}
+                Expr::Apply(_, args) => {
+                    for el in args {
+                        collect(el, accum);
+                    }
+                }
+                Expr::ApplyAgg(_, a_args, args) => {
+                    for el in args {
+                        collect(el, accum);
+                    }
+                    for el in a_args {
+                        collect(el, accum);
+                    }
+                }
+                Expr::FieldAcc(_, arg) => {
+                    collect(arg, accum);
+                }
+                Expr::IdxAcc(_, arg) => {
+                    collect(arg, accum);
+                }
+                Expr::IfExpr(args) => {
+                    let (if_p, then_p, else_p) = args.as_ref();
+                    collect(if_p, accum);
+                    collect(then_p, accum);
+                    collect(else_p, accum);
+                }
+                Expr::SwitchExpr(args) => {
+                    for (cond, el) in args {
+                        collect(cond, accum);
+                        collect(el, accum);
+                    }
+                }
+                ex => panic!("Unsupported on optimized expression: {:?}", ex),
+            }
+        }
+        collect(self, &mut ret);
+        ret
+    }
     pub(crate) fn into_static(self) -> StaticExpr {
         match self {
             Expr::Const(v) => Expr::Const(v.into_static()),

src/data/tuple.rs

@@ -207,6 +207,23 @@ impl<T: AsRef<[u8]>> Tuple<T> {
             .cmp(other.iter().map(|v| v.expect("Key comparison failed")))
     }
 
+    #[inline]
+    pub(crate) fn key_part_truncate_cmp<T2: AsRef<[u8]>>(
+        &self,
+        other: &Tuple<T2>,
+        n: usize,
+    ) -> Ordering {
+        self.iter()
+            .take(n)
+            .map(|v| v.expect("Key comparison failed"))
+            .cmp(
+                other
+                    .iter()
+                    .take(n)
+                    .map(|v| v.expect("Key comparison failed")),
+            )
+    }
+
     #[inline]
     pub(crate) fn new(data: T) -> Self {
         Self {

src/data/tuple_set.rs

@@ -141,6 +141,24 @@ impl TupleSet {
             target.push_bytes(v.as_ref());
         }
     }
+
+    pub(crate) fn padded_tset(padding: (usize, usize)) -> Self {
+        let mut ret = TupleSet {
+            keys: Vec::with_capacity(padding.0),
+            vals: Vec::with_capacity(padding.1),
+        };
+
+        for _ in 0..padding.0 {
+            ret.push_key(OwnTuple::empty_tuple().into());
+        }
+
+        for _ in 0..padding.1 {
+            ret.push_val(OwnTuple::empty_tuple().into());
+        }
+
+        ret
+    }
+
     pub(crate) fn decode_from_tuple<T: AsRef<[u8]>>(source: &Tuple<T>) -> Result<Self> {
         let gen_err = || DecodeFailure(source.to_owned());
         let k_len = source.get(0)?.get_int().ok_or_else(gen_err)? as usize;
@@ -315,6 +333,12 @@ pub(crate) struct BindingMap {
     pub(crate) val_size: usize,
 }
 
+impl BindingMap {
+    pub(crate) fn kv_size(&self) -> (usize, usize) {
+        (self.key_size, self.val_size)
+    }
+}
+
 pub(crate) fn merge_binding_maps(bmaps: impl Iterator<Item = BindingMap>) -> BindingMap {
     let mut ret: BindingMap = Default::default();