#include <ParserNode.h>

Inheritance diagram for Parser::InsertIntoTableAsSelectStmt:

Collaboration diagram for Parser::InsertIntoTableAsSelectStmt:

Classes
struct	DistributedConnector

struct	LocalConnector

Public Member Functions
	InsertIntoTableAsSelectStmt (const std::string table_name, const std::string select_query, std::list< std::string * > *c)

void	populateData (QueryStateProxy, bool is_temporary, bool validate_table)

void	execute (const Catalog_Namespace::SessionInfo &session) override

std::string &	get_table ()

std::string &	get_select_query ()

Public Member Functions inherited from Parser::DDLStmt
void	setColumnDescriptor (ColumnDescriptor &cd, const ColumnDef *coldef)

Public Member Functions inherited from Parser::Node
virtual	~Node ()

Public Attributes
DistributedConnector *	leafs_connector_ = nullptr

Protected Attributes
std::vector< std::unique_ptr < std::string > >	column_list_

std::string	table_name_

std::string	select_query_

Detailed Description

Definition at line 1007 of file ParserNode.h.

Constructor & Destructor Documentation

Parser::InsertIntoTableAsSelectStmt::InsertIntoTableAsSelectStmt	(	const std::string *	table_name,
		const std::string *	select_query,
		std::list< std::string * > *	c
	)

inline

Definition at line 1010 of file ParserNode.h.

References column_list_.

       : table_name_(*table_name), select_query_(*select_query) {
     if (c) {
       for (auto e : *c) {
         column_list_.emplace_back(e);
       }
       delete c;
     }
 
     delete table_name;
     delete select_query;
   }

Member Function Documentation

void Parser::InsertIntoTableAsSelectStmt::execute ( const Catalog_Namespace::SessionInfo & session )

overridevirtual

Implements Parser::DDLStmt.

Reimplemented in Parser::CreateTableAsSelectStmt.

Definition at line 2768 of file ParserNode.cpp.

References query_state::QueryState::create(), and STDLOG.

                                                                                      {
   auto session_copy = session;
   auto session_ptr = std::shared_ptr<Catalog_Namespace::SessionInfo>(
       &session_copy, boost::null_deleter());
   auto query_state = query_state::QueryState::create(session_ptr, select_query_);
   auto stdlog = STDLOG(query_state);
   populateData(query_state->createQueryStateProxy(), false, true);
 }

Here is the call graph for this function:

std::string& Parser::InsertIntoTableAsSelectStmt::get_select_query ( )

inline

Definition at line 1030 of file ParserNode.h.

References select_query_.

1030 { return select_query_; }

Parser::InsertIntoTableAsSelectStmt::select_query_

std::string select_query_

Definition: ParserNode.h:1062

std::string& Parser::InsertIntoTableAsSelectStmt::get_table ( )

inline

Definition at line 1028 of file ParserNode.h.

References table_name_.

1028 { return table_name_; }

Parser::InsertIntoTableAsSelectStmt::table_name_

std::string table_name_

Definition: ParserNode.h:1061

void Parser::InsertIntoTableAsSelectStmt::populateData	(	QueryStateProxy	query_state_proxy,
		bool	is_temporary,
		bool	validate_table
	)

Definition at line 2402 of file ParserNode.cpp.

                                                                     {
   auto const session = query_state_proxy.getQueryState().getConstSessionInfo();
   LocalConnector local_connector;
 
   bool populate_table = false;
 
   if (leafs_connector_) {
     populate_table = true;
   } else {
     leafs_connector_ = &local_connector;
     if (!g_cluster) {
       populate_table = true;
     }
   }
 
   auto& catalog = session->getCatalog();
   auto get_target_column_descriptors = [this, &catalog](const TableDescriptor* td) {
     std::vector<const ColumnDescriptor*> target_column_descriptors;
     if (column_list_.empty()) {
       auto list = catalog.getAllColumnMetadataForTable(td->tableId, false, false, false);
       target_column_descriptors = {std::begin(list), std::end(list)};
 
     } else {
       for (auto& c : column_list_) {
         const ColumnDescriptor* cd = catalog.getMetadataForColumn(td->tableId, *c);
         if (cd == nullptr) {
           throw std::runtime_error("Column " + *c + " does not exist.");
         }
         target_column_descriptors.push_back(cd);
       }
     }
 
     return target_column_descriptors;
   };
 
   if (validate_table) {
     // check access privileges
     const TableDescriptor* td = catalog.getMetadataForTable(table_name_);
 
     if (td == nullptr) {
       throw std::runtime_error("Table " + table_name_ + " does not exist.");
     }
     if (td->isView) {
       throw std::runtime_error("Insert to views is not supported yet.");
     }
 
     if (!session->checkDBAccessPrivileges(DBObjectType::TableDBObjectType,
                                           AccessPrivileges::INSERT_INTO_TABLE,
                                           table_name_)) {
       throw std::runtime_error("User has no insert privileges on " + table_name_ + ".");
     }
 
     // only validate the select query so we get the target types
     // correctly, but do not populate the result set
     auto result = local_connector.query(query_state_proxy, select_query_, true);
     auto source_column_descriptors = local_connector.getColumnDescriptors(result, false);
 
     std::vector<const ColumnDescriptor*> target_column_descriptors =
         get_target_column_descriptors(td);
     if (catalog.getAllColumnMetadataForTable(td->tableId, false, false, false).size() !=
         target_column_descriptors.size()) {
       throw std::runtime_error("Insert into a subset of columns is not supported yet.");
     }
 
     if (source_column_descriptors.size() != target_column_descriptors.size()) {
       throw std::runtime_error("The number of source and target columns does not match.");
     }
 
     for (int i = 0; i < source_column_descriptors.size(); i++) {
       const ColumnDescriptor* source_cd =
           &(*std::next(source_column_descriptors.begin(), i));
       const ColumnDescriptor* target_cd = target_column_descriptors.at(i);
 
       if ((!source_cd->columnType.is_string() && !target_cd->columnType.is_string()) &&
           (source_cd->columnType.get_type() != target_cd->columnType.get_type())) {
         throw std::runtime_error("Source '" + source_cd->columnName + " " +
                                  source_cd->columnType.get_type_name() +
                                  "' and target '" + target_cd->columnName + " " +
                                  target_cd->columnType.get_type_name() +
                                  "' column types do not match.");
       }
       if (source_cd->columnType.is_array()) {
         if (source_cd->columnType.get_subtype() != target_cd->columnType.get_subtype()) {
           throw std::runtime_error("Source '" + source_cd->columnName + " " +
                                    source_cd->columnType.get_type_name() +
                                    "' and target '" + target_cd->columnName + " " +
                                    target_cd->columnType.get_type_name() +
                                    "' array column element types do not match.");
         }
       }
 
       if (source_cd->columnType.is_decimal() ||
           source_cd->columnType.get_elem_type().is_decimal()) {
         SQLTypeInfo sourceType = source_cd->columnType;
         SQLTypeInfo targetType = target_cd->columnType;
 
         if (source_cd->columnType.is_array()) {
           sourceType = source_cd->columnType.get_elem_type();
           targetType = target_cd->columnType.get_elem_type();
         }
 
         if ((sourceType.get_dimension() != targetType.get_dimension()) ||
             (sourceType.get_scale() != targetType.get_scale())) {
           throw std::runtime_error(
               "Source '" + source_cd->columnName + " " +
               source_cd->columnType.get_type_name() + "' and target '" +
               target_cd->columnName + " " + target_cd->columnType.get_type_name() +
               "' decimal columns dimension and or scale do not match.");
         }
       }
 
       if (source_cd->columnType.is_string()) {
         if (source_cd->columnType.get_compression() !=
             target_cd->columnType.get_compression()) {
           throw std::runtime_error("Source '" + source_cd->columnName + " " +
                                    source_cd->columnType.get_type_name() +
                                    "' and target '" + target_cd->columnName + " " +
                                    target_cd->columnType.get_type_name() +
                                    "' columns string encodings do not match.");
         }
       }
 
       if (source_cd->columnType.is_timestamp() && target_cd->columnType.is_timestamp()) {
         if (source_cd->columnType.get_dimension() !=
             target_cd->columnType.get_dimension()) {
           throw std::runtime_error("Source '" + source_cd->columnName + " " +
                                    source_cd->columnType.get_type_name() +
                                    "' and target '" + target_cd->columnName + " " +
                                    target_cd->columnType.get_type_name() +
                                    "' timestamp column precisions do not match.");
         }
       }
 
       if (!source_cd->columnType.is_string() &&
           source_cd->columnType.get_size() > target_cd->columnType.get_size()) {
         throw std::runtime_error("Source '" + source_cd->columnName + " " +
                                  source_cd->columnType.get_type_name() +
                                  "' and target '" + target_cd->columnName + " " +
                                  target_cd->columnType.get_type_name() +
                                  "' column encoding sizes do not match.");
       }
     }
   }
 
   if (!populate_table) {
     return;
   }
 
   const TableDescriptor* created_td = catalog.getMetadataForTable(table_name_);
   Fragmenter_Namespace::InsertDataLoader insertDataLoader(*leafs_connector_);
   AggregatedResult res = leafs_connector_->query(query_state_proxy, select_query_);
   auto target_column_descriptors = get_target_column_descriptors(created_td);
   auto result_rows = res.rs;
   result_rows->setGeoReturnType(ResultSet::GeoReturnType::GeoTargetValue);
   const auto num_rows = result_rows->rowCount();
 
   if (0 == num_rows) {
     return;
   }
 
   size_t leaf_count = leafs_connector_->leafCount();
 
   size_t max_number_of_rows_per_package = std::min(num_rows / leaf_count, 64UL * 1024UL);
 
   size_t start_row = 0;
   size_t num_rows_to_process = std::min(num_rows, max_number_of_rows_per_package);
 
   // ensure that at least one row is being processed
   num_rows_to_process = std::max(num_rows_to_process, 1UL);
 
   std::vector<std::unique_ptr<TargetValueConverter>> value_converters;
 
   TargetValueConverterFactory factory;
 
   const int num_worker_threads = std::thread::hardware_concurrency();
 
   std::vector<size_t> thread_start_idx(num_worker_threads),
       thread_end_idx(num_worker_threads);
   bool can_go_parallel = !result_rows->isTruncated() && num_rows_to_process > 20000;
 
   std::atomic<size_t> row_idx{0};
 
   auto convert_function = [&result_rows,
                            &value_converters,
                            &row_idx,
                            &num_rows_to_process,
                            &thread_start_idx,
                            &thread_end_idx](const int thread_id) {
     const int num_cols = value_converters.size();
     const size_t start = thread_start_idx[thread_id];
     const size_t end = thread_end_idx[thread_id];
     size_t idx = 0;
     for (idx = start; idx < end; ++idx) {
       const auto result_row = result_rows->getRowAtNoTranslations(idx);
       if (!result_row.empty()) {
         size_t target_row = row_idx.fetch_add(1);
 
         if (target_row >= num_rows_to_process) {
           break;
         }
 
         for (unsigned int col = 0; col < num_cols; col++) {
           const auto& mapd_variant = result_row[col];
           value_converters[col]->convertToColumnarFormat(target_row, &mapd_variant);
         }
       }
     }
 
     thread_start_idx[thread_id] = idx;
   };
 
   auto single_threaded_convert_function = [&result_rows,
                                            &value_converters,
                                            &row_idx,
                                            &num_rows_to_process,
                                            &thread_start_idx,
                                            &thread_end_idx](const int thread_id) {
     const int num_cols = value_converters.size();
     const size_t start = thread_start_idx[thread_id];
     const size_t end = thread_end_idx[thread_id];
     size_t idx = 0;
     for (idx = start; idx < end; ++idx) {
       size_t target_row = row_idx.fetch_add(1);
 
       if (target_row >= num_rows_to_process) {
         break;
       }
       const auto result_row = result_rows->getNextRow(false, false);
       CHECK(!result_row.empty());
       for (unsigned int col = 0; col < num_cols; col++) {
         const auto& mapd_variant = result_row[col];
         value_converters[col]->convertToColumnarFormat(target_row, &mapd_variant);
       }
     }
 
     thread_start_idx[thread_id] = idx;
   };
 
   if (can_go_parallel) {
     const size_t entryCount = result_rows->entryCount();
     for (size_t i = 0,
                 start_entry = 0,
                 stride = (entryCount + num_worker_threads - 1) / num_worker_threads;
          i < num_worker_threads && start_entry < entryCount;
          ++i, start_entry += stride) {
       const auto end_entry = std::min(start_entry + stride, entryCount);
       thread_start_idx[i] = start_entry;
       thread_end_idx[i] = end_entry;
     }
 
   } else {
     thread_start_idx[0] = 0;
     thread_end_idx[0] = result_rows->entryCount();
   }
 
   std::shared_ptr<Executor> executor;
 
   if (g_enable_experimental_string_functions) {
     executor = Executor::getExecutor(catalog.getCurrentDB().dbId);
   }
 
   while (start_row < num_rows) {
     try {
       value_converters.clear();
       row_idx = 0;
       int colNum = 0;
       for (const auto targetDescriptor : target_column_descriptors) {
         auto sourceDataMetaInfo = res.targets_meta[colNum++];
 
         ConverterCreateParameter param{
             num_rows_to_process,
             catalog,
             sourceDataMetaInfo,
             targetDescriptor,
             targetDescriptor->columnType,
             !targetDescriptor->columnType.get_notnull(),
             result_rows->getRowSetMemOwner()->getLiteralStringDictProxy(),
             g_enable_experimental_string_functions
                 ? executor->getStringDictionaryProxy(
                       sourceDataMetaInfo.get_type_info().get_comp_param(),
                       result_rows->getRowSetMemOwner(),
                       true)
                 : nullptr};
         auto converter = factory.create(param);
         value_converters.push_back(std::move(converter));
       }
 
       if (can_go_parallel) {
         std::vector<std::future<void>> worker_threads;
         for (int i = 0; i < num_worker_threads; ++i) {
           worker_threads.push_back(std::async(std::launch::async, convert_function, i));
         }
 
         for (auto& child : worker_threads) {
           child.wait();
         }
         for (auto& child : worker_threads) {
           child.get();
         }
 
       } else {
         single_threaded_convert_function(0);
       }
 
       // finalize the insert data
       {
         auto finalizer_func =
             [](std::unique_ptr<TargetValueConverter>::pointer targetValueConverter) {
               targetValueConverter->finalizeDataBlocksForInsertData();
             };
         std::vector<std::future<void>> worker_threads;
         for (auto& converterPtr : value_converters) {
           worker_threads.push_back(
               std::async(std::launch::async, finalizer_func, converterPtr.get()));
         }
 
         for (auto& child : worker_threads) {
           child.wait();
         }
         for (auto& child : worker_threads) {
           child.get();
         }
       }
 
       Fragmenter_Namespace::InsertData insert_data;
       insert_data.databaseId = catalog.getCurrentDB().dbId;
       CHECK(created_td);
       insert_data.tableId = created_td->tableId;
       insert_data.numRows = num_rows_to_process;
 
       for (int col_idx = 0; col_idx < target_column_descriptors.size(); col_idx++) {
         value_converters[col_idx]->addDataBlocksToInsertData(insert_data);
       }
 
       insertDataLoader.insertData(*session, insert_data);
     } catch (...) {
       try {
         if (created_td->nShards) {
           const auto shard_tables = catalog.getPhysicalTablesDescriptors(created_td);
           for (const auto ptd : shard_tables) {
             leafs_connector_->rollback(*session, ptd->tableId);
           }
         }
         leafs_connector_->rollback(*session, created_td->tableId);
       } catch (...) {
         // eat it
       }
       throw;
     }
     start_row += num_rows_to_process;
     num_rows_to_process = std::min(num_rows - start_row, max_number_of_rows_per_package);
   }
 
   if (!is_temporary) {
     if (created_td->nShards) {
       const auto shard_tables = catalog.getPhysicalTablesDescriptors(created_td);
       for (const auto ptd : shard_tables) {
         leafs_connector_->checkpoint(*session, ptd->tableId);
       }
     }
     leafs_connector_->checkpoint(*session, created_td->tableId);
   }
 }

Here is the call graph for this function:

Member Data Documentation

std::vector<std::unique_ptr<std::string> > Parser::InsertIntoTableAsSelectStmt::column_list_

protected

Definition at line 1060 of file ParserNode.h.

Referenced by InsertIntoTableAsSelectStmt().

DistributedConnector* Parser::InsertIntoTableAsSelectStmt::leafs_connector_ = nullptr

Definition at line 1057 of file ParserNode.h.

std::string Parser::InsertIntoTableAsSelectStmt::select_query_

protected

Definition at line 1062 of file ParserNode.h.

Referenced by get_select_query().

std::string Parser::InsertIntoTableAsSelectStmt::table_name_

protected

Definition at line 1061 of file ParserNode.h.

Referenced by get_table().

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

/home/jenkins-slave/workspace/core-os-doxygen/Parser/ParserNode.h
/home/jenkins-slave/workspace/core-os-doxygen/Parser/ParserNode.cpp

Classes

Public Member Functions

Public Attributes

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation