Merge operator failed subcode (#11231)

Summary:
From HISTORY.md: 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.

This is a followup to https://github.com/facebook/rocksdb/issues/11092, where we gave users the ability to keep running a DB despite merge operator failing. Now that the DB keeps running despite such failures, they want to be able to distinguish such failures from real corruptions.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/11231

Test Plan: updated unit test

Reviewed By: akankshamahajan15

Differential Revision: D43396607

Pulled By: ajkr

fbshipit-source-id: 17fbcc779ad724dafada8abd73efd38e1c5208b9

diff --git a/HISTORY.md b/HISTORY.md
index 15e755088a97f170998f15db6fc9403f8d94116a..d041b3fe9947e4a176486b4fd3c0af0ef4a21735 100644 (file)
--- 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 f3a8b8cb923ea1d57905cc25fb0db2609c46db3e..19c7bd1e80f47a3bae21870bad888849576999f4 100644 (file)
--- 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 fcff8f0acf5309c706033c7a446ccf95a0a147ae..eceb9bcb8421a3a840a390b49d4d5809e38d24e2 100644 (file)
--- 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 d0ade94d5827ad50ac61796e92d9a9bd66255c53..a61fc3dbd23a5026288bb846fc29510ce3b5115c 100644 (file)
--- 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 8041c91fd64d091579906ef3f639bcf24d2f58f3..188c2e2f95034fdfd4383486dae46ace2737954b 100644 (file)
--- 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 39af9455991c6e9cb3bbdf2afe95326ccacbcbdc..447c3b9fefd18dd46937b2b76f93e9444d4002fd 100644 (file)
--- 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 f6acb17a917b88b3380dc578b377d2550b58b088..7e33e3567d125dfba333747c797e4cb5226a36e8 100644 (file)
--- 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 10371529bc398aad0813ada94f9a75b0bfa8dbe7..528cd14fd89ce81f4eedfa45692a2f3dca7be7cc 100644 (file)
--- 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 1156b10ef4988840c75225ded3d0c485c92779b4..ead315848d3aa64cace5d0eecaceeeb710c2d1eb 100644 (file)
--- 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,