Fix the wrong calculation of largest_key in import_column_family_job (#11381)

Summary:
When calculating the largest_key in ImportColumnFamilyJob::GetIngestedFileInfo, only the first element of range_del_iter is calculated. If range_del_iter has multiple elements, the largest_key will be wrong

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

Reviewed By: cbi42

Differential Revision: D44981450

Pulled By: ajkr

fbshipit-source-id: 584bc7da86295568a96984d2951644f289e578c7

db/import_column_family_job.cc

@@ -302,15 +302,24 @@ Status ImportColumnFamilyJob::GetIngestedFileInfo(
           return Status::Corruption("Corrupted key in external file. ",
                                     pik_status.getState());
         }
-        RangeTombstone tombstone(key, range_del_iter->value());
-        InternalKey start_key = tombstone.SerializeKey();
+        RangeTombstone first_tombstone(key, range_del_iter->value());
+        InternalKey start_key = first_tombstone.SerializeKey();
         const InternalKeyComparator* icmp = &cfd_->internal_comparator();
         if (!bound_set ||
             icmp->Compare(start_key, file_to_import->smallest_internal_key) <
                 0) {
           file_to_import->smallest_internal_key = start_key;
         }
-        InternalKey end_key = tombstone.SerializeEndKey();
+
+        range_del_iter->SeekToLast();
+        pik_status = ParseInternalKey(range_del_iter->key(), &key,
+                                      db_options_.allow_data_in_errors);
+        if (!pik_status.ok()) {
+          return Status::Corruption("Corrupted key in external file. ",
+                                    pik_status.getState());
+        }
+        RangeTombstone last_tombstone(key, range_del_iter->value());
+        InternalKey end_key = last_tombstone.SerializeEndKey();
         if (!bound_set ||
             icmp->Compare(end_key, file_to_import->largest_internal_key) > 0) {
           file_to_import->largest_internal_key = end_key;

db/import_column_family_test.cc

@@ -294,6 +294,8 @@ TEST_F(ImportColumnFamilyTest, ImportSSTFileWriterFilesWithRangeTombstone) {
   ASSERT_OK(sfw_cf1.Put("K1", "V1"));
   ASSERT_OK(sfw_cf1.Put("K2", "V2"));
   ASSERT_OK(sfw_cf1.DeleteRange("K3", "K4"));
+  ASSERT_OK(sfw_cf1.DeleteRange("K7", "K9"));
+
   ASSERT_OK(sfw_cf1.Finish());
 
   // Import sst file corresponding to cf1 onto a new cf and verify
@@ -319,7 +321,7 @@ TEST_F(ImportColumnFamilyTest, ImportSSTFileWriterFilesWithRangeTombstone) {
   ASSERT_TRUE(file_meta != nullptr);
   InternalKey largest;
   largest.DecodeFrom(file_meta->largest);
-  ASSERT_EQ(largest.user_key(), "K4");
+  ASSERT_EQ(largest.user_key(), "K9");
 
   std::string value;
   ASSERT_OK(db_->Get(ReadOptions(), import_cfh_, "K1", &value));