diff --git a/crates/dbx-core/src/db/questdb.rs b/crates/dbx-core/src/db/questdb.rs index 672290b2b..420084570 100644 --- a/crates/dbx-core/src/db/questdb.rs +++ b/crates/dbx-core/src/db/questdb.rs @@ -23,12 +23,23 @@ pub async fn list_objects(pool: &Pool, schema: &str) -> Result, .collect()) } -/// try query `table`, `view` and `materialized view` using statement supported by the newer version. -/// if there is an error, rollback to the previous version of the statement -pub async fn list_tables(pool: &Pool, _schema: &str) -> Result, String> { - match list_tables_new_version(pool, _schema).await { - Ok(ddl) => Ok(ddl), - Err(_) => list_tables_older_version(pool, _schema).await, +/// Query the richest table metadata supported by the connected QuestDB version. +pub async fn list_tables(pool: &Pool, schema: &str) -> Result, String> { + match list_tables_new_version(pool, schema).await { + Ok(tables) => Ok(tables), + Err(new_version_error) => match list_tables_mat_view_version(pool, schema).await { + Ok(tables) => Ok(tables), + Err(mat_view_version_error) => { + // QuestDB 8.2.x predates both `matView` and `table_type`, but + // still exposes `table_name`; retain basic table browsing. + list_tables_basic(pool, schema).await.map_err(|basic_error| { + format!( + "QuestDB table metadata queries failed: table_type={new_version_error}; \ + matView={mat_view_version_error}; table_name={basic_error}" + ) + }) + } + }, } } @@ -59,13 +70,13 @@ async fn list_tables_new_version(pool: &Pool, _schema: &str) -> Result &'static str { - "SELECT table_name, table_type FROM tables" + "SELECT table_name, table_type FROM tables()" } -async fn list_tables_older_version(pool: &Pool, _schema: &str) -> Result, String> { +async fn list_tables_mat_view_version(pool: &Pool, _schema: &str) -> Result, String> { let client = pool.get().await.map_err(|e| e.to_string())?; - let stmt = client.prepare_cached(questdb_tables_sql_older_version()).await.map_err(|e| e.to_string())?; + let stmt = client.prepare_cached(questdb_tables_sql_mat_view_version()).await.map_err(|e| e.to_string())?; let rows = client.query(&stmt, &[]).await.map_err(|e| e.to_string())?; Ok(rows @@ -84,8 +95,29 @@ async fn list_tables_older_version(pool: &Pool, _schema: &str) -> Result &'static str { - "SELECT table_name, matView FROM tables" +fn questdb_tables_sql_mat_view_version() -> &'static str { + "SELECT table_name, matView FROM tables()" +} + +async fn list_tables_basic(pool: &Pool, _schema: &str) -> Result, String> { + let client = pool.get().await.map_err(|e| e.to_string())?; + let stmt = client.prepare_cached(questdb_tables_sql_basic()).await.map_err(|e| e.to_string())?; + let rows = client.query(&stmt, &[]).await.map_err(|e| e.to_string())?; + + Ok(rows + .iter() + .map(|row| TableInfo { + name: row.get::<_, String>(0), + table_type: "TABLE".to_string(), + comment: None, + parent_schema: None, + parent_name: None, + }) + .collect()) +} + +fn questdb_tables_sql_basic() -> &'static str { + "SELECT table_name FROM tables()" } pub async fn get_columns(pool: &Pool, _schema: &str, table: &str) -> Result, String> { @@ -178,6 +210,18 @@ async fn questdb_normal_view_ddl(pool: &Pool, view: &str) -> Result Result { result .rows diff --git a/crates/dbx-core/tests/live_questdb.rs b/crates/dbx-core/tests/live_questdb.rs new file mode 100644 index 000000000..f2fc6ab7f --- /dev/null +++ b/crates/dbx-core/tests/live_questdb.rs @@ -0,0 +1,32 @@ +use std::time::Duration; + +use dbx_core::db; + +#[tokio::test] +#[ignore = "requires DBX_TEST_QUESTDB_URL pointing at a writable QuestDB database"] +async fn questdb_lists_tables_across_metadata_generations() { + let url = std::env::var("DBX_TEST_QUESTDB_URL").expect("DBX_TEST_QUESTDB_URL"); + let pool = db::postgres::connect(&url, Duration::from_secs(10)).await.expect("connect QuestDB"); + let table = format!("dbx_questdb_82_{}", uuid::Uuid::new_v4().simple()); + + db::postgres::execute_query( + &pool, + &format!("CREATE TABLE {table} (ts TIMESTAMP, value DOUBLE) TIMESTAMP(ts) PARTITION BY DAY WAL"), + ) + .await + .expect("create QuestDB fixture"); + + let exercise = async { + let tables = db::questdb::list_tables(&pool, "public").await?; + let listed = tables.iter().find(|candidate| candidate.name == table).ok_or("fixture table was not listed")?; + assert_eq!(listed.table_type, "TABLE"); + + let columns = db::questdb::get_columns(&pool, "public", &table).await?; + assert_eq!(columns.iter().map(|column| column.name.as_str()).collect::>(), vec!["ts", "value"]); + Ok::<_, String>(()) + } + .await; + + db::postgres::execute_query(&pool, &format!("DROP TABLE {table}")).await.expect("drop QuestDB fixture"); + exercise.expect("list QuestDB table metadata"); +}