From e6aa09d23de31edac04815380d69ebb2e59691f9 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Wed, 19 Aug 2026 21:55:11 +0800 Subject: [PATCH 1/2] [server] Remove lookup lock for PaimonLakeTableLookuper --- .../lookup/PaimonLakeTableLookuper.java | 402 ++++++++++-------- .../lookup/PaimonLakeTableLookuperTest.java | 92 ++++ 2 files changed, 308 insertions(+), 186 deletions(-) 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..2ae15fe465 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 @@ -25,7 +25,6 @@ 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; @@ -62,10 +61,12 @@ 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 java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Supplier; import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; @@ -85,11 +86,17 @@ *

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. File scans and {@link LocalTableQuery#refreshFiles} are synchronized on the query + * instance to coordinate with Paimon 1.3's synchronized lookup. Paimon 2.0 does not acquire that + * monitor for lookup and coordinates lookup with refresh through its internal bucket locks. + * + *

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,25 +107,11 @@ public class PaimonLakeTableLookuper implements LakeTableLookuper { private final long lookupCacheMaxDiskBytes; private final Runnable diskWriteGuard; - private final Set initializedBuckets; - - 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 CompactedKeyDecoder compactedKeyDecoder; - private @Nullable PaimonKeyEncoder paimonKeyEncoder; - private boolean hasCachedValueEncoder; - private short cachedValueSchemaId; - private @Nullable RowEncoder cachedValueRowEncoder; - private @Nullable InternalRow.FieldGetter[] cachedValueFieldGetters; + private final AtomicLong lookupFileDownloadCount; + private final Object initializationLock; + + private volatile @Nullable QueryState queryState; + // Guarded by initializationLock. private boolean closed; /** Creates a lookuper with the specified local lookup cache limit. */ @@ -137,29 +130,23 @@ 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.lookupFileDownloadCount = new AtomicLong(); + 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()); + QueryState state = 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()); + LookupRow lookupRow = createLookupRow(state, key, context); + initializeFilesIfNeeded(state, lookupRow.partition, context.bucketId()); - long downloadCountBeforeLookup = lookupFileDownloadCount; + long downloadCountBeforeLookup = lookupFileDownloadCount.get(); long lookupStartNanos = System.nanoTime(); - org.apache.paimon.data.InternalRow paimonRow; try { - paimonRow = - lookupWithFileRefresh( - partition, context.bucketId(), keyRow, context.valueRowType()); + return lookupWithFileRefresh(state, lookupRow, context); } catch (Exception e) { DiskWriteLockedException diskWriteLockedException = ExceptionUtils.findThrowable(e, DiskWriteLockedException.class).orElse(null); @@ -168,30 +155,27 @@ public PaimonLakeTableLookuper( } throw e; } finally { + // TODO: Attribute downloads to the request that triggered them. The global counter is + // thread-safe but may report another concurrent request's download for this lookup. context.lookupMetricRecorder() .recordLookup( System.nanoTime() - lookupStartNanos, - // An increase means this lookup downloaded at least one lookup file - // through the tracking IO manager. - lookupFileDownloadCount > downloadCountBeforeLookup); - } - if (paimonRow == null) { - return null; + lookupFileDownloadCount.get() > downloadCountBeforeLookup); } - 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; + if (queryState != null) { + queryState.close(); + queryState = 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() { @@ -200,11 +184,18 @@ private void checkNotClosed() { } } - private void ensureInitialized(RowType valueRowType) throws Exception { - if (localTableQuery != null) { - return; + private QueryState ensureInitialized(RowType valueRowType) throws Exception { + if (queryState == null) { + synchronized (initializationLock) { + if (queryState == null) { + queryState = createQueryState(valueRowType); + } + } } + return queryState; + } + private QueryState createQueryState(RowType valueRowType) throws Exception { Catalog newCatalog = null; IOManager newIOManager = null; LocalTableQuery newLocalTableQuery = null; @@ -221,16 +212,10 @@ 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 trimmedPrimaryKeys = + 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 @@ -242,21 +227,25 @@ private void ensureInitialized(RowType valueRowType) throws Exception { // 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); } - // Publish the newly created state only after every initialization step succeeds. - catalog = newCatalog; - fileStoreTable = newFileStoreTable; - ioManager = newIOManager; - partitionKeyExtractor = newPartitionKeyExtractor; - primaryKeyFieldCount = newPrimaryKeyFieldCount; - compactedKeyDecoder = newCompactedKeyDecoder; - paimonKeyEncoder = newPaimonKeyEncoder; - // localTableQuery is the initialization marker, so publish it last. - localTableQuery = newLocalTableQuery; + newIOManager = createIOManager(ioTmpDir); + newLocalTableQuery = + newFileStoreTable + .newLocalTableQuery() + .withValueProjection(businessFieldProjection(newFileStoreTable)) + .withIOManager(newIOManager); + + QueryState newQueryState = + new QueryState( + newCatalog, + newFileStoreTable, + newIOManager, + newLocalTableQuery, + trimmedPrimaryKeys, + newCompactedKeyDecoder); initialized = true; + return newQueryState; } finally { if (!initialized) { IOUtils.closeQuietly(newLocalTableQuery, "Paimon local table query"); @@ -292,88 +281,69 @@ 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 LookupRow createLookupRow(QueryState state, byte[] key, LookupContext context) { + // Both generated helpers reuse mutable writers or projections, so keep them confined to + // this lookup call. + RowPartitionKeyExtractor partitionKeyExtractor = + new RowPartitionKeyExtractor(state.fileStoreTable.schema()); + org.apache.paimon.data.BinaryRow partition = + toPaimonPartition( + context.partitionSpec(), + context.valueRowType(), + state.fileStoreTable.schema().logicalRowType(), + partitionKeyExtractor::partition) + .copy(); - private org.apache.paimon.data.BinaryRow toPaimonLookupKey(byte[] key) { 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); + if (state.compactedKeyDecoder != null) { + InternalRow decodedKey = state.compactedKeyDecoder.decodeKey(key); + RowType keyRowType = context.valueRowType().project(state.trimmedPrimaryKeys); + PaimonKeyEncoder paimonKeyEncoder = + new PaimonKeyEncoder(keyRowType, state.trimmedPrimaryKeys); + paimonKey = paimonKeyEncoder.encodeKey(decodedKey); } org.apache.paimon.data.BinaryRow keyRow = - new org.apache.paimon.data.BinaryRow(primaryKeyFieldCount); + new org.apache.paimon.data.BinaryRow(state.trimmedPrimaryKeys.size()); keyRow.pointTo(MemorySegment.wrap(paimonKey), 0, paimonKey.length); - return keyRow; + return new LookupRow(partition, keyRow); } - private void initializeFilesIfNeeded(org.apache.paimon.data.BinaryRow partition, int bucketId) { + private void initializeFilesIfNeeded( + QueryState state, 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; + Supplier exists = () -> state.initializedBucketFiles.containsKey(partitionBucket); + if (!exists.get()) { + synchronized (state.localTableQuery) { + if (!exists.get()) { + List currentFiles = + scanCurrentDataFiles(state.fileStoreTable, partition, bucketId); + state.localTableQuery.refreshFiles( + partition, bucketId, Collections.emptyList(), currentFiles); + // Publish only after Paimon accepted the complete file set. + state.initializedBucketFiles.put(partitionBucket, currentFiles); + } } - 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[] lookupWithFileRefresh( + QueryState state, LookupRow lookupRow, LookupContext context) throws Exception { + int bucketId = context.bucketId(); + PaimonPartitionBucket partitionBucket = + new PaimonPartitionBucket(lookupRow.partition, bucketId); + List filesBeforeLookup = registeredFiles(state, partitionBucket); try { - return localTableQuery().lookup(partition, bucketId, keyRow); + return lookupAndEncode(state, lookupRow, 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. + // do not repeatedly refresh Paimon lookup state within one request. try { - refreshFiles(partition, bucketId, valueRowType); - return localTableQuery().lookup(partition, bucketId, keyRow); + refreshFilesIfUnchanged( + state, lookupRow.partition, bucketId, partitionBucket, filesBeforeLookup); + return lookupAndEncode(state, lookupRow, context); } catch (IOException retryError) { retryError.addSuppressed(e); // Historical Paimon point lookup is part of the Fluss KV lookup path. Expose a @@ -388,23 +358,59 @@ private org.apache.paimon.data.InternalRow lookupWithFileRefresh( } } - 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 @Nullable byte[] lookupAndEncode( + QueryState state, LookupRow lookupRow, LookupContext context) throws IOException { + org.apache.paimon.data.InternalRow paimonRow = + state.localTableQuery.lookup( + lookupRow.partition, context.bucketId(), lookupRow.keyRow); + if (paimonRow == null) { + return null; + } + return encodeValue(paimonRow, context.schemaId(), context.valueRowType()); + } + + private List registeredFiles( + QueryState state, PaimonPartitionBucket partitionBucket) { + return checkNotNull( + state.initializedBucketFiles.get(partitionBucket), + "Partition-bucket files must be initialized."); + } + + private void refreshFilesIfUnchanged( + QueryState state, + org.apache.paimon.data.BinaryRow partition, + int bucketId, + PaimonPartitionBucket partitionBucket, + List filesBeforeLookup) { + synchronized (state.localTableQuery) { + if (state.initializedBucketFiles.get(partitionBucket) != filesBeforeLookup) { + return; + } + List latestFiles = + scanCurrentDataFiles(state.fileStoreTable, partition, bucketId); + state.localTableQuery.refreshFiles(partition, bucketId, filesBeforeLookup, latestFiles); + // The immutable list reference is also the bucket's refresh version. Publishing a new + // reference lets concurrent failures observe that refresh has already completed. + state.initializedBucketFiles.put(partitionBucket, latestFiles); + } + } + + private static List scanCurrentDataFiles( + FileStoreTable fileStoreTable, + org.apache.paimon.data.BinaryRow partition, + int bucketId) { + LinkedHashMap dataFilesByName = new LinkedHashMap<>(); + InnerTableScan tableScan = + fileStoreTable + .newScan() + .withPartitionFilter(Collections.singletonList(partition)) + .withBucket(bucketId); + for (Split split : tableScan.plan().splits()) { + if (split instanceof DataSplit) { + addFilesByName(dataFilesByName, ((DataSplit) split).dataFiles()); + } + } + return Collections.unmodifiableList(new ArrayList<>(dataFilesByName.values())); } private static void addFilesByName( @@ -417,14 +423,8 @@ private static void addFilesByName( 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,28 +436,52 @@ private byte[] encodeValue( } } - private void ensureValueEncoder(short schemaId, RowType valueRowType) { - if (hasCachedValueEncoder && cachedValueSchemaId == schemaId) { - return; - } + private static final class LookupRow { + private final org.apache.paimon.data.BinaryRow partition; + private final org.apache.paimon.data.BinaryRow keyRow; - 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 LookupRow( + org.apache.paimon.data.BinaryRow partition, + org.apache.paimon.data.BinaryRow keyRow) { + this.partition = partition; + this.keyRow = keyRow; + } } - private LocalTableQuery localTableQuery() { - return checkNotNull(localTableQuery, "localTableQuery must be initialized."); - } + private static final class QueryState { + private final Catalog catalog; + private final FileStoreTable fileStoreTable; + private final IOManager ioManager; + private final LocalTableQuery localTableQuery; + private final List trimmedPrimaryKeys; + private final Map> initializedBucketFiles; + + // CompactedKeyDecoder contains immutable type metadata and creates all decode state per + // invocation, so it can be shared by concurrent lookups. + private final @Nullable CompactedKeyDecoder compactedKeyDecoder; + + private QueryState( + Catalog catalog, + FileStoreTable fileStoreTable, + IOManager ioManager, + LocalTableQuery localTableQuery, + List trimmedPrimaryKeys, + @Nullable CompactedKeyDecoder compactedKeyDecoder) { + this.catalog = catalog; + this.fileStoreTable = fileStoreTable; + this.ioManager = ioManager; + this.localTableQuery = localTableQuery; + this.trimmedPrimaryKeys = trimmedPrimaryKeys; + this.initializedBucketFiles = new ConcurrentHashMap<>(); + this.compactedKeyDecoder = compactedKeyDecoder; + } - private RowPartitionKeyExtractor partitionKeyExtractor() { - return checkNotNull(partitionKeyExtractor, "partitionKeyExtractor must be initialized."); + private void close() { + IOUtils.closeQuietly(localTableQuery, "Paimon local table query"); + IOUtils.closeQuietly(ioManager, "Paimon lookup IO manager"); + IOUtils.closeQuietly(catalog, "Paimon catalog"); + initializedBucketFiles.clear(); + } } /** Tracks creation of Paimon lookup files while delegating all local I/O operations. */ @@ -483,7 +507,7 @@ 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++; + lookupFileDownloadCount.incrementAndGet(); return delegate.createChannel(prefix); } @@ -492,6 +516,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/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..8473172c05 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,6 +64,11 @@ 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 static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; @@ -345,6 +350,93 @@ 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)) { + LakeTableLookuper.LookupContext context = + lookupContext(schema, "20240101", 0, SCHEMA_ID); + 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(); + } + } + } + @Test void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception { TablePath tablePath = TablePath.of(DB, "retry_initialization"); From 471a215b1dba2da2887ba76bb70806723d3a18f3 Mon Sep 17 00:00:00 2001 From: zhangjunfan Date: Thu, 20 Aug 2026 11:51:00 +0800 Subject: [PATCH 2/2] better refactor --- .../lookup/PaimonLakeTableLookuper.java | 291 +++++------------- .../paimon/lookup/PaimonLocalTableQuery.java | 187 +++++++++++ .../lookup/PaimonLakeTableLookuperTest.java | 17 +- 3 files changed, 283 insertions(+), 212 deletions(-) create mode 100644 fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/PaimonLocalTableQuery.java 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 2ae15fe465..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,7 +23,6 @@ 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.TablePath; import org.apache.fluss.row.BinaryRow; @@ -44,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; @@ -61,12 +56,7 @@ import java.io.UncheckedIOException; 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 java.util.concurrent.atomic.AtomicLong; -import java.util.function.Supplier; import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; @@ -91,9 +81,8 @@ * *

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. File scans and {@link LocalTableQuery#refreshFiles} are synchronized on the query - * instance to coordinate with Paimon 1.3's synchronized lookup. Paimon 2.0 does not acquire that - * monitor for lookup and coordinates lookup with refresh through its internal bucket locks. + * 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. @@ -107,12 +96,21 @@ public class PaimonLakeTableLookuper implements LakeTableLookuper { private final long lookupCacheMaxDiskBytes; private final Runnable diskWriteGuard; - private final AtomicLong lookupFileDownloadCount; + private final ThreadLocal lookupFileDownloaded; private final Object initializationLock; - private volatile @Nullable QueryState queryState; + private @Nullable Catalog catalog; + private @Nullable FileStoreTable fileStoreTable; + private @Nullable IOManager ioManager; + 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 volatile @Nullable PaimonLocalTableQuery localTableQuery; // Guarded by initializationLock. - private boolean closed; + private volatile boolean closed; /** Creates a lookuper with the specified local lookup cache limit. */ public PaimonLakeTableLookuper( @@ -130,7 +128,7 @@ public PaimonLakeTableLookuper( lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0."); this.lookupCacheMaxDiskBytes = lookupCacheMaxDiskBytes; this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); - this.lookupFileDownloadCount = new AtomicLong(); + this.lookupFileDownloaded = new ThreadLocal<>(); this.initializationLock = new Object(); } @@ -138,15 +136,13 @@ public PaimonLakeTableLookuper( public @Nullable byte[] lookup(byte[] key, LookupContext context) throws Exception { checkNotNull(key, "key must not be null."); checkNotNull(context, "context must not be null."); - QueryState state = ensureInitialized(context.valueRowType()); - - LookupRow lookupRow = createLookupRow(state, key, context); - initializeFilesIfNeeded(state, lookupRow.partition, context.bucketId()); + checkNotClosed(); + ensureInitialized(context.valueRowType()); - long downloadCountBeforeLookup = lookupFileDownloadCount.get(); + lookupFileDownloaded.set(false); long lookupStartNanos = System.nanoTime(); try { - return lookupWithFileRefresh(state, lookupRow, context); + return lookupInternal(key, context); } catch (Exception e) { DiskWriteLockedException diskWriteLockedException = ExceptionUtils.findThrowable(e, DiskWriteLockedException.class).orElse(null); @@ -155,12 +151,10 @@ public PaimonLakeTableLookuper( } throw e; } finally { - // TODO: Attribute downloads to the request that triggered them. The global counter is - // thread-safe but may report another concurrent request's download for this lookup. + boolean fileDownloaded = lookupFileDownloaded.get(); + lookupFileDownloaded.remove(); context.lookupMetricRecorder() - .recordLookup( - System.nanoTime() - lookupStartNanos, - lookupFileDownloadCount.get() > downloadCountBeforeLookup); + .recordLookup(System.nanoTime() - lookupStartNanos, fileDownloaded); } } @@ -171,10 +165,15 @@ public void close() { return; } closed = true; - if (queryState != null) { - queryState.close(); - queryState = null; - } + 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; } } @@ -184,18 +183,17 @@ private void checkNotClosed() { } } - private QueryState ensureInitialized(RowType valueRowType) throws Exception { - if (queryState == null) { + private void ensureInitialized(RowType valueRowType) throws Exception { + if (localTableQuery == null) { synchronized (initializationLock) { - if (queryState == null) { - queryState = createQueryState(valueRowType); + if (localTableQuery == null) { + initialize(valueRowType); } } } - return queryState; } - private QueryState createQueryState(RowType valueRowType) throws Exception { + private void initialize(RowType valueRowType) throws Exception { Catalog newCatalog = null; IOManager newIOManager = null; LocalTableQuery newLocalTableQuery = null; @@ -212,7 +210,7 @@ private QueryState createQueryState(RowType valueRowType) throws Exception { "Point lookup is only supported for primary-key Paimon tables."); } - List trimmedPrimaryKeys = + List newTrimmedPrimaryKeys = Collections.unmodifiableList( new ArrayList<>(newFileStoreTable.schema().trimmedPrimaryKeys())); CompactedKeyDecoder newCompactedKeyDecoder = null; @@ -221,12 +219,12 @@ private QueryState createQueryState(RowType valueRowType) throws Exception { // 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); + CompactedKeyDecoder.createKeyDecoder(valueRowType, newTrimmedPrimaryKeys); } newIOManager = createIOManager(ioTmpDir); @@ -236,16 +234,16 @@ private QueryState createQueryState(RowType valueRowType) throws Exception { .withValueProjection(businessFieldProjection(newFileStoreTable)) .withIOManager(newIOManager); - QueryState newQueryState = - new QueryState( - newCatalog, - newFileStoreTable, - newIOManager, - newLocalTableQuery, - trimmedPrimaryKeys, - newCompactedKeyDecoder); + PaimonLocalTableQuery newLookupEngine = + new PaimonLocalTableQuery(newFileStoreTable, newLocalTableQuery); + catalog = newCatalog; + fileStoreTable = newFileStoreTable; + ioManager = newIOManager; + trimmedPrimaryKeys = newTrimmedPrimaryKeys; + compactedKeyDecoder = newCompactedKeyDecoder; + // Keep this volatile write last to publish all initialized fields together. + localTableQuery = newLookupEngine; initialized = true; - return newQueryState; } finally { if (!initialized) { IOUtils.closeQuietly(newLocalTableQuery, "Paimon local table query"); @@ -281,145 +279,59 @@ private static int[] businessFieldProjection(FileStoreTable fileStoreTable) { return projection; } - private LookupRow createLookupRow(QueryState state, byte[] key, LookupContext context) { + 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(state.fileStoreTable.schema()); + new RowPartitionKeyExtractor(fileStoreTable.schema()); org.apache.paimon.data.BinaryRow partition = toPaimonPartition( context.partitionSpec(), context.valueRowType(), - state.fileStoreTable.schema().logicalRowType(), + fileStoreTable.schema().logicalRowType(), partitionKeyExtractor::partition) .copy(); + return partition; + } + private org.apache.paimon.data.BinaryRow getKey(byte[] key, LookupContext context) { byte[] paimonKey = key; - if (state.compactedKeyDecoder != null) { - InternalRow decodedKey = state.compactedKeyDecoder.decodeKey(key); - RowType keyRowType = context.valueRowType().project(state.trimmedPrimaryKeys); + if (compactedKeyDecoder != null) { + InternalRow decodedKey = compactedKeyDecoder.decodeKey(key); + RowType keyRowType = context.valueRowType().project(trimmedPrimaryKeys); PaimonKeyEncoder paimonKeyEncoder = - new PaimonKeyEncoder(keyRowType, state.trimmedPrimaryKeys); + new PaimonKeyEncoder(keyRowType, trimmedPrimaryKeys); paimonKey = paimonKeyEncoder.encodeKey(decodedKey); } org.apache.paimon.data.BinaryRow keyRow = - new org.apache.paimon.data.BinaryRow(state.trimmedPrimaryKeys.size()); + new org.apache.paimon.data.BinaryRow(trimmedPrimaryKeys.size()); keyRow.pointTo(MemorySegment.wrap(paimonKey), 0, paimonKey.length); - return new LookupRow(partition, keyRow); + return keyRow; } - private void initializeFilesIfNeeded( - QueryState state, org.apache.paimon.data.BinaryRow partition, int bucketId) { - PaimonPartitionBucket partitionBucket = new PaimonPartitionBucket(partition, bucketId); - Supplier exists = () -> state.initializedBucketFiles.containsKey(partitionBucket); - if (!exists.get()) { - synchronized (state.localTableQuery) { - if (!exists.get()) { - List currentFiles = - scanCurrentDataFiles(state.fileStoreTable, partition, bucketId); - state.localTableQuery.refreshFiles( - partition, bucketId, Collections.emptyList(), currentFiles); - // Publish only after Paimon accepted the complete file set. - state.initializedBucketFiles.put(partitionBucket, currentFiles); - } - } - } - } - - private @Nullable byte[] lookupWithFileRefresh( - QueryState state, LookupRow lookupRow, LookupContext context) throws Exception { - int bucketId = context.bucketId(); - PaimonPartitionBucket partitionBucket = - new PaimonPartitionBucket(lookupRow.partition, bucketId); - List filesBeforeLookup = registeredFiles(state, partitionBucket); + private @Nullable byte[] lookupInternal(byte[] key, LookupContext context) { + org.apache.paimon.data.InternalRow paimonRow; try { - return lookupAndEncode(state, lookupRow, context); + 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 refresh Paimon lookup state within one request. - try { - refreshFilesIfUnchanged( - state, lookupRow.partition, bucketId, partitionBucket, filesBeforeLookup); - return lookupAndEncode(state, lookupRow, context); - } 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 @Nullable byte[] lookupAndEncode( - QueryState state, LookupRow lookupRow, LookupContext context) throws IOException { - org.apache.paimon.data.InternalRow paimonRow = - state.localTableQuery.lookup( - lookupRow.partition, context.bucketId(), lookupRow.keyRow); if (paimonRow == null) { return null; } return encodeValue(paimonRow, context.schemaId(), context.valueRowType()); } - private List registeredFiles( - QueryState state, PaimonPartitionBucket partitionBucket) { - return checkNotNull( - state.initializedBucketFiles.get(partitionBucket), - "Partition-bucket files must be initialized."); - } - - private void refreshFilesIfUnchanged( - QueryState state, - org.apache.paimon.data.BinaryRow partition, - int bucketId, - PaimonPartitionBucket partitionBucket, - List filesBeforeLookup) { - synchronized (state.localTableQuery) { - if (state.initializedBucketFiles.get(partitionBucket) != filesBeforeLookup) { - return; - } - List latestFiles = - scanCurrentDataFiles(state.fileStoreTable, partition, bucketId); - state.localTableQuery.refreshFiles(partition, bucketId, filesBeforeLookup, latestFiles); - // The immutable list reference is also the bucket's refresh version. Publishing a new - // reference lets concurrent failures observe that refresh has already completed. - state.initializedBucketFiles.put(partitionBucket, latestFiles); - } - } - - private static List scanCurrentDataFiles( - FileStoreTable fileStoreTable, - org.apache.paimon.data.BinaryRow partition, - int bucketId) { - LinkedHashMap dataFilesByName = new LinkedHashMap<>(); - InnerTableScan tableScan = - fileStoreTable - .newScan() - .withPartitionFilter(Collections.singletonList(partition)) - .withBucket(bucketId); - for (Split split : tableScan.plan().splits()) { - if (split instanceof DataSplit) { - addFilesByName(dataFilesByName, ((DataSplit) split).dataFiles()); - } - } - return Collections.unmodifiableList(new ArrayList<>(dataFilesByName.values())); - } - - private static void addFilesByName( - LinkedHashMap filesByName, List files) { - for (DataFileMeta file : files) { - filesByName.put(file.fileName(), file); - } - } - private byte[] encodeValue( org.apache.paimon.data.InternalRow paimonRow, short schemaId, RowType valueRowType) { PaimonRowAsFlussRow flussRow = new PaimonRowAsFlussRow(paimonRow); @@ -436,54 +348,6 @@ private byte[] encodeValue( } } - private static final class LookupRow { - private final org.apache.paimon.data.BinaryRow partition; - private final org.apache.paimon.data.BinaryRow keyRow; - - private LookupRow( - org.apache.paimon.data.BinaryRow partition, - org.apache.paimon.data.BinaryRow keyRow) { - this.partition = partition; - this.keyRow = keyRow; - } - } - - private static final class QueryState { - private final Catalog catalog; - private final FileStoreTable fileStoreTable; - private final IOManager ioManager; - private final LocalTableQuery localTableQuery; - private final List trimmedPrimaryKeys; - private final Map> initializedBucketFiles; - - // CompactedKeyDecoder contains immutable type metadata and creates all decode state per - // invocation, so it can be shared by concurrent lookups. - private final @Nullable CompactedKeyDecoder compactedKeyDecoder; - - private QueryState( - Catalog catalog, - FileStoreTable fileStoreTable, - IOManager ioManager, - LocalTableQuery localTableQuery, - List trimmedPrimaryKeys, - @Nullable CompactedKeyDecoder compactedKeyDecoder) { - this.catalog = catalog; - this.fileStoreTable = fileStoreTable; - this.ioManager = ioManager; - this.localTableQuery = localTableQuery; - this.trimmedPrimaryKeys = trimmedPrimaryKeys; - this.initializedBucketFiles = new ConcurrentHashMap<>(); - this.compactedKeyDecoder = compactedKeyDecoder; - } - - private void close() { - IOUtils.closeQuietly(localTableQuery, "Paimon local table query"); - IOUtils.closeQuietly(ioManager, "Paimon lookup IO manager"); - IOUtils.closeQuietly(catalog, "Paimon catalog"); - initializedBucketFiles.clear(); - } - } - /** Tracks creation of Paimon lookup files while delegating all local I/O operations. */ private final class TrackingIOManager implements IOManager { @@ -507,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.incrementAndGet(); - 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 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 8473172c05..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 @@ -70,6 +70,7 @@ 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; @@ -395,8 +396,18 @@ void testConcurrentLookupWithRequestScopedEncoders() throws Exception { 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); + 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); @@ -434,6 +445,10 @@ void testConcurrentLookupWithRequestScopedEncoders() throws Exception { } 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); } }