This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 976396b39f [vector] Add MRR hybrid search ranker (#8351)
976396b39f is described below
commit 976396b39f3704f298c8c7c27a77f6978925e10d
Author: Jingsong Lee <[email protected]>
AuthorDate: Thu Jun 25 20:44:44 2026 +0800
[vector] Add MRR hybrid search ranker (#8351)
Adds `mrr` as a built-in hybrid search ranker. MRR uses each route
result order and sums `weight / rank` for matching rows, then returns
the final top-k rows.
---
.../global-index/hybrid-search.mdx | 57 +++++++++++++++++++++-
.../paimon/globalindex/HybridSearchRanker.java | 36 +++++++++++++-
.../paimon/globalindex/HybridSearchRankerTest.java | 23 +++++++++
.../pypaimon/table/source/hybrid_search_builder.py | 28 ++++++++---
.../pypaimon/tests/hybrid_search_ranker_test.py | 18 ++++++-
.../plans/logical/PaimonTableValuedFunctions.scala | 3 +-
.../plans/logical/VectorSearchQueryTest.scala | 22 +++++++++
7 files changed, 173 insertions(+), 14 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
index e4923bdec8..bdc44314dc 100644
--- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
+++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
@@ -60,8 +60,8 @@ CALL sys.create_global_index(
For Spark SQL, use the `hybrid_search(table_name, vector_routes,
full_text_routes, limit[, ranker])`
table-valued function. The fourth argument is the final number of ranked
results to return.
-The optional fifth argument selects the ranker. Supported rankers are `rrf` and
-`weighted_score`; the default is `rrf`.
+The optional fifth argument selects the ranker. Supported rankers are `rrf`,
`weighted_score`,
+and `mrr`; the default is `rrf`.
Rankers combine route scores differently:
@@ -69,6 +69,7 @@ Rankers combine route scores differently:
|---|---|---|
| `rrf` | Combining routes with different score scales | Reciprocal rank
fusion uses each route's rank order, so vector and full-text scores do not need
to be normalized. |
| `weighted_score` | Weighting routes by normalized score, not just rank |
Min-max normalizes each route's scores to `[0, 1]`, then sums them weighted by
route `weight`, so weights (not raw score magnitude) control each route's
influence. The exposed `__paimon_search_score` is therefore a per-query
relative value in `[0, sum of weights]`, not a raw similarity or BM25 score. |
+| `mrr` | Emphasizing top-ranked hits from each route | Weighted
reciprocal-rank fusion sums `weight / rank` for each row returned by a route,
where rank starts at 1. |
The second argument is an array of vector route configs created by
`named_struct`:
@@ -148,6 +149,24 @@ FROM hybrid_search(
'weighted_score');
```
+Use `mrr` when you want weighted reciprocal-rank fusion:
+
+```sql
+SELECT id, __paimon_search_score
+FROM hybrid_search(
+ 'my_table',
+ array(
+ named_struct(
+ 'field', 'title_embedding',
+ 'query_vector', array(1.0f, 0.0f, 0.0f),
+ 'limit', 50,
+ 'weight', 2.0f,
+ 'options', map('ivf.nprobe', '32'))),
+ array(),
+ 10,
+ 'mrr');
+```
+
Spark SQL adds `__paimon_search_score` to expose the ranked score.
</TabItem>
@@ -205,6 +224,22 @@ GlobalIndexResult vectorOnlyResult =
.executeLocal();
```
+Use `withRanker("mrr")` for weighted reciprocal-rank fusion:
+
+```java
+GlobalIndexResult mrrResult =
+ table.newHybridSearchBuilder()
+ .addVectorRoute(
+ "title_embedding",
+ new float[] {1.0f, 0.0f, 0.0f},
+ 50,
+ 1.0f,
+ java.util.Collections.singletonMap("ivf.nprobe", "32"))
+ .withLimit(10)
+ .withRanker("mrr")
+ .executeLocal();
+```
+
For Java, use `Table.newHybridSearchBuilder()` to configure routes, final
limit, and ranker
directly. `VectorSearch` and `HybridSearch` are internal pushdown
representations.
@@ -267,6 +302,24 @@ result = (
)
```
+Use `with_ranker("mrr")` for weighted reciprocal-rank fusion:
+
+```python
+result = (
+ table.new_hybrid_search_builder()
+ .add_vector_route(
+ "title_embedding",
+ [1.0, 0.0, 0.0],
+ limit=50,
+ weight=1.0,
+ options={"ivf.nprobe": "32"},
+ )
+ .with_limit(10)
+ .with_ranker("mrr")
+ .execute_local()
+)
+```
+
</TabItem>
</Tabs>
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/HybridSearchRanker.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/HybridSearchRanker.java
index 675e4b1a38..e3c910aa4b 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/HybridSearchRanker.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/HybridSearchRanker.java
@@ -33,6 +33,7 @@ public class HybridSearchRanker {
public static final String RRF_RANKER = "rrf";
public static final String WEIGHTED_SCORE_RANKER = "weighted_score";
+ public static final String MRR_RANKER = "mrr";
private static final float RRF_K = 60.0f;
@@ -49,8 +50,11 @@ public class HybridSearchRanker {
public static ScoredGlobalIndexResult rank(
String ranker, List<WeightedResult> results, int limit) {
- if (WEIGHTED_SCORE_RANKER.equals(normalizeRanker(ranker))) {
+ String normalized = normalizeRanker(ranker);
+ if (WEIGHTED_SCORE_RANKER.equals(normalized)) {
return weightedScore(results, limit);
+ } else if (MRR_RANKER.equals(normalized)) {
+ return mrr(results, limit);
}
return rrf(results, limit);
}
@@ -60,7 +64,9 @@ public class HybridSearchRanker {
return RRF_RANKER;
}
String normalized = ranker.trim().toLowerCase();
- if (!RRF_RANKER.equals(normalized) &&
!WEIGHTED_SCORE_RANKER.equals(normalized)) {
+ if (!RRF_RANKER.equals(normalized)
+ && !WEIGHTED_SCORE_RANKER.equals(normalized)
+ && !MRR_RANKER.equals(normalized)) {
throw new IllegalArgumentException("Unsupported hybrid ranker: " +
ranker);
}
return normalized;
@@ -134,6 +140,32 @@ public class HybridSearchRanker {
return topK(scores, limit);
}
+ public static ScoredGlobalIndexResult mrr(
+ List<ScoredGlobalIndexResult> results, float[] weights, int limit)
{
+ List<WeightedResult> weightedResults = new ArrayList<>(results.size());
+ for (int i = 0; i < results.size(); i++) {
+ weightedResults.add(new WeightedResult(results.get(i),
weightAt(weights, i)));
+ }
+ return mrr(weightedResults, limit);
+ }
+
+ public static ScoredGlobalIndexResult mrr(List<WeightedResult> results,
int limit) {
+ Map<Long, Float> scores = new HashMap<>();
+ for (WeightedResult weightedResult : results) {
+ ScoredGlobalIndexResult result = weightedResult.result();
+ float weight = weightedResult.weight();
+ List<Long> ranked = rankedRowIds(result);
+ for (int rank = 0; rank < ranked.size(); rank++) {
+ Long rowId = ranked.get(rank);
+ float contribution = weight / (rank + 1.0f);
+ scores.compute(
+ rowId,
+ (k, oldScore) -> oldScore == null ? contribution :
oldScore + contribution);
+ }
+ }
+ return topK(scores, limit);
+ }
+
private static List<Long> rankedRowIds(ScoredGlobalIndexResult result) {
List<Long> rowIds = new ArrayList<>();
for (long rowId : result.results()) {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/HybridSearchRankerTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/HybridSearchRankerTest.java
index 0e3ad8cac5..a1c286f8d4 100644
---
a/paimon-common/src/test/java/org/apache/paimon/globalindex/HybridSearchRankerTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/HybridSearchRankerTest.java
@@ -147,6 +147,29 @@ public class HybridSearchRankerTest {
assertThat(ranked.scoreGetter().score(2L)).isCloseTo(2.0f,
within(0.000001f));
}
+ @Test
+ public void testMrrFavorsRowsWithStrongRanksAcrossRoutes() {
+ ScoredGlobalIndexResult first = result(new long[] {1, 2}, new float[]
{0.9f, 0.8f});
+ ScoredGlobalIndexResult second = result(new long[] {2, 3}, new float[]
{0.7f, 0.6f});
+
+ assertThat(HybridSearchRanker.normalizeRanker("mrr"))
+ .isEqualTo(HybridSearchRanker.MRR_RANKER);
+
+ ScoredGlobalIndexResult ranked =
+ HybridSearchRanker.rank(
+ HybridSearchRanker.MRR_RANKER,
+ Arrays.asList(
+ new HybridSearchRanker.WeightedResult(first,
1.0f),
+ new HybridSearchRanker.WeightedResult(second,
2.0f)),
+ 2);
+
+ assertThat(ranked.results()).contains(1L, 2L);
+ assertThat(ranked.results()).doesNotContain(3L);
+ assertThat(ranked.scoreGetter().score(2L)).isCloseTo(2.5f,
within(0.000001f));
+ assertThat(ranked.scoreGetter().score(1L)).isCloseTo(1.0f,
within(0.000001f));
+
assertThat(ranked.scoreGetter().score(2L)).isGreaterThan(ranked.scoreGetter().score(1L));
+ }
+
@Test
public void testRejectNonFiniteWeights() {
ScoredGlobalIndexResult result = result(new long[] {1}, new float[]
{1.0f});
diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
index 8edeeac251..b1875c43c5 100644
--- a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
+++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
@@ -33,6 +33,7 @@ from pypaimon.globalindex.vector_search_result import (
RRF_RANKER = "rrf"
WEIGHTED_SCORE_RANKER = "weighted_score"
+MRR_RANKER = "mrr"
_RRF_K = 60.0
@@ -46,7 +47,7 @@ def _normalize_ranker(ranker: Optional[str]) -> str:
if ranker is None or not ranker.strip():
return RRF_RANKER
normalized = ranker.strip().lower()
- if normalized not in (RRF_RANKER, WEIGHTED_SCORE_RANKER):
+ if normalized not in (RRF_RANKER, WEIGHTED_SCORE_RANKER, MRR_RANKER):
raise ValueError("Unsupported hybrid ranker: %s" % ranker)
return normalized
@@ -322,6 +323,8 @@ class HybridSearchBuilderImpl(HybridSearchBuilder):
]
if self._ranker == WEIGHTED_SCORE_RANKER:
return self._weighted_score(non_empty)
+ if self._ranker == MRR_RANKER:
+ return self._mrr(non_empty)
return self._rrf(non_empty)
def _validate_search(self):
@@ -364,17 +367,19 @@ class HybridSearchBuilderImpl(HybridSearchBuilder):
def _rrf(self, route_results):
scores = {}
for route_result in route_results:
- result = route_result.result
- score_getter = result.score_getter()
- row_ids = sorted(
- result.results(),
- key=lambda row_id: (
- -(score_getter(row_id) or 0.0), row_id))
- for rank, row_id in enumerate(row_ids):
+ for rank, row_id in
enumerate(_ranked_row_ids(route_result.result)):
contribution = route_result.route.weight / (_RRF_K + rank +
1.0)
scores[row_id] = scores.get(row_id, 0.0) + contribution
return _top_k(scores, self._limit)
+ def _mrr(self, route_results):
+ scores = {}
+ for route_result in route_results:
+ for rank, row_id in
enumerate(_ranked_row_ids(route_result.result)):
+ contribution = route_result.route.weight / (rank + 1.0)
+ scores[row_id] = scores.get(row_id, 0.0) + contribution
+ return _top_k(scores, self._limit)
+
def _weighted_score(self, route_results):
scores = {}
for route_result in route_results:
@@ -456,6 +461,13 @@ class HybridSearchBuilderImpl(HybridSearchBuilder):
return predicate.new_index(name_to_idx[predicate.field])
+def _ranked_row_ids(result):
+ score_getter = result.score_getter()
+ return sorted(
+ result.results(),
+ key=lambda row_id: (-(score_getter(row_id) or 0.0), row_id))
+
+
def _top_k(scores, limit):
if not scores:
return ScoredGlobalIndexResult.create_empty()
diff --git a/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
b/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
index 47dc85a428..4647fa44d0 100644
--- a/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
+++ b/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
@@ -24,7 +24,8 @@ import unittest
from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
from pypaimon.table.source.hybrid_search_builder import (
- HybridSearchBuilderImpl, HybridSearchRoute, HybridSearchRouteResult)
+ HybridSearchBuilderImpl, HybridSearchRoute, HybridSearchRouteResult,
+ MRR_RANKER)
def _route_result(weight, id_to_scores):
@@ -82,6 +83,21 @@ class HybridSearchRankerTest(unittest.TestCase):
# Rank-based fusion respects the 5x vector weight: rowId 1 wins.
self.assertGreater(getter(1), getter(2))
+ def test_mrr_favors_rows_with_strong_ranks_across_routes(self):
+ first = _route_result(1.0, {1: 0.9, 2: 0.8})
+ second = _route_result(2.0, {2: 0.7, 3: 0.6})
+ builder = _builder(2).with_ranker(MRR_RANKER)
+ builder._routes = [first.route, second.route]
+
+ ranked = builder.rank([first, second])
+ getter = ranked.score_getter()
+
+ self.assertEqual({1, 2}, set(ranked.results()))
+ self.assertNotIn(3, set(ranked.results()))
+ self.assertAlmostEqual(getter(2), 2.5, places=6)
+ self.assertAlmostEqual(getter(1), 1.0, places=6)
+ self.assertGreater(getter(2), getter(1))
+
def test_route_rejects_non_finite_weight(self):
for weight in (math.nan, math.inf, -math.inf):
with self.subTest(weight=weight):
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
index ad3675f369..ee3a930386 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/plans/logical/PaimonTableValuedFunctions.scala
@@ -644,7 +644,8 @@ case class VectorSearchQuery(override val args:
Seq[Expression])
* - vector_routes: route config array with field, query_vector, limit,
weight, and options fields
* - full_text_routes: route config array with query, limit, weight, and
empty options fields
* - limit: the final number of ranked top results to return
- * - ranker: optional ranker for combining results from multiple routes
+ * - ranker: optional ranker for combining results from multiple routes:
rrf, weighted_score, or
+ * mrr
*/
case class HybridSearchQuery(override val args: Seq[Expression])
extends PaimonTableValueFunction(HYBRID_SEARCH) {
diff --git
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/catalyst/plans/logical/VectorSearchQueryTest.scala
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/catalyst/plans/logical/VectorSearchQueryTest.scala
index 08bc308182..6ec37ac159 100644
---
a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/catalyst/plans/logical/VectorSearchQueryTest.scala
+++
b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/catalyst/plans/logical/VectorSearchQueryTest.scala
@@ -101,6 +101,28 @@ class VectorSearchQueryTest extends AnyFunSuite {
assert(search.routes().get(0).options().isEmpty)
}
+ test("create hybrid search with mrr ranker") {
+ val search = HybridSearchQuery(Seq.empty).createHybridSearch(
+ innerTable,
+ Seq(
+ CreateArray(
+ Seq(
+ CreateNamedStruct(
+ Seq(
+ Literal("vector_column"),
+ Literal("title_vec"),
+ Literal("query_vector"),
+ CreateArray(Seq(Literal(1.0f), Literal(0.0f)))
+ )))),
+ CreateArray(Seq.empty),
+ Literal(7),
+ Literal("mrr")
+ )
+ )
+
+ assert(search.ranker() == "mrr")
+ }
+
test("create hybrid search with full-text route configs") {
val search = HybridSearchQuery(Seq.empty).createHybridSearch(
innerTable,
