#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, Importer_NS::CopyParams copy_params)

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

TRowDescriptor	get_row_descriptor ()

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

Private Member Functions
void	createConnection (const ThriftClientConnection &con)

void	closeConnection ()

void	wait_disconnet_reconnnect_retry (size_t tries, Importer_NS::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< MapDClient >	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 329 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_, 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 363 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 379 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 (TMapDException& 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 Importer_NS::CopyParams &	copy_params
	)

Definition at line 285 of file RowToColumnLoader.cpp.

References array_column_type_info_, column_type_info_, logger::ERROR, input_columns_, kARRAY, LOG, Importer_NS::CopyParams::null_str, Importer_NS::DelimitedParserUtils::parseStringArray(), 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;
           Importer_NS::DelimitedParserUtils::parseStringArray(
               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 367 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 MapDClient(conn_details_.get_protocol()));
 
   try {
     client_->connect(session_, user_name_, passwd_, db_name_);
   } catch (TMapDException& 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,
		Importer_NS::CopyParams	copy_params
	)

Definition at line 401 of file RowToColumnLoader.cpp.

References client_, input_columns_, Importer_NS::CopyParams::retry_count, row_desc_, session_, 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 (TMapDException& 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 281 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 Importer_NS::CopyParams &	copy_params
	)

Definition at line 111 of file RowToColumnLoader.cpp.

References Importer_NS::CopyParams::delimiter, and out.

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,
		Importer_NS::CopyParams	copy_params
	)

private

Definition at line 389 of file RowToColumnLoader.cpp.

References closeConnection(), conn_details_, createConnection(), Importer_NS::CopyParams::retry_count, and Importer_NS::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_);
 }