SQL: Translate MIN/MAX on keyword fields as FIRST/LAST (#41247)

Although the translation rule was implemented in the `Optimizer`,
the rule was not added in the list of rules to be executed.

Relates to #41195
Follows #37936


(cherry picked from commit f426a33)

Author: matriv

x-pack/plugin/sql/qa/src/main/resources/agg.csv-spec

@@ -414,6 +414,26 @@ SELECT COUNT(ALL last_name)=COUNT(ALL first_name) AS areEqual, COUNT(ALL first_n
 false          |90             |100
 ;
 
+topHitsAsMinAndMax
+schema::min:s|max:s|first:s|last:s
+SELECT MIN(first_name) as min, MAX(first_name) as max, FIRST(first_name) as first, LAST(first_name) as last FROM test_emp;
+
+    min        |   max         |   first      |   last
+---------------+---------------+--------------+----------
+   Alejandro   |   Zvonko      |   Alejandro  |   Zvonko
+;
+
+topHitsAsMinAndMaxAndGroupBy
+schema::gender:s|min:s|max:s|first:s|last:s
+SELECT gender, MIN(first_name) as min, MAX(first_name) as max, FIRST(first_name) as first, LAST(first_name) as last FROM test_emp GROUP BY gender ORDER BY gender;
+
+    gender     |   min         |   max        |   first       |   last
+---------------+---------------+--------------+---------------+----------
+null           |   Berni       |   Patricio   |   Berni       |   Patricio
+F              |   Alejandro   |   Xinglin    |   Alejandro   |   Xinglin
+M              |   Amabile     |   Zvonko     |   Amabile     |   Zvonko
+;
+
 topHitsWithOneArgAndGroupBy
 schema::gender:s|first:s|last:s
 SELECT gender, FIRST(first_name) as first, LAST(first_name) as last FROM test_emp GROUP BY gender ORDER BY gender;

x-pack/plugin/sql/src/main/java/org/elasticsearch/xpack/sql/optimizer/Optimizer.java

@@ -149,6 +149,7 @@ protected Iterable<RuleExecutor<LogicalPlan>.Batch> batches() {
 
         Batch aggregate = new Batch("Aggregation Rewrite",
                 //new ReplaceDuplicateAggsWithReferences(),
+                new ReplaceMinMaxWithTopHits(),
                 new ReplaceAggsWithMatrixStats(),
                 new ReplaceAggsWithExtendedStats(),
                 new ReplaceAggsWithStats(),

x-pack/plugin/sql/src/test/java/org/elasticsearch/xpack/sql/planner/QueryTranslatorTests.java

@@ -763,6 +763,18 @@ public void testTopHitsAggregationWithOneArg() {
                         "\"explain\":false,\"docvalue_fields\":[{\"field\":\"keyword\"}]," +
                         "\"sort\":[{\"keyword\":{\"order\":\"asc\",\"missing\":\"_last\",\"unmapped_type\":\"keyword\"}}]}}}}}"));
         }
+        {
+            PhysicalPlan p = optimizeAndPlan("SELECT MIN(keyword) FROM test");
+            assertEquals(EsQueryExec.class, p.getClass());
+            EsQueryExec eqe = (EsQueryExec) p;
+            assertEquals(1, eqe.output().size());
+            assertEquals("MIN(keyword)", eqe.output().get(0).qualifiedName());
+            assertEquals(DataType.KEYWORD, eqe.output().get(0).dataType());
+            assertThat(eqe.queryContainer().aggs().asAggBuilder().toString().replaceAll("\\s+", ""),
+                endsWith("\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false," +
+                    "\"explain\":false,\"docvalue_fields\":[{\"field\":\"keyword\"}]," +
+                    "\"sort\":[{\"keyword\":{\"order\":\"asc\",\"missing\":\"_last\",\"unmapped_type\":\"keyword\"}}]}}}}}"));
+        }
         {
             PhysicalPlan p = optimizeAndPlan("SELECT LAST(date) FROM test");
             assertEquals(EsQueryExec.class, p.getClass());
@@ -775,6 +787,18 @@ public void testTopHitsAggregationWithOneArg() {
                     "\"explain\":false,\"docvalue_fields\":[{\"field\":\"date\",\"format\":\"epoch_millis\"}]," +
                     "\"sort\":[{\"date\":{\"order\":\"desc\",\"missing\":\"_last\",\"unmapped_type\":\"date\"}}]}}}}}"));
         }
+        {
+            PhysicalPlan p = optimizeAndPlan("SELECT MAX(keyword) FROM test");
+            assertEquals(EsQueryExec.class, p.getClass());
+            EsQueryExec eqe = (EsQueryExec) p;
+            assertEquals(1, eqe.output().size());
+            assertEquals("MAX(keyword)", eqe.output().get(0).qualifiedName());
+            assertEquals(DataType.KEYWORD, eqe.output().get(0).dataType());
+            assertThat(eqe.queryContainer().aggs().asAggBuilder().toString().replaceAll("\\s+", ""),
+                endsWith("\"top_hits\":{\"from\":0,\"size\":1,\"version\":false,\"seq_no_primary_term\":false," +
+                    "\"explain\":false,\"docvalue_fields\":[{\"field\":\"keyword\"}]," +
+                    "\"sort\":[{\"keyword\":{\"order\":\"desc\",\"missing\":\"_last\",\"unmapped_type\":\"keyword\"}}]}}}}}"));
+        }
     }
 
     public void testTopHitsAggregationWithTwoArgs() {