#include <AbstractTextFileDataWrapper.h>

Inheritance diagram for foreign_storage::AbstractTextFileDataWrapper:

Collaboration diagram for foreign_storage::AbstractTextFileDataWrapper:

Public Member Functions
	AbstractTextFileDataWrapper ()

	AbstractTextFileDataWrapper (const int db_id, const ForeignTable *foreign_table)

	AbstractTextFileDataWrapper (const int db_id, const ForeignTable foreign_table, const UserMapping user_mapping, const bool disable_cache)

void	populateChunkMetadata (ChunkMetadataVector &chunk_metadata_vector) override

void	populateChunkBuffers (const ChunkToBufferMap &required_buffers, const ChunkToBufferMap &optional_buffers, AbstractBuffer *delete_buffer) override

std::string	getSerializedDataWrapper () const override

void	restoreDataWrapperInternals (const std::string &file_path, const ChunkMetadataVector &chunk_metadata) override

bool	isRestored () const override

ParallelismLevel	getCachedParallelismLevel () const override

ParallelismLevel	getNonCachedParallelismLevel () const override

void	createRenderGroupAnalyzers () override
	Create RenderGroupAnalyzers for poly columns. More...

Public Member Functions inherited from foreign_storage::AbstractFileStorageDataWrapper
	AbstractFileStorageDataWrapper ()

void	validateServerOptions (const ForeignServer *foreign_server) const override

void	validateTableOptions (const ForeignTable *foreign_table) const override

const std::set < std::string_view > &	getSupportedTableOptions () const override

void	validateUserMappingOptions (const UserMapping user_mapping, const ForeignServer foreign_server) const override

const std::set < std::string_view > &	getSupportedUserMappingOptions () const override

Public Member Functions inherited from foreign_storage::ForeignDataWrapper
	ForeignDataWrapper ()=default

virtual	~ForeignDataWrapper ()=default

virtual const std::set < std::string >	getAlterableTableOptions () const

virtual void	validateSchema (const std::list< ColumnDescriptor > &columns) const

Protected Member Functions
virtual const TextFileBufferParser &	getFileBufferParser () const =0

Private Member Functions
	AbstractTextFileDataWrapper (const ForeignTable *foreign_table)

void	populateChunks (std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map, int fragment_id, AbstractBuffer *delete_buffer)

void	populateChunkMapForColumns (const std::set< const ColumnDescriptor * > &columns, const int fragment_id, const ChunkToBufferMap &buffers, std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map)

void	updateMetadata (std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map, int fragment_id)

Private Attributes
std::map< ChunkKey, std::shared_ptr< ChunkMetadata > >	chunk_metadata_map_

std::map< int, FileRegions >	fragment_id_to_file_regions_map_

std::unique_ptr< FileReader >	file_reader_

const int	db_id_

const ForeignTable *	foreign_table_

std::map< ChunkKey, std::unique_ptr < ForeignStorageBuffer > >	chunk_encoder_buffers_

size_t	num_rows_

size_t	append_start_offset_

bool	is_restored_

const UserMapping *	user_mapping_

const bool	disable_cache_

RenderGroupAnalyzerMap	render_group_analyzer_map_

Additional Inherited Members
Public Types inherited from foreign_storage::ForeignDataWrapper
enum	ParallelismLevel { NONE, INTRA_FRAGMENT, INTER_FRAGMENT }

Static Public Attributes inherited from foreign_storage::AbstractFileStorageDataWrapper
static const std::string	STORAGE_TYPE_KEY = "STORAGE_TYPE"

static const std::string	BASE_PATH_KEY = "BASE_PATH"

static const std::string	FILE_PATH_KEY = "FILE_PATH"

static const std::string	REGEX_PATH_FILTER_KEY = "REGEX_PATH_FILTER"

static const std::string	LOCAL_FILE_STORAGE_TYPE = "LOCAL_FILE"

static const std::string	S3_STORAGE_TYPE = "AWS_S3"

static const std::string	FILE_SORT_ORDER_BY_KEY = shared::FILE_SORT_ORDER_BY_KEY

static const std::string	FILE_SORT_REGEX_KEY = shared::FILE_SORT_REGEX_KEY

static const std::array < std::string, 1 >	supported_storage_types

Static Protected Member Functions inherited from foreign_storage::AbstractFileStorageDataWrapper
static std::string	getFullFilePath (const ForeignTable *foreign_table)
	Returns the path to the source file/dir of the table. Depending on options this may result from a concatenation of server and table path options. More...

Detailed Description

Definition at line 32 of file AbstractTextFileDataWrapper.h.

Constructor & Destructor Documentation

foreign_storage::AbstractTextFileDataWrapper::AbstractTextFileDataWrapper ( )

Definition at line 36 of file AbstractTextFileDataWrapper.cpp.

     : db_id_(-1)
     , foreign_table_(nullptr)
     , user_mapping_(nullptr)
     , disable_cache_(false) {}

foreign_storage::AbstractTextFileDataWrapper::AbstractTextFileDataWrapper	(	const int	db_id,
		const ForeignTable *	foreign_table
	)

Definition at line 42 of file AbstractTextFileDataWrapper.cpp.

     : db_id_(db_id)
     , foreign_table_(foreign_table)
     , is_restored_(false)
     , user_mapping_(nullptr)
     , disable_cache_(false) {}

foreign_storage::AbstractTextFileDataWrapper::AbstractTextFileDataWrapper	(	const int	db_id,
		const ForeignTable *	foreign_table,
		const UserMapping *	user_mapping,
		const bool	disable_cache
	)

Definition at line 51 of file AbstractTextFileDataWrapper.cpp.

     : db_id_(db_id)
     , foreign_table_(foreign_table)
     , is_restored_(false)
     , user_mapping_(user_mapping)
     , disable_cache_(disable_cache) {}

foreign_storage::AbstractTextFileDataWrapper::AbstractTextFileDataWrapper ( const ForeignTable * foreign_table )

private

Member Function Documentation

void foreign_storage::AbstractTextFileDataWrapper::createRenderGroupAnalyzers ( )

overridevirtual

Create RenderGroupAnalyzers for poly columns.

Reimplemented from foreign_storage::ForeignDataWrapper.

Definition at line 1071 of file AbstractTextFileDataWrapper.cpp.

References CHECK, CHECK_GE, db_id_, foreign_table_, Catalog_Namespace::SysCatalog::getCatalog(), Catalog_Namespace::SysCatalog::instance(), IS_GEO_POLY, render_group_analyzer_map_, and TableDescriptor::tableId.

                                                              {
   // must have these
   CHECK_GE(db_id_, 0);
   CHECK(foreign_table_);
 
   // populate map for all poly columns in this table
   auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
   CHECK(catalog);
   auto columns =
       catalog->getAllColumnMetadataForTable(foreign_table_->tableId, false, false, true);
   for (auto const& column : columns) {
     if (IS_GEO_POLY(column->columnType.get_type())) {
       CHECK(render_group_analyzer_map_
                 .try_emplace(column->columnId,
                              std::make_unique<import_export::RenderGroupAnalyzer>())
                 .second);
     }
   }
 }

Here is the call graph for this function:

ParallelismLevel foreign_storage::AbstractTextFileDataWrapper::getCachedParallelismLevel ( ) const

inlineoverridevirtual

Gets the desired level of parallelism for the data wrapper when a cache is in use. This affects the optional buffers that the data wrapper is made aware of during data requests.

Reimplemented from foreign_storage::ForeignDataWrapper.

Definition at line 55 of file AbstractTextFileDataWrapper.h.

References foreign_storage::ForeignDataWrapper::INTRA_FRAGMENT.

55 { return INTRA_FRAGMENT; }

foreign_storage::ForeignDataWrapper::INTRA_FRAGMENT

Definition: ForeignDataWrapper.h:142

virtual const TextFileBufferParser& foreign_storage::AbstractTextFileDataWrapper::getFileBufferParser ( ) const

protectedpure virtual

Implemented in foreign_storage::CsvDataWrapper, and foreign_storage::RegexParserDataWrapper.

Referenced by populateChunkMetadata(), populateChunks(), and restoreDataWrapperInternals().

Here is the caller graph for this function:

ParallelismLevel foreign_storage::AbstractTextFileDataWrapper::getNonCachedParallelismLevel ( ) const

inlineoverridevirtual

Gets the desired level of parallelism for the data wrapper when no cache is in use. This affects the optional buffers that the data wrapper is made aware of during data requests.

Reimplemented from foreign_storage::ForeignDataWrapper.

Definition at line 57 of file AbstractTextFileDataWrapper.h.

References foreign_storage::ForeignDataWrapper::INTRA_FRAGMENT.

                                                                  {
     return INTRA_FRAGMENT;
   }

std::string foreign_storage::AbstractTextFileDataWrapper::getSerializedDataWrapper ( ) const

overridevirtual

Serialize internal state of wrapper into file at given path if implemented

Implements foreign_storage::ForeignDataWrapper.

Definition at line 995 of file AbstractTextFileDataWrapper.cpp.

References foreign_storage::json_utils::add_value_to_object(), append_start_offset_, file_reader_, fragment_id_to_file_regions_map_, num_rows_, and foreign_storage::json_utils::write_to_string().

                                                                       {
   rapidjson::Document d;
   d.SetObject();
 
   // Save fragment map
   json_utils::add_value_to_object(d,
                                   fragment_id_to_file_regions_map_,
                                   "fragment_id_to_file_regions_map",
                                   d.GetAllocator());
 
   // Save reader metadata
   rapidjson::Value reader_metadata(rapidjson::kObjectType);
   file_reader_->serialize(reader_metadata, d.GetAllocator());
   d.AddMember("reader_metadata", reader_metadata, d.GetAllocator());
 
   json_utils::add_value_to_object(d, num_rows_, "num_rows", d.GetAllocator());
   json_utils::add_value_to_object(
       d, append_start_offset_, "append_start_offset", d.GetAllocator());
 
   return json_utils::write_to_string(d);
 }

Here is the call graph for this function:

bool foreign_storage::AbstractTextFileDataWrapper::isRestored ( ) const

overridevirtual

Implements foreign_storage::ForeignDataWrapper.

Definition at line 1064 of file AbstractTextFileDataWrapper.cpp.

References is_restored_.

                                                    {
   return is_restored_;
 }

void foreign_storage::AbstractTextFileDataWrapper::populateChunkBuffers	(	const ChunkToBufferMap &	required_buffers,
		const ChunkToBufferMap &	optional_buffers,
		AbstractBuffer *	delete_buffer
	)

overridevirtual

Populates given chunk buffers identified by chunk keys. All provided chunk buffers are expected to be for the same fragment.

Parameters

required_buffers	- chunk buffers that must always be populated
optional_buffers	- chunk buffers that can be optionally populated, if the data wrapper has to scan through chunk data anyways (typically for row wise data formats)
delete_buffer	- chunk buffer for fragment's delete column, if non-null data wrapper is expected to mark deleted rows in buffer and continue processing

Implements foreign_storage::ForeignDataWrapper.

Definition at line 98 of file AbstractTextFileDataWrapper.cpp.

References CHECK, CHUNK_KEY_FRAGMENT_IDX, db_id_, DEBUG_TIMER, foreign_table_, foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::get_columns(), Catalog_Namespace::SysCatalog::getCatalog(), Catalog_Namespace::SysCatalog::instance(), populateChunkMapForColumns(), populateChunks(), TableDescriptor::tableId, and updateMetadata().

                                    {
   auto timer = DEBUG_TIMER(__func__);
   auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
   CHECK(catalog);
   CHECK(!required_buffers.empty());
 
   auto fragment_id = required_buffers.begin()->first[CHUNK_KEY_FRAGMENT_IDX];
   auto required_columns =
       get_columns(required_buffers, *catalog, foreign_table_->tableId, fragment_id);
   std::map<int, Chunk_NS::Chunk> column_id_to_chunk_map;
   populateChunkMapForColumns(
       required_columns, fragment_id, required_buffers, column_id_to_chunk_map);
 
   if (!optional_buffers.empty()) {
     auto optional_columns =
         get_columns(optional_buffers, *catalog, foreign_table_->tableId, fragment_id);
     populateChunkMapForColumns(
         optional_columns, fragment_id, optional_buffers, column_id_to_chunk_map);
   }
   populateChunks(column_id_to_chunk_map, fragment_id, delete_buffer);
   updateMetadata(column_id_to_chunk_map, fragment_id);
 }

Here is the call graph for this function:

void foreign_storage::AbstractTextFileDataWrapper::populateChunkMapForColumns	(	const std::set< const ColumnDescriptor * > &	columns,
		const int	fragment_id,
		const ChunkToBufferMap &	buffers,
		std::map< int, Chunk_NS::Chunk > &	column_id_to_chunk_map
	)

private

Definition at line 83 of file AbstractTextFileDataWrapper.cpp.

References chunk_metadata_map_, db_id_, foreign_table_, foreign_storage::init_chunk_for_column(), and TableDescriptor::tableId.

Referenced by populateChunkBuffers().

                                                         {
   for (const auto column : columns) {
     ChunkKey data_chunk_key = {
         db_id_, foreign_table_->tableId, column->columnId, fragment_id};
     init_chunk_for_column(data_chunk_key,
                           chunk_metadata_map_,
                           buffers,
                           column_id_to_chunk_map[column->columnId]);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::AbstractTextFileDataWrapper::populateChunkMetadata ( ChunkMetadataVector & chunk_metadata_vector )

overridevirtual

Populates provided chunk metadata vector with metadata for table specified in given chunk key. Metadata scan for text file(s) configured for foreign table occurs in parallel whenever appropriate. Parallel processing involves the main thread creating ParseBufferRequest objects, which contain buffers with text content read from file and adding these request objects to a queue that is consumed by a fixed number of threads. After request processing, request objects are put back into a pool for reuse for subsequent requests in order to avoid unnecessary allocation of new buffers.

Parameters

chunk_metadata_vector - vector to be populated with chunk metadata

Implements foreign_storage::ForeignDataWrapper.

Definition at line 840 of file AbstractTextFileDataWrapper.cpp.

References foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::add_placeholder_metadata(), append_start_offset_, threading_serial::async(), foreign_storage::MetadataScanMultiThreadingParams::cached_chunks, CHECK, CHECK_EQ, foreign_storage::MetadataScanMultiThreadingParams::chunk_encoder_buffers, chunk_encoder_buffers_, CHUNK_KEY_COLUMN_IDX, chunk_metadata_map_, foreign_storage::MetadataScanMultiThreadingParams::continue_processing, db_id_, DEBUG_TIMER, foreign_storage::MetadataScanMultiThreadingParams::disable_cache, disable_cache_, foreign_storage::dispatch_metadata_scan_requests(), file_reader_, foreign_storage::AbstractFileStorageDataWrapper::FILE_SORT_ORDER_BY_KEY, foreign_storage::AbstractFileStorageDataWrapper::FILE_SORT_REGEX_KEY, foreign_storage::ForeignTable::foreign_server, foreign_table_, fragment_id_to_file_regions_map_, foreign_storage::get_buffer_size(), foreign_storage::get_thread_count(), Catalog_Namespace::SysCatalog::getCatalog(), getFileBufferParser(), foreign_storage::AbstractFileStorageDataWrapper::getFullFilePath(), foreign_storage::OptionsContainer::getOption(), Catalog_Namespace::SysCatalog::instance(), foreign_storage::ForeignTable::isAppendMode(), foreign_storage::AbstractFileStorageDataWrapper::LOCAL_FILE_STORAGE_TYPE, num_rows_, foreign_storage::OptionsContainer::options, run_benchmark_import::parser, foreign_storage::MetadataScanMultiThreadingParams::pending_requests_condition, foreign_storage::MetadataScanMultiThreadingParams::pending_requests_mutex, foreign_storage::AbstractFileStorageDataWrapper::REGEX_PATH_FILTER_KEY, foreign_storage::MetadataScanMultiThreadingParams::request_pool, foreign_storage::scan_metadata(), foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::skip_metadata_scan(), foreign_storage::AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY, TableDescriptor::tableId, UNREACHABLE, and foreign_storage::TextFileBufferParser::validateAndGetCopyParams().

                                                 {
   auto timer = DEBUG_TIMER(__func__);
 
   const auto copy_params = getFileBufferParser().validateAndGetCopyParams(foreign_table_);
   const auto file_path = getFullFilePath(foreign_table_);
   auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
   CHECK(catalog);
   auto& parser = getFileBufferParser();
   auto& server_options = foreign_table_->foreign_server->options;
   if (foreign_table_->isAppendMode() && file_reader_ != nullptr) {
     parser.validateFiles(file_reader_.get(), foreign_table_);
     if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
       file_reader_->checkForMoreRows(append_start_offset_);
     } else {
       UNREACHABLE();
     }
   } else {
     // Should only be called once for non-append tables
     CHECK(chunk_metadata_map_.empty());
     CHECK(fragment_id_to_file_regions_map_.empty());
     if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
       file_reader_ = std::make_unique<LocalMultiFileReader>(
           file_path,
           copy_params,
           foreign_table_->getOption(
               AbstractFileStorageDataWrapper::REGEX_PATH_FILTER_KEY),
           foreign_table_->getOption(
               AbstractFileStorageDataWrapper::FILE_SORT_ORDER_BY_KEY),
           foreign_table_->getOption(AbstractFileStorageDataWrapper::FILE_SORT_REGEX_KEY));
     } else {
       UNREACHABLE();
     }
     parser.validateFiles(file_reader_.get(), foreign_table_);
     num_rows_ = 0;
     append_start_offset_ = 0;
   }
 
   auto columns =
       catalog->getAllColumnMetadataForTable(foreign_table_->tableId, false, false, true);
   std::map<int32_t, const ColumnDescriptor*> column_by_id{};
   for (auto column : columns) {
     column_by_id[column->columnId] = column;
   }
   MetadataScanMultiThreadingParams multi_threading_params;
   multi_threading_params.disable_cache = disable_cache_;
 
   // Restore previous chunk data
   if (foreign_table_->isAppendMode()) {
     multi_threading_params.chunk_encoder_buffers = std::move(chunk_encoder_buffers_);
   }
 
   std::set<int> columns_to_scan;
   for (auto column : columns) {
     if (!skip_metadata_scan(column)) {
       columns_to_scan.insert(column->columnId);
     }
   }
 
   // Track where scan started for appends
   int start_row = num_rows_;
   if (!file_reader_->isScanFinished()) {
     auto buffer_size = get_buffer_size(copy_params,
                                        file_reader_->isRemainingSizeKnown(),
                                        file_reader_->getRemainingSize());
     auto thread_count = get_thread_count(copy_params,
                                          file_reader_->isRemainingSizeKnown(),
                                          file_reader_->getRemainingSize(),
                                          buffer_size);
     multi_threading_params.continue_processing = true;
 
     std::vector<std::future<void>> futures{};
     for (size_t i = 0; i < thread_count; i++) {
       multi_threading_params.request_pool.emplace(buffer_size,
                                                   copy_params,
                                                   db_id_,
                                                   foreign_table_,
                                                   columns_to_scan,
                                                   getFullFilePath(foreign_table_),
                                                   nullptr,
                                                   disable_cache_);
       // TODO: when the cache is renabled for the import case, the above
       // relationship between `disable_cache_` and `track_rejected_rows`
       // will no longer hold and will need to be addressed using a different
       // approach
 
       futures.emplace_back(std::async(std::launch::async,
                                       scan_metadata,
                                       std::ref(multi_threading_params),
                                       std::ref(fragment_id_to_file_regions_map_),
                                       std::ref(parser)));
     }
 
     try {
       dispatch_metadata_scan_requests(buffer_size,
                                       file_path,
                                       (*file_reader_),
                                       copy_params,
                                       multi_threading_params,
                                       num_rows_,
                                       append_start_offset_,
                                       getFileBufferParser());
     } catch (...) {
       {
         std::unique_lock<std::mutex> pending_requests_lock(
             multi_threading_params.pending_requests_mutex);
         multi_threading_params.continue_processing = false;
       }
       multi_threading_params.pending_requests_condition.notify_all();
       throw;
     }
 
     for (auto& future : futures) {
       // get() instead of wait() because we need to propagate potential exceptions.
       future.get();
     }
   }
 
   for (auto& [chunk_key, buffer] : multi_threading_params.chunk_encoder_buffers) {
     auto column_entry = column_by_id.find(chunk_key[CHUNK_KEY_COLUMN_IDX]);
     CHECK(column_entry != column_by_id.end());
     const auto& column_type = column_entry->second->columnType;
     auto chunk_metadata = buffer->getEncoder()->getMetadata(column_type);
     chunk_metadata->numElements = buffer->getEncoder()->getNumElems();
     const auto& cached_chunks = multi_threading_params.cached_chunks;
     if (!column_type.is_varlen_indeed()) {
       chunk_metadata->numBytes = column_type.get_size() * chunk_metadata->numElements;
     } else if (auto chunk_entry = cached_chunks.find(chunk_key);
                chunk_entry != cached_chunks.end()) {
       auto buffer = chunk_entry->second.getBuffer();
       CHECK(buffer);
       chunk_metadata->numBytes = buffer->size();
     } else {
       CHECK_EQ(chunk_metadata->numBytes, static_cast<size_t>(0));
     }
     chunk_metadata_map_[chunk_key] = chunk_metadata;
   }
 
   for (auto column : columns) {
     if (skip_metadata_scan(column)) {
       add_placeholder_metadata(
           column, foreign_table_, db_id_, start_row, num_rows_, chunk_metadata_map_);
     }
   }
 
   for (auto& [chunk_key, chunk_metadata] : chunk_metadata_map_) {
     chunk_metadata_vector.emplace_back(chunk_key, chunk_metadata);
   }
 
   // Save chunk data
   if (foreign_table_->isAppendMode()) {
     chunk_encoder_buffers_ = std::move(multi_threading_params.chunk_encoder_buffers);
   }
 }

Here is the call graph for this function:

void foreign_storage::AbstractTextFileDataWrapper::populateChunks	(	std::map< int, Chunk_NS::Chunk > &	column_id_to_chunk_map,
		int	fragment_id,
		AbstractBuffer *	delete_buffer
	)

private

Populates provided chunks with appropriate data by parsing all file regions containing chunk data.

Parameters

column_id_to_chunk_map	- map of column id to chunks to be populated
fragment_id	- fragment id of given chunks
delete_buffer	- optional buffer to store deleted row indices

Definition at line 270 of file AbstractTextFileDataWrapper.cpp.

References Data_Namespace::AbstractBuffer::append(), threading_serial::async(), CHECK, db_id_, file_reader_, foreign_storage::ForeignTable::foreign_server, foreign_table_, fragment_id_to_file_regions_map_, foreign_storage::get_buffer_size(), foreign_storage::get_thread_count(), getFileBufferParser(), foreign_storage::AbstractFileStorageDataWrapper::getFullFilePath(), foreign_storage::AbstractFileStorageDataWrapper::LOCAL_FILE_STORAGE_TYPE, foreign_storage::OptionsContainer::options, foreign_storage::parse_file_regions(), run_benchmark_import::parser, render_group_analyzer_map_, Data_Namespace::AbstractBuffer::reserve(), run_benchmark_import::result, foreign_storage::AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY, UNREACHABLE, and foreign_storage::TextFileBufferParser::validateAndGetCopyParams().

Referenced by populateChunkBuffers().

                                    {
   const auto copy_params = getFileBufferParser().validateAndGetCopyParams(foreign_table_);
 
   CHECK(!column_id_to_chunk_map.empty());
   const auto& file_regions = fragment_id_to_file_regions_map_[fragment_id];
   CHECK(!file_regions.empty());
 
   const auto buffer_size = get_buffer_size(file_regions);
   const auto thread_count = get_thread_count(copy_params, file_regions);
 
   const int batch_size = (file_regions.size() + thread_count - 1) / thread_count;
 
   std::vector<ParseBufferRequest> parse_file_requests{};
   parse_file_requests.reserve(thread_count);
   std::vector<std::future<ParseFileRegionResult>> futures{};
   std::set<int> column_filter_set;
   for (const auto& pair : column_id_to_chunk_map) {
     column_filter_set.insert(pair.first);
   }
 
   std::vector<std::unique_ptr<FileReader>> file_readers;
   rapidjson::Value reader_metadata(rapidjson::kObjectType);
   rapidjson::Document d;
   auto& server_options = foreign_table_->foreign_server->options;
   file_reader_->serialize(reader_metadata, d.GetAllocator());
   const auto file_path = getFullFilePath(foreign_table_);
   auto& parser = getFileBufferParser();
 
   for (size_t i = 0; i < file_regions.size(); i += batch_size) {
     parse_file_requests.emplace_back(buffer_size,
                                      copy_params,
                                      db_id_,
                                      foreign_table_,
                                      column_filter_set,
                                      file_path,
                                      &render_group_analyzer_map_,
                                      delete_buffer != nullptr);
     auto start_index = i;
     auto end_index =
         std::min<size_t>(start_index + batch_size - 1, file_regions.size() - 1);
 
     if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
       file_readers.emplace_back(std::make_unique<LocalMultiFileReader>(
           file_path, copy_params, reader_metadata));
     } else {
       UNREACHABLE();
     }
 
     futures.emplace_back(std::async(std::launch::async,
                                     parse_file_regions,
                                     std::ref(file_regions),
                                     start_index,
                                     end_index,
                                     std::ref(*(file_readers.back())),
                                     std::ref(parse_file_requests.back()),
                                     std::ref(column_id_to_chunk_map),
                                     std::ref(parser)));
   }
 
   for (auto& future : futures) {
     future.wait();
   }
 
   std::vector<ParseFileRegionResult> load_file_region_results{};
   for (auto& future : futures) {
     load_file_region_results.emplace_back(future.get());
   }
 
   std::set<size_t> chunk_rejected_row_indices;
   size_t chunk_offset = 0;
   for (auto result : load_file_region_results) {
     for (auto& [column_id, chunk] : column_id_to_chunk_map) {
       chunk.appendData(
           result.column_id_to_data_blocks_map[column_id], result.row_count, 0);
     }
     for (const auto& rejected_row_index : result.rejected_row_indices) {
       chunk_rejected_row_indices.insert(rejected_row_index + chunk_offset);
     }
     chunk_offset += result.row_count;
   }
 
   if (delete_buffer) {
     auto chunk_element_count = chunk_offset;
     delete_buffer->reserve(chunk_element_count);
     for (size_t i = 0; i < chunk_element_count; ++i) {
       if (chunk_rejected_row_indices.find(i) != chunk_rejected_row_indices.end()) {
         int8_t true_byte = true;
         delete_buffer->append(&true_byte, 1);
       } else {
         int8_t false_byte = false;
         delete_buffer->append(&false_byte, 1);
       }
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::AbstractTextFileDataWrapper::restoreDataWrapperInternals	(	const std::string &	file_path,
		const ChunkMetadataVector &	chunk_metadata
	)

overridevirtual

Restore internal state of datawrapper

Parameters

file_path	- location of file created by serializeMetadata
chunk_metadata_vector	- vector of chunk metadata recovered from disk

Implements foreign_storage::ForeignDataWrapper.

Definition at line 1017 of file AbstractTextFileDataWrapper.cpp.

References append_start_offset_, CHECK, chunk_encoder_buffers_, chunk_metadata_map_, file_reader_, foreign_storage::ForeignTable::foreign_server, foreign_table_, fragment_id_to_file_regions_map_, foreign_storage::json_utils::get_value_from_object(), getFileBufferParser(), foreign_storage::AbstractFileStorageDataWrapper::getFullFilePath(), is_restored_, foreign_storage::ForeignTable::isAppendMode(), foreign_storage::AbstractFileStorageDataWrapper::LOCAL_FILE_STORAGE_TYPE, num_rows_, foreign_storage::OptionsContainer::options, foreign_storage::json_utils::read_from_file(), foreign_storage::AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY, UNREACHABLE, and foreign_storage::TextFileBufferParser::validateAndGetCopyParams().

                                                {
   auto d = json_utils::read_from_file(file_path);
   CHECK(d.IsObject());
 
   // Restore fragment map
   json_utils::get_value_from_object(
       d, fragment_id_to_file_regions_map_, "fragment_id_to_file_regions_map");
 
   // Construct reader with metadta
   CHECK(d.HasMember("reader_metadata"));
   const auto copy_params = getFileBufferParser().validateAndGetCopyParams(foreign_table_);
   const auto full_file_path = getFullFilePath(foreign_table_);
   auto& server_options = foreign_table_->foreign_server->options;
   if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
     file_reader_ = std::make_unique<LocalMultiFileReader>(
         full_file_path, copy_params, d["reader_metadata"]);
   } else {
     UNREACHABLE();
   }
 
   json_utils::get_value_from_object(d, num_rows_, "num_rows");
   json_utils::get_value_from_object(d, append_start_offset_, "append_start_offset");
 
   // Now restore the internal metadata maps
   CHECK(chunk_metadata_map_.empty());
   CHECK(chunk_encoder_buffers_.empty());
 
   for (auto& pair : chunk_metadata) {
     chunk_metadata_map_[pair.first] = pair.second;
 
     if (foreign_table_->isAppendMode()) {
       // Restore encoder state for append mode
       chunk_encoder_buffers_[pair.first] = std::make_unique<ForeignStorageBuffer>();
       chunk_encoder_buffers_[pair.first]->initEncoder(pair.second->sqlType);
       chunk_encoder_buffers_[pair.first]->setSize(pair.second->numBytes);
       chunk_encoder_buffers_[pair.first]->getEncoder()->setNumElems(
           pair.second->numElements);
       chunk_encoder_buffers_[pair.first]->getEncoder()->resetChunkStats(
           pair.second->chunkStats);
       chunk_encoder_buffers_[pair.first]->setUpdated();
     }
   }
   is_restored_ = true;
 }

Here is the call graph for this function:

void foreign_storage::AbstractTextFileDataWrapper::updateMetadata	(	std::map< int, Chunk_NS::Chunk > &	column_id_to_chunk_map,
		int	fragment_id
	)

private

Definition at line 125 of file AbstractTextFileDataWrapper.cpp.

References CHECK, chunk_metadata_map_, db_id_, foreign_table_, shared::get_from_map(), Catalog_Namespace::SysCatalog::getCatalog(), Catalog_Namespace::SysCatalog::instance(), foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::skip_metadata_scan(), and TableDescriptor::tableId.

Referenced by populateChunkBuffers().

                      {
   auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
   CHECK(catalog);
   for (auto& entry : column_id_to_chunk_map) {
     const auto& column =
         catalog->getMetadataForColumn(foreign_table_->tableId, entry.first);
     if (skip_metadata_scan(column)) {
       ChunkKey data_chunk_key = {
           db_id_, foreign_table_->tableId, column->columnId, fragment_id};
       if (column->columnType.is_varlen_indeed()) {
         data_chunk_key.emplace_back(1);
       }
       CHECK(chunk_metadata_map_.find(data_chunk_key) != chunk_metadata_map_.end());
       // Allocate new shared_ptr for metadata so we dont modify old one which may be
       // used by executor
       auto cached_metadata_previous =
           shared::get_from_map(chunk_metadata_map_, data_chunk_key);
       shared::get_from_map(chunk_metadata_map_, data_chunk_key) =
           std::make_shared<ChunkMetadata>();
       auto cached_metadata = shared::get_from_map(chunk_metadata_map_, data_chunk_key);
       *cached_metadata = *cached_metadata_previous;
       auto chunk_metadata =
           entry.second.getBuffer()->getEncoder()->getMetadata(column->columnType);
       cached_metadata->chunkStats.max = chunk_metadata->chunkStats.max;
       cached_metadata->chunkStats.min = chunk_metadata->chunkStats.min;
       cached_metadata->chunkStats.has_nulls = chunk_metadata->chunkStats.has_nulls;
       cached_metadata->numBytes = entry.second.getBuffer()->size();
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

size_t foreign_storage::AbstractTextFileDataWrapper::append_start_offset_

private

Definition at line 102 of file AbstractTextFileDataWrapper.h.

Referenced by getSerializedDataWrapper(), populateChunkMetadata(), and restoreDataWrapperInternals().

std::map<ChunkKey, std::unique_ptr<ForeignStorageBuffer> > foreign_storage::AbstractTextFileDataWrapper::chunk_encoder_buffers_

private

Definition at line 98 of file AbstractTextFileDataWrapper.h.

Referenced by populateChunkMetadata(), and restoreDataWrapperInternals().

std::map<ChunkKey, std::shared_ptr<ChunkMetadata> > foreign_storage::AbstractTextFileDataWrapper::chunk_metadata_map_

private

Definition at line 89 of file AbstractTextFileDataWrapper.h.

Referenced by populateChunkMapForColumns(), populateChunkMetadata(), restoreDataWrapperInternals(), and updateMetadata().

const int foreign_storage::AbstractTextFileDataWrapper::db_id_

private

Definition at line 94 of file AbstractTextFileDataWrapper.h.

Referenced by createRenderGroupAnalyzers(), populateChunkBuffers(), populateChunkMapForColumns(), populateChunkMetadata(), populateChunks(), and updateMetadata().

const bool foreign_storage::AbstractTextFileDataWrapper::disable_cache_

private

Definition at line 109 of file AbstractTextFileDataWrapper.h.

Referenced by populateChunkMetadata().

std::unique_ptr<FileReader> foreign_storage::AbstractTextFileDataWrapper::file_reader_

private

Definition at line 92 of file AbstractTextFileDataWrapper.h.

Referenced by getSerializedDataWrapper(), populateChunkMetadata(), populateChunks(), and restoreDataWrapperInternals().

const ForeignTable* foreign_storage::AbstractTextFileDataWrapper::foreign_table_

private

Definition at line 95 of file AbstractTextFileDataWrapper.h.

Referenced by createRenderGroupAnalyzers(), populateChunkBuffers(), populateChunkMapForColumns(), populateChunkMetadata(), populateChunks(), restoreDataWrapperInternals(), and updateMetadata().

std::map<int, FileRegions> foreign_storage::AbstractTextFileDataWrapper::fragment_id_to_file_regions_map_

private

Definition at line 90 of file AbstractTextFileDataWrapper.h.

Referenced by getSerializedDataWrapper(), populateChunkMetadata(), populateChunks(), and restoreDataWrapperInternals().

bool foreign_storage::AbstractTextFileDataWrapper::is_restored_

private

Definition at line 104 of file AbstractTextFileDataWrapper.h.

Referenced by isRestored(), and restoreDataWrapperInternals().

size_t foreign_storage::AbstractTextFileDataWrapper::num_rows_

private

Definition at line 100 of file AbstractTextFileDataWrapper.h.

Referenced by getSerializedDataWrapper(), populateChunkMetadata(), and restoreDataWrapperInternals().

RenderGroupAnalyzerMap foreign_storage::AbstractTextFileDataWrapper::render_group_analyzer_map_

private

Definition at line 114 of file AbstractTextFileDataWrapper.h.

Referenced by createRenderGroupAnalyzers(), and populateChunks().

const UserMapping* foreign_storage::AbstractTextFileDataWrapper::user_mapping_

private

Definition at line 106 of file AbstractTextFileDataWrapper.h.

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

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

Public Member Functions

Protected Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation