anonymous_namespace{ForeignDataImporter.cpp} Namespace Reference

Classes
struct	FragmentBuffers

Functions
ChunkMetadataVector	metadata_scan (foreign_storage::ForeignDataWrapper *data_wrapper, foreign_storage::ForeignTable *foreign_table)
std::string	get_import_id (const import_export::CopyParams &copy_params, const std::string &copy_from_source)
void	validate_copy_params (const import_export::CopyParams &copy_params)
std::unique_ptr< FragmentBuffers >	create_fragment_buffers (const int32_t fragment_id, Catalog_Namespace::Catalog &catalog, const TableDescriptor *table)
void	load_foreign_data_buffers (Fragmenter_Namespace::InsertDataLoader::InsertConnector *connector, Catalog_Namespace::Catalog &catalog, const TableDescriptor *table, const Catalog_Namespace::SessionInfo *session_info, const import_export::CopyParams &copy_params, const std::string &copy_from_source, import_export::ImportStatus &import_status, std::mutex &communication_mutex, bool &continue_loading, bool &load_failed, bool &data_wrapper_error_occured, std::condition_variable &buffers_to_load_condition, std::list< std::unique_ptr< FragmentBuffers >> &buffers_to_load)
import_export::ImportStatus	import_foreign_data (const int32_t max_fragment_id, Fragmenter_Namespace::InsertDataLoader::InsertConnector *connector, Catalog_Namespace::Catalog &catalog, const TableDescriptor *table, foreign_storage::ForeignDataWrapper *data_wrapper, const Catalog_Namespace::SessionInfo *session_info, const import_export::CopyParams &copy_params, const std::string &copy_from_source, const size_t maximum_num_fragments_buffered)

Function Documentation

std::unique_ptr<FragmentBuffers> anonymous_namespace{ForeignDataImporter.cpp}::create_fragment_buffers	(	const int32_t	fragment_id,
		Catalog_Namespace::Catalog &	catalog,
		const TableDescriptor *	table
	)

Definition at line 94 of file ForeignDataImporter.cpp.

References shared::get_from_map(), Catalog_Namespace::Catalog::getAllColumnMetadataForTable(), Catalog_Namespace::Catalog::getDatabaseId(), and TableDescriptor::tableId.

Referenced by import_foreign_data().

                                  {
  auto columns = catalog.getAllColumnMetadataForTable(table->tableId, false, false, true);

  std::set<ChunkKey> fragment_keys;
  for (const auto col_desc : columns) {
    ChunkKey key{
        catalog.getDatabaseId(), table->tableId, col_desc->columnId, fragment_id};
    if (col_desc->columnType.is_varlen_indeed()) {
      auto data_key = key;
      data_key.push_back(1);
      fragment_keys.insert(data_key);
      auto index_key = key;
      index_key.push_back(2);
      fragment_keys.insert(index_key);
    } else {
      fragment_keys.insert(key);
    }
  }

  // create buffers
  std::unique_ptr<FragmentBuffers> frag_buffers = std::make_unique<FragmentBuffers>();
  frag_buffers->delete_buffer = std::make_unique<foreign_storage::ForeignStorageBuffer>();
  for (const auto& key : fragment_keys) {
    frag_buffers->fragment_buffers_owner[key] =
        std::make_unique<foreign_storage::ForeignStorageBuffer>();
    frag_buffers->fragment_buffers[key] =
        shared::get_from_map(frag_buffers->fragment_buffers_owner, key).get();
  }

  return frag_buffers;
}

Here is the call graph for this function:

Here is the caller graph for this function:

std::string anonymous_namespace{ForeignDataImporter.cpp}::get_import_id	(	const import_export::CopyParams &	copy_params,
		const std::string &	copy_from_source
	)

Definition at line 67 of file ForeignDataImporter.cpp.

References import_export::kDelimitedFile, import_export::kParquetFile, import_export::kRegexParsedFile, and import_export::CopyParams::source_type.

Referenced by load_foreign_data_buffers().

                                                             {
  if (copy_params.source_type == import_export::SourceType::kDelimitedFile ||
      copy_params.source_type == import_export::SourceType::kRegexParsedFile
#ifdef ENABLE_IMPORT_PARQUET
      || copy_params.source_type == import_export::SourceType::kParquetFile
#endif
  ) {
    return boost::filesystem::path(copy_from_source).filename().string();
  }

  return copy_from_source;
}

Here is the caller graph for this function:

import_export::ImportStatus anonymous_namespace{ForeignDataImporter.cpp}::import_foreign_data	(	const int32_t	max_fragment_id,
		Fragmenter_Namespace::InsertDataLoader::InsertConnector *	connector,
		Catalog_Namespace::Catalog &	catalog,
		const TableDescriptor *	table,
		foreign_storage::ForeignDataWrapper *	data_wrapper,
		const Catalog_Namespace::SessionInfo *	session_info,
		const import_export::CopyParams &	copy_params,
		const std::string &	copy_from_source,
		const size_t	maximum_num_fragments_buffered
	)

Definition at line 242 of file ForeignDataImporter.cpp.

References threading_serial::async(), create_fragment_buffers(), load_foreign_data_buffers(), and foreign_storage::ForeignDataWrapper::populateChunkBuffers().

Referenced by import_export::ForeignDataImporter::importGeneral().

                                                 {
  import_export::ImportStatus import_status;

  std::mutex communication_mutex;
  bool continue_loading =
      true;  // when false, signals that the last buffer to load has been added, and
             // loading should cease after loading remaining buffers
  bool load_failed =
      false;  // signals loading has failed and buffer population should cease
  bool data_wrapper_error_occured = false;  // signals an error occured during buffer
                                            // population and loading should cease
  std::condition_variable buffers_to_load_condition;
  std::list<std::unique_ptr<FragmentBuffers>> buffers_to_load;

  // launch separate thread to load processed fragment buffers
  auto load_future = std::async(std::launch::async,
                                load_foreign_data_buffers,
                                connector,
                                std::ref(catalog),
                                table,
                                session_info,
                                std::cref(copy_params),
                                std::cref(copy_from_source),
                                std::ref(import_status),
                                std::ref(communication_mutex),
                                std::ref(continue_loading),
                                std::ref(load_failed),
                                std::ref(data_wrapper_error_occured),
                                std::ref(buffers_to_load_condition),
                                std::ref(buffers_to_load));

  for (int32_t fragment_id = 0; fragment_id <= max_fragment_id; ++fragment_id) {
    {
      std::unique_lock communication_lock(communication_mutex);
      buffers_to_load_condition.wait(communication_lock, [&]() {
        return buffers_to_load.size() < maximum_num_fragments_buffered || load_failed;
      });
      if (load_failed) {
        break;
      }
    }
    auto grouped_fragment_buffers = create_fragment_buffers(fragment_id, catalog, table);
    auto& fragment_buffers = grouped_fragment_buffers->fragment_buffers;
    auto& delete_buffer = grouped_fragment_buffers->delete_buffer;

    // get the buffers, accounting for the possibility of the requested fragment id being
    // out of bounds
    try {
      data_wrapper->populateChunkBuffers(fragment_buffers, {}, delete_buffer.get());
    } catch (const foreign_storage::RequestedFragmentIdOutOfBoundsException& except) {
      break;
    } catch (...) {
      std::unique_lock communication_lock(communication_mutex);
      data_wrapper_error_occured = true;
      buffers_to_load_condition.notify_all();
      throw;
    }

    std::unique_lock communication_lock(communication_mutex);
    buffers_to_load.emplace_back(std::move(grouped_fragment_buffers));
    buffers_to_load_condition.notify_all();
  }

  {  // data wrapper processing has finished, notify loading thread
    std::unique_lock communication_lock(communication_mutex);
    continue_loading = false;
    buffers_to_load_condition.notify_all();
  }

  // any exceptions in separate loading thread will occur here
  load_future.get();

  return import_status;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void anonymous_namespace{ForeignDataImporter.cpp}::load_foreign_data_buffers	(	Fragmenter_Namespace::InsertDataLoader::InsertConnector *	connector,
		Catalog_Namespace::Catalog &	catalog,
		const TableDescriptor *	table,
		const Catalog_Namespace::SessionInfo *	session_info,
		const import_export::CopyParams &	copy_params,
		const std::string &	copy_from_source,
		import_export::ImportStatus &	import_status,
		std::mutex &	communication_mutex,
		bool &	continue_loading,
		bool &	load_failed,
		bool &	data_wrapper_error_occured,
		std::condition_variable &	buffers_to_load_condition,
		std::list< std::unique_ptr< FragmentBuffers >> &	buffers_to_load
	)

Definition at line 129 of file ForeignDataImporter.cpp.

References CHECK, CHECK_LE, CHUNK_KEY_COLUMN_IDX, CHUNK_KEY_TABLE_IDX, CHUNK_KEY_VARLEN_IDX, shared::get_from_map(), get_import_id(), Chunk_NS::Chunk::getChunk(), Catalog_Namespace::Catalog::getDatabaseId(), Catalog_Namespace::Catalog::getMetadataForColumn(), Fragmenter_Namespace::InsertDataLoader::insertChunks(), import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, import_export::CopyParams::max_reject, import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_rejected, import_export::Importer::set_import_status(), and TableDescriptor::tableId.

Referenced by import_foreign_data().

                                                              {
  Fragmenter_Namespace::InsertDataLoader insert_data_loader(*connector);
  while (true) {
    {
      std::unique_lock communication_lock(communication_mutex);
      buffers_to_load_condition.wait(communication_lock, [&]() {
        return !buffers_to_load.empty() || !continue_loading ||
               data_wrapper_error_occured;
      });
      if ((buffers_to_load.empty() && !continue_loading) || data_wrapper_error_occured) {
        return;
      }
    }

    CHECK(!buffers_to_load.empty());

    try {
      std::unique_ptr<FragmentBuffers> grouped_fragment_buffers;
      {
        std::unique_lock communication_lock(communication_mutex);
        grouped_fragment_buffers.reset(buffers_to_load.front().release());
        buffers_to_load.pop_front();
        buffers_to_load_condition.notify_all();
      }
      auto& fragment_buffers = grouped_fragment_buffers->fragment_buffers;
      auto& delete_buffer = grouped_fragment_buffers->delete_buffer;

      // get chunks for import
      Fragmenter_Namespace::InsertChunks insert_chunks{
          table->tableId, catalog.getDatabaseId(), {}, {}};

      // create chunks from buffers
      for (const auto& [key, buffer] : fragment_buffers) {
        const auto col_id = key[CHUNK_KEY_COLUMN_IDX];
        const auto table_id = key[CHUNK_KEY_TABLE_IDX];
        const auto col_desc = catalog.getMetadataForColumn(table_id, col_id);

        if (col_desc->columnType.is_varlen_indeed()) {
          CHECK(key.size() > CHUNK_KEY_VARLEN_IDX);  // check for varlen key
          if (key[CHUNK_KEY_VARLEN_IDX] == 1) {      // data key
            auto index_key = key;
            index_key[CHUNK_KEY_VARLEN_IDX] = 2;
            insert_chunks.chunks[col_id] = Chunk_NS::Chunk::getChunk(
                col_desc,
                buffer,
                shared::get_from_map(fragment_buffers, index_key),
                false);
          }
        } else {  // regular non-varlen case with no index buffer
          insert_chunks.chunks[col_id] =
              Chunk_NS::Chunk::getChunk(col_desc, buffer, nullptr, false);
        }
      }

      // mark which row indices are valid for import
      auto row_count = fragment_buffers.begin()
                           ->second->getEncoder()
                           ->getNumElems();  // assume all chunks have same number of
                                             // rows, this is validated at a lower level
      insert_chunks.valid_row_indices.reserve(row_count);
      for (size_t irow = 0; irow < row_count; ++irow) {
        if (delete_buffer->size() > 0) {
          CHECK_LE(irow, delete_buffer->size());
          if (delete_buffer->getMemoryPtr()[irow]) {
            continue;
          }
        }
        insert_chunks.valid_row_indices.emplace_back(irow);
      }

      // import chunks
      insert_data_loader.insertChunks(*session_info, insert_chunks);

      CHECK_LE(insert_chunks.valid_row_indices.size(), row_count);
      import_status.rows_rejected += row_count - insert_chunks.valid_row_indices.size();
      import_status.rows_completed += insert_chunks.valid_row_indices.size();
      if (import_status.rows_rejected > copy_params.max_reject) {
        import_status.load_failed = true;
        import_status.load_msg =
            "Load was cancelled due to max reject rows being reached";
        import_export::Importer::set_import_status(
            get_import_id(copy_params, copy_from_source), import_status);
        std::unique_lock communication_lock(communication_mutex);
        load_failed = true;
        buffers_to_load_condition.notify_all();
        return;
      }
      import_export::Importer::set_import_status(
          get_import_id(copy_params, copy_from_source), import_status);
    } catch (...) {
      {
        std::unique_lock communication_lock(communication_mutex);
        load_failed = true;
        buffers_to_load_condition.notify_all();
      }
      throw;
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

ChunkMetadataVector anonymous_namespace{ForeignDataImporter.cpp}::metadata_scan	(	foreign_storage::ForeignDataWrapper *	data_wrapper,
		foreign_storage::ForeignTable *	foreign_table
	)

Definition at line 43 of file ForeignDataImporter.cpp.

References TableDescriptor::maxFragRows, foreign_storage::MetadataScanInfeasibleFragmentSizeException::min_feasible_fragment_size_, and foreign_storage::ForeignDataWrapper::populateChunkMetadata().

Referenced by import_export::ForeignDataImporter::importGeneral().

                                                                              {
  ChunkMetadataVector metadata_vector;
  try {
    data_wrapper->populateChunkMetadata(
        metadata_vector);  // explicitly invoke a metadata scan on data wrapper
  } catch (const foreign_storage::MetadataScanInfeasibleFragmentSizeException&
               metadata_scan_exception) {
    // if a metadata scan exception is thrown, check to see if we can adjust
    // the fragment size and retry

    auto min_feasible_fragment_size = metadata_scan_exception.min_feasible_fragment_size_;
    if (min_feasible_fragment_size < 0) {
      throw;  // no valid fragment size returned by exception
    }
    foreign_table->maxFragRows = min_feasible_fragment_size;
    data_wrapper->populateChunkMetadata(
        metadata_vector);  // attempt another metadata scan, note, we assume that the
                           // metadata scan can be reentered safely after throwing the
                           // exception
  }
  return metadata_vector;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void anonymous_namespace{ForeignDataImporter.cpp}::validate_copy_params

(

const import_export::CopyParams &

copy_params

)

Definition at line 81 of file ForeignDataImporter.cpp.

References import_export::kRegexParsedFile, import_export::CopyParams::source_type, and foreign_storage::validate_regex_parser_options().

Referenced by import_export::ForeignDataImporter::importGeneral().

                                                                      {
  if (copy_params.source_type == import_export::SourceType::kRegexParsedFile) {
    foreign_storage::validate_regex_parser_options(copy_params);
  }
}

Here is the call graph for this function:

Here is the caller graph for this function: