Fix lookup support in adjacency matrix (#59099)

This request:
```
POST /_search
{
  "aggs": {
    "a": {
      "adjacency_matrix": {
        "filters": {
          "1": {
            "terms": { "t": { "index": "lookup", "id": "1", "path": "t" } }
          }
        }
      }
    }
  }
}
```

Would fail with a 500 error and a message like:
```
{
  "error": {
    "root_cause": [
      {
        "type": "illegal_state_exception",
        "reason":"async actions are left after rewrite"
      }
    ]
  }
}
```

This fixes that by moving the query rewrite phase from a synchronous
call on the data nodes into the standard aggregation rewrite phase which
can properly handle the asynchronous actions.

docs/reference/aggregations/bucket/adjacency-matrix-aggregation.asciidoc

@@ -110,5 +110,4 @@ where examining interactions _over time_ becomes important.
 ==== Limitations
 For N filters the matrix of buckets produced can be N²/2 which can be costly.
 The circuit breaker settings prevent results producing too many buckets and to avoid excessive disk seeks
-the `indices.query.bool.max_clause_count` setting is used to limit the number of filters. 
-imposed of 100 filters . This setting can be changed using the `index.max_adjacency_matrix_filters` index-level setting.
+the `indices.query.bool.max_clause_count` setting is used to limit the number of filters.

rest-api-spec/src/main/resources/rest-api-spec/test/search.aggregation/70_adjacency_matrix.yml

@@ -12,36 +12,104 @@ setup:
                   type: integer
 
   - do:
-        index:
-          index: test
-          id: 1
-          body: { "num": [1, 2] }
+      bulk:
+        index: test
+        refresh: true
+        body: |
+          { "index": {"_id": "1"}}
+          { "num": [1, 2] }
+          { "index": {"_id": "2"}}
+          { "num": [2, 3] }
+          { "index": {"_id": "3"}}
+          { "num": [3, 4] }
 
   - do:
-      index:
-        index: test
-        id: 2
-        body: { "num": [2, 3] }
+      indices.refresh: {}
+---
+"Filters intersections":
 
   - do:
-      index:
+      search:
         index: test
-        id: 3
-        body: { "num": [3, 4] }
+        body:
+          size: 0
+          aggs:
+            conns:
+              adjacency_matrix:
+                filters:
+                  1:
+                    term:
+                      num: 1
+                  2:
+                    term:
+                      num: 2
+                  4:
+                    term:
+                      num: 4
 
-  - do:
-      indices.refresh: {}
+  - match: { hits.total.value: 3 }
+
+  - length: { aggregations.conns.buckets: 4 }
+
+  - match: { aggregations.conns.buckets.0.doc_count: 1 }
+  - match: { aggregations.conns.buckets.0.key: "1" }
+
+  - match: { aggregations.conns.buckets.1.doc_count: 1 }
+  - match: { aggregations.conns.buckets.1.key: "1&2" }
+
+  - match: { aggregations.conns.buckets.2.doc_count: 2 }
+  - match: { aggregations.conns.buckets.2.key: "2" }
+
+  - match: { aggregations.conns.buckets.3.doc_count: 1 }
+  - match: { aggregations.conns.buckets.3.key: "4" }
 
 
 ---
-"Filters intersections":
+"Terms lookup":
+  - skip:
+      version: " - 7.99.99"
+      reason:  fixed in 8.0.0 (backporting to 7.9.0)
 
+  - do:
+      bulk:
+        index: lookup
+        refresh: true
+        body: |
+          { "index": {"_id": 1} }
+          { "num": [1] }
+          { "index": {"_id": 2} }
+          { "num": [2] }
+          { "index": {"_id": 4} }
+          { "num": [4] }
   - do:
       search:
-        rest_total_hits_as_int: true
-        body: { "size": 0, "aggs": { "conns": { "adjacency_matrix": {  "filters": { "1": { "term": { "num": 1 } }, "2": { "term": { "num": 2 } }, "4": { "term": { "num": 4 } } } } } } }
+        index: test
+        body:
+          size: 0
+          aggs:
+            conns:
+              adjacency_matrix:
+                filters:
+                  1:
+                    terms:
+                      num:
+                        index: lookup
+                        id: "1"
+                        path: num
+                  2:
+                    terms:
+                      num:
+                        index: lookup
+                        id: "2"
+                        path: num
+                  4:
+                    terms:
+                      num:
+                        index: lookup
+                        id: "4"
+                        path: num
 
-  - match: { hits.total: 3 }
+  - match: { hits.total.value: 3 }
 
   - length: { aggregations.conns.buckets: 4 }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/adjacency/AdjacencyMatrixAggregationBuilder.java

@@ -27,6 +27,7 @@ import org.elasticsearch.common.xcontent.ObjectParser;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentParser;
 import org.elasticsearch.index.query.QueryBuilder;
+import org.elasticsearch.index.query.QueryRewriteContext;
 import org.elasticsearch.index.query.QueryShardContext;
 import org.elasticsearch.index.query.Rewriteable;
 import org.elasticsearch.search.SearchModule;
@@ -69,31 +70,6 @@ public class AdjacencyMatrixAggregationBuilder extends AbstractAggregationBuilde
         return result;
     }
 
-    protected void checkConsistency() {
-        if ((filters == null) || (filters.size() == 0)) {
-            throw new IllegalStateException("[" + name  + "] is missing : " + FILTERS_FIELD.getPreferredName() + " parameter");
-        }
-    }
-
-    protected void setFiltersAsMap(Map<String, QueryBuilder> filters) {
-        // Convert uniquely named objects into internal KeyedFilters
-        this.filters = new ArrayList<>(filters.size());
-        for (Entry<String, QueryBuilder> kv : filters.entrySet()) {
-            this.filters.add(new KeyedFilter(kv.getKey(), kv.getValue()));
-        }
-        // internally we want to have a fixed order of filters, regardless of
-        // the order of the filters in the request
-        Collections.sort(this.filters, Comparator.comparing(KeyedFilter::key));
-    }
-
-    protected void setFiltersAsList(List<KeyedFilter> filters) {
-        this.filters = new ArrayList<>(filters);
-        // internally we want to have a fixed order of filters, regardless of
-        // the order of the filters in the request
-        Collections.sort(this.filters, Comparator.comparing(KeyedFilter::key));
-    }
-
-
     /**
      * @param name
      *            the name of this aggregation
@@ -163,6 +139,33 @@ public class AdjacencyMatrixAggregationBuilder extends AbstractAggregationBuilde
         }
     }
 
+    private void checkConsistency() {
+        if ((filters == null) || (filters.size() == 0)) {
+            throw new IllegalStateException("[" + name  + "] is missing : " + FILTERS_FIELD.getPreferredName() + " parameter");
+        }
+    }
+
+    private void setFiltersAsMap(Map<String, QueryBuilder> filters) {
+        // Convert uniquely named objects into internal KeyedFilters
+        this.filters = new ArrayList<>(filters.size());
+        for (Entry<String, QueryBuilder> kv : filters.entrySet()) {
+            this.filters.add(new KeyedFilter(kv.getKey(), kv.getValue()));
+        }
+        // internally we want to have a fixed order of filters, regardless of
+        // the order of the filters in the request
+        Collections.sort(this.filters, Comparator.comparing(KeyedFilter::key));
+    }
+
+    private AdjacencyMatrixAggregationBuilder setFiltersAsList(List<KeyedFilter> filters) {
+        this.filters = new ArrayList<>(filters);
+        // internally we want to have a fixed order of filters, regardless of
+        // the order of the filters in the request
+        Collections.sort(this.filters, Comparator.comparing(KeyedFilter::key));
+        return this;
+    }
+
+
+
     /**
      * Set the separator used to join pairs of bucket keys
      */
@@ -192,6 +195,20 @@ public class AdjacencyMatrixAggregationBuilder extends AbstractAggregationBuilde
         return result;
     }
 
+    @Override
+    protected AdjacencyMatrixAggregationBuilder doRewrite(QueryRewriteContext queryShardContext) throws IOException {
+        boolean modified = false;
+        List<KeyedFilter> rewrittenFilters = new ArrayList<>(filters.size());
+        for (KeyedFilter kf : filters) {
+            QueryBuilder rewritten = Rewriteable.rewrite(kf.filter(), queryShardContext);
+            modified = modified || rewritten != kf.filter();
+            rewrittenFilters.add(new KeyedFilter(kf.key(), rewritten));
+        }
+        if (modified) {
+            return new AdjacencyMatrixAggregationBuilder(name).separator(separator).setFiltersAsList(rewrittenFilters);
+        }
+        return this;
+    }
 
     @Override
     protected AggregatorFactory doBuild(QueryShardContext queryShardContext, AggregatorFactory parent, Builder subFactoriesBuilder)
@@ -204,13 +221,7 @@ public class AdjacencyMatrixAggregationBuilder extends AbstractAggregationBuilde
                             + "This limit can be set by changing the [" + SearchModule.INDICES_MAX_CLAUSE_COUNT_SETTING.getKey()
                             + "] setting.");
         }
-
-        List<KeyedFilter> rewrittenFilters = new ArrayList<>(filters.size());
-        for (KeyedFilter kf : filters) {
-            rewrittenFilters.add(new KeyedFilter(kf.key(), Rewriteable.rewrite(kf.filter(), queryShardContext, true)));
-        }
-
-        return new AdjacencyMatrixAggregatorFactory(name, rewrittenFilters, separator, queryShardContext, parent,
+        return new AdjacencyMatrixAggregatorFactory(name, filters, separator, queryShardContext, parent,
                 subFactoriesBuilder, metadata);
     }
 

server/src/main/java/org/elasticsearch/search/aggregations/bucket/adjacency/AdjacencyMatrixAggregator.java

@@ -124,7 +124,7 @@ public class AdjacencyMatrixAggregator extends BucketsAggregator {
     }
 
     private final String[] keys;
-    private Weight[] filters;
+    private final Weight[] filters;
     private final int totalNumKeys;
     private final int totalNumIntersections;
     private final String separator;

server/src/main/java/org/elasticsearch/search/aggregations/bucket/adjacency/AdjacencyMatrixAggregatorFactory.java

@@ -51,9 +51,9 @@ public class AdjacencyMatrixAggregatorFactory extends AggregatorFactory {
         keys = new String[filters.size()];
         for (int i = 0; i < filters.size(); ++i) {
             KeyedFilter keyedFilter = filters.get(i);
-            this.keys[i] = keyedFilter.key();
+            keys[i] = keyedFilter.key();
             Query filter = keyedFilter.filter().toQuery(queryShardContext);
-            this.weights[i] = contextSearcher.createWeight(contextSearcher.rewrite(filter), ScoreMode.COMPLETE_NO_SCORES, 1f);
+            weights[i] = contextSearcher.createWeight(contextSearcher.rewrite(filter), ScoreMode.COMPLETE_NO_SCORES, 1f);
         }
     }