MDEV-16791 mariabackup : Support DDL commands during backup

Author: vaintroub

extra/mariabackup/backup_copy.cc

@@ -1388,6 +1388,30 @@ backup_files(const char *from, bool prep_mode)
 	return(ret);
 }
 
+void backup_fix_ddl(void);
+
+#define LSN_PREFIX_IN_SHOW_STATUS  "\nLog sequence number "
+static lsn_t get_current_lsn(MYSQL *connection) {
+	MYSQL_RES *res = xb_mysql_query(connection, "SHOW ENGINE INNODB STATUS", true, false);
+	if (!res)
+		return 0;
+	MYSQL_ROW row = mysql_fetch_row(res);
+	DBUG_ASSERT(row);
+	if (row) {
+		const char *p = strstr(row[2],LSN_PREFIX_IN_SHOW_STATUS);
+		DBUG_ASSERT(p);
+		if (p)
+		{
+			p += sizeof(LSN_PREFIX_IN_SHOW_STATUS) - 1;
+			return (lsn_t)strtoll(p, NULL, 10);
+		}
+	}
+	mysql_free_result(res);
+	return 0;
+}
+
+lsn_t server_lsn_after_lock;
+extern void backup_wait_for_lsn(lsn_t lsn);
 /** Start --backup */
 bool backup_start()
 {
@@ -1407,6 +1431,7 @@ bool backup_start()
 		if (!lock_tables(mysql_connection)) {
 			return(false);
 		}
+		server_lsn_after_lock = get_current_lsn(mysql_connection);
 	}
 
 	if (!backup_files(fil_path_to_mysql_datadir, false)) {
@@ -1421,6 +1446,10 @@ bool backup_start()
 		rocksdb_create_checkpoint();
 	}
 
+	msg_ts("Waiting for log copy thread to read lsn %llu\n", (ulonglong)server_lsn_after_lock);
+	backup_wait_for_lsn(server_lsn_after_lock);
+	backup_fix_ddl();
+
 	// There is no need to stop slave thread before coping non-Innodb data when
 	// --no-lock option is used because --no-lock option requires that no DDL or
 	// DML to non-transaction tables can occur.
@@ -2230,6 +2259,7 @@ static void rocksdb_lock_checkpoint()
 		msg_ts("Could not obtain rocksdb checkpont lock\n");
 		exit(EXIT_FAILURE);
 	}
+	mysql_free_result(res);
 }
 
 static void rocksdb_unlock_checkpoint()

extra/mariabackup/backup_mysql.cc

@@ -1794,7 +1794,12 @@ mdl_lock_table(ulint space_id)
     std::ostringstream lock_query;
     lock_query << "SELECT 1 FROM " << full_table_name  << " LIMIT 0";
     msg_ts("Locking MDL for %s\n", full_table_name.c_str());
-    xb_mysql_query(mdl_con, lock_query.str().c_str(), false, true);
+    if (mysql_query(mdl_con, lock_query.str().c_str())) {
+      msg_ts("Warning : locking MDL failed for space id %zu, name %s\n", space_id, full_table_name.c_str());
+    } else {
+      MYSQL_RES *r = mysql_store_result(mdl_con);
+      mysql_free_result(r);
+    }
   }
 
   pthread_mutex_unlock(&mdl_lock_con_mutex);

extra/mariabackup/datasink.c

@@ -108,6 +108,9 @@ Write to a datasink file.
 int
 ds_write(ds_file_t *file, const void *buf, size_t len)
 {
+	if (len == 0) {
+		return 0;
+	}
 	return file->datasink->write(file, (const uchar *)buf, len);
 }
 

extra/mariabackup/fil_cur.cc

@@ -131,14 +131,15 @@ Open a source file cursor and initialize the associated read filter.
 be skipped and XB_FIL_CUR_ERROR on error. */
 xb_fil_cur_result_t
 xb_fil_cur_open(
-/*============*/
+	/*============*/
 	xb_fil_cur_t*	cursor,		/*!< out: source file cursor */
 	xb_read_filt_t*	read_filter,	/*!< in/out: the read filter */
 	fil_node_t*	node,		/*!< in: source tablespace node */
-	uint		thread_n)	/*!< thread number for diagnostics */
+	uint		thread_n,	/*!< thread number for diagnostics */
+	ulonglong max_file_size)
 {
 	bool	success;
-
+	int err;
 	/* Initialize these first so xb_fil_cur_close() handles them correctly
 	in case of error */
 	cursor->orig_buf = NULL;
@@ -173,7 +174,7 @@ xb_fil_cur_open(
 			    "tablespace %s\n",
 			    thread_n, cursor->abs_path);
 
-			return(XB_FIL_CUR_ERROR);
+			return(XB_FIL_CUR_SKIP);
 		}
 		mutex_enter(&fil_system->mutex);
 
@@ -194,14 +195,31 @@ xb_fil_cur_open(
 
 	cursor->node = node;
 	cursor->file = node->handle;
-
-	if (stat(cursor->abs_path, &cursor->statinfo)) {
-		msg("[%02u] mariabackup: error: cannot stat %s\n",
+#ifdef _WIN32
+    HANDLE hDup;
+    DuplicateHandle(GetCurrentProcess(),cursor->file.m_file,
+		GetCurrentProcess(), &hDup, 0, FALSE, DUPLICATE_SAME_ACCESS);
+	int filenr = _open_osfhandle((intptr_t)hDup, 0);
+	if (filenr < 0) {
+		err = EINVAL;
+	}
+	else {
+		err = _fstat64(filenr, &cursor->statinfo);
+		close(filenr);
+	}
+#else
+	err = fstat(cursor->file.m_file, &cursor->statinfo);
+#endif
+	if (max_file_size < (ulonglong)cursor->statinfo.st_size) {
+		cursor->statinfo.st_size = (ulonglong)max_file_size;
+	}
+	if (err) {
+		msg("[%02u] mariabackup: error: cannot fstat %s\n",
 		    thread_n, cursor->abs_path);
 
 		xb_fil_cur_close(cursor);
 
-		return(XB_FIL_CUR_ERROR);
+		return(XB_FIL_CUR_SKIP);
 	}
 
 	if (srv_file_flush_method == SRV_O_DIRECT
@@ -374,7 +392,9 @@ xb_fil_cur_close(
 /*=============*/
 	xb_fil_cur_t *cursor)	/*!< in/out: source file cursor */
 {
-	cursor->read_filter->deinit(&cursor->read_filter_ctxt);
+	if (cursor->read_filter) {
+		cursor->read_filter->deinit(&cursor->read_filter_ctxt);
+	}
 
 	free(cursor->orig_buf);
 

extra/mariabackup/fil_cur.h

@@ -57,7 +57,7 @@ struct xb_fil_cur_t {
 	ulint		space_size;	/*!< space size in pages */
 
 	/** TODO: remove this default constructor */
-	xb_fil_cur_t() : page_size(0), read_filter_ctxt() {}
+	xb_fil_cur_t() : page_size(0), read_filter(0), read_filter_ctxt() {}
 
 	/** @return whether this is not a file-per-table tablespace */
 	bool is_system() const
@@ -86,7 +86,8 @@ xb_fil_cur_open(
 	xb_fil_cur_t*	cursor,		/*!< out: source file cursor */
 	xb_read_filt_t*	read_filter,	/*!< in/out: the read filter */
 	fil_node_t*	node,		/*!< in: source tablespace node */
-	uint		thread_n);	/*!< thread number for diagnostics */
+	uint		thread_n,	/*!< thread number for diagnostics */
+	ulonglong max_file_size = ULLONG_MAX);
 
 /************************************************************************
 Reads and verifies the next block of pages from the source