add schema parameter to table provider factory create method

datafusion/core/src/test_util.rs

@@ -289,6 +289,7 @@ impl TableProviderFactory for TestTableFactory {
     ) -> datafusion_common::Result<Arc<dyn TableProvider>> {
         Ok(Arc::new(TestTableProvider {
             url: cmd.location.to_string(),
+            schema: Arc::new(cmd.schema.as_ref().into()),
         }))
     }
 }
@@ -297,6 +298,8 @@ impl TableProviderFactory for TestTableFactory {
 pub struct TestTableProvider {
     /// URL of table files or folder
     pub url: String,
+    /// test table schema
+    pub schema: SchemaRef,
 }
 
 impl TestTableProvider {}
@@ -308,11 +311,7 @@ impl TableProvider for TestTableProvider {
     }
 
     fn schema(&self) -> SchemaRef {
-        let schema = Schema::new(vec![
-            Field::new("a", DataType::Int64, true),
-            Field::new("b", DataType::Decimal128(15, 2), true),
-        ]);
-        Arc::new(schema)
+        self.schema.clone()
     }
 
     fn table_type(&self) -> TableType {

datafusion/core/tests/sql/create_drop.rs

@@ -387,6 +387,36 @@ async fn create_custom_table() -> Result<()> {
     Ok(())
 }
 
+#[tokio::test]
+async fn create_external_table_with_ddl() -> Result<()> {
+    let mut table_factories: HashMap<String, Arc<dyn TableProviderFactory>> =
+        HashMap::new();
+    table_factories.insert("mocktable".to_string(), Arc::new(TestTableFactory {}));
+    let cfg = RuntimeConfig::new().with_table_factories(table_factories);
+    let env = RuntimeEnv::new(cfg).unwrap();
+    let ses = SessionConfig::new();
+    let ctx = SessionContext::with_config_rt(ses, Arc::new(env));
+
+    let sql = "CREATE EXTERNAL TABLE dt (a_id integer, a_str string, a_bool boolean) STORED AS MOCKTABLE LOCATION 'mockprotocol://path/to/table';";
+    ctx.sql(sql).await.unwrap();
+
+    let cat = ctx.catalog("datafusion").unwrap();
+    let schema = cat.schema("public").unwrap();
+
+    let exists = schema.table_exist("dt");
+    assert!(exists, "Table should have been created!");
+
+    let table_schema = schema.table("dt").unwrap().schema();
+
+    assert_eq!(3, table_schema.fields().len());
+
+    assert_eq!(&DataType::Int32, table_schema.field(0).data_type());
+    assert_eq!(&DataType::Utf8, table_schema.field(1).data_type());
+    assert_eq!(&DataType::Boolean, table_schema.field(2).data_type());
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn create_bad_custom_table() {
     let ctx = SessionContext::new();

datafusion/proto/src/lib.rs

@@ -172,13 +172,16 @@ mod roundtrip_tests {
         fn try_decode_table_provider(
             &self,
             buf: &[u8],
-            _schema: SchemaRef,
+            schema: SchemaRef,
             _ctx: &SessionContext,
         ) -> Result<Arc<dyn TableProvider>, DataFusionError> {
             let msg = TestTableProto::decode(buf).map_err(|_| {
                 DataFusionError::Internal("Error decoding test table".to_string())
             })?;
-            let provider = TestTableProvider { url: msg.url };
+            let provider = TestTableProvider {
+                url: msg.url,
+                schema,
+            };
             Ok(Arc::new(provider))
         }