[Transform] Integration test for the case of empty key in the terms agg in transform (#90941)

Author: Przemysław Witek

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

@@ -252,6 +252,114 @@ public void testRestrictiveBucketSelector() throws Exception {
         verifyDestIndexHitsCount(indexName, "special_pivot_bucket_selector-10000", 10000, 41);
     }
 
+    public void testEmptyKeyInTermsAgg() throws Exception {
+        String indexName = REVIEWS_INDEX_NAME;
+
+        {
+            final Request request = new Request("PUT", indexName + "/_doc/strange-business-id-1");
+            request.addParameter("refresh", "true");
+            request.setJsonEntity("""
+                {
+                  "user_id": "user_0",
+                  "business_id": ""
+                }""");
+            client().performRequest(request);
+        }
+        {
+            final Request request = new Request("PUT", indexName + "/_doc/strange-business-id-2");
+            request.addParameter("refresh", "true");
+            request.setJsonEntity("""
+                {
+                  "user_id": "user_0",
+                  "business_id": "business_x."
+                }""");
+            client().performRequest(request);
+        }
+        {
+            final Request request = new Request("PUT", indexName + "/_doc/strange-business-id-3");
+            request.addParameter("refresh", "true");
+            request.setJsonEntity("""
+                {
+                  "user_id": "user_0",
+                  "business_id": ".business_y"
+                }""");
+            client().performRequest(request);
+        }
+        {
+            final Request request = new Request("PUT", indexName + "/_doc/strange-business-id-4");
+            request.addParameter("refresh", "true");
+            request.setJsonEntity("""
+                {
+                  "user_id": "user_0",
+                  "business_id": "..."
+                }""");
+            client().performRequest(request);
+        }
+
+        String transformIndex = "empty-terms-agg-key";
+        String transformId = "empty-terms-agg-key";
+        final Request createTransformRequest = new Request("PUT", getTransformEndpoint() + transformId);
+        final String config = formatted("""
+            {
+              "source": {
+                "index": "%s"
+              },
+              "dest": {
+                "index": "%s"
+              },
+              "pivot": {
+                "group_by": {
+                  "reviewer": {
+                    "terms": {
+                      "field": "user_id"
+                    }
+                  }
+                },
+                "aggregations": {
+                  "businesses": {
+                    "terms": {
+                      "field": "business_id"
+                    }
+                  }
+                }
+              }
+            }""", indexName, transformIndex);
+        createTransformRequest.setJsonEntity(config);
+        Map<String, Object> createTransformResponse = entityAsMap(client().performRequest(createTransformRequest));
+        assertThat(createTransformResponse.get("acknowledged"), equalTo(Boolean.TRUE));
+        startAndWaitForTransform(transformId, transformIndex);
+        assertTrue(indexExists(transformIndex));
+        Map<String, Object> searchResult = getAsMap(transformIndex + "/_search?q=reviewer:user_0");
+        long count = (Integer) XContentMapValues.extractValue("hits.total.value", searchResult);
+        assertThat(count, is(equalTo(1L)));
+        assertThat(XContentMapValues.extractValue("hits.hits._source.reviewer", searchResult), is(equalTo(List.of("user_0"))));
+        assertThat(
+            XContentMapValues.extractValue("hits.hits._source.businesses", searchResult),
+            is(equalTo(List.of(Map.of("business_0", 278, "", 1, "business_x.", 1, ".business_y", 1, "...", 1))))
+        );
+
+        {
+            final Request request = new Request("DELETE", indexName + "/_doc/strange-business-id-1");
+            request.addParameter("refresh", "true");
+            client().performRequest(request);
+        }
+        {
+            final Request request = new Request("DELETE", indexName + "/_doc/strange-business-id-2");
+            request.addParameter("refresh", "true");
+            client().performRequest(request);
+        }
+        {
+            final Request request = new Request("DELETE", indexName + "/_doc/strange-business-id-3");
+            request.addParameter("refresh", "true");
+            client().performRequest(request);
+        }
+        {
+            final Request request = new Request("DELETE", indexName + "/_doc/strange-business-id-4");
+            request.addParameter("refresh", "true");
+            client().performRequest(request);
+        }
+    }
+
     private void verifyDestIndexHitsCount(String sourceIndex, String transformId, int maxPageSearchSize, long expectedDestIndexCount)
         throws Exception {
         String transformIndex = transformId;