[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 "env/mock_env.h"
#include "port/stack_trace.h"

namespace ROCKSDB_NAMESPACE {

class DBTestXactLogIterator : public DBTestBase {
 public:
  DBTestXactLogIterator()
      : DBTestBase("db_log_iter_test", /*env_do_fsync=*/true) {}

  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, bool expect_ok = true) {
  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();
  }
  if (expect_ok) {
    EXPECT_OK(iter->status());
  } else {
    EXPECT_NOK(iter->status());
  }
  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);
}
}  // anonymous namespace

TEST_F(DBTestXactLogIterator, TransactionLogIterator) {
  do {
    Options options = OptionsForLogIterTest();
    DestroyAndReopen(options);
    CreateAndReopenWithCF({"pikachu"}, options);
    ASSERT_OK(Put(0, "key1", DummyString(1024)));
    ASSERT_OK(Put(1, "key2", DummyString(1024)));
    ASSERT_OK(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);
    {
      ASSERT_OK(Put(0, "key4", DummyString(1024)));
      ASSERT_OK(Put(1, "key5", DummyString(1024)));
      ASSERT_OK(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_NAMESPACE::SyncPoint::GetInstance()->LoadDependency({
        {sync_points[test][0], sync_points[test][1]},
        {sync_points[test][2], sync_points[test][3]},
    });

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

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

      ROCKSDB_NAMESPACE::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;
      ASSERT_OK(dbfull()->Flush(flush_options));

      // "key5" would be written in a new memtable and log
      ASSERT_OK(Put("key5", DummyString(1024)));
      ASSERT_OK(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);
    ASSERT_OK(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());
    ASSERT_OK(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);
    ASSERT_OK(Put("key1", DummyString(1024)));
    ASSERT_OK(Put("key2", DummyString(1023)));
    ASSERT_OK(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++) {
      ASSERT_OK(Put("key" + std::to_string(i), DummyString(10)));
    }

    ASSERT_OK(Flush());
    ASSERT_OK(db_->FlushWAL(false));

    // Corrupt this log to create a gap
    ASSERT_OK(db_->DisableFileDeletions());

    VectorLogPtr wal_files;
    ASSERT_OK(db_->GetSortedWalFiles(wal_files));
    ASSERT_FALSE(wal_files.empty());

    const auto logfile_path = dbname_ + "/" + wal_files.front()->PathName();
    ASSERT_OK(test::TruncateFile(env_, logfile_path,
                                 wal_files.front()->SizeFileBytes() / 2));

    ASSERT_OK(db_->EnableFileDeletions());

    // Insert a new entry to a new log file
    ASSERT_OK(Put("key1025", DummyString(10)));
    ASSERT_OK(db_->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 = 0;
    SequenceNumber last_sequence_read = ReadRecords(iter, count, false);
    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;
    ASSERT_OK(batch.Put(handles_[1], "key1", DummyString(1024)));
    ASSERT_OK(batch.Put(handles_[0], "key2", DummyString(1024)));
    ASSERT_OK(batch.Put(handles_[1], "key3", DummyString(1024)));
    ASSERT_OK(batch.Delete(handles_[0], "key2"));
    ASSERT_OK(dbfull()->Write(WriteOptions(), &batch));
    ASSERT_OK(Flush(1));
    ASSERT_OK(Flush(0));
    ReopenWithColumnFamilies({"default", "pikachu"}, options);
    ASSERT_OK(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;
    ASSERT_OK(batch.Put(handles_[1], "key1", DummyString(1024)));
    ASSERT_OK(batch.Put(handles_[0], "key2", DummyString(1024)));
    ASSERT_OK(batch.PutLogData(Slice("blob1")));
    ASSERT_OK(batch.Put(handles_[1], "key3", DummyString(1024)));
    ASSERT_OK(batch.PutLogData(Slice("blob2")));
    ASSERT_OK(batch.Delete(handles_[0], "key2"));
    ASSERT_OK(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(" + std::to_string(cf) + ", " + key.ToString() + ", " +
              std::to_string(value.size()) + ")";
      return Status::OK();
    }
    Status MergeCF(uint32_t cf, const Slice& key, const Slice& value) override {
      seen += "Merge(" + std::to_string(cf) + ", " + key.ToString() + ", " +
              std::to_string(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(" + std::to_string(cf) + ", " + key.ToString() + ")";
      return Status::OK();
    }
  } handler;
  ASSERT_OK(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_NAMESPACE

#endif  // !defined(ROCKSDB_LITE)

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