Rewrite Queries/Filter in FilterAggregationBuilder and ensure client usage marks query as non-cachable (#21303)

`FilterAggregationBuilder` today misses to rewrite queries which causes failures
if a query that uses a client for instance to lookup terms since it must be rewritten first.
This change also ensures that if a client is used from the rewrite context we mark the query as
non-cacheable.

Closes #21301

core/src/main/java/org/elasticsearch/index/query/QueryRewriteContext.java

@@ -67,7 +67,7 @@ public class QueryRewriteContext implements ParseFieldMatcherSupplier {
     /**
      * Returns a clients to fetch resources from local or remove nodes.
      */
-    public final Client getClient() {
+    public Client getClient() {
         return client;
     }
 

core/src/main/java/org/elasticsearch/index/query/QueryShardContext.java

@@ -421,4 +421,9 @@ public class QueryShardContext extends QueryRewriteContext {
         return super.nowInMillis();
     }
 
+    @Override
+    public Client getClient() {
+        failIfFrozen(); // we somebody uses a terms filter with lookup for instance can't be cached...
+        return super.getClient();
+    }
 }

core/src/main/java/org/elasticsearch/search/aggregations/bucket/filter/FilterAggregationBuilder.java

@@ -72,7 +72,9 @@ public class FilterAggregationBuilder extends AbstractAggregationBuilder<FilterA
     @Override
     protected AggregatorFactory<?> doBuild(AggregationContext context, AggregatorFactory<?> parent,
             AggregatorFactories.Builder subFactoriesBuilder) throws IOException {
-        return new FilterAggregatorFactory(name, type, filter, context, parent, subFactoriesBuilder, metaData);
+        // TODO this sucks we need a rewrite phase for aggregations too
+        final QueryBuilder rewrittenFilter = QueryBuilder.rewriteQuery(filter, context.searchContext().getQueryShardContext());
+        return new FilterAggregatorFactory(name, type, rewrittenFilter, context, parent, subFactoriesBuilder, metaData);
     }
 
     @Override

core/src/test/java/org/elasticsearch/index/query/MoreLikeThisQueryBuilderTests.java

@@ -301,6 +301,11 @@ public class MoreLikeThisQueryBuilderTests extends AbstractQueryTestCase<MoreLik
         assertEquals(expectedItem, newItem);
     }
 
+    @Override
+    protected boolean isCachable(MoreLikeThisQueryBuilder queryBuilder) {
+        return queryBuilder.likeItems().length == 0; // items are always fetched
+    }
+
     public void testFromJson() throws IOException {
         String json =
                 "{\n" +

core/src/test/java/org/elasticsearch/index/query/TermsQueryBuilderTests.java

@@ -289,5 +289,12 @@ public class TermsQueryBuilderTests extends AbstractQueryTestCase<TermsQueryBuil
         assertEquals("Geo fields do not support exact searching, use dedicated geo queries instead: [mapped_geo_point]",
                 e.getMessage());
     }
+
+    @Override
+    protected boolean isCachable(TermsQueryBuilder queryBuilder) {
+        // even though we use a terms lookup here we do this during rewrite and that means we are cachable on toQuery
+        // that's why we return true here all the time
+        return super.isCachable(queryBuilder);
+    }
 }
 

core/src/test/java/org/elasticsearch/indices/IndicesRequestCacheIT.java

@@ -362,7 +362,7 @@ public class IndicesRequestCacheIT extends ESIntegTestCase {
     public void testCanCache() throws Exception {
         assertAcked(client().admin().indices().prepareCreate("index").addMapping("type", "s", "type=date")
                 .setSettings(IndicesRequestCache.INDEX_CACHE_REQUEST_ENABLED_SETTING.getKey(), true, IndexMetaData.SETTING_NUMBER_OF_SHARDS,
-                        5, IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
+                        2, IndexMetaData.SETTING_NUMBER_OF_REPLICAS, 0)
                 .get());
         indexRandom(true, client().prepareIndex("index", "type", "1").setRouting("1").setSource("s", "2016-03-19"),
                 client().prepareIndex("index", "type", "2").setRouting("1").setSource("s", "2016-03-20"),
@@ -411,7 +411,7 @@ public class IndicesRequestCacheIT extends ESIntegTestCase {
         assertThat(client().admin().indices().prepareStats("index").setRequestCache(true).get().getTotal().getRequestCache().getMissCount(),
                 equalTo(0L));
 
-        // If the request has an aggregation containng now we should not cache
+        // If the request has an aggregation containing now we should not cache
         final SearchResponse r4 = client().prepareSearch("index").setSearchType(SearchType.QUERY_THEN_FETCH).setSize(0)
                 .setRequestCache(true).setQuery(QueryBuilders.rangeQuery("s").gte("2016-03-20").lte("2016-03-26"))
                 .addAggregation(filter("foo", QueryBuilders.rangeQuery("s").from("now-10y").to("now"))).get();
@@ -441,7 +441,7 @@ public class IndicesRequestCacheIT extends ESIntegTestCase {
         assertThat(client().admin().indices().prepareStats("index").setRequestCache(true).get().getTotal().getRequestCache().getHitCount(),
                 equalTo(0L));
         assertThat(client().admin().indices().prepareStats("index").setRequestCache(true).get().getTotal().getRequestCache().getMissCount(),
-                equalTo(5L));
+                equalTo(2L));
     }
 
     public void testCacheWithFilteredAlias() {

rest-api-spec/src/main/resources/rest-api-spec/test/search.aggregation/50_filter.yaml

@@ -0,0 +1,80 @@
+setup:
+  - do:
+      indices.create:
+          index: test
+          body:
+            settings:
+              number_of_shards: 1
+              number_of_replicas: 0
+            mappings:
+              post:
+                properties:
+                  mentions:
+                    type: keyword
+              user:
+                properties:
+                  notifications:
+                    type: keyword
+
+  - do:
+        index:
+          index: test
+          type: test
+          id: foo|bar|baz0
+          body: { "notifications" : ["abc"] }
+
+  - do:
+      index:
+        index: test
+        type: test
+        id: foo|bar|baz1
+        body: { "mentions" : ["abc"] }
+
+  - do:
+      indices.refresh: {}
+
+---
+"Filter aggs with terms lookup ensure not cached":
+  - skip:
+    version: " - 5.0.0"
+    reason:  This using filter aggs that needs rewriting, this was fixed in 5.0.1
+
+  - do:
+      search:
+        size: 0
+        request_cache: true
+        body: {"aggs": { "itemsNotify": { "filter": { "terms": { "mentions": { "index": "test", "type": "test", "id": "foo|bar|baz0", "path": "notifications"}}}, "aggs": { "mentions" : {"terms" : { "field" : "mentions" }}}}}}
+
+  # validate result
+  - match: { hits.total: 2 }
+  - match: { aggregations.itemsNotify.doc_count: 1 }
+  - length: { aggregations.itemsNotify.mentions.buckets: 1 }
+  - match: { aggregations.itemsNotify.mentions.buckets.0.key: "abc" }
+  # we are using a lookup - this should not cache
+  - do:
+      indices.stats: { index: test, metric: request_cache}
+  - match: { _shards.total: 1 }
+  - match: { _all.total.request_cache.hit_count: 0 }
+  - match: { _all.total.request_cache.miss_count: 0 }
+  - is_true: indices.test
+
+---
+"Filter aggs no lookup and ensure it's cached":
+  # now run without lookup and ensure we get cached or at least do the lookup
+  - do:
+      search:
+        size: 0
+        request_cache: true
+        body: {"aggs": { "itemsNotify": { "filter": { "terms": { "mentions": ["abc"]}}, "aggs": { "mentions" : {"terms" : { "field" : "mentions" }}}}}}
+
+  - match: { hits.total: 2 }
+  - match: { aggregations.itemsNotify.doc_count: 1 }
+  - length: { aggregations.itemsNotify.mentions.buckets: 1 }
+  - match: { aggregations.itemsNotify.mentions.buckets.0.key: "abc" }
+  - do:
+      indices.stats: { index: test, metric: request_cache}
+  - match: { _shards.total: 1 }
+  - match: { _all.total.request_cache.hit_count: 0 }
+  - match: { _all.total.request_cache.miss_count: 1 }
+  - is_true: indices.test
+