Rename cudf-polars max_io_threads configuration and change meaning

python/cudf_polars/cudf_polars/dsl/tracing.py

@@ -56,6 +56,7 @@ class Scope(enum.StrEnum):
 
     PLAN = "plan"
     ACTOR = "actor"
+    IO_TASK = "io_task"
     EVALUATE_IR_NODE = "evaluate_ir_node"
 
 

python/cudf_polars/cudf_polars/engine/options.py

@@ -223,6 +223,11 @@ class StreamingOptions:
         Env: ``CUDF_POLARS__EXECUTOR__NUM_PY_EXECUTORS``.
         Default: ``8``.
         Category: executor.
+    max_concurrent_io_tasks
+        Maximum concurrent IO tasks for each scan node.
+        Env: ``CUDF_POLARS__EXECUTOR__MAX_CONCURRENT_IO_TASKS``.
+        Default: ``2``.
+        Category: executor.
     fallback_mode
         Fallback behavior (``"warn"``, ``"raise"``, ``"silent"``).
         Env: ``CUDF_POLARS__EXECUTOR__FALLBACK_MODE``.
@@ -340,6 +345,9 @@ class StreamingOptions:
     num_py_executors: int | Unspecified = _opt(
         "executor", "CUDF_POLARS__EXECUTOR__NUM_PY_EXECUTORS", int
     )
+    max_concurrent_io_tasks: int | Unspecified = _opt(
+        "executor", "CUDF_POLARS__EXECUTOR__MAX_CONCURRENT_IO_TASKS", int
+    )
     fallback_mode: str | Unspecified = _opt(
         "executor", "CUDF_POLARS__EXECUTOR__FALLBACK_MODE"
     )
@@ -532,6 +540,7 @@ def _get(attr: str) -> Any:
             unbounded_file_read_cache=_get("unbounded_file_read_cache"),
             hardware_binding=_get("hardware_binding"),
             num_py_executors=_get("num_py_executors"),
+            max_concurrent_io_tasks=_get("max_concurrent_io_tasks"),
             fallback_mode=_get("fallback_mode"),
             max_rows_per_partition=_get("max_rows_per_partition"),
             broadcast_limit=_get("broadcast_limit"),
@@ -695,6 +704,16 @@ def _add_cli_args(parser: argparse.ArgumentParser) -> None:
                 Env: CUDF_POLARS__EXECUTOR__NUM_PY_EXECUTORS.
                 Built-in default: 8."""),
         )
+        g.add_argument(
+            "--max-concurrent-io-tasks",
+            dest="max_concurrent_io_tasks",
+            default=None,
+            type=int,
+            help=textwrap.dedent("""\
+                Maximum concurrent IO tasks for each scan node.
+                Env: CUDF_POLARS__EXECUTOR__MAX_CONCURRENT_IO_TASKS.
+                Built-in default: 2."""),
+        )
         g.add_argument(
             "--raise-on-fail",
             dest="raise_on_fail",

python/cudf_polars/cudf_polars/streaming/actor_graph/core.py

@@ -13,7 +13,6 @@
 
 import cudf_polars.dsl.tracing
 from cudf_polars.dsl.ir import (
-    DataFrameScan,
     Join,
     Union,
 )
@@ -23,7 +22,6 @@
     generate_ir_sub_network_wrapper,
     metadata_drain_node,
 )
-from cudf_polars.streaming.io import StreamingScan
 from cudf_polars.streaming.over import Over
 from cudf_polars.utils.config import SPMDContext
 
@@ -250,22 +248,15 @@ def generate_network(
     -------
     The network nodes and output hook.
     """
-    # Count the number of IO nodes and the number of IR dependencies
-    num_io_nodes: int = 0
+    # Count the number of IR dependencies
     ir_dep_count: defaultdict[IR, int] = defaultdict(int)
     for node in traversal([ir]):
-        if isinstance(node, (DataFrameScan, StreamingScan)):
-            num_io_nodes += 1
         for child in node.children:
             ir_dep_count[child] += 1
 
     # Determine which nodes need fanout
     fanout_nodes = determine_fanout_nodes(ir, partition_info, ir_dep_count)
 
-    # Get max_io_threads from config (default: 2)
-    max_io_threads_global = config_options.executor.max_io_threads
-    max_io_threads_local = max(1, max_io_threads_global // max(1, num_io_nodes))
-
     # Generate the network
     state: GenState = {
         "context": context,
@@ -274,7 +265,7 @@ def generate_network(
         "partition_info": partition_info,
         "fanout_nodes": fanout_nodes,
         "ir_context": ir_context,
-        "max_io_threads": max_io_threads_local,
+        "max_concurrent_io_tasks": config_options.executor.max_concurrent_io_tasks,
         "stats": stats,
         "collective_id_map": collective_id_map,
     }

python/cudf_polars/cudf_polars/streaming/actor_graph/dispatch.py

@@ -1,4 +1,4 @@
-# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES.
+# SPDX-FileCopyrightText: Copyright (c) 2025-2026, NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 # SPDX-License-Identifier: Apache-2.0
 """Dispatching for the RapidsMPF streaming runtime."""
 
@@ -51,9 +51,8 @@ class GenState(TypedDict):
         Dictionary mapping IR nodes to fanout information.
     ir_context
         The execution context for the IR node.
-    max_io_threads
-        The maximum number of IO threads to use for
-        a single IO node.
+    max_concurrent_io_tasks
+        The maximum number of concurrent IO tasks to use for a single IO node.
     stats
         Statistics collector.
     collective_id_map
@@ -66,7 +65,7 @@ class GenState(TypedDict):
     partition_info: MutableMapping[IR, PartitionInfo]
     fanout_nodes: dict[IR, FanoutInfo]
     ir_context: IRExecutionContext
-    max_io_threads: int
+    max_concurrent_io_tasks: int
     stats: StatsCollector
     collective_id_map: dict[IR, list[int]]