fix empty csv schema (#3272)

datafusion/core/src/datasource/file_format/csv.rs

@@ -116,10 +116,10 @@ impl FileFormat for CsvFormat {
                 Some(records_to_read),
                 self.has_header,
             )?;
+            schemas.push(schema.clone());
             if records_read == 0 {
                 continue;
             }
-            schemas.push(schema.clone());
             records_to_read -= records_read;
             if records_to_read == 0 {
                 break;

datafusion/core/tests/empty.csv

@@ -0,0 +1 @@
+c1,c2,c3

datafusion/core/tests/sql/create_drop.rs

@@ -261,3 +261,29 @@ async fn create_pipe_delimited_csv_table() -> Result<()> {
 
     Ok(())
 }
+
+#[tokio::test]
+async fn create_csv_table_empty_file() -> Result<()> {
+    let ctx =
+        SessionContext::with_config(SessionConfig::new().with_information_schema(true));
+
+    let sql = "CREATE EXTERNAL TABLE empty STORED AS CSV WITH HEADER ROW LOCATION 'tests/empty.csv'";
+    ctx.sql(sql).await.unwrap();
+    let sql =
+        "select column_name, data_type, ordinal_position from information_schema.columns";
+    let results = execute_to_batches(&ctx, sql).await;
+
+    let expected = vec![
+        "+-------------+-----------+------------------+",
+        "| column_name | data_type | ordinal_position |",
+        "+-------------+-----------+------------------+",
+        "| c1          | Utf8      | 0                |",
+        "| c2          | Utf8      | 1                |",
+        "| c3          | Utf8      | 2                |",
+        "+-------------+-----------+------------------+",
+    ];
+
+    assert_batches_eq!(expected, &results);
+
+    Ok(())
+}