[新機能]Dynamic TableのSCHEDULER attributeが一般提供となったのでdbtと合わせて試してみた

-- 作成時に指定
CREATE DYNAMIC TABLE my_dt
  WAREHOUSE = my_wh
  SCHEDULER = DISABLE          -- TARGET_LAG は指定しない
  AS SELECT ...;

-- 既存DTの変更
ALTER DYNAMIC TABLE my_dt SET SCHEDULER = DISABLE;

-- ENABLE に戻す場合は TARGET_LAG も再設定
ALTER DYNAMIC TABLE my_dt SET SCHEDULER = ENABLE TARGET_LAG = '10 minutes';

-- 手動リフレッシュ
ALTER DYNAMIC TABLE my_dt REFRESH;

CREATE OR REPLACE DYNAMIC TABLE KAWABATA_MART_DB.PUBLIC.DT_ORDERS_SCHEDULER_ENABLED
  TARGET_LAG = '10 minutes'
  WAREHOUSE = COMPUTE_WH
  REFRESH_MODE = FULL
  INITIALIZE = ON_CREATE
  SCHEDULER = ENABLE
  AS
    SELECT
      o.ID AS ORDER_ID,
      o.CUSTOMER AS CUSTOMER_ID,
      o.ORDERED_AT,
      o.STORE_ID,
      o.ORDER_TOTAL,
      c.NAME AS CUSTOMER_NAME
    FROM KAWABATA_MART_DB.KAWABATA_RAW.RAW_ORDERS o
    JOIN KAWABATA_MART_DB.KAWABATA_RAW.RAW_CUSTOMERS c
      ON o.CUSTOMER = c.ID;

CREATE OR REPLACE DYNAMIC TABLE KAWABATA_MART_DB.PUBLIC.DT_ORDERS_SCHEDULER_DISABLED
  WAREHOUSE = COMPUTE_WH
  REFRESH_MODE = FULL
  INITIALIZE = ON_CREATE
  SCHEDULER = DISABLE
  AS
    SELECT
      o.ID AS ORDER_ID,
      o.CUSTOMER AS CUSTOMER_ID,
      o.ORDERED_AT,
      o.STORE_ID,
      o.ORDER_TOTAL,
      c.NAME AS CUSTOMER_NAME
    FROM KAWABATA_MART_DB.KAWABATA_RAW.RAW_ORDERS o
    JOIN KAWABATA_MART_DB.KAWABATA_RAW.RAW_CUSTOMERS c
      ON o.CUSTOMER = c.ID;

-- =============================================
-- 検証3: SCHEDULER=DISABLE のDTを手動リフレッシュ
--   → テストデータ挿入後に手動リフレッシュし、更新結果の反映を確認
-- =============================================

-- 3a: リフレッシュ前の件数を確認
SELECT COUNT(*) AS ROW_COUNT_BEFORE_INSERT FROM KAWABATA_MART_DB.PUBLIC.DT_ORDERS_SCHEDULER_DISABLED;

-- 3b: ソーステーブルにテストデータを挿入
INSERT INTO KAWABATA_MART_DB.KAWABATA_RAW.RAW_CUSTOMERS (ID, NAME)
VALUES ('test-customer-001', 'Test User For DT Verification');

INSERT INTO KAWABATA_MART_DB.KAWABATA_RAW.RAW_ORDERS (ID, CUSTOMER, ORDERED_AT, STORE_ID, SUBTOTAL, TAX_PAID, ORDER_TOTAL)
VALUES
  ('test-order-001', 'test-customer-001', '2026-03-31', '1', 1000, 100, 1100),
  ('test-order-002', 'test-customer-001', '2026-03-31', '2', 2000, 200, 2200);

-- 3c: 手動リフレッシュ実行
ALTER DYNAMIC TABLE KAWABATA_MART_DB.PUBLIC.DT_ORDERS_SCHEDULER_DISABLED REFRESH;

-- 3d: リフレッシュ後の件数を確認（+2 されていれば手動リフレッシュにより更新結果が反映されたと判断できる）
SELECT COUNT(*) AS ROW_COUNT_AFTER_REFRESH FROM KAWABATA_MART_DB.PUBLIC.DT_ORDERS_SCHEDULER_DISABLED;

-- 3e: 挿入したテストデータがDTに反映されているか確認
SELECT * FROM KAWABATA_MART_DB.PUBLIC.DT_ORDERS_SCHEDULER_DISABLED
WHERE CUSTOMER_ID = 'test-customer-001';

-- =============================================
-- 検証4: タスクによる SCHEDULER=DISABLE DTの定期リフレッシュ
--   → CRON スケジュールで手動リフレッシュを自動化
-- =============================================

-- 4a: タスクを作成（1時間ごとに手動リフレッシュを実行）
CREATE OR REPLACE TASK KAWABATA_MART_DB.PUBLIC.TASK_REFRESH_DT_ORDERS
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = 'USING CRON 0 * * * * UTC'
  COMMENT = 'Hourly manual refresh of DT_ORDERS_SCHEDULER_DISABLED'
  AS
    ALTER DYNAMIC TABLE KAWABATA_MART_DB.PUBLIC.DT_ORDERS_SCHEDULER_DISABLED REFRESH;

-- 4b: タスクを有効化
ALTER TASK KAWABATA_MART_DB.PUBLIC.TASK_REFRESH_DT_ORDERS RESUME;

-- 4c: タスクの状態確認
SHOW TASKS LIKE 'TASK_REFRESH_DT_ORDERS' IN SCHEMA KAWABATA_MART_DB.PUBLIC;

-- 4d: タスクを即時実行（スケジュールを待たずにすぐ実行）
EXECUTE TASK KAWABATA_MART_DB.PUBLIC.TASK_REFRESH_DT_ORDERS;

-- 4e: タスク実行履歴の確認（実行後しばらくしてから確認）
SELECT NAME, STATE, SCHEDULED_TIME, COMPLETED_TIME, ERROR_CODE, ERROR_MESSAGE
FROM TABLE(KAWABATA_MART_DB.INFORMATION_SCHEMA.TASK_HISTORY(
  TASK_NAME => 'TASK_REFRESH_DT_ORDERS',
  SCHEDULED_TIME_RANGE_START => DATEADD('hour', -2, CURRENT_TIMESTAMP())
))
ORDER BY SCHEDULED_TIME DESC
LIMIT 5;

-- 4f: タスクを停止（検証完了後）
ALTER TASK KAWABATA_MART_DB.PUBLIC.TASK_REFRESH_DT_ORDERS SUSPEND;

-- ENABLE → DISABLE
ALTER DYNAMIC TABLE ... SET SCHEDULER = DISABLE;

-- DISABLE → ENABLE（TARGET_LAG の再設定が必要）
ALTER DYNAMIC TABLE ... SET SCHEDULER = ENABLE TARGET_LAG = '10 minutes';

jaffle_shop_incr_dt/
├── dbt_project.yml
├── profiles.yml
├── packages.yml
├── macros/
│   └── refresh_dynamic_table.sql
└── models/
    ├── sources.yml
    ├── schema.yml
    ├── staging/
    │   ├── stg_orders.sql          (view)
    │   ├── stg_customers.sql       (view)
    │   ├── stg_items.sql           (view)
    │   └── stg_products.sql        (view)
    ├── intermediate/
    │   └── int_order_items_dynamic_table.sql   ← dynamic_table（実行後に SCHEDULER = DISABLE へ変更）
    └── marts/
        └── fct_orders_from_dynamic_table.sql   (view)

version: 2

sources:
  - name: jaffle_raw
    database: KAWABATA_MART_DB
    schema: KAWABATA_RAW
    tables:
      - name: RAW_CUSTOMERS
      - name: RAW_ORDERS
      - name: RAW_ITEMS
      - name: RAW_PRODUCTS
      - name: RAW_STORES
      - name: RAW_SUPPLIES

-- models/intermediate/int_order_items_dynamic_table.sql
{{
    config(
        materialized='dynamic_table',
        snowflake_warehouse='COMPUTE_WH',
        target_lag='DOWNSTREAM',
        pre_hook=[
            "ALTER DYNAMIC TABLE IF EXISTS {{ this }} SET SCHEDULER = ENABLE TARGET_LAG = DOWNSTREAM"
        ],
        post_hook=[
            "ALTER DYNAMIC TABLE {{ this }} REFRESH",
            "ALTER DYNAMIC TABLE {{ this }} SET SCHEDULER = DISABLE"
        ]
    )
}}

SELECT
    i.ITEM_ID,
    i.ORDER_ID,
    i.SKU,
    o.CUSTOMER_ID,
    o.ORDERED_AT,
    o.STORE_ID,
    o.ORDER_TOTAL,
    p.PRODUCT_NAME,
    p.PRODUCT_TYPE,
    p.PRICE AS UNIT_PRICE
FROM {{ ref('stg_items') }} i
JOIN {{ ref('stg_orders') }} o ON i.ORDER_ID = o.ORDER_ID
JOIN {{ ref('stg_products') }} p ON i.SKU = p.SKU

# 初回作成（pre_hook は IF EXISTS により安全にスキップされる）
dbt run --select int_order_items_dynamic_table --full-refresh

# 2回目以降の差分更新（dbt run だけでOK）
dbt run --select int_order_items_dynamic_table

1. pre_hook:   ALTER ... IF EXISTS ... SET SCHEDULER = ENABLE TARGET_LAG = DOWNSTREAM
               → dbt の ALTER 比較でエラーにならないようにする
2. dbt 本体:   設定比較 → 変更なし → skip（DDL 発行なし）
3. post_hook①: ALTER DYNAMIC TABLE ... REFRESH
               → 手動リフレッシュ実行（差分データ更新）
4. post_hook②: ALTER ... SET SCHEDULER = DISABLE
               → 自動リフレッシュを再度無効化