Add show external tables

benchmarks/src/bin/h2o.rs

@@ -83,7 +83,7 @@ async fn group_by(opt: &GroupBy) -> Result<()> {
         let listing_config = ListingTableConfig::new(ListingTableUrl::parse(path)?)
             .with_listing_options(ListingOptions::new(Arc::new(CsvFormat::default())))
             .with_schema(Arc::new(schema));
-        let csv = ListingTable::try_new(listing_config)?;
+        let csv = ListingTable::try_new(listing_config, None)?;
         let partition_size = num_cpus::get();
         let memtable =
             MemTable::load(Arc::new(csv), Some(partition_size), &ctx.state()).await?;

benchmarks/src/bin/tpch.rs

@@ -429,7 +429,7 @@ fn get_table(
         .with_listing_options(options)
         .with_schema(schema);
 
-    Ok(Arc::new(ListingTable::try_new(config)?))
+    Ok(Arc::new(ListingTable::try_new(config, None)?))
 }
 
 fn get_schema(table: &str) -> Schema {

datafusion-examples/examples/parquet_sql_multiple_files.rs

@@ -49,6 +49,7 @@ async fn main() -> Result<()> {
         &format!("file://{}", testdata),
         listing_options,
         None,
+        None,
     )
     .await
     .unwrap();

datafusion/core/benches/sort_limit_query_sql.rs

@@ -74,7 +74,7 @@ fn create_context() -> Arc<Mutex<SessionContext>> {
         .with_listing_options(listing_options)
         .with_schema(schema);
 
-    let csv = async { ListingTable::try_new(config).unwrap() };
+    let csv = async { ListingTable::try_new(config, None).unwrap() };
 
     let rt = Runtime::new().unwrap();
 

datafusion/core/src/catalog/schema.rs

@@ -184,7 +184,7 @@ mod tests {
             .infer(&ctx.state())
             .await
             .unwrap();
-        let table = ListingTable::try_new(config).unwrap();
+        let table = ListingTable::try_new(config, None).unwrap();
 
         schema
             .register_table("alltypes_plain".to_string(), Arc::new(table))

datafusion/core/src/datasource/listing/table.rs

@@ -246,6 +246,7 @@ pub struct ListingTable {
     /// File fields + partition columns
     table_schema: SchemaRef,
     options: ListingOptions,
+    definition: Option<String>,
 }
 
 impl ListingTable {
@@ -256,7 +257,10 @@ impl ListingTable {
     /// If the schema is provided then it must be resolved before creating the table
     /// and should contain the fields of the file without the table
     /// partitioning columns.
-    pub fn try_new(config: ListingTableConfig) -> Result<Self> {
+    pub fn try_new(
+        config: ListingTableConfig,
+        definition: Option<String>,
+    ) -> Result<Self> {
         let file_schema = config
             .file_schema
             .ok_or_else(|| DataFusionError::Internal("No schema provided.".into()))?;
@@ -280,6 +284,7 @@ impl ListingTable {
             file_schema,
             table_schema: Arc::new(Schema::new(table_fields)),
             options,
+            definition,
         };
 
         Ok(table)
@@ -358,6 +363,10 @@ impl TableProvider for ListingTable {
             Ok(TableProviderFilterPushDown::Inexact)
         }
     }
+
+    fn get_table_definition(&self) -> Option<&str> {
+        self.definition.as_deref()
+    }
 }
 
 impl ListingTable {
@@ -460,7 +469,7 @@ mod tests {
         let config = ListingTableConfig::new(table_path)
             .with_listing_options(opt)
             .with_schema(schema);
-        let table = ListingTable::try_new(config)?;
+        let table = ListingTable::try_new(config, None)?;
 
         let exec = table.scan(&state, &None, &[], None).await?;
         assert_eq!(exec.statistics().num_rows, Some(8));
@@ -489,7 +498,7 @@ mod tests {
         let config = ListingTableConfig::new(table_path)
             .with_listing_options(opt)
             .with_schema(file_schema);
-        let table = ListingTable::try_new(config)?;
+        let table = ListingTable::try_new(config, None)?;
 
         assert_eq!(
             columns(&table.schema()),
@@ -660,7 +669,7 @@ mod tests {
         let config = ListingTableConfig::new(table_path)
             .infer(&ctx.state())
             .await?;
-        let table = ListingTable::try_new(config)?;
+        let table = ListingTable::try_new(config, None)?;
         Ok(Arc::new(table))
     }
 
@@ -692,7 +701,7 @@ mod tests {
             .with_listing_options(opt)
             .with_schema(Arc::new(schema));
 
-        let table = ListingTable::try_new(config)?;
+        let table = ListingTable::try_new(config, None)?;
 
         let (file_list, _) = table.list_files_for_scan(&ctx.state(), &[], None).await?;
 
@@ -732,7 +741,7 @@ mod tests {
             .with_listing_options(opt)
             .with_schema(Arc::new(schema));
 
-        let table = ListingTable::try_new(config)?;
+        let table = ListingTable::try_new(config, None)?;
 
         let (file_list, _) = table.list_files_for_scan(&ctx.state(), &[], None).await?;
 

datafusion/core/src/execution/context.rs

@@ -244,6 +244,7 @@ impl SessionContext {
                 ref delimiter,
                 ref table_partition_cols,
                 ref if_not_exists,
+                definition,
             }) => {
                 let (file_format, file_extension) = match file_type {
                     FileType::CSV => (
@@ -292,6 +293,7 @@ impl SessionContext {
                             location,
                             options,
                             provided_schema,
+                            definition,
                         )
                         .await?;
                         let plan = LogicalPlanBuilder::empty(false).build()?;
@@ -558,7 +560,7 @@ impl SessionContext {
         let config = ListingTableConfig::new(table_path)
             .with_listing_options(listing_options)
             .with_schema(resolved_schema);
-        let provider = ListingTable::try_new(config)?;
+        let provider = ListingTable::try_new(config, None)?;
         self.read_table(Arc::new(provider))
     }
 
@@ -584,7 +586,7 @@ impl SessionContext {
         let config = ListingTableConfig::new(table_path)
             .with_listing_options(listing_options)
             .with_schema(resolved_schema);
-        let provider = ListingTable::try_new(config)?;
+        let provider = ListingTable::try_new(config, None)?;
 
         self.read_table(Arc::new(provider))
     }
@@ -618,7 +620,7 @@ impl SessionContext {
             .with_listing_options(listing_options)
             .with_schema(resolved_schema);
 
-        let provider = ListingTable::try_new(config)?;
+        let provider = ListingTable::try_new(config, None)?;
         self.read_table(Arc::new(provider))
     }
 
@@ -642,7 +644,7 @@ impl SessionContext {
             .with_listing_options(listing_options)
             .with_schema(resolved_schema);
 
-        let provider = ListingTable::try_new(config)?;
+        let provider = ListingTable::try_new(config, None)?;
         self.read_table(Arc::new(provider))
     }
 
@@ -664,6 +666,7 @@ impl SessionContext {
         table_path: impl AsRef<str>,
         options: ListingOptions,
         provided_schema: Option<SchemaRef>,
+        sql: Option<String>,
     ) -> Result<()> {
         let table_path = ListingTableUrl::parse(table_path)?;
         let resolved_schema = match provided_schema {
@@ -673,7 +676,7 @@ impl SessionContext {
         let config = ListingTableConfig::new(table_path)
             .with_listing_options(options)
             .with_schema(resolved_schema);
-        let table = ListingTable::try_new(config)?;
+        let table = ListingTable::try_new(config, sql)?;
         self.register_table(name, Arc::new(table))?;
         Ok(())
     }
@@ -694,6 +697,7 @@ impl SessionContext {
             table_path,
             listing_options,
             options.schema.map(|s| Arc::new(s.to_owned())),
+            None,
         )
         .await?;
 
@@ -711,8 +715,14 @@ impl SessionContext {
         let listing_options =
             options.to_listing_options(self.copied_config().target_partitions);
 
-        self.register_listing_table(name, table_path, listing_options, options.schema)
-            .await?;
+        self.register_listing_table(
+            name,
+            table_path,
+            listing_options,
+            options.schema,
+            None,
+        )
+        .await?;
         Ok(())
     }
 
@@ -732,7 +742,7 @@ impl SessionContext {
             .parquet_pruning(parquet_pruning)
             .to_listing_options(target_partitions);
 
-        self.register_listing_table(name, table_path, listing_options, None)
+        self.register_listing_table(name, table_path, listing_options, None, None)
             .await?;
         Ok(())
     }
@@ -748,8 +758,14 @@ impl SessionContext {
         let listing_options =
             options.to_listing_options(self.copied_config().target_partitions);
 
-        self.register_listing_table(name, table_path, listing_options, options.schema)
-            .await?;
+        self.register_listing_table(
+            name,
+            table_path,
+            listing_options,
+            options.schema,
+            None,
+        )
+        .await?;
         Ok(())
     }
 

datafusion/core/tests/path_partition.rs

@@ -441,7 +441,7 @@ fn register_partitioned_aggregate_csv(
     let config = ListingTableConfig::new(table_path)
         .with_listing_options(options)
         .with_schema(file_schema);
-    let table = ListingTable::try_new(config).unwrap();
+    let table = ListingTable::try_new(config, None).unwrap();
 
     ctx.register_table("t", Arc::new(table))
         .expect("registering listing table failed");
@@ -479,7 +479,7 @@ async fn register_partitioned_alltypes_parquet(
         .with_listing_options(options)
         .with_schema(file_schema);
 
-    let table = ListingTable::try_new(config).unwrap();
+    let table = ListingTable::try_new(config, None).unwrap();
 
     ctx.register_table("t", Arc::new(table))
         .expect("registering listing table failed");

datafusion/core/tests/sql/information_schema.rs

@@ -605,6 +605,29 @@ async fn show_create_table() {
     assert_batches_eq!(expected, &results);
 }
 
+#[tokio::test]
+async fn show_external_create_table() {
+    let ctx =
+        SessionContext::with_config(SessionConfig::new().with_information_schema(true));
+
+    let table_sql =
+        "CREATE EXTERNAL TABLE abc STORED AS CSV WITH HEADER ROW LOCATION '../../testing/data/csv/aggregate_test_100.csv'";
+    plan_and_collect(&ctx, table_sql).await.unwrap();
+
+    let result_sql = "SHOW CREATE TABLE abc";
+    let results = plan_and_collect(&ctx, result_sql).await.unwrap();
+
+    let expected = vec![
+        "+---------------+--------------+------------+-------------------------------------------------------------------------------------------------+",
+        "| table_catalog | table_schema | table_name | definition                                                                                      |",
+        "+---------------+--------------+------------+-------------------------------------------------------------------------------------------------+",
+        "| datafusion    | public       | abc        | CREATE EXTERNAL TABLE abc STORED AS CSV LOCATION ../../testing/data/csv/aggregate_test_100.csv  |",
+        "+---------------+--------------+------------+-------------------------------------------------------------------------------------------------+",
+    ];
+
+    assert_batches_eq!(expected, &results);
+}
+
 /// Execute SQL and return results
 async fn plan_and_collect(ctx: &SessionContext, sql: &str) -> Result<Vec<RecordBatch>> {
     ctx.sql(sql).await?.collect().await

datafusion/expr/src/logical_plan/plan.rs

@@ -1227,6 +1227,21 @@ pub enum FileType {
     Avro,
 }
 
+impl fmt::Display for FileType {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "{}",
+            match self {
+                FileType::NdJson => "NDJSON",
+                FileType::Parquet => "PARQUET",
+                FileType::CSV => "CSV",
+                FileType::Avro => "AVRO",
+            }
+        )
+    }
+}
+
 /// Creates an external table.
 #[derive(Clone)]
 pub struct CreateExternalTable {
@@ -1246,6 +1261,8 @@ pub struct CreateExternalTable {
     pub table_partition_cols: Vec<String>,
     /// Option to not error if table already exists
     pub if_not_exists: bool,
+    /// SQL used to create the view, if available
+    pub definition: Option<String>,
 }
 
 /// Produces a relation with string representations of

datafusion/proto/proto/datafusion.proto

@@ -153,6 +153,7 @@ message CreateExternalTableNode {
   repeated string table_partition_cols = 6;
   bool if_not_exists = 7;
   string delimiter = 8;
+  string definition = 9;
 }
 
 message CreateCatalogSchemaNode {

datafusion/proto/src/logical_plan.rs

@@ -430,7 +430,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                         .with_listing_options(options)
                         .with_schema(Arc::new(schema));
 
-                let provider = ListingTable::try_new(config)?;
+                let provider = ListingTable::try_new(config, None)?;
 
                 LogicalPlanBuilder::scan_with_filters(
                     &scan.table_name,
@@ -493,6 +493,11 @@ impl AsLogicalPlan for LogicalPlanNode {
 
                 let pb_file_type: protobuf::FileType =
                     create_extern_table.file_type.try_into()?;
+                let definition = if !create_extern_table.definition.is_empty() {
+                    Some(create_extern_table.definition.clone())
+                } else {
+                    None
+                };
 
                 Ok(LogicalPlan::CreateExternalTable(CreateExternalTable {
                     schema: pb_schema.try_into()?,
@@ -507,6 +512,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                         .table_partition_cols
                         .clone(),
                     if_not_exists: create_extern_table.if_not_exists,
+                    definition,
                 }))
             }
             LogicalPlanType::CreateView(create_view) => {
@@ -1055,6 +1061,7 @@ impl AsLogicalPlan for LogicalPlanNode {
                 schema: df_schema,
                 table_partition_cols,
                 if_not_exists,
+                definition,
             }) => {
                 use datafusion::logical_plan::FileType;
 
@@ -1076,6 +1083,9 @@ impl AsLogicalPlan for LogicalPlanNode {
                             table_partition_cols: table_partition_cols.clone(),
                             if_not_exists: *if_not_exists,
                             delimiter: String::from(*delimiter),
+                            definition: definition
+                                .clone()
+                                .unwrap_or_else(|| "".to_string()),
                         },
                     )),
                 })