Support global ords in top_metrics (elastic#64967)

Adds support for fetching `keyword` and `ip` fields in `top_metrics`.
This only works if we can get global ordinals for those fields so we
can't support the entire `keyword` family, but this gets us *somewhere*.

Closes elastic#64774

server/src/main/java/org/elasticsearch/search/sort/BucketedSort.java

@@ -171,7 +171,7 @@ public int getBucketSize() {
      */
     @FunctionalInterface
     public interface ResultBuilder<T> {
-        T build(long index, SortValue sortValue);
+        T build(long index, SortValue sortValue) throws IOException;
     }
 
     /**
@@ -180,7 +180,7 @@ public interface ResultBuilder<T> {
      * @param builder builds results. See {@link ExtraData} for how to store
      *                data along side the sort for this to extract.
      */
-    public final <T extends Comparable<T>> List<T> getValues(long bucket, ResultBuilder<T> builder) {
+    public final <T extends Comparable<T>> List<T> getValues(long bucket, ResultBuilder<T> builder) throws IOException {
         long rootIndex = bucket * bucketSize;
         if (rootIndex >= values().size()) {
             // We've never seen this bucket.
@@ -201,7 +201,7 @@ public final <T extends Comparable<T>> List<T> getValues(long bucket, ResultBuil
      * Get the values for a bucket if it has been collected. If it hasn't
      * then returns an empty array.
      */
-    public final List<SortValue> getValues(long bucket) {
+    public final List<SortValue> getValues(long bucket) throws IOException {
         return getValues(bucket, (i, sv) -> sv);
     }
 

server/src/main/java/org/elasticsearch/search/sort/SortValue.java

@@ -19,6 +19,8 @@
 
 package org.elasticsearch.search.sort;
 
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.io.stream.NamedWriteable;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
@@ -48,13 +50,22 @@ public static SortValue from(long l) {
         return new LongSortValue(l);
     }
 
+    /**
+     * Get a {@linkplain SortValue} for bytes. Callers should be sure that they
+     * have a {@link BytesRef#deepCopyOf} of any mutable references.
+     */
+    public static SortValue from(BytesRef bytes) {
+        return new BytesSortValue(bytes);
+    }
+
     /**
      * Get the list of {@linkplain NamedWriteable}s that this class needs.
      */
     public static List<NamedWriteableRegistry.Entry> namedWriteables() {
         return Arrays.asList(
                 new NamedWriteableRegistry.Entry(SortValue.class, DoubleSortValue.NAME, DoubleSortValue::new),
-                new NamedWriteableRegistry.Entry(SortValue.class, LongSortValue.NAME, LongSortValue::new));
+                new NamedWriteableRegistry.Entry(SortValue.class, LongSortValue.NAME, LongSortValue::new),
+                new NamedWriteableRegistry.Entry(SortValue.class, BytesSortValue.NAME, BytesSortValue::new));
     }
 
     private SortValue() {
@@ -200,7 +211,7 @@ private static class LongSortValue extends SortValue {
             this.key = key;
         }
 
-        LongSortValue(StreamInput in) throws IOException {
+        private LongSortValue(StreamInput in) throws IOException {
             key = in.readLong();
         }
 
@@ -259,4 +270,80 @@ public Number numberValue() {
             return key;
         }
     }
+
+    private static class BytesSortValue extends SortValue {
+        public static final String NAME = "bytes";
+
+        private final BytesRef key;
+
+        BytesSortValue(BytesRef key) {
+            this.key = key;
+        }
+
+        private BytesSortValue(StreamInput in) throws IOException {
+            key = in.readBytesRef();
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            if (out.getVersion().before(Version.V_7_11_0)) {
+                throw new IllegalArgumentException(
+                    "versions of Elasticsearch before 7.11.0 can't handle non-numeric sort values and attempted to send to ["
+                        + out.getVersion()
+                        + "]"
+                );
+            }
+            out.writeBytesRef(key);
+        }
+
+        @Override
+        public String getWriteableName() {
+            return NAME;
+        }
+
+        @Override
+        public Object getKey() {
+            return key;
+        }
+
+        @Override
+        public String format(DocValueFormat format) {
+            return format.format(key).toString();
+        }
+
+        @Override
+        protected XContentBuilder rawToXContent(XContentBuilder builder) throws IOException {
+            return builder.value(key.utf8ToString());
+        }
+
+        @Override
+        protected int compareToSameType(SortValue obj) {
+            BytesSortValue other = (BytesSortValue) obj;
+            return key.compareTo(other.key);
+        }
+
+        @Override
+        public boolean equals(Object obj) {
+            if (obj == null || false == getClass().equals(obj.getClass())) {
+                return false;
+            }
+            BytesSortValue other = (BytesSortValue) obj;
+            return key.equals(other.key);
+        }
+
+        @Override
+        public int hashCode() {
+            return key.hashCode();
+        }
+
+        @Override
+        public String toString() {
+            return key.toString();
+        }
+
+        @Override
+        public Number numberValue() {
+            throw new UnsupportedOperationException();
+        }
+    }
 }

server/src/test/java/org/elasticsearch/search/sort/BucketedSortTestCase.java

@@ -73,7 +73,7 @@ private T build(SortOrder order, int bucketSize, double[] values) {
         return build(order, format, bucketSize, BucketedSort.NOOP_EXTRA_DATA, values);
     }
 
-    public final void testNeverCalled() {
+    public final void testNeverCalled() throws IOException {
         SortOrder order = randomFrom(SortOrder.values());
         DocValueFormat format = randomFrom(DocValueFormat.RAW, DocValueFormat.BINARY, DocValueFormat.BOOLEAN);
         try (T sort = build(order, format, 1, BucketedSort.NOOP_EXTRA_DATA, new double[] {})) {

server/src/test/java/org/elasticsearch/search/sort/SortValueTests.java

@@ -19,14 +19,19 @@
 
 package org.elasticsearch.search.sort;
 
+import org.apache.lucene.document.InetAddressPoint;
+import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.Version;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
+import org.elasticsearch.common.network.InetAddresses;
 import org.elasticsearch.common.time.DateFormatter;
 import org.elasticsearch.common.xcontent.ToXContentFragment;
 import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.index.mapper.DateFieldMapper;
 import org.elasticsearch.search.DocValueFormat;
 import org.elasticsearch.test.AbstractNamedWriteableTestCase;
+import org.elasticsearch.test.VersionUtils;
 
 import java.io.IOException;
 import java.time.ZoneId;
@@ -51,7 +56,12 @@ protected NamedWriteableRegistry getNamedWriteableRegistry() {
 
     @Override
     protected SortValue createTestInstance() {
-        return randomBoolean() ? SortValue.from(randomDouble()) : SortValue.from(randomLong());
+        switch (between(0, 2)) {
+            case 0: return SortValue.from(randomDouble());
+            case 1: return SortValue.from(randomLong());
+            case 2: return SortValue.from(new BytesRef(randomAlphaOfLength(5)));
+            default: throw new AssertionError();
+        }
     }
 
     @Override
@@ -81,11 +91,23 @@ public void testToXContentLong() {
         assertThat(toXContent(SortValue.from(1), STRICT_DATE_TIME), equalTo("{\"test\":\"1970-01-01T00:00:00.001Z\"}"));
     }
 
+    public void testToXContentBytes() {
+        assertThat(toXContent(SortValue.from(new BytesRef("cat")), DocValueFormat.RAW), equalTo("{\"test\":\"cat\"}"));
+        assertThat(
+            toXContent(SortValue.from(new BytesRef(InetAddressPoint.encode(InetAddresses.forString("127.0.0.1")))), DocValueFormat.IP),
+            equalTo("{\"test\":\"127.0.0.1\"}")
+        );
+    }
+
     public void testCompareDifferentTypes() {
         assertThat(SortValue.from(1.0), lessThan(SortValue.from(1)));
         assertThat(SortValue.from(Double.MAX_VALUE), lessThan(SortValue.from(Long.MIN_VALUE)));
         assertThat(SortValue.from(1), greaterThan(SortValue.from(1.0)));
         assertThat(SortValue.from(Long.MIN_VALUE), greaterThan(SortValue.from(Double.MAX_VALUE)));
+        assertThat(SortValue.from(new BytesRef("cat")), lessThan(SortValue.from(1)));
+        assertThat(SortValue.from(1), greaterThan(SortValue.from(new BytesRef("cat"))));
+        assertThat(SortValue.from(new BytesRef("cat")), lessThan(SortValue.from(1.0)));
+        assertThat(SortValue.from(1.0), greaterThan(SortValue.from(new BytesRef("cat"))));
     }
 
     public void testCompareDoubles() {
@@ -102,6 +124,25 @@ public void testCompareLongs() {
         assertThat(SortValue.from(r), greaterThan(SortValue.from(r - 1)));
     }
 
+    public void testBytes() {
+        String r = randomAlphaOfLength(5);
+        assertThat(SortValue.from(new BytesRef(r)), equalTo(SortValue.from(new BytesRef(r))));
+        assertThat(SortValue.from(new BytesRef(r)), lessThan(SortValue.from(new BytesRef(r + "with_suffix"))));
+        assertThat(SortValue.from(new BytesRef(r)), greaterThan(SortValue.from(new BytesRef(new byte[] {}))));
+    }
+
+    public void testSerializeBytesToOldVersion() {
+        SortValue value = SortValue.from(new BytesRef("can't send me!"));
+        Version version = VersionUtils.randomVersionBetween(random(), Version.V_7_0_0, Version.V_7_10_1);
+        Exception e = expectThrows(IllegalArgumentException.class, () -> copyInstance(value, version));
+        assertThat(
+            e.getMessage(),
+            equalTo(
+                "versions of Elasticsearch before 7.11.0 can't handle non-numeric sort values and attempted to send to [" + version + "]"
+            )
+        );
+    }
+
     public String toXContent(SortValue sortValue, DocValueFormat format) {
         return Strings.toString(new ToXContentFragment() {
             @Override

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

@@ -114,7 +114,8 @@ public List<AggregationSpec> getAggregations() {
                 TopMetricsAggregationBuilder.NAME,
                 TopMetricsAggregationBuilder::new,
                 usage.track(AnalyticsStatsAction.Item.TOP_METRICS, checkLicense(TopMetricsAggregationBuilder.PARSER)))
-                .addResultReader(InternalTopMetrics::new),
+                .addResultReader(InternalTopMetrics::new)
+                .setAggregatorRegistrar(TopMetricsAggregationBuilder::registerAggregators),
             new AggregationSpec(
                 TTestAggregationBuilder.NAME,
                 TTestAggregationBuilder::new,

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

@@ -18,7 +18,10 @@
 import org.elasticsearch.search.aggregations.AggregatorFactories.Builder;
 import org.elasticsearch.search.aggregations.AggregatorFactory;
 import org.elasticsearch.search.aggregations.support.AggregationContext;
+import org.elasticsearch.search.aggregations.support.CoreValuesSourceType;
 import org.elasticsearch.search.aggregations.support.MultiValuesSourceFieldConfig;
+import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry;
+import org.elasticsearch.search.aggregations.support.ValuesSourceRegistry.RegistryKey;
 import org.elasticsearch.search.sort.SortBuilder;
 
 import java.io.IOException;
@@ -34,6 +37,33 @@ public class TopMetricsAggregationBuilder extends AbstractAggregationBuilder<Top
     public static final String NAME = "top_metrics";
     public static final ParseField METRIC_FIELD = new ParseField("metrics");
 
+    static final RegistryKey<TopMetricsAggregator.MetricValuesSupplier> REGISTRY_KEY = new RegistryKey<>(
+        TopMetricsAggregationBuilder.NAME,
+        TopMetricsAggregator.MetricValuesSupplier.class
+    );
+
+    public static void registerAggregators(ValuesSourceRegistry.Builder registry) {
+        registry.registerUsage(NAME);
+        registry.register(
+            REGISTRY_KEY,
+            org.elasticsearch.common.collect.List.of(CoreValuesSourceType.NUMERIC),
+            TopMetricsAggregator::buildNumericMetricValues,
+            false
+        );
+        registry.register(
+            REGISTRY_KEY,
+            org.elasticsearch.common.collect.List.of(CoreValuesSourceType.BOOLEAN, CoreValuesSourceType.DATE),
+            TopMetricsAggregator.LongMetricValues::new,
+            false
+        );
+        registry.register(
+            REGISTRY_KEY,
+            org.elasticsearch.common.collect.List.of(CoreValuesSourceType.BYTES, CoreValuesSourceType.IP),
+            TopMetricsAggregator.GlobalOrdsValues::new,
+            false
+        );
+    }
+
     /**
      * Default to returning only a single top metric.
      */