From 82401493ce48c2365f9920aa32562c45b2e22074 Mon Sep 17 00:00:00 2001
From: "xiaolei.zl" <xiaolei.zl@alibaba-inc.com>
Date: Fri, 20 Dec 2024 10:22:36 +0800
Subject: [PATCH] fix bug

Committed-by: xiaolei.zl from Dev container
---
 .../runtime/adhoc/operators/group_by.cc       | 31 +++++++++++++++++++
 .../graph_db/runtime/common/operators/scan.cc |  3 +-
 flex/engines/graph_db/runtime/common/rt_any.h |  1 +
 3 files changed, 33 insertions(+), 2 deletions(-)

diff --git a/flex/engines/graph_db/runtime/adhoc/operators/group_by.cc b/flex/engines/graph_db/runtime/adhoc/operators/group_by.cc
index 14f070759c1e..20cff565f700 100644
--- a/flex/engines/graph_db/runtime/adhoc/operators/group_by.cc
+++ b/flex/engines/graph_db/runtime/adhoc/operators/group_by.cc
@@ -413,6 +413,29 @@ std::shared_ptr<IContextColumn> string_to_list(
   return builder.finish();
 }
 
+template <typename T>
+std::shared_ptr<IContextColumn> scalar_to_list(
+    const Var& var, const std::vector<std::vector<size_t>>& to_aggregate) {
+  ListValueColumnBuilder<T> builder;
+  size_t col_size = to_aggregate.size();
+  builder.reserve(col_size);
+  std::vector<std::shared_ptr<ListImplBase>> impls;
+  for (size_t k = 0; k < col_size; ++k) {
+    auto& vec = to_aggregate[k];
+
+    std::vector<T> elem;
+    for (auto idx : vec) {
+      elem.push_back(TypedConverter<T>::to_typed(var.get(idx)));
+    }
+    auto impl = ListImpl<T>::make_list_impl(std::move(elem));
+    auto list = List::make_list(impl);
+    impls.emplace_back(impl);
+    builder.push_back_opt(list);
+  }
+  builder.set_list_impls(impls);
+  return builder.finish();
+}
+
 bl::result<std::shared_ptr<IContextColumn>> apply_reduce(
     const AggFunc& func, const std::vector<std::vector<size_t>>& to_aggregate) {
   if (func.aggregate == AggrKind::kSum) {
@@ -497,6 +520,14 @@ bl::result<std::shared_ptr<IContextColumn>> apply_reduce(
       return tuple_to_list(var, to_aggregate);
     } else if (var.type() == RTAnyType::kStringValue) {
       return string_to_list(var, to_aggregate);
+    } else if (var.type() == RTAnyType::kI32Value) {
+      return scalar_to_list<int32_t>(var, to_aggregate);
+    } else if (var.type() == RTAnyType::kI64Value) {
+      return scalar_to_list<int64_t>(var, to_aggregate);
+    } else if (var.type() == RTAnyType::kU64Value) {
+      return scalar_to_list<uint64_t>(var, to_aggregate);
+    } else if (var.type() == RTAnyType::kF64Value) {
+      return scalar_to_list<double>(var, to_aggregate);
     } else {
       LOG(FATAL) << "not support" << static_cast<int>(var.type().type_enum_);
     }
diff --git a/flex/engines/graph_db/runtime/common/operators/scan.cc b/flex/engines/graph_db/runtime/common/operators/scan.cc
index 9b260008cd71..d61e3fd776ae 100644
--- a/flex/engines/graph_db/runtime/common/operators/scan.cc
+++ b/flex/engines/graph_db/runtime/common/operators/scan.cc
@@ -47,11 +47,10 @@ bl::result<Context> Scan::find_vertex_with_id(const ReadTransaction& txn,
     }
     if (GlobalId::get_label_id(gid) == label) {
       vid = GlobalId::get_vid(gid);
+      builder.push_back_opt(vid);
     } else {
       LOG(ERROR) << "Global id " << gid << " does not match label " << label;
-      return Context();
     }
-    builder.push_back_opt(vid);
     Context ctx;
     ctx.set(alias, builder.finish());
     return ctx;
diff --git a/flex/engines/graph_db/runtime/common/rt_any.h b/flex/engines/graph_db/runtime/common/rt_any.h
index cc5371e4005f..3b73f06e3998 100644
--- a/flex/engines/graph_db/runtime/common/rt_any.h
+++ b/flex/engines/graph_db/runtime/common/rt_any.h
@@ -420,6 +420,7 @@ class ListImpl : ListImplBase {
   static std::shared_ptr<ListImplBase> make_list_impl(std::vector<T>&& vals) {
     auto new_list = new ListImpl<T>();
     new_list->list_ = std::move(vals);
+    new_list->is_valid_.resize(new_list->list_.size(), true);
     return std::shared_ptr<ListImplBase>(static_cast<ListImplBase*>(new_list));
   }