Do not reuse the SetBackedScalingCuckooFilter during merging (#75111)

Fixes an IAE that is thrown when the underlaying circuit breaker is closed.

Author: iverase

server/src/internalClusterTest/java/org/elasticsearch/search/aggregations/bucket/terms/RareTermsIT.java

@@ -0,0 +1,66 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0 and the Server Side Public License, v 1; you may not use this file except
+ * in compliance with, at your election, the Elastic License 2.0 or the Server
+ * Side Public License, v 1.
+ */
+
+package org.elasticsearch.search.aggregations.bucket.terms;
+
+import org.elasticsearch.action.bulk.BulkRequestBuilder;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.search.SearchRequestBuilder;
+import org.elasticsearch.action.search.SearchResponse;
+import org.elasticsearch.cluster.metadata.IndexMetadata;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.common.xcontent.XContentType;
+import org.elasticsearch.test.ESSingleNodeTestCase;
+import org.hamcrest.Matchers;
+
+import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertNoFailures;
+
+/**
+ * Test that index enough data to trigger the creation of Cuckoo filters.
+ */
+
+public class RareTermsIT extends ESSingleNodeTestCase {
+
+    private static final String index = "idx";
+
+    private void indexDocs(int numDocs) {
+        final BulkRequestBuilder bulk = client().prepareBulk();
+        for (int i = 0; i < numDocs; ++i) {
+            bulk.add(new IndexRequest(index).source("{\"str_value\" : \"s" + i + "\"}", XContentType.JSON));
+        }
+        assertNoFailures(bulk.get());
+    }
+
+    public void testSingleValuedString() {
+        final Settings.Builder settings = Settings.builder()
+            .put(IndexMetadata.INDEX_NUMBER_OF_SHARDS_SETTING.getKey(), 2)
+            .put(IndexMetadata.INDEX_NUMBER_OF_REPLICAS_SETTING.getKey(), 0);
+        createIndex(index, settings.build());
+        // We want to trigger the usage of cuckoo filters that happen only when there are
+        // more than 10k distinct values in one shard.
+        final int numDocs = randomIntBetween(12000, 17000);
+        // Index every value 3 times
+        for (int i = 0; i < 3; i++) {
+            indexDocs(numDocs);
+            assertNoFailures(client().admin().indices().prepareRefresh(index).get());
+        }
+        // There are no rare terms that only appear in one document
+        assertNumRareTerms(1, 0);
+        // All terms have a cardinality lower than 10
+        assertNumRareTerms(10, numDocs);
+    }
+
+    private void assertNumRareTerms(int maxDocs, int rareTerms)  {
+        final SearchRequestBuilder requestBuilder = client().prepareSearch(index);
+        requestBuilder.addAggregation(new RareTermsAggregationBuilder("rareTerms").field("str_value.keyword").maxDocCount(maxDocs));
+        final SearchResponse response = requestBuilder.get();
+        assertNoFailures(response);
+        final RareTerms terms = response.getAggregations().get("rareTerms");
+        assertThat(terms.getBuckets().size(), Matchers.equalTo(rareTerms));
+    }
+}

server/src/main/java/org/elasticsearch/common/util/SetBackedScalingCuckooFilter.java

@@ -103,23 +103,6 @@ public SetBackedScalingCuckooFilter(int threshold, Random rng, double fpp) {
         this.fpp = fpp;
     }
 
-    public SetBackedScalingCuckooFilter(SetBackedScalingCuckooFilter other) {
-        this.threshold = other.threshold;
-        this.isSetMode = other.isSetMode;
-        this.rng = other.rng;
-        this.breaker = other.breaker;
-        this.capacity = other.capacity;
-        this.fpp = other.fpp;
-        if (isSetMode) {
-            this.hashes = new HashSet<>(other.hashes);
-        } else {
-            this.filters = new ArrayList<>(other.filters);
-            this.numBuckets = filters.get(0).getNumBuckets();
-            this.fingerprintMask = filters.get(0).getFingerprintMask();
-            this.bitsPerEntry = filters.get(0).getBitsPerEntry();
-        }
-    }
-
     public SetBackedScalingCuckooFilter(StreamInput in, Random rng) throws IOException {
         this.threshold = in.readVInt();
         this.isSetMode = in.readBoolean();
@@ -150,6 +133,27 @@ public void writeTo(StreamOutput out) throws IOException {
         }
     }
 
+    /**
+     * Returns the number of distinct values that are tracked before converting to an approximate representation.
+     * */
+    public int getThreshold() {
+        return threshold;
+    }
+
+    /**
+     * Returns the random number generator used for the cuckoo hashing process.
+     * */
+    public Random getRng() {
+        return rng;
+    }
+
+    /**
+     * Returns the false-positive rate used for the cuckoo filters.
+     * */
+    public double getFpp() {
+        return fpp;
+    }
+
     /**
      * Registers a circuit breaker with the datastructure.
      *

server/src/main/java/org/elasticsearch/search/aggregations/bucket/terms/InternalMappedRareTerms.java

@@ -107,10 +107,10 @@ public InternalAggregation reduce(List<InternalAggregation> aggregations, Reduce
 
             SetBackedScalingCuckooFilter otherFilter = ((InternalMappedRareTerms)aggregation).getFilter();
             if (filter == null) {
-                filter = new SetBackedScalingCuckooFilter(otherFilter);
-            } else {
-                filter.merge(otherFilter);
+                filter = new SetBackedScalingCuckooFilter(otherFilter.getThreshold(), otherFilter.getRng(), otherFilter.getFpp());
             }
+            filter.merge(otherFilter);
+
         }
 
         final List<B> rare = new ArrayList<>();

server/src/test/java/org/elasticsearch/common/util/SetBackedScalingCuckooFilterTests.java

@@ -18,6 +18,7 @@
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
+import static org.hamcrest.Matchers.in;
 import static org.hamcrest.Matchers.lessThanOrEqualTo;
 
 public class SetBackedScalingCuckooFilterTests extends AbstractWireSerializingTestCase<SetBackedScalingCuckooFilter> {
@@ -41,7 +42,9 @@ protected Writeable.Reader<SetBackedScalingCuckooFilter> instanceReader() {
 
     @Override
     protected SetBackedScalingCuckooFilter mutateInstance(SetBackedScalingCuckooFilter instance) throws IOException {
-        SetBackedScalingCuckooFilter newInstance = new SetBackedScalingCuckooFilter(instance);
+        SetBackedScalingCuckooFilter newInstance =
+            new SetBackedScalingCuckooFilter(instance.getThreshold(), instance.getRng(), instance.getFpp());
+        newInstance.merge(instance);
         int num = randomIntBetween(1, 10);
         for (int i = 0; i < num; i++) {
             newInstance.add(randomLong());