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

Fixes elastic#126253
Fixes elastic#126297

`inputDataType` may be null when in mixed cluster (<8.14). So `validateTypes()` should take that into account. Similar fix to elastic#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

@@ -177,7 +177,7 @@ public ThreadContext getThreadContext() {
     /**
      * Build a list of queries to perform inside the actual lookup.
      */
-    protected abstract QueryList queryList(T request, SearchExecutionContext context, Block inputBlock, DataType inputDataType);
+    protected abstract QueryList queryList(T request, SearchExecutionContext context, Block inputBlock, @Nullable DataType inputDataType);
 
     /**
      * Build the response.
@@ -193,7 +193,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

@@ -23,6 +23,7 @@
 import org.elasticsearch.compute.data.BlockStreamInput;
 import org.elasticsearch.compute.data.Page;
 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;
@@ -98,7 +99,12 @@ protected TransportRequest transportRequest(EnrichLookupService.Request request,
     }
 
     @Override
-    protected QueryList queryList(TransportRequest request, SearchExecutionContext context, Block inputBlock, DataType inputDataType) {
+    protected QueryList queryList(
+        TransportRequest request,
+        SearchExecutionContext context,
+        Block inputBlock,
+        @Nullable DataType inputDataType
+    ) {
         MappedFieldType fieldType = context.getFieldType(request.matchField);
         validateTypes(inputDataType, fieldType);
         return switch (request.matchType) {
@@ -121,8 +127,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) {
@@ -135,7 +141,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

@@ -18,6 +18,7 @@
 import org.elasticsearch.compute.data.BlockStreamInput;
 import org.elasticsearch.compute.data.Page;
 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;
@@ -76,7 +77,12 @@ protected TransportRequest transportRequest(LookupFromIndexService.Request reque
     }
 
     @Override
-    protected QueryList queryList(TransportRequest request, SearchExecutionContext context, Block inputBlock, DataType inputDataType) {
+    protected QueryList queryList(
+        TransportRequest request,
+        SearchExecutionContext context,
+        Block inputBlock,
+        @Nullable DataType inputDataType
+    ) {
         return termQueryList(context.getFieldType(request.matchField), context, inputBlock, inputDataType).onlySingleValues();
     }