class CountAction extends Action<CountRequest> {

    @Override
    public ActionResponse execute(CountRequest parameter) throws QueryStoreException {
        parameter.setTable(ElasticsearchUtils.getValidTableName(parameter.getTable()));
        if (null == parameter.getFilters() || parameter.getFilters().isEmpty()) {
            parameter.setFilters(Lists.<Filter>newArrayList(new AnyFilter(parameter.getTable())));
        }

        // Null field implies complete doc count
        if (parameter.getField() != null) {
            parameter.getFilters().add(new ExistsFilter(parameter.getField()));
        }

        try {
            if (parameter.isDistinct()){
                SearchRequestBuilder query = getConnection().getClient()
                        .prepareSearch(ElasticsearchUtils.getIndices(parameter.getTable(), parameter))
                        .setSearchType(SearchType.COUNT)
                        .setQuery(new ElasticSearchQueryGenerator(FilterCombinerType.and)
                                .genFilter(parameter.getFilters()))
                        .addAggregation(AggregationBuilders
                                .cardinality(Utils.sanitizeFieldForAggregation(parameter.getField()))
                                .field(parameter.getField())
                        );
                SearchResponse response = query.execute().actionGet();
                Aggregations aggregations = response.getAggregations();
                Cardinality cardinality = aggregations.get(Utils.sanitizeFieldForAggregation(parameter.getField()));
                if (cardinality == null){
                    return new CountResponse(0);
                } else {
                    return new CountResponse(cardinality.getValue());
                }
            } else {
                CountRequestBuilder countRequestBuilder = getConnection().getClient()
                        .prepareCount(ElasticsearchUtils.getIndices(parameter.getTable(), parameter))
                        .setQuery(new ElasticSearchQueryGenerator(FilterCombinerType.and).genFilter(parameter.getFilters()));
                org.elasticsearch.action.count.CountResponse countResponse = countRequestBuilder.execute().actionGet();
                return new CountResponse(countResponse.getCount());
            }
        } catch (Exception e) {
            logger.error("Error running count: ", e);
            throw new QueryStoreException(QueryStoreException.ErrorCode.QUERY_EXECUTION_ERROR,
                    "Error running count query.", e);
        }
    }

    public  CountAction(CountRequest parameter,
                       TableMetadataManager tableMetadataManager,
                       DataStore dataStore,
                       QueryStore queryStore,
                       ElasticsearchConnection connection,
                       String cacheToken);

}