Report kvikio I/O statistics per rank in the streaming engines

[madsbk]

This PR enables KvikIO statistics on every rank and gathers them on the client. Depends on rapidsai/kvikio#1036, which adds the monitor that does the counting.

Each rank turns on counting when the engine is configured with statistics=True, and StreamingEngine.gather_io_summary() brings back one kvikio.Summary per rank, keyed by rank index.

options = StreamingOptions(statistics=True)
with SPMDEngine(
    rapidsmpf_options=options.to_rapidsmpf_options(),
    executor_options=options.to_executor_options(),
    engine_options=options.to_engine_options(),
) as engine:
    pl.scan_parquet(path).select(pl.col("a").sum()).collect(engine=engine)
    for rank, summary in engine.gather_io_summary().items():
        print(f"--- rank {rank} ---")
        print(summary)

What a rank reports

KvikIO renders the report, so this PR formats nothing. A single-rank run of a parquet scan:

KvikIO I/O summary
  wall time            122.55 ms
  busy time            18.40 ms (15.02 % of the wall time)
  busy bandwidth       66.44 MB/s
  operations           12 (12 read, 0 write)
  mean duration        3.90 ms
  bytes                1.17 MiB of 1.17 MiB requested (1.17 MiB read, 0 B written)
  errors               0
  backend POSIX        1.17 MiB in 12 ops, 46.83 ms, 26.11 MB/s
  backend GDS          unused
  backend MMAP         unused
  backend REMOTE_HTTP  unused
  backend REMOTE_HDFS  unused

Busy time counts only the stretches with a read in flight, so busy bandwidth measures the storage rather than the query: this scan spent 15 % of its wall time reading, and dividing by the whole span would have reported it at a tenth of the rate the disk was really giving.

In the benchmarks

The PDS runners record the per-rank summaries on each iteration's record when --rapidsmpf-statistics is passed, so I/O stays queryable across a whole sweep rather than being printed once and lost.

print_results_file.py (new file) reads a results file back and prints it, since nothing existed that could. Timings and I/O side by side, one row per rank per iteration:

$ python -m cudf_polars.streaming.benchmarks.print_results_file pdsh-output.json

==============================================================================
run       : b6274527-a5d4-4791-9aa0-2a7a4ee3d307  (2026-08-20T08:35:11+00:00)
engine    : cudf-polars  frontend=ray
dataset   : /datasets/tpch-rs/scale-10-duckdb/  scale=10
workers   : 2  iterations=3
==============================================================================

Timings
   query  iters        min        max       mean
       1      3    0.1364s    0.3168s    0.2166s
       3      3    0.1616s    0.2154s    0.1824s
   total                                 0.3990s

I/O per rank
   query  iter  rank      ops         read       busy   busy%    bandwidth  backends
       1     0     0      796   312.09 MiB     64.0ms    9.8%     5.11GB/s  POSIX
       1     0     1      793   310.21 MiB     75.4ms   11.5%     4.31GB/s  POSIX
       1     1     0      788   306.54 MiB     40.1ms   29.1%     8.01GB/s  POSIX
       1     1     1      793   310.21 MiB     54.3ms   39.5%     5.99GB/s  POSIX
       1     2     0      788   306.54 MiB     37.3ms   18.9%     8.61GB/s  POSIX
       1     2     1      793   310.21 MiB     49.7ms   25.1%     6.54GB/s  POSIX
       3     0     0     1485   553.21 MiB     91.2ms   42.0%     6.36GB/s  POSIX
       3     0     1      793   339.17 MiB     82.1ms   37.8%     4.33GB/s  POSIX
       3     1     0     1474   549.74 MiB     68.6ms   40.0%     8.41GB/s  POSIX
       3     1     1      793   339.17 MiB     59.5ms   34.7%     5.98GB/s  POSIX
       3     2     0     1474   549.74 MiB     67.1ms   41.1%     8.59GB/s  POSIX
       3     2     1      793   339.17 MiB     61.3ms   37.6%     5.80GB/s  POSIX

  widest read skew: 1.63x (query 3, iteration 0)

Caveats worth knowing

• Counting is per process. With Ray and Dask each rank has a process to itself, so a summary covers only cudf-polars. With SPMD cudf-polars shares your script's process, so KvikIO operations your own code performs are counted too.
• Not all I/O is observed. Per KvikIO's Monitor docs the cuFile async API on a working GDS system and the batch API report nothing, and anything cudf-polars reads outside KvikIO is invisible.
• Timestamps are per rank. Each monitor takes its own clock anchor, so across hosts the start and end times carry whatever NTP skew exists. Ratios within a rank, busy_fraction and the bandwidths, are unaffected.

[copy-pr-bot]

This pull request requires additional validation before any workflows can run on NVIDIA's runners.

Pull request vetters can view their responsibilities here.

Contributors can view more details about this message here.

[copy-pr-bot]

Auto-sync is disabled for draft pull requests in this repository. Workflows must be run manually.

Contributors can view more details about this message here.