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, ScopeId};
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    /// Identifies this context as the owner of correlated references to its columns.
130    pub scope: ScopeId,
131}
132
133/// Holds the context for the `BindContext`'s `ColumnGroup`s.
134#[derive(Default, Debug, Clone)]
135pub struct ColumnGroupContext {
136    // Maps naturally-joined/USING columns to their column group id
137    pub mapping: HashMap<usize, u32>,
138    // Maps column group ids to their column group data
139    // We use a BTreeMap to ensure that iteration over the groups is ordered.
140    pub groups: BTreeMap<u32, ColumnGroup>,
141
142    next_group_id: u32,
143}
144
145/// When binding a natural join or a join with USING, a `ColumnGroup` contains the columns with the
146/// same name.
147#[derive(Default, Debug, Clone)]
148pub struct ColumnGroup {
149    /// Indices of the columns in the group
150    pub indices: HashSet<usize>,
151    /// A non-nullable column is never NULL.
152    /// If `None`, ambiguous references to the column name will be resolved to a `COALESCE(col1,
153    /// col2, ..., coln)` over each column in the group
154    pub non_nullable_column: Option<usize>,
155
156    pub column_name: Option<String>,
157}
158
159impl BindContext {
160    pub fn get_column_binding_index(
161        &self,
162        schema_name: &Option<String>,
163        table_name: &Option<String>,
164        column_name: &String,
165    ) -> LiteResult<usize> {
166        match &self.get_column_binding_indices(schema_name, table_name, column_name)?[..] {
167            [] => unreachable!(),
168            [idx] => Ok(*idx),
169            _ => Err(ErrorCode::InvalidReference(format!(
170                "column reference \"{}\" is ambiguous",
171                column_name
172            ))),
173        }
174    }
175
176    /// If return Vec has len > 1, it means we have an unqualified reference to a column which has
177    /// been naturally joined upon, wherein none of the columns are min-nullable. This will be
178    /// handled in downstream as a `COALESCE` expression
179    pub fn get_column_binding_indices(
180        &self,
181        schema_name: &Option<String>,
182        table_name: &Option<String>,
183        column_name: &String,
184    ) -> LiteResult<Vec<usize>> {
185        match table_name {
186            Some(table_name) => {
187                if let Some(group_id_str) = table_name.strip_prefix(COLUMN_GROUP_PREFIX) {
188                    let group_id = group_id_str.parse::<u32>().map_err(|_|ErrorCode::InternalError(
189                        format!("Could not parse {:?} as virtual table name `{COLUMN_GROUP_PREFIX}[group_id]`", table_name)))?;
190                    self.get_indices_with_group_id(group_id, column_name)
191                } else {
192                    Ok(vec![self.get_index_with_table_name(
193                        column_name,
194                        table_name,
195                        schema_name,
196                    )?])
197                }
198            }
199            None => self.get_unqualified_indices(column_name),
200        }
201    }
202
203    pub fn get_table_alias(
204        &self,
205        schema_name: &String,
206        table_name: &String,
207        column_name: &String,
208    ) -> LiteResult<Option<usize>> {
209        let column_indexes = self
210            .indices_of
211            .get(column_name)
212            .ok_or_else(|| ErrorCode::ItemNotFound(format!("Invalid column: {}", column_name)))?;
213        for index in column_indexes {
214            let column = &self.columns[*index];
215            if let (Some(schema), Some(table_alias)) = (&column.schema_name, &column.table_alias)
216                && schema == schema_name
217                && table_alias == table_name
218            {
219                return Ok(Some(*index));
220            }
221        }
222        Ok(None)
223    }
224
225    fn get_indices_with_group_id(
226        &self,
227        group_id: u32,
228        column_name: &String,
229    ) -> LiteResult<Vec<usize>> {
230        // The group-id qualifier is also reachable from a quoted user identifier, so a missing
231        // group or a group of another column is "not found" rather than an internal invariant.
232        let Some(group) = self
233            .column_group_context
234            .groups
235            .get(&group_id)
236            .filter(|group| {
237                group
238                    .column_name
239                    .as_ref()
240                    .is_none_or(|name| name == column_name)
241            })
242        else {
243            return Err(ErrorCode::ItemNotFound(format!(
244                "Invalid column: {column_name}"
245            )));
246        };
247        if let Some(non_nullable) = &group.non_nullable_column {
248            Ok(vec![*non_nullable])
249        } else {
250            // These will be converted to a `COALESCE(col1, col2, ..., coln)`
251            let mut indices: Vec<_> = group.indices.iter().copied().collect();
252            indices.sort(); // ensure a deterministic result
253            Ok(indices)
254        }
255    }
256
257    pub fn get_unqualified_indices(&self, column_name: &String) -> LiteResult<Vec<usize>> {
258        let columns = self
259            .indices_of
260            .get(column_name)
261            .ok_or_else(|| ErrorCode::ItemNotFound(format!("Invalid column: {column_name}")))?;
262        if columns.len() > 1 {
263            // If there is some group containing the columns and the ambiguous columns are all in
264            // the group
265            if let Some(group_id) = self.column_group_context.mapping.get(&columns[0]) {
266                let group = self.column_group_context.groups.get(group_id).unwrap();
267                if columns.iter().all(|idx| group.indices.contains(idx)) {
268                    if let Some(non_nullable) = &group.non_nullable_column {
269                        return Ok(vec![*non_nullable]);
270                    } else {
271                        // These will be converted to a `COALESCE(col1, col2, ..., coln)`
272                        return Ok(columns.clone());
273                    }
274                }
275            }
276            Err(ErrorCode::InvalidReference(format!(
277                "column reference \"{}\" is ambiguous",
278                column_name
279            )))
280        } else {
281            Ok(columns.clone())
282        }
283    }
284
285    pub fn check_catalog_name(&self, exposed_relation_name: &str) -> Result<()> {
286        if self.cte_relation_names.contains(exposed_relation_name) {
287            return Err(ErrorCode::DuplicateRelationName(format!(
288                "table name \"{}\" specified more than once",
289                exposed_relation_name
290            ))
291            .into());
292        }
293        Ok(())
294    }
295
296    pub fn add_cte_name(&mut self, exposed_relation_name: String) {
297        self.cte_relation_names.insert(exposed_relation_name);
298    }
299
300    fn has_relation_name(&self, relation_name: &str) -> bool {
301        self.range_of
302            .keys()
303            .any(|(_, existing_name)| existing_name == relation_name)
304    }
305
306    pub fn check_relation_name_conflict(&self, relation_name: &str) -> Result<()> {
307        if self.has_relation_name(relation_name) {
308            return Err(ErrorCode::DuplicateRelationName(format!(
309                "table name \"{}\" specified more than once",
310                relation_name
311            ))
312            .into());
313        }
314        Ok(())
315    }
316
317    fn check_cte_relation_name_conflict(&self, other: &Self) -> Result<()> {
318        // `self` may bind a CTE in a separate context, while `other` already has a catalog
319        // relation with the same exposed name.
320        for cte_relation_name in &self.cte_relation_names {
321            other.check_relation_name_conflict(cte_relation_name)?;
322        }
323        // `other` may bind a CTE in a separate context, while `self` already has a catalog
324        // relation with the same exposed name.
325        for cte_relation_name in &other.cte_relation_names {
326            self.check_relation_name_conflict(cte_relation_name)?;
327        }
328        Ok(())
329    }
330
331    /// Identifies two columns as being in the same group. Additionally, possibly provides one of
332    /// the columns as being `non_nullable`
333    pub fn add_natural_columns(
334        &mut self,
335        left: usize,
336        right: usize,
337        non_nullable_column: Option<usize>,
338    ) {
339        match (
340            self.column_group_context.mapping.get(&left).copied(),
341            self.column_group_context.mapping.get(&right).copied(),
342        ) {
343            (None, None) => {
344                let group_id = self.column_group_context.next_group_id;
345                self.column_group_context.next_group_id += 1;
346
347                let group = ColumnGroup {
348                    indices: HashSet::from([left, right]),
349                    non_nullable_column,
350                    column_name: Some(self.columns[left].field.name.clone()),
351                };
352                self.column_group_context.groups.insert(group_id, group);
353                self.column_group_context.mapping.insert(left, group_id);
354                self.column_group_context.mapping.insert(right, group_id);
355            }
356            (Some(group_id), None) => {
357                let group = self.column_group_context.groups.get_mut(&group_id).unwrap();
358                group.indices.insert(right);
359                if group.non_nullable_column.is_none() {
360                    group.non_nullable_column = non_nullable_column;
361                }
362                self.column_group_context.mapping.insert(right, group_id);
363            }
364            (None, Some(group_id)) => {
365                let group = self.column_group_context.groups.get_mut(&group_id).unwrap();
366                group.indices.insert(left);
367                if group.non_nullable_column.is_none() {
368                    group.non_nullable_column = non_nullable_column;
369                }
370                self.column_group_context.mapping.insert(left, group_id);
371            }
372            (Some(l_group_id), Some(r_group_id)) => {
373                if r_group_id == l_group_id {
374                    return;
375                }
376
377                let r_group = self
378                    .column_group_context
379                    .groups
380                    .remove(&r_group_id)
381                    .unwrap();
382                let l_group = self
383                    .column_group_context
384                    .groups
385                    .get_mut(&l_group_id)
386                    .unwrap();
387
388                for idx in &r_group.indices {
389                    *self.column_group_context.mapping.get_mut(idx).unwrap() = l_group_id;
390                    l_group.indices.insert(*idx);
391                }
392                if l_group.non_nullable_column.is_none() {
393                    l_group.non_nullable_column = if r_group.non_nullable_column.is_none() {
394                        non_nullable_column
395                    } else {
396                        r_group.non_nullable_column
397                    };
398                }
399            }
400        }
401    }
402
403    /// Resolve a qualified column reference (`table.column` or `schema.table.column`)
404    /// to one concrete column index in the current bind scope.
405    ///
406    /// This function intentionally uses a two-step strategy:
407    /// 1. Resolve relation first (`resolve_relation_range`), including ambiguity checks.
408    /// 2. Resolve column inside that unique relation range (`resolve_column_in_range`).
409    fn get_index_with_table_name(
410        &self,
411        column_name: &String,
412        table_name: &String,
413        schema_name: &Option<String>,
414    ) -> LiteResult<usize> {
415        let chosen_range = self.resolve_relation_range(table_name, schema_name)?;
416        self.resolve_column_in_range(column_name, table_name, chosen_range)
417    }
418
419    /// Resolve the relation range for a `table_name` in the current bind scope.
420    ///
421    /// Alias matches are prioritized over base relation name matches.
422    pub(super) fn resolve_relation_range(
423        &self,
424        table_name: &String,
425        schema_name: &Option<String>,
426    ) -> LiteResult<(usize, usize)> {
427        let mut alias_ranges = Vec::new();
428        let mut base_ranges = Vec::new();
429
430        for ((rel_schema, rel_name), range) in &self.range_of {
431            if rel_name != table_name
432                || !schema_name
433                    .as_deref()
434                    .is_none_or(|s| rel_schema.as_deref() == Some(s))
435            {
436                continue;
437            }
438
439            // Empty ranges have no column binding from which to read `table_alias`. Explicit
440            // aliases are stored without a schema in `range_of`.
441            let is_alias = if range.0 == range.1 {
442                rel_schema.is_none()
443            } else {
444                self.columns[range.0].table_alias.is_some()
445            };
446            if is_alias {
447                alias_ranges.push(*range);
448            } else {
449                base_ranges.push(*range);
450            }
451        }
452
453        let chosen_ranges = if alias_ranges.is_empty() {
454            base_ranges
455        } else {
456            alias_ranges
457        };
458
459        match chosen_ranges.as_slice() {
460            [] => Err(ErrorCode::ItemNotFound(format!(
461                "missing FROM-clause entry for table \"{}\"",
462                table_name
463            ))),
464            [range] => Ok(*range),
465            _ => Err(ErrorCode::InvalidReference(format!(
466                "table reference \"{}\" is ambiguous",
467                table_name
468            ))),
469        }
470    }
471
472    /// Resolve `column_name` inside a pre-resolved relation range.
473    fn resolve_column_in_range(
474        &self,
475        column_name: &String,
476        table_name: &String,
477        range: (usize, usize),
478    ) -> LiteResult<usize> {
479        let idxs = self
480            .indices_of
481            .get(column_name)
482            .ok_or_else(|| ErrorCode::ItemNotFound(format!("Invalid column: {}", column_name)))?;
483
484        let matched: Vec<_> = idxs
485            .iter()
486            .copied()
487            .filter(|index| (range.0..range.1).contains(index))
488            .collect();
489
490        match matched.as_slice() {
491            [] => Err(ErrorCode::ItemNotFound(format!(
492                "missing FROM-clause entry for table \"{}\"",
493                table_name
494            ))),
495            [column_index] => Ok(*column_index),
496            _ => Err(ErrorCode::InvalidReference(format!(
497                "column reference \"{}\" is ambiguous",
498                column_name
499            ))),
500        }
501    }
502
503    /// Merges two `BindContext`s which are adjacent. For instance, the `BindContext` of two
504    /// adjacent cross-joined tables.
505    pub fn merge_context(&mut self, other: Self) -> Result<()> {
506        self.check_cte_relation_name_conflict(&other)?;
507
508        let begin = self.columns.len();
509        self.columns.extend(other.columns.into_iter().map(|mut c| {
510            c.index += begin;
511            c
512        }));
513        for (k, v) in other.indices_of {
514            let entry = self.indices_of.entry(k).or_default();
515            entry.extend(v.into_iter().map(|x| x + begin));
516        }
517        for (k, (x, y)) in other.range_of {
518            match self.range_of.entry(k) {
519                Entry::Occupied(e) => {
520                    return Err(ErrorCode::InternalError(format!(
521                        "Duplicated table name while merging adjacent contexts: {}",
522                        e.key().1
523                    ))
524                    .into());
525                }
526                Entry::Vacant(entry) => {
527                    entry.insert((begin + x, begin + y));
528                }
529            }
530        }
531        self.cte_relation_names.extend(other.cte_relation_names);
532        // To merge the column_group_contexts, we just need to offset RHS
533        // with the next_group_id of LHS.
534        let ColumnGroupContext {
535            mapping,
536            groups,
537            next_group_id,
538        } = other.column_group_context;
539
540        let offset = self.column_group_context.next_group_id;
541        for (idx, group_id) in mapping {
542            self.column_group_context
543                .mapping
544                .insert(begin + idx, offset + group_id);
545        }
546        for (group_id, mut group) in groups {
547            group.indices = group.indices.into_iter().map(|idx| idx + begin).collect();
548            if let Some(col) = &mut group.non_nullable_column {
549                *col += begin;
550            }
551            self.column_group_context
552                .groups
553                .insert(offset + group_id, group);
554        }
555        self.column_group_context.next_group_id += next_group_id;
556
557        // we assume that the clause is contained in the outer-level context
558        Ok(())
559    }
560}
561
562impl BindContext {
563    pub fn new() -> Self {
564        Self::default()
565    }
566}
risingwave_frontend/binder/bind_context.rs

risingwave_frontend/binder/
bind_context.rs
