ESQL: Push more ==s on text fields to lucene (backport) (elastic#128156)

If you do:
```
| WHERE text_field == "cat"
```
we can't push to the text field because it's search index is for
individual words. But most text fields have a `.keyword` sub field and
we *can* query it's index. EXCEPT! It's normal for these fields to have
`ignore_above` in their mapping. In that case we don't push to the
field. Very sad.

With this change we can push down `==`, but only when the right hand
side is shorter than the `ignore_above`.

This has pretty much infinite speed gain. An example using a million
documents:
```
Before:  "took" : 391,
 After:  "took" :   4,
```

But this is going from totally un-indexed linear scans to totally
indexed. You can make the "Before" number as high as you want by loading
more data.



Reenables `text ==` pushdown and adds support for `text !=` pushdown.

It does so by making `TranslationAware#translatable` return something
we can turn into a tri-valued function. It has these values:
* `YES`
* `NO`
* `RECHECK`

`YES` means the `Expression` is entirely pushable into Lucene. They will
be pushed into Lucene and removed from the plan.

`NO` means the `Expression` can't be pushed to Lucene at all and will stay
in the plan.

`RECHECK` mean the `Expression` can push a query that makes *candidate*
matches but must be rechecked. Documents that don't match the query won't
match the expression, but documents that match the query might not match
the expression. These are pushed to Lucene *and* left in the plan.

This is required because `txt != "b"` can build a *candidate* query
against the `txt.keyword` subfield but it can't be sure of the match
without loading the `_source` - which we do in the compute engine.

I haven't plugged rally into this, but here's some basic
performance tests:
```
Before:
not text eq {"took":460,"documents_found":1000000}
    text eq {"took":432,"documents_found":1000000}

After:
    text eq {"took":5,"documents_found":1}
not text eq {"took":351,"documents_found":800000}
```

This comes from:
```
rm -f /tmp/bulk*
for a in {1..1000}; do
    echo '{"index":{}}' >> /tmp/bulk
    echo '{"text":"text '$(printf $(($a % 5)))'"}' >> /tmp/bulk
done
ls -l /tmp/bulk*

passwd="redacted"
curl -sk -uelastic:$passwd -HContent-Type:application/json -XDELETE https://localhost:9200/test
curl -sk -uelastic:$passwd -HContent-Type:application/json -XPUT https://localhost:9200/test -d'{
    "settings": {
        "index.codec": "best_compression",
        "index.refresh_interval": -1
    },
    "mappings": {
        "properties": {
            "many": {
                "enabled": false
            }
        }
    }
}'
for a in {1..1000}; do
    printf %04d: $a
    curl -sk -uelastic:$passwd -HContent-Type:application/json -XPOST https://localhost:9200/test/_bulk?pretty --data-binary @/tmp/bulk | grep errors
done
curl -sk -uelastic:$passwd -HContent-Type:application/json -XPOST https://localhost:9200/test/_forcemerge?max_num_segments=1
curl -sk -uelastic:$passwd -HContent-Type:application/json -XPOST https://localhost:9200/test/_refresh
echo
curl -sk -uelastic:$passwd https://localhost:9200/_cat/indices?v

text_eq() {
    echo -n "    text eq "
    curl -sk -uelastic:$passwd -HContent-Type:application/json -XPOST 'https://localhost:9200/_query?pretty' -d'{
        "query": "FROM test | WHERE text == \"text 1\" | STATS COUNT(*)",
        "pragma": {
            "data_partitioning": "shard"
        }
    }' | jq -c '{took, documents_found}'
}

not_text_eq() {
    echo -n "not text eq "
    curl -sk -uelastic:$passwd -HContent-Type:application/json -XPOST 'https://localhost:9200/_query?pretty' -d'{
        "query": "FROM test | WHERE NOT text == \"text 1\" | STATS COUNT(*)",
        "pragma": {
            "data_partitioning": "shard"
        }
    }' | jq -c '{took, documents_found}'
}

for a in {1..100}; do
    text_eq
    not_text_eq
done
```

Author: nik9000

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: []

docs/changelog/127199.yaml

@@ -0,0 +1,6 @@
+pr: 127199
+summary: Disable a bugged commit
+area: ES|QL
+type: bug
+issues:
+ - 127197

docs/changelog/127355.yaml

@@ -0,0 +1,5 @@
+pr: 127355
+summary: '`text ==` and `text !=` pushdown'
+area: ES|QL
+type: enhancement
+issues: []

docs/changelog/128156.yaml

@@ -0,0 +1,5 @@
+pr: 128156
+summary: Push more ==s on text fields to lucene (backport)
+area: ES|QL
+type: feature
+issues: []

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

@@ -986,6 +986,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,344 @@
+/*
+ * 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.ParametersFactory;
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakFilters;
+
+import org.elasticsearch.client.Request;
+import org.elasticsearch.client.Response;
+import org.elasticsearch.client.ResponseException;
+import org.elasticsearch.common.collect.Iterators;
+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.hamcrest.Matcher;
+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 java.util.stream.Stream;
+
+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.anyOf;
+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();
+
+    @ParametersFactory(argumentFormatting = "%1s")
+    public static List<Object[]> args() {
+        return Stream.of("auto", "text", "match_only_text", "semantic_text").map(s -> new Object[] { s }).toList();
+    }
+
+    private final String type;
+
+    public PushQueriesIT(String type) {
+        this.type = type;
+    }
+
+    public void testEquality() throws IOException {
+        String value = "v".repeat(between(0, 256));
+        String esqlQuery = """
+            FROM test
+            | WHERE test == "%value"
+            """;
+        String luceneQuery = switch (type) {
+            case "text", "auto" -> "#test.keyword:%value -_ignored:test.keyword";
+            case "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;
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), filterInCompute, true);
+    }
+
+    public void testEqualityTooBigToPush() throws IOException {
+        String value = "a".repeat(between(257, 1000));
+        String esqlQuery = """
+            FROM test
+            | WHERE test == "%value"
+            """;
+        String luceneQuery = switch (type) {
+            case "text", "auto", "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);
+    }
+
+    /**
+     * Turns into an {@code IN} which isn't currently pushed.
+     */
+    public void testEqualityOrTooBig() throws IOException {
+        String value = "v".repeat(between(0, 256));
+        String tooBig = "a".repeat(between(257, 1000));
+        String esqlQuery = """
+            FROM test
+            | WHERE test == "%value" OR test == "%tooBig"
+            """.replace("%tooBig", tooBig);
+        String luceneQuery = switch (type) {
+            case "text", "auto", "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);
+    }
+
+    public void testEqualityOrOther() throws IOException {
+        String value = "v".repeat(between(0, 256));
+        String esqlQuery = """
+            FROM test
+            | 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 "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;
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), filterInCompute, true);
+    }
+
+    public void testEqualityAndOther() throws IOException {
+        String value = "v".repeat(between(0, 256));
+        String esqlQuery = """
+            FROM test
+            | 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 "semantic_text" ->
+                /*
+                 * single_value_match is here because there are extra documents hiding in the index
+                 * that don't have the `foo` field.
+                 */
+                List.of("#foo:[1 TO 1] #single_value_match(foo)", "foo:[1 TO 1]");
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        boolean filterInCompute = switch (type) {
+            case "text", "auto" -> false;
+            case "match_only_text", "semantic_text" -> true;
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, luceneQueryOptions, filterInCompute, true);
+    }
+
+    public void testInequality() throws IOException {
+        String value = "v".repeat(between(0, 256));
+        String esqlQuery = """
+            FROM test
+            | WHERE test != "%different_value"
+            """;
+        String luceneQuery = switch (type) {
+            case "text", "auto" -> "(-test.keyword:%different_value #*:*) _ignored:test.keyword";
+            case "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);
+    }
+
+    public void testInequalityTooBigToPush() throws IOException {
+        String value = "a".repeat(between(257, 1000));
+        String esqlQuery = """
+            FROM test
+            | WHERE test != "%value"
+            """;
+        String luceneQuery = switch (type) {
+            case "text", "auto", "match_only_text" -> "*:*";
+            case "semantic_text" -> "FieldExistsQuery [field=_primary_term]";
+            default -> throw new UnsupportedOperationException("unknown type [" + type + "]");
+        };
+        testPushQuery(value, esqlQuery, List.of(luceneQuery), true, false);
+    }
+
+    public void testCaseInsensitiveEquality() throws IOException {
+        String value = "a".repeat(between(0, 256));
+        String esqlQuery = """
+            FROM test
+            | WHERE TO_LOWER(test) == "%value"
+            """;
+        String luceneQuery = switch (type) {
+            case "text", "auto", "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);
+    }
+
+    private void testPushQuery(String value, String esqlQuery, List<String> luceneQueryOptions, boolean filterInCompute, boolean found)
+        throws IOException {
+        indexValue(value);
+        String differentValue = randomValueOtherThan(value, () -> randomAlphaOfLength(value.isEmpty() ? 1 : value.length()));
+
+        String replacedQuery = esqlQuery.replaceAll("%value", value).replaceAll("%different_value", differentValue);
+        RestEsqlTestCase.RequestObjectBuilder builder = requestObjectBuilder().query(replacedQuery + "\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))),
+            matchesList().item(matchesMap().entry("name", "test").entry("type", "text")),
+            equalTo(found ? List.of(List.of(value)) : List.of())
+        );
+        Matcher<String> luceneQueryMatcher = anyOf(
+            () -> Iterators.map(
+                luceneQueryOptions.iterator(),
+                (String s) -> equalTo(s.replaceAll("%value", value).replaceAll("%different_value", differentValue))
+            )
+        );
+
+        @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, luceneQueryMatcher));
+            }
+            String description = p.get("task_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" -> {
+                    if (sig.contains("LimitOperator")) {
+                        // TODO figure out why this is sometimes here and sometimes not
+                        assertMap(sig, matchesList().item("ExchangeSourceOperator").item("LimitOperator").item("ExchangeSinkOperator"));
+                    } else {
+                        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 {
+        try {
+            // Delete the index if it has already been created.
+            client().performRequest(new Request("DELETE", "test"));
+        } catch (ResponseException e) {
+            if (e.getResponse().getStatusLine().getStatusCode() != 404) {
+                throw e;
+            }
+        }
+
+        Request createIndex = new Request("PUT", "test");
+        String json = """
+            {
+              "settings": {
+                "index": {
+                  "number_of_shards": 1
+                }
+              }""";
+        if (false == "auto".equals(type)) {
+            json += """
+                ,
+                "mappings": {
+                  "properties": {
+                    "test": {
+                      "type": "%type",
+                        "fields": {
+                          "keyword": {
+                            "type": "keyword",
+                            "ignore_above": 256
+                          }
+                        }
+                      }
+                    }
+                  }
+                }""".replace("%type", type);
+        }
+        json += "}";
+        createIndex.setJsonEntity(json);
+        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","foo":1}
+            """, 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, Matcher<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();
+    }
+
+    @Override
+    protected boolean preserveClusterUponCompletion() {
+        // Preserve the cluser to speed up the semantic_text tests
+        return true;
+    }
+}