risingwave_frontend/binder/relation/
mod.rs

1// Copyright 2022 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::hash_map::Entry;
16use std::ops::Deref;
17
18use itertools::{EitherOrBoth, Itertools};
19use risingwave_common::bail;
20use risingwave_common::catalog::{Field, TableId};
21use risingwave_sqlparser::ast::{
22    AsOf, Expr as ParserExpr, FunctionArg, FunctionArgExpr, Ident, ObjectName, TableAlias,
23    TableFactor,
24};
25use thiserror::Error;
26use thiserror_ext::AsReport;
27
28use super::bind_context::ColumnBinding;
29use super::statement::RewriteExprsRecursive;
30use crate::binder::Binder;
31use crate::binder::bind_context::{BindingCte, BindingCteState};
32use crate::error::{ErrorCode, Result, RwError};
33use crate::expr::{ExprImpl, InputRef};
34
35mod gap_fill;
36mod join;
37mod share;
38mod subquery;
39mod table_function;
40mod table_or_source;
41mod watermark;
42mod window_table_function;
43
44pub use gap_fill::BoundGapFill;
45pub use join::BoundJoin;
46pub use share::{BoundShare, BoundShareInput};
47pub use subquery::BoundSubquery;
48pub use table_or_source::{
49    BoundBaseTable, BoundIcebergMetadataTable, BoundSource, BoundSystemTable,
50};
51pub use watermark::BoundWatermark;
52pub use window_table_function::{BoundWindowTableFunction, WindowTableFunctionKind};
53
54use crate::expr::{CorrelatedId, Depth};
55
56/// A validated item that refers to a table-like entity, including base table, subquery, join, etc.
57/// It is usually part of the `from` clause.
58#[derive(Debug, Clone)]
59pub enum Relation {
60    Source(Box<BoundSource>),
61    BaseTable(Box<BoundBaseTable>),
62    SystemTable(Box<BoundSystemTable>),
63    IcebergMetadataTable(Box<BoundIcebergMetadataTable>),
64    Subquery(Box<BoundSubquery>),
65    Join(Box<BoundJoin>),
66    Apply(Box<BoundJoin>),
67    WindowTableFunction(Box<BoundWindowTableFunction>),
68    /// Table function or scalar function.
69    TableFunction {
70        expr: ExprImpl,
71        with_ordinality: bool,
72    },
73    Watermark(Box<BoundWatermark>),
74    Share(Box<BoundShare>),
75    GapFill(Box<BoundGapFill>),
76}
77
78impl RewriteExprsRecursive for Relation {
79    fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
80        match self {
81            Relation::Subquery(inner) => inner.rewrite_exprs_recursive(rewriter),
82            Relation::Join(inner) => inner.rewrite_exprs_recursive(rewriter),
83            Relation::Apply(inner) => inner.rewrite_exprs_recursive(rewriter),
84            Relation::WindowTableFunction(inner) => inner.rewrite_exprs_recursive(rewriter),
85            Relation::Watermark(inner) => inner.rewrite_exprs_recursive(rewriter),
86            Relation::Share(inner) => inner.rewrite_exprs_recursive(rewriter),
87            Relation::TableFunction { expr: inner, .. } => {
88                *inner = rewriter.rewrite_expr(inner.take())
89            }
90            _ => {}
91        }
92    }
93}
94
95impl Relation {
96    pub fn is_correlated_by_depth(&self, depth: Depth) -> bool {
97        match self {
98            Relation::Subquery(subquery) => subquery.query.is_correlated_by_depth(depth),
99            Relation::Join(join) => {
100                join.cond.has_correlated_input_ref_by_depth(depth)
101                    || join.left.is_correlated_by_depth(depth)
102                    || join.right.is_correlated_by_depth(depth)
103            }
104            // The right side of an `Apply` is bound in a scope extended by its left input. When
105            // looking for a reference owned by an enclosing `Apply`, cross that scope boundary.
106            Relation::Apply(join) => {
107                join.cond.has_correlated_input_ref_by_depth(depth)
108                    || join.left.is_correlated_by_depth(depth)
109                    || join.right.is_correlated_by_depth(depth + 1)
110            }
111            Relation::TableFunction {
112                expr: table_function,
113                with_ordinality: _,
114            } => table_function.has_correlated_input_ref_by_depth(depth + 1),
115            Relation::Share(share) => match &share.input {
116                BoundShareInput::Query(query) => query.is_correlated_by_depth(depth),
117                BoundShareInput::ChangeLog(change_log) => change_log.is_correlated_by_depth(depth),
118            },
119            _ => false,
120        }
121    }
122
123    pub fn is_correlated_by_correlated_id(&self, correlated_id: CorrelatedId) -> bool {
124        match self {
125            Relation::Subquery(subquery) => {
126                subquery.query.is_correlated_by_correlated_id(correlated_id)
127            }
128            Relation::Join(join) | Relation::Apply(join) => {
129                join.cond
130                    .has_correlated_input_ref_by_correlated_id(correlated_id)
131                    || join.left.is_correlated_by_correlated_id(correlated_id)
132                    || join.right.is_correlated_by_correlated_id(correlated_id)
133            }
134            Relation::TableFunction {
135                expr: table_function,
136                with_ordinality: _,
137            } => table_function.has_correlated_input_ref_by_correlated_id(correlated_id),
138            Relation::Share(share) => match &share.input {
139                BoundShareInput::Query(query) => {
140                    query.is_correlated_by_correlated_id(correlated_id)
141                }
142                BoundShareInput::ChangeLog(change_log) => {
143                    change_log.is_correlated_by_correlated_id(correlated_id)
144                }
145            },
146            _ => false,
147        }
148    }
149
150    pub fn collect_correlated_indices_by_depth_and_assign_id(
151        &mut self,
152        depth: Depth,
153        correlated_id: CorrelatedId,
154    ) -> Vec<usize> {
155        match self {
156            Relation::Subquery(subquery) => subquery
157                .query
158                .collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id),
159            Relation::Join(join) => {
160                let mut correlated_indices = vec![];
161                correlated_indices.extend(
162                    join.cond
163                        .collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id),
164                );
165                correlated_indices.extend(
166                    join.left
167                        .collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id),
168                );
169                correlated_indices.extend(
170                    join.right
171                        .collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id),
172                );
173                correlated_indices
174            }
175            Relation::Apply(join) => {
176                let mut correlated_indices = vec![];
177                correlated_indices.extend(
178                    join.cond
179                        .collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id),
180                );
181                correlated_indices.extend(
182                    join.left
183                        .collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id),
184                );
185                correlated_indices.extend(
186                    join.right
187                        .collect_correlated_indices_by_depth_and_assign_id(
188                            depth + 1,
189                            correlated_id,
190                        ),
191                );
192                correlated_indices
193            }
194            Relation::TableFunction {
195                expr: table_function,
196                with_ordinality: _,
197            } => table_function
198                .collect_correlated_indices_by_depth_and_assign_id(depth + 1, correlated_id),
199            Relation::Share(share) => match &mut share.input {
200                BoundShareInput::Query(query) => {
201                    query.collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id)
202                }
203                BoundShareInput::ChangeLog(change_log) => change_log
204                    .collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id),
205            },
206            _ => vec![],
207        }
208    }
209}
210
211#[derive(Debug)]
212#[non_exhaustive]
213pub enum ResolveQualifiedNameErrorKind {
214    QualifiedNameTooLong,
215    NotCurrentDatabase,
216}
217
218#[derive(Debug, Error)]
219pub struct ResolveQualifiedNameError {
220    qualified: String,
221    kind: ResolveQualifiedNameErrorKind,
222}
223
224impl std::fmt::Display for ResolveQualifiedNameError {
225    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
226        match self.kind {
227            ResolveQualifiedNameErrorKind::QualifiedNameTooLong => write!(
228                f,
229                "improper qualified name (too many dotted names): {}",
230                self.qualified
231            ),
232            ResolveQualifiedNameErrorKind::NotCurrentDatabase => write!(
233                f,
234                "cross-database references are not implemented: \"{}\"",
235                self.qualified
236            ),
237        }
238    }
239}
240
241impl ResolveQualifiedNameError {
242    pub fn new(qualified: String, kind: ResolveQualifiedNameErrorKind) -> Self {
243        Self { qualified, kind }
244    }
245}
246
247impl From<ResolveQualifiedNameError> for RwError {
248    fn from(e: ResolveQualifiedNameError) -> Self {
249        ErrorCode::InvalidInputSyntax(format!("{}", e.as_report())).into()
250    }
251}
252
253impl Binder {
254    /// return (`schema_name`, `name`)
255    pub fn resolve_schema_qualified_name(
256        db_name: &str,
257        name: &ObjectName,
258    ) -> std::result::Result<(Option<String>, String), ResolveQualifiedNameError> {
259        let formatted_name = name.to_string();
260        let mut identifiers = name.0.clone();
261
262        if identifiers.len() > 3 {
263            return Err(ResolveQualifiedNameError::new(
264                formatted_name,
265                ResolveQualifiedNameErrorKind::QualifiedNameTooLong,
266            ));
267        }
268
269        let name = identifiers.pop().unwrap().real_value();
270
271        let schema_name = identifiers.pop().map(|ident| ident.real_value());
272        let database_name = identifiers.pop().map(|ident| ident.real_value());
273
274        if let Some(database_name) = database_name
275            && database_name != db_name
276        {
277            return Err(ResolveQualifiedNameError::new(
278                formatted_name,
279                ResolveQualifiedNameErrorKind::NotCurrentDatabase,
280            ));
281        }
282
283        Ok((schema_name, name))
284    }
285
286    /// check whether the name is a cross-database reference
287    pub fn validate_cross_db_reference(
288        db_name: &str,
289        name: &ObjectName,
290    ) -> std::result::Result<(), ResolveQualifiedNameError> {
291        let formatted_name = name.to_string();
292        let identifiers = &name.0;
293        if identifiers.len() > 3 {
294            return Err(ResolveQualifiedNameError::new(
295                formatted_name,
296                ResolveQualifiedNameErrorKind::QualifiedNameTooLong,
297            ));
298        }
299
300        if identifiers.len() == 3 && identifiers[0].real_value() != db_name {
301            return Err(ResolveQualifiedNameError::new(
302                formatted_name,
303                ResolveQualifiedNameErrorKind::NotCurrentDatabase,
304            ));
305        }
306
307        Ok(())
308    }
309
310    /// return (`database_name`, `schema_name`, `name`)
311    pub fn resolve_db_schema_qualified_name(
312        name: &ObjectName,
313    ) -> std::result::Result<(Option<String>, Option<String>, String), ResolveQualifiedNameError>
314    {
315        let formatted_name = name.to_string();
316        let mut identifiers = name.0.clone();
317
318        if identifiers.len() > 3 {
319            return Err(ResolveQualifiedNameError::new(
320                formatted_name,
321                ResolveQualifiedNameErrorKind::QualifiedNameTooLong,
322            ));
323        }
324
325        let name = identifiers.pop().unwrap().real_value();
326        let schema_name = identifiers.pop().map(|ident| ident.real_value());
327        let database_name = identifiers.pop().map(|ident| ident.real_value());
328
329        Ok((database_name, schema_name, name))
330    }
331
332    /// return first name in identifiers, must have only one name.
333    fn resolve_single_name(mut identifiers: Vec<Ident>, ident_desc: &str) -> Result<String> {
334        if identifiers.len() > 1 {
335            bail!("{} must contain 1 argument", ident_desc);
336        }
337        let name = identifiers.pop().unwrap().real_value();
338
339        Ok(name)
340    }
341
342    /// return the `database_name`
343    pub fn resolve_database_name(name: ObjectName) -> Result<String> {
344        Self::resolve_single_name(name.0, "database name")
345    }
346
347    /// return the `schema_name`
348    pub fn resolve_schema_name(name: ObjectName) -> Result<String> {
349        Self::resolve_single_name(name.0, "schema name")
350    }
351
352    /// return the `index_name`
353    pub fn resolve_index_name(name: ObjectName) -> Result<String> {
354        Self::resolve_single_name(name.0, "index name")
355    }
356
357    /// return the `view_name`
358    pub fn resolve_view_name(name: ObjectName) -> Result<String> {
359        Self::resolve_single_name(name.0, "view name")
360    }
361
362    /// return the `sink_name`
363    pub fn resolve_sink_name(name: ObjectName) -> Result<String> {
364        Self::resolve_single_name(name.0, "sink name")
365    }
366
367    /// return the `subscription_name`
368    pub fn resolve_subscription_name(name: ObjectName) -> Result<String> {
369        Self::resolve_single_name(name.0, "subscription name")
370    }
371
372    /// return the `table_name`
373    pub fn resolve_table_name(name: ObjectName) -> Result<String> {
374        Self::resolve_single_name(name.0, "table name")
375    }
376
377    /// return the `source_name`
378    pub fn resolve_source_name(name: ObjectName) -> Result<String> {
379        Self::resolve_single_name(name.0, "source name")
380    }
381
382    /// return the `user_name`
383    pub fn resolve_user_name(name: ObjectName) -> Result<String> {
384        Self::resolve_single_name(name.0, "user name")
385    }
386
387    /// Fill the [`BindContext`](super::BindContext) for table.
388    pub(super) fn bind_table_to_context(
389        &mut self,
390        columns: impl IntoIterator<Item = (bool, Field)>, // bool indicates if the field is hidden
391        table_name: String,
392        schema_name: Option<String>,
393        alias: Option<&TableAlias>,
394    ) -> Result<()> {
395        const EMPTY: [Ident; 0] = [];
396        let (resolved_schema_name, table_name, column_aliases, table_alias) = match alias {
397            None => (schema_name.clone(), table_name, &EMPTY[..], None),
398            Some(TableAlias { name, columns }) => (
399                None,
400                name.real_value(),
401                columns.as_slice(),
402                Some(table_name),
403            ),
404        };
405
406        let num_col_aliases = column_aliases.len();
407
408        let begin = self.context.columns.len();
409        // Column aliases can be less than columns, but not more.
410        // It also needs to skip hidden columns.
411        let mut alias_iter = column_aliases.iter().fuse();
412        let mut index = 0;
413        columns.into_iter().for_each(|(is_hidden, mut field)| {
414            let name = match is_hidden {
415                true => field.name.clone(),
416                false => alias_iter
417                    .next()
418                    .map(|t| t.real_value())
419                    .unwrap_or_else(|| field.name.clone()),
420            };
421            field.name.clone_from(&name);
422            self.context.columns.push(ColumnBinding::new(
423                table_name.clone(),
424                schema_name.clone(),
425                table_alias.clone(),
426                begin + index,
427                is_hidden,
428                field,
429            ));
430            self.context
431                .indices_of
432                .entry(name)
433                .or_default()
434                .push(self.context.columns.len() - 1);
435            index += 1;
436        });
437
438        let num_cols = index;
439        if num_cols < num_col_aliases {
440            return Err(ErrorCode::BindError(format!(
441                "table \"{table_name}\" has {num_cols} columns available but {num_col_aliases} column aliases specified",
442            ))
443            .into());
444        }
445
446        match self
447            .context
448            .range_of
449            .entry((resolved_schema_name, table_name.clone()))
450        {
451            Entry::Occupied(_) => Err(ErrorCode::InternalError(format!(
452                "Duplicated table name while binding table to context: {}",
453                table_name
454            ))
455            .into()),
456            Entry::Vacant(entry) => {
457                entry.insert((begin, self.context.columns.len()));
458                Ok(())
459            }
460        }
461    }
462
463    /// Binds a relation, which can be:
464    /// - a table/source/materialized view
465    /// - a reference to a CTE
466    /// - a logical view
467    pub fn bind_relation_by_name(
468        &mut self,
469        name: &ObjectName,
470        alias: Option<&TableAlias>,
471        as_of: Option<&AsOf>,
472        allow_cross_db: bool,
473    ) -> Result<Relation> {
474        let (db_name, schema_name, table_name) = if allow_cross_db {
475            Self::resolve_db_schema_qualified_name(name)?
476        } else {
477            let (schema_name, table_name) =
478                Self::resolve_schema_qualified_name(&self.db_name, name)?;
479            (None, schema_name, table_name)
480        };
481
482        if schema_name.is_none()
483            // the `table_name` here is the name of the currently binding cte.
484            && let Some(item) = self.context.cte_to_relation.get(&table_name)
485        {
486            // Handles CTE
487
488            if as_of.is_some() {
489                return Err(ErrorCode::BindError(
490                    "Right table of a temporal join should not be a CTE. \
491                 It should be a table, index, or materialized view"
492                        .to_owned(),
493                )
494                .into());
495            }
496
497            let BindingCte {
498                share_id,
499                state: cte_state,
500                alias: mut original_alias,
501            } = item.deref().borrow().clone();
502
503            // The original CTE alias ought to be its table name.
504            debug_assert_eq!(original_alias.name.real_value(), table_name);
505
506            if let Some(from_alias) = alias {
507                original_alias.name = from_alias.name.clone();
508                original_alias.columns = original_alias
509                    .columns
510                    .into_iter()
511                    .zip_longest(from_alias.columns.iter().cloned())
512                    .map(EitherOrBoth::into_right)
513                    .collect();
514            }
515
516            let exposed_table_name = original_alias.name.real_value();
517            self.context
518                .check_relation_name_conflict(&exposed_table_name)?;
519
520            match cte_state {
521                BindingCteState::Bound { query } => {
522                    let input = BoundShareInput::Query(query);
523                    self.bind_table_to_context(
524                        input.fields()?,
525                        table_name,
526                        None,
527                        Some(&original_alias),
528                    )?;
529                    self.context.add_cte_name(exposed_table_name);
530                    // we could always share the cte,
531                    // no matter it's recursive or not.
532                    Ok(Relation::Share(Box::new(BoundShare { share_id, input })))
533                }
534                BindingCteState::ChangeLog { table } => {
535                    let input = BoundShareInput::ChangeLog(table);
536                    self.bind_table_to_context(
537                        input.fields()?,
538                        table_name,
539                        None,
540                        Some(&original_alias),
541                    )?;
542                    self.context.add_cte_name(exposed_table_name);
543                    Ok(Relation::Share(Box::new(BoundShare { share_id, input })))
544                }
545            }
546        } else {
547            let exposed_table_name = alias
548                .map(|alias| alias.name.real_value())
549                .unwrap_or_else(|| table_name.clone());
550            self.context.check_catalog_name(&exposed_table_name)?;
551            self.bind_catalog_relation_by_name(
552                db_name.as_deref(),
553                schema_name.as_deref(),
554                &table_name,
555                alias,
556                as_of,
557                false,
558            )
559        }
560    }
561
562    // Bind a relation provided a function arg.
563    fn bind_relation_by_function_arg(
564        &mut self,
565        arg: Option<&FunctionArg>,
566        err_msg: &str,
567    ) -> Result<(Relation, ObjectName)> {
568        let Some(FunctionArg::Unnamed(FunctionArgExpr::Expr(expr))) = arg else {
569            return Err(ErrorCode::BindError(err_msg.to_owned()).into());
570        };
571        let table_name = match expr {
572            ParserExpr::Identifier(ident) => Ok::<_, RwError>(ObjectName(vec![ident.clone()])),
573            ParserExpr::CompoundIdentifier(idents) => Ok(ObjectName(idents.clone())),
574            _ => Err(ErrorCode::BindError(err_msg.to_owned()).into()),
575        }?;
576
577        Ok((
578            self.bind_relation_by_name(&table_name, None, None, true)?,
579            table_name,
580        ))
581    }
582
583    // Bind column provided a function arg.
584    fn bind_column_by_function_args(
585        &mut self,
586        arg: Option<&FunctionArg>,
587        err_msg: &str,
588    ) -> Result<Box<InputRef>> {
589        if let Some(time_col_arg) = arg
590            && let Some(ExprImpl::InputRef(time_col)) =
591                self.bind_function_arg(time_col_arg)?.into_iter().next()
592        {
593            Ok(time_col)
594        } else {
595            Err(ErrorCode::BindError(err_msg.to_owned()).into())
596        }
597    }
598
599    /// `rw_table(table_id[,schema_name])` which queries internal table
600    fn bind_internal_table(
601        &mut self,
602        args: &[FunctionArg],
603        alias: Option<&TableAlias>,
604    ) -> Result<Relation> {
605        if args.is_empty() || args.len() > 2 {
606            return Err(
607                ErrorCode::BindError("usage: rw_table(table_id[,schema_name])".to_owned()).into(),
608            );
609        }
610
611        let table_id: TableId = args[0]
612            .to_string()
613            .parse::<u32>()
614            .map_err(|err| {
615                RwError::from(ErrorCode::BindError(format!(
616                    "invalid table id: {}",
617                    err.as_report()
618                )))
619            })?
620            .into();
621
622        let schema = args.get(1).map(|arg| arg.to_string());
623
624        let table_name = self.catalog.get_table_name_by_id(table_id)?;
625        self.bind_catalog_relation_by_name(None, schema.as_deref(), &table_name, alias, None, false)
626    }
627
628    pub(super) fn bind_table_factor(&mut self, table_factor: &TableFactor) -> Result<Relation> {
629        match table_factor {
630            TableFactor::Table { name, alias, as_of } => {
631                self.bind_relation_by_name(name, alias.as_ref(), as_of.as_ref(), true)
632            }
633            TableFactor::TableFunction {
634                name,
635                alias,
636                args,
637                with_ordinality,
638            } => {
639                let visibility = self.mark_lateral_contexts_visible();
640                let result = self.bind_table_function(name, alias.as_ref(), args, *with_ordinality);
641                self.restore_lateral_contexts_visibility(visibility);
642                result
643            }
644            TableFactor::Derived {
645                lateral,
646                subquery,
647                alias,
648            } => {
649                if *lateral {
650                    let visibility = self.mark_lateral_contexts_visible();
651
652                    // Bind lateral subquery here.
653                    let result = self.bind_subquery_relation(subquery, alias.as_ref(), true);
654
655                    self.restore_lateral_contexts_visibility(visibility);
656                    result.map(|subquery| Relation::Subquery(Box::new(subquery)))
657                } else {
658                    // Non-lateral subqueries to not have access to the join-tree context.
659                    self.push_lateral_context();
660                    let bound_subquery =
661                        self.bind_subquery_relation(subquery, alias.as_ref(), false)?;
662                    self.pop_and_merge_lateral_context()?;
663                    Ok(Relation::Subquery(Box::new(bound_subquery)))
664                }
665            }
666            TableFactor::NestedJoin(table_with_joins) => {
667                self.push_lateral_context();
668                let bound_join = self.bind_table_with_joins(table_with_joins)?;
669                self.pop_and_merge_lateral_context()?;
670                Ok(bound_join)
671            }
672        }
673    }
674}
risingwave_frontend/binder/relation/mod.rs

risingwave_frontend/binder/relation/
mod.rs
