risingwave_frontend/binder/mod.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::collections::{HashMap, HashSet};
16use std::sync::Arc;
17
18use itertools::Itertools;
19use parking_lot::RwLock;
20use risingwave_common::catalog::FunctionId;
21use risingwave_common::session_config::{SearchPath, SessionConfig};
22use risingwave_common::types::DataType;
23use risingwave_common::util::iter_util::ZipEqDebug;
24use risingwave_sqlparser::ast::Statement;
25
26use crate::error::Result;
27
28mod bind_context;
29mod bind_param;
30mod create;
31mod create_view;
32mod declare_cursor;
33mod delete;
34mod expr;
35pub mod fetch_cursor;
36mod for_system;
37mod gap_fill_binder;
38mod insert;
39mod query;
40mod relation;
41mod select;
42mod set_expr;
43mod statement;
44mod struct_field;
45mod update;
46mod values;
47
48pub use bind_context::{BindContext, Clause, LateralBindContext};
49pub use create_view::BoundCreateView;
50pub use delete::BoundDelete;
51pub use expr::bind_data_type;
52pub use gap_fill_binder::BoundFillStrategy;
53pub use insert::BoundInsert;
54use pgwire::pg_server::{Session, SessionId};
55pub use query::BoundQuery;
56pub use relation::{
57 BoundBaseTable, BoundGapFill, BoundIcebergMetadataTable, BoundJoin, BoundMatchRecognize,
58 BoundMeasure, BoundShare, BoundShareInput, BoundSource, BoundSymbolDefinition,
59 BoundSystemTable, BoundWatermark, BoundWindowTableFunction, MeasureSlotKind, Relation,
60 ResolveQualifiedNameError, WindowTableFunctionKind,
61};
62// Re-export common types
63pub use risingwave_common::gap_fill::FillStrategy;
64use risingwave_common::id::ObjectId;
65pub use select::{BoundDistinct, BoundSelect};
66pub use set_expr::*;
67pub use statement::BoundStatement;
68pub use update::{BoundUpdate, UpdateProject};
69pub use values::BoundValues;
70
71use crate::catalog::catalog_service::CatalogReadGuard;
72use crate::catalog::root_catalog::SchemaPath;
73use crate::catalog::schema_catalog::SchemaCatalog;
74use crate::catalog::{CatalogResult, DatabaseId, SecretId, ViewId};
75use crate::error::ErrorCode;
76use crate::session::{AuthContext, SessionImpl, StagingCatalogManager, TemporarySourceManager};
77use crate::user::user_service::UserInfoReadGuard;
78
79pub type ShareId = usize;
80
81/// The type of binding statement.
82enum BindFor {
83 /// Binding MV/SINK
84 Stream,
85 /// Binding a batch query
86 Batch,
87 /// Binding a DDL (e.g. CREATE TABLE/SOURCE)
88 Ddl,
89 /// Binding a system query (e.g. SHOW)
90 System,
91}
92
93/// `Binder` binds the identifiers in AST to columns in relations
94pub struct Binder {
95 // TODO: maybe we can only lock the database, but not the whole catalog.
96 catalog: CatalogReadGuard,
97 user: UserInfoReadGuard,
98 db_name: String,
99 database_id: DatabaseId,
100 session_id: SessionId,
101 context: BindContext,
102 auth_context: Arc<AuthContext>,
103 /// A stack holding contexts of outer queries when binding a subquery.
104 /// It also holds all of the lateral contexts for each respective
105 /// subquery.
106 ///
107 /// See [`Binder::bind_subquery_expr`] for details.
108 upper_subquery_contexts: Vec<(BindContext, Vec<LateralBindContext>)>,
109
110 /// A stack holding contexts of left-lateral `TableFactor`s.
111 ///
112 /// We need a separate stack as `CorrelatedInputRef` depth is
113 /// determined by the upper subquery context depth, not the lateral context stack depth.
114 lateral_contexts: Vec<LateralBindContext>,
115
116 next_subquery_id: usize,
117 next_values_id: usize,
118 /// The `ShareId` is used to identify the share relation which could be a CTE, a source, a view
119 /// and so on.
120 next_share_id: ShareId,
121
122 session_config: Arc<RwLock<SessionConfig>>,
123
124 search_path: SearchPath,
125 /// The type of binding statement.
126 bind_for: BindFor,
127
128 /// `ShareId`s identifying shared views.
129 shared_views: HashMap<ViewId, ShareId>,
130
131 /// The included relations while binding a query.
132 included_relations: HashSet<ObjectId>,
133
134 /// The included user-defined functions while binding a query.
135 included_udfs: HashSet<FunctionId>,
136
137 /// The included secrets while binding a query (e.g., secret refs in UDF arguments).
138 included_secrets: HashSet<SecretId>,
139
140 param_types: ParameterTypes,
141
142 /// The temporary sources that will be used during binding phase
143 temporary_source_manager: TemporarySourceManager,
144
145 /// The staging catalogs that will be used during binding phase
146 staging_catalog_manager: StagingCatalogManager,
147
148 /// Information for `secure_compare` function. It's ONLY available when binding the
149 /// `VALIDATE` clause of Webhook source i.e. `VALIDATE SECRET ... AS SECURE_COMPARE(...)`.
150 secure_compare_context: Option<SecureCompareContext>,
151}
152
153pub const WEBHOOK_PAYLOAD_FIELD_NAME: &str = "payload";
154
155// There are hidden names reserved for webhook validation expressions:
156// - `headers`, whose type is `JSONB`
157// - `payload`, whose type is `BYTEA`
158#[derive(Default, Clone, Debug)]
159pub struct SecureCompareContext {
160 /// The identifier used to reference the raw webhook payload during validation.
161 pub payload_name: String,
162 /// The secret (usually a token provided by the webhook source user) to validate the calls
163 pub secret_name: Option<String>,
164}
165
166/// `ParameterTypes` is used to record the types of the parameters during binding prepared stataments.
167/// It works by following the rules:
168/// 1. At the beginning, it contains the user specified parameters type.
169/// 2. When the binder encounters a parameter, it will record it as unknown(call `record_new_param`)
170/// if it didn't exist in `ParameterTypes`.
171/// 3. When the binder encounters a cast on parameter, if it's a unknown type, the cast function
172/// will record the target type as infer type for that parameter(call `record_infer_type`). If the
173/// parameter has been inferred, the cast function will act as a normal cast.
174/// 4. After bind finished:
175/// (a) parameter not in `ParameterTypes` means that the user didn't specify it and it didn't
176/// occur in the query. `export` will return error if there is a kind of
177/// parameter. This rule is compatible with PostgreSQL
178/// (b) parameter is None means that it's a unknown type. The user didn't specify it
179/// and we can't infer it in the query. We will treat it as VARCHAR type finally. This rule is
180/// compatible with PostgreSQL.
181/// (c) parameter is Some means that it's a known type.
182#[derive(Clone, Debug)]
183pub struct ParameterTypes(Arc<RwLock<HashMap<u64, Option<DataType>>>>);
184
185impl ParameterTypes {
186 pub fn new(specified_param_types: Vec<Option<DataType>>) -> Self {
187 let map = specified_param_types
188 .into_iter()
189 .enumerate()
190 .map(|(index, data_type)| ((index + 1) as u64, data_type))
191 .collect::<HashMap<u64, Option<DataType>>>();
192 Self(Arc::new(RwLock::new(map)))
193 }
194
195 pub fn has_infer(&self, index: u64) -> bool {
196 self.0.read().get(&index).unwrap().is_some()
197 }
198
199 pub fn read_type(&self, index: u64) -> Option<DataType> {
200 self.0.read().get(&index).unwrap().clone()
201 }
202
203 pub fn record_new_param(&mut self, index: u64) {
204 self.0.write().entry(index).or_insert(None);
205 }
206
207 pub fn record_infer_type(&mut self, index: u64, data_type: &DataType) {
208 assert!(
209 !self.has_infer(index),
210 "The parameter has been inferred, should not be inferred again."
211 );
212 self.0
213 .write()
214 .get_mut(&index)
215 .unwrap()
216 .replace(data_type.clone());
217 }
218
219 pub fn export(&self) -> Result<Vec<DataType>> {
220 let types = self
221 .0
222 .read()
223 .clone()
224 .into_iter()
225 .sorted_by_key(|(index, _)| *index)
226 .collect::<Vec<_>>();
227
228 // Check if all the parameters have been inferred.
229 for ((index, _), expect_index) in types.iter().zip_eq_debug(1_u64..=types.len() as u64) {
230 if *index != expect_index {
231 return Err(ErrorCode::InvalidInputSyntax(format!(
232 "Cannot infer the type of the parameter {}.",
233 expect_index
234 ))
235 .into());
236 }
237 }
238
239 Ok(types
240 .into_iter()
241 .map(|(_, data_type)| data_type.unwrap_or(DataType::Varchar))
242 .collect::<Vec<_>>())
243 }
244}
245
246impl Binder {
247 fn new(session: &SessionImpl, bind_for: BindFor) -> Binder {
248 Binder {
249 catalog: session.env().catalog_reader().read_guard(),
250 user: session.env().user_info_reader().read_guard(),
251 db_name: session.database(),
252 database_id: session.database_id(),
253 session_id: session.id(),
254 context: BindContext::new(),
255 auth_context: session.auth_context(),
256 upper_subquery_contexts: vec![],
257 lateral_contexts: vec![],
258 next_subquery_id: 0,
259 next_values_id: 0,
260 next_share_id: 0,
261 session_config: session.shared_config(),
262 search_path: session.config().search_path(),
263 bind_for,
264 shared_views: HashMap::new(),
265 included_relations: HashSet::new(),
266 included_udfs: HashSet::new(),
267 included_secrets: HashSet::new(),
268 param_types: ParameterTypes::new(vec![]),
269 temporary_source_manager: session.temporary_source_manager(),
270 staging_catalog_manager: session.staging_catalog_manager(),
271 secure_compare_context: None,
272 }
273 }
274
275 pub fn new_for_batch(session: &SessionImpl) -> Binder {
276 Self::new(session, BindFor::Batch)
277 }
278
279 pub fn new_for_stream(session: &SessionImpl) -> Binder {
280 Self::new(session, BindFor::Stream)
281 }
282
283 pub fn new_for_ddl(session: &SessionImpl) -> Binder {
284 Self::new(session, BindFor::Ddl)
285 }
286
287 pub fn new_for_system(session: &SessionImpl) -> Binder {
288 Self::new(session, BindFor::System)
289 }
290
291 /// Set the specified parameter types.
292 pub fn with_specified_params_types(mut self, param_types: Vec<Option<DataType>>) -> Self {
293 self.param_types = ParameterTypes::new(param_types);
294 self
295 }
296
297 /// Set the secure compare context.
298 pub fn with_secure_compare(mut self, ctx: SecureCompareContext) -> Self {
299 self.secure_compare_context = Some(ctx);
300 self
301 }
302
303 fn is_for_stream(&self) -> bool {
304 matches!(self.bind_for, BindFor::Stream)
305 }
306
307 #[expect(dead_code)]
308 fn is_for_batch(&self) -> bool {
309 matches!(self.bind_for, BindFor::Batch)
310 }
311
312 fn is_for_ddl(&self) -> bool {
313 matches!(self.bind_for, BindFor::Ddl)
314 }
315
316 /// Bind a [`Statement`].
317 pub fn bind(&mut self, stmt: Statement) -> Result<BoundStatement> {
318 self.bind_statement(stmt)
319 }
320
321 pub fn export_param_types(&self) -> Result<Vec<DataType>> {
322 self.param_types.export()
323 }
324
325 /// Get included relations in the query after binding. This is used for resolving relation
326 /// dependencies. Note that it only contains referenced relations discovered during binding.
327 /// After the plan is built, the referenced relations may be changed. We cannot rely on the
328 /// collection result of plan, because we still need to record the dependencies that have been
329 /// optimised away.
330 pub fn included_relations(&self) -> &HashSet<ObjectId> {
331 &self.included_relations
332 }
333
334 /// Get included user-defined functions in the query after binding.
335 pub fn included_udfs(&self) -> &HashSet<FunctionId> {
336 &self.included_udfs
337 }
338
339 /// Get included secrets in the query after binding (e.g., secret refs in UDF arguments).
340 pub fn included_secrets(&self) -> &HashSet<SecretId> {
341 &self.included_secrets
342 }
343
344 fn push_context(&mut self) {
345 let new_context = std::mem::take(&mut self.context);
346 self.context
347 .cte_to_relation
348 .clone_from(&new_context.cte_to_relation);
349 self.context.disable_security_invoker = new_context.disable_security_invoker;
350 let new_lateral_contexts = std::mem::take(&mut self.lateral_contexts);
351 self.upper_subquery_contexts
352 .push((new_context, new_lateral_contexts));
353 }
354
355 fn pop_context(&mut self) -> Result<()> {
356 let (old_context, old_lateral_contexts) = self
357 .upper_subquery_contexts
358 .pop()
359 .ok_or_else(|| ErrorCode::InternalError("Popping non-existent context".to_owned()))?;
360 self.context = old_context;
361 self.lateral_contexts = old_lateral_contexts;
362 Ok(())
363 }
364
365 fn push_lateral_context(&mut self) {
366 let new_context = std::mem::take(&mut self.context);
367 self.context
368 .cte_to_relation
369 .clone_from(&new_context.cte_to_relation);
370 self.context.disable_security_invoker = new_context.disable_security_invoker;
371 self.lateral_contexts.push(LateralBindContext {
372 is_visible: false,
373 context: new_context,
374 });
375 }
376
377 fn pop_and_merge_lateral_context(&mut self) -> Result<()> {
378 let mut old_context = self
379 .lateral_contexts
380 .pop()
381 .ok_or_else(|| ErrorCode::InternalError("Popping non-existent context".to_owned()))?
382 .context;
383 old_context.merge_context(self.context.clone())?;
384 self.context = old_context;
385 Ok(())
386 }
387
388 /// Make every enclosing left-hand `FROM` context visible while binding a lateral table
389 /// factor. A lateral factor nested in a join tree may refer not only to its immediate left
390 /// sibling, but also to left inputs of enclosing joins.
391 fn mark_lateral_contexts_visible(&mut self) -> Vec<bool> {
392 self.lateral_contexts
393 .iter_mut()
394 .map(|ctx| std::mem::replace(&mut ctx.is_visible, true))
395 .collect()
396 }
397
398 fn restore_lateral_contexts_visibility(&mut self, visibility: Vec<bool>) {
399 // Some table-factor binders return early on an error without unwinding their temporary
400 // query context. The whole binder is discarded in that case, so there is no visibility
401 // state to restore on the active stack.
402 if self.lateral_contexts.len() != visibility.len() {
403 return;
404 }
405 for (ctx, is_visible) in self.lateral_contexts.iter_mut().zip_eq_debug(visibility) {
406 ctx.is_visible = is_visible;
407 }
408 }
409
410 /// Returns a reverse iterator over the upper subquery contexts that are visible to the current
411 /// context. Not to be confused with `is_visible` in [`LateralBindContext`].
412 ///
413 /// In most cases, this should include all the upper subquery contexts. However, when binding
414 /// SQL UDFs, we should avoid resolving the context outside the UDF for hygiene.
415 fn visible_upper_subquery_contexts_rev(
416 &self,
417 ) -> impl Iterator<Item = &(BindContext, Vec<LateralBindContext>)> + '_ {
418 self.upper_subquery_contexts
419 .iter()
420 .rev()
421 .take_while(|(context, _)| context.sql_udf_arguments.is_none())
422 }
423
424 fn next_subquery_id(&mut self) -> usize {
425 let id = self.next_subquery_id;
426 self.next_subquery_id += 1;
427 id
428 }
429
430 fn next_values_id(&mut self) -> usize {
431 let id = self.next_values_id;
432 self.next_values_id += 1;
433 id
434 }
435
436 fn next_share_id(&mut self) -> ShareId {
437 let id = self.next_share_id;
438 self.next_share_id += 1;
439 id
440 }
441
442 fn first_valid_schema(&self) -> CatalogResult<&SchemaCatalog> {
443 self.catalog.first_valid_schema(
444 &self.db_name,
445 &self.search_path,
446 &self.auth_context.user_name,
447 )
448 }
449
450 fn bind_schema_path<'a>(&'a self, schema_name: Option<&'a str>) -> SchemaPath<'a> {
451 SchemaPath::new(schema_name, &self.search_path, &self.auth_context.user_name)
452 }
453
454 pub fn set_clause(&mut self, clause: Option<Clause>) {
455 self.context.clause = clause;
456 }
457}
458
459/// The column name stored in [`BindContext`] for a column without an alias.
460pub const UNNAMED_COLUMN: &str = "?column?";
461/// The table name stored in [`BindContext`] for a subquery without an alias.
462const UNNAMED_SUBQUERY: &str = "?subquery?";
463/// The table name stored in [`BindContext`] for a column group.
464const COLUMN_GROUP_PREFIX: &str = "?column_group_id?";
465
466#[cfg(test)]
467pub mod test_utils {
468 use risingwave_common::types::DataType;
469
470 use super::Binder;
471 use crate::session::SessionImpl;
472
473 pub fn mock_binder() -> Binder {
474 mock_binder_with_param_types(vec![])
475 }
476
477 pub fn mock_binder_with_param_types(param_types: Vec<Option<DataType>>) -> Binder {
478 Binder::new_for_batch(&SessionImpl::mock()).with_specified_params_types(param_types)
479 }
480}
481
482#[cfg(test)]
483mod tests {
484 use expect_test::expect;
485
486 use super::test_utils::*;
487
488 #[tokio::test]
489 async fn test_bind_approx_percentile() {
490 let stmt = risingwave_sqlparser::parser::Parser::parse_sql(
491 "SELECT approx_percentile(0.5, 0.01) WITHIN GROUP (ORDER BY generate_series) FROM generate_series(1, 100)",
492 ).unwrap().into_iter().next().unwrap();
493 let parse_expected = expect![[r#"
494 Query(
495 Query {
496 with: None,
497 body: Select(
498 Select {
499 distinct: All,
500 projection: [
501 UnnamedExpr(
502 Function(
503 Function {
504 scalar_as_agg: false,
505 name: ObjectName(
506 [
507 Ident {
508 value: "approx_percentile",
509 quote_style: None,
510 },
511 ],
512 ),
513 arg_list: FunctionArgList {
514 distinct: false,
515 args: [
516 Unnamed(
517 Expr(
518 Value(
519 Number(
520 "0.5",
521 ),
522 ),
523 ),
524 ),
525 Unnamed(
526 Expr(
527 Value(
528 Number(
529 "0.01",
530 ),
531 ),
532 ),
533 ),
534 ],
535 variadic: false,
536 order_by: [],
537 ignore_nulls: false,
538 },
539 within_group: Some(
540 OrderByExpr {
541 expr: Identifier(
542 Ident {
543 value: "generate_series",
544 quote_style: None,
545 },
546 ),
547 asc: None,
548 nulls_first: None,
549 },
550 ),
551 filter: None,
552 over: None,
553 },
554 ),
555 ),
556 ],
557 from: [
558 TableWithJoins {
559 relation: TableFunction {
560 name: ObjectName(
561 [
562 Ident {
563 value: "generate_series",
564 quote_style: None,
565 },
566 ],
567 ),
568 alias: None,
569 args: [
570 Unnamed(
571 Expr(
572 Value(
573 Number(
574 "1",
575 ),
576 ),
577 ),
578 ),
579 Unnamed(
580 Expr(
581 Value(
582 Number(
583 "100",
584 ),
585 ),
586 ),
587 ),
588 ],
589 with_ordinality: false,
590 },
591 joins: [],
592 },
593 ],
594 lateral_views: [],
595 selection: None,
596 group_by: [],
597 having: None,
598 window: [],
599 },
600 ),
601 order_by: [],
602 limit: None,
603 offset: None,
604 fetch: None,
605 },
606 )"#]];
607 parse_expected.assert_eq(&format!("{:#?}", stmt));
608
609 let mut binder = mock_binder();
610 let bound = binder.bind(stmt).unwrap();
611
612 let expected = expect![[r#"
613 Query(
614 BoundQuery {
615 body: Select(
616 BoundSelect {
617 distinct: All,
618 select_items: [
619 AggCall(
620 AggCall {
621 agg_type: Builtin(
622 ApproxPercentile,
623 ),
624 return_type: Float64,
625 args: [
626 FunctionCall(
627 FunctionCall {
628 func_type: Cast,
629 return_type: Float64,
630 inputs: [
631 InputRef(
632 InputRef {
633 index: 0,
634 data_type: Int32,
635 },
636 ),
637 ],
638 },
639 ),
640 ],
641 filter: Condition {
642 conjunctions: [],
643 },
644 distinct: false,
645 order_by: OrderBy {
646 sort_exprs: [
647 OrderByExpr {
648 expr: InputRef(
649 InputRef {
650 index: 0,
651 data_type: Int32,
652 },
653 ),
654 order_type: OrderType {
655 direction: Ascending,
656 nulls_are: Largest,
657 },
658 },
659 ],
660 },
661 direct_args: [
662 Literal {
663 data: Some(
664 Float64(
665 0.5,
666 ),
667 ),
668 data_type: Some(
669 Float64,
670 ),
671 },
672 Literal {
673 data: Some(
674 Float64(
675 0.01,
676 ),
677 ),
678 data_type: Some(
679 Float64,
680 ),
681 },
682 ],
683 },
684 ),
685 ],
686 aliases: [
687 Some(
688 "approx_percentile",
689 ),
690 ],
691 from: Some(
692 TableFunction {
693 expr: TableFunction(
694 FunctionCall {
695 function_type: GenerateSeries,
696 return_type: Int32,
697 args: [
698 Literal(
699 Literal {
700 data: Some(
701 Int32(
702 1,
703 ),
704 ),
705 data_type: Some(
706 Int32,
707 ),
708 },
709 ),
710 Literal(
711 Literal {
712 data: Some(
713 Int32(
714 100,
715 ),
716 ),
717 data_type: Some(
718 Int32,
719 ),
720 },
721 ),
722 ],
723 },
724 ),
725 with_ordinality: false,
726 },
727 ),
728 where_clause: None,
729 group_by: GroupKey(
730 [],
731 ),
732 having: None,
733 window: {},
734 schema: Schema {
735 fields: [
736 approx_percentile:Float64,
737 ],
738 },
739 },
740 ),
741 order: [],
742 limit: None,
743 offset: None,
744 with_ties: false,
745 extra_order_exprs: [],
746 },
747 )"#]];
748
749 expected.assert_eq(&format!("{:#?}", bound));
750 }
751}