foreign_storage::AbstractTextFileDataWrapper Class Reference

#include <AbstractTextFileDataWrapper.h>

Inheritance diagram for foreign_storage::AbstractTextFileDataWrapper:

Collaboration diagram for foreign_storage::AbstractTextFileDataWrapper:

Public Member Functions
	AbstractTextFileDataWrapper ()
	AbstractTextFileDataWrapper (const int db_id, const ForeignTable *foreign_table)
void	populateChunkMetadata (ChunkMetadataVector &chunk_metadata_vector) override
void	populateChunkBuffers (const ChunkToBufferMap &required_buffers, const ChunkToBufferMap &optional_buffers) override
std::string	getSerializedDataWrapper () const override
void	restoreDataWrapperInternals (const std::string &file_path, const ChunkMetadataVector &chunk_metadata) override
bool	isRestored () const override
ParallelismLevel	getCachedParallelismLevel () const override
ParallelismLevel	getNonCachedParallelismLevel () const override
Public Member Functions inherited from foreign_storage::AbstractFileStorageDataWrapper
	AbstractFileStorageDataWrapper ()
void	validateServerOptions (const ForeignServer *foreign_server) const override
void	validateTableOptions (const ForeignTable *foreign_table) const override
const std::set < std::string_view > &	getSupportedTableOptions () const override
void	validateUserMappingOptions (const UserMapping *user_mapping, const ForeignServer *foreign_server) const override
const std::set < std::string_view > &	getSupportedUserMappingOptions () const override
Public Member Functions inherited from foreign_storage::ForeignDataWrapper
	ForeignDataWrapper ()=default
virtual	~ForeignDataWrapper ()=default
virtual void	validateSchema (const std::list< ColumnDescriptor > &columns) const

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

Private Member Functions
	AbstractTextFileDataWrapper (const ForeignTable *foreign_table)
void	populateChunks (std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map, int fragment_id)
void	populateChunkMapForColumns (const std::set< const ColumnDescriptor * > &columns, const int fragment_id, const ChunkToBufferMap &buffers, std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map)
void	updateMetadata (std::map< int, Chunk_NS::Chunk > &column_id_to_chunk_map, int fragment_id)

Private Attributes
std::map< ChunkKey, std::shared_ptr< ChunkMetadata > >	chunk_metadata_map_
std::map< int, FileRegions >	fragment_id_to_file_regions_map_
std::unique_ptr< FileReader >	file_reader_
const int	db_id_
const ForeignTable *	foreign_table_
std::map< ChunkKey, std::unique_ptr < ForeignStorageBuffer > >	chunk_encoder_buffers_
size_t	num_rows_
size_t	append_start_offset_
bool	is_restored_

Additional Inherited Members
Public Types inherited from foreign_storage::ForeignDataWrapper
enum	ParallelismLevel { NONE, INTRA_FRAGMENT, INTER_FRAGMENT }
Static Public Attributes inherited from foreign_storage::AbstractFileStorageDataWrapper
static const std::string	STORAGE_TYPE_KEY = "STORAGE_TYPE"
static const std::string	BASE_PATH_KEY = "BASE_PATH"
static const std::string	FILE_PATH_KEY = "FILE_PATH"
static const std::string	REGEX_PATH_FILTER_KEY = "REGEX_PATH_FILTER"
static const std::string	LOCAL_FILE_STORAGE_TYPE = "LOCAL_FILE"
static const std::string	S3_STORAGE_TYPE = "AWS_S3"
static const std::string	FILE_SORT_ORDER_BY_KEY = shared::FILE_SORT_ORDER_BY_KEY
static const std::string	FILE_SORT_REGEX_KEY = shared::FILE_SORT_REGEX_KEY
static const std::array < std::string, 1 >	supported_storage_types
Static Protected Member Functions inherited from foreign_storage::AbstractFileStorageDataWrapper
static std::string	getFullFilePath (const ForeignTable *foreign_table)
	Returns the path to the source file/dir of the table. Depending on options this may result from a concatenation of server and table path options. More...

Detailed Description

Definition at line 32 of file AbstractTextFileDataWrapper.h.

Constructor & Destructor Documentation

foreign_storage::AbstractTextFileDataWrapper::AbstractTextFileDataWrapper

(

)

Definition at line 34 of file AbstractTextFileDataWrapper.cpp.

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

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

Definition at line 37 of file AbstractTextFileDataWrapper.cpp.

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

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

private

Member Function Documentation

ParallelismLevel foreign_storage::AbstractTextFileDataWrapper::getCachedParallelismLevel ( ) const

inlineoverridevirtual

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

Reimplemented from foreign_storage::ForeignDataWrapper.

Definition at line 49 of file AbstractTextFileDataWrapper.h.

References foreign_storage::ForeignDataWrapper::INTRA_FRAGMENT.

{ return INTRA_FRAGMENT; }

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

protectedpure virtual

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

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

Here is the caller graph for this function:

ParallelismLevel foreign_storage::AbstractTextFileDataWrapper::getNonCachedParallelismLevel ( ) const

inlineoverridevirtual

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

Reimplemented from foreign_storage::ForeignDataWrapper.

Definition at line 51 of file AbstractTextFileDataWrapper.h.

References foreign_storage::ForeignDataWrapper::INTRA_FRAGMENT.

                                                                 {
    return INTRA_FRAGMENT;
  }

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

overridevirtual

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

Implements foreign_storage::ForeignDataWrapper.

Definition at line 932 of file AbstractTextFileDataWrapper.cpp.

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

                                                                      {
  rapidjson::Document d;
  d.SetObject();

  // Save fragment map
  json_utils::add_value_to_object(d,
                                  fragment_id_to_file_regions_map_,
                                  "fragment_id_to_file_regions_map",
                                  d.GetAllocator());

  // Save reader metadata
  rapidjson::Value reader_metadata(rapidjson::kObjectType);
  file_reader_->serialize(reader_metadata, d.GetAllocator());
  d.AddMember("reader_metadata", reader_metadata, d.GetAllocator());

  json_utils::add_value_to_object(d, num_rows_, "num_rows", d.GetAllocator());
  json_utils::add_value_to_object(
      d, append_start_offset_, "append_start_offset", d.GetAllocator());

  return json_utils::write_to_string(d);
}

Here is the call graph for this function:

bool foreign_storage::AbstractTextFileDataWrapper::isRestored ( ) const

overridevirtual

Implements foreign_storage::ForeignDataWrapper.

Definition at line 1001 of file AbstractTextFileDataWrapper.cpp.

References is_restored_.

                                                   {
  return is_restored_;
}

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

overridevirtual

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

Parameters

required_buffers	- chunk buffers that must always be populated
optional_buffers	- chunk buffers that can be optionally populated, if the data wrapper has to scan through chunk data anyways (typically for row wise data formats)

Implements foreign_storage::ForeignDataWrapper.

Definition at line 78 of file AbstractTextFileDataWrapper.cpp.

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

                                              {
  auto timer = DEBUG_TIMER(__func__);
  auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
  CHECK(catalog);
  CHECK(!required_buffers.empty());

  auto fragment_id = required_buffers.begin()->first[CHUNK_KEY_FRAGMENT_IDX];
  auto required_columns =
      get_columns(required_buffers, *catalog, foreign_table_->tableId, fragment_id);
  std::map<int, Chunk_NS::Chunk> column_id_to_chunk_map;
  populateChunkMapForColumns(
      required_columns, fragment_id, required_buffers, column_id_to_chunk_map);

  if (!optional_buffers.empty()) {
    auto optional_columns =
        get_columns(optional_buffers, *catalog, foreign_table_->tableId, fragment_id);
    populateChunkMapForColumns(
        optional_columns, fragment_id, optional_buffers, column_id_to_chunk_map);
  }
  populateChunks(column_id_to_chunk_map, fragment_id);
  updateMetadata(column_id_to_chunk_map, fragment_id);
  for (auto& entry : column_id_to_chunk_map) {
    entry.second.setBuffer(nullptr);
    entry.second.setIndexBuffer(nullptr);
  }
}

Here is the call graph for this function:

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

private

Definition at line 63 of file AbstractTextFileDataWrapper.cpp.

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

Referenced by populateChunkBuffers().

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

Here is the call graph for this function:

Here is the caller graph for this function:

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

overridevirtual

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

Parameters

chunk_metadata_vector

- vector to be populated with chunk metadata

Implements foreign_storage::ForeignDataWrapper.

Definition at line 784 of file AbstractTextFileDataWrapper.cpp.

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

                                                {
  auto timer = DEBUG_TIMER(__func__);

  const auto copy_params = getFileBufferParser().validateAndGetCopyParams(foreign_table_);
  const auto file_path = getFullFilePath(foreign_table_);
  auto catalog = Catalog_Namespace::SysCatalog::instance().getCatalog(db_id_);
  CHECK(catalog);
  auto& parser = getFileBufferParser();
  auto& server_options = foreign_table_->foreign_server->options;
  if (foreign_table_->isAppendMode() && file_reader_ != nullptr) {
    parser.validateFiles(file_reader_.get(), foreign_table_);
    if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
      file_reader_->checkForMoreRows(append_start_offset_);
    } else {
      UNREACHABLE();
    }
  } else {
    // Should only be called once for non-append tables
    CHECK(chunk_metadata_map_.empty());
    CHECK(fragment_id_to_file_regions_map_.empty());
    if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
      file_reader_ = std::make_unique<LocalMultiFileReader>(
          file_path,
          copy_params,
          foreign_table_->getOption(
              AbstractFileStorageDataWrapper::REGEX_PATH_FILTER_KEY),
          foreign_table_->getOption(
              AbstractFileStorageDataWrapper::FILE_SORT_ORDER_BY_KEY),
          foreign_table_->getOption(AbstractFileStorageDataWrapper::FILE_SORT_REGEX_KEY));
    } else {
      UNREACHABLE();
    }
    parser.validateFiles(file_reader_.get(), foreign_table_);
    num_rows_ = 0;
    append_start_offset_ = 0;
  }

  auto columns =
      catalog->getAllColumnMetadataForTable(foreign_table_->tableId, false, false, true);
  std::map<int32_t, const ColumnDescriptor*> column_by_id{};
  for (auto column : columns) {
    column_by_id[column->columnId] = column;
  }
  MetadataScanMultiThreadingParams multi_threading_params;

  // Restore previous chunk data
  if (foreign_table_->isAppendMode()) {
    multi_threading_params.chunk_encoder_buffers = std::move(chunk_encoder_buffers_);
  }

  std::set<int> columns_to_scan;
  for (auto column : columns) {
    if (!skip_metadata_scan(column)) {
      columns_to_scan.insert(column->columnId);
    }
  }

  // Track where scan started for appends
  int start_row = num_rows_;
  if (!file_reader_->isScanFinished()) {
    auto buffer_size = get_buffer_size(copy_params,
                                       file_reader_->isRemainingSizeKnown(),
                                       file_reader_->getRemainingSize());
    auto thread_count = get_thread_count(copy_params,
                                         file_reader_->isRemainingSizeKnown(),
                                         file_reader_->getRemainingSize(),
                                         buffer_size);
    multi_threading_params.continue_processing = true;

    std::vector<std::future<void>> futures{};
    for (size_t i = 0; i < thread_count; i++) {
      multi_threading_params.request_pool.emplace(buffer_size,
                                                  copy_params,
                                                  db_id_,
                                                  foreign_table_,
                                                  columns_to_scan,
                                                  getFullFilePath(foreign_table_));

      futures.emplace_back(std::async(std::launch::async,
                                      scan_metadata,
                                      std::ref(multi_threading_params),
                                      std::ref(fragment_id_to_file_regions_map_),
                                      std::ref(parser)));
    }

    try {
      dispatch_metadata_scan_requests(buffer_size,
                                      file_path,
                                      (*file_reader_),
                                      copy_params,
                                      multi_threading_params,
                                      num_rows_,
                                      append_start_offset_,
                                      getFileBufferParser());
    } catch (...) {
      {
        std::unique_lock<std::mutex> pending_requests_lock(
            multi_threading_params.pending_requests_mutex);
        multi_threading_params.continue_processing = false;
      }
      multi_threading_params.pending_requests_condition.notify_all();
      throw;
    }

    for (auto& future : futures) {
      // get() instead of wait() because we need to propagate potential exceptions.
      future.get();
    }
  }

  for (auto& [chunk_key, buffer] : multi_threading_params.chunk_encoder_buffers) {
    auto column_entry = column_by_id.find(chunk_key[CHUNK_KEY_COLUMN_IDX]);
    CHECK(column_entry != column_by_id.end());
    const auto& column_type = column_entry->second->columnType;
    auto chunk_metadata = buffer->getEncoder()->getMetadata(column_type);
    chunk_metadata->numElements = buffer->getEncoder()->getNumElems();
    const auto& cached_chunks = multi_threading_params.cached_chunks;
    if (!column_type.is_varlen_indeed()) {
      chunk_metadata->numBytes = column_type.get_size() * chunk_metadata->numElements;
    } else if (auto chunk_entry = cached_chunks.find(chunk_key);
               chunk_entry != cached_chunks.end()) {
      auto buffer = chunk_entry->second.getBuffer();
      CHECK(buffer);
      chunk_metadata->numBytes = buffer->size();
    } else {
      CHECK_EQ(chunk_metadata->numBytes, static_cast<size_t>(0));
    }
    chunk_metadata_map_[chunk_key] = chunk_metadata;
  }

  for (auto column : columns) {
    if (skip_metadata_scan(column)) {
      add_placeholder_metadata(
          column, foreign_table_, db_id_, start_row, num_rows_, chunk_metadata_map_);
    }
  }

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

  // Save chunk data
  if (foreign_table_->isAppendMode()) {
    chunk_encoder_buffers_ = std::move(multi_threading_params.chunk_encoder_buffers);
  }
}

Here is the call graph for this function:

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

private

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

Parameters

column_id_to_chunk_map	- map of column id to chunks to be populated
fragment_id	- fragment id of given chunks

Definition at line 251 of file AbstractTextFileDataWrapper.cpp.

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

Referenced by populateChunkBuffers().

                     {
  const auto copy_params = getFileBufferParser().validateAndGetCopyParams(foreign_table_);

  CHECK(!column_id_to_chunk_map.empty());
  const auto& file_regions = fragment_id_to_file_regions_map_[fragment_id];
  CHECK(!file_regions.empty());

  const auto buffer_size = get_buffer_size(file_regions);
  const auto thread_count = get_thread_count(copy_params, file_regions);

  const int batch_size = (file_regions.size() + thread_count - 1) / thread_count;

  std::vector<ParseBufferRequest> parse_file_requests{};
  parse_file_requests.reserve(thread_count);
  std::vector<std::future<ParseFileRegionResult>> futures{};
  std::set<int> column_filter_set;
  for (const auto& pair : column_id_to_chunk_map) {
    column_filter_set.insert(pair.first);
  }

  std::vector<std::unique_ptr<FileReader>> file_readers;
  rapidjson::Value reader_metadata(rapidjson::kObjectType);
  rapidjson::Document d;
  auto& server_options = foreign_table_->foreign_server->options;
  file_reader_->serialize(reader_metadata, d.GetAllocator());
  const auto file_path = getFullFilePath(foreign_table_);
  auto& parser = getFileBufferParser();

  for (size_t i = 0; i < file_regions.size(); i += batch_size) {
    parse_file_requests.emplace_back(
        buffer_size, copy_params, db_id_, foreign_table_, column_filter_set, file_path);
    auto start_index = i;
    auto end_index =
        std::min<size_t>(start_index + batch_size - 1, file_regions.size() - 1);

    if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
      file_readers.emplace_back(std::make_unique<LocalMultiFileReader>(
          file_path, copy_params, reader_metadata));
    } else {
      UNREACHABLE();
    }

    futures.emplace_back(std::async(std::launch::async,
                                    parse_file_regions,
                                    std::ref(file_regions),
                                    start_index,
                                    end_index,
                                    std::ref(*(file_readers.back())),
                                    std::ref(parse_file_requests.back()),
                                    std::ref(column_id_to_chunk_map),
                                    std::ref(parser)));
  }

  for (auto& future : futures) {
    future.wait();
  }

  std::vector<ParseFileRegionResult> load_file_region_results{};
  for (auto& future : futures) {
    load_file_region_results.emplace_back(future.get());
  }

  for (auto result : load_file_region_results) {
    for (auto& [column_id, chunk] : column_id_to_chunk_map) {
      chunk.appendData(
          result.column_id_to_data_blocks_map[column_id], result.row_count, 0);
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

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

overridevirtual

Restore internal state of datawrapper

Parameters

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

Implements foreign_storage::ForeignDataWrapper.

Definition at line 954 of file AbstractTextFileDataWrapper.cpp.

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

                                               {
  auto d = json_utils::read_from_file(file_path);
  CHECK(d.IsObject());

  // Restore fragment map
  json_utils::get_value_from_object(
      d, fragment_id_to_file_regions_map_, "fragment_id_to_file_regions_map");

  // Construct reader with metadta
  CHECK(d.HasMember("reader_metadata"));
  const auto copy_params = getFileBufferParser().validateAndGetCopyParams(foreign_table_);
  const auto full_file_path = getFullFilePath(foreign_table_);
  auto& server_options = foreign_table_->foreign_server->options;
  if (server_options.find(STORAGE_TYPE_KEY)->second == LOCAL_FILE_STORAGE_TYPE) {
    file_reader_ = std::make_unique<LocalMultiFileReader>(
        full_file_path, copy_params, d["reader_metadata"]);
  } else {
    UNREACHABLE();
  }

  json_utils::get_value_from_object(d, num_rows_, "num_rows");
  json_utils::get_value_from_object(d, append_start_offset_, "append_start_offset");

  // Now restore the internal metadata maps
  CHECK(chunk_metadata_map_.empty());
  CHECK(chunk_encoder_buffers_.empty());

  for (auto& pair : chunk_metadata) {
    chunk_metadata_map_[pair.first] = pair.second;

    if (foreign_table_->isAppendMode()) {
      // Restore encoder state for append mode
      chunk_encoder_buffers_[pair.first] = std::make_unique<ForeignStorageBuffer>();
      chunk_encoder_buffers_[pair.first]->initEncoder(pair.second->sqlType);
      chunk_encoder_buffers_[pair.first]->setSize(pair.second->numBytes);
      chunk_encoder_buffers_[pair.first]->getEncoder()->setNumElems(
          pair.second->numElements);
      chunk_encoder_buffers_[pair.first]->getEncoder()->resetChunkStats(
          pair.second->chunkStats);
      chunk_encoder_buffers_[pair.first]->setUpdated();
    }
  }
  is_restored_ = true;
}

Here is the call graph for this function:

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

private

Definition at line 108 of file AbstractTextFileDataWrapper.cpp.

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

Referenced by populateChunkBuffers().

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

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

size_t foreign_storage::AbstractTextFileDataWrapper::append_start_offset_

private

Definition at line 92 of file AbstractTextFileDataWrapper.h.

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

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

private

Definition at line 88 of file AbstractTextFileDataWrapper.h.

Referenced by populateChunkMetadata(), and restoreDataWrapperInternals().

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

private

Definition at line 79 of file AbstractTextFileDataWrapper.h.

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

const int foreign_storage::AbstractTextFileDataWrapper::db_id_

private

Definition at line 84 of file AbstractTextFileDataWrapper.h.

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

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

private

Definition at line 82 of file AbstractTextFileDataWrapper.h.

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

const ForeignTable* foreign_storage::AbstractTextFileDataWrapper::foreign_table_

private

Definition at line 85 of file AbstractTextFileDataWrapper.h.

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

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

private

Definition at line 80 of file AbstractTextFileDataWrapper.h.

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

bool foreign_storage::AbstractTextFileDataWrapper::is_restored_

private

Definition at line 94 of file AbstractTextFileDataWrapper.h.

Referenced by isRestored(), and restoreDataWrapperInternals().

size_t foreign_storage::AbstractTextFileDataWrapper::num_rows_

private

Definition at line 90 of file AbstractTextFileDataWrapper.h.

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

---

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

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