分散結合とシャッフル・ブロードキャスト・コロケーション

分散結合で本当に効くのはネットワーク

単一ノードの結合では I/O と CPU が支配項でした（結合アルゴリズム参照）。ところが MPP データウェアハウスや分散SQLでは、表が複数ノードにシャード（水平分割）されているため、結合の前にまず「結合相手の行を同じノードへ集める」必要があります。このノード間のデータ移動がボトルネックです。ネットワーク帯域はメモリ・ローカルディスクより桁で遅く、転送量がそのまま実行時間を支配します。

前提を固定します。ノード数を P、左表 R の総行数を M、右表 S の総行数を N とし、両表は何らかのキーで各ノードへ均等に分散済みとします。結合は等値結合 R.k = S.k を考えます。問題は「結合キー k が、その表のシャードキーと一致しているか」です。一致の有無で、行を動かさずに済むか、動かすなら誰をどれだけ動かすかが決まります。

コロケート結合：行を1バイトも動かさない

両表が同じ結合キー k で同じ方式（同じハッシュ関数・同じノード割当）でシャードされているなら、R.k = S.k を満たす行は定義上必ず同じノードに同居しています。したがって各ノードは自分が持つ R と S の断片だけをローカルに結合すればよく、ノード間転送は発生しません。これが**コロケート結合（colocated / local join）**です。

-- 各ノード i で独立に実行（通信なし）
for node i in 0..P-1:
    local_join(R_i, S_i)        -- R_i, S_i は node i 上の断片
emit ∪ 各ノードの結果

正しさの核心は、結合キーが等しい行は同じハッシュ値を持ち、同じノードへ割り当てられるという点です。異なるノードにまたがる一致ペアは存在しえないので、ローカル結合だけで完全な結果が得られます。ネットワーク転送量は 0。これが分散結合の理想形であり、スキーマ設計で「よく結合する表を同じキーでシャードする」ことの最大の動機です。多くの分散DB（Citus の create_distributed_table(..., colocate_with=...)、CockroachDB の interleaved/locality、Spanner の親子テーブル）はこの同居を明示的に作る機能を持ちます。

シャッフル結合：両表を結合キーで配り直す

結合キーがどちらのシャードキーとも一致しないとき（例：R は user_id、S は product_id でシャードされ、結合は別の order_id で行う）、行は同居していません。そこで**両表を結合キー k のハッシュで全ノードへ再分配（repartition）し、同じ k の行を同じノードへ集めてから局所結合します。これがシャッフル結合（shuffle / repartition join）**で、単一ノードのグレース／ハイブリッドハッシュ結合を分散に拡張したものです。

-- フェーズ1: 再分配（ネットワーク転送が発生）
for each row r in R:  送る → node[ h(r.k) mod P ]
for each row s in S:  送る → node[ h(s.k) mod P ]

-- フェーズ2: 各ノードで局所ハッシュ結合
for node i in 0..P-1:
    build hashtable from received R rows
    probe with received S rows → emit

転送量の見積もりが要点です。各行は自分の h(k) が指すノードへ送られます。すでに正しいノードにある行（確率 1/P）は送らずに済みますが、残りの (P-1)/P の行はネットワークを渡ります。よって転送される行数はおおむね (M + N) × (P-1)/P、P が大きければ実質 M + N のほぼ全行が一度ネットワークを流れます。両表が大きいほど、これが分散結合で最も重いプランになります。

ブロードキャスト結合：小さい表を全員に配る

片方の表（仮に S）が十分小さいなら、S を再分配するのでなく**全ノードへ丸ごと複製（broadcast / replicate）**し、R は1行も動かさない方が安くなります。各ノードは手元の R 断片と、複製された S 全体を局所結合します。

-- S 全体を全ノードへ複製（R は動かさない）
broadcast(S) → 全ノードが S のフルコピーを持つ

-- 各ノードで局所結合（R は移動なし）
for node i in 0..P-1:
    build hashtable from full S
    probe with local R_i → emit

転送量は N × P（小表 S の N 行を P ノードへ）です。これをシャッフルの M + N（厳密には (M+N)(P-1)/P）と比べると、ブロードキャストが有利になる条件は N × P が M + N を下回るとき、つまり S が R に対して桁違いに小さいときです。大表 R を動かさずに済むのが効きます。スター型スキーマの「巨大ファクト表 × 小さいディメンション表」が典型的な適用先で、ディメンション表を各ノードへ撒けば大表のシャッフルを完全に回避できます。

分散プランナの選択原理

分散オプティマイザは、各プランの総コストをネットワーク転送量＋各ノードの局所結合コストとして見積もり、最小のものを選びます。判定はおおむね次の優先順位です。

if 結合キーが両表のシャードキーと一致:
    → コロケート結合（転送 0、最優先）
elif 片方の表が小さい（推定サイズ × P が他方を下回る）:
    → ブロードキャスト結合（小表のみ複製）
else:
    → シャッフル結合（両表を再分配・最も汎用だが最も重い）

この判定が基数推定（行数・サイズの見積もり）に強く依存する点が実務上の急所です（基数推定参照）。「小さいはず」とブロードキャストした表が、選択述語の見積もり誤りで実は巨大だと、N × P の転送とメモリ確保が膨張し、各ノードでハッシュテーブルが作業メモリを溢れてスピルし、シャッフルより遥かに遅くなります。逆に小表を過大評価してシャッフルを選ぶと、避けられたはずの大表移動を払うことになります。

多段結合とコストの伝播

3表以上の結合では、ある結合の出力がそのまま次の結合の入力になります。ここでコロケーションは伝播し、また失われます。たとえば A と B を k でコロケート結合した結果は、依然 k でシャードされた状態でノードに残るので、続けて C を k で結合すれば再びコロケートできます。しかし B と C を別キー j で結合すると、A⋈B の結果を j で再シャッフルしてからでないと C と結合できません。

つまり結合順序（結合順序の列挙参照）は、単に局所コストだけでなく「中間結果がどのキーでシャードされて出てくるか」まで含めて評価されます。良い分散プランナは、再シャッフルの回数を減らすよう結合順序とパーティション方式を同時に最適化します。中間結果の分布（distribution / partitioning property）を物理プロパティとして持ち回り、不要な再分配を**交換演算子（exchange / motion ノード）**として明示的にプランへ挿入するのが現代の実装の定石です。

実務での読み方

分散DBの実行計画には、ノード間データ移動を表す専用ノードが現れます。Greenplum/Redshift 系の Redistribute Motion・Broadcast Motion、Spark の Exchange (hashpartitioning)・BroadcastExchange、Citus の repartition がそれです。結合の直下にこれらの Motion/Exchange が無ければコロケートが効いており理想的。Broadcast が出ていれば小表複製、Redistribute/hashpartitioning Exchange が両側に出ていればフルシャッフルです。チューニングの定石は次の順です。

• 頻出の結合キーで両表のシャードキーを揃え、コロケートに持ち込む（設計段階が最も効く）。
• 小表側に確実なサイズ上限があるならブロードキャストを誘導（多くのエンジンに broadcast ヒントがある）。ただし基数推定が外れると逆効果。
• シャッフルが避けられないなら、スキューの解消（偏ったキーの分離・salting）でストラグラーを潰す。
• 中間結果の分布を意識し、再シャッフルの段数を結合順序で最小化する。

母集合そのものを絞るパーティショニングや、列指向のデータウェアハウス設計と組み合わせると、転送量をさらに下げられます。

まとめ

• 分散結合のコストはネットワーク転送量が支配項。コロケート（0）＜ブロードキャスト（N×P）＜シャッフル（約 M+N）の順に重い。
• コロケート結合は結合キー＝両表のシャードキーのとき成立し、行を動かさず各ノードで局所結合する。設計で作り込むのが最善手。
• ブロードキャスト結合は小表を全ノードへ複製し大表を動かさない。N×P が M+N を下回る、小表が十分小さいときに有利。
• シャッフル結合は両表を結合キーで再分配する汎用手段だが、ほぼ全行がネットワークを流れ最も重い。スキューに弱い。
• プランナは基数推定に基づきコスト最小を選ぶ。推定誤りでブロードキャストが破綻すると最悪化する。多段結合では中間結果の分布を持ち回り、再シャッフルを最小化する。