fix: Filter empty search results to prevent rerank panic

issue: #44909

This change adds validation to filter out empty search results
before reranking operations in both regular and advanced search
scenarios.

When QueryNode returns empty search results (with no IDs or
fields), the rerank process would panic when trying to access
result data. This happens in hybrid search scenarios where some
sub-requests may return empty results.

Changes include:
- Add empty result filtering in reduceAdvanceGroupBy
- Add empty result filtering in reduceSearchResultDataWithGroupBy
- Add empty result filtering in reduceSearchResultDataNoGroupBy
- Improve newRerankOutputs to safely handle empty fieldData
- Add comprehensive unit tests for empty result handling

This fix ensures empty results are filtered out before being
passed to ranking functions, preventing the panic.

Signed-off-by: Wei Liu <[email protected]>

Author: weiliu1031

internal/proxy/search_reduce_util.go

@@ -75,11 +75,25 @@ func reduceAdvanceGroupBy(ctx context.Context, subSearchResultData []*schemapb.S
 	nq int64, topK int64, pkType schemapb.DataType, metricType string,
 ) (*milvuspb.SearchResults, error) {
 	log.Ctx(ctx).Debug("reduceAdvanceGroupBY", zap.Int("len(subSearchResultData)", len(subSearchResultData)), zap.Int64("nq", nq))
+
+	validSubSearchResultData := []*schemapb.SearchResultData{}
+	for _, result := range subSearchResultData {
+		size := typeutil.GetSizeOfIDs(result.GetIds())
+		if result == nil || size == 0 {
+			continue
+		}
+		validSubSearchResultData = append(validSubSearchResultData, result)
+	}
+
+	if len(validSubSearchResultData) == 0 {
+		return fillInEmptyResult(nq), nil
+	}
+
 	// for advance group by, offset is not applied, so just return when there's only one channel
-	if len(subSearchResultData) == 1 {
+	if len(validSubSearchResultData) == 1 {
 		return &milvuspb.SearchResults{
 			Status:  merr.Success(),
-			Results: subSearchResultData[0],
+			Results: validSubSearchResultData[0],
 		}, nil
 	}
 
@@ -95,45 +109,45 @@ func reduceAdvanceGroupBy(ctx context.Context, subSearchResultData []*schemapb.S
 	}
 
 	var limit int64
-	if allSearchCount, hitNum, err := checkResultDatas(ctx, subSearchResultData, nq, topK); err != nil {
+	if allSearchCount, hitNum, err := checkResultDatas(ctx, validSubSearchResultData, nq, topK); err != nil {
 		log.Ctx(ctx).Warn("invalid search results", zap.Error(err))
 		return ret, err
 	} else {
 		ret.GetResults().AllSearchCount = allSearchCount
 		limit = int64(hitNum)
-		ret.GetResults().FieldsData = typeutil.PrepareResultFieldData(subSearchResultData[0].GetFieldsData(), limit)
+		ret.GetResults().FieldsData = typeutil.PrepareResultFieldData(validSubSearchResultData[0].GetFieldsData(), limit)
 	}
 
 	if err := setupIdListForSearchResult(ret, pkType, limit); err != nil {
 		return ret, nil
 	}
 
 	var (
-		subSearchNum = len(subSearchResultData)
+		subSearchNum = len(validSubSearchResultData)
 		// for results of each subSearchResultData, storing the start offset of each query of nq queries
 		subSearchNqOffset = make([][]int64, subSearchNum)
 	)
 	for i := 0; i < subSearchNum; i++ {
-		subSearchNqOffset[i] = make([]int64, subSearchResultData[i].GetNumQueries())
+		subSearchNqOffset[i] = make([]int64, validSubSearchResultData[i].GetNumQueries())
 		for j := int64(1); j < nq; j++ {
-			subSearchNqOffset[i][j] = subSearchNqOffset[i][j-1] + subSearchResultData[i].Topks[j-1]
+			subSearchNqOffset[i][j] = subSearchNqOffset[i][j-1] + validSubSearchResultData[i].Topks[j-1]
 		}
 	}
 
-	gpFieldBuilder, err := typeutil.NewFieldDataBuilder(subSearchResultData[0].GetGroupByFieldValue().GetType(), true, int(limit))
+	gpFieldBuilder, err := typeutil.NewFieldDataBuilder(validSubSearchResultData[0].GetGroupByFieldValue().GetType(), true, int(limit))
 	if err != nil {
 		return ret, merr.WrapErrServiceInternal("failed to construct group by field data builder, this is abnormal as segcore should always set up a group by field, no matter data status, check code on qn", err.Error())
 	}
 	// reducing nq * topk results
 	for nqIdx := int64(0); nqIdx < nq; nqIdx++ {
 		dataCount := int64(0)
 		for subIdx := 0; subIdx < subSearchNum; subIdx += 1 {
-			subData := subSearchResultData[subIdx]
+			subData := validSubSearchResultData[subIdx]
 			subPks := subData.GetIds()
 			subScores := subData.GetScores()
 			subGroupByVals := subData.GetGroupByFieldValue()
 
-			nqTopK := subData.Topks[nqIdx]
+			nqTopK := validSubSearchResultData[subIdx].Topks[nqIdx]
 			groupByValIterator := typeutil.GetDataIterator(subGroupByVals)
 
 			for i := int64(0); i < nqTopK; i++ {
@@ -188,12 +202,25 @@ func reduceSearchResultDataWithGroupBy(ctx context.Context, subSearchResultData
 		zap.Int64("limit", limit),
 		zap.String("metricType", metricType))
 
+	validSubSearchResultData := []*schemapb.SearchResultData{}
+	for _, result := range subSearchResultData {
+		size := typeutil.GetSizeOfIDs(result.GetIds())
+		if result == nil || size == 0 {
+			continue
+		}
+		validSubSearchResultData = append(validSubSearchResultData, result)
+	}
+
+	if len(validSubSearchResultData) == 0 {
+		return fillInEmptyResult(nq), nil
+	}
+
 	ret := &milvuspb.SearchResults{
 		Status: merr.Success(),
 		Results: &schemapb.SearchResultData{
 			NumQueries: nq,
 			TopK:       topk,
-			FieldsData: typeutil.PrepareResultFieldData(subSearchResultData[0].GetFieldsData(), limit),
+			FieldsData: typeutil.PrepareResultFieldData(validSubSearchResultData[0].GetFieldsData(), limit),
 			Scores:     []float32{},
 			Ids:        &schemapb.IDs{},
 			Topks:      []int64{},
@@ -204,30 +231,30 @@ func reduceSearchResultDataWithGroupBy(ctx context.Context, subSearchResultData
 		return ret, err
 	}
 
-	if allSearchCount, _, err := checkResultDatas(ctx, subSearchResultData, nq, topk); err != nil {
+	if allSearchCount, _, err := checkResultDatas(ctx, validSubSearchResultData, nq, topk); err != nil {
 		log.Ctx(ctx).Warn("invalid search results", zap.Error(err))
 		return ret, err
 	} else {
 		ret.GetResults().AllSearchCount = allSearchCount
 	}
 
 	var (
-		subSearchNum = len(subSearchResultData)
+		subSearchNum = len(validSubSearchResultData)
 		// for results of each subSearchResultData, storing the start offset of each query of nq queries
 		subSearchNqOffset                 = make([][]int64, subSearchNum)
 		totalResCount               int64 = 0
 		subSearchGroupByValIterator       = make([]func(int) any, subSearchNum)
 	)
 	for i := 0; i < subSearchNum; i++ {
-		subSearchNqOffset[i] = make([]int64, subSearchResultData[i].GetNumQueries())
+		subSearchNqOffset[i] = make([]int64, validSubSearchResultData[i].GetNumQueries())
 		for j := int64(1); j < nq; j++ {
-			subSearchNqOffset[i][j] = subSearchNqOffset[i][j-1] + subSearchResultData[i].Topks[j-1]
+			subSearchNqOffset[i][j] = subSearchNqOffset[i][j-1] + validSubSearchResultData[i].Topks[j-1]
 		}
 		totalResCount += subSearchNqOffset[i][nq-1]
-		subSearchGroupByValIterator[i] = typeutil.GetDataIterator(subSearchResultData[i].GetGroupByFieldValue())
+		subSearchGroupByValIterator[i] = typeutil.GetDataIterator(validSubSearchResultData[i].GetGroupByFieldValue())
 	}
 
-	gpFieldBuilder, err := typeutil.NewFieldDataBuilder(subSearchResultData[0].GetGroupByFieldValue().GetType(), true, int(limit))
+	gpFieldBuilder, err := typeutil.NewFieldDataBuilder(validSubSearchResultData[0].GetGroupByFieldValue().GetType(), true, int(limit))
 	if err != nil {
 		return ret, merr.WrapErrServiceInternal("failed to construct group by field data builder, this is abnormal as segcore should always set up a group by field, no matter data status, check code on qn", err.Error())
 	}
@@ -251,11 +278,11 @@ func reduceSearchResultDataWithGroupBy(ctx context.Context, subSearchResultData
 		)
 
 		for j = 0; j < groupBound; {
-			subSearchIdx, resultDataIdx := selectHighestScoreIndex(ctx, subSearchResultData, subSearchNqOffset, cursors, i)
+			subSearchIdx, resultDataIdx := selectHighestScoreIndex(ctx, validSubSearchResultData, subSearchNqOffset, cursors, i)
 			if subSearchIdx == -1 {
 				break
 			}
-			subSearchRes := subSearchResultData[subSearchIdx]
+			subSearchRes := validSubSearchResultData[subSearchIdx]
 
 			id := typeutil.GetPK(subSearchRes.GetIds(), resultDataIdx)
 			score := subSearchRes.GetScores()[resultDataIdx]
@@ -288,7 +315,7 @@ func reduceSearchResultDataWithGroupBy(ctx context.Context, subSearchResultData
 		for _, groupVal := range groupByValList {
 			groupEntities := groupByValMap[groupVal]
 			for _, groupEntity := range groupEntities {
-				subResData := subSearchResultData[groupEntity.subSearchIdx]
+				subResData := validSubSearchResultData[groupEntity.subSearchIdx]
 				retSize += typeutil.AppendFieldData(ret.Results.FieldsData, subResData.FieldsData, groupEntity.resultIdx)
 				typeutil.AppendPKs(ret.Results.Ids, groupEntity.id)
 				ret.Results.Scores = append(ret.Results.Scores, groupEntity.score)
@@ -331,12 +358,25 @@ func reduceSearchResultDataNoGroupBy(ctx context.Context, subSearchResultData []
 		zap.Int64("limit", limit),
 		zap.String("metricType", metricType))
 
+	validSubSearchResultData := []*schemapb.SearchResultData{}
+	for _, result := range subSearchResultData {
+		size := typeutil.GetSizeOfIDs(result.GetIds())
+		if result == nil || size == 0 {
+			continue
+		}
+		validSubSearchResultData = append(validSubSearchResultData, result)
+	}
+
+	if len(validSubSearchResultData) == 0 {
+		return fillInEmptyResult(nq), nil
+	}
+
 	ret := &milvuspb.SearchResults{
 		Status: merr.Success(),
 		Results: &schemapb.SearchResultData{
 			NumQueries: nq,
 			TopK:       topk,
-			FieldsData: typeutil.PrepareResultFieldData(subSearchResultData[0].GetFieldsData(), limit),
+			FieldsData: typeutil.PrepareResultFieldData(validSubSearchResultData[0].GetFieldsData(), limit),
 			Scores:     []float32{},
 			Ids:        &schemapb.IDs{},
 			Topks:      []int64{},
@@ -347,20 +387,20 @@ func reduceSearchResultDataNoGroupBy(ctx context.Context, subSearchResultData []
 		return ret, nil
 	}
 
-	if allSearchCount, _, err := checkResultDatas(ctx, subSearchResultData, nq, topk); err != nil {
+	if allSearchCount, _, err := checkResultDatas(ctx, validSubSearchResultData, nq, topk); err != nil {
 		log.Ctx(ctx).Warn("invalid search results", zap.Error(err))
 		return ret, err
 	} else {
 		ret.GetResults().AllSearchCount = allSearchCount
 	}
 
-	subSearchNum := len(subSearchResultData)
+	subSearchNum := len(validSubSearchResultData)
 	if subSearchNum == 1 && offset == 0 {
 		// sorting is not needed if there is only one shard and no offset, assigning the result directly.
 		//  we still need to adjust the scores later.
-		ret.Results = subSearchResultData[0]
+		ret.Results = validSubSearchResultData[0]
 		// realTopK is the topK of the nq-th query, it is used in proxy but not handled by delegator.
-		topks := subSearchResultData[0].Topks
+		topks := validSubSearchResultData[0].Topks
 		if len(topks) > 0 {
 			ret.Results.TopK = topks[len(topks)-1]
 		}
@@ -371,9 +411,9 @@ func reduceSearchResultDataNoGroupBy(ctx context.Context, subSearchResultData []
 		// for results of each subSearchResultData, storing the start offset of each query of nq queries
 		subSearchNqOffset := make([][]int64, subSearchNum)
 		for i := 0; i < subSearchNum; i++ {
-			subSearchNqOffset[i] = make([]int64, subSearchResultData[i].GetNumQueries())
+			subSearchNqOffset[i] = make([]int64, validSubSearchResultData[i].GetNumQueries())
 			for j := int64(1); j < nq; j++ {
-				subSearchNqOffset[i][j] = subSearchNqOffset[i][j-1] + subSearchResultData[i].Topks[j-1]
+				subSearchNqOffset[i][j] = subSearchNqOffset[i][j-1] + validSubSearchResultData[i].Topks[j-1]
 			}
 		}
 		maxOutputSize := paramtable.Get().QuotaConfig.MaxOutputSize.GetAsInt64()
@@ -388,7 +428,7 @@ func reduceSearchResultDataNoGroupBy(ctx context.Context, subSearchResultData []
 
 			// skip offset results
 			for k := int64(0); k < offset; k++ {
-				subSearchIdx, _ := selectHighestScoreIndex(ctx, subSearchResultData, subSearchNqOffset, cursors, i)
+				subSearchIdx, _ := selectHighestScoreIndex(ctx, validSubSearchResultData, subSearchNqOffset, cursors, i)
 				if subSearchIdx == -1 {
 					break
 				}
@@ -401,14 +441,14 @@ func reduceSearchResultDataNoGroupBy(ctx context.Context, subSearchResultData []
 				// From all the sub-query result sets of the i-th query vector,
 				//   find the sub-query result set index of the score j-th data,
 				//   and the index of the data in schemapb.SearchResultData
-				subSearchIdx, resultDataIdx := selectHighestScoreIndex(ctx, subSearchResultData, subSearchNqOffset, cursors, i)
+				subSearchIdx, resultDataIdx := selectHighestScoreIndex(ctx, validSubSearchResultData, subSearchNqOffset, cursors, i)
 				if subSearchIdx == -1 {
 					break
 				}
-				score := subSearchResultData[subSearchIdx].Scores[resultDataIdx]
+				score := validSubSearchResultData[subSearchIdx].Scores[resultDataIdx]
 
-				retSize += typeutil.AppendFieldData(ret.Results.FieldsData, subSearchResultData[subSearchIdx].FieldsData, resultDataIdx)
-				typeutil.CopyPk(ret.Results.Ids, subSearchResultData[subSearchIdx].GetIds(), int(resultDataIdx))
+				retSize += typeutil.AppendFieldData(ret.Results.FieldsData, validSubSearchResultData[subSearchIdx].FieldsData, resultDataIdx)
+				typeutil.CopyPk(ret.Results.Ids, validSubSearchResultData[subSearchIdx].GetIds(), int(resultDataIdx))
 				ret.Results.Scores = append(ret.Results.Scores, score)
 				cursors[subSearchIdx]++
 			}