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