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