#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, const bool do_metadata_stats_validation=true)

	ParquetDataWrapper (const ForeignTable *foreign_table, std::shared_ptr< arrow::fs::FileSystem > file_system)
	Constructor intended for detect use-case only. More...

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_vector) override

bool	isRestored () const override

ParallelismLevel	getCachedParallelismLevel () const override

ParallelismLevel	getNonCachedParallelismLevel () const override

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

DataPreview	getDataPreview (const size_t num_rows)

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

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, AbstractBuffer *delete_buffer)

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
const bool	do_metadata_stats_validation_

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_

std::mutex	delete_buffer_mutex_

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

Constructor & Destructor Documentation

foreign_storage::ParquetDataWrapper::ParquetDataWrapper ( )

Definition at line 76 of file ParquetDataWrapper.cpp.

77 : do_metadata_stats_validation_(true), db_id_(-1), foreign_table_(nullptr) {}

foreign_storage::ParquetDataWrapper::do_metadata_stats_validation_

const bool do_metadata_stats_validation_

Definition: ParquetDataWrapper.h:107

foreign_storage::ParquetDataWrapper::foreign_table_

const ForeignTable * foreign_table_

Definition: ParquetDataWrapper.h:111

foreign_storage::ParquetDataWrapper::db_id_

const int db_id_

Definition: ParquetDataWrapper.h:110

foreign_storage::ParquetDataWrapper::ParquetDataWrapper	(	const int	db_id,
		const ForeignTable *	foreign_table,
		const bool	do_metadata_stats_validation = `true`
	)

Definition at line 92 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.

     : do_metadata_stats_validation_(do_metadata_stats_validation)
     , 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();
   }
 }

foreign_storage::ParquetDataWrapper::ParquetDataWrapper	(	const ForeignTable *	foreign_table,
		std::shared_ptr< arrow::fs::FileSystem >	file_system
	)

Constructor intended for detect use-case only.

Definition at line 79 of file ParquetDataWrapper.cpp.

     : do_metadata_stats_validation_(false)
     , db_id_(-1)
     , foreign_table_(foreign_table)
     , last_fragment_index_(0)
     , last_fragment_row_count_(0)
     , total_row_count_(0)
     , last_row_group_(0)
     , is_restored_(false)
     , file_system_(file_system)
     , file_reader_cache_(std::make_unique<FileReaderMap>()) {}

Member Function Documentation

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

private

Definition at line 226 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 200 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::createRenderGroupAnalyzers ( )

overridevirtual

Create RenderGroupAnalyzers for poly columns.

Reimplemented from foreign_storage::ForeignDataWrapper.

Definition at line 623 of file ParquetDataWrapper.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:

void foreign_storage::ParquetDataWrapper::fetchChunkMetadata ( )

private

Definition at line 240 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_in_file_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_in_file_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 195 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 302 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(), and getDataPreview().

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

References foreign_storage::ForeignDataWrapper::INTER_FRAGMENT.

68 { return INTER_FRAGMENT; }

foreign_storage::ForeignDataWrapper::INTER_FRAGMENT

Definition: ForeignDataWrapper.h:142

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

private

Definition at line 124 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:

DataPreview foreign_storage::ParquetDataWrapper::getDataPreview ( const size_t num_rows )

Definition at line 609 of file ParquetDataWrapper.cpp.

References file_reader_cache_, file_system_, foreign_table_, getAllFilePaths(), foreign_storage::AbstractFileStorageDataWrapper::getFullFilePath(), and render_group_analyzer_map_.

                                                                     {
   LazyParquetChunkLoader chunk_loader(
       file_system_, file_reader_cache_.get(), &render_group_analyzer_map_);
   auto file_paths = getAllFilePaths();
   if (file_paths.empty()) {
     throw ForeignStorageException{"No file found at \"" +
                                   getFullFilePath(foreign_table_) + "\""};
   }
   return chunk_loader.previewFiles(file_paths, num_rows);
 }

Here is the call 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 70 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 292 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 566 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 141 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, false};
     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) ||
           column->columnType.is_geometry()) {
         // non-dictionary string or geometry WKT string
         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 212 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 605 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,
		AbstractBuffer *	delete_buffer
	)

private

Definition at line 382 of file ParquetDataWrapper.cpp.

References Data_Namespace::AbstractBuffer::append(), CHECK, CHECK_GT, chunk_metadata_map_, ColumnDescriptor::columnType, db_id_, delete_buffer_mutex_, foreign_storage::Interval< T >::end, file_reader_cache_, file_system_, foreign_table_, fragment_to_row_group_interval_map_, SQLTypeInfo::get_comp_param(), SQLTypeInfo::get_elem_type(), shared::get_from_map(), SQLTypeInfo::get_physical_cols(), Catalog_Namespace::SysCatalog::getCatalog(), Data_Namespace::AbstractBuffer::getMemoryPtr(), initializeChunkBuffers(), Catalog_Namespace::SysCatalog::instance(), SQLTypeInfo::is_array(), SQLTypeInfo::is_dict_encoded_string(), SQLTypeInfo::is_geometry(), foreign_storage::LazyParquetChunkLoader::loadChunk(), render_group_analyzer_map_, schema_, Data_Namespace::AbstractBuffer::size(), 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, false};
     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);
   }
 
   std::unique_ptr<RejectedRowIndices> rejected_row_indices;
   if (delete_buffer) {
     rejected_row_indices = std::make_unique<RejectedRowIndices>();
   }
   LazyParquetChunkLoader chunk_loader(
       file_system_, file_reader_cache_.get(), &render_group_analyzer_map_);
   auto metadata = chunk_loader.loadChunk(row_group_intervals,
                                          parquet_column_index,
                                          chunks,
                                          string_dictionary,
                                          rejected_row_indices.get());
 
   if (delete_buffer) {
     // all modifying operations on `delete_buffer` must be synchronized as it is a
     // shared buffer
     std::unique_lock<std::mutex> delete_buffer_lock(delete_buffer_mutex_);
 
     CHECK(!chunks.empty());
     CHECK(chunks.begin()->getBuffer()->hasEncoder());
     auto num_rows_in_chunk = chunks.begin()->getBuffer()->getEncoder()->getNumElems();
 
     // ensure delete buffer is sized appropriately
     if (delete_buffer->size() < num_rows_in_chunk) {
       auto remaining_rows = num_rows_in_chunk - delete_buffer->size();
       std::vector<int8_t> data(remaining_rows, false);
       delete_buffer->append(data.data(), remaining_rows);
     }
 
     // compute a logical OR with current `delete_buffer` contents and this chunks
     // rejected indices
     CHECK(rejected_row_indices);
     auto delete_buffer_data = delete_buffer->getMemoryPtr();
     for (const auto& rejected_index : *rejected_row_indices) {
       CHECK_GT(delete_buffer->size(), static_cast<size_t>(rejected_index));
       delete_buffer_data[rejected_index] = true;
     }
   }
 
   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 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);
     }
   }
 }

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 320 of file ParquetDataWrapper.cpp.

References addNewFile(), addNewFragment(), CHECK, CHECK_EQ, chunk_metadata_map_, db_id_, do_metadata_stats_validation_, 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(), nullptr);
   auto row_group_metadata =
       chunk_loader.metadataScan(file_paths, *schema_, do_metadata_stats_validation_);
   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 369 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,
		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 513 of file ParquetDataWrapper.cpp.

References CHECK, CHECK_EQ, CHUNK_KEY_COLUMN_IDX, CHUNK_KEY_FRAGMENT_IDX, foreign_storage::create_futures_for_workers(), g_max_import_threads, loadBuffersUsingLazyParquetChunkLoader(), 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]);
   }
 
   std::function<void(const std::set<ForeignStorageMgr::ParallelismHint>&)> lambda =
       [&, this](const std::set<ForeignStorageMgr::ParallelismHint>& hint_set) {
         for (const auto& [col_id, frag_id] : hint_set) {
           loadBuffersUsingLazyParquetChunkLoader(
               col_id, frag_id, buffers_to_load, delete_buffer);
         }
       };
   auto futures = create_futures_for_workers(col_frag_hints, g_max_import_threads, lambda);
 
   // We wait on all futures, then call get because we want all threads to have finished
   // before we propagate a potential exception.
   for (auto& future : futures) {
     future.wait();
   }
 
   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 374 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 113 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 584 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 109 of file ParquetDataWrapper.h.

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

const int foreign_storage::ParquetDataWrapper::db_id_

private

Definition at line 110 of file ParquetDataWrapper.h.

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

std::mutex foreign_storage::ParquetDataWrapper::delete_buffer_mutex_

private

Definition at line 121 of file ParquetDataWrapper.h.

Referenced by loadBuffersUsingLazyParquetChunkLoader().

const bool foreign_storage::ParquetDataWrapper::do_metadata_stats_validation_

private

Definition at line 107 of file ParquetDataWrapper.h.

Referenced by metadataScanFiles().

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

private

Definition at line 119 of file ParquetDataWrapper.h.

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

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

private

Definition at line 118 of file ParquetDataWrapper.h.

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

const ForeignTable* foreign_storage::ParquetDataWrapper::foreign_table_

private

Definition at line 111 of file ParquetDataWrapper.h.

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

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

private

Definition at line 108 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 116 of file ParquetDataWrapper.h.

Referenced by isRestored(), and restoreDataWrapperInternals().

int foreign_storage::ParquetDataWrapper::last_fragment_index_

private

Definition at line 112 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 113 of file ParquetDataWrapper.h.

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

int foreign_storage::ParquetDataWrapper::last_row_group_

private

Definition at line 115 of file ParquetDataWrapper.h.

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

RenderGroupAnalyzerMap foreign_storage::ParquetDataWrapper::render_group_analyzer_map_

private

Definition at line 126 of file ParquetDataWrapper.h.

Referenced by createRenderGroupAnalyzers(), getDataPreview(), and loadBuffersUsingLazyParquetChunkLoader().

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

private

Definition at line 117 of file ParquetDataWrapper.h.

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

size_t foreign_storage::ParquetDataWrapper::total_row_count_

private

Definition at line 114 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