PDGGK opened a new pull request, #9279:
URL: https://github.com/apache/paimon/pull/9279
### Purpose
`LookupMergeFunction.getResult` scans the candidate buffer twice and
re-identifies the high level record by object identity:
```java
// 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:
```java
// 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:
```java
if (kv.level() > 0) {
if (highLevel == null || kv.level() < highLevel.level()) {
highLevel = kv;
highLevelIndex = index;
}
}
index++;
```
```java
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
