CREATE TABLE source_kafka_table (
    `timestamp` BIGINT,
    `status` INTEGER,
    `contentLength` BIGINT,
    `latency` BIGINT,
    `groupUuid` STRING,
    `apiUuid` STRING
) WITH (
    'connector.type' = 'KAFKA',
    'format.encode' = 'JSON',
    'connector.topic' = 'xxxxxxxxx__bsc-source',
    'connector.properties.bootstrap.servers' = 'kafka.bd.baidubce.com:9071',
    'connector.properties.ssl.filename' = 'kafka-key_bd.zip',
    'connector.properties.group.id' = 'test_group',
    'connector.read.startup.mode' = 'latest',
    'watermark.field' = 'timestamp',
    'watermark.threshold' = '1 minutes'
);

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://xxxxxxx.tsdb-ej9v6mg6q8z9.tsdb.iot.bj.baidubce.com',
    'connector.write.max-message-num-per-batch' = '2000'
);

INSERT INTO
   sink_tsdb_table
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`,
                   `status` AS `status`
               FROM
                   (
                       SELECT
                           `timestamp`,
                           `contentLength`,
                           `latency`,
                           `apiUuid`,
                           `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_kafka_table
                   ) AS taba
               GROUP BY
                   TUMBLE(`timestamp`, INTERVAL '1' MINUTE),
                   `apiUuid`,
                   `groupUuid`,
                   `status`
           ) AS tabb
   ) AS tabc