Parser::CopyTableStmt Class Reference

#include <ParserNode.h>

Inheritance diagram for Parser::CopyTableStmt:

Collaboration diagram for Parser::CopyTableStmt:

Public Member Functions
	CopyTableStmt (std::string *t, std::string *f, std::list< NameValueAssign * > *o)
	CopyTableStmt (const rapidjson::Value &payload)
void	execute (const Catalog_Namespace::SessionInfo &session, bool read_only_mode) override
void	execute (const Catalog_Namespace::SessionInfo &session, bool read_only_mode, const std::function< std::unique_ptr< import_export::AbstractImporter >(Catalog_Namespace::Catalog &, const TableDescriptor *, const std::string &, const import_export::CopyParams &)> &importer_factory)
std::string &	get_table () const
bool	get_success () const
bool	was_deferred_copy_from () const
void	get_deferred_copy_from_payload (std::string &table, std::string &file_name, import_export::CopyParams &copy_params, std::string &partitions)
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< std::string >	return_message

Private Attributes
std::unique_ptr< std::string >	table_
std::unique_ptr< std::string >	copy_from_source_pattern_
bool	success_
std::list< std::unique_ptr < NameValueAssign > >	options_
bool	was_deferred_copy_from_ = false
std::string	deferred_copy_from_file_name_
import_export::CopyParams	deferred_copy_from_copy_params_
std::string	deferred_copy_from_partitions_

Detailed Description

Definition at line 1473 of file ParserNode.h.

Constructor & Destructor Documentation

Parser::CopyTableStmt::CopyTableStmt	(	std::string *	t,
		std::string *	f,
		std::list< NameValueAssign * > *	o
	)

Definition at line 5394 of file ParserNode.cpp.

References options_.

    : table_(t), copy_from_source_pattern_(f), success_(true) {
  if (o) {
    for (const auto e : *o) {
      options_.emplace_back(e);
    }
    delete o;
  }
}

Parser::CopyTableStmt::CopyTableStmt

(

const rapidjson::Value &

payload

)

Definition at line 5406 of file ParserNode.cpp.

References CHECK, copy_from_source_pattern_, json_str(), options_, Parser::anonymous_namespace{ParserNode.cpp}::parse_options(), and table_.

                                                          : success_(true) {
  CHECK(payload.HasMember("table"));
  table_ = std::make_unique<std::string>(json_str(payload["table"]));

  CHECK(payload.HasMember("filePath"));
  std::string fs = json_str(payload["filePath"]);
  // strip leading/trailing spaces/quotes/single quotes
  boost::algorithm::trim_if(fs, boost::is_any_of(" \"'`"));
  copy_from_source_pattern_ = std::make_unique<std::string>(fs);

  parse_options(payload, options_);
}

Here is the call graph for this function:

Member Function Documentation

void Parser::CopyTableStmt::execute ( const Catalog_Namespace::SessionInfo &  session, bool  read_only_mode  )

overridevirtual

Implements Parser::DDLStmt.

Definition at line 5419 of file ParserNode.cpp.

References import_export::create_importer().

Referenced by heavydb.cursor.Cursor::executemany(), and QueryRunner::QueryRunner::runImport().

                                                 {
  if (read_only_mode) {
    throw std::runtime_error("IMPORT invalid in read only mode.");
  }
  auto importer_factory = [](Catalog_Namespace::Catalog& catalog,
                             const TableDescriptor* td,
                             const std::string& copy_from_source,
                             const import_export::CopyParams& copy_params)
      -> std::unique_ptr<import_export::AbstractImporter> {
    return import_export::create_importer(catalog, td, copy_from_source, copy_params);
  };
  return execute(session, read_only_mode, importer_factory);
}

Here is the call graph for this function:

Here is the caller graph for this function:

void Parser::CopyTableStmt::execute	(	const Catalog_Namespace::SessionInfo &	session,
		bool	read_only_mode,
		const std::function< std::unique_ptr< import_export::AbstractImporter >(Catalog_Namespace::Catalog &, const TableDescriptor *, const std::string &, const import_export::CopyParams &)> &	importer_factory
	)

Definition at line 5434 of file ParserNode.cpp.

References CHECK, Executor::clearExternalCaches(), copy_from_source_pattern_, deferred_copy_from_copy_params_, deferred_copy_from_file_name_, deferred_copy_from_partitions_, logger::ERROR, measure< TimeT >::execution(), legacylockmgr::ExecutorOuterLock, g_enable_non_kernel_time_query_interrupt, Catalog_Namespace::SessionInfo::get_currentUser(), Catalog_Namespace::SessionInfo::get_session_id(), Catalog_Namespace::SessionInfo::getCatalog(), Executor::getExecutor(), legacylockmgr::LockMgr< MutexType, KeyType >::getMutex(), lockmgr::TableLockMgrImpl< T >::getWriteLockForTable(), ddl_utils::IMPORT, logger::INFO, AccessPrivileges::INSERT_INTO_TABLE, import_export::kGeoFile, import_export::kOdbc, import_export::kRasterFile, import_export::ImportStatus::load_failed, import_export::ImportStatus::load_msg, DBObject::loadKey(), LOG, import_export::CopyParams::max_reject, options_, Parser::anonymous_namespace{ParserNode.cpp}::parse_copy_params(), return_message, import_export::ImportStatus::rows_completed, import_export::ImportStatus::rows_rejected, DBObject::setPrivileges(), import_export::CopyParams::source_type, import_export::CopyParams::sql_order_by, import_export::CopyParams::sql_select, run_benchmark_import::start_time, success_, table_, TableDBObjectType, TableDescriptor::tableName, to_string(), toString(), Executor::UNITARY_EXECUTOR_ID, Catalog_Namespace::UserMetadata::userLoggable(), ddl_utils::validate_allowed_file_path(), and was_deferred_copy_from_.

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

                                                            {
  if (read_only_mode) {
    throw std::runtime_error("COPY FROM invalid in read only mode.");
  }

  size_t total_time = 0;

  // Prevent simultaneous import / truncate (see TruncateTableStmt::execute)
  const auto execute_read_lock =
      heavyai::shared_lock<legacylockmgr::WrapperType<heavyai::shared_mutex>>(
          *legacylockmgr::LockMgr<heavyai::shared_mutex, bool>::getMutex(
              legacylockmgr::ExecutorOuterLock, true));

  const TableDescriptor* td{nullptr};
  std::unique_ptr<lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>> td_with_lock;
  std::unique_ptr<lockmgr::WriteLock> insert_data_lock;

  auto& catalog = session.getCatalog();

  try {
    td_with_lock = std::make_unique<lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>>(
        lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>::acquireTableDescriptor(
            catalog, *table_));
    td = (*td_with_lock)();
    insert_data_lock = std::make_unique<lockmgr::WriteLock>(
        lockmgr::InsertDataLockMgr::getWriteLockForTable(catalog, *table_));
  } catch (const std::runtime_error& e) {
    // noop
    // TODO(adb): We're really only interested in whether the table exists or not.
    // Create a more refined exception.
  }

  // if the table already exists, it's locked, so check access privileges
  if (td) {
    std::vector<DBObject> privObjects;
    DBObject dbObject(*table_, TableDBObjectType);
    dbObject.loadKey(catalog);
    dbObject.setPrivileges(AccessPrivileges::INSERT_INTO_TABLE);
    privObjects.push_back(dbObject);
    if (!SysCatalog::instance().checkPrivileges(session.get_currentUser(), privObjects)) {
      throw std::runtime_error("Violation of access privileges: user " +
                               session.get_currentUser().userLoggable() +
                               " has no insert privileges for table " + *table_ + ".");
    }

    // invalidate cached item
    Executor::clearExternalCaches(true, td, catalog.getCurrentDB().dbId);
  }

  import_export::CopyParams copy_params;
  std::vector<std::string> warnings;
  parse_copy_params(options_, copy_params, warnings, deferred_copy_from_partitions_);

  boost::regex non_local_file_regex{R"(^\s*(s3|http|https)://.+)",
                                    boost::regex::extended | boost::regex::icase};
  if (!boost::regex_match(*copy_from_source_pattern_, non_local_file_regex) &&
      copy_params.source_type != import_export::SourceType::kOdbc) {
    ddl_utils::validate_allowed_file_path(
        *copy_from_source_pattern_, ddl_utils::DataTransferType::IMPORT, true);
  }
  // since we'll have not only posix file names but also s3/hdfs/... url
  // we do not expand wildcard or check file existence here.
  // from here on, copy_from_source contains something which may be a url
  // a wildcard of file names, or a sql select statement;
  std::string copy_from_source = *copy_from_source_pattern_;

  if (copy_params.source_type == import_export::SourceType::kOdbc) {
    copy_params.sql_select = copy_from_source;
    if (copy_params.sql_order_by.empty()) {
      throw std::runtime_error(
          "Option \"SQL ORDER BY\" must be specified when copying from an ODBC source.");
    }
  }

  std::string tr;

  for (auto const& warning : warnings) {
    tr += warning + "\n";
  }

  if (copy_params.source_type == import_export::SourceType::kGeoFile ||
      copy_params.source_type == import_export::SourceType::kRasterFile) {
    // geo import
    // we do nothing here, except stash the parameters so we can
    // do the import when we unwind to the top of the handler
    deferred_copy_from_file_name_ = copy_from_source;
    deferred_copy_from_copy_params_ = copy_params;
    was_deferred_copy_from_ = true;

    // the result string
    // @TODO simon.eves put something more useful in here
    // except we really can't because we haven't done the import yet!
    if (td) {
      tr += std::string("Appending geo to table '") + *table_ + std::string("'...");
    } else {
      tr += std::string("Creating table '") + *table_ +
            std::string("' and importing geo...");
    }
  } else {
    if (td) {
      CHECK(td_with_lock);

      // regular import
      auto importer = importer_factory(catalog, td, copy_from_source, copy_params);
      auto start_time = ::toString(std::chrono::system_clock::now());
      auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID).get();
      auto query_session = session.get_session_id();
      auto query_str = "COPYING " + td->tableName;
      if (g_enable_non_kernel_time_query_interrupt) {
        executor->enrollQuerySession(query_session,
                                     query_str,
                                     start_time,
                                     Executor::UNITARY_EXECUTOR_ID,
                                     QuerySessionStatus::QueryStatus::RUNNING_IMPORTER);
      }

      ScopeGuard clearInterruptStatus =
          [executor, &query_str, &query_session, &start_time, &importer] {
            // reset the runtime query interrupt status
            if (g_enable_non_kernel_time_query_interrupt) {
              executor->clearQuerySessionStatus(query_session, start_time);
            }
          };
      import_export::ImportStatus import_result;
      auto ms =
          measure<>::execution([&]() { import_result = importer->import(&session); });
      total_time += ms;
      // results
      if (!import_result.load_failed &&
          import_result.rows_rejected > copy_params.max_reject) {
        LOG(ERROR) << "COPY exited early due to reject records count during multi file "
                      "processing ";
        // if we have crossed the truncated load threshold
        import_result.load_failed = true;
        import_result.load_msg =
            "COPY exited early due to reject records count during multi file "
            "processing ";
        success_ = false;
      }
      if (!import_result.load_failed) {
        tr += std::string(
            "Loaded: " + std::to_string(import_result.rows_completed) +
            " recs, Rejected: " + std::to_string(import_result.rows_rejected) +
            " recs in " + std::to_string((double)total_time / 1000.0) + " secs");
      } else {
        tr += std::string("Loader Failed due to : " + import_result.load_msg + " in " +
                          std::to_string((double)total_time / 1000.0) + " secs");
      }
    } else {
      throw std::runtime_error("Table '" + *table_ + "' must exist before COPY FROM");
    }
  }
  return_message.reset(new std::string(tr));
  LOG(INFO) << tr;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void Parser::CopyTableStmt::get_deferred_copy_from_payload ( std::string &  table, std::string &  file_name, import_export::CopyParams &  copy_params, std::string &  partitions  )

inline

Definition at line 1498 of file ParserNode.h.

References deferred_copy_from_copy_params_, deferred_copy_from_file_name_, deferred_copy_from_partitions_, table_, and was_deferred_copy_from_.

                                                             {
    table = *table_;
    file_name = deferred_copy_from_file_name_;
    copy_params = deferred_copy_from_copy_params_;
    partitions = deferred_copy_from_partitions_;
    was_deferred_copy_from_ = false;
  }

bool Parser::CopyTableStmt::get_success ( ) const

inline

Definition at line 1494 of file ParserNode.h.

References success_.

{ return success_; }

std::string& Parser::CopyTableStmt::get_table ( ) const

inline

Definition at line 1489 of file ParserNode.h.

References CHECK, and table_.

                               {
    CHECK(table_);
    return *table_;
  }

bool Parser::CopyTableStmt::was_deferred_copy_from ( ) const

inline

Definition at line 1496 of file ParserNode.h.

References was_deferred_copy_from_.

{ return was_deferred_copy_from_; }

Member Data Documentation

std::unique_ptr<std::string> Parser::CopyTableStmt::copy_from_source_pattern_

private

Definition at line 1511 of file ParserNode.h.

Referenced by CopyTableStmt(), and execute().

import_export::CopyParams Parser::CopyTableStmt::deferred_copy_from_copy_params_

private

Definition at line 1517 of file ParserNode.h.

Referenced by execute(), and get_deferred_copy_from_payload().

std::string Parser::CopyTableStmt::deferred_copy_from_file_name_

private

Definition at line 1516 of file ParserNode.h.

Referenced by execute(), and get_deferred_copy_from_payload().

std::string Parser::CopyTableStmt::deferred_copy_from_partitions_

private

Definition at line 1518 of file ParserNode.h.

Referenced by execute(), and get_deferred_copy_from_payload().

std::list<std::unique_ptr<NameValueAssign> > Parser::CopyTableStmt::options_

private

Definition at line 1513 of file ParserNode.h.

Referenced by CopyTableStmt(), and execute().

std::unique_ptr<std::string> Parser::CopyTableStmt::return_message

Definition at line 1487 of file ParserNode.h.

Referenced by execute().

bool Parser::CopyTableStmt::success_

private

Definition at line 1512 of file ParserNode.h.

Referenced by execute(), and get_success().

std::unique_ptr<std::string> Parser::CopyTableStmt::table_

private

Definition at line 1510 of file ParserNode.h.

Referenced by CopyTableStmt(), execute(), get_deferred_copy_from_payload(), and get_table().

bool Parser::CopyTableStmt::was_deferred_copy_from_ = false

private

Definition at line 1515 of file ParserNode.h.

Referenced by execute(), get_deferred_copy_from_payload(), and was_deferred_copy_from().

---

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