Repository: alibaba/GraphScope (language: python)

Issue:
refactor(interactive): Migrating Physical Layer into the new Calcite-based Compilation Stack.
**Is your feature request related to a problem? Please describe.**

The GIE IR layer is re-factored based on Calcite. However, the physical layer is still in the pre-IR route. Migrate it into the new Compilation Stack.

To migrate the physical layer:

1. The following operators need to be converted to Physical Plan (in proto)
Graph Operators:

- [X] Scan
- [X] IndexScan
- [X] EdgeExpand
- [X] GetV
- [X] PathExpand
- [X] ExpandDegree
- [X] PhysicalEdgeExpand
- [X] PhysicalGetV
- [x] CommonTableScan

Relational Operators:
- [X] Filter
- [X] Project 
- [X] Aggregate
- [X] Dedup 
- [X] Order
- [X] Join
- [x] Intersect (MultiJoin)
- [x] Union
- [ ] Apply

Others:
- [ ] Sink
   - [X] SinkDefault
   - [ ] SinkVineyard

2. The following physical rules should be implemented 

- [X] ExpandGetVFusion
  - [X] BasicExpandGetVFusion
  - [X] PathbaseExpandGetVFusion
- [x] Repartition
- [x] PropertyFetch
  - [x] PreCache (for edge properties)
  - [x] LateProject (for vertex properties)



refactor(interactive): Migrating Physical Layer into the new Calcite-based Compilation Stack.
**Is your feature request related to a problem? Please describe.**

The GIE IR layer is re-factored based on Calcite. However, the physical layer is still in the pre-IR route. Migrate it into the new Compilation Stack.

To migrate the physical layer:

1. The following operators need to be converted to Physical Plan (in proto)
Graph Operators:

- [X] Scan
- [X] IndexScan
- [X] EdgeExpand
- [X] GetV
- [X] PathExpand
- [X] ExpandDegree
- [X] PhysicalEdgeExpand
- [X] PhysicalGetV
- [x] CommonTableScan

Relational Operators:
- [X] Filter
- [X] Project 
- [X] Aggregate
- [X] Dedup 
- [X] Order
- [X] Join
- [x] Intersect (MultiJoin)
- [x] Union
- [ ] Apply

Others:
- [ ] Sink
   - [X] SinkDefault
   - [ ] SinkVineyard

2. The following physical rules should be implemented 

- [X] ExpandGetVFusion
  - [X] BasicExpandGetVFusion
  - [X] PathbaseExpandGetVFusion
- [x] Repartition
- [x] PropertyFetch
  - [x] PreCache (for edge properties)
  - [x] LateProject (for vertex properties)





Discussion / hints:

Fix patch:
diff --git a/charts/gie-standalone/templates/frontend/statefulset.yaml b/charts/gie-standalone/templates/frontend/statefulset.yaml
index 889172c334ac..5f3c0ce11c33 100644
--- a/charts/gie-standalone/templates/frontend/statefulset.yaml
+++ b/charts/gie-standalone/templates/frontend/statefulset.yaml
@@ -126,7 +126,9 @@ spec:
             - name: FRONTEND_QUERY_PER_SECOND_LIMIT
               value: {{ .Values.frontendQueryPerSecondLimit | quote }}
             - name: GREMLIN_SCRIPT_LANGUAGE_NAME
-              value: {{ .Values.gremlinScriptLanguageName }}
+              value: {{ .Values.gremlinScriptLanguageName | quote }}
+            - name: PHYSICAL_OPT_CONFIG
+              value: {{ .Values.physicalOptConfig}}
           ports:
             - name: gremlin
               containerPort: {{ .Values.frontend.service.gremlinPort }}
diff --git a/interactive_engine/compiler/Makefile b/interactive_engine/compiler/Makefile
index 04f0970c2390..071ae0ff7de5 100644
--- a/interactive_engine/compiler/Makefile
+++ b/interactive_engine/compiler/Makefile
@@ -57,6 +57,8 @@ run:
 	  -Dgraph.schema=${graph.schema} \
 	  -Dgraph.store=${graph.store} \
 	  -Dpegasus.hosts=${pegasus.hosts} \
+	  -Dgremlin.script.language.name=${gremlin.script.language.name} \
+	  -Dphysical.opt.config=${physical.opt.config} \
 	  com.alibaba.graphscope.GraphServer ${config.path}
 
 # make physical_plan config.path='<path to the config file>'
diff --git a/interactive_engine/compiler/conf/ir.compiler.properties b/interactive_engine/compiler/conf/ir.compiler.properties
index 851f6c0eaf17..902fb27bb4b1 100644
--- a/interactive_engine/compiler/conf/ir.compiler.properties
+++ b/interactive_engine/compiler/conf/ir.compiler.properties
@@ -52,3 +52,6 @@ calcite.default.charset: UTF-8
 
 # antlr_gremlin_traversal(default) or antlr_gremlin_calcite
 # gremlin.script.language.name: antlr_gremlin_traversal
+
+# the output plan format, can be ffi(default) or proto
+# physical.opt.config: ffi
\ No newline at end of file
diff --git a/interactive_engine/compiler/ir_experimental_ci.sh b/interactive_engine/compiler/ir_experimental_ci.sh
index 7b8a8938636e..5602c2ab6694 100755
--- a/interactive_engine/compiler/ir_experimental_ci.sh
+++ b/interactive_engine/compiler/ir_experimental_ci.sh
@@ -17,8 +17,7 @@ fi
 
 # restart compiler service
 ps -ef | grep "com.alibaba.graphscope.GraphServer" | grep -v grep | awk '{print $2}' | xargs kill -9 || true
-export gremlin.script.language.name=antlr_gremlin_calcite
-cd ${base_dir} && make run &
+cd ${base_dir} && make run gremlin.script.language.name=antlr_gremlin_calcite &
 sleep 5s
 # run gremlin standard tests to test calcite-based IR layer
 cd ${base_dir} && make gremlin_calcite_test
@@ -29,6 +28,19 @@ if [ $exit_code -ne 0 ]; then
     exit 1
 fi
 
+# restart compiler service
+ps -ef | grep "com.alibaba.graphscope.GraphServer" | grep -v grep | awk '{print $2}' | xargs kill -9 || true
+cd ${base_dir} && make run gremlin.script.language.name=antlr_gremlin_calcite physical.opt.config=proto &
+sleep 5s
+# run gremlin standard tests to test calcite-based IR layer
+cd ${base_dir} && make gremlin_calcite_test
+exit_code=$?
+# report test result
+if [ $exit_code -ne 0 ]; then
+    echo "ir\(calcite-based\) gremlin integration with proto physical test on experimental store fail"
+    exit 1
+fi
+
 # clean service
 ps -ef | grep "com.alibaba.graphscope.GraphServer" | grep -v grep | awk '{print $2}' | xargs kill -9 || true
 ps -ef | grep "start_rpc_server" | grep -v grep | awk '{print $2}' | xargs kill -9
diff --git a/interactive_engine/compiler/set_properties.sh b/interactive_engine/compiler/set_properties.sh
index d3c0bf416e03..7ccecb159d6c 100755
--- a/interactive_engine/compiler/set_properties.sh
+++ b/interactive_engine/compiler/set_properties.sh
@@ -34,6 +34,8 @@ frontend_query_per_second_limit="frontend.query.per.second.limit: $FRONTEND_QUER
 
 gremlin_script_language_name="gremlin.script.language.name: $GREMLIN_SCRIPT_LANGUAGE_NAME"
 
+physical_opt_config="physical.opt.config: $PHYSICAL_OPT_CONFIG"
+
 count=1;
 while (($count<$SERVERSSIZE))
 do
@@ -45,6 +47,6 @@ done
 
 graph_schema="graph.schema: $GRAPH_SCHEMA"
 
-properties="$worker_num\n$timeout\n$batch_size\n$output_capacity\n$hosts\n$server_num\n$graph_schema\n$gremlin_server_port\n$cypher_server_port\n$frontend_query_per_second_limit\n$gremlin_script_language_name"
+properties="$worker_num\n$timeout\n$batch_size\n$output_capacity\n$hosts\n$server_num\n$graph_schema\n$gremlin_server_port\n$cypher_server_port\n$frontend_query_per_second_limit\n$gremlin_script_language_name\n$physical_opt_config"
 
 echo -e $properties > ./conf/ir.compiler.properties
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/config/FrontendConfig.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/config/FrontendConfig.java
index b5fea9963013..a07ad7a517de 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/config/FrontendConfig.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/config/FrontendConfig.java
@@ -54,4 +54,7 @@ public class FrontendConfig {
 
     public static final Config<String> GREMLIN_SCRIPT_LANGUAGE_NAME =
             Config.stringConfig("gremlin.script.language.name", "antlr_gremlin_traversal");
+
+    public static final Config<String> PHYSICAL_OPT_CONFIG =
+            Config.stringConfig("physical.opt.config", "ffi");
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/DegreeFusionRule.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/DegreeFusionRule.java
index c40309c3b8fd..2db53f5f46f1 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/DegreeFusionRule.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/DegreeFusionRule.java
@@ -18,8 +18,8 @@
 
 import com.alibaba.graphscope.common.ir.rel.GraphLogicalAggregate;
 import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalExpand;
-import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalExpandDegree;
 import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalGetV;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphPhysicalExpand;
 import com.alibaba.graphscope.common.ir.rel.type.group.GraphAggCall;
 import com.alibaba.graphscope.common.ir.tools.GraphBuilder;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
@@ -48,11 +48,13 @@ protected DegreeFusionRule(C config) {
     protected RelNode transform(
             GraphLogicalAggregate count, GraphLogicalExpand expand, GraphBuilder builder) {
         RelNode expandDegree =
-                GraphLogicalExpandDegree.create(
+                GraphPhysicalExpand.create(
                         (GraphOptCluster) expand.getCluster(),
                         ImmutableList.of(),
                         expand.getInput(0),
                         expand,
+                        null,
+                        GraphOpt.PhysicalExpandOpt.DEGREE,
                         null);
         builder.push(expandDegree);
         Preconditions.checkArgument(
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/ExpandGetVFusionRule.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/ExpandGetVFusionRule.java
new file mode 100644
index 000000000000..fa2d9c288651
--- /dev/null
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/ExpandGetVFusionRule.java
@@ -0,0 +1,274 @@
+/*
+ * 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.
+ */
+
+package com.alibaba.graphscope.common.ir.planner.rules;
+
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalExpand;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalGetV;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalPathExpand;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphPhysicalExpand;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphPhysicalGetV;
+import com.alibaba.graphscope.common.ir.tools.AliasInference;
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
+import com.google.common.collect.ImmutableList;
+
+import org.apache.calcite.plan.GraphOptCluster;
+import org.apache.calcite.plan.RelOptRuleCall;
+import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.rules.TransformationRule;
+import org.apache.calcite.tools.RelBuilderFactory;
+import org.apache.commons.lang3.ObjectUtils;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+// This rule try to fuse GraphLogicalExpand and GraphLogicalGetV if GraphLogicalExpand has no alias
+// (that it won't be visited individually later):
+// 1. if GraphLogicalGetV has no filters, then:
+//      GraphLogicalExpand + GraphLogicalGetV -> GraphPhysicalExpand(ExpandV)
+// 2. if GraphLogicalGetV has filters, then:
+//      GraphLogicalExpand + GraphLogicalGetV -> GraphPhysicalExpand(ExpandV)  +
+// GraphPhysicalGetV(VertexFilter)
+public abstract class ExpandGetVFusionRule<C extends RelRule.Config> extends RelRule<C>
+        implements TransformationRule {
+
+    protected ExpandGetVFusionRule(C config) {
+        super(config);
+    }
+
+    protected RelNode transform(GraphLogicalGetV getV, GraphLogicalExpand expand, RelNode input) {
+        if (expand.getOpt().equals(GraphOpt.Expand.OUT) && getV.getOpt().equals(GraphOpt.GetV.END)
+                || expand.getOpt().equals(GraphOpt.Expand.IN)
+                        && getV.getOpt().equals(GraphOpt.GetV.START)
+                || expand.getOpt().equals(GraphOpt.Expand.BOTH)
+                        && getV.getOpt().equals(GraphOpt.GetV.OTHER)) {
+            GraphPhysicalExpand physicalExpand =
+                    GraphPhysicalExpand.create(
+                            expand.getCluster(),
+                            expand.getHints(),
+                            input,
+                            expand,
+                            getV,
+                            GraphOpt.PhysicalExpandOpt.VERTEX,
+                            getV.getAliasName());
+            if (ObjectUtils.isEmpty(getV.getFilters())) {
+                return physicalExpand;
+            } else {
+                // If with filters, then create a GraphPhysicalGetV to do the filtering.
+                // We set alias of getV to null to avoid alias conflict (with expand's alias)
+                GraphPhysicalGetV physicalGetV =
+                        GraphPhysicalGetV.create(
+                                getV.getCluster(),
+                                getV.getHints(),
+                                physicalExpand,
+                                getV,
+                                null,
+                                GraphOpt.PhysicalGetVOpt.ITSELF);
+                return physicalGetV;
+            }
+        } else {
+            return getV;
+        }
+    }
+
+    // transform expande + getv to GraphPhysicalExpandGetV
+    public static class BasicExpandGetVFusionRule
+            extends ExpandGetVFusionRule<BasicExpandGetVFusionRule.Config> {
+        protected BasicExpandGetVFusionRule(Config config) {
+            super(config);
+        }
+
+        @Override
+        public void onMatch(RelOptRuleCall call) {
+            GraphLogicalGetV getV = call.rel(0);
+            GraphLogicalExpand expand = call.rel(1);
+            call.transformTo(transform(getV, expand, expand.getInput(0)));
+        }
+
+        public static class Config implements RelRule.Config {
+            public static BasicExpandGetVFusionRule.Config DEFAULT =
+                    new Config()
+                            .withOperandSupplier(
+                                    b0 ->
+                                            b0.operand(GraphLogicalGetV.class)
+                                                    .oneInput(
+                                                            b1 ->
+                                                                    b1.operand(
+                                                                                    GraphLogicalExpand
+                                                                                            .class)
+                                                                            .predicate(
+                                                                                    (GraphLogicalExpand
+                                                                                                    expand) -> {
+                                                                                        int alias =
+                                                                                                expand
+                                                                                                        .getAliasId();
+                                                                                        return alias
+                                                                                                == AliasInference
+                                                                                                        .DEFAULT_ID;
+                                                                                    })
+                                                                            .anyInputs()));
+
+            private RelRule.OperandTransform operandSupplier;
+            private @Nullable String description;
+            private RelBuilderFactory builderFactory;
+
+            @Override
+            public RelRule toRule() {
+                return new BasicExpandGetVFusionRule(this);
+            }
+
+            @Override
+            public Config withRelBuilderFactory(RelBuilderFactory relBuilderFactory) {
+                this.builderFactory = relBuilderFactory;
+                return this;
+            }
+
+            @Override
+            public Config withDescription(
+                    @org.checkerframework.checker.nullness.qual.Nullable String s) {
+                this.description = s;
+                return this;
+            }
+
+            @Override
+            public Config withOperandSupplier(OperandTransform operandTransform) {
+                this.operandSupplier = operandTransform;
+                return this;
+            }
+
+            @Override
+            public OperandTransform operandSupplier() {
+                return this.operandSupplier;
+            }
+
+            @Override
+            public @org.checkerframework.checker.nullness.qual.Nullable String description() {
+                return this.description;
+            }
+
+            @Override
+            public RelBuilderFactory relBuilderFactory() {
+                return this.builderFactory;
+            }
+        }
+    }
+
+    // transform expande + getv to GraphPhysicalExpandGetV in PathExpand
+    public static class PathBaseExpandGetVFusionRule
+            extends ExpandGetVFusionRule<PathBaseExpandGetVFusionRule.Config> {
+        protected PathBaseExpandGetVFusionRule(Config config) {
+            super(config);
+        }
+
+        @Override
+        public void onMatch(RelOptRuleCall call) {
+            GraphLogicalPathExpand pathExpand = call.rel(0);
+            GraphLogicalGetV getV = (GraphLogicalGetV) pathExpand.getGetV();
+            GraphLogicalExpand expand = (GraphLogicalExpand) pathExpand.getExpand();
+            RelNode fused = transform(getV, expand, null);
+            GraphLogicalPathExpand afterPathExpand =
+                    GraphLogicalPathExpand.create(
+                            (GraphOptCluster) pathExpand.getCluster(),
+                            ImmutableList.of(),
+                            pathExpand.getInput(0),
+                            fused,
+                            pathExpand.getOffset(),
+                            pathExpand.getFetch(),
+                            pathExpand.getResultOpt(),
+                            pathExpand.getPathOpt(),
+                            pathExpand.getAliasName(),
+                            pathExpand.getStartAlias());
+            call.transformTo(afterPathExpand);
+        }
+
+        public static class Config implements RelRule.Config {
+            public static PathBaseExpandGetVFusionRule.Config DEFAULT =
+                    new Config()
+                            .withOperandSupplier(
+                                    b0 ->
+                                            b0.operand(GraphLogicalPathExpand.class)
+                                                    .predicate(
+                                                            (GraphLogicalPathExpand pathExpand) -> {
+                                                                if (GraphOpt.PathExpandResult
+                                                                        .ALL_V_E
+                                                                        .equals(
+                                                                                pathExpand
+                                                                                        .getResultOpt())) {
+                                                                    return false;
+                                                                } else {
+                                                                    if (pathExpand.getExpand()
+                                                                            instanceof
+                                                                            GraphLogicalExpand) {
+                                                                        GraphLogicalExpand expand =
+                                                                                (GraphLogicalExpand)
+                                                                                        pathExpand
+                                                                                                .getExpand();
+                                                                        int alias =
+                                                                                expand.getAliasId();
+                                                                        return alias
+                                                                                == AliasInference
+                                                                                        .DEFAULT_ID;
+                                                                    } else {
+                                                                        return false;
+                                                                    }
+                                                                }
+                                                            })
+                                                    .anyInputs());
+
+            private RelRule.OperandTransform operandSupplier;
+            private @Nullable String description;
+            private RelBuilderFactory builderFactory;
+
+            @Override
+            public RelRule toRule() {
+                return new PathBaseExpandGetVFusionRule(this);
+            }
+
+            @Override
+            public Config withRelBuilderFactory(RelBuilderFactory relBuilderFactory) {
+                this.builderFactory = relBuilderFactory;
+                return this;
+            }
+
+            @Override
+            public Config withDescription(
+                    @org.checkerframework.checker.nullness.qual.Nullable String s) {
+                this.description = s;
+                return this;
+            }
+
+            @Override
+            public Config withOperandSupplier(OperandTransform operandTransform) {
+                this.operandSupplier = operandTransform;
+                return this;
+            }
+
+            @Override
+            public OperandTransform operandSupplier() {
+                return this.operandSupplier;
+            }
+
+            @Override
+            public @org.checkerframework.checker.nullness.qual.Nullable String description() {
+                return this.description;
+            }
+
+            @Override
+            public RelBuilderFactory relBuilderFactory() {
+                return this.builderFactory;
+            }
+        }
+    }
+}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalAggregate.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalAggregate.java
index 829bccf9fb39..c7474735c09b 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalAggregate.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalAggregate.java
@@ -134,8 +134,8 @@ public List<GraphAggCall> getAggCalls() {
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalDedupBy.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalDedupBy.java
index 81f478582656..53e16d2c2f1b 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalDedupBy.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalDedupBy.java
@@ -20,6 +20,7 @@
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelTraitSet;
 import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelShuttle;
 import org.apache.calcite.rel.RelWriter;
 import org.apache.calcite.rel.SingleRel;
 import org.apache.calcite.rex.RexNode;
@@ -60,4 +61,12 @@ public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
         return new GraphLogicalDedupBy(
                 this.getCluster(), traitSet, inputs.get(0), this.dedupByKeys);
     }
+
+    @Override
+    public RelNode accept(RelShuttle shuttle) {
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
+        }
+        return shuttle.visit(this);
+    }
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalProject.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalProject.java
index a12423855474..a114357028f1 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalProject.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalProject.java
@@ -84,8 +84,8 @@ public RelWriter explainTerms(RelWriter pw) {
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalSort.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalSort.java
index 837fb9ace294..5691925bc526 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalSort.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphLogicalSort.java
@@ -82,8 +82,8 @@ public Sort copy(
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelShuttle.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelShuttle.java
index 7b8203a26119..3c973bcf2ca5 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelShuttle.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelShuttle.java
@@ -32,7 +32,7 @@ public interface GraphRelShuttle {
 
     RelNode visit(GraphLogicalExpand expand);
 
-    RelNode visit(GraphLogicalExpandDegree expandCount);
+    RelNode visit(GraphPhysicalExpand physicalExpand);
 
     RelNode visit(GraphLogicalGetV getV);
 
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelShuttleWrapper.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelShuttleWrapper.java
index be11d98cc1d1..2039e272e029 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelShuttleWrapper.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelShuttleWrapper.java
@@ -75,8 +75,6 @@ public RelNode visit(RelNode relNode) {
             return relShuttle.visit((GraphLogicalMultiMatch) relNode);
         } else if (relNode instanceof GraphLogicalPathExpand) {
             return relShuttle.visit((GraphLogicalPathExpand) relNode);
-        } else if (relNode instanceof GraphLogicalExpandDegree) {
-            return relShuttle.visit((GraphLogicalExpandDegree) relNode);
         } else if (relNode instanceof GraphLogicalDedupBy) {
             return relShuttle.visit((GraphLogicalDedupBy) relNode);
         } else {
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelVisitor.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphShuttle.java
similarity index 67%
rename from interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelVisitor.java
rename to interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphShuttle.java
index bdb0a4354dca..bcec982547a4 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphRelVisitor.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/GraphShuttle.java
@@ -16,7 +16,12 @@
 
 package com.alibaba.graphscope.common.ir.rel;
 
-import com.alibaba.graphscope.common.ir.rel.graph.*;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalExpand;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalGetV;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalPathExpand;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalSource;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphPhysicalExpand;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphPhysicalGetV;
 import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalMultiMatch;
 import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalSingleMatch;
 
@@ -24,9 +29,9 @@
 import org.apache.calcite.rel.RelShuttleImpl;
 
 /**
- * the class is used to visit a graph relation tree
+ * this class provides new visit method for graph rel node
  */
-public abstract class GraphRelVisitor extends RelShuttleImpl {
+public abstract class GraphShuttle extends RelShuttleImpl {
     public RelNode visit(GraphLogicalSource source) {
         return source;
     }
@@ -35,10 +40,6 @@ public RelNode visit(GraphLogicalExpand expand) {
         return visitChildren(expand);
     }
 
-    public RelNode visit(GraphLogicalExpandDegree degree) {
-        return visitChildren(degree);
-    }
-
     public RelNode visit(GraphLogicalGetV getV) {
         return visitChildren(getV);
     }
@@ -55,15 +56,27 @@ public RelNode visit(GraphLogicalMultiMatch match) {
         return match;
     }
 
-    public RelNode visit(GraphLogicalAggregate aggregate) {
-        return visitChildren(aggregate);
-    }
-
     public RelNode visit(GraphLogicalProject project) {
         return visitChildren(project);
     }
 
+    public RelNode visit(GraphLogicalAggregate aggregate) {
+        return visitChildren(aggregate);
+    }
+
     public RelNode visit(GraphLogicalSort sort) {
         return visitChildren(sort);
     }
+
+    public RelNode visit(GraphPhysicalExpand physicalExpand) {
+        return visitChildren(physicalExpand);
+    }
+
+    public RelNode visit(GraphPhysicalGetV physicalGetV) {
+        return visitChildren(physicalGetV);
+    }
+
+    public RelNode visit(GraphLogicalDedupBy dedupBy) {
+        return visitChildren(dedupBy);
+    }
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/PushFilterVisitor.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/PushFilterVisitor.java
index f79238237b52..932647e2098d 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/PushFilterVisitor.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/PushFilterVisitor.java
@@ -37,7 +37,7 @@
 /**
  * visit a graph relation tree and push filter to the corresponding source
  */
-public class PushFilterVisitor extends GraphRelVisitor {
+public class PushFilterVisitor extends GraphShuttle {
     private final GraphBuilder builder;
     private final RexNode condition;
     private final List<Integer> distinctAliasIds;
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalExpand.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalExpand.java
index d2c12d851dce..bf24da6bb6c9 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalExpand.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalExpand.java
@@ -16,7 +16,7 @@
 
 package com.alibaba.graphscope.common.ir.rel.graph;
 
-import com.alibaba.graphscope.common.ir.rel.GraphRelVisitor;
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
 import com.alibaba.graphscope.common.ir.rel.type.AliasNameWithId;
 import com.alibaba.graphscope.common.ir.rel.type.TableConfig;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
@@ -86,8 +86,8 @@ public GraphLogicalExpand copy(RelTraitSet traitSet, List<RelNode> inputs) {
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalExpandDegree.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalExpandDegree.java
deleted file mode 100644
index 58141cb5eb75..000000000000
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalExpandDegree.java
+++ /dev/null
@@ -1,112 +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.ir.rel.graph;
-
-import com.alibaba.graphscope.common.ir.rel.GraphRelVisitor;
-import com.alibaba.graphscope.common.ir.tools.AliasInference;
-import com.google.common.collect.ImmutableList;
-
-import org.apache.calcite.plan.GraphOptCluster;
-import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelShuttle;
-import org.apache.calcite.rel.RelWriter;
-import org.apache.calcite.rel.SingleRel;
-import org.apache.calcite.rel.hint.RelHint;
-import org.apache.calcite.rel.type.*;
-import org.apache.calcite.sql.type.SqlTypeName;
-import org.apache.commons.lang3.ObjectUtils;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-import java.util.List;
-import java.util.Objects;
-
-// fuse expand + count after applying 'DegreeFusionRule'
-public class GraphLogicalExpandDegree extends SingleRel {
-    private final GraphLogicalExpand fusedExpand;
-    private final String aliasName;
-    private final int aliasId;
-
-    protected GraphLogicalExpandDegree(
-            GraphOptCluster cluster,
-            List<RelHint> hints,
-            RelNode input,
-            GraphLogicalExpand fusedExpand,
-            @Nullable String aliasName) {
-        super(cluster, RelTraitSet.createEmpty(), input);
-        this.fusedExpand = Objects.requireNonNull(fusedExpand);
-        this.aliasName =
-                AliasInference.inferDefault(
-                        aliasName, AliasInference.getUniqueAliasList(input, true));
-        this.aliasId = cluster.getIdGenerator().generate(this.aliasName);
-    }
-
-    public static GraphLogicalExpandDegree create(
-            GraphOptCluster cluster,
-            List<RelHint> hints,
-            RelNode input,
-            GraphLogicalExpand innerExpand,
-            @Nullable String alias) {
-        return new GraphLogicalExpandDegree(cluster, hints, input, innerExpand, alias);
-    }
-
-    @Override
-    public RelDataType deriveRowType() {
-        RelDataTypeFactory typeFactory = getCluster().getTypeFactory();
-        RelDataTypeField field =
-                new RelDataTypeFieldImpl(
-                        this.aliasName,
-                        this.aliasId,
-                        typeFactory.createSqlType(SqlTypeName.BIGINT));
-        return new RelRecordType(ImmutableList.of(field));
-    }
-
-    public GraphLogicalExpand getFusedExpand() {
-        return fusedExpand;
-    }
-
-    public String getAliasName() {
-        return aliasName;
-    }
-
-    public int getAliasId() {
-        return aliasId;
-    }
-
-    @Override
-    public RelWriter explainTerms(RelWriter pw) {
-        return pw.itemIf("input", input, !Objects.isNull(input))
-                .item("tableConfig", fusedExpand.tableConfig)
-                .item("alias", AliasInference.SIMPLE_NAME(getAliasName()))
-                .itemIf(
-                        "fusedProject",
-                        fusedExpand.project,
-                        !ObjectUtils.isEmpty(fusedExpand.project))
-                .itemIf(
-                        "fusedFilter",
-                        fusedExpand.filters,
-                        !ObjectUtils.isEmpty(fusedExpand.filters));
-    }
-
-    @Override
-    public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
-        }
-        return shuttle.visit(this);
-    }
-}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalGetV.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalGetV.java
index cfc94f954940..4001bbbf21f9 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalGetV.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalGetV.java
@@ -16,7 +16,7 @@
 
 package com.alibaba.graphscope.common.ir.rel.graph;
 
-import com.alibaba.graphscope.common.ir.rel.GraphRelVisitor;
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
 import com.alibaba.graphscope.common.ir.rel.type.AliasNameWithId;
 import com.alibaba.graphscope.common.ir.rel.type.TableConfig;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
@@ -86,8 +86,8 @@ public GraphLogicalGetV copy(RelTraitSet traitSet, List<RelNode> inputs) {
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java
index 3b382e2da503..9a3a26d58657 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java
@@ -16,11 +16,12 @@
 
 package com.alibaba.graphscope.common.ir.rel.graph;
 
-import com.alibaba.graphscope.common.ir.rel.GraphRelVisitor;
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
 import com.alibaba.graphscope.common.ir.rel.type.AliasNameWithId;
 import com.alibaba.graphscope.common.ir.tools.AliasInference;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
 import com.alibaba.graphscope.common.ir.type.GraphPathType;
+import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 
 import org.apache.calcite.plan.GraphOptCluster;
@@ -45,6 +46,9 @@ public class GraphLogicalPathExpand extends SingleRel {
     private final RelNode expand;
     private final RelNode getV;
 
+    // fuse expand with getV to a single operator
+    private final RelNode fused;
+
     private final @Nullable RexNode offset;
     private final @Nullable RexNode fetch;
 
@@ -72,6 +76,33 @@ protected GraphLogicalPathExpand(
         super(cluster, RelTraitSet.createEmpty(), input);
         this.expand = Objects.requireNonNull(expand);
         this.getV = Objects.requireNonNull(getV);
+        this.fused = null;
+        this.offset = offset;
+        this.fetch = fetch;
+        this.resultOpt = resultOpt;
+        this.pathOpt = pathOpt;
+        this.aliasName =
+                AliasInference.inferDefault(
+                        aliasName, AliasInference.getUniqueAliasList(input, true));
+        this.aliasId = cluster.getIdGenerator().generate(this.aliasName);
+        this.startAlias = Objects.requireNonNull(startAlias);
+    }
+
+    protected GraphLogicalPathExpand(
+            GraphOptCluster cluster,
+            @Nullable List<RelHint> hints,
+            RelNode input,
+            RelNode fused,
+            @Nullable RexNode offset,
+            @Nullable RexNode fetch,
+            GraphOpt.PathExpandResult resultOpt,
+            GraphOpt.PathExpandPath pathOpt,
+            @Nullable String aliasName,
+            AliasNameWithId startAlias) {
+        super(cluster, RelTraitSet.createEmpty(), input);
+        this.expand = null;
+        this.getV = null;
+        this.fused = Objects.requireNonNull(fused);
         this.offset = offset;
         this.fetch = fetch;
         this.resultOpt = resultOpt;
@@ -109,11 +140,39 @@ public static GraphLogicalPathExpand create(
                 startAlias);
     }
 
+    public static GraphLogicalPathExpand create(
+            GraphOptCluster cluster,
+            List<RelHint> hints,
+            RelNode input,
+            RelNode fused,
+            @Nullable RexNode offset,
+            @Nullable RexNode fetch,
+            GraphOpt.PathExpandResult resultOpt,
+            GraphOpt.PathExpandPath pathOpt,
+            String aliasName,
+            AliasNameWithId startAlias) {
+        Preconditions.checkArgument(
+                resultOpt != GraphOpt.PathExpandResult.ALL_V_E,
+                "can not fuse expand with getV if result opt is set to " + resultOpt.name());
+        return new GraphLogicalPathExpand(
+                cluster,
+                hints,
+                input,
+                fused,
+                offset,
+                fetch,
+                resultOpt,
+                pathOpt,
+                aliasName,
+                startAlias);
+    }
+
     @Override
     public RelWriter explainTerms(RelWriter pw) {
         return super.explainTerms(pw)
-                .item("expand", RelOptUtil.toString(expand))
-                .item("getV", RelOptUtil.toString(getV))
+                .itemIf("expand", RelOptUtil.toString(expand), expand != null)
+                .itemIf("getV", RelOptUtil.toString(getV), getV != null)
+                .itemIf("fused", RelOptUtil.toString(fused), fused != null)
                 .itemIf("offset", offset, offset != null)
                 .itemIf("fetch", fetch, fetch != null)
                 .item("path_opt", getPathOpt())
@@ -149,6 +208,10 @@ public RelNode getGetV() {
         return getV;
     }
 
+    public @Nullable RelNode getFused() {
+        return fused;
+    }
+
     public @Nullable RexNode getOffset() {
         return offset;
     }
@@ -163,35 +226,72 @@ public AliasNameWithId getStartAlias() {
 
     @Override
     protected RelDataType deriveRowType() {
-        ObjectUtils.requireNonEmpty(
-                this.expand.getRowType().getFieldList(),
-                "data type of expand operator should have at least one column field");
-        ObjectUtils.requireNonEmpty(
-                this.getV.getRowType().getFieldList(),
-                "data type of getV operator should have at least one column field");
         return new RelRecordType(
                 ImmutableList.of(
                         new RelDataTypeFieldImpl(
                                 getAliasName(),
                                 getAliasId(),
-                                new GraphPathType(
-                                        new GraphPathType.ElementType(
-                                                this.expand
-                                                        .getRowType()
-                                                        .getFieldList()
-                                                        .get(0)
-                                                        .getType(),
-                                                this.getV
-                                                        .getRowType()
-                                                        .getFieldList()
-                                                        .get(0)
-                                                        .getType())))));
+                                new GraphPathType(getElementType()))));
+    }
+
+    private GraphPathType.ElementType getElementType() {
+        switch (resultOpt) {
+            case ALL_V:
+            case END_V:
+                RelNode getV = this.fused != null ? this.fused : this.getV;
+                ObjectUtils.requireNonEmpty(
+                        getV.getRowType().getFieldList(),
+                        "data type of getV operator should have at least one column field");
+                return new GraphPathType.ElementType(
+                        getV.getRowType().getFieldList().get(0).getType());
+            case ALL_V_E:
+            default:
+                ObjectUtils.requireNonEmpty(
+                        this.expand.getRowType().getFieldList(),
+                        "data type of expand operator should have at least one column field");
+                ObjectUtils.requireNonEmpty(
+                        this.getV.getRowType().getFieldList(),
+                        "data type of getV operator should have at least one column field");
+                return new GraphPathType.ElementType(
+                        this.expand.getRowType().getFieldList().get(0).getType(),
+                        this.getV.getRowType().getFieldList().get(0).getType());
+        }
+    }
+
+    @Override
+    public GraphLogicalPathExpand copy(RelTraitSet traitSet, List<RelNode> inputs) {
+        if (this.fused != null) {
+            return new GraphLogicalPathExpand(
+                    (GraphOptCluster) getCluster(),
+                    ImmutableList.of(),
+                    inputs.get(0),
+                    this.fused,
+                    getOffset(),
+                    getFetch(),
+                    getResultOpt(),
+                    getPathOpt(),
+                    getAliasName(),
+                    getStartAlias());
+        } else {
+            return new GraphLogicalPathExpand(
+                    (GraphOptCluster) getCluster(),
+                    ImmutableList.of(),
+                    inputs.get(0),
+                    this.expand,
+                    this.getV,
+                    getOffset(),
+                    getFetch(),
+                    getResultOpt(),
+                    getPathOpt(),
+                    getAliasName(),
+                    getStartAlias());
+        }
     }
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalSource.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalSource.java
index f7c331c23e49..321b4109cc2f 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalSource.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalSource.java
@@ -16,7 +16,7 @@
 
 package com.alibaba.graphscope.common.ir.rel.graph;
 
-import com.alibaba.graphscope.common.ir.rel.GraphRelVisitor;
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
 import com.alibaba.graphscope.common.ir.rel.type.TableConfig;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
 
@@ -67,8 +67,8 @@ public RelWriter explainTerms(RelWriter pw) {
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphPhysicalExpand.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphPhysicalExpand.java
new file mode 100644
index 000000000000..478586969cd7
--- /dev/null
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphPhysicalExpand.java
@@ -0,0 +1,153 @@
+/*
+ * 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.ir.rel.graph;
+
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
+import com.alibaba.graphscope.common.ir.rel.type.AliasNameWithId;
+import com.alibaba.graphscope.common.ir.tools.AliasInference;
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
+import com.google.common.collect.ImmutableList;
+
+import org.apache.calcite.plan.GraphOptCluster;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelShuttle;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.SingleRel;
+import org.apache.calcite.rel.hint.RelHint;
+import org.apache.calcite.rel.type.*;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.commons.lang3.ObjectUtils;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.List;
+import java.util.Objects;
+
+public class GraphPhysicalExpand extends SingleRel {
+    private final GraphOpt.PhysicalExpandOpt physicalOpt;
+    private final GraphLogicalExpand fusedExpand;
+    private final GraphLogicalGetV fusedGetV;
+    private final String aliasName;
+    private final int aliasId;
+
+    protected GraphPhysicalExpand(
+            RelOptCluster cluster,
+            List<RelHint> hints,
+            RelNode input,
+            GraphLogicalExpand fusedExpand,
+            GraphLogicalGetV fusedGetV,
+            GraphOpt.PhysicalExpandOpt physicalOpt,
+            String aliasName) {
+        super(cluster, RelTraitSet.createEmpty(), input);
+        this.physicalOpt = physicalOpt;
+        this.fusedExpand = fusedExpand;
+        this.fusedGetV = fusedGetV;
+        this.aliasName =
+                AliasInference.inferDefault(
+                        aliasName, AliasInference.getUniqueAliasList(input, true));
+        this.aliasId = ((GraphOptCluster) cluster).getIdGenerator().generate(this.aliasName);
+    }
+
+    public static GraphPhysicalExpand create(
+            RelOptCluster cluster,
+            List<RelHint> hints,
+            RelNode input,
+            GraphLogicalExpand fusedExpand,
+            GraphLogicalGetV fusedGetV,
+            GraphOpt.PhysicalExpandOpt physicalOpt,
+            String alias) {
+        return new GraphPhysicalExpand(
+                cluster, hints, input, fusedExpand, fusedGetV, physicalOpt, alias);
+    }
+
+    public GraphOpt.PhysicalExpandOpt getPhysicalOpt() {
+        return this.physicalOpt;
+    }
+
+    public GraphLogicalExpand getFusedExpand() {
+        return fusedExpand;
+    }
+
+    public AliasNameWithId getStartAlias() {
+        return fusedExpand.getStartAlias();
+    }
+
+    public String getAliasName() {
+        return aliasName;
+    }
+
+    public int getAliasId() {
+        return aliasId;
+    }
+
+    public @Nullable ImmutableList<RexNode> getFilters() {
+        return fusedExpand.getFilters();
+    }
+
+    @Override
+    public RelDataType deriveRowType() {
+        switch (physicalOpt) {
+            case EDGE:
+                return fusedExpand.deriveRowType();
+            case DEGREE:
+                {
+                    RelDataTypeFactory typeFactory = getCluster().getTypeFactory();
+                    RelDataTypeField field =
+                            new RelDataTypeFieldImpl(
+                                    aliasName,
+                                    aliasId,
+                                    typeFactory.createSqlType(SqlTypeName.BIGINT));
+                    return new RelRecordType(ImmutableList.of(field));
+                }
+            case VERTEX:
+            default:
+                return fusedGetV.deriveRowType();
+        }
+    }
+
+    @Override
+    public List<RelNode> getInputs() {
+        return this.input == null ? ImmutableList.of() : ImmutableList.of(this.input);
+    }
+
+    @Override
+    public RelWriter explainTerms(RelWriter pw) {
+        return pw.itemIf("input", input, !Objects.isNull(input))
+                .item("tableConfig", fusedExpand.tableConfig)
+                .item("alias", AliasInference.SIMPLE_NAME(getAliasName()))
+                .itemIf(
+                        "startAlias",
+                        fusedExpand.getStartAlias(),
+                        fusedExpand.getStartAlias().getAliasName() != AliasInference.DEFAULT_NAME)
+                .itemIf(
+                        "fusedFilter",
+                        fusedExpand.getFilters(),
+                        !ObjectUtils.isEmpty(fusedExpand.getFilters()))
+                .item("opt", fusedExpand.getOpt())
+                .item("physicalOpt", getPhysicalOpt());
+    }
+
+    @Override
+    public RelNode accept(RelShuttle shuttle) {
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
+        }
+        return shuttle.visit(this);
+    }
+}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphPhysicalGetV.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphPhysicalGetV.java
new file mode 100644
index 000000000000..9d044ed7db77
--- /dev/null
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphPhysicalGetV.java
@@ -0,0 +1,141 @@
+/*
+ * 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.ir.rel.graph;
+
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
+import com.alibaba.graphscope.common.ir.rel.type.AliasNameWithId;
+import com.alibaba.graphscope.common.ir.tools.AliasInference;
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
+import com.google.common.collect.ImmutableList;
+
+import org.apache.calcite.plan.GraphOptCluster;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelShuttle;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.SingleRel;
+import org.apache.calcite.rel.hint.RelHint;
+import org.apache.commons.lang3.ObjectUtils;
+
+import java.util.List;
+import java.util.Objects;
+
+public class GraphPhysicalGetV extends SingleRel {
+    private final GraphOpt.PhysicalGetVOpt physicalOpt;
+    private final GraphLogicalGetV fusedGetV;
+
+    protected GraphPhysicalGetV(
+            RelOptCluster cluster,
+            List<RelHint> hints,
+            RelNode input,
+            GraphLogicalGetV fusedGetV,
+            GraphOpt.PhysicalGetVOpt physicalOpt) {
+        super(cluster, RelTraitSet.createEmpty(), input);
+        this.physicalOpt = physicalOpt;
+        this.fusedGetV = fusedGetV;
+    }
+
+    public static GraphPhysicalGetV create(
+            RelOptCluster cluster,
+            List<RelHint> hints,
+            RelNode input,
+            GraphLogicalGetV innerGetV,
+            String aliasName,
+            GraphOpt.PhysicalGetVOpt physicalOpt) {
+        // build a new getV if a new aliasName is given, to make sure the derived row type is
+        // correct (which is derived by getV)
+        GraphLogicalGetV newGetV;
+        if (innerGetV.getAliasName().equals(aliasName)) {
+            newGetV = innerGetV;
+        } else {
+            newGetV =
+                    GraphLogicalGetV.create(
+                            (GraphOptCluster) innerGetV.getCluster(),
+                            innerGetV.getHints(),
+                            input,
+                            innerGetV.getOpt(),
+                            innerGetV.getTableConfig(),
+                            aliasName,
+                            innerGetV.getStartAlias());
+            newGetV.setFilters(innerGetV.getFilters());
+        }
+        return new GraphPhysicalGetV(cluster, hints, input, newGetV, physicalOpt);
+    }
+
+    public GraphOpt.PhysicalGetVOpt getPhysicalOpt() {
+        return this.physicalOpt;
+    }
+
+    public AliasNameWithId getStartAlias() {
+        return fusedGetV.getStartAlias();
+    }
+
+    public String getAliasName() {
+        return fusedGetV.getAliasName();
+    }
+
+    public int getAliasId() {
+        return fusedGetV.getAliasId();
+    }
+
+    public GraphLogicalGetV getFusedGetV() {
+        return fusedGetV;
+    }
+
+    @Override
+    public List<RelNode> getInputs() {
+        return this.input == null ? ImmutableList.of() : ImmutableList.of(this.input);
+    }
+
+    @Override
+    public RelWriter explainTerms(RelWriter pw) {
+        return pw.itemIf("input", input, !Objects.isNull(input))
+                .item("tableConfig", fusedGetV.tableConfig)
+                .item("alias", AliasInference.SIMPLE_NAME(fusedGetV.getAliasName()))
+                .itemIf(
+                        "startAlias",
+                        fusedGetV.getStartAlias(),
+                        fusedGetV.getStartAlias().getAliasName() != AliasInference.DEFAULT_NAME)
+                .itemIf(
+                        "fusedFilter",
+                        fusedGetV.getFilters(),
+                        !ObjectUtils.isEmpty(fusedGetV.getFilters()))
+                .item("opt", fusedGetV.getOpt())
+                .item("physicalOpt", getPhysicalOpt());
+    }
+
+    @Override
+    public GraphPhysicalGetV copy(RelTraitSet traitSet, List<RelNode> inputs) {
+        GraphPhysicalGetV copy =
+                new GraphPhysicalGetV(
+                        getCluster(),
+                        fusedGetV.getHints(),
+                        inputs.get(0),
+                        fusedGetV,
+                        getPhysicalOpt());
+        return copy;
+    }
+
+    @Override
+    public RelNode accept(RelShuttle shuttle) {
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
+        }
+        return shuttle.visit(this);
+    }
+}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/match/GraphLogicalMultiMatch.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/match/GraphLogicalMultiMatch.java
index f02089cfc8d3..c09036103d15 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/match/GraphLogicalMultiMatch.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/match/GraphLogicalMultiMatch.java
@@ -16,7 +16,7 @@
 
 package com.alibaba.graphscope.common.ir.rel.graph.match;
 
-import com.alibaba.graphscope.common.ir.rel.GraphRelVisitor;
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Sets;
 
@@ -95,8 +95,8 @@ public RelDataType deriveRowType() {
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/match/GraphLogicalSingleMatch.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/match/GraphLogicalSingleMatch.java
index 7e8dc9181570..cb1983efddd7 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/match/GraphLogicalSingleMatch.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/match/GraphLogicalSingleMatch.java
@@ -16,7 +16,7 @@
 
 package com.alibaba.graphscope.common.ir.rel.graph.match;
 
-import com.alibaba.graphscope.common.ir.rel.GraphRelVisitor;
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
 import com.google.common.collect.Lists;
 
@@ -75,8 +75,8 @@ public RelDataType deriveRowType() {
 
     @Override
     public RelNode accept(RelShuttle shuttle) {
-        if (shuttle instanceof GraphRelVisitor) {
-            return ((GraphRelVisitor) shuttle).visit(this);
+        if (shuttle instanceof GraphShuttle) {
+            return ((GraphShuttle) shuttle).visit(this);
         }
         return shuttle.visit(this);
     }
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 2c875c275244..a42c294c7c70 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
@@ -93,7 +93,7 @@ public void appendNode(PhysicalNode<Pointer> node) {
             checkFfiResult(
                     LIB.appendScanOperator(ptrPlan, node.getNode(), oprIdx.getValue(), oprIdx));
         } else if (original instanceof GraphLogicalExpand
-                || original instanceof GraphLogicalExpandDegree) {
+                || original instanceof GraphPhysicalExpand) {
             checkFfiResult(
                     LIB.appendEdgexpdOperator(ptrPlan, node.getNode(), oprIdx.getValue(), oprIdx));
         } else if (original instanceof GraphLogicalGetV) {
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java
index d0b08d0677c9..762d733eff46 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java
@@ -123,26 +123,27 @@ public RelNode visit(GraphLogicalExpand expand) {
     }
 
     @Override
-    public RelNode visit(GraphLogicalExpandDegree expandCount) {
-        GraphLogicalExpand fusedExpand = expandCount.getFusedExpand();
-        Pointer ptrExpandCount =
+    public RelNode visit(GraphPhysicalExpand physicalExpand) {
+        GraphLogicalExpand fusedExpand = physicalExpand.getFusedExpand();
+        Pointer ptrPhysicalExpand =
                 LIB.initEdgexpdOperator(
-                        FfiExpandOpt.Degree, Utils.ffiDirection(fusedExpand.getOpt()));
-        checkFfiResult(LIB.setEdgexpdParams(ptrExpandCount, ffiQueryParams(fusedExpand)));
-        if (expandCount.getAliasId() != AliasInference.DEFAULT_ID) {
+                        Utils.ffiPhysicalExpandOpt(physicalExpand.getPhysicalOpt()),
+                        Utils.ffiDirection(fusedExpand.getOpt()));
+        checkFfiResult(LIB.setEdgexpdParams(ptrPhysicalExpand, ffiQueryParams(fusedExpand)));
+        if (physicalExpand.getAliasId() != AliasInference.DEFAULT_ID) {
             checkFfiResult(
                     LIB.setEdgexpdAlias(
-                            ptrExpandCount, ArgUtils.asAlias(expandCount.getAliasId())));
+                            ptrPhysicalExpand, ArgUtils.asAlias(physicalExpand.getAliasId())));
         }
         checkFfiResult(
                 LIB.setEdgexpdMeta(
-                        ptrExpandCount,
+                        ptrPhysicalExpand,
                         new FfiPbPointer.ByValue(
                                 com.alibaba.graphscope.common.ir.runtime.proto.Utils.protoRowType(
-                                                expandCount.getRowType(), isColumnId)
+                                                physicalExpand.getRowType(), isColumnId)
                                         .get(0)
                                         .toByteArray())));
-        return new PhysicalNode(expandCount, ptrExpandCount);
+        return new PhysicalNode(physicalExpand, ptrPhysicalExpand);
     }
 
     @Override
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/Utils.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/Utils.java
index 8696b754d5dd..fa7655eea0a7 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/Utils.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/Utils.java
@@ -51,6 +51,18 @@ public static final FfiDirection ffiDirection(GraphOpt.Expand opt) {
         }
     }
 
+    public static final FfiExpandOpt ffiPhysicalExpandOpt(GraphOpt.PhysicalExpandOpt opt) {
+        switch (opt) {
+            case DEGREE:
+                return FfiExpandOpt.Degree;
+            case EDGE:
+                return FfiExpandOpt.Edge;
+            case VERTEX:
+            default:
+                return FfiExpandOpt.Vertex;
+        }
+    }
+
     public static final FfiVOpt ffiVOpt(GraphOpt.GetV opt) {
         switch (opt) {
             case START:
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/GraphRelProtoPhysicalBuilder.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/GraphRelProtoPhysicalBuilder.java
new file mode 100644
index 000000000000..5fcb905b2195
--- /dev/null
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/GraphRelProtoPhysicalBuilder.java
@@ -0,0 +1,96 @@
+/*
+ * 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.
+ */
+
+package com.alibaba.graphscope.common.ir.runtime.proto;
+
+import com.alibaba.graphscope.common.config.Configs;
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
+import com.alibaba.graphscope.common.ir.runtime.PhysicalBuilder;
+import com.alibaba.graphscope.common.ir.runtime.PhysicalPlan;
+import com.alibaba.graphscope.common.ir.tools.LogicalPlan;
+import com.alibaba.graphscope.common.store.IrMeta;
+import com.alibaba.graphscope.gaia.proto.GraphAlgebra;
+import com.alibaba.graphscope.gaia.proto.GraphAlgebraPhysical;
+import com.google.protobuf.util.JsonFormat;
+
+import org.apache.calcite.rel.RelNode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Objects;
+
+/**
+ * build physical plan from logical plan of a regular query, the physical plan
+ * is a protobuf message {@code GraphAlgebraPhysical.PhysicalPlan}
+ */
+public class GraphRelProtoPhysicalBuilder extends PhysicalBuilder {
+    private static final Logger logger =
+            LoggerFactory.getLogger(GraphRelProtoPhysicalBuilder.class);
+    protected GraphShuttle relShuttle;
+    GraphAlgebraPhysical.PhysicalPlan.Builder physicalBuilder;
+
+    public GraphRelProtoPhysicalBuilder(
+            Configs graphConfig, IrMeta irMeta, LogicalPlan logicalPlan) {
+        super(logicalPlan);
+        this.physicalBuilder = GraphAlgebraPhysical.PhysicalPlan.newBuilder();
+        this.relShuttle =
+                new GraphRelToProtoConverter(
+                        irMeta.getSchema().isColumnId(), graphConfig, this.physicalBuilder);
+    }
+
+    @Override
+    public PhysicalPlan build() {
+        String plan = null;
+        try {
+            RelNode regularQuery = this.logicalPlan.getRegularQuery();
+            regularQuery.accept(this.relShuttle);
+            appendDefaultSink();
+            plan = getPlanAsJson(physicalBuilder.build());
+            int planId = Objects.hash(logicalPlan);
+            physicalBuilder.setPlanId(planId);
+            GraphAlgebraPhysical.PhysicalPlan physicalPlan = physicalBuilder.build();
+            byte[] bytes = physicalPlan.toByteArray();
+            return new PhysicalPlan(bytes, plan);
+        } catch (Exception e) {
+            logger.error("ir physical plan {}, error {}", plan, e);
+            throw new RuntimeException(e);
+        }
+    }
+
+    private void appendDefaultSink() {
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.Sink.Builder sinkBuilder = GraphAlgebraPhysical.Sink.newBuilder();
+        GraphAlgebra.Sink.SinkTarget.Builder sinkTargetBuilder =
+                GraphAlgebra.Sink.SinkTarget.newBuilder();
+        sinkTargetBuilder.setSinkDefault(GraphAlgebra.SinkDefault.newBuilder().build());
+        sinkBuilder.setSinkTarget(sinkTargetBuilder);
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setSink(sinkBuilder));
+        physicalBuilder.addPlan(oprBuilder);
+    }
+
+    private String getPlanAsJson(GraphAlgebraPhysical.PhysicalPlan physicalPlan) {
+        try {
+            return JsonFormat.printer().print(physicalPlan);
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    @Override
+    public void close() throws Exception {}
+}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/GraphRelToProtoConverter.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/GraphRelToProtoConverter.java
new file mode 100644
index 000000000000..d1eb95cbf205
--- /dev/null
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/GraphRelToProtoConverter.java
@@ -0,0 +1,686 @@
+/*
+ * 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.
+ */
+
+package com.alibaba.graphscope.common.ir.runtime.proto;
+
+import com.alibaba.graphscope.common.config.Configs;
+import com.alibaba.graphscope.common.ir.rel.GraphLogicalAggregate;
+import com.alibaba.graphscope.common.ir.rel.GraphLogicalDedupBy;
+import com.alibaba.graphscope.common.ir.rel.GraphLogicalProject;
+import com.alibaba.graphscope.common.ir.rel.GraphLogicalSort;
+import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
+import com.alibaba.graphscope.common.ir.rel.graph.*;
+import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalMultiMatch;
+import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalSingleMatch;
+import com.alibaba.graphscope.common.ir.rel.type.group.GraphAggCall;
+import com.alibaba.graphscope.common.ir.rel.type.group.GraphGroupKeys;
+import com.alibaba.graphscope.common.ir.rel.type.order.GraphFieldCollation;
+import com.alibaba.graphscope.common.ir.rex.RexGraphVariable;
+import com.alibaba.graphscope.common.ir.tools.AliasInference;
+import com.alibaba.graphscope.common.ir.tools.GraphPlanner;
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt.PhysicalGetVOpt;
+import com.alibaba.graphscope.common.ir.type.GraphLabelType;
+import com.alibaba.graphscope.common.ir.type.GraphSchemaType;
+import com.alibaba.graphscope.gaia.proto.GraphAlgebra;
+import com.alibaba.graphscope.gaia.proto.GraphAlgebraPhysical;
+import com.alibaba.graphscope.gaia.proto.OuterExpression;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rel.logical.LogicalJoin;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.*;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.commons.lang3.ObjectUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import javax.annotation.Nullable;
+
+public class GraphRelToProtoConverter extends GraphShuttle {
+    private static final Logger logger = LoggerFactory.getLogger(GraphRelToProtoConverter.class);
+    private final boolean isColumnId;
+    private final RexBuilder rexBuilder;
+    private final Configs graphConfig;
+    private GraphAlgebraPhysical.PhysicalPlan.Builder physicalBuilder;
+
+    public GraphRelToProtoConverter(
+            boolean isColumnId,
+            Configs configs,
+            GraphAlgebraPhysical.PhysicalPlan.Builder physicalBuilder) {
+        this.isColumnId = isColumnId;
+        this.rexBuilder = GraphPlanner.rexBuilderFactory.apply(configs);
+        this.graphConfig = configs;
+        this.physicalBuilder = physicalBuilder;
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalSource source) {
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.Scan.Builder scanBuilder = GraphAlgebraPhysical.Scan.newBuilder();
+        RexNode uniqueKeyFilters = source.getUniqueKeyFilters();
+        if (uniqueKeyFilters != null) {
+            GraphAlgebra.IndexPredicate indexPredicate = buildIndexPredicates(uniqueKeyFilters);
+            scanBuilder.setIdxPredicate(indexPredicate);
+        }
+        scanBuilder.setParams(buildQueryParams(source));
+        if (source.getAliasId() != AliasInference.DEFAULT_ID) {
+            scanBuilder.setAlias(Utils.asAliasId(source.getAliasId()));
+        }
+        scanBuilder.setScanOpt(Utils.protoScanOpt(source.getOpt()));
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setScan(scanBuilder));
+        oprBuilder.addAllMetaData(Utils.physicalProtoRowType(source.getRowType(), isColumnId));
+        physicalBuilder.addPlan(oprBuilder.build());
+        return source;
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalExpand expand) {
+        visitChildren(expand);
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.EdgeExpand.Builder edgeExpand = buildEdgeExpand(expand);
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setEdge(edgeExpand));
+        oprBuilder.addAllMetaData(Utils.physicalProtoRowType(expand.getRowType(), isColumnId));
+        physicalBuilder.addPlan(oprBuilder.build());
+        return expand;
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalGetV getV) {
+        visitChildren(getV);
+        // convert getV:
+        // if there is no filter, build getV(adj)
+        // otherwise, build getV(adj) + auxilia(filter)
+        if (ObjectUtils.isEmpty(getV.getFilters())) {
+            GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                    GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+            GraphAlgebraPhysical.GetV.Builder getVertex = buildGetV(getV);
+            oprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setVertex(getVertex));
+            oprBuilder.addAllMetaData(Utils.physicalProtoRowType(getV.getRowType(), isColumnId));
+            physicalBuilder.addPlan(oprBuilder.build());
+            return getV;
+        } else {
+            // build getV(adj) + auxilia(filter) if there is a filter in getV
+            GraphAlgebraPhysical.PhysicalOpr.Builder adjOprBuilder =
+                    GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+            GraphAlgebraPhysical.GetV.Builder adjVertexBuilder =
+                    GraphAlgebraPhysical.GetV.newBuilder();
+            adjVertexBuilder.setOpt(
+                    Utils.protoGetVOpt(PhysicalGetVOpt.valueOf(getV.getOpt().name())));
+            // 1. build adjV without filter
+            GraphAlgebra.QueryParams.Builder adjParamsBuilder = defaultQueryParams();
+            addQueryTables(adjParamsBuilder, getGraphLabels(getV).getLabelsEntry());
+            adjVertexBuilder.setParams(adjParamsBuilder);
+            if (getV.getStartAlias().getAliasId() != AliasInference.DEFAULT_ID) {
+                adjVertexBuilder.setTag(Utils.asAliasId(getV.getStartAlias().getAliasId()));
+            }
+            if (getV.getAliasId() != AliasInference.DEFAULT_ID) {
+                adjVertexBuilder.setAlias(Utils.asAliasId(getV.getAliasId()));
+            }
+            adjOprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
+                            .setVertex(adjVertexBuilder));
+            adjOprBuilder.addAllMetaData(Utils.physicalProtoRowType(getV.getRowType(), isColumnId));
+            physicalBuilder.addPlan(adjOprBuilder.build());
+
+            // 2. build auxilia(filter)
+            GraphAlgebraPhysical.PhysicalOpr.Builder auxiliaOprBuilder =
+                    GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+            GraphAlgebraPhysical.GetV.Builder auxiliaBuilder =
+                    GraphAlgebraPhysical.GetV.newBuilder();
+            auxiliaBuilder.setOpt(Utils.protoGetVOpt(PhysicalGetVOpt.ITSELF));
+            GraphAlgebra.QueryParams.Builder auxiliaParamsBuilder = defaultQueryParams();
+            addQueryFilters(auxiliaParamsBuilder, getV.getFilters());
+            auxiliaBuilder.setParams(auxiliaParamsBuilder);
+            auxiliaOprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
+                            .setVertex(auxiliaBuilder));
+            auxiliaOprBuilder.addAllMetaData(
+                    Utils.physicalProtoRowType(getV.getRowType(), isColumnId));
+            physicalBuilder.addPlan(auxiliaOprBuilder.build());
+            return getV;
+        }
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalPathExpand pxd) {
+        visitChildren(pxd);
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.PathExpand.Builder pathExpandBuilder =
+                GraphAlgebraPhysical.PathExpand.newBuilder();
+        GraphAlgebraPhysical.PathExpand.ExpandBase.Builder expandBaseBuilder =
+                GraphAlgebraPhysical.PathExpand.ExpandBase.newBuilder();
+        RelNode fused = pxd.getFused();
+        if (fused != null) {
+            // the case that expand base is fused
+            if (fused instanceof GraphPhysicalGetV) {
+                // fused into expand + auxilia
+                GraphPhysicalGetV fusedGetV = (GraphPhysicalGetV) fused;
+                GraphAlgebraPhysical.GetV.Builder auxilia = buildAuxilia(fusedGetV);
+                expandBaseBuilder.setGetV(auxilia);
+                if (fusedGetV.getInput() instanceof GraphPhysicalExpand) {
+                    GraphPhysicalExpand fusedExpand = (GraphPhysicalExpand) fusedGetV.getInput();
+                    GraphAlgebraPhysical.EdgeExpand.Builder expand = buildEdgeExpand(fusedExpand);
+                    expandBaseBuilder.setEdgeExpand(expand);
+                } else {
+                    throw new UnsupportedOperationException(
+                            "unsupported fused plan in path expand base: "
+                                    + fusedGetV.getInput().getClass().getName());
+                }
+            } else if (fused instanceof GraphPhysicalExpand) {
+                // fused into expand
+                GraphPhysicalExpand fusedExpand = (GraphPhysicalExpand) fused;
+                GraphAlgebraPhysical.EdgeExpand.Builder expand = buildEdgeExpand(fusedExpand);
+                expandBaseBuilder.setEdgeExpand(expand);
+            } else {
+                throw new UnsupportedOperationException(
+                        "unsupported fused plan in path expand base");
+            }
+        } else {
+            // the case that expand base is not fused
+            GraphAlgebraPhysical.EdgeExpand.Builder expand =
+                    buildEdgeExpand((GraphLogicalExpand) pxd.getExpand());
+            GraphAlgebraPhysical.GetV.Builder getV = buildGetV((GraphLogicalGetV) pxd.getGetV());
+            expandBaseBuilder.setEdgeExpand(expand);
+            expandBaseBuilder.setGetV(getV);
+        }
+        pathExpandBuilder.setBase(expandBaseBuilder);
+        pathExpandBuilder.setPathOpt(Utils.protoPathOpt(pxd.getPathOpt()));
+        pathExpandBuilder.setResultOpt(Utils.protoPathResultOpt(pxd.getResultOpt()));
+        GraphAlgebra.Range range = buildRange(pxd.getOffset(), pxd.getFetch());
+        pathExpandBuilder.setHopRange(range);
+        if (pxd.getAliasId() != AliasInference.DEFAULT_ID) {
+            pathExpandBuilder.setAlias(Utils.asAliasId(pxd.getAliasId()));
+        }
+        if (pxd.getStartAlias().getAliasId() != AliasInference.DEFAULT_ID) {
+            pathExpandBuilder.setStartTag(Utils.asAliasId(pxd.getStartAlias().getAliasId()));
+        }
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setPath(pathExpandBuilder));
+
+        physicalBuilder.addPlan(oprBuilder.build());
+        return pxd;
+    }
+
+    @Override
+    public RelNode visit(GraphPhysicalExpand physicalExpand) {
+        visitChildren(physicalExpand);
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.EdgeExpand.Builder edgeExpand = buildEdgeExpand(physicalExpand);
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setEdge(edgeExpand));
+        oprBuilder.addAllMetaData(
+                Utils.physicalProtoRowType(physicalExpand.getRowType(), isColumnId));
+        physicalBuilder.addPlan(oprBuilder.build());
+        return physicalExpand;
+    }
+
+    @Override
+    public RelNode visit(GraphPhysicalGetV physicalGetV) {
+        visitChildren(physicalGetV);
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.GetV.Builder auxilia = buildAuxilia(physicalGetV);
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setVertex(auxilia));
+        oprBuilder.addAllMetaData(
+                Utils.physicalProtoRowType(physicalGetV.getRowType(), isColumnId));
+        physicalBuilder.addPlan(oprBuilder.build());
+        return physicalGetV;
+    }
+
+    @Override
+    public RelNode visit(LogicalFilter filter) {
+        visitChildren(filter);
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebra.Select.Builder selectBuilder = GraphAlgebra.Select.newBuilder();
+        OuterExpression.Expression exprProto =
+                filter.getCondition()
+                        .accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder));
+        selectBuilder.setPredicate(exprProto);
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setSelect(selectBuilder));
+        physicalBuilder.addPlan(oprBuilder.build());
+        return filter;
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalProject project) {
+        visitChildren(project);
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.Project.Builder projectBuilder =
+                GraphAlgebraPhysical.Project.newBuilder();
+        projectBuilder.setIsAppend(project.isAppend());
+        List<RelDataTypeField> fields = project.getRowType().getFieldList();
+        for (int i = 0; i < project.getProjects().size(); ++i) {
+            OuterExpression.Expression expression =
+                    project.getProjects()
+                            .get(i)
+                            .accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder));
+            int aliasId = fields.get(i).getIndex();
+            GraphAlgebraPhysical.Project.ExprAlias.Builder projectExprAliasBuilder =
+                    GraphAlgebraPhysical.Project.ExprAlias.newBuilder();
+            projectExprAliasBuilder.setExpr(expression);
+            if (aliasId != AliasInference.DEFAULT_ID) {
+                projectExprAliasBuilder.setAlias(Utils.asAliasId(aliasId));
+            }
+            projectBuilder.addMappings(projectExprAliasBuilder.build());
+        }
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setProject(projectBuilder));
+        oprBuilder.addAllMetaData(Utils.physicalProtoRowType(project.getRowType(), isColumnId));
+        physicalBuilder.addPlan(oprBuilder.build());
+        return project;
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalAggregate aggregate) {
+        visitChildren(aggregate);
+        List<RelDataTypeField> fields = aggregate.getRowType().getFieldList();
+        List<GraphAggCall> groupCalls = aggregate.getAggCalls();
+        GraphGroupKeys keys = aggregate.getGroupKey();
+        if (groupCalls.isEmpty()) { // transform to project + dedup by keys
+            Preconditions.checkArgument(
+                    keys.groupKeyCount() > 0,
+                    "group keys should not be empty while group calls is empty");
+            GraphAlgebraPhysical.Project.Builder projectBuilder =
+                    GraphAlgebraPhysical.Project.newBuilder();
+            for (int i = 0; i < keys.groupKeyCount(); ++i) {
+                RexNode var = keys.getVariables().get(i);
+                Preconditions.checkArgument(
+                        var instanceof RexGraphVariable,
+                        "each group key should be type %s, but is %s",
+                        RexGraphVariable.class,
+                        var.getClass());
+                OuterExpression.Expression expr =
+                        var.accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder));
+                int aliasId;
+                if (i >= fields.size()
+                        || (aliasId = fields.get(i).getIndex()) == AliasInference.DEFAULT_ID) {
+                    throw new IllegalArgumentException(
+                            "each group key should have an alias if need dedup");
+                }
+                GraphAlgebraPhysical.Project.ExprAlias.Builder projectExprAliasBuilder =
+                        GraphAlgebraPhysical.Project.ExprAlias.newBuilder();
+                projectExprAliasBuilder.setExpr(expr);
+                if (aliasId != AliasInference.DEFAULT_ID) {
+                    projectExprAliasBuilder.setAlias(Utils.asAliasId(aliasId));
+                }
+                projectBuilder.addMappings(projectExprAliasBuilder.build());
+            }
+            GraphAlgebra.Dedup.Builder dedupBuilder = GraphAlgebra.Dedup.newBuilder();
+            for (int i = 0; i < keys.groupKeyCount(); ++i) {
+                RelDataTypeField field = fields.get(i);
+                RexVariable rexVar =
+                        RexGraphVariable.of(
+                                field.getIndex(),
+                                AliasInference.DEFAULT_COLUMN_ID,
+                                field.getName(),
+                                field.getType());
+                OuterExpression.Variable exprVar =
+                        rexVar.accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
+                                .getOperators(0)
+                                .getVar();
+                dedupBuilder.addKeys(exprVar);
+            }
+            GraphAlgebraPhysical.PhysicalOpr.Builder projectOprBuilder =
+                    GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+            projectOprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
+                            .setProject(projectBuilder));
+            GraphAlgebraPhysical.PhysicalOpr.Builder dedupOprBuilder =
+                    GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+            dedupOprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setDedup(dedupBuilder));
+            physicalBuilder.addPlan(projectOprBuilder.build());
+            physicalBuilder.addPlan(dedupOprBuilder.build());
+        } else {
+            GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                    GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+            GraphAlgebraPhysical.GroupBy.Builder groupByBuilder =
+                    GraphAlgebraPhysical.GroupBy.newBuilder();
+            for (int i = 0; i < keys.groupKeyCount(); ++i) {
+                RexNode var = keys.getVariables().get(i);
+                Preconditions.checkArgument(
+                        var instanceof RexGraphVariable,
+                        "each group key should be type %s, but is %s",
+                        RexGraphVariable.class,
+                        var.getClass());
+                OuterExpression.Variable exprVar =
+                        var.accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
+                                .getOperators(0)
+                                .getVar();
+                int aliasId = fields.get(i).getIndex();
+                GraphAlgebraPhysical.GroupBy.KeyAlias.Builder keyAliasBuilder =
+                        GraphAlgebraPhysical.GroupBy.KeyAlias.newBuilder();
+                keyAliasBuilder.setKey(exprVar);
+                if (aliasId != AliasInference.DEFAULT_ID) {
+                    keyAliasBuilder.setAlias(Utils.asAliasId(aliasId));
+                }
+                groupByBuilder.addMappings(keyAliasBuilder);
+            }
+            for (int i = 0; i < groupCalls.size(); ++i) {
+                List<RexNode> operands = groupCalls.get(i).getOperands();
+                if (operands.isEmpty()) {
+                    throw new IllegalArgumentException(
+                            "operands in aggregate call should not be empty");
+                } else if (operands.size() > 1) {
+                    throw new UnsupportedOperationException(
+                            "aggregate on multiple variables is unsupported yet");
+                }
+                GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate aggOpt =
+                        Utils.protoAggOpt(groupCalls.get(i));
+                int aliasId = fields.get(i + keys.groupKeyCount()).getIndex();
+                Preconditions.checkArgument(
+                        operands.get(0) instanceof RexGraphVariable,
+                        "each expression in aggregate call should be type %s, but is %s",
+                        RexGraphVariable.class,
+                        operands.get(0).getClass());
+                OuterExpression.Variable var =
+                        operands.get(0)
+                                .accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
+                                .getOperators(0)
+                                .getVar();
+                GraphAlgebraPhysical.GroupBy.AggFunc.Builder aggFnAliasBuilder =
+                        GraphAlgebraPhysical.GroupBy.AggFunc.newBuilder();
+                aggFnAliasBuilder.setAggregate(aggOpt);
+                aggFnAliasBuilder.addVars(var);
+                if (aliasId != AliasInference.DEFAULT_ID) {
+                    aggFnAliasBuilder.setAlias(Utils.asAliasId(aliasId));
+                }
+                groupByBuilder.addFunctions(aggFnAliasBuilder);
+            }
+            oprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
+                            .setGroupBy(groupByBuilder));
+            oprBuilder.addAllMetaData(
+                    Utils.physicalProtoRowType(aggregate.getRowType(), isColumnId));
+            physicalBuilder.addPlan(oprBuilder.build());
+        }
+        return aggregate;
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalDedupBy dedupBy) {
+        visitChildren(dedupBy);
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebra.Dedup.Builder dedupBuilder = GraphAlgebra.Dedup.newBuilder();
+        for (RexNode expr : dedupBy.getDedupByKeys()) {
+            Preconditions.checkArgument(
+                    expr instanceof RexGraphVariable,
+                    "each expression in dedup by should be type %s, but is %s",
+                    RexGraphVariable.class,
+                    expr.getClass());
+            OuterExpression.Variable var =
+                    expr.accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
+                            .getOperators(0)
+                            .getVar();
+            dedupBuilder.addKeys(var);
+        }
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setDedup(dedupBuilder));
+        physicalBuilder.addPlan(oprBuilder.build());
+        return dedupBy;
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalSort sort) {
+        visitChildren(sort);
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        List<RelFieldCollation> collations = sort.getCollation().getFieldCollations();
+        if (!collations.isEmpty()) {
+            GraphAlgebra.OrderBy.Builder orderByBuilder = GraphAlgebra.OrderBy.newBuilder();
+            for (int i = 0; i < collations.size(); ++i) {
+                GraphAlgebra.OrderBy.OrderingPair.Builder orderingPairBuilder =
+                        GraphAlgebra.OrderBy.OrderingPair.newBuilder();
+                RexGraphVariable expr = ((GraphFieldCollation) collations.get(i)).getVariable();
+                OuterExpression.Variable var =
+                        expr.accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
+                                .getOperators(0)
+                                .getVar();
+
+                orderingPairBuilder.setKey(var);
+                orderingPairBuilder.setOrder(Utils.protoOrderOpt(collations.get(i).getDirection()));
+                orderByBuilder.addPairs(orderingPairBuilder.build());
+            }
+            if (sort.offset != null || sort.fetch != null) {
+                orderByBuilder.setLimit(buildRange(sort.offset, sort.fetch));
+            }
+            oprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
+                            .setOrderBy(orderByBuilder));
+        } else {
+            GraphAlgebra.Limit.Builder limitBuilder = GraphAlgebra.Limit.newBuilder();
+            limitBuilder.setRange(buildRange(sort.offset, sort.fetch));
+            oprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setLimit(limitBuilder));
+        }
+        physicalBuilder.addPlan(oprBuilder.build());
+        return sort;
+    }
+
+    @Override
+    public RelNode visit(LogicalJoin join) {
+        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.Join.Builder joinBuilder = GraphAlgebraPhysical.Join.newBuilder();
+        joinBuilder.setJoinKind(Utils.protoJoinKind(join.getJoinType()));
+        List<RexNode> conditions = RelOptUtil.conjunctions(join.getCondition());
+        String errorMessage =
+                "join condition in ir core should be 'AND' of equal conditions, each equal"
+                        + " condition has two variables as operands";
+        Preconditions.checkArgument(!conditions.isEmpty(), errorMessage);
+        for (RexNode condition : conditions) {
+            List<RexGraphVariable> leftRightVars = getLeftRightVariables(condition);
+            Preconditions.checkArgument(leftRightVars.size() == 2, errorMessage);
+            OuterExpression.Variable leftVar =
+                    leftRightVars
+                            .get(0)
+                            .accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
+                            .getOperators(0)
+                            .getVar();
+            OuterExpression.Variable rightVar =
+                    leftRightVars
+                            .get(1)
+                            .accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
+                            .getOperators(0)
+                            .getVar();
+            joinBuilder.addLeftKeys(leftVar);
+            joinBuilder.addRightKeys(rightVar);
+        }
+
+        GraphAlgebraPhysical.PhysicalPlan.Builder leftPlanBuilder =
+                GraphAlgebraPhysical.PhysicalPlan.newBuilder();
+        GraphAlgebraPhysical.PhysicalPlan.Builder rightPlanBuilder =
+                GraphAlgebraPhysical.PhysicalPlan.newBuilder();
+
+        RelNode left = join.getLeft();
+        left.accept(new GraphRelToProtoConverter(isColumnId, graphConfig, leftPlanBuilder));
+        RelNode right = join.getRight();
+        right.accept(new GraphRelToProtoConverter(isColumnId, graphConfig, rightPlanBuilder));
+        joinBuilder.setLeftPlan(leftPlanBuilder);
+        joinBuilder.setRightPlan(rightPlanBuilder);
+        oprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setJoin(joinBuilder));
+        physicalBuilder.addPlan(oprBuilder.build());
+        return join;
+    }
+
+    private List<RexGraphVariable> getLeftRightVariables(RexNode condition) {
+        List<RexGraphVariable> vars = Lists.newArrayList();
+        if (condition instanceof RexCall) {
+            RexCall call = (RexCall) condition;
+            if (call.getOperator().getKind() == SqlKind.EQUALS) {
+                RexNode left = call.getOperands().get(0);
+                RexNode right = call.getOperands().get(1);
+                if (left instanceof RexGraphVariable && right instanceof RexGraphVariable) {
+                    vars.add((RexGraphVariable) left);
+                    vars.add((RexGraphVariable) right);
+                }
+            }
+        }
+        return vars;
+    }
+
+    private GraphAlgebraPhysical.EdgeExpand.Builder buildEdgeExpand(
+            GraphLogicalExpand expand, GraphOpt.PhysicalExpandOpt opt, int aliasId) {
+        GraphAlgebraPhysical.EdgeExpand.Builder expandBuilder =
+                GraphAlgebraPhysical.EdgeExpand.newBuilder();
+        expandBuilder.setDirection(Utils.protoExpandDirOpt(expand.getOpt()));
+        expandBuilder.setParams(buildQueryParams(expand));
+        if (aliasId != AliasInference.DEFAULT_ID) {
+            expandBuilder.setAlias(Utils.asAliasId(aliasId));
+        }
+        if (expand.getStartAlias().getAliasId() != AliasInference.DEFAULT_ID) {
+            expandBuilder.setVTag(Utils.asAliasId(expand.getStartAlias().getAliasId()));
+        }
+        expandBuilder.setExpandOpt(Utils.protoExpandOpt(opt));
+        return expandBuilder;
+    }
+
+    private GraphAlgebraPhysical.EdgeExpand.Builder buildEdgeExpand(GraphLogicalExpand expand) {
+        return buildEdgeExpand(expand, GraphOpt.PhysicalExpandOpt.EDGE, expand.getAliasId());
+    }
+
+    private GraphAlgebraPhysical.EdgeExpand.Builder buildEdgeExpand(
+            GraphPhysicalExpand physicalExpand) {
+        return buildEdgeExpand(
+                physicalExpand.getFusedExpand(),
+                physicalExpand.getPhysicalOpt(),
+                physicalExpand.getAliasId());
+    }
+
+    private GraphAlgebraPhysical.GetV.Builder buildVertex(
+            GraphLogicalGetV getV, GraphOpt.PhysicalGetVOpt opt) {
+        GraphAlgebraPhysical.GetV.Builder vertexBuilder = GraphAlgebraPhysical.GetV.newBuilder();
+        vertexBuilder.setOpt(Utils.protoGetVOpt(opt));
+        vertexBuilder.setParams(buildQueryParams(getV));
+        if (getV.getAliasId() != AliasInference.DEFAULT_ID) {
+            vertexBuilder.setAlias(Utils.asAliasId(getV.getAliasId()));
+        }
+        if (getV.getStartAlias().getAliasId() != AliasInference.DEFAULT_ID) {
+            vertexBuilder.setTag(Utils.asAliasId(getV.getStartAlias().getAliasId()));
+        }
+        return vertexBuilder;
+    }
+
+    private GraphAlgebraPhysical.GetV.Builder buildGetV(GraphLogicalGetV getV) {
+        return buildVertex(getV, PhysicalGetVOpt.valueOf(getV.getOpt().name()));
+    }
+
+    private GraphAlgebraPhysical.GetV.Builder buildAuxilia(GraphPhysicalGetV getV) {
+        return buildVertex(getV.getFusedGetV(), PhysicalGetVOpt.ITSELF);
+    }
+
+    private GraphAlgebra.Range buildRange(RexNode offset, RexNode fetch) {
+        if (offset != null && !(offset instanceof RexLiteral)
+                || fetch != null && !(fetch instanceof RexLiteral)) {
+            throw new IllegalArgumentException(
+                    "can not get INTEGER hops from types instead of RexLiteral");
+        }
+        GraphAlgebra.Range.Builder rangeBuilder = GraphAlgebra.Range.newBuilder();
+        rangeBuilder.setLower(
+                offset == null ? 0 : ((Number) ((RexLiteral) offset).getValue()).intValue());
+        rangeBuilder.setUpper(
+                fetch == null
+                        ? Integer.MAX_VALUE
+                        : ((Number) ((RexLiteral) fetch).getValue()).intValue());
+        return rangeBuilder.build();
+    }
+
+    private GraphAlgebra.IndexPredicate buildIndexPredicates(RexNode uniqueKeyFilters) {
+        GraphAlgebra.IndexPredicate indexPredicate =
+                uniqueKeyFilters.accept(
+                        new RexToIndexPbConverter(true, this.isColumnId, this.rexBuilder));
+        return indexPredicate;
+    }
+
+    private GraphLabelType getGraphLabels(AbstractBindableTableScan tableScan) {
+        List<RelDataTypeField> fields = tableScan.getRowType().getFieldList();
+        Preconditions.checkArgument(
+                !fields.isEmpty() && fields.get(0).getType() instanceof GraphSchemaType,
+                "data type of graph operators should be %s ",
+                GraphSchemaType.class);
+        GraphSchemaType schemaType = (GraphSchemaType) fields.get(0).getType();
+        return schemaType.getLabelType();
+    }
+
+    private GraphAlgebra.QueryParams.Builder defaultQueryParams() {
+        GraphAlgebra.QueryParams.Builder paramsBuilder = GraphAlgebra.QueryParams.newBuilder();
+        // TODO: currently no sample rate fused into tableScan, so directly set 1.0 as default.
+        paramsBuilder.setSampleRatio(1.0);
+        return paramsBuilder;
+    }
+
+    private void addQueryTables(
+            GraphAlgebra.QueryParams.Builder paramsBuilder, List<GraphLabelType.Entry> labels) {
+        Set<Integer> uniqueLabelIds =
+                labels.stream().map(k -> k.getLabelId()).collect(Collectors.toSet());
+        uniqueLabelIds.forEach(
+                k -> {
+                    paramsBuilder.addTables(Utils.asNameOrId(k));
+                });
+    }
+
+    private void addQueryFilters(
+            GraphAlgebra.QueryParams.Builder paramsBuilder,
+            @Nullable ImmutableList<RexNode> filters) {
+        if (ObjectUtils.isNotEmpty(filters)) {
+            OuterExpression.Expression expression =
+                    filters.get(0)
+                            .accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder));
+            paramsBuilder.setPredicate(expression);
+        }
+    }
+
+    private GraphAlgebra.QueryParams.Builder buildQueryParams(AbstractBindableTableScan tableScan) {
+        GraphAlgebra.QueryParams.Builder paramsBuilder = defaultQueryParams();
+        addQueryTables(paramsBuilder, getGraphLabels(tableScan).getLabelsEntry());
+        addQueryFilters(paramsBuilder, tableScan.getFilters());
+        return paramsBuilder;
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalSingleMatch match) {
+        throw new UnsupportedOperationException("Unimplemented method 'visit'");
+    }
+
+    @Override
+    public RelNode visit(GraphLogicalMultiMatch match) {
+        throw new UnsupportedOperationException("Unimplemented method 'visit'");
+    }
+}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/Utils.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/Utils.java
index dfd45fd92de3..e57e88721a3c 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/Utils.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/Utils.java
@@ -27,11 +27,14 @@
 import com.alibaba.graphscope.gaia.proto.Common;
 import com.alibaba.graphscope.gaia.proto.DataType;
 import com.alibaba.graphscope.gaia.proto.GraphAlgebra;
+import com.alibaba.graphscope.gaia.proto.GraphAlgebraPhysical;
 import com.alibaba.graphscope.gaia.proto.OuterExpression;
 import com.google.common.base.Preconditions;
 import com.google.protobuf.Int32Value;
 
 import org.apache.calcite.avatica.util.TimeUnit;
+import org.apache.calcite.rel.RelFieldCollation;
+import org.apache.calcite.rel.core.JoinRelType;
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexLiteral;
 import org.apache.calcite.sql.SqlOperator;
@@ -433,4 +436,188 @@ public static final Aggregate protoAggFn(GraphAggCall aggCall) {
                                 + " is unsupported yet");
         }
     }
+
+    public static final GraphAlgebra.OrderBy.OrderingPair.Order protoOrderOpt(
+            RelFieldCollation.Direction direction) {
+        switch (direction) {
+            case ASCENDING:
+                return GraphAlgebra.OrderBy.OrderingPair.Order.ASC;
+            case DESCENDING:
+                return GraphAlgebra.OrderBy.OrderingPair.Order.DESC;
+            case CLUSTERED:
+                return GraphAlgebra.OrderBy.OrderingPair.Order.SHUFFLE;
+            default:
+                throw new UnsupportedOperationException(
+                        "direction " + direction + " in order is unsupported yet");
+        }
+    }
+
+    public static final GraphAlgebraPhysical.EdgeExpand.Direction protoExpandDirOpt(
+            GraphOpt.Expand opt) {
+        switch (opt) {
+            case OUT:
+                return GraphAlgebraPhysical.EdgeExpand.Direction.OUT;
+            case IN:
+                return GraphAlgebraPhysical.EdgeExpand.Direction.IN;
+            case BOTH:
+                return GraphAlgebraPhysical.EdgeExpand.Direction.BOTH;
+            default:
+                throw new UnsupportedOperationException(
+                        "opt " + opt + " in expand is unsupported yet");
+        }
+    }
+
+    public static final GraphAlgebraPhysical.EdgeExpand.ExpandOpt protoExpandOpt(
+            GraphOpt.PhysicalExpandOpt opt) {
+        switch (opt) {
+            case EDGE:
+                return GraphAlgebraPhysical.EdgeExpand.ExpandOpt.EDGE;
+            case VERTEX:
+                return GraphAlgebraPhysical.EdgeExpand.ExpandOpt.VERTEX;
+            case DEGREE:
+                return GraphAlgebraPhysical.EdgeExpand.ExpandOpt.DEGREE;
+            default:
+                throw new UnsupportedOperationException(
+                        "opt " + opt + " in expand is unsupported yet");
+        }
+    }
+
+    public static final GraphAlgebraPhysical.GetV.VOpt protoGetVOpt(GraphOpt.PhysicalGetVOpt opt) {
+        switch (opt) {
+            case START:
+                return GraphAlgebraPhysical.GetV.VOpt.START;
+            case END:
+                return GraphAlgebraPhysical.GetV.VOpt.END;
+            case OTHER:
+                return GraphAlgebraPhysical.GetV.VOpt.OTHER;
+            case BOTH:
+                return GraphAlgebraPhysical.GetV.VOpt.BOTH;
+            case ITSELF:
+                return GraphAlgebraPhysical.GetV.VOpt.ITSELF;
+            default:
+                throw new UnsupportedOperationException(
+                        "opt " + opt + " in getV is unsupported yet");
+        }
+    }
+
+    public static final GraphAlgebraPhysical.PathExpand.PathOpt protoPathOpt(
+            GraphOpt.PathExpandPath opt) {
+        switch (opt) {
+            case ARBITRARY:
+                return GraphAlgebraPhysical.PathExpand.PathOpt.ARBITRARY;
+            case SIMPLE:
+                return GraphAlgebraPhysical.PathExpand.PathOpt.SIMPLE;
+            default:
+                throw new UnsupportedOperationException(
+                        "opt " + opt + " in path is unsupported yet");
+        }
+    }
+
+    public static final GraphAlgebraPhysical.PathExpand.ResultOpt protoPathResultOpt(
+            GraphOpt.PathExpandResult opt) {
+        switch (opt) {
+            case END_V:
+                return GraphAlgebraPhysical.PathExpand.ResultOpt.END_V;
+            case ALL_V:
+                return GraphAlgebraPhysical.PathExpand.ResultOpt.ALL_V;
+            case ALL_V_E:
+                return GraphAlgebraPhysical.PathExpand.ResultOpt.ALL_V_E;
+            default:
+                throw new UnsupportedOperationException(
+                        "result opt " + opt + " in path is unsupported yet");
+        }
+    }
+
+    public static final GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate protoAggOpt(
+            GraphAggCall aggCall) {
+        switch (aggCall.getAggFunction().kind) {
+            case COUNT:
+                return aggCall.isDistinct()
+                        ? GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.COUNT_DISTINCT
+                        : GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.COUNT;
+            case COLLECT:
+                return aggCall.isDistinct()
+                        ? GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.TO_SET
+                        : GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.TO_LIST;
+            case SUM:
+                return GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.SUM;
+            case SUM0:
+                return GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.SUM;
+            case AVG:
+                return GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.AVG;
+            case MIN:
+                return GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.MIN;
+            case MAX:
+                return GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.MAX;
+            case FIRST_VALUE:
+                return GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate.FIRST;
+            default:
+                throw new UnsupportedOperationException(
+                        "aggregate opt " + aggCall.getAggFunction().kind + " is unsupported yet");
+        }
+    }
+
+    public static final GraphAlgebraPhysical.Join.JoinKind protoJoinKind(JoinRelType joinRelType) {
+        switch (joinRelType) {
+            case INNER:
+                return GraphAlgebraPhysical.Join.JoinKind.INNER;
+            case LEFT:
+                return GraphAlgebraPhysical.Join.JoinKind.LEFT_OUTER;
+            case RIGHT:
+                return GraphAlgebraPhysical.Join.JoinKind.RIGHT_OUTER;
+            case FULL:
+                return GraphAlgebraPhysical.Join.JoinKind.FULL_OUTER;
+            case SEMI:
+                return GraphAlgebraPhysical.Join.JoinKind.SEMI;
+            case ANTI:
+                return GraphAlgebraPhysical.Join.JoinKind.ANTI;
+            default:
+                throw new UnsupportedOperationException(
+                        "join type " + joinRelType + " is unsupported yet");
+        }
+    }
+
+    public static final GraphAlgebraPhysical.Scan.ScanOpt protoScanOpt(GraphOpt.Source opt) {
+        switch (opt) {
+            case VERTEX:
+                return GraphAlgebraPhysical.Scan.ScanOpt.VERTEX;
+            case EDGE:
+                return GraphAlgebraPhysical.Scan.ScanOpt.EDGE;
+            default:
+                throw new UnsupportedOperationException("scan opt " + opt + " is unsupported yet");
+        }
+    }
+
+    public static Common.NameOrId asNameOrId(int id) {
+        Common.NameOrId.Builder builder = Common.NameOrId.newBuilder();
+        builder.setId(id);
+        return builder.build();
+    }
+
+    public static com.google.protobuf.Int32Value asAliasId(int id) {
+        return com.google.protobuf.Int32Value.of(id);
+    }
+
+    public static final List<GraphAlgebraPhysical.PhysicalOpr.MetaData> physicalProtoRowType(
+            RelDataType rowType, boolean isColumnId) {
+        switch (rowType.getSqlTypeName()) {
+            case ROW:
+                return rowType.getFieldList().stream()
+                        .map(
+                                k ->
+                                        GraphAlgebraPhysical.PhysicalOpr.MetaData.newBuilder()
+                                                .setType(
+                                                        com.alibaba.graphscope.common.ir.runtime
+                                                                .proto.Utils.protoIrDataType(
+                                                                k.getType(), isColumnId))
+                                                .setAlias(k.getIndex())
+                                                .build())
+                        .collect(Collectors.toList());
+            default:
+                throw new UnsupportedOperationException(
+                        "convert type "
+                                + rowType.getSqlTypeName()
+                                + " to List<MetaData> is unsupported");
+        }
+    }
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphPlanner.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphPlanner.java
index d6808af22879..dc692e7539b6 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphPlanner.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphPlanner.java
@@ -17,12 +17,14 @@
 package com.alibaba.graphscope.common.ir.tools;
 
 import com.alibaba.graphscope.common.config.Configs;
+import com.alibaba.graphscope.common.config.FrontendConfig;
 import com.alibaba.graphscope.common.config.PlannerConfig;
 import com.alibaba.graphscope.common.ir.meta.procedure.StoredProcedureMeta;
 import com.alibaba.graphscope.common.ir.meta.reader.LocalMetaDataReader;
 import com.alibaba.graphscope.common.ir.meta.schema.GraphOptSchema;
 import com.alibaba.graphscope.common.ir.meta.schema.IrGraphSchema;
 import com.alibaba.graphscope.common.ir.planner.rules.DegreeFusionRule;
+import com.alibaba.graphscope.common.ir.planner.rules.ExpandGetVFusionRule;
 import com.alibaba.graphscope.common.ir.planner.rules.FieldTrimRule;
 import com.alibaba.graphscope.common.ir.planner.rules.FilterMatchRule;
 import com.alibaba.graphscope.common.ir.planner.rules.NotMatchToAntiJoinRule;
@@ -30,6 +32,7 @@
 import com.alibaba.graphscope.common.ir.runtime.PhysicalPlan;
 import com.alibaba.graphscope.common.ir.runtime.ProcedurePhysicalBuilder;
 import com.alibaba.graphscope.common.ir.runtime.ffi.FfiPhysicalBuilder;
+import com.alibaba.graphscope.common.ir.runtime.proto.GraphRelProtoPhysicalBuilder;
 import com.alibaba.graphscope.common.ir.type.GraphTypeFactoryImpl;
 import com.alibaba.graphscope.common.store.ExperimentalMetaFetcher;
 import com.alibaba.graphscope.common.store.IrMeta;
@@ -39,7 +42,9 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
 
-import org.apache.calcite.plan.*;
+import org.apache.calcite.plan.GraphOptCluster;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelRule;
 import org.apache.calcite.plan.hep.HepPlanner;
 import org.apache.calcite.plan.hep.HepProgram;
 import org.apache.calcite.plan.hep.HepProgramBuilder;
@@ -139,11 +144,23 @@ public PhysicalPlan planPhysical(LogicalPlan logicalPlan) {
             if (logicalPlan.isReturnEmpty()) {
                 return PhysicalPlan.createEmpty();
             } else if (logicalPlan.getRegularQuery() != null) {
-                try (PhysicalBuilder physicalBuilder =
-                        new FfiPhysicalBuilder(graphConfig, irMeta, logicalPlan)) {
-                    return physicalBuilder.build();
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
+                String physicalOpt = FrontendConfig.PHYSICAL_OPT_CONFIG.get(graphConfig);
+                if ("proto".equals(physicalOpt.toLowerCase())) {
+                    logger.info("physical type is proto");
+                    try (GraphRelProtoPhysicalBuilder physicalBuilder =
+                            new GraphRelProtoPhysicalBuilder(graphConfig, irMeta, logicalPlan)) {
+                        return physicalBuilder.build();
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
+                } else {
+                    logger.info("physical type is ffi");
+                    try (PhysicalBuilder physicalBuilder =
+                            new FfiPhysicalBuilder(graphConfig, irMeta, logicalPlan)) {
+                        return physicalBuilder.build();
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    }
                 }
             } else {
                 return new ProcedurePhysicalBuilder(logicalPlan).build();
@@ -180,6 +197,7 @@ private RelOptPlanner createRelOptPlanner(
                             .getRules()
                             .forEach(
                                     k -> {
+                                        // logical rules
                                         if (k.equals(
                                                 FilterJoinRule.FilterIntoJoinRule.class
                                                         .getSimpleName())) {
@@ -199,6 +217,17 @@ private RelOptPlanner createRelOptPlanner(
                                                     DegreeFusionRule.ExpandGetVDegreeFusionRule
                                                             .Config.DEFAULT);
                                         }
+                                        // physical rules
+                                        else if (k.equals(
+                                                ExpandGetVFusionRule.class.getSimpleName())) {
+                                            ruleConfigs.add(
+                                                    ExpandGetVFusionRule.BasicExpandGetVFusionRule
+                                                            .Config.DEFAULT);
+                                            ruleConfigs.add(
+                                                    ExpandGetVFusionRule
+                                                            .PathBaseExpandGetVFusionRule.Config
+                                                            .DEFAULT);
+                                        }
                                     });
                     HepProgramBuilder hepBuilder = HepProgram.builder();
                     ruleConfigs.forEach(
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/config/GraphOpt.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/config/GraphOpt.java
index b8a46264fd52..0b2e5aefc4f6 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/config/GraphOpt.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/config/GraphOpt.java
@@ -30,6 +30,12 @@ public enum Expand {
         BOTH,
     }
 
+    public enum PhysicalExpandOpt {
+        EDGE,
+        VERTEX,
+        DEGREE,
+    }
+
     public enum GetV {
         START,
         END,
@@ -37,6 +43,14 @@ public enum GetV {
         BOTH
     }
 
+    public enum PhysicalGetVOpt {
+        START,
+        END,
+        OTHER,
+        BOTH,
+        ITSELF,
+    }
+
     public enum Match {
         INNER,
         ANTI, // the sentence is anti, i.e. `not(as("a").out().as("b"))` in gremlin query
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/type/GraphPathType.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/type/GraphPathType.java
index a5a12bc31a99..43dc8005bd13 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/type/GraphPathType.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/type/GraphPathType.java
@@ -20,6 +20,9 @@
 import org.apache.calcite.sql.type.AbstractSqlType;
 import org.apache.calcite.sql.type.ArraySqlType;
 import org.apache.calcite.sql.type.SqlTypeName;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+import java.util.Objects;
 
 public class GraphPathType extends ArraySqlType {
     public GraphPathType(ElementType elementType) {
@@ -35,17 +38,21 @@ public ElementType getComponentType() {
      * each element type of path expand
      */
     public static class ElementType extends AbstractSqlType {
-        private final RelDataType expandType;
+        private final @Nullable RelDataType expandType;
         private final RelDataType getVType;
 
+        public ElementType(RelDataType getVType) {
+            this(null, getVType);
+        }
+
         public ElementType(RelDataType expandType, RelDataType getVType) {
             super(SqlTypeName.OTHER, false, null);
             this.expandType = expandType;
-            this.getVType = getVType;
-            this.computeDigest();
+            this.getVType = Objects.requireNonNull(getVType);
+            computeDigest();
         }
 
-        public RelDataType getExpandType() {
+        public @Nullable RelDataType getExpandType() {
             return expandType;
         }
 


Test patch:
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/GraphTypeInferenceTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/GraphTypeInferenceTest.java
index 03f8f548fa2b..ac801b05c9cc 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/GraphTypeInferenceTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/GraphTypeInferenceTest.java
@@ -109,12 +109,7 @@ public void schema_type_test_5() {
                                 Utils.mockGraphBuilder("schema/ldbc.json"))
                         .build();
         GraphPathType pathType = (GraphPathType) match.getRowType().getFieldList().get(0).getType();
-        // check label type of expand in path_expand
-        Assert.assertEquals(
-                "[EdgeLabel(KNOWS, PERSON, PERSON)]",
-                ((GraphSchemaType) pathType.getComponentType().getExpandType())
-                        .getLabelType()
-                        .toString());
+        // check label type of getV in path_expand
         Assert.assertEquals(
                 "[VertexLabel(PERSON)]",
                 ((GraphSchemaType) pathType.getComponentType().getGetVType())
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/plan/DegreeFusionTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/plan/DegreeFusionTest.java
index 6ca40ef150b2..5e7d28d70bb2 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/plan/DegreeFusionTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/plan/DegreeFusionTest.java
@@ -66,8 +66,8 @@ public void degree_fusion_0_test() {
                 "GraphLogicalAggregate(keys=[{variables=[], aliases=[]}],"
                         + " values=[[{operands=[DEFAULT], aggFunction=$SUM0, alias='cnt',"
                         + " distinct=false}]])\n"
-                        + "  GraphLogicalExpandDegree(tableConfig=[{isAll=false, tables=[knows]}],"
-                        + " alias=[DEFAULT])\n"
+                        + "  GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT], physicalOpt=[DEGREE])\n"
                         + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
                         + " alias=[DEFAULT], opt=[VERTEX])",
                 after.explain().trim());
@@ -105,8 +105,8 @@ public void degree_fusion_1_test() {
                 "GraphLogicalAggregate(keys=[{variables=[], aliases=[]}],"
                         + " values=[[{operands=[DEFAULT], aggFunction=$SUM0, alias='cnt',"
                         + " distinct=false}]])\n"
-                        + "  GraphLogicalExpandDegree(tableConfig=[{isAll=false, tables=[knows]}],"
-                        + " alias=[DEFAULT])\n"
+                        + "  GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT], physicalOpt=[DEGREE])\n"
                         + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
                         + " alias=[DEFAULT], opt=[VERTEX])",
                 after.explain().trim());
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/plan/ExpandGetVFusionTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/plan/ExpandGetVFusionTest.java
new file mode 100644
index 000000000000..aafe9b8b0fe2
--- /dev/null
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/plan/ExpandGetVFusionTest.java
@@ -0,0 +1,504 @@
+/*
+ * 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.ir.plan;
+
+import com.alibaba.graphscope.common.ir.Utils;
+import com.alibaba.graphscope.common.ir.planner.rules.ExpandGetVFusionRule;
+import com.alibaba.graphscope.common.ir.tools.GraphBuilder;
+import com.alibaba.graphscope.common.ir.tools.GraphStdOperatorTable;
+import com.alibaba.graphscope.common.ir.tools.config.*;
+
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.rel.RelNode;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class ExpandGetVFusionTest {
+    // g.V().hasLabel("person").outE("knows").inV(), can be fused into GraphPhysicalExpand
+    @Test
+    public void expand_getv_fusion_0_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}], alias=[DEFAULT],"
+                        + " opt=[OUT], physicalOpt=[VERTEX])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // g.V().hasLabel("person").outE("knows").as("a").inV(), can not be fused
+    @Test
+    public void expand_getv_fusion_1_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows"),
+                                        "a"))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[a], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[a], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // g.V().hasLabel("person").outE("knows").inV().as("a"), can be fused into GraphPhysicalExpand
+    @Test
+    public void expand_getv_fusion_2_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "a"))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[a], opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}], alias=[a],"
+                        + " opt=[OUT], physicalOpt=[VERTEX])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // g.V().hasLabel("person").outE("knows").has("weight",0.5).inV(), can be fused into
+    // GraphPhysicalExpand
+    @Test
+    public void expand_getv_fusion_3_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable(null, "weight"),
+                                        builder.literal(0.5)))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}], alias=[DEFAULT],"
+                        + " opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], fusedFilter=[[=(DEFAULT.weight, 5E-1)]], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}], alias=[DEFAULT],"
+                    + " fusedFilter=[[=(DEFAULT.weight, 5E-1)]], opt=[OUT], physicalOpt=[VERTEX])\n"
+                    + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                    + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // g.V().hasLabel("person").outE("knows").inV().as("a").has("age",10), can be fused into
+    // GraphPhysicalExpand + GraphPhysicalGetV
+    @Test
+    public void expand_getv_fusion_4_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "a"))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable(null, "age"),
+                                        builder.literal(10)))
+                        .build();
+
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}], alias=[a],"
+                        + " fusedFilter=[[=(DEFAULT.age, 10)]], opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphPhysicalGetV(tableConfig=[{isAll=false, tables=[person]}], alias=[DEFAULT],"
+                        + " fusedFilter=[[=(DEFAULT.age, 10)]], opt=[END], physicalOpt=[ITSELF])\n"
+                        + "  GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[a], opt=[OUT], physicalOpt=[VERTEX])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // g.V().hasLabel("person").outE("knows").has("weight",0.5).inV().as("a").has("age",10), can be
+    // fused into
+    // GraphPhysicalExpand + GraphPhysicalGetV
+    @Test
+    public void expand_getv_fusion_5_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable(null, "weight"),
+                                        builder.literal(0.5)))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "a"))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable(null, "age"),
+                                        builder.literal(10)))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}], alias=[a],"
+                        + " fusedFilter=[[=(DEFAULT.age, 10)]], opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], fusedFilter=[[=(DEFAULT.weight, 5E-1)]], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphPhysicalGetV(tableConfig=[{isAll=false, tables=[person]}], alias=[DEFAULT],"
+                        + " fusedFilter=[[=(DEFAULT.age, 10)]], opt=[END], physicalOpt=[ITSELF])\n"
+                        + "  GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[a], fusedFilter=[[=(DEFAULT.weight, 5E-1)]], opt=[OUT],"
+                        + " physicalOpt=[VERTEX])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // g.V().hasLabel("person").outE("knows").outV(), can not be fused
+    @Test
+    public void expand_getv_fusion_6_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.START,
+                                        new LabelConfig(false).addLabel("person")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[START])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[START])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // path expand: g.V().hasLabel("person").out('1..3', "knows").with('PATH_OPT',
+    // SIMPLE).with('RESULT_OPT', ALL_V)
+    @Test
+    public void path_expand_getv_fusion_0_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        PathExpandConfig.Builder pxdBuilder = PathExpandConfig.newBuilder(builder);
+        PathExpandConfig pxdConfig =
+                pxdBuilder
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .range(1, 3)
+                        .pathOpt(GraphOpt.PathExpandPath.SIMPLE)
+                        .resultOpt(GraphOpt.PathExpandResult.ALL_V)
+                        .build();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .pathExpand(pxdConfig)
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalPathExpand(expand=[GraphLogicalExpand(tableConfig=[{isAll=false,"
+                        + " tables=[knows]}], alias=[DEFAULT], opt=[OUT])\n"
+                        + "], getV=[GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[END])\n"
+                        + "], offset=[1], fetch=[3], path_opt=[SIMPLE], result_opt=[ALL_V],"
+                        + " alias=[DEFAULT])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.PathBaseExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphLogicalPathExpand(fused=[GraphPhysicalExpand(tableConfig=[{isAll=false,"
+                        + " tables=[knows]}], alias=[DEFAULT], opt=[OUT], physicalOpt=[VERTEX])\n],"
+                        + " offset=[1], fetch=[3], path_opt=[SIMPLE], result_opt=[ALL_V],"
+                        + " alias=[DEFAULT])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // path expand: g.V().hasLabel("person").out('1..3', "knows").has("age",
+    // eq(10)).with('PATH_OPT', SIMPLE).with('RESULT_OPT', ALL_V)
+    @Test
+    public void path_expand_getv_fusion_1_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        PathExpandConfig.Builder pxdBuilder = PathExpandConfig.newBuilder(builder);
+        PathExpandConfig pxdConfig =
+                pxdBuilder
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .filter(
+                                pxdBuilder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        pxdBuilder.variable(null, "age"),
+                                        pxdBuilder.literal(10)))
+                        .range(1, 3)
+                        .pathOpt(GraphOpt.PathExpandPath.SIMPLE)
+                        .resultOpt(GraphOpt.PathExpandResult.ALL_V)
+                        .build();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .pathExpand(pxdConfig)
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalPathExpand(expand=[GraphLogicalExpand(tableConfig=[{isAll=false,"
+                        + " tables=[knows]}], alias=[DEFAULT], opt=[OUT])\n"
+                        + "], getV=[GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], fusedFilter=[[=(DEFAULT.age, 10)]], opt=[END])\n"
+                        + "], offset=[1], fetch=[3], path_opt=[SIMPLE], result_opt=[ALL_V],"
+                        + " alias=[DEFAULT])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.PathBaseExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphLogicalPathExpand(fused=[GraphPhysicalGetV(tableConfig=[{isAll=false,"
+                        + " tables=[person]}], alias=[DEFAULT], fusedFilter=[[=(DEFAULT.age, 10)]],"
+                        + " opt=[END], physicalOpt=[ITSELF])\n"
+                        + "  GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT], physicalOpt=[VERTEX])\n"
+                        + "], offset=[1], fetch=[3], path_opt=[SIMPLE], result_opt=[ALL_V],"
+                        + " alias=[DEFAULT])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+
+    // path expand with edge filters
+    @Test
+    public void path_expand_getv_fusion_2_test() {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        PathExpandConfig.Builder pxdBuilder = PathExpandConfig.newBuilder(builder);
+        PathExpandConfig pxdConfig =
+                pxdBuilder
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .filter(
+                                pxdBuilder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        pxdBuilder.variable(null, "weight"),
+                                        pxdBuilder.literal(0.5)))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .range(1, 3)
+                        .pathOpt(GraphOpt.PathExpandPath.SIMPLE)
+                        .resultOpt(GraphOpt.PathExpandResult.ALL_V)
+                        .build();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .pathExpand(pxdConfig)
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalPathExpand(expand=[GraphLogicalExpand(tableConfig=[{isAll=false,"
+                    + " tables=[knows]}], alias=[DEFAULT], fusedFilter=[[=(DEFAULT.weight, 5E-1)]],"
+                    + " opt=[OUT])\n"
+                    + "], getV=[GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                    + " alias=[DEFAULT], opt=[END])\n"
+                    + "], offset=[1], fetch=[3], path_opt=[SIMPLE], result_opt=[ALL_V],"
+                    + " alias=[DEFAULT])\n"
+                    + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                    + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.PathBaseExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphLogicalPathExpand(fused=[GraphPhysicalExpand(tableConfig=[{isAll=false,"
+                    + " tables=[knows]}], alias=[DEFAULT], fusedFilter=[[=(DEFAULT.weight, 5E-1)]],"
+                    + " opt=[OUT], physicalOpt=[VERTEX])\n"
+                    + "], offset=[1], fetch=[3], path_opt=[SIMPLE], result_opt=[ALL_V],"
+                    + " alias=[DEFAULT])\n"
+                    + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                    + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+    }
+}
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/GraphRelToProtoTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/GraphRelToProtoTest.java
new file mode 100644
index 000000000000..4a65025ce2f6
--- /dev/null
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/GraphRelToProtoTest.java
@@ -0,0 +1,765 @@
+/*
+ * Copyright 2024 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.ir.runtime;
+
+import com.alibaba.graphscope.common.config.Configs;
+import com.alibaba.graphscope.common.ir.Utils;
+import com.alibaba.graphscope.common.ir.planner.rules.DegreeFusionRule;
+import com.alibaba.graphscope.common.ir.planner.rules.ExpandGetVFusionRule;
+import com.alibaba.graphscope.common.ir.runtime.proto.GraphRelProtoPhysicalBuilder;
+import com.alibaba.graphscope.common.ir.tools.GraphBuilder;
+import com.alibaba.graphscope.common.ir.tools.GraphStdOperatorTable;
+import com.alibaba.graphscope.common.ir.tools.LogicalPlan;
+import com.alibaba.graphscope.common.ir.tools.config.ExpandConfig;
+import com.alibaba.graphscope.common.ir.tools.config.GetVConfig;
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
+import com.alibaba.graphscope.common.ir.tools.config.LabelConfig;
+import com.alibaba.graphscope.common.ir.tools.config.PathExpandConfig;
+import com.alibaba.graphscope.common.ir.tools.config.SourceConfig;
+import com.alibaba.graphscope.common.utils.FileUtils;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.tools.RelBuilder.AggCall;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+
+public class GraphRelToProtoTest {
+    @Test
+    public void scan_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode scan =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}], alias=[x],"
+                        + " opt=[VERTEX])",
+                scan.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(scan))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/scan_test.json"), plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void scan_edge_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode scan =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.EDGE,
+                                        new LabelConfig(false).addLabel("knows"),
+                                        "x"))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalSource(tableConfig=[{isAll=false, tables=[knows]}], alias=[x],"
+                        + " opt=[EDGE])",
+                scan.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(scan))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/scan_edge_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void scan_filter_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode scan =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable(null, "age"),
+                                        builder.literal(10)))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}], alias=[x],"
+                        + " fusedFilter=[[=(DEFAULT.age, 10)]], opt=[VERTEX])",
+                scan.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(scan))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/scan_filter_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void edge_expand_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode expand =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}], alias=[DEFAULT],"
+                        + " opt=[OUT])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                expand.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(expand))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/edge_expand_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void get_v_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode getV =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}], alias=[DEFAULT],"
+                        + " opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                getV.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(getV))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/getv_test.json"), plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void get_v_with_filter_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode getV =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable(null, "age"),
+                                        builder.literal(10)))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}], alias=[DEFAULT],"
+                        + " fusedFilter=[[=(DEFAULT.age, 10)]], opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                getV.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(getV))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/getv_with_filter_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void path_expand_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        PathExpandConfig.Builder pxdBuilder = PathExpandConfig.newBuilder(builder);
+        PathExpandConfig pxdConfig =
+                pxdBuilder
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .range(1, 3)
+                        .pathOpt(GraphOpt.PathExpandPath.SIMPLE)
+                        .resultOpt(GraphOpt.PathExpandResult.ALL_V)
+                        .build();
+        RelNode pxd =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .pathExpand(pxdConfig)
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalPathExpand(expand=[GraphLogicalExpand(tableConfig=[{isAll=false,"
+                        + " tables=[knows]}], alias=[DEFAULT], opt=[OUT])\n"
+                        + "], getV=[GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[END])\n"
+                        + "], offset=[1], fetch=[3], path_opt=[SIMPLE], result_opt=[ALL_V],"
+                        + " alias=[DEFAULT])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                pxd.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(pxd))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/path_expand_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void project_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode project =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .project(
+                                ImmutableList.of(builder.variable("x", "name")),
+                                ImmutableList.of("name"))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalProject(name=[x.name], isAppend=[false])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                project.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(project))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/project_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void filter_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode filter =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable("x", "name"),
+                                        builder.literal("marko")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}], alias=[x],"
+                        + " fusedFilter=[[=(DEFAULT.name, _UTF-8'marko')]], opt=[VERTEX])",
+                filter.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(filter))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/filter_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void aggregate_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode aggregate =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .aggregate(
+                                builder.groupKey(builder.variable("x", "name")),
+                                builder.count(builder.variable("x")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalAggregate(keys=[{variables=[x.name], aliases=[name]}],"
+                        + " values=[[{operands=[x], aggFunction=COUNT, alias='$f1',"
+                        + " distinct=false}]])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                aggregate.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(aggregate))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/aggregate_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void dedup_test_1() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        Iterable<AggCall> emptyIterable = Collections::emptyIterator;
+        RelNode dedup =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .aggregate(builder.groupKey(builder.variable("x", "name")), emptyIterable)
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalAggregate(keys=[{variables=[x.name], aliases=[name]}],"
+                        + " values=[[]])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                dedup.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(dedup))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/dedup_test_1.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void dedup_test_2() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode dedup =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .dedupBy(ImmutableList.of(builder.variable("x", "name")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalDedupBy(dedupByKeys=[[x.name]])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                dedup.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(dedup))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/dedup_test_2.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void join_test_1() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode source1 =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .build();
+        RelNode source2 =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .build();
+        RexNode condition = builder.getJoinCondition(source1, source2);
+        builder.push(source1);
+        builder.push(source2);
+        RelNode join = builder.join(JoinRelType.INNER, condition).build();
+        Assert.assertEquals(
+                "LogicalJoin(condition=[=(x, x)], joinType=[inner])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                join.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(join))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/join_test_1.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void join_test_2() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode expand1 =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "a"))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows"),
+                                        "b"))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x",
+                                        "b"))
+                        .build();
+        RelNode expand2 =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("software"),
+                                        "c"))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.IN,
+                                        new LabelConfig(false).addLabel("created"),
+                                        "d"))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.START,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x",
+                                        "d"))
+                        .build();
+        RexNode condition = builder.getJoinCondition(expand1, expand2);
+        builder.push(expand1);
+        builder.push(expand2);
+        RelNode join = builder.join(JoinRelType.INNER, condition).build();
+        Assert.assertEquals(
+                "LogicalJoin(condition=[=(x, x)], joinType=[inner])\n"
+                        + "  GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], startAlias=[b], opt=[END])\n"
+                        + "    GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[b], opt=[OUT])\n"
+                        + "      GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[a], opt=[VERTEX])\n"
+                        + "  GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], startAlias=[d], opt=[START])\n"
+                        + "    GraphLogicalExpand(tableConfig=[{isAll=false, tables=[created]}],"
+                        + " alias=[d], opt=[IN])\n"
+                        + "      GraphLogicalSource(tableConfig=[{isAll=false, tables=[software]}],"
+                        + " alias=[c], opt=[VERTEX])",
+                join.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(join))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/join_test_2.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void sort_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode sort =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .sort(builder.variable(null, "name"))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalSort(sort0=[DEFAULT.name], dir0=[ASC])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                sort.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(sort))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/sort_test.json"), plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void limit_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode limit =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .limit(0, 10)
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalSort(fetch=[10])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                limit.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(limit))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/limit_test.json"), plan.explain().trim());
+        }
+    }
+
+    // g.V().hasLabel("person").out("knows").count()
+    @Test
+    public void expand_degree_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .aggregate(builder.groupKey(), builder.countStar("cnt"))
+                        .build();
+        RelOptPlanner planner =
+                Utils.mockPlanner(DegreeFusionRule.ExpandGetVDegreeFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphLogicalAggregate(keys=[{variables=[], aliases=[]}],"
+                        + " values=[[{operands=[DEFAULT], aggFunction=$SUM0, alias='cnt',"
+                        + " distinct=false}]])\n"
+                        + "  GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT], physicalOpt=[DEGREE])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(after))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/edge_expand_degree_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    // g.V().hasLabel("person").outE("knows").inV()
+    @Test
+    public void expand_vertex_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[END])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[DEFAULT], opt=[OUT])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                before.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}], alias=[DEFAULT],"
+                        + " opt=[OUT], physicalOpt=[VERTEX])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(after))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/edge_expand_vertex_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    // g.V().hasLabel("person").outE("knows").has("weight", 0.5).inV().has("age", 10)
+    @Test
+    public void expand_vertex_with_filters_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode before =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person")))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable(null, "weight"),
+                                        builder.literal(0.5)))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "a"))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.EQUALS,
+                                        builder.variable(null, "age"),
+                                        builder.literal(10)))
+                        .build();
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.BasicExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(before);
+        RelNode after = planner.findBestExp();
+        Assert.assertEquals(
+                "GraphPhysicalGetV(tableConfig=[{isAll=false, tables=[person]}], alias=[DEFAULT],"
+                        + " fusedFilter=[[=(DEFAULT.age, 10)]], opt=[END], physicalOpt=[ITSELF])\n"
+                        + "  GraphPhysicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[a], fusedFilter=[[=(DEFAULT.weight, 5E-1)]], opt=[OUT],"
+                        + " physicalOpt=[VERTEX])\n"
+                        + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[VERTEX])",
+                after.explain().trim());
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(after))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource(
+                            "proto/edge_expand_vertex_with_filters_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void path_expand_fused_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        PathExpandConfig.Builder pxdBuilder = PathExpandConfig.newBuilder(builder);
+        PathExpandConfig pxdConfig =
+                pxdBuilder
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows")))
+                        .getV(
+                                new GetVConfig(
+                                        GraphOpt.GetV.END,
+                                        new LabelConfig(false).addLabel("person")))
+                        .range(1, 3)
+                        .pathOpt(GraphOpt.PathExpandPath.SIMPLE)
+                        .resultOpt(GraphOpt.PathExpandResult.ALL_V)
+                        .build();
+        RelNode pxd =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .pathExpand(pxdConfig)
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalPathExpand(expand=[GraphLogicalExpand(tableConfig=[{isAll=false,"
+                        + " tables=[knows]}], alias=[DEFAULT], opt=[OUT])\n"
+                        + "], getV=[GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[DEFAULT], opt=[END])\n"
+                        + "], offset=[1], fetch=[3], path_opt=[SIMPLE], result_opt=[ALL_V],"
+                        + " alias=[DEFAULT])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[x], opt=[VERTEX])",
+                pxd.explain().trim());
+        RelOptPlanner planner =
+                Utils.mockPlanner(ExpandGetVFusionRule.PathBaseExpandGetVFusionRule.Config.DEFAULT);
+        planner.setRoot(pxd);
+        RelNode after = planner.findBestExp();
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(after))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/path_fused_expand_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    private Configs getMockGraphConfig() {
+        return new Configs(ImmutableMap.of("servers", "1", "workers", "1"));
+    }
+}
diff --git a/interactive_engine/compiler/src/test/resources/proto/aggregate_test.json b/interactive_engine/compiler/src/test/resources/proto/aggregate_test.json
new file mode 100644
index 000000000000..86723fb046a3
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/aggregate_test.json
@@ -0,0 +1,115 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "groupBy": {
+        "mappings": [{
+          "key": {
+            "tag": {
+              "id": 0
+            },
+            "property": {
+              "key": {
+                "name": "name"
+              }
+            },
+            "nodeType": {
+              "dataType": "STRING"
+            }
+          },
+          "alias": 1
+        }],
+        "functions": [{
+          "vars": [{
+            "tag": {
+              "id": 0
+            },
+            "nodeType": {
+              "graphType": {
+                "graphDataType": [{
+                  "label": {
+                  },
+                  "props": [{
+                    "propId": {
+                      "name": "id"
+                    },
+                    "type": "INT64"
+                  }, {
+                    "propId": {
+                      "name": "name"
+                    },
+                    "type": "STRING"
+                  }, {
+                    "propId": {
+                      "name": "age"
+                    },
+                    "type": "INT32"
+                  }]
+                }]
+              }
+            }
+          }],
+          "aggregate": "COUNT",
+          "alias": 2
+        }]
+      }
+    },
+    "metaData": [{
+      "type": {
+        "dataType": "STRING"
+      },
+      "alias": 1
+    }, {
+      "type": {
+        "dataType": "INT64"
+      },
+      "alias": 2
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/dedup_test_1.json b/interactive_engine/compiler/src/test/resources/proto/dedup_test_1.json
new file mode 100644
index 000000000000..d19ad8cccadb
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/dedup_test_1.json
@@ -0,0 +1,91 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "project": {
+        "mappings": [{
+          "expr": {
+            "operators": [{
+              "var": {
+                "tag": {
+                  "id": 0
+                },
+                "property": {
+                  "key": {
+                    "name": "name"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "STRING"
+                }
+              },
+              "nodeType": {
+                "dataType": "STRING"
+              }
+            }]
+          },
+          "alias": 1
+        }]
+      }
+    }
+  }, {
+    "opr": {
+      "dedup": {
+        "keys": [{
+          "tag": {
+            "id": 1
+          },
+          "nodeType": {
+            "dataType": "STRING"
+          }
+        }]
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/dedup_test_2.json b/interactive_engine/compiler/src/test/resources/proto/dedup_test_2.json
new file mode 100644
index 000000000000..4b9c5288cb05
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/dedup_test_2.json
@@ -0,0 +1,68 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "dedup": {
+        "keys": [{
+          "tag": {
+            "id": 0
+          },
+          "property": {
+            "key": {
+              "name": "name"
+            }
+          },
+          "nodeType": {
+            "dataType": "STRING"
+          }
+        }]
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/edge_expand_degree_test.json b/interactive_engine/compiler/src/test/resources/proto/edge_expand_degree_test.json
new file mode 100644
index 000000000000..13624a1f0839
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/edge_expand_degree_test.json
@@ -0,0 +1,86 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        },
+        "expandOpt": "DEGREE"
+      }
+    },
+    "metaData": [{
+      "type": {
+        "dataType": "INT64"
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "groupBy": {
+        "functions": [{
+          "vars": [{
+            "nodeType": {
+              "dataType": "INT64"
+            }
+          }],
+          "alias": 0
+        }]
+      }
+    },
+    "metaData": [{
+      "type": {
+        "dataType": "INT64"
+      }
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/edge_expand_test.json b/interactive_engine/compiler/src/test/resources/proto/edge_expand_test.json
new file mode 100644
index 000000000000..f4a0957c24d0
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/edge_expand_test.json
@@ -0,0 +1,82 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        },
+        "expandOpt": "EDGE"
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "elementOpt": "EDGE",
+          "graphDataType": [{
+            "label": {
+              "srcLabel": 0,
+              "dstLabel": 0
+            },
+            "props": [{
+              "propId": {
+                "name": "weight"
+              },
+              "type": "DOUBLE"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_test.json b/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_test.json
new file mode 100644
index 000000000000..9354d27c58b4
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_test.json
@@ -0,0 +1,88 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_with_filters_test.json b/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_with_filters_test.json
new file mode 100644
index 000000000000..7b7dcc13999e
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_with_filters_test.json
@@ -0,0 +1,184 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "predicate": {
+            "operators": [{
+              "var": {
+                "property": {
+                  "key": {
+                    "name": "weight"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "DOUBLE"
+                }
+              },
+              "nodeType": {
+                "dataType": "DOUBLE"
+              }
+            }, {
+              "logical": "EQ",
+              "nodeType": {
+                "dataType": "BOOLEAN"
+              }
+            }, {
+              "const": {
+                "f64": 0.5
+              },
+              "nodeType": {
+                "dataType": "DOUBLE"
+              }
+            }]
+          },
+          "sampleRatio": 1.0
+        },
+        "alias": 0
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "vertex": {
+        "opt": "ITSELF",
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "predicate": {
+            "operators": [{
+              "var": {
+                "property": {
+                  "key": {
+                    "name": "age"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "INT32"
+                }
+              },
+              "nodeType": {
+                "dataType": "INT32"
+              }
+            }, {
+              "logical": "EQ",
+              "nodeType": {
+                "dataType": "BOOLEAN"
+              }
+            }, {
+              "const": {
+                "i32": 10
+              },
+              "nodeType": {
+                "dataType": "INT32"
+              }
+            }]
+          },
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/filter_test.json b/interactive_engine/compiler/src/test/resources/proto/filter_test.json
new file mode 100644
index 000000000000..959f76895e29
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/filter_test.json
@@ -0,0 +1,79 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "predicate": {
+            "operators": [{
+              "var": {
+                "property": {
+                  "key": {
+                    "name": "name"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "STRING"
+                }
+              },
+              "nodeType": {
+                "dataType": "STRING"
+              }
+            }, {
+              "logical": "EQ",
+              "nodeType": {
+                "dataType": "BOOLEAN"
+              }
+            }, {
+              "const": {
+                "str": "marko"
+              },
+              "nodeType": {
+                "dataType": "STRING"
+              }
+            }]
+          },
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/getv_test.json b/interactive_engine/compiler/src/test/resources/proto/getv_test.json
new file mode 100644
index 000000000000..69b6137d724c
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/getv_test.json
@@ -0,0 +1,121 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        },
+        "expandOpt": "EDGE"
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "elementOpt": "EDGE",
+          "graphDataType": [{
+            "label": {
+              "srcLabel": 0,
+              "dstLabel": 0
+            },
+            "props": [{
+              "propId": {
+                "name": "weight"
+              },
+              "type": "DOUBLE"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "vertex": {
+        "opt": "END",
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/getv_with_filter_test.json b/interactive_engine/compiler/src/test/resources/proto/getv_with_filter_test.json
new file mode 100644
index 000000000000..52f3d0b95ce9
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/getv_with_filter_test.json
@@ -0,0 +1,186 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        },
+        "expandOpt": "EDGE"
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "elementOpt": "EDGE",
+          "graphDataType": [{
+            "label": {
+              "srcLabel": 0,
+              "dstLabel": 0
+            },
+            "props": [{
+              "propId": {
+                "name": "weight"
+              },
+              "type": "DOUBLE"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "vertex": {
+        "opt": "END",
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "vertex": {
+        "opt": "ITSELF",
+        "params": {
+          "predicate": {
+            "operators": [{
+              "var": {
+                "property": {
+                  "key": {
+                    "name": "age"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "INT32"
+                }
+              },
+              "nodeType": {
+                "dataType": "INT32"
+              }
+            }, {
+              "logical": "EQ",
+              "nodeType": {
+                "dataType": "BOOLEAN"
+              }
+            }, {
+              "const": {
+                "i32": 10
+              },
+              "nodeType": {
+                "dataType": "INT32"
+              }
+            }]
+          },
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/join_test_1.json b/interactive_engine/compiler/src/test/resources/proto/join_test_1.json
new file mode 100644
index 000000000000..157d55ed5a24
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/join_test_1.json
@@ -0,0 +1,157 @@
+{
+  "plan": [{
+    "opr": {
+      "join": {
+        "leftKeys": [{
+          "tag": {
+            "id": 0
+          },
+          "nodeType": {
+            "graphType": {
+              "graphDataType": [{
+                "label": {
+                },
+                "props": [{
+                  "propId": {
+                    "name": "id"
+                  },
+                  "type": "INT64"
+                }, {
+                  "propId": {
+                    "name": "name"
+                  },
+                  "type": "STRING"
+                }, {
+                  "propId": {
+                    "name": "age"
+                  },
+                  "type": "INT32"
+                }]
+              }]
+            }
+          }
+        }],
+        "rightKeys": [{
+          "tag": {
+            "id": 0
+          },
+          "nodeType": {
+            "graphType": {
+              "graphDataType": [{
+                "label": {
+                },
+                "props": [{
+                  "propId": {
+                    "name": "id"
+                  },
+                  "type": "INT64"
+                }, {
+                  "propId": {
+                    "name": "name"
+                  },
+                  "type": "STRING"
+                }, {
+                  "propId": {
+                    "name": "age"
+                  },
+                  "type": "INT32"
+                }]
+              }]
+            }
+          }
+        }],
+        "leftPlan": {
+          "plan": [{
+            "opr": {
+              "scan": {
+                "alias": 0,
+                "params": {
+                  "tables": [{
+                    "id": 0
+                  }],
+                  "sampleRatio": 1.0
+                }
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "graphDataType": [{
+                    "label": {
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "id"
+                      },
+                      "type": "INT64"
+                    }, {
+                      "propId": {
+                        "name": "name"
+                      },
+                      "type": "STRING"
+                    }, {
+                      "propId": {
+                        "name": "age"
+                      },
+                      "type": "INT32"
+                    }]
+                  }]
+                }
+              }
+            }]
+          }]
+        },
+        "rightPlan": {
+          "plan": [{
+            "opr": {
+              "scan": {
+                "alias": 0,
+                "params": {
+                  "tables": [{
+                    "id": 0
+                  }],
+                  "sampleRatio": 1.0
+                }
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "graphDataType": [{
+                    "label": {
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "id"
+                      },
+                      "type": "INT64"
+                    }, {
+                      "propId": {
+                        "name": "name"
+                      },
+                      "type": "STRING"
+                    }, {
+                      "propId": {
+                        "name": "age"
+                      },
+                      "type": "INT32"
+                    }]
+                  }]
+                }
+              }
+            }]
+          }]
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/join_test_2.json b/interactive_engine/compiler/src/test/resources/proto/join_test_2.json
new file mode 100644
index 000000000000..373e73a9a6c5
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/join_test_2.json
@@ -0,0 +1,313 @@
+{
+  "plan": [{
+    "opr": {
+      "join": {
+        "leftKeys": [{
+          "tag": {
+            "id": 2
+          },
+          "nodeType": {
+            "graphType": {
+              "graphDataType": [{
+                "label": {
+                },
+                "props": [{
+                  "propId": {
+                    "name": "id"
+                  },
+                  "type": "INT64"
+                }, {
+                  "propId": {
+                    "name": "name"
+                  },
+                  "type": "STRING"
+                }, {
+                  "propId": {
+                    "name": "age"
+                  },
+                  "type": "INT32"
+                }]
+              }]
+            }
+          }
+        }],
+        "rightKeys": [{
+          "tag": {
+            "id": 2
+          },
+          "nodeType": {
+            "graphType": {
+              "graphDataType": [{
+                "label": {
+                },
+                "props": [{
+                  "propId": {
+                    "name": "id"
+                  },
+                  "type": "INT64"
+                }, {
+                  "propId": {
+                    "name": "name"
+                  },
+                  "type": "STRING"
+                }, {
+                  "propId": {
+                    "name": "age"
+                  },
+                  "type": "INT32"
+                }]
+              }]
+            }
+          }
+        }],
+        "leftPlan": {
+          "plan": [{
+            "opr": {
+              "scan": {
+                "alias": 0,
+                "params": {
+                  "tables": [{
+                    "id": 0
+                  }],
+                  "sampleRatio": 1.0
+                }
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "graphDataType": [{
+                    "label": {
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "id"
+                      },
+                      "type": "INT64"
+                    }, {
+                      "propId": {
+                        "name": "name"
+                      },
+                      "type": "STRING"
+                    }, {
+                      "propId": {
+                        "name": "age"
+                      },
+                      "type": "INT32"
+                    }]
+                  }]
+                }
+              }
+            }]
+          }, {
+            "opr": {
+              "edge": {
+                "params": {
+                  "tables": [{
+                    "id": 0
+                  }],
+                  "sampleRatio": 1.0
+                },
+                "alias": 1,
+                "expandOpt": "EDGE"
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "elementOpt": "EDGE",
+                  "graphDataType": [{
+                    "label": {
+                      "srcLabel": 0,
+                      "dstLabel": 0
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "weight"
+                      },
+                      "type": "DOUBLE"
+                    }]
+                  }]
+                }
+              },
+              "alias": 1
+            }]
+          }, {
+            "opr": {
+              "vertex": {
+                "tag": 1,
+                "opt": "END",
+                "params": {
+                  "tables": [{
+                    "id": 0
+                  }],
+                  "sampleRatio": 1.0
+                },
+                "alias": 2
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "graphDataType": [{
+                    "label": {
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "id"
+                      },
+                      "type": "INT64"
+                    }, {
+                      "propId": {
+                        "name": "name"
+                      },
+                      "type": "STRING"
+                    }, {
+                      "propId": {
+                        "name": "age"
+                      },
+                      "type": "INT32"
+                    }]
+                  }]
+                }
+              },
+              "alias": 2
+            }]
+          }]
+        },
+        "rightPlan": {
+          "plan": [{
+            "opr": {
+              "scan": {
+                "alias": 3,
+                "params": {
+                  "tables": [{
+                    "id": 1
+                  }],
+                  "sampleRatio": 1.0
+                }
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "graphDataType": [{
+                    "label": {
+                      "label": 1
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "id"
+                      },
+                      "type": "INT64"
+                    }, {
+                      "propId": {
+                        "name": "name"
+                      },
+                      "type": "STRING"
+                    }, {
+                      "propId": {
+                        "name": "lang"
+                      },
+                      "type": "STRING"
+                    }, {
+                      "propId": {
+                        "name": "creationDate"
+                      },
+                      "type": "DATE32"
+                    }]
+                  }]
+                }
+              },
+              "alias": 3
+            }]
+          }, {
+            "opr": {
+              "edge": {
+                "direction": "IN",
+                "params": {
+                  "tables": [{
+                    "id": 1
+                  }],
+                  "sampleRatio": 1.0
+                },
+                "alias": 4,
+                "expandOpt": "EDGE"
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "elementOpt": "EDGE",
+                  "graphDataType": [{
+                    "label": {
+                      "label": 1,
+                      "srcLabel": 0,
+                      "dstLabel": 1
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "weight"
+                      },
+                      "type": "DOUBLE"
+                    }]
+                  }]
+                }
+              },
+              "alias": 4
+            }]
+          }, {
+            "opr": {
+              "vertex": {
+                "tag": 4,
+                "params": {
+                  "tables": [{
+                    "id": 0
+                  }],
+                  "sampleRatio": 1.0
+                },
+                "alias": 2
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "graphDataType": [{
+                    "label": {
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "id"
+                      },
+                      "type": "INT64"
+                    }, {
+                      "propId": {
+                        "name": "name"
+                      },
+                      "type": "STRING"
+                    }, {
+                      "propId": {
+                        "name": "age"
+                      },
+                      "type": "INT32"
+                    }]
+                  }]
+                }
+              },
+              "alias": 2
+            }]
+          }]
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/limit_test.json b/interactive_engine/compiler/src/test/resources/proto/limit_test.json
new file mode 100644
index 000000000000..cbc7135f3b8b
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/limit_test.json
@@ -0,0 +1,58 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "limit": {
+        "range": {
+          "upper": 10
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/path_expand_test.json b/interactive_engine/compiler/src/test/resources/proto/path_expand_test.json
new file mode 100644
index 000000000000..cc18d0cac59b
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/path_expand_test.json
@@ -0,0 +1,81 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "path": {
+        "base": {
+          "edgeExpand": {
+            "params": {
+              "tables": [{
+                "id": 0
+              }],
+              "sampleRatio": 1.0
+            },
+            "expandOpt": "EDGE"
+          },
+          "getV": {
+            "opt": "END",
+            "params": {
+              "tables": [{
+                "id": 0
+              }],
+              "sampleRatio": 1.0
+            }
+          }
+        },
+        "hopRange": {
+          "lower": 1,
+          "upper": 3
+        },
+        "pathOpt": "SIMPLE",
+        "resultOpt": "ALL_V"
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/path_fused_expand_test.json b/interactive_engine/compiler/src/test/resources/proto/path_fused_expand_test.json
new file mode 100644
index 000000000000..9eb9011fecfc
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/path_fused_expand_test.json
@@ -0,0 +1,71 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "path": {
+        "base": {
+          "edgeExpand": {
+            "params": {
+              "tables": [{
+                "id": 0
+              }],
+              "sampleRatio": 1.0
+            }
+          }
+        },
+        "hopRange": {
+          "lower": 1,
+          "upper": 3
+        },
+        "pathOpt": "SIMPLE",
+        "resultOpt": "ALL_V"
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/project_test.json b/interactive_engine/compiler/src/test/resources/proto/project_test.json
new file mode 100644
index 000000000000..8449baf778ba
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/project_test.json
@@ -0,0 +1,84 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "project": {
+        "mappings": [{
+          "expr": {
+            "operators": [{
+              "var": {
+                "tag": {
+                  "id": 0
+                },
+                "property": {
+                  "key": {
+                    "name": "name"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "STRING"
+                }
+              },
+              "nodeType": {
+                "dataType": "STRING"
+              }
+            }]
+          },
+          "alias": 1
+        }]
+      }
+    },
+    "metaData": [{
+      "type": {
+        "dataType": "STRING"
+      },
+      "alias": 1
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/scan_edge_test.json b/interactive_engine/compiler/src/test/resources/proto/scan_edge_test.json
new file mode 100644
index 000000000000..cd01cb223be5
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/scan_edge_test.json
@@ -0,0 +1,44 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "scanOpt": "EDGE",
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "elementOpt": "EDGE",
+          "graphDataType": [{
+            "label": {
+              "srcLabel": 0,
+              "dstLabel": 0
+            },
+            "props": [{
+              "propId": {
+                "name": "weight"
+              },
+              "type": "DOUBLE"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/scan_filter_test.json b/interactive_engine/compiler/src/test/resources/proto/scan_filter_test.json
new file mode 100644
index 000000000000..9bb3096c2fac
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/scan_filter_test.json
@@ -0,0 +1,79 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "predicate": {
+            "operators": [{
+              "var": {
+                "property": {
+                  "key": {
+                    "name": "age"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "INT32"
+                }
+              },
+              "nodeType": {
+                "dataType": "INT32"
+              }
+            }, {
+              "logical": "EQ",
+              "nodeType": {
+                "dataType": "BOOLEAN"
+              }
+            }, {
+              "const": {
+                "i32": 10
+              },
+              "nodeType": {
+                "dataType": "INT32"
+              }
+            }]
+          },
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/scan_test.json b/interactive_engine/compiler/src/test/resources/proto/scan_test.json
new file mode 100644
index 000000000000..11fe411986ac
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/scan_test.json
@@ -0,0 +1,50 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "alias": 0,
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/sort_test.json b/interactive_engine/compiler/src/test/resources/proto/sort_test.json
new file mode 100644
index 000000000000..d8741401371f
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/sort_test.json
@@ -0,0 +1,68 @@
+{
+  "plan": [{
+    "opr": {
+      "scan": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      },
+      "alias": -1
+    }]
+  }, {
+    "opr": {
+      "orderBy": {
+        "pairs": [{
+          "key": {
+            "property": {
+              "key": {
+                "name": "name"
+              }
+            },
+            "nodeType": {
+              "dataType": "STRING"
+            }
+          },
+          "order": "ASC"
+        }]
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file