Release v0.3.58

Cargo.toml

@@ -4,7 +4,7 @@ resolver = "2"
 
 [workspace.package]
 edition = "2021"
-version = "0.3.57"
+version = "0.3.58"
 description = "Tower is the best way to host Python data apps in production"
 rust-version = "1.81"
 authors = ["Brad Heller <brad@tower.dev>"]

crates/tower-cmd/src/session.rs

@@ -132,8 +132,12 @@ fn finalize_session(
 ) {
     let mut session = Session::from_api_session(&session_response.session);
 
-    // we have to copy in the tower URL so that we save it for later on!
-    session.tower_url = config.tower_url.clone();
+    let mut url = config.tower_url.clone();
+    let local = matches!(url.host_str(), Some("localhost" | "127.0.0.1" | "::1"));
+    if url.scheme() == "http" && !local {
+        let _ = url.set_scheme("https");
+    }
+    session.tower_url = url;
 
     if let Err(err) = session.save() {
         spinner.failure();

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "tower"
-version = "0.3.57"
+version = "0.3.58"
 description = "Tower CLI and runtime environment for Tower."
 authors = [{ name = "Tower Computing Inc.", email = "brad@tower.dev" }]
 readme = "README.md"
@@ -48,7 +48,7 @@ tower = "tower.cli:main"
 
 [project.optional-dependencies]
 ai = ["huggingface-hub>=0.34.3", "ollama>=0.5.3"]
-iceberg = ["polars==1.27.1", "pyarrow==19.0.1", "pyiceberg==0.9.1"]
+iceberg = ["polars>=1.39.3", "pyarrow>=23.0.1", "pyiceberg>=0.11.1"]
 dbt = ["dbt-core>=1.9,<1.10"] # Can not use dbt 1.10+ due to dependency conflicts
 all = ["tower[ai,iceberg,dbt]"]
 
@@ -102,5 +102,5 @@ dev = [
   "mypy==1.13.0",
   "openapi-python-client==0.28.1",
   "pre-commit==4.0.1",
-  "pyiceberg[sql-sqlite]==0.9.1",
+  "pyiceberg[sql-sqlite]>=0.11.1",
 ]

src/tower/_tables.py

@@ -93,7 +93,9 @@ def read(self) -> pl.DataFrame:
         """
         # We call `collect` here to force the execution of the query and get
         # the result as a DataFrame.
-        return pl.scan_iceberg(self._table).collect()
+        # Note: reader_override="pyiceberg" ensures AWS S3 vended credentials are passed through
+        # properly. The native rust reader is faster but doesn't pass credentials from pyiceberg.
+        return pl.scan_iceberg(self._table, reader_override="pyiceberg").collect()
 
     def to_polars(self) -> pl.LazyFrame:
         """
@@ -121,7 +123,9 @@ def to_polars(self) -> pl.LazyFrame:
             >>> # Execute the plan
             >>> final_df = result.collect()
         """
-        return pl.scan_iceberg(self._table)
+        # Note: reader_override="pyiceberg" ensures AWS S3 vended credentials are passed through
+        # properly. The native rust reader is faster but doesn't pass credentials from pyiceberg.
+        return pl.scan_iceberg(self._table, reader_override="pyiceberg")
 
     def rows_affected(self) -> RowsAffectedInformation:
         """
@@ -149,9 +153,21 @@ def rows_affected(self) -> RowsAffectedInformation:
         """
         return self._stats
 
-    def insert(self, data: pa.Table) -> TTable:
+    @staticmethod
+    def _validate_retry_args(max_retries: int, retry_delay_seconds: float) -> None:
+        if max_retries < 0:
+            raise ValueError("max_retries must be >= 0")
+        if retry_delay_seconds < 0:
+            raise ValueError("retry_delay_seconds must be >= 0")
+
+    def insert(
+        self,
+        data: pa.Table,
+        max_retries: int = 5,
+        retry_delay_seconds: float = 0.5,
+    ) -> TTable:
         """
-        Inserts new rows into the Iceberg table.
+        Inserts new rows into the Iceberg table. In case of commit conflicts, reloads the metadata and retries.
 
         This method appends the provided data to the table. The data must be provided as a
         PyArrow table with a schema that matches the table's schema. The operation is
@@ -160,10 +176,17 @@ def insert(self, data: pa.Table) -> TTable:
         Args:
             data (pa.Table): The data to insert into the table. The schema of this table
                 must match the schema of the target table.
+            max_retries (int): Maximum number of retry attempts on commit conflicts.
+                Defaults to 5.
+            retry_delay_seconds (float): Wait time in seconds between retries.
+                Defaults to 0.5 seconds.
 
         Returns:
             TTable: The table instance with the newly inserted rows, allowing for method chaining.
 
+        Raises:
+            CommitFailedException: If all retry attempts are exhausted.
+
         Example:
             >>> table = tables("my_table").load()
             >>> # Create a PyArrow table with new data
@@ -178,9 +201,25 @@ def insert(self, data: pa.Table) -> TTable:
             >>> stats = table.rows_affected()
             >>> print(f"Inserted {stats.inserts} rows")
         """
-        self._table.append(data)
-        self._stats.inserts += data.num_rows
-        return self
+        self._validate_retry_args(max_retries, retry_delay_seconds)
+
+        last_exception = None
+
+        for attempt in range(max_retries + 1):
+            try:
+                if attempt > 0:
+                    self._table.refresh()
+
+                self._table.append(data)
+                self._stats.inserts += data.num_rows
+                return self
+
+            except CommitFailedException as e:
+                last_exception = e
+                if attempt < max_retries:
+                    time.sleep(retry_delay_seconds)
+
+        raise last_exception
 
     def upsert(
         self,
@@ -235,6 +274,8 @@ def upsert(
             >>> print(f"Updated {stats.updates} rows")
             >>> print(f"Inserted {stats.inserts} rows")
         """
+        self._validate_retry_args(max_retries, retry_delay_seconds)
+
         last_exception = None
 
         for attempt in range(max_retries + 1):
@@ -264,9 +305,15 @@ def upsert(
 
         raise last_exception
 
-    def delete(self, filters: Union[str, List[pc.Expression]]) -> TTable:
+    def delete(
+        self,
+        filters: Union[str, List[pc.Expression]],
+        max_retries: int = 5,
+        retry_delay_seconds: float = 0.5,
+    ) -> TTable:
         """
         Deletes rows from the Iceberg table that match the specified filter conditions.
+        In case of commit conflicts, reloads the metadata and retries.
 
         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
@@ -278,10 +325,17 @@ def delete(self, filters: Union[str, List[pc.Expression]]) -> TTable:
                 - A single PyArrow compute expression
                 - A list of PyArrow compute expressions (combined with AND)
                 - A string expression
+            max_retries (int): Maximum number of retry attempts on commit conflicts.
+                Defaults to 5.
+            retry_delay_seconds (float): Wait time in seconds between retries.
+                Defaults to 0.5 seconds.
 
         Returns:
             TTable: The table instance with the deleted rows, allowing for method chaining.
 
+        Raises:
+            CommitFailedException: If all retry attempts are exhausted.
+
         Note:
             - The operation is always case-sensitive
             - The number of deleted rows cannot be tracked in the table statistics
@@ -299,22 +353,37 @@ def delete(self, filters: Union[str, List[pc.Expression]]) -> TTable:
             >>> # Delete rows using a string expression
             >>> table.delete("age > 30 AND department = 'IT'")
         """
+        self._validate_retry_args(max_retries, retry_delay_seconds)
 
         if isinstance(filters, list):
             # We need to convert the pc.Expression into PyIceberg
             next_filters = convert_pyarrow_expressions(filters)
             filters = next_filters
 
-        self._table.delete(
-            delete_filter=filters,
-            # We want this to always be the case. Not sure why you wouldn't?
-            case_sensitive=True,
-        )
+        last_exception = None
 
-        # NOTE: There is, unfortunately, no way to get the number of rows
-        # deleted besides comparing the two snapshots that were created.
+        for attempt in range(max_retries + 1):
+            try:
+                if attempt > 0:
+                    self._table.refresh()
+
+                self._table.delete(
+                    delete_filter=filters,
+                    # We want this to always be the case. Not sure why you wouldn't?
+                    case_sensitive=True,
+                )
+
+                # NOTE: There is, unfortunately, no way to get the number of rows
+                # deleted besides comparing the two snapshots that were created.
+
+                return self
 
-        return self
+            except CommitFailedException as e:
+                last_exception = e
+                if attempt < max_retries:
+                    time.sleep(retry_delay_seconds)
+
+        raise last_exception
 
     def schema(self) -> pa.Schema:
         """