1use std::collections::{BTreeMap, HashMap, HashSet};
19use std::path::Path;
20use std::str::FromStr;
21use std::sync::Arc;
22
23use crate::parser::{
24 CopyToSource, CopyToStatement, CreateExternalTable, DFParser, ExplainStatement,
25 LexOrdering, ResetStatement, Statement as DFStatement,
26};
27use crate::planner::{
28 ContextProvider, PlannerContext, SqlToRel, object_name_to_qualifier,
29};
30use crate::utils::normalize_ident;
31
32use arrow::datatypes::{Field, FieldRef, Fields};
33use datafusion_common::error::_plan_err;
34use datafusion_common::format::ExplainStatementOptions;
35use datafusion_common::parsers::CompressionTypeVariant;
36use datafusion_common::tree_node::{Transformed, TreeNode, TreeNodeRecursion};
37use datafusion_common::{
38 Column, Constraint, Constraints, DFSchema, DFSchemaRef, DataFusionError, Result,
39 ScalarValue, SchemaError, SchemaReference, TableReference, ToDFSchema, exec_err,
40 internal_err, not_impl_err, plan_datafusion_err, plan_err, schema_err,
41 unqualified_field_not_found,
42};
43use datafusion_expr::dml::{
44 CopyTo, InsertOp, MergeIntoAction, MergeIntoClause, MergeIntoClauseKind, MergeIntoOp,
45};
46use datafusion_expr::expr_rewriter::normalize_col_with_schemas_and_ambiguity_check;
47use datafusion_expr::logical_plan::DdlStatement;
48use datafusion_expr::logical_plan::builder::project;
49use datafusion_expr::utils::expr_to_columns;
50use datafusion_expr::{
51 Analyze, CreateCatalog, CreateCatalogSchema,
52 CreateExternalTable as PlanCreateExternalTable, CreateFunction, CreateFunctionBody,
53 CreateIndex as PlanCreateIndex, CreateMemoryTable, CreateView, Deallocate,
54 DescribeTable, DmlStatement, DropCatalogSchema, DropFunction, DropTable, DropView,
55 EmptyRelation, Execute, Explain, ExplainFormat, Expr, ExprSchemable, Filter,
56 LogicalPlan, LogicalPlanBuilder, OperateFunctionArg, PlanType, Prepare,
57 ResetVariable, SetVariable, SortExpr, Statement as PlanStatement, ToStringifiedPlan,
58 TransactionAccessMode, TransactionConclusion, TransactionEnd,
59 TransactionIsolationLevel, TransactionStart, Volatility, WriteOp, cast,
60};
61use sqlparser::ast::{
62 self, BeginTransactionKind, CheckConstraint, ForeignKeyConstraint, IndexColumn,
63 IndexType, NullsDistinctOption, OrderByExpr, OrderByOptions, PrimaryKeyConstraint,
64 Set, ShowStatementIn, ShowStatementOptions, SqliteOnConflict, TableObject,
65 UniqueConstraint, Update, UpdateTableFromKind, ValueWithSpan,
66};
67use sqlparser::ast::{
68 Assignment, AssignmentTarget, ColumnDef, CreateIndex, CreateTable,
69 CreateTableOptions, Delete, DescribeAlias, Expr as SQLExpr, FromTable, Ident, Insert,
70 ObjectName, ObjectType, Query, SchemaName, SetExpr, ShowCreateObject,
71 ShowStatementFilter, Statement, TableConstraint, TableFactor, TableWithJoins,
72 TransactionMode, UnaryOperator, Value,
73};
74use sqlparser::parser::ParserError::ParserError;
75
76fn ident_to_string(ident: &Ident) -> String {
77 normalize_ident(ident.to_owned())
78}
79
80fn object_name_to_string(object_name: &ObjectName) -> String {
81 object_name
82 .0
83 .iter()
84 .map(|object_name_part| {
85 object_name_part
86 .as_ident()
87 .map_or_else(String::new, ident_to_string)
90 })
91 .collect::<Vec<String>>()
92 .join(".")
93}
94
95fn get_schema_name(schema_name: &SchemaName) -> String {
96 match schema_name {
97 SchemaName::Simple(schema_name) => object_name_to_string(schema_name),
98 SchemaName::UnnamedAuthorization(auth) => ident_to_string(auth),
99 SchemaName::NamedAuthorization(schema_name, auth) => format!(
100 "{}.{}",
101 object_name_to_string(schema_name),
102 ident_to_string(auth)
103 ),
104 }
105}
106
107fn calc_inline_constraints_from_columns(columns: &[ColumnDef]) -> Vec<TableConstraint> {
110 let mut constraints: Vec<TableConstraint> = vec![];
111 for column in columns {
112 for ast::ColumnOptionDef { name, option } in &column.options {
113 match option {
114 ast::ColumnOption::Unique(UniqueConstraint {
115 characteristics,
116 name,
117 index_name: _index_name,
118 index_type_display: _index_type_display,
119 index_type: _index_type,
120 columns: _column,
121 index_options: _index_options,
122 nulls_distinct: _nulls_distinct,
123 }) => constraints.push(TableConstraint::Unique(UniqueConstraint {
124 name: name.clone(),
125 index_name: None,
126 index_type_display: ast::KeyOrIndexDisplay::None,
127 index_type: None,
128 columns: vec![IndexColumn {
129 column: OrderByExpr {
130 expr: SQLExpr::Identifier(column.name.clone()),
131 options: OrderByOptions {
132 asc: None,
133 nulls_first: None,
134 },
135 with_fill: None,
136 },
137 operator_class: None,
138 }],
139 index_options: vec![],
140 characteristics: *characteristics,
141 nulls_distinct: NullsDistinctOption::None,
142 })),
143 ast::ColumnOption::PrimaryKey(PrimaryKeyConstraint {
144 characteristics,
145 name: _name,
146 index_name: _index_name,
147 index_type: _index_type,
148 columns: _columns,
149 index_options: _index_options,
150 }) => {
151 constraints.push(TableConstraint::PrimaryKey(PrimaryKeyConstraint {
152 name: name.clone(),
153 index_name: None,
154 index_type: None,
155 columns: vec![IndexColumn {
156 column: OrderByExpr {
157 expr: SQLExpr::Identifier(column.name.clone()),
158 options: OrderByOptions {
159 asc: None,
160 nulls_first: None,
161 },
162 with_fill: None,
163 },
164 operator_class: None,
165 }],
166 index_options: vec![],
167 characteristics: *characteristics,
168 }))
169 }
170 ast::ColumnOption::ForeignKey(ForeignKeyConstraint {
171 foreign_table,
172 referred_columns,
173 on_delete,
174 on_update,
175 characteristics,
176 name: _name,
177 index_name: _index_name,
178 columns: _columns,
179 match_kind: _match_kind,
180 }) => {
181 constraints.push(TableConstraint::ForeignKey(ForeignKeyConstraint {
182 name: name.clone(),
183 index_name: None,
184 columns: vec![],
185 foreign_table: foreign_table.clone(),
186 referred_columns: referred_columns.clone(),
187 on_delete: *on_delete,
188 on_update: *on_update,
189 match_kind: None,
190 characteristics: *characteristics,
191 }))
192 }
193 ast::ColumnOption::Check(CheckConstraint {
194 name,
195 expr,
196 enforced: _enforced,
197 }) => constraints.push(TableConstraint::Check(CheckConstraint {
198 name: name.clone(),
199 expr: expr.clone(),
200 enforced: None,
201 })),
202 ast::ColumnOption::Default(_)
203 | ast::ColumnOption::Null
204 | ast::ColumnOption::NotNull
205 | ast::ColumnOption::DialectSpecific(_)
206 | ast::ColumnOption::CharacterSet(_)
207 | ast::ColumnOption::Generated { .. }
208 | ast::ColumnOption::Comment(_)
209 | ast::ColumnOption::Options(_)
210 | ast::ColumnOption::OnUpdate(_)
211 | ast::ColumnOption::Materialized(_)
212 | ast::ColumnOption::Ephemeral(_)
213 | ast::ColumnOption::Identity(_)
214 | ast::ColumnOption::OnConflict(_)
215 | ast::ColumnOption::Policy(_)
216 | ast::ColumnOption::Tags(_)
217 | ast::ColumnOption::Alias(_)
218 | ast::ColumnOption::Srid(_)
219 | ast::ColumnOption::Collation(_)
220 | ast::ColumnOption::Invisible => {}
221 }
222 }
223 }
224 constraints
225}
226
227impl<S: ContextProvider> SqlToRel<'_, S> {
228 pub fn statement_to_plan(&self, statement: DFStatement) -> Result<LogicalPlan> {
230 match statement {
231 DFStatement::CreateExternalTable(s) => self.external_table_to_plan(s),
232 DFStatement::Statement(s) => self.sql_statement_to_plan(*s),
233 DFStatement::CopyTo(s) => self.copy_to_plan(s),
234 DFStatement::Explain(ExplainStatement { options, statement }) => {
235 self.explain_to_plan(options, *statement)
236 }
237 DFStatement::Reset(statement) => self.reset_statement_to_plan(statement),
238 }
239 }
240
241 pub fn sql_statement_to_plan(&self, statement: Statement) -> Result<LogicalPlan> {
243 self.sql_statement_to_plan_with_context_impl(
244 statement,
245 &mut PlannerContext::new(),
246 )
247 }
248
249 pub fn sql_statement_to_plan_with_context(
251 &self,
252 statement: Statement,
253 planner_context: &mut PlannerContext,
254 ) -> Result<LogicalPlan> {
255 self.sql_statement_to_plan_with_context_impl(statement, planner_context)
256 }
257
258 fn sql_statement_to_plan_with_context_impl(
259 &self,
260 statement: Statement,
261 planner_context: &mut PlannerContext,
262 ) -> Result<LogicalPlan> {
263 match statement {
264 Statement::ExplainTable {
265 describe_alias: DescribeAlias::Describe | DescribeAlias::Desc, table_name,
267 ..
268 } => self.describe_table_to_plan(table_name),
269 Statement::Explain {
270 describe_alias: DescribeAlias::Describe | DescribeAlias::Desc, statement,
272 ..
273 } => match *statement {
274 Statement::Query(query) => self.describe_query_to_plan(*query),
275 _ => {
276 not_impl_err!("Describing statements other than SELECT not supported")
277 }
278 },
279 Statement::Explain {
280 verbose,
281 statement,
282 analyze,
283 format,
284 ..
285 } => {
286 let format = format
287 .map(|format| ExplainFormat::from_str(&format.to_string()))
288 .transpose()?;
289 let statement = DFStatement::Statement(statement);
290 let options = ExplainStatementOptions {
291 analyze,
292 verbose,
293 format,
294 analyze_level: None,
295 analyze_categories: None,
296 show_statistics: None,
297 };
298 self.explain_to_plan(options, statement)
299 }
300 Statement::Query(query) => self.query_to_plan(*query, planner_context),
301 Statement::ShowVariable { variable } => self.show_variable_to_plan(&variable),
302 Statement::Set(statement) => self.set_statement_to_plan(statement),
303 Statement::CreateTable(CreateTable {
304 temporary,
305 external,
306 global,
307 transient,
308 volatile,
309 hive_distribution,
310 hive_formats,
311 file_format,
312 location,
313 query,
314 name,
315 columns,
316 constraints,
317 if_not_exists,
318 or_replace,
319 without_rowid,
320 like,
321 clone,
322 comment,
323 on_commit,
324 on_cluster,
325 primary_key,
326 order_by,
327 partition_by,
328 cluster_by,
329 clustered_by,
330 strict,
331 copy_grants,
332 enable_schema_evolution,
333 change_tracking,
334 data_retention_time_in_days,
335 max_data_extension_time_in_days,
336 default_ddl_collation,
337 with_aggregation_policy,
338 with_row_access_policy,
339 with_tags,
340 iceberg,
341 external_volume,
342 base_location,
343 catalog,
344 catalog_sync,
345 storage_serialization_policy,
346 inherits,
347 table_options: CreateTableOptions::None,
348 dynamic,
349 version,
350 target_lag,
351 warehouse,
352 refresh_mode,
353 initialize,
354 require_user,
355 partition_of,
356 for_values,
357 snapshot,
358 with_storage_lifecycle_policy,
359 diststyle,
360 distkey,
361 sortkey,
362 backup,
363 }) => {
364 if temporary {
365 return not_impl_err!("Temporary tables not supported");
366 }
367 if external {
368 return not_impl_err!("External tables not supported");
369 }
370 if global.is_some() {
371 return not_impl_err!("Global tables not supported");
372 }
373 if transient {
374 return not_impl_err!("Transient tables not supported");
375 }
376 if volatile {
377 return not_impl_err!("Volatile tables not supported");
378 }
379 if hive_distribution != ast::HiveDistributionStyle::NONE {
380 return not_impl_err!(
381 "Hive distribution not supported: {hive_distribution:?}"
382 );
383 }
384 if hive_formats.is_some()
385 && !matches!(
386 hive_formats,
387 Some(ast::HiveFormat {
388 row_format: None,
389 serde_properties: None,
390 storage: None,
391 location: None,
392 })
393 )
394 {
395 return not_impl_err!("Hive formats not supported: {hive_formats:?}");
396 }
397 if file_format.is_some() {
398 return not_impl_err!("File format not supported");
399 }
400 if location.is_some() {
401 return not_impl_err!("Location not supported");
402 }
403 if without_rowid {
404 return not_impl_err!("Without rowid not supported");
405 }
406 if like.is_some() {
407 return not_impl_err!("Like not supported");
408 }
409 if clone.is_some() {
410 return not_impl_err!("Clone not supported");
411 }
412 if comment.is_some() {
413 return not_impl_err!("Comment not supported");
414 }
415 if on_commit.is_some() {
416 return not_impl_err!("On commit not supported");
417 }
418 if on_cluster.is_some() {
419 return not_impl_err!("On cluster not supported");
420 }
421 if primary_key.is_some() {
422 return not_impl_err!("Primary key not supported");
423 }
424 if order_by.is_some() {
425 return not_impl_err!("Order by not supported");
426 }
427 if partition_by.is_some() {
428 return not_impl_err!("Partition by not supported");
429 }
430 if cluster_by.is_some() {
431 return not_impl_err!("Cluster by not supported");
432 }
433 if clustered_by.is_some() {
434 return not_impl_err!("Clustered by not supported");
435 }
436 if strict {
437 return not_impl_err!("Strict not supported");
438 }
439 if copy_grants {
440 return not_impl_err!("Copy grants not supported");
441 }
442 if enable_schema_evolution.is_some() {
443 return not_impl_err!("Enable schema evolution not supported");
444 }
445 if change_tracking.is_some() {
446 return not_impl_err!("Change tracking not supported");
447 }
448 if data_retention_time_in_days.is_some() {
449 return not_impl_err!("Data retention time in days not supported");
450 }
451 if max_data_extension_time_in_days.is_some() {
452 return not_impl_err!(
453 "Max data extension time in days not supported"
454 );
455 }
456 if default_ddl_collation.is_some() {
457 return not_impl_err!("Default DDL collation not supported");
458 }
459 if with_aggregation_policy.is_some() {
460 return not_impl_err!("With aggregation policy not supported");
461 }
462 if with_row_access_policy.is_some() {
463 return not_impl_err!("With row access policy not supported");
464 }
465 if with_tags.is_some() {
466 return not_impl_err!("With tags not supported");
467 }
468 if iceberg {
469 return not_impl_err!("Iceberg not supported");
470 }
471 if external_volume.is_some() {
472 return not_impl_err!("External volume not supported");
473 }
474 if base_location.is_some() {
475 return not_impl_err!("Base location not supported");
476 }
477 if catalog.is_some() {
478 return not_impl_err!("Catalog not supported");
479 }
480 if catalog_sync.is_some() {
481 return not_impl_err!("Catalog sync not supported");
482 }
483 if storage_serialization_policy.is_some() {
484 return not_impl_err!("Storage serialization policy not supported");
485 }
486 if inherits.is_some() {
487 return not_impl_err!("Table inheritance not supported");
488 }
489 if dynamic {
490 return not_impl_err!("Dynamic tables not supported");
491 }
492 if version.is_some() {
493 return not_impl_err!("Version not supported");
494 }
495 if target_lag.is_some() {
496 return not_impl_err!("Target lag not supported");
497 }
498 if warehouse.is_some() {
499 return not_impl_err!("Warehouse not supported");
500 }
501 if refresh_mode.is_some() {
502 return not_impl_err!("Refresh mode not supported");
503 }
504 if initialize.is_some() {
505 return not_impl_err!("Initialize not supported");
506 }
507 if require_user {
508 return not_impl_err!("Require user not supported");
509 }
510 if partition_of.is_some() {
511 return not_impl_err!("PARTITION OF not supported");
512 }
513 if for_values.is_some() {
514 return not_impl_err!("PARTITION OF .. FOR VALUES .. not supported");
515 }
516 if snapshot {
517 return not_impl_err!("Snapshot tables not supported");
518 }
519 if with_storage_lifecycle_policy.is_some() {
520 return not_impl_err!("WITH STORAGE LIFECYCLE POLICY not supported");
521 }
522 if diststyle.is_some() {
523 return not_impl_err!("DISTSTYLE not supported");
524 }
525 if distkey.is_some() {
526 return not_impl_err!("DISTKEY not supported");
527 }
528 if sortkey.is_some() {
529 return not_impl_err!("SORTKEY not supported");
530 }
531 if backup.is_some() {
532 return not_impl_err!("BACKUP not supported");
533 }
534 let mut all_constraints = constraints;
536 let inline_constraints = calc_inline_constraints_from_columns(&columns);
537 all_constraints.extend(inline_constraints);
538 let column_defaults =
540 self.build_column_defaults(&columns, planner_context)?;
541
542 let has_columns = !columns.is_empty();
543 let schema = self.build_schema(columns)?.to_dfschema_ref()?;
544 if has_columns {
545 planner_context.set_table_schema(Some(Arc::clone(&schema)));
546 }
547
548 match query {
549 Some(query) => {
550 let plan = self.query_to_plan(*query, planner_context)?;
551 let input_schema = plan.schema();
552
553 let plan = if has_columns {
554 if schema.fields().len() != input_schema.fields().len() {
555 return plan_err!(
556 "Mismatch: {} columns specified, but result has {} columns",
557 schema.fields().len(),
558 input_schema.fields().len()
559 );
560 }
561 let input_columns = input_schema.columns();
562 let project_exprs = schema
563 .fields()
564 .iter()
565 .zip(input_columns)
566 .map(|(field, input_column)| {
567 cast(
568 Expr::Column(input_column),
569 field.data_type().clone(),
570 )
571 .alias(field.name())
572 })
573 .collect::<Vec<_>>();
574
575 LogicalPlanBuilder::from(plan.clone())
576 .project(project_exprs)?
577 .build()?
578 } else {
579 plan
580 };
581
582 let constraints = self.new_constraint_from_table_constraints(
583 &all_constraints,
584 plan.schema(),
585 )?;
586
587 Ok(LogicalPlan::Ddl(DdlStatement::CreateMemoryTable(
588 CreateMemoryTable {
589 name: self.object_name_to_table_reference(name)?,
590 constraints,
591 input: Arc::new(plan),
592 if_not_exists,
593 or_replace,
594 column_defaults,
595 temporary,
596 },
597 )))
598 }
599
600 None => {
601 let plan = EmptyRelation {
602 produce_one_row: false,
603 schema,
604 };
605 let plan = LogicalPlan::EmptyRelation(plan);
606 let constraints = self.new_constraint_from_table_constraints(
607 &all_constraints,
608 plan.schema(),
609 )?;
610 Ok(LogicalPlan::Ddl(DdlStatement::CreateMemoryTable(
611 CreateMemoryTable {
612 name: self.object_name_to_table_reference(name)?,
613 constraints,
614 input: Arc::new(plan),
615 if_not_exists,
616 or_replace,
617 column_defaults,
618 temporary,
619 },
620 )))
621 }
622 }
623 }
624 Statement::CreateView(ast::CreateView {
625 or_replace,
626 materialized,
627 name,
628 columns,
629 query,
630 options: CreateTableOptions::None,
631 cluster_by,
632 comment,
633 with_no_schema_binding,
634 if_not_exists,
635 temporary,
636 to,
637 params,
638 or_alter,
639 secure,
640 name_before_not_exists,
641 copy_grants,
642 }) => {
643 if materialized {
644 return not_impl_err!("Materialized views not supported")?;
645 }
646 if !cluster_by.is_empty() {
647 return not_impl_err!("Cluster by not supported")?;
648 }
649 if comment.is_some() {
650 return not_impl_err!("Comment not supported")?;
651 }
652 if with_no_schema_binding {
653 return not_impl_err!("With no schema binding not supported")?;
654 }
655 if if_not_exists {
656 return not_impl_err!("If not exists not supported")?;
657 }
658 if to.is_some() {
659 return not_impl_err!("To not supported")?;
660 }
661 if copy_grants {
662 return not_impl_err!("COPY GRANTS not supported")?;
663 }
664
665 let stmt = Statement::CreateView(ast::CreateView {
668 or_replace,
669 materialized,
670 name,
671 columns,
672 query,
673 options: CreateTableOptions::None,
674 cluster_by,
675 comment,
676 with_no_schema_binding,
677 if_not_exists,
678 temporary,
679 to,
680 params,
681 or_alter,
682 secure,
683 name_before_not_exists,
684 copy_grants,
685 });
686 let sql = stmt.to_string();
687 let Statement::CreateView(ast::CreateView {
688 name,
689 columns,
690 query,
691 or_replace,
692 temporary,
693 ..
694 }) = stmt
695 else {
696 return internal_err!("Unreachable code in create view");
697 };
698
699 let columns = columns
700 .into_iter()
701 .map(|view_column_def| {
702 if let Some(options) = view_column_def.options {
703 plan_err!(
704 "Options not supported for view columns: {options:?}"
705 )
706 } else {
707 Ok(view_column_def.name)
708 }
709 })
710 .collect::<Result<Vec<_>>>()?;
711
712 let mut plan = self.query_to_plan(*query, &mut PlannerContext::new())?;
713 plan = self.apply_expr_alias(plan, columns)?;
714
715 Ok(LogicalPlan::Ddl(DdlStatement::CreateView(CreateView {
716 name: self.object_name_to_table_reference(name)?,
717 input: Arc::new(plan),
718 or_replace,
719 definition: Some(sql),
720 temporary,
721 })))
722 }
723 Statement::ShowCreate { obj_type, obj_name } => match obj_type {
724 ShowCreateObject::Table => self.show_create_table_to_plan(obj_name),
725 _ => {
726 not_impl_err!("Only `SHOW CREATE TABLE ...` statement is supported")
727 }
728 },
729 Statement::CreateSchema {
730 schema_name,
731 if_not_exists,
732 ..
733 } => Ok(LogicalPlan::Ddl(DdlStatement::CreateCatalogSchema(
734 CreateCatalogSchema {
735 schema_name: get_schema_name(&schema_name),
736 if_not_exists,
737 schema: Arc::new(DFSchema::empty()),
738 },
739 ))),
740 Statement::CreateDatabase {
741 db_name,
742 if_not_exists,
743 ..
744 } => Ok(LogicalPlan::Ddl(DdlStatement::CreateCatalog(
745 CreateCatalog {
746 catalog_name: object_name_to_string(&db_name),
747 if_not_exists,
748 schema: Arc::new(DFSchema::empty()),
749 },
750 ))),
751 Statement::Drop {
752 object_type,
753 if_exists,
754 mut names,
755 cascade,
756 restrict: _,
757 purge: _,
758 temporary: _,
759 table: _,
760 } => {
761 let name = match names.len() {
764 0 => Err(ParserError("Missing table name.".to_string()).into()),
765 1 => self.object_name_to_table_reference(names.pop().unwrap()),
766 _ => {
767 Err(ParserError("Multiple objects not supported".to_string())
768 .into())
769 }
770 }?;
771
772 match object_type {
773 ObjectType::Table => {
774 Ok(LogicalPlan::Ddl(DdlStatement::DropTable(DropTable {
775 name,
776 if_exists,
777 schema: DFSchemaRef::new(DFSchema::empty()),
778 })))
779 }
780 ObjectType::View => {
781 Ok(LogicalPlan::Ddl(DdlStatement::DropView(DropView {
782 name,
783 if_exists,
784 schema: DFSchemaRef::new(DFSchema::empty()),
785 })))
786 }
787 ObjectType::Schema => {
788 let name = match name {
789 TableReference::Bare { table } => {
790 Ok(SchemaReference::Bare { schema: table })
791 }
792 TableReference::Partial { schema, table } => {
793 Ok(SchemaReference::Full {
794 schema: table,
795 catalog: schema,
796 })
797 }
798 TableReference::Full {
799 catalog: _,
800 schema: _,
801 table: _,
802 } => Err(ParserError(
803 "Invalid schema specifier (has 3 parts)".to_string(),
804 )),
805 }?;
806 Ok(LogicalPlan::Ddl(DdlStatement::DropCatalogSchema(
807 DropCatalogSchema {
808 name,
809 if_exists,
810 cascade,
811 schema: DFSchemaRef::new(DFSchema::empty()),
812 },
813 )))
814 }
815 _ => not_impl_err!(
816 "Only `DROP TABLE/VIEW/SCHEMA ...` statement is supported currently"
817 ),
818 }
819 }
820 Statement::Prepare {
821 name,
822 data_types,
823 statement,
824 } => {
825 let mut fields: Vec<FieldRef> = data_types
827 .into_iter()
828 .map(|t| self.convert_data_type_to_field(&t))
829 .collect::<Result<_>>()?;
830
831 let mut planner_context = PlannerContext::new()
833 .with_prepare_param_data_types(
834 fields.iter().cloned().map(Some).collect(),
835 );
836
837 let plan = self.sql_statement_to_plan_with_context_impl(
839 *statement,
840 &mut planner_context,
841 )?;
842
843 if fields.is_empty() {
844 let map_types = plan.get_parameter_fields()?;
845 let param_types: Vec<_> = (1..=map_types.len())
846 .filter_map(|i| {
847 let key = format!("${i}");
848 map_types.get(&key).and_then(|opt| opt.clone())
849 })
850 .collect();
851 fields.extend(param_types.iter().cloned());
852 planner_context.with_prepare_param_data_types(
853 param_types.into_iter().map(Some).collect(),
854 );
855 }
856
857 Ok(LogicalPlan::Statement(PlanStatement::Prepare(Prepare {
858 name: ident_to_string(&name),
859 fields,
860 input: Arc::new(plan),
861 })))
862 }
863 Statement::Execute {
864 name,
865 parameters,
866 using,
867 has_parentheses: _,
870 immediate,
871 into,
872 output,
873 default,
874 } => {
875 if !using.is_empty() {
877 return not_impl_err!(
878 "Execute statement with USING is not supported"
879 );
880 }
881 if immediate {
882 return not_impl_err!(
883 "Execute statement with IMMEDIATE is not supported"
884 );
885 }
886 if !into.is_empty() {
887 return not_impl_err!("Execute statement with INTO is not supported");
888 }
889 if output {
890 return not_impl_err!(
891 "Execute statement with OUTPUT is not supported"
892 );
893 }
894 if default {
895 return not_impl_err!(
896 "Execute statement with DEFAULT is not supported"
897 );
898 }
899 let name = name.ok_or_else(|| {
900 plan_datafusion_err!("EXECUTE statement requires a name")
901 })?;
902
903 let empty_schema = DFSchema::empty();
904 let parameters = parameters
905 .into_iter()
906 .map(|expr| self.sql_to_expr(expr, &empty_schema, planner_context))
907 .collect::<Result<Vec<Expr>>>()?;
908
909 Ok(LogicalPlan::Statement(PlanStatement::Execute(Execute {
910 name: object_name_to_string(&name),
911 parameters,
912 })))
913 }
914 Statement::Deallocate {
915 name,
916 prepare: _,
918 } => Ok(LogicalPlan::Statement(PlanStatement::Deallocate(
919 Deallocate {
920 name: ident_to_string(&name),
921 },
922 ))),
923
924 Statement::ShowTables {
925 extended,
926 full,
927 terse,
928 history,
929 external,
930 show_options,
931 } => {
932 if extended {
935 return not_impl_err!("SHOW TABLES EXTENDED not supported")?;
936 }
937 if full {
938 return not_impl_err!("SHOW FULL TABLES not supported")?;
939 }
940 if terse {
941 return not_impl_err!("SHOW TERSE TABLES not supported")?;
942 }
943 if history {
944 return not_impl_err!("SHOW TABLES HISTORY not supported")?;
945 }
946 if external {
947 return not_impl_err!("SHOW EXTERNAL TABLES not supported")?;
948 }
949 let ShowStatementOptions {
950 show_in,
951 starts_with,
952 limit,
953 limit_from,
954 filter_position,
955 } = show_options;
956 if show_in.is_some() {
957 return not_impl_err!("SHOW TABLES IN not supported")?;
958 }
959 if starts_with.is_some() {
960 return not_impl_err!("SHOW TABLES LIKE not supported")?;
961 }
962 if limit.is_some() {
963 return not_impl_err!("SHOW TABLES LIMIT not supported")?;
964 }
965 if limit_from.is_some() {
966 return not_impl_err!("SHOW TABLES LIMIT FROM not supported")?;
967 }
968 if filter_position.is_some() {
969 return not_impl_err!("SHOW TABLES FILTER not supported")?;
970 }
971 self.show_tables_to_plan()
972 }
973
974 Statement::ShowColumns {
975 extended,
976 full,
977 show_options,
978 } => {
979 let ShowStatementOptions {
980 show_in,
981 starts_with,
982 limit,
983 limit_from,
984 filter_position,
985 } = show_options;
986 if starts_with.is_some() {
987 return not_impl_err!("SHOW COLUMNS LIKE not supported")?;
988 }
989 if limit.is_some() {
990 return not_impl_err!("SHOW COLUMNS LIMIT not supported")?;
991 }
992 if limit_from.is_some() {
993 return not_impl_err!("SHOW COLUMNS LIMIT FROM not supported")?;
994 }
995 if filter_position.is_some() {
996 return not_impl_err!(
997 "SHOW COLUMNS with WHERE or LIKE is not supported"
998 )?;
999 }
1000 let Some(ShowStatementIn {
1001 clause: _,
1004 parent_type,
1005 parent_name,
1006 }) = show_in
1007 else {
1008 return plan_err!("SHOW COLUMNS requires a table name");
1009 };
1010
1011 if let Some(parent_type) = parent_type {
1012 return not_impl_err!("SHOW COLUMNS IN {parent_type} not supported");
1013 }
1014 let Some(table_name) = parent_name else {
1015 return plan_err!("SHOW COLUMNS requires a table name");
1016 };
1017
1018 self.show_columns_to_plan(extended, full, table_name)
1019 }
1020
1021 Statement::ShowFunctions { filter, .. } => {
1022 self.show_functions_to_plan(filter)
1023 }
1024
1025 Statement::Insert(Insert {
1026 or,
1027 into,
1028 columns,
1029 overwrite,
1030 source,
1031 partitioned,
1032 after_columns,
1033 table,
1034 on,
1035 returning,
1036 ignore,
1037 table_alias,
1038 mut replace_into,
1039 priority,
1040 insert_alias,
1041 assignments,
1042 has_table_keyword,
1043 settings,
1044 format_clause,
1045 insert_token: _, optimizer_hints,
1047 output,
1048 multi_table_insert_type,
1049 multi_table_into_clauses,
1050 multi_table_when_clauses,
1051 multi_table_else_clause,
1052 }) => {
1053 let table_name = match table {
1054 TableObject::TableName(table_name) => table_name,
1055 TableObject::TableFunction(_) => {
1056 return not_impl_err!(
1057 "INSERT INTO Table functions not supported"
1058 );
1059 }
1060 TableObject::TableQuery(_) => {
1061 return not_impl_err!(
1062 "INSERT INTO subquery target not supported"
1063 );
1064 }
1065 };
1066 if let Some(or) = or {
1067 match or {
1068 SqliteOnConflict::Replace => replace_into = true,
1069 _ => plan_err!("Inserts with {or} clause is not supported")?,
1070 }
1071 }
1072 if partitioned.is_some() {
1073 plan_err!("Partitioned inserts not yet supported")?;
1074 }
1075 if !after_columns.is_empty() {
1076 plan_err!("After-columns clause not supported")?;
1077 }
1078 if on.is_some() {
1079 plan_err!("Insert-on clause not supported")?;
1080 }
1081 if returning.is_some() {
1082 plan_err!("Insert-returning clause not supported")?;
1083 }
1084 if ignore {
1085 plan_err!("Insert-ignore clause not supported")?;
1086 }
1087 let Some(source) = source else {
1088 plan_err!("Inserts without a source not supported")?
1089 };
1090 if let Some(table_alias) = table_alias {
1091 plan_err!(
1092 "Inserts with a table alias not supported: {table_alias:?}"
1093 )?
1094 };
1095 if let Some(priority) = priority {
1096 plan_err!(
1097 "Inserts with a `PRIORITY` clause not supported: {priority:?}"
1098 )?
1099 };
1100 if insert_alias.is_some() {
1101 plan_err!("Inserts with an alias not supported")?;
1102 }
1103 if !assignments.is_empty() {
1104 plan_err!("Inserts with assignments not supported")?;
1105 }
1106 if settings.is_some() {
1107 plan_err!("Inserts with settings not supported")?;
1108 }
1109 if format_clause.is_some() {
1110 plan_err!("Inserts with format clause not supported")?;
1111 }
1112 if !optimizer_hints.is_empty() {
1113 plan_err!("Optimizer hints not supported")?;
1114 }
1115 if output.is_some() {
1116 plan_err!("Insert OUTPUT clause not supported")?;
1117 }
1118 if multi_table_insert_type.is_some()
1119 || !multi_table_into_clauses.is_empty()
1120 || !multi_table_when_clauses.is_empty()
1121 || multi_table_else_clause.is_some()
1122 {
1123 plan_err!("Multi-table INSERT not supported")?;
1124 }
1125 let _ = into;
1127 let _ = has_table_keyword;
1128 self.insert_to_plan(table_name, columns, source, overwrite, replace_into)
1129 }
1130 Statement::Update(Update {
1131 table,
1132 assignments,
1133 from,
1134 selection,
1135 returning,
1136 or,
1137 limit,
1138 update_token: _,
1139 optimizer_hints,
1140 output,
1141 order_by,
1142 }) => {
1143 let from_clauses =
1144 from.map(|update_table_from_kind| match update_table_from_kind {
1145 UpdateTableFromKind::BeforeSet(from_clauses) => from_clauses,
1146 UpdateTableFromKind::AfterSet(from_clauses) => from_clauses,
1147 });
1148 if from_clauses.as_ref().is_some_and(|f| f.len() > 1) {
1150 not_impl_err!(
1151 "Multiple tables in UPDATE SET FROM not yet supported"
1152 )?;
1153 }
1154 let update_from = from_clauses.and_then(|mut f| f.pop());
1155
1156 if update_from.is_some() {
1159 return not_impl_err!("UPDATE ... FROM is not supported");
1160 }
1161
1162 if returning.is_some() {
1163 plan_err!("Update-returning clause not yet supported")?;
1164 }
1165 if or.is_some() {
1166 plan_err!("ON conflict not supported")?;
1167 }
1168 if limit.is_some() {
1169 return not_impl_err!("Update-limit clause not supported")?;
1170 }
1171 if !optimizer_hints.is_empty() {
1172 plan_err!("Optimizer hints not supported")?;
1173 }
1174 if output.is_some() {
1175 plan_err!("Update OUTPUT clause not supported")?;
1176 }
1177 if !order_by.is_empty() {
1178 plan_err!("Update ORDER BY not supported")?;
1179 }
1180 self.update_to_plan(table, &assignments, update_from, selection)
1181 }
1182
1183 Statement::Delete(Delete {
1184 tables,
1185 using,
1186 selection,
1187 returning,
1188 from,
1189 order_by,
1190 limit,
1191 delete_token: _,
1192 optimizer_hints,
1193 output,
1194 }) => {
1195 if !tables.is_empty() {
1196 plan_err!("DELETE <TABLE> not supported")?;
1197 }
1198
1199 if using.is_some() {
1200 plan_err!("Using clause not supported")?;
1201 }
1202
1203 if returning.is_some() {
1204 plan_err!("Delete-returning clause not yet supported")?;
1205 }
1206
1207 if !order_by.is_empty() {
1208 plan_err!("Delete-order-by clause not yet supported")?;
1209 }
1210
1211 if !optimizer_hints.is_empty() {
1212 plan_err!("Optimizer hints not supported")?;
1213 }
1214 if output.is_some() {
1215 plan_err!("Delete OUTPUT clause not supported")?;
1216 }
1217
1218 let table_name = self.get_delete_target(from)?;
1219 self.delete_to_plan(&table_name, selection, limit)
1220 }
1221
1222 Statement::Merge(merge) => self.merge_to_plan(merge),
1223
1224 Statement::StartTransaction {
1225 modes,
1226 begin: false,
1227 modifier,
1228 transaction,
1229 statements,
1230 has_end_keyword,
1231 exception,
1232 } => {
1233 if let Some(modifier) = modifier {
1234 return not_impl_err!(
1235 "Transaction modifier not supported: {modifier}"
1236 );
1237 }
1238 if !statements.is_empty() {
1239 return not_impl_err!(
1240 "Transaction with multiple statements not supported"
1241 );
1242 }
1243 if exception.is_some() {
1244 return not_impl_err!(
1245 "Transaction with exception statements not supported"
1246 );
1247 }
1248 if has_end_keyword {
1249 return not_impl_err!("Transaction with END keyword not supported");
1250 }
1251 self.validate_transaction_kind(transaction.as_ref())?;
1252 let isolation_level: ast::TransactionIsolationLevel = modes
1253 .iter()
1254 .filter_map(|m: &TransactionMode| match m {
1255 TransactionMode::AccessMode(_) => None,
1256 TransactionMode::IsolationLevel(level) => Some(level),
1257 })
1258 .next_back()
1259 .copied()
1260 .unwrap_or(ast::TransactionIsolationLevel::Serializable);
1261 let access_mode: ast::TransactionAccessMode = modes
1262 .iter()
1263 .filter_map(|m: &TransactionMode| match m {
1264 TransactionMode::AccessMode(mode) => Some(mode),
1265 TransactionMode::IsolationLevel(_) => None,
1266 })
1267 .next_back()
1268 .copied()
1269 .unwrap_or(ast::TransactionAccessMode::ReadWrite);
1270 let isolation_level = match isolation_level {
1271 ast::TransactionIsolationLevel::ReadUncommitted => {
1272 TransactionIsolationLevel::ReadUncommitted
1273 }
1274 ast::TransactionIsolationLevel::ReadCommitted => {
1275 TransactionIsolationLevel::ReadCommitted
1276 }
1277 ast::TransactionIsolationLevel::RepeatableRead => {
1278 TransactionIsolationLevel::RepeatableRead
1279 }
1280 ast::TransactionIsolationLevel::Serializable => {
1281 TransactionIsolationLevel::Serializable
1282 }
1283 ast::TransactionIsolationLevel::Snapshot => {
1284 TransactionIsolationLevel::Snapshot
1285 }
1286 };
1287 let access_mode = match access_mode {
1288 ast::TransactionAccessMode::ReadOnly => {
1289 TransactionAccessMode::ReadOnly
1290 }
1291 ast::TransactionAccessMode::ReadWrite => {
1292 TransactionAccessMode::ReadWrite
1293 }
1294 };
1295 let statement = PlanStatement::TransactionStart(TransactionStart {
1296 access_mode,
1297 isolation_level,
1298 });
1299 Ok(LogicalPlan::Statement(statement))
1300 }
1301 Statement::Commit {
1302 chain,
1303 end,
1304 modifier,
1305 } => {
1306 if end {
1307 return not_impl_err!("COMMIT AND END not supported");
1308 };
1309 if let Some(modifier) = modifier {
1310 return not_impl_err!("COMMIT {modifier} not supported");
1311 };
1312 let statement = PlanStatement::TransactionEnd(TransactionEnd {
1313 conclusion: TransactionConclusion::Commit,
1314 chain,
1315 });
1316 Ok(LogicalPlan::Statement(statement))
1317 }
1318 Statement::Rollback { chain, savepoint } => {
1319 if savepoint.is_some() {
1320 plan_err!("Savepoints not supported")?;
1321 }
1322 let statement = PlanStatement::TransactionEnd(TransactionEnd {
1323 conclusion: TransactionConclusion::Rollback,
1324 chain,
1325 });
1326 Ok(LogicalPlan::Statement(statement))
1327 }
1328 Statement::CreateFunction(ast::CreateFunction {
1329 or_replace,
1330 temporary,
1331 name,
1332 args,
1333 return_type,
1334 function_body,
1335 behavior,
1336 language,
1337 ..
1338 }) => {
1339 let return_type = match return_type {
1340 Some(ast::FunctionReturnType::DataType(t)) => {
1341 Some(self.convert_data_type_to_field(&t)?)
1342 }
1343 Some(ast::FunctionReturnType::SetOf(_)) => {
1344 return not_impl_err!(
1345 "RETURNS SETOF in CREATE FUNCTION is not supported"
1346 );
1347 }
1348 None => None,
1349 };
1350 let mut planner_context = PlannerContext::new();
1351 let empty_schema = &DFSchema::empty();
1352
1353 let args = match args {
1354 Some(function_args) => {
1355 let function_args = function_args
1356 .into_iter()
1357 .map(|arg| {
1358 let data_type =
1359 self.convert_data_type_to_field(&arg.data_type)?;
1360
1361 let default_expr = match arg.default_expr {
1362 Some(expr) => Some(self.sql_to_expr(
1363 expr,
1364 empty_schema,
1365 &mut planner_context,
1366 )?),
1367 None => None,
1368 };
1369 Ok(OperateFunctionArg {
1370 name: arg.name,
1371 default_expr,
1372 data_type: data_type.data_type().clone(),
1373 })
1374 })
1375 .collect::<Result<Vec<OperateFunctionArg>>>();
1376 Some(function_args?)
1377 }
1378 None => None,
1379 };
1380 let first_default = match args.as_ref() {
1382 Some(arg) => arg.iter().position(|t| t.default_expr.is_some()),
1383 None => None,
1384 };
1385 let last_non_default = match args.as_ref() {
1386 Some(arg) => arg
1387 .iter()
1388 .rev()
1389 .position(|t| t.default_expr.is_none())
1390 .map(|reverse_pos| arg.len() - reverse_pos - 1),
1391 None => None,
1392 };
1393 if let (Some(pos_default), Some(pos_non_default)) =
1394 (first_default, last_non_default)
1395 && pos_non_default > pos_default
1396 {
1397 return plan_err!(
1398 "Non-default arguments cannot follow default arguments."
1399 );
1400 }
1401 let name = match &name.0[..] {
1403 [] => exec_err!("Function should have name")?,
1404 [n] => n.as_ident().unwrap().value.clone(),
1405 [..] => not_impl_err!("Qualified functions are not supported")?,
1406 };
1407 let arg_types = args.as_ref().map(|arg| {
1411 arg.iter()
1412 .map(|t| {
1413 let name = match t.name.clone() {
1414 Some(name) => name.value,
1415 None => "".to_string(),
1416 };
1417 Arc::new(Field::new(name, t.data_type.clone(), true))
1418 })
1419 .collect::<Vec<_>>()
1420 });
1421 if let Some(ref fields) = arg_types {
1423 let count_positional =
1424 fields.iter().filter(|f| f.name() == "").count();
1425 if !(count_positional == 0 || count_positional == fields.len()) {
1426 return plan_err!(
1427 "All function arguments must use either named or positional style."
1428 );
1429 }
1430 }
1431 let mut planner_context = PlannerContext::new()
1432 .with_prepare_param_data_types(
1433 arg_types
1434 .unwrap_or_default()
1435 .into_iter()
1436 .map(Some)
1437 .collect(),
1438 );
1439
1440 let function_body = match function_body {
1441 Some(r) => Some(self.sql_to_expr(
1442 match r {
1443 ast::CreateFunctionBody::AsBeforeOptions{body: expr, link_symbol: _link_symbol} => expr,
1445 ast::CreateFunctionBody::AsAfterOptions(expr) => expr,
1446 ast::CreateFunctionBody::Return(expr) => expr,
1447 ast::CreateFunctionBody::AsBeginEnd(_) => {
1448 return not_impl_err!(
1449 "BEGIN/END enclosed function body syntax is not supported"
1450 )?;
1451 }
1452 ast::CreateFunctionBody::AsReturnExpr(_)
1453 | ast::CreateFunctionBody::AsReturnSelect(_) => {
1454 return not_impl_err!(
1455 "AS RETURN function syntax is not supported"
1456 )?
1457 }
1458 },
1459 &DFSchema::empty(),
1460 &mut planner_context,
1461 )?),
1462 None => None,
1463 };
1464
1465 let params = CreateFunctionBody {
1466 language,
1467 behavior: behavior.map(|b| match b {
1468 ast::FunctionBehavior::Immutable => Volatility::Immutable,
1469 ast::FunctionBehavior::Stable => Volatility::Stable,
1470 ast::FunctionBehavior::Volatile => Volatility::Volatile,
1471 }),
1472 function_body,
1473 };
1474
1475 let statement = DdlStatement::CreateFunction(Box::new(CreateFunction {
1476 or_replace,
1477 temporary,
1478 name,
1479 return_type: return_type.map(|f| f.data_type().clone()),
1480 args,
1481 params,
1482 schema: DFSchemaRef::new(DFSchema::empty()),
1483 }));
1484
1485 Ok(LogicalPlan::Ddl(statement))
1486 }
1487 Statement::DropFunction(ast::DropFunction {
1488 if_exists,
1489 func_desc,
1490 drop_behavior: _,
1491 }) => {
1492 if let Some(desc) = func_desc.first() {
1495 let name = match &desc.name.0[..] {
1497 [] => exec_err!("Function should have name")?,
1498 [n] => n.as_ident().unwrap().value.clone(),
1499 [..] => not_impl_err!("Qualified functions are not supported")?,
1500 };
1501 let statement = DdlStatement::DropFunction(DropFunction {
1502 if_exists,
1503 name,
1504 schema: DFSchemaRef::new(DFSchema::empty()),
1505 });
1506 Ok(LogicalPlan::Ddl(statement))
1507 } else {
1508 exec_err!("Function name not provided")
1509 }
1510 }
1511 Statement::Truncate(ast::Truncate {
1512 table_names,
1513 partitions,
1514 identity,
1515 cascade,
1516 on_cluster,
1517 table,
1518 if_exists,
1519 }) => {
1520 let _ = table; if table_names.len() != 1 {
1522 return not_impl_err!(
1523 "TRUNCATE with multiple tables is not supported"
1524 );
1525 }
1526
1527 let target = &table_names[0];
1528 if target.only {
1529 return not_impl_err!("TRUNCATE with ONLY is not supported");
1530 }
1531 if partitions.is_some() {
1532 return not_impl_err!("TRUNCATE with PARTITION is not supported");
1533 }
1534 if identity.is_some() {
1535 return not_impl_err!(
1536 "TRUNCATE with RESTART/CONTINUE IDENTITY is not supported"
1537 );
1538 }
1539 if cascade.is_some() {
1540 return not_impl_err!(
1541 "TRUNCATE with CASCADE/RESTRICT is not supported"
1542 );
1543 }
1544 if on_cluster.is_some() {
1545 return not_impl_err!("TRUNCATE with ON CLUSTER is not supported");
1546 }
1547 if if_exists {
1548 return not_impl_err!("TRUNCATE .. with IF EXISTS is not supported");
1549 }
1550 let table = self.object_name_to_table_reference(target.name.clone())?;
1551 let source = self.context_provider.get_table_source(table.clone())?;
1552
1553 Ok(LogicalPlan::Dml(DmlStatement::new(
1556 table.clone(),
1557 source,
1558 WriteOp::Truncate,
1559 Arc::new(LogicalPlan::EmptyRelation(EmptyRelation {
1560 produce_one_row: false,
1561 schema: DFSchemaRef::new(DFSchema::empty()),
1562 })),
1563 )))
1564 }
1565 Statement::CreateIndex(CreateIndex {
1566 name,
1567 table_name,
1568 using,
1569 columns,
1570 unique,
1571 if_not_exists,
1572 ..
1573 }) => {
1574 let name: Option<String> = name.as_ref().map(object_name_to_string);
1575 let table = self.object_name_to_table_reference(table_name)?;
1576 let table_schema = self
1577 .context_provider
1578 .get_table_source(table.clone())?
1579 .schema()
1580 .to_dfschema_ref()?;
1581 let using: Option<String> =
1582 using.as_ref().map(|index_type| match index_type {
1583 IndexType::Custom(ident) => ident_to_string(ident),
1584 _ => index_type.to_string().to_ascii_lowercase(),
1585 });
1586 let order_by_exprs: Vec<OrderByExpr> =
1587 columns.into_iter().map(|col| col.column).collect();
1588 let columns = self.order_by_to_sort_expr(
1589 order_by_exprs,
1590 &table_schema,
1591 planner_context,
1592 false,
1593 None,
1594 )?;
1595 Ok(LogicalPlan::Ddl(DdlStatement::CreateIndex(
1596 PlanCreateIndex {
1597 name,
1598 table,
1599 using,
1600 columns,
1601 unique,
1602 if_not_exists,
1603 schema: DFSchemaRef::new(DFSchema::empty()),
1604 },
1605 )))
1606 }
1607 stmt => {
1608 not_impl_err!("Unsupported SQL statement: {stmt}")
1609 }
1610 }
1611 }
1612
1613 fn get_delete_target(&self, from: FromTable) -> Result<ObjectName> {
1614 let mut from = match from {
1615 FromTable::WithFromKeyword(v) => v,
1616 FromTable::WithoutKeyword(v) => v,
1617 };
1618
1619 if from.len() != 1 {
1620 return not_impl_err!(
1621 "DELETE FROM only supports single table, got {}: {from:?}",
1622 from.len()
1623 );
1624 }
1625 let table_factor = from.pop().unwrap();
1626 if !table_factor.joins.is_empty() {
1627 return not_impl_err!("DELETE FROM only supports single table, got: joins");
1628 }
1629 let TableFactor::Table { name, .. } = table_factor.relation else {
1630 return not_impl_err!(
1631 "DELETE FROM only supports single table, got: {table_factor:?}"
1632 );
1633 };
1634
1635 Ok(name)
1636 }
1637
1638 fn show_tables_to_plan(&self) -> Result<LogicalPlan> {
1640 if self.has_table("information_schema", "tables") {
1641 let query = "SELECT * FROM information_schema.tables;";
1642 let mut rewrite = DFParser::parse_sql(query)?;
1643 assert_eq!(rewrite.len(), 1);
1644 self.statement_to_plan(rewrite.pop_front().unwrap()) } else {
1646 plan_err!("SHOW TABLES is not supported unless information_schema is enabled")
1647 }
1648 }
1649
1650 fn describe_table_to_plan(&self, table_name: ObjectName) -> Result<LogicalPlan> {
1651 let table_ref = self.object_name_to_table_reference(table_name)?;
1652
1653 let table_source = self.context_provider.get_table_source(table_ref)?;
1654
1655 let schema = table_source.schema();
1656
1657 let output_schema = DFSchema::try_from(LogicalPlan::describe_schema()).unwrap();
1658
1659 Ok(LogicalPlan::DescribeTable(DescribeTable {
1660 schema,
1661 output_schema: Arc::new(output_schema),
1662 }))
1663 }
1664
1665 fn describe_query_to_plan(&self, query: Query) -> Result<LogicalPlan> {
1666 let plan = self.query_to_plan(query, &mut PlannerContext::new())?;
1667
1668 let schema = Arc::new(plan.schema().as_arrow().clone());
1669
1670 let output_schema = DFSchema::try_from(LogicalPlan::describe_schema()).unwrap();
1671
1672 Ok(LogicalPlan::DescribeTable(DescribeTable {
1673 schema,
1674 output_schema: Arc::new(output_schema),
1675 }))
1676 }
1677
1678 fn copy_to_plan(&self, statement: CopyToStatement) -> Result<LogicalPlan> {
1679 let copy_source = statement.source;
1681 let (input, input_schema, table_ref) = match copy_source {
1682 CopyToSource::Relation(object_name) => {
1683 let table_name = object_name_to_string(&object_name);
1684 let table_ref = self.object_name_to_table_reference(object_name)?;
1685 let table_source =
1686 self.context_provider.get_table_source(table_ref.clone())?;
1687 let plan =
1688 LogicalPlanBuilder::scan(table_name, table_source, None)?.build()?;
1689 let input_schema = Arc::clone(plan.schema());
1690 (plan, input_schema, Some(table_ref))
1691 }
1692 CopyToSource::Query(query) => {
1693 let plan = self.query_to_plan(*query, &mut PlannerContext::new())?;
1694 let input_schema = Arc::clone(plan.schema());
1695 (plan, input_schema, None)
1696 }
1697 };
1698
1699 let options_map = self.parse_options_map(statement.options, true)?;
1700
1701 let maybe_file_type = if let Some(stored_as) = &statement.stored_as {
1702 self.context_provider.get_file_type(stored_as).ok()
1703 } else {
1704 None
1705 };
1706
1707 let file_type = match maybe_file_type {
1708 Some(ft) => ft,
1709 None => {
1710 let e = || {
1711 DataFusionError::Configuration(
1712 "Format not explicitly set and unable to get file extension! Use STORED AS to define file format."
1713 .to_string(),
1714 )
1715 };
1716 let extension: &str = &Path::new(&statement.target)
1718 .extension()
1719 .ok_or_else(e)?
1720 .to_str()
1721 .ok_or_else(e)?
1722 .to_lowercase();
1723
1724 self.context_provider.get_file_type(extension)?
1725 }
1726 };
1727
1728 let partition_by = statement
1729 .partitioned_by
1730 .iter()
1731 .map(|col| input_schema.field_with_name(table_ref.as_ref(), col))
1732 .collect::<Result<Vec<_>>>()?
1733 .into_iter()
1734 .map(|f| f.name().to_owned())
1735 .collect();
1736
1737 Ok(LogicalPlan::Copy(CopyTo::new(
1738 Arc::new(input),
1739 statement.target,
1740 partition_by,
1741 file_type,
1742 options_map,
1743 )))
1744 }
1745
1746 fn build_order_by(
1747 &self,
1748 order_exprs: Vec<LexOrdering>,
1749 schema: &DFSchemaRef,
1750 planner_context: &mut PlannerContext,
1751 ) -> Result<Vec<Vec<SortExpr>>> {
1752 if !order_exprs.is_empty() && schema.fields().is_empty() {
1753 let results = order_exprs
1754 .iter()
1755 .map(|lex_order| {
1756 let result = lex_order
1757 .iter()
1758 .map(|order_by_expr| {
1759 let ordered_expr = &order_by_expr.expr;
1760 let ordered_expr = ordered_expr.to_owned();
1761 let ordered_expr = self.sql_expr_to_logical_expr(
1762 ordered_expr,
1763 schema,
1764 planner_context,
1765 )?;
1766 let asc = order_by_expr.options.asc.unwrap_or(true);
1767 let nulls_first =
1768 order_by_expr.options.nulls_first.unwrap_or_else(|| {
1769 self.options.default_null_ordering.nulls_first(asc)
1770 });
1771
1772 Ok(SortExpr::new(ordered_expr, asc, nulls_first))
1773 })
1774 .collect::<Result<Vec<SortExpr>>>()?;
1775 Ok(result)
1776 })
1777 .collect::<Result<Vec<Vec<SortExpr>>>>()?;
1778
1779 return Ok(results);
1780 }
1781
1782 let mut all_results = vec![];
1783 for expr in order_exprs {
1784 let expr_vec =
1786 self.order_by_to_sort_expr(expr, schema, planner_context, true, None)?;
1787 for sort in expr_vec.iter() {
1789 for column in sort.expr.column_refs().iter() {
1790 if !schema.has_column(column) {
1791 return plan_err!("Column {column} is not in schema");
1793 }
1794 }
1795 }
1796 all_results.push(expr_vec)
1798 }
1799 Ok(all_results)
1800 }
1801
1802 fn external_table_to_plan(
1804 &self,
1805 statement: CreateExternalTable,
1806 ) -> Result<LogicalPlan> {
1807 let definition = Some(statement.to_string());
1808 let CreateExternalTable {
1809 name,
1810 columns,
1811 file_type,
1812 locations,
1813 table_partition_cols,
1814 if_not_exists,
1815 temporary,
1816 order_exprs,
1817 unbounded,
1818 options,
1819 constraints,
1820 or_replace,
1821 } = statement;
1822
1823 let mut all_constraints = constraints;
1825 let inline_constraints = calc_inline_constraints_from_columns(&columns);
1826 all_constraints.extend(inline_constraints);
1827
1828 let options_map = self.parse_options_map(options, false)?;
1829
1830 let compression = options_map
1831 .get("format.compression")
1832 .map(|c| CompressionTypeVariant::from_str(c))
1833 .transpose()?;
1834 if (file_type == "PARQUET" || file_type == "AVRO" || file_type == "ARROW")
1835 && compression
1836 .map(|c| c != CompressionTypeVariant::UNCOMPRESSED)
1837 .unwrap_or(false)
1838 {
1839 plan_err!(
1840 "File compression type cannot be set for PARQUET, AVRO, or ARROW files."
1841 )?;
1842 }
1843
1844 let mut planner_context = PlannerContext::new();
1845
1846 let column_defaults = self
1847 .build_column_defaults(&columns, &mut planner_context)?
1848 .into_iter()
1849 .collect();
1850
1851 let schema = self.build_schema(columns)?;
1852 let df_schema = schema.to_dfschema_ref()?;
1853 df_schema.check_names()?;
1854
1855 let ordered_exprs =
1856 self.build_order_by(order_exprs, &df_schema, &mut planner_context)?;
1857
1858 let name = self.object_name_to_table_reference(name)?;
1859 let constraints =
1860 self.new_constraint_from_table_constraints(&all_constraints, &df_schema)?;
1861
1862 let Some(location) = locations.first().cloned() else {
1863 return plan_err!("CREATE EXTERNAL TABLE requires at least one location");
1864 };
1865
1866 Ok(LogicalPlan::Ddl(DdlStatement::CreateExternalTable(
1869 Box::new(
1870 PlanCreateExternalTable::builder(name, location, file_type, df_schema)
1871 .with_locations(locations)
1872 .with_partition_cols(table_partition_cols)
1873 .with_if_not_exists(if_not_exists)
1874 .with_or_replace(or_replace)
1875 .with_temporary(temporary)
1876 .with_definition(definition)
1877 .with_order_exprs(ordered_exprs)
1878 .with_unbounded(unbounded)
1879 .with_options(options_map)
1880 .with_constraints(constraints)
1881 .with_column_defaults(column_defaults)
1882 .build(),
1883 ),
1884 )))
1885 }
1886
1887 fn get_constraint_column_indices(
1890 &self,
1891 df_schema: &DFSchemaRef,
1892 columns: &[IndexColumn],
1893 constraint_name: &str,
1894 ) -> Result<Vec<usize>> {
1895 let field_names = df_schema.field_names();
1896 columns
1897 .iter()
1898 .map(|index_column| {
1899 let expr = &index_column.column.expr;
1900 let ident = if let SQLExpr::Identifier(ident) = expr {
1901 ident
1902 } else {
1903 return Err(plan_datafusion_err!(
1904 "Column name for {constraint_name} must be an identifier: {expr}"
1905 ));
1906 };
1907 let column = self.ident_normalizer.normalize(ident.clone());
1908 field_names
1909 .iter()
1910 .position(|item| *item == column)
1911 .ok_or_else(|| {
1912 plan_datafusion_err!(
1913 "Column for {constraint_name} not found in schema: {column}"
1914 )
1915 })
1916 })
1917 .collect::<Result<Vec<_>>>()
1918 }
1919
1920 pub fn new_constraint_from_table_constraints(
1922 &self,
1923 constraints: &[TableConstraint],
1924 df_schema: &DFSchemaRef,
1925 ) -> Result<Constraints> {
1926 let constraints = constraints
1927 .iter()
1928 .map(|c: &TableConstraint| match c {
1929 TableConstraint::Unique(UniqueConstraint {
1930 name,
1931 index_name: _,
1932 index_type_display: _,
1933 index_type: _,
1934 columns,
1935 index_options: _,
1936 characteristics: _,
1937 nulls_distinct: _,
1938 }) => {
1939 let constraint_name = match &name {
1940 Some(name) => &format!("unique constraint with name '{name}'"),
1941 None => "unique constraint",
1942 };
1943 let indices = self.get_constraint_column_indices(
1945 df_schema,
1946 columns,
1947 constraint_name,
1948 )?;
1949 Ok(Constraint::Unique(indices))
1950 }
1951 TableConstraint::PrimaryKey(PrimaryKeyConstraint {
1952 name: _,
1953 index_name: _,
1954 index_type: _,
1955 columns,
1956 index_options: _,
1957 characteristics: _,
1958 }) => {
1959 let indices = self.get_constraint_column_indices(
1961 df_schema,
1962 columns,
1963 "primary key",
1964 )?;
1965 Ok(Constraint::PrimaryKey(indices))
1966 }
1967 TableConstraint::ForeignKey { .. } => {
1968 _plan_err!("Foreign key constraints are not currently supported")
1969 }
1970 TableConstraint::Check { .. } => {
1971 _plan_err!("Check constraints are not currently supported")
1972 }
1973 TableConstraint::Index { .. } => {
1974 _plan_err!("Indexes are not currently supported")
1975 }
1976 TableConstraint::FulltextOrSpatial { .. } => {
1977 _plan_err!("Indexes are not currently supported")
1978 }
1979 TableConstraint::PrimaryKeyUsingIndex(_) => {
1980 _plan_err!(
1981 "PRIMARY KEY USING INDEX constraints are not currently supported"
1982 )
1983 }
1984 TableConstraint::UniqueUsingIndex(_) => {
1985 _plan_err!(
1986 "UNIQUE USING INDEX constraints are not currently supported"
1987 )
1988 }
1989 })
1990 .collect::<Result<Vec<_>>>()?;
1991 Ok(Constraints::new_unverified(constraints))
1992 }
1993
1994 fn parse_options_map(
1995 &self,
1996 options: Vec<(String, Value)>,
1997 allow_duplicates: bool,
1998 ) -> Result<HashMap<String, String>> {
1999 let mut options_map = HashMap::new();
2000 for (key, value) in options {
2001 if !allow_duplicates && options_map.contains_key(&key) {
2002 return plan_err!("Option {key} is specified multiple times");
2003 }
2004
2005 let Some(value_string) = crate::utils::value_to_string(&value) else {
2006 return plan_err!("Unsupported Value {}", value);
2007 };
2008
2009 if !(&key.contains('.')) {
2010 let renamed_key = format!("format.{key}");
2014 options_map.insert(renamed_key.to_lowercase(), value_string);
2015 } else {
2016 options_map.insert(key.to_lowercase(), value_string);
2017 }
2018 }
2019
2020 Ok(options_map)
2021 }
2022
2023 fn explain_to_plan(
2028 &self,
2029 opts: ExplainStatementOptions,
2030 statement: DFStatement,
2031 ) -> Result<LogicalPlan> {
2032 let plan = self.statement_to_plan(statement)?;
2033 if matches!(plan, LogicalPlan::Explain(_)) {
2034 return plan_err!("Nested EXPLAINs are not supported");
2035 }
2036
2037 let plan = Arc::new(plan);
2038 let schema = LogicalPlan::explain_schema();
2039 let schema = schema.to_dfschema_ref()?;
2040
2041 let ExplainStatementOptions {
2042 analyze,
2043 verbose,
2044 format,
2045 analyze_level,
2046 analyze_categories,
2047 show_statistics,
2048 } = opts;
2049
2050 if verbose && format.is_some() {
2052 return plan_err!("EXPLAIN VERBOSE with FORMAT is not supported");
2053 }
2054 if !analyze {
2055 if analyze_level.is_some() {
2056 return plan_err!("EXPLAIN option LEVEL requires ANALYZE");
2057 }
2058 if analyze_categories.is_some() {
2059 return plan_err!("EXPLAIN option METRICS requires ANALYZE");
2060 }
2061 }
2062 if analyze && show_statistics.is_some() {
2063 return plan_err!("EXPLAIN option COSTS cannot be combined with ANALYZE");
2064 }
2065
2066 let options = self.context_provider.options();
2072 let format = if verbose {
2073 ExplainFormat::Indent
2074 } else if let Some(format) = format {
2075 format
2076 } else if analyze {
2077 ExplainFormat::Indent
2078 } else {
2079 options.explain.format.clone()
2080 };
2081
2082 if analyze {
2083 match &format {
2084 ExplainFormat::Indent => {}
2085 ExplainFormat::PostgresJSON => {
2086 if options.explain.show_statistics {
2089 return plan_err!(
2090 "EXPLAIN ANALYZE with FORMAT pgjson does not support show_statistics"
2091 );
2092 }
2093 }
2094 ExplainFormat::Tree | ExplainFormat::Graphviz => {
2095 return plan_err!(
2096 "EXPLAIN ANALYZE with FORMAT {format} is not supported"
2097 );
2098 }
2099 }
2100 Ok(LogicalPlan::Analyze(Analyze {
2101 verbose,
2102 format,
2103 input: plan,
2104 schema,
2105 analyze_level,
2106 analyze_categories,
2107 }))
2108 } else {
2109 let stringified_plans =
2110 vec![plan.to_stringified(PlanType::InitialLogicalPlan)];
2111
2112 Ok(LogicalPlan::Explain(Explain {
2113 verbose,
2114 explain_format: format,
2115 plan,
2116 stringified_plans,
2117 schema,
2118 logical_optimization_succeeded: false,
2119 show_statistics,
2120 }))
2121 }
2122 }
2123
2124 fn show_variable_to_plan(&self, variable: &[Ident]) -> Result<LogicalPlan> {
2125 if !self.has_table("information_schema", "df_settings") {
2126 return plan_err!(
2127 "SHOW [VARIABLE] is not supported unless information_schema is enabled"
2128 );
2129 }
2130
2131 let verbose = variable
2132 .last()
2133 .map(|s| ident_to_string(s) == "verbose")
2134 .unwrap_or(false);
2135 let mut variable_vec = variable.to_vec();
2136 let mut columns: String = "name, value".to_owned();
2137
2138 if verbose {
2139 columns = format!("{columns}, description");
2140 variable_vec = variable_vec.split_at(variable_vec.len() - 1).0.to_vec();
2141 }
2142
2143 let variable = object_name_to_string(&ObjectName::from(variable_vec));
2144 let base_query = format!("SELECT {columns} FROM information_schema.df_settings");
2145 let query = if variable == "all" {
2146 format!("{base_query} ORDER BY name")
2148 } else if variable == "timezone" || variable == "time.zone" {
2149 format!("{base_query} WHERE name = 'datafusion.execution.time_zone'")
2151 } else {
2152 let is_valid_variable = self
2156 .context_provider
2157 .options()
2158 .entries()
2159 .iter()
2160 .any(|opt| opt.key == variable);
2161
2162 let is_runtime_variable = variable.starts_with("datafusion.runtime.");
2164
2165 if !is_valid_variable && !is_runtime_variable {
2166 return plan_err!(
2167 "'{variable}' is not a variable which can be viewed with 'SHOW'"
2168 );
2169 }
2170
2171 format!("{base_query} WHERE name = '{variable}'")
2172 };
2173
2174 let mut rewrite = DFParser::parse_sql(&query)?;
2175 assert_eq!(rewrite.len(), 1);
2176
2177 self.statement_to_plan(rewrite.pop_front().unwrap())
2178 }
2179
2180 fn set_statement_to_plan(&self, statement: Set) -> Result<LogicalPlan> {
2181 match statement {
2182 Set::SingleAssignment {
2183 scope,
2184 hivevar,
2185 variable,
2186 values,
2187 } => {
2188 if scope.is_some() {
2189 return not_impl_err!("SET with scope modifiers is not supported");
2190 }
2191
2192 if hivevar {
2193 return not_impl_err!("SET HIVEVAR is not supported");
2194 }
2195
2196 let variable = object_name_to_string(&variable);
2197 let mut variable_lower = variable.to_lowercase();
2198
2199 if variable_lower == "timezone" || variable_lower == "time.zone" {
2201 variable_lower = "datafusion.execution.time_zone".to_string();
2202 }
2203
2204 if values.len() != 1 {
2205 return plan_err!("SET only supports single value assignment");
2206 }
2207
2208 let value_string = match &values[0] {
2209 SQLExpr::Identifier(i) => ident_to_string(i),
2210 SQLExpr::Value(v) => match crate::utils::value_to_string(&v.value) {
2211 None => {
2212 return plan_err!("Unsupported value {:?}", v.value);
2213 }
2214 Some(s) => s,
2215 },
2216 SQLExpr::UnaryOp { op, expr } => match op {
2217 UnaryOperator::Plus => format!("+{expr}"),
2218 UnaryOperator::Minus => format!("-{expr}"),
2219 _ => return plan_err!("Unsupported unary op {:?}", op),
2220 },
2221 _ => return plan_err!("Unsupported expr {:?}", values[0]),
2222 };
2223
2224 Ok(LogicalPlan::Statement(PlanStatement::SetVariable(
2225 SetVariable {
2226 variable: variable_lower,
2227 value: value_string,
2228 },
2229 )))
2230 }
2231 other => not_impl_err!("SET variant not implemented yet: {other:?}"),
2232 }
2233 }
2234
2235 fn reset_statement_to_plan(&self, statement: ResetStatement) -> Result<LogicalPlan> {
2236 match statement {
2237 ResetStatement::Variable(variable) => {
2238 let variable = object_name_to_string(&variable);
2239 let mut variable_lower = variable.to_lowercase();
2240
2241 if variable_lower == "timezone" || variable_lower == "time.zone" {
2243 variable_lower = "datafusion.execution.time_zone".to_string();
2244 }
2245
2246 Ok(LogicalPlan::Statement(PlanStatement::ResetVariable(
2247 ResetVariable {
2248 variable: variable_lower,
2249 },
2250 )))
2251 }
2252 }
2253 }
2254
2255 fn delete_to_plan(
2256 &self,
2257 table_name: &ObjectName,
2258 predicate_expr: Option<SQLExpr>,
2259 limit: Option<SQLExpr>,
2260 ) -> Result<LogicalPlan> {
2261 let table_ref = self.object_name_to_table_reference(table_name.clone())?;
2263 let table_source = self.context_provider.get_table_source(table_ref.clone())?;
2264 let schema = DFSchema::try_from_qualified_schema(
2265 table_ref.clone(),
2266 &table_source.schema(),
2267 )?;
2268 let scan =
2269 LogicalPlanBuilder::scan(table_ref.clone(), Arc::clone(&table_source), None)?
2270 .build()?;
2271 let mut planner_context = PlannerContext::new();
2272
2273 let mut source = match predicate_expr {
2274 None => scan,
2275 Some(predicate_expr) => {
2276 let filter_expr =
2277 self.sql_to_expr(predicate_expr, &schema, &mut planner_context)?;
2278 let schema = Arc::new(schema);
2279 let mut using_columns = HashSet::new();
2280 expr_to_columns(&filter_expr, &mut using_columns)?;
2281 let filter_expr = normalize_col_with_schemas_and_ambiguity_check(
2282 filter_expr,
2283 &[&[&schema]],
2284 &[using_columns],
2285 )?;
2286 LogicalPlan::Filter(Filter::try_new(filter_expr, Arc::new(scan))?)
2287 }
2288 };
2289
2290 if let Some(limit) = limit {
2291 let empty_schema = DFSchema::empty();
2292 let limit = self.sql_to_expr(limit, &empty_schema, &mut planner_context)?;
2293 source = LogicalPlanBuilder::from(source)
2294 .limit_by_expr(None, Some(limit))?
2295 .build()?
2296 }
2297
2298 let plan = LogicalPlan::Dml(DmlStatement::new(
2299 table_ref,
2300 table_source,
2301 WriteOp::Delete,
2302 Arc::new(source),
2303 ));
2304 Ok(plan)
2305 }
2306
2307 fn update_to_plan(
2308 &self,
2309 table: TableWithJoins,
2310 assignments: &[Assignment],
2311 from: Option<TableWithJoins>,
2312 predicate_expr: Option<SQLExpr>,
2313 ) -> Result<LogicalPlan> {
2314 let (table_name, table_alias) = match &table.relation {
2315 TableFactor::Table { name, alias, .. } => (name.clone(), alias.clone()),
2316 _ => plan_err!("Cannot update non-table relation!")?,
2317 };
2318
2319 let table_name = self.object_name_to_table_reference(table_name)?;
2321 let table_source = self.context_provider.get_table_source(table_name.clone())?;
2322 let table_schema = Arc::new(DFSchema::try_from_qualified_schema(
2323 table_name.clone(),
2324 &table_source.schema(),
2325 )?);
2326
2327 let mut planner_context = PlannerContext::new();
2329 let mut assign_map = assignments
2330 .iter()
2331 .map(|assign| {
2332 let cols = match &assign.target {
2333 AssignmentTarget::ColumnName(cols) => cols,
2334 _ => plan_err!("Tuples are not supported")?,
2335 };
2336 let col_name: &Ident = cols
2337 .0
2338 .iter()
2339 .last()
2340 .ok_or_else(|| plan_datafusion_err!("Empty column id"))?
2341 .as_ident()
2342 .unwrap();
2343 table_schema.field_with_unqualified_name(&col_name.value)?;
2345 Ok((col_name.value.clone(), assign.value.clone()))
2346 })
2347 .collect::<Result<HashMap<String, SQLExpr>>>()?;
2348
2349 let mut input_tables = vec![table];
2351 input_tables.extend(from);
2352 let scan = self.plan_from_tables(input_tables, &mut planner_context)?;
2353
2354 let source = match predicate_expr {
2356 None => scan,
2357 Some(predicate_expr) => {
2358 let filter_expr = self.sql_to_expr(
2359 predicate_expr,
2360 scan.schema(),
2361 &mut planner_context,
2362 )?;
2363 let mut using_columns = HashSet::new();
2364 expr_to_columns(&filter_expr, &mut using_columns)?;
2365 let filter_expr = normalize_col_with_schemas_and_ambiguity_check(
2366 filter_expr,
2367 &[&[scan.schema()]],
2368 &[using_columns],
2369 )?;
2370 LogicalPlan::Filter(Filter::try_new(filter_expr, Arc::new(scan))?)
2371 }
2372 };
2373
2374 let exprs = table_schema
2376 .iter()
2377 .map(|(qualifier, field)| {
2378 let expr = match assign_map.remove(field.name()) {
2379 Some(new_value) => {
2380 let mut expr = self.sql_to_expr(
2381 new_value,
2382 source.schema(),
2383 &mut planner_context,
2384 )?;
2385 if let Expr::Placeholder(placeholder) = &mut expr {
2387 placeholder.field = placeholder
2388 .field
2389 .take()
2390 .or_else(|| Some(Arc::clone(field)));
2391 }
2392 expr.cast_to(field.data_type(), source.schema())?
2394 }
2395 None => {
2396 if let Some(alias) = &table_alias {
2398 Expr::Column(Column::new(
2399 Some(self.ident_normalizer.normalize(alias.name.clone())),
2400 field.name(),
2401 ))
2402 } else {
2403 Expr::Column(Column::from((qualifier, field)))
2404 }
2405 }
2406 };
2407 Ok(expr.alias(field.name()))
2408 })
2409 .collect::<Result<Vec<_>>>()?;
2410
2411 let source = project(source, exprs)?;
2412
2413 let plan = LogicalPlan::Dml(DmlStatement::new(
2414 table_name,
2415 table_source,
2416 WriteOp::Update,
2417 Arc::new(source),
2418 ));
2419 Ok(plan)
2420 }
2421
2422 fn merge_to_plan(&self, merge: ast::Merge) -> Result<LogicalPlan> {
2423 let ast::Merge {
2424 table,
2425 source,
2426 on,
2427 clauses,
2428 into: _,
2429 merge_token: _,
2430 optimizer_hints,
2431 output,
2432 } = merge;
2433
2434 if !optimizer_hints.is_empty() {
2435 plan_err!("Optimizer hints not supported")?;
2436 }
2437
2438 if output.is_some() {
2439 return not_impl_err!("MERGE OUTPUT clause is not supported");
2440 }
2441
2442 if clauses.is_empty() {
2443 return plan_err!("MERGE INTO requires at least one WHEN clause");
2444 }
2445
2446 let (target_table_name, target_alias) = match table {
2448 TableFactor::Table {
2449 name,
2450 alias,
2451 args,
2452 with_hints,
2453 version,
2454 with_ordinality,
2455 partitions,
2456 json_path,
2457 sample,
2458 index_hints,
2459 } => {
2460 if alias
2461 .as_ref()
2462 .is_some_and(|alias| !alias.columns.is_empty())
2463 {
2464 return not_impl_err!(
2465 "MERGE target alias column lists are not supported"
2466 );
2467 }
2468 if args.is_some()
2469 || !with_hints.is_empty()
2470 || version.is_some()
2471 || with_ordinality
2472 || !partitions.is_empty()
2473 || json_path.is_some()
2474 || sample.is_some()
2475 || !index_hints.is_empty()
2476 {
2477 return not_impl_err!(
2478 "MERGE target table modifiers are not supported"
2479 );
2480 }
2481 (name, alias)
2482 }
2483 _ => plan_err!("Cannot MERGE INTO non-table relation!")?,
2484 };
2485 let target_table_ref = self.object_name_to_table_reference(target_table_name)?;
2486 let target_table_source = self
2487 .context_provider
2488 .get_table_source(target_table_ref.clone())?;
2489 let target_qualifier = target_alias
2492 .as_ref()
2493 .map(|a| {
2494 TableReference::bare(self.ident_normalizer.normalize(a.name.clone()))
2495 })
2496 .unwrap_or_else(|| target_table_ref.clone());
2497 let target_schema = Arc::new(DFSchema::try_from_qualified_schema(
2498 target_qualifier.clone(),
2499 &target_table_source.schema(),
2500 )?);
2501
2502 let mut planner_context = PlannerContext::new();
2504 let source_table_with_joins = TableWithJoins {
2505 relation: source,
2506 joins: vec![],
2507 };
2508 let source_plan =
2509 self.plan_from_tables(vec![source_table_with_joins], &mut planner_context)?;
2510
2511 let combined_schema =
2513 Arc::new(target_schema.as_ref().join(source_plan.schema())?);
2514
2515 let on_expr = self.sql_to_expr(*on, &combined_schema, &mut planner_context)?;
2517
2518 let df_clauses = clauses
2520 .into_iter()
2521 .map(|clause| {
2522 self.merge_clause_to_plan(
2523 clause,
2524 &combined_schema,
2525 &target_schema,
2526 &target_qualifier,
2527 &mut planner_context,
2528 )
2529 })
2530 .collect::<Result<Vec<_>>>()?;
2531
2532 let mut merge_op = MergeIntoOp {
2539 on: on_expr,
2540 clauses: df_clauses,
2541 };
2542 if target_qualifier != target_table_ref {
2543 for expr in merge_op.exprs() {
2550 if Self::has_outer_reference_to_qualifier(expr, &target_qualifier)? {
2551 return not_impl_err!(
2552 "MERGE subqueries correlated to target alias \
2553 '{target_qualifier}' are not supported"
2554 );
2555 }
2556 }
2557
2558 if source_plan.schema().iter().any(|(qualifier, _)| {
2565 qualifier.is_some_and(|q| q.resolved_eq(&target_table_ref))
2566 }) {
2567 return plan_err!(
2568 "MERGE source may not use the target table name '{target_table_ref}' \
2569 as a qualifier while the target is aliased as '{target_qualifier}'; \
2570 use a different source alias"
2571 );
2572 }
2573 let canonical = merge_op
2574 .exprs()
2575 .into_iter()
2576 .cloned()
2577 .map(|expr| {
2578 Self::canonicalize_target_qualifier(
2579 expr,
2580 &target_qualifier,
2581 &target_table_ref,
2582 )
2583 })
2584 .collect::<Result<Vec<_>>>()?;
2585 merge_op = merge_op.with_new_exprs(canonical)?;
2586 }
2587
2588 Ok(LogicalPlan::Dml(DmlStatement::new(
2589 target_table_ref,
2590 target_table_source,
2591 WriteOp::MergeInto(Box::new(merge_op)),
2592 Arc::new(source_plan),
2593 )))
2594 }
2595
2596 fn canonicalize_target_qualifier(
2600 expr: Expr,
2601 from: &TableReference,
2602 to: &TableReference,
2603 ) -> Result<Expr> {
2604 expr.transform(|expr| match expr {
2605 Expr::Column(col) if col.relation.as_ref() == Some(from) => Ok(
2606 Transformed::yes(Expr::Column(Column::new(Some(to.clone()), col.name))),
2607 ),
2608 other => Ok(Transformed::no(other)),
2609 })
2610 .map(|transformed| transformed.data)
2611 }
2612
2613 fn has_outer_reference_to_qualifier(
2616 expr: &Expr,
2617 qualifier: &TableReference,
2618 ) -> Result<bool> {
2619 let mut found = false;
2620 expr.apply(|expr| {
2621 let subquery = match expr {
2622 Expr::Exists(exists) => Some(&exists.subquery),
2623 Expr::InSubquery(in_subquery) => Some(&in_subquery.subquery),
2624 Expr::SetComparison(set_comparison) => Some(&set_comparison.subquery),
2625 Expr::ScalarSubquery(subquery) => Some(subquery),
2626 _ => None,
2627 };
2628
2629 if let Some(subquery) = subquery {
2630 subquery.subquery.apply_with_subqueries(|plan| {
2631 plan.apply_expressions(|expr| {
2632 expr.apply(|expr| {
2633 if let Expr::OuterReferenceColumn(_, column) = expr
2634 && column.relation.as_ref() == Some(qualifier)
2635 {
2636 found = true;
2637 Ok(TreeNodeRecursion::Stop)
2638 } else {
2639 Ok(TreeNodeRecursion::Continue)
2640 }
2641 })
2642 })?;
2643 Ok(if found {
2644 TreeNodeRecursion::Stop
2645 } else {
2646 TreeNodeRecursion::Continue
2647 })
2648 })?;
2649 }
2650
2651 Ok(if found {
2652 TreeNodeRecursion::Stop
2653 } else {
2654 TreeNodeRecursion::Continue
2655 })
2656 })?;
2657 Ok(found)
2658 }
2659
2660 fn merge_target_column_name(
2661 &self,
2662 name: &ObjectName,
2663 target_qualifier: &TableReference,
2664 ) -> Result<String> {
2665 let part = name
2666 .0
2667 .iter()
2668 .last()
2669 .ok_or_else(|| plan_datafusion_err!("Empty column name"))?;
2670 let ident = part
2671 .as_ident()
2672 .cloned()
2673 .ok_or_else(|| plan_datafusion_err!("Expected simple identifier"))?;
2674
2675 if name.0.len() > 1 {
2676 let qualifier = self.object_name_to_table_reference(ObjectName(
2677 name.0[..name.0.len() - 1].to_vec(),
2678 ))?;
2679 if !qualifier.resolved_eq(target_qualifier) {
2680 return plan_err!(
2681 "MERGE assignment target '{name}' must reference target table \
2682 '{target_qualifier}'"
2683 );
2684 }
2685 }
2686
2687 Ok(self.ident_normalizer.normalize(ident))
2688 }
2689
2690 fn merge_clause_to_plan(
2691 &self,
2692 clause: ast::MergeClause,
2693 combined_schema: &DFSchema,
2694 target_schema: &DFSchema,
2695 target_qualifier: &TableReference,
2696 planner_context: &mut PlannerContext,
2697 ) -> Result<MergeIntoClause> {
2698 let kind = match clause.clause_kind {
2699 ast::MergeClauseKind::Matched => MergeIntoClauseKind::Matched,
2700 ast::MergeClauseKind::NotMatched => MergeIntoClauseKind::NotMatched,
2701 ast::MergeClauseKind::NotMatchedByTarget => {
2702 MergeIntoClauseKind::NotMatchedByTarget
2703 }
2704 ast::MergeClauseKind::NotMatchedBySource => {
2705 MergeIntoClauseKind::NotMatchedBySource
2706 }
2707 };
2708
2709 let predicate = clause
2710 .predicate
2711 .map(|p| self.sql_to_expr(p, combined_schema, planner_context))
2712 .transpose()?;
2713
2714 let action = match clause.action {
2715 ast::MergeAction::Update(update_expr) => {
2716 if update_expr.update_predicate.is_some() {
2717 return not_impl_err!(
2718 "MERGE UPDATE WHERE predicates are not supported"
2719 );
2720 }
2721 if update_expr.delete_predicate.is_some() {
2722 return not_impl_err!(
2723 "MERGE UPDATE DELETE WHERE predicates are not supported"
2724 );
2725 }
2726 let assignments = update_expr
2727 .assignments
2728 .into_iter()
2729 .map(|assign| {
2730 let col_name = match &assign.target {
2731 AssignmentTarget::ColumnName(cols) => {
2732 self.merge_target_column_name(cols, target_qualifier)?
2733 }
2734 _ => plan_err!("Tuples are not supported")?,
2735 };
2736 target_schema.field_with_unqualified_name(&col_name)?;
2738 let value = self.sql_to_expr(
2739 assign.value,
2740 combined_schema,
2741 planner_context,
2742 )?;
2743 Ok((col_name, value))
2744 })
2745 .collect::<Result<Vec<_>>>()?;
2746 let mut seen = HashSet::new();
2747 for (column, _) in &assignments {
2748 if !seen.insert(column.as_str()) {
2749 return plan_err!("Duplicate column '{column}' in MERGE UPDATE");
2750 }
2751 }
2752 MergeIntoAction::Update(assignments)
2753 }
2754 ast::MergeAction::Insert(insert_expr) => {
2755 if insert_expr.insert_predicate.is_some() {
2756 return not_impl_err!(
2757 "MERGE INSERT WHERE predicates are not supported"
2758 );
2759 }
2760 let columns: Vec<String> = insert_expr
2761 .columns
2762 .iter()
2763 .map(|c| self.merge_target_column_name(c, target_qualifier))
2764 .collect::<Result<Vec<_>>>()?;
2765
2766 let mut seen = HashSet::new();
2768 for col in &columns {
2769 if !seen.insert(col.as_str()) {
2770 return plan_err!("Duplicate column '{col}' in MERGE INSERT");
2771 }
2772 target_schema.field_with_unqualified_name(col)?;
2773 }
2774
2775 let num_target_cols = target_schema.fields().len();
2776
2777 let values = match insert_expr.kind {
2778 ast::MergeInsertKind::Values(values) => {
2779 if values.rows.len() != 1 {
2780 return plan_err!(
2781 "MERGE INSERT must have exactly one row of values"
2782 );
2783 }
2784 let row = values.rows.into_iter().next().unwrap().content;
2785 let expected = if columns.is_empty() {
2786 num_target_cols
2787 } else {
2788 columns.len()
2789 };
2790 if row.len() != expected {
2791 return plan_err!(
2792 "MERGE INSERT has {expected} column(s) but {} value(s)",
2793 row.len()
2794 );
2795 }
2796 row.into_iter()
2797 .map(|v| {
2798 self.sql_to_expr(v, combined_schema, planner_context)
2799 })
2800 .collect::<Result<Vec<_>>>()?
2801 }
2802 ast::MergeInsertKind::Row => {
2803 return not_impl_err!("MERGE INSERT ROW is not supported");
2804 }
2805 };
2806
2807 MergeIntoAction::Insert { columns, values }
2808 }
2809 ast::MergeAction::Delete { .. } => MergeIntoAction::Delete,
2810 };
2811
2812 Ok(MergeIntoClause {
2813 kind,
2814 predicate,
2815 action,
2816 })
2817 }
2818
2819 fn insert_to_plan(
2820 &self,
2821 table_name: ObjectName,
2822 columns: Vec<ObjectName>,
2823 source: Box<Query>,
2824 overwrite: bool,
2825 replace_into: bool,
2826 ) -> Result<LogicalPlan> {
2827 let table_name = self.object_name_to_table_reference(table_name)?;
2829 let table_source = self.context_provider.get_table_source(table_name.clone())?;
2830 let table_schema = DFSchema::try_from(table_source.schema())?;
2831
2832 let columns: Vec<Ident> = columns
2833 .into_iter()
2834 .map(|name| {
2835 if name.0.len() != 1 {
2836 return not_impl_err!(
2837 "Multi-part column names in INSERT not supported: {name}"
2838 );
2839 }
2840 let part = &name.0[0];
2841 let Some(ident) = part.as_ident() else {
2842 return not_impl_err!(
2843 "Non-identifier column name part in INSERT not supported: {part}"
2844 );
2845 };
2846 Ok(ident.clone())
2847 })
2848 .collect::<Result<Vec<_>>>()?;
2849
2850 let (fields, value_indices) = if columns.is_empty() {
2858 (
2860 table_schema.fields().clone(),
2861 (0..table_schema.fields().len())
2862 .map(Some)
2863 .collect::<Vec<_>>(),
2864 )
2865 } else {
2866 let mut value_indices = vec![None; table_schema.fields().len()];
2867 let fields = columns
2868 .into_iter()
2869 .enumerate()
2870 .map(|(i, c)| {
2871 let c = self.ident_normalizer.normalize(c);
2872 let column_index = table_schema
2873 .index_of_column_by_name(None, &c)
2874 .ok_or_else(|| unqualified_field_not_found(&c, &table_schema))?;
2875
2876 if value_indices[column_index].is_some() {
2877 return schema_err!(SchemaError::DuplicateUnqualifiedField {
2878 name: c,
2879 });
2880 } else {
2881 value_indices[column_index] = Some(i);
2882 }
2883 Ok(Arc::clone(table_schema.field(column_index)))
2884 })
2885 .collect::<Result<Vec<_>>>()?;
2886 (Fields::from(fields), value_indices)
2887 };
2888
2889 let mut prepare_param_data_types = BTreeMap::new();
2891 if let SetExpr::Values(ast::Values { rows, .. }) = (*source.body).clone() {
2892 for row in rows.iter() {
2893 for (idx, val) in row.content.iter().enumerate() {
2894 if let SQLExpr::Value(ValueWithSpan {
2895 value: Value::Placeholder(name),
2896 span: _,
2897 }) = val
2898 {
2899 let index = match name[1..].parse::<usize>().map_err(|_| {
2900 plan_datafusion_err!("Can't parse placeholder: {name}")
2901 })? {
2902 0 => {
2903 return plan_err!(
2904 "Invalid placeholder, zero is not a valid index: {name}"
2905 );
2906 }
2907 index => index - 1,
2908 };
2909 let field = fields.get(idx).ok_or_else(|| {
2910 plan_datafusion_err!(
2911 "Placeholder ${} refers to a non existent column",
2912 idx + 1
2913 )
2914 })?;
2915 let _ = prepare_param_data_types.insert(index, Arc::clone(field));
2916 }
2917 }
2918 }
2919 }
2920 let prepare_param_data_types = {
2921 let len = prepare_param_data_types.keys().last().map_or(0, |&k| k + 1);
2922 (0..len)
2923 .map(|i| prepare_param_data_types.remove(&i))
2924 .collect()
2925 };
2926
2927 let mut planner_context =
2929 PlannerContext::new().with_prepare_param_data_types(prepare_param_data_types);
2930 planner_context.set_table_schema(Some(DFSchemaRef::new(
2931 DFSchema::from_unqualified_fields(fields.clone(), Default::default())?,
2932 )));
2933 let source = self.query_to_plan(*source, &mut planner_context)?;
2934 if fields.len() != source.schema().fields().len() {
2935 plan_err!("Column count doesn't match insert query!")?;
2936 }
2937
2938 let exprs = value_indices
2939 .into_iter()
2940 .enumerate()
2941 .map(|(i, value_index)| {
2942 let target_field = table_schema.field(i);
2943 let expr = match value_index {
2944 Some(v) => {
2945 Expr::Column(Column::from(source.schema().qualified_field(v)))
2946 .cast_to(target_field.data_type(), source.schema())?
2947 }
2948 None => table_source
2950 .get_column_default(target_field.name())
2951 .cloned()
2952 .unwrap_or_else(|| {
2953 Expr::Literal(ScalarValue::Null, None)
2955 })
2956 .cast_to(target_field.data_type(), &DFSchema::empty())?,
2957 };
2958 Ok(expr.alias(target_field.name()))
2959 })
2960 .collect::<Result<Vec<Expr>>>()?;
2961 let source = project(source, exprs)?;
2962
2963 let insert_op = match (overwrite, replace_into) {
2964 (false, false) => InsertOp::Append,
2965 (true, false) => InsertOp::Overwrite,
2966 (false, true) => InsertOp::Replace,
2967 (true, true) => plan_err!(
2968 "Conflicting insert operations: `overwrite` and `replace_into` cannot both be true"
2969 )?,
2970 };
2971
2972 let plan = LogicalPlan::Dml(DmlStatement::new(
2973 table_name,
2974 Arc::clone(&table_source),
2975 WriteOp::Insert(insert_op),
2976 Arc::new(source),
2977 ));
2978 Ok(plan)
2979 }
2980
2981 fn show_columns_to_plan(
2982 &self,
2983 extended: bool,
2984 full: bool,
2985 sql_table_name: ObjectName,
2986 ) -> Result<LogicalPlan> {
2987 let where_clause = object_name_to_qualifier(
2989 &sql_table_name,
2990 self.options.enable_ident_normalization,
2991 )?;
2992
2993 if !self.has_table("information_schema", "columns") {
2994 return plan_err!(
2995 "SHOW COLUMNS is not supported unless information_schema is enabled"
2996 );
2997 }
2998
2999 let table_ref = self.object_name_to_table_reference(sql_table_name)?;
3001 let _ = self.context_provider.get_table_source(table_ref)?;
3002
3003 let select_list = if full || extended {
3005 "*"
3006 } else {
3007 "table_catalog, table_schema, table_name, column_name, data_type, is_nullable"
3008 };
3009
3010 let query = format!(
3011 "SELECT {select_list} FROM information_schema.columns WHERE {where_clause}"
3012 );
3013
3014 let mut rewrite = DFParser::parse_sql(&query)?;
3015 assert_eq!(rewrite.len(), 1);
3016 self.statement_to_plan(rewrite.pop_front().unwrap()) }
3018
3019 fn show_functions_to_plan(
3030 &self,
3031 filter: Option<ShowStatementFilter>,
3032 ) -> Result<LogicalPlan> {
3033 let where_clause = if let Some(filter) = filter {
3034 match filter {
3035 ShowStatementFilter::Like(like) => {
3036 format!("WHERE p.function_name like '{like}'")
3037 }
3038 _ => return plan_err!("Unsupported SHOW FUNCTIONS filter"),
3039 }
3040 } else {
3041 "".to_string()
3042 };
3043
3044 let where_clause = where_clause.replace("p.function_name", "sc.function_name");
3051 let query = format!(
3052 r#"
3053SELECT DISTINCT
3054 sc.function_name,
3055 sc.return_type,
3056 sc.parameters,
3057 sc.parameter_types,
3058 sc.function_type,
3059 sc.description,
3060 sc.syntax_example
3061FROM (
3062 SELECT
3063 p.function_name,
3064 p.return_type,
3065 p.parameters,
3066 p.parameter_types,
3067 r.function_type function_type,
3068 r.description description,
3069 r.syntax_example syntax_example
3070 FROM (
3071 SELECT
3072 o.specific_name function_name,
3073 o.data_type return_type,
3074 array_agg(i.parameter_name ORDER BY i.ordinal_position ASC) parameters,
3075 array_agg(i.data_type ORDER BY i.ordinal_position ASC) parameter_types
3076 FROM (
3077 SELECT
3078 specific_catalog,
3079 specific_schema,
3080 specific_name,
3081 ordinal_position,
3082 parameter_name,
3083 data_type,
3084 rid
3085 FROM
3086 information_schema.parameters
3087 WHERE
3088 parameter_mode = 'OUT'
3089 ) o
3090 LEFT JOIN
3091 (
3092 SELECT
3093 specific_catalog,
3094 specific_schema,
3095 specific_name,
3096 ordinal_position,
3097 parameter_name,
3098 data_type,
3099 rid
3100 FROM
3101 information_schema.parameters
3102 WHERE
3103 parameter_mode = 'IN'
3104 ) i
3105 ON i.specific_catalog = o.specific_catalog
3106 AND i.specific_schema = o.specific_schema
3107 AND i.specific_name = o.specific_name
3108 AND i.rid = o.rid
3109 GROUP BY 1, 2, o.rid
3110 ) as p
3111 JOIN information_schema.routines r
3112 ON p.function_name = r.routine_name
3113 AND r.function_type <> 'TABLE'
3114
3115 UNION ALL
3116
3117 SELECT
3118 routine_name function_name,
3119 data_type return_type,
3120 array_agg(NULL) FILTER (WHERE FALSE) parameters,
3121 array_agg(NULL) FILTER (WHERE FALSE) parameter_types,
3122 function_type,
3123 description,
3124 syntax_example
3125 FROM information_schema.routines
3126 WHERE function_type = 'TABLE'
3127 GROUP BY routine_name, data_type, function_type, description, syntax_example
3128) sc
3129{where_clause}
3130 "#
3131 );
3132 let mut rewrite = DFParser::parse_sql(&query)?;
3133 assert_eq!(rewrite.len(), 1);
3134 self.statement_to_plan(rewrite.pop_front().unwrap()) }
3136
3137 fn show_create_table_to_plan(
3138 &self,
3139 sql_table_name: ObjectName,
3140 ) -> Result<LogicalPlan> {
3141 if !self.has_table("information_schema", "tables") {
3142 return plan_err!(
3143 "SHOW CREATE TABLE is not supported unless information_schema is enabled"
3144 );
3145 }
3146 let where_clause = object_name_to_qualifier(
3148 &sql_table_name,
3149 self.options.enable_ident_normalization,
3150 )?;
3151
3152 let table_ref = self.object_name_to_table_reference(sql_table_name)?;
3154 let _ = self.context_provider.get_table_source(table_ref)?;
3155
3156 let query = format!(
3157 "SELECT table_catalog, table_schema, table_name, definition FROM information_schema.views WHERE {where_clause}"
3158 );
3159
3160 let mut rewrite = DFParser::parse_sql(&query)?;
3161 assert_eq!(rewrite.len(), 1);
3162 self.statement_to_plan(rewrite.pop_front().unwrap()) }
3164
3165 fn has_table(&self, schema: &str, table: &str) -> bool {
3167 let tables_reference = TableReference::Partial {
3168 schema: schema.into(),
3169 table: table.into(),
3170 };
3171 self.context_provider
3172 .get_table_source(tables_reference)
3173 .is_ok()
3174 }
3175
3176 fn validate_transaction_kind(
3177 &self,
3178 kind: Option<&BeginTransactionKind>,
3179 ) -> Result<()> {
3180 match kind {
3181 None => Ok(()),
3183 Some(BeginTransactionKind::Transaction) => Ok(()),
3185 Some(BeginTransactionKind::Work) | Some(BeginTransactionKind::Tran) => {
3186 not_impl_err!("Transaction kind not supported: {kind:?}")
3187 }
3188 }
3189 }
3190}