Release v0.3.75

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
 

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;

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();
+    }
+}

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")

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"))]

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};
 

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,21 +371,27 @@ 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:
             data (pa.Table): The data to upsert into the table. The schema of this table
                 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

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

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)

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)

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,14 +208,27 @@ 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"),
     [
         ("timestamp_ns", pa.timestamp("ns")),
         ("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)),
     ],