fix(schema): handle view DDL object types

This commit is contained in:
t8y2 committed 2026-06-17 19:29:24 +08:00
1 parent 7643148708
commit 22b8ba17bb
14 files changed
+175 -48

No files matched your search

+1
View File
@@ -370,6 +370,7 @@ async fn build_schema_prompt(agent_ctx: &AgentLoopContext, system_prompt: &str)
None,
Some(50), // smaller limit for prompt injection
None,
None,
)
.await;
+1
View File
@@ -241,6 +241,7 @@ async fn execute_list_tables(
None,
Some(LIST_TABLES_LIMIT + 1),
None,
None,
)
.await
.map_err(|e| format!("Failed to list tables: {e}"))?;
+3 -2
View File
@@ -2715,8 +2715,9 @@ mod tests {
let schemas = schema::list_schemas_core(&state, "kwdb-live", &database).await.unwrap();
assert!(schemas.iter().any(|schema| schema == test_schema));
let tables =
schema::list_tables_core(&state, "kwdb-live", &database, test_schema, None, None, None).await.unwrap();
let tables = schema::list_tables_core(&state, "kwdb-live", &database, test_schema, None, None, None, None)
.await
.unwrap();
assert!(tables.iter().any(|table| table.name == "devices" && table.table_type == "BASE TABLE"));
let columns = schema::get_columns_core(&state, "kwdb-live", &database, test_schema, "devices").await.unwrap();
let id_column = columns.iter().find(|column| column.name == "id").expect("id column should be listed");
+2
View File
@@ -383,6 +383,7 @@ pub async fn export_database_sql_core(
None,
None,
None,
None,
)
.await?;
let all_tables = filter_selected_table_infos(all_tables, &request.selected_tables);
@@ -476,6 +477,7 @@ pub async fn export_database_sql_core(
&request.database,
&request.schema,
table_name,
None,
)
.await
{
+131 -28
View File
@@ -423,9 +423,10 @@ pub async fn list_tables_core(
filter: Option<&str>,
limit: Option<usize>,
offset: Option<usize>,
object_types: Option<&[String]>,
) -> Result<Vec<db::TableInfo>, String> {
retry_metadata_connection(state, connection_id, Some(database), || {
list_tables_once(state, connection_id, database, schema, filter, limit, offset)
list_tables_once(state, connection_id, database, schema, filter, limit, offset, object_types)
})
.await
}
@@ -438,6 +439,7 @@ async fn list_tables_once(
filter: Option<&str>,
limit: Option<usize>,
offset: Option<usize>,
object_types: Option<&[String]>,
) -> Result<Vec<db::TableInfo>, String> {
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
#[cfg(feature = "duckdb-bundled")]
@@ -456,7 +458,7 @@ async fn list_tables_once(
})
.await
.map_err(|e| e.to_string())?
.map(|tables| filter_table_infos(tables, filter, limit, offset));
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types));
}
if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) {
let config = config.clone();
@@ -468,26 +470,35 @@ async fn list_tables_once(
serde_json::json!({ "connection": config.as_ref(), "database": database, "schema": schema }),
)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset));
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types));
}
#[cfg(feature = "duckdb-bundled")]
if let Some(con) = extract_pool!(&connections, &pool_key, DuckDb) {
drop(connections);
let con = con.lock().map_err(|e| e.to_string())?;
return duckdb_query_tables_in_database_with_attached(&con, database, schema, &duckdb_attached_names)
.map(|tables| filter_table_infos(tables, filter, limit, offset));
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types));
}
if let Some(client) = extract_pool!(&connections, &pool_key, ClickHouse) {
drop(connections);
return db::clickhouse_driver::list_tables(&client, clickhouse_metadata_database(database, schema))
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset));
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types));
}
if let Some(client) = extract_pool!(&connections, &pool_key, InfluxDb) {
drop(connections);
return db::influxdb_driver::list_tables(&client, database)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset));
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types));
}
if object_types.is_some() {
if let Some(client) = extract_pool!(&connections, &pool_key, SqlServer) {
drop(connections);
let mut client = client.lock().await;
return db::sqlserver::list_tables(&mut client, schema, filter, None, None)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types));
}
}
try_sqlserver!(connections, &pool_key, list_tables, schema, filter, limit, offset);
if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
@@ -495,14 +506,22 @@ async fn list_tables_once(
drop(connections);
let mut client = client.lock().await;
match client.list_tables::<Vec<db::TableInfo>>(database, schema).await {
Ok(tables) if !tables.is_empty() => return Ok(filter_table_infos(tables, filter, limit, offset)),
Ok(tables) if !tables.is_empty() => {
return Ok(filter_table_infos(tables, filter, limit, offset, object_types))
}
Ok(tables) => {
if let Some(config) = fallback_config.as_ref() {
match native_postgres_metadata_pool(state, connection_id, database, config).await {
Ok(Some(pool)) => {
return db::postgres::list_tables_filtered(&pool, schema, filter, limit, offset).await;
return if object_types.is_some() {
db::postgres::list_tables_filtered(&pool, schema, filter, None, None)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types))
} else {
db::postgres::list_tables_filtered(&pool, schema, filter, limit, offset).await
};
}
Ok(None) => return Ok(filter_table_infos(tables, filter, limit, offset)),
Ok(None) => return Ok(filter_table_infos(tables, filter, limit, offset, object_types)),
Err(error) => {
log::warn!(
"[schema][agent:list_tables:fallback-failed] connection_id={} database={} schema={} error={}",
@@ -514,20 +533,23 @@ async fn list_tables_once(
}
}
}
return Ok(filter_table_infos(tables, filter, limit, offset));
return Ok(filter_table_infos(tables, filter, limit, offset, object_types));
}
Err(agent_error) => {
if let Some(config) = fallback_config.as_ref() {
if let Some(pool) =
native_postgres_metadata_pool(state, connection_id, database, config).await?
{
return db::postgres::list_tables_filtered(&pool, schema, filter, limit, offset)
.await
.map_err(|fallback_error| {
format!(
"{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}"
)
});
let result = if object_types.is_some() {
db::postgres::list_tables_filtered(&pool, schema, filter, None, None)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types))
} else {
db::postgres::list_tables_filtered(&pool, schema, filter, limit, offset).await
};
return result.map_err(|fallback_error| {
format!("{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}")
});
}
}
return Err(agent_error);
@@ -543,30 +565,40 @@ async fn list_tables_once(
PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(is_doris_family_config) => {
db::mysql::list_tables_show(p, database)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset))
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types))
}
PoolKind::Mysql(p, mode) => {
dispatch_mysql!(p, mode, db::mysql::list_tables, db::ob_oracle::list_tables, schema)
.map(|tables| filter_table_infos(tables, filter, limit, offset))
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types))
}
PoolKind::Postgres(p) if db_config.as_ref().is_some_and(is_questdb_config) => {
db::questdb::list_tables(p, schema).await.map(|tables| filter_table_infos(tables, filter, limit, offset))
db::questdb::list_tables(p, schema)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types))
}
PoolKind::Postgres(p) => db::postgres::list_tables_filtered(p, schema, filter, limit, offset).await,
PoolKind::Sqlite(p) => {
db::sqlite::list_tables(p, schema).await.map(|tables| filter_table_infos(tables, filter, limit, offset))
PoolKind::Postgres(p) => {
if object_types.is_some() {
db::postgres::list_tables_filtered(p, schema, filter, None, None)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types))
} else {
db::postgres::list_tables_filtered(p, schema, filter, limit, offset).await
}
}
PoolKind::Sqlite(p) => db::sqlite::list_tables(p, schema)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types)),
PoolKind::Rqlite(client) => db::rqlite_driver::list_tables(client, schema)
.await
.map(|tables| filter_table_infos(tables, filter, limit, offset)),
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types)),
PoolKind::MongoDb(client) => db::mongo_driver::list_collections(client, database)
.await
.map(|names| collection_names_to_tables(names, "COLLECTION"))
.map(|tables| filter_table_infos(tables, filter, limit, offset)),
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types)),
PoolKind::Elasticsearch(client) => db::elasticsearch_driver::list_indices(client)
.await
.map(|names| collection_names_to_tables(names, "INDEX"))
.map(|tables| filter_table_infos(tables, filter, limit, offset)),
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types)),
_ => Ok(vec![]),
}
}
@@ -589,6 +621,7 @@ fn filter_table_infos(
filter: Option<&str>,
limit: Option<usize>,
offset: Option<usize>,
object_types: Option<&[String]>,
) -> Vec<db::TableInfo> {
let filter = filter.unwrap_or("").to_lowercase();
let limit = limit.unwrap_or(usize::MAX);
@@ -596,11 +629,40 @@ fn filter_table_infos(
tables
.into_iter()
.filter(|table| filter.is_empty() || table.name.to_lowercase().contains(&filter))
.filter(|table| table_info_matches_object_types(table, object_types))
.skip(offset)
.take(limit)
.collect()
}
fn table_info_matches_object_types(table: &db::TableInfo, object_types: Option<&[String]>) -> bool {
let Some(object_types) = object_types else {
return true;
};
if object_types.is_empty() {
return true;
}
let table_type = normalize_table_info_object_type(&table.table_type);
object_types.iter().any(|object_type| normalize_table_info_object_type(object_type) == table_type)
}
fn normalize_table_info_object_type(value: &str) -> String {
let upper = value.to_ascii_uppercase().replace(' ', "_");
if upper.contains("MATERIALIZED") && upper.contains("VIEW") {
return "MATERIALIZED_VIEW".to_string();
}
if upper.contains("VIEW") {
return "VIEW".to_string();
}
if upper.contains("COLLECTION") {
return "COLLECTION".to_string();
}
if upper.contains("INDEX") {
return "INDEX".to_string();
}
"TABLE".to_string()
}
#[cfg(test)]
mod tests {
use super::{
@@ -738,12 +800,40 @@ mod tests {
test_table_info("users"),
];
let filtered = filter_table_infos(tables, Some("audit"), Some(1), Some(1));
let filtered = filter_table_infos(tables, Some("audit"), Some(1), Some(1), None);
assert_eq!(filtered.len(), 1);
assert_eq!(filtered[0].name, "audit_record");
}
#[test]
fn filter_table_infos_filters_object_type_before_offset_and_limit() {
let tables = vec![
test_table_info("orders"),
super::db::TableInfo {
name: "active_orders".to_string(),
table_type: "VIEW".to_string(),
comment: None,
parent_schema: None,
parent_name: None,
},
test_table_info("users"),
super::db::TableInfo {
name: "active_users".to_string(),
table_type: "VIEW".to_string(),
comment: None,
parent_schema: None,
parent_name: None,
},
];
let object_types = vec!["VIEW".to_string()];
let filtered = filter_table_infos(tables, None, Some(1), Some(1), Some(&object_types));
assert_eq!(filtered.len(), 1);
assert_eq!(filtered[0].name, "active_users");
}
#[cfg(feature = "duckdb-bundled")]
#[test]
fn duckdb_list_databases_includes_attached_database() {
@@ -954,7 +1044,7 @@ async fn list_objects_once(
PoolKind::Postgres(p) => db::postgres::list_objects(p, schema).await,
_ => {
drop(connections);
Ok(list_tables_core(state, connection_id, database, schema, None, None, None)
Ok(list_tables_core(state, connection_id, database, schema, None, None, None, None)
.await?
.into_iter()
.map(|table| db::ObjectInfo {
@@ -1473,7 +1563,20 @@ pub async fn get_table_ddl_core(
database: &str,
schema: &str,
table: &str,
object_type: Option<db::ObjectSourceKind>,
) -> Result<String, String> {
if matches!(object_type, Some(db::ObjectSourceKind::View)) {
let source =
get_object_source_core(state, connection_id, database, schema, table, db::ObjectSourceKind::View).await?;
let database_type = connection_config(state, connection_id).await.map(|config| config.db_type);
return Ok(crate::object_source_sql::build_view_ddl_sql(crate::object_source_sql::BuildViewDdlInput {
database_type,
schema: if schema.trim().is_empty() { None } else { Some(schema.to_string()) },
name: table.to_string(),
source: source.source,
}));
}
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
{
+2
View File
@@ -2746,6 +2746,7 @@ where
Some(table),
Some(1),
None,
None,
)
.await
.unwrap_or_default()
@@ -2761,6 +2762,7 @@ where
Some(table),
Some(1),
None,
None,
)
.await
.map(|tables| !tables.is_empty())