Stop passing search query local node fanout through transport layer

We don't need to go through all the steps in the transport layer for executing query work on the local node.
Doing so adds considerable overhead. E.g. with security enabled, the amount of work on transport threads goes
up by about 4x for local only searches in the `geonames` track.
The simpler a query is to rewrite and can-match and the more shards are queried on the local node, the more
pronounced the overhead is in practice.

Author: original-brownbear

server/src/main/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncAction.java

@@ -16,6 +16,7 @@
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.search.SearchPhaseResult;
+import org.elasticsearch.search.SearchService;
 import org.elasticsearch.search.SearchShardTarget;
 import org.elasticsearch.search.dfs.AggregatedDfs;
 import org.elasticsearch.search.internal.AliasFilter;
@@ -40,6 +41,7 @@ class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<SearchPh
     private final int trackTotalHitsUpTo;
     private volatile BottomSortValuesCollector bottomSortCollector;
     private final Client client;
+    private final SearchService searchService;
 
     SearchQueryThenFetchAsyncAction(
         Logger logger,
@@ -57,7 +59,8 @@ class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<SearchPh
         ClusterState clusterState,
         SearchTask task,
         SearchResponse.Clusters clusters,
-        Client client
+        Client client,
+        SearchService searchService
     ) {
         super(
             "query",
@@ -82,7 +85,7 @@ class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<SearchPh
         this.trackTotalHitsUpTo = request.resolveTrackTotalHitsUpTo();
         this.progressListener = task.getProgressListener();
         this.client = client;
-
+        this.searchService = searchService;
         // don't build the SearchShard list (can be expensive) if the SearchProgressListener won't use it
         if (progressListener != SearchProgressListener.NOOP) {
             notifyListShards(progressListener, clusters, request.source());
@@ -95,7 +98,13 @@ protected void executePhaseOnShard(
         final SearchActionListener<SearchPhaseResult> listener
     ) {
         ShardSearchRequest request = rewriteShardSearchRequest(super.buildShardSearchRequest(shardIt, listener.requestIndex));
-        getSearchTransport().sendExecuteQuery(connection, request, getTask(), listener);
+        var searchTransport = getSearchTransport();
+        var task = getTask();
+        if (searchTransport.transportService().getLocalNodeConnection() == connection) {
+            searchService.executeQueryPhase(request, task, listener);
+            return;
+        }
+        searchTransport.sendExecuteQuery(connection, request, task, listener);
     }
 
     @Override

server/src/main/java/org/elasticsearch/action/search/SearchTransportService.java

@@ -122,6 +122,10 @@ public SearchTransportService(
         this.responseWrapper = responseWrapper;
     }
 
+    public TransportService transportService() {
+        return transportService;
+    }
+
     public void sendFreeContext(
         Transport.Connection connection,
         ShardSearchContextId contextId,

server/src/main/java/org/elasticsearch/action/search/TransportSearchAction.java

@@ -1565,7 +1565,8 @@ public void runNewSearchPhase(
                         clusterState,
                         task,
                         clusters,
-                        client
+                        client,
+                        searchService
                     );
                 }
                 success = true;

server/src/test/java/org/elasticsearch/action/search/SearchQueryThenFetchAsyncActionTests.java

@@ -39,6 +39,7 @@
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.test.InternalAggregationTestCase;
 import org.elasticsearch.transport.Transport;
+import org.elasticsearch.transport.TransportService;
 
 import java.util.Collections;
 import java.util.List;
@@ -51,6 +52,8 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.instanceOf;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 public class SearchQueryThenFetchAsyncActionTests extends ESTestCase {
     public void testBottomFieldSort() throws Exception {
@@ -83,7 +86,11 @@ private void testCase(boolean withScroll, boolean withCollapse) throws Exception
         AtomicInteger numWithTopDocs = new AtomicInteger();
         AtomicInteger successfulOps = new AtomicInteger();
         AtomicBoolean canReturnNullResponse = new AtomicBoolean(false);
-        SearchTransportService searchTransportService = new SearchTransportService(null, null, null) {
+        TransportService transportService = mock(TransportService.class);
+        when(transportService.getLocalNodeConnection()).thenReturn(
+            new SearchAsyncActionTests.MockConnection(DiscoveryNodeUtils.create("local"))
+        );
+        SearchTransportService searchTransportService = new SearchTransportService(transportService, null, null) {
             @Override
             public void sendExecuteQuery(
                 Transport.Connection connection,
@@ -201,6 +208,7 @@ public void sendExecuteQuery(
                 new ClusterState.Builder(new ClusterName("test")).build(),
                 task,
                 SearchResponse.Clusters.EMPTY,
+                null,
                 null
             ) {
                 @Override