From 905c5a622be5adeb13761991965db957fed2190e Mon Sep 17 00:00:00 2001 From: Priec Date: Sat, 12 Sep 2026 22:10:22 +0200 Subject: [PATCH] fixing problems --- client | 2 +- client-gui2 | 2 +- common/src/catalog_storage.rs | 211 ++++++++++++++++++++++++++++++++++ common/src/lib.rs | 1 + search/src/lib.rs | 48 +++----- server | 2 +- 6 files changed, 228 insertions(+), 38 deletions(-) create mode 100644 common/src/catalog_storage.rs diff --git a/client b/client index 56cc91df..3f814d87 160000 --- a/client +++ b/client @@ -1 +1 @@ -Subproject commit 56cc91dfe62e6f62748123063ccb93344f6cb5a4 +Subproject commit 3f814d87864e47bb3c53425bfa3a10bc07c81305 diff --git a/client-gui2 b/client-gui2 index 7e56fbea..4e73919b 160000 --- a/client-gui2 +++ b/client-gui2 @@ -1 +1 @@ -Subproject commit 7e56fbead71f70874ce3f0a9ca41024850288004 +Subproject commit 4e73919b20c716496ff98ccc5bb1357bb65fa32f diff --git a/common/src/catalog_storage.rs b/common/src/catalog_storage.rs new file mode 100644 index 00000000..9e2c19ed --- /dev/null +++ b/common/src/catalog_storage.rs @@ -0,0 +1,211 @@ +use serde::{Deserialize, Serialize}; + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum ColumnType { + BigInt, + Integer, + Numeric, + Date, + TextArray, + Boolean, + Text, + Uuid, + Timestamp, + Json, +} + +impl ColumnType { + pub fn from_postgres_type(kind: &str) -> Option { + match kind { + "int8" => Some(Self::BigInt), + "int4" => Some(Self::Integer), + "numeric" => Some(Self::Numeric), + "date" => Some(Self::Date), + "_text" => Some(Self::TextArray), + "bool" => Some(Self::Boolean), + "text" => Some(Self::Text), + "uuid" => Some(Self::Uuid), + "timestamptz" => Some(Self::Timestamp), + "jsonb" => Some(Self::Json), + _ => None, + } + } + + pub fn text_representation(self) -> bool { + matches!(self, Self::Uuid | Self::Json | Self::TextArray) + } + + pub fn field_type(self) -> &'static str { + match self { + Self::BigInt => "BIGINT", + Self::Integer => "INT", + Self::Numeric => "NUMERIC", + Self::Date => "DATE", + Self::TextArray => "TEXT", + Self::Boolean => "BOOLEAN", + Self::Text => "TEXT", + Self::Uuid => "TEXT", + Self::Timestamp => "instant(6)", + Self::Json => "TEXT", + } + } + + pub fn postgres_type(self) -> &'static str { + match self { + Self::BigInt => "int8", + Self::Integer => "int4", + Self::Numeric => "numeric", + Self::Date => "date", + Self::TextArray => "_text", + Self::Boolean => "bool", + Self::Text => "text", + Self::Uuid => "uuid", + Self::Timestamp => "timestamptz", + Self::Json => "jsonb", + } + } +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ColumnSpec { + pub physical_name: String, + pub column_type: ColumnType, + pub not_null: bool, +} + +impl ColumnSpec { + pub fn new(name: &str, column_type: ColumnType) -> Self { + Self { + physical_name: name.into(), + column_type, + not_null: false, + } + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum DeletionPolicy { + None, + SoftDelete, + ManagedSoftDelete, +} + +impl DeletionPolicy { + pub fn is_soft_delete(self) -> bool { + matches!(self, Self::SoftDelete | Self::ManagedSoftDelete) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HistoryPolicy { + CurrentOnly, + ManagedVersions, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum NavigationPolicy { + Direct, + Tracked, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum RowScope { + All, + Profile { column: String }, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum RowWritePolicy { + Module, + Catalog, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct LinkSpec { + pub column: String, + pub target_table: String, + pub version_column: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct ProjectionSpec { + pub output_column: String, + pub link_column: String, + pub source_column: String, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct StorageSpec { + pub schema: String, + pub relation: String, + pub columns: Vec, + pub row_scope: RowScope, + pub row_writes: RowWritePolicy, + pub links: Vec, + pub projections: Vec, + pub deletion: DeletionPolicy, + pub history: HistoryPolicy, + pub navigation: NavigationPolicy, + pub created_at: Option, +} + +impl StorageSpec { + pub fn catalog_managed(&self) -> bool { + self.row_writes == RowWritePolicy::Catalog + } + + pub fn read_sql(&self, schema_id: i64) -> String { + let storage = self; + let mut columns = vec!["id".to_string()]; + columns.extend(storage.columns.iter().map(|column| { + let name = quote(&column.physical_name); + if column.column_type.text_representation() { + format!("{name}::TEXT AS {name}") + } else { + name + } + })); + columns.extend(storage.links.iter().map(|link| quote(&link.version_column))); + columns.push(match storage.deletion { + DeletionPolicy::None => "FALSE AS deleted".into(), + DeletionPolicy::SoftDelete | DeletionPolicy::ManagedSoftDelete => "deleted".into(), + }); + match storage.history { + HistoryPolicy::CurrentOnly => columns.extend([ + "1::BIGINT AS version".into(), + if storage.deletion == DeletionPolicy::ManagedSoftDelete { + "row_revision".into() + } else { "1::BIGINT AS row_revision".into() }, + ]), + HistoryPolicy::ManagedVersions => { + columns.extend(["version".into(), "row_revision".into()]) + } + } + columns.push(match &storage.created_at { + Some(column) => format!("{} AS created_at", quote(column)), + None => "NULL::TIMESTAMPTZ AS created_at".into(), + }); + let predicate = match &storage.row_scope { + RowScope::All => String::new(), + RowScope::Profile { column } => { + format!(" WHERE {} = {}", quote(column), schema_id) + } + }; + let relation = format!("{}.{}", quote(&storage.schema), quote(&storage.relation)); + format!("(SELECT {} FROM {relation}{predicate})", columns.join(", ")) + } +} + +fn quote(identifier: &str) -> String { + format!("\"{}\"", identifier.replace('"', "\"\"")) +} diff --git a/common/src/lib.rs b/common/src/lib.rs index 5ae9ede4..da5a8ade 100644 --- a/common/src/lib.rs +++ b/common/src/lib.rs @@ -3,6 +3,7 @@ #[cfg_attr(not(feature = "tantivy"), path = "search_light.rs")] pub mod search; pub mod alias; +pub mod catalog_storage; pub mod decimal; pub mod grpc_error; pub mod money; diff --git a/search/src/lib.rs b/search/src/lib.rs index d4610e04..703c5ca9 100644 --- a/search/src/lib.rs +++ b/search/src/lib.rs @@ -815,10 +815,11 @@ async fn qualified_visible_table( profile_name: &str, table_name: &str, ) -> Result { - let resolved = sqlx::query_as::<_, (String, String)>( - r#"SELECT owner.name, definition.table_name + let resolved = sqlx::query_as::<_, (i64, String, String, Option)>( + r#"SELECT definition.schema_id, owner.name, definition.table_name, binding.storage FROM table_definitions definition JOIN schemas owner ON owner.id = definition.schema_id + LEFT JOIN table_storage_bindings binding ON binding.table_definition_id = definition.id WHERE definition.canonical_table_name = $2 AND definition.deleted = FALSE AND (owner.name = $1 OR definition.is_global = TRUE) @@ -831,8 +832,15 @@ async fn qualified_visible_table( .await .map_err(|error| Status::internal(format!("Table storage lookup failed: {error}")))? .ok_or_else(|| Status::not_found(format!("Table '{table_name}' was not found")))?; - let (storage_schema, stored_table_name) = resolved; - Ok(qualify_profile_table(&storage_schema, &stored_table_name)) + let (schema_id, storage_schema, stored_table_name, storage) = resolved; + match storage { + Some(storage) => { + let storage = serde_json::from_value::(storage) + .map_err(|error| Status::internal(format!("Invalid table storage binding: {error}")))?; + Ok(storage.read_sql(schema_id)) + } + None => Ok(qualify_profile_table(&storage_schema, &stored_table_name)), + } } fn normalize_request(req: SearchRequest) -> Result { @@ -1289,6 +1297,7 @@ async fn resolve_order_column( .or_else(|| { (is_system_column(&requested_key) && !is_internal_column(&requested_key) + && requested_key != common::system_column::ACCOUNT_REFERENCE_COLUMN && !physical_to_display.contains_key(&requested_key)) .then(|| requested_key.clone()) }) @@ -1299,37 +1308,6 @@ async fn resolve_order_column( )) })?; - let physical_column = sqlx::query_scalar::<_, String>( - r#" - SELECT column_name - FROM information_schema.columns - WHERE table_schema = ( - SELECT owner.name - FROM table_definitions definition - JOIN schemas owner ON owner.id = definition.schema_id - WHERE definition.table_name = $2 - AND definition.deleted = FALSE - AND (owner.name = $1 OR definition.is_global = TRUE) - ORDER BY definition.is_global ASC - LIMIT 1 - ) - AND table_name = $2 - AND LOWER(column_name) = LOWER($3) - "#, - ) - .bind(profile_name) - .bind(table_name) - .bind(&physical_column) - .fetch_optional(pool) - .await - .map_err(|e| Status::internal(format!("Order column lookup failed: {}", e)))?; - let Some(physical_column) = physical_column else { - return Err(Status::invalid_argument(format!( - "Order column '{}' was not found in table '{}.{}'", - requested_column, profile_name, table_name - ))); - }; - Ok(ResolvedOrderColumn::Column(physical_column)) } diff --git a/server b/server index 3c9829e6..5bad22db 160000 --- a/server +++ b/server @@ -1 +1 @@ -Subproject commit 3c9829e6df6e5c7fa6de571b06050081c9663c8d +Subproject commit 5bad22db532af93955bcc013b2486dd38a5d78a9