risingwave_frontend/binder/relation/
join.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 risingwave_pb::plan_common::JoinType;
16use risingwave_sqlparser::ast::{
17    BinaryOperator, Expr, Ident, JoinConstraint, JoinOperator, TableFactor, TableWithJoins, Value,
18};
19
20use crate::binder::bind_context::BindContext;
21use crate::binder::statement::RewriteExprsRecursive;
22use crate::binder::{Binder, COLUMN_GROUP_PREFIX, Clause, Relation};
23use crate::error::{ErrorCode, Result};
24use crate::expr::{ExprImpl, ScopeId};
25
26#[derive(Debug, Clone)]
27pub struct BoundJoin {
28    pub join_type: JoinType,
29    pub left: Relation,
30    pub right: Relation,
31    pub cond: ExprImpl,
32    /// The scope holding the columns of `left` while `right` was bound. References from `right`
33    /// owned by it make this join an `Apply`.
34    pub left_scope: ScopeId,
35}
36
37impl RewriteExprsRecursive for BoundJoin {
38    fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
39        self.left.rewrite_exprs_recursive(rewriter);
40        self.right.rewrite_exprs_recursive(rewriter);
41        self.cond = rewriter.rewrite_expr(self.cond.take());
42    }
43}
44
45impl Binder {
46    pub(crate) fn bind_vec_table_with_joins(
47        &mut self,
48        from: &[TableWithJoins],
49    ) -> Result<Option<Relation>> {
50        let mut from_iter = from.iter();
51        let first = match from_iter.next() {
52            Some(t) => t,
53            None => return Ok(None),
54        };
55        self.push_lateral_context();
56        let mut root = self.bind_table_with_joins(first)?;
57        self.pop_and_merge_lateral_context()?;
58        for t in from_iter {
59            self.push_lateral_context();
60            let right = self.bind_table_with_joins(t)?;
61            self.pop_and_merge_lateral_context()?;
62
63            let left_scope = self.context.scope;
64            root = if Self::requires_apply(&right, left_scope) {
65                Relation::Apply(Box::new(BoundJoin {
66                    join_type: JoinType::Inner,
67                    left: root,
68                    right,
69                    cond: ExprImpl::literal_bool(true),
70                    left_scope,
71                }))
72            } else {
73                Relation::Join(Box::new(BoundJoin {
74                    join_type: JoinType::Inner,
75                    left: root,
76                    right,
77                    cond: ExprImpl::literal_bool(true),
78                    left_scope,
79                }))
80            }
81        }
82        Ok(Some(root))
83    }
84
85    fn requires_apply(relation: &Relation, left_scope: ScopeId) -> bool {
86        matches!(relation, Relation::Subquery(subquery) if subquery.lateral)
87            || matches!(relation, Relation::TableFunction { .. })
88            || relation.is_correlated_by_scope(left_scope)
89    }
90
91    pub(crate) fn bind_table_with_joins(&mut self, table: &TableWithJoins) -> Result<Relation> {
92        let mut root = self.bind_table_factor(&table.relation)?;
93        for join in &table.joins {
94            let (constraint, join_type) = match &join.join_operator {
95                JoinOperator::Inner(constraint) => (constraint, JoinType::Inner),
96                JoinOperator::LeftOuter(constraint) => (constraint, JoinType::LeftOuter),
97                JoinOperator::RightOuter(constraint) => (constraint, JoinType::RightOuter),
98                JoinOperator::FullOuter(constraint) => (constraint, JoinType::FullOuter),
99                // Cross join equals to inner join with with no constraint.
100                JoinOperator::CrossJoin => (&JoinConstraint::None, JoinType::Inner),
101                JoinOperator::AsOfInner(constraint) => (constraint, JoinType::AsofInner),
102                JoinOperator::AsOfLeft(constraint) => (constraint, JoinType::AsofLeftOuter),
103            };
104            let right: Relation;
105            let cond: ExprImpl;
106            if matches!(
107                constraint.clone(),
108                JoinConstraint::Using(_) | JoinConstraint::Natural
109            ) {
110                let option_rel: Option<Relation>;
111                (cond, option_rel) =
112                    self.bind_join_constraint(constraint, Some(&join.relation), join_type)?;
113                right = option_rel.unwrap();
114            } else {
115                right = self.bind_table_factor(&join.relation)?;
116                (cond, _) = self.bind_join_constraint(constraint, None, join_type)?;
117            }
118
119            let left_scope = self.context.scope;
120            root = if Self::requires_apply(&right, left_scope) {
121                match join_type {
122                    JoinType::Inner | JoinType::LeftOuter => {}
123                    _ => {
124                        return Err(ErrorCode::InvalidInputSyntax("The combining JOIN type must be INNER or LEFT for a LATERAL reference.".to_owned())
125                            .into());
126                    }
127                }
128
129                Relation::Apply(Box::new(BoundJoin {
130                    join_type,
131                    left: root,
132                    right,
133                    cond,
134                    left_scope,
135                }))
136            } else {
137                Relation::Join(Box::new(BoundJoin {
138                    join_type,
139                    left: root,
140                    right,
141                    cond,
142                    left_scope,
143                }))
144            };
145        }
146
147        Ok(root)
148    }
149
150    fn bind_join_constraint(
151        &mut self,
152        constraint: &JoinConstraint,
153        table_factor: Option<&TableFactor>,
154        join_type: JoinType,
155    ) -> Result<(ExprImpl, Option<Relation>)> {
156        Ok(match constraint {
157            JoinConstraint::None => (ExprImpl::literal_bool(true), None),
158            c @ JoinConstraint::Natural | c @ JoinConstraint::Using(_) => {
159                // First, we identify columns with the same name.
160                let old_context = self.context.clone();
161                let l_len = old_context.columns.len();
162                // Bind this table factor to an empty context
163                self.push_lateral_context();
164                let table_factor = table_factor.unwrap();
165                let relation = self.bind_table_factor(table_factor)?;
166
167                let using_columns = match c {
168                    JoinConstraint::Natural => None,
169                    JoinConstraint::Using(cols) => {
170                        // sanity check
171                        for col in cols {
172                            if !old_context.indices_of.contains_key(&col.real_value()) {
173                                return Err(ErrorCode::ItemNotFound(format!("column \"{}\" specified in USING clause does not exist in left table", col.real_value())).into());
174                            }
175                            if !self.context.indices_of.contains_key(&col.real_value()) {
176                                return Err(ErrorCode::ItemNotFound(format!("column \"{}\" specified in USING clause does not exist in right table", col.real_value())).into());
177                            }
178                        }
179                        Some(cols)
180                    }
181                    _ => unreachable!(),
182                };
183
184                let mut columns = self
185                    .context
186                    .indices_of
187                    .iter()
188                    .filter(|(_, idxs)| idxs.iter().all(|i| !self.context.columns[*i].is_hidden))
189                    .map(|(s, idxes)| (Ident::from_real_value(s), idxes))
190                    .collect::<Vec<_>>();
191                columns.sort_by_key(|a| a.0.real_value());
192
193                let mut col_indices = Vec::new();
194                let mut binary_expr = Expr::Value(Value::Boolean(true));
195
196                // Walk the RHS cols, checking to see if any share a name with any LHS cols
197                for (column, indices_r) in columns {
198                    // TODO: is it ok to ignore quote style?
199                    // If we have a `USING` constraint, we only bind the columns appearing in the
200                    // constraint.
201                    if let Some(cols) = &using_columns
202                        && !cols.contains(&column)
203                    {
204                        continue;
205                    }
206                    let indices_l = match old_context.get_unqualified_indices(&column.real_value())
207                    {
208                        Err(e) => {
209                            if let ErrorCode::ItemNotFound(_) = e {
210                                continue;
211                            } else {
212                                return Err(e.into());
213                            }
214                        }
215                        Ok(idxs) => idxs,
216                    };
217                    // Select at most one column from each natural column group from left and right
218                    col_indices.push((indices_l[0], indices_r[0] + l_len));
219                    let left_expr = Self::get_identifier_from_indices(
220                        &old_context,
221                        &indices_l,
222                        column.clone(),
223                    )?;
224                    let right_expr = Self::get_identifier_from_indices(
225                        &self.context,
226                        indices_r,
227                        column.clone(),
228                    )?;
229                    binary_expr = Expr::BinaryOp {
230                        left: Box::new(binary_expr),
231                        op: BinaryOperator::And,
232                        right: Box::new(Expr::BinaryOp {
233                            left: Box::new(left_expr),
234                            op: BinaryOperator::Eq,
235                            right: Box::new(right_expr),
236                        }),
237                    }
238                }
239                self.pop_and_merge_lateral_context()?;
240                // Bind the expression first, before allowing disambiguation of the columns involved
241                // in the join
242                let expr = self.bind_expr(&binary_expr)?;
243                for (l, r) in col_indices {
244                    let non_nullable = match join_type {
245                        JoinType::LeftOuter | JoinType::Inner => Some(l),
246                        JoinType::RightOuter => Some(r),
247                        JoinType::FullOuter => None,
248                        _ => unreachable!(),
249                    };
250                    self.context.add_natural_columns(l, r, non_nullable);
251                }
252                (expr, Some(relation))
253            }
254            JoinConstraint::On(expr) => {
255                let clause = self.context.clause;
256                self.context.clause = Some(Clause::JoinOn);
257                let bound_expr: ExprImpl = self
258                    .bind_expr(expr)
259                    .and_then(|expr| expr.enforce_bool_clause("JOIN ON"))?;
260                self.context.clause = clause;
261                (bound_expr, None)
262            }
263        })
264    }
265
266    fn get_identifier_from_indices(
267        context: &BindContext,
268        indices: &[usize],
269        column: Ident,
270    ) -> Result<Expr> {
271        if indices.len() == 1 {
272            let right_table = context.columns[indices[0]].table_name.as_ref();
273            Ok(Expr::CompoundIdentifier(vec![
274                Ident::from_real_value(right_table),
275                column,
276            ]))
277        } else if let Some(group_id) = context.column_group_context.mapping.get(&indices[0]) {
278            Ok(Expr::CompoundIdentifier(vec![
279                Ident::from_real_value(&format!("{COLUMN_GROUP_PREFIX}{}", group_id)),
280                column,
281            ]))
282        } else {
283            Err(
284                ErrorCode::InternalError(format!("Ambiguous column name: {}", column.real_value()))
285                    .into(),
286            )
287        }
288    }
289}
risingwave_frontend/binder/relation/join.rs

risingwave_frontend/binder/relation/
join.rs
