1use crate::ast::{
10 ColumnType, FunctionBinding, FunctionParamMode, FunctionReturns, RoutineInvocationBinding,
11};
12use crate::routines::{routine_returns_anonymous_record, RoutineResolution, SQLUserFunction};
13use crate::type_resolution::{BuiltinFunctionOverload, FunctionTypeResolver};
14use crate::{RowSchema, SQLError, SQLParam, ScalarExpr};
15use std::sync::Arc;
16
17pub struct ResolvedUserTableFunction {
18 pub function: Arc<SQLUserFunction>,
19 pub binding: FunctionBinding,
20}
21
22pub fn user_function_output_columns_for(function: &SQLUserFunction) -> Option<Vec<String>> {
23 let outputs = function.def.output_params();
24 if outputs.is_empty() {
25 return None;
26 }
27 Some(
28 outputs
29 .iter()
30 .enumerate()
31 .map(|(position, parameter)| {
32 if parameter.name.is_empty() {
33 format!("column{}", position + 1)
34 } else {
35 parameter.name.clone()
36 }
37 })
38 .collect(),
39 )
40}
41
42fn validate_user_table_function_column_definition(
43 function: &SQLUserFunction,
44 declared_types: &[String],
45) -> Result<(), SQLError> {
46 let returns_anonymous_record = routine_returns_anonymous_record(&function.def);
47 if declared_types.is_empty() {
48 if returns_anonymous_record {
49 return Err(SQLError::Routine {
50 sqlstate: "42601".into(),
51 message: "a column definition list is required for functions returning \"record\""
52 .into(),
53 });
54 }
55 return Ok(());
56 }
57 if returns_anonymous_record {
58 return Ok(());
59 }
60 if function.def.output_params().len() > 1 {
61 return Err(redundant_out_column_definition_error());
62 }
63 Err(SQLError::Routine {
64 sqlstate: "42601".into(),
65 message: "a column definition list is only allowed for functions returning \"record\""
66 .into(),
67 })
68}
69
70pub fn validate_table_function_column_definition(
71 name: &str,
72 binding: Option<&FunctionBinding>,
73 user_function: Option<&SQLUserFunction>,
74 declared_types: &[String],
75) -> Result<(), SQLError> {
76 if let Some(function) = user_function {
77 return validate_user_table_function_column_definition(function, declared_types);
78 }
79 if declared_types.is_empty() || binding.is_some_and(|binding| !binding.builtin) {
80 return Ok(());
81 }
82 let builtin = crate::semantics::builtin_function_dispatch_name(&name.to_ascii_lowercase());
83 if matches!(
84 builtin.as_str(),
85 "json_each"
86 | "jsonb_each"
87 | "json_each_text"
88 | "jsonb_each_text"
89 | "pg_get_sequence_data"
90 | "pg_sequence_parameters"
91 ) {
92 return Err(redundant_out_column_definition_error());
93 }
94 Ok(())
95}
96
97fn redundant_out_column_definition_error() -> SQLError {
98 SQLError::Routine {
99 sqlstate: "42601".into(),
100 message: "a column definition list is redundant for a function with OUT parameters".into(),
101 }
102}
103
104pub fn resolve_user_table_function(
105 routines: &dyn RoutineResolution,
106 name: &str,
107 binding: Option<&FunctionBinding>,
108 args: &[ScalarExpr],
109 input_schema: &RowSchema,
110 params: &[SQLParam],
111 resolver: &dyn FunctionTypeResolver,
112) -> Result<Option<ResolvedUserTableFunction>, SQLError> {
113 let Some(binding) = resolve_table_function_binding(
114 routines,
115 name,
116 binding,
117 args,
118 input_schema,
119 params,
120 resolver,
121 )?
122 else {
123 return Ok(None);
124 };
125 if binding.builtin {
126 return Ok(None);
127 }
128 let (argument_names, argument_types, explicit_variadic) =
129 crate::function_call_argument_signature(args, input_schema, params, Some(resolver))?;
130 let Some(matched) = routines.resolve_static_sql_function_match(
131 name,
132 Some(&binding),
133 &argument_names,
134 &argument_types,
135 explicit_variadic,
136 )?
137 else {
138 return Ok(None);
139 };
140 Ok(Some(ResolvedUserTableFunction {
141 binding: matched.binding(),
142 function: matched.function,
143 }))
144}
145
146pub fn resolve_table_function_binding(
147 routines: &dyn RoutineResolution,
148 name: &str,
149 binding: Option<&FunctionBinding>,
150 args: &[ScalarExpr],
151 input_schema: &RowSchema,
152 params: &[SQLParam],
153 resolver: &dyn FunctionTypeResolver,
154) -> Result<Option<FunctionBinding>, SQLError> {
155 if let Some(binding) = binding {
156 return Ok(Some(binding.clone()));
157 }
158 let identity = name.to_ascii_lowercase();
159 let builtin = crate::semantics::builtin_function_dispatch_name(&identity);
160 let (argument_names, argument_types, explicit_variadic) =
161 crate::function_call_argument_signature(args, input_schema, params, Some(resolver))?;
162 let builtins = builtin_table_function_overloads(&builtin, &argument_types);
163 if !builtins.is_empty() || has_builtin_table_function_overloads(&builtin) {
164 return routines
165 .resolve_table_function_overload_with_builtins(
166 name,
167 None,
168 &argument_names,
169 &argument_types,
170 explicit_variadic,
171 &builtins,
172 )
173 .map(|resolved| resolved.map(|resolved| resolved.binding));
174 }
175 let builtin_surface = is_builtin_table_function(&builtin)
176 || crate::registry::is_operator_join_table_function(&builtin)
177 || routines.has_registered_table_function(&identity);
178 if routines.lookup_visible_sql_functions(name)?.is_none() {
179 return Ok(None);
180 }
181 match routines.resolve_static_sql_function_match(
182 name,
183 None,
184 &argument_names,
185 &argument_types,
186 explicit_variadic,
187 ) {
188 Ok(Some(function)) => Ok(Some(function.binding())),
189 Ok(None) => Ok(None),
190 Err(error) if builtin_surface && error.sqlstate() == Some("42883") => Ok(None),
191 Err(error) => Err(error),
192 }
193}
194
195fn has_builtin_table_function_overloads(name: &str) -> bool {
196 matches!(
197 name,
198 "generate_series"
199 | "pg_listening_channels"
200 | "unnest"
201 | "regexp_split_to_table"
202 | "string_to_table"
203 | "json_array_elements"
204 | "jsonb_array_elements"
205 | "json_array_elements_text"
206 | "jsonb_array_elements_text"
207 | "json_object_keys"
208 | "jsonb_object_keys"
209 | "json_each"
210 | "jsonb_each"
211 | "json_each_text"
212 | "jsonb_each_text"
213 | "pg_get_sequence_data"
214 | "pg_sequence_parameters"
215 )
216}
217
218fn builtin_table_function_overloads(
219 name: &str,
220 argument_types: &[Option<ColumnType>],
221) -> Vec<BuiltinFunctionOverload> {
222 let canonical_name = format!("pg_catalog.{name}");
223 let overload = |argument_types: Vec<ColumnType>,
224 default_arguments: usize,
225 return_type: ColumnType| BuiltinFunctionOverload {
226 name: canonical_name.clone(),
227 argument_names: vec![None; argument_types.len()],
228 argument_types,
229 default_arguments,
230 return_type,
231 };
232 match name {
233 "pg_listening_channels" => vec![overload(Vec::new(), 0, ColumnType::Text)],
234 "generate_series" => vec![
235 overload(
236 vec![
237 ColumnType::Integer,
238 ColumnType::Integer,
239 ColumnType::Integer,
240 ],
241 1,
242 ColumnType::Integer,
243 ),
244 overload(
245 vec![
246 ColumnType::BigInteger,
247 ColumnType::BigInteger,
248 ColumnType::BigInteger,
249 ],
250 1,
251 ColumnType::BigInteger,
252 ),
253 ],
254 "unnest" => {
255 let [Some(argument)] = argument_types else {
256 return Vec::new();
257 };
258 let Some(element) = crate::type_resolution::array_element_type(argument) else {
259 return Vec::new();
260 };
261 vec![overload(vec![ColumnType::AnyArray], 0, element.clone())]
262 }
263 "regexp_split_to_table" | "string_to_table" => vec![overload(
264 vec![ColumnType::Text, ColumnType::Text],
265 0,
266 ColumnType::Text,
267 )],
268 "json_array_elements" => vec![overload(vec![ColumnType::Json], 0, ColumnType::Json)],
269 "jsonb_array_elements" => vec![overload(vec![ColumnType::JsonB], 0, ColumnType::JsonB)],
270 "json_array_elements_text" | "json_object_keys" => {
271 vec![overload(vec![ColumnType::Json], 0, ColumnType::Text)]
272 }
273 "jsonb_array_elements_text" | "jsonb_object_keys" => {
274 vec![overload(vec![ColumnType::JsonB], 0, ColumnType::Text)]
275 }
276 "json_each" | "json_each_text" => {
277 vec![overload(vec![ColumnType::Json], 0, ColumnType::Record)]
278 }
279 "jsonb_each" | "jsonb_each_text" => {
280 vec![overload(vec![ColumnType::JsonB], 0, ColumnType::Record)]
281 }
282 "pg_get_sequence_data" => vec![overload(vec![ColumnType::Regclass], 0, ColumnType::Record)],
283 "pg_sequence_parameters" => {
284 vec![overload(vec![ColumnType::Oid], 0, ColumnType::Record)]
285 }
286 _ => Vec::new(),
287 }
288}
289
290pub fn is_builtin_table_function(name: &str) -> bool {
291 matches!(
292 name,
293 "generate_series"
294 | "pg_listening_channels"
295 | "unnest"
296 | "regexp_split_to_table"
297 | "string_to_table"
298 | "json_array_elements"
299 | "jsonb_array_elements"
300 | "json_array_elements_text"
301 | "jsonb_array_elements_text"
302 | "json_object_keys"
303 | "jsonb_object_keys"
304 | "json_each"
305 | "jsonb_each"
306 | "json_each_text"
307 | "jsonb_each_text"
308 | "pg_get_sequence_data"
309 | "pg_sequence_parameters"
310 | "create_analyzer"
311 | "drop_analyzer"
312 | "list_analyzers"
313 | "analyze_text"
314 | "fts_index_stats"
315 | "set_table_analyzer"
316 | "pagerank"
317 | "graph_pagerank"
318 | "hits"
319 | "graph_hits"
320 | "betweenness"
321 | "graph_betweenness"
322 | "graph_edges"
323 | "rpq"
324 | "cypher"
325 )
326}
327
328pub fn table_function_empty_schema(
329 name: &str,
330 output_name: &str,
331 alias: Option<&str>,
332 column_aliases: &[String],
333 output_width: usize,
334 ordinality: bool,
335) -> Vec<String> {
336 let lower = crate::semantics::builtin_function_dispatch_name(&name.to_ascii_lowercase());
337 let columns = if lower == "unnest" {
338 let width = output_width.max(1);
339 let default_column = if width == 1 {
340 alias.unwrap_or(output_name)
341 } else {
342 output_name
343 };
344 vec![default_column.to_string(); width]
345 } else {
346 match lower.as_str() {
347 "json_each" | "jsonb_each" | "json_each_text" | "jsonb_each_text" => {
348 vec!["key".into(), "value".into()]
349 }
350 "pg_get_sequence_data" => vec!["last_value".into(), "is_called".into()],
351 "pg_sequence_parameters" => vec![
352 "start_value".into(),
353 "minimum_value".into(),
354 "maximum_value".into(),
355 "increment".into(),
356 "cycle_option".into(),
357 "cache_size".into(),
358 "data_type".into(),
359 ],
360 "pagerank" | "graph_pagerank" | "hits" | "graph_hits" | "betweenness"
361 | "graph_betweenness" => vec!["_doc_id".into(), "_score".into()],
362 "rpq" => vec!["vertex_id".into()],
363 "list_analyzers" => vec!["analyzer_name".into()],
364 "analyze_text" => vec!["analysis".into()],
365 "fts_index_stats" => vec![
366 "table_name".into(),
367 "field".into(),
368 "analyzer".into(),
369 "posting_count".into(),
370 "doc_length_count".into(),
371 "indexed_doc_count".into(),
372 "term_count".into(),
373 "total_field_length".into(),
374 ],
375 "text_similarity_join"
376 | "vector_similarity_join"
377 | "graph_join"
378 | "hybrid_join"
379 | "cross_paradigm_join" => {
380 vec!["left_doc_id".into(), "right_doc_id".into(), "_score".into()]
381 }
382 "generate_series"
383 | "pg_listening_channels"
384 | "regexp_split_to_table"
385 | "string_to_table"
386 | "json_array_elements"
387 | "jsonb_array_elements"
388 | "json_array_elements_text"
389 | "jsonb_array_elements_text"
390 | "json_object_keys"
391 | "jsonb_object_keys" => vec![scalar_table_function_default_column(
392 &lower,
393 output_name,
394 alias,
395 &[],
396 )],
397 _ => {
398 let minimum_width = 1;
399 let aliased_value_width = if ordinality && column_aliases.len() > minimum_width {
400 column_aliases.len() - 1
401 } else {
402 column_aliases.len()
403 };
404 vec![
405 scalar_table_function_default_column(&lower, output_name, alias, &[]);
406 minimum_width.max(aliased_value_width)
407 ]
408 }
409 }
410 };
411 apply_table_function_aliases(columns, column_aliases, ordinality)
412}
413
414pub fn apply_table_function_aliases(
415 mut columns: Vec<String>,
416 column_aliases: &[String],
417 ordinality: bool,
418) -> Vec<String> {
419 if ordinality {
420 columns.push("ordinality".into());
421 }
422 for (column, alias) in columns.iter_mut().zip(column_aliases) {
423 column.clone_from(alias);
424 }
425 columns
426}
427
428pub fn validate_table_function_alias_count(
429 table_alias: &str,
430 available: usize,
431 specified: usize,
432) -> Result<(), SQLError> {
433 if specified <= available {
434 return Ok(());
435 }
436 Err(SQLError::Routine {
437 sqlstate: "42P10".into(),
438 message: format!(
439 "table \"{table_alias}\" has {available} columns available but {specified} columns specified"
440 ),
441 })
442}
443
444pub struct TableFunctionTypeRequest<'a> {
445 pub name: &'a str,
446 pub args: &'a [ScalarExpr],
447 pub user_function: Option<&'a SQLUserFunction>,
448 pub user_invocation: Option<&'a RoutineInvocationBinding>,
449 pub declared_types: &'a [String],
450 pub columns: &'a [String],
451 pub ordinality: bool,
452}
453
454#[expect(
455 clippy::too_many_lines,
456 reason = "preserves source schema and row identity"
457)]
458pub fn table_function_column_types(
459 routines: &dyn RoutineResolution,
460 request: TableFunctionTypeRequest<'_>,
461 input_schema: &crate::RowSchema,
462 params: &[SQLParam],
463 resolver: &dyn FunctionTypeResolver,
464) -> Vec<Option<ColumnType>> {
465 let TableFunctionTypeRequest {
466 name,
467 args,
468 user_function,
469 user_invocation,
470 declared_types,
471 columns,
472 ordinality,
473 } = request;
474 let value_columns = if ordinality {
475 columns
476 .get(..columns.len().saturating_sub(1))
477 .unwrap_or(&[])
478 } else {
479 columns
480 };
481 let align = |types: Vec<Option<ColumnType>>| {
482 if types.len() == value_columns.len() {
483 types
484 } else if let [ty] = types.as_slice() {
485 vec![ty.clone(); value_columns.len()]
486 } else {
487 vec![None; value_columns.len()]
488 }
489 };
490 let mut types = if !declared_types.is_empty() {
491 align(
492 declared_types
493 .iter()
494 .map(|ty| ColumnType::from_sql_name(ty).ok())
495 .collect(),
496 )
497 } else if let Some(function) = user_function {
498 align(user_function_column_types(
499 routines,
500 function,
501 user_invocation,
502 ))
503 } else {
504 let normalized =
505 crate::semantics::builtin_function_dispatch_name(&name.to_ascii_lowercase());
506 let argument_type = |position: usize| {
507 args.get(position)
508 .and_then(|argument| {
509 crate::scalar_type_with_resolver(argument, input_schema, params, resolver).ok()
510 })
511 .flatten()
512 };
513 align(match normalized.as_str() {
514 "pg_listening_channels" => vec![Some(ColumnType::Text)],
515 "generate_series" => vec![argument_type(0)],
516 "analyze_text" => vec![Some(ColumnType::JsonB)],
517 "unnest" => args
518 .iter()
519 .map(|argument| {
520 crate::scalar_type_with_resolver(argument, input_schema, params, resolver)
521 .ok()
522 .flatten()
523 .and_then(|ty| crate::type_resolution::array_element_type(&ty).cloned())
524 })
525 .collect(),
526 "regexp_split_to_table"
527 | "string_to_table"
528 | "json_object_keys"
529 | "jsonb_object_keys" => vec![Some(ColumnType::Text)],
530 "json_array_elements" => vec![Some(ColumnType::Json)],
531 "jsonb_array_elements" => vec![Some(ColumnType::JsonB)],
532 "json_array_elements_text" | "jsonb_array_elements_text" => {
533 vec![Some(ColumnType::Text)]
534 }
535 "json_each" => vec![Some(ColumnType::Text), Some(ColumnType::Json)],
536 "jsonb_each" => vec![Some(ColumnType::Text), Some(ColumnType::JsonB)],
537 "json_each_text" | "jsonb_each_text" => {
538 vec![Some(ColumnType::Text), Some(ColumnType::Text)]
539 }
540 "pg_get_sequence_data" => {
541 vec![Some(ColumnType::BigInteger), Some(ColumnType::Boolean)]
542 }
543 "pg_sequence_parameters" => vec![
544 Some(ColumnType::BigInteger),
545 Some(ColumnType::BigInteger),
546 Some(ColumnType::BigInteger),
547 Some(ColumnType::BigInteger),
548 Some(ColumnType::Boolean),
549 Some(ColumnType::BigInteger),
550 Some(ColumnType::Oid),
551 ],
552 "pagerank" | "graph_pagerank" | "hits" | "graph_hits" | "betweenness"
553 | "graph_betweenness" => vec![
554 Some(ColumnType::BigInteger),
555 Some(ColumnType::DoublePrecision),
556 ],
557 "rpq" => vec![Some(ColumnType::BigInteger)],
558 "text_similarity_join"
559 | "vector_similarity_join"
560 | "graph_join"
561 | "hybrid_join"
562 | "cross_paradigm_join" => vec![
563 Some(ColumnType::BigInteger),
564 Some(ColumnType::BigInteger),
565 Some(ColumnType::DoublePrecision),
566 ],
567 _ => user_table_function_column_types(
568 routines,
569 name,
570 args,
571 input_schema,
572 params,
573 resolver,
574 ),
575 })
576 };
577 if ordinality {
578 types.push(Some(ColumnType::BigInteger));
579 }
580 types
581}
582
583fn user_table_function_column_types(
584 routines: &dyn RoutineResolution,
585 name: &str,
586 args: &[ScalarExpr],
587 input_schema: &crate::RowSchema,
588 params: &[SQLParam],
589 resolver: &dyn FunctionTypeResolver,
590) -> Vec<Option<ColumnType>> {
591 let Ok((argument_names, argument_types, explicit_variadic)) =
592 crate::function_call_argument_signature(args, input_schema, params, Some(resolver))
593 else {
594 return Vec::new();
595 };
596 let Ok(Some(matched)) = routines.resolve_static_sql_function_match(
597 name,
598 None,
599 &argument_names,
600 &argument_types,
601 explicit_variadic,
602 ) else {
603 return Vec::new();
604 };
605 user_function_column_types(routines, &matched.function, Some(&matched.invocation))
606}
607
608fn user_function_column_types(
609 routines: &dyn RoutineResolution,
610 function: &SQLUserFunction,
611 invocation: Option<&RoutineInvocationBinding>,
612) -> Vec<Option<ColumnType>> {
613 let outputs = function
614 .def
615 .params
616 .iter()
617 .enumerate()
618 .filter(|(_, parameter)| {
619 matches!(
620 parameter.mode,
621 FunctionParamMode::Out | FunctionParamMode::InOut | FunctionParamMode::Table
622 )
623 })
624 .collect::<Vec<_>>();
625 if !outputs.is_empty() {
626 return outputs
627 .into_iter()
628 .map(|(index, parameter)| {
629 let type_name = invocation
630 .and_then(|binding| binding.parameter_types.get(index))
631 .unwrap_or(¶meter.type_name);
632 resolve_table_function_column_type(routines, type_name)
633 })
634 .collect();
635 }
636 match &function.def.returns {
637 FunctionReturns::Scalar { type_name } | FunctionReturns::SetOf { type_name } => {
638 let type_name = invocation
639 .and_then(|binding| binding.return_type.as_ref())
640 .unwrap_or(type_name);
641 vec![resolve_table_function_column_type(routines, type_name)]
642 }
643 FunctionReturns::None | FunctionReturns::Table => Vec::new(),
644 }
645}
646
647fn resolve_table_function_column_type(
648 routines: &dyn RoutineResolution,
649 type_name: &str,
650) -> Option<ColumnType> {
651 routines
652 .resolve_type_name(type_name)
653 .ok()
654 .flatten()
655 .or_else(|| ColumnType::from_sql_name(type_name).ok())
656}
657
658pub fn is_json_array_table_function(name: &str) -> bool {
659 matches!(
660 name,
661 "json_array_elements"
662 | "jsonb_array_elements"
663 | "json_array_elements_text"
664 | "jsonb_array_elements_text"
665 )
666}
667
668pub fn scalar_table_function_default_column(
669 normalized_name: &str,
670 output_name: &str,
671 alias: Option<&str>,
672 column_aliases: &[String],
673) -> String {
674 column_aliases.first().cloned().unwrap_or_else(|| {
675 if is_json_array_table_function(normalized_name) {
676 "value".into()
677 } else {
678 alias.unwrap_or(output_name).to_string()
679 }
680 })
681}
682pub fn join_alias_columns(
683 schema: &crate::RowSchema,
684 alias: &str,
685 column_aliases: &[String],
686) -> Result<Vec<String>, SQLError> {
687 let available = schema.len();
688 let specified = column_aliases.len();
689 if specified > available {
690 return Err(SQLError::Routine {
691 sqlstate: "42P10".into(),
692 message: format!(
693 "join expression \"{alias}\" has {available} columns available but {specified} columns specified"
694 ),
695 });
696 }
697 Ok(schema
698 .columns()
699 .iter()
700 .enumerate()
701 .map(|(position, column)| {
702 column_aliases
703 .get(position)
704 .cloned()
705 .unwrap_or_else(|| schema.public_name(position).unwrap_or(column).to_string())
706 })
707 .collect())
708}
709
710pub fn alias_join_schema(
711 schema: &crate::RowSchema,
712 alias: Option<&str>,
713 column_aliases: &[String],
714) -> Result<crate::RowSchema, SQLError> {
715 let Some(alias) = alias else {
716 if column_aliases.is_empty() {
717 return Ok(schema.clone());
718 }
719 return Err(SQLError::Internal(
720 "JOIN column aliases exist without a relation alias".into(),
721 ));
722 };
723 let columns = join_alias_columns(schema, alias, column_aliases)?;
724 Ok(crate::RowSchema::with_qualified_types(
725 alias,
726 columns,
727 schema.column_types().to_vec(),
728 ))
729}