laskoviymishka commented on code in PR #3071:
URL: https://github.com/apache/iceberg-rust/pull/3071#discussion_r4087496136
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -382,6 +379,35 @@ impl SchemaVersion {
SchemaVersion::V0 => None,
}
}
+
+ /// Migrates the catalog table to this schema version.
+ ///
+ /// Another catalog instance may complete the same migration after version
detection but
+ /// before this statement runs. In that case, accept the statement error
once re-detection
+ /// confirms that the requested version is already installed.
+ async fn migrate(self, pool: &AnyPool) -> Result<()> {
+ let Some(migration_sql) = self.migration_sql() else {
+ return Ok(());
+ };
+
+ match sqlx::query(&migration_sql).execute(pool).await {
+ Ok(_) => Ok(()),
+ Err(error) => {
+ // A competing DDL statement may still be committing when this
statement fails.
+ // Re-acquiring a connection for each probe lets schema-lock
and duplicate-column
+ // races converge on the installed version instead of failing
initialization.
+ for _ in 0..3 {
+ if matches!(Self::detect(pool).await, Ok(detected) if
detected == self) {
+ tracing::debug!(
Review Comment:
This fires once a competing instance has already applied the migration, but
at debug it's below the default prod level — on a rolling deploy most instances
hit this path and log nothing. I'd bump it to `info!` and include the original
error string, so the "race resolved" case is actually visible and
distinguishable from a real failure.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -338,22 +338,19 @@ pub enum SchemaVersion {
impl SchemaVersion {
/// Detect the schema version of an existing catalog table by
introspecting its columns.
async fn detect(pool: &AnyPool) -> Result<Self> {
- let catalog_table_description = pool
+ let mut connection = pool.acquire().await.map_err(from_sqlx_error)?;
+ let v1_probe =
+ format!("SELECT {CATALOG_FIELD_RECORD_TYPE} FROM
{CATALOG_TABLE_NAME} WHERE 1 = 0");
+ if connection.describe(&v1_probe).await.is_ok() {
Review Comment:
`is_ok()` here treats every probe failure as "column absent", so a transient
connection drop or a permissions/proxy error on the V1 probe reads as V0 even
when the table is actually V1. I'd match on the error kind and only fall
through on a genuine "no such column" — anything else should propagate.
(PyIceberg propagates this probe error rather than collapsing it to V0.)
The failure that worries me: the V1 probe errors transiently, the `SELECT *`
probe on the same connection succeeds, and we return V0 for a V1 catalog. That
flips `record_type_filter()` to empty and starts leaking view rows through
`list_tables`/`load_table`, and refuses `create_table` against a schema that
actually supports it. Even in the case where `migrate` later re-detects V1 and
recovers, the whole chain runs on a false premise with nothing logged — so at
minimum log the discarded error before falling through, so this is diagnosable.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -2699,6 +2720,44 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_v0_schema_migration_is_concurrent_safe() {
Review Comment:
Good that this pins the happy concurrent path. The gap is that SQLite
serializes both writers at the file lock, so the loser always re-detects a
committed V1 — the test never exercises the retry loop's actual window
(competing DDL still committing) or the genuine-error
`Err(from_sqlx_error(error))` arm, which is currently uncovered. I'd add a unit
test that injects a non-concurrent DDL failure and asserts it propagates with
the original cause, so a regression in `detect` can't turn a real error into a
silent success.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -382,6 +379,35 @@ impl SchemaVersion {
SchemaVersion::V0 => None,
}
}
+
+ /// Migrates the catalog table to this schema version.
+ ///
+ /// Another catalog instance may complete the same migration after version
detection but
+ /// before this statement runs. In that case, accept the statement error
once re-detection
+ /// confirms that the requested version is already installed.
+ async fn migrate(self, pool: &AnyPool) -> Result<()> {
+ let Some(migration_sql) = self.migration_sql() else {
+ return Ok(());
+ };
+
+ match sqlx::query(&migration_sql).execute(pool).await {
+ Ok(_) => Ok(()),
+ Err(error) => {
+ // A competing DDL statement may still be committing when this
statement fails.
+ // Re-acquiring a connection for each probe lets schema-lock
and duplicate-column
+ // races converge on the installed version instead of failing
initialization.
+ for _ in 0..3 {
Review Comment:
The three probes fire back-to-back with no delay, so on a networked backend
they can all complete inside the window where the winning instance's DDL is
committed but not yet visible to a fresh connection — the loop exhausts, we
return the original error, and init fails on exactly the concurrent-migration
case this is meant to handle. I'd put a short backoff between iterations
(something like 50/100/200ms) so a mid-commit DDL has a chance to land. The
SQLite test doesn't surface this because the file lock serializes the two
writers, so the loser always re-detects a committed V1.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
