fixing problems

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<Self> {
+        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<ColumnSpec>,
+    pub row_scope: RowScope,
+    pub row_writes: RowWritePolicy,
+    pub links: Vec<LinkSpec>,
+    pub projections: Vec<ProjectionSpec>,
+    pub deletion: DeletionPolicy,
+    pub history: HistoryPolicy,
+    pub navigation: NavigationPolicy,
+    pub created_at: Option<String>,
+}
+
+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('"', "\"\""))
+}

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;

search/src/lib.rs

@@ -815,10 +815,11 @@ async fn qualified_visible_table(
     profile_name: &str,
     table_name: &str,
 ) -> Result<String, Status> {
-    let resolved = sqlx::query_as::<_, (String, String)>(
+    let resolved = sqlx::query_as::<_, (i64, String, String, Option<serde_json::Value>)>(
-        r#"SELECT owner.name, definition.table_name
+        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;
+    let (schema_id, storage_schema, stored_table_name, storage) = resolved;
-    Ok(qualify_profile_table(&storage_schema, &stored_table_name))
+    match storage {
+        Some(storage) => {
+            let storage = serde_json::from_value::<common::catalog_storage::StorageSpec>(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<NormalizedSearchRequest, Status> {
@@ -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))
 }