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/interactive_engine/compiler/ir_experimental_ci.sh b/interactive_engine/compiler/ir_experimental_ci.sh
index 5602c2ab6694..600079df6589 100755
--- a/interactive_engine/compiler/ir_experimental_ci.sh
+++ b/interactive_engine/compiler/ir_experimental_ci.sh
@@ -45,6 +45,27 @@ fi
 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
 
+# start distributed engine service and load modern graph
+cd ${base_dir}/../executor/ir/target/release &&
+RUST_LOG=info DATA_PATH=/tmp/gstest/modern_graph_exp_bin PARTITION_ID=0 ./start_rpc_server --config ${base_dir}/../executor/ir/integrated/config/distributed/server_0 &
+cd ${base_dir}/../executor/ir/target/release &&
+RUST_LOG=info DATA_PATH=/tmp/gstest/modern_graph_exp_bin PARTITION_ID=1 ./start_rpc_server --config ${base_dir}/../executor/ir/integrated/config/distributed/server_1 &
+# start compiler service
+cd ${base_dir} && make run gremlin.script.language.name=antlr_gremlin_calcite physical.opt.config=proto pegasus.hosts:=127.0.0.1:1234,127.0.0.1:1235 &
+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 distributed 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
+
 cd ${base_dir}/../executor/ir/target/release && DATA_PATH=/tmp/gstest/movie_graph_exp_bin RUST_LOG=info ./start_rpc_server --config ${base_dir}/../executor/ir/integrated/config &
 sleep 5s
 # start compiler service
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
index d1eb95cbf205..53f045c1c415 100644
--- 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
@@ -17,6 +17,7 @@
 package com.alibaba.graphscope.common.ir.runtime.proto;
 
 import com.alibaba.graphscope.common.config.Configs;
+import com.alibaba.graphscope.common.config.PegasusConfig;
 import com.alibaba.graphscope.common.ir.rel.GraphLogicalAggregate;
 import com.alibaba.graphscope.common.ir.rel.GraphLogicalDedupBy;
 import com.alibaba.graphscope.common.ir.rel.GraphLogicalProject;
@@ -34,6 +35,7 @@
 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.GraphNameOrId;
 import com.alibaba.graphscope.common.ir.type.GraphSchemaType;
 import com.alibaba.graphscope.gaia.proto.GraphAlgebra;
 import com.alibaba.graphscope.gaia.proto.GraphAlgebraPhysical;
@@ -55,6 +57,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
@@ -66,6 +69,8 @@ public class GraphRelToProtoConverter extends GraphShuttle {
     private final RexBuilder rexBuilder;
     private final Configs graphConfig;
     private GraphAlgebraPhysical.PhysicalPlan.Builder physicalBuilder;
+    private final boolean isPartitioned;
+    private boolean preCacheEdgeProps;
 
     public GraphRelToProtoConverter(
             boolean isColumnId,
@@ -75,6 +80,12 @@ public GraphRelToProtoConverter(
         this.rexBuilder = GraphPlanner.rexBuilderFactory.apply(configs);
         this.graphConfig = configs;
         this.physicalBuilder = physicalBuilder;
+        this.isPartitioned =
+                !(PegasusConfig.PEGASUS_HOSTS.get(configs).split(",").length == 1
+                        && PegasusConfig.PEGASUS_WORKER_NUM.get(configs) == 1);
+        // currently, since the store doesn't support get properties from edges, we always need to
+        // precache edge properties.
+        this.preCacheEdgeProps = true;
     }
 
     @Override
@@ -87,6 +98,12 @@ public RelNode visit(GraphLogicalSource source) {
             GraphAlgebra.IndexPredicate indexPredicate = buildIndexPredicates(uniqueKeyFilters);
             scanBuilder.setIdxPredicate(indexPredicate);
         }
+        GraphAlgebra.QueryParams.Builder queryParamsBuilder = buildQueryParams(source);
+        if (preCacheEdgeProps && GraphOpt.Source.EDGE.equals(source.getOpt())) {
+            addQueryColumns(
+                    queryParamsBuilder,
+                    Utils.extractColumnsFromRelDataType(source.getRowType(), isColumnId));
+        }
         scanBuilder.setParams(buildQueryParams(source));
         if (source.getAliasId() != AliasInference.DEFAULT_ID) {
             scanBuilder.setAlias(Utils.asAliasId(source.getAliasId()));
@@ -108,6 +125,9 @@ public RelNode visit(GraphLogicalExpand expand) {
         oprBuilder.setOpr(
                 GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setEdge(edgeExpand));
         oprBuilder.addAllMetaData(Utils.physicalProtoRowType(expand.getRowType(), isColumnId));
+        if (isPartitioned) {
+            addRepartitionToAnother(expand.getStartAlias().getAliasId());
+        }
         physicalBuilder.addPlan(oprBuilder.build());
         return expand;
     }
@@ -165,6 +185,9 @@ public RelNode visit(GraphLogicalGetV getV) {
                             .setVertex(auxiliaBuilder));
             auxiliaOprBuilder.addAllMetaData(
                     Utils.physicalProtoRowType(getV.getRowType(), isColumnId));
+            if (isPartitioned) {
+                addRepartitionToAnother(getV.getAliasId());
+            }
             physicalBuilder.addPlan(auxiliaOprBuilder.build());
             return getV;
         }
@@ -226,7 +249,9 @@ public RelNode visit(GraphLogicalPathExpand pxd) {
         }
         oprBuilder.setOpr(
                 GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setPath(pathExpandBuilder));
-
+        if (isPartitioned) {
+            addRepartitionToAnother(pxd.getStartAlias().getAliasId());
+        }
         physicalBuilder.addPlan(oprBuilder.build());
         return pxd;
     }
@@ -241,6 +266,9 @@ public RelNode visit(GraphPhysicalExpand physicalExpand) {
                 GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setEdge(edgeExpand));
         oprBuilder.addAllMetaData(
                 Utils.physicalProtoRowType(physicalExpand.getRowType(), isColumnId));
+        if (isPartitioned) {
+            addRepartitionToAnother(physicalExpand.getStartAlias().getAliasId());
+        }
         physicalBuilder.addPlan(oprBuilder.build());
         return physicalExpand;
     }
@@ -255,6 +283,9 @@ public RelNode visit(GraphPhysicalGetV physicalGetV) {
                 GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setVertex(auxilia));
         oprBuilder.addAllMetaData(
                 Utils.physicalProtoRowType(physicalGetV.getRowType(), isColumnId));
+        if (isPartitioned) {
+            addRepartitionToAnother(physicalGetV.getStartAlias().getAliasId());
+        }
         physicalBuilder.addPlan(oprBuilder.build());
         return physicalGetV;
     }
@@ -271,6 +302,18 @@ public RelNode visit(LogicalFilter filter) {
         selectBuilder.setPredicate(exprProto);
         oprBuilder.setOpr(
                 GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setSelect(selectBuilder));
+        oprBuilder.addAllMetaData(Utils.physicalProtoRowType(filter.getRowType(), isColumnId));
+        if (isPartitioned) {
+            Map<Integer, Set<GraphNameOrId>> tagColumns =
+                    Utils.extractTagColumnsFromRexNodes(List.of(filter.getCondition()));
+            if (preCacheEdgeProps) {
+                Utils.removeEdgeProperties(
+                        com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(
+                                filter.getInput()),
+                        tagColumns);
+            }
+            lazyPropertyFetching(tagColumns, true);
+        }
         physicalBuilder.addPlan(oprBuilder.build());
         return filter;
     }
@@ -284,6 +327,7 @@ public RelNode visit(GraphLogicalProject project) {
                 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()
@@ -301,6 +345,17 @@ public RelNode visit(GraphLogicalProject project) {
         oprBuilder.setOpr(
                 GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setProject(projectBuilder));
         oprBuilder.addAllMetaData(Utils.physicalProtoRowType(project.getRowType(), isColumnId));
+        if (isPartitioned) {
+            Map<Integer, Set<GraphNameOrId>> tagColumns =
+                    Utils.extractTagColumnsFromRexNodes(project.getProjects());
+            if (preCacheEdgeProps) {
+                Utils.removeEdgeProperties(
+                        com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(
+                                project.getInput()),
+                        tagColumns);
+            }
+            lazyPropertyFetching(tagColumns, true);
+        }
         physicalBuilder.addPlan(oprBuilder.build());
         return project;
     }
@@ -364,6 +419,17 @@ public RelNode visit(GraphLogicalAggregate aggregate) {
                     GraphAlgebraPhysical.PhysicalOpr.newBuilder();
             dedupOprBuilder.setOpr(
                     GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setDedup(dedupBuilder));
+            if (isPartitioned) {
+                Map<Integer, Set<GraphNameOrId>> tagColumns =
+                        Utils.extractTagColumnsFromRexNodes(keys.getVariables());
+                if (preCacheEdgeProps) {
+                    Utils.removeEdgeProperties(
+                            com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(
+                                    aggregate.getInput()),
+                            tagColumns);
+                }
+                lazyPropertyFetching(tagColumns);
+            }
             physicalBuilder.addPlan(projectOprBuilder.build());
             physicalBuilder.addPlan(dedupOprBuilder.build());
         } else {
@@ -396,27 +462,28 @@ public RelNode visit(GraphLogicalAggregate aggregate) {
                 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();
+                for (RexNode operand : operands) {
+                    Preconditions.checkArgument(
+                            operand instanceof RexGraphVariable,
+                            "each expression in aggregate call should be type %s, but is %s",
+                            RexGraphVariable.class,
+                            operand.getClass());
+                    OuterExpression.Variable var =
+                            operand.accept(
+                                            new RexToProtoConverter(
+                                                    true, isColumnId, this.rexBuilder))
+                                    .getOperators(0)
+                                    .getVar();
+                    aggFnAliasBuilder.addVars(var);
+                }
+                GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate aggOpt =
+                        Utils.protoAggOpt(groupCalls.get(i));
                 aggFnAliasBuilder.setAggregate(aggOpt);
-                aggFnAliasBuilder.addVars(var);
+                int aliasId = fields.get(i + keys.groupKeyCount()).getIndex();
                 if (aliasId != AliasInference.DEFAULT_ID) {
                     aggFnAliasBuilder.setAlias(Utils.asAliasId(aliasId));
                 }
@@ -427,6 +494,23 @@ public RelNode visit(GraphLogicalAggregate aggregate) {
                             .setGroupBy(groupByBuilder));
             oprBuilder.addAllMetaData(
                     Utils.physicalProtoRowType(aggregate.getRowType(), isColumnId));
+            if (isPartitioned) {
+                List<RexNode> keysAndAggs = Lists.newArrayList();
+                keysAndAggs.addAll(keys.getVariables());
+                keysAndAggs.addAll(
+                        groupCalls.stream()
+                                .flatMap(k -> k.getOperands().stream())
+                                .collect(Collectors.toList()));
+                Map<Integer, Set<GraphNameOrId>> tagColumns =
+                        Utils.extractTagColumnsFromRexNodes(keysAndAggs);
+                if (preCacheEdgeProps) {
+                    Utils.removeEdgeProperties(
+                            com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(
+                                    aggregate.getInput()),
+                            tagColumns);
+                }
+                lazyPropertyFetching(tagColumns);
+            }
             physicalBuilder.addPlan(oprBuilder.build());
         }
         return aggregate;
@@ -452,6 +536,18 @@ public RelNode visit(GraphLogicalDedupBy dedupBy) {
         }
         oprBuilder.setOpr(
                 GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setDedup(dedupBuilder));
+        oprBuilder.addAllMetaData(Utils.physicalProtoRowType(dedupBy.getRowType(), isColumnId));
+        if (isPartitioned) {
+            Map<Integer, Set<GraphNameOrId>> tagColumns =
+                    Utils.extractTagColumnsFromRexNodes(dedupBy.getDedupByKeys());
+            if (preCacheEdgeProps) {
+                Utils.removeEdgeProperties(
+                        com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(
+                                dedupBy.getInput()),
+                        tagColumns);
+            }
+            lazyPropertyFetching(tagColumns);
+        }
         physicalBuilder.addPlan(oprBuilder.build());
         return dedupBy;
     }
@@ -483,6 +579,20 @@ public RelNode visit(GraphLogicalSort sort) {
             oprBuilder.setOpr(
                     GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
                             .setOrderBy(orderByBuilder));
+            if (isPartitioned) {
+                Map<Integer, Set<GraphNameOrId>> tagColumns =
+                        Utils.extractTagColumnsFromRexNodes(
+                                collations.stream()
+                                        .map(k -> ((GraphFieldCollation) k).getVariable())
+                                        .collect(Collectors.toList()));
+                if (preCacheEdgeProps) {
+                    Utils.removeEdgeProperties(
+                            com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(
+                                    sort.getInput()),
+                            tagColumns);
+                }
+                lazyPropertyFetching(tagColumns);
+            }
         } else {
             GraphAlgebra.Limit.Builder limitBuilder = GraphAlgebra.Limit.newBuilder();
             limitBuilder.setRange(buildRange(sort.offset, sort.fetch));
@@ -500,13 +610,18 @@ public RelNode visit(LogicalJoin join) {
         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);
+        Preconditions.checkArgument(
+                !conditions.isEmpty(), "join condition in physical should not be empty");
+        List<RexNode> leftKeys = Lists.newArrayList();
+        List<RexNode> rightKeys = Lists.newArrayList();
         for (RexNode condition : conditions) {
             List<RexGraphVariable> leftRightVars = getLeftRightVariables(condition);
-            Preconditions.checkArgument(leftRightVars.size() == 2, errorMessage);
+            Preconditions.checkArgument(
+                    leftRightVars.size() == 2,
+                    "join condition in physical should have two operands, while it is %s",
+                    leftRightVars.size());
+            leftKeys.add(leftRightVars.get(0));
+            rightKeys.add(leftRightVars.get(1));
             OuterExpression.Variable leftVar =
                     leftRightVars
                             .get(0)
@@ -532,6 +647,23 @@ public RelNode visit(LogicalJoin join) {
         left.accept(new GraphRelToProtoConverter(isColumnId, graphConfig, leftPlanBuilder));
         RelNode right = join.getRight();
         right.accept(new GraphRelToProtoConverter(isColumnId, graphConfig, rightPlanBuilder));
+        if (isPartitioned) {
+
+            Map<Integer, Set<GraphNameOrId>> leftTagColumns =
+                    Utils.extractTagColumnsFromRexNodes(leftKeys);
+            Map<Integer, Set<GraphNameOrId>> rightTagColumns =
+                    Utils.extractTagColumnsFromRexNodes(rightKeys);
+            if (preCacheEdgeProps) {
+                Utils.removeEdgeProperties(
+                        com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(join.getLeft()),
+                        leftTagColumns);
+                Utils.removeEdgeProperties(
+                        com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(join.getRight()),
+                        rightTagColumns);
+            }
+            lazyPropertyFetching(leftPlanBuilder, leftTagColumns, false);
+            lazyPropertyFetching(rightPlanBuilder, rightTagColumns, false);
+        }
         joinBuilder.setLeftPlan(leftPlanBuilder);
         joinBuilder.setRightPlan(rightPlanBuilder);
         oprBuilder.setOpr(
@@ -561,7 +693,13 @@ private GraphAlgebraPhysical.EdgeExpand.Builder buildEdgeExpand(
         GraphAlgebraPhysical.EdgeExpand.Builder expandBuilder =
                 GraphAlgebraPhysical.EdgeExpand.newBuilder();
         expandBuilder.setDirection(Utils.protoExpandDirOpt(expand.getOpt()));
-        expandBuilder.setParams(buildQueryParams(expand));
+        GraphAlgebra.QueryParams.Builder queryParamsBuilder = buildQueryParams(expand);
+        if (preCacheEdgeProps && GraphOpt.PhysicalExpandOpt.EDGE.equals(opt)) {
+            addQueryColumns(
+                    queryParamsBuilder,
+                    Utils.extractColumnsFromRelDataType(expand.getRowType(), isColumnId));
+        }
+        expandBuilder.setParams(queryParamsBuilder);
         if (aliasId != AliasInference.DEFAULT_ID) {
             expandBuilder.setAlias(Utils.asAliasId(aliasId));
         }
@@ -613,12 +751,12 @@ private GraphAlgebra.Range buildRange(RexNode offset, RexNode fetch) {
                     "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());
+        int lower = (offset == null) ? 0 : ((Number) ((RexLiteral) offset).getValue()).intValue();
+        rangeBuilder.setLower(lower);
         rangeBuilder.setUpper(
                 fetch == null
                         ? Integer.MAX_VALUE
-                        : ((Number) ((RexLiteral) fetch).getValue()).intValue());
+                        : lower + ((Number) ((RexLiteral) fetch).getValue()).intValue());
         return rangeBuilder.build();
     }
 
@@ -667,6 +805,13 @@ private void addQueryFilters(
         }
     }
 
+    private void addQueryColumns(
+            GraphAlgebra.QueryParams.Builder paramsBuilder, Set<GraphNameOrId> columns) {
+        for (GraphNameOrId column : columns) {
+            paramsBuilder.addColumns(Utils.protoNameOrId(column));
+        }
+    }
+
     private GraphAlgebra.QueryParams.Builder buildQueryParams(AbstractBindableTableScan tableScan) {
         GraphAlgebra.QueryParams.Builder paramsBuilder = defaultQueryParams();
         addQueryTables(paramsBuilder, getGraphLabels(tableScan).getLabelsEntry());
@@ -674,6 +819,67 @@ private GraphAlgebra.QueryParams.Builder buildQueryParams(AbstractBindableTableS
         return paramsBuilder;
     }
 
+    private void addRepartitionToAnother(int repartitionKey) {
+        addRepartitionToAnother(physicalBuilder, repartitionKey);
+    }
+
+    private void addRepartitionToAnother(
+            GraphAlgebraPhysical.PhysicalPlan.Builder physicalBuilder, int repartitionKey) {
+        GraphAlgebraPhysical.PhysicalOpr.Builder repartitionOprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.Repartition repartition =
+                Utils.protoShuffleRepartition(repartitionKey);
+        repartitionOprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setRepartition(repartition));
+        physicalBuilder.addPlan(repartitionOprBuilder.build());
+    }
+
+    private void addAuxilia(
+            GraphAlgebraPhysical.PhysicalPlan.Builder physicalBuilder,
+            Integer tag,
+            Set<GraphNameOrId> columns) {
+        GraphAlgebraPhysical.PhysicalOpr.Builder auxiliaOprBuilder =
+                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+        GraphAlgebraPhysical.GetV.Builder vertexBuilder = GraphAlgebraPhysical.GetV.newBuilder();
+        vertexBuilder.setOpt(Utils.protoGetVOpt(PhysicalGetVOpt.ITSELF));
+        GraphAlgebra.QueryParams.Builder paramsBuilder = defaultQueryParams();
+        addQueryColumns(paramsBuilder, columns);
+        vertexBuilder.setParams(paramsBuilder);
+        if (tag != AliasInference.DEFAULT_ID) {
+            vertexBuilder.setTag(Utils.asAliasId(tag));
+            vertexBuilder.setAlias(Utils.asAliasId(tag));
+        }
+        auxiliaOprBuilder.setOpr(
+                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setVertex(vertexBuilder));
+        physicalBuilder.addPlan(auxiliaOprBuilder.build());
+    }
+
+    private void lazyPropertyFetching(Map<Integer, Set<GraphNameOrId>> columns) {
+        // by default, we cache the result of the tagColumns, i.e., the optimizedNoCaching is false
+        lazyPropertyFetching(columns, false);
+    }
+
+    private void lazyPropertyFetching(
+            Map<Integer, Set<GraphNameOrId>> tagColumns, boolean optimizedNoCaching) {
+        lazyPropertyFetching(physicalBuilder, tagColumns, optimizedNoCaching);
+    }
+
+    private void lazyPropertyFetching(
+            GraphAlgebraPhysical.PhysicalPlan.Builder physicalBuilder,
+            Map<Integer, Set<GraphNameOrId>> tagColumns,
+            boolean optimizedNoCaching) {
+        if (tagColumns.isEmpty()) {
+            return;
+        } else if (tagColumns.size() == 1 && optimizedNoCaching) {
+            addRepartitionToAnother(physicalBuilder, tagColumns.keySet().iterator().next());
+        } else {
+            for (Map.Entry<Integer, Set<GraphNameOrId>> tagColumn : tagColumns.entrySet()) {
+                addRepartitionToAnother(physicalBuilder, tagColumn.getKey());
+                addAuxilia(physicalBuilder, tagColumn.getKey(), tagColumn.getValue());
+            }
+        }
+    }
+
     @Override
     public RelNode visit(GraphLogicalSingleMatch 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 e57e88721a3c..69b2a6692eaa 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
@@ -16,9 +16,9 @@
 
 package com.alibaba.graphscope.common.ir.runtime.proto;
 
-import static com.alibaba.graphscope.gaia.proto.GraphAlgebra.GroupBy.AggFunc.Aggregate;
-
 import com.alibaba.graphscope.common.ir.rel.type.group.GraphAggCall;
+import com.alibaba.graphscope.common.ir.rex.RexVariableAliasCollector;
+import com.alibaba.graphscope.common.ir.tools.AliasInference;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
 import com.alibaba.graphscope.common.ir.type.GraphLabelType;
 import com.alibaba.graphscope.common.ir.type.GraphNameOrId;
@@ -27,6 +27,7 @@
 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.GraphAlgebra.GroupBy.AggFunc.Aggregate;
 import com.alibaba.graphscope.gaia.proto.GraphAlgebraPhysical;
 import com.alibaba.graphscope.gaia.proto.OuterExpression;
 import com.google.common.base.Preconditions;
@@ -36,15 +37,22 @@
 import org.apache.calcite.rel.RelFieldCollation;
 import org.apache.calcite.rel.core.JoinRelType;
 import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.SqlOperator;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.util.NlsString;
 import org.apache.calcite.util.Sarg;
+import org.javatuples.Pair;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 /**
@@ -620,4 +628,97 @@ public static final List<GraphAlgebraPhysical.PhysicalOpr.MetaData> physicalProt
                                 + " to List<MetaData> is unsupported");
         }
     }
+
+    public static GraphAlgebraPhysical.Repartition protoShuffleRepartition(int keyId) {
+        GraphAlgebraPhysical.Repartition.Shuffle.Builder shuffleBuilder =
+                GraphAlgebraPhysical.Repartition.Shuffle.newBuilder();
+        if (keyId != AliasInference.DEFAULT_ID) {
+            shuffleBuilder.setShuffleKey(asAliasId(keyId));
+        }
+        return GraphAlgebraPhysical.Repartition.newBuilder().setToAnother(shuffleBuilder).build();
+    }
+
+    public static Map<Integer, Set<GraphNameOrId>> extractTagColumnsFromRexNodes(
+            List<RexNode> exprs) {
+        return exprs.stream()
+                .map(
+                        expr ->
+                                expr.accept(
+                                        new RexVariableAliasCollector<Pair<Integer, GraphNameOrId>>(
+                                                true,
+                                                var -> {
+                                                    if (var.getProperty() != null
+                                                            && (GraphProperty.Opt.ALL.equals(
+                                                                            var.getProperty()
+                                                                                    .getOpt())
+                                                                    || GraphProperty.Opt.KEY.equals(
+                                                                            var.getProperty()
+                                                                                    .getOpt()))) {
+                                                        return Pair.with(
+                                                                var.getAliasId(),
+                                                                var.getProperty().getKey());
+
+                                                    } else return Pair.with(null, null);
+                                                })))
+                .flatMap(List::stream)
+                .filter(k -> k.getValue0() != null && k.getValue1() != null)
+                .collect(
+                        Collectors.groupingBy(
+                                pair -> pair.getValue0(),
+                                Collectors.mapping(pair -> pair.getValue1(), Collectors.toSet())));
+    }
+
+    // extract columns from relDataType, and return e.g., {name, age}
+    public static Set<GraphNameOrId> extractColumnsFromRelDataType(
+            RelDataType relDataType, boolean isColumnId) {
+        List<RelDataTypeField> recordColumns = relDataType.getFieldList();
+        Set<GraphNameOrId> columns = new HashSet<>();
+        for (int i = 0; i < recordColumns.size(); ++i) {
+            RelDataType recordColumnType = recordColumns.get(i).getType();
+            // if current column is a graph schema type, we extract all the fields (i.e., property
+            // types) from it
+            if (recordColumnType instanceof GraphSchemaType) {
+                List<RelDataTypeField> propertyTypes =
+                        ((GraphSchemaType) recordColumnType).getFieldList();
+                for (RelDataTypeField propertyType : propertyTypes) {
+                    if (isColumnId) {
+                        columns.add(new GraphNameOrId(propertyType.getIndex()));
+                    } else {
+                        columns.add(new GraphNameOrId(propertyType.getName()));
+                    }
+                }
+            }
+        }
+        return columns;
+    }
+
+    // remove edge properties from columns by checking if the tags refers to edge type
+    public static void removeEdgeProperties(
+            RelDataType inputDataType, Map<Integer, Set<GraphNameOrId>> tagColumns) {
+        List<RelDataTypeField> fieldTypes = inputDataType.getFieldList();
+        Set<Integer> tags = tagColumns.keySet();
+        // first, process the *HEAD* separately since it is a special case
+        if (tags.contains(AliasInference.DEFAULT_ID)) {
+            RelDataTypeField headFieldType = fieldTypes.get(fieldTypes.size() - 1);
+            if (headFieldType.getType() instanceof GraphSchemaType
+                    && GraphOpt.Source.EDGE.equals(
+                            ((GraphSchemaType) headFieldType.getType()).getScanOpt())) {
+                tags.remove(AliasInference.DEFAULT_ID);
+            }
+        }
+        if (tags.isEmpty()) {
+            return;
+        }
+        // then, process other tags by checking if they are of edge type
+        List<Integer> removeKeys = new ArrayList<>();
+        for (RelDataTypeField fieldType : fieldTypes) {
+            if (tags.contains(fieldType.getIndex())
+                    && fieldType.getType() instanceof GraphSchemaType
+                    && GraphOpt.Source.EDGE.equals(
+                            ((GraphSchemaType) fieldType.getType()).getScanOpt())) {
+                removeKeys.add(fieldType.getIndex());
+            }
+        }
+        tagColumns.keySet().removeAll(removeKeys);
+    }
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/Utils.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/Utils.java
index 7923394ba0e6..130bd4f33547 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/Utils.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/Utils.java
@@ -31,7 +31,6 @@
 
 import java.util.List;
 import java.util.Set;
-import java.util.stream.Collectors;
 
 public class Utils {
     public static RelDataType getOutputType(RelNode topNode) {
@@ -45,16 +44,20 @@ public static RelDataType getOutputType(RelNode topNode) {
             }
             inputs.addAll(cur.getInputs());
         }
-        Set<String> fieldNames = Sets.newHashSet();
-        List<RelDataTypeField> dedup =
-                outputFields.stream()
-                        .filter(
-                                k -> {
-                                    boolean notExist = !fieldNames.contains(k.getName());
-                                    fieldNames.add(k.getName());
-                                    return notExist;
-                                })
-                        .collect(Collectors.toList());
+        Set<String> uniqueNames = Sets.newHashSet();
+        // if field name is duplicated, we dedup it and keep the last one
+        List<RelDataTypeField> dedup = Lists.newArrayList();
+        for (int i = outputFields.size() - 1; i >= 0; i--) {
+            RelDataTypeField field = outputFields.get(i);
+            // specific implementation for gremlin `head`, DEFAULT can only denote the last field
+            if (field.getName() == AliasInference.DEFAULT_NAME && i != outputFields.size() - 1) {
+                continue;
+            }
+            if (!uniqueNames.contains(field.getName())) {
+                uniqueNames.add(field.getName());
+                dedup.add(0, field);
+            }
+        }
         return new RelRecordType(StructKind.FULLY_QUALIFIED, dedup);
     }
 
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/graph/calcite/RemoteTestGraph.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/graph/calcite/RemoteTestGraph.java
index 0a4d92befb0d..23af172497f7 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/graph/calcite/RemoteTestGraph.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/graph/calcite/RemoteTestGraph.java
@@ -1031,18 +1031,18 @@
 // "g_VX1X_outEXknowsX_asXhereX_hasXweight_1X_asXfakeX_inV_hasXname_joshX_selectXhereX",
 //        test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.SelectTest",
 //        reason = "returned label is id")
-// @Graph.OptOut(
-//        method = "g_VX1X_outEXknowsX_asXhereX_hasXweight_1X_inV_hasXname_joshX_selectXhereX",
-//        test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.SelectTest",
-//        reason = "returned label is id")
-// @Graph.OptOut(
-//        method = "g_VX1X_outEXknowsX_hasXweight_1X_asXhereX_inV_hasXname_joshX_selectXhereX",
-//        test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.SelectTest",
-//        reason = "returned label is id")
-// @Graph.OptOut(
-//        method = "g_VX1X_outE_asXhereX_inV_hasXname_vadasX_selectXhereX",
-//        test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.SelectTest",
-//        reason = "returned label is id")
+@Graph.OptOut(
+        method = "g_VX1X_outEXknowsX_asXhereX_hasXweight_1X_inV_hasXname_joshX_selectXhereX",
+        test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.SelectTest",
+        reason = "tested in IrGremlinQueryTest")
+@Graph.OptOut(
+        method = "g_VX1X_outEXknowsX_hasXweight_1X_asXhereX_inV_hasXname_joshX_selectXhereX",
+        test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.SelectTest",
+        reason = "tested in IrGremlinQueryTest")
+@Graph.OptOut(
+        method = "g_VX1X_outE_asXhereX_inV_hasXname_vadasX_selectXhereX",
+        test = "org.apache.tinkerpop.gremlin.process.traversal.step.map.SelectTest",
+        reason = "tested in IrGremlinQueryTest")
 
 // add more ignored tests which are out of ir range
 @Graph.OptOut(
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/standard/IrGremlinQueryTest.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/standard/IrGremlinQueryTest.java
index 8ca155d066ae..c206dd1ab928 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/standard/IrGremlinQueryTest.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/standard/IrGremlinQueryTest.java
@@ -113,6 +113,17 @@ public abstract class IrGremlinQueryTest extends AbstractGremlinProcessTest {
 
     public abstract Traversal<Vertex, Object> get_g_V_has_fold_select_as_unfold_values();
 
+    public abstract Traversal<Vertex, Edge>
+            get_g_VX1X_outE_asXhereX_inV_hasXname_vadasX_selectXhereX(final Object v1Id);
+
+    public abstract Traversal<Vertex, Edge>
+            get_g_VX1X_outEXknowsX_hasXweight_1X_asXhereX_inV_hasXname_joshX_selectXhereX(
+                    final Object v1Id);
+
+    public abstract Traversal<Vertex, Edge>
+            get_g_VX1X_outEXknowsX_asXhereX_hasXweight_1X_inV_hasXname_joshX_selectXhereX(
+                    final Object v1Id);
+
     @LoadGraphWith(LoadGraphWith.GraphData.MODERN)
     @Test
     public void g_V_group_by_by_dedup_count_test() {
@@ -646,6 +657,50 @@ public void g_V_has_fold_select_as_unfold_values() {
         Assert.assertEquals(1, counter);
     }
 
+    @Test
+    @LoadGraphWith(LoadGraphWith.GraphData.MODERN)
+    public void g_VX1X_outE_asXhereX_inV_hasXname_vadasX_selectXhereX() {
+        final Traversal<Vertex, Edge> traversal =
+                get_g_VX1X_outE_asXhereX_inV_hasXname_vadasX_selectXhereX(
+                        convertToVertexId("marko"));
+        printTraversalForm(traversal);
+        final Edge edge = traversal.next();
+        Assert.assertEquals("knows", edge.label());
+        Assert.assertEquals(convertToVertexId("vadas"), edge.inVertex().id());
+        Assert.assertEquals(convertToVertexId("marko"), edge.outVertex().id());
+        Assert.assertFalse(traversal.hasNext());
+    }
+
+    @Test
+    @LoadGraphWith(LoadGraphWith.GraphData.MODERN)
+    public void g_VX1X_outEXknowsX_hasXweight_1X_asXhereX_inV_hasXname_joshX_selectXhereX() {
+        final Traversal<Vertex, Edge> traversal =
+                get_g_VX1X_outEXknowsX_hasXweight_1X_asXhereX_inV_hasXname_joshX_selectXhereX(
+                        convertToVertexId("marko"));
+        printTraversalForm(traversal);
+        Assert.assertTrue(traversal.hasNext());
+        final Edge edge = traversal.next();
+        Assert.assertEquals("knows", edge.label());
+        Assert.assertEquals(convertToVertexId("josh"), edge.inVertex().id());
+        Assert.assertEquals(convertToVertexId("marko"), edge.outVertex().id());
+        Assert.assertFalse(traversal.hasNext());
+    }
+
+    @Test
+    @LoadGraphWith(LoadGraphWith.GraphData.MODERN)
+    public void g_VX1X_outEXknowsX_asXhereX_hasXweight_1X_inV_hasXname_joshX_selectXhereX() {
+        final Traversal<Vertex, Edge> traversal =
+                get_g_VX1X_outEXknowsX_asXhereX_hasXweight_1X_inV_hasXname_joshX_selectXhereX(
+                        convertToVertexId("marko"));
+        printTraversalForm(traversal);
+        Assert.assertTrue(traversal.hasNext());
+        final Edge edge = traversal.next();
+        Assert.assertEquals("knows", edge.label());
+        Assert.assertEquals(convertToVertexId("josh"), edge.inVertex().id());
+        Assert.assertEquals(convertToVertexId("marko"), edge.outVertex().id());
+        Assert.assertFalse(traversal.hasNext());
+    }
+
     public static class Traversals extends IrGremlinQueryTest {
 
         @Override
@@ -863,5 +918,37 @@ public Traversal<Vertex, Object> get_g_V_has_fold_select_as_unfold_values() {
                     .unfold()
                     .values("id");
         }
+
+        @Override
+        public Traversal<Vertex, Edge>
+                get_g_VX1X_outEXknowsX_hasXweight_1X_asXhereX_inV_hasXname_joshX_selectXhereX(
+                        final Object v1Id) {
+            return g.V(v1Id)
+                    .outE("knows")
+                    .has("weight", 1.0d)
+                    .as("here")
+                    .inV()
+                    .has("name", "josh")
+                    .select("here");
+        }
+
+        @Override
+        public Traversal<Vertex, Edge>
+                get_g_VX1X_outEXknowsX_asXhereX_hasXweight_1X_inV_hasXname_joshX_selectXhereX(
+                        final Object v1Id) {
+            return g.V(v1Id)
+                    .outE("knows")
+                    .as("here")
+                    .has("weight", 1.0d)
+                    .inV()
+                    .has("name", "josh")
+                    .select("here");
+        }
+
+        @Override
+        public Traversal<Vertex, Edge> get_g_VX1X_outE_asXhereX_inV_hasXname_vadasX_selectXhereX(
+                final Object v1Id) {
+            return g.V(v1Id).outE().as("here").inV().has("name", "vadas").select("here");
+        }
     }
 }


Test patch:
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
index 4a65025ce2f6..0f166f05d267 100644
--- 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
@@ -37,7 +37,6 @@
 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;
@@ -149,6 +148,17 @@ public void edge_expand_test() throws Exception {
                     FileUtils.readJsonFromResource("proto/edge_expand_test.json"),
                     plan.explain().trim());
         }
+
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(expand))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_edge_expand_test.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -184,6 +194,15 @@ public void get_v_test() throws Exception {
             Assert.assertEquals(
                     FileUtils.readJsonFromResource("proto/getv_test.json"), plan.explain().trim());
         }
+
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(), Utils.schemaMeta, new LogicalPlan(getV))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_getv_test.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -299,6 +318,64 @@ public void project_test() throws Exception {
                     FileUtils.readJsonFromResource("proto/project_test.json"),
                     plan.explain().trim());
         }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(project))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_project_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void project_02_test() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode project =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows"),
+                                        "y"))
+                        .project(
+                                ImmutableList.of(
+                                        builder.variable("x", "name"),
+                                        builder.variable("y", "weight")),
+                                ImmutableList.of("name"))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalProject(name=[x.name], weight=[y.weight],"
+                        + " isAppend=[false])\n"
+                        + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
+                        + " alias=[y], opt=[OUT])\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_2.json"),
+                    plan.explain().trim());
+        }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(project))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_project_test_2.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -330,6 +407,53 @@ public void filter_test() throws Exception {
         }
     }
 
+    @Test
+    public void filter_test_2() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode filter =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows"),
+                                        "y"))
+                        .filter(
+                                builder.call(
+                                        GraphStdOperatorTable.GREATER_THAN,
+                                        builder.variable("x", "age"),
+                                        builder.variable("y", "weight")))
+                        .build();
+        Assert.assertEquals(
+                "LogicalFilter(condition=[>(x.age, y.weight)])\n"
+                    + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}], alias=[y],"
+                    + " opt=[OUT])\n"
+                    + "    GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                    + " alias=[x], 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_2.json"),
+                    plan.explain().trim());
+        }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(filter))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_filter_test_2.json"),
+                    plan.explain().trim());
+        }
+    }
+
     @Test
     public void aggregate_test() throws Exception {
         GraphBuilder builder = Utils.mockGraphBuilder();
@@ -358,6 +482,68 @@ public void aggregate_test() throws Exception {
                     FileUtils.readJsonFromResource("proto/aggregate_test.json"),
                     plan.explain().trim());
         }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(aggregate))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_aggregate_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    @Test
+    public void aggregate_test_2() throws Exception {
+        GraphBuilder builder = Utils.mockGraphBuilder();
+        RelNode aggregate =
+                builder.source(
+                                new SourceConfig(
+                                        GraphOpt.Source.VERTEX,
+                                        new LabelConfig(false).addLabel("person"),
+                                        "x"))
+                        .expand(
+                                new ExpandConfig(
+                                        GraphOpt.Expand.OUT,
+                                        new LabelConfig(false).addLabel("knows"),
+                                        "y"))
+                        .aggregate(
+                                builder.groupKey(
+                                        builder.variable("x", "name"),
+                                        builder.variable("y", "weight")),
+                                builder.collect(
+                                        ImmutableList.of(
+                                                builder.variable("x", "name"),
+                                                builder.variable("y", "weight"))))
+                        .build();
+        Assert.assertEquals(
+                "GraphLogicalAggregate(keys=[{variables=[x.name, y.weight], aliases=[name,"
+                    + " weight]}], values=[[{operands=[x.name, y.weight], aggFunction=COLLECT,"
+                    + " alias='$f2', distinct=false}]])\n"
+                    + "  GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}], alias=[y],"
+                    + " opt=[OUT])\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_2.json"),
+                    plan.explain().trim());
+        }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(aggregate))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_aggregate_test_2.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -386,6 +572,16 @@ public void dedup_test_1() throws Exception {
                     FileUtils.readJsonFromResource("proto/dedup_test_1.json"),
                     plan.explain().trim());
         }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(dedup))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_dedup_test_1.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -412,6 +608,16 @@ public void dedup_test_2() throws Exception {
                     FileUtils.readJsonFromResource("proto/dedup_test_2.json"),
                     plan.explain().trim());
         }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(dedup))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_dedup_test_2.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -429,18 +635,24 @@ public void join_test_1() throws Exception {
                                 new SourceConfig(
                                         GraphOpt.Source.VERTEX,
                                         new LabelConfig(false).addLabel("person"),
-                                        "x"))
+                                        "y"))
                         .build();
-        RexNode condition = builder.getJoinCondition(source1, source2);
+
         builder.push(source1);
         builder.push(source2);
-        RelNode join = builder.join(JoinRelType.INNER, condition).build();
+        RelNode join =
+                builder.join(
+                                JoinRelType.INNER,
+                                builder.equals(
+                                        builder.variable("x", "name"),
+                                        builder.variable("y", "name")))
+                        .build();
         Assert.assertEquals(
-                "LogicalJoin(condition=[=(x, x)], joinType=[inner])\n"
+                "LogicalJoin(condition=[=(x.name, y.name)], joinType=[inner])\n"
                         + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
                         + " alias=[x], opt=[VERTEX])\n"
                         + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
-                        + " alias=[x], opt=[VERTEX])",
+                        + " alias=[y], opt=[VERTEX])",
                 join.explain().trim());
         try (PhysicalBuilder protoBuilder =
                 new GraphRelProtoPhysicalBuilder(
@@ -450,6 +662,14 @@ public void join_test_1() throws Exception {
                     FileUtils.readJsonFromResource("proto/join_test_1.json"),
                     plan.explain().trim());
         }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(), Utils.schemaMeta, new LogicalPlan(join))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_join_test_1.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -488,15 +708,20 @@ public void join_test_2() throws Exception {
                                 new GetVConfig(
                                         GraphOpt.GetV.START,
                                         new LabelConfig(false).addLabel("person"),
-                                        "x",
+                                        "y",
                                         "d"))
                         .build();
-        RexNode condition = builder.getJoinCondition(expand1, expand2);
         builder.push(expand1);
         builder.push(expand2);
-        RelNode join = builder.join(JoinRelType.INNER, condition).build();
+        RelNode join =
+                builder.join(
+                                JoinRelType.INNER,
+                                builder.equals(
+                                        builder.variable("x", "name"),
+                                        builder.variable("y", "name")))
+                        .build();
         Assert.assertEquals(
-                "LogicalJoin(condition=[=(x, x)], joinType=[inner])\n"
+                "LogicalJoin(condition=[=(x.name, y.name)], joinType=[inner])\n"
                         + "  GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
                         + " alias=[x], startAlias=[b], opt=[END])\n"
                         + "    GraphLogicalExpand(tableConfig=[{isAll=false, tables=[knows]}],"
@@ -504,7 +729,7 @@ public void join_test_2() throws Exception {
                         + "      GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
                         + " alias=[a], opt=[VERTEX])\n"
                         + "  GraphLogicalGetV(tableConfig=[{isAll=false, tables=[person]}],"
-                        + " alias=[x], startAlias=[d], opt=[START])\n"
+                        + " alias=[y], startAlias=[d], opt=[START])\n"
                         + "    GraphLogicalExpand(tableConfig=[{isAll=false, tables=[created]}],"
                         + " alias=[d], opt=[IN])\n"
                         + "      GraphLogicalSource(tableConfig=[{isAll=false, tables=[software]}],"
@@ -518,6 +743,14 @@ public void join_test_2() throws Exception {
                     FileUtils.readJsonFromResource("proto/join_test_2.json"),
                     plan.explain().trim());
         }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(), Utils.schemaMeta, new LogicalPlan(join))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_join_test_2.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -542,6 +775,14 @@ public void sort_test() throws Exception {
             Assert.assertEquals(
                     FileUtils.readJsonFromResource("proto/sort_test.json"), plan.explain().trim());
         }
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(), Utils.schemaMeta, new LogicalPlan(sort))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/partitioned_sort_test.json"),
+                    plan.explain().trim());
+        }
     }
 
     @Test
@@ -608,6 +849,18 @@ public void expand_degree_test() throws Exception {
                     FileUtils.readJsonFromResource("proto/edge_expand_degree_test.json"),
                     plan.explain().trim());
         }
+
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(after))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource(
+                            "proto/partitioned_edge_expand_degree_test.json"),
+                    plan.explain().trim());
+        }
     }
 
     // g.V().hasLabel("person").outE("knows").inV()
@@ -654,6 +907,78 @@ public void expand_vertex_test() throws Exception {
                     FileUtils.readJsonFromResource("proto/edge_expand_vertex_test.json"),
                     plan.explain().trim());
         }
+
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(after))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource(
+                            "proto/partitioned_edge_expand_vertex_test.json"),
+                    plan.explain().trim());
+        }
+    }
+
+    // g.V().hasLabel("person").outE("knows").inV().has("age",10), can be fused into
+    // GraphPhysicalExpand + GraphPhysicalGetV
+    @Test
+    public void expand_vertex_filter_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"),
+                                        "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();
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(after))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("proto/edge_expand_vertex_filter_test.json"),
+                    plan.explain().trim());
+        }
+
+        try (PhysicalBuilder protoBuilder =
+                new GraphRelProtoPhysicalBuilder(
+                        getMockPartitionedGraphConfig(),
+                        Utils.schemaMeta,
+                        new LogicalPlan(after))) {
+            PhysicalPlan plan = protoBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource(
+                            "proto/partitioned_edge_expand_vertex_filter_test.json"),
+                    plan.explain().trim());
+        }
     }
 
     // g.V().hasLabel("person").outE("knows").has("weight", 0.5).inV().has("age", 10)
@@ -760,6 +1085,11 @@ public void path_expand_fused_test() throws Exception {
     }
 
     private Configs getMockGraphConfig() {
-        return new Configs(ImmutableMap.of("servers", "1", "workers", "1"));
+        return new Configs(ImmutableMap.of("pegasus.hosts", "1", "pegasus.worker.num", "1"));
+    }
+
+    private Configs getMockPartitionedGraphConfig() {
+        return new Configs(
+                ImmutableMap.of("pegasus.hosts", "host1,host2", "pegasus.worker.num", "2"));
     }
 }
diff --git a/interactive_engine/compiler/src/test/resources/proto/aggregate_test_2.json b/interactive_engine/compiler/src/test/resources/proto/aggregate_test_2.json
new file mode 100644
index 000000000000..11b234e9d4a2
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/aggregate_test_2.json
@@ -0,0 +1,166 @@
+{
+  "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
+          }],
+          "columns": [{
+            "name": "weight"
+          }],
+          "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": {
+      "groupBy": {
+        "mappings": [{
+          "key": {
+            "tag": {
+              "id": 0
+            },
+            "property": {
+              "key": {
+                "name": "name"
+              }
+            },
+            "nodeType": {
+              "dataType": "STRING"
+            }
+          },
+          "alias": 2
+        }, {
+          "key": {
+            "tag": {
+              "id": 1
+            },
+            "property": {
+              "key": {
+                "name": "weight"
+              }
+            },
+            "nodeType": {
+              "dataType": "DOUBLE"
+            }
+          },
+          "alias": 3
+        }],
+        "functions": [{
+          "vars": [{
+            "tag": {
+              "id": 0
+            },
+            "property": {
+              "key": {
+                "name": "name"
+              }
+            },
+            "nodeType": {
+              "dataType": "STRING"
+            }
+          }, {
+            "tag": {
+              "id": 1
+            },
+            "property": {
+              "key": {
+                "name": "weight"
+              }
+            },
+            "nodeType": {
+              "dataType": "DOUBLE"
+            }
+          }],
+          "aggregate": "TO_LIST",
+          "alias": 4
+        }]
+      }
+    },
+    "metaData": [{
+      "type": {
+        "dataType": "STRING"
+      },
+      "alias": 2
+    }, {
+      "type": {
+        "dataType": "DOUBLE"
+      },
+      "alias": 3
+    }, {
+      "type": {
+      },
+      "alias": 4
+    }]
+  }, {
+    "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
index 4b9c5288cb05..724da2c6c941 100644
--- a/interactive_engine/compiler/src/test/resources/proto/dedup_test_2.json
+++ b/interactive_engine/compiler/src/test/resources/proto/dedup_test_2.json
@@ -54,7 +54,33 @@
           }
         }]
       }
-    }
+    },
+    "metaData": [{
+      "type": {
+        "graphType": {
+          "graphDataType": [{
+            "label": {
+            },
+            "props": [{
+              "propId": {
+                "name": "id"
+              },
+              "type": "INT64"
+            }, {
+              "propId": {
+                "name": "name"
+              },
+              "type": "STRING"
+            }, {
+              "propId": {
+                "name": "age"
+              },
+              "type": "INT32"
+            }]
+          }]
+        }
+      }
+    }]
   }, {
     "opr": {
       "sink": {
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
index f4a0957c24d0..012ed5797438 100644
--- a/interactive_engine/compiler/src/test/resources/proto/edge_expand_test.json
+++ b/interactive_engine/compiler/src/test/resources/proto/edge_expand_test.json
@@ -44,6 +44,9 @@
           "tables": [{
             "id": 0
           }],
+          "columns": [{
+            "name": "weight"
+          }],
           "sampleRatio": 1.0
         },
         "expandOpt": "EDGE"
diff --git a/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_filter_test.json b/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_filter_test.json
new file mode 100644
index 000000000000..c5331ee3a18e
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/edge_expand_vertex_filter_test.json
@@ -0,0 +1,155 @@
+{
+  "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
+        },
+        "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_2.json b/interactive_engine/compiler/src/test/resources/proto/filter_test_2.json
new file mode 100644
index 000000000000..e3864551d52a
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/filter_test_2.json
@@ -0,0 +1,153 @@
+{
+  "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
+          }],
+          "columns": [{
+            "name": "weight"
+          }],
+          "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": {
+      "select": {
+        "predicate": {
+          "operators": [{
+            "var": {
+              "tag": {
+                "id": 0
+              },
+              "property": {
+                "key": {
+                  "name": "age"
+                }
+              },
+              "nodeType": {
+                "dataType": "INT32"
+              }
+            },
+            "nodeType": {
+              "dataType": "INT32"
+            }
+          }, {
+            "logical": "GT",
+            "nodeType": {
+              "dataType": "BOOLEAN"
+            }
+          }, {
+            "var": {
+              "tag": {
+                "id": 1
+              },
+              "property": {
+                "key": {
+                  "name": "weight"
+                }
+              },
+              "nodeType": {
+                "dataType": "DOUBLE"
+              }
+            },
+            "nodeType": {
+              "dataType": "DOUBLE"
+            }
+          }]
+        }
+      }
+    },
+    "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/getv_test.json b/interactive_engine/compiler/src/test/resources/proto/getv_test.json
index 69b6137d724c..019ba459844a 100644
--- a/interactive_engine/compiler/src/test/resources/proto/getv_test.json
+++ b/interactive_engine/compiler/src/test/resources/proto/getv_test.json
@@ -44,6 +44,9 @@
           "tables": [{
             "id": 0
           }],
+          "columns": [{
+            "name": "weight"
+          }],
           "sampleRatio": 1.0
         },
         "expandOpt": "EDGE"
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
index 52f3d0b95ce9..d0d3240ca53f 100644
--- 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
@@ -44,6 +44,9 @@
           "tables": [{
             "id": 0
           }],
+          "columns": [{
+            "name": "weight"
+          }],
           "sampleRatio": 1.0
         },
         "expandOpt": "EDGE"
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
index 157d55ed5a24..3c0be266b6b9 100644
--- a/interactive_engine/compiler/src/test/resources/proto/join_test_1.json
+++ b/interactive_engine/compiler/src/test/resources/proto/join_test_1.json
@@ -6,58 +6,26 @@
           "tag": {
             "id": 0
           },
-          "nodeType": {
-            "graphType": {
-              "graphDataType": [{
-                "label": {
-                },
-                "props": [{
-                  "propId": {
-                    "name": "id"
-                  },
-                  "type": "INT64"
-                }, {
-                  "propId": {
-                    "name": "name"
-                  },
-                  "type": "STRING"
-                }, {
-                  "propId": {
-                    "name": "age"
-                  },
-                  "type": "INT32"
-                }]
-              }]
+          "property": {
+            "key": {
+              "name": "name"
             }
+          },
+          "nodeType": {
+            "dataType": "STRING"
           }
         }],
         "rightKeys": [{
           "tag": {
-            "id": 0
+            "id": 1
           },
-          "nodeType": {
-            "graphType": {
-              "graphDataType": [{
-                "label": {
-                },
-                "props": [{
-                  "propId": {
-                    "name": "id"
-                  },
-                  "type": "INT64"
-                }, {
-                  "propId": {
-                    "name": "name"
-                  },
-                  "type": "STRING"
-                }, {
-                  "propId": {
-                    "name": "age"
-                  },
-                  "type": "INT32"
-                }]
-              }]
+          "property": {
+            "key": {
+              "name": "name"
             }
+          },
+          "nodeType": {
+            "dataType": "STRING"
           }
         }],
         "leftPlan": {
@@ -105,7 +73,7 @@
           "plan": [{
             "opr": {
               "scan": {
-                "alias": 0,
+                "alias": 1,
                 "params": {
                   "tables": [{
                     "id": 0
@@ -138,7 +106,8 @@
                     }]
                   }]
                 }
-              }
+              },
+              "alias": 1
             }]
           }]
         }
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
index 373e73a9a6c5..76e9d5fe8679 100644
--- a/interactive_engine/compiler/src/test/resources/proto/join_test_2.json
+++ b/interactive_engine/compiler/src/test/resources/proto/join_test_2.json
@@ -6,58 +6,26 @@
           "tag": {
             "id": 2
           },
-          "nodeType": {
-            "graphType": {
-              "graphDataType": [{
-                "label": {
-                },
-                "props": [{
-                  "propId": {
-                    "name": "id"
-                  },
-                  "type": "INT64"
-                }, {
-                  "propId": {
-                    "name": "name"
-                  },
-                  "type": "STRING"
-                }, {
-                  "propId": {
-                    "name": "age"
-                  },
-                  "type": "INT32"
-                }]
-              }]
+          "property": {
+            "key": {
+              "name": "name"
             }
+          },
+          "nodeType": {
+            "dataType": "STRING"
           }
         }],
         "rightKeys": [{
           "tag": {
-            "id": 2
+            "id": 5
           },
-          "nodeType": {
-            "graphType": {
-              "graphDataType": [{
-                "label": {
-                },
-                "props": [{
-                  "propId": {
-                    "name": "id"
-                  },
-                  "type": "INT64"
-                }, {
-                  "propId": {
-                    "name": "name"
-                  },
-                  "type": "STRING"
-                }, {
-                  "propId": {
-                    "name": "age"
-                  },
-                  "type": "INT32"
-                }]
-              }]
+          "property": {
+            "key": {
+              "name": "name"
             }
+          },
+          "nodeType": {
+            "dataType": "STRING"
           }
         }],
         "leftPlan": {
@@ -106,6 +74,9 @@
                   "tables": [{
                     "id": 0
                   }],
+                  "columns": [{
+                    "name": "weight"
+                  }],
                   "sampleRatio": 1.0
                 },
                 "alias": 1,
@@ -229,6 +200,9 @@
                   "tables": [{
                     "id": 1
                   }],
+                  "columns": [{
+                    "name": "weight"
+                  }],
                   "sampleRatio": 1.0
                 },
                 "alias": 4,
@@ -266,7 +240,7 @@
                   }],
                   "sampleRatio": 1.0
                 },
-                "alias": 2
+                "alias": 5
               }
             },
             "metaData": [{
@@ -294,7 +268,7 @@
                   }]
                 }
               },
-              "alias": 2
+              "alias": 5
             }]
           }]
         }
diff --git a/interactive_engine/compiler/src/test/resources/proto/partitioned_aggregate_test.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_aggregate_test.json
new file mode 100644
index 000000000000..6efc88dd5d12
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_aggregate_test.json
@@ -0,0 +1,137 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+          "shuffleKey": 0
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "vertex": {
+        "tag": 0,
+        "opt": "ITSELF",
+        "params": {
+          "columns": [{
+            "name": "name"
+          }],
+          "sampleRatio": 1.0
+        },
+        "alias": 0
+      }
+    }
+  }, {
+    "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/partitioned_aggregate_test_2.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_aggregate_test_2.json
new file mode 100644
index 000000000000..3b786b2e1b05
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_aggregate_test_2.json
@@ -0,0 +1,195 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "columns": [{
+            "name": "weight"
+          }],
+          "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": {
+      "repartition": {
+        "toAnother": {
+          "shuffleKey": 0
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "vertex": {
+        "tag": 0,
+        "opt": "ITSELF",
+        "params": {
+          "columns": [{
+            "name": "name"
+          }],
+          "sampleRatio": 1.0
+        },
+        "alias": 0
+      }
+    }
+  }, {
+    "opr": {
+      "groupBy": {
+        "mappings": [{
+          "key": {
+            "tag": {
+              "id": 0
+            },
+            "property": {
+              "key": {
+                "name": "name"
+              }
+            },
+            "nodeType": {
+              "dataType": "STRING"
+            }
+          },
+          "alias": 2
+        }, {
+          "key": {
+            "tag": {
+              "id": 1
+            },
+            "property": {
+              "key": {
+                "name": "weight"
+              }
+            },
+            "nodeType": {
+              "dataType": "DOUBLE"
+            }
+          },
+          "alias": 3
+        }],
+        "functions": [{
+          "vars": [{
+            "tag": {
+              "id": 0
+            },
+            "property": {
+              "key": {
+                "name": "name"
+              }
+            },
+            "nodeType": {
+              "dataType": "STRING"
+            }
+          }, {
+            "tag": {
+              "id": 1
+            },
+            "property": {
+              "key": {
+                "name": "weight"
+              }
+            },
+            "nodeType": {
+              "dataType": "DOUBLE"
+            }
+          }],
+          "aggregate": "TO_LIST",
+          "alias": 4
+        }]
+      }
+    },
+    "metaData": [{
+      "type": {
+        "dataType": "STRING"
+      },
+      "alias": 2
+    }, {
+      "type": {
+        "dataType": "DOUBLE"
+      },
+      "alias": 3
+    }, {
+      "type": {
+      },
+      "alias": 4
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/partitioned_dedup_test_1.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_dedup_test_1.json
new file mode 100644
index 000000000000..044b7d64c330
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_dedup_test_1.json
@@ -0,0 +1,113 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+          "shuffleKey": 0
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "vertex": {
+        "tag": 0,
+        "opt": "ITSELF",
+        "params": {
+          "columns": [{
+            "name": "name"
+          }],
+          "sampleRatio": 1.0
+        },
+        "alias": 0
+      }
+    }
+  }, {
+    "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/partitioned_dedup_test_2.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_dedup_test_2.json
new file mode 100644
index 000000000000..78f170cc1291
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_dedup_test_2.json
@@ -0,0 +1,116 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+          "shuffleKey": 0
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "vertex": {
+        "tag": 0,
+        "opt": "ITSELF",
+        "params": {
+          "columns": [{
+            "name": "name"
+          }],
+          "sampleRatio": 1.0
+        },
+        "alias": 0
+      }
+    }
+  }, {
+    "opr": {
+      "dedup": {
+        "keys": [{
+          "tag": {
+            "id": 0
+          },
+          "property": {
+            "key": {
+              "name": "name"
+            }
+          },
+          "nodeType": {
+            "dataType": "STRING"
+          }
+        }]
+      }
+    },
+    "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/partitioned_edge_expand_degree_test.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_edge_expand_degree_test.json
new file mode 100644
index 000000000000..651127d1ed42
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_edge_expand_degree_test.json
@@ -0,0 +1,93 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "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/partitioned_edge_expand_test.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_edge_expand_test.json
new file mode 100644
index 000000000000..864e887cd7ef
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_edge_expand_test.json
@@ -0,0 +1,92 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "columns": [{
+            "name": "weight"
+          }],
+          "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/partitioned_edge_expand_vertex_filter_test.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_edge_expand_vertex_filter_test.json
new file mode 100644
index 000000000000..c2d71b76ba90
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_edge_expand_vertex_filter_test.json
@@ -0,0 +1,169 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "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/partitioned_edge_expand_vertex_test.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_edge_expand_vertex_test.json
new file mode 100644
index 000000000000..73be464d1957
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_edge_expand_vertex_test.json
@@ -0,0 +1,95 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "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/partitioned_filter_test_2.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_filter_test_2.json
new file mode 100644
index 000000000000..fb01b4e7ed8b
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_filter_test_2.json
@@ -0,0 +1,168 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "columns": [{
+            "name": "weight"
+          }],
+          "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": {
+      "repartition": {
+        "toAnother": {
+          "shuffleKey": 0
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "select": {
+        "predicate": {
+          "operators": [{
+            "var": {
+              "tag": {
+                "id": 0
+              },
+              "property": {
+                "key": {
+                  "name": "age"
+                }
+              },
+              "nodeType": {
+                "dataType": "INT32"
+              }
+            },
+            "nodeType": {
+              "dataType": "INT32"
+            }
+          }, {
+            "logical": "GT",
+            "nodeType": {
+              "dataType": "BOOLEAN"
+            }
+          }, {
+            "var": {
+              "tag": {
+                "id": 1
+              },
+              "property": {
+                "key": {
+                  "name": "weight"
+                }
+              },
+              "nodeType": {
+                "dataType": "DOUBLE"
+              }
+            },
+            "nodeType": {
+              "dataType": "DOUBLE"
+            }
+          }]
+        }
+      }
+    },
+    "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/partitioned_getv_test.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_getv_test.json
new file mode 100644
index 000000000000..e1856e2114ac
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_getv_test.json
@@ -0,0 +1,131 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "columns": [{
+            "name": "weight"
+          }],
+          "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/partitioned_join_test_1.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_join_test_1.json
new file mode 100644
index 000000000000..5e13bd2762fe
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_join_test_1.json
@@ -0,0 +1,170 @@
+{
+  "plan": [{
+    "opr": {
+      "join": {
+        "leftKeys": [{
+          "tag": {
+            "id": 0
+          },
+          "property": {
+            "key": {
+              "name": "name"
+            }
+          },
+          "nodeType": {
+            "dataType": "STRING"
+          }
+        }],
+        "rightKeys": [{
+          "tag": {
+            "id": 1
+          },
+          "property": {
+            "key": {
+              "name": "name"
+            }
+          },
+          "nodeType": {
+            "dataType": "STRING"
+          }
+        }],
+        "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": {
+              "repartition": {
+                "toAnother": {
+                  "shuffleKey": 0
+                }
+              }
+            }
+          }, {
+            "opr": {
+              "vertex": {
+                "tag": 0,
+                "opt": "ITSELF",
+                "params": {
+                  "columns": [{
+                    "name": "name"
+                  }],
+                  "sampleRatio": 1.0
+                },
+                "alias": 0
+              }
+            }
+          }]
+        },
+        "rightPlan": {
+          "plan": [{
+            "opr": {
+              "scan": {
+                "alias": 1,
+                "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": {
+              "repartition": {
+                "toAnother": {
+                  "shuffleKey": 1
+                }
+              }
+            }
+          }, {
+            "opr": {
+              "vertex": {
+                "tag": 1,
+                "opt": "ITSELF",
+                "params": {
+                  "columns": [{
+                    "name": "name"
+                  }],
+                  "sampleRatio": 1.0
+                },
+                "alias": 1
+              }
+            }
+          }]
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/partitioned_join_test_2.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_join_test_2.json
new file mode 100644
index 000000000000..9bb6e40ca3bf
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_join_test_2.json
@@ -0,0 +1,345 @@
+{
+  "plan": [{
+    "opr": {
+      "join": {
+        "leftKeys": [{
+          "tag": {
+            "id": 2
+          },
+          "property": {
+            "key": {
+              "name": "name"
+            }
+          },
+          "nodeType": {
+            "dataType": "STRING"
+          }
+        }],
+        "rightKeys": [{
+          "tag": {
+            "id": 5
+          },
+          "property": {
+            "key": {
+              "name": "name"
+            }
+          },
+          "nodeType": {
+            "dataType": "STRING"
+          }
+        }],
+        "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": {
+              "repartition": {
+                "toAnother": {
+                }
+              }
+            }
+          }, {
+            "opr": {
+              "edge": {
+                "params": {
+                  "tables": [{
+                    "id": 0
+                  }],
+                  "columns": [{
+                    "name": "weight"
+                  }],
+                  "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
+            }]
+          }, {
+            "opr": {
+              "repartition": {
+                "toAnother": {
+                  "shuffleKey": 2
+                }
+              }
+            }
+          }, {
+            "opr": {
+              "vertex": {
+                "tag": 2,
+                "opt": "ITSELF",
+                "params": {
+                  "columns": [{
+                    "name": "name"
+                  }],
+                  "sampleRatio": 1.0
+                },
+                "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": {
+              "repartition": {
+                "toAnother": {
+                }
+              }
+            }
+          }, {
+            "opr": {
+              "edge": {
+                "direction": "IN",
+                "params": {
+                  "tables": [{
+                    "id": 1
+                  }],
+                  "columns": [{
+                    "name": "weight"
+                  }],
+                  "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": 5
+              }
+            },
+            "metaData": [{
+              "type": {
+                "graphType": {
+                  "graphDataType": [{
+                    "label": {
+                    },
+                    "props": [{
+                      "propId": {
+                        "name": "id"
+                      },
+                      "type": "INT64"
+                    }, {
+                      "propId": {
+                        "name": "name"
+                      },
+                      "type": "STRING"
+                    }, {
+                      "propId": {
+                        "name": "age"
+                      },
+                      "type": "INT32"
+                    }]
+                  }]
+                }
+              },
+              "alias": 5
+            }]
+          }, {
+            "opr": {
+              "repartition": {
+                "toAnother": {
+                  "shuffleKey": 5
+                }
+              }
+            }
+          }, {
+            "opr": {
+              "vertex": {
+                "tag": 5,
+                "opt": "ITSELF",
+                "params": {
+                  "columns": [{
+                    "name": "name"
+                  }],
+                  "sampleRatio": 1.0
+                },
+                "alias": 5
+              }
+            }
+          }]
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/partitioned_project_test.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_project_test.json
new file mode 100644
index 000000000000..2a2af6d9b8e0
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_project_test.json
@@ -0,0 +1,92 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+          "shuffleKey": 0
+        }
+      }
+    }
+  }, {
+    "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/partitioned_project_test_2.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_project_test_2.json
new file mode 100644
index 000000000000..f8840b20e38b
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_project_test_2.json
@@ -0,0 +1,162 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "edge": {
+        "params": {
+          "tables": [{
+            "id": 0
+          }],
+          "columns": [{
+            "name": "weight"
+          }],
+          "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": {
+      "repartition": {
+        "toAnother": {
+          "shuffleKey": 0
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "project": {
+        "mappings": [{
+          "expr": {
+            "operators": [{
+              "var": {
+                "tag": {
+                  "id": 0
+                },
+                "property": {
+                  "key": {
+                    "name": "name"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "STRING"
+                }
+              },
+              "nodeType": {
+                "dataType": "STRING"
+              }
+            }]
+          },
+          "alias": 2
+        }, {
+          "expr": {
+            "operators": [{
+              "var": {
+                "tag": {
+                  "id": 1
+                },
+                "property": {
+                  "key": {
+                    "name": "weight"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "DOUBLE"
+                }
+              },
+              "nodeType": {
+                "dataType": "DOUBLE"
+              }
+            }]
+          },
+          "alias": 3
+        }]
+      }
+    },
+    "metaData": [{
+      "type": {
+        "dataType": "STRING"
+      },
+      "alias": 2
+    }, {
+      "type": {
+        "dataType": "DOUBLE"
+      },
+      "alias": 3
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file
diff --git a/interactive_engine/compiler/src/test/resources/proto/partitioned_sort_test.json b/interactive_engine/compiler/src/test/resources/proto/partitioned_sort_test.json
new file mode 100644
index 000000000000..1a4eb6e52390
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/partitioned_sort_test.json
@@ -0,0 +1,87 @@
+{
+  "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": {
+      "repartition": {
+        "toAnother": {
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "vertex": {
+        "opt": "ITSELF",
+        "params": {
+          "columns": [{
+            "name": "name"
+          }],
+          "sampleRatio": 1.0
+        }
+      }
+    }
+  }, {
+    "opr": {
+      "orderBy": {
+        "pairs": [{
+          "key": {
+            "property": {
+              "key": {
+                "name": "name"
+              }
+            },
+            "nodeType": {
+              "dataType": "STRING"
+            }
+          },
+          "order": "ASC"
+        }]
+      }
+    }
+  }, {
+    "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
index cc18d0cac59b..339d34540f53 100644
--- a/interactive_engine/compiler/src/test/resources/proto/path_expand_test.json
+++ b/interactive_engine/compiler/src/test/resources/proto/path_expand_test.json
@@ -46,6 +46,9 @@
               "tables": [{
                 "id": 0
               }],
+              "columns": [{
+                "name": "weight"
+              }],
               "sampleRatio": 1.0
             },
             "expandOpt": "EDGE"
@@ -62,7 +65,7 @@
         },
         "hopRange": {
           "lower": 1,
-          "upper": 3
+          "upper": 4
         },
         "pathOpt": "SIMPLE",
         "resultOpt": "ALL_V"
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
index 9eb9011fecfc..3805b180f88d 100644
--- 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
@@ -52,7 +52,7 @@
         },
         "hopRange": {
           "lower": 1,
-          "upper": 3
+          "upper": 4
         },
         "pathOpt": "SIMPLE",
         "resultOpt": "ALL_V"
diff --git a/interactive_engine/compiler/src/test/resources/proto/project_test_2.json b/interactive_engine/compiler/src/test/resources/proto/project_test_2.json
new file mode 100644
index 000000000000..63d2a05c8233
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/proto/project_test_2.json
@@ -0,0 +1,147 @@
+{
+  "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
+          }],
+          "columns": [{
+            "name": "weight"
+          }],
+          "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": {
+      "project": {
+        "mappings": [{
+          "expr": {
+            "operators": [{
+              "var": {
+                "tag": {
+                  "id": 0
+                },
+                "property": {
+                  "key": {
+                    "name": "name"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "STRING"
+                }
+              },
+              "nodeType": {
+                "dataType": "STRING"
+              }
+            }]
+          },
+          "alias": 2
+        }, {
+          "expr": {
+            "operators": [{
+              "var": {
+                "tag": {
+                  "id": 1
+                },
+                "property": {
+                  "key": {
+                    "name": "weight"
+                  }
+                },
+                "nodeType": {
+                  "dataType": "DOUBLE"
+                }
+              },
+              "nodeType": {
+                "dataType": "DOUBLE"
+              }
+            }]
+          },
+          "alias": 3
+        }]
+      }
+    },
+    "metaData": [{
+      "type": {
+        "dataType": "STRING"
+      },
+      "alias": 2
+    }, {
+      "type": {
+        "dataType": "DOUBLE"
+      },
+      "alias": 3
+    }]
+  }, {
+    "opr": {
+      "sink": {
+        "sinkTarget": {
+          "sinkDefault": {
+          }
+        }
+      }
+    }
+  }]
+}
\ No newline at end of file