public class GetIndicesVersionAction extends Action<GetIndicesVersionRequest, GetIndicesVersionResponse, GetIndicesVersionRequestBuilder> {

  public static final GetIndicesVersionAction INSTANCE = new GetIndicesVersionAction();
  public static final String NAME = "indices:admin/version/get";

  private GetIndicesVersionAction() {
    super(NAME);
  }

  @Override
  public GetIndicesVersionResponse newResponse() {
    return new GetIndicesVersionResponse();
  }

  @Override
  public GetIndicesVersionRequestBuilder newRequestBuilder(ElasticsearchClient client) {
    return new GetIndicesVersionRequestBuilder(client, this);
  }

} public class IndexVersionShardService extends AbstractIndexShardComponent implements Closeable {

  private final AtomicLong version;

  private final IndexShard indexShard;
  private final VersioningIndexingOperationListener versioningIndexingOperationListener;

  @Inject
  public IndexVersionShardService(ShardId shardId, Settings indexSettings, IndexShard indexShard) {
    super(shardId, indexSettings);
    this.indexShard = indexShard;
    this.versioningIndexingOperationListener = new VersioningIndexingOperationListener();
    indexShard.indexingService().addListener(versioningIndexingOperationListener);
    version = new AtomicLong(System.nanoTime()); // initialise version number based on time to ensure uniqueness even if shard restarted
  }

  public long getVersion() {
    return version.get();
  }

  @Override
  public void close() {
    indexShard.indexingService().removeListener(versioningIndexingOperationListener);
  }

  private class VersioningIndexingOperationListener extends IndexingOperationListener {

    @Override
    public void postIndexUnderLock(Engine.Index index) {
      version.incrementAndGet();
    }

    @Override
    public void postDeleteUnderLock(Engine.Delete delete) {
      version.incrementAndGet();
    }

  }


} public class TransportGetIndicesVersionAction extends TransportBroadcastByNodeAction<GetIndicesVersionRequest, GetIndicesVersionResponse, ShardIndexVersion> {

  private final IndicesService indicesService;

  @Inject
  public TransportGetIndicesVersionAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
                                          TransportService transportService, IndicesService indicesService,
                                          ActionFilters actionFilters, IndexNameExpressionResolver indexNameExpressionResolver) {
    super(settings, GetIndicesVersionAction.NAME, threadPool, clusterService, transportService, actionFilters, indexNameExpressionResolver,
            GetIndicesVersionRequest.class, ThreadPool.Names.MANAGEMENT);
    this.indicesService = indicesService;
  }

  @Override
  protected ShardIndexVersion readShardResult(StreamInput in) throws IOException {
    return ShardIndexVersion.readShardIndexVersion(in);
  }

  @Override
  protected GetIndicesVersionResponse newResponse(GetIndicesVersionRequest request, int totalShards, int successfulShards, int failedShards, List<ShardIndexVersion> shardVersions, List<ShardOperationFailedException> shardFailures, ClusterState clusterState) {
    return new GetIndicesVersionResponse(shardVersions.toArray(new ShardIndexVersion[shardVersions.size()]), totalShards, successfulShards, failedShards, shardFailures);
  }

  @Override
  protected GetIndicesVersionRequest readRequestFrom(StreamInput in) throws IOException {
    GetIndicesVersionRequest request = new GetIndicesVersionRequest();
    request.readFrom(in);
    return request;
  }

  @Override
  protected ShardIndexVersion shardOperation(GetIndicesVersionRequest request, ShardRouting shardRouting) throws IOException {
    IndexService indexService = indicesService.indexServiceSafe(shardRouting.getIndex());
    IndexShard indexShard = indexService.shardSafe(shardRouting.id());

    // Get the IndexVersionShardService associated to this shard
    Injector injector = indexService.shardInjectorSafe(shardRouting.id());
    IndexVersionShardService indexVersionService = injector.getBinding(IndexVersionShardService.class).getProvider().get();
    long version = indexVersionService.getVersion();

    return new ShardIndexVersion(indexShard.routingEntry(), version);
  }

  /**
   * Goes across *all* shards, i.e., primaries and replicas.
   *
   * When elasticsearch will have resolved issue #10708, we will be able to compute an index version based on the
   * primary shards only.
   */
  @Override
  protected ShardsIterator shards(ClusterState clusterState, GetIndicesVersionRequest request, String[] concreteIndices) {
    return clusterState.routingTable().allShards(concreteIndices);
  }

  @Override
  protected ClusterBlockException checkGlobalBlock(ClusterState state, GetIndicesVersionRequest request) {
    return state.blocks().globalBlockedException(ClusterBlockLevel.METADATA_READ);
  }

  @Override
  protected ClusterBlockException checkRequestBlock(ClusterState state, GetIndicesVersionRequest request, String[] concreteIndices) {
    return state.blocks().indicesBlockedException(ClusterBlockLevel.METADATA_READ, concreteIndices);
  }

} public class CoordinateMultiSearchAction extends Action<MultiSearchRequest, MultiSearchResponse, MultiSearchRequestBuilder> {

  public static final CoordinateMultiSearchAction INSTANCE = new CoordinateMultiSearchAction();
  public static final String NAME = "indices:data/read/coordinate-msearch";

  private CoordinateMultiSearchAction() {
    super(NAME);
  }

  @Override
  public MultiSearchRequestBuilder newRequestBuilder(ElasticsearchClient client) {
    return new CoordinateMultiSearchRequestBuilder(client);
  }

  @Override
  public MultiSearchResponse newResponse() {
    return new CoordinateMultiSearchResponse();
  }

} public class CoordinateSearchAction extends Action<SearchRequest, SearchResponse, SearchRequestBuilder> {

  public static final CoordinateSearchAction INSTANCE = new CoordinateSearchAction();
  public static final String NAME = "indices:data/read/coordinate-search";

  private CoordinateSearchAction() {
    super(NAME);
  }

  @Override
  public SearchRequestBuilder newRequestBuilder(ElasticsearchClient client) {
    return new CoordinateSearchRequestBuilder(client);
  }

  @Override
  public SearchResponse newResponse() {
    return new CoordinateSearchResponse();
  }
} public class TransportCoordinateMultiSearchAction extends BaseTransportCoordinateSearchAction<MultiSearchRequest, MultiSearchResponse> {

  private final ClusterService clusterService;

  private final TransportSearchAction searchAction;

  private final FilterJoinCacheService cacheService;

  @Inject
  public TransportCoordinateMultiSearchAction(Settings settings, ThreadPool threadPool,
                                              TransportService transportService, ClusterService clusterService,
                                              FilterJoinCacheService cacheService,
                                              TransportSearchAction search, ActionFilters actionFilters,
                                              IndexNameExpressionResolver indexNameExpressionResolver, Client client) {
    super(settings, CoordinateMultiSearchAction.NAME, threadPool, transportService, actionFilters,
            indexNameExpressionResolver, client, MultiSearchRequest.class);
    this.searchAction = search;
    this.clusterService = clusterService;
    this.cacheService = cacheService;
  }

  @Override
  protected void doExecute(final MultiSearchRequest request, final ActionListener<MultiSearchResponse> listener) {
    logger.debug("{}: Execute coordinated multi-search action", Thread.currentThread().getName());

    List<CoordinateSearchMetadata> metadatas = new ArrayList<>(request.requests().size());
    this.doExecuteFilterJoins(request, metadatas);
    this.doExecuteRequest(request, listener, metadatas);

    logger.debug("{}: Coordinated multi-search action completed", Thread.currentThread().getName());
  }

  private void doExecuteFilterJoins(final MultiSearchRequest request,
                                    final List<CoordinateSearchMetadata> metadatas) {
    FilterJoinCache cache = cacheService.getCacheInstance();

    for (int i = 0; i < request.requests().size(); i++) {
      // Parse query source
      Tuple<XContentType, Map<String, Object>> parsedSource = this.parseSource(request.requests().get(i).source());
      Map<String, Object> map = parsedSource.v2();

      // Unwrap "wrapper" queries
      WrapperQueryVisitor wrapperVisitor = new WrapperQueryVisitor(map);
      wrapperVisitor.traverse();

      // Query planning and execution of filter joins
      SourceMapVisitor mapVisitor = new SourceMapVisitor(map);
      mapVisitor.traverse();
      FilterJoinVisitor joinVisitor = new FilterJoinVisitor(client, mapVisitor.getFilterJoinTree(), cache, request);
      joinVisitor.traverse();
      metadatas.add(joinVisitor.getMetadata());

      // Filter joins have been replaced by a binary terms filter
      // Rebuild the query source, and delegate the execution of the search action
      request.requests().get(i).source(this.buildSource(parsedSource.v1().xContent(), map));
    }
  }

  private void doExecuteRequest(final MultiSearchRequest request, final ActionListener<MultiSearchResponse> listener,
                                final List<CoordinateSearchMetadata> metadatas) {
    ClusterState clusterState = clusterService.state();
    clusterState.blocks().globalBlockedRaiseException(ClusterBlockLevel.READ);

    final AtomicArray<CoordinateMultiSearchResponse.Item> responses = new AtomicArray<>(request.requests().size());
    final AtomicInteger counter = new AtomicInteger(responses.length());
    for (int i = 0; i < responses.length(); i++) {
      final int index = i;
      SearchRequest searchRequest = new SearchRequest(request.requests().get(i), request);
      searchAction.execute(searchRequest, new ActionListener<SearchResponse>() {

        @Override
        public void onResponse(SearchResponse searchResponse) {
          responses.set(index, new CoordinateMultiSearchResponse.Item(new CoordinateSearchResponse(searchResponse, metadatas.get(index)), null));
          if (counter.decrementAndGet() == 0) {
            finishHim();
          }
        }

        @Override
        public void onFailure(Throwable e) {
          responses.set(index, new CoordinateMultiSearchResponse.Item(null, ExceptionsHelper.detailedMessage(e)));
          if (counter.decrementAndGet() == 0) {
            finishHim();
          }
        }

        private void finishHim() {
          listener.onResponse(new CoordinateMultiSearchResponse(responses.toArray(new CoordinateMultiSearchResponse.Item[responses.length()])));
        }

      });
    }
  }

} public class TransportCoordinateSearchAction extends BaseTransportCoordinateSearchAction<SearchRequest, SearchResponse> {

  private final TransportSearchAction searchAction;

  private final FilterJoinCacheService cacheService;

  @Inject
  public TransportCoordinateSearchAction(Settings settings, ThreadPool threadPool,
                                         TransportService transportService, FilterJoinCacheService cacheService,
                                         ActionFilters actionFilters, TransportSearchAction searchAction,
                                         IndexNameExpressionResolver indexNameExpressionResolver, Client client) {
    super(settings, CoordinateSearchAction.NAME, threadPool, transportService, actionFilters,
            indexNameExpressionResolver, client, SearchRequest.class);
    this.searchAction = searchAction;
    this.cacheService = cacheService;
  }

  @Override
  protected void doExecute(final SearchRequest request, final ActionListener<SearchResponse> listener) {
    logger.debug("{}: Execute coordinated search action", Thread.currentThread().getName());

    // A reference to the listener that will be used - can be overwritten to reference a CoordinateSearchListener
    ActionListener<SearchResponse> actionListener = listener;

    // Retrieve the singleton instance of the filterjoin cache
    FilterJoinCache cache = cacheService.getCacheInstance();

    // Parse query source
    Tuple<XContentType, Map<String, Object>> parsedSource = this.parseSource(request.source());
    if (parsedSource != null) { // can be null if this is a uri search (query parameter in extraSource)
      Map<String, Object> map = parsedSource.v2();

      // Unwrap "wrapper" queries
      WrapperQueryVisitor wrapperVisitor = new WrapperQueryVisitor(map);
      wrapperVisitor.traverse();

      // Query planning and execution of filter joins
      SourceMapVisitor mapVisitor = new SourceMapVisitor(map);
      mapVisitor.traverse();
      FilterJoinVisitor joinVisitor = new FilterJoinVisitor(client, mapVisitor.getFilterJoinTree(), cache, request);
      joinVisitor.traverse();

      // Wraps the listener with our own to inject metadata information in the response
      CoordinateSearchListener coordinateSearchListener = new CoordinateSearchListener(listener);
      coordinateSearchListener.setMetadata(joinVisitor.getMetadata());
      actionListener = coordinateSearchListener;

      // Filter joins have been replaced by a binary terms filter
      // Rebuild the query source, and delegate the execution of the search action
      request.source(this.buildSource(parsedSource.v1().xContent(), map));
    }

    // Delegate the execution of the request to the original search action
    this.searchAction.execute(request, actionListener);

    logger.debug("{}: Coordinated search action completed", Thread.currentThread().getName());
  }

  /**
   * Wrapper around a listener that is responsible for injecting the coordinate search metadata
   * into the search response.
   */
  public static class CoordinateSearchListener implements ActionListener<SearchResponse> {

    private final ActionListener<SearchResponse> actionListener;

    private CoordinateSearchMetadata metadata;

    public CoordinateSearchListener(final ActionListener<SearchResponse> listener) {
      this.actionListener = listener;
    }

    public void setMetadata(CoordinateSearchMetadata metadata) {
      this.metadata = metadata;
    }

    @Override
    public final void onResponse(SearchResponse response) {
      CoordinateSearchResponse r = new CoordinateSearchResponse(response, metadata);
      this.actionListener.onResponse(r);
    }

    @Override
    public final void onFailure(Throwable e) {
      this.actionListener.onFailure(e);
    }

  }

} public class TermsByQueryAction extends Action<TermsByQueryRequest, TermsByQueryResponse, TermsByQueryRequestBuilder> {

  public static final TermsByQueryAction INSTANCE = new TermsByQueryAction();
  public static final String NAME = "indices:data/read/search/termsbyquery";

  /**
   * Default constructor
   */
  private TermsByQueryAction() {
    super(NAME);
  }

  /**
   * Gets a new {@link TermsByQueryResponse} object
   *
   * @return the new {@link TermsByQueryResponse}.
   */
  @Override
  public TermsByQueryResponse newResponse() {
    return new TermsByQueryResponse();
  }

  /**
   * Set transport options specific to a terms by query request.
   * Enabling compression here does not really reduce data transfer, even increase it on the contrary.
   *
   * @param settings node settings
   * @return the request options.
   */
  @Override
  public TransportRequestOptions transportOptions(Settings settings) {
    return TransportRequestOptions.builder()
            .withType(TransportRequestOptions.Type.REG)
            .build();
  }

  /**
   * Get a new {@link TermsByQueryRequestBuilder}
   *
   * @param client the client responsible for executing the request.
   * @return the new {@link TermsByQueryRequestBuilder}
   */
  @Override
  public TermsByQueryRequestBuilder newRequestBuilder(ElasticsearchClient client) {
    return new TermsByQueryRequestBuilder(client, this);
  }

} public class TransportTermsByQueryAction extends TransportBroadcastAction<TermsByQueryRequest, TermsByQueryResponse, TermsByQueryShardRequest, TermsByQueryShardResponse> {

  private final IndicesService indicesService;
  private final ScriptService scriptService;
  private final PageCacheRecycler pageCacheRecycler;
  private final BigArrays bigArrays;
  private final CircuitBreakerService breakerService;
  private final Client client;

  /**
   * Constructor
   */
  @Inject
  public TransportTermsByQueryAction(Settings settings, ThreadPool threadPool, ClusterService clusterService,
                                     TransportService transportService, IndicesService indicesService,
                                     CircuitBreakerService breakerService,
                                     ScriptService scriptService, PageCacheRecycler pageCacheRecycler,
                                     BigArrays bigArrays, ActionFilters actionFilters,
                                     IndexNameExpressionResolver indexNameExpressionResolver, Client client) {
    super(settings, TermsByQueryAction.NAME, threadPool, clusterService, transportService, actionFilters,
            indexNameExpressionResolver, TermsByQueryRequest.class, TermsByQueryShardRequest.class,
            // Use the generic threadpool which is cached, as we can end up with deadlock with the SEARCH threadpool
            ThreadPool.Names.GENERIC);
    this.indicesService = indicesService;
    this.scriptService = scriptService;
    this.pageCacheRecycler = pageCacheRecycler;
    this.bigArrays = bigArrays;
    this.breakerService = breakerService;
    this.client = client;
  }

  /**
   * Executes the actions.
   */
  @Override
  protected void doExecute(Task task, TermsByQueryRequest request, ActionListener<TermsByQueryResponse> listener) {
    request.nowInMillis(System.currentTimeMillis()); // set time to be used in scripts
    super.doExecute(task, request, listener);
  }

  /**
   * Creates a new {@link TermsByQueryShardRequest}
   */
  @Override
  protected TermsByQueryShardRequest newShardRequest(int numShards, ShardRouting shard, TermsByQueryRequest request) {
    String[] filteringAliases = indexNameExpressionResolver.filteringAliases(clusterService.state(), shard.index(), request.indices());
    return new TermsByQueryShardRequest(shard.shardId(), filteringAliases, request);
  }

  /**
   * Creates a new {@link TermsByQueryShardResponse}
   */
  @Override
  protected TermsByQueryShardResponse newShardResponse() {
    return new TermsByQueryShardResponse(breakerService.getBreaker(CircuitBreaker.REQUEST));
  }

  /**
   * The shards this request will execute against.
   */
  @Override
  protected GroupShardsIterator shards(ClusterState clusterState, TermsByQueryRequest request, String[] concreteIndices) {
    Map<String, Set<String>> routingMap = indexNameExpressionResolver.resolveSearchRouting(clusterState, request.routing(), request.indices());
    return clusterService.operationRouting().searchShards(clusterState, concreteIndices, routingMap, request.preference());
  }

  @Override
  protected ClusterBlockException checkGlobalBlock(ClusterState state, TermsByQueryRequest request) {
    return state.blocks().globalBlockedException(ClusterBlockLevel.READ);
  }

  @Override
  protected ClusterBlockException checkRequestBlock(ClusterState state, TermsByQueryRequest request, String[] concreteIndices) {
    return state.blocks().indicesBlockedException(ClusterBlockLevel.READ, concreteIndices);
  }

  /**
   * Merges the individual shard responses and returns the final {@link TermsByQueryResponse}.
   */
  @Override
  protected TermsByQueryResponse newResponse(TermsByQueryRequest request, AtomicReferenceArray shardsResponses, ClusterState clusterState) {
    int successfulShards = 0;
    int failedShards = 0;
    int numTerms = 0;
    TermsSet[] termsSets = new TermsSet[shardsResponses.length()];
    List<ShardOperationFailedException> shardFailures = null;

    // we check each shard response
    for (int i = 0; i < shardsResponses.length(); i++) {
      Object shardResponse = shardsResponses.get(i);
      if (shardResponse == null) {
        // simply ignore non active shards
      } else if (shardResponse instanceof BroadcastShardOperationFailedException) {
        failedShards++;
        if (shardFailures == null) {
          shardFailures = new ArrayList<>();
        }
        logger.error("Shard operation failed", (BroadcastShardOperationFailedException) shardResponse);
        shardFailures.add(new DefaultShardOperationFailedException((BroadcastShardOperationFailedException) shardResponse));
      } else {
        // on successful shard response, just add to the array or responses so we can process them below
        // we calculate the total number of terms gathered across each shard so we can use it during
        // initialization of the final TermsResponse below (to avoid rehashing during merging)
        TermsByQueryShardResponse shardResp = ((TermsByQueryShardResponse) shardResponse);
        TermsSet terms = shardResp.getTerms();
        termsSets[i] = terms;
        numTerms += terms.size();
        successfulShards++;
      }
    }

    // Merge the responses

    try {
      // NumericTermsSet is responsible for the merge, set size to avoid rehashing on certain implementations.
      long expectedElements = request.expectedTerms() != null ? request.expectedTerms() : numTerms;
      TermsSet termsSet = TermsSet.newTermsSet(expectedElements, request.termsEncoding(), breakerService.getBreaker(CircuitBreaker.REQUEST));

      TermsByQueryResponse rsp;
      try {
        for (int i = 0; i < termsSets.length; i++) {
          TermsSet terms = termsSets[i];
          if (terms != null) {
            termsSet.merge(terms);
            terms.release(); // release the shard terms set and adjust the circuit breaker
            termsSets[i] = null;
          }
        }

        long tookInMillis = System.currentTimeMillis() - request.nowInMillis();

        rsp = new TermsByQueryResponse(termsSet, tookInMillis, shardsResponses.length(), successfulShards, failedShards, shardFailures);
      }
      finally {
        // we can now release the terms set and adjust the circuit breaker, since the TermsByQueryResponse holds an
        // encoded version of the terms set
        termsSet.release();
      }

      return rsp;
    }
    finally { // If something happens, release the terms sets and adjust the circuit breaker
      for (int i = 0; i < termsSets.length; i++) {
        TermsSet terms = termsSets[i];
        if (terms != null) {
          terms.release();
        }
      }
    }
  }

  /**
   * The operation that executes the query and generates a {@link TermsByQueryShardResponse} for each shard.
   */
  @Override
  protected TermsByQueryShardResponse shardOperation(TermsByQueryShardRequest shardRequest) throws ElasticsearchException {
    IndexService indexService = indicesService.indexServiceSafe(shardRequest.shardId().getIndex());
    IndexShard indexShard = indexService.shardSafe(shardRequest.shardId().id());
    TermsByQueryRequest request = shardRequest.request();
    OrderByShardOperation orderByOperation = OrderByShardOperation.get(request.getOrderBy(), request.maxTermsPerShard());

    SearchShardTarget shardTarget = new SearchShardTarget(clusterService.localNode().id(),
                                                          shardRequest.shardId().getIndex(),
                                                          shardRequest.shardId().id());

    ShardSearchRequest shardSearchRequest = new ShardSearchLocalRequest(request.types(), request.nowInMillis(),
                                                                        shardRequest.filteringAliases());

    SearchContext context = new DefaultSearchContext(0, shardSearchRequest, shardTarget,
      indexShard.acquireSearcher("termsByQuery"), indexService, indexShard, scriptService,
      pageCacheRecycler, bigArrays, threadPool.estimatedTimeInMillisCounter(), parseFieldMatcher,
      SearchService.NO_TIMEOUT);
    SearchContext.setCurrent(context);

    try {
      MappedFieldType fieldType = context.smartNameFieldType(request.field());
      if (fieldType == null) {
        throw new SearchContextException(context, "[termsByQuery] field '" + request.field() +
                "' not found for types " + Arrays.toString(request.types()));
      }

      IndexFieldData indexFieldData = context.fieldData().getForField(fieldType);

      BytesReference querySource = request.querySource();
      if (querySource != null && querySource.length() > 0) {
        XContentParser queryParser = null;
        try {
          queryParser = XContentFactory.xContent(querySource).createParser(querySource);
          QueryParseContext.setTypes(request.types());
          ParsedQuery parsedQuery = orderByOperation.getParsedQuery(queryParser, indexService);
          if (parsedQuery != null) {
            context.parsedQuery(parsedQuery);
          }
        }
        finally {
          QueryParseContext.removeTypes();
          if (queryParser != null) {
            queryParser.close();
          }
        }
      }

      context.preProcess();

      // execute the search only gathering the hit count and bitset for each segment
      logger.debug("{}: Executes search for collecting terms {}", Thread.currentThread().getName(),
        shardRequest.shardId());

      TermsCollector termsCollector = this.getTermsCollector(request.termsEncoding(), indexFieldData, context);
      if (request.expectedTerms() != null) termsCollector.setExpectedTerms(request.expectedTerms());
      if (request.maxTermsPerShard() != null) termsCollector.setMaxTerms(request.maxTermsPerShard());
      HitStream hitStream = orderByOperation.getHitStream(context);
      TermsSet terms = termsCollector.collect(hitStream);

      logger.debug("{}: Returns terms response with {} terms for shard {}", Thread.currentThread().getName(),
        terms.size(), shardRequest.shardId());

      return new TermsByQueryShardResponse(shardRequest.shardId(), terms);
    }
    catch (Throwable e) {
      logger.error("[termsByQuery] Error executing shard operation", e);
      throw new QueryPhaseExecutionException(context, "[termsByQuery] Failed to execute query", e);
    }
    finally {
      // this will also release the index searcher
      context.close();
      SearchContext.removeCurrent();
    }
  }

  private TermsCollector getTermsCollector(TermsByQueryRequest.TermsEncoding termsEncoding,
                                           IndexFieldData indexFieldData, SearchContext context) {
    switch (termsEncoding) {
      case LONG:
        return new LongTermsCollector(indexFieldData, context, breakerService.getBreaker(CircuitBreaker.REQUEST));
      case INTEGER:
        return new IntegerTermsCollector(indexFieldData, context, breakerService.getBreaker(CircuitBreaker.REQUEST));
      case BLOOM:
        return new BloomFilterTermsCollector(indexFieldData, context, breakerService.getBreaker(CircuitBreaker.REQUEST));
      case BYTES:
        return new BytesRefTermsCollector(indexFieldData, context, breakerService.getBreaker(CircuitBreaker.REQUEST));
      default:
        throw new IllegalArgumentException("[termsByQuery] Invalid terms encoding: " + termsEncoding.name());
    }
  }

  /**
   * Abstraction over the logic of the order by operation.
   */
  private static abstract class OrderByShardOperation {

    protected final Integer maxTermsPerShard;

    private OrderByShardOperation(Integer maxTermsPerShard) {
      this.maxTermsPerShard = maxTermsPerShard;
    }

    /**
     * Returns the {@link ParsedQuery} associated to this order by operation.
     */
    protected ParsedQuery getParsedQuery(final XContentParser queryParser, final IndexService indexService) {
      return indexService.queryParserService().parse(queryParser);
    }

    /**
     * Returns the {@link HitStream} associated to this order by operation.
     */
    protected abstract HitStream getHitStream(final SearchContext context) throws IOException;

    /**
     * Instantiates the appropriate {@link OrderByShardOperation} for the given
     * {@link TermsByQueryRequest.Ordering}.
     * Default to {@link TermsByQueryRequest.Ordering#DEFAULT}.
     */
    private static OrderByShardOperation get(final TermsByQueryRequest.Ordering orderBy, final Integer maxTermsPerShard) {
      // By default, no ordering
      TermsByQueryRequest.Ordering ordering = orderBy != null ? orderBy : TermsByQueryRequest.Ordering.DEFAULT;
      switch (ordering) {
        case DEFAULT:
          return new OrderByDefaultShardOperation(maxTermsPerShard);

        case DOC_SCORE:
          return new OrderByDocScoreShardOperation(maxTermsPerShard);

        default:
          throw new ElasticsearchParseException("[termsByQuery] unknown ordering " + ordering.name());
      }
    }

  }

  /**
   * The default order by operation. Document score will not be computed, and documents will be ordered
   * by their id.
   */
  private static class OrderByDefaultShardOperation extends OrderByShardOperation {

    private OrderByDefaultShardOperation(final Integer maxTermsPerShard) {
      super(maxTermsPerShard);
    }

    @Override
    protected HitStream getHitStream(SearchContext context) throws IOException {
      return new BitSetHitStream(context.query(), context.searcher());
    }

  }

  /**
   * Order by operation based on document score. Document score will be computed, and documents will be ordered
   * by their score.
   */
  private static class OrderByDocScoreShardOperation extends OrderByShardOperation {

    private OrderByDocScoreShardOperation(final Integer maxTermsPerShard) {
      super(maxTermsPerShard);
    }

    @Override
    protected HitStream getHitStream(SearchContext context) throws IOException {
      if (maxTermsPerShard == null) {
        throw new ElasticsearchParseException("[termsByQuery] maxTermsPerShard parameter is null");
      }
      return new TopHitStream(maxTermsPerShard, context.query(), context.searcher());
    }
  }

} public class FieldDataTermsQueryParser implements QueryParser {

  public static final String NAME = "fielddata_terms";

  private static final ESLogger logger = Loggers.getLogger(FieldDataTermsQueryParser.class);

  public FieldDataTermsQueryParser() {}

  @Override
  public String[] names() {
    return new String[]{NAME};
  }

  @Override
  public Query parse(QueryParseContext parseContext) throws IOException, QueryParsingException {
    XContentParser parser = parseContext.parser();

    XContentParser.Token token = parser.nextToken();
    if (token != XContentParser.Token.FIELD_NAME) {
        throw new QueryParsingException(parseContext, "[fielddata_terms] a field name is required");
    }
    String fieldName = parser.currentName();

    String queryName = null;
    byte[] value = null;
    Long cacheKey = null;

    token = parser.nextToken();
    if (token == XContentParser.Token.START_OBJECT) {
      String currentFieldName = null;
      while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
        if (token == XContentParser.Token.FIELD_NAME) {
          currentFieldName = parser.currentName();
        } else {
          if ("value".equals(currentFieldName)) {
            value = parser.binaryValue();
          } else if ("_name".equals(currentFieldName)) {
            queryName = parser.text();
          } else if ("_cache_key".equals(currentFieldName) || "_cacheKey".equals(currentFieldName)) {
            cacheKey = parser.longValue();
          } else {
            throw new QueryParsingException(parseContext, "[fielddata_terms] filter does not support [" + currentFieldName + "]");
          }
        }
      }
      parser.nextToken();
    } else {
      value = parser.binaryValue();
      // move to the next token
      parser.nextToken();
    }

    if (value == null) {
      throw new QueryParsingException(parseContext, "[fielddata_terms] a binary value is required");
    }
    if (cacheKey == null) { // cache key is mandatory - see #170
      throw new QueryParsingException(parseContext, "[fielddata_terms] a cache key is required");
    }

    if (fieldName == null) {
      throw new QueryParsingException(parseContext, "[fielddata_terms] a field name is required");
    }

    MappedFieldType fieldType = parseContext.fieldMapper(fieldName);
    if (fieldType == null) {
      return new MatchNoDocsQuery();
    }

    IndexFieldData fieldData = parseContext.getForField(fieldType);
    Query query = this.toFieldDataTermsQuery(fieldType, fieldData, value, cacheKey);

    if (queryName != null) {
      parseContext.addNamedQuery(queryName, query);
    }

    return query;
  }

  private final Query toFieldDataTermsQuery(MappedFieldType fieldType, IndexFieldData fieldData,
                                            byte[] encodedTerms, long cacheKey) {
    Query query = null;

    if (fieldType instanceof NumberFieldMapper.NumberFieldType) {
      query = FieldDataTermsQuery.newLongs(encodedTerms, (IndexNumericFieldData) fieldData, cacheKey);
    } else if (fieldType instanceof StringFieldMapper.StringFieldType) {
      query = FieldDataTermsQuery.newBytes(encodedTerms, fieldData, cacheKey);
    } else {
      throw new ElasticsearchParseException("[fielddata_terms] query does not support field data type " + fieldType.fieldDataType().getType());
    }

    return query;
  }

} public class TermsEnumTermsQueryParser implements QueryParser {

  public static final String NAME = "termsenum_terms";

  private static final ESLogger logger = Loggers.getLogger(TermsEnumTermsQueryParser.class);

  public TermsEnumTermsQueryParser() {}

  @Override
  public String[] names() {
    return new String[]{NAME};
  }

  @Override
  public Query parse(QueryParseContext parseContext) throws IOException, QueryParsingException {
    XContentParser parser = parseContext.parser();

    XContentParser.Token token = parser.nextToken();
    if (token != XContentParser.Token.FIELD_NAME) {
        throw new QueryParsingException(parseContext, "[termsenum_terms] a field name is required");
    }
    String fieldName = parser.currentName();

    String queryName = null;
    byte[] value = null;
    Long cacheKey = null;

    token = parser.nextToken();
    if (token == XContentParser.Token.START_OBJECT) {
      String currentFieldName = null;
      while ((token = parser.nextToken()) != XContentParser.Token.END_OBJECT) {
        if (token == XContentParser.Token.FIELD_NAME) {
          currentFieldName = parser.currentName();
        } else {
          if ("value".equals(currentFieldName)) {
              value = parser.binaryValue();
          } else if ("_name".equals(currentFieldName)) {
              queryName = parser.text();
          } else if ("_cache_key".equals(currentFieldName) || "_cacheKey".equals(currentFieldName)) {
              cacheKey = parser.longValue();
          } else {
            throw new QueryParsingException(parseContext, "[termsenum_terms] filter does not support [" + currentFieldName + "]");
          }
        }
      }
      parser.nextToken();
    } else {
      value = parser.binaryValue();
      // move to the next token
      parser.nextToken();
    }

    if (value == null) {
      throw new QueryParsingException(parseContext, "[termsenum_terms] a binary value is required");
    }
    if (cacheKey == null) { // cache key is mandatory - see #170
      throw new QueryParsingException(parseContext, "[termsenum_terms] a cache key is required");
    }

    if (fieldName == null) {
      throw new QueryParsingException(parseContext, "[termsenum_terms] a field name is required");
    }

    MappedFieldType fieldType = parseContext.fieldMapper(fieldName);
    if (fieldType == null) {
      return new MatchNoDocsQuery();
    }

    Query query = new TermsEnumTermsQuery(value, fieldName, cacheKey);

    if (queryName != null) {
      parseContext.addNamedQuery(queryName, query);
    }

    return query;
  }

} public class RestClearFilterJoinCacheAction extends BaseRestHandler {

  @Inject
  public RestClearFilterJoinCacheAction(final Settings settings, final RestController controller, final Client client) {
    super(settings, controller, client);
    controller.registerHandler(RestRequest.Method.POST, "/_filter_join/cache/clear", this);
    controller.registerHandler(RestRequest.Method.GET, "/_filter_join/cache/clear", this);
  }

  @Override
  protected void handleRequest(RestRequest request, RestChannel channel, Client client) throws Exception {
    ClearFilterJoinCacheRequest clearFilterJoinCacheRequest = new ClearFilterJoinCacheRequest();
    client.execute(ClearFilterJoinCacheAction.INSTANCE, clearFilterJoinCacheRequest, new RestToXContentListener<ClearFilterJoinCacheResponse>(channel));
  }

} public class RestCoordinateMultiSearchAction extends BaseRestHandler {

  private final boolean allowExplicitIndex;

  @Inject
  public RestCoordinateMultiSearchAction(final Settings settings, final RestController controller, final Client client) {
    super(settings, controller, client);
    controller.registerHandler(GET, "/_coordinate_msearch", this);
    controller.registerHandler(POST, "/_coordinate_msearch", this);
    controller.registerHandler(GET, "/{index}/_coordinate_msearch", this);
    controller.registerHandler(POST, "/{index}/_coordinate_msearch", this);
    controller.registerHandler(GET, "/{index}/{type}/_coordinate_msearch", this);
    controller.registerHandler(POST, "/{index}/{type}/_coordinate_msearch", this);

    this.allowExplicitIndex = settings.getAsBoolean("rest.action.multi.allow_explicit_index", true);
  }

  @Override
  public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) throws Exception {
    MultiSearchRequest multiSearchRequest = new MultiSearchRequest();

    String[] indices = Strings.splitStringByCommaToArray(request.param("index"));
    String[] types = Strings.splitStringByCommaToArray(request.param("type"));
    String path = request.path();
    boolean isTemplateRequest = isTemplateRequest(path);
    IndicesOptions indicesOptions = IndicesOptions.fromRequest(request, multiSearchRequest.indicesOptions());
    multiSearchRequest.add(RestActions.getRestContent(request), isTemplateRequest, indices, types, request.param("search_type"), request.param("routing"), indicesOptions, allowExplicitIndex);

    client.execute(CoordinateMultiSearchAction.INSTANCE, multiSearchRequest, new RestToXContentListener<MultiSearchResponse>(channel));
  }

  private boolean isTemplateRequest(String path) {
    return (path != null && path.endsWith("/template"));
  }

} public class RestCoordinateSearchAction extends BaseRestHandler {

  @Inject
  public RestCoordinateSearchAction(final Settings settings, final RestController controller, final Client client) {
    super(settings, controller, client);
    controller.registerHandler(GET, "/_coordinate_search", this);
    controller.registerHandler(POST, "/_coordinate_search", this);
    controller.registerHandler(GET, "/{index}/_coordinate_search", this);
    controller.registerHandler(POST, "/{index}/_coordinate_search", this);
    controller.registerHandler(GET, "/{index}/{type}/_coordinate_search", this);
    controller.registerHandler(POST, "/{index}/{type}/_coordinate_search", this);
    controller.registerHandler(GET, "/_coordinate_search/template", this);
    controller.registerHandler(POST, "/_coordinate_search/template", this);
    controller.registerHandler(GET, "/{index}/_coordinate_search/template", this);
    controller.registerHandler(POST, "/{index}/_coordinate_search/template", this);
    controller.registerHandler(GET, "/{index}/{type}/_coordinate_search/template", this);
    controller.registerHandler(POST, "/{index}/{type}/_coordinate_search/template", this);

    // TODO: Redirects to original rest exists action, therefore it will not support filterjoin filter. It would be better to have our own coordinate exists action.
    RestExistsAction restExistsAction = new RestExistsAction(settings, controller, client);
    controller.registerHandler(GET, "/_coordinate_search/exists", restExistsAction);
    controller.registerHandler(POST, "/_coordinate_search/exists", restExistsAction);
    controller.registerHandler(GET, "/{index}/_coordinate_search/exists", restExistsAction);
    controller.registerHandler(POST, "/{index}/_coordinate_search/exists", restExistsAction);
    controller.registerHandler(GET, "/{index}/{type}/_coordinate_search/exists", restExistsAction);
    controller.registerHandler(POST, "/{index}/{type}/_coordinate_search/exists", restExistsAction);
  }

  @Override
  public void handleRequest(final RestRequest request, final RestChannel channel, final Client client) {
    SearchRequest searchRequest = new SearchRequest();
    RestSearchAction.parseSearchRequest(searchRequest, request, parseFieldMatcher, null);
    client.execute(CoordinateSearchAction.INSTANCE, searchRequest, new RestStatusToXContentListener<SearchResponse>(channel));
  }

} public class RestStatsFilterJoinCacheAction extends BaseRestHandler {

  @Inject
  public RestStatsFilterJoinCacheAction(final Settings settings, final RestController controller, final Client client) {
    super(settings, controller, client);
    controller.registerHandler(RestRequest.Method.POST, "/_filter_join/cache/stats", this);
    controller.registerHandler(RestRequest.Method.GET, "/_filter_join/cache/stats", this);
  }

  @Override
  protected void handleRequest(RestRequest request, RestChannel channel, Client client) throws Exception {
    StatsFilterJoinCacheRequest statsFilterJoinCacheRequest = new StatsFilterJoinCacheRequest();
    client.execute(StatsFilterJoinCacheAction.INSTANCE, statsFilterJoinCacheRequest, new RestToXContentListener<StatsFilterJoinCacheResponse>(channel));
  }

}
import org.elasticsearch.action.ActionModule;
import org.elasticsearch.cluster.node.DiscoveryNode;
import org.elasticsearch.common.inject.Module;
import org.elasticsearch.index.cache.IndexCacheModule;
import org.elasticsearch.indices.IndicesModule;
import org.elasticsearch.plugins.Plugin;
import org.elasticsearch.rest.RestModule;
import org.elasticsearch.common.inject.Inject;
import org.elasticsearch.common.settings.Settings;
import solutions.siren.join.action.admin.cache.*;
import solutions.siren.join.action.admin.version.GetIndicesVersionAction;
import solutions.siren.join.action.admin.version.IndexVersionShardService;
import solutions.siren.join.action.admin.version.TransportGetIndicesVersionAction;
import solutions.siren.join.action.coordinate.CoordinateMultiSearchAction;
import solutions.siren.join.action.coordinate.CoordinateSearchAction;
import solutions.siren.join.action.coordinate.TransportCoordinateMultiSearchAction;
import solutions.siren.join.action.coordinate.TransportCoordinateSearchAction;
import solutions.siren.join.action.terms.TermsByQueryAction;
import solutions.siren.join.action.terms.TransportTermsByQueryAction;
import solutions.siren.join.index.query.FieldDataTermsQueryParser;
import solutions.siren.join.index.query.TermsEnumTermsQueryParser;
import solutions.siren.join.rest.RestClearFilterJoinCacheAction;
import solutions.siren.join.rest.RestCoordinateMultiSearchAction;
import solutions.siren.join.rest.RestCoordinateSearchAction;
import solutions.siren.join.rest.RestStatsFilterJoinCacheAction;
import java.io.Closeable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
/**
 * Copyright (c) 2016, SIREn Solutions. All Rights Reserved.
 *
 * This file is part of the SIREn project.
 *
 * SIREn is a free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as
 * published by the Free Software Foundation, either version 3 of
 * the License, or (at your option) any later version.
 *
 * SIREn is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU Affero General Public License for more details.
 *
 * You should have received a copy of the GNU Affero General Public
 * License along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
package solutions.siren.join;




/**
 * The SIREn Join plugin.
 */
public class SirenJoinPlugin extends Plugin {

  private final boolean isEnabled;

  @Inject
  public SirenJoinPlugin(Settings settings) {
    if (DiscoveryNode.clientNode(settings)) {
      this.isEnabled = "node".equals(settings.get("client.type"));
    }
    else {
      this.isEnabled = true;
    }
  }

  public void onModule(ActionModule module) {
    module.registerAction(TermsByQueryAction.INSTANCE, TransportTermsByQueryAction.class);
    module.registerAction(CoordinateSearchAction.INSTANCE, TransportCoordinateSearchAction.class);
    module.registerAction(CoordinateMultiSearchAction.INSTANCE, TransportCoordinateMultiSearchAction.class);
    module.registerAction(ClearFilterJoinCacheAction.INSTANCE, TransportClearFilterJoinCacheAction.class);
    module.registerAction(StatsFilterJoinCacheAction.INSTANCE, TransportStatsFilterJoinCacheAction.class);
    module.registerAction(GetIndicesVersionAction.INSTANCE, TransportGetIndicesVersionAction.class);
  }

  public void onModule(IndicesModule module) {
    module.registerQueryParser(FieldDataTermsQueryParser.class);
    module.registerQueryParser(TermsEnumTermsQueryParser.class);
  }

  public void onModule(RestModule module) {
    module.addRestAction(RestCoordinateSearchAction.class);