Parser::InsertIntoTableAsSelectStmt Class Reference

#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

InsertIntoTableAsSelectStmt()

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.

      : 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

execute()

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

overridevirtual

Implements Parser::DDLStmt.

Reimplemented in Parser::CreateTableAsSelectStmt.

Definition at line 2957 of file ParserNode.cpp.

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

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

                                                                                     {
  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:

Here is the caller graph for this function:

get_select_query()

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

inline

Definition at line 1095 of file ParserNode.h.

{ return select_query_; }

get_table()

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

inline

Definition at line 1093 of file ParserNode.h.

{ return table_name_; }

populateData()

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

Definition at line 2534 of file ParserNode.cpp.

References CHECK, ColumnDescriptor::columnName, ColumnDescriptor::columnType, TargetValueConverterFactory::create(), Fragmenter_Namespace::InsertData::databaseId, parse_ast::end, logger::ERROR, 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(), LOG, 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;

  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);
      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) {
          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);

          start_row += num_rows_to_process;
          num_rows_to_process =
              std::min(num_rows - start_row, max_number_of_rows_per_package);
        }
      }
    }
  } 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:

Member Data Documentation

column_list_

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

protected

Definition at line 1100 of file ParserNode.h.

leafs_connector_

DistributedConnector* Parser::InsertIntoTableAsSelectStmt::leafs_connector_ = nullptr

Definition at line 1097 of file ParserNode.h.

select_query_

std::string Parser::InsertIntoTableAsSelectStmt::select_query_

protected

Definition at line 1102 of file ParserNode.h.

table_name_

std::string Parser::InsertIntoTableAsSelectStmt::table_name_

protected

Definition at line 1101 of file ParserNode.h.

---

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