Fix backward compatibility with 2.5 through 2.7 (#9189)

Summary:
A bug in https://github.com/facebook/rocksdb/issues/9163 can cause checksum verification to fail if
parsing a properties block fails.

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

Test Plan:
check_format_compatible.sh (never quite works locally but
this particular case seems fixed using variants of SHORT_TEST=1).
And added new unit test case.

Reviewed By: ajkr

Differential Revision: D32574626

Pulled By: pdillinger

fbshipit-source-id: 6fa5c8595737b71a3c3d011a52daf6d6c08715d7

db/db_table_properties_test.cc

@@ -171,6 +171,31 @@ TEST_F(DBTablePropertiesTest, GetPropertiesOfAllTablesTest) {
   SyncPoint::GetInstance()->DisableProcessing();
 }
 
+TEST_F(DBTablePropertiesTest, InvalidIgnored) {
+  // RocksDB versions 2.5 - 2.7 generate some properties that Block considers
+  // invalid in some way. This approximates that.
+
+  // Inject properties block data that Block considers invalid
+  SyncPoint::GetInstance()->SetCallBack(
+      "BlockBasedTableBuilder::WritePropertiesBlock:BlockData",
+      [&](void* block_data) {
+        *reinterpret_cast<Slice*>(block_data) = Slice("X");
+      });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  // Build file
+  for (int i = 0; i < 10; ++i) {
+    ASSERT_OK(db_->Put(WriteOptions(), ToString(i), "val"));
+  }
+  ASSERT_OK(db_->Flush(FlushOptions()));
+
+  SyncPoint::GetInstance()->DisableProcessing();
+
+  // Not crashing is good enough
+  TablePropertiesCollection props;
+  ASSERT_OK(db_->GetPropertiesOfAllTables(&props));
+}
+
 TEST_F(DBTablePropertiesTest, CreateOnDeletionCollectorFactory) {
   ConfigOptions options;
   options.ignore_unsupported_options = false;

table/block_based/block_based_table_builder.cc

@@ -1715,8 +1715,11 @@ void BlockBasedTableBuilder::WritePropertiesBlock(
                                          rep_->ioptions.logger,
                                          &property_block_builder);
 
-    WriteRawBlock(property_block_builder.Finish(), kNoCompression,
-                  &properties_block_handle, BlockType::kProperties);
+    Slice block_data = property_block_builder.Finish();
+    TEST_SYNC_POINT_CALLBACK(
+        "BlockBasedTableBuilder::WritePropertiesBlock:BlockData", &block_data);
+    WriteRawBlock(block_data, kNoCompression, &properties_block_handle,
+                  BlockType::kProperties);
   }
   if (ok()) {
 #ifndef NDEBUG

table/meta_blocks.cc

@@ -251,6 +251,9 @@ Status ReadTablePropertiesHelper(
     return s;
   }
 
+  // Unfortunately, Block::size() might not equal block_contents.data.size(),
+  // and Block hides block_contents
+  uint64_t block_size = block_contents.data.size();
   Block properties_block(std::move(block_contents));
   DataBlockIter iter;
   properties_block.NewDataIterator(BytewiseComparator(),
@@ -377,8 +380,7 @@ Status ReadTablePropertiesHelper(
   // (See write_global_seqno comment above)
   if (s.ok() && footer.GetBlockTrailerSize() > 0) {
     s = VerifyBlockChecksum(footer.checksum(), properties_block.data(),
-                            properties_block.size(), file->file_name(),
-                            handle.offset());
+                            block_size, file->file_name(), handle.offset());
     if (s.IsCorruption()) {
       const auto seqno_pos_iter = new_table_properties->properties_offsets.find(
           ExternalSstFilePropertyNames::kGlobalSeqno);
@@ -387,9 +389,8 @@ Status ReadTablePropertiesHelper(
                             block_fetcher.GetBlockSizeWithTrailer());
         uint64_t global_seqno_offset = seqno_pos_iter->second - handle.offset();
         EncodeFixed64(&tmp_buf[static_cast<size_t>(global_seqno_offset)], 0);
-        s = VerifyBlockChecksum(footer.checksum(), tmp_buf.data(),
-                                properties_block.size(), file->file_name(),
-                                handle.offset());
+        s = VerifyBlockChecksum(footer.checksum(), tmp_buf.data(), block_size,
+                                file->file_name(), handle.offset());
       }
     }
   }