Fragmenter_Namespace::InsertDataLoader Class Reference

#include <InsertDataLoader.h>

Collaboration diagram for Fragmenter_Namespace::InsertDataLoader:

Classes
class	InsertConnector

Public Member Functions
	InsertDataLoader (InsertConnector &connector)
void	insertData (const Catalog_Namespace::SessionInfo &session_info, InsertData &insert_data)
void	insertChunks (const Catalog_Namespace::SessionInfo &session_info, const InsertChunks &insert_chunks)
size_t	getLeafCount () const

Private Member Functions
size_t	moveToNextLeaf ()

Private Attributes
size_t	leaf_count_
size_t	current_leaf_index_
InsertConnector &	connector_
std::shared_mutex	current_leaf_index_mutex_

Detailed Description

Definition at line 25 of file InsertDataLoader.h.

Constructor & Destructor Documentation

Fragmenter_Namespace::InsertDataLoader::InsertDataLoader ( InsertConnector &  connector )

inline

Definition at line 46 of file InsertDataLoader.h.

      : leaf_count_(connector.leafCount())
      , current_leaf_index_(0)
      , connector_(connector) {}

Member Function Documentation

size_t Fragmenter_Namespace::InsertDataLoader::getLeafCount ( ) const

inline

Definition at line 57 of file InsertDataLoader.h.

References leaf_count_.

Referenced by RelAlgExecutor::executeSimpleInsert().

{ return leaf_count_; }

Here is the caller graph for this function:

void Fragmenter_Namespace::InsertDataLoader::insertChunks	(	const Catalog_Namespace::SessionInfo &	session_info,
		const InsertChunks &	insert_chunks
	)

Definition at line 418 of file InsertDataLoader.cpp.

References threading_serial::async(), cat(), CHECK, Fragmenter_Namespace::compute_row_indices_of_shards(), connector_, Fragmenter_Namespace::copy_data_of_shard(), Catalog_Namespace::SessionInfo::getCatalog(), Fragmenter_Namespace::InsertDataLoader::InsertConnector::insertChunksToLeaf(), Fragmenter_Namespace::InsertDataLoader::InsertConnector::leafCount(), moveToNextLeaf(), and Fragmenter_Namespace::InsertChunks::table_id.

Referenced by anonymous_namespace{ForeignDataImporter.cpp}::load_foreign_data_buffers().

                                                                       {
  const auto& cat = session_info.getCatalog();
  const auto* td = cat.getMetadataForTable(insert_chunks.table_id);

  CHECK(td);
  if (td->nShards == 0) {
    connector_.insertChunksToLeaf(session_info, moveToNextLeaf(), insert_chunks);
  } else {
    // we have a sharded target table, start spreading to physical tables
    auto row_indices_of_shards =
        compute_row_indices_of_shards(cat, connector_.leafCount(), insert_chunks);

    auto insert_shard_data =
        [this, &session_info, &insert_chunks, &cat, &td, &row_indices_of_shards](
            size_t shardId) {
          const auto shard_tables = cat.getPhysicalTablesDescriptors(td);
          auto stard_table_idx = shardId % td->nShards;
          auto shard_leaf_idx = shardId / td->nShards;

          const auto& row_indices_of_shard = row_indices_of_shards[shardId];

          auto [buffers, shard_insert_chunks] = copy_data_of_shard(
              cat, insert_chunks, stard_table_idx, row_indices_of_shard);
          connector_.insertChunksToLeaf(
              session_info, shard_leaf_idx, shard_insert_chunks);
        };

    std::vector<std::future<void>> worker_threads;
    for (size_t shard_id = 0; shard_id < row_indices_of_shards.size(); shard_id++) {
      if (row_indices_of_shards[shard_id].size() > 0) {
        worker_threads.push_back(
            std::async(std::launch::async, insert_shard_data, shard_id));
      }
    }
    for (auto& child : worker_threads) {
      child.wait();
    }
    for (auto& child : worker_threads) {
      child.get();
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void Fragmenter_Namespace::InsertDataLoader::insertData	(	const Catalog_Namespace::SessionInfo &	session_info,
		InsertData &	insert_data
	)

Definition at line 462 of file InsertDataLoader.cpp.

References threading_serial::async(), cat(), CHECK, Fragmenter_Namespace::computeRowIndicesOfShards(), connector_, Fragmenter_Namespace::copyDataOfShard(), Catalog_Namespace::SessionInfo::getCatalog(), Fragmenter_Namespace::InsertDataLoader::InsertConnector::insertDataToLeaf(), Fragmenter_Namespace::InsertDataLoader::InsertConnector::leafCount(), moveToNextLeaf(), Fragmenter_Namespace::InsertData::numRows, and Fragmenter_Namespace::InsertData::tableId.

Referenced by RelAlgExecutor::executeSimpleInsert(), and Parser::InsertIntoTableAsSelectStmt::populateData().

                                                           {
  const auto& cat = session_info.getCatalog();
  const auto* td = cat.getMetadataForTable(insert_data.tableId);

  CHECK(td);
  if (td->nShards == 0) {
    connector_.insertDataToLeaf(session_info, moveToNextLeaf(), insert_data);
  } else {
    // we have a sharded target table, start spreading to physical tables
    auto rowIndicesOfShards =
        computeRowIndicesOfShards(cat, connector_.leafCount(), insert_data);

    auto insertShardData =
        [this, &session_info, &insert_data, &cat, &td, &rowIndicesOfShards](
            size_t shardId) {
          const auto shard_tables = cat.getPhysicalTablesDescriptors(td);
          auto stardTableIdx = shardId % td->nShards;
          auto shardLeafIdx = shardId / td->nShards;

          const auto& rowIndicesOfShard = rowIndicesOfShards[shardId];
          ShardDataOwner shardDataOwner;

          InsertData shardData = copyDataOfShard(
              cat, shardDataOwner, insert_data, stardTableIdx, rowIndicesOfShard);
          CHECK(shardData.numRows > 0);
          connector_.insertDataToLeaf(session_info, shardLeafIdx, shardData);
        };

    std::vector<std::future<void>> worker_threads;
    for (size_t shardId = 0; shardId < rowIndicesOfShards.size(); shardId++) {
      if (rowIndicesOfShards[shardId].size() > 0) {
        worker_threads.push_back(
            std::async(std::launch::async, insertShardData, shardId));
      }
    }
    for (auto& child : worker_threads) {
      child.wait();
    }
    for (auto& child : worker_threads) {
      child.get();
    }
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

size_t Fragmenter_Namespace::InsertDataLoader::moveToNextLeaf ( )

private

Move to the next available leaf index internally. Done under a lock to prevent contention.

Returns

the current leaf index (prior to moving to the next index)

Definition at line 408 of file InsertDataLoader.cpp.

References current_leaf_index_, current_leaf_index_mutex_, and leaf_count_.

Referenced by insertChunks(), and insertData().

                                        {
  std::unique_lock current_leaf_index_lock(current_leaf_index_mutex_);
  size_t starting_leaf_index = current_leaf_index_;
  current_leaf_index_++;
  if (current_leaf_index_ >= leaf_count_) {
    current_leaf_index_ = 0;
  }
  return starting_leaf_index;
}

Here is the caller graph for this function:

Member Data Documentation

InsertConnector& Fragmenter_Namespace::InsertDataLoader::connector_

private

Definition at line 70 of file InsertDataLoader.h.

Referenced by insertChunks(), and insertData().

size_t Fragmenter_Namespace::InsertDataLoader::current_leaf_index_

private

Definition at line 69 of file InsertDataLoader.h.

Referenced by moveToNextLeaf().

std::shared_mutex Fragmenter_Namespace::InsertDataLoader::current_leaf_index_mutex_

private

Definition at line 71 of file InsertDataLoader.h.

Referenced by moveToNextLeaf().

size_t Fragmenter_Namespace::InsertDataLoader::leaf_count_

private

Definition at line 68 of file InsertDataLoader.h.

Referenced by getLeafCount(), and moveToNextLeaf().

---

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

• /home/jenkins-slave/workspace/core-os-doxygen/Fragmenter/InsertDataLoader.h
• /home/jenkins-slave/workspace/core-os-doxygen/Fragmenter/InsertDataLoader.cpp