Fix GetCorruptionStatus() crash with separated KV blocks (#15006)

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

Block::GetCorruptionStatus() uses restart_offset_ in error state to decide whether it can re-decode the data block footer for a more specific corruption status: nonzero means the footer decode failed and restart_offset_ stores the original block size; zero means the corruption was detected after the footer decoded.

For separated-KV data blocks, Block::Block() first sets restart_offset_ to the restart-array offset, then validates footer.values_section_offset. If that offset is past the restart array, the block is marked corrupt with size = 0, but restart_offset_ was left nonzero. NewDataIterator() then called GetCorruptionStatus(), which treated the stale restart-array offset as an original block size, re-decoded a syntactically valid footer, and hit DEBUG_FAIL("ok status on presumed bad block contents").

Reset restart_offset_ to 0 on this semantic-corruption path so GetCorruptionStatus() reports generic block corruption instead of trying to re-decode a footer that already decoded successfully.

The regression test builds a separated-KV hash-index data block, corrupts only the encoded values_section_offset to point past the block, and verifies NewDataIterator() returns "Corruption: bad block contents". Removing the fix line makes the test reproduce the original assertion.

This update also applies the clang-format layout required by check-format for the new regression test.

Reviewed By: joshkang97

Differential Revision: D113561303

fbshipit-source-id: 8d989a38761063c62694837976aad98f1868f884

table/block_based/block.cc

@@ -1356,6 +1356,9 @@ Block::Block(BlockContents&& contents, size_t read_amp_bytes_per_bit,
     // Set up values_section_ from footer if separated KV storage is used
     if (size != 0 && footer.separated_kv) {
       if (footer.values_section_offset > restart_offset_) {
+        // The footer decoded fine; the corruption is a semantic one that
+        // re-decoding cannot rediscover.
+        restart_offset_ = 0;
         size = 0;  // Error marker
       } else {
         values_section_ = data() + footer.values_section_offset;

table/block_based/block_test.cc

@@ -2119,6 +2119,51 @@ TEST_P(MetaBlockEntryCorruptionTest, CorruptedValueLengthPastValueEnd) {
   ASSERT_TRUE(iter->status().IsCorruption());
 }
 
+TEST_F(BlockTest, SeparatedKVInvalidValuesSectionOffset) {
+  BlockBuilder builder(16 /* restart_interval */, true /* use_delta_encoding */,
+                       false /* use_value_delta_encoding */,
+                       BlockBasedTableOptions::kDataBlockBinaryAndHash,
+                       0.75 /* hash_ratio */, 0 /* ts_sz */,
+                       true /* persist_user_defined_timestamps */,
+                       false /* is_user_key */, true /* separate_key_value */);
+
+  for (int i = 0; i < 5; i++) {
+    std::string key = "key" + std::to_string(i);
+    AppendInternalKeyFooter(&key, 100 - i, kTypeValue);
+    builder.Add(key, "value" + std::to_string(i));
+  }
+
+  Slice raw_block = builder.Finish();
+  std::string block_data = raw_block.ToString();
+
+  Slice footer_input(block_data);
+  DataBlockFooter footer;
+  ASSERT_OK(footer.DecodeFrom(&footer_input));
+  ASSERT_TRUE(footer.separated_kv);
+  footer.values_section_offset = static_cast<uint32_t>(block_data.size());
+
+  std::string encoded_footer;
+  footer.EncodeTo(&encoded_footer);
+  ASSERT_GE(block_data.size(), encoded_footer.size());
+  block_data.replace(block_data.size() - encoded_footer.size(),
+                     encoded_footer.size(), encoded_footer);
+
+  BlockContents contents;
+  contents.data = Slice(block_data);
+
+  Block block(std::move(contents), 0 /* read_amp_bytes_per_bit */,
+              nullptr /* statistics */, 0 /* restart_interval */);
+
+  ASSERT_EQ(block.size(), 0);
+
+  std::unique_ptr<DataBlockIter> iter(
+      block.NewDataIterator(BytewiseComparator(), kDisableGlobalSequenceNumber,
+                            nullptr, nullptr, false, true));
+  ASSERT_FALSE(iter->Valid());
+  ASSERT_TRUE(iter->status().IsCorruption());
+  ASSERT_EQ(iter->status().ToString(), "Corruption: bad block contents");
+}
+
 }  // namespace ROCKSDB_NAMESPACE
 
 int main(int argc, char** argv) {