fix: flush FilterExec output on pending inut for unbounded inputs

[goutamadwant]

Which issue does this PR close?

• Part of Document streaming usecase (like UNBOUNDED tables) #9016.

Rationale for this change

While prototyping the minimal push-based streaming example proposed in #9016, I found that FilterExec could buffer small filtered results indefinitely when its input was unbounded. Because an unbounded source may never finish, consumers could not observe those results while continuing to push input batches.

This PR extracts the smallest prerequisite fix so the streaming example and user-guide documentation can follow as separate, focused PRs.

This is complementary to #23856: that PR changes how filtered batches are supplied to the coalescer, while this PR controls when buffered rows become observable for an unbounded input.

What changes are included in this PR?

• Add a non-final LimitedBatchCoalescer flush operation that emits buffered rows while still allowing subsequent input.
• Preserve idempotent coalescer finalization.
• Flush FilterExec output after each input batch when its input is unbounded.
• Preserve the existing coalescing behavior for bounded inputs.
• Preserve existing fetch-limit behavior.
• Add regression coverage for:
  • emitting results while an unbounded input remains open;
  • multiple input pushes;
  • a fully filtered batch followed by a matching batch;
  • bounded-input coalescing;
  • flushing without preventing subsequent input.

Planned follow-up PRs:

1. Add the simplest possible push-based streaming example under datafusion-examples, showing how to feed batches into a running query and consume results incrementally.
2. Add a “Using DataFusion for streaming” section to the library user guide, based on that example and documenting relevant boundedness and execution considerations.

Are these changes tested?

Yes.

• Verified that the new unbounded-input regression test fails before the implementation and passes afterward.
• cargo test -p datafusion-physical-plan
• cargo test -p datafusion
• cargo test -p datafusion-cli
• cargo test --profile=ci --test sqllogictests
• Contributor-guide extended workspace test suite with avro,json,backtrace,extended_tests,recursive_protection,parquet_encryption
• cargo clippy --all-targets --all-features -- -D warnings
• cargo fmt --all -- --check
• Rust documentation, license-header, and typo checks

Are there any user-facing changes?

Yes. FilterExec now emits filtered results incrementally for unbounded inputs instead of waiting for the target batch size or input completion.

Bounded-input coalescing remains unchanged. There are no public API or breaking changes.

[2010YOUY01]

I found that FilterExec could buffer small filtered results indefinitely when its input was unbounded

Was that only to buffer up to batch_size from the configuration? I think this is still intended for streaming input. It's a convention for each operator to produce reasonably large output batches, which helps downstream operators vectorize.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 79.31034% with 48 lines in your changes missing coverage. Please review.
✅ Project coverage is 81.61%. Comparing base (35f58f5) to head (ba07df5).

Files with missing lines	Patch %	Lines
datafusion/physical-plan/src/filter.rs	77.22%	15 Missing and 31 partials ⚠️
datafusion/physical-plan/src/coalesce/mod.rs	93.33%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #24353      +/-   ##
==========================================
- Coverage   81.61%   81.61%   -0.01%     
==========================================
  Files        1124     1124              
  Lines      411978   412179     +201     
  Branches   411978   412179     +201     
==========================================
+ Hits       336236   336384     +148     
- Misses      55936    55957      +21     
- Partials    19806    19838      +32     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[jayzhan211]

It seems like belonged to the issue under #24265

[2010YOUY01]

I think this behavior is a design change, and now most operators don't follow.

I don't understand why this is the expected behavior, the existing behavior seems more reasonable, as micro-batching helps vectorize with small latency penalties. I suggest to have some further discussion before proceeding.

[alamb]

Thank you @goutamadwant -- i think this makes sense, but we should get alignment on what the intention of Unbounded input is and make sure @2010YOUY01 agrees / understands the rationale before we merge this

[alamb]

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)

datafusion/datafusion/physical-plan/src/execution_plan.rs

Lines 1289 to 1314 in caa3cb4

	/// Represents whether a stream of data **generated** by an operator is bounded (finite)
	/// or unbounded (infinite).
	///
	/// This is used to determine whether an execution plan will eventually complete
	/// processing all its data (bounded) or could potentially run forever (unbounded).
	///
	/// For unbounded streams, it also tracks whether the operator requires finite memory
	/// to process the stream or if memory usage could grow unbounded.
	///
	/// Boundedness of the output stream is based on the boundedness of the input stream and the nature of
	/// the operator. For example, limit or topk with fetch operator can convert an unbounded stream to a bounded stream.
	#[derive(Debug, Clone, Copy, PartialEq, Eq)]
	pub enum Boundedness {
	/// The data stream is bounded (finite) and will eventually complete
	Bounded,
	/// The data stream is unbounded (infinite) and could run forever
	Unbounded {
	/// Whether this operator requires infinite memory to process the unbounded stream.
	/// If false, the operator can process an infinite stream with bounded memory.
	/// If true, memory usage may grow unbounded while processing the stream.
	///
	/// For example, `Median` requires infinite memory to compute the median of an unbounded stream.
	/// `Min/Max` requires infinite memory if the stream is unordered, but can be computed with bounded memory if the stream is ordered.
	requires_infinite_memory: bool,
	},
	}

[alamb]

I recommend renaming this to something like input_unbounded to reflect what it represents rather than the behavior that it controls

[alamb]

maybe we could add a comment here explaining the rationale -- something like

// When the input is unbounded / streaming, flush any internal buffered 
// batches so the rest of the pipeline can produce output if possible.

[alamb]

I wonder if we can use a shared fixture somewhere for this -- it seems pretty general rather than specific to filter

[2010YOUY01]

Thank you @goutamadwant -- i think this makes sense, but we should get alignment on what the intention of Unbounded input is and make sure @2010YOUY01 agrees / understands the rationale before we merge this

Yes, I want to know more about application level motivation, and why the existing behavior won't satisfy the requirement, and next see what should we do.

Another concern is if we start to incorporate such no buffering behavior, extra implementation complexity will be added to all existing operators, for this FilterExec it's perfectly fine since it's simpler, but it would be very hard to do for other operators.

Is it possible to split it to a StreamingFilterStream, or even maintain that variant in downstream.

[sap1ens]

Is it possible to split it to a StreamingFilterStream, or even maintain that variant in downstream.

That's exactly what I've done, but @alamb's point is that more projects use DataFusion in the streaming context, so it would be nice to support this out of the box.

[jayzhan211]

Thank you @goutamadwant -- i think this makes sense, but we should get alignment on what the intention of Unbounded input is and make sure @2010YOUY01 agrees / understands the rationale before we merge this

Yes, I want to know more about application level motivation, and why the existing behavior won't satisfy the requirement, and next see what should we do.

Another concern is if we start to incorporate such no buffering behavior, extra implementation complexity will be added to all existing operators, for this FilterExec it's perfectly fine since it's simpler, but it would be very hard to do for other operators.

Is it possible to split it to a StreamingFilterStream, or even maintain that variant in downstream.

#24044 shows the issue that arises if we don't support "streaming".

If we agree on supporting more "streaming" execution in DataFusion, I think we should extend that support across all existing operators. +1 from me on the overall direction. How to design this while keeping maintenance complexity low could be a separate discussion — maybe in #24265

[2010YOUY01]

@jayzhan211 thanks for the pointer to that EPIC

My main blocking suggestion for this PR is to completely separate the bounded and unbounded execution paths, rather than adding the unbounded behavior into the existing implementation.

If we can do that, I think this PR should be good to go. I explained the reasoning in more detail in the comment in #24265 (comment)

[alamb]

#24044 shows the issue that arises if we don't support "streaming".

If we agree on supporting more "streaming" execution in DataFusion, I think we should extend that support across all existing operators. +1 from me on the overall direction. How to design this while keeping maintenance complexity low could be a separate discussion — maybe in #24265

Yeah -- in my mind DataFusion is in a half way state now -- it has some features / support for streaming (e.g. Boundedness), but as shown in #24044 since there is no clear design / tests / documentation implicit assumptions that streaming systems may be counting on can (and do) get broken

So I think we should first agree on if we want to try and make DataFusion more useful for building streaming engines -- I think it is important and there are a bunch of people already doing so (and have done so for a while -- e.g. Arroyo and earlier versions of Synnada). Rewriting Apache Flink in rust seems to be all the rage now too (e.g. @jordepic 's StreamFusion, etc) , and many people are using DataFusion to try.

@jayzhan211 thanks for the pointer to that EPIC

My main blocking suggestion for this PR is to completely separate the bounded and unbounded execution paths, rather than adding the unbounded behavior into the existing implementation.

If we can do that, I think this PR should be good to go. I explained the reasoning in more detail in the comment in #24265 (comment)

[alamb]

I propose putting this stuff into a new crate (datafusion-streaming)

• [EPIC] Better support / documentation / tests for Streaming SQL Engines #24265 (comment)

[jayzhan211]

ba07df5 overall LGTM

[2010YOUY01]

Was that ready for review? Seems still WIP from the latest change 🤔