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 4d54b5a86d [core] Match the lookup high level record by position, not
by identity (#9279)
4d54b5a86d is described below
commit 4d54b5a86dfbfa7c1189d8e8ce0b780d818e55c1
Author: ZIHAN DAI <[email protected]>
AuthorDate: Thu Aug 20 12:02:16 2026 +1000
[core] Match the lookup high level record by position, not by identity
(#9279)
---
.../mergetree/compact/LookupMergeFunction.java | 30 +++++++---
.../mergetree/compact/LookupMergeFunctionTest.java | 66 ++++++++++++++++++++++
2 files changed, 87 insertions(+), 9 deletions(-)
diff --git
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
b/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
index 309d97a903..a98baacf59 100644
---
a/paimon-core/src/main/java/org/apache/paimon/mergetree/compact/LookupMergeFunction.java
+++
b/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 class LookupMergeFunction implements
MergeFunction<KeyValue> {
candidates.reset();
currentKey = null;
containLevel0 = false;
+ highLevelIndex = -1;
}
@Override
@@ -75,19 +79,22 @@ public class LookupMergeFunction implements
MergeFunction<KeyValue> {
@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 class LookupMergeFunction implements
MergeFunction<KeyValue> {
@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);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java
b/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java
index 71ac78259d..2087eb6656 100644
---
a/paimon-core/src/test/java/org/apache/paimon/mergetree/compact/LookupMergeFunctionTest.java
+++
b/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 @@ class LookupMergeFunctionTest {
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();
+ }
}
