diff --git a/db/column_family.cc b/db/column_family.cc
index ae8256f15..68c1875c7 100644
--- a/db/column_family.cc
+++ b/db/column_family.cc
@@ -446,7 +446,11 @@ void ColumnFamilyData::RecalculateWriteStallConditions(
     } else if (vstorage->l0_delay_trigger_count() >=
                mutable_cf_options.level0_stop_writes_trigger) {
       write_controller_token_ = write_controller->GetStopToken();
-      internal_stats_->AddCFStats(InternalStats::LEVEL0_NUM_FILES, 1);
+      internal_stats_->AddCFStats(InternalStats::LEVEL0_NUM_FILES_TOTAL, 1);
+      if (compaction_picker_->IsLevel0CompactionInProgress()) {
+        internal_stats_->AddCFStats(
+            InternalStats::LEVEL0_NUM_FILES_WITH_COMPACTION, 1);
+      }
       Log(InfoLogLevel::WARN_LEVEL, ioptions_.info_log,
           "[%s] Stopping writes because we have %d level-0 files",
           name_.c_str(), vstorage->l0_delay_trigger_count());
@@ -454,7 +458,11 @@ void ColumnFamilyData::RecalculateWriteStallConditions(
                vstorage->l0_delay_trigger_count() >=
                    mutable_cf_options.level0_slowdown_writes_trigger) {
       write_controller_token_ = write_controller->GetDelayToken();
-      internal_stats_->AddCFStats(InternalStats::LEVEL0_SLOWDOWN, 1);
+      internal_stats_->AddCFStats(InternalStats::LEVEL0_SLOWDOWN_TOTAL, 1);
+      if (compaction_picker_->IsLevel0CompactionInProgress()) {
+        internal_stats_->AddCFStats(
+            InternalStats::LEVEL0_SLOWDOWN_WITH_COMPACTION, 1);
+      }
       Log(InfoLogLevel::WARN_LEVEL, ioptions_.info_log,
           "[%s] Stalling writes because we have %d level-0 files",
           name_.c_str(), vstorage->l0_delay_trigger_count());
diff --git a/db/compaction_job.cc b/db/compaction_job.cc
index 7eb2361e8..2fcdd2862 100644
--- a/db/compaction_job.cc
+++ b/db/compaction_job.cc
@@ -316,6 +316,8 @@ void CompactionJob::Prepare() {
       Slice* end = i == boundaries_.size() ? nullptr : &boundaries_[i];
       compact_->sub_compact_states.emplace_back(c, start, end, sizes_[i]);
     }
+    MeasureTime(stats_, NUM_SUBCOMPACTIONS_SCHEDULED,
+                compact_->sub_compact_states.size());
   } else {
     compact_->sub_compact_states.emplace_back(c, nullptr, nullptr);
   }
diff --git a/db/compaction_picker.h b/db/compaction_picker.h
index 1d1abe30c..e7d8bf6db 100644
--- a/db/compaction_picker.h
+++ b/db/compaction_picker.h
@@ -110,6 +110,11 @@ class CompactionPicker {
   // overlapping.
   bool IsInputNonOverlapping(Compaction* c);
 
+  // Is there currently a compaction involving level 0 taking place
+  bool IsLevel0CompactionInProgress() const {
+    return !level0_compactions_in_progress_.empty();
+  }
+
  protected:
   int NumberLevels() const { return ioptions_.num_levels; }
 
diff --git a/db/internal_stats.cc b/db/internal_stats.cc
index c9d0b9bf4..370dc4ed3 100644
--- a/db/internal_stats.cc
+++ b/db/internal_stats.cc
@@ -669,12 +669,11 @@ void InternalStats::DumpCFStats(std::string* value) {
     total_files += files;
     total_files_being_compacted += files_being_compacted[level];
     if (comp_stats_[level].micros > 0 || files > 0) {
-      uint64_t stalls = level == 0 ?
-        (cf_stats_count_[LEVEL0_SLOWDOWN] +
-         cf_stats_count_[LEVEL0_NUM_FILES] +
-         cf_stats_count_[MEMTABLE_COMPACTION])
-        : (stall_leveln_slowdown_count_soft_[level] +
-           stall_leveln_slowdown_count_hard_[level]);
+      uint64_t stalls = level == 0 ? (cf_stats_count_[LEVEL0_SLOWDOWN_TOTAL] +
+                                      cf_stats_count_[LEVEL0_NUM_FILES_TOTAL] +
+                                      cf_stats_count_[MEMTABLE_COMPACTION])
+                                   : (stall_leveln_slowdown_count_soft_[level] +
+                                      stall_leveln_slowdown_count_hard_[level]);
 
       stats_sum.Add(comp_stats_[level]);
       total_file_size += vstorage->NumLevelBytes(level);
@@ -718,11 +717,16 @@ void InternalStats::DumpCFStats(std::string* value) {
 
   snprintf(buf, sizeof(buf),
            "Stalls(count): %" PRIu64 " level0_slowdown, "
-           "%" PRIu64 " level0_numfiles, %" PRIu64 " memtable_compaction, "
+           "%" PRIu64 " level0_slowdown_with_compaction, "
+           "%" PRIu64 " level0_numfiles, "
+           "%" PRIu64 " level0_numfiles_with_compaction, "
+           "%" PRIu64 " memtable_compaction, "
            "%" PRIu64 " leveln_slowdown_soft, "
            "%" PRIu64 " leveln_slowdown_hard\n",
-           cf_stats_count_[LEVEL0_SLOWDOWN],
-           cf_stats_count_[LEVEL0_NUM_FILES],
+           cf_stats_count_[LEVEL0_SLOWDOWN_TOTAL],
+           cf_stats_count_[LEVEL0_SLOWDOWN_WITH_COMPACTION],
+           cf_stats_count_[LEVEL0_NUM_FILES_TOTAL],
+           cf_stats_count_[LEVEL0_NUM_FILES_WITH_COMPACTION],
            cf_stats_count_[MEMTABLE_COMPACTION],
            total_slowdown_count_soft, total_slowdown_count_hard);
   value->append(buf);
diff --git a/db/internal_stats.h b/db/internal_stats.h
index 1d5512c94..139a4544b 100644
--- a/db/internal_stats.h
+++ b/db/internal_stats.h
@@ -80,9 +80,11 @@ extern DBPropertyType GetPropertyType(const Slice& property,
 class InternalStats {
  public:
   enum InternalCFStatsType {
-    LEVEL0_SLOWDOWN,
+    LEVEL0_SLOWDOWN_TOTAL,
+    LEVEL0_SLOWDOWN_WITH_COMPACTION,
     MEMTABLE_COMPACTION,
-    LEVEL0_NUM_FILES,
+    LEVEL0_NUM_FILES_TOTAL,
+    LEVEL0_NUM_FILES_WITH_COMPACTION,
     WRITE_STALLS_ENUM_MAX,
     BYTES_FLUSHED,
     INTERNAL_CF_STATS_ENUM_MAX,
@@ -350,9 +352,11 @@ class InternalStats {
 class InternalStats {
  public:
   enum InternalCFStatsType {
-    LEVEL0_SLOWDOWN,
+    LEVEL0_SLOWDOWN_TOTAL,
+    LEVEL0_SLOWDOWN_WITH_COMPACTION,
     MEMTABLE_COMPACTION,
-    LEVEL0_NUM_FILES,
+    LEVEL0_NUM_FILES_TOTAL,
+    LEVEL0_NUM_FILES_WITH_COMPACTION,
     WRITE_STALLS_ENUM_MAX,
     BYTES_FLUSHED,
     INTERNAL_CF_STATS_ENUM_MAX,
diff --git a/include/rocksdb/statistics.h b/include/rocksdb/statistics.h
index 5d610c15a..10982fe3d 100644
--- a/include/rocksdb/statistics.h
+++ b/include/rocksdb/statistics.h
@@ -271,6 +271,8 @@ enum Histograms : uint32_t {
   DB_SEEK,
   WRITE_STALL,
   SST_READ_MICROS,
+  // The number of subcompactions actually scheduled during a compaction
+  NUM_SUBCOMPACTIONS_SCHEDULED,
   HISTOGRAM_ENUM_MAX,  // TODO(ldemailly): enforce HistogramsNameMap match
 };
 
@@ -297,6 +299,7 @@ const std::vector<std::pair<Histograms, std::string>> HistogramsNameMap = {
     {DB_SEEK, "rocksdb.db.seek.micros"},
     {WRITE_STALL, "rocksdb.db.write.stall"},
     {SST_READ_MICROS, "rocksdb.sst.read.micros"},
+    {NUM_SUBCOMPACTIONS_SCHEDULED, "rocksdb.num.subcompactions.scheduled"},
 };
 
 struct HistogramData {