Skip to content

Commit f3cc119

Browse files
jovezhongjovezhong
authored
New Emit Polices (#300)
* WIP update for the EMIT policies * Move TIMEOUT to common emit, update SQL in react * Update query-syntax.md fix heading * Update query-syntax.md * Update query-syntax.md * Update v2-release-notes.md * link to 2.7.6, make clear EMIT ON UPDATE WITH DELAY
1 parent 51113e2 commit f3cc119

File tree

3 files changed

+93
-111
lines changed

3 files changed

+93
-111
lines changed

‎docs/query-syntax.md

+76-110
Original file line numberDiff line numberDiff line change
@@ -36,20 +36,73 @@ Timeplus supports some advanced `SETTINGS` to fine tune the streaming query proc
3636

3737
As an advanced feature, Timeplus supports various policies to emit results during streaming query.
3838

39-
The syntax is:
39+
:::info
40+
Please note, we updated the EMIT syntax in [Timeplus Enterprise 2.7.6](/enterprise-v2.7#2_7_6). Please upgrade to the latest version to use those refined emit polices.
41+
:::
42+
43+
For [global aggregations](/stream-query#global-aggregation), the syntax is:
4044

4145
```sql
42-
EMIT
43-
[AFTER WATERMARK [WITH DELAY <interval>]
46+
EMIT [STREAM|CHANGELOG]
4447
[PERIODIC <interval> [REPEAT]]
45-
[ON UPDATE]
46-
- [[ AND ]TIMEOUT <interval>]
47-
- [[ AND ]LAST <interval> [ON PROCTIME]]
48+
[ON UPDATE [WITH BATCH <interval>] ]
49+
```
50+
51+
By default `EMIT STREAM` and `PERIODIC 2s` are applied. Advanced settings:
52+
* `EMIT CHANGELOG` works for [global aggregations](/stream-query#global-aggregation) and [non-aggregation tail/filter](/stream-query#non-aggregation). It will output `+1` or `-1` for `_tp_delta` column.
53+
54+
For [time-window aggregations](/stream-query#window-aggregation), the syntax is:
55+
56+
```sql
57+
EMIT
58+
[AFTER WINDOW CLOSE [WITH DELAY <interval> [AND TIMEOUT <interval>]]]
59+
[PERIODIC <interval> [REPEAT] [WITH DELAY <interval> [AND TIMEOUT <interval>]]]
60+
[ON UPDATE [WITH BATCH <interval>] [WITH DELAY <interval> [AND TIMEOUT <interval>]]]
61+
```
62+
You can only choose one of the emit policies: `AFTER WINDOW CLOSE`, `PERIODIC`, or `ON UPDATE`. If you omit any of them, the default policy is `AFTER WINDOW CLOSE`.
63+
64+
Examples:
65+
```sql
66+
EMIT STREAM AFTER WINDOW CLOSE WITH DELAY 1s AND TIMEOUT 5s
67+
EMIT STREAM PERIODIC 1s REPEAT WITH DELAY 1s AND TIMEOUT 5s
68+
EMIT ON UPDATE WITH DELAY 1s AND TIMEOUT 5s
69+
EMIT ON UPDATE WITH BATCH 1s WITH DELAY 1s AND TIMEOUT 5s
70+
```
71+
72+
### EMIT .. WITH DELAY {#emit_delay}
73+
74+
`WITH DELAY` and `AND TIMEOUT` only can be applied to time-window based aggregations.
75+
76+
By default, the query engine will emit the results immediately when the window is closed or other conditions are met. This behavior can be customized using the `WITH DELAY` clause. It allows you to specify extra time to progress the watermark, which can be useful for handling late data.
77+
78+
For example, if you want to wait for 1 second before emitting the results, you can use the following syntax:
79+
80+
```sql
81+
EMIT AFTER WINDOW CLOSE WITH DELAY 1s
4882
```
4983

50-
### EMIT AFTER WATERMARK {#emit_after_wm}
84+
Please check the interactive demo on [Understanding Watermark](/understanding-watermark).
85+
86+
### EMIT .. WITH DELAY AND TIMEOUT {#emit_timeout}
5187

52-
You can omit `EMIT AFTER WATERMARK`, since this is the default behavior for time window aggregations. For example:
88+
For time window based aggregations, when the window is closed is decided by the watermark. A new event outside the window will progress the watermark and inform the query engine to close the previous window and to emit aggregation results.
89+
90+
Say you only get one event for the time window. Since there is no more event, the watermark cannot be moved so the window won't be closed.
91+
92+
`EMIT .. TIMEOUT` is to force the window close, with a timeout after seeing last event.
93+
94+
Please note, if there no single event in the data stream, or in the time window, Proton won't emit result. For example, in the following SQL, you won't get 0 as the count:
95+
96+
```sql
97+
SELECT window_start, count() as count FROM tumble(stream,2s)
98+
GROUP BY window_start
99+
```
100+
101+
Even you add `EMIT .. TIMEOUT` in the SQL, it won't trigger timeout, because the query engine doesn't see any event in the window. If you need to detect such missing event for certain time window, one workaround is to create a heartbeat stream and use `UNION` to create a subquery to combine both heartbeat stream and target stream, for a time window, if all observed events are from heartbeat stream, this means there is no event in the target stream. Please discuss more with us in community slack.
102+
103+
### EMIT AFTER WINDOW CLOSE {#emit_after}
104+
105+
You can omit `EMIT AFTER WINDOW CLOSE`, since this is the default behavior for time window aggregations. For example:
53106

54107
```sql
55108
SELECT device, max(cpu_usage)
@@ -59,15 +112,15 @@ GROUP BY device, window_end
59112

60113
The above example SQL continuously aggregates max cpu usage per device per tumble window for the stream `devices_utils`. Every time a window is closed, Timeplus Proton emits the aggregation results. How to determine the window should be closed? This is done by [Watermark](/stream-query#window-watermark), which is an internal timestamp. It is guaranteed to be increased monotonically per stream query.
61114

62-
### EMIT AFTER WATERMARK WITH DELAY {#emit_after_wm_with_delay}
115+
### EMIT AFTER WINDOW CLOSE WITH DELAY {#emit_after_with_delay}
63116

64117
Example:
65118

66119
```sql
67120
SELECT device, max(cpu_usage)
68121
FROM tumble(device_utils, 5s)
69122
GROUP BY device, widnow_end
70-
EMIT AFTER WATERMARK WITH DELAY 2s;
123+
EMIT AFTER WINDOW CLOSE WITH DELAY 2s;
71124
```
72125

73126
The above example SQL continuously aggregates max cpu usage per device per tumble window for the stream `device_utils`. Every time a window is closed, Timeplus Proton waits for another 2 seconds and then emits the aggregation results.
@@ -110,12 +163,6 @@ EMIT PERIODIC 3s REPEAT
110163

111164
### EMIT ON UPDATE {#emit_on_update}
112165

113-
:::info
114-
115-
This is a new emit policy added in Proton 1.5.
116-
117-
:::
118-
119166
Since Proton 1.5, you can apply `EMIT ON UPDATE` in time windows, such as tumble/hop/session, with `GROUP BY` keys. For example:
120167

121168
```sql
@@ -132,106 +179,25 @@ EMIT ON UPDATE
132179

133180
During the 5 second tumble window, even the window is not closed, as long as the aggregation value(`cnt`) for the same `cid` is different , the results will be emitted.
134181

135-
### EMIT PERIODIC .. ON UPDATE {#emit_periodic_on_update}
136-
137-
:::info
138-
139-
This is a new emit policy added in Proton 1.5.
140-
141-
:::
142-
143-
You can combine `EMIT PERIODIC` and `EMIT ON UPDATE` together. In this case, even the window is not closed, Proton will check the intermediate aggregation result at the specified interval and emit rows if the result is changed.
144-
145-
### EMIT TIMEOUT{#emit_timeout}
146-
147-
For time window based aggregations, when the window is closed is decided by the watermark. A new event outside the window will progress the watermark and inform the query engine to close the previous window and to emit aggregation results.
148-
149-
Say you only get one event for the time window. Since there is no more event, the watermark cannot be moved so the window won't be closed.
150-
151-
`EMIT TIMEOUT` is to force the window close, with a timeout after seeing last event.
152-
153-
Please note, if there no single event in the data stream, or in the time window, Proton won't emit result. For example, in the following SQL, you won't get 0 as the count:
154-
155-
```sql
156-
SELECT window_start, count() as count FROM tumble(stream,2s)
157-
GROUP BY window_start
158-
```
159-
160-
Even you add `EMIT TIMEOUT` in the SQL, it won't trigger timeout, because the query engine doesn't see any event in the window. If you need to detect such missing event for certain time window, one workaround is to create a heartbeat stream and use `UNION` to create a subquery to combine both heartbeat stream and target stream, for a time window, if all observed events are from heartbeat stream, this means there is no event in the target stream. Please discuss more with us in community slack.
161-
162-
### EMIT LAST
163-
164-
In streaming processing, there is one typical query which is processing the last X seconds / minutes / hours of data. For example, show me the cpu usage per device in the last 1 hour. We call this type of processing `Last X Streaming Processing` in Timeplus and Timeplus provides a specialized SQL extension for ease of use: `EMIT LAST <n><UNIT>`. As in other parts of streaming queries, users can use interval shortcuts here.
165-
166-
:::info
167-
168-
By default, `EMIT LAST` uses the event time. Timeplus Proton will seek both streaming storage and historical to backfill data in last X time range. `EMIT LAST .. ON PROCTIME` uses the wall clock time to do the seek.
169-
170-
:::
182+
### EMIT ON UPDATE WITH DELAY {#emit_on_update_with_delay}
171183

172-
#### EMIT LAST for Streaming Tail
173-
174-
Tailing events whose event timestamps are in the last X range.
175-
176-
Examples
177-
178-
```sql
179-
SELECT *
180-
FROM device_utils
181-
WHERE cpu_usage > 80
182-
EMIT LAST 5m
183-
```
184-
185-
The above example filters events in the `device_utils` stream where `cpu_usage` is greater than 80% and events are appended in the last 5 minutes. Internally, Timeplus seeks streaming storage back to 5 minutes (wall-clock time from now) and tailing the data from there.
186-
187-
#### EMIT LAST for Global Aggregation
184+
Adding the `WITH DELAY` to `EMIT ON UPDATE` will allow late event for the window aggregation.
188185

189186
```sql
190-
SELECT <column_name1>, <column_name2>, <aggr_function>
191-
FROM <stream_name>
192-
[WHERE clause]
193-
GROUP BY ...
194-
EMIT LAST INTERVAL <n> <UNIT>
195-
SETTINGS max_keep_windows=<window_count>
196-
```
197-
198-
**Note** Internally Timeplus chops streaming data into small windows and does the aggregation in each small window and as time goes, it slides out old small windows to keep the overall time window fixed and keep the incremental aggregation efficient. By default, the maximum keeping windows is 100. If the last X interval is very big and the periodic emit interval is small,
199-
then users will need to explicitly set up a bigger max window : `last_x_interval / periodic_emit_interval`.
200-
201-
Examples
202-
203-
```sql
204-
SELECT device, count(*)
205-
FROM device_utils
206-
WHERE cpu_usage > 80
207-
GROUP BY device
208-
EMIT PERIODIC 5s AND LAST 1h
209-
SETTINGS max_keep_windows=720;
210-
```
211-
212-
#### EMIT LAST for Windowed Aggregation
213-
214-
```sql
215-
SELECT <column_name1>, <column_name2>, <aggr_function>
216-
FROM <streaming_window_function>(<stream_name>, [<time_column>], [<window_size>], ...)
217-
[WHERE clause]
218-
GROUP BY ...
219-
EMIT LAST INTERVAL <n> <UNIT>
220-
SETTINGS max_keep_windows=<window_count>
187+
SELECT
188+
window_start, cid, count() AS cnt
189+
FROM
190+
tumble(car_live_data, 5s)
191+
WHERE
192+
cid IN ('c00033', 'c00022')
193+
GROUP BY
194+
window_start, cid
195+
EMIT ON UPDATE WITH DELAY 2s
221196
```
222197

223-
Examples
198+
### EMIT ON UPDATE WITH BATCH {#emit_on_update_with_batch}
224199

225-
```sql
226-
SELECT device, window_end, count(*)
227-
FROM tumble(device_utils, 5s)
228-
WHERE cpu_usage > 80
229-
GROUP BY device, window_end
230-
EMIT LAST 1h
231-
SETTINGS max_keep_windows=720;
232-
```
233-
234-
Similarly, we can apply the last X on hopping window.
200+
You can combine `EMIT PERIODIC` and `EMIT ON UPDATE` together. In this case, even the window is not closed, Proton will check the intermediate aggregation result at the specified interval and emit rows if the result is changed.
235201

236202
## PARTITION BY
237203

‎docs/v2-release-notes.md

+16
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,22 @@
22

33
This page summarizes changes for Timeplus Enterprise and Timeplus Proton, on a biweekly basis, including new features and important bug fixes.
44

5+
## Apr 28, 2025
6+
7+
### Timeplus Proton v1.6.15
8+
* fixed the issue that meta store raft service is hardcoded to listen on ipv6
9+
10+
### Timeplus Connect v0.8.17
11+
[timeplus-connect](https://github.com/timeplus-io/timeplus-connect) provides Python connector to interact with Timeplus Enterprise or Timeplus Proton. In this release:
12+
* added support for new `json` data type
13+
* other bug fixes and enhancements
14+
15+
### Timeplus Go Driver v2.1.0
16+
The [proton-go-driver](https://github.com/timeplus-io/proton-go-driver) provides Go connector to interact with Timeplus Enterprise or Timeplus Proton. In this release:
17+
* added support for new `json` data type
18+
* updated protocol to support receiving query id from the server
19+
* other refinements and sample code updates
20+
521
## Apr 14, 2025
622

723
### Timeplus Enterprise v2.6.8, v2.7.4, v2.7.5

‎src/components/TimeplusWatermarkVisualization.js

+1-1
Original file line numberDiff line numberDiff line change
@@ -376,7 +376,7 @@ const TimeplusWatermarkVisualization = () => {
376376
GROUP BY
377377
</span> window_start <br />
378378
<span style={{ color: "#569CD6" }}>
379-
EMIT AFTER WATERMARK
379+
EMIT AFTER WINDOW CLOSE
380380
</span>{" "}
381381
<span style={{ color: "#569CD6" }}>WITH DELAY</span> {lag / 1000}s
382382
</code>

0 commit comments

Comments
 (0)