Sink data from RisingWave to Snowflake

CREATE SINK [ IF NOT EXISTS ] sink_name
[FROM sink_from | AS select_query]
WITH (
   connector='snowflake',
   connector_parameter = 'value', ...
);

CREATE MATERIALIZED VIEW ss_mv AS
WITH sub AS CHANGELOG FROM user_behaviors
SELECT
    user_id,
    target_id,
    event_timestamp AT TIME ZONE 'America/Indiana/Indianapolis' AS event_timestamp,
    changelog_op AS __op,
    _changelog_row_id::BIGINT AS __row_id
FROM sub;

CREATE SINK snowflake_sink FROM ss_mv WITH (
    connector            = 'snowflake',
    type                 = 'append-only',
    s3.bucket_name       = 'EXAMPLE_S3_BUCKET',
    s3.credentials.access= 'EXAMPLE_AWS_ACCESS',
    s3.credentials.secret= 'EXAMPLE_AWS_SECRET',
    s3.region_name       = 'EXAMPLE_REGION',
    s3.path              = 'EXAMPLE_S3_PATH'
);

CREATE WAREHOUSE test_warehouse;

-- 1. Create a table to store the results
CREATE TABLE target_table(xxx, xxx);

-- 2. Create a table to store the offset.
CREATE TABLE offset_table (table_name string, offset_row_id int); 

-- 3. Create a task to perform incremental updates on a regular basis.
CREATE OR REPLACE TASK multi_statement_task
WAREHOUSE = COMPUTE_WH
SCHEDULE = '1 HOUR'
AS
BEGIN
    -- 1. Query the offset from the table
    LET my_offset INT;
    
    SELECT COALESCE(MAX(offset_row_id), 0) INTO :my_offset
    FROM offset_table
    WHERE table_name = 'cdc_table'
    LIMIT 1;

    -- 2. Query max row id
    LET max_row_id INT;
    
    SELECT COALESCE(MAX(__row_id), 0) INTO :max_row_id
    FROM cdc_table;
    
    -- 3. Use MERGE INTO statement to merge the data from cdc_table to target_table
    MERGE INTO target_table AS target
    USING (
        SELECT *
        FROM (
            SELECT *,
                   ROW_NUMBER() OVER (PARTITION BY v1 ORDER BY __row_id DESC) AS dedupe_id
            FROM cdc_table
            WHERE __row_id > :my_offset and __row_id <= :max_row_id 
        ) AS subquery
        WHERE dedupe_id = 1
    ) AS source
    ON target.v1= source.v1
    WHEN MATCHED AND source.__op IN (2, 4) THEN DELETE
    WHEN MATCHED AND source.__op IN (1, 3) THEN UPDATE SET target.v1= source.v1, target.v2 = source.v2
    WHEN NOT MATCHED AND source.__op IN (1, 3) THEN INSERT (v1, v2) VALUES (source.v1, source.v2);
    
    -- 4. Update offset_table
    MERGE INTO offset_table AS target
    USING (SELECT 'cdc_table' AS table_name, :max_row_id AS offset_row_id) AS source
    ON target.table_name = source.table_name
    WHEN MATCHED THEN
      UPDATE SET offset_row_id = source.offset_row_id
    WHEN NOT MATCHED THEN
      INSERT (table_name, offset_row_id) VALUES (source.table_name, source.offset_row_id);
      
    -- 5. Delete old data from cdc_table
    DELETE FROM cdc_table
    WHERE __row_id <= :max_row_id;
END;

-- 4. Start the task
ALTER TASK multi_statement_task RESUME;
-- if you want execute the task once, you can
EXECUTE TASK multi_statement_task;

CREATE OR REPLACE DYNAMIC TABLE target_table
TARGET_LAG = '1 hour'
WAREHOUSE  = test_warehouse
AS
SELECT *
FROM (
    SELECT
        *,
        ROW_NUMBER() OVER (PARTITION BY {primary_key} ORDER BY __row_id DESC) AS dedupe_id
    FROM cdc_table              -- cdc_table is the table populated by the sink
) AS sub
WHERE dedupe_id = 1
  AND (__op = 1 OR __op = 3);