nodedb_sql/resolver/
columns.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! Column and table resolution against the catalog.
4
5use std::collections::HashMap;
6
7use nodedb_types::DatabaseId;
8
9use crate::error::{Result, SqlError};
10use crate::parser::normalize::{
11    normalize_ident, normalize_object_name_checked, table_name_from_factor,
12};
13use crate::types::{
14    ArrayCatalogView, CollectionInfo, ColumnInfo, EngineType, SqlCatalog, SqlDataType,
15};
16use crate::types_array::{ArrayAttrType, ArrayDimType};
17
18/// Resolved table reference: name, alias, and catalog info.
19#[derive(Debug, Clone)]
20pub struct ResolvedTable {
21    pub name: String,
22    pub alias: Option<String>,
23    pub info: CollectionInfo,
24}
25
26impl ResolvedTable {
27    /// The name to use for qualified column references.
28    pub fn ref_name(&self) -> &str {
29        self.alias.as_deref().unwrap_or(&self.name)
30    }
31}
32
33/// Context built during FROM clause resolution.
34#[derive(Debug, Default)]
35pub struct TableScope {
36    /// Tables by reference name (alias or table name).
37    pub tables: HashMap<String, ResolvedTable>,
38    /// Insertion order for unambiguous column resolution.
39    order: Vec<String>,
40}
41
42impl TableScope {
43    pub fn new() -> Self {
44        Self::default()
45    }
46
47    /// Add a resolved table. Returns error if name conflicts.
48    pub fn add(&mut self, table: ResolvedTable) -> Result<()> {
49        let key = table.ref_name().to_string();
50        if self.tables.contains_key(&key) {
51            return Err(SqlError::Parse {
52                detail: format!("duplicate table reference: {key}"),
53            });
54        }
55        self.order.push(key.clone());
56        self.tables.insert(key, table);
57        Ok(())
58    }
59
60    /// Resolve a column name, optionally qualified with a table reference.
61    ///
62    /// For schemaless collections, any column is accepted (dynamic fields).
63    /// For typed collections, the column must exist in the schema.
64    pub fn resolve_column(
65        &self,
66        table_ref: Option<&str>,
67        column: &str,
68    ) -> Result<(String, String)> {
69        let col = column.to_lowercase();
70
71        if let Some(tref) = table_ref {
72            let tref_lower = tref.to_lowercase();
73            let table = self
74                .tables
75                .get(&tref_lower)
76                .ok_or_else(|| SqlError::UnknownTable {
77                    name: tref_lower.clone(),
78                })?;
79            self.validate_column(table, &col)?;
80            return Ok((table.name.clone(), col));
81        }
82
83        // Unqualified: search all tables.
84        let mut matches = Vec::new();
85        for key in &self.order {
86            let table = &self.tables[key];
87            if self.column_exists(table, &col) {
88                matches.push(table.name.clone());
89            }
90        }
91
92        match matches.len() {
93            0 => {
94                // For single-table queries with schemaless, accept anything.
95                if self.tables.len() == 1 {
96                    let table = self
97                        .tables
98                        .values()
99                        .next()
100                        .expect("invariant: self.tables.len() == 1 checked immediately above");
101                    if table.info.engine == EngineType::DocumentSchemaless {
102                        return Ok((table.name.clone(), col));
103                    }
104                }
105                Err(SqlError::UnknownColumn {
106                    table: self
107                        .order
108                        .first()
109                        .cloned()
110                        .unwrap_or_else(|| "<unknown>".into()),
111                    column: col,
112                })
113            }
114            1 => Ok((
115                matches
116                    .into_iter()
117                    .next()
118                    .expect("invariant: matches.len() == 1 guaranteed by this match arm"),
119                col,
120            )),
121            _ => Err(SqlError::AmbiguousColumn { column: col }),
122        }
123    }
124
125    fn column_exists(&self, table: &ResolvedTable, column: &str) -> bool {
126        // Schemaless accepts any column.
127        if table.info.engine == EngineType::DocumentSchemaless {
128            return true;
129        }
130        table.info.columns.iter().any(|c| c.name == column)
131    }
132
133    fn validate_column(&self, table: &ResolvedTable, column: &str) -> Result<()> {
134        if self.column_exists(table, column) {
135            Ok(())
136        } else {
137            Err(SqlError::UnknownColumn {
138                table: table.name.clone(),
139                column: column.into(),
140            })
141        }
142    }
143
144    /// Get the single table in scope (for single-table queries).
145    pub fn single_table(&self) -> Option<&ResolvedTable> {
146        if self.tables.len() == 1 {
147            self.tables.values().next()
148        } else {
149            Option::None
150        }
151    }
152
153    /// Resolve tables from a FROM clause.
154    pub fn resolve_from(
155        catalog: &dyn SqlCatalog,
156        from: &[sqlparser::ast::TableWithJoins],
157    ) -> Result<Self> {
158        let mut scope = Self::new();
159        for table_with_joins in from {
160            scope.resolve_table_factor(catalog, &table_with_joins.relation)?;
161            for join in &table_with_joins.joins {
162                scope.resolve_table_factor(catalog, &join.relation)?;
163            }
164        }
165        Ok(scope)
166    }
167
168    fn resolve_table_factor(
169        &mut self,
170        catalog: &dyn SqlCatalog,
171        factor: &sqlparser::ast::TableFactor,
172    ) -> Result<()> {
173        // ARRAY_*(...) table-valued function: synthesize a ResolvedTable
174        // from the array's dim+attr schema so equi-join keys against the
175        // TVF's output rows resolve.
176        if let Some(resolved) = resolve_array_tvf(catalog, factor)? {
177            self.add(resolved)?;
178            return Ok(());
179        }
180        // LATERAL derived subquery: register the alias as a schemaless
181        // collection so qualified column references (`alias.col`) resolve
182        // without a catalog lookup. The actual inner plan is built separately.
183        if let sqlparser::ast::TableFactor::Derived {
184            lateral: true,
185            alias: Some(alias),
186            ..
187        } = factor
188        {
189            let alias_str = normalize_ident(&alias.name);
190            self.add(ResolvedTable {
191                name: alias_str.clone(),
192                alias: Some(alias_str.clone()),
193                info: CollectionInfo {
194                    name: alias_str,
195                    engine: EngineType::DocumentSchemaless,
196                    columns: Vec::new(),
197                    primary_key: None,
198                    has_auto_tier: false,
199                    indexes: Vec::new(),
200                    bitemporal: false,
201                    primary: nodedb_types::PrimaryEngine::Document,
202                    vector_primary: None,
203                    partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
204                },
205            })?;
206            return Ok(());
207        }
208        if let Some((name, alias)) = table_name_from_factor(factor)? {
209            let info = catalog
210                .resolve_relation(DatabaseId::DEFAULT, &name)?
211                .ok_or_else(|| SqlError::UnknownTable { name: name.clone() })?;
212            self.add(ResolvedTable { name, alias, info })?;
213        }
214        Ok(())
215    }
216}
217
218/// If `factor` is `ARRAY_*(name, ...)`, look up the array via the
219/// catalog and build a `ResolvedTable` whose columns mirror the array's
220/// dims + attrs. Returns `Ok(None)` for any non-array-TVF factor.
221fn resolve_array_tvf(
222    catalog: &dyn SqlCatalog,
223    factor: &sqlparser::ast::TableFactor,
224) -> Result<Option<ResolvedTable>> {
225    let (fn_name, args, alias) = match factor {
226        sqlparser::ast::TableFactor::Table {
227            name,
228            args: Some(args),
229            alias,
230            ..
231        } => (
232            normalize_object_name_checked(name)?,
233            args,
234            alias.as_ref().map(|a| normalize_ident(&a.name)),
235        ),
236        _ => return Ok(None),
237    };
238    if !matches!(
239        fn_name.as_str(),
240        "array_slice" | "array_project" | "array_agg" | "array_elementwise"
241    ) {
242        return Ok(None);
243    }
244
245    // First positional arg is the array name as a string literal.
246    let first = args.args.first().ok_or_else(|| SqlError::Unsupported {
247        detail: format!("{fn_name}: missing array-name argument"),
248    })?;
249    let array_name = extract_string_literal_arg(first).ok_or_else(|| SqlError::Unsupported {
250        detail: format!("{fn_name}: array-name argument must be a string literal"),
251    })?;
252    let view = catalog
253        .lookup_array(&array_name)
254        .ok_or_else(|| SqlError::UnknownTable {
255            name: array_name.clone(),
256        })?;
257
258    let info = CollectionInfo {
259        name: view.name.clone(),
260        engine: EngineType::Array,
261        columns: array_columns(&view),
262        primary_key: None,
263        has_auto_tier: false,
264        indexes: Vec::new(),
265        bitemporal: false,
266        primary: nodedb_types::PrimaryEngine::Document,
267        vector_primary: None,
268        partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
269    };
270    Ok(Some(ResolvedTable {
271        name: view.name,
272        alias,
273        info,
274    }))
275}
276
277fn array_columns(view: &ArrayCatalogView) -> Vec<ColumnInfo> {
278    let mut cols = Vec::with_capacity(view.dims.len() + view.attrs.len());
279    for d in &view.dims {
280        cols.push(ColumnInfo {
281            name: d.name.clone(),
282            data_type: dim_type_to_sql(d.dtype),
283            nullable: false,
284            is_primary_key: false,
285            default: None,
286            raw_type: None,
287        });
288    }
289    for a in &view.attrs {
290        cols.push(ColumnInfo {
291            name: a.name.clone(),
292            data_type: attr_type_to_sql(a.dtype),
293            nullable: a.nullable,
294            is_primary_key: false,
295            default: None,
296            raw_type: None,
297        });
298    }
299    cols
300}
301
302fn dim_type_to_sql(t: ArrayDimType) -> SqlDataType {
303    match t {
304        ArrayDimType::Int64 => SqlDataType::Int64,
305        ArrayDimType::Float64 => SqlDataType::Float64,
306        ArrayDimType::TimestampMs => SqlDataType::Timestamp,
307        ArrayDimType::String => SqlDataType::String,
308    }
309}
310
311fn attr_type_to_sql(t: ArrayAttrType) -> SqlDataType {
312    match t {
313        ArrayAttrType::Int64 => SqlDataType::Int64,
314        ArrayAttrType::Float64 => SqlDataType::Float64,
315        ArrayAttrType::String => SqlDataType::String,
316        ArrayAttrType::Bytes => SqlDataType::Bytes,
317    }
318}
319
320fn extract_string_literal_arg(arg: &sqlparser::ast::FunctionArg) -> Option<String> {
321    use sqlparser::ast::{Expr, FunctionArg, FunctionArgExpr, Value};
322    let expr = match arg {
323        FunctionArg::Unnamed(FunctionArgExpr::Expr(e)) => e,
324        FunctionArg::Named {
325            arg: FunctionArgExpr::Expr(e),
326            ..
327        } => e,
328        _ => return None,
329    };
330    match expr {
331        Expr::Value(v) => match &v.value {
332            Value::SingleQuotedString(s) => Some(s.clone()),
333            _ => None,
334        },
335        _ => None,
336    }
337}
338
339#[cfg(test)]
340mod tests {
341    use super::*;
342    use crate::types::{CollectionInfo, ColumnInfo, EngineType, SqlDataType};
343    use nodedb_types::PrimaryEngine;
344
345    fn strict_collection(name: &str, columns: Vec<&str>) -> CollectionInfo {
346        CollectionInfo {
347            name: name.into(),
348            engine: EngineType::DocumentStrict,
349            columns: columns
350                .into_iter()
351                .map(|c| ColumnInfo {
352                    name: c.into(),
353                    data_type: SqlDataType::String,
354                    nullable: true,
355                    is_primary_key: false,
356                    default: None,
357                    raw_type: None,
358                })
359                .collect(),
360            primary_key: None,
361            has_auto_tier: false,
362            indexes: Vec::new(),
363            bitemporal: false,
364            primary: PrimaryEngine::Document,
365            vector_primary: None,
366            partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
367        }
368    }
369
370    fn schemaless_collection(name: &str) -> CollectionInfo {
371        CollectionInfo {
372            name: name.into(),
373            engine: EngineType::DocumentSchemaless,
374            columns: Vec::new(),
375            primary_key: None,
376            has_auto_tier: false,
377            indexes: Vec::new(),
378            bitemporal: false,
379            primary: PrimaryEngine::Document,
380            vector_primary: None,
381            partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
382        }
383    }
384
385    fn scope_with(info: CollectionInfo) -> TableScope {
386        let mut scope = TableScope::new();
387        scope
388            .add(ResolvedTable {
389                name: info.name.clone(),
390                alias: None,
391                info,
392            })
393            .expect("add failed");
394        scope
395    }
396
397    /// A double-quoted identifier resolves as a column name (case-preserved).
398    /// `"userId"` is parsed by the SQL layer as `Expr::Identifier` with
399    /// `quote_style = Some('"')` and `value = "userId"`.  At the
400    /// `TableScope` level the column name arrives lowercase (strict schema
401    /// columns are stored lowercase), so the resolved name is `"userid"`.
402    ///
403    /// This test confirms the resolution path, not just `convert_expr`.
404    #[test]
405    fn quoted_identifier_resolves_as_column() {
406        let scope = scope_with(strict_collection("users", vec!["userid", "email"]));
407        let (table, col) = scope
408            .resolve_column(None, "userid")
409            .expect("should resolve");
410        assert_eq!(table, "users");
411        assert_eq!(col, "userid");
412    }
413
414    /// An unrecognized column in a strict collection must yield
415    /// `SqlError::UnknownColumn`, NOT `SqlError::Unsupported`.
416    /// This verifies that a double-quoted identifier like `"ghost_col"`
417    /// that maps to `SqlExpr::Column { name: "ghost_col" }` surfaces the
418    /// right error variant when resolved against a strict schema.
419    #[test]
420    fn unknown_column_in_strict_collection_yields_unknown_column_error() {
421        let scope = scope_with(strict_collection("users", vec!["id", "email"]));
422        let err = scope
423            .resolve_column(None, "ghost_col")
424            .expect_err("should fail for unknown column");
425        assert!(
426            matches!(err, SqlError::UnknownColumn { ref column, .. } if column == "ghost_col"),
427            "expected UnknownColumn(ghost_col), got {err:?}"
428        );
429        // Must NOT be Unsupported — that would be the wrong error variant.
430        assert!(
431            !matches!(err, SqlError::Unsupported { .. }),
432            "must not surface Unsupported for a missing column"
433        );
434    }
435
436    /// Schemaless collections accept any column, including ones that look
437    /// like they could be misidentified double-quoted identifiers.
438    #[test]
439    fn any_column_accepted_in_schemaless_collection() {
440        let scope = scope_with(schemaless_collection("events"));
441        let (table, col) = scope
442            .resolve_column(None, "ghost_col")
443            .expect("schemaless should accept any column");
444        assert_eq!(table, "events");
445        assert_eq!(col, "ghost_col");
446    }
447
448    /// Qualified column reference: `"t"."col"` → table `t`, column `col`.
449    #[test]
450    fn qualified_column_resolves_correctly() {
451        let scope = scope_with(strict_collection("t", vec!["col", "other"]));
452        let (table, col) = scope
453            .resolve_column(Some("t"), "col")
454            .expect("qualified column should resolve");
455        assert_eq!(table, "t");
456        assert_eq!(col, "col");
457    }
458
459    /// Qualified reference to an unknown column in a strict collection must
460    /// yield `SqlError::UnknownColumn`, not `Unsupported`.
461    #[test]
462    fn qualified_unknown_column_in_strict_collection() {
463        let scope = scope_with(strict_collection("t", vec!["id"]));
464        let err = scope
465            .resolve_column(Some("t"), "missing")
466            .expect_err("should fail");
467        assert!(
468            matches!(err, SqlError::UnknownColumn { .. }),
469            "expected UnknownColumn, got {err:?}"
470        );
471    }
472}
nodedb_sql/resolver/columns.rs

nodedb_sql/resolver/
columns.rs
