risingwave_frontend/handler/
create_view.rs1use either::Either;
18use pgwire::pg_response::{PgResponse, StatementType};
19use risingwave_common::util::iter_util::ZipEqFast;
20use risingwave_pb::catalog::PbView;
21use risingwave_sqlparser::ast::{Ident, ObjectName, Query};
22
23use super::RwPgResponse;
24use crate::binder::{Binder, BoundStatement};
25use crate::error::Result;
26use crate::handler::HandlerArgs;
27use crate::handler::util::reject_internal_table_dependencies;
28use crate::optimizer::{OptimizerContext, RelationCollectorVisitor};
29use crate::planner::Planner;
30
31pub async fn handle_create_view(
32 handler_args: HandlerArgs,
33 if_not_exists: bool,
34 name: ObjectName,
35 columns: Vec<Ident>,
36 query: Query,
37) -> Result<RwPgResponse> {
38 let session = handler_args.session.clone();
39 let db_name = &session.database();
40 let (schema_name, view_name) = Binder::resolve_schema_qualified_name(db_name, &name)?;
41
42 let (database_id, schema_id) = session.get_database_and_schema_id_for_create(schema_name)?;
43
44 let properties = handler_args.with_options.clone();
45
46 if let Either::Right(resp) = session.check_relation_name_duplicated(
47 name.clone(),
48 StatementType::CREATE_VIEW,
49 if_not_exists,
50 )? {
51 return Ok(resp);
52 }
53
54 let (dependent_relations, dependent_secrets, schema) = {
58 let mut binder = Binder::new_for_batch(&session);
59 let bound_query = binder.bind_query(&query)?;
60 let dependent_relations = binder.included_relations().clone();
61 let dependent_secrets = binder.included_secrets().clone();
62
63 let context = OptimizerContext::from_handler_args(handler_args);
64 let logical = Planner::new_for_batch_dql(context.into())
65 .plan(BoundStatement::Query(bound_query.into()))?;
66 let schema = logical.schema();
67 let batch_plan = logical.gen_batch_plan()?;
68 let dependent_relations =
69 RelationCollectorVisitor::collect_with(dependent_relations, batch_plan.plan);
70
71 reject_internal_table_dependencies(&session, &dependent_relations, "CREATE VIEW")?;
72
73 (dependent_relations, dependent_secrets, schema)
74 };
75
76 let columns = if columns.is_empty() {
77 schema.fields().to_vec()
78 } else {
79 if columns.len() != schema.fields().len() {
80 return Err(crate::error::ErrorCode::InternalError(
81 "view has different number of columns than the query's columns".to_owned(),
82 )
83 .into());
84 }
85 schema
86 .fields()
87 .iter()
88 .zip_eq_fast(columns)
89 .map(|(f, c)| {
90 let mut field = f.clone();
91 field.name = c.real_value();
92 field
93 })
94 .collect()
95 };
96
97 let (properties, secret_refs, connection_refs) = properties.into_parts();
98 if !secret_refs.is_empty() || !connection_refs.is_empty() {
99 return Err(crate::error::ErrorCode::InvalidParameterValue(
100 "Secret reference and Connection reference are not allowed in create view options"
101 .to_owned(),
102 )
103 .into());
104 }
105
106 let view = PbView {
107 id: 0.into(),
108 schema_id,
109 database_id,
110 name: view_name,
111 properties,
112 owner: session.user_id(),
113 sql: format!("{}", query),
114 columns: columns.into_iter().map(|f| f.to_prost()).collect(),
115 created_at_epoch: None,
116 created_at_cluster_version: None,
117 };
118
119 let catalog_writer = session.catalog_writer()?;
120 catalog_writer
121 .create_view(
122 view,
123 dependent_relations
124 .into_iter()
125 .chain(
126 dependent_secrets
127 .iter()
128 .copied()
129 .map(|id| id.as_object_id()),
130 )
131 .collect(),
132 )
133 .await?;
134
135 Ok(PgResponse::empty_result(StatementType::CREATE_VIEW))
136}