Fix failing tests

Author: berkaysynnada

datafusion/common/src/config.rs

@@ -1566,7 +1566,7 @@ config_namespace! {
     pub struct CsvOptions {
         /// Specifies whether there is a CSV header (i.e. the first line
         /// consists of is column names). If not specified, uses default from
-        /// the `CREATE TABLE` command, if any.
+        /// the session state, if any.
         pub has_header: Option<bool>, default = None
         pub delimiter: u8, default = b','
         pub quote: u8, default = b'"'
@@ -1609,8 +1609,8 @@ impl CsvOptions {
 
     /// Returns true if the first line is a header. If format options does not
     /// specify whether there is a header, consults the configuration.
-    pub fn has_header(&self, config_opt: &ConfigOptions) -> bool {
-        self.has_header.unwrap_or(config_opt.catalog.has_header)
+    pub fn has_header(&self) -> Option<bool> {
+        self.has_header
     }
 
     /// The character separating values within a row.

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

@@ -40,7 +40,7 @@ use arrow::array::RecordBatch;
 use arrow::csv::WriterBuilder;
 use arrow::datatypes::SchemaRef;
 use arrow::datatypes::{DataType, Field, Fields, Schema};
-use datafusion_common::config::{ConfigOptions, CsvOptions};
+use datafusion_common::config::CsvOptions;
 use datafusion_common::file_options::csv_writer::CsvWriterOptions;
 use datafusion_common::{exec_err, not_impl_err, DataFusionError, FileType};
 use datafusion_execution::TaskContext;
@@ -142,8 +142,8 @@ impl CsvFormat {
     }
 
     /// True if the first line is a header.
-    pub fn has_header(&self, config_opt: &ConfigOptions) -> bool {
-        self.options.has_header(config_opt)
+    pub fn has_header(&self) -> Option<bool> {
+        self.options.has_header
     }
 
     /// The character separating values within a row.
@@ -245,7 +245,9 @@ impl FileFormat for CsvFormat {
             conf,
             // If format options does not specify whether there is a header,
             // we consult configuration options.
-            self.options.has_header(state.config_options()),
+            self.options
+                .has_header
+                .unwrap_or(state.config_options().catalog.has_header),
             self.options.delimiter,
             self.options.quote,
             self.options.escape,
@@ -303,7 +305,10 @@ impl CsvFormat {
         while let Some(chunk) = stream.next().await.transpose()? {
             let format = arrow::csv::reader::Format::default()
                 .with_header(
-                    self.options.has_header(state.config_options()) && first_chunk,
+                    self.options
+                        .has_header
+                        .unwrap_or(state.config_options().catalog.has_header)
+                        && first_chunk,
                 )
                 .with_delimiter(self.options.delimiter);
 

datafusion/core/src/datasource/stream.rs

@@ -30,7 +30,7 @@ use arrow_schema::SchemaRef;
 use async_trait::async_trait;
 use futures::StreamExt;
 
-use datafusion_common::{plan_err, Constraints, DataFusionError, Result};
+use datafusion_common::{config_err, plan_err, Constraints, DataFusionError, Result};
 use datafusion_common_runtime::SpawnedTask;
 use datafusion_execution::{SendableRecordBatchStream, TaskContext};
 use datafusion_expr::{CreateExternalTable, Expr, TableType};
@@ -58,11 +58,22 @@ impl TableProviderFactory for StreamTableFactory {
         let schema: SchemaRef = Arc::new(cmd.schema.as_ref().into());
         let location = cmd.location.clone();
         let encoding = cmd.file_type.parse()?;
+        let header = if let Ok(opt) = cmd
+            .options
+            .get("format.has_header")
+            .map(|has_header| bool::from_str(has_header))
+            .transpose()
+        {
+            opt.unwrap_or(false)
+        } else {
+            return config_err!("format.has_header can either be true or false");
+        };
 
         let config = StreamConfig::new_file(schema, location.into())
             .with_encoding(encoding)
             .with_order(cmd.order_exprs.clone())
             .with_batch_size(state.config().batch_size())
+            .with_header(header)
             .with_constraints(cmd.constraints.clone());
 
         Ok(Arc::new(StreamTable(Arc::new(config))))

datafusion/core/tests/sql/mod.rs

@@ -85,7 +85,7 @@ async fn register_aggregate_csv_by_sql(ctx: &SessionContext) {
         c13 VARCHAR NOT NULL
     )
     STORED AS CSV
-    LOCATION '{testdata}/csv/aggregate_test_100.csv
+    LOCATION '{testdata}/csv/aggregate_test_100.csv'
     OPTIONS ('format.has_header' 'true')
     "
         ))

datafusion/proto/tests/cases/roundtrip_logical_plan.rs

@@ -238,7 +238,7 @@ async fn roundtrip_custom_listing_tables() -> Result<()> {
             STORED AS CSV
             WITH ORDER (a ASC, b ASC)
             WITH ORDER (c ASC)
-            LOCATION '../core/tests/data/window_2.csv';
+            LOCATION '../core/tests/data/window_2.csv'
             OPTIONS ('format.has_header' 'true')";
 
     let plan = ctx.state().create_logical_plan(query).await?;
@@ -268,7 +268,7 @@ async fn roundtrip_logical_plan_aggregation_with_pk() -> Result<()> {
             STORED AS CSV
             WITH ORDER (a ASC, b ASC)
             WITH ORDER (c ASC)
-            LOCATION '../core/tests/data/window_2.csv';
+            LOCATION '../core/tests/data/window_2.csv'
             OPTIONS ('format.has_header' 'true')",
     )
     .await?;