fix(sql): propagate catalog commit errors

crates/catalog/sql/src/catalog.rs

@@ -564,9 +564,12 @@ impl SqlCatalog {
             Some(t) => sqlx_query.execute(&mut **t).await.map_err(from_sqlx_error),
             None => {
                 let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
-                let result = sqlx_query.execute(&mut *tx).await.map_err(from_sqlx_error);
-                let _ = tx.commit().await.map_err(from_sqlx_error);
-                result
+                let result = sqlx_query
+                    .execute(&mut *tx)
+                    .await
+                    .map_err(from_sqlx_error)?;
+                tx.commit().await.map_err(from_sqlx_error)?;
+                Ok(result)
             }
         }
     }
@@ -1242,10 +1245,10 @@ mod tests {
 
     use crate::catalog::{
         CATALOG_FIELD_RECORD_TYPE, CATALOG_TABLE_NAME, NAMESPACE_LOCATION_PROPERTY_KEY,
-        SQL_CATALOG_PROP_BIND_STYLE, SQL_CATALOG_PROP_BIND_STYLE_LEGACY,
+        NAMESPACE_TABLE_NAME, SQL_CATALOG_PROP_BIND_STYLE, SQL_CATALOG_PROP_BIND_STYLE_LEGACY,
         SQL_CATALOG_PROP_SCHEMA_VERSION, SQL_CATALOG_PROP_URI, SQL_CATALOG_PROP_WAREHOUSE,
     };
-    use crate::{SchemaVersion, SqlBindStyle, SqlCatalogBuilder};
+    use crate::{SchemaVersion, SqlBindStyle, SqlCatalog, SqlCatalogBuilder};
 
     const UUID_REGEX_STR: &str = "[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}";
 
@@ -1386,6 +1389,108 @@ mod tests {
         new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
     }
 
+    async fn new_commit_error_catalog() -> SqlCatalog {
+        let sql_lite_uri = format!("sqlite:{}", temp_path());
+        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
+        let catalog = SqlCatalogBuilder::default()
+            .with_storage_factory(Arc::new(LocalFsStorageFactory))
+            .prop("pool.max-connections", "1")
+            .load(
+                "iceberg",
+                HashMap::from_iter([
+                    (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri),
+                    (SQL_CATALOG_PROP_WAREHOUSE.to_string(), temp_path()),
+                ]),
+            )
+            .await
+            .unwrap();
+
+        catalog
+            .connection
+            .execute("PRAGMA foreign_keys = ON")
+            .await
+            .unwrap();
+        // This deferred constraint lets an INSERT succeed while COMMIT fails.
+        catalog
+            .connection
+            .execute("CREATE TABLE parent(id INTEGER PRIMARY KEY)")
+            .await
+            .unwrap();
+        catalog
+            .connection
+            .execute(
+                "CREATE TABLE child(parent_id INTEGER REFERENCES parent(id) \
+                 DEFERRABLE INITIALLY DEFERRED)",
+            )
+            .await
+            .unwrap();
+
+        catalog
+    }
+
+    #[tokio::test]
+    async fn test_execute_returns_commit_error() {
+        let catalog = new_commit_error_catalog().await;
+
+        // Make the public namespace operation insert a child row whose deferred
+        // foreign-key constraint succeeds during execution but fails at commit.
+        let trigger = format!(
+            "CREATE TRIGGER fail_namespace_commit
+             AFTER INSERT ON {NAMESPACE_TABLE_NAME}
+             BEGIN INSERT INTO child VALUES (1); END"
+        );
+        catalog.connection.execute(trigger.as_str()).await.unwrap();
+
+        let failed_namespace = NamespaceIdent::new("failed".into());
+        let error = catalog
+            .create_namespace(&failed_namespace, HashMap::new())
+            .await
+            .unwrap_err();
+        assert_eq!(error.kind(), ErrorKind::Unexpected);
+        assert!(!catalog.namespace_exists(&failed_namespace).await.unwrap());
+
+        // A valid relationship confirms that successful transactions still commit.
+        catalog
+            .connection
+            .execute("INSERT INTO parent VALUES (1)")
+            .await
+            .unwrap();
+        let committed_namespace = NamespaceIdent::new("committed".into());
+        catalog
+            .create_namespace(&committed_namespace, HashMap::new())
+            .await
+            .unwrap();
+        assert!(
+            catalog
+                .namespace_exists(&committed_namespace)
+                .await
+                .unwrap()
+        );
+    }
+
+    #[tokio::test]
+    async fn test_execute_with_external_transaction_returns_commit_error() {
+        let catalog = new_commit_error_catalog().await;
+
+        // When the caller owns the transaction, execute returns the successful
+        // statement result and the caller receives the commit error.
+        let mut transaction = catalog.connection.begin().await.unwrap();
+        catalog
+            .execute(
+                "INSERT INTO child VALUES (1)",
+                vec![],
+                Some(&mut transaction),
+            )
+            .await
+            .unwrap();
+        assert!(transaction.commit().await.is_err());
+        let child_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM child")
+            .fetch_one(&catalog.connection)
+            .await
+            .unwrap();
+        assert_eq!(child_count, 0);
+    }
+
     // Regression test: storage-backend props set on the catalog must reach
     // the FileIO; otherwise authenticated backends fail with 401s on writes.
     #[tokio::test]