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    pub fn new_for_stream_with_param_types(
391        session: &SessionImpl,
392        param_types: Vec<Option<DataType>>,
393    ) -> Binder {
394        Self::new_inner(session, BindFor::Stream, param_types)
395    }
396
397    fn is_for_stream(&self) -> bool {
398        matches!(self.bind_for, BindFor::Stream)
399    }
400
401    #[allow(dead_code)]
402    fn is_for_batch(&self) -> bool {
403        matches!(self.bind_for, BindFor::Batch)
404    }
405
406    fn is_for_ddl(&self) -> bool {
407        matches!(self.bind_for, BindFor::Ddl)
408    }
409
410    /// Bind a [`Statement`].
411    pub fn bind(&mut self, stmt: Statement) -> Result<BoundStatement> {
412        self.bind_statement(stmt)
413    }
414
415    pub fn export_param_types(&self) -> Result<Vec<DataType>> {
416        self.param_types.export()
417    }
418
419    /// Get included relations in the query after binding. This is used for resolving relation
420    /// dependencies. Note that it only contains referenced relations discovered during binding.
421    /// After the plan is built, the referenced relations may be changed. We cannot rely on the
422    /// collection result of plan, because we still need to record the dependencies that have been
423    /// optimised away.
424    pub fn included_relations(&self) -> &HashSet<TableId> {
425        &self.included_relations
426    }
427
428    /// Get included user-defined functions in the query after binding.
429    pub fn included_udfs(&self) -> &HashSet<FunctionId> {
430        &self.included_udfs
431    }
432
433    fn push_context(&mut self) {
434        let new_context = std::mem::take(&mut self.context);
435        self.context
436            .cte_to_relation
437            .clone_from(&new_context.cte_to_relation);
438        self.context.disable_security_invoker = new_context.disable_security_invoker;
439        let new_lateral_contexts = std::mem::take(&mut self.lateral_contexts);
440        self.upper_subquery_contexts
441            .push((new_context, new_lateral_contexts));
442    }
443
444    fn pop_context(&mut self) -> Result<()> {
445        let (old_context, old_lateral_contexts) = self
446            .upper_subquery_contexts
447            .pop()
448            .ok_or_else(|| ErrorCode::InternalError("Popping non-existent context".to_owned()))?;
449        self.context = old_context;
450        self.lateral_contexts = old_lateral_contexts;
451        Ok(())
452    }
453
454    fn push_lateral_context(&mut self) {
455        let new_context = std::mem::take(&mut self.context);
456        self.context
457            .cte_to_relation
458            .clone_from(&new_context.cte_to_relation);
459        self.context.disable_security_invoker = new_context.disable_security_invoker;
460        self.lateral_contexts.push(LateralBindContext {
461            is_visible: false,
462            context: new_context,
463        });
464    }
465
466    fn pop_and_merge_lateral_context(&mut self) -> Result<()> {
467        let mut old_context = self
468            .lateral_contexts
469            .pop()
470            .ok_or_else(|| ErrorCode::InternalError("Popping non-existent context".to_owned()))?
471            .context;
472        old_context.merge_context(self.context.clone())?;
473        self.context = old_context;
474        Ok(())
475    }
476
477    fn try_mark_lateral_as_visible(&mut self) {
478        if let Some(mut ctx) = self.lateral_contexts.pop() {
479            ctx.is_visible = true;
480            self.lateral_contexts.push(ctx);
481        }
482    }
483
484    fn try_mark_lateral_as_invisible(&mut self) {
485        if let Some(mut ctx) = self.lateral_contexts.pop() {
486            ctx.is_visible = false;
487            self.lateral_contexts.push(ctx);
488        }
489    }
490
491    fn next_subquery_id(&mut self) -> usize {
492        let id = self.next_subquery_id;
493        self.next_subquery_id += 1;
494        id
495    }
496
497    fn next_values_id(&mut self) -> usize {
498        let id = self.next_values_id;
499        self.next_values_id += 1;
500        id
501    }
502
503    fn next_share_id(&mut self) -> ShareId {
504        let id = self.next_share_id;
505        self.next_share_id += 1;
506        id
507    }
508
509    fn first_valid_schema(&self) -> CatalogResult<&SchemaCatalog> {
510        self.catalog.first_valid_schema(
511            &self.db_name,
512            &self.search_path,
513            &self.auth_context.user_name,
514        )
515    }
516
517    fn bind_schema_path<'a>(&'a self, schema_name: Option<&'a str>) -> SchemaPath<'a> {
518        SchemaPath::new(schema_name, &self.search_path, &self.auth_context.user_name)
519    }
520
521    pub fn set_clause(&mut self, clause: Option<Clause>) {
522        self.context.clause = clause;
523    }
524
525    pub fn udf_context_mut(&mut self) -> &mut UdfContext {
526        &mut self.udf_context
527    }
528}
529
530/// The column name stored in [`BindContext`] for a column without an alias.
531pub const UNNAMED_COLUMN: &str = "?column?";
532/// The table name stored in [`BindContext`] for a subquery without an alias.
533const UNNAMED_SUBQUERY: &str = "?subquery?";
534/// The table name stored in [`BindContext`] for a column group.
535const COLUMN_GROUP_PREFIX: &str = "?column_group_id?";
536
537#[cfg(test)]
538pub mod test_utils {
539    use risingwave_common::types::DataType;
540
541    use super::Binder;
542    use crate::session::SessionImpl;
543
544    #[cfg(test)]
545    pub fn mock_binder() -> Binder {
546        Binder::new(&SessionImpl::mock())
547    }
548
549    #[cfg(test)]
550    pub fn mock_binder_with_param_types(param_types: Vec<Option<DataType>>) -> Binder {
551        Binder::new_with_param_types(&SessionImpl::mock(), param_types)
552    }
553}
554
555#[cfg(test)]
556mod tests {
557    use expect_test::expect;
558
559    use super::test_utils::*;
560
561    #[tokio::test]
562    async fn test_rcte() {
563        let stmt = risingwave_sqlparser::parser::Parser::parse_sql(
564            "WITH RECURSIVE t1 AS (SELECT 1 AS a UNION ALL SELECT a + 1 FROM t1 WHERE a < 10) SELECT * FROM t1",
565        ).unwrap().into_iter().next().unwrap();
566        let mut binder = mock_binder();
567        let bound = binder.bind(stmt).unwrap();
568
569        let expected = expect![[r#"
570            Query(
571                BoundQuery {
572                    body: Select(
573                        BoundSelect {
574                            distinct: All,
575                            select_items: [
576                                InputRef(
577                                    InputRef {
578                                        index: 0,
579                                        data_type: Int32,
580                                    },
581                                ),
582                            ],
583                            aliases: [
584                                Some(
585                                    "a",
586                                ),
587                            ],
588                            from: Some(
589                                Share(
590                                    BoundShare {
591                                        share_id: 0,
592                                        input: Query(
593                                            Right(
594                                                RecursiveUnion {
595                                                    all: true,
596                                                    base: Select(
597                                                        BoundSelect {
598                                                            distinct: All,
599                                                            select_items: [
600                                                                Literal(
601                                                                    Literal {
602                                                                        data: Some(
603                                                                            Int32(
604                                                                                1,
605                                                                            ),
606                                                                        ),
607                                                                        data_type: Some(
608                                                                            Int32,
609                                                                        ),
610                                                                    },
611                                                                ),
612                                                            ],
613                                                            aliases: [
614                                                                Some(
615                                                                    "a",
616                                                                ),
617                                                            ],
618                                                            from: None,
619                                                            where_clause: None,
620                                                            group_by: GroupKey(
621                                                                [],
622                                                            ),
623                                                            having: None,
624                                                            schema: Schema {
625                                                                fields: [
626                                                                    a:Int32,
627                                                                ],
628                                                            },
629                                                        },
630                                                    ),
631                                                    recursive: Select(
632                                                        BoundSelect {
633                                                            distinct: All,
634                                                            select_items: [
635                                                                FunctionCall(
636                                                                    FunctionCall {
637                                                                        func_type: Add,
638                                                                        return_type: Int32,
639                                                                        inputs: [
640                                                                            InputRef(
641                                                                                InputRef {
642                                                                                    index: 0,
643                                                                                    data_type: Int32,
644                                                                                },
645                                                                            ),
646                                                                            Literal(
647                                                                                Literal {
648                                                                                    data: Some(
649                                                                                        Int32(
650                                                                                            1,
651                                                                                        ),
652                                                                                    ),
653                                                                                    data_type: Some(
654                                                                                        Int32,
655                                                                                    ),
656                                                                                },
657                                                                            ),
658                                                                        ],
659                                                                    },
660                                                                ),
661                                                            ],
662                                                            aliases: [
663                                                                None,
664                                                            ],
665                                                            from: Some(
666                                                                BackCteRef(
667                                                                    BoundBackCteRef {
668                                                                        share_id: 0,
669                                                                        base: Select(
670                                                                            BoundSelect {
671                                                                                distinct: All,
672                                                                                select_items: [
673                                                                                    Literal(
674                                                                                        Literal {
675                                                                                            data: Some(
676                                                                                                Int32(
677                                                                                                    1,
678                                                                                                ),
679                                                                                            ),
680                                                                                            data_type: Some(
681                                                                                                Int32,
682                                                                                            ),
683                                                                                        },
684                                                                                    ),
685                                                                                ],
686                                                                                aliases: [
687                                                                                    Some(
688                                                                                        "a",
689                                                                                    ),
690                                                                                ],
691                                                                                from: None,
692                                                                                where_clause: None,
693                                                                                group_by: GroupKey(
694                                                                                    [],
695                                                                                ),
696                                                                                having: None,
697                                                                                schema: Schema {
698                                                                                    fields: [
699                                                                                        a:Int32,
700                                                                                    ],
701                                                                                },
702                                                                            },
703                                                                        ),
704                                                                    },
705                                                                ),
706                                                            ),
707                                                            where_clause: Some(
708                                                                FunctionCall(
709                                                                    FunctionCall {
710                                                                        func_type: LessThan,
711                                                                        return_type: Boolean,
712                                                                        inputs: [
713                                                                            InputRef(
714                                                                                InputRef {
715                                                                                    index: 0,
716                                                                                    data_type: Int32,
717                                                                                },
718                                                                            ),
719                                                                            Literal(
720                                                                                Literal {
721                                                                                    data: Some(
722                                                                                        Int32(
723                                                                                            10,
724                                                                                        ),
725                                                                                    ),
726                                                                                    data_type: Some(
727                                                                                        Int32,
728                                                                                    ),
729                                                                                },
730                                                                            ),
731                                                                        ],
732                                                                    },
733                                                                ),
734                                                            ),
735                                                            group_by: GroupKey(
736                                                                [],
737                                                            ),
738                                                            having: None,
739                                                            schema: Schema {
740                                                                fields: [
741                                                                    ?column?:Int32,
742                                                                ],
743                                                            },
744                                                        },
745                                                    ),
746                                                    schema: Schema {
747                                                        fields: [
748                                                            a:Int32,
749                                                        ],
750                                                    },
751                                                },
752                                            ),
753                                        ),
754                                    },
755                                ),
756                            ),
757                            where_clause: None,
758                            group_by: GroupKey(
759                                [],
760                            ),
761                            having: None,
762                            schema: Schema {
763                                fields: [
764                                    a:Int32,
765                                ],
766                            },
767                        },
768                    ),
769                    order: [],
770                    limit: None,
771                    offset: None,
772                    with_ties: false,
773                    extra_order_exprs: [],
774                },
775            )"#]];
776
777        expected.assert_eq(&format!("{:#?}", bound));
778    }
779
780    #[tokio::test]
781    async fn test_bind_approx_percentile() {
782        let stmt = risingwave_sqlparser::parser::Parser::parse_sql(
783            "SELECT approx_percentile(0.5, 0.01) WITHIN GROUP (ORDER BY generate_series) FROM generate_series(1, 100)",
784        ).unwrap().into_iter().next().unwrap();
785        let parse_expected = expect![[r#"
786            Query(
787                Query {
788                    with: None,
789                    body: Select(
790                        Select {
791                            distinct: All,
792                            projection: [
793                                UnnamedExpr(
794                                    Function(
795                                        Function {
796                                            scalar_as_agg: false,
797                                            name: ObjectName(
798                                                [
799                                                    Ident {
800                                                        value: "approx_percentile",
801                                                        quote_style: None,
802                                                    },
803                                                ],
804                                            ),
805                                            arg_list: FunctionArgList {
806                                                distinct: false,
807                                                args: [
808                                                    Unnamed(
809                                                        Expr(
810                                                            Value(
811                                                                Number(
812                                                                    "0.5",
813                                                                ),
814                                                            ),
815                                                        ),
816                                                    ),
817                                                    Unnamed(
818                                                        Expr(
819                                                            Value(
820                                                                Number(
821                                                                    "0.01",
822                                                                ),
823                                                            ),
824                                                        ),
825                                                    ),
826                                                ],
827                                                variadic: false,
828                                                order_by: [],
829                                                ignore_nulls: false,
830                                            },
831                                            within_group: Some(
832                                                OrderByExpr {
833                                                    expr: Identifier(
834                                                        Ident {
835                                                            value: "generate_series",
836                                                            quote_style: None,
837                                                        },
838                                                    ),
839                                                    asc: None,
840                                                    nulls_first: None,
841                                                },
842                                            ),
843                                            filter: None,
844                                            over: None,
845                                        },
846                                    ),
847                                ),
848                            ],
849                            from: [
850                                TableWithJoins {
851                                    relation: TableFunction {
852                                        name: ObjectName(
853                                            [
854                                                Ident {
855                                                    value: "generate_series",
856                                                    quote_style: None,
857                                                },
858                                            ],
859                                        ),
860                                        alias: None,
861                                        args: [
862                                            Unnamed(
863                                                Expr(
864                                                    Value(
865                                                        Number(
866                                                            "1",
867                                                        ),
868                                                    ),
869                                                ),
870                                            ),
871                                            Unnamed(
872                                                Expr(
873                                                    Value(
874                                                        Number(
875                                                            "100",
876                                                        ),
877                                                    ),
878                                                ),
879                                            ),
880                                        ],
881                                        with_ordinality: false,
882                                    },
883                                    joins: [],
884                                },
885                            ],
886                            lateral_views: [],
887                            selection: None,
888                            group_by: [],
889                            having: None,
890                        },
891                    ),
892                    order_by: [],
893                    limit: None,
894                    offset: None,
895                    fetch: None,
896                },
897            )"#]];
898        parse_expected.assert_eq(&format!("{:#?}", stmt));
899
900        let mut binder = mock_binder();
901        let bound = binder.bind(stmt).unwrap();
902
903        let expected = expect![[r#"
904            Query(
905                BoundQuery {
906                    body: Select(
907                        BoundSelect {
908                            distinct: All,
909                            select_items: [
910                                AggCall(
911                                    AggCall {
912                                        agg_type: Builtin(
913                                            ApproxPercentile,
914                                        ),
915                                        return_type: Float64,
916                                        args: [
917                                            FunctionCall(
918                                                FunctionCall {
919                                                    func_type: Cast,
920                                                    return_type: Float64,
921                                                    inputs: [
922                                                        InputRef(
923                                                            InputRef {
924                                                                index: 0,
925                                                                data_type: Int32,
926                                                            },
927                                                        ),
928                                                    ],
929                                                },
930                                            ),
931                                        ],
932                                        filter: Condition {
933                                            conjunctions: [],
934                                        },
935                                        distinct: false,
936                                        order_by: OrderBy {
937                                            sort_exprs: [
938                                                OrderByExpr {
939                                                    expr: InputRef(
940                                                        InputRef {
941                                                            index: 0,
942                                                            data_type: Int32,
943                                                        },
944                                                    ),
945                                                    order_type: OrderType {
946                                                        direction: Ascending,
947                                                        nulls_are: Largest,
948                                                    },
949                                                },
950                                            ],
951                                        },
952                                        direct_args: [
953                                            Literal {
954                                                data: Some(
955                                                    Float64(
956                                                        OrderedFloat(
957                                                            0.5,
958                                                        ),
959                                                    ),
960                                                ),
961                                                data_type: Some(
962                                                    Float64,
963                                                ),
964                                            },
965                                            Literal {
966                                                data: Some(
967                                                    Float64(
968                                                        OrderedFloat(
969                                                            0.01,
970                                                        ),
971                                                    ),
972                                                ),
973                                                data_type: Some(
974                                                    Float64,
975                                                ),
976                                            },
977                                        ],
978                                    },
979                                ),
980                            ],
981                            aliases: [
982                                Some(
983                                    "approx_percentile",
984                                ),
985                            ],
986                            from: Some(
987                                TableFunction {
988                                    expr: TableFunction(
989                                        FunctionCall {
990                                            function_type: GenerateSeries,
991                                            return_type: Int32,
992                                            args: [
993                                                Literal(
994                                                    Literal {
995                                                        data: Some(
996                                                            Int32(
997                                                                1,
998                                                            ),
999                                                        ),
1000                                                        data_type: Some(
1001                                                            Int32,
1002                                                        ),
1003                                                    },
1004                                                ),
1005                                                Literal(
1006                                                    Literal {
1007                                                        data: Some(
1008                                                            Int32(
1009                                                                100,
1010                                                            ),
1011                                                        ),
1012                                                        data_type: Some(
1013                                                            Int32,
1014                                                        ),
1015                                                    },
1016                                                ),
1017                                            ],
1018                                        },
1019                                    ),
1020                                    with_ordinality: false,
1021                                },
1022                            ),
1023                            where_clause: None,
1024                            group_by: GroupKey(
1025                                [],
1026                            ),
1027                            having: None,
1028                            schema: Schema {
1029                                fields: [
1030                                    approx_percentile:Float64,
1031                                ],
1032                            },
1033                        },
1034                    ),
1035                    order: [],
1036                    limit: None,
1037                    offset: None,
1038                    with_ties: false,
1039                    extra_order_exprs: [],
1040                },
1041            )"#]];
1042
1043        expected.assert_eq(&format!("{:#?}", bound));
1044    }
1045}