Time series based workload desc order optimization through reverse segment read

CHANGELOG.md

@@ -104,6 +104,7 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 - Introduce full support for searchable snapshots ([#5087](https://github.com/opensearch-project/OpenSearch/issues/5087))
 - Introduce full support for Segment Replication ([#5147](https://github.com/opensearch-project/OpenSearch/issues/5147))
 - [Remote Store] Add support for enabling Remote Store at Cluster level by default. ([#6932](https://github.com/opensearch-project/OpenSearch/pull/6932))
+- Time series based workload desc order optimization through reverse segment read. ([#7244](https://github.com/opensearch-project/OpenSearch/pull/7244))
 
 ### Dependencies
 - Bump `org.apache.logging.log4j:log4j-core` from 2.18.0 to 2.20.0 ([#6490](https://github.com/opensearch-project/OpenSearch/pull/6490))
@@ -146,4 +147,4 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 ### Security
 
 [Unreleased 3.0]: https://github.com/opensearch-project/OpenSearch/compare/2.x...HEAD
-[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.5...2.x
+[Unreleased 2.x]: https://github.com/opensearch-project/OpenSearch/compare/2.5...2.x

sandbox/plugins/concurrent-search/src/test/java/org/opensearch/search/profile/query/QueryProfilerTests.java

@@ -96,7 +96,8 @@ public void setUp() throws Exception {
             IndexSearcher.getDefaultQueryCache(),
             ALWAYS_CACHE_POLICY,
             true,
-            executor
+            executor,
+            false
         );
     }
 

sandbox/plugins/concurrent-search/src/test/java/org/opensearch/search/query/QueryPhaseTests.java

@@ -1152,7 +1152,8 @@ private static ContextIndexSearcher newContextSearcher(IndexReader reader, Execu
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            executor
+            executor,
+            false
         );
     }
 
@@ -1164,7 +1165,8 @@ private static ContextIndexSearcher newEarlyTerminationContextSearcher(IndexRead
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            executor
+            executor,
+            false
         ) {
 
             @Override

sandbox/plugins/concurrent-search/src/test/java/org/opensearch/search/query/QueryProfilePhaseTests.java

@@ -1164,7 +1164,8 @@ private static ContextIndexSearcher newContextSearcher(IndexReader reader, Execu
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            executor
+            executor,
+            false
         );
     }
 
@@ -1176,7 +1177,8 @@ private static ContextIndexSearcher newEarlyTerminationContextSearcher(IndexRead
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            executor
+            executor,
+            false
         ) {
 
             @Override

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

@@ -488,6 +488,7 @@ public void apply(Settings value, Settings current, Settings previous) {
                 SearchService.MAX_OPEN_SCROLL_CONTEXT,
                 SearchService.MAX_OPEN_PIT_CONTEXT,
                 SearchService.MAX_PIT_KEEPALIVE_SETTING,
+                SearchService.SEARCH_SEGMENTS_REVERSE_ORDER_OPTIMIZATION,
                 CreatePitController.PIT_INIT_KEEP_ALIVE,
                 Node.WRITE_PORTS_FILE_SETTING,
                 Node.NODE_NAME_SETTING,

server/src/main/java/org/opensearch/search/DefaultSearchContext.java

@@ -188,7 +188,8 @@ final class DefaultSearchContext extends SearchContext {
         boolean lowLevelCancellation,
         Version minNodeVersion,
         boolean validate,
-        Executor executor
+        Executor executor,
+        boolean searchSegmentOrderReversed
     ) throws IOException {
         this.readerContext = readerContext;
         this.request = request;
@@ -210,7 +211,8 @@ final class DefaultSearchContext extends SearchContext {
             engineSearcher.getQueryCache(),
             engineSearcher.getQueryCachingPolicy(),
             lowLevelCancellation,
-            executor
+            executor,
+            searchSegmentOrderReversed
         );
         this.relativeTimeSupplier = relativeTimeSupplier;
         this.timeout = timeout;

server/src/main/java/org/opensearch/search/SearchService.java

@@ -133,6 +133,7 @@
 import org.opensearch.search.sort.MinAndMax;
 import org.opensearch.search.sort.SortAndFormats;
 import org.opensearch.search.sort.SortBuilder;
+import org.opensearch.search.sort.SortOrder;
 import org.opensearch.search.suggest.Suggest;
 import org.opensearch.search.suggest.completion.CompletionSuggestion;
 import org.opensearch.threadpool.Scheduler.Cancellable;
@@ -234,6 +235,13 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
         Property.NodeScope
     );
 
+    public static final Setting<Boolean> SEARCH_SEGMENTS_REVERSE_ORDER_OPTIMIZATION = Setting.boolSetting(
+        "search.search_segments_reverse_order_optimization",
+        false,
+        Property.Dynamic,
+        Property.NodeScope
+    );
+
     /**
      * This setting defines the maximum number of active PIT reader contexts in the node , since each PIT context
      * has a resource cost attached to it. This setting is less than scroll since users are
@@ -282,6 +290,8 @@ public class SearchService extends AbstractLifecycleComponent implements IndexEv
 
     private volatile int maxOpenScrollContext;
 
+    private volatile boolean searchSegmentOrderReversed;
+
     private volatile int maxOpenPitContext;
 
     private final Cancellable keepAliveReaper;
@@ -356,6 +366,10 @@ public SearchService(
 
         lowLevelCancellation = LOW_LEVEL_CANCELLATION_SETTING.get(settings);
         clusterService.getClusterSettings().addSettingsUpdateConsumer(LOW_LEVEL_CANCELLATION_SETTING, this::setLowLevelCancellation);
+
+        searchSegmentOrderReversed = SEARCH_SEGMENTS_REVERSE_ORDER_OPTIMIZATION.get(settings);
+        clusterService.getClusterSettings()
+            .addSettingsUpdateConsumer(SEARCH_SEGMENTS_REVERSE_ORDER_OPTIMIZATION, this::setSearchSegmentOrderReversed);
     }
 
     private void validateKeepAlives(TimeValue defaultKeepAlive, TimeValue maxKeepAlive) {
@@ -432,6 +446,14 @@ private void setLowLevelCancellation(Boolean lowLevelCancellation) {
         this.lowLevelCancellation = lowLevelCancellation;
     }
 
+    private void setSearchSegmentOrderReversed(boolean reversed) {
+        this.searchSegmentOrderReversed = reversed;
+    }
+
+    private boolean getSearchSegmentOrderReversed() {
+        return this.searchSegmentOrderReversed;
+    }
+
     @Override
     public void afterIndexRemoved(Index index, IndexSettings indexSettings, IndexRemovalReason reason) {
         // once an index is removed due to deletion or closing, we can just clean up all the pending search context information
@@ -1038,7 +1060,8 @@ private DefaultSearchContext createSearchContext(ReaderContext reader, ShardSear
                 lowLevelCancellation,
                 clusterService.state().nodes().getMinNodeVersion(),
                 validate,
-                indexSearcherExecutor
+                indexSearcherExecutor,
+                shouldReverseLeafReaderContexts(request)
             );
             // we clone the query shard context here just for rewriting otherwise we
             // might end up with incorrect state since we are using now() or script services
@@ -1058,6 +1081,26 @@ private DefaultSearchContext createSearchContext(ReaderContext reader, ShardSear
         return searchContext;
     }
 
+    private boolean shouldReverseLeafReaderContexts(ShardSearchRequest request) {
+        // By default, if search query is on desc order, we will search leaves in reverse order.
+        // This is beneficial for desc order sort queries on time series based workload,
+        // where recent data is always on new segments which are in last.
+        // This won't regress or impact other type of workload where data is randomly distributed
+        // across segments. So turning it on by default.
+        // searchSegmentOrderReversed is true by default
+        if (searchSegmentOrderReversed) {
+            // Only reverse order for desc order sort queries
+            if (request != null
+                && request.source() != null
+                && request.source().sorts() != null
+                && request.source().sorts().size() > 0
+                && request.source().sorts().get(0).order() == SortOrder.DESC) {
+                return true;
+            }
+        }
+        return false;
+    }
+
     private void freeAllContextForIndex(Index index) {
         assert index != null;
         for (ReaderContext ctx : activeReaders.values()) {

server/src/main/java/org/opensearch/search/internal/ContextIndexSearcher.java

@@ -76,6 +76,7 @@
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Objects;
@@ -98,15 +99,31 @@ public class ContextIndexSearcher extends IndexSearcher implements Releasable {
     private QueryProfiler profiler;
     private MutableQueryTimeout cancellable;
 
+    /**
+     * Certain queries can benefit if we reverse the segment read order,
+     * for example time series based queries if searched for desc sort order
+     */
+    private boolean reverseLeafReaderContexts;
+
     public ContextIndexSearcher(
         IndexReader reader,
         Similarity similarity,
         QueryCache queryCache,
         QueryCachingPolicy queryCachingPolicy,
         boolean wrapWithExitableDirectoryReader,
-        Executor executor
+        Executor executor,
+        boolean reverseLeafReaderContexts
     ) throws IOException {
-        this(reader, similarity, queryCache, queryCachingPolicy, new MutableQueryTimeout(), wrapWithExitableDirectoryReader, executor);
+        this(
+            reader,
+            similarity,
+            queryCache,
+            queryCachingPolicy,
+            new MutableQueryTimeout(),
+            wrapWithExitableDirectoryReader,
+            executor,
+            reverseLeafReaderContexts
+        );
     }
 
     private ContextIndexSearcher(
@@ -116,13 +133,15 @@ private ContextIndexSearcher(
         QueryCachingPolicy queryCachingPolicy,
         MutableQueryTimeout cancellable,
         boolean wrapWithExitableDirectoryReader,
-        Executor executor
+        Executor executor,
+        boolean reverseLeafReaderContexts
     ) throws IOException {
         super(wrapWithExitableDirectoryReader ? new ExitableDirectoryReader((DirectoryReader) reader, cancellable) : reader, executor);
         setSimilarity(similarity);
         setQueryCache(queryCache);
         setQueryCachingPolicy(queryCachingPolicy);
         this.cancellable = cancellable;
+        this.reverseLeafReaderContexts = reverseLeafReaderContexts;
     }
 
     public void setProfiler(QueryProfiler profiler) {
@@ -246,6 +265,12 @@ public void search(
 
     @Override
     protected void search(List<LeafReaderContext> leaves, Weight weight, Collector collector) throws IOException {
+        if (reverseLeafReaderContexts) {
+            // reverse the segment search order if this flag is true.
+            // We need to make a copy, so we don't corrupt DirectoryReader leaves order
+            leaves = new ArrayList<>(leaves);
+            Collections.reverse(leaves);
+        }
         for (LeafReaderContext ctx : leaves) { // search each subreader
             searchLeaf(ctx, weight, collector);
         }

server/src/test/java/org/opensearch/search/DefaultSearchContextTests.java

@@ -213,7 +213,8 @@ protected Engine.Searcher acquireSearcherInternal(String source) {
                 false,
                 Version.CURRENT,
                 false,
-                executor
+                executor,
+                false
             );
             contextWithoutScroll.from(300);
             contextWithoutScroll.close();
@@ -255,7 +256,8 @@ protected Engine.Searcher acquireSearcherInternal(String source) {
                 false,
                 Version.CURRENT,
                 false,
-                executor
+                executor,
+                false
             );
             context1.from(300);
             exception = expectThrows(IllegalArgumentException.class, () -> context1.preProcess(false));
@@ -325,7 +327,8 @@ protected Engine.Searcher acquireSearcherInternal(String source) {
                 false,
                 Version.CURRENT,
                 false,
-                executor
+                executor,
+                false
             );
 
             SliceBuilder sliceBuilder = mock(SliceBuilder.class);
@@ -364,7 +367,8 @@ protected Engine.Searcher acquireSearcherInternal(String source) {
                 false,
                 Version.CURRENT,
                 false,
-                executor
+                executor,
+                false
             );
             ParsedQuery parsedQuery = ParsedQuery.parsedMatchAllQuery();
             context3.sliceBuilder(null).parsedQuery(parsedQuery).preProcess(false);
@@ -399,7 +403,8 @@ protected Engine.Searcher acquireSearcherInternal(String source) {
                 false,
                 Version.CURRENT,
                 false,
-                executor
+                executor,
+                false
             );
             context4.sliceBuilder(new SliceBuilder(1, 2)).parsedQuery(parsedQuery).preProcess(false);
             Query query1 = context4.query();
@@ -429,7 +434,8 @@ protected Engine.Searcher acquireSearcherInternal(String source) {
                 false,
                 Version.CURRENT,
                 false,
-                executor
+                executor,
+                false
             );
             int numSlicesForPit = maxSlicesPerPit + randomIntBetween(1, 100);
             when(sliceBuilder.getMax()).thenReturn(numSlicesForPit);
@@ -526,7 +532,8 @@ protected Engine.Searcher acquireSearcherInternal(String source) {
                 false,
                 Version.CURRENT,
                 false,
-                executor
+                executor,
+                false
             );
             assertThat(context.searcher().hasCancellations(), is(false));
             context.searcher().addQueryCancellation(() -> {});

server/src/test/java/org/opensearch/search/SearchCancellationTests.java

@@ -109,7 +109,8 @@ public void testAddingCancellationActions() throws IOException {
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            null
+            null,
+            false
         );
         NullPointerException npe = expectThrows(NullPointerException.class, () -> searcher.addQueryCancellation(null));
         assertEquals("cancellation runnable should not be null", npe.getMessage());
@@ -129,7 +130,8 @@ public void testCancellableCollector() throws IOException {
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            null
+            null,
+            false
         );
 
         searcher.search(new MatchAllDocsQuery(), collector1);
@@ -157,7 +159,8 @@ public void testExitableDirectoryReader() throws IOException {
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            null
+            null,
+            false
         );
         searcher.addQueryCancellation(cancellation);
         CompiledAutomaton automaton = new CompiledAutomaton(new RegExp("a.*").toAutomaton());

server/src/test/java/org/opensearch/search/internal/ContextIndexSearcherTests.java

@@ -259,7 +259,8 @@ public void onRemoval(ShardId shardId, Accountable accountable) {
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            null
+            null,
+            false
         );
 
         for (LeafReaderContext context : searcher.getIndexReader().leaves()) {

server/src/test/java/org/opensearch/search/profile/query/QueryProfilerTests.java

@@ -102,7 +102,8 @@ public void setUp() throws Exception {
             IndexSearcher.getDefaultQueryCache(),
             ALWAYS_CACHE_POLICY,
             true,
-            null
+            null,
+            false
         );
     }
 

server/src/test/java/org/opensearch/search/query/QueryPhaseTests.java

@@ -1170,7 +1170,8 @@ private static ContextIndexSearcher newContextSearcher(IndexReader reader) throw
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            null
+            null,
+            false
         );
     }
 
@@ -1181,7 +1182,8 @@ private static ContextIndexSearcher newEarlyTerminationContextSearcher(IndexRead
             IndexSearcher.getDefaultQueryCache(),
             IndexSearcher.getDefaultQueryCachingPolicy(),
             true,
-            null
+            null,
+            false
         ) {
 
             @Override