Rename cudf-polars max_io_threads configuration and change meaning - #23569
Conversation
wence-
added
improvement
github-actions
Bot
added
Python
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughSummary by CodeRabbit
WalkthroughThe PR adds configurable concurrent streaming I/O tasks with memory admission tracing. It renames the executor option, removes native Parquet scan support and bindings, updates documentation and interfaces, and adds integration coverage for I/O-task admission ordering. ChangesStreaming I/O concurrency and scan execution
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 2 | ❌ 3❌ Failed checks (3 warnings)
✅ Passed checks (2 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 6
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py (1)
480-483: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftAdd a deprecation path for removed configuration options.
RunConfigis public. These lines removenative_parquetandmax_io_threadsfrom its constructor and the benchmark CLI. Existing callers will fail without a migration message.Retain deprecated compatibility handling for one release. Emit a deprecation warning. Do not silently map
max_io_threadstomax_concurrent_io_tasks, because the option now has per-Scansemantics. Give--native-parquetan explicit migration error after the warning because no equivalent remains.As per coding guidelines, “Detect and flag API breaking changes to public methods/attributes without deprecation warnings.”
Also applies to: 627-632, 658-661, 707-710, 737-740, 2032-2035
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py` around lines 480 - 483, Add one-release compatibility handling to the public RunConfig constructor and benchmark CLI for removed native_parquet and max_io_threads options. Emit deprecation warnings for both; reject native_parquet with an explicit migration error because no equivalent exists, and accept max_io_threads only as deprecated input without mapping it to max_concurrent_io_tasks, whose semantics are per-Scan. Apply this consistently to the referenced argument-parsing and configuration paths.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@python/cudf_polars/cudf_polars/dsl/tracing.py`:
- Line 59: Update the affected type unions in engine/options.py to place None
last, resolving all three Ruff RUF036 errors while preserving the existing types
and behavior.
In `@python/cudf_polars/cudf_polars/streaming/io.py`:
- Line 531: Update the type annotations at lines 752 and 809 in
python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py to use pl.GPUEngine
| None, resolving both Ruff RUF036 errors; no direct change is required at
python/cudf_polars/cudf_polars/streaming/io.py:531-531.
In `@python/cudf_polars/cudf_polars/utils/config.py`:
- Around line 810-812: The public configuration rename must retain deprecated
compatibility. In python/cudf_polars/cudf_polars/utils/config.py:810-812, accept
max_io_threads as an alias for max_concurrent_io_tasks with deterministic
conflict handling, DeprecationWarning, and environment-variable migration; in
python/cudf_polars/cudf_polars/engine/options.py:348-350, preserve the same
alias across StreamingOptions, argparse, and environment parsing; in
python/cudf_polars/cudf_polars/utils/config.py:274-303, retain
ParquetOptions.use_rapidsmpf_native or detect its legacy environment variable
and issue a clear migration warning. Document the deprecation period and planned
removal version.
- Around line 900-901: Update the validation for max_concurrent_io_tasks in the
relevant configuration class to accept only positive, non-boolean integers;
reject 0, negative values, and False with the existing TypeError behavior, and
add coverage for each rejected case.
In `@python/cudf_polars/docs/overview.md`:
- Around line 413-417: Update the documentation around max_concurrent_io_tasks
to explicitly note that max_io_threads was renamed and that concurrency is now
limited independently per scan node rather than globally. Add migration guidance
telling users to replace the removed option with max_concurrent_io_tasks so
existing configurations do not silently use the default.
In `@python/cudf_polars/tests/streaming/test_tracing.py`:
- Around line 180-187: Update the subprocess handling around Popen and
communicate in the test helper to catch TimeoutExpired, call proc.kill(),
collect remaining output with communicate(), and fail the test after cleanup.
Preserve the existing returncode and output handling when communicate completes
within the timeout, ensuring Popen.__exit__ cannot block indefinitely.
---
Outside diff comments:
In `@python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py`:
- Around line 480-483: Add one-release compatibility handling to the public
RunConfig constructor and benchmark CLI for removed native_parquet and
max_io_threads options. Emit deprecation warnings for both; reject
native_parquet with an explicit migration error because no equivalent exists,
and accept max_io_threads only as deprecated input without mapping it to
max_concurrent_io_tasks, whose semantics are per-Scan. Apply this consistently
to the referenced argument-parsing and configuration paths.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 4dd0f261-0935-4d78-b688-d5cdf2a31def
📒 Files selected for processing (19)
python/cudf_polars/cudf_polars/dsl/tracing.pypython/cudf_polars/cudf_polars/engine/options.pypython/cudf_polars/cudf_polars/streaming/actor_graph/core.pypython/cudf_polars/cudf_polars/streaming/actor_graph/dispatch.pypython/cudf_polars/cudf_polars/streaming/actor_graph/io.pypython/cudf_polars/cudf_polars/streaming/benchmarks/utils.pypython/cudf_polars/cudf_polars/streaming/io.pypython/cudf_polars/cudf_polars/utils/config.pypython/cudf_polars/docs/cudf-polars-mp.mdpython/cudf_polars/docs/overview.mdpython/cudf_polars/tests/streaming/test_options.pypython/cudf_polars/tests/streaming/test_parallel.pypython/cudf_polars/tests/streaming/test_scan.pypython/cudf_polars/tests/streaming/test_tracing.pypython/cudf_polars/tests/test_config.pypython/cudf_streaming/cudf_streaming/__init__.pypython/cudf_streaming/cudf_streaming/parquet.pyipython/cudf_streaming/cudf_streaming/parquet.pyxpython/cudf_streaming/cudf_streaming/tests/test_read_parquet.py
💤 Files with no reviewable changes (5)
- python/cudf_streaming/cudf_streaming/parquet.pyi
- python/cudf_streaming/cudf_streaming/parquet.pyx
- python/cudf_streaming/cudf_streaming/init.py
- python/cudf_polars/tests/streaming/test_scan.py
- python/cudf_streaming/cudf_streaming/tests/test_read_parquet.py
423a531 to
512d408
Compare
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@python/cudf_streaming/CMakeLists.txt`:
- Line 40: Retain cudf_streaming/parquet.pyx in the CMake cython_sources list
because io.py still imports Filter and read_parquet from cudf_streaming.parquet.
Do not remove the Parquet binding until that consumer is removed.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 7c2c3c0c-119d-46bd-acbf-e3e419891e2b
📒 Files selected for processing (20)
python/cudf_polars/cudf_polars/dsl/tracing.pypython/cudf_polars/cudf_polars/engine/options.pypython/cudf_polars/cudf_polars/streaming/actor_graph/core.pypython/cudf_polars/cudf_polars/streaming/actor_graph/dispatch.pypython/cudf_polars/cudf_polars/streaming/actor_graph/io.pypython/cudf_polars/cudf_polars/streaming/benchmarks/utils.pypython/cudf_polars/cudf_polars/streaming/io.pypython/cudf_polars/cudf_polars/utils/config.pypython/cudf_polars/docs/cudf-polars-mp.mdpython/cudf_polars/docs/overview.mdpython/cudf_polars/tests/streaming/test_options.pypython/cudf_polars/tests/streaming/test_parallel.pypython/cudf_polars/tests/streaming/test_scan.pypython/cudf_polars/tests/streaming/test_tracing.pypython/cudf_polars/tests/test_config.pypython/cudf_streaming/CMakeLists.txtpython/cudf_streaming/cudf_streaming/__init__.pypython/cudf_streaming/cudf_streaming/parquet.pyipython/cudf_streaming/cudf_streaming/parquet.pyxpython/cudf_streaming/cudf_streaming/tests/test_read_parquet.py
💤 Files with no reviewable changes (5)
- python/cudf_streaming/cudf_streaming/parquet.pyi
- python/cudf_streaming/cudf_streaming/tests/test_read_parquet.py
- python/cudf_polars/tests/streaming/test_scan.py
- python/cudf_streaming/cudf_streaming/parquet.pyx
- python/cudf_streaming/cudf_streaming/init.py
🚧 Files skipped from review as they are similar to previous changes (13)
- python/cudf_polars/docs/cudf-polars-mp.md
- python/cudf_polars/cudf_polars/streaming/actor_graph/dispatch.py
- python/cudf_polars/cudf_polars/dsl/tracing.py
- python/cudf_polars/tests/streaming/test_parallel.py
- python/cudf_polars/cudf_polars/streaming/actor_graph/core.py
- python/cudf_polars/tests/streaming/test_options.py
- python/cudf_polars/docs/overview.md
- python/cudf_polars/cudf_polars/utils/config.py
- python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py
- python/cudf_polars/cudf_polars/streaming/actor_graph/io.py
- python/cudf_polars/cudf_polars/engine/options.py
- python/cudf_polars/cudf_polars/streaming/io.py
- python/cudf_polars/tests/test_config.py
This was only implemented for some C++-only benchmarks. It is not used by default (or under most circumstances) in cudf-polars and has a number of performance problems. So we can purge a load of code we're not using.
512d408 to
dcd022b
Compare
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
There was a problem hiding this comment.
🧹 Nitpick comments (2)
python/cudf_polars/cudf_polars/streaming/actor_graph/io.py (2)
530-540: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument the
2 *reservation factor.
_process_and_send_chunkexplains each reservation multiplier in an inline comment. This branch uses an unexplained factor of two for non-DataFrameScanreads. Add a short comment that states the second estimate covers decode temporaries, so future tuning does not guess.♻️ Proposed comment
+ # Reserve the retained output plus, for file scans, one extra estimate for + # transient decode buffers. DataFrameScan slices need no decode scratch. reservation_bytes = ( estimated_chunk_bytes if isinstance(scan, DataFrameScan) else 2 * estimated_chunk_bytes )🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/cudf_polars/cudf_polars/streaming/actor_graph/io.py` around lines 530 - 540, Add a concise inline comment beside the non-DataFrameScan reservation multiplier in the reservation_bytes calculation, stating that the additional estimated chunk allocation covers decode temporaries. Leave the existing reservation logic unchanged.
683-697: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueCollapse the leftover
nodesconstruction.Only one node remains after the native Parquet path removal. Build
nodesas a literal instead of an empty dict followed by an assignment. Thech_outtemporary can also be inlined.♻️ Proposed simplification
- ch_out = channels[ir].reserve_input_slot() - nodes: dict[IR, list[Any]] = {} - - nodes[ir] = [ - scan_node( - rec.state["context"], - ir, - rec.state["ir_context"], - ch_out, - num_producers=num_producers, - estimated_chunk_bytes=( - plan.estimated_chunk_bytes or executor.target_partition_size - ), - ) - ] + nodes: dict[IR, list[Any]] = { + ir: [ + scan_node( + rec.state["context"], + ir, + rec.state["ir_context"], + channels[ir].reserve_input_slot(), + num_producers=num_producers, + estimated_chunk_bytes=( + plan.estimated_chunk_bytes or executor.target_partition_size + ), + ) + ] + }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/cudf_polars/cudf_polars/streaming/actor_graph/io.py` around lines 683 - 697, In the surrounding graph-building function, simplify the single-node setup by constructing nodes directly as a literal containing the scan_node result, and inline the reserve_input_slot() call instead of assigning ch_out. Preserve all existing scan_node arguments and the nodes mapping shape.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@python/cudf_polars/cudf_polars/streaming/actor_graph/io.py`:
- Around line 530-540: Add a concise inline comment beside the non-DataFrameScan
reservation multiplier in the reservation_bytes calculation, stating that the
additional estimated chunk allocation covers decode temporaries. Leave the
existing reservation logic unchanged.
- Around line 683-697: In the surrounding graph-building function, simplify the
single-node setup by constructing nodes directly as a literal containing the
scan_node result, and inline the reserve_input_slot() call instead of assigning
ch_out. Preserve all existing scan_node arguments and the nodes mapping shape.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: e39eebdb-cfcd-4600-a0e3-df859c5b9c0a
📒 Files selected for processing (20)
python/cudf_polars/cudf_polars/dsl/tracing.pypython/cudf_polars/cudf_polars/engine/options.pypython/cudf_polars/cudf_polars/streaming/actor_graph/core.pypython/cudf_polars/cudf_polars/streaming/actor_graph/dispatch.pypython/cudf_polars/cudf_polars/streaming/actor_graph/io.pypython/cudf_polars/cudf_polars/streaming/benchmarks/utils.pypython/cudf_polars/cudf_polars/streaming/io.pypython/cudf_polars/cudf_polars/utils/config.pypython/cudf_polars/docs/cudf-polars-mp.mdpython/cudf_polars/docs/overview.mdpython/cudf_polars/tests/streaming/test_options.pypython/cudf_polars/tests/streaming/test_parallel.pypython/cudf_polars/tests/streaming/test_scan.pypython/cudf_polars/tests/streaming/test_tracing.pypython/cudf_polars/tests/test_config.pypython/cudf_streaming/CMakeLists.txtpython/cudf_streaming/cudf_streaming/__init__.pypython/cudf_streaming/cudf_streaming/parquet.pyipython/cudf_streaming/cudf_streaming/parquet.pyxpython/cudf_streaming/cudf_streaming/tests/test_read_parquet.py
💤 Files with no reviewable changes (5)
- python/cudf_polars/tests/streaming/test_scan.py
- python/cudf_streaming/cudf_streaming/parquet.pyx
- python/cudf_streaming/cudf_streaming/init.py
- python/cudf_streaming/cudf_streaming/parquet.pyi
- python/cudf_streaming/cudf_streaming/tests/test_read_parquet.py
🚧 Files skipped from review as they are similar to previous changes (13)
- python/cudf_polars/cudf_polars/streaming/actor_graph/dispatch.py
- python/cudf_polars/tests/streaming/test_parallel.py
- python/cudf_polars/cudf_polars/streaming/io.py
- python/cudf_streaming/CMakeLists.txt
- python/cudf_polars/tests/streaming/test_options.py
- python/cudf_polars/cudf_polars/dsl/tracing.py
- python/cudf_polars/docs/overview.md
- python/cudf_polars/cudf_polars/streaming/actor_graph/core.py
- python/cudf_polars/tests/test_config.py
- python/cudf_polars/cudf_polars/utils/config.py
- python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py
- python/cudf_polars/docs/cudf-polars-mp.md
- python/cudf_polars/cudf_polars/engine/options.py
There was a problem hiding this comment.
Definitely makes sense to rename this and to remove the native-IO path (for now).
We originally limited the number of producers on each scan to 1 when the number of total scan actors was >= max_io_threads. With these changes, the total number of producers will be much larger for queries with multiple scan actors. Are you expecting the new memory-reservation tweaks to make this less risky? Have you tested this on a single H100 at all?
This now limits the number of concurrent scan tasks per logical parquet scan node in the execution DAG.
Previously we only waited for a memory reservation for the estimated output size. But reading files also typically needs some temporary allocations, so estimate that at the size of the output chunk too.
dcd022b to
0e8cd37
Compare
I ran on H100 at SF1K. On main, with all defaults the pdsh benchmarks take ~148s lukewarm. On this branch, with max-concurrent-io-tasks set to 2 (the default) and all defaults, we take ~138s. So a small gain. |
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (2)
python/cudf_polars/cudf_polars/streaming/io.py (1)
531-531: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDocument the streaming executor’s
chunkedoverride.
ParquetOptions.chunkedis public and defaults toTrue, but streaming scan lowering always replaces it withFalse. Update its documentation to state this behavior, or warn when streaming receiveschunked=True.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/cudf_polars/cudf_polars/streaming/io.py` at line 531, Document in the ParquetOptions API that streaming scan lowering overrides chunked=True to False, or add a warning when the streaming executor receives chunked=True; anchor the change around the parquet_options replacement in the streaming scan lowering path and preserve the existing chunked=False behavior.Source: Coding guidelines
python/cudf_polars/cudf_polars/streaming/actor_graph/io.py (1)
11-26: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the redundant
IRimport from theTYPE_CHECKINGblock.Scope.IO_TASK.valueis correctly defined as"io_task".🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/cudf_polars/cudf_polars/streaming/actor_graph/io.py` around lines 11 - 26, Remove the unused IR import from the TYPE_CHECKING imports in the module, leaving the existing Scope.IO_TASK.value usage unchanged.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@python/cudf_polars/tests/streaming/test_tracing.py`:
- Line 164: Update the generated query string in the test around q and
pl.scan_parquet so source is embedded using Python repr formatting rather than
raw interpolation, ensuring paths containing quotes or backslashes remain valid
Python string literals.
---
Nitpick comments:
In `@python/cudf_polars/cudf_polars/streaming/actor_graph/io.py`:
- Around line 11-26: Remove the unused IR import from the TYPE_CHECKING imports
in the module, leaving the existing Scope.IO_TASK.value usage unchanged.
In `@python/cudf_polars/cudf_polars/streaming/io.py`:
- Line 531: Document in the ParquetOptions API that streaming scan lowering
overrides chunked=True to False, or add a warning when the streaming executor
receives chunked=True; anchor the change around the parquet_options replacement
in the streaming scan lowering path and preserve the existing chunked=False
behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: c2d009d2-2c57-414e-99df-24a37e015b74
📒 Files selected for processing (20)
python/cudf_polars/cudf_polars/dsl/tracing.pypython/cudf_polars/cudf_polars/engine/options.pypython/cudf_polars/cudf_polars/streaming/actor_graph/core.pypython/cudf_polars/cudf_polars/streaming/actor_graph/dispatch.pypython/cudf_polars/cudf_polars/streaming/actor_graph/io.pypython/cudf_polars/cudf_polars/streaming/benchmarks/utils.pypython/cudf_polars/cudf_polars/streaming/io.pypython/cudf_polars/cudf_polars/utils/config.pypython/cudf_polars/docs/cudf-polars-mp.mdpython/cudf_polars/docs/overview.mdpython/cudf_polars/tests/streaming/test_options.pypython/cudf_polars/tests/streaming/test_parallel.pypython/cudf_polars/tests/streaming/test_scan.pypython/cudf_polars/tests/streaming/test_tracing.pypython/cudf_polars/tests/test_config.pypython/cudf_streaming/CMakeLists.txtpython/cudf_streaming/cudf_streaming/__init__.pypython/cudf_streaming/cudf_streaming/parquet.pyipython/cudf_streaming/cudf_streaming/parquet.pyxpython/cudf_streaming/cudf_streaming/tests/test_read_parquet.py
💤 Files with no reviewable changes (5)
- python/cudf_polars/tests/streaming/test_scan.py
- python/cudf_streaming/cudf_streaming/tests/test_read_parquet.py
- python/cudf_streaming/cudf_streaming/init.py
- python/cudf_streaming/cudf_streaming/parquet.pyx
- python/cudf_streaming/cudf_streaming/parquet.pyi
🚧 Files skipped from review as they are similar to previous changes (12)
- python/cudf_polars/docs/overview.md
- python/cudf_polars/docs/cudf-polars-mp.md
- python/cudf_streaming/CMakeLists.txt
- python/cudf_polars/tests/streaming/test_options.py
- python/cudf_polars/cudf_polars/dsl/tracing.py
- python/cudf_polars/cudf_polars/streaming/actor_graph/core.py
- python/cudf_polars/tests/streaming/test_parallel.py
- python/cudf_polars/cudf_polars/streaming/actor_graph/dispatch.py
- python/cudf_polars/cudf_polars/engine/options.py
- python/cudf_polars/tests/test_config.py
- python/cudf_polars/cudf_polars/utils/config.py
- python/cudf_polars/cudf_polars/streaming/benchmarks/utils.py
| parquet_options = config_options.parquet_options | ||
| partition_info = rec.state["partition_info"][ir] | ||
| num_producers = rec.state["max_io_threads"] | ||
| num_producers = rec.state["max_concurrent_io_tasks"] |
There was a problem hiding this comment.
cc @Matt711 I think we'll need to thread this new option to apply to SplitScan/FusedScan nodes for the hybrid reader?
There was a problem hiding this comment.
A scan_node is made with this many producers. In this branch, that feeds in to the number the read_chunk call. Each read_chunk then suspends for a memory reservation. So if you're keeping that same setup I don't think you need to do anything.
There was a problem hiding this comment.
Yeah, I don't think the hybrid-scan work is adding/changing any references to "max_io_threads" or it's behavior, so you should be good.
With that said, it may be worth using this branch to run a cloud benchmark (if it's not too painful). The new default behavior seems to be a win for local storage, but I don't think we've measured in the cloud yet?
|
/merge |
7 participants
Description
The cudf-polars streaming configuration option
max_io_threadsis misnamed for two reasons:The idea behind
max_io_threadsis to limit to maximum amount of concurrency available in read tasks during streaming execution. The idea is that we find the number of uniqueScannodes in an execution DAG and then let them each have at mostmax(1, max_io_threads // nscans)read tasks in flight. At the point we assign concurrency to reads we don't take into consideration how large the file is: does it even need more than one IO task, for example. Additionally, if we have more thanmax_io_threadsuniqueScannodes, then we can actually havenscansconcurrent IO tasks.Really what we're trying to do is avoid launching too many read tasks to blow through memory limits. In
read_chunkwe're already suspending before launching a read to obtain a reservation for the output size, so we already have most of the machinery we want.To fix the above deficiencies, we make two changes:
max_io_threadstomax_concurrent_io_tasksand document that it applies per-Scanrather than globally;Additionally, I add logging to the reads.
Finally, since the rapidsmpf C++
read_parquetactor has never supported all the use cases we use in cudf-polars, and had different behaviour in terms of numbers of concurrent tasks, just remove Python bindings to it. There were no other Python consumers, the existing cudf-streaming C++ benchmarks can still use the C++ actor.max_io_threadsfrom the public API and derive it automatically #23552Checklist