From 6072b9f0c139c3c069136fdc9aeafc291ac0d188 Mon Sep 17 00:00:00 2001 From: Brad Heller Date: Mon, 14 Sep 2026 21:40:57 +0100 Subject: [PATCH 1/4] feat: let the server decide the maximum bundle size (#365) * feat: let the server decide the maximum bundle size The CLI used to hard-reject any package over 50 MB before uploading. The server enforces its own bundle-size limit and reports it in its error response, so the client-side check was redundant, blocked servers configured with higher limits, and required a CLI release to change. The check is now a non-fatal warning above 50 MB, and MAX_PACKAGE_SIZE is removed from tower-package in favor of a local advisory threshold in the deploy path. * feat: raise the large-package warning threshold to 500 MB * fix: address deploy size warning review feedback --- crates/tower-cmd/src/util/deploy.rs | 60 +++++++++++++++++++++++++---- crates/tower-package/src/core.rs | 2 - crates/tower-package/src/lib.rs | 2 +- 3 files changed, 54 insertions(+), 10 deletions(-) diff --git a/crates/tower-cmd/src/util/deploy.rs b/crates/tower-cmd/src/util/deploy.rs index 3ce5c5de..660c2abe 100644 --- a/crates/tower-cmd/src/util/deploy.rs +++ b/crates/tower-cmd/src/util/deploy.rs @@ -14,6 +14,12 @@ use tower_api::apis::Error; use tower_api::apis::ResponseContent; use tower_api::models::DeployAppResponse; +/// Advisory only: the server enforces the actual bundle-size limit and reports +/// it in its error response when a bundle is too large. +const LARGE_PACKAGE_WARNING_THRESHOLD: u64 = 500 * 1024 * 1024; + +/// Streams a package to the server, reporting progress and leaving size enforcement +/// to the server. Large-package warnings are suppressed in JSON output mode. pub async fn upload_file_with_progress( out: &output::Out, api_config: &Configuration, @@ -36,13 +42,10 @@ pub async fn upload_file_with_progress( let metadata = file.metadata().await?; let file_size = metadata.len(); - // Check if bundle size exceeds the maximum allowed size - if file_size > tower_package::MAX_PACKAGE_SIZE { - let size_mb = file_size as f64 / (1024.0 * 1024.0); - let max_mb = tower_package::MAX_PACKAGE_SIZE as f64 / (1024.0 * 1024.0); - out.die(&format!( - "Your App is too big! ({:.2} MB) exceeds maximum allowed size ({:.0} MB). Please consider reducing app size by removing unnecessary files or import_paths in the Towerfile.", - size_mb, max_mb + if file_size > LARGE_PACKAGE_WARNING_THRESHOLD { + out.note(&format!( + "Warning: Your app package is large ({}). The server may reject it depending on its configured maximum bundle size. You can reduce app size by removing unnecessary files or import_paths in the Towerfile.\n", + indicatif::BinaryBytes(file_size) )); } @@ -57,6 +60,7 @@ pub async fn upload_file_with_progress( .header("X-Tower-Checksum-SHA256", package_hash) .header("Content-Type", content_type) .header("Content-Encoding", "gzip") + .header(reqwest::header::CONTENT_LENGTH, file_size) .body(Body::wrap_stream(progress_stream)); // When supplied, ask the server to reuse an existing AppVersion that was @@ -174,3 +178,45 @@ pub async fn deploy_app_package( Ok(response) } + +#[cfg(test)] +mod tests { + use super::*; + use axum::{body::Bytes, http::HeaderMap, routing::post, Json, Router}; + use std::io::Write; + + #[tokio::test] + async fn upload_advertises_exact_streamed_length() { + let _ = rustls::crypto::ring::default_provider().install_default(); + let payload = b"test package contents"; + let mut package = tempfile::NamedTempFile::new().unwrap(); + package.write_all(payload).unwrap(); + + let app = Router::new().route( + "/deploy", + post(|headers: HeaderMap, body: Bytes| async move { + assert_eq!(headers["content-length"], "21"); + assert_eq!(headers["content-encoding"], "gzip"); + assert!(!headers.contains_key("transfer-encoding")); + assert_eq!(body.as_ref(), b"test package contents"); + Json(DeployAppResponse::default()) + }), + ); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let address = listener.local_addr().unwrap(); + let server = tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let result = upload_file_with_progress( + &output::Out::sink(), + &Configuration::default(), + format!("http://{address}/deploy"), + package.path().to_path_buf(), + "application/tar", + None, + Box::new(|_, _| {}), + ) + .await; + server.abort(); + result.unwrap(); + } +} diff --git a/crates/tower-package/src/core.rs b/crates/tower-package/src/core.rs index b376f83a..69383c4f 100644 --- a/crates/tower-package/src/core.rs +++ b/crates/tower-package/src/core.rs @@ -15,8 +15,6 @@ use crate::towerfile::{Parameter, Towerfile}; // 3 - Change checksum algorithm to be cross-platform pub const CURRENT_PACKAGE_VERSION: i32 = 3; -pub const MAX_PACKAGE_SIZE: u64 = 50 * 1024 * 1024; - #[derive(Debug, Snafu)] pub enum Error { #[snafu(display("Invalid path"))] diff --git a/crates/tower-package/src/lib.rs b/crates/tower-package/src/lib.rs index d876067b..3f8520ed 100644 --- a/crates/tower-package/src/lib.rs +++ b/crates/tower-package/src/lib.rs @@ -3,7 +3,7 @@ mod towerfile; pub use core::{ build_package, compute_sha256_bytes, compute_sha256_package, normalize_path, BuiltPackage, - Entry, Manifest, PackageInputs, CURRENT_PACKAGE_VERSION, MAX_PACKAGE_SIZE, + Entry, Manifest, PackageInputs, CURRENT_PACKAGE_VERSION, }; pub use towerfile::{App, Parameter, Towerfile}; From 2656cc1dac79caafca3682634800af5c5c26b2e4 Mon Sep 17 00:00:00 2001 From: Ben Lovell Date: Tue, 15 Sep 2026 12:22:38 +0200 Subject: [PATCH 2/4] fix: follow a finished run's logs through the stream endpoint (#363) * test: model the stored-logs persistence lag in the mock API server * fix: follow a finished run's logs through the stream endpoint --- crates/tower-cmd/src/apps.rs | 26 +++++++-------------- tests/integration/features/cli_runs.feature | 2 +- tests/mock-api-server/main.py | 22 +++++++++++++++-- 3 files changed, 30 insertions(+), 20 deletions(-) diff --git a/crates/tower-cmd/src/apps.rs b/crates/tower-cmd/src/apps.rs index bccc33cd..e2cd81be 100644 --- a/crates/tower-cmd/src/apps.rs +++ b/crates/tower-cmd/src/apps.rs @@ -405,30 +405,22 @@ async fn drain_stream_with_grace( .await; } -/// Follows the logs of a run: prints stored logs for a finished run, waits for -/// a not-yet-started run, and otherwise attaches to the live log stream with -/// reconnects, dedup, and independent completion detection. +/// Follows the logs of a run: waits for a not-yet-started run, then attaches +/// to the live log stream. Finished runs take the same path — the stream +/// replays from the start and the server closes it — because stored logs can +/// trail a run's terminal status. The post-stream catch-up covers runs whose +/// stream has expired. async fn follow_run_logs(out: &output::Out, config: &Config, name: &str, seq: i64) { let mut tracker = LineTracker::new(); let run = describe_run_or_die(out, config, name, seq).await; - - match run_phase(&run.status) { - RunPhase::Terminal => { - print_stored_logs(out, config, name, seq, &mut tracker).await; - return; - } - RunPhase::NotStarted => match wait_for_run_start(out, config, name, seq).await { - WaitOutcome::Started => {} - WaitOutcome::Finished => { - print_stored_logs(out, config, name, seq, &mut tracker).await; - return; - } + if run_phase(&run.status) == RunPhase::NotStarted { + match wait_for_run_start(out, config, name, seq).await { + WaitOutcome::Started | WaitOutcome::Finished => {} WaitOutcome::TimedOut => { out.die("Timed out waiting for the run to start. The runner may be unavailable."); } - }, - RunPhase::InProgress => {} + } } stream_logs_with_reconnect(out, config, name, seq, &run.dollar_link, &mut tracker).await; diff --git a/tests/integration/features/cli_runs.feature b/tests/integration/features/cli_runs.feature index 1a3c91f7..d559fe52 100644 --- a/tests/integration/features/cli_runs.feature +++ b/tests/integration/features/cli_runs.feature @@ -56,7 +56,7 @@ Feature: CLI Run Commands And the output should show "Warning: No new logs available" Scenario: CLI apps logs --follow on a finished run prints stored logs exactly once - Given I have a simple hello world application named "app-logs-after-completion" + Given I have a simple hello world application named "app-logs-finished" When I run "tower deploy --create" via CLI And I run "tower run --detached" via CLI and capture the run number And I wait for 2 seconds diff --git a/tests/mock-api-server/main.py b/tests/mock-api-server/main.py index 421f58b4..6e12f486 100644 --- a/tests/mock-api-server/main.py +++ b/tests/mock-api-server/main.py @@ -364,13 +364,15 @@ async def describe_run(name: str, seq: int): # For logs-after-completion test apps, complete quickly to test log draining # Use 1 second so CLI has time to start streaming before completion - completion_threshold = 1.0 if "logs-after-completion" in name else 5.0 + quick = "logs-after-completion" in name or "logs-finished" in name + completion_threshold = 1.0 if quick else 5.0 if elapsed > completion_threshold: run_data["status"] = "exited" run_data["status_group"] = "successful" run_data["exit_code"] = 0 - run_data["ended_at"] = now_time.isoformat() + if not run_data.get("ended_at"): + run_data["ended_at"] = now_time.isoformat() return { "run": run_data, @@ -685,12 +687,28 @@ def make_warning_event(content: str, timestamp: str, end_of_stream: bool = False STREAM_COMPLETE_WARNING = "stream complete" +# The real server persists log lines a beat after a run turns terminal; +# fetch too soon and the list is empty. Mirror that window. +STORED_LOGS_PERSIST_LAG = datetime.timedelta(seconds=2.5) + + @app.get("/v1/apps/{name}/runs/{seq}/logs") async def describe_run_logs(name: str, seq: int): """Mock endpoint for getting run logs.""" if name not in mock_apps_db: raise HTTPException(status_code=404, detail=f"App '{name}' not found") + for run_data in mock_runs_db.values(): + if run_data["app_name"] == name and run_data["number"] == seq: + ended = run_data.get("ended_at") + if ( + not ended + or datetime.datetime.now() - datetime.datetime.fromisoformat(ended) + < STORED_LOGS_PERSIST_LAG + ): + return {"log_lines": []} + break + return { "log_lines": [ make_log_data(seq, line_num, content, timestamp) From c44f008bcd5c11ab344bb694243fa77c7b89834d Mon Sep 17 00:00:00 2001 From: Konstantinos St Date: Fri, 18 Sep 2026 02:51:10 +0200 Subject: [PATCH 3/4] fix: become compatible with pyiceberg 0.12 (#368) --- INSTALL-AND-REFERENCE.md | 16 ++++- src/tower/_tables.py | 30 ++++++-- tests/tower/test_table_retries.py | 6 +- tests/tower/test_table_schemas.py | 22 +++++- tests/tower/test_tables.py | 115 ++++++++++++------------------ 5 files changed, 107 insertions(+), 82 deletions(-) diff --git a/INSTALL-AND-REFERENCE.md b/INSTALL-AND-REFERENCE.md index 0cccf67e..e5ab37b1 100644 --- a/INSTALL-AND-REFERENCE.md +++ b/INSTALL-AND-REFERENCE.md @@ -146,9 +146,19 @@ PyIceberg, which assigns field IDs and preserves nested nullability and `b"doc"` metadata. Timestamp units from seconds through microseconds, UTC-zoned microsecond timestamps, `time64[us]`, `date32`, and Decimal128 values up to precision 38 are supported. Nanosecond timestamps are rejected by default instead of being silently downcast, as are -`time32`, `time64[ns]`, `date64`, Float16, Decimal256, and non-UTC zoned timestamps. Convert -those fields explicitly before creating the table when the loss is acceptable. PyIceberg's -native validation exceptions propagate unchanged. +`time32`, `time64[ns]`, `date64`, Decimal256, and non-UTC zoned timestamps. Float16 handling +follows the installed PyIceberg version: PyIceberg 0.12 and newer accept it and widen it +losslessly to Iceberg `float` (Arrow float32 on read), while PyIceberg 0.11 rejects it. +Convert unsupported fields explicitly before creating the table when the loss is acceptable. +PyIceberg's native validation exceptions propagate unchanged. + +With PyIceberg 0.12 and newer, compatible commit races are retried internally. When an +upsert or delete was planned from stale data and conflicts with a concurrent change, +PyIceberg raises `pyiceberg.exceptions.ValidationException`; Tower does not automatically +re-run the complete operation with last-writer-wins behavior. The failed operation does not +commit. Reload, reconcile the newer data, and explicitly retry only when that is appropriate +for the application. Tower's per-call retry options still handle a surfaced +`CommitFailedException`, including for supported PyIceberg 0.11 installations. ### dbt Core support diff --git a/src/tower/_tables.py b/src/tower/_tables.py index 40d5b53b..78fd73ce 100644 --- a/src/tower/_tables.py +++ b/src/tower/_tables.py @@ -294,6 +294,8 @@ def _commit_with_retry( initial_retry_ceiling_seconds, _MAX_COMMIT_RETRY_DELAY_SECONDS ) + # A PyIceberg ValidationException means that data changed incompatibly. + # It deliberately propagates instead of becoming last-writer-wins. for attempt in range(max_retries + 1): try: return operation() @@ -369,12 +371,17 @@ def upsert( retry_delay_seconds: float = 0.5, ) -> TTable: """ - Performs an upsert operation (update or insert) on the Iceberg table. In case of commit conflicts, reloads the metadata and retries. + Performs an upsert operation (update or insert) on the Iceberg table. + + PyIceberg 0.12 and newer retry compatible commit races internally. If + PyIceberg detects that a concurrent write changed data relevant to this + upsert, its ``ValidationException`` propagates instead of Tower retrying + the complete operation with last-writer-wins behavior. This method will: - Update existing rows if they match the join columns - Insert new rows if no match is found - - Retry for max_retries if commits fail + - Preserve PyIceberg's concurrency validation semantics All operations are case-sensitive by default. Args: @@ -382,8 +389,9 @@ def upsert( must match the schema of the target table. join_cols (Optional[list[str]]): The columns that form the key to match rows on. If not provided, all columns will be used for matching. - max_retries (int): Maximum number of retry attempts on commit conflicts. - Defaults to 5. + max_retries (int): Maximum number of Tower retry attempts when PyIceberg + surfaces a ``CommitFailedException``. This does not apply to validated + data conflicts. Defaults to 5. retry_delay_seconds (float): Maximum randomized wait before the first retry, in seconds. The maximum doubles after each conflict but never exceeds 30 seconds; values above 30 are treated as 30. Defaults to 0.5 seconds. @@ -393,6 +401,8 @@ def upsert( Raises: CommitFailedException: If all retry attempts are exhausted. + pyiceberg.exceptions.ValidationException: If PyIceberg detects an + incompatible concurrent write. Note: - The operation is always case-sensitive @@ -445,7 +455,10 @@ def delete( ) -> TTable: """ Deletes rows from the Iceberg table that match the specified filter conditions. - In case of commit conflicts, reloads the metadata and retries. + PyIceberg 0.12 and newer retry compatible commit races internally, but reject + incompatible concurrent data changes with ``ValidationException``. Tower lets + that conflict propagate rather than retrying the complete delete against newer + data. This method removes rows from the table based on the provided filter expressions. The operation is always case-sensitive. Note that the number of deleted rows @@ -454,8 +467,9 @@ def delete( Args: filters (str | BooleanExpression): A SQL-like string or a PyIceberg boolean expression. Use ``Table.column()`` to construct expressions. - max_retries (int): Maximum number of retry attempts on commit conflicts. - Defaults to 5. + max_retries (int): Maximum number of Tower retry attempts when PyIceberg + surfaces a ``CommitFailedException``. This does not apply to validated + data conflicts. Defaults to 5. retry_delay_seconds (float): Maximum randomized wait before the first retry, in seconds. The maximum doubles after each conflict but never exceeds 30 seconds; values above 30 are treated as 30. Defaults to 0.5 seconds. @@ -465,6 +479,8 @@ def delete( Raises: CommitFailedException: If all retry attempts are exhausted. + pyiceberg.exceptions.ValidationException: If PyIceberg detects an + incompatible concurrent write. Note: - The operation is always case-sensitive diff --git a/tests/tower/test_table_retries.py b/tests/tower/test_table_retries.py index 6de9a76e..0df1be57 100644 --- a/tests/tower/test_table_retries.py +++ b/tests/tower/test_table_retries.py @@ -13,6 +13,7 @@ ServerError, ServiceUnavailableError, UnauthorizedError, + ValidationException, WaitingForLockException, ) @@ -217,14 +218,13 @@ def test_commit_retry_exhaustion_preserves_final_exception(monkeypatch, max_retr ServerError, BadRequestError, WaitingForLockException, + ValidationException, httpx.TimeoutException, httpx.ConnectError, ValueError, ], ) -def test_mutation_errors_other_than_commit_conflicts_are_not_retried( - monkeypatch, exception_type -): +def test_non_retryable_mutation_errors_are_not_retried(monkeypatch, exception_type): failure = exception_type("not retryable") iceberg_table = FakeMutationTable([failure]) table = make_table(iceberg_table) diff --git a/tests/tower/test_table_schemas.py b/tests/tower/test_table_schemas.py index 2683f05e..7bf5751e 100644 --- a/tests/tower/test_table_schemas.py +++ b/tests/tower/test_table_schemas.py @@ -1,3 +1,5 @@ +from importlib.metadata import version + import pyarrow as pa import pytest from pyiceberg import types as iceberg_types @@ -9,6 +11,11 @@ from tower._context import TowerContext +_PYICEBERG_SUPPORTS_FLOAT16 = tuple( + int(component) for component in version("pyiceberg").split(".")[:2] +) >= (0, 12) + + class RecordingCatalog: def __init__(self): self.schemas = [] @@ -201,6 +208,20 @@ def test_pyiceberg_accepts_supported_arrow_precision( assert table.schema().find_field("value").field_type == iceberg_type +def test_float16_schema_follows_pyiceberg_version(in_memory_schema_catalog): + schema = pa.schema([pa.field("value", pa.float16())]) + + if not _PYICEBERG_SUPPORTS_FLOAT16: + with pytest.raises(UnsupportedPyArrowTypeException): + make_reference(in_memory_schema_catalog, "float16").create(schema) + return + + make_reference(in_memory_schema_catalog, "float16").create(schema) + + table = in_memory_schema_catalog.load_table("default.float16") + assert table.schema().find_field("value").field_type == iceberg_types.FloatType() + + @pytest.mark.parametrize( ("name", "arrow_type"), [ @@ -208,7 +229,6 @@ def test_pyiceberg_accepts_supported_arrow_precision( ("timestamp_non_utc", pa.timestamp("us", tz="Europe/Berlin")), ("time32", pa.time32("s")), ("time_ns", pa.time64("ns")), - ("float16", pa.float16()), ("date64", pa.date64()), ("decimal256", pa.decimal256(38, 10)), ], diff --git a/tests/tower/test_tables.py b/tests/tower/test_tables.py index acd428f7..3d574988 100644 --- a/tests/tower/test_tables.py +++ b/tests/tower/test_tables.py @@ -3,6 +3,7 @@ import datetime import tempfile import pathlib +from importlib.metadata import version from urllib.parse import urljoin from urllib.request import pathname2url import threading @@ -12,7 +13,7 @@ import pyarrow as pa from pyiceberg.catalog.memory import InMemoryCatalog from pyiceberg.catalog.sql import SqlCatalog -from pyiceberg.exceptions import CommitFailedException +from pyiceberg.exceptions import CommitFailedException, ValidationException import concurrent.futures @@ -28,6 +29,11 @@ ) +_PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION = tuple( + int(component) for component in version("pyiceberg").split(".")[:2] +) >= (0, 12) + + class FakeLoadedTable: def __init__(self, mode: str, identifier: str): self.mode = mode @@ -622,8 +628,9 @@ def test_upsert_to_tables(in_memory_catalog): assert res["age"].item() == 26 -def test_upsert_concurrent_writes_with_retry(sql_catalog): - """Test that concurrent upserts succeed with retry logic handling conflicts.""" +def test_stale_upserts_to_different_rows_follow_pyiceberg_conflict_semantics( + sql_catalog, +): schema = pa.schema( [ pa.field("ticker", pa.string()), @@ -645,40 +652,28 @@ def test_upsert_concurrent_writes_with_retry(sql_catalog): ) table.insert(initial_data) - retry_count = {"value": 0} - retry_lock = threading.Lock() - - def upsert_ticker(ticker: str, new_price: float): - t = tower.tables("concurrent_test", catalog=sql_catalog).load() - - original_refresh = t._table.refresh + first_writer = tower.tables("concurrent_test", catalog=sql_catalog).load() + stale_writer = tower.tables("concurrent_test", catalog=sql_catalog).load() - def tracked_refresh(): - with retry_lock: - retry_count["value"] += 1 - return original_refresh() - - t._table.refresh = tracked_refresh - - data = pa.Table.from_pylist( - [{"ticker": ticker, "date": "2024-01-01", "price": new_price}], + first_writer.upsert( + pa.Table.from_pylist( + [{"ticker": "AAPL", "date": "2024-01-01", "price": 150.0}], schema=schema, - ) - t.upsert(data, join_cols=["ticker", "date"]) - return ticker - - with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor: - futures = [ - executor.submit(upsert_ticker, "AAPL", 150.0), - executor.submit(upsert_ticker, "GOOGL", 250.0), - executor.submit(upsert_ticker, "MSFT", 350.0), - ] - results = [f.result() for f in concurrent.futures.as_completed(futures)] + ), + join_cols=["ticker", "date"], + ) - assert len(results) == 3 - assert ( - retry_count["value"] > 0 - ), "Expected at least one retry due to concurrent conflicts" + stale_upsert = pa.Table.from_pylist( + [{"ticker": "GOOGL", "date": "2024-01-01", "price": 250.0}], + schema=schema, + ) + if _PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION: + with pytest.raises(ValidationException): + stale_writer.upsert(stale_upsert, join_cols=["ticker", "date"]) + else: + # PyIceberg 0.11 surfaces CommitFailedException, so Tower refreshes and + # replans the whole operation for supported older installations. + stale_writer.upsert(stale_upsert, join_cols=["ticker", "date"]) final_table = tower.tables("concurrent_test", catalog=sql_catalog).load() df = final_table.read() @@ -688,12 +683,13 @@ def tracked_refresh(): ticker_prices = {row["ticker"]: row["price"] for row in df.iter_rows(named=True)} assert ticker_prices["AAPL"] == 150.0 - assert ticker_prices["GOOGL"] == 250.0 - assert ticker_prices["MSFT"] == 350.0 + assert ticker_prices["GOOGL"] == ( + 200.0 if _PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION else 250.0 + ) + assert ticker_prices["MSFT"] == 300.0 -def test_upsert_concurrent_writes_same_row(sql_catalog): - """Test concurrent upserts to the SAME row - last write wins.""" +def test_stale_upserts_to_same_row_follow_pyiceberg_conflict_semantics(sql_catalog): schema = pa.schema( [ pa.field("id", pa.int64()), @@ -710,37 +706,20 @@ def test_upsert_concurrent_writes_same_row(sql_catalog): ) table.insert(initial_data) - retry_count = {"value": 0} - retry_lock = threading.Lock() - - def upsert_counter(value: int): - t = tower.tables("concurrent_same_row_test", catalog=sql_catalog).load() - - original_refresh = t._table.refresh - - def tracked_refresh(): - with retry_lock: - retry_count["value"] += 1 - return original_refresh() - - t._table.refresh = tracked_refresh + first_writer = tower.tables("concurrent_same_row_test", catalog=sql_catalog).load() + stale_writer = tower.tables("concurrent_same_row_test", catalog=sql_catalog).load() - data = pa.Table.from_pylist( - [{"id": 1, "counter": value}], - schema=schema, - ) - t.upsert(data, join_cols=["id"]) - return value - - with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor: - futures = [executor.submit(upsert_counter, i) for i in range(1, 6)] - results = [f.result() for f in concurrent.futures.as_completed(futures)] - - assert len(results) == 5 + first_writer.upsert( + pa.Table.from_pylist([{"id": 1, "counter": 1}], schema=schema), + join_cols=["id"], + ) - assert ( - retry_count["value"] > 0 - ), "Expected at least one retry due to concurrent conflicts" + stale_upsert = pa.Table.from_pylist([{"id": 1, "counter": 2}], schema=schema) + if _PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION: + with pytest.raises(ValidationException): + stale_writer.upsert(stale_upsert, join_cols=["id"]) + else: + stale_writer.upsert(stale_upsert, join_cols=["id"]) final_table = tower.tables("concurrent_same_row_test", catalog=sql_catalog).load() df = final_table.read() @@ -748,7 +727,7 @@ def tracked_refresh(): assert len(df) == 1 final_counter = df.select("counter").item() - assert final_counter in [1, 2, 3, 4, 5] + assert final_counter == (1 if _PYICEBERG_HAS_WRITE_CONFLICT_VALIDATION else 2) def test_insert_concurrent_writes_with_retry(sql_catalog): From d5c14aea1c6863db934e9a17349e85f8b52d2eeb Mon Sep 17 00:00:00 2001 From: Brad Heller Date: Fri, 18 Sep 2026 17:21:08 -0700 Subject: [PATCH 4/4] fix: pull MinIO test image from Quay --- crates/tower-duckdb/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/tower-duckdb/src/lib.rs b/crates/tower-duckdb/src/lib.rs index 57ab1b11..64754fc7 100644 --- a/crates/tower-duckdb/src/lib.rs +++ b/crates/tower-duckdb/src/lib.rs @@ -1241,7 +1241,7 @@ mod tests { } let bucket = "warehouse"; - let image = GenericImage::new("minio/minio", MINIO_IMAGE_TAG) + let image = GenericImage::new("quay.io/minio/minio", MINIO_IMAGE_TAG) .with_wait_for(WaitFor::seconds(1)) .with_exposed_port(9000.tcp()) .with_entrypoint("sh")