tschwarzinger opened a new pull request, #24921: URL: https://github.com/apache/datafusion/pull/24921
## Which issue does this PR close? Related to: - https://github.com/apache/datafusion/issues/18942 - https://github.com/apache/datafusion/pull/19761 - https://github.com/apache/datafusion/issues/20778 Happy to make a separate issue if further discussion is needed. ## Rationale for this change Currently, executing a deep join tree is done sequentially. First table `A` is scanned, the hash table is being computed, dynamic filters are updated, and then table `B` is scanned. This is in general not a bad thing. The rows that become available during scanning table `A` may help us to significantly prune the scan on table `B` due to dynamic filters, speeding up the overall query. However, when a deep join tree requires scanning many --- possibly small tables --- the sequential scans will lead to high latency in an object-store-based environment. First table `A` is scanned, after 150 ms table `B` is scanned, etc. Especially in graph query languages like SPARQL (which we use DF for), deep join trees are quite common as the graph patterns get translated into a series of joins. This can cause high query latencies when using object stores. Ideally, we would like to execute queries by i) scanning all small table scans eagerly and ii) delaying larger scans until dynamic filters have been computed. The rationale is that, when the scans are small (e.g., a single request), dynamic filters likely will not improve the performance by a large margin (maybe except if the filter remove all data). Then, all small scans (and the first scan in the join tree) are fetched in parallel, while large scans wait until their dynamic filters have been computed, avoiding the problems of why #19761 is not yet the default beahvior in DF. #19761 is a related ticket that, in general, aims to eagerly buffer the probe side of hash joins to cut down latency. Currently, this is an optional feature and can be enabled by setting `config.execution.hash_join_buffering_capacity` to non-zero. This will cause the optimizer to insert `BufferExec` at the probe sides of the join. Firstly, I was thinking about extending this solution and basically add a `BufferExec` more aggressively by default when the stats suggest that the result is small. This might be the better approach. However, it could be that the results is small but requires much computations (e.g., a join where the result of the join is small but the intermediate results are big). Therefore, I decided against this approach. The hope is that the eager fetch on the `DataSourceExec` is likely instantly waiting for I/O and not doing much CPU-bound work. Furthermore, we are relying on statistics for adaptively executing the stream eagerly. I am guessing that the cardinality estimates on the DataSources are easier to get right or at least that underestimations are rare. In general: - understimates of the byte size can lead to performance degradation (we trigger an expensive scan eagerly, even though dynamic filters could have prevented it). - overestimates of the byte size leads to not eagerly computing the result of a small scan (status quo) - we assume that small scans require few sequential I/O ops (otherwise dynamic filters could again be worth the wait) Other possibilities: - Push the decision "eager" fetch into the `DataSource`. This would allow, for example, Parquet to optimize for the file format or have finer control about what should be done during the prefetching (e.g., I/O requests) and what should be delayed to the actual polling (e.g., decoding). - A general prefetching infrastructure for scans - Go the `BufferExec` on the probe-side path. (I don't think these are mutually exclusive) ## What changes are included in this PR? - Add `data_source_small_scan_partition_threshold` - Instantiate a `MemoryBufferedStream` in `DataSourceExec::execute` if the scan is "small" ## What is the testing strategy for this PR? - Existing tests - Demonstrate speed-ups in benchmarks Could be difficult to test this in isolation but I could make a unit tests that asserts the stream type. ## Are there any user-facing changes? - New configuration: `data_source_small_scan_partition_threshold`. Set to 0 for disabling. Current default value: 1 MiB fetches are done eagerly. - Less latency for some queries - Higher scanned bytes for some queries (where dynamic filters would have pruned the result) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
