refactor(query): refactor hash join spill (#15746)

* chore: refactor HashJoinBuildState

* chore: refactor HashJoinProbeState

* chore: refactor HashJoinState

* refactor: unify hash join spiller and optimize partition

* refactor: improve SpillBuffer

* refactor: new step transition mechanism.

* chore: merge main

* chore: fix fast return

* chore: make lint

* chore: fix WaitCollect and remove useless code

* chore: fix unsplitted data blocks

* chore:  fix add_splitted_data_blocks

* chore: fix add_splitted_data_blocks

* chore(code): fix typos

* chore(query): refine spill buffer and some code

* chore(query): minimum threshold for each partition

* chore(query): update minimum threshold of partition

* chore(query): fix can probe first round

* chore(query): improve HashJoinSpiller

* chore(query): use 1024 instead of 1000 in SpillBuffer

* chore(query): refine comment

* chore(query): refine spill buffer partition_threshold

* chore(test): add tpch join spill test

* chore(test): add more test

* chore(query): refine code

* chore(query): remove probe first round

* chore(query): fix is_left_related_join_type

* chore(test): refine test

* chore(test): refine test

* chore(query): fix can_fast_return

* chore(query): fix cross join spill

* chore(test): refine tpch test

* chore(test): add tpch spill test

* chore(test): add tpcds spill test

* chore(test): add tpch spill test

* chore(test): add tpcds spill test

* chore(test): disable tpcds aggregate spill test

* chore(test): disable aggregate in tpch spill test

* chore(query): disable aggregate spill by default

* chore(query): support disable spill

* chore(query): stricter atomic ordering

* chore(query): refine atomic ordering

* chore(query): remove unused code

* chore(query): add #[allow(unused)]

Author: Dousir9

src/query/service/src/pipelines/builders/builder_join.rs

@@ -28,12 +28,10 @@ use databend_common_sql::IndexType;
 use crate::pipelines::processors::transforms::range_join::RangeJoinState;
 use crate::pipelines::processors::transforms::range_join::TransformRangeJoinLeft;
 use crate::pipelines::processors::transforms::range_join::TransformRangeJoinRight;
-use crate::pipelines::processors::transforms::BuildSpillState;
 use crate::pipelines::processors::transforms::HashJoinBuildState;
 use crate::pipelines::processors::transforms::HashJoinProbeState;
 use crate::pipelines::processors::transforms::MaterializedCteSink;
 use crate::pipelines::processors::transforms::MaterializedCteState;
-use crate::pipelines::processors::transforms::ProbeSpillState;
 use crate::pipelines::processors::transforms::TransformHashJoinBuild;
 use crate::pipelines::processors::transforms::TransformHashJoinProbe;
 use crate::pipelines::processors::HashJoinDesc;
@@ -161,19 +159,9 @@ impl PipelineBuilder {
         )?;
 
         let create_sink_processor = |input| {
-            let spill_state = if join_state.enable_spill {
-                Some(Box::new(BuildSpillState::create(
-                    self.ctx.clone(),
-                    build_state.clone(),
-                )?))
-            } else {
-                None
-            };
-
             Ok(ProcessorPtr::create(TransformHashJoinBuild::try_create(
                 input,
                 build_state.clone(),
-                spill_state,
             )?))
         };
         // for distributed merge into when source as build side.
@@ -192,7 +180,6 @@ impl PipelineBuilder {
 
         let max_block_size = self.settings.get_max_block_size()? as usize;
         let barrier = Barrier::new(self.main_pipeline.output_len());
-        let restore_barrier = Barrier::new(self.main_pipeline.output_len());
         let probe_state = Arc::new(HashJoinProbeState::create(
             self.ctx.clone(),
             self.func_ctx.clone(),
@@ -203,29 +190,18 @@ impl PipelineBuilder {
             &join.join_type,
             self.main_pipeline.output_len(),
             barrier,
-            restore_barrier,
         )?);
         let mut has_string_column = false;
         for field in join.output_schema()?.fields() {
             has_string_column |= field.data_type().is_string_column();
         }
 
         self.main_pipeline.add_transform(|input, output| {
-            let probe_spill_state = if state.enable_spill {
-                Some(Box::new(ProbeSpillState::create(
-                    self.ctx.clone(),
-                    probe_state.clone(),
-                )?))
-            } else {
-                None
-            };
-
             Ok(ProcessorPtr::create(TransformHashJoinProbe::create(
                 input,
                 output,
                 join.projections.clone(),
                 probe_state.clone(),
-                probe_spill_state,
                 max_block_size,
                 self.func_ctx.clone(),
                 &join.join_type,