1use std::pin::Pin;
16use std::sync::Arc;
17use std::task::{Context, Poll};
18
19use futures::stream::{self, BoxStream};
20use futures::{Stream, StreamExt};
21use itertools::Itertools;
22use pgwire::pg_field_descriptor::PgFieldDescriptor;
23use pgwire::pg_response::StatementType::{self, ABORT, BEGIN, COMMIT, ROLLBACK, START_TRANSACTION};
24use pgwire::pg_response::{PgResponse, PgResponseBuilder, RowSetResult};
25use pgwire::pg_server::BoxedError;
26use pgwire::types::{Format, Row};
27use risingwave_common::catalog::{AlterDatabaseParam, ICEBERG_SINK_PREFIX};
28use risingwave_common::types::Fields;
29use risingwave_common::util::iter_util::ZipEqFast;
30use risingwave_common::{bail, bail_not_implemented};
31use risingwave_pb::meta::PbThrottleTarget;
32use risingwave_sqlparser::ast::*;
33use thiserror_ext::AsReport;
34use util::get_table_catalog_by_table_name;
35
36use self::util::{DataChunkToRowSetAdapter, SourceSchemaCompatExt};
37use crate::catalog::table_catalog::TableType;
38use crate::error::{ErrorCode, Result};
39use crate::handler::cancel_job::handle_cancel;
40use crate::handler::kill_process::handle_kill;
41use crate::scheduler::{DistributedQueryStream, LocalQueryStream};
42use crate::session::SessionImpl;
43use crate::utils::WithOptions;
44
45mod alter_compaction_group;
46mod alter_connection_props;
47mod alter_database_param;
48mod alter_mv;
49mod alter_owner;
50mod alter_parallelism;
51mod alter_rename;
52mod alter_resource_group;
53mod alter_secret;
54mod alter_set_schema;
55mod alter_sink_props;
56mod alter_source_column;
57mod alter_source_props;
58mod alter_source_with_sr;
59mod alter_streaming_config;
60mod alter_streaming_enable_unaligned_join;
61mod alter_streaming_rate_limit;
62mod alter_subscription_retention;
63mod alter_swap_rename;
64mod alter_system;
65mod alter_table_column;
66pub mod alter_table_drop_connector;
67pub mod alter_table_props;
68mod alter_table_with_sr;
69pub mod alter_user;
70mod alter_utils;
71mod alter_watermark;
72mod backup;
73pub mod cancel_job;
74pub mod close_cursor;
75mod comment;
76pub mod create_aggregate;
77pub mod create_connection;
78mod create_database;
79pub mod create_function;
80pub mod create_index;
81pub mod create_mv;
82pub mod create_schema;
83pub mod create_secret;
84pub mod create_sink;
85pub mod create_source;
86pub mod create_sql_function;
87pub mod create_subscription;
88pub mod create_table;
89pub mod create_table_as;
90pub mod create_user;
91pub mod create_view;
92pub mod declare_cursor;
93mod delete_meta_snapshot;
94pub mod describe;
95pub mod discard;
96mod drop_connection;
97mod drop_database;
98pub mod drop_function;
99mod drop_index;
100pub mod drop_mv;
101mod drop_schema;
102pub mod drop_secret;
103pub mod drop_sink;
104pub mod drop_source;
105pub mod drop_subscription;
106pub mod drop_table;
107pub mod drop_user;
108mod drop_view;
109pub mod explain;
110pub mod explain_analyze_stream_job;
111pub mod extended_handle;
112pub mod fetch_cursor;
113mod flush;
114pub mod handle_privilege;
115pub mod kill_process;
116mod prepared_statement;
117pub mod privilege;
118pub mod query;
119mod recover;
120mod refresh;
121mod reset_source;
122pub mod show;
123mod transaction;
124mod use_db;
125pub mod util;
126pub mod vacuum;
127pub mod variable;
128mod wait;
129
130fn rate_limit_type_to_throttle_type(
131 rate_limit_type: AlterRateLimitType,
132) -> risingwave_pb::common::PbThrottleType {
133 match rate_limit_type {
134 AlterRateLimitType::Source => risingwave_pb::common::PbThrottleType::Source,
135 AlterRateLimitType::Backfill => risingwave_pb::common::PbThrottleType::Backfill,
136 AlterRateLimitType::Dml => risingwave_pb::common::PbThrottleType::Dml,
137 AlterRateLimitType::Sink => risingwave_pb::common::PbThrottleType::Sink,
138 }
139}
140
141fn ensure_rate_limit_type_supported(
142 rate_limit: &AlterRateLimit,
143 supported_types: &[AlterRateLimitType],
144 target: &str,
145) -> Result<()> {
146 if supported_types.contains(&rate_limit.rate_limit_type) {
147 Ok(())
148 } else {
149 Err(ErrorCode::InvalidInputSyntax(format!(
150 "ALTER {target} SET {} is not supported",
151 rate_limit.rate_limit_type
152 ))
153 .into())
154 }
155}
156
157pub use alter_table_column::{
158 fetch_table_catalog_for_alter, get_new_table_definition_for_cdc_table, get_replace_table_plan,
159};
160
161pub type RwPgResponseBuilder = PgResponseBuilder<PgResponseStream>;
163
164pub type RwPgResponse = PgResponse<PgResponseStream>;
166
167#[easy_ext::ext(RwPgResponseBuilderExt)]
168impl RwPgResponseBuilder {
169 pub fn rows<T: Fields>(self, rows: impl IntoIterator<Item = T>) -> Self {
171 let fields = T::fields();
172 self.values(
173 rows.into_iter()
174 .map(|row| {
175 Row::new(
176 row.into_owned_row()
177 .into_iter()
178 .zip_eq_fast(&fields)
179 .map(|(datum, (_, ty))| {
180 datum.map(|scalar| {
181 scalar.as_scalar_ref_impl().text_format(ty).into()
182 })
183 })
184 .collect(),
185 )
186 })
187 .collect_vec()
188 .into(),
189 fields_to_descriptors(fields),
190 )
191 }
192}
193
194pub fn fields_to_descriptors(
195 fields: Vec<(&str, risingwave_common::types::DataType)>,
196) -> Vec<PgFieldDescriptor> {
197 fields
198 .iter()
199 .map(|(name, ty)| PgFieldDescriptor::new(name.to_string(), ty.to_oid(), ty.type_len()))
200 .collect()
201}
202
203pub enum PgResponseStream {
204 LocalQuery(DataChunkToRowSetAdapter<LocalQueryStream>),
205 DistributedQuery(DataChunkToRowSetAdapter<DistributedQueryStream>),
206 Rows(BoxStream<'static, RowSetResult>),
207}
208
209impl Stream for PgResponseStream {
210 type Item = std::result::Result<Vec<Row>, BoxedError>;
211
212 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
213 match &mut *self {
214 PgResponseStream::LocalQuery(inner) => inner.poll_next_unpin(cx),
215 PgResponseStream::DistributedQuery(inner) => inner.poll_next_unpin(cx),
216 PgResponseStream::Rows(inner) => inner.poll_next_unpin(cx),
217 }
218 }
219}
220
221impl From<Vec<Row>> for PgResponseStream {
222 fn from(rows: Vec<Row>) -> Self {
223 Self::Rows(stream::iter(vec![Ok(rows)]).boxed())
224 }
225}
226
227#[derive(Clone)]
228pub struct HandlerArgs {
229 pub session: Arc<SessionImpl>,
230 pub sql: Arc<str>,
231 pub normalized_sql: String,
232 pub with_options: WithOptions,
233}
234
235impl HandlerArgs {
236 pub fn new(session: Arc<SessionImpl>, stmt: &Statement, sql: Arc<str>) -> Result<Self> {
237 Ok(Self {
238 session,
239 sql,
240 with_options: WithOptions::try_from(stmt)?,
241 normalized_sql: Self::normalize_sql(stmt),
242 })
243 }
244
245 fn normalize_sql(stmt: &Statement) -> String {
251 let mut stmt = stmt.clone();
252 match &mut stmt {
253 Statement::CreateView {
254 or_replace,
255 if_not_exists,
256 ..
257 } => {
258 *or_replace = false;
259 *if_not_exists = false;
260 }
261 Statement::CreateTable {
262 or_replace,
263 if_not_exists,
264 ..
265 } => {
266 *or_replace = false;
267 *if_not_exists = false;
268 }
269 Statement::CreateIndex { if_not_exists, .. } => {
270 *if_not_exists = false;
271 }
272 Statement::CreateSource {
273 stmt: CreateSourceStatement { if_not_exists, .. },
274 ..
275 } => {
276 *if_not_exists = false;
277 }
278 Statement::CreateSink {
279 stmt:
280 CreateSinkStatement {
281 or_replace,
282 if_not_exists,
283 ..
284 },
285 } => {
286 *or_replace = false;
287 *if_not_exists = false;
288 }
289 Statement::CreateSubscription {
290 stmt: CreateSubscriptionStatement { if_not_exists, .. },
291 } => {
292 *if_not_exists = false;
293 }
294 Statement::CreateConnection {
295 stmt: CreateConnectionStatement { if_not_exists, .. },
296 } => {
297 *if_not_exists = false;
298 }
299 _ => {}
300 }
301 stmt.to_string()
302 }
303}
304
305pub async fn handle(
306 session: Arc<SessionImpl>,
307 stmt: Statement,
308 sql: Arc<str>,
309 formats: Vec<Format>,
310) -> Result<RwPgResponse> {
311 session.clear_cancel_query_flag();
312 let _guard = session.txn_begin_implicit();
313 let handler_args = HandlerArgs::new(session, &stmt, sql)?;
314
315 check_ban_ddl_for_iceberg_engine_table(handler_args.session.clone(), &stmt)?;
316
317 match stmt {
318 Statement::Explain {
319 statement,
320 analyze,
321 options,
322 } => {
323 Box::pin(explain::handle_explain(
324 handler_args,
325 *statement,
326 options,
327 analyze,
328 ))
329 .await
330 }
331 Statement::ExplainAnalyzeStreamJob {
332 target,
333 duration_secs,
334 } => {
335 explain_analyze_stream_job::handle_explain_analyze_stream_job(
336 handler_args,
337 target,
338 duration_secs,
339 )
340 .await
341 }
342 Statement::CreateSource { stmt } => {
343 create_source::handle_create_source(handler_args, stmt).await
344 }
345 Statement::CreateSink { stmt } => {
346 create_sink::handle_create_sink(handler_args, stmt, false).await
347 }
348 Statement::CreateSubscription { stmt } => {
349 create_subscription::handle_create_subscription(handler_args, stmt).await
350 }
351 Statement::CreateConnection { stmt } => {
352 create_connection::handle_create_connection(handler_args, stmt).await
353 }
354 Statement::CreateSecret { stmt } => {
355 create_secret::handle_create_secret(handler_args, stmt).await
356 }
357 Statement::CreateFunction {
358 or_replace,
359 temporary,
360 if_not_exists,
361 name,
362 args,
363 returns,
364 params,
365 with_options,
366 } => {
367 if params.language.is_none()
370 || !params
371 .language
372 .as_ref()
373 .unwrap()
374 .real_value()
375 .eq_ignore_ascii_case("sql")
376 {
377 create_function::handle_create_function(
378 handler_args,
379 or_replace,
380 temporary,
381 if_not_exists,
382 name,
383 args,
384 returns,
385 params,
386 with_options,
387 )
388 .await
389 } else {
390 create_sql_function::handle_create_sql_function(
391 handler_args,
392 or_replace,
393 temporary,
394 if_not_exists,
395 name,
396 args,
397 returns,
398 params,
399 )
400 .await
401 }
402 }
403 Statement::CreateAggregate {
404 or_replace,
405 if_not_exists,
406 name,
407 args,
408 returns,
409 params,
410 ..
411 } => {
412 create_aggregate::handle_create_aggregate(
413 handler_args,
414 or_replace,
415 if_not_exists,
416 name,
417 args,
418 returns,
419 params,
420 )
421 .await
422 }
423 Statement::CreateTable {
424 name,
425 columns,
426 wildcard_idx,
427 constraints,
428 query,
429 with_options: _, or_replace,
432 temporary,
433 if_not_exists,
434 format_encode,
435 source_watermarks,
436 append_only,
437 on_conflict,
438 with_version_columns,
439 cdc_table_info,
440 include_column_options,
441 webhook_info,
442 engine,
443 } => {
444 if or_replace {
445 bail_not_implemented!("CREATE OR REPLACE TABLE");
446 }
447 if temporary {
448 bail_not_implemented!("CREATE TEMPORARY TABLE");
449 }
450 if let Some(query) = query {
451 return create_table_as::handle_create_as(
452 handler_args,
453 name,
454 if_not_exists,
455 query,
456 columns,
457 append_only,
458 on_conflict,
459 with_version_columns
460 .iter()
461 .map(|col| col.real_value())
462 .collect(),
463 engine,
464 )
465 .await;
466 }
467 let format_encode = format_encode.map(|s| s.into_v2_with_warning());
468 Box::pin(create_table::handle_create_table(
469 handler_args,
470 name,
471 columns,
472 wildcard_idx,
473 constraints,
474 if_not_exists,
475 format_encode,
476 source_watermarks,
477 append_only,
478 on_conflict,
479 with_version_columns
480 .iter()
481 .map(|col| col.real_value())
482 .collect(),
483 cdc_table_info,
484 include_column_options,
485 webhook_info,
486 engine,
487 ))
488 .await
489 }
490 Statement::CreateDatabase {
491 db_name,
492 if_not_exists,
493 owner,
494 resource_group,
495 barrier_interval_ms,
496 checkpoint_frequency,
497 } => {
498 create_database::handle_create_database(
499 handler_args,
500 db_name,
501 if_not_exists,
502 owner,
503 resource_group,
504 barrier_interval_ms,
505 checkpoint_frequency,
506 )
507 .await
508 }
509 Statement::CreateSchema {
510 schema_name,
511 if_not_exists,
512 owner,
513 } => {
514 create_schema::handle_create_schema(handler_args, schema_name, if_not_exists, owner)
515 .await
516 }
517 Statement::CreateUser(stmt) => create_user::handle_create_user(handler_args, stmt).await,
518 Statement::DeclareCursor { stmt } => {
519 declare_cursor::handle_declare_cursor(handler_args, stmt).await
520 }
521 Statement::FetchCursor { stmt } => {
522 fetch_cursor::handle_fetch_cursor(handler_args, stmt, &formats).await
523 }
524 Statement::CloseCursor { stmt } => {
525 close_cursor::handle_close_cursor(handler_args, stmt).await
526 }
527 Statement::AlterUser(stmt) => alter_user::handle_alter_user(handler_args, stmt).await,
528 Statement::Grant { .. } => {
529 handle_privilege::handle_grant_privilege(handler_args, stmt).await
530 }
531 Statement::Revoke { .. } => {
532 handle_privilege::handle_revoke_privilege(handler_args, stmt).await
533 }
534 Statement::Describe { name, kind } => match kind {
535 DescribeKind::Fragments => {
536 describe::handle_describe_fragments(handler_args, name).await
537 }
538 DescribeKind::Plain => describe::handle_describe(handler_args, name),
539 },
540 Statement::DescribeFragment { fragment_id } => {
541 describe::handle_describe_fragment(handler_args, fragment_id.into()).await
542 }
543 Statement::Discard(..) => discard::handle_discard(handler_args),
544 Statement::ShowObjects {
545 object: show_object,
546 filter,
547 } => show::handle_show_object(handler_args, show_object, filter).await,
548 Statement::ShowCreateObject { create_type, name } => {
549 show::handle_show_create_object(handler_args, create_type, name)
550 }
551 Statement::ShowTransactionIsolationLevel => {
552 transaction::handle_show_isolation_level(handler_args)
553 }
554 Statement::Drop(DropStatement {
555 object_type,
556 object_name,
557 if_exists,
558 drop_mode,
559 }) => {
560 let cascade = if let AstOption::Some(DropMode::Cascade) = drop_mode {
561 match object_type {
562 ObjectType::MaterializedView
563 | ObjectType::View
564 | ObjectType::Sink
565 | ObjectType::Source
566 | ObjectType::Subscription
567 | ObjectType::Index
568 | ObjectType::Table
569 | ObjectType::Schema
570 | ObjectType::Connection
571 | ObjectType::Secret => true,
572 ObjectType::Database | ObjectType::User => {
573 bail_not_implemented!("DROP CASCADE");
574 }
575 }
576 } else {
577 false
578 };
579 match object_type {
580 ObjectType::Table => {
581 drop_table::handle_drop_table(handler_args, object_name, if_exists, cascade)
582 .await
583 }
584 ObjectType::MaterializedView => {
585 drop_mv::handle_drop_mv(handler_args, object_name, if_exists, cascade).await
586 }
587 ObjectType::Index => {
588 drop_index::handle_drop_index(handler_args, object_name, if_exists, cascade)
589 .await
590 }
591 ObjectType::Source => {
592 drop_source::handle_drop_source(handler_args, object_name, if_exists, cascade)
593 .await
594 }
595 ObjectType::Sink => {
596 drop_sink::handle_drop_sink(handler_args, object_name, if_exists, cascade).await
597 }
598 ObjectType::Subscription => {
599 drop_subscription::handle_drop_subscription(
600 handler_args,
601 object_name,
602 if_exists,
603 cascade,
604 )
605 .await
606 }
607 ObjectType::Database => {
608 drop_database::handle_drop_database(handler_args, object_name, if_exists).await
609 }
610 ObjectType::Schema => {
611 drop_schema::handle_drop_schema(handler_args, object_name, if_exists, cascade)
612 .await
613 }
614 ObjectType::User => {
615 drop_user::handle_drop_user(handler_args, object_name, if_exists).await
616 }
617 ObjectType::View => {
618 drop_view::handle_drop_view(handler_args, object_name, if_exists, cascade).await
619 }
620 ObjectType::Connection => {
621 drop_connection::handle_drop_connection(
622 handler_args,
623 object_name,
624 if_exists,
625 cascade,
626 )
627 .await
628 }
629 ObjectType::Secret => {
630 drop_secret::handle_drop_secret(handler_args, object_name, if_exists, cascade)
631 .await
632 }
633 }
634 }
635 Statement::DropFunction {
637 if_exists,
638 func_desc,
639 option,
640 } => {
641 drop_function::handle_drop_function(handler_args, if_exists, func_desc, option, false)
642 .await
643 }
644 Statement::DropAggregate {
645 if_exists,
646 func_desc,
647 option,
648 } => {
649 drop_function::handle_drop_function(handler_args, if_exists, func_desc, option, true)
650 .await
651 }
652 Statement::Query(_)
653 | Statement::Insert { .. }
654 | Statement::Delete { .. }
655 | Statement::Update { .. } => query::handle_query(handler_args, stmt, formats).await,
656 Statement::Copy {
657 entity: CopyEntity::Query(query),
658 target: CopyTarget::Stdout,
659 } => {
660 let response =
661 query::handle_query(handler_args, Statement::Query(query), vec![Format::Text])
662 .await?;
663 Ok(response.into_copy_query_to_stdout())
664 }
665 Statement::CreateView {
666 materialized,
667 if_not_exists,
668 name,
669 columns,
670 query,
671 with_options: _, or_replace, emit_mode,
674 } => {
675 if or_replace {
676 bail_not_implemented!("CREATE OR REPLACE VIEW");
677 }
678 if materialized {
679 create_mv::handle_create_mv(
680 handler_args,
681 if_not_exists,
682 name,
683 *query,
684 columns,
685 emit_mode,
686 )
687 .await
688 } else {
689 create_view::handle_create_view(handler_args, if_not_exists, name, columns, *query)
690 .await
691 }
692 }
693 Statement::Flush => flush::handle_flush(handler_args).await,
694 Statement::Wait(target) => wait::handle_wait(handler_args, target).await,
695 Statement::Backup => backup::handle_backup(handler_args).await,
696 Statement::DeleteMetaSnapshots { snapshot_ids } => {
697 delete_meta_snapshot::handle_delete_meta_snapshots(handler_args, snapshot_ids).await
698 }
699 Statement::Recover => recover::handle_recover(handler_args).await,
700 Statement::SetVariable {
701 local: _,
702 variable,
703 value,
704 } => {
705 if variable.real_value().eq_ignore_ascii_case("database") {
707 let x = variable::set_var_to_param_str(&value);
708 let res = use_db::handle_use_db(
709 handler_args,
710 ObjectName::from(vec![Ident::from_real_value(
711 x.as_deref().unwrap_or("default"),
712 )]),
713 )?;
714 let mut builder = RwPgResponse::builder(StatementType::SET_VARIABLE);
715 for notice in res.notices() {
716 builder = builder.notice(notice);
717 }
718 return Ok(builder.into());
719 }
720 variable::handle_set(handler_args, variable, value)
721 }
722 Statement::SetTimeZone { local: _, value } => {
723 variable::handle_set_time_zone(handler_args, value)
724 }
725 Statement::ShowVariable { variable } => variable::handle_show(handler_args, variable),
726 Statement::CreateIndex {
727 name,
728 table_name,
729 method,
730 columns,
731 include,
732 distributed_by,
733 unique,
734 if_not_exists,
735 with_properties: _,
736 } => {
737 if unique {
738 bail_not_implemented!("create unique index");
739 }
740
741 create_index::handle_create_index(
742 handler_args,
743 if_not_exists,
744 name,
745 table_name,
746 method,
747 columns.clone(),
748 include,
749 distributed_by,
750 )
751 .await
752 }
753 Statement::AlterDatabase { name, operation } => match operation {
754 AlterDatabaseOperation::RenameDatabase { database_name } => {
755 alter_rename::handle_rename_database(handler_args, name, database_name).await
756 }
757 AlterDatabaseOperation::ChangeOwner { new_owner_name } => {
758 alter_owner::handle_alter_owner(
759 handler_args,
760 name,
761 new_owner_name,
762 StatementType::ALTER_DATABASE,
763 None,
764 )
765 .await
766 }
767 AlterDatabaseOperation::SetParam(config_param) => {
768 let ConfigParam { param, value } = config_param;
769
770 let database_param = match param.real_value().to_uppercase().as_str() {
771 "BARRIER_INTERVAL_MS" => {
772 let barrier_interval_ms = match value {
773 SetVariableValue::Default => None,
774 SetVariableValue::Single(SetVariableValueSingle::Literal(
775 Value::Number(num),
776 )) => {
777 let num = num.parse::<u32>().map_err(|e| {
778 ErrorCode::InvalidInputSyntax(format!(
779 "barrier_interval_ms must be a u32 integer: {}",
780 e.as_report()
781 ))
782 })?;
783 Some(num)
784 }
785 _ => {
786 return Err(ErrorCode::InvalidInputSyntax(
787 "barrier_interval_ms must be a u32 integer or DEFAULT"
788 .to_owned(),
789 )
790 .into());
791 }
792 };
793 AlterDatabaseParam::BarrierIntervalMs(barrier_interval_ms)
794 }
795 "CHECKPOINT_FREQUENCY" => {
796 let checkpoint_frequency = match value {
797 SetVariableValue::Default => None,
798 SetVariableValue::Single(SetVariableValueSingle::Literal(
799 Value::Number(num),
800 )) => {
801 let num = num.parse::<u64>().map_err(|e| {
802 ErrorCode::InvalidInputSyntax(format!(
803 "checkpoint_frequency must be a u64 integer: {}",
804 e.as_report()
805 ))
806 })?;
807 Some(num)
808 }
809 _ => {
810 return Err(ErrorCode::InvalidInputSyntax(
811 "checkpoint_frequency must be a u64 integer or DEFAULT"
812 .to_owned(),
813 )
814 .into());
815 }
816 };
817 AlterDatabaseParam::CheckpointFrequency(checkpoint_frequency)
818 }
819 _ => {
820 return Err(ErrorCode::InvalidInputSyntax(format!(
821 "Unsupported database config parameter: {}",
822 param.real_value()
823 ))
824 .into());
825 }
826 };
827
828 alter_database_param::handle_alter_database_param(
829 handler_args,
830 name,
831 database_param,
832 )
833 .await
834 }
835 AlterDatabaseOperation::SetResourceGroup {
836 resource_group,
837 deferred,
838 } => {
839 alter_database_param::handle_alter_database_resource_group(
840 handler_args,
841 name,
842 resource_group,
843 deferred,
844 )
845 .await
846 }
847 },
848 Statement::AlterSchema { name, operation } => match operation {
849 AlterSchemaOperation::RenameSchema { schema_name } => {
850 alter_rename::handle_rename_schema(handler_args, name, schema_name).await
851 }
852 AlterSchemaOperation::ChangeOwner { new_owner_name } => {
853 alter_owner::handle_alter_owner(
854 handler_args,
855 name,
856 new_owner_name,
857 StatementType::ALTER_SCHEMA,
858 None,
859 )
860 .await
861 }
862 AlterSchemaOperation::SwapRenameSchema { target_schema } => {
863 alter_swap_rename::handle_swap_rename(
864 handler_args,
865 name,
866 target_schema,
867 StatementType::ALTER_SCHEMA,
868 )
869 .await
870 }
871 },
872 Statement::AlterTable { name, operation } => match operation {
873 AlterTableOperation::AddColumn { .. }
874 | AlterTableOperation::DropColumn { .. }
875 | AlterTableOperation::AlterColumn { .. } => {
876 Box::pin(alter_table_column::handle_alter_table_column(
877 handler_args,
878 name,
879 operation,
880 ))
881 .await
882 }
883 AlterTableOperation::AlterWatermark {
884 column_name,
885 expr,
886 with_ttl,
887 } => {
888 Box::pin(alter_watermark::handle_alter_watermark(
889 handler_args,
890 name,
891 column_name,
892 expr,
893 with_ttl,
894 ))
895 .await
896 }
897 AlterTableOperation::RenameTable { table_name } => {
898 alter_rename::handle_rename_table(handler_args, TableType::Table, name, table_name)
899 .await
900 }
901 AlterTableOperation::ChangeOwner { new_owner_name } => {
902 alter_owner::handle_alter_owner(
903 handler_args,
904 name,
905 new_owner_name,
906 StatementType::ALTER_TABLE,
907 None,
908 )
909 .await
910 }
911 AlterTableOperation::SetParallelism {
912 parallelism,
913 deferred,
914 } => {
915 alter_parallelism::handle_alter_parallelism(
916 handler_args,
917 name,
918 parallelism,
919 StatementType::ALTER_TABLE,
920 deferred,
921 )
922 .await
923 }
924 AlterTableOperation::SetBackfillParallelism {
925 parallelism,
926 deferred,
927 } => {
928 alter_parallelism::handle_alter_backfill_parallelism(
929 handler_args,
930 name,
931 parallelism,
932 StatementType::ALTER_TABLE,
933 deferred,
934 )
935 .await
936 }
937 AlterTableOperation::SetSchema { new_schema_name } => {
938 alter_set_schema::handle_alter_set_schema(
939 handler_args,
940 name,
941 new_schema_name,
942 StatementType::ALTER_TABLE,
943 None,
944 )
945 .await
946 }
947 AlterTableOperation::RefreshSchema => {
948 Box::pin(alter_table_with_sr::handle_refresh_schema(
949 handler_args,
950 name,
951 ))
952 .await
953 }
954 AlterTableOperation::AlterRateLimit(rate_limit) => {
955 ensure_rate_limit_type_supported(
956 &rate_limit,
957 &[
958 AlterRateLimitType::Source,
959 AlterRateLimitType::Backfill,
960 AlterRateLimitType::Dml,
961 ],
962 "TABLE",
963 )?;
964 alter_streaming_rate_limit::handle_alter_streaming_rate_limit(
965 handler_args,
966 PbThrottleTarget::Table,
967 rate_limit_type_to_throttle_type(rate_limit.rate_limit_type),
968 name,
969 rate_limit.rate_limit,
970 )
971 .await
972 }
973 AlterTableOperation::DropConnector => {
974 Box::pin(
975 alter_table_drop_connector::handle_alter_table_drop_connector(
976 handler_args,
977 name,
978 ),
979 )
980 .await
981 }
982 AlterTableOperation::SetConfig { entries } => {
983 alter_streaming_config::handle_alter_streaming_set_config(
984 handler_args,
985 name,
986 entries,
987 StatementType::ALTER_TABLE,
988 )
989 .await
990 }
991 AlterTableOperation::ResetConfig { keys } => {
992 alter_streaming_config::handle_alter_streaming_reset_config(
993 handler_args,
994 name,
995 keys,
996 StatementType::ALTER_TABLE,
997 )
998 .await
999 }
1000 AlterTableOperation::SwapRenameTable { target_table } => {
1001 alter_swap_rename::handle_swap_rename(
1002 handler_args,
1003 name,
1004 target_table,
1005 StatementType::ALTER_TABLE,
1006 )
1007 .await
1008 }
1009 AlterTableOperation::AlterConnectorProps { alter_props } => {
1010 alter_table_props::handle_alter_table_props(handler_args, name, alter_props).await
1011 }
1012 AlterTableOperation::AddConstraint { .. }
1013 | AlterTableOperation::DropConstraint { .. }
1014 | AlterTableOperation::RenameColumn { .. }
1015 | AlterTableOperation::ChangeColumn { .. }
1016 | AlterTableOperation::RenameConstraint { .. } => {
1017 bail_not_implemented!(
1018 "Unhandled statement: {}",
1019 Statement::AlterTable { name, operation }
1020 )
1021 }
1022 },
1023 Statement::AlterIndex { name, operation } => match operation {
1024 AlterIndexOperation::RenameIndex { index_name } => {
1025 alter_rename::handle_rename_index(handler_args, name, index_name).await
1026 }
1027 AlterIndexOperation::SetParallelism {
1028 parallelism,
1029 deferred,
1030 } => {
1031 alter_parallelism::handle_alter_parallelism(
1032 handler_args,
1033 name,
1034 parallelism,
1035 StatementType::ALTER_INDEX,
1036 deferred,
1037 )
1038 .await
1039 }
1040 AlterIndexOperation::SetBackfillParallelism {
1041 parallelism,
1042 deferred,
1043 } => {
1044 alter_parallelism::handle_alter_backfill_parallelism(
1045 handler_args,
1046 name,
1047 parallelism,
1048 StatementType::ALTER_INDEX,
1049 deferred,
1050 )
1051 .await
1052 }
1053 AlterIndexOperation::SetResourceGroup {
1054 resource_group,
1055 deferred,
1056 } => {
1057 alter_resource_group::handle_alter_resource_group(
1058 handler_args,
1059 name,
1060 resource_group,
1061 StatementType::ALTER_INDEX,
1062 deferred,
1063 )
1064 .await
1065 }
1066 AlterIndexOperation::SetConfig { entries } => {
1067 alter_streaming_config::handle_alter_streaming_set_config(
1068 handler_args,
1069 name,
1070 entries,
1071 StatementType::ALTER_INDEX,
1072 )
1073 .await
1074 }
1075 AlterIndexOperation::ResetConfig { keys } => {
1076 alter_streaming_config::handle_alter_streaming_reset_config(
1077 handler_args,
1078 name,
1079 keys,
1080 StatementType::ALTER_INDEX,
1081 )
1082 .await
1083 }
1084 },
1085 Statement::AlterView {
1086 materialized,
1087 name,
1088 operation,
1089 } => {
1090 let statement_type = if materialized {
1091 StatementType::ALTER_MATERIALIZED_VIEW
1092 } else {
1093 StatementType::ALTER_VIEW
1094 };
1095 match operation {
1096 AlterViewOperation::RenameView { view_name } => {
1097 if materialized {
1098 alter_rename::handle_rename_table(
1099 handler_args,
1100 TableType::MaterializedView,
1101 name,
1102 view_name,
1103 )
1104 .await
1105 } else {
1106 alter_rename::handle_rename_view(handler_args, name, view_name).await
1107 }
1108 }
1109 AlterViewOperation::SetParallelism {
1110 parallelism,
1111 deferred,
1112 } => {
1113 if !materialized {
1114 bail_not_implemented!("ALTER VIEW SET PARALLELISM");
1115 }
1116 alter_parallelism::handle_alter_parallelism(
1117 handler_args,
1118 name,
1119 parallelism,
1120 statement_type,
1121 deferred,
1122 )
1123 .await
1124 }
1125 AlterViewOperation::SetBackfillParallelism {
1126 parallelism,
1127 deferred,
1128 } => {
1129 if !materialized {
1130 bail_not_implemented!("ALTER VIEW SET BACKFILL PARALLELISM");
1131 }
1132 alter_parallelism::handle_alter_backfill_parallelism(
1133 handler_args,
1134 name,
1135 parallelism,
1136 statement_type,
1137 deferred,
1138 )
1139 .await
1140 }
1141 AlterViewOperation::SetResourceGroup {
1142 resource_group,
1143 deferred,
1144 } => {
1145 if !materialized {
1146 bail_not_implemented!("ALTER VIEW SET RESOURCE GROUP");
1147 }
1148 alter_resource_group::handle_alter_resource_group(
1149 handler_args,
1150 name,
1151 resource_group,
1152 statement_type,
1153 deferred,
1154 )
1155 .await
1156 }
1157 AlterViewOperation::ChangeOwner { new_owner_name } => {
1158 alter_owner::handle_alter_owner(
1159 handler_args,
1160 name,
1161 new_owner_name,
1162 statement_type,
1163 None,
1164 )
1165 .await
1166 }
1167 AlterViewOperation::SetSchema { new_schema_name } => {
1168 alter_set_schema::handle_alter_set_schema(
1169 handler_args,
1170 name,
1171 new_schema_name,
1172 statement_type,
1173 None,
1174 )
1175 .await
1176 }
1177 AlterViewOperation::AlterRateLimit(rate_limit) => {
1178 ensure_rate_limit_type_supported(
1179 &rate_limit,
1180 &[AlterRateLimitType::Backfill],
1181 "MATERIALIZED VIEW",
1182 )?;
1183 if !materialized {
1184 bail_not_implemented!("ALTER VIEW SET BACKFILL RATE LIMIT");
1185 }
1186 alter_streaming_rate_limit::handle_alter_streaming_rate_limit(
1187 handler_args,
1188 PbThrottleTarget::Mv,
1189 rate_limit_type_to_throttle_type(rate_limit.rate_limit_type),
1190 name,
1191 rate_limit.rate_limit,
1192 )
1193 .await
1194 }
1195 AlterViewOperation::SwapRenameView { target_view } => {
1196 alter_swap_rename::handle_swap_rename(
1197 handler_args,
1198 name,
1199 target_view,
1200 statement_type,
1201 )
1202 .await
1203 }
1204 AlterViewOperation::SetStreamingEnableUnalignedJoin { enable } => {
1205 if !materialized {
1206 bail!(
1207 "ALTER VIEW SET STREAMING_ENABLE_UNALIGNED_JOIN is not supported. Only supported for materialized views"
1208 );
1209 }
1210 alter_streaming_enable_unaligned_join::handle_alter_streaming_enable_unaligned_join(handler_args, name, enable).await
1211 }
1212 AlterViewOperation::AsQuery { query } => {
1213 if !materialized {
1214 bail_not_implemented!("ALTER VIEW AS QUERY");
1215 }
1216 if !cfg!(debug_assertions) {
1218 bail_not_implemented!("ALTER MATERIALIZED VIEW AS QUERY");
1219 }
1220 alter_mv::handle_alter_mv(handler_args, name, query).await
1221 }
1222 AlterViewOperation::SetConfig { entries } => {
1223 if !materialized {
1224 bail!("SET CONFIG is only supported for materialized views");
1225 }
1226 alter_streaming_config::handle_alter_streaming_set_config(
1227 handler_args,
1228 name,
1229 entries,
1230 statement_type,
1231 )
1232 .await
1233 }
1234 AlterViewOperation::ResetConfig { keys } => {
1235 if !materialized {
1236 bail!("RESET CONFIG is only supported for materialized views");
1237 }
1238 alter_streaming_config::handle_alter_streaming_reset_config(
1239 handler_args,
1240 name,
1241 keys,
1242 statement_type,
1243 )
1244 .await
1245 }
1246 }
1247 }
1248
1249 Statement::AlterSink { name, operation } => match operation {
1250 AlterSinkOperation::AlterConnectorProps {
1251 alter_props: changed_props,
1252 } => alter_sink_props::handle_alter_sink_props(handler_args, name, changed_props).await,
1253 AlterSinkOperation::RenameSink { sink_name } => {
1254 alter_rename::handle_rename_sink(handler_args, name, sink_name).await
1255 }
1256 AlterSinkOperation::ChangeOwner { new_owner_name } => {
1257 alter_owner::handle_alter_owner(
1258 handler_args,
1259 name,
1260 new_owner_name,
1261 StatementType::ALTER_SINK,
1262 None,
1263 )
1264 .await
1265 }
1266 AlterSinkOperation::SetSchema { new_schema_name } => {
1267 alter_set_schema::handle_alter_set_schema(
1268 handler_args,
1269 name,
1270 new_schema_name,
1271 StatementType::ALTER_SINK,
1272 None,
1273 )
1274 .await
1275 }
1276 AlterSinkOperation::SetParallelism {
1277 parallelism,
1278 deferred,
1279 } => {
1280 alter_parallelism::handle_alter_parallelism(
1281 handler_args,
1282 name,
1283 parallelism,
1284 StatementType::ALTER_SINK,
1285 deferred,
1286 )
1287 .await
1288 }
1289 AlterSinkOperation::SetBackfillParallelism {
1290 parallelism,
1291 deferred,
1292 } => {
1293 alter_parallelism::handle_alter_backfill_parallelism(
1294 handler_args,
1295 name,
1296 parallelism,
1297 StatementType::ALTER_SINK,
1298 deferred,
1299 )
1300 .await
1301 }
1302 AlterSinkOperation::SetResourceGroup {
1303 resource_group,
1304 deferred,
1305 } => {
1306 alter_resource_group::handle_alter_resource_group(
1307 handler_args,
1308 name,
1309 resource_group,
1310 StatementType::ALTER_SINK,
1311 deferred,
1312 )
1313 .await
1314 }
1315 AlterSinkOperation::SetConfig { entries } => {
1316 alter_streaming_config::handle_alter_streaming_set_config(
1317 handler_args,
1318 name,
1319 entries,
1320 StatementType::ALTER_SINK,
1321 )
1322 .await
1323 }
1324 AlterSinkOperation::ResetConfig { keys } => {
1325 alter_streaming_config::handle_alter_streaming_reset_config(
1326 handler_args,
1327 name,
1328 keys,
1329 StatementType::ALTER_SINK,
1330 )
1331 .await
1332 }
1333 AlterSinkOperation::SwapRenameSink { target_sink } => {
1334 alter_swap_rename::handle_swap_rename(
1335 handler_args,
1336 name,
1337 target_sink,
1338 StatementType::ALTER_SINK,
1339 )
1340 .await
1341 }
1342 AlterSinkOperation::AlterRateLimit(rate_limit) => {
1343 ensure_rate_limit_type_supported(
1344 &rate_limit,
1345 &[AlterRateLimitType::Sink, AlterRateLimitType::Backfill],
1346 "SINK",
1347 )?;
1348 alter_streaming_rate_limit::handle_alter_streaming_rate_limit(
1349 handler_args,
1350 PbThrottleTarget::Sink,
1351 rate_limit_type_to_throttle_type(rate_limit.rate_limit_type),
1352 name,
1353 rate_limit.rate_limit,
1354 )
1355 .await
1356 }
1357 AlterSinkOperation::SetStreamingEnableUnalignedJoin { enable } => {
1358 alter_streaming_enable_unaligned_join::handle_alter_streaming_enable_unaligned_join(
1359 handler_args,
1360 name,
1361 enable,
1362 )
1363 .await
1364 }
1365 },
1366 Statement::AlterSubscription { name, operation } => match operation {
1367 AlterSubscriptionOperation::RenameSubscription { subscription_name } => {
1368 alter_rename::handle_rename_subscription(handler_args, name, subscription_name)
1369 .await
1370 }
1371 AlterSubscriptionOperation::ChangeOwner { new_owner_name } => {
1372 alter_owner::handle_alter_owner(
1373 handler_args,
1374 name,
1375 new_owner_name,
1376 StatementType::ALTER_SUBSCRIPTION,
1377 None,
1378 )
1379 .await
1380 }
1381 AlterSubscriptionOperation::SetSchema { new_schema_name } => {
1382 alter_set_schema::handle_alter_set_schema(
1383 handler_args,
1384 name,
1385 new_schema_name,
1386 StatementType::ALTER_SUBSCRIPTION,
1387 None,
1388 )
1389 .await
1390 }
1391 AlterSubscriptionOperation::SetRetention { retention } => {
1392 alter_subscription_retention::handle_alter_subscription_retention(
1393 handler_args,
1394 name,
1395 retention,
1396 )
1397 .await
1398 }
1399 AlterSubscriptionOperation::SwapRenameSubscription {
1400 target_subscription,
1401 } => {
1402 alter_swap_rename::handle_swap_rename(
1403 handler_args,
1404 name,
1405 target_subscription,
1406 StatementType::ALTER_SUBSCRIPTION,
1407 )
1408 .await
1409 }
1410 },
1411 Statement::AlterSource { name, operation } => match operation {
1412 AlterSourceOperation::AlterConnectorProps { alter_props } => {
1413 alter_source_props::handle_alter_source_connector_props(
1414 handler_args,
1415 name,
1416 alter_props,
1417 )
1418 .await
1419 }
1420 AlterSourceOperation::RenameSource { source_name } => {
1421 alter_rename::handle_rename_source(handler_args, name, source_name).await
1422 }
1423 AlterSourceOperation::AddColumn { .. } => {
1424 alter_source_column::handle_alter_source_column(handler_args, name, operation).await
1425 }
1426 AlterSourceOperation::ChangeOwner { new_owner_name } => {
1427 alter_owner::handle_alter_owner(
1428 handler_args,
1429 name,
1430 new_owner_name,
1431 StatementType::ALTER_SOURCE,
1432 None,
1433 )
1434 .await
1435 }
1436 AlterSourceOperation::SetSchema { new_schema_name } => {
1437 alter_set_schema::handle_alter_set_schema(
1438 handler_args,
1439 name,
1440 new_schema_name,
1441 StatementType::ALTER_SOURCE,
1442 None,
1443 )
1444 .await
1445 }
1446 AlterSourceOperation::FormatEncode { format_encode } => {
1447 alter_source_with_sr::handle_alter_source_with_sr(handler_args, name, format_encode)
1448 .await
1449 }
1450 AlterSourceOperation::RefreshSchema => {
1451 alter_source_with_sr::handler_refresh_schema(handler_args, name).await
1452 }
1453 AlterSourceOperation::AlterRateLimit(rate_limit) => {
1454 ensure_rate_limit_type_supported(
1455 &rate_limit,
1456 &[AlterRateLimitType::Source],
1457 "SOURCE",
1458 )?;
1459 alter_streaming_rate_limit::handle_alter_streaming_rate_limit(
1460 handler_args,
1461 PbThrottleTarget::Source,
1462 rate_limit_type_to_throttle_type(rate_limit.rate_limit_type),
1463 name,
1464 rate_limit.rate_limit,
1465 )
1466 .await
1467 }
1468 AlterSourceOperation::SwapRenameSource { target_source } => {
1469 alter_swap_rename::handle_swap_rename(
1470 handler_args,
1471 name,
1472 target_source,
1473 StatementType::ALTER_SOURCE,
1474 )
1475 .await
1476 }
1477 AlterSourceOperation::SetParallelism {
1478 parallelism,
1479 deferred,
1480 } => {
1481 alter_parallelism::handle_alter_parallelism(
1482 handler_args,
1483 name,
1484 parallelism,
1485 StatementType::ALTER_SOURCE,
1486 deferred,
1487 )
1488 .await
1489 }
1490 AlterSourceOperation::SetBackfillParallelism {
1491 parallelism,
1492 deferred,
1493 } => {
1494 alter_parallelism::handle_alter_backfill_parallelism(
1495 handler_args,
1496 name,
1497 parallelism,
1498 StatementType::ALTER_SOURCE,
1499 deferred,
1500 )
1501 .await
1502 }
1503 AlterSourceOperation::SetConfig { entries } => {
1504 alter_streaming_config::handle_alter_streaming_set_config(
1505 handler_args,
1506 name,
1507 entries,
1508 StatementType::ALTER_SOURCE,
1509 )
1510 .await
1511 }
1512 AlterSourceOperation::ResetConfig { keys } => {
1513 alter_streaming_config::handle_alter_streaming_reset_config(
1514 handler_args,
1515 name,
1516 keys,
1517 StatementType::ALTER_SOURCE,
1518 )
1519 .await
1520 }
1521 AlterSourceOperation::ResetSource => {
1522 reset_source::handle_reset_source(handler_args, name).await
1523 }
1524 },
1525 Statement::AlterFunction {
1526 name,
1527 args,
1528 operation,
1529 } => match operation {
1530 AlterFunctionOperation::SetSchema { new_schema_name } => {
1531 alter_set_schema::handle_alter_set_schema(
1532 handler_args,
1533 name,
1534 new_schema_name,
1535 StatementType::ALTER_FUNCTION,
1536 args,
1537 )
1538 .await
1539 }
1540 AlterFunctionOperation::ChangeOwner { new_owner_name } => {
1541 alter_owner::handle_alter_owner(
1542 handler_args,
1543 name,
1544 new_owner_name,
1545 StatementType::ALTER_FUNCTION,
1546 args,
1547 )
1548 .await
1549 }
1550 },
1551 Statement::AlterConnection { name, operation } => match operation {
1552 AlterConnectionOperation::SetSchema { new_schema_name } => {
1553 alter_set_schema::handle_alter_set_schema(
1554 handler_args,
1555 name,
1556 new_schema_name,
1557 StatementType::ALTER_CONNECTION,
1558 None,
1559 )
1560 .await
1561 }
1562 AlterConnectionOperation::ChangeOwner { new_owner_name } => {
1563 alter_owner::handle_alter_owner(
1564 handler_args,
1565 name,
1566 new_owner_name,
1567 StatementType::ALTER_CONNECTION,
1568 None,
1569 )
1570 .await
1571 }
1572 AlterConnectionOperation::AlterConnectorProps { alter_props } => {
1573 alter_connection_props::handle_alter_connection_connector_props(
1574 handler_args,
1575 name,
1576 alter_props,
1577 )
1578 .await
1579 }
1580 },
1581 Statement::AlterSystem { param, value } => {
1582 alter_system::handle_alter_system(handler_args, param, value).await
1583 }
1584 Statement::AlterSecret { name, operation } => match operation {
1585 AlterSecretOperation::ChangeCredential {
1586 with_options,
1587 new_credential,
1588 } => {
1589 alter_secret::handle_alter_secret(handler_args, name, with_options, new_credential)
1590 .await
1591 }
1592 AlterSecretOperation::ChangeOwner { new_owner_name } => {
1593 alter_owner::handle_alter_owner(
1594 handler_args,
1595 name,
1596 new_owner_name,
1597 StatementType::ALTER_SECRET,
1598 None,
1599 )
1600 .await
1601 }
1602 },
1603 Statement::AlterFragment {
1604 fragment_ids,
1605 operation,
1606 } => match operation {
1607 AlterFragmentOperation::AlterRateLimit(rate_limit) => {
1608 let [fragment_id] = fragment_ids.as_slice() else {
1609 return Err(ErrorCode::InvalidInputSyntax(
1610 "ALTER FRAGMENT ... SET rate limit supports exactly one fragment id"
1611 .to_owned(),
1612 )
1613 .into());
1614 };
1615 alter_streaming_rate_limit::handle_alter_streaming_rate_limit_by_id(
1616 &handler_args.session,
1617 PbThrottleTarget::Fragment,
1618 rate_limit_type_to_throttle_type(rate_limit.rate_limit_type),
1619 *fragment_id,
1620 rate_limit.rate_limit,
1621 StatementType::SET_VARIABLE,
1622 )
1623 .await
1624 }
1625 AlterFragmentOperation::SetParallelism { parallelism } => {
1626 alter_parallelism::handle_alter_fragment_parallelism(
1627 handler_args,
1628 fragment_ids.into_iter().map_into().collect(),
1629 parallelism,
1630 )
1631 .await
1632 }
1633 },
1634 Statement::AlterDefaultPrivileges { .. } => {
1635 handle_privilege::handle_alter_default_privileges(handler_args, stmt).await
1636 }
1637 Statement::AlterCompactionGroup {
1638 group_ids,
1639 operation,
1640 } => {
1641 alter_compaction_group::handle_alter_compaction_group(
1642 handler_args,
1643 group_ids,
1644 operation,
1645 )
1646 .await
1647 }
1648 Statement::StartTransaction { modes } => {
1649 transaction::handle_begin(handler_args, START_TRANSACTION, modes)
1650 }
1651 Statement::Begin { modes } => transaction::handle_begin(handler_args, BEGIN, modes),
1652 Statement::Commit { chain } => {
1653 transaction::handle_commit(handler_args, COMMIT, chain).await
1654 }
1655 Statement::Abort => transaction::handle_rollback(handler_args, ABORT, false).await,
1656 Statement::Rollback { chain } => {
1657 transaction::handle_rollback(handler_args, ROLLBACK, chain).await
1658 }
1659 Statement::SetTransaction {
1660 modes,
1661 snapshot,
1662 session,
1663 } => transaction::handle_set(handler_args, modes, snapshot, session),
1664 Statement::CancelJobs(jobs) => handle_cancel(handler_args, jobs).await,
1665 Statement::Kill(worker_process_id) => handle_kill(handler_args, worker_process_id).await,
1666 Statement::Comment {
1667 object_type,
1668 object_name,
1669 comment,
1670 } => comment::handle_comment(handler_args, object_type, object_name, comment).await,
1671 Statement::Use { db_name } => use_db::handle_use_db(handler_args, db_name),
1672 Statement::Prepare {
1673 name,
1674 data_types,
1675 statement,
1676 } => prepared_statement::handle_prepare(name, data_types, statement),
1677 Statement::Deallocate { name, prepare } => {
1678 prepared_statement::handle_deallocate(name, prepare)
1679 }
1680 Statement::Vacuum { object_name, full } => {
1681 vacuum::handle_vacuum(handler_args, object_name, full).await
1682 }
1683 Statement::Refresh { table_name } => {
1684 refresh::handle_refresh(handler_args, table_name).await
1685 }
1686 _ => bail_not_implemented!("Unhandled statement: {}", stmt),
1687 }
1688}
1689
1690fn check_ban_ddl_for_iceberg_engine_table(
1691 session: Arc<SessionImpl>,
1692 stmt: &Statement,
1693) -> Result<()> {
1694 if let Statement::AlterTable { name, operation } = stmt {
1695 let (table, schema_name) = get_table_catalog_by_table_name(session.as_ref(), name)?;
1696 if table.is_iceberg_engine_table() {
1697 let has_auto_refresh_schema_sink = if matches!(
1698 operation,
1699 AlterTableOperation::AddColumn { .. } | AlterTableOperation::DropColumn { .. }
1700 ) {
1701 let catalog_reader = session.env().catalog_reader().read_guard();
1702 let db_name = session.database();
1703 let sink_name = format!("{}{}", ICEBERG_SINK_PREFIX, table.name());
1704 let sink = catalog_reader
1705 .get_schema_by_name(&db_name, &schema_name)
1706 .ok()
1707 .and_then(|schema| schema.get_created_sink_by_name(&sink_name));
1708 sink.and_then(|s| s.auto_refresh_schema_from_table)
1709 .is_some()
1710 } else {
1711 false
1712 };
1713
1714 check_ban_alter_table_operation_for_iceberg_engine_table(
1715 operation,
1716 &schema_name,
1717 name,
1718 has_auto_refresh_schema_sink,
1719 )?;
1720 }
1721 }
1722
1723 Ok(())
1724}
1725
1726fn check_ban_alter_table_operation_for_iceberg_engine_table(
1727 operation: &AlterTableOperation,
1728 schema_name: &str,
1729 table_name: &ObjectName,
1730 has_auto_refresh_schema_sink: bool,
1731) -> Result<()> {
1732 match operation {
1733 AlterTableOperation::AddColumn { .. } => {
1734 if !has_auto_refresh_schema_sink {
1735 bail!(
1736 "ALTER TABLE {} is not supported for iceberg table without auto schema change sink: {}.{}",
1737 operation,
1738 schema_name,
1739 table_name
1740 );
1741 }
1742 }
1743 AlterTableOperation::DropColumn { .. } => {
1744 if !has_auto_refresh_schema_sink {
1745 bail!(
1746 "ALTER TABLE {} is not supported for iceberg table without auto schema change sink: {}.{}",
1747 operation,
1748 schema_name,
1749 table_name
1750 );
1751 }
1752 }
1753 AlterTableOperation::RenameColumn { .. }
1754 | AlterTableOperation::ChangeColumn { .. }
1755 | AlterTableOperation::AlterColumn { .. } => {
1756 bail!(
1757 "ALTER TABLE {} is not supported for iceberg table: {}.{}. Existing column schema mutation is not supported currently",
1758 operation,
1759 schema_name,
1760 table_name
1761 );
1762 }
1763 AlterTableOperation::RenameTable { .. } => {
1764 bail!(
1765 "ALTER TABLE RENAME is not supported for iceberg table: {}.{}",
1766 schema_name,
1767 table_name
1768 );
1769 }
1770 AlterTableOperation::SetParallelism { .. } => {
1771 bail!(
1772 "ALTER TABLE SET PARALLELISM is not supported for iceberg table: {}.{}",
1773 schema_name,
1774 table_name
1775 );
1776 }
1777 AlterTableOperation::SetBackfillParallelism { .. } => {
1778 bail!(
1779 "ALTER TABLE SET BACKFILL PARALLELISM is not supported for iceberg table: {}.{}",
1780 schema_name,
1781 table_name
1782 );
1783 }
1784 AlterTableOperation::SetSchema { .. } => {
1785 bail!(
1786 "ALTER TABLE SET SCHEMA is not supported for iceberg table: {}.{}",
1787 schema_name,
1788 table_name
1789 );
1790 }
1791 AlterTableOperation::RefreshSchema => {
1792 bail!(
1793 "ALTER TABLE REFRESH SCHEMA is not supported for iceberg table: {}.{}",
1794 schema_name,
1795 table_name
1796 );
1797 }
1798 AlterTableOperation::AlterRateLimit(rate_limit)
1799 if rate_limit.rate_limit_type == AlterRateLimitType::Source =>
1800 {
1801 bail!(
1802 "ALTER TABLE SET SOURCE RATE LIMIT is not supported for iceberg table: {}.{}",
1803 schema_name,
1804 table_name
1805 );
1806 }
1807 AlterTableOperation::AlterWatermark { .. } => {
1808 bail!(
1809 "ALTER TABLE ALTER WATERMARK is not supported for iceberg table: {}.{}",
1810 schema_name,
1811 table_name
1812 );
1813 }
1814 _ => {}
1815 }
1816 Ok(())
1817}