ESQL: Fix validation NPE in Enrich and add extra @nullable annotations (#128260)

Fixes #126253
Fixes #126297

`inputDataType` may be null when in mixed cluster (<8.14). So `validateTypes()` should take that into account. Similar fix to #116583

Author: ivancea

docs/changelog/128260.yaml

@@ -0,0 +1,7 @@
+pr: 128260
+summary: Fix validation NPE in Enrich and add extra @Nullable annotations
+area: ES|QL
+type: bug
+issues:
+ - 126297
+ - 126253

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/AbstractLookupService.java

@@ -182,7 +182,7 @@ protected abstract QueryList queryList(
         T request,
         SearchExecutionContext context,
         Block inputBlock,
-        DataType inputDataType,
+        @Nullable DataType inputDataType,
         Warnings warnings
     );
 
@@ -200,7 +200,7 @@ protected static QueryList termQueryList(
         MappedFieldType field,
         SearchExecutionContext searchExecutionContext,
         Block block,
-        DataType inputDataType
+        @Nullable DataType inputDataType
     ) {
         return switch (inputDataType) {
             case IP -> QueryList.ipTermQueryList(field, searchExecutionContext, (BytesRefBlock) block);

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/EnrichLookupService.java

@@ -24,6 +24,7 @@
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.Warnings;
 import org.elasticsearch.compute.operator.lookup.QueryList;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.index.mapper.MappedFieldType;
 import org.elasticsearch.index.mapper.RangeFieldMapper;
@@ -103,7 +104,7 @@ protected QueryList queryList(
         TransportRequest request,
         SearchExecutionContext context,
         Block inputBlock,
-        DataType inputDataType,
+        @Nullable DataType inputDataType,
         Warnings warnings
     ) {
         MappedFieldType fieldType = context.getFieldType(request.matchField);
@@ -128,8 +129,8 @@ protected LookupResponse readLookupResponse(StreamInput in, BlockFactory blockFa
         return new LookupResponse(in, blockFactory);
     }
 
-    private static void validateTypes(DataType inputDataType, MappedFieldType fieldType) {
-        if (fieldType instanceof RangeFieldMapper.RangeFieldType rangeType) {
+    private static void validateTypes(@Nullable DataType inputDataType, MappedFieldType fieldType) {
+        if (fieldType instanceof RangeFieldMapper.RangeFieldType rangeType && inputDataType != null) {
             // For range policy types, the ENRICH index field type will be one of a list of supported range types,
             // which need to match the input data type (eg. ip-range -> ip, date-range -> date, etc.)
             if (rangeTypesCompatible(rangeType.rangeType(), inputDataType) == false) {
@@ -142,7 +143,7 @@ private static void validateTypes(DataType inputDataType, MappedFieldType fieldT
         // For geo_match, type validation is done earlier, in the Analyzer.
     }
 
-    private static boolean rangeTypesCompatible(RangeType rangeType, DataType inputDataType) {
+    private static boolean rangeTypesCompatible(RangeType rangeType, @Nullable DataType inputDataType) {
         if (inputDataType.noText() == DataType.KEYWORD) {
             // We allow runtime parsing of string types to numeric types
             return true;

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/enrich/LookupFromIndexService.java

@@ -19,6 +19,7 @@
 import org.elasticsearch.compute.data.Page;
 import org.elasticsearch.compute.operator.Warnings;
 import org.elasticsearch.compute.operator.lookup.QueryList;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.index.query.SearchExecutionContext;
 import org.elasticsearch.index.shard.ShardId;
@@ -83,7 +84,7 @@ protected QueryList queryList(
         TransportRequest request,
         SearchExecutionContext context,
         Block inputBlock,
-        DataType inputDataType,
+        @Nullable DataType inputDataType,
         Warnings warnings
     ) {
         return termQueryList(context.getFieldType(request.matchField), context, inputBlock, inputDataType).onlySingleValues(