Only transport the Source text, instead of the full Configuration

Author: ivancea

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

@@ -175,7 +175,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion COHERE_BIT_EMBEDDING_TYPE_SUPPORT_ADDED_BACKPORT_8_X = def(8_840_0_01);
     public static final TransportVersion ELASTICSEARCH_9_0 = def(9_000_0_00);
     public static final TransportVersion COHERE_BIT_EMBEDDING_TYPE_SUPPORT_ADDED = def(9_001_0_00);
-    public static final TransportVersion ESQL_LOOKUP_JOIN_CONFIGURATION_IN_REQUEST = def(9_002_0_00);
+    public static final TransportVersion ESQL_LOOKUP_JOIN_SOURCE_TEXT = def(9_002_0_00);
     /*
      * STOP! READ THIS FIRST! No, really,
      *        ____ _____ ___  ____  _        ____  _____    _    ____    _____ _   _ ___ ____    _____ ___ ____  ____ _____ _

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

@@ -86,7 +86,6 @@ public Operator get(DriverContext driverContext) {
 
     private final LookupFromIndexService lookupService;
     private final String sessionId;
-    private final Configuration configuration;
     private final CancellableTask parentTask;
     private final int inputChannel;
     private final DataType inputDataType;
@@ -106,7 +105,6 @@ public Operator get(DriverContext driverContext) {
 
     public LookupFromIndexOperator(
         String sessionId,
-        Configuration configuration,
         DriverContext driverContext,
         CancellableTask parentTask,
         int maxOutstandingRequests,
@@ -120,7 +118,6 @@ public LookupFromIndexOperator(
     ) {
         super(driverContext, lookupService.getThreadContext(), maxOutstandingRequests);
         this.sessionId = sessionId;
-        this.configuration = configuration;
         this.parentTask = parentTask;
         this.inputChannel = inputChannel;
         this.lookupService = lookupService;
@@ -137,7 +134,6 @@ protected void performAsync(Page inputPage, ActionListener<OngoingJoin> listener
         totalTerms += inputBlock.getTotalValueCount();
         LookupFromIndexService.Request request = new LookupFromIndexService.Request(
             sessionId,
-            configuration,
             lookupIndex,
             inputDataType,
             matchField,

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

@@ -71,7 +71,6 @@ public LookupFromIndexService(
     protected TransportRequest transportRequest(LookupFromIndexService.Request request, ShardId shardId) {
         return new TransportRequest(
             request.sessionId,
-            request.configuration,
             shardId,
             request.inputDataType,
             request.inputPage,
@@ -107,12 +106,10 @@ protected AbstractLookupService.LookupResponse readLookupResponse(StreamInput in
     }
 
     public static class Request extends AbstractLookupService.Request {
-        private final Configuration configuration;
         private final String matchField;
 
         Request(
             String sessionId,
-            Configuration configuration,
             String index,
             DataType inputDataType,
             String matchField,
@@ -121,18 +118,15 @@ public static class Request extends AbstractLookupService.Request {
             Source source
         ) {
             super(sessionId, index, inputDataType, inputPage, extractFields, source);
-            this.configuration = configuration;
             this.matchField = matchField;
         }
     }
 
     protected static class TransportRequest extends AbstractLookupService.TransportRequest {
-        private final Configuration configuration;
         private final String matchField;
 
         TransportRequest(
             String sessionId,
-            Configuration configuration,
             ShardId shardId,
             DataType inputDataType,
             Page inputPage,
@@ -142,39 +136,33 @@ protected static class TransportRequest extends AbstractLookupService.TransportR
             Source source
         ) {
             super(sessionId, shardId, inputDataType, inputPage, toRelease, extractFields, source);
-            this.configuration = configuration;
             this.matchField = matchField;
         }
 
         static TransportRequest readFrom(StreamInput in, BlockFactory blockFactory) throws IOException {
             TaskId parentTaskId = TaskId.readFromStream(in);
             String sessionId = in.readString();
-            Configuration configuration = null;
-            if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_LOOKUP_JOIN_CONFIGURATION_IN_REQUEST)) {
-                configuration = new Configuration(
-                    // TODO make EsqlConfiguration Releasable
-                    new BlockStreamInput(
-                        in,
-                        new BlockFactory(new NoopCircuitBreaker(CircuitBreaker.REQUEST), BigArrays.NON_RECYCLING_INSTANCE)
-                    )
-                );
-            }
             ShardId shardId = new ShardId(in);
             DataType inputDataType = DataType.fromTypeName(in.readString());
             Page inputPage;
             try (BlockStreamInput bsi = new BlockStreamInput(in, blockFactory)) {
                 inputPage = new Page(bsi);
             }
-            PlanStreamInput planIn = new PlanStreamInput(in, in.namedWriteableRegistry(), configuration);
+            PlanStreamInput planIn = new PlanStreamInput(in, in.namedWriteableRegistry(), null);
             List<NamedExpression> extractFields = planIn.readNamedWriteableCollectionAsList(NamedExpression.class);
             String matchField = in.readString();
             var source = Source.EMPTY;
             if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_ENRICH_RUNTIME_WARNINGS)) {
                 source = Source.readFrom(planIn);
             }
+            // Source.readFrom() requires the query from the Configuration passed to PlanStreamInput.
+            // As we don't have the Configuration here, and it may be heavy to serialize, we directly pass the Source text.
+            if (in.getTransportVersion().onOrAfter(TransportVersions.ESQL_LOOKUP_JOIN_SOURCE_TEXT)) {
+                String sourceText = in.readString();
+                source = new Source(source.source(), sourceText);
+            }
             TransportRequest result = new TransportRequest(
                 sessionId,
-                configuration,
                 shardId,
                 inputDataType,
                 inputPage,
@@ -191,9 +179,6 @@ static TransportRequest readFrom(StreamInput in, BlockFactory blockFactory) thro
         public void writeTo(StreamOutput out) throws IOException {
             super.writeTo(out);
             out.writeString(sessionId);
-            if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_LOOKUP_JOIN_CONFIGURATION_IN_REQUEST)) {
-                configuration.writeTo(out);
-            }
             out.writeWriteable(shardId);
             out.writeString(inputDataType.typeName());
             out.writeWriteable(inputPage);
@@ -203,6 +188,9 @@ public void writeTo(StreamOutput out) throws IOException {
             if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_ENRICH_RUNTIME_WARNINGS)) {
                 source.writeTo(planOut);
             }
+            if (out.getTransportVersion().onOrAfter(TransportVersions.ESQL_LOOKUP_JOIN_SOURCE_TEXT)) {
+                out.writeString(source.text());
+            }
         }
 
         @Override