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 c63da64049 [global-index] Normalize hybrid weighted score fusion
(#8294)
c63da64049 is described below
commit c63da640497de96f1b3c51f7fd91b2acf11fa290
Author: QuakeWang <[email protected]>
AuthorDate: Sat Jun 20 16:04:39 2026 +0800
[global-index] Normalize hybrid weighted score fusion (#8294)
`weighted_score` previously fused hybrid search routes by directly
summing `weight * raw_score`.
This is incorrect for heterogeneous routes because vector similarity
scores and full-text/BM25 scores use different scales. A route with
larger numeric scores can dominate the final ranking, making route
weights ineffective.
---
.../global-index/hybrid-search.mdx | 2 +-
.../paimon/globalindex/HybridSearchRanker.java | 19 ++++-
.../paimon/globalindex/HybridSearchRankerTest.java | 54 +++++++++++++-
.../pypaimon/table/source/hybrid_search_builder.py | 25 ++++++-
.../pypaimon/tests/hybrid_search_ranker_test.py | 86 ++++++++++++++++++++++
.../apache/paimon/spark/sql/HybridSearchTest.scala | 69 +++++++++++++++++
6 files changed, 248 insertions(+), 7 deletions(-)
diff --git a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
index bde17d02b6..7c3f8395e3 100644
--- a/docs/docs/multimodal-table/global-index/hybrid-search.mdx
+++ b/docs/docs/multimodal-table/global-index/hybrid-search.mdx
@@ -68,7 +68,7 @@ Rankers combine route scores differently:
| Ranker | Best For | Description |
|---|---|---|
| `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` | Combining routes with comparable score scales | Adds
weighted scores from each route. Tune route weights carefully when score
distributions differ. |
+| `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. |
The second argument is an array of vector route configs created by
`named_struct`:
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 34c8a09c20..eb2f1d45ac 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
@@ -105,8 +105,25 @@ public class HybridSearchRanker {
ScoredGlobalIndexResult result = weightedResult.result();
float weight = weightedResult.weight();
ScoreGetter scoreGetter = result.scoreGetter();
+
+ // Route score scales are heterogeneous (e.g. bounded vector
similarity vs unbounded
+ // BM25), so raw scores are not comparable across routes. Min-max
normalize each route
+ // into [0, 1] before weighting, so that weights -- not a route's
numeric magnitude --
+ // decide its influence on the fused score.
+ float min = Float.POSITIVE_INFINITY;
+ float max = Float.NEGATIVE_INFINITY;
+ for (long rowId : result.results()) {
+ float score = scoreGetter.score(rowId);
+ min = Math.min(min, score);
+ max = Math.max(max, score);
+ }
+ float range = max - min;
+
for (long rowId : result.results()) {
- float contribution = weight * scoreGetter.score(rowId);
+ // No spread within the route (single hit or all ties) carries
no relative signal,
+ // so every hit maps to 1.0 rather than being zeroed out.
+ float normalized = range > 0.0f ? (scoreGetter.score(rowId) -
min) / range : 1.0f;
+ float contribution = weight * normalized;
scores.compute(
rowId,
(k, oldScore) -> oldScore == null ? contribution :
oldScore + contribution);
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 02b835134c..f927d0cf8a 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
@@ -71,7 +71,59 @@ public class HybridSearchRankerTest {
assertThat(ranked.results().getIntCardinality()).isEqualTo(1);
assertThat(ranked.results()).contains(1L);
- assertThat(ranked.scoreGetter().score(1L)).isCloseTo(0.9f,
within(0.000001f));
+ // Within the route {0.3, 0.2} min-max maps 0.3 -> 1.0, so score =
weight (3.0) * 1.0.
+ assertThat(ranked.scoreGetter().score(1L)).isCloseTo(3.0f,
within(0.000001f));
+ }
+
+ @Test
+ public void testWeightedScoreNormalizesHeterogeneousRouteScales() {
+ // Vector-like route: rowId 1 strongly matches, rowId 2 barely matches
(bounded ~[0, 1]).
+ ScoredGlobalIndexResult vectorRoute = result(new long[] {1, 2}, new
float[] {0.95f, 0.10f});
+ // BM25-like route: rowId 2 strongly matches, rowId 1 weakly
(unbounded, larger magnitude).
+ ScoredGlobalIndexResult textRoute = result(new long[] {1, 2}, new
float[] {2.0f, 25.0f});
+
+ // Vector route weighted 5x. Without normalization the BM25 magnitude
would dominate; with
+ // normalization each route maps its best hit to 1.0 and worst to 0.0,
so weights decide.
+ ScoredGlobalIndexResult ranked =
+ HybridSearchRanker.weightedScore(
+ Arrays.asList(vectorRoute, textRoute), new float[]
{5.0f, 1.0f}, 2);
+
+ // rowId 1: 5 * 1.0 (vector best) + 1 * 0.0 (text worst) = 5.0
+ // rowId 2: 5 * 0.0 (vector worst) + 1 * 1.0 (text best) = 1.0
+ assertThat(ranked.scoreGetter().score(1L)).isCloseTo(5.0f,
within(0.000001f));
+ assertThat(ranked.scoreGetter().score(2L)).isCloseTo(1.0f,
within(0.000001f));
+
assertThat(ranked.scoreGetter().score(1L)).isGreaterThan(ranked.scoreGetter().score(2L));
+ }
+
+ @Test
+ public void testWeightedScoreSingleHitRouteMapsToFullWeight() {
+ // A route with a single hit has no spread (min == max); it must not
produce NaN/Infinity
+ // and must keep contributing rather than being zeroed out.
+ ScoredGlobalIndexResult singleHit = result(new long[] {7}, new float[]
{42.0f});
+
+ ScoredGlobalIndexResult ranked =
+ HybridSearchRanker.weightedScore(
+ Collections.singletonList(
+ new
HybridSearchRanker.WeightedResult(singleHit, 2.0f)),
+ 1);
+
+ assertThat(ranked.scoreGetter().score(7L)).isCloseTo(2.0f,
within(0.000001f));
+ }
+
+ @Test
+ public void testWeightedScoreAllTiedRouteMapsEachHitToFullWeight() {
+ // All hits share the same score (no relative signal); each maps to
1.0 -> weight.
+ ScoredGlobalIndexResult tied = result(new long[] {1, 2, 3}, new
float[] {5.0f, 5.0f, 5.0f});
+
+ ScoredGlobalIndexResult ranked =
+ HybridSearchRanker.weightedScore(
+ Collections.singletonList(
+ new HybridSearchRanker.WeightedResult(tied,
2.0f)),
+ 3);
+
+ assertThat(ranked.scoreGetter().score(1L)).isCloseTo(2.0f,
within(0.000001f));
+ assertThat(ranked.scoreGetter().score(2L)).isCloseTo(2.0f,
within(0.000001f));
+ assertThat(ranked.scoreGetter().score(3L)).isCloseTo(2.0f,
within(0.000001f));
}
private ScoredGlobalIndexResult result(long[] rowIds, float[] scores) {
diff --git a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
index 39a93a0167..bb5a32cced 100644
--- a/paimon-python/pypaimon/table/source/hybrid_search_builder.py
+++ b/paimon-python/pypaimon/table/source/hybrid_search_builder.py
@@ -376,11 +376,28 @@ class HybridSearchBuilderImpl(HybridSearchBuilder):
scores = {}
for route_result in route_results:
result = route_result.result
+ weight = route_result.route.weight
score_getter = result.score_getter()
- for row_id in result.results():
- contribution = route_result.route.weight * (
- score_getter(row_id) or 0.0)
- scores[row_id] = scores.get(row_id, 0.0) + contribution
+
+ # Route score scales are heterogeneous (e.g. bounded vector
similarity
+ # vs unbounded BM25), so raw scores are not comparable across
routes.
+ # Min-max normalize each route into [0, 1] before weighting, so
that
+ # weights -- not a route's numeric magnitude -- decide its
influence on
+ # the fused score. This mirrors the Java HybridSearchRanker.
+ route_scores = {
+ row_id: (score_getter(row_id) or 0.0)
+ for row_id in result.results()
+ }
+ if not route_scores:
+ continue
+ min_score = min(route_scores.values())
+ score_range = max(route_scores.values()) - min_score
+ for row_id, raw in route_scores.items():
+ # No spread within the route (single hit or all ties) carries
no
+ # relative signal, so every hit maps to 1.0 rather than zeroed
out.
+ normalized = (
+ (raw - min_score) / score_range if score_range > 0.0 else
1.0)
+ scores[row_id] = scores.get(row_id, 0.0) + weight * normalized
return _top_k(scores, self._limit)
def _split_partition_filter(self, predicate):
diff --git a/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
b/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
new file mode 100644
index 0000000000..aa4936dec2
--- /dev/null
+++ b/paimon-python/pypaimon/tests/hybrid_search_ranker_test.py
@@ -0,0 +1,86 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Tests for HybridSearchBuilderImpl ranking, mirroring the Java
+HybridSearchRankerTest so the weighted_score ranker stays consistent across
+languages (per-route min-max normalization before weighting)."""
+
+import unittest
+
+from pypaimon.globalindex.vector_search_result import
DictBasedScoredIndexResult
+from pypaimon.table.source.hybrid_search_builder import (
+ HybridSearchBuilderImpl, HybridSearchRoute, HybridSearchRouteResult)
+
+
+def _route_result(weight, id_to_scores):
+ route = HybridSearchRoute.vector_route("f", [1.0], 10, weight=weight)
+ return HybridSearchRouteResult(route,
DictBasedScoredIndexResult(id_to_scores))
+
+
+def _builder(limit):
+ builder = HybridSearchBuilderImpl(table=None)
+ builder._limit = limit
+ return builder
+
+
+class HybridSearchRankerTest(unittest.TestCase):
+
+ def test_weighted_score_normalizes_heterogeneous_route_scales(self):
+ # Vector-like route (bounded ~[0, 1]) and BM25-like route (unbounded,
larger).
+ vector_route = _route_result(5.0, {1: 0.95, 2: 0.10})
+ text_route = _route_result(1.0, {1: 2.0, 2: 25.0})
+
+ ranked = _builder(2)._weighted_score([vector_route, text_route])
+ getter = ranked.score_getter()
+
+ # rowId 1: 5 * 1.0 (vector best) + 1 * 0.0 (text worst) = 5.0
+ # rowId 2: 5 * 0.0 (vector worst) + 1 * 1.0 (text best) = 1.0
+ self.assertAlmostEqual(getter(1), 5.0, places=6)
+ self.assertAlmostEqual(getter(2), 1.0, places=6)
+ self.assertGreater(getter(1), getter(2))
+
+ def test_weighted_score_single_hit_route_maps_to_full_weight(self):
+ # min == max: must not divide by zero and must keep contributing.
+ single_hit = _route_result(2.0, {7: 42.0})
+
+ ranked = _builder(1)._weighted_score([single_hit])
+
+ self.assertAlmostEqual(ranked.score_getter()(7), 2.0, places=6)
+
+ def test_weighted_score_all_tied_route_maps_each_hit_to_full_weight(self):
+ tied = _route_result(2.0, {1: 5.0, 2: 5.0, 3: 5.0})
+
+ ranked = _builder(3)._weighted_score([tied])
+ getter = ranked.score_getter()
+
+ self.assertAlmostEqual(getter(1), 2.0, places=6)
+ self.assertAlmostEqual(getter(2), 2.0, places=6)
+ self.assertAlmostEqual(getter(3), 2.0, places=6)
+
+ def test_rrf_unaffected_and_respects_weight_on_same_input(self):
+ vector_route = _route_result(5.0, {1: 0.95, 2: 0.10})
+ text_route = _route_result(1.0, {1: 2.0, 2: 25.0})
+
+ ranked = _builder(2)._rrf([vector_route, text_route])
+ getter = ranked.score_getter()
+
+ # Rank-based fusion respects the 5x vector weight: rowId 1 wins.
+ self.assertGreater(getter(1), getter(2))
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala
index 7bfbfcfa18..bfc39cad3a 100644
---
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/HybridSearchTest.scala
@@ -84,6 +84,75 @@ class HybridSearchTest extends PaimonSparkTestBase {
}
}
+ test("weighted_score exposes per-route min-max normalized fused scores") {
+ withTable("T") {
+ spark.sql("""
+ |CREATE TABLE T (id INT, vec_a ARRAY<FLOAT>, vec_b
ARRAY<FLOAT>)
+ |TBLPROPERTIES (
+ | 'bucket' = '-1',
+ | 'global-index.row-count-per-shard' = '10000',
+ | 'row-tracking.enabled' = 'true',
+ | 'data-evolution.enabled' = 'true',
+ | 'test.vector.dimension' = '2',
+ | 'test.vector.required-option.key' = 'ivf.nprobe',
+ | 'test.vector.required-option.value' = '16')
+ |""".stripMargin)
+
+ // Both columns hold identical vectors, so each route ranks id0 best and
id2 worst.
+ // Scores vs query [1, 0] are strictly ordered id0 > id1 > id2 under
every supported
+ // metric, so min-max maps id0 -> 1.0 and id2 -> 0.0 in BOTH routes
regardless of metric.
+ // id2's raw score is > 0, so a fused 0.0 can only come from
normalization, not a raw sum.
+ spark.sql("""
+ |INSERT INTO T VALUES
+ | (0, array(1.0f, 0.0f), array(1.0f, 0.0f)),
+ | (1, array(0.8f, 0.6f), array(0.8f, 0.6f)),
+ | (2, array(0.6f, 0.8f), array(0.6f, 0.8f))
+ |""".stripMargin)
+
+ spark
+ .sql(s"CALL sys.create_global_index(table => 'test.T', index_column =>
'vec_a', " +
+ s"index_type => '${TestVectorGlobalIndexerFactory.IDENTIFIER}')")
+ .collect()
+ spark
+ .sql(s"CALL sys.create_global_index(table => 'test.T', index_column =>
'vec_b', " +
+ s"index_type => '${TestVectorGlobalIndexerFactory.IDENTIFIER}')")
+ .collect()
+
+ val scores = spark
+ .sql("""
+ |SELECT id, __paimon_search_score
+ |FROM hybrid_search(
+ | 'T',
+ | array(
+ | named_struct(
+ | 'field', 'vec_a',
+ | 'query_vector', array(1.0f, 0.0f),
+ | 'limit', 3,
+ | 'weight', 2.0f,
+ | 'options', map('ivf.nprobe', '16')),
+ | named_struct(
+ | 'field', 'vec_b',
+ | 'query_vector', array(1.0f, 0.0f),
+ | 'limit', 3,
+ | 'weight', 1.0f,
+ | 'options', map('ivf.nprobe', '16'))),
+ | array(),
+ | 3,
+ | 'weighted_score')
+ |""".stripMargin)
+ .collect()
+ .map(row => row.getInt(0) -> row.getFloat(1))
+ .toMap
+
+ // id0 is the top hit in both routes -> 1.0 each -> 2.0 * 1.0 + 1.0 *
1.0 = sum(weights).
+ assert(math.abs(scores(0) - 3.0f) < 1e-5)
+ // id2 is the worst hit in both routes -> min-max maps it to 0.0 each ->
fused 0.0.
+ // Without normalization this would be 2.0 * raw_a + 1.0 * raw_b > 0.
+ assert(math.abs(scores(2) - 0.0f) < 1e-5)
+ assert(scores(0) > scores(2))
+ }
+ }
+
test("hybrid search ranks vector and full-text routes") {
withTable("T") {
spark.sql("""
