#include <FileReader.h>

Inheritance diagram for foreign_storage::CompressedFileReader:

Collaboration diagram for foreign_storage::CompressedFileReader:

Public Member Functions
	CompressedFileReader (const std::string &file_path, const import_export::CopyParams &copy_params)

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

size_t	read (void *buffer, size_t max_size) override

size_t	readRegion (void *buffer, size_t offset, size_t size) override

bool	isScanFinished () const override

bool	isRemainingSizeKnown () override

size_t	getRemainingSize () override

void	serialize (rapidjson::Value &value, rapidjson::Document::AllocatorType &allocator) const override

Public Member Functions inherited from foreign_storage::SingleFileReader
	SingleFileReader (const std::string &file_path, const import_export::CopyParams &copy_params)

	~SingleFileReader () override=default

FirstLineByFilePath	getFirstLineForEachFile () const override

bool	isEndOfLastFile () override

std::string	getCurrentFilePath () const override

Public Member Functions inherited from foreign_storage::FileReader
	FileReader (const std::string &file_path, const import_export::CopyParams &copy_params)

virtual	~FileReader ()=default

Private Member Functions
void	resetArchive ()

void	checkForMoreRows (size_t file_offset, const shared::FilePathOptions &options, const ForeignServer server_options, const UserMapping user_mapping) override

void	nextEntry ()

void	skipHeader () override

void	skipBytes (size_t n_bytes)

size_t	readInternal (void *buffer, size_t read_size, size_t buffer_size)

std::string	getFirstLine () const override

void	consumeFirstLine (std::optional< std::string > &dest_str)

Private Attributes
ArchiveWrapper	archive_

bool	initial_scan_

bool	scan_finished_

size_t	current_offset_

int	current_index_

std::vector< size_t >	cumulative_sizes_

std::vector< std::string >	sourcenames_

std::vector< int >	archive_entry_index_

Additional Inherited Members
Protected Attributes inherited from foreign_storage::FileReader
import_export::CopyParams	copy_params_

std::string	file_path_

Static Protected Attributes inherited from foreign_storage::SingleFileReader
static constexpr size_t	DEFAULT_HEADER_READ_SIZE {1024}

Detailed Description

Definition at line 269 of file FileReader.h.

Constructor & Destructor Documentation

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

Definition at line 249 of file FileReader.cpp.

References nextEntry().

     : SingleFileReader(file_path, copy_params)
     , archive_(file_path)
     , initial_scan_(true)
     , scan_finished_(false)
     , current_offset_(0)
     , current_index_(-1) {
   // Initialize first entry
   nextEntry();
 }

Here is the call graph for this function:

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

Definition at line 261 of file FileReader.cpp.

References archive_entry_index_, cumulative_sizes_, foreign_storage::json_utils::get_value_from_object(), initial_scan_, scan_finished_, and sourcenames_.

     : CompressedFileReader(file_path, copy_params) {
   scan_finished_ = true;
   initial_scan_ = false;
   sourcenames_.clear();
   archive_entry_index_.clear();
   cumulative_sizes_.clear();
   json_utils::get_value_from_object(value, sourcenames_, "sourcenames");
   json_utils::get_value_from_object(value, cumulative_sizes_, "cumulative_sizes");
   json_utils::get_value_from_object(value, archive_entry_index_, "archive_entry_index");
 }

Here is the call graph for this function:

Member Function Documentation

void foreign_storage::CompressedFileReader::checkForMoreRows	(	size_t	file_offset,
		const shared::FilePathOptions &	options,
		const ForeignServer *	server_options,
		const UserMapping *	user_mapping
	)

overrideprivatevirtual

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 files
user_mapping	- only needed for S3 backed files

Reimplemented from foreign_storage::FileReader.

Definition at line 422 of file FileReader.cpp.

References archive_, archive_entry_index_, CHECK, cumulative_sizes_, current_index_, current_offset_, foreign_storage::ArchiveWrapper::currentEntryDataAvailable(), foreign_storage::ArchiveWrapper::currentEntryFinished(), foreign_storage::ArchiveWrapper::entryName(), foreign_storage::FileReader::file_path_, initial_scan_, foreign_storage::ArchiveWrapper::nextEntry(), nextEntry(), foreign_storage::ArchiveWrapper::resetArchive(), scan_finished_, skipBytes(), skipHeader(), foreign_storage::ArchiveWrapper::skipToEntry(), and sourcenames_.

                                                                              {
   CHECK(initial_scan_ == false);
   size_t initial_entries = archive_entry_index_.size();
 
   // Reset all entry indexes for existing items
   for (size_t index = 0; index < archive_entry_index_.size(); index++) {
     archive_entry_index_[index] = -1;
   }
 
   // Read headers and determine location of existing and new files
   int entry_number = 0;
   archive_.resetArchive();
   while (archive_.nextEntry()) {
     auto it = find(sourcenames_.begin(), sourcenames_.end(), archive_.entryName());
     if (it != sourcenames_.end()) {
       // Record new index of already read file
       auto index = it - sourcenames_.begin();
       archive_entry_index_[index] = entry_number;
     } else {
       // Append new source file
       sourcenames_.emplace_back(archive_.entryName());
       archive_entry_index_.emplace_back(entry_number);
     }
     entry_number++;
   }
 
   // Error if we are missing a file from a previous scan
   for (size_t index = 0; index < archive_entry_index_.size(); index++) {
     if (archive_entry_index_[index] == -1) {
       throw std::runtime_error{
           "Foreign table refreshed with APPEND mode missing archive entry \"" +
           sourcenames_[index] + "\" from file \"" +
           boost::filesystem::path(file_path_).filename().string() + "\"."};
     }
   }
 
   archive_.resetArchive();
   if (initial_entries < archive_entry_index_.size()) {
     // We found more files
     current_index_ = static_cast<int>(initial_entries) - 1;
     current_offset_ = cumulative_sizes_[current_index_];
     // iterate through new entries until we get one with data
     do {
       nextEntry();
     } while (archive_.currentEntryFinished() &&
              current_index_ < static_cast<int>(archive_entry_index_.size()));
 
     if (archive_.currentEntryDataAvailable()) {
       scan_finished_ = false;
     }
   } else {
     // No new files but this may be an archive of a single file
     // Check if we only have one file and check if it has more data
     // May have still have multiple entries with some empty that are ignored
     // like directories
     size_t last_size = 0;
     size_t file_index = -1;
     size_t num_file_entries = 0;
     for (size_t index = 0; index < cumulative_sizes_.size(); index++) {
       if (cumulative_sizes_[index] > last_size) {
         file_index = index;
         num_file_entries++;
         last_size = cumulative_sizes_[index];
       }
     }
     if (num_file_entries == 1) {
       current_index_ = static_cast<int>(file_index);
       current_offset_ = 0;
       size_t last_eof = cumulative_sizes_[file_index];
 
       // reset cumulative_sizes_ with initial zero sizes
       auto old_cumulative_sizes = std::move(cumulative_sizes_);
       cumulative_sizes_ = {};
       for (size_t zero_index = 0; zero_index < file_index; zero_index++) {
         cumulative_sizes_.emplace_back(0);
       }
 
       // Go to Index of file and read to where we left off
       archive_.skipToEntry(archive_entry_index_[file_index]);
       skipHeader();
       skipBytes(last_eof);
       if (!archive_.currentEntryFinished()) {
         scan_finished_ = false;
       } else {
         // There was no new data, so put back the old data structure
         cumulative_sizes_ = std::move(old_cumulative_sizes);
       }
     }
   }
 }

Here is the call graph for this function:

void foreign_storage::CompressedFileReader::consumeFirstLine ( std::optional< std::string > & dest_str )

private

Definition at line 384 of file FileReader.cpp.

References archive_, foreign_storage::ArchiveWrapper::consumeDataFromCurrentEntry(), foreign_storage::FileReader::copy_params_, foreign_storage::ArchiveWrapper::currentEntryFinished(), import_export::CopyParams::line_delim, and foreign_storage::ArchiveWrapper::peekNextChar().

Referenced by skipHeader().

                                                                             {
   char* dest_buffer = nullptr;
   while (!archive_.currentEntryFinished()) {
     if (dest_str.has_value()) {
       auto& str = dest_str.value();
       str.resize(str.length() + 1);
       dest_buffer = str.data() + str.length() - 1;
     }
     if (archive_.peekNextChar() == copy_params_.line_delim) {
       archive_.consumeDataFromCurrentEntry(1, dest_buffer);
       break;
     }
     archive_.consumeDataFromCurrentEntry(1, dest_buffer);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::string foreign_storage::CompressedFileReader::getFirstLine ( ) const

overrideprivatevirtual

Implements foreign_storage::SingleFileReader.

Definition at line 376 of file FileReader.cpp.

References foreign_storage::FileReader::copy_params_, foreign_storage::SingleFileReader::DEFAULT_HEADER_READ_SIZE, and foreign_storage::FileReader::file_path_.

                                                    {
   CompressedFileReader reader{file_path_, copy_params_};
   auto first_line = std::make_optional<std::string>();
   first_line.value().reserve(DEFAULT_HEADER_READ_SIZE);
   reader.consumeFirstLine(first_line);
   return first_line.value();
 }

size_t foreign_storage::CompressedFileReader::getRemainingSize ( )

inlineoverridevirtual

Returns: size of the remaining content to be read

Implements foreign_storage::FileReader.

Definition at line 283 of file FileReader.h.

283 { return 0; }

bool foreign_storage::CompressedFileReader::isRemainingSizeKnown ( )

inlineoverridevirtual

Returns: if remaining size is known

Implements foreign_storage::FileReader.

Definition at line 282 of file FileReader.h.

282 { return false; };

bool foreign_storage::CompressedFileReader::isScanFinished ( ) const

inlineoverridevirtual

Returns: true if the entire file has been read

Implements foreign_storage::FileReader.

Definition at line 280 of file FileReader.h.

References scan_finished_.

Referenced by readRegion().

280 { return scan_finished_; }

foreign_storage::CompressedFileReader::scan_finished_

bool scan_finished_

Definition: FileReader.h:327

Here is the caller graph for this function:

void foreign_storage::CompressedFileReader::nextEntry ( )

private

Go to next archive entry/header with valid data

Definition at line 333 of file FileReader.cpp.

References archive_, archive_entry_index_, CHECK, cumulative_sizes_, current_index_, current_offset_, foreign_storage::ArchiveWrapper::currentEntryFinished(), foreign_storage::ArchiveWrapper::entryName(), foreign_storage::ArchiveWrapper::getCurrentEntryIndex(), initial_scan_, foreign_storage::ArchiveWrapper::nextEntry(), scan_finished_, skipHeader(), foreign_storage::ArchiveWrapper::skipToEntry(), and sourcenames_.

Referenced by checkForMoreRows(), CompressedFileReader(), and readInternal().

                                      {
   do {
     // Go to the next index
     current_index_++;
     if (static_cast<int>(cumulative_sizes_.size()) < current_index_) {
       cumulative_sizes_.push_back(current_offset_);
     }
     if (!initial_scan_) {
       // Entry # in the archive is known and might not be the next one in the file
       if (static_cast<int>(archive_entry_index_.size()) > current_index_) {
         archive_.skipToEntry(archive_entry_index_[current_index_]);
         skipHeader();
       } else {
         scan_finished_ = true;
         return;
       }
     } else {
       // Read next header in archive and save the sourcename
       if (archive_.nextEntry()) {
         // read headers until one has data
         CHECK(sourcenames_.size() == archive_entry_index_.size());
         sourcenames_.emplace_back(archive_.entryName());
         archive_entry_index_.emplace_back(archive_.getCurrentEntryIndex());
         skipHeader();
       } else {
         scan_finished_ = true;
         initial_scan_ = false;
         return;
       }
     }
   } while (archive_.currentEntryFinished());
 }

Here is the call graph for this function:

Here is the caller graph for this function:

size_t foreign_storage::CompressedFileReader::read	(	void *	buffer,
		size_t	max_size
	)

overridevirtual

Read up to max_size bytes from archive into buffer starting starting from the end of the last read

Parameters

buffer	- buffer to load into
max_size	- maximum number of bytes to read into the buffer

Returns: number of bytes actually read

Implements foreign_storage::FileReader.

Definition at line 301 of file FileReader.cpp.

References readInternal().

                                                                {
   // Leave one extra char in case we need to insert a delimiter
   size_t bytes_read = readInternal(buffer, max_size - 1, max_size);
   return bytes_read;
 }

Here is the call graph for this function:

size_t foreign_storage::CompressedFileReader::readInternal	(	void *	buffer,
		size_t	read_size,
		size_t	buffer_size
	)

private

Definition at line 275 of file FileReader.cpp.

References foreign_storage::anonymous_namespace{FileReader.cpp}::adjust_eof(), archive_, foreign_storage::ArchiveWrapper::consumeDataFromCurrentEntry(), foreign_storage::FileReader::copy_params_, current_offset_, foreign_storage::ArchiveWrapper::currentEntryDataAvailable(), foreign_storage::ArchiveWrapper::currentEntryFinished(), run_benchmark_import::dest, import_export::CopyParams::line_delim, and nextEntry().

Referenced by read(), and readRegion().

                                                               {
   size_t remaining_size = read_size;
   char* dest = static_cast<char*>(buffer);
   while (remaining_size > 0 && !archive_.currentEntryFinished()) {
     size_t copy_size = (archive_.currentEntryDataAvailable() < remaining_size)
                            ? archive_.currentEntryDataAvailable()
                            : remaining_size;
     // copy data into dest
     archive_.consumeDataFromCurrentEntry(copy_size, dest);
     remaining_size -= copy_size;
     dest += copy_size;
   }
   size_t bytes_read = read_size - remaining_size;
   if (archive_.currentEntryFinished() && (bytes_read < read_size)) {
     adjust_eof(
         bytes_read, buffer_size, static_cast<char*>(buffer), copy_params_.line_delim);
     current_offset_ += bytes_read;
     nextEntry();
   } else {
     current_offset_ += bytes_read;
   }
   return bytes_read;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

size_t foreign_storage::CompressedFileReader::readRegion	(	void *	buffer,
		size_t	offset,
		size_t	size
	)

overridevirtual

Read up to max_size bytes from archive, starting at given offset isScanFinished() must return true to use readRegion

Parameters

buffer	- buffer to load into
offset	- starting point into the archive to read
size	- maximum number of bytes to read into the buffer

Returns: number of bytes actually read

Implements foreign_storage::FileReader.

Definition at line 307 of file FileReader.cpp.

References archive_, archive_entry_index_, CHECK, cumulative_sizes_, current_index_, current_offset_, foreign_storage::ArchiveWrapper::getCurrentEntryIndex(), isScanFinished(), foreign_storage::anonymous_namespace{FileReader.cpp}::offset_to_index(), readInternal(), skipBytes(), skipHeader(), and foreign_storage::ArchiveWrapper::skipToEntry().

                                                                                 {
   CHECK(isScanFinished());
 
   // Determine where in the archive we are
   size_t index = offset_to_index(cumulative_sizes_, offset);
   CHECK(archive_entry_index_.size() > index);
   auto archive_entry = archive_entry_index_[index];
   current_index_ = static_cast<int>(index);
 
   // If we are in the wrong entry or too far in the right one skip to the correct entry
   if (archive_entry != archive_.getCurrentEntryIndex() ||
       (archive_entry == archive_.getCurrentEntryIndex() && offset < current_offset_)) {
     archive_.skipToEntry(archive_entry);
     skipHeader();
     current_offset_ = 0;
     if (index > 0) {
       current_offset_ = cumulative_sizes_[index - 1];
     }
   }
   skipBytes(offset - current_offset_);
   return readInternal(buffer, size, size);
 }

Here is the call graph for this function:

void foreign_storage::CompressedFileReader::resetArchive ( )

private

Reopen file and reset back to the beginning

void foreign_storage::CompressedFileReader::serialize	(	rapidjson::Value &	value,
		rapidjson::Document::AllocatorType &	allocator
	)		const

overridevirtual

Serialize internal state to given json object This Json will later be used to restore the reader state through a constructor must be called when isScanFinished() is true

Parameters

value	- json object to store needed state to this function can store any needed data or none
allocator	- allocator to use for json contruction

Implements foreign_storage::FileReader.

Definition at line 516 of file FileReader.cpp.

References foreign_storage::json_utils::add_value_to_object(), archive_entry_index_, CHECK, cumulative_sizes_, initial_scan_, scan_finished_, and sourcenames_.

                                                      {
   // Should be done initial scan
   CHECK(scan_finished_);
   CHECK(!initial_scan_);
 
   json_utils::add_value_to_object(value, sourcenames_, "sourcenames", allocator);
   json_utils::add_value_to_object(
       value, cumulative_sizes_, "cumulative_sizes", allocator);
   json_utils::add_value_to_object(
       value, archive_entry_index_, "archive_entry_index", allocator);
 };

Here is the call graph for this function:

void foreign_storage::CompressedFileReader::skipBytes ( size_t n_bytes )

private

Skip forward N bytes in current entry without reading the data

Parameters

n_bytes - number of bytes to skip

Skip forward N bytes without reading the data in current entry

Parameters

n_bytes - number of bytes to skip

Definition at line 404 of file FileReader.cpp.

References archive_, foreign_storage::ArchiveWrapper::consumeDataFromCurrentEntry(), current_offset_, and foreign_storage::ArchiveWrapper::currentEntryDataAvailable().

Referenced by checkForMoreRows(), and readRegion().

                                                    {
   current_offset_ += n_bytes;
   while (n_bytes > 0) {
     if (!archive_.currentEntryDataAvailable()) {
       // We've reached the end of the entry
       return;
     }
     // Keep fetching blocks/entries until we've gone through N bytes
     if (archive_.currentEntryDataAvailable() <= n_bytes) {
       n_bytes -= archive_.currentEntryDataAvailable();
       archive_.consumeDataFromCurrentEntry(archive_.currentEntryDataAvailable());
     } else {
       archive_.consumeDataFromCurrentEntry(n_bytes);
       n_bytes = 0;
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::CompressedFileReader::skipHeader ( )

overrideprivatevirtual

Skip Header of file

Skip file header

Implements foreign_storage::SingleFileReader.

Definition at line 369 of file FileReader.cpp.

References consumeFirstLine(), foreign_storage::FileReader::copy_params_, import_export::CopyParams::has_header, and import_export::kNoHeader.

Referenced by checkForMoreRows(), nextEntry(), and readRegion().

                                       {
   if (copy_params_.has_header != import_export::ImportHeaderRow::kNoHeader) {
     std::optional<std::string> str = std::nullopt;
     consumeFirstLine(str);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

ArchiveWrapper foreign_storage::CompressedFileReader::archive_

private

Definition at line 322 of file FileReader.h.

Referenced by checkForMoreRows(), consumeFirstLine(), nextEntry(), readInternal(), readRegion(), and skipBytes().

std::vector<int> foreign_storage::CompressedFileReader::archive_entry_index_

private

Definition at line 342 of file FileReader.h.

Referenced by checkForMoreRows(), CompressedFileReader(), nextEntry(), readRegion(), and serialize().

std::vector<size_t> foreign_storage::CompressedFileReader::cumulative_sizes_

private

Definition at line 337 of file FileReader.h.

Referenced by checkForMoreRows(), CompressedFileReader(), nextEntry(), readRegion(), and serialize().

int foreign_storage::CompressedFileReader::current_index_

private

Definition at line 334 of file FileReader.h.

Referenced by checkForMoreRows(), nextEntry(), and readRegion().

size_t foreign_storage::CompressedFileReader::current_offset_

private

Definition at line 330 of file FileReader.h.

Referenced by checkForMoreRows(), nextEntry(), readInternal(), readRegion(), and skipBytes().

bool foreign_storage::CompressedFileReader::initial_scan_

private

Definition at line 325 of file FileReader.h.

Referenced by checkForMoreRows(), CompressedFileReader(), nextEntry(), and serialize().

bool foreign_storage::CompressedFileReader::scan_finished_

private

Definition at line 327 of file FileReader.h.

Referenced by checkForMoreRows(), CompressedFileReader(), isScanFinished(), nextEntry(), and serialize().

std::vector<std::string> foreign_storage::CompressedFileReader::sourcenames_

private

Definition at line 339 of file FileReader.h.

Referenced by checkForMoreRows(), CompressedFileReader(), nextEntry(), and serialize().

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

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

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation