fix(pgwire): support ActiveRecord catalog reflection

Author: farhan-syah

docs/protocols.md

@@ -28,9 +28,9 @@ psql -h localhost -p 6432
 
 **SQL coverage:** Everything in the [query language reference](query-language.md).
 
-**Driver compatibility:** NodeDB advertises a libpq-parseable `server_version` (`15.0 (NodeDB <version>)`) and the matching PostgreSQL-compatible `server_version_num` in the startup parameter burst. It also supports the probes drivers issue on connect: `version()` returns a PostgreSQL-compatible string, `current_setting(name [, missing_ok])` resolves the same settings as `SHOW`, and `::regclass`/`::regtype` casts, `ANY(current_schemas(...))`, and cross-catalog-table JOINs evaluate PostgreSQL-identically. Binary result formats requested at Bind are honored per column; types the encoder cannot yet emit in binary (timestamp, numeric, json/jsonb, arrays) downgrade to text, with the Describe-phase RowDescription kept in sync.
+**Driver compatibility:** NodeDB advertises a libpq-parseable `server_version` (`15.0 (NodeDB <version>)`) and the matching PostgreSQL-compatible `server_version_num` in the startup parameter burst. It also supports the probes drivers issue on connect: `version()` returns a PostgreSQL-compatible string, `current_setting(name [, missing_ok])` resolves the same settings as `SHOW`, `current_schemas(...)` returns PostgreSQL `TEXT[]` syntax, and `::regclass`/`::regtype` casts, `ANY(...)`, and ActiveRecord-style cross-catalog-table JOINs and projections evaluate PostgreSQL-identically. Binary result formats requested at Bind are honored per column; types the encoder cannot yet emit in binary (timestamp, numeric, json/jsonb, arrays) downgrade to text, with the Describe-phase RowDescription kept in sync.
 
-**Introspection:** NodeDB exposes PostgreSQL-compatible `pg_catalog` virtual tables (e.g., `pg_class`, `pg_namespace`, `pg_attribute`, `pg_type`) so that standard Postgres clients, ORMs, and business intelligence tools can introspect the database schema without modification (`psql \d`, driver type caches, ORM bootstraps). Queries against `pg_catalog.*` tables are transparently rewritten to pull from NodeDB's internal catalog.
+**Introspection:** NodeDB exposes PostgreSQL-compatible `pg_catalog` virtual tables (including `pg_class`, `pg_namespace`, `pg_attribute`, `pg_attrdef`, `pg_type`, `pg_range`, and `pg_collation`) so that standard Postgres clients, ORMs, and business intelligence tools can introspect the database schema without modification (`psql \d`, driver type caches, ORM bootstraps). Column defaults and nullability are reflected, and the metadata functions used by PostgreSQL clients (`format_type`, `pg_get_expr`, and `col_description`) are supported. Queries against `pg_catalog.*` tables are transparently rewritten to pull from NodeDB's internal catalog.
 
 **Streaming:** Unordered multi-row SELECTs stream lazily to the client — rows are not buffered and merged on the coordinator first. Ordered, aggregate, point-get, and search queries use the materialized path.
 

nodedb-physical/src/physical_plan/query.rs

@@ -186,6 +186,12 @@ pub enum QueryOp {
         post_aggregates: Vec<(String, String)>,
         /// Post-join projection: column names to keep (empty = all).
         projection: Vec<JoinProjection>,
+        /// MessagePack-encoded computed projection expressions. When present,
+        /// these represent the complete SELECT list in output order.
+        computed_projection: Vec<u8>,
+        /// Residual `ON` predicates evaluated for each equi-key candidate
+        /// before outer-join match accounting (MessagePack).
+        join_filters: Vec<u8>,
         /// Post-join WHERE filter predicates (MessagePack).
         post_filters: Vec<u8>,
         /// Resolved child plan for the left side. An Exchange child during

nodedb-physical/src/physical_plan/wire.rs

@@ -99,3 +99,39 @@ pub fn encode_batch(plans: &Vec<PhysicalPlan>) -> Result<Vec<u8>, WireError> {
 pub fn decode_batch(bytes: &[u8]) -> Result<Vec<PhysicalPlan>, WireError> {
     zerompk::from_msgpack(bytes).map_err(|e| WireError::Codec(format!("batch decode: {e}")))
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::physical_plan::JoinProjection;
+
+    #[test]
+    fn hash_join_tail_semantics_roundtrip() {
+        let plan = PhysicalPlan::Query(QueryOp::HashJoin {
+            left_collection: "left".into(),
+            right_collection: "right".into(),
+            left_alias: Some("l".into()),
+            right_alias: Some("r".into()),
+            on: vec![("id".into(), "id".into())],
+            join_type: "left".into(),
+            limit: 10,
+            post_group_by: Vec::new(),
+            post_aggregates: Vec::new(),
+            projection: vec![JoinProjection {
+                source: "l.id".into(),
+                output: "id".into(),
+            }],
+            computed_projection: vec![1, 2],
+            join_filters: vec![3, 4],
+            post_filters: vec![5, 6],
+            left_input: None,
+            right_input: None,
+            left_bitmap: None,
+            right_bitmap: None,
+        });
+
+        let encoded = encode(&plan).expect("hash join encodes");
+        let decoded = decode(&encoded).expect("hash join decodes");
+        assert_eq!(decoded, plan);
+    }
+}

nodedb-query/src/functions/system.rs

@@ -5,9 +5,67 @@
 use nodedb_types::Value;
 
 /// Evaluate a system scalar function; returns `None` if `name` is not handled.
-pub(crate) fn try_eval(name: &str, _args: &[Value]) -> Option<Value> {
+pub(crate) fn try_eval(name: &str, args: &[Value]) -> Option<Value> {
     match name {
         "version" => Some(Value::String(nodedb_types::pg_compat::version_string())),
+        "format_type" => Some(format_type(args)),
+        "pg_get_expr" => Some(args.first().cloned().unwrap_or(Value::Null)),
+        // NodeDB does not persist per-column comments yet.
+        "col_description" => Some(Value::Null),
         _ => None,
     }
 }
+
+fn format_type(args: &[Value]) -> Value {
+    let Some(Value::Integer(oid)) = args.first() else {
+        return Value::Null;
+    };
+    let name = match *oid {
+        16 => "boolean",
+        17 => "bytea",
+        20 => "bigint",
+        21 => "smallint",
+        23 => "integer",
+        25 => "text",
+        26 => "oid",
+        114 => "json",
+        700 => "real",
+        701 => "double precision",
+        1042 => "character",
+        1043 => "character varying",
+        1082 => "date",
+        1083 => "time without time zone",
+        1114 => "timestamp without time zone",
+        1184 => "timestamp with time zone",
+        1186 => "interval",
+        1700 => "numeric",
+        2950 => "uuid",
+        3802 => "jsonb",
+        _ => return Value::String("text".into()),
+    };
+    Value::String(name.into())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn postgres_catalog_metadata_functions() {
+        assert_eq!(
+            try_eval("format_type", &[Value::Integer(23), Value::Integer(-1)]),
+            Some(Value::String("integer".into()))
+        );
+        assert_eq!(
+            try_eval(
+                "pg_get_expr",
+                &[Value::String("'untitled'".into()), Value::Integer(42)]
+            ),
+            Some(Value::String("'untitled'".into()))
+        );
+        assert_eq!(
+            try_eval("col_description", &[Value::Integer(42), Value::Integer(1)]),
+            Some(Value::Null)
+        );
+    }
+}

nodedb-sql/src/ddl_ast/collection_type.rs

@@ -306,10 +306,11 @@ pub fn parse_column_type_str_full(type_str: &str) -> (String, bool, bool, Option
     // type_str may look like: "TEXT DEFAULT upper('x')" or "INT NOT NULL DEFAULT 1 + 2".
     let default_expr = if let Some(def_pos) = find_ascii_case_insensitive(type_str, "DEFAULT") {
         let after = type_str[def_pos + "DEFAULT".len()..].trim();
-        if after.is_empty() {
+        let expression = &after[..default_expression_end(after)];
+        if expression.trim().is_empty() {
             None
         } else {
-            Some(after.to_string())
+            Some(expression.trim().to_string())
         }
     } else {
         None
@@ -325,6 +326,62 @@ pub fn parse_column_type_str_full(type_str: &str) -> (String, bool, bool, Option
     (bare.to_string(), is_pk, is_not_null, default_expr)
 }
 
+/// Return the byte offset where a trailing column constraint begins. Constraint
+/// keywords inside quoted strings or parenthesized expressions belong to the
+/// default expression and are intentionally ignored.
+fn default_expression_end(input: &str) -> usize {
+    const CONSTRAINTS: &[&str] = &[
+        "NOT NULL",
+        "PRIMARY KEY",
+        "UNIQUE",
+        "CHECK",
+        "REFERENCES",
+        "COLLATE",
+        "GENERATED",
+        "CONSTRAINT",
+    ];
+
+    let bytes = input.as_bytes();
+    let mut depth = 0usize;
+    let mut quote = None;
+    let mut index = 0usize;
+    while index < bytes.len() {
+        let byte = bytes[index];
+        if let Some(quoted_by) = quote {
+            if byte == quoted_by {
+                if index + 1 < bytes.len() && bytes[index + 1] == quoted_by {
+                    index += 2;
+                    continue;
+                }
+                quote = None;
+            }
+            index += 1;
+            continue;
+        }
+        match byte {
+            b'\'' | b'"' => quote = Some(byte),
+            b'(' => depth += 1,
+            b')' => depth = depth.saturating_sub(1),
+            _ if depth == 0 && (index == 0 || bytes[index - 1].is_ascii_whitespace()) => {
+                let tail = &input[index..];
+                if CONSTRAINTS.iter().any(|keyword| {
+                    tail.get(..keyword.len())
+                        .is_some_and(|prefix| prefix.eq_ignore_ascii_case(keyword))
+                        && tail
+                            .as_bytes()
+                            .get(keyword.len())
+                            .is_none_or(|next| next.is_ascii_whitespace() || *next == b'(')
+                }) {
+                    return index;
+                }
+            }
+            _ => {}
+        }
+        index += 1;
+    }
+    input.len()
+}
+
 /// Parse TTL from the options list.
 ///
 /// Looks for a key `ttl` whose value is either:
@@ -444,6 +501,18 @@ mod tests {
         assert_eq!(default_expr.as_deref(), Some("42"));
     }
 
+    #[test]
+    fn default_expression_excludes_trailing_constraints() {
+        let (_, _, is_not_null, default_expr) = parse_column_type_str_full(
+            "TEXT DEFAULT concat('NOT NULL', upper('(CHECK)')) NOT NULL UNIQUE",
+        );
+        assert!(is_not_null);
+        assert_eq!(
+            default_expr.as_deref(),
+            Some("concat('NOT NULL', upper('(CHECK)'))")
+        );
+    }
+
     // ── engine name → CollectionType variant ─────────────────────────────
 
     #[test]

nodedb-sql/src/functions/arg_types.rs

@@ -267,6 +267,8 @@ pub static REPLACE_ARGS: &[ArgTypeSpec] =
 
 pub static MAKE_ARRAY_ARGS: &[ArgTypeSpec] = &[any_variadic("expr")];
 
+pub static PG_CATALOG_2_ARGS: &[ArgTypeSpec] = &[any("value"), any("context")];
+
 // ── PostgreSQL JSON operators ─────────────────────────────────────────────────
 
 pub static PG_JSON_2_ARGS: &[ArgTypeSpec] = &[any("json_col"), any("key")];

nodedb-sql/src/functions/builtins/scalars/misc.rs

@@ -56,5 +56,32 @@ pub(super) fn misc_functions() -> Vec<FunctionMeta> {
             Some(ColumnType::String),
             arg_types::NO_ARGS,
         ),
+        m(
+            "format_type",
+            Scalar,
+            2,
+            2,
+            no_trigger(),
+            Some(ColumnType::String),
+            arg_types::PG_CATALOG_2_ARGS,
+        ),
+        m(
+            "pg_get_expr",
+            Scalar,
+            2,
+            2,
+            no_trigger(),
+            Some(ColumnType::String),
+            arg_types::PG_CATALOG_2_ARGS,
+        ),
+        m(
+            "col_description",
+            Scalar,
+            2,
+            2,
+            no_trigger(),
+            Some(ColumnType::String),
+            arg_types::PG_CATALOG_2_ARGS,
+        ),
     ]
 }

nodedb-sql/src/planner/const_fold.rs

@@ -47,6 +47,11 @@ pub fn fold_constant_default(expr: &SqlExpr) -> Option<SqlValue> {
 pub fn fold_constant(expr: &SqlExpr, registry: &FunctionRegistry) -> Option<SqlValue> {
     match expr {
         SqlExpr::Literal(v) => Some(v.clone()),
+        SqlExpr::ArrayLiteral(items) => items
+            .iter()
+            .map(|item| fold_constant(item, registry))
+            .collect::<Option<Vec<_>>>()
+            .map(SqlValue::Array),
         SqlExpr::UnaryOp {
             op: UnaryOp::Neg,
             expr,
@@ -335,6 +340,22 @@ mod tests {
         assert!(fold_constant(&expr, &registry).is_none());
     }
 
+    #[test]
+    fn fold_array_literal_recursively() {
+        let registry = FunctionRegistry::new();
+        let expr = SqlExpr::ArrayLiteral(vec![
+            SqlExpr::Literal(SqlValue::String("public".into())),
+            SqlExpr::Literal(SqlValue::Int(42)),
+        ]);
+        assert_eq!(
+            fold_constant(&expr, &registry),
+            Some(SqlValue::Array(vec![
+                SqlValue::String("public".into()),
+                SqlValue::Int(42),
+            ]))
+        );
+    }
+
     #[test]
     fn fold_literal_arithmetic_still_works() {
         let registry = FunctionRegistry::new();

nodedb/src/control/planner/sql_plan_convert/aggregate/mod.rs

@@ -18,7 +18,7 @@ pub(in crate::control::planner::sql_plan_convert) use plan::{
 };
 pub(in crate::control::planner::sql_plan_convert) use projection::{
     extract_computed_columns, extract_join_projection_specs, extract_projection_names,
-    serialize_window_functions,
+    serialize_join_computed_projection, serialize_window_functions,
 };
 pub(in crate::control::planner::sql_plan_convert) use spec::{
     agg_expr_to_pair, extract_collection_name, extract_scan_alias, inline_join_side,

nodedb/src/control/planner/sql_plan_convert/aggregate/plan.rs

@@ -123,6 +123,8 @@ pub(in crate::control::planner::sql_plan_convert) fn convert_aggregate(
                 post_group_by: group_strs,
                 post_aggregates: agg_pairs,
                 projection: Vec::new(),
+                computed_projection: Vec::new(),
+                join_filters: Vec::new(),
                 post_filters: Vec::new(),
                 left_input,
                 right_input,