ceph/src/rocksdb/db/import_column_family_test.cc

   1 #ifndef ROCKSDB_LITE
   2
   3 #include <functional>
   4 #include "db/db_test_util.h"
   5 #include "port/port.h"
   6 #include "port/stack_trace.h"
   7 #include "rocksdb/sst_file_writer.h"
   8 #include "test_util/testutil.h"
   9
  10 namespace ROCKSDB_NAMESPACE {
  11
  12 class ImportColumnFamilyTest : public DBTestBase {
  13  public:
  14   ImportColumnFamilyTest() : DBTestBase("/import_column_family_test") {
  15     sst_files_dir_ = dbname_ + "/sst_files/";
  16     DestroyAndRecreateExternalSSTFilesDir();
  17     export_files_dir_ = test::TmpDir(env_) + "/export";
  18     import_cfh_ = nullptr;
  19     import_cfh2_ = nullptr;
  20     metadata_ptr_ = nullptr;
  21   }
  22
  23   ~ImportColumnFamilyTest() {
  24     if (import_cfh_) {
  25       db_->DropColumnFamily(import_cfh_);
  26       db_->DestroyColumnFamilyHandle(import_cfh_);
  27       import_cfh_ = nullptr;
  28     }
  29     if (import_cfh2_) {
  30       db_->DropColumnFamily(import_cfh2_);
  31       db_->DestroyColumnFamilyHandle(import_cfh2_);
  32       import_cfh2_ = nullptr;
  33     }
  34     if (metadata_ptr_) {
  35       delete metadata_ptr_;
  36       metadata_ptr_ = nullptr;
  37     }
  38     test::DestroyDir(env_, sst_files_dir_);
  39     test::DestroyDir(env_, export_files_dir_);
  40   }
  41
  42   void DestroyAndRecreateExternalSSTFilesDir() {
  43     test::DestroyDir(env_, sst_files_dir_);
  44     env_->CreateDir(sst_files_dir_);
  45     test::DestroyDir(env_, export_files_dir_);
  46   }
  47
  48   LiveFileMetaData LiveFileMetaDataInit(std::string name, std::string path,
  49                                         int level,
  50                                         SequenceNumber smallest_seqno,
  51                                         SequenceNumber largest_seqno) {
  52     LiveFileMetaData metadata;
  53     metadata.name = name;
  54     metadata.db_path = path;
  55     metadata.smallest_seqno = smallest_seqno;
  56     metadata.largest_seqno = largest_seqno;
  57     metadata.level = level;
  58     return metadata;
  59   }
  60
  61  protected:
  62   std::string sst_files_dir_;
  63   std::string export_files_dir_;
  64   ColumnFamilyHandle* import_cfh_;
  65   ColumnFamilyHandle* import_cfh2_;
  66   ExportImportFilesMetaData* metadata_ptr_;
  67 };
  68
  69 TEST_F(ImportColumnFamilyTest, ImportSSTFileWriterFiles) {
  70   Options options = CurrentOptions();
  71   CreateAndReopenWithCF({"koko"}, options);
  72
  73   SstFileWriter sfw_cf1(EnvOptions(), options, handles_[1]);
  74   SstFileWriter sfw_unknown(EnvOptions(), options);
  75
  76   // cf1.sst
  77   const std::string cf1_sst_name = "cf1.sst";
  78   const std::string cf1_sst = sst_files_dir_ + cf1_sst_name;
  79   ASSERT_OK(sfw_cf1.Open(cf1_sst));
  80   ASSERT_OK(sfw_cf1.Put("K1", "V1"));
  81   ASSERT_OK(sfw_cf1.Put("K2", "V2"));
  82   ASSERT_OK(sfw_cf1.Finish());
  83
  84   // cf_unknown.sst
  85   const std::string unknown_sst_name = "cf_unknown.sst";
  86   const std::string unknown_sst = sst_files_dir_ + unknown_sst_name;
  87   ASSERT_OK(sfw_unknown.Open(unknown_sst));
  88   ASSERT_OK(sfw_unknown.Put("K3", "V1"));
  89   ASSERT_OK(sfw_unknown.Put("K4", "V2"));
  90   ASSERT_OK(sfw_unknown.Finish());
  91
  92   {
  93     // Import sst file corresponding to cf1 onto a new cf and verify
  94     ExportImportFilesMetaData metadata;
  95     metadata.files.push_back(
  96         LiveFileMetaDataInit(cf1_sst_name, sst_files_dir_, 0, 10, 19));
  97     metadata.db_comparator_name = options.comparator->Name();
  98
  99     ASSERT_OK(db_->CreateColumnFamilyWithImport(
 100         options, "toto", ImportColumnFamilyOptions(), metadata, &import_cfh_));
 101     ASSERT_NE(import_cfh_, nullptr);
 102
 103     std::string value;
 104     db_->Get(ReadOptions(), import_cfh_, "K1", &value);
 105     ASSERT_EQ(value, "V1");
 106     db_->Get(ReadOptions(), import_cfh_, "K2", &value);
 107     ASSERT_EQ(value, "V2");
 108     ASSERT_OK(db_->DropColumnFamily(import_cfh_));
 109     ASSERT_OK(db_->DestroyColumnFamilyHandle(import_cfh_));
 110     import_cfh_ = nullptr;
 111   }
 112
 113   {
 114     // Import sst file corresponding to unknown cf onto a new cf and verify
 115     ExportImportFilesMetaData metadata;
 116     metadata.files.push_back(
 117         LiveFileMetaDataInit(unknown_sst_name, sst_files_dir_, 0, 20, 29));
 118     metadata.db_comparator_name = options.comparator->Name();
 119
 120     ASSERT_OK(db_->CreateColumnFamilyWithImport(
 121         options, "yoyo", ImportColumnFamilyOptions(), metadata, &import_cfh_));
 122     ASSERT_NE(import_cfh_, nullptr);
 123
 124     std::string value;
 125     db_->Get(ReadOptions(), import_cfh_, "K3", &value);
 126     ASSERT_EQ(value, "V1");
 127     db_->Get(ReadOptions(), import_cfh_, "K4", &value);
 128     ASSERT_EQ(value, "V2");
 129   }
 130 }
 131
 132 TEST_F(ImportColumnFamilyTest, ImportSSTFileWriterFilesWithOverlap) {
 133   Options options = CurrentOptions();
 134   CreateAndReopenWithCF({"koko"}, options);
 135
 136   SstFileWriter sfw_cf1(EnvOptions(), options, handles_[1]);
 137
 138   // file3.sst
 139   const std::string file3_sst_name = "file3.sst";
 140   const std::string file3_sst = sst_files_dir_ + file3_sst_name;
 141   ASSERT_OK(sfw_cf1.Open(file3_sst));
 142   for (int i = 0; i < 100; ++i) {
 143     sfw_cf1.Put(Key(i), Key(i) + "_val");
 144   }
 145   ASSERT_OK(sfw_cf1.Finish());
 146
 147   // file2.sst
 148   const std::string file2_sst_name = "file2.sst";
 149   const std::string file2_sst = sst_files_dir_ + file2_sst_name;
 150   ASSERT_OK(sfw_cf1.Open(file2_sst));
 151   for (int i = 0; i < 100; i += 2) {
 152     sfw_cf1.Put(Key(i), Key(i) + "_overwrite1");
 153   }
 154   ASSERT_OK(sfw_cf1.Finish());
 155
 156   // file1a.sst
 157   const std::string file1a_sst_name = "file1a.sst";
 158   const std::string file1a_sst = sst_files_dir_ + file1a_sst_name;
 159   ASSERT_OK(sfw_cf1.Open(file1a_sst));
 160   for (int i = 0; i < 52; i += 4) {
 161     sfw_cf1.Put(Key(i), Key(i) + "_overwrite2");
 162   }
 163   ASSERT_OK(sfw_cf1.Finish());
 164
 165   // file1b.sst
 166   const std::string file1b_sst_name = "file1b.sst";
 167   const std::string file1b_sst = sst_files_dir_ + file1b_sst_name;
 168   ASSERT_OK(sfw_cf1.Open(file1b_sst));
 169   for (int i = 52; i < 100; i += 4) {
 170     sfw_cf1.Put(Key(i), Key(i) + "_overwrite2");
 171   }
 172   ASSERT_OK(sfw_cf1.Finish());
 173
 174   // file0a.sst
 175   const std::string file0a_sst_name = "file0a.sst";
 176   const std::string file0a_sst = sst_files_dir_ + file0a_sst_name;
 177   ASSERT_OK(sfw_cf1.Open(file0a_sst));
 178   for (int i = 0; i < 100; i += 16) {
 179     sfw_cf1.Put(Key(i), Key(i) + "_overwrite3");
 180   }
 181   ASSERT_OK(sfw_cf1.Finish());
 182
 183   // file0b.sst
 184   const std::string file0b_sst_name = "file0b.sst";
 185   const std::string file0b_sst = sst_files_dir_ + file0b_sst_name;
 186   ASSERT_OK(sfw_cf1.Open(file0b_sst));
 187   for (int i = 0; i < 100; i += 16) {
 188     sfw_cf1.Put(Key(i), Key(i) + "_overwrite4");
 189   }
 190   ASSERT_OK(sfw_cf1.Finish());
 191
 192   // Import sst files and verify
 193   ExportImportFilesMetaData metadata;
 194   metadata.files.push_back(
 195       LiveFileMetaDataInit(file3_sst_name, sst_files_dir_, 3, 10, 19));
 196   metadata.files.push_back(
 197       LiveFileMetaDataInit(file2_sst_name, sst_files_dir_, 2, 20, 29));
 198   metadata.files.push_back(
 199       LiveFileMetaDataInit(file1a_sst_name, sst_files_dir_, 1, 30, 34));
 200   metadata.files.push_back(
 201       LiveFileMetaDataInit(file1b_sst_name, sst_files_dir_, 1, 35, 39));
 202   metadata.files.push_back(
 203       LiveFileMetaDataInit(file0a_sst_name, sst_files_dir_, 0, 40, 49));
 204   metadata.files.push_back(
 205       LiveFileMetaDataInit(file0b_sst_name, sst_files_dir_, 0, 50, 59));
 206   metadata.db_comparator_name = options.comparator->Name();
 207
 208   ASSERT_OK(db_->CreateColumnFamilyWithImport(
 209       options, "toto", ImportColumnFamilyOptions(), metadata, &import_cfh_));
 210   ASSERT_NE(import_cfh_, nullptr);
 211
 212   for (int i = 0; i < 100; i++) {
 213     std::string value;
 214     db_->Get(ReadOptions(), import_cfh_, Key(i), &value);
 215     if (i % 16 == 0) {
 216       ASSERT_EQ(value, Key(i) + "_overwrite4");
 217     } else if (i % 4 == 0) {
 218       ASSERT_EQ(value, Key(i) + "_overwrite2");
 219     } else if (i % 2 == 0) {
 220       ASSERT_EQ(value, Key(i) + "_overwrite1");
 221     } else {
 222       ASSERT_EQ(value, Key(i) + "_val");
 223     }
 224   }
 225
 226   for (int i = 0; i < 100; i += 5) {
 227     ASSERT_OK(
 228         db_->Put(WriteOptions(), import_cfh_, Key(i), Key(i) + "_overwrite5"));
 229   }
 230
 231   // Flush and check again
 232   ASSERT_OK(db_->Flush(FlushOptions(), import_cfh_));
 233   for (int i = 0; i < 100; i++) {
 234     std::string value;
 235     db_->Get(ReadOptions(), import_cfh_, Key(i), &value);
 236     if (i % 5 == 0) {
 237       ASSERT_EQ(value, Key(i) + "_overwrite5");
 238     } else if (i % 16 == 0) {
 239       ASSERT_EQ(value, Key(i) + "_overwrite4");
 240     } else if (i % 4 == 0) {
 241       ASSERT_EQ(value, Key(i) + "_overwrite2");
 242     } else if (i % 2 == 0) {
 243       ASSERT_EQ(value, Key(i) + "_overwrite1");
 244     } else {
 245       ASSERT_EQ(value, Key(i) + "_val");
 246     }
 247   }
 248
 249   // Compact and check again.
 250   ASSERT_OK(
 251       db_->CompactRange(CompactRangeOptions(), import_cfh_, nullptr, nullptr));
 252   for (int i = 0; i < 100; i++) {
 253     std::string value;
 254     db_->Get(ReadOptions(), import_cfh_, Key(i), &value);
 255     if (i % 5 == 0) {
 256       ASSERT_EQ(value, Key(i) + "_overwrite5");
 257     } else if (i % 16 == 0) {
 258       ASSERT_EQ(value, Key(i) + "_overwrite4");
 259     } else if (i % 4 == 0) {
 260       ASSERT_EQ(value, Key(i) + "_overwrite2");
 261     } else if (i % 2 == 0) {
 262       ASSERT_EQ(value, Key(i) + "_overwrite1");
 263     } else {
 264       ASSERT_EQ(value, Key(i) + "_val");
 265     }
 266   }
 267 }
 268
 269 TEST_F(ImportColumnFamilyTest, ImportExportedSSTFromAnotherCF) {
 270   Options options = CurrentOptions();
 271   CreateAndReopenWithCF({"koko"}, options);
 272
 273   for (int i = 0; i < 100; ++i) {
 274     Put(1, Key(i), Key(i) + "_val");
 275   }
 276   ASSERT_OK(Flush(1));
 277
 278   ASSERT_OK(
 279       db_->CompactRange(CompactRangeOptions(), handles_[1], nullptr, nullptr));
 280
 281   // Overwrite the value in the same set of keys.
 282   for (int i = 0; i < 100; ++i) {
 283     Put(1, Key(i), Key(i) + "_overwrite");
 284   }
 285
 286   // Flush to create L0 file.
 287   ASSERT_OK(Flush(1));
 288   for (int i = 0; i < 100; ++i) {
 289     Put(1, Key(i), Key(i) + "_overwrite2");
 290   }
 291
 292   // Flush again to create another L0 file. It should have higher sequencer.
 293   ASSERT_OK(Flush(1));
 294
 295   Checkpoint* checkpoint;
 296   ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
 297   ASSERT_OK(checkpoint->ExportColumnFamily(handles_[1], export_files_dir_,
 298                                            &metadata_ptr_));
 299   ASSERT_NE(metadata_ptr_, nullptr);
 300   delete checkpoint;
 301
 302   ImportColumnFamilyOptions import_options;
 303   import_options.move_files = false;
 304   ASSERT_OK(db_->CreateColumnFamilyWithImport(options, "toto", import_options,
 305                                               *metadata_ptr_, &import_cfh_));
 306   ASSERT_NE(import_cfh_, nullptr);
 307
 308   import_options.move_files = true;
 309   ASSERT_OK(db_->CreateColumnFamilyWithImport(options, "yoyo", import_options,
 310                                               *metadata_ptr_, &import_cfh2_));
 311   ASSERT_NE(import_cfh2_, nullptr);
 312   delete metadata_ptr_;
 313   metadata_ptr_ = NULL;
 314
 315   std::string value1, value2;
 316
 317   for (int i = 0; i < 100; ++i) {
 318     db_->Get(ReadOptions(), import_cfh_, Key(i), &value1);
 319     ASSERT_EQ(Get(1, Key(i)), value1);
 320   }
 321
 322   for (int i = 0; i < 100; ++i) {
 323     db_->Get(ReadOptions(), import_cfh2_, Key(i), &value2);
 324     ASSERT_EQ(Get(1, Key(i)), value2);
 325   }
 326
 327   // Modify keys in cf1 and verify.
 328   for (int i = 0; i < 25; i++) {
 329     ASSERT_OK(db_->Delete(WriteOptions(), import_cfh_, Key(i)));
 330   }
 331   for (int i = 25; i < 50; i++) {
 332     ASSERT_OK(
 333         db_->Put(WriteOptions(), import_cfh_, Key(i), Key(i) + "_overwrite3"));
 334   }
 335   for (int i = 0; i < 25; ++i) {
 336     ASSERT_TRUE(
 337         db_->Get(ReadOptions(), import_cfh_, Key(i), &value1).IsNotFound());
 338   }
 339   for (int i = 25; i < 50; ++i) {
 340     db_->Get(ReadOptions(), import_cfh_, Key(i), &value1);
 341     ASSERT_EQ(Key(i) + "_overwrite3", value1);
 342   }
 343   for (int i = 50; i < 100; ++i) {
 344     db_->Get(ReadOptions(), import_cfh_, Key(i), &value1);
 345     ASSERT_EQ(Key(i) + "_overwrite2", value1);
 346   }
 347
 348   for (int i = 0; i < 100; ++i) {
 349     db_->Get(ReadOptions(), import_cfh2_, Key(i), &value2);
 350     ASSERT_EQ(Get(1, Key(i)), value2);
 351   }
 352
 353   // Compact and check again.
 354   ASSERT_OK(db_->Flush(FlushOptions(), import_cfh_));
 355   ASSERT_OK(
 356       db_->CompactRange(CompactRangeOptions(), import_cfh_, nullptr, nullptr));
 357
 358   for (int i = 0; i < 25; ++i) {
 359     ASSERT_TRUE(
 360         db_->Get(ReadOptions(), import_cfh_, Key(i), &value1).IsNotFound());
 361   }
 362   for (int i = 25; i < 50; ++i) {
 363     db_->Get(ReadOptions(), import_cfh_, Key(i), &value1);
 364     ASSERT_EQ(Key(i) + "_overwrite3", value1);
 365   }
 366   for (int i = 50; i < 100; ++i) {
 367     db_->Get(ReadOptions(), import_cfh_, Key(i), &value1);
 368     ASSERT_EQ(Key(i) + "_overwrite2", value1);
 369   }
 370
 371   for (int i = 0; i < 100; ++i) {
 372     db_->Get(ReadOptions(), import_cfh2_, Key(i), &value2);
 373     ASSERT_EQ(Get(1, Key(i)), value2);
 374   }
 375 }
 376
 377 TEST_F(ImportColumnFamilyTest, ImportExportedSSTFromAnotherDB) {
 378   Options options = CurrentOptions();
 379   CreateAndReopenWithCF({"koko"}, options);
 380
 381   for (int i = 0; i < 100; ++i) {
 382     Put(1, Key(i), Key(i) + "_val");
 383   }
 384   ASSERT_OK(Flush(1));
 385
 386   // Compact to create a L1 file.
 387   ASSERT_OK(
 388       db_->CompactRange(CompactRangeOptions(), handles_[1], nullptr, nullptr));
 389
 390   // Overwrite the value in the same set of keys.
 391   for (int i = 0; i < 50; ++i) {
 392     Put(1, Key(i), Key(i) + "_overwrite");
 393   }
 394
 395   // Flush to create L0 file.
 396   ASSERT_OK(Flush(1));
 397
 398   for (int i = 0; i < 25; ++i) {
 399     Put(1, Key(i), Key(i) + "_overwrite2");
 400   }
 401
 402   // Flush again to create another L0 file. It should have higher sequencer.
 403   ASSERT_OK(Flush(1));
 404
 405   Checkpoint* checkpoint;
 406   ASSERT_OK(Checkpoint::Create(db_, &checkpoint));
 407   ASSERT_OK(checkpoint->ExportColumnFamily(handles_[1], export_files_dir_,
 408                                            &metadata_ptr_));
 409   ASSERT_NE(metadata_ptr_, nullptr);
 410   delete checkpoint;
 411
 412   // Create a new db and import the files.
 413   DB* db_copy;
 414   test::DestroyDir(env_, dbname_ + "/db_copy");
 415   ASSERT_OK(DB::Open(options, dbname_ + "/db_copy", &db_copy));
 416   ColumnFamilyHandle* cfh = nullptr;
 417   ASSERT_OK(db_copy->CreateColumnFamilyWithImport(ColumnFamilyOptions(), "yoyo",
 418                                                   ImportColumnFamilyOptions(),
 419                                                   *metadata_ptr_, &cfh));
 420   ASSERT_NE(cfh, nullptr);
 421
 422   for (int i = 0; i < 100; ++i) {
 423     std::string value;
 424     db_copy->Get(ReadOptions(), cfh, Key(i), &value);
 425     ASSERT_EQ(Get(1, Key(i)), value);
 426   }
 427   db_copy->DropColumnFamily(cfh);
 428   db_copy->DestroyColumnFamilyHandle(cfh);
 429   delete db_copy;
 430   test::DestroyDir(env_, dbname_ + "/db_copy");
 431 }
 432
 433 TEST_F(ImportColumnFamilyTest, ImportColumnFamilyNegativeTest) {
 434   Options options = CurrentOptions();
 435   CreateAndReopenWithCF({"koko"}, options);
 436
 437   {
 438     // Create column family with existing cf name.
 439     ExportImportFilesMetaData metadata;
 440
 441     ASSERT_EQ(db_->CreateColumnFamilyWithImport(ColumnFamilyOptions(), "koko",
 442                                                 ImportColumnFamilyOptions(),
 443                                                 metadata, &import_cfh_),
 444               Status::InvalidArgument("Column family already exists"));
 445     ASSERT_EQ(import_cfh_, nullptr);
 446   }
 447
 448   {
 449     // Import with no files specified.
 450     ExportImportFilesMetaData metadata;
 451
 452     ASSERT_EQ(db_->CreateColumnFamilyWithImport(ColumnFamilyOptions(), "yoyo",
 453                                                 ImportColumnFamilyOptions(),
 454                                                 metadata, &import_cfh_),
 455               Status::InvalidArgument("The list of files is empty"));
 456     ASSERT_EQ(import_cfh_, nullptr);
 457   }
 458
 459   {
 460     // Import with overlapping keys in sst files.
 461     ExportImportFilesMetaData metadata;
 462     SstFileWriter sfw_cf1(EnvOptions(), options, handles_[1]);
 463     const std::string file1_sst_name = "file1.sst";
 464     const std::string file1_sst = sst_files_dir_ + file1_sst_name;
 465     ASSERT_OK(sfw_cf1.Open(file1_sst));
 466     ASSERT_OK(sfw_cf1.Put("K1", "V1"));
 467     ASSERT_OK(sfw_cf1.Put("K2", "V2"));
 468     ASSERT_OK(sfw_cf1.Finish());
 469     const std::string file2_sst_name = "file2.sst";
 470     const std::string file2_sst = sst_files_dir_ + file2_sst_name;
 471     ASSERT_OK(sfw_cf1.Open(file2_sst));
 472     ASSERT_OK(sfw_cf1.Put("K2", "V2"));
 473     ASSERT_OK(sfw_cf1.Put("K3", "V3"));
 474     ASSERT_OK(sfw_cf1.Finish());
 475
 476     metadata.files.push_back(
 477         LiveFileMetaDataInit(file1_sst_name, sst_files_dir_, 1, 10, 19));
 478     metadata.files.push_back(
 479         LiveFileMetaDataInit(file2_sst_name, sst_files_dir_, 1, 10, 19));
 480     metadata.db_comparator_name = options.comparator->Name();
 481
 482     ASSERT_EQ(db_->CreateColumnFamilyWithImport(ColumnFamilyOptions(), "yoyo",
 483                                                 ImportColumnFamilyOptions(),
 484                                                 metadata, &import_cfh_),
 485               Status::InvalidArgument("Files have overlapping ranges"));
 486     ASSERT_EQ(import_cfh_, nullptr);
 487   }
 488
 489   {
 490     // Import with a mismatching comparator, should fail with appropriate error.
 491     ExportImportFilesMetaData metadata;
 492     Options mismatch_options = CurrentOptions();
 493     mismatch_options.comparator = ReverseBytewiseComparator();
 494     SstFileWriter sfw_cf1(EnvOptions(), mismatch_options, handles_[1]);
 495     const std::string file1_sst_name = "file1.sst";
 496     const std::string file1_sst = sst_files_dir_ + file1_sst_name;
 497     ASSERT_OK(sfw_cf1.Open(file1_sst));
 498     ASSERT_OK(sfw_cf1.Put("K2", "V2"));
 499     ASSERT_OK(sfw_cf1.Put("K1", "V1"));
 500     ASSERT_OK(sfw_cf1.Finish());
 501
 502     metadata.files.push_back(
 503         LiveFileMetaDataInit(file1_sst_name, sst_files_dir_, 1, 10, 19));
 504     metadata.db_comparator_name = mismatch_options.comparator->Name();
 505
 506     ASSERT_EQ(db_->CreateColumnFamilyWithImport(ColumnFamilyOptions(), "coco",
 507                                                 ImportColumnFamilyOptions(),
 508                                                 metadata, &import_cfh_),
 509               Status::InvalidArgument("Comparator name mismatch"));
 510     ASSERT_EQ(import_cfh_, nullptr);
 511   }
 512
 513   {
 514     // Import with non existent sst file should fail with appropriate error
 515     ExportImportFilesMetaData metadata;
 516     SstFileWriter sfw_cf1(EnvOptions(), options, handles_[1]);
 517     const std::string file1_sst_name = "file1.sst";
 518     const std::string file1_sst = sst_files_dir_ + file1_sst_name;
 519     ASSERT_OK(sfw_cf1.Open(file1_sst));
 520     ASSERT_OK(sfw_cf1.Put("K1", "V1"));
 521     ASSERT_OK(sfw_cf1.Put("K2", "V2"));
 522     ASSERT_OK(sfw_cf1.Finish());
 523     const std::string file3_sst_name = "file3.sst";
 524
 525     metadata.files.push_back(
 526         LiveFileMetaDataInit(file1_sst_name, sst_files_dir_, 1, 10, 19));
 527     metadata.files.push_back(
 528         LiveFileMetaDataInit(file3_sst_name, sst_files_dir_, 1, 10, 19));
 529     metadata.db_comparator_name = options.comparator->Name();
 530
 531     ASSERT_EQ(db_->CreateColumnFamilyWithImport(ColumnFamilyOptions(), "yoyo",
 532                                                 ImportColumnFamilyOptions(),
 533                                                 metadata, &import_cfh_),
 534               Status::IOError("No such file or directory"));
 535     ASSERT_EQ(import_cfh_, nullptr);
 536
 537     // Test successful import after a failure with the same CF name. Ensures
 538     // there is no side effect with CF when there is a failed import
 539     metadata.files.pop_back();
 540     metadata.db_comparator_name = options.comparator->Name();
 541
 542     ASSERT_OK(db_->CreateColumnFamilyWithImport(ColumnFamilyOptions(), "yoyo",
 543                                                 ImportColumnFamilyOptions(),
 544                                                 metadata, &import_cfh_));
 545     ASSERT_NE(import_cfh_, nullptr);
 546   }
 547 }
 548
 549 }  // namespace ROCKSDB_NAMESPACE
 550
 551 int main(int argc, char** argv) {
 552   ROCKSDB_NAMESPACE::port::InstallStackTraceHandler();
 553   ::testing::InitGoogleTest(&argc, argv);
 554   return RUN_ALL_TESTS();
 555 }
 556
 557 #else
 558 #include <stdio.h>
 559
 560 int main(int /*argc*/, char** /*argv*/) {
 561   fprintf(stderr,
 562           "SKIPPED as External SST File Writer and Import are not supported "
 563           "in ROCKSDB_LITE\n");
 564   return 0;
 565 }
 566
 567 #endif  // !ROCKSDB_LITE