edge

alibaba · Jan 19, 2025 · 1a3820d · 1a3820d
1 parent b67c292
commit 1a3820d
Show file tree

Hide file tree

Showing 5 changed files with 107 additions and 23 deletions.
diff --git a/flex/engines/graph_db/runtime/common/operators/retrieve/edge_expand.cc b/flex/engines/graph_db/runtime/common/operators/retrieve/edge_expand.cc
@@ -143,6 +143,47 @@ static Context expand_edge_without_predicate_optional_impl(
                        }
                      });
 
+      ctx.set_with_reshuffle(params.alias, builder.finish(), shuffle_offset);
+      return ctx;
+    } else if (params.dir == Direction::kIn) {
+      auto& input_vertex_list =
+          *std::dynamic_pointer_cast<IVertexColumn>(ctx.get(params.v_tag));
+      CHECK(!input_vertex_list.is_optional())
+          << "not support optional vertex column as input currently";
+      auto& triplet = params.labels[0];
+      auto props = graph.schema().get_edge_properties(
+          triplet.src_label, triplet.dst_label, triplet.edge_label);
+      PropertyType pt = PropertyType::kEmpty;
+      if (!props.empty()) {
+        pt = props[0];
+      }
+      if (props.size() > 1) {
+        pt = PropertyType::kRecordView;
+      }
+      OptionalSDSLEdgeColumnBuilder builder(Direction::kIn, triplet, pt);
+      foreach_vertex(input_vertex_list,
+                     [&](size_t index, label_t label, vid_t v) {
+                       if (label == triplet.dst_label) {
+                         auto ie_iter = graph.GetInEdgeIterator(
+                             label, v, triplet.src_label, triplet.edge_label);
+                         bool has_edge = false;
+                         while (ie_iter.IsValid()) {
+                           auto nbr = ie_iter.GetNeighbor();
+                           builder.push_back_opt(nbr, v, ie_iter.GetData());
+                           shuffle_offset.push_back(index);
+                           ie_iter.Next();
+                           has_edge = true;
+                         }
+                         if (!has_edge) {
+                           builder.push_back_null();
+                           shuffle_offset.push_back(index);
+                         }
+                       } else {
+                         builder.push_back_null();
+                         shuffle_offset.push_back(index);
+                       }
+                     });
+
       ctx.set_with_reshuffle(params.alias, builder.finish(), shuffle_offset);
       return ctx;
     }

diff --git a/flex/engines/graph_db/runtime/common/operators/retrieve/edge_expand.h b/flex/engines/graph_db/runtime/common/operators/retrieve/edge_expand.h
@@ -260,7 +260,7 @@ class EdgeExpand {
                                const EdgeExpandParams& params,
                                const PRED_T& pred) {
     if (params.is_optional) {
-      LOG(FATAL) << "not support optional edge expand";
+      LOG(FATAL) << "not support optional edge expand with predicate";
     }
     std::shared_ptr<IVertexColumn> input_vertex_list =
         std::dynamic_pointer_cast<IVertexColumn>(ctx.get(params.v_tag));

diff --git a/flex/engines/graph_db/runtime/common/operators/retrieve/edge_expand_impl.cc b/flex/engines/graph_db/runtime/common/operators/retrieve/edge_expand_impl.cc
@@ -213,8 +213,12 @@ expand_vertex_without_predicate_optional_impl(
     }
   }
   LOG(INFO) << "ed_types.size() " << se << " " << sp;
-  LOG(FATAL) << "not implemented" << label_dirs.size();
-  return std::make_pair(nullptr, std::vector<size_t>());
+  int label_num = graph.schema().vertex_label_num();
+  std::vector<std::vector<std::tuple<label_t, label_t, Direction>>> _label_dirs(
+      label_num);
+  _label_dirs[input_label] = label_dirs;
+  return expand_vertex_optional_impl<DummyPredicate<Any>>(
+      graph, input, _label_dirs, DummyPredicate<Any>());
 }
 
 std::pair<std::shared_ptr<IContextColumn>, std::vector<size_t>>
@@ -369,12 +373,8 @@ expand_vertex_without_predicate_optional_impl(
       }
     }
   }
-  bool se = true;
   for (auto& vec : label_dirs) {
     grape::DistinctSort(vec);
-    if (vec.size() > 1) {
-      se = false;
-    }
   }
   bool sp = true;
   if (ed_types.size() == 0) {
@@ -394,16 +394,24 @@ expand_vertex_without_predicate_optional_impl(
   if (sp && (!check_exist_special_edge(graph, labels, dir))) {
     const PropertyType& ed_type = ed_types[0];
     if (ed_type == PropertyType::Empty()) {
-      if (se) {
-        LOG(FATAL) << "not implemented";
-      } else {
-        return expand_vertex_np_me_sp_optional<
-            grape::EmptyType, DummyPredicate<grape::EmptyType>>(
-            graph, input, label_dirs, DummyPredicate<grape::EmptyType>());
-      }
+      return expand_vertex_np_me_sp_optional<grape::EmptyType,
+                                             DummyPredicate<grape::EmptyType>>(
+          graph, input, label_dirs, DummyPredicate<grape::EmptyType>());
+    } else if (ed_type == PropertyType::Date()) {
+      return expand_vertex_np_me_sp_optional<Date, DummyPredicate<Date>>(
+          graph, input, label_dirs, DummyPredicate<Date>());
+    } else if (ed_type == PropertyType::Int32()) {
+      return expand_vertex_np_me_sp_optional<int, DummyPredicate<int>>(
+          graph, input, label_dirs, DummyPredicate<int>());
+    } else if (ed_type == PropertyType::Int64()) {
+      return expand_vertex_np_me_sp_optional<int64_t, DummyPredicate<int64_t>>(
+          graph, input, label_dirs, DummyPredicate<int64_t>());
+    } else {
+      LOG(INFO) << "type - " << ed_type << " - not implemented, fallback";
     }
   }
-  return {nullptr, {}};
+  return expand_vertex_optional_impl<DummyPredicate<Any>>(
+      graph, input, label_dirs, DummyPredicate<Any>());
 }
 
 std::pair<std::shared_ptr<IContextColumn>, std::vector<size_t>>

diff --git a/flex/engines/graph_db/runtime/common/operators/retrieve/edge_expand_impl.h b/flex/engines/graph_db/runtime/common/operators/retrieve/edge_expand_impl.h
@@ -288,7 +288,6 @@ expand_vertex_np_me_sp_optional(
   }
   if (single_nbr_label) {
     OptionalSLVertexColumnBuilder builder(nbr_labels[0]);
-#if 1
     foreach_vertex(input, [&](size_t idx, label_t l, vid_t v) {
       if (!input.has_value(idx)) {
         builder.push_back_null();
@@ -318,8 +317,7 @@ expand_vertex_np_me_sp_optional(
       }
       ++idx;
     });
-#else
-#endif
+
     col = builder.finish();
   } else {
     OptionalMLVertexColumnBuilder builder;
@@ -919,6 +917,48 @@ expand_vertex_np_me_mp(
   return std::make_pair(builder.finish(), std::move(offsets));
 }
 
+template <typename PRED_T>
+inline std::pair<std::shared_ptr<IContextColumn>, std::vector<size_t>>
+expand_vertex_optional_impl(
+    const GraphReadInterface& graph, const IVertexColumn& input,
+    const std::vector<std::vector<std::tuple<label_t, label_t, Direction>>>&
+        label_dirs,
+    const PRED_T& pred) {
+  OptionalMLVertexColumnBuilder builder;
+  std::vector<size_t> offsets;
+  foreach_vertex(input, [&](size_t idx, label_t label, vid_t v) {
+    if (!input.has_value(idx)) {
+      builder.push_back_null();
+      offsets.push_back(idx);
+      return;
+    }
+    bool has_nbr = false;
+    for (auto& t : label_dirs[label]) {
+      label_t nbr_label = std::get<0>(t);
+      label_t edge_label = std::get<1>(t);
+      Direction dir = std::get<2>(t);
+      auto it =
+          (dir == Direction::kOut)
+              ? (graph.GetOutEdgeIterator(label, v, nbr_label, edge_label))
+              : (graph.GetInEdgeIterator(label, v, nbr_label, edge_label));
+      while (it.IsValid()) {
+        auto nbr = it.GetNeighbor();
+        if (pred(label, v, nbr_label, nbr, edge_label, dir, it.GetData())) {
+          builder.push_back_vertex({nbr_label, nbr});
+          offsets.push_back(idx);
+          has_nbr = true;
+        }
+        it.Next();
+      }
+    }
+    if (!has_nbr) {
+      builder.push_back_null();
+      offsets.push_back(idx);
+    }
+  });
+  return std::make_pair(builder.finish(), std::move(offsets));
+}
+
 template <typename GPRED_T, typename EDATA_T>
 struct GPredWrapper {
   GPredWrapper(const GPRED_T& gpred) : gpred_(gpred) {}

diff --git a/flex/engines/graph_db/runtime/execute/ops/retrieve/project.cc b/flex/engines/graph_db/runtime/execute/ops/retrieve/project.cc
@@ -1457,11 +1457,6 @@ ProjectOrderByOprBuilder::Build(const gs::Schema& schema,
                               std::move(exprs), index_set, order_by_pairs,
                               lower, upper, first_tuple),
                           ret_meta);
-    /*return std::make_pair(
-        std::make_unique<ProjectOrderByOpr>(
-            plan.plan(op_idx).opr().project(),
-            plan.plan(op_idx + 1).opr().order_by(), data_types),
-        ret_meta);*/
   } else {
     return std::make_pair(nullptr, ContextMeta());
   }