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(ColumnType::Array(element))] = argument_types else {
256 return Vec::new();
257 };
258 vec![overload(vec![ColumnType::AnyArray], 0, (**element).clone())]
259 }
260 "regexp_split_to_table" | "string_to_table" => vec![overload(
261 vec![ColumnType::Text, ColumnType::Text],
262 0,
263 ColumnType::Text,
264 )],
265 "json_array_elements" => vec![overload(vec![ColumnType::Json], 0, ColumnType::Json)],
266 "jsonb_array_elements" => vec![overload(vec![ColumnType::JsonB], 0, ColumnType::JsonB)],
267 "json_array_elements_text" | "json_object_keys" => {
268 vec![overload(vec![ColumnType::Json], 0, ColumnType::Text)]
269 }
270 "jsonb_array_elements_text" | "jsonb_object_keys" => {
271 vec![overload(vec![ColumnType::JsonB], 0, ColumnType::Text)]
272 }
273 "json_each" | "json_each_text" => {
274 vec![overload(vec![ColumnType::Json], 0, ColumnType::Record)]
275 }
276 "jsonb_each" | "jsonb_each_text" => {
277 vec![overload(vec![ColumnType::JsonB], 0, ColumnType::Record)]
278 }
279 "pg_get_sequence_data" => vec![overload(vec![ColumnType::Regclass], 0, ColumnType::Record)],
280 "pg_sequence_parameters" => {
281 vec![overload(vec![ColumnType::Oid], 0, ColumnType::Record)]
282 }
283 _ => Vec::new(),
284 }
285}
286
287pub fn is_builtin_table_function(name: &str) -> bool {
288 matches!(
289 name,
290 "generate_series"
291 | "pg_listening_channels"
292 | "unnest"
293 | "regexp_split_to_table"
294 | "string_to_table"
295 | "json_array_elements"
296 | "jsonb_array_elements"
297 | "json_array_elements_text"
298 | "jsonb_array_elements_text"
299 | "json_object_keys"
300 | "jsonb_object_keys"
301 | "json_each"
302 | "jsonb_each"
303 | "json_each_text"
304 | "jsonb_each_text"
305 | "pg_get_sequence_data"
306 | "pg_sequence_parameters"
307 | "create_analyzer"
308 | "drop_analyzer"
309 | "list_analyzers"
310 | "analyze_text"
311 | "fts_index_stats"
312 | "set_table_analyzer"
313 | "pagerank"
314 | "graph_pagerank"
315 | "hits"
316 | "graph_hits"
317 | "betweenness"
318 | "graph_betweenness"
319 | "graph_edges"
320 | "rpq"
321 | "cypher"
322 )
323}
324
325pub fn table_function_empty_schema(
326 name: &str,
327 output_name: &str,
328 alias: Option<&str>,
329 column_aliases: &[String],
330 output_width: usize,
331 ordinality: bool,
332) -> Vec<String> {
333 let lower = crate::semantics::builtin_function_dispatch_name(&name.to_ascii_lowercase());
334 let columns = if lower == "unnest" {
335 let width = output_width.max(1);
336 let default_column = if width == 1 {
337 alias.unwrap_or(output_name)
338 } else {
339 output_name
340 };
341 vec![default_column.to_string(); width]
342 } else {
343 match lower.as_str() {
344 "json_each" | "jsonb_each" | "json_each_text" | "jsonb_each_text" => {
345 vec!["key".into(), "value".into()]
346 }
347 "pg_get_sequence_data" => vec!["last_value".into(), "is_called".into()],
348 "pg_sequence_parameters" => vec![
349 "start_value".into(),
350 "minimum_value".into(),
351 "maximum_value".into(),
352 "increment".into(),
353 "cycle_option".into(),
354 "cache_size".into(),
355 "data_type".into(),
356 ],
357 "pagerank" | "graph_pagerank" | "hits" | "graph_hits" | "betweenness"
358 | "graph_betweenness" => vec!["_doc_id".into(), "_score".into()],
359 "rpq" => vec!["vertex_id".into()],
360 "list_analyzers" => vec!["analyzer_name".into()],
361 "analyze_text" => vec!["analysis".into()],
362 "fts_index_stats" => vec![
363 "table_name".into(),
364 "field".into(),
365 "analyzer".into(),
366 "posting_count".into(),
367 "doc_length_count".into(),
368 "indexed_doc_count".into(),
369 "term_count".into(),
370 "total_field_length".into(),
371 ],
372 "text_similarity_join"
373 | "vector_similarity_join"
374 | "graph_join"
375 | "hybrid_join"
376 | "cross_paradigm_join" => {
377 vec!["left_doc_id".into(), "right_doc_id".into(), "_score".into()]
378 }
379 "generate_series"
380 | "pg_listening_channels"
381 | "regexp_split_to_table"
382 | "string_to_table"
383 | "json_array_elements"
384 | "jsonb_array_elements"
385 | "json_array_elements_text"
386 | "jsonb_array_elements_text"
387 | "json_object_keys"
388 | "jsonb_object_keys" => vec![scalar_table_function_default_column(
389 &lower,
390 output_name,
391 alias,
392 &[],
393 )],
394 _ => {
395 let minimum_width = 1;
396 let aliased_value_width = if ordinality && column_aliases.len() > minimum_width {
397 column_aliases.len() - 1
398 } else {
399 column_aliases.len()
400 };
401 vec![
402 scalar_table_function_default_column(&lower, output_name, alias, &[]);
403 minimum_width.max(aliased_value_width)
404 ]
405 }
406 }
407 };
408 apply_table_function_aliases(columns, column_aliases, ordinality)
409}
410
411pub fn apply_table_function_aliases(
412 mut columns: Vec<String>,
413 column_aliases: &[String],
414 ordinality: bool,
415) -> Vec<String> {
416 if ordinality {
417 columns.push("ordinality".into());
418 }
419 for (column, alias) in columns.iter_mut().zip(column_aliases) {
420 column.clone_from(alias);
421 }
422 columns
423}
424
425pub fn validate_table_function_alias_count(
426 table_alias: &str,
427 available: usize,
428 specified: usize,
429) -> Result<(), SQLError> {
430 if specified <= available {
431 return Ok(());
432 }
433 Err(SQLError::Routine {
434 sqlstate: "42P10".into(),
435 message: format!(
436 "table \"{table_alias}\" has {available} columns available but {specified} columns specified"
437 ),
438 })
439}
440
441pub struct TableFunctionTypeRequest<'a> {
442 pub name: &'a str,
443 pub args: &'a [ScalarExpr],
444 pub user_function: Option<&'a SQLUserFunction>,
445 pub user_invocation: Option<&'a RoutineInvocationBinding>,
446 pub declared_types: &'a [String],
447 pub columns: &'a [String],
448 pub ordinality: bool,
449}
450
451#[expect(
452 clippy::too_many_lines,
453 reason = "preserves source schema and row identity"
454)]
455pub fn table_function_column_types(
456 routines: &dyn RoutineResolution,
457 request: TableFunctionTypeRequest<'_>,
458 input_schema: &crate::RowSchema,
459 params: &[SQLParam],
460 resolver: &dyn FunctionTypeResolver,
461) -> Vec<Option<ColumnType>> {
462 let TableFunctionTypeRequest {
463 name,
464 args,
465 user_function,
466 user_invocation,
467 declared_types,
468 columns,
469 ordinality,
470 } = request;
471 let value_columns = if ordinality {
472 columns
473 .get(..columns.len().saturating_sub(1))
474 .unwrap_or(&[])
475 } else {
476 columns
477 };
478 let align = |types: Vec<Option<ColumnType>>| {
479 if types.len() == value_columns.len() {
480 types
481 } else if let [ty] = types.as_slice() {
482 vec![ty.clone(); value_columns.len()]
483 } else {
484 vec![None; value_columns.len()]
485 }
486 };
487 let mut types = if !declared_types.is_empty() {
488 align(
489 declared_types
490 .iter()
491 .map(|ty| ColumnType::from_sql_name(ty).ok())
492 .collect(),
493 )
494 } else if let Some(function) = user_function {
495 align(user_function_column_types(
496 routines,
497 function,
498 user_invocation,
499 ))
500 } else {
501 let normalized =
502 crate::semantics::builtin_function_dispatch_name(&name.to_ascii_lowercase());
503 let argument_type = |position: usize| {
504 args.get(position)
505 .and_then(|argument| {
506 crate::scalar_type_with_resolver(argument, input_schema, params, resolver).ok()
507 })
508 .flatten()
509 };
510 align(match normalized.as_str() {
511 "pg_listening_channels" => vec![Some(ColumnType::Text)],
512 "generate_series" => vec![argument_type(0)],
513 "analyze_text" => vec![Some(ColumnType::JsonB)],
514 "unnest" => args
515 .iter()
516 .map(|argument| {
517 crate::scalar_type_with_resolver(argument, input_schema, params, resolver)
518 .ok()
519 .flatten()
520 .and_then(|ty| match ty {
521 ColumnType::Array(element) => Some(*element),
522 _ => None,
523 })
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}