[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.

Author: QuakeWang

paimon-common/src/main/java/org/apache/paimon/globalindex/ScoredGlobalIndexResult.java

@@ -85,17 +85,22 @@ default ScoredGlobalIndexResult topK(int k) {
         }
 
         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);
             }

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);
+    }
+}