fix(interactive): Support sinking paths in Interactive (#4541)

zhanglei1949 · liulx20 · web-flow · commit 3b212c20fe92 · 2025-03-05T10:53:40.000+08:00
Fix #4540 --------- Co-authored-by: liulx20 <519459125@qq.com>
diff --git a/flex/engines/graph_db/runtime/common/operators/retrieve/path_expand.cc b/flex/engines/graph_db/runtime/common/operators/retrieve/path_expand.cc
@@ -203,7 +203,8 @@ bl::result<Context> PathExpand::edge_expand_p(const GraphReadInterface& graph,
                                                       label_triplet.edge_label);
               while (oe_iter.IsValid()) {
                 std::unique_ptr<PathImpl> new_path = path->expand(
-                    label_triplet.dst_label, oe_iter.GetNeighbor());
+                    label_triplet.edge_label, label_triplet.dst_label,
+                    oe_iter.GetNeighbor());
                 output.emplace_back(std::move(new_path), index);
                 oe_iter.Next();
               }
@@ -250,7 +251,8 @@ bl::result<Context> PathExpand::edge_expand_p(const GraphReadInterface& graph,
                                                      label_triplet.edge_label);
               while (ie_iter.IsValid()) {
                 std::unique_ptr<PathImpl> new_path = path->expand(
-                    label_triplet.src_label, ie_iter.GetNeighbor());
+                    label_triplet.edge_label, label_triplet.src_label,
+                    ie_iter.GetNeighbor());
                 output.emplace_back(std::move(new_path), index);
                 ie_iter.Next();
               }
@@ -296,7 +298,8 @@ bl::result<Context> PathExpand::edge_expand_p(const GraphReadInterface& graph,
                                                       label_triplet.dst_label,
                                                       label_triplet.edge_label);
               while (oe_iter.IsValid()) {
-                auto new_path = path->expand(label_triplet.dst_label,
+                auto new_path = path->expand(label_triplet.edge_label,
+                                             label_triplet.dst_label,
                                              oe_iter.GetNeighbor());
                 output.emplace_back(std::move(new_path), index);
                 oe_iter.Next();
@@ -307,7 +310,8 @@ bl::result<Context> PathExpand::edge_expand_p(const GraphReadInterface& graph,
                                                      label_triplet.src_label,
                                                      label_triplet.edge_label);
               while (ie_iter.IsValid()) {
-                auto new_path = path->expand(label_triplet.src_label,
+                auto new_path = path->expand(label_triplet.edge_label,
+                                             label_triplet.src_label,
                                              ie_iter.GetNeighbor());
                 output.emplace_back(std::move(new_path), index);
                 ie_iter.Next();
@@ -503,7 +507,8 @@ bl::result<Context> PathExpand::single_source_single_dest_shortest_path(
                                                      dest.second, path)) {
       builder.push_back_opt(dest.second);
       shuffle_offset.push_back(index);
-      auto impl = PathImpl::make_path_impl(label_triplet.src_label, path);
+      auto impl = PathImpl::make_path_impl(label_triplet.src_label,
+                                           label_triplet.edge_label, path);
       path_builder.push_back_opt(Path(impl.get()));
       arena->emplace_back(std::move(impl));
     }
@@ -752,7 +757,8 @@ bl::result<Context> PathExpand::all_shortest_paths_with_given_source_and_dest(
     all_shortest_path_with_given_source_and_dest_impl(graph, params, v,
                                                       dest.second, paths);
     for (auto& path : paths) {
-      auto ptr = PathImpl::make_path_impl(label_triplet.src_label, path);
+      auto ptr = PathImpl::make_path_impl(label_triplet.src_label,
+                                          label_triplet.edge_label, path);
       builder.push_back_opt(dest.second);
       path_builder.push_back_opt(Path(ptr.get()));
       arena->emplace_back(std::move(ptr));
diff --git a/flex/engines/graph_db/runtime/common/operators/retrieve/path_expand_impl.h b/flex/engines/graph_db/runtime/common/operators/retrieve/path_expand_impl.h
@@ -177,7 +177,7 @@ path_expand_vertex_without_predicate_impl(
 
 template <typename EDATA_T, typename PRED_T>
 void sssp_dir(const GraphReadInterface::graph_view_t<EDATA_T>& view,
-              label_t v_label, vid_t v,
+              label_t v_label, vid_t v, label_t e_label,
               const GraphReadInterface::vertex_set_t& vertices, size_t idx,
               int lower, int upper, SLVertexColumnBuilder& dest_col_builder,
               GeneralPathColumnBuilder& path_col_builder, Arena& path_impls,
@@ -202,7 +202,7 @@ void sssp_dir(const GraphReadInterface::graph_view_t<EDATA_T>& view,
             }
 
             dest_col_builder.push_back_opt(u);
-            auto impl = PathImpl::make_path_impl(v_label, path);
+            auto impl = PathImpl::make_path_impl(v_label, e_label, path);
             path_col_builder.push_back_opt(Path(impl.get()));
             path_impls.emplace_back(std::move(impl));
             offsets.push_back(idx);
@@ -219,7 +219,7 @@ void sssp_dir(const GraphReadInterface::graph_view_t<EDATA_T>& view,
             }
 
             dest_col_builder.push_back_opt(u);
-            auto impl = PathImpl::make_path_impl(v_label, path);
+            auto impl = PathImpl::make_path_impl(v_label, e_label, path);
             path_col_builder.push_back_opt(Path(impl.get()));
             path_impls.emplace_back(std::move(impl));
             offsets.push_back(idx);
@@ -253,7 +253,7 @@ void sssp_dir(const GraphReadInterface::graph_view_t<EDATA_T>& view,
 template <typename EDATA_T, typename PRED_T>
 void sssp_both_dir(const GraphReadInterface::graph_view_t<EDATA_T>& view0,
                    const GraphReadInterface::graph_view_t<EDATA_T>& view1,
-                   label_t v_label, vid_t v,
+                   label_t v_label, vid_t v, label_t e_label,
                    const GraphReadInterface::vertex_set_t& vertices, size_t idx,
                    int lower, int upper,
                    SLVertexColumnBuilder& dest_col_builder,
@@ -280,7 +280,7 @@ void sssp_both_dir(const GraphReadInterface::graph_view_t<EDATA_T>& view0,
             }
 
             dest_col_builder.push_back_opt(u);
-            auto impl = PathImpl::make_path_impl(v_label, path);
+            auto impl = PathImpl::make_path_impl(v_label, e_label, path);
             path_col_builder.push_back_opt(Path(impl.get()));
             path_impls.emplace_back(std::move(impl));
             offsets.push_back(idx);
@@ -297,7 +297,7 @@ void sssp_both_dir(const GraphReadInterface::graph_view_t<EDATA_T>& view0,
             }
 
             dest_col_builder.push_back_opt(u);
-            auto impl = PathImpl::make_path_impl(v_label, path);
+            auto impl = PathImpl::make_path_impl(v_label, e_label, path);
             path_col_builder.push_back_opt(Path(impl.get()));
             path_impls.emplace_back(std::move(impl));
             offsets.push_back(idx);
@@ -467,8 +467,8 @@ single_source_shortest_path_impl(const GraphReadInterface& graph,
             ? graph.GetIncomingGraphView<EDATA_T>(v_label, v_label, e_label)
             : graph.GetOutgoingGraphView<EDATA_T>(v_label, v_label, e_label);
     foreach_vertex(input, [&](size_t idx, label_t label, vid_t v) {
-      sssp_dir(view, label, v, vertices, idx, lower, upper, dest_col_builder,
-               path_col_builder, *path_impls, offsets, pred);
+      sssp_dir(view, label, v, e_label, vertices, idx, lower, upper,
+               dest_col_builder, path_col_builder, *path_impls, offsets, pred);
     });
   } else {
     CHECK(dir == Direction::kBoth);
@@ -477,9 +477,9 @@ single_source_shortest_path_impl(const GraphReadInterface& graph,
     auto ie_view =
         graph.GetIncomingGraphView<EDATA_T>(v_label, v_label, e_label);
     foreach_vertex(input, [&](size_t idx, label_t label, vid_t v) {
-      sssp_both_dir(oe_view, ie_view, v_label, v, vertices, idx, lower, upper,
-                    dest_col_builder, path_col_builder, *path_impls, offsets,
-                    pred);
+      sssp_both_dir(oe_view, ie_view, v_label, v, e_label, vertices, idx, lower,
+                    upper, dest_col_builder, path_col_builder, *path_impls,
+                    offsets, pred);
     });
   }
   return std::make_tuple(dest_col_builder.finish(nullptr),
@@ -529,44 +529,55 @@ default_single_source_shortest_path_impl(
         SLVertexColumnBuilder::builder(*dest_labels.begin());
 
     foreach_vertex(input, [&](size_t idx, label_t label, vid_t v) {
-      std::vector<std::pair<label_t, vid_t>> cur;
-      std::vector<std::pair<label_t, vid_t>> next;
-      cur.emplace_back(label, v);
-      std::map<std::pair<label_t, vid_t>, std::pair<label_t, vid_t>> parent;
+      std::vector<std::tuple<label_t, label_t, vid_t>> cur;
+      std::vector<std::tuple<label_t, label_t, vid_t>> next;
+      cur.emplace_back(std::numeric_limits<label_t>::max(), label, v);
+      std::map<std::tuple<label_t, label_t, vid_t>,
+               std::tuple<label_t, label_t, vid_t>>
+          parent;
+      std::set<std::pair<label_t, vid_t>> visited;
+      visited.insert(std::make_pair(label, v));
       int depth = 0;
       while (depth < upper && !cur.empty()) {
-        for (auto u : cur) {
-          if (depth >= lower && pred(u.first, u.second)) {
+        for (auto [edge_label, v_label, vid] : cur) {
+          if (depth >= lower && pred(v_label, vid)) {
             std::vector<VertexRecord> path;
-            auto x = u;
-            while (!(x.first == label && x.second == v)) {
-              path.emplace_back(VertexRecord{x.first, x.second});
+            std::vector<label_t> edge_labels;
+            auto x = std::tie(edge_label, label, vid);
+            while (!(v_label == label && vid == v)) {
+              path.push_back(VertexRecord{std::get<1>(x), std::get<2>(x)});
+              edge_labels.push_back(std::get<0>(x));
               x = parent[x];
             }
             path.emplace_back(VertexRecord{label, v});
+            std::reverse(edge_labels.begin(), edge_labels.end());
             std::reverse(path.begin(), path.end());
 
             if (path.size() > 1) {
-              auto impl = PathImpl::make_path_impl(std::move(path));
+              auto impl = PathImpl::make_path_impl(edge_labels, path);
+
               path_col_builder.push_back_opt(Path(impl.get()));
               path_impls->emplace_back(std::move(impl));
 
-              dest_col_builder.push_back_opt(u.second);
+              dest_col_builder.push_back_opt(vid);
               offsets.push_back(idx);
             }
           }
 
-          for (auto& l : labels_map[u.first]) {
+          for (auto& l : labels_map[v_label]) {
             label_t nbr_label = std::get<0>(l);
             auto iter = (std::get<2>(l) == Direction::kOut)
-                            ? graph.GetOutEdgeIterator(
-                                  u.first, u.second, nbr_label, std::get<1>(l))
-                            : graph.GetInEdgeIterator(
-                                  u.first, u.second, nbr_label, std::get<1>(l));
+                            ? graph.GetOutEdgeIterator(v_label, vid, nbr_label,
+                                                       std::get<1>(l))
+                            : graph.GetInEdgeIterator(v_label, vid, nbr_label,
+                                                      std::get<1>(l));
             while (iter.IsValid()) {
-              auto nbr = std::make_pair(nbr_label, iter.GetNeighbor());
-              if (parent.find(nbr) == parent.end()) {
-                parent[nbr] = u;
+              auto nbr = std::make_tuple(std::get<1>(l), nbr_label,
+                                         iter.GetNeighbor());
+              auto vertex = std::make_pair(nbr_label, iter.GetNeighbor());
+              if (visited.find(vertex) == visited.end()) {
+                visited.insert(vertex);
+                parent[nbr] = std::tie(edge_label, v_label, vid);
                 next.push_back(nbr);
               }
               iter.Next();
@@ -585,45 +596,54 @@ default_single_source_shortest_path_impl(
     auto dest_col_builder = MLVertexColumnBuilder::builder();
 
     foreach_vertex(input, [&](size_t idx, label_t label, vid_t v) {
-      std::vector<std::pair<label_t, vid_t>> cur;
-      std::vector<std::pair<label_t, vid_t>> next;
-      cur.emplace_back(label, v);
-      std::map<std::pair<label_t, vid_t>, std::pair<label_t, vid_t>> parent;
+      std::vector<std::tuple<label_t, label_t, vid_t>> cur;
+      std::vector<std::tuple<label_t, label_t, vid_t>> next;
+      cur.emplace_back(std::numeric_limits<label_t>::max(), label, v);
+      std::map<std::tuple<label_t, label_t, vid_t>,
+               std::tuple<label_t, label_t, vid_t>>
+          parent;
+      std::set<std::pair<label_t, vid_t>> visited;
+      visited.insert(std::make_pair(label, v));
       int depth = 0;
       while (depth < upper && !cur.empty()) {
-        for (auto u : cur) {
-          if (depth >= lower && pred(u.first, u.second)) {
+        for (auto [edge_label, v_label, vid] : cur) {
+          if (depth >= lower && pred(v_label, vid)) {
             std::vector<VertexRecord> path;
-            auto x = u;
-            while (!(x.first == label && x.second == v)) {
-              path.emplace_back(VertexRecord{x.first, x.second});
+            std::vector<label_t> edge_labels;
+            auto x = std::tie(edge_label, v_label, vid);
+            while (!(v_label == label && vid == v)) {
+              path.push_back(VertexRecord{std::get<1>(x), std::get<2>(x)});
+              edge_labels.push_back(std::get<0>(x));
               x = parent[x];
             }
             path.emplace_back(VertexRecord{label, v});
+            std::reverse(edge_labels.begin(), edge_labels.end());
             std::reverse(path.begin(), path.end());
 
             if (path.size() > 1) {
-              auto impl = PathImpl::make_path_impl(std::move(path));
-
+              auto impl = PathImpl::make_path_impl(edge_labels, path);
               path_col_builder.push_back_opt(Path(impl.get()));
               path_impls->emplace_back(std::move(impl));
 
-              dest_col_builder.push_back_vertex({u.first, u.second});
+              dest_col_builder.push_back_vertex({v_label, vid});
               offsets.push_back(idx);
             }
           }
 
-          for (auto& l : labels_map[u.first]) {
+          for (auto& l : labels_map[v_label]) {
             label_t nbr_label = std::get<0>(l);
             auto iter = (std::get<2>(l) == Direction::kOut)
-                            ? graph.GetOutEdgeIterator(
-                                  u.first, u.second, nbr_label, std::get<1>(l))
-                            : graph.GetInEdgeIterator(
-                                  u.first, u.second, nbr_label, std::get<1>(l));
+                            ? graph.GetOutEdgeIterator(v_label, vid, nbr_label,
+                                                       std::get<1>(l))
+                            : graph.GetInEdgeIterator(v_label, vid, nbr_label,
+                                                      std::get<1>(l));
             while (iter.IsValid()) {
-              auto nbr = std::make_pair(nbr_label, iter.GetNeighbor());
-              if (parent.find(nbr) == parent.end()) {
-                parent[nbr] = u;
+              auto nbr = std::make_tuple(std::get<1>(l), nbr_label,
+                                         iter.GetNeighbor());
+              auto vertex = std::make_pair(nbr_label, iter.GetNeighbor());
+              if (visited.find(vertex) == visited.end()) {
+                visited.insert(vertex);
+                parent[nbr] = std::tie(edge_label, v_label, vid);
                 next.push_back(nbr);
               }
               iter.Next();
diff --git a/flex/engines/graph_db/runtime/common/rt_any.cc b/flex/engines/graph_db/runtime/common/rt_any.cc
@@ -846,8 +846,43 @@ void RTAny::sink(const GraphReadInterface& graph, int id,
       }
     }
   } else if (type_ == RTAnyType::kPath) {
-    LOG(FATAL) << "not support path sink";
-
+    auto mutable_path =
+        col->mutable_entry()->mutable_element()->mutable_graph_path();
+    auto path_nodes = this->as_path().nodes();
+    auto edge_labels = this->as_path().edge_labels();
+    // same label for all edges
+    if (edge_labels.size() == 1) {
+      for (size_t i = 0; i + 2 < path_nodes.size(); ++i) {
+        edge_labels.emplace_back(edge_labels[0]);
+      }
+    }
+    assert(edge_labels.size() + 1 == path_nodes.size());
+    size_t len = path_nodes.size();
+    for (size_t i = 0; i + 1 < len; ++i) {
+      auto vertex_in_path = mutable_path->add_path();
+
+      auto node = vertex_in_path->mutable_vertex();
+      node->mutable_label()->set_id(path_nodes[i].label());
+      node->set_id(
+          encode_unique_vertex_id(path_nodes[i].label(), path_nodes[i].vid()));
+      auto edge_in_path = mutable_path->add_path();
+
+      auto edge = edge_in_path->mutable_edge();
+      edge->mutable_src_label()->set_id(path_nodes[i].label());
+      edge->mutable_dst_label()->set_id(path_nodes[i + 1].label());
+      edge->mutable_label()->set_id(edge_labels[i]);
+      edge->set_id(encode_unique_edge_id(edge_labels[i], path_nodes[i].vid(),
+                                         path_nodes[i + 1].vid()));
+      edge->set_src_id(
+          encode_unique_vertex_id(path_nodes[i].label(), path_nodes[i].vid()));
+      edge->set_dst_id(encode_unique_vertex_id(path_nodes[i + 1].label(),
+                                               path_nodes[i + 1].vid()));
+    }
+    auto vertex_in_path = mutable_path->add_path();
+    auto node = vertex_in_path->mutable_vertex();
+    node->mutable_label()->set_id(path_nodes[len - 1].label());
+    node->set_id(encode_unique_vertex_id(path_nodes[len - 1].label(),
+                                         path_nodes[len - 1].vid()));
   } else {
     sink_impl(col->mutable_entry()->mutable_element()->mutable_object());
   }
diff --git a/flex/engines/graph_db/runtime/common/rt_any.h b/flex/engines/graph_db/runtime/common/rt_any.h
@@ -91,24 +91,27 @@ class PathImpl : public CObject {
     return new_path;
   }
   static std::unique_ptr<PathImpl> make_path_impl(
-      label_t label, std::vector<vid_t>& path_ids) {
+      label_t label, label_t edge_label, std::vector<vid_t>& path_ids) {
     auto new_path = std::make_unique<PathImpl>();
     for (auto id : path_ids) {
       new_path->path_.push_back({label, id});
     }
+    new_path->edge_labels_.push_back(edge_label);
     return new_path;
   }
-
   static std::unique_ptr<PathImpl> make_path_impl(
+      const std::vector<label_t>& edge_labels,
       const std::vector<VertexRecord>& path) {
     auto new_path = std::make_unique<PathImpl>();
-    new_path->path_.insert(new_path->path_.end(), path.begin(), path.end());
+    new_path->path_ = path;
+    new_path->edge_labels_ = edge_labels;
     return new_path;
   }
-
-  std::unique_ptr<PathImpl> expand(label_t label, vid_t v) const {
+  std::unique_ptr<PathImpl> expand(label_t edge_label, label_t label,
+                                   vid_t v) const {
     auto new_path = std::make_unique<PathImpl>();
     new_path->path_ = path_;
+    new_path->edge_labels_.emplace_back(edge_label);
     new_path->path_.push_back({label, v});
     return new_path;
   }
@@ -130,6 +133,7 @@ class PathImpl : public CObject {
   bool operator<(const PathImpl& p) const { return path_ < p.path_; }
   bool operator==(const PathImpl& p) const { return path_ == p.path_; }
   std::vector<VertexRecord> path_;
+  std::vector<label_t> edge_labels_;
 };
 class Path {
  public:
@@ -155,6 +159,8 @@ class Path {
 
   std::vector<VertexRecord> nodes() { return impl_->path_; }
 
+  std::vector<label_t> edge_labels() const { return impl_->edge_labels_; }
+
   VertexRecord get_start() const { return impl_->get_start(); }
   bool operator<(const Path& p) const { return *impl_ < *(p.impl_); }
   bool operator==(const Path& p) const { return *(impl_) == *(p.impl_); }
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
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
