alamb opened a new issue, #24265: URL: https://github.com/apache/datafusion/issues/24265
### 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. 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: <img width="1799" height="967" alt="Image" src="https://github.com/user-attachments/assets/e0c68699-2cde-48c6-b8d9-1bce2b22ecc2"; /> It turns out that unfortunately we are making it *worse* for streaming engines recently by making things better for batch execution - Buffering in RepartitioNExec https://github.com/apache/datafusion/issues/24044 - More Buffering in FilterExec 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) - See related PR https://github.com/apache/datafusion/pull/13823) - However, given that we are regressing behavior for streaming I think Here are some example systems using DataFusion for streaming - **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 try and add some way to configure which behavior is desired. ### 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` 2. A flag on operators like FilterExec and RepartitionExec that controls their emission behavior (aka should they flush at the earliest opportunity) ## Better propagation of defaults @ahirner / @calvinchengx I think are suggesting more automated propagation of unbounded sources > 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? ### Additional context We have discussed this in the past - #4285 - #11404 (discussion) - #10895 - #9016 -- 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]
