WIP: Emulate "fields" on older versions

We introduced the new "fields" option in search with version 7.10. With this
change we are trying to do a best-effort attempt at emulating this new behaviour
in mixed cluster or CCS scenarios where search requests using the "fields"
option also target older nodes or clusters. In that case, currently we don't
return anything in the "fields" section of the response. This change tried to
emulate the fields behaviour by modifying the request to include the respective
"_source" fields and then parsing them back into the "fields" section on return.
This will not be fully equivalent to the post-7.10 "fields" functionality but at
least try to include whatever we find in "_source" in earlier versions.

Currently Draft only, needs more testing in CCS scenarios but I'm opening this
here already to get some test coverage on the modifications so far.

Author: Christoph Büscher

qa/mixed-cluster/src/test/java/org/elasticsearch/backwards/FieldsOptionEmulationIT.java

@@ -0,0 +1,103 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+package org.elasticsearch.backwards;
+
+import org.apache.http.HttpHost;
+import org.elasticsearch.Version;
+import org.elasticsearch.backwards.IndexingIT.Node;
+import org.elasticsearch.backwards.IndexingIT.Nodes;
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.test.rest.yaml.ObjectPath;
+import org.junit.Before;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * In mixed cluster scenarios on 7.x we try to emulate the "fields" option introduced in 7.10
+ * by running a request with "source" enabled for the requested patterns on older nodes and convert
+ * the resulting source entries back into the "fields" section. These tests check this in mixed cluster
+ * scenarios.
+ */
+public class FieldsOptionEmulationIT extends ESRestTestCase {
+
+    private static String index = "test_field_newversion";
+    private static String index_old = "test_field_oldversion";
+    private static Nodes nodes;
+    private static List<Node> bwcNodes;
+    private static List<Node> newNodes;
+    private static String oldNodeName;
+    private static String newNodeName;
+
+    @Before
+    public void prepareTestData() throws IOException {
+        nodes = IndexingIT.buildNodeAndVersions(client());
+        bwcNodes = new ArrayList<>(nodes.getBWCNodes());
+        newNodes = new ArrayList<>(nodes.getNewNodes());
+        oldNodeName = bwcNodes.get(0).getNodeName();
+        newNodeName = newNodes.get(0).getNodeName();
+        createIndexOnNode(index, newNodeName);
+        createIndexOnNode(index_old, oldNodeName);
+        refreshAllIndices();
+    }
+
+    private void createIndexOnNode(String indexName, String nodeName) throws IOException {
+        if (indexExists(indexName) == false) {
+            createIndex(indexName, Settings.builder()
+                .put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 1)
+                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                .put(IndexMetadata.INDEX_ROUTING_REQUIRE_GROUP_PREFIX + "._name", nodeName).build());
+            for (int i = 0; i < 5; i++) {
+                Request request = new Request("PUT", indexName + "/_doc/" + i);
+                request.setJsonEntity(
+                    "{\"test\": \"test_" + randomAlphaOfLength(2) + "\"," + "\"obj\" : { \"foo\" : \"value_" + i + "\"} }"
+                );
+                assertOK(client().performRequest(request));
+            }
+            ensureGreen(indexName);
+            flush(indexName, true);
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public void testFieldOptionAdapter() throws Exception {
+        Request matchAllRequest = new Request("POST",
+            "test_field_*/_search");
+        matchAllRequest.setJsonEntity("{\"_source\":false,\"fields\":[\"*\"]}");
+        try (
+            RestClient client = buildClient(restClientSettings(), newNodes.stream().map(Node::getPublishAddress).toArray(HttpHost[]::new))
+        ) {
+            Response response = client.performRequest(matchAllRequest);
+            ObjectPath responseObject = ObjectPath.createFromResponse(response);
+            System.out.println(Strings.toString(responseObject.toXContentBuilder(XContentType.JSON.xContent())));
+            List<Map<String, Object>> hits = responseObject.evaluate("hits.hits");
+            assertEquals(10, hits.size());
+            for (Map<String, Object> hit : hits) {
+                Map<String, Object> fieldsMap = (Map<String, Object>) hit.get("fields");
+                assertNotNull(fieldsMap);
+                assertNotNull(fieldsMap.get("test"));
+                assertTrue(((List<?>) fieldsMap.get("test")).get(0).toString().startsWith("test_"));
+                assertNotNull(fieldsMap.get("obj.foo"));
+                assertTrue(((List<?>) fieldsMap.get("obj.foo")).get(0).toString().startsWith("value_"));
+                if (bwcNodes.get(0).getVersion().onOrAfter(Version.V_7_10_0)) {
+                    // if all nodes are > 7.10 we should get full "fields" output even for subfields
+                    assertTrue(((List<?>) fieldsMap.get("test.keyword")).get(0).toString().startsWith("test_"));
+                }
+            }
+        }
+    }
+}

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

@@ -8,10 +8,12 @@
 package org.elasticsearch.action.search;
 
 import com.carrotsearch.hppc.IntArrayList;
+
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.message.ParameterizedMessage;
 import org.apache.lucene.search.ScoreDoc;
 import org.elasticsearch.action.OriginalIndices;
+import org.elasticsearch.action.search.TransportSearchAction.FieldsOptionSourceAdapter;
 import org.elasticsearch.common.util.concurrent.AbstractRunnable;
 import org.elasticsearch.common.util.concurrent.AtomicArray;
 import org.elasticsearch.search.RescoreDocIds;
@@ -162,11 +164,20 @@ private void executeFetch(final int shardIndex, final SearchShardTarget shardTar
                               final CountedCollector<FetchSearchResult> counter,
                               final ShardFetchSearchRequest fetchSearchRequest, final QuerySearchResult querySearchResult,
                               final Transport.Connection connection) {
+        final FieldsOptionSourceAdapter adapter;
+        if (querySearchResult instanceof WrappedQuerySearchResult) {
+            adapter = ((WrappedQuerySearchResult) querySearchResult).getAdapter();
+        } else {
+            adapter = null;
+        }
         context.getSearchTransport().sendExecuteFetch(connection, fetchSearchRequest, context.getTask(),
             new SearchActionListener<FetchSearchResult>(shardTarget, shardIndex) {
                 @Override
                 public void innerOnResponse(FetchSearchResult result) {
                     try {
+                        if (adapter != null) {
+                            adapter.adaptResponse(result.hits().getHits());
+                        }
                         progressListener.notifyFetchResult(shardIndex);
                         counter.onResult(result);
                     } catch (Exception e) {

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

@@ -10,7 +10,9 @@
 
 import org.apache.logging.log4j.Logger;
 import org.apache.lucene.search.TopFieldDocs;
+import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.search.TransportSearchAction.FieldsOptionSourceAdapter;
 import org.elasticsearch.cluster.ClusterState;
 import org.elasticsearch.cluster.routing.GroupShardsIterator;
 import org.elasticsearch.search.SearchPhaseResult;
@@ -20,6 +22,7 @@
 import org.elasticsearch.search.internal.ShardSearchRequest;
 import org.elasticsearch.search.query.QuerySearchResult;
 import org.elasticsearch.transport.Transport;
+import org.elasticsearch.transport.Transport.Connection;
 
 import java.util.Map;
 import java.util.concurrent.Executor;
@@ -64,11 +67,40 @@ class SearchQueryThenFetchAsyncAction extends AbstractSearchAsyncAction<SearchPh
             SearchProgressListener.buildSearchShards(toSkipShardsIts), clusters, hasFetchPhase);
     }
 
+    @Override
     protected void executePhaseOnShard(final SearchShardIterator shardIt,
                                        final SearchShardTarget shard,
                                        final SearchActionListener<SearchPhaseResult> listener) {
         ShardSearchRequest request = rewriteShardSearchRequest(super.buildShardSearchRequest(shardIt, listener.requestIndex));
-        getSearchTransport().sendExecuteQuery(getConnection(shard.getClusterAlias(), shard.getNodeId()), request, getTask(), listener);
+        final FieldsOptionSourceAdapter fieldsOptionAdapter;
+        Connection connection = getConnection(shard.getClusterAlias(), shard.getNodeId());
+        if (connection.getVersion().before(Version.V_7_10_0)) {
+            fieldsOptionAdapter = TransportSearchAction.createFieldsOptionAdapter(connection, request.source());
+            fieldsOptionAdapter.adaptRequest(request.source(), request::source);
+        } else {
+            fieldsOptionAdapter = null;
+        }
+
+        getSearchTransport().sendExecuteQuery(
+            connection,
+            request,
+            getTask(),
+            new SearchActionListener<SearchPhaseResult>(shard, listener.requestIndex) {
+
+                @Override
+                public void onFailure(Exception e) {
+                    listener.onFailure(e);
+                }
+
+                @Override
+                protected void innerOnResponse(SearchPhaseResult response) {
+                    if (response instanceof QuerySearchResult) {
+                        response = new WrappedQuerySearchResult((QuerySearchResult) response, fieldsOptionAdapter);
+                    }
+                    listener.onResponse(response);
+                }
+            }
+        );
     }
 
     @Override