#include <ParquetDataWrapper.h>

Inheritance diagram for foreign_storage::ParquetDataWrapper:

Collaboration diagram for foreign_storage::ParquetDataWrapper:

Public Member Functions
	ParquetDataWrapper ()

	ParquetDataWrapper (const int db_id, const ForeignTable *foreign_table)

void	populateChunkMetadata (ChunkMetadataVector &chunk_metadata_vector) override

void	populateChunkBuffers (const ChunkToBufferMap &required_buffers, const ChunkToBufferMap &optional_buffers) override

std::string	getSerializedDataWrapper () const override

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

bool	isRestored () const override

ParallelismLevel	getCachedParallelismLevel () const override

ParallelismLevel	getNonCachedParallelismLevel () const override

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 void	validateSchema (const std::list< ColumnDescriptor > &columns) const

Private Member Functions
std::list< const ColumnDescriptor * >	getColumnsToInitialize (const Interval< ColumnType > &column_interval)

void	initializeChunkBuffers (const int fragment_index, const Interval< ColumnType > &column_interval, const ChunkToBufferMap &required_buffers, const bool reserve_buffers_and_set_stats=false)

void	fetchChunkMetadata ()

void	loadBuffersUsingLazyParquetChunkLoader (const int logical_column_id, const int fragment_id, const ChunkToBufferMap &required_buffers)

std::set< std::string >	getProcessedFilePaths ()

std::vector< std::string >	getAllFilePaths ()

bool	moveToNextFragment (size_t new_rows_count) const

void	finalizeFragmentMap ()

void	addNewFragment (int row_group, const std::string &file_path)

bool	isNewFile (const std::string &file_path) const

void	addNewFile (const std::string &file_path)

void	resetParquetMetadata ()

void	metadataScanFiles (const std::vector< std::string > &file_paths)

Private Attributes
std::map< int, std::vector < RowGroupInterval > >	fragment_to_row_group_interval_map_

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

const int	db_id_

const ForeignTable *	foreign_table_

int	last_fragment_index_

size_t	last_fragment_row_count_

size_t	total_row_count_

int	last_row_group_

bool	is_restored_

std::unique_ptr < ForeignTableSchema >	schema_

std::shared_ptr < arrow::fs::FileSystem >	file_system_

std::unique_ptr< FileReaderMap >	file_reader_cache_

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 35 of file ParquetDataWrapper.h.

Constructor & Destructor Documentation

foreign_storage::ParquetDataWrapper::ParquetDataWrapper ( )

Definition at line 72 of file ParquetDataWrapper.cpp.

72 : db_id_(-1), foreign_table_(nullptr) {}

foreign_storage::ParquetDataWrapper::foreign_table_

const ForeignTable * foreign_table_

Definition: ParquetDataWrapper.h:91

foreign_storage::ParquetDataWrapper::db_id_

const int db_id_

Definition: ParquetDataWrapper.h:90

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

Definition at line 74 of file ParquetDataWrapper.cpp.

References file_system_, foreign_storage::ForeignTable::foreign_server, foreign_storage::AbstractFileStorageDataWrapper::LOCAL_FILE_STORAGE_TYPE, foreign_storage::OptionsContainer::options, foreign_storage::AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY, and UNREACHABLE.

     : db_id_(db_id)
     , foreign_table_(foreign_table)
     , last_fragment_index_(0)
     , last_fragment_row_count_(0)
     , total_row_count_(0)
     , last_row_group_(0)
     , is_restored_(false)
     , schema_(std::make_unique<ForeignTableSchema>(db_id, foreign_table))
     , file_reader_cache_(std::make_unique<FileReaderMap>()) {
   auto& server_options = foreign_table->foreign_server->options;
   if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
     file_system_ = std::make_shared<arrow::fs::LocalFileSystem>();
   } else {
     UNREACHABLE();
   }
 }

Member Function Documentation

void foreign_storage::ParquetDataWrapper::addNewFile ( const std::string & file_path )

private

Definition at line 203 of file ParquetDataWrapper.cpp.

References CHECK, CHECK_EQ, fragment_to_row_group_interval_map_, last_fragment_index_, and last_row_group_.

Referenced by metadataScanFiles().

                                                               {
   const auto last_fragment_entry =
       fragment_to_row_group_interval_map_.find(last_fragment_index_);
   CHECK(last_fragment_entry != fragment_to_row_group_interval_map_.end());
 
   // The entry for the first fragment starts out as an empty vector
   if (last_fragment_entry->second.empty()) {
     CHECK_EQ(last_fragment_index_, 0);
   } else {
     last_fragment_entry->second.back().end_index = last_row_group_;
   }
   last_fragment_entry->second.emplace_back(RowGroupInterval{file_path, 0});
 }

Here is the caller graph for this function:

void foreign_storage::ParquetDataWrapper::addNewFragment	(	int	row_group,
		const std::string &	file_path
	)

private

Definition at line 177 of file ParquetDataWrapper.cpp.

References CHECK, fragment_to_row_group_interval_map_, last_fragment_index_, last_fragment_row_count_, and last_row_group_.

Referenced by metadataScanFiles().

                                                                                  {
   const auto last_fragment_entry =
       fragment_to_row_group_interval_map_.find(last_fragment_index_);
   CHECK(last_fragment_entry != fragment_to_row_group_interval_map_.end());
 
   last_fragment_entry->second.back().end_index = last_row_group_;
   last_fragment_index_++;
   last_fragment_row_count_ = 0;
   fragment_to_row_group_interval_map_[last_fragment_index_].emplace_back(
       RowGroupInterval{file_path, row_group});
 }

Here is the caller graph for this function:

void foreign_storage::ParquetDataWrapper::fetchChunkMetadata ( )

private

Definition at line 217 of file ParquetDataWrapper.cpp.

References CHECK, CHECK_EQ, chunk_metadata_map_, shared::contains(), db_id_, file_reader_cache_, file_system_, foreign_table_, getAllFilePaths(), Catalog_Namespace::SysCatalog::getCatalog(), getProcessedFilePaths(), Catalog_Namespace::SysCatalog::instance(), foreign_storage::ForeignTable::isAppendMode(), metadataScanFiles(), resetParquetMetadata(), foreign_storage::throw_removed_file_error(), foreign_storage::throw_removed_row_error(), and total_row_count_.

Referenced by populateChunkMetadata().

                                             {
   auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
   CHECK(catalog);
   std::vector<std::string> new_file_paths;
   auto processed_file_paths = getProcessedFilePaths();
   if (foreign_table_->isAppendMode() && !processed_file_paths.empty()) {
     auto all_file_paths = getAllFilePaths();
     for (const auto& file_path : processed_file_paths) {
       if (!shared::contains(all_file_paths, file_path)) {
         throw_removed_file_error(file_path);
       }
     }
 
     for (const auto& file_path : all_file_paths) {
       if (!shared::contains(processed_file_paths, file_path)) {
         new_file_paths.emplace_back(file_path);
       }
     }
 
     // Single file append
     // If an append occurs with multiple files, then we assume any existing files have
     // not been altered.  If an append occurs on a single file, then we check to see if
     // it has changed.
     if (new_file_paths.empty() && all_file_paths.size() == 1) {
       CHECK_EQ(processed_file_paths.size(), static_cast<size_t>(1));
       const auto& file_path = *all_file_paths.begin();
       CHECK_EQ(*processed_file_paths.begin(), file_path);
 
       // Since an existing file is being appended to we need to update the cached
       // FileReader as the existing one will be out of date.
       auto reader = file_reader_cache_->insert(file_path, file_system_);
       size_t row_count = reader->parquet_reader()->metadata()->num_rows();
 
       if (row_count < total_row_count_) {
         throw_removed_row_error(file_path);
       } else if (row_count > total_row_count_) {
         new_file_paths = all_file_paths;
         chunk_metadata_map_.clear();
         resetParquetMetadata();
       }
     }
   } else {
     CHECK(chunk_metadata_map_.empty());
     new_file_paths = getAllFilePaths();
     resetParquetMetadata();
   }
 
   if (!new_file_paths.empty()) {
     metadataScanFiles(new_file_paths);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::ParquetDataWrapper::finalizeFragmentMap ( )

private

Definition at line 172 of file ParquetDataWrapper.cpp.

References fragment_to_row_group_interval_map_, last_fragment_index_, and last_row_group_.

Referenced by metadataScanFiles().

                                              {
   fragment_to_row_group_interval_map_[last_fragment_index_].back().end_index =
       last_row_group_;
 }

Here is the caller graph for this function:

std::vector< std::string > foreign_storage::ParquetDataWrapper::getAllFilePaths ( )

private

Definition at line 279 of file ParquetDataWrapper.cpp.

References DEBUG_TIMER, foreign_storage::AbstractFileStorageDataWrapper::FILE_SORT_ORDER_BY_KEY, foreign_storage::AbstractFileStorageDataWrapper::FILE_SORT_REGEX_KEY, foreign_storage::ForeignTable::foreign_server, foreign_table_, foreign_storage::AbstractFileStorageDataWrapper::getFullFilePath(), foreign_storage::OptionsContainer::getOption(), foreign_storage::AbstractFileStorageDataWrapper::LOCAL_FILE_STORAGE_TYPE, shared::local_glob_filter_sort_files(), foreign_storage::OptionsContainer::options, foreign_storage::AbstractFileStorageDataWrapper::REGEX_PATH_FILTER_KEY, foreign_storage::AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY, and UNREACHABLE.

Referenced by fetchChunkMetadata().

                                                          {
   auto timer = DEBUG_TIMER(__func__);
   std::vector<std::string> found_file_paths;
   auto file_path = getFullFilePath(foreign_table_);
   const auto& regex_pattern = foreign_table_->getOption(REGEX_PATH_FILTER_KEY);
   const auto& sort_by = foreign_table_->getOption(FILE_SORT_ORDER_BY_KEY);
   const auto& sort_regex = foreign_table_->getOption(FILE_SORT_REGEX_KEY);
 
   auto& server_options = foreign_table_->foreign_server->options;
   if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
     found_file_paths = shared::local_glob_filter_sort_files(
         file_path, regex_pattern, sort_by, sort_regex);
   } else {
     UNREACHABLE();
   }
   return found_file_paths;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ParallelismLevel foreign_storage::ParquetDataWrapper::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 54 of file ParquetDataWrapper.h.

References foreign_storage::ForeignDataWrapper::INTER_FRAGMENT.

54 { return INTER_FRAGMENT; }

foreign_storage::ForeignDataWrapper::INTER_FRAGMENT

Definition: ForeignDataWrapper.h:124

std::list< const ColumnDescriptor * > foreign_storage::ParquetDataWrapper::getColumnsToInitialize ( const Interval< ColumnType > & column_interval )

private

Definition at line 103 of file ParquetDataWrapper.cpp.

References CHECK, db_id_, foreign_storage::Interval< T >::end, Catalog_Namespace::SysCatalog::getCatalog(), Catalog_Namespace::SysCatalog::instance(), schema_, and foreign_storage::Interval< T >::start.

Referenced by initializeChunkBuffers().

                                                  {
   const auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
   CHECK(catalog);
   const auto& columns = schema_->getLogicalAndPhysicalColumns();
   auto column_start = column_interval.start;
   auto column_end = column_interval.end;
   std::list<const ColumnDescriptor*> columns_to_init;
   for (const auto column : columns) {
     auto column_id = column->columnId;
     if (column_id >= column_start && column_id <= column_end) {
       columns_to_init.push_back(column);
     }
   }
   return columns_to_init;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

ParallelismLevel foreign_storage::ParquetDataWrapper::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 56 of file ParquetDataWrapper.h.

References foreign_storage::ForeignDataWrapper::INTRA_FRAGMENT.

                                                                  {
     return INTRA_FRAGMENT;
   }

std::set< std::string > foreign_storage::ParquetDataWrapper::getProcessedFilePaths ( )

private

Definition at line 269 of file ParquetDataWrapper.cpp.

References fragment_to_row_group_interval_map_.

Referenced by fetchChunkMetadata().

                                                             {
   std::set<std::string> file_paths;
   for (const auto& entry : fragment_to_row_group_interval_map_) {
     for (const auto& row_group_interval : entry.second) {
       file_paths.emplace(row_group_interval.file_path);
     }
   }
   return file_paths;
 }

Here is the caller graph for this function:

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

overridevirtual

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

Implements foreign_storage::ForeignDataWrapper.

Definition at line 507 of file ParquetDataWrapper.cpp.

References foreign_storage::json_utils::add_value_to_object(), fragment_to_row_group_interval_map_, last_fragment_index_, last_fragment_row_count_, last_row_group_, total_row_count_, and foreign_storage::json_utils::write_to_string().

                                                              {
   rapidjson::Document d;
   d.SetObject();
 
   json_utils::add_value_to_object(d,
                                   fragment_to_row_group_interval_map_,
                                   "fragment_to_row_group_interval_map",
                                   d.GetAllocator());
   json_utils::add_value_to_object(d, last_row_group_, "last_row_group", d.GetAllocator());
   json_utils::add_value_to_object(
       d, last_fragment_index_, "last_fragment_index", d.GetAllocator());
   json_utils::add_value_to_object(
       d, last_fragment_row_count_, "last_fragment_row_count", d.GetAllocator());
   json_utils::add_value_to_object(
       d, total_row_count_, "total_row_count", d.GetAllocator());
   return json_utils::write_to_string(d);
 }

Here is the call graph for this function:

void foreign_storage::ParquetDataWrapper::initializeChunkBuffers	(	const int	fragment_index,
		const Interval< ColumnType > &	column_interval,
		const ChunkToBufferMap &	required_buffers,
		const bool	reserve_buffers_and_set_stats = `false`
	)

private

Definition at line 120 of file ParquetDataWrapper.cpp.

References CHECK, chunk_metadata_map_, db_id_, foreign_table_, shared::get_from_map(), getColumnsToInitialize(), kENCODING_NONE, and TableDescriptor::tableId.

Referenced by loadBuffersUsingLazyParquetChunkLoader().

                                               {
   for (const auto column : getColumnsToInitialize(column_interval)) {
     Chunk_NS::Chunk chunk{column};
     ChunkKey data_chunk_key;
     if (column->columnType.is_varlen_indeed()) {
       data_chunk_key = {
           db_id_, foreign_table_->tableId, column->columnId, fragment_index, 1};
       CHECK(required_buffers.find(data_chunk_key) != required_buffers.end());
       auto data_buffer = shared::get_from_map(required_buffers, data_chunk_key);
       chunk.setBuffer(data_buffer);
 
       ChunkKey index_chunk_key{
           db_id_, foreign_table_->tableId, column->columnId, fragment_index, 2};
       CHECK(required_buffers.find(index_chunk_key) != required_buffers.end());
       auto index_buffer = shared::get_from_map(required_buffers, index_chunk_key);
       chunk.setIndexBuffer(index_buffer);
     } else {
       data_chunk_key = {
           db_id_, foreign_table_->tableId, column->columnId, fragment_index};
       CHECK(required_buffers.find(data_chunk_key) != required_buffers.end());
       auto data_buffer = shared::get_from_map(required_buffers, data_chunk_key);
       chunk.setBuffer(data_buffer);
     }
     chunk.initEncoder();
     if (reserve_buffers_and_set_stats) {
       const auto metadata_it = chunk_metadata_map_.find(data_chunk_key);
       CHECK(metadata_it != chunk_metadata_map_.end());
       auto buffer = chunk.getBuffer();
       auto& metadata = metadata_it->second;
       auto encoder = buffer->getEncoder();
       encoder->resetChunkStats(metadata->chunkStats);
       encoder->setNumElems(metadata->numElements);
       if (column->columnType.is_string() &&
           column->columnType.get_compression() == kENCODING_NONE) {
         auto index_buffer = chunk.getIndexBuf();
         index_buffer->reserve(sizeof(StringOffsetT) * (metadata->numElements + 1));
       } else if (!column->columnType.is_fixlen_array() && column->columnType.is_array()) {
         auto index_buffer = chunk.getIndexBuf();
         index_buffer->reserve(sizeof(ArrayOffsetT) * (metadata->numElements + 1));
       } else {
         size_t num_bytes_to_reserve =
             metadata->numElements * column->columnType.get_size();
         buffer->reserve(num_bytes_to_reserve);
       }
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool foreign_storage::ParquetDataWrapper::isNewFile ( const std::string & file_path ) const

private

Definition at line 189 of file ParquetDataWrapper.cpp.

References CHECK, CHECK_EQ, fragment_to_row_group_interval_map_, and last_fragment_index_.

Referenced by metadataScanFiles().

                                                                    {
   const auto last_fragment_entry =
       fragment_to_row_group_interval_map_.find(last_fragment_index_);
   CHECK(last_fragment_entry != fragment_to_row_group_interval_map_.end());
 
   // The entry for the first fragment starts out as an empty vector
   if (last_fragment_entry->second.empty()) {
     CHECK_EQ(last_fragment_index_, 0);
     return true;
   } else {
     return (last_fragment_entry->second.back().file_path != file_path);
   }
 }

Here is the caller graph for this function:

bool foreign_storage::ParquetDataWrapper::isRestored ( ) const

overridevirtual

Implements foreign_storage::ForeignDataWrapper.

Definition at line 546 of file ParquetDataWrapper.cpp.

References is_restored_.

                                           {
   return is_restored_;
 }

void foreign_storage::ParquetDataWrapper::loadBuffersUsingLazyParquetChunkLoader	(	const int	logical_column_id,
		const int	fragment_id,
		const ChunkToBufferMap &	required_buffers
	)

private

Definition at line 358 of file ParquetDataWrapper.cpp.

References CHECK, chunk_metadata_map_, ColumnDescriptor::columnType, db_id_, foreign_storage::Interval< T >::end, file_reader_cache_, file_system_, foreign_table_, fragment_to_row_group_interval_map_, TableDescriptor::fragmenter, SQLTypeInfo::get_comp_param(), SQLTypeInfo::get_elem_type(), shared::get_from_map(), SQLTypeInfo::get_physical_cols(), Catalog_Namespace::SysCatalog::getCatalog(), initializeChunkBuffers(), Catalog_Namespace::SysCatalog::instance(), SQLTypeInfo::is_array(), SQLTypeInfo::is_dict_encoded_string(), SQLTypeInfo::is_geometry(), foreign_storage::LazyParquetChunkLoader::loadChunk(), schema_, foreign_storage::Interval< T >::start, and TableDescriptor::tableId.

Referenced by populateChunkBuffers().

                                               {
   auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
   CHECK(catalog);
   const ColumnDescriptor* logical_column =
       schema_->getColumnDescriptor(logical_column_id);
   auto parquet_column_index = schema_->getParquetColumnIndex(logical_column_id);
 
   const Interval<ColumnType> column_interval = {
       logical_column_id,
       logical_column_id + logical_column->columnType.get_physical_cols()};
   initializeChunkBuffers(fragment_id, column_interval, required_buffers, true);
 
   const auto& row_group_intervals =
       shared::get_from_map(fragment_to_row_group_interval_map_, fragment_id);
 
   const bool is_dictionary_encoded_string_column =
       logical_column->columnType.is_dict_encoded_string() ||
       (logical_column->columnType.is_array() &&
        logical_column->columnType.get_elem_type().is_dict_encoded_string());
 
   StringDictionary* string_dictionary = nullptr;
   if (is_dictionary_encoded_string_column) {
     auto dict_descriptor =
         catalog->getMetadataForDict(logical_column->columnType.get_comp_param(), true);
     CHECK(dict_descriptor);
     string_dictionary = dict_descriptor->stringDict.get();
   }
 
   std::list<Chunk_NS::Chunk> chunks;
   for (int column_id = column_interval.start; column_id <= column_interval.end;
        ++column_id) {
     auto column_descriptor = schema_->getColumnDescriptor(column_id);
     Chunk_NS::Chunk chunk{column_descriptor};
     if (column_descriptor->columnType.is_varlen_indeed()) {
       ChunkKey data_chunk_key = {
           db_id_, foreign_table_->tableId, column_id, fragment_id, 1};
       auto buffer = shared::get_from_map(required_buffers, data_chunk_key);
       chunk.setBuffer(buffer);
       ChunkKey index_chunk_key = {
           db_id_, foreign_table_->tableId, column_id, fragment_id, 2};
       auto index_buffer = shared::get_from_map(required_buffers, index_chunk_key);
       chunk.setIndexBuffer(index_buffer);
     } else {
       ChunkKey chunk_key = {db_id_, foreign_table_->tableId, column_id, fragment_id};
       auto buffer = shared::get_from_map(required_buffers, chunk_key);
       chunk.setBuffer(buffer);
     }
     chunks.emplace_back(chunk);
   }
 
   LazyParquetChunkLoader chunk_loader(file_system_, file_reader_cache_.get());
   auto metadata = chunk_loader.loadChunk(
       row_group_intervals, parquet_column_index, chunks, string_dictionary);
   auto fragmenter = foreign_table_->fragmenter;
 
   auto metadata_iter = metadata.begin();
   for (int column_id = column_interval.start; column_id <= column_interval.end;
        ++column_id, ++metadata_iter) {
     auto column = schema_->getColumnDescriptor(column_id);
     ChunkKey data_chunk_key = {db_id_, foreign_table_->tableId, column_id, 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;
 
     CHECK(required_buffers.find(data_chunk_key) != required_buffers.end());
     cached_metadata->numBytes =
         shared::get_from_map(required_buffers, data_chunk_key)->size();
 
     // for certain types, update the metadata statistics
     // should update the fragmenter, cache, and the internal chunk_metadata_map_
     if (is_dictionary_encoded_string_column || logical_column->columnType.is_geometry()) {
       CHECK(metadata_iter != metadata.end());
       auto& chunk_metadata_ptr = *metadata_iter;
       cached_metadata->chunkStats.max = chunk_metadata_ptr->chunkStats.max;
       cached_metadata->chunkStats.min = chunk_metadata_ptr->chunkStats.min;
 
       // Update stats on buffer so it is saved in cache
       shared::get_from_map(required_buffers, data_chunk_key)
           ->getEncoder()
           ->resetChunkStats(cached_metadata->chunkStats);
     }
 
     if (fragmenter) {
       fragmenter->updateColumnChunkMetadata(column, fragment_id, cached_metadata);
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::ParquetDataWrapper::metadataScanFiles ( const std::vector< std::string > & file_paths )

private

Definition at line 297 of file ParquetDataWrapper.cpp.

References addNewFile(), addNewFragment(), CHECK, CHECK_EQ, chunk_metadata_map_, db_id_, file_reader_cache_, file_system_, finalizeFragmentMap(), foreign_table_, isNewFile(), last_fragment_index_, last_fragment_row_count_, last_row_group_, foreign_storage::LazyParquetChunkLoader::metadataScan(), moveToNextFragment(), foreign_storage::anonymous_namespace{ParquetDataWrapper.cpp}::reduce_metadata(), schema_, TableDescriptor::tableId, and total_row_count_.

Referenced by fetchChunkMetadata().

                                                                                  {
   LazyParquetChunkLoader chunk_loader(file_system_, file_reader_cache_.get());
   auto row_group_metadata = chunk_loader.metadataScan(file_paths, *schema_);
   auto column_interval =
       Interval<ColumnType>{schema_->getLogicalAndPhysicalColumns().front()->columnId,
                            schema_->getLogicalAndPhysicalColumns().back()->columnId};
 
   for (const auto& row_group_metadata_item : row_group_metadata) {
     const auto& column_chunk_metadata = row_group_metadata_item.column_chunk_metadata;
     CHECK(static_cast<int>(column_chunk_metadata.size()) ==
           schema_->numLogicalAndPhysicalColumns());
     auto column_chunk_metadata_iter = column_chunk_metadata.begin();
     const int64_t import_row_count = (*column_chunk_metadata_iter)->numElements;
     int row_group = row_group_metadata_item.row_group_index;
     const auto& file_path = row_group_metadata_item.file_path;
     if (moveToNextFragment(import_row_count)) {
       addNewFragment(row_group, file_path);
     } else if (isNewFile(file_path)) {
       CHECK_EQ(row_group, 0);
       addNewFile(file_path);
     }
     last_row_group_ = row_group;
 
     for (int column_id = column_interval.start; column_id <= column_interval.end;
          column_id++, column_chunk_metadata_iter++) {
       CHECK(column_chunk_metadata_iter != column_chunk_metadata.end());
       const auto column_descriptor = schema_->getColumnDescriptor(column_id);
 
       const auto& type_info = column_descriptor->columnType;
       ChunkKey chunk_key{
           db_id_, foreign_table_->tableId, column_id, last_fragment_index_};
       ChunkKey data_chunk_key = chunk_key;
       if (type_info.is_varlen_indeed()) {
         data_chunk_key.emplace_back(1);
       }
       std::shared_ptr<ChunkMetadata> chunk_metadata = *column_chunk_metadata_iter;
       if (chunk_metadata_map_.find(data_chunk_key) == chunk_metadata_map_.end()) {
         chunk_metadata_map_[data_chunk_key] = chunk_metadata;
       } else {
         reduce_metadata(chunk_metadata_map_[data_chunk_key], chunk_metadata);
       }
     }
     last_fragment_row_count_ += import_row_count;
     total_row_count_ += import_row_count;
   }
   finalizeFragmentMap();
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool foreign_storage::ParquetDataWrapper::moveToNextFragment ( size_t new_rows_count ) const

private

Definition at line 345 of file ParquetDataWrapper.cpp.

References foreign_table_, last_fragment_row_count_, and TableDescriptor::maxFragRows.

Referenced by metadataScanFiles().

                                                                        {
   return (last_fragment_row_count_ + new_rows_count) >
          static_cast<size_t>(foreign_table_->maxFragRows);
 }

Here is the caller graph for this function:

void foreign_storage::ParquetDataWrapper::populateChunkBuffers	(	const ChunkToBufferMap &	required_buffers,
		const ChunkToBufferMap &	optional_buffers
	)

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)

Implements foreign_storage::ForeignDataWrapper.

Definition at line 459 of file ParquetDataWrapper.cpp.

References threading_serial::async(), CHECK, CHECK_EQ, CHUNK_KEY_COLUMN_IDX, CHUNK_KEY_FRAGMENT_IDX, g_max_import_threads, loadBuffersUsingLazyParquetChunkLoader(), foreign_storage::partition_for_threads(), and schema_.

                                                                                         {
   ChunkToBufferMap buffers_to_load;
   buffers_to_load.insert(required_buffers.begin(), required_buffers.end());
   buffers_to_load.insert(optional_buffers.begin(), optional_buffers.end());
 
   CHECK(!buffers_to_load.empty());
 
   std::set<ForeignStorageMgr::ParallelismHint> col_frag_hints;
   for (const auto& [chunk_key, buffer] : buffers_to_load) {
     CHECK_EQ(buffer->size(), static_cast<size_t>(0));
     col_frag_hints.emplace(
         schema_->getLogicalColumn(chunk_key[CHUNK_KEY_COLUMN_IDX])->columnId,
         chunk_key[CHUNK_KEY_FRAGMENT_IDX]);
   }
 
   auto hints_per_thread = partition_for_threads(col_frag_hints, g_max_import_threads);
 
   std::vector<std::future<void>> futures;
   for (const auto& hint_set : hints_per_thread) {
     futures.emplace_back(std::async(std::launch::async, [&, hint_set, this] {
       for (const auto& [col_id, frag_id] : hint_set) {
         loadBuffersUsingLazyParquetChunkLoader(col_id, frag_id, buffers_to_load);
       }
     }));
   }
 
   for (auto& future : futures) {
     future.get();
   }
 }

Here is the call graph for this function:

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

overridevirtual

Populates given chunk metadata vector with metadata for all chunks in related foreign table.

Parameters

chunk_metadata_vector - vector that will be populated with chunk metadata

Implements foreign_storage::ForeignDataWrapper.

Definition at line 350 of file ParquetDataWrapper.cpp.

References chunk_metadata_map_, and fetchChunkMetadata().

                                                 {
   fetchChunkMetadata();
   for (const auto& [chunk_key, chunk_metadata] : chunk_metadata_map_) {
     chunk_metadata_vector.emplace_back(chunk_key, chunk_metadata);
   }
 }

Here is the call graph for this function:

void foreign_storage::ParquetDataWrapper::resetParquetMetadata ( )

private

Definition at line 92 of file ParquetDataWrapper.cpp.

References file_reader_cache_, fragment_to_row_group_interval_map_, last_fragment_index_, last_fragment_row_count_, last_row_group_, and total_row_count_.

Referenced by fetchChunkMetadata().

                                               {
   fragment_to_row_group_interval_map_.clear();
   fragment_to_row_group_interval_map_[0] = {};
 
   last_row_group_ = 0;
   last_fragment_index_ = 0;
   last_fragment_row_count_ = 0;
   total_row_count_ = 0;
   file_reader_cache_->clear();
 }

Here is the caller graph for this function:

void foreign_storage::ParquetDataWrapper::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 525 of file ParquetDataWrapper.cpp.

References CHECK, chunk_metadata_map_, fragment_to_row_group_interval_map_, foreign_storage::json_utils::get_value_from_object(), is_restored_, last_fragment_index_, last_fragment_row_count_, last_row_group_, foreign_storage::json_utils::read_from_file(), and total_row_count_.

                                                       {
   auto d = json_utils::read_from_file(file_path);
   CHECK(d.IsObject());
 
   json_utils::get_value_from_object(
       d, fragment_to_row_group_interval_map_, "fragment_to_row_group_interval_map");
   json_utils::get_value_from_object(d, last_row_group_, "last_row_group");
   json_utils::get_value_from_object(d, last_fragment_index_, "last_fragment_index");
   json_utils::get_value_from_object(
       d, last_fragment_row_count_, "last_fragment_row_count");
   json_utils::get_value_from_object(d, total_row_count_, "total_row_count");
 
   CHECK(chunk_metadata_map_.empty());
   for (const auto& [chunk_key, chunk_metadata] : chunk_metadata_vector) {
     chunk_metadata_map_[chunk_key] = chunk_metadata;
   }
   is_restored_ = true;
 }

Here is the call graph for this function:

Member Data Documentation

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

private

Definition at line 89 of file ParquetDataWrapper.h.

Referenced by fetchChunkMetadata(), initializeChunkBuffers(), loadBuffersUsingLazyParquetChunkLoader(), metadataScanFiles(), populateChunkMetadata(), and restoreDataWrapperInternals().

const int foreign_storage::ParquetDataWrapper::db_id_

private

Definition at line 90 of file ParquetDataWrapper.h.

Referenced by fetchChunkMetadata(), getColumnsToInitialize(), initializeChunkBuffers(), loadBuffersUsingLazyParquetChunkLoader(), and metadataScanFiles().

std::unique_ptr<FileReaderMap> foreign_storage::ParquetDataWrapper::file_reader_cache_

private

Definition at line 99 of file ParquetDataWrapper.h.

Referenced by fetchChunkMetadata(), loadBuffersUsingLazyParquetChunkLoader(), metadataScanFiles(), and resetParquetMetadata().

std::shared_ptr<arrow::fs::FileSystem> foreign_storage::ParquetDataWrapper::file_system_

private

Definition at line 98 of file ParquetDataWrapper.h.

Referenced by fetchChunkMetadata(), loadBuffersUsingLazyParquetChunkLoader(), metadataScanFiles(), and ParquetDataWrapper().

const ForeignTable* foreign_storage::ParquetDataWrapper::foreign_table_

private

Definition at line 91 of file ParquetDataWrapper.h.

Referenced by fetchChunkMetadata(), getAllFilePaths(), initializeChunkBuffers(), loadBuffersUsingLazyParquetChunkLoader(), metadataScanFiles(), and moveToNextFragment().

std::map<int, std::vector<RowGroupInterval> > foreign_storage::ParquetDataWrapper::fragment_to_row_group_interval_map_

private

Definition at line 88 of file ParquetDataWrapper.h.

Referenced by addNewFile(), addNewFragment(), finalizeFragmentMap(), getProcessedFilePaths(), getSerializedDataWrapper(), isNewFile(), loadBuffersUsingLazyParquetChunkLoader(), resetParquetMetadata(), and restoreDataWrapperInternals().

bool foreign_storage::ParquetDataWrapper::is_restored_

private

Definition at line 96 of file ParquetDataWrapper.h.

Referenced by isRestored(), and restoreDataWrapperInternals().

int foreign_storage::ParquetDataWrapper::last_fragment_index_

private

Definition at line 92 of file ParquetDataWrapper.h.

Referenced by addNewFile(), addNewFragment(), finalizeFragmentMap(), getSerializedDataWrapper(), isNewFile(), metadataScanFiles(), resetParquetMetadata(), and restoreDataWrapperInternals().

size_t foreign_storage::ParquetDataWrapper::last_fragment_row_count_

private

Definition at line 93 of file ParquetDataWrapper.h.

Referenced by addNewFragment(), getSerializedDataWrapper(), metadataScanFiles(), moveToNextFragment(), resetParquetMetadata(), and restoreDataWrapperInternals().

int foreign_storage::ParquetDataWrapper::last_row_group_

private

Definition at line 95 of file ParquetDataWrapper.h.

Referenced by addNewFile(), addNewFragment(), finalizeFragmentMap(), getSerializedDataWrapper(), metadataScanFiles(), resetParquetMetadata(), and restoreDataWrapperInternals().

std::unique_ptr<ForeignTableSchema> foreign_storage::ParquetDataWrapper::schema_

private

Definition at line 97 of file ParquetDataWrapper.h.

Referenced by getColumnsToInitialize(), loadBuffersUsingLazyParquetChunkLoader(), metadataScanFiles(), and populateChunkBuffers().

size_t foreign_storage::ParquetDataWrapper::total_row_count_

private

Definition at line 94 of file ParquetDataWrapper.h.

Referenced by fetchChunkMetadata(), getSerializedDataWrapper(), metadataScanFiles(), resetParquetMetadata(), and restoreDataWrapperInternals().

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

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

Public Member Functions

Private Member Functions

Private Attributes

Additional Inherited Members

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation