laskoviymishka commented on code in PR #3071:
URL: https://github.com/apache/iceberg-rust/pull/3071#discussion_r4148873674
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -338,22 +338,35 @@ 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");
+ match connection.describe(&v1_probe).await {
+ Ok(_) => return Ok(SchemaVersion::V1),
+ Err(error) if Self::is_missing_record_type_column(&error) => {}
+ Err(error) => return Err(from_sqlx_error(error)),
+ }
+
+ // Distinguish a V0 schema from a missing or inaccessible catalog
table.
+ connection
.describe(&format!("SELECT * FROM {CATALOG_TABLE_NAME}"))
.await
.map_err(from_sqlx_error)?;
+ Ok(SchemaVersion::V0)
+ }
- let has_type_column =
catalog_table_description.columns().iter().any(|column| {
- column
- .name()
- .eq_ignore_ascii_case(CATALOG_FIELD_RECORD_TYPE)
- });
+ fn is_missing_record_type_column(error: &sqlx::Error) -> bool {
+ let sqlx::Error::Database(error) = error else {
+ return false;
+ };
- Ok(if has_type_column {
- SchemaVersion::V1
- } else {
- SchemaVersion::V0
- })
+ match error.code().as_deref() {
+ // PostgreSQL undefined_column and MySQL/MariaDB
ER_BAD_FIELD_ERROR.
+ Some("42703" | "42S22") => true,
+ // SQLite uses SQLITE_ERROR for many failures, so also check the
exact message.
+ Some("1") => error.message() == format!("no such column:
{CATALOG_FIELD_RECORD_TYPE}"),
Review Comment:
Exact-`==` on the SQLite message is fragile — if a future SQLite build
appends context (`no such column: iceberg_type (did you mean ...)`) this
returns false and `detect` propagates the error instead of falling through to
V0, silently breaking V0→V1 migration on that build. I'd switch to
`starts_with`. While we're here, declaring `CATALOG_FIELD_RECORD_TYPE` as
`const` instead of `static` lets this be `starts_with(concat!("no such column:
", CATALOG_FIELD_RECORD_TYPE))` — a zero-cost `&str` compare instead of a
`format!` heap alloc on every code-1 error.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -382,6 +395,37 @@ 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 delay_ms in [50, 100, 200] {
Review Comment:
The retry loop fires on every DDL failure, not just the concurrent-migration
race — a readonly connection or a missing ALTER privilege now eats the full
50+100+200ms across three probes that can never see V1 before the original
error propagates. `test_v0_schema_migration_preserves_ddl_error` pays exactly
that 350ms today.
I'd classify the failure before sleeping: duplicate-column (Postgres
`42701`, MySQL `42S21`, SQLite code 1 + `duplicate column name:`) is a definite
race, so probe once immediately; genuinely non-transient errors (readonly,
permission-denied) return right away; only ambiguous lock cases go through the
timed loop. That also drops the 50ms floor the SQLite concurrent-loser pays
when the column already landed.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -382,6 +395,37 @@ 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 delay_ms in [50, 100, 200] {
+ tokio::time::sleep(Duration::from_millis(delay_ms)).await;
+ if matches!(Self::detect(pool).await, Ok(detected) if
detected == self) {
Review Comment:
`matches!(Self::detect(pool).await, Ok(detected) if detected == self)` folds
a detect `Err` into the same `false` as a legit `Ok(V0)`, so if all three
probes fail on pool exhaustion or a reset connection we return only the
original DDL error with no trace of why re-detection never confirmed. I'd
expand this to a `match` and log the `Err` arm (a `tracing::warn!`/`debug!` is
plenty) so a stuck migration is diagnosable.
##########
crates/catalog/sql/Cargo.toml:
##########
@@ -35,6 +35,7 @@ iceberg = { workspace = true }
iceberg-property-macro = { workspace = true }
sqlx = { version = "0.8.1", features = ["any"], default-features = false }
strum = { workspace = true }
+tokio = { workspace = true, features = ["time"] }
tracing = { workspace = true }
Review Comment:
Moving `tokio` to main deps with just `time` is right, but this drops the
dev-dep line, and the tests lean on `macros` (`#[tokio::test]`), `sync`
(`Barrier`), and `join!` — features that now only reach the test build
transitively through sqlx's `runtime-tokio`. I'd re-add `tokio = { workspace =
true, features = ["sync", "macros"] }` under `[dev-dependencies]` so the test
surface doesn't silently ride on another crate's feature selection.
--
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]
