ahirner commented on issue #24044: URL: https://github.com/apache/datafusion/issues/24044#issuecomment-5230267858
> size-or-deadline policy I think there is a more sensible default without new `deadlines` if the user doesn't specify an explicit intent. Namely, bias unbounded execution towards low latency. What if to drain all `Poll::Ready` batches, i.e. until there is an await point? This might achieve low-latency and optimal performance more likely by default (batches are hot). I assume, pure compute nodes generally don't insert `Pending`. If so, one could go look for optimizations as long as they don't play directly against cooperative scheduling. > FilterExec looks like it has the same shape — it holds a LimitedBatchCoalescer and I did not see a boundedness check near it. https://github.com/apache/datafusion/blob/b225ded7338459a356d6afaba2756aecce7f98db/datafusion/physical-plan/src/filter.rs#L1256-L1316 I think `FilterExec` does have the same unfortunate behavior. To fix unbounded behavior it could more look like this: ```rust match self.input.poll_next_unpin(cx) { Poll::Pending if !self.batch_coalescer.is_empty() => { self.batch_coalescer.flush_buffered_batch()?; // drain but don't finish continue; } Poll::Pending => return Poll::Pending, Poll::Ready(None) => { self.batch_coalescer.finish()?; // ... } Poll::Ready(Some(batch)) => { // ... } } ``` Non-flushing of unbounded streams is not new though. It was introduced in #18604 and DataFusion 52. Before, it returned each non-empty filtered batch immediately. AI researched details: <summary> <details> <pre> When Change Effect ━━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Dec 30, 2020 Original CoalesceBatchesExec commit (https:// The optimizer wrapped FilterExec in a github.com/apache/datafusion/ coalescer. It flushed at target size or EOS, commit/13dbe8ad9b958ad20c07602dc9b8d5ef3eb37c7 but returned Pending without flushing. 1), released in 4.0.0 ────────────── ──────────────────────────────────────────────── ───────────────────────────────────────────────── Dec 28, 2022 PR #4694 (https://github.com/apache/ Added official infinite-input support. datafusion/pull/4694), released in 16.0.0 CoalesceBatchesExec declared itself compatible with unbounded input merely by propagating unbounded_output, without addressing residual flushing. ────────────── ──────────────────────────────────────────────── ───────────────────────────────────────────────── Nov 11, 2025 PR #18604 (https://github.com/apache/ Moved the coalescer inside FilterExec, making datafusion/pull/18604), released in 52.0.0 the behavior inherent to the operator. ────────────── ──────────────────────────────────────────────── ───────────────────────────────────────────────── Nov 14, 2025 PR #18630 (https://github.com/apache/ Removed the now-redundant optimizer wrapper and datafusion/pull/18630) added the current LimitedBatchCoalescer; the non-flushing behavior remained. </pre> </details> </summary> -- 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]
