RDDT: Group-by loop on proxy

go.sum

@@ -984,6 +984,7 @@ github.com/shoenig/go-m1cpu v0.2.2 h1:4nc55oVv7nygGnfI9bhLCLzUEs4794y0Bkqx4q2zy7
 github.com/shoenig/go-m1cpu v0.2.2/go.mod h1:KkDOw6m3ZJQAPHbrzkZki4hnx+pDRR1Lo+ldA56wD5w=
 github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU=
 github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k=
+github.com/shoenig/test v1.7.0 h1:eWcHtTXa6QLnBvm0jgEabMRN/uJ4DMV3M8xUGgRkZmk=
 github.com/shopspring/decimal v1.4.0/go.mod h1:gawqmDU56v4yIKSwfBSFip1HdCCXN8/+DMd9qYNcwME=
 github.com/shurcooL/go v0.0.0-20180423040247-9e1955d9fb6e/go.mod h1:TDJrrUr11Vxrven61rcy3hJMUqaf/CLWYhHNPmT14Lk=
 github.com/shurcooL/httpfs v0.0.0-20181222201310-74dc9339e414/go.mod h1:ZY1cvUeJuFPAdZ/B6v7RHavJWZn2YPVFQ1OSXhCGOkg=

internal/core/src/common/QueryInfo.h

@@ -43,6 +43,7 @@ struct SearchInfo {
     tracer::TraceContext trace_ctx_;
     bool materialized_view_involved = false;
     bool iterative_filter_execution = false;
+    bool proxy_group_by_refill_ = false;
     std::optional<SearchIteratorV2Info> iterator_v2_info_ = std::nullopt;
     std::optional<std::string> json_path_;
     std::optional<milvus::DataType> json_type_;

internal/core/src/exec/operator/GroupByNode.cpp

@@ -15,10 +15,14 @@
 // limitations under the License.
 
 #include "GroupByNode.h"
+#include <numeric>
+
 #include "common/Tracer.h"
 #include "fmt/format.h"
 
+#include "exec/operator/Utils.h"
 #include "exec/operator/groupby/SearchGroupByOperator.h"
+#include "log/Log.h"
 #include "monitor/Monitor.h"
 namespace milvus {
 namespace exec {
@@ -66,7 +70,8 @@ PhyGroupByNode::GetOutput() {
 
     auto op_context = query_context_->get_op_context();
     auto search_result = query_context_->get_search_result();
-    if (search_result.vector_iterators_.has_value()) {
+    if (UseVectorIterator(search_info_) &&
+        search_result.vector_iterators_.has_value()) {
         AssertInfo(search_result.vector_iterators_.value().size() ==
                        search_result.total_nq_,
                    "Vector Iterators' count must be equal to total_nq_, Check "
@@ -88,6 +93,24 @@ PhyGroupByNode::GetOutput() {
                    "equal to search_result.seg_offsets.size:{}",
                    search_result.group_by_values_.value().size(),
                    search_result.seg_offsets_.size());
+    } else {
+        std::vector<GroupByValueType> group_by_values;
+        milvus::exec::PopulateGroupByValues(op_context,
+                                            search_info_,
+                                            group_by_values,
+                                            *segment_,
+                                            search_result.seg_offsets_);
+        search_result.group_by_values_ = std::move(group_by_values);
+        search_result.group_size_ = search_info_.group_size_;
+        if (search_result.topk_per_nq_prefix_sum_.empty()) {
+            std::vector<size_t> topks(search_result.total_nq_,
+                                      search_result.unity_topK_);
+            search_result.topk_per_nq_prefix_sum_.resize(
+                search_result.total_nq_ + 1);
+            std::partial_sum(topks.begin(),
+                             topks.end(),
+                             search_result.topk_per_nq_prefix_sum_.begin() + 1);
+        }
     }
     tracer::AddEvent(
         fmt::format("grouped_results: {}", search_result.seg_offsets_.size()));
@@ -109,4 +132,4 @@ PhyGroupByNode::IsFinished() {
 }
 
 }  // namespace exec
-}  // namespace milvus
+}  // namespace milvus

internal/core/src/exec/operator/Utils.h

@@ -33,8 +33,16 @@ namespace exec {
 
 static bool
 UseVectorIterator(const SearchInfo& search_info) {
-    return search_info.group_by_field_id_.has_value() ||
-           search_info.iterative_filter_execution;
+    if (search_info.iterative_filter_execution) {
+        return true;
+    }
+    // Default group-by uses vector iterators at core. Proxy-side refill disables
+    // this path and handles incomplete groups with follow-up searches instead.
+    if (search_info.group_by_field_id_.has_value() &&
+        !search_info.proxy_group_by_refill_) {
+        return true;
+    }
+    return false;
 }
 
 static bool