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

[PDGGK]

Purpose

LookupMergeFunction.getResult scans the candidate buffer twice and re-identifies the high level record by object identity:

// LookupMergeFunction:107-118
public KeyValue getResult() {
    mergeFunction.reset();
    KeyValue highLevel = pickHighLevel();          // first scan
    try (CloseableIterator<KeyValue> iterator = candidates.iterator()) {   // second scan
        while (iterator.hasNext()) {
            KeyValue kv = iterator.next();
            if (kv.level() <= 0 || kv == highLevel) {
                mergeFunction.add(kv);
            }
        }
    }
    ...

The candidates are held in a KeyValueBuffer.HybridBuffer, which keeps them in an ArrayList until there are more than lookup.merge-records-threshold of them for the key and then spills to a BinaryBuffer (KeyValueBuffer:91). A spilled buffer hands out a new object on every call:

// KeyValueBuffer:192
return kvSerializer.fromRow(iterator.getRow().copy());

So after the spill the second scan never sees the object the first scan returned. kv == highLevel is false for every record, and the high level record is dropped from the merge.

The threshold counts candidates for a single key in one merge, so this needs a key with more than lookup.merge-records-threshold records across the files being merged — 1024 by default, lower for anyone who has tuned the option down to bound memory.

What it does

LookupMergeFunctionTest.testKeepLowestHighLevel — two high level records, expecting the lower level to win — is the shape that breaks. Run unchanged except that the candidates spill, it returns null instead of the level-1 value: neither record is level 0 and neither matches by identity, so nothing at all reaches the wrapped merge function.

In the pipeline this runs through LookupChangelogMergeFunctionWrapper.getResult (:104-146), and the two halves then disagree. The wrapper holds its own non-null highLevel from its own pickHighLevel() call at :106 and passes it to setChangelog as the before image at :142, while the after image it is compared against was merged without that record. For a merge engine that folds the persisted row into the result — partial-update, aggregation — the merged row loses the columns that only the high level record carried, and the changelog describes an update against a base row that was never part of it.

What changes

Match the record by its position in the buffer instead of by object identity. pickHighLevel records the index of the record it chose, and getResult compares indices on the second scan:

if (kv.level() > 0) {
    if (highLevel == null || kv.level() < highLevel.level()) {
        highLevel = kv;
        highLevelIndex = index;
    }
}
index++;

pickHighLevel();
int index = 0;
...
    if (kv.level() <= 0 || index == highLevelIndex) {
        mergeFunction.add(kv);
    }
    index++;

Both scans walk the same buffer with no intervening writes, and the buffer iterates in insertion order: ListBuffer over an ArrayList, and BinaryBuffer over a RowBuffer whose newIterator() builds a fresh RandomAccessInputView from the start of the record segments and reads forward (InMemoryBuffer:119-127). Both spilled flavours are covered by the tests below rather than only by that reading. highLevelIndex is reset in reset() alongside the other per-key state.

That same code is why identity can never hold once spilled: the iterator's getRow() returns a reused BinaryRow, so BinaryBuffer has to copy() it before deserialising.

This is deliberately not a value comparison: two candidates for one key can share a level, and comparing on (level, sequenceNumber) would rest on an assumption about sequence numbers that the position does not need.

Test evidence

Two tests, each run against both spilled buffer flavours — ioManager == null, which spills to an InMemoryBuffer, and a real IOManager, which spills to an ExternalBuffer:

• testKeepLowestHighLevelWhenCandidatesHaveSpilled — the existing testKeepLowestHighLevel scenario with lookup.merge-records-threshold = 1.
• testPicksTheLowestHighLevelFromTheMiddleWhenCandidatesHaveSpilled — levels 3, 1, 2 inserted in that order, so the answer is neither the first nor the last candidate and an off-by-one in the index bookkeeping would not pass.

Mutation control, on a forced clean rebuild of paimon-core (rm -rf target/classes target/test-classes) so this is not an incremental-build artefact: with the tests kept and LookupMergeFunction reverted, both new tests fail — Expecting actual not to be null — while the two pre-existing tests in the class stay green, which is the point: they do not spill.

Wider run: *Lookup*Test, *MergeTree*Test, *MergeFunction*Test, KeyValueBufferTest and *Compact*Test across paimon-core — 286 tests, 0 failures.

API and Format

No change to any option, on-disk format or public signature. Below the spill threshold the two scans return the same objects and the behaviour is byte-for-byte what it was.