luoyuxia opened a new issue, #9483:
URL: https://github.com/apache/paimon/issues/9483

   ### Search before asking
   
   - [x] I searched in the [issues](https://github.com/apache/paimon/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:
   
   ```java
   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](https://github.com/apache/fluss/actions/runs/33309807839/job/99260308512?pr=4120).
 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`:
   
   ```java
   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:
   
   ```java
   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](https://github.com/apache/paimon/pull/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](https://github.com/apache/paimon/issues/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!
   


-- 
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]

Reply via email to