[SPARK-56279][CORE] Enable zero-copy sendfile for FileRegion in native Netty transports

### What changes were proposed in this pull request?

This PR modifies `MessageEncoder` to emit the header `ByteBuf` and `FileRegion` as **separate objects** in the outbound message list when the body is a `FileRegion` backed by `FileSegmentManagedBuffer`, instead of wrapping them together in a `MessageWithHeader`.

Previously, all messages with a body were unconditionally wrapped in `MessageWithHeader`. This caused native transports (EPOLL, KQUEUE) to fall into a generic `FileRegion.transferTo()` fallback path that copies data through user-space, bypassing the optimized `sendfile()` / `splice()` zero-copy path that Netty's native transports provide.

The split is only applied when the `ManagedBuffer` is a `FileSegmentManagedBuffer`, whose `release()` is a no-op, making it safe to emit the `FileRegion` independently of write lifecycle management. Other `ManagedBuffer` types (e.g., `BlockManagerManagedBuffer`) still use the `MessageWithHeader` wrapper because they perform resource cleanup in `release()` that must be tied to `MessageWithHeader.deallocate()`.

### Why are the changes needed?

When using native transports (AUTO/EPOLL on Linux), file-backed shuffle fetch performance was severely degraded compared to NIO mode. The root cause lies in how Netty's native transports dispatch `FileRegion` writes.

In `AbstractEpollStreamChannel.doWriteSingle()` (and the analogous KQueue path), Netty uses an `instanceof` check to choose between two write strategies:

https://github.com/netty/netty/blob/eeb5674526f0b49a142580686a5a9a7147ddadec/transport-classes-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java#L474-L493

```java
} else if (msg instanceof DefaultFileRegion) {
    return writeDefaultFileRegion(in, (DefaultFileRegion) msg);  // → socket.sendFile() (zero-copy)
} else if (msg instanceof FileRegion) {
    return writeFileRegion(in, (FileRegion) msg);                // → region.transferTo() (user-space copy)
}
```

- **`writeDefaultFileRegion()`** calls `socket.sendFile()`, which maps directly to the Linux `sendfile()` syscall — a true zero-copy path where data is transferred from the file page cache to the socket buffer entirely within the kernel, with no user-space copy.

https://github.com/netty/netty/blob/eeb5674526f0b49a142580686a5a9a7147ddadec/transport-classes-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java#L367-L386

```java
    private int writeDefaultFileRegion(ChannelOutboundBuffer in, DefaultFileRegion region) throws Exception {
        final long offset = region.transferred();
        final long regionCount = region.count();
        if (offset >= regionCount) {
            in.remove();
            return 0;
        }

        final long flushedAmount = socket.sendFile(region, region.position(), offset, regionCount - offset);
        if (flushedAmount > 0) {
            in.progress(flushedAmount);
            if (region.transferred() >= regionCount) {
                in.remove();
            }
            return 1;
        } else if (flushedAmount == 0) {
            validateFileRegion(region, offset);
        }
        return WRITE_STATUS_SNDBUF_FULL;
    }
```

- **`writeFileRegion()`** falls back to `region.transferTo(WritableByteChannel)`, which writes data through a `SocketWritableByteChannel` wrapper — effectively a user-space copy path.

https://github.com/netty/netty/blob/eeb5674526f0b49a142580686a5a9a7147ddadec/transport-classes-epoll/src/main/java/io/netty/channel/epoll/AbstractEpollStreamChannel.java#L402-L420

```java
private int writeFileRegion(ChannelOutboundBuffer in, FileRegion region) throws Exception {
        if (region.transferred() >= region.count()) {
            in.remove();
            return 0;
        }

        if (byteChannel == null) {
            byteChannel = new EpollSocketWritableByteChannel();
        }
        final long flushedAmount = region.transferTo(byteChannel, region.transferred());
        if (flushedAmount > 0) {
            in.progress(flushedAmount);
            if (region.transferred() >= region.count()) {
                in.remove();
            }
            return 1;
        }
        return WRITE_STATUS_SNDBUF_FULL;
    }
```

Spark's `MessageWithHeader extends AbstractFileRegion` (not `DefaultFileRegion`). When `MessageEncoder` wraps a `DefaultFileRegion` body inside `MessageWithHeader`, the resulting object is a generic `FileRegion` from Netty's perspective. This means Netty dispatches it to the `writeFileRegion()` fallback, which calls `MessageWithHeader.transferTo()`:

```java
// MessageWithHeader.java, line 121
if (body instanceof FileRegion fileRegion) {
    writtenBody = fileRegion.transferTo(target, totalBytesTransferred - headerLength);
}
```

Here, even though the inner body is a `DefaultFileRegion`, its `transferTo()` is invoked with a `WritableByteChannel` (not a file descriptor), so the data is read from the file into a user-space buffer and then written to the socket — **the zero-copy opportunity is lost**.

By emitting the `DefaultFileRegion` directly into Netty's outbound buffer (instead of wrapping it in `MessageWithHeader`), Netty's native transport recognizes it via `instanceof DefaultFileRegion` and routes it to `socket.sendFile()`, restoring the zero-copy `sendfile()` path.

**Benchmark results (File-Backed Shuffle Fetch) show dramatic improvement:**

| Scenario | Before (ms) | After (ms) | Improvement |
|---|---|---|---|
| AUTO, sequential fetch (JDK17) | 2075 | 362 | **~5.7x faster** |
| AUTO, parallel fetch (JDK17) | 676 | 202 | **~3.3x faster** |
| AUTO, sequential fetch (JDK21) | 2093 | 360 | **~5.8x faster** |
| AUTO, parallel fetch (JDK21) | 677 | 204 | **~3.3x faster** |
| AUTO, sequential fetch (JDK25) | 1918 | 411 | **~4.7x faster** |
| AUTO, parallel fetch (JDK25) | 652 | 196 | **~3.3x faster** |

With this fix, AUTO (EPOLL) mode now matches or exceeds NIO performance for file-backed shuffle fetches, achieving the expected ~1.0X relative performance (previously ~0.2X).

### Does this PR introduce _any_ user-facing change?

No. This is an internal optimization to the Netty transport layer. Users benefit from improved shuffle fetch performance when using native transports (the default on Linux) without any configuration changes.

### How was this patch tested?

- Updated `MergedBlockMetaSuccessSuite` to validate the new encoder output format (2 separate objects: `ByteBuf` header + `DefaultFileRegion`, instead of a single `MessageWithHeader`).
- Re-ran `NettyTransportBenchmark` across JDK 17, JDK 21, and JDK 25 to confirm the performance improvement. Updated benchmark result files accordingly.

### Was this patch authored or co-authored using generative AI tooling?
Generated-by:  Claude Code 4.6

Closes #55087 from LuciferYang/SPARK-56279.

Authored-by: yangjie01 <yangjie01@baidu.com>
Signed-off-by: yangjie01 <yangjie01@baidu.com>

Author: LuciferYang

common/network-common/src/main/java/org/apache/spark/network/protocol/MessageEncoder.java

@@ -22,12 +22,14 @@
 import io.netty.buffer.ByteBuf;
 import io.netty.channel.ChannelHandler;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.FileRegion;
 import io.netty.handler.codec.MessageToMessageEncoder;
 
 import org.apache.spark.internal.LogKeys;
 import org.apache.spark.internal.SparkLogger;
 import org.apache.spark.internal.SparkLoggerFactory;
 import org.apache.spark.internal.MDC;
+import org.apache.spark.network.buffer.FileSegmentManagedBuffer;
 
 /**
  * Encoder used by the server side to encode server-to-client responses.
@@ -89,9 +91,24 @@ public void encode(ChannelHandlerContext ctx, Message in, List<Object> out) thro
     assert header.writableBytes() == 0;
 
     if (body != null) {
-      // We transfer ownership of the reference on in.body() to MessageWithHeader.
-      // This reference will be freed when MessageWithHeader.deallocate() is called.
-      out.add(new MessageWithHeader(in.body(), header, body, bodyLength));
+      if (body instanceof FileRegion && in.body() instanceof FileSegmentManagedBuffer) {
+        // Emit header and FileRegion as separate objects so that native transports
+        // (EPOLL, KQUEUE) can apply zero-copy sendfile/splice on the FileRegion directly.
+        // When wrapped in MessageWithHeader, native transports fall into a generic
+        // FileRegion.transferTo() fallback that copies data through user-space, bypassing
+        // the optimized sendfile() path.
+        //
+        // This split is only safe when the ManagedBuffer is FileSegmentManagedBuffer,
+        // whose release() is a no-op. Other ManagedBuffer types (e.g.,
+        // BlockManagerManagedBuffer) perform resource cleanup in release() that must be
+        // tied to the write lifecycle via MessageWithHeader.deallocate().
+        out.add(header);
+        out.add(body);
+      } else {
+        // We transfer ownership of the reference on in.body() to MessageWithHeader.
+        // This reference will be freed when MessageWithHeader.deallocate() is called.
+        out.add(new MessageWithHeader(in.body(), header, body, bodyLength));
+      }
     } else {
       out.add(header);
     }

common/network-common/src/test/java/org/apache/spark/network/protocol/MergedBlockMetaSuccessSuite.java

@@ -28,6 +28,7 @@
 import io.netty.buffer.ByteBufAllocator;
 import io.netty.buffer.Unpooled;
 import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.DefaultFileRegion;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.roaringbitmap.RoaringBitmap;
@@ -70,14 +71,25 @@ public void testMergedBlocksMetaEncodeDecode() throws Exception {
     when(context.alloc()).thenReturn(ByteBufAllocator.DEFAULT);
 
     MessageEncoder.INSTANCE.encode(context, expectedMeta, out);
-    Assertions.assertEquals(1, out.size());
-    MessageWithHeader msgWithHeader = (MessageWithHeader) out.remove(0);
+    // FileSegmentManagedBuffer.convertToNetty() returns a DefaultFileRegion,
+    // so MessageEncoder splits the output into header ByteBuf + FileRegion.
+    Assertions.assertEquals(2, out.size());
+    Assertions.assertInstanceOf(ByteBuf.class, out.get(0));
+    Assertions.assertInstanceOf(DefaultFileRegion.class, out.get(1));
+    ByteBuf headerBuf = (ByteBuf) out.remove(0);
+    DefaultFileRegion fileRegion = (DefaultFileRegion) out.remove(0);
 
-    ByteArrayWritableChannel writableChannel =
-      new ByteArrayWritableChannel((int) msgWithHeader.count());
-    while (msgWithHeader.transfered() < msgWithHeader.count()) {
-      msgWithHeader.transferTo(writableChannel, msgWithHeader.transfered());
+    int totalLen = headerBuf.readableBytes() + (int) fileRegion.count();
+    ByteArrayWritableChannel writableChannel = new ByteArrayWritableChannel(totalLen);
+    // Write header
+    writableChannel.write(headerBuf.nioBuffer());
+    // Write file region body
+    fileRegion.open();
+    while (fileRegion.transferred() < fileRegion.count()) {
+      fileRegion.transferTo(writableChannel, fileRegion.transferred());
     }
+    fileRegion.release();
+    headerBuf.release();
     ByteBuf messageBuf = Unpooled.wrappedBuffer(writableChannel.getData());
     messageBuf.readLong(); // frame length
     MessageDecoder.INSTANCE.decode(mock(ChannelHandlerContext.class), messageBuf, out);

core/benchmarks/NettyTransportBenchmark-jdk21-results.txt

@@ -6,7 +6,7 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 RPC Latency (1 KB):                       Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-1 KB payload                                        493            539          31          0.0       98616.6       1.0X
+1 KB payload                                        495            512          13          0.0       99079.2       1.0X
 
 
 ================================================================================================
@@ -17,7 +17,7 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 RPC Latency (64 KB):                      Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-64 KB payload                                       737            758          18          0.0      147330.6       1.0X
+64 KB payload                                       690            715          15          0.0      138099.8       1.0X
 
 
 ================================================================================================
@@ -28,7 +28,7 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 RPC Latency (1 MB):                       Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-1 MB payload                                        829            899          90          0.0      828572.3       1.0X
+1 MB payload                                        884            944          57          0.0      883906.3       1.0X
 
 
 ================================================================================================
@@ -39,7 +39,7 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 RPC Latency (16 MB):                      Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-16 MB payload                                      1271           1351          72          0.0    12713142.1       1.0X
+16 MB payload                                      1275           1303          22          0.0    12752225.6       1.0X
 
 
 ================================================================================================
@@ -50,10 +50,10 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 Concurrent RPC Throughput:                Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-1 client(s)                                        1936           2030          85          0.0       96813.3       1.0X
-4 client(s)                                         726            748          32          0.0       36305.7       2.7X
-8 client(s)                                         548            550           1          0.0       27417.0       3.5X
-16 client(s)                                        445            458          14          0.0       22238.6       4.4X
+1 client(s)                                        1883           1927          46          0.0       94151.5       1.0X
+4 client(s)                                         726            742          16          0.0       36282.9       2.6X
+8 client(s)                                         523            530           9          0.0       26136.4       3.6X
+16 client(s)                                        424            437          12          0.0       21221.7       4.4X
 
 
 ================================================================================================
@@ -64,8 +64,8 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 IOMode Comparison:                        Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-NIO (8 clients)                                     792            803          19          0.0       39590.9       1.0X
-AUTO (8 clients)                                    835            851          16          0.0       41747.2       0.9X
+NIO (8 clients)                                     811            829          17          0.0       40567.8       1.0X
+AUTO (8 clients)                                    848            874          45          0.0       42380.6       1.0X
 
 
 ================================================================================================
@@ -76,11 +76,11 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 Server Thread Scaling:                    Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-2 server threads                                    481            491          12          0.0       24050.8       1.0X
-4 server threads                                    391            413          29          0.1       19537.1       1.2X
-8 server threads                                    416            434          15          0.0       20819.9       1.2X
-16 server threads                                   449            473          26          0.0       22455.0       1.1X
-32 server threads                                   461            499          33          0.0       23043.3       1.0X
+2 server threads                                    465            477          15          0.0       23261.5       1.0X
+4 server threads                                    386            397          14          0.1       19278.9       1.2X
+8 server threads                                    415            431          21          0.0       20738.6       1.1X
+16 server threads                                   435            472          34          0.0       21760.2       1.1X
+32 server threads                                   479            497          28          0.0       23925.6       1.0X
 
 
 ================================================================================================
@@ -91,9 +91,9 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 Multi-Connection Throughput:              Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-1 conn(s), 4 threads                               1601           1667         108          0.0      320283.2       1.0X
-2 conn(s), 4 threads                               1189           1505         274          0.0      237803.7       1.3X
-4 conn(s), 4 threads                               1754           1793          34          0.0      350832.7       0.9X
+1 conn(s), 4 threads                               1853           2013         167          0.0      370521.8       1.0X
+2 conn(s), 4 threads                               1557           1588          28          0.0      311449.1       1.2X
+4 conn(s), 4 threads                               1756           1822          57          0.0      351214.2       1.1X
 
 
 ================================================================================================
@@ -104,9 +104,9 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 Async Write Throughput:                   Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-1 KB async burst                                     37             40           3          0.1        7439.9       1.0X
-64 KB async burst                                   145            155          13          0.0       29093.7       0.3X
-1 MB async burst                                   1539           1571          33          0.0      307765.1       0.0X
+1 KB async burst                                     34             48          21          0.1        6745.3       1.0X
+64 KB async burst                                   163            167           6          0.0       32507.6       0.2X
+1 MB async burst                                   1512           1574          75          0.0      302329.1       0.0X
 
 
 ================================================================================================
@@ -117,8 +117,8 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 16 MB Block Transfer:                     Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-Sequential sends                                    627            834         304          0.0     6267294.2       1.0X
-4-thread parallel sends                             452            457           7          0.0     4520142.4       1.4X
+Sequential sends                                    658           1087         373          0.0     6575971.7       1.0X
+4-thread parallel sends                             419            477          88          0.0     4187438.8       1.6X
 
 
 ================================================================================================
@@ -129,9 +129,9 @@ OpenJDK 64-Bit Server VM 21.0.10+7-LTS on Linux 6.17.0-1008-azure
 AMD EPYC 7763 64-Core Processor
 File-Backed Shuffle Fetch:                Best Time(ms)   Avg Time(ms)   Stdev(ms)    Rate(M/s)   Per Row(ns)   Relative
 ------------------------------------------------------------------------------------------------------------------------
-NIO, sequential fetch                               370            384          13          0.0     3700370.3       1.0X
-NIO, parallel fetch (4 clients)                     205            213           7          0.0     2054174.4       1.8X
-AUTO, sequential fetch                             2093           2189          85          0.0    20929166.7       0.2X
-AUTO, parallel fetch (4 clients)                    677            787         113          0.0     6769621.7       0.5X
+NIO, sequential fetch                               366            372           9          0.0     3655386.3       1.0X
+NIO, parallel fetch (4 clients)                     200            211          11          0.0     2002026.0       1.8X
+AUTO, sequential fetch                              360            369          13          0.0     3600497.3       1.0X
+AUTO, parallel fetch (4 clients)                    204            206           2          0.0     2041001.7       1.8X