(&self, cmd: CreateMemoryTable)
| 866 | } |
| 867 | |
| 868 | async fn create_memory_table(&self, cmd: CreateMemoryTable) -> Result<DataFrame> { |
| 869 | let CreateMemoryTable { |
| 870 | name, |
| 871 | input, |
| 872 | if_not_exists, |
| 873 | or_replace, |
| 874 | constraints, |
| 875 | column_defaults, |
| 876 | temporary, |
| 877 | } = cmd; |
| 878 | |
| 879 | let input = Arc::unwrap_or_clone(input); |
| 880 | let input = self.state().optimize(&input)?; |
| 881 | |
| 882 | if temporary { |
| 883 | return not_impl_err!("Temporary tables not supported"); |
| 884 | } |
| 885 | |
| 886 | let table = self.table(name.clone()).await; |
| 887 | match (if_not_exists, or_replace, table) { |
| 888 | (true, false, Ok(_)) => self.return_empty_dataframe(), |
| 889 | (false, true, Ok(_)) => { |
| 890 | Self::ensure_unique_column_names(input.schema())?; |
| 891 | self.deregister_table(name.clone())?; |
| 892 | let schema = Arc::clone(input.schema().inner()); |
| 893 | let physical = DataFrame::new(self.state(), input); |
| 894 | |
| 895 | let batches: Vec<_> = physical.collect_partitioned().await?; |
| 896 | let table = Arc::new( |
| 897 | // pass constraints and column defaults to the mem table. |
| 898 | MemTable::try_new(schema, batches)? |
| 899 | .with_constraints(constraints) |
| 900 | .with_column_defaults(column_defaults.into_iter().collect()), |
| 901 | ); |
| 902 | |
| 903 | self.register_table(name.clone(), table)?; |
| 904 | self.return_empty_dataframe() |
| 905 | } |
| 906 | (true, true, Ok(_)) => { |
| 907 | exec_err!("'IF NOT EXISTS' cannot coexist with 'REPLACE'") |
| 908 | } |
| 909 | (_, _, Err(_)) => { |
| 910 | Self::ensure_unique_column_names(input.schema())?; |
| 911 | let schema = Arc::clone(input.schema().inner()); |
| 912 | let physical = DataFrame::new(self.state(), input); |
| 913 | |
| 914 | let batches: Vec<_> = physical.collect_partitioned().await?; |
| 915 | let table = Arc::new( |
| 916 | // pass constraints and column defaults to the mem table. |
| 917 | MemTable::try_new(schema, batches)? |
| 918 | .with_constraints(constraints) |
| 919 | .with_column_defaults(column_defaults.into_iter().collect()), |
| 920 | ); |
| 921 | |
| 922 | self.register_table(name, table)?; |
| 923 | self.return_empty_dataframe() |
| 924 | } |
| 925 | (false, false, Ok(_)) => exec_err!("Table '{name}' already exists"), |
no test coverage detected