apache / apache/datafusion

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

Open
#24,265 9 comments 8 reactions 0 assignees View on GitHub
enhancement PROPOSAL EPIC
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### 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](https://www.youtube.com/watch?v=0-BIHyzODH8&t=2s) from @sap1ens also mentioned this mismatch between streaming and batch oriented workflows for several built in operators:

Image

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` https://github.com/apache/datafusion/issues/24044
- Made Buffering required in FilterExec (e.g. #18604 moved coalescing inside `FilterExec` (DF 52), #18630 / #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](https://docs.rs/datafusion/latest/datafusion/physical_plan/execution_plan/enum.Boundedness.html) / [EmissionType](https://docs.rs/datafusion/latest/datafusion/physical_plan/execution_plan/enum.EmissionType.html)
- [ExecutionPlanProperties::boundedness()](https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlanProperties.html#tymethod.boundedness)
- [ExecutionPlanProperties::pipeline_behavior() (returns EmissionType)](https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlanProperties.html#tymethod.pipeline_behavior)

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](https://github.com/apache/datafusion/issues/4285)
- **[Streamling](https://github.com/goldsky-io/streamling)** (Goldsky, @sap1ens) — data streaming runtime on Rust/Arrow/DataFusion; powers Goldsky Turbo Pipelines. See [Introducing Streamling](https://www.streamingdata.tech/p/introducing-streamling) and [talk](https://www.youtube.com/watch?v=0-BIHyzODH8)
- **[StreamFusion](https://github.com/datafusion-contrib/StreamFusion)** (@jordepic) — Flink accelerator: swaps supported Flink SQL operators for native DataFusion execution over JNI
- **[Arroyo](https://www.arroyo.dev/blog/why-arrow-and-datafusion/)** — distributed stream processing engine; SQL engine built on Arrow + DataFusion since 0.10
- **[Denormalized](https://github.com/probably-nothing-labs/denormalized)** — embeddable "DuckDB for streaming"; Kafka, windowed aggregates, stream joins. See [The future of DataFusion is Streaming](https://www.denormalized.io/blog/streaming-datafusion)
- **[ArkFlow](https://github.com/arkflow-rs/arkflow)** — Rust stream processing engine with a DataFusion-based SQL processor
- **[LaminarDB](https://github.com/laminardb/laminardb)** — embedded streaming SQL database; integrates DataFusion in its SQL layer
- **[Kamu](https://github.com/kamu-data/kamu-cli)** — 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
- https://github.com/apache/datafusion/issues/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](https://datafusion.apache.org/user-guide/configs.html) ([source](https://github.com/apache/datafusion/blob/128ef37d6256c278418f1425da28624816a8e95b/datafusion/common/src/config.rs#L758)) -- since #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 https://github.com/apache/datafusion/issues/24044:

1. **Skip coalescing entirely when the input is unbounded** (proposed by @goutamadwant in [this comment](https://github.com/apache/datafusion/issues/24044#issuecomment-5224906471), the approach of
https://github.com/apache/datafusion/pull/24193 for `RepartitionExec`): simple and fixes the
correctness issue, but @calvinchengx ([comment](https://github.com/apache/datafusion/issues/24044#issuecomment-5225141918)) and @jayzhan211 ([comment](https://github.com/apache/datafusion/issues/24044#issuecomment-5228886190)) note it may not be applicable to all cases
2. **Drain-on-pending** (suggested by @ahirner in [this comment](https://github.com/apache/datafusion/issues/24044#issuecomment-5230267858)): 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](https://github.com/apache/datafusion/issues/24044#issuecomment-5229019257)) / @calvinchengx ([comment](https://github.com/apache/datafusion/issues/24044#issuecomment-5229095294))): flush at `batch_size` OR after a
configurable maximum batch age, whichever comes first.
4. **Richer emission triggers** (suggested by @jayzhan211 in [this comment](https://github.com/apache/datafusion/issues/24044#issuecomment-5158290091)): 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
- #4285
- #11404 (discussion)
- #10895
- See related PR https://github.com/apache/datafusion/pull/13823

### Related Issues
- [ ] https://github.com/apache/datafusion/issues/9016
- [x] https://github.com/apache/datafusion/issues/24044 (fix proposed in https://github.com/apache/datafusion/pull/24193)
- [ ] https://github.com/apache/datafusion/issues/19481
- [ ] https://github.com/apache/datafusion/issues/22090
- [ ] https://github.com/apache/datafusion/issues/13807
- [ ] https://github.com/apache/datafusion/issues/11365
- [ ] https://github.com/apache/datafusion/issues/11369
- [ ] https://github.com/apache/datafusion/issues/11364

Contributor guide

Open the contributing guide

Research direction

Start with the documented Boundedness and EmissionType APIs, ExecutionPlanProperties, and the coalesce_batches configuration in datafusion/common/src/config.rs. Review RepartitionExec and FilterExec alongside issues #24044 and #9016 to understand the existing streaming behavior and open design choices. Done requires an agreed strategy, supporting tests, and documentation or examples that define streaming behavior.

Written by the indexing model from the issue text.

Assessment

Tech stack
rust, sql
Domain
data-engineering, stream-processing
Issue type
Feature
Difficulty
5/5
Estimated time
Over a week
Activity status
Active
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.