[EPIC] Better support / documentation / tests for Streaming SQL Engines

[alamb]

Is your feature request related to a problem or challenge?

DataFusion has traditionally been developed using techniques from Batch processing systems and tries to maximize throughput by being as efficient as possible per tuple. This often involves techniques such as batching rows together to amortize the overhead of each row

However, it also seems increasingly common to try and use DataFusion for streaming systems, similar to Apache Flink, which typically have "unbounded queries" that run indefinitely and place a high value on values being produced quickly (to minimize latency) rather than simply raw per-row computation.

I think we have a bunch of features for streaming documented / tested in isolation already (see below) but the overall design and goal is not clearly documented or tested and thus the overall experience is regressing

The recent presentation on youtube from @sap1ens also mentioned this mismatch between streaming and batch oriented workflows for several built in operators:

It turns out that unfortunately we are making it worse for streaming engines recently by making things better for batch execution (typically by moving buffering into the execution nodes)

• Buffering in RepartitionExec RepartitionExec withholds all output from an unbounded input until batch_size rows accumulate #24044
• Made Buffering required in FilterExec (e.g. Coalesce batches inside FilterExec #18604 moved coalescing inside FilterExec (DF 52), Remove FilterExec from CoalesceBatches optimization rule, add fetch support #18630 / Remove coalesce batches rule and deprecate CoalesceBatchesExec #19622 removed CoalesceBatches entirely so buffering is now inside the operator

There is some basic infrastructure today, but it isn't used / respected everywhere and there isn't an over arching strategy / description of its use

• Boundedness / EmissionType
• ExecutionPlanProperties::boundedness()
• ExecutionPlanProperties::pipeline_behavior() (returns EmissionType)

However, given that we are regressing behavior for streaming I think we should consider formalizing this functionality more

Example Streaming Systems

• Synnada — streaming-first data products; authors of the original streaming roadmap #4285
• Streamling (Goldsky, @sap1ens) — data streaming runtime on Rust/Arrow/DataFusion; powers Goldsky Turbo Pipelines. See Introducing Streamling and talk
• StreamFusion (@jordepic) — Flink accelerator: swaps supported Flink SQL operators for native DataFusion execution over JNI
• Arroyo — distributed stream processing engine; SQL engine built on Arrow + DataFusion since 0.10
• Denormalized — embeddable "DuckDB for streaming"; Kafka, windowed aggregates, stream joins. See The future of DataFusion is Streaming
• ArkFlow — Rust stream processing engine with a DataFusion-based SQL processor
• LaminarDB — embedded streaming SQL database; integrates DataFusion in its SQL layer
• Kamu — planet-scale streaming data pipeline

Describe the solution you'd like

Given how many systems seem to want to use DataFusion for streaming systems (rather than batch oriented ones) and the natural tension between batching (maximize throughput) vs streaming (minimize latency) I think if we should formalize / test / document the best way to configure DataFusion for streaming.

Describe alternatives you've considered

Better documentation / tests

Maybe part of this epic could be just to document what we have better

• Document streaming usecase (like UNBOUNDED tables) #9016
• Maybe some more tests for filtering/repartition, etc and ensuring it is treaming / low latency
• Maybe a blog / datafusion-example about how to configure DataFusion for streaming mode

Config Flag

Maybe it would be worth some sort of "bounded configuration mode" setting to make implementing streaming systems easier

For example,

1. A global config setting like streaming or optimize_for_latency (maybe repurpose datafusion.execution.coalesce_batches (source) -- since Remove coalesce batches rule and deprecate CoalesceBatchesExec #19622 I don't think anything reads it
2. A flag on operators like FilterExec and RepartitionExec that controls their emission behavior (aka should they flush at the earliest opportunity)

Better propagation / policies

Several potential policies for when operators should flush partially-full batches came up in the context of #24044:

1. Skip coalescing entirely when the input is unbounded (proposed by @goutamadwant in this comment, the approach of
   fix: avoid buffering unbounded repartition output indefinitely #24193 for RepartitionExec): simple and fixes the
   correctness issue, but @calvinchengx (comment) and @jayzhan211 (comment) note it may not be applicable to all cases
2. Drain-on-pending (suggested by @ahirner in this comment): keep coalescing while input is Poll::Ready, but flush the
   residual whenever the input returns Pending instead of holding it.
3. Size-or-deadline (suggested by @goutamadwant (comment) / @calvinchengx (comment)): flush at batch_size OR after a
   configurable maximum batch age, whichever comes first.
4. Richer emission triggers (suggested by @jayzhan211 in this comment): time-based, idle-based (N empty polls), or
   watermark / checkpoint-barrier-based, if DataFusion later grows control-message propagation

Make more operators respect buffering

Additional context

We have discussed this in the past

• [Proposal] Streaming execution support roadmap #4285
• [DISCUSSION] Support for Streaming in DataFusion #11404 (discussion)
• Real-time streaming support #10895
• See related PR Replace execution_mode with emission_type and boundedness #13823

Related Issues

• Document streaming usecase (like UNBOUNDED tables) #9016
• RepartitionExec withholds all output from an unbounded input until batch_size rows accumulate #24044 (fix proposed in fix: avoid buffering unbounded repartition output indefinitely #24193)
• Operators should respect batch_size instead of Emit::All #19481
• RepartitionExec channels grow unboundedly with one slow consumer #22090
• Convert sort to partial_sort when the table is unbounded #13807
• StateBackend in DataFusion's RuntimeEnv #11365
• Make Accumulators and ScalarValue serializable #11369
• Deterministic IDs for ExecutionPlan #11364

[sap1ens]

Thanks for creating this issue, @alamb!

I think getting something like "Size-or-deadline" flush trigger into as many operators as possible (with values configured globally), can be the first step into better streaming semantics. You can avoid a global "streaming" flag by simply choosing how to flush batches, for example:

• Batch system: size: 8k, deadline: disabled.
• Streaming system: size: 1k, deadline: 100ms.

Control-message propagation also would be a really cool building block: you could build high-level functionality like checkpointing, windowing, etc.