import org.elasticsearch.search.Scroll; //导入依赖的package包/类
protected void innerReadFrom(StreamInput in) throws IOException {
    shardId = ShardId.readShardId(in);
    searchType = SearchType.fromId(in.readByte());
    numberOfShards = in.readVInt();
    scroll = in.readOptionalWriteable(Scroll::new);
    source = in.readOptionalWriteable(SearchSourceBuilder::new);
    types = in.readStringArray();
    aliasFilter = new AliasFilter(in);
    if (in.getVersion().onOrAfter(Version.V_5_2_0_UNRELEASED)) {
        indexBoost = in.readFloat();
    } else {
        // Nodes < 5.2.0 doesn't send index boost. Read it from source.
        if (source != null) {
            Optional<SearchSourceBuilder.IndexBoost> boost = source.indexBoosts()
                .stream()
                .filter(ib -> ib.getIndex().equals(shardId.getIndexName()))
                .findFirst();
            indexBoost = boost.isPresent() ? boost.get().getBoost() : 1.0f;
        } else {
            indexBoost = 1.0f;
        }
    }
    nowInMillis = in.readVLong();
    requestCache = in.readOptionalBoolean();
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
@Override
public RestChannelConsumer prepareRequest(final RestRequest request, final NodeClient client) throws IOException {
    String scrollId = request.param("scroll_id");
    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    searchScrollRequest.scrollId(scrollId);
    String scroll = request.param("scroll");
    if (scroll != null) {
        searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
    }

    request.withContentOrSourceParamParserOrNull(xContentParser -> {
        if (xContentParser != null) {
            // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
            try {
                buildFromContent(xContentParser, searchScrollRequest);
            } catch (IOException e) {
                throw new IllegalArgumentException("Failed to parse request body", e);
            }
        }});
    return channel -> client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<>(channel));
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public static void buildFromContent(XContentParser parser, SearchScrollRequest searchScrollRequest) throws IOException {
    if (parser.nextToken() != XContentParser.Token.START_OBJECT) {
        throw new IllegalArgumentException("Malformed content, must start with an object");
    } else {
        XContentParser.Token token;
        String currentFieldName = null;
        while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
            if (token == XContentParser.Token.FIELD_NAME) {
                currentFieldName = parser.currentName();
            } else if ("scroll_id".equals(currentFieldName) && token == XContentParser.Token.VALUE_STRING) {
                searchScrollRequest.scrollId(parser.text());
            } else if ("scroll".equals(currentFieldName) && token == XContentParser.Token.VALUE_STRING) {
                searchScrollRequest.scroll(new Scroll(TimeValue.parseTimeValue(parser.text(), null, "scroll")));
            } else {
                throw new IllegalArgumentException("Unknown parameter [" + currentFieldName
                        + "] in request body or parameter is of the wrong type[" + token + "] ");
            }
        }
    }
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    searchType = SearchType.fromId(in.readByte());
    indices = new String[in.readVInt()];
    for (int i = 0; i < indices.length; i++) {
        indices[i] = in.readString();
    }
    routing = in.readOptionalString();
    preference = in.readOptionalString();
    scroll = in.readOptionalWriteable(Scroll::new);
    source = in.readOptionalWriteable(SearchSourceBuilder::new);
    types = in.readStringArray();
    indicesOptions = IndicesOptions.readIndicesOptions(in);
    requestCache = in.readOptionalBoolean();
    batchedReduceSize = in.readVInt();
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public void testDocIdSort() throws Exception {
    int numShards = setupIndex(true);
    SearchResponse sr = client().prepareSearch("test")
        .setQuery(matchAllQuery())
        .setSize(0)
        .get();
    int numDocs = (int) sr.getHits().getTotalHits();
    assertThat(numDocs, equalTo(NUM_DOCS));
    int max = randomIntBetween(2, numShards*3);
    for (String field : new String[]{"_uid", "random_int", "static_int"}) {
        int fetchSize = randomIntBetween(10, 100);
        SearchRequestBuilder request = client().prepareSearch("test")
            .setQuery(matchAllQuery())
            .setScroll(new Scroll(TimeValue.timeValueSeconds(10)))
            .setSize(fetchSize)
            .addSort(SortBuilders.fieldSort("_doc"));
        assertSearchSlicesWithScroll(request, field, max);
    }
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public void testNumericSort() throws Exception {
    int numShards = setupIndex(true);
    SearchResponse sr = client().prepareSearch("test")
        .setQuery(matchAllQuery())
        .setSize(0)
        .get();
    int numDocs = (int) sr.getHits().getTotalHits();
    assertThat(numDocs, equalTo(NUM_DOCS));

    int max = randomIntBetween(2, numShards*3);
    for (String field : new String[]{"_uid", "random_int", "static_int"}) {
        int fetchSize = randomIntBetween(10, 100);
        SearchRequestBuilder request = client().prepareSearch("test")
            .setQuery(matchAllQuery())
            .setScroll(new Scroll(TimeValue.timeValueSeconds(10)))
            .addSort(SortBuilders.fieldSort("random_int"))
            .setSize(fetchSize);
        assertSearchSlicesWithScroll(request, field, max);
    }
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public void testInvalidFields() throws Exception {
    setupIndex(false);
    SearchPhaseExecutionException exc = expectThrows(SearchPhaseExecutionException.class,
        () -> client().prepareSearch("test")
            .setQuery(matchAllQuery())
            .setScroll(new Scroll(TimeValue.timeValueSeconds(10)))
            .slice(new SliceBuilder("invalid_random_int", 0, 10))
            .get());
    Throwable rootCause = findRootCause(exc);
    assertThat(rootCause.getClass(), equalTo(IllegalArgumentException.class));
    assertThat(rootCause.getMessage(),
        startsWith("cannot load numeric doc values"));

    exc = expectThrows(SearchPhaseExecutionException.class, () -> client().prepareSearch("test")
        .setQuery(matchAllQuery())
        .setScroll(new Scroll(TimeValue.timeValueSeconds(10)))
        .slice(new SliceBuilder("invalid_random_kw", 0, 10))
        .get());
    rootCause = findRootCause(exc);
    assertThat(rootCause.getClass(), equalTo(IllegalArgumentException.class));
    assertThat(rootCause.getMessage(),
        startsWith("cannot load numeric doc values"));
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public CrateSearchContext(long id,
                          final long nowInMillis,
                          SearchShardTarget shardTarget,
                          Engine.Searcher engineSearcher,
                          IndexService indexService,
                          final IndexShard indexShard,
                          ScriptService scriptService,
                          PageCacheRecycler pageCacheRecycler,
                          BigArrays bigArrays,
                          Counter timeEstimateCounter,
                          Optional<Scroll> scroll) {
    super(id, new CrateSearchShardRequest(nowInMillis, scroll, indexShard),
            shardTarget, engineSearcher, indexService,
            indexShard, scriptService, pageCacheRecycler,
            bigArrays, timeEstimateCounter, ParseFieldMatcher.STRICT, SearchService.NO_TIMEOUT);
    this.engineSearcher = engineSearcher;
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) {
    String scrollId = request.param("scroll_id");
    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    searchScrollRequest.scrollId(scrollId);
    String scroll = request.param("scroll");
    if (scroll != null) {
        searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
    }

    if (RestActions.hasBodyContent(request)) {
        XContentType type = XContentFactory.xContentType(RestActions.getRestContent(request));
        if (type == null) {
            if (scrollId == null) {
                scrollId = RestActions.getRestContent(request).toUtf8();
                searchScrollRequest.scrollId(scrollId);
            }
        } else {
            // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
            buildFromContent(RestActions.getRestContent(request), searchScrollRequest);
        }
    }
    client.searchScroll(searchScrollRequest, new RestStatusToXContentListener<SearchResponse>(channel));
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public static void buildFromContent(BytesReference content, SearchScrollRequest searchScrollRequest) {
    try (XContentParser parser = XContentHelper.createParser(content)) {
        if (parser.nextToken() != XContentParser.Token.START_OBJECT) {
            throw new IllegalArgumentException("Malforrmed content, must start with an object");
        } else {
            XContentParser.Token token;
            String currentFieldName = null;
            while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
                if (token == XContentParser.Token.FIELD_NAME) {
                    currentFieldName = parser.currentName();
                } else if ("scroll_id".equals(currentFieldName) && token == XContentParser.Token.VALUE_STRING) {
                    searchScrollRequest.scrollId(parser.text());
                } else if ("scroll".equals(currentFieldName) && token == XContentParser.Token.VALUE_STRING) {
                    searchScrollRequest.scroll(new Scroll(TimeValue.parseTimeValue(parser.text(), null, "scroll")));
                } else {
                    throw new IllegalArgumentException("Unknown parameter [" + currentFieldName + "] in request body or parameter is of the wrong type[" + token + "] ");
                }
            }
        }
    } catch (IOException e) {
        throw new IllegalArgumentException("Failed to parse request body", e);
    }
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
@Override
public void byPages(int pageSize, Callback callback) {

	SearchResponse searchResponse = client.prepareSearch(index).setSize(pageSize).setFrom(0).setScroll(new Scroll(TimeValue.timeValueMinutes(10))).get();
	boolean loop = true;
	try {
		while (loop) {
			loop = callback.on(StreamSupport.stream(searchResponse.getHits().spliterator(), true).map(mapper).collect(Collectors.toList()));
			searchResponse = client.searchScroll(Requests.searchScrollRequest(searchResponse.getScrollId())).actionGet();

			if (searchResponse.getHits().hits() == null) {
				break;
			}
		}
	} finally {
		client.prepareClearScroll().addScrollId(searchResponse.getScrollId()).execute();
	}
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
/**
 * Parses the rest request on top of the SearchRequest, preserving values that are not overridden by the rest request.
 *
 * @param requestContentParser body of the request to read. This method does not attempt to read the body from the {@code request}
 *        parameter
 */
public static void parseSearchRequest(SearchRequest searchRequest, RestRequest request,
                                      XContentParser requestContentParser) throws IOException {

    if (searchRequest.source() == null) {
        searchRequest.source(new SearchSourceBuilder());
    }
    searchRequest.indices(Strings.splitStringByCommaToArray(request.param("index")));
    if (requestContentParser != null) {
        QueryParseContext context = new QueryParseContext(requestContentParser);
        searchRequest.source().parseXContent(context);
    }

    final int batchedReduceSize = request.paramAsInt("batched_reduce_size", searchRequest.getBatchedReduceSize());
    searchRequest.setBatchedReduceSize(batchedReduceSize);

    // do not allow 'query_and_fetch' or 'dfs_query_and_fetch' search types
    // from the REST layer. these modes are an internal optimization and should
    // not be specified explicitly by the user.
    String searchType = request.param("search_type");
    if ("query_and_fetch".equals(searchType) ||
            "dfs_query_and_fetch".equals(searchType)) {
        throw new IllegalArgumentException("Unsupported search type [" + searchType + "]");
    } else {
        searchRequest.searchType(searchType);
    }
    parseSearchSource(searchRequest.source(), request);
    searchRequest.requestCache(request.paramAsBoolean("request_cache", null));

    String scroll = request.param("scroll");
    if (scroll != null) {
        searchRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
    }

    searchRequest.types(Strings.splitStringByCommaToArray(request.param("type")));
    searchRequest.routing(request.param("routing"));
    searchRequest.preference(request.param("preference"));
    searchRequest.indicesOptions(IndicesOptions.fromRequest(request, searchRequest.indicesOptions()));
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public CrateSearchContext createContext(
        int jobSearchContextId,
        IndexShard indexshard,
        Engine.Searcher engineSearcher,
        WhereClause whereClause) {

    ShardId shardId = indexshard.shardId();
    SearchShardTarget searchShardTarget = new SearchShardTarget(
            clusterService.state().nodes().localNodeId(),
            shardId.getIndex(),
            shardId.id()
    );
    IndexService indexService = indexshard.indexService();
    CrateSearchContext searchContext = new CrateSearchContext(
            jobSearchContextId,
            System.currentTimeMillis(),
            searchShardTarget,
            engineSearcher,
            indexService,
            indexshard,
            scriptService,
            pageCacheRecycler,
            bigArrays,
            threadPool.estimatedTimeInMillisCounter(),
            Optional.<Scroll>absent()
    );
    LuceneQueryBuilder.Context context = luceneQueryBuilder.convert(
            whereClause,  indexService.mapperService(), indexService.fieldData(), indexService.cache());
    searchContext.parsedQuery(new ParsedQuery(context.query(), EMPTY_NAMED_FILTERS));

    Float minScore = context.minScore();
    if (minScore != null) {
        searchContext.minimumScore(minScore);
    }

    return searchContext;
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
private CrateSearchShardRequest(long nowInMillis, Optional<Scroll> scroll,
                                IndexShard indexShard) {
    this.nowInMillis = nowInMillis;
    this.scroll = scroll.orNull();
    this.index = indexShard.indexService().index().name();
    this.shardId = indexShard.shardId().id();
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
@Builder
private ScrollingSearchHitSpliterator(@NonNull final SearchRequestBuilder searchRequest,
		@NonNull final Client client, @NonNull final Scroll scroll, final int size) {
	super(searchRequest.setScroll(scroll).setSize(size).execute());
	this.client = client;
	this.scroll = scroll;
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public void emit() {
    // 构建查询
    Scroll esScroll = new Scroll(new TimeValue(scroll, TimeUnit.MINUTES));
    SearchRequestBuilder searchRequestBuilder = client.prepareSearch(index).setScroll(esScroll).setQuery(query).setSize(size);
    if (StringUtils.isNotEmpty(type)) {
        searchRequestBuilder.setTypes(type);
    }

    if (logger.isDebugEnabled()) {
        logger.debug(searchRequestBuilder.toString());
    }

    String scrollId = null;
    try {
        SearchResponse searchResponse = searchRequestBuilder.execute().actionGet();
        scrollId = searchResponse.getScrollId();

        long total = searchResponse.getHits().getTotalHits();
        // 处理第一次search查询
        int messageCount = submitMessage(searchResponse);

        // 计算剩余的批次
        long remainTotal = total - messageCount;
        long batchNum = remainTotal % size == 0 ? remainTotal / size : (remainTotal / size) + 1;

        for (long i = 0; !stop && i < batchNum; i++) {
            // 按批查询数据
            SearchScrollRequestBuilder scrollRequestBuilder = client.prepareSearchScroll(scrollId).setScroll(esScroll);
            searchResponse = scrollRequestBuilder.execute().actionGet();
            submitMessage(searchResponse);
        }
    } catch (Exception e) {
        logger.error("query error", e);
    } finally {
        if (StringUtils.isNotEmpty(scrollId)) {
            client.prepareClearScroll().addScrollId(scrollId).execute().actionGet();
        }
    }
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
@Override
public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) {
    String scrollId = request.param("scroll_id");
    SearchScrollRequest searchScrollRequest = new SearchScrollRequest();
    searchScrollRequest.scrollId(scrollId);
    String scroll = request.param("scroll");
    if (scroll != null) {
        searchScrollRequest.scroll(new Scroll(parseTimeValue(scroll, null, "scroll")));
    }

    if (RestActions.hasBodyContent(request)) {
        XContentType type = XContentFactory.xContentType(RestActions.getRestContent(request));
        if (type == null) {
            if (scrollId == null) {
                scrollId = RestActions.getRestContent(request).toUtf8();
                searchScrollRequest.scrollId(scrollId);
            }
        } else {
            // NOTE: if rest request with xcontent body has request parameters, these parameters override xcontent values
            RestSearchScrollAction.buildFromContent(RestActions.getRestContent(request), searchScrollRequest);
        }
    }

    final String sparseFormat = request.param("sparse_format", "dict");
    client.searchScroll(searchScrollRequest, new RestBuilderListener<SearchResponse>(channel) {
        @Override
        public RestResponse buildResponse(SearchResponse resp, XContentBuilder builder) throws Exception {
            SearchVectorizeResponse searchVectorizeResponse = new SearchVectorizeResponse(resp);
            searchVectorizeResponse.setFormat(sparseFormat);
            searchVectorizeResponse.toXContent(builder, ToXContent.EMPTY_PARAMS);
            return new BytesRestResponse(OK, builder);
        }
    });
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
/**
 * If set, will enable scrolling of the search request.
 */
public NoopSearchRequestBuilder setScroll(Scroll scroll) {
    request.scroll(scroll);
    return this;
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
public Scroll scroll() {
    return scroll;
}
 

import org.elasticsearch.search.Scroll; //导入依赖的package包/类
@Override
public void readFrom(StreamInput in) throws IOException {
    super.readFrom(in);
    id = in.readLong();
    scroll = in.readOptionalWriteable(Scroll::new);
}