fix: deduplicate element-level search results by pk and element offset (#49411)

issue: https://github.com/milvus-io/milvus/issues/49357
ref: https://github.com/milvus-io/milvus/issues/42148

- Fix element-level search reduce to deduplicate by `(pk,
element_index)` instead of `pk` only.
- Apply the same semantics in C++ reduce, group-by reduce, and Go search
result reduce paths.
- Reject legacy search iterator for element-level ArrayOfVector search;
require search iterator v2.

---------

Signed-off-by: SpadeA <tangchenjie1210@gmail.com>

internal/core/src/segcore/reduce/GroupReduce.cpp

@@ -32,9 +32,11 @@ namespace {
 // Contract: group_size >= 1 (enforced by PlanProto normalization).
 bool
 TryAcceptCompositeGroup(
-    const PkType& pk,
+    const SearchResultPair& result,
     const CompositeGroupKey& composite_key,
     std::unordered_set<PkType>& pk_set,
+    std::unordered_set<ElementSearchResultKey, ElementSearchResultKeyHash>&
+        element_result_set,
     std::unordered_map<CompositeGroupKey, int64_t, CompositeGroupKeyHash>&
         composite_group_by_map,
     int64_t topk,
@@ -42,9 +44,28 @@ TryAcceptCompositeGroup(
     AssertInfo(group_size >= 1,
                "group_size must be >= 1 (PlanProto normalizes), got {}",
                group_size);
-    if (pk_set.count(pk) != 0) {
-        return false;
+
+    auto search_result = result.search_result_;
+    ElementSearchResultKey element_key{result.primary_key_, -1};
+    if (search_result->element_level_) {
+        AssertInfo(
+            result.offset_ >= 0 && static_cast<size_t>(result.offset_) <
+                                       search_result->element_indices_.size(),
+            "invalid element-level search result offset {}, "
+            "element_indices size {}",
+            result.offset_,
+            search_result->element_indices_.size());
+        element_key.element_index =
+            search_result->element_indices_[result.offset_];
+        if (element_result_set.count(element_key) != 0) {
+            return false;
+        }
+    } else {
+        if (pk_set.count(result.primary_key_) != 0) {
+            return false;
+        }
     }
+
     auto [it, inserted] = composite_group_by_map.try_emplace(composite_key, 0);
     if (inserted &&
         static_cast<int64_t>(composite_group_by_map.size()) > topk) {
@@ -57,7 +78,11 @@ TryAcceptCompositeGroup(
         return false;
     }
     it->second += 1;
-    pk_set.insert(pk);
+    if (search_result->element_level_) {
+        element_result_set.insert(std::move(element_key));
+    } else {
+        pk_set.insert(result.primary_key_);
+    }
     return true;
 }
 
@@ -172,6 +197,7 @@ GroupReduceHelper::ReduceSearchResultForOneNQ(int64_t qi,
                         SearchResultPairComparator>
         heap;
     pk_set_.clear();
+    element_result_set_.clear();
     pairs_.clear();
     pairs_.reserve(num_segments_);
 
@@ -230,9 +256,10 @@ GroupReduceHelper::ReduceSearchResultForOneNQ(int64_t qi,
             pilot->search_result_->composite_group_by_values_
                 .value()[pilot->offset_];
 
-        if (TryAcceptCompositeGroup(pk,
+        if (TryAcceptCompositeGroup(*pilot,
                                     composite_key,
                                     pk_set_,
+                                    element_result_set_,
                                     composite_group_by_map,
                                     topk,
                                     group_size)) {

internal/core/src/segcore/reduce/Reduce.cpp

@@ -785,6 +785,27 @@ ReduceHelper::FillEntryData() {
     }
 }
 
+bool
+ReduceHelper::TryAcceptSearchResult(const SearchResultPair& result) {
+    auto search_result = result.search_result_;
+    if (!search_result->element_level_) {
+        return pk_set_.insert(result.primary_key_).second;
+    }
+
+    AssertInfo(
+        result.offset_ >= 0 && static_cast<size_t>(result.offset_) <
+                                   search_result->element_indices_.size(),
+        "invalid element-level search result offset {}, "
+        "element_indices size {}",
+        result.offset_,
+        search_result->element_indices_.size());
+
+    return element_result_set_
+        .insert({result.primary_key_,
+                 search_result->element_indices_[result.offset_]})
+        .second;
+}
+
 int64_t
 ReduceHelper::ReduceSearchResultForOneNQ(int64_t qi,
                                          int64_t topk,
@@ -794,6 +815,7 @@ ReduceHelper::ReduceSearchResultForOneNQ(int64_t qi,
                         SearchResultPairComparator>
         heap;
     pk_set_.clear();
+    element_result_set_.clear();
     pairs_.clear();
 
     pairs_.reserve(num_segments_);
@@ -828,13 +850,12 @@ ReduceHelper::ReduceSearchResultForOneNQ(int64_t qi,
         if (pk == INVALID_PK) {
             break;
         }
-        // remove duplicates
-        if (pk_set_.count(pk) == 0) {
+        // Row-level search is deduplicated by PK. Element-level search has
+        // multiple result identities per row, so use (PK, element_index).
+        if (TryAcceptSearchResult(*pilot)) {
             pilot->search_result_->result_offsets_.push_back(offset++);
             final_search_records_[index][qi].push_back(pilot->offset_);
-            pk_set_.insert(pk);
         } else {
-            // skip entity with same primary key
             dup_cnt++;
         }
         pilot->advance();

internal/core/src/segcore/reduce/Reduce.h

@@ -13,6 +13,7 @@
 
 #include <cstddef>
 #include <cstdint>
+#include <functional>
 #include <memory>
 #include <string>
 #include <unordered_set>
@@ -38,6 +39,26 @@ struct SearchResultDataBlobs {
     std::vector<StorageCost> costs;        // the cost of each slice
 };
 
+struct ElementSearchResultKey {
+    milvus::PkType pk;
+    int32_t element_index;
+
+    bool
+    operator==(const ElementSearchResultKey& other) const {
+        return pk == other.pk && element_index == other.element_index;
+    }
+};
+
+struct ElementSearchResultKeyHash {
+    size_t
+    operator()(const ElementSearchResultKey& key) const {
+        auto seed = std::hash<milvus::PkType>{}(key.pk);
+        seed ^= std::hash<int32_t>{}(key.element_index) + 0x9e3779b9 +
+                (seed << 6) + (seed >> 2);
+        return seed;
+    }
+};
+
 class ReduceHelper {
  public:
     explicit ReduceHelper(
@@ -155,6 +176,9 @@ class ReduceHelper {
     GetSearchResultDataSlice(const int slice_index,
                              const StorageCost& total_cost);
 
+    bool
+    TryAcceptSearchResult(const SearchResultPair& result);
+
     void
     GetTotalStorageCost();
 
@@ -170,6 +194,8 @@ class ReduceHelper {
     // define these here to avoid allocating them for each query
     std::vector<SearchResultPair> pairs_;
     std::unordered_set<milvus::PkType> pk_set_;
+    std::unordered_set<ElementSearchResultKey, ElementSearchResultKeyHash>
+        element_result_set_;
     // dim0: num_segments_; dim1: total_nq_; dim2: offset
     std::vector<std::vector<std::vector<int64_t>>> final_search_records_;
     std::vector<int64_t> slice_nqs_;

internal/core/src/segcore/reduce_c_test.cpp

@@ -35,6 +35,7 @@
 #include "segcore/Collection.h"
 #include "segcore/ReduceStructure.h"
 #include "segcore/collection_c.h"
+#include "segcore/reduce/GroupReduce.h"
 #include "segcore/reduce/Reduce.h"
 #include "segcore/reduce_c.h"
 #include "segcore/segment_c.h"
@@ -56,6 +57,12 @@ class TestReduceHelper : public ReduceHelper {
     using ReduceHelper::CanUseGlobalRefine;
     using ReduceHelper::IsSearchResultRefineEnabled;
     using ReduceHelper::ReduceHelper;
+    using ReduceHelper::ReduceResultData;
+
+    const std::vector<std::vector<std::vector<int64_t>>>&
+    FinalSearchRecordsForTest() const {
+        return final_search_records_;
+    }
 
     void
     TruncateForTest() {
@@ -114,6 +121,24 @@ class TestReduceHelper : public ReduceHelper {
         search_result_refine_enabled_by_result_for_test_;
 };
 
+class TestGroupReduceHelper : public GroupReduceHelper {
+ public:
+    using GroupReduceHelper::GroupReduceHelper;
+    using GroupReduceHelper::ReduceResultData;
+
+    const std::vector<std::vector<std::vector<int64_t>>>&
+    FinalSearchRecordsForTest() const {
+        return final_search_records_;
+    }
+};
+
+CompositeGroupKey
+MakeInt64CompositeKey(int64_t v) {
+    CompositeGroupKey key;
+    key.Add(GroupByValueType(std::in_place, v));
+    return key;
+}
+
 }  // namespace
 
 TEST(CApiTest, ReduceNullResult) {
@@ -335,6 +360,83 @@ TEST(CApiTest, ReduceRemoveDuplicates) {
     DeleteSegment(segment);
 }
 
+TEST(CApiTest, ReduceKeepsDistinctElementLevelHitsWithSamePK) {
+    SearchResult seg0;
+    seg0.total_nq_ = 1;
+    seg0.topk_per_nq_prefix_sum_ = {0, 4};
+    seg0.primary_keys_ = {
+        int64_t(5),
+        int64_t(5),
+        int64_t(5),
+        int64_t(6),
+    };
+    seg0.distances_ = {0.99f, 0.98f, 0.97f, 0.96f};
+    seg0.seg_offsets_ = {10, 10, 10, 11};
+    seg0.element_level_ = true;
+    seg0.element_indices_ = {0, 1, 1, 0};
+
+    std::vector<SearchResult*> search_results{&seg0};
+    std::vector<int64_t> slice_nqs{1};
+    std::vector<int64_t> slice_topks{4};
+    TestReduceHelper helper(search_results,
+                            nullptr,
+                            nullptr,
+                            slice_nqs.data(),
+                            slice_topks.data(),
+                            slice_nqs.size(),
+                            nullptr);
+
+    helper.ReduceResultData();
+
+    EXPECT_EQ(seg0.result_offsets_, std::vector<int64_t>({0, 1, 2}));
+    ASSERT_EQ(helper.FinalSearchRecordsForTest().size(), 1);
+    ASSERT_EQ(helper.FinalSearchRecordsForTest()[0].size(), 1);
+    EXPECT_EQ(helper.FinalSearchRecordsForTest()[0][0],
+              std::vector<int64_t>({0, 1, 3}));
+}
+
+TEST(CApiTest, GroupReduceKeepsDistinctElementLevelHitsWithSamePK) {
+    SearchResult seg0;
+    seg0.total_nq_ = 1;
+    seg0.topk_per_nq_prefix_sum_ = {0, 4};
+    seg0.primary_keys_ = {
+        int64_t(5),
+        int64_t(5),
+        int64_t(5),
+        int64_t(6),
+    };
+    seg0.distances_ = {0.99f, 0.98f, 0.97f, 0.96f};
+    seg0.seg_offsets_ = {10, 10, 10, 11};
+    seg0.element_level_ = true;
+    seg0.element_indices_ = {0, 1, 1, 0};
+    seg0.group_size_ = 2;
+    seg0.composite_group_by_values_ = std::vector<CompositeGroupKey>{
+        MakeInt64CompositeKey(5),
+        MakeInt64CompositeKey(5),
+        MakeInt64CompositeKey(5),
+        MakeInt64CompositeKey(6),
+    };
+
+    std::vector<SearchResult*> search_results{&seg0};
+    std::vector<int64_t> slice_nqs{1};
+    std::vector<int64_t> slice_topks{2};
+    TestGroupReduceHelper helper(search_results,
+                                 nullptr,
+                                 nullptr,
+                                 slice_nqs.data(),
+                                 slice_topks.data(),
+                                 slice_nqs.size(),
+                                 nullptr);
+
+    helper.ReduceResultData();
+
+    EXPECT_EQ(seg0.result_offsets_, std::vector<int64_t>({0, 1, 2}));
+    ASSERT_EQ(helper.FinalSearchRecordsForTest().size(), 1);
+    ASSERT_EQ(helper.FinalSearchRecordsForTest()[0].size(), 1);
+    EXPECT_EQ(helper.FinalSearchRecordsForTest()[0][0],
+              std::vector<int64_t>({0, 1, 3}));
+}
+
 template <class TraitType>
 void
 testReduceSearchWithExpr(int N,
@@ -1310,4 +1412,4 @@ TEST(CApiTest, GlobalRefineRejectsGroupBy) {
         search_results, &plan, nullptr, slice_nqs, slice_topks, 1, nullptr);
 
     ASSERT_THROW(helper.Reduce(), std::runtime_error);
-}
+}

internal/proxy/task_search.go

@@ -913,7 +913,7 @@ func (t *searchTask) initSearchRequest(ctx context.Context) error {
 
 	// For ArrayOfVector fields, the placeholder type decides the search semantics:
 	// - Element-level (plain vector placeholder): behaves like a normal single-vector
-	//   search; supports range search, iterator, and group by primary key.
+	//   search; supports range search, search iterator v2, and group by primary key.
 	// - Embedding-list-level (multi-search-multi): does not support range search,
 	//   iterator, or group by (other than the PK case above).
 	annsField := typeutil.GetField(t.schema.CollectionSchema, t.FieldId)
@@ -929,6 +929,9 @@ func (t *searchTask) initSearchRequest(ctx context.Context) error {
 				return merr.WrapErrParameterInvalid("", "",
 					"search iterator is not supported for multi-search-multi on embedding list fields")
 			}
+		} else if t.isIterator && queryInfo.GetSearchIteratorV2Info() == nil {
+			return merr.WrapErrParameterInvalid("", "",
+				"legacy search iterator is not supported for element-level search on embedding list fields; use search iterator v2")
 		}
 
 		groupByFieldIDs := queryInfo.GetGroupByFieldIds()

internal/proxy/task_search_test.go

@@ -6089,9 +6089,9 @@ func TestSearchTask_ArrayOfVectorSimpleSearch(t *testing.T) {
 		return bs
 	}
 
-	// paramsJSON is inlined into the ParamsKey field; setting withIterator adds
-	// the iterator flag as a separate KV pair.
-	makeTask := func(annsField string, phType commonpb.PlaceholderType, paramsJSON string, withIterator bool) *searchTask {
+	// paramsJSON is inlined into the ParamsKey field; iterator flags are
+	// separate KV pairs to match SDK requests.
+	makeTask := func(annsField string, phType commonpb.PlaceholderType, paramsJSON string, withIterator bool, withIteratorV2 bool) *searchTask {
 		params := []*commonpb.KeyValuePair{
 			{Key: AnnsFieldKey, Value: annsField},
 			{Key: TopKKey, Value: "10"},
@@ -6101,6 +6101,12 @@ func TestSearchTask_ArrayOfVectorSimpleSearch(t *testing.T) {
 		if withIterator {
 			params = append(params, &commonpb.KeyValuePair{Key: IteratorField, Value: "True"})
 		}
+		if withIteratorV2 {
+			params = append(params,
+				&commonpb.KeyValuePair{Key: SearchIterV2Key, Value: "True"},
+				&commonpb.KeyValuePair{Key: SearchIterBatchSizeKey, Value: "10"},
+			)
+		}
 
 		phgBytes := makePlaceholderGroup(phType)
 		return &searchTask{
@@ -6134,19 +6140,27 @@ func TestSearchTask_ArrayOfVectorSimpleSearch(t *testing.T) {
 	const plainParams = `{"nprobe": 10}`
 
 	t.Run("element-level range search should succeed", func(t *testing.T) {
-		task := makeTask("emb_vec", commonpb.PlaceholderType_FloatVector, rangeParams, false)
+		task := makeTask("emb_vec", commonpb.PlaceholderType_FloatVector, rangeParams, false, false)
 		err := task.initSearchRequest(ctx)
 		assert.NoError(t, err)
 	})
 
-	t.Run("element-level iterator should succeed", func(t *testing.T) {
-		task := makeTask("emb_vec", commonpb.PlaceholderType_FloatVector, plainParams, true)
+	t.Run("element-level legacy iterator should fail", func(t *testing.T) {
+		task := makeTask("emb_vec", commonpb.PlaceholderType_FloatVector, plainParams, true, false)
+		err := task.initSearchRequest(ctx)
+		assert.Error(t, err)
+		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
+		assert.Contains(t, err.Error(), "legacy search iterator is not supported for element-level search")
+	})
+
+	t.Run("element-level iterator v2 should succeed", func(t *testing.T) {
+		task := makeTask("emb_vec", commonpb.PlaceholderType_FloatVector, plainParams, true, true)
 		err := task.initSearchRequest(ctx)
 		assert.NoError(t, err)
 	})
 
 	t.Run("emblist range search should fail", func(t *testing.T) {
-		task := makeTask("emb_vec", commonpb.PlaceholderType_EmbListFloatVector, rangeParams, false)
+		task := makeTask("emb_vec", commonpb.PlaceholderType_EmbListFloatVector, rangeParams, false, false)
 		err := task.initSearchRequest(ctx)
 		assert.Error(t, err)
 		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
@@ -6154,7 +6168,7 @@ func TestSearchTask_ArrayOfVectorSimpleSearch(t *testing.T) {
 	})
 
 	t.Run("emblist iterator should fail", func(t *testing.T) {
-		task := makeTask("emb_vec", commonpb.PlaceholderType_EmbListFloatVector, plainParams, true)
+		task := makeTask("emb_vec", commonpb.PlaceholderType_EmbListFloatVector, plainParams, true, false)
 		err := task.initSearchRequest(ctx)
 		assert.Error(t, err)
 		assert.ErrorIs(t, err, merr.ErrParameterInvalid)
@@ -6163,7 +6177,7 @@ func TestSearchTask_ArrayOfVectorSimpleSearch(t *testing.T) {
 
 	t.Run("regular vector range search should succeed", func(t *testing.T) {
 		// Regression: new checks must not impact plain FloatVector fields.
-		task := makeTask("regular_vec", commonpb.PlaceholderType_FloatVector, rangeParams, false)
+		task := makeTask("regular_vec", commonpb.PlaceholderType_FloatVector, rangeParams, false, false)
 		err := task.initSearchRequest(ctx)
 		assert.NoError(t, err)
 	})

internal/querynodev2/segments/search_reduce.go

@@ -19,6 +19,25 @@ type SearchReduce interface {
 	ReduceSearchResultData(ctx context.Context, searchResultData []*schemapb.SearchResultData, info *reduce.ResultInfo) (*schemapb.SearchResultData, error)
 }
 
+type elementSearchResultKey struct {
+	pk           interface{}
+	elementIndex int64
+}
+
+func getSearchResultDedupKey(data *schemapb.SearchResultData, idx int64, pk interface{}, hasElementIndices bool) (interface{}, error) {
+	if !hasElementIndices {
+		return pk, nil
+	}
+	elementIndices := data.GetElementIndices()
+	if elementIndices == nil || idx < 0 || idx >= int64(len(elementIndices.GetData())) {
+		return nil, fmt.Errorf("element-level search result missing element index at offset %d", idx)
+	}
+	return elementSearchResultKey{
+		pk:           pk,
+		elementIndex: elementIndices.GetData()[idx],
+	}, nil
+}
+
 type SearchCommonReduce struct{}
 
 func (scr *SearchCommonReduce) ReduceSearchResultData(ctx context.Context, searchResultData []*schemapb.SearchResultData, info *reduce.ResultInfo) (*schemapb.SearchResultData, error) {
@@ -76,18 +95,23 @@ func (scr *SearchCommonReduce) ReduceSearchResultData(ctx context.Context, searc
 
 	resultOffsets := make([][]int64, len(searchResultData))
 	totalOffsetElements := 0
-	for i := range searchResultData {
-		totalOffsetElements += len(searchResultData[i].Topks)
+	for i, data := range searchResultData {
+		if int64(len(data.Topks)) < nq {
+			return nil, fmt.Errorf("invalid search result topks length at index %d: got %d, expected at least %d", i, len(data.Topks), nq)
+		}
+		totalOffsetElements += len(data.Topks)
 	}
 	offsetBacking := make([]int64, totalOffsetElements)
 	for i := 0; i < len(searchResultData); i++ {
-		n := len(searchResultData[i].Topks)
+		data := searchResultData[i]
+		topks := data.Topks
+		n := len(topks)
 		resultOffsets[i] = offsetBacking[:n:n]
 		offsetBacking = offsetBacking[n:]
-		for j := int64(1); j < nq; j++ {
-			resultOffsets[i][j] = resultOffsets[i][j-1] + searchResultData[i].Topks[j-1]
+		for j := 1; j < n; j++ {
+			resultOffsets[i][j] = resultOffsets[i][j-1] + topks[j-1]
 		}
-		ret.AllSearchCount += searchResultData[i].GetAllSearchCount()
+		ret.AllSearchCount += data.GetAllSearchCount()
 	}
 
 	idxComputers := make([]*typeutil.FieldDataIdxComputer, len(searchResultData))
@@ -101,7 +125,7 @@ func (scr *SearchCommonReduce) ReduceSearchResultData(ctx context.Context, searc
 	maxOutputSize := paramtable.Get().QuotaConfig.MaxOutputSize.GetAsInt64()
 	for i := int64(0); i < nq; i++ {
 		offsets := make([]int64, numResults)
-		idSet := make(map[interface{}]struct{})
+		dedupSet := make(map[interface{}]struct{})
 		var j int64
 		for j = 0; j < topk; {
 			sel := SelectSearchResultData(searchResultData, resultOffsets, offsets, i)
@@ -112,9 +136,13 @@ func (scr *SearchCommonReduce) ReduceSearchResultData(ctx context.Context, searc
 
 			id := typeutil.GetPK(searchResultData[sel].GetIds(), idx)
 			score := searchResultData[sel].Scores[idx]
+			dedupKey, err := getSearchResultDedupKey(searchResultData[sel], idx, id, hasElementIndices)
+			if err != nil {
+				return nil, err
+			}
 
 			// remove duplicates
-			if _, ok := idSet[id]; !ok {
+			if _, ok := dedupSet[dedupKey]; !ok {
 				fieldsData := searchResultData[sel].FieldsData
 				fieldIdxs := idxComputers[sel].Compute(idx)
 				retSize += typeutil.AppendFieldData(ret.FieldsData, fieldsData, idx, fieldIdxs...)
@@ -123,10 +151,10 @@ func (scr *SearchCommonReduce) ReduceSearchResultData(ctx context.Context, searc
 				if searchResultData[sel].ElementIndices != nil && ret.ElementIndices != nil {
 					ret.ElementIndices.Data = append(ret.ElementIndices.Data, searchResultData[sel].ElementIndices.Data[idx])
 				}
-				idSet[id] = struct{}{}
+				dedupSet[dedupKey] = struct{}{}
 				j++
 			} else {
-				// skip entity with same id
+				// skip the same row-level entity or the same element-level hit
 				skipDupCnt++
 			}
 			offsets[sel]++
@@ -205,18 +233,23 @@ func (sbr *SearchGroupByReduce) ReduceSearchResultData(ctx context.Context, sear
 
 	resultOffsets := make([][]int64, len(searchResultData))
 	totalOffsetElements := 0
-	for i := range searchResultData {
-		totalOffsetElements += len(searchResultData[i].Topks)
+	for i, data := range searchResultData {
+		if int64(len(data.Topks)) < nq {
+			return nil, fmt.Errorf("invalid search result topks length at index %d: got %d, expected at least %d", i, len(data.Topks), nq)
+		}
+		totalOffsetElements += len(data.Topks)
 	}
 	offsetBacking := make([]int64, totalOffsetElements)
 	for i := range searchResultData {
-		n := len(searchResultData[i].Topks)
+		data := searchResultData[i]
+		topks := data.Topks
+		n := len(topks)
 		resultOffsets[i] = offsetBacking[:n:n]
 		offsetBacking = offsetBacking[n:]
-		for j := int64(1); j < nq; j++ {
-			resultOffsets[i][j] = resultOffsets[i][j-1] + searchResultData[i].Topks[j-1]
+		for j := 1; j < n; j++ {
+			resultOffsets[i][j] = resultOffsets[i][j-1] + topks[j-1]
 		}
-		ret.AllSearchCount += searchResultData[i].GetAllSearchCount()
+		ret.AllSearchCount += data.GetAllSearchCount()
 	}
 
 	idxComputers := make([]*typeutil.FieldDataIdxComputer, len(searchResultData))
@@ -251,12 +284,16 @@ func (sbr *SearchGroupByReduce) ReduceSearchResultData(ctx context.Context, sear
 
 	for i := int64(0); i < info.GetNq(); i++ {
 		var j, fdelta, rsize int64
+		var err error
 		if singleField {
-			j, fdelta, rsize = reduceGroupBySinglePerNq(searchResultData, resultOffsets, i,
-				info.GetTopK(), groupSize, groupBound, singleIters, idxComputers, ret, &acceptedRows)
+			j, fdelta, rsize, err = reduceGroupBySinglePerNq(searchResultData, resultOffsets, i,
+				info.GetTopK(), groupSize, groupBound, singleIters, idxComputers, ret, &acceptedRows, hasElementIndices)
 		} else {
-			j, fdelta, rsize = reduceGroupByMultiPerNq(searchResultData, resultOffsets, i,
-				info.GetTopK(), groupSize, groupBound, multiExtractors, idxComputers, ret, &acceptedRows)
+			j, fdelta, rsize, err = reduceGroupByMultiPerNq(searchResultData, resultOffsets, i,
+				info.GetTopK(), groupSize, groupBound, multiExtractors, idxComputers, ret, &acceptedRows, hasElementIndices)
+		}
+		if err != nil {
+			return nil, err
 		}
 		filteredCount += fdelta
 		retSize += rsize
@@ -314,9 +351,10 @@ func reduceGroupBySinglePerNq(
 	idxComputers []*typeutil.FieldDataIdxComputer,
 	ret *schemapb.SearchResultData,
 	acceptedRows *[]reduce.RowRef,
-) (j int64, filtered int64, retSize int64) {
+	hasElementIndices bool,
+) (j int64, filtered int64, retSize int64, err error) {
 	offsets := make([]int64, len(searchResultData))
-	idSet := make(map[interface{}]struct{})
+	dedupSet := make(map[interface{}]struct{})
 	groupCounts := make(map[any]int64)
 
 	for j = 0; j < groupBound; {
@@ -329,7 +367,11 @@ func reduceGroupBySinglePerNq(
 		id := typeutil.GetPK(searchResultData[sel].GetIds(), idx)
 		val := iterators[sel](int(idx))
 		score := searchResultData[sel].Scores[idx]
-		if _, ok := idSet[id]; !ok {
+		dedupKey, dedupErr := getSearchResultDedupKey(searchResultData[sel], idx, id, hasElementIndices)
+		if dedupErr != nil {
+			return j, filtered, retSize, dedupErr
+		}
+		if _, ok := dedupSet[dedupKey]; !ok {
 			cnt, exists := groupCounts[val]
 			switch {
 			case !exists && int64(len(groupCounts)) >= topK:
@@ -345,7 +387,7 @@ func reduceGroupBySinglePerNq(
 					ret.ElementIndices.Data = append(ret.ElementIndices.Data, searchResultData[sel].ElementIndices.Data[idx])
 				}
 				groupCounts[val] = cnt + 1
-				idSet[id] = struct{}{}
+				dedupSet[dedupKey] = struct{}{}
 				*acceptedRows = append(*acceptedRows, reduce.RowRef{ResultIdx: sel, RowIdx: idx})
 				j++
 			}
@@ -368,9 +410,10 @@ func reduceGroupByMultiPerNq(
 	idxComputers []*typeutil.FieldDataIdxComputer,
 	ret *schemapb.SearchResultData,
 	acceptedRows *[]reduce.RowRef,
-) (j int64, filtered int64, retSize int64) {
+	hasElementIndices bool,
+) (j int64, filtered int64, retSize int64, err error) {
 	offsets := make([]int64, len(searchResultData))
-	idSet := make(map[interface{}]struct{})
+	dedupSet := make(map[interface{}]struct{})
 	groupBuckets := make(map[uint64][]*reduceGroupEntry)
 	totalGroups := int64(0)
 
@@ -384,7 +427,11 @@ func reduceGroupByMultiPerNq(
 		id := typeutil.GetPK(searchResultData[sel].GetIds(), idx)
 		hash, values := extractors[sel](int(idx))
 		score := searchResultData[sel].Scores[idx]
-		if _, ok := idSet[id]; !ok {
+		dedupKey, dedupErr := getSearchResultDedupKey(searchResultData[sel], idx, id, hasElementIndices)
+		if dedupErr != nil {
+			return j, filtered, retSize, dedupErr
+		}
+		if _, ok := dedupSet[dedupKey]; !ok {
 			entry := findReduceGroupEntry(groupBuckets[hash], values)
 			isNewGroup := entry == nil
 			switch {
@@ -406,7 +453,7 @@ func reduceGroupByMultiPerNq(
 					ret.ElementIndices.Data = append(ret.ElementIndices.Data, searchResultData[sel].ElementIndices.Data[idx])
 				}
 				entry.count++
-				idSet[id] = struct{}{}
+				dedupSet[dedupKey] = struct{}{}
 				*acceptedRows = append(*acceptedRows, reduce.RowRef{ResultIdx: sel, RowIdx: idx})
 				j++
 			}

internal/querynodev2/segments/search_reduce_test.go

@@ -561,6 +561,59 @@ func (suite *SearchReduceSuite) TestElementIndices_BackfillNilForEmptyResult() {
 	})
 }
 
+func (suite *SearchReduceSuite) TestElementLevelDedupUsesPKAndElementIndex() {
+	const (
+		nq   = 1
+		topk = 4
+	)
+
+	makeData := func() *schemapb.SearchResultData {
+		data := mock_segcore.GenSearchResultData(nq, topk,
+			[]int64{5, 5, 5, 6},
+			[]float32{0.99, 0.98, 0.97, 0.96},
+			[]int64{4},
+		)
+		data.ElementIndices = &schemapb.LongArray{Data: []int64{0, 1, 1, 0}}
+		return data
+	}
+
+	suite.Run("common_reduce", func() {
+		reduceInfo := reduce.NewReduceSearchResultInfo(nq, topk).WithGroupSize(1)
+		searchReduce := &SearchCommonReduce{}
+		res, err := searchReduce.ReduceSearchResultData(context.TODO(), []*schemapb.SearchResultData{makeData()}, reduceInfo)
+		suite.NoError(err)
+
+		suite.Equal([]int64{5, 5, 6}, res.Ids.GetIntId().Data)
+		suite.Equal([]float32{0.99, 0.98, 0.96}, res.Scores)
+		suite.Equal([]int64{0, 1, 0}, res.ElementIndices.GetData())
+		suite.Equal([]int64{3}, res.Topks)
+	})
+
+	suite.Run("group_by_reduce", func() {
+		data := makeData()
+		data.GroupByFieldValue = &schemapb.FieldData{
+			Type: schemapb.DataType_Int64,
+			Field: &schemapb.FieldData_Scalars{
+				Scalars: &schemapb.ScalarField{
+					Data: &schemapb.ScalarField_LongData{
+						LongData: &schemapb.LongArray{Data: []int64{5, 5, 5, 6}},
+					},
+				},
+			},
+		}
+
+		reduceInfo := reduce.NewReduceSearchResultInfo(nq, 2).WithGroupSize(2).WithGroupByFieldIdsFromProto(101, nil)
+		searchReduce := &SearchGroupByReduce{}
+		res, err := searchReduce.ReduceSearchResultData(context.TODO(), []*schemapb.SearchResultData{data}, reduceInfo)
+		suite.NoError(err)
+
+		suite.Equal([]int64{5, 5, 6}, res.Ids.GetIntId().Data)
+		suite.Equal([]float32{0.99, 0.98, 0.96}, res.Scores)
+		suite.Equal([]int64{0, 1, 0}, res.ElementIndices.GetData())
+		suite.Equal([]int64{3}, res.Topks)
+	})
+}
+
 func (suite *SearchReduceSuite) TestElementIndices_NoElementLevel() {
 	// When no result has ElementIndices, ret.ElementIndices should remain nil.
 	const (

tests/python_client/milvus_client/test_milvus_client_struct_array_element_search.py

@@ -5789,9 +5789,17 @@ class TestMilvusClientStructArrayElementSearchNoFilter(TestMilvusClientV2Base):
         assert check
         assert len(results[0]) == limit
 
-        # No duplicate IDs (each row appears at most once)
+        # Element-level search returns element hits, so the same row id may
+        # appear multiple times. If pymilvus exposes element offset, the
+        # element identity should still be unique.
         ids = [hit["id"] for hit in results[0]]
-        assert len(ids) == len(set(ids)), f"Duplicate IDs in results: {len(ids)} total, {len(set(ids))} unique"
+        offsets = [hit.get("offset") for hit in results[0]]
+        if any(offset is not None for offset in offsets):
+            assert all(offset is not None for offset in offsets), "Some element hits are missing offset"
+            element_keys = list(zip(ids, offsets))
+            assert len(element_keys) == len(set(element_keys)), (
+                f"Duplicate element hits in results: {len(element_keys)} total, {len(set(element_keys))} unique"
+            )
 
         # Distances monotonically decreasing
         distances = [hit["distance"] for hit in results[0]]