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

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

INSERT INTO
    bytes_log
VALUES
    (NULL, '2023-01-01 00:00:01'),
    (300, '2023-01-01 00:00:31');

Affected Rows: 2

-- should return error
ADMIN FLUSH_FLOW('find_approx_rate');

Error: 1002(Unexpected), Failed to execute admin function flush_flow: Execution error: Internal error: 1003

DROP FLOW find_approx_rate;

Affected Rows: 0

DROP TABLE bytes_log;

Affected Rows: 0

DROP TABLE approx_rate;

Affected Rows: 0

