feat: improve do_write_fragments concurrency

[lichuang]

Problem Statement

The original do_write_fragments function processed data sequentially:

1. Read a batch of data
2. Write it to a file
3. Flush if file size/row limits exceeded
4. Repeat

This sequential approach had several limitations:

• I/O Bottlenecks: Writing one file blocked all other operations
• Poor Resource Utilization: CPU and network bandwidth were underutilized during I/O waits
• High Latency: Total time was sum of all write operations rather than parallel execution
• No Concurrency: Multiple files couldn't be written simultaneously even when resources were available

Solution: Concurrent Multi-File Write

The optimized version (do_write_fragments_concurrent) implements a parallel processing pipeline:

Architecture

┌─────────────────────────────────────────────────────────────┐
│                     Data Stream Input                        │
└────────────────────┬────────────────────────────────────────┘
                     │
                     ▼
          ┌──────────────────────┐
          │   Batch Reader       │
          └──────┬───────────────┘
                 │
          ┌──────▼───────────────┐
          │  Load Balancer       │
          │  (Round-Robin)       │
          └──────┬───────────────┘
                 │
        ┌────────┼────────┬────────┬────────┐
        ▼        ▼        ▼        ▼        ▼
   ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐
   │Worker 1│ │Worker 2│ │Worker 3│ │Worker 4│
   │Writer  │ │Writer  │ │Writer  │ │Writer  │
   └────┬───┘ └────┬───┘ └────┬───┘ └────┬───┘
        │          │          │          │
        ▼          ▼          ▼          ▼
    File 1     File 2     File 3     File 4

Key Components

1. Active Writer Pool: Maintains up to MAX_CONCURRENT_WRITERS (default: 4) active file writers
2. Smart Dispatch System: Distributes batches across writers using round-robin with capacity awareness
3. Asynchronous Completion: When a file reaches limits, finishing happens in parallel while new writes continue
4. Backpressure Control: Limits how many operations can be in-flight to prevent memory exhaustion

[wjones127]

Hi @lichuang. This PR is interesting. I was thinking about doing something similar soon.

However, one drawback to doing this as written is if you have a small stream of data, then you can end up writing several small files, which is sub-optimal.

There's no way to know just from a stream how big it is. But if the input is materialized or otherwise provides information about how big the input is, then we can make smarter decisions about parallelization. That's why I wanted to implement #4583 first, before we implement a parallelized writer.