#include <ParquetDataWrapper.h>

Inheritance diagram for foreign_storage::ParquetDataWrapper:

Collaboration diagram for foreign_storage::ParquetDataWrapper:

Public Member Functions
	ParquetDataWrapper (const int db_id, const ForeignTable *foreign_table)

void	populateChunkMetadata (ChunkMetadataVector &chunk_metadata_vector) override

void	populateChunkBuffers (std::map< ChunkKey, AbstractBuffer * > &required_buffers, std::map< ChunkKey, AbstractBuffer * > &optional_buffers) override

void	serializeDataWrapperInternals (const std::string &file_path) const override

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

bool	isRestored () const override

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

virtual	~ForeignDataWrapper ()=default

Static Public Member Functions
static void	validateOptions (const ForeignTable *foreign_table)

static std::vector < std::string_view >	getSupportedOptions ()

Private Member Functions
	ParquetDataWrapper (const ForeignTable *foreign_table)

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

void	initializeChunkBuffers (const int fragment_index, const Interval< ColumnType > &column_interval, std::map< ChunkKey, AbstractBuffer * > &required_buffers, const bool reserve_buffers_and_set_stats=false)

void	fetchChunkMetadata ()

void	loadBuffersUsingLazyParquetChunkLoader (const int logical_column_id, const int fragment_id, std::map< ChunkKey, AbstractBuffer * > &required_buffers)

void	validateFilePath () const

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

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

import_export::CopyParams	validateAndGetCopyParams () const

std::string	validateAndGetStringWithLength (const std::string &option_name, const size_t expected_num_chars) const

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::set< 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_

Static Private Attributes
static constexpr std::array < char const *, 0 >	supported_options_ {}

Detailed Description

Definition at line 33 of file ParquetDataWrapper.h.

Constructor & Destructor Documentation

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

Definition at line 79 of file ParquetDataWrapper.cpp.

References file_system_, foreign_storage::ForeignTable::foreign_server, foreign_storage::ForeignServer::LOCAL_FILE_STORAGE_TYPE, foreign_storage::OptionsContainer::options, foreign_storage::ForeignServer::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)) {
   auto& server_options = foreign_table->foreign_server->options;
   if (server_options.find(ForeignServer::STORAGE_TYPE_KEY)->second ==
       ForeignServer::LOCAL_FILE_STORAGE_TYPE) {
     file_system_ = std::make_shared<arrow::fs::LocalFileSystem>();
   } else {
     UNREACHABLE();
   }
 }

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

private

Definition at line 97 of file ParquetDataWrapper.cpp.

98 : db_id_(-1), foreign_table_(foreign_table) {}

foreign_storage::ParquetDataWrapper::foreign_table_

const ForeignTable * foreign_table_

Definition: ParquetDataWrapper.h:104

foreign_storage::ParquetDataWrapper::db_id_

const int db_id_

Definition: ParquetDataWrapper.h:103

Member Function Documentation

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

private

Definition at line 239 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 213 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 253 of file ParquetDataWrapper.cpp.

References CHECK_EQ, Catalog_Namespace::SysCatalog::checkedGetCatalog(), chunk_metadata_map_, db_id_, file_system_, foreign_table_, getAllFilePaths(), getProcessedFilePaths(), Catalog_Namespace::SysCatalog::instance(), foreign_storage::ForeignTable::isAppendMode(), metadataScanFiles(), foreign_storage::open_parquet_table(), 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().checkedGetCatalog(db_id_);
   std::set<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 (all_file_paths.find(file_path) == all_file_paths.end()) {
         throw_removed_file_error(file_path);
       }
     }
 
     for (const auto& file_path : all_file_paths) {
       if (processed_file_paths.find(file_path) == processed_file_paths.end()) {
         new_file_paths.emplace(file_path);
       }
     }
 
     // Single file append
     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);
 
       std::unique_ptr<parquet::arrow::FileReader> reader;
       open_parquet_table(file_path, reader, 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 {
     new_file_paths = getAllFilePaths();
     chunk_metadata_map_.clear();
     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 208 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::set< std::string > foreign_storage::ParquetDataWrapper::getAllFilePaths ( )

private

Definition at line 310 of file ParquetDataWrapper.cpp.

References DEBUG_TIMER, file_system_, foreign_table_, and foreign_storage::ForeignTable::getFullFilePath().

Referenced by fetchChunkMetadata().

                                                       {
   auto timer = DEBUG_TIMER(__func__);
   std::set<std::string> file_paths;
   arrow::fs::FileSelector file_selector{};
   std::string base_path = foreign_table_->getFullFilePath();
   file_selector.base_dir = base_path;
   file_selector.recursive = true;
 
   auto file_info_result = file_system_->GetFileInfo(file_selector);
   if (!file_info_result.ok()) {
     // This is expected when `base_path` points to a single file.
     file_paths.emplace(base_path);
   } else {
     auto& file_info_vector = file_info_result.ValueOrDie();
     for (const auto& file_info : file_info_vector) {
       if (file_info.type() == arrow::fs::FileType::File) {
         file_paths.emplace(file_info.path());
       }
     }
     if (file_paths.empty()) {
       throw std::runtime_error{"No file found at given path \"" + base_path + "\"."};
     }
   }
   return file_paths;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

private

Definition at line 139 of file ParquetDataWrapper.cpp.

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

Referenced by initializeChunkBuffers().

                                                  {
   const auto catalog =
       Catalog_Namespace::SysCatalog::instance().checkedGetCatalog(db_id_);
   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:

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

private

Definition at line 300 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::vector< std::string_view > foreign_storage::ParquetDataWrapper::getSupportedOptions ( )

static

Definition at line 115 of file ParquetDataWrapper.cpp.

References supported_options_.

                                                                   {
   return std::vector<std::string_view>{supported_options_.begin(),
                                        supported_options_.end()};
 }

void foreign_storage::ParquetDataWrapper::initializeChunkBuffers	(	const int	fragment_index,
		const Interval< ColumnType > &	column_interval,
		std::map< ChunkKey, AbstractBuffer * > &	required_buffers,
		const bool	reserve_buffers_and_set_stats = `false`
	)

private

Definition at line 156 of file ParquetDataWrapper.cpp.

References CHECK, chunk_metadata_map_, db_id_, foreign_table_, 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};
       auto data_buffer = required_buffers[data_chunk_key];
       CHECK(data_buffer);
       chunk.setBuffer(data_buffer);
 
       ChunkKey index_chunk_key{
           db_id_, foreign_table_->tableId, column->columnId, fragment_index, 2};
       auto index_buffer = required_buffers[index_chunk_key];
       CHECK(index_buffer);
       chunk.setIndexBuffer(index_buffer);
     } else {
       data_chunk_key = {
           db_id_, foreign_table_->tableId, column->columnId, fragment_index};
       auto data_buffer = required_buffers[data_chunk_key];
       CHECK(data_buffer);
       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 225 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 584 of file ParquetDataWrapper.cpp.

References is_restored_.

                                           {
   return is_restored_;
 }

void foreign_storage::ParquetDataWrapper::loadBuffersUsingLazyParquetChunkLoader	(	const int	logical_column_id,
		const int	fragment_id,
		std::map< ChunkKey, AbstractBuffer * > &	required_buffers
	)

private

Definition at line 421 of file ParquetDataWrapper.cpp.

References CHECK, Catalog_Namespace::SysCatalog::checkedGetCatalog(), chunk_metadata_map_, ColumnDescriptor::columnType, db_id_, foreign_storage::Interval< T >::end, file_system_, foreign_table_, fragment_to_row_group_interval_map_, TableDescriptor::fragmenter, SQLTypeInfo::get_comp_param(), SQLTypeInfo::get_elem_type(), SQLTypeInfo::get_physical_cols(), 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().checkedGetCatalog(db_id_);
   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 = 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->getMetadataForDictUnlocked(
         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 = required_buffers[data_chunk_key];
       CHECK(buffer);
       chunk.setBuffer(buffer);
       ChunkKey index_chunk_key = {
           db_id_, foreign_table_->tableId, column_id, fragment_id, 2};
       auto index_buffer = required_buffers[index_chunk_key];
       CHECK(index_buffer);
       chunk.setIndexBuffer(index_buffer);
     } else {
       ChunkKey chunk_key = {db_id_, foreign_table_->tableId, column_id, fragment_id};
       auto buffer = required_buffers[chunk_key];
       CHECK(buffer);
       chunk.setBuffer(buffer);
     }
     chunks.emplace_back(chunk);
   }
 
   LazyParquetChunkLoader chunk_loader(file_system_);
   auto metadata = chunk_loader.loadChunk(
       row_group_intervals, parquet_column_index, chunks, string_dictionary);
   auto fragmenter = foreign_table_->fragmenter;
   if (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());
       auto cached_metadata = chunk_metadata_map_[data_chunk_key];
       auto updated_metadata = std::make_shared<ChunkMetadata>();
       *updated_metadata = *cached_metadata;
       // for certain types, update the metadata statistics
       if (is_dictionary_encoded_string_column ||
           logical_column->columnType.is_geometry()) {
         CHECK(metadata_iter != metadata.end());
         auto& chunk_metadata_ptr = *metadata_iter;
         updated_metadata->chunkStats.max = chunk_metadata_ptr->chunkStats.max;
         updated_metadata->chunkStats.min = chunk_metadata_ptr->chunkStats.min;
       }
       CHECK(required_buffers.find(data_chunk_key) != required_buffers.end());
       updated_metadata->numBytes = required_buffers[data_chunk_key]->size();
       fragmenter->updateColumnChunkMetadata(column, fragment_id, updated_metadata);
     }
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

private

Definition at line 360 of file ParquetDataWrapper.cpp.

References addNewFile(), addNewFragment(), CHECK, CHECK_EQ, chunk_metadata_map_, db_id_, 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_);
   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 408 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	(	std::map< ChunkKey, AbstractBuffer * > &	required_buffers,
		std::map< ChunkKey, AbstractBuffer * > &	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 507 of file ParquetDataWrapper.cpp.

References CHECK, CHECK_EQ, CHUNK_KEY_COLUMN_IDX, CHUNK_KEY_FRAGMENT_IDX, loadBuffersUsingLazyParquetChunkLoader(), and schema_.

                                                          {
   CHECK(!required_buffers.empty());
   auto fragment_id = required_buffers.begin()->first[CHUNK_KEY_FRAGMENT_IDX];
 
   std::set<int> logical_column_ids;
   for (const auto& [chunk_key, buffer] : required_buffers) {
     CHECK_EQ(fragment_id, chunk_key[CHUNK_KEY_FRAGMENT_IDX]);
     CHECK_EQ(buffer->size(), static_cast<size_t>(0));
     const auto column_id =
         schema_->getLogicalColumn(chunk_key[CHUNK_KEY_COLUMN_IDX])->columnId;
     logical_column_ids.emplace(column_id);
   }
 
   for (const auto column_id : logical_column_ids) {
     loadBuffersUsingLazyParquetChunkLoader(column_id, fragment_id, required_buffers);
   }
 }

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

References 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;
 }

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 563 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:

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

overridevirtual

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

Parameters

file_path - location to save file to

Implements foreign_storage::ForeignDataWrapper.

Definition at line 543 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_file().

                                       {
   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());
 
   json_utils::write_to_file(d, file_path);
 }

Here is the call graph for this function:

import_export::CopyParams foreign_storage::ParquetDataWrapper::validateAndGetCopyParams ( ) const

private

Definition at line 336 of file ParquetDataWrapper.cpp.

References import_export::CopyParams::file_type.

                                                                            {
   import_export::CopyParams copy_params{};
   // The file_type argument is never utilized in the context of FSI,
   // for completeness, set the file_type
   copy_params.file_type = import_export::FileType::PARQUET;
   return copy_params;
 }

std::string foreign_storage::ParquetDataWrapper::validateAndGetStringWithLength	(	const std::string &	option_name,
		const size_t	expected_num_chars
	)		const

private

Validates that the value of given table option has the expected number of characters. An exception is thrown if the number of characters do not match.

Parameters

option_name	- name of table option whose value is validated and returned
expected_num_chars	- expected number of characters for option value

Returns: value of the option if the number of characters match. Returns an empty string if table options do not contain provided option.

Definition at line 344 of file ParquetDataWrapper.cpp.

References foreign_table_, foreign_storage::OptionsContainer::options, and to_string().

                                            {
   if (auto it = foreign_table_->options.find(option_name);
       it != foreign_table_->options.end()) {
     if (it->second.length() != expected_num_chars) {
       throw std::runtime_error{"Value of \"" + option_name +
                                "\" foreign table option has the wrong number of "
                                "characters. Expected " +
                                std::to_string(expected_num_chars) + " character(s)."};
     }
     return it->second;
   }
   return "";
 }

Here is the call graph for this function:

void foreign_storage::ParquetDataWrapper::validateFilePath ( ) const

private

Definition at line 120 of file ParquetDataWrapper.cpp.

References foreign_storage::ForeignTable::foreign_server, foreign_table_, foreign_storage::ForeignTable::getFullFilePath(), ddl_utils::IMPORT, foreign_storage::ForeignServer::LOCAL_FILE_STORAGE_TYPE, foreign_storage::OptionsContainer::options, foreign_storage::ForeignServer::STORAGE_TYPE_KEY, and ddl_utils::validate_allowed_file_path().

                                                 {
   auto& server_options = foreign_table_->foreign_server->options;
   if (server_options.find(ForeignServer::STORAGE_TYPE_KEY)->second ==
       ForeignServer::LOCAL_FILE_STORAGE_TYPE) {
     ddl_utils::validate_allowed_file_path(foreign_table_->getFullFilePath(),
                                           ddl_utils::DataTransferType::IMPORT);
   }
 }

Here is the call graph for this function:

void foreign_storage::ParquetDataWrapper::validateOptions ( const ForeignTable * foreign_table )

static

Definition at line 100 of file ParquetDataWrapper.cpp.

References foreign_storage::OptionsContainer::options, foreign_storage::ForeignTable::supported_options, and supported_options_.

                                                                           {
   for (const auto& entry : foreign_table->options) {
     const auto& table_options = foreign_table->supported_options;
     if (std::find(table_options.begin(), table_options.end(), entry.first) ==
             table_options.end() &&
         std::find(supported_options_.begin(), supported_options_.end(), entry.first) ==
             supported_options_.end()) {
       throw std::runtime_error{"Invalid foreign table option \"" + entry.first + "\"."};
     }
   }
   ParquetDataWrapper data_wrapper{foreign_table};
   data_wrapper.validateAndGetCopyParams();
   data_wrapper.validateFilePath();
 }

Member Data Documentation

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

private

Definition at line 102 of file ParquetDataWrapper.h.

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

const int foreign_storage::ParquetDataWrapper::db_id_

private

Definition at line 103 of file ParquetDataWrapper.h.

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

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

private

Definition at line 111 of file ParquetDataWrapper.h.

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

const ForeignTable* foreign_storage::ParquetDataWrapper::foreign_table_

private

Definition at line 104 of file ParquetDataWrapper.h.

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

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

private

Definition at line 101 of file ParquetDataWrapper.h.

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

bool foreign_storage::ParquetDataWrapper::is_restored_

private

Definition at line 109 of file ParquetDataWrapper.h.

Referenced by isRestored(), and restoreDataWrapperInternals().

int foreign_storage::ParquetDataWrapper::last_fragment_index_

private

Definition at line 105 of file ParquetDataWrapper.h.

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

size_t foreign_storage::ParquetDataWrapper::last_fragment_row_count_

private

Definition at line 106 of file ParquetDataWrapper.h.

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

int foreign_storage::ParquetDataWrapper::last_row_group_

private

Definition at line 108 of file ParquetDataWrapper.h.

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

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

private

Definition at line 110 of file ParquetDataWrapper.h.

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

constexpr std::array<char const*, 0> foreign_storage::ParquetDataWrapper::supported_options_ {}

staticprivate

Definition at line 113 of file ParquetDataWrapper.h.

Referenced by getSupportedOptions(), and validateOptions().

size_t foreign_storage::ParquetDataWrapper::total_row_count_

private

Definition at line 107 of file ParquetDataWrapper.h.

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

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

Static Public Member Functions

Private Member Functions

Private Attributes

Static Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation