anonymous_namespace{TableArchiver.cpp} Namespace Reference

Functions
std::string	abs_path (const File_Namespace::GlobalFileMgr *global_file_mgr)
std::string	run (const std::string &cmd, const std::string &chdir="")
std::string	simple_file_cat (const std::string &archive_path, const std::string &file_name, const std::string &compression)
std::string	get_table_schema (const std::string &archive_path, const std::string &table, const std::string &compression)
void	adjust_altered_table_files (const std::string &temp_data_dir, const std::unordered_map< int, int > &column_ids_map)
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)

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

Function Documentation

abs_path()

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

inline

Definition at line 66 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:

adjust_altered_table_files()

void anonymous_namespace{TableArchiver.cpp}::adjust_altered_table_files	(	const std::string &	temp_data_dir,
		const std::unordered_map< int, int > &	column_ids_map
	)

Definition at line 157 of file TableArchiver.cpp.

References CHECK, ThreadController_NS::SimpleThreadController< FutureReturnType >::checkThreadsStatus(), cpu_threads(), omnisci::file_size(), ThreadController_NS::SimpleThreadController< FutureReturnType >::finish(), MAPD_FILE_EXT, simple_file_closer, split(), ThreadController_NS::SimpleThreadController< FutureReturnType >::startThread(), and to_string().

Referenced by 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_regular_file(fit->status())) {
      const std::string file_path = fit->path().string();
      const std::string file_name = fit->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 && MAPD_FILE_EXT == "." + tokens[2]) {
        thread_controller.startThread([file_name, file_path, tokens, &column_ids_map] {
          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));
          }
          // ref. FileInfo::openExistingFile for hint of chunk header layout
          for (size_t page = 0; page < file_size / page_size; ++page) {
            int ints[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(ints, sizeof ints, 1, fp.get())) {
              throw std::runtime_error("Failed to read " + file_path + ": " +
                                       std::strerror(errno));
            }
            if (ints[0] > 0) {  // header size
              auto cit = column_ids_map.find(ints[3]);
              CHECK(cit != column_ids_map.end());
              if (ints[3] != cit->second) {
                ints[3] = cit->second;
                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(ints, sizeof ints, 1, fp.get())) {
                  throw std::runtime_error("Failed to write " + file_path + ": " +
                                           std::strerror(errno));
                }
              }
            }
          }
        });
        thread_controller.checkThreadsStatus();
      }
    }
  }
  thread_controller.finish();
}

Here is the call graph for this function:

Here is the caller graph for this function:

get_table_schema()

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 145 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:

rename_table_directories()

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 216 of file TableArchiver.cpp.

References abs_path().

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));
        }
      }
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

run()

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

inline

Definition at line 70 of file TableArchiver.cpp.

References logger::ERROR, error_code, measure< TimeT >::execution(), LOG, 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) {
    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 {
      throw std::runtime_error("Failed to run command: " + cmd +
                               "\nexit code: " + std::to_string(rcode) + "\nerrors:\n" +
                               (rcode ? errors : ec.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:

simple_file_cat()

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

inline

Definition at line 124 of file TableArchiver.cpp.

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

Referenced by 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());
  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:

Variable Documentation

simple_file_closer

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

inline

Definition at line 64 of file TableArchiver.cpp.

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