foreign_storage::LocalMultiFileReader Class Reference

#include <CsvReader.h>

Inheritance diagram for foreign_storage::LocalMultiFileReader:

Collaboration diagram for foreign_storage::LocalMultiFileReader:

Public Member Functions
	LocalMultiFileReader (const std::string &file_path, const import_export::CopyParams &copy_params)
	LocalMultiFileReader (const std::string &file_path, const import_export::CopyParams &copy_params, const rapidjson::Value &value)
void	checkForMoreRows (size_t file_offset, const ForeignServer *server_options, const UserMapping *user_mapping) override
Public Member Functions inherited from foreign_storage::MultiFileReader
	MultiFileReader (const std::string &file_path, const import_export::CopyParams &copy_params)
	MultiFileReader (const std::string &file_path, const import_export::CopyParams &copy_params, const rapidjson::Value &value)
size_t	getRemainingSize () override
bool	isRemainingSizeKnown () override
size_t	read (void *buffer, size_t max_size) override
size_t	readRegion (void *buffer, size_t offset, size_t size) override
bool	isScanFinished () override
void	serialize (rapidjson::Value &value, rapidjson::Document::AllocatorType &allocator) const override
Public Member Functions inherited from foreign_storage::CsvReader
	CsvReader (const std::string &file_path, const import_export::CopyParams &copy_params)
virtual	~CsvReader ()=default

Private Member Functions
void	insertFile (std::string location)

Additional Inherited Members
Protected Attributes inherited from foreign_storage::MultiFileReader
std::vector< std::unique_ptr < CsvReader > >	files_
std::vector< std::string >	file_locations_
std::vector< size_t >	cumulative_sizes_
size_t	current_index_
size_t	current_offset_
Protected Attributes inherited from foreign_storage::CsvReader
import_export::CopyParams	copy_params_
std::string	file_path_

Detailed Description

Definition at line 325 of file CsvReader.h.

Constructor & Destructor Documentation

foreign_storage::LocalMultiFileReader::LocalMultiFileReader	(	const std::string &	file_path,
		const import_export::CopyParams &	copy_params
	)

Definition at line 529 of file CsvReader.cpp.

References insertFile().

    : MultiFileReader(file_path, copy_params) {
  std::set<std::string> file_locations;
  if (boost::filesystem::is_directory(file_path)) {
    // Find all files in this directory
    for (boost::filesystem::recursive_directory_iterator it(file_path), eit; it != eit;
         ++it) {
      if (!boost::filesystem::is_directory(it->path())) {
        file_locations.insert(it->path().string());
      }
    }
  } else {
    file_locations.insert(file_path);
  }
  for (const auto& location : file_locations) {
    insertFile(location);
  }
}

Here is the call graph for this function:

foreign_storage::LocalMultiFileReader::LocalMultiFileReader	(	const std::string &	file_path,
		const import_export::CopyParams &	copy_params,
		const rapidjson::Value &	value
	)

Definition at line 569 of file CsvReader.cpp.

References foreign_storage::CsvReader::copy_params_, foreign_storage::MultiFileReader::file_locations_, foreign_storage::MultiFileReader::files_, and foreign_storage::anonymous_namespace{CsvReader.cpp}::is_compressed_file().

    : MultiFileReader(file_path, copy_params, value) {
  // Constructs file from files_metadata
  for (size_t index = 0; index < file_locations_.size(); index++) {
    if (is_compressed_file(file_locations_[index])) {
      files_.emplace_back(std::make_unique<CompressedFileReader>(
          file_locations_[index],
          copy_params_,
          value["files_metadata"].GetArray()[index]));
    } else {
      files_.emplace_back(
          std::make_unique<SingleFileReader>(file_locations_[index],
                                             copy_params_,
                                             value["files_metadata"].GetArray()[index]));
    }
  }
}

Here is the call graph for this function:

Member Function Documentation

void foreign_storage::LocalMultiFileReader::checkForMoreRows ( size_t  file_offset, const ForeignServer *  server_options, const UserMapping *  user_mapping  )

overridevirtual

Rescan the target files Throws an exception if the rescan fails (ie files are not in a valid appended state or not supported)

Parameters

file_offset	- where to resume the scan from (end of the last row) as not all of the bytes may have been consumed by the upstream compoennet
server_options	- only needed for S3 backed CSV
user_mapping	- only needed for S3 backed CSV

Reimplemented from foreign_storage::CsvReader.

Definition at line 603 of file CsvReader.cpp.

References CHECK, foreign_storage::MultiFileReader::cumulative_sizes_, foreign_storage::MultiFileReader::current_index_, foreign_storage::MultiFileReader::current_offset_, foreign_storage::MultiFileReader::file_locations_, foreign_storage::CsvReader::file_path_, foreign_storage::MultiFileReader::files_, insertFile(), foreign_storage::MultiFileReader::isScanFinished(), and foreign_storage::throw_removed_file_error().

                                                                             {
  // Look for new files
  std::set<std::string> new_locations;
  CHECK(isScanFinished());
  CHECK(file_offset == current_offset_);
  if (boost::filesystem::is_directory(file_path_)) {
    // Find all files in this directory
    std::set<std::string> all_file_paths;
    for (boost::filesystem::recursive_directory_iterator it(file_path_), eit; it != eit;
         ++it) {
      bool new_file =
          std::find(file_locations_.begin(), file_locations_.end(), it->path()) ==
          file_locations_.end();
      if (!boost::filesystem::is_directory(it->path()) && new_file) {
        new_locations.insert(it->path().string());
      }
      all_file_paths.emplace(it->path().string());
    }

    for (const auto& file_path : file_locations_) {
      if (all_file_paths.find(file_path) == all_file_paths.end()) {
        throw_removed_file_error(file_path);
      }
    }
  }
  if (new_locations.size() > 0) {
    for (const auto& location : new_locations) {
      insertFile(location);
    }
  } else if (files_.size() == 1) {
    // Single file, check if it has new data
    files_[0].get()->checkForMoreRows(file_offset);
    if (!files_[0].get()->isScanFinished()) {
      current_index_ = 0;
      cumulative_sizes_ = {};
    }
  }
}

Here is the call graph for this function:

void foreign_storage::LocalMultiFileReader::insertFile ( std::string  location )

private

Definition at line 589 of file CsvReader.cpp.

References foreign_storage::CsvReader::copy_params_, foreign_storage::MultiFileReader::file_locations_, foreign_storage::MultiFileReader::files_, and foreign_storage::anonymous_namespace{CsvReader.cpp}::is_compressed_file().

Referenced by checkForMoreRows(), and LocalMultiFileReader().

                                                        {
  if (is_compressed_file(location)) {
    files_.emplace_back(std::make_unique<CompressedFileReader>(location, copy_params_));
  } else {
    files_.emplace_back(std::make_unique<SingleFileReader>(location, copy_params_));
  }
  if (files_.back()->isScanFinished()) {
    // skip any initially empty files
    files_.pop_back();
  } else {
    file_locations_.push_back(location);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

---

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

• /home/jenkins-slave/workspace/core-os-doxygen/DataMgr/ForeignStorage/CsvReader.h
• /home/jenkins-slave/workspace/core-os-doxygen/DataMgr/ForeignStorage/CsvReader.cpp