feat(query): reclaim memory on hash join finish

src/query/service/src/pipelines/processors/transforms/new_hash_join/join.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use databend_common_base::base::ProgressValues;
+use databend_common_exception::ErrorCode;
 use databend_common_exception::Result;
 use databend_common_expression::DataBlock;
 
@@ -66,3 +67,25 @@ impl JoinStream for OneBlockJoinStream {
         Ok(self.0.take())
     }
 }
+
+pub struct FinishedJoin;
+
+impl FinishedJoin {
+    pub fn create() -> Box<dyn Join> {
+        Box::new(FinishedJoin)
+    }
+}
+
+impl Join for FinishedJoin {
+    fn add_block(&mut self, _: Option<DataBlock>) -> Result<()> {
+        Err(ErrorCode::Internal("Join is finished"))
+    }
+
+    fn final_build(&mut self) -> Result<Option<ProgressValues>> {
+        Err(ErrorCode::Internal("Join is finished"))
+    }
+
+    fn probe_block(&mut self, _: DataBlock) -> Result<Box<dyn JoinStream + '_>> {
+        Err(ErrorCode::Internal("Join is finished"))
+    }
+}

src/query/service/src/pipelines/processors/transforms/new_hash_join/transform_hash_join.rs

@@ -31,6 +31,7 @@ use databend_common_pipeline::core::ProcessorPtr;
 use log::info;
 
 use crate::pipelines::processors::transforms::RuntimeFilterLocalBuilder;
+use crate::pipelines::processors::transforms::new_hash_join::join::FinishedJoin;
 use crate::pipelines::processors::transforms::new_hash_join::join::Join;
 use crate::pipelines::processors::transforms::new_hash_join::join::JoinStream;
 use crate::pipelines::processors::transforms::new_hash_join::runtime_filter::RuntimeFiltersDesc;
@@ -106,6 +107,11 @@ impl Processor for TransformHashJoin {
 
             if !matches!(self.stage, Stage::Finished) {
                 self.stage = Stage::Finished;
+
+                let mut finished = FinishedJoin::create();
+                std::mem::swap(&mut finished, &mut self.join);
+                drop(finished);
+
                 self.stage_sync_barrier.reduce_quorum(1);
             }
 
@@ -287,6 +293,11 @@ impl Processor for TransformHashJoin {
                     );
 
                     self.instant = Instant::now();
+
+                    let mut finished = FinishedJoin::create();
+                    std::mem::swap(&mut finished, &mut self.join);
+                    drop(finished);
+
                     Stage::Finished
                 }
                 false => {