Implement live database schema introspection via pg_catalog

crates/pg-analysis/src/index.rs

@@ -55,8 +55,22 @@ impl WorkspaceIndex {
 
         // Remove old entries, then insert new ones.
         self.remove_file(uri);
+        self.index_by_name(&symbol_arcs);
+        self.definitions.insert(uri.to_string(), symbol_arcs);
+        self.references.insert(uri.to_string(), ref_arcs);
+    }
+
+    /// Insert pre-built symbols (e.g., from database introspection).
+    /// Uses the provided URI to group them, allowing removal via `remove_file`.
+    pub fn load_symbols(&self, uri: &str, symbols: Vec<Symbol>) {
+        self.remove_file(uri);
+        let symbol_arcs: Vec<Arc<Symbol>> = symbols.into_iter().map(Arc::new).collect();
+        self.index_by_name(&symbol_arcs);
+        self.definitions.insert(uri.to_string(), symbol_arcs);
+    }
 
-        for sym in &symbol_arcs {
+    fn index_by_name(&self, symbols: &[Arc<Symbol>]) {
+        for sym in symbols {
             let key = (sym.kind, sym.name.name.to_lowercase());
             self.by_name.entry(key).or_default().push(Arc::clone(sym));
 
@@ -66,9 +80,6 @@ impl WorkspaceIndex {
                 self.by_name.entry(child_key).or_default().push(child_arc);
             }
         }
-
-        self.definitions.insert(uri.to_string(), symbol_arcs);
-        self.references.insert(uri.to_string(), ref_arcs);
     }
 
     /// Remove all entries for a file.

crates/pg-lsp/Cargo.toml

@@ -13,6 +13,7 @@ path = "src/main.rs"
 pg-parse = { path = "../pg-parse" }
 pg-analysis = { path = "../pg-analysis" }
 pg-format = { path = "../pg-format" }
+pg-schema = { path = "../pg-schema" }
 tree-sitter.workspace = true
 tree-sitter-postgres.workspace = true
 tower-lsp.workspace = true
@@ -23,4 +24,4 @@ serde.workspace = true
 serde_json.workspace = true
 tracing.workspace = true
 tracing-subscriber.workspace = true
-clap = { version = "4", features = ["derive"] }
+clap = { version = "4", features = ["derive", "env"] }

crates/pg-lsp/src/main.rs

@@ -21,11 +21,23 @@ struct Cli {
         help = "Formatting style: river, mozilla, aweber, dbt, gitlab, kickstarter, mattmc3"
     )]
     format_style: Style,
+
+    /// PostgreSQL connection URL for live schema introspection
+    #[arg(
+        long,
+        short = 'd',
+        env = "DATABASE_URL",
+        help = "e.g., postgres://user:pass@localhost/dbname"
+    )]
+    database_url: Option<String>,
 }
 
 fn parse_style(s: &str) -> Result<Style, String> {
-    s.parse::<Style>()
-        .map_err(|_| format!("unknown style '{s}'; options: river, mozilla, aweber, dbt, gitlab, kickstarter, mattmc3"))
+    s.parse::<Style>().map_err(|_| {
+        format!(
+            "unknown style '{s}'; options: river, mozilla, aweber, dbt, gitlab, kickstarter, mattmc3"
+        )
+    })
 }
 
 #[tokio::main]
@@ -40,7 +52,9 @@ async fn main() {
         .init();
 
     let format_style = cli.format_style;
-    let (service, socket) =
-        LspService::new(move |client| server::Backend::new(client, format_style));
+    let database_url = cli.database_url;
+    let (service, socket) = LspService::new(move |client| {
+        server::Backend::new(client, format_style, database_url.clone())
+    });
     Server::new(stdin(), stdout(), socket).serve(service).await;
 }

crates/pg-lsp/src/server.rs

@@ -26,16 +26,18 @@ pub struct Backend {
     documents: DashMap<String, Document>,
     index: Arc<WorkspaceIndex>,
     format_style: Style,
+    database_url: Option<String>,
 }
 
 impl Backend {
-    pub fn new(client: Client, format_style: Style) -> Self {
+    pub fn new(client: Client, format_style: Style, database_url: Option<String>) -> Self {
         Self {
             client,
             pool: Arc::new(ParserPool::new()),
             documents: DashMap::new(),
             index: Arc::new(WorkspaceIndex::new()),
             format_style,
+            database_url,
         }
     }
 
@@ -275,6 +277,18 @@ impl LanguageServer for Backend {
 
     async fn initialized(&self, _params: InitializedParams) {
         info!("pg-lsp initialized");
+
+        // Load database schema in the background so it doesn't block other LSP requests.
+        if let Some(ref db_url) = self.database_url {
+            let index = Arc::clone(&self.index);
+            let url = db_url.clone();
+            tokio::spawn(async move {
+                match pg_schema::load_database_schema(&url, &index).await {
+                    Ok(count) => info!("loaded {count} symbols from database"),
+                    Err(e) => tracing::warn!("failed to load database schema: {e}"),
+                }
+            });
+        }
     }
 
     async fn shutdown(&self) -> Result<()> {

crates/pg-schema/src/catalog.rs

@@ -1 +1,221 @@
-// Live database introspection via pg_catalog — Phase 7
+use std::collections::HashMap;
+
+use pg_analysis::symbols::{QualifiedName, Symbol, SymbolKind};
+use tokio_postgres::Client;
+
+/// The synthetic URI used for database-sourced symbols.
+pub const DB_URI: &str = "pg-catalog://database";
+
+/// Namespace exclusion filter — skip system and information_schema namespaces.
+const NS_FILTER: &str = "n.nspname NOT LIKE 'pg_%' AND n.nspname != 'information_schema'";
+
+/// Load all schemas, tables, columns, functions, types, and sequences
+/// from a live PostgreSQL database via `pg_catalog` queries.
+pub async fn load_catalog(client: &Client) -> Result<Vec<Symbol>, CatalogError> {
+    let mut symbols = Vec::new();
+
+    load_schemas(client, &mut symbols).await?;
+    load_tables_and_columns(client, &mut symbols).await?;
+    load_functions(client, &mut symbols).await?;
+    load_types(client, &mut symbols).await?;
+    load_sequences(client, &mut symbols).await?;
+
+    Ok(symbols)
+}
+
+async fn load_schemas(client: &Client, symbols: &mut Vec<Symbol>) -> Result<(), CatalogError> {
+    let query = format!("SELECT nspname FROM pg_catalog.pg_namespace n WHERE {NS_FILTER}");
+    let rows = client.query(&query, &[]).await?;
+
+    for row in rows {
+        let name: String = row.get(0);
+        symbols.push(make_symbol(SymbolKind::Schema, None, &name, ""));
+    }
+
+    Ok(())
+}
+
+async fn load_tables_and_columns(
+    client: &Client,
+    symbols: &mut Vec<Symbol>,
+) -> Result<(), CatalogError> {
+    // Single batch query for tables/views and their columns (avoids N+1).
+    let query = format!(
+        "SELECT n.nspname, c.relname, c.relkind, \
+                a.attname, format_type(a.atttypid, a.atttypmod) \
+         FROM pg_catalog.pg_class c \
+         JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
+         LEFT JOIN pg_catalog.pg_attribute a \
+           ON a.attrelid = c.oid AND a.attnum > 0 AND NOT a.attisdropped \
+         WHERE c.relkind IN ('r', 'v', 'm', 'f') AND {NS_FILTER} \
+         ORDER BY n.nspname, c.relname, a.attnum"
+    );
+    let rows = client.query(&query, &[]).await?;
+
+    // Group columns by (schema, table).
+    let mut tables: Vec<(String, String, SymbolKind)> = Vec::new();
+    let mut columns: HashMap<(String, String), Vec<Symbol>> = HashMap::new();
+
+    for row in &rows {
+        let schema: String = row.get(0);
+        let name: String = row.get(1);
+        // PostgreSQL "char" type maps to i8 in tokio_postgres; all relkind values are ASCII.
+        let relkind: i8 = row.get(2);
+        let col_name: Option<String> = row.get(3);
+        let col_type: Option<String> = row.get(4);
+
+        let kind = match relkind as u8 as char {
+            'r' => SymbolKind::Table,
+            'v' => SymbolKind::View,
+            'm' => SymbolKind::MaterializedView,
+            'f' => SymbolKind::ForeignTable,
+            _ => continue,
+        };
+
+        let key = (schema.clone(), name.clone());
+        if !columns.contains_key(&key) {
+            tables.push((schema.clone(), name.clone(), kind));
+            columns.insert(key.clone(), Vec::new());
+        }
+
+        if let (Some(cn), Some(ct)) = (col_name, col_type) {
+            let col_def = format!("{cn} {ct}");
+            columns.get_mut(&key).unwrap().push(make_symbol(
+                SymbolKind::Column,
+                None,
+                &cn,
+                &col_def,
+            ));
+        }
+    }
+
+    for (schema, name, kind) in tables {
+        let def_text = format!("{kind_label} {schema}.{name}", kind_label = kind.label());
+        let mut sym = make_symbol(kind, Some(&schema), &name, &def_text);
+        sym.children = columns.remove(&(schema, name)).unwrap_or_default();
+        symbols.push(sym);
+    }
+
+    Ok(())
+}
+
+async fn load_functions(client: &Client, symbols: &mut Vec<Symbol>) -> Result<(), CatalogError> {
+    let query = format!(
+        "SELECT n.nspname, p.proname, \
+                pg_catalog.pg_get_function_arguments(p.oid), \
+                pg_catalog.pg_get_function_result(p.oid), \
+                p.prokind \
+         FROM pg_catalog.pg_proc p \
+         JOIN pg_catalog.pg_namespace n ON n.oid = p.pronamespace \
+         WHERE {NS_FILTER}"
+    );
+    let rows = client.query(&query, &[]).await?;
+
+    for row in rows {
+        let schema: String = row.get(0);
+        let name: String = row.get(1);
+        let args: String = row.get(2);
+        let ret: String = row.get(3);
+        let prokind: i8 = row.get(4);
+
+        let kind = match prokind as u8 as char {
+            'p' => SymbolKind::Procedure,
+            _ => SymbolKind::Function,
+        };
+
+        let def_text = format!(
+            "{kind_label} {schema}.{name}({args}) RETURNS {ret}",
+            kind_label = kind.label()
+        );
+        symbols.push(make_symbol(kind, Some(&schema), &name, &def_text));
+    }
+
+    Ok(())
+}
+
+async fn load_types(client: &Client, symbols: &mut Vec<Symbol>) -> Result<(), CatalogError> {
+    let query = format!(
+        "SELECT n.nspname, t.typname, t.typtype \
+         FROM pg_catalog.pg_type t \
+         JOIN pg_catalog.pg_namespace n ON n.oid = t.typnamespace \
+         WHERE {NS_FILTER} \
+           AND t.typtype IN ('c', 'e', 'r', 'd') \
+           AND NOT EXISTS (SELECT 1 FROM pg_catalog.pg_class c WHERE c.reltype = t.oid)"
+    );
+    let rows = client.query(&query, &[]).await?;
+
+    for row in rows {
+        let schema: String = row.get(0);
+        let name: String = row.get(1);
+        let typtype: i8 = row.get(2);
+
+        let kind = match typtype as u8 as char {
+            'd' => SymbolKind::Domain,
+            _ => SymbolKind::Type,
+        };
+
+        let def_text = format!("{kind_label} {schema}.{name}", kind_label = kind.label());
+        symbols.push(make_symbol(kind, Some(&schema), &name, &def_text));
+    }
+
+    Ok(())
+}
+
+async fn load_sequences(client: &Client, symbols: &mut Vec<Symbol>) -> Result<(), CatalogError> {
+    let query = format!(
+        "SELECT n.nspname, c.relname \
+         FROM pg_catalog.pg_class c \
+         JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace \
+         WHERE c.relkind = 'S' AND {NS_FILTER}"
+    );
+    let rows = client.query(&query, &[]).await?;
+
+    for row in rows {
+        let schema: String = row.get(0);
+        let name: String = row.get(1);
+        let def_text = format!("sequence {schema}.{name}");
+        symbols.push(make_symbol(
+            SymbolKind::Sequence,
+            Some(&schema),
+            &name,
+            &def_text,
+        ));
+    }
+
+    Ok(())
+}
+
+fn make_symbol(
+    kind: SymbolKind,
+    schema: Option<&str>,
+    name: &str,
+    definition_text: &str,
+) -> Symbol {
+    let qname = match schema {
+        Some(s) => QualifiedName::with_schema(s.to_string(), name.to_string()),
+        None => QualifiedName::new(name.to_string()),
+    };
+    Symbol {
+        kind,
+        name: qname,
+        uri: DB_URI.to_string(),
+        start_byte: 0,
+        end_byte: 0,
+        start_line: 0,
+        start_col: 0,
+        end_line: 0,
+        end_col: 0,
+        name_start_line: 0,
+        name_start_col: 0,
+        name_end_line: 0,
+        name_end_col: 0,
+        definition_text: definition_text.to_string(),
+        children: Vec::new(),
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum CatalogError {
+    #[error("database error: {0}")]
+    Database(#[from] tokio_postgres::Error),
+}

crates/pg-schema/src/lib.rs

@@ -1 +1,29 @@
 pub mod catalog;
+
+use pg_analysis::WorkspaceIndex;
+use tracing::info;
+
+pub use catalog::{CatalogError, DB_URI};
+
+/// Load schema from a live PostgreSQL database and merge into the workspace index.
+pub async fn load_database_schema(
+    database_url: &str,
+    index: &WorkspaceIndex,
+) -> Result<usize, CatalogError> {
+    let (client, connection) = tokio_postgres::connect(database_url, tokio_postgres::NoTls).await?;
+
+    // Spawn the connection handler.
+    tokio::spawn(async move {
+        if let Err(e) = connection.await {
+            tracing::error!("database connection error: {e}");
+        }
+    });
+
+    let symbols = catalog::load_catalog(&client).await?;
+    let count = symbols.len();
+
+    info!("loaded {count} symbols from database");
+    index.load_symbols(DB_URI, symbols);
+
+    Ok(count)
+}