1use std::collections::{HashMap, HashSet};
16use std::fmt::Debug;
17
18use itertools::Itertools;
19use risingwave_common::catalog::{Field, Schema};
20use risingwave_common::types::ScalarImpl;
21use risingwave_common::util::iter_util::ZipEqFast;
22use risingwave_sqlparser::ast::{
23 DataType as AstDataType, Distinct, Expr, Select, SelectItem, Value, WindowSpec,
24};
25
26use super::bind_context::{Clause, ColumnBinding};
27use super::statement::RewriteExprsRecursive;
28use super::{BoundShareInput, UNNAMED_COLUMN};
29use crate::binder::{Binder, Relation};
30use crate::catalog::check_column_name_not_reserved;
31use crate::error::{ErrorCode, Result, RwError};
32use crate::expr::{CorrelatedId, Depth, Expr as _, ExprImpl, ExprType, FunctionCall, InputRef};
33use crate::optimizer::plan_node::generic::CHANGELOG_OP;
34use crate::utils::group_by::GroupBy;
35
36#[derive(Debug, Clone)]
37pub struct BoundSelect {
38 pub distinct: BoundDistinct,
39 pub select_items: Vec<ExprImpl>,
40 pub aliases: Vec<Option<String>>,
41 pub from: Option<Relation>,
42 pub where_clause: Option<ExprImpl>,
43 pub group_by: GroupBy,
44 pub having: Option<ExprImpl>,
45 pub window: HashMap<String, WindowSpec>,
46 pub schema: Schema,
47}
48
49impl RewriteExprsRecursive for BoundSelect {
50 fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
51 self.distinct.rewrite_exprs_recursive(rewriter);
52
53 let new_select_items = std::mem::take(&mut self.select_items)
54 .into_iter()
55 .map(|expr| rewriter.rewrite_expr(expr))
56 .collect::<Vec<_>>();
57 self.select_items = new_select_items;
58
59 if let Some(from) = &mut self.from {
60 from.rewrite_exprs_recursive(rewriter);
61 }
62
63 self.where_clause =
64 std::mem::take(&mut self.where_clause).map(|expr| rewriter.rewrite_expr(expr));
65
66 let new_group_by = match &mut self.group_by {
67 GroupBy::GroupKey(group_key) => GroupBy::GroupKey(
68 std::mem::take(group_key)
69 .into_iter()
70 .map(|expr| rewriter.rewrite_expr(expr))
71 .collect::<Vec<_>>(),
72 ),
73 GroupBy::GroupingSets(grouping_sets) => GroupBy::GroupingSets(
74 std::mem::take(grouping_sets)
75 .into_iter()
76 .map(|set| {
77 set.into_iter()
78 .map(|expr| rewriter.rewrite_expr(expr))
79 .collect()
80 })
81 .collect::<Vec<_>>(),
82 ),
83 GroupBy::Rollup(rollup) => GroupBy::Rollup(
84 std::mem::take(rollup)
85 .into_iter()
86 .map(|set| {
87 set.into_iter()
88 .map(|expr| rewriter.rewrite_expr(expr))
89 .collect()
90 })
91 .collect::<Vec<_>>(),
92 ),
93 GroupBy::Cube(cube) => GroupBy::Cube(
94 std::mem::take(cube)
95 .into_iter()
96 .map(|set| {
97 set.into_iter()
98 .map(|expr| rewriter.rewrite_expr(expr))
99 .collect()
100 })
101 .collect::<Vec<_>>(),
102 ),
103 };
104 self.group_by = new_group_by;
105
106 self.having = std::mem::take(&mut self.having).map(|expr| rewriter.rewrite_expr(expr));
107 }
108}
109
110impl BoundSelect {
111 pub fn schema(&self) -> &Schema {
113 &self.schema
114 }
115
116 pub fn exprs(&self) -> impl Iterator<Item = &ExprImpl> {
117 self.select_items
118 .iter()
119 .chain(self.group_by.iter())
120 .chain(self.where_clause.iter())
121 .chain(self.having.iter())
122 }
123
124 pub fn exprs_mut(&mut self) -> impl Iterator<Item = &mut ExprImpl> {
125 self.select_items
126 .iter_mut()
127 .chain(self.group_by.iter_mut())
128 .chain(self.where_clause.iter_mut())
129 .chain(self.having.iter_mut())
130 }
131
132 pub fn is_correlated_by_depth(&self, depth: Depth) -> bool {
133 self.exprs()
134 .any(|expr| expr.has_correlated_input_ref_by_depth(depth))
135 || match self.from.as_ref() {
136 Some(relation) => relation.is_correlated_by_depth(depth),
137 None => false,
138 }
139 }
140
141 pub fn is_correlated_by_correlated_id(&self, correlated_id: CorrelatedId) -> bool {
142 self.exprs()
143 .any(|expr| expr.has_correlated_input_ref_by_correlated_id(correlated_id))
144 || match self.from.as_ref() {
145 Some(relation) => relation.is_correlated_by_correlated_id(correlated_id),
146 None => 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 let mut correlated_indices = self
156 .exprs_mut()
157 .flat_map(|expr| {
158 expr.collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id)
159 })
160 .collect_vec();
161
162 if let Some(relation) = self.from.as_mut() {
163 correlated_indices.extend(
164 relation.collect_correlated_indices_by_depth_and_assign_id(depth, correlated_id),
165 );
166 }
167
168 correlated_indices
169 }
170}
171
172#[derive(Debug, Clone)]
173pub enum BoundDistinct {
174 All,
175 Distinct,
176 DistinctOn(Vec<ExprImpl>),
177}
178
179impl RewriteExprsRecursive for BoundDistinct {
180 fn rewrite_exprs_recursive(&mut self, rewriter: &mut impl crate::expr::ExprRewriter) {
181 if let Self::DistinctOn(exprs) = self {
182 let new_exprs = std::mem::take(exprs)
183 .into_iter()
184 .map(|expr| rewriter.rewrite_expr(expr))
185 .collect::<Vec<_>>();
186 exprs.extend(new_exprs);
187 }
188 }
189}
190
191impl BoundDistinct {
192 pub const fn is_all(&self) -> bool {
193 matches!(self, Self::All)
194 }
195
196 pub const fn is_distinct(&self) -> bool {
197 matches!(self, Self::Distinct)
198 }
199}
200
201impl Binder {
202 pub(super) fn bind_select(&mut self, select: Select) -> Result<BoundSelect> {
203 let from = self.bind_vec_table_with_joins(select.from)?;
205
206 let mut named_windows = HashMap::new();
208 for named_window in &select.window {
209 let window_name = named_window.name.real_value();
210 if named_windows.contains_key(&window_name) {
211 return Err(ErrorCode::InvalidInputSyntax(format!(
212 "window \"{}\" is already defined",
213 window_name
214 ))
215 .into());
216 }
217 named_windows.insert(window_name, named_window.window_spec.clone());
218 }
219
220 self.context.named_windows = named_windows.clone();
222
223 let (select_items, aliases) = self.bind_select_list(select.projection)?;
225 let out_name_to_index = Self::build_name_to_index(aliases.iter().filter_map(Clone::clone));
226
227 let distinct = self.bind_distinct_on(select.distinct, &out_name_to_index, &select_items)?;
229
230 self.context.clause = Some(Clause::Where);
232 let selection = select
233 .selection
234 .map(|expr| {
235 self.bind_expr(expr)
236 .and_then(|expr| expr.enforce_bool_clause("WHERE"))
237 })
238 .transpose()?;
239 self.context.clause = None;
240
241 self.context.clause = Some(Clause::GroupBy);
243
244 let group_by = if select.group_by.len() == 1
246 && let Expr::GroupingSets(grouping_sets) = &select.group_by[0]
247 {
248 GroupBy::GroupingSets(self.bind_grouping_items_expr_in_select(
249 grouping_sets.clone(),
250 &out_name_to_index,
251 &select_items,
252 )?)
253 } else if select.group_by.len() == 1
254 && let Expr::Rollup(rollup) = &select.group_by[0]
255 {
256 GroupBy::Rollup(self.bind_grouping_items_expr_in_select(
257 rollup.clone(),
258 &out_name_to_index,
259 &select_items,
260 )?)
261 } else if select.group_by.len() == 1
262 && let Expr::Cube(cube) = &select.group_by[0]
263 {
264 GroupBy::Cube(self.bind_grouping_items_expr_in_select(
265 cube.clone(),
266 &out_name_to_index,
267 &select_items,
268 )?)
269 } else {
270 if select.group_by.iter().any(|expr| {
271 matches!(expr, Expr::GroupingSets(_))
272 || matches!(expr, Expr::Rollup(_))
273 || matches!(expr, Expr::Cube(_))
274 }) {
275 return Err(ErrorCode::BindError(
276 "Only support one grouping item in group by clause".to_owned(),
277 )
278 .into());
279 }
280 GroupBy::GroupKey(
281 select
282 .group_by
283 .into_iter()
284 .map(|expr| {
285 self.bind_group_by_expr_in_select(expr, &out_name_to_index, &select_items)
286 })
287 .try_collect()?,
288 )
289 };
290 self.context.clause = None;
291
292 self.context.clause = Some(Clause::Having);
294 let having = select
295 .having
296 .map(|expr| {
297 self.bind_expr(expr)
298 .and_then(|expr| expr.enforce_bool_clause("HAVING"))
299 })
300 .transpose()?;
301 self.context.clause = None;
302
303 let fields = select_items
305 .iter()
306 .zip_eq_fast(aliases.iter())
307 .map(|(s, a)| {
308 let name = a.clone().unwrap_or_else(|| UNNAMED_COLUMN.to_owned());
309 Ok(Field::with_name(s.return_type(), name))
310 })
311 .collect::<Result<Vec<Field>>>()?;
312
313 if let Some(Relation::Share(bound)) = &from {
314 if matches!(bound.input, BoundShareInput::ChangeLog(_))
315 && fields.iter().filter(|&x| x.name.eq(CHANGELOG_OP)).count() > 1
316 {
317 return Err(ErrorCode::BindError(
318 "The source table of changelog cannot have `changelog_op`, please rename it first".to_owned()
319 )
320 .into());
321 }
322 }
323
324 Ok(BoundSelect {
325 distinct,
326 select_items,
327 aliases,
328 from,
329 where_clause: selection,
330 group_by,
331 having,
332 window: named_windows,
333 schema: Schema { fields },
334 })
335 }
336
337 pub fn bind_select_list(
338 &mut self,
339 select_items: Vec<SelectItem>,
340 ) -> Result<(Vec<ExprImpl>, Vec<Option<String>>)> {
341 let mut select_list = vec![];
342 let mut aliases = vec![];
343 for item in select_items {
344 match item {
345 SelectItem::UnnamedExpr(expr) => {
346 let alias = derive_alias(&expr);
347 let bound = self.bind_expr(expr)?;
348 select_list.push(bound);
349 aliases.push(alias);
350 }
351 SelectItem::ExprWithAlias { expr, alias } => {
352 check_column_name_not_reserved(&alias.real_value())?;
353
354 let expr = self.bind_expr(expr)?;
355 select_list.push(expr);
356 aliases.push(Some(alias.real_value()));
357 }
358 SelectItem::QualifiedWildcard(obj_name, except) => {
359 let table_name = &obj_name.0.last().unwrap().real_value();
360 let except_indices = self.generate_except_indices(except)?;
361 let (begin, end) = self.context.range_of.get(table_name).ok_or_else(|| {
362 ErrorCode::ItemNotFound(format!("relation \"{}\"", table_name))
363 })?;
364 let (exprs, names) = Self::iter_bound_columns(
365 self.context.columns[*begin..*end]
366 .iter()
367 .filter(|c| !c.is_hidden && !except_indices.contains(&c.index)),
368 );
369 select_list.extend(exprs);
370 aliases.extend(names);
371 }
372 SelectItem::ExprQualifiedWildcard(expr, prefix) => {
373 let (exprs, names) = self.bind_wildcard_field_column(expr, prefix)?;
374 select_list.extend(exprs);
375 aliases.extend(names);
376 }
377 SelectItem::Wildcard(except) => {
378 if self.context.range_of.is_empty() {
379 return Err(ErrorCode::BindError(
380 "SELECT * with no tables specified is not valid".into(),
381 )
382 .into());
383 }
384
385 let (exprs, names) = self.iter_column_groups();
389 select_list.extend(exprs);
390 aliases.extend(names);
391
392 let except_indices = self.generate_except_indices(except)?;
393
394 let (exprs, names) =
396 Self::iter_bound_columns(self.context.columns[..].iter().filter(|c| {
397 !c.is_hidden
398 && !self
399 .context
400 .column_group_context
401 .mapping
402 .contains_key(&c.index)
403 && !except_indices.contains(&c.index)
404 }));
405
406 select_list.extend(exprs);
407 aliases.extend(names);
408 }
417 }
418 }
419 assert_eq!(select_list.len(), aliases.len());
420 Ok((select_list, aliases))
421 }
422
423 fn bind_group_by_expr_in_select(
445 &mut self,
446 expr: Expr,
447 name_to_index: &HashMap<String, usize>,
448 select_items: &[ExprImpl],
449 ) -> Result<ExprImpl> {
450 let name = match &expr {
451 Expr::Identifier(ident) => Some(ident.real_value()),
452 _ => None,
453 };
454 match self.bind_expr(expr) {
455 Ok(ExprImpl::Literal(lit)) => match lit.get_data() {
456 Some(ScalarImpl::Int32(idx)) => idx
457 .saturating_sub(1)
458 .try_into()
459 .ok()
460 .and_then(|i: usize| select_items.get(i).cloned())
461 .ok_or_else(|| {
462 ErrorCode::BindError(format!(
463 "GROUP BY position {idx} is not in select list"
464 ))
465 .into()
466 }),
467 _ => Err(ErrorCode::BindError("non-integer constant in GROUP BY".into()).into()),
468 },
469 Ok(e) => Ok(e),
470 Err(e) => match name {
471 None => Err(e),
472 Some(name) => match name_to_index.get(&name) {
473 None => Err(e),
474 Some(&usize::MAX) => Err(ErrorCode::BindError(format!(
475 "GROUP BY \"{name}\" is ambiguous"
476 ))
477 .into()),
478 Some(out_idx) => Ok(select_items[*out_idx].clone()),
479 },
480 },
481 }
482 }
483
484 fn bind_grouping_items_expr_in_select(
485 &mut self,
486 grouping_items: Vec<Vec<Expr>>,
487 name_to_index: &HashMap<String, usize>,
488 select_items: &[ExprImpl],
489 ) -> Result<Vec<Vec<ExprImpl>>> {
490 let mut result = vec![];
491 for set in grouping_items {
492 let mut set_exprs = vec![];
493 for expr in set {
494 let name = match &expr {
495 Expr::Identifier(ident) => Some(ident.real_value()),
496 _ => None,
497 };
498 let expr_impl = match self.bind_expr(expr) {
499 Ok(ExprImpl::Literal(lit)) => match lit.get_data() {
500 Some(ScalarImpl::Int32(idx)) => idx
501 .saturating_sub(1)
502 .try_into()
503 .ok()
504 .and_then(|i: usize| select_items.get(i).cloned())
505 .ok_or_else(|| {
506 ErrorCode::BindError(format!(
507 "GROUP BY position {idx} is not in select list"
508 ))
509 .into()
510 }),
511 _ => Err(
512 ErrorCode::BindError("non-integer constant in GROUP BY".into()).into(),
513 ),
514 },
515 Ok(e) => Ok(e),
516 Err(e) => match name {
517 None => Err(e),
518 Some(name) => match name_to_index.get(&name) {
519 None => Err(e),
520 Some(&usize::MAX) => Err(ErrorCode::BindError(format!(
521 "GROUP BY \"{name}\" is ambiguous"
522 ))
523 .into()),
524 Some(out_idx) => Ok(select_items[*out_idx].clone()),
525 },
526 },
527 };
528
529 set_exprs.push(expr_impl?);
530 }
531 result.push(set_exprs);
532 }
533 Ok(result)
534 }
535
536 pub fn bind_returning_list(
537 &mut self,
538 returning_items: Vec<SelectItem>,
539 ) -> Result<(Vec<ExprImpl>, Vec<Field>)> {
540 let (returning_list, aliases) = self.bind_select_list(returning_items)?;
541 if returning_list
542 .iter()
543 .any(|expr| expr.has_agg_call() || expr.has_window_function())
544 {
545 return Err(RwError::from(ErrorCode::BindError(
546 "should not have agg/window in the `RETURNING` list".to_owned(),
547 )));
548 }
549
550 let fields = returning_list
551 .iter()
552 .zip_eq_fast(aliases.iter())
553 .map(|(s, a)| {
554 let name = a.clone().unwrap_or_else(|| UNNAMED_COLUMN.to_owned());
555 Ok::<Field, RwError>(Field::with_name(s.return_type(), name))
556 })
557 .try_collect()?;
558 Ok((returning_list, fields))
559 }
560
561 pub fn iter_bound_columns<'a>(
562 column_binding: impl Iterator<Item = &'a ColumnBinding>,
563 ) -> (Vec<ExprImpl>, Vec<Option<String>>) {
564 column_binding
565 .map(|c| {
566 (
567 InputRef::new(c.index, c.field.data_type.clone()).into(),
568 Some(c.field.name.clone()),
569 )
570 })
571 .unzip()
572 }
573
574 pub fn iter_column_groups(&self) -> (Vec<ExprImpl>, Vec<Option<String>>) {
575 self.context
576 .column_group_context
577 .groups
578 .values()
579 .rev() .map(|g| {
581 if let Some(col) = &g.non_nullable_column {
582 let c = &self.context.columns[*col];
583 (
584 InputRef::new(c.index, c.field.data_type.clone()).into(),
585 Some(c.field.name.clone()),
586 )
587 } else {
588 let mut input_idxes = g.indices.iter().collect::<Vec<_>>();
589 input_idxes.sort();
590 let inputs = input_idxes
591 .into_iter()
592 .map(|index| {
593 let column = &self.context.columns[*index];
594 InputRef::new(column.index, column.field.data_type.clone()).into()
595 })
596 .collect::<Vec<_>>();
597 let c = &self.context.columns[*g.indices.iter().next().unwrap()];
598 (
599 FunctionCall::new(ExprType::Coalesce, inputs)
600 .expect("Failure binding COALESCE function call")
601 .into(),
602 Some(c.field.name.clone()),
603 )
604 }
605 })
606 .unzip()
607 }
608
609 fn bind_distinct_on(
624 &mut self,
625 distinct: Distinct,
626 name_to_index: &HashMap<String, usize>,
627 select_items: &[ExprImpl],
628 ) -> Result<BoundDistinct> {
629 Ok(match distinct {
630 Distinct::All => BoundDistinct::All,
631 Distinct::Distinct => BoundDistinct::Distinct,
632 Distinct::DistinctOn(exprs) => {
633 let mut bound_exprs = vec![];
634 for expr in exprs {
635 let expr_impl = match expr {
636 Expr::Identifier(name)
637 if let Some(index) = name_to_index.get(&name.real_value()) =>
638 {
639 match *index {
640 usize::MAX => {
641 return Err(ErrorCode::BindError(format!(
642 "DISTINCT ON \"{}\" is ambiguous",
643 name.real_value()
644 ))
645 .into());
646 }
647 _ => select_items[*index].clone(),
648 }
649 }
650 Expr::Value(Value::Number(number)) => match number.parse::<usize>() {
651 Ok(index) if 1 <= index && index <= select_items.len() => {
652 let idx_from_0 = index - 1;
653 select_items[idx_from_0].clone()
654 }
655 _ => {
656 return Err(ErrorCode::InvalidInputSyntax(format!(
657 "Invalid ordinal number in DISTINCT ON: {}",
658 number
659 ))
660 .into());
661 }
662 },
663 expr => self.bind_expr(expr)?,
664 };
665 bound_exprs.push(expr_impl);
666 }
667 BoundDistinct::DistinctOn(bound_exprs)
668 }
669 })
670 }
671
672 fn generate_except_indices(&mut self, except: Option<Vec<Expr>>) -> Result<HashSet<usize>> {
673 let mut except_indices: HashSet<usize> = HashSet::new();
674 if let Some(exprs) = except {
675 for expr in exprs {
676 let bound = self.bind_expr(expr)?;
677 match bound {
678 ExprImpl::InputRef(inner) => {
679 if !except_indices.insert(inner.index) {
680 return Err(ErrorCode::BindError(
681 "Duplicate entry in except list".into(),
682 )
683 .into());
684 }
685 }
686 _ => {
687 return Err(ErrorCode::BindError(
688 "Only support column name in except list".into(),
689 )
690 .into());
691 }
692 }
693 }
694 }
695 Ok(except_indices)
696 }
697}
698
699fn derive_alias(expr: &Expr) -> Option<String> {
700 match expr.clone() {
701 Expr::Identifier(ident) => Some(ident.real_value()),
702 Expr::CompoundIdentifier(idents) => idents.last().map(|ident| ident.real_value()),
703 Expr::FieldIdentifier(_, idents) => idents.last().map(|ident| ident.real_value()),
704 Expr::Function(func) => Some(func.name.real_value()),
705 Expr::Extract { .. } => Some("extract".to_owned()),
706 Expr::Case { .. } => Some("case".to_owned()),
707 Expr::Cast { expr, data_type } => {
708 derive_alias(&expr).or_else(|| data_type_to_alias(&data_type))
709 }
710 Expr::TypedString { data_type, .. } => data_type_to_alias(&data_type),
711 Expr::Value(Value::Interval { .. }) => Some("interval".to_owned()),
712 Expr::Row(_) => Some("row".to_owned()),
713 Expr::Array(_) => Some("array".to_owned()),
714 Expr::Index { obj, index: _ } => derive_alias(&obj),
715 _ => None,
716 }
717}
718
719fn data_type_to_alias(data_type: &AstDataType) -> Option<String> {
720 let alias = match data_type {
721 AstDataType::Char(_) => "bpchar".to_owned(),
722 AstDataType::Varchar => "varchar".to_owned(),
723 AstDataType::Uuid => "uuid".to_owned(),
724 AstDataType::Decimal(_, _) => "numeric".to_owned(),
725 AstDataType::Real | AstDataType::Float(Some(1..=24)) => "float4".to_owned(),
726 AstDataType::Double | AstDataType::Float(Some(25..=53) | None) => "float8".to_owned(),
727 AstDataType::Float(Some(0 | 54..)) => unreachable!(),
728 AstDataType::SmallInt => "int2".to_owned(),
729 AstDataType::Int => "int4".to_owned(),
730 AstDataType::BigInt => "int8".to_owned(),
731 AstDataType::Boolean => "bool".to_owned(),
732 AstDataType::Date => "date".to_owned(),
733 AstDataType::Time(tz) => format!("time{}", if *tz { "z" } else { "" }),
734 AstDataType::Timestamp(tz) => {
735 format!("timestamp{}", if *tz { "tz" } else { "" })
736 }
737 AstDataType::Interval => "interval".to_owned(),
738 AstDataType::Regclass => "regclass".to_owned(),
739 AstDataType::Regproc => "regproc".to_owned(),
740 AstDataType::Text => "text".to_owned(),
741 AstDataType::Bytea => "bytea".to_owned(),
742 AstDataType::Jsonb => "jsonb".to_owned(),
743 AstDataType::Array(ty) => return data_type_to_alias(ty),
744 AstDataType::Custom(ty) => format!("{}", ty),
745 AstDataType::Struct(_) | AstDataType::Map(_) => {
746 return None;
748 }
749 };
750
751 Some(alias)
752}