mariadb/extra/mariabackup/ddl_log.cc

#include "ddl_log.h"
#include "common.h"
#include "my_sys.h"
#include "sql_table.h"
#include "backup_copy.h"
#include "xtrabackup.h"
#include <unordered_set>
#include <functional>
#include <memory>
#include <cstddef>

namespace ddl_log {

struct Entry {
	enum Type {
		CREATE,
		ALTER,
		RENAME,
		REPAIR,
		OPTIMIZE,
		DROP,
		TRUNCATE,
		CHANGE_INDEX,
		BULK_INSERT
	};
	Type type;
	std::string date;
	std::string engine;
	bool partitioned;
	std::string db;
	std::string table;
	std::string id;
	std::string new_engine;
	bool new_partitioned;
	std::string new_db;
	std::string new_table;
	std::string new_id;
};

typedef std::vector<std::unique_ptr<Entry>> entries_t;
typedef std::function<bool(std::unique_ptr<Entry>)> store_entry_func_t;

const char *aria_engine_name = "Aria";
static const char *frm_ext = ".frm";
static const char *database_keyword = "DATABASE";

const std::unordered_map<std::string, std::vector<const char *>> engine_exts =
{
	{"Aria", {".MAD", ".MAI"}},
	{"MyISAM", {".MYD", ".MYI"}},
	{"MRG_MyISAM", {".MRG"}},
	{"ARCHIVE", {".ARM", ".ARZ"}},
	{"CSV", {".CSM", ".CSV"}}
};

static inline bool known_engine(const std::string &engine) {
	return engine_exts.count(engine);
}

// TODO: add error messages
size_t parse(const uchar *buf, size_t buf_size, bool &error_flag,
	store_entry_func_t &store_entry_func) {
	DBUG_ASSERT(buf);
	static constexpr char token_delimiter = '\t';
	static constexpr char line_delimiter = '\n';
	enum {
		TOKEN_FIRST = 0,
		TOKEN_DATE = TOKEN_FIRST,
		TOKEN_TYPE,
		TOKEN_ENGINE,
		TOKEN_PARTITIONED,
		TOKEN_DB,
		TOKEN_TABLE,
		TOKEN_ID,
		TOKEN_MANDATORY = TOKEN_ID,
		TOKEN_NEW_ENGINE,
		TOKEN_NEW_PARTITIONED,
		TOKEN_NEW_DB,
		TOKEN_NEW_TABLE,
		TOKEN_NEW_ID,
		TOKEN_LAST = TOKEN_NEW_ID
	};
	const size_t string_offsets[TOKEN_LAST + 1] = {
		offsetof(Entry, date),
		offsetof(Entry, type), // not a string, be careful
		offsetof(Entry, engine),
		offsetof(Entry, partitioned), // not a string, be careful
		offsetof(Entry, db),
		offsetof(Entry, table),
		offsetof(Entry, id),
		offsetof(Entry, new_engine),
		offsetof(Entry, new_partitioned), // not a string, be careful
		offsetof(Entry, new_db),
		offsetof(Entry, new_table),
		offsetof(Entry, new_id)
	};
	const std::unordered_map<std::string, Entry::Type> str_to_type = {
		{"CREATE", Entry::CREATE},
		{"ALTER", Entry::ALTER},
		{"RENAME", Entry::RENAME},
		// TODO: fix to use uppercase-only
		{"repair", Entry::REPAIR},
		{"optimize", Entry::OPTIMIZE},
		{"DROP", Entry::DROP},
		{"TRUNCATE", Entry::TRUNCATE},
		{"CHANGE_INDEX", Entry::CHANGE_INDEX},
		{"BULK_INSERT", Entry::BULK_INSERT}
	};

	const uchar *new_line = buf;
	const uchar *token_start = buf;
	unsigned token_num = TOKEN_FIRST;

	error_flag = false;

	std::unique_ptr<Entry> entry(new Entry());

	for (const uchar *ptr = buf; ptr < buf + buf_size; ++ptr) {

		if (*ptr != token_delimiter && *ptr != line_delimiter)
			continue;

		if (token_start != ptr) {
			std::string token(token_start, ptr);

			if (token_num == TOKEN_TYPE) {
				const auto type_it = str_to_type.find(token);
				if (type_it == str_to_type.end()) {
					error_flag = true;
					goto exit;
				}
				entry->type = type_it->second;
			}
			else if (token_num == TOKEN_PARTITIONED) {
				entry->partitioned = token[0] - '0';
			}
			else if (token_num == TOKEN_NEW_PARTITIONED) {
				entry->new_partitioned = token[0] - '0';
			}
			else if (token_num <= TOKEN_LAST) {
				DBUG_ASSERT(token_num != TOKEN_TYPE);
				DBUG_ASSERT(token_num != TOKEN_PARTITIONED);
				DBUG_ASSERT(token_num != TOKEN_NEW_PARTITIONED);
				reinterpret_cast<std::string *>
					(reinterpret_cast<uchar *>(entry.get()) + string_offsets[token_num])->
					assign(std::move(token));
			}
			else {
				error_flag = true;
				goto exit;
			}
		}
		token_start = ptr + 1;

		if (*ptr == line_delimiter) {
			if (token_num < TOKEN_MANDATORY) {
				error_flag = true;
				goto exit;
			}
			if (!store_entry_func(std::move(entry))) {
				error_flag = true;
				goto exit;
			}
			entry.reset(new Entry());
			token_num = TOKEN_FIRST;
			new_line = ptr + 1;
		} else
			++token_num;
	}

exit:
	return new_line - buf;
}

bool parse(const char *file_path, store_entry_func_t store_entry_func) {
	DBUG_ASSERT(file_path);
	DBUG_ASSERT(store_entry_func);
	File file= -1;
	bool result = true;
	uchar buf[1024];
	size_t bytes_read = 0;
	size_t buf_read_offset = 0;

	if ((file= my_open(file_path, O_RDONLY | O_SHARE | O_NOFOLLOW | O_CLOEXEC,
		MYF(MY_WME))) < 0) {
		msg("DDL log file %s open failed: %d", file_path, my_errno);
		result = false;
		goto exit;
	}

	while((bytes_read = my_read(
		file, &buf[buf_read_offset], sizeof(buf) - buf_read_offset, MY_WME)) > 0) {
		if (bytes_read == size_t(-1)) {
			msg("DDL log file %s read error: %d", file_path, my_errno);
			result = false;
			break;
		}
		bytes_read += buf_read_offset;
		bool parse_error_flag = false;
		size_t bytes_parsed = parse(
			buf, bytes_read, parse_error_flag, store_entry_func);
		if (parse_error_flag) {
			result = false;
			break;
		}
		size_t rest_size = bytes_read - bytes_parsed;
		if (rest_size)
			memcpy(buf, buf + bytes_parsed, rest_size);
		buf_read_offset = rest_size;
	}

exit:
	if (file >= 0)
		my_close(file, MYF(MY_WME));
	return result;
};


static
bool process_database(
	const char *datadir_path,
	ds_ctxt_t *ds,
	const Entry &entry,
	std::unordered_set<std::string> &dropped_databases) {

	if (entry.type == Entry::Type::CREATE ||
		entry.type == Entry::Type::ALTER) {
		std::string opt_file(datadir_path);
		opt_file.append("/").append(entry.db).append("/db.opt");
		if (!ds->copy_file(opt_file.c_str(), opt_file.c_str(), 0, true)) {
			msg("Failed to re-copy %s.", opt_file.c_str());
			return false;
		}
		if (entry.type == Entry::Type::CREATE)
			dropped_databases.erase(entry.db);
		return true;
	}

	DBUG_ASSERT(entry.type == Entry::Type::DROP);

	std::string db_path(datadir_path);
	db_path.append("/").append(entry.db);
	const char	*dst_path = convert_dst(db_path.c_str());
	if (!ds_remove(ds, dst_path)) {
		dropped_databases.insert(entry.db);
		return true;
	}
	return false;
}

static
std::unique_ptr<std::vector<std::string>>
	find_table_files(
		const char *dir_path,
		const std::string &db,
		const std::string &table) {

	std::unique_ptr<std::vector<std::string>>
		result(new std::vector<std::string>());

	std::string prefix = convert_tablename_to_filepath(dir_path, db, table);
	foreach_file_in_db_dirs(dir_path, [&](const char *file_name)->bool {
		if (!strncmp(file_name, prefix.c_str(), prefix.size())) {
			DBUG_ASSERT(strlen(file_name) >= prefix.size());
			if (file_name[prefix.size()] == '.' ||
				!strncmp(file_name + prefix.size(), "#P#", strlen("#P#")))
				result->push_back(std::string(file_name));
		}
		return true;
	});

	return result;
}

static
bool process_remove(
	const char *datadir_path,
	ds_ctxt_t *ds,
	const Entry &entry,
	bool remove_frm) {

	if (check_if_skip_table(
		std::string(entry.db).append("/").append(entry.table).c_str()))
		return true;

	auto ext_it = engine_exts.find(entry.engine);
	if (ext_it == engine_exts.end())
		return true;

	std::string file_preffix = convert_tablename_to_filepath(datadir_path,
		entry.db, entry.table);
	const char *dst_preffix = convert_dst(file_preffix.c_str());

	for (const char *ext : ext_it->second) {
		std::string old_name(dst_preffix);
		if (!entry.partitioned)
			old_name.append(ext);
		else
			old_name.append("#P#*");
		if (ds_remove(ds, old_name.c_str())) {
			msg("Failed to remove %s.", old_name.c_str());
			return false;
		}
	}

	if (remove_frm) {
		std::string old_frm_name(dst_preffix);
		old_frm_name.append(frm_ext);
		if (ds_remove(ds, old_frm_name.c_str())) {
			msg("Failed to remove %s.", old_frm_name.c_str());
			return false;
		}
	}
	return true;

}

static
bool process_recopy(
	const char *datadir_path,
	ds_ctxt_t *ds,
	const Entry &entry,
	const tables_t &tables) {

	if (check_if_skip_table(
		std::string(entry.db).append("/").append(entry.table).c_str()))
		return true;

	const std::string &new_table_id =
		entry.new_id.empty() ? entry.id : entry.new_id;
	DBUG_ASSERT(!new_table_id.empty());
	const std::string &new_table =
		entry.new_table.empty() ? entry.table : entry.new_table;
	DBUG_ASSERT(!new_table.empty());
	const std::string &new_db =
		entry.new_db.empty() ? entry.db : entry.new_db;
	DBUG_ASSERT(!new_db.empty());
	const std::string &new_engine =
		entry.new_engine.empty() ? entry.engine : entry.new_engine;
	DBUG_ASSERT(!new_engine.empty());

	if (entry.type != Entry::Type::BULK_INSERT) {
		auto table_it = tables.find(table_key(new_db, new_table));
		if (table_it != tables.end() &&
			table_it->second == new_table_id)
			return true;
	}

	if (!entry.new_engine.empty() &&
		entry.engine != entry.new_engine &&
		!known_engine(entry.new_engine)) {
		return process_remove(datadir_path, ds, entry, false);
	}

	if ((entry.partitioned || entry.new_partitioned) &&
		!process_remove(datadir_path, ds, entry, false))
		return false;

	if (entry.partitioned || entry.new_partitioned) {
		auto files = find_table_files(datadir_path, new_db, new_table);
		if (!files.get())
			return true;
		for (const auto &file : *files) {
			const char *dst_path = convert_dst(file.c_str());
			if (!ds->copy_file(file.c_str(), dst_path, 0, true)) {
				msg("Failed to re-copy %s.", file.c_str());
				return false;
			}
		}
		return true;
	}

	auto ext_it = engine_exts.find(new_engine);
	if (ext_it == engine_exts.end())
		return false;

	for (const char *ext : ext_it->second) {
		std::string file_name =
			convert_tablename_to_filepath(datadir_path, new_db, new_table).
				append(ext);
		const char *dst_path = convert_dst(file_name.c_str());
		if (file_exists(file_name.c_str()) &&
			!ds->copy_file(file_name.c_str(), dst_path, 0, true)) {
			msg("Failed to re-copy %s.", file_name.c_str());
			return false;
		}
	}

	std::string frm_file =
		convert_tablename_to_filepath(datadir_path, new_db, new_table).
			append(frm_ext);
	const char *frm_dst_path = convert_dst(frm_file.c_str());
	if (file_exists(frm_file.c_str()) &&
		!ds->copy_file(frm_file.c_str(), frm_dst_path, 0, true)) {
		msg("Failed to re-copy %s.", frm_file.c_str());
		return false;
	}

	return true;
}

static
bool process_rename(
	const char *datadir_path,
	ds_ctxt_t *ds,
	const Entry &entry) {

	if (check_if_skip_table(
		std::string(entry.db).append("/").append(entry.table).c_str()))
		return true;

	DBUG_ASSERT(entry.db != "partition");

	auto ext_it = engine_exts.find(entry.engine);
	if (ext_it == engine_exts.end())
		return false;

	std::string new_preffix = convert_tablename_to_filepath(datadir_path,
		entry.new_db, entry.new_table);
	const char	*dst_path = convert_dst(new_preffix.c_str());

	std::string old_preffix = convert_tablename_to_filepath(datadir_path,
		entry.db, entry.table);
	const char	*src_path = convert_dst(old_preffix.c_str());

	for (const char *ext : ext_it->second) {
		std::string old_name(src_path);
		old_name.append(ext);
		std::string new_name(dst_path);
		new_name.append(ext);
		if (ds_rename(ds, old_name.c_str(), new_name.c_str())) {
			msg("Failed to rename %s to %s.",
				old_name.c_str(), new_name.c_str());
			return false;
		}
	}

	std::string new_frm_file = new_preffix + frm_ext;
	const char	*new_frm_dst = convert_dst(new_frm_file.c_str());
	if (file_exists(new_frm_file.c_str()) &&
		!ds->copy_file(new_frm_file.c_str(), new_frm_dst, 0, true)) {
		msg("Failed to re-copy %s.", new_frm_file.c_str());
		return false;
	}

// TODO: return this code if .frm is copied not under BLOCK_DDL
/*
	std::string old_frm_name(src_path);
	old_frm_name.append(frm_ext);
	std::string new_frm_name(dst_path);
	new_frm_name.append(frm_ext);
	if (ds_rename(ds, old_frm_name.c_str(), new_frm_name.c_str())) {
		msg("Failed to rename %s to %s.",
			old_frm_name.c_str(), new_frm_name.c_str());
		return false;
	}
*/
	return true;
}

bool backup(
	const char *datadir_path,
	ds_ctxt_t *ds,
	const tables_t &tables) {
	DBUG_ASSERT(datadir_path);
	DBUG_ASSERT(ds);
	char ddl_log_path[FN_REFLEN];
	fn_format(ddl_log_path, "ddl", datadir_path, ".log", 0);
	std::vector<std::unique_ptr<Entry>> entries;

	std::unordered_set<std::string> processed_tables;
	std::unordered_set<std::string> dropped_databases;

	bool parsing_result =
		parse(ddl_log_path, [&](std::unique_ptr<Entry> entry)->bool {

			if (entry->engine == database_keyword)
				return process_database(datadir_path, ds, *entry, dropped_databases);

			if (!known_engine(entry->engine) && !known_engine(entry->new_engine))
				return true;

			if (entry->type == Entry::Type::CREATE ||
					(entry->type == Entry::Type::ALTER &&
					!entry->new_engine.empty() &&
					entry->engine != entry->new_engine)) {
				if (!process_recopy(datadir_path, ds, *entry, tables))
					return false;
				processed_tables.insert(table_key(entry->db, entry->table));
				if (entry->type == Entry::Type::ALTER)
					processed_tables.insert(table_key(entry->new_db, entry->new_table));
				return true;
			}

			if (entry->type == Entry::Type::DROP) {
				if (!process_remove(datadir_path, ds, *entry, true))
					return false;
				processed_tables.insert(table_key(entry->db, entry->table));
				return true;
			}
			if (entry->type == Entry::Type::RENAME) {
				if (entry->partitioned) {
					if (!process_remove(datadir_path, ds, *entry, true))
						return false;
					Entry recopy_entry {
						entry->type,
						{},
						entry->new_engine.empty() ? entry->engine : entry->new_engine,
						true,
						entry->new_db,
						entry->new_table,
						entry->new_id,
						{}, true, {}, {}, {}
					};
					if (!process_recopy(datadir_path, ds, recopy_entry, tables))
						return false;
				}
				else if (!process_rename(datadir_path, ds, *entry))
					return false;
				processed_tables.insert(table_key(entry->db, entry->table));
				processed_tables.insert(table_key(entry->new_db, entry->new_table));
				return true;
			}

			entries.push_back(std::move(entry));
			return true;

		});

	if (!parsing_result)
		return false;


	while (!entries.empty()) {
		auto entry = std::move(entries.back());
		entries.pop_back();
		auto tk = table_key(
			entry->new_db.empty() ? entry->db : entry->new_db,
			entry->new_table.empty() ? entry->table : entry->new_table);
		if (dropped_databases.count(entry->db) ||
			dropped_databases.count(entry->new_db))
			continue;
		if (processed_tables.count(tk))
			continue;
		processed_tables.insert(std::move(tk));
		if (!process_recopy(datadir_path, ds, *entry, tables))
			return false;
	}

	return true;
}

} // namespace ddl_log