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 8b2911b5b9 [core] Stabilize scored global index top-k tie-breaking
(#8301)
8b2911b5b9 is described below
commit 8b2911b5b9ff43b8fe1e76684e1f8096b67801a4
Author: QuakeWang <[email protected]>
AuthorDate: Mon Jun 22 17:29:48 2026 +0800
[core] Stabilize scored global index top-k tie-breaking (#8301)
`ScoredGlobalIndexResult.topK` used a min-heap ordered only by score,
and replaced the heap head only when a new row had a strictly higher
score. When rows tied around the top-k boundary, the retained row IDs
could differ from the expected `score desc, rowId asc` ranking
semantics.
---
.../globalindex/ScoredGlobalIndexResult.java | 17 +++--
.../globalindex/ScoredGlobalIndexResultTest.java | 77 ++++++++++++++++++++++
2 files changed, 88 insertions(+), 6 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/globalindex/ScoredGlobalIndexResult.java
b/paimon-common/src/main/java/org/apache/paimon/globalindex/ScoredGlobalIndexResult.java
index cc75b207a5..3faeeee7ba 100644
---
a/paimon-common/src/main/java/org/apache/paimon/globalindex/ScoredGlobalIndexResult.java
+++
b/paimon-common/src/main/java/org/apache/paimon/globalindex/ScoredGlobalIndexResult.java
@@ -85,17 +85,22 @@ public interface ScoredGlobalIndexResult extends
GlobalIndexResult {
}
ScoreGetter scoreGetter = scoreGetter();
- // Min-heap by score: the head is the smallest score so we can evict
it when a
- // higher-scored row arrives. This gives O(n log k) instead of O(n log
n).
- PriorityQueue<long[]> minHeap =
- new PriorityQueue<>(
- k + 1, Comparator.comparingDouble(a ->
Float.intBitsToFloat((int) a[1])));
+ // Min-heap whose ordering matches the global index ranking semantics
(score desc,
+ // rowId asc): the head is the weakest candidate currently kept, i.e.
the lowest
+ // score and, among ties, the largest rowId. A new row replaces the
head only when
+ // it is strictly stronger, so the retained set equals a full "score
desc, rowId asc"
+ // sort truncated to k, while keeping O(n log k) instead of O(n log n).
+ // entry: [rowId, rawScoreBits]
+ Comparator<long[]> weakestFirst =
+ Comparator.<long[]>comparingDouble(a ->
Float.intBitsToFloat((int) a[1]))
+ .thenComparing(Comparator.comparingLong((long[] a) ->
a[0]).reversed());
+ PriorityQueue<long[]> minHeap = new PriorityQueue<>(k + 1,
weakestFirst);
for (long rowId : rowIds) {
float score = scoreGetter.score(rowId);
long[] entry = new long[] {rowId, Float.floatToRawIntBits(score)};
if (minHeap.size() < k) {
minHeap.offer(entry);
- } else if (score > Float.intBitsToFloat((int) minHeap.peek()[1])) {
+ } else if (weakestFirst.compare(entry, minHeap.peek()) > 0) {
minHeap.poll();
minHeap.offer(entry);
}
diff --git
a/paimon-common/src/test/java/org/apache/paimon/globalindex/ScoredGlobalIndexResultTest.java
b/paimon-common/src/test/java/org/apache/paimon/globalindex/ScoredGlobalIndexResultTest.java
new file mode 100644
index 0000000000..756e7733cf
--- /dev/null
+++
b/paimon-common/src/test/java/org/apache/paimon/globalindex/ScoredGlobalIndexResultTest.java
@@ -0,0 +1,77 @@
+/*
+ * 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.
+ */
+
+package org.apache.paimon.globalindex;
+
+import org.apache.paimon.utils.RoaringNavigableMap64;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ScoredGlobalIndexResult}. */
+public class ScoredGlobalIndexResultTest {
+
+ @Test
+ public void testTopKBreaksBoundaryTiesByRowId() {
+ ScoredGlobalIndexResult result =
+ result(new long[] {1, 2, 3, 4}, new float[] {0.5f, 0.5f, 0.5f,
0.9f});
+
+ RoaringNavigableMap64 topK = result.topK(2).results();
+
+ assertThat(topK.getIntCardinality()).isEqualTo(2);
+ assertThat(topK).contains(1L, 4L);
+ assertThat(topK).doesNotContain(2L, 3L);
+ }
+
+ @Test
+ public void testTopKBreaksTieGroupBySmallerRowId() {
+ ScoredGlobalIndexResult result =
+ result(new long[] {1, 2, 3, 4, 5}, new float[] {0.5f, 0.5f,
0.5f, 0.5f, 0.9f});
+
+ // The high-scored row 5 forces a heap eviction within the 0.5 tie
group, so this
+ // distinguishes the score-only heap ({2, 3, 5}) from the fixed heap
({1, 2, 5}).
+ RoaringNavigableMap64 topK = result.topK(3).results();
+
+ assertThat(topK.getIntCardinality()).isEqualTo(3);
+ assertThat(topK).contains(1L, 2L, 5L);
+ assertThat(topK).doesNotContain(3L, 4L);
+ }
+
+ @Test
+ public void testTopKReturnsSameResultWhenCardinalityDoesNotExceedK() {
+ ScoredGlobalIndexResult result =
+ result(new long[] {1, 2, 3}, new float[] {0.1f, 0.2f, 0.3f});
+
+ assertThat(result.topK(3)).isSameAs(result);
+ assertThat(result.topK(5)).isSameAs(result);
+ }
+
+ private ScoredGlobalIndexResult result(long[] rowIds, float[] scores) {
+ RoaringNavigableMap64 bitmap = new RoaringNavigableMap64();
+ Map<Long, Float> scoreMap = new HashMap<>();
+ for (int i = 0; i < rowIds.length; i++) {
+ bitmap.add(rowIds[i]);
+ scoreMap.put(rowIds[i], scores[i]);
+ }
+ return ScoredGlobalIndexResult.create(bitmap, scoreMap::get);
+ }
+}
