clickhouse-migrations

ClickHouse マイグレーション

包括的なパターン、クラスターのセットアップ、例、インジェストレイヤーの詳細については、posthog/clickhouse/migrations/AGENTS.md を参照してください。

クイックリファレンス

マイグレーションの構造

operations = [
    run_sql_with_exceptions(
        SQL_FUNCTION(),
        node_roles=[...],
        sharded=False,  # True for sharded tables
        is_alter_on_replicated_table=False  # True for ALTER on replicated tables
    ),
]

ノードの役割 (オブジェクトを所有するクラスターで選ぶ)

posthog/clickhouse/client/connection.py の NodeRole のメンバーのみが有効です: ALL、DATA、 INGESTION_EVENTS、INGESTION_SMALL、INGESTION_MEDIUM、ENDPOINTS、LOGS、APM、AI_EVENTS、 AUX、BATCH_EXPORTS、OPS、SESSIONS。この enum にない名前はインポート時に失敗し、それによって マイグレーションの検出が中断され、マイグレーションを実行するすべてのジョブが巻き添えで停止します。

役割は、その型だけではなく、どのクラスターがそのオブジェクトを所有するかで選んでください:

• [NodeRole.DATA]: メインクラスターに属するオブジェクト — シャーディングされたテーブル、シャーディングされたテーブルや衛星テーブルの前段にある Distributed 読み取り テーブル、ビュー、ディクショナリ。DATA 上の非シャーディングのレプリケートテーブルは、メインクラスターのすべてのノードに完全なコピーを保持するため、 製品のロールアップや事前集計の置き場所としては不適切です (以下を参照)。
• [NodeRole.INGESTION_SMALL]: 書き込み可能なテーブル、Kafka テーブル、インジェストレイヤー上のマテリアライズドビュー
• [NodeRole.APM]: ingestion-apm ノード、メトリクスのインジェストレイヤー。Kafka テーブル、Null 入力 テーブル、マテリアライズドビュー、および LOGS 上のストレージに書き込む書き込み可能な Distributed テーブル。 他のインジェストノードと同様にステートレスであるため、ストレージテーブルをここに置くことはありません。
• [NodeRole.OPS]、[NodeRole.LOGS]、[NodeRole.AUX]、[NodeRole.AI_EVENTS]、[NodeRole.SESSIONS]、 [NodeRole.BATCH_EXPORTS]: 衛星クラスター上に存在するオブジェクト。これらのいずれかのためのテーブルを DATA に置くと間違ったノードに配置され、本来のクラスターにはそれが存在しなくなるため、DATA を既定にする前に そのオブジェクトがどこで読み書きされるかを確認してください。Dev は US/EU 本番と同じ衛星クラスターを実行します — dev に異なるレイアウトを与えるために CLOUD_DEPLOYMENT で分岐しないでください。
• [NodeRole.ALL]: ほとんど使用されません

事前集計とロールアップは衛星上に保存され、DATA 上の Distributed テーブルを通じて読み取られます (また、そこでのアドホックな読み取りのためにストレージ衛星上でも)。ただし、アプリがすでにその衛星に 直接接続している場合は除きます。衛星自身のデータのロールアップはその 衛星に留まります。一般的な製品の事前集計は AUX に置きます。以下のライブチェックで衛星を確認し、 hclexp locate で現在の例の HCL レイヤーを見つけてください。

役割を選ぶ前にライブクラスターを確認する

古いマイグレーション、HCL の local レイヤー、共有ヘルパーは、dev と本番で実行されているものに遅れをとることがあります。マイグレーションを書く前に、ライブクラスターに対して配置を確認してください:

1. ClickHouse MCP を使用して、dev、prod-us、prod-eu のそれぞれを確認します: 比較可能なテーブルがどこに存在するか、 メインクラスターが Distributed テーブルを通じて読み取るのはどの衛星テーブルか、そして新しいテーブル名が 未使用であること。セットアップと使用方法については、Housekeeper MCP ランブックを参照してください。
2. 前例が一致しない場合は、そのテーブルファミリーの最新のマイグレーションに従い、それが 3 つの環境すべてで 実行されている内容と一致することを確認してください。一部の環境でのみまだ実行されているレイアウトはマイグレーション途中であり、 コピーすべきパターンではありません。
3. PR の説明に、新しい各オブジェクトがどのクラスターに配置されるかを明記し、その選択の根拠となる参照マイグレーション またはライブチェックを挙げてください。

テーブルエンジンのクイックリファレンス

MergeTree エンジン:

• シャーディングされたテーブルには AggregatingMergeTree(table, replication_scheme=ReplicationScheme.SHARDED)
• シャーディングされていないテーブルには ReplacingMergeTree(table, replication_scheme=ReplicationScheme.REPLICATED)
• その他のバリアント: CollapsingMergeTree、ReplacingMergeTreeDeleted

分散エンジン:

• シャーディングあり: Distributed(data_table="sharded_events", sharding_key="sipHash64(person_id)")
• シャーディングなし: Distributed(data_table="my_table", cluster=settings.CLICKHOUSE_SINGLE_SHARD_CLUSTER)

Kafka エンジン — kafka_engine() は ENGINE 句のみを返すため、各テーブルはそれぞれ独自の SETTINGS を追加します。次のベースラインから始めてください:

SETTINGS kafka_skip_broken_messages = 100,
         kafka_num_consumers = {kafka_num_consumers(1)},
         kafka_thread_per_consumer = 1,
         kafka_poll_timeout_ms = 10000,
         kafka_max_block_size = 100000

• kafka_skip_broken_messages のデフォルトは 0 で、不正な形式のメッセージが 1 つあるとコンシューマーが完全に停止します。
• kafka_poll_timeout_ms は WarpStream では 10000 です。WarpStream は fetch.min.bytes をサポートしていません。
• kafka_max_block_size のデフォルトは約 100 万行であるため、100000 は削減になります。幅の広い 行ではさらに下げてください。ただし約 1000 を下回らないようにしてください。
• kafka_num_consumers はノードごとの値であるため、ロールのノード数がそれを掛け合わせます。合計は トピックのパーティション数を超えてはなりません。
• カラムと設定は作成時に固定されます。MV とともにテーブルを削除して再作成してください。SYNC は不要です。 ALTER TABLE <mv> MODIFY QUERY は MV-SELECT のみの変更に対応します。 kafka_events_json_ws / events_json_ws_mv には決して使用しないでください — 重要なルールを参照してください。

posthog/clickhouse/migrations/AGENTS.md にサイジングのルールとその他の詳細があります。

重要なルール

• SQL ステートメントで ON CLUSTER 句を決して使用しないでください
• 常に IF EXISTS / IF NOT EXISTS 句を使用してください
• 同じマイグレーションでレプリケートテーブルを削除して再作成する場合は、DROP TABLE IF EXISTS ... SYNC を使用してください
• SQL を生成する関数に on_cluster パラメータがある場合は、常に on_cluster=False を設定してください
• シャーディングされたテーブルを変更する場合は sharded=True を使用してください
• シャーディングされていないレプリケートテーブルを変更する場合は is_alter_on_replicated_table=True を使用してください
• カラムに CODEC(ZSTD(1)) を決して記述しないでください — サーバーはすでにすべてのカラムを ZSTD で圧縮しているため、何も得られません。このデフォルトを上回る場合にのみ CODEC を宣言し、まず ORDER BY を確認してください: Delta/DoubleDelta はカラムがストレージ順でほぼソートされていること（先頭のソートキープレフィックス）を必要とし、キーがそのカラムをバケット化するだけ、あるいは省略している場合は不利になります。T64/Gorilla は順序に依存しません。ストレージテーブルにのみ設定してください — Distributed テーブルや Kafka テーブル上の CODEC は、それが前面に立つシャーディングテーブルから乖離する不活性なメタデータです。
• DROP COLUMN マイグレーションを自分で決して記述しないでください — DROP COLUMN は ClickHouse でスタックし、リリースをブロックする可能性があります。カラムの削除は 2 段階のプロセスです: (1) ClickHouse チームがクラスター上で直接カラムを削除し、次に (2) コードベースのスキーマが同期を保つように、一致する DROP COLUMN を含むマイグレーションを記述します。ClickHouse チームがステップ 1 を先に実行していない限り、マイグレーションから削除を開始しないでください。
• kafka_events_json_ws または events_json_ws_mv を決して削除または再作成しないでください — これらのテーブルは立入禁止区域です。MV の定義は US prod、EU prod、dev で大きく異なり（環境固有の mat_* カラムが多数あります）、それらの違いはリポジトリに反映されていません。リポジトリの SQL から削除および再作成すると、環境固有のスキーマが破壊され、イベントの取り込みが壊れます。変更は必ず ClickHouse チームを通してください。

PR のスコープ

ClickHouse マイグレーションを含む PR は、マイグレーションのみでなければなりません。同じ PR でマイグレーションファイルと機能コード、API の変更、モデルの変更、またはフロントエンドの変更を混在させないでください。マイグレーション関連のファイルは次のとおりです:

• マイグレーションファイル自体 (posthog/clickhouse/migrations/0NNN_*.py)
• マイグレーションが依存する SQL 定義ファイル（例: posthog/clickhouse/sql/*.py、テーブルエンジンのヘルパー）
• マイグレーション、またはそれが触れる SQL 定義を直接テストするテスト

スキーマの変更と、その新しいスキーマを使用するアプリケーションコードの両方が必要な場合は、まずマイグレーションを単独の PR で出荷し、アプリケーションコードの PR より先にマージしてください。

ローカルセットアップの同等性

クラウドにのみ存在するテーブルがあってはなりません。 マイグレーションで作成されたすべてのテーブルは、ローカルの開発環境にも存在している必要があります。

一部のマイグレーションはクラウド用にガードされており、ローカル/ホビーの開発ではスキップされます。settings.CLOUD_DEPLOYMENT を直接参照するのではなく、必ず posthog.run_mode で判定してください。clickhouse-migrations-use-run-mode semgrep ルールが生の比較をブロックするためです。

from posthog.run_mode import RunMode, run_mode

operations = (
    []
    if not run_mode().is_deployed_cloud  # US/EU/DEV
    else [...]
)

述語	真になる対象
run_mode().is_deployed_cloud	US、EU、DEV（通常のマイグレーションゲート）
run_mode().is_prod_cloud	US、EU（ステージングを除く）
run_mode() is RunMode.CLOUD_US	1 つのリージョン（CLOUD_EU、CLOUD_DEV も同様）

run_mode().is_cloud は E2E もカウントするため、マイグレーションには適していません。それは posthog.cloud_utils.is_cloud に一致します。

モジュールレベルの定数に代入するのではなく、必要な箇所で run_mode() を呼び出してください。posthog/clickhouse/test/test_migrations.py は、パッチを適用した posthog.settings.CLOUD_DEPLOYMENT の下で毎回すべてのマイグレーションを再インポートし、各デプロイの分岐に紛れ込んだ ON CLUSTER がないか確認しますが、キャッシュされた値ではそのカバレッジが黙ってスキップされてしまいます。

このようなガードの内部で新しいテーブルを作成する場合は、その SQL 関数を posthog/clickhouse/schema.py の適切なタプルにも追加して、テーブルがローカルで作成されるようにする必要があります。

テーブルの種類	schema.py のタプル
MergeTree / ベーステーブル	CREATE_MERGETREE_TABLE_QUERIES
Distributed / 書き込み可能	CREATE_DISTRIBUTED_TABLE_QUERIES
Kafka コンシューマー	CREATE_KAFKA_TABLE_QUERIES
マテリアライズドビュー	CREATE_MV_TABLE_QUERIES
非マテリアライズドビュー	CREATE_VIEW_QUERIES
辞書	CREATE_DICTIONARY_QUERIES

唯一の例外は、定義が環境ごとに意図的に異なり、リポジトリで追跡されていないテーブルです（例: 立入禁止区域の events_json_ws_mv テーブル）。

辞書の認証情報: 辞書が SOURCE(CLICKHOUSE(...)) を使用する場合、ソースのユーザー/パスワードを get_clickhouse_creds(ClickHouseUser.DICT_READER) 経由で解決し、それらを USER/PASSWORD 句に補間してください。default/CLICKHOUSE_USER をハードコードしたり、認証情報を省略したりしないでください。これにより、辞書の認証は専用の低権限 dict_reader ユーザーに保たれ、default から切り離されます。環境変数が設定されていない場合は default の認証情報にフォールバックします。パターンについては posthog/models/exchange_rate/sql.py を参照してください。

宣言的スキーマ（HCL）も一致させる必要があります

一部のロールは posthog/clickhouse/hcl/ でも宣言されており、マルチノードマイグレーションのスモークテストはマイグレーション後のクラスターを再イントロスペクトし、ライブスキーマとコミットされたゴールデンの間にドリフトがあれば失敗します。したがって、管理対象ロール上でマイグレーションによって追加されたテーブルは、同じ PR で HCL も更新する必要があります。そうしないと、CI は check_live_hcl で DRIFT: ブロックとともに新しいオブジェクトを名指しして失敗します。

manifest.hcl に local-multi ブロックがあるすべてのロールが管理対象であり、メインクラスターの data ロールおよびほとんどのサテライトおよびインジェストのロールが含まれます。対象としているロールについてそのファイルを確認してください。管理対象ロール上の新しいテーブルも該当します。マイグレーションを書いた後は次のようにします。

HCL=posthog/clickhouse/hcl
# add the table to the layer that composes it (hclexp locate a sibling table to find that layer)
bash $HCL/gen-golden.sh && bash $HCL/gen-sql.sh   # refresh generated artifacts
bash $HCL/check.sh                                # must exit 0

レイヤーを編集する前に posthog/clickhouse/hcl/README.md を読んでください。どのレイヤーを選ぶべきか、抽象カラムリスト、およびマイグレーションを自動生成するコード生成パスについて説明されています。

テスト

マイグレーションを再実行するには、infi_clickhouse_orm_migrations テーブルからエントリを削除してください。