Parser::InsertIntoTableAsSelectStmt Class Reference

#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) 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 1110 of file ParserNode.h.

Constructor & Destructor Documentation

Parser::InsertIntoTableAsSelectStmt::InsertIntoTableAsSelectStmt

(

const rapidjson::Value &

payload

)

Definition at line 2898 of file ParserNode.cpp.

References CHECK, and json_str().

                                   {
  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 1114 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 3413 of file ParserNode.cpp.

References anonymous_namespace{Utm.h}::a, query_state::QueryState::create(), legacylockmgr::ExecutorOuterLock, legacylockmgr::LockMgr< MutexType, KeyType >::getMutex(), pg_shim(), run_benchmark_import::result, gpu_enabled::sort(), STDLOG, and run_benchmark_import::tables.

Referenced by omnisci.cursor.Cursor::executemany(), and QueryRunner::QueryRunner::runSQL().

                                                                                     {
  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 = mapd_shared_lock<mapd_shared_mutex>(
      *legacylockmgr::LockMgr<mapd_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.");
  }

  lockmgr::LockedTableDescriptors locks;
  std::vector<std::string> tables;

  // get the table info
  auto calcite_mgr = catalog.getCalciteMgr();

  // TODO MAT this should actually get the global or the session parameter for
  // view optimization
  const auto result = calcite_mgr->process(query_state->createQueryStateProxy(),
                                           pg_shim(select_query_),
                                           {},
                                           true,
                                           false,
                                           false,
                                           true);

  for (auto& tab : result.resolved_accessed_objects.tables_selected_from) {
    tables.emplace_back(tab[0]);
  }
  tables.emplace_back(table_name_);

  // force sort into tableid order in case of name change to guarantee fixed order of
  // mutex access
  std::sort(tables.begin(),
            tables.end(),
            [&catalog](const std::string& a, const std::string& b) {
              return catalog.getMetadataForTable(a, false)->tableId <
                     catalog.getMetadataForTable(b, false)->tableId;
            });

  tables.erase(unique(tables.begin(), tables.end()), tables.end());
  for (const auto& table : tables) {
    locks.emplace_back(
        std::make_unique<lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>>(
            lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>::acquireTableDescriptor(
                catalog, table)));
    if (table == table_name_) {
      // Aquire an insert data lock for updates/deletes, consistent w/ insert. The
      // table data lock will be aquired in the fragmenter during checkpoint.
      locks.emplace_back(
          std::make_unique<lockmgr::TableInsertLockContainer<lockmgr::WriteLock>>(
              lockmgr::TableInsertLockContainer<lockmgr::WriteLock>::acquire(
                  catalog.getDatabaseId(), (*locks.back())())));
    } else {
      locks.emplace_back(
          std::make_unique<lockmgr::TableDataLockContainer<lockmgr::ReadLock>>(
              lockmgr::TableDataLockContainer<lockmgr::ReadLock>::acquire(
                  catalog.getDatabaseId(), (*locks.back())())));
    }
  }

  const TableDescriptor* td = catalog.getMetadataForTable(table_name_);
  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 1137 of file ParserNode.h.

References select_query_.

{ return select_query_; }

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

inline

Definition at line 1135 of file ParserNode.h.

References table_name_.

{ return table_name_; }

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

Definition at line 2918 of file ParserNode.cpp.

References threading_serial::async(), CHECK, anonymous_namespace{Importer.cpp}::check_session_interrupted(), ColumnDescriptor::columnName, ColumnDescriptor::columnType, TargetValueConverterFactory::create(), Fragmenter_Namespace::InsertData::databaseId, logger::ERROR, import_export::fill_missing_columns(), g_cluster, g_enable_experimental_string_functions, g_enable_non_kernel_time_query_interrupt, 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(), query_state::QueryState::getQuerySubmittedTime(), i, 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(), TableDescriptor::isView, LOG, Fragmenter_Namespace::InsertData::numRows, Parser::LocalConnector::query(), run_benchmark_import::res, run_benchmark_import::result, run_benchmark_import::start_time, table_is_temporary(), 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 Parser::CreateTableAsSelectStmt::execute().

                                                              {
  auto const session = query_state_proxy.getQueryState().getConstSessionInfo();
  auto& catalog = session->getCatalog();
  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);

  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, 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 (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();
        }

        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_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, 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 1142 of file ParserNode.h.

Referenced by InsertIntoTableAsSelectStmt().

DistributedConnector* Parser::InsertIntoTableAsSelectStmt::leafs_connector_ = nullptr

Definition at line 1139 of file ParserNode.h.

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

std::string Parser::InsertIntoTableAsSelectStmt::select_query_

protected

Definition at line 1144 of file ParserNode.h.

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

std::string Parser::InsertIntoTableAsSelectStmt::table_name_

protected

Definition at line 1143 of file ParserNode.h.

Referenced by Parser::CreateTableAsSelectStmt::execute(), and 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