MDEV-31693 Add Catalogs support to RocksDB

storage/rocksdb/ha_rocksdb.cc

@@ -5073,11 +5073,12 @@ static bool rocksdb_rollback_to_savepoint_can_release_mdl(
 */
 static void rocksdb_update_table_stats(
     /* per-table stats callback */
-    void (*cb)(const char *db, const char *tbl, bool is_partition,
-               my_io_perf_t *r, my_io_perf_t *w, my_io_perf_t *r_blob,
-               my_io_perf_t *r_primary, my_io_perf_t *r_secondary,
-               page_stats_t *page_stats, comp_stats_t *comp_stats,
-               int n_lock_wait, int n_lock_wait_timeout, int n_lock_deadlock,
+    void (*cb)(const char *catalog, const char *db, const char *tbl,
+               bool is_partition, my_io_perf_t *r, my_io_perf_t *w,
+               my_io_perf_t *r_blob, my_io_perf_t *r_primary,
+               my_io_perf_t *r_secondary, page_stats_t *page_stats,
+               comp_stats_t *comp_stats, int n_lock_wait,
+               int n_lock_wait_timeout, int n_lock_deadlock,
                const char *engine)) {
   my_io_perf_t io_perf_read;
   my_io_perf_t io_perf_write;
@@ -5102,7 +5103,8 @@ static void rocksdb_update_table_stats(
 
   for (const auto &it : tablenames) {
     Rdb_table_handler *table_handler;
-    std::string str, dbname, tablename, partname;
+    std::string str, catname, dbname, tablename, partname;
+    char catname_sys[NAME_LEN + 1];
     char dbname_sys[NAME_LEN + 1];
     char tablename_sys[NAME_LEN + 1];
     bool is_partition;
@@ -5116,7 +5118,8 @@ static void rocksdb_update_table_stats(
       return;
     }
 
-    if (rdb_split_normalized_tablename(str, &dbname, &tablename, &partname)) {
+    if (rdb_split_normalized_tablename(str, &catname &dbname, &tablename,
+        &partname)) {
       continue;
     }
 
@@ -5157,11 +5160,13 @@ static void rocksdb_update_table_stats(
       Table stats expects our database and table name to be in system encoding,
       not filename format. Convert before calling callback.
      */
+    my_core::filename_to_tablename(catname.c_str(), catname_sys,
+                                   sizeof(catname_sys));
     my_core::filename_to_tablename(dbname.c_str(), dbname_sys,
                                    sizeof(dbname_sys));
     my_core::filename_to_tablename(tablename.c_str(), tablename_sys,
                                    sizeof(tablename_sys));
-    (*cb)(dbname_sys, tablename_sys, is_partition, &io_perf_read,
+    (*cb)(catname_sys, dbname_sys, tablename_sys, is_partition, &io_perf_read,
           &io_perf_write, &io_perf, &io_perf, &io_perf, &page_stats,
           &comp_stats, lock_wait_stats, lock_wait_timeout_stats, deadlock_stats,
           rocksdb_hton_name);
@@ -7629,6 +7634,20 @@ int rdb_normalize_tablename(const std::string &tablename,
 
   *strbuf = tablename.substr(2, pos - 2) + "." + tablename.substr(pos + 1);
 
+  if (using_catalogs) {
+    std::string tablename2 = *strbuf;
+    size_t pos2 = tablename2.find_first_of(FN_LIBCHAR, pos + 1);
+    if (pos2 == std::string::npos) {
+      pos2 = tablename2.find_first_of(FN_LIBCHAR2, pos + 1);
+    }
+
+    if (pos2 == std::string::npos) {
+      DBUG_ASSERT(0);  // We were not passed table name?
+      return HA_ERR_ROCKSDB_INVALID_TABLE;
+    }
+    *strbuf = tablename2.substr(0, pos2) + "." + tablename2.substr(pos2 + 1);
+  }
+
   return HA_EXIT_SUCCESS;
 }
 
@@ -7686,22 +7705,39 @@ bool ha_rocksdb::contains_foreign_key(THD *const thd) {
   @brief
   splits the normalized table name of <dbname>.<tablename>#P#<part_no> into
   the <dbname>, <tablename> and <part_no> components.
+  For Catalogs this is <catname>.<dbname>.<tablename>.
 
   @param dbbuf returns database name/table_schema
   @param tablebuf returns tablename
   @param partitionbuf returns partition suffix if there is one
   @return HA_EXIT_SUCCESS on success, non-zero on failure to split
 */
 int rdb_split_normalized_tablename(const std::string &fullname,
+                                   std::string *const cat,
                                    std::string *const db,
                                    std::string *const table,
                                    std::string *const partition) {
   DBUG_ASSERT(!fullname.empty());
 
+  ssize_t dotpos;
+  size_t start = 0;
 #define RDB_PARTITION_STR "#P#"
 
+  if (using_catalogs) {
+    dotpos = fullname.find('.');
+
+    if (dotpos == std::string::npos) {
+      return HA_ERR_ROCKSDB_INVALID_TABLE;
+    }
+
+    if (cat != nullptr) {
+      *cat = fullname.substr(0, dotpos);
+    }
+    start = dotpos + 1;
+  }
+
   /* Normalize returns dbname.tablename. */
-  size_t dotpos = fullname.find('.');
+  dotpos = fullname.find('.', start);
 
   /* Invalid table name? */
   if (dotpos == std::string::npos) {
@@ -7712,7 +7748,7 @@ int rdb_split_normalized_tablename(const std::string &fullname,
   DBUG_ASSERT(dotpos > 0);
 
   if (db != nullptr) {
-    *db = fullname.substr(0, dotpos);
+    *db = fullname.substr(start, dotpos - start);
   }
 
   dotpos++;
@@ -11885,6 +11921,8 @@ int ha_rocksdb::rename_table(const char *const from, const char *const to) {
 
   std::string from_str;
   std::string to_str;
+  std::string from_cat;
+  std::string to_cat;
   std::string from_db;
   std::string to_db;
 
@@ -11893,7 +11931,7 @@ int ha_rocksdb::rename_table(const char *const from, const char *const to) {
     DBUG_RETURN(rc);
   }
 
-  rc = rdb_split_normalized_tablename(from_str, &from_db);
+  rc = rdb_split_normalized_tablename(from_str, &from_cat, &from_db);
   if (rc != HA_EXIT_SUCCESS) {
     DBUG_RETURN(rc);
   }
@@ -11903,14 +11941,15 @@ int ha_rocksdb::rename_table(const char *const from, const char *const to) {
     DBUG_RETURN(rc);
   }
 
-  rc = rdb_split_normalized_tablename(to_str, &to_db);
+  rc = rdb_split_normalized_tablename(to_str, &to_cat, &to_db);
   if (rc != HA_EXIT_SUCCESS) {
     DBUG_RETURN(rc);
   }
 
   // If the user changed the database part of the name then validate that the
   // 'to' database exists.
-  if (from_db != to_db && !rdb_database_exists(to_db)) {
+  // Rename will always be within a single DB.
+  if (from_db != to_db && !rdb_database_exists(to_cat, to_db)) {
     // If we return a RocksDB specific error code here we get
     // "error: 206 - Unknown error 206".  InnoDB gets
     // "error -1 - Unknown error -1" so let's match them.

storage/rocksdb/ha_rocksdb_proto.h

@@ -49,7 +49,8 @@ void rdb_handle_io_error(const rocksdb::Status status,
 int rdb_normalize_tablename(const std::string &tablename, std::string *str)
     MY_ATTRIBUTE((__nonnull__, __warn_unused_result__));
 
-int rdb_split_normalized_tablename(const std::string &fullname, std::string *db,
+int rdb_split_normalized_tablename(const std::string &fullname,
+                                   std::string *cat, std::string *db,
                                    std::string *table = nullptr,
                                    std::string *partition = nullptr)
     MY_ATTRIBUTE((__warn_unused_result__));

storage/rocksdb/mysql-test/rocksdb/include/have_rocksdb.inc

@@ -3,7 +3,6 @@ if (`SELECT COUNT(*) = 0 FROM INFORMATION_SCHEMA.ENGINES WHERE engine = 'rocksdb
   --skip Test requires engine RocksDB.
 }
 
---source include/not_catalogs.inc
 --disable_query_log
 # Table statistics can vary depending on when the memtables are flushed, so
 # flush them at the beginning of the test to ensure the test runs consistently.

storage/rocksdb/mysql-test/rocksdb/r/perf_context.result

@@ -5,7 +5,7 @@ SET GLOBAL rocksdb_perf_context_level=3;
 CREATE TABLE t1 (i INT, j INT, PRIMARY KEY (i)) ENGINE = ROCKSDB;
 CREATE TABLE t2 (k INT, PRIMARY KEY (k)) ENGINE = ROCKSDB;
 INSERT INTO t1 VALUES (1,1), (2,2), (3,3), (4,4), (5,5);
-SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT WHERE TABLE_NAME = 't1';
+SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, STAT_TYPE, VALUE FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT WHERE TABLE_NAME = 't1';
 TABLE_SCHEMA	TABLE_NAME	PARTITION_NAME	STAT_TYPE	VALUE
 test	t1	NULL	USER_KEY_COMPARISON_COUNT	#
 test	t1	NULL	BLOCK_CACHE_HIT_COUNT	#
@@ -123,7 +123,7 @@ IO_WRITE_NANOS	#
 IO_READ_NANOS	#
 IO_RANGE_SYNC_NANOS	#
 IO_LOGGER_NANOS	#
-SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
+SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, STAT_TYPE, VALUE FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
 WHERE TABLE_NAME = 't1'
 AND STAT_TYPE in ('INTERNAL_KEY_SKIPPED_COUNT', 'INTERNAL_DELETE_SKIPPED_COUNT');
 TABLE_SCHEMA	TABLE_NAME	PARTITION_NAME	STAT_TYPE	VALUE
@@ -136,7 +136,7 @@ i	j
 3	3
 4	4
 5	5
-SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
+SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, STAT_TYPE, VALUE FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
 WHERE TABLE_NAME = 't1'
 AND STAT_TYPE in ('INTERNAL_KEY_SKIPPED_COUNT', 'INTERNAL_DELETE_SKIPPED_COUNT');
 TABLE_SCHEMA	TABLE_NAME	PARTITION_NAME	STAT_TYPE	VALUE
@@ -149,7 +149,7 @@ i	j
 3	3
 4	4
 5	5
-SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
+SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, STAT_TYPE, VALUE FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
 WHERE TABLE_NAME = 't1'
 AND STAT_TYPE in ('INTERNAL_KEY_SKIPPED_COUNT', 'INTERNAL_DELETE_SKIPPED_COUNT');
 TABLE_SCHEMA	TABLE_NAME	PARTITION_NAME	STAT_TYPE	VALUE

storage/rocksdb/mysql-test/rocksdb/t/duplicate_table.test

@@ -7,10 +7,11 @@ INSERT INTO t values (1), (2), (3);
 --error ER_TABLE_EXISTS_ERROR 
 CREATE TABLE t(id int primary key) engine=rocksdb;
 FLUSH TABLES;
-move_file $MYSQLTEST_VARDIR/mysqld.1/data/test/t.frm $MYSQLTEST_VARDIR/mysqld.1/data/test/t.frm.tmp;
+move_file $MARIADB_DATADIR/test/t.frm $MARIADB_DATADIR/test/t.frm.tmp;
+--replace_regex /Table '.*' does/Table 'test.t' does/ /if '.*' exists/if '.\/test\/t.frm' exists/
 --error ER_METADATA_INCONSISTENCY
 CREATE TABLE t(id int primary key) engine=rocksdb;
-move_file $MYSQLTEST_VARDIR/mysqld.1/data/test/t.frm.tmp $MYSQLTEST_VARDIR/mysqld.1/data/test/t.frm;
+move_file $MARIADB_DATADIR/test/t.frm.tmp $MARIADB_DATADIR/test/t.frm;
 FLUSH TABLES;
 SELECT * FROM t;
 DROP TABLE t;

storage/rocksdb/mysql-test/rocksdb/t/i_s_deadlock.test

@@ -33,22 +33,26 @@ echo Deadlock #1;
 --source include/simple_deadlock.inc
 connection default;
 --replace_column 1 DEADLOCK_ID 2 TIMESTAMP 3 TRANSACTION_ID 5 WAITING_KEY
+--replace_regex /.*test/test/
 select * from information_schema.rocksdb_deadlock;
 
 echo Deadlock #2;
 --source include/simple_deadlock.inc
 connection default;
 --replace_column 1 DEADLOCK_ID 2 TIMESTAMP 3 TRANSACTION_ID 5 WAITING_KEY
+--replace_regex /.*test/test/
 select * from information_schema.rocksdb_deadlock;
 set global rocksdb_max_latest_deadlocks = 10;
 
 echo Deadlock #3;
 --source include/simple_deadlock.inc
 connection default;
 --replace_column 1 DEADLOCK_ID 2 TIMESTAMP 3 TRANSACTION_ID 5 WAITING_KEY
+--replace_regex /.*test/test/
 select * from information_schema.rocksdb_deadlock;
 set global rocksdb_max_latest_deadlocks = 1;
 --replace_column 1 DEADLOCK_ID 2 TIMESTAMP 3 TRANSACTION_ID 5 WAITING_KEY
+--replace_regex /.*test/test/
 select * from information_schema.rocksdb_deadlock;
 
 connection con3;
@@ -139,6 +143,7 @@ rollback;
 
 connection default;
 --replace_column 1 DEADLOCK_ID 2 TIMESTAMP 3 TRANSACTION_ID 5 WAITING_KEY
+--replace_regex /.*test/test/
 select * from information_schema.rocksdb_deadlock;
 
 disconnect con1;

storage/rocksdb/mysql-test/rocksdb/t/perf_context.test

@@ -24,7 +24,7 @@ INSERT INTO t1 VALUES (1,1), (2,2), (3,3), (4,4), (5,5);
 
 # Check per-table perf context
 --replace_column 5 #
-SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT WHERE TABLE_NAME = 't1';
+SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, STAT_TYPE, VALUE FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT WHERE TABLE_NAME = 't1';
 
 # Check global perf context
 --replace_column 2 #
@@ -34,19 +34,19 @@ SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT_GLOBAL;
 # Test iteration skip counters
 ###############################################################################
 
-SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
+SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, STAT_TYPE, VALUE FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
 WHERE TABLE_NAME = 't1'
 AND STAT_TYPE in ('INTERNAL_KEY_SKIPPED_COUNT', 'INTERNAL_DELETE_SKIPPED_COUNT');
 
 SELECT * FROM t1;
 
-SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
+SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, STAT_TYPE, VALUE FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
 WHERE TABLE_NAME = 't1'
 AND STAT_TYPE in ('INTERNAL_KEY_SKIPPED_COUNT', 'INTERNAL_DELETE_SKIPPED_COUNT');
 
 SELECT * FROM t1 WHERE j BETWEEN 1 AND 5;
 
-SELECT * FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
+SELECT TABLE_SCHEMA, TABLE_NAME, PARTITION_NAME, STAT_TYPE, VALUE FROM INFORMATION_SCHEMA.ROCKSDB_PERF_CONTEXT
 WHERE TABLE_NAME = 't1'
 AND STAT_TYPE in ('INTERNAL_KEY_SKIPPED_COUNT', 'INTERNAL_DELETE_SKIPPED_COUNT');
 

storage/rocksdb/mysql-test/rocksdb/t/rocksdb_datadir.test

@@ -1,3 +1,6 @@
+# Bootstrap command doesn't work with Catalogs
+--source include/not_catalogs.inc
+
 --source include/have_rocksdb.inc
 let $ddir = $MYSQL_TMP_DIR/.rocksdb_datadir.test.install.db;
 let $rdb_ddir = $MYSQL_TMP_DIR/.rocksdb_datadir.test;

storage/rocksdb/mysql-test/rocksdb/t/skip_validate_tmp_table.test

@@ -6,7 +6,7 @@
 create table t1 (pk int primary key) engine=rocksdb;
 
 # Create a .frm file without a matching table
---exec cp $MYSQLTEST_VARDIR/mysqld.$_server_id/data/test/t1.frm $MYSQLTEST_VARDIR/mysqld.$_server_id/data/test/t1#sql-test.frm
+--exec cp $MARIADB_DATADIR/test/t1.frm $MARIADB_DATADIR/test/t1#sql-test.frm
 
 --source include/restart_mysqld.inc
 
@@ -23,8 +23,8 @@ set session debug_dbug= "-d,gen_sql_table_name";
 show tables;
 
 # Remove the corresponding .frm files
---remove_files_wildcard $MYSQLTEST_VARDIR/mysqld.$_server_id/data/test *t1*.frm
---remove_files_wildcard $MYSQLTEST_VARDIR/mysqld.$_server_id/data/test *t2*.frm
+--remove_files_wildcard $MARIADB_DATADIR/test *t1*.frm
+--remove_files_wildcard $MARIADB_DATADIR/test *t2*.frm
 
 # Restart the server with a table registered in RocksDB but does not have a .frm file
 --source include/restart_mysqld.inc

storage/rocksdb/rdb_datadic.cc

@@ -3614,8 +3614,14 @@ time_t Rdb_tbl_def::get_create_time() {
     // Read it from the .frm file. It's not a problem if several threads do this
     // concurrently
     char path[FN_REFLEN];
-    snprintf(path, sizeof(path), "%s/%s/%s%s", mysql_data_home,
-             m_dbname.c_str(), m_tablename.c_str(), reg_ext);
+    if (using_catalogs) {
+      snprintf(path, sizeof(path), "%s/%s/%s/%s%s", mysql_data_home,
+               m_catname.c_str(), m_dbname.c_str(), m_tablename.c_str(),
+               reg_ext);
+    } else {
+      snprintf(path, sizeof(path), "%s/%s/%s%s", mysql_data_home,
+               m_dbname.c_str(), m_tablename.c_str(), reg_ext);
+    }
     unpack_filename(path,path);
     MY_STAT f_stat;
     if (my_stat(path, &f_stat, MYF(0)))
@@ -3700,8 +3706,8 @@ void Rdb_tbl_def::set_name(const std::string &name) {
   int err MY_ATTRIBUTE((__unused__));
 
   m_dbname_tablename = name;
-  err = rdb_split_normalized_tablename(name, &m_dbname, &m_tablename,
-                                       &m_partition);
+  err = rdb_split_normalized_tablename(name, &m_catname, &m_dbname,
+                                       &m_tablename, &m_partition);
   DBUG_ASSERT(err == 0);
 
   check_if_is_mysql_system_table();
@@ -3895,8 +3901,8 @@ bool Rdb_validate_tbls::scan_for_frms(const std::string &datadir,
 bool Rdb_validate_tbls::compare_to_actual_tables(const std::string &datadir,
                                                  bool *has_errors) {
   bool result = true;
-  struct st_my_dir *dir_info;
-  struct fileinfo *file_info;
+  struct st_my_dir *dir_info, *l2_dir_info = NULL;
+  struct fileinfo *file_info, *l2_file_info;
 
   dir_info = my_dir(datadir.c_str(), MYF(MY_DONT_SORT | MY_WANT_STAT));
   if (dir_info == nullptr) {
@@ -3913,10 +3919,27 @@ bool Rdb_validate_tbls::compare_to_actual_tables(const std::string &datadir,
     /* Ignore all non-directory files */
     if (!MY_S_ISDIR(file_info->mystat->st_mode)) continue;
 
-    /* Scan all the .frm files in the directory */
-    if (!scan_for_frms(datadir, file_info->name, has_errors)) {
-      result = false;
-      break;
+    if (!using_catalogs) {
+      /* Scan all the .frm files in the directory */
+      if (!scan_for_frms(datadir, file_info->name, has_errors)) {
+        result = false;
+        break;
+      }
+    } else {
+      /* We got a Catalog here, scan that for DBs */
+      std::string catdir = datadir + FN_LIBCHAR + file_info->name + FN_LIBCHAR;
+      l2_dir_info = my_dir(catdir.c_str(), MYF(MY_DONT_SORT | MY_WANT_STAT));
+      l2_file_info = l2_dir_info->dir_entry;
+      for (size_t jj = 0; jj < l2_dir_info->number_of_files; jj++,
+           l2_file_info++) {
+        if (l2_file_info->name[0] == '.') continue;
+        if (!MY_S_ISDIR(l2_file_info->mystat->st_mode)) continue;
+        if (!scan_for_frms(catdir, l2_file_info->name, has_errors)) {
+          result = false;
+          break;
+        }
+      }
+      my_dirend(l2_dir_info);
     }
   }