From 77d645ebdc0f19d392a0c38548a3e497006d65ef Mon Sep 17 00:00:00 2001 From: Zihan Dai Date: Tue, 18 Aug 2026 04:11:47 +1000 Subject: [PATCH] [core] Match the lookup high level record by position, not by identity LookupMergeFunction.getResult scans the candidate buffer twice and re-identifies the high level record with kv == highLevel. Once the candidates exceed lookup.merge-records-threshold they spill to a BinaryBuffer, whose iterator deserializes a new KeyValue on every call, so the second scan never sees the object the first scan returned and the high level record is dropped from the merge. Record the position pickHighLevel chose and compare positions instead. --- .../compact/LookupMergeFunction.java | 30 ++++++--- .../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 309d97a9031a..a98baacf5980 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 { 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 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 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 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 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 71ac78259d74..2087eb66566c 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 @@ 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(); + } }