Add basic runtime profiling in cudf-polars

[rjzamora]

Description

• Closes [FEA] Add IR-node profiling for the rapidsmpf runtime #21244
• Part of [FEA][Story] Dynamic planning in cuDF-Polars #20482

Checklist

• I am familiar with the Contributing Guidelines.
• New or existing tests cover these changes.
• The documentation is up to date with these changes.

[rjzamora]

cc @TomAugspurger

[TomAugspurger]

Could you share a bit of background on the goal here, and what type of information we're trying to collect, and what our desired output looks like?

And I'll look more closely, but I'm especially curious whether we'd be able to integrate this into our cudf-polars tracing at https://github.com/rapidsai/cudf/blob/main/python/cudf_polars/cudf_polars/dsl/tracing.py. I think that we might already have some things like node count (if we mean the same thing for "node": a cudf-polars IR node):

❯ CUDF_POLARS_LOG_TRACES=1 python -m cudf_polars.experimental.benchmarks.pdsh --path /datasets/toaugspurger/tpch-rs/scale-10/ --suffix "" --iterations 1 1 --no-print-results --no-summarize
2026-01-30 13:50:01 [info     ] Execute IR                     count_frames_input=0 count_frames_output=1 frames_input=[] frames_output=[{'shape': (59142609, 7), 'size': 2957130458}] nvml_current_bytes_input=320864256 nvml_current_bytes_output=331350016 overhead_duration=2412492 rmm_current_bytes_input=0 rmm_current_bytes_output=2957130458 rmm_current_count_input=0 rmm_current_count_output=9 rmm_peak_bytes_input=0 rmm_peak_bytes_output=8592351777 rmm_peak_count_input=0 rmm_peak_count_output=120 rmm_total_bytes_input=0 rmm_total_bytes_output=10498498090 rmm_total_count_input=0 rmm_total_count_output=225 start=14767367239405135 stop=14767367600357750 total_bytes_input=0 total_bytes_output=2957130458 type=Scan
2026-01-30 13:50:01 [info     ] Execute IR                     count_frames_input=0 count_frames_output=1 frames_input=[] frames_output=[{'shape': (59142609, 8), 'size': 3430271330}] nvml_current_bytes_input=331350016 nvml_current_bytes_output=333447168 overhead_duration=1024010 rmm_current_bytes_input=2957130458 rmm_current_bytes_output=3430271330 rmm_current_count_input=9 rmm_current_count_output=10 rmm_peak_bytes_input=8592351777 rmm_peak_bytes_output=8592351777 rmm_peak_count_input=120 rmm_peak_count_output=120 rmm_total_bytes_input=10498498090 rmm_total_bytes_output=11444779860 rmm_total_count_input=225 rmm_total_count_output=232 start=14767367602411727 stop=14767367615460387 total_bytes_input=0 total_bytes_output=3430271330 type=HStack
2026-01-30 13:50:01 [info     ] Execute IR                     count_frames_input=1 count_frames_output=1 frames_input=[{'shape': (59142609, 8), 'size': 3430271330}] frames_output=[{'shape': (4, 10), 'size': 288}] nvml_current_bytes_input=333447168 nvml_current_bytes_output=339738624 overhead_duration=1327672 rmm_current_bytes_input=3430271330 rmm_current_bytes_output=3430271618 rmm_current_count_input=10 rmm_current_count_output=22 rmm_peak_bytes_input=8592351777 rmm_peak_bytes_output=8592351777 rmm_peak_count_input=120 rmm_peak_count_output=120 rmm_total_bytes_input=11444779860 rmm_total_bytes_output=13574741458 rmm_total_count_input=232 rmm_total_count_output=276 start=14767367616781167 stop=14767367739343388 total_bytes_input=3430271330 total_bytes_output=288 type=GroupBy
...

The list under frames_output have the shapes / sizes.

That's built on structlog, which I think gives us a couple of additional nice things:

1. We can rely on the normal logging config for collection / persisting to file
2. We can use context variables to propagate node-specific data we want to log (like a runtime decision on what join algorithm to use) from the IR.do_evaluate to the site where the record is logged without having to pass around object everywhere.

[rjzamora]

Right now, this just allows us to view the plan after the run with real row- and chunk-count statistics. For example, if I pass in the --profile-output-path to pdsh, I get something like this for q6 (running on 4 ranks):

SELECT ('revenue',) rows=4 chunks=4
  SELECT ('_______________1',) rows=4 chunks=4
    REPARTITION ('_______________0',) rows=204 chunks=4
      SELECT ('_______________0',) rows=51 chunks=51
        HSTACK ('l_extendedprice', 'l_discount', 'l_quantity', 'l_shipdate', 'revenue') rows=34.26 M chunks=51
          SCAN PARQUET ('l_extendedprice', 'l_discount', 'l_quantity', 'l_shipdate') rows=34.26 M chunks=51

This is "nice to have", but the my biggest motivation for pushing this in now is that I've also been traking a decision= metric in my dynamic-planning work. This way I can see what kind of algorithm was used for GroupBy/Join/Distinct. Otherwise, there is currently no way to know without print statements.

With that said, I'm very happy to do something different here. In the short term, I just need a mechanism to review an adaptive plan.

[rjzamora]

Closing in favor of #21309