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

//  Copyright (c) 2011-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).
//
// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file. See the AUTHORS file for names of contributors.

// Introduction of SyncPoint effectively disabled building and running this test
// in Release build.
// which is a pity, it is a good test
#if !defined(ROCKSDB_LITE)

#include "db/db_test_util.h"
#include "port/stack_trace.h"

namespace rocksdb {

class DBTestXactLogIterator : public DBTestBase {
 public:
  DBTestXactLogIterator() : DBTestBase("/db_log_iter_test") {}

  std::unique_ptr<TransactionLogIterator> OpenTransactionLogIter(
      const SequenceNumber seq) {
    std::unique_ptr<TransactionLogIterator> iter;
    Status status = dbfull()->GetUpdatesSince(seq, &iter);
    EXPECT_OK(status);
    EXPECT_TRUE(iter->Valid());
    return iter;
  }
};

namespace {
SequenceNumber ReadRecords(
    std::unique_ptr<TransactionLogIterator>& iter,
    int& count) {
  count = 0;
  SequenceNumber lastSequence = 0;
  BatchResult res;
  while (iter->Valid()) {
    res = iter->GetBatch();
    EXPECT_TRUE(res.sequence > lastSequence);
    ++count;
    lastSequence = res.sequence;
    EXPECT_OK(iter->status());
    iter->Next();
  }
  return res.sequence;
}

void ExpectRecords(
    const int expected_no_records,
    std::unique_ptr<TransactionLogIterator>& iter) {
  int num_records;
  ReadRecords(iter, num_records);
  ASSERT_EQ(num_records, expected_no_records);
}
}  // namespace

TEST_F(DBTestXactLogIterator, TransactionLogIterator) {
  do {
    Options options = OptionsForLogIterTest();
    DestroyAndReopen(options);
    CreateAndReopenWithCF({"pikachu"}, options);
    Put(0, "key1", DummyString(1024));
    Put(1, "key2", DummyString(1024));
    Put(1, "key2", DummyString(1024));
    ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 3U);
    {
      auto iter = OpenTransactionLogIter(0);
      ExpectRecords(3, iter);
    }
    ReopenWithColumnFamilies({"default", "pikachu"}, options);
    env_->SleepForMicroseconds(2 * 1000 * 1000);
    {
      Put(0, "key4", DummyString(1024));
      Put(1, "key5", DummyString(1024));
      Put(0, "key6", DummyString(1024));
    }
    {
      auto iter = OpenTransactionLogIter(0);
      ExpectRecords(6, iter);
    }
  } while (ChangeCompactOptions());
}

#ifndef NDEBUG  // sync point is not included with DNDEBUG build
TEST_F(DBTestXactLogIterator, TransactionLogIteratorRace) {
  static const int LOG_ITERATOR_RACE_TEST_COUNT = 2;
  static const char* sync_points[LOG_ITERATOR_RACE_TEST_COUNT][4] = {
      {"WalManager::GetSortedWalFiles:1",  "WalManager::PurgeObsoleteFiles:1",
       "WalManager::PurgeObsoleteFiles:2", "WalManager::GetSortedWalFiles:2"},
      {"WalManager::GetSortedWalsOfType:1",
       "WalManager::PurgeObsoleteFiles:1",
       "WalManager::PurgeObsoleteFiles:2",
       "WalManager::GetSortedWalsOfType:2"}};
  for (int test = 0; test < LOG_ITERATOR_RACE_TEST_COUNT; ++test) {
    // Setup sync point dependency to reproduce the race condition of
    // a log file moved to archived dir, in the middle of GetSortedWalFiles
    rocksdb::SyncPoint::GetInstance()->LoadDependency(
      { { sync_points[test][0], sync_points[test][1] },
        { sync_points[test][2], sync_points[test][3] },
      });

    do {
      rocksdb::SyncPoint::GetInstance()->ClearTrace();
      rocksdb::SyncPoint::GetInstance()->DisableProcessing();
      Options options = OptionsForLogIterTest();
      DestroyAndReopen(options);
      Put("key1", DummyString(1024));
      dbfull()->Flush(FlushOptions());
      Put("key2", DummyString(1024));
      dbfull()->Flush(FlushOptions());
      Put("key3", DummyString(1024));
      dbfull()->Flush(FlushOptions());
      Put("key4", DummyString(1024));
      ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 4U);
      dbfull()->FlushWAL(false);

      {
        auto iter = OpenTransactionLogIter(0);
        ExpectRecords(4, iter);
      }

      rocksdb::SyncPoint::GetInstance()->EnableProcessing();
      // trigger async flush, and log move. Well, log move will
      // wait until the GetSortedWalFiles:1 to reproduce the race
      // condition
      FlushOptions flush_options;
      flush_options.wait = false;
      dbfull()->Flush(flush_options);

      // "key5" would be written in a new memtable and log
      Put("key5", DummyString(1024));
      dbfull()->FlushWAL(false);
      {
        // this iter would miss "key4" if not fixed
        auto iter = OpenTransactionLogIter(0);
        ExpectRecords(5, iter);
      }
    } while (ChangeCompactOptions());
  }
}
#endif

TEST_F(DBTestXactLogIterator, TransactionLogIteratorStallAtLastRecord) {
  do {
    Options options = OptionsForLogIterTest();
    DestroyAndReopen(options);
    Put("key1", DummyString(1024));
    auto iter = OpenTransactionLogIter(0);
    ASSERT_OK(iter->status());
    ASSERT_TRUE(iter->Valid());
    iter->Next();
    ASSERT_TRUE(!iter->Valid());
    ASSERT_OK(iter->status());
    Put("key2", DummyString(1024));
    iter->Next();
    ASSERT_OK(iter->status());
    ASSERT_TRUE(iter->Valid());
  } while (ChangeCompactOptions());
}

TEST_F(DBTestXactLogIterator, TransactionLogIteratorCheckAfterRestart) {
  do {
    Options options = OptionsForLogIterTest();
    DestroyAndReopen(options);
    Put("key1", DummyString(1024));
    Put("key2", DummyString(1023));
    dbfull()->Flush(FlushOptions());
    Reopen(options);
    auto iter = OpenTransactionLogIter(0);
    ExpectRecords(2, iter);
  } while (ChangeCompactOptions());
}

TEST_F(DBTestXactLogIterator, TransactionLogIteratorCorruptedLog) {
  do {
    Options options = OptionsForLogIterTest();
    DestroyAndReopen(options);
    for (int i = 0; i < 1024; i++) {
      Put("key"+ToString(i), DummyString(10));
    }
    dbfull()->Flush(FlushOptions());
    dbfull()->FlushWAL(false);
    // Corrupt this log to create a gap
    rocksdb::VectorLogPtr wal_files;
    ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
    const auto logfile_path = dbname_ + "/" + wal_files.front()->PathName();
    if (mem_env_) {
      mem_env_->Truncate(logfile_path, wal_files.front()->SizeFileBytes() / 2);
    } else {
      ASSERT_EQ(0, truncate(logfile_path.c_str(),
                   wal_files.front()->SizeFileBytes() / 2));
    }

    // Insert a new entry to a new log file
    Put("key1025", DummyString(10));
    dbfull()->FlushWAL(false);
    // Try to read from the beginning. Should stop before the gap and read less
    // than 1025 entries
    auto iter = OpenTransactionLogIter(0);
    int count;
    SequenceNumber last_sequence_read = ReadRecords(iter, count);
    ASSERT_LT(last_sequence_read, 1025U);
    // Try to read past the gap, should be able to seek to key1025
    auto iter2 = OpenTransactionLogIter(last_sequence_read + 1);
    ExpectRecords(1, iter2);
  } while (ChangeCompactOptions());
}

TEST_F(DBTestXactLogIterator, TransactionLogIteratorBatchOperations) {
  do {
    Options options = OptionsForLogIterTest();
    DestroyAndReopen(options);
    CreateAndReopenWithCF({"pikachu"}, options);
    WriteBatch batch;
    batch.Put(handles_[1], "key1", DummyString(1024));
    batch.Put(handles_[0], "key2", DummyString(1024));
    batch.Put(handles_[1], "key3", DummyString(1024));
    batch.Delete(handles_[0], "key2");
    dbfull()->Write(WriteOptions(), &batch);
    Flush(1);
    Flush(0);
    ReopenWithColumnFamilies({"default", "pikachu"}, options);
    Put(1, "key4", DummyString(1024));
    auto iter = OpenTransactionLogIter(3);
    ExpectRecords(2, iter);
  } while (ChangeCompactOptions());
}

TEST_F(DBTestXactLogIterator, TransactionLogIteratorBlobs) {
  Options options = OptionsForLogIterTest();
  DestroyAndReopen(options);
  CreateAndReopenWithCF({"pikachu"}, options);
  {
    WriteBatch batch;
    batch.Put(handles_[1], "key1", DummyString(1024));
    batch.Put(handles_[0], "key2", DummyString(1024));
    batch.PutLogData(Slice("blob1"));
    batch.Put(handles_[1], "key3", DummyString(1024));
    batch.PutLogData(Slice("blob2"));
    batch.Delete(handles_[0], "key2");
    dbfull()->Write(WriteOptions(), &batch);
    ReopenWithColumnFamilies({"default", "pikachu"}, options);
  }

  auto res = OpenTransactionLogIter(0)->GetBatch();
  struct Handler : public WriteBatch::Handler {
    std::string seen;
    Status PutCF(uint32_t cf, const Slice& key, const Slice& value) override {
      seen += "Put(" + ToString(cf) + ", " + key.ToString() + ", " +
              ToString(value.size()) + ")";
      return Status::OK();
    }
    Status MergeCF(uint32_t cf, const Slice& key, const Slice& value) override {
      seen += "Merge(" + ToString(cf) + ", " + key.ToString() + ", " +
              ToString(value.size()) + ")";
      return Status::OK();
    }
    void LogData(const Slice& blob) override {
      seen += "LogData(" + blob.ToString() + ")";
    }
    Status DeleteCF(uint32_t cf, const Slice& key) override {
      seen += "Delete(" + ToString(cf) + ", " + key.ToString() + ")";
      return Status::OK();
    }
  } handler;
  res.writeBatchPtr->Iterate(&handler);
  ASSERT_EQ(
      "Put(1, key1, 1024)"
      "Put(0, key2, 1024)"
      "LogData(blob1)"
      "Put(1, key3, 1024)"
      "LogData(blob2)"
      "Delete(0, key2)",
      handler.seen);
}
}  // namespace rocksdb

#endif  // !defined(ROCKSDB_LITE)

int main(int argc, char** argv) {
#if !defined(ROCKSDB_LITE)
  rocksdb::port::InstallStackTraceHandler();
  ::testing::InitGoogleTest(&argc, argv);
  return RUN_ALL_TESTS();
#else
  (void) argc;
  (void) argv;
  return 0;
#endif
}
Commit	Line	Data
7c673cae	1	// Copyright (c) 2011-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	// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
	7	// Use of this source code is governed by a BSD-style license that can be
	8	// found in the LICENSE file. See the AUTHORS file for names of contributors.
	9
	10	// Introduction of SyncPoint effectively disabled building and running this test
	11	// in Release build.
	12	// which is a pity, it is a good test
	13	#if !defined(ROCKSDB_LITE)
	14
	15	#include "db/db_test_util.h"
	16	#include "port/stack_trace.h"
	17
	18	namespace rocksdb {
	19
	20	class DBTestXactLogIterator : public DBTestBase {
	21	public:
	22	DBTestXactLogIterator() : DBTestBase("/db_log_iter_test") {}
	23
	24	std::unique_ptr<TransactionLogIterator> OpenTransactionLogIter(
	25	const SequenceNumber seq) {
494da23a	26	std::unique_ptr<TransactionLogIterator> iter;
7c673cae FG	27	Status status = dbfull()->GetUpdatesSince(seq, &iter);
	28	EXPECT_OK(status);
	29	EXPECT_TRUE(iter->Valid());
	30	return iter;
	31	}
	32	};
	33
	34	namespace {
	35	SequenceNumber ReadRecords(
	36	std::unique_ptr<TransactionLogIterator>& iter,
	37	int& count) {
	38	count = 0;
	39	SequenceNumber lastSequence = 0;
	40	BatchResult res;
	41	while (iter->Valid()) {
	42	res = iter->GetBatch();
	43	EXPECT_TRUE(res.sequence > lastSequence);
	44	++count;
	45	lastSequence = res.sequence;
	46	EXPECT_OK(iter->status());
	47	iter->Next();
	48	}
	49	return res.sequence;
	50	}
	51
	52	void ExpectRecords(
	53	const int expected_no_records,
	54	std::unique_ptr<TransactionLogIterator>& iter) {
	55	int num_records;
	56	ReadRecords(iter, num_records);
	57	ASSERT_EQ(num_records, expected_no_records);
	58	}
	59	} // namespace
	60
	61	TEST_F(DBTestXactLogIterator, TransactionLogIterator) {
	62	do {
	63	Options options = OptionsForLogIterTest();
	64	DestroyAndReopen(options);
	65	CreateAndReopenWithCF({"pikachu"}, options);
	66	Put(0, "key1", DummyString(1024));
	67	Put(1, "key2", DummyString(1024));
	68	Put(1, "key2", DummyString(1024));
	69	ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 3U);
	70	{
	71	auto iter = OpenTransactionLogIter(0);
	72	ExpectRecords(3, iter);
	73	}
	74	ReopenWithColumnFamilies({"default", "pikachu"}, options);
	75	env_->SleepForMicroseconds(2 * 1000 * 1000);
	76	{
	77	Put(0, "key4", DummyString(1024));
	78	Put(1, "key5", DummyString(1024));
	79	Put(0, "key6", DummyString(1024));
	80	}
	81	{
	82	auto iter = OpenTransactionLogIter(0);
	83	ExpectRecords(6, iter);
	84	}
	85	} while (ChangeCompactOptions());
	86	}
	87
	88	#ifndef NDEBUG // sync point is not included with DNDEBUG build
	89	TEST_F(DBTestXactLogIterator, TransactionLogIteratorRace) {
	90	static const int LOG_ITERATOR_RACE_TEST_COUNT = 2;
91	static const char* sync_points[LOG_ITERATOR_RACE_TEST_COUNT][4] = {
92	{"WalManager::GetSortedWalFiles:1", "WalManager::PurgeObsoleteFiles:1",
93	"WalManager::PurgeObsoleteFiles:2", "WalManager::GetSortedWalFiles:2"},
94	{"WalManager::GetSortedWalsOfType:1",
95	"WalManager::PurgeObsoleteFiles:1",
96	"WalManager::PurgeObsoleteFiles:2",
97	"WalManager::GetSortedWalsOfType:2"}};
98	for (int test = 0; test < LOG_ITERATOR_RACE_TEST_COUNT; ++test) {
99	// Setup sync point dependency to reproduce the race condition of
100	// a log file moved to archived dir, in the middle of GetSortedWalFiles
101	rocksdb::SyncPoint::GetInstance()->LoadDependency(
102	{ { sync_points[test][0], sync_points[test][1] },
103	{ sync_points[test][2], sync_points[test][3] },
104	});
105
106	do {
107	rocksdb::SyncPoint::GetInstance()->ClearTrace();
108	rocksdb::SyncPoint::GetInstance()->DisableProcessing();
109	Options options = OptionsForLogIterTest();
110	DestroyAndReopen(options);
111	Put("key1", DummyString(1024));
112	dbfull()->Flush(FlushOptions());
113	Put("key2", DummyString(1024));
114	dbfull()->Flush(FlushOptions());
115	Put("key3", DummyString(1024));
116	dbfull()->Flush(FlushOptions());
117	Put("key4", DummyString(1024));
118	ASSERT_EQ(dbfull()->GetLatestSequenceNumber(), 4U);
11fdf7f2	119	dbfull()->FlushWAL(false);
7c673cae FG	120
	121	{
	122	auto iter = OpenTransactionLogIter(0);
	123	ExpectRecords(4, iter);
	124	}
	125
	126	rocksdb::SyncPoint::GetInstance()->EnableProcessing();
	127	// trigger async flush, and log move. Well, log move will
	128	// wait until the GetSortedWalFiles:1 to reproduce the race
	129	// condition
	130	FlushOptions flush_options;
	131	flush_options.wait = false;
	132	dbfull()->Flush(flush_options);
	133
	134	// "key5" would be written in a new memtable and log
	135	Put("key5", DummyString(1024));
11fdf7f2	136	dbfull()->FlushWAL(false);
7c673cae FG	137	{
	138	// this iter would miss "key4" if not fixed
	139	auto iter = OpenTransactionLogIter(0);
	140	ExpectRecords(5, iter);
	141	}
	142	} while (ChangeCompactOptions());
	143	}
	144	}
	145	#endif
	146
	147	TEST_F(DBTestXactLogIterator, TransactionLogIteratorStallAtLastRecord) {
	148	do {
	149	Options options = OptionsForLogIterTest();
	150	DestroyAndReopen(options);
	151	Put("key1", DummyString(1024));
	152	auto iter = OpenTransactionLogIter(0);
	153	ASSERT_OK(iter->status());
	154	ASSERT_TRUE(iter->Valid());
	155	iter->Next();
	156	ASSERT_TRUE(!iter->Valid());
	157	ASSERT_OK(iter->status());
	158	Put("key2", DummyString(1024));
	159	iter->Next();
	160	ASSERT_OK(iter->status());
	161	ASSERT_TRUE(iter->Valid());
	162	} while (ChangeCompactOptions());
	163	}
	164
	165	TEST_F(DBTestXactLogIterator, TransactionLogIteratorCheckAfterRestart) {
	166	do {
	167	Options options = OptionsForLogIterTest();
	168	DestroyAndReopen(options);
	169	Put("key1", DummyString(1024));
	170	Put("key2", DummyString(1023));
	171	dbfull()->Flush(FlushOptions());
	172	Reopen(options);
	173	auto iter = OpenTransactionLogIter(0);
	174	ExpectRecords(2, iter);
	175	} while (ChangeCompactOptions());
	176	}
	177
	178	TEST_F(DBTestXactLogIterator, TransactionLogIteratorCorruptedLog) {
	179	do {
	180	Options options = OptionsForLogIterTest();
	181	DestroyAndReopen(options);
	182	for (int i = 0; i < 1024; i++) {
	183	Put("key"+ToString(i), DummyString(10));
	184	}
	185	dbfull()->Flush(FlushOptions());
11fdf7f2	186	dbfull()->FlushWAL(false);
7c673cae FG	187	// Corrupt this log to create a gap
	188	rocksdb::VectorLogPtr wal_files;
	189	ASSERT_OK(dbfull()->GetSortedWalFiles(wal_files));
	190	const auto logfile_path = dbname_ + "/" + wal_files.front()->PathName();
	191	if (mem_env_) {
	192	mem_env_->Truncate(logfile_path, wal_files.front()->SizeFileBytes() / 2);
	193	} else {
	194	ASSERT_EQ(0, truncate(logfile_path.c_str(),
	195	wal_files.front()->SizeFileBytes() / 2));
	196	}
	197
	198	// Insert a new entry to a new log file
	199	Put("key1025", DummyString(10));
11fdf7f2	200	dbfull()->FlushWAL(false);
7c673cae FG	201	// Try to read from the beginning. Should stop before the gap and read less
	202	// than 1025 entries
	203	auto iter = OpenTransactionLogIter(0);
	204	int count;
	205	SequenceNumber last_sequence_read = ReadRecords(iter, count);
	206	ASSERT_LT(last_sequence_read, 1025U);
	207	// Try to read past the gap, should be able to seek to key1025
	208	auto iter2 = OpenTransactionLogIter(last_sequence_read + 1);
	209	ExpectRecords(1, iter2);
	210	} while (ChangeCompactOptions());
	211	}
	212
	213	TEST_F(DBTestXactLogIterator, TransactionLogIteratorBatchOperations) {
	214	do {
	215	Options options = OptionsForLogIterTest();
	216	DestroyAndReopen(options);
	217	CreateAndReopenWithCF({"pikachu"}, options);
	218	WriteBatch batch;
	219	batch.Put(handles_[1], "key1", DummyString(1024));
	220	batch.Put(handles_[0], "key2", DummyString(1024));
	221	batch.Put(handles_[1], "key3", DummyString(1024));
	222	batch.Delete(handles_[0], "key2");
	223	dbfull()->Write(WriteOptions(), &batch);
	224	Flush(1);
	225	Flush(0);
	226	ReopenWithColumnFamilies({"default", "pikachu"}, options);
	227	Put(1, "key4", DummyString(1024));
	228	auto iter = OpenTransactionLogIter(3);
	229	ExpectRecords(2, iter);
	230	} while (ChangeCompactOptions());
	231	}
	232
	233	TEST_F(DBTestXactLogIterator, TransactionLogIteratorBlobs) {
	234	Options options = OptionsForLogIterTest();
	235	DestroyAndReopen(options);
	236	CreateAndReopenWithCF({"pikachu"}, options);
	237	{
	238	WriteBatch batch;
	239	batch.Put(handles_[1], "key1", DummyString(1024));
	240	batch.Put(handles_[0], "key2", DummyString(1024));
	241	batch.PutLogData(Slice("blob1"));
	242	batch.Put(handles_[1], "key3", DummyString(1024));
	243	batch.PutLogData(Slice("blob2"));
	244	batch.Delete(handles_[0], "key2");
	245	dbfull()->Write(WriteOptions(), &batch);
	246	ReopenWithColumnFamilies({"default", "pikachu"}, options);
	247	}
	248
	249	auto res = OpenTransactionLogIter(0)->GetBatch();
	250	struct Handler : public WriteBatch::Handler {
	251	std::string seen;
494da23a	252	Status PutCF(uint32_t cf, const Slice& key, const Slice& value) override {
7c673cae FG	253	seen += "Put(" + ToString(cf) + ", " + key.ToString() + ", " +
	254	ToString(value.size()) + ")";
	255	return Status::OK();
	256	}
494da23a	257	Status MergeCF(uint32_t cf, const Slice& key, const Slice& value) override {
7c673cae FG	258	seen += "Merge(" + ToString(cf) + ", " + key.ToString() + ", " +
	259	ToString(value.size()) + ")";
	260	return Status::OK();
	261	}
494da23a	262	void LogData(const Slice& blob) override {
7c673cae FG	263	seen += "LogData(" + blob.ToString() + ")";
7c673cae FG	264	}
494da23a	265	Status DeleteCF(uint32_t cf, const Slice& key) override {
7c673cae FG	266	seen += "Delete(" + ToString(cf) + ", " + key.ToString() + ")";
	267	return Status::OK();
	268	}
	269	} handler;
	270	res.writeBatchPtr->Iterate(&handler);
	271	ASSERT_EQ(
	272	"Put(1, key1, 1024)"
	273	"Put(0, key2, 1024)"
	274	"LogData(blob1)"
	275	"Put(1, key3, 1024)"
	276	"LogData(blob2)"
	277	"Delete(0, key2)",
	278	handler.seen);
	279	}
	280	} // namespace rocksdb
	281
	282	#endif // !defined(ROCKSDB_LITE)
	283
	284	int main(int argc, char** argv) {
	285	#if !defined(ROCKSDB_LITE)
	286	rocksdb::port::InstallStackTraceHandler();
	287	::testing::InitGoogleTest(&argc, argv);
	288	return RUN_ALL_TESTS();
	289	#else
11fdf7f2 TL	290	(void) argc;
11fdf7f2 TL	291	(void) argv;
7c673cae FG	292	return 0;
	293	#endif
	294	}