[core] Match the lookup high level record by position, not by identity

paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java

@@ -42,6 +42,9 @@ public class LookupMergeFunction implements MergeFunction<KeyValue> {
     private boolean containLevel0;
     private InternalRow currentKey;
 
+    /** Position of the record {@link #pickHighLevel} chose, -1 when there is none. */
+    private int highLevelIndex = -1;
+
     public LookupMergeFunction(
             MergeFunction<KeyValue> mergeFunction,
             CoreOptions options,
@@ -57,6 +60,7 @@ public void reset() {
         candidates.reset();
         currentKey = null;
         containLevel0 = false;
+        highLevelIndex = -1;
     }
 
     @Override
@@ -75,19 +79,22 @@ public boolean containLevel0() {
     @Nullable
     public KeyValue pickHighLevel() {
         KeyValue highLevel = null;
+        highLevelIndex = -1;
+        int index = 0;
         try (CloseableIterator<KeyValue> iterator = candidates.iterator()) {
             while (iterator.hasNext()) {
                 KeyValue kv = iterator.next();
                 // records that has not been stored on the disk yet, such as the data in the write
                 // buffer being at level -1
-                if (kv.level() <= 0) {
-                    continue;
-                }
-                // For high-level comparison logic (not involving Level 0), only the value of the
-                // minimum Level should be selected
-                if (highLevel == null || kv.level() < highLevel.level()) {
-                    highLevel = kv;
+                if (kv.level() > 0) {
+                    // For high-level comparison logic (not involving Level 0), only the value of
+                    // the minimum Level should be selected
+                    if (highLevel == null || kv.level() < highLevel.level()) {
+                        highLevel = kv;
+                        highLevelIndex = index;
+                    }
                 }
+                index++;
             }
         } catch (Exception e) {
             throw new RuntimeException(e);
@@ -106,15 +113,20 @@ public void insertInto(KeyValue highLevel, Comparator<KeyValue> comparator) {
     @Override
     public KeyValue getResult() {
         mergeFunction.reset();
-        KeyValue highLevel = pickHighLevel();
+        // match the high level record by its position: once the candidates have spilled, every
+        // iteration deserializes fresh KeyValue instances, so the one picked above is never the
+        // same object as the one seen here
+        pickHighLevel();
+        int index = 0;
         try (CloseableIterator<KeyValue> iterator = candidates.iterator()) {
             while (iterator.hasNext()) {
                 KeyValue kv = iterator.next();
                 // records that has not been stored on the disk yet, such as the data in the write
                 // buffer being at level -1
-                if (kv.level() <= 0 || kv == highLevel) {
+                if (kv.level() <= 0 || index == highLevelIndex) {
                     mergeFunction.add(kv);
                 }
+                index++;
             }
         } catch (Exception e) {
             throw new RuntimeException(e);

paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java

@@ -18,9 +18,19 @@
 
 package org.apache.paimon.mergetree.compact;
 
+import org.apache.paimon.CoreOptions;
 import org.apache.paimon.KeyValue;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
 
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import javax.annotation.Nullable;
+
+import java.nio.file.Path;
 
 import static org.apache.paimon.io.DataFileTestUtils.row;
 import static org.apache.paimon.types.RowKind.INSERT;
@@ -57,4 +67,60 @@ public void testLevelNegative() {
         assertThat(kv).isNotNull();
         assertThat(kv.value().getInt(0)).isEqualTo(1);
     }
+
+    @TempDir Path tempDir;
+
+    /**
+     * Same scenario as {@link #testKeepLowestHighLevel()}, but with the candidates spilled: every
+     * iteration over a spilled buffer deserializes fresh instances, so an identity check against
+     * the record picked by a previous iteration can never match.
+     */
+    @Test
+    public void testKeepLowestHighLevelWhenCandidatesHaveSpilled() {
+        for (boolean withIoManager : new boolean[] {false, true}) {
+            LookupMergeFunction function = spillingFunction(withIoManager);
+            function.reset();
+            function.add(new KeyValue().replace(row(1), 1, INSERT, row(2)).setLevel(1));
+            function.add(new KeyValue().replace(row(1), 1, INSERT, row(1)).setLevel(2));
+            KeyValue kv = function.getResult();
+            assertThat(kv).as("spilled, ioManager=%s", withIoManager).isNotNull();
+            assertThat(kv.value().getInt(0)).isEqualTo(2);
+        }
+    }
+
+    /**
+     * The lowest high level record is not simply the first or the last one, so this also covers the
+     * position bookkeeping rather than only "some high level record was merged".
+     */
+    @Test
+    public void testPicksTheLowestHighLevelFromTheMiddleWhenCandidatesHaveSpilled() {
+        for (boolean withIoManager : new boolean[] {false, true}) {
+            LookupMergeFunction function = spillingFunction(withIoManager);
+            function.reset();
+            function.add(new KeyValue().replace(row(1), 1, INSERT, row(30)).setLevel(3));
+            function.add(new KeyValue().replace(row(1), 2, INSERT, row(10)).setLevel(1));
+            function.add(new KeyValue().replace(row(1), 3, INSERT, row(20)).setLevel(2));
+            KeyValue kv = function.getResult();
+            assertThat(kv).as("spilled, ioManager=%s", withIoManager).isNotNull();
+            assertThat(kv.value().getInt(0)).isEqualTo(10);
+        }
+    }
+
+    private LookupMergeFunction spillingFunction(boolean withIoManager) {
+        Options options = new Options();
+        // spill as soon as there is more than one candidate for the key
+        options.set(CoreOptions.LOOKUP_MERGE_RECORDS_THRESHOLD, 1);
+        RowType keyType = RowType.builder().field("k", DataTypes.INT()).build();
+        RowType valueType = RowType.builder().field("v", DataTypes.INT()).build();
+        LookupMergeFunction.Factory factory =
+                (LookupMergeFunction.Factory)
+                        LookupMergeFunction.wrap(
+                                DeduplicateMergeFunction.factory(),
+                                new CoreOptions(options),
+                                keyType,
+                                valueType);
+        @Nullable IOManager ioManager = withIoManager ? IOManager.create(tempDir.toString()) : null;
+        factory.withIOManager(ioManager);
+        return (LookupMergeFunction) factory.create();
+    }
 }