BigQueryのトランザクションについて掘り下げて検証してみた

BigQueryのトランザクションについて調べる機会があったので、ドキュメントを読みつつ実際に検証してみました。

BigQueryのトランザクションについて

BigQueryのトランザクションはスナップショット分離をサポートしており、実装としてMVCC（Multi-version Concurrency Control; マルチバージョン並行性制御）を採用しています。トランザクション分離レベルについては、「データ指向アプリケーションデザイン」第7章（トランザクション）に詳細かつわかりやすく説明されているため本記事では省略しますが、スナップショット分離はそれぞれのトランザクションがデータベースの一貫性のあるスナップショットから読み取りを行います。トランザクション内で読み取るデータは、そのトランザクション開始時点でデータベースにコミットされているものとなります。そのため、別トランザクションでデータ更新された際のデータ不整合の問題（ノンリピータブルリード）が起こりません。また、スナップショット分離はデータ書き込み時にロックはかけますが読み取りはロックをとらないため、バックアップや分析（長時間実行されるクエリ）に向いています。

Google Cloudブログによると、BigQueryはテーブルロックではなく、楽観的並行性制御という方法を使用します。トランザクション開始時点で排他処理をせず、DMLの処理結果をコミットする前にこのテーブル内で行われた変更内容と競合しないかを確認し、競合があった場合にはリトライされるとのことです。

DMLによる競合は、同じパーティションで変更（UPDATE、MERGE）しようとした場合に発生します。ほかのパーティションにおける変更や、データ追加（INSERT）は競合しません。

検証してみた

以下のパターンにおいて挙動を調べてみました。

1. 長時間のデータ更新中にデータ取得した場合
   i. データ更新中のトランザクションなし
   ii. データ更新中のトランザクションあり
2. トランザクション中、長時間のデータ取得中にデータ更新した場合
3. 同じパーティション内で異なるテーブル行での書き込みをする場合
4. 異なるパーティションでテーブル行での書き込みをする場合
5. トランザクション中、データ更新中にデータ追加する場合

準備

USリージョンで、データセット analytics とテーブル test-table1 を作成します。

CREATE OR REPLACE TABLE `analytics.test-table1` (
  id INT64,
  name STRING,
  updated_at TIMESTAMP
);
INSERT INTO `analytics.test-table1` (id, name) VALUES (1, 'foo');
INSERT INTO `analytics.test-table1` (id, name) VALUES (2, 'bar');

$ bq query \
  --use_legacy_sql=false \
  'SELECT * FROM `analytics.test-table1` ORDER BY id'
+----+------+------------+
| id | name | updated_at |
+----+------+------------+
|  1 | foo  |       NULL |
|  2 | bar  |       NULL |
+----+------+------------+

Sleep処理について

トランザクション処理を検証するうえでSleep処理を入れたいのですが、BigQueryではSleep関数が用意されていません。プロシージャを定義することでSleep処理を入れることはできますが、処理内で用いている CURRENT_TIMESTAMP 関数はトランザクション開始時点のタイムスタンプを返すため、意図した動作になりません（永遠にループします）。

The CURRENT_TIMESTAMP, CURRENT_DATE, and CURRENT_TIME functions return the timestamp when the transaction started.
https://cloud.google.com/bigquery/docs/reference/standard-sql/transactions#datetime_functions_in_transactions

そのため、原始的な方法ですが以下のように変数 x をひたすらインクリメントすることで時間を稼ぎます。おおよそ3000ぐらいインクリメントするのに1分ぐらいかかります。

BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
END;

1. 長時間のデータ更新中にデータ取得した場合

長期間、UPDATEなどでデータ更新している間に、データ取得した場合の挙動について確認します。想定されるケースとしては、長時間のETL処理している最中にデータ分析する場合です。

i. データ更新中のトランザクションなし

テーブル test-table1 に対してデータ更新した後、約1分間Sleep状態にします。この処理にはトランザクションを貼りません。

BEGIN
  DECLARE x INT64 DEFAULT 0;
  UPDATE `analytics.test-table1` SET name="hogehoge", updated_at=CURRENT_TIMESTAMP() WHERE id=1;
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
END;

Sleepしている間に test-table1 の中身を確認すると、（当たり前ですが）データが更新されています。

$ bq query \
  --use_legacy_sql=false \
  'SELECT * FROM `analytics.test-table1` ORDER BY id'
+----+----------+---------------------+
| id |   name   |     updated_at      |
+----+----------+---------------------+
|  1 | hogehoge | 2022-06-24 11:04:26 |
|  2 | bar      |                NULL |
+----+----------+---------------------+

次の検証の前にデータを戻しておきます。

UPDATE `analytics.test-table1` SET name="foo", updated_at=null WHERE id=1;

ii. データ更新中のトランザクションあり

BEGIN TRANSACTION ~ COMMIT TRANSACTION でトランザクションを挟むようにします。

BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- トランザクション開始
  BEGIN TRANSACTION;
  UPDATE `analytics.test-table1` SET name="hogehoge", updated_at=CURRENT_TIMESTAMP() WHERE id=1;
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
  -- コミット
  COMMIT TRANSACTION;
END;

Sleepしている間に test-table1 の中身を確認します。コミット前かつトランザクション外からのデータ参照であるため、反映前のデータが取得できました。

# Sleep中（コミット前）
$ bq query \
  --use_legacy_sql=false \
  'SELECT * FROM `analytics.test-table1` ORDER BY id'
+----+------+------------+
| id | name | updated_at |
+----+------+------------+
|  1 | foo  |       NULL |
|  2 | bar  |       NULL |
+----+------+------------+

コミット後に再度実行すると、反映後のデータが返ってきました。

# Sleep後（コミット後）
$ bq query \
  --use_legacy_sql=false \
  'SELECT * FROM `analytics.test-table1` ORDER BY id'
+----+----------+---------------------+
| id |   name   |     updated_at      |
+----+----------+---------------------+
|  1 | hogehoge | 2022-06-24 11:06:17 |
|  2 | bar      |                NULL |
+----+----------+---------------------+

次の検証の前にデータを戻しておきます。

UPDATE `analytics.test-table1` SET name="foo", updated_at=null WHERE id=1;

2. トランザクション中、長時間のデータ取得中にデータ更新した場合

BigQueryのトランザクションはスナップショット分離であると冒頭に説明しました。スナップショット分離はノンリピータブルリード（Non-Repeatable Read）あるいは読み取りスキュー（Read Skew）を防ぎます。つまり、トランザクション内でレコードを取得した後に別のトランザクションで同じレコードを更新・コミットした場合に、2回目以降のレコード取得も同じ結果を返します。本当にスナップショット分離が実装されているかを確認します。

以下のクエリ（トランザクションA）の実行中に、もう一つのトランザクションB（UPDATE）を実行します。トランザクションA内で取得するデータが変わっていないことを確認します。

-- トランザクションA
BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- トランザクション開始
  BEGIN TRANSACTION;
  -- データを取得
  SELECT * FROM `analytics.test-table1` ORDER BY id;
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
  -- 再度データを取得
  -- 読み取りスキューが発生するとデータが変わってしまう
  SELECT * FROM `analytics.test-table1` ORDER BY id;
  -- トランザクションのコミット
  COMMIT TRANSACTION;
END;

-- トランザクションB
BEGIN
  -- トランザクション開始
  BEGIN TRANSACTION;
  UPDATE `analytics.test-table1` SET name="hogehoge", updated_at=CURRENT_TIMESTAMP() WHERE id=1;
  -- トランザクションのコミット
  COMMIT TRANSACTION;
END;

1回目（UPDATE実行前）、2回目（UPDATE実行後）ともに同じ結果が返ってきました。つまり、トランザクション内であればデータは一貫していることが保証されていることが確認できました。

$ bq query \
  --use_legacy_sql=false \
  'SELECT * FROM `analytics.test-table1` ORDER BY id'
+----+------+------------+
| id | name | updated_at |
+----+------+------------+
|  1 | foo  |       NULL |
|  2 | bar  |       NULL |
+----+------+------------+

次の検証の前にデータを戻しておきます。

UPDATE `analytics.test-table1` SET name="foo", updated_at=null WHERE id=1;

3. 同じパーティション内で異なるテーブル行での書き込みをする場合

以下のクエリ（トランザクションA）によるデータ更新とは別のトランザクションBでデータ更新をした場合の挙動を確認します。

-- トランザクションA
BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- トランザクション開始
  BEGIN TRANSACTION;
  UPDATE `analytics.test-table1` SET name="hogehoge", updated_at=CURRENT_TIMESTAMP() WHERE id=1;
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
  -- トランザクションのコミット
  COMMIT TRANSACTION;
END;

-- トランザクションB
BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- トランザクション開始
  BEGIN TRANSACTION;
  -- トランザクションAとは異なる行を更新
  UPDATE `analytics.test-table1` SET name="fugafuga", updated_at=CURRENT_TIMESTAMP() WHERE id=2;
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
  -- トランザクションのコミット
  COMMIT TRANSACTION;
END;

トランザクションAを先に実行した後トランザクションBを実行すると、UPDATEの箇所で以下のエラーになりました。同じテーブルかつ同じパーティション内の更新なのでトランザクションがキャンセルされたようです。

Query error: Transaction is aborted due to concurrent update against table analytics.test-table1. Transaction ID: 62b3ae6d-0000-297d-abbd-24058872ade0 at [6:3]

4. 異なるパーティションでテーブル行での書き込みをする場合

updated_at カラムをパーティション分割する test-table2 テーブルを作成します。

CREATE OR REPLACE TABLE analysis.test-table2 (
  id INT64,
  name STRING,
  updated_at TIMESTAMP
)
PARTITION BY
  DATETIME_TRUNC(updated_at, DAY)
;
INSERT INTO `analytics.test-table2` (id, name, updated_at) VALUES (1, 'foo', TIMESTAMP("2022-06-27 00:00:00+00"));
INSERT INTO `analytics.test-table2` (id, name, updated_at) VALUES (2, 'bar', TIMESTAMP("2022-06-28 00:00:00+00"));

パーティション毎にデータが入っているのを確認できます。

$ bq query \
  --use_legacy_sql=false \
  'SELECT
     table_name, partition_id, total_rows
   FROM analytics.INFORMATION_SCHEMA.PARTITIONS
   WHERE table_name = "test-table2"'
+-------------+--------------+------------+
| table_name  | partition_id | total_rows |
+-------------+--------------+------------+
| test-table2 | 20220627     |          1 |
| test-table2 | 20220628     |          1 |
+-------------+--------------+------------+

「3. 同じパーティション内で異なるテーブル行での書き込みをする場合」と同じように実行してみます。

-- トランザクションA
BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- トランザクション開始
  BEGIN TRANSACTION;
  UPDATE `analytics.test-table2` SET name="hogehoge" WHERE DATE(updated_at) = "2022-06-27";
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
  -- トランザクションのコミット
  COMMIT TRANSACTION;
END;

-- トランザクションB（トランザクションAの後に実行）
BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- トランザクション開始
  BEGIN TRANSACTION;
  -- トランザクションAとは異なる行を更新
  UPDATE `analytics.test-table2` SET name="fugafuga" WHERE DATE(updated_at) = "2022-06-28";
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
  -- トランザクションのコミット
  COMMIT TRANSACTION;
END;

結果は先ほどと同じエラーになってしまいました。異なるパーティションならデータ更新は成功すると思っていたので想定と異なります。

試しに、トランザクションを張らずに以下のクエリのみ実行した場合、トランザクションAの完了まで待機し、完了後に実行されました。

UPDATE `analytics.test-table2` SET name="fugafuga" WHERE DATE(updated_at) = "2022-06-28";

誰かわかる人いたらコメントください。

5. トランザクション中、データ更新中にデータ追加する場合

追記（INSERT）はDML競合しないとのことなので検証します。

先ほどと同じクエリ（トランザクションA）の実行中にトランザクションB（INSERT）を実行します。

-- トランザクションA
BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- トランザクション開始
  BEGIN TRANSACTION;
  UPDATE `analytics.test-table1` SET name="hogehoge", updated_at=CURRENT_TIMESTAMP() WHERE id=1;
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
  -- トランザクションのコミット
  COMMIT TRANSACTION;
END;

-- トランザクションB（INSERT）
BEGIN
  DECLARE x INT64 DEFAULT 0;
  -- トランザクション開始
  BEGIN TRANSACTION;
  -- データ追加
  INSERT INTO `analytics.test-table1` (id, name, updated_at) VALUES (3, 'komekome', CURRENT_TIMESTAMP());
  -- 約1分間、Sleep状態にする
  REPEAT
    SET x = x + 1;
    UNTIL x >= 3000
  END REPEAT;
  -- トランザクションのコミット
  COMMIT TRANSACTION;
END;

トランザクションA、Bともに実行できてコミット後に反映されていました。

$ bq query \
  --use_legacy_sql=false \
  'SELECT * FROM `analytics.test-table1` ORDER BY id'
+----+----------+---------------------+
| id |   name   |     updated_at      |
+----+----------+---------------------+
|  1 | hogehoge | 2022-06-25 08:45:45 |
|  2 | bar      |                NULL |
|  3 | komekome | 2022-06-25 08:45:59 |
+----+----------+---------------------+

参考URL

• Multi-statement transactions  |  BigQuery  |  Google Cloud
• Using data manipulation language (DML)  |  BigQuery  |  Google Cloud
• BigQuery で DML が無制限で利用可能に | Google Cloud Blog
• データ指向アプリケーションデザイン | Amazon
• Google Cloudではじめる実践データエンジニアリング入門 | Amazon
  
• BigQuery でトランザクション処理がサポート開始！！ | DevelopersIO