mirror of
https://github.com/GreptimeTeam/greptimedb.git
synced 2026-01-09 06:42:57 +00:00
a0900f5b90
* feat: use flow batching engine broken: try using logical plan fix: use dummy catalog for logical plan fix: insert plan exec&sqlness grpc addr feat: use frontend instance in flownode in standalone feat: flow type in metasrv&fix: flush flow out of sync& column name alias tests: sqlness update tests: sqlness flow rebuild udpate chore: per review refactor: keep chnl mgr refactor: use catalog mgr for get table tests: use valid sql fix: add more check refactor: put flow type determine to frontend * chore: update proto * chore: update proto to main branch * fix: add locks for create/drop flow&docs: update docs * feat: flush_flow flush all ranges now * test: add align time window test * docs: explain `nodeid` use in check task * refactor: AddAutoColumnRewriter check for Projection * refactor: per review * fix: query without time window also clean dirty time window * chore: better logging * chore: add comments per review * refactor: per review * chore: per review * chore: per review rename args * refactor: per review partially * chore: update docs * chore: use better error variant * chore: better error variant * refactor: rename FlowWorkerManager to FlowStreamingEngine * rename again * refactor: per review * chore: rebase after #5963 merged * refactor: rename all flow_worker_manager occurs * docs: rm resolved TODO
177 lines
6.0 KiB
SQL
177 lines
6.0 KiB
SQL
CREATE TABLE numbers_input_df_func (
|
|
number INT,
|
|
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY(number),
|
|
TIME INDEX(ts)
|
|
);
|
|
|
|
-- call `sum(abs(number))` where `abs` is DataFusion Function and `sum` is flow function
|
|
CREATE FLOW test_numbers_df_func
|
|
SINK TO out_num_cnt_df_func
|
|
AS
|
|
SELECT sum(abs(number)), date_bin(INTERVAL '1 second', ts, '2021-07-01 00:00:00') as time_window FROM numbers_input_df_func GROUP BY time_window;
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
INSERT INTO numbers_input_df_func
|
|
VALUES
|
|
(-20, "2021-07-01 00:00:00.200"),
|
|
(22, "2021-07-01 00:00:00.600");
|
|
|
|
-- flush flow to make sure that table is created and data is inserted
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
-- note that this quote-unquote column is a column-name, **not** a aggregation expr, generated by datafusion
|
|
SELECT "sum(abs(numbers_input_df_func.number))", time_window FROM out_num_cnt_df_func;
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
INSERT INTO numbers_input_df_func
|
|
VALUES
|
|
(23,"2021-07-01 00:00:01.000"),
|
|
(-24,"2021-07-01 00:00:01.500");
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
-- note that this quote-unquote column is a column-name, **not** a aggregation expr, generated by datafusion
|
|
SELECT "sum(abs(numbers_input_df_func.number))", time_window FROM out_num_cnt_df_func;
|
|
|
|
DROP FLOW test_numbers_df_func;
|
|
DROP TABLE numbers_input_df_func;
|
|
DROP TABLE out_num_cnt_df_func;
|
|
|
|
CREATE TABLE numbers_input_df_func (
|
|
number INT,
|
|
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY(number),
|
|
TIME INDEX(ts)
|
|
);
|
|
|
|
-- call `abs(sum(number))`to make sure that calling `abs` function(impl by datafusion) on `sum` function(impl by flow) is working
|
|
CREATE FLOW test_numbers_df_func
|
|
SINK TO out_num_cnt_df_func
|
|
AS
|
|
SELECT abs(sum(number)), date_bin(INTERVAL '1 second', ts, '2021-07-01 00:00:00') as time_window FROM numbers_input_df_func GROUP BY time_window;
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
INSERT INTO numbers_input_df_func
|
|
VALUES
|
|
(-20, "2021-07-01 00:00:00.200"),
|
|
(22, "2021-07-01 00:00:00.600");
|
|
|
|
-- flush flow to make sure that table is created and data is inserted
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
SELECT "abs(sum(numbers_input_df_func.number))", time_window FROM out_num_cnt_df_func;
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
INSERT INTO numbers_input_df_func
|
|
VALUES
|
|
(23,"2021-07-01 00:00:01.000"),
|
|
(-24,"2021-07-01 00:00:01.500");
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
SELECT "abs(sum(numbers_input_df_func.number))", time_window FROM out_num_cnt_df_func;
|
|
|
|
DROP FLOW test_numbers_df_func;
|
|
DROP TABLE numbers_input_df_func;
|
|
DROP TABLE out_num_cnt_df_func;
|
|
|
|
-- test date_bin
|
|
CREATE TABLE numbers_input_df_func (
|
|
number INT,
|
|
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY(number),
|
|
TIME INDEX(ts)
|
|
);
|
|
|
|
CREATE FLOW test_numbers_df_func
|
|
SINK TO out_num_cnt_df_func
|
|
AS
|
|
SELECT max(number) - min(number) as maxmin, date_bin(INTERVAL '1 second', ts, '2021-07-01 00:00:00'::Timestamp) as time_window FROM numbers_input_df_func GROUP BY time_window;
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
INSERT INTO numbers_input_df_func
|
|
VALUES
|
|
(20, "2021-07-01 00:00:00.200"),
|
|
(22, "2021-07-01 00:00:00.600");
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
SELECT maxmin, time_window FROM out_num_cnt_df_func;
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
INSERT INTO numbers_input_df_func
|
|
VALUES
|
|
(23,"2021-07-01 00:00:01.000"),
|
|
(24,"2021-07-01 00:00:01.500");
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
SELECT maxmin, time_window FROM out_num_cnt_df_func;
|
|
|
|
DROP FLOW test_numbers_df_func;
|
|
DROP TABLE numbers_input_df_func;
|
|
DROP TABLE out_num_cnt_df_func;
|
|
|
|
|
|
-- test date_trunc
|
|
CREATE TABLE numbers_input_df_func (
|
|
number INT,
|
|
ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
PRIMARY KEY(number),
|
|
TIME INDEX(ts)
|
|
);
|
|
|
|
CREATE FLOW test_numbers_df_func
|
|
SINK TO out_num_cnt
|
|
AS
|
|
SELECT date_trunc('second', ts) as time_window, sum(number) as sum_num FROM numbers_input_df_func GROUP BY date_trunc('second', ts);
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
INSERT INTO numbers_input_df_func
|
|
VALUES
|
|
(20, "2021-07-01 00:00:00.200"),
|
|
(22, "2021-07-01 00:00:00.600");
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
SELECT time_window, sum_num FROM out_num_cnt;
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
INSERT INTO numbers_input_df_func
|
|
VALUES
|
|
(23,"2021-07-01 00:00:01.000"),
|
|
(24,"2021-07-01 00:00:01.500");
|
|
|
|
-- SQLNESS REPLACE (ADMIN\sFLUSH_FLOW\('\w+'\)\s+\|\n\+-+\+\n\|\s+)[0-9]+\s+\| $1 FLOW_FLUSHED |
|
|
ADMIN FLUSH_FLOW('test_numbers_df_func');
|
|
|
|
SELECT time_window, sum_num FROM out_num_cnt;
|
|
|
|
DROP FLOW test_numbers_df_func;
|
|
DROP TABLE numbers_input_df_func;
|
|
DROP TABLE out_num_cnt;
|