[8.x] Fix NPE in EnrichLookupService on mixed clusters with <8.14 versions (elastic#116583) (elastic#116607)

* Fix NPE in EnrichLookupService on mixed clusters with <8.14 versions (elastic#116583)

Fixes elastic#116529
Fixes elastic#116544

* Changed switch type for 8.x

Author: ivancea

docs/changelog/116583.yaml

@@ -0,0 +1,7 @@
+pr: 116583
+summary: Fix NPE in `EnrichLookupService` on mixed clusters with <8.14 versions
+area: ES|QL
+type: bug
+issues:
+ - 116529
+ - 116544

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

@@ -45,6 +45,7 @@
 import org.elasticsearch.compute.operator.lookup.MergePositionsOperator;
 import org.elasticsearch.compute.operator.lookup.QueryList;
 import org.elasticsearch.core.AbstractRefCounted;
+import org.elasticsearch.core.Nullable;
 import org.elasticsearch.core.RefCounted;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.Releasables;
@@ -182,6 +183,9 @@ protected static QueryList termQueryList(
         Block block,
         DataType inputDataType
     ) {
+        if (inputDataType == null) {
+            return QueryList.rawTermQueryList(field, searchExecutionContext, block);
+        }
         return switch (inputDataType) {
             case IP -> QueryList.ipTermQueryList(field, searchExecutionContext, (BytesRefBlock) block);
             case DATETIME -> QueryList.dateTermQueryList(field, searchExecutionContext, (LongBlock) block);
@@ -459,6 +463,10 @@ abstract static class Request {
     abstract static class TransportRequest extends org.elasticsearch.transport.TransportRequest implements IndicesRequest {
         final String sessionId;
         final ShardId shardId;
+        /**
+         * For mixed clusters with nodes <8.14, this will be null.
+         */
+        @Nullable
         final DataType inputDataType;
         final Page inputPage;
         final List<NamedExpression> extractFields;

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

@@ -127,9 +127,9 @@ static TransportRequest readFrom(StreamInput in, BlockFactory blockFactory) thro
             TaskId parentTaskId = TaskId.readFromStream(in);
             String sessionId = in.readString();
             ShardId shardId = new ShardId(in);
-            DataType inputDataType = DataType.fromTypeName(
-                (in.getTransportVersion().onOrAfter(TransportVersions.V_8_14_0)) ? in.readString() : "unknown"
-            );
+            DataType inputDataType = (in.getTransportVersion().onOrAfter(TransportVersions.V_8_14_0))
+                ? DataType.fromTypeName(in.readString())
+                : null;
             String matchType = in.readString();
             String matchField = in.readString();
             Page inputPage;