diff --git a/table/block_based/block.cc b/table/block_based/block.cc index f36883fd68..1c5be79984 100644 --- a/table/block_based/block.cc +++ b/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; diff --git a/table/block_based/block_test.cc b/table/block_based/block_test.cc index 1165a43513..9f28f5501b 100644 --- a/table/block_based/block_test.cc +++ b/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(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 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) {