[TEST] Add coverage for field caps and ES|QL to LogsDB QA testing (#114505)

* Add coverage for field caps and ES|QL to LogsDB QA testing

* address comments

* address comments

* address comments

Author: kkrik-es

modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/logsdb/qa/AbstractChallengeRestTest.java

@@ -276,6 +276,33 @@ private Response query(final SearchSourceBuilder search, final Supplier<String>
         return client.performRequest(request);
     }
 
+    public Response esqlBaseline(final String query) throws IOException {
+        return esql(query, this::getBaselineDataStreamName);
+    }
+
+    public Response esqlContender(final String query) throws IOException {
+        return esql(query, this::getContenderDataStreamName);
+    }
+
+    private Response esql(final String query, final Supplier<String> dataStreamNameSupplier) throws IOException {
+        final Request request = new Request("POST", "/_query");
+        request.setJsonEntity("{\"query\": \"" + query.replace("$index", dataStreamNameSupplier.get()) + "\"}");
+        return client.performRequest(request);
+    }
+
+    public Response fieldCapsBaseline() throws IOException {
+        return fieldCaps(this::getBaselineDataStreamName);
+    }
+
+    public Response fieldCapsContender() throws IOException {
+        return fieldCaps(this::getContenderDataStreamName);
+    }
+
+    private Response fieldCaps(final Supplier<String> dataStreamNameSupplier) throws IOException {
+        final Request request = new Request("GET", "/" + dataStreamNameSupplier.get() + "/_field_caps?fields=*");
+        return client.performRequest(request);
+    }
+
     public String getBaselineDataStreamName() {
         return baselineDataStreamName;
     }

modules/data-streams/src/javaRestTest/java/org/elasticsearch/datastreams/logsdb/qa/StandardVersusLogsIndexModeChallengeRestIT.java

@@ -42,6 +42,7 @@
 import java.util.Comparator;
 import java.util.List;
 import java.util.Map;
+import java.util.TreeMap;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.greaterThan;
@@ -271,6 +272,51 @@ public void testDateHistogramAggregation() throws IOException {
         assertTrue(matchResult.getMessage(), matchResult.isMatch());
     }
 
+    public void testEsqlSource() throws IOException {
+        int numberOfDocuments = ESTestCase.randomIntBetween(100, 200);
+        final List<XContentBuilder> documents = generateDocuments(numberOfDocuments);
+
+        indexDocuments(documents);
+
+        final String query = "FROM $index METADATA _source, _id | KEEP _source, _id | LIMIT " + numberOfDocuments;
+        final MatchResult matchResult = Matcher.matchSource()
+            .mappings(getContenderMappings(), getBaselineMappings())
+            .settings(getContenderSettings(), getBaselineSettings())
+            .expected(getEsqlSourceResults(esqlBaseline(query)))
+            .ignoringSort(true)
+            .isEqualTo(getEsqlSourceResults(esqlContender(query)));
+        assertTrue(matchResult.getMessage(), matchResult.isMatch());
+    }
+
+    public void testEsqlTermsAggregation() throws IOException {
+        int numberOfDocuments = ESTestCase.randomIntBetween(100, 200);
+        final List<XContentBuilder> documents = generateDocuments(numberOfDocuments);
+
+        indexDocuments(documents);
+
+        final String query = "FROM $index | STATS count(*) BY host.name | SORT host.name | LIMIT " + numberOfDocuments;
+        final MatchResult matchResult = Matcher.mappings(getContenderMappings(), getBaselineMappings())
+            .settings(getContenderSettings(), getBaselineSettings())
+            .expected(getEsqlStatsResults(esqlBaseline(query)))
+            .ignoringSort(true)
+            .isEqualTo(getEsqlStatsResults(esqlContender(query)));
+        assertTrue(matchResult.getMessage(), matchResult.isMatch());
+    }
+
+    public void testFieldCaps() throws IOException {
+        int numberOfDocuments = ESTestCase.randomIntBetween(20, 50);
+        final List<XContentBuilder> documents = generateDocuments(numberOfDocuments);
+
+        indexDocuments(documents);
+
+        final MatchResult matchResult = Matcher.mappings(getContenderMappings(), getBaselineMappings())
+            .settings(getContenderSettings(), getBaselineSettings())
+            .expected(getFields(fieldCapsBaseline()))
+            .ignoringSort(true)
+            .isEqualTo(getFields(fieldCapsContender()));
+        assertTrue(matchResult.getMessage(), matchResult.isMatch());
+    }
+
     @Override
     public Response indexBaselineDocuments(CheckedSupplier<List<XContentBuilder>, IOException> documentsSupplier) throws IOException {
         var response = super.indexBaselineDocuments(documentsSupplier);
@@ -329,6 +375,40 @@ private static List<Map<String, Object>> getQueryHits(final Response response) t
             .toList();
     }
 
+    @SuppressWarnings("unchecked")
+    private static Map<String, Object> getFields(final Response response) throws IOException {
+        final Map<String, Object> map = XContentHelper.convertToMap(XContentType.JSON.xContent(), response.getEntity().getContent(), true);
+        final Map<String, Object> fields = (Map<String, Object>) map.get("fields");
+        assertThat(fields.size(), greaterThan(0));
+        return new TreeMap<>(fields);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static List<Map<String, Object>> getEsqlSourceResults(final Response response) throws IOException {
+        final Map<String, Object> map = XContentHelper.convertToMap(XContentType.JSON.xContent(), response.getEntity().getContent(), true);
+        final List<List<Object>> values = (List<List<Object>>) map.get("values");
+        assertThat(values.size(), greaterThan(0));
+
+        // Results contain a list of [source, id] lists.
+        return values.stream()
+            .sorted(Comparator.comparingInt((List<Object> value) -> Integer.parseInt((String) value.get(1))))
+            .map(value -> (Map<String, Object>) value.get(0))
+            .toList();
+    }
+
+    @SuppressWarnings("unchecked")
+    private static List<Map<String, Object>> getEsqlStatsResults(final Response response) throws IOException {
+        final Map<String, Object> map = XContentHelper.convertToMap(XContentType.JSON.xContent(), response.getEntity().getContent(), true);
+        final List<List<Object>> values = (List<List<Object>>) map.get("values");
+        assertThat(values.size(), greaterThan(0));
+
+        // Results contain a list of [agg value, group name] lists.
+        return values.stream()
+            .sorted(Comparator.comparing((List<Object> value) -> (String) value.get(1)))
+            .map(value -> Map.of((String) value.get(1), value.get(0)))
+            .toList();
+    }
+
     @SuppressWarnings("unchecked")
     private static List<Map<String, Object>> getAggregationBuckets(final Response response, final String aggName) throws IOException {
         final Map<String, Object> map = XContentHelper.convertToMap(XContentType.JSON.xContent(), response.getEntity().getContent(), true);