diff --git a/crates/catalog/sql/src/catalog.rs b/crates/catalog/sql/src/catalog.rs index 76cfc11bc8..73671f5c4b 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) } } } @@ -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,85 @@ 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() + ); + } + // Regression test: storage-backend props set on the catalog must reach // the FileIO; otherwise authenticated backends fail with 401s on writes. #[tokio::test]