JingsongLi commented on code in PR #8356:
URL: https://github.com/apache/paimon/pull/8356#discussion_r3475105886
##########
paimon-core/src/main/java/org/apache/paimon/mergetree/LookupLevels.java:
##########
@@ -144,29 +147,84 @@ private T lookup(InternalRow key, SortedRun level) throws
IOException {
@Nullable
private T lookup(InternalRow key, DataFileMeta file) throws IOException {
- LookupFile lookupFile = lookupFileCache.getIfPresent(file.fileName());
+ byte[] keyBytes = serializeKey(key);
+ LookupResult lookupResult = lookupFile(file, keyBytes);
+ byte[] valueBytes = lookupResult.valueBytes;
+ if (valueBytes == null) {
+ return null;
+ }
- boolean newCreatedLookupFile = false;
- if (lookupFile == null) {
- lookupFile = createLookupFile(file);
- newCreatedLookupFile = true;
+ return readFromDisk(
+ getOrCreateProcessor(lookupResult.schemaId,
lookupResult.serVersion),
+ key,
+ lookupResult.level,
+ valueBytes,
+ file.fileName());
+ }
+
+ private LookupResult lookupFile(DataFileMeta file, byte[] keyBytes) throws
IOException {
+ String fileName = file.fileName();
+ LookupFile lookupFile = lookupFileCache.getIfPresent(fileName);
+ LookupResult lookupResult = lookupCachedFile(fileName, lookupFile,
keyBytes);
+ if (lookupResult != null) {
+ return lookupResult;
}
- byte[] valueBytes;
- try {
- byte[] keyBytes = keySerializer.serializeToBytes(key);
- valueBytes = lookupFile.get(keyBytes);
- } finally {
- if (newCreatedLookupFile) {
- addLocalFile(file, lookupFile);
+ Object lock = lookupFileLocks.computeIfAbsent(fileName, ignored -> new
Object());
+ synchronized (lock) {
+ try {
+ lookupFile = lookupFileCache.getIfPresent(fileName);
+ lookupResult = lookupCachedFile(fileName, lookupFile,
keyBytes);
+ if (lookupResult != null) {
+ return lookupResult;
+ }
+
+ lookupFile = createLookupFile(file);
+
+ try {
+ return LookupResult.of(lookupFile,
lookupFile.get(keyBytes));
+ } finally {
+ addLocalFile(file, lookupFile);
+ }
+ } finally {
+ lookupFileLocks.remove(fileName, lock);
Review 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.
--
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]
