fix-myrocks-version.diff

diff --git a/storage/rocksdb/ha_rocksdb.cc b/storage/rocksdb/ha_rocksdb.cc
index e8f54485197..30e7f4e728e 100644
--- a/storage/rocksdb/ha_rocksdb.cc
+++ b/storage/rocksdb/ha_rocksdb.cc
@@ -223,10 +223,12 @@ void dbug_dump_database(rocksdb::DB *db);
 static handler *rocksdb_create_handler(my_core::handlerton *hton,
                                        my_core::TABLE_SHARE *table_arg,
                                        my_core::MEM_ROOT *mem_root);
+
 void save_table_version(rocksdb::WriteBatch *wb, const char *path,
                         ulonglong version);
-
 ulonglong get_table_version(const char *path);
+void delete_table_version(rocksdb::WriteBatch *wb,
+                          const char *path);
 
 static rocksdb::CompactRangeOptions getCompactRangeOptions(
     int concurrency = 0) {
@@ -7755,7 +7757,7 @@ int ha_rocksdb::create_table(const std::string &table_name,
     dict_manager.unlock();
     goto error;
   }
-  //psergey:
+
   save_table_version(batch, table_arg->s->path.str, 0);
   err = dict_manager.commit(batch);
   if (err != HA_EXIT_SUCCESS) {
@@ -11707,11 +11709,14 @@ int ha_rocksdb::delete_table(Rdb_tbl_def *const tbl) {
 
   dict_manager.add_drop_table(tbl->m_key_descr_arr, tbl->m_key_count, batch);
 
+  std::string path = std::string("./") + tbl->base_dbname() +
+                     "/" + tbl->base_tablename();
   /*
     Remove the table entry in data dictionary (this will also remove it from
     the persistent data dictionary).
   */
   ddl_manager.remove(tbl, batch, true);
+  delete_table_version(batch, path.c_str());
 
   int err = dict_manager.commit(batch);
   if (err) {
@@ -13054,12 +13059,12 @@ bool ha_rocksdb::commit_inplace_alter_table(
       */
       ddl_manager.remove_uncommitted_keydefs(ctx->m_added_indexes);
     }
-    //psergey21-todo: bump the version # here...
-    ulonglong table_ver= table_version();
-    fprintf(stderr, "MYROCKS-VER: ha_rocksdb::commit_inplace_alter_table(): path=%s, version update: %llu -> %llu\n",
-            table->s->path.str, table_ver, table_ver+1);
+
+    /*
+      Increment the table version.
+    */
+    ulonglong table_ver = get_table_version(table->s->path.str);
     table_ver++;
-    //TODO: handling for errors?
     save_table_version(batch, table->s->path.str, table_ver);
 
     if (dict_manager.commit(batch)) {
@@ -14602,11 +14607,20 @@ void ha_rocksdb::print_error(int error, myf errflag) {
 
 std::string make_table_version_lookup_key(const char *path) {
   std::string res;
-  res.append("MariaDB-");
+  res.append("MariaDB:table-version:");
   res.append(path);
   return res;
 }
 
+
+/*
+  Save the table version in the data dictionary
+
+  @param  wb      Where to write
+  @param  path    Table's path "./db_name/table_name"
+  @param  version Version
+*/
+
 void save_table_version(rocksdb::WriteBatch *wb, const char *path,
                         ulonglong version) {
   ulonglong val= htobe64(version);
@@ -14615,15 +14629,23 @@ void save_table_version(rocksdb::WriteBatch *wb, const char *path,
           rocksdb::Slice((const char*)&val, sizeof(val)));
 }
 
+/*
+  @brief Read table's version from the data dictionary
+
+  @param path  The table's path, "./db_name/table_name"
+
+  @return
+    number  TTable version as stored in the data dictionary
+    0       If there's no table version stored.
+   -1       Read error
+*/
+
 ulonglong get_table_version(const char *path) {
-  // TODO: load the data from disk. (Do we have path?)
-  // TABLE->path or TABLE->normalized_path
   auto lookup_key = make_table_version_lookup_key(path);
   std::string value;
   ulonglong res;
 
-  // Then, make a lookup.
-  auto s = dict_manager.get_value(rocksdb::Slice(lookup_key), &value); 
+  auto s = dict_manager.get_value(rocksdb::Slice(lookup_key), &value);
   if (s.IsNotFound()) {
     res = 0;
   } else if (s.ok()) {
@@ -14632,7 +14654,7 @@ ulonglong get_table_version(const char *path) {
       memcpy(&res, value.data(), sizeof(res));
       res = be64toh(res);
     }
-    else 
+    else
       res = ulonglong(-1);
   } else {
     res = ulonglong(-1);
@@ -14640,6 +14662,11 @@ ulonglong get_table_version(const char *path) {
   return res;
 }
 
+void delete_table_version(rocksdb::WriteBatch *wb, const char *path) {
+  auto lookup_key = make_table_version_lookup_key(path);
+  wb->Delete(dict_manager.get_system_cf(), lookup_key);
+}
+
 ulonglong ha_rocksdb::table_version() const {
   return get_table_version(table->s->path.str);
 }