Avoid eager input materialization for GPU range shuffle

[sdrp713]

Fixes #15940 and fixes #15444.

Description

This PR changes GPU range shuffles to consume chunked Parquet input incrementally
instead of eagerly materializing multiple decoded full row batches.

CachedGpuBatchIterator normally drains a GPU data producer before returning, and
size based coalescing can retain an additional lookahead batch. For wide range-shuffle
inputs, this leaves multiple decoded batches live while the current batch is being
partitioned.

That behavior caused avoidable GPU spill, re-materialization, and semaphore
contention in wide Delta liquid-clustering workloads.

Implementation

• Adds an execution scoped marker around upstream iterator calls made by a GPU range
  shuffle.
• Stops size based coalescing after the first available input batch while that marker
  is active.
• Adds an internal opt in capability for GpuDataProducer implementations that can
  remain open across semaphore releases.
• Enables that capability for the chunked Parquet reader.
• Streams eligible Parquet output one batch at a time and reacquires the GPU semaphore
  before each producer interaction.
• Preserves eager caching for non range consumers and producers that do not explicitly
  opt into streaming.
• Closes the producer on exhaustion, failure, or Spark task completion.

Performance Results

1. A plain append with optimized writes and automatic compaction disabled.
2. An OPTIMIZE FULL operation on a table clustered by a column.
3. GPU range partitioning into 1,000 destination ranges.

The input contained:

• 5,887,163 rows
• 585 columns
• 30 input tasks
• 938 decoded input batches
• Approximately 58.32 GiB of wide-row input

The following configuration was used for both runs:

• Spark 3.5.3
• Two g4dn.8xlarge GPU executors
• 4 executor cores
• 60 GiB executor memory
• spark.sql.files.maxPartitionBytes=2g
• 512 MiB target GPU batch size
• RAPIDS shuffle manager
• Optimized writes disabled
• Automatic compaction disabled
• Key-only range-boundary sampling enabled

Only the range-shuffle input consumption implementation changed between the runs. The original implementation eagerly drained and retained decoded batches from the chunked Parquet reader. The optimized implementation streams one decoded batch at a time to range partitioning and avoids coalescing a subsequent wide batch while the current batch remains live.

Stage 24 metric	Eager input materialization	Streaming input batches	Improvement
Wall-clock duration	460.196 s	285.447 s	1.61× faster
End-to-end OPTIMIZE FULL duration	676.698 s	499.823 s	1.35× faster
Aggregate executor runtime	3,480.055 s	2,185.316 s	37.2% lower
Input rows	5,887,163	5,887,163	Unchanged
Decoded input batches	938	938	Unchanged
GPU decode time	326.788 s	189.608 s	42.0% lower
Scan time	2,910.188 s	1,234.929 s	2.36× lower
Threaded-writer input-fetch time	1,714.742 s	1,064.432 s	37.9% lower
GPU semaphore wait	1,698.724 s	253.192 s	85.1% lower
Host spill	83.18 GiB	0	Eliminated
Disk spill	21.57 GiB	0	Eliminated
Aggregate per-task peak GPU footprint	278.36 GiB	95.51 GiB	65.7% lower
Maximum task duration	177.991 s	112.793 s	36.6% lower
Maximum concurrent GPU tasks	1	4	Increased concurrency

Checklists

Documentation

• Updated for new or modified user-facing features or behaviors
• No user-facing change

Testing

• Added or modified tests to cover new code paths
• Covered by existing tests
  (Please provide the names of the existing tests in the PR description.)
• Not required

Performance

• Tests ran and results are added in the PR description
• Issue filed with a link in the PR description
• Not required

[greptile-apps]

Confidence Score: 4/5

The batching behavior appears sound, but the explicit GPU-resource cleanup requirement must be satisfied before merging; partial-consumption cleanup coverage should also be added.

Summary

• Preserves eager producer caching for consumers outside GPU range shuffles.
• Keeps eligible Parquet producers open across range-partitioning batches.
• Adds unit coverage for marker scope, one-batch coalescing, exhaustion cleanup, and fallback to eager caching.
• Requires ARM-compliant producer cleanup and stronger coverage of partial-consumption task cleanup.

Diagram

%%{init: {'theme': 'neutral'}}%%
flowchart LR
  A[Range shuffle requests upstream input] --> B[Activate RangeInputBatching marker]
  B --> C[Coalescer reads first available batch]
  C --> D{Producer supports semaphore release?}
  D -->|Yes: chunked Parquet| E[Stream one producer output]
  D -->|No| F[Eagerly cache producer outputs]
  E --> G[Range-partition current batch]
  G --> H[Semaphore may be released]
  H --> A
  E --> I[Close on exhaustion, failure, or task completion]

Loading

[gerashegalov]

The new stop-after-one condition applies to every AbstractGpuCoalesceIterator, including RequireSingleBatch and RequireSingleBatchWithFilter, even though the intended optimization is only for size-based, independently consumable batches.

[gerashegalov]

Returning RangeGpuDataProducerIterator here defers every producer.next until later, after the caller's retry block has returned. Both production Parquet paths deliberately create and drain CachedGpuBatchIterator inside RmmRapidsRetryIterator.withRetryNoSplit

[revans2]

My concern is that the performance numbers are for a single bad use case, but it does not show what happens on other use cases. I get that for a wide schema spilling can be problematic but what happens as we change the structure of the data? More rows, fewer columns/etc.