Remove DFS_QUERY_AND_FETCH as a search type (#22787)

This commit removes the search type `dfs_query_and_fetch` without a
replacement. We don't allow to use this type via REST since 2.x
but still keep it around for no particular reason. There we no users
complaining about the availability. This should now be removed from the
codebase. `query_and_fetch` is still used internally to safe a roundtrip
if there is only one shard but it can't be used via the rest interface.

core/src/main/java/org/elasticsearch/action/search/SearchDfsQueryAndFetchAsyncAction.java

@@ -1,149 +0,0 @@
-/*
- * Licensed to Elasticsearch under one or more contributor
- * license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright
- * ownership. Elasticsearch 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.elasticsearch.action.search;
-
-import org.apache.logging.log4j.Logger;
-import org.apache.logging.log4j.message.ParameterizedMessage;
-import org.apache.logging.log4j.util.Supplier;
-import org.elasticsearch.action.ActionListener;
-import org.elasticsearch.action.ActionRunnable;
-import org.elasticsearch.cluster.node.DiscoveryNode;
-import org.elasticsearch.cluster.routing.GroupShardsIterator;
-import org.elasticsearch.common.util.concurrent.AtomicArray;
-import org.elasticsearch.search.dfs.AggregatedDfs;
-import org.elasticsearch.search.dfs.DfsSearchResult;
-import org.elasticsearch.search.fetch.QueryFetchSearchResult;
-import org.elasticsearch.search.internal.AliasFilter;
-import org.elasticsearch.search.internal.InternalSearchResponse;
-import org.elasticsearch.search.internal.ShardSearchTransportRequest;
-import org.elasticsearch.search.query.QuerySearchRequest;
-import org.elasticsearch.transport.Transport;
-
-import java.io.IOException;
-import java.util.Map;
-import java.util.concurrent.Executor;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Function;
-
-class SearchDfsQueryAndFetchAsyncAction extends AbstractSearchAsyncAction<DfsSearchResult> {
-
-    private final AtomicArray<QueryFetchSearchResult> queryFetchResults;
-    private final SearchPhaseController searchPhaseController;
-    SearchDfsQueryAndFetchAsyncAction(Logger logger, SearchTransportService searchTransportService,
-                                      Function<String, Transport.Connection> nodeIdToConnection,
-                                      Map<String, AliasFilter> aliasFilter, Map<String, Float> concreteIndexBoosts,
-                                      SearchPhaseController searchPhaseController, Executor executor, SearchRequest request,
-                                      ActionListener<SearchResponse> listener,  GroupShardsIterator shardsIts,
-                                      long startTime, long clusterStateVersion, SearchTask task) {
-        super(logger, searchTransportService, nodeIdToConnection, aliasFilter, concreteIndexBoosts, executor,
-                request, listener, shardsIts, startTime, clusterStateVersion, task);
-        this.searchPhaseController = searchPhaseController;
-        queryFetchResults = new AtomicArray<>(firstResults.length());
-    }
-
-    @Override
-    protected String firstPhaseName() {
-        return "dfs";
-    }
-
-    @Override
-    protected void sendExecuteFirstPhase(Transport.Connection connection, ShardSearchTransportRequest request,
-                                         ActionListener<DfsSearchResult> listener) {
-        searchTransportService.sendExecuteDfs(connection, request, task, listener);
-    }
-
-    @Override
-    protected void moveToSecondPhase() {
-        final AggregatedDfs dfs = searchPhaseController.aggregateDfs(firstResults);
-        final AtomicInteger counter = new AtomicInteger(firstResults.asList().size());
-
-        for (final AtomicArray.Entry<DfsSearchResult> entry : firstResults.asList()) {
-            DfsSearchResult dfsResult = entry.value;
-            Transport.Connection connection = nodeIdToConnection.apply(dfsResult.shardTarget().getNodeId());
-            QuerySearchRequest querySearchRequest = new QuerySearchRequest(request, dfsResult.id(), dfs);
-            executeSecondPhase(entry.index, dfsResult, counter, connection, querySearchRequest);
-        }
-    }
-
-    void executeSecondPhase(final int shardIndex, final DfsSearchResult dfsResult, final AtomicInteger counter,
-                            final Transport.Connection connection, final QuerySearchRequest querySearchRequest) {
-        searchTransportService.sendExecuteFetch(connection, querySearchRequest, task, new ActionListener<QueryFetchSearchResult>() {
-            @Override
-            public void onResponse(QueryFetchSearchResult result) {
-                result.shardTarget(dfsResult.shardTarget());
-                queryFetchResults.set(shardIndex, result);
-                if (counter.decrementAndGet() == 0) {
-                    finishHim();
-                }
-            }
-
-            @Override
-            public void onFailure(Exception t) {
-                try {
-                    onSecondPhaseFailure(t, querySearchRequest, shardIndex, dfsResult, counter);
-                } finally {
-                    // the query might not have been executed at all (for example because thread pool rejected execution)
-                    // and the search context that was created in dfs phase might not be released.
-                    // release it again to be in the safe side
-                    sendReleaseSearchContext(querySearchRequest.id(), connection);
-                }
-            }
-        });
-    }
-
-    void onSecondPhaseFailure(Exception e, QuerySearchRequest querySearchRequest, int shardIndex, DfsSearchResult dfsResult,
-                              AtomicInteger counter) {
-        if (logger.isDebugEnabled()) {
-            logger.debug((Supplier<?>) () -> new ParameterizedMessage("[{}] Failed to execute query phase", querySearchRequest.id()), e);
-        }
-        this.addShardFailure(shardIndex, dfsResult.shardTarget(), e);
-        successfulOps.decrementAndGet();
-        if (counter.decrementAndGet() == 0) {
-            finishHim();
-        }
-    }
-
-    private void finishHim() {
-        getExecutor().execute(new ActionRunnable<SearchResponse>(listener) {
-            @Override
-            public void doRun() throws IOException {
-                sortedShardDocs = searchPhaseController.sortDocs(true, queryFetchResults);
-                final InternalSearchResponse internalResponse = searchPhaseController.merge(true, sortedShardDocs, queryFetchResults,
-                    queryFetchResults);
-                String scrollId = null;
-                if (request.scroll() != null) {
-                    scrollId = TransportSearchHelper.buildScrollId(request.searchType(), firstResults);
-                }
-                listener.onResponse(new SearchResponse(internalResponse, scrollId, expectedSuccessfulOps, successfulOps.get(),
-                    buildTookInMillis(), buildShardFailures()));
-            }
-
-            @Override
-            public void onFailure(Exception e) {
-                ReduceSearchPhaseException failure = new ReduceSearchPhaseException("query_fetch", "", e, buildShardFailures());
-                if (logger.isDebugEnabled()) {
-                    logger.debug("failed to reduce search", failure);
-                }
-                super.onFailure(e);
-            }
-        });
-
-    }
-}

core/src/main/java/org/elasticsearch/action/search/SearchType.java

@@ -37,11 +37,7 @@ public enum SearchType {
      * are fetched. This is very handy when the index has a lot of shards (not replicas, shard id groups).
      */
     QUERY_THEN_FETCH((byte) 1),
-    /**
-     * Same as {@link #QUERY_AND_FETCH}, except for an initial scatter phase which goes and computes the distributed
-     * term frequencies for more accurate scoring.
-     */
-    DFS_QUERY_AND_FETCH((byte) 2),
+    // 2 used to be DFS_QUERY_AND_FETCH
     /**
      * The most naive (and possibly fastest) implementation is to simply execute the query on all relevant shards
      * and return the results. Each shard returns size results. Since each shard already returns size hits, this
@@ -75,8 +71,6 @@ public enum SearchType {
             return DFS_QUERY_THEN_FETCH;
         } else if (id == 1) {
             return QUERY_THEN_FETCH;
-        } else if (id == 2) {
-            return DFS_QUERY_AND_FETCH;
         } else if (id == 3) {
             return QUERY_AND_FETCH;
         } else {
@@ -95,8 +89,6 @@ public enum SearchType {
         }
         if ("dfs_query_then_fetch".equals(searchType)) {
             return SearchType.DFS_QUERY_THEN_FETCH;
-        } else if ("dfs_query_and_fetch".equals(searchType)) {
-            return SearchType.DFS_QUERY_AND_FETCH;
         } else if ("query_then_fetch".equals(searchType)) {
             return SearchType.QUERY_THEN_FETCH;
         } else if ("query_and_fetch".equals(searchType)) {

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

@@ -191,7 +191,6 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
             // disable request cache if we have only suggest
             searchRequest.requestCache(false);
             switch (searchRequest.searchType()) {
-                case DFS_QUERY_AND_FETCH:
                 case DFS_QUERY_THEN_FETCH:
                     // convert to Q_T_F if we have only suggest
                     searchRequest.searchType(QUERY_THEN_FETCH);
@@ -270,11 +269,6 @@ public class TransportSearchAction extends HandledTransportAction<SearchRequest,
                     aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
                     clusterStateVersion, task);
                 break;
-            case DFS_QUERY_AND_FETCH:
-                searchAsyncAction = new SearchDfsQueryAndFetchAsyncAction(logger, searchTransportService, connectionLookup,
-                    aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,
-                    clusterStateVersion, task);
-                break;
             case QUERY_AND_FETCH:
                 searchAsyncAction = new SearchQueryAndFetchAsyncAction(logger, searchTransportService, connectionLookup,
                     aliasFilter, concreteIndexBoosts, searchPhaseController, executor, searchRequest, listener, shardIterators, startTime,

core/src/main/java/org/elasticsearch/action/search/TransportSearchHelper.java

@@ -37,7 +37,7 @@ final class TransportSearchHelper {
     static String buildScrollId(SearchType searchType, AtomicArray<? extends SearchPhaseResult> searchPhaseResults) throws IOException {
         if (searchType == SearchType.DFS_QUERY_THEN_FETCH || searchType == SearchType.QUERY_THEN_FETCH) {
             return buildScrollId(ParsedScrollId.QUERY_THEN_FETCH_TYPE, searchPhaseResults);
-        } else if (searchType == SearchType.QUERY_AND_FETCH || searchType == SearchType.DFS_QUERY_AND_FETCH) {
+        } else if (searchType == SearchType.QUERY_AND_FETCH) {
             return buildScrollId(ParsedScrollId.QUERY_AND_FETCH_TYPE, searchPhaseResults);
         } else {
             throw new IllegalStateException("search_type [" + searchType + "] not supported");

core/src/main/java/org/elasticsearch/rest/action/search/RestSearchAction.java

@@ -93,7 +93,7 @@ public class RestSearchAction extends BaseRestHandler {
         // not be specified explicitly by the user.
         String searchType = request.param("search_type");
         if (SearchType.fromString(searchType).equals(SearchType.QUERY_AND_FETCH) ||
-                SearchType.fromString(searchType).equals(SearchType.DFS_QUERY_AND_FETCH)) {
+                "dfs_query_and_fetch".equals(searchType)) {
             throw new IllegalArgumentException("Unsupported search type [" + searchType + "]");
         } else {
             searchRequest.searchType(searchType);

core/src/main/java/org/elasticsearch/search/query/QueryPhase.java

@@ -262,8 +262,7 @@ public class QueryPhase implements SearchPhase {
                             }
                             switch (searchType) {
                             case QUERY_AND_FETCH:
-                            case DFS_QUERY_AND_FETCH:
-                                // for (DFS_)QUERY_AND_FETCH, we already know the last emitted doc
+                                // for QUERY_AND_FETCH, we already know the last emitted doc
                                 if (topDocs.scoreDocs.length > 0) {
                                     // set the last emitted doc
                                     scrollContext.lastEmittedDoc = topDocs.scoreDocs[topDocs.scoreDocs.length - 1];

core/src/main/java/org/elasticsearch/search/rescore/Rescorer.java

@@ -66,7 +66,6 @@ public interface Rescorer {
     /**
      * Extracts all terms needed to execute this {@link Rescorer}. This method
      * is executed in a distributed frequency collection roundtrip for
-     * {@link SearchType#DFS_QUERY_AND_FETCH} and
      * {@link SearchType#DFS_QUERY_THEN_FETCH}
      */
     void extractTerms(SearchContext context, RescoreSearchContext rescoreContext, Set<Term> termsSet);

core/src/test/java/org/elasticsearch/action/IndicesRequestIT.java

@@ -598,27 +598,6 @@ public class IndicesRequestIT extends ESIntegTestCase {
         assertSameIndicesOptionalRequests(searchRequest, SearchTransportService.FREE_CONTEXT_ACTION_NAME);
     }
 
-    public void testSearchDfsQueryAndFetch() throws Exception {
-        interceptTransportActions(SearchTransportService.QUERY_QUERY_FETCH_ACTION_NAME,
-                SearchTransportService.FREE_CONTEXT_ACTION_NAME);
-
-        String[] randomIndicesOrAliases = randomIndicesOrAliases();
-        for (int i = 0; i < randomIndicesOrAliases.length; i++) {
-            client().prepareIndex(randomIndicesOrAliases[i], "type", "id-" + i).setSource("field", "value").get();
-        }
-        refresh();
-
-        SearchRequest searchRequest = new SearchRequest(randomIndicesOrAliases).searchType(SearchType.DFS_QUERY_AND_FETCH);
-        SearchResponse searchResponse = internalCluster().coordOnlyNodeClient().search(searchRequest).actionGet();
-        assertNoFailures(searchResponse);
-        assertThat(searchResponse.getHits().totalHits(), greaterThan(0L));
-
-        clearInterceptedActions();
-        assertSameIndices(searchRequest, SearchTransportService.QUERY_QUERY_FETCH_ACTION_NAME);
-        //free context messages are not necessarily sent, but if they are, check their indices
-        assertSameIndicesOptionalRequests(searchRequest, SearchTransportService.FREE_CONTEXT_ACTION_NAME);
-    }
-
     private static void assertSameIndices(IndicesRequest originalRequest, String... actions) {
         assertSameIndices(originalRequest, false, actions);
     }

core/src/test/java/org/elasticsearch/search/SearchWithRejectionsIT.java

@@ -54,7 +54,7 @@ public class SearchWithRejectionsIT extends ESIntegTestCase {
 
         int numSearches = 10;
         Future<SearchResponse>[] responses = new Future[numSearches];
-        SearchType searchType = randomFrom(SearchType.DEFAULT, SearchType.QUERY_AND_FETCH, SearchType.QUERY_THEN_FETCH, SearchType.DFS_QUERY_AND_FETCH, SearchType.DFS_QUERY_THEN_FETCH);
+        SearchType searchType = randomFrom(SearchType.DEFAULT, SearchType.QUERY_AND_FETCH, SearchType.QUERY_THEN_FETCH, SearchType.DFS_QUERY_THEN_FETCH);
         logger.info("search type is {}", searchType);
         for (int i = 0; i < numSearches; i++) {
             responses[i] = client().prepareSearch()

core/src/test/java/org/elasticsearch/search/basic/TransportTwoNodesSearchIT.java

@@ -48,7 +48,6 @@ import java.util.HashSet;
 import java.util.Set;
 import java.util.TreeSet;
 
-import static org.elasticsearch.action.search.SearchType.DFS_QUERY_AND_FETCH;
 import static org.elasticsearch.action.search.SearchType.DFS_QUERY_THEN_FETCH;
 import static org.elasticsearch.action.search.SearchType.QUERY_AND_FETCH;
 import static org.elasticsearch.action.search.SearchType.QUERY_THEN_FETCH;
@@ -319,58 +318,6 @@ public class TransportTwoNodesSearchIT extends ESIntegTestCase {
         assertThat("make sure we got all [" + expectedIds + "]", expectedIds.size(), equalTo(0));
     }
 
-    public void testDfsQueryAndFetch() throws Exception {
-        prepareData(3);
-
-        SearchSourceBuilder source = searchSource()
-                .query(termQuery("multi", "test"))
-                .from(0).size(20).explain(true);
-
-        Set<String> expectedIds = new HashSet<>();
-        for (int i = 0; i < 100; i++) {
-            expectedIds.add(Integer.toString(i));
-        }
-
-
-        //SearchResponse searchResponse = client().search(searchRequest("test").source(source).searchType(DFS_QUERY_AND_FETCH).scroll(new Scroll(timeValueMinutes(10)))).actionGet();
-        SearchResponse searchResponse = client().prepareSearch("test").setSearchType(DFS_QUERY_AND_FETCH).setScroll("10m").setSource(source).get();
-        assertNoFailures(searchResponse);
-        assertThat(searchResponse.getHits().totalHits(), equalTo(100L));
-        assertThat(searchResponse.getHits().hits().length, equalTo(60)); // 20 per shard
-        for (int i = 0; i < 60; i++) {
-            SearchHit hit = searchResponse.getHits().hits()[i];
-//            System.out.println(hit.shard() + ": " +  hit.explanation());
-            assertThat(hit.explanation(), notNullValue());
-            assertThat(hit.explanation().getDetails().length, equalTo(1));
-            assertThat(hit.explanation().getDetails()[0].getDetails().length, equalTo(2));
-            assertThat(hit.explanation().getDetails()[0].getDetails()[0].getDetails().length, equalTo(2));
-            assertThat(hit.explanation().getDetails()[0].getDetails()[0].getDetails()[0].getDescription(),
-                equalTo("docFreq"));
-            assertThat(hit.explanation().getDetails()[0].getDetails()[0].getDetails()[0].getValue(),
-                equalTo(100.0f));
-            assertThat(hit.explanation().getDetails()[0].getDetails()[0].getDetails()[1].getDescription(),
-                equalTo("docCount"));
-            assertThat(hit.explanation().getDetails()[0].getDetails()[0].getDetails()[1].getValue(),
-                equalTo(100.0f));
-//            assertThat("id[" + hit.id() + "]", hit.id(), equalTo(Integer.toString(100 - i - 1)));
-            assertThat("make sure we don't have duplicates", expectedIds.remove(hit.id()), notNullValue());
-        }
-
-        do {
-            searchResponse = client().prepareSearchScroll(searchResponse.getScrollId()).setScroll("10m").get();
-
-            assertThat(searchResponse.getHits().totalHits(), equalTo(100L));
-            assertThat(searchResponse.getHits().hits().length, lessThanOrEqualTo(40));
-            for (int i = 0; i < searchResponse.getHits().hits().length; i++) {
-                SearchHit hit = searchResponse.getHits().hits()[i];
-                // we don't do perfect sorting when it comes to scroll with Query+Fetch
-                assertThat("make sure we don't have duplicates", expectedIds.remove(hit.id()), notNullValue());
-            }
-        } while (searchResponse.getHits().hits().length > 0);
-        clearScroll(searchResponse.getScrollId());
-        assertThat("make sure we got all [" + expectedIds + "]", expectedIds.size(), equalTo(0));
-    }
-
     public void testSimpleFacets() throws Exception {
         prepareData();
 
@@ -421,7 +368,7 @@ public class TransportTwoNodesSearchIT extends ESIntegTestCase {
         SearchSourceBuilder source = searchSource()
                 .query(termQuery("multi", "test"))
                 .from(1000).size(20).explain(true);
-        SearchResponse response = client().search(searchRequest("test").searchType(DFS_QUERY_AND_FETCH).source(source)).actionGet();
+        SearchResponse response = client().search(searchRequest("test").searchType(DFS_QUERY_THEN_FETCH).source(source)).actionGet();
         assertThat(response.getHits().hits().length, equalTo(0));
         assertThat(response.getTotalShards(), equalTo(test.numPrimaries));
         assertThat(response.getSuccessfulShards(), equalTo(test.numPrimaries));
@@ -431,7 +378,7 @@ public class TransportTwoNodesSearchIT extends ESIntegTestCase {
         assertNoFailures(response);
         assertThat(response.getHits().hits().length, equalTo(0));
 
-        response = client().search(searchRequest("test").searchType(DFS_QUERY_AND_FETCH).source(source)).actionGet();
+        response = client().search(searchRequest("test").searchType(DFS_QUERY_THEN_FETCH).source(source)).actionGet();
         assertNoFailures(response);
         assertThat(response.getHits().hits().length, equalTo(0));
 

core/src/test/java/org/elasticsearch/search/scroll/DuelScrollIT.java

@@ -97,38 +97,6 @@ public class DuelScrollIT extends ESIntegTestCase {
         clearScroll(scrollId);
     }
 
-    public void testDuelQueryAndFetch() throws Exception {
-        // *_QUERY_AND_FETCH search types are tricky: the ordering can be incorrect, since it returns num_shards * (from + size)
-        // a subsequent scroll call can return hits that should have been in the hits of the first scroll call.
-
-        TestContext context = create(SearchType.DFS_QUERY_AND_FETCH, SearchType.QUERY_AND_FETCH);
-        SearchResponse searchScrollResponse = client().prepareSearch("index")
-                .setSearchType(context.searchType)
-                .addSort(context.sort)
-                .setSize(context.scrollRequestSize)
-                .setScroll("10m").get();
-
-        assertNoFailures(searchScrollResponse);
-        assertThat(searchScrollResponse.getHits().getTotalHits(), equalTo((long) context.numDocs));
-
-        int counter = searchScrollResponse.getHits().hits().length;
-        String scrollId = searchScrollResponse.getScrollId();
-        while (true) {
-            searchScrollResponse = client().prepareSearchScroll(scrollId).setScroll("10m").get();
-            assertNoFailures(searchScrollResponse);
-            assertThat(searchScrollResponse.getHits().getTotalHits(), equalTo((long) context.numDocs));
-            if (searchScrollResponse.getHits().hits().length == 0) {
-                break;
-            }
-
-            counter += searchScrollResponse.getHits().hits().length;
-            scrollId = searchScrollResponse.getScrollId();
-        }
-
-        assertThat(counter, equalTo(context.numDocs));
-        clearScroll(scrollId);
-    }
-
 
     private TestContext create(SearchType... searchTypes) throws Exception {
         assertAcked(prepareCreate("index").addMapping("type", jsonBuilder().startObject().startObject("type").startObject("properties")
@@ -289,7 +257,7 @@ public class DuelScrollIT extends ESIntegTestCase {
     }
 
     public void testDuelIndexOrderQueryAndFetch() throws Exception {
-        final SearchType searchType = RandomPicks.randomFrom(random(), Arrays.asList(SearchType.QUERY_AND_FETCH, SearchType.DFS_QUERY_AND_FETCH));
+        final SearchType searchType = SearchType.QUERY_AND_FETCH;
         // QUERY_AND_FETCH only works with a single shard
         final int numDocs = createIndex(true);
         testDuelIndexOrder(searchType, false, numDocs);