二人の書き手が同じテーブルに同時にコミットする — 誰が勝ち、何が残り、何が失われるか

目標

pyicebergで、2人のライターが同じmetadataを読んだあと順番にコミットする競合を、確実に再現します。append(追加)は自動リトライで両方入り、リトライを切ると片方が失敗して、書きかけのファイルが孤立ファイルとして残ることを確認します。条件付きの上書き(overwrite)は、リトライしても検証に引っかかり、そのあと新しく読んで計算し直さないと更新の喪失が起きることを、カウンター1つで確認します。

なぜ重要なのか

Icebergにはロックがありません。ライターは、それぞれファイルをすべて書き終えて新しいmetadataを作ったあと、カタログで「自分が読んだmainがまだそのままなら書き換える」という条件付き置換を試みます。2人が同時に来ると1人だけが勝ち、負けた側は新しいmetadataを読み直して、自分の変更を載せ直します。これが楽観的同時実行制御です。ほとんどはぶつからないと信じて、ぶつかったらやり直します。 問題は、「載せ直してよい変更」と「そうしてはいけない変更」があることです。新しいファイルを加えることは、間に誰が何をしていても、載せ直してかまいません。ところが「この条件に合う行をこの値に変える」という変更は、その間に誰かが同じ条件の行を変えていたら、間違った結果になります。ライブラリはファイルレベルでそれを検証して拒否してくれますが、皆さんのコードが以前に読んだ値で計算し直さずにそのまま書き直すと、誰も止めてくれません。

ステップ

1. /root/ice/conc/common.pyに1日分のCSVをpyarrowで読むヘルパーを置き(order_tsはUTCタイムゾーン)、/root/ice/conc/setup.pyでlake.conc.ordersを作成して2026-03-01を入れてください。
2. /root/ice/conc/race.pyで、2つのTableオブジェクトa・bを先に両方読んでから、aが03-02を、bが03-03を追加するようにし、結果を、/root/ice/conc/out/race.jsonに書いてください。
3. /root/ice/conc/noretry.pyでテーブルプロパティcommit.retry.num-retriesを0にして、同じ競合(03-04・03-05)をもう一度起こしたあと、負けた側の例外名を、/root/ice/conc/out/noretry.txtに書いてください。
4. /root/ice/conc/orphans.pyで、テーブルの場所のdataディレクトリにあるのに、どのスナップショットも指していないParquetファイルを探して、/root/ice/conc/out/orphans.jsonに書いてください。
5. /root/ice/conc/conflict.pyでカウンターテーブルlake.conc.counters(hits = 10)を作成し、2人のワーカーが同じ値を読んだあとそれぞれ+5を上書きするようにして、負けた側の例外名を、/root/ice/conc/out/conflict.txtに書いてください。
6. /root/ice/conc/retry.pyで、負けた側の+5を正しくやり直し、カウンターを20にしてください。
7. /root/ice/conc/report.mdに、## 자동 재시도、## 실패한 커밋의 흔적、## 잃어버린 갱신の3つの節を書いてください(3つの見出しは順に、韓国語で「自動リトライ」「失敗したコミットの痕跡」「更新の喪失」を意味する語句です)。

参考

• スクリプトはcd /root/ice/conc && python3 race.pyのように実行します(from common import …)。
• このラボのコミットは、すべてpyicebergです。pyicebergは、条件付き置換に負けると、テーブルプロパティのリトライ設定の回数だけ、新しく読んで再試行し、そのとき「Commit failed due to a concurrent update, retrying」を出力します。
• 例外名はtype(exc).__name__で得られます。
• よくある間違いは、ステップ6で例外を捕まえたあと、以前に読んだ値(10)に5を足した15を書き直すことです。コミットは成功しますが、カウンターは15で止まります。ステップ1–4を最初からやり直すには、spark-sql -e "DROP TABLE lake.conc.orders PURGE"を実行してから、ステップ1に戻ってください。
• 公式ドキュメント: Reliability — Concurrent write operations・Configuration — Table behavior properties・Spec — Optimistic Concurrency・pyiceberg — Write support

Pythonで作ったテーブル

/root/ice/conc/common.pyに、day("YYYY-MM-DD")がその日のCSVをpyarrowテーブルとして返すようにし(amountはint32、order_tsはUTCタイムゾーンのtimestamp)、/root/ice/conc/setup.pyでlake.conc.ordersをformat-version 2で作成して、2026-03-01を入れてください。

pyicebergのcreate_table(이름, schema=pyarrow_스키마)(プレースホルダーは順に、名前とpyarrowスキーマです)は、列ごとにフィールドIDを振ります。時刻にタイムゾーンを付けると、Sparkが作るテーブルと同じtimestamptzになります。採点ツールは、最初のスナップショットが3月1日の行数を加えたものであるかを確認します。

競合: appendは載せ直せばよい

/root/ice/conc/race.pyでa = load_table(…)、b = load_table(…)を両方先に作ってから、a.append(03-02)、b.append(03-03)の順にコミットし、スナップショット数と行数を、/root/ice/conc/out/race.jsonに{"snapshots", "rows"}の形で書いてください。

bはaのコミット前のmetadataを持っているので、最初のコミット試行が条件に引っかかります。appendは、間に何が入っていてもその上に載せ直してかまわないので、pyicebergが新しく読んで再試行し、成功します。採点ツールは、3つのスナップショットが1本の線でつながっているか(分岐していないか)と、行数を確認します。

リトライを切ると、負けた側は失敗する

/root/ice/conc/noretry.pyでlake.conc.ordersのプロパティcommit.retry.num-retriesを"0"に変えたあと、ステップ2のようにa・bを先に作って、aが03-04、bが03-05を追加するようにしてください。bの例外名を、/root/ice/conc/out/noretry.txtの1行目に書いてください。

リトライが0なら、条件付き置換に負けた瞬間に例外が上がります。ところがbは、コミットを試みる前に、データファイルをすでにすべて書き終えています。採点ツールは、例外名とプロパティの値、3月4日は入っていて3月5日はテーブルにないことを確認します。

失敗したコミットが残したファイル

/root/ice/conc/orphans.pyで、すべてのスナップショットが指すファイル一覧と、テーブルの場所(t.location())の下のdataディレクトリにある実際のParquetファイルを比較し、一覧にないファイル(孤立ファイル)のパスを、/root/ice/conc/out/orphans.jsonに{"orphans": [경로, …]}(プレースホルダーはパスです)の形で書いてください。

孤立ファイルはテーブルの一部ではないので読まれませんが、容量を占めます。現在のスナップショットではなく、すべてのスナップショットが指すファイルと比較する必要があります。古いスナップショットのファイルは、タイムトラベル用に生きているファイルです。このようなファイルを片づけるのが、次のモジュールのremove_orphan_filesです。

上書きは検証に引っかかる

/root/ice/conc/conflict.pyでlake.conc.counters(name STRING, value BIGINT)をhits = 10の1行で(あれば削除して)新しく作り、a・bが両方値を読んだあと、aが읽은 값 + 5(韓国語で「読んだ値に5を足したもの」という意味の式です)をoverwrite(…, overwrite_filter=EqualTo("name", "hits"))で書き、bも同じように書くようにしてください。bの例外名を、/root/ice/conc/out/conflict.txtの1行目に書いてください。

bの最初の試行は条件付き置換に負け、リトライでは「その間に自分の条件(name = hits)に合うファイルが新しく入ってきた」という検証に引っかかって止まります。appendと違って、上書きはそのまま載せ直すとaの結果を消してしまうからです。採点ツールは、例外名と、カウンターが一時15だったスナップショットがあるかを確認します。

新しく読んで計算し直す

/root/ice/conc/retry.pyで、bの+5をやり直してください。試行のたびにテーブルを新しく読み、そのとき読んだ値に5を足して条件付き上書きを試み、失敗したら最初からやり直します。終わったら、hitsは20でなければなりません。

ライブラリのリトライは、コミットを載せ直すだけで、皆さんが以前に読んだ値を読み直してはくれません。以前に読んだ10で計算した15をそのまま書き直すと、コミットは成功し、aの+5が消えます。これが更新の喪失です。採点ツールは、カウンターがちょうど20であるかを確認します。

同時書き込みのルールをチームのルールにする

/root/ice/conc/report.mdに、## 자동 재시도、## 실패한 커밋의 흔적、## 잃어버린 갱신の3つの節を書いてください(3つの見出しは順に、韓国語で「自動リトライ」「失敗したコミットの痕跡」「更新の喪失」を意味する語句です)。2つ目の節にはステップ4で見つけた孤立ファイル数を、3つ目の節には最終的なカウンターの値を、数字で入れてください。

同じテーブルに書き込むジョブが2つ以上あるなら、どのジョブはリトライに任せてよく、どのジョブは読み取りからやり直すべきか、失敗したジョブのファイルは誰がいつ片づけるのかを、書いてみてください。