tschwarzinger commented on code in PR #24921:
URL: https://github.com/apache/datafusion/pull/24921#discussion_r3925034901
##########
datafusion/datasource/src/source.rs:
##########
@@ -694,6 +698,52 @@ impl DataSourceExec {
.map(|source| (file_scan_conf, source))
})
}
+
+ /// Tries to evaluate the stream in a separate task, depending on whether
the statistics show
+ /// that this is a small scan.
+ ///
+ /// For some scenarios, this can significantly improve the latency, as
object store requests
+ /// are triggered earlier. For example, while a join is building its build
side, the eager poll
+ /// can already trigger network requests, cutting down on the latency of
the overall join.
+ ///
+ /// Note that this can pessimize queries if the dynamic filters have not
yet been computed by
+ /// other operators and can therefore not be used for pruning. To
alleviate this problem,
+ /// the eager fetches are only executed if the scan is "small" and thus
the savings from dynamic
+ /// filter pushdown are negligible. Nevertheless, inaccurate or wrong
statistics could pessimize
+ /// queries.
+ fn try_buffer_stream(
+ &self,
+ partition: usize,
+ context: &Arc<TaskContext>,
+ stream: SendableRecordBatchStream,
+ ) -> Result<SendableRecordBatchStream> {
+ let small_scan_threshold = context
+ .session_config()
+ .options()
+ .execution
+ .data_source_small_scan_partition_threshold;
+ if small_scan_threshold == 0 {
+ return Ok(stream);
+ }
+
+ let partition_stats =
self.data_source.partition_statistics(Some(partition))?;
+ let Some(actual_size) = partition_stats.total_byte_size.get_value()
else {
+ return Ok(stream);
+ };
+
+ let stream = if *actual_size <= small_scan_threshold {
+ let mem_reservation =
+
MemoryConsumer::new(format!("DataSourceExecEagerExecution[{partition}]"))
+ .register(context.memory_pool());
+ Box::pin(RecordBatchStreamAdapter::new(
+ stream.schema(),
+ MemoryBufferedStream::new(stream, *actual_size,
mem_reservation),
Review Comment:
Note: `actual_size` will likely underestimate the size of the record batch
after decoding. Maybe `small_scan_threshold` would be a more sensible size?
--
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]
