Add a dynamic setting to change skip_cache_factor and min_frequency for querycache

Signed-off-by: kkewwei <kewei.11@bytedance.com>
Signed-off-by: kkewwei <kkewwei@163.com>

test

Author: kkewwei

CHANGELOG.md

@@ -28,6 +28,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Use QueryCoordinatorContext for the rewrite in validate API. ([#18272](https://github.com/opensearch-project/OpenSearch/pull/18272))
 - Upgrade crypto kms plugin dependencies for AWS SDK v2.x. ([#18268](https://github.com/opensearch-project/OpenSearch/pull/18268))
 - Add support for `matched_fields` with the unified highlighter ([#18164](https://github.com/opensearch-project/OpenSearch/issues/18164))
+- Add a dynamic setting to change skip_cache_factor and min_frequency for querycache ([#18274]( ([#18274](URL_ADDRESS.com/opensearch-project/OpenSearch/pull/18274))
 
 ### Changed
 - Create generic DocRequest to better categorize ActionRequests ([#18269](https://github.com/opensearch-project/OpenSearch/pull/18269)))

server/src/main/java/org/opensearch/common/settings/ClusterSettings.java

@@ -297,6 +297,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 IndicesQueryCache.INDICES_CACHE_QUERY_SIZE_SETTING,
                 IndicesQueryCache.INDICES_CACHE_QUERY_COUNT_SETTING,
                 IndicesQueryCache.INDICES_QUERIES_CACHE_ALL_SEGMENTS_SETTING,
+                IndicesQueryCache.INDICES_QUERIES_CACHE_SKIP_CACHE_FACTOR,
                 IndicesService.CLUSTER_DEFAULT_INDEX_MAX_MERGE_AT_ONCE_SETTING,
                 IndicesService.CLUSTER_DEFAULT_INDEX_REFRESH_INTERVAL_SETTING,
                 IndicesService.CLUSTER_MINIMUM_INDEX_REFRESH_INTERVAL_SETTING,

server/src/main/java/org/opensearch/common/settings/IndexScopedSettings.java

@@ -185,6 +185,7 @@ public final class IndexScopedSettings extends AbstractScopedSettings {
                 IndexSettings.INDEX_SEARCH_IDLE_AFTER,
                 IndexSettings.INDEX_SEARCH_THROTTLED,
                 IndexSettings.INDEX_UNREFERENCED_FILE_CLEANUP,
+                IndexSettings.INDEX_QUERY_CACHE_MIN_FREQUENCY,
                 IndexFieldDataService.INDEX_FIELDDATA_CACHE_KEY,
                 FieldMapper.IGNORE_MALFORMED_SETTING,
                 FieldMapper.COERCE_SETTING,

server/src/main/java/org/opensearch/index/IndexSettings.java

@@ -783,6 +783,14 @@ public static IndexMergePolicy fromString(String text) {
         Property.IndexScope
     );
 
+    public static final Setting<Integer> INDEX_QUERY_CACHE_MIN_FREQUENCY = Setting.intSetting(
+        "index.query_cache.min_frequency",
+        5,
+        1,
+        Property.Dynamic,
+        Property.IndexScope
+    );
+
     private final Index index;
     private final Version version;
     private final Logger logger;
@@ -832,6 +840,7 @@ public static IndexMergePolicy fromString(String text) {
     private final RemoteStorePathStrategy remoteStorePathStrategy;
     private final boolean isTranslogMetadataEnabled;
     private volatile boolean allowDerivedField;
+    private volatile int queryCacheMinFrequency;
 
     /**
      * The maximum age of a retention lease before it is considered expired.
@@ -1079,6 +1088,7 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
         setEnableFuzzySetForDocId(scopedSettings.get(INDEX_DOC_ID_FUZZY_SET_ENABLED_SETTING));
         setDocIdFuzzySetFalsePositiveProbability(scopedSettings.get(INDEX_DOC_ID_FUZZY_SET_FALSE_POSITIVE_PROBABILITY_SETTING));
         isCompositeIndex = scopedSettings.get(StarTreeIndexSettings.IS_COMPOSITE_INDEX_SETTING);
+        queryCacheMinFrequency = scopedSettings.get(INDEX_QUERY_CACHE_MIN_FREQUENCY);
         scopedSettings.addSettingsUpdateConsumer(
             TieredMergePolicyProvider.INDEX_COMPOUND_FORMAT_SETTING,
             tieredMergePolicyProvider::setNoCFSRatio
@@ -1201,6 +1211,15 @@ public IndexSettings(final IndexMetadata indexMetadata, final Settings nodeSetti
             IndexMetadata.INDEX_REMOTE_TRANSLOG_REPOSITORY_SETTING,
             this::setRemoteStoreTranslogRepository
         );
+        scopedSettings.addSettingsUpdateConsumer(INDEX_QUERY_CACHE_MIN_FREQUENCY, this::setQueryCacheMinFrequency);
+    }
+
+    private void setQueryCacheMinFrequency(int queryCacheMinFrequency) {
+        this.queryCacheMinFrequency = queryCacheMinFrequency;
+    }
+
+    public int getQueryCacheMinFrequency() {
+        return queryCacheMinFrequency;
     }
 
     private void setSearchIdleAfter(TimeValue searchIdleAfter) {

server/src/main/java/org/opensearch/index/shard/IndexShard.java

@@ -43,10 +43,14 @@
 import org.apache.lucene.index.LeafReader;
 import org.apache.lucene.index.SegmentInfos;
 import org.apache.lucene.index.Term;
+import org.apache.lucene.search.BooleanQuery;
+import org.apache.lucene.search.DisjunctionMaxQuery;
+import org.apache.lucene.search.MultiTermQuery;
 import org.apache.lucene.search.Query;
 import org.apache.lucene.search.QueryCachingPolicy;
 import org.apache.lucene.search.ReferenceManager;
 import org.apache.lucene.search.Sort;
+import org.apache.lucene.search.TermInSetQuery;
 import org.apache.lucene.search.UsageTrackingQueryCachingPolicy;
 import org.apache.lucene.store.AlreadyClosedException;
 import org.apache.lucene.store.BufferedChecksumIndexInput;
@@ -495,7 +499,7 @@ public boolean shouldCache(Query query) {
                 }
             };
         } else {
-            cachingPolicy = new UsageTrackingQueryCachingPolicy();
+            cachingPolicy = new OpenseachUsageTrackingQueryCachingPolicy(() -> indexSettings.getQueryCacheMinFrequency());
         }
         indexShardOperationPermits = new IndexShardOperationPermits(shardId, threadPool);
         readerWrapper = indexReaderWrapper;
@@ -5578,4 +5582,52 @@ public void stopRefreshTask() {
     public AsyncShardRefreshTask getRefreshTask() {
         return refreshTask;
     }
+
+    /**
+     * Custom caching policy for Opensearch.
+     */
+    public static class OpenseachUsageTrackingQueryCachingPolicy extends UsageTrackingQueryCachingPolicy {
+        private Supplier<Integer> queryCacheMinFrequency;
+
+        public OpenseachUsageTrackingQueryCachingPolicy(Supplier<Integer> queryCacheMinFrequency) {
+            this.queryCacheMinFrequency = queryCacheMinFrequency;
+        }
+
+        @Override
+        protected int minFrequencyToCache(Query query) {
+            if (isCostly(query)) {
+                return 2;
+            }
+            int minFrequency = queryCacheMinFrequency.get();
+            if (query instanceof BooleanQuery || query instanceof DisjunctionMaxQuery) {
+                --minFrequency;
+            }
+
+            return Math.max(1, minFrequency);
+        }
+
+        /**
+         * Same to Lucene's UsageTrackingQueryCachingPolicy.isCostly, it's not public in Lucene.
+         * Given that lucene doesn't give the desired extensibility at this point.
+         * Also, we can extension point if needed.
+          */
+        private boolean isCostly(Query query) {
+            return query instanceof MultiTermQuery
+                || query.getClass().getSimpleName().equals("MultiTermQueryConstantScoreBlendedWrapper")
+                || query.getClass().getSimpleName().equals("MultiTermQueryConstantScoreWrapper")
+                || query instanceof TermInSetQuery
+                || isPointQuery(query);
+        }
+
+        // Same to Lucene's UsageTrackingQueryCachingPolicy.isPointQuery
+        private boolean isPointQuery(Query query) {
+            for (Class<?> clazz = query.getClass(); clazz != Query.class; clazz = clazz.getSuperclass()) {
+                final String simpleName = clazz.getSimpleName();
+                if (simpleName.startsWith("Point") && simpleName.endsWith("Query")) {
+                    return true;
+                }
+            }
+            return false;
+        }
+    }
 }

server/src/main/java/org/opensearch/indices/IndicesQueryCache.java

@@ -44,6 +44,7 @@
 import org.apache.lucene.search.Weight;
 import org.opensearch.common.annotation.PublicApi;
 import org.opensearch.common.lucene.ShardCoreKeyMap;
+import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Setting;
 import org.opensearch.common.settings.Setting.Property;
 import org.opensearch.common.settings.Settings;
@@ -91,6 +92,15 @@ public class IndicesQueryCache implements QueryCache, Closeable {
         Property.NodeScope
     );
 
+    // dynamic change the skipCacheFactor for query_cache
+    public static final Setting<Float> INDICES_QUERIES_CACHE_SKIP_CACHE_FACTOR = Setting.floatSetting(
+        "indices.queries.cache.skip_cache_factor",
+        10,
+        1,
+        Property.NodeScope,
+        Property.Dynamic
+    );
+
     private final LRUQueryCache cache;
     private final ShardCoreKeyMap shardKeyMap = new ShardCoreKeyMap();
     private final Map<ShardId, Stats> shardStats = new ConcurrentHashMap<>();
@@ -101,18 +111,43 @@ public class IndicesQueryCache implements QueryCache, Closeable {
     // See onDocIdSetEviction for more info
     private final Map<Object, StatsAndCount> stats2 = Collections.synchronizedMap(new IdentityHashMap<>());
 
+    // Compatible for public api
     public IndicesQueryCache(Settings settings) {
+        this(settings, null);
+    }
+
+    public IndicesQueryCache(Settings settings, ClusterSettings clusterSettings) {
         final ByteSizeValue size = INDICES_CACHE_QUERY_SIZE_SETTING.get(settings);
         final int count = INDICES_CACHE_QUERY_COUNT_SETTING.get(settings);
-        logger.debug("using [node] query cache with size [{}] max filter count [{}]", size, count);
+        float skipCacheFactor = INDICES_QUERIES_CACHE_SKIP_CACHE_FACTOR.get(settings);
+        logger.debug("using [node] query cache with size [{}] max filter count [{}] skipCacheFactor [{}]", size, count, skipCacheFactor);
         if (INDICES_QUERIES_CACHE_ALL_SEGMENTS_SETTING.get(settings)) {
             cache = new OpenSearchLRUQueryCache(count, size.getBytes(), context -> true, 1f);
         } else {
             cache = new OpenSearchLRUQueryCache(count, size.getBytes());
+            cache.setSkipCacheFactor(skipCacheFactor);
+            if (clusterSettings != null) {
+                clusterSettings.addSettingsUpdateConsumer(INDICES_QUERIES_CACHE_SKIP_CACHE_FACTOR, this::setSkipCacheFactor);
+            } else {
+                logger.warn("clusterSettings is null, so {} is not dynamic", INDICES_QUERIES_CACHE_SKIP_CACHE_FACTOR.getKey());
+            }
         }
         sharedRamBytesUsed = 0;
     }
 
+    public void setSkipCacheFactor(float skipCacheFactor) {
+        if (skipCacheFactor < 1) {
+            throw new IllegalArgumentException("skipCacheFactor must be no less than 1, get " + skipCacheFactor);
+        }
+        logger.debug(
+            "set cluster settings {} {} -> {}",
+            INDICES_QUERIES_CACHE_SKIP_CACHE_FACTOR.getKey(),
+            this.cache.getSkipCacheFactor(),
+            skipCacheFactor
+        );
+        cache.setSkipCacheFactor(skipCacheFactor);
+    }
+
     /** Get usage statistics for the given shard. */
     public QueryCacheStats getStats(ShardId shard) {
         final Map<ShardId, QueryCacheStats> stats = new HashMap<>();

server/src/main/java/org/opensearch/indices/IndicesService.java

@@ -471,7 +471,7 @@ public IndicesService(
             }
             return Optional.of(new IndexShardCacheEntity(indexService.getShardOrNull(shardId.id())));
         }), cacheService, threadPool, clusterService, nodeEnv);
-        this.indicesQueryCache = new IndicesQueryCache(settings);
+        this.indicesQueryCache = new IndicesQueryCache(settings, clusterService.getClusterSettings());
         this.mapperRegistry = mapperRegistry;
         this.namedWriteableRegistry = namedWriteableRegistry;
         indexingMemoryController = new IndexingMemoryController(
@@ -1195,7 +1195,7 @@ public synchronized void verifyIndexMetadata(IndexMetadata metadata, IndexMetada
             IndicesFieldDataCache indicesFieldDataCache = new IndicesFieldDataCache(settings, new IndexFieldDataCache.Listener() {
             });
             closeables.add(indicesFieldDataCache);
-            IndicesQueryCache indicesQueryCache = new IndicesQueryCache(settings);
+            IndicesQueryCache indicesQueryCache = new IndicesQueryCache(settings, clusterService.getClusterSettings());
             closeables.add(indicesQueryCache);
             // this will also fail if some plugin fails etc. which is nice since we can verify that early
             final IndexService service = createIndexService(

server/src/test/java/org/opensearch/indices/IndicesQueryCacheTests.java

@@ -33,9 +33,14 @@
 package org.opensearch.indices;
 
 import org.apache.lucene.document.Document;
+import org.apache.lucene.document.IntPoint;
 import org.apache.lucene.index.DirectoryReader;
 import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.IndexWriterConfig;
 import org.apache.lucene.index.LeafReaderContext;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.BooleanClause;
+import org.apache.lucene.search.BooleanQuery;
 import org.apache.lucene.search.ConstantScoreScorer;
 import org.apache.lucene.search.ConstantScoreWeight;
 import org.apache.lucene.search.DocIdSetIterator;
@@ -48,16 +53,23 @@
 import org.apache.lucene.search.ScoreMode;
 import org.apache.lucene.search.Scorer;
 import org.apache.lucene.search.ScorerSupplier;
+import org.apache.lucene.search.TermQuery;
+import org.apache.lucene.search.TotalHitCountCollector;
 import org.apache.lucene.search.Weight;
 import org.apache.lucene.store.Directory;
+import org.apache.lucene.util.BytesRef;
 import org.opensearch.common.lucene.index.OpenSearchDirectoryReader;
+import org.opensearch.common.settings.ClusterSettings;
 import org.opensearch.common.settings.Settings;
 import org.opensearch.common.util.io.IOUtils;
 import org.opensearch.core.index.shard.ShardId;
 import org.opensearch.index.cache.query.QueryCacheStats;
+import org.opensearch.index.mapper.KeywordFieldMapper;
+import org.opensearch.index.shard.IndexShard;
 import org.opensearch.test.OpenSearchTestCase;
 
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class IndicesQueryCacheTests extends OpenSearchTestCase {
 
@@ -492,4 +504,72 @@ public void testDelegatesCount() throws Exception {
         cache.onClose(shard);
         cache.close();
     }
+
+    public void testDynamicChangeSettings() throws IOException {
+        Directory dir = newDirectory();
+        IndexWriterConfig conf = newIndexWriterConfig();
+        conf.setMaxBufferedDocs(100000);
+        IndexWriter w = new IndexWriter(dir, conf);
+        // lucene will not cache segment whose docs is smaller than 10,000
+        for (int i = 0; i < 10001; i++) {
+            Document document = new Document();
+            document.add(new IntPoint("age", i));
+            final BytesRef binaryValue = new BytesRef(String.valueOf(i));
+            document.add(new KeywordFieldMapper.KeywordField("name", binaryValue, KeywordFieldMapper.Defaults.FIELD_TYPE));
+            w.addDocument(document);
+        }
+        DirectoryReader r = DirectoryReader.open(w);
+        w.close();
+        ShardId shard = new ShardId("index", "_na_", 0);
+        r = OpenSearchDirectoryReader.wrap(r, shard);
+
+        AtomicInteger queryCacheMinFrequency = new AtomicInteger(5);
+        IndexSearcher searcher = new IndexSearcher(r);
+        searcher.setQueryCachingPolicy(new IndexShard.OpenseachUsageTrackingQueryCachingPolicy(() -> queryCacheMinFrequency.get()));
+        Settings settings = Settings.builder().build();
+        ClusterSettings clusterSettings = new ClusterSettings(settings, ClusterSettings.BUILT_IN_CLUSTER_SETTINGS);
+        IndicesQueryCache cache = new IndicesQueryCache(settings, clusterSettings);
+        searcher.setQueryCache(cache);
+        final TotalHitCountCollector collector = new TotalHitCountCollector();
+
+        // test changing queryCacheMinFrequency
+        {
+            BooleanQuery.Builder booleanQuery = new BooleanQuery.Builder();
+            booleanQuery.add(IntPoint.newRangeQuery("age", 1, 9999), BooleanClause.Occur.FILTER);
+            booleanQuery.add(new TermQuery(new Term("name", "1")), BooleanClause.Occur.FILTER);
+            searcher.search(booleanQuery.build(), collector);
+            QueryCacheStats stats = cache.getStats(shard);
+            assertEquals(0L, stats.getCacheSize());
+            assertEquals(0L, stats.getCacheCount());
+            assertEquals(0L, stats.getHitCount());
+            assertEquals(6L, stats.getMissCount());
+
+            queryCacheMinFrequency.set(2);
+            searcher.search(booleanQuery.build(), collector);
+            stats = cache.getStats(shard);
+            // ensure result is cached
+            assertEquals(1L, stats.getCacheSize());
+            assertEquals(1L, stats.getCacheCount());
+        }
+
+        // test changing skipCacheFactor
+        {
+            // make sure the TermInSetQuery can be cached, because queryCacheMinFrequency is 2
+            queryCacheMinFrequency.set(4);
+            BooleanQuery.Builder booleanQuery = new BooleanQuery.Builder();
+            booleanQuery.add(IntPoint.newRangeQuery("age", 2, 9999), BooleanClause.Occur.FILTER);
+            booleanQuery.add(new TermQuery(new Term("name", "3")), BooleanClause.Occur.MUST);
+            cache.setSkipCacheFactor(20000);
+            searcher.search(booleanQuery.build(), collector);
+            searcher.search(booleanQuery.build(), collector);
+            QueryCacheStats stats = cache.getStats(shard);
+            // only range can be cached, assert the range query has been cached because of skipCacheFactor
+            assertEquals(2L, stats.getCacheSize());
+            assertEquals(2L, stats.getCacheCount());
+            assertEquals(0L, stats.getHitCount());
+        }
+        IOUtils.close(r, dir);
+        cache.onClose(shard);
+        cache.close(); // this triggers some assertions
+    }
 }