MCPcopy Create free account
hub / github.com/apache/datafusion / create_memory_table

Method create_memory_table

datafusion/core/src/execution/context/mod.rs:868–927  ·  view source on GitHub ↗
(&self, cmd: CreateMemoryTable)

Source from the content-addressed store, hash-verified

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"),

Callers 1

execute_logical_planMethod · 0.80

Calls 15

newFunction · 0.85
collect_partitionedMethod · 0.80
collectMethod · 0.80
optimizeMethod · 0.45
stateMethod · 0.45
tableMethod · 0.45
cloneMethod · 0.45
schemaMethod · 0.45
deregister_tableMethod · 0.45
innerMethod · 0.45
with_column_defaultsMethod · 0.45

Tested by

no test coverage detected