Convert index_id into (cf_id, index_id).

Summary:
Eventually we want to be able to import tables and the index_id values from the other tables
may collide with existing index_ids.  Thus we are converting the global index identifier from a single
integer to the combination of a column family and index id.

@update-submodule: rocksdb

Test Plan: MTR

Reviewers: yoshinorim, spetrunia, hermanlee4, maykov

Reviewed By: maykov

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

Author: jkedgar

storage/rocksdb/ha_rocksdb.cc

@@ -21,6 +21,9 @@
 /* The C++ file's header */
 #include "./ha_rocksdb.h"
 
+/* C++ standard header files */
+#include <map>
+
 /* MySQL includes */
 #include "./my_bit.h"
 #include "./my_stacktrace.h"
@@ -979,7 +982,7 @@ static bool stop_drop_index_thread;
 class Rdb_CompactionFilter : public rocksdb::CompactionFilter
 {
 public:
-  Rdb_CompactionFilter() {}
+  explicit Rdb_CompactionFilter(uint32_t _cf_id) : cf_id(_cf_id) {}
   ~Rdb_CompactionFilter()
   {
     print_compaction_status();
@@ -996,18 +999,21 @@ class Rdb_CompactionFilter : public rocksdb::CompactionFilter
                       bool* value_changed) const override {
 
     DBUG_ASSERT(key.size() >= sizeof(uint32));
-    uint32 index= read_big_uint4((const uchar*)key.data());
-    DBUG_ASSERT(index >= 1);
 
-    if (index != prev_index) // processing new index id
+    GL_INDEX_ID gl_index_id;
+    gl_index_id.cf_id= cf_id;
+    gl_index_id.index_id= read_big_uint4((const uchar*)key.data());
+    DBUG_ASSERT(gl_index_id.index_id >= 1);
+
+    if (gl_index_id != prev_index)  // processing new index id
     {
       if (num_deleted > 0)
       {
         print_compaction_status();
         num_deleted= 0;
       }
-      should_delete= dict_manager.is_drop_index_ongoing(index);
-      prev_index= index;
+      should_delete= dict_manager.is_drop_index_ongoing(gl_index_id);
+      prev_index= gl_index_id;
     }
 
     if (should_delete)
@@ -1024,11 +1030,14 @@ class Rdb_CompactionFilter : public rocksdb::CompactionFilter
   {
     if (num_deleted > 0)
       sql_print_information("RocksDB: Compacting away elements from dropped "
-                            "index %u: %llu", prev_index, num_deleted);
+                            "index (%u,%u): %llu", prev_index.cf_id,
+                            prev_index.index_id, num_deleted);
   }
 
+  // Column family for this compaction filter
+  const uint32_t cf_id;
   // Index id of the previous record
-  mutable uint32 prev_index= 0;
+  mutable GL_INDEX_ID prev_index= {0, 0};
   // Number of rows deleted for the same index id
   mutable uint64 num_deleted= 0;
   // Current index id should be deleted or not (should be deleted if true)
@@ -1049,7 +1058,7 @@ class Rdb_CompactionFilterFactory : public rocksdb::CompactionFilterFactory
   std::unique_ptr<rocksdb::CompactionFilter> CreateCompactionFilter(
       const rocksdb::CompactionFilter::Context& context) {
     return std::unique_ptr<rocksdb::CompactionFilter>(
-      new Rdb_CompactionFilter);
+      new Rdb_CompactionFilter(context.column_family_id));
   }
 };
 
@@ -3041,18 +3050,16 @@ int ha_rocksdb::create_key_defs(TABLE *table_arg, const char *db_table,
   for (i= 0; i < table_arg->s->keys; i++)
   {
     uint index_id= ddl_manager.get_and_update_next_number(&dict_manager);
-    uint16_t index_dict_version;
+    uint16_t index_dict_version= RDBSE_KEYDEF::INDEX_INFO_VERSION_GLOBAL_ID;
     uchar index_type;
     uint16_t kv_version;
     if (i == table_arg->s->primary_key)
     {
-      index_dict_version= RDBSE_KEYDEF::INDEX_CF_MAPPING_VERSION_KV_FORMAT;
       index_type= RDBSE_KEYDEF::INDEX_TYPE_PRIMARY;
       kv_version= RDBSE_KEYDEF::PRIMARY_FORMAT_VERSION_INITIAL;
     }
     else
     {
-      index_dict_version= RDBSE_KEYDEF::INDEX_CF_MAPPING_VERSION_KV_FORMAT;
       index_type= RDBSE_KEYDEF::INDEX_TYPE_SECONDARY;
       kv_version= RDBSE_KEYDEF::SECONDARY_FORMAT_VERSION_INITIAL;
     }
@@ -5549,43 +5556,31 @@ void* drop_index_thread(void*)
     assert(ret == 0 || ret == ETIMEDOUT);
     mysql_mutex_unlock(&drop_index_interrupt_mutex);
 
-    std::vector<uint32> indices;
-    dict_manager.get_drop_indexes_ongoing(indices);
+    std::vector<GL_INDEX_ID> indices;
+    dict_manager.get_drop_indexes_ongoing(&indices);
     if (!indices.empty()) {
-      std::unordered_set<uint32> finished;
+      std::unordered_set<GL_INDEX_ID> finished;
       rocksdb::ReadOptions read_opts;
       read_opts.total_order_seek = true; // disable bloom filter
 
       for (auto d : indices) {
-        uint16 index_dict_version=0;
-        uchar index_type= 0;
-        uint16 kv_version= 0;
-        uint32 cf_id= 0;
         uint32 cf_flags= 0;
-        if (!dict_manager.get_cf_id(d, &index_dict_version, &index_type,
-                                    &kv_version, &cf_id))
-        {
-          sql_print_error("RocksDB: Failed to get column family id "
-                          "from index id %u. MyRocks data dictionary may "
-                          "get corrupted.", d);
-          abort_with_stack_traces();
-        }
-        if (!dict_manager.get_cf_flags(cf_id, &cf_flags))
+        if (!dict_manager.get_cf_flags(d.cf_id, &cf_flags))
         {
           sql_print_error("RocksDB: Failed to get column family flags "
                           "from cf id %u. MyRocks data dictionary may "
-                          "get corrupted.", cf_id);
+                          "get corrupted.", d.cf_id);
           abort_with_stack_traces();
         }
-        rocksdb::ColumnFamilyHandle* cfh= cf_manager.get_cf(cf_id);
+        rocksdb::ColumnFamilyHandle* cfh= cf_manager.get_cf(d.cf_id);
         DBUG_ASSERT(cfh);
         bool is_reverse_cf= cf_flags & RDBSE_KEYDEF::REVERSE_CF_FLAG;
         std::unique_ptr<rocksdb::Iterator> it(
           rdb->NewIterator(read_opts, cfh));
 
         bool index_removed= false;
         uchar key_buf[RDBSE_KEYDEF::INDEX_NUMBER_SIZE]= {0};
-        store_big_uint4(key_buf, d);
+        store_big_uint4(key_buf, d.index_id);
         rocksdb::Slice key = rocksdb::Slice((char*)key_buf, sizeof(key_buf));
         it->Seek(key);
         if (is_reverse_cf)
@@ -5938,7 +5933,7 @@ void compute_optimizer_stats()
     }
   }
 
-  std::map<uint32_t, MyRocksTablePropertiesCollector::IndexStats>
+  std::map<GL_INDEX_ID, MyRocksTablePropertiesCollector::IndexStats>
     statsFromProps;
   for (auto cf : cfs) {
     rocksdb::TablePropertiesCollection props;
@@ -5953,13 +5948,13 @@ void compute_optimizer_stats()
   std::vector<MyRocksTablePropertiesCollector::IndexStats> stats;
   for (auto& keydef : keydefs)
   {
-    auto it = statsFromProps.find(keydef->get_index_number());
+    auto it = statsFromProps.find(keydef->get_gl_index_id());
     if (it != statsFromProps.end()) {
       stats.push_back(it->second);
     } else {
       // This index didn't have any associated stats in SST properties.
       // It might be an empty table, for example.
-      stats.emplace_back(keydef->get_index_number());
+      stats.emplace_back(keydef->get_gl_index_id());
     }
 
     uchar buf[2*RDBSE_KEYDEF::INDEX_NUMBER_SIZE];

storage/rocksdb/ha_rocksdb.h

@@ -115,6 +115,53 @@ class Rdb_transaction;
 
 const char *const rocksdb_hton_name= "ROCKSDB";
 
+typedef struct _gl_index_id_s {
+  uint32_t cf_id;
+  uint32_t index_id;
+  bool operator==(const struct _gl_index_id_s& other) const
+  {
+    return cf_id == other.cf_id && index_id == other.index_id;
+  }
+  bool operator!=(const struct _gl_index_id_s& other) const
+  {
+    return cf_id != other.cf_id || index_id != other.index_id;
+  }
+  bool operator<(const struct _gl_index_id_s& other) const
+  {
+    return cf_id < other.cf_id ||
+        (cf_id == other.cf_id && index_id < other.index_id);
+  }
+  bool operator<=(const struct _gl_index_id_s& other) const
+  {
+    return cf_id < other.cf_id ||
+        (cf_id == other.cf_id && index_id <= other.index_id);
+  }
+  bool operator>(const struct _gl_index_id_s& other) const
+  {
+    return cf_id > other.cf_id ||
+        (cf_id == other.cf_id && index_id > other.index_id);
+  }
+  bool operator>=(const struct _gl_index_id_s& other) const
+  {
+    return cf_id > other.cf_id ||
+        (cf_id == other.cf_id && index_id >= other.index_id);
+  }
+} GL_INDEX_ID;
+
+/* Provide hash function for GL_INDEX_ID so we can include it in sets */
+namespace std {
+  template <>
+  struct hash<GL_INDEX_ID>
+  {
+    std::size_t operator()(const GL_INDEX_ID& gl_index_id) const
+    {
+      uint64_t val = ((uint64_t) gl_index_id.cf_id << 32 |
+                      (uint64_t) gl_index_id.index_id);
+      return std::hash<uint64_t>()(val);
+    }
+  };
+}  // namespace std
+
 /** @brief
   Class definition for the storage engine
 */

storage/rocksdb/properties_collector.cc

@@ -17,6 +17,9 @@
 /* This C++ file's header file */
 #include "./properties_collector.h"
 
+/* Standard C++ header files */
+#include <map>
+
 /* MySQL header files */
 #include "./sql_array.h"
 
@@ -31,8 +34,10 @@ uint64_t rocksdb_num_sst_entry_other = 0;
 
 MyRocksTablePropertiesCollector::MyRocksTablePropertiesCollector(
   Table_ddl_manager* ddl_manager,
-  CompactionParams params
+  CompactionParams params,
+  uint32_t cf_id
 ) :
+    cf_id_(cf_id),
     ddl_manager_(ddl_manager),
     rows_(0l), deleted_rows_(0l), max_deleted_rows_(0l),
     params_(params)
@@ -70,15 +75,17 @@ MyRocksTablePropertiesCollector::AddUserKey(
       break;
     }
 
-
-    uint32_t index_number = read_big_uint4((const uchar*)key.data());
-    if (stats_.empty() || index_number != stats_.back().index_number) {
+    GL_INDEX_ID gl_index_id;
+    gl_index_id.cf_id = cf_id_;
+    gl_index_id.index_id = read_big_uint4((const uchar*)key.data());
+    if (stats_.empty() || gl_index_id != stats_.back().gl_index_id)
+    {
       keydef_ = NULL;
       // starting a new table
       // add the new element into stats_
-      stats_.push_back(IndexStats(index_number));
+      stats_.push_back(IndexStats(gl_index_id));
       if (ddl_manager_) {
-        keydef_ = ddl_manager_->get_copy_of_keydef(index_number);
+        keydef_ = ddl_manager_->get_copy_of_keydef(gl_index_id);
       }
       if (keydef_) {
         // resize the array to the number of columns.
@@ -154,12 +161,12 @@ rocksdb::Status
 MyRocksTablePropertiesCollector::Finish(
   rocksdb::UserCollectedProperties* properties
 ) {
-  std::vector<uint32_t> changed_indexes;
+  std::vector<GL_INDEX_ID> changed_indexes;
   changed_indexes.resize(stats_.size());
   std::transform(
     stats_.begin(), stats_.end(),
     changed_indexes.begin(),
-    [](const IndexStats& s) {return s.index_number;}
+    [](const IndexStats& s) {return s.gl_index_id;}
   );
   ddl_manager_->add_changed_indexes(changed_indexes);
   properties->insert({INDEXSTATS_KEY, IndexStats::materialize(stats_)});
@@ -203,11 +210,12 @@ MyRocksTablePropertiesCollector::GetReadableStats(
   const MyRocksTablePropertiesCollector::IndexStats& it
 ) {
   std::string s;
-  s.append(std::to_string(it.index_number));
-  s.append(":{name:");
+  s.append("(");
+  s.append(std::to_string(it.gl_index_id.cf_id));
+  s.append(", ");
+  s.append(std::to_string(it.gl_index_id.index_id));
+  s.append("):{name:");
   s.append(it.name);
-  s.append(", number:");
-  s.append(std::to_string(it.index_number));
   s.append(", size:");
   s.append(std::to_string(it.data_size));
   s.append(", rows:");
@@ -244,8 +252,8 @@ MyRocksTablePropertiesCollector::GetStatsFromTableProperties(
 */
 void MyRocksTablePropertiesCollector::GetStats(
   const rocksdb::TablePropertiesCollection& collection,
-  const std::unordered_set<uint32_t>& index_numbers,
-  std::map<uint32_t, MyRocksTablePropertiesCollector::IndexStats>& ret
+  const std::unordered_set<GL_INDEX_ID>& index_numbers,
+  std::map<GL_INDEX_ID, MyRocksTablePropertiesCollector::IndexStats>& ret
 ) {
   for (auto it : collection) {
     const auto& user_properties = it.second->user_collected_properties;
@@ -254,8 +262,8 @@ void MyRocksTablePropertiesCollector::GetStats(
       std::vector<IndexStats> stats;
       if (IndexStats::unmaterialize(it2->second, stats) == 0) {
         for (auto it3 : stats) {
-          if (index_numbers.count(it3.index_number) != 0) {
-            ret[it3.index_number].merge(it3);
+          if (index_numbers.count(it3.gl_index_id) != 0) {
+            ret[it3.gl_index_id].merge(it3);
           }
         }
       }
@@ -272,7 +280,8 @@ std::string MyRocksTablePropertiesCollector::IndexStats::materialize(
   String ret;
   write_short(&ret, INDEX_STATS_VERSION);
   for (auto i : stats) {
-    write_int(&ret, i.index_number);
+    write_int(&ret, i.gl_index_id.cf_id);
+    write_int(&ret, i.gl_index_id.index_id);
     assert(sizeof i.data_size <= 8);
     write_int64(&ret, i.data_size);
     write_int64(&ret, i.rows);
@@ -302,15 +311,17 @@ int MyRocksTablePropertiesCollector::IndexStats::unmaterialize(
   while (p < p2) {
     IndexStats stats;
     if (p+
-       sizeof(stats.index_number)+
+       sizeof(stats.gl_index_id.cf_id)+
+       sizeof(stats.gl_index_id.index_id)+
        sizeof(stats.data_size)+
        sizeof(stats.rows)+
        sizeof(stats.approximate_size)+
        sizeof(uint64) > p2)
     {
       return 1;
     }
-    stats.index_number = read_int(&p);
+    stats.gl_index_id.cf_id = read_int(&p);
+    stats.gl_index_id.index_id = read_int(&p);
     stats.data_size = read_int64(&p);
     stats.rows = read_int64(&p);
     stats.approximate_size = read_int64(&p);
@@ -334,7 +345,7 @@ int MyRocksTablePropertiesCollector::IndexStats::unmaterialize(
   for the index based on stats for each sst
 */
 void MyRocksTablePropertiesCollector::IndexStats::merge(const IndexStats& s) {
-  index_number = s.index_number;
+  gl_index_id = s.gl_index_id;
   rows += s.rows;
   data_size += s.data_size;
   if (distinct_keys_per_prefix.size() < s.distinct_keys_per_prefix.size()) {