ESQL: Fix functions emitting warnings with no source (#122821)

Fixes https://github.com/elastic/elasticsearch/issues/122588

- Replaced `Source.EMPTY.writeTo(out)` to `source().writeTo(out)` in functions emitting warnings
  - Did the same on all aggs, as Top emits an error on type resolution. This is not a bug, as type resolution errors should only happen in the coordinator. Another option would be changing Top to not generate that error there, and make it implement instead `PostAnalysisVerificationAware`
- In some cases, we don't even serialize an empty source. So I had to add a new `TransportVersion` to do so
  - As an special case, `ToLower` and `ToUpper` weren't serializing a source, but they don't emit warnings. As they were the only remaining functions not serializing the source, I added it there too

docs/changelog/122821.yaml

@@ -0,0 +1,6 @@
+pr: 122821
+summary: Fix functions emitting warnings with no source
+area: ES|QL
+type: bug
+issues:
+ - 122588

server/src/main/java/org/elasticsearch/TransportVersions.java

@@ -203,6 +203,7 @@ public class TransportVersions {
     public static final TransportVersion RERANKER_FAILURES_ALLOWED = def(9_013_0_00);
     public static final TransportVersion VOYAGE_AI_INTEGRATION_ADDED = def(9_014_0_00);
     public static final TransportVersion BYTE_SIZE_VALUE_ALWAYS_USES_BYTES = def(9_015_0_00);
+    public static final TransportVersion ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS = def(9_016_0_00);
 
     /*
      * STOP! READ THIS FIRST! No, really,

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

@@ -287,6 +287,26 @@ emp_no:integer | is_rehired:boolean       | a1:boolean
 10005          | [false,false,false,true] | false
 ;
 
+mvSliceWarnings
+required_capability: functions_source_serialization_warnings
+
+FROM employees
+| SORT first_name ASC
+| EVAL
+    start = CASE(first_name == "Alejandro", 1, 0),
+    end = CASE(first_name == "Alejandro", 0, 1),
+    result = MV_SLICE(is_rehired, start, end)
+| KEEP first_name, result
+| LIMIT 1
+;
+
+warning:Line 6:10: evaluation of [MV_SLICE(is_rehired, start, end)] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 6:10: org.elasticsearch.xpack.esql.core.InvalidArgumentException: Start offset is greater than end offset
+
+first_name:keyword | result:boolean
+Alejandro          | null
+;
+
 values
 required_capability: agg_values
 

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

@@ -366,6 +366,27 @@ date1:date               | dd_ms:integer
 2023-12-02T11:00:00.000Z | 1000
 ;
 
+dateDiffTestWarnings
+required_capability: functions_source_serialization_warnings
+
+FROM employees
+| WHERE first_name IN ("Alejandro", "Mary")
+| SORT first_name ASC
+| EVAL date = TO_DATETIME("2023-12-02T11:00:00.000Z")
+| EVAL dd_ms = DATE_DIFF(first_name, date, date)
+| KEEP date, dd_ms
+| LIMIT 2
+;
+
+warning:Line 5:16: evaluation of [DATE_DIFF(first_name, date, date)] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 5:16: java.lang.IllegalArgumentException: A value of [YEAR, QUARTER, MONTH, DAYOFYEAR, DAY, WEEK, WEEKDAY, HOUR, MINUTE, SECOND, MILLISECOND, MICROSECOND, NANOSECOND] or their aliases is required; received [Alejandro]
+warning:Line 5:16: java.lang.IllegalArgumentException: Received value [Mary] is not valid date part to add; did you mean [day]? 
+
+date:date                | dd_ms:integer
+2023-12-02T11:00:00.000Z | null
+2023-12-02T11:00:00.000Z | null
+;
+
 evalDateDiffMonthAsWhole0Months#[skip:-8.14.1, reason:omitting millis/timezone not allowed before 8.14]
 
 ROW from=TO_DATETIME("2023-12-31T23:59:59.999Z"), to=TO_DATETIME("2024-01-01T00:00:00")

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

@@ -1009,6 +1009,33 @@ mvsum:unsigned_long
 null
 ;
 
+mvSumOverflowOnNode
+required_capability: functions_source_serialization_warnings
+
+FROM employees
+| SORT first_name ASC
+| EVAL
+    ints = CASE(first_name == "Alejandro", [0, 1, 2147483647], 0),
+    longs = CASE(first_name == "Alejandro", [0, 1, 9223372036854775807], 0),
+    ulongs = CASE(first_name == "Alejandro", [0, 1, 18446744073709551615], 0),
+    intsResult = mv_sum(ints),
+    longsResult = mv_sum(longs),
+    ulongsResult = mv_sum(ulongs)
+| KEEP first_name, intsResult, longsResult, ulongsResult
+| LIMIT 1
+;
+
+warning:Line 7:14: evaluation of [mv_sum(ints)] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 7:14: java.lang.ArithmeticException: integer overflow
+warning:Line 8:15: evaluation of [mv_sum(longs)] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 8:15: java.lang.ArithmeticException: long overflow
+warning:Line 9:16: evaluation of [mv_sum(ulongs)] failed, treating result as null. Only first 20 failures recorded.
+warning:Line 9:16: java.lang.ArithmeticException: unsigned_long overflow
+
+first_name:keyword | intsResult:integer | longsResult:long | ulongsResult:unsigned_long
+Alejandro          | null               | null             | null
+;
+
 e
 // tag::e[]
 ROW E()

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

@@ -1068,6 +1068,25 @@ Gatewayinstances|Gateway instances
 null            |null             
 ;
 
+replaceWarnings
+required_capability: functions_source_serialization_warnings
+
+FROM employees
+| SORT first_name ASC
+| EVAL
+    regex = CASE(first_name == "Alejandro", "(", ""),
+    result = replace(first_name, regex, "")
+| KEEP first_name, result
+| LIMIT 1
+;
+
+warningRegex:Line 5:10: evaluation of \[replace\(first_name, regex, \\"\\"\)\] failed, treating result as null. Only first 20 failures recorded.
+warningRegex:Line 5:10: java.util.regex.PatternSyntaxException: Unclosed group near index 1(%0D)?%0A\(
+
+first_name:keyword | result:keyword
+Alejandro          | null
+;
+
 left
 // tag::left[]
 FROM employees

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

@@ -204,6 +204,11 @@ public class EsqlCapabilities {
          */
         FN_ROUND_UL_FIXES,
 
+        /**
+         * Fixes for multiple functions not serializing their source, and emitting warnings with wrong line number and text.
+         */
+        FUNCTIONS_SOURCE_SERIALIZATION_WARNINGS,
+
         /**
          * All functions that take TEXT should never emit TEXT, only KEYWORD. #114334
          */

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/expression/function/aggregate/AggregateFunction.java

@@ -69,7 +69,7 @@ public abstract class AggregateFunction extends Function implements PostAnalysis
 
     @Override
     public final void writeTo(StreamOutput out) throws IOException {
-        Source.EMPTY.writeTo(out);
+        source().writeTo(out);
         out.writeNamedWriteable(field);
         if (out.getTransportVersion().onOrAfter(TransportVersions.V_8_16_0)) {
             out.writeNamedWriteable(filter);

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

@@ -197,7 +197,7 @@ public class DateDiff extends EsqlScalarFunction {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        Source.EMPTY.writeTo(out);
+        source().writeTo(out);
         out.writeNamedWriteable(unit);
         out.writeNamedWriteable(startTimestamp);
         out.writeNamedWriteable(endTimestamp);

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

@@ -41,7 +41,7 @@ public abstract class AbstractMultivalueFunction extends UnaryScalarFunction {
 
     @Override
     public final void writeTo(StreamOutput out) throws IOException {
-        Source.EMPTY.writeTo(out);
+        source().writeTo(out);
         out.writeNamedWriteable(field);
     }
 

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

@@ -132,7 +132,7 @@ public class MvSlice extends EsqlScalarFunction implements OptionalArgument, Eva
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
-        Source.EMPTY.writeTo(out);
+        source().writeTo(out);
         out.writeNamedWriteable(field);
         out.writeNamedWriteable(start);
         out.writeOptionalNamedWriteable(end);

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

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.expression.function.scalar.spatial;
 
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.geometry.Geometry;
@@ -20,6 +21,7 @@ import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
 import org.elasticsearch.xpack.esql.core.util.SpatialCoordinateTypes;
 import org.elasticsearch.xpack.esql.expression.EsqlTypeResolutions;
+import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 import org.elasticsearch.xpack.esql.optimizer.rules.physical.local.LucenePushdownPredicates;
 
 import java.io.IOException;
@@ -63,7 +65,9 @@ public abstract class BinarySpatialFunction extends BinaryScalarFunction impleme
     protected BinarySpatialFunction(StreamInput in, boolean leftDocValues, boolean rightDocValues, boolean pointsOnly) throws IOException {
         // The doc-values fields are only used on data nodes local planning, and therefor never serialized
         this(
-            Source.EMPTY,
+            in.getTransportVersion().onOrAfter(TransportVersions.ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS)
+                ? Source.readFrom((PlanStreamInput) in)
+                : Source.EMPTY,
             in.readNamedWriteable(Expression.class),
             in.readNamedWriteable(Expression.class),
             leftDocValues,
@@ -74,6 +78,9 @@ public abstract class BinarySpatialFunction extends BinaryScalarFunction impleme
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS)) {
+            source().writeTo(out);
+        }
         out.writeNamedWriteable(left());
         out.writeNamedWriteable(right());
         // The doc-values fields are only used on data nodes local planning, and therefor never serialized

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

@@ -8,6 +8,7 @@
 package org.elasticsearch.xpack.esql.expression.function.scalar.string;
 
 import org.apache.lucene.util.BytesRef;
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -22,6 +23,7 @@ import org.elasticsearch.xpack.esql.expression.function.Example;
 import org.elasticsearch.xpack.esql.expression.function.FunctionInfo;
 import org.elasticsearch.xpack.esql.expression.function.Param;
 import org.elasticsearch.xpack.esql.expression.function.scalar.EsqlScalarFunction;
+import org.elasticsearch.xpack.esql.io.stream.PlanStreamInput;
 
 import java.io.IOException;
 import java.util.Arrays;
@@ -66,7 +68,9 @@ public class Replace extends EsqlScalarFunction {
 
     private Replace(StreamInput in) throws IOException {
         this(
-            Source.EMPTY,
+            in.getTransportVersion().onOrAfter(TransportVersions.ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS)
+                ? Source.readFrom((PlanStreamInput) in)
+                : Source.EMPTY,
             in.readNamedWriteable(Expression.class),
             in.readNamedWriteable(Expression.class),
             in.readNamedWriteable(Expression.class)
@@ -75,6 +79,9 @@ public class Replace extends EsqlScalarFunction {
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS)) {
+            source().writeTo(out);
+        }
         out.writeNamedWriteable(str);
         out.writeNamedWriteable(regex);
         out.writeNamedWriteable(newStr);

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

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.expression.function.scalar.string;
 
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -42,11 +43,20 @@ public class ToLower extends ChangeCase {
     }
 
     private ToLower(StreamInput in) throws IOException {
-        this(Source.EMPTY, in.readNamedWriteable(Expression.class), ((PlanStreamInput) in).configuration());
+        this(
+            in.getTransportVersion().onOrAfter(TransportVersions.ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS)
+                ? Source.readFrom((PlanStreamInput) in)
+                : Source.EMPTY,
+            in.readNamedWriteable(Expression.class),
+            ((PlanStreamInput) in).configuration()
+        );
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS)) {
+            source().writeTo(out);
+        }
         out.writeNamedWriteable(field());
     }
 

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

@@ -7,6 +7,7 @@
 
 package org.elasticsearch.xpack.esql.expression.function.scalar.string;
 
+import org.elasticsearch.TransportVersions;
 import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -42,11 +43,20 @@ public class ToUpper extends ChangeCase {
     }
 
     private ToUpper(StreamInput in) throws IOException {
-        this(Source.EMPTY, in.readNamedWriteable(Expression.class), ((PlanStreamInput) in).configuration());
+        this(
+            in.getTransportVersion().onOrAfter(TransportVersions.ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS)
+                ? Source.readFrom((PlanStreamInput) in)
+                : Source.EMPTY,
+            in.readNamedWriteable(Expression.class),
+            ((PlanStreamInput) in).configuration()
+        );
     }
 
     @Override
     public void writeTo(StreamOutput out) throws IOException {
+        if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_SERIALIZE_SOURCE_FUNCTIONS_WARNINGS)) {
+            source().writeTo(out);
+        }
         out.writeNamedWriteable(field());
     }
 

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/TestCaseSupplier.java

@@ -117,6 +117,9 @@ public record TestCaseSupplier(String name, List<DataType> types, Supplier<TestC
     @Override
     public TestCase get() {
         TestCase supplied = supplier.get();
+        if (types.size() != supplied.getData().size()) {
+            throw new IllegalStateException(name + ": type/data size mismatch " + types.size() + "/" + supplied.getData().size());
+        }
         for (int i = 0; i < types.size(); i++) {
             if (supplied.getData().get(i).type() != types.get(i)) {
                 throw new IllegalStateException(

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/AvgSerializationTests.java

@@ -21,9 +21,4 @@ public class AvgSerializationTests extends AbstractExpressionSerializationTests<
     protected Avg mutateInstance(Avg instance) throws IOException {
         return new Avg(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/CountDistinctSerializationTests.java

@@ -34,9 +34,4 @@ public class CountDistinctSerializationTests extends AbstractExpressionSerializa
         }
         return new CountDistinct(source, field, precision);
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/CountSerializationTests.java

@@ -21,9 +21,4 @@ public class CountSerializationTests extends AbstractExpressionSerializationTest
     protected Count mutateInstance(Count instance) throws IOException {
         return new Count(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/MaxSerializationTests.java

@@ -21,9 +21,4 @@ public class MaxSerializationTests extends AbstractExpressionSerializationTests<
     protected Max mutateInstance(Max instance) throws IOException {
         return new Max(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/MedianAbsoluteDeviationSerializationTests.java

@@ -24,9 +24,4 @@ public class MedianAbsoluteDeviationSerializationTests extends AbstractExpressio
             randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild)
         );
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/MedianSerializationTests.java

@@ -21,9 +21,4 @@ public class MedianSerializationTests extends AbstractExpressionSerializationTes
     protected Median mutateInstance(Median instance) throws IOException {
         return new Median(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/MinSerializationTests.java

@@ -21,9 +21,4 @@ public class MinSerializationTests extends AbstractExpressionSerializationTests<
     protected Min mutateInstance(Min instance) throws IOException {
         return new Min(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/PercentileSerializationTests.java

@@ -34,9 +34,4 @@ public class PercentileSerializationTests extends AbstractExpressionSerializatio
         }
         return new Percentile(source, field, percentile);
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/RateSerializationTests.java

@@ -36,9 +36,4 @@ public class RateSerializationTests extends AbstractExpressionSerializationTests
         }
         return new Rate(source, field, timestamp, unit);
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/SpatialCentroidSerializationTests.java

@@ -24,9 +24,4 @@ public class SpatialCentroidSerializationTests extends AbstractExpressionSeriali
             randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild)
         );
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/SumSerializationTests.java

@@ -21,9 +21,4 @@ public class SumSerializationTests extends AbstractExpressionSerializationTests<
     protected Sum mutateInstance(Sum instance) throws IOException {
         return new Sum(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/TopSerializationTests.java

@@ -36,9 +36,4 @@ public class TopSerializationTests extends AbstractExpressionSerializationTests<
         }
         return new Top(source, field, limit, order);
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/TopTests.java

@@ -126,7 +126,7 @@ public class TopTests extends AbstractAggregationTestCase {
                     )
                 ),
                 new TestCaseSupplier(
-                    List.of(DataType.IP),
+                    List.of(DataType.IP, DataType.INTEGER, DataType.KEYWORD),
                     () -> new TestCaseSupplier.TestCase(
                         List.of(
                             TestCaseSupplier.TypedData.multiRow(
@@ -215,7 +215,7 @@ public class TopTests extends AbstractAggregationTestCase {
                     )
                 ),
                 new TestCaseSupplier(
-                    List.of(DataType.IP),
+                    List.of(DataType.IP, DataType.INTEGER, DataType.KEYWORD),
                     () -> new TestCaseSupplier.TestCase(
                         List.of(
                             TestCaseSupplier.TypedData.multiRow(

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/aggregate/ValuesSerializationTests.java

@@ -21,9 +21,4 @@ public class ValuesSerializationTests extends AbstractExpressionSerializationTes
     protected Values mutateInstance(Values instance) throws IOException {
         return new Values(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/conditional/CaseTests.java

@@ -532,7 +532,7 @@ public class CaseTests extends AbstractScalarFunctionTestCase {
                 suppliers.add(
                     new TestCaseSupplier(
                         "partial foldable 1 " + TestCaseSupplier.nameFrom(Arrays.asList(cond1, type, cond2, type)),
-                        List.of(DataType.BOOLEAN, type, DataType.BOOLEAN, type),
+                        List.of(DataType.BOOLEAN, type, DataType.BOOLEAN, type, type),
                         () -> {
                             Object r1 = randomLiteral(type).value();
                             Object r2 = randomLiteral(type).value();

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/date/DateDiffSerializationTests.java

@@ -36,9 +36,4 @@ public class DateDiffSerializationTests extends AbstractExpressionSerializationT
         }
         return new DateDiff(source, unit, startTimestamp, endTimestamp);
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvAvgSerializationTests.java

@@ -21,9 +21,4 @@ public class MvAvgSerializationTests extends AbstractExpressionSerializationTest
     protected MvAvg mutateInstance(MvAvg instance) throws IOException {
         return new MvAvg(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvCountSerializationTests.java

@@ -21,9 +21,4 @@ public class MvCountSerializationTests extends AbstractExpressionSerializationTe
     protected MvCount mutateInstance(MvCount instance) throws IOException {
         return new MvCount(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvDedupeSerializationTests.java

@@ -21,9 +21,4 @@ public class MvDedupeSerializationTests extends AbstractExpressionSerializationT
     protected MvDedupe mutateInstance(MvDedupe instance) throws IOException {
         return new MvDedupe(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvFirstSerializationTests.java

@@ -21,9 +21,4 @@ public class MvFirstSerializationTests extends AbstractExpressionSerializationTe
     protected MvFirst mutateInstance(MvFirst instance) throws IOException {
         return new MvFirst(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvLastSerializationTests.java

@@ -21,9 +21,4 @@ public class MvLastSerializationTests extends AbstractExpressionSerializationTes
     protected MvLast mutateInstance(MvLast instance) throws IOException {
         return new MvLast(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvMaxSerializationTests.java

@@ -21,9 +21,4 @@ public class MvMaxSerializationTests extends AbstractExpressionSerializationTest
     protected MvMax mutateInstance(MvMax instance) throws IOException {
         return new MvMax(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvMedianAbsoluteDeviationSerializationTests.java

@@ -24,9 +24,4 @@ public class MvMedianAbsoluteDeviationSerializationTests extends AbstractExpress
             randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild)
         );
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvMedianSerializationTests.java

@@ -21,9 +21,4 @@ public class MvMedianSerializationTests extends AbstractExpressionSerializationT
     protected MvMedian mutateInstance(MvMedian instance) throws IOException {
         return new MvMedian(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvMinSerializationTests.java

@@ -21,9 +21,4 @@ public class MvMinSerializationTests extends AbstractExpressionSerializationTest
     protected MvMin mutateInstance(MvMin instance) throws IOException {
         return new MvMin(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvSliceSerializationTests.java

@@ -36,9 +36,4 @@ public class MvSliceSerializationTests extends AbstractExpressionSerializationTe
         }
         return new MvSlice(source, field, start, end);
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvSliceTests.java

@@ -13,6 +13,7 @@ import com.carrotsearch.randomizedtesting.annotations.ParametersFactory;
 import org.apache.lucene.util.BytesRef;
 import org.elasticsearch.geo.GeometryTestUtils;
 import org.elasticsearch.geo.ShapeTestUtils;
+import org.elasticsearch.xpack.esql.core.InvalidArgumentException;
 import org.elasticsearch.xpack.esql.core.expression.Expression;
 import org.elasticsearch.xpack.esql.core.tree.Source;
 import org.elasticsearch.xpack.esql.core.type.DataType;
@@ -44,6 +45,64 @@ public class MvSliceTests extends AbstractScalarFunctionTestCase {
         longs(suppliers);
         doubles(suppliers);
         bytesRefs(suppliers);
+
+        // Warnings cases
+        suppliers.stream().toList().forEach(supplier -> {
+            DataType firstArgumentType = supplier.types().get(0);
+            String evaluatorTypePart = switch (firstArgumentType) {
+                case BOOLEAN -> "Boolean";
+                case INTEGER -> "Int";
+                case LONG, DATE_NANOS, DATETIME, UNSIGNED_LONG -> "Long";
+                case DOUBLE -> "Double";
+                case KEYWORD, TEXT, SEMANTIC_TEXT, IP, VERSION, GEO_POINT, CARTESIAN_POINT, GEO_SHAPE, CARTESIAN_SHAPE -> "BytesRef";
+                default -> throw new IllegalArgumentException("Unsupported type: " + firstArgumentType);
+            };
+
+            // Start offset greater than end offset
+            suppliers.add(new TestCaseSupplier(List.of(firstArgumentType, DataType.INTEGER, DataType.INTEGER), () -> {
+                int end = randomIntBetween(0, 10);
+                int start = randomIntBetween(end + 1, end + 10);
+                return new TestCaseSupplier.TestCase(
+                    List.of(
+                        new TestCaseSupplier.TypedData(List.of(randomLiteral(firstArgumentType).value()), firstArgumentType, "field"),
+                        new TestCaseSupplier.TypedData(start, DataType.INTEGER, "start"),
+                        new TestCaseSupplier.TypedData(end, DataType.INTEGER, "end")
+                    ),
+                    "MvSlice"
+                        + evaluatorTypePart
+                        + "Evaluator[field=Attribute[channel=0], start=Attribute[channel=1], end=Attribute[channel=2]]",
+                    firstArgumentType,
+                    nullValue()
+                ).withFoldingException(InvalidArgumentException.class, "Start offset is greater than end offset")
+                    .withWarning("Line 1:1: evaluation of [source] failed, treating result as null. Only first 20 failures recorded.")
+                    .withWarning(
+                        "Line 1:1: org.elasticsearch.xpack.esql.core.InvalidArgumentException: Start offset is greater than end offset"
+                    );
+            }));
+
+            // Negative start with positive end
+            suppliers.add(new TestCaseSupplier(List.of(firstArgumentType, DataType.INTEGER, DataType.INTEGER), () -> {
+                int start = randomIntBetween(-10, -1);
+                int end = randomIntBetween(0, 10);
+                return new TestCaseSupplier.TestCase(
+                    List.of(
+                        new TestCaseSupplier.TypedData(List.of(randomLiteral(firstArgumentType).value()), firstArgumentType, "field"),
+                        new TestCaseSupplier.TypedData(start, DataType.INTEGER, "start"),
+                        new TestCaseSupplier.TypedData(end, DataType.INTEGER, "end")
+                    ),
+                    "MvSlice"
+                        + evaluatorTypePart
+                        + "Evaluator[field=Attribute[channel=0], start=Attribute[channel=1], end=Attribute[channel=2]]",
+                    firstArgumentType,
+                    nullValue()
+                ).withFoldingException(InvalidArgumentException.class, "Start and end offset have different signs")
+                    .withWarning("Line 1:1: evaluation of [source] failed, treating result as null. Only first 20 failures recorded.")
+                    .withWarning(
+                        "Line 1:1: org.elasticsearch.xpack.esql.core.InvalidArgumentException: Start and end offset have different signs"
+                    );
+            }));
+        });
+
         return parameterSuppliersFromTypedData(
             anyNullIsNull(
                 suppliers,

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/multivalue/MvSumSerializationTests.java

@@ -21,9 +21,4 @@ public class MvSumSerializationTests extends AbstractExpressionSerializationTest
     protected MvSum mutateInstance(MvSum instance) throws IOException {
         return new MvSum(instance.source(), randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild));
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/spatial/AbstractBinarySpatialFunctionSerializationTestCase.java

@@ -38,9 +38,4 @@ public abstract class AbstractBinarySpatialFunctionSerializationTestCase<T exten
         }
         return build(source, left, right);
     }
-
-    @Override
-    protected final boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ReplaceSerializationTests.java

@@ -36,9 +36,4 @@ public class ReplaceSerializationTests extends AbstractExpressionSerializationTe
         }
         return new Replace(source, str, regex, newStr);
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToLowerSerializationTests.java

@@ -25,9 +25,4 @@ public class ToLowerSerializationTests extends AbstractExpressionSerializationTe
         Expression child = randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild);
         return new ToLower(source, child, configuration());
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/expression/function/scalar/string/ToUpperSerializationTests.java

@@ -25,9 +25,4 @@ public class ToUpperSerializationTests extends AbstractExpressionSerializationTe
         Expression child = randomValueOtherThan(instance.field(), AbstractExpressionSerializationTests::randomChild);
         return new ToUpper(source, child, configuration());
     }
-
-    @Override
-    protected boolean alwaysEmptySource() {
-        return true;
-    }
 }

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

@@ -26,7 +26,9 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Locale;
 import java.util.function.Supplier;
+import java.util.stream.IntStream;
 
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.of;
 import static org.elasticsearch.xpack.esql.EsqlTestUtils.randomLiteral;
@@ -91,7 +93,7 @@ public class InTests extends AbstractFunctionTestCase {
     }
 
     private static void booleans(List<TestCaseSupplier> suppliers, int items) {
-        suppliers.add(new TestCaseSupplier("boolean", List.of(DataType.BOOLEAN, DataType.BOOLEAN), () -> {
+        suppliers.add(new TestCaseSupplier("boolean", typesList(DataType.BOOLEAN, DataType.BOOLEAN, items), () -> {
             List<Boolean> inlist = randomList(items, items, () -> randomBoolean());
             boolean field = randomBoolean();
             List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -109,7 +111,7 @@ public class InTests extends AbstractFunctionTestCase {
     }
 
     private static void numerics(List<TestCaseSupplier> suppliers, int items) {
-        suppliers.add(new TestCaseSupplier("integer", List.of(DataType.INTEGER, DataType.INTEGER), () -> {
+        suppliers.add(new TestCaseSupplier("integer", typesList(DataType.INTEGER, DataType.INTEGER, items), () -> {
             List<Integer> inlist = randomList(items, items, () -> randomInt());
             int field = inlist.get(inlist.size() - 1);
             List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -125,7 +127,7 @@ public class InTests extends AbstractFunctionTestCase {
             );
         }));
 
-        suppliers.add(new TestCaseSupplier("long", List.of(DataType.LONG, DataType.LONG), () -> {
+        suppliers.add(new TestCaseSupplier("long", typesList(DataType.LONG, DataType.LONG, items), () -> {
             List<Long> inlist = randomList(items, items, () -> randomLong());
             long field = randomLong();
             List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -141,7 +143,7 @@ public class InTests extends AbstractFunctionTestCase {
             );
         }));
 
-        suppliers.add(new TestCaseSupplier("double", List.of(DataType.DOUBLE, DataType.DOUBLE), () -> {
+        suppliers.add(new TestCaseSupplier("double", typesList(DataType.DOUBLE, DataType.DOUBLE, items), () -> {
             List<Double> inlist = randomList(items, items, () -> randomDouble());
             double field = inlist.get(0);
             List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -159,58 +161,10 @@ public class InTests extends AbstractFunctionTestCase {
     }
 
     private static void bytesRefs(List<TestCaseSupplier> suppliers, int items) {
-        suppliers.add(new TestCaseSupplier("keyword", List.of(DataType.KEYWORD, DataType.KEYWORD), () -> {
-            List<Object> inlist = randomList(items, items, () -> randomLiteral(DataType.KEYWORD).value());
-            Object field = inlist.get(inlist.size() - 1);
-            List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
-            for (Object i : inlist) {
-                args.add(new TestCaseSupplier.TypedData(i, DataType.KEYWORD, "inlist" + i));
-            }
-            args.add(new TestCaseSupplier.TypedData(field, DataType.KEYWORD, "field"));
-            return new TestCaseSupplier.TestCase(
-                args,
-                matchesPattern("InBytesRefEvaluator.*"),
-                DataType.BOOLEAN,
-                equalTo(inlist.contains(field))
-            );
-        }));
-
-        suppliers.add(new TestCaseSupplier("text", List.of(DataType.TEXT, DataType.TEXT), () -> {
-            List<Object> inlist = randomList(items, items, () -> randomLiteral(DataType.TEXT).value());
-            Object field = inlist.get(0);
-            List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
-            for (Object i : inlist) {
-                args.add(new TestCaseSupplier.TypedData(i, DataType.TEXT, "inlist" + i));
-            }
-            args.add(new TestCaseSupplier.TypedData(field, DataType.TEXT, "field"));
-            return new TestCaseSupplier.TestCase(
-                args,
-                matchesPattern("InBytesRefEvaluator.*"),
-                DataType.BOOLEAN,
-                equalTo(inlist.contains(field))
-            );
-        }));
-
-        suppliers.add(new TestCaseSupplier("semantic_text", List.of(DataType.SEMANTIC_TEXT, DataType.SEMANTIC_TEXT), () -> {
-            List<Object> inlist = randomList(items, items, () -> randomLiteral(DataType.SEMANTIC_TEXT).value());
-            Object field = inlist.get(0);
-            List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
-            for (Object i : inlist) {
-                args.add(new TestCaseSupplier.TypedData(i, DataType.SEMANTIC_TEXT, "inlist" + i));
-            }
-            args.add(new TestCaseSupplier.TypedData(field, DataType.SEMANTIC_TEXT, "field"));
-            return new TestCaseSupplier.TestCase(
-                args,
-                matchesPattern("InBytesRefEvaluator.*"),
-                DataType.BOOLEAN,
-                equalTo(inlist.contains(field))
-            );
-        }));
-
         for (DataType type1 : DataType.stringTypes()) {
             for (DataType type2 : DataType.stringTypes()) {
-                if (type1 == type2 || items > 1) continue;
-                suppliers.add(new TestCaseSupplier(type1 + " " + type2, List.of(type1, type2), () -> {
+                String name = type1 == type2 ? type1.toString() : type1 + " " + type2;
+                suppliers.add(new TestCaseSupplier(name.toLowerCase(Locale.ROOT), typesList(type1, type2, items), () -> {
                     List<Object> inlist = randomList(items, items, () -> randomLiteral(type1).value());
                     Object field = randomLiteral(type2).value();
                     List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -227,7 +181,7 @@ public class InTests extends AbstractFunctionTestCase {
                 }));
             }
         }
-        suppliers.add(new TestCaseSupplier("ip", List.of(DataType.IP, DataType.IP), () -> {
+        suppliers.add(new TestCaseSupplier("ip", typesList(DataType.IP, DataType.IP, items), () -> {
             List<Object> inlist = randomList(items, items, () -> randomLiteral(DataType.IP).value());
             Object field = randomLiteral(DataType.IP).value();
             List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -243,7 +197,7 @@ public class InTests extends AbstractFunctionTestCase {
             );
         }));
 
-        suppliers.add(new TestCaseSupplier("version", List.of(DataType.VERSION, DataType.VERSION), () -> {
+        suppliers.add(new TestCaseSupplier("version", typesList(DataType.VERSION, DataType.VERSION, items), () -> {
             List<Object> inlist = randomList(items, items, () -> randomLiteral(DataType.VERSION).value());
             Object field = randomLiteral(DataType.VERSION).value();
             List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -259,7 +213,7 @@ public class InTests extends AbstractFunctionTestCase {
             );
         }));
 
-        suppliers.add(new TestCaseSupplier("geo_point", List.of(GEO_POINT, GEO_POINT), () -> {
+        suppliers.add(new TestCaseSupplier("geo_point", typesList(GEO_POINT, GEO_POINT, items), () -> {
             List<Object> inlist = randomList(items, items, () -> new BytesRef(GEO.asWkt(GeometryTestUtils.randomPoint())));
             Object field = inlist.get(0);
             List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -275,7 +229,7 @@ public class InTests extends AbstractFunctionTestCase {
             );
         }));
 
-        suppliers.add(new TestCaseSupplier("geo_shape", List.of(GEO_SHAPE, GEO_SHAPE), () -> {
+        suppliers.add(new TestCaseSupplier("geo_shape", typesList(GEO_SHAPE, GEO_SHAPE, items), () -> {
             List<Object> inlist = randomList(
                 items,
                 items,
@@ -295,7 +249,7 @@ public class InTests extends AbstractFunctionTestCase {
             );
         }));
 
-        suppliers.add(new TestCaseSupplier("cartesian_point", List.of(CARTESIAN_POINT, CARTESIAN_POINT), () -> {
+        suppliers.add(new TestCaseSupplier("cartesian_point", typesList(CARTESIAN_POINT, CARTESIAN_POINT, items), () -> {
             List<Object> inlist = randomList(items, items, () -> new BytesRef(CARTESIAN.asWkt(ShapeTestUtils.randomPoint())));
             Object field = new BytesRef(CARTESIAN.asWkt(ShapeTestUtils.randomPoint()));
             List<TestCaseSupplier.TypedData> args = new ArrayList<>(inlist.size() + 1);
@@ -311,7 +265,7 @@ public class InTests extends AbstractFunctionTestCase {
             );
         }));
 
-        suppliers.add(new TestCaseSupplier("cartesian_shape", List.of(CARTESIAN_SHAPE, CARTESIAN_SHAPE), () -> {
+        suppliers.add(new TestCaseSupplier("cartesian_shape", typesList(CARTESIAN_SHAPE, CARTESIAN_SHAPE, items), () -> {
             List<Object> inlist = randomList(
                 items,
                 items,
@@ -332,6 +286,16 @@ public class InTests extends AbstractFunctionTestCase {
         }));
     }
 
+    /**
+     * Returns a list with N dataType1, followed by 1 dataType2.
+     */
+    private static List<DataType> typesList(DataType inListType, DataType fieldType, int n) {
+        List<DataType> types = new ArrayList<>(n + 1);
+        IntStream.range(0, n).forEach(i -> types.add(inListType));
+        types.add(fieldType);
+        return types;
+    }
+
     @Override
     protected Expression build(Source source, List<Expression> args) {
         return new In(source, args.get(args.size() - 1), args.subList(0, args.size() - 1));