ES|QL: fix filtering on TEXT fieds with MV value

docs/changelog/136948.yaml

@@ -0,0 +1,5 @@
+pr: 136948
+summary: Fix filtering on TEXT fieds with MV value
+area: ES|QL
+type: bug
+issues: []

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

@@ -2532,6 +2532,18 @@ warning:Line 2:9: java.lang.IllegalArgumentException: single-value function enco
 2023-10-23T13:55:01.544Z|Connected to 10.1.0.1
 ;
 
+mvStringEqualsMv
+required_capability: fix_text_pushdown_mv
+FROM mv_text
+| WHERE message == ["Connected to 10.1.0.1", "Banana"]
+| KEEP @timestamp, message
+;
+warning:Line 2:9: evaluation of [message == [\"Connected to 10.1.0.1\", \"Banana\"]] 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
+;
+
 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"

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/action/EsqlCapabilities.java

@@ -1560,6 +1560,12 @@ public enum Cap {
         FORBID_LIMIT_BEFORE_INLINE_STATS(INLINE_STATS.enabled),
         // Last capability should still have a comma for fewer merge conflicts when adding new ones :)
         // This comment prevents the semicolon from being on the previous capability when Spotless formats the file.
+
+        /**
+         * https://github.com/elastic/elasticsearch/issues/136939
+         */
+        FIX_TEXT_PUSHDOWN_MV,
+
         ;
 
         private final boolean enabled;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/predicate/operator/comparison/Equals.java

@@ -138,9 +138,9 @@ public Equals(Source source, Expression left, Expression right, ZoneId zoneId) {
 
     @Override
     public Translatable translatable(LucenePushdownPredicates pushdownPredicates) {
-        if (right() instanceof Literal lit) {
+        if (right() instanceof Literal lit && lit.value() instanceof BytesRef bytesRefValue) {
             if (left().dataType() == DataType.TEXT && left() instanceof FieldAttribute fa) {
-                if (pushdownPredicates.canUseEqualityOnSyntheticSourceDelegate(fa, ((BytesRef) lit.value()).utf8ToString())) {
+                if (pushdownPredicates.canUseEqualityOnSyntheticSourceDelegate(fa, bytesRefValue.utf8ToString())) {
                     return Translatable.YES_BUT_RECHECK_NEGATED;
                 }
             }
@@ -150,9 +150,9 @@ public Translatable translatable(LucenePushdownPredicates pushdownPredicates) {
 
     @Override
     public Query asQuery(LucenePushdownPredicates pushdownPredicates, TranslatorHandler handler) {
-        if (right() instanceof Literal lit) {
+        if (right() instanceof Literal lit && lit.value() instanceof BytesRef bytesRefValue) {
             if (left().dataType() == DataType.TEXT && left() instanceof FieldAttribute fa) {
-                String value = ((BytesRef) lit.value()).utf8ToString();
+                String value = bytesRefValue.utf8ToString();
                 if (pushdownPredicates.canUseEqualityOnSyntheticSourceDelegate(fa, value)) {
                     String name = handler.nameOf(fa);
                     return new SingleValueQuery(new EqualsSyntheticSourceDelegate(source(), name, value), name, true);

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/PhysicalPlanOptimizerTests.java

@@ -3052,6 +3052,23 @@ public void testTextWithRawFilterPushDown() {
         assertThat(qb.field(), equalTo("job.raw"));
     }
 
+    // https://github.com/elastic/elasticsearch/issues/136939
+    public void testNoErrorWithTextMvFilter() {
+        var plan = physicalPlan("""
+            from test
+            | where job == ["foo", "bar"]
+            """);
+
+        var optimized = optimizedPlan(plan);
+        var limit = as(optimized, LimitExec.class);
+        var exchange = asRemoteExchange(limit.child());
+        var project = as(exchange.child(), ProjectExec.class);
+        var extract = as(project.child(), FieldExtractExec.class);
+        var source = as(extract.child(), EsQueryExec.class);
+        var qb = as(source.query(), SingleValueQuery.Builder.class);
+        assertThat(qb.field(), equalTo("job.raw"));
+    }
+
     public void testNoTextSortPushDown() {
         var plan = physicalPlan("""
             from test