[sqllogictest] Run tests on Windows (#5870)

datafusion/core/tests/sqllogictests/src/engines/datafusion/mod.rs

@@ -24,7 +24,6 @@ use self::error::{DFSqlLogicTestError, Result};
 use async_trait::async_trait;
 use datafusion::arrow::record_batch::RecordBatch;
 use datafusion::prelude::SessionContext;
-use datafusion_sql::parser::DFParser;
 use sqllogictest::DBOutput;
 
 mod error;
@@ -64,7 +63,7 @@ impl sqllogictest::AsyncDB for DataFusion {
 
     /// [`Runner`] calls this function to perform sleep.
     ///
-    /// The default implementation is `std::thread::sleep`, which is universial to any async runtime
+    /// The default implementation is `std::thread::sleep`, which is universal to any async runtime
     /// but would block the current thread. If you are running in tokio runtime, you should override
     /// this by `tokio::time::sleep`.
     async fn sleep(dur: Duration) {
@@ -73,12 +72,7 @@ impl sqllogictest::AsyncDB for DataFusion {
 }
 
 async fn run_query(ctx: &SessionContext, sql: impl Into<String>) -> Result<DFOutput> {
-    let sql = sql.into();
-    // check if the sql is more than one statement
-    if let Ok(mut statements) = DFParser::parse_sql(&sql) {
-        statements.pop_front().expect("at least one SQL statement");
-    }
-    let df = ctx.sql(sql.as_str()).await?;
+    let df = ctx.sql(sql.into().as_str()).await?;
 
     let types = normalize::convert_schema_to_types(df.schema().fields());
     let results: Vec<RecordBatch> = df.collect().await?;

datafusion/core/tests/sqllogictests/src/engines/datafusion/normalize.rs

@@ -135,10 +135,18 @@ fn workspace_root() -> object_store::path::Path {
         .expect("parent of datafusion")
         .to_string_lossy();
 
-    object_store::path::Path::parse(workspace_root).unwrap()
+    let sanitized_workplace_root = if cfg!(windows) {
+        // Object store paths are delimited with `/`, e.g. `D:/a/arrow-datafusion/arrow-datafusion/testing/data/csv/aggregate_test_100.csv`.
+        // The default windows delimiter is `\`, so the workplace path is `D:\a\arrow-datafusion\arrow-datafusion`.
+        workspace_root.replace(std::path::MAIN_SEPARATOR, object_store::path::DELIMITER)
+    } else {
+        workspace_root.to_string()
+    };
+
+    object_store::path::Path::parse(sanitized_workplace_root).unwrap()
 }
 
-// holds the root directory (
+// holds the root directory
 lazy_static! {
     static ref WORKSPACE_ROOT: object_store::path::Path = workspace_root();
 }

datafusion/core/tests/sqllogictests/src/main.rs

@@ -17,6 +17,8 @@
 
 use std::error::Error;
 use std::path::{Path, PathBuf};
+#[cfg(target_family = "windows")]
+use std::thread;
 
 use log::info;
 use sqllogictest::strict_column_validator;
@@ -33,16 +35,31 @@ mod utils;
 const TEST_DIRECTORY: &str = "tests/sqllogictests/test_files/";
 const PG_COMPAT_FILE_PREFIX: &str = "pg_compat_";
 
-#[tokio::main]
 #[cfg(target_family = "windows")]
-pub async fn main() -> Result<(), Box<dyn Error>> {
-    println!("Skipping test on windows");
-    Ok(())
+pub fn main() {
+    // Tests from `tpch.slt` fail with stackoverflow with the default stack size.
+    thread::Builder::new()
+        .stack_size(2 * 1024 * 1024) // 2 MB
+        .spawn(move || {
+            tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .unwrap()
+                .block_on(async { run_tests().await })
+                .unwrap()
+        })
+        .unwrap()
+        .join()
+        .unwrap();
 }
 
 #[tokio::main]
 #[cfg(not(target_family = "windows"))]
 pub async fn main() -> Result<(), Box<dyn Error>> {
+    run_tests().await
+}
+
+async fn run_tests() -> Result<(), Box<dyn Error>> {
     // Enable logging (e.g. set RUST_LOG=debug to see debug logs)
     env_logger::init();