Improve merge performance

[Bukhtawar]

Description

Optimizations Applied

1. Zero-allocation row comparison via arrow::row::RowConverter (row_compare.rs, cursor_optimized.rs)

The original merge extracted sort keys into a freshly allocated Vec on every row comparison. Flamegraph profiling showed 50% of sort key extraction time was spent in malloc. The fix: RowConverter pre-computes a comparable byte representation per batch at load time. Row comparison becomes a memcmp — no allocation, no type dispatch, no dynamic downcasting.

2. Precomputed column index mapping (schema_optimized.rs)

The original pad_batch_to_schema did schema.index_of(field.name()) (linear name scan) for every field on every batch. ColumnMapping precomputes the source→target index mapping once per cursor and reuses it via indexed lookup.

3. Tokio-based prefetch pool separation (cursor_optimized.rs, io_task.rs)

The original cursor prefetched next batches on the same 4-thread Rayon pool used for parallel column encoding. Under load, prefetch IO and CPU encoding competed for threads. The fix: prefetch uses tokio::spawn_blocking on the existing Tokio runtime (auto-scaling blocking pool), keeping the Rayon pool 100% available for CPU-bound encoding.

4. Single-chunk concat_batches skip (context.rs)

When only one batch is buffered before flush, the original still called concat_batches (full memcpy). Now it pops the single batch directly.

Benchmark Results
Config	Baseline	Optimized	Improvement
3 files × 50k (150k rows)	295 ms	236 ms	19.9% faster
5 files × 100k (500k rows)	869 ms	676 ms	22.2% faster
10 files × 100k (1M rows)	1586 ms	1368 ms	13.7% faster

Related Issues

Resolves # opensearch-project#21079

Check List

• Functionality includes testing.
• API changes companion pull request created, if applicable.
• Public documentation issue/PR created, if applicable.

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[Shailesh-Kumar-Singh]

Given that for nearly-sorted Parquet files we'd mostly hit Tier 2 (emit whole batch, 1 comparison) or Tier 3 (binary search for boundary to emit a prefix, ~17 comparisons for 100K batch), we're converting the sort columns for 100K rows but only comparing a handful.
Would this still be more efficient than on-demand extraction for just the rows we need?

[Shailesh-Kumar-Singh]

The initial plan we discussed was to parallelize column decoding on Rayon (similar to how we parallelize column encoding via into_par_iter during flush), but the current Parquet reader API doesn't expose per-column readers, reader.next() decodes all columns sequentially in a single call.

Agreed on moving to tokio and spawning a thread here so that prefetch IO and CPU encoding do not compete.
Now that the tokio pool is shared between disk writes and prefetch, lets keep a cap on the blocking thread count via .max_blocking_threads(), Open to inputs on the max cap we should keep here.