fix dtl interm result leak when use vectorized spf and px batch rescan

src/sql/dtl/ob_dtl_interm_result_manager.cpp

@@ -396,6 +396,7 @@ int ObDTLIntermResultManager::generate_monitor_info_rows(observer::ObDTLIntermRe
   if (OB_FAIL(map_.foreach_refactored(monitor_info_getter))) {
     LOG_WARN("fail to generate monitor info array from map", K(ret));
   }
+  LOG_TRACE("generate monitor info rows", K(ret), K(map_.size()));
   return ret;
 }
 

src/sql/engine/px/ob_px_coord_op.cpp

@@ -132,7 +132,8 @@ ObPxCoordOp::ObPxCoordOp(ObExecContext &exec_ctx, const ObOpSpec &spec, ObOpInpu
   px_dop_(1),
   time_recorder_(0),
   batch_rescan_param_version_(0),
-  server_alive_checker_(coord_info_.dfo_mgr_, exec_ctx.get_my_session()->get_process_query_time())
+  server_alive_checker_(coord_info_.dfo_mgr_, exec_ctx.get_my_session()->get_process_query_time()),
+  last_px_batch_rescan_size_(0)
 {}
 
 
@@ -889,6 +890,7 @@ int ObPxCoordOp::receive_channel_root_dfo(
         if (enable_px_batch_rescan()) {
           ch->set_interm_result(true);
           ch->set_batch_id(get_batch_id());
+          last_px_batch_rescan_size_ = max(get_batch_id() + 1, get_rescan_param_count());
         }
       }
       LOG_TRACE("link qc-task channel and registered to qc msg loop. ready to receive task data msg",
@@ -961,6 +963,7 @@ int ObPxCoordOp::receive_channel_root_dfo(
         if (enable_px_batch_rescan()) {
           ch->set_interm_result(true);
           ch->set_batch_id(get_batch_id());
+          last_px_batch_rescan_size_ = max(get_batch_id() + 1, get_rescan_param_count());
         }
       }
       LOG_TRACE("link qc-task channel and registered to qc msg loop. ready to receive task data msg",
@@ -1049,12 +1052,13 @@ int ObPxCoordOp::erase_dtl_interm_result()
         LOG_WARN("fail get channel info", K(ret));
       } else {
         key.channel_id_ = ci.chid_;
-        for (int j = 0; j < max(get_batch_id() + 1, get_rescan_param_count()); ++j) {
+        for (int j = 0; j < last_px_batch_rescan_size_; ++j) {
           key.batch_id_ = j;
           if (OB_FAIL(ObDTLIntermResultManager::getInstance().erase_interm_result_info(key))) {
             LOG_TRACE("fail to release recieve internal result", K(ret));
           }
         }
+        last_px_batch_rescan_size_ = 0;
       }
     }
   }

src/sql/engine/px/ob_px_coord_op.h

@@ -159,6 +159,7 @@ protected:
   int64_t time_recorder_;
   int64_t batch_rescan_param_version_;
   ObExtraServerAliveCheck server_alive_checker_;
+  int64_t last_px_batch_rescan_size_;
 };
 
 class ObPxCoordSpec : public ObPxReceiveSpec