From bc4c4a371d601bebdaa8f05046175640c18611aa Mon Sep 17 00:00:00 2001
From: obdev <obdev@oceanbase.com>
Date: Fri, 9 Feb 2024 08:31:09 +0000
Subject: [PATCH] Fix bug in multipart upload in plus archive mode

---
 src/storage/backup/ob_backup_task.cpp | 50 +++++++++++++++------------
 src/storage/backup/ob_backup_task.h   |  3 +-
 2 files changed, 29 insertions(+), 24 deletions(-)

diff --git a/src/storage/backup/ob_backup_task.cpp b/src/storage/backup/ob_backup_task.cpp
index d223fa0a8..cf59b52bb 100644
--- a/src/storage/backup/ob_backup_task.cpp
+++ b/src/storage/backup/ob_backup_task.cpp
@@ -5591,41 +5591,49 @@ int ObLSBackupComplementLogTask::inner_backup_complement_log_(
 int ObLSBackupComplementLogTask::transfer_clog_file_(const ObBackupPath &src_path, const ObBackupPath &dst_path)
 {
   int ret = OB_SUCCESS;
+  int tmp_ret = OB_SUCCESS;
+  int64_t dst_len = 0;
   int64_t transfer_len = 0;
-  while (OB_SUCC(ret)) {
-    if (OB_FAIL(inner_transfer_clog_file_(src_path, dst_path, transfer_len))) {
-      LOG_WARN("failed to inner transfer clog file", K(ret), K(src_path), K(dst_path));
+  ObIOFd fd;
+  ObBackupIoAdapter util;
+  ObIODevice *device_handle = NULL;
+  if (OB_FAIL(util.open_with_access_type(
+          device_handle, fd, backup_dest_.get_storage_info(), dst_path.get_obstr(), OB_STORAGE_ACCESS_MULTIPART_WRITER))) {
+    LOG_WARN("failed to open with access type", K(ret));
+  } else {
+    while (OB_SUCC(ret)) {
+      if (OB_FAIL(inner_transfer_clog_file_(src_path, dst_path, device_handle, fd, dst_len, transfer_len))) {
+        LOG_WARN("failed to inner transfer clog file", K(ret), K(src_path), K(dst_path));
+      } else {
+        dst_len += transfer_len;
+      }
+      if (0 == transfer_len) { //at this point, last part is still held in memory
+        LOG_INFO("transfer ended", K(ret), K(src_path), K(dst_path));
+        break;
+      }
     }
-    if (0 == transfer_len) {
-      LOG_INFO("transfer ended", K(ret), K(src_path), K(dst_path));
-      break;
+
+    if (OB_SUCCESS != (tmp_ret = util.close_device_and_fd(device_handle, fd))) {
+      LOG_WARN("fail to close file", K(ret), K_(backup_dest), K(dst_path));
+      ret = OB_SUCCESS == ret ? tmp_ret : ret;
     }
   }
   return ret;
 }
 
-int ObLSBackupComplementLogTask::inner_transfer_clog_file_(
-    const ObBackupPath &src_path, const ObBackupPath &dst_path, int64_t &transfer_len)
+int ObLSBackupComplementLogTask::inner_transfer_clog_file_(const ObBackupPath &src_path, const ObBackupPath &dst_path,
+    ObIODevice *&device_handle, ObIOFd &fd, const int64_t dst_len, int64_t &transfer_len)
 {
   int ret = OB_SUCCESS;
-  int tmp_ret = OB_SUCCESS;
   transfer_len = 0;
-  ObIOFd fd;
   ObBackupIoAdapter util;
-  ObIODevice *device_handle = NULL;
   int64_t write_size = -1;
   ObArenaAllocator allocator;
   int64_t src_len = 0;
-  int64_t dst_len = 0;
   char *buf = NULL;
   int64_t read_len = 0;
-  if (OB_FAIL(util.open_with_access_type(
-          device_handle, fd, backup_dest_.get_storage_info(), dst_path.get_obstr(), OB_STORAGE_ACCESS_MULTIPART_WRITER))) {
-    LOG_WARN("failed to open with access type", K(ret));
-  } else if (OB_FAIL(get_file_length_(src_path.get_obstr(), archive_dest_.get_storage_info(), src_len))) {
+  if (OB_FAIL(get_file_length_(src_path.get_obstr(), archive_dest_.get_storage_info(), src_len))) {
     LOG_WARN("failed to get file length", K(ret), K(src_path));
-  } else if (OB_FAIL(get_file_length_(dst_path.get_obstr(), backup_dest_.get_storage_info(), dst_len))) {
-    LOG_WARN("failed to get file length", K(ret), K(dst_path));
   } else if (dst_len == src_len) {
     transfer_len = 0;
   } else if (dst_len > src_len) {
@@ -5642,11 +5650,7 @@ int ObLSBackupComplementLogTask::inner_transfer_clog_file_(
     ret = OB_ERR_UNEXPECTED;
     LOG_WARN("read len not expected", K(ret), K(read_len), K(transfer_len));
   } else if (OB_FAIL(device_handle->pwrite(fd, dst_len, transfer_len, buf, write_size))) {
-    LOG_WARN("failed to write appender file", K(ret));
-  }
-  if (OB_SUCCESS != (tmp_ret = util.close_device_and_fd(device_handle, fd))) {
-    LOG_WARN("failed to close storage appender", K(ret), KR(tmp_ret));
-    ret = OB_SUCCESS == ret ? tmp_ret : ret;
+    LOG_WARN("failed to write multipart upload file", K(ret));
   }
   return ret;
 }
diff --git a/src/storage/backup/ob_backup_task.h b/src/storage/backup/ob_backup_task.h
index 1acc2c091..cd7ae5953 100644
--- a/src/storage/backup/ob_backup_task.h
+++ b/src/storage/backup/ob_backup_task.h
@@ -721,7 +721,8 @@ private:
   int backup_complement_log_(const common::ObIArray<BackupPieceFile> &path);
   int inner_backup_complement_log_(const share::ObBackupPath &src_path, const share::ObBackupPath &dst_path);
   int transfer_clog_file_(const share::ObBackupPath &src_path, const share::ObBackupPath &dst_path);
-  int inner_transfer_clog_file_(const share::ObBackupPath &src_path, const share::ObBackupPath &dst_path, int64_t &transfer_len);
+  int inner_transfer_clog_file_(const ObBackupPath &src_path, const ObBackupPath &dst_path,
+      ObIODevice *&device_handle, ObIOFd &fd, const int64_t dst_len, int64_t &transfer_len);
   int get_transfer_length_(const int64_t delta_len, int64_t &transfer_len);
   int get_file_length_(const common::ObString &path, const share::ObBackupStorageInfo *storage_info, int64_t &length);
   int get_copy_src_and_dest_(const BackupPieceFile &piece_file, share::ObBackupDest &src, share::ObBackupDest &dest);