Repository: alibaba/GraphScope (language: python)

Issue:
 请增加打印多跳路径具体业务号的功能
比如打印  点3-4跳路径的中间所有点，并可以指定业务号和属性来显示，而不是目前单一展示gid。
interactive.execute("g.V().has('trmnl','trmnl_black_flag',1).as('start').both('3..4', 'register').with('PATH_OPT', 'ARBITRARY') \
.with('RESULT_OPT', 'ALL_V').as('allV').select('start','allV'))").one()
结果为：
[{'start': v[2305843009213693975],
  'allV': [v[2305843009213693975],
   v[2886807361144487959],
   v[2305843009213693975],
   v[2886807361144487959]]},
目前打印路径展示的gid没有实际业务意义，希望看到的是起点的终端号，和中间的注册号，没有办法直接展示业务号，需要循环搜索g.V(gid).valueMap()来查询gid对应的业务号，过程有些繁琐。
所以，能否增加打印多跳路径具体业务号的功能，业务号本身就是唯一的。打印效果如
点trmnl，a,b,c,d,e
点code，1,2,3,4
  a-->3-->b-->2-->c-->1        
请支撑直观打印业务号，而不是[v[2305843009213693975],v[2886807361144487959], v[2305843009213693975],v[2886807361144487959]，这样无法看清具体路径，谢谢~


Discussion / hints:
Thanks for opening your first issue here! Be sure to follow the issue template! And a maintainer will get back to you shortly!
Please feel free to contact us on [DingTalk](https://h5.dingtalk.com/circle/healthCheckin.html?dtaction=os&corpId=dingb4b76e0fa9ccd8d516fd0f0acdf19f55&68e71190=0fe34ec0&cbdbhh=qwertyuiop), WeChat account(graphscope) or [Slack](http://slack.graphscope.io/). We are happy to answer your questions responsively.


Fix patch:
diff --git a/docs/interactive_engine/tinkerpop/supported_gremlin_steps.md b/docs/interactive_engine/tinkerpop/supported_gremlin_steps.md
index 27ab04e5e416..cb2bb60237c6 100644
--- a/docs/interactive_engine/tinkerpop/supported_gremlin_steps.md
+++ b/docs/interactive_engine/tinkerpop/supported_gremlin_steps.md
@@ -578,7 +578,7 @@ The following steps are extended to denote more complex situations.
 In Graph querying, expanding a multiple-hops path from a starting point is called `PathExpand`, which is commonly used in graph scenarios. In addition, there are different requirements for expanding strategies in different scenarios, i.e. it is required to output a simple path or all vertices explored along the expanding path. We introduce the with()-step to configure the corresponding behaviors of the `PathExpand`-step.
 
 #### out()
-Expand a multiple-hops path along the outgoing edges, which length is within the given range.
+Expand a multiple-hops path along the outgoing edges, which length is within the given range. 
 
 Parameters: </br>
 lengthRange - the lower and the upper bounds of the path length, </br> edgeLabels - the edge labels to traverse.
@@ -603,6 +603,9 @@ g.V().out("1..10", "knows")
 # expand hops within the range of [1, 10) along the outgoing edges which label is `knows` or `created`,
 # vertices can be duplicated and only the end vertex should be kept
 g.V().out("1..10", "knows", "created")
+# expand hops within the range of [1, 10) along the outgoing edges,
+# and project the properties "id" and "name" of every vertex along the path
+g.V().out("1..10").with('RESULT_OPT', 'ALL_V').values("name")
 ```
 Running Example:
 ```bash
@@ -615,6 +618,12 @@ gremlin> g.V().out("1..3", "knows").with('RESULT_OPT', 'ALL_V_E')
 gremlin> g.V().out("1..3", "knows").with('RESULT_OPT', 'END_V').endV()
 ==>v[2]
 ==>v[4]
+gremlin> g.V().out("1..3", "knows").with('RESULT_OPT', 'ALL_V').values("name")
+==>[marko, vadas]
+==>[marko, josh]
+gremlin> g.V().out("1..3", "knows").with('RESULT_OPT', 'ALL_V').valueMap("id","name")
+==>{id=[[1, 2]], name=[[marko, vadas]]}
+==>{id=[[1, 4]], name=[[marko, josh]]}
 ```
 #### in()
 Expand a multiple-hops path along the incoming edges, which length is within the given range.
diff --git a/interactive_engine/executor/ir/core/src/plan/logical.rs b/interactive_engine/executor/ir/core/src/plan/logical.rs
index 9f57e4bcdf2a..1f6e140fdb11 100644
--- a/interactive_engine/executor/ir/core/src/plan/logical.rs
+++ b/interactive_engine/executor/ir/core/src/plan/logical.rs
@@ -1295,10 +1295,6 @@ impl AsLogical for pb::PathExpand {
             let tag_id = get_or_set_tag_id(alias, plan_meta)?;
             plan_meta.set_tag_nodes(tag_id, vec![plan_meta.get_curr_node()]);
         }
-        // PathExpand would never require adding columns
-        plan_meta
-            .curr_node_meta_mut()
-            .set_columns_opt(ColumnsOpt::None);
 
         Ok(())
     }
diff --git a/interactive_engine/executor/ir/core/src/plan/physical.rs b/interactive_engine/executor/ir/core/src/plan/physical.rs
index b05f8372c022..f8fdef18ed03 100644
--- a/interactive_engine/executor/ir/core/src/plan/physical.rs
+++ b/interactive_engine/executor/ir/core/src/plan/physical.rs
@@ -236,14 +236,16 @@ impl AsPhysical for pb::PathExpand {
         if range.upper <= range.lower || range.lower < 0 || range.upper <= 0 {
             Err(IrError::InvalidRange(range.lower, range.upper))?
         }
+        // post_process for path_expand, including add repartition, and the properties need to cache, if necessary.
+        let mut path_expand = self.clone();
+        path_expand.post_process(builder, plan_meta)?;
         // PathExpand includes cases of:
-        //  1) EdgeExpand(Opt=Edge) + GetV(NoFilter),
+        //  1) EdgeExpand(Opt=Edge) + GetV(NoFilterNorColumn),
         //  This would be translated into EdgeExpand(Opt=Vertex);
-        //  2) EdgeExpand(Opt=Edge) + GetV(WithFilter),
+        //  2) EdgeExpand(Opt=Edge) + GetV(WithFilterOrColumn),
         //  This would be translated into EdgeExpand(Opt=Vertex) + GetV(Opt=Self);
         //  3) EdgeExpand(Opt=Vertex) + GetV(WithFilter and Opt=Self) TODO: would this case exist after match?
         //  This would be remain unchanged.
-        let mut path_expand = self.clone();
         if let Some(expand_base) = path_expand.base.as_mut() {
             let edge_expand = expand_base.edge_expand.as_mut();
             let getv = expand_base.get_v.as_mut();
@@ -286,8 +288,6 @@ impl AsPhysical for pb::PathExpand {
                     edge_expand, getv
                 )));
             }
-
-            path_expand.post_process(builder, plan_meta)?;
             builder.path_expand(path_expand);
 
             Ok(())
@@ -297,9 +297,142 @@ impl AsPhysical for pb::PathExpand {
     }
 
     fn post_process(&mut self, builder: &mut PlanBuilder, plan_meta: &mut PlanMeta) -> IrResult<()> {
-        post_process_vars(builder, plan_meta, false)?;
         if plan_meta.is_partition() {
             builder.shuffle(self.start_tag.clone());
+            if let Some(node_meta) = plan_meta.get_curr_node_meta() {
+                let columns = node_meta.get_columns();
+                let is_all_columns = node_meta.is_all_columns();
+                if !columns.is_empty() || is_all_columns {
+                    let new_params = pb::QueryParams {
+                        tables: vec![],
+                        columns: columns
+                            .clone()
+                            .into_iter()
+                            .map(|column| column.into())
+                            .collect(),
+                        is_all_columns,
+                        limit: None,
+                        predicate: None,
+                        sample_ratio: 1.0,
+                        extra: Default::default(),
+                    };
+                    // Notice that, when properties of a `Path` is needed, we need to cache the properties of the vertices/edges in the path.
+                    // For example, `g.V().out("1..3").with("RESULT_OPT, "ALL_V").values("name")`, we need to cache the property of "name" in all the vertices in the path.
+                    // If "RESULT_OPT" is "ALL_V_E", we assume the property of the edges in the path is also needed.
+
+                    // first, cache properties on the path start vertex.
+                    let start_auxilia = pb::GetV {
+                        tag: self.start_tag.clone(),
+                        opt: 4, //ItSelf
+                        params: Some(new_params.clone()),
+                        alias: self.start_tag.clone(),
+                        meta_data: None,
+                    };
+                    builder.get_v(start_auxilia);
+
+                    // then, cache properties during the path expanding.
+                    let result_opt: pb::path_expand::ResultOpt =
+                        unsafe { std::mem::transmute(self.result_opt) };
+                    let expand_base = self
+                        .base
+                        .as_mut()
+                        .ok_or(IrError::MissingData("PathExpand::base".to_string()))?;
+                    let getv = expand_base.get_v.as_mut();
+                    let edge_expand = expand_base
+                        .edge_expand
+                        .as_mut()
+                        .ok_or(IrError::MissingData("PathExpand::base.edge_expand".to_string()))?;
+                    match result_opt {
+                        pb::path_expand::ResultOpt::EndV => {
+                            // do nothing
+                        }
+                        // if the result_opt is ALL_V or ALL_V_E, we need to cache the properties of the vertices, or vertices and edges, in the path.
+                        pb::path_expand::ResultOpt::AllV => {
+                            if let Some(getv) = getv {
+                                // case 1:expand (edge) + getv, then cache properties in getv
+                                if let Some(params) = getv.params.as_mut() {
+                                    params.columns = columns
+                                        .clone()
+                                        .into_iter()
+                                        .map(|column| column.into())
+                                        .collect();
+                                    params.is_all_columns = is_all_columns;
+                                } else {
+                                    getv.params = Some(new_params.clone());
+                                }
+                            } else {
+                                // case 2: expand (vertex) + no getv, then cache properties with an extra getv (self)
+                                if edge_expand.expand_opt != pb::edge_expand::ExpandOpt::Vertex as i32 {
+                                    return Err(IrError::ParsePbError(
+                                        format!("Unexpected ExpandBase in PathExpand {:?}", expand_base)
+                                            .into(),
+                                    ));
+                                }
+
+                                let auxilia = pb::GetV {
+                                    tag: None,
+                                    opt: 4, //ItSelf
+                                    params: Some(new_params.clone()),
+                                    alias: edge_expand.alias.clone(),
+                                    meta_data: edge_expand.meta_data.clone(),
+                                };
+                                expand_base.get_v = Some(auxilia);
+                            }
+                        }
+                        pb::path_expand::ResultOpt::AllVE => {
+                            if let Some(getv) = getv {
+                                // case 1:expand (edge) + getv, then cache properties in both expand and getv.
+                                if let Some(params) = getv.params.as_mut() {
+                                    params.columns = columns
+                                        .clone()
+                                        .into_iter()
+                                        .map(|column| column.into())
+                                        .collect();
+                                    params.is_all_columns = is_all_columns;
+                                } else {
+                                    getv.params = Some(new_params.clone());
+                                }
+                                if let Some(params) = edge_expand.params.as_mut() {
+                                    params.columns = columns
+                                        .clone()
+                                        .into_iter()
+                                        .map(|column| column.into())
+                                        .collect();
+                                    params.is_all_columns = is_all_columns;
+                                } else {
+                                    edge_expand.params = Some(new_params.clone());
+                                }
+                            } else {
+                                // case 2: expand (vertex) + no getv, then cache properties of edges in expand, and properties of vertices with an extra getv (self)
+                                if edge_expand.expand_opt != pb::edge_expand::ExpandOpt::Vertex as i32 {
+                                    return Err(IrError::ParsePbError(
+                                        format!("Unexpected ExpandBase in PathExpand {:?}", expand_base)
+                                            .into(),
+                                    ));
+                                }
+                                if let Some(params) = edge_expand.params.as_mut() {
+                                    params.columns = columns
+                                        .clone()
+                                        .into_iter()
+                                        .map(|column| column.into())
+                                        .collect();
+                                    params.is_all_columns = is_all_columns;
+                                } else {
+                                    edge_expand.params = Some(new_params.clone());
+                                }
+                                let auxilia = pb::GetV {
+                                    tag: None,
+                                    opt: 4, //ItSelf
+                                    params: Some(new_params.clone()),
+                                    alias: edge_expand.alias.clone(),
+                                    meta_data: edge_expand.meta_data.clone(),
+                                };
+                                expand_base.get_v = Some(auxilia);
+                            }
+                        }
+                    }
+                }
+            }
         }
         Ok(())
     }
@@ -2917,4 +3050,89 @@ mod test {
 
         assert_eq!(builder, expected_builder);
     }
+
+    #[test]
+    fn path_expand_project_as_physical() {
+        let source_opr = pb::Scan {
+            scan_opt: 0,
+            alias: None,
+            params: Some(query_params(vec!["person".into()], vec![])),
+            idx_predicate: None,
+            meta_data: None,
+        };
+
+        let edge_expand = pb::EdgeExpand {
+            v_tag: None,
+            direction: 0,
+            params: Some(query_params(vec!["knows".into()], vec![])),
+            expand_opt: 0, // vertex
+            alias: None,
+            meta_data: None,
+        };
+
+        let path_opr = pb::PathExpand {
+            base: Some(edge_expand.clone().into()),
+            start_tag: None,
+            alias: None,
+            hop_range: Some(pb::Range { lower: 1, upper: 4 }),
+            path_opt: 0,   // ARBITRARY
+            result_opt: 1, // ALL_V
+            condition: None,
+        };
+
+        let project_opr = pb::Project {
+            mappings: vec![ExprAlias {
+                expr: Some(str_to_expr_pb("@.name".to_string()).unwrap()),
+                alias: None,
+            }],
+            is_append: true,
+            meta_data: vec![],
+        };
+
+        let mut logical_plan = LogicalPlan::with_node(Node::new(0, source_opr.clone().into()));
+        logical_plan
+            .append_operator_as_node(path_opr.clone().into(), vec![0])
+            .unwrap(); // node 1
+        logical_plan
+            .append_operator_as_node(project_opr.clone().into(), vec![1])
+            .unwrap(); // node 2
+
+        // Case without partition
+        let mut builder = PlanBuilder::default();
+        let mut plan_meta = logical_plan.get_meta().clone();
+        logical_plan
+            .add_job_builder(&mut builder, &mut plan_meta)
+            .unwrap();
+
+        let mut expected_builder = PlanBuilder::default();
+        expected_builder.add_scan_source(source_opr.clone());
+        expected_builder.path_expand(path_opr.clone());
+        expected_builder.project(project_opr.clone());
+
+        assert_eq!(builder, expected_builder);
+
+        // Case with partition
+        let mut builder = PlanBuilder::default();
+        let mut plan_meta = logical_plan.get_meta().clone().with_partition();
+        logical_plan
+            .add_job_builder(&mut builder, &mut plan_meta)
+            .unwrap();
+
+        // translate `PathExpand(out("knows"))` to `auxilia("name") + PathExpand() with ExpandBase of out("knows")+auxilia("name")`
+        let mut path_expand = path_opr.clone();
+        path_expand.base.as_mut().unwrap().get_v =
+            Some(build_auxilia_with_tag_alias_columns(None, None, vec!["name".into()]));
+
+        let mut expected_builder = PlanBuilder::default();
+        expected_builder.add_scan_source(source_opr);
+        expected_builder.shuffle(None);
+        // post process for path expand: 1. cache properties of path start vertex; 2.
+        expected_builder.get_v(build_auxilia_with_tag_alias_columns(None, None, vec!["name".into()]));
+        expected_builder.path_expand(path_expand);
+        // postprocess for project
+        expected_builder.shuffle(None);
+        expected_builder.get_v(build_auxilia_with_tag_alias_columns(None, None, vec![]));
+        expected_builder.project(project_opr);
+        assert_eq!(builder, expected_builder);
+    }
 }
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 9d091825a787..f123cfc1f582 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
@@ -29,6 +29,7 @@ use pegasus_common::downcast::AsAny;
 use pegasus_common::impl_as_any;
 
 use crate::apis::{Edge, Element, GraphElement, PropertyValue, Vertex, ID};
+use crate::utils::expr::eval::Context;
 
 #[derive(Clone, Debug, Hash, PartialEq, PartialOrd)]
 pub enum VertexOrEdge {
@@ -219,6 +220,12 @@ impl GraphElement for VertexOrEdge {
     }
 }
 
+impl Context<VertexOrEdge> for VertexOrEdge {
+    fn get(&self, _tag: Option<&NameOrId>) -> Option<&VertexOrEdge> {
+        Some(&self)
+    }
+}
+
 impl Element for GraphPath {
     fn as_graph_element(&self) -> Option<&dyn GraphElement> {
         Some(self)
@@ -256,11 +263,34 @@ impl GraphElement for GraphPath {
     }
 
     fn get_property(&self, key: &NameOrId) -> Option<PropertyValue> {
-        self.get_path_end().get_property(key)
+        match self {
+            GraphPath::AllPath(path) | GraphPath::SimpleAllPath(path) => {
+                let mut properties = vec![];
+                for v_or_e in path {
+                    if let Some(p) = v_or_e.get_property(key) {
+                        properties.push(p.try_to_owned().unwrap());
+                    }
+                }
+                Some(PropertyValue::Owned(Object::Vector(properties)))
+            }
+
+            GraphPath::EndV((v_or_e, _)) | GraphPath::SimpleEndV((v_or_e, _, _)) => {
+                v_or_e.get_property(key)
+            }
+        }
     }
 
     fn get_all_properties(&self) -> Option<HashMap<NameOrId, Object>> {
-        self.get_path_end().get_all_properties()
+        match self {
+            GraphPath::AllPath(_) | GraphPath::SimpleAllPath(_) => {
+                // not supported yet.
+                None
+            }
+
+            GraphPath::EndV((v_or_e, _)) | GraphPath::SimpleEndV((v_or_e, _, _)) => {
+                v_or_e.get_all_properties()
+            }
+        }
     }
 }
 
diff --git a/interactive_engine/executor/ir/runtime/src/assembly.rs b/interactive_engine/executor/ir/runtime/src/assembly.rs
index fb61c8accdaf..017a922fc5aa 100644
--- a/interactive_engine/executor/ir/runtime/src/assembly.rs
+++ b/interactive_engine/executor/ir/runtime/src/assembly.rs
@@ -159,6 +159,10 @@ impl<P: PartitionInfo, C: ClusterInfo> FnGenerator<P, C> {
         Ok(opr.gen_map()?)
     }
 
+    fn gen_path_condition(&self, opr: pb::PathExpand) -> FnGenResult<RecordFilter> {
+        Ok(opr.gen_filter()?)
+    }
+
     fn gen_coin(&self, opr: algebra_pb::Sample) -> FnGenResult<RecordFilter> {
         Ok(opr.gen_filter()?)
     }
@@ -604,11 +608,9 @@ impl<P: PartitionInfo, C: ClusterInfo> IRJobAssembly<P, C> {
                     }
                     let times = range.upper - range.lower - 1;
                     if times > 0 {
-                        if let Some(condition) = path.condition.as_ref() {
+                        if path.condition.is_some() {
                             let mut until = IterCondition::max_iters(times as u32);
-                            let func = self
-                                .udf_gen
-                                .gen_filter(algebra_pb::Select { predicate: Some(condition.clone()) })?;
+                            let func = self.udf_gen.gen_path_condition(path.clone())?;
                             until.set_until(func);
                             // Notice that if UNTIL condition set, we expand path without `Emit`
                             stream = stream
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/filter/mod.rs b/interactive_engine/executor/ir/runtime/src/process/operator/filter/mod.rs
index c52b5e626e3b..e3f403a420c6 100644
--- a/interactive_engine/executor/ir/runtime/src/process/operator/filter/mod.rs
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/filter/mod.rs
@@ -13,6 +13,7 @@
 //! See the License for the specific language governing permissions and
 //! limitations under the License.
 mod coin;
+mod path_condition;
 mod select;
 
 use pegasus::api::function::FilterFunction;
diff --git a/interactive_engine/executor/ir/runtime/src/process/operator/filter/path_condition.rs b/interactive_engine/executor/ir/runtime/src/process/operator/filter/path_condition.rs
new file mode 100644
index 000000000000..1a349296e282
--- /dev/null
+++ b/interactive_engine/executor/ir/runtime/src/process/operator/filter/path_condition.rs
@@ -0,0 +1,66 @@
+//
+//! Copyright 2023 Alibaba Group Holding Limited.
+//!
+//! Licensed under the Apache License, Version 2.0 (the "License");
+//! you may not use this file except in compliance with the License.
+//! You may obtain a copy of the License at
+//!
+//! http://www.apache.org/licenses/LICENSE-2.0
+//!
+//! Unless required by applicable law or agreed to in writing, software
+//! distributed under the License is distributed on an "AS IS" BASIS,
+//! WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//! See the License for the specific language governing permissions and
+//! limitations under the License.
+
+use std::convert::TryInto;
+
+use graph_proxy::utils::expr::eval_pred::{EvalPred, PEvaluator};
+use ir_common::error::ParsePbError;
+use ir_common::generated::physical as pb;
+use pegasus::api::function::{FilterFunction, FnResult};
+
+use crate::error::{FnExecError, FnGenResult};
+use crate::process::entry::Entry;
+use crate::process::operator::filter::FilterFuncGen;
+use crate::process::record::Record;
+
+/// a filter for path until condition
+#[derive(Debug)]
+struct PathConditionOperator {
+    pub filter: PEvaluator,
+}
+
+impl FilterFunction<Record> for PathConditionOperator {
+    fn test(&self, input: &Record) -> FnResult<bool> {
+        if let Some(entry) = input.get(None) {
+            if let Some(path) = entry.as_graph_path() {
+                // we assume the until condition must be tested on the path end
+                let path_end = path.get_path_end();
+                let res = self
+                    .filter
+                    .eval_bool(Some(path_end))
+                    .map_err(|e| FnExecError::from(e))?;
+                return Ok(res);
+            }
+        }
+        Err(FnExecError::unexpected_data_error(&format!(
+            "unexpected input for path until condition {:?}",
+            input.get(None),
+        )))?
+    }
+}
+
+impl FilterFuncGen for pb::PathExpand {
+    fn gen_filter(self) -> FnGenResult<Box<dyn FilterFunction<Record>>> {
+        if let Some(predicate) = self.condition {
+            let path_condition_operator = PathConditionOperator { filter: predicate.try_into()? };
+            if log_enabled!(log::Level::Debug) && pegasus::get_current_worker().index == 0 {
+                debug!("Runtime path condition operator: {:?}", path_condition_operator);
+            }
+            Ok(Box::new(path_condition_operator))
+        } else {
+            Err(ParsePbError::EmptyFieldError("empty path condition pb".to_string()).into())
+        }
+    }
+}


Test patch:
diff --git a/interactive_engine/executor/ir/integrated/tests/pathxd_test.rs b/interactive_engine/executor/ir/integrated/tests/pathxd_test.rs
index 8580ae8cb667..60cf06645a97 100644
--- a/interactive_engine/executor/ir/integrated/tests/pathxd_test.rs
+++ b/interactive_engine/executor/ir/integrated/tests/pathxd_test.rs
@@ -19,6 +19,7 @@ mod common;
 
 #[cfg(test)]
 mod test {
+    use dyn_type::{object, Object};
     use graph_proxy::apis::{Element, GraphElement, ID};
     use ir_common::expr_parse::str_to_expr_pb;
     use ir_common::generated::algebra as pb;
@@ -873,4 +874,85 @@ mod test {
     fn path_expand_exactly_whole_v_e_w2_test() {
         path_expand_exactly_whole_v_e_query(2)
     }
+
+    // g.V().hasLabel("person").both("2..3", "knows").values("name")
+    fn init_path_expand_project_request(result_opt: i32) -> JobRequest {
+        let source_opr = pb::Scan {
+            scan_opt: 0,
+            alias: None,
+            params: Some(query_params(vec![PERSON_LABEL.into()], vec![], None)),
+            idx_predicate: None,
+            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,
+        };
+
+        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,
+        };
+
+        let project_opr = pb::Project {
+            mappings: vec![pb::project::ExprAlias {
+                expr: Some(str_to_expr_pb("@.name".to_string()).unwrap()),
+                alias: None,
+            }],
+            is_append: true,
+            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()
+    }
+
+    #[test]
+    fn path_expand_allv_project_test() {
+        initialize();
+        let request = init_path_expand_project_request(1); // 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(properties) = entry.get(None).unwrap().as_object() {
+                        result_collection.push(properties.clone());
+                    }
+                }
+                Err(e) => {
+                    panic!("err result {:?}", e);
+                }
+            }
+        }
+        expected_result_paths.sort();
+        result_collection.sort();
+        assert_eq!(result_collection, expected_result_paths);
+    }
 }