#include <TableArchiver.h>

Collaboration diagram for TableArchiver:

Public Member Functions
	TableArchiver (Catalog_Namespace::Catalog *cat)

void	dumpTable (const TableDescriptor *td, const std::string &archive_path, const std::string &compression)

void	restoreTable (const Catalog_Namespace::SessionInfo &session, const TableDescriptor *td, const std::string &archive_path, const std::string &compression)

void	restoreTable (const Catalog_Namespace::SessionInfo &session, const std::string &table_name, const std::string &archive_path, const std::string &compression)

Private Attributes
Catalog_Namespace::Catalog *	cat_

Detailed Description

Definition at line 24 of file TableArchiver.h.

Constructor & Destructor Documentation

TableArchiver::TableArchiver ( Catalog_Namespace::Catalog * cat )

inline

Definition at line 26 of file TableArchiver.h.

26 : cat_(cat){};

TableArchiver::cat_

Catalog_Namespace::Catalog * cat_

Definition: TableArchiver.h:43

Member Function Documentation

void TableArchiver::dumpTable	(	const TableDescriptor *	td,
		const std::string &	archive_path,
		const std::string &	compression
	)

Definition at line 240 of file TableArchiver.cpp.

Referenced by Parser::DumpTableStmt::execute().

                                                             {
   filename_security_check(archive_path);
   if (g_cluster) {
     throw std::runtime_error("DUMP/RESTORE is not supported yet on distributed setup.");
   }
   if (boost::filesystem::exists(archive_path)) {
     throw std::runtime_error("Archive " + archive_path + " already exists.");
   }
   if (td->isView || td->persistenceLevel != Data_Namespace::MemoryLevel::DISK_LEVEL) {
     throw std::runtime_error("Dumping view or temporary table is not supported.");
   }
   // collect paths of files to archive
   const auto global_file_mgr = cat_->getDataMgr().getGlobalFileMgr();
   std::vector<std::string> file_paths;
   auto file_writer = [&file_paths, global_file_mgr](const std::string& file_name,
                                                     const std::string& file_type,
                                                     const std::string& file_data) {
     const auto file_path = abs_path(global_file_mgr) + "/" + file_name;
     std::unique_ptr<FILE, decltype(simple_file_closer)> fp(
         std::fopen(file_path.c_str(), "w"), simple_file_closer);
     if (!fp) {
       throw std::runtime_error("Failed to create " + file_type + " file '" + file_path +
                                "': " + std::strerror(errno));
     }
     if (std::fwrite(file_data.data(), 1, file_data.size(), fp.get()) < file_data.size()) {
       throw std::runtime_error("Failed to write " + file_type + " file '" + file_path +
                                "': " + std::strerror(errno));
     }
     file_paths.push_back(file_name);
   };
   // Prevent modification of the table schema during a dump operation, while allowing
   // concurrent inserts.
   auto table_read_lock =
       lockmgr::TableSchemaLockMgr::getReadLockForTable(*cat_, td->tableName);
   const auto table_name = td->tableName;
   {
     // - gen schema file
     const auto schema_str = cat_->dumpSchema(td);
     file_writer(table_schema_filename, "table schema", schema_str);
     // - gen column-old-info file
     const auto cds = cat_->getAllColumnMetadataForTable(td->tableId, true, true, true);
     std::vector<std::string> column_oldinfo;
     std::transform(cds.begin(),
                    cds.end(),
                    std::back_inserter(column_oldinfo),
                    [&](const auto cd) -> std::string {
                      return cd->columnName + ":" + std::to_string(cd->columnId) + ":" +
                             cat_->getColumnDictDirectory(cd);
                    });
     const auto column_oldinfo_str = boost::algorithm::join(column_oldinfo, " ");
     file_writer(table_oldinfo_filename, "table old info", column_oldinfo_str);
     // - gen table epoch
     const auto epoch = cat_->getTableEpoch(cat_->getCurrentDB().dbId, td->tableId);
     file_writer(table_epoch_filename, "table epoch", std::to_string(epoch));
     // - collect table data file paths ...
     const auto data_file_dirs = cat_->getTableDataDirectories(td);
     file_paths.insert(file_paths.end(), data_file_dirs.begin(), data_file_dirs.end());
     // - collect table dict file paths ...
     const auto dict_file_dirs = cat_->getTableDictDirectories(td);
     file_paths.insert(file_paths.end(), dict_file_dirs.begin(), dict_file_dirs.end());
     // tar takes time. release cat lock to yield the cat to concurrent CREATE statements.
   }
   // run tar to archive the files ... this may take a while !!
   run("tar " + compression + " -cvf " + get_quoted_string(archive_path) + " " +
           boost::algorithm::join(file_paths, " "),
       abs_path(global_file_mgr));
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void TableArchiver::restoreTable	(	const Catalog_Namespace::SessionInfo &	session,
		const TableDescriptor *	td,
		const std::string &	archive_path,
		const std::string &	compression
	)

Definition at line 311 of file TableArchiver.cpp.

Referenced by Parser::RestoreTableStmt::execute(), and restoreTable().

                                                                {
   filename_security_check(archive_path);
   if (g_cluster) {
     throw std::runtime_error("DUMP/RESTORE is not supported yet on distributed setup.");
   }
   if (!boost::filesystem::exists(archive_path)) {
     throw std::runtime_error("Archive " + archive_path + " does not exist.");
   }
   if (td->isView || td->persistenceLevel != Data_Namespace::MemoryLevel::DISK_LEVEL) {
     throw std::runtime_error("Restoring view or temporary table is not supported.");
   }
   // Obtain table schema read lock to prevent modification of the schema during
   // restoration
   const auto table_read_lock =
       lockmgr::TableSchemaLockMgr::getReadLockForTable(*cat_, td->tableName);
   // prevent concurrent inserts into table during restoration
   const auto insert_data_lock =
       lockmgr::InsertDataLockMgr::getWriteLockForTable(*cat_, td->tableName);
 
   // untar takes time. no grab of cat lock to yield to concurrent CREATE stmts.
   const auto global_file_mgr = cat_->getDataMgr().getGlobalFileMgr();
   // dirs where src files are untarred and dst files are backed up
   constexpr static const auto temp_data_basename = "_data";
   constexpr static const auto temp_back_basename = "_back";
   const auto temp_data_dir = abs_path(global_file_mgr) + "/" + temp_data_basename;
   const auto temp_back_dir = abs_path(global_file_mgr) + "/" + temp_back_basename;
   // clean up tmp dirs and files in any case
   auto tmp_files_cleaner = [&](void*) {
     run("rm -rf " + temp_data_dir + " " + temp_back_dir);
     run("rm -f " + abs_path(global_file_mgr) + "/" + table_schema_filename);
     run("rm -f " + abs_path(global_file_mgr) + "/" + table_oldinfo_filename);
     run("rm -f " + abs_path(global_file_mgr) + "/" + table_epoch_filename);
   };
   std::unique_ptr<decltype(tmp_files_cleaner), decltype(tmp_files_cleaner)> tfc(
       &tmp_files_cleaner, tmp_files_cleaner);
   // extract & parse schema
   const auto schema_str = get_table_schema(archive_path, td->tableName, compression);
   const auto create_table_stmt =
       Parser::parseDDL<Parser::CreateTableStmt>("table schema", schema_str);
   // verify compatibility between source and destination schemas
   TableDescriptor src_td;
   std::list<ColumnDescriptor> src_columns;
   std::vector<Parser::SharedDictionaryDef> shared_dict_defs;
   create_table_stmt->executeDryRun(session, src_td, src_columns, shared_dict_defs);
   // - sanity check table-level compatibility
   if (src_td.hasDeletedCol != td->hasDeletedCol) {
     // TODO: allow the case, in which src data enables vacuum while
     // dst doesn't, by simply discarding src $deleted column data.
     throw std::runtime_error("Incompatible table VACCUM option");
   }
   if (src_td.nShards != td->nShards) {
     // TODO: allow different shard numbers if they have a "GCD",
     // by splitting/merging src data files before drop into dst.
     throw std::runtime_error("Unmatched number of table shards");
   }
   // - sanity check column-level compatibility (based on column names)
   const auto dst_columns =
       cat_->getAllColumnMetadataForTable(td->tableId, false, false, false);
   if (dst_columns.size() != src_columns.size()) {
     throw std::runtime_error("Unmatched number of table columns");
   }
   for (const auto& [src_cd, dst_cd] : boost::combine(src_columns, dst_columns)) {
     if (src_cd.columnType.get_type_name() != dst_cd->columnType.get_type_name() ||
         src_cd.columnType.get_compression_name() !=
             dst_cd->columnType.get_compression_name()) {
       throw std::runtime_error("Incompatible types on column " + src_cd.columnName);
     }
   }
   // extract src table column ids (ALL columns incl. system/virtual/phy geo cols)
   const auto all_src_oldinfo_str =
       simple_file_cat(archive_path, table_oldinfo_filename, compression);
   std::vector<std::string> src_oldinfo_strs;
   boost::algorithm::split(src_oldinfo_strs,
                           all_src_oldinfo_str,
                           boost::is_any_of(" "),
                           boost::token_compress_on);
   auto all_dst_columns =
       cat_->getAllColumnMetadataForTable(td->tableId, true, true, true);
   if (src_oldinfo_strs.size() != all_dst_columns.size()) {
     throw std::runtime_error("Source table has a unmatched number of columns: " +
                              std::to_string(src_oldinfo_strs.size()) + " vs " +
                              std::to_string(all_dst_columns.size()));
   }
   // build a map of src column ids and dst column ids, just in case src table has been
   // ALTERed before and chunk keys of src table needs to be adjusted accordingly.
   // note: this map is used only for the case of migrating a table and not for restoring
   // a table. When restoring a table, the two tables must have the same column ids.
   //
   // also build a map of src dict paths and dst dict paths for relocating src dicts
   std::unordered_map<int, int> column_ids_map;
   std::unordered_map<std::string, std::string> dict_paths_map;
   // sort inputs of transform in lexical order of column names for correct mappings
   std::list<std::vector<std::string>> src_oldinfo_tokens;
   std::transform(
       src_oldinfo_strs.begin(),
       src_oldinfo_strs.end(),
       std::back_inserter(src_oldinfo_tokens),
       [](const auto& src_oldinfo_str) -> auto {
         std::vector<std::string> tokens;
         boost::algorithm::split(
             tokens, src_oldinfo_str, boost::is_any_of(":"), boost::token_compress_on);
         return tokens;
       });
   src_oldinfo_tokens.sort(
       [](const auto& lhs, const auto& rhs) { return lhs[0].compare(rhs[0]) < 0; });
   all_dst_columns.sort(
       [](auto a, auto b) { return a->columnName.compare(b->columnName) < 0; });
   // transform inputs into the maps
   std::transform(src_oldinfo_tokens.begin(),
                  src_oldinfo_tokens.end(),
                  all_dst_columns.begin(),
                  std::inserter(column_ids_map, column_ids_map.end()),
                  [&](const auto& tokens, const auto& cd) -> std::pair<int, int> {
                    VLOG(3) << boost::algorithm::join(tokens, ":") << " ==> "
                            << cd->columnName << ":" << cd->columnId;
                    dict_paths_map[tokens[2]] = cat_->getColumnDictDirectory(cd);
                    return {boost::lexical_cast<int>(tokens[1]), cd->columnId};
                  });
   bool was_table_altered = false;
   std::for_each(column_ids_map.begin(), column_ids_map.end(), [&](auto& it) {
     was_table_altered = was_table_altered || it.first != it.second;
   });
   VLOG(3) << "was_table_altered = " << was_table_altered;
   // extract all data files to a temp dir. will swap with dst table dir after all set,
   // otherwise will corrupt table in case any bad thing happens in the middle.
   run("rm -rf " + temp_data_dir);
   run("mkdir -p " + temp_data_dir);
   run("tar " + compression + " -xvf " + get_quoted_string(archive_path), temp_data_dir);
   // if table was ever altered after it was created, update column ids in chunk headers.
   if (was_table_altered) {
     const auto time_ms = measure<>::execution(
         [&]() { adjust_altered_table_files(temp_data_dir, column_ids_map); });
     VLOG(3) << "adjust_altered_table_files: " << time_ms << " ms";
   }
   // finally,,, swap table data/dict dirs!
   const auto data_file_dirs = cat_->getTableDataDirectories(td);
   const auto dict_file_dirs = cat_->getTableDictDirectories(td);
   // move current target dirs, if exists, to backup dir
   std::vector<std::string> both_file_dirs;
   std::merge(data_file_dirs.begin(),
              data_file_dirs.end(),
              dict_file_dirs.begin(),
              dict_file_dirs.end(),
              std::back_inserter(both_file_dirs));
   bool backup_completed = false;
   try {
     run("rm -rf " + temp_back_dir);
     run("mkdir -p " + temp_back_dir);
     for (const auto& dir : both_file_dirs) {
       const auto dir_full_path = abs_path(global_file_mgr) + "/" + dir;
       if (boost::filesystem::is_directory(dir_full_path)) {
         run("mv " + dir_full_path + " " + temp_back_dir);
       }
     }
     backup_completed = true;
     // accord src data dirs to dst
     rename_table_directories(
         cat_->getDataMgr().getGlobalFileMgr(), temp_data_dir, data_file_dirs, "table_");
     // accord src dict dirs to dst
     for (const auto& dit : dict_paths_map) {
       if (!dit.first.empty() && !dit.second.empty()) {
         const auto src_dict_path = temp_data_dir + "/" + dit.first;
         const auto dst_dict_path = abs_path(global_file_mgr) + "/" + dit.second;
         run("mv " + src_dict_path + " " + dst_dict_path);
       }
     }
     // throw if sanity test forces a rollback
     if (g_test_rollback_dump_restore) {
       throw std::runtime_error("lol!");
     }
   } catch (...) {
     // once backup is completed, whatever in abs_path(global_file_mgr) is the "src"
     // dirs that are to be rolled back and discarded
     if (backup_completed) {
       run("rm -rf " + boost::algorithm::join(both_file_dirs, " "),
           abs_path(global_file_mgr));
     }
     // complete rollback by recovering original "dst" table dirs from backup dir
     boost::filesystem::path base_path(temp_back_dir);
     boost::filesystem::directory_iterator end_it;
     for (boost::filesystem::directory_iterator fit(base_path); fit != end_it; ++fit) {
       run("mv " + fit->path().string() + " .", abs_path(global_file_mgr));
     }
     throw;
   }
   // set for reloading table from the restored/migrated files
   const auto epoch = simple_file_cat(archive_path, table_epoch_filename, compression);
   cat_->setTableEpoch(
       cat_->getCurrentDB().dbId, td->tableId, boost::lexical_cast<int>(epoch));
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void TableArchiver::restoreTable	(	const Catalog_Namespace::SessionInfo &	session,
		const std::string &	table_name,
		const std::string &	archive_path,
		const std::string &	compression
	)

Definition at line 507 of file TableArchiver.cpp.

References cat_, anonymous_namespace{TableArchiver.cpp}::get_table_schema(), Catalog_Namespace::Catalog::getMetadataForTable(), and restoreTable().

                                                                {
   // replace table name and drop foreign dict references
   const auto schema_str = get_table_schema(archive_path, table_name, compression);
   Parser::parseDDL<Parser::CreateTableStmt>("table schema", schema_str)->execute(session);
   try {
     restoreTable(
         session, cat_->getMetadataForTable(table_name), archive_path, compression);
   } catch (...) {
     Parser::parseDDL<Parser::DropTableStmt>("statement",
                                             "DROP TABLE IF EXISTS " + table_name + ";")
         ->execute(session);
     throw;
   }
 }

Here is the call graph for this function:

Member Data Documentation

Catalog_Namespace::Catalog* TableArchiver::cat_

private

Definition at line 43 of file TableArchiver.h.

Referenced by dumpTable(), and restoreTable().

The documentation for this class was generated from the following files:

/home/jenkins-slave/workspace/core-os-doxygen/TableArchiver/TableArchiver.h
/home/jenkins-slave/workspace/core-os-doxygen/TableArchiver/TableArchiver.cpp

Public Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation