2217 lines
89 KiB
Rust
2217 lines
89 KiB
Rust
use crate::connection::{connection_url_for_endpoint, database_connection_config, AppState, MysqlMode, PoolKind};
|
|
use crate::db;
|
|
use crate::models::connection::{ConnectionConfig, DatabaseType};
|
|
use crate::query::{agent_execute_query_params, QueryExecutionOptions};
|
|
use std::collections::HashSet;
|
|
use std::future::Future;
|
|
use std::sync::Arc;
|
|
use std::time::Duration;
|
|
|
|
macro_rules! extract_pool {
|
|
($connections:expr, $key:expr, $variant:ident) => {
|
|
$connections.get($key).and_then(|v| match v {
|
|
PoolKind::$variant(val) => Some(val.clone()),
|
|
_ => None,
|
|
})
|
|
};
|
|
}
|
|
|
|
macro_rules! dispatch_mysql {
|
|
($p:expr, $mode:expr, $mysql:path, $ob:path $(, $arg:expr)*) => {
|
|
if *$mode == MysqlMode::OceanBaseOracle {
|
|
$ob($p $(, $arg)*).await
|
|
} else {
|
|
$mysql($p $(, $arg)*).await
|
|
}
|
|
};
|
|
}
|
|
|
|
macro_rules! try_sqlserver {
|
|
($connections:expr, $pool_key:expr, $method:ident $(, $arg:expr)*) => {
|
|
if let Some(client) = extract_pool!(&$connections, $pool_key, SqlServer) {
|
|
drop($connections);
|
|
let mut client = client.lock().await;
|
|
return db::sqlserver::$method(&mut client $(, $arg)*).await;
|
|
}
|
|
};
|
|
}
|
|
|
|
macro_rules! try_agent {
|
|
($connections:expr, $pool_key:expr, $method:ident $(, $arg:expr)*) => {
|
|
if let Some(client) = extract_pool!(&$connections, $pool_key, Agent) {
|
|
drop($connections);
|
|
let mut client = client.lock().await;
|
|
return client.$method($($arg),*).await;
|
|
}
|
|
};
|
|
}
|
|
|
|
pub fn duckdb_query_tables(con: &duckdb::Connection) -> Result<Vec<db::TableInfo>, String> {
|
|
duckdb_query_tables_in_database(con, "main", "main")
|
|
}
|
|
|
|
pub fn duckdb_query_tables_in_database(
|
|
con: &duckdb::Connection,
|
|
database: &str,
|
|
schema: &str,
|
|
) -> Result<Vec<db::TableInfo>, String> {
|
|
duckdb_query_tables_in_database_with_attached(con, database, schema, &[])
|
|
}
|
|
|
|
pub fn duckdb_query_tables_in_database_with_attached(
|
|
con: &duckdb::Connection,
|
|
database: &str,
|
|
schema: &str,
|
|
attached_names: &[String],
|
|
) -> Result<Vec<db::TableInfo>, String> {
|
|
let database = duckdb_catalog_name(con, database, attached_names)?;
|
|
let mut stmt = con.prepare(
|
|
"SELECT table_name, table_type FROM information_schema.tables WHERE table_catalog = ? AND table_schema = ? ORDER BY table_name"
|
|
).map_err(|e| e.to_string())?;
|
|
let rows = stmt
|
|
.query_map((database.as_str(), schema), |row| {
|
|
Ok(db::TableInfo {
|
|
name: row.get::<_, String>(0)?,
|
|
table_type: row.get::<_, String>(1)?,
|
|
comment: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
})
|
|
})
|
|
.map_err(|e| e.to_string())?;
|
|
Ok(rows.filter_map(|r| r.ok()).collect())
|
|
}
|
|
|
|
pub fn duckdb_attach_database(con: &duckdb::Connection, name: &str, path: &str) -> Result<(), String> {
|
|
let name = name.trim();
|
|
let path = path.trim();
|
|
if name.is_empty() || path.is_empty() {
|
|
return Err("DuckDB attached database name and path are required".to_string());
|
|
}
|
|
let sql = format!("ATTACH {} AS {}", duckdb_quote_string(path), duckdb_quote_ident(name));
|
|
con.execute_batch(&sql).map_err(|e| e.to_string())
|
|
}
|
|
|
|
pub fn duckdb_list_databases(con: &duckdb::Connection) -> Result<Vec<db::DatabaseInfo>, String> {
|
|
duckdb_list_databases_with_attached(con, &[])
|
|
}
|
|
|
|
pub fn duckdb_list_databases_with_attached(
|
|
con: &duckdb::Connection,
|
|
attached_names: &[String],
|
|
) -> Result<Vec<db::DatabaseInfo>, String> {
|
|
let primary = duckdb_primary_catalog(con, attached_names)?;
|
|
let mut stmt = con.prepare("SHOW DATABASES").map_err(|e| e.to_string())?;
|
|
let rows = stmt
|
|
.query_map([], |row| {
|
|
let name = row.get::<_, String>(0)?;
|
|
Ok(db::DatabaseInfo { name: if name == primary { "main".to_string() } else { name } })
|
|
})
|
|
.map_err(|e| e.to_string())?;
|
|
Ok(rows.filter_map(|row| row.ok()).collect())
|
|
}
|
|
|
|
pub fn duckdb_list_schemas(con: &duckdb::Connection, database: &str) -> Result<Vec<String>, String> {
|
|
duckdb_list_schemas_with_attached(con, database, &[])
|
|
}
|
|
|
|
pub fn duckdb_list_schemas_with_attached(
|
|
con: &duckdb::Connection,
|
|
database: &str,
|
|
attached_names: &[String],
|
|
) -> Result<Vec<String>, String> {
|
|
let database = duckdb_catalog_name(con, database, attached_names)?;
|
|
let mut stmt = con
|
|
.prepare(
|
|
"SELECT schema_name FROM information_schema.schemata WHERE catalog_name = ? AND schema_name NOT IN ('information_schema', 'pg_catalog') ORDER BY schema_name",
|
|
)
|
|
.map_err(|e| e.to_string())?;
|
|
let rows = stmt.query_map([database.as_str()], |row| row.get::<_, String>(0)).map_err(|e| e.to_string())?;
|
|
Ok(rows.filter_map(|r| r.ok()).collect())
|
|
}
|
|
|
|
fn duckdb_catalog_name(con: &duckdb::Connection, database: &str, attached_names: &[String]) -> Result<String, String> {
|
|
if database.trim().is_empty() || database == "main" {
|
|
return duckdb_primary_catalog(con, attached_names);
|
|
}
|
|
Ok(database.to_string())
|
|
}
|
|
|
|
pub fn duckdb_primary_catalog(con: &duckdb::Connection, attached_names: &[String]) -> Result<String, String> {
|
|
if attached_names.is_empty() {
|
|
return duckdb_current_database(con);
|
|
}
|
|
let attached: std::collections::HashSet<String> = attached_names.iter().map(|name| name.to_lowercase()).collect();
|
|
let mut stmt = con.prepare("SHOW DATABASES").map_err(|e| e.to_string())?;
|
|
let rows = stmt.query_map([], |row| row.get::<_, String>(0)).map_err(|e| e.to_string())?;
|
|
for row in rows {
|
|
let name = row.map_err(|e| e.to_string())?;
|
|
if !attached.contains(&name.to_lowercase()) {
|
|
return Ok(name);
|
|
}
|
|
}
|
|
duckdb_current_database(con)
|
|
}
|
|
|
|
fn duckdb_current_database(con: &duckdb::Connection) -> Result<String, String> {
|
|
con.query_row("SELECT current_database()", [], |row| row.get::<_, String>(0)).map_err(|e| e.to_string())
|
|
}
|
|
|
|
fn duckdb_quote_ident(value: &str) -> String {
|
|
format!("\"{}\"", value.replace('"', "\"\""))
|
|
}
|
|
|
|
fn duckdb_quote_string(value: &str) -> String {
|
|
format!("'{}'", value.replace('\'', "''"))
|
|
}
|
|
|
|
pub fn duckdb_query_columns(con: &duckdb::Connection, table: &str) -> Result<Vec<db::ColumnInfo>, String> {
|
|
duckdb_query_columns_in_database(con, "main", "main", table)
|
|
}
|
|
|
|
pub fn duckdb_query_columns_in_database(
|
|
con: &duckdb::Connection,
|
|
database: &str,
|
|
schema: &str,
|
|
table: &str,
|
|
) -> Result<Vec<db::ColumnInfo>, String> {
|
|
duckdb_query_columns_in_database_with_attached(con, database, schema, table, &[])
|
|
}
|
|
|
|
pub fn duckdb_query_columns_in_database_with_attached(
|
|
con: &duckdb::Connection,
|
|
database: &str,
|
|
schema: &str,
|
|
table: &str,
|
|
attached_names: &[String],
|
|
) -> Result<Vec<db::ColumnInfo>, String> {
|
|
let database = duckdb_catalog_name(con, database, attached_names)?;
|
|
let mut pk_stmt = con
|
|
.prepare(
|
|
"SELECT kcu.column_name
|
|
FROM information_schema.table_constraints tc
|
|
JOIN information_schema.key_column_usage kcu
|
|
ON tc.constraint_name = kcu.constraint_name
|
|
AND tc.table_schema = kcu.table_schema
|
|
AND tc.table_name = kcu.table_name
|
|
WHERE tc.constraint_type = 'PRIMARY KEY'
|
|
AND tc.table_catalog = ?
|
|
AND tc.table_schema = ?
|
|
AND tc.table_name = ?
|
|
ORDER BY kcu.ordinal_position",
|
|
)
|
|
.map_err(|e| e.to_string())?;
|
|
let pk_rows = pk_stmt
|
|
.query_map((database.as_str(), schema, table), |row| row.get::<_, String>(0))
|
|
.map_err(|e| e.to_string())?;
|
|
let primary_keys: std::collections::HashSet<String> = pk_rows.filter_map(|r| r.ok()).collect();
|
|
|
|
let mut stmt = con
|
|
.prepare(
|
|
"SELECT column_name, data_type, is_nullable, column_default
|
|
FROM information_schema.columns
|
|
WHERE table_catalog = ? AND table_schema = ? AND table_name = ?
|
|
ORDER BY ordinal_position",
|
|
)
|
|
.map_err(|e| e.to_string())?;
|
|
let rows = stmt
|
|
.query_map((database.as_str(), schema, table), |row| {
|
|
let name = row.get::<_, String>(0)?;
|
|
Ok(db::ColumnInfo {
|
|
is_primary_key: primary_keys.contains(&name),
|
|
name,
|
|
data_type: row.get::<_, String>(1)?,
|
|
is_nullable: row.get::<_, String>(2).unwrap_or_default() == "YES",
|
|
column_default: row.get::<_, Option<String>>(3)?,
|
|
extra: None,
|
|
comment: None,
|
|
numeric_precision: None,
|
|
numeric_scale: None,
|
|
character_maximum_length: None,
|
|
})
|
|
})
|
|
.map_err(|e| e.to_string())?;
|
|
Ok(rows.filter_map(|r| r.ok()).collect())
|
|
}
|
|
|
|
async fn duckdb_attached_database_names(state: &AppState, connection_id: &str) -> Vec<String> {
|
|
state
|
|
.configs
|
|
.read()
|
|
.await
|
|
.get(connection_id)
|
|
.map(|config| config.attached_databases.iter().map(|database| database.name.clone()).collect())
|
|
.unwrap_or_default()
|
|
}
|
|
|
|
fn clickhouse_metadata_database<'a>(database: &'a str, schema: &'a str) -> &'a str {
|
|
if database.is_empty() {
|
|
schema
|
|
} else {
|
|
database
|
|
}
|
|
}
|
|
|
|
pub async fn list_databases_core(state: &AppState, connection_id: &str) -> Result<Vec<db::DatabaseInfo>, String> {
|
|
retry_metadata_connection(state, connection_id, None, || list_databases_once(state, connection_id)).await
|
|
}
|
|
|
|
async fn list_databases_once(state: &AppState, connection_id: &str) -> Result<Vec<db::DatabaseInfo>, String> {
|
|
log::info!("[list_databases] connection_id={connection_id}");
|
|
{
|
|
let connections = state.connections.read().await;
|
|
if extract_pool!(&connections, connection_id, ExternalTabular).is_some() {
|
|
return Ok(vec![db::DatabaseInfo { name: "main".to_string() }]);
|
|
}
|
|
if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(connection_id) {
|
|
let config = config.clone();
|
|
let session = session.clone();
|
|
drop(connections);
|
|
return session
|
|
.invoke::<Vec<db::DatabaseInfo>>("listDatabases", serde_json::json!({ "connection": config.as_ref() }))
|
|
.await;
|
|
}
|
|
if let Some(client) = extract_pool!(&connections, connection_id, ClickHouse) {
|
|
drop(connections);
|
|
return db::clickhouse_driver::list_databases(&client).await;
|
|
}
|
|
try_sqlserver!(connections, connection_id, list_databases);
|
|
if let Some(client) = extract_pool!(&connections, connection_id, Agent) {
|
|
let is_mongo =
|
|
state.configs.read().await.get(connection_id).is_some_and(|c| c.db_type == DatabaseType::MongoDb);
|
|
if is_mongo {
|
|
drop(connections);
|
|
let dbs = crate::mongo_ops::mongo_list_databases_core(state, connection_id).await?;
|
|
return Ok(dbs.into_iter().map(|name| db::DatabaseInfo { name }).collect());
|
|
}
|
|
drop(connections);
|
|
let mut client = client.lock().await;
|
|
return client.list_databases().await;
|
|
}
|
|
}
|
|
|
|
let duckdb_attached_names = duckdb_attached_database_names(state, connection_id).await;
|
|
let db_config = connection_config(state, connection_id).await;
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(connection_id).ok_or("Connection not found")?;
|
|
|
|
match pool {
|
|
PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(is_doris_family_config) => {
|
|
db::mysql::list_databases_show(p).await
|
|
}
|
|
PoolKind::Mysql(p, mode) => dispatch_mysql!(p, mode, db::mysql::list_databases, db::ob_oracle::list_databases),
|
|
PoolKind::Postgres(p) => db::postgres::list_databases(p).await,
|
|
PoolKind::Sqlite(p) => db::sqlite::list_databases(p).await,
|
|
PoolKind::Rqlite(client) => db::rqlite_driver::list_databases(client).await,
|
|
PoolKind::DuckDb(con) => {
|
|
let con = con.lock().map_err(|e| e.to_string())?;
|
|
duckdb_list_databases_with_attached(&con, &duckdb_attached_names)
|
|
}
|
|
_ => Ok(vec![]),
|
|
}
|
|
}
|
|
|
|
pub async fn list_schemas_core(state: &AppState, connection_id: &str, database: &str) -> Result<Vec<String>, String> {
|
|
retry_metadata_connection(state, connection_id, Some(database), || {
|
|
list_schemas_once(state, connection_id, database)
|
|
})
|
|
.await
|
|
}
|
|
|
|
async fn list_schemas_once(state: &AppState, connection_id: &str, database: &str) -> Result<Vec<String>, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
let db_config = connection_config(state, connection_id).await;
|
|
|
|
{
|
|
let connections = state.connections.read().await;
|
|
if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) {
|
|
let config = config.clone();
|
|
let session = session.clone();
|
|
drop(connections);
|
|
return session
|
|
.invoke::<Vec<String>>(
|
|
"listSchemas",
|
|
serde_json::json!({ "connection": config.as_ref(), "database": database }),
|
|
)
|
|
.await;
|
|
}
|
|
try_sqlserver!(connections, &pool_key, list_schemas);
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
|
|
let fallback_config = db_config.clone();
|
|
drop(connections);
|
|
let mut client = client.lock().await;
|
|
match client.list_schemas::<Vec<String>>(database).await {
|
|
Ok(schemas) if !schemas.is_empty() => return Ok(schemas),
|
|
Ok(schemas) => {
|
|
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_schemas(&pool).await,
|
|
Ok(None) => return Ok(schemas),
|
|
Err(error) => {
|
|
log::warn!(
|
|
"[schema][agent:list_schemas:fallback-failed] connection_id={} database={} error={}",
|
|
connection_id,
|
|
database,
|
|
error
|
|
);
|
|
}
|
|
}
|
|
}
|
|
return Ok(schemas);
|
|
}
|
|
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_schemas(&pool).await.map_err(|fallback_error| {
|
|
format!("{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}")
|
|
});
|
|
}
|
|
}
|
|
return Err(agent_error);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
|
|
match pool {
|
|
PoolKind::Postgres(p) => db::postgres::list_schemas(p).await,
|
|
PoolKind::DuckDb(con) => {
|
|
let duckdb_attached_names = duckdb_attached_database_names(state, connection_id).await;
|
|
let con = con.lock().map_err(|e| e.to_string())?;
|
|
duckdb_list_schemas_with_attached(&con, database, &duckdb_attached_names)
|
|
}
|
|
_ => Ok(vec![]),
|
|
}
|
|
}
|
|
|
|
pub async fn list_tables_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
filter: Option<&str>,
|
|
limit: Option<usize>,
|
|
) -> Result<Vec<db::TableInfo>, String> {
|
|
retry_metadata_connection(state, connection_id, Some(database), || {
|
|
list_tables_once(state, connection_id, database, schema, filter, limit)
|
|
})
|
|
.await
|
|
}
|
|
|
|
async fn list_tables_once(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
filter: Option<&str>,
|
|
limit: Option<usize>,
|
|
) -> Result<Vec<db::TableInfo>, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
let duckdb_attached_names = duckdb_attached_database_names(state, connection_id).await;
|
|
let db_config = connection_config(state, connection_id).await;
|
|
|
|
{
|
|
let connections = state.connections.read().await;
|
|
if let Some(ext_pool) = extract_pool!(&connections, &pool_key, ExternalTabular) {
|
|
drop(connections);
|
|
let cache = ext_pool.cache.clone();
|
|
return tokio::task::spawn_blocking(move || {
|
|
let con = cache.lock().map_err(|e| e.to_string())?;
|
|
duckdb_query_tables(&con)
|
|
})
|
|
.await
|
|
.map_err(|e| e.to_string())?;
|
|
}
|
|
if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) {
|
|
let config = config.clone();
|
|
let session = session.clone();
|
|
drop(connections);
|
|
return session
|
|
.invoke::<Vec<db::TableInfo>>(
|
|
"listTables",
|
|
serde_json::json!({ "connection": config.as_ref(), "database": database, "schema": schema }),
|
|
)
|
|
.await;
|
|
}
|
|
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);
|
|
}
|
|
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;
|
|
}
|
|
try_sqlserver!(connections, &pool_key, list_tables, schema, filter, limit);
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
|
|
let fallback_config = db_config.clone();
|
|
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_for_config(tables, filter, limit, db_config.as_ref()))
|
|
}
|
|
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(&pool, schema).await.map(|tables| {
|
|
filter_table_infos_for_config(tables, filter, limit, db_config.as_ref())
|
|
});
|
|
}
|
|
Ok(None) => {
|
|
return Ok(filter_table_infos_for_config(tables, filter, limit, db_config.as_ref()))
|
|
}
|
|
Err(error) => {
|
|
log::warn!(
|
|
"[schema][agent:list_tables:fallback-failed] connection_id={} database={} schema={} error={}",
|
|
connection_id,
|
|
database,
|
|
schema,
|
|
error
|
|
);
|
|
}
|
|
}
|
|
}
|
|
return Ok(filter_table_infos_for_config(tables, filter, limit, db_config.as_ref()));
|
|
}
|
|
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(&pool, schema)
|
|
.await
|
|
.map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref()))
|
|
.map_err(|fallback_error| {
|
|
format!(
|
|
"{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}"
|
|
)
|
|
});
|
|
}
|
|
}
|
|
return Err(agent_error);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
|
|
match pool {
|
|
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_for_config(tables, filter, limit, db_config.as_ref()))
|
|
}
|
|
PoolKind::Mysql(p, mode) => {
|
|
dispatch_mysql!(p, mode, db::mysql::list_tables, db::ob_oracle::list_tables, schema)
|
|
.map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref()))
|
|
}
|
|
PoolKind::Postgres(p) => db::postgres::list_tables(p, schema)
|
|
.await
|
|
.map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref())),
|
|
PoolKind::Sqlite(p) => db::sqlite::list_tables(p, schema)
|
|
.await
|
|
.map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref())),
|
|
PoolKind::Rqlite(client) => db::rqlite_driver::list_tables(client, schema)
|
|
.await
|
|
.map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref())),
|
|
PoolKind::MongoDb(client) => db::mongo_driver::list_collections(client, database)
|
|
.await
|
|
.map(|names| collection_names_to_tables(names, "COLLECTION"))
|
|
.map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref())),
|
|
PoolKind::Elasticsearch(client) => db::elasticsearch_driver::list_indices(client)
|
|
.await
|
|
.map(|names| collection_names_to_tables(names, "INDEX"))
|
|
.map(|tables| filter_table_infos_for_config(tables, filter, limit, db_config.as_ref())),
|
|
_ => Ok(vec![]),
|
|
}
|
|
}
|
|
|
|
fn collection_names_to_tables(names: Vec<String>, table_type: &str) -> Vec<db::TableInfo> {
|
|
names
|
|
.into_iter()
|
|
.map(|name| db::TableInfo {
|
|
name,
|
|
table_type: table_type.to_string(),
|
|
comment: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn filter_table_infos(tables: Vec<db::TableInfo>, filter: Option<&str>, limit: Option<usize>) -> Vec<db::TableInfo> {
|
|
let filter = filter.unwrap_or("").to_lowercase();
|
|
let limit = limit.unwrap_or(usize::MAX);
|
|
tables
|
|
.into_iter()
|
|
.filter(|table| filter.is_empty() || table.name.to_lowercase().contains(&filter))
|
|
.take(limit)
|
|
.collect()
|
|
}
|
|
|
|
fn filter_table_infos_for_config(
|
|
tables: Vec<db::TableInfo>,
|
|
filter: Option<&str>,
|
|
limit: Option<usize>,
|
|
config: Option<&ConnectionConfig>,
|
|
) -> Vec<db::TableInfo> {
|
|
filter_table_infos(filter_yashandb_recyclebin_tables(tables, config), filter, limit)
|
|
}
|
|
|
|
fn filter_yashandb_recyclebin_tables(
|
|
tables: Vec<db::TableInfo>,
|
|
config: Option<&ConnectionConfig>,
|
|
) -> Vec<db::TableInfo> {
|
|
if !is_yashandb_config(config) {
|
|
return tables;
|
|
}
|
|
tables.into_iter().filter(|table| !is_recyclebin_object_name(&table.name)).collect()
|
|
}
|
|
|
|
fn filter_yashandb_recyclebin_objects(
|
|
objects: Vec<db::ObjectInfo>,
|
|
config: Option<&ConnectionConfig>,
|
|
) -> Vec<db::ObjectInfo> {
|
|
if !is_yashandb_config(config) {
|
|
return objects;
|
|
}
|
|
objects.into_iter().filter(|object| !is_recyclebin_object_name(&object.name)).collect()
|
|
}
|
|
|
|
fn is_yashandb_config(config: Option<&ConnectionConfig>) -> bool {
|
|
config.is_some_and(|config| config.db_type == DatabaseType::Yashandb)
|
|
}
|
|
|
|
fn is_recyclebin_object_name(name: &str) -> bool {
|
|
name.to_ascii_uppercase().starts_with("BIN$")
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::{
|
|
clickhouse_metadata_database, deduplicate_column_infos, duckdb_attach_database, duckdb_list_databases,
|
|
duckdb_query_tables_in_database, filter_objects_by_types, filter_yashandb_recyclebin_objects,
|
|
filter_yashandb_recyclebin_tables, is_agent_postgres_metadata_fallback_config,
|
|
};
|
|
use crate::models::connection::{ConnectionConfig, DatabaseType};
|
|
|
|
fn test_column(name: &str, comment: Option<&str>, is_primary_key: bool) -> super::db::ColumnInfo {
|
|
super::db::ColumnInfo {
|
|
name: name.to_string(),
|
|
data_type: "VARCHAR".to_string(),
|
|
is_nullable: true,
|
|
column_default: None,
|
|
is_primary_key,
|
|
extra: None,
|
|
comment: comment.map(|value| value.to_string()),
|
|
numeric_precision: None,
|
|
numeric_scale: None,
|
|
character_maximum_length: None,
|
|
}
|
|
}
|
|
|
|
fn test_connection_config(db_type: DatabaseType) -> ConnectionConfig {
|
|
ConnectionConfig {
|
|
id: "test".to_string(),
|
|
name: "test".to_string(),
|
|
db_type,
|
|
driver_profile: None,
|
|
driver_label: None,
|
|
url_params: None,
|
|
host: "127.0.0.1".to_string(),
|
|
port: 5432,
|
|
username: "user".to_string(),
|
|
password: "secret".to_string(),
|
|
database: Some("demo".to_string()),
|
|
visible_databases: None,
|
|
attached_databases: Vec::new(),
|
|
color: None,
|
|
transport_layers: Vec::new(),
|
|
connect_timeout_secs: 5,
|
|
query_timeout_secs: 30,
|
|
ssl: false,
|
|
ca_cert_path: String::new(),
|
|
client_cert_path: String::new(),
|
|
client_key_path: String::new(),
|
|
sysdba: false,
|
|
oracle_connection_type: None,
|
|
connection_string: None,
|
|
redis_connection_mode: None,
|
|
redis_sentinel_master: String::new(),
|
|
redis_sentinel_nodes: String::new(),
|
|
redis_sentinel_username: String::new(),
|
|
redis_sentinel_password: String::new(),
|
|
redis_sentinel_tls: false,
|
|
redis_cluster_nodes: String::new(),
|
|
etcd_endpoints: String::new(),
|
|
external_config: None,
|
|
jdbc_driver_class: None,
|
|
jdbc_driver_paths: Vec::new(),
|
|
one_time: false,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn duckdb_list_databases_includes_attached_database() {
|
|
let unique = uuid::Uuid::new_v4();
|
|
let path = std::env::temp_dir().join(format!("dbx-attached-{unique}.duckdb"));
|
|
let _ = std::fs::remove_file(&path);
|
|
let con = duckdb::Connection::open_in_memory().unwrap();
|
|
|
|
duckdb_attach_database(&con, "analytics", path.to_str().unwrap()).unwrap();
|
|
let databases = duckdb_list_databases(&con).unwrap();
|
|
|
|
assert!(databases.iter().any(|database| database.name == "main"));
|
|
assert!(databases.iter().any(|database| database.name == "analytics"));
|
|
|
|
let _ = std::fs::remove_file(path);
|
|
}
|
|
|
|
#[test]
|
|
fn duckdb_query_tables_filters_by_attached_database() {
|
|
let unique = uuid::Uuid::new_v4();
|
|
let path = std::env::temp_dir().join(format!("dbx-attached-tables-{unique}.duckdb"));
|
|
let _ = std::fs::remove_file(&path);
|
|
let con = duckdb::Connection::open_in_memory().unwrap();
|
|
|
|
con.execute_batch("CREATE TABLE main_table(id INTEGER);").unwrap();
|
|
duckdb_attach_database(&con, "analytics", path.to_str().unwrap()).unwrap();
|
|
con.execute_batch("CREATE TABLE analytics.attached_table(id INTEGER);").unwrap();
|
|
|
|
let main_tables = duckdb_query_tables_in_database(&con, "main", "main").unwrap();
|
|
let attached_tables = duckdb_query_tables_in_database(&con, "analytics", "main").unwrap();
|
|
|
|
assert!(main_tables.iter().any(|table| table.name == "main_table"));
|
|
assert!(!main_tables.iter().any(|table| table.name == "attached_table"));
|
|
assert!(attached_tables.iter().any(|table| table.name == "attached_table"));
|
|
assert!(!attached_tables.iter().any(|table| table.name == "main_table"));
|
|
|
|
let _ = std::fs::remove_file(path);
|
|
}
|
|
|
|
#[test]
|
|
fn clickhouse_metadata_uses_schema_when_database_is_empty() {
|
|
assert_eq!(clickhouse_metadata_database("", "testdb"), "testdb");
|
|
assert_eq!(clickhouse_metadata_database("testdb", ""), "testdb");
|
|
assert_eq!(clickhouse_metadata_database("default", "testdb"), "default");
|
|
}
|
|
|
|
#[test]
|
|
fn deduplicates_columns_and_preserves_later_comment() {
|
|
let columns = deduplicate_column_infos(vec![
|
|
test_column("ID", None, false),
|
|
test_column("ID", Some("源主键"), true),
|
|
test_column("TFBH", Some(""), false),
|
|
test_column("TFBH", Some("台账编号"), false),
|
|
]);
|
|
|
|
assert_eq!(columns.len(), 2);
|
|
assert_eq!(columns[0].name, "ID");
|
|
assert_eq!(columns[0].comment.as_deref(), Some("源主键"));
|
|
assert!(columns[0].is_primary_key);
|
|
assert_eq!(columns[1].name, "TFBH");
|
|
assert_eq!(columns[1].comment.as_deref(), Some("台账编号"));
|
|
}
|
|
|
|
#[test]
|
|
fn postgres_like_agent_metadata_fallback_targets_pg_compatible_agents() {
|
|
assert!(is_agent_postgres_metadata_fallback_config(&test_connection_config(DatabaseType::Kingbase)));
|
|
assert!(is_agent_postgres_metadata_fallback_config(&test_connection_config(DatabaseType::Highgo)));
|
|
assert!(is_agent_postgres_metadata_fallback_config(&test_connection_config(DatabaseType::Vastbase)));
|
|
assert!(!is_agent_postgres_metadata_fallback_config(&test_connection_config(DatabaseType::Postgres)));
|
|
assert!(!is_agent_postgres_metadata_fallback_config(&test_connection_config(DatabaseType::Mysql)));
|
|
}
|
|
|
|
#[test]
|
|
fn filters_list_objects_by_normalized_object_types() {
|
|
let objects = vec![
|
|
super::db::ObjectInfo {
|
|
name: "orders".to_string(),
|
|
object_type: "BASE TABLE".to_string(),
|
|
schema: None,
|
|
comment: None,
|
|
created_at: None,
|
|
updated_at: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
},
|
|
super::db::ObjectInfo {
|
|
name: "active_orders".to_string(),
|
|
object_type: "MATERIALIZED VIEW".to_string(),
|
|
schema: None,
|
|
comment: None,
|
|
created_at: None,
|
|
updated_at: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
},
|
|
super::db::ObjectInfo {
|
|
name: "payroll".to_string(),
|
|
object_type: "PACKAGE BODY".to_string(),
|
|
schema: None,
|
|
comment: None,
|
|
created_at: None,
|
|
updated_at: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
},
|
|
];
|
|
|
|
let filtered = filter_objects_by_types(objects, Some(&["VIEW".to_string(), "PACKAGE_BODY".to_string()]));
|
|
|
|
assert_eq!(
|
|
filtered.iter().map(|object| object.name.as_str()).collect::<Vec<_>>(),
|
|
["active_orders", "payroll"]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn filters_yashandb_recyclebin_tables() {
|
|
let tables = vec![
|
|
super::db::TableInfo {
|
|
name: "USERS".to_string(),
|
|
table_type: "TABLE".to_string(),
|
|
comment: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
},
|
|
super::db::TableInfo {
|
|
name: "BIN$abc123==$0".to_string(),
|
|
table_type: "TABLE".to_string(),
|
|
comment: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
},
|
|
];
|
|
|
|
let filtered =
|
|
filter_yashandb_recyclebin_tables(tables.clone(), Some(&test_connection_config(DatabaseType::Yashandb)));
|
|
let oracle = filter_yashandb_recyclebin_tables(tables, Some(&test_connection_config(DatabaseType::Oracle)));
|
|
|
|
assert_eq!(filtered.iter().map(|table| table.name.as_str()).collect::<Vec<_>>(), ["USERS"]);
|
|
assert_eq!(oracle.len(), 2);
|
|
}
|
|
|
|
#[test]
|
|
fn filters_yashandb_recyclebin_objects() {
|
|
let objects = vec![
|
|
super::db::ObjectInfo {
|
|
name: "ORDERS".to_string(),
|
|
object_type: "TABLE".to_string(),
|
|
schema: Some("HR".to_string()),
|
|
comment: None,
|
|
created_at: None,
|
|
updated_at: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
},
|
|
super::db::ObjectInfo {
|
|
name: "bin$deleted".to_string(),
|
|
object_type: "TABLE".to_string(),
|
|
schema: Some("HR".to_string()),
|
|
comment: None,
|
|
created_at: None,
|
|
updated_at: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
},
|
|
];
|
|
|
|
let filtered =
|
|
filter_yashandb_recyclebin_objects(objects, Some(&test_connection_config(DatabaseType::Yashandb)));
|
|
|
|
assert_eq!(filtered.iter().map(|object| object.name.as_str()).collect::<Vec<_>>(), ["ORDERS"]);
|
|
}
|
|
}
|
|
|
|
pub async fn list_objects_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
object_types: Option<Vec<String>>,
|
|
) -> Result<Vec<db::ObjectInfo>, String> {
|
|
retry_metadata_connection(state, connection_id, Some(database), || {
|
|
list_objects_once(state, connection_id, database, schema, object_types.as_deref())
|
|
})
|
|
.await
|
|
}
|
|
|
|
pub async fn list_completion_objects_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
) -> Result<Vec<db::ObjectInfo>, String> {
|
|
retry_metadata_connection(state, connection_id, Some(database), || {
|
|
list_completion_objects_once(state, connection_id, database, schema)
|
|
})
|
|
.await
|
|
}
|
|
|
|
async fn list_objects_once(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
object_types: Option<&[String]>,
|
|
) -> Result<Vec<db::ObjectInfo>, String> {
|
|
let db_config = connection_config(state, connection_id).await;
|
|
list_objects_once_unfiltered(state, connection_id, database, schema)
|
|
.await
|
|
.map(|objects| filter_yashandb_recyclebin_objects(objects, db_config.as_ref()))
|
|
.map(|objects| filter_objects_by_types(objects, object_types))
|
|
}
|
|
|
|
async fn list_objects_once_unfiltered(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
) -> Result<Vec<db::ObjectInfo>, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
let db_config = connection_config(state, connection_id).await;
|
|
|
|
{
|
|
let connections = state.connections.read().await;
|
|
if let Some(ext_pool) = extract_pool!(&connections, &pool_key, ExternalTabular) {
|
|
drop(connections);
|
|
let cache = ext_pool.cache.clone();
|
|
return tokio::task::spawn_blocking(move || {
|
|
let con = cache.lock().map_err(|e| e.to_string())?;
|
|
Ok(duckdb_query_tables(&con)?
|
|
.into_iter()
|
|
.map(|table| db::ObjectInfo {
|
|
name: table.name,
|
|
object_type: table.table_type,
|
|
schema: None,
|
|
comment: table.comment,
|
|
created_at: None,
|
|
updated_at: None,
|
|
parent_schema: table.parent_schema,
|
|
parent_name: table.parent_name,
|
|
})
|
|
.collect())
|
|
})
|
|
.await
|
|
.map_err(|e| e.to_string())?;
|
|
}
|
|
if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) {
|
|
let config = config.clone();
|
|
let session = session.clone();
|
|
drop(connections);
|
|
return session
|
|
.invoke::<Vec<db::ObjectInfo>>(
|
|
"listObjects",
|
|
serde_json::json!({ "connection": config.as_ref(), "database": database, "schema": schema }),
|
|
)
|
|
.await;
|
|
}
|
|
try_sqlserver!(connections, &pool_key, list_objects, schema);
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
|
|
let oracle_object_options = db_config.as_ref().and_then(oracle_agent_object_options);
|
|
let fallback_config = db_config.clone();
|
|
drop(connections);
|
|
if let Some(options) = oracle_object_options {
|
|
return oracle_agent_list_objects(client, database, schema, options).await;
|
|
}
|
|
let mut client = client.lock().await;
|
|
match client.list_objects::<Vec<db::ObjectInfo>>(database, schema).await {
|
|
Ok(objects) if !objects.is_empty() => return Ok(objects),
|
|
Ok(objects) => {
|
|
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_objects(&pool, schema).await,
|
|
Ok(None) => return Ok(objects),
|
|
Err(error) => {
|
|
log::warn!(
|
|
"[schema][agent:list_objects:fallback-failed] connection_id={} database={} schema={} error={}",
|
|
connection_id,
|
|
database,
|
|
schema,
|
|
error
|
|
);
|
|
}
|
|
}
|
|
}
|
|
return Ok(objects);
|
|
}
|
|
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_objects(&pool, schema).await.map_err(|fallback_error| {
|
|
format!("{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}")
|
|
});
|
|
}
|
|
}
|
|
return Err(agent_error);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
|
|
match pool {
|
|
PoolKind::Mysql(p, mode) => {
|
|
// Note: mysql and ob_oracle take different second args (database vs schema)
|
|
if *mode == MysqlMode::OceanBaseOracle {
|
|
db::ob_oracle::list_objects(p, schema).await
|
|
} else if db_config.as_ref().is_some_and(is_doris_family_config) {
|
|
db::mysql::list_table_objects_show(p, database).await
|
|
} else {
|
|
db::mysql::list_objects(p, database).await
|
|
}
|
|
}
|
|
PoolKind::Postgres(p) => db::postgres::list_objects(p, schema).await,
|
|
_ => {
|
|
drop(connections);
|
|
Ok(list_tables_core(state, connection_id, database, schema, None, None)
|
|
.await?
|
|
.into_iter()
|
|
.map(|table| db::ObjectInfo {
|
|
name: table.name,
|
|
object_type: table.table_type,
|
|
schema: if schema.is_empty() { None } else { Some(schema.to_string()) },
|
|
comment: table.comment,
|
|
created_at: None,
|
|
updated_at: None,
|
|
parent_schema: table.parent_schema,
|
|
parent_name: table.parent_name,
|
|
})
|
|
.collect())
|
|
}
|
|
}
|
|
}
|
|
|
|
fn filter_objects_by_types(objects: Vec<db::ObjectInfo>, object_types: Option<&[String]>) -> Vec<db::ObjectInfo> {
|
|
let Some(object_types) = object_types else {
|
|
return objects;
|
|
};
|
|
if object_types.is_empty() {
|
|
return objects;
|
|
}
|
|
let wanted: HashSet<String> =
|
|
object_types.iter().map(|object_type| normalize_object_info_type(object_type)).collect();
|
|
objects.into_iter().filter(|object| wanted.contains(&normalize_object_info_type(&object.object_type))).collect()
|
|
}
|
|
|
|
fn normalize_object_info_type(object_type: &str) -> String {
|
|
let value = object_type.to_ascii_uppercase().replace(' ', "_");
|
|
if value.contains("PACKAGE_BODY") {
|
|
"PACKAGE_BODY".to_string()
|
|
} else if value.contains("PACKAGE") {
|
|
"PACKAGE".to_string()
|
|
} else if value.contains("VIEW") {
|
|
"VIEW".to_string()
|
|
} else if value.contains("PROC") {
|
|
"PROCEDURE".to_string()
|
|
} else if value.contains("FUNC") {
|
|
"FUNCTION".to_string()
|
|
} else {
|
|
"TABLE".to_string()
|
|
}
|
|
}
|
|
|
|
async fn list_completion_objects_once(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
) -> Result<Vec<db::ObjectInfo>, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
let db_config = connection_config(state, connection_id).await;
|
|
|
|
let connections = state.connections.read().await;
|
|
if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) {
|
|
let config = config.clone();
|
|
let session = session.clone();
|
|
drop(connections);
|
|
return session
|
|
.invoke::<Vec<db::ObjectInfo>>(
|
|
"listObjects",
|
|
serde_json::json!({ "connection": config.as_ref(), "database": database, "schema": schema }),
|
|
)
|
|
.await
|
|
.map(filter_completion_objects);
|
|
}
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
|
|
let oracle_object_options = db_config.as_ref().and_then(oracle_agent_object_options);
|
|
let fallback_config = db_config.clone();
|
|
drop(connections);
|
|
let objects = if let Some(options) = oracle_object_options {
|
|
oracle_agent_list_objects(client, database, schema, options).await?
|
|
} else {
|
|
let mut client = client.lock().await;
|
|
match client.list_objects::<Vec<db::ObjectInfo>>(database, schema).await {
|
|
Ok(objects) if !objects.is_empty() => objects,
|
|
Ok(objects) => {
|
|
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_objects(&pool, schema).await.map(filter_completion_objects)
|
|
}
|
|
Ok(None) => objects,
|
|
Err(error) => {
|
|
log::warn!(
|
|
"[schema][agent:list_completion_objects:fallback-failed] connection_id={} database={} schema={} error={}",
|
|
connection_id,
|
|
database,
|
|
schema,
|
|
error
|
|
);
|
|
objects
|
|
}
|
|
}
|
|
} else {
|
|
objects
|
|
}
|
|
}
|
|
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_objects(&pool, schema)
|
|
.await
|
|
.map(filter_completion_objects)
|
|
.map_err(|fallback_error| {
|
|
format!(
|
|
"{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}"
|
|
)
|
|
});
|
|
}
|
|
}
|
|
return Err(agent_error);
|
|
}
|
|
}
|
|
};
|
|
let objects = filter_yashandb_recyclebin_objects(objects, fallback_config.as_ref());
|
|
return Ok(filter_completion_objects(objects));
|
|
}
|
|
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
match pool {
|
|
PoolKind::Mysql(p, mode) if *mode != MysqlMode::OceanBaseOracle => {
|
|
db::mysql::list_completion_objects(p, database).await
|
|
}
|
|
PoolKind::Mysql(p, mode) if *mode == MysqlMode::OceanBaseOracle => {
|
|
db::ob_oracle::list_objects(p, schema).await.map(filter_completion_objects)
|
|
}
|
|
PoolKind::Postgres(p) => db::postgres::list_objects(p, schema).await.map(filter_completion_objects),
|
|
PoolKind::SqlServer(_) => {
|
|
drop(connections);
|
|
let objects = list_objects_once(state, connection_id, database, schema, None).await?;
|
|
Ok(filter_completion_objects(objects))
|
|
}
|
|
_ => Ok(Vec::new()),
|
|
}
|
|
}
|
|
|
|
fn filter_completion_objects(objects: Vec<db::ObjectInfo>) -> Vec<db::ObjectInfo> {
|
|
objects
|
|
.into_iter()
|
|
.filter(|object| {
|
|
let object_type = object.object_type.to_ascii_uppercase();
|
|
object_type.contains("PROCEDURE") || object_type.contains("FUNCTION") || object_type.contains("TRIGGER")
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
fn is_agent_postgres_metadata_fallback_config(config: &ConnectionConfig) -> bool {
|
|
matches!(config.db_type, DatabaseType::Kingbase | DatabaseType::Highgo | DatabaseType::Vastbase)
|
|
}
|
|
|
|
async fn native_postgres_metadata_pool(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
config: &ConnectionConfig,
|
|
) -> Result<Option<deadpool_postgres::Pool>, String> {
|
|
if !is_agent_postgres_metadata_fallback_config(config) {
|
|
return Ok(None);
|
|
}
|
|
|
|
let mut postgres_config = database_connection_config(config, Some(database));
|
|
postgres_config.db_type = DatabaseType::Postgres;
|
|
let (host, port) = state.connection_host_port(connection_id, &postgres_config).await?;
|
|
let url = connection_url_for_endpoint(&postgres_config, &host, port);
|
|
let connect_timeout = Duration::from_secs(postgres_config.effective_connect_timeout_secs());
|
|
db::postgres::connect(&url, connect_timeout).await.map(Some)
|
|
}
|
|
|
|
async fn retry_metadata_connection<T, F, Fut>(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: Option<&str>,
|
|
mut operation: F,
|
|
) -> Result<T, String>
|
|
where
|
|
F: FnMut() -> Fut,
|
|
Fut: Future<Output = Result<T, String>>,
|
|
{
|
|
let result = operation().await;
|
|
match result {
|
|
Err(error) if crate::query::is_connection_error(&error) => {
|
|
state.reconnect_pool(connection_id, database).await?;
|
|
operation().await
|
|
}
|
|
_ => result,
|
|
}
|
|
}
|
|
|
|
pub async fn get_columns_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
table: &str,
|
|
) -> Result<Vec<db::ColumnInfo>, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
let duckdb_attached_names = duckdb_attached_database_names(state, connection_id).await;
|
|
let db_config = connection_config(state, connection_id).await;
|
|
|
|
{
|
|
let connections = state.connections.read().await;
|
|
if let Some(ext_pool) = extract_pool!(&connections, &pool_key, ExternalTabular) {
|
|
drop(connections);
|
|
let cache = ext_pool.cache.clone();
|
|
let table = table.to_string();
|
|
return tokio::task::spawn_blocking(move || {
|
|
let con = cache.lock().map_err(|e| e.to_string())?;
|
|
duckdb_query_columns(&con, &table)
|
|
})
|
|
.await
|
|
.map_err(|e| e.to_string())?;
|
|
}
|
|
if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) {
|
|
let config = config.clone();
|
|
let session = session.clone();
|
|
drop(connections);
|
|
let columns = session
|
|
.invoke::<Vec<db::ColumnInfo>>(
|
|
"getColumns",
|
|
serde_json::json!({
|
|
"connection": config.as_ref(),
|
|
"database": database,
|
|
"schema": schema,
|
|
"table": table,
|
|
}),
|
|
)
|
|
.await?;
|
|
return Ok(deduplicate_column_infos(columns));
|
|
}
|
|
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_columns_in_database_with_attached(
|
|
&con,
|
|
database,
|
|
schema,
|
|
table,
|
|
&duckdb_attached_names,
|
|
);
|
|
}
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, ClickHouse) {
|
|
drop(connections);
|
|
return db::clickhouse_driver::get_columns(&client, clickhouse_metadata_database(database, schema), table)
|
|
.await
|
|
.map(deduplicate_column_infos);
|
|
}
|
|
try_sqlserver!(connections, &pool_key, get_columns, schema, table);
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
|
|
let fallback_config = db_config.clone();
|
|
drop(connections);
|
|
let mut client = client.lock().await;
|
|
match client.get_columns::<Vec<db::ColumnInfo>>(database, schema, table).await {
|
|
Ok(columns) if !columns.is_empty() => return Ok(deduplicate_column_infos(columns)),
|
|
Ok(columns) => {
|
|
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::get_columns(&pool, schema, table)
|
|
.await
|
|
.map(deduplicate_column_infos);
|
|
}
|
|
Ok(None) => return Ok(deduplicate_column_infos(columns)),
|
|
Err(error) => {
|
|
log::warn!(
|
|
"[schema][agent:get_columns:fallback-failed] connection_id={} database={} schema={} table={} error={}",
|
|
connection_id,
|
|
database,
|
|
schema,
|
|
table,
|
|
error
|
|
);
|
|
}
|
|
}
|
|
}
|
|
return Ok(deduplicate_column_infos(columns));
|
|
}
|
|
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::get_columns(&pool, schema, table)
|
|
.await
|
|
.map(deduplicate_column_infos)
|
|
.map_err(|fallback_error| {
|
|
format!(
|
|
"{agent_error}\n\nNative PostgreSQL metadata fallback failed: {fallback_error}"
|
|
)
|
|
});
|
|
}
|
|
}
|
|
return Err(agent_error);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
|
|
match pool {
|
|
PoolKind::Mysql(p, _) if db_config.as_ref().is_some_and(is_doris_family_config) => {
|
|
db::mysql::get_columns_show(p, database, table).await.map(deduplicate_column_infos)
|
|
}
|
|
PoolKind::Mysql(p, mode) => {
|
|
dispatch_mysql!(p, mode, db::mysql::get_columns, db::ob_oracle::get_columns, database, table)
|
|
.map(deduplicate_column_infos)
|
|
}
|
|
PoolKind::Postgres(p) => db::postgres::get_columns(p, schema, table).await.map(deduplicate_column_infos),
|
|
PoolKind::Sqlite(p) => db::sqlite::get_columns(p, schema, table).await.map(deduplicate_column_infos),
|
|
PoolKind::Rqlite(client) => {
|
|
db::rqlite_driver::get_columns(client, schema, table).await.map(deduplicate_column_infos)
|
|
}
|
|
PoolKind::Elasticsearch(client) => {
|
|
db::elasticsearch_driver::get_columns(client, table).await.map(deduplicate_column_infos)
|
|
}
|
|
_ => Ok(vec![]),
|
|
}
|
|
}
|
|
|
|
fn deduplicate_column_infos(columns: Vec<db::ColumnInfo>) -> Vec<db::ColumnInfo> {
|
|
let mut result: Vec<db::ColumnInfo> = Vec::with_capacity(columns.len());
|
|
for column in columns {
|
|
if let Some(existing) = result.iter_mut().find(|existing| existing.name == column.name) {
|
|
existing.is_primary_key |= column.is_primary_key;
|
|
existing.is_nullable &= column.is_nullable;
|
|
merge_optional_string(&mut existing.column_default, column.column_default);
|
|
merge_optional_string(&mut existing.extra, column.extra);
|
|
merge_optional_string(&mut existing.comment, column.comment);
|
|
if existing.numeric_precision.is_none() {
|
|
existing.numeric_precision = column.numeric_precision;
|
|
}
|
|
if existing.numeric_scale.is_none() {
|
|
existing.numeric_scale = column.numeric_scale;
|
|
}
|
|
if existing.character_maximum_length.is_none() {
|
|
existing.character_maximum_length = column.character_maximum_length;
|
|
}
|
|
if existing.data_type.trim().is_empty() && !column.data_type.trim().is_empty() {
|
|
existing.data_type = column.data_type;
|
|
}
|
|
} else {
|
|
result.push(column);
|
|
}
|
|
}
|
|
result
|
|
}
|
|
|
|
fn merge_optional_string(target: &mut Option<String>, candidate: Option<String>) {
|
|
let Some(candidate) = candidate else {
|
|
return;
|
|
};
|
|
if candidate.trim().is_empty() {
|
|
if target.is_none() {
|
|
*target = Some(candidate);
|
|
}
|
|
return;
|
|
}
|
|
if target.as_ref().is_none_or(|value| value.trim().is_empty()) {
|
|
*target = Some(candidate);
|
|
}
|
|
}
|
|
|
|
pub async fn list_indexes_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
table: &str,
|
|
) -> Result<Vec<db::IndexInfo>, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
|
|
{
|
|
let connections = state.connections.read().await;
|
|
try_sqlserver!(connections, &pool_key, list_indexes, schema, table);
|
|
try_agent!(connections, &pool_key, list_indexes, database, schema, table);
|
|
}
|
|
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
|
|
match pool {
|
|
PoolKind::Mysql(p, mode) => {
|
|
dispatch_mysql!(p, mode, db::mysql::list_indexes, db::ob_oracle::list_indexes, schema, table)
|
|
}
|
|
PoolKind::Postgres(p) => db::postgres::list_indexes(p, schema, table).await,
|
|
PoolKind::Sqlite(p) => db::sqlite::list_indexes(p, schema, table).await,
|
|
PoolKind::Rqlite(client) => db::rqlite_driver::list_indexes(client, schema, table).await,
|
|
PoolKind::MongoDb(client) => db::mongo_driver::list_indexes(client, database, table).await,
|
|
_ => Ok(vec![]),
|
|
}
|
|
}
|
|
|
|
pub async fn list_foreign_keys_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
table: &str,
|
|
) -> Result<Vec<db::ForeignKeyInfo>, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
|
|
{
|
|
let connections = state.connections.read().await;
|
|
try_sqlserver!(connections, &pool_key, list_foreign_keys, schema, table);
|
|
try_agent!(connections, &pool_key, list_foreign_keys, database, schema, table);
|
|
}
|
|
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
|
|
match pool {
|
|
PoolKind::Mysql(p, mode) => {
|
|
dispatch_mysql!(p, mode, db::mysql::list_foreign_keys, db::ob_oracle::list_foreign_keys, schema, table)
|
|
}
|
|
PoolKind::Postgres(p) => db::postgres::list_foreign_keys(p, schema, table).await,
|
|
PoolKind::Sqlite(p) => db::sqlite::list_foreign_keys(p, schema, table).await,
|
|
PoolKind::Rqlite(client) => db::rqlite_driver::list_foreign_keys(client, schema, table).await,
|
|
_ => Ok(vec![]),
|
|
}
|
|
}
|
|
|
|
pub async fn list_triggers_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
table: &str,
|
|
) -> Result<Vec<db::TriggerInfo>, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
|
|
{
|
|
let connections = state.connections.read().await;
|
|
try_sqlserver!(connections, &pool_key, list_triggers, schema, table);
|
|
try_agent!(connections, &pool_key, list_triggers, database, schema, table);
|
|
}
|
|
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
|
|
match pool {
|
|
PoolKind::Mysql(p, mode) => {
|
|
dispatch_mysql!(p, mode, db::mysql::list_triggers, db::ob_oracle::list_triggers, schema, table)
|
|
}
|
|
PoolKind::Postgres(p) => db::postgres::list_triggers(p, schema, table).await,
|
|
PoolKind::Sqlite(p) => db::sqlite::list_triggers(p, schema, table).await,
|
|
PoolKind::Rqlite(client) => db::rqlite_driver::list_triggers(client, schema, table).await,
|
|
_ => Ok(vec![]),
|
|
}
|
|
}
|
|
|
|
pub async fn get_table_ddl_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
table: &str,
|
|
) -> Result<String, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
|
|
{
|
|
let connections = state.connections.read().await;
|
|
if let Some(con) = extract_pool!(&connections, &pool_key, DuckDb) {
|
|
drop(connections);
|
|
let tbl = table.replace('\'', "''");
|
|
let con = con.lock().map_err(|e| e.to_string())?;
|
|
let mut stmt = con
|
|
.prepare(&format!("SELECT sql FROM duckdb_tables() WHERE table_name = '{tbl}'"))
|
|
.map_err(|e| e.to_string())?;
|
|
let mut rows = stmt.query([]).map_err(|e| e.to_string())?;
|
|
if let Some(row) = rows.next().map_err(|e| e.to_string())? {
|
|
return row.get::<_, String>(0).map_err(|e| e.to_string());
|
|
}
|
|
return Err("Table not found".to_string());
|
|
}
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, ClickHouse) {
|
|
drop(connections);
|
|
let clickhouse_database = clickhouse_metadata_database(database, schema);
|
|
let result = db::clickhouse_driver::execute_query(
|
|
&client,
|
|
clickhouse_database,
|
|
&format!("SHOW CREATE TABLE `{table}`"),
|
|
)
|
|
.await?;
|
|
return result
|
|
.rows
|
|
.first()
|
|
.and_then(|r| r.first())
|
|
.and_then(|v| v.as_str())
|
|
.map(|s| s.to_string())
|
|
.ok_or_else(|| "Table not found".to_string());
|
|
}
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, SqlServer) {
|
|
drop(connections);
|
|
let mut client = client.lock().await;
|
|
return build_sqlserver_ddl(&mut client, schema, table).await;
|
|
}
|
|
try_agent!(connections, &pool_key, get_table_ddl, database, schema, table);
|
|
}
|
|
|
|
let connections = state.connections.read().await;
|
|
let pool = connections.get(&pool_key).ok_or("Pool not found")?;
|
|
let db_config = connection_config(state, connection_id).await;
|
|
|
|
match pool {
|
|
PoolKind::Mysql(p, _) => mysql_ddl(p, table).await,
|
|
PoolKind::Postgres(p) if db_config.as_ref().is_some_and(is_opengauss_family_config) => {
|
|
match opengauss_table_ddl(p, schema, table).await {
|
|
Ok(ddl) => Ok(ddl),
|
|
Err(_) => pg_ddl(p, schema, table).await,
|
|
}
|
|
}
|
|
PoolKind::Postgres(p) => pg_ddl(p, schema, table).await,
|
|
PoolKind::Sqlite(p) => sqlite_ddl(p, table).await,
|
|
PoolKind::Rqlite(client) => db::rqlite_driver::table_ddl(client, table).await,
|
|
_ => Err("DDL not supported for this database type".to_string()),
|
|
}
|
|
}
|
|
|
|
async fn connection_config(state: &AppState, connection_id: &str) -> Option<ConnectionConfig> {
|
|
state.configs.read().await.get(connection_id).cloned()
|
|
}
|
|
|
|
fn is_opengauss_family_config(config: &ConnectionConfig) -> bool {
|
|
matches!(config.db_type, DatabaseType::OpenGauss | DatabaseType::Gaussdb)
|
|
|| matches!(config.driver_profile.as_deref(), Some("opengauss" | "gaussdb"))
|
|
}
|
|
|
|
fn is_doris_family_config(config: &ConnectionConfig) -> bool {
|
|
matches!(config.db_type, DatabaseType::Doris | DatabaseType::StarRocks)
|
|
|| matches!(config.driver_profile.as_deref(), Some("doris" | "selectdb" | "starrocks"))
|
|
}
|
|
|
|
fn sql_string(value: &str) -> String {
|
|
format!("'{}'", value.replace('\'', "''"))
|
|
}
|
|
|
|
fn pg_ident(value: &str) -> String {
|
|
format!("\"{}\"", value.replace('"', "\"\""))
|
|
}
|
|
|
|
fn mysql_ident(value: &str) -> String {
|
|
format!("`{}`", value.replace('`', "``"))
|
|
}
|
|
|
|
fn sqlite_object_type(kind: &db::ObjectSourceKind) -> &'static str {
|
|
match kind {
|
|
db::ObjectSourceKind::View => "view",
|
|
db::ObjectSourceKind::Procedure
|
|
| db::ObjectSourceKind::Function
|
|
| db::ObjectSourceKind::Sequence
|
|
| db::ObjectSourceKind::Package
|
|
| db::ObjectSourceKind::PackageBody => "routine",
|
|
}
|
|
}
|
|
|
|
fn sqlserver_object_type_filter(kind: &db::ObjectSourceKind) -> &'static str {
|
|
match kind {
|
|
db::ObjectSourceKind::View => "'V'",
|
|
db::ObjectSourceKind::Procedure => "'P'",
|
|
db::ObjectSourceKind::Function => "'FN','IF','TF','FS','FT'",
|
|
db::ObjectSourceKind::Sequence | db::ObjectSourceKind::Package | db::ObjectSourceKind::PackageBody => "''",
|
|
}
|
|
}
|
|
|
|
pub fn sqlserver_object_source_sql(schema: &str, name: &str, kind: &db::ObjectSourceKind) -> String {
|
|
format!(
|
|
"SELECT m.definition FROM sys.sql_modules m \
|
|
JOIN sys.objects o ON o.object_id = m.object_id \
|
|
JOIN sys.schemas s ON s.schema_id = o.schema_id \
|
|
WHERE s.name = {} AND o.name = {} AND o.type IN ({})",
|
|
sql_string(schema),
|
|
sql_string(name),
|
|
sqlserver_object_type_filter(kind)
|
|
)
|
|
}
|
|
|
|
pub fn postgres_object_source_sql(schema: &str, name: &str, kind: &db::ObjectSourceKind) -> String {
|
|
match kind {
|
|
db::ObjectSourceKind::View => {
|
|
format!(
|
|
"SELECT pg_get_viewdef(c.oid, 0) \
|
|
FROM pg_catalog.pg_class c \
|
|
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
|
|
WHERE n.nspname = {} AND c.relname = {} AND c.relkind IN ('v','m') \
|
|
ORDER BY c.oid LIMIT 1",
|
|
sql_string(schema),
|
|
sql_string(name)
|
|
)
|
|
}
|
|
db::ObjectSourceKind::Procedure | db::ObjectSourceKind::Function => {
|
|
let prokind = if matches!(kind, db::ObjectSourceKind::Procedure) { "p" } else { "f" };
|
|
format!(
|
|
"SELECT pg_get_functiondef(p.oid) \
|
|
FROM pg_proc p \
|
|
JOIN pg_namespace n ON n.oid = p.pronamespace \
|
|
WHERE n.nspname = {} AND p.proname = {} AND p.prokind = '{}' \
|
|
ORDER BY p.oid LIMIT 1",
|
|
sql_string(schema),
|
|
sql_string(name),
|
|
prokind
|
|
)
|
|
}
|
|
db::ObjectSourceKind::Sequence => {
|
|
format!(
|
|
"SELECT concat_ws(E'\\n\\n', \
|
|
'-- auto-generated definition' || E'\\n' || \
|
|
'create sequence ' || quote_ident(c.relname) || E'\\n' || \
|
|
' as ' || pg_catalog.format_type(s.seqtypid, NULL) || ';', \
|
|
'alter sequence ' || quote_ident(c.relname) || ' owner to ' || quote_ident(pg_get_userbyid(c.relowner)) || ';', \
|
|
CASE WHEN owned.relname IS NOT NULL AND a.attname IS NOT NULL \
|
|
THEN 'alter sequence ' || quote_ident(c.relname) || ' owned by ' || quote_ident(owned.relname) || '.' || quote_ident(a.attname) || ';' \
|
|
END \
|
|
) \
|
|
FROM pg_catalog.pg_class c \
|
|
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
|
|
JOIN pg_catalog.pg_sequence s ON s.seqrelid = c.oid \
|
|
LEFT JOIN pg_catalog.pg_depend d \
|
|
ON d.classid = 'pg_class'::regclass AND d.objid = c.oid AND d.deptype = 'a' \
|
|
LEFT JOIN pg_catalog.pg_class owned ON owned.oid = d.refobjid \
|
|
LEFT JOIN pg_catalog.pg_attribute a ON a.attrelid = d.refobjid AND a.attnum = d.refobjsubid \
|
|
WHERE n.nspname = {} AND c.relname = {} AND c.relkind = 'S' \
|
|
ORDER BY c.oid LIMIT 1",
|
|
sql_string(schema),
|
|
sql_string(name)
|
|
)
|
|
}
|
|
db::ObjectSourceKind::Package | db::ObjectSourceKind::PackageBody => "SELECT NULL WHERE FALSE".to_string(),
|
|
}
|
|
}
|
|
|
|
pub fn oracle_object_source_sql(schema: &str, name: &str, kind: &db::ObjectSourceKind) -> String {
|
|
let object_type = match kind {
|
|
db::ObjectSourceKind::View => "VIEW",
|
|
db::ObjectSourceKind::Procedure => "PROCEDURE",
|
|
db::ObjectSourceKind::Function => "FUNCTION",
|
|
db::ObjectSourceKind::Sequence => "SEQUENCE",
|
|
db::ObjectSourceKind::Package => "PACKAGE",
|
|
db::ObjectSourceKind::PackageBody => "PACKAGE_BODY",
|
|
};
|
|
if schema.trim().is_empty() {
|
|
format!("SELECT DBMS_METADATA.GET_DDL({}, {}) FROM DUAL", sql_string(object_type), sql_string(name))
|
|
} else {
|
|
format!(
|
|
"SELECT DBMS_METADATA.GET_DDL({}, {}, {}) FROM DUAL",
|
|
sql_string(object_type),
|
|
sql_string(name),
|
|
sql_string(schema)
|
|
)
|
|
}
|
|
}
|
|
|
|
pub fn sqlite_object_source_sql(name: &str, kind: &db::ObjectSourceKind) -> String {
|
|
format!(
|
|
"SELECT sql FROM sqlite_master WHERE type = {} AND name = {}",
|
|
sql_string(sqlite_object_type(kind)),
|
|
sql_string(name)
|
|
)
|
|
}
|
|
|
|
pub fn mysql_object_source_sql(name: &str, kind: &db::ObjectSourceKind) -> String {
|
|
match kind {
|
|
db::ObjectSourceKind::View => format!("SHOW CREATE VIEW {}", mysql_ident(name)),
|
|
db::ObjectSourceKind::Procedure => format!("SHOW CREATE PROCEDURE {}", mysql_ident(name)),
|
|
db::ObjectSourceKind::Function => format!("SHOW CREATE FUNCTION {}", mysql_ident(name)),
|
|
db::ObjectSourceKind::Sequence | db::ObjectSourceKind::Package | db::ObjectSourceKind::PackageBody => {
|
|
String::new()
|
|
}
|
|
}
|
|
}
|
|
|
|
pub fn postgres_view_source_fallback_sql(schema: &str, name: &str) -> String {
|
|
format!(
|
|
"SELECT definition \
|
|
FROM pg_catalog.pg_views \
|
|
WHERE schemaname = {} AND viewname = {} \
|
|
LIMIT 1",
|
|
sql_string(schema),
|
|
sql_string(name)
|
|
)
|
|
}
|
|
|
|
fn first_string_cell(result: db::QueryResult) -> Result<String, String> {
|
|
result
|
|
.rows
|
|
.first()
|
|
.and_then(|row| row.iter().find_map(|value| value.as_str().map(str::to_string)))
|
|
.filter(|value| !value.trim().is_empty())
|
|
.ok_or_else(|| "Object source not found".to_string())
|
|
}
|
|
|
|
async fn mysql_object_source(
|
|
pool: &db::mysql::MySqlPool,
|
|
name: &str,
|
|
kind: &db::ObjectSourceKind,
|
|
) -> Result<String, String> {
|
|
use mysql_async::prelude::*;
|
|
let sql = mysql_object_source_sql(name, kind);
|
|
let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?;
|
|
let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?;
|
|
let rows: Vec<mysql_async::Row> = result.collect_and_drop().await.map_err(|e| e.to_string())?;
|
|
let row = rows.first().ok_or("Object source not found")?;
|
|
let index = if matches!(kind, db::ObjectSourceKind::View) { 1 } else { 2 };
|
|
row.get_opt::<String, usize>(index)
|
|
.and_then(|result| result.ok())
|
|
.or_else(|| {
|
|
row.get_opt::<Vec<u8>, usize>(index)
|
|
.and_then(|result| result.ok())
|
|
.map(|b| String::from_utf8_lossy(&b).to_string())
|
|
})
|
|
.ok_or_else(|| "Failed to read object source".to_string())
|
|
}
|
|
|
|
pub async fn get_object_source_core(
|
|
state: &AppState,
|
|
connection_id: &str,
|
|
database: &str,
|
|
schema: &str,
|
|
name: &str,
|
|
object_type: db::ObjectSourceKind,
|
|
) -> Result<db::ObjectSource, String> {
|
|
let pool_key = state.get_or_create_pool(connection_id, Some(database)).await?;
|
|
let db_config = connection_config(state, connection_id).await;
|
|
let source = {
|
|
let connections = state.connections.read().await;
|
|
if let Some(PoolKind::ExternalDriver { config, session, .. }) = connections.get(&pool_key) {
|
|
let config = config.clone();
|
|
let session = session.clone();
|
|
drop(connections);
|
|
let result: db::ObjectSource = session
|
|
.invoke(
|
|
"getObjectSource",
|
|
serde_json::json!({
|
|
"connection": config.as_ref(),
|
|
"database": database,
|
|
"schema": schema,
|
|
"name": name,
|
|
"object_type": &object_type,
|
|
}),
|
|
)
|
|
.await?;
|
|
return Ok(result);
|
|
}
|
|
if let Some(client) = extract_pool!(&connections, &pool_key, SqlServer) {
|
|
drop(connections);
|
|
let mut client = client.lock().await;
|
|
first_string_cell(
|
|
db::sqlserver::execute_query(&mut client, &sqlserver_object_source_sql(schema, name, &object_type))
|
|
.await?,
|
|
)?
|
|
} else if let Some(client) = extract_pool!(&connections, &pool_key, Agent) {
|
|
drop(connections);
|
|
if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::Oracle)
|
|
&& matches!(object_type, db::ObjectSourceKind::Package | db::ObjectSourceKind::PackageBody)
|
|
{
|
|
oracle_agent_object_source(client, database, schema, name, &object_type).await?
|
|
} else {
|
|
let mut client = client.lock().await;
|
|
let result: db::ObjectSource = client.get_object_source(database, schema, name, &object_type).await?;
|
|
return Ok(result);
|
|
}
|
|
} else {
|
|
match connections.get(&pool_key).ok_or("Pool not found")? {
|
|
PoolKind::Mysql(pool, _) => mysql_object_source(pool, name, &object_type).await?,
|
|
PoolKind::Postgres(pool) => postgres_object_source(pool, schema, name, &object_type).await?,
|
|
PoolKind::Sqlite(pool) => first_string_cell(
|
|
db::sqlite::execute_query(pool, &sqlite_object_source_sql(name, &object_type)).await?,
|
|
)?,
|
|
PoolKind::Rqlite(client) => {
|
|
return db::rqlite_driver::object_source(client, name, &object_type).await;
|
|
}
|
|
PoolKind::ClickHouse(client) if matches!(object_type, db::ObjectSourceKind::View) => {
|
|
let result = db::clickhouse_driver::execute_query(
|
|
client,
|
|
database,
|
|
&format!("SHOW CREATE TABLE {}", mysql_ident(name)),
|
|
)
|
|
.await?;
|
|
first_string_cell(result)?
|
|
}
|
|
_ => return Err("Object source is not supported for this database type".to_string()),
|
|
}
|
|
}
|
|
};
|
|
|
|
Ok(db::ObjectSource {
|
|
name: name.to_string(),
|
|
object_type,
|
|
schema: if schema.is_empty() { None } else { Some(schema.to_string()) },
|
|
source,
|
|
})
|
|
}
|
|
|
|
fn oracle_owner_filter(schema: &str) -> String {
|
|
let schema = schema.trim();
|
|
if schema.is_empty() {
|
|
"USER".to_string()
|
|
} else {
|
|
sql_string(&schema.to_uppercase())
|
|
}
|
|
}
|
|
|
|
#[derive(Debug, Clone, Copy)]
|
|
struct OracleAgentObjectOptions {
|
|
hide_recyclebin_objects: bool,
|
|
}
|
|
|
|
fn oracle_agent_object_options(config: &ConnectionConfig) -> Option<OracleAgentObjectOptions> {
|
|
match config.db_type {
|
|
DatabaseType::Oracle => Some(OracleAgentObjectOptions { hide_recyclebin_objects: false }),
|
|
DatabaseType::Yashandb => Some(OracleAgentObjectOptions { hide_recyclebin_objects: true }),
|
|
_ => None,
|
|
}
|
|
}
|
|
|
|
pub fn oracle_list_objects_sql(schema: &str, hide_recyclebin_objects: bool) -> String {
|
|
let recyclebin_filter = if hide_recyclebin_objects { " AND object_name NOT LIKE 'BIN$%'" } else { "" };
|
|
format!(
|
|
"SELECT object_name, CASE object_type WHEN 'PACKAGE BODY' THEN 'PACKAGE_BODY' ELSE object_type END AS object_type, owner \
|
|
FROM all_objects \
|
|
WHERE owner = {} AND object_type IN ('TABLE', 'VIEW', 'PROCEDURE', 'FUNCTION', 'PACKAGE', 'PACKAGE BODY'){} \
|
|
ORDER BY CASE object_type WHEN 'TABLE' THEN 0 WHEN 'VIEW' THEN 1 WHEN 'PROCEDURE' THEN 2 WHEN 'FUNCTION' THEN 3 WHEN 'PACKAGE' THEN 4 ELSE 5 END, object_name",
|
|
oracle_owner_filter(schema),
|
|
recyclebin_filter
|
|
)
|
|
}
|
|
|
|
async fn oracle_agent_list_objects(
|
|
client: Arc<tokio::sync::Mutex<db::agent_driver::AgentDriverClient>>,
|
|
database: &str,
|
|
schema: &str,
|
|
options: OracleAgentObjectOptions,
|
|
) -> Result<Vec<db::ObjectInfo>, String> {
|
|
let sql = oracle_list_objects_sql(schema, options.hide_recyclebin_objects);
|
|
let params = agent_execute_query_params(
|
|
&sql,
|
|
if database.is_empty() { None } else { Some(database) },
|
|
if schema.is_empty() { None } else { Some(schema) },
|
|
QueryExecutionOptions { max_rows: Some(10_000), ..Default::default() },
|
|
);
|
|
let mut client = client.lock().await;
|
|
let result: db::QueryResult = client.execute_query(params).await?;
|
|
Ok(result
|
|
.rows
|
|
.into_iter()
|
|
.filter_map(|row| {
|
|
let name = row.first()?.as_str()?.to_string();
|
|
let object_type = row.get(1)?.as_str()?.to_string();
|
|
let schema = row.get(2).and_then(|value| value.as_str()).map(str::to_string);
|
|
Some(db::ObjectInfo {
|
|
name,
|
|
object_type,
|
|
schema,
|
|
comment: None,
|
|
created_at: None,
|
|
updated_at: None,
|
|
parent_schema: None,
|
|
parent_name: None,
|
|
})
|
|
})
|
|
.collect())
|
|
}
|
|
|
|
async fn oracle_agent_object_source(
|
|
client: Arc<tokio::sync::Mutex<db::agent_driver::AgentDriverClient>>,
|
|
database: &str,
|
|
schema: &str,
|
|
name: &str,
|
|
object_type: &db::ObjectSourceKind,
|
|
) -> Result<String, String> {
|
|
let sql = oracle_object_source_sql(schema, name, object_type);
|
|
let params = agent_execute_query_params(
|
|
&sql,
|
|
if database.is_empty() { None } else { Some(database) },
|
|
if schema.is_empty() { None } else { Some(schema) },
|
|
QueryExecutionOptions { max_rows: Some(1), ..Default::default() },
|
|
);
|
|
let mut client = client.lock().await;
|
|
let result: db::QueryResult = client.execute_query(params).await?;
|
|
first_string_cell(result)
|
|
}
|
|
|
|
async fn postgres_object_source(
|
|
pool: &deadpool_postgres::Pool,
|
|
schema: &str,
|
|
name: &str,
|
|
object_type: &db::ObjectSourceKind,
|
|
) -> Result<String, String> {
|
|
let sql = postgres_object_source_sql(schema, name, object_type);
|
|
match db::postgres::execute_query(pool, &sql).await.and_then(first_string_cell) {
|
|
Ok(source) => Ok(source),
|
|
Err(primary_err) if matches!(object_type, db::ObjectSourceKind::View) => {
|
|
let fallback_sql = postgres_view_source_fallback_sql(schema, name);
|
|
db::postgres::execute_query(pool, &fallback_sql)
|
|
.await
|
|
.and_then(first_string_cell)
|
|
.map_err(|fallback_err| format!("{primary_err}; fallback failed: {fallback_err}"))
|
|
}
|
|
Err(err) => Err(err),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod object_source_tests {
|
|
use super::*;
|
|
use crate::types::ObjectSourceKind;
|
|
|
|
#[test]
|
|
fn builds_sqlserver_object_source_sql_for_schema_scoped_routines() {
|
|
assert_eq!(
|
|
sqlserver_object_source_sql("dbo", "refresh_cache", &ObjectSourceKind::Procedure),
|
|
"SELECT m.definition FROM sys.sql_modules m JOIN sys.objects o ON o.object_id = m.object_id JOIN sys.schemas s ON s.schema_id = o.schema_id WHERE s.name = 'dbo' AND o.name = 'refresh_cache' AND o.type IN ('P')"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn builds_postgres_object_source_sql_for_views_and_functions() {
|
|
assert_eq!(
|
|
postgres_object_source_sql("public", "active_users", &ObjectSourceKind::View),
|
|
"SELECT pg_get_viewdef(c.oid, 0) FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace WHERE n.nspname = 'public' AND c.relname = 'active_users' AND c.relkind IN ('v','m') ORDER BY c.oid LIMIT 1"
|
|
);
|
|
assert_eq!(
|
|
postgres_object_source_sql("public", "recalc_score", &ObjectSourceKind::Function),
|
|
"SELECT pg_get_functiondef(p.oid) FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace WHERE n.nspname = 'public' AND p.proname = 'recalc_score' AND p.prokind = 'f' ORDER BY p.oid LIMIT 1"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn builds_postgres_object_source_sql_for_sequences() {
|
|
let sql = postgres_object_source_sql("tenant's schema", "order id seq", &ObjectSourceKind::Sequence);
|
|
|
|
assert!(sql.contains("-- auto-generated definition"));
|
|
assert!(sql.contains("create sequence"));
|
|
assert!(sql.contains("alter sequence"));
|
|
assert!(sql.contains("owner to"));
|
|
assert!(sql.contains("owned by"));
|
|
assert!(sql.contains("pg_catalog.pg_sequence"));
|
|
assert!(sql.contains("n.nspname = 'tenant''s schema'"));
|
|
assert!(sql.contains("c.relname = 'order id seq'"));
|
|
assert!(sql.contains("c.relkind = 'S'"));
|
|
assert!(!sql.contains("MINVALUE"));
|
|
assert!(!sql.contains("START WITH"));
|
|
assert!(!sql.contains("CACHE"));
|
|
assert!(!sql.contains("NO CYCLE"));
|
|
}
|
|
|
|
#[test]
|
|
fn builds_postgres_view_source_sql_without_regclass_cast() {
|
|
let sql = postgres_object_source_sql("tenant's schema", "active users", &ObjectSourceKind::View);
|
|
|
|
assert!(!sql.contains("::regclass"));
|
|
assert!(sql.contains("pg_get_viewdef(c.oid, 0)"));
|
|
assert!(sql.contains("n.nspname = 'tenant''s schema'"));
|
|
assert!(sql.contains("c.relname = 'active users'"));
|
|
assert!(sql.contains("c.relkind IN ('v','m')"));
|
|
}
|
|
|
|
#[test]
|
|
fn builds_postgres_view_source_fallback_sql_from_pg_views() {
|
|
assert_eq!(
|
|
postgres_view_source_fallback_sql("tenant's schema", "active users"),
|
|
"SELECT definition FROM pg_catalog.pg_views WHERE schemaname = 'tenant''s schema' AND viewname = 'active users' LIMIT 1"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn builds_oracle_object_source_sql_using_metadata_api() {
|
|
assert_eq!(
|
|
oracle_object_source_sql("HR", "ACTIVE_USERS", &ObjectSourceKind::View),
|
|
"SELECT DBMS_METADATA.GET_DDL('VIEW', 'ACTIVE_USERS', 'HR') FROM DUAL"
|
|
);
|
|
assert_eq!(
|
|
oracle_object_source_sql("HR", "PAYROLL", &ObjectSourceKind::PackageBody),
|
|
"SELECT DBMS_METADATA.GET_DDL('PACKAGE_BODY', 'PAYROLL', 'HR') FROM DUAL"
|
|
);
|
|
assert_eq!(
|
|
oracle_object_source_sql("", "PAYROLL", &ObjectSourceKind::Package),
|
|
"SELECT DBMS_METADATA.GET_DDL('PACKAGE', 'PAYROLL') FROM DUAL"
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn builds_oracle_list_objects_sql_with_packages() {
|
|
let sql = oracle_list_objects_sql("hr", false);
|
|
|
|
assert!(sql.contains("'PACKAGE'"));
|
|
assert!(sql.contains("'PACKAGE BODY'"));
|
|
assert!(sql.contains("CASE object_type WHEN 'PACKAGE BODY' THEN 'PACKAGE_BODY'"));
|
|
assert!(sql.contains("owner = 'HR'"));
|
|
assert!(!sql.contains("BIN$"));
|
|
}
|
|
|
|
#[test]
|
|
fn builds_yashandb_list_objects_sql_without_recyclebin_objects() {
|
|
let sql = oracle_list_objects_sql("hr", true);
|
|
|
|
assert!(sql.contains("object_name NOT LIKE 'BIN$%'"));
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod ddl_tests {
|
|
use super::*;
|
|
|
|
fn column(name: &str, data_type: &str) -> db::ColumnInfo {
|
|
db::ColumnInfo {
|
|
name: name.to_string(),
|
|
data_type: data_type.to_string(),
|
|
is_nullable: true,
|
|
column_default: None,
|
|
is_primary_key: false,
|
|
extra: None,
|
|
comment: None,
|
|
numeric_precision: None,
|
|
numeric_scale: None,
|
|
character_maximum_length: None,
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn postgres_table_ddl_includes_column_comments() {
|
|
let mut display_name = column("display_name", "text");
|
|
display_name.comment = Some("User's display name".to_string());
|
|
let columns = vec![display_name];
|
|
|
|
let ddl = render_postgres_table_ddl("public", "users", &columns, &[], &[]);
|
|
|
|
assert!(ddl.contains("COMMENT ON COLUMN \"public\".\"users\".\"display_name\" IS 'User''s display name';"));
|
|
}
|
|
|
|
#[test]
|
|
fn opengauss_table_ddl_uses_native_tabledef_function() {
|
|
assert_eq!(
|
|
opengauss_table_ddl_sql("tenant's schema", "active users"),
|
|
"SELECT pg_get_tabledef('\"tenant''s schema\".\"active users\"')"
|
|
);
|
|
}
|
|
}
|
|
|
|
pub async fn mysql_ddl(pool: &db::mysql::MySqlPool, table: &str) -> Result<String, String> {
|
|
use mysql_async::prelude::*;
|
|
let sql = format!("SHOW CREATE TABLE `{}`", table.replace('`', "``"));
|
|
let mut conn = pool.get_conn().await.map_err(|e| e.to_string())?;
|
|
let result = conn.query_iter(&sql).await.map_err(|e| e.to_string())?;
|
|
let rows: Vec<mysql_async::Row> = result.collect_and_drop().await.map_err(|e| e.to_string())?;
|
|
let row = rows.first().ok_or("DDL not found")?;
|
|
row.get_opt::<String, usize>(1)
|
|
.and_then(|result| result.ok())
|
|
.or_else(|| {
|
|
row.get_opt::<Vec<u8>, usize>(1)
|
|
.and_then(|result| result.ok())
|
|
.map(|b| String::from_utf8_lossy(&b).to_string())
|
|
})
|
|
.ok_or_else(|| "Failed to read DDL".to_string())
|
|
}
|
|
|
|
pub async fn sqlite_ddl(pool: &db::sqlite::SqliteHandle, table: &str) -> Result<String, String> {
|
|
let pool = pool.clone();
|
|
let table = table.to_string();
|
|
tokio::task::spawn_blocking(move || {
|
|
pool.with_connection(|conn| {
|
|
conn.query_row("SELECT sql FROM sqlite_master WHERE type='table' AND name=?1", [table], |row| {
|
|
row.get::<_, String>(0)
|
|
})
|
|
.map_err(|e| e.to_string())
|
|
})
|
|
})
|
|
.await
|
|
.map_err(|e| e.to_string())?
|
|
}
|
|
|
|
pub async fn opengauss_table_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -> Result<String, String> {
|
|
first_string_cell(db::postgres::execute_query(pool, &opengauss_table_ddl_sql(schema, table)).await?)
|
|
}
|
|
|
|
pub fn opengauss_table_ddl_sql(schema: &str, table: &str) -> String {
|
|
let qualified_name = format!("{}.{}", pg_ident(schema), pg_ident(table));
|
|
format!("SELECT pg_get_tabledef({})", sql_string(&qualified_name))
|
|
}
|
|
|
|
pub async fn pg_ddl(pool: &deadpool_postgres::Pool, schema: &str, table: &str) -> Result<String, String> {
|
|
let (columns, indexes, fkeys) = tokio::try_join!(
|
|
db::postgres::get_columns(pool, schema, table),
|
|
db::postgres::list_indexes(pool, schema, table),
|
|
db::postgres::list_foreign_keys(pool, schema, table),
|
|
)?;
|
|
|
|
Ok(render_postgres_table_ddl(schema, table, &columns, &indexes, &fkeys))
|
|
}
|
|
|
|
fn render_postgres_table_ddl(
|
|
schema: &str,
|
|
table: &str,
|
|
columns: &[db::ColumnInfo],
|
|
indexes: &[db::IndexInfo],
|
|
fkeys: &[db::ForeignKeyInfo],
|
|
) -> String {
|
|
let table_name = format!("{}.{}", pg_ident(schema), pg_ident(table));
|
|
let mut ddl = format!("CREATE TABLE {table_name} (\n");
|
|
let col_lines: Vec<String> = columns
|
|
.iter()
|
|
.map(|c| {
|
|
let mut line = format!(" {} {}", pg_ident(&c.name), c.data_type);
|
|
if !c.is_nullable {
|
|
line.push_str(" NOT NULL");
|
|
}
|
|
if let Some(ref def) = c.column_default {
|
|
line.push_str(&format!(" DEFAULT {def}"));
|
|
}
|
|
line
|
|
})
|
|
.collect();
|
|
ddl.push_str(&col_lines.join(",\n"));
|
|
|
|
let pks: Vec<&str> = columns.iter().filter(|c| c.is_primary_key).map(|c| c.name.as_str()).collect();
|
|
if !pks.is_empty() {
|
|
ddl.push_str(&format!(",\n PRIMARY KEY ({})", pks.iter().map(|k| pg_ident(k)).collect::<Vec<_>>().join(", ")));
|
|
}
|
|
for fk in fkeys {
|
|
ddl.push_str(&format!(
|
|
",\n CONSTRAINT {} FOREIGN KEY ({}) REFERENCES {}({})",
|
|
pg_ident(&fk.name),
|
|
pg_ident(&fk.column),
|
|
pg_ident(&fk.ref_table),
|
|
pg_ident(&fk.ref_column)
|
|
));
|
|
}
|
|
ddl.push_str("\n);\n");
|
|
|
|
for col in columns {
|
|
if let Some(comment) = col.comment.as_deref().filter(|comment| !comment.is_empty()) {
|
|
ddl.push_str(&format!(
|
|
"\nCOMMENT ON COLUMN {table_name}.{} IS {};",
|
|
pg_ident(&col.name),
|
|
sql_string(comment)
|
|
));
|
|
}
|
|
}
|
|
|
|
for idx in indexes {
|
|
if idx.is_primary {
|
|
continue;
|
|
}
|
|
let unique = if idx.is_unique { "UNIQUE " } else { "" };
|
|
let cols = idx.columns.iter().map(|c| pg_ident(c)).collect::<Vec<_>>().join(", ");
|
|
let using = idx.index_type.as_deref().map(|t| format!(" USING {t}")).unwrap_or_default();
|
|
let include = idx
|
|
.included_columns
|
|
.as_deref()
|
|
.filter(|c| !c.is_empty())
|
|
.map(|cols| format!(" INCLUDE ({})", cols.iter().map(|c| pg_ident(c)).collect::<Vec<_>>().join(", ")))
|
|
.unwrap_or_default();
|
|
let filter = idx.filter.as_deref().map(|f| format!(" WHERE {f}")).unwrap_or_default();
|
|
ddl.push_str(&format!(
|
|
"\nCREATE {unique}INDEX {} ON {table_name}{using} ({cols}){include}{filter};",
|
|
pg_ident(&idx.name)
|
|
));
|
|
if let Some(ref c) = idx.comment {
|
|
ddl.push_str(&format!(
|
|
"\nCOMMENT ON INDEX {}.{} IS {};",
|
|
pg_ident(schema),
|
|
pg_ident(&idx.name),
|
|
sql_string(c)
|
|
));
|
|
}
|
|
}
|
|
ddl
|
|
}
|
|
|
|
pub async fn build_sqlserver_ddl(
|
|
client: &mut db::sqlserver::SqlServerClient,
|
|
schema: &str,
|
|
table: &str,
|
|
) -> Result<String, String> {
|
|
let columns = db::sqlserver::get_columns(client, schema, table).await?;
|
|
let indexes = db::sqlserver::list_indexes(client, schema, table).await?;
|
|
let fkeys = db::sqlserver::list_foreign_keys(client, schema, table).await?;
|
|
|
|
let mut ddl = format!("CREATE TABLE [{schema}].[{table}] (\n");
|
|
let col_lines: Vec<String> = columns
|
|
.iter()
|
|
.map(|c| {
|
|
let mut line = format!(" [{}] {}", c.name, c.data_type);
|
|
if !c.is_nullable {
|
|
line.push_str(" NOT NULL");
|
|
}
|
|
if let Some(ref def) = c.column_default {
|
|
line.push_str(&format!(" DEFAULT {def}"));
|
|
}
|
|
line
|
|
})
|
|
.collect();
|
|
ddl.push_str(&col_lines.join(",\n"));
|
|
|
|
let pks: Vec<&str> = columns.iter().filter(|c| c.is_primary_key).map(|c| c.name.as_str()).collect();
|
|
if !pks.is_empty() {
|
|
ddl.push_str(&format!(
|
|
",\n PRIMARY KEY ({})",
|
|
pks.iter().map(|k| format!("[{k}]")).collect::<Vec<_>>().join(", ")
|
|
));
|
|
}
|
|
for fk in &fkeys {
|
|
ddl.push_str(&format!(
|
|
",\n CONSTRAINT [{}] FOREIGN KEY ([{}]) REFERENCES [{}]([{}])",
|
|
fk.name, fk.column, fk.ref_table, fk.ref_column
|
|
));
|
|
}
|
|
ddl.push_str("\n);\n");
|
|
|
|
for idx in &indexes {
|
|
if idx.is_primary {
|
|
continue;
|
|
}
|
|
let unique = if idx.is_unique { "UNIQUE " } else { "" };
|
|
let idx_type = idx.index_type.as_deref().map(|t| format!("{t} ")).unwrap_or_default();
|
|
let cols = idx.columns.iter().map(|c| format!("[{c}]")).collect::<Vec<_>>().join(", ");
|
|
let include = idx
|
|
.included_columns
|
|
.as_deref()
|
|
.filter(|c| !c.is_empty())
|
|
.map(|cols| format!(" INCLUDE ({})", cols.iter().map(|c| format!("[{c}]")).collect::<Vec<_>>().join(", ")))
|
|
.unwrap_or_default();
|
|
let filter = idx.filter.as_deref().map(|f| format!(" WHERE {f}")).unwrap_or_default();
|
|
ddl.push_str(&format!(
|
|
"\nCREATE {unique}{idx_type}INDEX [{}] ON [{schema}].[{table}] ({cols}){include}{filter};",
|
|
idx.name
|
|
));
|
|
}
|
|
Ok(ddl)
|
|
}
|