Reciprocal Rank Fusion (RRF) normalization technique in hybrid query

CHANGELOG.md

@@ -15,6 +15,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 
 ## [Unreleased 2.x](https://github.com/opensearch-project/neural-search/compare/2.17...2.x)
 ### Features
+- Implement Reciprocal Rank Fusion score normalization/combination technique in hybrid query ([#874](https://github.com/opensearch-project/neural-search/pull/874))
 ### Enhancements
 - Implement `ignore_missing` field in text chunking processors ([#907](https://github.com/opensearch-project/neural-search/pull/907))
 - Added rescorer in hybrid query ([#917](https://github.com/opensearch-project/neural-search/pull/917))

src/main/java/org/opensearch/neuralsearch/plugin/NeuralSearch.java

@@ -30,20 +30,22 @@
 import org.opensearch.neuralsearch.ml.MLCommonsClientAccessor;
 import org.opensearch.neuralsearch.processor.NeuralQueryEnricherProcessor;
 import org.opensearch.neuralsearch.processor.NeuralSparseTwoPhaseProcessor;
-import org.opensearch.neuralsearch.processor.NormalizationProcessor;
 import org.opensearch.neuralsearch.processor.NormalizationProcessorWorkflow;
 import org.opensearch.neuralsearch.processor.SparseEncodingProcessor;
 import org.opensearch.neuralsearch.processor.TextEmbeddingProcessor;
 import org.opensearch.neuralsearch.processor.TextChunkingProcessor;
 import org.opensearch.neuralsearch.processor.TextImageEmbeddingProcessor;
+import org.opensearch.neuralsearch.processor.RRFProcessor;
+import org.opensearch.neuralsearch.processor.NormalizationProcessor;
 import org.opensearch.neuralsearch.processor.combination.ScoreCombinationFactory;
 import org.opensearch.neuralsearch.processor.combination.ScoreCombiner;
 import org.opensearch.neuralsearch.processor.factory.TextChunkingProcessorFactory;
-import org.opensearch.neuralsearch.processor.factory.NormalizationProcessorFactory;
 import org.opensearch.neuralsearch.processor.factory.RerankProcessorFactory;
 import org.opensearch.neuralsearch.processor.factory.SparseEncodingProcessorFactory;
 import org.opensearch.neuralsearch.processor.factory.TextEmbeddingProcessorFactory;
 import org.opensearch.neuralsearch.processor.factory.TextImageEmbeddingProcessorFactory;
+import org.opensearch.neuralsearch.processor.factory.RRFProcessorFactory;
+import org.opensearch.neuralsearch.processor.factory.NormalizationProcessorFactory;
 import org.opensearch.neuralsearch.processor.normalization.ScoreNormalizationFactory;
 import org.opensearch.neuralsearch.processor.normalization.ScoreNormalizer;
 import org.opensearch.neuralsearch.processor.rerank.RerankProcessor;
@@ -154,7 +156,9 @@ public Map<String, org.opensearch.search.pipeline.Processor.Factory<SearchPhaseR
     ) {
         return Map.of(
             NormalizationProcessor.TYPE,
-            new NormalizationProcessorFactory(normalizationProcessorWorkflow, scoreNormalizationFactory, scoreCombinationFactory)
+            new NormalizationProcessorFactory(normalizationProcessorWorkflow, scoreNormalizationFactory, scoreCombinationFactory),
+            RRFProcessor.TYPE,
+            new RRFProcessorFactory(normalizationProcessorWorkflow, scoreNormalizationFactory, scoreCombinationFactory)
         );
     }
 

src/main/java/org/opensearch/neuralsearch/processor/NormalizationExecuteDTO.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.neuralsearch.processor;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NonNull;
+import org.opensearch.neuralsearch.processor.combination.ScoreCombinationTechnique;
+import org.opensearch.neuralsearch.processor.normalization.ScoreNormalizationTechnique;
+import org.opensearch.search.fetch.FetchSearchResult;
+import org.opensearch.search.query.QuerySearchResult;
+
+import java.util.List;
+import java.util.Optional;
+
+/**
+ * DTO object to hold data required for score normalization passed to execute() function
+ * in NormalizationProcessorWorkflow. Field rankConstant
+ * used in RRF but will be null for other normalization techniques
+ */
+@AllArgsConstructor
+@Builder
+@Getter
+public class NormalizationExecuteDTO {
+    @NonNull
+    private List<QuerySearchResult> querySearchResults;
+    @NonNull
+    private Optional<FetchSearchResult> fetchSearchResultOptional;
+    @NonNull
+    private ScoreNormalizationTechnique normalizationTechnique;
+    @NonNull
+    private ScoreCombinationTechnique combinationTechnique;
+}

src/main/java/org/opensearch/neuralsearch/processor/NormalizationProcessor.java

@@ -58,7 +58,15 @@ public <Result extends SearchPhaseResult> void process(
         }
         List<QuerySearchResult> querySearchResults = getQueryPhaseSearchResults(searchPhaseResult);
         Optional<FetchSearchResult> fetchSearchResult = getFetchSearchResults(searchPhaseResult);
-        normalizationWorkflow.execute(querySearchResults, fetchSearchResult, normalizationTechnique, combinationTechnique);
+        // Builds data transfer object to pass into execute, DTO has nullable field for rankConstant which
+        // is only used for RRF technique
+        NormalizationExecuteDTO normalizationExecuteDTO = NormalizationExecuteDTO.builder()
+            .querySearchResults(querySearchResults)
+            .fetchSearchResultOptional(fetchSearchResult)
+            .normalizationTechnique(normalizationTechnique)
+            .combinationTechnique(combinationTechnique)
+            .build();
+        normalizationWorkflow.execute(normalizationExecuteDTO);
     }
 
     @Override

src/main/java/org/opensearch/neuralsearch/processor/NormalizationProcessorWorkflow.java

@@ -47,26 +47,31 @@ public class NormalizationProcessorWorkflow {
 
     /**
      * Start execution of this workflow
-     * @param querySearchResults input data with QuerySearchResult from multiple shards
-     * @param normalizationTechnique technique for score normalization
-     * @param combinationTechnique technique for score combination
+     * @param normalizationExecuteDTO contains querySearchResults input data with QuerySearchResult
+     * from multiple shards, fetchSearchResultOptional, normalizationTechnique technique for score normalization
+     *  combinationTechnique technique for score combination, and nullable rankConstant only used in RRF technique
      */
-    public void execute(
-        final List<QuerySearchResult> querySearchResults,
-        final Optional<FetchSearchResult> fetchSearchResultOptional,
-        final ScoreNormalizationTechnique normalizationTechnique,
-        final ScoreCombinationTechnique combinationTechnique
-    ) {
+    public void execute(final NormalizationExecuteDTO normalizationExecuteDTO) {
+        final List<QuerySearchResult> querySearchResults = normalizationExecuteDTO.getQuerySearchResults();
+        final Optional<FetchSearchResult> fetchSearchResultOptional = normalizationExecuteDTO.getFetchSearchResultOptional();
+        final ScoreNormalizationTechnique normalizationTechnique = normalizationExecuteDTO.getNormalizationTechnique();
+        final ScoreCombinationTechnique combinationTechnique = normalizationExecuteDTO.getCombinationTechnique();
         // save original state
         List<Integer> unprocessedDocIds = unprocessedDocIds(querySearchResults);
 
         // pre-process data
         log.debug("Pre-process query results");
         List<CompoundTopDocs> queryTopDocs = getQueryTopDocs(querySearchResults);
 
+        // Data transfer object for score normalization used to pass nullable rankConstant which is only used in RRF
+        NormalizeScoresDTO normalizeScoresDTO = NormalizeScoresDTO.builder()
+            .queryTopDocs(queryTopDocs)
+            .normalizationTechnique(normalizationTechnique)
+            .build();
+
         // normalize
         log.debug("Do score normalization");
-        scoreNormalizer.normalizeScores(queryTopDocs, normalizationTechnique);
+        scoreNormalizer.normalizeScores(normalizeScoresDTO);
 
         CombineScoresDto combineScoresDTO = CombineScoresDto.builder()
             .queryTopDocs(queryTopDocs)

src/main/java/org/opensearch/neuralsearch/processor/NormalizeScoresDTO.java

@@ -0,0 +1,26 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.neuralsearch.processor;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NonNull;
+import org.opensearch.neuralsearch.processor.normalization.ScoreNormalizationTechnique;
+
+import java.util.List;
+
+/**
+ * DTO object to hold data required for score normalization.
+ */
+@AllArgsConstructor
+@Builder
+@Getter
+public class NormalizeScoresDTO {
+    @NonNull
+    private List<CompoundTopDocs> queryTopDocs;
+    @NonNull
+    private ScoreNormalizationTechnique normalizationTechnique;
+}

src/main/java/org/opensearch/neuralsearch/processor/RRFProcessor.java

@@ -0,0 +1,139 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.neuralsearch.processor;
+
+import static org.opensearch.neuralsearch.search.util.HybridSearchResultFormatUtil.isHybridQueryStartStopElement;
+
+import java.util.stream.Collectors;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+import lombok.Getter;
+import org.opensearch.neuralsearch.processor.combination.ScoreCombinationTechnique;
+import org.opensearch.neuralsearch.processor.normalization.ScoreNormalizationTechnique;
+import org.opensearch.search.fetch.FetchSearchResult;
+import org.opensearch.search.query.QuerySearchResult;
+
+import org.opensearch.action.search.QueryPhaseResultConsumer;
+import org.opensearch.action.search.SearchPhaseContext;
+import org.opensearch.action.search.SearchPhaseName;
+import org.opensearch.action.search.SearchPhaseResults;
+import org.opensearch.search.SearchPhaseResult;
+import org.opensearch.search.internal.SearchContext;
+import org.opensearch.search.pipeline.SearchPhaseResultsProcessor;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+
+/**
+ * Processor for implementing reciprocal rank fusion technique on post
+ * query search results. Updates query results with
+ * normalized and combined scores for next phase (typically it's FETCH)
+ * by using ranks from individual subqueries to calculate 'normalized'
+ * scores before combining results from subqueries into final results
+ */
+@Log4j2
+@AllArgsConstructor
+public class RRFProcessor implements SearchPhaseResultsProcessor {
+    public static final String TYPE = "score-ranker-processor";
+
+    @Getter
+    private final String tag;
+    @Getter
+    private final String description;
+    private final ScoreNormalizationTechnique normalizationTechnique;
+    private final ScoreCombinationTechnique combinationTechnique;
+    private final NormalizationProcessorWorkflow normalizationWorkflow;
+
+    /**
+     * Method abstracts functional aspect of score normalization and score combination. Exact methods for each processing stage
+     * are set as part of class constructor
+     * @param searchPhaseResult {@link SearchPhaseResults} DTO that has query search results. Results will be mutated as part of this method execution
+     * @param searchPhaseContext {@link SearchContext}
+     */
+    @Override
+    public <Result extends SearchPhaseResult> void process(
+        final SearchPhaseResults<Result> searchPhaseResult,
+        final SearchPhaseContext searchPhaseContext
+    ) {
+        if (shouldSkipProcessor(searchPhaseResult)) {
+            log.debug("Query results are not compatible with RRF processor");
+            return;
+        }
+        List<QuerySearchResult> querySearchResults = getQueryPhaseSearchResults(searchPhaseResult);
+        Optional<FetchSearchResult> fetchSearchResult = getFetchSearchResults(searchPhaseResult);
+
+        // make data transfer object to pass in, execute will get object with 4 or 5 fields, depending
+        // on coming from NormalizationProcessor or RRFProcessor
+        NormalizationExecuteDTO normalizationExecuteDTO = NormalizationExecuteDTO.builder()
+            .querySearchResults(querySearchResults)
+            .fetchSearchResultOptional(fetchSearchResult)
+            .normalizationTechnique(normalizationTechnique)
+            .combinationTechnique(combinationTechnique)
+            .build();
+        normalizationWorkflow.execute(normalizationExecuteDTO);
+    }
+
+    @Override
+    public SearchPhaseName getBeforePhase() {
+        return SearchPhaseName.QUERY;
+    }
+
+    @Override
+    public SearchPhaseName getAfterPhase() {
+        return SearchPhaseName.FETCH;
+    }
+
+    @Override
+    public String getType() {
+        return TYPE;
+    }
+
+    @Override
+    public boolean isIgnoreFailure() {
+        return false;
+    }
+
+    private <Result extends SearchPhaseResult> boolean shouldSkipProcessor(SearchPhaseResults<Result> searchPhaseResult) {
+        if (Objects.isNull(searchPhaseResult) || !(searchPhaseResult instanceof QueryPhaseResultConsumer queryPhaseResultConsumer)) {
+            return true;
+        }
+
+        return queryPhaseResultConsumer.getAtomicArray().asList().stream().filter(Objects::nonNull).noneMatch(this::isHybridQuery);
+    }
+
+    /**
+     * Return true if results are from hybrid query.
+     * @param searchPhaseResult
+     * @return true if results are from hybrid query
+     */
+    private boolean isHybridQuery(final SearchPhaseResult searchPhaseResult) {
+        // check for delimiter at the end of the score docs.
+        return Objects.nonNull(searchPhaseResult.queryResult())
+            && Objects.nonNull(searchPhaseResult.queryResult().topDocs())
+            && Objects.nonNull(searchPhaseResult.queryResult().topDocs().topDocs.scoreDocs)
+            && searchPhaseResult.queryResult().topDocs().topDocs.scoreDocs.length > 0
+            && isHybridQueryStartStopElement(searchPhaseResult.queryResult().topDocs().topDocs.scoreDocs[0]);
+    }
+
+    private <Result extends SearchPhaseResult> List<QuerySearchResult> getQueryPhaseSearchResults(
+        final SearchPhaseResults<Result> results
+    ) {
+        return results.getAtomicArray()
+            .asList()
+            .stream()
+            .map(result -> result == null ? null : result.queryResult())
+            .collect(Collectors.toList());
+    }
+
+    private <Result extends SearchPhaseResult> Optional<FetchSearchResult> getFetchSearchResults(
+        final SearchPhaseResults<Result> searchPhaseResults
+    ) {
+        Optional<Result> optionalFirstSearchPhaseResult = searchPhaseResults.getAtomicArray().asList().stream().findFirst();
+        return optionalFirstSearchPhaseResult.map(SearchPhaseResult::fetchResult);
+    }
+}

src/main/java/org/opensearch/neuralsearch/processor/combination/RRFScoreCombinationTechnique.java

@@ -0,0 +1,32 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.opensearch.neuralsearch.processor.combination;
+
+import lombok.ToString;
+import lombok.extern.log4j.Log4j2;
+
+import java.util.Map;
+
+@Log4j2
+/**
+ * Abstracts combination of scores based on reciprocal rank fusion algorithm
+ */
+@ToString(onlyExplicitlyIncluded = true)
+public class RRFScoreCombinationTechnique implements ScoreCombinationTechnique {
+    @ToString.Include
+    public static final String TECHNIQUE_NAME = "rrf";
+
+    // Not currently using weights for RRF, no need to modify or verify these params
+    public RRFScoreCombinationTechnique(final Map<String, Object> params, final ScoreCombinationUtil combinationUtil) {}
+
+    @Override
+    public float combine(final float[] scores) {
+        float sumScores = 0.0f;
+        for (float score : scores) {
+            sumScores += score;
+        }
+        return sumScores;
+    }
+}

src/main/java/org/opensearch/neuralsearch/processor/combination/ScoreCombinationFactory.java

@@ -25,7 +25,9 @@ public class ScoreCombinationFactory {
         HarmonicMeanScoreCombinationTechnique.TECHNIQUE_NAME,
         params -> new HarmonicMeanScoreCombinationTechnique(params, scoreCombinationUtil),
         GeometricMeanScoreCombinationTechnique.TECHNIQUE_NAME,
-        params -> new GeometricMeanScoreCombinationTechnique(params, scoreCombinationUtil)
+        params -> new GeometricMeanScoreCombinationTechnique(params, scoreCombinationUtil),
+        RRFScoreCombinationTechnique.TECHNIQUE_NAME,
+        params -> new RRFScoreCombinationTechnique(params, scoreCombinationUtil)
     );
 
     /**

src/main/java/org/opensearch/neuralsearch/processor/combination/ScoreCombinationUtil.java

@@ -25,7 +25,7 @@
 @Log4j2
 class ScoreCombinationUtil {
     private static final String PARAM_NAME_WEIGHTS = "weights";
-    private static final float DELTA_FOR_SCORE_ASSERTION = 0.01f;
+    private static final float DELTA_FOR_WEIGHTS_ASSERTION = 0.01f;
 
     /**
      * Get collection of weights based on user provided config
@@ -117,7 +117,7 @@ protected void validateIfWeightsMatchScores(final float[] scores, final List<Flo
      * @param weightsList
      */
     private void validateWeights(final List<Float> weightsList) {
-        boolean isOutOfRange = weightsList.stream().anyMatch(weight -> !Range.between(0.0f, 1.0f).contains(weight));
+        boolean isOutOfRange = weightsList.stream().anyMatch(weight -> !Range.of(0.0f, 1.0f).contains(weight));
         if (isOutOfRange) {
             throw new IllegalArgumentException(
                 String.format(
@@ -128,7 +128,7 @@ private void validateWeights(final List<Float> weightsList) {
             );
         }
         float sumOfWeights = weightsList.stream().reduce(0.0f, Float::sum);
-        if (!DoubleMath.fuzzyEquals(1.0f, sumOfWeights, DELTA_FOR_SCORE_ASSERTION)) {
+        if (!DoubleMath.fuzzyEquals(1.0f, sumOfWeights, DELTA_FOR_WEIGHTS_ASSERTION)) {
             throw new IllegalArgumentException(
                 String.format(
                     Locale.ROOT,