[server] Add support of sharable file cache capacity for historical partition lookup

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<Duration>

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;
-        }
-    }
 }

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;
+        }
+    }
+}

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<LakeTableLookupRuntime> {
+
+        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

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 {

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<PaimonSplit> 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");
+        }
     }
 }