laskoviymishka commented on code in PR #3071:
URL: https://github.com/apache/iceberg-rust/pull/3071#discussion_r4313538626
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -382,6 +393,58 @@ 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) => {
+ // Check the migration's result without interpreting
vendor-specific errors.
+ // A competing migration may still need time to commit after
the first probe.
+ let recovery = (|| async {
+ match Self::detect(pool).await {
+ Ok(detected) if detected == self => Ok(()),
+ Ok(_) => Err(()),
+ Err(probe_error) => {
+ tracing::warn!(
+ error = %probe_error,
+ migration_error = %error,
+ "failed to re-detect catalog schema after
migration to {self} failed"
+ );
+ Err(())
+ }
+ }
+ })
+ .retry(
+ ExponentialBuilder::new()
+ .with_min_delay(Duration::from_millis(100))
+ .with_max_delay(Duration::from_secs(60))
Review Comment:
Two of these settings have no effect. With `max_times(4)` the waits are
100/200/400/800ms, so the 60s cap never applies, and `2.0` is already backon's
default factor. I'd drop both and pull the min delay and retry count into named
consts, so the hard-coded `6` in the two readonly-failure tests can be derived
from them.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -338,22 +339,32 @@ 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
- .describe(&format!("SELECT * FROM {CATALOG_TABLE_NAME}"))
- .await
- .map_err(from_sqlx_error)?;
-
- let has_type_column =
catalog_table_description.columns().iter().any(|column| {
+ Self::probe(pool).await.map_err(from_sqlx_error)
+ }
+
+ async fn probe(pool: &AnyPool) -> sqlx::Result<Self> {
+ let mut transaction = pool.begin().await?;
+ // Run both reads as one unprepared batch in a transaction. The empty
read refreshes
+ // the schema before the second statement prepares its column
metadata. The second
+ // statement returns one NULL row even for an empty catalog.
+ let probe = format!(
+ "SELECT * FROM {CATALOG_TABLE_NAME} WHERE 1 = 0; \
+ SELECT catalog.* FROM (SELECT 1) AS probe \
+ LEFT JOIN {CATALOG_TABLE_NAME} AS catalog ON 1 = 0"
+ );
+ let catalog_row = transaction
+ .fetch_all(sqlx::raw_sql(&probe))
+ .await?
+ .pop()
+ .ok_or(sqlx::Error::RowNotFound)?;
Review Comment:
This runs on every catalog start, so if it fails, `SqlCatalog::new` fails
for everyone, not just during a migration race. The description shows it
passing on SQLite and on Postgres 18.6 locally. MySQL isn't covered, and
multi-statement text queries are the part most likely to behave differently
there. If the batch comes back without the trailing row, we return
`RowNotFound` and the catalog won't start.
I'd either add a MySQL run of `test_schema_detection_after_migration`, or
fall back to `describe` when `.pop()` finds no row, so an unexpected result
degrades instead of failing init.
##########
crates/catalog/sql/src/catalog.rs:
##########
@@ -2608,31 +2669,98 @@ mod tests {
async fn test_detect_schema_version() {
install_default_drivers();
- let detected_schema = {
- let (uri, _temp_dir) = create_v0_sqlite_db().await;
+ for expected_schema in [SchemaVersion::V0, SchemaVersion::V1] {
+ let (uri, _temp_dir) = match expected_schema {
+ SchemaVersion::V0 => create_v0_sqlite_db().await,
+ SchemaVersion::V1 => create_v1_sqlite_db().await,
+ };
let pool = sqlx::AnyPool::connect(&uri).await.unwrap();
- let detected_schema = SchemaVersion::detect(&pool).await.unwrap();
+ assert_eq!(
+ SchemaVersion::detect(&pool).await.unwrap(),
+ expected_schema,
+ "a populated catalog should be detected as {expected_schema}",
+ );
+ sqlx::query("DELETE FROM iceberg_tables")
+ .execute(&pool)
+ .await
+ .unwrap();
+ assert_eq!(
+ SchemaVersion::detect(&pool).await.unwrap(),
+ expected_schema,
+ "an empty catalog should be detected as {expected_schema}",
+ );
pool.close().await;
- detected_schema
- };
+ }
+ }
+
+ #[tokio::test]
+ async fn test_schema_detection_after_concurrent_migration() {
+ install_default_drivers();
+
+ let (uri, _temp_dir) = create_v0_sqlite_db().await;
+ let pool = AnyPoolOptions::new()
+ .min_connections(1)
+ .max_connections(1)
+ .connect(&uri)
+ .await
+ .unwrap();
assert_eq!(
- detected_schema,
- SchemaVersion::V0,
- "a catalog table without an iceberg_type column should be V0",
+ SchemaVersion::detect(&pool).await.unwrap(),
+ SchemaVersion::V0
);
- let detected_schema = {
- let (uri, _temp_dir) = create_v1_sqlite_db().await;
- let pool = sqlx::AnyPool::connect(&uri).await.unwrap();
- let detected_schema = SchemaVersion::detect(&pool).await.unwrap();
- pool.close().await;
- detected_schema
- };
+ let migrating_pool = sqlx::AnyPool::connect(&uri).await.unwrap();
+ SchemaVersion::V1.migrate(&migrating_pool).await.unwrap();
+ // Reuse the V0 connection without executing any DDL on it first.
assert_eq!(
- detected_schema,
- SchemaVersion::V1,
- "a catalog table with an iceberg_type column should be V1",
+ SchemaVersion::detect(&pool).await.unwrap(),
+ SchemaVersion::V1
);
+
+ pool.close().await;
+ migrating_pool.close().await;
+ }
+
+ #[tokio::test]
+ async fn test_schema_detection_after_migration() {
+ install_default_drivers();
+
+ let uri =
+ std::env::var("ICEBERG_SQL_TEST_URI").unwrap_or_else(|_|
"sqlite::memory:".to_string());
Review Comment:
With `ICEBERG_SQL_TEST_URI` unset this runs against `sqlite::memory:`, which
is what CI does, so the Postgres run from the description isn't repeated
anywhere. I'd add a CI job that sets the URI against a Postgres service
container (and MySQL, per the `probe` comment). Without that, the new
`postgres` dev-feature only adds compile time.
--
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]
