diff --git a/db/db_io_failure_test.cc b/db/db_io_failure_test.cc
index 4d66e1d1a3..3b13e7727c 100644
--- a/db/db_io_failure_test.cc
+++ b/db/db_io_failure_test.cc
@@ -111,6 +111,7 @@ TEST_F(DBIOFailureTest, NoSpaceCompactRange) {
     Status s = dbfull()->TEST_CompactRange(0, nullptr, nullptr, nullptr,
                                            true /* disallow trivial move */);
     ASSERT_TRUE(s.IsIOError());
+    ASSERT_TRUE(s.IsNoSpace());
 
     env_->no_space_.store(false, std::memory_order_release);
   } while (ChangeCompactOptions());
diff --git a/db/db_test_util.h b/db/db_test_util.h
index 645619602a..68dcde52f7 100644
--- a/db/db_test_util.h
+++ b/db/db_test_util.h
@@ -221,7 +221,7 @@ class SpecialEnv : public EnvWrapper {
           // Drop writes on the floor
           return Status::OK();
         } else if (env_->no_space_.load(std::memory_order_acquire)) {
-          return Status::IOError("No space left on device");
+          return Status::NoSpace("No space left on device");
         } else {
           env_->bytes_written_ += data.size();
           return base_->Append(data);
diff --git a/include/rocksdb/status.h b/include/rocksdb/status.h
index 03fd8895fe..139942362e 100644
--- a/include/rocksdb/status.h
+++ b/include/rocksdb/status.h
@@ -33,14 +33,14 @@ class Status {
   Status& operator=(const Status& s);
   Status(Status&& s)
 #if !(defined _MSC_VER) || ((defined _MSC_VER) && (_MSC_VER >= 1900))
-    noexcept
+      noexcept
 #endif
-  ;
+      ;
   Status& operator=(Status&& s)
 #if !(defined _MSC_VER) || ((defined _MSC_VER) && (_MSC_VER >= 1900))
-    noexcept
+      noexcept
 #endif
-  ;
+      ;
   bool operator==(const Status& rhs) const;
   bool operator!=(const Status& rhs) const;
 
@@ -58,7 +58,7 @@ class Status {
     kAborted = 10,
     kBusy = 11,
     kExpired = 12,
-    kTryAgain = 13,
+    kTryAgain = 13
   };
 
   Code code() const { return code_; }
@@ -68,6 +68,7 @@ class Status {
     kMutexTimeout = 1,
     kLockTimeout = 2,
     kLockLimit = 3,
+    kNoSpace = 4,
     kMaxSubCode
   };
 
@@ -157,6 +158,11 @@ class Status {
     return Status(kTryAgain, msg, msg2);
   }
 
+  static Status NoSpace() { return Status(kIOError, kNoSpace); }
+  static Status NoSpace(const Slice& msg, const Slice& msg2 = Slice()) {
+    return Status(kIOError, kNoSpace, msg, msg2);
+  }
+
   // Returns true iff the status indicates success.
   bool ok() const { return code() == kOk; }
 
@@ -200,6 +206,15 @@ class Status {
   // re-attempted.
   bool IsTryAgain() const { return code() == kTryAgain; }
 
+  // Returns true iff the status indicates a NoSpace error
+  // This is caused by an I/O error returning the specific "out of space"
+  // error condition. Stricto sensu, an NoSpace error is an I/O error
+  // with a specific subcode, enabling users to take the appropriate action
+  // if needed
+  bool IsNoSpace() const {
+    return (code() == kIOError) && (subcode() == kNoSpace);
+  }
+
   // Return a string representation of this status suitable for printing.
   // Returns the string "OK" for success.
   std::string ToString() const;
@@ -219,7 +234,10 @@ class Status {
   explicit Status(Code _code, SubCode _subcode = kNone)
       : code_(_code), subcode_(_subcode), state_(nullptr) {}
 
-  Status(Code _code, const Slice& msg, const Slice& msg2);
+  Status(Code _code, SubCode _subcode, const Slice& msg, const Slice& msg2);
+  Status(Code _code, const Slice& msg, const Slice& msg2)
+      : Status(_code, kNone, msg, msg2) {}
+
   static const char* CopyState(const char* s);
 };
 
@@ -229,7 +247,7 @@ inline Status::Status(const Status& s) : code_(s.code_), subcode_(s.subcode_) {
 inline Status& Status::operator=(const Status& s) {
   // The following condition catches both aliasing (when this == &s),
   // and the common case where both s and *this are ok.
-  if(this != &s) {
+  if (this != &s) {
     code_ = s.code_;
     subcode_ = s.subcode_;
     delete[] state_;
@@ -240,23 +258,23 @@ inline Status& Status::operator=(const Status& s) {
 
 inline Status::Status(Status&& s)
 #if !(defined _MSC_VER) || ((defined _MSC_VER) && (_MSC_VER >= 1900))
-noexcept
+    noexcept
 #endif
-  : Status() {
+    : Status() {
   *this = std::move(s);
 }
 
 inline Status& Status::operator=(Status&& s)
 #if !(defined _MSC_VER) || ((defined _MSC_VER) && (_MSC_VER >= 1900))
-noexcept
+    noexcept
 #endif
 {
-  if(this != &s) {
+  if (this != &s) {
     code_ = std::move(s.code_);
     s.code_ = kOk;
     subcode_ = std::move(s.subcode_);
     s.subcode_ = kNone;
-    delete [] state_;
+    delete[] state_;
     state_ = nullptr;
     std::swap(state_, s.state_);
   }
diff --git a/port/win/io_win.h b/port/win/io_win.h
index 8c3a4ba7a6..311cc35ff1 100644
--- a/port/win/io_win.h
+++ b/port/win/io_win.h
@@ -26,7 +26,9 @@ namespace port {
 std::string GetWindowsErrSz(DWORD err);
 
 inline Status IOErrorFromWindowsError(const std::string& context, DWORD err) {
-  return Status::IOError(context, GetWindowsErrSz(err));
+  return (err == ERROR_HANDLE_DISK_FULL) ?
+      Status::NoSpace(context, GetWindowsErrSz(err)) :
+      Status::IOError(context, GetWindowsErrSz(err));
 }
 
 inline Status IOErrorFromLastWindowsError(const std::string& context) {
@@ -34,7 +36,9 @@ inline Status IOErrorFromLastWindowsError(const std::string& context) {
 }
 
 inline Status IOError(const std::string& context, int err_number) {
-  return Status::IOError(context, strerror(err_number));
+  return (err_number == ENOSPC) ?
+      Status::NoSpace(context, strerror(err_number)) :      
+      Status::IOError(context, strerror(err_number));
 }
 
 // Note the below two do not set errno because they are used only here in this
diff --git a/util/env_hdfs.cc b/util/env_hdfs.cc
index f1d3f01480..cb85df9c90 100644
--- a/util/env_hdfs.cc
+++ b/util/env_hdfs.cc
@@ -36,7 +36,9 @@ namespace {
 
 // Log error message
 static Status IOError(const std::string& context, int err_number) {
-  return Status::IOError(context, strerror(err_number));
+  return (err_number == ENOSPC) ?
+      Status::NoSpace(context, strerror(err_number)) :
+      Status::IOError(context, strerror(err_number));
 }
 
 // assume that there is one global logger for now. It is not thread-safe,
diff --git a/util/io_posix.h b/util/io_posix.h
index 7db8e23d9e..ab9a8f796a 100644
--- a/util/io_posix.h
+++ b/util/io_posix.h
@@ -9,6 +9,7 @@
 #pragma once
 #include <unistd.h>
 #include <atomic>
+#include <errno.h>
 #include "rocksdb/env.h"
 
 // For non linux platform, the following macros are used only as place
@@ -24,7 +25,9 @@
 namespace rocksdb {
 
 static Status IOError(const std::string& context, int err_number) {
-  return Status::IOError(context, strerror(err_number));
+  return (err_number == ENOSPC) ?
+      Status::NoSpace(context, strerror(err_number)) :
+      Status::IOError(context, strerror(err_number));
 }
 
 class PosixHelper {
diff --git a/util/status.cc b/util/status.cc
index 93590d2d7f..c17006ea72 100644
--- a/util/status.cc
+++ b/util/status.cc
@@ -21,9 +21,10 @@ const char* Status::CopyState(const char* state) {
   return result;
 }
 
-Status::Status(Code _code, const Slice& msg, const Slice& msg2)
-    : code_(_code), subcode_(kNone) {
+Status::Status(Code _code, SubCode _subcode, const Slice& msg, const Slice& msg2)
+    : code_(_code), subcode_(_subcode) {
   assert(code_ != kOk);
+  assert(subcode_ != kMaxSubCode);
   const uint32_t len1 = static_cast<uint32_t>(msg.size());
   const uint32_t len2 = static_cast<uint32_t>(msg2.size());
   const uint32_t size = len1 + (len2 ? (2 + len2) : 0);
diff --git a/util/status_message.cc b/util/status_message.cc
index fc251a9b40..94ae651282 100644
--- a/util/status_message.cc
+++ b/util/status_message.cc
@@ -8,10 +8,11 @@
 namespace rocksdb {
 
 const char* Status::msgs[] = {
-    "",                                                  // kNone
-    "Timeout Acquiring Mutex",                           // kMutexTimeout
-    "Timeout waiting to lock key",                       // kLockTimeout
-    "Failed to acquire lock due to max_num_locks limit"  // kLockLimit
+    "",                                                   // kNone
+    "Timeout Acquiring Mutex",                            // kMutexTimeout
+    "Timeout waiting to lock key",                        // kLockTimeout
+    "Failed to acquire lock due to max_num_locks limit",  // kLockLimit
+    "No space left on device"                             // kNoSpace
 };
 
 }  // namespace rocksdb