risingwave_frontend/binder/expr/function/
builtin_scalar.rs

1// Copyright 2025 RisingWave Labs
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use 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                // NOTE: this will be further transformed during optimization. See the
92                // documentation of `Now`.
93                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        // `SESSION_USER` is the user name of the user that is connected to the database.
109        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        // `CURRENT_USER` is the user name of the user that is executing the command,
118        // `CURRENT_ROLE`, `USER` are synonyms for `CURRENT_USER`. Since we don't support
119        // `SET ROLE xxx` for now, they will all returns session user name.
120        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        // `CURRENT_DATABASE` is the name of the database you are currently connected to.
129        // `CURRENT_CATALOG` is a synonym for `CURRENT_DATABASE`.
130        fn current_database() -> Handle {
131            guard_by_len(|binder, []| Ok(ExprImpl::literal_varchar(binder.db_name.clone())))
132        }
133
134        // XXX: can we unify this with FUNC_SIG_MAP?
135        // For raw_call here, it seems unnecessary to declare it again here.
136        // For some functions, we have validation logic here. Is it still useful now?
137        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" is the function name used in PG.
166                ("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                // date and time
209                (
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                    // swap the first and second argument
225                    Ok(FunctionCall::new(ExprType::AtTimeZone, vec![arg1, arg0])?.into())
226                })),
227                ("to_date", raw_call(ExprType::CharToDate)),
228                // string
229                ("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                        // Support `quote_literal(any)` by converting it to `quote_literal(any::text)`
272                        // Ref. https://github.com/postgres/postgres/blob/REL_16_1/src/include/catalog/pg_proc.dat#L4641
273                        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                        // Support `quote_nullable(any)` by converting it to `quote_nullable(any::text)`
280                        // Ref. https://github.com/postgres/postgres/blob/REL_16_1/src/include/catalog/pg_proc.dat#L4650
281                        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                    // Similar to `cast` from string, return type is set explicitly rather than inferred.
306                    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
317                ("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                        // rewrite into `CASE WHEN 0 < arg1 AND arg1 <= array_ndims(arg0) THEN 1 END`
360                        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)), // `lower == 1` implies `upper == length`
379                ("array_dims", raw_call(ExprType::ArrayDims)),
380                // int256
381                ("hex_to_int256", raw_call(ExprType::HexToInt256)),
382                // jsonb
383                ("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
414                ("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                // vector
426                ("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                // Functions that return a constant value
433                ("pi", pi()),
434                // greatest and least
435                ("greatest", raw_call(ExprType::Greatest)),
436                ("least", raw_call(ExprType::Least)),
437                // System information operations.
438                (
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                        // TODO: implement pg_get_expr rather than just return empty as an workaround.
534                        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                    // Returns the OID of the current session's temporary schema, or zero if it has none (because it has not created any temporary tables).
544                    Ok(ExprImpl::literal_int(
545                        // always return 0, as we haven't supported temporary tables nor temporary schema yet
546                        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                    // TODO: report session config changes if necessary.
601                    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                    // FIXME: the session id is not global unique in multi-frontend env.
685                    Ok(ExprImpl::literal_int(binder.session_id.0))
686                })),
687                ("pg_cancel_backend", guard_by_len(|_binder, [_input]| {
688                    // TODO: implement real cancel rather than just return false as an workaround.
689                    Ok(ExprImpl::literal_bool(false))
690                })),
691                ("pg_terminate_backend", guard_by_len(|_binder, [_input]| {
692                    // TODO: implement real terminate rather than just return false as an
693                    // workaround.
694                    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                // TODO: really implement them.
706                // https://www.postgresql.org/docs/9.5/functions-info.html#FUNCTIONS-INFO-COMMENT-TABLE
707                // WARN: Hacked in [`Binder::bind_function`]!!!
708                ("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                // internal
715                ("rw_vnode", raw_call(ExprType::VnodeUser)),
716                ("rw_license", raw_call(ExprType::License)),
717                ("rw_test_paid_tier", raw_call(ExprType::TestFeature)), // deprecated, kept for compatibility
718                ("rw_test_feature", raw_call(ExprType::TestFeature)), // for testing purposes
719                // TODO: choose which pg version we should return.
720                ("version", raw_literal(ExprImpl::literal_varchar(current_cluster_version()))),
721                // non-deterministic
722                ("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                // TODO: implement pg_sleep_until
729                // ("pg_sleep_until", raw_call(ExprType::PgSleepUntil)),
730
731                // cast functions
732                // only functions required by the existing PostgreSQL tool are implemented
733                ("date", guard_by_len(|_binder, [input]| {
734                    input.cast_explicit(&DataType::Date).map_err(Into::into)
735                })),
736
737                // AI model functions
738                ("openai_embedding", guard_by_len(|_binder, [arg0, arg1]| {
739                    // check if the first two arguments are constants
740                    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            // TODO: Also hint other functinos, e.g., Agg or UDF.
757            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        // Note: for raw_call, we only check name here. The type check is done later.
785        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
858/// Make sure inputs only have 2 value and rewrite the arguments.
859/// Nullif(expr1,expr2) -> Case(Equal(expr1 = expr2),null,expr1).
860fn 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}