Feldera

Feldera is an incremental compute engine: you define tables and views in SQL, and Feldera keeps the views up to date as records arrive instead of recomputing them. S2 ships as a native transport in Feldera, so a table can read directly from an S2 stream and a view can append its updates to one.

s2_input feeds a table from an S2 stream, passing each record body to the configured Feldera input format (e.g. JSON). It checkpoints S2 sequence numbers, so a resumed pipeline picks up where it left off, and it reconnects with bounded backoff on transient errors. start_from selects where a fresh pipeline begins: "Beginning", "Tail", a SeqNum, a Timestamp, or a TailOffset.
s2_output appends a view’s changes to an S2 stream as one record per encoded row, batched into native S2 appends. It is not a fault-tolerant output: a failure between S2 accepting a batch and Feldera observing the acknowledgement can produce duplicates on retry.

Both connectors take a basin, stream, and auth_token, plus an optional endpoint. Full configuration tables are in the Feldera docs:

Feldera marks S2 support as experimental.

Example

A table reads JSON records from the orders stream, and a view appends every order of $1,000 or more to the large-orders stream as Feldera insert/delete envelopes: Replace commerce with your own basin name, and create both streams first (s2 create-stream s2://{basin}/orders and s2 create-stream s2://{basin}/large-orders) unless the basin has stream auto-creation enabled.

CREATE TABLE orders (
    id BIGINT,
    amount DECIMAL(12, 2)
) WITH (
    'connectors' = '[{
        "transport": {
            "name": "s2_input",
            "config": {
                "basin": "commerce",
                "stream": "orders",
                "auth_token": "${secret:kubernetes:s2/auth-token}",
                "start_from": "Beginning"
            }
        },
        "format": {
            "name": "json",
            "config": { "update_format": "raw" }
        }
    }]'
);

CREATE VIEW large_orders WITH (
    'connectors' = '[{
        "transport": {
            "name": "s2_output",
            "config": {
                "basin": "commerce",
                "stream": "large-orders",
                "auth_token": "${secret:kubernetes:s2/auth-token}"
            }
        },
        "format": {
            "name": "json",
            "config": { "update_format": "insert_delete" }
        }
    }]'
) AS
SELECT * FROM orders WHERE amount >= 1000;

auth_token is an S2 access token; the example passes it as a Feldera secret reference rather than inline. With the pipeline running, appending an order to orders:

echo '{"id":5,"amount":1500}' | s2 append s2://commerce/orders

shows up moments later on large-orders as an insert envelope:

s2 read s2://commerce/large-orders --format json

{"seq_num":2,"timestamp":1792778129264,"body":"{\"insert\":{\"id\":5,\"amount\":1500}}"}

Set endpoint to run against s2-lite locally, for example "endpoint": "http://localhost:7070". s2-lite ignores the access token, so any non-empty auth_token works.

Protocols

AI SDKs

Pipelines

Sync

Example

​Example

Example