diff --git a/HISTORY.md b/HISTORY.md
index 15e755088..d041b3fe9 100644
--- a/HISTORY.md
+++ b/HISTORY.md
@@ -27,6 +27,7 @@
 * Completely removed the following deprecated/obsolete statistics: the tickers `BLOCK_CACHE_INDEX_BYTES_EVICT`, `BLOCK_CACHE_FILTER_BYTES_EVICT`, `BLOOM_FILTER_MICROS`, `NO_FILE_CLOSES`, `STALL_L0_SLOWDOWN_MICROS`, `STALL_MEMTABLE_COMPACTION_MICROS`, `STALL_L0_NUM_FILES_MICROS`, `RATE_LIMIT_DELAY_MILLIS`, `NO_ITERATORS`, `NUMBER_FILTERED_DELETES`, `WRITE_TIMEDOUT`, `BLOB_DB_GC_NUM_KEYS_OVERWRITTEN`, `BLOB_DB_GC_NUM_KEYS_EXPIRED`, `BLOB_DB_GC_BYTES_OVERWRITTEN`, `BLOB_DB_GC_BYTES_EXPIRED`, `BLOCK_CACHE_COMPRESSION_DICT_BYTES_EVICT` as well as the histograms `STALL_L0_SLOWDOWN_COUNT`, `STALL_MEMTABLE_COMPACTION_COUNT`, `STALL_L0_NUM_FILES_COUNT`, `HARD_RATE_LIMIT_DELAY_COUNT`, `SOFT_RATE_LIMIT_DELAY_COUNT`, `BLOB_DB_GC_MICROS`, and `NUM_DATA_BLOCKS_READ_PER_LEVEL`. Note that as a result, the C++ enum values of the still supported statistics have changed. Developers are advised to not rely on the actual numeric values.
 * Deprecated IngestExternalFileOptions::write_global_seqno and change default to false. This option only needs to be set to true to generate a DB compatible with RocksDB versions before 5.16.0.
 * Remove deprecated APIs `GetColumnFamilyOptionsFrom{Map|String}(const ColumnFamilyOptions&, ..)`, `GetDBOptionsFrom{Map|String}(const DBOptions&, ..)`, `GetBlockBasedTableOptionsFrom{Map|String}(const BlockBasedTableOptions& table_options, ..)` and ` GetPlainTableOptionsFrom{Map|String}(const PlainTableOptions& table_options,..)`.
+* Added a subcode of `Status::Corruption`, `Status::SubCode::kMergeOperatorFailed`, for users to identify corruption failures originating in the merge operator, as opposed to RocksDB's internally identified data corruptions
 
 ### Build Changes
 * The `make` build now builds a shared library by default instead of a static library. Use `LIB_MODE=static` to override.
diff --git a/db/db_merge_operator_test.cc b/db/db_merge_operator_test.cc
index f3a8b8cb9..19c7bd1e8 100644
--- a/db/db_merge_operator_test.cc
+++ b/db/db_merge_operator_test.cc
@@ -231,7 +231,9 @@ TEST_F(DBMergeOperatorTest, MergeOperatorFailsWithMustMerge) {
       {
         std::string value;
         ASSERT_OK(db_->Get(ReadOptions(), "k0", &value));
-        ASSERT_TRUE(db_->Get(ReadOptions(), "k1", &value).IsCorruption());
+        Status s = db_->Get(ReadOptions(), "k1", &value);
+        ASSERT_TRUE(s.IsCorruption());
+        ASSERT_EQ(Status::SubCode::kMergeOperatorFailed, s.subcode());
         ASSERT_OK(db_->Get(ReadOptions(), "k2", &value));
       }
 
@@ -243,6 +245,8 @@ TEST_F(DBMergeOperatorTest, MergeOperatorFailsWithMustMerge) {
         ASSERT_EQ("k0", iter->key());
         iter->Next();
         ASSERT_TRUE(iter->status().IsCorruption());
+        ASSERT_EQ(Status::SubCode::kMergeOperatorFailed,
+                  iter->status().subcode());
 
         iter->SeekToLast();
         ASSERT_TRUE(iter->Valid());
diff --git a/db/merge_helper.cc b/db/merge_helper.cc
index fcff8f0ac..eceb9bcb8 100644
--- a/db/merge_helper.cc
+++ b/db/merge_helper.cc
@@ -113,7 +113,7 @@ Status MergeHelper::TimedFullMerge(
 
   if (!success) {
     RecordTick(statistics, NUMBER_MERGE_FAILURES);
-    return Status::Corruption("Error: Could not perform merge.");
+    return Status::Corruption(Status::SubCode::kMergeOperatorFailed);
   }
 
   return Status::OK();
diff --git a/db/version_set.cc b/db/version_set.cc
index d0ade94d5..a61fc3dbd 100644
--- a/db/version_set.cc
+++ b/db/version_set.cc
@@ -2407,6 +2407,9 @@ void Version::Get(const ReadOptions& read_options, const LookupKey& k,
             "Encounter unexpected blob index. Please open DB with "
             "ROCKSDB_NAMESPACE::blob_db::BlobDB instead.");
         return;
+      case GetContext::kMergeOperatorFailed:
+        *status = Status::Corruption(Status::SubCode::kMergeOperatorFailed);
+        return;
     }
     f = fp.GetNextFile();
   }
diff --git a/db/version_set_sync_and_async.h b/db/version_set_sync_and_async.h
index 8041c91fd..188c2e2f9 100644
--- a/db/version_set_sync_and_async.h
+++ b/db/version_set_sync_and_async.h
@@ -157,6 +157,10 @@ DEFINE_SYNC_AND_ASYNC(Status, Version::MultiGetFromSST)
             "ROCKSDB_NAMESPACE::blob_db::BlobDB instead.");
         file_range.MarkKeyDone(iter);
         continue;
+      case GetContext::kMergeOperatorFailed:
+        *status = Status::Corruption(Status::SubCode::kMergeOperatorFailed);
+        file_range.MarkKeyDone(iter);
+        continue;
     }
   }
 
diff --git a/include/rocksdb/status.h b/include/rocksdb/status.h
index 39af94559..447c3b9fe 100644
--- a/include/rocksdb/status.h
+++ b/include/rocksdb/status.h
@@ -113,6 +113,7 @@ class Status {
     kOverwritten = 12,
     kTxnNotPrepared = 13,
     kIOFenced = 14,
+    kMergeOperatorFailed = 15,
     kMaxSubCode
   };
 
diff --git a/table/get_context.cc b/table/get_context.cc
index f6acb17a9..7e33e3567 100644
--- a/table/get_context.cc
+++ b/table/get_context.cc
@@ -474,7 +474,11 @@ void GetContext::Merge(const Slice* value) {
       /* update_num_ops_stats */ true,
       /* op_failure_scope */ nullptr);
   if (!s.ok()) {
-    state_ = kCorrupt;
+    if (s.subcode() == Status::SubCode::kMergeOperatorFailed) {
+      state_ = kMergeOperatorFailed;
+    } else {
+      state_ = kCorrupt;
+    }
     return;
   }
 
@@ -514,7 +518,11 @@ void GetContext::MergeWithEntity(Slice entity) {
           /* update_num_ops_stats */ true,
           /* op_failure_scope */ nullptr);
       if (!s.ok()) {
-        state_ = kCorrupt;
+        if (s.subcode() == Status::SubCode::kMergeOperatorFailed) {
+          state_ = kMergeOperatorFailed;
+        } else {
+          state_ = kCorrupt;
+        }
         return;
       }
     }
@@ -533,7 +541,11 @@ void GetContext::MergeWithEntity(Slice entity) {
         &result, logger_, statistics_, clock_, /* update_num_ops_stats */ true,
         /* op_failure_scope */ nullptr);
     if (!s.ok()) {
-      state_ = kCorrupt;
+      if (s.subcode() == Status::SubCode::kMergeOperatorFailed) {
+        state_ = kMergeOperatorFailed;
+      } else {
+        state_ = kCorrupt;
+      }
       return;
     }
   }
diff --git a/table/get_context.h b/table/get_context.h
index 10371529b..528cd14fd 100644
--- a/table/get_context.h
+++ b/table/get_context.h
@@ -75,6 +75,7 @@ class GetContext {
     kCorrupt,
     kMerge,  // saver contains the current merge result (the operands)
     kUnexpectedBlobIndex,
+    kMergeOperatorFailed,
   };
   GetContextStats get_context_stats_;
 
diff --git a/util/status.cc b/util/status.cc
index 1156b10ef..ead315848 100644
--- a/util/status.cc
+++ b/util/status.cc
@@ -41,9 +41,10 @@ static const char* msgs[static_cast<int>(Status::kMaxSubCode)] = {
     "Insufficient capacity for merge operands",
     // kManualCompactionPaused
     "Manual compaction paused",
-    " (overwritten)",    // kOverwritten, subcode of OK
-    "Txn not prepared",  // kTxnNotPrepared
-    "IO fenced off",     // kIOFenced
+    " (overwritten)",         // kOverwritten, subcode of OK
+    "Txn not prepared",       // kTxnNotPrepared
+    "IO fenced off",          // kIOFenced
+    "Merge operator failed",  // kMergeOperatorFailed
 };
 
 Status::Status(Code _code, SubCode _subcode, const Slice& msg,