#include <ParserNode.h>

Inheritance diagram for Parser::InsertIntoTableAsSelectStmt:

Collaboration diagram for Parser::InsertIntoTableAsSelectStmt:

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

void	populateData (QueryStateProxy, 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 1072 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 1075 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 2969 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(), true);
 }

Here is the call graph for this function:

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

inline

Definition at line 1095 of file ParserNode.h.

References select_query_.

1095 { return select_query_; }

Parser::InsertIntoTableAsSelectStmt::select_query_

std::string select_query_

Definition: ParserNode.h:1102

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

inline

Definition at line 1093 of file ParserNode.h.

References table_name_.

1093 { return table_name_; }

Parser::InsertIntoTableAsSelectStmt::table_name_

std::string table_name_

Definition: ParserNode.h:1101

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

Definition at line 2535 of file ParserNode.cpp.

References CHECK(), ColumnDescriptor::columnName, ColumnDescriptor::columnType, TargetValueConverterFactory::create(), Fragmenter_Namespace::InsertData::databaseId, g_cluster, g_enable_experimental_string_functions, ResultSet::GeoTargetValue, SQLTypeInfo::get_compression(), SQLTypeInfo::get_dimension(), SQLTypeInfo::get_elem_type(), SQLTypeInfo::get_scale(), SQLTypeInfo::get_size(), SQLTypeInfo::get_subtype(), SQLTypeInfo::get_type(), SQLTypeInfo::get_type_name(), Parser::LocalConnector::getColumnDescriptors(), query_state::QueryState::getConstSessionInfo(), Executor::getExecutor(), query_state::QueryStateProxy::getQueryState(), lockmgr::TableLockMgrImpl< InsertDataLockMgr >::getWriteLockForTable(), AccessPrivileges::INSERT_INTO_TABLE, Fragmenter_Namespace::InsertDataLoader::insertData(), SQLTypeInfo::is_array(), SQLTypeInfo::is_date(), SQLTypeInfo::is_decimal(), SQLTypeInfo::is_geometry(), SQLTypeInfo::is_integer(), SQLTypeInfo::is_string(), SQLTypeInfo::is_time(), SQLTypeInfo::is_timestamp(), num_rows, Fragmenter_Namespace::InsertData::numRows, Parser::LocalConnector::query(), run_benchmark_import::res, run_benchmark_import::result, table_is_temporary(), TableDBObjectType, Fragmenter_Namespace::InsertData::tableId, logger::thread_id(), timer_start(), timer_stop(), Executor::UNITARY_EXECUTOR_ID, foreign_storage::validate_non_foreign_table_write(), and VLOG.

                                                                     {
   auto const session = query_state_proxy.getQueryState().getConstSessionInfo();
   auto& catalog = session->getCatalog();
   const auto td_with_lock =
       lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>::acquireTableDescriptor(
           catalog, table_name_);
   const auto td = td_with_lock();
   foreign_storage::validate_non_foreign_table_write(td);
 
   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 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;
   };
 
   bool is_temporary = table_is_temporary(td);
 
   // Don't allow simultaneous inserts
   const auto insert_data_lock =
       lockmgr::InsertDataLockMgr::getWriteLockForTable(catalog, table_name_);
 
   if (validate_table) {
     // check access privileges
     if (!td) {
       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.get_type() != target_cd->columnType.get_type()) {
         auto type_cannot_be_cast = [](const auto& col_type) {
           return (col_type.is_time() || col_type.is_geometry() || col_type.is_array() ||
                   col_type.is_boolean());
         };
 
         if (type_cannot_be_cast(source_cd->columnType) ||
             type_cannot_be_cast(target_cd->columnType)) {
           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_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 scales do not match.");
         }
       }
 
       if (source_cd->columnType.is_string()) {
         if (!target_cd->columnType.is_string()) {
           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.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.is_geometry() &&
           !source_cd->columnType.is_integer() && !source_cd->columnType.is_decimal() &&
           !source_cd->columnType.is_date() && !source_cd->columnType.is_time() &&
           !source_cd->columnType.is_timestamp() &&
           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;
   }
 
   int64_t total_row_count = 0;
   int64_t total_source_query_time_ms = 0;
   int64_t total_target_value_translate_time_ms = 0;
   int64_t total_data_load_time_ms = 0;
 
   Fragmenter_Namespace::InsertDataLoader insertDataLoader(*leafs_connector_);
   auto target_column_descriptors = get_target_column_descriptors(td);
 
   auto outer_frag_count =
       leafs_connector_->getOuterFragmentCount(query_state_proxy, select_query_);
 
   size_t outer_frag_end = outer_frag_count == 0 ? 1 : outer_frag_count;
 
   for (size_t outer_frag_idx = 0; outer_frag_idx < outer_frag_end; outer_frag_idx++) {
     std::vector<size_t> allowed_outer_fragment_indices;
 
     if (outer_frag_count) {
       allowed_outer_fragment_indices.push_back(outer_frag_idx);
     }
 
     const auto query_clock_begin = timer_start();
     std::vector<AggregatedResult> query_results = leafs_connector_->query(
         query_state_proxy, select_query_, allowed_outer_fragment_indices);
     total_source_query_time_ms += timer_stop(query_clock_begin);
 
     for (auto& res : query_results) {
       auto result_rows = res.rs;
       result_rows->setGeoReturnType(ResultSet::GeoReturnType::GeoTargetValue);
       const auto num_rows = result_rows->rowCount();
 
       if (0 == num_rows) {
         continue;
       }
 
       total_row_count += num_rows;
 
       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(Executor::UNITARY_EXECUTOR_ID);
       }
 
       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));
           }
 
           const auto translate_clock_begin = timer_start();
           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(td);
           insert_data.tableId = 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);
           }
           total_target_value_translate_time_ms += timer_stop(translate_clock_begin);
 
           const auto data_load_clock_begin = timer_start();
           insertDataLoader.insertData(*session, insert_data);
           total_data_load_time_ms += timer_stop(data_load_clock_begin);
 
         } catch (...) {
           try {
             if (td->nShards) {
               const auto shard_tables = catalog.getPhysicalTablesDescriptors(td);
               for (const auto ptd : shard_tables) {
                 leafs_connector_->rollback(*session, ptd->tableId);
               }
             }
             leafs_connector_->rollback(*session, 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);
       }
     }
   }
 
   int64_t total_time_ms = total_source_query_time_ms +
                           total_target_value_translate_time_ms + total_data_load_time_ms;
 
   VLOG(1) << "CTAS/ITAS " << total_row_count << " rows loaded in " << total_time_ms
           << "ms (outer_frag_count=" << outer_frag_count
           << ", query_time=" << total_source_query_time_ms
           << "ms, translation_time=" << total_target_value_translate_time_ms
           << "ms, data_load_time=" << total_data_load_time_ms
           << "ms)\nquery: " << select_query_;
 
   if (!is_temporary) {
     if (td->nShards) {
       const auto shard_tables = catalog.getPhysicalTablesDescriptors(td);
       for (const auto ptd : shard_tables) {
         leafs_connector_->checkpoint(*session, ptd->tableId);
       }
     }
     leafs_connector_->checkpoint(*session, 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 1100 of file ParserNode.h.

Referenced by InsertIntoTableAsSelectStmt().

DistributedConnector* Parser::InsertIntoTableAsSelectStmt::leafs_connector_ = nullptr

Definition at line 1097 of file ParserNode.h.

std::string Parser::InsertIntoTableAsSelectStmt::select_query_

protected

Definition at line 1102 of file ParserNode.h.

Referenced by get_select_query().

std::string Parser::InsertIntoTableAsSelectStmt::table_name_

protected

Definition at line 1101 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

Public Member Functions

Public Attributes

Protected Attributes

Detailed Description

Constructor & Destructor Documentation

Member Function Documentation

Member Data Documentation