Fix SIGSEGV on CREATE TABLE contention (issue mysql#104)

Summary:
MyRocks did strict dictionary consistency check that
newer index_id (passed by CREATE TABLE) must be larger than
max_index_id stored in data dictionary. This is not always true. Suppose
the following conditions.
CREATE TABLE t1:
1) index_id= ddl_manager.get_next_number();
4) dict_manager.update_max_index_id(batch, index_id)

CREATE TABLE t2:
2) index_id= ddl_manager.get_next_number();
3) dict_manager.update_max_index_id(batch, index_id)

MyRocks returned an error on CREATE TABLE t1 because of dictionary
condition check (dict_manager.update_max_index_id()). Then MyRocks
crashed later by memory violation, because ddl_manager.put_and_write()
was not cleaned up after failure.

This diff fixes the issue by updating max_index_id in dictionary
when getting new index id. This diff also renaming function name from
ddl_manager.get_next_number() to ddl_manager.get_and_update_next_number().

Test Plan: mtr

Reviewers: maykov, spetrunia, jkedgar, hermanlee4

Reviewed By: hermanlee4

Differential Revision: https://reviews.facebook.net/D45963

Author: yoshinorim

storage/rocksdb/ha_rocksdb.cc

@@ -2748,7 +2748,7 @@ int ha_rocksdb::create_key_defs(TABLE *table_arg, const char *db_table, uint len
     if (!cf_handle)
       goto error;
 
-    index_id= ddl_manager.get_next_number();
+    index_id= ddl_manager.get_and_update_next_number(&dict_manager);
     if (!(key_descr[i]= new RDBSE_KEYDEF(index_id, i,
                                          cf_handle,
                                          is_cf_name_reverse(comment),
@@ -2762,7 +2762,6 @@ int ha_rocksdb::create_key_defs(TABLE *table_arg, const char *db_table, uint len
   tbl_def->dbname_tablename.append(db_table, len);
   dict_manager.lock();
   write_err= ddl_manager.put_and_write(tbl_def, batch)
-             || dict_manager.update_max_index_id(batch, index_id)
              || dict_manager.commit(batch);
   dict_manager.unlock();
 

storage/rocksdb/rdb_datadic.cc

@@ -2199,3 +2199,16 @@ Dict_manager::get_stats(
 
   return MyRocksTablePropertiesCollector::IndexStats();
 }
+
+uint Sequence_generator::get_and_update_next_number(Dict_manager *dict)
+{
+  uint res;
+  mysql_mutex_lock(&mutex);
+  res= next_number++;
+  std::unique_ptr<rocksdb::WriteBatch> wb= dict->begin();
+  rocksdb::WriteBatch *batch= wb.get();
+  dict->update_max_index_id(batch, res);
+  dict->commit(batch);
+  mysql_mutex_unlock(&mutex);
+  return res;
+}

storage/rocksdb/rdb_datadic.h

@@ -552,14 +552,7 @@ class Sequence_generator
     next_number= initial_number;
   }
 
-  uint get_next_number()
-  {
-    uint res;
-    mysql_mutex_lock(&mutex);
-    res= next_number++;
-    mysql_mutex_unlock(&mutex);
-    return res;
-  }
+  uint get_and_update_next_number(Dict_manager *dict);
 
   void cleanup()
   {
@@ -609,7 +602,8 @@ class Table_ddl_manager
   bool rename(uchar *from, uint from_len, uchar *to, uint to_len,
               rocksdb::WriteBatch *batch);
 
-  uint get_next_number() { return sequence.get_next_number(); }
+  uint get_and_update_next_number(Dict_manager *dict)
+    { return sequence.get_and_update_next_number(dict); }
   void add_changed_indexes(const std::vector<uint32_t>& changed_indexes);
   std::unordered_set<uint32_t> get_changed_indexes();