Cassandra SE: more datatypes support

- Support mapping Cassandra's timestamp to INT64
- Support mapping Cassadnra's decimal to VARBINARY.

mysql-test/r/cassandra.result

@@ -343,3 +343,29 @@ thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cf9';
 select rowkey, hex(varint_col) from t2;
 ERROR HY000: Internal error: 'Unable to convert value for field `varint_col` from Cassandra's data format. Source data is 4 bytes, 0x12345678'
 drop table t2;
+#
+# Decimal datatype support
+#
+CREATE TABLE t2 (rowkey varchar(32) PRIMARY KEY, decimal_col varbinary(32)) ENGINE=CASSANDRA
+thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cf11';
+select rowkey, hex(decimal_col) from t2;
+rowkey	hex(decimal_col)
+val_1.5	000000010F
+val_0.5	0000000105
+val_1234	0000000004D2
+drop table t2;
+#
+# Mapping TIMESTAMP -> int64
+#
+CREATE TABLE t2 (rowkey bigint PRIMARY KEY, datecol timestamp) ENGINE=CASSANDRA
+thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cf4';
+insert into t2 values (1, '2012-08-29 01:23:45');
+INSERT INTO t2 VALUES (10,'2012-08-29 01:23:46');
+drop table t2;
+CREATE TABLE t2 (rowkey bigint PRIMARY KEY, datecol bigint) ENGINE=CASSANDRA
+thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cf4';
+select * from t2;
+rowkey	datecol
+1	1346189025000
+10	1346189026000
+drop table t2;

mysql-test/t/cassandra.test

@@ -73,6 +73,11 @@ insert into cf9 (rowkey, varint_col) values ('val-01', 1);
 insert into cf9 (rowkey, varint_col) values ('val-0x123456', 1193046);
 insert into cf9 (rowkey, varint_col) values ('val-0x12345678', 305419896);
 
+create columnfamily cf11 (rowkey varchar primary key, decimal_col decimal);
+insert into cf11 (rowkey, decimal_col) values ('val_0.5', 0.5);
+insert into cf11 (rowkey, decimal_col) values ('val_1.5', 1.5);
+insert into cf11 (rowkey, decimal_col) values ('val_1234', 1234);
+
 EOF
 --error 0,1,2
 --system cqlsh -3 -f $MYSQLTEST_VARDIR/cassandra_test_init.cql
@@ -436,6 +441,28 @@ CREATE TABLE t2 (rowkey varchar(32) PRIMARY KEY, varint_col varbinary(2)) ENGINE
 select rowkey, hex(varint_col) from t2;
 drop table t2;
 
+--echo #
+--echo # Decimal datatype support
+--echo #
+CREATE TABLE t2 (rowkey varchar(32) PRIMARY KEY, decimal_col varbinary(32)) ENGINE=CASSANDRA
+  thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cf11';
+select rowkey, hex(decimal_col) from t2;
+drop table t2;
+
+--echo #
+--echo # Mapping TIMESTAMP -> int64
+--echo #
+CREATE TABLE t2 (rowkey bigint PRIMARY KEY, datecol timestamp) ENGINE=CASSANDRA
+  thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cf4';
+insert into t2 values (1, '2012-08-29 01:23:45');
+INSERT INTO t2 VALUES (10,'2012-08-29 01:23:46');
+drop table t2;
+
+CREATE TABLE t2 (rowkey bigint PRIMARY KEY, datecol bigint) ENGINE=CASSANDRA
+  thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cf4';
+select * from t2;
+
+drop table t2;
 
 ############################################################################
 ## Cassandra cleanup

storage/cassandra/ha_cassandra.cc

@@ -850,6 +850,7 @@ const char * const validator_boolean= "org.apache.cassandra.db.marshal.BooleanTy
 
 /* VARINTs are stored as big-endian big numbers. */
 const char * const validator_varint= "org.apache.cassandra.db.marshal.IntegerType";
+const char * const validator_decimal= "org.apache.cassandra.db.marshal.DecimalType";
 
 
 ColumnDataConverter *map_field_to_validator(Field *field, const char *validator_name)
@@ -869,6 +870,7 @@ ColumnDataConverter *map_field_to_validator(Field *field, const char *validator_
     {
       bool is_counter= false;
       if (!strcmp(validator_name, validator_bigint) ||
+          !strcmp(validator_name, validator_timestamp) ||
           (is_counter= !strcmp(validator_name, validator_counter)))
         res= new BigintDataConverter(!is_counter);
       break;
@@ -913,7 +915,8 @@ ColumnDataConverter *map_field_to_validator(Field *field, const char *validator_
       */
       if (field->type() == MYSQL_TYPE_VARCHAR && 
           field->binary() &&
-          !strcmp(validator_name, validator_varint))
+          (!strcmp(validator_name, validator_varint) ||
+           !strcmp(validator_name, validator_decimal)))
       {
         res= new StringCopyConverter(field->field_length);
         break;