From 6178ed5115cb17b3ced7883c05f9025c68dfdcbe Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 27 May 2026 14:18:21 -0400 Subject: [PATCH 01/12] SS-148 repro from Dennis --- test/iceberg/idempotent-retry-setup.td | 40 +++++++ test/iceberg/idempotent-retry-verify.td | 16 +++ test/iceberg/mzcompose.py | 69 ++++++++++- test/iceberg/polaris_proxy.py | 145 ++++++++++++++++++++++++ 4 files changed, 267 insertions(+), 3 deletions(-) create mode 100644 test/iceberg/idempotent-retry-setup.td create mode 100644 test/iceberg/idempotent-retry-verify.td create mode 100644 test/iceberg/polaris_proxy.py diff --git a/test/iceberg/idempotent-retry-setup.td b/test/iceberg/idempotent-retry-setup.td new file mode 100644 index 0000000000000..8ad4c2279251c --- /dev/null +++ b/test/iceberg/idempotent-retry-setup.td @@ -0,0 +1,40 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +> CREATE SECRET access_key_secret AS '${arg.s3-access-key}' + +> CREATE CONNECTION aws_conn TO AWS ( + ACCESS KEY ID = 'tduser', + SECRET ACCESS KEY = SECRET access_key_secret, + ENDPOINT = '${arg.aws-endpoint}', + REGION = 'us-east-1' + ); + +> CREATE CONNECTION polaris TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'http://polaris-proxy:8181/api/catalog', + CREDENTIAL = 'root:root', + WAREHOUSE = 'default_catalog', + SCOPE = 'PRINCIPAL_ROLE:ALL' + ); + +> CREATE TABLE retry_src (id INT, val TEXT); + +> INSERT INTO retry_src VALUES (1, 'a'), (2, 'b'), (3, 'c'); + +> CREATE SINK retry_sink + FROM retry_src + INTO ICEBERG CATALOG CONNECTION polaris ( + NAMESPACE 'default_namespace', + TABLE 'retry_table' + ) + USING AWS CONNECTION aws_conn + KEY (id) NOT ENFORCED + MODE UPSERT + WITH (COMMIT INTERVAL '2s'); diff --git a/test/iceberg/idempotent-retry-verify.td b/test/iceberg/idempotent-retry-verify.td new file mode 100644 index 0000000000000..c54f765eca88f --- /dev/null +++ b/test/iceberg/idempotent-retry-verify.td @@ -0,0 +1,16 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +$ duckdb-execute name=iceberg +CREATE SECRET s3_secret (TYPE S3, KEY_ID 'tduser', SECRET '${arg.s3-access-key}', ENDPOINT '${arg.aws-endpoint}', URL_STYLE 'path', USE_SSL false, REGION 'minio'); +SET unsafe_enable_version_guessing = true; + +$ duckdb-query name=iceberg +SELECT COUNT(*) FROM iceberg_scan('s3://test-bucket/default_namespace/retry_table') +13 diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index f681670208f37..1f66a20006dc4 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -14,8 +14,10 @@ import urllib.error import urllib.request -from materialize.mzcompose.composition import Composition, Service +from materialize.mzcompose.composition import Composition +from materialize.mzcompose.composition import Service as UpService from materialize.mzcompose.helpers.iceberg import setup_polaris_for_iceberg +from materialize.mzcompose.service import Service from materialize.mzcompose.services.materialized import Materialized from materialize.mzcompose.services.minio import Mc, Minio from materialize.mzcompose.services.mz import Mz @@ -29,6 +31,21 @@ Minio(), PolarisBootstrap(), Polaris(), + Service( + "polaris-proxy", + { + "image": "python:3.11-slim", + "command": ["python", "-u", "polaris_proxy.py"], + "working_dir": "/workdir", + "volumes": [".:/workdir"], + "ports": [8181], + "environment": [ + "UPSTREAM_HOST=polaris", + "UPSTREAM_PORT=8181", + "PROXY_PORT=8181", + ], + }, + ), Materialized( depends_on=["minio"], sanity_restart=False, @@ -49,8 +66,8 @@ def _setup( c.up( "postgres", "materialized", - Service("polaris-bootstrap", idle=True), - Service("polaris", idle=True), + UpService("polaris-bootstrap", idle=True), + UpService("polaris", idle=True), ) _, key = setup_polaris_for_iceberg( c, vended=vended, static_credentials=static_credentials @@ -411,6 +428,52 @@ def modify_table_loop() -> None: ) +def workflow_idempotent_retry(c: Composition) -> None: + """Regression test: dropping a single catalog commit response must not + fence the sink off or cause duplicate row commits.""" + key = _setup(c) + c.invoke("up", "--detach", "--wait", "--no-recreate", "polaris-proxy") + + c.run_testdrive_files( + f"--var=s3-access-key={key}", + "--var=aws-endpoint=minio:9000", + "idempotent-retry-setup.td", + ) + + proxy_base = f"http://localhost:{c.port('polaris-proxy', 8181)}" + + def proxy_post(path: str) -> None: + with urllib.request.urlopen( + urllib.request.Request(f"{proxy_base}{path}", data=b"", method="POST") + ) as resp: + resp.read() + + time.sleep(15) + proxy_post("/__control/drop_next_commit") + + for i in range(10): + c.sql(f"INSERT INTO retry_src VALUES ({i + 4}, 'row_{i + 4}')") + time.sleep(1) + + time.sleep(60) + + status_rows = c.sql_query( + "SELECT s.status, COALESCE(s.error, '') " + "FROM mz_internal.mz_sink_statuses s " + "JOIN mz_sinks ON s.id = mz_sinks.id " + "WHERE mz_sinks.name = 'retry_sink'" + ) + assert status_rows, "retry_sink not found in mz_sink_statuses" + status, error = status_rows[0] + assert status == "running", f"retry_sink is {status!r} (error={error!r})" + + c.run_testdrive_files( + f"--var=s3-access-key={key}", + "--var=aws-endpoint=minio:9000", + "idempotent-retry-verify.td", + ) + + def workflow_large_upsert_batch(c: Composition) -> None: """Regression test for database-issues#11326: DeltaWriter seen_rows eviction caused equality deletes within the same snapshot, which diff --git a/test/iceberg/polaris_proxy.py b/test/iceberg/polaris_proxy.py new file mode 100644 index 0000000000000..d90e1eb65dc50 --- /dev/null +++ b/test/iceberg/polaris_proxy.py @@ -0,0 +1,145 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +import http.server +import os +import socketserver +import sys +import threading +import time +import urllib.error +import urllib.request + +UPSTREAM_HOST = os.environ.get("UPSTREAM_HOST", "polaris") +UPSTREAM_PORT = int(os.environ.get("UPSTREAM_PORT", "8181")) + +_lock = threading.Lock() +_drop_armed = False + + +def _log(msg: str) -> None: + ts = time.strftime("%H:%M:%S") + sys.stderr.write(f"[polaris-proxy {ts}] {msg}\n") + sys.stderr.flush() + + +class Handler(http.server.BaseHTTPRequestHandler): + def log_message(self, fmt, *args): # type: ignore[override] + return + + def _handle_control(self) -> bool: + global _drop_armed + if not self.path.startswith("/__control/"): + return False + if self.path == "/__control/drop_next_commit" and self.command == "POST": + with _lock: + _drop_armed = True + self.send_response(200) + self.send_header("Content-Length", "6") + self.end_headers() + self.wfile.write(b"armed\n") + return True + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return True + + def _is_table_commit_endpoint(self) -> bool: + parts = self.path.split("?", 1)[0].split("/") + if "tables" not in parts: + return False + idx = parts.index("tables") + return idx < len(parts) - 1 and parts[idx + 1] != "" + + def _forward(self, body: bytes | None) -> None: + global _drop_armed + upstream_url = f"http://{UPSTREAM_HOST}:{UPSTREAM_PORT}{self.path}" + headers = { + k: v + for k, v in self.headers.items() + if k.lower() not in ("host", "transfer-encoding", "content-length") + } + req = urllib.request.Request( + upstream_url, data=body, headers=headers, method=self.command + ) + try: + resp = urllib.request.urlopen(req) + resp_body = resp.read() + status = resp.status + resp_headers = list(resp.headers.items()) + except urllib.error.HTTPError as e: + resp_body = e.read() + status = e.code + resp_headers = list(e.headers.items()) + + should_drop = False + if ( + self.command == "POST" + and self._is_table_commit_endpoint() + and 200 <= status < 300 + ): + with _lock: + if _drop_armed: + _drop_armed = False + should_drop = True + + if should_drop: + _log(f"dropping response for {self.command} {self.path}") + self.send_response(502) + self.send_header("Content-Length", "0") + self.end_headers() + return + + self.send_response(status) + for k, v in resp_headers: + if k.lower() not in ("transfer-encoding", "connection", "content-length"): + self.send_header(k, v) + self.send_header("Content-Length", str(len(resp_body))) + self.end_headers() + if resp_body: + self.wfile.write(resp_body) + + def _do_with_body(self) -> None: + if self._handle_control(): + return + length = int(self.headers.get("Content-Length", 0)) + body = self.rfile.read(length) if length else None + self._forward(body) + + def _do_without_body(self) -> None: + if self._handle_control(): + return + self._forward(None) + + def do_GET(self): + self._do_without_body() + + def do_HEAD(self): + self._do_without_body() + + def do_DELETE(self): + self._do_with_body() + + def do_POST(self): + self._do_with_body() + + def do_PUT(self): + self._do_with_body() + + +class ThreadedHTTPServer(socketserver.ThreadingMixIn, http.server.HTTPServer): + daemon_threads = True + allow_reuse_address = True + + +if __name__ == "__main__": + port = int(os.environ.get("PROXY_PORT", "8181")) + server = ThreadedHTTPServer(("0.0.0.0", port), Handler) + _log(f"listening on :{port}, upstream={UPSTREAM_HOST}:{UPSTREAM_PORT}") + server.serve_forever() From 65c8c789921fdf5c43b4e00f54830e73795336d3 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 19 Aug 2026 14:03:05 -0400 Subject: [PATCH 02/12] Claude: test case polls instead of sleeping --- test/iceberg/mzcompose.py | 43 +++++++++++++++++++++++++++++++++-- test/iceberg/polaris_proxy.py | 19 +++++++++++++++- 2 files changed, 59 insertions(+), 3 deletions(-) diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index 1f66a20006dc4..cff32b250e1e4 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -13,6 +13,7 @@ import time import urllib.error import urllib.request +from collections.abc import Callable from materialize.mzcompose.composition import Composition from materialize.mzcompose.composition import Service as UpService @@ -448,14 +449,52 @@ def proxy_post(path: str) -> None: ) as resp: resp.read() - time.sleep(15) + def proxy_status() -> dict[str, int]: + with urllib.request.urlopen(f"{proxy_base}/__control/status") as resp: + return json.loads(resp.read()) + + def await_condition(what: str, timeout: float, check: Callable[[], bool]) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + if check(): + return + time.sleep(0.5) + raise AssertionError(f"timed out waiting for {what}") + + # Arm the drop only once the sink has a commit through the proxy, so the + # dropped commit is a steady-state one rather than table bootstrap. + await_condition( + "first sink commit", + timeout=60, + check=lambda: proxy_status()["commits_ok"] >= 1, + ) proxy_post("/__control/drop_next_commit") for i in range(10): c.sql(f"INSERT INTO retry_src VALUES ({i + 4}, 'row_{i + 4}')") time.sleep(1) - time.sleep(60) + await_condition( + "dropped commit response", + timeout=60, + check=lambda: proxy_status()["commits_dropped"] >= 1, + ) + + def messages_committed() -> int: + rows = c.sql_query( + "SELECT COALESCE(SUM(messages_committed), 0) " + "FROM mz_internal.mz_sink_statistics st " + "JOIN mz_sinks s ON st.id = s.id " + "WHERE s.name = 'retry_sink'" + ) + return int(rows[0][0]) + + # 3 initial rows + 10 inserted rows, all committed despite the dropped response. + await_condition( + "all 13 rows committed", + timeout=120, + check=lambda: messages_committed() >= 13, + ) status_rows = c.sql_query( "SELECT s.status, COALESCE(s.error, '') " diff --git a/test/iceberg/polaris_proxy.py b/test/iceberg/polaris_proxy.py index d90e1eb65dc50..1ce5e27843e10 100644 --- a/test/iceberg/polaris_proxy.py +++ b/test/iceberg/polaris_proxy.py @@ -8,6 +8,7 @@ # by the Apache License, Version 2.0. import http.server +import json import os import socketserver import sys @@ -21,6 +22,9 @@ _lock = threading.Lock() _drop_armed = False +# Successful (2xx upstream) table commits seen, including ones whose response we dropped. +_commits_ok = 0 +_commits_dropped = 0 def _log(msg: str) -> None: @@ -45,6 +49,17 @@ def _handle_control(self) -> bool: self.end_headers() self.wfile.write(b"armed\n") return True + if self.path == "/__control/status" and self.command == "GET": + with _lock: + payload = json.dumps( + {"commits_ok": _commits_ok, "commits_dropped": _commits_dropped} + ).encode() + self.send_response(200) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(payload))) + self.end_headers() + self.wfile.write(payload) + return True self.send_response(404) self.send_header("Content-Length", "0") self.end_headers() @@ -58,7 +73,7 @@ def _is_table_commit_endpoint(self) -> bool: return idx < len(parts) - 1 and parts[idx + 1] != "" def _forward(self, body: bytes | None) -> None: - global _drop_armed + global _drop_armed, _commits_ok, _commits_dropped upstream_url = f"http://{UPSTREAM_HOST}:{UPSTREAM_PORT}{self.path}" headers = { k: v @@ -85,9 +100,11 @@ def _forward(self, body: bytes | None) -> None: and 200 <= status < 300 ): with _lock: + _commits_ok += 1 if _drop_armed: _drop_armed = False should_drop = True + _commits_dropped += 1 if should_drop: _log(f"dropping response for {self.command} {self.path}") From 4a0cbc024e671ebf785d169385f97dd0533c815c Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 19 Aug 2026 14:56:47 -0400 Subject: [PATCH 03/12] Claude: test case, iceberg-rust still commits over newer writer --- test/iceberg/fenced-writer-setup.td | 40 ++++++++ test/iceberg/mzcompose.py | 153 +++++++++++++++++++++++++++- 2 files changed, 192 insertions(+), 1 deletion(-) create mode 100644 test/iceberg/fenced-writer-setup.td diff --git a/test/iceberg/fenced-writer-setup.td b/test/iceberg/fenced-writer-setup.td new file mode 100644 index 0000000000000..1c2e4f852b791 --- /dev/null +++ b/test/iceberg/fenced-writer-setup.td @@ -0,0 +1,40 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +> CREATE SECRET access_key_secret AS '${arg.s3-access-key}' + +> CREATE CONNECTION aws_conn TO AWS ( + ACCESS KEY ID = 'tduser', + SECRET ACCESS KEY = SECRET access_key_secret, + ENDPOINT = '${arg.aws-endpoint}', + REGION = 'us-east-1' + ); + +> CREATE CONNECTION polaris TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'http://polaris:8181/api/catalog', + CREDENTIAL = 'root:root', + WAREHOUSE = 'default_catalog', + SCOPE = 'PRINCIPAL_ROLE:ALL' + ); + +> CREATE TABLE fence_src (id INT, val TEXT); + +> INSERT INTO fence_src VALUES (1, 'a'), (2, 'b'), (3, 'c'); + +> CREATE SINK fence_sink + FROM fence_src + INTO ICEBERG CATALOG CONNECTION polaris ( + NAMESPACE 'default_namespace', + TABLE 'fence_table' + ) + USING AWS CONNECTION aws_conn + KEY (id) NOT ENFORCED + MODE UPSERT + WITH (COMMIT INTERVAL '2s'); diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index cff32b250e1e4..227b9d5e4d1ef 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -17,7 +17,10 @@ from materialize.mzcompose.composition import Composition from materialize.mzcompose.composition import Service as UpService -from materialize.mzcompose.helpers.iceberg import setup_polaris_for_iceberg +from materialize.mzcompose.helpers.iceberg import ( + get_polaris_access_token, + setup_polaris_for_iceberg, +) from materialize.mzcompose.service import Service from materialize.mzcompose.services.materialized import Materialized from materialize.mzcompose.services.minio import Mc, Minio @@ -513,6 +516,154 @@ def messages_committed() -> int: ) +def workflow_fenced_writer(c: Composition) -> None: + """Regression test: once a snapshot with a newer mz-sink-version is on the + table, the running sink must stop committing. + + Currently fails: iceberg-rust's commit path reloads the table and rebases + onto the newest snapshot before every attempt, so the fenced sink never + sees a conflict and commits right over the newer writer. Its snapshot then + becomes the latest one, so even the startup fencing check of a later + restart no longer notices the newer writer.""" + key = _setup(c) + + c.run_testdrive_files( + f"--var=s3-access-key={key}", + "--var=aws-endpoint=minio:9000", + "fenced-writer-setup.td", + ) + + token = get_polaris_access_token(c) + table_url = ( + f"http://localhost:{c.port('polaris', 8181)}" + "/api/catalog/v1/default_catalog/namespaces/default_namespace/tables/fence_table" + ) + headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + + def load_metadata() -> dict | None: + req = urllib.request.Request(table_url, headers=headers) + try: + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read())["metadata"] + except urllib.error.HTTPError as e: + if e.code == 404: + return None + raise + + def await_condition(what: str, timeout: float, check: Callable[[], bool]) -> None: + deadline = time.time() + timeout + while time.time() < deadline: + if check(): + return + time.sleep(0.5) + raise AssertionError(f"timed out waiting for {what}") + + def first_commit_done() -> bool: + meta = load_metadata() + return meta is not None and meta.get("current-snapshot-id") not in (None, -1) + + await_condition("first sink commit", timeout=60, check=first_commit_done) + + def forge_newer_writer_snapshot() -> int: + """Commit a snapshot that claims mz-sink-version 999, as a sink with a + newer version would. Reuses the current snapshot's manifest list so the + table stays readable without writing new files. Returns the forged + snapshot's sequence number.""" + meta = load_metadata() + assert meta is not None + current_id = meta["refs"]["main"]["snapshot-id"] + current = next(s for s in meta["snapshots"] if s["snapshot-id"] == current_id) + forged_seq = meta["last-sequence-number"] + 1 + body = json.dumps( + { + "requirements": [ + { + "type": "assert-ref-snapshot-id", + "ref": "main", + "snapshot-id": current_id, + } + ], + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": current_id + 1, + "parent-snapshot-id": current_id, + "sequence-number": forged_seq, + "timestamp-ms": int(time.time() * 1000), + "manifest-list": current["manifest-list"], + "schema-id": meta["current-schema-id"], + "summary": { + "operation": "append", + "mz-sink-id": "u0", + "mz-frontier": current["summary"]["mz-frontier"], + "mz-sink-version": "999", + }, + }, + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "type": "branch", + "snapshot-id": current_id + 1, + }, + ], + } + ).encode() + req = urllib.request.Request( + table_url, data=body, headers=headers, method="POST" + ) + with urllib.request.urlopen(req) as resp: + resp.read() + return forged_seq + + # The sink may commit between reading the metadata and posting the forged + # snapshot, failing our requirement. Retry on conflict with fresh metadata. + for attempt in range(5): + try: + forged_seq = forge_newer_writer_snapshot() + break + except urllib.error.HTTPError as e: + if e.code != 409 or attempt == 4: + raise + else: + raise AssertionError("unreachable") + + # Give the fenced sink something to commit. + for i in range(5): + c.sql(f"INSERT INTO fence_src VALUES ({i + 4}, 'row_{i + 4}')") + time.sleep(1) + + deadline = time.time() + 90 + while True: + meta = load_metadata() + assert meta is not None + for snapshot in meta["snapshots"]: + if ( + snapshot["sequence-number"] > forged_seq + and snapshot["summary"].get("mz-sink-version") != "999" + ): + raise AssertionError( + "fenced sink committed past the newer writer: " + f"snapshot {snapshot['snapshot-id']} " + f"summary {snapshot['summary']}" + ) + + status_rows = c.sql_query( + "SELECT s.status, COALESCE(s.error, '') " + "FROM mz_internal.mz_sink_statuses s " + "JOIN mz_sinks ON s.id = mz_sinks.id " + "WHERE mz_sinks.name = 'fence_sink'" + ) + assert status_rows, "fence_sink not found in mz_sink_statuses" + status, error = status_rows[0] + if status != "running" and ("another writer" in error or "Fenced off" in error): + return + if time.time() > deadline: + raise AssertionError(f"sink not fenced: status={status!r} error={error!r}") + time.sleep(1) + + def workflow_large_upsert_batch(c: Composition) -> None: """Regression test for database-issues#11326: DeltaWriter seen_rows eviction caused equality deletes within the same snapshot, which From bff588aab927cd7b99204ba8980f109941b49cc9 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Tue, 18 Aug 2026 15:20:03 -0400 Subject: [PATCH 04/12] Claude: bypass iceberg-rust Txn::commit. sink handles retries. Claude: on both retrial paths, check for other writers --- src/storage/src/sink/iceberg.rs | 257 ++++++++++++++++++++++++-------- 1 file changed, 195 insertions(+), 62 deletions(-) diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index eb6526aea0ddf..f7d6b07f5b104 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -105,7 +105,7 @@ use iceberg::spec::{ }; use iceberg::spec::{Schema, SchemaRef}; use iceberg::table::Table; -use iceberg::transaction::{ApplyTransactionAction, Transaction}; +use iceberg::transaction::{RowDeltaAction, TransactionAction}; use iceberg::writer::base_writer::data_file_writer::DataFileWriterBuilder; use iceberg::writer::base_writer::equality_delete_writer::{ EqualityDeleteFileWriterBuilder, EqualityDeleteWriterConfig, @@ -120,7 +120,7 @@ use iceberg::writer::file_writer::location_generator::{ }; use iceberg::writer::file_writer::rolling_writer::RollingFileWriterBuilder; use iceberg::writer::{IcebergWriter, IcebergWriterBuilder}; -use iceberg::{Catalog, NamespaceIdent, TableCreation, TableIdent}; +use iceberg::{Catalog, NamespaceIdent, TableCommit, TableCreation, TableIdent}; use itertools::Itertools; use mz_arrow_util::builder::{ARROW_EXTENSION_NAME_KEY, ArrowBuilder}; use mz_interchange::avro::DiffPair; @@ -155,7 +155,7 @@ use timely::dataflow::channels::pact::{Exchange, Pipeline}; use timely::dataflow::operators::vec::{Broadcast, Map, ToStream}; use timely::dataflow::operators::{CapabilitySet, Concatenate}; use timely::progress::{Antichain, Timestamp as _}; -use tracing::debug; +use tracing::{debug, info, warn}; use crate::healthcheck::{HealthStatusMessage, HealthStatusUpdate, StatusNamespace}; use crate::metrics::sink::iceberg::IcebergSinkMetrics; @@ -778,11 +778,63 @@ async fn reload_table( } } +/// After attempting a commit, we expect to know whether it succeeded. +enum CommitState { + /// The Catalog told us whether the commit succeeded and gave us an updated table + /// (which we can safely use for another attempt). + Known(Table), + /// We couldn't get an answer from the Catalog. + /// We don't have an updated table. We don't know if the commit succeeded or failed. + /// Before making any more attempts, we must reload the table + /// and check for ourselves whether the commit is there. + Unresolved(Table), +} + +/// Check whether the most recent Materialize snapshot on the table belongs to another writer. +fn check_fencing( + last: Option<&(Antichain, u64)>, + sink_version: u64, + frontier: &Antichain, + conn_table: &str, +) -> Result<(), anyhow::Error> { + let Some((last_frontier, last_version)) = last else { + return Ok(()); + }; + if *last_version > sink_version { + anyhow::bail!( + "Iceberg table '{}' has been modified by another writer \ + with version {}. Current sink version: {}. \ + Frontiers may be out of sync, aborting to avoid data loss.", + conn_table, + last_version, + sink_version, + ); + } + + // Check if someone has already written this batch (or an even later batch). + // If it was us, we should already know. + // (Either we received a success response or we reloaded the table and checked.) + // So it must be another writer. + if PartialOrder::less_equal(frontier, last_frontier) { + anyhow::bail!( + "Iceberg table '{}' has been modified by another writer. \ + Current frontier: {:?}, last frontier: {:?}.", + conn_table, + frontier, + last_frontier, + ); + } + Ok(()) +} + /// Attempt a single commit of a batch of data files to an Iceberg table. -/// On conflict or failure, reloads the table and returns a retryable error. -/// On success, returns the updated table state. +/// +/// If a previous attempt left the outcome unknown, first reload the table to establish whether +/// that attempt succeeded. Try again only once we're sure it failed. +/// On conflict, reload the table and return a retryable error. +/// On success, return the updated table state. async fn try_commit_batch( - mut table: Table, + state: CommitState, snapshot_properties: Vec<(String, String)>, data_files: Vec, delete_files: Vec, @@ -794,10 +846,71 @@ async fn try_commit_batch( batch_lower: &Antichain, batch_upper: &Antichain, metrics: &IcebergSinkMetrics, -) -> (Table, RetryResult<(), anyhow::Error>) { - let tx = Transaction::new(&table); - let mut action = tx - .row_delta() +) -> (CommitState, RetryResult<(), anyhow::Error>) { + let table = match state { + CommitState::Known(table) => table, + CommitState::Unresolved(stale) => { + let reloaded = match reload_table( + catalog, + conn_namespace.to_string(), + conn_table.to_string(), + stale.clone(), + ) + .await + { + Ok(reloaded) => reloaded, + // We can't proceed until we've checked the table. + Err(e) => { + return ( + CommitState::Unresolved(stale), + RetryResult::RetryableErr(anyhow!(e)), + ); + } + }; + + let mut snapshots: Vec<_> = reloaded.metadata().snapshots().cloned().collect(); + match retrieve_upper_from_snapshots(&mut snapshots) { + // Our own commit for this batch is already on the table. + // It landed and we never saw the response. + Ok(Some((last_frontier, last_version))) + if last_version == sink_version && last_frontier == *frontier => + { + info!( + namespace = %conn_namespace, + table = %conn_table, + lower = %batch_lower.pretty(), + upper = %batch_upper.pretty(), + "iceberg commit with lost response found on reload, treating as success" + ); + return (CommitState::Known(reloaded), RetryResult::Ok(())); + } + // Our commit isn't there. + // We can retry only if no other writer has taken over the table in the meantime. + Ok(last) => { + if let Err(e) = check_fencing(last.as_ref(), sink_version, frontier, conn_table) + { + return (CommitState::Known(reloaded), RetryResult::FatalErr(e)); + } + info!( + namespace = %conn_namespace, + table = %conn_table, + lower = %batch_lower.pretty(), + upper = %batch_upper.pretty(), + "iceberg commit with lost response not found on reload, committing again" + ); + reloaded + } + Err(e) => { + return ( + CommitState::Unresolved(stale), + RetryResult::RetryableErr(anyhow!(e)), + ); + } + } + } + }; + + let mut action = RowDeltaAction::new() .set_snapshot_properties(snapshot_properties.into_iter().collect()) .with_check_duplicate(false); @@ -807,13 +920,18 @@ async fn try_commit_batch( .add_delete_files(delete_files); } - let tx = match action - .apply(tx) - .context("Failed to apply data file addition to iceberg table transaction") - { - Ok(tx) => tx, + // Build the commit's metadata updates and requirements against our own view of + // the table and send them to the catalog directly, instead of going through + // `Transaction::commit`. That path reloads the table and rebases the commit onto + // whatever it finds, retrying conflicts internally, so it would silently commit + // over another writer. Generated this way, the requirements pin the table state + // we know, and any interleaved write surfaces as a commit conflict below, where + // `check_fencing` decides whether retrying is safe. + let mut action_commit = match Arc::new(action).commit(&table).await { + Ok(action_commit) => action_commit, Err(e) => { - match reload_table( + // Nothing was sent to the catalog, so this commit definitely didn't happen. + let reloaded = match reload_table( catalog, conn_namespace.to_string(), conn_table.to_string(), @@ -821,26 +939,32 @@ async fn try_commit_batch( ) .await { - Ok(reloaded) => table = reloaded, + Ok(reloaded) => reloaded, Err(reload_err) => { - return (table, RetryResult::RetryableErr(anyhow!(reload_err))); + return ( + CommitState::Known(table), + RetryResult::RetryableErr(anyhow!(reload_err)), + ); } - } + }; return ( - table, - RetryResult::RetryableErr(anyhow!( - "Failed to apply data file addition to iceberg table transaction: {}", - e - )), + CommitState::Known(reloaded), + RetryResult::RetryableErr(anyhow!("Failed to build iceberg table commit: {}", e)), ); } }; - let new_table = tx.commit(catalog).await; + let table_commit = TableCommit::builder() + .ident(table.identifier().clone()) + .updates(action_commit.take_updates()) + .requirements(action_commit.take_requirements()) + .build(); + + let new_table = catalog.update_table(table_commit).await; match new_table { Err(e) if matches!(e.kind(), ErrorKind::CatalogCommitConflicts) => { metrics.commit_conflicts.inc(); - match reload_table( + let table = match reload_table( catalog, conn_namespace.to_string(), conn_table.to_string(), @@ -848,9 +972,12 @@ async fn try_commit_batch( ) .await { - Ok(reloaded) => table = reloaded, + Ok(reloaded) => reloaded, Err(e) => { - return (table, RetryResult::RetryableErr(anyhow!(e))); + return ( + CommitState::Known(table), + RetryResult::RetryableErr(anyhow!(e)), + ); } }; @@ -859,41 +986,20 @@ async fn try_commit_batch( let last = match last { Ok(val) => val, Err(e) => { - return (table, RetryResult::RetryableErr(anyhow!(e))); + return ( + CommitState::Known(table), + RetryResult::RetryableErr(anyhow!(e)), + ); } }; // Check if another writer has advanced the frontier beyond ours (fencing check) - if let Some((last_frontier, last_version)) = last { - if last_version > sink_version { - return ( - table, - RetryResult::FatalErr(anyhow!( - "Iceberg table '{}' has been modified by another writer \ - with version {}. Current sink version: {}. \ - Frontiers may be out of sync, aborting to avoid data loss.", - conn_table, - last_version, - sink_version, - )), - ); - } - if PartialOrder::less_equal(frontier, &last_frontier) { - return ( - table, - RetryResult::FatalErr(anyhow!( - "Iceberg table '{}' has been modified by another writer. \ - Current frontier: {:?}, last frontier: {:?}.", - conn_table, - frontier, - last_frontier, - )), - ); - } + if let Err(e) = check_fencing(last.as_ref(), sink_version, frontier, conn_table) { + return (CommitState::Known(table), RetryResult::FatalErr(e)); } ( - table, + CommitState::Known(table), RetryResult::RetryableErr(anyhow!( "Commit conflict detected when committing batch [{}, {}) \ to Iceberg table '{}.{}'. Retrying...", @@ -904,11 +1010,29 @@ async fn try_commit_batch( )), ) } + // The catalog may have applied this commit before the success response was lost. + // Mark the outcome as unknown so the next attempt starts by reading the table to see what happened. + Err(e) if matches!(e.kind(), ErrorKind::Unexpected) => { + metrics.commit_failures.inc(); + warn!( + namespace = %conn_namespace, + table = %conn_table, + lower = %batch_lower.pretty(), + upper = %batch_upper.pretty(), + error = %e, + "iceberg commit outcome unknown, will reload the table to check before retrying" + ); + ( + CommitState::Unresolved(table), + RetryResult::RetryableErr(anyhow!(e)), + ) + } + // All other errors are definite: retrying will not change the outcome. Err(e) => { metrics.commit_failures.inc(); - (table, RetryResult::RetryableErr(anyhow!(e))) + (CommitState::Known(table), RetryResult::FatalErr(anyhow!(e))) } - Ok(new_table) => (new_table, RetryResult::Ok(())), + Ok(new_table) => (CommitState::Known(new_table), RetryResult::Ok(())), } } @@ -2671,9 +2795,9 @@ fn commit_to_iceberg<'scope>( ("mz-sink-version".to_string(), sink_version.to_string()), ]; - let (table_state, commit_result) = Retry::default() + let (commit_state, commit_result) = Retry::default() .max_tries(5) - .retry_async_with_state(table, |_, table| { + .retry_async_with_state(CommitState::Known(table), |_, commit_state| { let snapshot_properties = snapshot_properties.clone(); let data_files = data_files.clone(); let delete_files = delete_files.clone(); @@ -2686,7 +2810,7 @@ fn commit_to_iceberg<'scope>( let batch_upper = batch.1.clone(); async move { try_commit_batch( - table, + commit_state, snapshot_properties, data_files, delete_files, @@ -2709,12 +2833,21 @@ fn commit_to_iceberg<'scope>( connection.namespace, connection.table ) }); - table = table_state; let duration = instant.elapsed(); metrics .commit_duration_seconds .observe(duration.as_secs_f64()); commit_result?; + table = match commit_state { + CommitState::Known(table) => table, + CommitState::Unresolved(_) => { + anyhow::bail!( + "invariant: unresolved commit state but the commit succeeded, Iceberg table '{}.{}'", + connection.namespace, + connection.table + ) + } + }; debug!( ?sink_id, From 3eaa5e079b1e4c171c802b51e72a44f86d070c7b Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Thu, 20 Aug 2026 16:55:59 -0400 Subject: [PATCH 05/12] human rewrite --- src/storage/src/sink/iceberg.rs | 393 +++++++++++++------------------- 1 file changed, 156 insertions(+), 237 deletions(-) diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index f7d6b07f5b104..41da16e3c08ea 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -155,7 +155,7 @@ use timely::dataflow::channels::pact::{Exchange, Pipeline}; use timely::dataflow::operators::vec::{Broadcast, Map, ToStream}; use timely::dataflow::operators::{CapabilitySet, Concatenate}; use timely::progress::{Antichain, Timestamp as _}; -use tracing::{debug, info, warn}; +use tracing::{debug, info}; use crate::healthcheck::{HealthStatusMessage, HealthStatusUpdate, StatusNamespace}; use crate::metrics::sink::iceberg::IcebergSinkMetrics; @@ -743,7 +743,7 @@ async fn reload_table( catalog: &dyn Catalog, namespace: String, table_name: String, - current_table: Table, + current_table: &Table, ) -> anyhow::Result { let namespace_ident = NamespaceIdent::new(namespace.clone()); let table_ident = TableIdent::new(namespace_ident, table_name.clone()); @@ -778,138 +778,33 @@ async fn reload_table( } } -/// After attempting a commit, we expect to know whether it succeeded. -enum CommitState { - /// The Catalog told us whether the commit succeeded and gave us an updated table - /// (which we can safely use for another attempt). - Known(Table), - /// We couldn't get an answer from the Catalog. - /// We don't have an updated table. We don't know if the commit succeeded or failed. - /// Before making any more attempts, we must reload the table - /// and check for ourselves whether the commit is there. - Unresolved(Table), +/// A failed commit attempt, split by whether the commit request reached the catalog. +enum CommitError { + /// Failed while building the commit locally. Nothing was sent to the catalog, + /// so the commit definitely didn't happen. + Local(iceberg::Error), + /// The update request failed. Depending on the error kind, + /// it's still possible the catalog applied the commit. + Request(iceberg::Error), } -/// Check whether the most recent Materialize snapshot on the table belongs to another writer. -fn check_fencing( - last: Option<&(Antichain, u64)>, - sink_version: u64, - frontier: &Antichain, - conn_table: &str, -) -> Result<(), anyhow::Error> { - let Some((last_frontier, last_version)) = last else { - return Ok(()); - }; - if *last_version > sink_version { - anyhow::bail!( - "Iceberg table '{}' has been modified by another writer \ - with version {}. Current sink version: {}. \ - Frontiers may be out of sync, aborting to avoid data loss.", - conn_table, - last_version, - sink_version, - ); - } - - // Check if someone has already written this batch (or an even later batch). - // If it was us, we should already know. - // (Either we received a success response or we reloaded the table and checked.) - // So it must be another writer. - if PartialOrder::less_equal(frontier, last_frontier) { - anyhow::bail!( - "Iceberg table '{}' has been modified by another writer. \ - Current frontier: {:?}, last frontier: {:?}.", - conn_table, - frontier, - last_frontier, - ); - } - Ok(()) -} - -/// Attempt a single commit of a batch of data files to an Iceberg table. +/// Build a row delta commit against the given table and send it to the catalog. /// -/// If a previous attempt left the outcome unknown, first reload the table to establish whether -/// that attempt succeeded. Try again only once we're sure it failed. -/// On conflict, reload the table and return a retryable error. -/// On success, return the updated table state. -async fn try_commit_batch( - state: CommitState, +/// We can't use iceberg-rust's `Transaction::commit` (a retry wrapper for `Transaction::do_commit`) +/// because it automatically rebases the transaction onto the latest state of the table. +/// +/// That behavior leads to duplicate writes because there's no way to check: +/// 1. Have we already committed this data? +/// 2. Has another (newer) writer taken over? +/// +/// So this implementation doesn't do that (details inline). +async fn do_commit( + table: &Table, + catalog: &dyn Catalog, snapshot_properties: Vec<(String, String)>, data_files: Vec, delete_files: Vec, - catalog: &dyn Catalog, - conn_namespace: &str, - conn_table: &str, - sink_version: u64, - frontier: &Antichain, - batch_lower: &Antichain, - batch_upper: &Antichain, - metrics: &IcebergSinkMetrics, -) -> (CommitState, RetryResult<(), anyhow::Error>) { - let table = match state { - CommitState::Known(table) => table, - CommitState::Unresolved(stale) => { - let reloaded = match reload_table( - catalog, - conn_namespace.to_string(), - conn_table.to_string(), - stale.clone(), - ) - .await - { - Ok(reloaded) => reloaded, - // We can't proceed until we've checked the table. - Err(e) => { - return ( - CommitState::Unresolved(stale), - RetryResult::RetryableErr(anyhow!(e)), - ); - } - }; - - let mut snapshots: Vec<_> = reloaded.metadata().snapshots().cloned().collect(); - match retrieve_upper_from_snapshots(&mut snapshots) { - // Our own commit for this batch is already on the table. - // It landed and we never saw the response. - Ok(Some((last_frontier, last_version))) - if last_version == sink_version && last_frontier == *frontier => - { - info!( - namespace = %conn_namespace, - table = %conn_table, - lower = %batch_lower.pretty(), - upper = %batch_upper.pretty(), - "iceberg commit with lost response found on reload, treating as success" - ); - return (CommitState::Known(reloaded), RetryResult::Ok(())); - } - // Our commit isn't there. - // We can retry only if no other writer has taken over the table in the meantime. - Ok(last) => { - if let Err(e) = check_fencing(last.as_ref(), sink_version, frontier, conn_table) - { - return (CommitState::Known(reloaded), RetryResult::FatalErr(e)); - } - info!( - namespace = %conn_namespace, - table = %conn_table, - lower = %batch_lower.pretty(), - upper = %batch_upper.pretty(), - "iceberg commit with lost response not found on reload, committing again" - ); - reloaded - } - Err(e) => { - return ( - CommitState::Unresolved(stale), - RetryResult::RetryableErr(anyhow!(e)), - ); - } - } - } - }; - +) -> Result { let mut action = RowDeltaAction::new() .set_snapshot_properties(snapshot_properties.into_iter().collect()) .with_check_duplicate(false); @@ -920,39 +815,20 @@ async fn try_commit_batch( .add_delete_files(delete_files); } - // Build the commit's metadata updates and requirements against our own view of - // the table and send them to the catalog directly, instead of going through - // `Transaction::commit`. That path reloads the table and rebases the commit onto - // whatever it finds, retrying conflicts internally, so it would silently commit - // over another writer. Generated this way, the requirements pin the table state - // we know, and any interleaved write surfaces as a commit conflict below, where - // `check_fencing` decides whether retrying is safe. - let mut action_commit = match Arc::new(action).commit(&table).await { - Ok(action_commit) => action_commit, - Err(e) => { - // Nothing was sent to the catalog, so this commit definitely didn't happen. - let reloaded = match reload_table( - catalog, - conn_namespace.to_string(), - conn_table.to_string(), - table.clone(), - ) - .await - { - Ok(reloaded) => reloaded, - Err(reload_err) => { - return ( - CommitState::Known(table), - RetryResult::RetryableErr(anyhow!(reload_err)), - ); - } - }; - return ( - CommitState::Known(reloaded), - RetryResult::RetryableErr(anyhow!("Failed to build iceberg table commit: {}", e)), - ); - } - }; + // Divergence: `Transaction::do_commit` reloads the table and rebases the transaction onto it. + // We do not reload the table and rebase the transaction. + // The caller must check the transaction against the table before committing because + // the Iceberg Catalog's conflict check is not sufficient to prevent duplicate commits. + + let mut action_commit = Arc::new(action) + .commit(table) + .await + .map_err(CommitError::Local)?; + + // Divergence: `Transaction::do_commit` also checks each action's requirements + // against the local metadata and applies its updates to a local copy of the table, + // so the next action in the transaction can build on the result. + // We only commit a single action, so we skip that step. let table_commit = TableCommit::builder() .ident(table.identifier().clone()) @@ -960,79 +836,131 @@ async fn try_commit_batch( .requirements(action_commit.take_requirements()) .build(); - let new_table = catalog.update_table(table_commit).await; - match new_table { - Err(e) if matches!(e.kind(), ErrorKind::CatalogCommitConflicts) => { - metrics.commit_conflicts.inc(); - let table = match reload_table( - catalog, - conn_namespace.to_string(), - conn_table.to_string(), - table.clone(), - ) - .await - { - Ok(reloaded) => reloaded, - Err(e) => { - return ( - CommitState::Known(table), - RetryResult::RetryableErr(anyhow!(e)), - ); - } - }; - - let mut snapshots: Vec<_> = table.metadata().snapshots().cloned().collect(); - let last = retrieve_upper_from_snapshots(&mut snapshots); - let last = match last { - Ok(val) => val, - Err(e) => { - return ( - CommitState::Known(table), - RetryResult::RetryableErr(anyhow!(e)), - ); - } - }; - - // Check if another writer has advanced the frontier beyond ours (fencing check) - if let Err(e) = check_fencing(last.as_ref(), sink_version, frontier, conn_table) { - return (CommitState::Known(table), RetryResult::FatalErr(e)); - } + catalog + .update_table(table_commit) + .await + .map_err(CommitError::Request) +} - ( - CommitState::Known(table), - RetryResult::RetryableErr(anyhow!( - "Commit conflict detected when committing batch [{}, {}) \ - to Iceberg table '{}.{}'. Retrying...", - batch_lower.pretty(), - batch_upper.pretty(), - conn_namespace, - conn_table - )), - ) +/// Attempt a single commit of a batch of data files to an Iceberg table. +async fn try_commit_batch( + table: Table, + snapshot_properties: Vec<(String, String)>, + data_files: Vec, + delete_files: Vec, + catalog: &dyn Catalog, + conn_namespace: &str, + conn_table: &str, + sink_version: u64, + frontier: &Antichain, + batch_lower: &Antichain, + batch_upper: &Antichain, + metrics: &IcebergSinkMetrics, +) -> (Table, RetryResult<(), anyhow::Error>) { + // We begin the attempt by evaluating our current state. + // 1. Load the table from the catalog. + // 2. Check if the table metadata says it's safe to write to. + let table = match reload_table( + catalog, + conn_namespace.to_string(), + conn_table.to_string(), + &table, + ) + .await + { + Ok(table) => table, + Err(e) => { + // We can't proceed without a fresh view of the table, so we must retry. + return (table, RetryResult::RetryableErr(anyhow!(e))); } - // The catalog may have applied this commit before the success response was lost. - // Mark the outcome as unknown so the next attempt starts by reading the table to see what happened. - Err(e) if matches!(e.kind(), ErrorKind::Unexpected) => { - metrics.commit_failures.inc(); - warn!( + }; + + let mut snapshots: Vec<_> = table.metadata().snapshots().cloned().collect(); + let last = match retrieve_upper_from_snapshots(&mut snapshots) { + Ok(last) => last, + Err(e) => return (table, RetryResult::RetryableErr(anyhow!(e))), + }; + if let Some((last_frontier, last_version)) = last { + if last_version == sink_version && last_frontier == *frontier { + // Our own commit for this batch is already on the table. + // We must've missed the response. + info!( namespace = %conn_namespace, table = %conn_table, lower = %batch_lower.pretty(), upper = %batch_upper.pretty(), - error = %e, - "iceberg commit outcome unknown, will reload the table to check before retrying" + "found iceberg commit from previous attempt, treating as success" + ); + return (table, RetryResult::Ok(())); + } + + if last_version > sink_version { + // We've been superseded by a new version of the sink. + return ( + table, + RetryResult::FatalErr(anyhow!( + "Iceberg table '{}' has been modified by another writer \ + with version {}. Current sink version: {}. \ + Frontiers may be out of sync, aborting to avoid data loss.", + conn_table, + last_version, + sink_version, + )), + ); + } + + if PartialOrder::less_equal(frontier, &last_frontier) { + // Someone else has already written this batch (or an even later batch). + return ( + table, + RetryResult::FatalErr(anyhow!( + "Iceberg table '{}' has been modified by another writer. \ + Current frontier: {:?}, last frontier: {:?}.", + conn_table, + frontier, + last_frontier, + )), ); + } + } + + match do_commit( + &table, + catalog, + snapshot_properties, + data_files, + delete_files, + ) + .await + { + Ok(new_table) => (new_table, RetryResult::Ok(())), + Err(CommitError::Local(e)) => { + // Nothing was sent to the catalog, so this commit definitely didn't happen. + // Next attempt should reload the table and try again from scratch. ( - CommitState::Unresolved(table), - RetryResult::RetryableErr(anyhow!(e)), + table, + RetryResult::RetryableErr(anyhow!("Failed to build iceberg table commit: {}", e)), ) } - // All other errors are definite: retrying will not change the outcome. - Err(e) => { - metrics.commit_failures.inc(); - (CommitState::Known(table), RetryResult::FatalErr(anyhow!(e))) - } - Ok(new_table) => (CommitState::Known(new_table), RetryResult::Ok(())), + Err(CommitError::Request(e)) => match e.kind() { + ErrorKind::CatalogCommitConflicts => { + // Our view of the table was outdated. + // Next attempt should reload the table and try again from scratch. + metrics.commit_conflicts.inc(); + (table, RetryResult::RetryableErr(anyhow!(e))) + } + ErrorKind::Unexpected => { + // The catalog may have applied this commit before the success response was lost. + // Next attempt should reload the table and see if the commit landed. + metrics.commit_failures.inc(); + (table, RetryResult::RetryableErr(anyhow!(e))) + } + _ => { + // All other errors are definite: retrying will not change the outcome. + metrics.commit_failures.inc(); + (table, RetryResult::FatalErr(anyhow!(e))) + } + }, } } @@ -2795,9 +2723,9 @@ fn commit_to_iceberg<'scope>( ("mz-sink-version".to_string(), sink_version.to_string()), ]; - let (commit_state, commit_result) = Retry::default() + let (table_state, commit_result) = Retry::default() .max_tries(5) - .retry_async_with_state(CommitState::Known(table), |_, commit_state| { + .retry_async_with_state(table, |_, table| { let snapshot_properties = snapshot_properties.clone(); let data_files = data_files.clone(); let delete_files = delete_files.clone(); @@ -2810,7 +2738,7 @@ fn commit_to_iceberg<'scope>( let batch_upper = batch.1.clone(); async move { try_commit_batch( - commit_state, + table, snapshot_properties, data_files, delete_files, @@ -2833,21 +2761,12 @@ fn commit_to_iceberg<'scope>( connection.namespace, connection.table ) }); + table = table_state; let duration = instant.elapsed(); metrics .commit_duration_seconds .observe(duration.as_secs_f64()); commit_result?; - table = match commit_state { - CommitState::Known(table) => table, - CommitState::Unresolved(_) => { - anyhow::bail!( - "invariant: unresolved commit state but the commit succeeded, Iceberg table '{}.{}'", - connection.namespace, - connection.table - ) - } - }; debug!( ?sink_id, From d573ece1a3d47d2b0833d0a9b3493bd5ea95d95c Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 26 Aug 2026 16:17:48 -0400 Subject: [PATCH 06/12] check sink ID also --- src/storage/src/sink/iceberg.rs | 27 +++++++++++++++++++-------- 1 file changed, 19 insertions(+), 8 deletions(-) diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index 41da16e3c08ea..f12a709b1cedc 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -851,6 +851,7 @@ async fn try_commit_batch( catalog: &dyn Catalog, conn_namespace: &str, conn_table: &str, + sink_id: GlobalId, sink_version: u64, frontier: &Antichain, batch_lower: &Antichain, @@ -880,8 +881,9 @@ async fn try_commit_batch( Ok(last) => last, Err(e) => return (table, RetryResult::RetryableErr(anyhow!(e))), }; - if let Some((last_frontier, last_version)) = last { - if last_version == sink_version && last_frontier == *frontier { + if let Some((last_frontier, last_id, last_version)) = last { + // Just in case the sink was recreated, check both sink ID and version to see if it was us. + if last_id == sink_id && last_version == sink_version && last_frontier == *frontier { // Our own commit for this batch is already on the table. // We must've missed the response. info!( @@ -1030,28 +1032,36 @@ async fn load_or_create_table( } /// Find the most recent Materialize frontier from Iceberg snapshots. +/// Returns: (frontier, sink id, sink version) +/// /// We store the frontier in snapshot metadata to track where we left off after restarts. /// Snapshots with operation="replace" (compactions) don't have our metadata and are skipped. /// The input slice will be sorted by sequence number in descending order. fn retrieve_upper_from_snapshots( snapshots: &mut [Arc], -) -> anyhow::Result, u64)>> { +) -> anyhow::Result, GlobalId, u64)>> { snapshots.sort_by(|a, b| Ord::cmp(&b.sequence_number(), &a.sequence_number())); for snapshot in snapshots { let props = &snapshot.summary().additional_properties; - if let (Some(frontier_json), Some(sink_version_str)) = - (props.get("mz-frontier"), props.get("mz-sink-version")) - { + if let (Some(frontier_json), Some(sink_id_str), Some(sink_version_str)) = ( + props.get("mz-frontier"), + props.get("mz-sink-id"), + props.get("mz-sink-version"), + ) { let frontier: Vec = serde_json::from_str(frontier_json) .context("Failed to deserialize frontier from snapshot properties")?; let frontier = Antichain::from_iter(frontier); + let sink_id = sink_id_str + .parse::() + .context("Failed to parse mz-sink-id from snapshot properties")?; + let sink_version = sink_version_str .parse::() .context("Failed to parse mz-sink-version from snapshot properties")?; - return Ok(Some((frontier, sink_version))); + return Ok(Some((frontier, sink_id, sink_version))); } if snapshot.summary().operation.as_str() != "replace" { // This is a bad heuristic, but we have no real other way to identify compactions @@ -1236,7 +1246,7 @@ fn mint_batch_descriptions<'scope>( let mut snapshots: Vec<_> = table.metadata().snapshots().cloned().collect(); let resume = retrieve_upper_from_snapshots(&mut snapshots)?; let (resume_upper, resume_version) = match resume { - Some((f, v)) => (f, v), + Some((f, _, v)) => (f, v), None => (Antichain::from_elem(Timestamp::minimum()), 0), }; debug!( @@ -2745,6 +2755,7 @@ fn commit_to_iceberg<'scope>( catalog.as_ref(), &conn_namespace, &conn_table, + sink_id, sink_version, &frontier, &batch_lower, From 140290b0df15e6febc8a06a4ef05ab273053637b Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 26 Aug 2026 16:41:53 -0400 Subject: [PATCH 07/12] count commits that failed before reaching the catalog --- src/storage/src/sink/iceberg.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index f12a709b1cedc..4cc2eb66052ae 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -939,6 +939,7 @@ async fn try_commit_batch( Err(CommitError::Local(e)) => { // Nothing was sent to the catalog, so this commit definitely didn't happen. // Next attempt should reload the table and try again from scratch. + metrics.commit_failures.inc(); ( table, RetryResult::RetryableErr(anyhow!("Failed to build iceberg table commit: {}", e)), From 6d0d967d6d386fccd8d523b6531426019e7b7493 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 26 Aug 2026 16:45:27 -0400 Subject: [PATCH 08/12] update comment about writer fencing test --- test/iceberg/mzcompose.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index 227b9d5e4d1ef..875523b7f7bed 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -520,11 +520,13 @@ def workflow_fenced_writer(c: Composition) -> None: """Regression test: once a snapshot with a newer mz-sink-version is on the table, the running sink must stop committing. - Currently fails: iceberg-rust's commit path reloads the table and rebases - onto the newest snapshot before every attempt, so the fenced sink never + Root cause: iceberg-rust's built-in commit path reloads the table and rebases + onto the newest snapshot before every attempt, so the fenced-out sink never sees a conflict and commits right over the newer writer. Its snapshot then becomes the latest one, so even the startup fencing check of a later - restart no longer notices the newer writer.""" + restart no longer notices the newer writer. + + So we can't use iceberg-rust's built-in commit path.""" key = _setup(c) c.run_testdrive_files( From 69c79801a4582acc1f8c61678b09878771cbd691 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 26 Aug 2026 16:49:02 -0400 Subject: [PATCH 09/12] Claude: minor review fixes --- src/storage/src/sink/iceberg.rs | 6 +++--- test/iceberg/mzcompose.py | 37 +++++++++++++-------------------- 2 files changed, 18 insertions(+), 25 deletions(-) diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index 4cc2eb66052ae..ff19e5059e56e 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -902,8 +902,8 @@ async fn try_commit_batch( table, RetryResult::FatalErr(anyhow!( "Iceberg table '{}' has been modified by another writer \ - with version {}. Current sink version: {}. \ - Frontiers may be out of sync, aborting to avoid data loss.", + with version {}. Current sink version: {}. \ + Frontiers may be out of sync, aborting to avoid data loss.", conn_table, last_version, sink_version, @@ -917,7 +917,7 @@ async fn try_commit_batch( table, RetryResult::FatalErr(anyhow!( "Iceberg table '{}' has been modified by another writer. \ - Current frontier: {:?}, last frontier: {:?}.", + Current frontier: {:?}, last frontier: {:?}.", conn_table, frontier, last_frontier, diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index 875523b7f7bed..413df0576ddf9 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -15,13 +15,12 @@ import urllib.request from collections.abc import Callable -from materialize.mzcompose.composition import Composition -from materialize.mzcompose.composition import Service as UpService +from materialize.mzcompose.composition import Composition, Service from materialize.mzcompose.helpers.iceberg import ( get_polaris_access_token, setup_polaris_for_iceberg, ) -from materialize.mzcompose.service import Service +from materialize.mzcompose.service import Service as ServiceDefinition from materialize.mzcompose.services.materialized import Materialized from materialize.mzcompose.services.minio import Mc, Minio from materialize.mzcompose.services.mz import Mz @@ -35,7 +34,7 @@ Minio(), PolarisBootstrap(), Polaris(), - Service( + ServiceDefinition( "polaris-proxy", { "image": "python:3.11-slim", @@ -70,8 +69,8 @@ def _setup( c.up( "postgres", "materialized", - UpService("polaris-bootstrap", idle=True), - UpService("polaris", idle=True), + Service("polaris-bootstrap", idle=True), + Service("polaris", idle=True), ) _, key = setup_polaris_for_iceberg( c, vended=vended, static_credentials=static_credentials @@ -79,6 +78,16 @@ def _setup( return key +def await_condition(what: str, timeout: float, check: Callable[[], bool]) -> None: + """Poll `check` until it returns True, or raise after `timeout` seconds.""" + deadline = time.time() + timeout + while time.time() < deadline: + if check(): + return + time.sleep(0.5) + raise AssertionError(f"timed out waiting for {what}") + + def workflow_default(c: Composition) -> None: def process(name: str) -> None: if name == "default": @@ -456,14 +465,6 @@ def proxy_status() -> dict[str, int]: with urllib.request.urlopen(f"{proxy_base}/__control/status") as resp: return json.loads(resp.read()) - def await_condition(what: str, timeout: float, check: Callable[[], bool]) -> None: - deadline = time.time() + timeout - while time.time() < deadline: - if check(): - return - time.sleep(0.5) - raise AssertionError(f"timed out waiting for {what}") - # Arm the drop only once the sink has a commit through the proxy, so the # dropped commit is a steady-state one rather than table bootstrap. await_condition( @@ -552,14 +553,6 @@ def load_metadata() -> dict | None: return None raise - def await_condition(what: str, timeout: float, check: Callable[[], bool]) -> None: - deadline = time.time() + timeout - while time.time() < deadline: - if check(): - return - time.sleep(0.5) - raise AssertionError(f"timed out waiting for {what}") - def first_commit_done() -> bool: meta = load_metadata() return meta is not None and meta.get("current-snapshot-id") not in (None, -1) From a60afe21430962b605e2de3c05bc272cfbe86700 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Wed, 2 Sep 2026 15:49:45 -0400 Subject: [PATCH 10/12] Claude: test for overlapping writers with different commit intervals --- test/iceberg/mzcompose.py | 177 ++++++++++++++++++++++++++++ test/iceberg/prefix-writer-setup.td | 42 +++++++ 2 files changed, 219 insertions(+) create mode 100644 test/iceberg/prefix-writer-setup.td diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index 413df0576ddf9..a0b9a4f79ee9c 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -659,6 +659,183 @@ def forge_newer_writer_snapshot() -> int: time.sleep(1) +def workflow_prefix_writer(c: Composition) -> None: + """Regression test: a snapshot whose frontier falls strictly inside the + sink's pending batch must stop the sink from committing that batch. + + How this could happen in practice: + + Suppose we have a sink with commit interval 30s. + Then we alter the sink's commit interval to 1s. + + Let the table's frontier be X. + The new version of the sink can commit a batch [X, X + 1s). + The old version of the sink will try to commit a batch [X, X + 30s). + + Committing both batches would result in duplicate writes.""" + key = _setup(c) + + c.run_testdrive_files( + f"--var=s3-access-key={key}", + "--var=aws-endpoint=minio:9000", + "prefix-writer-setup.td", + ) + + token = get_polaris_access_token(c) + table_url = ( + f"http://localhost:{c.port('polaris', 8181)}" + "/api/catalog/v1/default_catalog/namespaces/default_namespace/tables/prefix_table" + ) + headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"} + + def load_metadata() -> dict | None: + req = urllib.request.Request(table_url, headers=headers) + try: + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read())["metadata"] + except urllib.error.HTTPError as e: + if e.code == 404: + return None + raise + + def load_current() -> tuple[dict, dict] | None: + """Table metadata plus its current snapshot, once the sink has committed.""" + meta = load_metadata() + if meta is None: + return None + snapshot_id = meta.get("current-snapshot-id") + if snapshot_id in (None, -1): + return None + snapshot = next(s for s in meta["snapshots"] if s["snapshot-id"] == snapshot_id) + return meta, snapshot + + def frontier_of(snapshot: dict) -> int: + elements = json.loads(snapshot["summary"]["mz-frontier"]) + assert len(elements) == 1, f"expected a single frontier element: {elements}" + return elements[0] + + def committed_frontier() -> int | None: + current = load_current() + return None if current is None else frontier_of(current[1]) + + await_condition( + "first sink commit", timeout=60, check=lambda: committed_frontier() is not None + ) + first_frontier = committed_frontier() + assert first_frontier is not None + + # The first commit covers the source snapshot and ends just past the as_of. + # The catch-up batch that follows it can be just as narrow, so wait for that + # one to commit too. Only then is the pending batch a full commit interval + # wide, which is what makes room for a frontier strictly inside it. + def catchup_committed() -> bool: + frontier = committed_frontier() + return frontier is not None and frontier > first_frontier + + await_condition("catch-up commit", timeout=120, check=catchup_committed) + + def forge_prefix_snapshot() -> tuple[int, int]: + """Commit a snapshot claiming the running sink's own ID and version, one + second past its last committed frontier: what a second incarnation with + a shorter commit interval leaves behind. Reuses the current snapshot's + manifest list so the table stays readable without writing new files. + Returns the forged sequence number and frontier.""" + current = load_current() + assert current is not None + meta, snapshot = current + current_id = snapshot["snapshot-id"] + forged_seq = meta["last-sequence-number"] + 1 + # Frontiers are milliseconds, so one second past the last commit. + forged_frontier = frontier_of(snapshot) + 1000 + body = json.dumps( + { + "requirements": [ + { + "type": "assert-ref-snapshot-id", + "ref": "main", + "snapshot-id": current_id, + } + ], + "updates": [ + { + "action": "add-snapshot", + "snapshot": { + "snapshot-id": current_id + 1, + "parent-snapshot-id": current_id, + "sequence-number": forged_seq, + "timestamp-ms": int(time.time() * 1000), + "manifest-list": snapshot["manifest-list"], + "schema-id": meta["current-schema-id"], + "summary": { + "operation": "append", + "mz-sink-id": snapshot["summary"]["mz-sink-id"], + "mz-frontier": json.dumps([forged_frontier]), + "mz-sink-version": snapshot["summary"][ + "mz-sink-version" + ], + }, + }, + }, + { + "action": "set-snapshot-ref", + "ref-name": "main", + "type": "branch", + "snapshot-id": current_id + 1, + }, + ], + } + ).encode() + req = urllib.request.Request( + table_url, data=body, headers=headers, method="POST" + ) + with urllib.request.urlopen(req) as resp: + resp.read() + return forged_seq, forged_frontier + + # The sink may commit between reading the metadata and posting the forged + # snapshot, failing our requirement. Retry on conflict with fresh metadata. + for attempt in range(5): + try: + forged_seq, forged_frontier = forge_prefix_snapshot() + break + except urllib.error.HTTPError as e: + if e.code != 409 or attempt == 4: + raise + else: + raise AssertionError("unreachable") + + # Put a row in the range the forged snapshot claims to have committed, so the + # overlap the sink must refuse is not empty. + c.sql("INSERT INTO prefix_src VALUES (4, 'd')") + + deadline = time.time() + 120 + while True: + meta = load_metadata() + assert meta is not None + for snapshot in meta["snapshots"]: + if snapshot["sequence-number"] > forged_seq: + raise AssertionError( + "sink committed a batch overlapping the forged writer's " + f"range: forged frontier {forged_frontier}, batch started at " + f"{forged_frontier - 1000}, snapshot {snapshot['snapshot-id']} " + f"summary {snapshot['summary']}" + ) + + status_rows = c.sql_query( + "SELECT s.status, COALESCE(s.error, '') " + "FROM mz_internal.mz_sink_statuses s " + "JOIN mz_sinks ON s.id = mz_sinks.id " + "WHERE mz_sinks.name = 'prefix_sink'" + ) + assert status_rows, "prefix_sink not found in mz_sink_statuses" + status, error = status_rows[0] + if status != "running" and ("another writer" in error or "Fenced off" in error): + return + if time.time() > deadline: + raise AssertionError(f"sink not fenced: status={status!r} error={error!r}") + time.sleep(1) + + def workflow_large_upsert_batch(c: Composition) -> None: """Regression test for database-issues#11326: DeltaWriter seen_rows eviction caused equality deletes within the same snapshot, which diff --git a/test/iceberg/prefix-writer-setup.td b/test/iceberg/prefix-writer-setup.td new file mode 100644 index 0000000000000..c7eeed825f673 --- /dev/null +++ b/test/iceberg/prefix-writer-setup.td @@ -0,0 +1,42 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +> CREATE SECRET access_key_secret AS '${arg.s3-access-key}' + +> CREATE CONNECTION aws_conn TO AWS ( + ACCESS KEY ID = 'tduser', + SECRET ACCESS KEY = SECRET access_key_secret, + ENDPOINT = '${arg.aws-endpoint}', + REGION = 'us-east-1' + ); + +> CREATE CONNECTION polaris TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'http://polaris:8181/api/catalog', + CREDENTIAL = 'root:root', + WAREHOUSE = 'default_catalog', + SCOPE = 'PRINCIPAL_ROLE:ALL' + ); + +> CREATE TABLE prefix_src (id INT, val TEXT); + +> INSERT INTO prefix_src VALUES (1, 'a'), (2, 'b'), (3, 'c'); + +# The long commit interval makes the sink's pending batch wide, so a forged +# frontier one second past the last committed one lands strictly inside it. +> CREATE SINK prefix_sink + FROM prefix_src + INTO ICEBERG CATALOG CONNECTION polaris ( + NAMESPACE 'default_namespace', + TABLE 'prefix_table' + ) + USING AWS CONNECTION aws_conn + KEY (id) NOT ENFORCED + MODE UPSERT + WITH (COMMIT INTERVAL '30s'); From 34cc35fbbdf7d2a9c7242e4c58881993bdfceb30 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Thu, 3 Sep 2026 15:06:56 -0400 Subject: [PATCH 11/12] Check that already committed frontier does not contain pending batch. --- src/storage/src/sink/iceberg.rs | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/src/storage/src/sink/iceberg.rs b/src/storage/src/sink/iceberg.rs index ff19e5059e56e..2f73e194f27ba 100644 --- a/src/storage/src/sink/iceberg.rs +++ b/src/storage/src/sink/iceberg.rs @@ -853,7 +853,6 @@ async fn try_commit_batch( conn_table: &str, sink_id: GlobalId, sink_version: u64, - frontier: &Antichain, batch_lower: &Antichain, batch_upper: &Antichain, metrics: &IcebergSinkMetrics, @@ -883,7 +882,7 @@ async fn try_commit_batch( }; if let Some((last_frontier, last_id, last_version)) = last { // Just in case the sink was recreated, check both sink ID and version to see if it was us. - if last_id == sink_id && last_version == sink_version && last_frontier == *frontier { + if last_id == sink_id && last_version == sink_version && last_frontier == *batch_upper { // Our own commit for this batch is already on the table. // We must've missed the response. info!( @@ -911,15 +910,17 @@ async fn try_commit_batch( ); } - if PartialOrder::less_equal(frontier, &last_frontier) { - // Someone else has already written this batch (or an even later batch). + if PartialOrder::less_equal(batch_upper, &last_frontier) + || PartialOrder::less_than(batch_lower, &last_frontier) + { + // This batch contains records someone else has already written. return ( table, RetryResult::FatalErr(anyhow!( "Iceberg table '{}' has been modified by another writer. \ Current frontier: {:?}, last frontier: {:?}.", conn_table, - frontier, + batch_upper, last_frontier, )), ); @@ -2744,7 +2745,6 @@ fn commit_to_iceberg<'scope>( let catalog = Arc::clone(&catalog); let conn_namespace = connection.namespace.clone(); let conn_table = connection.table.clone(); - let frontier = frontier.clone(); let batch_lower = batch.0.clone(); let batch_upper = batch.1.clone(); async move { @@ -2758,7 +2758,6 @@ fn commit_to_iceberg<'scope>( &conn_table, sink_id, sink_version, - &frontier, &batch_lower, &batch_upper, &metrics, From 9a8c0ae995a31efd84a5ae8c644f3079aae1ce84 Mon Sep 17 00:00:00 2001 From: Kynan Rilee Date: Tue, 15 Sep 2026 13:48:43 -0400 Subject: [PATCH 12/12] fix broken formatting in iceberg nightly test --- test/iceberg/storage-provider.td | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/test/iceberg/storage-provider.td b/test/iceberg/storage-provider.td index b09a99efdb5e6..f45e606dbdc80 100644 --- a/test/iceberg/storage-provider.td +++ b/test/iceberg/storage-provider.td @@ -31,7 +31,7 @@ ALTER SYSTEM SET enable_default_connection_validation = false > SHOW CREATE CONNECTION sp_default name create_sql --------------------------------- -materialize.public.sp_default "CREATE CONNECTION materialize.public.sp_default TO ICEBERG CATALOG (CATALOG TYPE = 'rest', CREDENTIAL = SECRET materialize.public.sp_cred, URL = 'https://example.invalid/api/catalog', WAREHOUSE = 'wh');" +materialize.public.sp_default "CREATE CONNECTION materialize.public.sp_default\nTO ICEBERG CATALOG\n (\n CREDENTIAL = SECRET materialize.public.sp_cred,\n CATALOG TYPE = 'REST',\n URL = 'https://example.invalid/api/catalog',\n WAREHOUSE = 'wh'\n );" # Each accepted value round-trips through the catalog. > CREATE CONNECTION sp_s3 TO ICEBERG CATALOG (