😊

Cloud Spanner の Partition Query における root-partitionable の判定方法について

2024/10/02に公開

日付が変わる前に BigQueryからCloud Spannerへの連携クエリをroot-partitionableにするという記事を見て、去年の Advent Calendar のために温めたのを忘れていたまま冷めたネタを思い出したので書き殴りで供養する。

これから始まるのは Cloud Spanner の root-partitionable についての細かいお話。

前提知識

とりあえず Cloud Spanner の実行計画を読めること。
これから学ぶなら Google Cloud カスタマーエンジニアの Yasui Michitaka さんによる
Cloud Spanner の実行計画を読み解くあたりが良さそう。

あと Partition Query とか Data Boost についてもほぼ知っている前提で話す。

root-partitionable とは何か？

root-partitionable とは、 Partition Query できるクエリの条件である。

Cloud Spanner のドキュメント中では Reads outside of transactions の Read data in parallel のセクションで説明されており、公式にはこれが現在最も正しい説明だと言える。

For a query to be root-partitionable, the query plan must satisfy one of the following conditions:

The first operator in the query execution plan is a distributed union and the query execution plan only contains one distributed union (excluding Local Distribution Unions). Your query plan can't contain any other distributed operators, such as distributed cross apply.

There are no distributed operators in the query plan.

他の場所でより正確ではない説明がされていることがあるので気をつけてほしい。

なお、 Cloud Spanner インスタンスのリソースではなくオンデマンドのリソースを使ってクエリを処理する Data Boost を利用するためには Partition Query である必要があるため、全く同じ条件が適用される。

また、 BigQuery の Spanner Federated Query では Partition Query を BigQuery Connection の CloudSpannerProperties にそれぞれ Partition Query は use_parallelism=true, Data Boost は useDataBoost=true を設定することで有効にできる。
これらを有効にした BigQuery Connection に対して EXTERNAL_QUERY で実行したクエリは通常の ExecuteStreamingSQL API ではなく PartitionQuery API と多数の ExecuteStreamingSQL API で処理されることになり、 root-partitionable が要求されることになる。

論文中の root-partitionable の定義

Cloud Spanner という製品としてアクセスできるようになったのとほぼ同じ時期の Spanner について説明した論文である "Spanner: Becoming a SQL System" により詳細な説明がある。

root-partitionable に関することを要約すると、下記のようなことが書かれている。

partitionability とは分散オペレータである Distributed Union よりも下に処理を push down できる特性である。
- 一般的に Op = OpFinal ◦ OpLocal のように partitionability を持つ部分と持たない部分に分けられて Op(DistributedUnion[shard ⊆ T](F(Scan(shard)))) = OpFinal(DistributedUnion[shard ⊆ T](OpLocal(F(Scan(shard)))) のようになる。
全ての処理が push down された結果、最上位の Distributed Union でそれぞれ shard を持つ leaf node にサブプランの処理を振り分けた後はその結果を集めれば完了するクエリが root-partitionable である。
Parallel-consumer API(製品としての Cloud Spanner の機能である Partition Query に対応) では root-partitionable なクエリの出力を partition ごとに複数のコンシュマーで処理することで処理を分散する。

ここで重要なのは、この論文の root-partitionable の定義は一切分散 JOIN には触れていないことである。
実際には Distributed Union の下に Distributed Cross Apply などの分散オペレータがあったり、 Distributed Union の下にさらに Distributed Union がある場合があるので、最上位に Distributed Union があったとしてもそれで全ての処理が leaf node で完結するようにはなっていない。

なのでこの論文に始まる root-partitionable の定義は厳密ではないが、誤った定義がまだ広く残っていることが混乱の根源だと考えている。

インスタンスに負荷を掛けずに Partition Query 可能かどうかを確認する方法

検証の結果、 Partition Query が実行不可能でエラーになるタイミングは PartitionQuery API を実行した時点であることがわかっている。partition からデータを ExecuteStreamingSQL API を使って読み始めるまでは実際にデータにアクセスする処理は行われず負荷はかからないことが想定される。

よって、このような処理を execspansql に --try-partition-query として実装した。

このオプションを使うことで Partition Query が成功するかどうかを実際にエラーが出るかどうかで見分けることができる。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="SELECT 1"                                                             
success
$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="SELECT * FROM Singers JOIN Concerts USING(SingerId)"
2024/10/02 01:45:57 rpc error: code = InvalidArgument desc = Query is not root partitionable since it does not have a DistributedUnion at the root. Please check the conditions for a query to be root-partitionable.
error details: name = Help desc = Conditions for a query to be root-partitionable. url = https://cloud.google.com/spanner/docs/reads#read_data_in_parallel

root-partitionable のそれぞれの条件を考察する

前に引用した root-partitionable の条件は細かく分けると次のようになる。

分散処理が root-partitionable になるには下記全てを満たす
a. The first operator in the query execution plan is a distributed union
b. The query execution plan only contains one distributed union (excluding Local Distribution Unions)
c. Your query plan can't contain any other distributed operators, such as distributed cross apply.
分散処理ではない場合は下記を満たせば良い。
a. There are no distributed operators in the query plan.

追記: 1 は「分散オペレータが最初(プランツリーの最上位)の Distributed Union だけである」と言い換えた方が良いように思う。

それぞれを見ていく。なお、後述する面倒臭い話があるので一旦 query optimizer version は2に固定する。また、 Cloud Spanner 公式ドキュメントでお馴染みのサンプルスキーマを使用する。

1.a. The first operator in the query execution plan is a distributed union

一番最初のオペレータが Distributed Union であるというのは当初からの root-partitionable の要件だが、次のようなクエリが違反する。

@{OPTIMIZER_VERSION=2}
SELECT *
FROM Singers
ORDER BY FirstName;

実行計画は spanner-cli や rendertree で一覧性があり共有が容易なテキスト形式で表示できる。

+----+------------------------------------------------------------------------------------+
| ID | Query_Execution_Plan                                                               |
+----+------------------------------------------------------------------------------------+
|  0 | Serialize Result                                                                   |
|  1 | +- Sort                                                                            |
|  2 |    +- Distributed Union (distribution_table: Singers, split_ranges_aligned: false) |
|  3 |       +- Local Distributed Union                                                   |
|  4 |          +- Table Scan (Full scan: true, Table: Singers, scan_method: Automatic)   |
+----+------------------------------------------------------------------------------------+

ソートは一般的に Distributed Union で root に集めた後で処理するために最上位が Distributed Union ではなくなるため root-partitionable ではない。
実際に失敗することが確認できる。

$ execspansql ${SPANNER_DATABASE} --try-partition-query  --sql="@{OPTIMIZER_VERSION=3} SELECT * FROM Singers ORDER BY FirstName"
2024/10/02 01:57:13 rpc error: code = InvalidArgument desc = Query is not root partitionable since it does not have a DistributedUnion at the root. Please check the conditions for a query to be root-partitionable.
error details: name = Help desc = Conditions for a query to be root-partitionable. url = https://cloud.google.com/spanner/docs/reads#read_data_in_parallel

1.b. The query execution plan only contains one distributed union (excluding Local Distribution Unions)

Local ではない Distributed Union が複数含まれるクエリは root-partitionable ではない。これは色々な場面で発生するが、一例としては次のようなクエリが該当する。

@{OPTIMIZER_VERSION=2}
SELECT SingerId,
ARRAY(SELECT AS STRUCT * FROM Concerts WHERE Singers.SingerId = Concerts.SingerId)
FROM Singers;

このクエリの実行計画は次のようになる。

+-----+----------------------------------------------------------------------------------------------+
| ID  | Query_Execution_Plan                                                                         |
+-----+----------------------------------------------------------------------------------------------+
|   0 | Distributed Union (distribution_table: SingersByFirstLastName, split_ranges_aligned: false)  |
|   1 | +- Local Distributed Union                                                                   |
|   2 |    +- Serialize Result                                                                       |
|   3 |       +- Index Scan (Full scan: true, Index: SingersByFirstLastName, scan_method: Automatic) |
|   6 |       +- [Scalar] Array Subquery                                                             |
|  *7 |          +- Distributed Union (distribution_table: Concerts, split_ranges_aligned: false)    |
|   8 |             +- Local Distributed Union                                                       |
|   9 |                +- Compute Struct                                                             |
| *10 |                   +- Filter Scan (seekable_key_size: 0)                                      |
|  11 |                      +- Table Scan (Full scan: true, Table: Concerts, scan_method: Scalar)   |
+-----+----------------------------------------------------------------------------------------------+
Predicates(identified by ID):
  7: Split Range: ($SingerId_1 = $SingerId)
 10: Residual Condition: ($SingerId_1 = $SingerId)

Singers と Concerts は INTERLEAVE 関係にない無関係なテーブルなため、分散 JOIN ではなく相関サブクエリだが、一旦別の Distributed Union で結果を集めてから処理することになる。
要するにこのクエリは root-partititonable ではないため、失敗する。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="@{OPTIMIZER_VERSION=2} SELECT SingerId, ARRAY(SELECT AS STRUCT * FROM Concerts WHERE Singers.SingerId = Concerts.SingerId) FROM Singers" 
2024/10/02 02:23:40 rpc error: code = InvalidArgument desc = Queries with more than one non-Local Distributed Union or Distributed Apply are currently considered not partitionable. Please check the conditions for a query to be root-partitionable.
error details: name = Help desc = Conditions for a query to be root-partitionable. url = https://cloud.google.com/spanner/docs/reads#read_data_in_parallel

ちなみに INTERLEAVE 関係がある Singers と Albums でほぼ同じクエリを実行してみよう。

@{OPTIMIZER_VERSION=2}
SELECT SingerId,
ARRAY(SELECT AS STRUCT * FROM Albums WHERE Singers.SingerId = Albums.SingerId)
FROM Singers

実行計画は次のようになる。

+-----+-------------------------------------------------------------------------------+
| ID  | Query_Execution_Plan                                                          |
+-----+-------------------------------------------------------------------------------+
|   0 | Distributed Union (distribution_table: Singers, split_ranges_aligned: true)   |
|   1 | +- Local Distributed Union                                                    |
|   2 |    +- Serialize Result                                                        |
|   3 |       +- Table Scan (Full scan: true, Table: Singers, scan_method: Automatic) |
|   6 |       +- [Scalar] Array Subquery                                              |
|   7 |          +- Local Distributed Union                                           |
|   8 |             +- Compute Struct                                                 |
|   9 |                +- Filter Scan (seekable_key_size: 0)                          |
| *10 |                   +- Table Scan (Table: Albums, scan_method: Scalar)          |
+-----+-------------------------------------------------------------------------------+
Predicates(identified by ID):
 10: Seek Condition: ($SingerId_1 = $SingerId)

この実行計画には Distributed Union が1つしかないため、 root-partitionable に違反しない。当然、 Partition Query は成功する。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="@{OPTIMIZER_VERSION=2} SELECT SingerId, ARRAY(SELECT AS STRUCT * FROM Albums WHERE Singers.SingerId = Albums.SingerId) FROM Singers"             
success

1.c. Your query plan can't contain any other distributed operators, such as distributed cross apply.

他の分散オペレータが存在する場合は常に失敗する。

分散オペレータとはドキュメントによると5種類ある。

例えばこのクエリは前述のように INTERLEAVE 関係にない2つのテーブル Singers と Concerts を分散 JOIN することになる。

@{OPTIMIZER_VERSION=2}
SELECT *
FROM Singers
INNER JOIN Concerts USING (SingerId)

よって最上位が Distributed Union ではあるとしても実行計画には Distributed Cross Apply が含まれる。

+-----+--------------------------------------------------------------------------------------+
| ID  | Query_Execution_Plan                                                                 |
+-----+--------------------------------------------------------------------------------------+
|   0 | Distributed Union (distribution_table: Concerts, split_ranges_aligned: false)        |
|  *1 | +- Distributed Cross Apply                                                           |
|   2 |    +- [Input] Create Batch                                                           |
|   3 |    |  +- Local Distributed Union                                                     |
|   4 |    |     +- Compute Struct                                                           |
|   5 |    |        +- Table Scan (Full scan: true, Table: Concerts, scan_method: Automatic) |
|  19 |    +- [Map] Serialize Result                                                         |
|  20 |       +- Cross Apply                                                                 |
|  21 |          +- [Input] KeyRangeAccumulator                                              |
|  22 |          |  +- Batch Scan (Batch: $v2, scan_method: Scalar)                          |
|  29 |          +- [Map] Local Distributed Union                                            |
|  30 |             +- Filter Scan (seekable_key_size: 0)                                    |
| *31 |                +- Table Scan (Table: Singers, scan_method: Scalar)                   |
+-----+--------------------------------------------------------------------------------------+
Predicates(identified by ID):
  1: Split Range: ($SingerId = $SingerId_1)
 31: Seek Condition: ($SingerId = $batched_SingerId_1)

root-partitionable の要件に違反するため、これも当然 Partition Query に失敗する。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="@{OPTIMIZER_VERSION=2} SELECT * FROM Singers INNER JOIN Concerts USING (SingerId)"                                                      
2024/10/02 02:29:47 rpc error: code = InvalidArgument desc = Queries with more than one non-Local Distributed Union or Distributed Apply are currently considered not partitionable. Please check the conditions for a query to be root-partitionable.
error details: name = Help desc = Conditions for a query to be root-partitionable. url = https://cloud.google.com/spanner/docs/reads#read_data_in_parallel

ちなみに INTERLEAVE 関係にある Singers と Albums テーブルの場合はどうなるだろうか？

+-----+------------------------------------------------------------------------------------------+
| ID  | Query_Execution_Plan                                                                     |
+-----+------------------------------------------------------------------------------------------+
|   0 | Distributed Union (distribution_table: Singers, split_ranges_aligned: true)              |
|   1 | +- Local Distributed Union                                                               |
|   2 |    +- Serialize Result                                                                   |
|   3 |       +- Cross Apply                                                                     |
|   4 |          +- [Input] Table Scan (Full scan: true, Table: Singers, scan_method: Automatic) |
|  10 |          +- [Map] Local Distributed Union                                                |
|  11 |             +- Filter Scan (seekable_key_size: 0)                                        |
| *12 |                +- Table Scan (Table: Albums, scan_method: Scalar)                        |
+-----+------------------------------------------------------------------------------------------+
Predicates(identified by ID):
 12: Seek Condition: ($SingerId_1 = $SingerId)

これは分散 JOIN ではなくなるため、Distributed Cross Apply ではなく Cross Apply だけで JOIN が完結する。
明らかに root-partitionable の要件を満たすため、やはり Partition Query は成功する。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="@{OPTIMIZER_VERSION=2} SELECT * FROM Singers INNER JOIN Albums USING (SingerId)"  
success

INTERLEAVE の重要性は分散 JOIN と分散コミットだけでも余りあるが、 1.b と合わせて Partition Query および Data Boost の実行可能性にも関わってくることが分かる。

2.a. There are no distributed operators in the query plan.

最後に、分散オペレータが一切存在しない場合は Partition Query で実行可能であると書かれている。これは何を意味しているだろうか？
データにアクセスするクエリは全て、それぞれの shard を持つサーバに処理を分散するための分散オペレータが必要となる。逆に言えば、データにアクセスしないクエリは分散オペレータが不要だ。

spanner> EXPLAIN SELECT 1;
+----+----------------------+
| ID | Query_Execution_Plan |
+----+----------------------+
|  0 | Serialize Result     |
|  1 | +- Unit Relation     |
+----+----------------------+

このクエリは明らかに分散オペレータが存在しないという条件を満たす。よって、 Partition Query で処理可能である。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="@{OPTIMIZER_VERSION=2} SELECT 1"                                                
success

実はこの条件は2023年12月19日のリリースノートに書かれているようにあとから追加されたものだ。

Cloud Spanner now supports partition queries whose query plans don't contain any distributed unions. To learn more about how to read data in parallel using partition queries, see Read data in parallel.

分散する処理が存在しないクエリの分散処理が失敗しないようになった喜びを噛み締めよう。

何の役に立つかは説明されては居ないが、データベーススキーマに関わらず成功するのでテスト用途で有用な気がする。実際有用だった。

落穂拾い

ここまででドキュメンテーションされた root-partitionable の要件を全て見ていったが、この条件だけでは説明できない挙動がいくつかある。

隠れた Distributed Merge Union が root-partitionable ではないのでは？

1.a. で ORDER BY を使った例の optimizer version の指定を外した次のようなクエリを想定する。

SELECT *
FROM Singers
ORDER BY FirstName

実行計画は次のようになる。

+----+-------------------------------------------------------------------------------------------------------------+
| ID | Query_Execution_Plan                                                                                        |
+----+-------------------------------------------------------------------------------------------------------------+
|  0 | Distributed Union (distribution_table: Singers, preserve_subquery_order: true, split_ranges_aligned: false) |
|  1 | +- Serialize Result                                                                                         |
|  2 |    +- Sort                                                                                                  |
|  3 |       +- Local Distributed Union                                                                            |
|  4 |          +- Table Scan (Full scan: true, Table: Singers, scan_method: Automatic)                            |
+----+-------------------------------------------------------------------------------------------------------------+

なんと、ソートがあるのに Distributed Union が最上位になった。
これはドキュメンテーションされた root-partitionable の条件を明らかに満たすので成功することが期待される。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="SELECT * FROM Singers ORDER BY FirstName"                
2024/10/02 02:46:45 rpc error: code = InvalidArgument desc = Query is not root partitionable since it does not have a DistributedUnion at the root. Please check the conditions for a query to be root-partitionable.
error details: name = Help desc = Conditions for a query to be root-partitionable. url = https://cloud.google.com/spanner/docs/reads#read_data_in_parallel

しかし、失敗する。エラーを見てもなぜ失敗したかは分からない。

これは何故なのかと言うと、このソートの処理は実際には Distributed Merge Union で処理されていて普通の Distributed Union と異なると考えられる。
Distributed Merge Union は query optimizer v3 で追加されたため、 v2 指定を外したことで使われるようになったということで説明できる。

その証拠に、Distributed Merge Union を使用しないように制御するヒントである ALLOW_DISTRIBUTED_MERGE=FALSE 指定するだけで、実行計画は元の Distributed Union の上に Sort がある形に戻る。

@{ALLOW_DISTRIBUTED_MERGE=FALSE}
SELECT *
FROM Singers
ORDER BY FirstName

+----+------------------------------------------------------------------------------------+
| ID | Query_Execution_Plan                                                               |
+----+------------------------------------------------------------------------------------+
|  0 | Serialize Result                                                                   |
|  1 | +- Sort                                                                            |
|  2 |    +- Distributed Union (distribution_table: Singers, split_ranges_aligned: false) |
|  3 |       +- Local Distributed Union                                                   |
|  4 |          +- Table Scan (Full scan: true, Table: Singers, scan_method: Automatic)   |
+----+------------------------------------------------------------------------------------+

ドキュメントにはこれが Merge Distributed Union であるとは明記されていないが、マージも行うため単純に集めるだけの処理にはなっておらず、 partition ごとに集めれば良いだけの root-partitionable ではなくなるのではないか。

おそらく失敗に直接関わっているのは preserve_subquery_order: true な気がしている。

|  0 | Distributed Union (distribution_table: Singers, preserve_subquery_order: true, split_ranges_aligned: false) |

通常のクエリと異なる実行計画が立てられる場合があるのでは？

以前見つけた例。

SELECT *                                 
FROM Singers
WHERE SingerId IN (
  SELECT SingerId
  FROM Albums
  WHERE AlbumTitle > ''
)

実行計画は次のようになる。

+-----+-----------------------------------------------------------------------------------------------------+
| ID  | Query_Execution_Plan                                                                                |
+-----+-----------------------------------------------------------------------------------------------------+
|  *0 | Distributed Cross Apply                                                                             |
|   1 | +- [Input] Create Batch                                                                             |
|   2 | |  +- Compute Struct                                                                                |
|   3 | |     +- Global Hash Aggregate                                                                      |
|  *4 | |        +- Distributed Union (distribution_table: AlbumsByAlbumTitle, split_ranges_aligned: false) |
|   5 | |           +- Local Hash Aggregate                                                                 |
|   6 | |              +- Local Distributed Union                                                           |
|   7 | |                 +- Filter Scan (seekable_key_size: 1)                                             |
|  *8 | |                    +- Index Scan (Index: AlbumsByAlbumTitle, scan_method: Scalar)                 |
|  22 | +- [Map] Serialize Result                                                                           |
|  23 |    +- Cross Apply                                                                                   |
|  24 |       +- [Input] KeyRangeAccumulator                                                                |
|  25 |       |  +- Batch Scan (Batch: $v2, scan_method: Scalar)                                            |
|  27 |       +- [Map] Local Distributed Union                                                              |
|  28 |          +- Filter Scan (seekable_key_size: 0)                                                      |
| *29 |             +- Table Scan (Table: Singers, scan_method: Scalar)                                     |
+-----+-----------------------------------------------------------------------------------------------------+
Predicates(identified by ID):
  0: Split Range: ($SingerId = $group_SingerId_1')
  4: Split Range: ($AlbumTitle > '')
  8: Seek Condition: ($AlbumTitle > '')
 29: Seek Condition: ($SingerId = $batched_SingerId_1)

Distributed Cross Apply が最上位にあるため明らかに root-partitionable ではないように見える。

しかしこのクエリは Partition Query に成功する。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="SELECT *                                 
FROM Singers
WHERE SingerId IN (
  SELECT SingerId
  FROM Albums
  WHERE AlbumTitle > ''
)"
success

何故？仮説としては、オプティマイザはいくつかの実行計画の候補がある時に、 Partition Query の時は root-partitionable になることを優先し、それ以外の場合は root-partitionable ではなくても効率的に実行できそうなことを優先するからではないだろうか。
オプティマイザが選べる実行計画の選択肢を減らすため、 FORCE_INDEX ヒントを利用して考察する。

Albums でインデックスを使わずにベーステーブルを使うことを強制してみる。

SELECT *
FROM Singers
WHERE SingerId IN (
  SELECT SingerId
  FROM Albums@{FORCE_INDEX=_BASE_TABLE}
  WHERE AlbumTitle > ""
)

実行計画は次のようになり、 Distributed Cross Apply が消える。

+-----+---------------------------------------------------------------------------------------+
| ID  | Query_Execution_Plan                                                                  |
+-----+---------------------------------------------------------------------------------------+
|   0 | Distributed Union (distribution_table: Singers, split_ranges_aligned: true)           |
|   1 | +- Serialize Result                                                                   |
|   2 |    +- Cross Apply                                                                     |
|   3 |       +- [Input] Stream Aggregate                                                     |
|   4 |       |  +- Local Distributed Union                                                   |
|  *5 |       |     +- Filter Scan (seekable_key_size: 0)                                     |
|   6 |       |        +- Table Scan (Full scan: true, Table: Albums, scan_method: Automatic) |
|  14 |       +- [Map] Local Distributed Union                                                |
|  15 |          +- Filter Scan (seekable_key_size: 0)                                        |
| *16 |             +- Table Scan (Table: Singers, scan_method: Scalar)                       |
+-----+---------------------------------------------------------------------------------------+
Predicates(identified by ID):
  5: Residual Condition: ($AlbumTitle > '')
 16: Seek Condition: ($SingerId = $group_SingerId_1)

この実行計画は明らかに root-partitionable であるため、 Parittion Query は成功する。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="SELECT *
FROM Singers
WHERE SingerId IN (
  SELECT SingerId
  FROM Albums@{FORCE_INDEX=_BASE_TABLE}
  WHERE AlbumTitle > ''
)"
success

逆に、 Albums の INTERLEAVE ではないインデックスを使うように強制してみる。

SELECT *
FROM Singers
WHERE SingerId IN (
  SELECT SingerId
  FROM Albums@{FORCE_INDEX=AlbumsByAlbumTitle}
  WHERE AlbumTitle > ''
)

+-----+-----------------------------------------------------------------------------------------------------+
| ID  | Query_Execution_Plan                                                                                |
+-----+-----------------------------------------------------------------------------------------------------+
|  *0 | Distributed Cross Apply                                                                             |
|   1 | +- [Input] Create Batch                                                                             |
|   2 | |  +- Compute Struct                                                                                |
|   3 | |     +- Global Hash Aggregate                                                                      |
|  *4 | |        +- Distributed Union (distribution_table: AlbumsByAlbumTitle, split_ranges_aligned: false) |
|   5 | |           +- Local Hash Aggregate                                                                 |
|   6 | |              +- Local Distributed Union                                                           |
|   7 | |                 +- Filter Scan (seekable_key_size: 1)                                             |
|  *8 | |                    +- Index Scan (Index: AlbumsByAlbumTitle, scan_method: Scalar)                 |
|  22 | +- [Map] Serialize Result                                                                           |
|  23 |    +- Cross Apply                                                                                   |
|  24 |       +- [Input] KeyRangeAccumulator                                                                |
|  25 |       |  +- Batch Scan (Batch: $v2, scan_method: Scalar)                                            |
|  27 |       +- [Map] Local Distributed Union                                                              |
|  28 |          +- Filter Scan (seekable_key_size: 0)                                                      |
| *29 |             +- Table Scan (Table: Singers, scan_method: Scalar)                                     |
+-----+-----------------------------------------------------------------------------------------------------+
Predicates(identified by ID):
  0: Split Range: ($SingerId = $group_SingerId_1')
  4: Split Range: ($AlbumTitle > '')
  8: Seek Condition: ($AlbumTitle > '')
 29: Seek Condition: ($SingerId = $batched_SingerId_1)

最初の実行計画と全く同じに戻った。Distributed Cross Apply が最上位にあるため、 root-partitionable の要件を満たさない。このクエリはちゃんと期待通り Partition Query に失敗する。

$ execspansql ${SPANNER_DATABASE} --try-partition-query --sql="SELECT *
FROM Singers
WHERE SingerId IN (
  SELECT SingerId
  FROM Albums@{FORCE_INDEX=AlbumsByAlbumTitle}
  WHERE AlbumTitle > ''
)"
2024/10/02 03:14:21 rpc error: code = InvalidArgument desc = Query is not root partitionable since it does not have a DistributedUnion at the root. Please check the conditions for a query to be root-partitionable.
error details: name = Help desc = Conditions for a query to be root-partitionable. url = https://cloud.google.com/spanner/docs/reads#read_data_in_parallel

ヒントがない最初のクエリは、普通に実行計画を立てた時はインデックスを使う実行計画になっているが、 Partition Query を実行する場合だけインデックスを使わないことで root-partitionable を満たすような実行計画を選ぶというようなことが行われているという仮説で矛盾していないのではないだろうか？

雑なまとめ

多分最初の期待よりも実行できないクエリは多い。
実行計画に関することは実際に実行計画を見て議論しよう。
Cloud Spanner が複雑になるにつれ root-partitionable の真の要件は変わっている。
- ドキュメントが間違っていると思ったらフィードバックするべき。
Partition Query でだけ実行計画が変わっていると思われることがある。
- Parititon Query 用の PLAN オプションがほしい。
- PLAN では root-partitioned なものが失敗することはないので別に良いかもしれない。
INTERLEAVE は本当に重要。
(追記:2024-10-04) BigQuery から Cloud Spanner のテーブルデータにアクセスしたいなら Spanner External Dataset 使えば root-partitionable を気にする必要もなく Data Boost も使われるので楽そう。

前提知識

root-partitionable とは何か？

論文中の root-partitionable の定義

インスタンスに負荷を掛けずに Partition Query 可能かどうかを確認する方法

root-partitionable のそれぞれの条件を考察する

1.a. The first operator in the query execution plan is a distributed union

1.b. The query execution plan only contains one distributed union (excluding Local Distribution Unions)

1.c. Your query plan can't contain any other distributed operators, such as distributed cross apply.

2.a. There are no distributed operators in the query plan.

落穂拾い

隠れた Distributed Merge Union が root-partitionable ではないのでは？

通常のクエリと異なる実行計画が立てられる場合があるのでは？

雑なまとめ

Discussion