Dispatch SplitScan to HybridScanReader in the streaming engine - #22857
Dispatch SplitScan to HybridScanReader in the streaming engine#22857Matt711 wants to merge 30 commits into
SplitScan to HybridScanReader in the streaming engine#22857Conversation
Matt711
added
feature request
|
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. |
github-actions
Bot
added
libcudf
|
Thanks @Matt711 ! Just a quick note that most real-world S3 datasets that I see have files on the smaller side. Therefore, it makes perfect sense to start with The |
| return scans | ||
|
|
||
|
|
||
| def _read_with_hybrid_scan( |
There was a problem hiding this comment.
Benchmarks: TPC-H Q6, SF1k, one iteration
On main: 2 sec
This PR: 10 sec
I'm seeing large gaps between SplitScan tasks with hybrid scan (54ms -> 249ms ~4x longer). I'm not sure if this is due to python overhead in the two-pass approach or something else. Note I dont see how this could be due to the synchronous plc.io.parquet_io_utils.fetch_byte_ranges_to_device since this is inter-scan not intra-scan. I'm noting here because that might be a thought you had since plc.io.parquet_io_utils.fetch_byte_ranges_to_device is synchronous. Thatisn't to say we arnt paying for that synchroneity, but 5x slower is do to the inter-scan cost is my current hypothesis.
Thanks, yup
Oh nice, I think we should do this |
|
/ok to test 78b3ac9 |
|
/ok to test 7f2cba1 |
| """ | ||
| with opaque_memory_usage( | ||
| await reserve_memory( | ||
| with nvtx_annotate_cudf_polars(message="reserve_memory"): |
There was a problem hiding this comment.
Just a note that the annotations produced nvtx.annotate (which this nvtx_annotate_cudf_polars uses) spanning an await can be confusing.
The await will suspend execution here until the future resolves. The thread executing this coroutine might move onto another read_chunk task, and push another "reserve_memory" nvtx range onto the stack, making it look like a child of the first nvtx.annotate.
The begin_range() / end_range() APIs might be less confusing. #22718 ran into similar issues.
There was a problem hiding this comment.
Ah good call out, thanks
|
/ok to test a027ad9 |
|
/ok to test 258c008 |
|
Superceded by #23317 |
5 participants
Description
Hybrid scan lets us have more control over what happens when executing a
Scan. By default, the streaming engine usescudf::io::read_parquetto compute eachScan. Usingcudf::io::parquet::experimental::HybridScanReaderin cudf_polars essentially lets us break up the I/O and compute that happens during a call tocudf::io::read_parquet. In hybrid-scan speak, we split the read into two passes: first we read only the filter columns and compute a row mask, then we read only the payload columns for rows that survive the filter and combine. The typical benefit is for a selective filter. We only transfer the payload columns for rows that pass, rather than reading everything upfront and discarding filtered rows afterward.For remote IO,
HybridScanReaderexposes the exact byte ranges needed for the filter columns and payload columns separately. We can prefetch the filter column byte ranges asynchronously into pinned host memory and transfer to device memory. For row groups that don't survive the filter, the payload byte ranges are never fetched at all. In this way, we aren't bottlenecked by transferring payload data that will only be discarded. This will be done in a folow-up PR.We only dispatch to
HybridScanReaderforSplitScans in the streaming engine.SplitScanis used by the streaming engine to parallelize reads of a single large file. This is good for the single-fileHybridScanReader.FusedScanfuses multiple smaller files into one (logically) and AFAICT would be a good candidate for the multi-file hybrid scan reader (see #22583). Finally for the in-memory engine, there isn't a good justification for using hybrid scan (prefer streaming over in-memory always).Dispatch is also conditional on having a filter predicate (we'll fallback to default libcudf parquet reader otherwise).
So for this first PR: Enabling
CUDF_POLARS__PARQUET_OPTIONS__USE_HYBRID_SCAN=1means dispatchingSplitScans toHybridScanReaderwhenever there's a filter predicate (that we support translating to libcudf).Checklist