fix: infer data type for null columns and empty tables as Utf8`

Author: dqkqd

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

@@ -253,16 +253,8 @@ mod tests {
         let projection = None;
         let root = "./tests/data/csv";
         let format = CsvFormat::default().with_has_header(true);
-        let exec = scan_format(
-            &state,
-            &format,
-            None,
-            root,
-            "aggregate_test_100_with_nulls.csv",
-            projection,
-            None,
-        )
-        .await?;
+        let exec =
+            get_exec(&state, "aggregate_test_100.csv", projection, None, true).await?;
 
         let x: Vec<String> = exec
             .schema()
@@ -285,8 +277,6 @@ mod tests {
                 "c11: Float64",
                 "c12: Float64",
                 "c13: Utf8",
-                "c14: Null",
-                "c15: Utf8"
             ],
             x
         );
@@ -295,56 +285,95 @@ mod tests {
     }
 
     #[tokio::test]
+    async fn infer_schema_with_null() -> Result<()> {
+        let session_ctx = SessionContext::new();
+        let state = session_ctx.state();
+        let variable_object_store = Arc::new(VariableStream::new(
+            Bytes::from(
+                r#"has_nulls,all_nulls,utf8
+1,,NULL
+,,
+2,,NULL"#,
+            ),
+            1,
+        ));
+        let object_meta = ObjectMeta {
+            location: Path::parse("/")?,
+            last_modified: DateTime::default(),
+            size: u64::MAX,
+            e_tag: None,
+            version: None,
+        };
+
+        let csv_format = CsvFormat::default().with_has_header(true);
+        let inferred_schema = csv_format
+            .infer_schema(
+                &state,
+                &(variable_object_store.clone() as Arc<dyn ObjectStore>),
+                &[object_meta],
+            )
+            .await?;
+
+        let actual_fields: Vec<_> = inferred_schema
+            .fields()
+            .iter()
+            .map(|f| format!("{}: {:?}", f.name(), f.data_type()))
+            .collect();
+        assert_eq!(
+            vec!["has_nulls: Int64", "all_nulls: Utf8", "utf8: Utf8"],
+            actual_fields
+        );
+
+        Ok(())
+    }
+
     async fn infer_schema_with_null_regex() -> Result<()> {
         let session_ctx = SessionContext::new();
         let state = session_ctx.state();
+        let variable_object_store = Arc::new(VariableStream::new(
+            Bytes::from(
+                r#"has_nulls,all_nulls,has_nulls_regex,all_nulls_regex
+1,,1,NULL
+,,NULL,NULL
+2,,2,NULL"#,
+            ),
+            1,
+        ));
+        let object_meta = ObjectMeta {
+            location: Path::parse("/")?,
+            last_modified: DateTime::default(),
+            size: u64::MAX,
+            e_tag: None,
+            version: None,
+        };
 
-        let projection = None;
-        let root = "./tests/data/csv";
-        let format = CsvFormat::default()
+        let csv_format = CsvFormat::default()
             .with_has_header(true)
             .with_null_regex(Some("^NULL$|^$".to_string()));
-        let exec = scan_format(
-            &state,
-            &format,
-            None,
-            root,
-            "aggregate_test_100_with_nulls.csv",
-            projection,
-            None,
-        )
-        .await?;
+        let inferred_schema = csv_format
+            .infer_schema(
+                &state,
+                &(variable_object_store.clone() as Arc<dyn ObjectStore>),
+                &[object_meta],
+            )
+            .await?;
 
-        let x: Vec<String> = exec
-            .schema()
+        let actual_fields: Vec<_> = inferred_schema
             .fields()
             .iter()
             .map(|f| format!("{}: {:?}", f.name(), f.data_type()))
             .collect();
         assert_eq!(
             vec![
-                "c1: Utf8",
-                "c2: Int64",
-                "c3: Int64",
-                "c4: Int64",
-                "c5: Int64",
-                "c6: Int64",
-                "c7: Int64",
-                "c8: Int64",
-                "c9: Int64",
-                "c10: Utf8",
-                "c11: Float64",
-                "c12: Float64",
-                "c13: Utf8",
-                "c14: Null",
-                "c15: Null"
+                "has_nulls: Int64",
+                "all_nulls: Utf8",
+                "has_nulls_regex: Int64",
+                "all_nulls_regex: Utf8"
             ],
-            x
+            actual_fields
         );
-
         Ok(())
     }
-
     #[tokio::test]
     async fn read_char_column() -> Result<()> {
         let session_ctx = SessionContext::new();
@@ -489,8 +518,7 @@ mod tests {
                 Bytes::from(
                     r#"c1,c2,c3
 1,1.0,
-,,
-"#,
+,,"#,
                 ),
                 1,
             )),
@@ -520,7 +548,7 @@ mod tests {
             .collect();
 
         // ensure null chunks don't skew type inference
-        assert_eq!(vec!["c1: Int64", "c2: Float64", "c3: Null"], actual_fields);
+        assert_eq!(vec!["c1: Int64", "c2: Float64", "c3: Utf8"], actual_fields);
         Ok(())
     }
 

datafusion/datasource-csv/src/file_format.rs

@@ -600,7 +600,6 @@ fn build_schema_helper(names: Vec<String>, types: Vec<HashSet<DataType>>) -> Sch
             data_type_possibilities.remove(&DataType::Null);
 
             match data_type_possibilities.len() {
-                0 => Field::new(field_name, DataType::Null, true),
                 1 => Field::new(
                     field_name,
                     data_type_possibilities.iter().next().unwrap().clone(),