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.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Enterprise

Run ID: f6434de3-befa-4602-83fe-902c58383e78

📥 Commits

Reviewing files that changed from the base of the PR and between ba43628 and c2dde72.

📒 Files selected for processing (2)

• python/cudf_polars/cudf_polars/engine/dask.py
• python/cudf_polars/cudf_polars/engine/ray.py

Included review availability: Your plan provides up to 12 included reviews per hour; 10 remain after this review.

---

📝 Walkthrough

Summary by CodeRabbit

• New Features

  • Added per-rank I/O statistics for streaming GPU executions across supported engines.
  • Added options to collect, clear, and restart I/O measurement periods.
  • Benchmark results now include I/O summaries and read-skew diagnostics.
  • Added a command-line tool to review benchmark runs, timings, and I/O activity.

• Documentation

  • Expanded API documentation for I/O summary reporting.
  • Added guidance on collecting and interpreting per-rank I/O statistics.
  • Reduced irrelevant documentation warnings.

Walkthrough

Adds KvikIO per-rank I/O monitoring to Dask, Ray, and SPMD streaming engines. Benchmark records store I/O summaries, and a CLI reports timing and I/O metrics. Documentation and tests cover collection, clearing, disabled statistics, and reporting.

Changes

KvikIO I/O statistics

Layer / File(s)	Summary
Monitoring helpers and engine contract python/cudf_polars/cudf_polars/engine/core.py	Adds KvikIO monitor lifecycle helpers and the abstract StreamingEngine.gather_io_summary() interface.
Engine lifecycle and summary collection python/cudf_polars/cudf_polars/engine/dask.py, python/cudf_polars/cudf_polars/engine/ray.py, python/cudf_polars/cudf_polars/engine/spmd.py	Integrates monitor setup, reset, shutdown, and per-rank summary collection across Dask, Ray, and SPMD engines.
Benchmark record capture python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py	Collects cleared I/O summaries before RapidsMPF statistics and stores them in successful iteration records with string rank keys.
Benchmark reporting, documentation, and validation python/cudf_polars/cudf_polars/streaming/benchmarks/print_results_file.py, python/cudf_polars/tests/streaming/benchmarks/test_print_results_file.py, python/cudf_polars/tests/streaming/test_statistics.py, docs/cudf/source/cudf_polars/api.md, docs/cudf/source/cudf_polars/profiling.md, docs/cudf/source/conf.py	Adds benchmark result parsing and I/O reporting, documents the APIs and summary fields, suppresses KvikIO nitpick warnings, and tests collection and reporting behavior.

Estimated code review effort: 4 (Complex) | ~60 minutes

Merge Risk: 🟡 Moderate · up to c2dde

The change can omit benchmark I/O summaries when statistics are enabled during engine reset, and teardown failures can leave monitoring active, causing incomplete reporting and lifecycle cleanup problems. Merge readiness remains moderate until these bounded paths are fixed or explicitly accepted.

Suggested reviewers: wence-, tomaugspurger, vyasr

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly summarizes the primary change: reporting per-rank KvikIO I/O statistics in streaming engines.
Description check	✅ Passed	The description accurately explains KvikIO monitoring, per-rank collection, benchmark recording, output, dependencies, and limitations.
Docstring Coverage	✅ Passed	Docstring check was indeterminate for this PR — some files could not be analyzed in time. Not blocking.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

_{Comment @coderabbitai help to get the list of available commands.}