Merge 10.2 into 10.3

extra/mariabackup/backup_copy.cc

@@ -1384,6 +1384,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()
 {
@@ -1403,6 +1427,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)) {
@@ -1417,6 +1442,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.
@@ -2226,6 +2255,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

@@ -1795,7 +1795,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

@@ -109,6 +109,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

@@ -130,14 +130,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;
@@ -172,7 +173,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);
 
@@ -193,14 +194,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
@@ -373,7 +391,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

@@ -58,7 +58,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
@@ -87,7 +87,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

extra/mariabackup/xtrabackup.cc

@@ -344,6 +344,25 @@ const char *opt_history = NULL;
 char mariabackup_exe[FN_REFLEN];
 char orig_argv1[FN_REFLEN];
 
+pthread_mutex_t backup_mutex;
+pthread_cond_t  scanned_lsn_cond;
+
+typedef std::map<space_id_t,std::string> space_id_to_name_t;
+
+struct ddl_tracker_t {
+	/** Tablspaces with their ID and name, as they were copied to backup.*/
+	space_id_to_name_t tables_in_backup;
+	/** Tablespaces for that optimized DDL without redo log was found.*/
+	std::set<space_id_t> optimized_ddl;
+	/** Drop operations found in redo log. */
+	std::set<space_id_t> drops;
+	/* For DDL operation found in redo log,  */
+	space_id_to_name_t id_to_name;
+};
+const space_id_t REMOVED_SPACE_ID = ULINT_MAX;
+static ddl_tracker_t ddl_tracker;
+
+
 /* Whether xtrabackup_binlog_info should be created on recovery */
 static bool recover_binlog_info;
 
@@ -536,49 +555,79 @@ void mdl_lock_all()
 		mdl_lock_table(node->space->id);
 	}
 	datafiles_iter_free(it);
-
-	DBUG_EXECUTE_IF("check_mdl_lock_works",
-		dbug_alter_thread_done =
-		  dbug_start_query_thread("ALTER TABLE test.t ADD COLUMN mdl_lock_column int",
-			 "Waiting for table metadata lock",1, ER_QUERY_INTERRUPTED););
 }
 
-/** Check if the space id belongs to the table which name should
-be skipped based on the --tables, --tables-file and --table-exclude
-options.
-@param[in]	space_id	space id to check
-@return true if the space id belongs to skip table/database list. */
-static bool backup_includes(space_id_t space_id)
-{
-	datafiles_iter_t *it = datafiles_iter_new();
-	if (!it)
-		return true;
 
-	while (fil_node_t *node = datafiles_iter_next(it)){
-		if (space_id == 0
-		    || (node->space->id == space_id
-			&& !check_if_skip_table(node->space->name))) {
+// Convert non-null terminated filename to space name
+std::string filename_to_spacename(const byte *filename, size_t len)
+{
+	// null- terminate filename
+	char *f = (char *)malloc(len + 1);
+	ut_a(f);
+	memcpy(f, filename, len);
+	f[len] = 0;
+	for (size_t i = 0; i < len; i++)
+		if (f[i] == '\\')
+			f[i] = '/';
+	char *p = strrchr(f, '.');
+	ut_a(p);
+	*p = 0;
+	char *table = strrchr(f, '/');
+	ut_a(table);
+	*table = 0;
+	char *db = strrchr(f, '/');
+	ut_a(db);
+	*table = '/';
+	return std::string(db+1);
+}
 
-			msg("mariabackup: Unsupported redo log detected "
-			"and it belongs to %s\n",
-			space_id ? node->name: "the InnoDB system tablespace");
+/** Report an operation to create, delete, or rename a file during backup.
+@param[in]	space_id	tablespace identifier
+@param[in]	flags		tablespace flags (NULL if not create)
+@param[in]	name		file name (not NUL-terminated)
+@param[in]	len		length of name, in bytes
+@param[in]	new_name	new file name (NULL if not rename)
+@param[in]	new_len		length of new_name, in bytes (0 if NULL) */
+void backup_file_op(ulint space_id, const byte* flags,
+	const byte* name, ulint len,
+	const byte* new_name, ulint new_len)
+{
 
-			msg("mariabackup: ALTER TABLE or OPTIMIZE TABLE "
-			"was being executed during the backup.\n");
+	ut_ad(!flags || !new_name);
+	ut_ad(name);
+	ut_ad(len);
+	ut_ad(!new_name == !new_len);
+	pthread_mutex_lock(&backup_mutex);
+
+	if (flags) {
+		ddl_tracker.id_to_name[space_id] = filename_to_spacename(name, len);
+		msg("DDL tracking :  create %zu \"%.*s\": %x\n",
+			space_id, int(len), name, mach_read_from_4(flags));
+	}
+	else if (new_name) {
+		ddl_tracker.id_to_name[space_id] = filename_to_spacename(new_name, new_len);
+		msg("DDL tracking : rename %zu \"%.*s\",\"%.*s\"\n",
+			space_id, int(len), name, int(new_len), new_name);
+	} else {
+		ddl_tracker.drops.insert(space_id);
+		msg("DDL tracking : delete %zu \"%.*s\"\n", space_id, int(len), name);
+	}
+	pthread_mutex_unlock(&backup_mutex);
+}
 
-			if (!opt_lock_ddl_per_table) {
-				msg("mariabackup: Use --lock-ddl-per-table "
-				"parameter to lock all the table before "
-				"backup operation.\n");
-			}
 
-			datafiles_iter_free(it);
-			return false;
-		}
-	}
+/** Callback whenever MLOG_INDEX_LOAD happens.
+@param[in]	space_id	space id to check
+@return false */
+void backup_optimized_ddl_op(ulint space_id)
+{
+	// TODO : handle incremental
+	if (xtrabackup_incremental)
+		return;
 
-	datafiles_iter_free(it);
-	return true;
+	pthread_mutex_lock(&backup_mutex);
+	ddl_tracker.optimized_ddl.insert(space_id);
+	pthread_mutex_unlock(&backup_mutex);
 }
 
 /* ======== Date copying thread context ======== */
@@ -2326,7 +2375,7 @@ xb_get_copy_action(const char *dflt)
 
 static
 my_bool
-xtrabackup_copy_datafile(fil_node_t* node, uint thread_n)
+xtrabackup_copy_datafile(fil_node_t* node, uint thread_n, const char *dest_name=0, ulonglong max_size=ULLONG_MAX)
 {
 	char			 dst_name[FN_REFLEN];
 	ds_file_t		*dstfile = NULL;
@@ -2356,20 +2405,35 @@ xtrabackup_copy_datafile(fil_node_t* node, uint thread_n)
 		return(FALSE);
 	}
 
+	bool was_dropped;
+	pthread_mutex_lock(&backup_mutex);
+	was_dropped = (ddl_tracker.drops.find(node->space->id) != ddl_tracker.drops.end());
+	pthread_mutex_unlock(&backup_mutex);
+	if (was_dropped) {
+		if (node->is_open()) {
+			mutex_enter(&fil_system.mutex);
+			node->close();
+			mutex_exit(&fil_system.mutex);
+		}
+		goto skip;
+	}
+
 	if (!changed_page_bitmap) {
 		read_filter = &rf_pass_through;
 	}
 	else {
 		read_filter = &rf_bitmap;
 	}
-	res = xb_fil_cur_open(&cursor, read_filter, node, thread_n);
+
+	res = xb_fil_cur_open(&cursor, read_filter, node, thread_n,max_size);
 	if (res == XB_FIL_CUR_SKIP) {
 		goto skip;
 	} else if (res == XB_FIL_CUR_ERROR) {
 		goto error;
 	}
 
-	strncpy(dst_name, cursor.rel_path, sizeof(dst_name));
+	strncpy(dst_name, (dest_name)?dest_name : cursor.rel_path, sizeof(dst_name));
+
 
 	/* Setup the page write filter */
 	if (xtrabackup_incremental) {
@@ -2421,6 +2485,10 @@ xtrabackup_copy_datafile(fil_node_t* node, uint thread_n)
 		goto error;
 	}
 
+	pthread_mutex_lock(&backup_mutex);
+	ddl_tracker.tables_in_backup[node->space->id] = node_name;
+	pthread_mutex_unlock(&backup_mutex);
+
 	/* close */
 	msg_ts("[%02u]        ...done\n", thread_n);
 	xb_fil_cur_close(&cursor);
@@ -2584,7 +2652,7 @@ static bool xtrabackup_copy_logfile(bool last = false)
 		if (!start_lsn) {
 			msg("mariabackup: Error: xtrabackup_copy_logfile()"
 			    " failed.\n");
-			return(true);
+			exit(EXIT_FAILURE);
 		}
 	} while (start_lsn == end_lsn);
 
@@ -2593,13 +2661,31 @@ static bool xtrabackup_copy_logfile(bool last = false)
 	msg_ts(">> log scanned up to (" LSN_PF ")\n", start_lsn);
 
 	/* update global variable*/
+	pthread_mutex_lock(&backup_mutex);
 	log_copy_scanned_lsn = start_lsn;
+	pthread_cond_broadcast(&scanned_lsn_cond);
+	pthread_mutex_unlock(&backup_mutex);
 
 	debug_sync_point("xtrabackup_copy_logfile_pause");
 	return(false);
 }
 
-static os_thread_ret_t DECLARE_THREAD(log_copying_thread)(void*)
+/**
+Wait until redo log copying thread processes given lsn
+*/
+void backup_wait_for_lsn(lsn_t lsn) {
+	bool completed = false;
+	pthread_mutex_lock(&backup_mutex);
+	do {
+		pthread_cond_wait(&scanned_lsn_cond, &backup_mutex);
+		completed = log_copy_scanned_lsn >= lsn;
+	} while (!completed);
+	pthread_mutex_unlock(&backup_mutex);
+}
+
+extern lsn_t server_lsn_after_lock;
+
+static os_thread_ret_t log_copying_thread(void*)
 {
 	/*
 	  Initialize mysys thread-specific memory so we can
@@ -2655,6 +2741,42 @@ static os_thread_ret_t DECLARE_THREAD(io_watching_thread)(void*)
 	return(0);
 }
 
+#ifndef DBUG_OFF
+/*
+In debug mode,  execute SQL statement that was passed via environment.
+To use this facility, you need to
+
+1. Add code DBUG_EXECUTE_MARIABACKUP_EVENT("my_event_name", key););
+  to the code. key is usually a table name
+2. Set environment variable my_event_name_$key SQL statement you want to execute
+   when event occurs, in DBUG_EXECUTE_IF from above.
+   In mtr , you can set environment via 'let' statement (do not use $ as the first char
+   for the variable)
+3. start mariabackup with --dbug=+d,debug_mariabackup_events
+*/
+static void dbug_mariabackup_event(const char *event,const char *key)
+{
+	char envvar[FN_REFLEN];
+	if (key) {
+		snprintf(envvar, sizeof(envvar), "%s_%s", event, key);
+		char *slash = strchr(envvar, '/');
+		if (slash)
+			*slash = '_';
+	} else {
+		strncpy(envvar, event, sizeof(envvar));
+	}
+	char *sql = getenv(envvar);
+	if (sql) {
+		msg("dbug_mariabackup_event : executing '%s'\n", sql);
+		xb_mysql_query(mysql_connection, sql, false, true);
+	}
+
+}
+#define DBUG_MARIABACKUP_EVENT(A, B) DBUG_EXECUTE_IF("mariabackup_events", dbug_mariabackup_event(A,B););
+#else
+#define DBUG_MARIABACKUP_EVENT(A,B)
+#endif
+
 /**************************************************************************
 Datafiles copying thread.*/
 static
@@ -2677,12 +2799,18 @@ DECLARE_THREAD(data_copy_thread_func)(
 
 	while ((node = datafiles_iter_next(ctxt->it)) != NULL) {
 
+		DBUG_MARIABACKUP_EVENT("before_copy", node->space->name);
+
+
 		/* copy the datafile */
 		if(xtrabackup_copy_datafile(node, num)) {
 			msg("[%02u] mariabackup: Error: "
 			    "failed to copy datafile.\n", num);
 			exit(EXIT_FAILURE);
 		}
+
+		DBUG_MARIABACKUP_EVENT("after_copy", node->space->name);
+
 	}
 
 	pthread_mutex_lock(&ctxt->count_mutex);
@@ -2854,6 +2982,7 @@ xb_load_single_table_tablespace(
 	Datafile *file = xb_new_datafile(name, is_remote);
 
 	if (file->open_read_only(true) != DB_SUCCESS) {
+		msg("Can't open datafile %s\n", name);
 		ut_free(name);
 		exit(EXIT_FAILURE);
 	}
@@ -3171,7 +3300,7 @@ xb_load_tablespaces()
 	}
 
 	debug_sync_point("xtrabackup_load_tablespaces_pause");
-
+	DBUG_MARIABACKUP_EVENT("after_load_tablespaces", 0);
 	return(DB_SUCCESS);
 }
 
@@ -3766,6 +3895,8 @@ xtrabackup_backup_func()
 	uint			 count;
 	pthread_mutex_t		 count_mutex;
 	data_thread_ctxt_t 	*data_threads;
+	pthread_mutex_init(&backup_mutex, NULL);
+	pthread_cond_init(&scanned_lsn_cond, NULL);
 
 #ifdef USE_POSIX_FADVISE
 	msg("mariabackup: uses posix_fadvise().\n");
@@ -3779,7 +3910,7 @@ xtrabackup_backup_func()
 		return(false);
 	}
 	msg("mariabackup: cd to %s\n", mysql_real_data_home);
-
+	encryption_plugin_backup_init(mysql_connection);
 	msg("mariabackup: open files limit requested %u, set to %u\n",
 	    (uint) xb_open_files_limit,
 	    xb_set_max_open_files(xb_open_files_limit));
@@ -3792,6 +3923,7 @@ xtrabackup_backup_func()
 	srv_read_only_mode = TRUE;
 
 	srv_operation = SRV_OPERATION_BACKUP;
+	log_file_op = backup_file_op;
 	metadata_to_lsn = 0;
 
 	if (xb_close_files)
@@ -3805,6 +3937,7 @@ xtrabackup_backup_func()
 fail:
 		metadata_to_lsn = log_copying_running;
 		stop_backup_threads();
+		log_file_op = NULL;
 		if (dst_log_file) {
 			ds_close(dst_log_file);
 			dst_log_file = NULL;
@@ -4054,6 +4187,7 @@ xtrabackup_backup_func()
 		goto fail_before_log_copying_thread_start;
 
 	log_copying_stop = os_event_create(0);
+	log_optimized_ddl_op = backup_optimized_ddl_op;
 	os_thread_create(log_copying_thread, NULL, &log_copying_thread_id);
 
 	/* FLUSH CHANGED_PAGE_BITMAPS call */
@@ -4085,6 +4219,11 @@ xtrabackup_backup_func()
 
 	if (opt_lock_ddl_per_table) {
 		mdl_lock_all();
+
+		DBUG_EXECUTE_IF("check_mdl_lock_works",
+			dbug_alter_thread_done =
+			dbug_start_query_thread("ALTER TABLE test.t ADD COLUMN mdl_lock_column int",
+				"Waiting for table metadata lock", 1, ER_QUERY_INTERRUPTED););
 	}
 
 	it = datafiles_iter_new();
@@ -4122,10 +4261,6 @@ xtrabackup_backup_func()
 	pthread_mutex_destroy(&count_mutex);
 	free(data_threads);
 	datafiles_iter_free(it);
-
-	if (changed_page_bitmap) {
-		xb_page_bitmap_deinit(changed_page_bitmap);
-	}
 	}
 
 	bool ok = backup_start();
@@ -4149,6 +4284,9 @@ xtrabackup_backup_func()
 		goto fail;
 	}
 
+	if (changed_page_bitmap) {
+		xb_page_bitmap_deinit(changed_page_bitmap);
+	}
 	xtrabackup_destroy_datasinks();
 
 	msg("mariabackup: Redo log (from LSN " LSN_PF " to " LSN_PF
@@ -4166,9 +4304,182 @@ xtrabackup_backup_func()
 	}
 
 	innodb_shutdown();
+	log_file_op = NULL;
+	pthread_mutex_destroy(&backup_mutex);
+	pthread_cond_destroy(&scanned_lsn_cond);
 	return(true);
 }
 
+
+/**
+This function handles DDL changes at the end of backup, under protection of
+FTWRL.  This ensures consistent backup in presence of DDL.
+
+- New tables, that were created during backup, are now copied into backup.
+  Also, tablespaces with optimized (no redo loggin DDL) are re-copied into 
+  backup. This tablespaces will get the extension ".new" in the backup
+
+- Tables that were renamed during backup, are marked as renamed
+  For these, file <old_name>.ren will be created.
+  The content of the file is the new tablespace name.
+
+- Tables that were deleted during backup, are marked as deleted
+  For these , an empty file <name>.del will be created
+
+  It is the responsibility of the prepare phase to deal with .new, .ren, and .del
+  files.
+*/
+void backup_fix_ddl(void)
+{
+	std::set<std::string> new_tables;
+	std::set<std::string> dropped_tables;
+	std::map<std::string, std::string> renamed_tables;
+
+	for (space_id_to_name_t::iterator iter = ddl_tracker.tables_in_backup.begin();
+		iter != ddl_tracker.tables_in_backup.end();
+		iter++) {
+
+		const std::string name = iter->second;
+		ulint id = iter->first;
+
+		if (ddl_tracker.drops.find(id) != ddl_tracker.drops.end()) {
+			dropped_tables.insert(name);
+			continue;
+		}
+
+		bool has_optimized_ddl =
+			ddl_tracker.optimized_ddl.find(id) != ddl_tracker.optimized_ddl.end();
+
+		if (ddl_tracker.id_to_name.find(id) == ddl_tracker.id_to_name.end()) {
+			if (has_optimized_ddl) {
+				new_tables.insert(name);
+			}
+			continue;
+		}
+
+		/* tablespace was affected by DDL. */
+		const std::string new_name = ddl_tracker.id_to_name[id];
+		if (new_name != name) {
+			if (has_optimized_ddl) {
+				/* table was renamed, but we need a full copy
+				of it because of optimized DDL. We emulate a drop/create.*/
+				dropped_tables.insert(name);
+				new_tables.insert(new_name);
+			} else {
+				/* Renamed, and no optimized DDL*/
+				renamed_tables[name] = new_name;
+			}
+		} else if (has_optimized_ddl) {
+			/* Table was recreated, or optimized DDL ran.
+			In both cases we need a full copy in the backup.*/
+			new_tables.insert(name);
+		}
+	}
+
+	/* Find tables that were created during backup (and not removed).*/
+	for(space_id_to_name_t::iterator iter = ddl_tracker.id_to_name.begin();
+		iter != ddl_tracker.id_to_name.end();
+		iter++) {
+
+		ulint id = iter->first;
+		std::string name = iter->second;
+
+		if (ddl_tracker.tables_in_backup.find(id) != ddl_tracker.tables_in_backup.end()) {
+			/* already processed above */
+			continue;
+		}
+
+		if (ddl_tracker.drops.find(id) == ddl_tracker.drops.end()) {
+			dropped_tables.erase(name);
+			new_tables.insert(name);
+		}
+	}
+
+	// Mark tablespaces for rename
+	for (std::map<std::string, std::string>::iterator iter = renamed_tables.begin();
+		iter != renamed_tables.end(); ++iter) {
+		const std::string old_name = iter->first;
+		std::string new_name = iter->second;
+		backup_file_printf((old_name + ".ren").c_str(), "%s", new_name.c_str());
+	}
+
+	// Mark tablespaces for drop
+	for (std::set<std::string>::iterator iter = dropped_tables.begin();
+		iter != dropped_tables.end();
+		iter++) {
+		const std::string name(*iter);
+		backup_file_printf((name + ".del").c_str(), "%s", "");
+	}
+
+	//  Load and copy new tables.
+	//  Close all datanodes first, reload only new tables.
+	std::vector<fil_node_t *> all_nodes;
+	datafiles_iter_t *it = datafiles_iter_new();
+	if (!it)
+		return;
+	while (fil_node_t *node = datafiles_iter_next(it)) {
+		all_nodes.push_back(node);
+	}
+	for (size_t i = 0; i < all_nodes.size(); i++) {
+		fil_node_t *n = all_nodes[i];
+		if (n->space->id == 0)
+			continue;
+		if (n->is_open()) {
+			mutex_enter(&fil_system.mutex);
+			n->close();
+			mutex_exit(&fil_system.mutex);
+		}
+		fil_space_free(n->space->id, false);
+	}
+
+
+	for (std::set<std::string>::iterator iter = new_tables.begin();
+		iter != new_tables.end(); iter++) {
+		const char *space_name = iter->c_str();
+		if (check_if_skip_table(space_name))
+			continue;
+		std::string name(*iter);
+		bool is_remote = access((name + ".ibd").c_str(), R_OK) != 0;
+		const char *extension = is_remote ? ".isl" : ".ibd";
+		name.append(extension);
+		char buf[FN_REFLEN];
+		strncpy(buf, name.c_str(), sizeof(buf));
+		const char *dbname = buf;
+		char *p = strchr(buf, '/');
+		if (p == 0) {
+			msg("Unexpected tablespace %s filename %s\n", space_name, name.c_str());
+			ut_a(0);
+		}
+		ut_a(p);
+		*p = 0;
+		const char *tablename = p + 1;
+		xb_load_single_table_tablespace(dbname, tablename, is_remote);
+	}
+
+	it = datafiles_iter_new();
+	if (!it)
+		return;
+
+	while (fil_node_t *node = datafiles_iter_next(it)) {
+		fil_space_t * space = node->space;
+		if (!fil_is_user_tablespace_id(space->id))
+			continue;
+		std::string dest_name(node->space->name);
+		dest_name.append(".new");
+#if 0
+		bool do_full_copy = ddl_tracker.optimized_ddl.find(n->space->id) != ddl_tracker.optimized_ddl.end();
+		if (do_full_copy) {
+			msg(
+				"Performing a full copy of the tablespace %s, because optimized (without redo logging) DDL operation"
+				"ran during backup. You can use set innodb_log_optimize_ddl=OFF to improve backup performance"
+				"in the future.\n",
+				n->space->name);
+		}
+#endif
+		xtrabackup_copy_datafile(node, 0, dest_name.c_str()/*, do_full_copy ? ULONGLONG_MAX:UNIV_PAGE_SIZE */);
+	}
+}
+
 /* ================= prepare ================= */
 
 /***********************************************************************
@@ -4681,6 +4992,27 @@ xtrabackup_apply_delta(
 	return FALSE;
 }
 
+
+std::string change_extension(std::string filename, std::string new_ext) {
+	DBUG_ASSERT(new_ext.size() == 3);
+	std::string new_name(filename);
+	new_name.resize(new_name.size() - new_ext.size());
+	new_name.append(new_ext);
+	return new_name;
+}
+
+
+static void rename_file(const char *from,const char *to) {
+	msg("Renaming %s to %s\n", from, to);
+	if (my_rename(from, to, MY_WME)) {
+		msg("Cannot rename %s to %s errno %d", from, to, errno);
+		exit(EXIT_FAILURE);
+	}
+}
+
+static void rename_file(const std::string& from, const std::string &to) {
+	rename_file(from.c_str(), to.c_str());
+}
 /************************************************************************
 Callback to handle datadir entry. Function of this type will be called
 for each entry which matches the mask by xb_process_datadir.
@@ -4692,6 +5024,37 @@ typedef ibool (*handle_datadir_entry_func_t)(
 	const char*	file_name,		/*!<in: file name with suffix */
 	void*		arg);			/*!<in: caller-provided data */
 
+/** Rename, and replace destination file, if exists */
+static void rename_force(const char *from, const char *to) {
+	if (access(to, R_OK) == 0) {
+		msg("Removing %s\n", to);
+		if (my_delete(to, MYF(MY_WME))) {
+			msg("Can't remove %s, errno %d", to, errno);
+			exit(EXIT_FAILURE);
+		}
+	}
+	rename_file(from,to);
+}
+
+/* During prepare phase, rename ".new" files , that were created in backup_fix_ddl(),
+  to ".ibd".*/
+static ibool prepare_handle_new_files(
+	const char*	data_home_dir,		/*!<in: path to datadir */
+	const char*	db_name,		/*!<in: database name */
+	const char*	file_name,		/*!<in: file name with suffix */
+	void *)
+{
+
+	std::string src_path = std::string(data_home_dir) + '/' + std::string(db_name) + '/' + file_name;
+	std::string dest_path = src_path;
+
+	size_t index = dest_path.find(".new");
+	DBUG_ASSERT(index != std::string::npos);
+	dest_path.replace(index, 4, ".ibd");
+	rename_force(src_path.c_str(),dest_path.c_str());
+	return TRUE;
+}
+
 /************************************************************************
 Callback to handle datadir entry. Deletes entry if it has no matching
 fil_space in fil_system directory.
@@ -4897,6 +5260,103 @@ store_binlog_info(const char* filename, const char* name, ulonglong pos)
 	return(true);
 }
 
+/** Check if file exists*/
+static bool file_exists(std::string name)
+{
+	return access(name.c_str(), R_OK) == 0 ;
+}
+
+/** Read file content into STL string */
+static std::string read_file_as_string(const std::string file) {
+	char content[FN_REFLEN];
+	FILE *f = fopen(file.c_str(), "r");
+	if (!f) {
+		msg("Can not open %s\n", file.c_str());
+	}
+	size_t len = fread(content, 1, FN_REFLEN, f);
+	fclose(f);
+	return std::string(content, len);
+}
+
+/** Delete file- Provide verbose diagnostics and exit, if operation fails. */
+static void delete_file(const std::string& file, bool if_exists = false) {
+	if (if_exists && !file_exists(file))
+		return;
+	if (my_delete(file.c_str(), MYF(MY_WME))) {
+		msg("Can't remove %s, errno %d", file.c_str(), errno);
+		exit(EXIT_FAILURE);
+	}
+}
+
+/**
+Rename tablespace during prepare.
+Backup in its end phase may generate some .ren files, recording
+tablespaces that should be renamed in --prepare.
+*/
+static void rename_table_in_prepare(const std::string &datadir, const std::string& from , const std::string& to,
+	const char *extension=0) {
+	if (!extension) {
+		static const char *extensions_nonincremental[] = { ".ibd", 0 };
+		static const char *extensions_incremental[] = { ".ibd.delta", ".ibd.meta", 0 };
+		const char **extensions = xtrabackup_incremental_dir ?
+			extensions_incremental : extensions_nonincremental;
+		for (size_t i = 0; extensions[i]; i++) {
+			rename_table_in_prepare(datadir, from, to, extensions[i]);
+		}
+		return;
+	}
+	std::string src = std::string(datadir) + "/" + from + extension;
+	std::string dest = std::string(datadir) + "/" + to + extension;
+	std::string ren2, tmp;
+	if (file_exists(dest)) {
+		ren2= std::string(datadir) + "/" + to + ".ren";
+		if (!file_exists(ren2)) {
+			msg("ERROR : File %s was not found, but expected during rename processing\n", ren2.c_str());
+			ut_a(0);
+		}
+		tmp = to + "#";
+		rename_table_in_prepare(datadir, to, tmp);
+	}
+	rename_file(src, dest);
+	if (ren2.size()) {
+		// Make sure the temp. renamed file is processed.
+		std::string to2 = read_file_as_string(ren2);
+		rename_table_in_prepare(datadir, tmp, to2);
+		delete_file(ren2);
+	}
+}
+
+static ibool prepare_handle_ren_files(const char *datadir, const char *db, const char *filename, void *) {
+
+	std::string ren_file = std::string(datadir) + "/" + db + "/" + filename;
+	if (!file_exists(ren_file))
+		return TRUE;
+
+	std::string to = read_file_as_string(ren_file);
+	std::string source_space_name = std::string(db) + "/" + filename;
+	source_space_name.resize(source_space_name.size() - 4); // remove extension
+
+	rename_table_in_prepare(datadir, source_space_name.c_str(), to.c_str());
+	delete_file(ren_file);
+	return TRUE;
+}
+
+/* Remove tablespaces during backup, based on */
+static ibool prepare_handle_del_files(const char *datadir, const char *db, const char *filename, void *) {
+	std::string del_file = std::string(datadir) + "/" + db + "/" + filename;
+	std::string path(del_file);
+	path.resize(path.size() - 4); // remove extension;
+	if (xtrabackup_incremental) {
+		delete_file(path + ".ibd.delta", true);
+		delete_file(path + ".ibd.meta", true);
+	}
+	else {
+		delete_file(path + ".ibd", true);
+	}
+	delete_file(del_file);
+	return TRUE;
+}
+
 /** Implement --prepare
 @return	whether the operation succeeded */
 static bool
@@ -4914,6 +5374,21 @@ xtrabackup_prepare_func(char** argv)
 	}
 	msg("mariabackup: cd to %s\n", xtrabackup_real_target_dir);
 
+	fil_path_to_mysql_datadir = ".";
+
+	if (xtrabackup_incremental_dir) {
+		xb_process_datadir(xtrabackup_incremental_dir, ".new.meta", prepare_handle_new_files);
+		xb_process_datadir(xtrabackup_incremental_dir, ".new.delta", prepare_handle_new_files);
+	}
+	else {
+		xb_process_datadir(".", ".new", prepare_handle_new_files);
+	}
+	xb_process_datadir(xtrabackup_incremental_dir? xtrabackup_incremental_dir:".",
+		".ren", prepare_handle_ren_files);
+	xb_process_datadir(xtrabackup_incremental_dir ? xtrabackup_incremental_dir : ".",
+		".del", prepare_handle_del_files);
+
+
 	int argc; for (argc = 0; argv[argc]; argc++) {}
 	encryption_plugin_prepare_init(argc, argv);
 
@@ -5204,7 +5679,6 @@ xb_init()
 			return(false);
 		}
 
-		encryption_plugin_backup_init(mysql_connection);
 		history_start_time = time(NULL);
 
 	}
@@ -5262,7 +5736,7 @@ handle_options(int argc, char **argv, char ***argv_client, char ***argv_server)
 	srv_operation = SRV_OPERATION_RESTORE;
 
 	files_charset_info = &my_charset_utf8_general_ci;
-	check_if_backup_includes = backup_includes;
+
 
 	setup_error_messages();
 	sys_var_init();

mysql-test/suite/mariabackup/create_during_backup.result

+# xtrabackup backup
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT COUNT(*) from t1;
+COUNT(*)
+10000
+DROP TABLE t1;

mysql-test/suite/mariabackup/create_during_backup.test

+--source include/have_debug.inc
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+mkdir $targetdir;
+
+# this will table and populate it, after backup has list of tables to be copied
+--let after_load_tablespaces =CREATE TABLE test.t1 ENGINE=INNODB SELECT UUID() from test.seq_1_to_10000
+
+echo # xtrabackup backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --dbug=+d,mariabackup_events;
+--enable_result_log
+
+--let after_load_tables=
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# Check that new table is there after restore.
+SELECT COUNT(*) from t1;
+DROP TABLE t1;
+rmdir $targetdir;
+
+

mysql-test/suite/mariabackup/create_with_data_directory_during_backup.result

+# xtrabackup backup
+# xtrabackup prepare
+DROP TABLE t;
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT * FROM t;
+i
+DROP TABLE t;

mysql-test/suite/mariabackup/create_with_data_directory_during_backup.test

+--source include/have_debug.inc
+let $table_data_dir=$MYSQLTEST_VARDIR/tmp/ddir;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+mkdir $table_data_dir;
+--replace_result $table_data_dir table_data_dir
+--let after_load_tablespaces=CREATE TABLE test.t(i int) ENGINE=INNODB DATA DIRECTORY='$table_data_dir'
+echo # xtrabackup backup;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+--disable_result_log
+exec   $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --target-dir=$targetdir --dbug=+d,mariabackup_events;
+--enable_result_log
+--source include/shutdown_mysqld.inc
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+--source include/start_mysqld.inc
+DROP TABLE t;
+rmdir $table_data_dir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+SELECT * FROM t;
+DROP TABLE t;
+rmdir $targetdir;
+rmdir $table_data_dir;

mysql-test/suite/mariabackup/disabled.def

-
+unsupported_redo    : MDEV-16791 allows optimized redo
\ No newline at end of file

mysql-test/suite/mariabackup/drop_table_during_backup.result

+CREATE TABLE t1 (i int) ENGINE=INNODB;
+CREATE TABLE t2 (i int) ENGINE=INNODB;
+CREATE TABLE t3 (i int) ENGINE=INNODB;
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+CREATE TABLE t2(i int);
+DROP TABLE t2;
+DROP TABLE t3;

mysql-test/suite/mariabackup/drop_table_during_backup.test

+--source include/have_debug.inc
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+CREATE TABLE t1 (i int) ENGINE=INNODB;
+CREATE TABLE t2 (i int) ENGINE=INNODB;
+CREATE TABLE t3 (i int) ENGINE=INNODB;
+--let before_copy_test_t1=DROP TABLE test.t1
+--let after_copy_test_t2=DROP TABLE test.t2;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf --backup --target-dir=$targetdir --dbug=+d,mariabackup_events;
+--enable_result_log
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+#check that the table t1 does not exist in backup
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+CREATE TABLE t2(i int);
+DROP TABLE t2;
+DROP TABLE t3;
+rmdir $targetdir;

mysql-test/suite/mariabackup/incremental_ddl_during_backup.result

+call mtr.add_suppression("InnoDB: New log files created");
+CREATE TABLE t1(i INT PRIMARY KEY) ENGINE INNODB;
+CREATE TABLE t2(i INT PRIMARY KEY) ENGINE INNODB;
+CREATE TABLE t3(i INT) ENGINE INNODB;
+# Create full backup , modify table, then create incremental/differential backup
+INSERT into t1 values(1);
+# Prepare full backup, apply incremental one
+# Restore and check results
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+SELECT * from t1_renamed;
+i
+1
+DROP TABLE t1_renamed;
+CREATE TABLE t2(i INT PRIMARY KEY) ENGINE INNODB;
+DROP TABLE t2;
+DROP TABLE t3;
+DROP TABLE t4;

mysql-test/suite/mariabackup/incremental_ddl_during_backup.test

+--source include/have_debug.inc
+
+call mtr.add_suppression("InnoDB: New log files created");
+
+let $basedir=$MYSQLTEST_VARDIR/tmp/backup;
+let $incremental_dir=$MYSQLTEST_VARDIR/tmp/backup_inc1;
+
+CREATE TABLE t1(i INT PRIMARY KEY) ENGINE INNODB;
+CREATE TABLE t2(i INT PRIMARY KEY) ENGINE INNODB;
+CREATE TABLE t3(i INT) ENGINE INNODB;
+
+echo # Create full backup , modify table, then create incremental/differential backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$basedir;
+--enable_result_log
+
+INSERT into t1 values(1);
+
+--let after_load_tablespaces=CREATE TABLE test.t4 ENGINE=INNODB SELECT UUID() from test.seq_1_to_10000
+--let after_copy_test_t1=RENAME TABLE  test.t1 TO test.t1_renamed
+--let after_copy_test_t2=DROP TABLE test.t2
+--let after_copy_test_t3=CREATE INDEX a_i ON test.t3(i);
+
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$incremental_dir --incremental-basedir=$basedir --dbug=+d,mariabackup_events;
+--let after_load_tablespaces=
+--disable_result_log
+echo # Prepare full backup, apply incremental one;
+exec $XTRABACKUP --apply-log-only --prepare --target-dir=$basedir;
+exec $XTRABACKUP --prepare --target-dir=$basedir --incremental-dir=$incremental_dir ;
+
+echo # Restore and check results;
+let $targetdir=$basedir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# Test that t1 does not exist, but t1_renamed does
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+
+SELECT * from t1_renamed;
+DROP TABLE t1_renamed;
+
+# Test that t2 does not exist;
+CREATE TABLE t2(i INT PRIMARY KEY) ENGINE INNODB;
+DROP TABLE t2;
+
+DROP TABLE t3;
+DROP TABLE t4;
+
+# Cleanup
+rmdir $basedir;
+rmdir $incremental_dir;

mysql-test/suite/mariabackup/mlog_index_load.result

+CREATE TABLE t1(i INT PRIMARY KEY auto_increment, a int) ENGINE INNODB;
+INSERT INTO t1(a) SELECT * from seq_1_to_10000;
+# xtrabackup backup
+t1.frm
+t1.ibd
+t1.new
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT COUNT(*) from t1;
+COUNT(*)
+10000
+DROP TABLE t1;

mysql-test/suite/mariabackup/mlog_index_load.test

+--source include/have_debug.inc
+
+CREATE TABLE t1(i INT PRIMARY KEY auto_increment, a int) ENGINE INNODB;
+INSERT INTO t1(a) SELECT * from seq_1_to_10000;
+
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+
+let after_copy_test_t1=CREATE INDEX a_ind ON test.t1(a) ALGORITHM=INPLACE;
+echo # xtrabackup backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --dbug=+d,mariabackup_events;
+--enable_result_log
+
+--list_files $targetdir/test t1*
+--let before_copy_test_t1=
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# Check that new table is there after restore.
+SELECT COUNT(*) from t1;
+DROP TABLE t1;
+rmdir $targetdir;
+

mysql-test/suite/mariabackup/recreate_table_during_backup.result

+CREATE TABLE t1(i int) ENGINE=INNODB;
+CREATE TABLE t2(i int) ENGINE=INNODB;
+CREATE TABLE t3(a CHAR(36)) ENGINE INNODB;
+INSERT INTO t3 SELECT UUID() FROM seq_1_to_1000;
+# xtrabackup backup
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+SELECT COUNT(*) from t1;
+COUNT(*)
+100
+SELECT COUNT(*) from t2;
+COUNT(*)
+1000
+SELECT COUNT(*) from t3;
+COUNT(*)
+1000
+DROP INDEX index_a ON t3;
+DROP TABLE t1;
+DROP TABLE t2;
+DROP TABLE t3;

mysql-test/suite/mariabackup/recreate_table_during_backup.test

+--source include/have_debug.inc
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+mkdir $targetdir;
+
+CREATE TABLE t1(i int) ENGINE=INNODB;
+CREATE TABLE t2(i int) ENGINE=INNODB;
+CREATE TABLE t3(a CHAR(36)) ENGINE INNODB;
+INSERT INTO t3 SELECT UUID() FROM seq_1_to_1000;
+
+# this will table and populate it, after backup has list of tables to be copied
+--let before_copy_test_t1=BEGIN NOT ATOMIC DROP TABLE test.t1;CREATE TABLE test.t1 ENGINE=INNODB SELECT UUID() from test.seq_1_to_100; END
+--let after_copy_test_t2=BEGIN NOT ATOMIC  DROP TABLE test.t2;CREATE TABLE test.t2 ENGINE=INNODB SELECT UUID() from test.seq_1_to_1000; END
+--let after_copy_test_t3=ALTER TABLE test.t3 ADD INDEX index_a(a),ALGORITHM=COPY
+echo # xtrabackup backup;
+--disable_result_log
+exec $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --close-files --dbug=+d,mariabackup_events;
+--enable_result_log
+--let after_load_tables=
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# Check that new table is there after restore.
+SELECT COUNT(*) from t1;
+SELECT COUNT(*) from t2;
+SELECT COUNT(*) from t3;
+DROP INDEX index_a ON t3;
+DROP TABLE t1;
+DROP TABLE t2;
+DROP TABLE t3;
+rmdir $targetdir;
+
+
+

mysql-test/suite/mariabackup/rename_during_backup.result

+CREATE TABLE t1(i int)   ENGINE INNODB;
+INSERT into t1 values(1);
+CREATE TABLE t2(i int)   ENGINE INNODB;
+INSERT INTO t2 values(2);
+CREATE TABLE t3(i int) ENGINE INNODB;
+CREATE TABLE t4(i int) ENGINE INNODB;
+CREATE TABLE a(a int)    ENGINE INNODB;
+INSERT INTO a values(1);
+CREATE TABLE b(b CHAR(1)) ENGINE INNODB;
+INSERT INTO b VALUES('b');
+CREATE TABLE a1(a1 int)    ENGINE INNODB;
+INSERT INTO a1 VALUES(1);
+CREATE TABLE b1(b1 CHAR(2)) ENGINE INNODB;
+INSERT INTO b1 VALUES('b1');
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+SELECT * from t1_renamed;
+i
+1
+DROP TABLE t1_renamed;
+CREATE TABLE t2(i int);
+DROP TABLE t2;
+SELECT * from t2_renamed;
+i
+2
+DROP TABLE t2_renamed;
+SELECT * from t3;
+i
+3
+DROP TABLE t3;
+SELECT * from t4;
+i
+DROP TABLE t4;
+CREATE TABLE tmp(i int);
+DROP TABLE tmp;
+SELECT * FROM a;
+b
+b
+SELECT * FROM b;
+a
+1
+SELECT * FROM a1;
+b1
+b1
+SELECT * FROM b1;
+a1
+1
+DROP TABLE a,b,a1,b1;

mysql-test/suite/mariabackup/rename_during_backup.test

+--source include/have_debug.inc
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
+mkdir $targetdir;
+
+CREATE TABLE t1(i int)   ENGINE INNODB;
+INSERT into t1 values(1);
+
+CREATE TABLE t2(i int)   ENGINE INNODB;
+INSERT INTO t2 values(2);
+
+CREATE TABLE t3(i int) ENGINE INNODB;
+CREATE TABLE t4(i int) ENGINE INNODB;
+
+CREATE TABLE a(a int)    ENGINE INNODB;
+INSERT INTO a values(1);
+CREATE TABLE b(b CHAR(1)) ENGINE INNODB;
+INSERT INTO b VALUES('b');
+
+CREATE TABLE a1(a1 int)    ENGINE INNODB;
+INSERT INTO a1 VALUES(1);
+
+CREATE TABLE b1(b1 CHAR(2)) ENGINE INNODB;
+INSERT INTO b1 VALUES('b1');
+
+# Test renames before of after copying tablespaces
+--let before_copy_test_t1=RENAME TABLE test.t1 TO test.t1_renamed
+--let after_copy_test_t2=RENAME TABLE  test.t2 TO test.t2_renamed
+--let after_copy_test_t3=BEGIN NOT ATOMIC RENAME TABLE test.t3 TO test.t3_tmp; INSERT INTO test.t3_tmp VALUES(3); RENAME TABLE test.t3_tmp TO test.t3; END
+--let before_copy_test_t4=RENAME TABLE test.t4 TO test.t4_tmp
+--let after_copy_test_t4=RENAME TABLE test.t4_tmp TO test.t4
+
+# Test circular renames
+--let before_copy_test_b=RENAME TABLE test.a to test.tmp, test.b to test.a, test.tmp to test.b
+--let after_copy_test_b1=RENAME TABLE test.a1 to test.tmp, test.b1 to test.a1, test.tmp to test.b1
+
+--disable_result_log 
+exec  $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup --target-dir=$targetdir  --dbug=+d,mariabackup_events;
+--enable_result_log
+
+--let before_copy_test_t1=''
+--let after_copy_test_t2=''
+--let before_copy_test_a=''
+--let after_copy_test_a1=''
+
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+
+# the table was renamed from t1 to t1_renamed
+# make sure t1 does not exist, and t1_renamed does.
+
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+
+SELECT * from t1_renamed;
+DROP TABLE t1_renamed;
+
+
+CREATE TABLE t2(i int);
+DROP TABLE t2;
+
+SELECT * from t2_renamed;
+DROP TABLE t2_renamed;
+
+#rename to itself
+SELECT * from t3;
+DROP TABLE t3;
+
+SELECT * from t4;
+DROP TABLE t4;
+
+# For circular renames , make sure intermediate tables do not exist
+CREATE TABLE tmp(i int);
+DROP TABLE tmp;
+
+SELECT * FROM a;
+SELECT * FROM b;
+SELECT * FROM a1;
+SELECT * FROM b1;
+
+DROP TABLE a,b,a1,b1;
+rmdir $targetdir;
+
+

mysql-test/suite/mariabackup/rename_during_mdl_lock.result

 CREATE TABLE t1(i int) ENGINE INNODB;
-FOUND 1 /failed to execute query SELECT 1 FROM/ in backup.log
+# xtrabackup prepare
+# shutdown server
+# remove datadir
+# xtrabackup move back
+# restart server
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+SELECT * from t2;
+i
 DROP TABLE t2;

mysql-test/suite/mariabackup/rename_during_mdl_lock.test

 --source include/have_debug.inc
-let $targetdir=$MYSQLTEST_VARDIR/backup;
+let $targetdir=$MYSQLTEST_VARDIR/tmp/backup;
 mkdir $targetdir;
 CREATE TABLE t1(i int) ENGINE INNODB;
---error 1
-exec  $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --lock-ddl-per-table --dbug=+d,rename_during_mdl_lock_table 2>$targetdir/backup.log;
+exec  $XTRABACKUP --defaults-file=$MYSQLTEST_VARDIR/my.cnf  --backup  --target-dir=$targetdir  --lock-ddl-per-table --dbug=+d,rename_during_mdl_lock_table;
 
-let SEARCH_FILE=$targetdir/backup.log;
-let SEARCH_PATTERN=failed to execute query SELECT 1 FROM;
-source include/search_pattern_in_file.inc;
+echo # xtrabackup prepare;
+--disable_result_log
+exec $XTRABACKUP  --prepare --target-dir=$targetdir;
+-- source include/restart_and_restore.inc
+--enable_result_log
+# the table was renamed from t1 to t2
+# make sure t1 does not exist, and t2 does
 
+CREATE TABLE t1(i int);
+DROP TABLE t1;
+SELECT * from t2;
 DROP TABLE t2;
+
 rmdir $targetdir;
+

mysql-test/suite/mariabackup/suite.opt

---innodb --loose-changed_page_bitmaps --innodb-sys-tables --innodb-flush-log-at-trx-commit=2
+--innodb --loose-changed_page_bitmaps --innodb-sys-tables --innodb-flush-log-at-trx-commit=2 --sequence

mysql-test/suite/mariabackup/xb_aws_key_management.result

@@ -11,4 +11,6 @@ INSERT INTO t VALUES('foobar1');
 SELECT * from t;
 c
 foobar1
+Warnings:
+Note	1105	AWS KMS plugin: loaded key 1, version 1, key length 128 bit
 DROP TABLE t;

storage/innobase/fil/fil0fil.cc

@@ -739,42 +739,35 @@ fil_node_open_file(
 	return(true);
 }
 
-/** Close a file node.
-@param[in,out]	node	File node */
-static
-void
-fil_node_close_file(
-	fil_node_t*	node)
+/** Close the file handle. */
+void fil_node_t::close()
 {
 	bool	ret;
 
-	ut_ad(mutex_own(&(fil_system.mutex)));
-	ut_a(node->is_open());
-	ut_a(node->n_pending == 0);
-	ut_a(node->n_pending_flushes == 0);
-	ut_a(!node->being_extended);
-	ut_a(node->modification_counter == node->flush_counter
-	     || node->space->purpose == FIL_TYPE_TEMPORARY
+	ut_ad(mutex_own(&fil_system.mutex));
+	ut_a(is_open());
+	ut_a(n_pending == 0);
+	ut_a(n_pending_flushes == 0);
+	ut_a(!being_extended);
+	ut_a(modification_counter == flush_counter
+	     || space->purpose == FIL_TYPE_TEMPORARY
 	     || srv_fast_shutdown == 2
 	     || !srv_was_started);
 
-	ret = os_file_close(node->handle);
+	ret = os_file_close(handle);
 	ut_a(ret);
 
-	/* printf("Closing file %s\n", node->name); */
+	/* printf("Closing file %s\n", name); */
 
-	node->handle = OS_FILE_CLOSED;
-	ut_ad(!node->is_open());
+	handle = OS_FILE_CLOSED;
+	ut_ad(!is_open());
 	ut_a(fil_system.n_open > 0);
 	fil_system.n_open--;
 	fil_n_file_opened--;
 
-	if (fil_space_belongs_in_lru(node->space)) {
-
+	if (fil_space_belongs_in_lru(space)) {
 		ut_a(UT_LIST_GET_LEN(fil_system.LRU) > 0);
-
-		/* The node is in the LRU list, remove it */
-		UT_LIST_REMOVE(fil_system.LRU, node);
+		UT_LIST_REMOVE(fil_system.LRU, this);
 	}
 }
 
@@ -810,7 +803,7 @@ fil_try_to_close_file_in_LRU(
 		    && node->n_pending_flushes == 0
 		    && !node->being_extended) {
 
-			fil_node_close_file(node);
+			node->close();
 
 			return(true);
 		}
@@ -1240,7 +1233,7 @@ fil_node_close_to_free(
 	ut_a(!node->being_extended);
 
 	if (node->is_open()) {
-		/* We fool the assertion in fil_node_close_file() to think
+		/* We fool the assertion in fil_node_t::close() to think
 		there are no unflushed modifications in the file */
 
 		node->modification_counter = node->flush_counter;
@@ -1259,7 +1252,7 @@ fil_node_close_to_free(
 			UT_LIST_REMOVE(fil_system.unflushed_spaces, space);
 		}
 
-		fil_node_close_file(node);
+		node->close();
 	}
 }
 
@@ -1750,7 +1743,7 @@ void fil_space_t::close()
 	     node != NULL;
 	     node = UT_LIST_GET_NEXT(chain, node)) {
 		if (node->is_open()) {
-			fil_node_close_file(node);
+			node->close();
 		}
 	}
 
@@ -1911,7 +1904,7 @@ fil_close_all_files(void)
 		     node = UT_LIST_GET_NEXT(chain, node)) {
 
 			if (node->is_open()) {
-				fil_node_close_file(node);
+				node->close();
 			}
 		}
 
@@ -1958,7 +1951,7 @@ fil_close_log_files(
 		     node = UT_LIST_GET_NEXT(chain, node)) {
 
 			if (node->is_open()) {
-				fil_node_close_file(node);
+				node->close();
 			}
 		}
 

storage/innobase/include/fil0fil.h

@@ -327,6 +327,9 @@ struct fil_node_t {
 	{
 		return(handle != OS_FILE_CLOSED);
 	}
+
+	/** Close the file handle. */
+	void close();
 };
 
 /** Value of fil_node_t::magic_n */

storage/innobase/include/lock0lock.h

@@ -806,69 +806,6 @@ lock_trx_has_expl_x_lock(
 	MY_ATTRIBUTE((nonnull, warn_unused_result));
 #endif /* UNIV_DEBUG */
 
-/**
-Allocate cached locks for the transaction.
-@param trx		allocate cached record locks for this transaction */
-void
-lock_trx_alloc_locks(trx_t* trx);
-
-/** Lock modes and types */
-/* @{ */
-#define LOCK_MODE_MASK	0xFUL	/*!< mask used to extract mode from the
-				type_mode field in a lock */
-/** Lock types */
-/* @{ */
-#define LOCK_TABLE	16U	/*!< table lock */
-#define	LOCK_REC	32U	/*!< record lock */
-#define LOCK_TYPE_MASK	0xF0UL	/*!< mask used to extract lock type from the
-				type_mode field in a lock */
-#if LOCK_MODE_MASK & LOCK_TYPE_MASK
-# error "LOCK_MODE_MASK & LOCK_TYPE_MASK"
-#endif
-
-#define LOCK_WAIT	256U	/*!< Waiting lock flag; when set, it
-				means that the lock has not yet been
-				granted, it is just waiting for its
-				turn in the wait queue */
-/* Precise modes */
-#define LOCK_ORDINARY	0	/*!< this flag denotes an ordinary
-				next-key lock in contrast to LOCK_GAP
-				or LOCK_REC_NOT_GAP */
-#define LOCK_GAP	512U	/*!< when this bit is set, it means that the
-				lock holds only on the gap before the record;
-				for instance, an x-lock on the gap does not
-				give permission to modify the record on which
-				the bit is set; locks of this type are created
-				when records are removed from the index chain
-				of records */
-#define LOCK_REC_NOT_GAP 1024U	/*!< this bit means that the lock is only on
-				the index record and does NOT block inserts
-				to the gap before the index record; this is
-				used in the case when we retrieve a record
-				with a unique key, and is also used in
-				locking plain SELECTs (not part of UPDATE
-				or DELETE) when the user has set the READ
-				COMMITTED isolation level */
-#define LOCK_INSERT_INTENTION 2048U/*!< this bit is set when we place a waiting
-				gap type record lock request in order to let
-				an insert of an index record to wait until
-				there are no conflicting locks by other
-				transactions on the gap; note that this flag
-				remains set when the waiting lock is granted,
-				or if the lock is inherited to a neighboring
-				record */
-#define LOCK_PREDICATE	8192U	/*!< Predicate lock */
-#define LOCK_PRDT_PAGE	16384U	/*!< Page lock */
-
-
-#if (LOCK_WAIT|LOCK_GAP|LOCK_REC_NOT_GAP|LOCK_INSERT_INTENTION|LOCK_PREDICATE|LOCK_PRDT_PAGE)&LOCK_MODE_MASK
-# error
-#endif
-#if (LOCK_WAIT|LOCK_GAP|LOCK_REC_NOT_GAP|LOCK_INSERT_INTENTION|LOCK_PREDICATE|LOCK_PRDT_PAGE)&LOCK_TYPE_MASK
-# error
-#endif
-/* @} */
-
 /** Lock operation struct */
 struct lock_op_t{
 	dict_table_t*	table;	/*!< table to be locked */

storage/innobase/include/lock0priv.h

@@ -42,19 +42,6 @@ those functions in lock/ */
 #define UINT32_MAX             (4294967295U)
 #endif
 
-/** A table lock */
-struct lock_table_t {
-	dict_table_t*	table;		/*!< database table in dictionary
-					cache */
-	UT_LIST_NODE_T(lock_t)
-			locks;		/*!< list of locks on the same
-					table */
-	/** Print the table lock into the given output stream
-	@param[in,out]	out	the output stream
-	@return the given output stream. */
-	std::ostream& print(std::ostream& out) const;
-};
-
 /** Print the table lock into the given output stream
 @param[in,out]	out	the output stream
 @return the given output stream. */
@@ -77,131 +64,11 @@ operator<<(std::ostream& out, const lock_table_t& lock)
 	return(lock.print(out));
 }
 
-/** Record lock for a page */
-struct lock_rec_t {
-	ib_uint32_t	space;		/*!< space id */
-	ib_uint32_t	page_no;	/*!< page number */
-	ib_uint32_t	n_bits;		/*!< number of bits in the lock
-					bitmap; NOTE: the lock bitmap is
-					placed immediately after the
-					lock struct */
-
-	/** Print the record lock into the given output stream
-	@param[in,out]	out	the output stream
-	@return the given output stream. */
-	std::ostream& print(std::ostream& out) const;
-};
-
-/** Print the record lock into the given output stream
-@param[in,out]	out	the output stream
-@return the given output stream. */
-inline
-std::ostream& lock_rec_t::print(std::ostream& out) const
-{
-	out << "[lock_rec_t: space=" << space << ", page_no=" << page_no
-		<< ", n_bits=" << n_bits << "]";
-	return(out);
-}
-
-inline
-std::ostream&
-operator<<(std::ostream& out, const lock_rec_t& lock)
-{
-	return(lock.print(out));
-}
-
-/** Lock struct; protected by lock_sys.mutex */
-struct lock_t {
-	trx_t*		trx;		/*!< transaction owning the
-					lock */
-	UT_LIST_NODE_T(lock_t)
-			trx_locks;	/*!< list of the locks of the
-					transaction */
-
-	dict_index_t*	index;		/*!< index for a record lock */
-
-	lock_t*		hash;		/*!< hash chain node for a record
-					lock. The link node in a singly linked
-					list, used during hashing. */
-
-	/* Statistics for how long lock has been held and time
-	how long this lock had to be waited before it was granted */
-	time_t		requested_time; /*!< Lock request time */
-	ulint		wait_time;	/*!< Time waited this lock or 0 */
-
-	union {
-		lock_table_t	tab_lock;/*!< table lock */
-		lock_rec_t	rec_lock;/*!< record lock */
-	} un_member;			/*!< lock details */
-
-	ib_uint32_t	type_mode;	/*!< lock type, mode, LOCK_GAP or
-					LOCK_REC_NOT_GAP,
-					LOCK_INSERT_INTENTION,
-					wait flag, ORed */
-
-	/** Determine if the lock object is a record lock.
-	@return true if record lock, false otherwise. */
-	bool is_record_lock() const
-	{
-		return(type() == LOCK_REC);
-	}
-
-	bool is_waiting() const
-	{
-		return(type_mode & LOCK_WAIT);
-	}
-
-	bool is_gap() const
-	{
-		return(type_mode & LOCK_GAP);
-	}
-
-	bool is_record_not_gap() const
-	{
-		return(type_mode & LOCK_REC_NOT_GAP);
-	}
-
-	bool is_insert_intention() const
-	{
-		return(type_mode & LOCK_INSERT_INTENTION);
-	}
-
-	ulint type() const {
-		return(type_mode & LOCK_TYPE_MASK);
-	}
-
-	enum lock_mode mode() const
-	{
-		return(static_cast<enum lock_mode>(type_mode & LOCK_MODE_MASK));
-	}
-
-	/** Print the lock object into the given output stream.
-	@param[in,out]	out	the output stream
-	@return the given output stream. */
-	std::ostream& print(std::ostream& out) const;
-
-	/** Convert the member 'type_mode' into a human readable string.
-	@return human readable string */
-	std::string type_mode_string() const;
-
-	const char* type_string() const
-	{
-		switch (type_mode & LOCK_TYPE_MASK) {
-		case LOCK_REC:
-			return("LOCK_REC");
-		case LOCK_TABLE:
-			return("LOCK_TABLE");
-		default:
-			ut_error;
-		}
-	}
-};
-
 /** Convert the member 'type_mode' into a human readable string.
 @return human readable string */
 inline
 std::string
-lock_t::type_mode_string() const
+ib_lock_t::type_mode_string() const
 {
 	std::ostringstream sout;
 	sout << type_string();
@@ -227,7 +94,7 @@ lock_t::type_mode_string() const
 
 inline
 std::ostream&
-lock_t::print(std::ostream& out) const
+ib_lock_t::print(std::ostream& out) const
 {
 	out << "[lock_t: type_mode=" << type_mode << "("
 		<< type_mode_string() << ")";
@@ -244,7 +111,7 @@ lock_t::print(std::ostream& out) const
 
 inline
 std::ostream&
-operator<<(std::ostream& out, const lock_t& lock)
+operator<<(std::ostream& out, const ib_lock_t& lock)
 {
 	return(lock.print(out));
 }

storage/innobase/include/lock0priv.ic

 /*****************************************************************************
 
 Copyright (c) 2007, 2014, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2018, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -388,17 +389,10 @@ lock_table_has(
 	const dict_table_t*	table,	/*!< in: table */
 	lock_mode		in_mode)/*!< in: lock mode */
 {
-	if (trx->lock.table_locks.empty()) {
-		return(NULL);
-	}
-
-	typedef lock_pool_t::const_reverse_iterator iterator;
-
-	iterator	end = trx->lock.table_locks.rend();
-
 	/* Look for stronger locks the same trx already has on the table */
 
-	for (iterator it = trx->lock.table_locks.rbegin(); it != end; ++it) {
+	for (lock_list::const_iterator it = trx->lock.table_locks.begin(),
+             end = trx->lock.table_locks.end(); it != end; ++it) {
 
 		const lock_t*	lock = *it;
 

storage/innobase/include/lock0types.h

 /*****************************************************************************
 
 Copyright (c) 1996, 2015, Oracle and/or its affiliates. All Rights Reserved.
+Copyright (c) 2018, MariaDB Corporation.
 
 This program is free software; you can redistribute it and/or modify it under
 the terms of the GNU General Public License as published by the Free Software
@@ -72,6 +73,195 @@ const char* lock_mode_string(enum lock_mode mode)
 	}
 }
 
-typedef UT_LIST_BASE_NODE_T(lock_t) trx_lock_list_t;
+/** A table lock */
+struct lock_table_t {
+	dict_table_t*	table;		/*!< database table in dictionary
+					cache */
+	UT_LIST_NODE_T(ib_lock_t)
+			locks;		/*!< list of locks on the same
+					table */
+	/** Print the table lock into the given output stream
+	@param[in,out]	out	the output stream
+	@return the given output stream. */
+	std::ostream& print(std::ostream& out) const;
+};
+
+/** Record lock for a page */
+struct lock_rec_t {
+	ib_uint32_t	space;		/*!< space id */
+	ib_uint32_t	page_no;	/*!< page number */
+	ib_uint32_t	n_bits;		/*!< number of bits in the lock
+					bitmap; NOTE: the lock bitmap is
+					placed immediately after the
+					lock struct */
+
+	/** Print the record lock into the given output stream
+	@param[in,out]	out	the output stream
+	@return the given output stream. */
+	std::ostream& print(std::ostream& out) const;
+};
+
+/** Print the record lock into the given output stream
+@param[in,out]	out	the output stream
+@return the given output stream. */
+inline
+std::ostream& lock_rec_t::print(std::ostream& out) const
+{
+	out << "[lock_rec_t: space=" << space << ", page_no=" << page_no
+		<< ", n_bits=" << n_bits << "]";
+	return(out);
+}
+
+inline
+std::ostream&
+operator<<(std::ostream& out, const lock_rec_t& lock)
+{
+	return(lock.print(out));
+}
+
+#define LOCK_MODE_MASK	0xFUL	/*!< mask used to extract mode from the
+				type_mode field in a lock */
+/** Lock types */
+/* @{ */
+#define LOCK_TABLE	16U	/*!< table lock */
+#define	LOCK_REC	32U	/*!< record lock */
+#define LOCK_TYPE_MASK	0xF0UL	/*!< mask used to extract lock type from the
+				type_mode field in a lock */
+#if LOCK_MODE_MASK & LOCK_TYPE_MASK
+# error "LOCK_MODE_MASK & LOCK_TYPE_MASK"
+#endif
+
+#define LOCK_WAIT	256U	/*!< Waiting lock flag; when set, it
+				means that the lock has not yet been
+				granted, it is just waiting for its
+				turn in the wait queue */
+/* Precise modes */
+#define LOCK_ORDINARY	0	/*!< this flag denotes an ordinary
+				next-key lock in contrast to LOCK_GAP
+				or LOCK_REC_NOT_GAP */
+#define LOCK_GAP	512U	/*!< when this bit is set, it means that the
+				lock holds only on the gap before the record;
+				for instance, an x-lock on the gap does not
+				give permission to modify the record on which
+				the bit is set; locks of this type are created
+				when records are removed from the index chain
+				of records */
+#define LOCK_REC_NOT_GAP 1024U	/*!< this bit means that the lock is only on
+				the index record and does NOT block inserts
+				to the gap before the index record; this is
+				used in the case when we retrieve a record
+				with a unique key, and is also used in
+				locking plain SELECTs (not part of UPDATE
+				or DELETE) when the user has set the READ
+				COMMITTED isolation level */
+#define LOCK_INSERT_INTENTION 2048U/*!< this bit is set when we place a waiting
+				gap type record lock request in order to let
+				an insert of an index record to wait until
+				there are no conflicting locks by other
+				transactions on the gap; note that this flag
+				remains set when the waiting lock is granted,
+				or if the lock is inherited to a neighboring
+				record */
+#define LOCK_PREDICATE	8192U	/*!< Predicate lock */
+#define LOCK_PRDT_PAGE	16384U	/*!< Page lock */
+
+
+#if (LOCK_WAIT|LOCK_GAP|LOCK_REC_NOT_GAP|LOCK_INSERT_INTENTION|LOCK_PREDICATE|LOCK_PRDT_PAGE)&LOCK_MODE_MASK
+# error
+#endif
+#if (LOCK_WAIT|LOCK_GAP|LOCK_REC_NOT_GAP|LOCK_INSERT_INTENTION|LOCK_PREDICATE|LOCK_PRDT_PAGE)&LOCK_TYPE_MASK
+# error
+#endif
+/* @} */
+
+/** Lock struct; protected by lock_sys.mutex */
+struct ib_lock_t
+{
+	trx_t*		trx;		/*!< transaction owning the
+					lock */
+	UT_LIST_NODE_T(ib_lock_t)
+			trx_locks;	/*!< list of the locks of the
+					transaction */
+
+	dict_index_t*	index;		/*!< index for a record lock */
+
+	ib_lock_t*	hash;		/*!< hash chain node for a record
+					lock. The link node in a singly linked
+					list, used during hashing. */
+
+	/* Statistics for how long lock has been held and time
+	how long this lock had to be waited before it was granted */
+	time_t		requested_time; /*!< Lock request time */
+	ulint		wait_time;	/*!< Time waited this lock or 0 */
+
+	union {
+		lock_table_t	tab_lock;/*!< table lock */
+		lock_rec_t	rec_lock;/*!< record lock */
+	} un_member;			/*!< lock details */
+
+	ib_uint32_t	type_mode;	/*!< lock type, mode, LOCK_GAP or
+					LOCK_REC_NOT_GAP,
+					LOCK_INSERT_INTENTION,
+					wait flag, ORed */
+
+	/** Determine if the lock object is a record lock.
+	@return true if record lock, false otherwise. */
+	bool is_record_lock() const
+	{
+		return(type() == LOCK_REC);
+	}
+
+	bool is_waiting() const
+	{
+		return(type_mode & LOCK_WAIT);
+	}
+
+	bool is_gap() const
+	{
+		return(type_mode & LOCK_GAP);
+	}
+
+	bool is_record_not_gap() const
+	{
+		return(type_mode & LOCK_REC_NOT_GAP);
+	}
+
+	bool is_insert_intention() const
+	{
+		return(type_mode & LOCK_INSERT_INTENTION);
+	}
+
+	ulint type() const {
+		return(type_mode & LOCK_TYPE_MASK);
+	}
+
+	enum lock_mode mode() const
+	{
+		return(static_cast<enum lock_mode>(type_mode & LOCK_MODE_MASK));
+	}
+
+	/** Print the lock object into the given output stream.
+	@param[in,out]	out	the output stream
+	@return the given output stream. */
+	std::ostream& print(std::ostream& out) const;
+
+	/** Convert the member 'type_mode' into a human readable string.
+	@return human readable string */
+	std::string type_mode_string() const;
+
+	const char* type_string() const
+	{
+		switch (type_mode & LOCK_TYPE_MASK) {
+		case LOCK_REC:
+			return("LOCK_REC");
+		case LOCK_TABLE:
+			return("LOCK_TABLE");
+		default:
+			ut_error;
+		}
+	}
+};
+
+typedef UT_LIST_BASE_NODE_T(ib_lock_t) trx_lock_list_t;
 
 #endif /* lock0types_h */

storage/innobase/include/log0recv.h

@@ -139,9 +139,21 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply);
 /** Moves the parsing buffer data left to the buffer start. */
 void recv_sys_justify_left_parsing_buf();
 
-/** Backup function checks whether the space id belongs to
-the skip table list given in the mariabackup option. */
-extern bool(*check_if_backup_includes)(ulint space_id);
+/** Report optimized DDL operation (without redo log), corresponding to MLOG_INDEX_LOAD.
+@param[in]	space_id	tablespace identifier
+*/
+extern void(*log_optimized_ddl_op)(ulint space_id);
+
+/** Report an operation to create, delete, or rename a file during backup.
+@param[in]	space_id	tablespace identifier
+@param[in]	flags		tablespace flags (NULL if not create)
+@param[in]	name		file name (not NUL-terminated)
+@param[in]	len		length of name, in bytes
+@param[in]	new_name	new file name (NULL if not rename)
+@param[in]	new_len		length of new_name, in bytes (0 if NULL) */
+extern void (*log_file_op)(ulint space_id, const byte* flags,
+			   const byte* name, ulint len,
+			   const byte* new_name, ulint new_len);
 
 /** Block of log record data */
 struct recv_data_t{

storage/innobase/include/trx0sys.h

@@ -646,8 +646,11 @@ class rw_trx_hash_t
     {
       mutex_enter(&element->mutex);
       lf_hash_search_unpin(pins);
-      if ((trx= element->trx))
-      {
+      trx= element->trx;
+      if (!trx);
+      else if (UNIV_UNLIKELY(trx_id != trx->id))
+        trx= NULL;
+      else {
         if (do_ref_count)
           trx->reference();
         ut_d(validate_element(trx));

storage/innobase/include/trx0trx.h

@@ -476,7 +476,9 @@ Check transaction state */
 @param t transaction handle */
 #define	assert_trx_is_free(t)	do {					\
 	ut_ad(trx_state_eq((t), TRX_STATE_NOT_STARTED));		\
-	ut_ad(!trx->has_logged());					\
+	ut_ad(!(t)->id);						\
+	ut_ad(!(t)->has_logged());					\
+	ut_ad(!(t)->is_referenced());					\
 	ut_ad(!(t)->read_view.is_open());				\
 	ut_ad((t)->lock.wait_thr == NULL);				\
 	ut_ad(UT_LIST_GET_LEN((t)->lock.trx_locks) == 0);		\
@@ -517,7 +519,7 @@ The transaction must have mysql_thd assigned. */
 # define assert_trx_nonlocking_or_in_list(trx) ((void)0)
 #endif /* UNIV_DEBUG */
 
-typedef std::vector<ib_lock_t*, ut_allocator<ib_lock_t*> >	lock_pool_t;
+typedef std::vector<ib_lock_t*, ut_allocator<ib_lock_t*> >	lock_list;
 
 /*******************************************************************//**
 Latching protocol for trx_lock_t::que_state.  trx_lock_t::que_state
@@ -579,13 +581,19 @@ struct trx_lock_t {
 					only be modified by the thread that is
 					serving the running transaction. */
 
-	lock_pool_t	rec_pool;	/*!< Pre-allocated record locks */
+	/** Pre-allocated record locks */
+	struct {
+		ib_lock_t lock; byte pad[256];
+	} rec_pool[8];
 
-	lock_pool_t	table_pool;	/*!< Pre-allocated table locks */
+	/** Pre-allocated table locks */
+	ib_lock_t	table_pool[8];
 
-	ulint		rec_cached;	/*!< Next free rec lock in pool */
+	/** Next available rec_pool[] entry */
+	unsigned	rec_cached;
 
-	ulint		table_cached;	/*!< Next free table lock in pool */
+	/** Next available table_pool[] entry */
+	unsigned	table_cached;
 
 	mem_heap_t*	lock_heap;	/*!< memory heap for trx_locks;
 					protected by lock_sys.mutex */
@@ -595,7 +603,7 @@ struct trx_lock_t {
 					and lock_sys.mutex; removals are
 					protected by lock_sys.mutex */
 
-	lock_pool_t	table_locks;	/*!< All table locks requested by this
+	lock_list	table_locks;	/*!< All table locks requested by this
 					transaction, including AUTOINC locks */
 
 	bool		cancel;		/*!< true if the transaction is being

storage/innobase/include/ut0pool.h

@@ -125,8 +125,7 @@ struct Pool {
 			elem = NULL;
 		}
 
-		m_lock_strategy.exit();
-
+#if defined HAVE_valgrind || defined __SANITIZE_ADDRESS__
 		if (elem) {
 			/* Unpoison the memory for AddressSanitizer */
 			MEM_UNDEFINED(&elem->m_type, sizeof elem->m_type);
@@ -135,10 +134,11 @@ struct Pool {
 			actually initialized; we checked that by
 			UNIV_MEM_ASSERT_RW() in mem_free() below. */
 			UNIV_MEM_VALID(&elem->m_type, sizeof elem->m_type);
-			return &elem->m_type;
 		}
+#endif
 
-		return NULL;
+		m_lock_strategy.exit();
+		return elem ? &elem->m_type : NULL;
 	}
 
 	/** Add the object to the pool.
@@ -151,8 +151,12 @@ struct Pool {
 		elem = reinterpret_cast<Element*>(p - sizeof(*elem));
 		UNIV_MEM_ASSERT_RW(&elem->m_type, sizeof elem->m_type);
 
-		elem->m_pool->put(elem);
+		elem->m_pool->m_lock_strategy.enter();
+
+		elem->m_pool->putl(elem);
 		MEM_NOACCESS(&elem->m_type, sizeof elem->m_type);
+
+		elem->m_pool->m_lock_strategy.exit();
 	}
 
 protected:
@@ -170,17 +174,13 @@ struct Pool {
 
 	/** Release the object to the free pool
 	@param elem element to free */
-	void put(Element* elem)
+	void putl(Element* elem)
 	{
-		m_lock_strategy.enter();
-
 		ut_ad(elem >= m_start && elem < m_last);
 
 		ut_ad(Factory::debug(&elem->m_type));
 
 		m_pqueue.push(elem);
-
-		m_lock_strategy.exit();
 	}
 
 	/** Initialise the elements.

storage/innobase/lock/lock0lock.cc

@@ -59,18 +59,6 @@ ulong innodb_lock_schedule_algorithm;
 /** The value of innodb_deadlock_detect */
 my_bool	innobase_deadlock_detect;
 
-/** Total number of cached record locks */
-static const ulint	REC_LOCK_CACHE = 8;
-
-/** Maximum record lock size in bytes */
-static const ulint	REC_LOCK_SIZE = sizeof(ib_lock_t) + 256;
-
-/** Total number of cached table locks */
-static const ulint	TABLE_LOCK_CACHE = 8;
-
-/** Size in bytes, of the table lock instance */
-static const ulint	TABLE_LOCK_SIZE = sizeof(ib_lock_t);
-
 /*********************************************************************//**
 Checks if a waiting record lock request still has to wait in a queue.
 @return lock that is causing the wait */
@@ -1409,13 +1397,13 @@ lock_rec_create_low(
 		}
 	}
 
- 	if (trx->lock.rec_cached >= trx->lock.rec_pool.size()
-	    || sizeof *lock + n_bytes > REC_LOCK_SIZE) {
+	if (trx->lock.rec_cached >= UT_ARR_SIZE(trx->lock.rec_pool)
+	    || sizeof *lock + n_bytes > sizeof *trx->lock.rec_pool) {
 		lock = static_cast<lock_t*>(
 			mem_heap_alloc(trx->lock.lock_heap,
 				       sizeof *lock + n_bytes));
 	} else {
-		lock = trx->lock.rec_pool[trx->lock.rec_cached++];
+		lock = &trx->lock.rec_pool[trx->lock.rec_cached++].lock;
 	}
 
 	lock->trx = trx;
@@ -3520,8 +3508,9 @@ lock_table_create(
 
 		ib_vector_push(trx->autoinc_locks, &lock);
 
-	} else if (trx->lock.table_cached < trx->lock.table_pool.size()) {
-		lock = trx->lock.table_pool[trx->lock.table_cached++];
+	} else if (trx->lock.table_cached
+		   < UT_ARR_SIZE(trx->lock.table_pool)) {
+		lock = &trx->lock.table_pool[trx->lock.table_cached++];
 	} else {
 
 		lock = static_cast<lock_t*>(
@@ -4373,24 +4362,15 @@ lock_trx_table_locks_remove(
 		ut_ad(trx_mutex_own(trx));
 	}
 
-	typedef lock_pool_t::reverse_iterator iterator;
-
-	iterator	end = trx->lock.table_locks.rend();
-
-	for (iterator it = trx->lock.table_locks.rbegin(); it != end; ++it) {
-
+	for (lock_list::iterator it = trx->lock.table_locks.begin(),
+             end = trx->lock.table_locks.end(); it != end; ++it) {
 		const lock_t*	lock = *it;
 
-		if (lock == NULL) {
-			continue;
-		}
-
-		ut_a(trx == lock->trx);
-		ut_a(lock_get_type_low(lock) & LOCK_TABLE);
-		ut_a(lock->un_member.tab_lock.table != NULL);
+		ut_ad(!lock || trx == lock->trx);
+		ut_ad(!lock || lock_get_type_low(lock) & LOCK_TABLE);
+		ut_ad(!lock || lock->un_member.tab_lock.table);
 
 		if (lock == lock_to_remove) {
-
 			*it = NULL;
 
 			if (!trx->lock.cancel) {
@@ -4807,11 +4787,8 @@ lock_trx_table_locks_find(
 
 	trx_mutex_enter(trx);
 
-	typedef lock_pool_t::const_reverse_iterator iterator;
-
-	iterator	end = trx->lock.table_locks.rend();
-
-	for (iterator it = trx->lock.table_locks.rbegin(); it != end; ++it) {
+	for (lock_list::const_iterator it = trx->lock.table_locks.begin(),
+             end = trx->lock.table_locks.end(); it != end; ++it) {
 
 		const lock_t*	lock = *it;
 
@@ -6337,6 +6314,9 @@ lock_trx_release_locks(
 	/*--------------------------------------*/
 	trx_mutex_enter(trx);
 	trx->state = TRX_STATE_COMMITTED_IN_MEMORY;
+	/* Ensure that rw_trx_hash_t::find() will no longer find
+	this transaction. */
+	trx->id = 0;
 	trx_mutex_exit(trx);
 	/*--------------------------------------*/
 
@@ -6547,10 +6527,8 @@ lock_trx_has_sys_table_locks(
 
 	lock_mutex_enter();
 
-	typedef lock_pool_t::const_reverse_iterator iterator;
-
-	iterator	end = trx->lock.table_locks.rend();
-	iterator	it = trx->lock.table_locks.rbegin();
+	const lock_list::const_iterator end = trx->lock.table_locks.end();
+	lock_list::const_iterator it = trx->lock.table_locks.begin();
 
 	/* Find a valid mode. Note: ib_vector_size() can be 0. */
 
@@ -7102,33 +7080,6 @@ DeadlockChecker::check_and_resolve(const lock_t* lock, trx_t* trx)
 	return(victim_trx);
 }
 
-/**
-Allocate cached locks for the transaction.
-@param trx		allocate cached record locks for this transaction */
-void
-lock_trx_alloc_locks(trx_t* trx)
-{
-	ulint	sz = REC_LOCK_SIZE * REC_LOCK_CACHE;
-	byte*	ptr = reinterpret_cast<byte*>(ut_malloc_nokey(sz));
-
-	/* We allocate one big chunk and then distribute it among
-	the rest of the elements. The allocated chunk pointer is always
-	at index 0. */
-
-	for (ulint i = 0; i < REC_LOCK_CACHE; ++i, ptr += REC_LOCK_SIZE) {
-		trx->lock.rec_pool.push_back(
-			reinterpret_cast<ib_lock_t*>(ptr));
-	}
-
-	sz = TABLE_LOCK_SIZE * TABLE_LOCK_CACHE;
-	ptr = reinterpret_cast<byte*>(ut_malloc_nokey(sz));
-
-	for (ulint i = 0; i < TABLE_LOCK_CACHE; ++i, ptr += TABLE_LOCK_SIZE) {
-		trx->lock.table_pool.push_back(
-			reinterpret_cast<ib_lock_t*>(ptr));
-	}
-
-}
 /*************************************************************//**
 Updates the lock table when a page is split and merged to
 two pages. */

storage/innobase/log/log0recv.cc

@@ -169,9 +169,21 @@ typedef std::map<
 
 static recv_spaces_t	recv_spaces;
 
-/** Backup function checks whether the space id belongs to
-the skip table list given in the mariabackup option. */
-bool(*check_if_backup_includes)(ulint space_id);
+/** Report optimized DDL operation (without redo log), corresponding to MLOG_INDEX_LOAD.
+@param[in]	space_id	tablespace identifier
+*/
+void (*log_optimized_ddl_op)(ulint space_id);
+
+/** Report an operation to create, delete, or rename a file during backup.
+@param[in]	space_id	tablespace identifier
+@param[in]	flags		tablespace flags (NULL if not create)
+@param[in]	name		file name (not NUL-terminated)
+@param[in]	len		length of name, in bytes
+@param[in]	new_name	new file name (NULL if not rename)
+@param[in]	new_len		length of new_name, in bytes (0 if NULL) */
+void (*log_file_op)(ulint space_id, const byte* flags,
+		    const byte* name, ulint len,
+		    const byte* new_name, ulint new_len);
 
 /** Process a file name from a MLOG_FILE_* record.
 @param[in,out]	name		file name
@@ -381,9 +393,13 @@ fil_name_parse(
 
 		fil_name_process(
 			reinterpret_cast<char*>(ptr), len, space_id, true);
-
-		break;
+		/* fall through */
 	case MLOG_FILE_CREATE2:
+		if (log_file_op) {
+			log_file_op(space_id,
+				    type == MLOG_FILE_CREATE2 ? ptr - 4 : NULL,
+				    ptr, len, NULL, 0);
+		}
 		break;
 	case MLOG_FILE_RENAME2:
 		if (corrupt) {
@@ -424,6 +440,11 @@ fil_name_parse(
 			reinterpret_cast<char*>(new_name), new_len,
 			space_id, false);
 
+		if (log_file_op) {
+			log_file_op(space_id, NULL,
+				    ptr, len, new_name, new_len);
+		}
+
 		if (!apply) {
 			break;
 		}
@@ -716,6 +737,15 @@ bool log_t::files::read_log_seg(lsn_t* start_lsn, lsn_t end_lsn)
 					  OS_FILE_LOG_BLOCK_SIZE, true);
 			}
 		}
+
+		ulint dl = log_block_get_data_len(buf);
+		if (dl < LOG_BLOCK_HDR_SIZE
+		    || (dl > OS_FILE_LOG_BLOCK_SIZE - LOG_BLOCK_TRL_SIZE
+			&& dl != OS_FILE_LOG_BLOCK_SIZE)) {
+			recv_sys->found_corrupt_log = true;
+			end_lsn = *start_lsn;
+			break;
+		}
 	}
 
 	if (recv_sys->report(ut_time())) {
@@ -2125,7 +2155,8 @@ recv_parse_log_rec(
 	case MLOG_MULTI_REC_END | MLOG_SINGLE_REC_FLAG:
 	case MLOG_DUMMY_RECORD | MLOG_SINGLE_REC_FLAG:
 	case MLOG_CHECKPOINT | MLOG_SINGLE_REC_FLAG:
-		ib::error() << "Incorrect log record type:" << *ptr;
+		ib::error() << "Incorrect log record type "
+			<< ib::hex(unsigned(*ptr));
 		recv_sys->found_corrupt_log = true;
 		return(0);
 	}
@@ -2144,7 +2175,6 @@ recv_parse_log_rec(
 		*type, new_ptr, end_ptr, *space, *page_no, apply, NULL, NULL);
 
 	if (UNIV_UNLIKELY(new_ptr == NULL)) {
-
 		return(0);
 	}
 
@@ -2201,30 +2231,30 @@ recv_report_corrupt_log(
 	ib::error() <<
 		"############### CORRUPT LOG RECORD FOUND ##################";
 
+	const ulint ptr_offset = ulint(ptr - recv_sys->buf);
+
 	ib::info() << "Log record type " << type << ", page " << space << ":"
 		<< page_no << ". Log parsing proceeded successfully up to "
 		<< recv_sys->recovered_lsn << ". Previous log record type "
 		<< recv_previous_parsed_rec_type << ", is multi "
 		<< recv_previous_parsed_rec_is_multi << " Recv offset "
-		<< (ptr - recv_sys->buf) << ", prev "
+		<< ptr_offset << ", prev "
 		<< recv_previous_parsed_rec_offset;
 
 	ut_ad(ptr <= recv_sys->buf + recv_sys->len);
 
 	const ulint	limit	= 100;
-	const ulint	before
-		= std::min(recv_previous_parsed_rec_offset, limit);
-	const ulint	after
-		= std::min(recv_sys->len - ulint(ptr - recv_sys->buf), limit);
+	const ulint	prev_offset = std::min(recv_previous_parsed_rec_offset,
+					       ptr_offset);
+	const ulint	before = std::min(prev_offset, limit);
+	const ulint	after = std::min(recv_sys->len - ptr_offset, limit);
 
 	ib::info() << "Hex dump starting " << before << " bytes before and"
 		" ending " << after << " bytes after the corrupted record:";
 
-	ut_print_buf(stderr,
-		     recv_sys->buf
-		     + recv_previous_parsed_rec_offset - before,
-		     ulint(ptr - recv_sys->buf) + before + after
-		     - recv_previous_parsed_rec_offset);
+	const byte* start = recv_sys->buf + prev_offset - before;
+
+	ut_print_buf(stderr, start, ulint(ptr - start) + after);
 	putc('\n', stderr);
 
 	if (!srv_force_recovery) {
@@ -2295,13 +2325,8 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply)
 		len = recv_parse_log_rec(&type, ptr, end_ptr, &space,
 					 &page_no, apply, &body);
 
-		if (len == 0) {
-			return(false);
-		}
-
 		if (recv_sys->found_corrupt_log) {
-			recv_report_corrupt_log(
-				ptr, type, space, page_no);
+			recv_report_corrupt_log(ptr, type, space, page_no);
 			return(true);
 		}
 
@@ -2309,6 +2334,10 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply)
 			return(true);
 		}
 
+		if (len == 0) {
+			return(false);
+		}
+
 		new_recovered_lsn = recv_calc_lsn_on_data_add(old_lsn, len);
 
 		if (new_recovered_lsn > recv_sys->scanned_lsn) {
@@ -2396,11 +2425,8 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply)
 			/* fall through */
 		case MLOG_INDEX_LOAD:
 			if (type == MLOG_INDEX_LOAD) {
-				if (check_if_backup_includes
-				    && !check_if_backup_includes(space)) {
-					ut_ad(srv_operation
-					      == SRV_OPERATION_BACKUP);
-					return true;
+				if (log_optimized_ddl_op) {
+					log_optimized_ddl_op(space);
 				}
 			}
 			/* fall through */
@@ -2433,13 +2459,10 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply)
 				&type, ptr, end_ptr, &space, &page_no,
 				false, &body);
 
-			if (len == 0) {
-				return(false);
-			}
-
 			if (recv_sys->found_corrupt_log
 			    || type == MLOG_CHECKPOINT
-			    || (*ptr & MLOG_SINGLE_REC_FLAG)) {
+			    || (ptr != end_ptr
+				&& (*ptr & MLOG_SINGLE_REC_FLAG))) {
 				recv_sys->found_corrupt_log = true;
 				recv_report_corrupt_log(
 					ptr, type, space, page_no);
@@ -2450,6 +2473,10 @@ bool recv_parse_log_recs(lsn_t checkpoint_lsn, store_t store, bool apply)
 				return(true);
 			}
 
+			if (len == 0) {
+				return(false);
+			}
+
 			recv_previous_parsed_rec_type = type;
 			recv_previous_parsed_rec_offset
 				= recv_sys->recovered_offset + total_len;

storage/innobase/mtr/mtr0log.cc

@@ -98,7 +98,11 @@ mlog_parse_initial_log_record(
 	}
 
 	*type = mlog_id_t(*ptr & ~MLOG_SINGLE_REC_FLAG);
-	ut_ad(*type <= MLOG_BIGGEST_TYPE || EXTRA_CHECK_MLOG_NUMBER(*type));
+	if (UNIV_UNLIKELY(*type > MLOG_BIGGEST_TYPE
+			  && !EXTRA_CHECK_MLOG_NUMBER(*type))) {
+		recv_sys->found_corrupt_log = true;
+		return NULL;
+	}
 
 	ptr++;
 

storage/innobase/page/page0cur.cc

@@ -2256,7 +2256,10 @@ page_cur_parse_delete_rec(
 	offset = mach_read_from_2(ptr);
 	ptr += 2;
 
-	ut_a(offset <= srv_page_size);
+	if (UNIV_UNLIKELY(offset >= srv_page_size)) {
+		recv_sys->found_corrupt_log = true;
+		return NULL;
+	}
 
 	if (block) {
 		page_t*		page		= buf_block_get_frame(block);

storage/innobase/trx/trx0trx.cc

@@ -118,8 +118,6 @@ trx_init(
 /*=====*/
 	trx_t*	trx)
 {
-	trx->id = 0;
-
 	trx->no = TRX_ID_MAX;
 
 	trx->state = TRX_STATE_NOT_STARTED;
@@ -197,11 +195,7 @@ struct TrxFactory {
 		the constructors of the trx_t members. */
 		new(&trx->mod_tables) trx_mod_tables_t();
 
-		new(&trx->lock.rec_pool) lock_pool_t();
-
-		new(&trx->lock.table_pool) lock_pool_t();
-
-		new(&trx->lock.table_locks) lock_pool_t();
+		new(&trx->lock.table_locks) lock_list();
 
 		new(&trx->read_view) ReadView();
 
@@ -225,8 +219,6 @@ struct TrxFactory {
 			&trx_named_savept_t::trx_savepoints);
 
 		mutex_create(LATCH_ID_TRX, &trx->mutex);
-
-		lock_trx_alloc_locks(trx);
 	}
 
 	/** Release resources held by the transaction object.
@@ -256,27 +248,7 @@ struct TrxFactory {
 
 		ut_ad(!trx->read_view.is_open());
 
-		if (!trx->lock.rec_pool.empty()) {
-
-			/* See lock_trx_alloc_locks() why we only free
-			the first element. */
-
-			ut_free(trx->lock.rec_pool[0]);
-		}
-
-		if (!trx->lock.table_pool.empty()) {
-
-			/* See lock_trx_alloc_locks() why we only free
-			the first element. */
-
-			ut_free(trx->lock.table_pool[0]);
-		}
-
-		trx->lock.rec_pool.~lock_pool_t();
-
-		trx->lock.table_pool.~lock_pool_t();
-
-		trx->lock.table_locks.~lock_pool_t();
+		trx->lock.table_locks.~lock_list();
 
 		trx->read_view.~ReadView();
 	}
@@ -412,7 +384,12 @@ trx_t *trx_create()
 
 	/* Should have been either just initialized or .clear()ed by
 	trx_free(). */
-	ut_a(trx->mod_tables.size() == 0);
+	ut_ad(trx->mod_tables.empty());
+	ut_ad(trx->lock.table_locks.empty());
+	ut_ad(UT_LIST_GET_LEN(trx->lock.trx_locks) == 0);
+	ut_ad(trx->lock.n_rec_locks == 0);
+	ut_ad(trx->lock.table_cached == 0);
+	ut_ad(trx->lock.rec_cached == 0);
 
 #ifdef WITH_WSREP
 	trx->wsrep_event = NULL;
@@ -993,8 +970,6 @@ trx_start_low(
 
 		trx_sys.register_rw(trx);
 	} else {
-		trx->id = 0;
-
 		if (!trx_is_autocommit_non_locking(trx)) {
 
 			/* If this is a read-only transaction that is writing
@@ -1250,9 +1225,6 @@ trx_update_mod_tables_timestamp(
 /*============================*/
 	trx_t*	trx)	/*!< in: transaction */
 {
-
-	ut_ad(trx->id != 0);
-
 	/* consider using trx->start_time if calling time() is too
 	expensive here */
 	time_t	now = ut_time();
@@ -1325,7 +1297,10 @@ trx_commit_in_memory(
 			trx_sys.deregister_rw(trx);
 		}
 
+		/* trx->id will be cleared in lock_trx_release_locks(trx). */
+		ut_ad(trx->read_only || !trx->rsegs.m_redo.rseg || trx->id);
 		lock_trx_release_locks(trx);
+		ut_ad(trx->id == 0);
 
 		/* Remove the transaction from the list of active
 		transactions now that it no longer holds any user locks. */

storage/rocksdb/build_rocksdb.cmake

@@ -376,9 +376,31 @@ SET(SOURCES)
 FOREACH(s ${ROCKSDB_SOURCES})
   list(APPEND SOURCES ${ROCKSDB_SOURCE_DIR}/${s})
 ENDFOREACH()
-IF(MSVC)
+
+if(MSVC)
   add_definitions(-DHAVE_SSE42 -DHAVE_PCLMUL)
-ENDIF()
+else()
+  set(CMAKE_REQUIRED_FLAGS "-msse4.2 -mpclmul ${CXX11_FLAGS}")
+
+  CHECK_CXX_SOURCE_COMPILES("
+#include <cstdint>
+#include <nmmintrin.h>
+#include <wmmintrin.h>
+int main() {
+  volatile uint32_t x = _mm_crc32_u32(0, 0);
+  const auto a = _mm_set_epi64x(0, 0);
+  const auto b = _mm_set_epi64x(0, 0);
+  const auto c = _mm_clmulepi64_si128(a, b, 0x00);
+  auto d = _mm_cvtsi128_si64(c);
+}
+" HAVE_SSE42)
+  if(HAVE_SSE42)
+    set_source_files_properties(${ROCKSDB_SOURCE_DIR}/util/crc32c.cc
+      PROPERTIES COMPILE_FLAGS "-DHAVE_SSE42 -DHAVE_PCLMUL -msse4.2 -mpclmul")
+  endif()
+  unset(CMAKE_REQUIRED_FLAGS)
+endif()
+
 IF(CMAKE_VERSION VERSION_GREATER "2.8.10")
   STRING(TIMESTAMP GIT_DATE_TIME "%Y-%m-%d %H:%M:%S")
 ENDIF()

storage/rocksdb/mysql-test/rocksdb/t/autoinc_debug-master.opt

---binlog-format=row
+--binlog-format=row --rocksdb-flush-log-at-trx-commit=1

storage/rocksdb/mysql-test/rocksdb/t/bulk_load_errors.test

@@ -60,6 +60,13 @@ SELECT * FROM t1;
 
 --source include/wait_until_count_sessions.inc
 
+# Note: in MariaDB, session count will be decremented *before* 
+#   myrocks::rocksdb_close_connection is called. This causes a race condition:
+#   we may grep the error log before bulk load is finalized.
+#   To prevent that, do a soft restart of the server (I wasnt able to find
+#   any other reliable way)
+--source include/restart_mysqld_with_option.inc
+
 --let SEARCH_FILE=$LOG2
 --let SEARCH_PATTERN=RocksDB: Error [0-9]+ finalizing last SST file while disconnecting
 --source include/search_pattern_in_file.inc