(&self, cmd: CreateCatalogSchema)
| 987 | } |
| 988 | |
| 989 | async fn create_catalog_schema(&self, cmd: CreateCatalogSchema) -> Result<DataFrame> { |
| 990 | let CreateCatalogSchema { |
| 991 | schema_name, |
| 992 | if_not_exists, |
| 993 | .. |
| 994 | } = cmd; |
| 995 | |
| 996 | // sqlparser doesn't accept database / catalog as parameter to CREATE SCHEMA |
| 997 | // so for now, we default to default catalog |
| 998 | let tokens: Vec<&str> = schema_name.split('.').collect(); |
| 999 | let (catalog, schema_name) = match tokens.len() { |
| 1000 | 1 => { |
| 1001 | let state = self.state.read(); |
| 1002 | let name = &state.config().options().catalog.default_catalog; |
| 1003 | let catalog = state.catalog_list().catalog(name).ok_or_else(|| { |
| 1004 | exec_datafusion_err!("Missing default catalog '{name}'") |
| 1005 | })?; |
| 1006 | (catalog, tokens[0]) |
| 1007 | } |
| 1008 | 2 => { |
| 1009 | let name = &tokens[0]; |
| 1010 | let catalog = self |
| 1011 | .catalog(name) |
| 1012 | .ok_or_else(|| exec_datafusion_err!("Missing catalog '{name}'"))?; |
| 1013 | (catalog, tokens[1]) |
| 1014 | } |
| 1015 | _ => return exec_err!("Unable to parse catalog from {schema_name}"), |
| 1016 | }; |
| 1017 | let schema = catalog.schema(schema_name); |
| 1018 | |
| 1019 | match (if_not_exists, schema) { |
| 1020 | (true, Some(_)) => self.return_empty_dataframe(), |
| 1021 | (true, None) | (false, None) => { |
| 1022 | let schema = Arc::new(MemorySchemaProvider::new()); |
| 1023 | catalog.register_schema(schema_name, schema)?; |
| 1024 | self.return_empty_dataframe() |
| 1025 | } |
| 1026 | (false, Some(_)) => exec_err!("Schema '{schema_name}' already exists"), |
| 1027 | } |
| 1028 | } |
| 1029 | |
| 1030 | async fn create_catalog(&self, cmd: CreateCatalog) -> Result<DataFrame> { |
| 1031 | let CreateCatalog { |
no test coverage detected