#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 std::string &table_name, const std::string &archive_path, const std::string &compression, const TableArchiverS3Options &s3_options)

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

Private Attributes
Catalog_Namespace::Catalog *	cat_

Detailed Description

Definition at line 32 of file TableArchiver.h.

Constructor & Destructor Documentation

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

inline

Definition at line 34 of file TableArchiver.h.

34 : cat_(cat){};

TableArchiver::cat_

Catalog_Namespace::Catalog * cat_

Definition: TableArchiver.h:52

Member Function Documentation

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

Definition at line 335 of file TableArchiver.cpp.

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

                                                             {
   if (td->is_system_table) {
     throw std::runtime_error("Dumping a system table is not supported.");
   }
   ddl_utils::validate_allowed_file_path(archive_path,
                                         ddl_utils::DataTransferType::EXPORT);
   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.");
   }
   // create a unique uuid for this table dump
   std::string uuid = boost::uuids::to_string(boost::uuids::random_generator()());
 
   // 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, uuid](const std::string& file_name,
                                          const std::string& file_type,
                                          const std::string& file_data) {
     std::unique_ptr<FILE, decltype(simple_file_closer)> fp(
         std::fopen(file_name.c_str(), "w"), simple_file_closer);
     if (!fp) {
       throw std::runtime_error("Failed to create " + file_type + " file '" + file_name +
                                "': " + 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_name +
                                "': " + std::strerror(errno));
     }
     file_paths.push_back(uuid / std::filesystem::path(file_name).filename());
   };
 
   const auto file_mgr_dir = std::filesystem::path(abs_path(global_file_mgr));
   const auto uuid_dir = file_mgr_dir / uuid;
 
   if (!std::filesystem::create_directory(uuid_dir)) {
     throw std::runtime_error("Failed to create work directory '" + uuid_dir.string() +
                              "' while dumping table.");
   }
 
   ScopeGuard cleanup_guard = [&] {
     if (std::filesystem::exists(uuid_dir)) {
       std::filesystem::remove_all(uuid_dir);
     }
   };
 
   const auto table_name = td->tableName;
   {
     // - gen schema file
     const auto schema_str = cat_->dumpSchema(td);
     file_writer(uuid_dir / 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(uuid_dir / table_oldinfo_filename, "table old info", column_oldinfo_str);
     // - gen table epoch
     const auto epoch = cat_->getTableEpoch(cat_->getCurrentDB().dbId, td->tableId);
     file_writer(uuid_dir / 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 + " --transform=s|" + uuid +
           std::filesystem::path::preferred_separator + "|| -cvf " +
           get_quoted_string(archive_path) + " " + boost::algorithm::join(file_paths, " "),
       file_mgr_dir);
 }

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,
		const TableArchiverS3Options &	s3_options
	)

Definition at line 670 of file TableArchiver.cpp.

References cat_, CHECK, Parser::create_stmt_for_query(), anonymous_namespace{TableArchiver.cpp}::get_table_schema(), Catalog_Namespace::Catalog::getMetadataForTable(), and shared::is_s3_uri().

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

                                                                            {
   auto local_archive_path = archive_path;
 #ifdef HAVE_AWS_S3
   const auto restore_dir_path = get_restore_dir_path();
   ScopeGuard archive_cleanup_guard = [&archive_path, &restore_dir_path] {
     if (shared::is_s3_uri(archive_path) && std::filesystem::exists(restore_dir_path)) {
       std::filesystem::remove_all(restore_dir_path);
     }
   };
   if (shared::is_s3_uri(archive_path)) {
     local_archive_path = download_s3_file(archive_path, s3_options, restore_dir_path);
   }
 #endif
 
   // replace table name and drop foreign dict references
   const auto schema_str = get_table_schema(local_archive_path, table_name, compression);
   std::unique_ptr<Parser::Stmt> stmt = Parser::create_stmt_for_query(schema_str, session);
   const auto create_table_stmt = dynamic_cast<Parser::CreateTableStmt*>(stmt.get());
   CHECK(create_table_stmt);
   create_table_stmt->execute(session, false /*read-only*/);
 
   try {
     restoreTable(
         session, cat_->getMetadataForTable(table_name), local_archive_path, compression);
   } catch (...) {
     const auto schema_str = "DROP TABLE IF EXISTS " + table_name + ";";
     std::unique_ptr<Parser::Stmt> stmt =
         Parser::create_stmt_for_query(schema_str, session);
     const auto drop_table_stmt = dynamic_cast<Parser::DropTableStmt*>(stmt.get());
     CHECK(drop_table_stmt);
     drop_table_stmt->execute(session, false /*read-only*/);
 
     throw;
   }
 }

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
	)

private

Definition at line 424 of file TableArchiver.cpp.

References anonymous_namespace{Utm.h}::a, anonymous_namespace{TableArchiver.cpp}::abs_path(), anonymous_namespace{TableArchiver.cpp}::adjust_altered_table_files(), cat_, CHECK, Parser::create_stmt_for_query(), Catalog_Namespace::DBMetadata::dbId, Data_Namespace::DISK_LEVEL, measure< TimeT >::execution(), g_cluster, g_test_rollback_dump_restore, get_quoted_string(), anonymous_namespace{TableArchiver.cpp}::get_table_schema(), Catalog_Namespace::Catalog::getAllColumnMetadataForTable(), Catalog_Namespace::Catalog::getColumnDictDirectory(), Catalog_Namespace::Catalog::getCurrentDB(), Catalog_Namespace::Catalog::getDataMgr(), Data_Namespace::DataMgr::getGlobalFileMgr(), lockmgr::TableLockMgrImpl< TableSchemaLockMgr >::getReadLockForTable(), Catalog_Namespace::Catalog::getTableDataDirectories(), Catalog_Namespace::Catalog::getTableDictDirectories(), lockmgr::TableLockMgrImpl< T >::getWriteLockForTable(), TableDescriptor::hasDeletedCol, ddl_utils::IMPORT, TableDescriptor::isView, join(), TableDescriptor::nShards, TableDescriptor::persistenceLevel, anonymous_namespace{TableArchiver.cpp}::rename_table_directories(), run, Catalog_Namespace::Catalog::setTableEpoch(), anonymous_namespace{TableArchiver.cpp}::simple_file_cat(), split(), table_epoch_filename, table_oldinfo_filename, TableDescriptor::tableId, TableDescriptor::tableName, to_string(), shared::transform(), ddl_utils::validate_allowed_file_path(), and VLOG.

                                                                {
   ddl_utils::validate_allowed_file_path(archive_path,
                                         ddl_utils::DataTransferType::IMPORT);
   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();
 
   // create a unique uuid for this table restore
   std::string uuid = boost::uuids::to_string(boost::uuids::random_generator()());
 
   const auto uuid_dir = std::filesystem::path(abs_path(global_file_mgr)) / uuid;
 
   if (!std::filesystem::create_directory(uuid_dir)) {
     throw std::runtime_error("Failed to create work directory '" + uuid_dir.string() +
                              "' while restoring table.");
   }
 
   ScopeGuard cleanup_guard = [&] {
     if (std::filesystem::exists(uuid_dir)) {
       std::filesystem::remove_all(uuid_dir);
     }
   };
 
   // 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 = uuid_dir / temp_data_basename;
   const auto temp_back_dir = uuid_dir / temp_back_basename;
 
   // extract & parse schema
   const auto schema_str = get_table_schema(archive_path, td->tableName, compression);
   std::unique_ptr<Parser::Stmt> stmt = Parser::create_stmt_for_query(schema_str, session);
   const auto create_table_stmt = dynamic_cast<Parser::CreateTableStmt*>(stmt.get());
   CHECK(create_table_stmt);
 
   // 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.string());
   run("mkdir -p " + temp_data_dir.string());
   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 epoch = boost::lexical_cast<int32_t>(
         simple_file_cat(archive_path, table_epoch_filename, compression));
     const auto time_ms = measure<>::execution(
         [&]() { adjust_altered_table_files(epoch, 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.string());
     run("mkdir -p " + temp_back_dir.string());
     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.string());
       }
     }
     backup_completed = true;
     // Move table directories from temp dir to main data directory.
     rename_table_directories(global_file_mgr, temp_data_dir, data_file_dirs, "table_");
     // Move dictionaries from temp dir to main dir.
     for (const auto& dit : dict_paths_map) {
       if (!dit.first.empty() && !dit.second.empty()) {
         const auto src_dict_path = temp_data_dir.string() + "/" + 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:

Member Data Documentation

Catalog_Namespace::Catalog* TableArchiver::cat_

private

Definition at line 52 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
/home/jenkins-slave/workspace/core-os-doxygen/TableArchiver/TableArchiverWindows.cpp

Public Member Functions

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation