feat(vector): add qdrant and milvus support

This commit is contained in:
t8y2 committed 2026-06-19 10:07:56 +08:00
1 parent fffdcb9db1
commit e27d34a34a
45 files changed
+1116 -29

No files matched your search

+24 -1
View File
@@ -67,6 +67,7 @@ pub enum PoolKind {
ClickHouse(db::clickhouse_driver::ChClient),
SqlServer(Arc<tokio::sync::Mutex<db::sqlserver::SqlServerClient>>),
Elasticsearch(db::elasticsearch_driver::EsClient),
VectorDb(db::vector_driver::VectorClient),
InfluxDb(db::influxdb_driver::InfluxdbClient),
Agent(Arc<tokio::sync::Mutex<db::agent_driver::AgentDriverClient>>),
ExternalTabular(ExternalTabularHandle),
@@ -601,6 +602,23 @@ impl AppState {
db::elasticsearch_driver::test_connection(&mut client, connect_timeout).await?;
PoolKind::Elasticsearch(client)
}
DatabaseType::Qdrant | DatabaseType::Milvus => {
let kind = match db_config.db_type {
DatabaseType::Qdrant => db::vector_driver::VectorDbKind::Qdrant,
DatabaseType::Milvus => db::vector_driver::VectorDbKind::Milvus,
_ => unreachable!(),
};
let client = db::vector_driver::VectorClient::new(
kind,
&url,
Some(&db_config.username),
Some(&db_config.password),
db_config.ssl,
connect_timeout,
);
db::vector_driver::test_connection(&client, connect_timeout).await?;
PoolKind::VectorDb(client)
}
DatabaseType::InfluxDb => {
let username = if db_config.username.is_empty() { None } else { Some(db_config.username.clone()) };
let password = if db_config.password.is_empty() { None } else { Some(db_config.password.clone()) };
@@ -1292,6 +1310,7 @@ enum KeepaliveTarget {
ClickHouse(db::clickhouse_driver::ChClient),
SqlServer(Arc<tokio::sync::Mutex<db::sqlserver::SqlServerClient>>),
Elasticsearch(db::elasticsearch_driver::EsClient),
VectorDb(db::vector_driver::VectorClient),
InfluxDb(db::influxdb_driver::InfluxdbClient),
}
@@ -1308,6 +1327,7 @@ fn keepalive_target_from_pool(pool: &PoolKind, config: &ConnectionConfig) -> Opt
PoolKind::ClickHouse(client) => Some(KeepaliveTarget::ClickHouse(client.clone())),
PoolKind::SqlServer(client) => Some(KeepaliveTarget::SqlServer(client.clone())),
PoolKind::Elasticsearch(client) => Some(KeepaliveTarget::Elasticsearch(client.clone())),
PoolKind::VectorDb(client) => Some(KeepaliveTarget::VectorDb(client.clone())),
PoolKind::InfluxDb(client) => Some(KeepaliveTarget::InfluxDb(client.clone())),
_ => None,
}
@@ -1334,6 +1354,7 @@ async fn ping_keepalive_target(target: &mut KeepaliveTarget, timeout: Duration)
db::sqlserver::test_connection(&mut client).await
}
KeepaliveTarget::Elasticsearch(client) => db::elasticsearch_driver::test_connection(client, timeout).await,
KeepaliveTarget::VectorDb(client) => db::vector_driver::test_connection(client, timeout).await,
KeepaliveTarget::InfluxDb(client) => db::influxdb_driver::test_connection(client, timeout).await,
}
}
@@ -1452,6 +1473,7 @@ pub async fn close_pool_kind(pool: PoolKind) {
PoolKind::ClickHouse(_) => {}
PoolKind::SqlServer(_) => {}
PoolKind::Elasticsearch(_) => {}
PoolKind::VectorDb(_) => {}
PoolKind::InfluxDb(_) => {}
PoolKind::Agent(client) => {
let mut client = client.lock().await;
@@ -1510,7 +1532,8 @@ fn base_pool_key_for(
) -> String {
let is_single_connection_pool = db_type.as_ref().is_some_and(|db_type| {
let is_single = database_capabilities::is_single_connection_pool(db_type)
|| (include_elasticsearch_single_pool && *db_type == DatabaseType::Elasticsearch);
|| (include_elasticsearch_single_pool
&& matches!(db_type, DatabaseType::Elasticsearch | DatabaseType::Qdrant | DatabaseType::Milvus));
is_single && (!database_capabilities::is_agent_type(db_type) || shares_database_pool_with_connection(db_type))
});
+1
View File
@@ -19,6 +19,7 @@ pub mod sqlserver;
pub mod ssh_tunnel;
pub mod transport_layer_tunnel;
pub mod turso_driver;
pub mod vector_driver;
pub mod wkb;
use reqwest::ClientBuilder;
+408
View File
@@ -0,0 +1,408 @@
use percent_encoding::{utf8_percent_encode, AsciiSet, CONTROLS};
use reqwest::Client as HttpClient;
use serde_json::{Map, Value};
use std::error::Error;
use std::time::{Duration, Instant};
use super::{http_client_builder, with_connection_timeout};
use crate::types::QueryResult;
const PATH_SEGMENT_ENCODE_SET: &AsciiSet = &CONTROLS
.add(b' ')
.add(b'"')
.add(b'#')
.add(b'%')
.add(b'/')
.add(b'<')
.add(b'>')
.add(b'?')
.add(b'[')
.add(b'\\')
.add(b']')
.add(b'^')
.add(b'`')
.add(b'{')
.add(b'|')
.add(b'}');
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum VectorDbKind {
Qdrant,
Milvus,
}
impl VectorDbKind {
fn label(self) -> &'static str {
match self {
VectorDbKind::Qdrant => "Qdrant",
VectorDbKind::Milvus => "Milvus",
}
}
}
#[derive(Clone)]
pub struct VectorClient {
kind: VectorDbKind,
http: HttpClient,
base_url: String,
auth: Option<VectorAuth>,
}
#[derive(Clone)]
enum VectorAuth {
Basic(String, String),
Bearer(String),
ApiKey(String),
}
impl VectorClient {
pub fn new(
kind: VectorDbKind,
url: &str,
username: Option<&str>,
password: Option<&str>,
accept_invalid_certs: bool,
timeout: Duration,
) -> Self {
let base_url = url.trim_end_matches('/').to_string();
let auth = vector_auth(kind, username, password);
let builder = http_client_builder(timeout).danger_accept_invalid_certs(accept_invalid_certs);
let http = builder.build().unwrap_or_else(|_| HttpClient::new());
Self { kind, http, base_url, auth }
}
fn get(&self, path: &str) -> reqwest::RequestBuilder {
self.with_auth(self.http.get(format!("{}{}", self.base_url, path)))
}
fn post(&self, path: &str) -> reqwest::RequestBuilder {
self.with_auth(self.http.post(format!("{}{}", self.base_url, path)))
}
fn put(&self, path: &str) -> reqwest::RequestBuilder {
self.with_auth(self.http.put(format!("{}{}", self.base_url, path)))
}
fn delete(&self, path: &str) -> reqwest::RequestBuilder {
self.with_auth(self.http.delete(format!("{}{}", self.base_url, path)))
}
fn with_auth(&self, req: reqwest::RequestBuilder) -> reqwest::RequestBuilder {
match &self.auth {
Some(VectorAuth::Basic(user, pass)) => req.basic_auth(user, Some(pass)),
Some(VectorAuth::Bearer(token)) => req.bearer_auth(token),
Some(VectorAuth::ApiKey(token)) => req.header("api-key", token),
None => req,
}
}
}
fn vector_auth(kind: VectorDbKind, username: Option<&str>, password: Option<&str>) -> Option<VectorAuth> {
let username = username.unwrap_or("").trim();
let password = password.unwrap_or("");
match kind {
VectorDbKind::Qdrant if !username.is_empty() => {
Some(VectorAuth::Basic(username.to_string(), password.to_string()))
}
VectorDbKind::Qdrant if !password.is_empty() => Some(VectorAuth::ApiKey(password.to_string())),
VectorDbKind::Qdrant => None,
VectorDbKind::Milvus if !username.is_empty() => Some(VectorAuth::Bearer(format!("{username}:{password}"))),
VectorDbKind::Milvus => None,
}
}
pub async fn test_connection(client: &VectorClient, timeout: Duration) -> Result<(), String> {
let label = client.kind.label();
let path = match client.kind {
VectorDbKind::Qdrant => "/collections",
VectorDbKind::Milvus => "/v2/vectordb/collections/list",
};
let request = match client.kind {
VectorDbKind::Qdrant => client.get(path),
VectorDbKind::Milvus => client.post(path).json(&serde_json::json!({ "dbName": "default" })),
};
let resp = with_connection_timeout(label, timeout, async {
request.send().await.map_err(|e| format!("{label} connection failed: {}", format_reqwest_error(&e)))
})
.await?;
ensure_success(label, resp).await.map(|_| ())
}
pub async fn list_collections(client: &VectorClient) -> Result<Vec<String>, String> {
match client.kind {
VectorDbKind::Qdrant => list_qdrant_collections(client).await,
VectorDbKind::Milvus => list_milvus_collections(client).await,
}
}
async fn list_qdrant_collections(client: &VectorClient) -> Result<Vec<String>, String> {
let body = send_json(client.get("/collections"), "Qdrant").await?;
let mut names: Vec<String> = body
.pointer("/result/collections")
.and_then(Value::as_array)
.into_iter()
.flatten()
.filter_map(|item| item.get("name").and_then(Value::as_str).map(str::to_string))
.collect();
names.sort();
Ok(names)
}
async fn list_milvus_collections(client: &VectorClient) -> Result<Vec<String>, String> {
let body = send_json(
client.post("/v2/vectordb/collections/list").json(&serde_json::json!({ "dbName": "default" })),
"Milvus",
)
.await?;
let mut names: Vec<String> = match body.get("data") {
Some(Value::Array(items)) => items.iter().filter_map(collection_name_from_milvus_item).collect(),
_ => Vec::new(),
};
names.sort();
Ok(names)
}
fn collection_name_from_milvus_item(item: &Value) -> Option<String> {
item.as_str()
.map(str::to_string)
.or_else(|| item.get("collectionName").and_then(Value::as_str).map(str::to_string))
.or_else(|| item.get("name").and_then(Value::as_str).map(str::to_string))
}
pub async fn find_documents(
client: &VectorClient,
collection: &str,
skip: u64,
limit: i64,
) -> Result<crate::db::mongo_driver::MongoDocumentResult, String> {
let query = match client.kind {
VectorDbKind::Qdrant => format!(
"POST /collections/{}/points/scroll\n{}",
path_segment(collection),
serde_json::json!({
"limit": limit.max(1) as u64,
"offset": if skip == 0 { Value::Null } else { Value::from(skip) },
"with_payload": true,
"with_vector": false,
})
),
VectorDbKind::Milvus => format!(
"POST /v2/vectordb/entities/query\n{}",
serde_json::json!({
"dbName": "default",
"collectionName": collection,
"filter": "",
"limit": limit.max(1) as u64,
"offset": skip,
"outputFields": ["*"],
})
),
};
let result = execute_rest_query(client, &query).await?;
let documents = result
.rows
.into_iter()
.map(|row| {
let mut map = Map::new();
for (idx, column) in result.columns.iter().enumerate() {
map.insert(column.clone(), row.get(idx).cloned().unwrap_or(Value::Null));
}
Value::Object(map)
})
.collect();
Ok(crate::db::mongo_driver::MongoDocumentResult { documents, total: result.affected_rows })
}
pub async fn execute_rest_query(client: &VectorClient, input: &str) -> Result<QueryResult, String> {
let start = Instant::now();
let request = parse_rest_query(client, input)?;
let resp = request.send().await.map_err(|e| format!("{} request failed: {e}", client.kind.label()))?;
let status = resp.status().as_u16();
let body = resp.json::<Value>().await.unwrap_or(Value::Null);
if !(200..300).contains(&status) {
let detail = serde_json::to_string_pretty(&body).unwrap_or_else(|_| body.to_string());
return Err(format!("{} error ({status}): {detail}", client.kind.label()));
}
Ok(json_to_query_result(status, body, start))
}
fn parse_rest_query(client: &VectorClient, input: &str) -> Result<reqwest::RequestBuilder, String> {
let trimmed = input.trim();
if trimmed.is_empty() {
return Err(format!("{} query cannot be empty", client.kind.label()));
}
if !starts_with_http_method(trimmed) {
return default_collection_query(client, trimmed);
}
let (head, body) = trimmed.split_once('\n').map_or((trimmed, ""), |(head, body)| (head.trim(), body.trim()));
let mut parts = head.split_whitespace();
let method = parts.next().unwrap_or("").to_ascii_uppercase();
let path = parts.next().ok_or_else(|| "Vector query path is required".to_string())?;
let path = if path.starts_with('/') { path.to_string() } else { format!("/{path}") };
let req = match method.as_str() {
"GET" => client.get(&path),
"POST" => client.post(&path),
"PUT" => client.put(&path),
"DELETE" => client.delete(&path),
other => return Err(format!("Unsupported vector REST method: {other}")),
};
if body.is_empty() {
Ok(req)
} else {
let json: Value = serde_json::from_str(body).map_err(|e| format!("Vector query body must be JSON: {e}"))?;
Ok(req.json(&json))
}
}
fn default_collection_query(client: &VectorClient, collection: &str) -> Result<reqwest::RequestBuilder, String> {
let collection = collection.trim();
if collection.is_empty() {
return Err("Vector collection name cannot be empty".to_string());
}
match client.kind {
VectorDbKind::Qdrant => Ok(client
.post(&format!("/collections/{}/points/scroll", path_segment(collection)))
.json(&serde_json::json!({ "limit": 100, "with_payload": true, "with_vector": false }))),
VectorDbKind::Milvus => Ok(client.post("/v2/vectordb/entities/query").json(&serde_json::json!({
"dbName": "default",
"collectionName": collection,
"filter": "",
"limit": 100,
"outputFields": ["*"],
}))),
}
}
fn starts_with_http_method(input: &str) -> bool {
["GET ", "POST ", "PUT ", "DELETE "].iter().any(|prefix| input.to_ascii_uppercase().starts_with(prefix))
}
fn path_segment(value: &str) -> String {
utf8_percent_encode(value, PATH_SEGMENT_ENCODE_SET).to_string()
}
async fn send_json(req: reqwest::RequestBuilder, label: &str) -> Result<Value, String> {
let resp = req.send().await.map_err(|e| format!("{label} request failed: {e}"))?;
let resp = ensure_success(label, resp).await?;
resp.json().await.map_err(|e| format!("{label} parse error: {e}"))
}
async fn ensure_success(label: &str, resp: reqwest::Response) -> Result<reqwest::Response, String> {
if resp.status().is_success() {
return Ok(resp);
}
let status = resp.status();
let body = resp.text().await.unwrap_or_default();
Err(format!("{label} error ({status}): {body}"))
}
fn json_to_query_result(status: u16, body: Value, start: Instant) -> QueryResult {
let rows_value =
body.pointer("/result/points").or_else(|| body.get("result")).or_else(|| body.get("data")).cloned();
if let Some(Value::Array(items)) = rows_value {
return values_to_query_result(items, start);
}
if let Some(Value::Array(items)) = body.pointer("/result/collections").cloned() {
return values_to_query_result(items, start);
}
QueryResult {
columns: vec!["status".to_string(), "response".to_string()],
column_types: Vec::new(),
column_sortables: vec![],
rows: vec![vec![
Value::Number(status.into()),
Value::String(serde_json::to_string_pretty(&body).unwrap_or_else(|_| body.to_string())),
]],
affected_rows: 0,
execution_time_ms: start.elapsed().as_millis(),
truncated: false,
session_id: None,
has_more: false,
}
}
fn values_to_query_result(items: Vec<Value>, start: Instant) -> QueryResult {
let docs: Vec<Map<String, Value>> = items.into_iter().map(normalize_row_object).collect();
let mut columns = Vec::<String>::new();
for doc in &docs {
for key in doc.keys() {
if !columns.contains(key) {
columns.push(key.clone());
}
}
}
if columns.is_empty() {
columns.push("value".to_string());
}
let rows: Vec<Vec<Value>> = docs
.iter()
.map(|doc| columns.iter().map(|column| doc.get(column).cloned().unwrap_or(Value::Null)).collect())
.collect();
QueryResult {
columns,
column_types: Vec::new(),
column_sortables: vec![],
affected_rows: rows.len() as u64,
rows,
execution_time_ms: start.elapsed().as_millis(),
truncated: false,
session_id: None,
has_more: false,
}
}
fn normalize_row_object(value: Value) -> Map<String, Value> {
match value {
Value::Object(mut object) => {
if let Some(Value::Object(payload)) = object.remove("payload") {
for (key, value) in payload {
object.entry(key).or_insert(value);
}
}
object
}
other => {
let mut object = Map::new();
object.insert("value".to_string(), other);
object
}
}
}
fn format_reqwest_error(err: &reqwest::Error) -> String {
let mut parts = vec![err.to_string()];
let mut source = err.source();
while let Some(err) = source {
let text = err.to_string();
if !text.is_empty() && !parts.iter().any(|part| part == &text) {
parts.push(text);
}
source = err.source();
}
parts.join(": ")
}
#[cfg(test)]
mod tests {
use super::{starts_with_http_method, values_to_query_result};
use serde_json::json;
use std::time::Instant;
#[test]
fn detects_rest_queries_case_insensitively() {
assert!(starts_with_http_method("post /collections/foo"));
assert!(starts_with_http_method("GET /collections"));
assert!(!starts_with_http_method("collection_name"));
}
#[test]
fn flattens_qdrant_payload_columns() {
let result =
values_to_query_result(vec![json!({"id": 1, "score": 0.9, "payload": {"title": "hello"}})], Instant::now());
assert!(result.columns.contains(&"id".to_string()));
assert!(result.columns.contains(&"score".to_string()));
assert!(result.columns.contains(&"title".to_string()));
}
}
+6 -2
View File
@@ -254,6 +254,10 @@ pub enum DatabaseType {
Oracle,
#[serde(rename = "elasticsearch")]
Elasticsearch,
#[serde(rename = "qdrant")]
Qdrant,
#[serde(rename = "milvus")]
Milvus,
Doris,
#[serde(rename = "starrocks")]
StarRocks,
@@ -725,7 +729,7 @@ impl ConnectionConfig {
format!("mongodb://{host}:{port}{db_part}{suffix}")
}
DatabaseType::Oracle => format!("oracle://{host}:{port}{db_part}"),
DatabaseType::Elasticsearch => {
DatabaseType::Elasticsearch | DatabaseType::Qdrant | DatabaseType::Milvus => {
let scheme = if self.ssl { "https" } else { "http" };
format!("{scheme}://{host}:{port}")
}
@@ -858,7 +862,7 @@ impl ConnectionConfig {
DatabaseType::Oracle => {
format!("oracle://{}:{}@{host}:{port}{db_part}", username, password)
}
DatabaseType::Elasticsearch => {
DatabaseType::Elasticsearch | DatabaseType::Qdrant | DatabaseType::Milvus => {
let scheme = if self.ssl { "https" } else { "http" };
format!("{scheme}://{host}:{port}")
}
+12 -4
View File
@@ -2,6 +2,7 @@ use crate::connection::{AppState, PoolKind};
use crate::db::agent_driver::mongo_document_id_params;
use crate::db::elasticsearch_driver;
use crate::db::mongo_driver::{self, MongoDocumentResult};
use crate::db::vector_driver;
fn sort_names(mut names: Vec<String>) -> Vec<String> {
names.sort_by(|left, right| {
@@ -28,7 +29,7 @@ pub async fn mongo_list_databases_core(state: &AppState, connection_id: &str) ->
}
Err(error) => Err(error),
},
PoolKind::Elasticsearch(_) => Ok(vec!["default".to_string()]),
PoolKind::Elasticsearch(_) | PoolKind::VectorDb(_) => Ok(vec!["default".to_string()]),
PoolKind::Agent(client) => {
let mut client = client.lock().await;
match client.mongo_list_databases::<Vec<serde_json::Value>>().await {
@@ -41,7 +42,7 @@ pub async fn mongo_list_databases_core(state: &AppState, connection_id: &str) ->
Err(error) => Err(error),
}
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
_ => Err("Not a MongoDB/Elasticsearch/vector connection".to_string()),
}
}
@@ -69,11 +70,12 @@ pub async fn mongo_list_collections_core(
match connections.get(connection_id).ok_or("Not found")? {
PoolKind::MongoDb(client) => mongo_driver::list_collections(client, database).await.map(sort_names),
PoolKind::Elasticsearch(client) => elasticsearch_driver::list_indices(client).await.map(sort_names),
PoolKind::VectorDb(client) => vector_driver::list_collections(client).await.map(sort_names),
PoolKind::Agent(client) => {
let mut client = client.lock().await;
client.mongo_list_collections(database).await.map(sort_names)
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
_ => Err("Not a MongoDB/Elasticsearch/vector connection".to_string()),
}
}
@@ -99,6 +101,12 @@ pub async fn document_find_documents_core(
drop(connections);
elasticsearch_driver::find_documents(&client, collection, skip, limit, filter, sort).await
}
PoolKind::VectorDb(client) => {
let client = client.clone();
drop(connections);
let _ = (database, filter, sort);
vector_driver::find_documents(&client, collection, skip, limit).await
}
PoolKind::Agent(client) => {
let mut client = client.lock().await;
client
@@ -112,7 +120,7 @@ pub async fn document_find_documents_core(
}))
.await
}
_ => Err("Not a MongoDB/Elasticsearch connection".to_string()),
_ => Err("Not a MongoDB/Elasticsearch/vector connection".to_string()),
}
}
+11
View File
@@ -957,6 +957,15 @@ pub async fn do_execute(
.await
.map(|result| truncate_result_with_max_rows(result, max_rows))
}
PoolKind::VectorDb(client) => {
let client = client.clone();
let sql = sql.to_string();
let max_rows = options.max_rows;
drop(connections);
wait_for_query_opt(cancel_token, query_timeout, db::vector_driver::execute_rest_query(&client, &sql))
.await
.map(|result| truncate_result_with_max_rows(result, max_rows))
}
PoolKind::Redis(_) => Err("Use Redis-specific commands".to_string()),
PoolKind::MongoDb(_) => Err("Use MongoDB-specific commands".to_string()),
PoolKind::MessageQueue => Err("Use Message Queue-specific commands".to_string()),
@@ -1681,6 +1690,7 @@ pub async fn execute_statements_in_transaction(
| PoolKind::Redis(_)
| PoolKind::MongoDb(_)
| PoolKind::Elasticsearch(_)
| PoolKind::VectorDb(_)
| PoolKind::InfluxDb(_)
| PoolKind::ExternalTabular(_)
| PoolKind::ExternalDriver { .. } => TxPath::None,
@@ -1689,6 +1699,7 @@ pub async fn execute_statements_in_transaction(
| PoolKind::Redis(_)
| PoolKind::MongoDb(_)
| PoolKind::Elasticsearch(_)
| PoolKind::VectorDb(_)
| PoolKind::InfluxDb(_)
| PoolKind::ExternalTabular(_)
| PoolKind::ExternalDriver { .. } => TxPath::None,
@@ -87,6 +87,8 @@ pub fn supports_sql_query(database_type: DatabaseType) -> bool {
DatabaseType::Redis
| DatabaseType::MongoDb
| DatabaseType::Elasticsearch
| DatabaseType::Qdrant
| DatabaseType::Milvus
| DatabaseType::InfluxDb
| DatabaseType::Neo4j
| DatabaseType::Etcd
+4
View File
@@ -678,6 +678,10 @@ async fn list_tables_once(
.await
.map(|names| collection_names_to_tables(names, "INDEX"))
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types)),
PoolKind::VectorDb(client) => db::vector_driver::list_collections(client)
.await
.map(|names| collection_names_to_tables(names, "COLLECTION"))
.map(|tables| filter_table_infos(tables, filter, limit, offset, object_types)),
_ => Ok(vec![]),
}
}
@@ -58,6 +58,9 @@ pub(in crate::schema) async fn list_tables(
PoolKind::Elasticsearch(client) => {
db::elasticsearch_driver::list_indices(client).await.map(|names| collection_names_to_tables(names, "INDEX"))
}
PoolKind::VectorDb(client) => {
db::vector_driver::list_collections(client).await.map(|names| collection_names_to_tables(names, "COLLECTION"))
}
_ => Ok(vec![]),
}
}
@@ -136,6 +139,7 @@ pub(in crate::schema) async fn get_columns(
PoolKind::Rqlite(client) => db::rqlite_driver::get_columns(client, schema, table).await,
PoolKind::Turso(client) => db::turso_driver::get_columns(client, schema, table).await,
PoolKind::Elasticsearch(client) => db::elasticsearch_driver::get_columns(client, table).await,
PoolKind::VectorDb(_) => Ok(vec![]),
_ => Ok(vec![]),
}
}
@@ -333,4 +337,4 @@ fn is_mysql_system_database(name: &str) -> bool {
fn is_questdb_config(config: &ConnectionConfig) -> bool {
matches!(config.db_type, DatabaseType::Questdb)
|| matches!(config.driver_profile.as_deref(), Some("questdb"))
}
}
@@ -82,6 +82,10 @@ pub fn build_table_data_select_sql(options: TableDataSelectSqlOptions) -> String
);
}
if database_type == Some(DatabaseType::OceanbaseOracle) {
return build_oceanbase_oracle_table_select_sql(&table_alias, &where_clause, &order, &select_columns, limit);
}
if database_type == Some(DatabaseType::Oracle) {
return format!("SELECT {select_columns} FROM {table_alias}{where_clause}{order}");
}
@@ -165,6 +169,11 @@ pub fn build_table_select_sql(options: TableSelectSqlOptions<'_>) -> String {
return format!("SELECT FIRST {limit} {select_columns} FROM {table}{order_by}");
}
if database_type == Some(DatabaseType::OceanbaseOracle) {
let inner_select = format!("SELECT {select_columns} FROM {table}{order_by}");
return format!("SELECT {select_columns} FROM ({inner_select}) WHERE ROWNUM <= {limit}");
}
if database_type.is_some_and(uses_fetch_first) {
return format!("SELECT {select_columns} FROM {table}{order_by} FETCH FIRST {limit} ROWS ONLY");
}
@@ -189,6 +198,17 @@ fn informix_row_limit_clause(limit: usize, offset: usize) -> String {
}
}
fn build_oceanbase_oracle_table_select_sql(
table: &str,
where_clause: &str,
order: &str,
select_columns: &str,
limit: usize,
) -> String {
let inner_select = format!("SELECT {select_columns} FROM {table}{where_clause}{order}");
format!("SELECT {select_columns} FROM ({inner_select}) WHERE ROWNUM <= {limit}")
}
pub(super) fn is_oracle_row_id(database_type: Option<DatabaseType>, name: &str) -> bool {
database_type == Some(DatabaseType::Oracle) && name.eq_ignore_ascii_case(DBX_ROWID_COLUMN)
}
+43
View File
@@ -90,6 +90,17 @@ fn builds_select_sql_with_limit_syntax_for_database_type() {
}),
"SELECT \"id\", \"name\" FROM \"DB2INST1\".\"USERS\" ORDER BY \"id\" ASC FETCH FIRST 100 ROWS ONLY"
);
assert_eq!(
build_table_select_sql(TableSelectSqlOptions {
database_type: Some(DatabaseType::OceanbaseOracle),
schema: Some("DBXTEST"),
table_name: "USERS",
columns: &columns,
order_columns: &keys,
limit: 100,
}),
"SELECT \"id\", \"name\" FROM (SELECT \"id\", \"name\" FROM \"DBXTEST\".\"USERS\" ORDER BY \"id\" ASC) WHERE ROWNUM <= 100"
);
// JDBC connections skip SQL-level row limiting — the JDBC agent handles
// it via Statement.setMaxRows() which is universally supported.
assert_eq!(
@@ -269,6 +280,38 @@ fn builds_table_data_where_and_schema_queries() {
}),
"SELECT * FROM \"DB2INST1\".\"ORDERS\" WHERE (amount > 10) FETCH FIRST 50 ROWS ONLY"
);
assert_eq!(
build_table_data_select_sql(TableDataSelectSqlOptions {
database_type: Some(DatabaseType::OceanbaseOracle),
schema: Some("DBXTEST".to_string()),
table_name: "ORDERS".to_string(),
primary_keys: Vec::new(),
columns: Vec::new(),
fallback_order_columns: Vec::new(),
order_by: None,
limit: Some(100),
offset: None,
where_input: None,
include_row_id: false,
}),
"SELECT * FROM (SELECT * FROM \"DBXTEST\".\"ORDERS\") WHERE ROWNUM <= 100"
);
assert_eq!(
build_table_data_select_sql(TableDataSelectSqlOptions {
database_type: Some(DatabaseType::OceanbaseOracle),
schema: Some("DBXTEST".to_string()),
table_name: "ORDERS".to_string(),
primary_keys: vec!["ID".to_string()],
columns: Vec::new(),
fallback_order_columns: Vec::new(),
order_by: None,
limit: Some(50),
offset: None,
where_input: Some("WHERE amount > 10".to_string()),
include_row_id: false,
}),
"SELECT * FROM (SELECT * FROM \"DBXTEST\".\"ORDERS\" WHERE (amount > 10) ORDER BY \"ID\" ASC) WHERE ROWNUM <= 50"
);
assert_eq!(
build_table_data_select_sql(TableDataSelectSqlOptions {
database_type: Some(DatabaseType::Db2),