[Bug] Concurrent LocalTableQuery lookups across files can return null

[luoyuxia]

Search before asking

• I searched in the issues and found nothing similar.

Paimon version

• Paimon 2.0.0
• Current master at bfeb7cf2de9c4462083c38c1fe2d73eb9a2fea65

Compute Engine

Java API (LocalTableQuery), observed through Apache Fluss historical lookup.

Minimal reproduce step

1. Create a partitioned primary-key table and commit one row to each of two different partitions, so the rows are stored in different data files.
2. Create one LocalTableQuery with an IOManager.
3. Call refreshFiles for both partition-buckets.
4. From a cold local lookup cache, start two threads together and let each thread look up the existing key from a different partition-bucket.
5. Repeat with a fresh LocalTableQuery and cache.

The stress test is equivalent to:

LocalTableQuery query = new LocalTableQuery(table).withIOManager(ioManager);
query.refreshFiles(partition1, 0, Collections.emptyList(), files1);
query.refreshFiles(partition2, 0, Collections.emptyList(), files2);

CountDownLatch start = new CountDownLatch(1);
Future<InternalRow> first =
        executor.submit(
                () -> {
                    start.await();
                    return query.lookup(partition1, 0, key1);
                });
Future<InternalRow> second =
        executor.submit(
                () -> {
                    start.await();
                    return query.lookup(partition2, 0, key2);
                });
start.countDown();

assertThat(first.get()).isNotNull();
assertThat(second.get()).isNotNull();

One of the two lookups intermittently returns null, although both keys exist. It is easier to reproduce when the two cold-cache lookup files are created and queried concurrently. The problem also reproduces with LocalTableQuery directly, without the Fluss lookup layer.

A Fluss CI occurrence is available at apache/fluss job 99260308512. Rerunning the job reproduced the failure again, with the missing partition varying between runs.

What doesn't meet your expectations?

Concurrent lookups for existing keys should return the same rows as sequential lookups. An existing key must not be reported as absent because another lookup is reading a different local lookup file concurrently.

LocalTableQuery creates one LookupStoreFactory and passes it one RowCompactedSerializer.SliceComparator:

this.lookupStoreFactory =
        LookupStoreFactory.create(
                options,
                new CacheManager(...),
                new RowCompactedSerializer(keyType).createSliceComparator());

The comparator keeps mutable RowReader instances in fields and repoints them during every comparison:

private final RowReader reader1;
private final RowReader reader2;

public int compare(MemorySlice slice1, MemorySlice slice2) {
    reader1.pointTo(slice1.segment(), slice1.offset());
    reader2.pointTo(slice2.segment(), slice2.offset());
    // Read fields from reader1 and reader2.
}

All SortLookupStoreReader instances created by that factory therefore share the same mutable comparator. The per-bucket locks in LocalTableQuery do not prevent lookups for different partition-buckets from using it concurrently. Interleaved pointTo and field reads can produce an incorrect comparison during binary search, resulting in a false miss.

This appears to have been exposed by PR #8356, which replaced query-level synchronization with per-bucket locking. The bucket locks protect each BucketLookupState, but the comparator is owned by the query-level LookupStoreFactory and is shared across buckets.

Anything else?

Related feature request: #3376. That issue asks for concurrent lookup support in general; this report covers a correctness problem in the current implementation after concurrent lookup support was added.

Possible fixes include making SliceComparator stateless/thread-safe, keeping its readers in thread-local state, or creating an independent comparator for each lookup-store reader. A regression test should run concurrent lookups against at least two different lookup files; testing concurrent access to only one file may not exercise the shared comparator race.

Are you willing to submit a PR?

• I'm willing to submit a PR!