Fix docCountError calculation for multiple reduces (#76391)

Fix docCountError calculation in case of multiple reduces. It fixes 2 mistakes
in #43874. The first error was introduced in the original PR, where unknown doc
count errors were initialized equal to 0, the second was introduced during in
order to fix the first one by ignoring these 0s, which essentially disabled the
original fix.

Fixes #75667

Author: imotov

benchmarks/src/main/java/org/elasticsearch/benchmark/search/aggregations/TermsReduceBenchmark.java

@@ -154,7 +154,7 @@ private StringTerms newTerms(Random rand, BytesRef[] dict, boolean withNested) {
                 true,
                 0,
                 buckets,
-                0L
+                null
             );
         }
 

benchmarks/src/main/java/org/elasticsearch/benchmark/search/aggregations/bucket/terms/StringTermsSerializationBenchmark.java

@@ -70,7 +70,7 @@ private StringTerms newTerms(boolean withNested) {
             false,
             100000,
             resultBuckets,
-            0L
+            null
         );
     }
 

server/src/internalClusterTest/java/org/elasticsearch/search/aggregations/bucket/TermsDocCountErrorIT.java

@@ -1034,7 +1034,6 @@ public void testFixedDocs() throws Exception {
      * Tests the upper bounds are correct when performing incremental reductions
      * See https://github.com/elastic/elasticsearch/issues/40005 for more details
      */
-    @AwaitsFix(bugUrl = "https://github.com/elastic/elasticsearch/issues/75667")
     public void testIncrementalReduction() {
         SearchResponse response = client().prepareSearch("idx_fixed_docs_3", "idx_fixed_docs_4", "idx_fixed_docs_5")
             .addAggregation(terms("terms")

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

@@ -133,7 +133,7 @@ private long getDocCountError(A terms) {
         if (size == 0 || size < terms.getShardSize() || isKeyOrder(terms.getOrder())) {
             return 0;
         } else if (InternalOrder.isCountDesc(terms.getOrder())) {
-            if (terms.getDocCountError() != null && terms.getDocCountError() > 0) {
+            if (terms.getDocCountError() != null) {
                 // If there is an existing docCountError for this agg then
                 // use this as the error for this aggregation
                 return terms.getDocCountError();
@@ -340,7 +340,7 @@ public InternalAggregation reduce(List<InternalAggregation> aggregations, Intern
 
     protected static XContentBuilder doXContentCommon(XContentBuilder builder,
                                                       Params params,
-                                                      long docCountError,
+                                                      Long docCountError,
                                                       long otherDocCount,
                                                       List<? extends AbstractTermsBucket> buckets) throws IOException {
         builder.field(DOC_COUNT_ERROR_UPPER_BOUND_FIELD_NAME.getPreferredName(), docCountError);

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

@@ -765,7 +765,7 @@ StringTerms buildResult(long owningBucketOrd, long otherDocCount, StringTerms.Bu
             }
             return new StringTerms(name, reduceOrder, order, bucketCountThresholds.getRequiredSize(),
                 bucketCountThresholds.getMinDocCount(), metadata(), format, bucketCountThresholds.getShardSize(), showTermDocCountError,
-                otherDocCount, Arrays.asList(topBuckets), 0L);
+                otherDocCount, Arrays.asList(topBuckets), null);
         }
 
         @Override

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

@@ -53,12 +53,13 @@ protected InternalMappedTerms(String name, BucketOrder reduceOrder, BucketOrder
     protected InternalMappedTerms(StreamInput in, Bucket.Reader<B> bucketReader) throws IOException {
         super(in);
         if (in.getVersion().onOrAfter(Version.V_8_0_0)) { // todo fix after backport
-            docCountError = in.readOptionalLong();
-        } else {
-            docCountError = in.readZLong();
-            if (docCountError == 0) {
+            if (in.readBoolean()) {
+                docCountError = in.readZLong();
+            } else {
                 docCountError = null;
             }
+        } else {
+            docCountError = in.readZLong();
         }
         format = in.readNamedWriteable(DocValueFormat.class);
         shardSize = readSize(in);
@@ -70,7 +71,12 @@ protected InternalMappedTerms(StreamInput in, Bucket.Reader<B> bucketReader) thr
     @Override
     protected final void writeTermTypeInfoTo(StreamOutput out) throws IOException {
         if (out.getVersion().onOrAfter(Version.V_8_0_0)) {  // todo fix after backport
-            out.writeOptionalLong(docCountError);
+            if (docCountError != null) {
+                out.writeBoolean(true);
+                out.writeZLong(docCountError);
+            } else {
+                out.writeBoolean(false);
+            }
         } else {
             out.writeZLong(docCountError == null ? 0 : docCountError);
         }

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

@@ -446,7 +446,7 @@ StringTerms buildResult(long owningBucketOrd, long otherDocCount, StringTerms.Bu
             }
             return new StringTerms(name, reduceOrder, order, bucketCountThresholds.getRequiredSize(),
                 bucketCountThresholds.getMinDocCount(), metadata(), format, bucketCountThresholds.getShardSize(), showTermDocCountError,
-                otherDocCount, Arrays.asList(topBuckets), 0L);
+                otherDocCount, Arrays.asList(topBuckets), null);
         }
 
         @Override

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

@@ -372,7 +372,7 @@ LongTerms buildResult(long owningBucketOrd, long otherDocCount, LongTerms.Bucket
                 showTermDocCountError,
                 otherDocCount,
                 List.of(topBuckets),
-                0L
+                null
             );
         }
 
@@ -454,7 +454,7 @@ DoubleTerms buildResult(long owningBucketOrd, long otherDocCount, DoubleTerms.Bu
                 showTermDocCountError,
                 otherDocCount,
                 List.of(topBuckets),
-                0L
+                null
             );
         }
 

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

@@ -227,7 +227,7 @@ protected boolean lessThan(OrdBucket a, OrdBucket b) {
             showTermDocCountError,
             otherDocsCount,
             buckets,
-            0L
+            null
         );
     }
 

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

@@ -97,7 +97,7 @@ public boolean isMapped() {
 
     @Override
     public final XContentBuilder doXContentBody(XContentBuilder builder, Params params) throws IOException {
-        return doXContentCommon(builder, params, 0, 0, Collections.emptyList());
+        return doXContentCommon(builder, params, 0L, 0, Collections.emptyList());
     }
 
     @Override