ESQL: Fix constant keyword optimization

Fixes the ESQL's detection of `constant_keyword` fields. We unplugged it
when we changed a function signature because we didn't have an
`@Override` annotation. This plugs it back in and adds it to the
integration tests we use for pushing queries to lucene. When you do
`| WHERE constant_keyword_field == "itsvalue"` then the whole is removed
from the query plan because *all* documents are equal.

Author: nik9000

x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/LocalSourceOperator.java

@@ -82,7 +82,10 @@ public Page getOutput() {
     }
 
     @Override
-    public void close() {
+    public void close() {}
 
+    @Override
+    public String toString() {
+        return "LocalSourceOperator";
     }
 }

x-pack/plugin/esql/qa/server/single-node/src/javaRestTest/java/org/elasticsearch/xpack/esql/qa/single_node/PushQueriesIT.java

@@ -57,7 +57,7 @@ public class PushQueriesIT extends ESRestTestCase {
 
     @ParametersFactory(argumentFormatting = "%1s")
     public static List<Object[]> args() {
-        return Stream.of("auto", "text", "match_only_text", "semantic_text").map(s -> new Object[] { s }).toList();
+        return Stream.of("auto", "text", "match_only_text", "semantic_text", "constant_keyword").map(s -> new Object[] { s }).toList();
     }
 
     private final String type;
@@ -74,16 +74,16 @@ public void testEquality() throws IOException {
             """;
         String luceneQuery = switch (type) {
             case "text", "auto" -> "#test.keyword:%value -_ignored:test.keyword";
-            case "match_only_text" -> "*:*";
+            case "constant_keyword", "match_only_text" -> "*:*";
             case "semantic_text" -> "FieldExistsQuery [field=_primary_term]";
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        boolean filterInCompute = switch (type) {
-            case "text", "auto" -> false;
-            case "match_only_text", "semantic_text" -> true;
+        ComputeSignature dataNodeSignature = switch (type) {
+            case "auto", "constant_keyword", "text" -> ComputeSignature.FILTER_IN_QUERY;
+            case "match_only_text", "semantic_text" -> ComputeSignature.FILTER_IN_COMPUTE;
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        testPushQuery(value, esqlQuery, List.of(luceneQuery), filterInCompute, true);
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), dataNodeSignature, true);
     }
 
     public void testEqualityTooBigToPush() throws IOException {
@@ -93,11 +93,16 @@ public void testEqualityTooBigToPush() throws IOException {
             | WHERE test == "%value"
             """;
         String luceneQuery = switch (type) {
-            case "text", "auto", "match_only_text" -> "*:*";
+            case "auto", "constant_keyword", "match_only_text", "text" -> "*:*";
             case "semantic_text" -> "FieldExistsQuery [field=_primary_term]";
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        testPushQuery(value, esqlQuery, List.of(luceneQuery), true, true);
+        ComputeSignature dataNodeSignature = switch (type) {
+            case "constant_keyword" -> ComputeSignature.FILTER_IN_QUERY;
+            case "auto", "match_only_text", "semantic_text", "text" -> ComputeSignature.FILTER_IN_COMPUTE;
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), dataNodeSignature, true);
     }
 
     /**
@@ -111,11 +116,16 @@ public void testEqualityOrTooBig() throws IOException {
             | WHERE test == "%value" OR test == "%tooBig"
             """.replace("%tooBig", tooBig);
         String luceneQuery = switch (type) {
-            case "text", "auto", "match_only_text" -> "*:*";
+            case "auto", "constant_keyword", "match_only_text", "text" -> "*:*";
             case "semantic_text" -> "FieldExistsQuery [field=_primary_term]";
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        testPushQuery(value, esqlQuery, List.of(luceneQuery), true, true);
+        ComputeSignature dataNodeSignature = switch (type) {
+            case "constant_keyword" -> ComputeSignature.FILTER_IN_QUERY;
+            case "auto", "match_only_text", "semantic_text", "text" -> ComputeSignature.FILTER_IN_COMPUTE;
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), dataNodeSignature, true);
     }
 
     public void testEqualityOrOther() throws IOException {
@@ -125,17 +135,17 @@ public void testEqualityOrOther() throws IOException {
             | WHERE test == "%value" OR foo == 2
             """;
         String luceneQuery = switch (type) {
-            case "text", "auto" -> "(#test.keyword:%value -_ignored:test.keyword) foo:[2 TO 2]";
-            case "match_only_text" -> "*:*";
+            case "auto", "text" -> "(#test.keyword:%value -_ignored:test.keyword) foo:[2 TO 2]";
+            case "constant_keyword", "match_only_text" -> "*:*";
             case "semantic_text" -> "FieldExistsQuery [field=_primary_term]";
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        boolean filterInCompute = switch (type) {
-            case "text", "auto" -> false;
-            case "match_only_text", "semantic_text" -> true;
+        ComputeSignature dataNodeSignature = switch (type) {
+            case "auto", "constant_keyword", "text" -> ComputeSignature.FILTER_IN_QUERY;
+            case "match_only_text", "semantic_text" -> ComputeSignature.FILTER_IN_COMPUTE;
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        testPushQuery(value, esqlQuery, List.of(luceneQuery), filterInCompute, true);
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), dataNodeSignature, true);
     }
 
     public void testEqualityAndOther() throws IOException {
@@ -145,8 +155,8 @@ public void testEqualityAndOther() throws IOException {
             | WHERE test == "%value" AND foo == 1
             """;
         List<String> luceneQueryOptions = switch (type) {
-            case "text", "auto" -> List.of("#test.keyword:%value -_ignored:test.keyword #foo:[1 TO 1]");
-            case "match_only_text" -> List.of("foo:[1 TO 1]");
+            case "auto", "text" -> List.of("#test.keyword:%value -_ignored:test.keyword #foo:[1 TO 1]");
+            case "constant_keyword", "match_only_text" -> List.of("foo:[1 TO 1]");
             case "semantic_text" ->
                 /*
                  * single_value_match is here because there are extra documents hiding in the index
@@ -158,12 +168,12 @@ public void testEqualityAndOther() throws IOException {
                 );
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        boolean filterInCompute = switch (type) {
-            case "text", "auto" -> false;
-            case "match_only_text", "semantic_text" -> true;
+        ComputeSignature dataNodeSignature = switch (type) {
+            case "auto", "constant_keyword", "text" -> ComputeSignature.FILTER_IN_QUERY;
+            case "match_only_text", "semantic_text" -> ComputeSignature.FILTER_IN_COMPUTE;
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        testPushQuery(value, esqlQuery, luceneQueryOptions, filterInCompute, true);
+        testPushQuery(value, esqlQuery, luceneQueryOptions, dataNodeSignature, true);
     }
 
     public void testInequality() throws IOException {
@@ -173,12 +183,17 @@ public void testInequality() throws IOException {
             | WHERE test != "%different_value"
             """;
         String luceneQuery = switch (type) {
-            case "text", "auto" -> "(-test.keyword:%different_value #*:*) _ignored:test.keyword";
-            case "match_only_text" -> "*:*";
+            case "auto", "text" -> "(-test.keyword:%different_value #*:*) _ignored:test.keyword";
+            case "constant_keyword", "match_only_text" -> "*:*";
             case "semantic_text" -> "FieldExistsQuery [field=_primary_term]";
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        testPushQuery(value, esqlQuery, List.of(luceneQuery), true, true);
+        ComputeSignature dataNodeSignature = switch (type) {
+            case "constant_keyword" -> ComputeSignature.FILTER_IN_QUERY;
+            case "auto", "match_only_text", "semantic_text", "text" -> ComputeSignature.FILTER_IN_COMPUTE;
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), dataNodeSignature, true);
     }
 
     public void testInequalityTooBigToPush() throws IOException {
@@ -188,11 +203,16 @@ public void testInequalityTooBigToPush() throws IOException {
             | WHERE test != "%value"
             """;
         String luceneQuery = switch (type) {
-            case "text", "auto", "match_only_text" -> "*:*";
+            case "auto", "constant_keyword", "match_only_text", "text" -> "*:*";
             case "semantic_text" -> "FieldExistsQuery [field=_primary_term]";
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        testPushQuery(value, esqlQuery, List.of(luceneQuery), true, false);
+        ComputeSignature dataNodeSignature = switch (type) {
+            case "constant_keyword" -> ComputeSignature.FIND_NONE;
+            case "auto", "match_only_text", "semantic_text", "text" -> ComputeSignature.FILTER_IN_COMPUTE;
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), dataNodeSignature, false);
     }
 
     public void testCaseInsensitiveEquality() throws IOException {
@@ -202,15 +222,49 @@ public void testCaseInsensitiveEquality() throws IOException {
             | WHERE TO_LOWER(test) == "%value"
             """;
         String luceneQuery = switch (type) {
-            case "text", "auto", "match_only_text" -> "*:*";
+            case "auto", "constant_keyword", "match_only_text", "text" -> "*:*";
             case "semantic_text" -> "FieldExistsQuery [field=_primary_term]";
             default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
         };
-        testPushQuery(value, esqlQuery, List.of(luceneQuery), true, true);
+        ComputeSignature dataNodeSignature = switch (type) {
+            case "constant_keyword" -> ComputeSignature.FILTER_IN_QUERY;
+            case "auto", "match_only_text", "semantic_text", "text" -> ComputeSignature.FILTER_IN_COMPUTE;
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), dataNodeSignature, true);
     }
 
-    private void testPushQuery(String value, String esqlQuery, List<String> luceneQueryOptions, boolean filterInCompute, boolean found)
-        throws IOException {
+    enum ComputeSignature {
+        FILTER_IN_COMPUTE(
+            matchesList().item("LuceneSourceOperator")
+                .item("ValuesSourceReaderOperator")
+                .item("FilterOperator")
+                .item("LimitOperator")
+                .item("ProjectOperator")
+                .item("ExchangeSinkOperator")
+        ),
+        FILTER_IN_QUERY(
+            matchesList().item("LuceneSourceOperator")
+                .item("ValuesSourceReaderOperator")
+                .item("ProjectOperator")
+                .item("ExchangeSinkOperator")
+        ),
+        FIND_NONE(matchesList().item("LocalSourceOperator").item("ExchangeSinkOperator"));
+
+        private final ListMatcher matcher;
+
+        ComputeSignature(ListMatcher sig) {
+            this.matcher = sig;
+        }
+    }
+
+    private void testPushQuery(
+        String value,
+        String esqlQuery,
+        List<String> luceneQueryOptions,
+        ComputeSignature dataNodeSignature,
+        boolean found
+    ) throws IOException {
         indexValue(value);
         String differentValue = randomValueOtherThan(value, () -> randomAlphaOfLength(value.isEmpty() ? 1 : value.length()));
 
@@ -226,7 +280,7 @@ private void testPushQuery(String value, String esqlQuery, List<String> luceneQu
                     .entry("planning", matchesMap().extraOk())
                     .entry("query", matchesMap().extraOk())
             ),
-            matchesList().item(matchesMap().entry("name", "test").entry("type", "text")),
+            matchesList().item(matchesMap().entry("name", "test").entry("type", anyOf(equalTo("text"), equalTo("keyword")))),
             equalTo(found ? List.of(List.of(value)) : List.of())
         );
         Matcher<String> luceneQueryMatcher = anyOf(
@@ -250,12 +304,7 @@ private void testPushQuery(String value, String esqlQuery, List<String> luceneQu
             String description = p.get("description").toString();
             switch (description) {
                 case "data" -> {
-                    ListMatcher matcher = matchesList().item("LuceneSourceOperator").item("ValuesSourceReaderOperator");
-                    if (filterInCompute) {
-                        matcher = matcher.item("FilterOperator").item("LimitOperator");
-                    }
-                    matcher = matcher.item("ProjectOperator").item("ExchangeSinkOperator");
-                    assertMap(sig, matcher);
+                    assertMap(sig, dataNodeSignature.matcher);
                 }
                 case "node_reduce" -> {
                     if (sig.contains("LimitOperator")) {
@@ -294,38 +343,10 @@ private void indexValue(String value) throws IOException {
               }""";
         json += switch (type) {
             case "auto" -> "";
-            case "semantic_text" -> """
-                ,
-                "mappings": {
-                  "properties": {
-                    "test": {
-                      "type": "semantic_text",
-                      "inference_id": "test",
-                      "fields": {
-                        "keyword": {
-                          "type": "keyword",
-                          "ignore_above": 256
-                        }
-                      }
-                    }
-                  }
-                }""";
-            default -> """
-                  ,
-                  "mappings": {
-                    "properties": {
-                      "test": {
-                        "type": "%type",
-                        "fields": {
-                          "keyword": {
-                            "type": "keyword",
-                            "ignore_above": 256
-                          }
-                        }
-                      }
-                    }
-                  }
-                }""".replace("%type", type);
+            case "constant_keyword" -> constantKeyword();
+            case "semantic_text" -> semanticText();
+            case "text", "match_only_text" -> subKeyword();
+            default -> throw new UnsupportedOperationException("unsupported type config: " + type);
         };
         json += "}";
         createIndex.setJsonEntity(json);
@@ -345,6 +366,57 @@ private void indexValue(String value) throws IOException {
         assertThat(entityToMap(bulkResponse.getEntity(), XContentType.JSON), matchesMap().entry("errors", false).extraOk());
     }
 
+    private String constantKeyword() {
+        return """
+            ,
+            "mappings": {
+              "properties": {
+                "test": {
+                  "type": "constant_keyword"
+                }
+              }
+            }
+            """;
+    }
+
+    private String semanticText() {
+        return """
+            ,
+            "mappings": {
+              "properties": {
+                "test": {
+                  "type": "semantic_text",
+                  "inference_id": "test",
+                  "fields": {
+                    "keyword": {
+                      "type": "keyword",
+                      "ignore_above": 256
+                    }
+                  }
+                }
+              }
+            }""";
+    }
+
+    private String subKeyword() {
+        return """
+              ,
+              "mappings": {
+                "properties": {
+                  "test": {
+                    "type": "%type",
+                    "fields": {
+                      "keyword": {
+                        "type": "keyword",
+                        "ignore_above": 256
+                      }
+                    }
+                  }
+                }
+              }
+            }""".replace("%type", type);
+    }
+
     private static final Pattern TO_NAME = Pattern.compile("\\[.+", Pattern.DOTALL);
 
     private static String checkOperatorProfile(Map<String, Object> o, Matcher<String> query) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/stats/SearchContextStats.java

@@ -149,6 +149,7 @@ public boolean hasExactSubfield(FieldName field) {
         return cache.computeIfAbsent(field.string(), this::makeFieldStats).config.hasExactSubfield;
     }
 
+    @Override
     public long count() {
         var count = new long[] { 0 };
         boolean completed = doWithContexts(r -> {
@@ -322,10 +323,11 @@ public boolean canUseEqualityOnSyntheticSourceDelegate(FieldAttribute.FieldName
         return true;
     }
 
-    public String constantValue(String name) {
+    @Override
+    public String constantValue(FieldAttribute.FieldName name) {
         String val = null;
         for (SearchExecutionContext ctx : contexts) {
-            MappedFieldType f = ctx.getFieldType(name);
+            MappedFieldType f = ctx.getFieldType(name.string());
             if (f == null) {
                 return null;
             }