Repository: jiangzhe/xngin (rust)

Issue:
Support plan structural assertion for unit test
Add OpKind enum and method to return plan shape.

Pull request:
Add OpKind and fix HAVING validation
close #16 

Patch:
diff --git a/xngin-plan/Cargo.toml b/xngin-plan/Cargo.toml
index 0b422b8..87c2c3d 100644
--- a/xngin-plan/Cargo.toml
+++ b/xngin-plan/Cargo.toml
@@ -14,6 +14,7 @@ xngin-datatype = { version = "0.1.0", path = "../xngin-datatype" }
 xngin-catalog = { version = "0.1.0", path = "../xngin-catalog" }
 xngin-frontend = { version = "0.1.0", path = "../xngin-frontend" }
 xngin-expr = { version = "0.1.0", path = "../xngin-expr" }
+
 slab = "0.4"
 thiserror = "1.0"
 smol_str = "0.1"
diff --git a/xngin-plan/src/builder.rs b/xngin-plan/src/builder.rs
index c10c686..73ec377 100644
--- a/xngin-plan/src/builder.rs
+++ b/xngin-plan/src/builder.rs
@@ -266,7 +266,7 @@ impl PlanBuilder {
                             ))
                         }
                         DerivedCol::Expr(expr, alias) => {
-                            let e = resolver.resolve_expr(expr, "field list", phc)?;
+                            let e = resolver.resolve_expr(expr, "field list", phc, false)?;
                             cols.push((e.clone(), alias.to_lower()));
                             // alias conflict is detected when referring, here no need to check duplication
                         }
@@ -318,7 +318,7 @@ impl PlanBuilder {
                 qs: &self.qs,
                 query_aliases,
             };
-            let pred = resolver.resolve_expr(filter, "where clause", phc)?;
+            let pred = resolver.resolve_expr(filter, "where clause", phc, false)?;
             Some(pred)
         } else {
             None
@@ -349,8 +349,8 @@ impl PlanBuilder {
                 query_aliases,
                 proj_cols: &proj_cols,
             };
-            let having = resolver.resolve_expr(having, "having clause", phc)?;
-            validate_having(&groups, scalar_aggr, &having)?;
+            let having = resolver.resolve_expr(having, "having clause", phc, false)?;
+            validate_having(&proj_cols, &groups, scalar_aggr, &having)?;
             Some(having)
         } else {
             None
@@ -416,7 +416,7 @@ impl PlanBuilder {
             }
         }
         // f) build Sort operator
-        if !order.is_empty() && !scalar_aggr {
+        if !order.is_empty() {
             // scalar aggr query only returns 1 row, ingore SORT operation
             root = Op::sort(order, root);
         }
@@ -468,7 +468,7 @@ impl PlanBuilder {
     ) -> Result<Vec<SortItem>> {
         let mut res = Vec::with_capacity(elems.len());
         for elem in elems {
-            let expr = resolver.resolve_expr(&elem.expr, "order by", phc)?;
+            let expr = resolver.resolve_expr(&elem.expr, "order by", phc, false)?;
             let item = SortItem {
                 expr,
                 desc: elem.desc,
@@ -527,7 +527,7 @@ impl PlanBuilder {
                     proj_cols.extend(es);
                 }
                 DerivedCol::Expr(e, alias) => {
-                    let e = resolver.resolve_expr(e, location, phc)?;
+                    let e = resolver.resolve_expr(e, location, phc, false)?;
                     proj_cols.push((e, alias.to_lower()))
                 }
             }
@@ -686,7 +686,8 @@ impl PlanBuilder {
                                     qs: &self.qs,
                                     query_aliases: &from_aliases,
                                 };
-                                let pred = resolver.resolve_expr(e, "join condition", phc)?;
+                                let pred =
+                                    resolver.resolve_expr(e, "join condition", phc, false)?;
                                 (pred, left_aliases)
                             }
                         }
@@ -792,7 +793,10 @@ impl PlanBuilder {
     ) -> Result<Vec<expr::Expr>> {
         let mut items = Vec::with_capacity(group_by.len());
         for g in group_by {
-            let e = resolver.resolve_expr(g, "group by", phc)?;
+            let e = resolver.resolve_expr(g, "group by", phc, false)?;
+            if e.contains_aggr_func() {
+                return Err(Error::InvalidUsageOfAggrFunc);
+            }
             // the alias is temporary, will be updated in select clause
             items.push(e)
         }
@@ -894,6 +898,7 @@ fn validate_proj_aggr(
 /// 3. For flat projection, aggr functions are disallowed.
 #[inline]
 fn validate_having(
+    proj_cols: &[(expr::Expr, SmolStr)],
     aggr_groups: &[expr::Expr],
     scalar_aggr: bool,
     having: &expr::Expr,
@@ -917,9 +922,18 @@ fn validate_having(
         return Ok(());
     }
     // aggr functions are not allowed.
-    if having.contains_aggr_func() {
+    let (non_aggr_cols, aggr) = having.collect_non_aggr_cols();
+    if aggr {
         return Err(Error::FieldsSelectedNotInGroupBy);
     }
+    // columns must match projected expressions
+    for nac in non_aggr_cols {
+        let ce = &expr::Expr::Col(nac);
+        if !proj_cols.iter().any(|(e, _)| e == ce) {
+            // todo: notify column name
+            return Err(Error::UnknownColumn("Unknown column".to_string()));
+        }
+    }
     Ok(())
 }
 
@@ -1049,7 +1063,9 @@ impl ExprResolve for ResolveGroup<'_> {
     }
 }
 
-/// Resolver for HAVING
+/// Resolver for HAVING or ORDER BY
+/// The difference between them is HAVING does not care about
+/// tables/queries in FROM clause.
 pub struct ResolveHavingOrOrder<'a> {
     catalog: &'a dyn QueryCatalog,
     qs: &'a QuerySet,
diff --git a/xngin-plan/src/error.rs b/xngin-plan/src/error.rs
index e058098..3c2b85c 100644
--- a/xngin-plan/src/error.rs
+++ b/xngin-plan/src/error.rs
@@ -56,6 +56,8 @@ pub enum Error {
     DulicatedAsterisksInFieldList,
     #[error("Aggregate functions in GROUP BY clause")]
     AggrFuncInGroupBy,
+    #[error("Invalid usage of aggregate function")]
+    InvalidUsageOfAggrFunc,
     #[error(transparent)]
     ParseFloatError(#[from] std::num::ParseFloatError),
     #[error(transparent)]
diff --git a/xngin-plan/src/explain.rs b/xngin-plan/src/explain.rs
index c66f8f7..5ea67d7 100644
--- a/xngin-plan/src/explain.rs
+++ b/xngin-plan/src/explain.rs
@@ -206,9 +206,9 @@ impl Explain for Const {
 impl Explain for Col {
     fn explain<F: Write>(&self, f: &mut F) -> fmt::Result {
         match self {
-            Col::TableCol(table_id, idx) => write!(f, "tc({},{})", table_id.value(), idx),
-            Col::QueryCol(query_id, idx) => write!(f, "qc({},{})", **query_id, idx),
-            Col::CorrelatedCol(query_id, idx) => write!(f, "cc({},{})", **query_id, idx),
+            Col::TableCol(table_id, idx) => write!(f, "t{}.{}", table_id.value(), idx),
+            Col::QueryCol(query_id, idx) => write!(f, "q{}.{}", **query_id, idx),
+            Col::CorrelatedCol(query_id, idx) => write!(f, "cq{}.{}", **query_id, idx),
         }
     }
 }
diff --git a/xngin-plan/src/op.rs b/xngin-plan/src/op.rs
index 69e8e4f..9d28fc5 100644
--- a/xngin-plan/src/op.rs
+++ b/xngin-plan/src/op.rs
@@ -12,6 +12,20 @@ use smol_str::SmolStr;
 use xngin_catalog::{SchemaID, TableID};
 use xngin_expr::{Expr, Pred, QueryID, Setq};
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum OpKind {
+    Proj,
+    Filt,
+    Aggr,
+    Join,
+    Sort,
+    Limit,
+    Row,
+    Subquery,
+    Table,
+    Setop,
+}
+
 /// Op stands for logical operator.
 /// This is the general enum containing all nodes of logical plan.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -40,6 +54,22 @@ pub enum Op {
 }
 
 impl Op {
+    #[inline]
+    pub fn kind(&self) -> OpKind {
+        match self {
+            Op::Proj(_) => OpKind::Proj,
+            Op::Filt(_) => OpKind::Filt,
+            Op::Aggr(_) => OpKind::Aggr,
+            Op::Join(_) => OpKind::Join,
+            Op::Sort(_) => OpKind::Sort,
+            Op::Limit(_) => OpKind::Limit,
+            Op::Row(_) => OpKind::Row,
+            Op::Subquery(_) => OpKind::Subquery,
+            Op::Table(..) => OpKind::Table,
+            Op::Setop(_) => OpKind::Setop,
+        }
+    }
+
     #[inline]
     pub fn proj(cols: Vec<(Expr, SmolStr)>, q: Setq, source: Op) -> Self {
         Op::Proj(Proj {
diff --git a/xngin-plan/src/query.rs b/xngin-plan/src/query.rs
index 72cee74..4c4c1fe 100644
--- a/xngin-plan/src/query.rs
+++ b/xngin-plan/src/query.rs
@@ -1,5 +1,5 @@
 use crate::error::{Error, Result};
-use crate::op::{Join, JoinOp, Op, OpMutVisitor, OpVisitor};
+use crate::op::{Join, JoinOp, Op, OpKind, OpMutVisitor, OpVisitor};
 use crate::scope::Scope;
 use fnv::FnvHashMap;
 use slab::Slab;
@@ -14,6 +14,20 @@ pub struct QueryPlan {
     pub root: QueryID,
 }
 
+impl QueryPlan {
+    /// Returns the shape of the query plan.
+    /// The shape is presented by a sequence of Operators
+    /// generated by preorder traversal on current plan.
+    /// For example, The plan of "SELECT c0 FROM t1 JOIN t2"
+    /// is represented as "[Proj, Join, Proj, Scan, Proj, Scan]"
+    #[inline]
+    pub fn shape(&self) -> Vec<OpKind> {
+        let mut shape = vec![];
+        generate_shape(&self.queries, &self.root, &mut shape);
+        shape
+    }
+}
+
 /// Query wraps logical operator with additional syntax information.
 /// group operators as a tree, with output column list.
 /// it's equivalent to a simple SELECT statement, like below:
@@ -236,3 +250,32 @@ impl<'a> OpMutVisitor for UpsertQuery<'a> {
         true
     }
 }
+
+fn generate_shape(qs: &QuerySet, root: &QueryID, shape: &mut Vec<OpKind>) {
+    if let Some(subq) = qs.get(root) {
+        let mut sg = ShapeGen { qs, shape };
+        subq.root.walk(&mut sg);
+    }
+}
+
+struct ShapeGen<'a> {
+    qs: &'a QuerySet,
+    shape: &'a mut Vec<OpKind>,
+}
+
+impl OpVisitor for ShapeGen<'_> {
+    #[inline]
+    fn enter(&mut self, op: &Op) -> bool {
+        if let Op::Subquery(query_id) = op {
+            generate_shape(self.qs, query_id, self.shape);
+        } else {
+            self.shape.push(op.kind());
+        }
+        true
+    }
+
+    #[inline]
+    fn leave(&mut self, _op: &Op) -> bool {
+        true
+    }
+}
diff --git a/xngin-plan/src/resolv.rs b/xngin-plan/src/resolv.rs
index 68985b3..fa4680e 100644
--- a/xngin-plan/src/resolv.rs
+++ b/xngin-plan/src/resolv.rs
@@ -21,9 +21,6 @@ pub trait ExprResolve {
     /// It should be implemented in all scenarios, except row subquery.
     fn catalog(&self) -> Option<&dyn QueryCatalog>;
 
-    /// Returns query set for subquery lookup.
-    // fn query_set(&self) -> Option<&QuerySet>;
-
     /// Search visible subquery(table treated as subquery) by alias.
     /// It should be implemented in all scenarios, except row subquery.
     fn find_query(&self, tbl_alias: &str) -> Option<(QueryID, &Subquery)>;
@@ -277,6 +274,18 @@ pub trait ExprResolve {
         e: &'a Expr<'a>,
         location: &'static str,
         phc: &mut PlaceholderCollector<'a>,
+        within_aggr: bool,
+    ) -> Result<expr::Expr> {
+        self.resolve_expr_by_default(e, location, phc, within_aggr)
+    }
+
+    #[inline]
+    fn resolve_expr_by_default<'a>(
+        &self,
+        e: &'a Expr<'a>,
+        location: &'static str,
+        phc: &mut PlaceholderCollector<'a>,
+        within_aggr: bool,
     ) -> Result<expr::Expr> {
         let res = match e {
             Expr::Literal(lit) => self.resolve_lit(lit)?,
@@ -290,25 +299,31 @@ pub trait ExprResolve {
                     }
                 }
             },
-            Expr::AggrFunc(af) => self.resolve_aggr_func(af, location, phc)?,
+            Expr::AggrFunc(af) => {
+                if within_aggr {
+                    return Err(Error::InvalidUsageOfAggrFunc);
+                } else {
+                    self.resolve_aggr_func(af, location, phc)?
+                }
+            }
             Expr::Unary(ue) => match ue.op {
                 UnaryOp::Neg => {
-                    let e = self.resolve_expr(&ue.arg, location, phc)?;
+                    let e = self.resolve_expr(&ue.arg, location, phc, within_aggr)?;
                     expr::Expr::func(FuncKind::Neg, vec![e])
                 }
                 UnaryOp::BitInv => {
-                    let e = self.resolve_expr(&ue.arg, location, phc)?;
+                    let e = self.resolve_expr(&ue.arg, location, phc, within_aggr)?;
                     expr::Expr::func(FuncKind::BitInv, vec![e])
                 }
                 UnaryOp::LogicalNot => {
-                    let e = self.resolve_expr(&ue.arg, location, phc)?;
+                    let e = self.resolve_expr(&ue.arg, location, phc, within_aggr)?;
                     let p = Pred::Not(e);
                     expr::Expr::pred(p)
                 }
             },
             Expr::Binary(be) => {
-                let lhs = self.resolve_expr(&be.lhs, location, phc)?;
-                let rhs = self.resolve_expr(&be.rhs, location, phc)?;
+                let lhs = self.resolve_expr(&be.lhs, location, phc, within_aggr)?;
+                let rhs = self.resolve_expr(&be.rhs, location, phc, within_aggr)?;
                 let kind = match be.op {
                     BinaryOp::Add => FuncKind::Add,
                     BinaryOp::Sub => FuncKind::Sub,
@@ -324,8 +339,8 @@ pub trait ExprResolve {
             }
             Expr::Predicate(pred) => match pred.as_ref() {
                 Predicate::Cmp(op, lhs, rhs) => {
-                    let lhs = self.resolve_expr(lhs, location, phc)?;
-                    let rhs = self.resolve_expr(rhs, location, phc)?;
+                    let lhs = self.resolve_expr(lhs, location, phc, within_aggr)?;
+                    let rhs = self.resolve_expr(rhs, location, phc, within_aggr)?;
                     let kind = match op {
                         CompareOp::Equal => PredFuncKind::Equal,
                         CompareOp::Greater => PredFuncKind::Greater,
@@ -338,7 +353,7 @@ pub trait ExprResolve {
                 }
                 Predicate::QuantCmp(..) => todo!(),
                 Predicate::Is(op, arg) => {
-                    let arg = self.resolve_expr(arg, location, phc)?;
+                    let arg = self.resolve_expr(arg, location, phc, within_aggr)?;
                     let kind = match op {
                         IsOp::Null => PredFuncKind::IsNull,
                         IsOp::NotNull => PredFuncKind::IsNotNull,
@@ -350,8 +365,8 @@ pub trait ExprResolve {
                     expr::Expr::pred_func(kind, vec![arg])
                 }
                 Predicate::Match(op, lhs, rhs) => {
-                    let lhs = self.resolve_expr(lhs, location, phc)?;
-                    let rhs = self.resolve_expr(rhs, location, phc)?;
+                    let lhs = self.resolve_expr(lhs, location, phc, within_aggr)?;
+                    let rhs = self.resolve_expr(rhs, location, phc, within_aggr)?;
                     let kind = match op {
                         MatchOp::SafeEqual => PredFuncKind::SafeEqual,
                         MatchOp::Like => PredFuncKind::Like,
@@ -362,45 +377,45 @@ pub trait ExprResolve {
                     expr::Expr::pred_func(kind, vec![lhs, rhs])
                 }
                 Predicate::InValues(lhs, vals) => {
-                    let lhs = self.resolve_expr(lhs, location, phc)?;
+                    let lhs = self.resolve_expr(lhs, location, phc, within_aggr)?;
                     let mut list = Vec::with_capacity(vals.len() + 1);
                     list.push(lhs);
                     for v in vals {
-                        let e = self.resolve_expr(v, location, phc)?;
+                        let e = self.resolve_expr(v, location, phc, within_aggr)?;
                         list.push(e);
                     }
                     expr::Expr::pred_func(PredFuncKind::InValues, list)
                 }
                 Predicate::NotInValues(lhs, vals) => {
-                    let lhs = self.resolve_expr(lhs, location, phc)?;
+                    let lhs = self.resolve_expr(lhs, location, phc, within_aggr)?;
                     let mut list = Vec::with_capacity(vals.len() + 1);
                     list.push(lhs);
                     for v in vals {
-                        let e = self.resolve_expr(v, location, phc)?;
+                        let e = self.resolve_expr(v, location, phc, within_aggr)?;
                         list.push(e);
                     }
                     expr::Expr::pred_func(PredFuncKind::NotInValues, list)
                 }
                 Predicate::InSubquery(lhs, subq) => {
-                    let lhs = self.resolve_expr(lhs, location, phc)?;
+                    let lhs = self.resolve_expr(lhs, location, phc, within_aggr)?;
                     let ph = phc.add_subquery(SubqKind::Table, subq.as_ref(), location);
                     expr::Expr::pred(Pred::InSubquery(lhs, ph))
                 }
                 Predicate::NotInSubquery(lhs, subq) => {
-                    let lhs = self.resolve_expr(lhs, location, phc)?;
+                    let lhs = self.resolve_expr(lhs, location, phc, within_aggr)?;
                     let ph = phc.add_subquery(SubqKind::Table, subq.as_ref(), location);
                     expr::Expr::pred(Pred::NotInSubquery(lhs, ph))
                 }
                 Predicate::Between(lhs, mhs, rhs) => {
-                    let lhs = self.resolve_expr(lhs, location, phc)?;
-                    let mhs = self.resolve_expr(mhs, location, phc)?;
-                    let rhs = self.resolve_expr(rhs, location, phc)?;
+                    let lhs = self.resolve_expr(lhs, location, phc, within_aggr)?;
+                    let mhs = self.resolve_expr(mhs, location, phc, within_aggr)?;
+                    let rhs = self.resolve_expr(rhs, location, phc, within_aggr)?;
                     expr::Expr::pred_func(PredFuncKind::Between, vec![lhs, mhs, rhs])
                 }
                 Predicate::NotBetween(lhs, mhs, rhs) => {
-                    let lhs = self.resolve_expr(lhs, location, phc)?;
-                    let mhs = self.resolve_expr(mhs, location, phc)?;
-                    let rhs = self.resolve_expr(rhs, location, phc)?;
+                    let lhs = self.resolve_expr(lhs, location, phc, within_aggr)?;
+                    let mhs = self.resolve_expr(mhs, location, phc, within_aggr)?;
+                    let rhs = self.resolve_expr(rhs, location, phc, within_aggr)?;
                     expr::Expr::pred_func(PredFuncKind::NotBetween, vec![lhs, mhs, rhs])
                 }
                 Predicate::Exists(subq) => {
@@ -410,7 +425,7 @@ pub trait ExprResolve {
                 Predicate::Conj(cs) => {
                     let mut list = Vec::with_capacity(cs.len());
                     for c in cs {
-                        let e = self.resolve_expr(c, location, phc)?;
+                        let e = self.resolve_expr(c, location, phc, within_aggr)?;
                         list.push(e);
                     }
                     expr::Expr::pred(Pred::Conj(list))
@@ -418,7 +433,7 @@ pub trait ExprResolve {
                 Predicate::Disj(ds) => {
                     let mut list = Vec::with_capacity(ds.len());
                     for d in ds {
-                        let e = self.resolve_expr(d, location, phc)?;
+                        let e = self.resolve_expr(d, location, phc, within_aggr)?;
                         list.push(e);
                     }
                     expr::Expr::pred(Pred::Disj(list))
@@ -426,7 +441,7 @@ pub trait ExprResolve {
                 Predicate::LogicalXor(xs) => {
                     let mut list = Vec::with_capacity(xs.len());
                     for x in xs {
-                        let e = self.resolve_expr(x, location, phc)?;
+                        let e = self.resolve_expr(x, location, phc, within_aggr)?;
                         list.push(e);
                     }
                     expr::Expr::pred(Pred::Xor(list))
@@ -438,17 +453,17 @@ pub trait ExprResolve {
             Expr::Builtin(bi) => match bi {
                 Builtin::Extract(unit, arg) => {
                     let unit = time_unit_from_ast(*unit);
-                    let e = self.resolve_expr(arg, location, phc)?;
+                    let e = self.resolve_expr(arg, location, phc, within_aggr)?;
                     expr::Expr::func(
                         expr::FuncKind::Extract,
                         vec![expr::Expr::Farg(Farg::TimeUnit(unit)), e],
                     )
                 }
                 Builtin::Substring(arg, start, end) => {
-                    let arg = self.resolve_expr(arg, location, phc)?;
-                    let start = self.resolve_expr(start, location, phc)?;
+                    let arg = self.resolve_expr(arg, location, phc, within_aggr)?;
+                    let start = self.resolve_expr(start, location, phc, within_aggr)?;
                     let end = match end {
-                        Some(e) => self.resolve_expr(e, location, phc)?,
+                        Some(e) => self.resolve_expr(e, location, phc, within_aggr)?,
                         None => expr::Expr::farg_none(),
                     };
                     expr::Expr::func(expr::FuncKind::Substring, vec![arg, start, end])
@@ -461,19 +476,19 @@ pub trait ExprResolve {
             }) => {
                 let mut args = Vec::with_capacity(branches.len() * 2 + 2);
                 let node = match operand {
-                    Some(e) => self.resolve_expr(e, location, phc)?,
+                    Some(e) => self.resolve_expr(e, location, phc, within_aggr)?,
                     None => expr::Expr::farg_none(),
                 };
                 args.push(node);
                 let fb = match fallback {
-                    Some(e) => self.resolve_expr(e, location, phc)?,
+                    Some(e) => self.resolve_expr(e, location, phc, within_aggr)?,
                     None => expr::Expr::farg_none(),
                 };
                 args.push(fb);
                 for (when, then) in branches {
-                    let when = self.resolve_expr(when, location, phc)?;
+                    let when = self.resolve_expr(when, location, phc, within_aggr)?;
                     args.push(when);
-                    let then = self.resolve_expr(then, location, phc)?;
+                    let then = self.resolve_expr(then, location, phc, within_aggr)?;
                     args.push(then);
                 }
                 expr::Expr::func(expr::FuncKind::Case, args)
@@ -551,7 +566,7 @@ pub trait ExprResolve {
         location: &'static str,
         phc: &mut PlaceholderCollector<'a>,
     ) -> Result<expr::Expr> {
-        let e = self.resolve_expr(&af.expr, location, phc)?;
+        let e = self.resolve_expr(&af.expr, location, phc, true)?;
         let q = match af.q {
             SetQuantifier::All => expr::Setq::All,
             SetQuantifier::Distinct => expr::Setq::Distinct,


Test patch:
diff --git a/xngin-plan/src/builder/tests.rs b/xngin-plan/src/builder/tests.rs
index 6b77b20..96891e3 100644
--- a/xngin-plan/src/builder/tests.rs
+++ b/xngin-plan/src/builder/tests.rs
@@ -1,4 +1,5 @@
 use super::*;
+use crate::op::OpKind;
 use std::sync::Arc;
 use xngin_catalog::mem_impl::{ColumnSpec, MemCatalogBuilder};
 use xngin_catalog::ColumnAttr;
@@ -6,9 +7,20 @@ use xngin_datatype::DataType;
 use xngin_frontend::parser::dialect::MySQL;
 use xngin_frontend::parser::parse_query;
 
+macro_rules! plan_shape {
+    ( $($ok:ident),* ) => {
+        vec![
+            $(
+                OpKind::$ok
+            ),*
+        ]
+    }
+}
+
 #[test]
 fn test_plan_build_select_row() {
     let cat = empty_catalog();
+    let shape = plan_shape![Row];
     for sql in vec![
         "select 1",
         "select 1, 2, 3",
@@ -17,21 +29,45 @@ fn test_plan_build_select_row() {
     ] {
         let builder = PlanBuilder::new(Arc::clone(&cat), "default").unwrap();
         let (_, qr) = parse_query(MySQL(sql)).unwrap();
-        builder.build_plan(&qr).unwrap();
+        let plan = builder.build_plan(&qr).unwrap();
+        assert_eq!(shape, plan.shape())
     }
 }
 
 #[test]
-fn test_plan_build_expr() {
+fn test_plan_build_row_expr() {
     let cat = tpch_catalog();
+    let shape = plan_shape![Row];
     for sql in vec![
         "select 1, 'abc', 'hello ' 'world', 1e20, 9223372036854775808",
-        "select count(*), count(distinct l_orderkey), sum(l_quantity), max(l_discount), min(l_tax), avg(l_extendedprice) from lineitem",
-        "select -l_orderkey, ~l_orderkey, not l_orderkey = 1 from lineitem",
         "select 1+2, 1-2, 1*2, 1/2",
         "select 1&2, 1|2, 1^2, 1<<2, 1>>2",
         "select true, false, 1 and 2, 1 or 2, 1 xor 2, 1=2, 1>=2, 1>2, 1<=2, 1<2, 1<>2",
         "select null, date '2021-01-01'",
+    ] {
+        let builder = PlanBuilder::new(Arc::clone(&cat), "tpch").unwrap();
+        let (_, qr) = parse_query(MySQL(sql)).unwrap();
+        let plan = builder.build_plan(&qr).unwrap();
+        assert_eq!(shape, plan.shape())
+    }
+}
+
+#[test]
+fn test_plan_build_aggr_expr() {
+    let cat = j_catalog();
+    let sql = "select count(*), count(distinct c0), sum(c0), max(c1), min(c1), avg(c1) from t1";
+    let builder = PlanBuilder::new(Arc::clone(&cat), "j").unwrap();
+    let (_, qr) = parse_query(MySQL(sql)).unwrap();
+    let plan = builder.build_plan(&qr).unwrap();
+    assert_eq!(plan_shape![Aggr, Proj, Table], plan.shape())
+}
+
+#[test]
+fn test_plan_build_expr_from_table() {
+    let cat = tpch_catalog();
+    let shape = plan_shape![Proj, Proj, Table];
+    for sql in vec![
+        "select -l_orderkey, ~l_orderkey, not l_orderkey = 1 from lineitem",
         "select l_tax is null, l_tax is not null, l_tax is true, l_tax is not true, l_tax is false, l_tax is not false from lineitem",
         "select l_shipdate <=> l_commitdate, l_comment like 'A%', l_comment not like 'A%', l_comment regexp '.*', l_comment not regexp '.*' from lineitem",
         "select l_quantity in (1,2,3), l_quantity not in (1,2,3), l_quantity between 1 and 3, l_quantity not between 1 and 3 from lineitem",
@@ -41,64 +77,69 @@ fn test_plan_build_expr() {
     ] {
         let builder = PlanBuilder::new(Arc::clone(&cat), "tpch").unwrap();
         let (_, qr) = parse_query(MySQL(sql)).unwrap();
-        builder.build_plan(&qr).unwrap();
+        let plan = builder.build_plan(&qr).unwrap();
+        assert_eq!(shape, plan.shape())
     }
 }
 
 #[test]
 fn test_plan_build_select_table() {
     let cat = tpch_catalog();
-    for (sql, n_cols) in vec![
-        ("select 1 from lineitem", 1),
-        ("select l_orderkey from lineitem", 1),
-        ("select l_orderkey from tpch.lineitem", 1),
-        ("select lineitem.l_orderkey from lineitem", 1),
-        ("select lineitem.l_orderkey from tpch.lineitem", 1),
-        ("select tpch.lineitem.l_orderkey from lineitem", 1),
-        ("select tpch.lineitem.l_orderkey, tpch.lineitem.l_quantity from lineitem", 2),
-        ("select tpch.lineitem.l_orderkey from lineitem, (select 1) b", 1),
-        ("select n, l_orderkey from lineitem, (select 1 as n) b", 2),
-        ("select l_orderkey, l_partkey from lineitem", 2),
-        ("select l_orderkey, l_partkey from lineitem as li", 2),
-        ("select * from lineitem", 16),
-        ("select lineitem.* from lineitem", 16),
-        ("select tpch.lineitem.* from lineitem", 16),
-        ("select l_quantity, tpch.lineitem.* from lineitem where l_quantity > 10", 17),
-        ("select l_quantity, tpch.lineitem.* from lineitem, (select 1) b where l_quantity > 10", 17),
-        ("select l_orderkey from lineitem where true", 1),
-        ("select l_orderkey from lineitem where 1", 1),
-        ("select l_orderkey from lineitem where l_comment is null", 1),
-        ("select tpch.lineitem.l_orderkey from lineitem, lineitem as l2 where lineitem.l_orderkey = l2.l_orderkey", 1),
-        ("select lineitem.l_orderkey from lineitem, lineitem as l2 where lineitem.l_orderkey = l2.l_orderkey", 1),
-        ("select l_orderkey from lineitem group by l_orderkey", 1),
-        ("select l_orderkey, count(*) from lineitem group by l_orderkey", 2),
-        ("select count(*), l_orderkey from lineitem group by l_orderkey", 2),
-        ("select l_orderkey from lineitem group by l_orderkey having count(*) > 1", 1),
-        ("select l_orderkey from lineitem having l_quantity > 10", 1),
-        ("select count(*) from lineitem having sum(1) > 0", 1),
-        ("select l_orderkey, count(*) from lineitem group by l_orderkey having l_orderkey > 0", 2),
-        ("select l_orderkey from lineitem where l_tax > 0 having l_quantity < 10", 1),
-        ("select l_orderkey k, count(*) from lineitem group by l_orderkey", 2),
-        ("select l_orderkey from lineitem order by l_orderkey", 1),
-        ("select l_orderkey from lineitem order by lineitem.l_orderkey", 1),
-        ("select l_orderkey from lineitem order by tpch.lineitem.l_orderkey", 1),
-        ("select l_orderkey from lineitem order by l_linenumber", 1),
-        ("select l_orderkey from lineitem order by l_linenumber desc", 1),
-        ("select l_orderkey k from lineitem order by k", 1),
-        ("select l_orderkey k, sum(1) from lineitem group by l_orderkey order by k", 2),
-        ("select l_orderkey k, sum(1) from lineitem group by k order by k", 2),
-        ("select l_orderkey k, sum(1) from lineitem group by l_orderkey order by l_orderkey", 2),
-        ("select l_orderkey, count(*) from lineitem group by l_orderkey order by l_orderkey", 2),
-        ("select l_orderkey, sum(l_tax) from lineitem group by l_orderkey order by sum(l_tax) desc, l_orderkey", 2),
-        ("select count(*) from lineitem order by count(*)", 1),
-        ("select l_orderkey from lineitem limit 10", 1),
-        ("select l_orderkey from lineitem limit 10 offset 10", 1),
-        ("select l_orderkey from lineitem order by l_quantity desc limit 10", 1),
-        ("select distinct l_orderkey from lineitem", 1),
-        ("select distinct count(*) from lineitem", 1),
-        ("select distinct l_orderkey from lineitem group by l_orderkey", 1),
-        ("select * from (select l_orderkey from lineitem order by l_orderkey limit 1) a, (select l_orderkey from lineitem order by l_orderkey desc limit 1) b", 2),
-        ("select a.* from (select l_orderkey, l_linenumber from lineitem) a, (select l_orderkey from lineitem limit 1) b", 2),
+    for (sql, n_cols, shape) in vec![
+        ("select 1 from lineitem", 1, plan_shape![Proj, Proj, Table]),
+        ("select l_orderkey from lineitem", 1, plan_shape![Proj, Proj, Table]),
+        ("select l_orderkey from tpch.lineitem", 1, plan_shape![Proj, Proj, Table]),
+        ("select lineitem.l_orderkey from lineitem", 1, plan_shape![Proj, Proj, Table]),
+        ("select lineitem.l_orderkey from tpch.lineitem", 1, plan_shape![Proj, Proj, Table]),
+        ("select tpch.lineitem.l_orderkey from lineitem", 1, plan_shape![Proj, Proj, Table]),
+        ("select tpch.lineitem.l_orderkey, tpch.lineitem.l_quantity from lineitem", 2, plan_shape![Proj, Proj, Table]),
+        ("select tpch.lineitem.l_orderkey from lineitem, (select 1) b", 1, plan_shape![Proj, Join, Proj, Table, Row]),
+        ("select n, l_orderkey from lineitem, (select 1 as n) b", 2, plan_shape![Proj, Join, Proj, Table, Row]),
+        ("select l_orderkey, l_partkey from lineitem", 2, plan_shape![Proj, Proj, Table]),
+        ("select l_orderkey, l_partkey from lineitem as li", 2, plan_shape![Proj, Proj, Table]),
+        ("select * from lineitem", 16, plan_shape![Proj, Proj, Table]),
+        ("select lineitem.* from lineitem", 16, plan_shape![Proj, Proj, Table]),
+        ("select tpch.lineitem.* from lineitem", 16, plan_shape![Proj, Proj, Table]),
+        ("select l_quantity, tpch.lineitem.* from lineitem where l_quantity > 10", 17, plan_shape![Proj, Filt, Proj, Table]),
+        ("select l_quantity, tpch.lineitem.* from lineitem, (select 1) b where l_quantity > 10", 17, plan_shape![Proj, Filt, Join, Proj, Table, Row]),
+        ("select l_orderkey from lineitem where true", 1, plan_shape![Proj, Filt, Proj, Table]),
+        ("select l_orderkey from lineitem where 1", 1, plan_shape![Proj, Filt, Proj, Table]),
+        ("select l_orderkey from lineitem where l_comment is null", 1, plan_shape![Proj, Filt, Proj, Table]),
+        ("select tpch.lineitem.l_orderkey from lineitem, lineitem as l2 where lineitem.l_orderkey = l2.l_orderkey", 1, plan_shape![Proj, Filt, Join, Proj, Table, Proj, Table]),
+        ("select lineitem.l_orderkey from lineitem, lineitem as l2 where lineitem.l_orderkey = l2.l_orderkey", 1, plan_shape![Proj, Filt, Join, Proj, Table, Proj, Table]),
+        ("select l_orderkey from lineitem group by l_orderkey", 1, plan_shape![Aggr, Proj, Table]),
+        ("select l_orderkey, count(*) from lineitem group by l_orderkey", 2, plan_shape![Aggr, Proj, Table]),
+        ("select count(*), l_orderkey from lineitem group by l_orderkey", 2, plan_shape![Aggr, Proj, Table]),
+        ("select l_orderkey, count(*) from lineitem where l_tax > 0.1 group by l_orderkey", 2, plan_shape![Aggr, Filt, Proj, Table]),
+        ("select l_orderkey from lineitem group by l_orderkey having count(*) > 1", 1, plan_shape![Aggr, Proj, Table]),
+        ("select l_orderkey from lineitem group by l_orderkey having sum(l_extendedprice * l_tax) > 1", 1, plan_shape![Aggr, Proj, Table]),
+        ("select l_orderkey from lineitem where l_shipdate is null group by l_orderkey having count(*) > 1", 1, plan_shape![Aggr, Filt, Proj, Table]),
+        ("select l_orderkey from lineitem having l_orderkey is not null", 1, plan_shape![Proj, Filt, Proj, Table]),
+        ("select count(*) from lineitem having sum(1) > 0", 1, plan_shape![Aggr, Proj, Table]),
+        ("select l_orderkey, count(*) from lineitem group by l_orderkey having l_orderkey > 0", 2, plan_shape![Aggr, Proj, Table]),
+        ("select l_orderkey from lineitem where l_tax > 0 having l_orderkey < 10", 1, plan_shape![Proj, Filt, Proj, Table]),
+        ("select l_orderkey k, count(*) from lineitem group by l_orderkey", 2, plan_shape![Aggr, Proj, Table]),
+        ("select l_orderkey from lineitem order by l_orderkey", 1, plan_shape![Sort, Proj, Proj, Table]),
+        ("select l_orderkey from lineitem order by lineitem.l_orderkey", 1, plan_shape![Sort, Proj, Proj, Table]),
+        ("select l_orderkey from lineitem order by tpch.lineitem.l_orderkey", 1, plan_shape![Sort, Proj, Proj, Table]),
+        ("select l_orderkey from lineitem order by l_linenumber", 1, plan_shape![Sort, Proj, Proj, Table]),
+        ("select l_orderkey from lineitem order by l_linenumber desc", 1, plan_shape![Sort, Proj, Proj, Table]),
+        ("select l_orderkey k from lineitem order by k", 1, plan_shape![Sort, Proj, Proj, Table]),
+        ("select l_orderkey k, sum(1) from lineitem group by l_orderkey order by k", 2, plan_shape![Sort, Aggr, Proj, Table]),
+        ("select l_orderkey k, sum(1) from lineitem group by k order by k", 2, plan_shape![Sort, Aggr, Proj, Table]),
+        ("select l_orderkey k, sum(1) from lineitem group by l_orderkey order by l_orderkey", 2, plan_shape![Sort, Aggr, Proj, Table]),
+        ("select l_orderkey, count(*) from lineitem group by l_orderkey order by l_orderkey", 2, plan_shape![Sort, Aggr, Proj, Table]),
+        ("select l_orderkey, sum(l_tax) from lineitem group by l_orderkey order by sum(l_tax) desc, l_orderkey", 2, plan_shape![Sort, Aggr, Proj, Table]),
+        ("select count(*) from lineitem order by count(*)", 1, plan_shape![Sort, Aggr, Proj, Table]),
+        ("select l_orderkey from lineitem limit 10", 1, plan_shape![Limit, Proj, Proj, Table]),
+        ("select l_orderkey from lineitem limit 10 offset 10", 1, plan_shape![Limit, Proj, Proj, Table]),
+        ("select l_orderkey from lineitem order by l_quantity desc limit 10", 1, plan_shape![Limit, Sort, Proj, Proj, Table]),
+        ("select distinct l_orderkey from lineitem", 1, plan_shape![Aggr, Proj, Table]),
+        ("select distinct l_orderkey from lineitem limit 1", 1, plan_shape![Limit, Aggr, Proj, Table]),
+        ("select distinct count(*) from lineitem", 1, plan_shape![Aggr, Proj, Table]),
+        ("select distinct l_orderkey from lineitem group by l_orderkey", 1, plan_shape![Aggr, Proj, Table]),
+        ("select * from (select l_orderkey from lineitem order by l_orderkey limit 1) a, (select l_orderkey from lineitem order by l_orderkey desc limit 1) b", 2, plan_shape![Proj, Join, Limit, Sort, Proj, Proj, Table, Limit, Sort, Proj, Proj, Table]),
+        ("select a.* from (select l_orderkey, l_linenumber from lineitem) a, (select l_orderkey from lineitem limit 1) b", 2, plan_shape![Proj, Join, Proj, Proj, Table, Limit, Proj, Proj, Table]),
     ] {
         let builder = PlanBuilder::new(Arc::clone(&cat), "tpch").unwrap();
         let (_, qr) = parse_query(MySQL(sql)).unwrap();
@@ -109,63 +150,126 @@ fn test_plan_build_select_table() {
                 panic!("{:?}", e)
             }
         };
+        assert_eq!(shape, plan.shape());
         let p = plan.queries.get(&plan.root).unwrap();
         assert_eq!(n_cols, p.scope.out_cols.len());
+        print_plan(&plan)
     }
 }
 
 #[test]
 fn test_plan_build_join() {
-    let cat = tpch_catalog();
-    for sql in vec![
-        "select * from lineitem l1 cross join lineitem l2",
-        "select l1.* from lineitem l1 cross join lineitem l2",
-        "select l1.l_orderkey from lineitem l1 cross join lineitem l2",
-        "select tpch.l1.* from lineitem l1 cross join lineitem l2",
-        "select tpch.l1.l_orderkey from lineitem l1 cross join lineitem l2",
-        "select l1.* from lineitem l1 join lineitem l2",
-        "select l1.l_orderkey from lineitem l1 join lineitem l2 on l1.l_orderkey = l2.l_orderkey",
-        "select l1.l_orderkey from lineitem l1 left join lineitem l2 on l1.l_orderkey = l2.l_orderkey",
-        "select l1.l_orderkey from lineitem l1 right join lineitem l2 on l1.l_orderkey = l2.l_orderkey",
-        "select l1.l_orderkey from lineitem l1 full join lineitem l2 on l1.l_orderkey = l2.l_orderkey",
-        "select l1.l_orderkey from lineitem l1 full join lineitem l2 on l1.l_orderkey = l2.l_orderkey and l1.l_quantity > 10",
-        "select l1.l_orderkey from lineitem l1 join lineitem l2 on l1.l_orderkey = l2.l_orderkey, lineitem l3",
-        "select l1.l_orderkey from lineitem l1, lineitem l2 join lineitem l3 on l2.l_orderkey = l3.l_orderkey",
-        "select l1.l_orderkey from lineitem l1 join lineitem l2 using (l_orderkey)",
-        "select l1.l_orderkey from lineitem l1 join lineitem l2 using (l_orderkey, l_linenumber)",
+    let cat = j_catalog();
+    for (sql, shape) in vec![
+        (
+            "select * from t0 cross join t1",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.* from t1 cross join t2 x2",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 cross join t2",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select j.l1.* from t1 l1 cross join t2 l2",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select j.l1.c0 from t1 l1 cross join t2 l2",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.* from t1 join t2",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1, t2, t3",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 join t2 join t3",
+            plan_shape![Proj, Join, Join, Proj, Table, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0, t2.c0 from t1 join t2 on t1.c0 = t2.c0",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 left join t2 on t1.c1 = t2.c1",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 right join t2 on t1.c1 = t2.c1",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 full join t2 on t1.c1 = t2.c1",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 full join t2 on t1.c1 = t2.c1 and t1.c0 > 10",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 join t2 on t1.c1 = t2.c1, t3",
+            plan_shape![Proj, Join, Join, Proj, Table, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1, t2 join t3 on t2.c2 = t3.c3",
+            plan_shape![Proj, Join, Proj, Table, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 join t2 using (c1)",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
+        (
+            "select t1.c0 from t1 join t2 using (c0, c1)",
+            plan_shape![Proj, Join, Proj, Table, Proj, Table],
+        ),
     ] {
-        let builder = PlanBuilder::new(Arc::clone(&cat), "tpch").unwrap();
+        let builder = PlanBuilder::new(Arc::clone(&cat), "j").unwrap();
         let (_, qr) = parse_query(MySQL(sql)).unwrap();
-        if let Err(e) = builder.build_plan(&qr) {
-            eprintln!("sql={}", sql);
-            panic!("{:?}", e)
-        }
+        let plan = match builder.build_plan(&qr) {
+            Err(e) => {
+                eprintln!("sql={}", sql);
+                panic!("{:?}", e)
+            }
+            Ok(plan) => plan,
+        };
+        assert_eq!(shape, plan.shape());
+        print_plan(&plan)
     }
 }
 
 #[test]
 fn test_plan_build_subquery() {
-    let cat = tpch_catalog();
+    let cat = j_catalog();
     for sql in vec![
-        "select * from lineitem where l_orderkey in (select l_orderkey from lineitem)",
-        "select * from lineitem where l_orderkey not in (select l_orderkey from lineitem)",
-        "select * from lineitem where not l_orderkey in (select l_orderkey from lineitem)",
-        "select * from lineitem where l_orderkey in (select l_orderkey from lineitem l2 where l2.l_linenumber = lineitem.l_linenumber)",
-        "select * from lineitem l1 where l_orderkey not in (select l_orderkey from lineitem l2 where l2.l_linenumber = l1.l_linenumber)",
-        "select * from lineitem where not l_orderkey in (select l_orderkey from lineitem l2 where l2.l_linenumber = tpch.lineitem.l_linenumber)",
-        "select * from lineitem where exists (select l_orderkey from lineitem)",
-        "select * from lineitem where not exists (select 1 from lineitem)",
-        "select * from lineitem where exists (select 1 from lineitem l2 where l2.l_orderkey = lineitem.l_orderkey)",
-        "select * from lineitem where not exists (select 1 from lineitem l2 where l2.l_orderkey = lineitem.l_orderkey)",
-        "select * from lineitem where exists (select 1 from lineitem l2 where exists (select 1 from lineitem))",
-        "select * from lineitem where exists (select 1 from lineitem l2 where exists (select 1 from lineitem l3 where l3.l_orderkey = l2.l_orderkey))",
-        "select * from lineitem where exists (select 1 from lineitem l2 where exists (select 1 from lineitem l3 where l3.l_orderkey = lineitem.l_orderkey))",
-        "select * from lineitem l1 where exists (select 1 from lineitem l2 where exists (select 1 from lineitem l3 where l3.l_orderkey = l1.l_orderkey))",
-        "select * from lineitem l1 where exists (select 1 from lineitem l2, (select l_orderkey from lineitem l3 where l3.l_orderkey = l1.l_orderkey) tt where tt.l_orderkey = l2.l_orderkey)",
-        "select * from lineitem where exists (with tmp as (select 1 from lineitem l2 where l2.l_orderkey = lineitem.l_orderkey) select * from tmp)",
-        "with cte as (select 1 as c0) select * from cte where exists (select 1 from lineitem where lineitem.l_orderkey = c0)",
+        "select * from t0 where c0 in (select c0 from t1)",
+        "select * from t0 where c0 not in (select c0 from t1)",
+        "select * from t0 where not c0 in (select c0 from t1)",
+        "select * from t1 where c0 in (select c0 from t2 where t2.c1 = t1.c1)",
+        "select * from t1 x where c0 in (select c0 from t2 where t2.c1 = x.c1)",
+        "select * from t1 where c0 not in (select c0 from t2 tx where tx.c1 = t1.c1)",
+        "select * from t1 where not c0 in (select c0 from t2 where t2.c1 = j.t1.c1)",
+        "select * from t1 where exists (select c1 from t2)",
+        "select * from t1 where not exists (select 1 from t2)",
+        "select * from t1 where exists (select 1 from t2 where t2.c0 = t1.c0)",
+        "select * from t1 where not exists (select 1 from t2 x2 where x2.c1 = t1.c1)",
+        "select * from t1 where exists (select 1 from t2 where exists (select 1 from t1))",
+        "select * from t1 where exists (select 1 from t2 x2 where exists (select 1 from t3 where t3.c0 = x2.c0))",
+        "select * from t3 where exists (select 1 from t2 where exists (select 1 from t1 where t1.c0 = t3.c3))",
+        "select * from t3 where exists (select 1 from t2 where exists (select 1 from t1 where t1.c0 = c3))",
+        "select * from t3 where exists (select 1 from t2 where exists (select 1 from t1 where t1.c0 = c2))",
+        "select * from t3 where exists (select 1 from t2 where exists (select 1 from t1 where c0 = t3.c0))",
+        "select * from t3 where exists (select 1 from t2, (select c0 from t1 where t1.c1 = t3.c1) tt where tt.c0 = t2.c0)",
+        "select * from t3 where exists (with tmp as (select 1 from t2 where t2.c0 = t3.c0) select * from tmp)",
+        "with cte as (select 1 as cx) select * from cte where exists (select 1 from t1 where t1.c0 = cx)",
     ] {
-        let builder = PlanBuilder::new(Arc::clone(&cat), "tpch").unwrap();
+        let builder = PlanBuilder::new(Arc::clone(&cat), "j").unwrap();
         let (_, qr) = parse_query(MySQL(sql)).unwrap();
         if let Err(e) = builder.build_plan(&qr) {
             eprintln!("sql={}", sql);
@@ -174,57 +278,137 @@ fn test_plan_build_subquery() {
     }
 }
 
+fn j_catalog() -> Arc<dyn QueryCatalog> {
+    let mut builder = MemCatalogBuilder::default();
+    builder.add_schema("j").unwrap();
+    builder
+        .add_table(
+            "j",
+            "t0",
+            &vec![ColumnSpec::new("c0", DataType::I32, ColumnAttr::empty())],
+        )
+        .unwrap();
+    builder
+        .add_table(
+            "j",
+            "t1",
+            &vec![
+                ColumnSpec::new("c0", DataType::I32, ColumnAttr::empty()),
+                ColumnSpec::new("c1", DataType::I32, ColumnAttr::empty()),
+            ],
+        )
+        .unwrap();
+    builder
+        .add_table(
+            "j",
+            "t2",
+            &vec![
+                ColumnSpec::new("c0", DataType::I32, ColumnAttr::empty()),
+                ColumnSpec::new("c1", DataType::I32, ColumnAttr::empty()),
+                ColumnSpec::new("c2", DataType::I32, ColumnAttr::empty()),
+            ],
+        )
+        .unwrap();
+    builder
+        .add_table(
+            "j",
+            "t3",
+            &vec![
+                ColumnSpec::new("c0", DataType::I32, ColumnAttr::empty()),
+                ColumnSpec::new("c1", DataType::I32, ColumnAttr::empty()),
+                ColumnSpec::new("c2", DataType::I32, ColumnAttr::empty()),
+                ColumnSpec::new("c3", DataType::I32, ColumnAttr::empty()),
+            ],
+        )
+        .unwrap();
+    let cat = builder.build();
+    Arc::new(cat)
+}
+
 #[test]
 fn test_plan_build_setop() {
-    use crate::explain::Explain;
     let cat = tpch_catalog();
-    for sql in vec![
-        "select * from lineitem union select * from lineitem",
-        "select * from lineitem union all select * from lineitem",
-        "select 1 union select 2",
-        "select l_orderkey from lineitem union select 1",
-        "select 1 union select l_orderkey from lineitem",
-        "select 1 union select 2 union select 3",
-        "select l_orderkey from lineitem union select 1 union select l_orderkey from lineitem",
-        "select 1 union select l_orderkey from lineitem union select 1",
-        "select l_orderkey from lineitem except select l_linenumber from lineitem",
-        "select l_orderkey from lineitem except all select l_linenumber from lineitem",
-        "select l_orderkey from lineitem intersect select l_linenumber from lineitem",
-        "select l_orderkey from lineitem intersect all select l_linenumber from lineitem",
+    for (sql, shape) in vec![
+        (
+            "select * from lineitem union select * from lineitem",
+            plan_shape![Setop, Proj, Proj, Table, Proj, Proj, Table],
+        ),
+        (
+            "select * from lineitem union all select * from lineitem",
+            plan_shape![Setop, Proj, Proj, Table, Proj, Proj, Table],
+        ),
+        ("select 1 union select 2", plan_shape![Setop, Row, Row]),
+        (
+            "select l_orderkey from lineitem union select 1",
+            plan_shape![Setop, Proj, Proj, Table, Row],
+        ),
+        (
+            "select 1 union select l_orderkey from lineitem",
+            plan_shape![Setop, Row, Proj, Proj, Table],
+        ),
+        (
+            "select 1 union select 2 union select 3",
+            plan_shape![Setop, Setop, Row, Row, Row],
+        ),
+        (
+            "select l_orderkey from lineitem union select 1 union select l_orderkey from lineitem",
+            plan_shape![Setop, Setop, Proj, Proj, Table, Row, Proj, Proj, Table],
+        ),
+        (
+            "select 1 union select l_orderkey from lineitem union select 1",
+            plan_shape![Setop, Setop, Row, Proj, Proj, Table, Row],
+        ),
+        (
+            "select l_orderkey from lineitem except select l_linenumber from lineitem",
+            plan_shape![Setop, Proj, Proj, Table, Proj, Proj, Table],
+        ),
+        (
+            "select l_orderkey from lineitem except all select l_linenumber from lineitem",
+            plan_shape![Setop, Proj, Proj, Table, Proj, Proj, Table],
+        ),
+        (
+            "select l_orderkey from lineitem intersect select l_linenumber from lineitem",
+            plan_shape![Setop, Proj, Proj, Table, Proj, Proj, Table],
+        ),
+        (
+            "select l_orderkey from lineitem intersect all select l_linenumber from lineitem",
+            plan_shape![Setop, Proj, Proj, Table, Proj, Proj, Table],
+        ),
     ] {
         let builder = PlanBuilder::new(Arc::clone(&cat), "tpch").unwrap();
         let (_, qr) = parse_query(MySQL(sql)).unwrap();
         let plan = builder.build_plan(&qr).unwrap();
-        let mut s = String::new();
-        plan.explain(&mut s).unwrap();
-        println!("Explain plan:\n{}", s)
+        assert_eq!(shape, plan.shape());
+        print_plan(&plan)
     }
 }
 
 #[test]
 fn test_plan_build_with() {
     let cat = tpch_catalog();
-    for (sql, n_cols) in vec![
-        ("with a as (select 1) select * from a", 1),
-        ("with a as (select 1) select 2", 1),
-        ("with a as (select 1, 2) select * from a", 2),
-        ("with a as (select count(*) from lineitem) select `count(*)` from a", 1),
-        ("with a as (select count(*) c from lineitem) select c from a", 1),
-        ("with a (c) as (select count(*) from lineitem) select a.c from a", 1),
-        ("with a (x, y) as (select l_orderkey, count(*) from lineitem group by l_orderkey) select x, y from a", 2),
-        ("with a as (select 1), b as (select 2) select * from a, b", 2),
-        ("with a as (select 1) select * from lineitem, a", 17),
-        ("with a as (select 1) select lineitem.* from lineitem, a", 16),
-        ("with a as (select 1) select tpch.lineitem.* from lineitem, a", 16),
-        ("with a as (select 1) select l_orderkey from lineitem, a", 1),
-        ("with a as (select l_orderkey, count(*) cnt from (select * from lineitem) li group by l_orderkey) select l_orderkey from a where cnt > 10", 1),
-        ("with a as (select l_orderkey, n from lineitem, (select 1 as n) b) select * from a", 2),
+    for (sql, n_cols, shape) in vec![
+        ("with a as (select 1) select * from a", 1, plan_shape![Proj, Row]),
+        ("with a as (select 1) select 2", 1, plan_shape![Row]),
+        ("with a as (select 1, 2) select * from a", 2, plan_shape![Proj, Row]),
+        ("with a as (select count(*) from lineitem) select `count(*)` from a", 1, plan_shape![Proj, Aggr, Proj, Table]),
+        ("with a as (select count(*) c from lineitem) select c from a", 1, plan_shape![Proj, Aggr, Proj, Table]),
+        ("with a (c) as (select count(*) from lineitem) select a.c from a", 1, plan_shape![Proj, Aggr, Proj, Table]),
+        ("with a (x, y) as (select l_orderkey, count(*) from lineitem group by l_orderkey) select x, y from a", 2, plan_shape![Proj, Aggr, Proj, Table]),
+        ("with a as (select 1), b as (select 2) select * from a, b", 2, plan_shape![Proj, Join, Row, Row]),
+        ("with a as (select 1) select * from lineitem, a", 17, plan_shape![Proj, Join, Proj, Table, Row]),
+        ("with a as (select 1) select lineitem.* from lineitem, a", 16, plan_shape![Proj, Join, Proj, Table, Row]),
+        ("with a as (select 1) select tpch.lineitem.* from lineitem, a", 16, plan_shape![Proj, Join, Proj, Table, Row]),
+        ("with a as (select 1) select l_orderkey from lineitem, a", 1, plan_shape![Proj, Join, Proj, Table, Row]),
+        ("with a as (select l_orderkey, count(*) cnt from (select * from lineitem) li group by l_orderkey) select l_orderkey from a where cnt > 10", 1, plan_shape![Proj, Filt, Aggr, Proj, Proj, Table]),
+        ("with a as (select l_orderkey, n from lineitem, (select 1 as n) b) select * from a", 2, plan_shape![Proj, Proj, Join, Proj, Table, Row]),
     ] {
         let builder = PlanBuilder::new(Arc::clone(&cat), "tpch").unwrap();
         let (_, qr) = parse_query(MySQL(sql)).unwrap();
         let plan = builder.build_plan(&qr).unwrap();
+        assert_eq!(shape, plan.shape());
         let p = plan.queries.get(&plan.root).unwrap();
         assert_eq!(n_cols, p.scope.out_cols.len());
+        print_plan(&plan)
     }
 }
 
@@ -267,3 +451,10 @@ pub(crate) fn tpch_catalog() -> Arc<dyn QueryCatalog> {
     let cat = builder.build();
     Arc::new(cat)
 }
+
+fn print_plan(plan: &QueryPlan) {
+    use crate::explain::Explain;
+    let mut s = String::new();
+    plan.explain(&mut s).unwrap();
+    println!("Explain plan:\n{}", s)
+}