1use std::collections::HashMap;
16use std::sync::LazyLock;
17
18use bk_tree::{BKTree, metrics};
19use itertools::Itertools;
20use risingwave_common::session_config::USER_NAME_WILD_CARD;
21use risingwave_common::types::{DataType, ListValue, ScalarImpl, Timestamptz};
22use risingwave_common::{bail_not_implemented, current_cluster_version, no_function};
23use thiserror_ext::AsReport;
24
25use crate::Binder;
26use crate::binder::Clause;
27use crate::error::{ErrorCode, Result};
28use crate::expr::{CastContext, Expr, ExprImpl, ExprType, FunctionCall, Literal, Now};
29
30impl Binder {
31 pub(super) fn bind_builtin_scalar_function(
32 &mut self,
33 function_name: &str,
34 inputs: Vec<ExprImpl>,
35 variadic: bool,
36 ) -> Result<ExprImpl> {
37 type Inputs = Vec<ExprImpl>;
38
39 type Handle = Box<dyn Fn(&mut Binder, Inputs) -> Result<ExprImpl> + Sync + Send>;
40
41 fn rewrite(r#type: ExprType, rewriter: fn(Inputs) -> Result<Inputs>) -> Handle {
42 Box::new(move |_binder, mut inputs| {
43 inputs = (rewriter)(inputs)?;
44 Ok(FunctionCall::new(r#type, inputs)?.into())
45 })
46 }
47
48 fn raw_call(r#type: ExprType) -> Handle {
49 rewrite(r#type, Ok)
50 }
51
52 fn guard_by_len(expected_len: usize, handle: Handle) -> Handle {
53 Box::new(move |binder, inputs| {
54 if inputs.len() == expected_len {
55 handle(binder, inputs)
56 } else {
57 Err(ErrorCode::ExprError("unexpected arguments number".into()).into())
58 }
59 })
60 }
61
62 fn raw<F: Fn(&mut Binder, Inputs) -> Result<ExprImpl> + Sync + Send + 'static>(
63 f: F,
64 ) -> Handle {
65 Box::new(f)
66 }
67
68 fn dispatch_by_len(mapping: Vec<(usize, Handle)>) -> Handle {
69 Box::new(move |binder, inputs| {
70 for (len, handle) in &mapping {
71 if inputs.len() == *len {
72 return handle(binder, inputs);
73 }
74 }
75 Err(ErrorCode::ExprError("unexpected arguments number".into()).into())
76 })
77 }
78
79 fn raw_literal(literal: ExprImpl) -> Handle {
80 Box::new(move |_binder, _inputs| Ok(literal.clone()))
81 }
82
83 fn now() -> Handle {
84 guard_by_len(
85 0,
86 raw(move |binder, _inputs| {
87 binder.ensure_now_function_allowed()?;
88 Ok(Now.into())
91 }),
92 )
93 }
94
95 fn pi() -> Handle {
96 raw_literal(ExprImpl::literal_f64(std::f64::consts::PI))
97 }
98
99 fn proctime() -> Handle {
100 Box::new(move |binder, inputs| {
101 binder.ensure_proctime_function_allowed()?;
102 raw_call(ExprType::Proctime)(binder, inputs)
103 })
104 }
105
106 fn session_user() -> Handle {
108 guard_by_len(
109 0,
110 raw(|binder, _inputs| {
111 Ok(ExprImpl::literal_varchar(
112 binder.auth_context.user_name.clone(),
113 ))
114 }),
115 )
116 }
117
118 fn current_user() -> Handle {
122 guard_by_len(
123 0,
124 raw(|binder, _inputs| {
125 Ok(ExprImpl::literal_varchar(
126 binder.auth_context.user_name.clone(),
127 ))
128 }),
129 )
130 }
131
132 fn current_database() -> Handle {
135 guard_by_len(
136 0,
137 raw(|binder, _inputs| Ok(ExprImpl::literal_varchar(binder.db_name.clone()))),
138 )
139 }
140
141 static HANDLES: LazyLock<HashMap<&'static str, Handle>> = LazyLock::new(|| {
145 [
146 (
147 "booleq",
148 rewrite(ExprType::Equal, rewrite_two_bool_inputs),
149 ),
150 (
151 "boolne",
152 rewrite(ExprType::NotEqual, rewrite_two_bool_inputs),
153 ),
154 ("coalesce", rewrite(ExprType::Coalesce, |inputs| {
155 if inputs.iter().any(ExprImpl::has_table_function) {
156 return Err(ErrorCode::BindError("table functions are not allowed in COALESCE".into()).into());
157 }
158 Ok(inputs)
159 })),
160 (
161 "nullif",
162 rewrite(ExprType::Case, rewrite_nullif_to_case_when),
163 ),
164 (
165 "round",
166 dispatch_by_len(vec![
167 (2, raw_call(ExprType::RoundDigit)),
168 (1, raw_call(ExprType::Round)),
169 ]),
170 ),
171 ("pow", raw_call(ExprType::Pow)),
172 ("power", raw_call(ExprType::Pow)),
174 ("ceil", raw_call(ExprType::Ceil)),
175 ("ceiling", raw_call(ExprType::Ceil)),
176 ("floor", raw_call(ExprType::Floor)),
177 ("trunc", raw_call(ExprType::Trunc)),
178 ("abs", raw_call(ExprType::Abs)),
179 ("exp", raw_call(ExprType::Exp)),
180 ("ln", raw_call(ExprType::Ln)),
181 ("log", raw_call(ExprType::Log10)),
182 ("log10", raw_call(ExprType::Log10)),
183 ("mod", raw_call(ExprType::Modulus)),
184 ("sin", raw_call(ExprType::Sin)),
185 ("cos", raw_call(ExprType::Cos)),
186 ("tan", raw_call(ExprType::Tan)),
187 ("cot", raw_call(ExprType::Cot)),
188 ("asin", raw_call(ExprType::Asin)),
189 ("acos", raw_call(ExprType::Acos)),
190 ("atan", raw_call(ExprType::Atan)),
191 ("atan2", raw_call(ExprType::Atan2)),
192 ("sind", raw_call(ExprType::Sind)),
193 ("cosd", raw_call(ExprType::Cosd)),
194 ("cotd", raw_call(ExprType::Cotd)),
195 ("tand", raw_call(ExprType::Tand)),
196 ("sinh", raw_call(ExprType::Sinh)),
197 ("cosh", raw_call(ExprType::Cosh)),
198 ("tanh", raw_call(ExprType::Tanh)),
199 ("coth", raw_call(ExprType::Coth)),
200 ("asinh", raw_call(ExprType::Asinh)),
201 ("acosh", raw_call(ExprType::Acosh)),
202 ("atanh", raw_call(ExprType::Atanh)),
203 ("asind", raw_call(ExprType::Asind)),
204 ("acosd", raw_call(ExprType::Acosd)),
205 ("atand", raw_call(ExprType::Atand)),
206 ("atan2d", raw_call(ExprType::Atan2d)),
207 ("degrees", raw_call(ExprType::Degrees)),
208 ("radians", raw_call(ExprType::Radians)),
209 ("sqrt", raw_call(ExprType::Sqrt)),
210 ("cbrt", raw_call(ExprType::Cbrt)),
211 ("sign", raw_call(ExprType::Sign)),
212 ("scale", raw_call(ExprType::Scale)),
213 ("min_scale", raw_call(ExprType::MinScale)),
214 ("trim_scale", raw_call(ExprType::TrimScale)),
215 (
217 "to_timestamp",
218 dispatch_by_len(vec![
219 (1, raw_call(ExprType::SecToTimestamptz)),
220 (2, raw_call(ExprType::CharToTimestamptz)),
221 ]),
222 ),
223 ("date_trunc", raw_call(ExprType::DateTrunc)),
224 ("date_bin", raw_call(ExprType::DateBin)),
225 ("date_part", raw_call(ExprType::DatePart)),
226 ("make_date", raw_call(ExprType::MakeDate)),
227 ("make_time", raw_call(ExprType::MakeTime)),
228 ("make_timestamp", raw_call(ExprType::MakeTimestamp)),
229 ("make_timestamptz", raw_call(ExprType::MakeTimestamptz)),
230 ("timezone", rewrite(ExprType::AtTimeZone, |mut inputs| {
231 if inputs.len() == 2 {
232 inputs.swap(0, 1);
233 Ok(inputs)
234 } else {
235 Err(ErrorCode::ExprError("unexpected arguments number".into()).into())
236 }
237 })),
238 ("to_date", raw_call(ExprType::CharToDate)),
239 ("substr", raw_call(ExprType::Substr)),
241 ("length", raw_call(ExprType::Length)),
242 ("upper", raw_call(ExprType::Upper)),
243 ("lower", raw_call(ExprType::Lower)),
244 ("trim", raw_call(ExprType::Trim)),
245 ("replace", raw_call(ExprType::Replace)),
246 ("overlay", raw_call(ExprType::Overlay)),
247 ("btrim", raw_call(ExprType::Trim)),
248 ("ltrim", raw_call(ExprType::Ltrim)),
249 ("rtrim", raw_call(ExprType::Rtrim)),
250 ("md5", raw_call(ExprType::Md5)),
251 ("to_char", raw_call(ExprType::ToChar)),
252 (
253 "concat",
254 rewrite(ExprType::ConcatWs, rewrite_concat_to_concat_ws),
255 ),
256 ("concat_ws", raw_call(ExprType::ConcatWs)),
257 ("format", raw_call(ExprType::Format)),
258 ("translate", raw_call(ExprType::Translate)),
259 ("split_part", raw_call(ExprType::SplitPart)),
260 ("char_length", raw_call(ExprType::CharLength)),
261 ("character_length", raw_call(ExprType::CharLength)),
262 ("repeat", raw_call(ExprType::Repeat)),
263 ("ascii", raw_call(ExprType::Ascii)),
264 ("octet_length", raw_call(ExprType::OctetLength)),
265 ("bit_length", raw_call(ExprType::BitLength)),
266 ("regexp_match", raw_call(ExprType::RegexpMatch)),
267 ("regexp_replace", raw_call(ExprType::RegexpReplace)),
268 ("regexp_count", raw_call(ExprType::RegexpCount)),
269 ("regexp_split_to_array", raw_call(ExprType::RegexpSplitToArray)),
270 ("chr", raw_call(ExprType::Chr)),
271 ("starts_with", raw_call(ExprType::StartsWith)),
272 ("initcap", raw_call(ExprType::Initcap)),
273 ("lpad", raw_call(ExprType::Lpad)),
274 ("rpad", raw_call(ExprType::Rpad)),
275 ("reverse", raw_call(ExprType::Reverse)),
276 ("strpos", raw_call(ExprType::Position)),
277 ("to_ascii", raw_call(ExprType::ToAscii)),
278 ("to_hex", raw_call(ExprType::ToHex)),
279 ("quote_ident", raw_call(ExprType::QuoteIdent)),
280 ("quote_literal", guard_by_len(1, raw(|_binder, mut inputs| {
281 if inputs[0].return_type() != DataType::Varchar {
282 FunctionCall::cast_mut(&mut inputs[0], DataType::Varchar, CastContext::Explicit)?;
285 }
286 Ok(FunctionCall::new_unchecked(ExprType::QuoteLiteral, inputs, DataType::Varchar).into())
287 }))),
288 ("quote_nullable", guard_by_len(1, raw(|_binder, mut inputs| {
289 if inputs[0].return_type() != DataType::Varchar {
290 FunctionCall::cast_mut(&mut inputs[0], DataType::Varchar, CastContext::Explicit)?;
293 }
294 Ok(FunctionCall::new_unchecked(ExprType::QuoteNullable, inputs, DataType::Varchar).into())
295 }))),
296 ("string_to_array", raw_call(ExprType::StringToArray)),
297 ("encode", raw_call(ExprType::Encode)),
298 ("decode", raw_call(ExprType::Decode)),
299 ("convert_from", raw_call(ExprType::ConvertFrom)),
300 ("convert_to", raw_call(ExprType::ConvertTo)),
301 ("sha1", raw_call(ExprType::Sha1)),
302 ("sha224", raw_call(ExprType::Sha224)),
303 ("sha256", raw_call(ExprType::Sha256)),
304 ("sha384", raw_call(ExprType::Sha384)),
305 ("sha512", raw_call(ExprType::Sha512)),
306 ("encrypt", raw_call(ExprType::Encrypt)),
307 ("decrypt", raw_call(ExprType::Decrypt)),
308 ("hmac", raw_call(ExprType::Hmac)),
309 ("secure_compare", raw_call(ExprType::SecureCompare)),
310 ("left", raw_call(ExprType::Left)),
311 ("right", raw_call(ExprType::Right)),
312 ("inet_aton", raw_call(ExprType::InetAton)),
313 ("inet_ntoa", raw_call(ExprType::InetNtoa)),
314 ("int8send", raw_call(ExprType::PgwireSend)),
315 ("int8recv", guard_by_len(1, raw(|_binder, mut inputs| {
316 let hint = if !inputs[0].is_untyped() && inputs[0].return_type() == DataType::Varchar {
318 " Consider `decode` or cast."
319 } else {
320 ""
321 };
322 inputs[0].cast_implicit_mut(DataType::Bytea).map_err(|e| {
323 ErrorCode::BindError(format!("{} in `recv`.{hint}", e.as_report()))
324 })?;
325 Ok(FunctionCall::new_unchecked(ExprType::PgwireRecv, inputs, DataType::Int64).into())
326 }))),
327 ("array_cat", raw_call(ExprType::ArrayCat)),
329 ("array_append", raw_call(ExprType::ArrayAppend)),
330 ("array_join", raw_call(ExprType::ArrayToString)),
331 ("array_prepend", raw_call(ExprType::ArrayPrepend)),
332 ("array_to_string", raw_call(ExprType::ArrayToString)),
333 ("array_distinct", raw_call(ExprType::ArrayDistinct)),
334 ("array_min", raw_call(ExprType::ArrayMin)),
335 ("array_sort", raw_call(ExprType::ArraySort)),
336 ("array_length", raw_call(ExprType::ArrayLength)),
337 ("cardinality", raw_call(ExprType::Cardinality)),
338 ("array_remove", raw_call(ExprType::ArrayRemove)),
339 ("array_replace", raw_call(ExprType::ArrayReplace)),
340 ("array_max", raw_call(ExprType::ArrayMax)),
341 ("array_sum", raw_call(ExprType::ArraySum)),
342 ("array_position", raw_call(ExprType::ArrayPosition)),
343 ("array_positions", raw_call(ExprType::ArrayPositions)),
344 ("array_contains", raw_call(ExprType::ArrayContains)),
345 ("arraycontains", raw_call(ExprType::ArrayContains)),
346 ("array_contained", raw_call(ExprType::ArrayContained)),
347 ("arraycontained", raw_call(ExprType::ArrayContained)),
348 ("array_flatten", guard_by_len(1, raw(|_binder, inputs| {
349 inputs[0].ensure_array_type().map_err(|_| ErrorCode::BindError("array_flatten expects `any[][]` input".into()))?;
350 let return_type = inputs[0].return_type().into_list_element_type();
351 if !return_type.is_array() {
352 return Err(ErrorCode::BindError("array_flatten expects `any[][]` input".into()).into());
353
354 }
355 Ok(FunctionCall::new_unchecked(ExprType::ArrayFlatten, inputs, return_type).into())
356 }))),
357 ("trim_array", raw_call(ExprType::TrimArray)),
358 (
359 "array_ndims",
360 guard_by_len(1, raw(|_binder, inputs| {
361 inputs[0].ensure_array_type()?;
362
363 let n = inputs[0].return_type().array_ndims()
364 .try_into().map_err(|_| ErrorCode::BindError("array_ndims integer overflow".into()))?;
365 Ok(ExprImpl::literal_int(n))
366 })),
367 ),
368 (
369 "array_lower",
370 guard_by_len(2, raw(|binder, inputs| {
371 let (arg0, arg1) = inputs.into_iter().next_tuple().unwrap();
372 let ndims_expr = binder.bind_builtin_scalar_function("array_ndims", vec![arg0], false)?;
374 let arg1 = arg1.cast_implicit(DataType::Int32)?;
375
376 FunctionCall::new(
377 ExprType::Case,
378 vec![
379 FunctionCall::new(
380 ExprType::And,
381 vec![
382 FunctionCall::new(ExprType::LessThan, vec![ExprImpl::literal_int(0), arg1.clone()])?.into(),
383 FunctionCall::new(ExprType::LessThanOrEqual, vec![arg1, ndims_expr])?.into(),
384 ],
385 )?.into(),
386 ExprImpl::literal_int(1),
387 ],
388 ).map(Into::into)
389 })),
390 ),
391 ("array_upper", raw_call(ExprType::ArrayLength)), ("array_dims", raw_call(ExprType::ArrayDims)),
393 ("hex_to_int256", raw_call(ExprType::HexToInt256)),
395 ("jsonb_object_field", raw_call(ExprType::JsonbAccess)),
397 ("jsonb_array_element", raw_call(ExprType::JsonbAccess)),
398 ("jsonb_object_field_text", raw_call(ExprType::JsonbAccessStr)),
399 ("jsonb_array_element_text", raw_call(ExprType::JsonbAccessStr)),
400 ("jsonb_extract_path", raw_call(ExprType::JsonbExtractPath)),
401 ("jsonb_extract_path_text", raw_call(ExprType::JsonbExtractPathText)),
402 ("jsonb_typeof", raw_call(ExprType::JsonbTypeof)),
403 ("jsonb_array_length", raw_call(ExprType::JsonbArrayLength)),
404 ("jsonb_concat", raw_call(ExprType::JsonbConcat)),
405 ("jsonb_object", raw_call(ExprType::JsonbObject)),
406 ("jsonb_pretty", raw_call(ExprType::JsonbPretty)),
407 ("jsonb_contains", raw_call(ExprType::JsonbContains)),
408 ("jsonb_contained", raw_call(ExprType::JsonbContained)),
409 ("jsonb_exists", raw_call(ExprType::JsonbExists)),
410 ("jsonb_exists_any", raw_call(ExprType::JsonbExistsAny)),
411 ("jsonb_exists_all", raw_call(ExprType::JsonbExistsAll)),
412 ("jsonb_delete", raw_call(ExprType::Subtract)),
413 ("jsonb_delete_path", raw_call(ExprType::JsonbDeletePath)),
414 ("jsonb_strip_nulls", raw_call(ExprType::JsonbStripNulls)),
415 ("to_jsonb", raw_call(ExprType::ToJsonb)),
416 ("jsonb_build_array", raw_call(ExprType::JsonbBuildArray)),
417 ("jsonb_build_object", raw_call(ExprType::JsonbBuildObject)),
418 ("jsonb_populate_record", raw_call(ExprType::JsonbPopulateRecord)),
419 ("jsonb_path_match", raw_call(ExprType::JsonbPathMatch)),
420 ("jsonb_path_exists", raw_call(ExprType::JsonbPathExists)),
421 ("jsonb_path_query_array", raw_call(ExprType::JsonbPathQueryArray)),
422 ("jsonb_path_query_first", raw_call(ExprType::JsonbPathQueryFirst)),
423 ("jsonb_set", raw_call(ExprType::JsonbSet)),
424 ("jsonb_populate_map", raw_call(ExprType::JsonbPopulateMap)),
425 ("map_from_entries", raw_call(ExprType::MapFromEntries)),
427 ("map_access", raw_call(ExprType::MapAccess)),
428 ("map_keys", raw_call(ExprType::MapKeys)),
429 ("map_values", raw_call(ExprType::MapValues)),
430 ("map_entries", raw_call(ExprType::MapEntries)),
431 ("map_from_key_values", raw_call(ExprType::MapFromKeyValues)),
432 ("map_cat", raw_call(ExprType::MapCat)),
433 ("map_contains", raw_call(ExprType::MapContains)),
434 ("map_delete", raw_call(ExprType::MapDelete)),
435 ("map_insert", raw_call(ExprType::MapInsert)),
436 ("map_length", raw_call(ExprType::MapLength)),
437 ("l2_distance", raw_call(ExprType::L2Distance)),
439 ("pi", pi()),
441 ("greatest", raw_call(ExprType::Greatest)),
443 ("least", raw_call(ExprType::Least)),
444 (
446 "pg_typeof",
447 guard_by_len(1, raw(|_binder, inputs| {
448 let input = &inputs[0];
449 let v = match input.is_untyped() {
450 true => "unknown".into(),
451 false => input.return_type().to_string(),
452 };
453 Ok(ExprImpl::literal_varchar(v))
454 })),
455 ),
456 ("current_catalog", current_database()),
457 ("current_database", current_database()),
458 ("current_schema", guard_by_len(0, raw(|binder, _inputs| {
459 Ok(binder
460 .first_valid_schema()
461 .map(|schema| ExprImpl::literal_varchar(schema.name()))
462 .unwrap_or_else(|_| ExprImpl::literal_null(DataType::Varchar)))
463 }))),
464 ("current_schemas", raw(|binder, mut inputs| {
465 let no_match_err = ErrorCode::ExprError(
466 "No function matches the given name and argument types. You might need to add explicit type casts.".into()
467 );
468 if inputs.len() != 1 {
469 return Err(no_match_err.into());
470 }
471 let input = inputs
472 .pop()
473 .unwrap()
474 .enforce_bool_clause("current_schemas")
475 .map_err(|_| no_match_err)?;
476
477 let ExprImpl::Literal(literal) = &input else {
478 bail_not_implemented!("Only boolean literals are supported in `current_schemas`.");
479 };
480
481 let Some(bool) = literal.get_data().as_ref().map(|bool| bool.clone().into_bool()) else {
482 return Ok(ExprImpl::literal_null(DataType::List(Box::new(DataType::Varchar))));
483 };
484
485 let paths = if bool {
486 binder.search_path.path()
487 } else {
488 binder.search_path.real_path()
489 };
490
491 let mut schema_names = vec![];
492 for path in paths {
493 let mut schema_name = path;
494 if schema_name == USER_NAME_WILD_CARD {
495 schema_name = &binder.auth_context.user_name;
496 }
497
498 if binder
499 .catalog
500 .get_schema_by_name(&binder.db_name, schema_name)
501 .is_ok()
502 {
503 schema_names.push(schema_name.as_str());
504 }
505 }
506
507 Ok(ExprImpl::literal_list(
508 ListValue::from_iter(schema_names),
509 DataType::Varchar,
510 ))
511 })),
512 ("session_user", session_user()),
513 ("current_role", current_user()),
514 ("current_user", current_user()),
515 ("user", current_user()),
516 ("pg_get_userbyid", raw_call(ExprType::PgGetUserbyid)),
517 ("pg_get_indexdef", raw_call(ExprType::PgGetIndexdef)),
518 ("pg_get_viewdef", raw_call(ExprType::PgGetViewdef)),
519 ("pg_index_column_has_property", raw_call(ExprType::PgIndexColumnHasProperty)),
520 ("pg_relation_size", raw(|_binder, mut inputs| {
521 if inputs.is_empty() {
522 return Err(ErrorCode::ExprError(
523 "function pg_relation_size() does not exist".into(),
524 )
525 .into());
526 }
527 inputs[0].cast_to_regclass_mut()?;
528 Ok(FunctionCall::new(ExprType::PgRelationSize, inputs)?.into())
529 })),
530 ("pg_get_serial_sequence", raw_literal(ExprImpl::literal_null(DataType::Varchar))),
531 ("pg_table_size", guard_by_len(1, raw(|_binder, mut inputs| {
532 inputs[0].cast_to_regclass_mut()?;
533 Ok(FunctionCall::new(ExprType::PgRelationSize, inputs)?.into())
534 }))),
535 ("pg_indexes_size", guard_by_len(1, raw(|_binder, mut inputs| {
536 inputs[0].cast_to_regclass_mut()?;
537 Ok(FunctionCall::new(ExprType::PgIndexesSize, inputs)?.into())
538 }))),
539 ("pg_get_expr", raw(|_binder, inputs| {
540 if inputs.len() == 2 || inputs.len() == 3 {
541 Ok(ExprImpl::literal_varchar("".into()))
543 } else {
544 Err(ErrorCode::ExprError(
545 "Too many/few arguments for pg_catalog.pg_get_expr()".into(),
546 )
547 .into())
548 }
549 })),
550 ("pg_my_temp_schema", guard_by_len(0, raw(|_binder, _inputs| {
551 Ok(ExprImpl::literal_int(
553 0,
555 ))
556 }))),
557 ("current_setting", guard_by_len(1, raw(|binder, inputs| {
558 let input = &inputs[0];
559 let input = if let ExprImpl::Literal(literal) = input &&
560 let Some(ScalarImpl::Utf8(input)) = literal.get_data()
561 {
562 input
563 } else {
564 return Err(ErrorCode::ExprError(
565 "Only literal is supported in `setting_name`.".into(),
566 )
567 .into());
568 };
569 let session_config = binder.session_config.read();
570 Ok(ExprImpl::literal_varchar(session_config.get(input.as_ref())?))
571 }))),
572 ("set_config", guard_by_len(3, raw(|binder, inputs| {
573 let setting_name = if let ExprImpl::Literal(literal) = &inputs[0] && let Some(ScalarImpl::Utf8(input)) = literal.get_data() {
574 input
575 } else {
576 return Err(ErrorCode::ExprError(
577 "Only string literal is supported in `setting_name`.".into(),
578 )
579 .into());
580 };
581
582 let new_value = if let ExprImpl::Literal(literal) = &inputs[1] && let Some(ScalarImpl::Utf8(input)) = literal.get_data() {
583 input
584 } else {
585 return Err(ErrorCode::ExprError(
586 "Only string literal is supported in `setting_name`.".into(),
587 )
588 .into());
589 };
590
591 let is_local = if let ExprImpl::Literal(literal) = &inputs[2] && let Some(ScalarImpl::Bool(input)) = literal.get_data() {
592 input
593 } else {
594 return Err(ErrorCode::ExprError(
595 "Only bool literal is supported in `is_local`.".into(),
596 )
597 .into());
598 };
599
600 if *is_local {
601 return Err(ErrorCode::ExprError(
602 "`is_local = true` is not supported now.".into(),
603 )
604 .into());
605 }
606
607 let mut session_config = binder.session_config.write();
608
609 session_config.set(setting_name, new_value.to_string(), &mut ())?;
611
612 Ok(ExprImpl::literal_varchar(new_value.to_string()))
613 }))),
614 ("format_type", raw_call(ExprType::FormatType)),
615 ("pg_table_is_visible", raw_call(ExprType::PgTableIsVisible)),
616 ("pg_type_is_visible", raw_literal(ExprImpl::literal_bool(true))),
617 ("pg_get_constraintdef", raw_literal(ExprImpl::literal_null(DataType::Varchar))),
618 ("pg_get_partkeydef", raw_literal(ExprImpl::literal_null(DataType::Varchar))),
619 ("pg_encoding_to_char", raw_literal(ExprImpl::literal_varchar("UTF8".into()))),
620 ("has_database_privilege", raw_literal(ExprImpl::literal_bool(true))),
621 ("has_table_privilege", raw(|binder, mut inputs| {
622 if inputs.len() == 2 {
623 inputs.insert(0, ExprImpl::literal_varchar(binder.auth_context.user_name.clone()));
624 }
625 if inputs.len() == 3 {
626 if inputs[1].return_type() == DataType::Varchar {
627 inputs[1].cast_to_regclass_mut()?;
628 }
629 Ok(FunctionCall::new(ExprType::HasTablePrivilege, inputs)?.into())
630 } else {
631 Err(ErrorCode::ExprError(
632 "Too many/few arguments for pg_catalog.has_table_privilege()".into(),
633 )
634 .into())
635 }
636 })),
637 ("has_any_column_privilege", raw(|binder, mut inputs| {
638 if inputs.len() == 2 {
639 inputs.insert(0, ExprImpl::literal_varchar(binder.auth_context.user_name.clone()));
640 }
641 if inputs.len() == 3 {
642 if inputs[1].return_type() == DataType::Varchar {
643 inputs[1].cast_to_regclass_mut()?;
644 }
645 Ok(FunctionCall::new(ExprType::HasAnyColumnPrivilege, inputs)?.into())
646 } else {
647 Err(ErrorCode::ExprError(
648 "Too many/few arguments for pg_catalog.has_any_column_privilege()".into(),
649 )
650 .into())
651 }
652 })),
653 ("has_schema_privilege", raw(|binder, mut inputs| {
654 if inputs.len() == 2 {
655 inputs.insert(0, ExprImpl::literal_varchar(binder.auth_context.user_name.clone()));
656 }
657 if inputs.len() == 3 {
658 Ok(FunctionCall::new(ExprType::HasSchemaPrivilege, inputs)?.into())
659 } else {
660 Err(ErrorCode::ExprError(
661 "Too many/few arguments for pg_catalog.has_schema_privilege()".into(),
662 )
663 .into())
664 }
665 })),
666 ("has_function_privilege", raw(|binder, mut inputs| {
667 if inputs.len() == 2 {
668 inputs.insert(0, ExprImpl::literal_varchar(binder.auth_context.user_name.clone()));
669 }
670 if inputs.len() == 3 {
671 Ok(FunctionCall::new(ExprType::HasFunctionPrivilege, inputs)?.into())
672 } else {
673 Err(ErrorCode::ExprError(
674 "Too many/few arguments for pg_catalog.has_function_privilege()".into(),
675 )
676 .into())
677 }
678 })),
679 ("pg_stat_get_numscans", raw_literal(ExprImpl::literal_bigint(0))),
680 ("pg_backend_pid", raw(|binder, _inputs| {
681 Ok(ExprImpl::literal_int(binder.session_id.0))
683 })),
684 ("pg_cancel_backend", guard_by_len(1, raw(|_binder, _inputs| {
685 Ok(ExprImpl::literal_bool(false))
687 }))),
688 ("pg_terminate_backend", guard_by_len(1, raw(|_binder, _inputs| {
689 Ok(ExprImpl::literal_bool(false))
692 }))),
693 ("pg_tablespace_location", guard_by_len(1, raw_literal(ExprImpl::literal_null(DataType::Varchar)))),
694 ("pg_postmaster_start_time", guard_by_len(0, raw(|_binder, _inputs| {
695 let server_start_time = risingwave_variables::get_server_start_time();
696 let datum = server_start_time.map(Timestamptz::from).map(ScalarImpl::from);
697 let literal = Literal::new(datum, DataType::Timestamptz);
698 Ok(literal.into())
699 }))),
700 ("col_description", raw_call(ExprType::ColDescription)),
704 ("obj_description", raw_literal(ExprImpl::literal_varchar("".to_owned()))),
705 ("shobj_description", raw_literal(ExprImpl::literal_varchar("".to_owned()))),
706 ("pg_is_in_recovery", raw_call(ExprType::PgIsInRecovery)),
707 ("rw_recovery_status", raw_call(ExprType::RwRecoveryStatus)),
708 ("rw_epoch_to_ts", raw_call(ExprType::RwEpochToTs)),
709 ("rw_vnode", raw_call(ExprType::VnodeUser)),
711 ("rw_license", raw_call(ExprType::License)),
712 ("rw_test_paid_tier", raw_call(ExprType::TestPaidTier)), ("version", raw_literal(ExprImpl::literal_varchar(current_cluster_version()))),
715 ("now", now()),
717 ("current_timestamp", now()),
718 ("proctime", proctime()),
719 ("pg_sleep", raw_call(ExprType::PgSleep)),
720 ("pg_sleep_for", raw_call(ExprType::PgSleepFor)),
721 ("date", guard_by_len(1, raw(|_binder, inputs| {
727 inputs[0].clone().cast_explicit(DataType::Date).map_err(Into::into)
728 }))),
729
730 ("openai_embedding", guard_by_len(3, raw(|_binder, inputs| {
732 if let ExprImpl::Literal(api_key) = &inputs[0] && let Some(ScalarImpl::Utf8(_api_key)) = api_key.get_data()
734 && let ExprImpl::Literal(model) = &inputs[1] && let Some(ScalarImpl::Utf8(_model)) = model.get_data() {
735 Ok(FunctionCall::new(ExprType::OpenaiEmbedding, inputs)?.into())
736 } else {
737 Err(ErrorCode::InvalidInputSyntax(
738 "`api_key` and `model` must be constant strings".to_owned(),
739 ).into())
740 }
741 }))),
742 ]
743 .into_iter()
744 .collect()
745 });
746
747 static FUNCTIONS_BKTREE: LazyLock<BKTree<&str>> = LazyLock::new(|| {
748 let mut tree = BKTree::new(metrics::Levenshtein);
749
750 for k in HANDLES.keys() {
752 tree.add(*k);
753 }
754
755 tree
756 });
757
758 if variadic {
759 let func = match function_name {
760 "format" => ExprType::FormatVariadic,
761 "concat" => ExprType::ConcatVariadic,
762 "concat_ws" => ExprType::ConcatWsVariadic,
763 "jsonb_build_array" => ExprType::JsonbBuildArrayVariadic,
764 "jsonb_build_object" => ExprType::JsonbBuildObjectVariadic,
765 "jsonb_extract_path" => ExprType::JsonbExtractPathVariadic,
766 "jsonb_extract_path_text" => ExprType::JsonbExtractPathTextVariadic,
767 _ => {
768 return Err(ErrorCode::BindError(format!(
769 "VARIADIC argument is not allowed in function \"{}\"",
770 function_name
771 ))
772 .into());
773 }
774 };
775 return Ok(FunctionCall::new(func, inputs)?.into());
776 }
777
778 match HANDLES.get(function_name) {
780 Some(handle) => handle(self, inputs),
781 None => {
782 let allowed_distance = if function_name.len() > 3 { 2 } else { 1 };
783
784 let candidates = FUNCTIONS_BKTREE
785 .find(function_name, allowed_distance)
786 .map(|(_idx, c)| c)
787 .join(" or ");
788
789 Err(no_function!(
790 candidates = (!candidates.is_empty()).then_some(candidates),
791 "{}({})",
792 function_name,
793 inputs.iter().map(|e| e.return_type()).join(", ")
794 )
795 .into())
796 }
797 }
798 }
799
800 fn ensure_now_function_allowed(&self) -> Result<()> {
801 if self.is_for_stream()
802 && !matches!(
803 self.context.clause,
804 Some(Clause::Where)
805 | Some(Clause::Having)
806 | Some(Clause::JoinOn)
807 | Some(Clause::From)
808 )
809 {
810 return Err(ErrorCode::InvalidInputSyntax(format!(
811 "For streaming queries, `NOW()` function is only allowed in `WHERE`, `HAVING`, `ON` and `FROM`. Found in clause: {:?}. \
812 Please please refer to https://www.risingwave.dev/docs/current/sql-pattern-temporal-filters/ for more information",
813 self.context.clause
814 ))
815 .into());
816 }
817 if matches!(self.context.clause, Some(Clause::GeneratedColumn)) {
818 return Err(ErrorCode::InvalidInputSyntax(
819 "Cannot use `NOW()` function in generated columns. Do you want `PROCTIME()`?"
820 .to_owned(),
821 )
822 .into());
823 }
824 Ok(())
825 }
826
827 fn ensure_proctime_function_allowed(&self) -> Result<()> {
828 if !self.is_for_ddl() {
829 return Err(ErrorCode::InvalidInputSyntax(
830 "Function `PROCTIME()` is only allowed in CREATE TABLE/SOURCE. Is `NOW()` what you want?".to_owned(),
831 )
832 .into());
833 }
834 Ok(())
835 }
836}
837
838fn rewrite_concat_to_concat_ws(inputs: Vec<ExprImpl>) -> Result<Vec<ExprImpl>> {
839 if inputs.is_empty() {
840 Err(ErrorCode::BindError(
841 "Function `concat` takes at least 1 arguments (0 given)".to_owned(),
842 )
843 .into())
844 } else {
845 let inputs = std::iter::once(ExprImpl::literal_varchar("".to_owned()))
846 .chain(inputs)
847 .collect();
848 Ok(inputs)
849 }
850}
851
852fn rewrite_nullif_to_case_when(inputs: Vec<ExprImpl>) -> Result<Vec<ExprImpl>> {
855 if inputs.len() != 2 {
856 Err(ErrorCode::BindError("Function `nullif` must contain 2 arguments".to_owned()).into())
857 } else {
858 let inputs = vec![
859 FunctionCall::new(ExprType::Equal, inputs.clone())?.into(),
860 Literal::new(None, inputs[0].return_type()).into(),
861 inputs[0].clone(),
862 ];
863 Ok(inputs)
864 }
865}
866
867fn rewrite_two_bool_inputs(mut inputs: Vec<ExprImpl>) -> Result<Vec<ExprImpl>> {
868 if inputs.len() != 2 {
869 return Err(
870 ErrorCode::BindError("function must contain only 2 arguments".to_owned()).into(),
871 );
872 }
873 let left = inputs.pop().unwrap();
874 let right = inputs.pop().unwrap();
875 Ok(vec![
876 left.cast_implicit(DataType::Boolean)?,
877 right.cast_implicit(DataType::Boolean)?,
878 ])
879}