ESQL: Fix `NULL` handling in `IN` clause (#125832)

This PR fixes #119950 where an `IN` query includes `NULL` values with non-NULL `DataType` appearing within the query range. An expression is considered `NULL` when its `DataType` is `NULL` or it is a `Literal` with a value of `null`.

docs/changelog/125832.yaml

@@ -0,0 +1,6 @@
+pr: 125832
+summary: "ESQL: Fix `NULL` handling in `IN` clause"
+area: ES|QL
+type: bug
+issues:
+  - 119950

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

@@ -203,3 +203,16 @@ FROM employees
 
 fullname:keyword | job_positions:keyword | salary:integer | salary_change:double
 ;
+
+inConvertedNull
+required_capability: filter_in_converted_null
+FROM employees
+| WHERE emp_no in (10021, 10022, null::int)
+| KEEP emp_no, first_name, last_name
+| SORT emp_no
+;
+
+emp_no:integer | first_name:keyword | last_name:keyword   
+10021          | Ramzi              | Erde
+10022          | Shahaf             | Famili
+;

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

@@ -954,6 +954,12 @@ public class EsqlCapabilities {
          */
         FIX_REPLACE_MISSING_FIELD_WITH_NULL_DUPLICATE_NAME_ID_IN_LAYOUT,
 
+        /**
+         * Support for filter in converted null.
+         * See <a href="https://github.com/elastic/elasticsearch/issues/125832"> ESQL: Fix `NULL` handling in `IN` clause #125832 </a>
+         */
+        FILTER_IN_CONVERTED_NULL,
+
         /**
          * When creating constant null blocks in {@link org.elasticsearch.compute.lucene.ValuesSourceReaderOperator}, we also handed off
          * the ownership of that block - but didn't account for the fact that the caller might close it, leading to double releases

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

@@ -478,7 +478,7 @@ public class In extends EsqlScalarFunction implements TranslationAware.SingleVal
         List<Query> queries = new ArrayList<>();
 
         for (Expression rhs : list()) {
-            if (DataType.isNull(rhs.dataType()) == false) {
+            if (Expressions.isGuaranteedNull(rhs) == false) {
                 if (needsTypeSpecificValueHandling(attribute.dataType())) {
                     // delegates to BinaryComparisons translator to ensure consistent handling of date and time values
                     // TODO:

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

@@ -13,12 +13,16 @@ import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.geo.GeometryTestUtils;
 import org.elasticsearch.geo.ShapeTestUtils;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
+import org.elasticsearch.xpack.esql.core.expression.FieldAttribute;
 import org.elasticsearch.xpack.esql.core.expression.FoldContext;
 import org.elasticsearch.xpack.esql.core.expression.Literal;
+import org.elasticsearch.xpack.esql.core.querydsl.query.TermsQuery;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
+import org.elasticsearch.xpack.esql.core.type.EsField;
 import org.elasticsearch.xpack.esql.expression.function.AbstractFunctionTestCase;
 import org.elasticsearch.xpack.esql.expression.function.TestCaseSupplier;
+import org.elasticsearch.xpack.esql.planner.TranslatorHandler;
 import org.junit.AfterClass;
 
 import java.io.IOException;
@@ -26,6 +30,8 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
+import java.util.Set;
 import java.util.function.Supplier;
 import java.util.stream.IntStream;
 
@@ -81,6 +87,16 @@ public class InTests extends AbstractFunctionTestCase {
         return of(EMPTY, value);
     }
 
+    public void testConvertedNull() {
+        In in = new In(
+            EMPTY,
+            new FieldAttribute(Source.EMPTY, "field", new EsField("suffix", DataType.KEYWORD, Map.of(), true)),
+            Arrays.asList(ONE, new Literal(Source.EMPTY, null, randomFrom(DataType.types())), THREE)
+        );
+        var query = in.asQuery(TranslatorHandler.TRANSLATOR_HANDLER);
+        assertEquals(new TermsQuery(EMPTY, "field", Set.of(1, 3)), query);
+    }
+
     @ParametersFactory
     public static Iterable<Object[]> parameters() {
         List<TestCaseSupplier> suppliers = new ArrayList<>();