Speed up block serialization (#124394) (#124840)

Currently, we use NamedWriteable for serializing blocks. While
convenient, it incurs a noticeable performance penalty when pages
contain thousands of blocks. Since block types are small and already
centered in ElementType, we can safely switch from NamedWriteable to
typed code. For example, the NamedWriteable alone of a small page with
10K fields would be 180KB, whereas the new method reduces it to 10KB.
Below are the serialization improvements with FROM idx | LIMIT 10000
where the target index has 10K fields:

- write_exchange_response executed 173 times took: 73.2ms -> 26.7ms
- read_exchange_response executed 173 times took: 49.4ms -> 25.8ms

(cherry picked from commit 79a1626)

Author: dnhatn

docs/changelog/124394.yaml

@@ -0,0 +1,5 @@
+pr: 124394
+summary: Avoid `NamedWritable` in block serialization
+area: ES|QL
+type: enhancement
+issues: []

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

@@ -194,6 +194,7 @@ static TransportVersion def(int id) {
     public static final TransportVersion RETRY_ILM_ASYNC_ACTION_REQUIRE_ERROR_8_19 = def(8_841_0_07);
     public static final TransportVersion INFERENCE_CONTEXT_8_X = def(8_841_0_08);
     public static final TransportVersion ML_INFERENCE_DEEPSEEK_8_19 = def(8_841_0_09);
+    public static final TransportVersion ESQL_SERIALIZE_BLOCK_TYPE_CODE = def(8_841_0_10);
 
     /*
      * STOP! READ THIS FIRST! No, really,

x-pack/plugin/esql/compute/src/main/generated-src/org/elasticsearch/compute/data/BooleanBlock.java

@@ -9,14 +9,17 @@
 
 import org.apache.lucene.util.Accountable;
 import org.apache.lucene.util.RamUsageEstimator;
-import org.elasticsearch.common.io.stream.NamedWriteable;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.core.RefCounted;
 import org.elasticsearch.core.Releasable;
 import org.elasticsearch.core.ReleasableIterator;
 import org.elasticsearch.core.Releasables;
 import org.elasticsearch.index.mapper.BlockLoader;
 
+import java.io.IOException;
+
 /**
  * A Block is a columnar representation of homogenous data. It has a position (row) count, and
  * various data retrieval methods for accessing the underlying data that is stored at a given
@@ -35,7 +38,7 @@
  * <p> Block are immutable and can be passed between threads as long as no two threads hold a reference to
  * the same block at the same time.
  */
-public interface Block extends Accountable, BlockLoader.Block, NamedWriteable, RefCounted, Releasable {
+public interface Block extends Accountable, BlockLoader.Block, Writeable, RefCounted, Releasable {
     /**
      * The maximum number of values that can be added to one position via lookup.
      * TODO maybe make this everywhere?
@@ -329,6 +332,30 @@ static Block[] buildAll(Block.Builder... builders) {
         }
     }
 
+    /**
+     * Writes only the data of the block to a stream output.
+     * This method should be used when the type of the block is known during reading.
+     */
+    void writeTo(StreamOutput out) throws IOException;
+
+    /**
+     * Writes the type of the block followed by the block data to a stream output.
+     * This should be paired with {@link #readTypedBlock(BlockStreamInput)}
+     */
+    static void writeTypedBlock(Block block, StreamOutput out) throws IOException {
+        block.elementType().writeTo(out);
+        block.writeTo(out);
+    }
+
+    /**
+     * Reads the block type and then the block data from a stream input
+     * This should be paired with {@link #writeTypedBlock(Block, StreamOutput)}
+     */
+    static Block readTypedBlock(BlockStreamInput in) throws IOException {
+        ElementType elementType = ElementType.readFrom(in);
+        return elementType.reader.readBlock(in);
+    }
+
     /**
      * Serialization type for blocks: 0 and 1 replace false/true used in pre-8.14
      */