github/dbx
4
0 Fork 0
mirror of https://github.com/t8y2/dbx.git synced 2026-11-15 06:51:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(plugin): expose a read-only schema metadata host API

Browse Source
This commit is contained in:
jearhe authored and GitHub committed 2026-09-23 12:51:23 +08:00
1 parent 2f960f104c
commit d5a05a9884
40 files changed
+1445 -86
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

crates/dbx-core/src/ai/agent_loop.rs
+27
View File
@@ -139,6 +139,33 @@ pub async fn run_agent_loop(
max_tokens: Option<u32>,
task_contract: Option<&AiTaskContract>,
is_agent_mode: bool,
) -> Result<String, String> {
run_agent_loop_inner(
config,
system_prompt,
messages,
agent_ctx,
on_event,
cancelled,
max_tokens,
task_contract,
is_agent_mode,
)
.boxed()
.await
}
#[allow(clippy::too_many_arguments)]
async fn run_agent_loop_inner(
config: &AiConfig,
system_prompt: &str,
messages: &[AiMessage],
agent_ctx: &AgentLoopContext,
on_event: impl Fn(AgentEvent) + Send + Sync + Clone + 'static,
cancelled: &Notify,
max_tokens: Option<u32>,
task_contract: Option<&AiTaskContract>,
is_agent_mode: bool,
) -> Result<String, String> {
let contract_system_prompt = augment_system_prompt_with_task_contract(system_prompt, task_contract, is_agent_mode);
let system_prompt = contract_system_prompt.as_str();
crates/dbx-core/src/connection/mod.rs
+49
View File
@@ -1406,6 +1406,55 @@ impl AppState {
self.connections.read().await.keys().any(|key| pool_key_belongs_to_connection(key, connection_id))
}
/// Find an already-registered metadata/workload pool for a metadata read.
/// Unlike `get_or_create_metadata_pool_for_session`, this is a pure lookup:
/// it never validates credentials, starts an agent, or opens a transport.
pub(crate) async fn existing_metadata_pool_key_for_session(
&self,
connection_id: &str,
database: Option<&str>,
client_session_id: Option<&str>,
) -> Option<String> {
let config = {
let configs = self.configs.read().await;
configs.get(connection_id).cloned()
}?;
let pool_database = metadata_pool_database(Some(&config), database);
let mut base_pool_keys =
vec![base_pool_key_for_with_catalog(Some(config.db_type), connection_id, pool_database, None, false)];
// MongoDB document operations use a connection-level pool and send the
// requested database in the command. A host connection can therefore
// legitimately be registered either under the selected database or
// under the connection-level key; probe both without creating either.
if config.db_type == DatabaseType::MongoDb {
let connection_pool_key =
base_pool_key_for_with_catalog(Some(config.db_type), connection_id, None, None, false);
if !base_pool_keys.contains(&connection_pool_key) {
base_pool_keys.push(connection_pool_key);
}
}
let connections = self.connections.read().await;
base_pool_keys
.into_iter()
.flat_map(|base_pool_key| {
[
pool_key_for_session_role(
Some(&config),
base_pool_key.clone(),
client_session_id,
AgentSessionRole::Metadata,
),
pool_key_for_session_role(
Some(&config),
base_pool_key,
client_session_id,
AgentSessionRole::Workload,
),
]
})
.find(|pool_key| connections.pools.contains_key(pool_key))
}
/// Mutate the registry atomically. The callback is deliberately
/// synchronous; asynchronous cleanup must use values returned from it.
pub async fn update_connection_pools<R>(&self, update: impl FnOnce(&mut ConnectionPoolRegistry) -> R) -> R {
crates/dbx-core/src/data/script_generator.rs
+3
View File
@@ -1446,6 +1446,7 @@ mod tests {
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
}
}
@@ -2399,6 +2400,7 @@ mod tests {
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
}),
target: Some(ColumnInfo {
name: "large_val".to_string(),
@@ -2416,6 +2418,7 @@ mod tests {
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
}),
changes: vec!["data_type BIGINT → INT".to_string()],
add_position: None,
crates/dbx-core/src/query/mongo_ops.rs
+14 -6
View File
@@ -232,15 +232,12 @@ pub async fn mongo_server_version_core(
}
}
pub async fn mongo_run_command_core(
state: &AppState,
connection_id: &str,
pub(crate) async fn mongo_run_command_with_existing_pool(
pool: &PoolKind,
database: &str,
command_json: &str,
) -> Result<MongoDocumentResult, String> {
ensure_document_pool(state, connection_id).await?;
let pool = state.pool_handle(connection_id).await.ok_or("Not found")?;
match &pool {
match pool {
PoolKind::MongoDb(client) => mongo_driver::run_command(client, database, command_json).await,
PoolKind::Agent(client) => {
let mut client = client.lock().await;
@@ -261,6 +258,17 @@ pub async fn mongo_run_command_core(
}
}
pub async fn mongo_run_command_core(
state: &AppState,
connection_id: &str,
database: &str,
command_json: &str,
) -> Result<MongoDocumentResult, String> {
ensure_document_pool(state, connection_id).await?;
let pool = state.pool_handle(connection_id).await.ok_or("Not found")?;
mongo_run_command_with_existing_pool(&pool, database, command_json).await
}
pub async fn mongo_show_databases_core(state: &AppState, connection_id: &str) -> Result<MongoDocumentResult, String> {
mongo_run_command_core(state, connection_id, MONGO_SHOW_DATABASES_DATABASE, MONGO_SHOW_DATABASES_COMMAND_JSON).await
}
crates/dbx-core/src/schema/mod.rs
+121 -24
View File
@@ -19,6 +19,9 @@ use std::time::{Duration, Instant};
mod kingbase;
mod mongodb_columns;
pub mod plugin_metadata;
#[cfg(test)]
mod plugin_metadata_tests;
macro_rules! extract_pool {
($pool:expr, $variant:ident) => {
@@ -6859,6 +6862,22 @@ async fn retry_metadata_connection_for_session<T, F, Fut>(
connection_id: &str,
database: Option<&str>,
client_session_id: Option<&str>,
operation: F,
) -> Result<T, String>
where
F: FnMut() -> Fut,
Fut: Future<Output = Result<T, String>>,
{
Box::pin(run_metadata_connection_for_session(state, connection_id, database, client_session_id, true, operation))
.await
}
async fn run_metadata_connection_for_session<T, F, Fut>(
state: &AppState,
connection_id: &str,
database: Option<&str>,
client_session_id: Option<&str>,
allow_recovery: bool,
mut operation: F,
) -> Result<T, String>
where
@@ -6875,6 +6894,9 @@ where
}
None => None,
};
if !allow_recovery {
return operation().await;
}
let mut retried = false;
let mut missing_pool_retry = false;
loop {
@@ -7083,21 +7105,77 @@ async fn get_columns_core_for_session_inner(
table: &str,
client_session_id: Option<&str>,
use_client_session_context: bool,
) -> Result<Vec<db::ColumnInfo>, String> {
get_columns_core_for_session_inner_with_pool(
state,
connection_id,
database,
schema,
table,
client_session_id,
use_client_session_context,
None,
true,
)
.await
}
async fn get_columns_core_for_existing_pool(
state: &AppState,
connection_id: &str,
database: &str,
schema: &str,
table: &str,
pool_key: &str,
) -> Result<Vec<db::ColumnInfo>, String> {
get_columns_core_for_session_inner_with_pool(
state,
connection_id,
database,
schema,
table,
None,
true,
Some(pool_key),
false,
)
.await
}
async fn get_columns_core_for_session_inner_with_pool(
state: &AppState,
connection_id: &str,
database: &str,
schema: &str,
table: &str,
client_session_id: Option<&str>,
use_client_session_context: bool,
existing_pool_key: Option<&str>,
allow_recovery: bool,
) -> Result<Vec<db::ColumnInfo>, String> {
let context_session_id = if use_client_session_context { client_session_id } else { None };
retry_metadata_connection_for_session(state, connection_id, Some(database), client_session_id, || async {
let pool_key = state
.get_or_create_metadata_pool_for_session(connection_id, Some(database), client_session_id)
.await?;
let existing_pool_key = existing_pool_key.map(str::to_owned);
let operation = || async {
let pool_key = if let Some(pool_key) = existing_pool_key.as_deref() {
pool_key.to_string()
} else {
state.get_or_create_metadata_pool_for_session(connection_id, Some(database), client_session_id).await?
};
let db_config = connection_config(state, connection_id).await;
if db_config.as_ref().is_some_and(|config| config.db_type == DatabaseType::MongoDb) {
let pool_handle = state.pool_handle(&pool_key).await.ok_or("Pool not found")?;
return mongodb_columns::get_columns_from_existing_pool(&pool_handle, database, table).await;
}
{
let pool_handle = state.pool_handle(&pool_key).await;
if let Some(PoolKind::ExternalDriver { config, session, .. }) = pool_handle.as_ref() {
let config = config.clone();
let session = session.clone();
if uses_presto_like_information_schema_tables(&config.db_type) {
return external_driver_presto_like_columns(session, config.as_ref(), database, schema, table).await;
if uses_presto_like_information_schema_tables(&config.db_type) {
return external_driver_presto_like_columns(session, config.as_ref(), database, schema, table)
.await;
}
let query_oracle_columns_first =
should_query_oracle_columns_via_sql_first(&config.db_type, context_session_id);
@@ -7168,25 +7246,29 @@ async fn get_columns_core_for_session_inner(
let database = database.to_string();
let schema = schema.to_string();
let table = table.to_string();
return client.list_columns(database, schema, table).await;
return client.list_columns(database, schema, table).await;
}
if let Some(client) = extract_pool!(pool_handle.as_ref(), ClickHouse) {
return db::clickhouse_driver::get_columns(&client, clickhouse_metadata_database(database, schema), table)
.await
.map(deduplicate_column_infos);
return db::clickhouse_driver::get_columns(
&client,
clickhouse_metadata_database(database, schema),
table,
)
.await
.map(deduplicate_column_infos);
}
if let Some(client) = extract_pool!(pool_handle.as_ref(), InfluxDb) {
return db::influxdb_driver::get_columns(&client, database, table).await.map(deduplicate_column_infos);
return db::influxdb_driver::get_columns(&client, database, table).await.map(deduplicate_column_infos);
}
if let Some(client) = extract_pool!(pool_handle.as_ref(), InfluxDb3) {
return db::influxdb3_driver::get_columns(&client, database, table).await.map(deduplicate_column_infos);
return db::influxdb3_driver::get_columns(&client, database, table).await.map(deduplicate_column_infos);
}
if let Some(client) = extract_pool!(pool_handle.as_ref(), VictoriaMetrics) {
return db::victoriametrics_driver::get_columns(&client, table).await.map(deduplicate_column_infos);
return db::victoriametrics_driver::get_columns(&client, table).await.map(deduplicate_column_infos);
}
if let Some(linked) = crate::sql_dialect::parse_sqlserver_linked_schema_ref(schema) {
if let Some(client) = extract_pool!(pool_handle.as_ref(), SqlServer) {
let mut client = lock_sqlserver_metadata_client(&client).await?;
let mut client = lock_sqlserver_metadata_client(&client).await?;
return db::sqlserver::get_linked_server_columns(
&mut client,
&linked.server,
@@ -7201,10 +7283,10 @@ async fn get_columns_core_for_session_inner(
try_sqlserver!(pool_handle, get_columns, schema, table);
if let Some(client) = extract_pool!(pool_handle.as_ref(), Agent) {
let fallback_config = db_config.clone();
let mut client = client.lock().await;
let oracle_sql_config = fallback_config.as_ref().filter(|config| {
should_query_oracle_columns_via_sql_first(&config.db_type, context_session_id)
});
let mut client = client.lock().await;
let oracle_sql_config = fallback_config
.as_ref()
.filter(|config| should_query_oracle_columns_via_sql_first(&config.db_type, context_session_id));
let query_oracle_columns_first = oracle_sql_config.is_some();
if let Some(config) = oracle_sql_config {
match oracle_columns_via_sql(
@@ -7252,7 +7334,9 @@ async fn get_columns_core_for_session_inner(
)
.await
{
Ok(fallback_columns) if !fallback_columns.is_empty() => return Ok(fallback_columns),
Ok(fallback_columns) if !fallback_columns.is_empty() => {
return Ok(fallback_columns)
}
Ok(_) => {}
Err(error) => {
log::warn!(
@@ -7266,6 +7350,8 @@ async fn get_columns_core_for_session_inner(
}
}
}
}
if let Some(config) = fallback_config.as_ref().filter(|_| existing_pool_key.is_none()) {
match native_postgres_metadata_pool(state, connection_id, database, config).await {
Ok(Some(pool)) => {
return db::postgres::get_columns(&pool, schema, table)
@@ -7288,7 +7374,7 @@ async fn get_columns_core_for_session_inner(
return Ok(deduplicate_column_infos(columns));
}
Err(agent_error) => {
if let Some(config) = fallback_config.as_ref() {
if let Some(config) = fallback_config.as_ref().filter(|_| existing_pool_key.is_none()) {
if let Some(pool) =
native_postgres_metadata_pool(state, connection_id, database, config).await?
{
@@ -7354,9 +7440,9 @@ async fn get_columns_core_for_session_inner(
PoolKind::Turso(client) => {
db::turso_driver::get_columns(client, schema, table).await.map(deduplicate_column_infos)
}
PoolKind::CloudflareD1(client) => db::cloudflare_d1_driver::get_columns(client, schema, table)
.await
.map(deduplicate_column_infos),
PoolKind::CloudflareD1(client) => {
db::cloudflare_d1_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)
}
@@ -7371,7 +7457,15 @@ async fn get_columns_core_for_session_inner(
}
_ => Ok(vec![]),
}
})
};
Box::pin(run_metadata_connection_for_session(
state,
connection_id,
Some(database),
client_session_id,
allow_recovery,
operation,
))
.await
}
@@ -7413,6 +7507,9 @@ fn deduplicate_column_infos(columns: Vec<db::ColumnInfo>) -> Vec<db::ColumnInfo>
if existing.data_type.trim().is_empty() && !column.data_type.trim().is_empty() {
existing.data_type = column.data_type;
}
if existing.metadata_capabilities != column.metadata_capabilities {
existing.metadata_capabilities = None;
}
} else {
result.push(column);
}
crates/dbx-core/src/schema/mongodb_columns.rs
+31 -10
View File
@@ -3,28 +3,48 @@ use std::collections::{BTreeMap, BTreeSet};
use mongodb::bson::Bson;
use serde_json::{json, Value};
use crate::connection::AppState;
use crate::connection::{AppState, PoolKind};
use crate::db::mongo_driver::MongoDocumentResult;
use crate::db::ColumnInfo;
const SAMPLE_SIZE: usize = 100;
const MAX_FIELDS: usize = 512;
fn sample_command(collection: &str) -> Value {
json!({
"find": collection,
"filter": {},
"limit": SAMPLE_SIZE,
"batchSize": SAMPLE_SIZE,
"singleBatch": true,
"maxTimeMS": 5000,
})
}
pub(super) async fn get_columns(
state: &AppState,
connection_id: &str,
database: &str,
collection: &str,
) -> Result<Vec<ColumnInfo>, String> {
let command = json!({
"find": collection,
"filter": {},
"limit": SAMPLE_SIZE,
"batchSize": SAMPLE_SIZE,
"singleBatch": true,
"maxTimeMS": 5000,
});
let result = crate::mongo_ops::mongo_run_command_core(state, connection_id, database, &command.to_string()).await?;
let result = crate::mongo_ops::mongo_run_command_core(
state,
connection_id,
database,
&sample_command(collection).to_string(),
)
.await?;
columns_from_sample(&result)
}
pub(super) async fn get_columns_from_existing_pool(
pool: &PoolKind,
database: &str,
collection: &str,
) -> Result<Vec<ColumnInfo>, String> {
let result =
crate::mongo_ops::mongo_run_command_with_existing_pool(pool, database, &sample_command(collection).to_string())
.await?;
columns_from_sample(&result)
}
@@ -68,6 +88,7 @@ fn columns_from_sample(result: &MongoDocumentResult) -> Result<Vec<ColumnInfo>,
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
})
.collect())
}
crates/dbx-core/src/schema/plugin_metadata.rs
+287
View File
@@ -0,0 +1,287 @@
//! Narrow, read-only schema metadata exposed through the plugin Host API.
//!
//! The request is the wire representation of the desktop `PluginTableContext`:
//! it carries table identity only, never a connection config or credential. The
//! implementation deliberately enters the existing schema abstraction instead
//! of issuing driver-specific metadata SQL from the plugin layer.
use serde::{Deserialize, Serialize};
use crate::connection::AppState;
use crate::db;
use crate::types::ColumnMetadataCapabilities;
pub const MAX_PLUGIN_SCHEMA_METADATA_NAME_CHARS: usize = 256;
pub const CONNECTION_NOT_OPEN_ERROR: &str = "Connection is not open";
pub const TABLE_METADATA_SESSION_NOT_OPEN_ERROR: &str = "Connection session is not open for the requested database";
/// The Rust wire DTO intentionally mirrors the public TypeScript
/// `PluginTableContext` field-for-field. It is not a second table identity
/// model; it is the camelCase JSON boundary used by Tauri and Web transports.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginTableContext {
pub connection_id: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub database: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub schema: Option<String>,
pub table: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginTableMetadata {
pub columns: Vec<PluginColumnMetadata>,
/// Reports whether structured optional fields are exposed by the provider
/// path. `unknown` is deliberate when an external provider gives DBX no
/// provenance; the host never guesses from a database type.
pub field_capabilities: PluginMetadataFieldCapabilities,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginColumnMetadata {
pub name: String,
pub data_type: String,
pub nullable: bool,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub length: Option<i32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub precision: Option<i32>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub scale: Option<i32>,
#[serde(rename = "default", default, skip_serializing_if = "Option::is_none")]
pub default_value: Option<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PluginMetadataFieldCapabilities {
pub length: PluginMetadataFieldAvailability,
pub precision: PluginMetadataFieldAvailability,
pub scale: PluginMetadataFieldAvailability,
#[serde(rename = "default")]
pub default_value: PluginMetadataFieldAvailability,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum PluginMetadataFieldAvailability {
Supported,
Unsupported,
Unknown,
}
/// Resolve one table's columns through DBX's existing metadata subsystem.
///
/// The open-connection check is intentionally before `get_columns_core`: that
/// ordinary DBX path is allowed to create metadata pools for the application,
/// but a plugin request must not turn a saved configuration into an implicit
/// connection. A closed request returns without touching the pool registry.
pub async fn get_table_metadata(state: &AppState, context: PluginTableContext) -> Result<PluginTableMetadata, String> {
let context = validate_plugin_table_context(context)?;
let config = {
let configs = state.configs.read().await;
configs.get(&context.connection_id).cloned()
}
.ok_or_else(|| "Connection config not found".to_string())?;
if !state.is_connection_open(&context.connection_id).await {
return Err(CONNECTION_NOT_OPEN_ERROR.to_string());
}
// A table contribution may omit a scope when the selected database does
// not expose one. When the host already knows a saved/default scope, use
// that existing identity; otherwise the schema subsystem receives the
// same empty scope used by its normal metadata path.
let database = context.database.as_deref().or_else(|| config.effective_database()).unwrap_or_default();
let schema = context.schema.as_deref().or(config.default_schema.as_deref()).unwrap_or_default();
let pool_key = state
.existing_metadata_pool_key_for_session(&context.connection_id, Some(database), None)
.await
.ok_or_else(|| TABLE_METADATA_SESSION_NOT_OPEN_ERROR.to_string())?;
let columns = super::get_columns_core_for_existing_pool(
state,
&context.connection_id,
database,
schema,
&context.table,
&pool_key,
)
.await?;
if columns.is_empty() {
return Err("Table metadata provider returned no columns".to_string());
}
let field_capabilities = field_capabilities(&columns);
Ok(PluginTableMetadata { columns: columns.into_iter().map(map_column).collect(), field_capabilities })
}
fn validate_plugin_table_context(mut context: PluginTableContext) -> Result<PluginTableContext, String> {
context.connection_id = required_identifier(context.connection_id, "connectionId")?;
context.table = required_identifier(context.table, "table")?;
context.database = optional_identifier(context.database, "database")?;
context.schema = optional_identifier(context.schema, "schema")?;
Ok(context)
}
fn required_identifier(value: String, name: &str) -> Result<String, String> {
let value = value.trim();
if value.is_empty() {
return Err(format!("{name} must not be empty"));
}
if value.chars().count() > MAX_PLUGIN_SCHEMA_METADATA_NAME_CHARS {
return Err(format!("{name} must be at most {MAX_PLUGIN_SCHEMA_METADATA_NAME_CHARS} characters"));
}
Ok(value.to_string())
}
fn optional_identifier(value: Option<String>, name: &str) -> Result<Option<String>, String> {
let Some(value) = value else {
return Ok(None);
};
let value = value.trim();
if value.is_empty() {
return Ok(None);
}
if value.chars().count() > MAX_PLUGIN_SCHEMA_METADATA_NAME_CHARS {
return Err(format!("{name} must be at most {MAX_PLUGIN_SCHEMA_METADATA_NAME_CHARS} characters"));
}
Ok(Some(value.to_string()))
}
fn map_column(column: db::ColumnInfo) -> PluginColumnMetadata {
PluginColumnMetadata {
name: column.name,
data_type: column.data_type,
nullable: column.is_nullable,
length: column.character_maximum_length,
precision: column.numeric_precision,
scale: column.numeric_scale,
default_value: column.column_default,
}
}
fn field_capabilities(columns: &[db::ColumnInfo]) -> PluginMetadataFieldCapabilities {
PluginMetadataFieldCapabilities {
length: field_capability(columns, |capabilities| capabilities.length),
precision: field_capability(columns, |capabilities| capabilities.precision),
scale: field_capability(columns, |capabilities| capabilities.scale),
default_value: field_capability(columns, |capabilities| capabilities.default),
}
}
fn field_capability(
columns: &[db::ColumnInfo],
field: impl Fn(ColumnMetadataCapabilities) -> bool,
) -> PluginMetadataFieldAvailability {
let Some(first) = columns.first().and_then(|column| column.metadata_capabilities) else {
return PluginMetadataFieldAvailability::Unknown;
};
if columns.iter().all(|column| column.metadata_capabilities.is_some_and(|capabilities| capabilities == first)) {
if field(first) {
PluginMetadataFieldAvailability::Supported
} else {
PluginMetadataFieldAvailability::Unsupported
}
} else {
PluginMetadataFieldAvailability::Unknown
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::storage::Storage;
fn config() -> crate::models::connection::ConnectionConfig {
serde_json::from_value(serde_json::json!({
"id": "conn-1",
"name": "Connection",
"db_type": "postgres",
"host": "127.0.0.1",
"port": 5432,
"username": "user",
"password": "secret"
}))
.unwrap()
}
fn column(name: &str) -> db::ColumnInfo {
db::ColumnInfo {
name: name.to_string(),
data_type: "character varying".to_string(),
is_nullable: true,
column_default: None,
numeric_precision: None,
numeric_scale: None,
character_maximum_length: Some(64),
metadata_capabilities: Some(ColumnMetadataCapabilities::all_supported()),
..Default::default()
}
}
#[test]
fn maps_column_info_to_the_narrow_public_contract() {
let metadata = PluginTableMetadata {
columns: vec![map_column(column("name"))],
field_capabilities: field_capabilities(&[column("name")]),
};
let value = serde_json::to_value(metadata).unwrap();
assert_eq!(value["columns"][0]["name"], "name");
assert_eq!(value["columns"][0]["dataType"], "character varying");
assert_eq!(value["columns"][0]["nullable"], true);
assert_eq!(value["columns"][0]["length"], 64);
assert!(value["columns"][0].get("default").is_none());
assert!(value["columns"][0].get("comment").is_none());
assert_eq!(value["fieldCapabilities"]["length"], "supported");
}
#[test]
fn does_not_guess_capabilities_without_provider_provenance() {
let column = db::ColumnInfo { name: "id".to_string(), data_type: "integer".to_string(), ..Default::default() };
let capabilities = field_capabilities(&[column]);
assert_eq!(capabilities.length, PluginMetadataFieldAvailability::Unknown);
assert_eq!(capabilities.precision, PluginMetadataFieldAvailability::Unknown);
}
#[test]
fn preserves_unsupported_structured_fields_without_fabricating_values() {
let mut column_info = column("id");
column_info.character_maximum_length = None;
column_info.numeric_precision = None;
column_info.numeric_scale = None;
column_info.metadata_capabilities = Some(ColumnMetadataCapabilities::default_only());
let metadata = PluginTableMetadata {
columns: vec![map_column(column_info)],
field_capabilities: field_capabilities(&[db::ColumnInfo {
metadata_capabilities: Some(ColumnMetadataCapabilities::default_only()),
..column("id")
}]),
};
let value = serde_json::to_value(metadata).unwrap();
assert!(value["columns"][0].get("length").is_none());
assert_eq!(value["fieldCapabilities"]["length"], "unsupported");
assert_eq!(value["fieldCapabilities"]["default"], "supported");
}
#[tokio::test]
async fn saved_but_disconnected_context_is_rejected_without_creating_a_pool() {
let dir = tempfile::tempdir().unwrap();
let storage = Storage::open(&dir.path().join("storage.db")).await.unwrap();
let state = AppState::new_with_plugin_dir(storage, dir.path().join("plugins"));
let config = config();
state.configs.write().await.insert(config.id.clone(), config);
let context = PluginTableContext {
connection_id: "conn-1".to_string(),
database: Some("app".to_string()),
schema: Some("public".to_string()),
table: "users".to_string(),
};
let error = get_table_metadata(&state, context).await.unwrap_err();
assert_eq!(error, CONNECTION_NOT_OPEN_ERROR);
assert!(state.with_connection_pools(|pools| pools.is_empty()).await);
}
}
crates/dbx-core/src/schema/plugin_metadata_tests.rs
+276
View File
@@ -0,0 +1,276 @@
use super::plugin_metadata::{get_table_metadata, PluginTableContext, TABLE_METADATA_SESSION_NOT_OPEN_ERROR};
use super::{
get_columns_core_for_existing_pool, get_columns_core_for_session_inner, run_metadata_connection_for_session,
};
use crate::connection::{AppState, PoolKind, METADATA_POOL_ACQUIRE_TIMEOUT};
use crate::db::agent_driver::{AgentDriverClient, AgentLaunchSpec, AgentRuntimeClient};
use crate::models::connection::{ConnectionConfig, DatabaseType};
use crate::storage::Storage;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
async fn state(db_type: DatabaseType) -> (tempfile::TempDir, AppState) {
let dir = tempfile::tempdir().unwrap();
let storage = Storage::open(&dir.path().join("storage.db")).await.unwrap();
let state = AppState::new_with_plugin_and_agent_dir_and_app_version(
storage,
dir.path().join("plugins"),
dir.path().join("agents"),
"test",
);
let config: ConnectionConfig = serde_json::from_value(serde_json::json!({
"id": "conn",
"name": "Metadata test",
"db_type": db_type,
"host": "127.0.0.1",
"port": 5432,
"username": "user",
"password": "secret",
"database": "app",
"connect_timeout_secs": 1
}))
.unwrap();
state.configs.write().await.insert(config.id.clone(), config);
(dir, state)
}
fn context() -> PluginTableContext {
PluginTableContext {
connection_id: "conn".to_string(),
database: Some("app".to_string()),
schema: Some("public".to_string()),
table: "users".to_string(),
}
}
struct ScriptedAgent(Arc<AgentRuntimeClient>);
impl Drop for ScriptedAgent {
fn drop(&mut self) {
self.0.kill();
}
}
async fn scripted_agent(dir: &std::path::Path, response: serde_json::Value) -> ScriptedAgent {
let script = dir.join("metadata-agent.py");
std::fs::write(
&script,
r#"import json, pathlib, sys
response = json.loads(sys.argv[1])
calls = pathlib.Path(sys.argv[2])
print(json.dumps({'ready': True}), flush=True)
for line in sys.stdin:
request = json.loads(line)
method = request['method']
if method == 'handshake':
body = {'result': {'protocolVersion': 2, 'agentProtocolVersion': 2, 'capabilities': ['multi_session']}}
elif method == 'get_columns':
with calls.open('a') as output:
output.write(method + '\n')
body = response
else:
body = {'result': {}}
print(json.dumps({'jsonrpc': '2.0', 'id': request['id'], **body}), flush=True)
"#,
)
.unwrap();
let python = if cfg!(windows) { "python" } else { "python3" };
let runtime = AgentRuntimeClient::spawn(
AgentLaunchSpec::new(python).with_args([
script.to_string_lossy().to_string(),
response.to_string(),
dir.join("calls").to_string_lossy().to_string(),
]),
"test",
)
.await
.unwrap();
ScriptedAgent(runtime)
}
#[tokio::test]
async fn existing_pool_never_connects_on_empty_or_error_but_default_app_keeps_fallback() {
for db_type in [DatabaseType::Highgo, DatabaseType::Vastbase] {
for provider_error in [false, true] {
let (dir, state) = state(db_type).await;
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
state.configs.write().await.get_mut("conn").unwrap().port = listener.local_addr().unwrap().port();
let response = if provider_error {
serde_json::json!({"error": {"code": -1, "message": "metadata denied", "data": {
"category": "sql", "sessionDisposition": "keep"
}}})
} else {
serde_json::json!({"result": []})
};
let agent = scripted_agent(dir.path(), response).await;
agent.0.increment_session_count();
let client = AgentDriverClient::shared_session(agent.0.clone(), "metadata-session".to_string());
let pool_key = "conn:app:role:metadata";
state.update_connection_pools(|pools| pools.insert(pool_key.to_string(), PoolKind::agent(client))).await;
let error = tokio::time::timeout(Duration::from_secs(3), async {
tokio::select! {
biased;
_ = listener.accept() => panic!("plugin metadata attempted a native connection"),
result = get_table_metadata(&state, context()) => result.unwrap_err(),
}
})
.await
.expect("plugin metadata must not wait for a native fallback");
if provider_error {
assert!(error.contains("metadata denied"), "{error}");
assert_eq!(
crate::db::agent_driver::try_agent_error_from_legacy(&error).and_then(|error| error.category()),
Some(crate::db::agent_driver::AgentErrorCategory::Sql)
);
} else {
assert_eq!(error, "Table metadata provider returned no columns");
}
assert_eq!(std::fs::read_to_string(dir.path().join("calls")).unwrap().lines().count(), 1);
assert!(state.pool_handle(pool_key).await.is_some());
assert_eq!(state.with_connection_pools(|pools| pools.len()).await, 1);
tokio::time::timeout(Duration::from_secs(3), async {
tokio::select! {
accepted = listener.accept() => { accepted.unwrap(); }
result = get_columns_core_for_session_inner(&state, "conn", "app", "public", "users", None, true) => {
panic!("ordinary app skipped its native fallback: {result:?}");
}
}
})
.await
.expect("ordinary app fallback must still connect");
assert_eq!(std::fs::read_to_string(dir.path().join("calls")).unwrap().lines().count(), 2);
agent.0.kill();
let stale_error = tokio::time::timeout(Duration::from_secs(3), async {
tokio::select! {
biased;
_ = listener.accept() => panic!("stale plugin pool triggered a native connection"),
result = get_table_metadata(&state, context()) => result.unwrap_err(),
}
})
.await
.expect("cancelled metadata must release the agent lock");
assert!(stale_error.contains("unavailable"), "{stale_error}");
assert!(state.pool_handle(pool_key).await.is_some());
assert_eq!(state.with_connection_pools(|pools| pools.len()).await, 1);
}
}
}
#[tokio::test]
async fn existing_pool_shares_app_concurrency_limit_and_queue_timeout() {
for db_type in [DatabaseType::Mysql, DatabaseType::Postgres] {
let (_dir, state) = state(db_type).await;
tokio::time::pause();
let entered = AtomicUsize::new(0);
let mut operations = Vec::new();
for index in 0..6 {
let operation = run_metadata_connection_for_session(&state, "conn", Some("app"), None, index < 3, || {
entered.fetch_add(1, Ordering::SeqCst);
std::future::pending::<Result<(), String>>()
});
let mut operation = Box::pin(operation);
assert!(futures::poll!(operation.as_mut()).is_pending());
operations.push(operation);
}
assert_eq!(entered.load(Ordering::SeqCst), 6);
let mut queued =
Box::pin(run_metadata_connection_for_session(&state, "conn", Some("app"), None, false, || {
entered.fetch_add(1, Ordering::SeqCst);
async { Ok(()) }
}));
assert!(futures::poll!(queued.as_mut()).is_pending());
tokio::time::advance(METADATA_POOL_ACQUIRE_TIMEOUT).await;
assert_eq!(queued.await.unwrap_err(), crate::query::METADATA_POOL_BUSY_ERROR);
assert_eq!(entered.load(Ordering::SeqCst), 6);
drop(operations);
for _ in 0..6 {
run_metadata_connection_for_session(&state, "conn", Some("app"), None, false, || async { Ok(()) })
.await
.unwrap();
}
tokio::time::resume();
}
}
#[tokio::test]
async fn existing_pool_entrypoint_waits_for_the_shared_budget_before_provider_lookup() {
let (_dir, state) = state(DatabaseType::Postgres).await;
let mut held = Vec::new();
for _ in 0..6 {
held.push(state.acquire_metadata_permit("conn", Some("app"), DatabaseType::Postgres, None).await.unwrap());
}
let mut lookup = Box::pin(get_columns_core_for_existing_pool(&state, "conn", "app", "public", "users", "conn:app"));
assert!(futures::poll!(lookup.as_mut()).is_pending());
drop(held.pop());
assert_eq!(lookup.await.unwrap_err(), "Pool not found");
let mut next = Box::pin(state.acquire_metadata_permit("conn", Some("app"), DatabaseType::Postgres, None));
assert!(matches!(futures::poll!(next.as_mut()), std::task::Poll::Ready(Ok(_))));
}
#[tokio::test]
async fn existing_pool_releases_permits_on_success_error_and_cancellation_without_retry() {
let (_dir, state) = state(DatabaseType::Postgres).await;
let mut held = Vec::new();
for _ in 0..5 {
held.push(state.acquire_metadata_permit("conn", Some("app"), DatabaseType::Postgres, None).await.unwrap());
}
for result in [Ok(()), Err("Pool not found"), Err("connection reset by peer"), Err("Query canceled")] {
let mut attempts = 0;
let actual = run_metadata_connection_for_session(&state, "conn", Some("app"), None, false, || {
attempts += 1;
async { result.map_err(str::to_string) }
})
.await;
assert_eq!(actual, result.map_err(str::to_string));
assert_eq!(attempts, 1);
}
let mut running = Box::pin(run_metadata_connection_for_session(&state, "conn", Some("app"), None, false, || {
std::future::pending::<Result<(), String>>()
}));
assert!(futures::poll!(running.as_mut()).is_pending());
let queued_calls = AtomicUsize::new(0);
let mut queued = Box::pin(run_metadata_connection_for_session(&state, "conn", Some("app"), None, false, || {
queued_calls.fetch_add(1, Ordering::SeqCst);
async { Ok(()) }
}));
assert!(futures::poll!(queued.as_mut()).is_pending());
drop(queued);
drop(running);
assert_eq!(queued_calls.load(Ordering::SeqCst), 0);
let mut next = Box::pin(state.acquire_metadata_permit("conn", Some("app"), DatabaseType::Postgres, None));
assert!(matches!(futures::poll!(next.as_mut()), std::task::Poll::Ready(Ok(_))));
}
#[tokio::test]
async fn invalid_scope_and_removed_pool_do_not_create_or_reconnect() {
let (_dir, state) = state(DatabaseType::Postgres).await;
let mut invalid = context();
invalid.table = " ".to_string();
assert_eq!(get_table_metadata(&state, invalid).await.unwrap_err(), "table must not be empty");
let mut unknown = context();
unknown.connection_id = "unknown".to_string();
assert_eq!(get_table_metadata(&state, unknown).await.unwrap_err(), "Connection config not found");
state
.update_connection_pools(|pools| {
pools.insert("conn:other".to_string(), PoolKind::agent(AgentDriverClient::test_stub()));
})
.await;
assert_eq!(get_table_metadata(&state, context()).await.unwrap_err(), TABLE_METADATA_SESSION_NOT_OPEN_ERROR);
state
.update_connection_pools(|pools| {
pools.remove("conn:other");
pools.insert("conn:app".to_string(), PoolKind::agent(AgentDriverClient::test_stub()));
})
.await;
let key = state.existing_metadata_pool_key_for_session("conn", Some("app"), None).await.unwrap();
state.update_connection_pools(|pools| pools.remove(&key)).await;
assert_eq!(
get_columns_core_for_existing_pool(&state, "conn", "app", "public", "users", &key).await.unwrap_err(),
"Pool not found"
);
assert!(state.with_connection_pools(|pools| pools.is_empty()).await);
}
crates/dbx-core/tests/api_contract_verification.rs
+2
View File
@@ -293,6 +293,7 @@ fn column_info_serialization_roundtrip() {
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
};
let json = serde_json::to_value(&col).unwrap();
assert_eq!(json.get("is_unique"), Some(&serde_json::json!(true)));
@@ -340,6 +341,7 @@ fn table_columns_result_serialization_contract() {
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
}],
error: Some("partial".to_string()),
};
crates/dbx-core/tests/bidirectional_diff_e2e.rs
+1
View File
@@ -27,6 +27,7 @@ fn col(name: &str, data_type: &str) -> ColumnInfo {
numeric_precision: None,
numeric_scale: None,
character_maximum_length: None,
metadata_capabilities: None,
enum_values: None,
character_set: None,
collation: None,
crates/dbx-core/tests/cross_dialect_integration.rs
+1
View File
@@ -30,6 +30,7 @@ fn col(name: &str, data_type: &str) -> ColumnInfo {
numeric_precision: None,
numeric_scale: None,
character_maximum_length: None,
metadata_capabilities: None,
enum_values: None,
character_set: None,
collation: None,
crates/dbx-core/tests/performance_benchmarks.rs
+1
View File
@@ -48,6 +48,7 @@ fn generate_details(tables: &[TableInfo], columns_per_table: usize) -> Vec<Table
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
})
.collect(),
indexes: vec![],
crates/dbx-core/tests/regression_compatibility.rs
+1
View File
@@ -36,6 +36,7 @@ fn col(name: &str, data_type: &str) -> ColumnInfo {
numeric_precision: None,
numeric_scale: None,
character_maximum_length: None,
metadata_capabilities: None,
enum_values: None,
character_set: None,
collation: None,
crates/dbx-core/tests/schema_diff_table_mapping.rs
+1
View File
@@ -29,6 +29,7 @@ fn column(name: &str, data_type: &str) -> ColumnInfo {
numeric_precision: None,
numeric_scale: None,
character_maximum_length: None,
metadata_capabilities: None,
enum_values: None,
character_set: None,
collation: None,
crates/dbx-drivers/src/db/meilisearch_driver.rs
+1
View File
@@ -395,6 +395,7 @@ pub async fn get_columns(client: &MeilisearchClient, index: &str) -> Result<Vec<
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
})
.collect())
}
crates/dbx-drivers/src/db/mysql.rs
+6 -4
View File
@@ -24,10 +24,10 @@ use crate::models::connection::{
};
use crate::sql::{starts_with_executable_sql_keyword, starts_with_executable_sql_keyword_for_database};
use crate::types::{
ColumnInfo, CompletionAssistantCandidate, CompletionAssistantCandidateKind, CompletionAssistantMatchMode,
CompletionAssistantObjectKind, CompletionAssistantRequest, CompletionAssistantResponse, DatabaseInfo,
ForeignKeyInfo, IndexInfo, LargeValueCell, ObjectInfo, ObjectStatistics, QueryMessage, QueryResult,
SpatialColumnBuilder, TableInfo, TriggerInfo,
ColumnInfo, ColumnMetadataCapabilities, CompletionAssistantCandidate, CompletionAssistantCandidateKind,
CompletionAssistantMatchMode, CompletionAssistantObjectKind, CompletionAssistantRequest,
CompletionAssistantResponse, DatabaseInfo, ForeignKeyInfo, IndexInfo, LargeValueCell, ObjectInfo, ObjectStatistics,
QueryMessage, QueryResult, SpatialColumnBuilder, TableInfo, TriggerInfo,
};
use dbx_types::metadata_filter::{table_name_filter_matches, TableNameFilter};
@@ -4238,6 +4238,7 @@ where
enum_values,
character_set: get_opt_str(row, "CHARACTER_SET_NAME").filter(|s| !s.is_empty()),
collation: get_opt_str(row, "COLLATION_NAME").filter(|s| !s.is_empty()),
metadata_capabilities: Some(ColumnMetadataCapabilities::all_supported()),
})
})
.collect();
@@ -4294,6 +4295,7 @@ where
.and_then(|c| c.split_once('_').map(|(charset, _)| charset.to_string()))
.filter(|s| !s.is_empty()),
collation,
metadata_capabilities: Some(ColumnMetadataCapabilities::default_only()),
})
})
.collect();
crates/dbx-drivers/src/db/mysql_compatible.rs
+1
View File
@@ -218,6 +218,7 @@ pub async fn get_columns_show_from(
.and_then(|c| c.split_once('_').map(|(charset, _)| charset.to_string()))
.filter(|s| !s.is_empty()),
collation,
metadata_capabilities: None,
})
})
.collect())
crates/dbx-drivers/src/db/postgres.rs
+9 -6
View File
@@ -33,12 +33,13 @@ use crate::execution::{await_stream_with_progress_timeout, DbOperationBudget, St
use crate::models::connection::DatabaseType;
use crate::sql::starts_with_executable_sql_keyword;
use crate::types::{
ColumnInfo, CompletionAssistantCandidate, CompletionAssistantCandidateKind, CompletionAssistantMatchMode,
CompletionAssistantObjectKind, CompletionAssistantRequest, CompletionAssistantResponse, ConstraintInfo,
CustomTypeDdl, CustomTypeDetails, CustomTypeDomainConstraint, CustomTypeKind, CustomTypeMember,
CustomTypeProperties, DatabaseInfo, DatabaseStorageInfo, ExtensionInfo, ForeignKeyInfo, FunctionInfo, IndexInfo,
ObjectInfo, ObjectStatistics, OwnerInfo, PgPartitionBound, PgPartitionKind, PgPartitionNode, PgTablePartitioning,
QueryMessage, QueryResult, RuleInfo, SchemaInfo, SequenceInfo, SpatialColumnBuilder, TableInfo, TriggerInfo,
ColumnInfo, ColumnMetadataCapabilities, CompletionAssistantCandidate, CompletionAssistantCandidateKind,
CompletionAssistantMatchMode, CompletionAssistantObjectKind, CompletionAssistantRequest,
CompletionAssistantResponse, ConstraintInfo, CustomTypeDdl, CustomTypeDetails, CustomTypeDomainConstraint,
CustomTypeKind, CustomTypeMember, CustomTypeProperties, DatabaseInfo, DatabaseStorageInfo, ExtensionInfo,
ForeignKeyInfo, FunctionInfo, IndexInfo, ObjectInfo, ObjectStatistics, OwnerInfo, PgPartitionBound,
PgPartitionKind, PgPartitionNode, PgTablePartitioning, QueryMessage, QueryResult, RuleInfo, SchemaInfo,
SequenceInfo, SpatialColumnBuilder, TableInfo, TriggerInfo,
};
pub const GAUSSDB_COMPATIBILITY_SQL: &str =
@@ -4711,6 +4712,7 @@ fn column_info_from_row_offset(row: &Row, offset: usize) -> ColumnInfo {
numeric_scale: row.try_get::<_, Option<i32>>(offset + 8).ok().flatten(),
character_maximum_length: row.try_get::<_, Option<i32>>(offset + 9).ok().flatten(),
enum_values: parse_enum_values_from_row(row, offset + 10),
metadata_capabilities: Some(ColumnMetadataCapabilities::all_supported()),
..Default::default()
}
}
@@ -7622,6 +7624,7 @@ fn redshift_columns_from_query_result(result: QueryResult) -> Vec<ColumnInfo> {
numeric_scale: query_result_i32(&row, 5),
character_maximum_length: query_result_i32(&row, 6),
enum_values: None,
metadata_capabilities: Some(ColumnMetadataCapabilities::all_supported()),
..Default::default()
})
})
crates/dbx-drivers/src/db/sqlite.rs
+5 -3
View File
@@ -18,9 +18,9 @@ extern crate libsqlite3_hotbundle;
use super::file_validator::validate_file_path;
use crate::sql::starts_with_executable_sql_keyword;
use crate::types::{
ColumnInfo, CompletionAssistantCandidate, CompletionAssistantCandidateKind, CompletionAssistantMatchMode,
CompletionAssistantObjectKind, CompletionAssistantRequest, CompletionAssistantResponse, DatabaseInfo,
ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo,
ColumnInfo, ColumnMetadataCapabilities, CompletionAssistantCandidate, CompletionAssistantCandidateKind,
CompletionAssistantMatchMode, CompletionAssistantObjectKind, CompletionAssistantRequest,
CompletionAssistantResponse, DatabaseInfo, ForeignKeyInfo, IndexInfo, QueryResult, TableInfo, TriggerInfo,
};
const SQLITE_DATABASE_HEADER: &[u8; 16] = b"SQLite format 3\0";
@@ -1588,6 +1588,7 @@ pub async fn get_columns(pool: &SqliteHandle, schema: &str, table: &str) -> Resu
numeric_scale: None,
character_maximum_length: None,
enum_values: None,
metadata_capabilities: Some(ColumnMetadataCapabilities::default_only()),
..Default::default()
}
})
@@ -1623,6 +1624,7 @@ pub async fn get_columns(pool: &SqliteHandle, schema: &str, table: &str) -> Resu
numeric_scale: None,
character_maximum_length: None,
enum_values: None,
metadata_capabilities: Some(ColumnMetadataCapabilities::default_only()),
..Default::default()
})
})
crates/dbx-mcp/src/backend.rs
+1
View File
@@ -2570,6 +2570,7 @@ fn infer_document_columns(documents: &[Value]) -> Vec<ColumnInfo> {
enum_values: None,
character_set: None,
collation: None,
metadata_capabilities: None,
})
.collect()
}
crates/dbx-plugin-runtime/src/plugins/manifest.rs
+35 -18
View File
@@ -9,11 +9,12 @@ pub const SUPPORTED_PLUGIN_MANIFEST_VERSION: u32 = 1;
///
/// 1.1 adds the plugin-initiated `host/requestUserInput` method (see
/// `plugins/runtime.rs`). 1.2 adds the plugin-initiated plan Host API
/// (`host.getPlanCapabilities` / `host.explainPlan`). Both are additive: older
/// plugins keep working, and a plugin that wants either capability must check
/// the advertised version (or the matching `capabilities` / `host.features`
/// (`host.getPlanCapabilities` / `host.explainPlan`). 1.3 adds read-only table
/// schema metadata (`host.getTableMetadata`). All are additive: older plugins
/// keep working, and a plugin that wants either capability must check the
/// advertised version (or the matching `capabilities` / `host.features`
/// entry) before calling it.
pub const SUPPORTED_PLUGIN_HOST_API_VERSION: &str = "1.2.0";
pub const SUPPORTED_PLUGIN_HOST_API_VERSION: &str = "1.3.0";
/// Capabilities the host advertises to a plugin backend at `plugin/initialize`.
pub const SUPPORTED_PLUGIN_HOST_FEATURES: &[&str] = &["host.requestUserInput"];
pub const SUPPORTED_PLUGIN_PROTOCOL_VERSION: u32 = 1;
@@ -21,8 +22,16 @@ pub const PLUGIN_CONNECTION_TEST_METHOD: &str = "connection/test";
pub const PLUGIN_CONNECTION_CONNECT_METHOD: &str = "connection/connect";
pub const PLUGIN_CONNECTION_DISCONNECT_METHOD: &str = "connection/disconnect";
pub const PLUGIN_CONNECTION_ACTION_METHOD: &str = "connection/action";
pub const SUPPORTED_PLUGIN_PERMISSIONS: &[&str] =
&["host.events", "host.binary", "host.workbench", "host.filesystem", "host.plans:read", "host.storage", "host.ai"];
pub const SUPPORTED_PLUGIN_PERMISSIONS: &[&str] = &[
"host.events",
"host.binary",
"host.workbench",
"host.filesystem",
"host.plans:read",
"host.schema:read",
"host.storage",
"host.ai",
];
/// Cap the number of `host.network:<origin>` entries so a manifest cannot bloat
/// the sandbox CSP or enumerate large origin lists.
@@ -1572,7 +1581,7 @@ mod tests {
}
#[test]
fn accepts_host_plans_read_and_still_rejects_unknown_permissions() {
fn accepts_read_only_host_permissions_and_still_rejects_unknown_permissions() {
let dir = tempfile::tempdir().unwrap();
std::fs::create_dir_all(dir.path().join("ui")).unwrap();
std::fs::write(dir.path().join("ui").join("index.html"), "<!doctype html>").unwrap();
@@ -1585,14 +1594,22 @@ mod tests {
"publisher": "example",
"engines": { "dbx": ">=0.1.0", "host_api": "^1.0" },
"entrypoints": { "ui": { "root": "ui", "entry": "ui/index.html" } },
"permissions": ["host.plans:read"]
"permissions": ["host.plans:read", "host.schema:read"]
}))
.unwrap();
let compatibility = manifest.compatibility(dir.path(), "0.1.0");
assert!(compatibility.compatible, "{:?}", compatibility.errors);
// The plan API is read-only: an execute-scoped scope must not be declared.
for permission in ["host.plans:execute", "host.plans", "host.plans:read:all", "host.plan:read"] {
// Both the plan and schema metadata APIs are read-only; execute/write
// scopes must not be declared as substitutes.
for permission in [
"host.plans:execute",
"host.plans",
"host.plans:read:all",
"host.plan:read",
"host.schema:write",
"host.schema",
] {
let manifest: PluginManifest = serde_json::from_value(serde_json::json!({
"manifest_version": 1,
"id": "io.dbx.example",
@@ -1635,22 +1652,22 @@ mod tests {
assert_eq!(declared, SUPPORTED_PLUGIN_PERMISSIONS.iter().map(|value| value.to_string()).collect::<Vec<_>>());
}
/// The reason for the 1.2.0 bump: `engines.host_api` is how a plugin states
/// "I need the plan API", so the advertised version has to satisfy `^1.2`
/// while a floor this host cannot meet stays rejected.
/// The reason for the 1.3.0 bump: `engines.host_api` is how a plugin states
/// "I need the schema metadata API", so the advertised version has to
/// satisfy `^1.3` while a floor this host cannot meet stays rejected.
#[test]
fn host_api_advertises_the_floor_a_plan_api_plugin_declares() {
fn host_api_advertises_the_floor_a_schema_metadata_plugin_declares() {
let advertised = semver::Version::parse(SUPPORTED_PLUGIN_HOST_API_VERSION)
.expect("the advertised Host API version must be semver");
assert!(
semver::VersionReq::parse("^1.2").unwrap().matches(&advertised),
"the host must satisfy the plan API floor it asks plugins to declare"
semver::VersionReq::parse("^1.3").unwrap().matches(&advertised),
"the host must satisfy the schema metadata API floor it asks plugins to declare"
);
for requirement in ["^1.0", "^1.1", "^1.2", ">=1.1.0, <2.0.0"] {
for requirement in ["^1.0", "^1.1", "^1.2", "^1.3", ">=1.1.0, <2.0.0"] {
assert!(host_api_requirement_errors(requirement).is_empty(), "{requirement} must be satisfiable");
}
for requirement in [">=1.3.0", "^2.0"] {
for requirement in [">=1.4.0", "^2.0"] {
assert!(!host_api_requirement_errors(requirement).is_empty(), "{requirement} must be rejected");
}
}
crates/dbx-sql/src/schema_diff.rs
+4
View File
@@ -6374,6 +6374,7 @@ mod tests {
numeric_precision: None,
numeric_scale: None,
character_maximum_length: None,
metadata_capabilities: None,
enum_values: None,
character_set: None,
collation: None,
@@ -9214,6 +9215,7 @@ mod tests {
numeric_precision: None,
numeric_scale: None,
character_maximum_length: None,
metadata_capabilities: None,
enum_values: None,
character_set: None,
collation: None,
@@ -12410,6 +12412,7 @@ mod tests {
numeric_precision: None,
numeric_scale: None,
character_maximum_length: None,
metadata_capabilities: None,
enum_values: None,
character_set: None,
collation: None,
@@ -12427,6 +12430,7 @@ mod tests {
numeric_precision: None,
numeric_scale: None,
character_maximum_length: None,
metadata_capabilities: None,
enum_values: None,
character_set: None,
collation: None,
crates/dbx-types/src/types.rs
+22
View File
@@ -181,6 +181,26 @@ pub struct ObjectSource {
pub editable: Option<bool>,
}
/// Provenance for structured metadata fields that are optional in [`ColumnInfo`].
/// This stays internal to the metadata mapping path and is not serialized.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ColumnMetadataCapabilities {
pub default: bool,
pub length: bool,
pub precision: bool,
pub scale: bool,
}
impl ColumnMetadataCapabilities {
pub const fn all_supported() -> Self {
Self { default: true, length: true, precision: true, scale: true }
}
pub const fn default_only() -> Self {
Self { default: true, length: false, precision: false, scale: false }
}
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct ColumnInfo {
pub name: String,
@@ -203,6 +223,8 @@ pub struct ColumnInfo {
pub character_set: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub collation: Option<String>,
#[serde(skip)]
pub metadata_capabilities: Option<ColumnMetadataCapabilities>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
crates/dbx-web/src/main.rs
+1
View File
@@ -566,6 +566,7 @@ async fn serve() {
.route("/schema/event-info", get(routes::schema::get_event_info))
.route("/schema/custom-type-details", get(routes::schema::get_custom_type_details))
.route("/schema/columns", get(routes::schema::list_columns))
.route("/plugin/table-metadata", post(routes::schema::get_plugin_table_metadata))
.route("/schema/all-columns", get(routes::schema::get_all_columns))
.route("/schema/data-types", get(routes::schema::list_data_types))
.route("/schema/indexes", get(routes::schema::list_indexes))
crates/dbx-web/src/routes/schema.rs
+9
View File
@@ -462,6 +462,15 @@ fn should_cache_columns(client_session_id: Option<&str>) -> bool {
client_session_id.is_none()
}
pub async fn get_plugin_table_metadata(
State(state): State<Arc<WebState>>,
Json(request): Json<dbx_core::schema::plugin_metadata::PluginTableContext>,
) -> Result<Json<dbx_core::schema::plugin_metadata::PluginTableMetadata>, AppError> {
let result =
dbx_core::schema::plugin_metadata::get_table_metadata(&state.app, request).await.map_err(AppError::from)?;
Ok(Json(result))
}
pub async fn list_columns(
State(state): State<Arc<WebState>>,
Query(q): Query<SchemaQuery>,