[VL] Support multiple segments per partition in columnar shuffle

[guowangy]

What changes are proposed in this pull request?

Introduces multi-segment-per-partition support in the Velox backend columnar shuffle writer, enabling incremental flushing of partition data to the final data file during processing — reducing peak memory usage without requiring full in-memory buffering or temporary spill files. The implementation can reduce total latency of TPC-H(SF6T) by ~16% using sort-based shuffle with low memory capacity in 2-socket Xeon 6960P system.

New index file format (ColumnarIndexShuffleBlockResolver)

Extends IndexShuffleBlockResolver with a new index format supporting multiple (offset, length) segments per partition:

[Partition Index: (N+1) × 8-byte big-endian offsets]
[Segment Data: per-partition list of (data_offset, length) pairs, each 8 bytes]
[1-byte end marker]  ← distinguishes from legacy format (size always multiple of 8)

ColumnarShuffleManager now uses this resolver. Multi-segment mode activates only when external shuffle service, push-based shuffle, and dictionary encoding are all disabled (dictionary encoding requires all-batches-complete before writing).

New I/O abstractions

• FileSegmentsInputStream — InputStream over non-contiguous (offset, size) file segments; supports zero-copy native reads via read(destAddress, maxSize)
• FileSegmentsManagedBuffer — ManagedBuffer backed by discontiguous segments; supports nioByteBuffer(), createInputStream(), convertToNetty()
• DiscontiguousFileRegion — Netty FileRegion mapping a logical range to multiple physical segments for zero-copy network transfer
• LowCopyFileSegmentsJniByteInputStream — zero-copy JNI wrapper over FileSegmentsInputStream; wired into JniByteInputStreams.create()

C++ LocalPartitionWriter changes

• usePartitionMultipleSegments_ flag + partitionSegments_ vector tracking (start, length) per partition
• flushCachedPayloads() — incremental flush after each hashEvict
• writeMemoryPayload() — direct write to final data file during sortEvict
• writeIndexFile() — serializes the new index at stop time
• PayloadCache::writeIncremental() — flushes completed (non-active) partitions without touching the in-use partition

JNI/JVM wiring

LocalPartitionWriterJniWrapper and JniWrapper.cc accept a new optional indexFile parameter; ColumnarShuffleWriter passes the temp index file path when multi-segment mode is active.

How was this patch tested?

New unit test suites:

• ColumnarIndexShuffleBlockResolverSuite — index format read/write, format detection, multi-segment block lookup
• FileSegmentsInputStreamSuite — sequential reads, multi-segment traversal, skip, zero-copy native reads
• FileSegmentsManagedBufferSuite — nioByteBuffer, createInputStream, convertToNetty, EOF and mmap edge cases
• DiscontiguousFileRegionSuite — Netty transfer across discontiguous segments, lazy open
• LowCopyFileSegmentsJniByteInputStreamTest — JNI wrapper correctness for ByteInputStream

Was this patch authored or co-authored using generative AI tooling?

[github-actions]

Run Gluten Clickhouse CI on x86

[marin-ma]

@guowangy Thanks for contributing this feature. Please check my comments below.

[marin-ma]

Can you add some c++ unit tests for the multi-segment partition write?

[marin-ma]

Please move the header to the top, and move htoll into anonymous namespace after the headers.

[marin-ma]

Please add a configuration to enable this feature.

[marin-ma]

Can you explain a bit more on how this can reduce the memory usage? Looks like the memory is still only get reclaimed by OOM and spilling.

[marin-ma]

The hashEvict is not only called for spilling. When the evictType is kCache, then it try to cache as much payload in memory as possible to reduce spilling.

And when the evitType is kSpill, the data will be written to a spilled data file. Two evict types can exist in the same job. Is evictType == kSpill being properly handled for multi-segments write?

[marin-ma]

The implementation can reduce total latency of TPC-H(SF6T) by ~16% using sort-based shuffle with low memory capacity in 2-socket Xeon 6960P system.

Can you explain where this improvement mainly comes from?

Currently we follow the same file layout as vanilla spark to have each partition output contiguous. I think one major benefit for this design is to reduce small random disk IO from the shuffle reader side. If memory is tight then the spill will be triggered more frequently, and it will be more likely to produce small output blocks for each partition. In this case this design will not be IO friendly.