(
&self,
catalog: &Arc<dyn Catalog>,
ct: &CreateTable,
partition_keys: Vec<String>,
)
| 589 | } |
| 590 | |
| 591 | async fn handle_create_table( |
| 592 | &self, |
| 593 | catalog: &Arc<dyn Catalog>, |
| 594 | ct: &CreateTable, |
| 595 | partition_keys: Vec<String>, |
| 596 | ) -> DFResult<DataFrame> { |
| 597 | if ct.external { |
| 598 | return Err(DataFusionError::Plan( |
| 599 | "CREATE EXTERNAL TABLE is not supported. Use CREATE TABLE instead.".to_string(), |
| 600 | )); |
| 601 | } |
| 602 | if ct.location.is_some() { |
| 603 | return Err(DataFusionError::Plan( |
| 604 | "LOCATION is not supported for Paimon tables. Table path is determined by the catalog warehouse.".to_string(), |
| 605 | )); |
| 606 | } |
| 607 | if ct.query.is_some() { |
| 608 | return Err(DataFusionError::Plan( |
| 609 | "CREATE TABLE AS SELECT is not yet supported for Paimon tables.".to_string(), |
| 610 | )); |
| 611 | } |
| 612 | |
| 613 | let identifier = self.resolve_table_name(&ct.name)?; |
| 614 | |
| 615 | let mut builder = paimon::spec::Schema::builder(); |
| 616 | |
| 617 | // Columns |
| 618 | for col in &ct.columns { |
| 619 | let paimon_type = column_def_to_paimon_type(col)?; |
| 620 | builder = builder.column(col.name.value.clone(), paimon_type); |
| 621 | } |
| 622 | |
| 623 | // Primary key from constraints: PRIMARY KEY (col, ...) |
| 624 | for constraint in &ct.constraints { |
| 625 | if let datafusion::sql::sqlparser::ast::TableConstraint::PrimaryKey(pk) = constraint { |
| 626 | let pk_cols: Vec<String> = pk |
| 627 | .columns |
| 628 | .iter() |
| 629 | .map(|c| c.column.expr.to_string()) |
| 630 | .collect(); |
| 631 | builder = builder.primary_key(pk_cols); |
| 632 | } |
| 633 | } |
| 634 | |
| 635 | // Partition keys (extracted and validated before parsing) |
| 636 | if !partition_keys.is_empty() { |
| 637 | let col_names: Vec<&str> = ct.columns.iter().map(|c| c.name.value.as_str()).collect(); |
| 638 | for pk in &partition_keys { |
| 639 | if !col_names.contains(&pk.as_str()) { |
| 640 | return Err(DataFusionError::Plan(format!( |
| 641 | "PARTITIONED BY column '{pk}' is not defined in the table" |
| 642 | ))); |
| 643 | } |
| 644 | } |
| 645 | builder = builder.partition_keys(partition_keys); |
| 646 | } |
| 647 | |
| 648 | // Table options from WITH ('key' = 'value', ...) |
no test coverage detected