Fix bug in BucketMetrics path traversal

When processing a top-level sibling pipeline, we destructively sublist
the path by assigning back onto the same variable.  But if aggs are
specified such:

A. Multi-bucket agg in the first entry of our internal list
B. Regular agg as the immediate child of the multi-bucket in A
C. Regular agg with the same name as B at the top level, listed as the
   second entry in our internal list
D. Finally, a pipeline agg with the path down to B

We'll get class cast exception.  The first agg will sublist the path
from [A,B] to [B], and then when we loop around to check agg C,
the sublisted path [B] matches the name of C and it fails.

The fix is simple: we just need to store the sublist in a new object
so that the old path remains valid for the rest of the aggs in the loop

Closes elastic#30608

server/src/main/java/org/elasticsearch/search/aggregations/pipeline/bucketmetrics/BucketMetricsPipelineAggregator.java

@@ -79,11 +79,11 @@ public final InternalAggregation doReduce(Aggregations aggregations, ReduceConte
         List<String> bucketsPath = AggregationPath.parse(bucketsPaths()[0]).getPathElementsAsStringList();
         for (Aggregation aggregation : aggregations) {
             if (aggregation.getName().equals(bucketsPath.get(0))) {
-                bucketsPath = bucketsPath.subList(1, bucketsPath.size());
+                List<String> sublistedPath = bucketsPath.subList(1, bucketsPath.size());
                 InternalMultiBucketAggregation<?, ?> multiBucketsAgg = (InternalMultiBucketAggregation<?, ?>) aggregation;
                 List<? extends InternalMultiBucketAggregation.InternalBucket> buckets = multiBucketsAgg.getBuckets();
                 for (InternalMultiBucketAggregation.InternalBucket bucket : buckets) {
-                    Double bucketValue = BucketHelpers.resolveBucketValue(multiBucketsAgg, bucket, bucketsPath, gapPolicy);
+                    Double bucketValue = BucketHelpers.resolveBucketValue(multiBucketsAgg, bucket, sublistedPath, gapPolicy);
                     if (bucketValue != null && !Double.isNaN(bucketValue)) {
                         collectBucketValue(bucket.getKeyAsString(), bucketValue);
                     }

server/src/test/java/org/elasticsearch/search/aggregations/pipeline/bucketmetrics/avg/AvgBucketAggregatorTests.java

@@ -0,0 +1,144 @@
+/*
+ * Licensed to Elasticsearch under one or more contributor
+ * license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright
+ * ownership. Elasticsearch licenses this file to you under
+ * the Apache License, Version 2.0 (the "License"); you may
+ * not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.search.aggregations.pipeline.bucketmetrics.avg;
+
+import org.apache.lucene.document.Document;
+import org.apache.lucene.document.SortedNumericDocValuesField;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.IndexReader;
+import org.apache.lucene.index.RandomIndexWriter;
+import org.apache.lucene.search.IndexSearcher;
+import org.apache.lucene.search.MatchAllDocsQuery;
+import org.apache.lucene.search.Query;
+import org.apache.lucene.store.Directory;
+import org.elasticsearch.index.mapper.DateFieldMapper;
+import org.elasticsearch.index.mapper.MappedFieldType;
+import org.elasticsearch.index.mapper.NumberFieldMapper;
+import org.elasticsearch.search.aggregations.Aggregation;
+import org.elasticsearch.search.aggregations.Aggregations;
+import org.elasticsearch.search.aggregations.AggregatorTestCase;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramAggregationBuilder;
+import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramInterval;
+import org.elasticsearch.search.aggregations.bucket.histogram.InternalDateHistogram;
+import org.elasticsearch.search.aggregations.metrics.avg.AvgAggregationBuilder;
+import org.elasticsearch.search.aggregations.metrics.avg.InternalAvg;
+import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+
+public class AvgBucketAggregatorTests extends AggregatorTestCase {
+    private static final String DATE_FIELD = "date";
+    private static final String VALUE_FIELD = "value";
+
+    private static final List<String> dataset = Arrays.asList(
+        "2010-03-12T01:07:45",
+        "2010-04-27T03:43:34",
+        "2012-05-18T04:11:00",
+        "2013-05-29T05:11:31",
+        "2013-10-31T08:24:05",
+        "2015-02-13T13:09:32",
+        "2015-06-24T13:47:43",
+        "2015-11-13T16:14:34",
+        "2016-03-04T17:09:50",
+        "2017-12-12T22:55:46");
+
+    /**
+     * Test for issue #30608.  Under the following circumstances:
+     *
+     * A. Multi-bucket agg in the first entry of our internal list
+     * B. Regular agg as the immediate child of the multi-bucket in A
+     * C. Regular agg with the same name as B at the top level, listed as the second entry in our internal list
+     * D. Finally, a pipeline agg with the path down to B
+     *
+     * BucketMetrics reduction would throw a class cast exception due to bad subpathing.  This test ensures
+     * it is fixed.
+     *
+     * Note: we have this test inside of the `avg_bucket` package so that we can get access to the package-private
+     * `doReduce()` needed for testing this
+     */
+    public void testSameAggNames() throws IOException {
+        Query query = new MatchAllDocsQuery();
+
+        AvgAggregationBuilder avgBuilder = new AvgAggregationBuilder("foo").field(VALUE_FIELD);
+        DateHistogramAggregationBuilder histo = new DateHistogramAggregationBuilder("histo")
+            .dateHistogramInterval(DateHistogramInterval.YEAR)
+            .field(DATE_FIELD)
+            .subAggregation(new AvgAggregationBuilder("foo").field(VALUE_FIELD));
+
+        AvgBucketPipelineAggregationBuilder avgBucketBuilder
+            = new AvgBucketPipelineAggregationBuilder("the_avg_bucket", "histo>foo");
+
+        try (Directory directory = newDirectory()) {
+            try (RandomIndexWriter indexWriter = new RandomIndexWriter(random(), directory)) {
+                Document document = new Document();
+                for (String date : dataset) {
+                    if (frequently()) {
+                        indexWriter.commit();
+                    }
+
+                    document.add(new SortedNumericDocValuesField(DATE_FIELD, asLong(date)));
+                    document.add(new SortedNumericDocValuesField(VALUE_FIELD, randomInt()));
+                    indexWriter.addDocument(document);
+                    document.clear();
+                }
+            }
+
+            InternalAvg avgResult;
+            InternalDateHistogram histogramResult;
+            try (IndexReader indexReader = DirectoryReader.open(directory)) {
+                IndexSearcher indexSearcher = newSearcher(indexReader, true, true);
+
+                DateFieldMapper.Builder builder = new DateFieldMapper.Builder("histo");
+                DateFieldMapper.DateFieldType fieldType = builder.fieldType();
+                fieldType.setHasDocValues(true);
+                fieldType.setName(DATE_FIELD);
+
+                MappedFieldType valueFieldType = new NumberFieldMapper.NumberFieldType(NumberFieldMapper.NumberType.LONG);
+                valueFieldType.setName(VALUE_FIELD);
+                valueFieldType.setHasDocValues(true);
+
+                avgResult = searchAndReduce(indexSearcher, query, avgBuilder, 10000, new MappedFieldType[]{fieldType, valueFieldType});
+                histogramResult = searchAndReduce(indexSearcher, query, histo, 10000,  new MappedFieldType[]{fieldType, valueFieldType});
+            }
+
+            // Finally, reduce the pipeline agg
+            PipelineAggregator avgBucketAgg = avgBucketBuilder.createInternal(Collections.emptyMap());
+            List<Aggregation> reducedAggs = new ArrayList<>(2);
+
+            // Histo has to go first to exercise the bug
+            reducedAggs.add(histogramResult);
+            reducedAggs.add(avgResult);
+            Aggregations aggregations = new Aggregations(reducedAggs);
+            InternalAggregation pipelineResult = ((AvgBucketPipelineAggregator)avgBucketAgg).doReduce(aggregations, null);
+            assertNotNull(pipelineResult);
+        }
+    }
+
+
+    private static long asLong(String dateTime) {
+        return DateFieldMapper.DEFAULT_DATE_TIME_FORMATTER.parser().parseDateTime(dateTime).getMillis();
+    }
+}