[Transform] Support `aggregate_metric_double` field type in transform aggregations (#91045)

docs/changelog/91045.yaml

@@ -0,0 +1,5 @@
+pr: 91045
+summary: Support `aggregate_metric_double` field type in transform aggregations
+area: Transform
+type: enhancement
+issues: []

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformPivotRestIT.java

@@ -1707,6 +1707,85 @@ public class TransformPivotRestIT extends TransformRestTestCase {
         assertEquals("business_3", actual);
     }
 
+    public void testPivotWithAggregateMetricDouble() throws Exception {
+        String transformId = "aggregate_metric_double_transform";
+        String transformIndex = "aggregate_metric_double_pivot_reviews";
+        String statsField = "stars_stats";
+        setupDataAccessRole(DATA_ACCESS_ROLE, REVIEWS_INDEX_NAME, transformIndex);
+
+        String config = formatted("""
+            {
+              "source": {
+                "index": "%s"
+              },
+              "dest": {
+                "index": "%s"
+              },
+              "pivot": {
+                "group_by": {
+                  "reviewer": {
+                    "terms": {
+                      "field": "user_id"
+                    }
+                  }
+                },
+                "aggregations": {
+                  "stats_min": {
+                    "min": {
+                      "field": "%s"
+                    }
+                  },
+                  "stats_max": {
+                    "max": {
+                      "field": "%s"
+                    }
+                  },
+                  "stats_sum": {
+                    "sum": {
+                      "field": "%s"
+                    }
+                  }
+                }
+              }
+            }""", REVIEWS_INDEX_NAME, transformIndex, statsField, statsField, statsField);
+
+        final Request createPreviewRequest = createRequestWithAuth("POST", getTransformEndpoint() + "_preview", null);
+        createPreviewRequest.setJsonEntity(config);
+        Map<String, Object> previewTransformResponse = entityAsMap(client().performRequest(createPreviewRequest));
+        @SuppressWarnings("unchecked")
+        Map<String, Object> mappingsProperties = (Map<String, Object>) XContentMapValues.extractValue(
+            "generated_dest_index.mappings.properties",
+            previewTransformResponse
+        );
+        assertThat(XContentMapValues.extractValue("stats_min.type", mappingsProperties), is(equalTo("double")));
+        assertThat(XContentMapValues.extractValue("stats_max.type", mappingsProperties), is(equalTo("double")));
+        assertThat(XContentMapValues.extractValue("stats_sum.type", mappingsProperties), is(equalTo("double")));
+
+        final Request createTransformRequest = createRequestWithAuth(
+            "PUT",
+            getTransformEndpoint() + transformId,
+            BASIC_AUTH_VALUE_TRANSFORM_ADMIN_WITH_SOME_DATA_ACCESS
+        );
+        createTransformRequest.setJsonEntity(config);
+        Map<String, Object> createTransformResponse = entityAsMap(client().performRequest(createTransformRequest));
+        assertThat(createTransformResponse.get("acknowledged"), equalTo(Boolean.TRUE));
+
+        startAndWaitForTransform(transformId, transformIndex, BASIC_AUTH_VALUE_TRANSFORM_ADMIN_WITH_SOME_DATA_ACCESS);
+        assertTrue(indexExists(transformIndex));
+
+        Map<String, Object> searchResult = getAsMap(transformIndex + "/_search?q=reviewer:user_4");
+        assertEquals(1, XContentMapValues.extractValue("hits.total.value", searchResult));
+        assertThat(((List<?>) XContentMapValues.extractValue("hits.hits._source.stats_min", searchResult)).get(0), is(equalTo(0.0)));
+        assertThat(((List<?>) XContentMapValues.extractValue("hits.hits._source.stats_max", searchResult)).get(0), is(equalTo(6.0)));
+        assertThat(((List<?>) XContentMapValues.extractValue("hits.hits._source.stats_sum", searchResult)).get(0), is(equalTo(1590.0)));
+
+        searchResult = getAsMap(transformIndex + "/_search?q=reviewer:user_1");
+        assertEquals(1, XContentMapValues.extractValue("hits.total.value", searchResult));
+        assertThat(((List<?>) XContentMapValues.extractValue("hits.hits._source.stats_min", searchResult)).get(0), is(equalTo(0.0)));
+        assertThat(((List<?>) XContentMapValues.extractValue("hits.hits._source.stats_max", searchResult)).get(0), is(equalTo(6.0)));
+        assertThat(((List<?>) XContentMapValues.extractValue("hits.hits._source.stats_sum", searchResult)).get(0), is(equalTo(2020.0)));
+    }
+
     public void testManyBucketsWithSmallPageSize() throws Exception {
         String transformId = "test_with_many_buckets";
         String transformIndex = transformId + "-idx";

x-pack/plugin/transform/qa/single-node-tests/src/javaRestTest/java/org/elasticsearch/xpack/transform/integration/TransformRestTestCase.java

@@ -107,6 +107,18 @@ public abstract class TransformRestTestCase extends ESRestTestCase {
             if ((user == userWithMissingBuckets && missingBucketField.equals("stars")) == false) {
                 bulk.append("\"stars\":").append(stars).append(",");
             }
+            if ((user == userWithMissingBuckets && missingBucketField.equals("stars_stats")) == false) {
+                bulk.append("\"stars_stats\": { ")
+                    .append("\"min\": ")
+                    .append(stars - 1)
+                    .append(",")
+                    .append("\"max\": ")
+                    .append(stars + 1)
+                    .append(",")
+                    .append("\"sum\": ")
+                    .append(10 * stars)
+                    .append("},");
+            }
             if ((user == userWithMissingBuckets && missingBucketField.equals("location")) == false) {
                 bulk.append("\"location\":\"").append(location).append("\",");
             }
@@ -177,6 +189,11 @@ public abstract class TransformRestTestCase extends ESRestTestCase {
                     .startObject("stars")
                     .field("type", randomFrom("integer", "long")) // gh#64347 unsigned_long disabled
                     .endObject()
+                    .startObject("stars_stats")
+                    .field("type", "aggregate_metric_double")
+                    .field("metrics", List.of("min", "max", "sum"))
+                    .field("default_metric", "max")
+                    .endObject()
                     .startObject("location")
                     .field("type", "geo_point")
                     .endObject()

x-pack/plugin/transform/src/main/java/org/elasticsearch/xpack/transform/transforms/pivot/TransformAggregations.java

@@ -38,6 +38,7 @@ public final class TransformAggregations {
     public static final String FLATTENED = "flattened";
     public static final String SCALED_FLOAT = "scaled_float";
     public static final String DOUBLE = "double";
+    public static final String AGGREGATE_METRIC_DOUBLE = "aggregate_metric_double";
     public static final String LONG = "long";
     public static final String GEO_SHAPE = "geo_shape";
     public static final String GEO_POINT = "geo_point";
@@ -175,6 +176,11 @@ public final class TransformAggregations {
                 return FLOAT;
             }
 
+            // min/max/sum aggregations over aggregate_metric_double return double
+            if (sourceType.equals(AGGREGATE_METRIC_DOUBLE)) {
+                return DOUBLE;
+            }
+
             return sourceType;
         }
 

x-pack/plugin/transform/src/test/java/org/elasticsearch/xpack/transform/transforms/pivot/TransformAggregationsTests.java

@@ -54,17 +54,20 @@ public class TransformAggregationsTests extends ESTestCase {
         // max
         assertEquals("int", TransformAggregations.resolveTargetMapping("max", "int"));
         assertEquals("double", TransformAggregations.resolveTargetMapping("max", "double"));
+        assertEquals("double", TransformAggregations.resolveTargetMapping("max", "aggregate_metric_double"));
         assertEquals("half_float", TransformAggregations.resolveTargetMapping("max", "half_float"));
         assertEquals("float", TransformAggregations.resolveTargetMapping("max", "scaled_float"));
 
         // min
         assertEquals("int", TransformAggregations.resolveTargetMapping("min", "int"));
         assertEquals("double", TransformAggregations.resolveTargetMapping("min", "double"));
+        assertEquals("double", TransformAggregations.resolveTargetMapping("min", "aggregate_metric_double"));
         assertEquals("half_float", TransformAggregations.resolveTargetMapping("min", "half_float"));
         assertEquals("float", TransformAggregations.resolveTargetMapping("min", "scaled_float"));
 
         // sum
         assertEquals("double", TransformAggregations.resolveTargetMapping("sum", "double"));
+        assertEquals("double", TransformAggregations.resolveTargetMapping("sum", "aggregate_metric_double"));
         assertEquals("double", TransformAggregations.resolveTargetMapping("sum", "half_float"));
         assertEquals("double", TransformAggregations.resolveTargetMapping("sum", null));