fix: tpcds spill test fail (#17597)

* splite queries.test

* fix

* update

* x

* x

* x

* x

* x

* x

* x

* xx

* x

* x

* x

* x

Author: forsaken628

.github/actions/test_sqllogic_cluster_linux/action.yml

@@ -9,6 +9,11 @@ inputs:
     description: "logic test handlers, mysql,http,clickhouse"
     required: false
     default: ""
+  parallel:
+    description: "logic test parallel"
+    required: false
+    default: ""
+
 runs:
   using: "composite"
   steps:
@@ -20,4 +25,6 @@ runs:
       shell: bash
       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_stage/action.yml

@@ -13,7 +13,7 @@ inputs:
     description: "storage backend for stage, choices: s3,fs"
     required: true
     default: ""
-  deducp:
+  dedup:
     description: "path type for dedup when copy, choices: full_path,sub_path"
     required: true
     default: ""

.github/actions/test_sqllogic_standalone_linux/action.yml

@@ -29,12 +29,20 @@ 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 }}
 
@@ -44,5 +52,6 @@ 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'
         CACHE_ENABLE_TABLE_META_CACHE: ${{ inputs.enable_table_meta_cache}}
       run: bash ./scripts/ci/ci-run-sqllogic-tests-native.sh ${{ inputs.dirs }}

.github/actions/test_sqllogic_standalone_linux/script.js

@@ -0,0 +1,12 @@
+module.exports = (core) => {
+    switch (process.env.DIRS) {
+        case 'tpcds':
+            const parquet = [
+                '--skip_file tpcds_spill_2.test,tpcds_spill_3.test',
+                '--skip_file tpcds_spill_1.test,tpcds_spill_3.test',
+                '--skip_file tpcds_spill_1.test,tpcds_spill_2.test',
+            ][Date.now() % 3];
+            core.setOutput('parquet', parquet)
+            return
+    }
+}

.github/workflows/reuse.sqllogic.yml

@@ -50,7 +50,7 @@ jobs:
           - { dirs: "crdb", runner: "2c8g" }
           - { dirs: "base", runner: "2c8g" }
           - { dirs: "ydb", runner: "2c8g" }
-          - { dirs: "tpcds", runner: "2c8g" }
+          - { dirs: "tpcds", runner: "4c16g", parallel: "1" }
           - { dirs: "tpch", runner: "2c8g" }
           - { dirs: "standalone", runner: "2c8g" }
         handler:
@@ -59,10 +59,11 @@ jobs:
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/test_sqllogic_standalone_linux
-        timeout-minutes: 15
+        timeout-minutes: 20
         with:
           dirs: ${{ matrix.tests.dirs }}
           handlers: ${{ matrix.handler }}
+          parallel: ${{ matrix.tests.parallel }}
           storage-format: all
       - name: Upload failure
         if: failure()
@@ -192,6 +193,7 @@ jobs:
         with:
           dirs: ${{ matrix.tests.dirs }}
           handlers: ${{ matrix.handler }}
+          parallel: ${{ matrix.tests.parallel }}
       - name: Upload failure
         if: failure()
         uses: ./.github/actions/artifact_failure

scripts/ci/ci-run-sqllogic-tests-cluster.sh

@@ -22,4 +22,4 @@ fi
 echo "Run suites using argument: $RUN_DIR"
 
 echo "Starting databend-sqllogic tests"
-target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --enable_sandbox --parallel ${TEST_PARALLEL} --skip_file tpcds_q64.test,tpcds_join_order.test
+target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --enable_sandbox --parallel ${TEST_PARALLEL} ${TEST_EXT_ARGS}

scripts/ci/ci-run-sqllogic-tests-native.sh

@@ -20,4 +20,4 @@ fi
 echo "Run suites using argument: $RUN_DIR"
 
 echo "Starting databend-sqllogic tests"
-target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,cluster,explain,tpch,ee --enable_sandbox --parallel ${TEST_PARALLEL}
+target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,cluster,explain,tpch,ee --enable_sandbox --parallel ${TEST_PARALLEL} ${TEST_EXT_ARGS}

scripts/ci/ci-run-sqllogic-tests.sh

@@ -21,6 +21,6 @@ echo "Run suites using argument: $RUN_DIR"
 
 echo "Starting databend-sqllogic tests"
 if [ -z "$RUN_DIR" ]; then
-	target/${BUILD_PROFILE}/databend-sqllogictests --run_dir temp_table --enable_sandbox --parallel ${TEST_PARALLEL}
+	target/${BUILD_PROFILE}/databend-sqllogictests --run_dir temp_table --enable_sandbox --parallel ${TEST_PARALLEL} ${TEST_EXT_ARGS}
 fi
-target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,explain_native,ee,temp_table --enable_sandbox --parallel ${TEST_PARALLEL}
+target/${BUILD_PROFILE}/databend-sqllogictests --handlers ${TEST_HANDLERS} ${RUN_DIR} --skip_dir management,explain_native,ee,temp_table --enable_sandbox --parallel ${TEST_PARALLEL} ${TEST_EXT_ARGS}

src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_probe_state.rs

@@ -114,11 +114,14 @@ impl HashJoinProbeState {
         }
         let hash_key_types = probe_keys
             .iter()
-            .map(|expr| {
-                expr.as_expr(&BUILTIN_FUNCTIONS)
-                    .data_type()
-                    .remove_nullable()
-                    .clone()
+            .zip(&hash_join_state.hash_join_desc.is_null_equal)
+            .map(|(expr, is_null_equal)| {
+                let expr = expr.as_expr(&BUILTIN_FUNCTIONS);
+                if *is_null_equal {
+                    expr.data_type().clone()
+                } else {
+                    expr.data_type().remove_nullable()
+                }
             })
             .collect::<Vec<_>>();
         let method = DataBlock::choose_hash_method_with_types(&hash_key_types)?;

src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_spiller.rs

@@ -40,6 +40,7 @@ pub struct HashJoinSpiller {
     partition_buffer: PartitionBuffer,
     partition_threshold: usize,
     join_type: JoinType,
+    join_state: Arc<HashJoinState>,
     is_build_side: bool,
     func_ctx: FunctionContext,
     /// Used for partition.
@@ -95,6 +96,7 @@ impl HashJoinSpiller {
             hash_keys,
             hash_method,
             join_type,
+            join_state,
             func_ctx: ctx.get_function_context()?,
             is_build_side,
             next_restore_file: 0,
@@ -235,6 +237,7 @@ impl HashJoinSpiller {
             &self.hash_method,
             join_type,
             self.is_build_side,
+            &self.join_state.hash_join_desc.is_null_equal,
             &mut hashes,
         )?;
         Ok(hashes)