Streaming SQL
Source of truth: https://docs.timeplus.com | Raw markdown: https://github.com/timeplus-io/docs/tree/main/docs
Naming rules
| Element |
Style |
Example |
| Keywords |
UPPERCASE |
SELECT, CREATE STREAM, EMIT, JOIN |
| Functions |
lowercase |
count(), tumble(), date_diff_within() |
| Data types |
lowercase |
int64, float64, string, datetime64 |
| Identifiers |
lowercase_with_underscores |
event_time, user_id |
| Reserved fields |
_tp_ prefix |
_tp_time (event timestamp), _tp_delta (changelog) |
Stream type decision table
| Need |
Type |
Syntax |
| Immutable events, time-series, high throughput |
append (default) |
CREATE STREAM ... ORDER BY |
| Updates/upserts, point queries, KV (first choice) |
mutable |
CREATE MUTABLE STREAM ... PRIMARY KEY |
| Version history, ASOF JOINs |
versioned_kv |
CREATE STREAM ... PRIMARY KEY ... SETTINGS mode='versioned_kv' |
CDC semantics, track deletes via _tp_delta |
changelog_kv |
CREATE STREAM ... PRIMARY KEY ... SETTINGS mode='changelog_kv' |
| External source (Kafka, Pulsar, etc.) |
external |
CREATE EXTERNAL STREAM ... SETTINGS type='kafka' |
Full details → references/stream-types.md
Query modes
SELECT FROM stream → streaming (continuous, future events)SELECT FROM table(stream) → historical (batch scan, returns once)
Three trigger types:
| Query type |
Trigger |
| Non-aggregation (tail/filter/transform) |
When events arrive |
| Window aggregation |
Window end + watermark |
| Global aggregation |
Fixed interval (default 2s if EMIT PERIODIC omitted) |
Window functions quick reference
| Function |
Signature |
Use case |
tumble |
tumble(stream, [time_col], interval, [tz]) |
Fixed non-overlapping windows |
hop |
hop(stream, [time_col], slide, size, [tz]) |
Sliding/overlapping windows |
session |
session(stream, [time_col], MAXSPAN x AND TIMEOUT y) |
Inactivity-based windows |
time_col defaults to _tp_time if omitted- Intervals:
1s, 5m, 2h, 3d, 1w, 1M, 1q, 1y
window_start, window_end auto-generated (left-closed, right-open [))- Hop: slide and size must use same unit; slide > size is unsupported
- Window nesting: max 2 levels; window-over-global is unsupported
EMIT policy quick reference
| Context |
Policy |
Effect |
| Window |
EMIT AFTER WINDOW CLOSE |
Default for windowed agg |
| Window |
EMIT AFTER WINDOW CLOSE WITH DELAY 2s |
Allow late events |
| Window |
EMIT AFTER WINDOW CLOSE WITH DELAY 1s AND TIMEOUT 3s |
Late events + force-close |
| Window |
EMIT ON UPDATE |
Emit when agg value changes per key |
| Window |
EMIT ON UPDATE WITH BATCH 2s |
Batched update detection |
| Global |
EMIT PERIODIC 5s |
Default (2s), batch periodic output |
| Global |
EMIT PERIODIC 5s REPEAT |
Emit even without new events |
| Global |
EMIT ON UPDATE |
Immediate on every change |
| Global |
EMIT CHANGELOG |
With _tp_delta (+1/-1) |
| Global |
EMIT PER EVENT |
Per-event (debug only, no parallelism) |
| Global |
EMIT AFTER KEY EXPIRE ... WITH MAXSPAN x AND TIMEOUT y |
Tracing/span aggregation |
Full formal syntax → references/emit-policies.md
JOIN quick reference
| Pattern |
Syntax key |
Use case |
| Static enrichment |
stream JOIN table(lookup) |
Enrich with historical data |
| Dynamic enrichment |
append JOIN versioned_kv USING(k) |
Latest version auto-picked |
| Bidirectional |
mutable JOIN mutable |
Both sides updatable |
| Range (time-bounded) |
stream JOIN stream ... AND date_diff_within(2m) |
Bounded stream-to-stream |
| ASOF |
append ASOF JOIN versioned_kv ON ... AND t1 >= t2 |
Closest version match |
| LATEST |
append LATEST JOIN versioned_kv ON ... |
Latest value only |
| Direct lookup |
stream JOIN mutable ... SETTINGS join_algorithm='direct' |
PK/index lookup, no full load |
| Dictionary |
stream JOIN dict ... SETTINGS join_algorithm='direct' |
External source lookup |
Supported: INNER, LEFT, FULL. Unsupported: RIGHT, CROSS.
Strictness: ALL (default), ASOF, LATEST.
Full examples → references/join-patterns.md
Materialized view checklist
Full config → references/mv-production.md
External stream (Kafka) quick reference
CREATE EXTERNAL STREAM events(raw string)
SETTINGS type='kafka', brokers='host:9092', topic='events';
Key settings: data_format, security_protocol, sasl_mechanism, kafka_schema_registry_url
Virtual columns: _tp_message_key, _tp_message_headers, _tp_sn (offset), _tp_shard (partition)
Query options: SETTINGS shards='0,2', seek_to='earliest'
Full details → references/external-streams.md
UDF quick reference
CREATE FUNCTION udf_name(param type) RETURNS type LANGUAGE JAVASCRIPT AS $$ ... $$;
Scalar: receives array of values (batched), returns array.
UDAF: implement initialize, process, finalize, serialize, deserialize, merge.
Full details → references/udf.md
References
- Stream type details and examples
- All EMIT policies with formal syntax
- JOIN patterns with full examples
- Production materialized view configuration
- External streams (Kafka, Pulsar)
- User-defined functions (UDF/UDAF)
1---2name: sql-usage3description: Timeplus streaming SQL covering stream types, EMIT policies, window functions, JOINs, materialized views, external streams, and UDFs. Make sure to use this skill for any SQL-related question including writing queries, debugging SQL errors, understanding streaming behavior, or designing stream processing pipelines, even if the user doesn't explicitly mention streaming SQL.4---56# Streaming SQL78Source of truth: https://docs.timeplus.com | Raw markdown: https://github.com/timeplus-io/docs/tree/main/docs910## Naming rules1112| Element | Style | Example |13|---------|-------|---------|14| Keywords | `UPPERCASE` | `SELECT`, `CREATE STREAM`, `EMIT`, `JOIN` |15| Functions | `lowercase` | `count()`, `tumble()`, `date_diff_within()` |16| Data types | `lowercase` | `int64`, `float64`, `string`, `datetime64` |17| Identifiers | `lowercase_with_underscores` | `event_time`, `user_id` |18| Reserved fields | `_tp_` prefix | `_tp_time` (event timestamp), `_tp_delta` (changelog) |1920## Stream type decision table2122| Need | Type | Syntax |23|------|------|--------|24| Immutable events, time-series, high throughput | append (default) | `CREATE STREAM ... ORDER BY` |25| Updates/upserts, point queries, KV (first choice) | mutable | `CREATE MUTABLE STREAM ... PRIMARY KEY` |26| Version history, ASOF JOINs | versioned_kv | `CREATE STREAM ... PRIMARY KEY ... SETTINGS mode='versioned_kv'` |27| CDC semantics, track deletes via `_tp_delta` | changelog_kv | `CREATE STREAM ... PRIMARY KEY ... SETTINGS mode='changelog_kv'` |28| External source (Kafka, Pulsar, etc.) | external | `CREATE EXTERNAL STREAM ... SETTINGS type='kafka'` |2930Full details → [references/stream-types.md](references/stream-types.md)3132## Query modes3334- `SELECT FROM stream` → streaming (continuous, future events)35- `SELECT FROM table(stream)` → historical (batch scan, returns once)3637Three trigger types:38| Query type | Trigger |39|-----------|---------|40| Non-aggregation (tail/filter/transform) | When events arrive |41| Window aggregation | Window end + watermark |42| Global aggregation | Fixed interval (default 2s if `EMIT PERIODIC` omitted) |4344## Window functions quick reference4546| Function | Signature | Use case |47|----------|-----------|----------|48| `tumble` | `tumble(stream, [time_col], interval, [tz])` | Fixed non-overlapping windows |49| `hop` | `hop(stream, [time_col], slide, size, [tz])` | Sliding/overlapping windows |50| `session` | `session(stream, [time_col], MAXSPAN x AND TIMEOUT y)` | Inactivity-based windows |5152- `time_col` defaults to `_tp_time` if omitted53- Intervals: `1s`, `5m`, `2h`, `3d`, `1w`, `1M`, `1q`, `1y`54- `window_start`, `window_end` auto-generated (left-closed, right-open `[)`)55- Hop: slide and size must use same unit; slide > size is unsupported56- Window nesting: max 2 levels; window-over-global is unsupported5758## EMIT policy quick reference5960| Context | Policy | Effect |61|---------|--------|--------|62| Window | `EMIT AFTER WINDOW CLOSE` | Default for windowed agg |63| Window | `EMIT AFTER WINDOW CLOSE WITH DELAY 2s` | Allow late events |64| Window | `EMIT AFTER WINDOW CLOSE WITH DELAY 1s AND TIMEOUT 3s` | Late events + force-close |65| Window | `EMIT ON UPDATE` | Emit when agg value changes per key |66| Window | `EMIT ON UPDATE WITH BATCH 2s` | Batched update detection |67| Global | `EMIT PERIODIC 5s` | Default (2s), batch periodic output |68| Global | `EMIT PERIODIC 5s REPEAT` | Emit even without new events |69| Global | `EMIT ON UPDATE` | Immediate on every change |70| Global | `EMIT CHANGELOG` | With `_tp_delta` (+1/-1) |71| Global | `EMIT PER EVENT` | Per-event (debug only, no parallelism) |72| Global | `EMIT AFTER KEY EXPIRE ... WITH MAXSPAN x AND TIMEOUT y` | Tracing/span aggregation |7374Full formal syntax → [references/emit-policies.md](references/emit-policies.md)7576## JOIN quick reference7778| Pattern | Syntax key | Use case |79|---------|-----------|----------|80| Static enrichment | `stream JOIN table(lookup)` | Enrich with historical data |81| Dynamic enrichment | `append JOIN versioned_kv USING(k)` | Latest version auto-picked |82| Bidirectional | `mutable JOIN mutable` | Both sides updatable |83| Range (time-bounded) | `stream JOIN stream ... AND date_diff_within(2m)` | Bounded stream-to-stream |84| ASOF | `append ASOF JOIN versioned_kv ON ... AND t1 >= t2` | Closest version match |85| LATEST | `append LATEST JOIN versioned_kv ON ...` | Latest value only |86| Direct lookup | `stream JOIN mutable ... SETTINGS join_algorithm='direct'` | PK/index lookup, no full load |87| Dictionary | `stream JOIN dict ... SETTINGS join_algorithm='direct'` | External source lookup |8889Supported: INNER, LEFT, FULL. Unsupported: RIGHT, CROSS.90Strictness: ALL (default), ASOF, LATEST.9192Full examples → [references/join-patterns.md](references/join-patterns.md)9394## Materialized view checklist9596- [ ] Stateless test default: use MatView without `INTO`, then verify with `table(mv)`97- [ ] Create target stream FIRST only when you need an extra sink stream98- [ ] Use explicit `INTO target` only when a target stream is required by the scenario99- [ ] Configure checkpointing: `SETTINGS checkpoint_interval=30`100- [ ] High-cardinality: `SETTINGS default_hash_table='hybrid', max_hot_keys=10000`101- [ ] Schema evolution (with target stream): `ALTER STREAM` target + `ALTER VIEW ... MODIFY QUERY`102- [ ] Cleanup order: `DROP VIEW` → (if created) `DROP STREAM` target → `DROP STREAM` source103104Full config → [references/mv-production.md](references/mv-production.md)105106## External stream (Kafka) quick reference107108```sql109CREATE EXTERNAL STREAM events(raw string)110SETTINGS type='kafka', brokers='host:9092', topic='events';111```112113Key settings: `data_format`, `security_protocol`, `sasl_mechanism`, `kafka_schema_registry_url`114Virtual columns: `_tp_message_key`, `_tp_message_headers`, `_tp_sn` (offset), `_tp_shard` (partition)115Query options: `SETTINGS shards='0,2'`, `seek_to='earliest'`116117Full details → [references/external-streams.md](references/external-streams.md)118119## UDF quick reference120121```sql122CREATE FUNCTION udf_name(param type) RETURNS type LANGUAGE JAVASCRIPT AS $$ ... $$;123```124125Scalar: receives array of values (batched), returns array.126UDAF: implement `initialize`, `process`, `finalize`, `serialize`, `deserialize`, `merge`.127128Full details → [references/udf.md](references/udf.md)129130## References131132- [Stream type details and examples](references/stream-types.md)133- [All EMIT policies with formal syntax](references/emit-policies.md)134- [JOIN patterns with full examples](references/join-patterns.md)135- [Production materialized view configuration](references/mv-production.md)136- [External streams (Kafka, Pulsar)](references/external-streams.md)137- [User-defined functions (UDF/UDAF)](references/udf.md)
timeplus-io