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

Fixes elastic#116529
Fixes elastic#116544

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;
@@ -185,7 +186,7 @@ protected static QueryList termQueryList(
         return switch (inputDataType) {
             case IP -> QueryList.ipTermQueryList(field, searchExecutionContext, (BytesRefBlock) block);
             case DATETIME -> QueryList.dateTermQueryList(field, searchExecutionContext, (LongBlock) block);
-            default -> QueryList.rawTermQueryList(field, searchExecutionContext, block);
+            case null, default -> QueryList.rawTermQueryList(field, searchExecutionContext, block);
         };
     }
 
@@ -459,6 +460,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;