From d8e7ed0642c6a2a29da2f98f4d332ea6b184e24e Mon Sep 17 00:00:00 2001 From: Matt Faltyn Date: Mon, 24 Aug 2026 22:16:12 +0200 Subject: [PATCH 1/4] fix(sql): propagate catalog commit errors --- crates/catalog/sql/src/catalog.rs | 73 +++++++++++++++++++++++++++++-- 1 file changed, 70 insertions(+), 3 deletions(-) diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs index 76cfc11bc8..2ab84de22d 100644 --- a/crates/catalog/sql/src/catalog.rs +++ b/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) } } } @@ -1386,6 +1389,70 @@ mod tests { new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await; } + #[tokio::test] + async fn test_execute_returns_commit_error() { + 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(); + 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(); + + assert!( + catalog + .execute("INSERT INTO child VALUES (1)", vec![], None) + .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); + + catalog + .connection + .execute("INSERT INTO parent VALUES (1)") + .await + .unwrap(); + catalog + .execute("INSERT INTO child VALUES (1)", vec![], None) + .await + .unwrap(); + let child_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM child") + .fetch_one(&catalog.connection) + .await + .unwrap(); + assert_eq!(child_count, 1); + } + // Regression test: storage-backend props set on the catalog must reach // the FileIO; otherwise authenticated backends fail with 401s on writes. #[tokio::test] From a72b73a939983bfa802bb089b8130b9efd570ab9 Mon Sep 17 00:00:00 2001 From: Matt Faltyn Date: Wed, 26 Aug 2026 08:32:59 +0200 Subject: [PATCH 2/4] test(sql): clarify transaction commit coverage --- crates/catalog/sql/src/catalog.rs | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs index 2ab84de22d..c706c45971 100644 --- a/crates/catalog/sql/src/catalog.rs +++ b/crates/catalog/sql/src/catalog.rs @@ -1406,6 +1406,8 @@ mod tests { .await .unwrap(); + // A deferred foreign-key constraint makes the INSERT succeed while COMMIT + // reliably fails, allowing both transaction ownership paths to be tested. catalog .connection .execute("PRAGMA foreign_keys = ON") @@ -1425,6 +1427,7 @@ mod tests { .await .unwrap(); + // When execute owns the transaction, it must return the commit error. assert!( catalog .execute("INSERT INTO child VALUES (1)", vec![], None) @@ -1437,6 +1440,25 @@ mod tests { .unwrap(); assert_eq!(child_count, 0); + // 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); + + // A valid relationship confirms that successful owned transactions commit. catalog .connection .execute("INSERT INTO parent VALUES (1)") From c324990b91d998b142c1b1c88c898b25b0b8dac7 Mon Sep 17 00:00:00 2001 From: Matt Faltyn Date: Wed, 26 Aug 2026 10:03:14 +0200 Subject: [PATCH 3/4] test(sql): exercise commit errors through catalog API --- crates/catalog/sql/src/catalog.rs | 76 +++++++++++++++++++------------ 1 file changed, 46 insertions(+), 30 deletions(-) diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs index c706c45971..67751aaa79 100644 --- a/crates/catalog/sql/src/catalog.rs +++ b/crates/catalog/sql/src/catalog.rs @@ -1245,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}"; @@ -1389,8 +1389,7 @@ mod tests { new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await; } - #[tokio::test] - async fn test_execute_returns_commit_error() { + 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() @@ -1406,13 +1405,12 @@ mod tests { .await .unwrap(); - // A deferred foreign-key constraint makes the INSERT succeed while COMMIT - // reliably fails, allowing both transaction ownership paths to be tested. 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)") @@ -1427,18 +1425,52 @@ mod tests { .await .unwrap(); - // When execute owns the transaction, it must return the commit error. + 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 - .execute("INSERT INTO child VALUES (1)", vec![], None) + .namespace_exists(&committed_namespace) .await - .is_err() + .unwrap() ); - let child_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM child") - .fetch_one(&catalog.connection) - .await - .unwrap(); - assert_eq!(child_count, 0); + } + + #[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. @@ -1457,22 +1489,6 @@ mod tests { .await .unwrap(); assert_eq!(child_count, 0); - - // A valid relationship confirms that successful owned transactions commit. - catalog - .connection - .execute("INSERT INTO parent VALUES (1)") - .await - .unwrap(); - catalog - .execute("INSERT INTO child VALUES (1)", vec![], None) - .await - .unwrap(); - let child_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM child") - .fetch_one(&catalog.connection) - .await - .unwrap(); - assert_eq!(child_count, 1); } // Regression test: storage-backend props set on the catalog must reach From 64e5aa50ce59d4f3935203b4263ce5bab7c18547 Mon Sep 17 00:00:00 2001 From: Matt Faltyn Date: Wed, 26 Aug 2026 10:19:33 +0200 Subject: [PATCH 4/4] test(sql): keep commit regression catalog-focused --- crates/catalog/sql/src/catalog.rs | 23 ----------------------- 1 file changed, 23 deletions(-) diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs index 67751aaa79..73671f5c4b 100644 --- a/crates/catalog/sql/src/catalog.rs +++ b/crates/catalog/sql/src/catalog.rs @@ -1468,29 +1468,6 @@ mod tests { ); } - #[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]