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