[CP] Fix win buf mem leak

src/sql/engine/px/datahub/components/ob_dh_winbuf.h

@@ -108,6 +108,10 @@ public:
       : ObPieceMsgCtx(op_id, task_cnt, timeout_ts), received_(0), tenant_id_(tenant_id), whole_msg_()
   {}
   ~ObWinbufPieceMsgCtx() = default;
+  virtual void destroy()
+  {
+    whole_msg_.reset();
+  }
   INHERIT_TO_STRING_KV("meta", ObPieceMsgCtx, K_(received));
   static int alloc_piece_msg_ctx(
       const ObWinbufPieceMsg& pkt, ObExecContext& ctx, int64_t task_cnt, ObPieceMsgCtx*& msg_ctx);

src/sql/engine/px/datahub/ob_dh_msg_ctx.h

@@ -23,6 +23,10 @@ public:
   ObPieceMsgCtx(uint64_t op_id, int64_t task_cnt, int64_t timeout_ts)
       : op_id_(op_id), task_cnt_(task_cnt), timeout_ts_(timeout_ts)
   {}
+  virtual ~ObPieceMsgCtx()
+  {}
+  virtual void destroy()
+  {}
   VIRTUAL_TO_STRING_KV(K_(op_id), K_(task_cnt));
   uint64_t op_id_;
   int64_t task_cnt_;
@@ -35,6 +39,12 @@ public:
   ~ObPieceMsgCtxMgr() = default;
   void reset()
   {
+    for (int i = 0; i < ctxs_.count(); ++i) {
+      if (OB_NOT_NULL(ctxs_[i])) {
+        ctxs_[i]->destroy();
+        ctxs_[i]->~ObPieceMsgCtx();
+      }
+    }
     ctxs_.reset();
   }
   int find_piece_ctx(uint64_t op_id, ObPieceMsgCtx*& ctx)