Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
92 changes: 87 additions & 5 deletions crates/catalog/sql/src/catalog.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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)
}
}
}
Expand Down Expand Up @@ -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}";

Expand Down Expand Up @@ -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]
Expand Down
Loading