Cassandra Storage Engine:

- Partially address review feedback.
- Update cassandra.test result result
- make cassandra.test timezone-agnostic

mysql-test/r/cassandra.result

 drop table if exists t0, t1;
 create table t1 (a int) engine=cassandra 
 thrift_host='localhost' keyspace='foo' column_family='colfam';
-ERROR 42000: Incorrect column name 'First column must be NOT NULL'
+ERROR 42000: This table type requires a primary key
 create table t1 (a int primary key, b int) engine=cassandra
 thrift_host='localhost' keyspace='foo' column_family='colfam';
 ERROR HY000: Unable to connect to foreign data source: Default TException. [Keyspace foo does not exist]
@@ -356,6 +356,8 @@ drop table t2;
 #
 # Mapping TIMESTAMP -> int64
 #
+set @save_tz= @@time_zone;
+set time_zone='UTC';
 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');
@@ -365,10 +367,11 @@ 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	1346192625000
-10	1346192626000
+1	1346203425000
+10	1346203426000
 delete from t2;
 drop table t2;
+set time_zone=@save_tz;
 #
 # Check whether changing parameters with ALTER TABLE works.
 #
@@ -555,3 +558,25 @@ ENGINE=CASSANDRA thrift_host='localhost' keyspace='mariadbtest2' column_family =
 insert into t1 values (1, NULL);
 delete from t1;
 DROP TABLE t1;
+#
+# strange side effect of Cassandra - remiving all columns of primary
+# key removes all row.
+#
+CREATE TABLE t1 (rowkey int PRIMARY KEY, dyn blob DYNAMIC_COLUMN_STORAGE=yes) 
+ENGINE=CASSANDRA thrift_host='localhost' keyspace='mariadbtest2' column_family = 'cfd2';
+INSERT INTO t1 VALUES(2,column_create("ab","ab"));
+select rowkey, column_json(dyn) from t1;
+rowkey	column_json(dyn)
+2	[{"ab":"ab"}]
+UPDATE t1 set dyn=NULL;
+select rowkey, column_json(dyn) from t1;
+rowkey	column_json(dyn)
+INSERT INTO t1 VALUES(2,column_create("ab","ab"));
+select rowkey, column_json(dyn) from t1;
+rowkey	column_json(dyn)
+2	[{"ab":"ab"}]
+UPDATE t1 set dyn="";
+select rowkey, column_json(dyn) from t1;
+rowkey	column_json(dyn)
+delete from t1;
+DROP TABLE t1;

mysql-test/t/cassandra.test

@@ -8,9 +8,9 @@ drop table if exists t0, t1;
 --enable_warnings
 
 # Test various errors on table creation.
---error ER_WRONG_COLUMN_NAME
+--error ER_REQUIRES_PRIMARY_KEY
 create table t1 (a int) engine=cassandra 
-  thrift_host='localhost' keyspace='foo' column_family='colfam';
+ thrift_host='localhost' keyspace='foo' column_family='colfam';
 
 --error ER_CONNECT_TO_FOREIGN_DATA_SOURCE
 create table t1 (a int primary key, b int) engine=cassandra
@@ -466,6 +466,8 @@ drop table t2;
 --echo #
 --echo # Mapping TIMESTAMP -> int64
 --echo #
+set @save_tz= @@time_zone;
+set time_zone='UTC';
 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');
@@ -477,6 +479,7 @@ CREATE TABLE t2 (rowkey bigint PRIMARY KEY, datecol bigint) ENGINE=CASSANDRA
 select * from t2;
 delete from t2;
 drop table t2;
+set time_zone=@save_tz;
 
 --echo #
 --echo # Check whether changing parameters with ALTER TABLE works.

storage/cassandra/CMakeLists.txt

@@ -14,7 +14,8 @@ SET(cassandra_sources
 #INCLUDE_DIRECTORIES(BEFORE ${Boost_INCLUDE_DIRS})
 
 #INCLUDE_DIRECTORIES(AFTER /usr/local/include/thrift)
-INCLUDE_DIRECTORIES(AFTER /home/buildbot/build/thrift-inst/include/thrift/) 
+#INCLUDE_DIRECTORIES(AFTER /home/buildbot/build/thrift-inst/include/thrift/) 
+INCLUDE_DIRECTORIES(AFTER /home/psergey/cassandra/thrift/include/thrift/) 
 
 # 
 STRING(REPLACE "-fno-exceptions" "" CMAKE_CXX_FLAGS ${CMAKE_CXX_FLAGS})

storage/cassandra/cassandra_se.cc

@@ -30,15 +30,6 @@ 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)
 */

storage/cassandra/ha_cassandra.cc

@@ -542,13 +542,6 @@ int ha_cassandra::create(const char *name, TABLE *table_arg,
   int res;
   DBUG_ENTER("ha_cassandra::create");
 
-  Field **pfield= table_arg->s->field;
-  if (!((*pfield)->flags & NOT_NULL_FLAG))
-  {
-    my_error(ER_WRONG_COLUMN_NAME, MYF(0), "First column must be NOT NULL");
-    DBUG_RETURN(HA_WRONG_CREATE_OPTION);
-  }
-
   if (table_arg->s->keys != 1 || table_arg->s->primary_key !=0 ||
       table_arg->key_info[0].key_parts != 1 ||
       table_arg->key_info[0].key_part[0].fieldnr != 1)
@@ -1680,7 +1673,7 @@ void ha_cassandra::print_conversion_error(const char *field_name,
   char buf[32];
   char *p= cass_value;
   size_t i= 0;
-  for (; (i < (int)sizeof(buf)-1) && (p < cass_value + cass_value_len); p++)
+  for (; (i < sizeof(buf)-1) && (p < cass_value + cass_value_len); p++)
   {
     buf[i++]= map2number[(*p >> 4) & 0xF];
     buf[i++]= map2number[*p & 0xF];
@@ -2163,7 +2156,7 @@ int ha_cassandra::info(uint flag)
   if (flag & HA_STATUS_VARIABLE)
   {
     stats.records= 1000;
-    //TODO: any other stats?
+    stats.deleted= 0;
   }
   if (flag & HA_STATUS_CONST)
   {
@@ -2346,55 +2339,6 @@ int ha_cassandra::multi_range_read_explain_info(uint mrr_mode, char *str, size_t
 }
 
 
-/////////////////////////////////////////////////////////////////////////////
-// Dummy implementations start
-/////////////////////////////////////////////////////////////////////////////
-
-
-int ha_cassandra::index_next(uchar *buf)
-{
-  int rc;
-  DBUG_ENTER("ha_cassandra::index_next");
-  rc= HA_ERR_WRONG_COMMAND;
-  DBUG_RETURN(rc);
-}
-
-
-int ha_cassandra::index_prev(uchar *buf)
-{
-  int rc;
-  DBUG_ENTER("ha_cassandra::index_prev");
-  rc= HA_ERR_WRONG_COMMAND;
-  DBUG_RETURN(rc);
-}
-
-
-int ha_cassandra::index_first(uchar *buf)
-{
-  int rc;
-  DBUG_ENTER("ha_cassandra::index_first");
-  rc= HA_ERR_WRONG_COMMAND;
-  DBUG_RETURN(rc);
-}
-
-int ha_cassandra::index_last(uchar *buf)
-{
-  int rc;
-  DBUG_ENTER("ha_cassandra::index_last");
-  rc= HA_ERR_WRONG_COMMAND;
-  DBUG_RETURN(rc);
-}
-
-
-ha_rows ha_cassandra::records_in_range(uint inx, key_range *min_key,
-                                     key_range *max_key)
-{
-  DBUG_ENTER("ha_cassandra::records_in_range");
-  //DBUG_RETURN(10);                         // low number to force index usage
-  DBUG_RETURN(HA_POS_ERROR);
-}
-
-
 class Column_name_enumerator_impl : public Column_name_enumerator
 {
   ha_cassandra *obj;
@@ -2550,13 +2494,6 @@ int ha_cassandra::update_row(const uchar *old_data, uchar *new_data)
 }
 
 
-int ha_cassandra::extra(enum ha_extra_function operation)
-{
-  DBUG_ENTER("ha_cassandra::extra");
-  DBUG_RETURN(0);
-}
-
-
 /* The following function was copied from ha_blackhole::store_lock: */
 THR_LOCK_DATA **ha_cassandra::store_lock(THD *thd,
                                          THR_LOCK_DATA **to,
@@ -2595,18 +2532,20 @@ THR_LOCK_DATA **ha_cassandra::store_lock(THD *thd,
 }
 
 
-int ha_cassandra::external_lock(THD *thd, int lock_type)
+ha_rows ha_cassandra::records_in_range(uint inx, key_range *min_key,
+                                       key_range *max_key)
 {
-  DBUG_ENTER("ha_cassandra::external_lock");
-  DBUG_RETURN(0);
+  DBUG_ENTER("ha_cassandra::records_in_range");
+  DBUG_RETURN(HA_POS_ERROR); /* Range scans are not supported */
 }
 
+
 int ha_cassandra::delete_table(const char *name)
 {
   DBUG_ENTER("ha_cassandra::delete_table");
   /*
     Cassandra table is just a view. Dropping it doesn't affect the underlying
-    column family.
+    column family, so we do nothing here.
   */
   DBUG_RETURN(0);
 }
@@ -2632,9 +2571,15 @@ bool ha_cassandra::check_if_incompatible_data(HA_CREATE_INFO *info,
 }
 
 
-/////////////////////////////////////////////////////////////////////////////
-// Dummy implementations end
-/////////////////////////////////////////////////////////////////////////////
+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
+  my_vsnprintf(err_buffer, sizeof(err_buffer), format, ap);
+  va_end(ap);
+}
+
 
 static int show_cassandra_vars(THD *thd, SHOW_VAR *var, char *buff)
 {
@@ -2655,6 +2600,7 @@ static struct st_mysql_show_var func_status[]=
   {0,0,SHOW_UNDEF}
 };
 
+
 maria_declare_plugin(cassandra)
 {
   MYSQL_STORAGE_ENGINE_PLUGIN,

storage/cassandra/ha_cassandra.h

@@ -139,18 +139,13 @@ class ha_cassandra: public handler
   */
   ulonglong table_flags() const
   {
-    /*
-      HA_BINLOG_STMT_CAPABLE
-        We are saying that this engine is just statement capable to have
-        an engine that can only handle statement-based logging. This is
-        used in testing.
-      HA_REC_NOT_IN_SEQ
-        If we don't set it, filesort crashes, because it assumes rowids are
-        1..8 byte numbers
-    */
     return HA_BINLOG_STMT_CAPABLE |
-           HA_REC_NOT_IN_SEQ;
-
+           HA_REC_NOT_IN_SEQ |
+           HA_NO_TRANSACTIONS |
+           HA_REQUIRE_PRIMARY_KEY |
+           HA_PRIMARY_KEY_IN_READ_INDEX |
+           HA_PRIMARY_KEY_REQUIRED_FOR_POSITION |
+           HA_NO_AUTO_INCREMENT;
   }
 
   /** @brief
@@ -239,65 +234,13 @@ class ha_cassandra: public handler
   CASSANDRA_TYPE_DEF * get_cassandra_field_def(char *cass_name,
                                                int cass_name_length);
 public:
+  int open(const char *name, int mode, uint test_if_locked);
+  int close(void);
 
-  /*
-    Everything below are methods that we implement in ha_example.cc.
-
-    Most of these methods are not obligatory, skip them and
-    MySQL will treat them as not implemented
-  */
-  /** @brief
-    We implement this in ha_example.cc; it's a required method.
-  */
-  int open(const char *name, int mode, uint test_if_locked);    // required
-
-  /** @brief
-    We implement this in ha_example.cc; it's a required method.
-  */
-  int close(void);                                              // required
-
-  /** @brief
-    We implement this in ha_example.cc. It's not an obligatory method;
-    skip it and and MySQL will treat it as not implemented.
-  */
   int write_row(uchar *buf);
-
-  /** @brief
-    We implement this in ha_example.cc. It's not an obligatory method;
-    skip it and and MySQL will treat it as not implemented.
-  */
   int update_row(const uchar *old_data, uchar *new_data);
-
-  /** @brief
-    We implement this in ha_example.cc. It's not an obligatory method;
-    skip it and and MySQL will treat it as not implemented.
-  */
   int delete_row(const uchar *buf);
 
-  /** @brief
-    We implement this in ha_example.cc. It's not an obligatory method;
-    skip it and and MySQL will treat it as not implemented.
-  */
-  int index_next(uchar *buf);
-
-  /** @brief
-    We implement this in ha_example.cc. It's not an obligatory method;
-    skip it and and MySQL will treat it as not implemented.
-  */
-  int index_prev(uchar *buf);
-
-  /** @brief
-    We implement this in ha_example.cc. It's not an obligatory method;
-    skip it and and MySQL will treat it as not implemented.
-  */
-  int index_first(uchar *buf);
-
-  /** @brief
-    We implement this in ha_example.cc. It's not an obligatory method;
-    skip it and and MySQL will treat it as not implemented.
-  */
-  int index_last(uchar *buf);
-
   /** @brief
     Unlike index_init(), rnd_init() can be called two consecutive times
     without rnd_end() in between (it only makes sense if scan=1). In this
@@ -312,8 +255,6 @@ class ha_cassandra: public handler
   int rnd_pos(uchar *buf, uchar *pos);                          ///< required
   void position(const uchar *record);                           ///< required
   int info(uint);                                               ///< required
-  int extra(enum ha_extra_function operation);
-  int external_lock(THD *thd, int lock_type);                   ///< required
   int delete_all_rows(void);
   ha_rows records_in_range(uint inx, key_range *min_key,
                            key_range *max_key);