[core] Make local table query lookup thread-safe#8356
Conversation
| addLocalFile(file, lookupFile); | ||
| } | ||
| } finally { | ||
| lookupFileLocks.remove(fileName, lock); |
There was a problem hiding this comment.
Removing the per-file lock here can split synchronization while other threads are already waiting on the old lock. If the first creator fails, or the newly-created lookup file is immediately evicted because of the disk-size limit, a later caller can install a new lock while an older waiter continues under the removed lock; then two creators may run for the same data file. With deterministic local file names this can surface as createNewFile() failures. Consider a ref-counted/striped keyed lock or a single-flight Future entry that is not removed until no waiters remain.
There was a problem hiding this comment.
Fixed by replacing the removable per-file lock map with fixed striped keyed locks. This avoids splitting waiters across old/new lock objects when lookup file creation fails or the cache entry is evicted. Added a regression test covering the first-creator-fails + old-waiter + later-caller scenario.
| public synchronized InternalRow lookup(BinaryRow partition, int bucket, InternalRow key) | ||
| throws IOException { | ||
| Map<Integer, LookupLevels<KeyValue>> buckets = tableView.get(partition); | ||
| public InternalRow lookup(BinaryRow partition, int bucket, InternalRow key) throws IOException { |
There was a problem hiding this comment.
By removing the method-level synchronization here, cache misses for different files in the same bucket can now enter LookupLevels concurrently. Those paths share the same KeyValueFileReaderFactory, whose formatReaderMappings cache is a plain HashMap mutated via computeIfAbsent during createRecordReader. The old synchronized lookup serialized that path; this change exposes the shared factory to concurrent mutation. Please make that cache thread-safe or serialize reader creation for the shared factory.
Stephen0421
force-pushed
the
lookup-concurrency-improve
branch
from
bb4b206 to
924696e
Compare
2 participants
Purpose
This PR makes local table lookup safer for concurrent access.
LocalTableQuerypreviously synchronized the wholelookupmethod. This PR replaces the coarse-grained query-level synchronization with bucket-level state and read/write locking, solookup()andrefreshFiles()have explicit concurrency semantics per bucket.It also hardens lookup file lifecycle handling:
LookupFileaccess and close idempotent/thread-safe.LookupFile.closeorLookupLevels.createLookupFilefails on checked or unchecked exceptions.This is intended as correctness groundwork for concurrent local lookup usage, rather than a standalone throughput benchmark change.
Tests
Added tests:
LookupFileTest#testCloseCleansLocalStateWhenReaderCloseFailsLookupLevelsTest#testCreateReaderRuntimeFailureCleansLocalFileLookupLevelsTestandPrimaryKeySimpleTableTestVerified by:
mvn -pl paimon-core -Pfast-build -Dtest=LookupFileTest,LookupLevelsTest testmvn -pl paimon-core -Pfast-build -Dtest=PrimaryKeySimpleTableTest#testTableQueryForLookup+testTableQueryForLookupLocalSortFile+testTableQueryForNormal+testTableQueryDownloadsRemoteLookupFile test