diff --git a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java index bb19ec7fd65..f098717bfc8 100644 --- a/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java +++ b/fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java @@ -428,7 +428,7 @@ public class ConfigOptions { .defaultValue(0.10) .withDescription( "The maximum fraction of the total capacity of the volume containing the first available data directory allocated to historical partition lookup caches on a TabletServer. " - + "Up to ten table lookupers are cached, and each receives one tenth of this capacity. Historical lookup cache files are stored under that data directory; additional data volumes are not used. " + + "Up to ten table lookupers share this capacity. Historical lookup cache files are stored under that data directory; additional data volumes are not used. " + "The valid range is (0.0, 1.0]."); public static final ConfigOption diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java index 446aa1184c4..bba509188f6 100644 --- a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java +++ b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeStorage.java @@ -18,14 +18,10 @@ package org.apache.fluss.lake.lakestorage; import org.apache.fluss.annotation.PublicEvolving; -import org.apache.fluss.config.TableConfig; import org.apache.fluss.lake.source.LakeSource; import org.apache.fluss.lake.writer.LakeTieringFactory; import org.apache.fluss.metadata.TablePath; -import static org.apache.fluss.utils.Preconditions.checkArgument; -import static org.apache.fluss.utils.Preconditions.checkNotNull; - /** * The LakeStorage interface defines how to implement lakehouse storage system such as Paimon and * Iceberg. It provides a method to create a lake tiering factory. @@ -56,64 +52,15 @@ public interface LakeStorage { LakeSource createLakeSource(TablePath tablePath); /** - * Creates a table-level point lookuper for the specified lake table. + * Creates a TabletServer-scoped runtime for lake table point lookup. * - * @param tablePath the logical path identifying the table in the lakehouse storage - * @param context runtime context for creating the lookuper - * @return a table-level point lookuper + * @param ioTmpDir local directory shared by lookupers for temporary files + * @param lookupCacheMaxDiskBytes maximum local lookup cache size in bytes + * @return the lookup runtime */ - default LakeTableLookuper createLakeTableLookuper( - TablePath tablePath, LookuperContext context) { + default LakeTableLookupRuntime createLakeTableLookupRuntime( + String ioTmpDir, long lookupCacheMaxDiskBytes) { throw new UnsupportedOperationException( "Point lookup is not supported for this lake storage."); } - - /** Runtime context for creating a lake table lookuper. */ - final class LookuperContext { - private final String ioTmpDir; - private final TableConfig tableConfig; - private final long lookupCacheMaxDiskBytes; - private final Runnable diskWriteGuard; - - /** - * Creates a lookuper context. - * - * @param ioTmpDir local directory for temporary files used by the lookuper - * @param tableConfig configuration of the Fluss table - * @param lookupCacheMaxDiskBytes maximum local lookup cache size in bytes - * @param diskWriteGuard guard invoked before creating a local lookup cache file - */ - public LookuperContext( - String ioTmpDir, - TableConfig tableConfig, - long lookupCacheMaxDiskBytes, - Runnable diskWriteGuard) { - this.ioTmpDir = checkNotNull(ioTmpDir, "ioTmpDir must not be null."); - this.tableConfig = checkNotNull(tableConfig, "tableConfig must not be null."); - checkArgument( - lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0."); - this.lookupCacheMaxDiskBytes = lookupCacheMaxDiskBytes; - this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); - } - - /** Returns the local directory for temporary files used by the lookuper. */ - public String ioTmpDir() { - return ioTmpDir; - } - - /** Returns the configuration of the Fluss table. */ - public TableConfig tableConfig() { - return tableConfig; - } - - /** Returns the maximum local lookup cache size in bytes. */ - public long lookupCacheMaxDiskBytes() { - return lookupCacheMaxDiskBytes; - } - - /** Returns the guard invoked before creating a local lookup cache file. */ - public Runnable diskWriteGuard() { - return diskWriteGuard; - } - } } diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookupRuntime.java b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookupRuntime.java new file mode 100644 index 00000000000..2eb41b2ee26 --- /dev/null +++ b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/LakeTableLookupRuntime.java @@ -0,0 +1,90 @@ +/* + * 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.lakestorage; + +import org.apache.fluss.annotation.PublicEvolving; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.config.TableConfig; +import org.apache.fluss.metadata.TablePath; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** TabletServer-scoped runtime for creating lake table lookupers. */ +@PublicEvolving +public interface LakeTableLookupRuntime extends AutoCloseable { + + /** + * Creates a table-level point lookuper for the specified lake table. + * + * @param tablePath the logical path identifying the table in the lakehouse storage + * @param context runtime context for creating the lookuper + * @return a table-level point lookuper + */ + LakeTableLookuper createLakeTableLookuper(TablePath tablePath, Context context); + + /** Updates the maximum local lookup cache size in bytes. */ + void updateLookupCacheMaxDiskBytes(long lookupCacheMaxDiskBytes); + + /** Runtime context for creating a lake table lookuper. */ + final class Context { + private final Configuration lakeConfiguration; + private final String cacheNamespace; + private final TableConfig tableConfig; + private final Runnable diskWriteGuard; + + /** + * Creates a lookuper context. + * + * @param lakeConfiguration configuration of the lake storage for this lookuper + * @param cacheNamespace namespace identifying cache entries owned by this lookuper + * @param tableConfig configuration of the Fluss table + * @param diskWriteGuard guard invoked before creating a local lookup cache file + */ + public Context( + Configuration lakeConfiguration, + String cacheNamespace, + TableConfig tableConfig, + Runnable diskWriteGuard) { + this.lakeConfiguration = + checkNotNull(lakeConfiguration, "lakeConfiguration must not be null."); + this.cacheNamespace = checkNotNull(cacheNamespace, "cacheNamespace must not be null."); + this.tableConfig = checkNotNull(tableConfig, "tableConfig must not be null."); + this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); + } + + /** Returns the lake storage configuration for this lookuper. */ + public Configuration lakeConfiguration() { + return lakeConfiguration; + } + + /** Returns the namespace identifying cache entries owned by this lookuper. */ + public String cacheNamespace() { + return cacheNamespace; + } + + /** Returns the configuration of the Fluss table. */ + public TableConfig tableConfig() { + return tableConfig; + } + + /** Returns the guard invoked before creating a local lookup cache file. */ + public Runnable diskWriteGuard() { + return diskWriteGuard; + } + } +} diff --git a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/PluginLakeStorageWrapper.java b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/PluginLakeStorageWrapper.java index 2f778bc0a90..dc16026a485 100644 --- a/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/PluginLakeStorageWrapper.java +++ b/fluss-common/src/main/java/org/apache/fluss/lake/lakestorage/PluginLakeStorageWrapper.java @@ -139,13 +139,54 @@ public LakeSource createLakeSource(TablePath tablePath) { } @Override - public LakeTableLookuper createLakeTableLookuper( - TablePath tablePath, LookuperContext context) { + public LakeTableLookupRuntime createLakeTableLookupRuntime( + String ioTmpDir, long lookupCacheMaxDiskBytes) { + try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(loader)) { + return new ClassLoaderFixingLakeTableLookupRuntime( + inner.createLakeTableLookupRuntime(ioTmpDir, lookupCacheMaxDiskBytes), + loader); + } + } + } + + static class ClassLoaderFixingLakeTableLookupRuntime + implements LakeTableLookupRuntime, WrappingProxy { + + private final LakeTableLookupRuntime inner; + private final ClassLoader loader; + + private ClassLoaderFixingLakeTableLookupRuntime( + LakeTableLookupRuntime inner, ClassLoader loader) { + this.inner = inner; + this.loader = loader; + } + + @Override + public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, Context context) { try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(loader)) { return new ClassLoaderFixingLakeTableLookuper( inner.createLakeTableLookuper(tablePath, context), loader); } } + + @Override + public void updateLookupCacheMaxDiskBytes(long lookupCacheMaxDiskBytes) { + try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(loader)) { + inner.updateLookupCacheMaxDiskBytes(lookupCacheMaxDiskBytes); + } + } + + @Override + public void close() throws Exception { + try (TemporaryClassLoaderContext ignored = TemporaryClassLoaderContext.of(loader)) { + inner.close(); + } + } + + @Override + public LakeTableLookupRuntime getWrappedDelegate() { + return inner; + } } static class ClassLoaderFixingLakeTableLookuper diff --git a/fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/LakeStorageTest.java b/fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/LakeStorageTest.java index 178ec37af2d..8b8a8b585b2 100644 --- a/fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/LakeStorageTest.java +++ b/fluss-common/src/test/java/org/apache/fluss/lake/lakestorage/LakeStorageTest.java @@ -91,6 +91,19 @@ void testWithPluginManager() throws Exception { ((PluginLakeStorageWrapper.ClassLoaderFixingLakeCatalog) lakeCatalog) .getWrappedDelegate()) .isInstanceOf(TestPaimonLakeCatalog.class); + + LakeTableLookupRuntime lookupRuntime = + lakeStorage.createLakeTableLookupRuntime("lookup-dir", 1024L); + assertThat(lookupRuntime) + .isInstanceOf( + PluginLakeStorageWrapper.ClassLoaderFixingLakeTableLookupRuntime.class); + TestLakeTableLookupRuntime innerLookupRuntime = + (TestLakeTableLookupRuntime) + ((PluginLakeStorageWrapper.ClassLoaderFixingLakeTableLookupRuntime) + lookupRuntime) + .getWrappedDelegate(); + lookupRuntime.close(); + assertThat(innerLookupRuntime.closed).isTrue(); } private static class TestingPluginManager implements PluginManager { @@ -124,7 +137,6 @@ public LakeStorage createLakeStorage(Configuration configuration) { } private static class TestPaimonLakeStorage implements LakeStorage { - public TestPaimonLakeStorage() {} @Override @@ -141,6 +153,30 @@ public TestPaimonLakeCatalog createLakeCatalog() { public LakeSource createLakeSource(TablePath tablePath) { throw new UnsupportedOperationException("Not implemented"); } + + @Override + public LakeTableLookupRuntime createLakeTableLookupRuntime( + String ioTmpDir, long lookupCacheMaxDiskBytes) { + return new TestLakeTableLookupRuntime(); + } + } + + private static class TestLakeTableLookupRuntime implements LakeTableLookupRuntime { + + private boolean closed; + + @Override + public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, Context context) { + throw new UnsupportedOperationException("Not implemented"); + } + + @Override + public void updateLookupCacheMaxDiskBytes(long lookupCacheMaxDiskBytes) {} + + @Override + public void close() { + closed = true; + } } private static class TestPaimonLakeCatalog implements LakeCatalog { diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java index 80e66398985..9cda75ffb56 100644 --- a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/PaimonLakeStorage.java @@ -19,8 +19,10 @@ import org.apache.fluss.config.Configuration; import org.apache.fluss.lake.lakestorage.LakeStorage; +import org.apache.fluss.lake.lakestorage.LakeTableLookupRuntime; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.paimon.lookup.PaimonLakeTableLookuper; +import org.apache.fluss.lake.paimon.lookup.SharedLookupFileCache; import org.apache.fluss.lake.paimon.source.PaimonLakeSource; import org.apache.fluss.lake.paimon.source.PaimonSplit; import org.apache.fluss.lake.paimon.tiering.PaimonCommittable; @@ -29,6 +31,14 @@ import org.apache.fluss.lake.source.LakeSource; import org.apache.fluss.lake.writer.LakeTieringFactory; import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.utils.IOUtils; + +import org.apache.paimon.CoreOptions; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.options.MemorySize; + +import static org.apache.fluss.utils.Preconditions.checkArgument; +import static org.apache.fluss.utils.Preconditions.checkNotNull; /** Paimon implementation of {@link LakeStorage}. */ public class PaimonLakeStorage implements LakeStorage { @@ -55,13 +65,50 @@ public LakeSource createLakeSource(TablePath tablePath) { } @Override - public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, LookuperContext context) { - return new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - context.ioTmpDir(), - context.tableConfig(), - context.lookupCacheMaxDiskBytes(), - context.diskWriteGuard()); + public LakeTableLookupRuntime createLakeTableLookupRuntime( + String ioTmpDir, long lookupCacheMaxDiskBytes) { + return new PaimonLakeTableLookupRuntime(ioTmpDir, lookupCacheMaxDiskBytes); + } + + /** Paimon lookup runtime sharing one I/O manager across table lookupers. */ + private static final class PaimonLakeTableLookupRuntime implements LakeTableLookupRuntime { + private final IOManager ioManager; + private final SharedLookupFileCache lookupFileCache; + + private PaimonLakeTableLookupRuntime(String ioTmpDir, long lookupCacheMaxDiskBytes) { + checkArgument( + lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0."); + this.ioManager = IOManager.create(checkNotNull(ioTmpDir, "ioTmpDir must not be null.")); + // ponytail: one runtime-wide retention; add a server option if this needs tuning. + this.lookupFileCache = + new SharedLookupFileCache( + CoreOptions.LOOKUP_CACHE_FILE_RETENTION.defaultValue(), + new MemorySize(lookupCacheMaxDiskBytes)); + } + + @Override + public LakeTableLookuper createLakeTableLookuper(TablePath tablePath, Context context) { + return new PaimonLakeTableLookuper( + new Configuration(context.lakeConfiguration()), + tablePath, + ioManager, + lookupFileCache, + context.cacheNamespace(), + context.tableConfig(), + context.diskWriteGuard()); + } + + @Override + public void updateLookupCacheMaxDiskBytes(long lookupCacheMaxDiskBytes) { + checkArgument( + lookupCacheMaxDiskBytes > 0, "lookupCacheMaxDiskBytes must be greater than 0."); + lookupFileCache.updateMaxDiskSize(new MemorySize(lookupCacheMaxDiskBytes)); + } + + @Override + public void close() { + IOUtils.closeQuietly(lookupFileCache, "shared Paimon lookup-file cache"); + IOUtils.closeQuietly(ioManager, "shared Paimon lookup IO manager"); + } } } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/NamespacedLookupFileCache.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/NamespacedLookupFileCache.java new file mode 100644 index 00000000000..0eb408101ff --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/NamespacedLookupFileCache.java @@ -0,0 +1,205 @@ +/* + * 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.paimon.mergetree.LookupFile; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Policy; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.stats.CacheStats; + +import javax.annotation.Nullable; + +import java.util.AbstractMap; +import java.util.HashSet; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentMap; +import java.util.function.Function; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; + +/** Namespace-scoped cache view delegating all storage to a shared lookup-file cache. */ +final class NamespacedLookupFileCache implements Cache { + + private final Cache sharedCache; + private final String namespace; + + NamespacedLookupFileCache( + Cache sharedCache, String namespace) { + this.sharedCache = checkNotNull(sharedCache, "sharedCache must not be null."); + this.namespace = checkNotNull(namespace, "namespace must not be null."); + } + + @Override + public @Nullable LookupFile getIfPresent(Object fileName) { + return fileName instanceof String ? sharedCache.getIfPresent(key((String) fileName)) : null; + } + + @Override + public LookupFile get( + String fileName, Function mappingFunction) { + return sharedCache.get(key(fileName), ignored -> mappingFunction.apply(fileName)); + } + + @Override + public Map getAllPresent(Iterable fileNames) { + Map result = new LinkedHashMap<>(); + for (Object fileName : fileNames) { + LookupFile lookupFile = getIfPresent(fileName); + if (lookupFile != null) { + result.put((String) fileName, lookupFile); + } + } + return result; + } + + @Override + public void put(String fileName, LookupFile lookupFile) { + sharedCache.put(key(fileName), lookupFile); + } + + @Override + public void putAll(Map entries) { + entries.forEach(this::put); + } + + @Override + public void invalidate(Object fileName) { + if (fileName instanceof String) { + sharedCache.invalidate(key((String) fileName)); + } + } + + @Override + public void invalidateAll(Iterable fileNames) { + for (Object fileName : fileNames) { + invalidate(fileName); + } + } + + @Override + public void invalidateAll() { + // ponytail: O(n) namespace scan; add a namespace index if cache cardinality makes close + // slow. + Set keys = new HashSet<>(); + for (SharedLookupFileCache.Key key : sharedCache.asMap().keySet()) { + if (key.namespace.equals(namespace)) { + keys.add(key); + } + } + sharedCache.invalidateAll(keys); + } + + @Override + public long estimatedSize() { + return sharedCache.asMap().keySet().stream() + .filter(key -> key.namespace.equals(namespace)) + .count(); + } + + @Override + public CacheStats stats() { + return sharedCache.stats(); + } + + @Override + public ConcurrentMap asMap() { + return new NamespacedMap(); + } + + @Override + public void cleanUp() { + sharedCache.cleanUp(); + } + + @Override + public Policy policy() { + throw new UnsupportedOperationException( + "Policy access is not supported by the namespaced cache view."); + } + + /** Concurrent-map view required by Paimon's closed-entry removal path. */ + private final class NamespacedMap extends AbstractMap + implements ConcurrentMap { + + @Override + public Set> entrySet() { + Set> entries = new HashSet<>(); + sharedCache + .asMap() + .forEach( + (key, lookupFile) -> { + if (key.namespace.equals(namespace)) { + entries.add( + new SimpleImmutableEntry<>(key.fileName, lookupFile)); + } + }); + return entries; + } + + @Override + public @Nullable LookupFile get(Object fileName) { + return fileName instanceof String + ? sharedCache.asMap().get(key((String) fileName)) + : null; + } + + @Override + public @Nullable LookupFile put(String fileName, LookupFile lookupFile) { + return sharedCache.asMap().put(key(fileName), lookupFile); + } + + @Override + public @Nullable LookupFile remove(Object fileName) { + return fileName instanceof String + ? sharedCache.asMap().remove(key((String) fileName)) + : null; + } + + @Override + public boolean remove(Object fileName, Object lookupFile) { + return fileName instanceof String + && sharedCache.asMap().remove(key((String) fileName), lookupFile); + } + + @Override + public @Nullable LookupFile putIfAbsent(String fileName, LookupFile lookupFile) { + return sharedCache.asMap().putIfAbsent(key(fileName), lookupFile); + } + + @Override + public boolean replace(String fileName, LookupFile oldValue, LookupFile newValue) { + return sharedCache.asMap().replace(key(fileName), oldValue, newValue); + } + + @Override + public @Nullable LookupFile replace(String fileName, LookupFile lookupFile) { + return sharedCache.asMap().replace(key(fileName), lookupFile); + } + + @Override + public void clear() { + invalidateAll(); + } + } + + private SharedLookupFileCache.Key key(String fileName) { + return new SharedLookupFileCache.Key(namespace, fileName); + } +} 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 01d29aac355..bca2c4ca353 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 @@ -18,14 +18,12 @@ package org.apache.fluss.lake.paimon.lookup; import org.apache.fluss.config.Configuration; -import org.apache.fluss.config.MemorySize; import org.apache.fluss.config.TableConfig; 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; @@ -37,7 +35,6 @@ import org.apache.fluss.utils.ExceptionUtils; import org.apache.fluss.utils.IOUtils; -import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.Catalog; import org.apache.paimon.catalog.CatalogContext; import org.apache.paimon.catalog.CatalogFactory; @@ -49,7 +46,6 @@ 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; @@ -62,104 +58,96 @@ 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 static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; import static org.apache.fluss.lake.paimon.PaimonLakeCatalog.LEGACY_SYSTEM_COLUMNS; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimonPartition; -import static org.apache.fluss.utils.Preconditions.checkArgument; import static org.apache.fluss.utils.Preconditions.checkNotNull; /** * Paimon implementation of {@link LakeTableLookuper} for primary-key tables. * - *

The catalog, table, local query, and I/O manager are initialized lazily on the first lookup. - * For each partition and bucket, the lookuper scans the latest Paimon snapshot once and registers - * its data files with {@link LocalTableQuery}. Paimon then creates local lookup files lazily as + *

The catalog, table, and local query are initialized lazily on the first lookup. The I/O + * manager is supplied by the lake storage and may be shared with other table lookupers. For each + * partition and bucket, the lookuper scans the latest Paimon snapshot once and registers its data + * files with {@link SharedLocalTableQuery}. Paimon then creates local lookup files lazily as * individual remote data files are queried. * *

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 concurrency is delegated to {@link SharedLocalTableQuery}. Older Paimon versions may + * serialize lookups internally, while Paimon 2.0 supports concurrent lookups without an additional + * Fluss-level lock. + * + *

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 { private final Configuration paimonConfig; private final TablePath tablePath; - private final String ioTmpDir; + private final IOManager ioManager; + private final SharedLookupFileCache lookupFileCache; + private final String cacheNamespace; private final TableConfig tableConfig; - private final long lookupCacheMaxDiskBytes; private final Runnable diskWriteGuard; - private final Set initializedBuckets; + private final ThreadLocal lookupFileDownloaded; + private final Object initializationLock; + private final Map> registeredFiles; 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; - - /** Creates a lookuper with the specified local lookup cache limit. */ + + private volatile @Nullable SharedLocalTableQuery localTableQuery; + // Guarded by initializationLock. + private volatile boolean closed; + + /** Creates a lookuper using an I/O manager shared with other table lookupers. */ public PaimonLakeTableLookuper( Configuration paimonConfig, TablePath tablePath, - String ioTmpDir, + IOManager ioManager, + SharedLookupFileCache lookupFileCache, + String cacheNamespace, TableConfig tableConfig, - long lookupCacheMaxDiskBytes, Runnable diskWriteGuard) { this.paimonConfig = checkNotNull(paimonConfig, "paimonConfig must not be null."); this.tablePath = checkNotNull(tablePath, "tablePath must not be null."); - this.ioTmpDir = checkNotNull(ioTmpDir, "ioTmpDir must not be null."); + this.lookupFileCache = checkNotNull(lookupFileCache, "lookupFileCache must not be null."); + this.cacheNamespace = checkNotNull(cacheNamespace, "cacheNamespace must not be null."); this.tableConfig = checkNotNull(tableConfig, "tableConfig must not be null."); - checkArgument( - 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.ioManager = + new TrackingIOManager(checkNotNull(ioManager, "ioManager must not be null.")); + this.initializationLock = new Object(); + this.registeredFiles = new ConcurrentHashMap<>(); } @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; - long lookupStartNanos = System.nanoTime(); - org.apache.paimon.data.InternalRow paimonRow; - try { - paimonRow = - lookupWithFileRefresh( - partition, context.bucketId(), keyRow, context.valueRowType()); + try (TrackingMetrics ignored = new TrackingMetrics(lookupFileDownloaded, context)) { + return lookupInternal(key, context); } catch (Exception e) { DiskWriteLockedException diskWriteLockedException = ExceptionUtils.findThrowable(e, DiskWriteLockedException.class).orElse(null); @@ -167,31 +155,25 @@ public PaimonLakeTableLookuper( throw diskWriteLockedException; } throw e; - } finally { - 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; - } - 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(catalog, "Paimon catalog"); + registeredFiles.clear(); + localTableQuery = null; + compactedKeyDecoder = null; + trimmedPrimaryKeys = 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,81 +183,67 @@ 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; + SharedLocalTableQuery newLocalTableQuery = null; boolean initialized = false; try { newCatalog = CatalogFactory.createCatalog( CatalogContext.create(Options.fromMap(paimonConfig.toMap()))); FileStoreTable newFileStoreTable = - withLookupCacheOptions( - (FileStoreTable) newCatalog.getTable(toPaimon(tablePath))); + (FileStoreTable) newCatalog.getTable(toPaimon(tablePath)); if (newFileStoreTable.primaryKeys().isEmpty()) { throw new UnsupportedOperationException( "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. + newLocalTableQuery = + new SharedLocalTableQuery(newFileStoreTable, lookupFileCache, cacheNamespace) + .withValueProjection(businessFieldProjection(newFileStoreTable)) + .withIOManager(ioManager); + 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. + // Keep this volatile write last to publish all initialized fields together. localTableQuery = newLocalTableQuery; initialized = true; } finally { if (!initialized) { IOUtils.closeQuietly(newLocalTableQuery, "Paimon local table query"); - IOUtils.closeQuietly(newIOManager, "Paimon lookup IO manager"); IOUtils.closeQuietly(newCatalog, "Paimon catalog"); } } } - private FileStoreTable withLookupCacheOptions(FileStoreTable table) { - String key = CoreOptions.LOOKUP_CACHE_MAX_DISK_SIZE.key(); - String maxDiskSize = new MemorySize(lookupCacheMaxDiskBytes).toString(); - return table.copy(Collections.singletonMap(key, maxDiskSize)); - } - - private IOManager createIOManager(String ioTmpDir) { - return new TrackingIOManager(IOManager.create(ioTmpDir)); - } - private static int[] businessFieldProjection(FileStoreTable fileStoreTable) { List fields = fileStoreTable.schema().logicalRowType().getFields(); List projectedFields = new ArrayList<>(); @@ -292,139 +260,134 @@ 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; + private @Nullable byte[] lookupInternal(byte[] key, LookupContext context) { + org.apache.paimon.data.InternalRow paimonRow; + try { + paimonRow = + lookupPaimon(getPartition(context), context.bucketId(), getKey(key, context)); + } catch (IOException 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 + + ".", + e); } - - 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()); + if (paimonRow == null) { + return null; } - - // 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); + return encodeValue(paimonRow, context.schemaId(), context.valueRowType()); } - private org.apache.paimon.data.InternalRow lookupWithFileRefresh( + private @Nullable org.apache.paimon.data.InternalRow lookupPaimon( org.apache.paimon.data.BinaryRow partition, - int bucketId, - org.apache.paimon.data.InternalRow keyRow, - RowType valueRowType) - throws Exception { + int bucket, + org.apache.paimon.data.InternalRow key) + throws IOException { + List filesBeforeLookup = initializeFiles(partition, bucket); try { - return localTableQuery().lookup(partition, bucketId, keyRow); - } 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. + return localTableQuery.lookup(partition, bucket, key); + } catch (IOException firstError) { + refreshFiles(partition, bucket, filesBeforeLookup); try { - refreshFiles(partition, bucketId, valueRowType); - return localTableQuery().lookup(partition, bucketId, keyRow); + return localTableQuery.lookup(partition, bucket, key); } 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); + retryError.addSuppressed(firstError); + throw retryError; } } } + private List initializeFiles( + org.apache.paimon.data.BinaryRow partition, int bucket) { + PaimonPartitionBucket partitionBucket = new PaimonPartitionBucket(partition, bucket); + return registeredFiles.computeIfAbsent( + partitionBucket, + ignored -> registerFiles(partition, bucket, Collections.emptyList())); + } + 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); + org.apache.paimon.data.BinaryRow partition, + int bucket, + List filesBeforeLookup) { + PaimonPartitionBucket partitionBucket = new PaimonPartitionBucket(partition, bucket); + registeredFiles.compute( + partitionBucket, + (ignored, currentFiles) -> { + List files = + checkNotNull( + currentFiles, "Partition-bucket files must be initialized."); + return files == filesBeforeLookup + ? registerFiles(partition, bucket, filesBeforeLookup) + : files; + }); + } + + private List registerFiles( + org.apache.paimon.data.BinaryRow partition, + int bucket, + List filesBeforeRefresh) { + List latestFiles = scanDataFiles(partition, bucket); + localTableQuery.refreshFiles(partition, bucket, filesBeforeRefresh, latestFiles); + return latestFiles; } - private static void addFilesByName( - LinkedHashMap filesByName, List files) { - for (DataFileMeta file : files) { - filesByName.put(file.fileName(), file); + private List scanDataFiles( + org.apache.paimon.data.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 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 +399,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 +422,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 @@ -514,9 +458,30 @@ public BufferFileReader createBufferFileReader(FileIOChannel.ID channelID) return delegate.createBufferFileReader(channelID); } + @Override + public void close() { + // The shared delegate is owned by PaimonLakeStorage. + } + } + + private final class TrackingMetrics implements AutoCloseable { + private final ThreadLocal lookupFileDownloaded; + private final long startNanoTime; + private final LookupContext context; + + private TrackingMetrics(ThreadLocal lookupFileDownloaded, LookupContext context) { + this.lookupFileDownloaded = lookupFileDownloaded; + this.lookupFileDownloaded.set(false); + this.startNanoTime = System.nanoTime(); + this.context = context; + } + @Override public void close() throws Exception { - delegate.close(); + boolean fileDownloaded = lookupFileDownloaded.get(); + lookupFileDownloaded.remove(); + context.lookupMetricRecorder() + .recordLookup(System.nanoTime() - startNanoTime, fileDownloaded); } } } diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/SharedLocalTableQuery.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/SharedLocalTableQuery.java new file mode 100644 index 00000000000..3e6ee7701a3 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/SharedLocalTableQuery.java @@ -0,0 +1,316 @@ +/* + * 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.paimon.CoreOptions; +import org.apache.paimon.FileStore; +import org.apache.paimon.KeyValue; +import org.apache.paimon.KeyValueFileStore; +import org.apache.paimon.data.BinaryRow; +import org.apache.paimon.data.InternalRow; +import org.apache.paimon.data.serializer.InternalRowSerializer; +import org.apache.paimon.data.serializer.InternalSerializers; +import org.apache.paimon.data.serializer.RowCompactedSerializer; +import org.apache.paimon.deletionvectors.DeletionVector; +import org.apache.paimon.disk.IOManager; +import org.apache.paimon.fs.FileIO; +import org.apache.paimon.io.DataFileMeta; +import org.apache.paimon.io.KeyValueFileReaderFactory; +import org.apache.paimon.io.cache.CacheManager; +import org.apache.paimon.lookup.LookupStoreFactory; +import org.apache.paimon.mergetree.Levels; +import org.apache.paimon.mergetree.LookupFile; +import org.apache.paimon.mergetree.LookupLevels; +import org.apache.paimon.mergetree.lookup.LookupSerializerFactory; +import org.apache.paimon.mergetree.lookup.PersistValueProcessor; +import org.apache.paimon.mergetree.lookup.RemoteLookupFileManager; +import org.apache.paimon.operation.metrics.PartialLookupMetrics; +import org.apache.paimon.options.Options; +import org.apache.paimon.reader.RecordReader; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; +import org.apache.paimon.table.FileStoreTable; +import org.apache.paimon.table.query.TableQuery; +import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.Filter; +import org.apache.paimon.utils.KeyComparatorSupplier; +import org.apache.paimon.utils.Preconditions; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.util.Comparator; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.locks.ReadWriteLock; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Supplier; + +import static org.apache.paimon.lookup.LookupStoreFactory.bloomFilterBuilderFactory; +import static org.apache.paimon.mergetree.LookupFile.localFilePrefix; + +/** + * Fluss-owned fork of Paimon 2.0.0's {@code LocalTableQuery}, using a shared namespaced lookup + * cache. + */ +final class SharedLocalTableQuery implements TableQuery { + + private final Map> tableView; + + private final CoreOptions options; + + private final Supplier> keyComparatorSupplier; + + private final KeyValueFileReaderFactory.Builder readerFactoryBuilder; + + private final LookupStoreFactory lookupStoreFactory; + + private final int startLevel; + + private IOManager ioManager; + + private final Cache lookupFileCache; + + private final RowType rowType; + private final RowType partitionType; + private final FileIO fileIO; + + @Nullable private Filter cacheRowFilter; + + @Nullable private PartialLookupMetrics partialLookupMetrics; + + SharedLocalTableQuery( + FileStoreTable table, + SharedLookupFileCache sharedLookupFileCache, + String cacheNamespace) { + this.options = table.coreOptions(); + this.lookupFileCache = sharedLookupFileCache.namespaced(cacheNamespace); + this.tableView = new ConcurrentHashMap<>(); + FileStore tableStore = table.store(); + if (!(tableStore instanceof KeyValueFileStore)) { + throw new UnsupportedOperationException( + "Table Query only supports table with primary key."); + } + KeyValueFileStore store = (KeyValueFileStore) tableStore; + + this.readerFactoryBuilder = store.newReaderFactoryBuilder(); + this.rowType = table.schema().logicalRowType(); + this.partitionType = table.schema().logicalPartitionType(); + this.fileIO = table.fileIO(); + RowType keyType = readerFactoryBuilder.keyType(); + this.keyComparatorSupplier = new KeyComparatorSupplier(readerFactoryBuilder.keyType()); + this.lookupStoreFactory = + LookupStoreFactory.create( + options, + new CacheManager( + options.lookupCacheMaxMemory(), + options.lookupCacheHighPrioPoolRatio()), + new RowCompactedSerializer(keyType).createSliceComparator()); + startLevel = options.needLookup() ? 1 : 0; + } + + public void refreshFiles( + BinaryRow partition, + int bucket, + List beforeFiles, + List dataFiles) { + // Both tableView and its nested bucket maps are ConcurrentHashMaps; this nested + // computeIfAbsent pattern relies on each map providing atomic insertion. + BucketLookupState state = + tableView + .computeIfAbsent(partition, k -> new ConcurrentHashMap<>()) + .computeIfAbsent(bucket, k -> new BucketLookupState()); + state.lock.writeLock().lock(); + try { + if (state.lookupLevels == null) { + // Initial phase: ignore beforeFiles as they represent deletions from previous state + state.lookupLevels = createLookupLevels(partition, bucket, dataFiles); + } else { + state.lookupLevels.getLevels().update(beforeFiles, dataFiles); + } + } finally { + state.lock.writeLock().unlock(); + } + } + + private LookupLevels createLookupLevels( + BinaryRow partition, int bucket, List dataFiles) { + Levels levels = new Levels(keyComparatorSupplier.get(), dataFiles, options.numLevels()); + // TODO pass DeletionVector factory + KeyValueFileReaderFactory factory = + readerFactoryBuilder.build(partition, bucket, DeletionVector.emptyFactory()); + Options options = this.options.toConfiguration(); + + RowType readValueType = readerFactoryBuilder.readValueType(); + LookupLevels lookupLevels = + new LookupLevels<>( + schemaId -> readValueType, + 0L, + levels, + keyComparatorSupplier.get(), + readerFactoryBuilder.keyType(), + PersistValueProcessor.factory(readValueType), + LookupSerializerFactory.INSTANCE.get(), + file -> { + RecordReader reader = factory.createRecordReader(file); + if (cacheRowFilter != null) { + reader = + reader.filter( + keyValue -> cacheRowFilter.test(keyValue.value())); + } + return reader; + }, + file -> + Preconditions.checkNotNull(ioManager, "IOManager is required.") + .createChannel( + localFilePrefix( + partitionType, partition, bucket, file)) + .getPathFile(), + lookupStoreFactory, + bloomFilterBuilderFactory(options), + lookupFileCache); + + // Optimization - download lookup files if already persisted to object store + // We download these files if three conditions are met + // 1) lookup.remote-file.enabled is true - files are persisted in the first place + // 2) deletion-vectors.enabled is false - SSTables only contain row positions, not values, + // when DVs are enabled + // 3) The client is accessing the full data row, as opposed to a projection + // - The persisted remote SSTable files are created during compaction and hold the entire + // data row value + // - We could deserialize and project in memory, but we'll have to read much more data, + // not as clear of a win + boolean fullValueRead = readerFactoryBuilder.readValueType().equals(rowType); + if (this.options.lookupRemoteFileEnabled() + && !this.options.deletionVectorsEnabled() + && fullValueRead) { + // Calling the constructor tells `lookupLevels` to load remote files + new RemoteLookupFileManager<>( + fileIO, + factory.pathFactory(), + lookupLevels, + this.options.lookupRemoteLevelThreshold()); + } + + return lookupLevels; + } + + @Nullable + @Override + public InternalRow lookup(BinaryRow partition, int bucket, InternalRow key) throws IOException { + PartialLookupMetrics currentMetrics = partialLookupMetrics; + LookupLevels.LookupContext context = + currentMetrics == null ? null : new LookupLevels.LookupContext(); + try { + return lookup(partition, bucket, key, context); + } finally { + if (currentMetrics != null) { + currentMetrics.reportLookup(context != null && context.remoteAccessed()); + } + } + } + + @Nullable + private InternalRow lookup( + BinaryRow partition, + int bucket, + InternalRow key, + @Nullable LookupLevels.LookupContext context) + throws IOException { + Map buckets = tableView.get(partition); + if (buckets == null || buckets.isEmpty()) { + return null; + } + BucketLookupState state = buckets.get(bucket); + if (state == null) { + return null; + } + + state.lock.readLock().lock(); + try { + LookupLevels lookupLevels = state.lookupLevels; + if (lookupLevels == null) { + return null; + } + + KeyValue kv = lookupLevels.lookup(key, startLevel, context); + if (kv == null || kv.valueKind().isRetract()) { + return null; + } else { + return kv.value(); + } + } finally { + state.lock.readLock().unlock(); + } + } + + @Override + public SharedLocalTableQuery withValueProjection(int[] projection) { + this.readerFactoryBuilder.withReadValueType(rowType.project(projection)); + return this; + } + + public SharedLocalTableQuery withIOManager(IOManager ioManager) { + this.ioManager = ioManager; + return this; + } + + public SharedLocalTableQuery withCacheRowFilter(Filter cacheRowFilter) { + this.cacheRowFilter = cacheRowFilter; + return this; + } + + public SharedLocalTableQuery withMetrics(@Nullable PartialLookupMetrics metrics) { + this.partialLookupMetrics = metrics; + return this; + } + + @Override + public InternalRowSerializer createValueSerializer() { + return InternalSerializers.create(readerFactoryBuilder.readValueType()); + } + + @Override + public void close() throws IOException { + // ConcurrentHashMap iteration is weakly consistent. close is expected not to race with + // refreshFiles for the same query instance; callers may rebuild this query after close. + for (Map.Entry> buckets : tableView.entrySet()) { + for (Map.Entry bucket : buckets.getValue().entrySet()) { + BucketLookupState state = bucket.getValue(); + state.lock.writeLock().lock(); + try { + if (state.lookupLevels != null) { + state.lookupLevels.close(); + } + } finally { + state.lock.writeLock().unlock(); + } + } + } + lookupFileCache.invalidateAll(); + tableView.clear(); + } + + private static class BucketLookupState { + + private final ReadWriteLock lock = new ReentrantReadWriteLock(); + + @Nullable private LookupLevels lookupLevels; + } +} diff --git a/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/SharedLookupFileCache.java b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/SharedLookupFileCache.java new file mode 100644 index 00000000000..1e99d914c83 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/lookup/SharedLookupFileCache.java @@ -0,0 +1,112 @@ +/* + * 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.annotation.Internal; + +import org.apache.paimon.mergetree.LookupFile; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Caffeine; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.RemovalCause; + +import javax.annotation.Nullable; + +import java.io.IOException; +import java.io.UncheckedIOException; +import java.time.Duration; +import java.util.Objects; + +import static org.apache.fluss.utils.Preconditions.checkNotNull; +import static org.apache.paimon.mergetree.LookupUtils.fileKibiBytes; + +/** A weighted lookup-file cache shared by multiple Paimon table lookupers. */ +@Internal +public final class SharedLookupFileCache implements AutoCloseable { + + private final Cache cache; + + /** Creates a shared lookup-file cache. */ + public SharedLookupFileCache(Duration fileRetention, MemorySize maxDiskSize) { + checkNotNull(fileRetention, "fileRetention must not be null."); + checkNotNull(maxDiskSize, "maxDiskSize must not be null."); + this.cache = + Caffeine.newBuilder() + .expireAfterAccess(fileRetention) + .maximumWeight(Math.max(1L, maxDiskSize.getKibiBytes())) + .weigher( + (Key key, LookupFile lookupFile) -> + Math.max(1, fileKibiBytes(lookupFile.localFile()))) + .removalListener(SharedLookupFileCache::removeLookupFile) + .executor(Runnable::run) + .build(); + } + + Cache namespaced(String namespace) { + return new NamespacedLookupFileCache(cache, namespace); + } + + /** Updates the maximum cache weight. */ + public void updateMaxDiskSize(MemorySize maxDiskSize) { + cache.policy().eviction().get().setMaximum(Math.max(1L, maxDiskSize.getKibiBytes())); + } + + @Override + public void close() { + cache.invalidateAll(); + cache.cleanUp(); + } + + private static void removeLookupFile( + @Nullable Key key, @Nullable LookupFile lookupFile, RemovalCause cause) { + if (lookupFile != null) { + try { + lookupFile.close(cause); + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + } + + static final class Key { + final String namespace; + final String fileName; + + Key(String namespace, String fileName) { + this.namespace = checkNotNull(namespace, "namespace must not be null."); + this.fileName = checkNotNull(fileName, "fileName must not be null."); + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof Key)) { + return false; + } + Key key = (Key) o; + return namespace.equals(key.namespace) && fileName.equals(key.fileName); + } + + @Override + public int hashCode() { + return Objects.hash(namespace, fileName); + } + } +} 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 1bf0ec3a16f..ee6e3c60a0d 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 @@ -22,9 +22,11 @@ import org.apache.fluss.config.MemorySize; import org.apache.fluss.config.TableConfig; import org.apache.fluss.exception.DiskWriteLockedException; +import org.apache.fluss.lake.lakestorage.LakeTableLookupRuntime; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.lakestorage.TestingLakeCatalogContext; import org.apache.fluss.lake.paimon.PaimonLakeCatalog; +import org.apache.fluss.lake.paimon.PaimonLakeStorage; import org.apache.fluss.metadata.KvFormat; import org.apache.fluss.metadata.ResolvedPartitionSpec; import org.apache.fluss.metadata.Schema; @@ -38,6 +40,7 @@ import org.apache.fluss.row.encode.ValueDecoder; import org.apache.fluss.row.encode.paimon.PaimonKeyEncoder; import org.apache.fluss.types.DataTypes; +import org.apache.fluss.utils.ExecutorUtils; import org.apache.paimon.CoreOptions; import org.apache.paimon.catalog.Catalog; @@ -60,11 +63,22 @@ import org.junit.jupiter.api.io.TempDir; import java.io.File; +import java.io.IOException; +import java.nio.file.Files; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; +import java.util.HashSet; import java.util.List; +import java.util.Set; +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.stream.Collectors; +import java.util.stream.Stream; import static org.apache.fluss.config.ConfigOptions.KV_FORMAT_VERSION_2; import static org.apache.fluss.lake.paimon.utils.PaimonConversions.toPaimon; @@ -90,6 +104,7 @@ class PaimonLakeTableLookuperTest { private Configuration paimonConfig; private PaimonLakeCatalog lakeCatalog; private Catalog paimonCatalog; + private LakeTableLookupRuntime lookupRuntime; @BeforeEach void setUp() { @@ -99,10 +114,17 @@ void setUp() { paimonCatalog = CatalogFactory.createCatalog( CatalogContext.create(Options.fromMap(paimonConfig.toMap()))); + lookupRuntime = + new PaimonLakeStorage(paimonConfig) + .createLakeTableLookupRuntime( + tempWarehouseDir.getAbsolutePath(), LOOKUP_CACHE_MAX_DISK_BYTES); } @AfterEach void tearDown() throws Exception { + if (lookupRuntime != null) { + lookupRuntime.close(); + } if (paimonCatalog != null) { paimonCatalog.close(); } @@ -123,13 +145,8 @@ void testLookupPartitionedPrimaryKeyTable() throws Exception { 0, Collections.singletonList(paimonRow(1, "20240101", "Alice")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper( + tablePath, tableConfig(KvFormat.COMPACTED), NO_OP_DISK_WRITE_GUARD)) { List lookupFileDownloads = new ArrayList<>(); LakeTableLookuper.LookupContext context = lookupContext( @@ -179,13 +196,7 @@ void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { }; try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - diskWriteGuard)) { + createLookuper(tablePath, tableConfig(KvFormat.COMPACTED), diskWriteGuard)) { LakeTableLookuper.LookupContext cachedPartition = lookupContext(schema, "20240101", 0, SCHEMA_ID); LakeTableLookuper.LookupContext uncachedPartition = @@ -210,6 +221,83 @@ void testDiskWriteLockBlocksOnlyLookupFileDownloads() throws Exception { } } + @Test + void testSharesIOManagerAndDeletesOnlyClosedLookuperFiles() throws Exception { + Schema schema = pkSchema(); + TablePath firstTablePath = TablePath.of(DB, "shared_io_first"); + TablePath secondTablePath = TablePath.of(DB, "shared_io_second"); + FileStoreTable firstTable = + createPaimonTable(firstTablePath, partitionedPkDescriptor(schema)); + FileStoreTable secondTable = + createPaimonTable(secondTablePath, partitionedPkDescriptor(schema)); + writeAndCommitData( + firstTable, + Collections.singletonMap( + 0, Collections.singletonList(paimonRow(1, "20240101", "Alice")))); + writeAndCommitData( + secondTable, + Collections.singletonMap( + 0, Collections.singletonList(paimonRow(2, "20240101", "Bob")))); + + File lookupDir = new File(tempWarehouseDir, "shared-lookup-cache"); + LakeTableLookupRuntime.Context firstLookuperContext = + new LakeTableLookupRuntime.Context( + paimonConfig, + "first-table", + tableConfig(KvFormat.COMPACTED), + NO_OP_DISK_WRITE_GUARD); + LakeTableLookupRuntime.Context secondLookuperContext = + new LakeTableLookupRuntime.Context( + paimonConfig, + "second-table", + tableConfig(KvFormat.COMPACTED), + NO_OP_DISK_WRITE_GUARD); + LakeTableLookupRuntime sharedLookupRuntime = + new PaimonLakeStorage(paimonConfig) + .createLakeTableLookupRuntime( + lookupDir.getAbsolutePath(), LOOKUP_CACHE_MAX_DISK_BYTES); + try { + try (LakeTableLookuper firstLookuper = + sharedLookupRuntime.createLakeTableLookuper( + firstTablePath, firstLookuperContext); + LakeTableLookuper secondLookuper = + sharedLookupRuntime.createLakeTableLookuper( + secondTablePath, secondLookuperContext)) { + assertThat( + firstLookuper.lookup( + paimonKey(schema, 1, "20240101"), + lookupContext(schema, "20240101", 0, SCHEMA_ID))) + .isNotNull(); + Set firstLookupFiles = regularFiles(lookupDir); + assertThat(firstLookupFiles).isNotEmpty(); + + assertThat( + secondLookuper.lookup( + paimonKey(schema, 2, "20240101"), + lookupContext(schema, "20240101", 0, SCHEMA_ID))) + .isNotNull(); + Set secondLookupFiles = regularFiles(lookupDir); + secondLookupFiles.removeAll(firstLookupFiles); + assertThat(secondLookupFiles).isNotEmpty(); + assertThat(lookupDir.listFiles(File::isDirectory)).hasSize(1); + + firstLookuper.close(); + assertThat(firstLookupFiles).allMatch(path -> !Files.exists(path)); + assertThat(secondLookupFiles).allMatch(Files::exists); + assertThat( + secondLookuper.lookup( + paimonKey(schema, 2, "20240101"), + lookupContext(schema, "20240101", 0, SCHEMA_ID))) + .isNotNull(); + } + assertThat(regularFiles(lookupDir)).isEmpty(); + assertThat(lookupDir.listFiles(File::isDirectory)).hasSize(1); + } finally { + sharedLookupRuntime.close(); + } + assertThat(lookupDir.listFiles(File::isDirectory)).isEmpty(); + } + @Test void testLookupPartitionsWithSameHashCode() throws Exception { // These distinct partition values produce the same BinaryRow hash code, reproducing the @@ -236,13 +324,8 @@ void testLookupPartitionsWithSameHashCode() throws Exception { 0, Collections.singletonList(paimonRow(2, secondPartition, "Bob")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper( + tablePath, tableConfig(KvFormat.COMPACTED), NO_OP_DISK_WRITE_GUARD)) { BinaryValue firstValue = decodeValue( lookuper.lookup( @@ -278,13 +361,7 @@ void testLookupWithIndexedKvFormat() throws Exception { 0, Collections.singletonList(paimonRow(1, "20240101", "Alice")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.INDEXED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper(tablePath, tableConfig(KvFormat.INDEXED), NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = lookupContext(schema, "20240101", 0, SCHEMA_ID); @@ -324,12 +401,9 @@ void testLookupKvFormatV2WithNonDefaultBucketKey() throws Exception { 0, Collections.singletonList(paimonRow(1, "sub-1", "20240101", "Alice")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, + createLookuper( 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); @@ -381,12 +455,9 @@ void testRetriesInitializationAfterLookupKeyConverterFailure() throws Exception .build(); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, + createLookuper( tablePath, - tempWarehouseDir.getAbsolutePath(), tableConfig(KvFormat.COMPACTED, KV_FORMAT_VERSION_2), - LOOKUP_CACHE_MAX_DISK_BYTES, NO_OP_DISK_WRITE_GUARD)) { // Inject a late initialization failure: the Paimon table requires sub_id in its // lookup key, but the first lookup's value row type deliberately omits that field. @@ -419,22 +490,43 @@ void testRefreshFilesAfterCompactionAndSnapshotExpiration() throws Exception { Collections.singletonMap( 0, Collections.singletonList(paimonRow(id, "20240101", "name-" + id)))); } + writeAndCommitData( + table, + Collections.singletonMap( + 0, Collections.singletonList(paimonRow(6, "20240102", "name-6")))); BinaryRow partition = BinaryRow.singleColumn(BinaryString.fromString("20240101")); List filesBeforeCompaction = dataFiles(table, partition, 0); assertThat(filesBeforeCompaction).hasSize(5); + AtomicBoolean blockDownloads = new AtomicBoolean(); + CountDownLatch downloadStarted = new CountDownLatch(1); + CountDownLatch continueDownload = new CountDownLatch(1); + Runnable diskWriteGuard = + () -> { + if (blockDownloads.get()) { + downloadStarted.countDown(); + try { + continueDownload.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new RuntimeException(e); + } + } + }; + try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { - LakeTableLookuper.LookupContext context = - lookupContext(schema, "20240101", 0, SCHEMA_ID); - assertThat(lookuper.lookup(paimonKey(schema, 5, "20240101"), context)).isNotNull(); + createLookuper(tablePath, tableConfig(KvFormat.COMPACTED), diskWriteGuard)) { + assertThat( + lookuper.lookup( + paimonKey(schema, 5, "20240101"), + lookupContext(schema, "20240101", 0, SCHEMA_ID))) + .isNotNull(); + assertThat( + lookuper.lookup( + paimonKey(schema, 6, "20240102"), + lookupContext(schema, "20240102", 0, SCHEMA_ID))) + .isNotNull(); new CompactHelper(table, new File(tempWarehouseDir, "compact")) .compactBucket(partition, 0) @@ -462,12 +554,55 @@ void testRefreshFilesAfterCompactionAndSnapshotExpiration() throws Exception { assertThat(table.store().snapshotManager().fileIO().exists(path)).isFalse(); } - BinaryValue decodedValue = - decodeValue( - lookuper.lookup(paimonKey(schema, 1, "20240101"), context), + List refreshedLookupDownloads = new ArrayList<>(); + List cachedLookupDownloads = new ArrayList<>(); + LakeTableLookuper.LookupContext refreshedContext = + lookupContext( + schema, + "20240101", + 0, SCHEMA_ID, - schema); - assertRow(decodedValue.row, 1, "20240101", "name-1"); + (lookupTimeNanos, lookupFileDownloaded) -> + refreshedLookupDownloads.add(lookupFileDownloaded)); + LakeTableLookuper.LookupContext cachedContext = + lookupContext( + schema, + "20240102", + 0, + SCHEMA_ID, + (lookupTimeNanos, lookupFileDownloaded) -> + cachedLookupDownloads.add(lookupFileDownloaded)); + ExecutorService executor = Executors.newFixedThreadPool(2); + blockDownloads.set(true); + try { + Future refreshedLookup = + executor.submit( + () -> + lookuper.lookup( + paimonKey(schema, 1, "20240101"), + refreshedContext)); + assertThat(downloadStarted.await(30, TimeUnit.SECONDS)).isTrue(); + + Future cachedLookup = + executor.submit( + () -> + lookuper.lookup( + paimonKey(schema, 6, "20240102"), cachedContext)); + BinaryValue cachedValue = + decodeValue(cachedLookup.get(30, TimeUnit.SECONDS), SCHEMA_ID, schema); + assertRow(cachedValue.row, 6, "20240102", "name-6"); + + continueDownload.countDown(); + BinaryValue refreshedValue = + decodeValue(refreshedLookup.get(30, TimeUnit.SECONDS), SCHEMA_ID, schema); + assertRow(refreshedValue.row, 1, "20240101", "name-1"); + } finally { + continueDownload.countDown(); + ExecutorUtils.gracefulShutdown(30, TimeUnit.SECONDS, executor); + } + + assertThat(refreshedLookupDownloads).containsExactly(true); + assertThat(cachedLookupDownloads).containsExactly(false); } } @@ -493,13 +628,8 @@ void testLookupWithNonStringPartitionKey() throws Exception { Collections.singletonMap(0, Collections.singletonList(paimonRow(1, 7, "Alice")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper( + tablePath, tableConfig(KvFormat.COMPACTED), NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = new LakeTableLookuper.LookupContext( ResolvedPartitionSpec.fromPartitionName( @@ -531,13 +661,8 @@ void testRejectAppendOnlyTable() throws Exception { createPaimonTable(tablePath, tableDescriptor); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper( + tablePath, tableConfig(KvFormat.COMPACTED), NO_OP_DISK_WRITE_GUARD)) { LakeTableLookuper.LookupContext context = new LakeTableLookuper.LookupContext( new ResolvedPartitionSpec( @@ -590,13 +715,8 @@ void testLookupAfterSchemaEvolutionPadsNewColumnsWithNull() throws Exception { Collections.singletonList(paimonRow(2, "20240101", "Bob", "new-value")))); try (LakeTableLookuper lookuper = - new PaimonLakeTableLookuper( - paimonConfig, - tablePath, - tempWarehouseDir.getAbsolutePath(), - tableConfig(KvFormat.COMPACTED), - LOOKUP_CACHE_MAX_DISK_BYTES, - NO_OP_DISK_WRITE_GUARD)) { + createLookuper( + tablePath, tableConfig(KvFormat.COMPACTED), NO_OP_DISK_WRITE_GUARD)) { BinaryValue oldSchemaValue = decodeValue( lookuper.lookup( @@ -664,6 +784,23 @@ private static TableConfig tableConfig(KvFormat kvFormat, int kvFormatVersion) { return new TableConfig(config); } + private LakeTableLookuper createLookuper( + TablePath tablePath, TableConfig tableConfig, Runnable diskWriteGuard) { + return lookupRuntime.createLakeTableLookuper( + tablePath, + new LakeTableLookupRuntime.Context( + paimonConfig, tablePath.toString(), tableConfig, diskWriteGuard)); + } + + private static Set regularFiles(File directory) throws IOException { + if (!directory.exists()) { + return new HashSet<>(); + } + try (Stream paths = Files.walk(directory.toPath())) { + return paths.filter(Files::isRegularFile).collect(Collectors.toSet()); + } + } + private static Schema pkSchema() { return Schema.newBuilder() .column("id", DataTypes.INT()) diff --git a/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/SharedLookupFileCacheTest.java b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/SharedLookupFileCacheTest.java new file mode 100644 index 00000000000..fa497dd3353 --- /dev/null +++ b/fluss-lake/fluss-lake-paimon/src/test/java/org/apache/fluss/lake/paimon/lookup/SharedLookupFileCacheTest.java @@ -0,0 +1,93 @@ +/* + * 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.paimon.lookup.LookupStoreReader; +import org.apache.paimon.mergetree.LookupFile; +import org.apache.paimon.options.MemorySize; +import org.apache.paimon.shade.caffeine2.com.github.benmanes.caffeine.cache.Cache; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; + +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.time.Duration; +import java.util.Arrays; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link SharedLookupFileCache}. */ +class SharedLookupFileCacheTest { + + @TempDir private File tempDir; + + @Test + void testNamespaceIsolationAndGlobalLimit() throws Exception { + File firstFile = lookupFile("first.lookup"); + File secondFile = lookupFile("second.lookup"); + File thirdFile = lookupFile("third.lookup"); + + try (SharedLookupFileCache sharedCache = + new SharedLookupFileCache(Duration.ofHours(1), MemorySize.ofKibiBytes(2))) { + Cache firstNamespace = sharedCache.namespaced("first"); + Cache secondNamespace = sharedCache.namespaced("second"); + LookupFile firstLookupFile = lookupFile(firstFile); + LookupFile secondLookupFile = lookupFile(secondFile); + + firstNamespace.put("same-file-name", firstLookupFile); + secondNamespace.put("same-file-name", secondLookupFile); + assertThat(firstNamespace.getIfPresent("same-file-name")).isSameAs(firstLookupFile); + assertThat(secondNamespace.getIfPresent("same-file-name")).isSameAs(secondLookupFile); + + firstNamespace.invalidateAll(); + assertThat(firstFile).doesNotExist(); + assertThat(secondFile).exists(); + + sharedCache.updateMaxDiskSize(MemorySize.ofKibiBytes(1)); + secondNamespace.put("third-file", lookupFile(thirdFile)); + assertThat(Arrays.asList(secondFile, thirdFile).stream().filter(File::exists).count()) + .isLessThanOrEqualTo(1L); + } + + assertThat(secondFile).doesNotExist(); + assertThat(thirdFile).doesNotExist(); + } + + private File lookupFile(String name) throws IOException { + File file = new File(tempDir, name); + try (RandomAccessFile randomAccessFile = new RandomAccessFile(file, "rw")) { + randomAccessFile.setLength(1024L); + } + return file; + } + + private static LookupFile lookupFile(File file) { + return new LookupFile(file, 1, 0L, "v1", new NoOpLookupStoreReader(), () -> {}); + } + + private static final class NoOpLookupStoreReader implements LookupStoreReader { + @Override + public byte[] lookup(byte[] key) { + return null; + } + + @Override + public void close() {} + } +} diff --git a/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java b/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java index 977b608402b..05f342e7129 100644 --- a/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java +++ b/fluss-server/src/main/java/org/apache/fluss/server/replica/HistoricalLakeLookupManager.java @@ -28,6 +28,7 @@ import org.apache.fluss.lake.lakestorage.LakeStorage; import org.apache.fluss.lake.lakestorage.LakeStoragePlugin; import org.apache.fluss.lake.lakestorage.LakeStoragePluginSetUp; +import org.apache.fluss.lake.lakestorage.LakeTableLookupRuntime; import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.metadata.DataLakeFormat; import org.apache.fluss.metadata.ResolvedPartitionSpec; @@ -76,6 +77,7 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; import java.util.stream.Stream; import static org.apache.fluss.server.utils.LakeStorageUtils.extractLakeProperties; @@ -97,8 +99,12 @@ * the current request. Active lookups can finish on the old lookuper, which is closed after its * last lookup releases it. * - *

Up to ten table lookupers are cached. Each lookuper receives one tenth of the server-level - * disk budget, and Caffeine evicts lookupers when the table limit is exceeded. + *

One lake-format-specific lookup runtime is initialized with this manager, or when Paimon is + * configured dynamically, and shared by all table lookupers. The runtime owns TabletServer-scoped + * resources such as Paimon's I/O manager. + * + *

Up to ten table lookupers are cached. Their lookup files share one server-level disk budget, + * and Caffeine evicts lookupers when the table limit is exceeded. * *

Historical lookup cache I/O participates in TabletServer disk write protection. Existing cache * hits remain available when the data disk is write-locked, while lookups that need to download new @@ -122,17 +128,16 @@ class HistoricalLakeLookupManager implements AutoCloseable { private static final Duration HISTORICAL_PARTITION_EXECUTOR_SHUTDOWN_TIMEOUT = Duration.ofSeconds(10); private static final String HISTORICAL_PARTITION_THREAD_NAME_PREFIX = "historical-partition-io"; - // TODO: Share one Paimon IOManager disk budget across all table lookupers and evict cached - // entries by data file instead of reserving fixed per-table capacity. See - // https://github.com/apache/fluss/issues/3955. private static final int MAX_CACHED_TABLES = 10; private volatile Configuration conf; private volatile long lakeConfigVersion; private final @Nullable PluginManager pluginManager; + private volatile @Nullable LakeTableLookupRuntime lookupRuntime; private final Counter capacityEvictions; private final int maxQueuedHistoricalRequests; private final Semaphore lookupPermits; + private final AtomicLong lookuperIdSequence; // Accepted lookup futures tracked so close() can cancel tasks left after executor shutdown. private final Set> pendingLookups; private final Cache lakeTableLookupers; @@ -144,7 +149,7 @@ class HistoricalLakeLookupManager implements AutoCloseable { // the cache to grow back to the maximum ratio after disk usage recovers. private final Runnable diskWriteGuard; - private volatile long lookupCacheMaxDiskBytesPerTable; + private volatile long lookupCacheMaxDiskBytes; private volatile long lookupCacheDiskSize; private volatile boolean started; @@ -187,8 +192,8 @@ class HistoricalLakeLookupManager implements AutoCloseable { checkArgument(dataDirVolumeBytes > 0, "dataDirVolumeBytes must be greater than 0."); this.dataDirVolumeBytes = dataDirVolumeBytes; this.diskWriteGuard = checkNotNull(diskWriteGuard, "diskWriteGuard must not be null."); - this.lookupCacheMaxDiskBytesPerTable = - cacheBytesPerTable( + this.lookupCacheMaxDiskBytes = + cacheBytes( conf.get( ConfigOptions .SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO)); @@ -222,7 +227,9 @@ class HistoricalLakeLookupManager implements AutoCloseable { .removalListener(this::onLookuperRemoved) .build(); this.lookupPermits = new Semaphore(maxQueuedHistoricalRequests); + this.lookuperIdSequence = new AtomicLong(); this.pendingLookups = ConcurrentHashMap.newKeySet(); + this.lookupRuntime = createLookupRuntime(conf); } private static com.github.benmanes.caffeine.cache.Scheduler createCacheScheduler( @@ -324,6 +331,12 @@ public void close() { pendingLookups.forEach(future -> future.cancel(true)); lakeTableLookupers.invalidateAll(); lakeTableLookupers.cleanUp(); + LakeTableLookupRuntime runtime; + synchronized (this) { + runtime = lookupRuntime; + lookupRuntime = null; + } + IOUtils.closeQuietly(runtime, "historical lake lookup runtime"); } private CompletableFuture submitLookup( @@ -374,6 +387,16 @@ int numInflightRequests() { return maxQueuedHistoricalRequests - lookupPermits.availablePermits(); } + @VisibleForTesting + boolean hasLookupRuntime() { + return lookupRuntime != null; + } + + @VisibleForTesting + long lookupCacheMaxDiskBytes() { + return lookupCacheMaxDiskBytes; + } + /** Applies dynamic historical lookup configuration changes. */ void reconfigure(Configuration newConf) { checkNotNull(newConf, "newConf must not be null."); @@ -385,13 +408,13 @@ void reconfigure(Configuration newConf) { ConfigOptions .SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS); synchronized (this) { - long newMaxBytesPerTable = - cacheBytesPerTable( + long newMaxDiskBytes = + cacheBytes( newConf.get( ConfigOptions .SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO)); - cacheLimitChanged = newMaxBytesPerTable != lookupCacheMaxDiskBytesPerTable; - lookupCacheMaxDiskBytesPerTable = newMaxBytesPerTable; + cacheLimitChanged = newMaxDiskBytes != lookupCacheMaxDiskBytes; + lookupCacheMaxDiskBytes = newMaxDiskBytes; lakeConfigChanged = hasLakeConfigChanged(conf, newConf); expirationChanged = @@ -399,6 +422,12 @@ void reconfigure(Configuration newConf) { conf.get( ConfigOptions .SERVER_HISTORICAL_PARTITION_LOOKUPER_CACHE_EXPIRE_AFTER_ACCESS)); + if (lakeConfigChanged && lookupRuntime == null) { + lookupRuntime = createLookupRuntime(newConf); + } + if (cacheLimitChanged && lookupRuntime != null) { + lookupRuntime.updateLookupCacheMaxDiskBytes(newMaxDiskBytes); + } // Publish the configuration before its version. A lookup that observes the new version // must also observe the matching configuration snapshot. conf = newConf; @@ -413,12 +442,11 @@ void reconfigure(Configuration newConf) { .get() .setExpiresAfter(newExpiration.toMillis(), TimeUnit.MILLISECONDS); } - if (lakeConfigChanged || cacheLimitChanged) { + if (lakeConfigChanged) { // Do not invalidate while holding this monitor: lookuper creation holds a cache key // lock before preparing the lookup directory under the same monitor. Invalidation // closes inactive lookupers immediately and active lookupers after their last lookup - // releases them. After a cache limit change, the next lookup creates a Paimon lookuper - // with the updated per-table limit. + // releases them. lakeTableLookupers.invalidateAll(); lakeTableLookupers.cleanUp(); } @@ -436,7 +464,6 @@ private LookupResultForBucket lookupInternal( createLookupContext(lookupData, tableInfo, schemaInfo, lookupMetricRecorder); long currentLakeConfigVersion = lakeConfigVersion; Configuration currentConf = conf; - long cacheSizeBytes = lookupCacheMaxDiskBytesPerTable; cachedLookuper = lakeTableLookupers .asMap() @@ -445,26 +472,22 @@ private LookupResultForBucket lookupInternal( (ignored, currentLookuper) -> { CachedLakeTableLookuper selectedLookuper = currentLookuper; // Create the lookuper lazily, and recreate it after schema, - // lake configuration, or server cache size changes so it - // reloads lake table/query state and uses the current - // settings. + // lake configuration changes so it reloads lake table/query + // state and uses the current settings. if (selectedLookuper == null || selectedLookuper.schemaId != context.schemaId || selectedLookuper.lakeConfigVersion - != currentLakeConfigVersion - || selectedLookuper.cacheSizeBytes - != cacheSizeBytes) { - File tableLookupDir = - FlussPaths.historicalLookupTableDir( - historicalLookupCacheRootDir, - context.tablePath, - context.tableId); + != currentLakeConfigVersion) { LakeTableLookuper lookuper = createLakeTableLookuper( context.tablePath, - tableLookupDir.getAbsolutePath(), tableInfo.getTableConfig(), - cacheSizeBytes, + cacheNamespace( + context.tableId, + context.schemaId, + currentLakeConfigVersion, + lookuperIdSequence + .getAndIncrement()), currentConf); selectedLookuper = new CachedLakeTableLookuper( @@ -472,8 +495,6 @@ private LookupResultForBucket lookupInternal( context.tablePath, context.schemaId, currentLakeConfigVersion, - cacheSizeBytes, - tableLookupDir, lookuper); } // Pin the lookuper before leaving the atomic cache update. @@ -556,9 +577,8 @@ private LookupContext createLookupContext( LakeTableLookuper createLakeTableLookuper( TablePath tablePath, - String ioTmpDir, TableConfig tableConfig, - long cacheSizeBytes, + String cacheNamespace, Configuration clusterConf) { DataLakeFormat dataLakeFormat = clusterConf.get(ConfigOptions.DATALAKE_FORMAT); if (dataLakeFormat == null) { @@ -578,14 +598,32 @@ LakeTableLookuper createLakeTableLookuper( "Historical lookup requires cluster lake storage properties to be configured."); } + LakeTableLookupRuntime runtime = lookupRuntime; + if (runtime == null) { + throw new LakeStorageNotConfiguredException( + "Historical lake lookup runtime has not been initialized."); + } + return runtime.createLakeTableLookuper( + tablePath, + new LakeTableLookupRuntime.Context( + Configuration.fromMap(lakeProperties), + cacheNamespace, + tableConfig, + diskWriteGuard)); + } + + private @Nullable LakeTableLookupRuntime createLookupRuntime(Configuration configuration) { + DataLakeFormat dataLakeFormat = configuration.get(ConfigOptions.DATALAKE_FORMAT); + Map lakeProperties = extractLakeProperties(configuration); + if (dataLakeFormat != DataLakeFormat.PAIMON || lakeProperties == null) { + return null; + } LakeStoragePlugin lakeStoragePlugin = LakeStoragePluginSetUp.fromDataLakeFormat(dataLakeFormat.toString(), pluginManager); LakeStorage lakeStorage = lakeStoragePlugin.createLakeStorage(Configuration.fromMap(lakeProperties)); - return lakeStorage.createLakeTableLookuper( - tablePath, - new LakeStorage.LookuperContext( - ioTmpDir, tableConfig, cacheSizeBytes, diskWriteGuard)); + return lakeStorage.createLakeTableLookupRuntime( + historicalLookupCacheRootDir.getAbsolutePath(), lookupCacheMaxDiskBytes); } private static boolean hasLakeConfigChanged(Configuration currentConf, Configuration newConf) { @@ -595,11 +633,16 @@ private static boolean hasLakeConfigChanged(Configuration currentConf, Configura extractLakeProperties(currentConf), extractLakeProperties(newConf)); } - private long cacheBytesPerTable(double ratio) { + private long cacheBytes(double ratio) { checkArgument(ratio > 0.0 && ratio <= 1.0, "ratio must be within (0.0, 1.0]."); long totalCacheBytes = Math.min(dataDirVolumeBytes, (long) Math.ceil(dataDirVolumeBytes * ratio)); - return Math.max(1L, totalCacheBytes / MAX_CACHED_TABLES); + return Math.max(1L, totalCacheBytes); + } + + private static String cacheNamespace( + long tableId, int schemaId, long lakeConfigVersion, long lookuperId) { + return tableId + "-" + schemaId + "-" + lakeConfigVersion + "-" + lookuperId; } /** Returns the most recently sampled historical lookup cache footprint, in bytes. */ @@ -635,28 +678,7 @@ private static long fileSize(Path path) { } private static void closeLookuper(CachedLakeTableLookuper cachedLookuper) { - closeLookuper(cachedLookuper.lookuper, cachedLookuper.tableLookupDir); - } - - private static void closeLookuper(LakeTableLookuper lookuper, File tableLookupDir) { - try { - IOUtils.closeQuietly(lookuper, "historical lake table lookuper"); - } finally { - deleteTableLookupDirIfEmpty(tableLookupDir); - } - } - - private static void deleteTableLookupDirIfEmpty(File tableLookupDir) { - if (FileUtils.isDirectoryEmpty(tableLookupDir)) { - try { - Files.deleteIfExists(tableLookupDir.toPath()); - } catch (IOException e) { - LOG.debug( - "Failed to delete empty historical lookup directory {}.", - tableLookupDir, - e); - } - } + IOUtils.closeQuietly(cachedLookuper.lookuper, "historical lake table lookuper"); } private static final class LookupContext { @@ -682,8 +704,6 @@ private static final class CachedLakeTableLookuper { private final TablePath tablePath; private final int schemaId; private final long lakeConfigVersion; - private final long cacheSizeBytes; - private final File tableLookupDir; private final LakeTableLookuper lookuper; private int activeLookups; private boolean invalidated; @@ -694,15 +714,11 @@ private CachedLakeTableLookuper( TablePath tablePath, int schemaId, long lakeConfigVersion, - long cacheSizeBytes, - File tableLookupDir, LakeTableLookuper lookuper) { this.tableId = tableId; this.tablePath = tablePath; this.schemaId = schemaId; this.lakeConfigVersion = lakeConfigVersion; - this.cacheSizeBytes = cacheSizeBytes; - this.tableLookupDir = tableLookupDir; this.lookuper = lookuper; } diff --git a/fluss-server/src/test/java/org/apache/fluss/server/lakehouse/TestingPaimonStoragePlugin.java b/fluss-server/src/test/java/org/apache/fluss/server/lakehouse/TestingPaimonStoragePlugin.java index 88ad329e04f..ca30f502291 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/lakehouse/TestingPaimonStoragePlugin.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/lakehouse/TestingPaimonStoragePlugin.java @@ -28,6 +28,8 @@ import org.apache.fluss.lake.lakestorage.LakeCatalog; import org.apache.fluss.lake.lakestorage.LakeStorage; import org.apache.fluss.lake.lakestorage.LakeStoragePlugin; +import org.apache.fluss.lake.lakestorage.LakeTableLookupRuntime; +import org.apache.fluss.lake.lakestorage.LakeTableLookuper; import org.apache.fluss.lake.serializer.SimpleVersionedSerializer; import org.apache.fluss.lake.source.LakeSource; import org.apache.fluss.lake.writer.LakeTieringFactory; @@ -86,6 +88,24 @@ public LakeCatalog createLakeCatalog() { public LakeSource createLakeSource(TablePath tablePath) { throw new UnsupportedOperationException("Not implemented"); } + + @Override + public LakeTableLookupRuntime createLakeTableLookupRuntime( + String ioTmpDir, long lookupCacheMaxDiskBytes) { + return new LakeTableLookupRuntime() { + @Override + public LakeTableLookuper createLakeTableLookuper( + TablePath tablePath, Context context) { + throw new UnsupportedOperationException("Not implemented"); + } + + @Override + public void updateLookupCacheMaxDiskBytes(long lookupCacheMaxDiskBytes) {} + + @Override + public void close() {} + }; + } } /** Paimon implementation of LakeCatalog for testing purpose. */ diff --git a/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java b/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java index afa53e5516b..59359ee7772 100644 --- a/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java +++ b/fluss-server/src/test/java/org/apache/fluss/server/replica/HistoricalLakeLookupManagerTest.java @@ -230,7 +230,6 @@ void testCleansAndCreatesLookupCacheDirectoryOnStartup() throws Exception { assertThat(staleLookupFile).doesNotExist(); assertThat(serverLookupDir).isDirectory(); lookupAndRun(manager, executor, PARTITION_TABLE_INFO); - assertThat(manager.createdIoTmpDirs.get(0)).startsWith(serverLookupDir.getAbsolutePath()); File liveLookupFile = new File(serverLookupDir, "live-lookup-file"); assertThat(liveLookupFile.createNewFile()).isTrue(); @@ -403,11 +402,39 @@ void testEvictsLookuperWhenCachedTableLimitIsExceeded() throws Exception { assertThat(manager.createdLookupers).hasSize(11); assertThat(manager.createdLookupers).filteredOn(lookuper -> lookuper.closed).hasSize(1); - assertThat(manager.createdCacheSizes).containsOnly(2L); + assertThat(manager.createdCacheNamespaces).doesNotHaveDuplicates(); + assertThat(manager.lookupCacheMaxDiskBytes()).isEqualTo(20L); assertThat(manager.cachedTableCount()).isEqualTo(10); assertThat(manager.capacityEvictions().getCount()).isEqualTo(1); } + @Test + void testUpdatesSharedCacheLimitWithoutReplacingLookuper() throws Exception { + ManualExecutor executor = new ManualExecutor(); + Configuration initialConf = conf(1); + initialConf.set( + ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO, 0.10); + TestingHistoricalLakeLookupManager manager = + new TestingHistoricalLakeLookupManager( + initialConf, + executor, + Ticker.systemTicker(), + Scheduler.disabledScheduler(), + 100L, + 0L); + manager.startup(NO_OP_SCHEDULER); + lookupAndRun(manager, executor, PARTITION_TABLE_INFO); + TestingLakeTableLookuper lookuper = manager.createdLookupers.get(0); + + Configuration newConf = new Configuration(initialConf); + newConf.set(ConfigOptions.SERVER_HISTORICAL_PARTITION_LOOKUP_CACHE_MAX_DISK_RATIO, 0.20); + manager.reconfigure(newConf); + + assertThat(manager.lookupCacheMaxDiskBytes()).isEqualTo(20L); + assertThat(lookuper.closed).isFalse(); + assertThat(manager.cachedTableCount()).isOne(); + } + @Test void testReconfiguresLakePropertiesAndInvalidatesLookuper() throws Exception { Configuration initialConf = conf(1); @@ -416,6 +443,7 @@ void testReconfiguresLakePropertiesAndInvalidatesLookuper() throws Exception { ManualExecutor executor = new ManualExecutor(); TestingHistoricalLakeLookupManager manager = new TestingHistoricalLakeLookupManager(initialConf, executor); + assertThat(manager.hasLookupRuntime()).isTrue(); manager.startup(NO_OP_SCHEDULER); lookupAndRun(manager, executor, PARTITION_TABLE_INFO); @@ -541,9 +569,8 @@ private static LookupResultForBucket lookupResultAndRun( private static final class TestingHistoricalLakeLookupManager extends HistoricalLakeLookupManager { private final List createdLookupers = new ArrayList<>(); - private final List createdIoTmpDirs = new ArrayList<>(); private final List createdTableConfigs = new ArrayList<>(); - private final List createdCacheSizes = new ArrayList<>(); + private final List createdCacheNamespaces = new ArrayList<>(); private final List createdClusterConfigs = new ArrayList<>(); private final long lookupCacheFileBytes; @@ -599,16 +626,17 @@ private TestingHistoricalLakeLookupManager( @Override LakeTableLookuper createLakeTableLookuper( TablePath tablePath, - String ioTmpDir, TableConfig tableConfig, - long cacheSizeBytes, + String cacheNamespace, Configuration clusterConf) { TestingLakeTableLookuper lookuper = - new TestingLakeTableLookuper(new File(ioTmpDir), lookupCacheFileBytes); + new TestingLakeTableLookuper( + FlussPaths.historicalLookupRootDir( + new File(clusterConf.get(ConfigOptions.DATA_DIR))), + lookupCacheFileBytes); createdLookupers.add(lookuper); - createdIoTmpDirs.add(ioTmpDir); createdTableConfigs.add(tableConfig); - createdCacheSizes.add(cacheSizeBytes); + createdCacheNamespaces.add(cacheNamespace); createdClusterConfigs.add(clusterConf); return lookuper; }