Repository: alibaba/GraphScope (language: python)

Issue:
Support to fetch properties of each elements in path in GOpt
**Is your feature request related to a problem? Please describe.**

As titled. In the new GOpt-based GIE, we need to support projecting  properties of each elements in path in GOpt. e.g.,
```
 g.V().both("1..3","knows").with('RESULT_OPT', 'ALL_V').values("id")
 g.V().both("1..3","knows").with('RESULT_OPT', 'ALL_V_E').values("id") // in which, "id" should exisit in both vertex's and edge's property schema
 g.V().both("1..3","knows").with('RESULT_OPT', 'ALL_V').valueMap("id")
 g.V().both("1..3","knows").with('RESULT_OPT', 'ALL_V_E').valueMap("id")
```

Support processing `None` value in GIE
**Is your feature request related to a problem? Please describe.**

As titled.



Discussion / hints:

Fix patch:
diff --git a/docs/interactive_engine/tinkerpop/supported_gremlin_steps.md b/docs/interactive_engine/tinkerpop/supported_gremlin_steps.md
index e20a27183e3e..6306b25e432f 100644
--- a/docs/interactive_engine/tinkerpop/supported_gremlin_steps.md
+++ b/docs/interactive_engine/tinkerpop/supported_gremlin_steps.md
@@ -692,6 +692,36 @@ g.V().out("1..10").with('RESULT_OPT', 'ALL_V')
 # unfold vertices in the path collection
 g.V().out("1..10").with('RESULT_OPT', 'ALL_V').endV()
 ```
+#### Getting Properites
+The properties of the elements (vertices and/or edges) in the path can be projected by `values()`-step or `valueMap()`-step.
+It is important to note that the specific elements targeted for property projection are determined by the `RESULT_OPT` setting. 
+For instance, if you configure `RESULT_OPT` as `ALL_V`,  `values()` or `valueMap()` will then project the properties of all vertices present in the path.
+```bash
+# get properties of each vertex in the path
+gremlin> g.V().both("1..3","knows").with('RESULT_OPT', 'ALL_V').values("name")
+==>[vadas, marko]
+==>[josh, marko]
+==>[marko, vadas]
+==>[marko, josh]
+==>[vadas, marko, vadas]
+==>[vadas, marko, josh]
+==>[josh, marko, vadas]
+==>[josh, marko, josh]
+==>[marko, vadas, marko]
+==>[marko, josh, marko]
+gremlin> g.V().both("1..3","knows").with('RESULT_OPT', 'ALL_V').valueMap("name","age")
+==>[{age=27, name=vadas}, {age=29, name=marko}]
+==>[{age=32, name=josh}, {age=29, name=marko}]
+==>[{age=29, name=marko}, {age=27, name=vadas}, {age=29, name=marko}]
+==>[{age=29, name=marko}, {age=32, name=josh}, {age=29, name=marko}]
+==>[{age=29, name=marko}, {age=27, name=vadas}]
+==>[{age=29, name=marko}, {age=32, name=josh}]
+==>[{age=27, name=vadas}, {age=29, name=marko}, {age=27, name=vadas}]
+==>[{age=27, name=vadas}, {age=29, name=marko}, {age=32, name=josh}]
+==>[{age=32, name=josh}, {age=29, name=marko}, {age=27, name=vadas}]
+==>[{age=32, name=josh}, {age=29, name=marko}, {age=32, name=josh}]
+```
+
 ### Expression
 
 Expressions, expressed via the `expr()` syntactic sugar, have been introduced to facilitate writing expressions directly within steps such as `select()`, `project()`, `where()`, and `group()`. This update is part of an ongoing effort to standardize Gremlin's expression syntax, making it more aligned with [SQL expression syntax](https://www.w3schools.com/sql/sql_operators.asp). The updated syntax, effective from version 0.27.0, streamlines user operations and enhances readability. Below, we detail the updated syntax definitions and point out key distinctions from the syntax used prior to version 0.26.0.
diff --git a/flex/engines/hqps_db/core/operator/sink.h b/flex/engines/hqps_db/core/operator/sink.h
index f935cdf71756..c852c9b7308d 100644
--- a/flex/engines/hqps_db/core/operator/sink.h
+++ b/flex/engines/hqps_db/core/operator/sink.h
@@ -234,7 +234,7 @@ void template_set_key_value(results::KeyValues* map,
   for (auto& kv : key_value) {
     auto cur_kv = map->add_key_values();
     cur_kv->mutable_key()->set_str(kv.first);
-    auto value = cur_kv->mutable_element();
+    auto value = cur_kv->mutable_value()->mutable_element();
     set_any_to_element(kv.second, value);
   }
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GraphIOProcessor.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GraphIOProcessor.java
index be525e0e8211..4b4d07cc6600 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GraphIOProcessor.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GraphIOProcessor.java
@@ -363,7 +363,9 @@ private PatternEdge visitAndAddPxdEdge(
                                     new ElementDetails(
                                             expandEdge.getElementDetails().getSelectivity(),
                                             new PathExpandRange(offset, fetch),
-                                            innerGetVTypes);
+                                            innerGetVTypes,
+                                            pxd.getResultOpt(),
+                                            pxd.getPathOpt());
                             expandEdge =
                                     (expandEdge instanceof SinglePatternEdge)
                                             ? new SinglePatternEdge(
@@ -825,10 +827,12 @@ private RelNode createExpandGetV(
                                                 edge.getElementDetails()),
                                         true),
                                 getVConfig.getAlias());
+                GraphOpt.PathExpandResult resultOpt = edge.getElementDetails().getResultOpt();
+                GraphOpt.PathExpandPath pathOpt = edge.getElementDetails().getPathOpt();
                 pxdBuilder
                         .getV(innerGetVConfig)
-                        .resultOpt(GraphOpt.PathExpandResult.END_V)
-                        .pathOpt(GraphOpt.PathExpandPath.ARBITRARY)
+                        .resultOpt(resultOpt == null ? GraphOpt.PathExpandResult.END_V : resultOpt)
+                        .pathOpt(pathOpt == null ? GraphOpt.PathExpandPath.ARBITRARY : pathOpt)
                         .alias(edgeValue.getAlias())
                         .startAlias(srcValue.getAlias())
                         .range(
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/ExpandGetVFusionRule.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/ExpandGetVFusionRule.java
index 551492ce773c..7967cfb30477 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/ExpandGetVFusionRule.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/ExpandGetVFusionRule.java
@@ -103,9 +103,12 @@ protected RelNode transform(GraphLogicalGetV getV, GraphLogicalExpand expand, Re
     }
 
     private boolean canFuse(GraphLogicalGetV getV, GraphLogicalExpand expand) {
-        // If GraphLogicalGetV has filters, then we cannot fuse them directly. Instead, we create a
+        // If GraphLogicalGetV has filters (or need to cache properties), then we cannot fuse them
+        // directly. Instead, we create a
         // EdgeExpand(V) with an Auxilia for the filters.
-        // If GraphLogicalGetV has no filters, then:
+        // todo: currently, we have not considered the property cache case yet. We will add this
+        // after FieldTrimmer is ready.
+        // If GraphLogicalGetV has no filters (or need to cache properties), then:
         // 1. if we want to expand "person-knows->person", and the type in getV is "person",
         // we can fuse them into one EdgeExpand with type "knows" (where "knows" will be given in
         // EdgeExpand's QueryParam in PhysicalPlan)
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/JoinDecompositionRule.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/JoinDecompositionRule.java
index 1e02d5041cf2..217bd91f4ae3 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/JoinDecompositionRule.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/JoinDecompositionRule.java
@@ -383,7 +383,9 @@ private PatternEdge createNewEdge(
                 new ElementDetails(
                         oldEdge.getElementDetails().getSelectivity(),
                         newRange,
-                        oldEdge.getElementDetails().getPxdInnerGetVTypes());
+                        oldEdge.getElementDetails().getPxdInnerGetVTypes(),
+                        oldEdge.getElementDetails().getResultOpt(),
+                        oldEdge.getElementDetails().getPathOpt());
         return (oldEdge instanceof SinglePatternEdge)
                 ? new SinglePatternEdge(
                         newSrc,
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java
index f7deb4f67bba..869eb27ded0d 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java
@@ -92,14 +92,7 @@ protected GraphLogicalPathExpand(
         this.aliasId = cluster.getIdGenerator().generate(this.aliasName);
         this.startAlias = Objects.requireNonNull(startAlias);
         this.optional = optional;
-        // Here, we set the result opt in a direct manner, where the opt represents what sort of
-        // vertex and edge information should be saved for the path.
-        // if the alias exists, then result opt is set to 'ALL_V_E' to cache all inner vertices or
-        // edges, otherwise set to 'END_V' to reduce the memory consumption.
-        this.resultOpt =
-                (this.aliasId == AliasInference.DEFAULT_ID)
-                        ? resultOpt
-                        : GraphOpt.PathExpandResult.ALL_V_E;
+        this.resultOpt = resultOpt;
         this.untilCondition = untilCondition;
     }
 
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/metadata/glogue/pattern/ElementDetails.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/metadata/glogue/pattern/ElementDetails.java
index d219ddd8e5bc..ac686dceb9a0 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/metadata/glogue/pattern/ElementDetails.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/metadata/glogue/pattern/ElementDetails.java
@@ -16,6 +16,7 @@
 
 package com.alibaba.graphscope.common.ir.rel.metadata.glogue.pattern;
 
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
 import com.google.common.collect.ImmutableList;
 
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -30,6 +31,8 @@ public class ElementDetails implements Comparable<ElementDetails> {
     private final @Nullable PathExpandRange range;
     // record inner getV types of path expand
     private final List<Integer> pxdInnerGetVTypes;
+    private final GraphOpt.PathExpandResult resultOpt;
+    private final GraphOpt.PathExpandPath pathOpt;
     private boolean optional;
 
     public ElementDetails() {
@@ -37,27 +40,35 @@ public ElementDetails() {
     }
 
     public ElementDetails(double selectivity) {
-        this(selectivity, null, ImmutableList.of());
+        this(selectivity, null, ImmutableList.of(), null, null);
     }
 
     public ElementDetails(
-            double selectivity, @Nullable PathExpandRange range, List<Integer> pxdInnerGetVTypes) {
-        this(selectivity, range, pxdInnerGetVTypes, false);
+            double selectivity,
+            @Nullable PathExpandRange range,
+            List<Integer> pxdInnerGetVTypes,
+            GraphOpt.PathExpandResult resultOpt,
+            GraphOpt.PathExpandPath pathOpt) {
+        this(selectivity, range, pxdInnerGetVTypes, resultOpt, pathOpt, false);
     }
 
     public ElementDetails(double selectivity, boolean optional) {
-        this(selectivity, null, ImmutableList.of(), optional);
+        this(selectivity, null, ImmutableList.of(), null, null, optional);
     }
 
     public ElementDetails(
             double selectivity,
             @Nullable PathExpandRange range,
             List<Integer> pxdInnerVertexTypes,
+            GraphOpt.PathExpandResult resultOpt,
+            GraphOpt.PathExpandPath pathOpt,
             boolean optional) {
         this.selectivity = selectivity;
         this.range = range;
         this.pxdInnerGetVTypes = pxdInnerVertexTypes;
         this.optional = optional;
+        this.resultOpt = resultOpt;
+        this.pathOpt = pathOpt;
     }
 
     @Override
@@ -96,6 +107,14 @@ public List<Integer> getPxdInnerGetVTypes() {
         return Collections.unmodifiableList(this.pxdInnerGetVTypes);
     }
 
+    public GraphOpt.PathExpandResult getResultOpt() {
+        return resultOpt;
+    }
+
+    public GraphOpt.PathExpandPath getPathOpt() {
+        return pathOpt;
+    }
+
     @Override
     public int compareTo(ElementDetails o) {
         int compare = Double.compare(this.selectivity, o.selectivity);
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 a642987e10cc..9dc5490701fb 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
@@ -218,18 +218,21 @@ public RelNode visit(GraphLogicalPathExpand pxd) {
         GraphAlgebraPhysical.PathExpand.ExpandBase.Builder expandBaseBuilder =
                 GraphAlgebraPhysical.PathExpand.ExpandBase.newBuilder();
         RelNode fused = pxd.getFused();
+        GraphAlgebraPhysical.EdgeExpand.Builder edgeExpandBuilder;
+        GraphAlgebraPhysical.GetV.Builder getVBuilder = null;
+        // preserve the original fused logical getv to cache properties if necessary
+        GraphLogicalGetV originalGetV;
         RelDataType rowType;
         if (fused != null) {
             // the case that expand base is fused
             if (fused instanceof GraphPhysicalGetV) {
                 // fused into expand + auxilia
                 GraphPhysicalGetV fusedGetV = (GraphPhysicalGetV) fused;
-                GraphAlgebraPhysical.GetV.Builder auxilia = buildAuxilia(fusedGetV);
-                expandBaseBuilder.setGetV(auxilia);
+                getVBuilder = buildAuxilia(fusedGetV);
+                originalGetV = fusedGetV.getFusedGetV();
                 if (fusedGetV.getInput() instanceof GraphPhysicalExpand) {
                     GraphPhysicalExpand fusedExpand = (GraphPhysicalExpand) fusedGetV.getInput();
-                    GraphAlgebraPhysical.EdgeExpand.Builder expand = buildEdgeExpand(fusedExpand);
-                    expandBaseBuilder.setEdgeExpand(expand);
+                    edgeExpandBuilder = buildEdgeExpand(fusedExpand);
                     rowType = fusedExpand.getRowType();
                 } else {
                     throw new UnsupportedOperationException(
@@ -239,8 +242,8 @@ public RelNode visit(GraphLogicalPathExpand pxd) {
             } else if (fused instanceof GraphPhysicalExpand) {
                 // fused into expand
                 GraphPhysicalExpand fusedExpand = (GraphPhysicalExpand) fused;
-                GraphAlgebraPhysical.EdgeExpand.Builder expand = buildEdgeExpand(fusedExpand);
-                expandBaseBuilder.setEdgeExpand(expand);
+                edgeExpandBuilder = buildEdgeExpand(fusedExpand);
+                originalGetV = fusedExpand.getFusedGetV();
                 rowType = fusedExpand.getFusedExpand().getRowType();
             } else {
                 throw new UnsupportedOperationException(
@@ -248,13 +251,55 @@ public RelNode visit(GraphLogicalPathExpand pxd) {
             }
         } else {
             // the case that expand base is not fused
-            GraphAlgebraPhysical.EdgeExpand.Builder expand =
-                    buildEdgeExpand((GraphLogicalExpand) pxd.getExpand());
-            GraphAlgebraPhysical.GetV.Builder getV = buildGetV((GraphLogicalGetV) pxd.getGetV());
-            expandBaseBuilder.setEdgeExpand(expand);
-            expandBaseBuilder.setGetV(getV);
+            edgeExpandBuilder = buildEdgeExpand((GraphLogicalExpand) pxd.getExpand());
+            originalGetV = (GraphLogicalGetV) pxd.getGetV();
+            getVBuilder = buildGetV(originalGetV);
             rowType = pxd.getExpand().getRowType();
         }
+        expandBaseBuilder.setEdgeExpand(edgeExpandBuilder);
+        // specifically, for path expand (except that it has PathExpandResult as END_V), we need to
+        // cache properties for the expanded vertices if necessary, instead of fetch them lazily
+        Set<GraphNameOrId> vertexColumns =
+                Utils.extractColumnsFromRelDataType(originalGetV.getRowType(), isColumnId);
+        if (isPartitioned
+                && !vertexColumns.isEmpty()
+                && pxd.getResultOpt() != GraphOpt.PathExpandResult.END_V) {
+            // 1. build an auxilia to cache necessary properties for the start vertex of the path
+            GraphAlgebraPhysical.GetV.Builder startAuxiliaBuilder =
+                    GraphAlgebraPhysical.GetV.newBuilder();
+            startAuxiliaBuilder.setOpt(Utils.protoGetVOpt(PhysicalGetVOpt.ITSELF));
+            GraphAlgebra.QueryParams.Builder startParamsBuilder = defaultQueryParams();
+            addQueryColumns(startParamsBuilder, vertexColumns);
+            startAuxiliaBuilder.setParams(startParamsBuilder);
+            if (pxd.getStartAlias().getAliasId() != AliasInference.DEFAULT_ID) {
+                startAuxiliaBuilder.setTag(Utils.asAliasId(pxd.getStartAlias().getAliasId()));
+                startAuxiliaBuilder.setAlias(Utils.asAliasId(pxd.getStartAlias().getAliasId()));
+            }
+            GraphAlgebraPhysical.PhysicalOpr.Builder startAuxiliaOprBuilder =
+                    GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+            startAuxiliaOprBuilder.setOpr(
+                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
+                            .setVertex(startAuxiliaBuilder));
+            startAuxiliaOprBuilder.addAllMetaData(
+                    Utils.physicalProtoRowType(originalGetV.getRowType(), isColumnId));
+            addRepartitionToAnother(pxd.getStartAlias().getAliasId());
+            physicalBuilder.addPlan(startAuxiliaOprBuilder.build());
+            // 2. build an auxilia to cache necessary properties for the expanded vertices during
+            // the path, in expand base
+            if (getVBuilder == null) {
+                // getVBuilder can be null if the expand base is ExpandV
+                // then create a new GetV to cache properties
+                getVBuilder = buildVertex(originalGetV, PhysicalGetVOpt.ITSELF);
+            }
+            GraphAlgebra.QueryParams.Builder paramsBuilder = getVBuilder.getParamsBuilder();
+            addQueryColumns(paramsBuilder, vertexColumns);
+            getVBuilder.setParams(paramsBuilder);
+            expandBaseBuilder.setGetV(getVBuilder);
+        } else {
+            if (getVBuilder != null) {
+                expandBaseBuilder.setGetV(getVBuilder);
+            }
+        }
         pathExpandBuilder.setBase(expandBaseBuilder);
         pathExpandBuilder.setPathOpt(Utils.protoPathOpt(pxd.getPathOpt()));
         pathExpandBuilder.setResultOpt(Utils.protoPathResultOpt(pxd.getResultOpt()));
@@ -1063,22 +1108,6 @@ private void addAuxilia(
         physicalBuilder.addPlan(auxiliaOprBuilder.build());
     }
 
-    private void addAuxilia(
-            GraphAlgebraPhysical.PhysicalPlan.Builder physicalBuilder,
-            GraphPhysicalGetV physicalGetV) {
-        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
-                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
-        GraphAlgebraPhysical.GetV.Builder auxilia = buildAuxilia(physicalGetV);
-        oprBuilder.setOpr(
-                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setVertex(auxilia));
-        oprBuilder.addAllMetaData(
-                Utils.physicalProtoRowType(physicalGetV.getRowType(), isColumnId));
-        if (isPartitioned) {
-            addRepartitionToAnother(physicalGetV.getStartAlias().getAliasId());
-        }
-        physicalBuilder.addPlan(oprBuilder.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);
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/RexToProtoConverter.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/RexToProtoConverter.java
index bddfc23f9ca4..92f505312157 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/RexToProtoConverter.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/RexToProtoConverter.java
@@ -62,6 +62,9 @@ public OuterExpression.Expression visitCall(RexCall call) {
         } else if (operator.getKind() == SqlKind.OTHER
                 && operator.getName().equals("PATH_CONCAT")) {
             return visitPathConcat(call);
+        } else if (operator.getKind() == SqlKind.OTHER
+                && operator.getName().equals("PATH_FUNCTION")) {
+            return visitPathFunction(call);
         } else if (operator.getKind() == SqlKind.EXTRACT) {
             return visitExtract(call);
         } else if (operator.getKind() == SqlKind.OTHER
@@ -74,6 +77,64 @@ public OuterExpression.Expression visitCall(RexCall call) {
         }
     }
 
+    private OuterExpression.Expression visitPathFunction(RexCall call) {
+        List<RexNode> operands = call.getOperands();
+        RexGraphVariable variable = (RexGraphVariable) operands.get(0);
+        OuterExpression.PathFunction.Builder builder = OuterExpression.PathFunction.newBuilder();
+        if (variable.getAliasId() != AliasInference.DEFAULT_ID) {
+            builder.setTag(Common.NameOrId.newBuilder().setId(variable.getAliasId()).build());
+        }
+        GraphOpt.PathExpandFunction funcOpt =
+                ((RexLiteral) operands.get(1)).getValueAs(GraphOpt.PathExpandFunction.class);
+        builder.setOpt(OuterExpression.PathFunction.FuncOpt.valueOf(funcOpt.name()));
+        setPathFunctionProjection(builder, operands.get(2));
+        return OuterExpression.Expression.newBuilder()
+                .addOperators(
+                        OuterExpression.ExprOpr.newBuilder()
+                                .setPathFunc(builder)
+                                .setNodeType(Utils.protoIrDataType(call.getType(), isColumnId)))
+                .build();
+    }
+
+    private void setPathFunctionProjection(
+            OuterExpression.PathFunction.Builder builder, RexNode projection) {
+        switch (projection.getKind()) {
+            case MAP_VALUE_CONSTRUCTOR:
+                List<RexNode> operands = ((RexCall) projection).getOperands();
+                OuterExpression.PathFunction.PathElementKeyValues.Builder keyValuesBuilder =
+                        OuterExpression.PathFunction.PathElementKeyValues.newBuilder();
+                for (int i = 0; i < operands.size(); i += 2) {
+                    keyValuesBuilder.addKeyVals(
+                            OuterExpression.PathFunction.PathElementKeyValues.PathElementKeyValue
+                                    .newBuilder()
+                                    .setKey(operands.get(i).accept(this).getOperators(0).getConst())
+                                    .setVal(visitPathFunctionProperty(operands.get(i + 1))));
+                }
+                builder.setMap(keyValuesBuilder);
+                break;
+            case ARRAY_VALUE_CONSTRUCTOR:
+                OuterExpression.PathFunction.PathElementKeys.Builder valuesBuilder =
+                        OuterExpression.PathFunction.PathElementKeys.newBuilder();
+                ((RexCall) projection)
+                        .getOperands()
+                        .forEach(
+                                operand ->
+                                        valuesBuilder.addKeys(visitPathFunctionProperty(operand)));
+                builder.setVars(valuesBuilder);
+                break;
+            default:
+                builder.setProperty(visitPathFunctionProperty(projection));
+        }
+    }
+
+    private OuterExpression.Property visitPathFunctionProperty(RexNode variable) {
+        Preconditions.checkArgument(
+                variable instanceof RexGraphVariable
+                        && ((RexGraphVariable) variable).getProperty() != null,
+                "cannot get path function property from variable [" + variable + "]");
+        return Utils.protoProperty(((RexGraphVariable) variable).getProperty());
+    }
+
     private OuterExpression.Expression visitPathConcat(RexCall call) {
         List<RexNode> operands = call.getOperands();
         return OuterExpression.Expression.newBuilder()
@@ -210,6 +271,9 @@ private OuterExpression.Expression visitMapValueConstructor(RexCall call) {
                 case MAP:
                     keyValueBuilder.setNested(valueOpr.getMap());
                     break;
+                case PATH_FUNC:
+                    keyValueBuilder.setPathFunc(valueOpr.getPathFunc());
+                    break;
                 default:
                     throw new IllegalArgumentException(
                             "can not convert value ["
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java
index 2f16b86c4264..b9d1c1ba37d1 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java
@@ -855,7 +855,8 @@ private boolean isCurrentSupported(SqlOperator operator) {
                 || (sqlKind == SqlKind.OTHER
                         && (operator.getName().equals("IN")
                                 || operator.getName().equals("DATETIME_MINUS")
-                                || operator.getName().equals("PATH_CONCAT")))
+                                || operator.getName().equals("PATH_CONCAT")
+                                || operator.getName().equals("PATH_FUNCTION")))
                 || sqlKind == SqlKind.ARRAY_CONCAT;
     }
 
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphStdOperatorTable.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphStdOperatorTable.java
index 1d8d64ba79e7..ec7277105c96 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphStdOperatorTable.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphStdOperatorTable.java
@@ -321,4 +321,18 @@ public static final SqlFunction USER_DEFINED_PROCEDURE(StoredProcedureMeta meta)
                                             .get(i),
                             i -> false),
                     SqlFunctionCategory.SYSTEM);
+
+    public static final SqlOperator PATH_FUNCTION =
+            new SqlFunction(
+                    "PATH_FUNCTION",
+                    SqlKind.OTHER,
+                    ReturnTypes.ARG2.andThen(SqlTypeTransforms.TO_ARRAY),
+                    null,
+                    GraphOperandTypes.operandMetadata(
+                            ImmutableList.of(
+                                    GraphTypeFamily.PATH, SqlTypeFamily.IGNORE, SqlTypeFamily.ANY),
+                            typeFactory -> ImmutableList.of(),
+                            i -> ImmutableList.of("Path", "FuncOpt", "PropertyProjection").get(i),
+                            i -> false),
+                    SqlFunctionCategory.SYSTEM);
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/config/GraphOpt.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/config/GraphOpt.java
index 36c8e5926097..27f464dc85f1 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/config/GraphOpt.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/config/GraphOpt.java
@@ -67,4 +67,10 @@ public enum PathExpandResult {
         ALL_V,
         ALL_V_E
     }
+
+    public enum PathExpandFunction {
+        VERTEX,
+        EDGE,
+        VERTEX_EDGE
+    }
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/antlr4/visitor/PathExpandBuilderVisitor.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/antlr4/visitor/PathExpandBuilderVisitor.java
index 4f2c376f7764..bfb91f3fce51 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/antlr4/visitor/PathExpandBuilderVisitor.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/antlr4/visitor/PathExpandBuilderVisitor.java
@@ -16,10 +16,7 @@
 
 package com.alibaba.graphscope.cypher.antlr4.visitor;
 
-import com.alibaba.graphscope.common.ir.tools.config.ExpandConfig;
-import com.alibaba.graphscope.common.ir.tools.config.GetVConfig;
-import com.alibaba.graphscope.common.ir.tools.config.LabelConfig;
-import com.alibaba.graphscope.common.ir.tools.config.PathExpandConfig;
+import com.alibaba.graphscope.common.ir.tools.config.*;
 import com.alibaba.graphscope.grammar.CypherGSBaseVisitor;
 import com.alibaba.graphscope.grammar.CypherGSParser;
 
@@ -46,6 +43,12 @@ public PathExpandConfig.Builder visitOC_RelationshipPattern(
         // set path expand alias
         if (expandConfig.getAlias() != null) {
             builder.alias(expandConfig.getAlias());
+            // Here, we set the result opt in a direct manner, where the opt represents what sort of
+            // vertex and edge information should be saved for the path.
+            // if the alias exists, then result opt is set to 'ALL_V_E' to cache all inner vertices
+            // or
+            // edges, otherwise keep the default value 'END_V' to reduce the memory consumption.
+            builder.resultOpt(GraphOpt.PathExpandResult.ALL_V_E);
         }
         // set path expand hops
         return visitOC_RangeLiteral(ctx.oC_RelationshipDetail().oC_RangeLiteral());
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/result/CypherRecordParser.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/result/CypherRecordParser.java
index ae2b90497d86..706019e0bb57 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/result/CypherRecordParser.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/result/CypherRecordParser.java
@@ -24,7 +24,6 @@
 import com.alibaba.graphscope.common.result.Utils;
 import com.alibaba.graphscope.gaia.proto.Common;
 import com.alibaba.graphscope.gaia.proto.IrResult;
-import com.alibaba.graphscope.gremlin.exception.GremlinResultParserException;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -120,30 +119,6 @@ protected AnyValue parseEntry(IrResult.Entry entry, @Nullable RelDataType dataTy
         }
     }
 
-    protected AnyValue parseElement(IrResult.KeyValues.KeyValue value, RelDataType type) {
-        switch (value.getValueCase()) {
-            case ELEMENT:
-                return parseElement(value.getElement(), type);
-            case NESTED:
-                if (type instanceof ArbitraryMapType) {
-                    return parseKeyValues(
-                            value.getNested(),
-                            ((ArbitraryMapType) type).getKeyTypes(),
-                            ((ArbitraryMapType) type).getValueTypes());
-                } else {
-                    return parseKeyValues(
-                            value.getNested(), type.getKeyType(), type.getValueType());
-                }
-            default:
-                throw new GremlinResultParserException(
-                        "keyValue ["
-                                + value
-                                + "] has invalid value type ["
-                                + value.getValueCase()
-                                + "]");
-        }
-    }
-
     protected AnyValue parseElement(IrResult.Element element, @Nullable RelDataType dataType) {
         switch (element.getInnerCase()) {
             case VERTEX:
@@ -220,7 +195,9 @@ protected AnyValue parseKeyValues(
                 .getKeyValuesList()
                 .forEach(
                         entry -> {
-                            valueMap.put(entry.getKey().getStr(), parseElement(entry, valueType));
+                            valueMap.put(
+                                    entry.getKey().getStr(),
+                                    parseEntry(entry.getValue(), valueType));
                         });
         return VirtualValues.fromMap(valueMap, valueMap.size(), 0);
     }
@@ -239,7 +216,7 @@ protected AnyValue parseKeyValues(
         Map<String, AnyValue> valueMap = Maps.newLinkedHashMap();
         for (int i = 0; i < entries.size(); ++i) {
             IrResult.KeyValues.KeyValue entry = entries.get(i);
-            valueMap.put(entry.getKey().getStr(), parseElement(entry, valueTypes.get(i)));
+            valueMap.put(entry.getKey().getStr(), parseEntry(entry.getValue(), valueTypes.get(i)));
         }
         return VirtualValues.fromMap(valueMap, valueMap.size(), 0);
     }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/ExpressionVisitor.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/ExpressionVisitor.java
index 2fb40abb31bf..9c003e03c1f7 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/ExpressionVisitor.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/ExpressionVisitor.java
@@ -16,28 +16,171 @@
 
 package com.alibaba.graphscope.gremlin.antlr4x.visitor;
 
+import com.alibaba.graphscope.common.ir.rex.RexGraphVariable;
 import com.alibaba.graphscope.common.ir.tools.GraphBuilder;
 import com.alibaba.graphscope.common.ir.tools.GraphStdOperatorTable;
+import com.alibaba.graphscope.common.ir.type.GraphPathType;
+import com.alibaba.graphscope.common.ir.type.GraphProperty;
+import com.alibaba.graphscope.common.ir.type.GraphSchemaType;
 import com.alibaba.graphscope.grammar.GremlinGSBaseVisitor;
 import com.alibaba.graphscope.grammar.GremlinGSParser;
+import com.alibaba.graphscope.gremlin.antlr4.TraversalEnumParser;
 import com.alibaba.graphscope.gremlin.exception.UnsupportedEvalException;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
 
+import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rex.RexNode;
+import org.apache.tinkerpop.gremlin.structure.T;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.math.BigDecimal;
 import java.util.List;
 import java.util.Objects;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public class ExpressionVisitor extends GremlinGSBaseVisitor<RexNode> {
     protected final RexNode propertyKey;
     protected final GraphBuilder builder;
+    // indicate whether to throw exception when property not found in the input expand or getV
+    protected final boolean throwsOnPropertyNotFound;
 
     public static final int SIMPLE_PREDICATE_CHILD_COUNT = 1, CONNECTIVE_PREDICATE_CHILD_COUNT = 6;
 
     public ExpressionVisitor(GraphBuilder builder, RexNode propertyKey) {
+        this(builder, propertyKey, true);
+    }
+
+    public ExpressionVisitor(
+            GraphBuilder builder, RexNode propertyKey, boolean throwsOnPropertyNotFound) {
         this.builder = Objects.requireNonNull(builder);
         this.propertyKey = Objects.requireNonNull(propertyKey);
+        this.throwsOnPropertyNotFound = throwsOnPropertyNotFound;
+    }
+
+    @Override
+    public RexNode visitTraversalMethod_valueMap(
+            GremlinGSParser.TraversalMethod_valueMapContext ctx) {
+        List<String> ctxProperties =
+                new LiteralList(ctx.oC_ListLiteral(), ctx.oC_Expression()).toList(String.class);
+        if (isPathFunction(ctxProperties)) {
+            return (new PathFunctionVisitor(builder, propertyKey))
+                    .visitTraversalMethod_valueMap(ctx);
+        }
+        String tag = getPropertyTag();
+        return builder.call(
+                GraphStdOperatorTable.MAP_VALUE_CONSTRUCTOR,
+                getProperties(ctxProperties, tag).stream()
+                        .flatMap(k -> Stream.of(builder.literal(k), builder.variable(tag, k)))
+                        .collect(Collectors.toList()));
+    }
+
+    @Override
+    public RexNode visitTraversalMethod_values(GremlinGSParser.TraversalMethod_valuesContext ctx) {
+        List<String> ctxProperties =
+                Lists.newArrayList((String) LiteralVisitor.INSTANCE.visit(ctx.StringLiteral()));
+        if (isPathFunction(ctxProperties)) {
+            return (new PathFunctionVisitor(builder, propertyKey)).visitTraversalMethod_values(ctx);
+        }
+        String tag = getPropertyTag();
+        List<String> properties = getProperties(ctxProperties, tag);
+        return properties.isEmpty() ? null : builder.variable(tag, properties.get(0));
+    }
+
+    @Override
+    public RexNode visitTraversalMethod_elementMap(
+            GremlinGSParser.TraversalMethod_elementMapContext ctx) {
+        List<String> ctxProperties =
+                new LiteralList(ctx.oC_ListLiteral(), ctx.oC_Expression()).toList(String.class);
+        if (isPathFunction(ctxProperties)) {
+            return (new PathFunctionVisitor(builder, propertyKey))
+                    .visitTraversalMethod_elementMap(ctx);
+        }
+        String tag = getPropertyTag();
+        return builder.call(
+                GraphStdOperatorTable.MAP_VALUE_CONSTRUCTOR,
+                getElementMapProperties(getProperties(ctxProperties, tag)).stream()
+                        .flatMap(k -> Stream.of(builder.literal(k), builder.variable(tag, k)))
+                        .collect(Collectors.toList()));
+    }
+
+    @Override
+    public RexNode visitTraversalMethod_selectby(
+            GremlinGSParser.TraversalMethod_selectbyContext byCtx) {
+        int byChildCount = byCtx.getChildCount();
+        if (byChildCount == 3) { // select(..).by()
+            return propertyKey;
+        }
+        String tag = getPropertyTag();
+        if (byChildCount == 4 && byCtx.StringLiteral() != null) { // select(..).by('name')
+            List<String> ctxProperties =
+                    Lists.newArrayList(
+                            (String) LiteralVisitor.INSTANCE.visit(byCtx.StringLiteral()));
+            if (isPathFunction(ctxProperties)) {
+                return (new PathFunctionVisitor(builder, propertyKey))
+                        .visitTraversalMethod_selectby(byCtx);
+            }
+            List<String> properties = getProperties(ctxProperties, tag);
+            return properties.isEmpty() ? null : builder.variable(tag, properties.get(0));
+        } else if (byChildCount == 4
+                && byCtx.traversalToken() != null) { // select(..).by(T.label/T.id)
+            T token =
+                    TraversalEnumParser.parseTraversalEnumFromContext(
+                            T.class, byCtx.traversalToken());
+            return builder.variable(tag, token.getAccessor());
+        } else if (byCtx.traversalMethod_valueMap() != null) { // select(..).by(valueMap('name'))
+            return new ExpressionVisitor(builder, propertyKey)
+                    .visitTraversalMethod_valueMap(byCtx.traversalMethod_valueMap());
+        } else if (byCtx.traversalMethod_elementMap()
+                != null) { // select(..).by(elementMap('name'))
+            return new ExpressionVisitor(builder, propertyKey)
+                    .visitTraversalMethod_elementMap(byCtx.traversalMethod_elementMap());
+        } else if (byCtx.nestedTraversal() != null) {
+            return Utils.convertExprToPair(
+                            (new NestedTraversalRexVisitor(this.builder, tag, byCtx))
+                                    .visitNestedTraversal(byCtx.nestedTraversal()))
+                    .getValue0();
+        }
+        throw new UnsupportedEvalException(
+                GremlinGSParser.TraversalMethod_selectbyContext.class,
+                byCtx.getText() + " is unsupported yet in select");
+    }
+
+    private boolean isPathFunction(List<String> ctxProperties) {
+        return (propertyKey.getType() instanceof GraphPathType)
+                && ctxProperties.stream().allMatch(k -> !k.equals(GraphProperty.LEN_KEY));
+    }
+
+    private List<String> getProperties(List<String> ctxProperties, @Nullable String tag) {
+        if (throwsOnPropertyNotFound || ctxProperties.isEmpty()) {
+            return ctxProperties.isEmpty() ? getAllProperties(tag) : ctxProperties;
+        }
+        ctxProperties.retainAll(getAllProperties(tag));
+        return ctxProperties;
+    }
+
+    private List<String> getElementMapProperties(List<String> properties) {
+        properties.add(0, GraphProperty.LABEL_KEY);
+        properties.add(1, GraphProperty.ID_KEY);
+        return properties;
+    }
+
+    private List<String> getAllProperties(@Nullable String tag) {
+        RexGraphVariable curVar = builder.variable(tag);
+        RelDataType dataType = curVar.getType();
+        Preconditions.checkArgument(
+                dataType instanceof GraphSchemaType, "can not get property from type=", dataType);
+        return dataType.getFieldList().stream().map(k -> k.getName()).collect(Collectors.toList());
+    }
+
+    private String getPropertyTag() {
+        Preconditions.checkArgument(
+                propertyKey instanceof RexGraphVariable
+                        && ((RexGraphVariable) propertyKey).getProperty() == null,
+                "variable: [%s] cannot denote a start tag",
+                propertyKey);
+        return ((RexGraphVariable) propertyKey).getName();
     }
 
     @Override
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/GraphBuilderVisitor.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/GraphBuilderVisitor.java
index a8fb65ab15d3..d45a9ab8e733 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/GraphBuilderVisitor.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/GraphBuilderVisitor.java
@@ -23,14 +23,12 @@
 import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalPathExpand;
 import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalSource;
 import com.alibaba.graphscope.common.ir.rel.type.group.GraphAggCall;
-import com.alibaba.graphscope.common.ir.rex.RexGraphVariable;
 import com.alibaba.graphscope.common.ir.rex.RexTmpVariableConverter;
 import com.alibaba.graphscope.common.ir.tools.AliasInference;
 import com.alibaba.graphscope.common.ir.tools.GraphBuilder;
 import com.alibaba.graphscope.common.ir.tools.GraphStdOperatorTable;
 import com.alibaba.graphscope.common.ir.tools.config.*;
 import com.alibaba.graphscope.common.ir.type.GraphProperty;
-import com.alibaba.graphscope.common.ir.type.GraphSchemaType;
 import com.alibaba.graphscope.grammar.GremlinGSBaseVisitor;
 import com.alibaba.graphscope.grammar.GremlinGSParser;
 import com.alibaba.graphscope.gremlin.antlr4.TraversalEnumParser;
@@ -46,7 +44,6 @@
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.Aggregate;
 import org.apache.calcite.rel.core.Project;
-import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
 import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexNode;
@@ -62,7 +59,6 @@
 import java.util.Objects;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 public class GraphBuilderVisitor extends GremlinGSBaseVisitor<GraphBuilder> {
     private final GraphBuilder builder;
@@ -409,15 +405,8 @@ public GraphBuilder visitTraversalMethod_as(GremlinGSParser.TraversalMethod_asCo
     public GraphBuilder visitTraversalMethod_valueMap(
             GremlinGSParser.TraversalMethod_valueMapContext ctx) {
         RexNode expr =
-                builder.call(
-                        GraphStdOperatorTable.MAP_VALUE_CONSTRUCTOR,
-                        getProperties(ctx, null).stream()
-                                .flatMap(
-                                        k ->
-                                                Stream.of(
-                                                        builder.literal(k),
-                                                        builder.variable(null, k)))
-                                .collect(Collectors.toList()));
+                new ExpressionVisitor(builder, builder.variable((String) null))
+                        .visitTraversalMethod_valueMap(ctx);
         return builder.project(ImmutableList.of(expr), ImmutableList.of(), true);
     }
 
@@ -426,8 +415,8 @@ public GraphBuilder visitTraversalMethod_values(
             GremlinGSParser.TraversalMethod_valuesContext ctx) {
         if (ctx.getChildCount() == 4 && ctx.StringLiteral() != null) {
             RexNode expr =
-                    builder.variable(
-                            null, (String) LiteralVisitor.INSTANCE.visit(ctx.StringLiteral()));
+                    new ExpressionVisitor(builder, builder.variable((String) null))
+                            .visitTraversalMethod_values(ctx);
             return builder.project(ImmutableList.of(expr), ImmutableList.of(), true);
         }
         throw new UnsupportedEvalException(ctx.getClass(), "supported pattern is [values('..')]");
@@ -437,15 +426,8 @@ public GraphBuilder visitTraversalMethod_values(
     public GraphBuilder visitTraversalMethod_elementMap(
             GremlinGSParser.TraversalMethod_elementMapContext ctx) {
         RexNode expr =
-                builder.call(
-                        GraphStdOperatorTable.MAP_VALUE_CONSTRUCTOR,
-                        getProperties(ctx, null).stream()
-                                .flatMap(
-                                        k ->
-                                                Stream.of(
-                                                        builder.literal(k),
-                                                        builder.variable(null, k)))
-                                .collect(Collectors.toList()));
+                new ExpressionVisitor(builder, builder.variable((String) null))
+                        .visitTraversalMethod_elementMap(ctx);
         return builder.project(ImmutableList.of(expr), ImmutableList.of(), true);
     }
 
@@ -931,40 +913,8 @@ private RexNode convertSelectByCtx(
             return builder.variable(tag);
         }
         GremlinGSParser.TraversalMethod_selectbyContext byCtx = byCtxs.get(i % ctxCnt);
-        int byChildCount = byCtx.getChildCount();
-        if (byChildCount == 3) { // select(..).by()
-            return builder.variable(tag);
-        } else if (byChildCount == 4 && byCtx.StringLiteral() != null) { // select(..).by('name')
-            return builder.variable(
-                    tag, (String) LiteralVisitor.INSTANCE.visit(byCtx.StringLiteral()));
-        } else if (byChildCount == 4
-                && byCtx.traversalToken() != null) { // select(..).by(T.label/T.id)
-            T token =
-                    TraversalEnumParser.parseTraversalEnumFromContext(
-                            T.class, byCtx.traversalToken());
-            return builder.variable(tag, token.getAccessor());
-        } else if (byCtx.traversalMethod_valueMap() != null) { // select(..).by(valueMap('name'))
-            return builder.call(
-                    GraphStdOperatorTable.MAP_VALUE_CONSTRUCTOR,
-                    getProperties(byCtx.traversalMethod_valueMap(), tag).stream()
-                            .flatMap(k -> Stream.of(builder.literal(k), builder.variable(tag, k)))
-                            .collect(Collectors.toList()));
-        } else if (byCtx.traversalMethod_elementMap()
-                != null) { // select(..).by(elementMap('name'))
-            return builder.call(
-                    GraphStdOperatorTable.MAP_VALUE_CONSTRUCTOR,
-                    getProperties(byCtx.traversalMethod_elementMap(), tag).stream()
-                            .flatMap(k -> Stream.of(builder.literal(k), builder.variable(tag, k)))
-                            .collect(Collectors.toList()));
-        } else if (byCtx.nestedTraversal() != null) {
-            return Utils.convertExprToPair(
-                            (new NestedTraversalRexVisitor(this.builder, tag, byCtx))
-                                    .visitNestedTraversal(byCtx.nestedTraversal()))
-                    .getValue0();
-        }
-        throw new UnsupportedEvalException(
-                GremlinGSParser.TraversalMethod_selectbyContext.class,
-                byCtx.getText() + " is unsupported yet in select");
+        return new ExpressionVisitor(builder, builder.variable(tag))
+                .visitTraversalMethod_selectby(byCtx);
     }
 
     private boolean pathExpandPattern(
@@ -991,33 +941,6 @@ private LabelConfig getLabelConfig(
         }
     }
 
-    private List<String> getProperties(
-            GremlinGSParser.TraversalMethod_valueMapContext ctx, @Nullable String tag) {
-        List<String> properties =
-                new LiteralList(ctx.oC_ListLiteral(), ctx.oC_Expression()).toList(String.class);
-        return properties.isEmpty() ? getAllProperties(tag) : properties;
-    }
-
-    private List<String> getProperties(
-            GremlinGSParser.TraversalMethod_elementMapContext ctx, @Nullable String tag) {
-        List<String> properties =
-                new LiteralList(ctx.oC_ListLiteral(), ctx.oC_Expression()).toList(String.class);
-        if (properties.isEmpty()) {
-            properties.addAll(getAllProperties(tag));
-        }
-        properties.add(0, GraphProperty.LABEL_KEY);
-        properties.add(1, GraphProperty.ID_KEY);
-        return properties;
-    }
-
-    private List<String> getAllProperties(@Nullable String tag) {
-        RexGraphVariable curVar = builder.variable(tag);
-        RelDataType dataType = curVar.getType();
-        Preconditions.checkArgument(
-                dataType instanceof GraphSchemaType, "can not get property from type=", dataType);
-        return dataType.getFieldList().stream().map(k -> k.getName()).collect(Collectors.toList());
-    }
-
     private GraphBuilder appendTailProject() {
         Preconditions.checkArgument(builder.size() > 0, "builder should not be empty");
         RelNode top = builder.peek();
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/PathFunctionVisitor.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/PathFunctionVisitor.java
new file mode 100644
index 000000000000..9756a41fddf5
--- /dev/null
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/antlr4x/visitor/PathFunctionVisitor.java
@@ -0,0 +1,317 @@
+/*
+ *
+ *  * Copyright 2020 Alibaba Group Holding Limited.
+ *  *
+ *  * Licensed under the Apache License, Version 2.0 (the "License");
+ *  * you may not use this file except in compliance with the License.
+ *  * You may obtain a copy of the License at
+ *  *
+ *  * http://www.apache.org/licenses/LICENSE-2.0
+ *  *
+ *  * Unless required by applicable law or agreed to in writing, software
+ *  * distributed under the License is distributed on an "AS IS" BASIS,
+ *  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  * See the License for the specific language governing permissions and
+ *  * limitations under the License.
+ *
+ */
+
+package com.alibaba.graphscope.gremlin.antlr4x.visitor;
+
+import com.alibaba.graphscope.common.ir.meta.schema.CommonOptTable;
+import com.alibaba.graphscope.common.ir.rel.CommonTableScan;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalPathExpand;
+import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalMultiMatch;
+import com.alibaba.graphscope.common.ir.rel.graph.match.GraphLogicalSingleMatch;
+import com.alibaba.graphscope.common.ir.rex.RexGraphVariable;
+import com.alibaba.graphscope.common.ir.tools.AliasInference;
+import com.alibaba.graphscope.common.ir.tools.GraphBuilder;
+import com.alibaba.graphscope.common.ir.tools.GraphStdOperatorTable;
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
+import com.alibaba.graphscope.common.ir.type.GraphPathType;
+import com.alibaba.graphscope.grammar.GremlinGSBaseVisitor;
+import com.alibaba.graphscope.grammar.GremlinGSParser;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Supplier;
+import com.google.common.collect.Lists;
+
+import org.apache.calcite.plan.GraphOptCluster;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * visit the antlr tree and generate the corresponding {@code RexNode} for path function
+ */
+public class PathFunctionVisitor extends GremlinGSBaseVisitor<RexNode> {
+    private final GraphBuilder parentBuilder;
+    private final RexNode variable;
+    private final GraphLogicalPathExpand pathExpand;
+    private final RelNode innerExpand;
+    private final RelNode innerGetV;
+
+    public PathFunctionVisitor(GraphBuilder parentBuilder, RexNode variable) {
+        Preconditions.checkArgument(
+                variable instanceof RexGraphVariable
+                        && ((RexGraphVariable) variable).getProperty() == null
+                        && variable.getType() instanceof GraphPathType,
+                "variable [" + variable + "] can not denote a path expand");
+        this.parentBuilder = parentBuilder;
+        this.variable = variable;
+        this.pathExpand = getPathExpand(parentBuilder.peek(), variable);
+        this.innerExpand =
+                Objects.requireNonNull(
+                        this.pathExpand.getExpand(), "expand in path expand can not be null");
+        this.innerGetV =
+                Objects.requireNonNull(
+                        this.pathExpand.getGetV(), "getV in path expand can not be null");
+    }
+
+    private GraphLogicalPathExpand getPathExpand(RelNode top, RexNode variable) {
+        int expectedAlias = ((RexGraphVariable) variable).getAliasId();
+        List<RelNode> inputQueue = Lists.newArrayList(top);
+        while (!inputQueue.isEmpty()) {
+            RelNode cur = inputQueue.remove(0);
+            if (cur instanceof GraphLogicalPathExpand
+                    && (expectedAlias == AliasInference.DEFAULT_ID
+                            || ((GraphLogicalPathExpand) cur).getAliasId() == expectedAlias)) {
+                return (GraphLogicalPathExpand) cur;
+            }
+            if (cur instanceof Project) {
+                Project project = (Project) cur;
+                RelDataType projectType = project.getRowType();
+                for (int i = 0; i < projectType.getFieldList().size(); ++i) {
+                    RelDataTypeField field = projectType.getFieldList().get(i);
+                    if (field.getIndex() == expectedAlias) {
+                        if (i < project.getProjects().size()
+                                && project.getProjects().get(i) instanceof RexGraphVariable) {
+                            return getPathExpand(project.getInput(), project.getProjects().get(i));
+                        }
+                        break;
+                    }
+                }
+            } else if (cur instanceof CommonTableScan) {
+                CommonOptTable optTable = (CommonOptTable) ((CommonTableScan) cur).getTable();
+                return getPathExpand(optTable.getCommon(), variable);
+            } else if (cur instanceof GraphLogicalSingleMatch) {
+                return getPathExpand(((GraphLogicalSingleMatch) cur).getSentence(), variable);
+            } else if (cur instanceof GraphLogicalMultiMatch) {
+                List<RelNode> sentences = ((GraphLogicalMultiMatch) cur).getSentences();
+                for (RelNode sentence : sentences) {
+                    try {
+                        return getPathExpand(sentence, variable);
+                    } catch (IllegalArgumentException e) {
+                        // ignore
+                    }
+                }
+            }
+            if (AliasInference.removeAlias(cur)) {
+                break;
+            }
+            inputQueue.addAll(cur.getInputs());
+        }
+        throw new IllegalArgumentException("can not find path expand by variable " + variable);
+    }
+
+    @Override
+    public RexNode visitTraversalMethod_valueMap(
+            GremlinGSParser.TraversalMethod_valueMapContext ctx) {
+        GraphBuilder builder =
+                GraphBuilder.create(
+                        parentBuilder.getContext(),
+                        (GraphOptCluster) parentBuilder.getCluster(),
+                        parentBuilder.getRelOptSchema());
+        GraphOpt.PathExpandFunction funcOpt = getFuncOpt();
+        // avoid to throw exception if the function opt is VERTEX_EDGE, to support the semantics
+        // like
+        // g.V().out('2..3').with('RESULT_OPT', 'ALL_V_E').valueMap('name', 'weight'),
+        // in nested path collection, each edge will return the property of 'weight' while each
+        // vertex will return the property of 'name'
+        boolean throwsOnPropertyNotFound =
+                (funcOpt == GraphOpt.PathExpandFunction.VERTEX_EDGE) ? false : true;
+        RexNode propertyProjection =
+                propertyProjection(
+                        () ->
+                                (new ExpressionVisitor(
+                                                builder.push(innerExpand),
+                                                builder.variable((String) null),
+                                                throwsOnPropertyNotFound))
+                                        .visitTraversalMethod_valueMap(ctx),
+                        () ->
+                                (new ExpressionVisitor(
+                                                builder.push(innerGetV),
+                                                builder.variable((String) null),
+                                                throwsOnPropertyNotFound))
+                                        .visitTraversalMethod_valueMap(ctx),
+                        builder);
+        return builder.call(
+                GraphStdOperatorTable.PATH_FUNCTION,
+                variable,
+                builder.getRexBuilder().makeFlag(funcOpt),
+                propertyProjection);
+    }
+
+    @Override
+    public RexNode visitTraversalMethod_values(GremlinGSParser.TraversalMethod_valuesContext ctx) {
+        GraphBuilder builder =
+                GraphBuilder.create(
+                        parentBuilder.getContext(),
+                        (GraphOptCluster) parentBuilder.getCluster(),
+                        parentBuilder.getRelOptSchema());
+        GraphOpt.PathExpandFunction funcOpt = getFuncOpt();
+        boolean throwsOnPropertyNotFound =
+                (funcOpt == GraphOpt.PathExpandFunction.VERTEX_EDGE) ? false : true;
+        RexNode propertyProjection =
+                propertyProjection(
+                        () ->
+                                (new ExpressionVisitor(
+                                                builder.push(innerExpand),
+                                                builder.variable((String) null),
+                                                throwsOnPropertyNotFound))
+                                        .visitTraversalMethod_values(ctx),
+                        () ->
+                                (new ExpressionVisitor(
+                                                builder.push(innerGetV),
+                                                builder.variable((String) null),
+                                                throwsOnPropertyNotFound))
+                                        .visitTraversalMethod_values(ctx),
+                        builder);
+        return builder.call(
+                GraphStdOperatorTable.PATH_FUNCTION,
+                variable,
+                builder.getRexBuilder().makeFlag(funcOpt),
+                propertyProjection);
+    }
+
+    @Override
+    public RexNode visitTraversalMethod_elementMap(
+            GremlinGSParser.TraversalMethod_elementMapContext ctx) {
+        GraphBuilder builder =
+                GraphBuilder.create(
+                        parentBuilder.getContext(),
+                        (GraphOptCluster) parentBuilder.getCluster(),
+                        parentBuilder.getRelOptSchema());
+        GraphOpt.PathExpandFunction funcOpt = getFuncOpt();
+        boolean throwsOnPropertyNotFound =
+                (funcOpt == GraphOpt.PathExpandFunction.VERTEX_EDGE) ? false : true;
+        RexNode propertyProjection =
+                propertyProjection(
+                        () ->
+                                (new ExpressionVisitor(
+                                                builder.push(innerExpand),
+                                                builder.variable((String) null),
+                                                throwsOnPropertyNotFound))
+                                        .visitTraversalMethod_elementMap(ctx),
+                        () ->
+                                (new ExpressionVisitor(
+                                                builder.push(innerGetV),
+                                                builder.variable((String) null),
+                                                throwsOnPropertyNotFound))
+                                        .visitTraversalMethod_elementMap(ctx),
+                        builder);
+        return builder.call(
+                GraphStdOperatorTable.PATH_FUNCTION,
+                variable,
+                builder.getRexBuilder().makeFlag(funcOpt),
+                propertyProjection);
+    }
+
+    @Override
+    public RexNode visitTraversalMethod_selectby(
+            GremlinGSParser.TraversalMethod_selectbyContext ctx) {
+        GraphBuilder builder =
+                GraphBuilder.create(
+                        parentBuilder.getContext(),
+                        (GraphOptCluster) parentBuilder.getCluster(),
+                        parentBuilder.getRelOptSchema());
+        GraphOpt.PathExpandFunction funcOpt = getFuncOpt();
+        boolean throwsOnPropertyNotFound =
+                (funcOpt == GraphOpt.PathExpandFunction.VERTEX_EDGE) ? false : true;
+        RexNode propertyProjection =
+                propertyProjection(
+                        () ->
+                                (new ExpressionVisitor(
+                                                builder.push(innerExpand),
+                                                builder.variable((String) null),
+                                                throwsOnPropertyNotFound))
+                                        .visitTraversalMethod_selectby(ctx),
+                        () ->
+                                (new ExpressionVisitor(
+                                                builder.push(innerGetV),
+                                                builder.variable((String) null),
+                                                throwsOnPropertyNotFound))
+                                        .visitTraversalMethod_selectby(ctx),
+                        builder);
+        return builder.call(
+                GraphStdOperatorTable.PATH_FUNCTION,
+                variable,
+                builder.getRexBuilder().makeFlag(funcOpt),
+                propertyProjection);
+    }
+
+    private RexNode propertyProjection(
+            Supplier<RexNode> expandSupplier,
+            Supplier<RexNode> getVSupplier,
+            GraphBuilder builder) {
+        GraphOpt.PathExpandFunction funcOpt = getFuncOpt();
+        switch (funcOpt) {
+            case VERTEX:
+                return getVSupplier.get();
+            case EDGE:
+                return expandSupplier.get();
+            case VERTEX_EDGE:
+            default:
+                RexNode expandProjection = expandSupplier.get();
+                RexNode getVProjection = getVSupplier.get();
+                return unionProjection(expandProjection, getVProjection, builder);
+        }
+    }
+
+    private RexNode unionProjection(
+            RexNode expandProjection, RexNode getVProjection, GraphBuilder builder) {
+        if (expandProjection == null || getVProjection == null) {
+            Preconditions.checkArgument(
+                    expandProjection != null || getVProjection != null,
+                    "invalid query given properties, not found in path expand");
+            return expandProjection != null ? expandProjection : getVProjection;
+        }
+        Preconditions.checkArgument(
+                expandProjection.getKind() == getVProjection.getKind(),
+                "expand projection ["
+                        + expandProjection
+                        + "] is not consistent with getV projection ["
+                        + getVProjection
+                        + "]");
+        if (expandProjection instanceof RexCall) {
+            RexCall expandCall = (RexCall) expandProjection;
+            RexCall getVCall = (RexCall) getVProjection;
+            List<RexNode> newOperands = Lists.newArrayList(expandCall.getOperands());
+            newOperands.addAll(getVCall.getOperands());
+            Preconditions.checkArgument(
+                    !newOperands.isEmpty(),
+                    "invalid query given properties, not found in path expand");
+            return builder.call(
+                    expandCall.getOperator(),
+                    newOperands.stream().distinct().collect(Collectors.toList()));
+        }
+        return expandProjection;
+    }
+
+    private GraphOpt.PathExpandFunction getFuncOpt() {
+        switch (pathExpand.getResultOpt()) {
+            case ALL_V:
+            case END_V:
+                return GraphOpt.PathExpandFunction.VERTEX;
+            case ALL_V_E:
+            default:
+                return GraphOpt.PathExpandFunction.VERTEX_EDGE;
+        }
+    }
+}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/graph/RemoteTestGraph.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/graph/RemoteTestGraph.java
index 10ebf37bdd74..5842710c43c9 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/graph/RemoteTestGraph.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/graph/RemoteTestGraph.java
@@ -1333,15 +1333,17 @@
 // @Graph.OptOut(method="g_V_hasXp_neqXvXX" ,
 // test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
 // property is unsupported")
-// @Graph.OptOut(method="g_V_hasXage_withoutX27X_count" ,
-// test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
-// property is unsupported")
+@Graph.OptOut(
+        method = "g_V_hasXage_withoutX27X_count",
+        test = "org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest",
+        reason = "None value is not supported yet")
 // @Graph.OptOut(method="g_VX1X_hasXcircumferenceX" ,
 // test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
 // property is unsupported")
-// @Graph.OptOut(method="g_V_hasXage_withoutX27_29X_count" ,
-// test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
-// property is unsupported")
+@Graph.OptOut(
+        method = "g_V_hasXage_withoutX27_29X_count",
+        test = "org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest",
+        reason = "None value is not supported yet")
 // @Graph.OptOut(method="g_VX1X_hasXnameX" ,
 // test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
 // property is unsupported")
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 3801a968ad55..6eea6e96a420 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
@@ -1284,15 +1284,17 @@
 // @Graph.OptOut(method="g_V_hasXp_neqXvXX" ,
 // test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
 // property is unsupported")
-// @Graph.OptOut(method="g_V_hasXage_withoutX27X_count" ,
-// test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
-// property is unsupported")
+@Graph.OptOut(
+        method = "g_V_hasXage_withoutX27X_count",
+        test = "org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest",
+        reason = "None value is not supported yet")
 // @Graph.OptOut(method="g_VX1X_hasXcircumferenceX" ,
 // test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
 // property is unsupported")
-// @Graph.OptOut(method="g_V_hasXage_withoutX27_29X_count" ,
-// test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
-// property is unsupported")
+@Graph.OptOut(
+        method = "g_V_hasXage_withoutX27_29X_count",
+        test = "org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest",
+        reason = "None value is not supported yet")
 // @Graph.OptOut(method="g_VX1X_hasXnameX" ,
 // test="org.apache.tinkerpop.gremlin.process.traversal.step.filter.HasTest", reason = "existence of
 // property is unsupported")
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/pattern/PatternQueryTest.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/pattern/PatternQueryTest.java
index 518e48ca2b61..ed6f140ac20e 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/pattern/PatternQueryTest.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/integration/suite/pattern/PatternQueryTest.java
@@ -546,6 +546,7 @@ public Traversal<Vertex, Long> get_st_path_test() {
                                             __.as("a")
                                                     .has("PERSON", "id", 2199023256684L)
                                                     .out("5..6", "KNOWS")
+                                                    .with("RESULT_OPT", "ALL_V")
                                                     .as("c"))
                                     .endV()
                                     .has("PERSON", "id", 8796093023060L)
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/result/ParserUtils.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/result/ParserUtils.java
index 9484adaf88a1..982983215c95 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/result/ParserUtils.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/result/ParserUtils.java
@@ -226,18 +226,6 @@ public static Object parseEntry(IrResult.Entry entry) {
     }
 
     public static Object parseElement(IrResult.KeyValues.KeyValue value) {
-        switch (value.getValueCase()) {
-            case ELEMENT:
-                return parseElement(value.getElement());
-            case NESTED:
-                return parseEntry(IrResult.Entry.newBuilder().setMap(value.getNested()).build());
-            default:
-                throw new GremlinResultParserException(
-                        "keyValue ["
-                                + value
-                                + "] has invalid value type ["
-                                + value.getValueCase()
-                                + "]");
-        }
+        return parseEntry(value.getValue());
     }
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/resultx/GremlinRecordParser.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/resultx/GremlinRecordParser.java
index fe490220539d..dab05c9e6451 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/resultx/GremlinRecordParser.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/gremlin/resultx/GremlinRecordParser.java
@@ -32,6 +32,7 @@
 
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.sql.type.MapSqlType;
 import org.apache.commons.lang3.NotImplementedException;
 import org.apache.tinkerpop.gremlin.structure.Edge;
 import org.apache.tinkerpop.gremlin.structure.Element;
@@ -161,7 +162,7 @@ private Map<Object, Object> parseKeyValues(
                 .getKeyValuesList()
                 .forEach(
                         entry -> {
-                            Object value = parseElement(entry, valueType);
+                            Object value = parseEntry(entry.getValue(), valueType);
                             if (value != null) {
                                 valueMap.put(parseMapKey(entry.getKey().getStr()), value);
                             }
@@ -183,7 +184,7 @@ private Map<Object, Object> parseKeyValues(
         Map<Object, Object> valueMap = Maps.newLinkedHashMap();
         for (int i = 0; i < entries.size(); ++i) {
             IrResult.KeyValues.KeyValue entry = entries.get(i);
-            Object value = parseElement(entry, valueTypes.get(i));
+            Object value = parseEntry(entry.getValue(), valueTypes.get(i));
             if (value != null) {
                 valueMap.put(parseMapKey(entry.getKey().getStr()), value);
             }
@@ -219,30 +220,6 @@ private Object parseElement(IrResult.Element element, RelDataType type) {
         }
     }
 
-    private Object parseElement(IrResult.KeyValues.KeyValue value, RelDataType type) {
-        switch (value.getValueCase()) {
-            case ELEMENT:
-                return parseElement(value.getElement(), type);
-            case NESTED:
-                if (type instanceof ArbitraryMapType) {
-                    return parseKeyValues(
-                            value.getNested(),
-                            ((ArbitraryMapType) type).getKeyTypes(),
-                            ((ArbitraryMapType) type).getValueTypes());
-                } else {
-                    return parseKeyValues(
-                            value.getNested(), type.getKeyType(), type.getValueType());
-                }
-            default:
-                throw new GremlinResultParserException(
-                        "keyValue ["
-                                + value
-                                + "] has invalid value type ["
-                                + value.getValueCase()
-                                + "]");
-        }
-    }
-
     private Vertex parseVertex(IrResult.Vertex vertex, RelDataType type) {
         return new DetachedVertex(
                 vertex.getId(),
@@ -311,6 +288,30 @@ private List<Element> parseGraphPath(IrResult.GraphPath path, @Nullable RelDataT
                 return value.getF64Array().getItemList();
             case STR_ARRAY:
                 return value.getStrArray().getItemList();
+            case PAIR_ARRAY:
+                if (dataType instanceof MapSqlType) {
+                    return value.getPairArray().getItemList().stream()
+                            .collect(
+                                    Collectors.toMap(
+                                            k -> parseValue(k.getKey(), dataType.getKeyType()),
+                                            v -> parseValue(v.getVal(), dataType.getValueType())));
+                } else if (dataType instanceof ArbitraryMapType) {
+                    List<Common.Pair> pairArray = value.getPairArray().getItemList();
+                    List<RelDataType> keyTypes = ((ArbitraryMapType) dataType).getKeyTypes();
+                    List<RelDataType> valueTypes = ((ArbitraryMapType) dataType).getValueTypes();
+                    Map map = Maps.newLinkedHashMap();
+                    for (int i = 0; i < pairArray.size(); ++i) {
+                        Common.Pair pair = pairArray.get(i);
+                        map.put(
+                                parseValue(
+                                        pair.getKey(),
+                                        i < keyTypes.size() ? keyTypes.get(i) : null),
+                                parseValue(
+                                        pair.getVal(),
+                                        i < valueTypes.size() ? valueTypes.get(i) : null));
+                    }
+                    return map;
+                }
             case NONE:
                 return null;
                 // todo: support temporal types
diff --git a/interactive_engine/executor/common/dyn_type/src/object.rs b/interactive_engine/executor/common/dyn_type/src/object.rs
index e8c3b90f7dfe..c4657c1c929c 100644
--- a/interactive_engine/executor/common/dyn_type/src/object.rs
+++ b/interactive_engine/executor/common/dyn_type/src/object.rs
@@ -926,6 +926,20 @@ impl Object {
         }
     }
 
+    pub fn take_vector(self) -> Result<Vec<Object>, CastError> {
+        match self {
+            Object::Vector(vec) => Ok(vec),
+            Object::DynOwned(mut x) => {
+                if let Some(v) = x.try_downcast_mut::<Vec<Object>>() {
+                    Ok(std::mem::replace(v, Vec::new()))
+                } else {
+                    Err(CastError::new::<Vec<Object>>(RawType::Unknown))
+                }
+            }
+            _ => Err(CastError::new::<Vec<Object>>(self.raw_type())),
+        }
+    }
+
     fn contains_single(&self, single: &Object) -> bool {
         contains_single!(self, single, Object)
     }
diff --git a/interactive_engine/executor/ir/core/src/plan/logical.rs b/interactive_engine/executor/ir/core/src/plan/logical.rs
index e05cf343b95b..f565d383f667 100644
--- a/interactive_engine/executor/ir/core/src/plan/logical.rs
+++ b/interactive_engine/executor/ir/core/src/plan/logical.rs
@@ -1117,6 +1117,9 @@ fn preprocess_expression(
                                 common_pb::variable_key_value::Value::Nested(_) => {
                                     return Err(IrError::Unsupported("nested value in Map".to_string()));
                                 }
+                                common_pb::variable_key_value::Value::PathFunc(_) => {
+                                    return Err(IrError::Unsupported("PathFunc in Map".to_string()));
+                                }
                             }
                         }
                     }
diff --git a/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/path.rs b/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/path.rs
index 1832e08dfb27..8b6be03bac22 100644
--- a/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/path.rs
+++ b/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/path.rs
@@ -166,6 +166,13 @@ impl GraphPath {
         }
     }
 
+    pub fn get_path_end_mut(&mut self) -> &mut VertexOrEdge {
+        match self {
+            GraphPath::AllPath(ref mut p) | GraphPath::SimpleAllPath(ref mut p) => p.last_mut().unwrap(),
+            GraphPath::EndV((ref mut e, _)) | GraphPath::SimpleEndV((ref mut e, _, _)) => e,
+        }
+    }
+
     pub fn get_path(&self) -> Option<&Vec<VertexOrEdge>> {
         match self {
             GraphPath::AllPath(p) | GraphPath::SimpleAllPath(p) => Some(p),
@@ -318,13 +325,22 @@ impl GraphElement for GraphPath {
                 for v_or_e in path {
                     if let Some(p) = v_or_e.get_property(key) {
                         properties.push(p.try_to_owned().unwrap());
+                    } else {
+                        // Fill with None if the property is not found. Otherwise, the property order will be inconsistent.
+                        properties.push(Object::None);
                     }
                 }
                 Some(PropertyValue::Owned(Object::Vector(properties)))
             }
 
             GraphPath::EndV((v_or_e, _)) | GraphPath::SimpleEndV((v_or_e, _, _)) => {
-                v_or_e.get_property(key)
+                if let Some(prop) = v_or_e.get_property(key) {
+                    if let Some(obj) = prop.try_to_owned() {
+                        // make the type consistent with the all path.
+                        return Some(PropertyValue::Owned(Object::Vector(vec![obj])));
+                    }
+                }
+                Some(PropertyValue::Owned(Object::Vector(vec![Object::None])))
             }
         }
     }
diff --git a/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/property.rs b/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/property.rs
index 715dbb10a53b..160c0662b873 100644
--- a/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/property.rs
+++ b/interactive_engine/executor/ir/graph_proxy/src/apis/graph/element/property.rs
@@ -26,6 +26,9 @@ use pegasus_common::codec::{Decode, Encode, ReadExt, WriteExt};
 use pegasus_common::downcast::*;
 use pegasus_common::impl_as_any;
 
+use super::Element;
+use crate::utils::expr::{ExprEvalError, ExprEvalResult};
+
 /// The three types of property to get
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 pub enum PropKey {
@@ -36,6 +39,51 @@ pub enum PropKey {
     Key(NameOrId),
 }
 
+impl PropKey {
+    pub fn get_key<E: Element>(&self, element: &E) -> ExprEvalResult<Object> {
+        let prop_obj = if let PropKey::Len = self {
+            element.len().into()
+        } else {
+            let graph_element = element
+                .as_graph_element()
+                .ok_or_else(|| ExprEvalError::UnexpectedDataType(self.into()))?;
+            match self {
+                PropKey::Id => graph_element.id().into(),
+                PropKey::Label => graph_element
+                    .label()
+                    .map(|label| label.into())
+                    .unwrap_or(Object::None),
+                PropKey::Len => unreachable!(),
+                PropKey::All => graph_element
+                    .get_all_properties()
+                    .map(|obj| {
+                        obj.into_iter()
+                            .map(|(key, value)| {
+                                let obj_key: Object = match key {
+                                    NameOrId::Str(str) => str.into(),
+                                    NameOrId::Id(id) => id.into(),
+                                };
+                                (obj_key, value)
+                            })
+                            .collect::<Vec<(Object, Object)>>()
+                            .into()
+                    })
+                    .unwrap_or(Object::None),
+                PropKey::Key(key) => {
+                    if let Some(prop_val) = graph_element.get_property(key) {
+                        prop_val.try_to_owned().ok_or_else(|| {
+                            ExprEvalError::OtherErr("cannot get `Object` from `BorrowObject`".to_string())
+                        })?
+                    } else {
+                        Object::None
+                    }
+                }
+            }
+        };
+        Ok(prop_obj)
+    }
+}
+
 impl TryFrom<pb::Property> for PropKey {
     type Error = ParsePbError;
 
diff --git a/interactive_engine/executor/ir/graph_proxy/src/utils/expr/eval.rs b/interactive_engine/executor/ir/graph_proxy/src/utils/expr/eval.rs
index aa507e07c401..6ff5bc199b37 100644
--- a/interactive_engine/executor/ir/graph_proxy/src/utils/expr/eval.rs
+++ b/interactive_engine/executor/ir/graph_proxy/src/utils/expr/eval.rs
@@ -157,6 +157,12 @@ impl From<&Operand> for OperatorDesc {
     }
 }
 
+impl From<&PropKey> for OperatorDesc {
+    fn from(prop: &PropKey) -> Self {
+        Self(format!("{:?}", prop))
+    }
+}
+
 /// A `Context` gives the behavior of obtaining a certain tag from the runtime
 /// for evaluating variables in an expression.
 pub trait Context<E: Element> {
@@ -629,6 +635,7 @@ impl TryFrom<common_pb::VariableKeyValues> for Operand {
                 match value {
                     common_pb::variable_key_value::Value::Val(val) => Operand::try_from(val)?,
                     common_pb::variable_key_value::Value::Nested(nested) => Operand::try_from(nested)?,
+                    common_pb::variable_key_value::Value::PathFunc(_path_func) => todo!(),
                 }
             } else {
                 return Err(ParsePbError::from("empty value provided in Map"));
@@ -707,47 +714,7 @@ impl Evaluate for Operand {
                 if let Some(ctxt) = context {
                     if let Some(element) = ctxt.get(tag.as_ref()) {
                         let result = if let Some(property) = prop_key {
-                            if let PropKey::Len = property {
-                                element.len().into()
-                            } else {
-                                let graph_element = element
-                                    .as_graph_element()
-                                    .ok_or_else(|| ExprEvalError::UnexpectedDataType(self.into()))?;
-                                match property {
-                                    PropKey::Id => graph_element.id().into(),
-                                    PropKey::Label => graph_element
-                                        .label()
-                                        .map(|label| label.into())
-                                        .unwrap_or(Object::None),
-                                    PropKey::Len => unreachable!(),
-                                    PropKey::All => graph_element
-                                        .get_all_properties()
-                                        .map(|obj| {
-                                            obj.into_iter()
-                                                .map(|(key, value)| {
-                                                    let obj_key: Object = match key {
-                                                        NameOrId::Str(str) => str.into(),
-                                                        NameOrId::Id(id) => id.into(),
-                                                    };
-                                                    (obj_key, value)
-                                                })
-                                                .collect::<Vec<(Object, Object)>>()
-                                                .into()
-                                        })
-                                        .unwrap_or(Object::None),
-                                    PropKey::Key(key) => {
-                                        if let Some(prop) = graph_element.get_property(key) {
-                                            prop.try_to_owned().ok_or_else(|| {
-                                                ExprEvalError::OtherErr(
-                                                    "cannot get `Object` from `BorrowObject`".to_string(),
-                                                )
-                                            })?
-                                        } else {
-                                            Object::None
-                                        }
-                                    }
-                                }
-                            }
+                            property.get_key(element)?
                         } else {
                             element
                                 .as_borrow_object()
@@ -1095,6 +1062,7 @@ mod tests {
             "{@0.name, @0.age}",                           // {"name": "John", "age": 31}
             "@0.~all", // {age = 31, name = John, birthday = 19900416, hobbies = [football, guitar]]}
             "{@0.~all}", // {~all, {age = 31, name = John, birthday = 19900416, hobbies = [football, guitar]]}}
+            "@0.not_exist", // Object::None
         ];
 
         let expected: Vec<Object> = vec![
@@ -1145,6 +1113,7 @@ mod tests {
                 .into_iter()
                 .collect(),
             ),
+            Object::None,
         ];
 
         for (case, expected) in cases.into_iter().zip(expected.into_iter()) {
@@ -1155,17 +1124,8 @@ mod tests {
 
     #[test]
     fn test_eval_errors() {
-        let cases: Vec<&str> = vec![
-            "@2",
-            "+",
-            "1 + ",
-            "1 1",
-            "1 1 2",
-            "1 1 + 2",
-            "1 + @1.age * 1 1 - 1 - 5",
-            "@2",
-            // "@0.not_exist", // if the property does not exist, it will be rejected by the type checker
-        ];
+        let cases: Vec<&str> =
+            vec!["@2", "+", "1 + ", "1 1", "1 1 2", "1 1 + 2", "1 + @1.age * 1 1 - 1 - 5", "@2"];
         let ctxt = prepare_context();
 
         let expected: Vec<ExprEvalError> = vec![
@@ -1181,7 +1141,6 @@ mod tests {
             ExprEvalError::OtherErr("invalid expression".to_string()),
             ExprEvalError::OtherErr("invalid expression".to_string()),
             ExprEvalError::GetNoneFromContext,
-            // ExprEvalError::GetNoneFromContext,
         ];
 
         let mut is_context = false;
diff --git a/interactive_engine/executor/ir/proto/expr.proto b/interactive_engine/executor/ir/proto/expr.proto
index 0d93d7914d88..2e2d10cf5c0e 100644
--- a/interactive_engine/executor/ir/proto/expr.proto
+++ b/interactive_engine/executor/ir/proto/expr.proto
@@ -117,6 +117,44 @@ message Variable {
   common.IrDataType node_type = 3;
 }
 
+// Path function to rechieve properties of each vertex or edge (or both) in a path.
+// e.g., project `a.name`, `a.[name, age]`, or `a.{name, age}`, where `a` refers to a path
+message PathFunction {
+  // rechieve properties from a path to a list
+  message PathElementKeys {
+    repeated Property keys = 1;
+  }
+  // rechieve properties from a path to a map
+  message PathElementKeyValues {
+    message PathElementKeyValue {
+      common.Value key = 1;
+      Property val = 2;
+    }
+    repeated PathElementKeyValue key_vals = 1;
+  }
+  // the option of PathFunction, e.g., rechieve the properties of each vertex, edge, or both, in a path
+  enum FuncOpt {
+    VERTEX = 0;
+    EDGE = 1;
+    VERTEX_EDGE = 2;
+  }
+  // The alias of a graph path
+  common.NameOrId tag = 1;
+  oneof path_key {
+    // e.g., project name, and return the property of each entry in the path, 
+    // and return [marko, josh]
+    Property property = 2;
+    // e.g., project [name, age] to a list of properties of each entry in the path, 
+    // and [[marko, 29], [josh, 27]]
+    PathElementKeys vars = 3;
+    // e.g., project {name, age} to a map of properties of each entry in the path,
+    // and return [{name: marko, age: 29}, {name: josh, age: 27}]
+    PathElementKeyValues map = 4;
+  }
+  FuncOpt opt = 5;
+  common.IrDataType node_type = 6;
+}
+
 message VariableKeys {
   repeated Variable keys = 1;
 }
@@ -125,14 +163,22 @@ message VariableKeyValue {
   common.Value key = 1;
   oneof value {
     Variable val = 2;
-    VariableKeyValues nested = 3;
+    PathFunction path_func = 3;
+    VariableKeyValues nested = 4;
   }
 }
 
 // A nested kv projection, which is used to project a nested structure of key-value pairs
-// e.g., to project [value(a.name), value(a.age)], or project [key_value('name', a.name), key_value('age', a.age)],
-// or more complex nested projection, 
-// e.g., to project [key_value('tagA', nested{[key_value('name', a.name), key_value('age', a.age)]}), key_value('tagB', nested{key_value(b.name), key_value(b.age)})]
+// e.g., for a single kv projection,
+// to project [VariableKeyValue('name', a.name), VariableKeyValue('age', a.age)], where `a` refers to a vertex,
+// it outputs a map of {'name': 'marko', 'age': 27};
+// or to project [VariableKeyValue('tagP', PathFunction{tag: a, path_key: PathElementKeys{keys: [name, age]}})], where `a` refers to a path (with two vertices in it),
+// it outputs a map of {'tagP': [{'name': 'marko', 'age': 27}, {'name': 'josh', 'age', 29}]}.
+// For a more complex nested projection, 
+// to project [VariableKeyValue('tagA', nested{[VariableKeyValue('name', a.name), VariableKeyValue('age', a.age)]}), 
+//             VariableKeyValue('tagP', nested{[VariableKeyValue('name', p.name), VariableKeyValue('age', p.age)]})],
+// where `a` refers to a vertex, and `p` refers to a path (with two vertices in it),
+// it outputs a map of {'tagA': {'name': 'marko', 'age': 27}, 'tagP': [{'name': 'marko', 'age': 27}, {'name': 'josh', 'age', 29}]}.
 message VariableKeyValues {
   repeated VariableKeyValue key_vals = 1;
 }
@@ -225,6 +271,7 @@ message ExprOpr {
     TimeInterval time_interval = 14;
     DateTimeMinus date_time_minus = 15;
     PathConcat path_concat = 16;
+    PathFunction path_func = 17;
   }
   // The data of type of ExprOpr
   common.IrDataType node_type = 12;
diff --git a/interactive_engine/executor/ir/proto/results.proto b/interactive_engine/executor/ir/proto/results.proto
index 9938fefc4792..2a2782684cd0 100644
--- a/interactive_engine/executor/ir/proto/results.proto
+++ b/interactive_engine/executor/ir/proto/results.proto
@@ -68,10 +68,11 @@ message Collection {
 message KeyValues {
   message KeyValue {
     common.Value key = 1;
-    oneof value {
-      Element element = 2;
-      KeyValues nested = 3;
-    }
+    // the value is an entry, which includes cases of:
+    // 1. element, e.g., results of vertex.valueMap('name');
+    // 2. collection, e.g., results of select('a','b'), where a is a Path, the result value of 'a' is a collection;
+    // 3. nested map, e.g., results of select('a','b').valueMap('name').
+    Entry value = 2;
   }
   repeated KeyValue key_values = 1;
 }
diff --git a/interactive_engine/executor/ir/runtime/src/assembly.rs b/interactive_engine/executor/ir/runtime/src/assembly.rs
index fa904b445e0d..4f8c66926782 100644
--- a/interactive_engine/executor/ir/runtime/src/assembly.rs
+++ b/interactive_engine/executor/ir/runtime/src/assembly.rs
@@ -817,7 +817,7 @@ impl<P: PartitionInfo, C: ClusterInfo> IRJobAssembly<P, C> {
                         None
                     };
                     // process get_v
-                    if let Some(getv) = base.get_v.take() {
+                    if let Some(mut getv) = base.get_v.take() {
                         if (pb::get_v::VOpt::Itself as i32) == getv.opt {
                             // the case of expandv + auxilia (to deal with filtering on vertices).
                             if let Some(repartition) = repartition {
@@ -826,9 +826,31 @@ impl<P: PartitionInfo, C: ClusterInfo> IRJobAssembly<P, C> {
                             base_expand_plan.push(getv.clone().into());
                         } else {
                             // the case of expande + getv
-                            base_expand_plan.push(getv.clone().into());
-                            if let Some(repartition) = repartition {
-                                base_expand_plan.push(repartition);
+                            // specifically, if getv has predicates or columns,
+                            // separate it as getv(getAdj) + auxilia (for filter/get columns)
+                            let needs_separate = getv
+                                .params
+                                .as_ref()
+                                .map_or(false, |params| params.has_predicates() || params.has_columns());
+                            if needs_separate {
+                                let alias = getv.alias.take();
+                                let params = getv.params.take();
+                                let auxilia = pb::GetV {
+                                    opt: pb::get_v::VOpt::Itself as i32,
+                                    tag: None,
+                                    params: params,
+                                    alias,
+                                };
+                                base_expand_plan.push(getv.clone().into());
+                                if let Some(repartition) = repartition {
+                                    base_expand_plan.push(repartition);
+                                }
+                                base_expand_plan.push(auxilia.into());
+                            } else {
+                                base_expand_plan.push(getv.clone().into());
+                                if let Some(repartition) = repartition {
+                                    base_expand_plan.push(repartition);
+                                }
                             }
                         }
                     } else {
diff --git a/interactive_engine/executor/ir/runtime/src/process/entry.rs b/interactive_engine/executor/ir/runtime/src/process/entry.rs
index 168c66404cd0..ef73406c2c9d 100644
--- a/interactive_engine/executor/ir/runtime/src/process/entry.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/entry.rs
@@ -576,6 +576,7 @@ impl TryFrom<result_pb::Element> for DynEntry {
     }
 }
 
+// this is for ci tests
 impl TryFrom<result_pb::Entry> for DynEntry {
     type Error = ParsePbError;
 
@@ -597,9 +598,9 @@ impl TryFrom<result_pb::Entry> for DynEntry {
                     let mut map = BTreeMap::new();
                     for key_val in kv.key_values {
                         let key = key_val.key.unwrap();
-                        let value_inner = key_val.value.unwrap();
+                        let value_inner = key_val.value.unwrap().inner.unwrap();
                         match value_inner {
-                            result_pb::key_values::key_value::Value::Element(val) => {
+                            result_pb::entry::Inner::Element(val) => {
                                 let val_inner = val.inner.unwrap();
                                 let val_obj: Object = match val_inner {
                                     result_pb::element::Inner::Object(obj) => Object::try_from(obj)?,
@@ -610,7 +611,8 @@ impl TryFrom<result_pb::Entry> for DynEntry {
                                 };
                                 map.insert(Object::try_from(key)?, val_obj);
                             }
-                            result_pb::key_values::key_value::Value::Nested(_) => todo!(),
+                            result_pb::entry::Inner::Collection(_) => todo!(),
+                            result_pb::entry::Inner::Map(_) => todo!(),
                         }
                     }
                     Ok(DynEntry::new(Object::KV(map)))
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/map/get_v.rs b/interactive_engine/executor/ir/runtime/src/process/operator/map/get_v.rs
index 359f818b0bfd..b07a23310bad 100644
--- a/interactive_engine/executor/ir/runtime/src/process/operator/map/get_v.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/map/get_v.rs
@@ -25,7 +25,7 @@ use ir_common::{KeyId, LabelId};
 use pegasus::api::function::{FilterMapFunction, FnResult};
 
 use crate::error::{FnExecError, FnExecResult, FnGenError, FnGenResult};
-use crate::process::entry::{DynEntry, Entry, EntryType};
+use crate::process::entry::{DynEntry, Entry};
 use crate::process::operator::map::FilterMapFuncGen;
 use crate::process::record::Record;
 
@@ -169,71 +169,83 @@ impl FilterMapFunction<Record, Record> for AuxiliaOperator {
                 }
             }
             // 2. Otherwise, filter after query store, e.g., the case of filter by columns.
-            match entry.get_type() {
-                EntryType::Vertex => {
-                    let graph = get_graph().ok_or_else(|| FnExecError::NullGraphError)?;
-                    let id = entry.id();
-                    if let Some(vertex) = graph
-                        .get_vertex(&[id], &self.query_params)?
-                        .next()
-                        .map(|vertex| DynEntry::new(vertex))
-                    {
-                        if let Some(alias) = self.alias {
-                            // append without moving head
-                            input
-                                .get_columns_mut()
-                                .insert(alias as usize, vertex.into());
-                        } else {
-                            input.append(vertex, self.alias.clone());
-                        }
-                    } else {
-                        return Ok(None);
-                    }
-                }
-                EntryType::Edge => {
-                    // TODO: This is a little bit tricky. Modify this logic to query store once query by eid is supported.
-                    // Currently, we support two cases:
-                    // 1. use auxilia to rename the edge to a new alias;
-                    // 2. use auxilia to filter the edge by predicates, where the necessary properties of the edge is assumed to be already pre-cached.
-                    let entry = entry.clone();
-                    if let Some(predicate) = &self.query_params.filter {
-                        let res = predicate
-                            .eval_bool(Some(&input))
-                            .map_err(|e| FnExecError::from(e))?;
-                        if !res {
-                            return Ok(None);
-                        }
-                    }
+            let graph = get_graph().ok_or_else(|| FnExecError::NullGraphError)?;
+            if let Some(v) = entry.as_vertex() {
+                let id = v.id();
+                if let Some(vertex) = graph
+                    .get_vertex(&[id], &self.query_params)?
+                    .next()
+                    .map(|vertex| DynEntry::new(vertex))
+                {
                     if let Some(alias) = self.alias {
                         // append without moving head
                         input
                             .get_columns_mut()
-                            .insert(alias as usize, entry);
+                            .insert(alias as usize, vertex.into());
                     } else {
-                        input.append_arc_entry(entry, self.alias.clone());
+                        input.append(vertex, self.alias.clone());
                     }
+                } else {
+                    return Ok(None);
                 }
-                EntryType::Path => {
-                    // Auxilia for vertices in Path is for filtering.
-                    let graph_path = entry
-                        .as_graph_path()
-                        .ok_or_else(|| FnExecError::Unreachable)?;
-                    let path_end = graph_path.get_path_end();
-                    let graph = get_graph().ok_or_else(|| FnExecError::NullGraphError)?;
-                    let id = path_end.id();
-                    if graph
-                        .get_vertex(&[id], &self.query_params)?
-                        .next()
-                        .is_none()
-                    {
+            } else if let Some(_edge) = entry.as_edge() {
+                // TODO: This is a little bit tricky. Modify this logic to query store once query by eid is supported.
+                // Currently, we support two cases:
+                // 1. use auxilia to rename the edge to a new alias;
+                // 2. use auxilia to filter the edge by predicates, where the necessary properties of the edge is assumed to be already pre-cached.
+                let entry = entry.clone();
+                if let Some(predicate) = &self.query_params.filter {
+                    let res = predicate
+                        .eval_bool(Some(&input))
+                        .map_err(|e| FnExecError::from(e))?;
+                    if !res {
                         return Ok(None);
                     }
                 }
-                _ => Err(FnExecError::unexpected_data_error(&format!(
+                if let Some(alias) = self.alias {
+                    // append without moving head
+                    input
+                        .get_columns_mut()
+                        .insert(alias as usize, entry);
+                } else {
+                    input.append_arc_entry(entry, self.alias.clone());
+                }
+            } else if let Some(graph_path) = entry.as_graph_path() {
+                // 1. Auxilia for vertices in Path for filtering.
+                // 2. Auxilia for vertices in Path for property caching.
+                let path_end = graph_path.get_path_end();
+                if let Some(v) = graph
+                    .get_vertex(&[path_end.id()], &self.query_params)?
+                    .next()
+                {
+                    if self.query_params.has_columns() {
+                        // for property caching
+                        let mut_graph_path = input
+                            .get_mut(self.tag)
+                            .ok_or_else(|| {
+                                FnExecError::unexpected_data_error(&format!(
+                                    "get_mut of GraphPath failed in {:?}",
+                                    self
+                                ))
+                            })?
+                            .as_any_mut()
+                            .downcast_mut::<GraphPath>()
+                            .ok_or_else(|| {
+                                FnExecError::unexpected_data_error(&format!("entry is not a path in GetV"))
+                            })?;
+                        let path_end = mut_graph_path.get_path_end_mut();
+                        *path_end = v.into();
+                    }
+                    return Ok(Some(input));
+                } else {
+                    return Ok(None);
+                }
+            } else {
+                Err(FnExecError::unexpected_data_error(&format!(
                     "neither Vertex nor Edge entry is accessed in `Auxilia` operator, the entry is {:?}",
                     entry
-                )))?,
-            };
+                )))?
+            }
             Ok(Some(input))
         } else {
             Ok(None)
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/map/project.rs b/interactive_engine/executor/ir/runtime/src/process/operator/map/project.rs
index 1ae5bf3c519b..fd77a6f5fa77 100644
--- a/interactive_engine/executor/ir/runtime/src/process/operator/map/project.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/map/project.rs
@@ -13,10 +13,12 @@
 //! See the License for the specific language governing permissions and
 //! limitations under the License.
 
-use std::convert::TryFrom;
+use std::collections::BTreeMap;
+use std::convert::{TryFrom, TryInto};
 
 use common_pb::path_concat::Endpoint;
 use dyn_type::Object;
+use graph_proxy::apis::PropKey;
 use graph_proxy::utils::expr::eval::{Evaluate, Evaluator};
 use ir_common::error::ParsePbError;
 use ir_common::generated::common as common_pb;
@@ -26,14 +28,29 @@ use pegasus::api::function::{FilterMapFunction, FnResult};
 
 use crate::error::FnExecError;
 use crate::error::{FnExecResult, FnGenResult};
-use crate::process::entry::CollectionEntry;
 use crate::process::entry::DynEntry;
 use crate::process::entry::Entry;
 use crate::process::entry::PairEntry;
+use crate::process::entry::{CollectionEntry, EntryType};
 use crate::process::operator::map::FilterMapFuncGen;
 use crate::process::operator::TagKey;
 use crate::process::record::Record;
 
+#[derive(Debug)]
+enum Projector {
+    ExprProjector(Evaluator),
+    GraphElementProjector(TagKey),
+    /// VecProjector will output a collection entry, which is a collection of projected graph elements (computed via TagKey).
+    VecProjector(Vec<TagKey>),
+    /// MapProjector will output a collection entry, which is a collection of key-value pairs. The key is a Object (preserve the user-given key), and the value is a projected graph element (computed via TagKey).
+    /// Besides, MapProjector supports nested map.
+    MapProjector(VariableKeyValues),
+    /// A simple concatenation of two paths.
+    PathConcatProjector((TagKey, Endpoint), (TagKey, Endpoint)),
+    /// PathValueProjector will output a collection of projected properties of the element in the path.
+    PathValueProjector(PathTagKeyValues),
+}
+
 /// Project entries with specified tags or further their properties.
 /// Notice that when projecting a single column, if the result is a None-Entry,
 /// Caused by either the given `tag` or the required properties do not exist, the record will be filtered.
@@ -46,6 +63,7 @@ struct ProjectOperator {
 #[derive(Debug)]
 enum VariableValue {
     Value(TagKey),
+    PathFunc(PathTagKeyValues),
     Nest(VariableKeyValues),
 }
 
@@ -68,6 +86,7 @@ impl VariableKeyValues {
             let value = match &kv.value {
                 VariableValue::Value(tag_key) => tag_key.get_arc_entry(input)?,
                 VariableValue::Nest(nest) => nest.exec_projector(input)?,
+                VariableValue::PathFunc(path_func) => path_func.exec_projector(input)?,
             };
             map_collection.push(PairEntry::new(key.into(), value).into());
         }
@@ -76,16 +95,107 @@ impl VariableKeyValues {
 }
 
 #[derive(Debug)]
-enum Projector {
-    ExprProjector(Evaluator),
-    GraphElementProjector(TagKey),
-    /// VecProjector will output a collection entry, which is a collection of projected graph elements (computed via TagKey).
-    VecProjector(Vec<TagKey>),
-    /// MapProjector will output a collection entry, which is a collection of key-value pairs. The key is a Object (preserve the user-given key), and the value is a projected graph element (computed via TagKey).
-    /// Besides, MapProjector supports nested map.
-    MapProjector(VariableKeyValues),
-    /// A simple concatenation of two paths.
-    PathConcatProjector((TagKey, Endpoint), (TagKey, Endpoint)),
+enum PathKey {
+    Property(PropKey),
+    Vec(Vec<PropKey>),
+    Map(Vec<(Object, PropKey)>),
+}
+
+impl PathKey {
+    // get the properties of the elements in path according to the path key.
+    // The properties are returned as a Object::Vector, e.g., project a.name where a is a path, the result is a vector of names.
+    fn get_key(&self, entry: &DynEntry) -> FnExecResult<Object> {
+        match self {
+            PathKey::Property(prop_key) => Ok(prop_key.get_key(entry)?),
+            PathKey::Vec(vec) => {
+                let prop_num = vec.len();
+                if prop_num == 0 {
+                    warn!("Empty Path Properties in PathKey::Vec");
+                    return Ok(Object::Vector(vec![]));
+                }
+                let prob_props = self.get_path_props(entry, &vec[0])?;
+                let mut prop_collection: Vec<Vec<Object>> = prob_props
+                    .into_iter()
+                    .map(|prop| {
+                        let mut inner_vec = Vec::with_capacity(prop_num);
+                        inner_vec.push(prop);
+                        inner_vec.extend((1..prop_num).map(|_| Object::None));
+                        inner_vec
+                    })
+                    .collect();
+                for (prop_idx, prop_key) in vec.into_iter().enumerate().skip(1) {
+                    let props = self.get_path_props(entry, &prop_key)?;
+                    for (path_idx, prop) in props.into_iter().enumerate() {
+                        prop_collection[path_idx][prop_idx] = prop;
+                    }
+                }
+                Ok(Object::Vector(
+                    prop_collection
+                        .into_iter()
+                        .map(|vec| Object::Vector(vec))
+                        .collect(),
+                ))
+            }
+            PathKey::Map(map) => {
+                let prop_num = map.len();
+                if prop_num == 0 {
+                    warn!("Empty Path Properties in PathKey::Vec");
+                    return Ok(Object::Vector(vec![]));
+                }
+                let prob_key = &map[0].0;
+                let prob_props = self.get_path_props(entry, &map[0].1)?;
+                let mut prop_collection = Vec::with_capacity(prob_props.len());
+                for prop_val in prob_props.into_iter() {
+                    let mut btree_map = BTreeMap::new();
+                    btree_map.insert(prob_key.clone(), prop_val);
+                    prop_collection.push(btree_map);
+                }
+                for (key_name, prop_key) in map.into_iter().skip(1) {
+                    let props = prop_key.get_key(entry)?.take_vector().unwrap();
+                    for (path_idx, prop) in props.into_iter().enumerate() {
+                        prop_collection[path_idx].insert(key_name.clone(), prop);
+                    }
+                }
+                Ok(Object::Vector(
+                    prop_collection
+                        .into_iter()
+                        .map(|map| Object::KV(map))
+                        .collect(),
+                ))
+            }
+        }
+    }
+
+    fn get_path_props(&self, entry: &DynEntry, prop_key: &PropKey) -> FnExecResult<Vec<Object>> {
+        Ok(prop_key
+            .get_key(entry)?
+            .take_vector()
+            .map_err(|e| FnExecError::ExprEvalError(e.into()))?)
+    }
+}
+
+#[derive(Debug)]
+struct PathTagKeyValues {
+    tag: Option<KeyId>,
+    val: PathKey,
+    // TODO: support function options.
+    // Currently, if the path is ALLV, return the properties of vertices; if the path is ALLVE, return the properties of both vertices and edges.
+    _opt: common_pb::path_function::FuncOpt,
+}
+
+impl PathTagKeyValues {
+    fn exec_projector(&self, input: &Record) -> FnExecResult<DynEntry> {
+        if let Some(entry) = input.get(self.tag) {
+            if EntryType::Path != entry.get_type() {
+                Err(FnExecError::unexpected_data_error("Apply PathTagKeyValues on a non-Path entry"))
+            } else {
+                let projected_properties_obj = self.val.get_key(&entry)?;
+                Ok(projected_properties_obj.into())
+            }
+        } else {
+            Ok(DynEntry::new(Object::Vector(vec![])))
+        }
+    }
 }
 
 // TODO:
@@ -183,6 +293,7 @@ fn exec_projector(input: &Record, projector: &Projector) -> FnExecResult<DynEntr
                 DynEntry::new(left_path)
             }
         }
+        Projector::PathValueProjector(path) => path.exec_projector(input)?,
     };
     Ok(entry)
 }
@@ -301,6 +412,13 @@ impl FilterMapFuncGen for pb::Project {
                             (TagKey::try_from(right_path_tag)?, right_endpoint),
                         )
                     }
+                    common_pb::ExprOpr {
+                        item: Some(common_pb::expr_opr::Item::PathFunc(path_func)),
+                        ..
+                    } => {
+                        let path_key_values = PathTagKeyValues::try_from(path_func.clone())?;
+                        Projector::PathValueProjector(path_key_values)
+                    }
                     _ => {
                         let evaluator = Evaluator::try_from(expr)?;
                         Projector::ExprProjector(evaluator)
@@ -341,6 +459,10 @@ impl TryFrom<common_pb::VariableKeyValues> for VariableKeyValues {
                         let nested = VariableKeyValues::try_from(nested_vals)?;
                         VariableValue::Nest(nested)
                     }
+                    common_pb::variable_key_value::Value::PathFunc(path_func) => {
+                        let path_key_values = path_func.try_into()?;
+                        VariableValue::PathFunc(path_key_values)
+                    }
                 }
             } else {
                 return Err(ParsePbError::from("empty value provided in Map"));
@@ -351,6 +473,50 @@ impl TryFrom<common_pb::VariableKeyValues> for VariableKeyValues {
     }
 }
 
+impl TryFrom<common_pb::PathFunction> for PathTagKeyValues {
+    type Error = ParsePbError;
+
+    fn try_from(path_func: common_pb::PathFunction) -> Result<Self, Self::Error> {
+        let path_key = path_func
+            .path_key
+            .ok_or_else(|| ParsePbError::from("empty path key"))?;
+        let func_opt = unsafe { std::mem::transmute(path_func.opt) };
+        let path_key_values = PathTagKeyValues {
+            tag: path_func
+                .tag
+                .map(|tag| KeyId::try_from(tag))
+                .transpose()?,
+            val: match path_key {
+                common_pb::path_function::PathKey::Property(prop) => {
+                    PathKey::Property(PropKey::try_from(prop)?)
+                }
+                common_pb::path_function::PathKey::Vars(vars) => PathKey::Vec(
+                    vars.keys
+                        .into_iter()
+                        .map(|prop| PropKey::try_from(prop))
+                        .collect::<Result<Vec<PropKey>, _>>()?,
+                ),
+                common_pb::path_function::PathKey::Map(map) => PathKey::Map(
+                    map.key_vals
+                        .into_iter()
+                        .map(|key_val| {
+                            let key = Object::try_from(key_val.key.unwrap());
+                            let value = PropKey::try_from(key_val.val.unwrap());
+                            if key.is_ok() && value.is_ok() {
+                                Ok((key.unwrap(), value.unwrap()))
+                            } else {
+                                Err(ParsePbError::from("invalid key-value pair in Map"))
+                            }
+                        })
+                        .collect::<Result<Vec<(Object, PropKey)>, _>>()?,
+                ),
+            },
+            _opt: func_opt,
+        };
+        Ok(path_key_values)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::vec;
@@ -370,8 +536,8 @@ mod tests {
     use crate::process::operator::map::FilterMapFuncGen;
     use crate::process::operator::tests::{
         init_source, init_source_with_multi_tags, init_source_with_tag, init_vertex1, init_vertex2,
-        to_expr_map_pb, to_expr_var_pb, to_expr_vars_pb, to_var_pb, PERSON_LABEL, TAG_A, TAG_B, TAG_C,
-        TAG_D, TAG_E, TAG_F, TAG_G,
+        to_expr_map_pb, to_expr_var_pb, to_expr_vars_pb, to_prop_pb, to_var_pb, PERSON_LABEL, TAG_A, TAG_B,
+        TAG_C, TAG_D, TAG_E, TAG_F, TAG_G,
     };
     use crate::process::record::Record;
 
@@ -1564,4 +1730,134 @@ mod tests {
             assert!(res.is_err());
         }
     }
+
+    fn init_path_record() -> Record {
+        let vertex1 = init_vertex1();
+        let vertex2 = init_vertex2();
+        let mut path =
+            GraphPath::new(vertex1, pb::path_expand::PathOpt::Arbitrary, pb::path_expand::ResultOpt::AllV);
+        path.append(vertex2);
+        Record::new(path, None)
+    }
+
+    fn to_path_func_pb(
+        tag: Option<NameOrId>, path_key: common_pb::path_function::PathKey,
+        opt: common_pb::path_function::FuncOpt,
+    ) -> common_pb::PathFunction {
+        common_pb::PathFunction {
+            tag: tag.map(|t| t.into()),
+            opt: opt as i32,
+            node_type: None,
+            path_key: Some(path_key),
+        }
+    }
+
+    fn to_expr_path_func_pb(
+        tag: Option<NameOrId>, path_key: common_pb::path_function::PathKey,
+        opt: common_pb::path_function::FuncOpt,
+    ) -> common_pb::Expression {
+        common_pb::Expression {
+            operators: vec![common_pb::ExprOpr {
+                node_type: None,
+                item: Some(common_pb::expr_opr::Item::PathFunc(to_path_func_pb(tag, path_key, opt))),
+            }],
+        }
+    }
+
+    // g.V().out(2..3).values('name')
+    #[test]
+    fn project_path_prop_test() {
+        let path_key = common_pb::path_function::PathKey::Property(to_prop_pb("name".into()));
+        let project_opr_pb = pb::Project {
+            mappings: vec![pb::project::ExprAlias {
+                expr: Some(to_expr_path_func_pb(None, path_key, common_pb::path_function::FuncOpt::Vertex)),
+                alias: Some(TAG_A.into()),
+            }],
+            is_append: false,
+        };
+        let mut result = project_test(vec![init_path_record()], project_opr_pb);
+
+        let mut object_result = Object::None;
+        if let Some(Ok(res)) = result.next() {
+            let a_entry = res.get(Some(TAG_A));
+            object_result = a_entry.unwrap().as_object().unwrap().clone();
+        }
+        assert!(result.next().is_none());
+        let expected_result = Object::Vector(vec![object!("marko"), object!("vadas")]);
+        assert_eq!(object_result, expected_result);
+    }
+
+    // g.V().out(2..3).values('name','age')
+    #[test]
+    fn project_path_prop_vars_test() {
+        let path_key = common_pb::path_function::PathKey::Vars(common_pb::path_function::PathElementKeys {
+            keys: vec![to_prop_pb("name".into()), to_prop_pb("age".into())],
+        });
+        let project_opr_pb = pb::Project {
+            mappings: vec![pb::project::ExprAlias {
+                expr: Some(to_expr_path_func_pb(None, path_key, common_pb::path_function::FuncOpt::Vertex)),
+                alias: Some(TAG_A.into()),
+            }],
+            is_append: false,
+        };
+        let mut result = project_test(vec![init_path_record()], project_opr_pb);
+        let mut object_result = Object::None;
+        if let Some(Ok(res)) = result.next() {
+            let a_entry = res.get(Some(TAG_A));
+            object_result = a_entry.unwrap().as_object().unwrap().clone();
+        }
+        assert!(result.next().is_none());
+        let expected_result = Object::Vector(vec![
+            Object::Vector(vec![object!("marko"), object!(29)]),
+            Object::Vector(vec![object!("vadas"), object!(27)]),
+        ]);
+        assert_eq!(object_result, expected_result);
+    }
+
+    // g.V().out(2..3).valueMap('name','age')
+    #[test]
+    fn project_path_prop_map_test() {
+        let path_key =
+            common_pb::path_function::PathKey::Map(common_pb::path_function::PathElementKeyValues {
+                key_vals: vec![
+                    common_pb::path_function::path_element_key_values::PathElementKeyValue {
+                        key: Some("name".to_string().into()),
+                        val: Some(to_prop_pb("name".into())),
+                    },
+                    common_pb::path_function::path_element_key_values::PathElementKeyValue {
+                        key: Some("age".to_string().into()),
+                        val: Some(to_prop_pb("age".into())),
+                    },
+                ],
+            });
+        let project_opr_pb = pb::Project {
+            mappings: vec![pb::project::ExprAlias {
+                expr: Some(to_expr_path_func_pb(None, path_key, common_pb::path_function::FuncOpt::Vertex)),
+                alias: Some(TAG_A.into()),
+            }],
+            is_append: false,
+        };
+        let mut result = project_test(vec![init_path_record()], project_opr_pb);
+
+        let mut object_result = Object::None;
+        if let Some(Ok(res)) = result.next() {
+            let a_entry = res.get(Some(TAG_A));
+            println!("{:?}", a_entry);
+            object_result = a_entry.unwrap().as_object().unwrap().clone();
+        }
+        assert!(result.next().is_none());
+        let expected_result = Object::Vector(vec![
+            Object::KV(
+                vec![(object!("name"), object!("marko")), (object!("age"), object!(29))]
+                    .into_iter()
+                    .collect(),
+            ),
+            Object::KV(
+                vec![(object!("name"), object!("vadas")), (object!("age"), object!(27))]
+                    .into_iter()
+                    .collect(),
+            ),
+        ]);
+        assert_eq!(object_result, expected_result);
+    }
 }
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/mod.rs b/interactive_engine/executor/ir/runtime/src/process/operator/mod.rs
index 9428cdb72cda..c069d479755f 100644
--- a/interactive_engine/executor/ir/runtime/src/process/operator/mod.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/mod.rs
@@ -29,13 +29,13 @@ pub mod subtask;
 use std::convert::TryFrom;
 
 use dyn_type::Object;
-use graph_proxy::apis::{Element, PropKey};
+use graph_proxy::apis::PropKey;
 use ir_common::error::ParsePbError;
 use ir_common::generated::common as common_pb;
-use ir_common::{KeyId, NameOrId};
+use ir_common::KeyId;
 use pegasus::codec::{Decode, Encode, ReadExt, WriteExt};
 
-use crate::error::{FnExecError, FnExecResult};
+use crate::error::FnExecResult;
 use crate::process::entry::DynEntry;
 use crate::process::record::Record;
 
@@ -50,8 +50,8 @@ impl TagKey {
     pub fn get_arc_entry(&self, input: &Record) -> FnExecResult<DynEntry> {
         if let Some(entry) = input.get(self.tag) {
             if let Some(prop_key) = self.key.as_ref() {
-                let prop = self.get_key(entry, prop_key)?;
-                Ok(prop)
+                let prop = prop_key.get_key(entry)?;
+                Ok(DynEntry::new(prop))
             } else {
                 Ok(entry.clone())
             }
@@ -59,58 +59,6 @@ impl TagKey {
             Ok(DynEntry::new(Object::None))
         }
     }
-
-    fn get_key(&self, entry: &DynEntry, prop_key: &PropKey) -> FnExecResult<DynEntry> {
-        if let PropKey::Len = prop_key {
-            let obj: Object = (entry.len() as u64).into();
-            Ok(DynEntry::new(obj))
-        } else {
-            if let Some(element) = entry.as_graph_element() {
-                let prop_obj = match prop_key {
-                    PropKey::Id => element.id().into(),
-                    PropKey::Label => element
-                        .label()
-                        .map(|label| label.into())
-                        .unwrap_or(Object::None),
-                    PropKey::Len => unreachable!(),
-                    PropKey::All => {
-                        if let Some(properties) = element.get_all_properties() {
-                            properties
-                                .into_iter()
-                                .map(|(key, value)| {
-                                    let obj_key: Object = match key {
-                                        NameOrId::Str(str) => str.into(),
-                                        NameOrId::Id(id) => id.into(),
-                                    };
-                                    (obj_key, value)
-                                })
-                                .collect::<Vec<(Object, Object)>>()
-                                .into()
-                        } else {
-                            Object::None
-                        }
-                    }
-                    PropKey::Key(key) => {
-                        if let Some(properties) = element.get_property(key) {
-                            properties.try_to_owned().ok_or_else(|| {
-                                FnExecError::unexpected_data_error("unable to own the `BorrowObject`")
-                            })?
-                        } else {
-                            Object::None
-                        }
-                    }
-                };
-
-                Ok(DynEntry::new(prop_obj))
-            } else {
-                Err(FnExecError::unexpected_data_error(&format!(
-                    "
-                Get key failed since get details from a none-graph element {:?} ",
-                    entry
-                )))
-            }
-        }
-    }
 }
 
 impl TryFrom<common_pb::Variable> for TagKey {
@@ -175,7 +123,7 @@ pub(crate) mod tests {
     use ahash::HashMap;
     use dyn_type::Object;
     use graph_proxy::apis::{DynDetails, GraphElement, Vertex};
-    use ir_common::{KeyId, LabelId};
+    use ir_common::{KeyId, LabelId, NameOrId};
 
     use super::*;
     use crate::process::entry::Entry;
@@ -245,11 +193,14 @@ pub(crate) mod tests {
         vec![r1]
     }
 
+    pub fn to_prop_pb(key: NameOrId) -> common_pb::Property {
+        common_pb::Property { item: Some(common_pb::property::Item::Key(key.into())) }
+    }
+
     pub fn to_var_pb(tag: Option<NameOrId>, key: Option<NameOrId>) -> common_pb::Variable {
         common_pb::Variable {
             tag: tag.map(|t| t.into()),
-            property: key
-                .map(|k| common_pb::Property { item: Some(common_pb::property::Item::Key(k.into())) }),
+            property: key.map(|k| to_prop_pb(k)),
             node_type: None,
         }
     }
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/sink/sink.rs b/interactive_engine/executor/ir/runtime/src/process/operator/sink/sink.rs
index 4e3227c5820d..93e61c09baae 100644
--- a/interactive_engine/executor/ir/runtime/src/process/operator/sink/sink.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/sink/sink.rs
@@ -115,6 +115,8 @@ impl RecordSinkEncoder {
             _ => {
                 if let Some(map_pb) = self.try_map_to_pb(e) {
                     Some(result_pb::entry::Inner::Map(map_pb))
+                } else if let Some(collection_pb) = self.try_collection_to_pb(e) {
+                    Some(result_pb::entry::Inner::Collection(collection_pb))
                 } else {
                     let element_pb = self.element_to_pb(e);
                     Some(result_pb::entry::Inner::Element(element_pb))
@@ -124,6 +126,24 @@ impl RecordSinkEncoder {
         Ok(result_pb::Entry { inner })
     }
 
+    // return if the given entry is a collection entry result from PathValueProjector eval, etc.
+    fn try_collection_to_pb(&self, e: &DynEntry) -> Option<result_pb::Collection> {
+        if let EntryType::Object = e.get_type() {
+            if let Object::Vector(vec) = e.as_object().unwrap() {
+                let mut collection_pb = Vec::with_capacity(vec.len());
+                for obj in vec {
+                    let obj_pb = self.object_to_pb(obj.clone());
+                    let element_pb =
+                        result_pb::Element { inner: Some(result_pb::element::Inner::Object(obj_pb)) };
+                    collection_pb.push(element_pb);
+                }
+                return Some(result_pb::Collection { collection: collection_pb });
+            }
+        }
+
+        None
+    }
+
     // return if the given entry is a map entry result from Map eval.
     fn try_map_to_pb(&self, e: &DynEntry) -> Option<result_pb::KeyValues> {
         if let EntryType::Object = e.get_type() {
@@ -140,9 +160,11 @@ impl RecordSinkEncoder {
                     let val_pb: common_pb::Value = val.clone().into();
                     key_values.push(result_pb::key_values::KeyValue {
                         key: Some(key_pb),
-                        value: Some(result_pb::key_values::key_value::Value::Element(result_pb::Element {
-                            inner: Some(result_pb::element::Inner::Object(val_pb)),
-                        })),
+                        value: Some(result_pb::Entry {
+                            inner: Some(result_pb::entry::Inner::Element(result_pb::Element {
+                                inner: Some(result_pb::element::Inner::Object(val_pb)),
+                            })),
+                        }),
                     })
                 }
                 return Some(result_pb::KeyValues { key_values });
@@ -167,17 +189,31 @@ impl RecordSinkEncoder {
                         .as_any_ref()
                         .downcast_ref::<CollectionEntry>()
                         .unwrap();
-                    let inner_collection_pb = self.collection_map_to_pb(inner_collection.clone())?;
+                    let inner_map_pb = self.collection_map_to_pb(inner_collection.clone())?;
                     key_values.push(result_pb::key_values::KeyValue {
                         key: Some(key_pb),
-                        value: Some(result_pb::key_values::key_value::Value::Nested(inner_collection_pb)),
+                        value: Some(result_pb::Entry {
+                            inner: Some(result_pb::entry::Inner::Map(inner_map_pb)),
+                        }),
                     })
                 } else {
-                    let val_pb = self.element_to_pb(pair.get_right());
-                    key_values.push(result_pb::key_values::KeyValue {
-                        key: Some(key_pb),
-                        value: Some(result_pb::key_values::key_value::Value::Element(val_pb)),
-                    })
+                    let right = pair.get_right();
+                    if let Some(collection) = self.try_collection_to_pb(right) {
+                        key_values.push(result_pb::key_values::KeyValue {
+                            key: Some(key_pb),
+                            value: Some(result_pb::Entry {
+                                inner: Some(result_pb::entry::Inner::Collection(collection)),
+                            }),
+                        });
+                    } else {
+                        let val_pb = self.element_to_pb(right);
+                        key_values.push(result_pb::key_values::KeyValue {
+                            key: Some(key_pb),
+                            value: Some(result_pb::Entry {
+                                inner: Some(result_pb::entry::Inner::Element(val_pb)),
+                            }),
+                        })
+                    }
                 }
             } else {
                 Err(FnExecError::unsupported_error(&format!(


Test patch:
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/gremlin/antlr4x/GraphBuilderTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/gremlin/antlr4x/GraphBuilderTest.java
index 467317573bb8..01bd4e217ea4 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/gremlin/antlr4x/GraphBuilderTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/gremlin/antlr4x/GraphBuilderTest.java
@@ -427,7 +427,7 @@ public void g_V_path_expand_test() {
                     + " tables=[knows]}], alias=[_], opt=[OUT])\n"
                     + "], getV=[GraphLogicalGetV(tableConfig=[{isAll=true, tables=[software,"
                     + " person]}], alias=[_], opt=[END])\n"
-                    + "], offset=[1], fetch=[1], path_opt=[SIMPLE], result_opt=[ALL_V_E],"
+                    + "], offset=[1], fetch=[1], path_opt=[SIMPLE], result_opt=[ALL_V],"
                     + " alias=[b])\n"
                     + "      GraphLogicalSource(tableConfig=[{isAll=true, tables=[software,"
                     + " person]}], alias=[a], opt=[VERTEX])",
@@ -1736,4 +1736,42 @@ public void g_V_select_expr_property_id() {
                         + " person]}], alias=[_], opt=[VERTEX])",
                 rel.explain().trim());
     }
+
+    @Test
+    public void g_V_path_as_a_select_a_valueMap() {
+        RelNode rel =
+                eval(
+                        "g.V().out('2..3').with('RESULT_OPT',"
+                                + " 'ALL_V_E').as('a').select('a').valueMap('~len')");
+        Assert.assertEquals(
+                "GraphLogicalProject($f0=[$f0], isAppend=[false])\n"
+                    + "  GraphLogicalProject($f0=[MAP(_UTF-8'~len', a.~len)], isAppend=[true])\n"
+                    + "    GraphLogicalPathExpand(expand=[GraphLogicalExpand(tableConfig=[{isAll=true,"
+                    + " tables=[created, knows]}], alias=[_], opt=[OUT])\n"
+                    + "], getV=[GraphLogicalGetV(tableConfig=[{isAll=true, tables=[software,"
+                    + " person]}], alias=[_], opt=[END])\n"
+                    + "], offset=[2], fetch=[1], path_opt=[ARBITRARY], result_opt=[ALL_V_E],"
+                    + " alias=[a])\n"
+                    + "      GraphLogicalSource(tableConfig=[{isAll=true, tables=[software,"
+                    + " person]}], alias=[_], opt=[VERTEX])",
+                rel.explain().trim());
+
+        rel =
+                eval(
+                        "g.V().out('2..3').with('RESULT_OPT',"
+                                + " 'ALL_V_E').as('a').select('a').valueMap('name', 'weight')");
+        Assert.assertEquals(
+                "GraphLogicalProject($f0=[$f0], isAppend=[false])\n"
+                    + "  GraphLogicalProject($f0=[PATH_FUNCTION(a, FLAG(VERTEX_EDGE),"
+                    + " MAP(_UTF-8'weight', a.weight, _UTF-8'name', a.name))], isAppend=[true])\n"
+                    + "    GraphLogicalPathExpand(expand=[GraphLogicalExpand(tableConfig=[{isAll=true,"
+                    + " tables=[created, knows]}], alias=[_], opt=[OUT])\n"
+                    + "], getV=[GraphLogicalGetV(tableConfig=[{isAll=true, tables=[software,"
+                    + " person]}], alias=[_], opt=[END])\n"
+                    + "], offset=[2], fetch=[1], path_opt=[ARBITRARY], result_opt=[ALL_V_E],"
+                    + " alias=[a])\n"
+                    + "      GraphLogicalSource(tableConfig=[{isAll=true, tables=[software,"
+                    + " person]}], alias=[_], opt=[VERTEX])",
+                rel.explain().trim());
+    }
 }
diff --git a/interactive_engine/executor/ir/integrated/tests/pathxd_test.rs b/interactive_engine/executor/ir/integrated/tests/pathxd_test.rs
index 94f79d088c3f..e77e2ed75075 100644
--- a/interactive_engine/executor/ir/integrated/tests/pathxd_test.rs
+++ b/interactive_engine/executor/ir/integrated/tests/pathxd_test.rs
@@ -23,9 +23,11 @@ mod test {
     use graph_proxy::apis::{Element, GraphElement, ID};
     use ir_common::expr_parse::str_to_expr_pb;
     use ir_common::generated::algebra as pb;
+    use ir_common::generated::common as common_pb;
     use ir_physical_client::physical_builder::*;
+    use pegasus_common::downcast::AsAny;
     use pegasus_server::JobRequest;
-    use runtime::process::entry::Entry;
+    use runtime::process::entry::{CollectionEntry, Entry};
 
     use crate::common::test::*;
 
@@ -954,8 +956,16 @@ mod test {
             match result {
                 Ok(res) => {
                     let entry = parse_result(res).unwrap();
-                    if let Some(properties) = entry.get(None).unwrap().as_object() {
-                        result_collection.push(properties.clone());
+                    if let Some(e) = entry.get(None) {
+                        let collection = e
+                            .as_any_ref()
+                            .downcast_ref::<CollectionEntry>()
+                            .unwrap();
+                        let mut path_values_collection = vec![];
+                        for v in collection.inner.iter() {
+                            path_values_collection.push(v.as_object().unwrap().clone());
+                        }
+                        result_collection.push(Object::Vector(path_values_collection));
                     }
                 }
                 Err(e) => {
@@ -1179,4 +1189,284 @@ mod test {
         result_collection.sort();
         assert_eq!(result_collection, expected_result_collection);
     }
+
+    // g.V().hasLabel("person").both("2..3", "knows").values("xx"), where values("xx") would be computed via PathFunc
+    fn init_path_expand_project_path_func_request(
+        result_opt: i32, path_key: common_pb::path_function::PathKey,
+    ) -> JobRequest {
+        let source_opr = pb::Scan {
+            scan_opt: 0,
+            alias: None,
+            params: Some(query_params(vec![PERSON_LABEL.into()], vec![], None)),
+            idx_predicate: None,
+            is_count_only: false,
+            meta_data: None,
+        };
+
+        let edge_expand = pb::EdgeExpand {
+            v_tag: None,
+            direction: 2,
+            params: Some(query_params(vec![KNOWS_LABEL.into()], vec![], None)),
+            expand_opt: 0,
+            alias: None,
+            meta_data: None,
+            is_optional: false,
+        };
+
+        let path_expand_opr = pb::PathExpand {
+            base: Some(edge_expand.into()),
+            start_tag: None,
+            alias: None,
+            hop_range: Some(pb::Range { lower: 2, upper: 3 }),
+            path_opt: 0, // Arbitrary
+            result_opt,
+            condition: None,
+            is_optional: false,
+        };
+
+        // to project path.name
+        let project_opr = pb::Project {
+            mappings: vec![pb::project::ExprAlias {
+                expr: Some(common_pb::Expression {
+                    operators: vec![common_pb::ExprOpr {
+                        node_type: None,
+                        item: Some(common_pb::expr_opr::Item::PathFunc(common_pb::PathFunction {
+                            tag: None,
+                            opt: 0, // opt is not supported for now
+                            node_type: None,
+                            path_key: Some(path_key),
+                        })),
+                    }],
+                }),
+                alias: Some(TAG_A.into()),
+            }],
+            is_append: false,
+            meta_data: vec![],
+        };
+
+        let mut job_builder = JobBuilder::default();
+        job_builder.add_scan_source(source_opr);
+        job_builder.shuffle(None);
+        job_builder.path_expand(path_expand_opr);
+        job_builder.project(project_opr);
+        job_builder.sink(default_sink_pb());
+
+        job_builder.build().unwrap()
+    }
+
+    // g.V().hasLabel("person").both("2..3", "knows").with("RESULT_OPT", "ALL_V").values("name"),
+    #[test]
+    fn path_expand_allv_project_path_func_test() {
+        initialize();
+        let path_key = common_pb::path_function::PathKey::Property(common_pb::Property {
+            item: Some(common_pb::property::Item::Key("name".into())),
+        });
+        let request = init_path_expand_project_path_func_request(1, path_key); // all v
+        let mut results = submit_query(request, 2);
+        let mut result_collection: Vec<Object> = vec![];
+        let mut expected_result_paths: Vec<Object> = vec![
+            Object::Vector(vec![object!("marko"), object!("vadas"), object!("marko")]),
+            Object::Vector(vec![object!("marko"), object!("josh"), object!("marko")]),
+            Object::Vector(vec![object!("vadas"), object!("marko"), object!("vadas")]),
+            Object::Vector(vec![object!("vadas"), object!("marko"), object!("josh")]),
+            Object::Vector(vec![object!("josh"), object!("marko"), object!("vadas")]),
+            Object::Vector(vec![object!("josh"), object!("marko"), object!("josh")]),
+        ];
+
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    if let Some(e) = entry.get(None) {
+                        let collection = e
+                            .as_any_ref()
+                            .downcast_ref::<CollectionEntry>()
+                            .unwrap();
+                        let mut path_values_collection = vec![];
+                        for v in collection.inner.iter() {
+                            path_values_collection.push(v.as_object().unwrap().clone());
+                        }
+                        result_collection.push(Object::Vector(path_values_collection));
+                    }
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        expected_result_paths.sort();
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_paths);
+    }
+
+    // g.V().hasLabel("person").both("2..3", "knows").with("RESULT_OPT", "END_V").values("name"),
+    #[test]
+    fn path_expand_endv_project_path_func_test() {
+        initialize();
+        let path_key = common_pb::path_function::PathKey::Property(common_pb::Property {
+            item: Some(common_pb::property::Item::Key("name".into())),
+        });
+        let request = init_path_expand_project_path_func_request(0, path_key); // endv
+        let mut results = submit_query(request, 2);
+        let mut result_collection: Vec<Object> = vec![];
+        let mut expected_result_paths: Vec<Object> = vec![
+            Object::Vector(vec![object!("marko")]),
+            Object::Vector(vec![object!("marko")]),
+            Object::Vector(vec![object!("vadas")]),
+            Object::Vector(vec![object!("josh")]),
+            Object::Vector(vec![object!("vadas")]),
+            Object::Vector(vec![object!("josh")]),
+        ];
+
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    if let Some(e) = entry.get(None) {
+                        let collection = e
+                            .as_any_ref()
+                            .downcast_ref::<CollectionEntry>()
+                            .unwrap();
+                        let mut path_values_collection = vec![];
+                        for v in collection.inner.iter() {
+                            path_values_collection.push(v.as_object().unwrap().clone());
+                        }
+                        result_collection.push(Object::Vector(path_values_collection));
+                    }
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        expected_result_paths.sort();
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_paths);
+    }
+
+    // g.V().hasLabel("person").both("2..3", "knows").with("RESULT_OPT", "ALL_VE").values("name", "weight"),
+    #[test]
+    fn path_expand_allve_project_path_func_test() {
+        initialize();
+        let path_key = common_pb::path_function::PathKey::Vars(common_pb::path_function::PathElementKeys {
+            keys: vec![
+                common_pb::Property { item: Some(common_pb::property::Item::Key("name".into())) },
+                common_pb::Property { item: Some(common_pb::property::Item::Key("weight".into())) },
+            ],
+        });
+        let request = init_path_expand_project_path_func_request(2, path_key); // all ve
+        let mut results = submit_query(request, 2);
+        let mut result_collection: Vec<Object> = vec![];
+        let not_exist = object!("");
+        let mut expected_result_paths: Vec<Object> = vec![
+            // marko - 0.5 - vadas - 0.5 - marko
+            Object::Vector(
+                vec![
+                    // e.g., for the vertex element, name: marko, weight, None
+                    vec![object!("marko"), not_exist.clone()],
+                    // e.g., for the edge element, name: None, weight: 0.5
+                    vec![not_exist.clone(), object!("0.5")],
+                    vec![object!("vadas"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("0.5")],
+                    vec![object!("marko"), not_exist.clone()],
+                ]
+                .into_iter()
+                .map(|vec_obj| Object::Vector(vec_obj))
+                .collect(),
+            ),
+            // marko - 1 - josh - 1 - marko
+            Object::Vector(
+                vec![
+                    vec![object!("marko"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("1")],
+                    vec![object!("josh"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("1")],
+                    vec![object!("marko"), not_exist.clone()],
+                ]
+                .into_iter()
+                .map(|vec_obj| Object::Vector(vec_obj))
+                .collect(),
+            ),
+            // vadas - 0.5 - marko - 0.5 - vadas
+            Object::Vector(
+                vec![
+                    vec![object!("vadas"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("0.5")],
+                    vec![object!("marko"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("0.5")],
+                    vec![object!("vadas"), not_exist.clone()],
+                ]
+                .into_iter()
+                .map(|vec_obj| Object::Vector(vec_obj))
+                .collect(),
+            ),
+            // vadas - 0.5 - marko - 1 - josh
+            Object::Vector(
+                vec![
+                    // e.g., for the vertex element, name: marko, weight, None
+                    vec![object!("vadas"), not_exist.clone()],
+                    // e.g., for the edge element, name: None, weight: 0.5
+                    vec![not_exist.clone(), object!("0.5")],
+                    vec![object!("marko"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("1")],
+                    vec![object!("josh"), not_exist.clone()],
+                ]
+                .into_iter()
+                .map(|vec_obj| Object::Vector(vec_obj))
+                .collect(),
+            ),
+            // josh - 1 - marko - 0.5 - vadas
+            Object::Vector(
+                vec![
+                    vec![object!("josh"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("1")],
+                    vec![object!("marko"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("0.5")],
+                    vec![object!("vadas"), not_exist.clone()],
+                ]
+                .into_iter()
+                .map(|vec_obj| Object::Vector(vec_obj))
+                .collect(),
+            ),
+            // josh - 1 - marko - 1 - josh
+            Object::Vector(
+                vec![
+                    vec![object!("josh"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("1")],
+                    vec![object!("marko"), not_exist.clone()],
+                    vec![not_exist.clone(), object!("1")],
+                    vec![object!("josh"), not_exist.clone()],
+                ]
+                .into_iter()
+                .map(|vec_obj| Object::Vector(vec_obj))
+                .collect(),
+            ),
+        ];
+
+        while let Some(result) = results.next() {
+            match result {
+                Ok(res) => {
+                    let entry = parse_result(res).unwrap();
+                    if let Some(e) = entry.get(None) {
+                        println!("entry {:?}", e);
+                        let collection = e
+                            .as_any_ref()
+                            .downcast_ref::<CollectionEntry>()
+                            .unwrap();
+                        let mut path_values_collection = vec![];
+                        for v in collection.inner.iter() {
+                            path_values_collection.push(v.as_object().unwrap().clone());
+                        }
+                        result_collection.push(Object::Vector(path_values_collection));
+                    }
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        expected_result_paths.sort();
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_paths);
+    }
 }