Add rocksdb_wsenv_path for initial WSEnvironment support

Summary:
Added rocksdb_wsenv_path to support WSEnvironment. At this point you'll need both specify rocksdb_wsenv_path and also builds with --wsenv to enable this feature since it requires linking with WSEnvironment in a separate repo.

Note under WSEnvironment, there are a few quirks:
* SyncWAL isn't supported yet so for now I'm pretending they just succeed (otherwise simple things like log rotation would fail) and only do Flush(false) as needed for rdb_dict_manager
* There is no local .rocksdb directory - so for now I'm disabling the bulk loading temp SST cleanup code
* WSEnvironment for some reason doesn't like '#' in the file names, for now I'm working around it by replacing '#' with '$' in SST file name for partitioned table during bulk loading (mostly working, though some tests are timing out)
* We should not rely on errno and instead should always stick to rocksdb::Status for error checking (addressed in earlier commit)
* We should make sure DBOptions/etc are cleaned up

Differential Revision: D22531061

----------------------------------------------------------------

Delete garbage temp sst files using RocksDB Env API

Summary:
In MyRocksOnWS scenarios, MyRocks writes bulk load tmp file into WS (through SSTFileWriter) but the cleanup code in Rdb_sst_info::init still assumes local directory and will enumerate from local directory to delete garbage temp files that weren't cleaned up during a crash. This would leave files in WS around and end up not cleaning anything.
This involves using RocksDB Env API to create / access the file instead of using MySQL API, and should work regardless whether it is WS or local disk.

Reviewed By: yizhang82

Differential Revision: D28680122

Author: yizhang82

mysql-test/r/mysqld--help-notwin.result

@@ -2122,6 +2122,8 @@ The following options may be given as the first argument:
  WriteOptions::ignore_missing_column_families for RocksDB
  --rocksdb-write-policy=name 
  DBOptions::write_policy for RocksDB
+ --rocksdb-wsenv-path=name 
+ Path for RocksDB WSEnv
  --rpl-read-size=#   The size for reads done from the binlog and relay log. It
  must be a multiple of 4kb. Making it larger might help
  with IO stalls while reading these files when they are
@@ -3145,6 +3147,7 @@ rocksdb-write-batch-max-bytes 0
 rocksdb-write-disable-wal FALSE
 rocksdb-write-ignore-missing-column-families FALSE
 rocksdb-write-policy write_committed
+rocksdb-wsenv-path 
 rpl-read-size 8192
 rpl-receive-buffer-size 2097152
 rpl-send-buffer-size 2097152

mysql-test/suite/rocksdb_sys_vars/r/rocksdb_wsenv_path_basic.result

@@ -0,0 +1,7 @@
+SET @start_global_value = @@global.ROCKSDB_WSENV_PATH;
+SELECT @start_global_value;
+@start_global_value
+
+"Trying to set variable @@global.ROCKSDB_WSENV_PATH to 444. It should fail because it is readonly."
+SET @@global.ROCKSDB_WSENV_PATH   = 444;
+ERROR HY000: Variable 'rocksdb_wsenv_path' is a read only variable

mysql-test/suite/rocksdb_sys_vars/t/rocksdb_wsenv_path_basic.test

@@ -0,0 +1,6 @@
+--source include/have_rocksdb.inc
+
+--let $sys_var=ROCKSDB_WSENV_PATH
+--let $read_only=1
+--let $session=0
+--source ../include/rocksdb_sys_var.inc

storage/rocksdb/CMakeLists.txt

@@ -141,6 +141,13 @@ IF(CMAKE_COMPILER_IS_GNUCXX AND NOT CMAKE_CXX_COMPILER_VERSION VERSION_LESS 12)
   ENDIF()
 ENDIF()
 
+IF(FB_WITH_WSENV)
+  ADD_DEFINITIONS(-DFB_HAVE_WSENV=1)
+  INCLUDE_DIRECTORIES(
+    ${FB_WITH_WSENV}
+  )
+ENDIF()
+
 INCLUDE_DIRECTORIES(
   ${ROCKSDB_ROOT}
   ${ROCKSDB_ROOT}/include

storage/rocksdb/ha_rocksdb.cc

@@ -43,6 +43,7 @@
 
 /* MySQL includes */
 #include <mysql/components/services/log_builtins.h>
+#include <mysql/psi/mysql_file.h>
 #include <mysql/psi/mysql_table.h>
 #include <mysql/thread_pool_priv.h>
 #include <mysys_err.h>
@@ -724,6 +725,7 @@ static unsigned long  // NOLINT(runtime/int)
 static ulong rocksdb_info_log_level;
 static char *rocksdb_wal_dir;
 static char *rocksdb_persistent_cache_path;
+static char *rocksdb_wsenv_path;
 static ulong rocksdb_index_type;
 static uint32_t rocksdb_flush_log_at_trx_commit;
 static uint32_t rocksdb_debug_optimizer_n_rows;
@@ -1596,6 +1598,10 @@ static MYSQL_SYSVAR_ULONG(
     nullptr, nullptr, rocksdb_persistent_cache_size_mb,
     /* min */ 0L, /* max */ ULONG_MAX, 0);
 
+static MYSQL_SYSVAR_STR(wsenv_path, rocksdb_wsenv_path,
+                        PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
+                        "Path for RocksDB WSEnv", nullptr, nullptr, "");
+
 static MYSQL_SYSVAR_STR(fault_injection_options,
                         opt_rocksdb_fault_injection_options,
                         PLUGIN_VAR_RQCMDARG | PLUGIN_VAR_READONLY,
@@ -2496,6 +2502,7 @@ static struct SYS_VAR *rocksdb_system_variables[] = {
     MYSQL_SYSVAR(persistent_cache_path),
     MYSQL_SYSVAR(persistent_cache_size_mb),
     MYSQL_SYSVAR(fault_injection_options),
+    MYSQL_SYSVAR(wsenv_path),
     MYSQL_SYSVAR(delete_obsolete_files_period_micros),
     MYSQL_SYSVAR(max_background_jobs),
     MYSQL_SYSVAR(max_background_flushes),
@@ -3998,7 +4005,8 @@ class Rdb_transaction_impl : public Rdb_transaction {
     tx_opts.write_batch_flush_threshold =
         THDVAR(m_thd, write_batch_flush_threshold);
 
-    write_opts.sync = (rocksdb_flush_log_at_trx_commit == FLUSH_LOG_SYNC);
+    write_opts.sync = (rocksdb_flush_log_at_trx_commit == FLUSH_LOG_SYNC) &&
+                      rdb_sync_wal_supported();
     write_opts.disableWAL = THDVAR(m_thd, write_disable_wal);
     write_opts.ignore_missing_column_families =
         THDVAR(m_thd, write_ignore_missing_column_families);
@@ -4291,7 +4299,8 @@ class Rdb_writebatch_impl : public Rdb_transaction {
 
   void start_tx() override {
     reset();
-    write_opts.sync = (rocksdb_flush_log_at_trx_commit == FLUSH_LOG_SYNC);
+    write_opts.sync = (rocksdb_flush_log_at_trx_commit == FLUSH_LOG_SYNC) &&
+                      rdb_sync_wal_supported();
     write_opts.disableWAL = THDVAR(m_thd, write_disable_wal);
     write_opts.ignore_missing_column_families =
         THDVAR(m_thd, write_ignore_missing_column_families);
@@ -4566,14 +4575,17 @@ static bool rocksdb_flush_wal(handlerton *const hton MY_ATTRIBUTE((__unused__)),
   if ((!binlog_group_flush && !rocksdb_db_options->allow_mmap_writes) ||
       rocksdb_flush_log_at_trx_commit != FLUSH_LOG_NEVER) {
     rocksdb_wal_group_syncs++;
-    s = rdb->FlushWAL(!binlog_group_flush ||
-                      rocksdb_flush_log_at_trx_commit == FLUSH_LOG_SYNC);
+    bool sync = rdb_sync_wal_supported() &&
+                (!binlog_group_flush ||
+                 rocksdb_flush_log_at_trx_commit == FLUSH_LOG_SYNC);
+    s = rdb->FlushWAL(sync);
   }
 
   if (!s.ok()) {
     rdb_log_status_error(s);
     return HA_EXIT_FAILURE;
   }
+
   return HA_EXIT_SUCCESS;
 }
 
@@ -5929,6 +5941,23 @@ static rocksdb::Status check_rocksdb_options_compatibility(
 
 static uint rocksdb_partition_flags() { return (HA_CANNOT_PARTITION_FK); }
 
+bool rdb_has_wsenv() {
+#if FB_HAVE_WSENV
+  return rocksdb_wsenv_path != nullptr && *rocksdb_wsenv_path;
+#else
+  return false;
+#endif
+}
+
+bool rdb_sync_wal_supported() {
+#if FB_HAVE_WSENV
+  // wsenv doesn't support SyncWAL=true yet
+  return !rdb_has_wsenv();
+#else
+  return true;
+#endif
+}
+
 /*
   Storage Engine initialization function, invoked when plugin is loaded.
 */
@@ -5965,6 +5994,33 @@ static int rocksdb_init_internal(void *const p) {
     DBUG_RETURN(HA_EXIT_FAILURE);
   });
 
+#ifdef FB_HAVE_WSENV
+  // Initialize WSEnv with rocksdb_ws_env_path
+  if (rdb_has_wsenv()) {
+    // NO_LINT_DEBUG
+    sql_print_information(
+        "RocksDB: Initializing WSEnvironment: rocksdb_wsenv_path = %s",
+        rocksdb_wsenv_path);
+
+    RegisterCustomObjectsSimple();
+    rocksdb::Env *ws_env = nullptr;
+    auto s = rocksdb::Env::LoadEnv(rocksdb_wsenv_path, &ws_env);
+    if (s.ok()) {
+      rocksdb_db_options->env = ws_env;
+    } else {
+      rdb_log_status_error(s, "Can't initialize WSEnvironment");
+      DBUG_RETURN(HA_EXIT_FAILURE);
+    }
+  }
+#else
+  if (rocksdb_wsenv_path != nullptr && *rocksdb_wsenv_path) {
+    // We've turned on WSEnv in the wrong build
+    // NO_LINT_DEBUG
+    sql_print_error("RocksDB: WSEnvironment not supported. ");
+    DBUG_RETURN(HA_EXIT_FAILURE);
+  }
+#endif
+
   if (opt_rocksdb_fault_injection_options != nullptr &&
       *opt_rocksdb_fault_injection_options != '\0') {
     bool retryable = false;
@@ -14773,7 +14829,8 @@ void Rdb_background_thread::run() {
     // background thread.
     if (rdb && (rocksdb_flush_log_at_trx_commit != FLUSH_LOG_SYNC) &&
         !rocksdb_db_options->allow_mmap_writes) {
-      const rocksdb::Status s = rdb->FlushWAL(true);
+      bool sync = rdb_sync_wal_supported();
+      const rocksdb::Status s = rdb->FlushWAL(sync);
       if (!s.ok()) {
         rdb_handle_io_error(s, RDB_IO_ERROR_BG_THREAD);
       }

storage/rocksdb/ha_rocksdb_proto.h

@@ -88,6 +88,12 @@ int rdb_dbug_set_ttl_read_filter_ts();
 bool rdb_dbug_set_ttl_ignore_pk();
 #endif
 
+/* Whether WSEnvironment is enabled */
+bool rdb_has_wsenv();
+
+/* Whether SyncWAL is supported in current scenario */
+bool rdb_sync_wal_supported();
+
 enum operation_type : int;
 void rdb_update_global_stats(const operation_type &type, uint count,
                              bool is_system_table = false);

storage/rocksdb/rdb_datadic.cc

@@ -5266,14 +5266,19 @@ int Rdb_dict_manager::commit(rocksdb::WriteBatch *const batch,
   if (!batch) return HA_ERR_ROCKSDB_COMMIT_FAILED;
   int res = HA_EXIT_SUCCESS;
   rocksdb::WriteOptions options;
-  options.sync = sync;
+  options.sync = (sync && rdb_sync_wal_supported());
   rocksdb::TransactionDBWriteOptimizations optimize;
   optimize.skip_concurrency_control = true;
   rocksdb::Status s = m_db->Write(options, optimize, batch);
   res = !s.ok();  // we return true when something failed
   if (res) {
     rdb_handle_io_error(s, RDB_IO_ERROR_DICT_COMMIT);
   }
+  if (!rdb_sync_wal_supported()) {
+    // If we don't support SyncWAL, do a flush at least
+    m_db->FlushWAL(false);
+  }
+
   batch->Clear();
   return res;
 }

storage/rocksdb/rdb_sst_info.cc

@@ -31,6 +31,8 @@
 
 /* RocksDB header files */
 #include "rocksdb/db.h"
+#include "rocksdb/file_system.h"
+#include "rocksdb/io_status.h"
 #include "rocksdb/options.h"
 
 /* MyRocks header files */
@@ -333,6 +335,11 @@ Rdb_sst_info::Rdb_sst_info(rocksdb::DB *const db, const std::string &tablename,
   // is loaded in parallel
   m_prefix += std::to_string(m_prefix_counter.fetch_add(1)) + "_";
 
+  if (rdb_has_wsenv()) {
+    // WSEnv doesn't like '#'
+    std::replace(m_prefix.begin(), m_prefix.end(), '#', '$');
+  }
+
   rocksdb::ColumnFamilyDescriptor cf_descr;
   const rocksdb::Status s = m_cf->GetDescriptor(&cf_descr);
   if (!s.ok()) {
@@ -519,32 +526,30 @@ void Rdb_sst_info::report_error_msg(const rocksdb::Status &s,
 }
 
 void Rdb_sst_info::init(const rocksdb::DB *const db) {
-  const std::string path = db->GetName() + FN_DIRSEP;
-  struct MY_DIR *const dir_info = my_dir(path.c_str(), MYF(MY_DONT_SORT));
+  const std::string dir = db->GetName();
+  const auto &fs = db->GetEnv()->GetFileSystem();
+  std::vector<std::string> files_in_dir;
 
-  // Access the directory
-  if (dir_info == nullptr) {
+  // Get the files in the specified directory
+  rocksdb::IOStatus s =
+      fs->GetChildren(dir, rocksdb::IOOptions(), &files_in_dir, nullptr);
+  if (!s.ok()) {
     // NO_LINT_DEBUG
     sql_print_warning("RocksDB: Could not access database directory: %s",
-                      path.c_str());
+                      dir.c_str());
     return;
   }
 
   // Scan through the files in the directory
-  const struct fileinfo *file_info = dir_info->dir_entry;
-  for (uint ii = 0; ii < dir_info->number_off_files; ii++, file_info++) {
-    // find any files ending with m_suffix ...
-    const std::string name = file_info->name;
-    const size_t pos = name.find(m_suffix);
-    if (pos != std::string::npos && name.size() - pos == m_suffix.size()) {
-      // ... and remove them
-      const std::string fullname = path + name;
-      my_delete(fullname.c_str(), MYF(0));
+  for (const auto &file : files_in_dir) {
+    // Find any files ending with m_suffix ...
+    const size_t pos = file.find(m_suffix);
+    if (pos != std::string::npos && file.size() - pos == m_suffix.size()) {
+      // Remove
+      const std::string fullname = dir + FN_DIRSEP + file;
+      fs->DeleteFile(fullname, rocksdb::IOOptions(), nullptr);
     }
   }
-
-  // Release the directory entry
-  my_dirend(dir_info);
 }
 
 std::atomic<uint64_t> Rdb_sst_info::m_prefix_counter(0);