[ceph.git] / ceph / src / rocksdb / db / repair_test.cc

//  Copyright (c) 2016-present, Facebook, Inc.  All rights reserved.
//  This source code is licensed under both the GPLv2 (found in the
//  COPYING file in the root directory) and Apache 2.0 License
//  (found in the LICENSE.Apache file in the root directory).

#ifndef ROCKSDB_LITE

#include <algorithm>
#include <string>
#include <vector>

#include "db/db_impl/db_impl.h"
#include "db/db_test_util.h"
#include "file/file_util.h"
#include "rocksdb/comparator.h"
#include "rocksdb/db.h"
#include "rocksdb/transaction_log.h"
#include "util/string_util.h"

namespace ROCKSDB_NAMESPACE {

#ifndef ROCKSDB_LITE
class RepairTest : public DBTestBase {
 public:
  RepairTest() : DBTestBase("/repair_test") {}

  std::string GetFirstSstPath() {
    uint64_t manifest_size;
    std::vector<std::string> files;
    db_->GetLiveFiles(files, &manifest_size);
    auto sst_iter =
        std::find_if(files.begin(), files.end(), [](const std::string& file) {
          uint64_t number;
          FileType type;
          bool ok = ParseFileName(file, &number, &type);
          return ok && type == kTableFile;
        });
    return sst_iter == files.end() ? "" : dbname_ + *sst_iter;
  }
};

TEST_F(RepairTest, LostManifest) {
  // Add a couple SST files, delete the manifest, and verify RepairDB() saves
  // the day.
  Put("key", "val");
  Flush();
  Put("key2", "val2");
  Flush();
  // Need to get path before Close() deletes db_, but delete it after Close() to
  // ensure Close() didn't change the manifest.
  std::string manifest_path =
      DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());

  Close();
  ASSERT_OK(env_->FileExists(manifest_path));
  ASSERT_OK(env_->DeleteFile(manifest_path));
  ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
  Reopen(CurrentOptions());

  ASSERT_EQ(Get("key"), "val");
  ASSERT_EQ(Get("key2"), "val2");
}

TEST_F(RepairTest, CorruptManifest) {
  // Manifest is in an invalid format. Expect a full recovery.
  Put("key", "val");
  Flush();
  Put("key2", "val2");
  Flush();
  // Need to get path before Close() deletes db_, but overwrite it after Close()
  // to ensure Close() didn't change the manifest.
  std::string manifest_path =
      DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());

  Close();
  ASSERT_OK(env_->FileExists(manifest_path));

  LegacyFileSystemWrapper fs(env_);
  CreateFile(&fs, manifest_path, "blah", false /* use_fsync */);
  ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
  Reopen(CurrentOptions());

  ASSERT_EQ(Get("key"), "val");
  ASSERT_EQ(Get("key2"), "val2");
}

TEST_F(RepairTest, IncompleteManifest) {
  // In this case, the manifest is valid but does not reference all of the SST
  // files. Expect a full recovery.
  Put("key", "val");
  Flush();
  std::string orig_manifest_path =
      DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
  CopyFile(orig_manifest_path, orig_manifest_path + ".tmp");
  Put("key2", "val2");
  Flush();
  // Need to get path before Close() deletes db_, but overwrite it after Close()
  // to ensure Close() didn't change the manifest.
  std::string new_manifest_path =
      DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());

  Close();
  ASSERT_OK(env_->FileExists(new_manifest_path));
  // Replace the manifest with one that is only aware of the first SST file.
  CopyFile(orig_manifest_path + ".tmp", new_manifest_path);
  ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
  Reopen(CurrentOptions());

  ASSERT_EQ(Get("key"), "val");
  ASSERT_EQ(Get("key2"), "val2");
}

TEST_F(RepairTest, PostRepairSstFileNumbering) {
  // Verify after a DB is repaired, new files will be assigned higher numbers
  // than old files.
  Put("key", "val");
  Flush();
  Put("key2", "val2");
  Flush();
  uint64_t pre_repair_file_num = dbfull()->TEST_Current_Next_FileNo();
  Close();

  ASSERT_OK(RepairDB(dbname_, CurrentOptions()));

  Reopen(CurrentOptions());
  uint64_t post_repair_file_num = dbfull()->TEST_Current_Next_FileNo();
  ASSERT_GE(post_repair_file_num, pre_repair_file_num);
}

TEST_F(RepairTest, LostSst) {
  // Delete one of the SST files but preserve the manifest that refers to it,
  // then verify the DB is still usable for the intact SST.
  Put("key", "val");
  Flush();
  Put("key2", "val2");
  Flush();
  auto sst_path = GetFirstSstPath();
  ASSERT_FALSE(sst_path.empty());
  ASSERT_OK(env_->DeleteFile(sst_path));

  Close();
  ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
  Reopen(CurrentOptions());

  // Exactly one of the key-value pairs should be in the DB now.
  ASSERT_TRUE((Get("key") == "val") != (Get("key2") == "val2"));
}

TEST_F(RepairTest, CorruptSst) {
  // Corrupt one of the SST files but preserve the manifest that refers to it,
  // then verify the DB is still usable for the intact SST.
  Put("key", "val");
  Flush();
  Put("key2", "val2");
  Flush();
  auto sst_path = GetFirstSstPath();
  ASSERT_FALSE(sst_path.empty());

  LegacyFileSystemWrapper fs(env_);
  CreateFile(&fs, sst_path, "blah", false /* use_fsync */);

  Close();
  ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
  Reopen(CurrentOptions());

  // Exactly one of the key-value pairs should be in the DB now.
  ASSERT_TRUE((Get("key") == "val") != (Get("key2") == "val2"));
}

TEST_F(RepairTest, UnflushedSst) {
  // This test case invokes repair while some data is unflushed, then verifies
  // that data is in the db.
  Put("key", "val");
  VectorLogPtr wal_files;
  ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
  ASSERT_EQ(wal_files.size(), 1);
  uint64_t total_ssts_size;
  GetAllSSTFiles(&total_ssts_size);
  ASSERT_EQ(total_ssts_size, 0);
  // Need to get path before Close() deletes db_, but delete it after Close() to
  // ensure Close() didn't change the manifest.
  std::string manifest_path =
      DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());

  Close();
  ASSERT_OK(env_->FileExists(manifest_path));
  ASSERT_OK(env_->DeleteFile(manifest_path));
  ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
  Reopen(CurrentOptions());

  ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
  ASSERT_EQ(wal_files.size(), 0);
  GetAllSSTFiles(&total_ssts_size);
  ASSERT_GT(total_ssts_size, 0);
  ASSERT_EQ(Get("key"), "val");
}

TEST_F(RepairTest, SeparateWalDir) {
  do {
    Options options = CurrentOptions();
    DestroyAndReopen(options);
    Put("key", "val");
    Put("foo", "bar");
    VectorLogPtr wal_files;
    ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
    ASSERT_EQ(wal_files.size(), 1);
    uint64_t total_ssts_size;
    GetAllSSTFiles(&total_ssts_size);
    ASSERT_EQ(total_ssts_size, 0);
    std::string manifest_path =
      DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());

    Close();
    ASSERT_OK(env_->FileExists(manifest_path));
    ASSERT_OK(env_->DeleteFile(manifest_path));
    ASSERT_OK(RepairDB(dbname_, options));

    // make sure that all WALs are converted to SSTables.
    options.wal_dir = "";

    Reopen(options);
    ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
    ASSERT_EQ(wal_files.size(), 0);
    GetAllSSTFiles(&total_ssts_size);
    ASSERT_GT(total_ssts_size, 0);
    ASSERT_EQ(Get("key"), "val");
    ASSERT_EQ(Get("foo"), "bar");

 } while(ChangeWalOptions());
}

TEST_F(RepairTest, RepairMultipleColumnFamilies) {
  // Verify repair logic associates SST files with their original column
  // families.
  const int kNumCfs = 3;
  const int kEntriesPerCf = 2;
  DestroyAndReopen(CurrentOptions());
  CreateAndReopenWithCF({"pikachu1", "pikachu2"}, CurrentOptions());
  for (int i = 0; i < kNumCfs; ++i) {
    for (int j = 0; j < kEntriesPerCf; ++j) {
      Put(i, "key" + ToString(j), "val" + ToString(j));
      if (j == kEntriesPerCf - 1 && i == kNumCfs - 1) {
        // Leave one unflushed so we can verify WAL entries are properly
        // associated with column families.
        continue;
      }
      Flush(i);
    }
  }

  // Need to get path before Close() deletes db_, but delete it after Close() to
  // ensure Close() doesn't re-create the manifest.
  std::string manifest_path =
      DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
  Close();
  ASSERT_OK(env_->FileExists(manifest_path));
  ASSERT_OK(env_->DeleteFile(manifest_path));

  ASSERT_OK(RepairDB(dbname_, CurrentOptions()));

  ReopenWithColumnFamilies({"default", "pikachu1", "pikachu2"},
                           CurrentOptions());
  for (int i = 0; i < kNumCfs; ++i) {
    for (int j = 0; j < kEntriesPerCf; ++j) {
      ASSERT_EQ(Get(i, "key" + ToString(j)), "val" + ToString(j));
    }
  }
}

TEST_F(RepairTest, RepairColumnFamilyOptions) {
  // Verify repair logic uses correct ColumnFamilyOptions when repairing a
  // database with different options for column families.
  const int kNumCfs = 2;
  const int kEntriesPerCf = 2;

  Options opts(CurrentOptions()), rev_opts(CurrentOptions());
  opts.comparator = BytewiseComparator();
  rev_opts.comparator = ReverseBytewiseComparator();

  DestroyAndReopen(opts);
  CreateColumnFamilies({"reverse"}, rev_opts);
  ReopenWithColumnFamilies({"default", "reverse"},
                           std::vector<Options>{opts, rev_opts});
  for (int i = 0; i < kNumCfs; ++i) {
    for (int j = 0; j < kEntriesPerCf; ++j) {
      Put(i, "key" + ToString(j), "val" + ToString(j));
      if (i == kNumCfs - 1 && j == kEntriesPerCf - 1) {
        // Leave one unflushed so we can verify RepairDB's flush logic
        continue;
      }
      Flush(i);
    }
  }
  Close();

  // RepairDB() records the comparator in the manifest, and DB::Open would fail
  // if a different comparator were used.
  ASSERT_OK(RepairDB(dbname_, opts, {{"default", opts}, {"reverse", rev_opts}},
                     opts /* unknown_cf_opts */));
  ASSERT_OK(TryReopenWithColumnFamilies({"default", "reverse"},
                                        std::vector<Options>{opts, rev_opts}));
  for (int i = 0; i < kNumCfs; ++i) {
    for (int j = 0; j < kEntriesPerCf; ++j) {
      ASSERT_EQ(Get(i, "key" + ToString(j)), "val" + ToString(j));
    }
  }

  // Examine table properties to verify RepairDB() used the right options when
  // converting WAL->SST
  TablePropertiesCollection fname_to_props;
  db_->GetPropertiesOfAllTables(handles_[1], &fname_to_props);
  ASSERT_EQ(fname_to_props.size(), 2U);
  for (const auto& fname_and_props : fname_to_props) {
    std::string comparator_name (
      InternalKeyComparator(rev_opts.comparator).Name());
    comparator_name = comparator_name.substr(comparator_name.find(':') + 1);
    ASSERT_EQ(comparator_name,
              fname_and_props.second->comparator_name);
  }
  Close();

  // Also check comparator when it's provided via "unknown" CF options
  ASSERT_OK(RepairDB(dbname_, opts, {{"default", opts}},
                     rev_opts /* unknown_cf_opts */));
  ASSERT_OK(TryReopenWithColumnFamilies({"default", "reverse"},
                                        std::vector<Options>{opts, rev_opts}));
  for (int i = 0; i < kNumCfs; ++i) {
    for (int j = 0; j < kEntriesPerCf; ++j) {
      ASSERT_EQ(Get(i, "key" + ToString(j)), "val" + ToString(j));
    }
  }
}

TEST_F(RepairTest, DbNameContainsTrailingSlash) {
  {
    bool tmp;
    if (env_->AreFilesSame("", "", &tmp).IsNotSupported()) {
      fprintf(stderr,
              "skipping RepairTest.DbNameContainsTrailingSlash due to "
              "unsupported Env::AreFilesSame\n");
      return;
    }
  }

  Put("key", "val");
  Flush();
  Close();

  ASSERT_OK(RepairDB(dbname_ + "/", CurrentOptions()));
  Reopen(CurrentOptions());
  ASSERT_EQ(Get("key"), "val");
}
#endif  // ROCKSDB_LITE
}  // namespace ROCKSDB_NAMESPACE

int main(int argc, char** argv) {
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
}

#else
#include <stdio.h>

int main(int /*argc*/, char** /*argv*/) {
  fprintf(stderr, "SKIPPED as RepairDB is not supported in ROCKSDB_LITE\n");
  return 0;
}

#endif  // ROCKSDB_LITE
Commit	Line	Data
7c673cae	1	// Copyright (c) 2016-present, Facebook, Inc. All rights reserved.
11fdf7f2 TL	2	// This source code is licensed under both the GPLv2 (found in the
	3	// COPYING file in the root directory) and Apache 2.0 License
	4	// (found in the LICENSE.Apache file in the root directory).
7c673cae FG	5
	6	#ifndef ROCKSDB_LITE
	7
	8	#include <algorithm>
	9	#include <string>
	10	#include <vector>
	11
f67539c2	12	#include "db/db_impl/db_impl.h"
7c673cae	13	#include "db/db_test_util.h"
f67539c2	14	#include "file/file_util.h"
7c673cae FG	15	#include "rocksdb/comparator.h"
	16	#include "rocksdb/db.h"
	17	#include "rocksdb/transaction_log.h"
7c673cae FG	18	#include "util/string_util.h"
7c673cae FG	19
f67539c2	20	namespace ROCKSDB_NAMESPACE {
7c673cae FG	21
	22	#ifndef ROCKSDB_LITE
	23	class RepairTest : public DBTestBase {
	24	public:
	25	RepairTest() : DBTestBase("/repair_test") {}
	26
	27	std::string GetFirstSstPath() {
	28	uint64_t manifest_size;
	29	std::vector<std::string> files;
	30	db_->GetLiveFiles(files, &manifest_size);
	31	auto sst_iter =
	32	std::find_if(files.begin(), files.end(), [](const std::string& file) {
	33	uint64_t number;
	34	FileType type;
	35	bool ok = ParseFileName(file, &number, &type);
	36	return ok && type == kTableFile;
	37	});
	38	return sst_iter == files.end() ? "" : dbname_ + *sst_iter;
	39	}
	40	};
	41
	42	TEST_F(RepairTest, LostManifest) {
	43	// Add a couple SST files, delete the manifest, and verify RepairDB() saves
	44	// the day.
	45	Put("key", "val");
	46	Flush();
	47	Put("key2", "val2");
	48	Flush();
	49	// Need to get path before Close() deletes db_, but delete it after Close() to
	50	// ensure Close() didn't change the manifest.
	51	std::string manifest_path =
	52	DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
	53
	54	Close();
	55	ASSERT_OK(env_->FileExists(manifest_path));
	56	ASSERT_OK(env_->DeleteFile(manifest_path));
	57	ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
	58	Reopen(CurrentOptions());
	59
	60	ASSERT_EQ(Get("key"), "val");
	61	ASSERT_EQ(Get("key2"), "val2");
	62	}
	63
	64	TEST_F(RepairTest, CorruptManifest) {
	65	// Manifest is in an invalid format. Expect a full recovery.
	66	Put("key", "val");
	67	Flush();
	68	Put("key2", "val2");
	69	Flush();
	70	// Need to get path before Close() deletes db_, but overwrite it after Close()
	71	// to ensure Close() didn't change the manifest.
	72	std::string manifest_path =
	73	DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
	74
	75	Close();
	76	ASSERT_OK(env_->FileExists(manifest_path));
f67539c2 TL	77
	78	LegacyFileSystemWrapper fs(env_);
	79	CreateFile(&fs, manifest_path, "blah", false /* use_fsync */);
7c673cae FG	80	ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
	81	Reopen(CurrentOptions());
	82
	83	ASSERT_EQ(Get("key"), "val");
	84	ASSERT_EQ(Get("key2"), "val2");
	85	}
	86
	87	TEST_F(RepairTest, IncompleteManifest) {
	88	// In this case, the manifest is valid but does not reference all of the SST
	89	// files. Expect a full recovery.
	90	Put("key", "val");
	91	Flush();
	92	std::string orig_manifest_path =
	93	DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
	94	CopyFile(orig_manifest_path, orig_manifest_path + ".tmp");
	95	Put("key2", "val2");
	96	Flush();
	97	// Need to get path before Close() deletes db_, but overwrite it after Close()
	98	// to ensure Close() didn't change the manifest.
	99	std::string new_manifest_path =
	100	DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
	101
	102	Close();
	103	ASSERT_OK(env_->FileExists(new_manifest_path));
	104	// Replace the manifest with one that is only aware of the first SST file.
	105	CopyFile(orig_manifest_path + ".tmp", new_manifest_path);
	106	ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
	107	Reopen(CurrentOptions());
	108
	109	ASSERT_EQ(Get("key"), "val");
	110	ASSERT_EQ(Get("key2"), "val2");
	111	}
	112
11fdf7f2 TL	113	TEST_F(RepairTest, PostRepairSstFileNumbering) {
	114	// Verify after a DB is repaired, new files will be assigned higher numbers
	115	// than old files.
	116	Put("key", "val");
	117	Flush();
	118	Put("key2", "val2");
	119	Flush();
	120	uint64_t pre_repair_file_num = dbfull()->TEST_Current_Next_FileNo();
	121	Close();
	122
	123	ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
	124
	125	Reopen(CurrentOptions());
	126	uint64_t post_repair_file_num = dbfull()->TEST_Current_Next_FileNo();
	127	ASSERT_GE(post_repair_file_num, pre_repair_file_num);
	128	}
	129
7c673cae FG	130	TEST_F(RepairTest, LostSst) {
	131	// Delete one of the SST files but preserve the manifest that refers to it,
	132	// then verify the DB is still usable for the intact SST.
	133	Put("key", "val");
	134	Flush();
	135	Put("key2", "val2");
	136	Flush();
	137	auto sst_path = GetFirstSstPath();
	138	ASSERT_FALSE(sst_path.empty());
	139	ASSERT_OK(env_->DeleteFile(sst_path));
	140
	141	Close();
	142	ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
	143	Reopen(CurrentOptions());
	144
	145	// Exactly one of the key-value pairs should be in the DB now.
	146	ASSERT_TRUE((Get("key") == "val") != (Get("key2") == "val2"));
	147	}
	148
	149	TEST_F(RepairTest, CorruptSst) {
	150	// Corrupt one of the SST files but preserve the manifest that refers to it,
	151	// then verify the DB is still usable for the intact SST.
	152	Put("key", "val");
	153	Flush();
	154	Put("key2", "val2");
	155	Flush();
	156	auto sst_path = GetFirstSstPath();
	157	ASSERT_FALSE(sst_path.empty());
f67539c2 TL	158
	159	LegacyFileSystemWrapper fs(env_);
	160	CreateFile(&fs, sst_path, "blah", false /* use_fsync */);
7c673cae FG	161
	162	Close();
	163	ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
	164	Reopen(CurrentOptions());
	165
	166	// Exactly one of the key-value pairs should be in the DB now.
	167	ASSERT_TRUE((Get("key") == "val") != (Get("key2") == "val2"));
	168	}
	169
	170	TEST_F(RepairTest, UnflushedSst) {
	171	// This test case invokes repair while some data is unflushed, then verifies
	172	// that data is in the db.
	173	Put("key", "val");
	174	VectorLogPtr wal_files;
	175	ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
	176	ASSERT_EQ(wal_files.size(), 1);
	177	uint64_t total_ssts_size;
	178	GetAllSSTFiles(&total_ssts_size);
	179	ASSERT_EQ(total_ssts_size, 0);
	180	// Need to get path before Close() deletes db_, but delete it after Close() to
	181	// ensure Close() didn't change the manifest.
	182	std::string manifest_path =
	183	DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
	184
	185	Close();
	186	ASSERT_OK(env_->FileExists(manifest_path));
	187	ASSERT_OK(env_->DeleteFile(manifest_path));
	188	ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
	189	Reopen(CurrentOptions());
	190
	191	ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
	192	ASSERT_EQ(wal_files.size(), 0);
	193	GetAllSSTFiles(&total_ssts_size);
	194	ASSERT_GT(total_ssts_size, 0);
	195	ASSERT_EQ(Get("key"), "val");
	196	}
	197
11fdf7f2 TL	198	TEST_F(RepairTest, SeparateWalDir) {
	199	do {
	200	Options options = CurrentOptions();
	201	DestroyAndReopen(options);
	202	Put("key", "val");
	203	Put("foo", "bar");
	204	VectorLogPtr wal_files;
	205	ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
	206	ASSERT_EQ(wal_files.size(), 1);
	207	uint64_t total_ssts_size;
	208	GetAllSSTFiles(&total_ssts_size);
	209	ASSERT_EQ(total_ssts_size, 0);
	210	std::string manifest_path =
	211	DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
	212
	213	Close();
	214	ASSERT_OK(env_->FileExists(manifest_path));
	215	ASSERT_OK(env_->DeleteFile(manifest_path));
	216	ASSERT_OK(RepairDB(dbname_, options));
	217
	218	// make sure that all WALs are converted to SSTables.
	219	options.wal_dir = "";
	220
	221	Reopen(options);
	222	ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
	223	ASSERT_EQ(wal_files.size(), 0);
	224	GetAllSSTFiles(&total_ssts_size);
	225	ASSERT_GT(total_ssts_size, 0);
	226	ASSERT_EQ(Get("key"), "val");
	227	ASSERT_EQ(Get("foo"), "bar");
	228
	229	} while(ChangeWalOptions());
	230	}
	231
7c673cae FG	232	TEST_F(RepairTest, RepairMultipleColumnFamilies) {
	233	// Verify repair logic associates SST files with their original column
	234	// families.
	235	const int kNumCfs = 3;
	236	const int kEntriesPerCf = 2;
	237	DestroyAndReopen(CurrentOptions());
	238	CreateAndReopenWithCF({"pikachu1", "pikachu2"}, CurrentOptions());
	239	for (int i = 0; i < kNumCfs; ++i) {
	240	for (int j = 0; j < kEntriesPerCf; ++j) {
	241	Put(i, "key" + ToString(j), "val" + ToString(j));
	242	if (j == kEntriesPerCf - 1 && i == kNumCfs - 1) {
	243	// Leave one unflushed so we can verify WAL entries are properly
	244	// associated with column families.
	245	continue;
	246	}
	247	Flush(i);
	248	}
	249	}
	250
	251	// Need to get path before Close() deletes db_, but delete it after Close() to
	252	// ensure Close() doesn't re-create the manifest.
	253	std::string manifest_path =
	254	DescriptorFileName(dbname_, dbfull()->TEST_Current_Manifest_FileNo());
	255	Close();
	256	ASSERT_OK(env_->FileExists(manifest_path));
	257	ASSERT_OK(env_->DeleteFile(manifest_path));
	258
	259	ASSERT_OK(RepairDB(dbname_, CurrentOptions()));
	260
	261	ReopenWithColumnFamilies({"default", "pikachu1", "pikachu2"},
	262	CurrentOptions());
	263	for (int i = 0; i < kNumCfs; ++i) {
	264	for (int j = 0; j < kEntriesPerCf; ++j) {
	265	ASSERT_EQ(Get(i, "key" + ToString(j)), "val" + ToString(j));
	266	}
	267	}
	268	}
	269
	270	TEST_F(RepairTest, RepairColumnFamilyOptions) {
	271	// Verify repair logic uses correct ColumnFamilyOptions when repairing a
	272	// database with different options for column families.
	273	const int kNumCfs = 2;
	274	const int kEntriesPerCf = 2;
	275
	276	Options opts(CurrentOptions()), rev_opts(CurrentOptions());
	277	opts.comparator = BytewiseComparator();
	278	rev_opts.comparator = ReverseBytewiseComparator();
	279
	280	DestroyAndReopen(opts);
	281	CreateColumnFamilies({"reverse"}, rev_opts);
	282	ReopenWithColumnFamilies({"default", "reverse"},
	283	std::vector<Options>{opts, rev_opts});
	284	for (int i = 0; i < kNumCfs; ++i) {
	285	for (int j = 0; j < kEntriesPerCf; ++j) {
	286	Put(i, "key" + ToString(j), "val" + ToString(j));
	287	if (i == kNumCfs - 1 && j == kEntriesPerCf - 1) {
	288	// Leave one unflushed so we can verify RepairDB's flush logic
	289	continue;
	290	}
	291	Flush(i);
	292	}
	293	}
	294	Close();
	295
296	// RepairDB() records the comparator in the manifest, and DB::Open would fail
297	// if a different comparator were used.
298	ASSERT_OK(RepairDB(dbname_, opts, {{"default", opts}, {"reverse", rev_opts}},
299	opts /* unknown_cf_opts */));
300	ASSERT_OK(TryReopenWithColumnFamilies({"default", "reverse"},
301	std::vector<Options>{opts, rev_opts}));
302	for (int i = 0; i < kNumCfs; ++i) {
303	for (int j = 0; j < kEntriesPerCf; ++j) {
304	ASSERT_EQ(Get(i, "key" + ToString(j)), "val" + ToString(j));
305	}
306	}
307
308	// Examine table properties to verify RepairDB() used the right options when
309	// converting WAL->SST
310	TablePropertiesCollection fname_to_props;
311	db_->GetPropertiesOfAllTables(handles_[1], &fname_to_props);
312	ASSERT_EQ(fname_to_props.size(), 2U);
313	for (const auto& fname_and_props : fname_to_props) {
314	std::string comparator_name (
315	InternalKeyComparator(rev_opts.comparator).Name());
316	comparator_name = comparator_name.substr(comparator_name.find(':') + 1);
317	ASSERT_EQ(comparator_name,
318	fname_and_props.second->comparator_name);
319	}
494da23a	320	Close();
7c673cae FG	321
	322	// Also check comparator when it's provided via "unknown" CF options
	323	ASSERT_OK(RepairDB(dbname_, opts, {{"default", opts}},
	324	rev_opts /* unknown_cf_opts */));
	325	ASSERT_OK(TryReopenWithColumnFamilies({"default", "reverse"},
	326	std::vector<Options>{opts, rev_opts}));
	327	for (int i = 0; i < kNumCfs; ++i) {
	328	for (int j = 0; j < kEntriesPerCf; ++j) {
	329	ASSERT_EQ(Get(i, "key" + ToString(j)), "val" + ToString(j));
	330	}
	331	}
	332	}
	333
11fdf7f2 TL	334	TEST_F(RepairTest, DbNameContainsTrailingSlash) {
	335	{
	336	bool tmp;
	337	if (env_->AreFilesSame("", "", &tmp).IsNotSupported()) {
	338	fprintf(stderr,
	339	"skipping RepairTest.DbNameContainsTrailingSlash due to "
	340	"unsupported Env::AreFilesSame\n");
	341	return;
	342	}
	343	}
	344
	345	Put("key", "val");
	346	Flush();
	347	Close();
	348
	349	ASSERT_OK(RepairDB(dbname_ + "/", CurrentOptions()));
	350	Reopen(CurrentOptions());
	351	ASSERT_EQ(Get("key"), "val");
	352	}
7c673cae	353	#endif // ROCKSDB_LITE
f67539c2	354	} // namespace ROCKSDB_NAMESPACE
7c673cae FG	355
	356	int main(int argc, char** argv) {
	357	::testing::InitGoogleTest(&argc, argv);
	358	return RUN_ALL_TESTS();
	359	}
	360
	361	#else
	362	#include <stdio.h>
	363
11fdf7f2	364	int main(int /argc/, char** /argv/) {
7c673cae FG	365	fprintf(stderr, "SKIPPED as RepairDB is not supported in ROCKSDB_LITE\n");
	366	return 0;
	367	}
	368
	369	#endif // ROCKSDB_LITE