alamb commented on code in PR #24353:
URL: https://github.com/apache/datafusion/pull/24353#discussion_r3824766717
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -606,6 +606,7 @@ impl ExecutionPlan for FilterExec {
context.task_id()
);
let metrics = FilterExecMetrics::new(&self.metrics, partition);
+ let flush_when_input_pending = self.input.boundedness().is_unbounded();
Review Comment:
I think this is the key change / question --
I think the assumption is that if the input is "unbounded" that signals we
are in " streaming mode" and thus the users are prioritizing latency over raw
throughput.
However, looking at the Boundedness documentation that is never explicitly
stated -- perhaps we should update it (maybe @jayzhan211 who introduced it in
#13823 could confirm the intent)
https://github.com/apache/datafusion/blob/caa3cb4af1bbf49403261e27714414bf7d41d71e/datafusion/physical-plan/src/execution_plan.rs#L1289-L1314
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -1203,6 +1205,8 @@ struct FilterExecStream {
projection: Option<ProjectionRef>,
/// Batch coalescer to combine small batches
batch_coalescer: LimitedBatchCoalescer,
+ /// Emit buffered rows when an unbounded input has no batch ready.
+ flush_when_input_pending: bool,
Review Comment:
I recommend renaming this to something like `input_unbounded` to reflect
what it represents rather than the behavior that it controls
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -1280,15 +1284,22 @@ impl Stream for FilterExecStream {
}
// Attempt to pull the next batch from the input stream.
- match ready!(self.input.poll_next_unpin(cx)) {
- None => {
+ match self.input.poll_next_unpin(cx) {
+ Poll::Pending => {
+ if self.flush_when_input_pending &&
!self.batch_coalescer.is_empty() {
Review Comment:
maybe we could add a comment here explaining the rationale -- something like
```rust
// When the input is unbounded / streaming, flush any internal buffered
// batches so the rest of the pipeline can produce output if possible.
```
##########
datafusion/physical-plan/src/filter.rs:
##########
@@ -1413,9 +1422,163 @@ mod tests {
use crate::empty::EmptyExec;
use crate::expressions::*;
use crate::statistics::{StatisticsArgs, StatisticsContext};
+ use crate::stream::RecordBatchStreamAdapter;
+ use crate::streaming::{PartitionStream, StreamingTableExec};
use crate::test;
use crate::test::exec::StatisticsExec;
+ use arrow::array::Int32Array;
use arrow::datatypes::{Field, Schema, UnionFields, UnionMode};
+ use futures::stream::poll_fn;
+ use std::fmt::{Debug, Formatter};
+ use std::sync::Mutex;
+ use std::time::Duration;
+ use tokio::sync::mpsc::{Receiver, Sender, channel};
+
+ struct ChannelPartition {
Review Comment:
I wonder if we can use a shared fixture somewhere for this -- it seems
pretty general rather than specific to filter
--
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]
