fix(ci): fix tpcds spill tests that no longer work (#18869)

* update

* update

Author: forsaken628

.github/actions/test_sqllogic_cluster_linux/action.yml

@@ -26,5 +26,4 @@ runs:
       env:
         TEST_HANDLERS: ${{ inputs.handlers }}
         TEST_PARALLEL: ${{ inputs.parallel }}
-        TEST_EXT_ARGS: '--skip_file tpcds_spill_1.test,tpcds_spill_2.test,tpcds_spill_3.test'
       run: bash ./scripts/ci/ci-run-sqllogic-tests-cluster.sh ${{ inputs.dirs }}

.github/actions/test_sqllogic_standalone_linux/action.yml

@@ -29,20 +29,12 @@ runs:
       with:
         artifacts: sqllogictests,meta,query
 
-    - uses: actions/github-script@v7
-      id: ext-args
-      env:
-        DIRS: ${{ inputs.dirs }}
-      with:
-        script: require('.github/actions/test_sqllogic_standalone_linux/script.js')(core)
-
     - name: Run sqllogic Tests with Standalone mode
       if: inputs.storage-format == 'all' || inputs.storage-format == 'parquet'
       shell: bash
       env:
         TEST_HANDLERS: ${{ inputs.handlers }}
         TEST_PARALLEL: ${{ inputs.parallel }}
-        TEST_EXT_ARGS: ${{ steps.ext-args.outputs.parquet }}
         CACHE_ENABLE_TABLE_META_CACHE: ${{ inputs.enable_table_meta_cache}}
       run: bash ./scripts/ci/ci-run-sqllogic-tests.sh ${{ inputs.dirs }}
 

.github/actions/test_sqllogic_standalone_linux/script.js

@@ -60,7 +60,7 @@ jobs:
           - { dirs: "crdb", runner: "2c8g" }
           - { dirs: "base", runner: "2c8g" }
           - { dirs: "ydb", runner: "2c8g" }
-          - { dirs: "tpcds", runner: "4c16g", parallel: "1" }
+          - { dirs: "tpcds", runner: "2c8g" }
           - { dirs: "tpch", runner: "2c8g" }
           - { dirs: "standalone", runner: "2c8g" }
         handler:

.github/workflows/reuse.sqllogic.yml

@@ -17,6 +17,7 @@ use std::sync::Arc;
 use std::time::Duration;
 use std::time::Instant;
 
+use regex::Regex;
 use reqwest::header::HeaderMap;
 use reqwest::header::HeaderValue;
 use reqwest::Client;
@@ -41,12 +42,43 @@ pub struct HttpClient {
 #[derive(serde::Deserialize, Debug)]
 struct QueryResponse {
     session: Option<HttpSessionConf>,
+    schema: Vec<SchemaItem>,
     data: Option<serde_json::Value>,
     next_uri: Option<String>,
 
     error: Option<serde_json::Value>,
 }
 
+#[derive(serde::Deserialize, Debug)]
+struct SchemaItem {
+    #[allow(dead_code)]
+    pub name: String,
+    pub r#type: String,
+}
+
+impl SchemaItem {
+    fn parse_type(&self) -> Result<DefaultColumnType> {
+        let nullable = Regex::new(r"^Nullable\((.+)\)$").unwrap();
+        let value = match nullable.captures(&self.r#type) {
+            Some(captures) => {
+                let (_, [value]) = captures.extract();
+                value
+            }
+            None => &self.r#type,
+        };
+        let typ = match value {
+            "String" => DefaultColumnType::Text,
+            "Int8" | "Int16" | "Int32" | "Int64" | "UInt8" | "UInt16" | "UInt32" | "UInt64" => {
+                DefaultColumnType::Integer
+            }
+            "Float32" | "Float64" => DefaultColumnType::FloatingPoint,
+            decimal if decimal.starts_with("Decimal") => DefaultColumnType::FloatingPoint,
+            _ => DefaultColumnType::Any,
+        };
+        Ok(typ)
+    }
+}
+
 // make error message the same with ErrorCode::display
 fn format_error(value: serde_json::Value) -> String {
     let value = value.as_object().unwrap();
@@ -125,14 +157,20 @@ impl HttpClient {
 
     pub async fn query(&mut self, sql: &str) -> Result<DBOutput<DefaultColumnType>> {
         let start = Instant::now();
+        let port = self.port;
+        let mut response = self
+            .post_query(sql, &format!("http://127.0.0.1:{port}/v1/query"))
+            .await?;
 
-        let url = format!("http://127.0.0.1:{}/v1/query", self.port);
+        let mut schema = std::mem::take(&mut response.schema);
         let mut parsed_rows = vec![];
-        let mut response = self.post_query(sql, &url).await?;
         self.handle_response(&response, &mut parsed_rows)?;
         while let Some(next_uri) = &response.next_uri {
-            let url = format!("http://127.0.0.1:{}{next_uri}", self.port);
-            let new_response = self.poll_query_result(&url).await?;
+            let url = format!("http://127.0.0.1:{port}{next_uri}");
+            let mut new_response = self.poll_query_result(&url).await?;
+            if schema.is_empty() && !new_response.schema.is_empty() {
+                schema = std::mem::take(&mut new_response.schema);
+            }
             if new_response.next_uri.is_some() {
                 self.handle_response(&new_response, &mut parsed_rows)?;
                 response = new_response;
@@ -143,11 +181,6 @@ impl HttpClient {
         if let Some(error) = response.error {
             return Err(format_error(error).into());
         }
-        // Todo: add types to compare
-        let mut types = vec![];
-        if !parsed_rows.is_empty() {
-            types = vec![DefaultColumnType::Any; parsed_rows[0].len()];
-        }
 
         if self.debug {
             println!(
@@ -156,6 +189,11 @@ impl HttpClient {
             );
         }
 
+        let types = schema
+            .iter()
+            .map(|item| item.parse_type().unwrap_or(DefaultColumnType::Any))
+            .collect();
+
         Ok(DBOutput::Rows {
             types,
             rows: parsed_rows,

tests/sqllogictests/src/client/http_client.rs

@@ -27,6 +27,8 @@ use sqllogictest::default_validator;
 use sqllogictest::parse_file;
 use sqllogictest::DBOutput;
 use sqllogictest::DefaultColumnType;
+use sqllogictest::Location;
+use sqllogictest::QueryExpect;
 use sqllogictest::Record;
 use sqllogictest::Runner;
 use sqllogictest::TestError;
@@ -319,7 +321,6 @@ async fn run_suits(args: SqlLogicTestArgs, client_type: ClientType) -> Result<()
 
             let col_separator = " ";
             let validator = default_validator;
-            let column_validator = default_column_validator;
             let mut runner =
                 Runner::new(|| async { create_databend(&client_type, &file_name).await });
             runner
@@ -328,7 +329,7 @@ async fn run_suits(args: SqlLogicTestArgs, client_type: ClientType) -> Result<()
                     col_separator,
                     validator,
                     sqllogictest::default_normalizer,
-                    column_validator,
+                    default_column_validator,
                 )
                 .await
                 .unwrap();
@@ -354,6 +355,33 @@ async fn run_suits(args: SqlLogicTestArgs, client_type: ClientType) -> Result<()
     Ok(())
 }
 
+fn column_validator(
+    loc: Location,
+    actual: Vec<DefaultColumnType>,
+    expected: Vec<DefaultColumnType>,
+) {
+    let equals = if actual.len() != expected.len() {
+        false
+    } else {
+        actual.iter().zip(expected.iter()).all(|x| {
+            use DefaultColumnType::*;
+            matches!(
+                x,
+                (Text, Text)
+                    | (Integer, Integer)
+                    | (FloatingPoint, FloatingPoint)
+                    | (Any, _)
+                    | (_, Any)
+            )
+        })
+    };
+    if !equals {
+        println!(
+            "warn: column type not match, actual: {actual:?}, expected: {expected:?}, loc: {loc}"
+        );
+    }
+}
+
 async fn run_parallel_async(
     tasks: Vec<impl Future<Output = std::result::Result<Vec<TestError>, TestError>>>,
     num_of_tests: usize,
@@ -399,22 +427,40 @@ async fn run_file_async(
             break;
         }
         // Capture error record and continue to run next records
-        if let Err(e) = runner.run_async(record).await {
-            // Skip query result error in bench
-            if bench
-                && matches!(
-                    e.kind(),
-                    sqllogictest::TestErrorKind::QueryResultMismatch { .. }
-                )
-            {
-                continue;
-            }
+        let expected_types = if let Record::Query {
+            loc,
+            expected: QueryExpect::Results { types, .. },
+            ..
+        } = &record
+        {
+            Some((loc.clone(), types.clone()))
+        } else {
+            None
+        };
+
+        match (runner.run_async(record).await, expected_types) {
+            (
+                Ok(sqllogictest::RecordOutput::Query { types: actual, .. }),
+                Some((loc, expected)),
+            ) => column_validator(loc, actual, expected),
+            (Err(e), _) => {
+                // Skip query result error in bench
+                if bench
+                    && matches!(
+                        e.kind(),
+                        sqllogictest::TestErrorKind::QueryResultMismatch { .. }
+                    )
+                {
+                    continue;
+                }
 
-            if no_fail_fast {
-                error_records.push(e);
-            } else {
-                return Err(e);
+                if no_fail_fast {
+                    error_records.push(e);
+                } else {
+                    return Err(e);
+                }
             }
+            _ => {}
         }
     }
     let run_file_status = match error_records.is_empty() {

tests/sqllogictests/src/main.rs

@@ -1,5 +1,5 @@
 # Q1
-query I
+query T
 WITH customer_total_return AS
   (SELECT sr_customer_sk AS ctr_customer_sk,
           sr_store_sk AS ctr_store_sk,

tests/sqllogictests/suites/tpcds/Q1

@@ -1,5 +1,5 @@
 # Q10
-query I
+query TTTIIITIIIIIII
 SELECT cd_gender,
        cd_marital_status,
        cd_education_status,

tests/sqllogictests/suites/tpcds/Q10

@@ -1,5 +1,5 @@
 # Q11
-query I
+query TTTT
 WITH year_total AS
   (SELECT c_customer_id customer_id,
           c_first_name customer_first_name,

tests/sqllogictests/suites/tpcds/Q11

@@ -1,5 +1,5 @@
 # Q12
-query I
+query TTTTRRR
 SELECT i_item_id,
        i_item_desc,
        i_category,