適応的クエリ実行とランタイム再最適化

静的な最適化の限界

コストベースの最適化（→ クエリオプティマイザの内部）は、実行を始める前に統計情報から各候補プランのコストを見積もり、最小コストの物理プランを1つ確定します。問題は、この見積もりが統計の精度に完全に依存する点です。複数条件の相関、結合をまたいだ選択率の伝播、UDF やパラメータ化された述語などにより、推定行数（カーディナリティ）は容易に1桁、2桁とずれます。

しかも誤差は結合木を上に伝わるほど増幅します。下流の1つの結合で出力行数の推定が10倍外れれば、その上のすべての結合・集約のコスト見積もりが連鎖的に狂い、不適切な結合アルゴリズムやパーティション数が選ばれます。静的最適化はこの誤差をプラン確定後に修正する手段を持ちません。**適応的クエリ実行（Adaptive Query Execution, AQE）**は、実行を進めながら得た「実測値」でプランの一部を作り直すことで、この構造的な弱点に耐性を与えます。

適応実行の基本原理

核心は単純です。実行を最後まで一気に流さず、材料化境界（materialization boundary）でいったん止め、そこまでの実測統計を集めて残りのプランを再決定します。

• 観測点: パイプラインが切れて中間結果がディスクやメモリに書き出される地点。分散処理ではシャッフル（再分配）の完了点が代表例。ここでは各パーティションの実際のバイトサイズと、ステージ全体の実出力行数が確定する。
• 再最適化: 確定した実測統計で、まだ実行していない上流ステージのプランを組み直す。推定値ではなく実測値を使うため、判断が当たりやすい。
• 制約: すでに走り終えたステージはやり直さない。再最適化できるのは「観測点より上流の未実行部分」に限られる。

Spark AQE が実行中に行う3つの調整

Spark の AQE は、各シャッフル完了時の実測統計（MapStatus が持つパーティションごとのバイトサイズと、ステージ単位で集計される実出力行数）を使い、代表的に次の3つを動的に適用します。

パーティション合体（coalesce）

シャッフルのパーティション数（spark.sql.shuffle.partitions、既定 200）は実行前に固定で決め打ちするため、実データが小さいとほぼ空のパーティションを処理する大量のタスクが無駄に起動します。AQE は各パーティションの実サイズを見て、目標サイズ（既定 64MB 程度）に届くよう隣接パーティションを連結し、タスク数を実データ量に見合った数まで減らします。スケジューリングのオーバーヘッドが下がり、小タスクの裾が消えます。

結合方式の動的切替

オプティマイザは静的にはフィルタ後のサイズを推定するしかなく、Broadcast の可否（片側が spark.sql.autoBroadcastJoinThreshold 以下か）を見誤りがちです。AQE はサブクエリやフィルタを通過した実際の出力サイズがシャッフル後に判明した時点で、片側が十分小さければ Sort-Merge Join を Broadcast Hash Join に切り替えます。これにより大きい側のシャッフルとソートを丸ごと省け、コストの効き方が大きく変わります（→ ハッシュ結合の内部：グレース法とハイブリッドハッシュ で扱う方式選択がそのまま実行時判断になる）。

スキュー結合の分割

結合キーの値分布が偏ると（ホットキー）、特定パーティションだけが巨大化し、その1タスクが全体の完了を律速します。AQE は中央値に対し突出して大きいパーティション（既定で中央値の5倍かつ 256MB 超など）を検出し、それを複数のサブパーティションに分割、相手側の対応パーティションを各サブに複製して並列度を回復します。

推定誤差への耐性という観点

AQE の本質的な価値は、特定の最適化メニューそのものより**「推定が外れても致命傷になりにくい」という耐性**にあります。静的最適化が抱える誤差の増幅（→ カーディナリティ推定とヒストグラム・スケッチ）に対し、適応実行は次の構造で対抗します。

静的最適化:  推定 → プラン確定 → 実行（誤差は最後まで残る）
適応実行:    推定 → 一部実行 → 実測でプラン補正 → 続行（境界ごとに誤差をリセット）

材料化境界を通過するたびに、それより下流のカーディナリティは推定ではなく実測に置き換わります。つまり誤差が伝播する距離が「次の境界まで」に限定され、深い結合木でも累積誤差が抑え込まれます。これは、相関や偏りで統計が当てにならないクエリほど効果が大きいことを意味します。

適用範囲と注意点

注意点として、再最適化は材料化境界の存在が前提なので、境界の少ないクエリでは出番がほとんどありません。また各調整には閾値パラメータ（目標パーティションサイズ、ブロードキャスト上限、スキュー判定倍率など）があり、これらが実データの特性とずれていると効果が出ない、あるいは過剰な分割でかえって遅くなることがあります。AQE は静的最適化や統計整備を置き換えるものではなく、推定が外れたときの保険として上に重なる層だと捉えるのが正確です。巨大表をあらかじめ適切に分割しておく設計（→ パーティショニング）と組み合わせれば、そもそもスキューや空タスクを生みにくくでき、適応実行の負担も軽くなります。

まとめ

適応的クエリ実行は、実行前の統計推定だけでプランを固定せず、シャッフルなどの材料化境界で得た実測のカーディナリティとサイズを使って、上流の未実行部分を作り直す仕組みです。Spark AQE では「小パーティションの合体」「Sort-Merge から Broadcast への結合切替」「スキューパーティションの分割」として具体化されます。最大の意義は個々の最適化ではなく、推定誤差が深い結合木で増幅するという静的最適化の構造的弱点に対し、境界ごとに誤差をリセットして耐性を与える点にあります。