1use sqlparser::ast::{self, Expr, SelectItem, SelectItemQualifiedWildcardKind, SetExpr};
6
7use crate::error::{Result, SqlError};
8use crate::functions::registry::FunctionRegistry;
9use crate::parser::normalize::{normalize_ident, normalize_object_name_checked};
10use crate::resolver::columns::TableScope;
11use crate::temporal::TemporalScope;
12use crate::types::{CollectionInfo, ColumnInfo, EngineType, SqlCatalog, SqlDataType, SqlPlan};
13
14pub fn infer_subquery_relation(
27 catalog: &dyn SqlCatalog,
28 alias: &str,
29 query: &ast::Query,
30 plan: Option<&SqlPlan>,
31 functions: &FunctionRegistry,
32 temporal: TemporalScope,
33) -> Result<CollectionInfo> {
34 let (mut columns, open_schema) = infer_projection(catalog, query, functions, temporal)?;
35 if plan.is_some_and(SqlPlan::carries_search_cells) {
43 if !columns.iter().any(|c| c.name == "distance") {
44 columns.push(synthetic_column("distance"));
45 }
46 if !columns.iter().any(|c| c.name == "_surrogate") {
47 columns.push(synthetic_column("_surrogate"));
48 }
49 }
50 Ok(CollectionInfo {
51 name: alias.to_string(),
52 engine: EngineType::DocumentSchemaless,
53 columns,
54 primary_key: Some("id".into()),
55 has_auto_tier: false,
56 indexes: Vec::new(),
57 bitemporal: false,
58 primary: nodedb_types::PrimaryEngine::Document,
59 vector_primary: None,
60 partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
61 open_schema,
62 })
63}
64
65pub fn open_subquery_relation(alias: &str) -> CollectionInfo {
71 CollectionInfo {
72 name: alias.to_string(),
73 engine: EngineType::DocumentSchemaless,
74 columns: Vec::new(),
75 primary_key: Some("id".into()),
76 has_auto_tier: false,
77 indexes: Vec::new(),
78 bitemporal: false,
79 primary: nodedb_types::PrimaryEngine::Document,
80 vector_primary: None,
81 partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
82 open_schema: true,
83 }
84}
85
86pub fn rename_output_columns(mut info: CollectionInfo, names: &[String]) -> CollectionInfo {
92 if names.is_empty() {
93 return info;
94 }
95 let inferred = std::mem::take(&mut info.columns);
96 info.columns = names
97 .iter()
98 .enumerate()
99 .map(|(index, name)| match inferred.get(index) {
100 Some(column) => ColumnInfo {
101 name: name.clone(),
102 ..column.clone()
103 },
104 None => synthetic_column(name),
105 })
106 .collect();
107 info
108}
109
110fn infer_projection(
113 catalog: &dyn SqlCatalog,
114 query: &ast::Query,
115 functions: &FunctionRegistry,
116 temporal: TemporalScope,
117) -> Result<(Vec<ColumnInfo>, bool)> {
118 infer_body(catalog, &query.body, functions, temporal)
119}
120
121fn infer_body(
122 catalog: &dyn SqlCatalog,
123 body: &SetExpr,
124 functions: &FunctionRegistry,
125 temporal: TemporalScope,
126) -> Result<(Vec<ColumnInfo>, bool)> {
127 match body {
128 SetExpr::Select(select) => infer_select_projection(catalog, select, functions, temporal),
129 SetExpr::Query(query) => infer_projection(catalog, query, functions, temporal),
130 SetExpr::SetOperation { left, .. } => infer_body(catalog, left, functions, temporal),
132 SetExpr::Values(_)
135 | SetExpr::Table(_)
136 | SetExpr::Insert(_)
137 | SetExpr::Update(_)
138 | SetExpr::Delete(_)
139 | SetExpr::Merge(_) => Ok((Vec::new(), true)),
140 }
141}
142
143fn infer_select_projection(
144 catalog: &dyn SqlCatalog,
145 select: &ast::Select,
146 functions: &FunctionRegistry,
147 temporal: TemporalScope,
148) -> Result<(Vec<ColumnInfo>, bool)> {
149 let scope = TableScope::resolve_from(catalog, functions, temporal, &select.from)?;
150 let mut columns = Vec::new();
151 let mut open = false;
152
153 for item in &select.projection {
154 match item {
155 SelectItem::Wildcard(_) => {
156 for table in scope.tables_in_order() {
157 columns.extend(table.info.columns.iter().cloned());
158 open |= table.info.open_schema;
159 }
160 }
161 SelectItem::QualifiedWildcard(kind, _) => {
162 let table_ref = match kind {
163 SelectItemQualifiedWildcardKind::ObjectName(name) => {
164 normalize_object_name_checked(name)?
165 }
166 SelectItemQualifiedWildcardKind::Expr(_) => {
169 open = true;
170 continue;
171 }
172 };
173 match scope.table_by_ref(&table_ref) {
174 Some(table) => {
175 columns.extend(table.info.columns.iter().cloned());
176 open |= table.info.open_schema;
177 }
178 None => open = true,
180 }
181 }
182 SelectItem::ExprWithAlias { alias, .. } => {
183 columns.push(synthetic_column(&normalize_ident(alias)));
184 }
185 SelectItem::UnnamedExpr(expr) => columns.push(unnamed_column(&scope, expr)),
186 SelectItem::ExprWithAliases { aliases, .. } => {
187 return Err(SqlError::Unsupported {
188 detail: format!(
189 "multi-alias projection ('AS' with {} names) is not supported; \
190 give the expression a single alias",
191 aliases.len()
192 ),
193 });
194 }
195 }
196 }
197
198 Ok((columns, open))
199}
200
201fn unnamed_column(scope: &TableScope, expr: &Expr) -> ColumnInfo {
207 let named = match expr {
208 Expr::Identifier(ident) => Some((None, normalize_ident(ident))),
209 Expr::CompoundIdentifier(parts) if parts.len() == 2 => {
210 Some((Some(normalize_ident(&parts[0])), normalize_ident(&parts[1])))
211 }
212 _ => None,
213 };
214 match named {
215 Some((qualifier, name)) => match scope.declared_column(qualifier.as_deref(), &name) {
216 Some(column) => ColumnInfo {
217 name,
218 ..column.clone()
219 },
220 None => synthetic_column(&name),
221 },
222 None => synthetic_column(&format!("{expr}").to_lowercase()),
223 }
224}
225
226fn synthetic_column(name: &str) -> ColumnInfo {
228 ColumnInfo {
229 name: name.to_string(),
230 data_type: SqlDataType::Unknown,
231 nullable: true,
232 is_primary_key: false,
233 default: None,
234 raw_type: None,
235 int_width: None,
236 float_width: None,
237 }
238}
239
240pub fn from_relation_names(from: &[ast::TableWithJoins]) -> Vec<String> {
243 let mut names = Vec::new();
244 let factors = from.iter().flat_map(|twj| {
245 std::iter::once(&twj.relation).chain(twj.joins.iter().map(|j| &j.relation))
246 });
247 for factor in factors {
248 match factor {
249 ast::TableFactor::Table { name, alias, .. } => match alias {
250 Some(alias) => names.push(normalize_ident(&alias.name)),
251 None => {
252 if let Ok(name) = normalize_object_name_checked(name) {
253 names.push(name);
254 }
255 }
256 },
257 ast::TableFactor::Derived {
258 alias: Some(alias), ..
259 } => names.push(normalize_ident(&alias.name)),
260 _ => {}
261 }
262 }
263 names
264}
265
266pub fn refuse_correlated_derived(
273 err: SqlError,
274 factor: &ast::TableFactor,
275 from_names: &[String],
276) -> SqlError {
277 let ast::TableFactor::Derived { lateral: false, .. } = factor else {
278 return err;
279 };
280 match err {
281 SqlError::UnknownTable { name } if from_names.contains(&name) => SqlError::Unsupported {
282 detail: format!(
283 "invalid reference to FROM-clause entry for table \"{name}\": a subquery \
284 in FROM cannot refer to other relations of the same query level; write \
285 JOIN LATERAL (SELECT ...) to correlate it"
286 ),
287 },
288 other => other,
289 }
290}
291
292#[cfg(test)]
293mod tests {
294 use super::*;
295 use crate::types::SqlCatalogError;
296 use nodedb_types::DatabaseId;
297 use sqlparser::ast::Statement;
298
299 struct TestCatalog;
300
301 fn strict(name: &str, columns: &[&str]) -> CollectionInfo {
302 CollectionInfo {
303 name: name.into(),
304 engine: EngineType::DocumentStrict,
305 columns: columns
306 .iter()
307 .map(|c| ColumnInfo {
308 name: (*c).into(),
309 data_type: SqlDataType::Int64,
310 nullable: false,
311 is_primary_key: false,
312 default: None,
313 raw_type: None,
314 int_width: None,
315 float_width: None,
316 })
317 .collect(),
318 primary_key: Some("a".into()),
319 has_auto_tier: false,
320 indexes: Vec::new(),
321 bitemporal: false,
322 primary: nodedb_types::PrimaryEngine::Document,
323 vector_primary: None,
324 partition_strategy: nodedb_types::PartitionStrategy::CollectionHomed,
325 open_schema: CollectionInfo::open_schema_for(EngineType::DocumentStrict),
326 }
327 }
328
329 impl SqlCatalog for TestCatalog {
330 fn get_collection(
331 &self,
332 _database_id: DatabaseId,
333 name: &str,
334 ) -> std::result::Result<Option<CollectionInfo>, SqlCatalogError> {
335 Ok(match name {
336 "src" => Some(strict("src", &["a", "b"])),
337 "loose" => {
338 let mut info = strict("loose", &["x"]);
339 info.engine = EngineType::DocumentSchemaless;
340 info.open_schema = true;
341 Some(info)
342 }
343 _ => None,
344 })
345 }
346 }
347
348 fn parse_query(sql: &str) -> ast::Query {
349 let stmts = crate::parser::statement::parse_sql(sql).expect("parse failed");
350 match &stmts[0] {
351 Statement::Query(query) => (**query).clone(),
352 other => panic!("expected a query, got {other:?}"),
353 }
354 }
355
356 fn infer(sql: &str) -> CollectionInfo {
357 let query = parse_query(sql);
358 let functions = crate::functions::registry::FunctionRegistry::new();
359 let plan = crate::planner::select::plan_query(
360 &query,
361 &TestCatalog,
362 &functions,
363 TemporalScope::default(),
364 )
365 .expect("plan failed");
366 infer_subquery_relation(
367 &TestCatalog,
368 "t",
369 &query,
370 Some(&plan),
371 &functions,
372 TemporalScope::default(),
373 )
374 .expect("inference failed")
375 }
376
377 #[test]
378 fn bare_identifier_keeps_the_declared_type() {
379 let info = infer("SELECT a FROM src");
380 assert_eq!(info.columns.len(), 1);
381 assert_eq!(info.columns[0].name, "a");
382 assert_eq!(info.columns[0].data_type, SqlDataType::Int64);
383 assert!(!info.open_schema);
384 }
385
386 #[test]
387 fn qualified_identifier_exposes_the_unqualified_name() {
388 let info = infer("SELECT i.b FROM src AS i");
389 assert_eq!(info.columns.len(), 1);
390 assert_eq!(info.columns[0].name, "b");
391 }
392
393 #[test]
394 fn alias_names_the_output_column() {
395 let info = infer("SELECT a + 1 AS total FROM src");
396 assert_eq!(info.columns[0].name, "total");
397 assert_eq!(info.columns[0].data_type, SqlDataType::Unknown);
398 }
399
400 #[test]
401 fn star_over_a_closed_source_stays_closed() {
402 let info = infer("SELECT * FROM src");
403 let names: Vec<&str> = info.columns.iter().map(|c| c.name.as_str()).collect();
404 assert_eq!(names, vec!["a", "b"]);
405 assert!(!info.open_schema);
406 }
407
408 #[test]
409 fn star_over_an_open_source_stays_open() {
410 let info = infer("SELECT * FROM loose");
411 assert!(info.open_schema);
412 }
413
414 #[test]
415 fn set_operation_takes_the_left_arm_shape() {
416 let info = infer("SELECT a FROM src UNION ALL SELECT b FROM src");
417 assert_eq!(info.columns.len(), 1);
418 assert_eq!(info.columns[0].name, "a");
419 }
420
421 #[test]
422 fn alias_column_list_renames_positionally() {
423 let info = rename_output_columns(infer("SELECT a, b FROM src"), &["p".into(), "q".into()]);
424 let names: Vec<&str> = info.columns.iter().map(|c| c.name.as_str()).collect();
425 assert_eq!(names, vec!["p", "q"]);
426 assert_eq!(info.columns[0].data_type, SqlDataType::Int64);
427 }
428
429 #[test]
430 fn vector_search_projection_declares_the_synthetic_distance_column() {
431 let info = infer("SELECT * FROM src ORDER BY vector_distance(b, ARRAY[0.1, 0.2]) LIMIT 2");
436 let names: Vec<&str> = info.columns.iter().map(|c| c.name.as_str()).collect();
437 assert!(names.contains(&"distance"), "columns: {names:?}");
438 }
439
440 #[test]
441 fn plain_ordered_projection_has_no_distance_column() {
442 let info = infer("SELECT * FROM src ORDER BY b LIMIT 2");
443 let names: Vec<&str> = info.columns.iter().map(|c| c.name.as_str()).collect();
444 assert!(!names.contains(&"distance"), "columns: {names:?}");
445 }
446
447 #[test]
450 fn every_vector_search_function_declares_the_distance_column() {
451 for call in [
452 "vector_distance(b, ARRAY[0.1])",
453 "vector_cosine_distance(b, ARRAY[0.1])",
454 "vector_neg_inner_product(b, ARRAY[0.1])",
455 ] {
456 let info = infer(&format!("SELECT * FROM src ORDER BY {call} LIMIT 2"));
457 let names: Vec<&str> = info.columns.iter().map(|c| c.name.as_str()).collect();
458 assert!(names.contains(&"distance"), "{call}: columns {names:?}");
459 }
460 }
461}