Problems understanding the mechanics of this extension

[dberardo]

Hi there,
Looking at the examples online, it seems that this extension works in a sort of mini batches way, meaning that it first waits for the pipes to finish loading and just then emits the partial computed result.

Is it possible to use this extension as an actual streaming engine, meaning that results are emitted after every new incoming row, or on multiple rows when using a window? To do so one need an incremental implementation of all primitive expressions such as distinct, count, etc.

Is this within the scope of this library and if so, where to find the list of supported incremental operators? And also, how are stream/stream and table/stream joins handled?

[turbolytics]

Hi! Thanks for your detailed question and interest in how the extension works under the hood. I'll provide a technical explanation of how windowing, streaming, and incremental operators are currently handled, along with links and example configurations.

Windowing & Emission Model

The extension (currently) processes records in a mini-batch style. Data is grouped into windows (such as hourly, tumbling, etc.), and each window is polled on a regular interval. When a window is considered closed—based on a time watermark or a SQL predicate—its results are emitted (to Kafka, console, etc.) and then cleaned up. This avoids unbounded growth and helps with stateful aggregation.

Example (tumbling window):

tables:
  sql: 
    - name: agg_cities_count
      manager:
        tumbling_window: 
          collect_closed_windows_sql: |
            SELECT ... WHERE bucket < (now() - INTERVAL '60' SECOND)
          delete_closed_windows_sql: |
            DELETE ... WHERE bucket < (now() - INTERVAL '60' SECOND)

Each window "bucket" (e.g., per hour) aggregates new rows using standard SQL and writes to a table. The manager periodically checks for closed buckets and outputs the aggregated result. Example aggregation:

INSERT INTO agg_cities_count
BY NAME
SELECT date_trunc('hour', timestamp) as bucket, city, count(*) as count
FROM batch
GROUP BY bucket, city
ON CONFLICT (bucket, city) DO UPDATE SET count = count + EXCLUDED.count;

Streaming / Per-row Incremental Processing

• The system does not emit results for every new row as a true streaming engine would, but rather on window closure or batch interval.
• Incremental operators (count, distinct, etc.) are supported via the SQL engine (DuckDB) using ON CONFLICT upserts and in-place aggregation.
• There isn't a dedicated API surface for per-row stateful incremental expressions—it's more oriented around SQL-style window aggregations.
• Truly incremental/continuous emission (per row) would require further architectural changes, particularly to state tracking and emission semantics, as well as support for each incremental operator.

Supported Incremental Operators

Currently, you can use standard SQL aggregate functions (count, sum, avg, min, max) in the context of a window/batch, with upserts for aggregating. Distinct/count-distinct and similar primitives are handled if they're supported by your SQL dialect and streaming logic.

There is not (as of now) a maintained list of "streamable" primitive operators with full incremental state. Most expressions supported by DuckDB SQL within windowed aggregation are available.

Stream/Stream and Table/Stream Joins

• Stream/stream or table/stream joins are expressed as SQL queries or windowed SQL joins.
• For true continuous/row-level joins, specialized window join logic would be needed. As of today, join support is oriented around what DuckDB can do inside a batch/window.

---

References:

• Window lifecycle inspiration (Flink)
• See also the tumbling window ADR in the repo

If you want true per-row streaming or a list of primitive incremental operators, let us know more about your use case! We are happy to discuss possible architectural extensions.

[turbolytics]

A working tumbling window is available for experimentation here:

https://github.com/turbolytics/sql-flow/blob/main/dev/config/examples/tumbling.window.yml