From 782e59f78f8372168cb6e8c78db4c5ecabdfcced Mon Sep 17 00:00:00 2001
From: obdev <obdev@oceanbase.com>
Date: Wed, 26 Apr 2023 20:22:44 +0800
Subject: [PATCH] [CP] fix ddl wrongly relying on macro_block_buf

---
 src/storage/ddl/ob_ddl_clog.cpp            | 22 ++++++++--------------
 src/storage/ddl/ob_ddl_redo_log_writer.cpp | 11 +++++++++--
 2 files changed, 17 insertions(+), 16 deletions(-)

diff --git a/src/storage/ddl/ob_ddl_clog.cpp b/src/storage/ddl/ob_ddl_clog.cpp
index b322898bdb..e81aa86e4c 100644
--- a/src/storage/ddl/ob_ddl_clog.cpp
+++ b/src/storage/ddl/ob_ddl_clog.cpp
@@ -156,20 +156,14 @@ int ObDDLMacroBlockClogCb::init(const share::ObLSID &ls_id,
     ret = OB_INVALID_ARGUMENT;
     LOG_WARN("invalid argument", K(ret), K(ls_id), K(redo_info), K(macro_block_id));
   } else {
-    void *buf = nullptr;
-    if (OB_ISNULL(buf = arena_.alloc(redo_info.data_buffer_.length()))) {
-      ret = OB_ALLOCATE_MEMORY_FAILED;
-      LOG_WARN("allocate memory failed", K(ret), K(redo_info.data_buffer_.length()));
-    } else {
-      redo_info_.data_buffer_.assign(const_cast<char *>(redo_info.data_buffer_.ptr()), redo_info.data_buffer_.length());
-      redo_info_.block_type_ = redo_info.block_type_;
-      redo_info_.logic_id_ = redo_info.logic_id_;
-      redo_info_.table_key_ = redo_info.table_key_;
-      redo_info_.start_scn_ = redo_info.start_scn_;
-      ls_id_ = ls_id;
-      macro_block_id_ = macro_block_id;
-      ddl_kv_mgr_handle_ = ddl_kv_mgr_handle;
-    }
+    redo_info_.data_buffer_.assign(const_cast<char *>(redo_info.data_buffer_.ptr()), redo_info.data_buffer_.length());
+    redo_info_.block_type_ = redo_info.block_type_;
+    redo_info_.logic_id_ = redo_info.logic_id_;
+    redo_info_.table_key_ = redo_info.table_key_;
+    redo_info_.start_scn_ = redo_info.start_scn_;
+    ls_id_ = ls_id;
+    macro_block_id_ = macro_block_id;
+    ddl_kv_mgr_handle_ = ddl_kv_mgr_handle;
   }
   return ret;
 }
diff --git a/src/storage/ddl/ob_ddl_redo_log_writer.cpp b/src/storage/ddl/ob_ddl_redo_log_writer.cpp
index ddb93e270c..0d8394ad90 100644
--- a/src/storage/ddl/ob_ddl_redo_log_writer.cpp
+++ b/src/storage/ddl/ob_ddl_redo_log_writer.cpp
@@ -613,6 +613,8 @@ int ObDDLRedoLogWriter::write(
                               + log.get_serialize_size();
   int64_t pos = 0;
   ObDDLMacroBlockClogCb *cb = nullptr;
+  ObDDLRedoLog tmp_log;
+  int64_t log_start_pos = 0;
 
   palf::LSN lsn;
   const bool need_nonblock= false;
@@ -633,14 +635,18 @@ int ObDDLRedoLogWriter::write(
   } else if (OB_ISNULL(cb = op_alloc(ObDDLMacroBlockClogCb))) {
     ret = OB_ALLOCATE_MEMORY_FAILED;
     LOG_WARN("fail to alloc memory", K(ret));
-  } else if (OB_FAIL(cb->init(ls_id, log.get_redo_info(), macro_block_id, ddl_kv_mgr_handle))) {
-    LOG_WARN("init ddl clog callback failed", K(ret));
   } else if (OB_FAIL(base_header.serialize(buffer, buffer_size, pos))) {
     LOG_WARN("failed to serialize log base header", K(ret));
   } else if (OB_FAIL(ddl_header.serialize(buffer, buffer_size, pos))) {
     LOG_WARN("fail to seriaize ddl redo log", K(ret));
+  } else if (FALSE_IT(log_start_pos = pos)) {
   } else if (OB_FAIL(log.serialize(buffer, buffer_size, pos))) {
     LOG_WARN("fail to seriaize ddl redo log", K(ret));
+  } else if (OB_FAIL(tmp_log.deserialize(buffer, buffer_size, log_start_pos))) {
+    LOG_WARN("fail to deserialize ddl redo log", K(ret));
+  /* use the ObString data_buffer_ in tmp_log.redo_info_, do not rely on the macro_block_buf in original log*/
+  } else if (OB_FAIL(cb->init(ls_id, tmp_log.get_redo_info(), macro_block_id, ddl_kv_mgr_handle))) {
+    LOG_WARN("init ddl clog callback failed", K(ret));
   } else if (OB_FAIL(log_handler->append(buffer,
                                          buffer_size,
                                          base_scn,
@@ -1385,6 +1391,7 @@ ObDDLRedoLogWriterCallback::ObDDLRedoLogWriterCallback()
 
 ObDDLRedoLogWriterCallback::~ObDDLRedoLogWriterCallback()
 {
+  (void)wait();
   if (nullptr != block_buffer_) {
     ob_free(block_buffer_);
     block_buffer_ = nullptr;