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