Minor: add flags for temporary ddl

datafusion/core/src/catalog_common/listing_schema.rs

@@ -136,6 +136,7 @@ impl ListingSchemaProvider {
                             file_type: self.format.clone(),
                             table_partition_cols: vec![],
                             if_not_exists: false,
+                            temporary: false,
                             definition: None,
                             order_exprs: vec![],
                             unbounded: false,

datafusion/core/src/datasource/listing_table_factory.rs

@@ -197,6 +197,7 @@ mod tests {
             schema: Arc::new(DFSchema::empty()),
             table_partition_cols: vec![],
             if_not_exists: false,
+            temporary: false,
             definition: None,
             order_exprs: vec![],
             unbounded: false,
@@ -236,6 +237,7 @@ mod tests {
             schema: Arc::new(DFSchema::empty()),
             table_partition_cols: vec![],
             if_not_exists: false,
+            temporary: false,
             definition: None,
             order_exprs: vec![],
             unbounded: false,

datafusion/core/src/execution/context/mod.rs

@@ -738,6 +738,11 @@ impl SessionContext {
         cmd: &CreateExternalTable,
     ) -> Result<DataFrame> {
         let exist = self.table_exist(cmd.name.clone())?;
+
+        if cmd.temporary {
+            return not_impl_err!("Temporary tables not supported");
+        }
+
         if exist {
             match cmd.if_not_exists {
                 true => return self.return_empty_dataframe(),
@@ -761,10 +766,16 @@ impl SessionContext {
             or_replace,
             constraints,
             column_defaults,
+            temporary,
         } = cmd;
 
         let input = Arc::unwrap_or_clone(input);
         let input = self.state().optimize(&input)?;
+
+        if temporary {
+            return not_impl_err!("Temporary tables not supported");
+        }
+
         let table = self.table(name.clone()).await;
         match (if_not_exists, or_replace, table) {
             (true, false, Ok(_)) => self.return_empty_dataframe(),
@@ -813,10 +824,15 @@ impl SessionContext {
             input,
             or_replace,
             definition,
+            temporary,
         } = cmd;
 
         let view = self.table(name.clone()).await;
 
+        if temporary {
+            return not_impl_err!("Temporary views not supported");
+        }
+
         match (or_replace, view) {
             (true, Ok(_)) => {
                 self.deregister_table(name.clone())?;

datafusion/expr/src/logical_plan/ddl.rs

@@ -202,6 +202,8 @@ pub struct CreateExternalTable {
     pub table_partition_cols: Vec<String>,
     /// Option to not error if table already exists
     pub if_not_exists: bool,
+    /// Whether the table is a temporary table
+    pub temporary: bool,
     /// SQL used to create the table, if available
     pub definition: Option<String>,
     /// Order expressions supplied by user
@@ -298,6 +300,8 @@ pub struct CreateMemoryTable {
     pub or_replace: bool,
     /// Default values for columns
     pub column_defaults: Vec<(String, Expr)>,
+    /// Wheter the table is `TableType::Temporary`
+    pub temporary: bool,
 }
 
 /// Creates a view.
@@ -311,6 +315,8 @@ pub struct CreateView {
     pub or_replace: bool,
     /// SQL used to create the view, if available
     pub definition: Option<String>,
+    /// Wheter the view is ephemeral
+    pub temporary: bool,
 }
 
 /// Creates a catalog (aka "Database").

datafusion/expr/src/logical_plan/plan.rs

@@ -993,6 +993,7 @@ impl LogicalPlan {
                 if_not_exists,
                 or_replace,
                 column_defaults,
+                temporary,
                 ..
             })) => {
                 self.assert_no_expressions(expr)?;
@@ -1005,13 +1006,15 @@ impl LogicalPlan {
                         if_not_exists: *if_not_exists,
                         or_replace: *or_replace,
                         column_defaults: column_defaults.clone(),
+                        temporary: *temporary,
                     },
                 )))
             }
             LogicalPlan::Ddl(DdlStatement::CreateView(CreateView {
                 name,
                 or_replace,
                 definition,
+                temporary,
                 ..
             })) => {
                 self.assert_no_expressions(expr)?;
@@ -1020,6 +1023,7 @@ impl LogicalPlan {
                     input: Arc::new(input),
                     name: name.clone(),
                     or_replace: *or_replace,
+                    temporary: *temporary,
                     definition: definition.clone(),
                 })))
             }

datafusion/expr/src/logical_plan/tree_node.rs

@@ -285,6 +285,7 @@ impl TreeNode for LogicalPlan {
                         if_not_exists,
                         or_replace,
                         column_defaults,
+                        temporary,
                     }) => rewrite_arc(input, f)?.update_data(|input| {
                         DdlStatement::CreateMemoryTable(CreateMemoryTable {
                             name,
@@ -293,19 +294,22 @@ impl TreeNode for LogicalPlan {
                             if_not_exists,
                             or_replace,
                             column_defaults,
+                            temporary,
                         })
                     }),
                     DdlStatement::CreateView(CreateView {
                         name,
                         input,
                         or_replace,
                         definition,
+                        temporary,
                     }) => rewrite_arc(input, f)?.update_data(|input| {
                         DdlStatement::CreateView(CreateView {
                             name,
                             input,
                             or_replace,
                             definition,
+                            temporary,
                         })
                     }),
                     // no inputs in these statements

datafusion/proto/proto/datafusion.proto

@@ -163,6 +163,7 @@ message CreateExternalTableNode {
   datafusion_common.DfSchema schema = 4;
   repeated string table_partition_cols = 5;
   bool if_not_exists = 6;
+  bool temporary = 14;
   string definition = 7;
   repeated SortExprNodeCollection order_exprs = 10;
   bool unbounded = 11;
@@ -200,6 +201,7 @@ message CreateViewNode {
   TableReference name = 5;
   LogicalPlanNode input = 2;
   bool or_replace = 3;
+  bool temporary = 6;
   string definition = 4;
 }
 

datafusion/proto/src/logical_plan/mod.rs

@@ -579,6 +579,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                             .clone(),
                         order_exprs,
                         if_not_exists: create_extern_table.if_not_exists,
+                        temporary: create_extern_table.temporary,
                         definition,
                         unbounded: create_extern_table.unbounded,
                         options: create_extern_table.options.clone(),
@@ -601,6 +602,7 @@ impl AsLogicalPlan for LogicalPlanNode {
 
                 Ok(LogicalPlan::Ddl(DdlStatement::CreateView(CreateView {
                     name: from_table_reference(create_view.name.as_ref(), "CreateView")?,
+                    temporary: create_view.temporary,
                     input: Arc::new(plan),
                     or_replace: create_view.or_replace,
                     definition,
@@ -1386,6 +1388,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                     options,
                     constraints,
                     column_defaults,
+                    temporary,
                 },
             )) => {
                 let mut converted_order_exprs: Vec<SortExprNodeCollection> = vec![];
@@ -1412,6 +1415,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                             schema: Some(df_schema.try_into()?),
                             table_partition_cols: table_partition_cols.clone(),
                             if_not_exists: *if_not_exists,
+                            temporary: *temporary,
                             order_exprs: converted_order_exprs,
                             definition: definition.clone().unwrap_or_default(),
                             unbounded: *unbounded,
@@ -1427,6 +1431,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                 input,
                 or_replace,
                 definition,
+                temporary,
             })) => Ok(protobuf::LogicalPlanNode {
                 logical_plan_type: Some(LogicalPlanType::CreateView(Box::new(
                     protobuf::CreateViewNode {
@@ -1436,6 +1441,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                             extension_codec,
                         )?)),
                         or_replace: *or_replace,
+                        temporary: *temporary,
                         definition: definition.clone().unwrap_or_default(),
                     },
                 ))),