ESQL: Push more ==s on text fields to lucene

docs/changelog/126641.yaml

@@ -0,0 +1,5 @@
+pr: 126641
+summary: Push more `==`s on text fields to lucene
+area: ES|QL
+type: enhancement
+issues: []

server/src/main/java/org/elasticsearch/index/mapper/TextFieldMapper.java

@@ -983,6 +983,21 @@ public boolean canUseSyntheticSourceDelegateForQuerying() {
                 && syntheticSourceDelegate.isIndexed();
         }
 
+        /**
+         * Returns true if the delegate sub-field can be used for querying only (ie. isIndexed must be true)
+         */
+        public boolean canUseSyntheticSourceDelegateForQueryingEquality(String str) {
+            if (syntheticSourceDelegate == null
+                // Can't push equality to an index if there isn't an index
+                || syntheticSourceDelegate.isIndexed() == false
+                // ESQL needs docs values to push equality
+                || syntheticSourceDelegate.hasDocValues() == false) {
+                return false;
+            }
+            // Can't push equality if the field we're checking for is so big we'd ignore it.
+            return str.length() <= syntheticSourceDelegate.ignoreAbove();
+        }
+
         @Override
         public BlockLoader blockLoader(BlockLoaderContext blContext) {
             if (canUseSyntheticSourceDelegateForLoading()) {

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

@@ -0,0 +1,168 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.qa.single_node;
+
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.test.ListMatcher;
+import org.elasticsearch.test.MapMatcher;
+import org.elasticsearch.test.TestClustersThreadFilter;
+import org.elasticsearch.test.cluster.ElasticsearchCluster;
+import org.elasticsearch.test.rest.ESRestTestCase;
+import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xpack.esql.AssertWarnings;
+import org.elasticsearch.xpack.esql.qa.rest.RestEsqlTestCase;
+import org.junit.ClassRule;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.regex.Pattern;
+
+import static org.elasticsearch.test.ListMatcher.matchesList;
+import static org.elasticsearch.test.MapMatcher.assertMap;
+import static org.elasticsearch.test.MapMatcher.matchesMap;
+import static org.elasticsearch.xpack.esql.EsqlTestUtils.entityToMap;
+import static org.elasticsearch.xpack.esql.qa.rest.RestEsqlTestCase.requestObjectBuilder;
+import static org.elasticsearch.xpack.esql.qa.rest.RestEsqlTestCase.runEsql;
+import static org.elasticsearch.xpack.esql.qa.single_node.RestEsqlIT.commonProfile;
+import static org.elasticsearch.xpack.esql.qa.single_node.RestEsqlIT.fixTypesOnProfile;
+import static org.hamcrest.Matchers.equalTo;
+import static org.hamcrest.Matchers.instanceOf;
+import static org.hamcrest.Matchers.startsWith;
+
+/**
+ * Tests for pushing queries to lucene.
+ */
+@ThreadLeakFilters(filters = TestClustersThreadFilter.class)
+public class PushQueriesIT extends ESRestTestCase {
+    @ClassRule
+    public static ElasticsearchCluster cluster = Clusters.testCluster();
+
+    public void testPushEqualityOnDefaults() throws IOException {
+        String value = "v".repeat(between(0, 256));
+        testPushQuery(value, """
+            FROM test
+            | WHERE test == "%value"
+            """, "#test.keyword:%value -_ignored:test.keyword", false);
+    }
+
+    public void testPushEqualityOnDefaultsTooBigToPush() throws IOException {
+        String value = "a".repeat(between(257, 1000));
+        testPushQuery(value, """
+            FROM test
+            | WHERE test == "%value"
+            """, "*:*", true);
+    }
+
+    public void testPushCaseInsensitiveEqualityOnDefaults() throws IOException {
+        String value = "a".repeat(between(0, 256));
+        testPushQuery(value, """
+            FROM test
+            | WHERE TO_LOWER(test) == "%value"
+            """, "*:*", true);
+    }
+
+    private void testPushQuery(String value, String esqlQuery, String luceneQuery, boolean filterInCompute) throws IOException {
+        indexValue(value);
+
+        RestEsqlTestCase.RequestObjectBuilder builder = requestObjectBuilder().query(
+            esqlQuery.replaceAll("%value", value) + "\n| KEEP test"
+        );
+        builder.profile(true);
+        Map<String, Object> result = runEsql(builder, new AssertWarnings.NoWarnings(), RestEsqlTestCase.Mode.SYNC);
+        assertResultMap(
+            result,
+            getResultMatcher(result).entry(
+                "profile",
+                matchesMap().entry("drivers", instanceOf(List.class))
+                    .entry("planning", matchesMap().extraOk())
+                    .entry("query", matchesMap().extraOk())
+            ),
+            matchesList().item(matchesMap().entry("name", "test").entry("type", "text")),
+            equalTo(List.of(List.of(value)))
+        );
+
+        @SuppressWarnings("unchecked")
+        List<Map<String, Object>> profiles = (List<Map<String, Object>>) ((Map<String, Object>) result.get("profile")).get("drivers");
+        for (Map<String, Object> p : profiles) {
+            fixTypesOnProfile(p);
+            assertThat(p, commonProfile());
+            List<String> sig = new ArrayList<>();
+            @SuppressWarnings("unchecked")
+            List<Map<String, Object>> operators = (List<Map<String, Object>>) p.get("operators");
+            for (Map<String, Object> o : operators) {
+                sig.add(checkOperatorProfile(o, luceneQuery.replaceAll("%value", value)));
+            }
+            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);
+                }
+                case "node_reduce" -> assertMap(sig, matchesList().item("ExchangeSourceOperator").item("ExchangeSinkOperator"));
+                case "final" -> assertMap(
+                    sig,
+                    matchesList().item("ExchangeSourceOperator").item("LimitOperator").item("ProjectOperator").item("OutputOperator")
+                );
+                default -> throw new IllegalArgumentException("can't match " + description);
+            }
+        }
+    }
+
+    private void indexValue(String value) throws IOException {
+        Request createIndex = new Request("PUT", "test");
+        createIndex.setJsonEntity("""
+            {
+              "settings": {
+                "index": {
+                  "number_of_shards": 1
+                }
+              }
+            }""");
+        Response createResponse = client().performRequest(createIndex);
+        assertThat(
+            entityToMap(createResponse.getEntity(), XContentType.JSON),
+            matchesMap().entry("shards_acknowledged", true).entry("index", "test").entry("acknowledged", true)
+        );
+
+        Request bulk = new Request("POST", "/_bulk");
+        bulk.addParameter("refresh", "");
+        bulk.setJsonEntity(String.format("""
+            {"create":{"_index":"test"}}
+            {"test":"%s"}
+            """, value));
+        Response bulkResponse = client().performRequest(bulk);
+        assertThat(entityToMap(bulkResponse.getEntity(), XContentType.JSON), matchesMap().entry("errors", false).extraOk());
+    }
+
+    private static final Pattern TO_NAME = Pattern.compile("\\[.+", Pattern.DOTALL);
+
+    private static String checkOperatorProfile(Map<String, Object> o, String query) {
+        String name = (String) o.get("operator");
+        name = TO_NAME.matcher(name).replaceAll("");
+        if (name.equals("LuceneSourceOperator")) {
+            MapMatcher expectedOp = matchesMap().entry("operator", startsWith(name))
+                .entry("status", matchesMap().entry("processed_queries", List.of(query)).extraOk());
+            assertMap(o, expectedOp);
+        }
+        return name;
+    }
+
+    @Override
+    protected String getTestRestCluster() {
+        return cluster.getHttpAddresses();
+    }
+}

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

@@ -648,7 +648,7 @@ public void testForceSleepsProfile() throws IOException {
         }
     }
 
-    private MapMatcher commonProfile() {
+    static MapMatcher commonProfile() {
         return matchesMap() //
             .entry("description", any(String.class))
             .entry("cluster_name", any(String.class))
@@ -669,7 +669,7 @@ private MapMatcher commonProfile() {
      * come back as integers and sometimes longs. This just promotes
      * them to long every time.
      */
-    private void fixTypesOnProfile(Map<String, Object> profile) {
+    static void fixTypesOnProfile(Map<String, Object> profile) {
         profile.put("iterations", ((Number) profile.get("iterations")).longValue());
         profile.put("cpu_nanos", ((Number) profile.get("cpu_nanos")).longValue());
         profile.put("took_nanos", ((Number) profile.get("took_nanos")).longValue());

x-pack/plugin/esql/qa/server/src/main/java/org/elasticsearch/xpack/esql/qa/rest/RestEsqlTestCase.java

@@ -1235,7 +1235,8 @@ public static Map<String, Object> runEsqlAsync(RequestObjectBuilder requestObjec
         return runEsqlAsync(requestObject, randomBoolean(), new AssertWarnings.NoWarnings());
     }
 
-    static Map<String, Object> runEsql(RequestObjectBuilder requestObject, AssertWarnings assertWarnings, Mode mode) throws IOException {
+    public static Map<String, Object> runEsql(RequestObjectBuilder requestObject, AssertWarnings assertWarnings, Mode mode)
+        throws IOException {
         if (mode == ASYNC) {
             return runEsqlAsync(requestObject, randomBoolean(), assertWarnings);
         } else {

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/CsvTestsDataLoader.java

@@ -139,6 +139,7 @@ public class CsvTestsDataLoader {
     private static final TestDataset BOOKS = new TestDataset("books").withSetting("books-settings.json");
     private static final TestDataset SEMANTIC_TEXT = new TestDataset("semantic_text").withInferenceEndpoint(true);
     private static final TestDataset LOGS = new TestDataset("logs");
+    private static final TestDataset MV_TEXT = new TestDataset("mv_text");
 
     public static final Map<String, TestDataset> CSV_DATASET_MAP = Map.ofEntries(
         Map.entry(EMPLOYEES.indexName, EMPLOYEES),
@@ -196,7 +197,8 @@ public class CsvTestsDataLoader {
         Map.entry(ADDRESSES.indexName, ADDRESSES),
         Map.entry(BOOKS.indexName, BOOKS),
         Map.entry(SEMANTIC_TEXT.indexName, SEMANTIC_TEXT),
-        Map.entry(LOGS.indexName, LOGS)
+        Map.entry(LOGS.indexName, LOGS),
+        Map.entry(MV_TEXT.indexName, MV_TEXT)
     );
 
     private static final EnrichConfig LANGUAGES_ENRICH = new EnrichConfig("languages_policy", "enrich-policy-languages.json");

x-pack/plugin/esql/qa/testFixtures/src/main/java/org/elasticsearch/xpack/esql/EsqlTestUtils.java

@@ -294,6 +294,11 @@ public byte[] max(String field, DataType dataType) {
         public boolean isSingleValue(String field) {
             return false;
         }
+
+        @Override
+        public boolean canUseEqualityOnSyntheticSourceDelegate(String name, String value) {
+            return false;
+        }
     }
 
     /**

x-pack/plugin/esql/qa/testFixtures/src/main/resources/data/mv_text.csv

@@ -0,0 +1,5 @@
+@timestamp:date         ,message:text
+2023-10-23T13:55:01.543Z,[Connected to 10.1.0.1, Banana]
+2023-10-23T13:55:01.544Z,Connected to 10.1.0.1
+2023-10-23T13:55:01.545Z,[Connected to 10.1.0.1, More than one hundred characters long so it isn't indexed by the sub keyword field with ignore_above:100]
+2023-10-23T13:55:01.546Z,More than one hundred characters long so it isn't indexed by the sub keyword field with ignore_above:100

x-pack/plugin/esql/qa/testFixtures/src/main/resources/mapping-basic-limited-raw.json

@@ -0,0 +1,42 @@
+{
+    "properties" : {
+        "emp_no" : {
+            "type" : "integer"
+        },
+        "first_name" : {
+            "type" : "keyword"
+        },
+        "gender" : {
+            "type" : "text"
+        },
+        "languages" : {
+            "type" : "byte"
+        },
+        "last_name" : {
+            "type" : "keyword"
+        },
+        "salary" : {
+            "type" : "integer"
+        },
+        "_meta_field": {
+            "type" : "keyword"
+        },
+        "hire_date": {
+          "type": "date"
+        },
+        "job": {
+            "type": "text",
+            "fields": {
+                "raw": {
+                  "type": "keyword",
+                  "ignore_above": 4
+                }
+            }
+        },
+        "long_noidx": {
+          "type": "long",
+          "index": false,
+          "doc_values": false
+        }
+    }
+}

x-pack/plugin/esql/qa/testFixtures/src/main/resources/mapping-mv_text.json

@@ -0,0 +1,16 @@
+{
+    "properties": {
+        "@timestamp": {
+            "type": "date"
+        },
+        "message": {
+            "type": "text",
+            "fields": {
+                "raw": {
+                    "type": "keyword",
+                    "ignore_above": 100
+                }
+            }
+        }
+    }
+}

x-pack/plugin/esql/qa/testFixtures/src/main/resources/string.csv-spec

@@ -2308,3 +2308,27 @@ message:keyword
 foo ( bar
 // end::rlikeEscapingTripleQuotes-result[]
 ;
+
+mvStringEquals
+FROM mv_text
+| WHERE message == "Connected to 10.1.0.1"
+| KEEP @timestamp, message
+;
+warning:Line 2:9: evaluation of [message == \"Connected to 10.1.0.1\"] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 2:9: java.lang.IllegalArgumentException: single-value function encountered multi-value
+
+    @timestamp:date     | message:text
+2023-10-23T13:55:01.544Z|Connected to 10.1.0.1
+;
+
+mvStringEqualsLongString
+FROM mv_text
+| WHERE message == "More than one hundred characters long so it isn't indexed by the sub keyword field with ignore_above:100"
+| KEEP @timestamp, message
+;
+warning:Line 2:9: evaluation of [message == \"More than one hundred characters long so it isn't indexed by the sub keyword field with ignore_above:100\"] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 2:9: java.lang.IllegalArgumentException: single-value function encountered multi-value
+
+    @timestamp:date     | message:text
+2023-10-23T13:55:01.546Z|More than one hundred characters long so it isn't indexed by the sub keyword field with ignore_above:100
+;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/capabilities/TranslationAware.java

@@ -31,7 +31,7 @@ public interface TranslationAware {
      * <p>and <b>not</b> this:</p>
      * <p>{@code Query childQuery = child.asQuery(handler);}</p>
      */
-    Query asQuery(TranslatorHandler handler);
+    Query asQuery(LucenePushdownPredicates pushdownPredicates, TranslatorHandler handler);
 
     /**
      * Subinterface for expressions that can only process single values (and null out on MVs).

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/FullTextFunction.java

@@ -162,7 +162,7 @@ public boolean translatable(LucenePushdownPredicates pushdownPredicates) {
     }
 
     @Override
-    public Query asQuery(TranslatorHandler handler) {
+    public Query asQuery(LucenePushdownPredicates pushdownPredicates, TranslatorHandler handler) {
         return queryBuilder != null ? new TranslationAwareExpressionQuery(source(), queryBuilder) : translate(handler);
     }
 

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/fulltext/QueryBuilderResolver.java

@@ -13,6 +13,7 @@
 import org.elasticsearch.index.query.QueryRewriteContext;
 import org.elasticsearch.index.query.Rewriteable;
 import org.elasticsearch.xpack.esql.core.util.Holder;
+import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates;
 import org.elasticsearch.xpack.esql.plan.logical.EsRelation;
 import org.elasticsearch.xpack.esql.plan.logical.LogicalPlan;
 import org.elasticsearch.xpack.esql.planner.TranslatorHandler;
@@ -76,7 +77,9 @@ public FullTextFunctionsRewritable rewrite(QueryRewriteContext ctx) throws IOExc
             Holder<Boolean> updated = new Holder<>(false);
             LogicalPlan newPlan = plan.transformExpressionsDown(FullTextFunction.class, f -> {
                 QueryBuilder builder = f.queryBuilder(), initial = builder;
-                builder = builder == null ? f.asQuery(TranslatorHandler.TRANSLATOR_HANDLER).toQueryBuilder() : builder;
+                builder = builder == null
+                    ? f.asQuery(LucenePushdownPredicates.DEFAULT, TranslatorHandler.TRANSLATOR_HANDLER).toQueryBuilder()
+                    : builder;
                 try {
                     builder = builder.rewrite(ctx);
                 } catch (IOException e) {

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/scalar/ip/CIDRMatch.java

@@ -184,7 +184,7 @@ public boolean translatable(LucenePushdownPredicates pushdownPredicates) {
     }
 
     @Override
-    public Query asQuery(TranslatorHandler handler) {
+    public Query asQuery(LucenePushdownPredicates pushdownPredicates, TranslatorHandler handler) {
         var fa = LucenePushdownPredicates.checkIsFieldAttribute(ipField);
         Check.isTrue(Expressions.foldable(matches), "Expected foldable matches, but got [{}]", matches);