Repository: alibaba/GraphScope (language: python)

Issue:
feat(interactive): Support count multiple distinct column.
For the following query, compiler throw error`java.lang.UnsupportedOperationException: aggregate on multiple variables is unsupported yet`

```cypher
MATCH (p:Paper)-[:Resolve]->(ch:Challenge),
(p1:Paper)-[:Resolve]->(ch),
(p)-[c:Citation]-(p1)
RETURN COUNT(distinct p, p1)
```


Discussion / hints:

Fix patch:
diff --git a/flex/engines/hqps_db/core/operator/group_by.h b/flex/engines/hqps_db/core/operator/group_by.h
index 40e391538abe..bf3315bbec2f 100644
--- a/flex/engines/hqps_db/core/operator/group_by.h
+++ b/flex/engines/hqps_db/core/operator/group_by.h
@@ -62,22 +62,40 @@ struct GroupKeyResT<CTX_T, GroupKey<col_id, T>> {
   using result_t = typename KeyedT<set_t, PropertySelector<T>>::keyed_set_t;
 };
 
-template <typename CTX_T, typename AGG_T>
+template <typename CTX_T, typename AGG_T, typename Enable = void>
 struct GroupValueResT;
 
+// The SET_T could b a single set or a tuple of sets.
 template <typename SET_T, AggFunc agg_func, typename SELECTOR_TUPLE>
 struct GroupValueResTImpl;
 
-template <typename CTX_T, AggFunc agg_func, typename... SELECTOR, int Is>
-struct GroupValueResT<CTX_T, AggregateProp<agg_func, std::tuple<SELECTOR...>,
-                                           std::integer_sequence<int, Is>>> {
+// Specialize for single set
+template <typename CTX_T, AggFunc agg_func, typename... SELECTOR, int... Is>
+struct GroupValueResT<CTX_T,
+                      AggregateProp<agg_func, std::tuple<SELECTOR...>,
+                                    std::integer_sequence<int, Is...>>,
+                      typename std::enable_if<(sizeof...(Is) == 1)>::type> {
   using old_set_t = std::remove_const_t<std::remove_reference_t<decltype(
-      std::declval<CTX_T>().template GetNode<Is>())>>;
+      std::declval<CTX_T>().template GetNode<FirstElement<Is...>::value>())>>;
   using result_t =
       typename GroupValueResTImpl<old_set_t, agg_func,
                                   std::tuple<SELECTOR...>>::result_t;
 };
 
+// Specialize for multiple sets
+template <typename CTX_T, AggFunc agg_func, typename... SELECTOR, int... Is>
+struct GroupValueResT<CTX_T,
+                      AggregateProp<agg_func, std::tuple<SELECTOR...>,
+                                    std::integer_sequence<int, Is...>>,
+                      typename std::enable_if<(sizeof...(Is) > 1)>::type> {
+  using old_set_tuple_t =
+      std::tuple<std::remove_const_t<std::remove_reference_t<decltype(
+          std::declval<CTX_T>().template GetNode<Is>())>>...>;
+  using result_t =
+      typename GroupValueResTImpl<old_set_tuple_t, agg_func,
+                                  std::tuple<SELECTOR...>>::result_t;
+};
+
 // specialization for count for single tag
 // TODO: count for pairs.
 template <typename SET_T>
@@ -92,6 +110,18 @@ struct GroupValueResTImpl<SET_T, AggFunc::COUNT_DISTINCT,
   using result_t = Collection<size_t>;
 };
 
+// PropSelectorTuple doesn't effect the result type.
+template <typename SET_TUPLE_T, typename PropSelectorTuple>
+struct GroupValueResTImpl<SET_TUPLE_T, AggFunc::COUNT_DISTINCT,
+                          PropSelectorTuple> {
+  using result_t = Collection<size_t>;
+};
+
+template <typename SET_TUPLE_T, typename PropSelectorTuple>
+struct GroupValueResTImpl<SET_TUPLE_T, AggFunc::COUNT, PropSelectorTuple> {
+  using result_t = Collection<size_t>;
+};
+
 template <typename T>
 struct GroupValueResTImpl<Collection<T>, AggFunc::SUM,
                           std::tuple<PropertySelector<grape::EmptyType>>> {
@@ -571,7 +601,7 @@ class GroupByOp {
 
   template <typename... SET_T, typename HEAD_T, AggFunc _agg_func, typename T,
             int tag_id>
-  static auto create_keyed_value_set_builder(
+  static auto create_keyed_value_set_builder_single_tag(
       const GRAPH_INTERFACE& graph, const std::tuple<SET_T...>& tuple,
       const HEAD_T& head,
       AggregateProp<_agg_func, std::tuple<PropertySelector<T>>,
@@ -591,6 +621,43 @@ class GroupByOp {
     }
   }
 
+  // For aggregate on multiple tags, we currently only support count distinct
+  // and count.
+  template <typename... SET_T, typename HEAD_T, AggFunc _agg_func,
+            typename PROP_TUPLE_T, int... tag_id>
+  static auto create_keyed_value_set_builder_multi_tag(
+      const GRAPH_INTERFACE& graph, const std::tuple<SET_T...>& tuple,
+      const HEAD_T& head,
+      AggregateProp<_agg_func, PROP_TUPLE_T,
+                    std::integer_sequence<int32_t, tag_id...>>& agg) {
+    // create const ref tuple from tuple and head using std::cref
+    // construct a const ref tuple from tuple
+    auto const_ref_tuple = make_tuple_of_const_refs(tuple);
+
+    auto old_set = std::tuple_cat(const_ref_tuple, std::tie(head));
+    // get the tuple from old_set, with tag_ids
+    auto old_set_tuple = std::tuple{gs::get_from_tuple<tag_id>(old_set)...};
+
+    return KeyedAggMultiColT<GRAPH_INTERFACE, decltype(old_set_tuple),
+                             _agg_func, PROP_TUPLE_T,
+                             std::integer_sequence<int32_t, tag_id...>>::
+        create_agg_builder(old_set_tuple, graph, agg.selectors_);
+  }
+
+  template <typename... SET_T, typename HEAD_T, AggFunc _agg_func,
+            typename PROP_SELECTOR_TUPLE, int... tag_ids>
+  static auto create_keyed_value_set_builder(
+      const GRAPH_INTERFACE& graph, const std::tuple<SET_T...>& tuple,
+      const HEAD_T& head,
+      AggregateProp<_agg_func, PROP_SELECTOR_TUPLE,
+                    std::integer_sequence<int32_t, tag_ids...>>& agg) {
+    if constexpr (sizeof...(tag_ids) == 1) {
+      return create_keyed_value_set_builder_single_tag(graph, tuple, head, agg);
+    } else {
+      return create_keyed_value_set_builder_multi_tag(graph, tuple, head, agg);
+    }
+  }
+
   // create builder for single key_alias
   template <typename... SET_T, typename HEAD_T, int col_id, typename KEY_PROP>
   static auto create_unkeyed_set_builder(
diff --git a/flex/engines/hqps_db/core/utils/hqps_utils.h b/flex/engines/hqps_db/core/utils/hqps_utils.h
index dd70b115afb5..0c2b6773a34a 100644
--- a/flex/engines/hqps_db/core/utils/hqps_utils.h
+++ b/flex/engines/hqps_db/core/utils/hqps_utils.h
@@ -378,6 +378,27 @@ inline auto make_offset_vector(size_t m, size_t n) {
   return offsets;
 }
 
+template <int I, int... Is>
+struct FirstElement {
+  static constexpr int value = I;
+};
+
+// Create a tuple of const references to the elements of a tuple.
+template <typename... Args>
+auto make_tuple_of_const_refs(const std::tuple<Args...>& t) {
+  return std::apply(
+      [](const Args&... args) { return std::make_tuple(std::cref(args)...); },
+      t);
+}
+
+template <typename T>
+struct ConstRefRemoveHelper;
+
+template <typename... T>
+struct ConstRefRemoveHelper<std::tuple<T...>> {
+  using type = std::tuple<std::remove_const_t<std::remove_reference_t<T>>...>;
+};
+
 // first n ele in tuple type
 
 template <int n, typename In, typename... Out>
@@ -429,6 +450,16 @@ constexpr auto tuple_slice(T&& t) {
                              std::make_index_sequence<r - l>{});
 }
 
+// [l, tuple_size - 1]
+template <size_t l, typename T>
+constexpr auto tuple_slice(T&& t) {
+  static_assert(std::tuple_size<std::decay_t<T>>::value > l,
+                "slice index out of bounds");
+  return tuple_slice_impl<l>(
+      std::forward<T>(t),
+      std::make_index_sequence<std::tuple_size<std::decay_t<T>>::value - l>{});
+}
+
 template <int Is, typename... T,
           typename std::enable_if<(Is >= 0)>::type* = nullptr>
 inline auto get_from_tuple(std::tuple<T...>& tuple) {
diff --git a/flex/engines/hqps_db/core/utils/keyed.h b/flex/engines/hqps_db/core/utils/keyed.h
index a386a991a33c..af22745eaab9 100644
--- a/flex/engines/hqps_db/core/utils/keyed.h
+++ b/flex/engines/hqps_db/core/utils/keyed.h
@@ -149,13 +149,21 @@ struct KeyedT<Collection<T>, PropertySelector<grape::EmptyType>> {
 // when keyed with aggregation function, (which we currently only support
 // collection)
 
-/// @brief Helper to get keyed set type with aggregation fnc
+/// @brief Helper to get keyed set type with aggregation func
 /// @tparam T
 /// @tparam ValueT Keyed prop type
 template <typename GI, typename T, AggFunc agg_func, typename Props,
           typename Tags>
 struct KeyedAggT;
 
+/// @brief Helper to get keyed set type with aggregation func, which is applied
+/// on multiple column
+/// @tparam T
+/// @tparam ValueT Keyed prop type
+template <typename GI, typename SET_TUPLE_T, AggFunc agg_func, typename Props,
+          typename Tags>
+struct KeyedAggMultiColT;
+
 template <typename GI, typename LabelT, typename VID_T, typename... T,
           typename PropT, int tag_id>
 struct KeyedAggT<GI, RowVertexSet<LabelT, VID_T, T...>, AggFunc::COUNT,
@@ -465,6 +473,39 @@ struct KeyedAggT<GI, FlatEdgeSet<VID_T, LabelT, EDATA_T>,
   }
 };
 
+template <typename GI, typename... SET_T, typename... PropSelectorT,
+          int... tag_ids>
+struct KeyedAggMultiColT<GI, std::tuple<SET_T...>, AggFunc::COUNT_DISTINCT,
+                         std::tuple<PropSelectorT...>,
+                         std::integer_sequence<int32_t, tag_ids...>> {
+  using agg_res_t = Collection<size_t>;
+  // get the tuple of sets from the tuple of tags.
+  using aggregate_res_builder_t =
+      MultiColDistinctCountBuilder<std::tuple<SET_T...>, tag_ids...>;
+
+  static aggregate_res_builder_t create_agg_builder(
+      const std::tuple<SET_T...>& set, const GI& graph,
+      std::tuple<PropSelectorT...>& selectors) {
+    return aggregate_res_builder_t();
+  }
+};
+
+template <typename GI, typename... SET_T, typename... PropSelectorT,
+          int... tag_ids>
+struct KeyedAggMultiColT<GI, std::tuple<SET_T...>, AggFunc::COUNT,
+                         std::tuple<PropSelectorT...>,
+                         std::integer_sequence<int32_t, tag_ids...>> {
+  using agg_res_t = Collection<size_t>;
+  // get the tuple of sets from the tuple of tags.
+  using aggregate_res_builder_t = MultiColCountBuilder<tag_ids...>;
+
+  static aggregate_res_builder_t create_agg_builder(
+      const std::tuple<SET_T...>& set, const GI& graph,
+      std::tuple<PropSelectorT...>& selectors) {
+    return aggregate_res_builder_t();
+  }
+};
+
 template <typename LabelT, typename KEY_T, typename VID_T, typename... T,
           typename ELE, typename DATA>
 static inline auto insert_into_builder_v2_impl(
diff --git a/flex/engines/hqps_db/structures/collection.h b/flex/engines/hqps_db/structures/collection.h
index f2f307a12d65..79b37c34400e 100644
--- a/flex/engines/hqps_db/structures/collection.h
+++ b/flex/engines/hqps_db/structures/collection.h
@@ -16,6 +16,7 @@
 #define ENGINES_HQPS_DS_COLLECTION_H_
 
 #include <tuple>
+#include <unordered_set>
 #include <vector>
 
 #include "flex/engines/hqps_db/core/null_record.h"
@@ -427,6 +428,39 @@ class CountBuilder {
   std::vector<size_t> vec_;
 };
 
+template <int... tag>
+class MultiColCountBuilder {
+ public:
+  MultiColCountBuilder() {}
+
+  // insert tuple at index ind.
+  // if the ele_value equal to invalid_value, then do not insert.
+  template <typename ELE_TUPLE, typename DATA_TUPLE>
+  void insert(size_t ind, const ELE_TUPLE& tuple, const DATA_TUPLE& data) {
+    auto cur_ele_tuple =
+        std::tuple_cat(tuple_slice<1>(gs::get_from_tuple<tag>(tuple))...);
+    while (vec_.size() <= ind) {
+      vec_.emplace_back(0);
+    }
+    using cur_ele_tuple_t =
+        std::remove_const_t<std::remove_reference_t<decltype(cur_ele_tuple)>>;
+    // remove the const and reference for each type in cur_ele_tuple_t
+    using cur_ele_tuple_rm_const_ref_t =
+        typename ConstRefRemoveHelper<cur_ele_tuple_t>::type;
+    if (cur_ele_tuple !=
+        NullRecordCreator<cur_ele_tuple_rm_const_ref_t>::GetNull()) {
+      ++vec_[ind];
+    } else {
+      VLOG(10) << "ele is null";
+    }
+  }
+
+  Collection<size_t> Build() { return Collection<size_t>(std::move(vec_)); }
+
+ private:
+  std::vector<size_t> vec_;
+};
+
 template <int tag_id, typename T, typename Enable = void>
 class DistinctCountBuilder;
 
@@ -581,6 +615,48 @@ class DistinctCountBuilder<tag_id, TwoLabelVertexSetImpl<VID_T, LabelT, T...>> {
   std::array<VID_T, 2> min_v, max_v, range_size;
 };
 
+// DistinctCountBuilder for multiple sets together
+template <typename SET_TUPLE_T, int... TAG_IDs>
+class MultiColDistinctCountBuilder;
+
+template <typename... SET_Ts, int... TAG_IDs>
+class MultiColDistinctCountBuilder<std::tuple<SET_Ts...>, TAG_IDs...> {
+ public:
+  using set_ele_t = std::tuple<typename SET_Ts::element_t...>;
+  MultiColDistinctCountBuilder() {}
+
+  template <typename ELE_TUPLE_T, typename DATA_TUPLE>
+  void insert(size_t ind, const ELE_TUPLE_T& tuple, const DATA_TUPLE& data) {
+    // construct the ref tuple from tuple,with TAG_IDS,
+    // get element tuple from index_ele_tuple_t
+    auto cur_ele_tuple =
+        std::tuple_cat(tuple_slice<1>(gs::get_from_tuple<TAG_IDs>(tuple))...);
+
+    while (vec_of_set_.size() <= ind) {
+      vec_of_set_.emplace_back(
+          std::unordered_set<set_ele_t, boost::hash<set_ele_t>>());
+    }
+    auto& cur_set = vec_of_set_[ind];
+    cur_set.insert(cur_ele_tuple);
+    VLOG(10) << "tuple: " << gs::to_string(cur_ele_tuple)
+             << ",all ele: " << gs::to_string(tuple) << "insert at ind: " << ind
+             << ", res: " << cur_set.size();
+  }
+
+  Collection<size_t> Build() {
+    std::vector<size_t> res;
+    res.reserve(vec_of_set_.size());
+    for (auto& set : vec_of_set_) {
+      res.emplace_back(set.size());
+    }
+    return Collection<size_t>(std::move(res));
+  }
+
+ private:
+  std::vector<std::unordered_set<set_ele_t, boost::hash<set_ele_t>>>
+      vec_of_set_;
+};
+
 template <typename T, int tag_id>
 class SumBuilder {
  public:
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java
index c7c229d72960..54305f7a8a3c 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/ffi/RelToFfiConverter.java
@@ -39,6 +39,7 @@
 import com.alibaba.graphscope.common.ir.type.GraphSchemaType;
 import com.alibaba.graphscope.common.jna.IrCoreLibrary;
 import com.alibaba.graphscope.common.jna.type.*;
+import com.alibaba.graphscope.gaia.proto.Common;
 import com.alibaba.graphscope.gaia.proto.GraphAlgebra;
 import com.alibaba.graphscope.gaia.proto.OuterExpression;
 import com.google.common.base.Preconditions;
@@ -365,32 +366,41 @@ public PhysicalNode visit(GraphLogicalAggregate aggregate) {
             if (operands.isEmpty()) {
                 throw new IllegalArgumentException(
                         "operands in aggregate call should not be empty");
-            } else if (operands.size() > 1) {
-                throw new UnsupportedOperationException(
-                        "aggregate on multiple variables is unsupported yet");
             }
-            FfiAggOpt ffiAggOpt = Utils.ffiAggOpt(groupCalls.get(i));
             int aliasId = fields.get(i + groupKeys.size()).getIndex();
-            FfiAlias.ByValue ffiAlias =
+            Common.NameOrId alias =
                     (aliasId == AliasInference.DEFAULT_ID)
-                            ? ArgUtils.asNoneAlias()
-                            : ArgUtils.asAlias(aliasId);
-            Preconditions.checkArgument(
-                    operands.get(0) instanceof RexGraphVariable,
-                    "each expression in aggregate call should be type %s, but is %s",
-                    RexGraphVariable.class,
-                    operands.get(0).getClass());
-            OuterExpression.Variable var =
-                    operands.get(0)
-                            .accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
-                            .getOperators(0)
-                            .getVar();
+                            ? Common.NameOrId.newBuilder().build()
+                            : Common.NameOrId.newBuilder().setId(aliasId).build();
+            List<OuterExpression.Variable> vars =
+                    operands.stream()
+                            .map(
+                                    k -> {
+                                        Preconditions.checkArgument(
+                                                k instanceof RexGraphVariable,
+                                                "each operand in aggregate call should be type %s,"
+                                                        + " but is %s",
+                                                RexGraphVariable.class,
+                                                k.getClass());
+                                        return k.accept(
+                                                        new RexToProtoConverter(
+                                                                true, isColumnId, this.rexBuilder))
+                                                .getOperators(0)
+                                                .getVar();
+                                    })
+                            .collect(Collectors.toList());
             checkFfiResult(
                     LIB.addGroupbyAggFnPb(
                             ptrGroup,
-                            new FfiPbPointer.ByValue(var.toByteArray()),
-                            ffiAggOpt,
-                            ffiAlias));
+                            new FfiPbPointer.ByValue(
+                                    GraphAlgebra.GroupBy.AggFunc.newBuilder()
+                                            .addAllVars(vars)
+                                            .setAggregate(
+                                                    com.alibaba.graphscope.common.ir.runtime.proto
+                                                            .Utils.protoAggFn(groupCalls.get(i)))
+                                            .setAlias(alias)
+                                            .build()
+                                            .toByteArray())));
         }
         com.alibaba.graphscope.common.ir.runtime.proto.Utils.protoRowType(
                         aggregate.getRowType(), isColumnId)
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/Utils.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/Utils.java
index c816b461302b..dfd45fd92de3 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/Utils.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/Utils.java
@@ -16,6 +16,9 @@
 
 package com.alibaba.graphscope.common.ir.runtime.proto;
 
+import static com.alibaba.graphscope.gaia.proto.GraphAlgebra.GroupBy.AggFunc.Aggregate;
+
+import com.alibaba.graphscope.common.ir.rel.type.group.GraphAggCall;
 import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
 import com.alibaba.graphscope.common.ir.type.GraphLabelType;
 import com.alibaba.graphscope.common.ir.type.GraphNameOrId;
@@ -405,4 +408,29 @@ public static final OuterExpression.Extract.Interval protoInterval(RexLiteral li
                 throw new UnsupportedOperationException("unsupported interval type " + timeUnit);
         }
     }
+
+    public static final Aggregate protoAggFn(GraphAggCall aggCall) {
+        switch (aggCall.getAggFunction().getKind()) {
+            case COUNT:
+                return aggCall.isDistinct() ? Aggregate.COUNT_DISTINCT : Aggregate.COUNT;
+            case COLLECT:
+                return aggCall.isDistinct() ? Aggregate.TO_SET : Aggregate.TO_LIST;
+            case SUM:
+            case SUM0:
+                return Aggregate.SUM;
+            case AVG:
+                return Aggregate.AVG;
+            case MIN:
+                return Aggregate.MIN;
+            case MAX:
+                return Aggregate.MAX;
+            case FIRST_VALUE:
+                return Aggregate.FIRST;
+            default:
+                throw new UnsupportedOperationException(
+                        "aggregate opt "
+                                + aggCall.getAggFunction().getKind()
+                                + " is unsupported yet");
+        }
+    }
 }
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/jna/IrCoreLibrary.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/jna/IrCoreLibrary.java
index d1aa912dc0a2..0a3afbd6e811 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/jna/IrCoreLibrary.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/jna/IrCoreLibrary.java
@@ -127,11 +127,7 @@ FfiResult.ByValue addGroupbyAggFn(
     FfiResult.ByValue addGroupbyKeyPbAlias(
             Pointer groupBy, FfiPbPointer.ByValue pbPointer, FfiAlias.ByValue alias);
 
-    FfiResult.ByValue addGroupbyAggFnPb(
-            Pointer group,
-            FfiPbPointer.ByValue pbPointer,
-            FfiAggOpt aggOpt,
-            FfiAlias.ByValue alias);
+    FfiResult.ByValue addGroupbyAggFnPb(Pointer group, FfiPbPointer.ByValue pbPointer);
 
     FfiResult.ByValue appendGroupbyOperator(
             Pointer plan, Pointer groupBy, int parent, IntByReference oprIdx);
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/movie/MovieQueries.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/movie/MovieQueries.java
index bd35cb4b655b..2cf5403843b0 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/movie/MovieQueries.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/cypher/integration/suite/movie/MovieQueries.java
@@ -333,4 +333,31 @@ public static QueryContext get_movie_query20_test() {
                 Arrays.asList("Record<{value: {tagline: \"Welcome to the Real World\", id: 0}}>");
         return new QueryContext(query, expected);
     }
+
+    public static QueryContext get_movie_query21_test() {
+        String query =
+                "Match (tom:Person {name: 'Tom Hanks'})-[:ACTED_IN]->(movie1:Movie) Return"
+                        + " count(tom, movie1);";
+        List<String> expected = Arrays.asList("Record<{$f0: 12}>");
+        return new QueryContext(query, expected);
+    }
+
+    public static QueryContext get_movie_query22_test() {
+        String query =
+                "Match (tom:Person {name: 'Tom"
+                        + " Hanks'})-[:ACTED_IN]->(movie1:Movie)<-[:ACTED_IN]-(p2:Person) Return"
+                        + " count(distinct tom, p2);";
+        List<String> expected = Arrays.asList("Record<{$f0: 35}>");
+        return new QueryContext(query, expected);
+    }
+
+    public static QueryContext get_movie_query23_test() {
+        String query =
+                "Match (tom:Person {name: 'Tom"
+                    + " Hanks'})-[:ACTED_IN]->(movie1:Movie)<-[:ACTED_IN]-(p2:Person {name: \"Tom"
+                    + " Hanks\"}) Return distinct tom.name, p2.name;";
+        List<String> expected =
+                Arrays.asList("Record<{name: \"Tom Hanks\", name0: \"Tom Hanks\"}>");
+        return new QueryContext(query, expected);
+    }
 }
diff --git a/interactive_engine/executor/ir/core/src/plan/ffi.rs b/interactive_engine/executor/ir/core/src/plan/ffi.rs
index 20f6c598507c..8d3d10bd36eb 100644
--- a/interactive_engine/executor/ir/core/src/plan/ffi.rs
+++ b/interactive_engine/executor/ir/core/src/plan/ffi.rs
@@ -1495,24 +1495,14 @@ mod groupby {
     /// Add the aggregate function for each group.
     /// The aggregation function is represented as a pb pointer.
     #[no_mangle]
-    pub extern "C" fn add_groupby_agg_fn_pb(
-        ptr_groupby: *const c_void, agg_val: FfiPbPointer, agg_opt: FfiAggOpt, alias: FfiAlias,
-    ) -> FfiResult {
+    pub extern "C" fn add_groupby_agg_fn_pb(ptr_groupby: *const c_void, agg_fn: FfiPbPointer) -> FfiResult {
         let mut result = FfiResult::success();
         let mut group = unsafe { Box::from_raw(ptr_groupby as *mut pb::GroupBy) };
-        let val_pb = ptr_to_pb::<common_pb::Variable>(agg_val);
-        let aggregate = unsafe { std::mem::transmute::<FfiAggOpt, i32>(agg_opt) };
-        let alias_pb = alias.try_into();
-        if val_pb.is_ok() && alias_pb.is_ok() {
-            group.functions.push(pb::group_by::AggFunc {
-                vars: vec![val_pb.unwrap()],
-                aggregate,
-                alias: alias_pb.unwrap(),
-            });
-        } else if val_pb.is_err() {
-            result = val_pb.err().unwrap();
+        let agg_fn_pb = ptr_to_pb::<pb::group_by::AggFunc>(agg_fn);
+        if agg_fn_pb.is_ok() {
+            group.functions.push(agg_fn_pb.unwrap());
         } else {
-            result = alias_pb.err().unwrap();
+            result = agg_fn_pb.err().unwrap();
         }
         std::mem::forget(group);
 


Test patch:
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java
index 9be7065deffa..3e4a0826fbbf 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/common/ir/runtime/FfiLogicalPlanTest.java
@@ -284,6 +284,22 @@ public void logical_plan_8_test() throws Exception {
         }
     }
 
+    @Test
+    public void logical_plan_9_test() throws Exception {
+        RelNode node =
+                com.alibaba.graphscope.cypher.antlr4.Utils.eval(
+                                "Match (a) Return count(distinct a.name, a.age)")
+                        .build();
+        try (PhysicalBuilder ffiBuilder =
+                new FfiPhysicalBuilder(
+                        getMockGraphConfig(), Utils.schemaMeta, new LogicalPlan(node))) {
+            PhysicalPlan physicalPlan = ffiBuilder.build();
+            Assert.assertEquals(
+                    FileUtils.readJsonFromResource("ffi_logical_plan_9.json"),
+                    physicalPlan.explain());
+        }
+    }
+
     private Configs getMockGraphConfig() {
         return new Configs(ImmutableMap.of("servers", "1", "workers", "1"));
     }
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/movie/MovieTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/movie/MovieTest.java
index 712d9dd148b7..9f93565034c8 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/movie/MovieTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/cypher/integration/movie/MovieTest.java
@@ -160,9 +160,29 @@ public void run_movie_query19_test() {
     public void run_movie_query20_test() {
         assumeTrue("pegasus".equals(System.getenv("ENGINE_TYPE")));
         QueryContext testQuery = MovieQueries.get_movie_query20_test();
-        Result result =
-                session.run(
-                        "Match (n:Movie {id: 0}) Return {id: n.id, tagline: n.tagline} as value;");
+        Result result = session.run(testQuery.getQuery());
+        Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
+    }
+
+    @Test
+    public void run_movie_query21_test() {
+        QueryContext testQuery = MovieQueries.get_movie_query21_test();
+        Result result = session.run(testQuery.getQuery());
+        Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
+    }
+
+    @Test
+    public void run_movie_query22_test() {
+        assumeTrue("hiactor".equals(System.getenv("ENGINE_TYPE")));
+        QueryContext testQuery = MovieQueries.get_movie_query22_test();
+        Result result = session.run(testQuery.getQuery());
+        Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
+    }
+
+    @Test
+    public void run_movie_query23_test() {
+        QueryContext testQuery = MovieQueries.get_movie_query23_test();
+        Result result = session.run(testQuery.getQuery());
         Assert.assertEquals(testQuery.getExpectedResult().toString(), result.list().toString());
     }
 
diff --git a/interactive_engine/compiler/src/test/resources/ffi_logical_plan_9.json b/interactive_engine/compiler/src/test/resources/ffi_logical_plan_9.json
new file mode 100644
index 000000000000..9f7927954fb4
--- /dev/null
+++ b/interactive_engine/compiler/src/test/resources/ffi_logical_plan_9.json
@@ -0,0 +1,225 @@
+{
+  "nodes": [
+    {
+      "opr": {
+        "opr": {
+          "Scan": {
+            "scan_opt": 0,
+            "alias": {
+              "item": {
+                "Id": 0
+              }
+            },
+            "params": {
+              "tables": [
+                {
+                  "item": {
+                    "Id": 0
+                  }
+                },
+                {
+                  "item": {
+                    "Id": 1
+                  }
+                }
+              ],
+              "columns": [],
+              "is_all_columns": false,
+              "limit": null,
+              "predicate": null,
+              "sample_ratio": 1.0,
+              "extra": {}
+            },
+            "idx_predicate": null,
+            "is_count_only": false,
+            "meta_data": {
+              "type": {
+                "type": {
+                  "GraphType": {
+                    "element_opt": 0,
+                    "graph_data_type": [
+                      {
+                        "label": {
+                          "label": 1,
+                          "src_label": null,
+                          "dst_label": null
+                        },
+                        "props": [
+                          {
+                            "prop_id": {
+                              "item": {
+                                "Name": "id"
+                              }
+                            },
+                            "type": 2
+                          },
+                          {
+                            "prop_id": {
+                              "item": {
+                                "Name": "name"
+                              }
+                            },
+                            "type": 4
+                          },
+                          {
+                            "prop_id": {
+                              "item": {
+                                "Name": "lang"
+                              }
+                            },
+                            "type": 4
+                          },
+                          {
+                            "prop_id": {
+                              "item": {
+                                "Name": "creationDate"
+                              }
+                            },
+                            "type": 12
+                          }
+                        ]
+                      },
+                      {
+                        "label": {
+                          "label": 0,
+                          "src_label": null,
+                          "dst_label": null
+                        },
+                        "props": [
+                          {
+                            "prop_id": {
+                              "item": {
+                                "Name": "id"
+                              }
+                            },
+                            "type": 2
+                          },
+                          {
+                            "prop_id": {
+                              "item": {
+                                "Name": "name"
+                              }
+                            },
+                            "type": 4
+                          },
+                          {
+                            "prop_id": {
+                              "item": {
+                                "Name": "age"
+                              }
+                            },
+                            "type": 1
+                          }
+                        ]
+                      }
+                    ]
+                  }
+                }
+              },
+              "alias": 0
+            }
+          }
+        }
+      },
+      "children": [
+        1
+      ]
+    },
+    {
+      "opr": {
+        "opr": {
+          "GroupBy": {
+            "mappings": [],
+            "functions": [
+              {
+                "vars": [
+                  {
+                    "tag": {
+                      "item": {
+                        "Id": 0
+                      }
+                    },
+                    "property": {
+                      "item": {
+                        "Key": {
+                          "item": {
+                            "Name": "name"
+                          }
+                        }
+                      }
+                    },
+                    "node_type": {
+                      "type": {
+                        "DataType": 4
+                      }
+                    }
+                  },
+                  {
+                    "tag": {
+                      "item": {
+                        "Id": 0
+                      }
+                    },
+                    "property": {
+                      "item": {
+                        "Key": {
+                          "item": {
+                            "Name": "age"
+                          }
+                        }
+                      }
+                    },
+                    "node_type": {
+                      "type": {
+                        "DataType": 1
+                      }
+                    }
+                  }
+                ],
+                "aggregate": 4,
+                "alias": {
+                  "item": {
+                    "Id": 1
+                  }
+                }
+              }
+            ],
+            "meta_data": [
+              {
+                "type": {
+                  "type": {
+                    "DataType": 2
+                  }
+                },
+                "alias": 1
+              }
+            ]
+          }
+        }
+      },
+      "children": [
+        2
+      ]
+    },
+    {
+      "opr": {
+        "opr": {
+          "Sink": {
+            "tags": [],
+            "sink_target": {
+              "inner": {
+                "SinkDefault": {
+                  "id_name_mappings": []
+                }
+              }
+            }
+          }
+        }
+      },
+      "children": []
+    }
+  ],
+  "roots": [
+    0
+  ]
+}