Repository: alibaba/GraphScope (language: python)

Issue:
[BUG] Incorrect Logical Plan Generated for Join of Multiple Matches in IR Core
**Describe the bug**
When attempting to perform a join of multiple matches using the IR Core FFI API, the logical plan generated is unexpected.

**To Reproduce**
Run the following java JNI codes to invoke FFI API provided by IR Core:
```java
Pointer ptrPlan = LIB.initLogicalPlan();

// append 'V()' operator
IntByReference oprScan = new IntByReference();
Pointer scan = LIB.initScanOperator(FfiScanOpt.Entity);
LIB.appendScanOperator(ptrPlan, scan, -1, oprScan);

// build sentence: as('a').out().as('b')
Pointer expand1 = LIB.initEdgexpdOperator(FfiExpandOpt.Vertex, FfiDirection.Out);
Pointer sentence1 = LIB.initPatternSentence(FfiJoinKind.Inner);
LIB.setSentenceStart(sentence1, ArgUtils.asNameOrId('a'));
LIB.addSentenceBinder(sentence1, expand1, FfiBinderOpt.Edge);
LIB.setSentenceEnd(sentence1, ArgUtils.asNameOrId('b'));

// add sentence to left match
IntByReference oprMatch1 = new IntByReference();
Pointer match1 = LIB.initPatternOperator();
LIB.addPatternSentence(match1, sentence1);
LIB.appendPatternOperator(ptrPlan, match1, oprScan.getValue(), oprMatch1);

// build sentence: as('c').out().as('b')
Pointer expand2 = LIB.initEdgexpdOperator(FfiExpandOpt.Vertex, FfiDirection.Out);
Pointer sentence2 = LIB.initPatternSentence(FfiJoinKind.Inner);
LIB.setSentenceStart(sentence2, ArgUtils.asNameOrId('c'));
LIB.addSentenceBinder(sentence2, expand2, FfiBinderOpt.Edge);
LIB.setSentenceEnd(sentence2, ArgUtils.asNameOrId('b'));

// add sentence to right match
IntByReference oprMatch2 = new IntByReference();
Pointer match2 = LIB.initPatternOperator();
LIB.addPatternSentence(match2, sentence2);
LIB.appendPatternOperator(ptrPlan, match2, oprScan.getValue(), oprMatch2);

// append join operator which left is match1, right is match2
IntByReference oprJoin = new IntByReference();
Pointer join = LIB.initJoinOperator(FfiJoinKind.Inner);
LIB.addJoinKeyPair(join, ArgUtils.asVar("b", ""), ArgUtils.asVar("b", ""));
LIB.appendJoinOperator(ptrPlan, join, oprMatch1.getValue(), oprMatch2.getValue(), oprJoin);

// append sink operator
LIB.appendSinkOperator(ptrPlan, LIB.initSinkOperator(), oprJoin.getValue(), new IntByReference());

System.out.println(LIB.printPlanAsJson(ptrPlan));
LIB.destroyLogicalPlan(ptrPlan);
```

**Screenshots**
match2 seems to be dropped:
```
FfiResult{code=Success, msg='{
  "nodes": [
    {
      "opr": {
        "opr": {
          "Scan": {
            "scan_opt": 0,
            "alias": {
              "item": {
                "Id": 97
              }
            },
            "params": {
              "tables": [],
              "columns": [],
              "is_all_columns": false,
              "limit": null,
              "predicate": null,
              "sample_ratio": 1.0,
              "extra": {}
            },
            "idx_predicate": null,
            "meta_data": null
          }
        }
      },
      "children": [
        1
      ]
    },
    {
      "opr": {
        "opr": {
          "Edge": {
            "v_tag": {
              "item": {
                "Id": 97
              }
            },
            "direction": 0,
            "params": {
              "tables": [],
              "columns": [],
              "is_all_columns": false,
              "limit": null,
              "predicate": null,
              "sample_ratio": 1.0,
              "extra": {}
            },
            "alias": null,
            "expand_opt": 1,
            "meta_data": null
          }
        }
      },
      "children": [
        2
      ]
    },
    {
      "opr": {
        "opr": {
          "Vertex": {
            "tag": null,
            "opt": 1,
            "params": null,
            "alias": {
              "item": {
                "Id": 98
              }
            },
            "meta_data": null
          }
        }
      },
      "children": []
    }
  ],
  "roots": []
}'}
```



Discussion / hints:
@longbinlai you can checkout to branch `ir_match_debug` and some rust example (test) is provided:
1. cd ~/interactive_engine/executor/ir
2. run the following tests:
`cargo test tests::test_match_ffi -- --nocapture`
`cargo test tests::test_match_join_match_logical -- --nocapture`
`cargo test tests::test_match_join_match_physical -- --nocapture`


Fix patch:
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/FfiPhysicalBuilder.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/FfiPhysicalBuilder.java
index 15824ae21c22..1673b6f80208 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/FfiPhysicalBuilder.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/FfiPhysicalBuilder.java
@@ -64,7 +64,6 @@ public FfiPhysicalBuilder(Configs graphConfig, IrMeta irMeta, LogicalPlan logica
         this.irMeta = irMeta;
         checkFfiResult(LIB.setSchema(irMeta.getSchema().schemaJson()));
         this.ptrPlan = LIB.initLogicalPlan();
-        this.lastIdx = -1;
         initialize();
     }
 
diff --git a/interactive_engine/executor/ir/clients/rust/client/src/physical_builder.rs b/interactive_engine/executor/ir/clients/rust/client/src/physical_builder.rs
index 2d95a6808992..345808c2ed65 100644
--- a/interactive_engine/executor/ir/clients/rust/client/src/physical_builder.rs
+++ b/interactive_engine/executor/ir/clients/rust/client/src/physical_builder.rs
@@ -52,7 +52,7 @@ impl DerefMut for PlanBuilder {
 
 impl PlanBuilder {
     pub fn add_dummy_source(&mut self) -> &mut Self {
-        let op = pb::physical_opr::operator::OpKind::Root(pb::RootScan {});
+        let op = pb::physical_opr::operator::OpKind::Root(pb::Root {});
         self.plan.push(op.into());
         self
     }
diff --git a/interactive_engine/executor/ir/common/src/utils.rs b/interactive_engine/executor/ir/common/src/utils.rs
index 5fc2d24e3824..c01a59a43dc2 100644
--- a/interactive_engine/executor/ir/common/src/utils.rs
+++ b/interactive_engine/executor/ir/common/src/utils.rs
@@ -479,6 +479,12 @@ impl From<pb::Scan> for pb::logical_plan::Operator {
     }
 }
 
+impl From<pb::Root> for pb::logical_plan::Operator {
+    fn from(opr: pb::Root) -> Self {
+        pb::logical_plan::Operator { opr: Some(pb::logical_plan::operator::Opr::Root(opr)) }
+    }
+}
+
 impl From<pb::logical_plan::Operator> for Option<pb::Scan> {
     fn from(opr: pb::logical_plan::Operator) -> Self {
         if let Some(opr) = opr.opr {
@@ -595,27 +601,6 @@ impl From<(pb::EdgeExpand, pb::GetV)> for pb::path_expand::ExpandBase {
     }
 }
 
-impl pb::logical_plan::Operator {
-    pub fn is_whole_graph(&self) -> bool {
-        if let Some(opr) = &self.opr {
-            match opr {
-                pb::logical_plan::operator::Opr::Scan(scan) => {
-                    scan.idx_predicate.is_none()
-                        && scan.alias.is_none()
-                        && scan
-                            .params
-                            .as_ref()
-                            .map(|params| !params.is_queryable())
-                            .unwrap_or(true)
-                }
-                _ => false,
-            }
-        } else {
-            false
-        }
-    }
-}
-
 impl pb::QueryParams {
     // is_queryable doesn't consider tables as we assume that the table info can be inferred directly from current data.
     pub fn is_queryable(&self) -> bool {
diff --git a/interactive_engine/executor/ir/core/src/plan/logical.rs b/interactive_engine/executor/ir/core/src/plan/logical.rs
index d4b439d34cbe..1c20a45c864f 100644
--- a/interactive_engine/executor/ir/core/src/plan/logical.rs
+++ b/interactive_engine/executor/ir/core/src/plan/logical.rs
@@ -23,7 +23,7 @@ use ir_common::error::ParsePbError;
 use ir_common::generated::algebra as pb;
 use ir_common::generated::algebra::pattern::binder::Item;
 use ir_common::generated::common as common_pb;
-use ir_common::{KeyId, NameOrId};
+use ir_common::{KeyId, LabelId, NameOrId};
 use vec_map::VecMap;
 
 use crate::error::{IrError, IrResult};
@@ -64,6 +64,20 @@ impl Node {
     pub fn add_parent(&mut self, parent_id: NodeId) {
         self.parents.insert(parent_id);
     }
+
+    pub fn is_root(&self) -> bool {
+        match self.opr.opr.as_ref() {
+            Some(pb::logical_plan::operator::Opr::Root(_dummy)) => true,
+            _ => false,
+        }
+    }
+
+    pub fn is_scan(&self) -> bool {
+        match self.opr.opr.as_ref() {
+            Some(pb::logical_plan::operator::Opr::Scan(_scan)) => true,
+            _ => false,
+        }
+    }
 }
 
 pub(crate) type NodeType = Rc<RefCell<Node>>;
@@ -71,7 +85,7 @@ pub(crate) type NodeType = Rc<RefCell<Node>>;
 /// An internal representation of the pb-[`LogicalPlan`].
 ///
 /// [`LogicalPlan`]: crate::generated::algebra::LogicalPlan
-#[derive(Default, Clone)]
+#[derive(Clone)]
 pub struct LogicalPlan {
     pub(crate) nodes: VecMap<NodeType>,
     /// To record the nodes' maximum id in the logical plan. Note that the nodes
@@ -81,6 +95,14 @@ pub struct LogicalPlan {
     pub(crate) meta: PlanMeta,
 }
 
+impl Default for LogicalPlan {
+    fn default() -> Self {
+        let dummy_opr = pb::logical_plan::operator::Opr::Root(pb::Root {});
+        let dummy_root = Node::new(0, pb::logical_plan::Operator { opr: Some(dummy_opr) });
+        LogicalPlan::with_root(dummy_root)
+    }
+}
+
 impl PartialEq for LogicalPlan {
     fn eq(&self, other: &Self) -> bool {
         if self.nodes.len() != other.nodes.len() {
@@ -140,13 +162,21 @@ impl TryFrom<pb::LogicalPlan> for LogicalPlan {
                     }
                     _ => {}
                 }
-                let parent_ids = parents
+                let mut parent_ids = parents
                     .get(&(id as NodeId))
                     .cloned()
                     .unwrap_or_default()
                     .into_iter()
                     .map(|old| id_map[&old])
                     .collect::<Vec<NodeId>>();
+
+                // Point the nodes of pb root to the dummy RootScan
+                // For example, pb::LogicalPlan: scan->out->...
+                //              LogicalPlan: dummy root->scan->out...
+                if pb.roots.contains(&(id as i32)) && parent_ids.is_empty() {
+                    parent_ids = vec![0];
+                }
+
                 let new_id = plan
                     .append_operator_as_node(opr, parent_ids)
                     .map_err(|err| ParsePbError::ParseError(format!("{:?}", err)))?;
@@ -155,7 +185,6 @@ impl TryFrom<pb::LogicalPlan> for LogicalPlan {
                 return Err(ParsePbError::EmptyFieldError("Node::opr".to_string()));
             }
         }
-
         Ok(plan)
     }
 }
@@ -312,6 +341,10 @@ impl LogicalPlan {
         self.max_node_id
     }
 
+    pub fn get_plan_meta(&self) -> PlanMeta {
+        self.meta.clone()
+    }
+
     /// Append a new node into the logical plan, with specified `parent_ids`
     /// as its parent nodes. In order to do so, all specified parents must present in the
     /// logical plan.
@@ -393,14 +426,19 @@ impl LogicalPlan {
 
     /// Append an operator into the logical plan, as a new node with `self.max_node_id` as its id.
     pub fn append_operator_as_node(
-        &mut self, mut opr: pb::logical_plan::Operator, parent_ids: Vec<NodeId>,
+        &mut self, mut opr: pb::logical_plan::Operator, mut parent_ids: Vec<NodeId>,
     ) -> IrResult<NodeId> {
         use pb::logical_plan::operator::Opr;
 
+        let inner_opr = opr
+            .opr
+            .as_ref()
+            .ok_or(IrError::MissingData("Operator::opr".to_string()))?;
+
+        let is_sink = if let pb::logical_plan::operator::Opr::Sink(_) = inner_opr { true } else { false };
+
         let old_curr_node = self.meta.get_curr_node();
-        if opr.opr.is_none() {
-            return Err(IrError::MissingData("Operator::opr".to_string()));
-        }
+
         // Set new current node as `self.max_node_id`
         let new_curr_node = self.max_node_id;
         self.meta.set_curr_node(new_curr_node);
@@ -416,8 +454,9 @@ impl LogicalPlan {
         if let Ok(store_meta) = STORE_META.read() {
             opr.preprocess(&store_meta, &mut self.meta)?;
         }
-        let new_curr_node_rst = match opr.opr.as_ref().unwrap() {
-            Opr::Pattern(pattern) => {
+
+        let new_curr_node_rst = match opr.opr.as_ref() {
+            Some(Opr::Pattern(pattern)) => {
                 if parent_ids.len() == 1 {
                     // We try to match via ExtendStrategy. If not supported, match via NaiveStrategy.
 
@@ -428,41 +467,40 @@ impl LogicalPlan {
                     // 4. if match binders of PathExpand exists, it should at least range from 1
                     let is_pattern_source_whole_graph = self
                         .get_opr(parent_ids[0])
-                        .map(|pattern_source| pattern_source.is_whole_graph())
+                        .map(|pattern_source| is_whole_graph(&pattern_source))
                         .ok_or(IrError::ParentNodeNotExist(parent_ids[0]))?;
                     let extend_strategy = if is_pattern_source_whole_graph {
                         ExtendStrategy::init(&pattern, &self.meta)
                     } else {
                         Err(IrPatternError::Unsupported("pattern source is not whole graph".to_string()))
                     };
-                    match extend_strategy {
+
+                    let match_plan = match extend_strategy {
                         Ok(extend_strategy) => {
                             debug!("pattern matching by ExtendStrategy");
-                            let plan = extend_strategy.build_logical_plan()?;
-                            let new_node_id = self.append_plan(plan, parent_ids.clone())?;
-                            // As we have added a new source op to scan with label efficiently in extend_strategy,
-                            // we remove the old source op.
-                            self.nodes.remove(0);
-                            Ok(new_node_id)
+                            // Extend Strategy can only add to the source dummy node
+                            parent_ids = vec![0];
+                            extend_strategy.build_logical_plan()
                         }
                         Err(err) => match err {
                             IrPatternError::Unsupported(_) => {
                                 // if is not supported in ExtendStrategy, try NaiveStrategy
                                 debug!("pattern matching by NaiveStrategy");
                                 let naive_strategy = NaiveStrategy::try_from(pattern.clone())?;
-                                let plan = naive_strategy.build_logical_plan()?;
-                                self.append_plan(plan, parent_ids.clone())
+                                naive_strategy.build_logical_plan()
                             }
                             _ => Err(err.into()),
                         },
-                    }
+                    }?;
+                    self.append_plan(match_plan, parent_ids)
                 } else {
                     Err(IrError::Unsupported(
                         "only one single parent is supported for the `Pattern` operator".to_string(),
                     ))
                 }
             }
-            _ => self.append_node(Node::new(new_curr_node, opr), parent_ids.clone()),
+            Some(_) => self.append_node(Node::new(new_curr_node, opr), parent_ids),
+            None => Err(IrError::MissingData("Operator::opr".to_string())),
         };
 
         // As in this case, the current id will not refer to any actual nodes, it is fine to
@@ -471,9 +509,40 @@ impl LogicalPlan {
             self.meta.set_curr_node(old_curr_node);
         }
 
+        if is_sink {
+            self.clean_redundant_nodes();
+        }
+
         new_curr_node_rst
     }
 
+    /// Remove redundant nodes
+    ///
+    /// There are two types of redundant nodes:
+    /// - Dummy root node with only single child
+    /// - Scan node with empty child
+    pub(crate) fn clean_redundant_nodes(&mut self) {
+        // Remove scan node with empty child
+        let mut scan_ids_to_clean = vec![];
+        for (_, node) in self.nodes.iter() {
+            // Empty Children Scan nodes
+            if node.borrow().is_scan() && node.borrow().children.len() == 0 {
+                let empty_child_scan_id = node.borrow().id;
+                scan_ids_to_clean.push(empty_child_scan_id);
+            }
+        }
+        for id in scan_ids_to_clean {
+            self.remove_node(id);
+        }
+        // Remove dummy node with single child
+        if let Some(node) = self.get_first_node() {
+            if node.borrow().is_root() && node.borrow().children.len() <= 1 {
+                let single_child_dummy_id = node.borrow().id;
+                self.remove_node(single_child_dummy_id);
+            }
+        }
+    }
+
     /// Remove a node from the logical plan, and do the following:
     /// * For each of its parent, if present, remove this node's id reference from its `children`.
     /// * For each of its children, remove this node's id reference from its `parent`, and if
@@ -493,7 +562,7 @@ impl LogicalPlan {
             for c in &n.borrow().children {
                 if let Some(child) = self.get_node(*c) {
                     child.borrow_mut().parents.remove(&id);
-                    if child.borrow().parents.is_empty() {
+                    if child.borrow().parents.is_empty() && !n.borrow().is_root() {
                         // Recursively remove the child
                         let _ = self.remove_node(*c);
                     }
@@ -644,6 +713,7 @@ impl LogicalPlan {
                     if let Some(plan) = subplan.as_mut() {
                         plan.append_node(curr_node.borrow().clone(), parent_ids)
                             .expect("append node to subplan error!");
+                        plan.clean_redundant_nodes();
                     }
                     subplan
                 } else {
@@ -1272,6 +1342,57 @@ impl AsLogical for pb::IndexPredicate {
     }
 }
 
+fn is_whole_graph(operator: &pb::logical_plan::Operator) -> bool {
+    if let Some(opr) = &operator.opr {
+        match opr {
+            pb::logical_plan::operator::Opr::Scan(scan) => {
+                scan.idx_predicate.is_none()
+                    && scan.alias.is_none()
+                    && scan
+                        .params
+                        .as_ref()
+                        .map(|params| !params.is_queryable() && is_params_all_labels(params))
+                        .unwrap_or(true)
+            }
+            pb::logical_plan::operator::Opr::Root(_) => true,
+            _ => false,
+        }
+    } else {
+        false
+    }
+}
+
+fn is_params_all_labels(params: &pb::QueryParams) -> bool {
+    params.tables.is_empty()
+        || if let Some(schema) = STORE_META
+            .read()
+            .ok()
+            .as_ref()
+            .and_then(|store_meta| store_meta.schema.as_ref())
+        {
+            let params_label_ids: BTreeSet<LabelId> = params
+                .tables
+                .iter()
+                .filter_map(|name_or_id| {
+                    if let Some(ir_common::generated::common::name_or_id::Item::Id(id)) =
+                        name_or_id.item.as_ref()
+                    {
+                        Some(*id)
+                    } else {
+                        None
+                    }
+                })
+                .collect();
+            let meta_label_ids: BTreeSet<LabelId> = schema
+                .entity_labels_iter()
+                .map(|(_, label_id)| label_id)
+                .collect();
+            params_label_ids.eq(&meta_label_ids)
+        } else {
+            false
+        }
+}
+
 impl AsLogical for pb::OrderBy {
     fn preprocess(&mut self, meta: &StoreMeta, plan_meta: &mut PlanMeta) -> IrResult<()> {
         for pair in self.pairs.iter_mut() {
@@ -1448,21 +1569,25 @@ mod test {
         };
         let mut plan = LogicalPlan::default();
 
-        let id = plan
-            .append_operator_as_node(opr.clone(), vec![])
-            .unwrap();
-        assert_eq!(id, 0);
+        // The default logical plan should contain a dummy node
         assert_eq!(plan.len(), 1);
         assert_eq!(plan.max_node_id, 1);
-        let node0 = plan.get_node(0).unwrap().clone();
 
         let id = plan
-            .append_operator_as_node(opr.clone(), vec![0])
+            .append_operator_as_node(opr.clone(), vec![])
             .unwrap();
         assert_eq!(id, 1);
         assert_eq!(plan.len(), 2);
         assert_eq!(plan.max_node_id, 2);
-        let node1 = plan.get_node(1).unwrap().clone();
+        let node0 = plan.get_node(1).unwrap().clone();
+
+        let id = plan
+            .append_operator_as_node(opr.clone(), vec![1])
+            .unwrap();
+        assert_eq!(id, 2);
+        assert_eq!(plan.len(), 3);
+        assert_eq!(plan.max_node_id, 3);
+        let node1 = plan.get_node(2).unwrap().clone();
 
         let parents = node1
             .borrow()
@@ -1470,7 +1595,7 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(parents, vec![0]);
+        assert_eq!(parents, vec![1]);
 
         let children = node0
             .borrow()
@@ -1478,15 +1603,15 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(children, vec![1]);
+        assert_eq!(children, vec![2]);
 
         let id = plan
-            .append_operator_as_node(opr.clone(), vec![0, 1])
+            .append_operator_as_node(opr.clone(), vec![1, 2])
             .unwrap();
-        assert_eq!(id, 2);
-        assert_eq!(plan.len(), 3);
-        assert_eq!(plan.max_node_id, 3);
-        let node2 = plan.get_node(2).unwrap().clone();
+        assert_eq!(id, 3);
+        assert_eq!(plan.len(), 4);
+        assert_eq!(plan.max_node_id, 4);
+        let node2 = plan.get_node(3).unwrap().clone();
 
         let parents = node2
             .borrow()
@@ -1494,7 +1619,7 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(parents, vec![0, 1]);
+        assert_eq!(parents, vec![1, 2]);
 
         let children = node0
             .borrow()
@@ -1502,7 +1627,7 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(children, vec![1, 2]);
+        assert_eq!(children, vec![2, 3]);
 
         let children = node1
             .borrow()
@@ -1510,19 +1635,19 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(children, vec![2]);
+        assert_eq!(children, vec![3]);
 
-        let node2 = plan.remove_node(2);
-        assert_eq!(node2.unwrap().borrow().id, 2);
-        assert_eq!(plan.len(), 2);
-        assert_eq!(plan.max_node_id, 3);
+        let node2 = plan.remove_node(3);
+        assert_eq!(node2.unwrap().borrow().id, 3);
+        assert_eq!(plan.len(), 3);
+        assert_eq!(plan.max_node_id, 4);
         let children = node0
             .borrow()
             .children
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(children, vec![1]);
+        assert_eq!(children, vec![2]);
 
         let children = node1
             .borrow()
@@ -1532,36 +1657,36 @@ mod test {
             .collect::<Vec<NodeId>>();
         assert!(children.is_empty());
 
-        let _id = plan.append_operator_as_node(opr.clone(), vec![0, 2]);
+        let _id = plan.append_operator_as_node(opr.clone(), vec![1, 3]);
         match _id.err().unwrap() {
-            IrError::ParentNodeNotExist(node) => assert_eq!(node, 2),
+            IrError::ParentNodeNotExist(node) => assert_eq!(node, 3),
             _ => panic!("wrong error type"),
         }
-        assert_eq!(plan.len(), 2);
-        assert_eq!(plan.max_node_id, 3);
+        assert_eq!(plan.len(), 3);
+        assert_eq!(plan.max_node_id, 4);
         let children = node0
             .borrow()
             .children
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(children, vec![1]);
+        assert_eq!(children, vec![2]);
 
         // add node2 back again for further testing recursive removal
         let _ = plan
-            .append_operator_as_node(opr.clone(), vec![0, 1])
+            .append_operator_as_node(opr.clone(), vec![1, 2])
             .unwrap();
-        let node3 = plan.get_node(3).unwrap();
-        let _ = plan.remove_node(1);
-        assert_eq!(plan.len(), 2);
-        assert_eq!(plan.max_node_id, 4);
+        let node3 = plan.get_node(4).unwrap();
+        let _ = plan.remove_node(2);
+        assert_eq!(plan.len(), 3);
+        assert_eq!(plan.max_node_id, 5);
         let children = node0
             .borrow()
             .children
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(children, vec![3]);
+        assert_eq!(children, vec![4]);
 
         let parents = node3
             .borrow()
@@ -1569,7 +1694,7 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(parents, vec![0]);
+        assert_eq!(parents, vec![1]);
     }
 
     #[test]
@@ -1577,16 +1702,21 @@ mod test {
         let opr = pb::logical_plan::Operator {
             opr: Some(pb::logical_plan::operator::Opr::As(pb::As { alias: None })),
         };
+        let sink_opr = pb::logical_plan::Operator {
+            opr: Some(pb::logical_plan::operator::Opr::Sink(pb::Sink::default())),
+        };
         let root_pb = pb::logical_plan::Node { opr: Some(opr.clone()), children: vec![1, 2] };
         let node1_pb = pb::logical_plan::Node { opr: Some(opr.clone()), children: vec![2] };
-        let node2_pb = pb::logical_plan::Node { opr: Some(opr.clone()), children: vec![] };
-        let plan_pb = pb::LogicalPlan { nodes: vec![root_pb, node1_pb, node2_pb], roots: vec![0] };
+        let node2_pb = pb::logical_plan::Node { opr: Some(opr.clone()), children: vec![3] };
+        let sink_pb = pb::logical_plan::Node { opr: Some(sink_opr), children: vec![] };
+        let plan_pb = pb::LogicalPlan { nodes: vec![root_pb, node1_pb, node2_pb, sink_pb], roots: vec![0] };
 
         let plan = LogicalPlan::try_from(plan_pb).unwrap();
-        assert_eq!(plan.len(), 3);
-        let node0 = plan.get_node(0).unwrap();
-        let node1 = plan.get_node(1).unwrap();
-        let node2 = plan.get_node(2).unwrap();
+        assert_eq!(plan.len(), 4);
+        let node0 = plan.get_node(1).unwrap();
+        let node1 = plan.get_node(2).unwrap();
+        let node2 = plan.get_node(3).unwrap();
+        let sink = plan.get_node(4).unwrap();
 
         let children = node0
             .borrow()
@@ -1594,7 +1724,7 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(children, vec![1, 2]);
+        assert_eq!(children, vec![2, 3]);
 
         let children = node1
             .borrow()
@@ -1602,7 +1732,7 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(children, vec![2]);
+        assert_eq!(children, vec![3]);
 
         let parents = node1
             .borrow()
@@ -1610,7 +1740,15 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(parents, vec![0]);
+        assert_eq!(parents, vec![1]);
+
+        let children = node2
+            .borrow()
+            .children
+            .iter()
+            .map(|x| *x)
+            .collect::<Vec<NodeId>>();
+        assert_eq!(children, vec![4]);
 
         let parents = node2
             .borrow()
@@ -1618,7 +1756,15 @@ mod test {
             .iter()
             .map(|x| *x)
             .collect::<Vec<NodeId>>();
-        assert_eq!(parents, vec![0, 1]);
+        assert_eq!(parents, vec![1, 2]);
+
+        let parents = sink
+            .borrow()
+            .parents
+            .iter()
+            .map(|x| *x)
+            .collect::<Vec<NodeId>>();
+        assert_eq!(parents, vec![3]);
     }
 
     #[test]
@@ -1629,16 +1775,18 @@ mod test {
         let mut plan = LogicalPlan::default();
 
         let _ = plan
-            .append_operator_as_node(opr.clone(), vec![])
+            .append_operator_as_node(opr.clone(), vec![0])
             .unwrap();
         let _ = plan
-            .append_operator_as_node(opr.clone(), vec![0])
+            .append_operator_as_node(opr.clone(), vec![1])
             .unwrap();
         let _ = plan
-            .append_operator_as_node(opr.clone(), vec![0])
+            .append_operator_as_node(opr.clone(), vec![1])
             .unwrap();
 
-        let _ = plan.remove_node(1);
+        plan.clean_redundant_nodes();
+
+        let _ = plan.remove_node(2);
 
         let plan_pb = pb::LogicalPlan::from(plan);
         assert_eq!(plan_pb.nodes.len(), 2);
@@ -1998,25 +2146,25 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         // .hasLabel("person")
         let select = pb::Select { predicate: str_to_expr_pb("@.~label == \"person\"".to_string()).ok() };
-        plan.append_operator_as_node(select.into(), vec![0])
+        plan.append_operator_as_node(select.into(), vec![1])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         // .has("age", 27)
         let select = pb::Select { predicate: str_to_expr_pb("@.age == 27".to_string()).ok() };
-        plan.append_operator_as_node(select.into(), vec![1])
+        plan.append_operator_as_node(select.into(), vec![2])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
         // The column "age" in a predicate ".age == 27" should not be added
         assert!(plan
             .meta
-            .get_node_meta(0)
+            .get_node_meta(1)
             .unwrap()
             .get_columns()
             .is_empty());
@@ -2030,12 +2178,12 @@ mod test {
             is_append: false,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![2])
+        plan.append_operator_as_node(project.into(), vec![3])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![3]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![4]);
         assert_eq!(
             plan.meta
-                .get_node_meta(0)
+                .get_node_meta(1)
                 .unwrap()
                 .get_columns(),
             vec!["age".into(), "id".into(), "name".into()]
@@ -2053,9 +2201,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         // .out().as("here")
         let expand = pb::EdgeExpand {
@@ -2066,15 +2214,15 @@ mod test {
             alias: Some("here".into()),
             meta_data: None,
         };
-        plan.append_operator_as_node(expand.into(), vec![0])
+        plan.append_operator_as_node(expand.into(), vec![1])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
 
         // .has("lang", "Java")
         let select = pb::Select { predicate: str_to_expr_pb("@.lang == \"Java\"".to_string()).ok() };
-        plan.append_operator_as_node(select.into(), vec![1])
+        plan.append_operator_as_node(select.into(), vec![2])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
         // The column "lang" in a predicate should not be added
         assert!(plan
             .meta
@@ -2092,9 +2240,9 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![2])
+        plan.append_operator_as_node(project.into(), vec![3])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
 
         // .values("name")
         let project = pb::Project {
@@ -2105,16 +2253,16 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![3])
+        plan.append_operator_as_node(project.into(), vec![4])
             .unwrap();
         assert_eq!(
             plan.meta
-                .get_node_meta(1)
+                .get_node_meta(2)
                 .unwrap()
                 .get_columns(),
             vec!["name".into()]
         );
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![4]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![5]);
     }
 
     #[test]
@@ -2130,9 +2278,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         // .outE().as(0)
         let expand = pb::EdgeExpand {
@@ -2143,11 +2291,11 @@ mod test {
             alias: Some("e".into()),
             meta_data: None,
         };
-        plan.append_operator_as_node(expand.into(), vec![0])
+        plan.append_operator_as_node(expand.into(), vec![1])
             .unwrap();
         let e_tag_id = plan.meta.get_tag_id("e").unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
-        assert_eq!(plan.meta.get_tag_nodes(e_tag_id), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
+        assert_eq!(plan.meta.get_tag_nodes(e_tag_id), &vec![2]);
 
         // .inV().as("v")
         let getv = pb::GetV {
@@ -2157,11 +2305,11 @@ mod test {
             alias: Some("v".into()),
             meta_data: None,
         };
-        plan.append_operator_as_node(getv.into(), vec![1])
+        plan.append_operator_as_node(getv.into(), vec![2])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![3]);
         let v_tag_id = plan.meta.get_tag_id("v").unwrap();
-        assert_eq!(plan.meta.get_tag_nodes(v_tag_id), &vec![2]);
+        assert_eq!(plan.meta.get_tag_nodes(v_tag_id), &vec![3]);
 
         // .select("e")
         let project = pb::Project {
@@ -2172,9 +2320,9 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![2])
+        plan.append_operator_as_node(project.into(), vec![3])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
 
         // .order().by("weight")
         let orderby = pb::OrderBy {
@@ -2190,12 +2338,12 @@ mod test {
             }],
             limit: None,
         };
-        plan.append_operator_as_node(orderby.into(), vec![3])
+        plan.append_operator_as_node(orderby.into(), vec![4])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
         assert_eq!(
             plan.meta
-                .get_node_meta(1)
+                .get_node_meta(2)
                 .unwrap()
                 .get_columns(),
             vec!["weight".into()]
@@ -2210,9 +2358,9 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![4])
+        plan.append_operator_as_node(project.into(), vec![5])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![3]);
 
         // .values("name")
         let project = pb::Project {
@@ -2223,12 +2371,12 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![5])
+        plan.append_operator_as_node(project.into(), vec![6])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![6]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![7]);
         assert_eq!(
             plan.meta
-                .get_node_meta(2)
+                .get_node_meta(3)
                 .unwrap()
                 .get_columns(),
             vec!["name".into()]
@@ -2252,20 +2400,20 @@ mod test {
             meta_data: None,
         };
         let mut opr_id = plan
-            .append_operator_as_node(scan.into(), vec![])
+            .append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         // .has("name", "John")
         let select = pb::Select { predicate: str_to_expr_pb("@.name == \"John\"".to_string()).ok() };
         opr_id = plan
             .append_operator_as_node(select.into(), vec![opr_id as NodeId])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
         // The column "name" in a predicate should not be added
         assert!(plan
             .meta
-            .get_node_meta(0)
+            .get_node_meta(1)
             .unwrap()
             .get_columns()
             .is_empty());
@@ -2275,9 +2423,9 @@ mod test {
         opr_id = plan
             .append_operator_as_node(as_opr.into(), vec![opr_id as NodeId])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
         let a_id = plan.meta.get_tag_id("a").unwrap();
-        assert_eq!(plan.meta.get_tag_nodes(a_id), &vec![0]);
+        assert_eq!(plan.meta.get_tag_nodes(a_id), &vec![1]);
 
         // outE("knows").as('b').has("date", 20200101)
         let expand = pb::EdgeExpand {
@@ -2390,7 +2538,7 @@ mod test {
             meta_data: None,
         };
 
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
         assert!(plan
             .meta
@@ -2419,7 +2567,7 @@ mod test {
         };
 
         let opr_id = plan
-            .append_operator_as_node(scan.into(), vec![])
+            .append_operator_as_node(scan.into(), vec![0])
             .unwrap();
 
         let project = pb::Project {
@@ -2435,7 +2583,7 @@ mod test {
             .unwrap();
         assert!(plan
             .meta
-            .get_node_meta(0)
+            .get_node_meta(1)
             .unwrap()
             .is_all_columns());
     }
@@ -2454,9 +2602,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         // .out().as("a")
         let expand = pb::EdgeExpand {
@@ -2467,9 +2615,9 @@ mod test {
             alias: Some("a".into()),
             meta_data: None,
         };
-        plan.append_operator_as_node(expand.into(), vec![0])
+        plan.append_operator_as_node(expand.into(), vec![1])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
 
         // .in()
         let expand = pb::EdgeExpand {
@@ -2480,9 +2628,9 @@ mod test {
             alias: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(expand.into(), vec![1])
+        plan.append_operator_as_node(expand.into(), vec![2])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![3]);
 
         // .select("a").as("b")
         let project = pb::Project {
@@ -2493,9 +2641,9 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![2])
+        plan.append_operator_as_node(project.into(), vec![3])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
 
         // select("b").by("name")
         let project = pb::Project {
@@ -2506,16 +2654,16 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![3])
+        plan.append_operator_as_node(project.into(), vec![4])
             .unwrap();
         assert_eq!(
             plan.meta
-                .get_node_meta(1)
+                .get_node_meta(2)
                 .unwrap()
                 .get_columns(),
             vec!["name".into()]
         );
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![4]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![5]);
     }
 
     #[test]
@@ -2531,9 +2679,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         let expand = pb::EdgeExpand {
             v_tag: None,
@@ -2544,9 +2692,9 @@ mod test {
             meta_data: None,
         };
         let oprid = plan
-            .append_operator_as_node(expand.into(), vec![])
+            .append_operator_as_node(expand.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
 
         // .where(out())
         let apply = pb::Apply {
@@ -2556,9 +2704,9 @@ mod test {
             alias: None,
         };
         let oprid = plan
-            .append_operator_as_node(apply.into(), vec![0])
+            .append_operator_as_node(apply.into(), vec![1])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         // .valueMap("age")
         let project = pb::Project {
@@ -2573,7 +2721,7 @@ mod test {
             .unwrap();
         assert_eq!(
             plan.meta
-                .get_node_meta(0)
+                .get_node_meta(1)
                 .unwrap()
                 .get_columns(),
             vec!["age".into()]
@@ -2594,9 +2742,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         let group = pb::GroupBy {
             mappings: vec![pb::group_by::KeyAlias {
@@ -2610,10 +2758,10 @@ mod test {
             }],
             meta_data: vec![],
         };
-        plan.append_operator_as_node(group.into(), vec![0])
+        plan.append_operator_as_node(group.into(), vec![1])
             .unwrap();
         let keys_tag_id = plan.meta.get_tag_id("~keys_2_0").unwrap();
-        assert_eq!(plan.meta.get_tag_nodes(keys_tag_id), &vec![0]);
+        assert_eq!(plan.meta.get_tag_nodes(keys_tag_id), &vec![1]);
 
         let order = pb::OrderBy {
             pairs: vec![pb::order_by::OrderingPair {
@@ -2628,11 +2776,11 @@ mod test {
             }],
             limit: None,
         };
-        plan.append_operator_as_node(order.into(), vec![1])
+        plan.append_operator_as_node(order.into(), vec![2])
             .unwrap();
         assert!(plan
             .meta
-            .get_node_meta(0)
+            .get_node_meta(1)
             .unwrap()
             .get_columns()
             .contains(&"name".into()));
@@ -2652,9 +2800,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         let group = pb::GroupBy {
             mappings: vec![pb::group_by::KeyAlias {
@@ -2668,11 +2816,11 @@ mod test {
             }],
             meta_data: vec![],
         };
-        plan.append_operator_as_node(group.into(), vec![0])
+        plan.append_operator_as_node(group.into(), vec![1])
             .unwrap();
         let keys_tag_id = plan.meta.get_tag_id("~keys_2_0").unwrap();
 
-        assert_eq!(plan.meta.get_tag_nodes(keys_tag_id), &vec![0]);
+        assert_eq!(plan.meta.get_tag_nodes(keys_tag_id), &vec![1]);
 
         let project = pb::Project {
             mappings: vec![pb::project::ExprAlias {
@@ -2682,7 +2830,7 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![1])
+        plan.append_operator_as_node(project.into(), vec![2])
             .unwrap();
     }
 
@@ -2699,9 +2847,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         let group = pb::GroupBy {
             mappings: vec![pb::group_by::KeyAlias {
@@ -2721,14 +2869,14 @@ mod test {
             }],
             meta_data: vec![],
         };
-        plan.append_operator_as_node(group.into(), vec![0])
+        plan.append_operator_as_node(group.into(), vec![1])
             .unwrap();
         let keys_tag_id = plan.meta.get_tag_id("a").unwrap();
-        assert_eq!(plan.meta.get_tag_nodes(keys_tag_id), &vec![1]);
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_tag_nodes(keys_tag_id), &vec![2]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
         assert_eq!(
             plan.meta
-                .get_nodes_meta(&[0])
+                .get_nodes_meta(&[1])
                 .unwrap()
                 .get_columns(),
             vec!["name".into()]
@@ -2742,7 +2890,7 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![1])
+        plan.append_operator_as_node(project.into(), vec![2])
             .unwrap();
     }
 
@@ -2758,9 +2906,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         // by(outE().count())
         let expand = pb::EdgeExpand {
@@ -2788,7 +2936,7 @@ mod test {
         plan.append_operator_as_node(group.into(), vec![subtask])
             .unwrap();
         // the tag "~values_0_1" maps to id 0
-        assert_eq!(plan.meta.get_tag_nodes(0), &vec![2]); // tag self
+        assert_eq!(plan.meta.get_tag_nodes(0), &vec![3]); // tag self
 
         let apply = pb::Apply {
             join_kind: 0, // join
@@ -2796,12 +2944,12 @@ mod test {
             subtask: subtask as PbNodeId,
             alias: Some("~apply".into()),
         };
-        plan.append_operator_as_node(apply.into(), vec![0])
+        plan.append_operator_as_node(apply.into(), vec![1])
             .unwrap();
         // do not change the head of current node, given that alias is given
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
         // the tag "~apply" maps to id 1
-        assert_eq!(plan.meta.get_tag_nodes(1), &vec![3]);
+        assert_eq!(plan.meta.get_tag_nodes(1), &vec![4]);
 
         let group = pb::GroupBy {
             mappings: vec![pb::group_by::KeyAlias {
@@ -2825,12 +2973,12 @@ mod test {
             }],
             meta_data: vec![],
         };
-        plan.append_operator_as_node(group.into(), vec![3])
+        plan.append_operator_as_node(group.into(), vec![4])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
         assert_eq!(
             plan.meta
-                .get_node_meta(0)
+                .get_node_meta(1)
                 .unwrap()
                 .get_columns(),
             vec!["name".into()]
@@ -2849,9 +2997,9 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![0]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
 
         let project = pb::Project {
             mappings: vec![pb::project::ExprAlias {
@@ -2861,12 +3009,12 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![0])
+        plan.append_operator_as_node(project.into(), vec![1])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
         assert!(plan
             .meta
-            .get_node_meta(0)
+            .get_node_meta(1)
             .unwrap()
             .get_columns()
             .contains(&"workFrom".into()));
@@ -2878,9 +3026,9 @@ mod test {
             }],
             limit: None,
         };
-        plan.append_operator_as_node(order.into(), vec![1])
+        plan.append_operator_as_node(order.into(), vec![2])
             .unwrap();
-        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![1]);
+        assert_eq!(plan.meta.get_curr_referred_nodes(), &vec![2]);
     }
 
     #[test]
@@ -2895,7 +3043,7 @@ mod test {
             idx_predicate: None,
             meta_data: None,
         };
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
 
         let expand1 = pb::EdgeExpand {
@@ -2911,14 +3059,14 @@ mod test {
         let expand2 = expand1.clone();
         let expand3 = expand1.clone();
         let id1 = plan
-            .append_operator_as_node(expand1.into(), vec![0])
+            .append_operator_as_node(expand1.into(), vec![1])
             .unwrap();
         let id1_f = plan
             .append_operator_as_node(filter.into(), vec![id1])
             .unwrap();
 
         let opr_id = plan
-            .append_operator_as_node(expand2.into(), vec![0])
+            .append_operator_as_node(expand2.into(), vec![1])
             .unwrap();
         let id2 = plan
             .append_operator_as_node(expand3.into(), vec![opr_id])
@@ -2993,7 +3141,7 @@ mod test {
         };
 
         let opr_id = plan
-            .append_operator_as_node(scan.into(), vec![])
+            .append_operator_as_node(scan.into(), vec![0])
             .unwrap();
 
         // .out().as("o")
@@ -3071,7 +3219,7 @@ mod test {
             meta_data: None,
         };
 
-        plan.append_operator_as_node(scan.into(), vec![])
+        plan.append_operator_as_node(scan.into(), vec![0])
             .unwrap();
 
         let expand = pb::EdgeExpand {
@@ -3087,7 +3235,7 @@ mod test {
             .unwrap();
 
         let apply = pb::Apply { join_kind: 5, tags: vec![], subtask: root_id as PbNodeId, alias: None };
-        plan.append_operator_as_node(apply.into(), vec![0])
+        plan.append_operator_as_node(apply.into(), vec![1])
             .unwrap();
 
         let project = pb::Project {
@@ -3098,7 +3246,7 @@ mod test {
             is_append: true,
             meta_data: vec![],
         };
-        plan.append_operator_as_node(project.into(), vec![2])
+        plan.append_operator_as_node(project.into(), vec![3])
             .unwrap();
 
         let sink = pb::Sink {
@@ -3109,15 +3257,15 @@ mod test {
                 })),
             }),
         };
-        plan.append_operator_as_node(sink.into(), vec![3])
+        plan.append_operator_as_node(sink.into(), vec![4])
             .unwrap();
 
         let subplan = plan
-            .extract_subplan(plan.get_node(2).unwrap())
+            .extract_subplan(plan.get_node(3).unwrap())
             .unwrap();
         assert_eq!(subplan.len(), 1);
         match subplan
-            .get_node(1)
+            .get_node(2)
             .unwrap()
             .borrow()
             .opr
@@ -3147,23 +3295,25 @@ mod test {
             opr: Some(pb::logical_plan::operator::Opr::As(pb::As { alias: None })),
         };
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(opr.clone(), vec![])
-            .unwrap(); // root
         plan.append_operator_as_node(opr.clone(), vec![0])
+            .unwrap(); // root
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 1
-        plan.append_operator_as_node(opr.clone(), vec![0])
-            .unwrap(); // node 2
         plan.append_operator_as_node(opr.clone(), vec![1])
+            .unwrap(); // node 2
+        plan.append_operator_as_node(opr.clone(), vec![2])
             .unwrap(); // node 3
-        plan.append_operator_as_node(opr.clone(), vec![1])
+        plan.append_operator_as_node(opr.clone(), vec![2])
             .unwrap(); // node 4
-        plan.append_operator_as_node(opr.clone(), vec![3, 4])
+        plan.append_operator_as_node(opr.clone(), vec![4, 5])
             .unwrap(); // node 5
-        plan.append_operator_as_node(opr.clone(), vec![2, 5])
+        plan.append_operator_as_node(opr.clone(), vec![3, 6])
             .unwrap(); // node 6
-        plan.append_operator_as_node(opr.clone(), vec![6])
+        plan.append_operator_as_node(opr.clone(), vec![7])
             .unwrap(); // node 7
 
+        plan.clean_redundant_nodes();
+
         plan
     }
 
@@ -3180,18 +3330,21 @@ mod test {
             opr: Some(pb::logical_plan::operator::Opr::As(pb::As { alias: None })),
         };
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(opr.clone(), vec![])
-            .unwrap(); // root
         plan.append_operator_as_node(opr.clone(), vec![0])
+            .unwrap(); // root
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 1
-        plan.append_operator_as_node(opr.clone(), vec![0])
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 2
-        plan.append_operator_as_node(opr.clone(), vec![0])
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 3
-        plan.append_operator_as_node(opr.clone(), vec![1, 2])
+        plan.append_operator_as_node(opr.clone(), vec![2, 3])
             .unwrap(); // node 4
-        plan.append_operator_as_node(opr.clone(), vec![3, 4])
+        plan.append_operator_as_node(opr.clone(), vec![4, 5])
             .unwrap(); // node 5
+
+        plan.clean_redundant_nodes();
+
         plan
     }
 
@@ -3208,53 +3361,56 @@ mod test {
             opr: Some(pb::logical_plan::operator::Opr::As(pb::As { alias: None })),
         };
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(opr.clone(), vec![])
-            .unwrap(); // root
         plan.append_operator_as_node(opr.clone(), vec![0])
+            .unwrap(); // root
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 1
-        plan.append_operator_as_node(opr.clone(), vec![0])
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 2
-        plan.append_operator_as_node(opr.clone(), vec![0])
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 3
-        plan.append_operator_as_node(opr.clone(), vec![1, 2])
-            .unwrap(); // node 4
         plan.append_operator_as_node(opr.clone(), vec![2, 3])
+            .unwrap(); // node 4
+        plan.append_operator_as_node(opr.clone(), vec![3, 4])
             .unwrap(); // node 5
-        plan.append_operator_as_node(opr.clone(), vec![4, 5])
+        plan.append_operator_as_node(opr.clone(), vec![5, 6])
             .unwrap(); // node 6
+
+        plan.clean_redundant_nodes();
+
         plan
     }
     #[test]
     fn test_get_merge_node1() {
         let plan = create_logical_plan1();
-        let merge_node = plan.get_merge_node(plan.get_node(1).unwrap());
-        assert_eq!(merge_node, plan.get_node(5));
-        let merge_node = plan.get_merge_node(plan.get_node(0).unwrap());
+        let merge_node = plan.get_merge_node(plan.get_node(2).unwrap());
         assert_eq!(merge_node, plan.get_node(6));
+        let merge_node = plan.get_merge_node(plan.get_node(1).unwrap());
+        assert_eq!(merge_node, plan.get_node(7));
         // Not a branch node
-        let merge_node = plan.get_merge_node(plan.get_node(2).unwrap());
+        let merge_node = plan.get_merge_node(plan.get_node(3).unwrap());
         assert!(merge_node.is_none());
     }
 
     #[test]
     fn test_get_merge_node2() {
         let plan = create_logical_plan2();
-        let merge_node = plan.get_merge_node(plan.get_node(0).unwrap());
-        assert_eq!(merge_node, plan.get_node(5));
+        let merge_node = plan.get_merge_node(plan.get_node(1).unwrap());
+        assert_eq!(merge_node, plan.get_node(6));
         // Not a branch node
-        let merge_node = plan.get_merge_node(plan.get_node(2).unwrap());
+        let merge_node = plan.get_merge_node(plan.get_node(3).unwrap());
         assert!(merge_node.is_none());
     }
 
     #[test]
     fn test_get_merge_node3() {
         let plan = create_logical_plan3();
-        let merge_node = plan.get_merge_node(plan.get_node(0).unwrap());
-        assert_eq!(merge_node, plan.get_node(6));
-        let merge_node = plan.get_merge_node(plan.get_node(2).unwrap());
-        assert_eq!(merge_node, plan.get_node(6));
-        // Not a branch node
         let merge_node = plan.get_merge_node(plan.get_node(1).unwrap());
+        assert_eq!(merge_node, plan.get_node(7));
+        let merge_node = plan.get_merge_node(plan.get_node(3).unwrap());
+        assert_eq!(merge_node, plan.get_node(7));
+        // Not a branch node
+        let merge_node = plan.get_merge_node(plan.get_node(2).unwrap());
         assert!(merge_node.is_none());
     }
 
@@ -3263,28 +3419,28 @@ mod test {
         let opr = pb::logical_plan::Operator {
             opr: Some(pb::logical_plan::operator::Opr::As(pb::As { alias: None })),
         };
-        let mut plan = LogicalPlan::with_root(Node::new(0, opr.clone()));
+        let mut plan = LogicalPlan::with_root(Node::new(1, opr.clone()));
 
-        let mut subplan1 = LogicalPlan::with_root(Node::new(1, opr.clone()));
+        let mut subplan1 = LogicalPlan::with_root(Node::new(2, opr.clone()));
         subplan1
-            .append_node(Node::new(3, opr.clone()), vec![1])
+            .append_node(Node::new(4, opr.clone()), vec![2])
             .unwrap();
         subplan1
-            .append_node(Node::new(4, opr.clone()), vec![1])
+            .append_node(Node::new(5, opr.clone()), vec![2])
             .unwrap();
         subplan1
-            .append_node(Node::new(5, opr.clone()), vec![3, 4])
+            .append_node(Node::new(6, opr.clone()), vec![4, 5])
             .unwrap();
 
-        let subplan2 = LogicalPlan::with_root(Node::new(2, opr.clone()));
+        let subplan2 = LogicalPlan::with_root(Node::new(3, opr.clone()));
 
-        plan.append_branch_plans(plan.get_node(0).unwrap(), vec![subplan1, subplan2]);
+        plan.append_branch_plans(plan.get_node(1).unwrap(), vec![subplan1, subplan2]);
         let mut expected_plan = create_logical_plan1();
-        expected_plan.remove_node(6);
+        expected_plan.remove_node(7);
 
-        plan.append_node(Node::new(6, opr.clone()), vec![2, 5])
+        plan.append_node(Node::new(7, opr.clone()), vec![3, 6])
             .unwrap();
-        plan.append_node(Node::new(7, opr.clone()), vec![6])
+        plan.append_node(Node::new(8, opr.clone()), vec![7])
             .unwrap();
 
         assert_eq!(plan, create_logical_plan1());
@@ -3296,27 +3452,27 @@ mod test {
         let opr = pb::logical_plan::Operator {
             opr: Some(pb::logical_plan::operator::Opr::As(pb::As { alias: None })),
         };
-        let subplan = plan.subplan(plan.get_node(2).unwrap(), plan.get_node(7).unwrap());
-        let mut expected_plan = LogicalPlan::with_root(Node::new(2, opr.clone()));
+        let subplan = plan.subplan(plan.get_node(3).unwrap(), plan.get_node(8).unwrap());
+        let mut expected_plan = LogicalPlan::with_root(Node::new(3, opr.clone()));
         expected_plan
-            .append_node(Node::new(6, opr.clone()), vec![2])
+            .append_node(Node::new(7, opr.clone()), vec![3])
             .unwrap();
         assert_eq!(subplan.unwrap(), expected_plan);
 
         // The node 3 is at one of the branches, which is incomplete and hence invalid subplan
-        let subplan = plan.subplan(plan.get_node(1).unwrap(), plan.get_node(3).unwrap());
+        let subplan = plan.subplan(plan.get_node(2).unwrap(), plan.get_node(4).unwrap());
         assert!(subplan.is_none());
 
-        let subplan = plan.subplan(plan.get_node(1).unwrap(), plan.get_node(6).unwrap());
-        let mut expected_plan = LogicalPlan::with_root(Node::new(1, opr.clone()));
+        let subplan = plan.subplan(plan.get_node(2).unwrap(), plan.get_node(7).unwrap());
+        let mut expected_plan = LogicalPlan::with_root(Node::new(2, opr.clone()));
         expected_plan
-            .append_node(Node::new(3, opr.clone()), vec![1])
+            .append_node(Node::new(4, opr.clone()), vec![2])
             .unwrap();
         expected_plan
-            .append_node(Node::new(4, opr.clone()), vec![1])
+            .append_node(Node::new(5, opr.clone()), vec![2])
             .unwrap();
         expected_plan
-            .append_node(Node::new(5, opr.clone()), vec![3, 4])
+            .append_node(Node::new(6, opr.clone()), vec![4, 5])
             .unwrap();
 
         assert_eq!(subplan.unwrap(), expected_plan);
@@ -3325,51 +3481,51 @@ mod test {
     #[test]
     fn get_branch_plans() {
         let plan = create_logical_plan1();
-        let (merge_node, subplans) = plan.get_branch_plans(plan.get_node(1).unwrap());
+        let (merge_node, subplans) = plan.get_branch_plans(plan.get_node(2).unwrap());
         let opr = pb::logical_plan::Operator {
             opr: Some(pb::logical_plan::operator::Opr::As(pb::As { alias: None })),
         };
 
-        let plan1 = LogicalPlan::with_root(Node::new(3, opr.clone()));
-        let plan2 = LogicalPlan::with_root(Node::new(4, opr.clone()));
+        let plan1 = LogicalPlan::with_root(Node::new(4, opr.clone()));
+        let plan2 = LogicalPlan::with_root(Node::new(5, opr.clone()));
 
-        assert_eq!(merge_node, plan.get_node(5));
+        assert_eq!(merge_node, plan.get_node(6));
         assert_eq!(subplans, vec![plan1, plan2]);
 
-        let (merge_node, subplans) = plan.get_branch_plans(plan.get_node(0).unwrap());
-        let mut plan1 = LogicalPlan::with_root(Node::new(1, opr.clone()));
+        let (merge_node, subplans) = plan.get_branch_plans(plan.get_node(1).unwrap());
+        let mut plan1 = LogicalPlan::with_root(Node::new(2, opr.clone()));
         plan1
-            .append_node(Node::new(3, opr.clone()), vec![1])
+            .append_node(Node::new(4, opr.clone()), vec![2])
             .unwrap();
         plan1
-            .append_node(Node::new(4, opr.clone()), vec![1])
+            .append_node(Node::new(5, opr.clone()), vec![2])
             .unwrap();
         plan1
-            .append_node(Node::new(5, opr.clone()), vec![3, 4])
+            .append_node(Node::new(6, opr.clone()), vec![4, 5])
             .unwrap();
-        let plan2 = LogicalPlan::with_root(Node::new(2, opr.clone()));
+        let plan2 = LogicalPlan::with_root(Node::new(3, opr.clone()));
 
-        assert_eq!(merge_node, plan.get_node(6));
+        assert_eq!(merge_node, plan.get_node(7));
         assert_eq!(subplans, vec![plan1, plan2]);
 
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(opr.clone(), vec![])
-            .unwrap(); // root
         plan.append_operator_as_node(opr.clone(), vec![0])
+            .unwrap(); // root
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 1
-        plan.append_operator_as_node(opr.clone(), vec![0])
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 2
-        plan.append_operator_as_node(opr.clone(), vec![0])
+        plan.append_operator_as_node(opr.clone(), vec![1])
             .unwrap(); // node 3
-        plan.append_operator_as_node(opr.clone(), vec![1, 2, 3])
+        plan.append_operator_as_node(opr.clone(), vec![2, 3, 4])
             .unwrap(); // node 4
 
-        let (merge_node, subplans) = plan.get_branch_plans(plan.get_node(0).unwrap());
-        let plan1 = LogicalPlan::with_root(Node::new(1, opr.clone()));
-        let plan2 = LogicalPlan::with_root(Node::new(2, opr.clone()));
-        let plan3 = LogicalPlan::with_root(Node::new(3, opr.clone()));
+        let (merge_node, subplans) = plan.get_branch_plans(plan.get_node(1).unwrap());
+        let plan1 = LogicalPlan::with_root(Node::new(2, opr.clone()));
+        let plan2 = LogicalPlan::with_root(Node::new(3, opr.clone()));
+        let plan3 = LogicalPlan::with_root(Node::new(4, opr.clone()));
 
-        assert_eq!(merge_node, plan.get_node(4));
+        assert_eq!(merge_node, plan.get_node(5));
         assert_eq!(subplans, vec![plan1, plan2, plan3]);
     }
 }
diff --git a/interactive_engine/executor/ir/core/src/plan/meta.rs b/interactive_engine/executor/ir/core/src/plan/meta.rs
index eac241823605..85f858626cb0 100644
--- a/interactive_engine/executor/ir/core/src/plan/meta.rs
+++ b/interactive_engine/executor/ir/core/src/plan/meta.rs
@@ -27,6 +27,7 @@ use ir_common::{KeyId, OneOrMany};
 use ir_common::{LabelId, NameOrId};
 
 use crate::error::{IrError, IrResult};
+use crate::glogue::DynIter;
 use crate::plan::logical::NodeId;
 use crate::JsonIO;
 
@@ -214,6 +215,20 @@ impl Schema {
         self.is_table_id
     }
 
+    pub fn entity_labels_iter(&self) -> DynIter<(String, LabelId)> {
+        Box::new(
+            self.table_name_to_id
+                .iter()
+                .filter_map(|(name, (key_type, label_id))| {
+                    if let KeyType::Entity = key_type {
+                        Some((name.clone(), *label_id))
+                    } else {
+                        None
+                    }
+                }),
+        )
+    }
+
     /// Check whether a given table contains a given column as a primary key.
     /// Also return the number of primary keys of the given table.
     pub fn check_primary_key(&self, table: &str, col: &str) -> (bool, usize) {
diff --git a/interactive_engine/executor/ir/core/src/plan/physical.rs b/interactive_engine/executor/ir/core/src/plan/physical.rs
index 5072bb852e23..68833f2010b2 100644
--- a/interactive_engine/executor/ir/core/src/plan/physical.rs
+++ b/interactive_engine/executor/ir/core/src/plan/physical.rs
@@ -583,6 +583,10 @@ impl AsPhysical for pb::logical_plan::Operator {
                 Union(_) => Ok(()),
                 Intersect(_) => Ok(()),
                 Unfold(unfold) => unfold.add_job_builder(builder, plan_meta),
+                Root(_) => {
+                    builder.add_dummy_source();
+                    Ok(())
+                }
                 Sample(sample) => sample.add_job_builder(builder, plan_meta),
                 _ => Err(IrError::Unsupported(format!("the operator {:?}", self))),
             }
@@ -1078,12 +1082,15 @@ mod test {
     fn post_process_edgexpd() {
         // g.V().outE()
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(1, vec![], None).into(), vec![0])
+        plan.append_operator_as_node(build_edgexpd(1, vec![], None).into(), vec![1])
             .unwrap();
-        plan.append_operator_as_node(build_sink().into(), vec![1])
+        plan.append_operator_as_node(build_sink().into(), vec![2])
             .unwrap();
+
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan.add_job_builder(&mut job_builder, &mut plan_meta)
@@ -1117,14 +1124,17 @@ mod test {
         // In this case, the Select will be translated into an GetV, to fetch and filter the
         // results in one single `FilterMap` pegasus operator.
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(0, vec![], None).into(), vec![0])
+        plan.append_operator_as_node(build_edgexpd(0, vec![], None).into(), vec![1])
             .unwrap();
-        plan.append_operator_as_node(build_select("@.birthday == 20220101").into(), vec![1])
+        plan.append_operator_as_node(build_select("@.birthday == 20220101").into(), vec![2])
             .unwrap();
-        plan.append_operator_as_node(build_sink().into(), vec![2])
+        plan.append_operator_as_node(build_sink().into(), vec![3])
             .unwrap();
+
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan.add_job_builder(&mut job_builder, &mut plan_meta)
@@ -1159,14 +1169,17 @@ mod test {
     fn post_process_edgexpd_label_filter() {
         // g.V().out().filter(@.~label == "person")
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(0, vec![], None).into(), vec![0])
+        plan.append_operator_as_node(build_edgexpd(0, vec![], None).into(), vec![1])
             .unwrap();
-        plan.append_operator_as_node(build_select("@.~label == \"person\"").into(), vec![1])
+        plan.append_operator_as_node(build_select("@.~label == \"person\"").into(), vec![2])
             .unwrap();
-        plan.append_operator_as_node(build_sink().into(), vec![2])
+        plan.append_operator_as_node(build_sink().into(), vec![3])
             .unwrap();
+
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan.add_job_builder(&mut job_builder, &mut plan_meta)
@@ -1189,16 +1202,19 @@ mod test {
         // the properties using `GetV` twice before filter, and can
         // finally execute the selection of "0.age > 1.age".
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(0, vec![], Some(0.into())).into(), vec![0])
+        plan.append_operator_as_node(build_edgexpd(0, vec![], Some(0.into())).into(), vec![1])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(0, vec![], Some(1.into())).into(), vec![1])
+        plan.append_operator_as_node(build_edgexpd(0, vec![], Some(1.into())).into(), vec![2])
             .unwrap();
-        plan.append_operator_as_node(build_select("@0.age > @1.age").into(), vec![2])
+        plan.append_operator_as_node(build_select("@0.age > @1.age").into(), vec![3])
             .unwrap();
-        plan.append_operator_as_node(build_sink().into(), vec![3])
+        plan.append_operator_as_node(build_sink().into(), vec![4])
             .unwrap();
+
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan.add_job_builder(&mut job_builder, &mut plan_meta)
@@ -1247,14 +1263,17 @@ mod test {
     fn post_process_edgexpd_project_auxilia() {
         // g.V().out().as(0).select(0).by(valueMap("name", "id", "age")
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(0, vec![], Some(0.into())).into(), vec![0])
+        plan.append_operator_as_node(build_edgexpd(0, vec![], Some(0.into())).into(), vec![1])
             .unwrap();
-        plan.append_operator_as_node(build_project("{@0.name, @0.id, @0.age}").into(), vec![1])
+        plan.append_operator_as_node(build_project("{@0.name, @0.id, @0.age}").into(), vec![2])
             .unwrap();
-        plan.append_operator_as_node(build_sink().into(), vec![2])
+        plan.append_operator_as_node(build_sink().into(), vec![3])
             .unwrap();
+
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan.add_job_builder(&mut job_builder, &mut plan_meta)
@@ -1296,14 +1315,17 @@ mod test {
     fn post_process_edgexpd_tag_no_auxilia() {
         // g.V().out().as('a').select('a')
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(0, vec![], Some(0.into())).into(), vec![0])
+        plan.append_operator_as_node(build_edgexpd(0, vec![], Some(0.into())).into(), vec![1])
             .unwrap();
-        plan.append_operator_as_node(build_project("@0").into(), vec![1])
+        plan.append_operator_as_node(build_project("@0").into(), vec![2])
             .unwrap();
-        plan.append_operator_as_node(build_sink().into(), vec![2])
+        plan.append_operator_as_node(build_sink().into(), vec![3])
             .unwrap();
+
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan_meta = plan_meta.with_partition();
@@ -1324,22 +1346,24 @@ mod test {
     fn post_process_scan() {
         let mut plan = LogicalPlan::default();
         // g.V().hasLabel("person").has("age", 27).valueMap("age", "name", "id")
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
         // .hasLabel("person")
-        plan.append_operator_as_node(build_select("@.~label == \"person\"").into(), vec![0])
+        plan.append_operator_as_node(build_select("@.~label == \"person\"").into(), vec![1])
             .unwrap();
         // .has("age", 27)
-        plan.append_operator_as_node(build_select("@.age == 27").into(), vec![1])
+        plan.append_operator_as_node(build_select("@.age == 27").into(), vec![2])
             .unwrap();
 
         // .valueMap("age", "name", "id")
-        plan.append_operator_as_node(build_project("{@.name, @.id}").into(), vec![2])
+        plan.append_operator_as_node(build_project("{@.name, @.id}").into(), vec![3])
             .unwrap();
 
-        plan.append_operator_as_node(build_sink().into(), vec![3])
+        plan.append_operator_as_node(build_sink().into(), vec![4])
             .unwrap();
 
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan.add_job_builder(&mut job_builder, &mut plan_meta)
@@ -1376,16 +1400,19 @@ mod test {
     fn post_process_getv_auxilia_projection() {
         // g.V().outE().inV().as('a').select('a').by(valueMap("name", "id", "age")
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(1, vec![], None).into(), vec![0])
+        plan.append_operator_as_node(build_edgexpd(1, vec![], None).into(), vec![1])
             .unwrap();
-        plan.append_operator_as_node(build_getv(Some(0.into())).into(), vec![1])
+        plan.append_operator_as_node(build_getv(Some(0.into())).into(), vec![2])
             .unwrap();
-        plan.append_operator_as_node(build_project("{@0.name, @0.id, @0.age}").into(), vec![2])
+        plan.append_operator_as_node(build_project("{@0.name, @0.id, @0.age}").into(), vec![3])
             .unwrap();
-        plan.append_operator_as_node(build_sink().into(), vec![3])
+        plan.append_operator_as_node(build_sink().into(), vec![4])
             .unwrap();
+
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan.add_job_builder(&mut job_builder, &mut plan_meta)
@@ -1427,9 +1454,9 @@ mod test {
     fn post_process_getv_auxilia_filter() {
         // g.V().outE().inV().filter('age > 10')
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(build_scan(vec![]).into(), vec![])
+        plan.append_operator_as_node(build_scan(vec![]).into(), vec![0])
             .unwrap();
-        plan.append_operator_as_node(build_edgexpd(1, vec![], None).into(), vec![0])
+        plan.append_operator_as_node(build_edgexpd(1, vec![], None).into(), vec![1])
             .unwrap();
         plan.append_operator_as_node(
             pb::GetV {
@@ -1448,11 +1475,14 @@ mod test {
                 meta_data: None,
             }
             .into(),
-            vec![1],
+            vec![2],
         )
         .unwrap();
-        plan.append_operator_as_node(build_sink().into(), vec![2])
+        plan.append_operator_as_node(build_sink().into(), vec![3])
             .unwrap();
+
+        plan.clean_redundant_nodes();
+
         let mut job_builder = PlanBuilder::default();
         let mut plan_meta = plan.meta.clone();
         plan.add_job_builder(&mut job_builder, &mut plan_meta)
@@ -1491,20 +1521,23 @@ mod test {
         let mut logical_plan = LogicalPlan::default();
 
         logical_plan
-            .append_operator_as_node(source_opr.clone().into(), vec![])
+            .append_operator_as_node(source_opr.clone().into(), vec![0])
             .unwrap(); // node 0
         logical_plan
-            .append_operator_as_node(select_opr.clone().into(), vec![0])
+            .append_operator_as_node(select_opr.clone().into(), vec![1])
             .unwrap(); // node 1
         logical_plan
-            .append_operator_as_node(expand_opr.clone().into(), vec![1])
+            .append_operator_as_node(expand_opr.clone().into(), vec![2])
             .unwrap(); // node 2
         logical_plan
-            .append_operator_as_node(limit_opr.clone().into(), vec![2])
+            .append_operator_as_node(limit_opr.clone().into(), vec![3])
             .unwrap(); // node 3
         logical_plan
-            .append_operator_as_node(build_sink().into(), vec![3])
+            .append_operator_as_node(build_sink().into(), vec![4])
             .unwrap();
+
+        logical_plan.clean_redundant_nodes();
+
         let mut builder = PlanBuilder::default();
         let mut plan_meta = logical_plan.meta.clone();
         let _ = logical_plan.add_job_builder(&mut builder, &mut plan_meta);
@@ -1866,7 +1899,7 @@ mod test {
         };
 
         let opr_id = plan
-            .append_operator_as_node(scan.clone().into(), vec![])
+            .append_operator_as_node(scan.clone().into(), vec![0])
             .unwrap();
 
         // .out().as("1")
@@ -1904,6 +1937,8 @@ mod test {
         plan.append_operator_as_node(project.clone().into(), vec![opr_id])
             .unwrap();
 
+        plan.clean_redundant_nodes();
+
         let mut builder = PlanBuilder::default();
         let mut meta = plan.meta.clone();
         plan.add_job_builder(&mut builder, &mut meta)
@@ -1953,7 +1988,7 @@ mod test {
         };
 
         let opr_id = plan
-            .append_operator_as_node(scan.clone().into(), vec![])
+            .append_operator_as_node(scan.clone().into(), vec![0])
             .unwrap();
 
         // .out().count()
@@ -1968,6 +2003,8 @@ mod test {
         plan.append_operator_as_node(apply.clone().into(), vec![opr_id])
             .unwrap();
 
+        plan.clean_redundant_nodes();
+
         let mut builder = PlanBuilder::default();
         let mut meta = plan.meta.clone();
         plan.add_job_builder(&mut builder, &mut meta)
@@ -2016,7 +2053,7 @@ mod test {
         };
 
         let opr_id = plan
-            .append_operator_as_node(scan.clone().into(), vec![])
+            .append_operator_as_node(scan.clone().into(), vec![0])
             .unwrap();
 
         // .select("0").out().count()
@@ -2042,6 +2079,8 @@ mod test {
         plan.append_operator_as_node(apply.clone(), vec![opr_id])
             .unwrap();
 
+        plan.clean_redundant_nodes();
+
         let mut builder = PlanBuilder::default();
         let mut meta = plan.meta.clone();
         plan.add_job_builder(&mut builder, &mut meta)
diff --git a/interactive_engine/executor/ir/proto/algebra.proto b/interactive_engine/executor/ir/proto/algebra.proto
index b59cf8b27108..5ab4c5ea92db 100644
--- a/interactive_engine/executor/ir/proto/algebra.proto
+++ b/interactive_engine/executor/ir/proto/algebra.proto
@@ -469,6 +469,9 @@ message SinkVineyard {
   schema.Schema graph_schema = 2;
 }
 
+// A dummy node to delegate a source opr for multiple scan cases.
+message Root {}
+
 // A logical plan of graph relation algebra forms a DAG, where each node is an operator and the
 // edge indicates the execution flow.
 message LogicalPlan {
@@ -489,7 +492,8 @@ message LogicalPlan {
       As as = 14;
       Intersect intersect = 15;
       Sink sink = 16;
-      Sample sample = 17;
+      Root root = 17;
+      Sample sample = 18;
       // Saving the room for relational operators
       GetV vertex = 30;
       EdgeExpand edge = 31;
diff --git a/interactive_engine/executor/ir/proto/physical.proto b/interactive_engine/executor/ir/proto/physical.proto
index 14ff3701e8c0..b5faf0f5ba3f 100644
--- a/interactive_engine/executor/ir/proto/physical.proto
+++ b/interactive_engine/executor/ir/proto/physical.proto
@@ -267,7 +267,7 @@ message Repartition {
 }
 
 // A dummy node to delegate a source opr for multiple scan cases.
-message RootScan {}
+message Root {}
 
 message PhysicalOpr {
   message Operator {
@@ -286,7 +286,7 @@ message PhysicalOpr {
       Union union = 12;
       Intersect intersect = 13;
       Repartition repartition = 14;
-      RootScan root = 16;
+      Root root = 16;
       algebra.Sample sample = 17;
       // Saving the room for relational operators
       GetV vertex = 30;
diff --git a/interactive_engine/executor/ir/runtime/src/assembly.rs b/interactive_engine/executor/ir/runtime/src/assembly.rs
index fcd5966c23ec..d2bf8773dcc7 100644
--- a/interactive_engine/executor/ir/runtime/src/assembly.rs
+++ b/interactive_engine/executor/ir/runtime/src/assembly.rs
@@ -186,7 +186,7 @@ impl<P: PartitionInfo, C: ClusterInfo> IRJobAssembly<P, C> {
     fn install(
         &self, mut stream: Stream<Record>, plan: &[pb::PhysicalOpr],
     ) -> Result<Stream<Record>, BuildJobError> {
-        let mut prev_op_kind = pb::physical_opr::operator::OpKind::Root(pb::RootScan {});
+        let mut prev_op_kind = pb::physical_opr::operator::OpKind::Root(pb::Root {});
         for op in &plan[..] {
             let op_kind = op.try_into().map_err(|e| FnGenError::from(e))?;
             match op_kind {


Test patch:
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/DedupOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/DedupOpTest.java
deleted file mode 100644
index afd7cf4e064f..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/DedupOpTest.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.intermediate.ArgUtils;
-import com.alibaba.graphscope.common.intermediate.InterOpCollection;
-import com.alibaba.graphscope.common.ir.Utils;
-import com.alibaba.graphscope.common.jna.type.FfiVariable;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.function.Function;
-
-public class DedupOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void dedupTest() throws IOException {
-        DedupOp op = new DedupOp();
-        FfiVariable.ByValue dedupKey = ArgUtils.asNoneVar();
-        op.setDedupKeys(new OpArg(Collections.singletonList(dedupKey), Function.identity()));
-
-        irPlan = getTestIrPlan(op);
-        Assert.assertEquals(FileUtils.readJsonFromResource("dedup.json"), irPlan.getPlanAsJson());
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-
-    public static IrPlan getTestIrPlan(InterOpBase op, InterOpBase... ops) {
-        InterOpCollection opCollection = new InterOpCollection();
-        opCollection.appendInterOp(op);
-        if (ops != null) {
-            for (InterOpBase op1 : ops) {
-                opCollection.appendInterOp(op1);
-            }
-        }
-        return new IrPlan(Utils.schemaMeta, opCollection);
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ExpandOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ExpandOpTest.java
deleted file mode 100644
index adf68b7ac0b6..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ExpandOpTest.java
+++ /dev/null
@@ -1,77 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.intermediate.ArgUtils;
-import com.alibaba.graphscope.common.jna.type.FfiDirection;
-import com.alibaba.graphscope.common.jna.type.FfiExpandOpt;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.function.Function;
-
-public class ExpandOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void edgeOptTest() throws IOException {
-        ExpandOp op = new ExpandOp();
-        op.setEdgeOpt(new OpArg<>(FfiExpandOpt.Edge, Function.identity()));
-        op.setDirection(new OpArg<>(FfiDirection.Out, Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("expand_edge_opt.json"), actual);
-    }
-
-    @Test
-    public void labelsTest() throws IOException {
-        ExpandOp op = new ExpandOp();
-        op.setEdgeOpt(new OpArg<>(FfiExpandOpt.Edge, Function.identity()));
-        op.setDirection(new OpArg<>(FfiDirection.Out, Function.identity()));
-
-        QueryParams params = new QueryParams();
-        params.addTable(ArgUtils.asNameOrId("knows"));
-        op.setParams(params);
-
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("expand_labels.json"), actual);
-    }
-
-    @Test
-    public void aliasTest() throws IOException {
-        ExpandOp op = new ExpandOp();
-        op.setEdgeOpt(new OpArg<>(FfiExpandOpt.Edge, Function.identity()));
-        op.setDirection(new OpArg<>(FfiDirection.Out, Function.identity()));
-        op.setAlias(new OpArg(ArgUtils.asAlias("a", true), Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("expand_alias.json"), actual);
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/GetVOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/GetVOpTest.java
deleted file mode 100644
index d92d901af91b..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/GetVOpTest.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.jna.type.FfiVOpt;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-
-public class GetVOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void getBothVTest() throws IOException {
-        GetVOp getVOp = new GetVOp();
-        getVOp.setGetVOpt(new OpArg(FfiVOpt.Both));
-        irPlan = DedupOpTest.getTestIrPlan(getVOp);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("get_bothV.json"), actual);
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/GroupOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/GroupOpTest.java
deleted file mode 100644
index 2d67665c83ec..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/GroupOpTest.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.intermediate.ArgAggFn;
-import com.alibaba.graphscope.common.intermediate.ArgUtils;
-import com.alibaba.graphscope.common.jna.type.*;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.javatuples.Pair;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.function.Function;
-
-public class GroupOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void countTest() throws IOException {
-        GroupOp op = new GroupOp();
-        op.setGroupByKeys(new OpArg(Collections.emptyList(), Function.identity()));
-        ArgAggFn aggFn = new ArgAggFn(FfiAggOpt.Count, ArgUtils.asAlias("values", false));
-        op.setGroupByValues(new OpArg(Collections.singletonList(aggFn), Function.identity()));
-
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        Assert.assertEquals(FileUtils.readJsonFromResource("count.json"), irPlan.getPlanAsJson());
-    }
-
-    @Test
-    public void countAsTest() throws IOException {
-        GroupOp op = new GroupOp();
-        op.setGroupByKeys(new OpArg(Collections.emptyList(), Function.identity()));
-        ArgAggFn aggFn = new ArgAggFn(FfiAggOpt.Count, ArgUtils.asAlias("a", true));
-        op.setGroupByValues(new OpArg(Collections.singletonList(aggFn), Function.identity()));
-
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        Assert.assertEquals(
-                FileUtils.readJsonFromResource("count_as.json"), irPlan.getPlanAsJson());
-    }
-
-    @Test
-    public void groupTest() throws IOException {
-        GroupOp op = new GroupOp();
-        Pair<FfiVariable.ByValue, FfiAlias.ByValue> groupKey =
-                Pair.with(ArgUtils.asNoneVar(), ArgUtils.asAlias("keys", false));
-        op.setGroupByKeys(new OpArg(Collections.singletonList(groupKey), Function.identity()));
-
-        ArgAggFn aggFn = new ArgAggFn(FfiAggOpt.ToList, ArgUtils.asAlias("values", false));
-        op.setGroupByValues(new OpArg(Collections.singletonList(aggFn), Function.identity()));
-
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        Assert.assertEquals(FileUtils.readJsonFromResource("group.json"), irPlan.getPlanAsJson());
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/LimitOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/LimitOpTest.java
deleted file mode 100644
index 8ad7a2d5ea86..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/LimitOpTest.java
+++ /dev/null
@@ -1,48 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.function.Function;
-
-public class LimitOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void limitOpTest() throws IOException {
-        LimitOp op = new LimitOp();
-        op.setLower(new OpArg<>(Integer.valueOf(1), Function.identity()));
-        op.setUpper(new OpArg<>(Integer.valueOf(2), Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("limit_range.json"), actual);
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/OrderOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/OrderOpTest.java
deleted file mode 100644
index 910188c92b92..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/OrderOpTest.java
+++ /dev/null
@@ -1,79 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.intermediate.ArgUtils;
-import com.alibaba.graphscope.common.jna.type.FfiOrderOpt;
-import com.alibaba.graphscope.common.jna.type.FfiVariable;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.javatuples.Pair;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.function.Function;
-
-public class OrderOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void orderTest() throws IOException {
-        OrderOp op = new OrderOp();
-        op.setOrderVarWithOrder(
-                new OpArg(
-                        Arrays.asList(Pair.with(ArgUtils.asNoneVar(), FfiOrderOpt.Asc)),
-                        Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        Assert.assertEquals(
-                FileUtils.readJsonFromResource("order_asc.json"), irPlan.getPlanAsJson());
-    }
-
-    @Test
-    public void orderByLabelTest() throws IOException {
-        OrderOp op = new OrderOp();
-        FfiVariable.ByValue var = ArgUtils.asVar("", "~label");
-        op.setOrderVarWithOrder(
-                new OpArg(Arrays.asList(Pair.with(var, FfiOrderOpt.Asc)), Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        Assert.assertEquals(
-                FileUtils.readJsonFromResource("order_label.json"), irPlan.getPlanAsJson());
-    }
-
-    @Test
-    public void orderLimitTest() throws IOException {
-        OrderOp op = new OrderOp();
-        FfiVariable.ByValue var = ArgUtils.asNoneVar();
-        op.setOrderVarWithOrder(
-                new OpArg(Arrays.asList(Pair.with(var, FfiOrderOpt.Asc)), Function.identity()));
-        op.setLower(new OpArg(Integer.valueOf(1), Function.identity()));
-        op.setUpper(new OpArg(Integer.valueOf(2), Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        Assert.assertEquals(
-                FileUtils.readJsonFromResource("order_limit.json"), irPlan.getPlanAsJson());
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/PathExpandOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/PathExpandOpTest.java
deleted file mode 100644
index 78bb6b97e168..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/PathExpandOpTest.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.jna.type.FfiDirection;
-import com.alibaba.graphscope.common.jna.type.FfiExpandOpt;
-import com.alibaba.graphscope.common.jna.type.PathOpt;
-import com.alibaba.graphscope.common.jna.type.ResultOpt;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.function.Function;
-
-public class PathExpandOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void expand_1_5_Test() throws IOException {
-        PathExpandOp op = new PathExpandOp();
-        op.setEdgeOpt(new OpArg<>(FfiExpandOpt.Vertex, Function.identity()));
-        op.setDirection(new OpArg<>(FfiDirection.Out, Function.identity()));
-        op.setLower(new OpArg(Integer.valueOf(1)));
-        op.setUpper(new OpArg(Integer.valueOf(5)));
-        op.setPathOpt(PathOpt.Simple);
-        op.setResultOpt(ResultOpt.AllV);
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("path_expand.json"), actual);
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ProjectOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ProjectOpTest.java
deleted file mode 100644
index ca2d4156f8d6..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ProjectOpTest.java
+++ /dev/null
@@ -1,78 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.intermediate.ArgUtils;
-import com.alibaba.graphscope.common.jna.type.FfiAlias;
-import com.alibaba.graphscope.common.jna.type.FfiScanOpt;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.javatuples.Pair;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.List;
-import java.util.function.Function;
-
-public class ProjectOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void projectTagKeyTest() throws IOException {
-        ScanFusionOp scanOp = new ScanFusionOp();
-        scanOp.setScanOpt(new OpArg<>(FfiScanOpt.Entity, Function.identity()));
-        scanOp.setAlias(new OpArg(ArgUtils.asAlias("a", true), Function.identity()));
-
-        ProjectOp projectOp = new ProjectOp();
-        String projectExpr = "@a.name";
-        FfiAlias.ByValue alias = ArgUtils.asAlias("a_name", false);
-        List<Pair> exprWithAlias = Collections.singletonList(Pair.with(projectExpr, alias));
-        projectOp.setExprWithAlias(new OpArg(exprWithAlias, Function.identity()));
-
-        irPlan = DedupOpTest.getTestIrPlan(scanOp, projectOp);
-        Assert.assertEquals(
-                FileUtils.readJsonFromResource("project_tag_key.json"), irPlan.getPlanAsJson());
-    }
-
-    @Test
-    public void projectTagKeysTest() throws IOException {
-        ScanFusionOp scanOp = new ScanFusionOp();
-        scanOp.setScanOpt(new OpArg<>(FfiScanOpt.Entity, Function.identity()));
-        scanOp.setAlias(new OpArg(ArgUtils.asAlias("a", true), Function.identity()));
-
-        ProjectOp op = new ProjectOp();
-        String projectExpr = "{@a.name, @a.id}";
-        FfiAlias.ByValue alias = ArgUtils.asAlias("a_{name, id}", false);
-        List<Pair> exprWithAlias = Collections.singletonList(Pair.with(projectExpr, alias));
-        op.setExprWithAlias(new OpArg(exprWithAlias, Function.identity()));
-
-        irPlan = DedupOpTest.getTestIrPlan(scanOp, op);
-        Assert.assertEquals(
-                FileUtils.readJsonFromResource("project_tag_keys.json"), irPlan.getPlanAsJson());
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ScanFusionOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ScanFusionOpTest.java
deleted file mode 100644
index c4be67ea50db..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/ScanFusionOpTest.java
+++ /dev/null
@@ -1,115 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.intermediate.ArgUtils;
-import com.alibaba.graphscope.common.jna.type.FfiConst;
-import com.alibaba.graphscope.common.jna.type.FfiScanOpt;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.List;
-import java.util.function.Function;
-
-public class ScanFusionOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void scanOptTest() throws IOException {
-        ScanFusionOp op = new ScanFusionOp();
-        op.setScanOpt(new OpArg<>(FfiScanOpt.Entity, Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("scan_opt.json"), actual);
-    }
-
-    @Test
-    public void addSnapshotTest() throws IOException {
-        ScanFusionOp op = new ScanFusionOp();
-        op.setScanOpt(new OpArg<>(FfiScanOpt.Entity, Function.identity()));
-
-        QueryParams params = new QueryParams();
-        params.addExtraParams("SID", "1");
-        op.setParams(params);
-
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("scan_snapshot.json"), actual);
-    }
-
-    @Test
-    public void predicateTest() throws IOException {
-        ScanFusionOp op = new ScanFusionOp();
-        op.setScanOpt(new OpArg<>(FfiScanOpt.Entity, Function.identity()));
-
-        QueryParams params = new QueryParams();
-        params.setPredicate("@.id == 1");
-        op.setParams(params);
-
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("scan_expr.json"), actual);
-    }
-
-    @Test
-    public void labelsTest() throws IOException {
-        ScanFusionOp op = new ScanFusionOp();
-        op.setScanOpt(new OpArg<>(FfiScanOpt.Entity, Function.identity()));
-
-        QueryParams params = new QueryParams();
-        params.addTable(ArgUtils.asNameOrId("person"));
-        op.setParams(params);
-
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("scan_labels.json"), actual);
-    }
-
-    @Test
-    public void idsTest() throws IOException {
-        ScanFusionOp op = new ScanFusionOp();
-        op.setScanOpt(new OpArg<>(FfiScanOpt.Entity, Function.identity()));
-        List<FfiConst.ByValue> values = Arrays.asList(ArgUtils.asConst(1L), ArgUtils.asConst(2L));
-        op.setIds(new OpArg<List, List>(values, Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("scan_ids.json"), actual);
-    }
-
-    @Test
-    public void aliasTest() throws IOException {
-        ScanFusionOp op = new ScanFusionOp();
-        op.setScanOpt(new OpArg<>(FfiScanOpt.Entity, Function.identity()));
-        op.setAlias(new OpArg(ArgUtils.asAlias("a", true), Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("scan_alias.json"), actual);
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/SelectOpTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/SelectOpTest.java
deleted file mode 100644
index bef689fcfa68..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/operator/SelectOpTest.java
+++ /dev/null
@@ -1,47 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.operator;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.utils.FileUtils;
-
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.io.IOException;
-import java.util.function.Function;
-
-public class SelectOpTest {
-    private IrPlan irPlan;
-
-    @Test
-    public void selectOpTest() throws IOException {
-        SelectOp op = new SelectOp();
-        op.setPredicate(new OpArg("@.id == 1 && @.name == \"marko\"", Function.identity()));
-        irPlan = DedupOpTest.getTestIrPlan(op);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("select_expr.json"), actual);
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/strategy/ElementFusionStrategyTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/strategy/ElementFusionStrategyTest.java
deleted file mode 100644
index 24e800b14701..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/strategy/ElementFusionStrategyTest.java
+++ /dev/null
@@ -1,80 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.strategy;
-
-import com.alibaba.graphscope.common.IrPlan;
-import com.alibaba.graphscope.common.intermediate.InterOpCollection;
-import com.alibaba.graphscope.common.ir.Utils;
-import com.alibaba.graphscope.common.utils.FileUtils;
-import com.alibaba.graphscope.gremlin.InterOpCollectionBuilder;
-import com.alibaba.graphscope.gremlin.integration.suite.utils.__;
-import com.alibaba.graphscope.gremlin.plugin.processor.IrStandardOpProcessor;
-import com.alibaba.graphscope.gremlin.plugin.traversal.IrCustomizedTraversalSource;
-
-import org.apache.tinkerpop.gremlin.process.traversal.Traversal;
-import org.apache.tinkerpop.gremlin.process.traversal.dsl.graph.GraphTraversalSource;
-import org.apache.tinkerpop.gremlin.structure.Graph;
-import org.apache.tinkerpop.gremlin.tinkergraph.structure.TinkerFactory;
-import org.junit.After;
-import org.junit.Assert;
-import org.junit.Test;
-
-public class ElementFusionStrategyTest {
-    private IrPlan irPlan;
-    private Graph graph = TinkerFactory.createModern();
-    private GraphTraversalSource g = graph.traversal(IrCustomizedTraversalSource.class);
-
-    // group().by(outE().count())
-    @Test
-    public void g_V_group_by_outE_count_test() {
-        Traversal traversal = g.V().group().by(__.outE().count());
-        IrStandardOpProcessor.applyStrategies(traversal);
-        InterOpCollection opCollection = (new InterOpCollectionBuilder(traversal)).build();
-        irPlan = new IrPlan(Utils.schemaMeta, opCollection);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("group_key_out_count.json"), actual);
-    }
-
-    // group().by(outE().has('name', 'marko').count())
-    @Test
-    public void g_V_group_by_outE_has_count_test() {
-        Traversal traversal = g.V().group().by(__.outE().has("name", "marko").count());
-        IrStandardOpProcessor.applyStrategies(traversal);
-        InterOpCollection opCollection = (new InterOpCollectionBuilder(traversal)).build();
-        irPlan = new IrPlan(Utils.schemaMeta, opCollection);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("group_key_out_has_count.json"), actual);
-    }
-
-    // group().by(out().count())
-    @Test
-    public void g_V_group_by_out_count_test() {
-        Traversal traversal = g.V().group().by(__.out().count());
-        IrStandardOpProcessor.applyStrategies(traversal);
-        InterOpCollection opCollection = (new InterOpCollectionBuilder(traversal)).build();
-        irPlan = new IrPlan(Utils.schemaMeta, opCollection);
-        String actual = irPlan.getPlanAsJson();
-        Assert.assertEquals(FileUtils.readJsonFromResource("group_key_out_count.json"), actual);
-    }
-
-    @After
-    public void after() {
-        if (irPlan != null) {
-            irPlan.close();
-        }
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/strategy/TopKStrategyTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/strategy/TopKStrategyTest.java
deleted file mode 100644
index ffa0d87cd287..000000000000
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/intermediate/strategy/TopKStrategyTest.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Copyright 2020 Alibaba Group Holding Limited.
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package com.alibaba.graphscope.common.intermediate.strategy;
-
-import com.alibaba.graphscope.common.intermediate.ArgUtils;
-import com.alibaba.graphscope.common.intermediate.InterOpCollection;
-import com.alibaba.graphscope.common.intermediate.operator.LimitOp;
-import com.alibaba.graphscope.common.intermediate.operator.OpArg;
-import com.alibaba.graphscope.common.intermediate.operator.OrderOp;
-import com.alibaba.graphscope.common.jna.type.FfiOrderOpt;
-
-import org.javatuples.Pair;
-import org.junit.Assert;
-import org.junit.Test;
-
-import java.util.Arrays;
-import java.util.List;
-import java.util.function.Function;
-
-public class TopKStrategyTest {
-    @Test
-    public void orderLimitTest() {
-        InterOpCollection opCollection = new InterOpCollection();
-
-        OrderOp orderOp = new OrderOp();
-        List<Pair> orderPair = Arrays.asList(Pair.with(ArgUtils.asNoneVar(), FfiOrderOpt.Asc));
-        orderOp.setOrderVarWithOrder(new OpArg(orderPair, Function.identity()));
-        opCollection.appendInterOp(orderOp);
-
-        LimitOp limitOp = new LimitOp();
-        limitOp.setLower(new OpArg<>(Integer.valueOf(1), Function.identity()));
-        limitOp.setUpper(new OpArg<>(Integer.valueOf(2), Function.identity()));
-        opCollection.appendInterOp(limitOp);
-
-        TopKStrategy.INSTANCE.apply(opCollection);
-        OrderOp orderLimit = (OrderOp) opCollection.unmodifiableCollection().get(0);
-
-        Assert.assertEquals(1, opCollection.unmodifiableCollection().size());
-        Assert.assertEquals(1, orderLimit.getLower().get().applyArg());
-        Assert.assertEquals(2, orderLimit.getUpper().get().applyArg());
-    }
-}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java
index a0c070d29a65..9bd601883ec6 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java
@@ -87,6 +87,7 @@ public void logical_plan_1_test() throws Exception {
                         getMockGraphConfig(),
                         Utils.schemaMeta,
                         new LogicalPlan(aggregate, false))) {
+            ffiBuilder.build();
             Assert.assertEquals(
                     FileUtils.readJsonFromResource("ffi_logical_plan_1.json"),
                     ffiBuilder.explain());
@@ -117,6 +118,7 @@ public void logical_plan_2_test() throws Exception {
         try (PhysicalBuilder<byte[]> ffiBuilder =
                 new FfiPhysicalBuilder(
                         getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(filter, false))) {
+            ffiBuilder.build();
             Assert.assertEquals(
                     FileUtils.readJsonFromResource("ffi_logical_plan_2.json"),
                     ffiBuilder.explain());
@@ -143,6 +145,7 @@ public void logical_plan_3_test() throws Exception {
         try (PhysicalBuilder<byte[]> ffiBuilder =
                 new FfiPhysicalBuilder(
                         getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(project, false))) {
+            ffiBuilder.build();
             Assert.assertEquals(
                     FileUtils.readJsonFromResource("ffi_logical_plan_3.json"),
                     ffiBuilder.explain());
@@ -155,6 +158,7 @@ public void logical_plan_4_test() throws Exception {
                 com.alibaba.graphscope.cypher.antlr4.Utils.evalLogicalPlan(
                         "Call query_ic2(10l, 20120112l)");
         try (PhysicalBuilder<byte[]> ffiBuilder = new ProcedurePhysicalBuilder(logicalPlan)) {
+            ffiBuilder.build();
             Assert.assertEquals(
                     FileUtils.readJsonFromResource("call_procedure.json"), ffiBuilder.explain());
         }
@@ -189,6 +193,7 @@ public void logical_plan_5_test() throws Exception {
         try (PhysicalBuilder<byte[]> ffiBuilder =
                 new FfiPhysicalBuilder(
                         getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(project, false))) {
+            ffiBuilder.build();
             Assert.assertEquals(
                     FileUtils.readJsonFromResource("case_when.json"), ffiBuilder.explain());
         }
diff --git a/interactive_engine/compiler/src/test/resources/auxilia_alias.json b/interactive_engine/compiler/src/test/resources/auxilia_alias.json
deleted file mode 100644
index 83404952dcef..000000000000
--- a/interactive_engine/compiler/src/test/resources/auxilia_alias.json
+++ /dev/null
@@ -1,28 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Auxilia": {
-            "params": {
-              "table_names": [],
-              "columns": [],
-              "limit": null,
-              "predicate": null,
-              "requirements": []
-            },
-            "alias": {
-              "item": {
-                "Name": "a"
-              }
-            }
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/auxilia_properties.json b/interactive_engine/compiler/src/test/resources/auxilia_properties.json
deleted file mode 100644
index c5cb5824ec4b..000000000000
--- a/interactive_engine/compiler/src/test/resources/auxilia_properties.json
+++ /dev/null
@@ -1,36 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Auxilia": {
-            "tag": null,
-            "params": {
-              "table_names": [],
-              "columns": [
-                {
-                  "item": {
-                    "Name": "age"
-                  }
-                },
-                {
-                  "item": {
-                    "Name": "name"
-                  }
-                }
-              ],
-              "limit": null,
-              "predicate": null,
-              "requirements": []
-            },
-            "alias": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/case_when.json b/interactive_engine/compiler/src/test/resources/case_when.json
index 5192d10d3b8d..cc9d70a4557d 100644
--- a/interactive_engine/compiler/src/test/resources/case_when.json
+++ b/interactive_engine/compiler/src/test/resources/case_when.json
@@ -218,6 +218,25 @@
           }
         }
       },
+      "children": [
+        2
+      ]
+    },
+    {
+      "opr": {
+        "opr": {
+          "Sink": {
+            "tags": [],
+            "sink_target": {
+              "inner": {
+                "SinkDefault": {
+                  "id_name_mappings": []
+                }
+              }
+            }
+          }
+        }
+      },
       "children": []
     }
   ],
diff --git a/interactive_engine/compiler/src/test/resources/count.json b/interactive_engine/compiler/src/test/resources/count.json
deleted file mode 100644
index 2e9a6b1a185f..000000000000
--- a/interactive_engine/compiler/src/test/resources/count.json
+++ /dev/null
@@ -1,35 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "GroupBy": {
-            "mappings": [],
-            "functions": [
-              {
-                "vars": [
-                  {
-                    "tag": null,
-                    "property": null,
-                    "node_type": null
-                  }
-                ],
-                "aggregate": 3,
-                "alias": {
-                  "item": {
-                    "Id": 0
-                  }
-                }
-              }
-            ],
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/count_as.json b/interactive_engine/compiler/src/test/resources/count_as.json
deleted file mode 100644
index 2e9a6b1a185f..000000000000
--- a/interactive_engine/compiler/src/test/resources/count_as.json
+++ /dev/null
@@ -1,35 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "GroupBy": {
-            "mappings": [],
-            "functions": [
-              {
-                "vars": [
-                  {
-                    "tag": null,
-                    "property": null,
-                    "node_type": null
-                  }
-                ],
-                "aggregate": 3,
-                "alias": {
-                  "item": {
-                    "Id": 0
-                  }
-                }
-              }
-            ],
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/dedup.json b/interactive_engine/compiler/src/test/resources/dedup.json
deleted file mode 100644
index 17f4df8590a5..000000000000
--- a/interactive_engine/compiler/src/test/resources/dedup.json
+++ /dev/null
@@ -1,23 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Dedup": {
-            "keys": [
-              {
-                "tag": null,
-                "property": null,
-                "node_type": null
-              }
-            ]
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/expand_alias.json b/interactive_engine/compiler/src/test/resources/expand_alias.json
deleted file mode 100644
index d287c3016e7c..000000000000
--- a/interactive_engine/compiler/src/test/resources/expand_alias.json
+++ /dev/null
@@ -1,34 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Edge": {
-            "v_tag": null,
-            "direction": 0,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "alias": {
-              "item": {
-                "Id": 0
-              }
-            },
-            "expand_opt": 1,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/expand_edge_opt.json b/interactive_engine/compiler/src/test/resources/expand_edge_opt.json
deleted file mode 100644
index 6214782095cb..000000000000
--- a/interactive_engine/compiler/src/test/resources/expand_edge_opt.json
+++ /dev/null
@@ -1,30 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Edge": {
-            "v_tag": null,
-            "direction": 0,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "alias": null,
-            "expand_opt": 1,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/expand_labels.json b/interactive_engine/compiler/src/test/resources/expand_labels.json
deleted file mode 100644
index 2c7b971f9cd5..000000000000
--- a/interactive_engine/compiler/src/test/resources/expand_labels.json
+++ /dev/null
@@ -1,36 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Edge": {
-            "v_tag": null,
-            "direction": 0,
-            "params": {
-              "tables": [
-                {
-                  "item": {
-                    "Id": 0
-                  }
-                }
-              ],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "alias": null,
-            "expand_opt": 1,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/ffi_logical_plan_1.json b/interactive_engine/compiler/src/test/resources/ffi_logical_plan_1.json
index 3ada39d5992f..0c8bab63ce63 100644
--- a/interactive_engine/compiler/src/test/resources/ffi_logical_plan_1.json
+++ b/interactive_engine/compiler/src/test/resources/ffi_logical_plan_1.json
@@ -422,6 +422,25 @@
           }
         }
       },
+      "children": [
+        4
+      ]
+    },
+    {
+      "opr": {
+        "opr": {
+          "Sink": {
+            "tags": [],
+            "sink_target": {
+              "inner": {
+                "SinkDefault": {
+                  "id_name_mappings": []
+                }
+              }
+            }
+          }
+        }
+      },
       "children": []
     }
   ],
diff --git a/interactive_engine/compiler/src/test/resources/ffi_logical_plan_2.json b/interactive_engine/compiler/src/test/resources/ffi_logical_plan_2.json
index 4fb9a4194f2a..371981b607c2 100644
--- a/interactive_engine/compiler/src/test/resources/ffi_logical_plan_2.json
+++ b/interactive_engine/compiler/src/test/resources/ffi_logical_plan_2.json
@@ -131,6 +131,25 @@
           }
         }
       },
+      "children": [
+        1
+      ]
+    },
+    {
+      "opr": {
+        "opr": {
+          "Sink": {
+            "tags": [],
+            "sink_target": {
+              "inner": {
+                "SinkDefault": {
+                  "id_name_mappings": []
+                }
+              }
+            }
+          }
+        }
+      },
       "children": []
     }
   ],
diff --git a/interactive_engine/compiler/src/test/resources/ffi_logical_plan_3.json b/interactive_engine/compiler/src/test/resources/ffi_logical_plan_3.json
index f070d01e5171..b461958aef4f 100644
--- a/interactive_engine/compiler/src/test/resources/ffi_logical_plan_3.json
+++ b/interactive_engine/compiler/src/test/resources/ffi_logical_plan_3.json
@@ -259,6 +259,25 @@
           }
         }
       },
+      "children": [
+        3
+      ]
+    },
+    {
+      "opr": {
+        "opr": {
+          "Sink": {
+            "tags": [],
+            "sink_target": {
+              "inner": {
+                "SinkDefault": {
+                  "id_name_mappings": []
+                }
+              }
+            }
+          }
+        }
+      },
       "children": []
     }
   ],
diff --git a/interactive_engine/compiler/src/test/resources/get_bothV.json b/interactive_engine/compiler/src/test/resources/get_bothV.json
deleted file mode 100644
index 66963199a948..000000000000
--- a/interactive_engine/compiler/src/test/resources/get_bothV.json
+++ /dev/null
@@ -1,29 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Vertex": {
-            "tag": null,
-            "opt": 3,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "alias": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/group.json b/interactive_engine/compiler/src/test/resources/group.json
deleted file mode 100644
index c7d2ad7b0d2f..000000000000
--- a/interactive_engine/compiler/src/test/resources/group.json
+++ /dev/null
@@ -1,48 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "GroupBy": {
-            "mappings": [
-              {
-                "key": {
-                  "tag": null,
-                  "property": null,
-                  "node_type": null
-                },
-                "alias": {
-                  "item": {
-                    "Id": 0
-                  }
-                }
-              }
-            ],
-            "functions": [
-              {
-                "vars": [
-                  {
-                    "tag": null,
-                    "property": null,
-                    "node_type": null
-                  }
-                ],
-                "aggregate": 5,
-                "alias": {
-                  "item": {
-                    "Id": 1
-                  }
-                }
-              }
-            ],
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/group_key.json b/interactive_engine/compiler/src/test/resources/group_key.json
deleted file mode 100644
index 8eb15f24fc67..000000000000
--- a/interactive_engine/compiler/src/test/resources/group_key.json
+++ /dev/null
@@ -1,50 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "GroupBy": {
-            "mappings": [
-              {
-                "key": {
-                  "tag": null,
-                  "property": {
-                    "item": {
-                      "Key": {
-                        "item": {
-                          "Name": "name"
-                        }
-                      }
-                    }
-                  },
-                  "node_type": null
-                },
-                "alias": {
-                  "item": {
-                    "Name": "keys_name"
-                  }
-                }
-              }
-            ],
-            "functions": [
-              {
-                "vars": [],
-                "aggregate": 5,
-                "alias": {
-                  "item": {
-                    "Name": "values"
-                  }
-                }
-              }
-            ],
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/group_key_count.json b/interactive_engine/compiler/src/test/resources/group_key_count.json
deleted file mode 100644
index b186de3487f6..000000000000
--- a/interactive_engine/compiler/src/test/resources/group_key_count.json
+++ /dev/null
@@ -1,50 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "GroupBy": {
-            "mappings": [
-              {
-                "key": {
-                  "tag": null,
-                  "property": {
-                    "item": {
-                      "Key": {
-                        "item": {
-                          "Name": "name"
-                        }
-                      }
-                    }
-                  },
-                  "node_type": null
-                },
-                "alias": {
-                  "item": {
-                    "Name": "keys_name"
-                  }
-                }
-              }
-            ],
-            "functions": [
-              {
-                "vars": [],
-                "aggregate": 3,
-                "alias": {
-                  "item": {
-                    "Name": "values"
-                  }
-                }
-              }
-            ],
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/group_key_out_count.json b/interactive_engine/compiler/src/test/resources/group_key_out_count.json
deleted file mode 100644
index 3cfc414a8f8d..000000000000
--- a/interactive_engine/compiler/src/test/resources/group_key_out_count.json
+++ /dev/null
@@ -1,119 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": null,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": [
-        2
-      ]
-    },
-    {
-      "opr": {
-        "opr": {
-          "Edge": {
-            "v_tag": null,
-            "direction": 0,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "alias": null,
-            "expand_opt": 2,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    },
-    {
-      "opr": {
-        "opr": {
-          "Apply": {
-            "join_kind": 0,
-            "tags": [],
-            "subtask": 1,
-            "alias": {
-              "item": {
-                "Id": 0
-              }
-            }
-          }
-        }
-      },
-      "children": [
-        3
-      ]
-    },
-    {
-      "opr": {
-        "opr": {
-          "GroupBy": {
-            "mappings": [
-              {
-                "key": {
-                  "tag": {
-                    "item": {
-                      "Id": 0
-                    }
-                  },
-                  "property": null,
-                  "node_type": null
-                },
-                "alias": {
-                  "item": {
-                    "Id": 0
-                  }
-                }
-              }
-            ],
-            "functions": [
-              {
-                "vars": [
-                  {
-                    "tag": null,
-                    "property": null,
-                    "node_type": null
-                  }
-                ],
-                "aggregate": 5,
-                "alias": {
-                  "item": {
-                    "Id": 1
-                  }
-                }
-              }
-            ],
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0,
-    1
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/group_key_out_has_count.json b/interactive_engine/compiler/src/test/resources/group_key_out_has_count.json
deleted file mode 100644
index c736c044007d..000000000000
--- a/interactive_engine/compiler/src/test/resources/group_key_out_has_count.json
+++ /dev/null
@@ -1,156 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": null,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": [
-        2
-      ]
-    },
-    {
-      "opr": {
-        "opr": {
-          "Edge": {
-            "v_tag": null,
-            "direction": 0,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": {
-                "operators": [
-                  {
-                    "node_type": null,
-                    "item": {
-                      "Var": {
-                        "tag": null,
-                        "property": {
-                          "item": {
-                            "Key": {
-                              "item": {
-                                "Name": "name"
-                              }
-                            }
-                          }
-                        },
-                        "node_type": null
-                      }
-                    }
-                  },
-                  {
-                    "node_type": null,
-                    "item": {
-                      "Logical": 0
-                    }
-                  },
-                  {
-                    "node_type": null,
-                    "item": {
-                      "Const": {
-                        "item": {
-                          "Str": "marko"
-                        }
-                      }
-                    }
-                  }
-                ]
-              },
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "alias": null,
-            "expand_opt": 2,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    },
-    {
-      "opr": {
-        "opr": {
-          "Apply": {
-            "join_kind": 0,
-            "tags": [],
-            "subtask": 1,
-            "alias": {
-              "item": {
-                "Id": 0
-              }
-            }
-          }
-        }
-      },
-      "children": [
-        3
-      ]
-    },
-    {
-      "opr": {
-        "opr": {
-          "GroupBy": {
-            "mappings": [
-              {
-                "key": {
-                  "tag": {
-                    "item": {
-                      "Id": 0
-                    }
-                  },
-                  "property": null,
-                  "node_type": null
-                },
-                "alias": {
-                  "item": {
-                    "Id": 0
-                  }
-                }
-              }
-            ],
-            "functions": [
-              {
-                "vars": [
-                  {
-                    "tag": null,
-                    "property": null,
-                    "node_type": null
-                  }
-                ],
-                "aggregate": 5,
-                "alias": {
-                  "item": {
-                    "Id": 1
-                  }
-                }
-              }
-            ],
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0,
-    1
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/limit_range.json b/interactive_engine/compiler/src/test/resources/limit_range.json
deleted file mode 100644
index 4d8068a9b2b8..000000000000
--- a/interactive_engine/compiler/src/test/resources/limit_range.json
+++ /dev/null
@@ -1,20 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Limit": {
-            "range": {
-              "lower": 1,
-              "upper": 2
-            }
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/order_asc.json b/interactive_engine/compiler/src/test/resources/order_asc.json
deleted file mode 100644
index 9ff8094d371b..000000000000
--- a/interactive_engine/compiler/src/test/resources/order_asc.json
+++ /dev/null
@@ -1,27 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "OrderBy": {
-            "pairs": [
-              {
-                "key": {
-                  "tag": null,
-                  "property": null,
-                  "node_type": null
-                },
-                "order": 1
-              }
-            ],
-            "limit": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/order_key.json b/interactive_engine/compiler/src/test/resources/order_key.json
deleted file mode 100644
index a4c4ed74ad03..000000000000
--- a/interactive_engine/compiler/src/test/resources/order_key.json
+++ /dev/null
@@ -1,35 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "OrderBy": {
-            "pairs": [
-              {
-                "key": {
-                  "tag": null,
-                  "property": {
-                    "item": {
-                      "Key": {
-                        "item": {
-                          "Name": "name"
-                        }
-                      }
-                    }
-                  },
-                  "node_type": null
-                },
-                "order": 1
-              }
-            ],
-            "limit": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/order_keys.json b/interactive_engine/compiler/src/test/resources/order_keys.json
deleted file mode 100644
index 9410b1829c58..000000000000
--- a/interactive_engine/compiler/src/test/resources/order_keys.json
+++ /dev/null
@@ -1,51 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "OrderBy": {
-            "pairs": [
-              {
-                "key": {
-                  "tag": null,
-                  "property": {
-                    "item": {
-                      "Key": {
-                        "item": {
-                          "Name": "name"
-                        }
-                      }
-                    }
-                  },
-                  "node_type": null
-                },
-                "order": 1
-              },
-              {
-                "key": {
-                  "tag": null,
-                  "property": {
-                    "item": {
-                      "Key": {
-                        "item": {
-                          "Name": "id"
-                        }
-                      }
-                    }
-                  },
-                  "node_type": null
-                },
-                "order": 2
-              }
-            ],
-            "limit": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/order_label.json b/interactive_engine/compiler/src/test/resources/order_label.json
deleted file mode 100644
index 44d10e7642a1..000000000000
--- a/interactive_engine/compiler/src/test/resources/order_label.json
+++ /dev/null
@@ -1,31 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "OrderBy": {
-            "pairs": [
-              {
-                "key": {
-                  "tag": null,
-                  "property": {
-                    "item": {
-                      "Label": {}
-                    }
-                  },
-                  "node_type": null
-                },
-                "order": 1
-              }
-            ],
-            "limit": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/order_limit.json b/interactive_engine/compiler/src/test/resources/order_limit.json
deleted file mode 100644
index 7b16ec87e1de..000000000000
--- a/interactive_engine/compiler/src/test/resources/order_limit.json
+++ /dev/null
@@ -1,30 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "OrderBy": {
-            "pairs": [
-              {
-                "key": {
-                  "tag": null,
-                  "property": null,
-                  "node_type": null
-                },
-                "order": 1
-              }
-            ],
-            "limit": {
-              "lower": 1,
-              "upper": 2
-            }
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/path_expand.json b/interactive_engine/compiler/src/test/resources/path_expand.json
deleted file mode 100644
index 3346b2d01708..000000000000
--- a/interactive_engine/compiler/src/test/resources/path_expand.json
+++ /dev/null
@@ -1,44 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Path": {
-            "base": {
-              "edge_expand": {
-                "v_tag": null,
-                "direction": 0,
-                "params": {
-                  "tables": [],
-                  "columns": [],
-                  "is_all_columns": false,
-                  "limit": null,
-                  "predicate": null,
-                  "sample_ratio": 1.0,
-                  "extra": {}
-                },
-                "alias": null,
-                "expand_opt": 0,
-                "meta_data": null
-              },
-              "get_v": null
-            },
-            "start_tag": null,
-            "alias": null,
-            "hop_range": {
-              "lower": 1,
-              "upper": 5
-            },
-            "path_opt": 1,
-            "result_opt": 1,
-            "condition": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/project_key.json b/interactive_engine/compiler/src/test/resources/project_key.json
deleted file mode 100644
index f6e92b7041a0..000000000000
--- a/interactive_engine/compiler/src/test/resources/project_key.json
+++ /dev/null
@@ -1,48 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Project": {
-            "mappings": [
-              {
-                "expr": {
-                  "operators": [
-                    {
-                      "item": {
-                        "Var": {
-                          "tag": null,
-                          "property": {
-                            "item": {
-                              "Key": {
-                                "item": {
-                                  "Name": "name"
-                                }
-                              }
-                            }
-                          },
-                          "node_type": null
-                        }
-                      }
-                    }
-                  ]
-                },
-                "alias": {
-                  "item": {
-                    "Name": "name"
-                  }
-                }
-              }
-            ],
-            "is_append": true,
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/project_tag.json b/interactive_engine/compiler/src/test/resources/project_tag.json
deleted file mode 100644
index 5a0789441d84..000000000000
--- a/interactive_engine/compiler/src/test/resources/project_tag.json
+++ /dev/null
@@ -1,44 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Project": {
-            "mappings": [
-              {
-                "expr": {
-                  "operators": [
-                    {
-                      "item": {
-                        "Var": {
-                          "tag": {
-                            "item": {
-                              "Name": "a"
-                            }
-                          },
-                          "property": null,
-                          "node_type": null
-                        }
-                      }
-                    }
-                  ]
-                },
-                "alias": {
-                  "item": {
-                    "Name": "project_a"
-                  }
-                }
-              }
-            ],
-            "is_append": true,
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/project_tag_key.json b/interactive_engine/compiler/src/test/resources/project_tag_key.json
deleted file mode 100644
index 6b9da770064f..000000000000
--- a/interactive_engine/compiler/src/test/resources/project_tag_key.json
+++ /dev/null
@@ -1,81 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": {
-              "item": {
-                "Id": 0
-              }
-            },
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": [
-        1
-      ]
-    },
-    {
-      "opr": {
-        "opr": {
-          "Project": {
-            "mappings": [
-              {
-                "expr": {
-                  "operators": [
-                    {
-                      "node_type": null,
-                      "item": {
-                        "Var": {
-                          "tag": {
-                            "item": {
-                              "Id": 0
-                            }
-                          },
-                          "property": {
-                            "item": {
-                              "Key": {
-                                "item": {
-                                  "Name": "name"
-                                }
-                              }
-                            }
-                          },
-                          "node_type": null
-                        }
-                      }
-                    }
-                  ]
-                },
-                "alias": {
-                  "item": {
-                    "Id": 1
-                  }
-                }
-              }
-            ],
-            "is_append": true,
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/project_tag_keys.json b/interactive_engine/compiler/src/test/resources/project_tag_keys.json
deleted file mode 100644
index a1def82b11ea..000000000000
--- a/interactive_engine/compiler/src/test/resources/project_tag_keys.json
+++ /dev/null
@@ -1,102 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": {
-              "item": {
-                "Id": 0
-              }
-            },
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": [
-        1
-      ]
-    },
-    {
-      "opr": {
-        "opr": {
-          "Project": {
-            "mappings": [
-              {
-                "expr": {
-                  "operators": [
-                    {
-                      "node_type": null,
-                      "item": {
-                        "VarMap": {
-                          "keys": [
-                            {
-                              "tag": {
-                                "item": {
-                                  "Id": 0
-                                }
-                              },
-                              "property": {
-                                "item": {
-                                  "Key": {
-                                    "item": {
-                                      "Name": "name"
-                                    }
-                                  }
-                                }
-                              },
-                              "node_type": null
-                            },
-                            {
-                              "tag": {
-                                "item": {
-                                  "Id": 0
-                                }
-                              },
-                              "property": {
-                                "item": {
-                                  "Key": {
-                                    "item": {
-                                      "Name": "id"
-                                    }
-                                  }
-                                }
-                              },
-                              "node_type": null
-                            }
-                          ]
-                        }
-                      }
-                    }
-                  ]
-                },
-                "alias": {
-                  "item": {
-                    "Id": 1
-                  }
-                }
-              }
-            ],
-            "is_append": true,
-            "meta_data": []
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/scan_alias.json b/interactive_engine/compiler/src/test/resources/scan_alias.json
deleted file mode 100644
index 34a88b4e98cb..000000000000
--- a/interactive_engine/compiler/src/test/resources/scan_alias.json
+++ /dev/null
@@ -1,33 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": {
-              "item": {
-                "Id": 0
-              }
-            },
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/scan_expr.json b/interactive_engine/compiler/src/test/resources/scan_expr.json
deleted file mode 100644
index e1e8475324f5..000000000000
--- a/interactive_engine/compiler/src/test/resources/scan_expr.json
+++ /dev/null
@@ -1,66 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": null,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": {
-                "operators": [
-                  {
-                    "node_type": null,
-                    "item": {
-                      "Var": {
-                        "tag": null,
-                        "property": {
-                          "item": {
-                            "Key": {
-                              "item": {
-                                "Name": "id"
-                              }
-                            }
-                          }
-                        },
-                        "node_type": null
-                      }
-                    }
-                  },
-                  {
-                    "node_type": null,
-                    "item": {
-                      "Logical": 0
-                    }
-                  },
-                  {
-                    "node_type": null,
-                    "item": {
-                      "Const": {
-                        "item": {
-                          "I64": 1
-                        }
-                      }
-                    }
-                  }
-                ]
-              },
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/scan_ids.json b/interactive_engine/compiler/src/test/resources/scan_ids.json
deleted file mode 100644
index e7decae1b91a..000000000000
--- a/interactive_engine/compiler/src/test/resources/scan_ids.json
+++ /dev/null
@@ -1,66 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": null,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": {
-              "or_predicates": [
-                {
-                  "predicates": [
-                    {
-                      "key": {
-                        "item": {
-                          "Id": {}
-                        }
-                      },
-                      "value": {
-                        "item": {
-                          "I64": 1
-                        }
-                      },
-                      "cmp": null
-                    }
-                  ]
-                },
-                {
-                  "predicates": [
-                    {
-                      "key": {
-                        "item": {
-                          "Id": {}
-                        }
-                      },
-                      "value": {
-                        "item": {
-                          "I64": 2
-                        }
-                      },
-                      "cmp": null
-                    }
-                  ]
-                }
-              ]
-            },
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/scan_labels.json b/interactive_engine/compiler/src/test/resources/scan_labels.json
deleted file mode 100644
index 40291880c008..000000000000
--- a/interactive_engine/compiler/src/test/resources/scan_labels.json
+++ /dev/null
@@ -1,35 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": null,
-            "params": {
-              "tables": [
-                {
-                  "item": {
-                    "Id": 0
-                  }
-                }
-              ],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/scan_opt.json b/interactive_engine/compiler/src/test/resources/scan_opt.json
deleted file mode 100644
index 913ecb4b9289..000000000000
--- a/interactive_engine/compiler/src/test/resources/scan_opt.json
+++ /dev/null
@@ -1,29 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": null,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {}
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/scan_snapshot.json b/interactive_engine/compiler/src/test/resources/scan_snapshot.json
deleted file mode 100644
index 7906a1bfcbe9..000000000000
--- a/interactive_engine/compiler/src/test/resources/scan_snapshot.json
+++ /dev/null
@@ -1,31 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Scan": {
-            "scan_opt": 0,
-            "alias": null,
-            "params": {
-              "tables": [],
-              "columns": [],
-              "is_all_columns": false,
-              "limit": null,
-              "predicate": null,
-              "sample_ratio": 1.0,
-              "extra": {
-                "SID": "1"
-              }
-            },
-            "idx_predicate": null,
-            "meta_data": null
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/compiler/src/test/resources/select_expr.json b/interactive_engine/compiler/src/test/resources/select_expr.json
deleted file mode 100644
index d2dc4369dad4..000000000000
--- a/interactive_engine/compiler/src/test/resources/select_expr.json
+++ /dev/null
@@ -1,94 +0,0 @@
-{
-  "nodes": [
-    {
-      "opr": {
-        "opr": {
-          "Select": {
-            "predicate": {
-              "operators": [
-                {
-                  "node_type": null,
-                  "item": {
-                    "Var": {
-                      "tag": null,
-                      "property": {
-                        "item": {
-                          "Key": {
-                            "item": {
-                              "Name": "id"
-                            }
-                          }
-                        }
-                      },
-                      "node_type": null
-                    }
-                  }
-                },
-                {
-                  "node_type": null,
-                  "item": {
-                    "Logical": 0
-                  }
-                },
-                {
-                  "node_type": null,
-                  "item": {
-                    "Const": {
-                      "item": {
-                        "I64": 1
-                      }
-                    }
-                  }
-                },
-                {
-                  "node_type": null,
-                  "item": {
-                    "Logical": 10
-                  }
-                },
-                {
-                  "node_type": null,
-                  "item": {
-                    "Var": {
-                      "tag": null,
-                      "property": {
-                        "item": {
-                          "Key": {
-                            "item": {
-                              "Name": "name"
-                            }
-                          }
-                        }
-                      },
-                      "node_type": null
-                    }
-                  }
-                },
-                {
-                  "node_type": null,
-                  "item": {
-                    "Logical": 0
-                  }
-                },
-                {
-                  "node_type": null,
-                  "item": {
-                    "Const": {
-                      "item": {
-                        "Str": "marko"
-                      }
-                    }
-                  }
-                }
-              ]
-            }
-          }
-        }
-      },
-      "children": []
-    }
-  ],
-  "roots": [
-    0
-  ]
-}
diff --git a/interactive_engine/executor/ir/core/tests/test_multi_source.rs b/interactive_engine/executor/ir/core/tests/test_multi_source.rs
new file mode 100644
index 000000000000..7a5f3d278157
--- /dev/null
+++ b/interactive_engine/executor/ir/core/tests/test_multi_source.rs
@@ -0,0 +1,481 @@
+//
+//! Copyright 2023 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+//!
+
+mod common;
+
+#[cfg(test)]
+mod tests {
+    use std::vec;
+
+    use ir_common::generated::algebra as pb;
+    use ir_common::generated::common as common_pb;
+    use ir_core::plan::logical::LogicalPlan;
+    use ir_core::plan::physical::AsPhysical;
+    use ir_physical_client::physical_builder::PlanBuilder;
+
+    fn default_sink_pb() -> pb::Sink {
+        pb::Sink {
+            tags: vec![common_pb::NameOrIdKey { key: None }],
+            sink_target: Some(pb::sink::SinkTarget {
+                inner: Some(pb::sink::sink_target::Inner::SinkDefault(pb::SinkDefault {
+                    id_name_mappings: vec![],
+                })),
+            }),
+        }
+    }
+
+    // join(scan.match1, scan.match2)
+    fn single_source_multi_match_join_logical_plan() -> LogicalPlan {
+        let scan_opr =
+            pb::Scan { scan_opt: 0, alias: None, params: None, idx_predicate: None, meta_data: None };
+
+        let expand_opr1 = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0, // out
+            params: None,
+            expand_opt: 0,
+            alias: None,
+            meta_data: None,
+        };
+
+        let expand_opr2 = expand_opr1.clone();
+
+        // build pattern 1: as('a').out().as('b')
+        let left_pattern = pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some("a".into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(expand_opr1)),
+                }],
+                end: Some("b".into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        };
+
+        // build pattern 2: as('c').out().as('b')
+        let right_pattern = pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some("c".into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(expand_opr2)),
+                }],
+                end: Some("b".into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        };
+
+        let join_opr = pb::Join {
+            kind: 0,
+            left_keys: vec![common_pb::Variable { tag: Some("b".into()), property: None, node_type: None }],
+            right_keys: vec![common_pb::Variable {
+                tag: Some("b".into()),
+                property: None,
+                node_type: None,
+            }],
+        };
+
+        let sink = default_sink_pb();
+
+        let mut plan = LogicalPlan::default();
+        let scan_id = plan
+            .append_operator_as_node(scan_opr.into(), vec![0])
+            .unwrap();
+        let left_match_id = plan
+            .append_operator_as_node(left_pattern.into(), vec![scan_id])
+            .unwrap();
+        let right_match_id = plan
+            .append_operator_as_node(right_pattern.into(), vec![scan_id])
+            .unwrap();
+        let join_id = plan
+            .append_operator_as_node(join_opr.into(), vec![left_match_id, right_match_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        plan
+    }
+
+    // join(scan1.match1, scan2.match2)
+    fn multi_source_multi_match_join_logical_plan() -> LogicalPlan {
+        let scan_opr =
+            pb::Scan { scan_opt: 0, alias: None, params: None, idx_predicate: None, meta_data: None };
+
+        let expand_opr1 = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0, // out
+            params: None,
+            expand_opt: 0,
+            alias: None,
+            meta_data: None,
+        };
+
+        let expand_opr2 = expand_opr1.clone();
+
+        // build pattern 1: as('a').out().as('b')
+        let left_pattern = pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some("a".into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(expand_opr1)),
+                }],
+                end: Some("b".into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        };
+
+        // build pattern 2: as('c').out().as('b')
+        let right_pattern = pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some("c".into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(expand_opr2)),
+                }],
+                end: Some("b".into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        };
+
+        let join_opr = pb::Join {
+            kind: 0,
+            left_keys: vec![common_pb::Variable { tag: Some("b".into()), property: None, node_type: None }],
+            right_keys: vec![common_pb::Variable {
+                tag: Some("b".into()),
+                property: None,
+                node_type: None,
+            }],
+        };
+
+        let sink = default_sink_pb();
+
+        let mut plan = LogicalPlan::default();
+        let left_scan_id = plan
+            .append_operator_as_node(scan_opr.clone().into(), vec![0])
+            .unwrap();
+        let left_match_id = plan
+            .append_operator_as_node(left_pattern.into(), vec![left_scan_id])
+            .unwrap();
+        let right_scan_id = plan
+            .append_operator_as_node(scan_opr.into(), vec![0])
+            .unwrap();
+        let right_match_id = plan
+            .append_operator_as_node(right_pattern.into(), vec![right_scan_id])
+            .unwrap();
+        let join_id = plan
+            .append_operator_as_node(join_opr.into(), vec![left_match_id, right_match_id])
+            .unwrap();
+
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        plan
+    }
+
+    // join(join(scan1.match1, scan2.match2), scan3.match3)
+    fn multi_source_multi_match_multi_join_logical_plan() -> LogicalPlan {
+        let scan_opr =
+            pb::Scan { scan_opt: 0, alias: None, params: None, idx_predicate: None, meta_data: None };
+
+        let expand_opr = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0, // out
+            params: None,
+            expand_opt: 0,
+            alias: None,
+            meta_data: None,
+        };
+
+        // build pattern 1: as('a').out().as('b')
+        let left_pattern = pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some("a".into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(expand_opr.clone())),
+                }],
+                end: Some("b".into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        };
+
+        // build pattern 2: as('c').out().as('b')
+        let right_pattern = pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some("c".into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(expand_opr.clone())),
+                }],
+                end: Some("b".into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        };
+
+        // build pattern 3: as('d').out().as('b')
+        let third_pattern = pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some("d".into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(expand_opr.clone())),
+                }],
+                end: Some("b".into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        };
+
+        let join_opr = pb::Join {
+            kind: 0,
+            left_keys: vec![common_pb::Variable { tag: Some("b".into()), property: None, node_type: None }],
+            right_keys: vec![common_pb::Variable {
+                tag: Some("b".into()),
+                property: None,
+                node_type: None,
+            }],
+        };
+
+        let sink = default_sink_pb();
+
+        let mut plan = LogicalPlan::default();
+        let left_scan_id = plan
+            .append_operator_as_node(scan_opr.clone().into(), vec![0])
+            .unwrap();
+        let left_match_id = plan
+            .append_operator_as_node(left_pattern.into(), vec![left_scan_id])
+            .unwrap();
+        let right_scan_id = plan
+            .append_operator_as_node(scan_opr.clone().into(), vec![0])
+            .unwrap();
+        let right_match_id = plan
+            .append_operator_as_node(right_pattern.into(), vec![right_scan_id])
+            .unwrap();
+        let join_id = plan
+            .append_operator_as_node(join_opr.clone().into(), vec![left_match_id, right_match_id])
+            .unwrap();
+
+        let third_scan_id = plan
+            .append_operator_as_node(scan_opr.into(), vec![0])
+            .unwrap();
+        let third_match_id = plan
+            .append_operator_as_node(third_pattern.into(), vec![third_scan_id])
+            .unwrap();
+        let second_join_id = plan
+            .append_operator_as_node(join_opr.into(), vec![join_id, third_match_id])
+            .unwrap();
+
+        plan.append_operator_as_node(sink.into(), vec![second_join_id])
+            .unwrap();
+
+        plan
+    }
+
+    #[test]
+    fn test_sinlge_source_multi_match_join_logical() {
+        let plan = single_source_multi_match_join_logical_plan();
+        // dummy, scan1, expand1, getv1, scan2, expand2, getv2, join1, sink
+        assert_eq!(plan.len(), 9);
+    }
+
+    #[test]
+    fn test_sinlge_source_multi_match_join_physical() {
+        let plan = single_source_multi_match_join_logical_plan();
+
+        let mut plan_meta = plan.get_plan_meta();
+        let mut builder = PlanBuilder::default();
+        let _ = plan.add_job_builder(&mut builder, &mut plan_meta);
+        let physical_plan = builder.take();
+        // dummy, join, sink
+        assert_eq!(physical_plan.len(), 3);
+    }
+
+    #[test]
+    fn test_multi_source_multi_match_join_logical() {
+        let plan = multi_source_multi_match_join_logical_plan();
+        // dummy, scan1, expand1, getv1, scan2, expand2, getv2, join1, sink
+        assert_eq!(plan.len(), 9);
+    }
+
+    #[test]
+    fn test_multi_source_multi_match_join_physical() {
+        let plan = multi_source_multi_match_join_logical_plan();
+
+        let mut plan_meta = plan.get_plan_meta();
+        let mut builder = PlanBuilder::default();
+        let _ = plan.add_job_builder(&mut builder, &mut plan_meta);
+        let physical_plan = builder.take();
+        // dummy, join, sink
+        assert_eq!(physical_plan.len(), 3);
+    }
+
+    #[test]
+    fn test_multi_source_multi_match_multi_join_logical() {
+        let plan = multi_source_multi_match_multi_join_logical_plan();
+        // dummy, scan1, expand1, getv1, scan2, expand2, getv2, join1, scan3, expand3, getv3, join2, sink
+        assert_eq!(plan.len(), 13);
+    }
+
+    #[test]
+    #[ignore]
+    // TODO: this test would be fixed when multiple joins is supported
+    fn test_multi_source_multi_match_multi_join_physical() {
+        let plan = multi_source_multi_match_multi_join_logical_plan();
+
+        let mut plan_meta = plan.get_plan_meta();
+        let mut builder = PlanBuilder::default();
+        let _res = plan.add_job_builder(&mut builder, &mut plan_meta);
+        let physical_plan = builder.take();
+        // dummy, join1, join2, sink
+        assert_eq!(physical_plan.len(), 4);
+    }
+
+    // join(join(scan1.match1, scan2.match2), scan3.match3)
+    fn multi_join_logical_plan() -> LogicalPlan {
+        let scan_opr =
+            pb::Scan { scan_opt: 0, alias: None, params: None, idx_predicate: None, meta_data: None };
+        let join_opr = pb::Join {
+            kind: 0,
+            left_keys: vec![common_pb::Variable { tag: None, property: None, node_type: None }],
+            right_keys: vec![common_pb::Variable { tag: None, property: None, node_type: None }],
+        };
+
+        let sink = default_sink_pb();
+
+        let mut plan = LogicalPlan::default();
+        let scan1_id = plan
+            .append_operator_as_node(scan_opr.clone().into(), vec![0])
+            .unwrap();
+        let scan2_id = plan
+            .append_operator_as_node(scan_opr.clone().into(), vec![0])
+            .unwrap();
+        let join1_id = plan
+            .append_operator_as_node(join_opr.clone().into(), vec![scan1_id, scan2_id])
+            .unwrap();
+        let scan3_id = plan
+            .append_operator_as_node(scan_opr.clone().into(), vec![0])
+            .unwrap();
+        let join2_id = plan
+            .append_operator_as_node(join_opr.clone().into(), vec![join1_id, scan3_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join2_id])
+            .unwrap();
+
+        plan
+    }
+
+    #[test]
+    fn test_multi_join_logical() {
+        let plan = multi_join_logical_plan();
+        // dummy, scan1, scan2,  join1, scan3, join2, sink
+        assert_eq!(plan.len(), 7);
+    }
+
+    #[test]
+    #[ignore]
+    // TODO: this test would be fixed when multiple joins is supported
+    fn test_multi_join_physical() {
+        let plan = multi_join_logical_plan();
+
+        let mut plan_meta = plan.get_plan_meta();
+        let mut builder = PlanBuilder::default();
+        let res = plan.add_job_builder(&mut builder, &mut plan_meta);
+        assert!(res.is_ok());
+        let physical_plan = builder.take();
+        // dummy, join2, sink
+        assert_eq!(physical_plan.len(), 3);
+    }
+
+    // join(join(scan1.match1, scan2.match2), scan3.match3) with branch op
+    fn multi_join_with_branch_logical_plan() -> LogicalPlan {
+        let scan1_opr = pb::Scan {
+            scan_opt: 0,
+            alias: Some(0.into()),
+            params: None,
+            idx_predicate: None,
+            meta_data: None,
+        };
+        let scan2_opr = pb::Scan {
+            scan_opt: 0,
+            alias: Some(1.into()),
+            params: None,
+            idx_predicate: None,
+            meta_data: None,
+        };
+        let scan3_opr = pb::Scan {
+            scan_opt: 0,
+            alias: Some(2.into()),
+            params: None,
+            idx_predicate: None,
+            meta_data: None,
+        };
+        let dummy_opr = pb::Root {};
+        let join_opr = pb::Join {
+            kind: 0,
+            left_keys: vec![common_pb::Variable { tag: None, property: None, node_type: None }],
+            right_keys: vec![common_pb::Variable { tag: None, property: None, node_type: None }],
+        };
+
+        let sink = default_sink_pb();
+
+        let mut plan = LogicalPlan::default();
+        let branch2_id = plan
+            .append_operator_as_node(dummy_opr.clone().into(), vec![0])
+            .unwrap();
+        let branch1_id = plan
+            .append_operator_as_node(dummy_opr.into(), vec![branch2_id])
+            .unwrap();
+        let scan1_id = plan
+            .append_operator_as_node(scan1_opr.clone().into(), vec![branch1_id])
+            .unwrap();
+        let scan2_id = plan
+            .append_operator_as_node(scan2_opr.clone().into(), vec![branch1_id])
+            .unwrap();
+        let join1_id = plan
+            .append_operator_as_node(join_opr.clone().into(), vec![scan1_id, scan2_id])
+            .unwrap();
+        let scan3_id = plan
+            .append_operator_as_node(scan3_opr.clone().into(), vec![branch2_id])
+            .unwrap();
+        let join2_id = plan
+            .append_operator_as_node(join_opr.clone().into(), vec![join1_id, scan3_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join2_id])
+            .unwrap();
+
+        plan
+    }
+
+    #[test]
+    fn test_multi_join_with_branch_logical() {
+        let plan = multi_join_with_branch_logical_plan();
+        //  branch2,  branch1, scan1, scan2,  join1, scan3, join2, sink
+        assert_eq!(plan.len(), 8);
+    }
+
+    #[test]
+    fn test_multi_join_with_branch_physical() {
+        let plan = multi_join_with_branch_logical_plan();
+        let mut plan_meta = plan.get_plan_meta();
+        let mut builder = PlanBuilder::default();
+        let res = plan.add_job_builder(&mut builder, &mut plan_meta);
+        assert!(res.is_ok());
+        let physical_plan = builder.take();
+        // dummy, join2, sink
+        assert_eq!(physical_plan.len(), 3);
+    }
+}
diff --git a/interactive_engine/executor/ir/integrated/tests/match_test.rs b/interactive_engine/executor/ir/integrated/tests/match_test.rs
index 564db11d1e29..a4adabf8ae55 100644
--- a/interactive_engine/executor/ir/integrated/tests/match_test.rs
+++ b/interactive_engine/executor/ir/integrated/tests/match_test.rs
@@ -25,6 +25,7 @@ mod test {
     use ir_common::expr_parse::str_to_expr_pb;
     use ir_common::generated::algebra as pb;
     use ir_common::generated::common as common_pb;
+    use ir_common::KeyId;
     use ir_core::plan::logical::LogicalPlan;
     use ir_core::plan::meta::set_schema_from_json;
     use ir_core::plan::physical::AsPhysical;
@@ -42,32 +43,60 @@ mod test {
         set_schema_from_json(ldbc_schema_file);
     }
 
-    // g.V().hasLabel("person").match(
-    //    __.as('a').has(age, gt(25)).out("created").as('b'),
-    //    __.as('a').out('knows').as('c'),
-    // )
-    fn init_match_case1_request(with_schema: bool) -> JobRequest {
-        if with_schema {
-            init_schema();
-        }
-        let source = pb::Scan {
+    fn build_job_request(plan: LogicalPlan) -> JobRequest {
+        let mut plan_builder = PlanBuilder::default();
+        let mut plan_meta = plan.get_meta().clone();
+        plan.add_job_builder(&mut plan_builder, &mut plan_meta)
+            .unwrap();
+        let job_builder = JobBuilder::with_plan(plan_builder);
+        job_builder.build().unwrap()
+    }
+
+    fn get_person_scan() -> pb::Scan {
+        pb::Scan {
             scan_opt: 0,
             alias: None,
             params: Some(query_params(vec![PERSON_LABEL.into()], vec![], None)),
             idx_predicate: None,
             meta_data: None,
-        };
+        }
+    }
+
+    fn get_out_edge() -> pb::EdgeExpand {
+        pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: None,
+            expand_opt: 0,
+            alias: None,
+            meta_data: None,
+        }
+    }
+
+    fn get_in_edge() -> pb::EdgeExpand {
+        pb::EdgeExpand {
+            v_tag: None,
+            direction: 1,
+            params: None,
+            expand_opt: 0,
+            alias: None,
+            meta_data: None,
+        }
+    }
 
-        let out_knows = pb::EdgeExpand {
+    fn get_out_knows() -> pb::EdgeExpand {
+        pb::EdgeExpand {
             v_tag: None,
             direction: 0,
             params: Some(query_params(vec![KNOWS_LABEL.into()], vec![], None)),
             expand_opt: 0,
             alias: None,
             meta_data: None,
-        };
+        }
+    }
 
-        let out_created = pb::EdgeExpand {
+    fn get_out_created() -> pb::EdgeExpand {
+        pb::EdgeExpand {
             v_tag: None,
             direction: 0,
             params: Some(query_params(
@@ -78,9 +107,35 @@ mod test {
             expand_opt: 0,
             alias: None,
             meta_data: None,
-        };
+        }
+    }
+
+    fn get_inner_join(tags: Vec<KeyId>) -> pb::Join {
+        let keys: Vec<common_pb::Variable> = tags
+            .into_iter()
+            .map(|tag| common_pb::Variable { tag: Some(tag.into()), property: None, node_type: None })
+            .collect();
+        pb::Join { left_keys: keys.clone(), right_keys: keys.clone(), kind: 0 }
+    }
+
+    fn get_sink(tags: Vec<KeyId>) -> pb::Sink {
+        pb::Sink {
+            tags: tags
+                .into_iter()
+                .map(|tag| common_pb::NameOrIdKey { key: Some(tag.into()) })
+                .collect(),
+            sink_target: default_sink_target(),
+        }
+    }
 
-        let pattern = pb::Pattern {
+    //    __.as('a').has(age, gt(25)).out("created").as('b'),
+    //    __.as('a').out('knows').as('c')
+    fn get_pattern_case1() -> pb::Pattern {
+        let out_knows = get_out_knows();
+
+        let out_created = get_out_created();
+
+        pb::Pattern {
             sentences: vec![
                 pb::pattern::Sentence {
                     start: Some(TAG_A.into()),
@@ -108,70 +163,18 @@ mod test {
                 },
             ],
             meta_data: vec![],
-        };
-
-        let sink = pb::Sink {
-            tags: vec![
-                common_pb::NameOrIdKey { key: Some(TAG_A.into()) },
-                common_pb::NameOrIdKey { key: Some(TAG_B.into()) },
-                common_pb::NameOrIdKey { key: Some(TAG_C.into()) },
-            ],
-            sink_target: default_sink_target(),
-        };
-
-        let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(source.into(), vec![])
-            .unwrap();
-        let id = plan
-            .append_operator_as_node(pattern.into(), vec![0])
-            .unwrap();
-        plan.append_operator_as_node(sink.into(), vec![id])
-            .unwrap();
-
-        let mut plan_builder = PlanBuilder::default();
-        let mut plan_meta = plan.get_meta().clone();
-        plan.add_job_builder(&mut plan_builder, &mut plan_meta)
-            .unwrap();
-        let job_builder = JobBuilder::with_plan(plan_builder);
-        job_builder.build().unwrap()
+        }
     }
 
-    // g.V().hasLabel("person").match(
     //    __.as('a').out("knows").as('b'),
     //    __.as('a').out('knows').as('c'),
     //    __.as('a').out('created').as('d')
-    // )
-    fn init_match_case2_request(with_schema: bool) -> JobRequest {
-        if with_schema {
-            init_schema();
-        }
-        let source = pb::Scan {
-            scan_opt: 0,
-            alias: None,
-            params: Some(query_params(vec![PERSON_LABEL.into()], vec![], None)),
-            idx_predicate: None,
-            meta_data: None,
-        };
+    fn get_pattern_case2() -> pb::Pattern {
+        let out_knows = get_out_knows();
 
-        let out_knows = pb::EdgeExpand {
-            v_tag: None,
-            direction: 0,
-            params: Some(query_params(vec![KNOWS_LABEL.into()], vec![], None)),
-            expand_opt: 0,
-            alias: None,
-            meta_data: None,
-        };
-
-        let out_created = pb::EdgeExpand {
-            v_tag: None,
-            direction: 0,
-            params: Some(query_params(vec![CREATED_LABEL.into()], vec![], None)),
-            expand_opt: 0,
-            alias: None,
-            meta_data: None,
-        };
+        let out_created = get_out_created();
 
-        let pattern = pb::Pattern {
+        pb::Pattern {
             sentences: vec![
                 pb::pattern::Sentence {
                     start: Some(TAG_A.into()),
@@ -199,38 +202,437 @@ mod test {
                 },
             ],
             meta_data: vec![],
-        };
-
-        let sink = pb::Sink {
-            tags: vec![
-                common_pb::NameOrIdKey { key: Some(TAG_A.into()) },
-                common_pb::NameOrIdKey { key: Some(TAG_B.into()) },
-                common_pb::NameOrIdKey { key: Some(TAG_C.into()) },
-                common_pb::NameOrIdKey { key: Some(TAG_D.into()) },
+        }
+    }
+
+    //    __.as('a').out("created").as('c'),
+    //    __.as('b').out('created').as('c'),
+    fn get_pattern_case3() -> pb::Pattern {
+        let out_created = get_out_created();
+
+        pb::Pattern {
+            sentences: vec![
+                pb::pattern::Sentence {
+                    start: Some(TAG_A.into()),
+                    binders: vec![pb::pattern::Binder {
+                        item: Some(pb::pattern::binder::Item::Edge(out_created.clone())),
+                    }],
+                    end: Some(TAG_C.into()),
+                    join_kind: 0,
+                },
+                pb::pattern::Sentence {
+                    start: Some(TAG_B.into()),
+                    binders: vec![pb::pattern::Binder {
+                        item: Some(pb::pattern::binder::Item::Edge(out_created.clone())),
+                    }],
+                    end: Some(TAG_C.into()),
+                    join_kind: 0,
+                },
             ],
-            sink_target: default_sink_target(),
-        };
+            meta_data: vec![],
+        }
+    }
+
+    //    __.as('a').out("knows").as('b'),
+    fn get_pattern_case4() -> pb::Pattern {
+        let out_knows = get_out_knows();
+
+        pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some(TAG_A.into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(out_knows.clone())),
+                }],
+                end: Some(TAG_B.into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        }
+    }
+
+    //    __.as('a').out().as('c'),
+    //    __.as('b').out().as('c'),
+    fn get_pattern_case5() -> pb::Pattern {
+        let out_edge = get_out_edge();
+
+        pb::Pattern {
+            sentences: vec![
+                pb::pattern::Sentence {
+                    start: Some(TAG_A.into()),
+                    binders: vec![pb::pattern::Binder {
+                        item: Some(pb::pattern::binder::Item::Edge(out_edge.clone())),
+                    }],
+                    end: Some(TAG_C.into()),
+                    join_kind: 0,
+                },
+                pb::pattern::Sentence {
+                    start: Some(TAG_B.into()),
+                    binders: vec![pb::pattern::Binder {
+                        item: Some(pb::pattern::binder::Item::Edge(out_edge.clone())),
+                    }],
+                    end: Some(TAG_C.into()),
+                    join_kind: 0,
+                },
+            ],
+            meta_data: vec![],
+        }
+    }
+
+    //    __.as('a').out().as('b'),
+    fn get_pattern_case6() -> pb::Pattern {
+        let out_edge = get_out_edge();
+
+        pb::Pattern {
+            sentences: vec![pb::pattern::Sentence {
+                start: Some(TAG_A.into()),
+                binders: vec![pb::pattern::Binder {
+                    item: Some(pb::pattern::binder::Item::Edge(out_edge.clone())),
+                }],
+                end: Some(TAG_B.into()),
+                join_kind: 0,
+            }],
+            meta_data: vec![],
+        }
+    }
+
+    // g.V().match(
+    //    __.as('a').has(age, gt(25)).out("created").as('b'),
+    //    __.as('a').out('knows').as('c'),
+    // )
+    // if define source, g.V().hasLabel('person').match(...)
+    fn init_match_case1_request(define_source: bool) -> JobRequest {
+        init_schema();
+        let source = if define_source { get_person_scan() } else { pb::Scan::default() };
+
+        let pattern = get_pattern_case1();
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C]);
 
         let mut plan = LogicalPlan::default();
-        plan.append_operator_as_node(source.into(), vec![])
+        let srouce_id = plan
+            .append_operator_as_node(source.into(), vec![0])
             .unwrap();
         let id = plan
-            .append_operator_as_node(pattern.into(), vec![0])
+            .append_operator_as_node(pattern.into(), vec![srouce_id])
             .unwrap();
         plan.append_operator_as_node(sink.into(), vec![id])
             .unwrap();
 
-        let mut plan_builder = PlanBuilder::default();
-        let mut plan_meta = plan.get_meta().clone();
-        plan.add_job_builder(&mut plan_builder, &mut plan_meta)
+        build_job_request(plan)
+    }
+
+    // g.V().match(
+    //    __.as('a').out("knows").as('b'),
+    //    __.as('a').out('knows').as('c'),
+    //    __.as('a').out('created').as('d')
+    // )
+    // if define source, g.V().hasLabel('person').match(...)
+    fn init_match_case2_request(define_source: bool) -> JobRequest {
+        init_schema();
+        let source = if define_source { get_person_scan() } else { pb::Scan::default() };
+
+        let pattern = get_pattern_case2();
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C, TAG_D]);
+
+        let mut plan = LogicalPlan::default();
+        let source_id = plan
+            .append_operator_as_node(source.into(), vec![0])
             .unwrap();
-        let job_builder = JobBuilder::with_plan(plan_builder);
-        job_builder.build().unwrap()
+        let id = plan
+            .append_operator_as_node(pattern.into(), vec![source_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![id])
+            .unwrap();
+
+        build_job_request(plan)
+    }
+
+    // g.V().match(
+    //    __.as('a').out("created").as('c'),
+    //    __.as('b').out('created').as('c'),
+    // ).join(
+    //    match(__.as('a').out("knows").as('b'))
+    // )
+    fn init_match_case3_request() -> JobRequest {
+        let source = pb::Scan::default();
+
+        let pattern_0 = get_pattern_case3();
+
+        let pattern_1 = get_pattern_case4();
+
+        let join = get_inner_join(vec![TAG_A, TAG_B]);
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C]);
+
+        let mut plan = LogicalPlan::default();
+        let source_id = plan
+            .append_operator_as_node(source.into(), vec![0])
+            .unwrap();
+
+        let pattern_0_id = plan
+            .append_operator_as_node(pattern_0.into(), vec![source_id])
+            .unwrap();
+        let pattern_1_id = plan
+            .append_operator_as_node(pattern_1.into(), vec![source_id])
+            .unwrap();
+
+        let join_id = plan
+            .append_operator_as_node(join.into(), vec![pattern_0_id, pattern_1_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        build_job_request(plan)
+    }
+
+    // g.V().in().match(
+    //    __.as('a').out("created").as('c'),
+    //    __.as('b').out('created').as('c'),
+    // ).join(
+    //    match(__.as('a').out("knows").as('b'))
+    // )
+    fn init_match_case4_request() -> JobRequest {
+        let source = pb::Scan::default();
+
+        let in_edge = get_in_edge();
+
+        let pattern_0 = get_pattern_case3();
+
+        let pattern_1 = get_pattern_case4();
+
+        let join = get_inner_join(vec![TAG_A, TAG_B]);
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C]);
+
+        let mut plan = LogicalPlan::default();
+        let source_id = plan
+            .append_operator_as_node(source.into(), vec![0])
+            .unwrap();
+
+        let in_edge_id = plan
+            .append_operator_as_node(in_edge.into(), vec![source_id])
+            .unwrap();
+
+        let pattern_0_id = plan
+            .append_operator_as_node(pattern_0.into(), vec![in_edge_id])
+            .unwrap();
+
+        let pattern_1_id = plan
+            .append_operator_as_node(pattern_1.into(), vec![in_edge_id])
+            .unwrap();
+
+        let join_id = plan
+            .append_operator_as_node(join.into(), vec![pattern_0_id, pattern_1_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        build_job_request(plan)
+    }
+
+    // g.V().in().match(
+    //    __.as('a').out("created").as('c'),
+    //    __.as('b').out('created').as('c'),
+    // ).join(
+    //    match(__.as('a').out("knows").as('b'))
+    // )
+    // first match follows the "in()" step
+    // second match follows the "V()"(source operator) step
+    fn init_match_case5_request() -> JobRequest {
+        let source = pb::Scan::default();
+
+        let in_edge = get_in_edge();
+
+        let pattern_0 = get_pattern_case3();
+
+        let pattern_1 = get_pattern_case4();
+
+        let join = get_inner_join(vec![TAG_A, TAG_B]);
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C]);
+
+        let mut plan = LogicalPlan::default();
+        let source_id = plan
+            .append_operator_as_node(source.into(), vec![0])
+            .unwrap();
+
+        let in_edge_id = plan
+            .append_operator_as_node(in_edge.into(), vec![source_id])
+            .unwrap();
+
+        let pattern_0_id = plan
+            .append_operator_as_node(pattern_0.into(), vec![in_edge_id])
+            .unwrap();
+
+        let pattern_1_id = plan
+            .append_operator_as_node(pattern_1.into(), vec![source_id])
+            .unwrap();
+
+        let join_id = plan
+            .append_operator_as_node(join.into(), vec![pattern_0_id, pattern_1_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        build_job_request(plan)
+    }
+
+    // g.V().match(
+    //    __.as('a').out().as('c'),
+    //    __.as('b').out().as('c'),
+    // ).join(
+    //    match(__.as('a').out().as('b'))
+    // )
+    fn init_match_case6_request() -> JobRequest {
+        let source = pb::Scan::default();
+
+        let pattern_0 = get_pattern_case5();
+
+        let pattern_1 = get_pattern_case6();
+
+        let join = get_inner_join(vec![TAG_A, TAG_B]);
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C]);
+
+        let mut plan = LogicalPlan::default();
+        let source_id = plan
+            .append_operator_as_node(source.into(), vec![0])
+            .unwrap();
+
+        let pattern_0_id = plan
+            .append_operator_as_node(pattern_0.into(), vec![source_id])
+            .unwrap();
+        let pattern_1_id = plan
+            .append_operator_as_node(pattern_1.into(), vec![source_id])
+            .unwrap();
+
+        let join_id = plan
+            .append_operator_as_node(join.into(), vec![pattern_0_id, pattern_1_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        build_job_request(plan)
+    }
+
+    // g.V().hasLabel('person').match(
+    //    __.as('a').out().as('c'),
+    //    __.as('b').out().as('c'),
+    // ).join(
+    //    match(__.as('a').out().as('b'))
+    // )
+    fn init_match_case7_request() -> JobRequest {
+        let source = get_person_scan();
+
+        let pattern_0 = get_pattern_case5();
+
+        let pattern_1 = get_pattern_case6();
+
+        let join = get_inner_join(vec![TAG_A, TAG_B]);
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C]);
+
+        let mut plan = LogicalPlan::default();
+        let source_id = plan
+            .append_operator_as_node(source.into(), vec![0])
+            .unwrap();
+
+        let pattern_0_id = plan
+            .append_operator_as_node(pattern_0.into(), vec![source_id])
+            .unwrap();
+        let pattern_1_id = plan
+            .append_operator_as_node(pattern_1.into(), vec![source_id])
+            .unwrap();
+
+        let join_id = plan
+            .append_operator_as_node(join.into(), vec![pattern_0_id, pattern_1_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        build_job_request(plan)
+    }
+
+    // g.V().hasLabel('person').match(
+    //    __.as('a').out("created").as('c'),
+    //    __.as('b').out("created").as('c'),
+    // ).join(
+    //    match(__.as('a').out().as('b'))
+    // )
+    fn init_match_case8_request() -> JobRequest {
+        let source = get_person_scan();
+
+        let pattern_0 = get_pattern_case3();
+
+        let pattern_1 = get_pattern_case6();
+
+        let join = get_inner_join(vec![TAG_A, TAG_B]);
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C]);
+
+        let mut plan = LogicalPlan::default();
+        let source_id = plan
+            .append_operator_as_node(source.into(), vec![0])
+            .unwrap();
+
+        let pattern_0_id = plan
+            .append_operator_as_node(pattern_0.into(), vec![source_id])
+            .unwrap();
+        let pattern_1_id = plan
+            .append_operator_as_node(pattern_1.into(), vec![source_id])
+            .unwrap();
+
+        let join_id = plan
+            .append_operator_as_node(join.into(), vec![pattern_0_id, pattern_1_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        build_job_request(plan)
+    }
+
+    // g.V().hasLabel('person').match(
+    //    __.as('a').out().as('c'),
+    //    __.as('b').out().as('c'),
+    // ).join(
+    //    match(__.as('a').out("knows").as('b'))
+    // )
+    fn init_match_case9_request() -> JobRequest {
+        let source = get_person_scan();
+
+        let pattern_0 = get_pattern_case5();
+
+        let pattern_1 = get_pattern_case4();
+
+        let join = get_inner_join(vec![TAG_A, TAG_B]);
+
+        let sink = get_sink(vec![TAG_A, TAG_B, TAG_C]);
+
+        let mut plan = LogicalPlan::default();
+        let source_id = plan
+            .append_operator_as_node(source.into(), vec![0])
+            .unwrap();
+
+        let pattern_0_id = plan
+            .append_operator_as_node(pattern_0.into(), vec![source_id])
+            .unwrap();
+        let pattern_1_id = plan
+            .append_operator_as_node(pattern_1.into(), vec![source_id])
+            .unwrap();
+
+        let join_id = plan
+            .append_operator_as_node(join.into(), vec![pattern_0_id, pattern_1_id])
+            .unwrap();
+        plan.append_operator_as_node(sink.into(), vec![join_id])
+            .unwrap();
+
+        build_job_request(plan)
     }
 
-    fn match_case1(with_schema: bool) {
+    fn match_case1(define_source: bool) {
         initialize();
-        let request = init_match_case1_request(with_schema);
+        let request = init_match_case1_request(define_source);
         let mut results = submit_query(request, 2);
         let mut result_collection = vec![];
         let expected_result_ids = vec![(1, 4, 1 << 56 | 3)];
@@ -255,17 +657,17 @@ mod test {
 
     #[test]
     fn naive_match_case1() {
-        match_case1(false)
+        match_case1(true)
     }
 
     #[test]
     fn extend_match_case1() {
-        match_case1(true)
+        match_case1(false)
     }
 
-    fn match_case2(with_schema: bool) {
+    fn match_case2(define_source: bool) {
         initialize();
-        let request = init_match_case2_request(with_schema);
+        let request = init_match_case2_request(define_source);
         let mut results = submit_query(request, 2);
         let mut result_collection = vec![];
         let expected_result_ids = vec![
@@ -301,11 +703,186 @@ mod test {
 
     #[test]
     fn naive_match_case2() {
-        match_case2(false)
+        match_case2(true)
     }
 
     #[test]
     fn extend_match_case2() {
-        match_case2(true)
+        match_case2(false)
+    }
+
+    #[test]
+    fn match_case3() {
+        initialize();
+        let request = init_match_case3_request();
+        let mut results = submit_query(request, 2);
+        let mut result_collection = vec![];
+        let expected_result_ids = vec![(1, 4, 1 << 56 | 3)];
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    let a = entry.get(Some(TAG_A)).unwrap().as_vertex();
+                    let b = entry.get(Some(TAG_B)).unwrap().as_vertex();
+                    let c = entry.get(Some(TAG_C)).unwrap().as_vertex();
+                    result_collection.push((a.unwrap().id(), b.unwrap().id(), c.unwrap().id()));
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_ids);
+    }
+
+    #[test]
+    fn match_case4() {
+        initialize();
+        let request = init_match_case4_request();
+        let mut results = submit_query(request, 2);
+        let mut result_collection = vec![];
+        let expected_result_ids = vec![(1, 4, 1 << 56 | 3); 9];
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    let a = entry.get(Some(TAG_A)).unwrap().as_vertex();
+                    let b = entry.get(Some(TAG_B)).unwrap().as_vertex();
+                    let c = entry.get(Some(TAG_C)).unwrap().as_vertex();
+                    result_collection.push((a.unwrap().id(), b.unwrap().id(), c.unwrap().id()));
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        result_collection.sort();
+        assert_eq!(expected_result_ids, result_collection);
+    }
+
+    #[test]
+    fn match_case5() {
+        initialize();
+        let request = init_match_case5_request();
+        let mut results = submit_query(request, 2);
+        let mut result_collection = vec![];
+        let expected_result_ids = vec![(1, 4, 1 << 56 | 3); 3];
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    let a = entry.get(Some(TAG_A)).unwrap().as_vertex();
+                    let b = entry.get(Some(TAG_B)).unwrap().as_vertex();
+                    let c = entry.get(Some(TAG_C)).unwrap().as_vertex();
+                    result_collection.push((a.unwrap().id(), b.unwrap().id(), c.unwrap().id()));
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        result_collection.sort();
+        assert_eq!(expected_result_ids, result_collection);
+    }
+
+    #[test]
+    fn match_case6() {
+        initialize();
+        let request = init_match_case6_request();
+        let mut results = submit_query(request, 2);
+        let mut result_collection = vec![];
+        let expected_result_ids = vec![(1, 4, 1 << 56 | 3)];
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    let a = entry.get(Some(TAG_A)).unwrap().as_vertex();
+                    let b = entry.get(Some(TAG_B)).unwrap().as_vertex();
+                    let c = entry.get(Some(TAG_C)).unwrap().as_vertex();
+                    result_collection.push((a.unwrap().id(), b.unwrap().id(), c.unwrap().id()));
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_ids);
+    }
+
+    #[test]
+    fn match_case7() {
+        initialize();
+        let request = init_match_case7_request();
+        let mut results = submit_query(request, 2);
+        let mut result_collection = vec![];
+        let expected_result_ids = vec![(1, 4, 1 << 56 | 3)];
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    let a = entry.get(Some(TAG_A)).unwrap().as_vertex();
+                    let b = entry.get(Some(TAG_B)).unwrap().as_vertex();
+                    let c = entry.get(Some(TAG_C)).unwrap().as_vertex();
+                    result_collection.push((a.unwrap().id(), b.unwrap().id(), c.unwrap().id()));
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_ids);
+    }
+
+    #[test]
+    fn match_case8() {
+        initialize();
+        let request = init_match_case8_request();
+        let mut results = submit_query(request, 2);
+        let mut result_collection = vec![];
+        let expected_result_ids = vec![(1, 4, 1 << 56 | 3)];
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    let a = entry.get(Some(TAG_A)).unwrap().as_vertex();
+                    let b = entry.get(Some(TAG_B)).unwrap().as_vertex();
+                    let c = entry.get(Some(TAG_C)).unwrap().as_vertex();
+                    result_collection.push((a.unwrap().id(), b.unwrap().id(), c.unwrap().id()));
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_ids);
+    }
+
+    #[test]
+    fn match_case9() {
+        initialize();
+        let request = init_match_case9_request();
+        let mut results = submit_query(request, 2);
+        let mut result_collection = vec![];
+        let expected_result_ids = vec![(1, 4, 1 << 56 | 3)];
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    let a = entry.get(Some(TAG_A)).unwrap().as_vertex();
+                    let b = entry.get(Some(TAG_B)).unwrap().as_vertex();
+                    let c = entry.get(Some(TAG_C)).unwrap().as_vertex();
+                    result_collection.push((a.unwrap().id(), b.unwrap().id(), c.unwrap().id()));
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_ids);
     }
 }