Cassandra SE: add support for reading counter type values

2 files changed, 37 insertions, 9 deletions

diff --git a/storage/cassandra/cassandra_se.cc b/storage/cassandra/cassandra_se.cc
index 692bfb595da..6eba141cf58 100644
--- a/storage/cassandra/cassandra_se.cc
+++ b/storage/cassandra/cassandra_se.cc
@@ -378,21 +378,37 @@ bool Cassandra_se_impl::get_slice(char *key, size_t key_len, bool *found)
 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))
+    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;
-  *name= (char*)cs.column.name.c_str();
-  *value= (char*)cs.column.value.c_str();
-  *value_len= cs.column.value.length();
+  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;
diff --git a/storage/cassandra/ha_cassandra.cc b/storage/cassandra/ha_cassandra.cc
index 7ef563114fd..f1a5916ffc6 100644
--- a/storage/cassandra/ha_cassandra.cc
+++ b/storage/cassandra/ha_cassandra.cc
@@ -501,23 +501,31 @@ static void flip64(const char *from, char* to)
 class BigintDataConverter : public ColumnDataConverter
 {
   longlong buf;
+  bool flip; /* is false when reading counter columns */
 public:
   void cassandra_to_mariadb(const char *cass_data, int cass_data_len)
   {
     longlong tmp;
     DBUG_ASSERT(cass_data_len == sizeof(longlong));
-    flip64(cass_data, (char*)&tmp);
+    if (flip)
+      flip64(cass_data, (char*)&tmp);
+    else
+      memcpy(&tmp, cass_data, sizeof(longlong));
     field->store(tmp);
   }
   
   bool mariadb_to_cassandra(char **cass_data, int *cass_data_len)
   {
     longlong tmp= field->val_int();
-    flip64((const char*)&tmp, (char*)&buf);
+    if (flip)
+      flip64((const char*)&tmp, (char*)&buf);
+    else
+      memcpy(&buf, &tmp, sizeof(longlong));
     *cass_data= (char*)&buf;
     *cass_data_len=sizeof(longlong);
     return false;
   }
+  BigintDataConverter(bool flip_arg) : flip(flip_arg) {}
   ~BigintDataConverter(){}
 };
 
@@ -723,6 +731,7 @@ const char * const validator_uuid= "org.apache.cassandra.db.marshal.UUIDType";
 
 const char * const validator_boolean= "org.apache.cassandra.db.marshal.BooleanType";
 
+
 ColumnDataConverter *map_field_to_validator(Field *field, const char *validator_name)
 {
   ColumnDataConverter *res= NULL;
@@ -737,10 +746,13 @@ ColumnDataConverter *map_field_to_validator(Field *field, const char *validator_
       /* fall through: */
     case MYSQL_TYPE_SHORT:
     case MYSQL_TYPE_LONGLONG:
-      if (!strcmp(validator_name, validator_bigint))
-        res= new BigintDataConverter;
+    {
+      bool is_counter= false;
+      if (!strcmp(validator_name, validator_bigint) ||
+          (is_counter= !strcmp(validator_name, validator_counter)))
+        res= new BigintDataConverter(!is_counter);
       break;
-
+    }
     case MYSQL_TYPE_FLOAT:
       if (!strcmp(validator_name, validator_float))
         res= new FloatDataConverter;