~maria-captains/maria/5.5-cassandra-parallel

// Cassandra includes:
#include <inttypes.h>
#include <netinet/in.h>
#include <sys/time.h>
#include <stdio.h>
#include <stdarg.h>

#include <boost/asio.hpp>
#include <boost/bind.hpp>

#include <async/TAsioAsync.h>

#include "Thrift.h"
#include "transport/TSocket.h"
#include "transport/TTransport.h"
#include "transport/TBufferTransports.h"
#include "protocol/TProtocol.h"
#include "protocol/TBinaryProtocol.h"
#include "gen-cpp-async/Cassandra.h"
// cassandra includes end

#include "cassandra_se.h"

using namespace std;
using namespace apache::thrift;
using namespace apache::thrift::transport;
using namespace apache::thrift::protocol;
using namespace org::apache::cassandra;

void Cassandra_se_interface::print_error(const char *format, ...)
{
  va_list ap;
  va_start(ap, format);
  // it's not a problem if output was truncated
  vsnprintf(err_buffer, sizeof(err_buffer), format, ap);
  va_end(ap);
}

/*
  Implementation of connection to one Cassandra column family (ie., table)
*/
class Cassandra_se_impl: public Cassandra_se_interface
{
  /* Parameters */ 
  std::vector<std::string> hosts_list;
  ConsistencyLevel::type cur_consistency_level;
  std::string column_family;
  std::string keyspace;
  
  /* Connection handlers */
  boost::asio::io_service io_service;
  boost::shared_ptr<protocol::TProtocolFactory> protocolFactory;

  //boost::shared_ptr<async::TAsioClient> client;
  std::vector<boost::shared_ptr<async::TAsioClient> > clients;

  std::vector<boost::shared_ptr<CassandraAsyncClient> > async_clients;
  
  // DELETE THIS: v
  CassandraClient *cass; /* Connection to cassandra */


  
  /* DDL data */
  KsDef ks_def; /* KeySpace we're using (TODO: put this in table->share) */
  CfDef cf_def; /* Column family we're using (TODO: put in table->share)*/
  std::vector<ColumnDef>::iterator column_ddl_it;

  /* The list that was returned by the last key lookup */
  std::vector<ColumnOrSuperColumn> column_data_vec;
  std::vector<ColumnOrSuperColumn>::iterator column_data_it;

  /* Insert preparation */
  typedef std::map<std::string, std::vector<Mutation> > ColumnFamilyToMutation;
  typedef std::map<std::string,  ColumnFamilyToMutation> KeyToCfMutationMap;
   
  std::vector<KeyToCfMutationMap> batch_mutations; /* Prepare operation here */
  KeyToCfMutationMap *batch_mutation; /* Prepare operation here */
  uint batch_mutation_idx;
  
  int64_t insert_timestamp; /* rows will be inserted with this timestamp */
  
  /* Current row mutation being filled */
  std::vector<Mutation>* insert_list;
   
  /* Resultset we're reading */
  std::vector<KeySlice> key_slice_vec;
  std::vector<KeySlice>::iterator key_slice_it;
  
  std::string rowkey; /* key of the record we're returning now */

  SlicePredicate slice_pred;
  bool get_slices_returned_less;
public:
  Cassandra_se_impl() /* : cass(NULL) */ {}
  virtual ~Cassandra_se_impl(){ /*delete cass; */ }
  
  /* Connection and DDL checks */
  bool make_hosts_list(const char *hosts);
  bool connect(const char *host, int port, const char *keyspace);
  void set_column_family(const char *cfname) { column_family.assign(cfname); }

  bool setup_ddl_checks();
  void first_ddl_column();
  bool next_ddl_column(char **name, int *name_len, char **value, int *value_len);
  void get_rowkey_type(char **name, char **type);

  /* Writes */
  void clear_insert_buffer();
  void start_row_insert(const char *key, int key_len);
  void add_insert_column(const char *name, const char *value, int value_len);
  bool do_insert(bool flush);

  /* Reads, point lookups */
  bool get_slice(char *key, size_t key_len, bool *found);
  bool get_next_read_column(char **name, char **value, int *value_len);
  void get_read_rowkey(char **value, int *value_len);

  /* Reads, multi-row scans */
private:
  bool have_rowkey_to_skip;
  std::string rowkey_to_skip;
public:
  bool get_range_slices(bool last_key_as_start_key);
  void finish_reading_range_slices();
  bool get_next_range_slice_row(bool *eof);

  /* Setup that's necessary before a multi-row read. (todo: use it before point lookups, too) */
  void clear_read_columns();
  void add_read_column(const char *name);
 
  /* Reads, MRR scans */
  void new_lookup_keys();
  int  add_lookup_key(const char *key, size_t key_len);
  bool multiget_slice();

private:
  std::vector<std::string> mrr_keys; /* TODO: can we use allocator to put them onto MRR buffer? */
  std::map<std::string, std::vector<ColumnOrSuperColumn> > mrr_result;
  std::map<std::string, std::vector<ColumnOrSuperColumn> >::iterator mrr_result_it;
public:
  bool get_next_multiget_row();

  bool truncate();
  bool remove_row();

private:
  /* Non-inherited utility functions: */
  int64_t get_i64_timestamp();

  void on_connected2(boost::shared_ptr< CassandraAsyncClient> client);
  void on_set_keyspace();
  void on_set_keyspace_fail();
  void on_describe_keyspace(org::apache::cassandra::KsDef ks_def_arg);
  void on_batch_mutate_done();

  uint connected;
  uint keyspace_set;
  bool describe_keyspace_done;
  uint batch_mutations_to_do;

  bool remove_op_done;
  void on_remove_done() { remove_op_done= true; }

  bool error_happened;
};


/////////////////////////////////////////////////////////////////////////////
// Connection and setup
/////////////////////////////////////////////////////////////////////////////
Cassandra_se_interface *get_cassandra_se()
{
  return new Cassandra_se_impl;
}


void Cassandra_se_impl::on_connected2(boost::shared_ptr<CassandraAsyncClient> client)
{
  async_clients.push_back(client);
  connected++;
  fprintf(stderr, "=Connected!\n");
}


void Cassandra_se_impl::on_set_keyspace()
{
  keyspace_set++;
  fprintf(stderr, "=Keyspace set!\n");
}

void Cassandra_se_impl::on_set_keyspace_fail()
{
  keyspace_set++;
  error_happened= true;
}

bool Cassandra_se_impl::make_hosts_list(const char *hosts)
{
  const char* p= hosts;
  const char* name_start=p;
  hosts_list.clear();
  while (1)
  {
    if (*p==0 || *p==',')
    {
      // [name_start; p) is the name
      size_t len= p - name_start;
      if (len==0)
        return true;
      hosts_list.push_back(std::string(name_start, len));
      name_start= p + 1;
      if (*p == 0)
        break;
    }
    p++;
  }
  if (hosts_list.size() > 0)
    return false; /* OK */
  else
    return true; /* Empty list: error */
}


bool Cassandra_se_impl::connect(const char *host, int port, const char *keyspace_arg)
{
  bool res= true;
  
  if (make_hosts_list(host))
    return true;

  keyspace.assign(keyspace_arg);

  try {
    protocolFactory= boost::shared_ptr<protocol::TProtocolFactory>(new protocol::TBinaryProtocolFactory());
    clients.resize(hosts_list.size());

    for (uint i= 0; i < hosts_list.size(); i++)
    {
      clients[i]= boost::shared_ptr<async::TAsioClient>(
                             new async::TAsioClient(io_service, 
                              protocolFactory, protocolFactory));

      boost::function<void (boost::shared_ptr<CassandraAsyncClient> c )> call_me;
      call_me= boost::bind(&Cassandra_se_impl::on_connected2, this, _1);

      clients[i]->connect(hosts_list[i], port, call_me);
    }
    connected= 0;

    //io_service.run();
    while (connected !=hosts_list.size())
    {
      size_t cnt= io_service.run_one();
      fprintf(stderr, "processed %d events\n", (int)cnt);
    }

    keyspace_set= 0;
    error_happened=false;
    for (uint i= 0; i < hosts_list.size(); i++)
    {
      boost::function< void () > call_on_connect;
      call_on_connect= boost::bind(&Cassandra_se_impl::on_set_keyspace, this);
      async_clients[i]->set_keyspace(keyspace_arg).
        setCallback(boost::bind(&Cassandra_se_impl::on_set_keyspace, this)).
        setErrback(boost::bind(&Cassandra_se_impl::on_set_keyspace_fail, this));
    }

    while (keyspace_set != hosts_list.size())
    {
      size_t cnt= io_service.run_one();
      fprintf(stderr, "processed %d events\n", (int)cnt);
    }
    if (error_happened)
      res= true;
    else
      res= false;

  }catch (...) {
    print_error("Unknown exception");
  }

#if 1  
  try {
    boost::shared_ptr<TTransport> socket = 
      boost::shared_ptr<TSocket>(new TSocket(hosts_list[0], port));
    boost::shared_ptr<TTransport> tr = 
      boost::shared_ptr<TFramedTransport>(new TFramedTransport (socket));
    boost::shared_ptr<TProtocol> p = 
      boost::shared_ptr<TBinaryProtocol>(new TBinaryProtocol(tr));
    
    cass= new CassandraClient(p);
    tr->open();
    cass->set_keyspace(keyspace_arg);

    res= false; // success
  }catch(TTransportException te){
    print_error("%s [%d]", te.what(), te.getType());
  }catch(InvalidRequestException ire){
    print_error("%s [%s]", ire.what(), ire.why.c_str());
  }catch(NotFoundException nfe){
    print_error("%s", nfe.what());
  }catch(TException e){
    print_error("Thrift exception: %s", e.what());
  }catch (...) {
    print_error("Unknown exception");
  }
#endif  

  cur_consistency_level= ConsistencyLevel::ONE;

  if (!res && setup_ddl_checks())
    res= true;
  return res;
}


void Cassandra_se_impl::on_describe_keyspace(org::apache::cassandra::KsDef ks_def_arg)
{
  fprintf(stderr, "=describe_keyspace_done");
  ks_def= ks_def_arg;
  describe_keyspace_done= true;
}


bool Cassandra_se_impl::setup_ddl_checks()
{
  describe_keyspace_done= false;
  async_clients[0]->describe_keyspace(keyspace).
                    setCallback(boost::bind(&Cassandra_se_impl::on_describe_keyspace,
                    this, _1));

  while (!describe_keyspace_done)
  {
    size_t cnt= io_service.run_one();
    fprintf(stderr, "processed %d events\n", (int)cnt);
  }

#if 0  
  try {
    cass->describe_keyspace(ks_def, keyspace);
  } catch (InvalidRequestException ire) {
    print_error("%s [%s]", ire.what(), ire.why.c_str());
  } catch (NotFoundException nfe) {
    print_error("keyspace not found: %s", nfe.what());
  }catch(TException e){
    print_error("Thrift exception: %s", e.what());
  } catch (...) {
    print_error("Unknown exception");
  }
#endif

  std::vector<CfDef>::iterator it;
  for (it= ks_def.cf_defs.begin(); it < ks_def.cf_defs.end(); it++)
  {
    cf_def= *it;
    if (!cf_def.name.compare(column_family))
      return false;
  }

  print_error("describe_keyspace() didn't return our column family");

  return true;
}


void Cassandra_se_impl::first_ddl_column()
{
  column_ddl_it= cf_def.column_metadata.begin();
}


bool Cassandra_se_impl::next_ddl_column(char **name, int *name_len, 
                                        char **type, int *type_len)
{
  if (column_ddl_it == cf_def.column_metadata.end())
    return true;

  *name= (char*)(*column_ddl_it).name.c_str();
  *name_len= (*column_ddl_it).name.length();

  *type= (char*)(*column_ddl_it).validation_class.c_str();
  *type_len= (*column_ddl_it).validation_class.length();

  column_ddl_it++;
  return false;
}


void Cassandra_se_impl::get_rowkey_type(char **name, char **type)
{
  if (cf_def.__isset.key_validation_class)
    *type= (char*)cf_def.key_validation_class.c_str();
  else
    *type= NULL;

  if (cf_def.__isset.key_alias)
    *name= (char*)cf_def.key_alias.c_str();
  else
    *name= NULL;
}


/////////////////////////////////////////////////////////////////////////////
// Data writes
/////////////////////////////////////////////////////////////////////////////
int64_t Cassandra_se_impl::get_i64_timestamp()
{
  struct timeval td;
  gettimeofday(&td, NULL);
  int64_t ms = td.tv_sec;
  ms = ms * 1000;
  int64_t usec = td.tv_usec;
  usec = usec / 1000;
  ms += usec;
  
  return ms;
}


void Cassandra_se_impl::clear_insert_buffer()
{
  batch_mutations.resize(hosts_list.size());
  for (uint i=0; i < hosts_list.size(); i++)
    batch_mutations[i]= KeyToCfMutationMap();

  batch_mutation_idx= 0;
  batch_mutation= &batch_mutations[batch_mutation_idx];
}


void Cassandra_se_impl::start_row_insert(const char *key, int key_len)
{
  std::string key_to_insert;
  key_to_insert.assign(key, key_len);
  (*batch_mutation)[key_to_insert]= ColumnFamilyToMutation();
  ColumnFamilyToMutation& cf_mut= (*batch_mutation)[key_to_insert];

  cf_mut[column_family]= std::vector<Mutation>();
  insert_list= &cf_mut[column_family];

  insert_timestamp= get_i64_timestamp();
}


void Cassandra_se_impl::add_insert_column(const char *name, const char *value, 
                                          int value_len)
{
  Mutation mut;
  mut.__isset.column_or_supercolumn= true;
  mut.column_or_supercolumn.__isset.column= true;

  Column& col=mut.column_or_supercolumn.column;
  col.name.assign(name);
  col.value.assign(value, value_len);
  col.timestamp= insert_timestamp;
  col.__isset.value= true;
  col.__isset.timestamp= true;
  insert_list->push_back(mut);
}


void Cassandra_se_impl::on_batch_mutate_done()
{
  batch_mutations_to_do--;
}


bool Cassandra_se_impl::do_insert(bool flush)
{
  bool res= true;
  size_t size= hosts_list.size();

  if (!flush)
  {
    /* Switch to constructing another mutation object */
    if (batch_mutation_idx != size - 1)
    {
      batch_mutation= &batch_mutations[++batch_mutation_idx];
      return false;
    }
  }

  for (uint i= 0; i < size; i++)
  {
    /*
      zero-size mutations are allowed by Cassandra's batch_mutate but lets not 
      do them (we may attempt to do it if there is a bulk insert that stores
      exactly @@cassandra_insert_batch_size*n elements.
    */
    if (batch_mutations[i].empty())
      continue;
 
    batch_mutations_to_do++;
    async_clients[i]->batch_mutate(batch_mutations[i], cur_consistency_level).
      setCallback(boost::bind(&Cassandra_se_impl::on_batch_mutate_done, this));

    cassandra_counters.row_inserts+= batch_mutations[i].size();
    cassandra_counters.row_insert_batches++;
  }
  
  while (batch_mutations_to_do != 0)
  {
    io_service.run_one();
    //fprintf(stderr, "bm processed %d events\n", (int)cnt);
  }
  res= false;

  clear_insert_buffer();

#if 0
  try {
    
    cass->batch_mutate(batch_mutation, cur_consistency_level);

    cassandra_counters.row_inserts+= batch_mutation.size();
    cassandra_counters.row_insert_batches++;

    clear_insert_buffer();
    res= false;

  } catch (InvalidRequestException ire) {
    print_error("%s [%s]", ire.what(), ire.why.c_str());
  } catch (UnavailableException ue) {
    print_error("UnavailableException: %s", ue.what());
  } catch (TimedOutException te) {
    print_error("TimedOutException: %s", te.what());
  }catch(TException e){
    print_error("Thrift exception: %s", e.what());
  } catch (...) {
    print_error("Unknown exception");
  }
#endif
  return res;
}


/////////////////////////////////////////////////////////////////////////////
// Reading data
/////////////////////////////////////////////////////////////////////////////

/*
  Make one key lookup. If the record is found, the result is stored locally and 
  the caller should iterate over it.
*/

bool Cassandra_se_impl::get_slice(char *key, size_t key_len, bool *found)
{
  ColumnParent cparent;
  cparent.column_family= column_family;

  rowkey.assign(key, key_len);

  SlicePredicate slice_pred;
  SliceRange sr;
  sr.start = "";
  sr.finish = "";
  slice_pred.__set_slice_range(sr);

  try {
    cass->get_slice(column_data_vec, rowkey, cparent, slice_pred, 
                    cur_consistency_level);

    if (column_data_vec.size() == 0)
    {
      /*
        No columns found. Cassandra doesn't allow records without any column =>
        this means the seach key doesn't exist
      */
      *found= false;
      return false;
    }
    *found= true;

  } catch (InvalidRequestException ire) {
    print_error("%s [%s]", ire.what(), ire.why.c_str());
    return true;
  } catch (UnavailableException ue) {
    print_error("UnavailableException: %s", ue.what());
    return true;
  } catch (TimedOutException te) {
    print_error("TimedOutException: %s", te.what());
    return true;
  }catch(TException e){
    print_error("Thrift exception: %s", e.what());
  } catch (...) {
    print_error("Unknown exception");
    return true;
  }

  column_data_it= column_data_vec.begin();
  return false;
}


bool Cassandra_se_impl::get_next_read_column(char **name, char **value, 
                                             int *value_len)
{
  bool use_counter=false;
  while (1)
  {
    if (column_data_it == column_data_vec.end())
      return true;

    if ((*column_data_it).__isset.column)
      break; /* Ok it's a real column. Should be always the case. */

    if ((*column_data_it).__isset.counter_column)
    {
      use_counter= true;
      break;
    }

    column_data_it++;
  }

  ColumnOrSuperColumn& cs= *column_data_it;
  if (use_counter)
  {
    *name= (char*)cs.counter_column.name.c_str();
    *value= (char*)&cs.counter_column.value;
    *value_len= sizeof(cs.counter_column.value);
  }
  else
  {
    *name= (char*)cs.column.name.c_str();
    *value= (char*)cs.column.value.c_str();
    *value_len= cs.column.value.length();
  }

  column_data_it++;
  return false;
}


/* Return the rowkey for the record that was read */

void Cassandra_se_impl::get_read_rowkey(char **value, int *value_len)
{
  *value= (char*)rowkey.c_str();
  *value_len= rowkey.length();
}


bool Cassandra_se_impl::get_range_slices(bool last_key_as_start_key)
{
  bool res= true;
  
  ColumnParent cparent;
  cparent.column_family= column_family;
  
  /* SlicePredicate can be used to limit columns we will retrieve */

  KeyRange key_range;
  key_range.__isset.start_key= true;
  key_range.__isset.end_key= true;

  if (last_key_as_start_key)
  {
    key_range.start_key= rowkey;

    have_rowkey_to_skip= true;
    rowkey_to_skip= rowkey;
  }
  else
  {
    have_rowkey_to_skip= false;
    key_range.start_key.assign("", 0);
  }

  key_range.end_key.assign("", 0);
  key_range.count= read_batch_size;
  try {
  
    cass->get_range_slices(key_slice_vec,
                           cparent, slice_pred, key_range, 
                           cur_consistency_level);
    res= false;

    if (key_slice_vec.size() < (uint)read_batch_size)
      get_slices_returned_less= true;
    else
      get_slices_returned_less= false;

  } catch (InvalidRequestException ire) {
    print_error("%s [%s]", ire.what(), ire.why.c_str());
  } catch (UnavailableException ue) {
    print_error("UnavailableException: %s", ue.what());
  } catch (TimedOutException te) {
    print_error("TimedOutException: %s", te.what());
  }catch(TException e){
    print_error("Thrift exception: %s", e.what());
  } catch (...) {
    print_error("Unknown exception");
  }

  key_slice_it= key_slice_vec.begin();
  return res;
}


/* Switch to next row. This may produce an error */
bool Cassandra_se_impl::get_next_range_slice_row(bool *eof)
{
restart:
  if (key_slice_it == key_slice_vec.end())
  {
    if (get_slices_returned_less)
    {
      *eof= true;
      return false;
    }

    /*
      We have read through all columns in this batch. Try getting the next
      batch.
    */
    if (get_range_slices(true))
      return true;

    if (key_slice_vec.empty())
    {
      *eof= true;
      return false;
    }
  }
  
  if (have_rowkey_to_skip && !rowkey_to_skip.compare(key_slice_it->key))
  {
    key_slice_it++;
    goto restart;
  }

  *eof= false;
  column_data_vec= key_slice_it->columns;
  rowkey= key_slice_it->key;
  column_data_it= column_data_vec.begin();
  key_slice_it++;
  return false;
}


void Cassandra_se_impl::finish_reading_range_slices()
{
  key_slice_vec.clear();
}


void Cassandra_se_impl::clear_read_columns()
{
  slice_pred.column_names.clear();
}


void Cassandra_se_impl::add_read_column(const char *name_arg)
{
  std::string name(name_arg);
  slice_pred.__isset.column_names= true;
  slice_pred.column_names.push_back(name);
}


bool Cassandra_se_impl::truncate()
{
  bool res= true;
  try {
    
    cass->truncate(column_family);
    res= false;

  } catch (InvalidRequestException ire) {
    print_error("%s [%s]", ire.what(), ire.why.c_str());
  } catch (UnavailableException ue) {
    print_error("UnavailableException: %s", ue.what());
  } catch (TimedOutException te) {
    print_error("TimedOutException: %s", te.what());
  }catch(TException e){
    print_error("Thrift exception: %s", e.what());
  } catch (...) {
    print_error("Unknown exception");
  }

  return res;
}

bool Cassandra_se_impl::remove_row()
{
  bool res= true;

  ColumnPath column_path;
  column_path.column_family= column_family;

  remove_op_done= false;
  async_clients[0]->remove(rowkey, column_path, get_i64_timestamp(), cur_consistency_level).
      setCallback(boost::bind(&Cassandra_se_impl::on_remove_done, this));
  
  
  while (!remove_op_done)
  {
    io_service.run_one();
  }
  res= false;

#if 0
  try {
    
    cass->remove(rowkey, column_path, get_i64_timestamp(), cur_consistency_level);
    res= false;

  } catch (InvalidRequestException ire) {
    print_error("%s [%s]", ire.what(), ire.why.c_str());
  } catch (UnavailableException ue) {
    print_error("UnavailableException: %s", ue.what());
  } catch (TimedOutException te) {
    print_error("TimedOutException: %s", te.what());
  }catch(TException e){
    print_error("Thrift exception: %s", e.what());
  } catch (...) {
    print_error("Unknown exception");
  }
#endif
  return res;
}

/////////////////////////////////////////////////////////////////////////////
// MRR reads
/////////////////////////////////////////////////////////////////////////////

void Cassandra_se_impl::new_lookup_keys()
{
  mrr_keys.clear();
}


int Cassandra_se_impl::add_lookup_key(const char *key, size_t key_len)
{
  mrr_keys.push_back(std::string(key, key_len));
  return mrr_keys.size();
}


bool Cassandra_se_impl::multiget_slice()
{
  ColumnParent cparent;
  cparent.column_family= column_family;

  SlicePredicate slice_pred;
  SliceRange sr;
  sr.start = "";
  sr.finish = "";
  slice_pred.__set_slice_range(sr);

  bool res= true;

  try {
    
    cassandra_counters.multiget_reads++;
    cassandra_counters.multiget_keys_scanned += mrr_keys.size();

    cass->multiget_slice(mrr_result, mrr_keys, cparent, slice_pred, 
                         cur_consistency_level);

    cassandra_counters.multiget_rows_read += mrr_result.size();
    
    res= false;
    mrr_result_it= mrr_result.begin();

  } catch (InvalidRequestException ire) {
    print_error("%s [%s]", ire.what(), ire.why.c_str());
  } catch (UnavailableException ue) {
    print_error("UnavailableException: %s", ue.what());
  } catch (TimedOutException te) {
    print_error("TimedOutException: %s", te.what());
  }catch(TException e){
    print_error("Thrift exception: %s", e.what());
  } catch (...) {
    print_error("Unknown exception");
  }

  return res;
}


bool Cassandra_se_impl::get_next_multiget_row()
{
  if (mrr_result_it == mrr_result.end())
    return true; /* EOF */

  column_data_vec= mrr_result_it->second;
  rowkey= mrr_result_it->first;

  column_data_it= column_data_vec.begin();
  mrr_result_it++;
  return false;
}