fix -1

Author: JingsongLi

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

@@ -106,7 +106,7 @@ public void add(KeyValue kv) {
     public ChangelogResult getResult() {
         // 1. Find the latest high level record and compute containLevel0
         KeyValue highLevel = mergeFunction.pickHighLevel();
-        boolean containLevel0 = mergeFunction.containLevel0();
+        boolean containLevel0 = containLevel0();
 
         // 2. Lookup if latest high level record is absent
         if (highLevel == null) {
@@ -138,6 +138,15 @@ public ChangelogResult getResult() {
         return reusedResult.setResult(result);
     }
 
+    public boolean containLevel0() {
+        for (KeyValue kv : mergeFunction.candidates()) {
+            if (kv.level() == 0) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     private void insertInto(LinkedList<KeyValue> candidates, KeyValue highLevel) {
         List<KeyValue> newCandidates = new ArrayList<>();
         for (KeyValue candidate : candidates) {

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

@@ -53,7 +53,9 @@ public void add(KeyValue kv) {
     public KeyValue pickHighLevel() {
         KeyValue highLevel = null;
         for (KeyValue kv : candidates) {
-            if (kv.level() == 0) {
+            // 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
@@ -65,15 +67,6 @@ public KeyValue pickHighLevel() {
         return highLevel;
     }
 
-    public boolean containLevel0() {
-        for (KeyValue kv : candidates) {
-            if (kv.level() == 0) {
-                return true;
-            }
-        }
-        return false;
-    }
-
     public InternalRow key() {
         return candidates.get(0).key();
     }
@@ -87,7 +80,9 @@ public KeyValue getResult() {
         mergeFunction.reset();
         KeyValue highLevel = pickHighLevel();
         for (KeyValue kv : candidates) {
-            if (kv.level() == 0 || kv == highLevel) {
+            // 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) {
                 mergeFunction.add(kv);
             }
         }