Reuse TantivySearcher across queries via searcher pool

Author: chenghuichen

paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexer.java

@@ -36,7 +36,9 @@ public class TantivyFullTextGlobalIndexer implements GlobalIndexer {
     private final TantivySearcherPool searcherPool;
 
     public TantivyFullTextGlobalIndexer() {
-        this(new TantivySearcherPool(Runtime.getRuntime().availableProcessors() * 2));
+        this(
+                new TantivySearcherPool(
+                        TantivyFullTextIndexOptions.SEARCHER_POOL_MAX_SIZE.defaultValue()));
     }
 
     public TantivyFullTextGlobalIndexer(TantivySearcherPool searcherPool) {

paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextGlobalIndexerFactory.java

@@ -32,9 +32,11 @@ public class TantivyFullTextGlobalIndexerFactory implements GlobalIndexerFactory
      * Shared pool across all indexers created by this factory. This factory instance is a JVM-level
      * singleton (loaded once via {@link java.util.ServiceLoader}), so the pool naturally survives
      * across queries and scanners.
+     *
+     * <p>The pool is initialized lazily on the first {@link #create} call so that the pool size can
+     * be read from user-supplied options.
      */
-    private final TantivySearcherPool searcherPool =
-            new TantivySearcherPool(Runtime.getRuntime().availableProcessors() * 2);
+    private volatile TantivySearcherPool searcherPool;
 
     @Override
     public String identifier() {
@@ -43,6 +45,14 @@ public String identifier() {
 
     @Override
     public GlobalIndexer create(DataField field, Options options) {
+        if (searcherPool == null) {
+            synchronized (this) {
+                if (searcherPool == null) {
+                    int maxSize = options.get(TantivyFullTextIndexOptions.SEARCHER_POOL_MAX_SIZE);
+                    searcherPool = new TantivySearcherPool(maxSize);
+                }
+            }
+        }
         return new TantivyFullTextGlobalIndexer(searcherPool);
     }
 }

paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivyFullTextIndexOptions.java

@@ -0,0 +1,37 @@
+/*
+ * 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.paimon.tantivy.index;
+
+import org.apache.paimon.options.ConfigOption;
+import org.apache.paimon.options.ConfigOptions;
+
+/** Options for the Tantivy full-text index. */
+public class TantivyFullTextIndexOptions {
+
+    public static final ConfigOption<Integer> SEARCHER_POOL_MAX_SIZE =
+            ConfigOptions.key("tantivy.searcher-pool.max-size")
+                    .intType()
+                    .defaultValue(32)
+                    .withDescription(
+                            "Maximum number of idle TantivySearcher instances kept in the pool "
+                                    + "across all index shards. Each entry holds the index open in "
+                                    + "Rust memory (including the FST term dictionary), so memory "
+                                    + "usage scales with this value times the index size per shard. "
+                                    + "Set to 0 to disable pooling.");
+}

paimon-tantivy/paimon-tantivy-index/src/main/java/org/apache/paimon/tantivy/index/TantivySearcherPool.java

@@ -22,24 +22,33 @@
 import org.apache.paimon.tantivy.TantivySearcher;
 import org.apache.paimon.utils.IOUtils;
 
+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 javax.annotation.Nullable;
 
 import java.io.Closeable;
 import java.io.IOException;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
 
 /**
  * Pool of {@link TantivySearcher} instances keyed by index file identity ({@code filePath@size}).
  *
  * <p>Each searcher holds the Tantivy index open in Rust memory (including the FST term dictionary).
  * Pooling avoids the repeated cost of loading the index on every query.
  *
+ * <p>At most one idle searcher is kept per key. Under concurrent queries on the same shard, the
+ * last entry to be returned wins; the others are closed immediately. Entries idle for more than
+ * {@link #EXPIRE_AFTER_ACCESS_MINUTES} minutes are evicted and closed automatically. The total
+ * number of idle entries across all keys is bounded by {@code maxSize}.
+ *
  * <p>Thread-safe. Borrow/return semantics guarantee at most one thread uses a given entry at a
  * time.
  */
 public class TantivySearcherPool {
 
+    static final long EXPIRE_AFTER_ACCESS_MINUTES = 30;
+
     /** A borrowed searcher + its backing stream. Both are returned together. */
     static final class PooledEntry implements Closeable {
         final TantivySearcher searcher;
@@ -57,12 +66,21 @@ public void close() throws IOException {
         }
     }
 
-    private final int maxSizePerKey;
-    private final ConcurrentHashMap<String, LinkedBlockingDeque<PooledEntry>> pool =
-            new ConcurrentHashMap<>();
+    @Nullable private final Cache<String, PooledEntry> idleCache;
 
-    public TantivySearcherPool(int maxSizePerKey) {
-        this.maxSizePerKey = maxSizePerKey;
+    public TantivySearcherPool(int maxSize) {
+        if (maxSize <= 0) {
+            this.idleCache = null;
+        } else {
+            Cache<String, PooledEntry> cache =
+                    Caffeine.newBuilder()
+                            .maximumSize(maxSize)
+                            .expireAfterAccess(EXPIRE_AFTER_ACCESS_MINUTES, TimeUnit.MINUTES)
+                            .executor(Runnable::run)
+                            .removalListener((k, v, c) -> IOUtils.closeQuietly((PooledEntry) v))
+                            .build();
+            this.idleCache = cache;
+        }
     }
 
     /**
@@ -73,24 +91,21 @@ public TantivySearcherPool(int maxSizePerKey) {
      */
     @Nullable
     public PooledEntry borrow(String key) {
-        LinkedBlockingDeque<PooledEntry> deque = pool.get(key);
-        return deque == null ? null : deque.pollFirst();
+        if (idleCache == null) {
+            return null;
+        }
+        return idleCache.asMap().remove(key);
     }
 
     /**
-     * Return a previously borrowed entry. If the pool is full, the entry is closed immediately.
-     *
-     * <p>The stream position after use is irrelevant — Rust always seeks before reading.
+     * Return a previously borrowed entry to the pool. Any entry displaced by size eviction, TTL
+     * expiry, or key replacement is closed automatically via the removal listener.
      */
     public void returnEntry(String key, PooledEntry entry) {
-        if (maxSizePerKey <= 0) {
+        if (idleCache == null) {
             IOUtils.closeQuietly(entry);
             return;
         }
-        LinkedBlockingDeque<PooledEntry> deque =
-                pool.computeIfAbsent(key, k -> new LinkedBlockingDeque<>(maxSizePerKey));
-        if (!deque.offerFirst(entry)) {
-            IOUtils.closeQuietly(entry);
-        }
+        idleCache.put(key, entry);
     }
 }