foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp} Namespace Reference

Functions
void	throw_fragment_id_out_of_bounds_error (const TableDescriptor *table, const int32_t fragment_id, const int32_t max_fragment_id)
std::set< const ColumnDescriptor * >	get_columns (const ChunkToBufferMap &buffers, const Catalog_Namespace::Catalog &catalog, const int32_t table_id, const int fragment_id)
bool	skip_metadata_scan (const ColumnDescriptor *column)
void	throw_unexpected_number_of_items (const size_t num_expected, const size_t num_loaded, const std::string &item_type, const std::string &foreign_table_name)
size_t	get_buffer_size (const import_export::CopyParams &copy_params, const bool size_known, const size_t file_size)
size_t	get_buffer_size (const FileRegions &file_regions)
size_t	get_thread_count (const import_export::CopyParams &copy_params, const bool size_known, const size_t file_size, const size_t buffer_size)
size_t	get_thread_count (const import_export::CopyParams &copy_params, const FileRegions &file_regions)
void	resize_delete_buffer (AbstractBuffer *delete_buffer, const size_t chunk_element_count)
bool	no_deferred_requests (MetadataScanMultiThreadingParams &multi_threading_params)
bool	is_file_scan_finished (const FileReader *file_reader, MetadataScanMultiThreadingParams &multi_threading_params)
foreign_storage::ForeignStorageCache *	get_cache_if_enabled (std::shared_ptr< Catalog_Namespace::Catalog > &catalog, const bool disable_cache)
void	add_placeholder_metadata (const ColumnDescriptor *column, const ForeignTable *foreign_table, const int db_id, const size_t start_row, const size_t total_num_rows, std::map< ChunkKey, std::shared_ptr< ChunkMetadata >> &chunk_metadata_map)
void	initialize_non_append_mode_scan (const std::map< ChunkKey, std::shared_ptr< ChunkMetadata >> &chunk_metadata_map, const std::map< int, FileRegions > &fragment_id_to_file_regions_map, const foreign_storage::OptionsMap &server_options, std::unique_ptr< FileReader > &file_reader, const std::string &file_path, const import_export::CopyParams &copy_params, const shared::FilePathOptions &file_path_options, const std::optional< size_t > &max_file_count, const foreign_storage::ForeignTable *foreign_table, const foreign_storage::UserMapping *user_mapping, const foreign_storage::TextFileBufferParser &parser, std::function< std::string()> get_s3_key, size_t &num_rows, size_t &append_start_offset)

Function Documentation

void foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::add_placeholder_metadata	(	const ColumnDescriptor *	column,
		const ForeignTable *	foreign_table,
		const int	db_id,
		const size_t	start_row,
		const size_t	total_num_rows,
		std::map< ChunkKey, std::shared_ptr< ChunkMetadata >> &	chunk_metadata_map
	)

Definition at line 1299 of file AbstractTextFileDataWrapper.cpp.

References CHUNK_KEY_FRAGMENT_IDX, ColumnDescriptor::columnId, ColumnDescriptor::columnType, foreign_storage::get_placeholder_metadata(), SQLTypeInfo::is_varlen_indeed(), TableDescriptor::maxFragRows, and TableDescriptor::tableId.

Referenced by foreign_storage::AbstractTextFileDataWrapper::populateChunkMetadata().

                                                                        {
  ChunkKey chunk_key = {db_id, foreign_table->tableId, column->columnId, 0};
  if (column->columnType.is_varlen_indeed()) {
    chunk_key.emplace_back(1);
  }

  // Create placeholder metadata for every fragment touched by this scan
  int start_fragment = start_row / foreign_table->maxFragRows;
  int end_fragment{0};
  if (total_num_rows > 0) {
    end_fragment = (total_num_rows - 1) / foreign_table->maxFragRows;
  }
  for (int fragment_id = start_fragment; fragment_id <= end_fragment; fragment_id++) {
    size_t num_elements = (static_cast<size_t>(foreign_table->maxFragRows *
                                               (fragment_id + 1)) > total_num_rows)
                              ? total_num_rows % foreign_table->maxFragRows
                              : foreign_table->maxFragRows;

    chunk_key[CHUNK_KEY_FRAGMENT_IDX] = fragment_id;
    chunk_metadata_map[chunk_key] =
        get_placeholder_metadata(column->columnType, num_elements);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::get_buffer_size	(	const import_export::CopyParams &	copy_params,
		const bool	size_known,
		const size_t	file_size
	)

Gets the appropriate buffer size to be used when processing file(s).

Definition at line 264 of file AbstractTextFileDataWrapper.cpp.

References import_export::CopyParams::buffer_size.

Referenced by foreign_storage::AbstractTextFileDataWrapper::iterativeFileScan(), foreign_storage::AbstractTextFileDataWrapper::populateChunkMetadata(), and foreign_storage::AbstractTextFileDataWrapper::populateChunks().

                                               {
  size_t buffer_size = copy_params.buffer_size;
  if (size_known && file_size < buffer_size) {
    buffer_size = file_size + 1;  // +1 for end of line character, if missing
  }
  return buffer_size;
}

Here is the caller graph for this function:

size_t foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::get_buffer_size

(

const FileRegions &

file_regions

)

Definition at line 274 of file AbstractTextFileDataWrapper.cpp.

References CHECK.

                                                        {
  size_t buffer_size = 0;
  for (const auto& file_region : file_regions) {
    buffer_size = std::max(buffer_size, file_region.region_size);
  }
  CHECK(buffer_size);
  return buffer_size;
}

foreign_storage::ForeignStorageCache* foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::get_cache_if_enabled	(	std::shared_ptr< Catalog_Namespace::Catalog > &	catalog,
		const bool	disable_cache
	)

Definition at line 593 of file AbstractTextFileDataWrapper.cpp.

Referenced by foreign_storage::cache_blocks().

                              {
  if (!disable_cache && catalog->getDataMgr()
                            .getPersistentStorageMgr()
                            ->getDiskCacheConfig()
                            .isEnabledForFSI()) {
    return catalog->getDataMgr().getPersistentStorageMgr()->getDiskCache();
  } else {
    return nullptr;
  }
}

Here is the caller graph for this function:

std::set<const ColumnDescriptor*> foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::get_columns	(	const ChunkToBufferMap &	buffers,
		const Catalog_Namespace::Catalog &	catalog,
		const int32_t	table_id,
		const int	fragment_id
	)

Definition at line 83 of file AbstractTextFileDataWrapper.cpp.

References CHECK, CHECK_EQ, CHUNK_KEY_COLUMN_IDX, CHUNK_KEY_FRAGMENT_IDX, and Catalog_Namespace::Catalog::getMetadataForColumn().

Referenced by foreign_storage::AbstractTextFileDataWrapper::populateChunkBuffers().

                                                                     {
  CHECK(!buffers.empty());
  std::set<const ColumnDescriptor*> columns;
  for (const auto& entry : buffers) {
    CHECK_EQ(fragment_id, entry.first[CHUNK_KEY_FRAGMENT_IDX]);
    const auto column_id = entry.first[CHUNK_KEY_COLUMN_IDX];
    const auto column = catalog.getMetadataForColumn(table_id, column_id);
    columns.emplace(column);
  }
  return columns;
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::get_thread_count	(	const import_export::CopyParams &	copy_params,
		const bool	size_known,
		const size_t	file_size,
		const size_t	buffer_size
	)

Gets the appropriate number of threads to be used for concurrent processing within the data wrapper.

Definition at line 287 of file AbstractTextFileDataWrapper.cpp.

References CHECK_GT, and import_export::CopyParams::threads.

Referenced by foreign_storage::AbstractTextFileDataWrapper::iterativeFileScan(), foreign_storage::AbstractTextFileDataWrapper::populateChunkMetadata(), and foreign_storage::AbstractTextFileDataWrapper::populateChunks().

                                                  {
  size_t thread_count = copy_params.threads;
  if (thread_count == 0) {
    thread_count = std::thread::hardware_concurrency();
  }
  if (size_known && file_size > 0) {
    size_t num_buffers_in_file = (file_size + buffer_size - 1) / buffer_size;
    if (num_buffers_in_file < thread_count) {
      thread_count = num_buffers_in_file;
    }
  }
  CHECK_GT(thread_count, static_cast<size_t>(0));
  return thread_count;
}

Here is the caller graph for this function:

size_t foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::get_thread_count	(	const import_export::CopyParams &	copy_params,
		const FileRegions &	file_regions
	)

Definition at line 305 of file AbstractTextFileDataWrapper.cpp.

References CHECK_GT, and import_export::CopyParams::threads.

                                                         {
  size_t thread_count = copy_params.threads;
  if (thread_count == 0) {
    thread_count =
        std::min<size_t>(std::thread::hardware_concurrency(), file_regions.size());
  }
  CHECK_GT(thread_count, static_cast<size_t>(0));
  return thread_count;
}

void foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::initialize_non_append_mode_scan	(	const std::map< ChunkKey, std::shared_ptr< ChunkMetadata >> &	chunk_metadata_map,
		const std::map< int, FileRegions > &	fragment_id_to_file_regions_map,
		const foreign_storage::OptionsMap &	server_options,
		std::unique_ptr< FileReader > &	file_reader,
		const std::string &	file_path,
		const import_export::CopyParams &	copy_params,
		const shared::FilePathOptions &	file_path_options,
		const std::optional< size_t > &	max_file_count,
		const foreign_storage::ForeignTable *	foreign_table,
		const foreign_storage::UserMapping *	user_mapping,
		const foreign_storage::TextFileBufferParser &	parser,
		std::function< std::string()>	get_s3_key,
		size_t &	num_rows,
		size_t &	append_start_offset
	)

Definition at line 1329 of file AbstractTextFileDataWrapper.cpp.

References CHECK, foreign_storage::AbstractFileStorageDataWrapper::LOCAL_FILE_STORAGE_TYPE, foreign_storage::AbstractFileStorageDataWrapper::STORAGE_TYPE_KEY, UNREACHABLE, and foreign_storage::TextFileBufferParser::validateFiles().

Referenced by foreign_storage::AbstractTextFileDataWrapper::iterativeFileScan(), and foreign_storage::AbstractTextFileDataWrapper::populateChunkMetadata().

                                 {
  // 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(foreign_storage::AbstractTextFileDataWrapper::STORAGE_TYPE_KEY)
          ->second ==
      foreign_storage::AbstractTextFileDataWrapper::LOCAL_FILE_STORAGE_TYPE) {
    file_reader = std::make_unique<LocalMultiFileReader>(
        file_path, copy_params, file_path_options, max_file_count);
  } else {
    UNREACHABLE();
  }
  parser.validateFiles(file_reader.get(), foreign_table);
  num_rows = 0;
  append_start_offset = 0;
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::is_file_scan_finished	(	const FileReader *	file_reader,
		MetadataScanMultiThreadingParams &	multi_threading_params
	)

Definition at line 331 of file AbstractTextFileDataWrapper.cpp.

References CHECK, foreign_storage::FileReader::isScanFinished(), and no_deferred_requests().

Referenced by foreign_storage::AbstractTextFileDataWrapper::iterativeFileScan(), and foreign_storage::AbstractTextFileDataWrapper::populateChunks().

                                                                                     {
  CHECK(file_reader);
  return file_reader->isScanFinished() && no_deferred_requests(multi_threading_params);
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::no_deferred_requests

(

MetadataScanMultiThreadingParams &

multi_threading_params

)

Definition at line 325 of file AbstractTextFileDataWrapper.cpp.

References foreign_storage::MetadataScanMultiThreadingParams::deferred_requests, and foreign_storage::MetadataScanMultiThreadingParams::deferred_requests_mutex.

Referenced by foreign_storage::dispatch_scan_requests(), and is_file_scan_finished().

                                                                                    {
  std::unique_lock<std::mutex> deferred_requests_lock(
      multi_threading_params.deferred_requests_mutex);
  return multi_threading_params.deferred_requests.empty();
}

Here is the caller graph for this function:

void foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::resize_delete_buffer	(	AbstractBuffer *	delete_buffer,
		const size_t	chunk_element_count
	)

Definition at line 316 of file AbstractTextFileDataWrapper.cpp.

References Data_Namespace::AbstractBuffer::append(), and Data_Namespace::AbstractBuffer::size().

Referenced by foreign_storage::AbstractTextFileDataWrapper::populateChunks(), and foreign_storage::update_delete_buffer().

                                                            {
  if (delete_buffer->size() < chunk_element_count) {
    auto remaining_rows = chunk_element_count - delete_buffer->size();
    std::vector<int8_t> data(remaining_rows, false);
    delete_buffer->append(data.data(), remaining_rows);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

bool foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::skip_metadata_scan

(

const ColumnDescriptor *

column

)

Definition at line 98 of file AbstractTextFileDataWrapper.cpp.

References ColumnDescriptor::columnType, and SQLTypeInfo::is_dict_encoded_type().

Referenced by foreign_storage::AbstractTextFileDataWrapper::populateChunkMetadata(), and foreign_storage::AbstractTextFileDataWrapper::updateMetadata().

                                                        {
  return column->columnType.is_dict_encoded_type();
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::throw_fragment_id_out_of_bounds_error	(	const TableDescriptor *	table,
		const int32_t	fragment_id,
		const int32_t	max_fragment_id
	)

Definition at line 73 of file AbstractTextFileDataWrapper.cpp.

References TableDescriptor::tableName, and to_string().

Referenced by foreign_storage::AbstractTextFileDataWrapper::populateChunks().

                                                                          {
  throw RequestedFragmentIdOutOfBoundsException{
      "Attempting to populate fragment id " + std::to_string(fragment_id) +
      " for foreign table " + table->tableName +
      " which is greater than the maximum fragment id of " +
      std::to_string(max_fragment_id) + "."};
}

Here is the call graph for this function:

Here is the caller graph for this function:

void foreign_storage::anonymous_namespace{AbstractTextFileDataWrapper.cpp}::throw_unexpected_number_of_items	(	const size_t	num_expected,
		const size_t	num_loaded,
		const std::string &	item_type,
		const std::string &	foreign_table_name
	)

Definition at line 196 of file AbstractTextFileDataWrapper.cpp.

References foreign_storage::throw_unexpected_number_of_items().

                                                                           {
  try {
    foreign_storage::throw_unexpected_number_of_items(
        num_expected, num_loaded, item_type);
  } catch (const foreign_storage::ForeignStorageException& except) {
    throw foreign_storage::ForeignStorageException(
        std::string(except.what()) + " Foreign table: " + foreign_table_name);
  }
}

Here is the call graph for this function: