risingwave_frontend/binder/
bind_context.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::cell::RefCell;
16use std::collections::hash_map::Entry;
17use std::collections::{BTreeMap, HashMap, HashSet};
18use std::rc::Rc;
19
20use parse_display::Display;
21use risingwave_common::catalog::Field;
22use risingwave_common::types::DataType;
23use risingwave_sqlparser::ast::{TableAlias, WindowSpec};
24
25use crate::binder::Relation;
26use crate::error::{ErrorCode, Result};
27use crate::expr::ExprImpl;
28
29type LiteResult<T> = std::result::Result<T, ErrorCode>;
30
31use crate::binder::{BoundQuery, COLUMN_GROUP_PREFIX, ShareId};
32
33#[derive(Debug, Clone)]
34pub struct ColumnBinding {
35    pub table_name: String,
36    pub schema_name: Option<String>,
37    /// if the table has table alias, `table_alias` store the original table name
38    pub table_alias: Option<String>,
39    pub index: usize,
40    pub is_hidden: bool,
41    pub field: Field,
42}
43
44impl ColumnBinding {
45    pub fn new(
46        table_name: String,
47        schema_name: Option<String>,
48        table_alias: Option<String>,
49        index: usize,
50        is_hidden: bool,
51        field: Field,
52    ) -> Self {
53        ColumnBinding {
54            table_name,
55            schema_name,
56            table_alias,
57            index,
58            is_hidden,
59            field,
60        }
61    }
62}
63
64#[derive(Clone, Copy, PartialEq, Eq, Hash, Debug, Display)]
65#[display(style = "TITLE CASE")]
66pub enum Clause {
67    Where,
68    Values,
69    GroupBy,
70    JoinOn,
71    Having,
72    Filter,
73    From,
74    GeneratedColumn,
75    Insert,
76}
77
78/// A `BindContext` that is only visible if the `LATERAL` keyword
79/// is provided.
80pub struct LateralBindContext {
81    pub is_visible: bool,
82    pub context: BindContext,
83}
84
85#[derive(Debug, Clone)]
86pub enum BindingCteState {
87    /// We get the whole bound result of the CTE.
88    Bound { query: BoundQuery },
89
90    ChangeLog {
91        table: Relation,
92        key_indices: Option<Vec<usize>>,
93    },
94}
95
96#[derive(Clone, Debug)]
97pub struct BindingCte {
98    pub share_id: ShareId,
99    pub state: BindingCteState,
100    pub alias: TableAlias,
101}
102
103#[derive(Default, Debug, Clone)]
104pub struct BindContext {
105    // Columns of all tables.
106    pub columns: Vec<ColumnBinding>,
107    // Mapping column name to indices in `columns`.
108    pub indices_of: HashMap<String, Vec<usize>>,
109    // Mapping (schema name, table name) to [begin, end) of its columns.
110    pub range_of: HashMap<(Option<String>, String), (usize, usize)>,
111    // `clause` identifies in what clause we are binding.
112    pub clause: Option<Clause>,
113    // The `BindContext`'s data on its column groups
114    pub column_group_context: ColumnGroupContext,
115    /// Map the cte's name to its binding state.
116    /// The `ShareId` in `BindingCte` of the value is used to help the planner identify the share plan.
117    pub cte_to_relation: HashMap<String, Rc<RefCell<BindingCte>>>,
118    /// Exposed relation names of CTE references in the current FROM scope.
119    pub cte_relation_names: HashSet<String>,
120    /// Current lambda functions's arguments
121    pub lambda_args: Option<HashMap<String, (usize, DataType)>>,
122    /// Whether the security invoker is set, currently only used for views.
123    pub disable_security_invoker: bool,
124    /// Named window definitions from the `WINDOW` clause
125    pub named_windows: HashMap<String, WindowSpec>,
126    /// Bound arguments for the current SQL UDF, if any.
127    // TODO: use enum for named or positional arguments
128    pub sql_udf_arguments: Option<HashMap<String, ExprImpl>>,
129}
130
131/// Holds the context for the `BindContext`'s `ColumnGroup`s.
132#[derive(Default, Debug, Clone)]
133pub struct ColumnGroupContext {
134    // Maps naturally-joined/USING columns to their column group id
135    pub mapping: HashMap<usize, u32>,
136    // Maps column group ids to their column group data
137    // We use a BTreeMap to ensure that iteration over the groups is ordered.
138    pub groups: BTreeMap<u32, ColumnGroup>,
139
140    next_group_id: u32,
141}
142
143/// When binding a natural join or a join with USING, a `ColumnGroup` contains the columns with the
144/// same name.
145#[derive(Default, Debug, Clone)]
146pub struct ColumnGroup {
147    /// Indices of the columns in the group
148    pub indices: HashSet<usize>,
149    /// A non-nullable column is never NULL.
150    /// If `None`, ambiguous references to the column name will be resolved to a `COALESCE(col1,
151    /// col2, ..., coln)` over each column in the group
152    pub non_nullable_column: Option<usize>,
153
154    pub column_name: Option<String>,
155}
156
157impl BindContext {
158    pub fn get_column_binding_index(
159        &self,
160        schema_name: &Option<String>,
161        table_name: &Option<String>,
162        column_name: &String,
163    ) -> LiteResult<usize> {
164        match &self.get_column_binding_indices(schema_name, table_name, column_name)?[..] {
165            [] => unreachable!(),
166            [idx] => Ok(*idx),
167            _ => Err(ErrorCode::InvalidReference(format!(
168                "column reference \"{}\" is ambiguous",
169                column_name
170            ))),
171        }
172    }
173
174    /// If return Vec has len > 1, it means we have an unqualified reference to a column which has
175    /// been naturally joined upon, wherein none of the columns are min-nullable. This will be
176    /// handled in downstream as a `COALESCE` expression
177    pub fn get_column_binding_indices(
178        &self,
179        schema_name: &Option<String>,
180        table_name: &Option<String>,
181        column_name: &String,
182    ) -> LiteResult<Vec<usize>> {
183        match table_name {
184            Some(table_name) => {
185                if let Some(group_id_str) = table_name.strip_prefix(COLUMN_GROUP_PREFIX) {
186                    let group_id = group_id_str.parse::<u32>().map_err(|_|ErrorCode::InternalError(
187                        format!("Could not parse {:?} as virtual table name `{COLUMN_GROUP_PREFIX}[group_id]`", table_name)))?;
188                    self.get_indices_with_group_id(group_id, column_name)
189                } else {
190                    Ok(vec![self.get_index_with_table_name(
191                        column_name,
192                        table_name,
193                        schema_name,
194                    )?])
195                }
196            }
197            None => self.get_unqualified_indices(column_name),
198        }
199    }
200
201    pub fn get_table_alias(
202        &self,
203        schema_name: &String,
204        table_name: &String,
205        column_name: &String,
206    ) -> LiteResult<Option<usize>> {
207        let column_indexes = self
208            .indices_of
209            .get(column_name)
210            .ok_or_else(|| ErrorCode::ItemNotFound(format!("Invalid column: {}", column_name)))?;
211        for index in column_indexes {
212            let column = &self.columns[*index];
213            if let (Some(schema), Some(table_alias)) = (&column.schema_name, &column.table_alias)
214                && schema == schema_name
215                && table_alias == table_name
216            {
217                return Ok(Some(*index));
218            }
219        }
220        Ok(None)
221    }
222
223    fn get_indices_with_group_id(
224        &self,
225        group_id: u32,
226        column_name: &String,
227    ) -> LiteResult<Vec<usize>> {
228        // The group-id qualifier is also reachable from a quoted user identifier, so a missing
229        // group or a group of another column is "not found" rather than an internal invariant.
230        let Some(group) = self
231            .column_group_context
232            .groups
233            .get(&group_id)
234            .filter(|group| {
235                group
236                    .column_name
237                    .as_ref()
238                    .is_none_or(|name| name == column_name)
239            })
240        else {
241            return Err(ErrorCode::ItemNotFound(format!(
242                "Invalid column: {column_name}"
243            )));
244        };
245        if let Some(non_nullable) = &group.non_nullable_column {
246            Ok(vec![*non_nullable])
247        } else {
248            // These will be converted to a `COALESCE(col1, col2, ..., coln)`
249            let mut indices: Vec<_> = group.indices.iter().copied().collect();
250            indices.sort(); // ensure a deterministic result
251            Ok(indices)
252        }
253    }
254
255    pub fn get_unqualified_indices(&self, column_name: &String) -> LiteResult<Vec<usize>> {
256        let columns = self
257            .indices_of
258            .get(column_name)
259            .ok_or_else(|| ErrorCode::ItemNotFound(format!("Invalid column: {column_name}")))?;
260        if columns.len() > 1 {
261            // If there is some group containing the columns and the ambiguous columns are all in
262            // the group
263            if let Some(group_id) = self.column_group_context.mapping.get(&columns[0]) {
264                let group = self.column_group_context.groups.get(group_id).unwrap();
265                if columns.iter().all(|idx| group.indices.contains(idx)) {
266                    if let Some(non_nullable) = &group.non_nullable_column {
267                        return Ok(vec![*non_nullable]);
268                    } else {
269                        // These will be converted to a `COALESCE(col1, col2, ..., coln)`
270                        return Ok(columns.clone());
271                    }
272                }
273            }
274            Err(ErrorCode::InvalidReference(format!(
275                "column reference \"{}\" is ambiguous",
276                column_name
277            )))
278        } else {
279            Ok(columns.clone())
280        }
281    }
282
283    pub fn check_catalog_name(&self, exposed_relation_name: &str) -> Result<()> {
284        if self.cte_relation_names.contains(exposed_relation_name) {
285            return Err(ErrorCode::DuplicateRelationName(format!(
286                "table name \"{}\" specified more than once",
287                exposed_relation_name
288            ))
289            .into());
290        }
291        Ok(())
292    }
293
294    pub fn add_cte_name(&mut self, exposed_relation_name: String) {
295        self.cte_relation_names.insert(exposed_relation_name);
296    }
297
298    fn has_relation_name(&self, relation_name: &str) -> bool {
299        self.range_of
300            .keys()
301            .any(|(_, existing_name)| existing_name == relation_name)
302    }
303
304    pub fn check_relation_name_conflict(&self, relation_name: &str) -> Result<()> {
305        if self.has_relation_name(relation_name) {
306            return Err(ErrorCode::DuplicateRelationName(format!(
307                "table name \"{}\" specified more than once",
308                relation_name
309            ))
310            .into());
311        }
312        Ok(())
313    }
314
315    fn check_cte_relation_name_conflict(&self, other: &Self) -> Result<()> {
316        // `self` may bind a CTE in a separate context, while `other` already has a catalog
317        // relation with the same exposed name.
318        for cte_relation_name in &self.cte_relation_names {
319            other.check_relation_name_conflict(cte_relation_name)?;
320        }
321        // `other` may bind a CTE in a separate context, while `self` already has a catalog
322        // relation with the same exposed name.
323        for cte_relation_name in &other.cte_relation_names {
324            self.check_relation_name_conflict(cte_relation_name)?;
325        }
326        Ok(())
327    }
328
329    /// Identifies two columns as being in the same group. Additionally, possibly provides one of
330    /// the columns as being `non_nullable`
331    pub fn add_natural_columns(
332        &mut self,
333        left: usize,
334        right: usize,
335        non_nullable_column: Option<usize>,
336    ) {
337        match (
338            self.column_group_context.mapping.get(&left).copied(),
339            self.column_group_context.mapping.get(&right).copied(),
340        ) {
341            (None, None) => {
342                let group_id = self.column_group_context.next_group_id;
343                self.column_group_context.next_group_id += 1;
344
345                let group = ColumnGroup {
346                    indices: HashSet::from([left, right]),
347                    non_nullable_column,
348                    column_name: Some(self.columns[left].field.name.clone()),
349                };
350                self.column_group_context.groups.insert(group_id, group);
351                self.column_group_context.mapping.insert(left, group_id);
352                self.column_group_context.mapping.insert(right, group_id);
353            }
354            (Some(group_id), None) => {
355                let group = self.column_group_context.groups.get_mut(&group_id).unwrap();
356                group.indices.insert(right);
357                if group.non_nullable_column.is_none() {
358                    group.non_nullable_column = non_nullable_column;
359                }
360                self.column_group_context.mapping.insert(right, group_id);
361            }
362            (None, Some(group_id)) => {
363                let group = self.column_group_context.groups.get_mut(&group_id).unwrap();
364                group.indices.insert(left);
365                if group.non_nullable_column.is_none() {
366                    group.non_nullable_column = non_nullable_column;
367                }
368                self.column_group_context.mapping.insert(left, group_id);
369            }
370            (Some(l_group_id), Some(r_group_id)) => {
371                if r_group_id == l_group_id {
372                    return;
373                }
374
375                let r_group = self
376                    .column_group_context
377                    .groups
378                    .remove(&r_group_id)
379                    .unwrap();
380                let l_group = self
381                    .column_group_context
382                    .groups
383                    .get_mut(&l_group_id)
384                    .unwrap();
385
386                for idx in &r_group.indices {
387                    *self.column_group_context.mapping.get_mut(idx).unwrap() = l_group_id;
388                    l_group.indices.insert(*idx);
389                }
390                if l_group.non_nullable_column.is_none() {
391                    l_group.non_nullable_column = if r_group.non_nullable_column.is_none() {
392                        non_nullable_column
393                    } else {
394                        r_group.non_nullable_column
395                    };
396                }
397            }
398        }
399    }
400
401    /// Resolve a qualified column reference (`table.column` or `schema.table.column`)
402    /// to one concrete column index in the current bind scope.
403    ///
404    /// This function intentionally uses a two-step strategy:
405    /// 1. Resolve relation first (`resolve_relation_range`), including ambiguity checks.
406    /// 2. Resolve column inside that unique relation range (`resolve_column_in_range`).
407    fn get_index_with_table_name(
408        &self,
409        column_name: &String,
410        table_name: &String,
411        schema_name: &Option<String>,
412    ) -> LiteResult<usize> {
413        let chosen_range = self.resolve_relation_range(table_name, schema_name)?;
414        self.resolve_column_in_range(column_name, table_name, chosen_range)
415    }
416
417    /// Resolve the relation range for a `table_name` in the current bind scope.
418    ///
419    /// Alias matches are prioritized over base relation name matches.
420    pub(super) fn resolve_relation_range(
421        &self,
422        table_name: &String,
423        schema_name: &Option<String>,
424    ) -> LiteResult<(usize, usize)> {
425        let mut alias_ranges = Vec::new();
426        let mut base_ranges = Vec::new();
427
428        for ((rel_schema, rel_name), range) in &self.range_of {
429            if rel_name != table_name
430                || !schema_name
431                    .as_deref()
432                    .is_none_or(|s| rel_schema.as_deref() == Some(s))
433            {
434                continue;
435            }
436
437            // Empty ranges have no column binding from which to read `table_alias`. Explicit
438            // aliases are stored without a schema in `range_of`.
439            let is_alias = if range.0 == range.1 {
440                rel_schema.is_none()
441            } else {
442                self.columns[range.0].table_alias.is_some()
443            };
444            if is_alias {
445                alias_ranges.push(*range);
446            } else {
447                base_ranges.push(*range);
448            }
449        }
450
451        let chosen_ranges = if alias_ranges.is_empty() {
452            base_ranges
453        } else {
454            alias_ranges
455        };
456
457        match chosen_ranges.as_slice() {
458            [] => Err(ErrorCode::ItemNotFound(format!(
459                "missing FROM-clause entry for table \"{}\"",
460                table_name
461            ))),
462            [range] => Ok(*range),
463            _ => Err(ErrorCode::InvalidReference(format!(
464                "table reference \"{}\" is ambiguous",
465                table_name
466            ))),
467        }
468    }
469
470    /// Resolve `column_name` inside a pre-resolved relation range.
471    fn resolve_column_in_range(
472        &self,
473        column_name: &String,
474        table_name: &String,
475        range: (usize, usize),
476    ) -> LiteResult<usize> {
477        let idxs = self
478            .indices_of
479            .get(column_name)
480            .ok_or_else(|| ErrorCode::ItemNotFound(format!("Invalid column: {}", column_name)))?;
481
482        let matched: Vec<_> = idxs
483            .iter()
484            .copied()
485            .filter(|index| (range.0..range.1).contains(index))
486            .collect();
487
488        match matched.as_slice() {
489            [] => Err(ErrorCode::ItemNotFound(format!(
490                "missing FROM-clause entry for table \"{}\"",
491                table_name
492            ))),
493            [column_index] => Ok(*column_index),
494            _ => Err(ErrorCode::InvalidReference(format!(
495                "column reference \"{}\" is ambiguous",
496                column_name
497            ))),
498        }
499    }
500
501    /// Merges two `BindContext`s which are adjacent. For instance, the `BindContext` of two
502    /// adjacent cross-joined tables.
503    pub fn merge_context(&mut self, other: Self) -> Result<()> {
504        self.check_cte_relation_name_conflict(&other)?;
505
506        let begin = self.columns.len();
507        self.columns.extend(other.columns.into_iter().map(|mut c| {
508            c.index += begin;
509            c
510        }));
511        for (k, v) in other.indices_of {
512            let entry = self.indices_of.entry(k).or_default();
513            entry.extend(v.into_iter().map(|x| x + begin));
514        }
515        for (k, (x, y)) in other.range_of {
516            match self.range_of.entry(k) {
517                Entry::Occupied(e) => {
518                    return Err(ErrorCode::InternalError(format!(
519                        "Duplicated table name while merging adjacent contexts: {}",
520                        e.key().1
521                    ))
522                    .into());
523                }
524                Entry::Vacant(entry) => {
525                    entry.insert((begin + x, begin + y));
526                }
527            }
528        }
529        self.cte_relation_names.extend(other.cte_relation_names);
530        // To merge the column_group_contexts, we just need to offset RHS
531        // with the next_group_id of LHS.
532        let ColumnGroupContext {
533            mapping,
534            groups,
535            next_group_id,
536        } = other.column_group_context;
537
538        let offset = self.column_group_context.next_group_id;
539        for (idx, group_id) in mapping {
540            self.column_group_context
541                .mapping
542                .insert(begin + idx, offset + group_id);
543        }
544        for (group_id, mut group) in groups {
545            group.indices = group.indices.into_iter().map(|idx| idx + begin).collect();
546            if let Some(col) = &mut group.non_nullable_column {
547                *col += begin;
548            }
549            self.column_group_context
550                .groups
551                .insert(offset + group_id, group);
552        }
553        self.column_group_context.next_group_id += next_group_id;
554
555        // we assume that the clause is contained in the outer-level context
556        Ok(())
557    }
558}
559
560impl BindContext {
561    pub fn new() -> Self {
562        Self::default()
563    }
564}
risingwave_frontend/binder/bind_context.rs

risingwave_frontend/binder/
bind_context.rs
