risingwave_frontend/binder/
mod.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, HashSet};
16use std::sync::Arc;
17
18use itertools::Itertools;
19use parking_lot::RwLock;
20use risingwave_common::catalog::FunctionId;
21use risingwave_common::session_config::{SearchPath, SessionConfig};
22use risingwave_common::types::DataType;
23use risingwave_common::util::iter_util::ZipEqDebug;
24use risingwave_sqlparser::ast::{Expr as AstExpr, SelectItem, SetExpr, Statement};
25
26use crate::error::Result;
27
28mod bind_context;
29mod bind_param;
30mod create;
31mod create_view;
32mod declare_cursor;
33mod delete;
34mod expr;
35pub mod fetch_cursor;
36mod for_system;
37mod insert;
38mod query;
39mod relation;
40mod select;
41mod set_expr;
42mod statement;
43mod struct_field;
44mod update;
45mod values;
46
47pub use bind_context::{BindContext, Clause, LateralBindContext};
48pub use create_view::BoundCreateView;
49pub use delete::BoundDelete;
50pub use expr::bind_data_type;
51pub use insert::BoundInsert;
52use pgwire::pg_server::{Session, SessionId};
53pub use query::BoundQuery;
54pub use relation::{
55    BoundBackCteRef, BoundBaseTable, BoundJoin, BoundShare, BoundShareInput, BoundSource,
56    BoundSystemTable, BoundWatermark, BoundWindowTableFunction, Relation,
57    ResolveQualifiedNameError, WindowTableFunctionKind,
58};
59pub use select::{BoundDistinct, BoundSelect};
60pub use set_expr::*;
61pub use statement::BoundStatement;
62pub use update::{BoundUpdate, UpdateProject};
63pub use values::BoundValues;
64
65use crate::catalog::catalog_service::CatalogReadGuard;
66use crate::catalog::root_catalog::SchemaPath;
67use crate::catalog::schema_catalog::SchemaCatalog;
68use crate::catalog::{CatalogResult, DatabaseId, TableId, ViewId};
69use crate::error::ErrorCode;
70use crate::expr::ExprImpl;
71use crate::session::{AuthContext, SessionImpl, TemporarySourceManager};
72use crate::user::user_service::UserInfoReadGuard;
73
74pub type ShareId = usize;
75
76/// The type of binding statement.
77enum BindFor {
78    /// Binding MV/SINK
79    Stream,
80    /// Binding a batch query
81    Batch,
82    /// Binding a DDL (e.g. CREATE TABLE/SOURCE)
83    Ddl,
84    /// Binding a system query (e.g. SHOW)
85    System,
86}
87
88/// `Binder` binds the identifiers in AST to columns in relations
89pub struct Binder {
90    // TODO: maybe we can only lock the database, but not the whole catalog.
91    catalog: CatalogReadGuard,
92    user: UserInfoReadGuard,
93    db_name: String,
94    database_id: DatabaseId,
95    session_id: SessionId,
96    context: BindContext,
97    auth_context: Arc<AuthContext>,
98    /// A stack holding contexts of outer queries when binding a subquery.
99    /// It also holds all of the lateral contexts for each respective
100    /// subquery.
101    ///
102    /// See [`Binder::bind_subquery_expr`] for details.
103    upper_subquery_contexts: Vec<(BindContext, Vec<LateralBindContext>)>,
104
105    /// A stack holding contexts of left-lateral `TableFactor`s.
106    ///
107    /// We need a separate stack as `CorrelatedInputRef` depth is
108    /// determined by the upper subquery context depth, not the lateral context stack depth.
109    lateral_contexts: Vec<LateralBindContext>,
110
111    next_subquery_id: usize,
112    next_values_id: usize,
113    /// The `ShareId` is used to identify the share relation which could be a CTE, a source, a view
114    /// and so on.
115    next_share_id: ShareId,
116
117    session_config: Arc<RwLock<SessionConfig>>,
118
119    search_path: SearchPath,
120    /// The type of binding statement.
121    bind_for: BindFor,
122
123    /// `ShareId`s identifying shared views.
124    shared_views: HashMap<ViewId, ShareId>,
125
126    /// The included relations while binding a query.
127    included_relations: HashSet<TableId>,
128
129    /// The included user-defined functions while binding a query.
130    included_udfs: HashSet<FunctionId>,
131
132    param_types: ParameterTypes,
133
134    /// The sql udf context that will be used during binding phase
135    udf_context: UdfContext,
136
137    /// The temporary sources that will be used during binding phase
138    temporary_source_manager: TemporarySourceManager,
139
140    /// Information for `secure_compare` function. It's ONLY available when binding the
141    /// `VALIDATE` clause of Webhook source i.e. `VALIDATE SECRET ... AS SECURE_COMPARE(...)`.
142    secure_compare_context: Option<SecureCompareContext>,
143}
144
145// There's one more hidden name, `HEADERS`, which is a reserved identifier for HTTP headers. Its type is `JSONB`.
146#[derive(Default, Clone, Debug)]
147pub struct SecureCompareContext {
148    /// The column name to store the whole payload in `JSONB`, but during validation it will be used as `bytea`
149    pub column_name: String,
150    /// The secret (usually a token provided by the webhook source user) to validate the calls
151    pub secret_name: Option<String>,
152}
153
154#[derive(Clone, Debug, Default)]
155pub struct UdfContext {
156    /// The mapping from `sql udf parameters` to a bound `ExprImpl` generated from `ast expressions`
157    /// Note: The expressions are constructed during runtime, correspond to the actual users' input
158    udf_param_context: HashMap<String, ExprImpl>,
159
160    /// The global counter that records the calling stack depth
161    /// of the current binding sql udf chain
162    udf_global_counter: u32,
163}
164
165impl UdfContext {
166    pub fn new() -> Self {
167        Self {
168            udf_param_context: HashMap::new(),
169            udf_global_counter: 0,
170        }
171    }
172
173    pub fn global_count(&self) -> u32 {
174        self.udf_global_counter
175    }
176
177    pub fn incr_global_count(&mut self) {
178        self.udf_global_counter += 1;
179    }
180
181    pub fn decr_global_count(&mut self) {
182        self.udf_global_counter -= 1;
183    }
184
185    pub fn _is_empty(&self) -> bool {
186        self.udf_param_context.is_empty()
187    }
188
189    pub fn update_context(&mut self, context: HashMap<String, ExprImpl>) {
190        self.udf_param_context = context;
191    }
192
193    pub fn _clear(&mut self) {
194        self.udf_global_counter = 0;
195        self.udf_param_context.clear();
196    }
197
198    pub fn get_expr(&self, name: &str) -> Option<&ExprImpl> {
199        self.udf_param_context.get(name)
200    }
201
202    pub fn get_context(&self) -> HashMap<String, ExprImpl> {
203        self.udf_param_context.clone()
204    }
205
206    /// A common utility function to extract sql udf
207    /// expression out from the input `ast`
208    pub fn extract_udf_expression(ast: Vec<Statement>) -> Result<AstExpr> {
209        if ast.len() != 1 {
210            return Err(ErrorCode::InvalidInputSyntax(
211                "the query for sql udf should contain only one statement".to_owned(),
212            )
213            .into());
214        }
215
216        // Extract the expression out
217        let Statement::Query(query) = ast[0].clone() else {
218            return Err(ErrorCode::InvalidInputSyntax(
219                "invalid function definition, please recheck the syntax".to_owned(),
220            )
221            .into());
222        };
223
224        let SetExpr::Select(select) = query.body else {
225            return Err(ErrorCode::InvalidInputSyntax(
226                "missing `select` body for sql udf expression, please recheck the syntax"
227                    .to_owned(),
228            )
229            .into());
230        };
231
232        if select.projection.len() != 1 {
233            return Err(ErrorCode::InvalidInputSyntax(
234                "`projection` should contain only one `SelectItem`".to_owned(),
235            )
236            .into());
237        }
238
239        let SelectItem::UnnamedExpr(expr) = select.projection[0].clone() else {
240            return Err(ErrorCode::InvalidInputSyntax(
241                "expect `UnnamedExpr` for `projection`".to_owned(),
242            )
243            .into());
244        };
245
246        Ok(expr)
247    }
248}
249
250/// `ParameterTypes` is used to record the types of the parameters during binding prepared stataments.
251/// It works by following the rules:
252/// 1. At the beginning, it contains the user specified parameters type.
253/// 2. When the binder encounters a parameter, it will record it as unknown(call `record_new_param`)
254///    if it didn't exist in `ParameterTypes`.
255/// 3. When the binder encounters a cast on parameter, if it's a unknown type, the cast function
256///    will record the target type as infer type for that parameter(call `record_infer_type`). If the
257///    parameter has been inferred, the cast function will act as a normal cast.
258/// 4. After bind finished:
259///    (a) parameter not in `ParameterTypes` means that the user didn't specify it and it didn't
260///    occur in the query. `export` will return error if there is a kind of
261///    parameter. This rule is compatible with PostgreSQL
262///    (b) parameter is None means that it's a unknown type. The user didn't specify it
263///    and we can't infer it in the query. We will treat it as VARCHAR type finally. This rule is
264///    compatible with PostgreSQL.
265///    (c) parameter is Some means that it's a known type.
266#[derive(Clone, Debug)]
267pub struct ParameterTypes(Arc<RwLock<HashMap<u64, Option<DataType>>>>);
268
269impl ParameterTypes {
270    pub fn new(specified_param_types: Vec<Option<DataType>>) -> Self {
271        let map = specified_param_types
272            .into_iter()
273            .enumerate()
274            .map(|(index, data_type)| ((index + 1) as u64, data_type))
275            .collect::<HashMap<u64, Option<DataType>>>();
276        Self(Arc::new(RwLock::new(map)))
277    }
278
279    pub fn has_infer(&self, index: u64) -> bool {
280        self.0.read().get(&index).unwrap().is_some()
281    }
282
283    pub fn read_type(&self, index: u64) -> Option<DataType> {
284        self.0.read().get(&index).unwrap().clone()
285    }
286
287    pub fn record_new_param(&mut self, index: u64) {
288        self.0.write().entry(index).or_insert(None);
289    }
290
291    pub fn record_infer_type(&mut self, index: u64, data_type: DataType) {
292        assert!(
293            !self.has_infer(index),
294            "The parameter has been inferred, should not be inferred again."
295        );
296        self.0.write().get_mut(&index).unwrap().replace(data_type);
297    }
298
299    pub fn export(&self) -> Result<Vec<DataType>> {
300        let types = self
301            .0
302            .read()
303            .clone()
304            .into_iter()
305            .sorted_by_key(|(index, _)| *index)
306            .collect::<Vec<_>>();
307
308        // Check if all the parameters have been inferred.
309        for ((index, _), expect_index) in types.iter().zip_eq_debug(1_u64..=types.len() as u64) {
310            if *index != expect_index {
311                return Err(ErrorCode::InvalidInputSyntax(format!(
312                    "Cannot infer the type of the parameter {}.",
313                    expect_index
314                ))
315                .into());
316            }
317        }
318
319        Ok(types
320            .into_iter()
321            .map(|(_, data_type)| data_type.unwrap_or(DataType::Varchar))
322            .collect::<Vec<_>>())
323    }
324}
325
326impl Binder {
327    fn new_inner(
328        session: &SessionImpl,
329        bind_for: BindFor,
330        param_types: Vec<Option<DataType>>,
331    ) -> Binder {
332        Binder {
333            catalog: session.env().catalog_reader().read_guard(),
334            user: session.env().user_info_reader().read_guard(),
335            db_name: session.database(),
336            database_id: session.database_id(),
337            session_id: session.id(),
338            context: BindContext::new(),
339            auth_context: session.auth_context(),
340            upper_subquery_contexts: vec![],
341            lateral_contexts: vec![],
342            next_subquery_id: 0,
343            next_values_id: 0,
344            next_share_id: 0,
345            session_config: session.shared_config(),
346            search_path: session.config().search_path(),
347            bind_for,
348            shared_views: HashMap::new(),
349            included_relations: HashSet::new(),
350            included_udfs: HashSet::new(),
351            param_types: ParameterTypes::new(param_types),
352            udf_context: UdfContext::new(),
353            temporary_source_manager: session.temporary_source_manager(),
354            secure_compare_context: None,
355        }
356    }
357
358    pub fn new(session: &SessionImpl) -> Binder {
359        Self::new_inner(session, BindFor::Batch, vec![])
360    }
361
362    pub fn new_with_param_types(
363        session: &SessionImpl,
364        param_types: Vec<Option<DataType>>,
365    ) -> Binder {
366        Self::new_inner(session, BindFor::Batch, param_types)
367    }
368
369    pub fn new_for_stream(session: &SessionImpl) -> Binder {
370        Self::new_inner(session, BindFor::Stream, vec![])
371    }
372
373    pub fn new_for_ddl(session: &SessionImpl) -> Binder {
374        Self::new_inner(session, BindFor::Ddl, vec![])
375    }
376
377    pub fn new_for_ddl_with_secure_compare(
378        session: &SessionImpl,
379        ctx: SecureCompareContext,
380    ) -> Binder {
381        let mut binder = Self::new_inner(session, BindFor::Ddl, vec![]);
382        binder.secure_compare_context = Some(ctx);
383        binder
384    }
385
386    pub fn new_for_system(session: &SessionImpl) -> Binder {
387        Self::new_inner(session, BindFor::System, vec![])
388    }
389
390    fn is_for_stream(&self) -> bool {
391        matches!(self.bind_for, BindFor::Stream)
392    }
393
394    #[allow(dead_code)]
395    fn is_for_batch(&self) -> bool {
396        matches!(self.bind_for, BindFor::Batch)
397    }
398
399    fn is_for_ddl(&self) -> bool {
400        matches!(self.bind_for, BindFor::Ddl)
401    }
402
403    /// Bind a [`Statement`].
404    pub fn bind(&mut self, stmt: Statement) -> Result<BoundStatement> {
405        self.bind_statement(stmt)
406    }
407
408    pub fn export_param_types(&self) -> Result<Vec<DataType>> {
409        self.param_types.export()
410    }
411
412    /// Get included relations in the query after binding. This is used for resolving relation
413    /// dependencies. Note that it only contains referenced relations discovered during binding.
414    /// After the plan is built, the referenced relations may be changed. We cannot rely on the
415    /// collection result of plan, because we still need to record the dependencies that have been
416    /// optimised away.
417    pub fn included_relations(&self) -> &HashSet<TableId> {
418        &self.included_relations
419    }
420
421    /// Get included user-defined functions in the query after binding.
422    pub fn included_udfs(&self) -> &HashSet<FunctionId> {
423        &self.included_udfs
424    }
425
426    fn push_context(&mut self) {
427        let new_context = std::mem::take(&mut self.context);
428        self.context
429            .cte_to_relation
430            .clone_from(&new_context.cte_to_relation);
431        self.context.disable_security_invoker = new_context.disable_security_invoker;
432        let new_lateral_contexts = std::mem::take(&mut self.lateral_contexts);
433        self.upper_subquery_contexts
434            .push((new_context, new_lateral_contexts));
435    }
436
437    fn pop_context(&mut self) -> Result<()> {
438        let (old_context, old_lateral_contexts) = self
439            .upper_subquery_contexts
440            .pop()
441            .ok_or_else(|| ErrorCode::InternalError("Popping non-existent context".to_owned()))?;
442        self.context = old_context;
443        self.lateral_contexts = old_lateral_contexts;
444        Ok(())
445    }
446
447    fn push_lateral_context(&mut self) {
448        let new_context = std::mem::take(&mut self.context);
449        self.context
450            .cte_to_relation
451            .clone_from(&new_context.cte_to_relation);
452        self.context.disable_security_invoker = new_context.disable_security_invoker;
453        self.lateral_contexts.push(LateralBindContext {
454            is_visible: false,
455            context: new_context,
456        });
457    }
458
459    fn pop_and_merge_lateral_context(&mut self) -> Result<()> {
460        let mut old_context = self
461            .lateral_contexts
462            .pop()
463            .ok_or_else(|| ErrorCode::InternalError("Popping non-existent context".to_owned()))?
464            .context;
465        old_context.merge_context(self.context.clone())?;
466        self.context = old_context;
467        Ok(())
468    }
469
470    fn try_mark_lateral_as_visible(&mut self) {
471        if let Some(mut ctx) = self.lateral_contexts.pop() {
472            ctx.is_visible = true;
473            self.lateral_contexts.push(ctx);
474        }
475    }
476
477    fn try_mark_lateral_as_invisible(&mut self) {
478        if let Some(mut ctx) = self.lateral_contexts.pop() {
479            ctx.is_visible = false;
480            self.lateral_contexts.push(ctx);
481        }
482    }
483
484    fn next_subquery_id(&mut self) -> usize {
485        let id = self.next_subquery_id;
486        self.next_subquery_id += 1;
487        id
488    }
489
490    fn next_values_id(&mut self) -> usize {
491        let id = self.next_values_id;
492        self.next_values_id += 1;
493        id
494    }
495
496    fn next_share_id(&mut self) -> ShareId {
497        let id = self.next_share_id;
498        self.next_share_id += 1;
499        id
500    }
501
502    fn first_valid_schema(&self) -> CatalogResult<&SchemaCatalog> {
503        self.catalog.first_valid_schema(
504            &self.db_name,
505            &self.search_path,
506            &self.auth_context.user_name,
507        )
508    }
509
510    fn bind_schema_path<'a>(&'a self, schema_name: Option<&'a str>) -> SchemaPath<'a> {
511        SchemaPath::new(schema_name, &self.search_path, &self.auth_context.user_name)
512    }
513
514    pub fn set_clause(&mut self, clause: Option<Clause>) {
515        self.context.clause = clause;
516    }
517
518    pub fn udf_context_mut(&mut self) -> &mut UdfContext {
519        &mut self.udf_context
520    }
521}
522
523/// The column name stored in [`BindContext`] for a column without an alias.
524pub const UNNAMED_COLUMN: &str = "?column?";
525/// The table name stored in [`BindContext`] for a subquery without an alias.
526const UNNAMED_SUBQUERY: &str = "?subquery?";
527/// The table name stored in [`BindContext`] for a column group.
528const COLUMN_GROUP_PREFIX: &str = "?column_group_id?";
529
530#[cfg(test)]
531pub mod test_utils {
532    use risingwave_common::types::DataType;
533
534    use super::Binder;
535    use crate::session::SessionImpl;
536
537    #[cfg(test)]
538    pub fn mock_binder() -> Binder {
539        Binder::new(&SessionImpl::mock())
540    }
541
542    #[cfg(test)]
543    pub fn mock_binder_with_param_types(param_types: Vec<Option<DataType>>) -> Binder {
544        Binder::new_with_param_types(&SessionImpl::mock(), param_types)
545    }
546}
547
548#[cfg(test)]
549mod tests {
550    use expect_test::expect;
551
552    use super::test_utils::*;
553
554    #[tokio::test]
555    async fn test_rcte() {
556        let stmt = risingwave_sqlparser::parser::Parser::parse_sql(
557            "WITH RECURSIVE t1 AS (SELECT 1 AS a UNION ALL SELECT a + 1 FROM t1 WHERE a < 10) SELECT * FROM t1",
558        ).unwrap().into_iter().next().unwrap();
559        let mut binder = mock_binder();
560        let bound = binder.bind(stmt).unwrap();
561
562        let expected = expect![[r#"
563            Query(
564                BoundQuery {
565                    body: Select(
566                        BoundSelect {
567                            distinct: All,
568                            select_items: [
569                                InputRef(
570                                    InputRef {
571                                        index: 0,
572                                        data_type: Int32,
573                                    },
574                                ),
575                            ],
576                            aliases: [
577                                Some(
578                                    "a",
579                                ),
580                            ],
581                            from: Some(
582                                Share(
583                                    BoundShare {
584                                        share_id: 0,
585                                        input: Query(
586                                            Right(
587                                                RecursiveUnion {
588                                                    all: true,
589                                                    base: Select(
590                                                        BoundSelect {
591                                                            distinct: All,
592                                                            select_items: [
593                                                                Literal(
594                                                                    Literal {
595                                                                        data: Some(
596                                                                            Int32(
597                                                                                1,
598                                                                            ),
599                                                                        ),
600                                                                        data_type: Some(
601                                                                            Int32,
602                                                                        ),
603                                                                    },
604                                                                ),
605                                                            ],
606                                                            aliases: [
607                                                                Some(
608                                                                    "a",
609                                                                ),
610                                                            ],
611                                                            from: None,
612                                                            where_clause: None,
613                                                            group_by: GroupKey(
614                                                                [],
615                                                            ),
616                                                            having: None,
617                                                            schema: Schema {
618                                                                fields: [
619                                                                    a:Int32,
620                                                                ],
621                                                            },
622                                                        },
623                                                    ),
624                                                    recursive: Select(
625                                                        BoundSelect {
626                                                            distinct: All,
627                                                            select_items: [
628                                                                FunctionCall(
629                                                                    FunctionCall {
630                                                                        func_type: Add,
631                                                                        return_type: Int32,
632                                                                        inputs: [
633                                                                            InputRef(
634                                                                                InputRef {
635                                                                                    index: 0,
636                                                                                    data_type: Int32,
637                                                                                },
638                                                                            ),
639                                                                            Literal(
640                                                                                Literal {
641                                                                                    data: Some(
642                                                                                        Int32(
643                                                                                            1,
644                                                                                        ),
645                                                                                    ),
646                                                                                    data_type: Some(
647                                                                                        Int32,
648                                                                                    ),
649                                                                                },
650                                                                            ),
651                                                                        ],
652                                                                    },
653                                                                ),
654                                                            ],
655                                                            aliases: [
656                                                                None,
657                                                            ],
658                                                            from: Some(
659                                                                BackCteRef(
660                                                                    BoundBackCteRef {
661                                                                        share_id: 0,
662                                                                        base: Select(
663                                                                            BoundSelect {
664                                                                                distinct: All,
665                                                                                select_items: [
666                                                                                    Literal(
667                                                                                        Literal {
668                                                                                            data: Some(
669                                                                                                Int32(
670                                                                                                    1,
671                                                                                                ),
672                                                                                            ),
673                                                                                            data_type: Some(
674                                                                                                Int32,
675                                                                                            ),
676                                                                                        },
677                                                                                    ),
678                                                                                ],
679                                                                                aliases: [
680                                                                                    Some(
681                                                                                        "a",
682                                                                                    ),
683                                                                                ],
684                                                                                from: None,
685                                                                                where_clause: None,
686                                                                                group_by: GroupKey(
687                                                                                    [],
688                                                                                ),
689                                                                                having: None,
690                                                                                schema: Schema {
691                                                                                    fields: [
692                                                                                        a:Int32,
693                                                                                    ],
694                                                                                },
695                                                                            },
696                                                                        ),
697                                                                    },
698                                                                ),
699                                                            ),
700                                                            where_clause: Some(
701                                                                FunctionCall(
702                                                                    FunctionCall {
703                                                                        func_type: LessThan,
704                                                                        return_type: Boolean,
705                                                                        inputs: [
706                                                                            InputRef(
707                                                                                InputRef {
708                                                                                    index: 0,
709                                                                                    data_type: Int32,
710                                                                                },
711                                                                            ),
712                                                                            Literal(
713                                                                                Literal {
714                                                                                    data: Some(
715                                                                                        Int32(
716                                                                                            10,
717                                                                                        ),
718                                                                                    ),
719                                                                                    data_type: Some(
720                                                                                        Int32,
721                                                                                    ),
722                                                                                },
723                                                                            ),
724                                                                        ],
725                                                                    },
726                                                                ),
727                                                            ),
728                                                            group_by: GroupKey(
729                                                                [],
730                                                            ),
731                                                            having: None,
732                                                            schema: Schema {
733                                                                fields: [
734                                                                    ?column?:Int32,
735                                                                ],
736                                                            },
737                                                        },
738                                                    ),
739                                                    schema: Schema {
740                                                        fields: [
741                                                            a:Int32,
742                                                        ],
743                                                    },
744                                                },
745                                            ),
746                                        ),
747                                    },
748                                ),
749                            ),
750                            where_clause: None,
751                            group_by: GroupKey(
752                                [],
753                            ),
754                            having: None,
755                            schema: Schema {
756                                fields: [
757                                    a:Int32,
758                                ],
759                            },
760                        },
761                    ),
762                    order: [],
763                    limit: None,
764                    offset: None,
765                    with_ties: false,
766                    extra_order_exprs: [],
767                },
768            )"#]];
769
770        expected.assert_eq(&format!("{:#?}", bound));
771    }
772
773    #[tokio::test]
774    async fn test_bind_approx_percentile() {
775        let stmt = risingwave_sqlparser::parser::Parser::parse_sql(
776            "SELECT approx_percentile(0.5, 0.01) WITHIN GROUP (ORDER BY generate_series) FROM generate_series(1, 100)",
777        ).unwrap().into_iter().next().unwrap();
778        let parse_expected = expect![[r#"
779            Query(
780                Query {
781                    with: None,
782                    body: Select(
783                        Select {
784                            distinct: All,
785                            projection: [
786                                UnnamedExpr(
787                                    Function(
788                                        Function {
789                                            scalar_as_agg: false,
790                                            name: ObjectName(
791                                                [
792                                                    Ident {
793                                                        value: "approx_percentile",
794                                                        quote_style: None,
795                                                    },
796                                                ],
797                                            ),
798                                            arg_list: FunctionArgList {
799                                                distinct: false,
800                                                args: [
801                                                    Unnamed(
802                                                        Expr(
803                                                            Value(
804                                                                Number(
805                                                                    "0.5",
806                                                                ),
807                                                            ),
808                                                        ),
809                                                    ),
810                                                    Unnamed(
811                                                        Expr(
812                                                            Value(
813                                                                Number(
814                                                                    "0.01",
815                                                                ),
816                                                            ),
817                                                        ),
818                                                    ),
819                                                ],
820                                                variadic: false,
821                                                order_by: [],
822                                                ignore_nulls: false,
823                                            },
824                                            within_group: Some(
825                                                OrderByExpr {
826                                                    expr: Identifier(
827                                                        Ident {
828                                                            value: "generate_series",
829                                                            quote_style: None,
830                                                        },
831                                                    ),
832                                                    asc: None,
833                                                    nulls_first: None,
834                                                },
835                                            ),
836                                            filter: None,
837                                            over: None,
838                                        },
839                                    ),
840                                ),
841                            ],
842                            from: [
843                                TableWithJoins {
844                                    relation: TableFunction {
845                                        name: ObjectName(
846                                            [
847                                                Ident {
848                                                    value: "generate_series",
849                                                    quote_style: None,
850                                                },
851                                            ],
852                                        ),
853                                        alias: None,
854                                        args: [
855                                            Unnamed(
856                                                Expr(
857                                                    Value(
858                                                        Number(
859                                                            "1",
860                                                        ),
861                                                    ),
862                                                ),
863                                            ),
864                                            Unnamed(
865                                                Expr(
866                                                    Value(
867                                                        Number(
868                                                            "100",
869                                                        ),
870                                                    ),
871                                                ),
872                                            ),
873                                        ],
874                                        with_ordinality: false,
875                                    },
876                                    joins: [],
877                                },
878                            ],
879                            lateral_views: [],
880                            selection: None,
881                            group_by: [],
882                            having: None,
883                        },
884                    ),
885                    order_by: [],
886                    limit: None,
887                    offset: None,
888                    fetch: None,
889                },
890            )"#]];
891        parse_expected.assert_eq(&format!("{:#?}", stmt));
892
893        let mut binder = mock_binder();
894        let bound = binder.bind(stmt).unwrap();
895
896        let expected = expect![[r#"
897            Query(
898                BoundQuery {
899                    body: Select(
900                        BoundSelect {
901                            distinct: All,
902                            select_items: [
903                                AggCall(
904                                    AggCall {
905                                        agg_type: Builtin(
906                                            ApproxPercentile,
907                                        ),
908                                        return_type: Float64,
909                                        args: [
910                                            FunctionCall(
911                                                FunctionCall {
912                                                    func_type: Cast,
913                                                    return_type: Float64,
914                                                    inputs: [
915                                                        InputRef(
916                                                            InputRef {
917                                                                index: 0,
918                                                                data_type: Int32,
919                                                            },
920                                                        ),
921                                                    ],
922                                                },
923                                            ),
924                                        ],
925                                        filter: Condition {
926                                            conjunctions: [],
927                                        },
928                                        distinct: false,
929                                        order_by: OrderBy {
930                                            sort_exprs: [
931                                                OrderByExpr {
932                                                    expr: InputRef(
933                                                        InputRef {
934                                                            index: 0,
935                                                            data_type: Int32,
936                                                        },
937                                                    ),
938                                                    order_type: OrderType {
939                                                        direction: Ascending,
940                                                        nulls_are: Largest,
941                                                    },
942                                                },
943                                            ],
944                                        },
945                                        direct_args: [
946                                            Literal {
947                                                data: Some(
948                                                    Float64(
949                                                        OrderedFloat(
950                                                            0.5,
951                                                        ),
952                                                    ),
953                                                ),
954                                                data_type: Some(
955                                                    Float64,
956                                                ),
957                                            },
958                                            Literal {
959                                                data: Some(
960                                                    Float64(
961                                                        OrderedFloat(
962                                                            0.01,
963                                                        ),
964                                                    ),
965                                                ),
966                                                data_type: Some(
967                                                    Float64,
968                                                ),
969                                            },
970                                        ],
971                                    },
972                                ),
973                            ],
974                            aliases: [
975                                Some(
976                                    "approx_percentile",
977                                ),
978                            ],
979                            from: Some(
980                                TableFunction {
981                                    expr: TableFunction(
982                                        FunctionCall {
983                                            function_type: GenerateSeries,
984                                            return_type: Int32,
985                                            args: [
986                                                Literal(
987                                                    Literal {
988                                                        data: Some(
989                                                            Int32(
990                                                                1,
991                                                            ),
992                                                        ),
993                                                        data_type: Some(
994                                                            Int32,
995                                                        ),
996                                                    },
997                                                ),
998                                                Literal(
999                                                    Literal {
1000                                                        data: Some(
1001                                                            Int32(
1002                                                                100,
1003                                                            ),
1004                                                        ),
1005                                                        data_type: Some(
1006                                                            Int32,
1007                                                        ),
1008                                                    },
1009                                                ),
1010                                            ],
1011                                        },
1012                                    ),
1013                                    with_ordinality: false,
1014                                },
1015                            ),
1016                            where_clause: None,
1017                            group_by: GroupKey(
1018                                [],
1019                            ),
1020                            having: None,
1021                            schema: Schema {
1022                                fields: [
1023                                    approx_percentile:Float64,
1024                                ],
1025                            },
1026                        },
1027                    ),
1028                    order: [],
1029                    limit: None,
1030                    offset: None,
1031                    with_ties: false,
1032                    extra_order_exprs: [],
1033                },
1034            )"#]];
1035
1036        expected.assert_eq(&format!("{:#?}", bound));
1037    }
1038}