Cloud Spanner 分散クエリ実行エンジン解体新書：数千ノードを束ねる「見えざる手」の制御法
こんにちは。テクニカルリードの私だ。
これまでのレビューで、君たちが書いたクエリやスキーマ設計を見ていて気になったことがある。「Cloud Spannerはリレーショナルデータベースでありながら、内部は巨大な分散ストレージだ」という事実を、どこまで意識してコードを書いているかという点だ。
「単一のトランザクションで安全に動くから」といって、何も考えずに数百万行をスキャンするようなクエリを投げ込んでいないか？ あるいは、スプリット（Split）やマルチテナントのシャード分割を無視して、特定のノードだけをホットスポットにしていないか？
今回は、Cloud Spannerのコア中のコア、「分散クエリ実行エンジン（Distributed Query Execution Engine）」の深層を紐解く。これが内部でどうタスクを分割し、データをシャッフルし、結合（Join）をさばいているのか。その物理法則を理解すれば、君たちの設計レビューの質は劇的に変わるはずだ。
—
1. 概念の破壊：Spannerのクエリは「RDBのそれ」とは別物である
まず、頭をアップデートしてほしい。Spannerの分散クエリエンジンは、単一ノードのPostgreSQLやMySQLの延長線上にはない。これは、「数千の独立したストレージノード（Spanserver）に分散したデータを、あたかも1台の巨大なメモリ上にあるかのように見せかけ、限界まで並列度を引き出す分散データフローエンジン」だ。
基本アーキテクチャの構成要素は以下の3つに大別される。
1. Root Coordinator（ルート・コーディネーター）: クライアントから受け取ったSQLを解析し、オプティマイザ（Query Optimizer）が生成した分散実行プラン（Distributed Execution Plan）の統括を行う。
2. Intermediate Nodes（中間ノード）: データの集約（Aggregation）、ソート、分散結合（Distributed Join）のバッファリングやマージを担当する。
3. Leaf Nodes（リーフノード / Spanserver）: 実際のデータを保持するストレージ（PangaeaファイルシステムとLSMツリー構造）に直接アクセスし、フィルタリングやローカルスキャンを行う。
クエリが発行されると、Spannerは単一のプロセスでそれを処理するのではなく、数千のタスクに分解し、データが眠る物理的な場所（Locality）へ計算のほうを飛ばす（Compute-to-Data）。このパラダイムシフトを理解することがすべての出発点だ。
—
2. 内部メカニズム：タスク分割・データシャッフル・中間結果集約の三相
分散クエリが実行されるとき、内部では厳密なパイプラインが組まれている。その動きを3つのフェーズで分解しよう。
[Client]
│ (SQL Request)
▼
[Root Coordinator] ──(プランの配給)──► [Intermediate Nodes (Shuffle/Join)]
▲
┌──────────────────────┴──────────────────────┐
│ (Parallel Leaf Scans) │
▼ ▼
[Leaf Node A (Split 1)] [Leaf Node B (Split 2)]
① タスク分割（Task Splitting & Pruning）
テーブルは数GB〜数十GB単位の「スプリット（Split）」と呼ばれる物理的なシャードに分割され、世界中のSpanserverに分散配置されている。
オプティマイザは、`WHERE` 句の条件（特にプライマリキーやインターリーブのプレフィックス）を見て、どのスプリットに目的のデータが存在するかを特定し、不要なスプリットへのリクエストを容赦なく枝刈り（Pruning）する。
残ったスプリットに対して、リーフノードで実行される「Leaf Scanタスク」が並列発行される。
② データシャッフル（Data Shuffling）
ここがパフォーマンスの分水嶺だ。例えば、異なるスプリットに存在するデータを `JOIN` する場合や、`GROUP BY` で再集約する必要がある場合、データはネットワークを越えてシャッフルされなければならない。
Spannerのエンジンは、特定のキーハッシュに基づいてデータを動的に再パーティショニングし、次のステージの中間ノードへストリーミング送信する。
このとき、不要な大容量データがネットワークをクロスしている場合（いわゆる「シャッフル地獄」）、CPUとネットワーク帯域が飽和し、クエリレイテンシは劇的に悪化する。
③ 中間結果の集約（Aggregation & Merge）
リーフおよび中間ノードで部分集約（Partial Aggregation）を行った後、ルートコーディネーター（あるいは上位の中間ノード）へ結果が送られ、最終的なマージが行われる。
ストリーミングアーキテクチャを採用しているため、全データが揃うのを待たずにパイプラインの川下へデータを流し込むことが可能になっている。
—
3. 実務で直面する「アンチパターン」と堅牢な設計パターン
理論はこれくらいにして、実際の設計レビューで私が指摘するポイントをコードを交えて解説しよう。
アンチパターン：非効率な分散JOINとフルスキャン
以下のクエリを見てほしい。
— 【アンチパターン】巨大な2つのテーブルを結合し、ローカル性を無視したクエリ
SELECT
u.user_id,
o.order_amount,
p.product_name
FROM Users u
JOIN Orders o ON u.user_id = o.user_id
JOIN Products p ON o.product_id = p.product_id
WHERE u.country_code = ‘JP’;
何が問題か？
`Users` と `Orders` が別々のスプリットに散らばっている場合、これらを結合するために膨大なデータシャッフルが発生する。さらに、`Products` テーブルのキー設計が適切でなければ、すべてのノード間でブロードキャストまたはハッシュシャッフルが起き、CPU使用率が100%に張り付く。
堅牢な設計パターン：インターリーブ（Interleave）とロカリティの最大化
Spannerの真価を引き出すには、「物理的なデータ配置（Locality）」をスキーマ設計段階で強制することだ。親子関係にあるテーブルは、`INTERLEAVE IN PARENT` を使って同じ物理スプリット内に同居させよ。
— 【堅牢な設計】インターリーブ構造によるコロケーション（Colocation）
CREATE TABLE Users (
user_id INT64 NOT NULL,
country_code STRING(2),
created_at TIMESTAMP,
) PRIMARY KEY (user_id);
CREATE TABLE Orders (
user_id INT64 NOT NULL,
order_id INT64 NOT NULL,
order_amount FLOAT64,
order_date TIMESTAMP,
) PRIMARY KEY (user_id, order_id),
INTERLEAVE IN PARENT Users ON DELETE CASCADE;
なぜこれが強いのか？
`Users` とその子である `Orders` は、同じ `user_id` をプレフィックスとして物理的に同一のストレージブロック（スプリット）に隣接して保存される。
そのため、`user_id` を軸にしたJOIN（親子の結合）は、ネットワークを跨ぐシャッフルを一切必要としない。リーフノード内部のメモリ・ディスク上でローカルに完結するため、爆速で処理される。これがSpannerにおける最強の設計パターンだ。
—
4. パフォーマンスチューニングの実務知見：EXPLAIN PLANを読め
「クエリが遅い」と嘆く前に、必ず `EXPLAIN`（または `EXPLAIN ANALYZE`）を実行し、分散クエリエンジンの実行計画を覗き見ることだ。
— 実行計画の取得
EXPLAIN
SELECT user_id, SUM(order_amount)
FROM Orders
WHERE order_date >= ‘2023-01-01’
GROUP BY user_id;
チューニングで見るべき3つのチェックポイント
1. Distributed Union / Distributed Cross Apply の存在
実行計画に `Distributed Union` が頻出している場合、クエリが多くのスプリットにまたがって並列スキャンを行っていることを示す。これが多すぎると、コーディネーターのオーバヘッドが増える。
2. Hash Join vs Apply Join (Nested Loop)
スプリットを跨ぐ結合で `Hash Join` が選ばれている場合、データシャッフルが発生している。可能であれば、インターリーブ設計やセカンダリインデックスの `STORING` 句を活用して、`Distributed Apply` やローカルな結合へ誘導できないか検討せよ。
3. Rows（行数）と Bytes（バイト数）の不一致
`EXPLAIN ANALYZE` で、初期のリーフスキャンで何百万行も読んでいるのに、最終的に数行しか返していないクエリがないか確認する。インデックスの貼る位置が間違っているか、`WHERE` 句のプッシュダウンが効いていない証拠だ。必要なカラムを網羅したカバーリングインデックス（Covering Index / STORING句付きインデックス）を導入し、ストレージからのデータフェッチ量を物理的に削れ。
— カバーリングインデックスの例：インデックス側だけでスキャンを完結させる
CREATE INDEX OrdersByDate
ON Orders(order_date)
STORING (order_amount);
このインデックスがあれば、`order_date` のフィルタと `order_amount` の集約が、ベーステーブルのデータ本体（Primary Data）にアクセスすることなく、インデックスのリーフノードだけで完結する。I/Oコストは劇的に下がる。
—
5. チーフアーキテクトからの総括
Cloud Spannerの分散クエリ実行エンジンは、私たちが書いたSQLを世界規模の巨大な並列計算機上で最適に翻訳・実行してくれる極めて洗練されたシステムだ。しかし、それは「魔法の箱」ではない。
- データを読むときは、どこに物理的に配置されているかを想像する。
- 結合するときは、ネットワークを跨ぐシャッフルを最小限にする（インターリーブの活用）。
- スキャンするときは、インデックスとストレージのI/Oを削る（カバーリングインデックスの活用）。
この3つを常に頭に置き、設計レビューに臨んでほしい。
コードの美しさは、分散システムの物理法則を理解した者だけに宿る。次のレビューで、君たちの洗練されたクエリとスキーマを見せてもらうのを期待している。
コメント