#include <ParserNode.h>

Inheritance diagram for Parser::InsertIntoTableAsSelectStmt:

Collaboration diagram for Parser::InsertIntoTableAsSelectStmt:

Public Member Functions
	InsertIntoTableAsSelectStmt (const rapidjson::Value &payload)

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

void	populateData (QueryStateProxy, const TableDescriptor *td, bool validate_table, bool for_CTAS=false)

void	execute (const Catalog_Namespace::SessionInfo &session, bool read_only_mode) 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
std::unique_ptr< QueryConnector >	leafs_connector_

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

std::string	table_name_

std::string	select_query_

Detailed Description

Definition at line 1126 of file ParserNode.h.

Constructor & Destructor Documentation

Parser::InsertIntoTableAsSelectStmt::InsertIntoTableAsSelectStmt ( const rapidjson::Value & payload )

Definition at line 3594 of file ParserNode.cpp.

References CHECK, column_list_, json_str(), select_query_, and table_name_.

                                    {
   CHECK(payload.HasMember("name"));
   table_name_ = json_str(payload["name"]);
 
   CHECK(payload.HasMember("query"));
   select_query_ = json_str(payload["query"]);
 
   boost::replace_all(select_query_, "\n", " ");
   select_query_ = "(" + select_query_ + ")";
 
   if (payload.HasMember("columns")) {
     CHECK(payload["columns"].IsArray());
     for (auto& column : payload["columns"].GetArray()) {
       std::string s = json_str(column);
       column_list_.emplace_back(std::unique_ptr<std::string>(new std::string(s)));
     }
   }
 }

Here is the call graph for this function:

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

inline

Definition at line 1130 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,
		bool	read_only_mode
	)

overridevirtual

Implements Parser::DDLStmt.

Reimplemented in Parser::CreateTableAsSelectStmt.

Definition at line 4180 of file ParserNode.cpp.

References Parser::anonymous_namespace{ParserNode.cpp}::acquire_query_table_locks(), Executor::clearExternalCaches(), query_state::QueryState::create(), Catalog_Namespace::DBMetadata::dbId, legacylockmgr::ExecutorOuterLock, Catalog_Namespace::Catalog::getCurrentDB(), Catalog_Namespace::Catalog::getMetadataForTable(), legacylockmgr::LockMgr< MutexType, KeyType >::getMutex(), populateData(), select_query_, STDLOG, and table_name_.

Referenced by heavydb.cursor.Cursor::executemany().

                                                                {
   if (read_only_mode) {
     throw std::runtime_error("INSERT INTO TABLE invalid in read only mode.");
   }
   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);
   auto& catalog = session_ptr->getCatalog();
 
   const auto execute_read_lock =
       heavyai::shared_lock<legacylockmgr::WrapperType<heavyai::shared_mutex>>(
           *legacylockmgr::LockMgr<heavyai::shared_mutex, bool>::getMutex(
               legacylockmgr::ExecutorOuterLock, true));
 
   if (catalog.getMetadataForTable(table_name_) == nullptr) {
     throw std::runtime_error("ITAS failed: table " + table_name_ + " does not exist.");
   }
 
   auto locks = acquire_query_table_locks(
       catalog, select_query_, query_state->createQueryStateProxy(), table_name_);
   const TableDescriptor* td = catalog.getMetadataForTable(table_name_);
 
   Executor::clearExternalCaches(true, td, catalog.getCurrentDB().dbId);
 
   try {
     populateData(query_state->createQueryStateProxy(), td, true, false);
   } catch (...) {
     throw;
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

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

inline

Definition at line 1154 of file ParserNode.h.

References select_query_.

1154 { return select_query_; }

Parser::InsertIntoTableAsSelectStmt::select_query_

std::string select_query_

Definition: ParserNode.h:1161

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

inline

Definition at line 1152 of file ParserNode.h.

References table_name_.

1152 { return table_name_; }

Parser::InsertIntoTableAsSelectStmt::table_name_

std::string table_name_

Definition: ParserNode.h:1160

void Parser::InsertIntoTableAsSelectStmt::populateData	(	QueryStateProxy	query_state_proxy,
		const TableDescriptor *	td,
		bool	validate_table,
		bool	for_CTAS = `false`
	)

Definition at line 3614 of file ParserNode.cpp.

References threading_serial::async(), CHECK, Parser::check_session_interrupted(), column_list_, ColumnDescriptor::columnName, ColumnDescriptor::columnType, TargetValueConverterFactory::create(), Fragmenter_Namespace::InsertData::databaseId, logger::ERROR, import_export::fill_missing_columns(), g_cluster, g_enable_assign_render_groups, g_enable_non_kernel_time_query_interrupt, g_enable_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::LocalQueryConnector::getColumnDescriptors(), query_state::QueryState::getConstSessionInfo(), Executor::getExecutor(), query_state::QueryStateProxy::getQueryState(), query_state::QueryState::getQuerySubmittedTime(), AccessPrivileges::INSERT_INTO_TABLE, Fragmenter_Namespace::InsertDataLoader::insertData(), SQLTypeInfo::is_array(), SQLTypeInfo::is_date(), SQLTypeInfo::is_decimal(), IS_GEO_POLY, SQLTypeInfo::is_geometry(), SQLTypeInfo::is_integer(), SQLTypeInfo::is_string(), SQLTypeInfo::is_time(), SQLTypeInfo::is_timestamp(), TableDescriptor::isView, leafs_connector_, LOG, Fragmenter_Namespace::InsertData::numRows, Parser::LocalQueryConnector::query(), run_benchmark_import::res, run_benchmark_import::result, select_query_, run_benchmark_import::start_time, table_is_temporary(), table_name_, TableDBObjectType, TableDescriptor::tableId, Fragmenter_Namespace::InsertData::tableId, logger::thread_id(), timer_start(), timer_stop(), Executor::UNITARY_EXECUTOR_ID, UNLIKELY, foreign_storage::validate_non_foreign_table_write(), and VLOG.

Referenced by execute(), and Parser::CreateTableAsSelectStmt::execute().

                                                               {
   auto const session = query_state_proxy.getQueryState().getConstSessionInfo();
   auto& catalog = session->getCatalog();
   foreign_storage::validate_non_foreign_table_write(td);
   bool populate_table = false;
 
   if (leafs_connector_) {
     populate_table = true;
   } else {
     leafs_connector_ = std::make_unique<LocalQueryConnector>();
     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);
 
   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
     LocalQueryConnector local_connector;
     auto result = local_connector.query(query_state_proxy, select_query_, {}, true, true);
     auto source_column_descriptors = local_connector.getColumnDescriptors(result, false);
 
     std::vector<const ColumnDescriptor*> target_column_descriptors =
         get_target_column_descriptors(td);
 
     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 (target_cd->columnType.is_string() && !source_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.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;
   auto query_session = session ? session->get_session_id() : "";
   auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID).get();
   std::string work_type_str = for_CTAS ? "CTAS" : "ITAS";
   try {
     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,
                                   g_enable_non_kernel_time_query_interrupt);
       total_source_query_time_ms += timer_stop(query_clock_begin);
 
       auto start_time = query_state_proxy.getQueryState().getQuerySubmittedTime();
       auto query_str = "INSERT_DATA for " + work_type_str;
       if (g_enable_non_kernel_time_query_interrupt) {
         // In the clean-up phase of the query execution for collecting aggregated result
         // of SELECT query, we remove its query session info, so we need to enroll the
         // session info again
         executor->enrollQuerySession(query_session,
                                      query_str,
                                      start_time,
                                      Executor::UNITARY_EXECUTOR_ID,
                                      QuerySessionStatus::QueryStatus::RUNNING_IMPORTER);
       }
 
       ScopeGuard clearInterruptStatus = [executor, &query_session, &start_time] {
         // this data population is non-kernel operation, so we manually cleanup
         // the query session info in the cleanup phase
         if (g_enable_non_kernel_time_query_interrupt) {
           executor->clearQuerySessionStatus(query_session, start_time);
         }
       };
 
       for (auto& res : query_results) {
         if (UNLIKELY(check_session_interrupted(query_session, executor))) {
           throw std::runtime_error(
               "Query execution has been interrupted while performing " + work_type_str);
         }
         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();
 
         // ensure that at least 1 row is processed per block up to a maximum of 65536 rows
         const size_t rows_per_block =
             std::max(std::min(num_rows / leaf_count, size_t(64 * 1024)), size_t(1));
 
         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() && rows_per_block > 20000;
 
         std::atomic<size_t> crt_row_idx{0};
 
         auto do_work = [&result_rows, &value_converters, &crt_row_idx](
                            const size_t idx,
                            const size_t block_end,
                            const size_t num_cols,
                            const size_t thread_id,
                            bool& stop_convert) {
           const auto result_row = result_rows->getRowAtNoTranslations(idx);
           if (!result_row.empty()) {
             size_t target_row = crt_row_idx.fetch_add(1);
             if (target_row >= block_end) {
               stop_convert = true;
               return;
             }
             for (unsigned int col = 0; col < num_cols; col++) {
               const auto& mapd_variant = result_row[col];
               value_converters[col]->convertToColumnarFormat(target_row, &mapd_variant);
             }
           }
         };
 
         auto convert_function = [&thread_start_idx,
                                  &thread_end_idx,
                                  &value_converters,
                                  &executor,
                                  &query_session,
                                  &work_type_str,
                                  &do_work](const int thread_id, const size_t block_end) {
           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;
           bool stop_convert = false;
           if (g_enable_non_kernel_time_query_interrupt) {
             size_t local_idx = 0;
             for (idx = start; idx < end; ++idx, ++local_idx) {
               if (UNLIKELY((local_idx & 0xFFFF) == 0 &&
                            check_session_interrupted(query_session, executor))) {
                 throw std::runtime_error(
                     "Query execution has been interrupted while performing " +
                     work_type_str);
               }
               do_work(idx, block_end, num_cols, thread_id, stop_convert);
               if (stop_convert) {
                 break;
               }
             }
           } else {
             for (idx = start; idx < end; ++idx) {
               do_work(idx, block_end, num_cols, thread_id, stop_convert);
               if (stop_convert) {
                 break;
               }
             }
           }
           thread_start_idx[thread_id] = idx;
         };
 
         auto single_threaded_value_converter =
             [&crt_row_idx, &value_converters, &result_rows](const size_t idx,
                                                             const size_t block_end,
                                                             const size_t num_cols,
                                                             bool& stop_convert) {
               size_t target_row = crt_row_idx.fetch_add(1);
               if (target_row >= block_end) {
                 stop_convert = true;
                 return;
               }
               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);
               }
             };
 
         auto single_threaded_convert_function = [&value_converters,
                                                  &thread_start_idx,
                                                  &thread_end_idx,
                                                  &executor,
                                                  &query_session,
                                                  &work_type_str,
                                                  &single_threaded_value_converter](
                                                     const int thread_id,
                                                     const size_t block_end) {
           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;
           bool stop_convert = false;
           if (g_enable_non_kernel_time_query_interrupt) {
             size_t local_idx = 0;
             for (idx = start; idx < end; ++idx, ++local_idx) {
               if (UNLIKELY((local_idx & 0xFFFF) == 0 &&
                            check_session_interrupted(query_session, executor))) {
                 throw std::runtime_error(
                     "Query execution has been interrupted while performing " +
                     work_type_str);
               }
               single_threaded_value_converter(idx, block_end, num_cols, stop_convert);
               if (stop_convert) {
                 break;
               }
             }
           } else {
             for (idx = start; idx < end; ++idx) {
               single_threaded_value_converter(idx, end, num_cols, stop_convert);
               if (stop_convert) {
                 break;
               }
             }
           }
           thread_start_idx[thread_id] = idx;
         };
 
         if (can_go_parallel) {
           const size_t entry_count = result_rows->entryCount();
           for (size_t
                    i = 0,
                    start_entry = 0,
                    stride = (entry_count + num_worker_threads - 1) / num_worker_threads;
                i < num_worker_threads && start_entry < entry_count;
                ++i, start_entry += stride) {
             const auto end_entry = std::min(start_entry + stride, entry_count);
             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();
         }
 
         RenderGroupAnalyzerMap render_group_analyzer_map;
 
         for (size_t block_start = 0; block_start < num_rows;
              block_start += rows_per_block) {
           const auto num_rows_this_itr = block_start + rows_per_block < num_rows
                                              ? rows_per_block
                                              : num_rows - block_start;
           crt_row_idx = 0;  // reset block tracker
           value_converters.clear();
           int colNum = 0;
           for (const auto targetDescriptor : target_column_descriptors) {
             auto sourceDataMetaInfo = res.targets_meta[colNum++];
             ConverterCreateParameter param{
                 num_rows_this_itr,
                 catalog,
                 sourceDataMetaInfo,
                 targetDescriptor,
                 targetDescriptor->columnType,
                 !targetDescriptor->columnType.get_notnull(),
                 result_rows->getRowSetMemOwner()->getLiteralStringDictProxy(),
                 g_enable_string_functions &&
                         sourceDataMetaInfo.get_type_info().is_dict_encoded_string()
                     ? executor->getStringDictionaryProxy(
                           sourceDataMetaInfo.get_type_info().get_comp_param(),
                           result_rows->getRowSetMemOwner(),
                           true)
                     : nullptr,
                 IS_GEO_POLY(targetDescriptor->columnType.get_type()) &&
                         g_enable_assign_render_groups
                     ? &render_group_analyzer_map
                     : 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, num_rows_this_itr));
             }
 
             for (auto& child : worker_threads) {
               child.wait();
             }
             for (auto& child : worker_threads) {
               child.get();
             }
 
           } else {
             single_threaded_convert_function(0, num_rows_this_itr);
           }
 
           // 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_this_itr;
 
           for (int col_idx = 0; col_idx < target_column_descriptors.size(); col_idx++) {
             if (UNLIKELY(g_enable_non_kernel_time_query_interrupt &&
                          check_session_interrupted(query_session, executor))) {
               throw std::runtime_error(
                   "Query execution has been interrupted while performing " +
                   work_type_str);
             }
             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();
           auto data_memory_holder =
               import_export::fill_missing_columns(&catalog, insert_data);
           insertDataLoader.insertData(*session, insert_data);
           total_data_load_time_ms += timer_stop(data_load_clock_begin);
         }
       }
     }
   } catch (...) {
     try {
       leafs_connector_->rollback(*session, td->tableId);
     } catch (std::exception& e) {
       LOG(ERROR) << "An error occurred during ITAS rollback attempt. Table id: "
                  << td->tableId << ", Error: " << e.what();
     }
     throw;
   }
 
   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) {
     leafs_connector_->checkpoint(*session, td->tableId);
   }
 }

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

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

protected

Definition at line 1159 of file ParserNode.h.

Referenced by InsertIntoTableAsSelectStmt(), and populateData().

std::unique_ptr<QueryConnector> Parser::InsertIntoTableAsSelectStmt::leafs_connector_

Definition at line 1156 of file ParserNode.h.

Referenced by Parser::CreateTableAsSelectStmt::execute(), and populateData().

std::string Parser::InsertIntoTableAsSelectStmt::select_query_

protected

Definition at line 1161 of file ParserNode.h.

Referenced by execute(), Parser::CreateTableAsSelectStmt::execute(), get_select_query(), InsertIntoTableAsSelectStmt(), and populateData().

std::string Parser::InsertIntoTableAsSelectStmt::table_name_

protected

Definition at line 1160 of file ParserNode.h.

Referenced by execute(), Parser::CreateTableAsSelectStmt::execute(), get_table(), InsertIntoTableAsSelectStmt(), and populateData().

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