[新機能]Row timestampsが一般提供となったのでdbtのincremental modelと合わせて試してみた

-- タイムスタンプ表示形式の設定
ALTER SESSION SET TIMESTAMP_TZ_OUTPUT_FORMAT = 'YYYY-MM-DDTHH:MI:SS.FF3 TZH';
ALTER SESSION SET TIMEZONE = 'UTC';

-- テスト用テーブルの作成（ROW_TIMESTAMP = TRUE）
CREATE OR REPLACE TABLE KAWABATA_MART_DB.DBT_TKAWABATA.ROW_TIMESTAMP_TEST (
    id INT,
    value STRING
) ROW_TIMESTAMP = TRUE;

-- データ挿入
INSERT INTO KAWABATA_MART_DB.DBT_TKAWABATA.ROW_TIMESTAMP_TEST VALUES (1, 'first');
INSERT INTO KAWABATA_MART_DB.DBT_TKAWABATA.ROW_TIMESTAMP_TEST VALUES (2, 'second');

-- METADATA$ROW_LAST_COMMIT_TIME を確認
SELECT 
    id, 
    value, 
    METADATA$ROW_LAST_COMMIT_TIME AS row_timestamp
FROM KAWABATA_MART_DB.DBT_TKAWABATA.ROW_TIMESTAMP_TEST
ORDER BY row_timestamp;

SHOW PARAMETERS LIKE 'ROW_TIMESTAMP_DEFAULT' IN SCHEMA KAWABATA_MART_DB.DBT_TKAWABATA;

ALTER SCHEMA KAWABATA_MART_DB.DBT_TKAWABATA SET ROW_TIMESTAMP_DEFAULT = TRUE;

-- 既存テーブルのコピーを作成してテスト
CREATE OR REPLACE TABLE KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_ROW_TS_TEST AS
SELECT * FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS LIMIT 100;

-- 新しい行を挿入
INSERT INTO KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_ROW_TS_TEST 
SELECT * FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS LIMIT 5;

-- 追加した5行にタイムスタンプが付与
SELECT 
    ORDER_ID,
    ORDER_DATE,
    METADATA$ROW_LAST_COMMIT_TIME AS row_timestamp
FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_ROW_TS_TEST
ORDER BY row_timestamp NULLS FIRST;

{% if is_incremental() %}
WHERE updated_at > (SELECT MAX(updated_at) FROM {{ this }})
{% endif %}

version: 2

sources:
  - name: raw
    database: KAWABATA_MART_DB
    schema: DBT_TKAWABATA
    tables:
      - name: ORDERS_ROW_TS_TEST
        description: "ROW_TIMESTAMP enabled orders table for incremental processing"

-- インクリメンタルモデル: Row Timestampを活用した増分処理
-- 初回実行時はフルロード、2回目以降は新規・更新行のみを処理

{#
  materialized='incremental': インクリメンタルモデルとして実行
  unique_key='ORDER_ID': MERGEのキー（重複時は更新）
  incremental_strategy='merge': MERGE文で既存行を更新、新規行を挿入
#}

{{
  config(
    materialized='incremental',   
    unique_key='ORDER_ID',        
    incremental_strategy='merge'  
  )
}}

SELECT
    ORDER_ID,
    LOCATION_ID,
    CUSTOMER_ID,
    SUBTOTAL,
    TAX_PAID,
    ORDER_TOTAL,
    ORDER_DATE,
    ORDER_COST,
    COUNT_ORDER_ITEMS,
    IS_FOOD_ORDER,
    IS_DRINK_ORDER,
    -- Snowflake Row Timestamp: 行がコミットされた時刻を自動取得
    METADATA$ROW_LAST_COMMIT_TIME AS ROW_TIMESTAMP
FROM {{ source('raw', 'ORDERS_ROW_TS_TEST') }}

-- 2回目以降の実行時のみ適用されるフィルタ条件
{% if is_incremental() %}
WHERE METADATA$ROW_LAST_COMMIT_TIME > (
    -- ターゲットテーブルの最新タイムスタンプより新しい行のみ取得
    SELECT COALESCE(MAX(ROW_TIMESTAMP), '1900-01-01'::TIMESTAMP_LTZ)
    FROM {{ this }}
)
{% endif %}

dbt run --select orders_incremental

-- レコードの追加
INSERT INTO KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_ROW_TS_TEST 
SELECT * FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS 
WHERE ORDER_ID NOT IN (SELECT ORDER_ID FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_ROW_TS_TEST)
LIMIT 10

dbt run --select orders_incremental

SELECT 
    DATE_TRUNC('minute', ROW_TIMESTAMP) AS commit_minute,
    COUNT(*) AS row_count
FROM KAWABATA_MART_DB.DBT_TKAWABATA.ORDERS_INCREMENTAL
WHERE ROW_TIMESTAMP IS NOT NULL
GROUP BY 1
ORDER BY 1;