fix tiflash assert failure (#9456)

dbms/src/Flash/Planner/Plans/PhysicalJoin.cpp

@@ -238,6 +238,7 @@ void PhysicalJoin::probeSideTransform(DAGPipeline & probe_pipeline, Context & co
             settings.max_block_size);
         stream->setExtraInfo(join_probe_extra_info);
     }
+    join_ptr->setCancellationHook([&] { return context.isCancelled(); });
 }
 
 void PhysicalJoin::buildSideTransform(DAGPipeline & build_pipeline, Context & context, size_t max_streams)

dbms/src/Flash/Planner/Plans/PhysicalJoinProbe.cpp

@@ -56,6 +56,14 @@ void PhysicalJoinProbe::buildPipelineExecGroupImpl(
             max_block_size,
             input_header));
     });
+<<<<<<< HEAD
+=======
+    // The `join_ptr->wait_build_finished_future` does not need to be added to exec_context here;
+    // it is only necessary to add it during the "restore build stage."
+    // The order of build/probe here is ensured by the event.
+    exec_context.addOneTimeFuture(join_ptr->wait_probe_finished_future);
+    join_ptr->setCancellationHook([&]() { return exec_context.isCancelled(); });
+>>>>>>> 78bd3f04dc (fix tiflash assert failure  (#9456))
     join_ptr.reset();
 }
 } // namespace DB

dbms/src/Interpreters/Join.cpp

@@ -1352,6 +1352,10 @@ Block Join::doJoinBlockHash(ProbeProcessInfo & probe_process_info) const
 
 Block Join::removeUselessColumn(Block & block) const
 {
+    // cancelled
+    if (!block)
+        return block;
+
     Block projected_block;
     for (const auto & name : tidb_output_column_names)
     {
@@ -2004,6 +2008,9 @@ Block Join::joinBlock(ProbeProcessInfo & probe_process_info, bool dry_run) const
     else
         block = joinBlockHash(probe_process_info);
 
+    // if cancelled, just return empty block
+    if (!block)
+        return block;
     /// for (cartesian)antiLeftSemi join, the meaning of "match-helper" is `non-matched` instead of `matched`.
     if (kind == LeftOuterAnti || kind == Cross_LeftOuterAnti)
     {