[Cosmos] Use deterministic ordinal ranks for hybrid-search RRF

sdk/cosmos/azure-cosmos-tests/src/test/resources/test-resource-hygiene-baseline.properties

@@ -40,7 +40,6 @@ com/azure/cosmos/rx/FullTextIndexTest.java=1
 com/azure/cosmos/rx/GatewayReadConsistencyStrategyE2ETest.java=1
 com/azure/cosmos/rx/GatewayReadConsistencyStrategySpyWireTest.java=1
 com/azure/cosmos/rx/GlobalSecondaryIndexContainerCrudTest.java=1
-com/azure/cosmos/rx/HybridSearchQueryTest.java=1
 com/azure/cosmos/rx/MultiMasterConflictResolutionTest.java=1
 com/azure/cosmos/rx/NonStreamingOrderByQueryVectorSearchTest.java=1
 com/azure/cosmos/rx/OfferQueryTest.java=1

sdk/cosmos/azure-cosmos/CHANGELOG.md

@@ -7,6 +7,7 @@
 #### Breaking Changes
 
 #### Bugs Fixed
+* Hybrid search now uses deterministic ordinal reciprocal-rank fusion: component scores `10, 5, 5, 5, 0` receive ranks `1, 2, 3, 4, 5`. Set `AZURE_COSMOS_HYBRID_SEARCH_USE_COMPETITION_RANKING=true` to use competition ranks `1, 2, 2, 2, 5` instead. Component ties use numeric document RID in the query-plan sort direction; fused-score ties use numeric document RID descending.
 * Fixed the Direct mode (RNTBD) encoding of the `x-ms-cosmos-workload-id` header, which was sent with the wrong token ID (`0x00DC`) and type (`Byte`) instead of `0x00E7` and `UShort`. - See [PR 50673](https://github.com/Azure/azure-sdk-for-java/pull/50673)
 
 #### Other Changes

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/Configs.java

@@ -269,6 +269,11 @@ public class Configs {
     private static final String MAX_ITEM_COUNT_FOR_HYBRID_SEARCH_VARIABLE = "COSMOS_MAX_ITEM_SIZE_FOR_HYBRID_SEARCH";
     public static final int DEFAULT_MAX_ITEM_COUNT_FOR_HYBRID_SEARCH = 1000;
 
+    private static final String HYBRID_SEARCH_USE_COMPETITION_RANKING =
+        "COSMOS.HYBRID_SEARCH_USE_COMPETITION_RANKING";
+    private static final String HYBRID_SEARCH_USE_COMPETITION_RANKING_VARIABLE =
+        "AZURE_COSMOS_HYBRID_SEARCH_USE_COMPETITION_RANKING";
+
     private static final String AZURE_COSMOS_DISABLE_NON_STREAMING_ORDER_BY = "COSMOS.AZURE_COSMOS_DISABLE_NON_STREAMING_ORDER_BY";
 
     private static final boolean DEFAULT_AZURE_COSMOS_DISABLE_NON_STREAMING_ORDER_BY = false;
@@ -1248,6 +1253,11 @@ public static int getMaxItemCountForHybridSearchSearch() {
         return DEFAULT_MAX_ITEM_COUNT_FOR_HYBRID_SEARCH;
     }
 
+    public static boolean isHybridSearchCompetitionRankingEnabled() {
+        return Boolean.parseBoolean(System.getProperty(HYBRID_SEARCH_USE_COMPETITION_RANKING,
+            firstNonNull(emptyToNull(System.getenv(HYBRID_SEARCH_USE_COMPETITION_RANKING_VARIABLE)), "false")));
+    }
+
     public static boolean getAzureCosmosNonStreamingOrderByDisabled() {
         if(logger.isTraceEnabled()) {
             logger.trace(

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/query/HybridSearchDocumentQueryExecutionContext.java

@@ -7,6 +7,7 @@
 import com.azure.cosmos.CosmosDiagnostics;
 import com.azure.cosmos.CosmosException;
 import com.azure.cosmos.implementation.ClientSideRequestStatistics;
+import com.azure.cosmos.implementation.Configs;
 import com.azure.cosmos.implementation.DiagnosticsClientContext;
 import com.azure.cosmos.implementation.Document;
 import com.azure.cosmos.implementation.DocumentClientRetryPolicy;
@@ -78,6 +79,7 @@ private static ImplementationBridgeHelpers.FeedResponseHelper.FeedResponseAccess
     private Flux<HybridSearchQueryResult<Document>> hybridObservable;
     private Mono<GlobalFullTextSearchQueryStatistics> aggregatedGlobalStatistics;
     private final SchedulingStopwatch hybridSearchSchedulingStopwatch;
+    private final boolean useCompetitionRanking;
 
     protected HybridSearchDocumentQueryExecutionContext(
         DiagnosticsClientContext diagnosticsClientContext,
@@ -97,6 +99,7 @@ protected HybridSearchDocumentQueryExecutionContext(
         this.tracker = new RequestChargeTracker();
         this.queryMetricMap = new ConcurrentHashMap<>();
         this.clientSideRequestStatistics = ConcurrentHashMap.newKeySet();
+        this.useCompetitionRanking = Configs.isHybridSearchCompetitionRankingEnabled();
 
         // Initialize the shared stopwatch for hybrid search timing
         this.hybridSearchSchedulingStopwatch = new HybridSearchCumulativeSchedulingStopWatch();
@@ -196,15 +199,15 @@ private Flux<HybridSearchQueryResult<Document>> hybridSearch(List<FeedRangeEpkIm
         // Run component queries, and retrieve component query results.
         Flux<Document> componentQueryResults = getComponentQueryResults(targetFeedRanges, initialPageSize, collection, rewrittenQueryInfos);
 
-        // Coalesce the results on unique _rids, and sort it based on the _rid
+        // Deduplicate by encoded RID, then order by numeric document RID.
         Mono<List<HybridSearchQueryResult<Document>>> coalescedAndSortedResults =
             coalesceAndSortResults(componentQueryResults).cache();
 
         // Compose component scores matrix, where each tuple is (score, index)
         Mono<List<List<ScoreTuple>>> componentScoresList = retrieveComponentScores(coalescedAndSortedResults, componentWeights);
 
-        // Compute Ranks
-        Mono<List<List<Integer>>> ranks = computeRanks(componentScoresList);
+        // Assign ordinal ranks by default, or competition ranks for equal component scores.
+        Mono<List<List<Integer>>> ranks = computeRanks(componentScoresList, this.useCompetitionRanking);
 
         // Compute the RRF scores
         return computeRRFScores(ranks, coalescedAndSortedResults, componentWeights)
@@ -332,9 +335,10 @@ public Flux<FeedResponse<Document>> apply(Flux<HybridSearchQueryResult<Document>
         }
     }
 
-    private static Flux<HybridSearchQueryResult<Document>> computeRRFScores(Mono<List<List<Integer>>> ranks,
-                                                                            Mono<List<HybridSearchQueryResult<Document>>> coalescedAndSortedResults,
-                                                                            List<ComponentWeight> componentWeights) {
+    private static Flux<HybridSearchQueryResult<Document>> computeRRFScores(
+        Mono<List<List<Integer>>> ranks,
+        Mono<List<HybridSearchQueryResult<Document>>> coalescedAndSortedResults,
+        List<ComponentWeight> componentWeights) {
         return ranks.zipWith(coalescedAndSortedResults)
             .map(tuple -> {
                 List<List<Integer>> ranksInternal = tuple.getT1();
@@ -347,13 +351,19 @@ private static Flux<HybridSearchQueryResult<Document>> computeRRFScores(Mono<Lis
                     }
                     results.get(index).setScore(rrfScore);
                 }
-                // Sort on the RRF scores to build the final result
-                results.sort(Comparator.comparing(HybridSearchQueryResult::getScore, Comparator.reverseOrder()));
+
+                // Fused scores and their numeric document RID tie-breaker are always descending.
+                results.sort((x, y) -> {
+                    int scoreComparison = Double.compare(y.getScore(), x.getScore());
+                    return scoreComparison != 0 ? scoreComparison
+                        : Long.compareUnsigned(y.getDocumentRid(), x.getDocumentRid());
+                });
                 return results;
             }).flatMapMany(Flux::fromIterable);
     }
 
-    private static Mono<List<List<Integer>>> computeRanks(Mono<List<List<ScoreTuple>>> componentScoresList) {
+    private static Mono<List<List<Integer>>> computeRanks(
+        Mono<List<List<ScoreTuple>>> componentScoresList, boolean useCompetitionRanking) {
         return componentScoresList.map(componentScores -> {
             // initialize ranks as an N-D list with zeros
             List<List<Integer>> ranksInternal = new ArrayList<>();
@@ -368,9 +378,11 @@ private static Mono<List<List<Integer>>> computeRanks(Mono<List<List<ScoreTuple>
             for (int componentIndex = 0; componentIndex < componentScores.size(); componentIndex++) {
                 int rank = 1; // ranks are 1 based
                 for (int index = 0; index < componentScores.get(componentIndex).size(); index++) {
-                    // Identical scores should have the same rank
-                    if ((index > 0) && (componentScores.get(componentIndex).get(index).getScore() != componentScores.get(componentIndex).get(index - 1).getScore())) {
-                        rank += 1;
+                    // Competition ties share the first position; the next distinct score skips tied positions.
+                    if (!useCompetitionRanking || (index > 0
+                        && componentScores.get(componentIndex).get(index).getScore().doubleValue()
+                            != componentScores.get(componentIndex).get(index - 1).getScore().doubleValue())) {
+                        rank = index + 1;
                     }
                     int rankIndex = componentScores.get(componentIndex).get(index).getIndex();
                     ranksInternal.get(componentIndex).set(rankIndex, rank);
@@ -380,10 +392,12 @@ private static Mono<List<List<Integer>>> computeRanks(Mono<List<List<ScoreTuple>
         });
     }
 
-    private static Mono<List<List<ScoreTuple>>> retrieveComponentScores(Mono<List<HybridSearchQueryResult<Document>>> coalescedAndSortedResults, List<ComponentWeight> componentWeights) {
+    private static Mono<List<List<ScoreTuple>>> retrieveComponentScores(
+        Mono<List<HybridSearchQueryResult<Document>>> coalescedAndSortedResults, List<ComponentWeight> componentWeights) {
         return coalescedAndSortedResults.map(results -> {
             List<List<ScoreTuple>> componentScoresInternal = new ArrayList<>();
-            for (int i = 0; i < results.get(0).getComponentScores().size(); i++) {
+            // The plan defines the dimensions, even when the first row has no component scores.
+            for (int i = 0; i < componentWeights.size(); i++) {
                 componentScoresInternal.add(new ArrayList<>());
             }
             List<Double> undefinedComponentScores = new ArrayList<>();
@@ -401,10 +415,16 @@ private static Mono<List<List<ScoreTuple>>> retrieveComponentScores(Mono<List<Hy
                     componentScoresInternal.get(j).add(scoreTuple);
                 }
             }
+
+            // Component ties follow the same query-plan direction as their scores.
             for (int i = 0; i < componentScoresInternal.size(); i++) {
-                final int componentIndex = i;
-                componentScoresInternal.get(i).sort((x,y) ->
-                    componentWeights.get(componentIndex).getComparator().compare(x.getScore(), y.getScore()));
+                ComponentWeight componentWeight = componentWeights.get(i);
+                componentScoresInternal.get(i).sort((x, y) -> {
+                    int scoreComparison = componentWeight.getComparator().compare(x.getScore(), y.getScore());
+                    return scoreComparison != 0 ? scoreComparison
+                        : componentWeight.comparisonFactor * Long.compareUnsigned(
+                            results.get(x.getIndex()).getDocumentRid(), results.get(y.getIndex()).getDocumentRid());
+                });
             }
             return componentScoresInternal;
         });
@@ -499,7 +519,7 @@ private Flux<QueryInfo> retrieveRewrittenQueryInfos(List<QueryInfo> componentQue
         });
     }
 
-    private Mono<List<HybridSearchQueryResult<Document>>> coalesceAndSortResults(Flux<Document> componentQueryResults) {
+    private static Mono<List<HybridSearchQueryResult<Document>>> coalesceAndSortResults(Flux<Document> componentQueryResults) {
         return componentQueryResults.collectList()
             .map(results -> {
                 Map<String, HybridSearchQueryResult<Document>> uniqueDocuments = new LinkedHashMap<>();
@@ -509,9 +529,9 @@ private Mono<List<HybridSearchQueryResult<Document>>> coalesceAndSortResults(Flu
                     uniqueDocuments.putIfAbsent(rid, result);
                 }
 
-                // Sort by _rid
+                // RID identity remains encoded; only ordering uses the cached document component.
                 List<HybridSearchQueryResult<Document>> coalescedResults = new ArrayList<>(uniqueDocuments.values());
-                coalescedResults.sort(Comparator.comparing(HybridSearchQueryResult::getRid));
+                coalescedResults.sort((x, y) -> Long.compareUnsigned(y.getDocumentRid(), x.getDocumentRid()));
                 return coalescedResults;
             });
     }
@@ -571,7 +591,7 @@ private GlobalFullTextSearchQueryStatistics aggregateStatistics(List<GlobalFullT
         return aggregatedStats;
     }
 
-    private List<ComponentWeight> retrieveComponentWeights(List<Double> componentWeightList, List<QueryInfo> componentQueryInfos) {
+    private static List<ComponentWeight> retrieveComponentWeights(List<Double> componentWeightList, List<QueryInfo> componentQueryInfos) {
         boolean useDefaultComponentWeight = componentWeightList == null || componentWeightList.isEmpty();
         List<ComponentWeight> componentWeights = new ArrayList<>();
         for (int i=0;i<componentQueryInfos.size();i++) {
@@ -586,11 +606,13 @@ private List<ComponentWeight> retrieveComponentWeights(List<Double> componentWei
     private static class ComponentWeight {
         private final Double weight;
         private final Comparator<Double> comparator;
+        private final int comparisonFactor;
 
         public ComponentWeight(Double weight, SortOrder sortOrder) {
             this.weight = weight;
 
-            int comparisonFactor = (sortOrder == SortOrder.Ascending) ? 1 : -1;
+            // Direction comes from the plan, not the (potentially normalized) weight.
+            this.comparisonFactor = (sortOrder == SortOrder.Ascending) ? 1 : -1;
             this.comparator = (x, y) -> comparisonFactor * Double.compare(x, y);
         }
 

sdk/cosmos/azure-cosmos/src/main/java/com/azure/cosmos/implementation/query/hybridsearch/HybridSearchQueryResult.java

@@ -5,6 +5,7 @@
 
 import com.azure.cosmos.implementation.Constants;
 import com.azure.cosmos.implementation.Document;
+import com.azure.cosmos.implementation.ResourceId;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 
@@ -16,11 +17,14 @@ public class HybridSearchQueryResult<T> extends Document {
     public List<Double> componentScores;
     public Document payload;
     public Double score;
+    private final long documentRid;
 
     public HybridSearchQueryResult(String jsonString) {
         super(jsonString);
         // Initialize rid
         this.rid = super.getResourceId();
+        // getDocument exposes the little-endian document component; compare it as unsigned.
+        this.documentRid = ResourceId.parse(this.rid).getDocument();
         // Initialize componentScores
         final Object outerObject = super.get(Constants.Properties.PAYLOAD);
         if (outerObject instanceof ObjectNode) {
@@ -48,6 +52,10 @@ public String getRid() {
         return this.rid;
     }
 
+    public long getDocumentRid() {
+        return this.documentRid;
+    }
+
     public void setRid(String rid) {
         super.setResourceId(rid);
     }