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