mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-10-16 00:42:29 +00:00
discord9
7f76ad5439
* feat: check schema on create Signed-off-by: discord9 <discord9@163.com> * chore: update sqlness Signed-off-by: discord9 <discord9@163.com> * fix(flow): avoid duplicate fields when matching sink schema Signed-off-by: discord9 <discord9@163.com> * fix: null handling Signed-off-by: discord9 <discord9@163.com> * chore: per review Signed-off-by: discord9 <discord9@163.com> * chore: debug log Signed-off-by: discord9 <discord9@163.com> --------- Signed-off-by: discord9 <discord9@163.com>
187 lines
5.7 KiB
Plaintext
187 lines
5.7 KiB
Plaintext
CREATE TABLE bytes_log (
|
|
byte INT,
|
|
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
-- event time
|
|
TIME INDEX(ts)
|
|
);
|
|
|
|
Affected Rows: 0
|
|
|
|
-- TODO(discord9): remove this after auto infer table's time index is impl
|
|
CREATE TABLE approx_rate (
|
|
rate DOUBLE NULL,
|
|
time_window TIMESTAMP,
|
|
update_at TIMESTAMP,
|
|
bb DOUBLE NULL,
|
|
TIME INDEX(time_window)
|
|
)with('merge_mode'='last_non_null');
|
|
|
|
Affected Rows: 0
|
|
|
|
INSERT INTO approx_rate(rate, time_window, update_at) VALUES (0.0, '2023-01-01 00:00:00', TIMESTAMP '2023-01-01 00:00:00.100');
|
|
|
|
Affected Rows: 1
|
|
|
|
INSERT INTO approx_rate(time_window, update_at, bb) VALUES ('2023-01-01 00:00:00', TIMESTAMP '2023-01-01 00:00:00.200', 50.0);
|
|
|
|
Affected Rows: 1
|
|
|
|
select * from approx_rate;
|
|
|
|
+------+---------------------+-------------------------+------+
|
|
| rate | time_window | update_at | bb |
|
|
+------+---------------------+-------------------------+------+
|
|
| 0.0 | 2023-01-01T00:00:00 | 2023-01-01T00:00:00.200 | 50.0 |
|
|
+------+---------------------+-------------------------+------+
|
|
|
|
CREATE FLOW find_approx_rate SINK TO approx_rate AS
|
|
SELECT
|
|
(max(byte) - min(byte)) / 30.0 as rate,
|
|
date_bin(INTERVAL '30 second', ts) as time_window,
|
|
TIMESTAMP '2023-01-01 00:00:10' as update_at
|
|
from
|
|
bytes_log
|
|
GROUP BY
|
|
time_window;
|
|
|
|
Affected Rows: 0
|
|
|
|
SHOW CREATE TABLE approx_rate;
|
|
|
|
+-------------+--------------------------------------------+
|
|
| Table | Create Table |
|
|
+-------------+--------------------------------------------+
|
|
| approx_rate | CREATE TABLE IF NOT EXISTS "approx_rate" ( |
|
|
| | "rate" DOUBLE NULL, |
|
|
| | "time_window" TIMESTAMP(3) NOT NULL, |
|
|
| | "update_at" TIMESTAMP(3) NULL, |
|
|
| | "bb" DOUBLE NULL, |
|
|
| | TIME INDEX ("time_window") |
|
|
| | ) |
|
|
| | |
|
|
| | ENGINE=mito |
|
|
| | WITH( |
|
|
| | merge_mode = 'last_non_null' |
|
|
| | ) |
|
|
+-------------+--------------------------------------------+
|
|
|
|
INSERT INTO
|
|
bytes_log
|
|
VALUES
|
|
(NULL, '2023-01-01 00:00:01'),
|
|
(300, '2023-01-01 00:00:31');
|
|
|
|
Affected Rows: 2
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('find_approx_rate');
|
|
|
|
+--------------------------------------+
|
|
| ADMIN FLUSH_FLOW('find_approx_rate') |
|
|
+--------------------------------------+
|
|
| FLOW_FLUSHED |
|
|
+--------------------------------------+
|
|
|
|
SELECT * FROM approx_rate;
|
|
|
|
+------+---------------------+---------------------+------+
|
|
| rate | time_window | update_at | bb |
|
|
+------+---------------------+---------------------+------+
|
|
| 0.0 | 2023-01-01T00:00:00 | 2023-01-01T00:00:10 | 50.0 |
|
|
| 0.0 | 2023-01-01T00:00:30 | 2023-01-01T00:00:10 | |
|
|
+------+---------------------+---------------------+------+
|
|
|
|
CREATE FLOW find_bb_only SINK TO approx_rate AS
|
|
SELECT
|
|
date_bin(INTERVAL '30 second', ts) as time_window,
|
|
TIMESTAMP '2023-01-01 00:00:59' as update_at,
|
|
CAST(max(byte) AS DOUBLE) as bb
|
|
from
|
|
bytes_log
|
|
GROUP BY
|
|
time_window;
|
|
|
|
Affected Rows: 0
|
|
|
|
-- make new windows dirty after flow is created
|
|
INSERT INTO bytes_log VALUES (600, '2023-01-01 00:00:10'), (320, '2023-01-01 00:00:35');
|
|
|
|
Affected Rows: 2
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('find_bb_only');
|
|
|
|
+----------------------------------+
|
|
| ADMIN FLUSH_FLOW('find_bb_only') |
|
|
+----------------------------------+
|
|
| FLOW_FLUSHED |
|
|
+----------------------------------+
|
|
|
|
SELECT * FROM approx_rate;
|
|
|
|
+------+---------------------+---------------------+-------+
|
|
| rate | time_window | update_at | bb |
|
|
+------+---------------------+---------------------+-------+
|
|
| 0.0 | 2023-01-01T00:00:00 | 2023-01-01T00:00:59 | 600.0 |
|
|
| 0.0 | 2023-01-01T00:00:30 | 2023-01-01T00:00:59 | 320.0 |
|
|
+------+---------------------+---------------------+-------+
|
|
|
|
DROP FLOW find_approx_rate;
|
|
|
|
Affected Rows: 0
|
|
|
|
DROP FLOW find_bb_only;
|
|
|
|
Affected Rows: 0
|
|
|
|
DROP TABLE bytes_log;
|
|
|
|
Affected Rows: 0
|
|
|
|
DROP TABLE approx_rate;
|
|
|
|
Affected Rows: 0
|
|
|
|
CREATE TABLE bytes_log (
|
|
byte INT,
|
|
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
-- event time
|
|
TIME INDEX(ts)
|
|
);
|
|
|
|
Affected Rows: 0
|
|
|
|
-- TODO(discord9): remove this after auto infer table's time index is impl
|
|
CREATE TABLE approx_rate (
|
|
rate DOUBLE NULL,
|
|
time_window TIMESTAMP,
|
|
update_at TIMESTAMP,
|
|
bb DOUBLE NULL,
|
|
TIME INDEX(time_window)
|
|
);
|
|
|
|
Affected Rows: 0
|
|
|
|
-- Without merge_mode=last_non_null, this partial output is rejected at CREATE FLOW time.
|
|
-- SQLNESS REPLACE (in\scontext:\sFailed\sto\srewrite\splan:\sError\sduring\splanning:.*) in context: Failed to rewrite plan
|
|
CREATE FLOW find_approx_rate SINK TO approx_rate AS
|
|
SELECT
|
|
(max(byte) - min(byte)) / 30.0 as rate,
|
|
date_bin(INTERVAL '30 second', ts) as time_window,
|
|
TIMESTAMP '2023-01-01 00:00:10' as update_at
|
|
from
|
|
bytes_log
|
|
GROUP BY
|
|
time_window;
|
|
|
|
Error: 3001(EngineExecuteQuery), Datafusion error: Plan("Flow output schema does not match sink table schema: found 3 flow output columns and 4 sink table columns. flow output columns: [\"rate\", \"time_window\", \"update_at\"], sink table columns: [\"rate\", \"time_window\", \"update_at\", \"bb\"], extra flow columns not in sink: [], missing sink columns from flow output: [\"bb\"]") in context: Failed to rewrite plan
|
|
|
|
DROP TABLE bytes_log;
|
|
|
|
Affected Rows: 0
|
|
|
|
DROP TABLE approx_rate;
|
|
|
|
Affected Rows: 0
|
|
|