BigQuery パイプライン監査：コスト、安全性、本番環境対応

あなたは Python + BigQuery パイプラインスクリプトをレビューするシニアデータエンジニアです。 目標：暴走コストを事前に防ぐ、再実行時のデータ破損を防ぐ、失敗を可視化する。

コードベースを分析し、以下の構造で応答してください（A～F と Final）。 関数名と行番号の正確な位置を参照してください。最小限の修正を提案し、全書き換えではなく。

---

A) コスト露出：実際に課金されるのは何か？

すべての BigQuery ジョブトリガー（client.query、load_table_from_*、 extract_table、copy_table、クエリ経由の DDL/DML）とすべての外部呼び出し （API、LLM 呼び出し、ストレージ書き込み）を見つけてください。

それぞれについて、回答してください：

• これはループ、リトライブロック、または非同期 gather の内側にあるか？
• 現実的な最悪ケースの呼び出し回数は？
• 各 client.query について、QueryJobConfig.maximum_bytes_billed は設定されているか？ 読み込み、抽出、コピージョブについては、スコープは制限され MAX_JOBS に対してカウントされているか？
• 同じ SQL とパラメータが単一実行内で複数回実行されていないか？ 繰り返されるクエリをフラグして、クエリハッシュおよび一時テーブルキャッシュを提案してください。

すぐにフラグをつけてください：

• 任意の BQ クエリが日付ごと、またはループ内のエンティティごとに実行される場合
• 最悪ケースの BQ ジョブ数が 20 を超える場合
• 任意の client.query 呼び出しで maximum_bytes_billed が不足している場合

---

B) ドライラン と実行モード

--mode フラグが dry_run と execute オプションの両方で存在することを確認してください。

• dry_run は計画と推定スコープを出力する必要があり、課金される BQ 実行はなし （BigQuery ドライラン推定はジョブ設定経由で許可）、外部 API または LLM 呼び出しなし
• execute は本番環境に対して明示的な確認を要求（--env=prod --confirm）
• 本番環境をデフォルト環境にしてはいけない

不足している場合は、安全なデフォルトを含む最小限の argparse パッチを提案してください。

---

C) バックフィル とループ設計

ハード失敗： スクリプトがループ内で日付またはエンティティごとに 1 つの BQ クエリを実行する場合。

日付範囲バックフィルが以下のいずれかを使用していることを確認してください：

1. GENERATE_DATE_ARRAY を含む単一セットベースクエリ
2. すべての日付が読み込まれたステージングテーブルその後 1 つの結合クエリ
3. ハード MAX_CHUNKS キャップを含む明示的なチャンク

また確認してください：

• 日付範囲はデフォルトで制限されているか（--override なしで最大 14 日を提案）？
• スクリプトが実行途中でクラッシュした場合、二重書き込みなしで再実行しても安全か？
• バックデート型シミュレーションについては、時間一貫性のあるスナップショットからデータが読み込まれていることを確認 （FOR SYSTEM_TIME AS OF、パーティション型 as-of テーブル、または日付スナップショットテーブル）。 バックデート モード実行時に「最新」またはバージョンなしテーブルから読み込む場合フラグをつけてください。

現在のアプローチが行ごとの場合、具体的な書き換えを提案してください。

---

D) クエリ安全性とスキャンサイズ

各クエリについて確認してください：

• パーティションフィルタ は DATE(ts)、CAST(...)、またはプルーニングを防ぐ関数ではなく、 生のカラムの上にあるか
• SELECT * なし：ダウンストリームで実際に使用される列のみ
• 結合が爆発しないか：結合キーが一意または适切にスコープされていることを確認し、 任意の多対多の可能性をフラグてください
• 高コスト操作（REGEXP、JSON_EXTRACT、UDF）はパーティションフィルタ後にのみ実行され、 全テーブルスキャンの上ではないか

チェックに失敗したクエリについて、具体的な SQL 修正を提供してください。

---

E) 安全な書き込みとべき等性

すべての書き込み操作を特定してください。 デデュプロジック なしの純粋な INSERT/追加をフラグしてください。

各書き込みは以下のいずれかを使用する必要があります：

1. 決定論的キー（例：entity_id + date + model_version）での MERGE
2. 実行にスコープされたステージングテーブルに書き込み、その後最終テーブルにスワップまたは結合
3. 追加のみで、デデュプビュー： QUALIFY ROW_NUMBER() OVER (PARTITION BY <key>) = 1

また確認してください：

• 再実行により重複する行が作成されるか？
• 書き込み配置（WRITE_TRUNCATE vs WRITE_APPEND）は意図的で ドキュメント化されているか？
• run_id がマージまたはデデュプキーの一部として使用されているか？その場合フラグしてください。 run_id はメタデータカラムとして保存される必要があり、一意性キーの一部としてではありません。 マルチ実行履歴を明示的に望む場合を除きます。

推奨アプローチと、このコードベースの正確なデデュプキーを述べてください。

---

F) 可観測性：失敗をデバッグできるか？

検証してください：

• 失敗は例外を発生させ、サイレント except: pass または警告のみなしで中止
• 各 BQ ジョブはログに記録：ジョブ ID、処理またはスキャンされたバイト（利用可能な場合）、 スロットミリ秒、期間
• 実行サマリーがログに記録または最後に書き込まれる： run_id、env、mode、date_range、テーブル書き込み、合計 BQ ジョブ、合計バイト
• run_id が存在し、すべてのログ行で一貫性がある

run_id が不足している場合、1 行の修正を提案してください： run_id = run_id or datetime.utcnow().strftime('%Y%m%dT%H%M%S')

---

Final

1. PASS / FAIL セクション別の具体的な理由（A～F）。 2. パッチリスト 変更対象の正確な関数を参照し、リスク順に並べられています。 3. FAIL の場合：上位 3 つのコストリスク 大まかな最悪ケース推定を含む （例：「90 日 × 3 リトライでループ = 270 BQ ジョブ」）。