OmniSciDB  a987f07e93
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
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 rewrite_column_ids_in_page_headers (const boost::filesystem::path &path, const std::unordered_map< int, int > &column_ids_map, const int32_t table_epoch)
 
void adjust_altered_table_files (const int32_t table_epoch, const std::string &temp_data_dir, const std::unordered_map< int, int > &column_ids_map)
 
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)
 

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

References File_Namespace::GlobalFileMgr::getBasePath().

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

74  {
75  return boost::filesystem::canonical(global_file_mgr->getBasePath()).string();
76 }
std::string getBasePath() const

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

References DATA_FILE_EXT, and File_Namespace::kLegacyDataFileExtension.

Referenced by rename_table_directories().

281  {
282  std::map<boost::filesystem::path, boost::filesystem::path> old_to_new_paths;
283  for (boost::filesystem::directory_iterator it(table_data_dir), end_it; it != end_it;
284  it++) {
285  const auto path = boost::filesystem::canonical(it->path());
286  if (path.extension().string() == DATA_FILE_EXT) {
287  auto old_path = path;
288  old_path.replace_extension(File_Namespace::kLegacyDataFileExtension);
289  // Add a symlink to data file, if one does not exist.
290  if (!boost::filesystem::exists(old_path)) {
291  old_to_new_paths[old_path] = path;
292  }
293  }
294  }
295  for (const auto& [old_path, new_path] : old_to_new_paths) {
296  boost::filesystem::create_symlink(new_path.filename(), old_path);
297  }
298 }
#define DATA_FILE_EXT
Definition: File.h:25
constexpr auto kLegacyDataFileExtension
Definition: File.h:36

+ Here is the caller graph for this function:

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

Definition at line 250 of file TableArchiver.cpp.

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

Referenced by TableArchiver::restoreTable().

252  {
253  boost::filesystem::path base_path(temp_data_dir);
254  boost::filesystem::recursive_directory_iterator end_it;
256  for (boost::filesystem::recursive_directory_iterator fit(base_path); fit != end_it;
257  ++fit) {
258  if (!boost::filesystem::is_symlink(fit->path()) &&
259  boost::filesystem::is_regular_file(fit->status())) {
260  thread_controller.startThread(
261  rewrite_column_ids_in_page_headers, fit->path(), column_ids_map, table_epoch);
262  thread_controller.checkThreadsStatus();
263  }
264  }
265  thread_controller.finish();
266 }
void rewrite_column_ids_in_page_headers(const boost::filesystem::path &path, const std::unordered_map< int, int > &column_ids_map, const int32_t table_epoch)
int cpu_threads()
Definition: thread_count.h:25

+ Here is the call graph for this function:

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

Referenced by rename_table_directories().

268  {
269  std::vector<boost::filesystem::path> symlinks;
270  for (boost::filesystem::directory_iterator it(table_data_dir), end_it; it != end_it;
271  it++) {
272  if (boost::filesystem::is_symlink(it->path())) {
273  symlinks.emplace_back(it->path());
274  }
275  }
276  for (const auto& symlink : symlinks) {
277  boost::filesystem::remove_all(symlink);
278  }
279 }

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

References simple_file_cat(), and table_schema_filename.

Referenced by TableArchiver::restoreTable().

170  {
171  const auto schema_str =
172  simple_file_cat(archive_path, table_schema_filename, compression);
173  std::regex regex("@T");
174  return std::regex_replace(schema_str, regex, table);
175 }
static constexpr char const * table_schema_filename
std::string simple_file_cat(const std::string &archive_path, const std::string &file_name, const std::string &compression)

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

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

Referenced by TableArchiver::restoreTable().

303  {
304  boost::filesystem::path base_path(temp_data_dir);
305  boost::filesystem::directory_iterator end_it;
306  int target_path_index = 0;
307  for (boost::filesystem::directory_iterator fit(base_path); fit != end_it; ++fit) {
308  if (!boost::filesystem::is_regular_file(fit->status())) {
309  const std::string file_path = fit->path().string();
310  const std::string file_name = fit->path().filename().string();
311  if (boost::istarts_with(file_name, name_prefix)) {
312  const std::string target_path =
313  abs_path(global_file_mgr) + "/" + target_paths[target_path_index++];
314  if (std::rename(file_path.c_str(), target_path.c_str())) {
315  throw std::runtime_error("Failed to rename file " + file_path + " to " +
316  target_path + ": " + std::strerror(errno));
317  }
318  // Delete any old/invalid symlinks contained in table dump.
319  delete_old_symlinks(target_path);
321  // For post-rebrand table dumps, symlinks need to be added here, since file mgr
322  // migration would already have been executed for the dumped table.
323  add_data_file_symlinks(target_path);
324  }
325  }
326  }
327 }
void delete_old_symlinks(const std::string &table_data_dir)
std::string abs_path(const File_Namespace::GlobalFileMgr *global_file_mgr)
static void renameAndSymlinkLegacyFiles(const std::string &table_data_dir)
Definition: FileMgr.cpp:1125
void add_data_file_symlinks(const std::string &table_data_dir)

+ Here is the call graph for this function:

+ Here is the caller graph for this function:

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

Definition at line 179 of file TableArchiver.cpp.

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

Referenced by adjust_altered_table_files().

182  {
183  const std::string file_path = path.string();
184  const std::string file_name = path.filename().string();
185  std::vector<std::string> tokens;
186  boost::split(tokens, file_name, boost::is_any_of("."));
187 
188  // ref. FileMgr::init for hint of data file name layout
189  if (tokens.size() <= 2 || !(DATA_FILE_EXT == "." + tokens[2] || tokens[2] == "mapd")) {
190  // We are only interested in files in the form <id>.<page_size>.<DATA_FILE_EXT>
191  return;
192  }
193 
194  const auto page_size = boost::lexical_cast<int64_t>(tokens[1]);
195  const auto file_size = boost::filesystem::file_size(file_path);
196  std::unique_ptr<FILE, decltype(simple_file_closer)> fp(
197  std::fopen(file_path.c_str(), "r+"), simple_file_closer);
198  if (!fp) {
199  throw std::runtime_error("Failed to open " + file_path +
200  " for update: " + std::strerror(errno));
201  }
202  // TODO(Misiu): Rather than reference an exernal layout we should de-duplicate this
203  // page-reading code in a single location. This will also reduce the need for comments
204  // below.
205  // ref. FileInfo::openExistingFile for hint of chunk header layout
206  for (size_t page = 0; page < file_size / page_size; ++page) {
207  int32_t header_info[8];
208  if (0 != std::fseek(fp.get(), page * page_size, SEEK_SET)) {
209  throw std::runtime_error("Failed to seek to page# " + std::to_string(page) +
210  file_path + " for read: " + std::strerror(errno));
211  }
212  if (1 != fread(header_info, sizeof header_info, 1, fp.get())) {
213  throw std::runtime_error("Failed to read " + file_path + ": " +
214  std::strerror(errno));
215  }
216  if (const auto header_size = header_info[0]; header_size > 0) {
217  // header_info[1] is the page's db_id; but can also be used as an "is deleted"
218  // indicator if negative.
219  auto& contingent = header_info[1];
220  // header_info[2] is the page's table_id; but can also used to store the page's
221  // epoch since the FileMgr stores table_id information separately.
222  auto& epoch = header_info[2];
223  auto& col_id = header_info[3];
225  table_epoch, epoch, contingent)) {
226  continue;
227  }
228  auto column_map_it = column_ids_map.find(col_id);
229  CHECK(column_map_it != column_ids_map.end()) << "could not find " << col_id;
230  // If a header contains a column id that is remapped to new location
231  // then write that change to the file.
232  if (const auto dest_col_id = column_map_it->second; col_id != dest_col_id) {
233  col_id = dest_col_id;
234  if (0 != std::fseek(fp.get(), page * page_size, SEEK_SET)) {
235  throw std::runtime_error("Failed to seek to page# " + std::to_string(page) +
236  file_path + " for write: " + std::strerror(errno));
237  }
238  if (1 != fwrite(header_info, sizeof header_info, 1, fp.get())) {
239  throw std::runtime_error("Failed to write " + file_path + ": " +
240  std::strerror(errno));
241  }
242  }
243  }
244  }
245 }
#define DATA_FILE_EXT
Definition: File.h:25
std::string to_string(char const *&&v)
std::vector< std::string > split(std::string_view str, std::string_view delim, std::optional< size_t > maxsplit)
split apart a string into a vector of substrings
::FILE * fopen(const char *filename, const char *mode)
Definition: heavyai_fs.cpp:74
bool is_page_deleted_with_checkpoint(int32_t table_epoch, int32_t page_epoch, int32_t contingent)
Definition: FileInfo.cpp:271
#define CHECK(condition)
Definition: Logger.h:289
size_t file_size(const int fd)
Definition: heavyai_fs.cpp:33

+ 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 = "" 
)
inline

Definition at line 78 of file TableArchiver.cpp.

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

78  {
79  VLOG(3) << "running cmd: " << cmd;
80  int rcode;
81  std::error_code ec;
82  std::string output, errors;
83  const auto time_ms = measure<>::execution([&]() {
84  using namespace boost::process;
85  ipstream stdout, stderr;
86  if (!chdir.empty()) {
87  rcode = system(cmd, std_out > stdout, std_err > stderr, ec, start_dir = chdir);
88  } else {
89  rcode = system(cmd, std_out > stdout, std_err > stderr, ec);
90  }
91  std::ostringstream ss_output, ss_errors;
92  stdout >> ss_output.rdbuf();
93  stderr >> ss_errors.rdbuf();
94  output = ss_output.str();
95  errors = ss_errors.str();
96  });
97  if (rcode || ec) {
98  LOG(ERROR) << "failed cmd: " << cmd;
99  LOG(ERROR) << "exit code: " << rcode;
100  LOG(ERROR) << "error code: " << ec.value() << " - " << ec.message();
101  LOG(ERROR) << "stdout: " << output;
102  LOG(ERROR) << "stderr: " << errors;
103 #if defined(__APPLE__)
104  // osx bsdtar options "--use-compress-program" and "--fast-read" together
105  // run into pipe write error after tar extracts the first occurrence of a
106  // file and closes the read end while the decompression program still writes
107  // to the pipe. bsdtar doesn't handle this situation well like gnu tar does.
108  if (1 == rcode && cmd.find("--fast-read") &&
109  (errors.find("cannot write decoded block") != std::string::npos ||
110  errors.find("Broken pipe") != std::string::npos)) {
111  // ignore this error, or lose speed advantage of "--fast-read" on osx.
112  LOG(ERROR) << "tar error ignored on osx for --fast-read";
113  } else
114 #endif
115  // circumvent tar warning on reading file that is "changed as we read it".
116  // this warning results from reading a table file under concurrent inserts
117  if (1 == rcode && errors.find("changed as we read") != std::string::npos) {
118  LOG(ERROR) << "tar error ignored under concurrent inserts";
119  } else {
120  int error_code;
121  std::string error_message;
122  if (ec) {
123  error_code = ec.value();
124  error_message = ec.message();
125  } else {
126  error_code = rcode;
127  // Show a more concise message for permission errors instead of the default
128  // verbose message. Error logs will still contain all details.
129  if (to_lower(errors).find("permission denied") != std::string::npos) {
130  error_message = "Insufficient file read/write permission.";
131  } else {
132  error_message = errors;
133  }
134  }
135  throw std::runtime_error(
136  "An error occurred while executing an internal command. Error code: " +
137  std::to_string(error_code) + ", message: " + error_message);
138  }
139  } else {
140  VLOG(3) << "finished cmd: " << cmd;
141  VLOG(3) << "time: " << time_ms << " ms";
142  VLOG(3) << "stdout: " << output;
143  }
144  return output;
145 }
std::string to_lower(const std::string &str)
static TimeT::rep execution(F func, Args &&...args)
Definition: sample.cpp:29
#define LOG(tag)
Definition: Logger.h:283
std::string to_string(char const *&&v)
#define VLOG(n)
Definition: Logger.h:383

+ 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 
)
inline

Definition at line 147 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().

149  {
152 #if defined(__APPLE__)
153  constexpr static auto opt_occurrence = "--fast-read";
154 #else
155  constexpr static auto opt_occurrence = "--occurrence=1";
156 #endif
157  boost::filesystem::path temp_dir =
158  boost::filesystem::temp_directory_path() / boost::filesystem::unique_path();
159  boost::filesystem::create_directories(temp_dir);
160  run("tar " + compression + " -xvf " + get_quoted_string(archive_path) + " " +
161  opt_occurrence + " " + file_name,
162  temp_dir.string());
163  const auto output = run("cat " + (temp_dir / file_name).string());
164  boost::filesystem::remove_all(temp_dir);
165  return output;
166 }
std::string get_quoted_string(const std::string &filename, char quote, char escape)
Quote a string while escaping any existing quotes in the string.
void validate_allowed_file_path(const std::string &file_path, const DataTransferType data_transfer_type, const bool allow_wildcards)
Definition: DdlUtils.cpp:785
static bool run

+ 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