feat(argo): add argodb connection type with routine browsing

This commit is contained in:
wangfengxiang authored and GitHub committed 2026-09-01 00:21:10 +08:00
1 parent 5eb071bc22
commit 2f71a1a8aa
34 files changed
+492 -48

No files matched your search

+2 -1
View File
@@ -661,7 +661,8 @@ fn append_agent_url_params(base: String, params: Option<&str>) -> String {
}
pub fn hive_uses_zookeeper_discovery(config: &ConnectionConfig) -> bool {
if !matches!(config.db_type, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala) {
if !matches!(config.db_type, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo)
{
return false;
}
+1
View File
@@ -249,6 +249,7 @@ macro_rules! agent_connection_pool_database_type {
| DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Spark
| DatabaseType::Db2
| DatabaseType::Informix
+5
View File
@@ -2052,6 +2052,10 @@ impl AgentDriverClient {
}
if let Some(object_types) = object_types {
params["object_types"] = serde_json::json!(object_types);
// Agent drivers (hive-go, oracle-go, etc.) read the object-type
// filter from a camelCase field; keep both spellings so existing
// agents don't need a coordinated rollout.
params["objectTypes"] = serde_json::json!(object_types);
}
self.call_method_with_timeout(AgentMethod::ListTables, params, timeout_duration).await
}
@@ -2087,6 +2091,7 @@ impl AgentDriverClient {
}
if let Some(object_types) = object_types {
params["object_types"] = serde_json::json!(object_types);
params["objectTypes"] = serde_json::json!(object_types);
}
self.call_method_with_timeout(AgentMethod::ListObjects, params, timeout_duration).await
}
+17 -14
View File
@@ -538,6 +538,7 @@ pub fn build_empty_table_sql(options: TableAdminSqlOptions) -> String {
DatabaseType::Cassandra
| DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Argo
| DatabaseType::Kylin
| DatabaseType::Questdb,
) => {
@@ -739,20 +740,22 @@ pub fn build_duplicate_table_structure_sql(options: DuplicateTableStructureSqlOp
);
let target =
qualified_duplicate_target_name(options.database_type, options.schema.as_deref(), &options.target_name);
let structure_sql =
if matches!(options.database_type, Some(DatabaseType::Mysql | DatabaseType::Kyuubi | DatabaseType::Impala)) {
format!("CREATE TABLE {target} LIKE {source};")
} else if options.database_type == Some(DatabaseType::Questdb) {
format!("CREATE TABLE {target} (LIKE {source});")
} else if options.database_type.is_some_and(is_postgres_like_structure_copy) {
format!("CREATE TABLE {target} (LIKE {source} INCLUDING ALL);")
} else if options.database_type == Some(DatabaseType::SqlServer) {
format!("SELECT TOP 0 * INTO {target} FROM {source};")
} else if options.database_type.is_some_and(uses_false_predicate_duplicate_structure) {
format!("CREATE TABLE {target} AS SELECT * FROM {source} WHERE 1=0")
} else {
format!("CREATE TABLE {target} AS SELECT * FROM {source} WHERE 0;")
};
let structure_sql = if matches!(
options.database_type,
Some(DatabaseType::Mysql | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo)
) {
format!("CREATE TABLE {target} LIKE {source};")
} else if options.database_type == Some(DatabaseType::Questdb) {
format!("CREATE TABLE {target} (LIKE {source});")
} else if options.database_type.is_some_and(is_postgres_like_structure_copy) {
format!("CREATE TABLE {target} (LIKE {source} INCLUDING ALL);")
} else if options.database_type == Some(DatabaseType::SqlServer) {
format!("SELECT TOP 0 * INTO {target} FROM {source};")
} else if options.database_type.is_some_and(uses_false_predicate_duplicate_structure) {
format!("CREATE TABLE {target} AS SELECT * FROM {source} WHERE 1=0")
} else {
format!("CREATE TABLE {target} AS SELECT * FROM {source} WHERE 0;")
};
let mut comment_sql = Vec::new();
if let Some(database_type) =
+2 -2
View File
@@ -1089,7 +1089,7 @@ impl ConnectionConfig {
DatabaseType::Snowflake => format!("snowflake://{host}/{db_part}"),
DatabaseType::Trino => format!("trino://{host}:{port}{db_part}"),
DatabaseType::PrestoSql => format!("prestosql://{host}:{port}{db_part}"),
DatabaseType::Hive => format!("hive://{host}:{port}{db_part}"),
DatabaseType::Hive | DatabaseType::Argo => format!("hive://{host}:{port}{db_part}"),
DatabaseType::Kyuubi => format!("kyuubi://{host}:{port}{db_part}"),
DatabaseType::Impala => format!("impala://{host}:{port}{db_part}"),
DatabaseType::Spark => format!("spark://{host}:{port}{db_part}"),
@@ -1315,7 +1315,7 @@ impl ConnectionConfig {
DatabaseType::PrestoSql => {
format!("prestosql://{}:{}@{host}:{port}{db_part}", username, password)
}
DatabaseType::Hive => {
DatabaseType::Hive | DatabaseType::Argo => {
format!("hive://{}:{}@{host}:{port}{db_part}", username, password)
}
DatabaseType::Kyuubi => {
@@ -55,6 +55,7 @@ pub fn is_schema_aware(database_type: DatabaseType) -> bool {
| DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Spark
| DatabaseType::Db2
| DatabaseType::Informix
@@ -707,9 +707,9 @@ pub fn profile_for(db_type: DatabaseType) -> DdlDialectProfile {
ClickHouse => clickhouse_profile(),
DuckDb | Questdb | SapHana | Teradata | Snowflake | Trino | PrestoSql | Hive | Kyuubi | Impala | Spark
| Db2 | Informix | Bigquery | Spanner | Kylin | Ignite | Ignite3 | Oscar | Tdengine | Iotdb | Databricks
| Jdbc => conservative_ansi(db_type),
DuckDb | Questdb | SapHana | Teradata | Snowflake | Trino | PrestoSql | Hive | Kyuubi | Impala | Argo
| Spark | Db2 | Informix | Bigquery | Spanner | Kylin | Ignite | Ignite3 | Oscar | Tdengine | Iotdb
| Databricks | Jdbc => conservative_ansi(db_type),
// Non-tabular / not applicable for relational CREATE TABLE
Redis | MongoDb | DynamoDb | Elasticsearch | Easysearch | Meilisearch | Qdrant | Milvus | Weaviate
@@ -131,6 +131,7 @@ pub fn quote_table_identifier(database_type: Option<DatabaseType>, name: &str) -
| DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Spark
| DatabaseType::Databricks
| DatabaseType::Databend
@@ -388,6 +389,7 @@ pub(crate) fn quote_transfer_identifier(name: &str, database_type: &DatabaseType
| DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::Spark
| DatabaseType::Questdb => format!("`{}`", name.replace('`', "``")),
DatabaseType::SqlServer => format!("[{}]", name.replace(']', "]]")),
@@ -610,7 +610,13 @@ pub(super) fn build_select_columns(
}
if !matches!(
database_type,
Some(DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::InfluxDb)
Some(
DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
| DatabaseType::InfluxDb
)
) {
return "*".to_string();
}
+5
View File
@@ -35,6 +35,11 @@ fn quotes_identifiers_by_database_type() {
assert_eq!(quote_table_identifier(Some(DatabaseType::Jdbc), "user name"), "user name");
assert_eq!(quote_table_identifier(Some(DatabaseType::Iotdb), "root.test.device2"), "root.test.device2");
assert_eq!(quote_table_identifier(Some(DatabaseType::Spanner), "user`name"), "`user``name`");
// ArgoDB shares the Hive-family dialect: backticks quote identifiers and
// double quotes are string literals, and schemas qualify table names.
assert_eq!(quote_table_identifier(Some(DatabaseType::Argo), "user`name"), "`user``name`");
assert_eq!(quote_transfer_identifier("user`name", &DatabaseType::Argo), "`user``name`");
assert!(is_schema_aware(DatabaseType::Argo));
}
/// Spanner databases are created in one of two immutable dialects. The connected
+18 -6
View File
@@ -2959,7 +2959,7 @@ pub fn map_column_type(source_type: &str, _source_db: &DatabaseType, target_db:
// Extract basic type, `bigint unsigned` -> `bigint`
base = base.split(' ').next().unwrap_or(base).trim();
if matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala) {
if matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo) {
return match base {
"tinyint" => "TINYINT".into(),
"smallint" | "int2" => "SMALLINT".into(),
@@ -3192,7 +3192,11 @@ fn generate_create_table_ddl_with_column_quoting(
line.push(' ');
line.push_str(&default_clause);
}
if !c.is_nullable && !matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala)
if !c.is_nullable
&& !matches!(
target_db,
DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo
)
{
line.push_str(" NOT NULL");
}
@@ -3216,7 +3220,7 @@ fn generate_create_table_ddl_with_column_quoting(
}
let mut pks = Vec::with_capacity(columns.iter().filter(|c| c.is_primary_key).count());
if !matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala) {
if !matches!(target_db, DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo) {
for c in columns {
if c.is_primary_key {
let qname = transfer_column_identifier(&c.name, target_db, quote_target_column_names);
@@ -3748,7 +3752,7 @@ fn generate_upsert_typed_for_transfer(
fn max_transfer_write_rows(db_type: &DatabaseType, mode: &TransferMode) -> usize {
match (db_type, mode) {
(DatabaseType::SqlServer, TransferMode::Append | TransferMode::Overwrite) => MAX_SQLSERVER_INSERT_ROWS,
(DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala, _) => 500,
(DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala | DatabaseType::Argo, _) => 500,
(DatabaseType::Oracle, TransferMode::Append | TransferMode::Overwrite) => MAX_ORACLE_INSERT_ALL_ROWS,
(DatabaseType::Oracle, TransferMode::Upsert) => MAX_ORACLE_MERGE_ROWS,
_ => usize::MAX,
@@ -7548,7 +7552,11 @@ where
|| (request.mode == TransferMode::Upsert
&& !matches!(
target_db_type,
DatabaseType::ClickHouse | DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala
DatabaseType::ClickHouse
| DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
))
|| matches!(target_db_type, DatabaseType::Postgres | DatabaseType::Dameng);
let target_columns = if needs_target_columns {
@@ -7622,7 +7630,11 @@ where
let (effective_mode, pk_columns) = if request.mode == TransferMode::Upsert {
if matches!(
target_db_type,
DatabaseType::ClickHouse | DatabaseType::Hive | DatabaseType::Kyuubi | DatabaseType::Impala
DatabaseType::ClickHouse
| DatabaseType::Hive
| DatabaseType::Kyuubi
| DatabaseType::Impala
| DatabaseType::Argo
) {
log::warn!("[transfer] upsert not supported for {:?}, falling back to append", target_db_type);
(TransferMode::Append, vec![])