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

risingwave_frontend/binder/relation/
join.rs
