CREATE TABLE sink_tsdb_table (
     `datapoints` ARRAY < ROW(
        `timestamp` BIGINT,
        `metric` STRING,
        `value` BIGINT,
        `tags` MAP < STRING, STRING >
    ) >
) WITH (
    'connector.type' = 'TSDB',
    'connector.emit' = 'batch',
    'connector.url' = 'http://xxxxxx.tsdb-ej9v6mg6q8z9.tsdb.iot.bj.baidubce.com',
    'connector.net.max-connection-total' = '10',
    'connector.net.max-connection-per-route' = '10',
    'connector.net.connection-timeout-ms' = '60000',
    'connector.write.max-message-num-per-batch' = '1000'
);

CREATE TABLE sink_tsdb_table (
    `datapoints` ARRAY < ROW(
        `timestamp` BIGINT,
        `metric` STRING,
        `value` BIGINT,
        `tags` MAP < STRING, STRING >
    ) >
) WITH (
    'connector.type' = 'TSDB',
    'format.encode' = 'JSON',
    'connector.emit' = 'BATCH',
    'connector.url' = 'http://bscwgtest.tsdb-ej9v6mg6q8z9.tsdb.iot.bj.baidubce.com',
    'connector.write.max-message-num-per-batch' = '2000'
);

INSERT INTO
    tsdb_sink
SELECT
    ARRAY [
        ROW(`timestamp`, `count_name` , `count`, `common_tags`),
        ROW(`timestamp`, `traffic_name`, `traffic`, `common_tags`),
        ROW(`timestamp`, `latency_name`, `latency`, `common_tags`)
    ]
FROM
    (
        SELECT
            `timestamp`,
            'count' AS `count_name`,
            `count`,
            'traffic' AS `traffic_name`,
            `traffic`,
            'latency' AS `latency_name`,
            `latency`,
            MAP ['apiUuid', `apiUuid`, 'groupUuid', `groupUuid`, 'status', `status`] AS `common_tags`
        FROM
            (
                SELECT
                    TO_BIGINT(TUMBLE_START(`timestamp`, INTERVAL '1' MINUTE)) AS `timestamp`,
                    COUNT(1) AS `count`,
                    SUM(contentLength) AS `traffic`,
                    SUM(latency) AS `latency`,
                    apiUuid AS `apiUuid`,
                    groupUuid AS `groupUuid`,
                    CASE
                        WHEN status >= 200
                        AND status < 300 THEN '2xx'
                        WHEN status >= 300
                        AND status < 200 THEN '3xx'
                        WHEN status >= 400
                        AND status < 500 THEN '4xx'
                        WHEN status >= 500
                        AND status < 600 THEN '5xx'
                        ELSE 'oth'
                    END AS `status`
                FROM
                    source_table
                GROUP BY
                    TUMBLE(`timestamp`, INTERVAL '1' MINUTE),
                    apiUuid,
                    groupUuid,
                    status
            )
    )