Snowflake

Snowflake Connectorは、Snowflakeのデータベースからテーブル、ビュー、ステージのファイルを読み、レコードとテーブルを書き戻し、ステージにコンテンツをロードし、プロシージャとタスクを実行します。解約リスクの例のDWHがこれです。利用データはここにあり、Connectorは重い部分の処理（絞り込みと集計）をSnowflakeに任せたうえで、レコードをエンジンに届けます。

読み取り元Copy Anchor Link

設定したデータベースとスキーマのすべてのテーブルとビューが読み取り元です。snowflake.usage_30d、snowflake.events、snowflake.accounts_dim のように参照します。列と型はSnowflakeの情報スキーマから取ります。テーブルのキーは、宣言されている PRIMARY KEY か UNIQUE 制約です。Snowflakeは通常のテーブルでこれらの制約を強制しないので、Connectorは事実ではなく検証すべき主張として扱います（Effectと操作を参照）。

内部ステージと外部ステージのファイルも読み取り元です。LIST_FILES(snowflake, glob => r'@exports/*.csv') が resource、version（ファイルのMD5）、path で一覧を返し、resource と version でJOINした snowflake.contents がファイルを content_ref として読みます。

別のスキーマを読むには、プログラムの中で名前を修飾するのではなく、snowflake_raw のような2つ目の接続を設定してください。

型Copy Anchor Link

書き込みは同じ規則で逆方向に変換します。対象の列に収まらない値（精度、長さ、型）は、静的に分かるなら check で、そうでなければその項目が失敗します。黙って切り詰めることはありません。

Effectと操作Copy Anchor Link

レコードのEffectはキーで書きます。

• insert — EFFECT INSERT snowflake.account_scores (col, …)。
• update — EFFECT UPDATE snowflake.account_scores ID account_id SET (…)。
• upsert — EFFECT UPSERT snowflake.account_scores ID account_id SET (…)。MERGE として適用します。
• delete — EFFECT DELETE snowflake.account_scores ID account_id。

ID で指定するキーは、ちょうど1レコードを識別しなければなりません。どの列がキーになるかは、そのテーブルに対するConnectorのMappingが宣言します。Snowflakeは一意性を強制しないので、Connectorはプレビューを作るときに検証し、複数のレコードに一致するキーの項目は failed になります。単一レコードの変更として適用されることはありません。同じキーに対して元のレコードが2つの異なる値を持つ更新は、check で拒否されます。

テーブルとコンテンツ：

• create table — |> CREATE [OR REPLACE] TABLE snowflake.active_usage は、レコードをテーブル全体として書きます。
• store — EFFECT STORE snowflake.stages SET (stage = '@exports', path = …, content = content) は、レコードの content（SERIALIZE または contents の読み取り元から来たもの）をステージに置きます。

操作：

• call — EFFECT COMMAND snowflake.call ID key INPUT (procedure = 'ANALYTICS.REBUILD', args = [...]) は、ストアドプロシージャを呼びます。
• execute_task — EFFECT COMMAND snowflake.execute_task TARGET task_name ID key は、タスクを1回実行します。

論理的な項目と物理的な文は別のものです。Connectorは100件の UPSERT の項目を1つの MERGE で適用してもかまいませんが、各項目は自分の識別子、プレビューの行、結果を持ち続けます。

セットアップCopy Anchor Link

"snowflake": {
  "connector": "snowflake",
  "account": "acme-eu",
  "warehouse": "ANALYTICS",
  "database": "PRODUCT",
  "schema": "USAGE",
  "role": "ACTQL_READER",
  "user": "ACTQL",
  "credentials": {"private_key": {"env": "SNOWFLAKE_PRIVATE_KEY"}}
}

既定はキーペア認証です。パスワードも同じ形で渡せます。ロールには、ウェアハウス、データベース、スキーマへの USAGE と、プログラムが読むテーブルへの SELECT が必要です。書き込みには、同じロールに INSERT、UPDATE、DELETE、CREATE TABLE とステージの権限を与えるか、書き込み用のロールを持つ2つ目の接続（snowflake_write、ロール ACTQL_WRITER）を設定してEffectをそこに向けます。後者なら、ポリシーが2つを区別できます。

ActQL Cloudでも接続の項目は同じです。秘密鍵は環境変数ではなくCloudのシークレットストアに保存します。

CapabilityとプッシュダウンCopy Anchor Link

explain は、プッシュダウンされた各ステップをSite connector:snowflake、Engine Snowflake SQL で表示します。ActQLの意味を保ったままSnowflakeで実行できないものはHostで実行し、CloudやリモートのHostはそのためにDuckDBを選ぶことがあります。

プレビューと突き合わせCopy Anchor Link

プレビュー。 更新、UPSERT、削除は full_diff でプレビューします。Connectorは対象のレコードをキーで読み、現在の値を記録し、プレビューにはレコードごとに現在の値と新しい値が出ます。Snowflakeは READ COMMITTED で動くので、先に読むだけでは足りません。Connectorは各レコードの METADATA$ROW_ID と現在の値のハッシュも記録し、送る MERGE や UPDATE をそれらの条件付きにします。プレビューと適用の間に変わったレコードは更新されません。その項目は failed（conflict）で終わり、続けるよう求められればRunは新しいプレビューを作ります。追加とテーブル作成は targeted（何をどこに書くか）でプレビューし、CREATE OR REPLACE は置き換えられる件数をプレビューします。

突き合わせ。 送信前に、ConnectorはEffectの項目を、実行する文に結び付けます。自分で生成するSQL APIの requestId と、分かった時点で statementHandle とクエリIDです。応答が失われた場合は、その文をSQL APIと QUERY_HISTORY でIDによって探します。履歴にある成功した文でも、Effectの requestId を持たないものは証跡になりません。バッチを明示的なトランザクションで実行する場合は、文の成功とコミットの成功を別々に確認します。コミットされた MERGE は、どの項目に影響したかを報告します。非同期の文は、最終状態になるまでハンドルで追跡します。これらで結果が確定しなければ、項目は unknown のままで、何も再送しません。

上限と費用Copy Anchor Link

• プッシュダウンされたクエリは設定したウェアハウスで動き、そこのほかのクエリと同じように課金されます。explain はウェアハウスの名前を示します。1回の読み取りが返せる量は、プログラムの件数とバイト数の上限で抑えられます。
• 結果はSQL APIのパーティション単位で取得します。上限に達したConnectorは、途中までの結果を返すのではなく、エラーで止まります。
• 情報スキーマの QUERY_HISTORY は7日間を対象とし、アカウント利用状況のビューは最大45分遅れます。突き合わせはまず文のハンドルを使い、それらの期間内で履歴に切り替えます。
• 大文字小文字：Snowflakeは引用符のない識別子を大文字にします。リソース名と列名は、Snowflakeに表示されるとおりにプログラムに書いてください。引用はConnectorが行います。

例Copy Anchor Link

直近30日に利用が落ちた取引先を、セグメント別に数えます。

FROM snowflake.usage_30d
|> WHERE trend = 'down'
|> AGGREGATE COUNT(*) AS accounts, AVG(active_users) AS avg_users
     GROUP BY segment
|> ORDER BY accounts DESC;

LLMの判定でスコアのテーブルを最新に保ちます。

SELECT a.id AS account_id, a.description AS notes, u.active_users, u.trend
FROM salesforce.accounts AS a
JOIN snowflake.usage_30d AS u ON a.id = u.account_id
|> INFER `model-task:churn-risk.v1`
     INPUT (notes, active_users, trend) AS risk
|> WHERE risk.outcome = 'ok'
|> SELECT account_id, risk.value.level AS level, risk.value.reason AS reason
|> EFFECT UPSERT snowflake.account_scores ID account_id SET (level = level, reason = reason);

DWHの中のデータからレポート用のテーブルを作り直します。

FROM snowflake.events
|> WHERE occurred_at >= DATE_SUB(CURRENT_DATE(), INTERVAL 30 DAY)
|> AGGREGATE COUNT(*) AS events GROUP BY account_id
|> CREATE OR REPLACE TABLE snowflake.events_30d;