RowToColumnLoader Class Reference

#include <RowToColumnLoader.h>

Collaboration diagram for RowToColumnLoader:

Public Member Functions
	RowToColumnLoader (const ThriftClientConnection &conn_details, const std::string &user_name, const std::string &passwd, const std::string &db_name, const std::string &table_name)
	~RowToColumnLoader ()
void	do_load (int &nrows, int &nskipped, import_export::CopyParams copy_params)
bool	convert_string_to_column (std::vector< TStringValue > row, const import_export::CopyParams &copy_params)
TRowDescriptor	get_row_descriptor ()
std::string	print_row_with_delim (std::vector< TStringValue > row, const import_export::CopyParams &copy_params)

Private Member Functions
void	createConnection (const ThriftClientConnection &con)
void	closeConnection ()
void	wait_disconnet_reconnnect_retry (size_t tries, import_export::CopyParams copy_params)

Private Attributes
std::string	user_name_
std::string	passwd_
std::string	db_name_
std::string	table_name_
ThriftClientConnection	conn_details_
std::vector< TColumn >	input_columns_
std::vector< SQLTypeInfo >	column_type_info_
std::vector< SQLTypeInfo >	array_column_type_info_
TRowDescriptor	row_desc_
mapd::shared_ptr< OmniSciClient >	client_
TSessionId	session_

Detailed Description

Definition at line 62 of file RowToColumnLoader.h.

Constructor & Destructor Documentation

RowToColumnLoader::RowToColumnLoader	(	const ThriftClientConnection &	conn_details,
		const std::string &	user_name,
		const std::string &	passwd,
		const std::string &	db_name,
		const std::string &	table_name
	)

Definition at line 336 of file RowToColumnLoader.cpp.

References array_column_type_info_, client_, column_type_info_, conn_details_, create_array_sql_type_info_from_col_type(), create_sql_type_info_from_col_type(), createConnection(), input_columns_, row_desc_, session_, generate_TableFunctionsFactory_init::t, and table_name_.

    : user_name_(user_name)
    , passwd_(passwd)
    , db_name_(db_name)
    , table_name_(table_name)
    , conn_details_(conn_details) {
  createConnection(conn_details_);

  TTableDetails table_details;
  client_->get_table_details(table_details, session_, table_name_);

  row_desc_ = table_details.row_desc;

  // create vector with column details
  for (TColumnType ct : row_desc_) {
    column_type_info_.push_back(create_sql_type_info_from_col_type(ct));
  }

  // create vector with array column details presented as real column for easier resue
  // of othe code
  for (TColumnType ct : row_desc_) {
    array_column_type_info_.push_back(create_array_sql_type_info_from_col_type(ct));
  }

  // create vector for storage of the actual column data
  for (TColumnType column : row_desc_) {
    TColumn t;
    input_columns_.push_back(t);
  }
}

Here is the call graph for this function:

RowToColumnLoader::~RowToColumnLoader

(

)

Definition at line 370 of file RowToColumnLoader.cpp.

References closeConnection().

                                      {
  closeConnection();
}

Here is the call graph for this function:

Member Function Documentation

void RowToColumnLoader::closeConnection ( )

private

Definition at line 386 of file RowToColumnLoader.cpp.

References client_, and session_.

Referenced by wait_disconnet_reconnnect_retry(), and ~RowToColumnLoader().

                                        {
  try {
    client_->disconnect(session_);  // disconnect from omnisci_server
  } catch (TOmniSciException& e) {
    std::cerr << e.error_msg << std::endl;
  } catch (TException& te) {
    std::cerr << "Thrift error on close: " << te.what() << std::endl;
  }
}

Here is the caller graph for this function:

bool RowToColumnLoader::convert_string_to_column	(	std::vector< TStringValue >	row,
		const import_export::CopyParams &	copy_params
	)

Definition at line 292 of file RowToColumnLoader.cpp.

References array_column_type_info_, column_type_info_, logger::ERROR, input_columns_, kARRAY, LOG, import_export::CopyParams::null_str, import_export::delimited_parser::parse_string_array(), populate_TColumn(), print_row_with_delim(), and remove_partial_row().

Referenced by msg_consume(), and stream_insert().

                                                {
  // create datum and push data to column structure from row data
  uint curr_col = 0;
  for (TStringValue ts : row) {
    try {
      switch (column_type_info_[curr_col].get_type()) {
        case SQLTypes::kARRAY: {
          std::vector<std::string> arr_ele;
          import_export::delimited_parser::parse_string_array(
              ts.str_val, copy_params, arr_ele);
          TColumn array_tcol;
          for (std::string item : arr_ele) {
            boost::algorithm::trim(item);
            TStringValue tsa;
            tsa.str_val = item;
            tsa.is_null = (tsa.str_val.empty() || tsa.str_val == copy_params.null_str);
            // now put into TColumn
            populate_TColumn(
                tsa, array_column_type_info_[curr_col], array_tcol, copy_params);
          }
          input_columns_[curr_col].nulls.push_back(false);
          input_columns_[curr_col].data.arr_col.push_back(array_tcol);

          break;
        }
        default:
          populate_TColumn(
              ts, column_type_info_[curr_col], input_columns_[curr_col], copy_params);
      }
    } catch (const std::exception& e) {
      remove_partial_row(curr_col, column_type_info_, input_columns_);
      // import_status.rows_rejected++;
      LOG(ERROR) << "Input exception thrown: " << e.what()
                 << ". Row discarded, issue at column : " << (curr_col + 1)
                 << " data :" << print_row_with_delim(row, copy_params);
      return false;
    }
    curr_col++;
  }
  return true;
}

Here is the call graph for this function:

Here is the caller graph for this function:

void RowToColumnLoader::createConnection ( const ThriftClientConnection &  con )

private

Definition at line 374 of file RowToColumnLoader.cpp.

References client_, conn_details_, db_name_, ThriftClientConnection::get_protocol(), passwd_, session_, and user_name_.

Referenced by RowToColumnLoader(), and wait_disconnet_reconnnect_retry().

                                                                          {
  client_.reset(new OmniSciClient(conn_details_.get_protocol()));

  try {
    client_->connect(session_, user_name_, passwd_, db_name_);
  } catch (TOmniSciException& e) {
    std::cerr << e.error_msg << std::endl;
  } catch (TException& te) {
    std::cerr << "Thrift error on connect: " << te.what() << std::endl;
  }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void RowToColumnLoader::do_load	(	int &	nrows,
		int &	nskipped,
		import_export::CopyParams	copy_params
	)

Definition at line 408 of file RowToColumnLoader.cpp.

References client_, input_columns_, import_export::CopyParams::retry_count, row_desc_, session_, generate_TableFunctionsFactory_init::t, table_name_, and wait_disconnet_reconnnect_retry().

Referenced by kafka_insert(), and stream_insert().

                                                                     {
  for (size_t tries = 0; tries < copy_params.retry_count;
       tries++) {  // allow for retries in case of insert failure
    try {
      client_->load_table_binary_columnar(session_, table_name_, input_columns_, {});
      //      client->load_table(session, table_name, input_rows);
      nrows += input_columns_[0].nulls.size();
      std::cout << nrows << " Rows Inserted, " << nskipped << " rows skipped."
                << std::endl;
      // we successfully loaded the data, lets move on
      input_columns_.clear();
      // create vector for storage of the actual column data
      for (TColumnType column : row_desc_) {
        TColumn t;
        input_columns_.push_back(t);
      }
      return;
    } catch (TOmniSciException& e) {
      std::cerr << "Exception trying to insert data " << e.error_msg << std::endl;
      wait_disconnet_reconnnect_retry(tries, copy_params);
    } catch (TException& te) {
      std::cerr << "Exception trying to insert data " << te.what() << std::endl;
      wait_disconnet_reconnnect_retry(tries, copy_params);
    }
  }
  std::cerr << "Retries exhausted program terminated" << std::endl;
  exit(1);
}

Here is the call graph for this function:

Here is the caller graph for this function:

TRowDescriptor RowToColumnLoader::get_row_descriptor

(

)

Definition at line 288 of file RowToColumnLoader.cpp.

References row_desc_.

Referenced by msg_consume(), and stream_insert().

                                                     {
  return row_desc_;
};

Here is the caller graph for this function:

std::string RowToColumnLoader::print_row_with_delim	(	std::vector< TStringValue >	row,
		const import_export::CopyParams &	copy_params
	)

Definition at line 113 of file RowToColumnLoader.cpp.

References import_export::CopyParams::delimiter.

Referenced by convert_string_to_column(), msg_consume(), and stream_insert().

                                                {
  std::ostringstream out;
  bool first = true;
  for (TStringValue ts : row) {
    if (first) {
      first = false;
    } else {
      out << copy_params.delimiter;
    }
    out << ts.str_val;
  }
  return out.str();
}

Here is the caller graph for this function:

void RowToColumnLoader::wait_disconnet_reconnnect_retry ( size_t  tries, import_export::CopyParams  copy_params  )

private

Definition at line 396 of file RowToColumnLoader.cpp.

References closeConnection(), conn_details_, createConnection(), import_export::CopyParams::retry_count, and import_export::CopyParams::retry_wait.

Referenced by do_load().

                                         {
  std::cout << "  Waiting  " << copy_params.retry_wait
            << " secs to retry Inserts , will try " << (copy_params.retry_count - tries)
            << " times more " << std::endl;
  sleep(copy_params.retry_wait);

  closeConnection();
  createConnection(conn_details_);
}

Here is the call graph for this function:

Here is the caller graph for this function:

Member Data Documentation

std::vector<SQLTypeInfo> RowToColumnLoader::array_column_type_info_

private

Definition at line 86 of file RowToColumnLoader.h.

Referenced by convert_string_to_column(), and RowToColumnLoader().

mapd::shared_ptr<OmniSciClient> RowToColumnLoader::client_

private

Definition at line 90 of file RowToColumnLoader.h.

Referenced by closeConnection(), createConnection(), do_load(), and RowToColumnLoader().

std::vector<SQLTypeInfo> RowToColumnLoader::column_type_info_

private

Definition at line 85 of file RowToColumnLoader.h.

Referenced by convert_string_to_column(), and RowToColumnLoader().

ThriftClientConnection RowToColumnLoader::conn_details_

private

Definition at line 82 of file RowToColumnLoader.h.

Referenced by createConnection(), RowToColumnLoader(), and wait_disconnet_reconnnect_retry().

std::string RowToColumnLoader::db_name_

private

Definition at line 80 of file RowToColumnLoader.h.

Referenced by createConnection().

std::vector<TColumn> RowToColumnLoader::input_columns_

private

Definition at line 84 of file RowToColumnLoader.h.

Referenced by convert_string_to_column(), do_load(), and RowToColumnLoader().

std::string RowToColumnLoader::passwd_

private

Definition at line 79 of file RowToColumnLoader.h.

Referenced by createConnection().

TRowDescriptor RowToColumnLoader::row_desc_

private

Definition at line 88 of file RowToColumnLoader.h.

Referenced by do_load(), get_row_descriptor(), and RowToColumnLoader().

TSessionId RowToColumnLoader::session_

private

Definition at line 91 of file RowToColumnLoader.h.

Referenced by closeConnection(), createConnection(), do_load(), and RowToColumnLoader().

std::string RowToColumnLoader::table_name_

private

Definition at line 81 of file RowToColumnLoader.h.

Referenced by do_load(), and RowToColumnLoader().

std::string RowToColumnLoader::user_name_

private

Definition at line 78 of file RowToColumnLoader.h.

Referenced by createConnection().

---

The documentation for this class was generated from the following files:

• /home/jenkins-slave/workspace/core-os-doxygen/ImportExport/RowToColumnLoader.h
• /home/jenkins-slave/workspace/core-os-doxygen/ImportExport/RowToColumnLoader.cpp