#include <LazyParquetChunkLoader.h>

Collaboration diagram for foreign_storage::LazyParquetChunkLoader:

Public Member Functions
	LazyParquetChunkLoader (std::shared_ptr< arrow::fs::FileSystem > file_system, FileReaderMap *file_reader_cache)

std::list< std::unique_ptr < ChunkMetadata > >	loadChunk (const std::vector< RowGroupInterval > &row_group_intervals, const int parquet_column_index, std::list< Chunk_NS::Chunk > &chunks, StringDictionary *string_dictionary=nullptr)

std::list< RowGroupMetadata >	metadataScan (const std::set< std::string > &file_paths, const ForeignTableSchema &schema)
	Perform a metadata scan for the paths specified. More...

Static Public Member Functions
static bool	isColumnMappingSupported (const ColumnDescriptor omnisci_column, const parquet::ColumnDescriptor parquet_column)

Static Public Attributes
static const int	batch_reader_num_elements = 4096

Private Member Functions
std::list< std::unique_ptr < ChunkMetadata > >	appendRowGroups (const std::vector< RowGroupInterval > &row_group_intervals, const int parquet_column_index, const ColumnDescriptor column_descriptor, std::list< Chunk_NS::Chunk > &chunks, StringDictionary string_dictionary)

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

FileReaderMap *	file_reader_cache_

Detailed Description

A lazy parquet to chunk loader

Definition at line 35 of file LazyParquetChunkLoader.h.

Constructor & Destructor Documentation

foreign_storage::LazyParquetChunkLoader::LazyParquetChunkLoader	(	std::shared_ptr< arrow::fs::FileSystem >	file_system,
		FileReaderMap *	file_reader_cache
	)

Definition at line 1478 of file LazyParquetChunkLoader.cpp.

1481 : file_system_(file_system), file_reader_cache_(file_map) {}

foreign_storage::LazyParquetChunkLoader::file_reader_cache_

FileReaderMap * file_reader_cache_

Definition: LazyParquetChunkLoader.h:99

foreign_storage::LazyParquetChunkLoader::file_system_

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

Definition: LazyParquetChunkLoader.h:98

Member Function Documentation

std::list< std::unique_ptr< ChunkMetadata > > foreign_storage::LazyParquetChunkLoader::appendRowGroups	(	const std::vector< RowGroupInterval > &	row_group_intervals,
		const int	parquet_column_index,
		const ColumnDescriptor *	column_descriptor,
		std::list< Chunk_NS::Chunk > &	chunks,
		StringDictionary *	string_dictionary
	)

private

Definition at line 1359 of file LazyParquetChunkLoader.cpp.

References batch_reader_num_elements, CHECK, foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::create_parquet_encoder(), DEBUG_TIMER, file_reader_cache_, file_system_, foreign_storage::get_column_descriptor(), foreign_storage::get_parquet_table_size(), foreign_storage::FileReaderMap::getOrInsert(), foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::resize_values_buffer(), to_string(), foreign_storage::validate_equal_column_descriptor(), and foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::validate_max_repetition_and_definition_level().

Referenced by loadChunk().

                                          {
   auto timer = DEBUG_TIMER(__func__);
   std::list<std::unique_ptr<ChunkMetadata>> chunk_metadata;
   // `def_levels` and `rep_levels` below are used to store the read definition
   // and repetition levels of the Dremel encoding implemented by the Parquet
   // format
   std::vector<int16_t> def_levels(LazyParquetChunkLoader::batch_reader_num_elements);
   std::vector<int16_t> rep_levels(LazyParquetChunkLoader::batch_reader_num_elements);
   std::vector<int8_t> values;
 
   CHECK(!row_group_intervals.empty());
   const auto& first_file_path = row_group_intervals.front().file_path;
 
   auto first_file_reader = file_reader_cache_->getOrInsert(first_file_path, file_system_);
   auto first_parquet_column_descriptor =
       get_column_descriptor(first_file_reader, parquet_column_index);
   resize_values_buffer(column_descriptor, first_parquet_column_descriptor, values);
   auto encoder = create_parquet_encoder(column_descriptor,
                                         first_parquet_column_descriptor,
                                         chunks,
                                         string_dictionary,
                                         chunk_metadata);
   CHECK(encoder.get());
 
   for (const auto& row_group_interval : row_group_intervals) {
     const auto& file_path = row_group_interval.file_path;
     auto file_reader = file_reader_cache_->getOrInsert(file_path, file_system_);
 
     auto [num_row_groups, num_columns] = get_parquet_table_size(file_reader);
     CHECK(row_group_interval.start_index >= 0 &&
           row_group_interval.end_index < num_row_groups);
     CHECK(parquet_column_index >= 0 && parquet_column_index < num_columns);
 
     parquet::ParquetFileReader* parquet_reader = file_reader->parquet_reader();
     auto parquet_column_descriptor =
         get_column_descriptor(file_reader, parquet_column_index);
     validate_equal_column_descriptor(first_parquet_column_descriptor,
                                      parquet_column_descriptor,
                                      first_file_path,
                                      file_path);
 
     validate_max_repetition_and_definition_level(column_descriptor,
                                                  parquet_column_descriptor);
     int64_t values_read = 0;
     for (int row_group_index = row_group_interval.start_index;
          row_group_index <= row_group_interval.end_index;
          ++row_group_index) {
       auto group_reader = parquet_reader->RowGroup(row_group_index);
       std::shared_ptr<parquet::ColumnReader> col_reader =
           group_reader->Column(parquet_column_index);
 
       try {
         while (col_reader->HasNext()) {
           int64_t levels_read =
               parquet::ScanAllValues(LazyParquetChunkLoader::batch_reader_num_elements,
                                      def_levels.data(),
                                      rep_levels.data(),
                                      reinterpret_cast<uint8_t*>(values.data()),
                                      &values_read,
                                      col_reader.get());
           encoder->appendData(def_levels.data(),
                               rep_levels.data(),
                               values_read,
                               levels_read,
                               !col_reader->HasNext(),
                               values.data());
         }
       } catch (const std::exception& error) {
         throw ForeignStorageException(
             std::string(error.what()) + " Row group: " + std::to_string(row_group_index) +
             ", Parquet column: '" + col_reader->descr()->path()->ToDotString() +
             "', Parquet file: '" + file_path + "'");
       }
     }
   }
   return chunk_metadata;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

bool foreign_storage::LazyParquetChunkLoader::isColumnMappingSupported	(	const ColumnDescriptor *	omnisci_column,
		const parquet::ColumnDescriptor *	parquet_column
	)

static

Determine if a Parquet to OmniSci column mapping is supported.

Parameters

omnisci_column	- the column descriptor of the OmniSci column
parquet_column	- the column descriptor of the Parquet column

Returns: true if the column mapping is supported by LazyParquetChunkLoader, false otherwise

Definition at line 1442 of file LazyParquetChunkLoader.cpp.

Referenced by foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::validate_allowed_mapping(), and foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::validate_array_mapping().

                                                    {
   if (validate_geospatial_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_array_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_decimal_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_floating_point_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_integral_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_none_type_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_timestamp_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_time_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_date_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   if (validate_string_mapping(omnisci_column, parquet_column)) {
     return true;
   }
   return false;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::list< std::unique_ptr< ChunkMetadata > > foreign_storage::LazyParquetChunkLoader::loadChunk	(	const std::vector< RowGroupInterval > &	row_group_intervals,
		const int	parquet_column_index,
		std::list< Chunk_NS::Chunk > &	chunks,
		StringDictionary *	string_dictionary = `nullptr`
	)

Load a number of row groups of a column in a parquet file into a chunk

Parameters

row_group_interval	- an inclusive interval [start,end] that specifies row groups to load
parquet_column_index	- the logical column index in the parquet file (and omnisci db) of column to load
chunks	- a list containing the chunks to load
string_dictionary	- a string dictionary for the column corresponding to the column, if applicable

Returns: An empty list when no metadata update is applicable, otherwise a list of ChunkMetadata shared pointers with which to update the corresponding column chunk metadata. NOTE: Only ChunkMetadata.sqlType and the min & max values of the ChunkMetadata.chunkStats are valid, other values are not set.

NOTE: if more than one chunk is supplied, the first chunk is required to be the chunk corresponding to the logical column, while the remaining chunks correspond to physical columns (in ascending order of column id.) Similarly, if a metada update is expected, the list of ChunkMetadata shared pointers returned will correspond directly to the list chunks.

Definition at line 1483 of file LazyParquetChunkLoader.cpp.

References appendRowGroups(), and CHECK.

Referenced by foreign_storage::ParquetDataWrapper::loadBuffersUsingLazyParquetChunkLoader().

                                          {
   CHECK(!chunks.empty());
   auto const& chunk = *chunks.begin();
   auto column_descriptor = chunk.getColumnDesc();
   auto buffer = chunk.getBuffer();
   CHECK(buffer);
 
   try {
     auto metadata = appendRowGroups(row_group_intervals,
                                     parquet_column_index,
                                     column_descriptor,
                                     chunks,
                                     string_dictionary);
     return metadata;
   } catch (const std::exception& error) {
     throw ForeignStorageException(error.what());
   }
 
   return {};
 }

Here is the call graph for this function:

Here is the caller graph for this function:

std::list< RowGroupMetadata > foreign_storage::LazyParquetChunkLoader::metadataScan	(	const std::set< std::string > &	file_paths,
		const ForeignTableSchema &	schema
	)

Perform a metadata scan for the paths specified.

Parameters

file_paths	- (ordered) files of the metadata scan
schema	- schema of the foreign table to perform metadata scan for

Returns: a list of the row group metadata extracted from file_paths

Definition at line 1508 of file LazyParquetChunkLoader.cpp.

References CHECK, DEBUG_TIMER, file_reader_cache_, file_system_, g_max_import_threads, foreign_storage::get_parquet_table_size(), foreign_storage::ForeignTableSchema::getLogicalAndPhysicalColumns(), foreign_storage::FileReaderMap::initializeIfEmpty(), foreign_storage::FileReaderMap::insert(), foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::metadata_scan_rowgroup_interval(), foreign_storage::partition_for_threads(), foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::populate_encoder_map(), foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::validate_equal_schema(), and foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::validate_parquet_metadata().

Referenced by foreign_storage::ParquetDataWrapper::metadataScanFiles().

                                       {
   auto timer = DEBUG_TIMER(__func__);
   auto column_interval =
       Interval<ColumnType>{schema.getLogicalAndPhysicalColumns().front()->columnId,
                            schema.getLogicalAndPhysicalColumns().back()->columnId};
   CHECK(!file_paths.empty());
 
   // The encoder map needs to be populated before we can start scanning rowgroups, so we
   // peel the first file_path out of the async loop below to perform population.
   const auto& first_path = *file_paths.begin();
   auto first_reader = file_reader_cache_->insert(first_path, file_system_);
   validate_parquet_metadata(
       first_reader->parquet_reader()->metadata(), first_path, schema);
   auto encoder_map = populate_encoder_map(column_interval, schema, first_reader);
   const auto num_row_groups = get_parquet_table_size(first_reader).first;
   auto row_group_metadata = metadata_scan_rowgroup_interval(
       encoder_map, {first_path, 0, num_row_groups - 1}, first_reader, schema);
 
   // We want each (filepath->FileReader) pair in the cache to be initialized before we
   // multithread so that we are not adding keys in a concurrent environment, so we add
   // cache entries for each path and initialize to an empty unique_ptr if the file has not
   // yet been opened.
   // Since we have already performed the first iteration, we skip it in the thread groups
   // so as not to process it twice.
   std::set<std::string> cache_subset;
   for (auto path_it = ++(file_paths.begin()); path_it != file_paths.end(); ++path_it) {
     file_reader_cache_->initializeIfEmpty(*path_it);
     cache_subset.insert(*path_it);
   }
 
   // Iterate asyncronously over any paths beyond the first.
   auto paths_per_thread = partition_for_threads(cache_subset, g_max_import_threads);
   std::vector<std::future<std::list<RowGroupMetadata>>> futures;
   for (const auto& path_group : paths_per_thread) {
     futures.emplace_back(std::async(
         std::launch::async,
         [&](const auto& paths, const auto& file_reader_cache) {
           std::list<RowGroupMetadata> reduced_metadata;
           for (const auto& path : paths.get()) {
             auto reader = file_reader_cache.get().getOrInsert(path, file_system_);
             validate_equal_schema(first_reader, reader, first_path, path);
             validate_parquet_metadata(reader->parquet_reader()->metadata(), path, schema);
             const auto num_row_groups = get_parquet_table_size(reader).first;
             const auto interval = RowGroupInterval{path, 0, num_row_groups - 1};
             reduced_metadata.splice(
                 reduced_metadata.end(),
                 metadata_scan_rowgroup_interval(encoder_map, interval, reader, schema));
           }
           return reduced_metadata;
         },
         std::ref(path_group),
         std::ref(*file_reader_cache_)));
   }
 
   // Reduce all the row_group results.
   for (auto& future : futures) {
     row_group_metadata.splice(row_group_metadata.end(), future.get());
   }
   return row_group_metadata;
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

const int foreign_storage::LazyParquetChunkLoader::batch_reader_num_elements = 4096

static

Definition at line 40 of file LazyParquetChunkLoader.h.

Referenced by appendRowGroups(), and foreign_storage::anonymous_namespace{LazyParquetChunkLoader.cpp}::resize_values_buffer().

FileReaderMap* foreign_storage::LazyParquetChunkLoader::file_reader_cache_

private

Definition at line 99 of file LazyParquetChunkLoader.h.

Referenced by appendRowGroups(), and metadataScan().

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

private

Definition at line 98 of file LazyParquetChunkLoader.h.

Referenced by appendRowGroups(), and metadataScan().

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

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

Public Member Functions

Static Public Member Functions

Static Public Attributes

Private Member Functions

Private Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation