diff --git a/mysql-test/suite/rocksdb/r/rocksdb.result b/mysql-test/suite/rocksdb/r/rocksdb.result
index 41d087a2538e..78cf42ebcd3c 100644
--- a/mysql-test/suite/rocksdb/r/rocksdb.result
+++ b/mysql-test/suite/rocksdb/r/rocksdb.result
@@ -900,6 +900,7 @@ rocksdb_error_if_exists	OFF
 rocksdb_flush_log_at_trx_commit	1
 rocksdb_flush_memtable_on_analyze	ON
 rocksdb_force_compute_memtable_stats	ON
+rocksdb_force_flush_memtable_and_lzero_now	OFF
 rocksdb_force_flush_memtable_now	OFF
 rocksdb_force_index_records_in_range	0
 rocksdb_hash_index_allow_collision	ON
diff --git a/mysql-test/suite/rocksdb_sys_vars/r/rocksdb_force_flush_memtable_and_lzero_now_basic.result b/mysql-test/suite/rocksdb_sys_vars/r/rocksdb_force_flush_memtable_and_lzero_now_basic.result
new file mode 100644
index 000000000000..68cfeb07fc71
--- /dev/null
+++ b/mysql-test/suite/rocksdb_sys_vars/r/rocksdb_force_flush_memtable_and_lzero_now_basic.result
@@ -0,0 +1,50 @@
+CREATE TABLE valid_values (value varchar(255)) ENGINE=myisam;
+INSERT INTO valid_values VALUES(1);
+INSERT INTO valid_values VALUES(0);
+INSERT INTO valid_values VALUES('on');
+CREATE TABLE invalid_values (value varchar(255)) ENGINE=myisam;
+SET @start_global_value = @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW;
+SELECT @start_global_value;
+@start_global_value
+0
+'# Setting to valid values in global scope#'
+"Trying to set variable @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW to 1"
+SET @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW   = 1;
+SELECT @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW;
+@@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW
+0
+"Setting the global scope variable back to default"
+SET @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW = DEFAULT;
+SELECT @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW;
+@@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW
+0
+"Trying to set variable @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW to 0"
+SET @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW   = 0;
+SELECT @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW;
+@@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW
+0
+"Setting the global scope variable back to default"
+SET @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW = DEFAULT;
+SELECT @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW;
+@@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW
+0
+"Trying to set variable @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW to on"
+SET @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW   = on;
+SELECT @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW;
+@@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW
+0
+"Setting the global scope variable back to default"
+SET @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW = DEFAULT;
+SELECT @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW;
+@@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW
+0
+"Trying to set variable @@session.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW to 444. It should fail because it is not session."
+SET @@session.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW   = 444;
+ERROR HY000: Variable 'rocksdb_force_flush_memtable_and_lzero_now' is a GLOBAL variable and should be set with SET GLOBAL
+'# Testing with invalid values in global scope #'
+SET @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW = @start_global_value;
+SELECT @@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW;
+@@global.ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW
+0
+DROP TABLE valid_values;
+DROP TABLE invalid_values;
diff --git a/mysql-test/suite/rocksdb_sys_vars/t/rocksdb_force_flush_memtable_and_lzero_now_basic.test b/mysql-test/suite/rocksdb_sys_vars/t/rocksdb_force_flush_memtable_and_lzero_now_basic.test
new file mode 100644
index 000000000000..7b55ac7126ac
--- /dev/null
+++ b/mysql-test/suite/rocksdb_sys_vars/t/rocksdb_force_flush_memtable_and_lzero_now_basic.test
@@ -0,0 +1,17 @@
+--source include/have_rocksdb.inc
+
+CREATE TABLE valid_values (value varchar(255)) ENGINE=myisam;
+INSERT INTO valid_values VALUES(1);
+INSERT INTO valid_values VALUES(0);
+INSERT INTO valid_values VALUES('on');
+
+CREATE TABLE invalid_values (value varchar(255)) ENGINE=myisam;
+
+--let $sys_var=ROCKSDB_FORCE_FLUSH_MEMTABLE_AND_LZERO_NOW
+--let $read_only=0
+--let $session=0
+--let $sticky=1
+--source suite/sys_vars/inc/rocksdb_sys_var.inc
+
+DROP TABLE valid_values;
+DROP TABLE invalid_values;
diff --git a/storage/rocksdb/ha_rocksdb.cc b/storage/rocksdb/ha_rocksdb.cc
index f82ec2c00b51..528ff3336aa7 100644
--- a/storage/rocksdb/ha_rocksdb.cc
+++ b/storage/rocksdb/ha_rocksdb.cc
@@ -283,11 +283,47 @@ static void rocksdb_force_flush_memtable_now_stub(
 static int rocksdb_force_flush_memtable_now(
     THD *const thd, struct st_mysql_sys_var *const var, void *const var_ptr,
     struct st_mysql_value *const value) {
-  sql_print_information("RocksDB: Manual memtable flush\n");
+  sql_print_information("RocksDB: Manual memtable flush.");
   rocksdb_flush_all_memtables();
   return HA_EXIT_SUCCESS;
 }
 
+static void rocksdb_force_flush_memtable_and_lzero_now_stub(
+    THD *const thd, struct st_mysql_sys_var *const var, void *const var_ptr,
+    const void *const save) {}
+
+static int rocksdb_force_flush_memtable_and_lzero_now(
+    THD *const thd, struct st_mysql_sys_var *const var, void *const var_ptr,
+    struct st_mysql_value *const value) {
+  sql_print_information("RocksDB: Manual memtable and L0 flush.");
+  rocksdb_flush_all_memtables();
+
+  const Rdb_cf_manager &cf_manager = rdb_get_cf_manager();
+  const rocksdb::CompactionOptions c_options = rocksdb::CompactionOptions();
+  for (const auto &cf_handle : cf_manager.get_all_cf()) {
+    rocksdb::ColumnFamilyMetaData metadata;
+    rdb->GetColumnFamilyMetaData(cf_handle, &metadata);
+
+    DBUG_ASSERT(metadata.levels[0].level == 0);
+    std::vector<std::string> file_names;
+    for (auto &file : metadata.levels[0].files) {
+      file_names.emplace_back(file.db_path + file.name);
+    }
+
+    if (!file_names.empty()) {
+      rocksdb::Status s;
+      s = rdb->CompactFiles(c_options, cf_handle, file_names, 1);
+
+      if (!s.ok() && !s.IsAborted()) {
+        rdb_handle_io_error(s, RDB_IO_ERROR_GENERAL);
+        return HA_EXIT_FAILURE;
+      }
+    }
+  }
+
+  return HA_EXIT_SUCCESS;
+}
+
 static void rocksdb_drop_index_wakeup_thread(
     my_core::THD *const thd MY_ATTRIBUTE((__unused__)),
     struct st_mysql_sys_var *const var MY_ATTRIBUTE((__unused__)),
@@ -385,6 +421,7 @@ static my_bool rocksdb_enable_2pc = 0;
 static char *rocksdb_strict_collation_exceptions;
 static my_bool rocksdb_collect_sst_properties = 1;
 static my_bool rocksdb_force_flush_memtable_now_var = 0;
+static my_bool rocksdb_force_flush_memtable_and_lzero_now_var = 0;
 static uint64_t rocksdb_number_stat_computes = 0;
 static uint32_t rocksdb_seconds_between_stat_computes = 3600;
 static long long rocksdb_compaction_sequential_deletes = 0l;
@@ -1086,6 +1123,13 @@ static MYSQL_SYSVAR_BOOL(
     rocksdb_force_flush_memtable_now, rocksdb_force_flush_memtable_now_stub,
     FALSE);
 
+static MYSQL_SYSVAR_BOOL(
+    force_flush_memtable_and_lzero_now,
+    rocksdb_force_flush_memtable_and_lzero_now_var, PLUGIN_VAR_RQCMDARG,
+    "Acts similar to force_flush_memtable_now, but also compacts all L0 files.",
+    rocksdb_force_flush_memtable_and_lzero_now,
+    rocksdb_force_flush_memtable_and_lzero_now_stub, FALSE);
+
 static MYSQL_THDVAR_BOOL(
     flush_memtable_on_analyze, PLUGIN_VAR_RQCMDARG,
     "Forces memtable flush on ANALZYE table to get accurate cardinality",
@@ -1299,6 +1343,7 @@ static struct st_mysql_sys_var *rocksdb_system_variables[] = {
     MYSQL_SYSVAR(strict_collation_exceptions),
     MYSQL_SYSVAR(collect_sst_properties),
     MYSQL_SYSVAR(force_flush_memtable_now),
+    MYSQL_SYSVAR(force_flush_memtable_and_lzero_now),
     MYSQL_SYSVAR(flush_memtable_on_analyze),
     MYSQL_SYSVAR(seconds_between_stat_computes),