fix bugs to make subquery work

Signed-off-by: Runji Wang <wangrunji0408@163.com>

src/binder_v2/table.rs

@@ -115,6 +115,7 @@ impl Binder {
     fn bind_join_constraint(&mut self, constraint: JoinConstraint) -> Result {
         match constraint {
             JoinConstraint::On(expr) => self.bind_expr(expr),
+            JoinConstraint::None => Ok(self.egraph.add(Node::true_())),
             _ => todo!("Support more join constraints"),
         }
     }

src/executor_v2/mod.rs

@@ -167,14 +167,13 @@ impl<S: Storage> Builder<S> {
     /// Resolve the column index of `expr` in `plan`.
     fn resolve_column_index(&self, expr: Id, plan: Id) -> RecExpr {
         let schema = self.egraph[plan].data.schema.as_ref().expect("no schema");
-        self.node(expr).build_recexpr(|id| {
-            if let Some(idx) = schema.iter().position(|x| {
-                if let Expr::As([alias, _]) = self.node(*x) {
-                    *alias == id
-                } else {
-                    *x == id
-                }
-            }) {
+        self.node(expr).build_recexpr(|mut id| {
+            // skip AS
+            if let Expr::As([_, expr]) = self.node(id) {
+                id = *expr;
+            }
+            // resolve column index
+            if let Some(idx) = schema.iter().position(|x| *x == id) {
                 return Expr::ColumnIndex(ColumnIndex(idx as _));
             }
             match self.node(id) {

src/planner/mod.rs

@@ -158,6 +158,13 @@ impl Expr {
         v.clone()
     }
 
+    pub fn as_as(&self) -> Option<Id> {
+        match self {
+            Self::As([alias, _]) => Some(*alias),
+            _ => None,
+        }
+    }
+
     pub const fn binary_op(&self) -> Option<(BinaryOperator, Id, Id)> {
         use BinaryOperator as Op;
         #[allow(clippy::match_ref_pats)]

src/planner/rules/mod.rs

@@ -89,7 +89,11 @@ impl Analysis<Expr> for ExprAnalysis {
         Data {
             constant: expr::eval_constant(egraph, enode),
             columns: plan::analyze_columns(egraph, enode),
-            schema: schema::analyze_schema(enode, |i| egraph[*i].data.schema.clone()),
+            schema: schema::analyze_schema(
+                enode,
+                |i| egraph[*i].data.schema.clone(),
+                |i| egraph[*i].nodes[0].as_as(),
+            ),
             rows: rows::analyze_rows(egraph, enode),
         }
     }
@@ -150,7 +154,11 @@ impl Analysis<Expr> for TypeSchemaAnalysis {
     fn make(egraph: &egg::EGraph<Expr, Self>, enode: &Expr) -> Self::Data {
         TypeSchema {
             type_: type_::analyze_type(enode, |i| egraph[*i].data.type_.clone(), &egraph.analysis),
-            schema: schema::analyze_schema(enode, |i| egraph[*i].data.schema.clone()),
+            schema: schema::analyze_schema(
+                enode,
+                |i| egraph[*i].data.schema.clone(),
+                |i| egraph[*i].nodes[0].as_as(),
+            ),
             aggs: agg::analyze_aggs(enode, |i| egraph[*i].data.aggs.clone()),
         }
     }

src/planner/rules/plan.rs

@@ -44,10 +44,6 @@ fn merge_rules() -> Vec<Rewrite> { vec![
         "(filter ?cond1 (filter ?cond2 ?child))" =>
         "(filter (and ?cond1 ?cond2) ?child)"
     ),
-    rw!("proj-merge";
-        "(proj ?exprs1 (proj ?exprs2 ?child))" =>
-        "(proj ?exprs1 ?child)"
-    ),
 ]}
 
 #[rustfmt::skip]

src/planner/rules/schema.rs

@@ -6,7 +6,11 @@ use super::*;
 pub type Schema = Option<Vec<Id>>;
 
 /// Returns the output expressions for plan node.
-pub fn analyze_schema(enode: &Expr, x: impl Fn(&Id) -> Schema) -> Schema {
+pub fn analyze_schema(
+    enode: &Expr,
+    x: impl Fn(&Id) -> Schema,
+    as_as: impl Fn(&Id) -> Option<Id>,
+) -> Schema {
     use Expr::*;
     let concat = |v1: Vec<Id>, v2: Vec<Id>| v1.into_iter().chain(v2.into_iter()).collect();
     Some(match enode {
@@ -17,7 +21,13 @@ pub fn analyze_schema(enode: &Expr, x: impl Fn(&Id) -> Schema) -> Schema {
         Join([_, _, l, r]) | HashJoin([_, _, _, l, r]) => concat(x(l)?, x(r)?),
 
         // list is the source for the following nodes
-        List(ids) => ids.to_vec(),
+        List(ids) => ids
+            .iter()
+            .map(|id| {
+                // only keep alias for `(as alias _)`
+                as_as(id).unwrap_or(*id)
+            })
+            .collect(),
 
         // plans that change schema
         Scan([_, columns]) => x(columns)?,