Snowflake で日別・店別の集計テーブルを、新しく入った行と直した行の分だけ更新する｜ストリームとタスクで組む一式と、Dynamic Tables を選ぶ場合

日別・店別の集計テーブルを、毎朝、前の日に増えた注文と直された注文の分だけ足して更新しているとします。全件を集計し直すより処理する行は少なく済みますが、足し方を一度誤ると、その日の数はずれたまま残ります。MERGE がエラーにならなければ、集計テーブルを見ても気づけません。

この記事は、その更新を Snowflake のストリームとタスクで組み、ずれを毎朝見つけてメールで知らせるまでの SQL の一式です。差が出た日に、その日の分だけを作り直す手順も入れました。集計だけが目的なら Dynamic Tables のほうが作るものは少ないので、どちらを選ぶかの決め手も書きます。

載せた結果は、Snowflake の導入を支援している当社が、試用のアカウントで実際に出したものです。注文は架空で、渋谷店と新宿店の分を作りました。

差分を足すタスクのあとに、全件から集計し直して比べるアラートを置く

差分の更新は、ストリームとタスクで組みます。ストリームは、テーブルの変更のうち、まだ使っていない分を返すオブジェクトです。タスクは、決めた時刻に SQL を実行する仕組みで、ここでは MERGE を実行します。MERGE は、一致する行があれば更新し、無ければ入れる文です。

この記事では、ストリームを SELECT するだけのことを「中身を見る」と書きます。MERGE のような DML の文でストリームを使うと、使った変更はもう返されません。これを「消費する」と書きます。消費すると、ストリームが持っている位置（オフセット）が先へ進みます。

足したあとの確かめには、アラートを使います。アラートは、決めた時刻に条件の SQL を実行し、結果が1行以上あれば、決めた処理を実行する仕組みです。ここでは、注文テーブルから集計し直した値と集計テーブルが合わない日・店があれば、メールを送ります。

差分で足すなら、全件と比べるアラートを組にする。これが当社の判断です。USING の書き方を誤っても、MERGE はエラーにならなかったからです。

毎朝全件から集計し直して比べるなら、最初から毎朝全件で作り直せばよいのでは、という疑問もあると思います。全件で作り直して、取り込みとのあいだの時間に収まるなら、当社もそのほうがよいと考えます。作るものが少なく、足し方の誤りも起きないからです。

差分にするのは、全件の集計が間に合わなくなってからで足ります。そのときの突き合わせは、比べる日を直近に絞るか、比べる回数を減らします。この形は試していないので、SQL は載せていません。この記事では仕組みを見せるため、毎朝全件で比べています。

例にするのは、注文テーブル ORDERS と、集計テーブル DAILY_STORE_SALES です。ORDERS は注文番号・注文日・店・金額の4列で、毎朝、前の日の分が入ります。DAILY_STORE_SALES は、注文日と店ごとに、件数 ORDERS_N と金額 AMOUNT_SUM を持ちます。毎朝の取り込みそのものをタスクにする形は、取り込みの自動化の記事に書きました。

検証用のデータベースに注文テーブルと集計テーブルを作り、ストリームの中身を見る

ここから試す SQL は、どれも試すときの文です。本番とは別の、test_stream という検証用のデータベースで、記事の順に実行します。本番に入れる一式は、試すための文を除いて後半にまとめました。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE ROLE ACCOUNTADMIN;
USE SECONDARY ROLES NONE;
USE WAREHOUSE COMPUTE_WH;
CREATE DATABASE IF NOT EXISTS test_stream;
CREATE SCHEMA IF NOT EXISTS test_stream.raw;
USE SCHEMA test_stream.raw;
CREATE OR REPLACE TABLE ORDERS (ORDER_ID NUMBER, ORDER_DATE DATE, STORE STRING, AMOUNT NUMBER);
INSERT INTO ORDERS
  SELECT SEQ4() + 1, DATEADD(DAY, FLOOR(SEQ4() / 100), '2026-10-19'::DATE),
         DECODE(MOD(SEQ4(), 2), 0, '渋谷店', '新宿店'),
         1000 + MOD((SEQ4() + 1) * 7919, 9000)
  FROM TABLE(GENERATOR(ROWCOUNT => 300));
CREATE OR REPLACE TABLE DAILY_STORE_SALES (ORDER_DATE DATE, STORE STRING, ORDERS_N NUMBER, AMOUNT_SUM NUMBER, UPDATED_AT TIMESTAMP_LTZ);
INSERT INTO DAILY_STORE_SALES
  SELECT ORDER_DATE, STORE, COUNT(*), SUM(AMOUNT), CURRENT_TIMESTAMP() FROM ORDERS GROUP BY ORDER_DATE, STORE;
CREATE OR REPLACE STREAM ORDERS_STREAM ON TABLE ORDERS;
SELECT ORDER_DATE, STORE, ORDERS_N, AMOUNT_SUM FROM DAILY_STORE_SALES ORDER BY ORDER_DATE, STORE;

注文は19日から21日まで、1日100行ずつです。店は渋谷店と新宿店が交互で、金額は注文番号から式で決めています。集計テーブルには、この300行を日と店ごとに集計した6行が入ります。

集計テーブルに値を入れてから、ストリームを作ります。ストリームが返すのは、作った時点よりあとの変更だけだからです。ここで作ったのは標準ストリームで、行の追加・更新・削除をすべて返します。APPEND_ONLY を付けると追加だけになり、金額の直しや削除は入りません。注文を直すことがあるなら、標準ストリームを使います。

次に、22日分の80行を入れます。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
INSERT INTO ORDERS
  SELECT SEQ4() + 301, '2026-10-22'::DATE, DECODE(MOD(SEQ4(), 2), 0, '渋谷店', '新宿店'),
         1000 + MOD((SEQ4() + 301) * 7919, 9000)
  FROM TABLE(GENERATOR(ROWCOUNT => 80));

この80行は、ストリームの中身を見ても減らず、このあとタスクの MERGE を実行するまで残っていました。中身を見ただけでは、ストリームは消費されません。位置が進むのは DML の文で使われたときだけだと、Snowflake のドキュメントにもあります。

MERGE はストリームの差分を日と店でまとめて足し、タスクで毎朝実行する

ストリームの中身には、元のテーブルの列のほかに、METADATA$ で始まる列が3つ付きます。METADATA$ACTION・METADATA$ISUPDATE・METADATA$ROW_ID で、この記事で使うのは前の2つです。METADATA$ACTION の値は INSERT と DELETE の2つだけで、UPDATE という値はありません。

注文の金額を直すと、ストリームには、直す前の行の DELETE と、直したあとの行の INSERT が組で入ります。どちらも METADATA$ISUPDATE は TRUE です。そこで MERGE では、INSERT の行を足し、DELETE の行を引きます。

日と店ごとにまとめた差分を MERGE で集計テーブルに足す文を、タスクの本体にします。まとめないとどうなるかは、あとで試します。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
CREATE OR REPLACE TASK REFRESH_DAILY_SALES
  WAREHOUSE = COMPUTE_WH
  SCHEDULE = 'USING CRON 15 6 * * * Asia/Tokyo'
  WHEN SYSTEM$STREAM_HAS_DATA('ORDERS_STREAM')
  AS
  MERGE INTO DAILY_STORE_SALES T
  USING (
    SELECT ORDER_DATE, STORE,
           SUM(IFF(METADATA$ACTION = 'INSERT', 1, -1)) AS N_DELTA,
           SUM(IFF(METADATA$ACTION = 'INSERT', AMOUNT, -AMOUNT)) AS AMOUNT_DELTA
    FROM ORDERS_STREAM
    GROUP BY ORDER_DATE, STORE
  ) S
  ON T.ORDER_DATE = S.ORDER_DATE AND T.STORE = S.STORE
  WHEN MATCHED THEN UPDATE SET
    ORDERS_N = T.ORDERS_N + S.N_DELTA,
    AMOUNT_SUM = T.AMOUNT_SUM + S.AMOUNT_DELTA,
    UPDATED_AT = CURRENT_TIMESTAMP()
  WHEN NOT MATCHED THEN INSERT (ORDER_DATE, STORE, ORDERS_N, AMOUNT_SUM, UPDATED_AT)
    VALUES (S.ORDER_DATE, S.STORE, S.N_DELTA, S.AMOUNT_DELTA, CURRENT_TIMESTAMP());
EXECUTE TASK REFRESH_DAILY_SALES;

USING の中では、件数の差を、INSERT の行は1、DELETE の行は−1 として足しています。金額の差も、DELETE の行は符号を逆にします。集計テーブルに同じ日と店の行があれば差を足し、無ければ新しい行として入れます。

SCHEDULE の USING CRON 15 6 * * * Asia/Tokyo は、日本時間の毎朝6時15分です。CRON の式は、分・時・日・月・曜日の順に5つの欄で書きます。WHEN の SYSTEM$STREAM_HAS_DATA は、ストリームに変更があるときだけ MERGE を実行する条件です。

試すときは、EXECUTE TASK でタスクを手動で1回実行します。22日分が入った状態での1回目は SUCCEEDED で、ストリームは0行になりました。続けて、空のストリームでもう一度実行し、タスクの実行履歴を見ました。画面の時刻は、日本時間に直しています。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
EXECUTE TASK REFRESH_DAILY_SALES;
SELECT NAME, STATE, ERROR_CODE, COALESCE(ERROR_MESSAGE, 'null') AS ERROR_MESSAGE,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', SCHEDULED_TIME), 'YYYY-MM-DD HH24:MI:SS') AS SCHEDULED_JST
FROM TABLE(INFORMATION_SCHEMA.TASK_HISTORY(TASK_NAME => 'REFRESH_DAILY_SALES',
       SCHEDULED_TIME_RANGE_START => DATEADD(HOUR, -2, CURRENT_TIMESTAMP())))
ORDER BY SCHEDULED_TIME DESC;

2回目の STATE は SKIPPED で、ERROR_CODE は 0040003 でした。ERROR_MESSAGE は Conditional expression for task evaluated to false. です。WHEN の条件が偽になり、MERGE は実行されていません。

SKIPPED の回はウェアハウスを起動しない、とドキュメントにあります。WHEN の判定はクラウドサービスという層で行われ、判定のたびにわずかな料金がたまります。そのため、1日1回しかデータが入らないテーブルを、1分ごとに判定する設定は勧められていません。クラウドサービスの料金が請求される条件は、料金とコストの記事で説明しています。

全件から集計し直した値と比べ、金額の直しと削除でも合うかを見る

足し方が合っているかは、注文テーブルを全件から集計し直した値と、集計テーブルを比べて見ます。試すときは、次の SELECT で、比べた日・店の数と、合わなかった数を1行に出しました。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
WITH FULL_CALC AS (
  SELECT ORDER_DATE, STORE, COUNT(*) AS ORDERS_N, SUM(AMOUNT) AS AMOUNT_SUM FROM ORDERS GROUP BY ORDER_DATE, STORE
), CMP AS (
  SELECT COALESCE(F.ORDER_DATE, D.ORDER_DATE) AS ORDER_DATE, COALESCE(F.STORE, D.STORE) AS STORE,
         EQUAL_NULL(F.ORDERS_N, D.ORDERS_N) AND EQUAL_NULL(F.AMOUNT_SUM, D.AMOUNT_SUM) AS SAME,
         COALESCE(F.ORDERS_N, 0) - COALESCE(D.ORDERS_N, 0) AS N_GAP,
         COALESCE(F.AMOUNT_SUM, 0) - COALESCE(D.AMOUNT_SUM, 0) AS AMOUNT_GAP
  FROM FULL_CALC F FULL OUTER JOIN DAILY_STORE_SALES D ON F.ORDER_DATE = D.ORDER_DATE AND F.STORE = D.STORE
)
SELECT COUNT(*) AS CELLS, COUNT_IF(SAME) AS SAME_CELLS, COUNT_IF(NOT SAME) AS DIFF_CELLS,
       (SELECT COUNT(*) FROM ORDERS_STREAM) AS STREAM_ROWS,
       (SELECT SUM(ORDERS_N) FROM DAILY_STORE_SALES) AS TABLE_ORDERS,
       (SELECT COUNT(*) FROM ORDERS) AS SOURCE_ORDERS,
       COALESCE(LISTAGG(IFF(SAME, NULL, TO_CHAR(ORDER_DATE) || ' ' || STORE || ' 件数差' || N_GAP || ' 金額差' || AMOUNT_GAP), ' / '), '') AS DIFF_DETAIL
FROM CMP;

FULL OUTER JOIN にしたのは、どちらか片方にしか無い日・店も、合わないものとして数えるためです。EQUAL_NULL は、両方が NULL のときも等しいとみなして比べる関数です。

CELLS は比べた日・店の数、SAME_CELLS は合った数、DIFF_CELLS は合わなかった数です。STREAM_ROWS は、ストリームに残っている行の数です。TABLE_ORDERS は集計テーブルの件数の合計で、SOURCE_ORDERS は注文テーブルの行の数です。DIFF_DETAIL には、合わない日・店と差が並びます。

22日分を足したあとは、8つの日・店がすべて合い、件数はどちらも380でした。

次に、金額の直しと削除を試しました。5行の金額に1,000ずつ足し、1行を消してから、ストリームの中身を数えています。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
UPDATE ORDERS SET AMOUNT = AMOUNT + 1000 WHERE ORDER_ID BETWEEN 1 AND 5;
DELETE FROM ORDERS WHERE ORDER_ID = 10;
SELECT METADATA$ACTION AS ACTION, METADATA$ISUPDATE AS ISUPDATE, COUNT(*) AS ROWS_N, SUM(AMOUNT) AS AMOUNT_SUM
FROM ORDERS_STREAM GROUP BY 1, 2 ORDER BY 1, 2;

消した1行は DELETE で、METADATA$ISUPDATE は FALSE でした。直した5行は、DELETE と INSERT が5行ずつで、どちらも TRUE です。このあと、上と同じ EXECUTE TASK の文でタスクを実行して比べると、8つとも合い、件数はどちらも379でした。

キャンセルした注文を数えないなど、集計に条件があるときは、その条件を差分の式の中に入れます。キャンセルの行の件数の差と金額の差を、IFF や CASE で0にするやり方です。注文がキャンセルに直されると、直す前の行は引かれ、直したあとの行は0で足されるので、その注文は集計から抜けるはずです。全件から集計し直す側にも、同じ条件を入れます。この書き方は試していないので、SQL は載せていません。

ある回に、ある日・店で変わった行の金額がすべて NULL だと、金額の差が NULL になります。そのとき集計テーブルの金額も NULL になるはずです。注文テーブルの金額が必ず入るかも、先に確かめておきます。

USING の書き方を誤ると、外した日を取りこぼし、まとめないと MERGE が止まった

取りこぼしは、USING に WHERE を付けたときに起きました。23日の分だけを先に集計に入れたい、という場面を考えて試しています。23日と24日の分を50行ずつ入れ、23日だけに絞った MERGE を手で実行しました。

-- 試すときの文。わざと USING に WHERE を付けて、取りこぼしを作った例
USE SCHEMA test_stream.raw;
INSERT INTO ORDERS
  SELECT SEQ4() + 401, DATEADD(DAY, FLOOR(SEQ4() / 50), '2026-10-23'::DATE), DECODE(MOD(SEQ4(), 2), 0, '渋谷店', '新宿店'),
         1000 + MOD((SEQ4() + 401) * 7919, 9000)
  FROM TABLE(GENERATOR(ROWCOUNT => 100));
MERGE INTO DAILY_STORE_SALES T
USING (
  SELECT ORDER_DATE, STORE,
         SUM(IFF(METADATA$ACTION = 'INSERT', 1, -1)) AS N_DELTA,
         SUM(IFF(METADATA$ACTION = 'INSERT', AMOUNT, -AMOUNT)) AS AMOUNT_DELTA
  FROM ORDERS_STREAM
  WHERE ORDER_DATE = '2026-10-23'
  GROUP BY ORDER_DATE, STORE
) S
ON T.ORDER_DATE = S.ORDER_DATE AND T.STORE = S.STORE
WHEN MATCHED THEN UPDATE SET ORDERS_N = T.ORDERS_N + S.N_DELTA, AMOUNT_SUM = T.AMOUNT_SUM + S.AMOUNT_DELTA, UPDATED_AT = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN INSERT (ORDER_DATE, STORE, ORDERS_N, AMOUNT_SUM, UPDATED_AT)
  VALUES (S.ORDER_DATE, S.STORE, S.N_DELTA, S.AMOUNT_DELTA, CURRENT_TIMESTAMP());

このあと突き合わせの SELECT を実行した結果が、次の画面です。

比べた12の日・店のうち、合わないのは2つで、24日の渋谷店と新宿店でした。集計テーブルの件数は429で、注文テーブルの479より50少ない数です。ストリームの行は0でした。

ストリームを使った DML の文は、コミットすると、ストリームの変更をすべて消費します。WHERE で外した24日の分も、この MERGE で消費されました。この50行は、ストリームからは二度と取り出せません。

23日の分だけを先に見たいなら、ストリームの変更はその回に全部を足し、集計テーブルを使う側で日を絞ります。同じ変更を別の処理でも使うなら、その処理のためのストリームを、同じ注文テーブルにもう1つ作ります。ドキュメントも、変更を使う処理ごとに別のストリームを作るよう勧めています。

取りこぼした日は、その日の分を注文テーブルから集計し直して入れ替えます。入れ替える前に、突き合わせの SELECT の STREAM_ROWS が0かを見ます。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
BEGIN;
DELETE FROM DAILY_STORE_SALES WHERE ORDER_DATE = '2026-10-24';
INSERT INTO DAILY_STORE_SALES
  SELECT ORDER_DATE, STORE, COUNT(*), SUM(AMOUNT), CURRENT_TIMESTAMP() FROM ORDERS WHERE ORDER_DATE = '2026-10-24' GROUP BY ORDER_DATE, STORE;
COMMIT;

その日の行を消してから入れ直すまでを、BEGIN と COMMIT で1つのトランザクションにしています。入れ替えたあとは、12の日・店がすべて合い、件数はどちらも479でした。

STREAM_ROWS が0のときだけ入れ替えるのは、当社の判断です。ストリームに変更が残ったまま入れ替えると、その変更は入れ替えた値にもう入っています。そこへ次のタスクが同じ差分を足すので、二重になるはずです。

次は、USING で日と店ごとにまとめない MERGE です。22日の渋谷店に2行を足し、ストリームの行をそのまま MERGE に渡しました。

-- 試すときの文。わざと日と店でまとめずに MERGE した例で、エラーになる
USE SCHEMA test_stream.raw;
INSERT INTO ORDERS VALUES (501, '2026-10-22', '渋谷店', 3000), (502, '2026-10-22', '渋谷店', 4000);
MERGE INTO DAILY_STORE_SALES T
USING (
  SELECT ORDER_DATE, STORE, IFF(METADATA$ACTION = 'INSERT', 1, -1) AS N_DELTA, IFF(METADATA$ACTION = 'INSERT', AMOUNT, -AMOUNT) AS AMOUNT_DELTA
  FROM ORDERS_STREAM
) S
ON T.ORDER_DATE = S.ORDER_DATE AND T.STORE = S.STORE
WHEN MATCHED THEN UPDATE SET ORDERS_N = T.ORDERS_N + S.N_DELTA, AMOUNT_SUM = T.AMOUNT_SUM + S.AMOUNT_DELTA, UPDATED_AT = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN INSERT (ORDER_DATE, STORE, ORDERS_N, AMOUNT_SUM, UPDATED_AT)
  VALUES (S.ORDER_DATE, S.STORE, S.N_DELTA, S.AMOUNT_DELTA, CURRENT_TIMESTAMP());

MERGE は、次のエラーで止まりました。Row Values の 20748 は、日付を数で出したものです。

Duplicate row detected during DML action Row Values: [20748, "渋谷店", 40, 224400, 1792514243471000000]

集計テーブルの1行に、ストリームの2行が当たったためです。パラメータ ERROR_ON_NONDETERMINISTIC_MERGE が既定の TRUE だと、こうした MERGE はエラーになります。

MERGE が失敗したので、ストリームの2行は消費されずに残りました。タスクの節と同じ EXECUTE TASK の文でタスクを実行すると、2行とも集計テーブルに入りました。

集計テーブルにまだ無い日・店では、まとめずに渡してもエラーにならず、渡した数だけ行が入ります。このときは MERGE が止まらないので、突き合わせで初めて差として出るはずです。USING では、いつも日と店でまとめます。

差がある日は、アラートからメールで知らせる

突き合わせを、毎朝のアラートにします。条件に使うのは、合わない日・店だけを出すビュー SALES_CHECK です。処理では、ストアドプロシージャ NOTIFY_SALES_DIFF を呼びます。ストアドプロシージャは、SQL で書いた処理をまとめて保存したものです。NOTIFY_SALES_DIFF は、差のある日・店を一覧にして、SYSTEM$SEND_EMAIL でメールを送ります。

メールの宛先にできるのは、同じアカウントの Snowflake ユーザーで、メールアドレスを確認済みの人だけです。送るには、メール用の通知インテグレーションも要ります。通知インテグレーションは、Snowflake から外へ知らせを送るための設定です。

宛先のユーザーと通知インテグレーション LOAD_EMAIL_INT は、取り込みの自動化の記事の手順で作ったものを使いました。その記事の片付けでは、この2つを最後に削除しています。そちらを最後まで試したなら、先に作り直しておきます。宛先の data-team@example.com は例の書き方で、自社では確認済みのユーザーのアドレスにします。

ビュー・ストアドプロシージャ・アラートは、後半の本番の一式の3・5・6と同じ文です。<…> の所を次の値にして、test_stream で作りました。ウェアハウスは COMPUTE_WH、アラートの時刻は毎朝6時45分、通知インテグレーションは LOAD_EMAIL_INT です。件名と本文の1行目には、テスト送信と分かる文を入れました。

作ったら、EXECUTE ALERT でアラートを手動で実行します。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
EXECUTE ALERT CHECK_SALES_DIFF;

1回目は、差が無かったので CONDITION_FALSE でした。条件の SQL が0行を返し、処理は動いていません。

次に、わざと差を作りました。25日分の40行を入れてから、取りこぼしを試したときと同じ、23日だけに絞った MERGE を実行しています。ストリームにあるのは25日分だけなので、40行とも足されずに消費されます。

-- 試すときの文。わざと差を作り、SALES_CHECK の中身を見る
USE SCHEMA test_stream.raw;
INSERT INTO ORDERS
  SELECT SEQ4() + 601, '2026-10-25'::DATE, DECODE(MOD(SEQ4(), 2), 0, '渋谷店', '新宿店'),
         1000 + MOD((SEQ4() + 601) * 7919, 9000)
  FROM TABLE(GENERATOR(ROWCOUNT => 40));
MERGE INTO DAILY_STORE_SALES T
USING (
  SELECT ORDER_DATE, STORE,
         SUM(IFF(METADATA$ACTION = 'INSERT', 1, -1)) AS N_DELTA,
         SUM(IFF(METADATA$ACTION = 'INSERT', AMOUNT, -AMOUNT)) AS AMOUNT_DELTA
  FROM ORDERS_STREAM
  WHERE ORDER_DATE = '2026-10-23'
  GROUP BY ORDER_DATE, STORE
) S
ON T.ORDER_DATE = S.ORDER_DATE AND T.STORE = S.STORE
WHEN MATCHED THEN UPDATE SET ORDERS_N = T.ORDERS_N + S.N_DELTA, AMOUNT_SUM = T.AMOUNT_SUM + S.AMOUNT_DELTA, UPDATED_AT = CURRENT_TIMESTAMP()
WHEN NOT MATCHED THEN INSERT (ORDER_DATE, STORE, ORDERS_N, AMOUNT_SUM, UPDATED_AT)
  VALUES (S.ORDER_DATE, S.STORE, S.N_DELTA, S.AMOUNT_DELTA, CURRENT_TIMESTAMP());
SELECT * FROM SALES_CHECK;

SALES_CHECK には、25日の渋谷店と新宿店が、20件ずつの差で出ました。金額の差は、渋谷店が115,600、新宿店が102,980です。

ここで、上と同じ EXECUTE ALERT の文をもう一度実行しました。このときのストアドプロシージャは最初に作った版で、LISTAGG の区切りを CHR(10) にしていました。一式の5の直した形で作っていれば、ここで TRIGGERED になるはずです。

この版での2回目は、ACTION_FAILED でした。条件は成り立ったものの、処理の SQL が失敗したという意味です。エラー文は、アラートの実行履歴の SQL_ERROR_MESSAGE にあります。長いので、次の SELECT で100字ずつに分けて出しました。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
SELECT STATE, SQL_ERROR_CODE, G.VALUE::INT AS PART, SUBSTR(SQL_ERROR_MESSAGE, G.VALUE::INT * 100 + 1, 100) AS TEXT
FROM TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(SCHEDULED_TIME_RANGE_START => DATEADD(HOUR, -1, CURRENT_TIMESTAMP()))) H,
     TABLE(FLATTEN(ARRAY_GENERATE_RANGE(0, CEIL(LENGTH(H.SQL_ERROR_MESSAGE) / 100)::INT))) G
WHERE NAME = 'CHECK_SALES_DIFF' AND STATE = 'ACTION_FAILED'
ORDER BY PART;

SQL_ERROR_CODE は1015でした。エラー文の終わりは、to function LISTAGG needs to be constant, found 'CHR(10)' です。LISTAGG の区切りには、定数を書くことになっています。CHR(10) は関数なので、定数ではありません。

区切りを、引用符のあいだで改行しただけの文字列に直して、ストアドプロシージャを作り直しました。本番の一式の5は、この直した形です。本文の行をつなぐ所の CHR(10) は、そのままで動いています。同じ EXECUTE ALERT の文で3回目を実行すると、TRIGGERED でした。

アラートの実行履歴と送信の記録は、次の SELECT で並べて見ました。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
SELECT 'ALERT' AS SRC, NAME AS VALUE1, STATE AS VALUE2,
       TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', SCHEDULED_TIME), 'YYYY-MM-DD HH24:MI:SS') AS AT_JST
FROM TABLE(INFORMATION_SCHEMA.ALERT_HISTORY(SCHEDULED_TIME_RANGE_START => DATEADD(HOUR, -1, CURRENT_TIMESTAMP())))
WHERE NAME = 'CHECK_SALES_DIFF'
UNION ALL
SELECT 'MAIL', '(送信の記録)', STATUS, TO_CHAR(CONVERT_TIMEZONE('Asia/Tokyo', CREATED), 'YYYY-MM-DD HH24:MI:SS')
FROM TABLE(INFORMATION_SCHEMA.NOTIFICATION_HISTORY(START_TIME => DATEADD(MINUTE, -30, CURRENT_TIMESTAMP())))
ORDER BY AT_JST DESC;

送信の記録 NOTIFICATION_HISTORY の STATUS は、SUCCESS でした。SUCCESS は、Snowflake の側で送ったという記録です。この回のメールが受信箱に届いたかは、確かめていません。同じ通知インテグレーションから別の検証で送ったメールは、受信箱に届いています。

ストアドプロシージャもアラートも、作るときには、この誤りでエラーになりませんでした。差が無い日のアラートは処理を動かさないので、CONDITION_FALSE だけを見ても、処理の誤りには気づけません。作ったら、検証用のデータベースで差を作り、TRIGGERED になるまで確かめます。

アラートは、続けて失敗しても自動では一時停止しません。処理が失敗した日は、メールも来ません。ウェアハウスで動かすアラートは、メールを送るだけでも、少なくとも1分ぶんのウェアハウスの料金がかかります。条件の SQL が全件を集計する時間も、その料金に入ります。

Dynamic Tables で同じ集計を作ると、SELECT 1本で注文テーブルに追いついた

同じ集計を、Dynamic Tables でも作りました。Dynamic Tables は、日本語版のドキュメントでは動的テーブルと呼ばれています。SELECT の定義と TARGET_LAG を決めておくと、Snowflake がその結果を新しく保つテーブルです。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
CREATE OR REPLACE DYNAMIC TABLE DAILY_STORE_SALES_DT
  TARGET_LAG = '1 minute'
  WAREHOUSE = COMPUTE_WH
  AS
  SELECT ORDER_DATE, STORE, COUNT(*) AS ORDERS_N, SUM(AMOUNT) AS AMOUNT_SUM
  FROM ORDERS
  GROUP BY ORDER_DATE, STORE;
SHOW DYNAMIC TABLES LIKE 'DAILY_STORE_SALES_DT' IN SCHEMA test_stream.raw;
SELECT "name", "target_lag", "refresh_mode", "refresh_mode_reason", "scheduling_state" FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()));

今回は REFRESH_MODE を書かずに作りました。SHOW DYNAMIC TABLES で見ると、refresh_mode は INCREMENTAL で、差分で更新する方式に決まっていました。書かないときの既定は AUTO で、方式は作るときに一度だけ決まります。本番では、ADAPTIVE・FULL・INCREMENTAL のどれかを明示するよう、ドキュメントは勧めています。

差分で更新する集計には ADAPTIVE を第一に勧め、取り込みのたびにほぼ全部の行が入れ替わるなら FULL が向く、とドキュメントにあります。後半の本番の形には、今回決まった INCREMENTAL を書きました。この行を書いて作ったことは、まだありません。

SELECT に CURRENT_TIMESTAMP() を書くと、差分では更新できません。AUTO なら全件の更新に決まり、INCREMENTAL と書けば作る時点でエラーになります。そのため、集計テーブルにある UPDATED_AT の列は、Dynamic Table には入れていません。

26日分の60行を入れて、注文テーブルと比べました。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
INSERT INTO ORDERS
  SELECT SEQ4() + 701, '2026-10-26'::DATE, DECODE(MOD(SEQ4(), 2), 0, '渋谷店', '新宿店'),
         1000 + MOD((SEQ4() + 701) * 7919, 9000)
  FROM TABLE(GENERATOR(ROWCOUNT => 60));
SELECT 'DT' AS SRC, COUNT(*) AS CELLS, SUM(ORDERS_N) AS ORDERS_TOTAL, SUM(AMOUNT_SUM) AS AMOUNT_TOTAL FROM DAILY_STORE_SALES_DT
UNION ALL
SELECT 'ORDERS', COUNT(DISTINCT ORDER_DATE || STORE), COUNT(*), SUM(AMOUNT) FROM ORDERS;

入れた直後の Dynamic Table は、14の日・店で521件でした。注文テーブルの16の日・店・581件より、古いままです。約1分20秒後には16の日・店・581件になり、金額の合計も3,207,520でそろいました。

更新の履歴では、変更が無い回は NO_DATA でした。変更が無ければウェアハウスは止まったまま、とドキュメントにあります。

TARGET_LAG は、更新の間隔ではなく、データがどれだけ古くてよいかの目安です。最小は1分で、必ず守られるわけではありません。今回の1分は試すための値で、TARGET_LAG を短くするほど、変更を確かめる回数が増えます。

Dynamic Table は読み取り専用で、INSERT・UPDATE・DELETE はできません。CRON で時刻を決めることもできず、決めた時刻に更新するなら手動の REFRESH を使います。定義に、ストアドプロシージャも書けません。

試し終えた Dynamic Table は、次の stale の SQL の1行目で一時停止します。

どちらで作るかは、表の「こういうとき」に当てはまるかで決めます。

注文テーブルを CREATE OR REPLACE で作り直すと、ストリームは使えなくなる

ストリームは、元のテーブルの変更の履歴を使って差分を返します。元のテーブルの履歴が無くなると、ストリームは使えなくなります。この状態を stale と言い、日本語版のドキュメントでは「古くなる」と訳されています。状態は、SHOW STREAMS の stale と stale_after の列で見られます。

-- 試すときの文（検証用のデータベース test_stream で実行する）
USE SCHEMA test_stream.raw;
ALTER DYNAMIC TABLE DAILY_STORE_SALES_DT SUSPEND;
SHOW STREAMS LIKE 'ORDERS_STREAM' IN SCHEMA test_stream.raw;
SELECT 'BEFORE' AS WHEN_, "name", "stale", "stale_after", "mode" FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()));

作り直す前は、stale が false でした。stale_after には、最後にストリームを消費してから約14日後の日時が出ていました。

次に、注文テーブルを CREATE OR REPLACE TABLE で作り直しました。中身の行は同じです。

-- 試すときの文。わざと注文テーブルを作り直した例
USE SCHEMA test_stream.raw;
CREATE OR REPLACE TABLE ORDERS AS SELECT * FROM ORDERS;
SHOW STREAMS LIKE 'ORDERS_STREAM' IN SCHEMA test_stream.raw;
SELECT 'AFTER' AS WHEN_, "name", "stale", "stale_after", "mode" FROM TABLE(RESULT_SCAN(LAST_QUERY_ID()));

stale は true に変わり、stale_after は null になりました。SELECT COUNT(*) FROM ORDERS_STREAM; でストリームの中身を見ると、次のエラーになりました。データベース名は、試したときの名前なので <DB> に置き換えています。

Base table '<DB>.RAW.ORDERS' dropped, cannot read from stream 'ORDERS_STREAM' in line 1 position 21.

CREATE OR REPLACE TABLE で作り直すと、テーブルの履歴が消え、その上のストリームは stale になります。stale になったストリームは作り直すしかなく、まだ消費していなかった変更も取り出せなくなります。そこで、本番の一式の1を、使っている集計テーブルの名前のままもう一度実行し、集計テーブルとストリームを作り直します。stale のときは、捨てることになる変更がもう取り出せないので、1をもう一度実行してかまいません。毎朝の取り込みは、注文テーブルを作り直さず、行を足すやり方にします。

消費しないまま置いておいても、ストリームは stale になります。既定の設定では、最後に消費してから約14日です。テーブルのデータ保持期間が14日より短くても、Snowflake が最大14日まで延ばすためです。データ保持期間については、Time Travel と UNDROP の記事に書きました。

SYSTEM$STREAM_HAS_DATA を呼んだときも、ストリームが空で FALSE が返れば、stale になるのを防げます。毎朝タスクが動いていれば、注文の来ない日が続いても stale にはならないはずです。タスクを一時停止したまま2週間ほど置くと、stale になるおそれがあります。

本番に入れる一式と、差のメールが来た日・定義を直すときの手順

本番に入れる SQL を、ここに1つにまとめました。試すための注文の INSERT、わざと差を作る MERGE、EXECUTE TASK と EXECUTE ALERT は入れていません。名前・宛先・件名・時刻は、<…> の形にしています。

各文は、試したときに実行したものと同じ形です。通知インテグレーションを作る文だけは、取り込みの自動化の検証で実行しました。このまとまりで通して実行したことはなく、最後の RESUME と SHOW も実行していません。

-- 本番に入れる一式。この形のまとまりでは実行していない
-- <…> の所を、自社の名前・宛先・件名・時刻に書き換える
USE SCHEMA <データベース>.<スキーマ>;

-- 1. 集計テーブルに全件の値を入れてから、ストリームを作る
CREATE OR REPLACE TABLE DAILY_STORE_SALES (ORDER_DATE DATE, STORE STRING, ORDERS_N NUMBER, AMOUNT_SUM NUMBER, UPDATED_AT TIMESTAMP_LTZ);
INSERT INTO DAILY_STORE_SALES
  SELECT ORDER_DATE, STORE, COUNT(*), SUM(AMOUNT), CURRENT_TIMESTAMP() FROM ORDERS GROUP BY ORDER_DATE, STORE;
CREATE OR REPLACE STREAM ORDERS_STREAM ON TABLE ORDERS;

-- 2. 差分を日と店でまとめて足すタスク（毎朝の取り込みが終わったあとの時刻に）
CREATE OR REPLACE TASK REFRESH_DAILY_SALES
  WAREHOUSE = <ウェアハウス>
  SCHEDULE = 'USING CRON <分> <時> * * * Asia/Tokyo'
  WHEN SYSTEM$STREAM_HAS_DATA('ORDERS_STREAM')
  AS
  MERGE INTO DAILY_STORE_SALES T
  USING (
    SELECT ORDER_DATE, STORE,
           SUM(IFF(METADATA$ACTION = 'INSERT', 1, -1)) AS N_DELTA,
           SUM(IFF(METADATA$ACTION = 'INSERT', AMOUNT, -AMOUNT)) AS AMOUNT_DELTA
    FROM ORDERS_STREAM
    GROUP BY ORDER_DATE, STORE
  ) S
  ON T.ORDER_DATE = S.ORDER_DATE AND T.STORE = S.STORE
  WHEN MATCHED THEN UPDATE SET
    ORDERS_N = T.ORDERS_N + S.N_DELTA,
    AMOUNT_SUM = T.AMOUNT_SUM + S.AMOUNT_DELTA,
    UPDATED_AT = CURRENT_TIMESTAMP()
  WHEN NOT MATCHED THEN INSERT (ORDER_DATE, STORE, ORDERS_N, AMOUNT_SUM, UPDATED_AT)
    VALUES (S.ORDER_DATE, S.STORE, S.N_DELTA, S.AMOUNT_DELTA, CURRENT_TIMESTAMP());

-- 3. 全件から集計し直した値と合わない日・店だけを出すビュー
CREATE OR REPLACE VIEW SALES_CHECK AS
WITH FULL_CALC AS (
  SELECT ORDER_DATE, STORE, COUNT(*) AS ORDERS_N, SUM(AMOUNT) AS AMOUNT_SUM FROM ORDERS GROUP BY ORDER_DATE, STORE
)
SELECT COALESCE(F.ORDER_DATE, D.ORDER_DATE) AS ORDER_DATE, COALESCE(F.STORE, D.STORE) AS STORE,
       COALESCE(F.ORDERS_N, 0) - COALESCE(D.ORDERS_N, 0) AS N_GAP,
       COALESCE(F.AMOUNT_SUM, 0) - COALESCE(D.AMOUNT_SUM, 0) AS AMOUNT_GAP
FROM FULL_CALC F FULL OUTER JOIN DAILY_STORE_SALES D ON F.ORDER_DATE = D.ORDER_DATE AND F.STORE = D.STORE
WHERE NOT (EQUAL_NULL(F.ORDERS_N, D.ORDERS_N) AND EQUAL_NULL(F.AMOUNT_SUM, D.AMOUNT_SUM));

-- 4. メール用の通知インテグレーション（宛先は、同じアカウントのメールアドレスを確認済みのユーザー）
CREATE NOTIFICATION INTEGRATION IF NOT EXISTS <通知インテグレーション>
  TYPE = EMAIL
  ENABLED = TRUE
  ALLOWED_RECIPIENTS = ('<宛先のアドレス>');

-- 5. 差のある日・店を一覧にしてメールで送るストアドプロシージャ
--    LISTAGG の区切りは CHR(10) ではなく、引用符のあいだで改行した文字列にする
CREATE OR REPLACE PROCEDURE NOTIFY_SALES_DIFF()
RETURNS STRING
LANGUAGE SQL
AS
$$
DECLARE
  diff_cells NUMBER;
  detail STRING;
  body STRING;
BEGIN
  SELECT COUNT(*), LISTAGG(TO_CHAR(ORDER_DATE) || ' ' || STORE || ' 件数の差 ' || N_GAP || ' 金額の差 ' || AMOUNT_GAP, '
')
    INTO :diff_cells, :detail
    FROM SALES_CHECK;
  body := '<本文の1行目>' || CHR(10) ||
          '日別・店別の集計テーブルと、注文テーブルから集計し直した値に差があります。' || CHR(10) ||
          '差のある日と店: ' || diff_cells || CHR(10) || COALESCE(detail, '');
  CALL SYSTEM$SEND_EMAIL(
    '<通知インテグレーション>',
    '<宛先のアドレス>',
    '<件名>',
    :body);
  RETURN diff_cells;
END;
$$;

-- 6. 毎朝ビューを見て、行があればメールを送るアラート（タスクが終わったあとの時刻に）
CREATE OR REPLACE ALERT CHECK_SALES_DIFF
  WAREHOUSE = <ウェアハウス>
  SCHEDULE = 'USING CRON <分> <時> * * * Asia/Tokyo'
  IF (EXISTS (SELECT 1 FROM SALES_CHECK))
  THEN
    CALL NOTIFY_SALES_DIFF();

-- 7. 動かし始める。2・6 を作り直したときも、ここをもう一度実行する（RESUME と SHOW は今回は実行していない）
ALTER TASK REFRESH_DAILY_SALES RESUME;
ALTER ALERT CHECK_SALES_DIFF RESUME;
SHOW TASKS LIKE 'REFRESH_DAILY_SALES';
SHOW ALERTS LIKE 'CHECK_SALES_DIFF';

書き換える所は、次の表のとおりです。番号は、上の一式のコメントの番号です。

1は、注文テーブルに行が入らない時間に、続けて実行します。最初の INSERT とストリームの作成のあいだに入った行は、集計にもストリームにも入らないはずだからです。もしそうなっても、突き合わせで差として出ます。

1は最初の1回だけ実行します。もう一度実行すると、集計テーブルとストリームが作り直され、まだ消費していない変更は捨てられます。ストリームが stale になったときだけは、例外として1をもう一度実行します。

4は IF NOT EXISTS なので、同じ名前の通知インテグレーションがすでにあれば、何も変わりません。ALLOWED_RECIPIENTS も前の宛先のままなので、宛先を変えたいなら、別の名前で作ります。

タスクやアラートの定義を直すときは、RESUME し直すところまでが手順です。CREATE OR REPLACE TASK で作り直したタスクは、一時停止の状態から始まり、次の予定の回も取り消されます。アラートも、作ったときは一時停止の状態です。CREATE OR REPLACE は古いものを消して新しく作る文なので、作り直したアラートも同じはずです。

直したら、一式の7をもう一度実行します。SHOW TASKS と SHOW ALERTS の state が、suspended のままになっていないかを見ます。時刻だけを変えるなら、ALTER TASK … SET SCHEDULE もあります。ほかのタスクとつないでいない単独のタスクは、先に SUSPEND しておく必要があります。

ACCOUNTADMIN 以外のロールで組むときは、作るものごとに次の権限が要ります。今回は ACCOUNTADMIN で試しました。

• ストリーム：スキーマの CREATE STREAM と、注文テーブルの SELECT が要ります。最初のストリームを作ると、注文テーブルで変更の追跡が有効になります。まだ有効でなければ、最初のストリームを作れるのは、テーブルを所有するロールだけです。
• タスク：スキーマの CREATE TASK と、ウェアハウスの USAGE が要ります。RESUME するとき、所有するロールにアカウントの EXECUTE TASK があるかを Snowflake が確かめます。タスクは所有するロールの権限で動くので、ストリームの SELECT と、集計テーブルを更新する権限も要るはずです。
• アラート：アカウントの EXECUTE ALERT と、スキーマの USAGE と CREATE ALERT が要ります。ウェアハウスの USAGE、条件で見る SALES_CHECK の SELECT も要ります。EXECUTE ALERT を付与できるのは、ACCOUNTADMIN のロールだけです。
• 通知インテグレーション：作るには、アカウントの権限が要ります。CREATE INTEGRATION か CREATE NOTIFICATION INTEGRATION のどちらかです。ストアドプロシージャは既定で所有するロールの権限で動くので、そのロールに通知インテグレーションの USAGE を付与します。

動かし始めた週は、タスクの実行履歴に、毎朝 SUCCEEDED か SKIPPED の行があるかを見ます。アラートの実行履歴には、CONDITION_FALSE か TRIGGERED の行があるかを見ます。

差のメールが来るのは、足し方を誤った日だけではありません。タスクが失敗した日や、6時15分のあとに注文が入った日・直された日もあります。ストリームに変更が残ったまま比べるので、差として出るはずです。

その日は、記事の最初の図の手順のとおり、まず突き合わせの SELECT で STREAM_ROWS を見ます。0でなければ EXECUTE TASK でタスクを実行し、もう一度比べます。0なのに差が残る日・店だけを、取りこぼしを直したときの SQL で、日付を書き換えて入れ替えます。タスクを実行してから比べ直す手順は、通しでは試していません。

取り込みを朝のうちにやり直した日も、そのあとタスクを実行すれば、その日のうちに集計テーブルへ入るはずです。ある日・ある店の注文がすべて消えた日は、集計テーブルに件数0・金額0の行が残るはずです。全件の集計にはその日・店の行が無いので、突き合わせでは合わない行として出て、メールには「件数の差 0 金額の差 0」と載ります。その日の分を入れ替える SQL を実行すれば、この行は消えるはずです。

Dynamic Tables を選ぶなら、本番の形は次のとおりです。

-- Dynamic Tables を選ぶ場合の本番の形。この形のまとまりでは実行していない
USE SCHEMA <データベース>.<スキーマ>;
-- REFRESH_MODE の行は、今回は書かずに作り INCREMENTAL に決まった。この行は実行していない
CREATE OR REPLACE DYNAMIC TABLE DAILY_STORE_SALES_DT
  TARGET_LAG = '<どれだけ古くてよいか>'
  WAREHOUSE = <ウェアハウス>
  REFRESH_MODE = INCREMENTAL
  AS
  SELECT ORDER_DATE, STORE, COUNT(*) AS ORDERS_N, SUM(AMOUNT) AS AMOUNT_SUM
  FROM ORDERS
  GROUP BY ORDER_DATE, STORE;
SHOW DYNAMIC TABLES LIKE 'DAILY_STORE_SALES_DT' IN SCHEMA <データベース>.<スキーマ>;
-- 止めるときと、また動かすとき（RESUME は今回は実行していない）
ALTER DYNAMIC TABLE DAILY_STORE_SALES_DT SUSPEND;
ALTER DYNAMIC TABLE DAILY_STORE_SALES_DT RESUME;

TARGET_LAG は、集計を使う人がどれだけ古い数を許せるかで決めます。あとで定義を CREATE OR REPLACE で変えると、全件から作り直しになります。

検証用の test_stream を片付けたら、いま差分で足している集計テーブルを一度、全件と比べる

片付けの SQL は、検証用の test_stream にだけ使い、本番の名前に書き換えません。タスク・アラート・Dynamic Table を一時停止してから削除します。

-- 片付け。検証用のデータベース test_stream だけを消す
USE ROLE ACCOUNTADMIN;
USE SCHEMA test_stream.raw;
ALTER TASK test_stream.raw.REFRESH_DAILY_SALES SUSPEND;
ALTER ALERT test_stream.raw.CHECK_SALES_DIFF SUSPEND;
ALTER DYNAMIC TABLE DAILY_STORE_SALES_DT SUSPEND;
ALTER DATABASE test_stream SET DATA_RETENTION_TIME_IN_DAYS = 0;
DROP DATABASE test_stream;

データ保持期間を0にしてから削除したのは、削除したあとに Time Travel で戻せる期間を残さないためです。宛先のユーザーと通知インテグレーションはアカウントのものなので、データベースを削除しても残ります。

試したのは2026年10月21日の深夜で、ウェアハウスは X-Small の1つだけを使っています。試用のアカウントは Enterprise Edition です。ストリーム・タスク・Dynamic Tables は、どれも Standard Edition から使えるとドキュメントにあります。

今回試していないこともあります。RESUME して予定の時刻に動くところは見ておらず、タスクもアラートも手動で実行しただけです。APPEND_ONLY のストリームと、1つの注文テーブルに複数のストリームを作る形も試していません。BEGIN と COMMIT の中でストリームを使う形と、14日たって stale になるところも見ていません。Dynamic Table の全件の更新と、TARGET_LAG を長くした場合も見ていません。料金も測っていません。本文で「はずです」と書いた所は、どれも試していません。

すでに差分で集計テーブルを更新しているなら、まず一度、注文テーブルから全件で集計し直した値と比べます。突き合わせの SELECT の、テーブルと列の名前を書き換えれば使えます。ストリームを使っていないなら、STREAM_ROWS の行は外します。DIFF_CELLS が0で、件数の合計がそろっていれば、その時点の数は合っています。差が出たら、DIFF_DETAIL の日・店から、USING の条件やまとめ方を見直します。

差の原因の切り分けや、Dynamic Tables への移し替えまで手が回らないときは、ご相談ください。増分ロードと更新頻度の設計は、当社が Snowflake 導入支援で受けている内容です。