Fix the transport version of PlanStreamOutput (#103758) (#103767) (#103768)

The backward compatibility of ESQL is not handled correctly because
PlanStreamOutput doesn't return the transport version from its delegate.
The reason we didn't notice these failures is that the mixed cluster QA
doesn't upgrade two nodes to the current version, leaving all nodes in
the cluster with BWC versions.

Author: dnhatn

docs/changelog/103758.yaml

@@ -0,0 +1,5 @@
+pr: 103758
+summary: Fix the transport version of `PlanStreamOutput`
+area: ES|QL
+type: bug
+issues: []

x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/io/stream/PlanStreamOutput.java

@@ -7,7 +7,7 @@
 
 package org.elasticsearch.xpack.esql.io.stream;
 
-import org.elasticsearch.common.io.stream.OutputStreamStreamOutput;
+import org.elasticsearch.TransportVersion;
 import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.xpack.esql.io.stream.PlanNameRegistry.PlanWriter;
 import org.elasticsearch.xpack.esql.plan.physical.PhysicalPlan;
@@ -24,18 +24,19 @@
  * A customized stream output used to serialize ESQL physical plan fragments. Complements stream
  * output with methods that write plan nodes, Attributes, Expressions, etc.
  */
-public final class PlanStreamOutput extends OutputStreamStreamOutput {
+public final class PlanStreamOutput extends StreamOutput {
 
+    private final StreamOutput delegate;
     private final PlanNameRegistry registry;
 
     private final Function<Class<?>, String> nameSupplier;
 
-    public PlanStreamOutput(StreamOutput streamOutput, PlanNameRegistry registry) {
-        this(streamOutput, registry, PlanNamedTypes::name);
+    public PlanStreamOutput(StreamOutput delegate, PlanNameRegistry registry) {
+        this(delegate, registry, PlanNamedTypes::name);
     }
 
-    public PlanStreamOutput(StreamOutput streamOutput, PlanNameRegistry registry, Function<Class<?>, String> nameSupplier) {
-        super(streamOutput);
+    public PlanStreamOutput(StreamOutput delegate, PlanNameRegistry registry, Function<Class<?>, String> nameSupplier) {
+        this.delegate = delegate;
         this.registry = registry;
         this.nameSupplier = nameSupplier;
     }
@@ -89,4 +90,35 @@ public <T> void writeNamed(Class<T> type, T value) throws IOException {
         writeString(name);
         writer.write(this, value);
     }
+
+    @Override
+    public void writeByte(byte b) throws IOException {
+        delegate.writeByte(b);
+    }
+
+    @Override
+    public void writeBytes(byte[] b, int offset, int length) throws IOException {
+        delegate.writeBytes(b, offset, length);
+    }
+
+    @Override
+    public void flush() throws IOException {
+        delegate.flush();
+    }
+
+    @Override
+    public void close() throws IOException {
+        delegate.close();
+    }
+
+    @Override
+    public TransportVersion getTransportVersion() {
+        return delegate.getTransportVersion();
+    }
+
+    @Override
+    public void setTransportVersion(TransportVersion version) {
+        delegate.setTransportVersion(version);
+        super.setTransportVersion(version);
+    }
 }

x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/io/stream/PlanStreamOutputTests.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.esql.io.stream;
+
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.common.io.stream.BytesStreamOutput;
+import org.elasticsearch.test.ESTestCase;
+import org.elasticsearch.test.TransportVersionUtils;
+
+import static org.hamcrest.Matchers.equalTo;
+
+public class PlanStreamOutputTests extends ESTestCase {
+
+    public void testTransportVersion() {
+        BytesStreamOutput out = new BytesStreamOutput();
+        TransportVersion v1 = TransportVersionUtils.randomCompatibleVersion(random());
+        out.setTransportVersion(v1);
+        PlanStreamOutput planOut = new PlanStreamOutput(out, PlanNameRegistry.INSTANCE);
+        assertThat(planOut.getTransportVersion(), equalTo(v1));
+        TransportVersion v2 = TransportVersionUtils.randomCompatibleVersion(random());
+        planOut.setTransportVersion(v2);
+        assertThat(planOut.getTransportVersion(), equalTo(v2));
+        assertThat(out.getTransportVersion(), equalTo(v2));
+    }
+}