Repository: alibaba/GraphScope (language: python)

Issue:
[BUG] It seems like the Interactive Engine doesn't support some Cypher statements
**Describe the bug**

`MATCH p=()-[:write]->() RETURN p LIMIT 25;`
![image](https://github.com/alibaba/GraphScope/assets/10703060/d8c24a75-43bd-4943-9d1a-b7eba8138408)

`Match (a)-[c*1..3]->(b) Return a, c, b`

![image](https://github.com/alibaba/GraphScope/assets/10703060/9c4e5594-9926-4872-845e-b9b0cdba5f16)

`Match (n) return n limit 10` this statement executed successfully.

![image](https://github.com/alibaba/GraphScope/assets/10703060/2a724855-c8ac-4db6-aa2f-b4772fa9489b)



**To Reproduce**
Steps to reproduce the behavior:
 
Refer to this document: https://graphscope.io/docs/flex/interactive/custom_graph_data




Discussion / hints:
For the first query, `p = xxx` such kind of query is not supported yet. We have no plan to support it in a short time. We  should update the doc: https://graphscope.io/docs/latest/interactive_engine/neo4j/supported_cypher, to include more known limitations @shirly121 . 

Fix patch:
diff --git a/flex/engines/hqps_db/core/null_record.h b/flex/engines/hqps_db/core/null_record.h
index 371d4616982e..291a5885d12b 100644
--- a/flex/engines/hqps_db/core/null_record.h
+++ b/flex/engines/hqps_db/core/null_record.h
@@ -18,6 +18,7 @@
 #include <limits>
 #include <tuple>
 #include "flex/engines/hqps_db/core/utils/hqps_utils.h"
+#include "flex/engines/hqps_db/structures/path.h"
 
 namespace gs {
 
@@ -45,6 +46,13 @@ struct NullRecordCreator<std::tuple<T...>> {
   }
 };
 
+template <typename VID_T, typename LabelT>
+struct NullRecordCreator<Path<VID_T, LabelT>> {
+  static inline Path<VID_T, LabelT> GetNull() {
+    return Path<VID_T, LabelT>::Null();
+  }
+};
+
 template <size_t Ind = 0, typename... T>
 static inline bool IsNull(const std::tuple<T...>& tuple) {
   if constexpr (Ind + 1 < sizeof...(T)) {
diff --git a/flex/engines/hqps_db/core/operator/get_v.h b/flex/engines/hqps_db/core/operator/get_v.h
index 59dce5e85a52..f590bf1ba97a 100644
--- a/flex/engines/hqps_db/core/operator/get_v.h
+++ b/flex/engines/hqps_db/core/operator/get_v.h
@@ -230,7 +230,8 @@ class GetVertex {
       std::array<label_id_t, num_labels>& req_labels,
       Filter<EXPRESSION, SELECTOR...>& filter) {
     auto req_label_vec = array_to_vec(req_labels);
-    auto labels = set.GetLabels();
+    std::vector<label_id_t> labels = set.GetLabels(v_opt);
+
     // remove duplicate from labels
     std::sort(labels.begin(), labels.end());
     labels.erase(std::unique(labels.begin(), labels.end()), labels.end());
@@ -260,6 +261,16 @@ class GetVertex {
     return set.GetVertices(v_opt, filter.expr_, property_getters_array);
   }
 
+  // get no prop v from path set.
+  template <size_t num_labels, typename EXPRESSION, typename... SELECTOR>
+  static auto GetNoPropVFromPathSetImpl(
+      const GRAPH_INTERFACE& graph, const PathSet<vertex_id_t, label_id_t>& set,
+      VOpt v_opt, std::array<label_id_t, num_labels>& req_labels,
+      Filter<EXPRESSION, SELECTOR...>& filter) {
+    auto req_label_vec = array_to_vec(req_labels);
+    return set.GetVertices(v_opt, req_label_vec);
+  }
+
   // User-defined expression
   // for vertex set with multiple labels, i.e. two_label or general vertex set.
   // do project.
diff --git a/flex/engines/hqps_db/core/operator/path_expand.h b/flex/engines/hqps_db/core/operator/path_expand.h
index b08cceb3f39d..c44ea8f32ded 100644
--- a/flex/engines/hqps_db/core/operator/path_expand.h
+++ b/flex/engines/hqps_db/core/operator/path_expand.h
@@ -74,6 +74,100 @@ class PathExpand {
     return std::make_pair(std::move(path_set), std::move(offsets));
   }
 
+  // PathExpand Path with multiple edge triplet.
+  template <typename VERTEX_SET_T, typename LabelT, size_t get_v_num_labels,
+            typename EDGE_FILTER_T, typename VERTEX_FILTER_T>
+  static auto PathExpandP(
+      const GRAPH_INTERFACE& graph, const VERTEX_SET_T& vertex_set,
+      PathExpandVMultiTripletOpt<LabelT, EDGE_FILTER_T, get_v_num_labels,
+                                 VERTEX_FILTER_T>&& path_expand_opt) {
+    auto& range = path_expand_opt.range_;
+    auto& edge_expand_opt = path_expand_opt.edge_expand_opt_;
+    auto& get_v_opt = path_expand_opt.get_v_opt_;
+    auto& edge_triplets = edge_expand_opt.edge_label_triplets_;
+    auto& vertex_other_labels = get_v_opt.v_labels_;
+    auto vertex_other_labels_vec = array_to_vec(vertex_other_labels);
+
+    std::vector<std::vector<vertex_id_t>> other_vertices;
+    std::vector<std::vector<label_id_t>> other_labels_vec;
+    std::vector<std::vector<offset_t>> other_offsets;
+    auto& vertices_vec = vertex_set.GetVertices();
+    std::vector<label_id_t> src_label_id_vec;
+    std::vector<label_id_t> src_labels_set;
+    static_assert(VERTEX_SET_T::is_row_vertex_set ||
+                      VERTEX_SET_T::is_two_label_set ||
+                      VERTEX_SET_T::is_general_set,
+                  "Unsupported vertex set type");
+    if constexpr (VERTEX_SET_T::is_row_vertex_set) {
+      auto src_label = vertex_set.GetLabel();
+      src_label_id_vec =
+          std::vector<label_id_t>(vertices_vec.size(), vertex_set.GetLabel());
+      src_labels_set = {src_label};
+    } else if constexpr (VERTEX_SET_T::is_two_label_set) {
+      auto src_label_vec = vertex_set.GetLabelVec();
+      src_labels_set = array_to_vec(vertex_set.GetLabels());
+      src_label_id_vec = label_key_vec_2_label_id_vec(src_label_vec);
+    } else {
+      src_labels_set = vertex_set.GetLabels();
+      src_label_id_vec = label_key_vec_2_label_id_vec(vertex_set.GetLabelVec());
+    }
+    std::tie(other_vertices, other_labels_vec, other_offsets) =
+        path_expandp_multi_triplet(graph, edge_triplets,
+                                   vertex_other_labels_vec,
+                                   edge_expand_opt.direction_, vertices_vec,
+                                   src_labels_set, src_label_id_vec, range);
+
+    // The path are stored in a compressed way, and we flat it.
+    std::vector<offset_t> res_offsets;
+    res_offsets.reserve(vertices_vec.size() + 1);
+    res_offsets.emplace_back(0);
+
+    std::vector<std::vector<Path<vid_t, LabelT>>> cur_path, next_path;
+    std::vector<std::vector<Path<vid_t, LabelT>>> prev_path;
+    // indexed by src_vid,
+    for (size_t i = 0; i < vertices_vec.size(); ++i) {
+      std::vector<Path<vid_t, LabelT>> tmp_path;
+      tmp_path.emplace_back(vertices_vec[i], src_label_id_vec[i]);
+      cur_path.emplace_back(tmp_path);
+    }
+    prev_path.resize(vertices_vec.size());
+
+    for (auto j = range.start_; j < range.limit_; ++j) {
+      next_path.clear();
+      auto& cur_offset_vec = other_offsets[j];
+      CHECK(cur_path.size() == vertices_vec.size());
+      next_path.resize(vertices_vec.size());
+
+      for (auto i = 0; i < vertices_vec.size(); ++i) {
+        auto& tmp_path = cur_path[i];
+        auto start = cur_offset_vec[i];
+        auto end = cur_offset_vec[i + 1];
+        for (auto k = start; k < end; ++k) {
+          auto& next_vid = other_vertices[j][k];
+          auto& label = other_labels_vec[j][k];
+          for (auto& path : tmp_path) {
+            path.EmplaceBack(next_vid, label);
+            next_path[i].emplace_back(path);
+            path.PopBack();
+          }
+        }
+        // push all next_path[i] to tmp_path
+        prev_path[i].insert(prev_path[i].end(), next_path[i].begin(),
+                            next_path[i].end());
+        next_path[i].swap(cur_path[i]);
+      }
+    }
+
+    std::vector<Path<vid_t, LabelT>> res_path;
+    for (auto i = 0; i < vertices_vec.size(); ++i) {
+      auto& tmp_path = prev_path[i];
+      res_path.insert(res_path.end(), tmp_path.begin(), tmp_path.end());
+      res_offsets.emplace_back(res_path.size());
+    }
+    return std::make_pair(PathSet<vertex_id_t, label_id_t>(std::move(res_path)),
+                          std::move(res_offsets));
+  }
+
   // Path expand to vertices with columns.
   // PathExpand to vertices with vertex properties also retreived
   template <typename... V_SET_T, typename VERTEX_FILTER_T, typename LabelT,
@@ -741,7 +835,7 @@ class PathExpand {
   // expand from vertices, with multiple edge triplets.
   // The intermediate vertices can also have multiple labels, and expand with
   // multiple edge triplet.
-  static auto path_expandv_multi_triplet(
+  static auto path_expandp_multi_triplet(
       const GRAPH_INTERFACE& graph,
       const std::vector<std::array<label_id_t, 3>>&
           edge_label_triplets,  // src, dst, edge
@@ -749,13 +843,6 @@ class PathExpand {
       const std::vector<vertex_id_t>& vertices_vec,
       const std::vector<label_id_t>& src_labels_set,
       const std::vector<label_id_t>& src_v_labels_vec, const Range& range) {
-    // (range, other_label_ind, vertices)
-    LOG(INFO) << "PathExpandV with multiple edge triplets: "
-              << gs::to_string(edge_label_triplets)
-              << ", direction: " << gs::to_string(direction)
-              << ", vertices size: " << vertices_vec.size()
-              << ", src_labels_set: " << gs::to_string(src_labels_set)
-              << ", range: " << range.start_ << ", " << range.limit_;
     std::vector<std::vector<vertex_id_t>> other_vertices;
     std::vector<std::vector<label_id_t>> other_labels_vec;
     std::vector<std::vector<offset_t>> other_offsets;
@@ -924,6 +1011,35 @@ class PathExpand {
             tmp_cur_offset[other_offsets[cur_hop - 1][i]]);
       }
     }
+    return std::make_tuple(std::move(other_vertices),
+                           std::move(other_labels_vec),
+                           std::move(other_offsets));
+  }
+
+  static auto path_expandv_multi_triplet(
+      const GRAPH_INTERFACE& graph,
+      const std::vector<std::array<label_id_t, 3>>&
+          edge_label_triplets,  // src, dst, edge
+      const std::vector<label_id_t>& get_v_labels, const Direction& direction,
+      const std::vector<vertex_id_t>& vertices_vec,
+      const std::vector<label_id_t>& src_labels_set,
+      const std::vector<label_id_t>& src_v_labels_vec, const Range& range) {
+    // (range, other_label_ind, vertices)
+    LOG(INFO) << "PathExpandV with multiple edge triplets: "
+              << gs::to_string(edge_label_triplets)
+              << ", direction: " << gs::to_string(direction)
+              << ", vertices size: " << vertices_vec.size()
+              << ", src_labels_set: " << gs::to_string(src_labels_set)
+              << ", range: " << range.start_ << ", " << range.limit_;
+
+    std::vector<std::vector<vertex_id_t>> other_vertices;
+    std::vector<std::vector<label_id_t>> other_labels_vec;
+    std::vector<std::vector<offset_t>> other_offsets;
+
+    std::tie(other_vertices, other_labels_vec, other_offsets) =
+        path_expandp_multi_triplet(graph, edge_label_triplets, get_v_labels,
+                                   direction, vertices_vec, src_labels_set,
+                                   src_v_labels_vec, range);
 
     // select vertices that are in range and are in vertex_other_labels.
     std::vector<vertex_id_t> res_vertices;
diff --git a/flex/engines/hqps_db/core/operator/sink.h b/flex/engines/hqps_db/core/operator/sink.h
index 503f639eac21..1662c251d94e 100644
--- a/flex/engines/hqps_db/core/operator/sink.h
+++ b/flex/engines/hqps_db/core/operator/sink.h
@@ -926,10 +926,11 @@ class SinkOp {
   }
 
   // sink for compressed path set
-  template <size_t Ind, size_t act_tag_id, typename VID_T, typename LabelT>
+  template <size_t Ind, size_t act_tag_id, typename PATH_SET_T,
+            typename std::enable_if_t<(PATH_SET_T::is_path_set)>* = nullptr>
   static void sink_col_impl(const GRAPH_INTERFACE& graph,
                             results::CollectiveResults& results_vec,
-                            const CompressedPathSet<VID_T, LabelT>& path_set,
+                            const PATH_SET_T& path_set,
                             const std::vector<size_t>& repeat_offsets,
                             int32_t tag_id) {
     if (repeat_offsets.empty()) {
@@ -981,16 +982,24 @@ class SinkOp {
     }
   }
 
+  // TODO(zhanglei): This is temporary solution for sink path to results.
+  // Out physical plan will only generate EndV option, so we can only sink the
+  // end vertex to results.
+  // If we sink all vertices to results, cypher driver seems failed to parse
+  // paths of different lengths or what:
+  // <Tried to construct a path that is not built like a path: even number of
+  // elements>
   static void add_path_to_pb(const Path<vid_t, label_id_t>& path,
                              results::GraphPath& mutable_path) {
-    for (size_t i = 0; i <= path.length(); ++i) {
-      // path's length + 1 = number of nodes
-      vid_t vid;
-      label_id_t label;
-      std::tie(label, vid) = path.GetNode(i);
-      mutable_path.add_path()->mutable_vertex()->set_id(
-          encode_unique_vertex_id(label, vid));
+    if (path.length() <= 0) {
+      return;
     }
+    vid_t vid;
+    label_id_t label;
+    std::tie(label, vid) = path.GetNode(path.length() - 1);
+    auto vertex = mutable_path.add_path()->mutable_vertex();
+    vertex->set_id(encode_unique_vertex_id(label, vid));
+    vertex->mutable_label()->set_id(label);
   }
 
   static vid_t encode_unique_vertex_id(label_id_t label_id, vid_t vid) {
diff --git a/flex/engines/hqps_db/core/sync_engine.h b/flex/engines/hqps_db/core/sync_engine.h
index 6a7e853a98fb..dbeeae969d03 100644
--- a/flex/engines/hqps_db/core/sync_engine.h
+++ b/flex/engines/hqps_db/core/sync_engine.h
@@ -520,6 +520,32 @@ class SyncEngine : public BaseEngine {
     // old context will be abandon here.
   }
 
+  /// Expand to Path
+  template <AppendOpt opt, int alias_to_use, typename CTX_HEAD_T, int cur_alias,
+            int base_tag, typename... CTX_PREV, typename LabelT,
+            typename EDGE_FILTER_T, size_t get_v_num_labels,
+            typename VERTEX_FILTER_T>
+  static auto PathExpandP(
+      const GRAPH_INTERFACE& graph,
+      Context<CTX_HEAD_T, cur_alias, base_tag, CTX_PREV...>&& ctx,
+      PathExpandVMultiTripletOpt<LabelT, EDGE_FILTER_T, get_v_num_labels,
+                                 VERTEX_FILTER_T>&& path_expand_opt) {
+    if (path_expand_opt.path_opt_ != PathOpt::Arbitrary) {
+      LOG(FATAL) << "Only support Arbitrary path now";
+    }
+    if (path_expand_opt.result_opt_ != ResultOpt::EndV) {
+      LOG(FATAL) << "Only support EndV now";
+    }
+    auto& select_node = gs::Get<alias_to_use>(ctx);
+    auto pair = PathExpand<GRAPH_INTERFACE>::PathExpandP(
+        graph, select_node, std::move(path_expand_opt));
+
+    // create new context node, update offsets.
+    return ctx.template AddNode<opt>(std::move(pair.first),
+                                     std::move(pair.second), alias_to_use);
+    // old context will be abandon here.
+  }
+
   // get no props, just filter
   template <
       AppendOpt opt, int alias_to_use, typename CTX_HEAD_T, int cur_alias,
diff --git a/flex/engines/hqps_db/core/utils/keyed.h b/flex/engines/hqps_db/core/utils/keyed.h
index 0eb1bb563145..db6076203113 100644
--- a/flex/engines/hqps_db/core/utils/keyed.h
+++ b/flex/engines/hqps_db/core/utils/keyed.h
@@ -25,6 +25,7 @@
 #include "flex/engines/hqps_db/structures/multi_vertex_set/keyed_row_vertex_set.h"
 #include "flex/engines/hqps_db/structures/multi_vertex_set/row_vertex_set.h"
 #include "flex/engines/hqps_db/structures/multi_vertex_set/two_label_vertex_set.h"
+#include "flex/engines/hqps_db/structures/path.h"
 
 namespace gs {
 
@@ -536,6 +537,34 @@ struct KeyedAggT<GI, FlatEdgeSet<VID_T, LabelT, EDATA_T>, AggFunc::COUNT,
   }
 };
 
+template <typename GI, typename VID_T, typename LabelT, int tag_id>
+struct KeyedAggT<GI, CompressedPathSet<VID_T, LabelT>, AggFunc::COUNT,
+                 std::tuple<grape::EmptyType>,
+                 std::integer_sequence<int32_t, tag_id>> {
+  using agg_res_t = Collection<size_t>;
+  using aggregate_res_builder_t = CountBuilder<tag_id>;
+
+  static aggregate_res_builder_t create_agg_builder(
+      const CompressedPathSet<VID_T, LabelT>& set, const GI& graph,
+      std::tuple<PropertySelector<grape::EmptyType>>& selectors) {
+    return CountBuilder<tag_id>();
+  }
+};
+
+template <typename GI, typename VID_T, typename LabelT, int tag_id>
+struct KeyedAggT<GI, PathSet<VID_T, LabelT>, AggFunc::COUNT,
+                 std::tuple<grape::EmptyType>,
+                 std::integer_sequence<int32_t, tag_id>> {
+  using agg_res_t = Collection<size_t>;
+  using aggregate_res_builder_t = CountBuilder<tag_id>;
+
+  static aggregate_res_builder_t create_agg_builder(
+      const PathSet<VID_T, LabelT>& set, const GI& graph,
+      std::tuple<PropertySelector<grape::EmptyType>>& selectors) {
+    return CountBuilder<tag_id>();
+  }
+};
+
 // COUNT DISTINCT for EdgeSets.
 template <typename GI, typename VID_T, typename LabelT, int tag_id>
 struct KeyedAggT<GI, UnTypedEdgeSet<VID_T, LabelT, typename GI::sub_graph_t>,
@@ -585,6 +614,36 @@ struct KeyedAggT<GI, FlatEdgeSet<VID_T, LabelT, EDATA_T>,
   }
 };
 
+template <typename GI, typename VID_T, typename LabelT, int tag_id>
+struct KeyedAggT<GI, CompressedPathSet<VID_T, LabelT>, AggFunc::COUNT_DISTINCT,
+                 std::tuple<grape::EmptyType>,
+                 std::integer_sequence<int32_t, tag_id>> {
+  using agg_res_t = Collection<size_t>;
+  using path_set_t = CompressedPathSet<VID_T, LabelT>;
+  using aggregate_res_builder_t = DistinctCountBuilder<tag_id, path_set_t>;
+
+  static aggregate_res_builder_t create_agg_builder(
+      const path_set_t& set, const GI& graph,
+      std::tuple<PropertySelector<grape::EmptyType>>& selectors) {
+    return aggregate_res_builder_t(set);
+  }
+};
+
+template <typename GI, typename VID_T, typename LabelT, int tag_id>
+struct KeyedAggT<GI, PathSet<VID_T, LabelT>, AggFunc::COUNT_DISTINCT,
+                 std::tuple<grape::EmptyType>,
+                 std::integer_sequence<int32_t, tag_id>> {
+  using agg_res_t = Collection<size_t>;
+  using path_set_t = PathSet<VID_T, LabelT>;
+  using aggregate_res_builder_t = DistinctCountBuilder<tag_id, path_set_t>;
+
+  static aggregate_res_builder_t create_agg_builder(
+      const path_set_t& set, const GI& graph,
+      std::tuple<PropertySelector<grape::EmptyType>>& selectors) {
+    return aggregate_res_builder_t(set);
+  }
+};
+
 template <typename GI, typename... SET_T, typename... PropSelectorT,
           int... tag_ids>
 struct KeyedAggMultiColT<GI, std::tuple<SET_T...>, AggFunc::COUNT_DISTINCT,
diff --git a/flex/engines/hqps_db/structures/collection.h b/flex/engines/hqps_db/structures/collection.h
index 781fe38e214b..fa53317fd549 100644
--- a/flex/engines/hqps_db/structures/collection.h
+++ b/flex/engines/hqps_db/structures/collection.h
@@ -652,6 +652,48 @@ class DistinctCountBuilder<tag_id, TwoLabelVertexSetImpl<VID_T, LabelT, T...>> {
   std::array<VID_T, 2> min_v, max_v, range_size;
 };
 
+// DistinctCountBuilder for PathSet
+template <int tag_id, typename PATH_SET_T>
+class DistinctCountBuilder<
+    tag_id, PATH_SET_T,
+    typename std::enable_if<PATH_SET_T::is_path_set>::type> {
+ public:
+  using path_set_t = PATH_SET_T;
+  using index_ele_t = typename path_set_t::index_ele_tuple_t;
+  DistinctCountBuilder(const path_set_t& path_set) {
+    paths_num_ = path_set.Size();
+  }
+
+  template <typename ELE_TUPLE_T, typename DATA_TUPLE>
+  void insert(size_t ind, const ELE_TUPLE_T& tuple, const DATA_TUPLE& data) {
+    auto& cur_ind_ele = gs::get_from_tuple<tag_id>(tuple);
+    while (vec_.size() <= ind) {
+      vec_.emplace_back(grape::Bitset(paths_num_));
+    }
+    auto& cur_bitset = vec_[ind];
+    auto cur_ind = std::get<0>(cur_ind_ele);
+    if (cur_ind < paths_num_) {
+      cur_bitset.set_bit(cur_ind);
+    } else {
+      LOG(FATAL) << "Invalid path set index: " << cur_ind
+                 << ", path set num: " << paths_num_;
+    }
+  }
+
+  Collection<size_t> Build() {
+    std::vector<size_t> res;
+    res.reserve(vec_.size());
+    for (auto& bitset : vec_) {
+      res.emplace_back(bitset.count());
+    }
+    return Collection<size_t>(std::move(res));
+  }
+
+ private:
+  std::vector<grape::Bitset> vec_;
+  size_t paths_num_;
+};
+
 // DistinctCountBuilder for multiple sets together
 template <typename SET_TUPLE_T, int... TAG_IDs>
 class MultiColDistinctCountBuilder;
diff --git a/flex/engines/hqps_db/structures/multi_vertex_set/row_vertex_set.h b/flex/engines/hqps_db/structures/multi_vertex_set/row_vertex_set.h
index 87add4aa548f..3deef580fc44 100644
--- a/flex/engines/hqps_db/structures/multi_vertex_set/row_vertex_set.h
+++ b/flex/engines/hqps_db/structures/multi_vertex_set/row_vertex_set.h
@@ -19,12 +19,14 @@
 #include <tuple>
 #include <vector>
 
-#include "flex/engines/hqps_db/structures/collection.h"
 #include "grape/util.h"
 
 // Vertex set in with data in rows.
 namespace gs {
 
+template <typename T>
+class Collection;
+
 namespace internal {
 
 template <size_t Proj_Is, size_t My_Is, typename... DATA_T, typename... PropT>
diff --git a/flex/engines/hqps_db/structures/multi_vertex_set/two_label_vertex_set.h b/flex/engines/hqps_db/structures/multi_vertex_set/two_label_vertex_set.h
index 401d949c8793..c532ef94480e 100644
--- a/flex/engines/hqps_db/structures/multi_vertex_set/two_label_vertex_set.h
+++ b/flex/engines/hqps_db/structures/multi_vertex_set/two_label_vertex_set.h
@@ -644,6 +644,7 @@ class TwoLabelVertexSetImpl {
 
   static constexpr bool is_vertex_set = true;
   static constexpr bool is_general_set = false;
+  static constexpr bool is_row_vertex_set = false;
   static constexpr bool is_two_label_set = true;
   static constexpr size_t num_labels = 2;
   static constexpr size_t num_props = sizeof...(T);
@@ -943,6 +944,7 @@ class TwoLabelVertexSetImpl<VID_T, LabelT, grape::EmptyType> {
   static constexpr bool is_vertex_set = true;
   static constexpr bool is_general_set = false;
   static constexpr bool is_two_label_set = true;
+  static constexpr bool is_row_vertex_set = false;
   static constexpr size_t num_labels = 2;
   static constexpr size_t num_props = 0;
   static constexpr bool is_collection = false;
diff --git a/flex/engines/hqps_db/structures/path.h b/flex/engines/hqps_db/structures/path.h
index 3d067b2a2ce0..8b8b4d899737 100644
--- a/flex/engines/hqps_db/structures/path.h
+++ b/flex/engines/hqps_db/structures/path.h
@@ -19,6 +19,7 @@ limitations under the License.
 #include <memory>
 #include <string>
 #include <vector>
+#include "flex/engines/hqps_db/structures/multi_vertex_set/general_vertex_set.h"
 
 namespace gs {
 
@@ -34,6 +35,13 @@ struct Path {
   std::vector<VID_T> vids_;
   std::vector<label_id_t> label_ids_;
 
+  static Path<VID_T, LabelT> Null() {
+    static Path<VID_T, LabelT> null_path;
+    return null_path;
+  }
+
+  Path() = default;
+
   Path(const Path<VID_T, LabelT>& other)
       : vids_(other.vids_), label_ids_(other.label_ids_) {}
 
@@ -45,18 +53,25 @@ struct Path {
   Path(std::vector<VID_T>&& vids, std::vector<label_id_t>&& label_ids)
       : vids_(std::move(vids)), label_ids_(std::move(label_ids)) {}
 
-  inline int32_t length() const {
-    if (vids_.size() > 0) {
-      return vids_.size() - 1;
-    }
-    return 0;
-  }
+  inline int32_t length() const { return vids_.size(); }
 
   void EmplaceBack(VID_T vid, label_id_t label_id) {
     vids_.emplace_back(vid);
     label_ids_.emplace_back(label_id);
   }
 
+  std::pair<LabelT, VID_T> GetNode(size_t index) const {
+    CHECK(index < vids_.size());
+    return std::make_pair(label_ids_[index], vids_[index]);
+  }
+
+  void PopBack() {
+    if (vids_.size() > 0) {
+      vids_.pop_back();
+      label_ids_.pop_back();
+    }
+  }
+
   inline const std::vector<VID_T>& GetVertices() const { return vids_; }
 
   inline const std::vector<label_id_t>& GetLabels() const { return label_ids_; }
@@ -65,15 +80,26 @@ struct Path {
     CHECK(vids_.size() > 0);
     return vids_.back();
   }
+
+  inline LabelT GetEndLabel() const {
+    CHECK(label_ids_.size() > 0);
+    return label_ids_.back();
+  }
+
   inline VID_T GetStart() const {
     CHECK(vids_.size() > 0);
     return vids_.front();
   }
 
+  inline LabelT GetStartLabel() const {
+    CHECK(label_ids_.size() > 0);
+    return label_ids_.front();
+  }
+
   std::string to_string() const {
     std::stringstream ss;
     for (size_t i = 0; i < vids_.size(); ++i) {
-      ss << vids_[i];
+      ss << vids_[i] << "(label:" << std::to_string(label_ids_[i]) << ")";
       if (i + 1 < vids_.size()) {
         ss << "->";
       }
@@ -94,6 +120,28 @@ struct Path {
   }
 };
 
+template <typename VID_T, typename LabelT>
+inline bool operator==(const Path<VID_T, LabelT>& lhs,
+                       const Path<VID_T, LabelT>& rhs) {
+  if (lhs.vids_.size() != rhs.vids_.size() ||
+      lhs.label_ids_.size() != rhs.label_ids_.size()) {
+    return false;
+  }
+  for (size_t i = 0; i < lhs.vids_.size(); ++i) {
+    if (lhs.vids_[i] != rhs.vids_[i] ||
+        lhs.label_ids_[i] != rhs.label_ids_[i]) {
+      return false;
+    }
+  }
+  return true;
+}
+
+template <typename VID_T, typename LabelT>
+inline bool operator!=(const Path<VID_T, LabelT>& lhs,
+                       const Path<VID_T, LabelT>& rhs) {
+  return !(lhs == rhs);
+}
+
 template <typename VID_T, typename LabelT>
 class PathSetIter {
  public:
@@ -108,7 +156,7 @@ class PathSetIter {
   inline Path<VID_T, LabelT> GetData() const { return paths_[ind_]; }
 
   inline index_ele_tuple_t GetIndexElement() const {
-    return std::make_tuple(ind_, paths_[ind_]);
+    return std::make_pair(ind_, paths_[ind_]);
   }
 
   inline const self_type_t& operator++() {
@@ -147,6 +195,8 @@ class PathSet {
   using data_tuple_t = Path<VID_T, LabelT>;
   using index_ele_tuple_t = std::pair<size_t, Path<VID_T, LabelT>>;
 
+  static constexpr bool is_path_set = true;
+
   PathSet(std::vector<Path<VID_T, LabelT>>&& paths)
       : paths_(std::move(paths)) {}
 
@@ -165,6 +215,113 @@ class PathSet {
 
   iterator end() const { return iterator(paths_, paths_.size()); }
 
+  std::vector<LabelT> GetLabels(VOpt v_opt) const {
+    std::vector<bool> label_set(sizeof(LabelT) * 8, false);
+    if (v_opt == VOpt::End) {
+      for (auto i = 0; i < paths_.size(); ++i) {
+        label_set[paths_[i].GetLabels().back()] = true;
+      }
+    } else if (v_opt == VOpt::Start) {
+      for (auto i = 0; i < paths_.size(); ++i) {
+        label_set[paths_[i].GetLabels().front()] = true;
+      }
+    } else {
+      LOG(ERROR) << "Not supported vopt: " << gs::to_string(v_opt);
+      return {};
+    }
+    std::vector<LabelT> res;
+    for (size_t i = 0; i < label_set.size(); ++i) {
+      if (label_set[i]) {
+        res.emplace_back(i);
+      }
+    }
+    return res;
+  }
+
+  auto GetVertices(VOpt vopt, const std::vector<LabelT>& req_labels) const {
+    std::vector<vid_t> vids;
+    std::vector<offset_t> offsets;
+    offsets.reserve(Size() + 1);
+    offsets.emplace_back(0);
+
+    // intersect with req_labels
+    std::vector<bool> label_set(sizeof(LabelT) * 8, false);
+    std::vector<LabelT> labels;
+    std::vector<int32_t> label_to_index(sizeof(LabelT) * 8, -1);
+    if (req_labels.size() > 0) {
+      for (auto label : req_labels) {
+        label_set[label] = true;
+      }
+    } else {
+      std::fill(label_set.begin(), label_set.end(), true);  // empty means all
+    }
+    size_t valid_labels = 0;
+    for (int i = 0; i < label_set.size(); ++i) {
+      if (label_set[i]) {
+        labels.emplace_back(i);
+        label_to_index[i] = valid_labels++;
+      }
+    }
+
+    std::vector<grape::Bitset> label_bitsets;
+    label_bitsets.resize(labels.size());
+    for (auto i = 0; i < labels.size(); ++i) {
+      label_bitsets[i].init(paths_.size());
+    }
+    for (auto iter : *this) {
+      auto element = iter.GetElement();
+      if (vopt == VOpt::End) {
+        auto label_ind = label_to_index[element.GetEndLabel()];
+        if (label_ind >= 0) {
+          label_bitsets[label_ind].set_bit(vids.size());
+          vids.emplace_back(element.GetEnd());
+        }
+      } else if (vopt == VOpt::Start) {
+        // label_bitsets[element.GetStartLabel()].set_bit(vids.size());
+        auto label_ind = label_to_index[element.GetStartLabel()];
+        if (label_ind >= 0) {
+          label_bitsets[label_ind].set_bit(vids.size());
+          vids.emplace_back(element.GetStart());
+        }
+      } else {
+        LOG(ERROR) << "Not supported vopt: " << gs::to_string(vopt);
+      }
+      offsets.emplace_back(vids.size());
+    }
+
+    for (auto i = 0; i < labels.size(); ++i) {
+      label_bitsets[i].resize(vids.size());
+    }
+    auto general_set = make_general_set(std::move(vids), std::move(labels),
+                                        std::move(label_bitsets));
+    return std::make_pair(general_set, std::move(offsets));
+  }
+
+  // project my self.
+  template <int tag_id, int Fs,
+            typename std::enable_if<Fs == -1>::type* = nullptr>
+  self_type_t ProjectWithRepeatArray(const std::vector<size_t>& repeat_array,
+                                     KeyAlias<tag_id, Fs>& key_alias) const {
+    std::vector<Path<VID_T, LabelT>> res;
+    for (size_t i = 0; i < repeat_array.size(); ++i) {
+      for (size_t j = 0; j < repeat_array[i]; ++j) {
+        res.push_back(paths_[i]);
+      }
+    }
+    return self_type_t(std::move(res));
+  }
+
+  template <size_t col_ind, typename... index_ele_tuple_t_>
+  flat_t Flat(std::vector<std::tuple<index_ele_tuple_t_...>>& index_ele_tuple) {
+    static_assert(col_ind <
+                  std::tuple_size_v<std::tuple<index_ele_tuple_t_...>>);
+    std::vector<Path<VID_T, LabelT>> res;
+    for (auto& ele : index_ele_tuple) {
+      res.push_back(std::get<1>(std::get<col_ind>(ele)));
+    }
+    return flat_t(std::move(res));
+  }
+
  private:
   std::vector<Path<VID_T, LabelT>> paths_;
 };
@@ -237,7 +394,7 @@ class CompressedPathSet {
         offsets_(std::move(offsets)),
         labels_(std::move(labels)),
         min_len_(min_len) {
-    CHECK(min_len_ < vids_.size() && min_len_ > 0);
+    CHECK(min_len_ < vids_.size());
     // vids.size() is the max length of path.
     // offset[i].back() is the number of paths with length i.
     CHECK(vids_.size() == offsets_.size())
@@ -272,12 +429,23 @@ class CompressedPathSet {
 
   const std::vector<LabelT>& GetLabels() const { return labels_; }
 
+  std::vector<LabelT> GetLabels(VOpt v_opt) const {
+    if (v_opt == VOpt::End) {
+      return {labels_.back()};
+    } else if (v_opt == VOpt::Start) {
+      return {labels_.front()};
+    } else {
+      LOG(ERROR) << "Not supported vopt: " << gs::to_string(v_opt);
+      return {};
+    }
+  }
+
   template <typename FILTER_T, typename PROP_GETTER_T>
   std::pair<DefaultRowVertexSet<LabelT, vid_t>, std::vector<offset_t>>
   GetVertices(VOpt vopt, const FILTER_T& expr,
               const std::vector<PROP_GETTER_T>& prop_getters) const {
-    // get vertices from path set, current we only have one label is path, so we
-    // don't have label params.
+    // get vertices from path set, current we only have one label is path, so
+    // we don't have label params.
     std::vector<vid_t> vids;
     std::vector<offset_t> offsets;
     offsets.reserve(Size() + 1);
@@ -362,6 +530,20 @@ class CompressedPathSet {
     return res;
   }
 
+  template <int tag_id, int Fs,
+            typename std::enable_if<Fs == -1>::type* = nullptr>
+  flat_t ProjectWithRepeatArray(const std::vector<size_t>& repeat_array,
+                                KeyAlias<tag_id, Fs>& key_alias) const {
+    std::vector<Path<VID_T, LabelT>> res;
+    auto valid_paths = get_all_valid_paths();
+    for (auto i = 0; i < repeat_array.size(); ++i) {
+      for (auto j = 0; j < repeat_array[i]; ++j) {
+        res.push_back(valid_paths[i]);
+      }
+    }
+    return flat_t(std::move(res));
+  }
+
  private:
   std::vector<std::vector<VID_T>> vids_;
   std::vector<std::vector<offset_t>> offsets_;
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/simple/SimpleMatchQueries.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/simple/SimpleMatchQueries.java
index 2a5cf5894f9a..de26d63988aa 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/simple/SimpleMatchQueries.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/simple/SimpleMatchQueries.java
@@ -180,4 +180,39 @@ public static QueryContext get_simple_match_query_13_test() {
                         "Record<{aId: 0, bId: 1454}>");
         return new QueryContext(query, expected);
     }
+
+    public static QueryContext get_simple_match_query_14_test() {
+        String query =
+                "Match (a)-[c*0..2]->(b) RETURN a.id AS aId, c, b.id AS bId ORDER BY aId"
+                        + " ASC, bId ASC LIMIT 5;";
+        List<String> expected =
+                Arrays.asList(
+                        "Record<{aId: 0, c: path[], bId: 0}>",
+                        "Record<{aId: 0, c: path[], bId: 0}>",
+                        "Record<{aId: 0, c: path[], bId: 0}>",
+                        "Record<{aId: 0, c: path[], bId: 0}>",
+                        "Record<{aId: 0, c: path[], bId: 3}>");
+        return new QueryContext(query, expected);
+    }
+
+    public static QueryContext get_simple_match_query_15_test() {
+        String query =
+                "Match (a:PERSON)-[c:KNOWS*1..2]->(b:PERSON) RETURN a.id AS aId, c, b.id AS bId"
+                        + " ORDER BY aId ASC, bId ASC LIMIT 5;";
+
+        List<String> expected =
+                Arrays.asList(
+                        "Record<{aId: 94, c: path[], bId: 987}>",
+                        "Record<{aId: 94, c: path[], bId: 2199023256154}>",
+                        "Record<{aId: 94, c: path[], bId: 2199023256776}>",
+                        "Record<{aId: 94, c: path[], bId: 6597069767432}>",
+                        "Record<{aId: 94, c: path[], bId: 8796093022369}>");
+        return new QueryContext(query, expected);
+    }
+
+    public static QueryContext get_simple_match_query_16_test() {
+        String query = "Match (a)-[c*0..2]->(b) RETURN COUNT (DISTINCT c);";
+        List<String> expected = Arrays.asList("Record<{$f0: 1552803}>");
+        return new QueryContext(query, expected);
+    }
 }


Test patch:
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/ldbc/SimpleMatchTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/ldbc/SimpleMatchTest.java
index b8f105eadb9a..fd10f8df16b3 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/ldbc/SimpleMatchTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/ldbc/SimpleMatchTest.java
@@ -128,6 +128,27 @@ public void run_simple_match_13_test() {
         Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
     }
 
+    @Test
+    public void run_simple_match_14_test() {
+        QueryContext testQuery = SimpleMatchQueries.get_simple_match_query_14_test();
+        Result result = session.run(testQuery.getQuery());
+        Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
+    }
+
+    @Test
+    public void run_simple_match_15_test() {
+        QueryContext testQuery = SimpleMatchQueries.get_simple_match_query_15_test();
+        Result result = session.run(testQuery.getQuery());
+        Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
+    }
+
+    @Test
+    public void run_simple_match_16_test() {
+        QueryContext testQuery = SimpleMatchQueries.get_simple_match_query_16_test();
+        Result result = session.run(testQuery.getQuery());
+        Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
+    }
+
     @AfterClass
     public static void afterClass() {
         if (session != null) {