diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java index 9e04ce15ee..0b2685fd7c 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuper.java @@ -23,9 +23,7 @@ import org.apache.fluss.exception.DiskWriteLockedException; import org.apache.fluss.exception.KvStorageException; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; -import org.apache.fluss.lake.paimon.utils.PaimonPartitionBucket; import org.apache.fluss.lake.paimon.utils.PaimonRowAsFlussRow; -import org.apache.fluss.metadata.ResolvedPartitionSpec; import org.apache.fluss.metadata.TablePath; import org.apache.fluss.row.BinaryRow; import org.apache.fluss.row.InternalRow; @@ -45,15 +43,11 @@ import org.apache.paimon.disk.BufferFileWriter; import org.apache.paimon.disk.FileIOChannel; import org.apache.paimon.disk.IOManager; -import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.memory.MemorySegment; import org.apache.paimon.options.Options; import org.apache.paimon.table.FileStoreTable; import org.apache.paimon.table.query.LocalTableQuery; import org.apache.paimon.table.sink.RowPartitionKeyExtractor; -import org.apache.paimon.table.source.DataSplit; -import org.apache.paimon.table.source.InnerTableScan; -import org.apache.paimon.table.source.Split; import org.apache.paimon.types.DataField; import javax.annotation.Nullable; @@ -62,10 +56,7 @@ import java.io.UncheckedIOException; import java.util.ArrayList; import java.util.Collections; -import java.util.HashSet; -import java.util.LinkedHashMap; import java.util.List; -import java.util.Set; import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; @@ -85,11 +76,16 @@ *

A cached partition-bucket file set can become stale when Paimon compaction replaces its data * files and snapshot expiration physically deletes the old files. Because {@code FileIO} * implementations may represent a missing file with different {@link IOException} types, the first - * lookup I/O failure closes the cached query state, reopens the table from the latest snapshot, and + * lookup I/O failure refreshes that partition-bucket with the files from the latest snapshot and * retries once. * - *

Lookup and close operations are synchronized because they share mutable Paimon query, local - * cache, and value-encoding state. + *

Lookup calls do not acquire a Fluss-level lock, allowing a Paimon version that supports + * concurrent lookup to serve them concurrently. Lazy initialization is synchronized only on its + * slow path. A version-aware lookup engine uses Paimon 1.3's query monitor or Paimon 2.0's internal + * bucket locks as appropriate. + * + *

Close is expected only after the owner has drained active lookups. It is synchronized with + * lazy initialization, but deliberately does not add a lifecycle lock to every lookup. */ public class PaimonLakeTableLookuper implements LakeTableLookuper { @@ -100,26 +96,21 @@ public class PaimonLakeTableLookuper implements LakeTableLookuper { private final long lookupCacheMaxDiskBytes; private final Runnable diskWriteGuard; - private final Set initializedBuckets; + private final ThreadLocal lookupFileDownloaded; + private final Object initializationLock; private @Nullable Catalog catalog; private @Nullable FileStoreTable fileStoreTable; private @Nullable IOManager ioManager; - private @Nullable LocalTableQuery localTableQuery; - private @Nullable RowPartitionKeyExtractor partitionKeyExtractor; - private int primaryKeyFieldCount; - private long lookupFileDownloadCount; - - // Both encoders are initialized only for a kv-format-v2 table whose bucket key differs from - // its physical primary key. They remain null when the incoming Fluss key already uses Paimon's - // BinaryRow encoding and no conversion is needed. + private @Nullable List trimmedPrimaryKeys; + + // CompactedKeyDecoder contains immutable type metadata and creates all decode state per + // invocation, so it can be shared by concurrent lookups. private @Nullable CompactedKeyDecoder compactedKeyDecoder; - private @Nullable PaimonKeyEncoder paimonKeyEncoder; - private boolean hasCachedValueEncoder; - private short cachedValueSchemaId; - private @Nullable RowEncoder cachedValueRowEncoder; - private @Nullable InternalRow.FieldGetter[] cachedValueFieldGetters; - private boolean closed; + + private volatile @Nullable PaimonLocalTableQuery localTableQuery; + // Guarded by initializationLock. + private volatile boolean closed; /** Creates a lookuper with the specified local lookup cache limit. */ public PaimonLakeTableLookuper( @@ -137,29 +128,21 @@ public PaimonLakeTableLookuper( lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0."); this.lookupCacheMaxDiskBytes = lookupCacheMaxDiskBytes; this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); - this.initializedBuckets = new HashSet<>(); + this.lookupFileDownloaded = new ThreadLocal<>(); + this.initializationLock = new Object(); } @Override - public synchronized @Nullable byte[] lookup(byte[] key, LookupContext context) - throws Exception { + public @Nullable byte[] lookup(byte[] key, LookupContext context) throws Exception { checkNotNull(key, "key must not be null."); checkNotNull(context, "context must not be null."); checkNotClosed(); ensureInitialized(context.valueRowType()); - org.apache.paimon.data.BinaryRow partition = - convertPartition(context.partitionSpec(), context.valueRowType()); - org.apache.paimon.data.BinaryRow keyRow = toPaimonLookupKey(key); - initializeFilesIfNeeded(partition, context.bucketId()); - - long downloadCountBeforeLookup = lookupFileDownloadCount; + lookupFileDownloaded.set(false); long lookupStartNanos = System.nanoTime(); - org.apache.paimon.data.InternalRow paimonRow; try { - paimonRow = - lookupWithFileRefresh( - partition, context.bucketId(), keyRow, context.valueRowType()); + return lookupInternal(key, context); } catch (Exception e) { DiskWriteLockedException diskWriteLockedException = ExceptionUtils.findThrowable(e, DiskWriteLockedException.class).orElse(null); @@ -168,30 +151,30 @@ public PaimonLakeTableLookuper( } throw e; } finally { + boolean fileDownloaded = lookupFileDownloaded.get(); + lookupFileDownloaded.remove(); context.lookupMetricRecorder() - .recordLookup( - System.nanoTime() - lookupStartNanos, - // An increase means this lookup downloaded at least one lookup file - // through the tracking IO manager. - lookupFileDownloadCount > downloadCountBeforeLookup); + .recordLookup(System.nanoTime() - lookupStartNanos, fileDownloaded); } - if (paimonRow == null) { - return null; - } - return encodeValue(paimonRow, context.schemaId(), context.valueRowType()); } @Override - public synchronized void close() { - if (closed) { - return; + public void close() { + synchronized (initializationLock) { + if (closed) { + return; + } + closed = true; + IOUtils.closeQuietly(localTableQuery, "Paimon lookup engine"); + IOUtils.closeQuietly(ioManager, "Paimon lookup IO manager"); + IOUtils.closeQuietly(catalog, "Paimon catalog"); + localTableQuery = null; + compactedKeyDecoder = null; + trimmedPrimaryKeys = null; + ioManager = null; + fileStoreTable = null; + catalog = null; } - closed = true; - IOUtils.closeQuietly(cachedValueRowEncoder, "Fluss value row encoder"); - IOUtils.closeQuietly(localTableQuery, "Paimon local table query"); - IOUtils.closeQuietly(ioManager, "Paimon lookup IO manager"); - IOUtils.closeQuietly(catalog, "Paimon catalog"); - initializedBuckets.clear(); } private void checkNotClosed() { @@ -201,10 +184,16 @@ private void checkNotClosed() { } private void ensureInitialized(RowType valueRowType) throws Exception { - if (localTableQuery != null) { - return; + if (localTableQuery == null) { + synchronized (initializationLock) { + if (localTableQuery == null) { + initialize(valueRowType); + } + } } + } + private void initialize(RowType valueRowType) throws Exception { Catalog newCatalog = null; IOManager newIOManager = null; LocalTableQuery newLocalTableQuery = null; @@ -221,41 +210,39 @@ private void ensureInitialized(RowType valueRowType) throws Exception { "Point lookup is only supported for primary-key Paimon tables."); } - newIOManager = createIOManager(ioTmpDir); - newLocalTableQuery = newFileStoreTable.newLocalTableQuery(); - newLocalTableQuery.withValueProjection(businessFieldProjection(newFileStoreTable)); - newLocalTableQuery.withIOManager(newIOManager); - RowPartitionKeyExtractor newPartitionKeyExtractor = - new RowPartitionKeyExtractor(newFileStoreTable.schema()); - List trimmedPrimaryKeys = newFileStoreTable.schema().trimmedPrimaryKeys(); - int newPrimaryKeyFieldCount = trimmedPrimaryKeys.size(); + List newTrimmedPrimaryKeys = + Collections.unmodifiableList( + new ArrayList<>(newFileStoreTable.schema().trimmedPrimaryKeys())); CompactedKeyDecoder newCompactedKeyDecoder = null; - PaimonKeyEncoder newPaimonKeyEncoder = null; // Legacy/v1 tables and v2 tables with a default bucket key already encode Fluss // lookup keys with Paimon's key encoder. Only v2 tables with a non-default bucket // key use the compacted key encoding and need conversion before querying Paimon. if (tableConfig.getKvFormatVersion().orElse(1) == KV_FORMAT_VERSION_2 - && !newFileStoreTable.schema().bucketKeys().equals(trimmedPrimaryKeys)) { + && !newFileStoreTable.schema().bucketKeys().equals(newTrimmedPrimaryKeys)) { // Kv-format-v2 tables with a non-default bucket key store Fluss keys using the // compacted encoding to support prefix lookup. Paimon's LocalTableQuery expects // its own BinaryRow encoding, so convert the key at the lake lookup boundary. newCompactedKeyDecoder = - CompactedKeyDecoder.createKeyDecoder(valueRowType, trimmedPrimaryKeys); - RowType keyRowType = valueRowType.project(trimmedPrimaryKeys); - newPaimonKeyEncoder = new PaimonKeyEncoder(keyRowType, trimmedPrimaryKeys); + CompactedKeyDecoder.createKeyDecoder(valueRowType, newTrimmedPrimaryKeys); } - // Publish the newly created state only after every initialization step succeeds. + newIOManager = createIOManager(ioTmpDir); + newLocalTableQuery = + newFileStoreTable + .newLocalTableQuery() + .withValueProjection(businessFieldProjection(newFileStoreTable)) + .withIOManager(newIOManager); + + PaimonLocalTableQuery newLookupEngine = + new PaimonLocalTableQuery(newFileStoreTable, newLocalTableQuery); catalog = newCatalog; fileStoreTable = newFileStoreTable; ioManager = newIOManager; - partitionKeyExtractor = newPartitionKeyExtractor; - primaryKeyFieldCount = newPrimaryKeyFieldCount; + trimmedPrimaryKeys = newTrimmedPrimaryKeys; compactedKeyDecoder = newCompactedKeyDecoder; - paimonKeyEncoder = newPaimonKeyEncoder; - // localTableQuery is the initialization marker, so publish it last. - localTableQuery = newLocalTableQuery; + // Keep this volatile write last to publish all initialized fields together. + localTableQuery = newLookupEngine; initialized = true; } finally { if (!initialized) { @@ -292,139 +279,64 @@ private static int[] businessFieldProjection(FileStoreTable fileStoreTable) { return projection; } - private org.apache.paimon.data.BinaryRow convertPartition( - ResolvedPartitionSpec partitionSpec, RowType valueRowType) { - // The generated partition projection reuses its mutable output, while lookup caches retain - // the returned row as a hash key. Copy it before it escapes to keep those keys stable. - return toPaimonPartition( - partitionSpec, - valueRowType, - fileStoreTable().schema().logicalRowType(), - partitionKeyExtractor()::partition) - .copy(); + private org.apache.paimon.data.BinaryRow getPartition(LookupContext context) { + // Both generated helpers reuse mutable writers or projections, so keep them confined to + // this lookup call. + RowPartitionKeyExtractor partitionKeyExtractor = + new RowPartitionKeyExtractor(fileStoreTable.schema()); + org.apache.paimon.data.BinaryRow partition = + toPaimonPartition( + context.partitionSpec(), + context.valueRowType(), + fileStoreTable.schema().logicalRowType(), + partitionKeyExtractor::partition) + .copy(); + return partition; } - private org.apache.paimon.data.BinaryRow toPaimonLookupKey(byte[] key) { + private org.apache.paimon.data.BinaryRow getKey(byte[] key, LookupContext context) { byte[] paimonKey = key; if (compactedKeyDecoder != null) { - // A non-null decoder means the Fluss lookup key uses compacted encoding. Decode it - // first, then re-encode it as the Paimon BinaryRow expected by LocalTableQuery. InternalRow decodedKey = compactedKeyDecoder.decodeKey(key); - paimonKey = - checkNotNull(paimonKeyEncoder, "Paimon key encoder must be initialized.") - .encodeKey(decodedKey); + RowType keyRowType = context.valueRowType().project(trimmedPrimaryKeys); + PaimonKeyEncoder paimonKeyEncoder = + new PaimonKeyEncoder(keyRowType, trimmedPrimaryKeys); + paimonKey = paimonKeyEncoder.encodeKey(decodedKey); } org.apache.paimon.data.BinaryRow keyRow = - new org.apache.paimon.data.BinaryRow(primaryKeyFieldCount); + new org.apache.paimon.data.BinaryRow(trimmedPrimaryKeys.size()); keyRow.pointTo(MemorySegment.wrap(paimonKey), 0, paimonKey.length); return keyRow; } - private void initializeFilesIfNeeded(org.apache.paimon.data.BinaryRow partition, int bucketId) { - PaimonPartitionBucket partitionBucket = new PaimonPartitionBucket(partition, bucketId); - if (initializedBuckets.contains(partitionBucket)) { - return; - } - - LinkedHashMap dataFilesByName = new LinkedHashMap<>(); - - InnerTableScan tableScan = - fileStoreTable() - .newScan() - .withPartitionFilter(Collections.singletonList(partition)) - .withBucket(bucketId); - for (Split split : tableScan.plan().splits()) { - if (!(split instanceof DataSplit)) { - continue; - } - DataSplit dataSplit = (DataSplit) split; - addFilesByName(dataFilesByName, dataSplit.dataFiles()); - } - - // TODO: Refresh the file set if writes to expired partitions are supported in the future. - // Historical lookup is triggered only after the original Fluss partition has expired and - // been dropped. This PR does not support writes to expired partitions, so no new rows are - // expected and initializing the file set once is sufficient. Compaction-related missing - // files are handled by the IOException refresh path below. - // This partition-bucket has no registered lookup levels yet, so there are no old files to - // remove when building its lookup state from the active data files. - localTableQuery() - .refreshFiles( - partition, - bucketId, - Collections.emptyList(), - new ArrayList<>(dataFilesByName.values())); - initializedBuckets.add(partitionBucket); - } - - private org.apache.paimon.data.InternalRow lookupWithFileRefresh( - org.apache.paimon.data.BinaryRow partition, - int bucketId, - org.apache.paimon.data.InternalRow keyRow, - RowType valueRowType) - throws Exception { + private @Nullable byte[] lookupInternal(byte[] key, LookupContext context) { + org.apache.paimon.data.InternalRow paimonRow; try { - return localTableQuery().lookup(partition, bucketId, keyRow); + paimonRow = + localTableQuery.lookup( + getPartition(context), context.bucketId(), getKey(key, context)); } catch (IOException e) { - // FileIO only guarantees IOException and storage plugins may use different exception - // types for a missing file. The missing old file after compaction may therefore - // surface as any IOException. Refresh and retry only once so persistent I/O failures - // do not repeatedly rebuild Paimon lookup state within one request. - try { - refreshFiles(partition, bucketId, valueRowType); - return localTableQuery().lookup(partition, bucketId, keyRow); - } catch (IOException retryError) { - retryError.addSuppressed(e); - // Historical Paimon point lookup is part of the Fluss KV lookup path. Expose a - // persistent I/O failure as a retriable KV error so the existing KV RPC retry - // semantics can handle it consistently. - throw new KvStorageException( - "Failed to lookup historical data from Paimon after refreshing files for " - + tablePath - + ".", - retryError); - } + // Historical Paimon point lookup is part of the Fluss KV lookup path. Expose a + // persistent I/O failure as a retriable KV error so the existing KV RPC retry + // semantics can handle it consistently. + throw new KvStorageException( + "Failed to lookup historical data from Paimon after refreshing files for " + + tablePath + + ".", + e); } - } - - private void refreshFiles( - org.apache.paimon.data.BinaryRow partition, int bucketId, RowType valueRowType) - throws Exception { - IOUtils.closeQuietly(localTableQuery, "Paimon local table query"); - IOUtils.closeQuietly(ioManager, "Paimon lookup IO manager"); - IOUtils.closeQuietly(catalog, "Paimon catalog"); - localTableQuery = null; - ioManager = null; - catalog = null; - fileStoreTable = null; - partitionKeyExtractor = null; - primaryKeyFieldCount = 0; - compactedKeyDecoder = null; - paimonKeyEncoder = null; - initializedBuckets.clear(); - ensureInitialized(valueRowType); - initializeFilesIfNeeded(partition, bucketId); - } - - private static void addFilesByName( - LinkedHashMap filesByName, List files) { - for (DataFileMeta file : files) { - filesByName.put(file.fileName(), file); + if (paimonRow == null) { + return null; } + return encodeValue(paimonRow, context.schemaId(), context.valueRowType()); } private byte[] encodeValue( org.apache.paimon.data.InternalRow paimonRow, short schemaId, RowType valueRowType) { PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow); - try { - ensureValueEncoder(schemaId, valueRowType); - RowEncoder rowEncoder = - checkNotNull(cachedValueRowEncoder, "cachedValueRowEncoder must not be null."); - InternalRow.FieldGetter[] fieldGetters = - checkNotNull( - cachedValueFieldGetters, "cachedValueFieldGetters must not be null."); - + InternalRow.FieldGetter[] fieldGetters = InternalRow.createFieldGetters(valueRowType); + try (RowEncoder rowEncoder = RowEncoder.create(tableConfig.getKvFormat(), valueRowType)) { rowEncoder.startNewRow(); for (int i = 0; i < fieldGetters.length; i++) { rowEncoder.encodeField(i, fieldGetters[i].getFieldOrNull(flussRow)); @@ -436,30 +348,6 @@ private byte[] encodeValue( } } - private void ensureValueEncoder(short schemaId, RowType valueRowType) { - if (hasCachedValueEncoder && cachedValueSchemaId == schemaId) { - return; - } - - IOUtils.closeQuietly(cachedValueRowEncoder, "Fluss value row encoder"); - cachedValueRowEncoder = RowEncoder.create(tableConfig.getKvFormat(), valueRowType); - cachedValueFieldGetters = InternalRow.createFieldGetters(valueRowType); - cachedValueSchemaId = schemaId; - hasCachedValueEncoder = true; - } - - private FileStoreTable fileStoreTable() { - return checkNotNull(fileStoreTable, "fileStoreTable must be initialized."); - } - - private LocalTableQuery localTableQuery() { - return checkNotNull(localTableQuery, "localTableQuery must be initialized."); - } - - private RowPartitionKeyExtractor partitionKeyExtractor() { - return checkNotNull(partitionKeyExtractor, "partitionKeyExtractor must be initialized."); - } - /** Tracks creation of Paimon lookup files while delegating all local I/O operations. */ private final class TrackingIOManager implements IOManager { @@ -483,8 +371,13 @@ public FileIOChannel.ID createChannel(String prefix) { // I/O boundary here and unwrap the retriable Fluss exception in lookup(). throw new UncheckedIOException(new IOException(e)); } - lookupFileDownloadCount++; - return delegate.createChannel(prefix); + FileIOChannel.ID channel = delegate.createChannel(prefix); + // Paimon creates lookup files synchronously in the lookup thread, so this marks only + // the request that caused this channel to be created. + if (lookupFileDownloaded.get() != null) { + lookupFileDownloaded.set(true); + } + return channel; } @Override @@ -492,6 +385,12 @@ public String[] tempDirs() { return delegate.tempDirs(); } + // Paimon 2.0 adds this method to IOManager. Do not add @Override so the same source also + // compiles against Paimon 1.3. This lookuper configures exactly one temporary directory. + public String pickTempDir() { + return delegate.tempDirs()[0]; + } + @Override public FileIOChannel.Enumerator createChannelEnumerator() { return delegate.createChannelEnumerator(); diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLocalTableQuery.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLocalTableQuery.java new file mode 100644 index 0000000000..e43d83df02 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLocalTableQuery.java @@ -0,0 +1,187 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.lake.paimon.lookup; + +import org.apache.fluss.lake.paimon.utils.PaimonPartitionBucket; + +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.query.LocalTableQuery; +import org.apache.paimon.table.source.DataSplit; +import org.apache.paimon.table.source.InnerTableScan; +import org.apache.paimon.table.source.Split; + +import javax.annotation.Nullable; + +import java.io.Closeable; +import java.io.IOException; +import java.lang.reflect.Method; +import java.lang.reflect.Modifier; +import java.util.ArrayList; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** + * Adapts Paimon's {@link LocalTableQuery} to Fluss's long-lived historical lookup lifecycle. + * + *

{@code LocalTableQuery} does not discover source files by itself. Fluss must register the + * files for each partition-bucket before its first lookup and refresh that registration when + * compaction and snapshot expiration make a cached file stale. This class owns that file state and + * retries a failed lookup once, while ensuring that concurrent failures trigger only one refresh. + * + *

It also isolates synchronization differences between supported Paimon versions. Paimon 1.3 + * synchronizes lookup on the {@code LocalTableQuery} instance, so file registration must use the + * same monitor. Paimon 2.0 supports concurrent lookup and coordinates access by bucket, allowing + * registration to use a per-bucket lock instead. Keeping these concerns here lets {@link + * PaimonLakeTableLookuper} focus on Fluss key, partition, and value conversion. + */ +class PaimonLocalTableQuery implements Closeable { + + private static final boolean CONCURRENT_LOOKUP_SUPPORTED = supportsConcurrentLookup(); + + private final FileStoreTable fileStoreTable; + private final LocalTableQuery localTableQuery; + private final Map bucketStates; + + public PaimonLocalTableQuery(FileStoreTable fileStoreTable, LocalTableQuery localTableQuery) { + this.fileStoreTable = checkNotNull(fileStoreTable, "fileStoreTable must not be null."); + this.localTableQuery = checkNotNull(localTableQuery, "localTableQuery must not be null."); + this.bucketStates = new ConcurrentHashMap<>(); + } + + final @Nullable InternalRow lookup(BinaryRow partition, int bucket, InternalRow key) + throws IOException { + List filesBeforeLookup = initializeFiles(partition, bucket); + try { + return localTableQuery.lookup(partition, bucket, key); + } catch (IOException firstError) { + refreshFiles(partition, bucket, filesBeforeLookup); + try { + return localTableQuery.lookup(partition, bucket, key); + } catch (IOException retryError) { + retryError.addSuppressed(firstError); + throw retryError; + } + } + } + + private List initializeFiles(BinaryRow partition, int bucket) { + PaimonPartitionBucket partitionBucket = new PaimonPartitionBucket(partition, bucket); + BucketState bucketState = + bucketStates.computeIfAbsent(partitionBucket, ignored -> new BucketState()); + List files = bucketState.files; + if (files != null) { + return files; + } + + Object lockScope = fileRegistrationLock(bucketState); + synchronized (lockScope) { + files = bucketState.files; + if (files == null) { + files = registerFiles(partition, bucket, Collections.emptyList()); + bucketState.files = files; + } + return files; + } + } + + private void refreshFiles( + BinaryRow partition, int bucket, List filesBeforeLookup) { + PaimonPartitionBucket partitionBucket = new PaimonPartitionBucket(partition, bucket); + BucketState bucketState = + checkNotNull( + bucketStates.get(partitionBucket), + "Partition-bucket files must be initialized."); + if (bucketState.files != filesBeforeLookup) { + return; + } + + Object lockScope = fileRegistrationLock(bucketState); + synchronized (lockScope) { + if (bucketState.files != filesBeforeLookup) { + return; + } + bucketState.files = registerFiles(partition, bucket, filesBeforeLookup); + } + } + + Object fileRegistrationLock(BucketState bucketState) { + if (CONCURRENT_LOOKUP_SUPPORTED) { + return bucketState; + } + return localTableQuery; + } + + @Override + public final void close() throws IOException { + try { + localTableQuery.close(); + } finally { + bucketStates.clear(); + } + } + + private List registerFiles( + BinaryRow partition, int bucket, List filesBeforeRefresh) { + List latestFiles = scanDataFiles(partition, bucket); + localTableQuery.refreshFiles(partition, bucket, filesBeforeRefresh, latestFiles); + return latestFiles; + } + + final List scanDataFiles(BinaryRow partition, int bucket) { + LinkedHashMap dataFilesByName = new LinkedHashMap<>(); + InnerTableScan tableScan = + fileStoreTable + .newScan() + .withPartitionFilter(Collections.singletonList(partition)) + .withBucket(bucket); + for (Split split : tableScan.plan().splits()) { + if (split instanceof DataSplit) { + for (DataFileMeta file : ((DataSplit) split).dataFiles()) { + dataFilesByName.put(file.fileName(), file); + } + } + } + return Collections.unmodifiableList(new ArrayList<>(dataFilesByName.values())); + } + + private static boolean supportsConcurrentLookup() { + try { + Method lookupMethod = + LocalTableQuery.class.getMethod( + "lookup", BinaryRow.class, int.class, InternalRow.class); + return !Modifier.isSynchronized(lookupMethod.getModifiers()); + } catch (NoSuchMethodException e) { + throw new IllegalStateException( + "Unsupported Paimon LocalTableQuery lookup signature.", e); + } + } + + private static final class BucketState { + // Mirrors the file membership retained by LocalTableQuery. Disk lookup-file eviction does + // not remove that query state, so this list remains the next refreshFiles' beforeFiles. + private volatile @Nullable List files; + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java index 1bf0ec3a16..bba761c0c5 100644 --- a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/PaimonLakeTableLookuperTest.java @@ -64,7 +64,13 @@ import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; @@ -345,6 +351,107 @@ void testLookupKvFormatV2WithNonDefaultBucketKey() throws Exception { } } + @Test + void testConcurrentLookupWithRequestScopedEncoders() throws Exception { + TablePath tablePath = TablePath.of(DB, "concurrent_non_default_bucket_key"); + Schema schema = + Schema.newBuilder() + .column("id", DataTypes.INT()) + .column("sub_id", DataTypes.STRING()) + .column("dt", DataTypes.STRING()) + .column("name", DataTypes.STRING()) + .primaryKey("id", "sub_id", "dt") + .build(); + TableDescriptor tableDescriptor = + TableDescriptor.builder() + .schema(schema) + .partitionedBy("dt") + .distributedBy(1, "id") + .build(); + FileStoreTable table = createPaimonTable(tablePath, tableDescriptor); + writeAndCommitData( + table, + Collections.singletonMap( + 0, + Arrays.asList( + paimonRow(1, "sub-1", "20240101", "Alice"), + paimonRow(2, "sub-2", "20240101", "Bob")))); + + List compactedKeys = Arrays.asList("id", "sub_id"); + CompactedKeyEncoder keyEncoder = + CompactedKeyEncoder.createKeyEncoder(schema.getRowType(), compactedKeys); + byte[][] keys = + new byte[][] { + keyEncoder.encodeKey(row(1, "sub-1", "20240101", "")), + keyEncoder.encodeKey(row(2, "sub-2", "20240101", "")) + }; + String[] subIds = new String[] {"sub-1", "sub-2"}; + String[] names = new String[] {"Alice", "Bob"}; + + try (LakeTableLookuper lookuper = + new PaimonLakeTableLookuper( + paimonConfig, + tablePath, + tempWarehouseDir.getAbsolutePath(), + tableConfig(KvFormat.COMPACTED, KV_FORMAT_VERSION_2), + LOOKUP_CACHE_MAX_DISK_BYTES, + NO_OP_DISK_WRITE_GUARD)) { + AtomicInteger lookupFileDownloads = new AtomicInteger(); + LakeTableLookuper.LookupContext context = + lookupContext( + schema, + "20240101", + 0, + SCHEMA_ID, + (lookupTimeNanos, lookupFileDownloaded) -> { + if (lookupFileDownloaded) { + lookupFileDownloads.incrementAndGet(); + } + }); + int threadCount = 8; + CountDownLatch start = new CountDownLatch(1); + ExecutorService executor = Executors.newFixedThreadPool(threadCount); + try { + List> futures = new ArrayList<>(); + for (int thread = 0; thread < threadCount; thread++) { + final int taskId = thread; + futures.add( + executor.submit( + () -> { + start.await(); + for (int lookup = 0; lookup < 50; lookup++) { + int rowIndex = (taskId + lookup) % keys.length; + BinaryValue value = + decodeValue( + lookuper.lookup( + keys[rowIndex], context), + SCHEMA_ID, + schema); + assertRow( + value.row, + rowIndex + 1, + subIds[rowIndex], + "20240101", + names[rowIndex]); + } + return null; + })); + } + + start.countDown(); + for (Future future : futures) { + future.get(30, TimeUnit.SECONDS); + } + } finally { + executor.shutdownNow(); + } + + // All concurrent requests query the same remote data file. Only the request that + // creates its local lookup file should be attributed with the download. + assertThat(lookupFileDownloads).hasValue(1); + } + } + @Test void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception { TablePath tablePath = TablePath.of(DB, "retry_initialization");