Functions
std::string	abs_path (const File_Namespace::GlobalFileMgr *global_file_mgr)

std::string	run (const std::string &cmd, const std::string &chdir="", const bool log_failure=true)

std::string	simple_file_cat (const std::string &archive_path, const std::string &file_name, const std::string &compression, const bool log_failure=true)

std::string	get_table_schema (const std::string &archive_path, const std::string &table, const std::string &compression)

void	update_or_drop_column_ids_in_page_headers (const boost::filesystem::path &path, const std::unordered_map< int, int > &column_ids_map, const int32_t table_epoch, const bool drop_not_update)

void	update_or_drop_column_ids_in_table_files (const int32_t table_epoch, const std::string &temp_data_dir, const std::unordered_map< int, int > &column_ids_map, const bool drop_not_update)

void	delete_old_symlinks (const std::string &table_data_dir)

void	add_data_file_symlinks (const std::string &table_data_dir)

void	rename_table_directories (const File_Namespace::GlobalFileMgr *global_file_mgr, const std::string &temp_data_dir, const std::vector< std::string > &target_paths, const std::string &name_prefix)

std::unordered_map< int, int >	find_render_group_columns (const std::list< ColumnDescriptor > &src_columns, std::vector< std::string > &src_oldinfo_strs, const std::string &archive_path)

void	drop_render_group_columns (const std::unordered_map< int, int > &render_group_column_ids, const std::string &archive_path, const std::string &temp_data_dir, const std::string &compression)

Variables
auto	simple_file_closer = [](FILE* f) { std::fclose(f); }

Function Documentation

std::string anonymous_namespace{TableArchiver.cpp}::abs_path ( const File_Namespace::GlobalFileMgr * global_file_mgr )

inline

Definition at line 82 of file TableArchiver.cpp.

References File_Namespace::GlobalFileMgr::getBasePath().

Referenced by TableArchiver::dumpTable(), rename_table_directories(), and TableArchiver::restoreTable().

                                                                               {
   return boost::filesystem::canonical(global_file_mgr->getBasePath()).string();
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void anonymous_namespace{TableArchiver.cpp}::add_data_file_symlinks ( const std::string & table_data_dir )

Definition at line 322 of file TableArchiver.cpp.

References DATA_FILE_EXT, and File_Namespace::kLegacyDataFileExtension.

Referenced by rename_table_directories().

                                                              {
   std::map<boost::filesystem::path, boost::filesystem::path> old_to_new_paths;
   for (boost::filesystem::directory_iterator it(table_data_dir), end_it; it != end_it;
        it++) {
     const auto path = boost::filesystem::canonical(it->path());
     if (path.extension().string() == DATA_FILE_EXT) {
       auto old_path = path;
       old_path.replace_extension(File_Namespace::kLegacyDataFileExtension);
       // Add a symlink to data file, if one does not exist.
       if (!boost::filesystem::exists(old_path)) {
         old_to_new_paths[old_path] = path;
       }
     }
   }
   for (const auto& [old_path, new_path] : old_to_new_paths) {
     boost::filesystem::create_symlink(new_path.filename(), old_path);
   }
 }

Here is the caller graph for this function:

void anonymous_namespace{TableArchiver.cpp}::delete_old_symlinks ( const std::string & table_data_dir )

Definition at line 309 of file TableArchiver.cpp.

Referenced by rename_table_directories().

                                                           {
   std::vector<boost::filesystem::path> symlinks;
   for (boost::filesystem::directory_iterator it(table_data_dir), end_it; it != end_it;
        it++) {
     if (boost::filesystem::is_symlink(it->path())) {
       symlinks.emplace_back(it->path());
     }
   }
   for (const auto& symlink : symlinks) {
     boost::filesystem::remove_all(symlink);
   }
 }

Here is the caller graph for this function:

void anonymous_namespace{TableArchiver.cpp}::drop_render_group_columns	(	const std::unordered_map< int, int > &	render_group_column_ids,
		const std::string &	archive_path,
		const std::string &	temp_data_dir,
		const std::string &	compression
	)

Definition at line 420 of file TableArchiver.cpp.

References measure< TimeT >::execution(), simple_file_cat(), table_epoch_filename, update_or_drop_column_ids_in_table_files(), and VLOG.

Referenced by TableArchiver::restoreTable().

                                   {
   // rewrite page files to drop the columns with IDs that are the keys of the map
   if (render_group_column_ids.size()) {
     const auto epoch = boost::lexical_cast<int32_t>(
         simple_file_cat(archive_path, table_epoch_filename, compression));
     const auto time_ms = measure<>::execution([&]() {
       update_or_drop_column_ids_in_table_files(
           epoch, temp_data_dir, render_group_column_ids, true /* drop */);
     });
     VLOG(3) << "drop render group columns: " << time_ms << " ms";
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::unordered_map<int, int> anonymous_namespace{TableArchiver.cpp}::find_render_group_columns	(	const std::list< ColumnDescriptor > &	src_columns,
		std::vector< std::string > &	src_oldinfo_strs,
		const std::string &	archive_path
	)

Definition at line 370 of file TableArchiver.cpp.

References logger::INFO, kMULTIPOLYGON, kPOLYGON, LOG, and split().

Referenced by TableArchiver::restoreTable().

                                    {
   // scan for poly or mpoly columns and collect their names
   std::vector<std::string> poly_column_names;
   for (auto const& src_column : src_columns) {
     auto const sqltype = src_column.columnType.get_type();
     if (sqltype == kPOLYGON || sqltype == kMULTIPOLYGON) {
       poly_column_names.push_back(src_column.columnName);
     }
   }
 
   // remove any matching render group columns from the source list
   // and capture their IDs in the keys of a map (value is ignored)
   std::unordered_map<int, int> column_ids_to_drop;
   auto last_itr = std::remove_if(
       src_oldinfo_strs.begin(),
       src_oldinfo_strs.end(),
       [&](const std::string& v) -> bool {
         // tokenize
         std::vector<std::string> tokens;
         boost::algorithm::split(
             tokens, v, boost::is_any_of(":"), boost::token_compress_on);
         // extract name and ID
         if (tokens.size() < 2) {
           throw std::runtime_error(
               "Dump " + archive_path +
               " has invalid oldinfo file contents. Dump may be corrupt.");
         }
         auto const& column_name = tokens[0];
         auto const column_id = std::stoi(tokens[1]);
         for (auto const& poly_column_name : poly_column_names) {
           // is it a render group column?
           auto const render_group_column_name = poly_column_name + "_render_group";
           if (column_name == render_group_column_name) {
             LOG(INFO) << "RESTORE TABLE dropping render group column '"
                       << render_group_column_name << "' from dump " << archive_path;
             // add to "set"
             column_ids_to_drop[column_id] = -1;
             return true;
           }
         }
         return false;
       });
   src_oldinfo_strs.erase(last_itr, src_oldinfo_strs.end());
 
   return column_ids_to_drop;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::string anonymous_namespace{TableArchiver.cpp}::get_table_schema	(	const std::string &	archive_path,
		const std::string &	table,
		const std::string &	compression
	)

inline

Definition at line 182 of file TableArchiver.cpp.

References simple_file_cat(), and table_schema_filename.

Referenced by TableArchiver::restoreTable().

                                                                   {
   const auto schema_str =
       simple_file_cat(archive_path, table_schema_filename, compression);
   std::regex regex("@T");
   return std::regex_replace(schema_str, regex, table);
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void anonymous_namespace{TableArchiver.cpp}::rename_table_directories	(	const File_Namespace::GlobalFileMgr *	global_file_mgr,
		const std::string &	temp_data_dir,
		const std::vector< std::string > &	target_paths,
		const std::string &	name_prefix
	)

Definition at line 341 of file TableArchiver.cpp.

References abs_path(), add_data_file_symlinks(), delete_old_symlinks(), and File_Namespace::FileMgr::renameAndSymlinkLegacyFiles().

Referenced by TableArchiver::restoreTable().

                                                             {
   boost::filesystem::path base_path(temp_data_dir);
   boost::filesystem::directory_iterator end_it;
   int target_path_index = 0;
   for (boost::filesystem::directory_iterator fit(base_path); fit != end_it; ++fit) {
     if (!boost::filesystem::is_regular_file(fit->status())) {
       const std::string file_path = fit->path().string();
       const std::string file_name = fit->path().filename().string();
       if (boost::istarts_with(file_name, name_prefix)) {
         const std::string target_path =
             abs_path(global_file_mgr) + "/" + target_paths[target_path_index++];
         if (std::rename(file_path.c_str(), target_path.c_str())) {
           throw std::runtime_error("Failed to rename file " + file_path + " to " +
                                    target_path + ": " + std::strerror(errno));
         }
         // Delete any old/invalid symlinks contained in table dump.
         delete_old_symlinks(target_path);
         File_Namespace::FileMgr::renameAndSymlinkLegacyFiles(target_path);
         // For post-rebrand table dumps, symlinks need to be added here, since file mgr
         // migration would already have been executed for the dumped table.
         add_data_file_symlinks(target_path);
       }
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::string anonymous_namespace{TableArchiver.cpp}::run	(	const std::string &	cmd,
		const std::string &	chdir = `""`,
		const bool	log_failure = `true`
	)

inline

Definition at line 86 of file TableArchiver.cpp.

References logger::ERROR, report::error_code(), measure< TimeT >::execution(), LOG, to_lower(), to_string(), and VLOG.

                                                       {
   VLOG(3) << "running cmd: " << cmd;
   int rcode;
   std::error_code ec;
   std::string output, errors;
   const auto time_ms = measure<>::execution([&]() {
     using namespace boost::process;
     ipstream stdout, stderr;
     if (!chdir.empty()) {
       rcode = system(cmd, std_out > stdout, std_err > stderr, ec, start_dir = chdir);
     } else {
       rcode = system(cmd, std_out > stdout, std_err > stderr, ec);
     }
     std::ostringstream ss_output, ss_errors;
     stdout >> ss_output.rdbuf();
     stderr >> ss_errors.rdbuf();
     output = ss_output.str();
     errors = ss_errors.str();
   });
   if (rcode || ec) {
     if (log_failure) {
       LOG(ERROR) << "failed cmd: " << cmd;
       LOG(ERROR) << "exit code: " << rcode;
       LOG(ERROR) << "error code: " << ec.value() << " - " << ec.message();
       LOG(ERROR) << "stdout: " << output;
       LOG(ERROR) << "stderr: " << errors;
     }
 #if defined(__APPLE__)
     // osx bsdtar options "--use-compress-program" and "--fast-read" together
     // run into pipe write error after tar extracts the first occurrence of a
     // file and closes the read end while the decompression program still writes
     // to the pipe. bsdtar doesn't handle this situation well like gnu tar does.
     if (1 == rcode && cmd.find("--fast-read") &&
         (errors.find("cannot write decoded block") != std::string::npos ||
          errors.find("Broken pipe") != std::string::npos)) {
       // ignore this error, or lose speed advantage of "--fast-read" on osx.
       LOG(ERROR) << "tar error ignored on osx for --fast-read";
     } else
 #endif
       // circumvent tar warning on reading file that is "changed as we read it".
       // this warning results from reading a table file under concurrent inserts
       if (1 == rcode && errors.find("changed as we read") != std::string::npos) {
         LOG(ERROR) << "tar error ignored under concurrent inserts";
       } else {
         int error_code;
         std::string error_message;
         if (ec) {
           error_code = ec.value();
           error_message = ec.message();
         } else {
           error_code = rcode;
           // Show a more concise message for permission errors instead of the default
           // verbose message. Error logs will still contain all details.
           if (to_lower(errors).find("permission denied") != std::string::npos) {
             error_message = "Insufficient file read/write permission.";
           } else {
             error_message = errors;
           }
         }
         throw std::runtime_error(
             "An error occurred while executing an internal command. Error code: " +
             std::to_string(error_code) + ", message: " + error_message);
       }
   } else {
     VLOG(3) << "finished cmd: " << cmd;
     VLOG(3) << "time: " << time_ms << " ms";
     VLOG(3) << "stdout: " << output;
   }
   return output;
 }

Here is the call graph for this function:

std::string anonymous_namespace{TableArchiver.cpp}::simple_file_cat	(	const std::string &	archive_path,
		const std::string &	file_name,
		const std::string &	compression,
		const bool	log_failure = `true`
	)

inline

Definition at line 159 of file TableArchiver.cpp.

References get_quoted_string(), ddl_utils::IMPORT, run, and ddl_utils::validate_allowed_file_path().

Referenced by drop_render_group_columns(), get_table_schema(), and TableArchiver::restoreTable().

                                                                   {
   ddl_utils::validate_allowed_file_path(archive_path,
                                         ddl_utils::DataTransferType::IMPORT);
 #if defined(__APPLE__)
   constexpr static auto opt_occurrence = "--fast-read";
 #else
   constexpr static auto opt_occurrence = "--occurrence=1";
 #endif
   boost::filesystem::path temp_dir =
       boost::filesystem::temp_directory_path() / boost::filesystem::unique_path();
   boost::filesystem::create_directories(temp_dir);
   run("tar " + compression + " -xvf " + get_quoted_string(archive_path) + " " +
           opt_occurrence + " " + file_name,
       temp_dir.string(),
       log_failure);
   const auto output = run("cat " + (temp_dir / file_name).string());
   boost::filesystem::remove_all(temp_dir);
   return output;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void anonymous_namespace{TableArchiver.cpp}::update_or_drop_column_ids_in_page_headers	(	const boost::filesystem::path &	path,
		const std::unordered_map< int, int > &	column_ids_map,
		const int32_t	table_epoch,
		const bool	drop_not_update
	)

Definition at line 193 of file TableArchiver.cpp.

References DATA_FILE_EXT, heavyai::file_size(), heavyai::fopen(), File_Namespace::is_page_deleted_with_checkpoint(), simple_file_closer, split(), and to_string().

Referenced by update_or_drop_column_ids_in_table_files().

                                 {
   const std::string file_path = path.string();
   const std::string file_name = path.filename().string();
   std::vector<std::string> tokens;
   boost::split(tokens, file_name, boost::is_any_of("."));
 
   // ref. FileMgr::init for hint of data file name layout
   if (tokens.size() <= 2 || !(DATA_FILE_EXT == "." + tokens[2] || tokens[2] == "mapd")) {
     // We are only interested in files in the form <id>.<page_size>.<DATA_FILE_EXT>
     return;
   }
 
   const auto page_size = boost::lexical_cast<int64_t>(tokens[1]);
   const auto file_size = boost::filesystem::file_size(file_path);
   std::unique_ptr<FILE, decltype(simple_file_closer)> fp(
       std::fopen(file_path.c_str(), "r+"), simple_file_closer);
   if (!fp) {
     throw std::runtime_error("Failed to open " + file_path +
                              " for update: " + std::strerror(errno));
   }
   // TODO(Misiu): Rather than reference an exernal layout we should de-duplicate this
   // page-reading code in a single location.  This will also reduce the need for comments
   // below.
   // ref. FileInfo::openExistingFile for hint of chunk header layout
   for (size_t page = 0; page < file_size / page_size; ++page) {
     int32_t header_info[8];
     if (0 != std::fseek(fp.get(), page * page_size, SEEK_SET)) {
       throw std::runtime_error("Failed to seek to page# " + std::to_string(page) +
                                file_path + " for read: " + std::strerror(errno));
     }
     if (1 != fread(header_info, sizeof header_info, 1, fp.get())) {
       throw std::runtime_error("Failed to read " + file_path + ": " +
                                std::strerror(errno));
     }
     if (const auto header_size = header_info[0]; header_size > 0) {
       // header_info[1] is the page's db_id; but can also be used as an "is deleted"
       // indicator if negative.
       auto& contingent = header_info[1];
       // header_info[2] is the page's table_id; but can also used to store the page's
       // epoch since the FileMgr stores table_id information separately.
       auto& epoch = header_info[2];
       auto& col_id = header_info[3];
       if (File_Namespace::is_page_deleted_with_checkpoint(
               table_epoch, epoch, contingent)) {
         continue;
       }
       auto column_map_it = column_ids_map.find(col_id);
       bool rewrite_header = false;
       if (drop_not_update) {
         // if the header contains a column ID that is a key of the map
         // erase the entire header so that column is effectively dropped
         // the value of the map is ignored, thus allowing us to use the
         // same function for both operations
         if (column_map_it != column_ids_map.end()) {
           // clear the entire header
           std::memset(header_info, 0, sizeof(header_info));
           rewrite_header = true;
         }
       } else {
         if (column_map_it == column_ids_map.end()) {
           throw std::runtime_error("Page " + std::to_string(page) + " in " + file_path +
                                    " has unexpected Column ID " + std::to_string(col_id) +
                                    ". Dump may be corrupt.");
         }
         // If a header contains a column id that is remapped to new location
         // then write that change to the file.
         if (const auto dest_col_id = column_map_it->second; col_id != dest_col_id) {
           col_id = dest_col_id;
           rewrite_header = true;
         }
       }
       if (rewrite_header) {
         if (0 != std::fseek(fp.get(), page * page_size, SEEK_SET)) {
           throw std::runtime_error("Failed to seek to page# " + std::to_string(page) +
                                    file_path + " for write: " + std::strerror(errno));
         }
         if (1 != fwrite(header_info, sizeof header_info, 1, fp.get())) {
           throw std::runtime_error("Failed to write " + file_path + ": " +
                                    std::strerror(errno));
         }
       }
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void anonymous_namespace{TableArchiver.cpp}::update_or_drop_column_ids_in_table_files	(	const int32_t	table_epoch,
		const std::string &	temp_data_dir,
		const std::unordered_map< int, int > &	column_ids_map,
		const bool	drop_not_update
	)

Definition at line 286 of file TableArchiver.cpp.

References ThreadController_NS::SimpleThreadController< FutureReturnType >::checkThreadsStatus(), cpu_threads(), ThreadController_NS::SimpleThreadController< FutureReturnType >::finish(), ThreadController_NS::SimpleThreadController< FutureReturnType >::startThread(), and update_or_drop_column_ids_in_page_headers().

Referenced by drop_render_group_columns(), and TableArchiver::restoreTable().

                                 {
   boost::filesystem::path base_path(temp_data_dir);
   boost::filesystem::recursive_directory_iterator end_it;
   ThreadController_NS::SimpleThreadController<> thread_controller(cpu_threads());
   for (boost::filesystem::recursive_directory_iterator fit(base_path); fit != end_it;
        ++fit) {
     if (!boost::filesystem::is_symlink(fit->path()) &&
         boost::filesystem::is_regular_file(fit->status())) {
       thread_controller.startThread(update_or_drop_column_ids_in_page_headers,
                                     fit->path(),
                                     column_ids_map,
                                     table_epoch,
                                     drop_not_update);
       thread_controller.checkThreadsStatus();
     }
   }
   thread_controller.finish();
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Variable Documentation

auto anonymous_namespace{TableArchiver.cpp}::simple_file_closer = [](FILE* f) { std::fclose(f); }

inline

Definition at line 80 of file TableArchiver.cpp.

Referenced by TableArchiver::dumpTable(), and update_or_drop_column_ids_in_page_headers().

Functions

Variables

Function Documentation

Variable Documentation