alamb commented on issue #9016:
URL: https://github.com/apache/datafusion/issues/9016#issuecomment-5252501106
Some background on the current state of how a data source declares itself
unbounded / streaming, which such documentation could cover.
There are three layers, from lowest to highest:
**1. The fundamental mechanism — `PlanProperties` on any `ExecutionPlan`**
A source (or any operator) declares itself unbounded by returning
`Boundedness::Unbounded { requires_infinite_memory: false }` in the
[`PlanProperties`](https://docs.rs/datafusion/latest/datafusion/physical_plan/struct.PlanProperties.html)
it builds in `properties()` (the constructor takes `boundedness`; there is
also `with_boundedness()`):
-
[`Boundedness`](https://docs.rs/datafusion/latest/datafusion/physical_plan/execution_plan/enum.Boundedness.html)
and
[`EmissionType`](https://docs.rs/datafusion/latest/datafusion/physical_plan/execution_plan/enum.EmissionType.html),
added in #13823
- Accessors:
[`ExecutionPlanProperties::boundedness()`](https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlanProperties.html#tymethod.boundedness)
and
[`pipeline_behavior()`](https://docs.rs/datafusion/latest/datafusion/physical_plan/trait.ExecutionPlanProperties.html#tymethod.pipeline_behavior)
**2. The ready-made physical operator — `StreamingTableExec`**
The intended way to expose a custom stream: implement `PartitionStream` and
pass `infinite: true` to
[`StreamingTableExec::try_new`](https://docs.rs/datafusion/latest/datafusion/physical_plan/streaming/struct.StreamingTableExec.html).
That flag is the *only* thing that becomes `Unbounded` — see
[`compute_properties`](https://github.com/apache/datafusion/blob/128ef37d6256c278418f1425da28624816a8e95b/datafusion/physical-plan/src/streaming.rs#L171-L187)
(emission is hard-coded `Incremental`).
For generated data there is also
[`LazyBatchGenerator::boundedness()`](https://docs.rs/datafusion/latest/datafusion/physical_plan/memory/trait.LazyBatchGenerator.html)
(default `Bounded`, override to declare unbounded).
**3. The `TableProvider` / SQL layer**
-
[`StreamingTable`](https://docs.rs/datafusion/latest/datafusion/catalog/streaming/struct.StreamingTable.html)
with `.with_infinite_table(true)` (default is `false`)
- In SQL: `CREATE UNBOUNDED EXTERNAL TABLE` routes to `StreamTableFactory`,
whose [`scan()` hardcodes `infinite =
true`](https://github.com/apache/datafusion/blob/128ef37d6256c278418f1425da28624816a8e95b/datafusion/catalog/src/stream.rs#L344-L351)
— user guide: https://datafusion.apache.org/user-guide/sql/ddl.html
Two gaps worth noting:
- `DataSourceExec` (i.e., every `FileScanConfig`-based file source) *cannot*
declare unboundedness — it [hardcodes
`Boundedness::Bounded`](https://github.com/apache/datafusion/blob/128ef37d6256c278418f1425da28624816a8e95b/datafusion/datasource/src/source.rs#L597-L605)
- [`JsonReadOptions::infinite` /
`mark_infinite()`](https://github.com/apache/datafusion/blob/128ef37d6256c278418f1425da28624816a8e95b/datafusion/core/src/datasource/file_format/options.rs#L475)
appears to be dead code with no consumer
There is currently no narrative "how to build a streaming source" doc — the
closest existing docs are the `StreamingTableExec` rustdoc and the DDL page.
--
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]
