CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
    "user_id" VARCHAR(100),
    "activity_name" VARCHAR(100),
    "activity_status" VARCHAR(100),
    "time_at" TIMESTAMP,
    "message" VARCHAR(2000));
CREATE OR REPLACE PUMP "STREAM_PUMP" AS
INSERT INTO "DESTINATION_SQL_STREAM"
SELECT STREAM
    "user_id",
    "activity_name",
    "activity_status",
    "time_at",
    "message"
FROM "SOURCE_SQL_STREAM_001";

CREATE OR REPLACE STREAM "DESTINATION_SQL_STREAM" (
    "time_to_minute" TIMESTAMP,
    "user_id" VARCHAR(100),
    "activity_name" VARCHAR(100),
    "activity_count" INTEGER);
CREATE OR REPLACE PUMP "STREAM_PUMP" AS
INSERT INTO "DESTINATION_SQL_STREAM"
SELECT STREAM
    FLOOR("SOURCE_SQL_STREAM_001"."time_at" TO MINUTE) AS "time_to_minute",
    "user_id",
    "activity_name",
    COUNT(*) AS "activity_count"
FROM "SOURCE_SQL_STREAM_001"
GROUP BY
    "user_id",
    "activity_name",
    FLOOR("SOURCE_SQL_STREAM_001".ROWTIME TO MINUTE),
    FLOOR("SOURCE_SQL_STREAM_001"."time_at" TO MINUTE)
ORDER BY
    FLOOR("SOURCE_SQL_STREAM_001".ROWTIME TO MINUTE),
    "user_id",
    "activity_name";