Skip to content

fix(sql): propagate catalog commit errors #3062

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
blackmwk merged 5 commits into from
Aug 28, 2026
Merged

fix(sql): propagate catalog commit errors #3062

Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
d8e7ed0
fix(sql): propagate catalog commit errors
mattfaltyn Aug 24, 2026
a72b73a
test(sql): clarify transaction commit coverage
mattfaltyn Aug 26, 2026
e47fb82
Merge branch 'main' into fix/sql-catalog-commit-errors
mattfaltyn Aug 26, 2026
c324990
test(sql): exercise commit errors through catalog API
mattfaltyn Aug 26, 2026
64e5aa5
test(sql): keep commit regression catalog-focused
mattfaltyn Aug 26, 2026
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
73 changes: 70 additions & 3 deletions crates/catalog/sql/src/catalog.rs
View file
Open in desktop
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 @@ -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);
}

@dannycjones dannycjones Aug 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think this test needs a narrative comment to explain why we're writing some complex SQL unrelated to Iceberg. It's a bit weird, although I don't know a better way to exercise the execute error handling.

Something to this effect would help:

// Create a table that will reliably fail at database transaction commit.
// Table has a foreign key relationship, where child column reference parent ID field.
// This allows to exercise error handling where part of the table operation succeeds, but the transaction itself fails.

I think if we structure it a bit, it'll come clear what exactly we're testing.

@dannycjones dannycjones Aug 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Would it be possible to extend this test to also cover the scenario where the transaction is passed in outside of execute?

@mattfaltyn mattfaltyn Aug 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Great suggestions—thank you! I added narrative comments explaining the deferred foreign-key setup and extended the regression test to cover a caller-owned transaction, including its commit-time failure.

@blackmwk blackmwk Aug 26, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Also this ut tests using private api. I hope to see a ut where it uses public api only and get the propogated error (You could still use internal connection to construct test case)

@mattfaltyn mattfaltyn Aug 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Good point—thanks! The regression now sets up the deferred failure internally, invokes public create_namespace, asserts the propagated error, and verifies rollback through public namespace_exists.


// 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