re-factor top-metrics to use a generic multi value output interface (#71903)

re-factor top-metrics to return generic(non-numeric) multi value
aggregation results

server/src/main/java/org/elasticsearch/search/aggregations/metrics/InternalMultiValueAggregation.java

@@ -0,0 +1,36 @@
+/*
+ * 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.metrics;
+
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.search.aggregations.InternalAggregation;
+import org.elasticsearch.search.aggregations.support.AggregationPath;
+
+import java.io.IOException;
+import java.util.Iterator;
+import java.util.Map;
+
+public abstract class InternalMultiValueAggregation extends InternalAggregation implements MultiValueAggregation {
+
+    protected InternalMultiValueAggregation(String name, Map<String, Object> metadata) {
+        super(name, metadata);
+    }
+
+    /**
+     * Read from a stream.
+     */
+    protected InternalMultiValueAggregation(StreamInput in) throws IOException {
+        super(in);
+    }
+
+    @Override
+    public final double sortValue(AggregationPath.PathElement head, Iterator<AggregationPath.PathElement> tail) {
+        throw new IllegalArgumentException("Metrics aggregations cannot have sub-aggregations (at [>" + head + "]");
+    }
+}

server/src/main/java/org/elasticsearch/search/aggregations/metrics/MultiValueAggregation.java

@@ -0,0 +1,34 @@
+/*
+ * 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.metrics;
+
+import org.elasticsearch.search.aggregations.Aggregation;
+
+import java.util.List;
+
+public interface MultiValueAggregation extends Aggregation {
+
+    /**
+     * Return an iterable over all value names this multi value aggregation provides.
+     *
+     * The iterable might be created on the fly, if you need to call this multiple times, please
+     * cache the result in a variable on caller side..
+     *
+     * @return iterable over all value names
+     */
+    Iterable<String> valueNames();
+
+    /**
+     * Return a list of all results with the specified name
+     *
+     * @param name of the value
+     * @return list of all values formatted as string
+     */
+    List<String> getValuesAsStrings(String name);
+}

test/framework/src/main/java/org/elasticsearch/search/aggregations/AggregatorTestCase.java

@@ -104,6 +104,7 @@ import org.elasticsearch.search.aggregations.InternalAggregation.ReduceContext;
 import org.elasticsearch.search.aggregations.MultiBucketConsumerService.MultiBucketConsumer;
 import org.elasticsearch.search.aggregations.bucket.nested.NestedAggregationBuilder;
 import org.elasticsearch.search.aggregations.metrics.MetricsAggregator;
+import org.elasticsearch.search.aggregations.metrics.MultiValueAggregation;
 import org.elasticsearch.search.aggregations.metrics.NumericMetricsAggregation;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator;
 import org.elasticsearch.search.aggregations.pipeline.PipelineAggregator.PipelineTree;
@@ -674,12 +675,20 @@ public abstract class AggregatorTestCase extends ESTestCase {
             return;
         }
 
-        assert agg instanceof NumericMetricsAggregation.MultiValue : "only multi value aggs are supported";
-
-        NumericMetricsAggregation.MultiValue multiValueAgg = (NumericMetricsAggregation.MultiValue) agg;
         Set<String> valueNames = new HashSet<>();
-        for (String name : multiValueAgg.valueNames()) {
-            valueNames.add(name);
+
+        if (agg instanceof NumericMetricsAggregation.MultiValue) {
+            NumericMetricsAggregation.MultiValue multiValueAgg = (NumericMetricsAggregation.MultiValue) agg;
+            for (String name : multiValueAgg.valueNames()) {
+                valueNames.add(name);
+            }
+        } else if (agg instanceof MultiValueAggregation) {
+            MultiValueAggregation multiValueAgg = (MultiValueAggregation) agg;
+            for (String name : multiValueAgg.valueNames()) {
+                valueNames.add(name);
+            }
+        } else {
+            assert false : "only multi value aggs are supported";
         }
 
         assertEquals(aggregationBuilder.getOutputFieldNames().get(), valueNames);

x-pack/plugin/analytics/src/main/java/org/elasticsearch/xpack/analytics/topmetrics/InternalTopMetrics.java

@@ -15,22 +15,24 @@ import org.elasticsearch.common.xcontent.ToXContent;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.search.aggregations.InternalAggregation;
-import org.elasticsearch.search.aggregations.metrics.InternalNumericMetricsAggregation;
+import org.elasticsearch.search.aggregations.metrics.InternalMultiValueAggregation;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.search.sort.SortValue;
 
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 import static java.util.Collections.emptyList;
 import static org.elasticsearch.search.builder.SearchSourceBuilder.SORT_FIELD;
 import static  org.elasticsearch.xpack.analytics.topmetrics.TopMetricsAggregationBuilder.METRIC_FIELD;
 
 
-public class InternalTopMetrics extends InternalNumericMetricsAggregation.MultiValue {
+public class InternalTopMetrics extends InternalMultiValueAggregation {
     private final SortOrder sortOrder;
     private final int size;
     private final List<String> metricNames;
@@ -162,28 +164,53 @@ public class InternalTopMetrics extends InternalNumericMetricsAggregation.MultiV
     }
 
     @Override
-    public double value(String name) {
-        int index = metricNames.indexOf(name);
+    public final double sortValue(String key) {
+        int index = metricNames.indexOf(key);
         if (index < 0) {
-            throw new IllegalArgumentException("unknown metric [" + name + "]");
+            throw new IllegalArgumentException("unknown metric [" + key + "]");
         }
         if (topMetrics.isEmpty()) {
             return Double.NaN;
         }
-        assert topMetrics.size() == 1 : "property paths should only resolve against top metrics with size == 1.";
-        // TODO it'd probably be nicer to have "compareTo" instead of assuming a double.
+
         MetricValue value = topMetrics.get(0).metricValues.get(index);
         if (value == null) {
             return Double.NaN;
         }
+
+        // TODO it'd probably be nicer to have "compareTo" instead of assuming a double.
+        // non-numeric fields always return NaN
         return value.numberValue().doubleValue();
     }
 
+    @Override
+    public List<String> getValuesAsStrings(String name) {
+        int index = metricNames.indexOf(name);
+        if (index < 0) {
+            throw new IllegalArgumentException("unknown metric [" + name + "]");
+        }
+        if (topMetrics.isEmpty()) {
+            return Collections.emptyList();
+        }
+        return topMetrics.stream().map(r -> {
+            MetricValue value = r.metricValues.get(index);
+            if (value == null) {
+                return "null";
+            }
+            return value.getValue().format(value.getFormat());
+        }).collect(Collectors.toList());
+    }
+
     @Override
     public Iterable<String> valueNames() {
         return metricNames;
     }
 
+    @Override
+    protected boolean mustReduceOnSingleInternalAgg() {
+        return false;
+    }
+
     SortOrder getSortOrder() {
         return sortOrder;
     }

x-pack/plugin/analytics/src/test/java/org/elasticsearch/xpack/analytics/topmetrics/InternalTopMetricsTests.java

@@ -225,12 +225,12 @@ public class InternalTopMetricsTests extends InternalAggregationTestCase<Interna
         assertThat((Double) metrics.getProperty("null"), notANumber());
     }
 
-    public void testValue() {
+    public void testGetValuesAsStrings() {
         InternalTopMetrics metrics = resultWithAllTypes();
-        assertThat(metrics.value("int"), equalTo(1.0));
-        assertThat(metrics.value("double"), equalTo(5.0));
-        assertThat(metrics.value("bytes"), notANumber());
-        assertThat(metrics.value("null"), notANumber());
+        assertThat(metrics.getValuesAsStrings("int"), equalTo(Collections.singletonList("1")));
+        assertThat(metrics.getValuesAsStrings("double"), equalTo(Collections.singletonList("5.0")));
+        assertThat(metrics.getValuesAsStrings("bytes"), equalTo(Collections.singletonList("cat")));
+        assertThat(metrics.getValuesAsStrings("null"), equalTo(Collections.singletonList("null")));
     }
 
     private InternalTopMetrics resultWithAllTypes() {