1 // Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
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).
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 #include "db/db_test_util.h"
11
12 #include "db/forward_iterator.h"
13 #include "env/mock_env.h"
14 #include "rocksdb/convenience.h"
15 #include "rocksdb/env_encryption.h"
16 #include "rocksdb/unique_id.h"
17 #include "rocksdb/utilities/object_registry.h"
18 #include "util/random.h"
19
20 namespace ROCKSDB_NAMESPACE {
21
22 namespace {
MaybeCurrentTime(Env * env)23 int64_t MaybeCurrentTime(Env* env) {
24 int64_t time = 1337346000; // arbitrary fallback default
25 env->GetCurrentTime(&time).PermitUncheckedError();
26 return time;
27 }
28 } // namespace
29
30 // Special Env used to delay background operations
31
SpecialEnv(Env * base,bool time_elapse_only_sleep)32 SpecialEnv::SpecialEnv(Env* base, bool time_elapse_only_sleep)
33 : EnvWrapper(base),
34 maybe_starting_time_(MaybeCurrentTime(base)),
35 rnd_(301),
36 sleep_counter_(this),
37 time_elapse_only_sleep_(time_elapse_only_sleep),
38 no_slowdown_(time_elapse_only_sleep) {
39 delay_sstable_sync_.store(false, std::memory_order_release);
40 drop_writes_.store(false, std::memory_order_release);
41 no_space_.store(false, std::memory_order_release);
42 non_writable_.store(false, std::memory_order_release);
43 count_random_reads_ = false;
44 count_sequential_reads_ = false;
45 manifest_sync_error_.store(false, std::memory_order_release);
46 manifest_write_error_.store(false, std::memory_order_release);
47 log_write_error_.store(false, std::memory_order_release);
48 no_file_overwrite_.store(false, std::memory_order_release);
49 random_file_open_counter_.store(0, std::memory_order_relaxed);
50 delete_count_.store(0, std::memory_order_relaxed);
51 num_open_wal_file_.store(0);
52 log_write_slowdown_ = 0;
53 bytes_written_ = 0;
54 sync_counter_ = 0;
55 non_writeable_rate_ = 0;
56 new_writable_count_ = 0;
57 non_writable_count_ = 0;
58 table_write_callback_ = nullptr;
59 }
DBTestBase(const std::string path,bool env_do_fsync)60 DBTestBase::DBTestBase(const std::string path, bool env_do_fsync)
61 : mem_env_(nullptr), encrypted_env_(nullptr), option_config_(kDefault) {
62 Env* base_env = Env::Default();
63 ConfigOptions config_options;
64 EXPECT_OK(test::CreateEnvFromSystem(config_options, &base_env, &env_guard_));
65 EXPECT_NE(nullptr, base_env);
66 if (getenv("MEM_ENV")) {
67 mem_env_ = MockEnv::Create(base_env, base_env->GetSystemClock());
68 }
69 #ifndef ROCKSDB_LITE
70 if (getenv("ENCRYPTED_ENV")) {
71 std::shared_ptr<EncryptionProvider> provider;
72 std::string provider_id = getenv("ENCRYPTED_ENV");
73 if (provider_id.find("=") == std::string::npos &&
74 !EndsWith(provider_id, "://test")) {
75 provider_id = provider_id + "://test";
76 }
77 EXPECT_OK(EncryptionProvider::CreateFromString(ConfigOptions(), provider_id,
78 &provider));
79 encrypted_env_ = NewEncryptedEnv(mem_env_ ? mem_env_ : base_env, provider);
80 }
81 #endif // !ROCKSDB_LITE
82 env_ = new SpecialEnv(encrypted_env_ ? encrypted_env_
83 : (mem_env_ ? mem_env_ : base_env));
84 env_->SetBackgroundThreads(1, Env::LOW);
85 env_->SetBackgroundThreads(1, Env::HIGH);
86 env_->skip_fsync_ = !env_do_fsync;
87 dbname_ = test::PerThreadDBPath(env_, path);
88 alternative_wal_dir_ = dbname_ + "/wal";
89 alternative_db_log_dir_ = dbname_ + "/db_log_dir";
90 auto options = CurrentOptions();
91 options.env = env_;
92 auto delete_options = options;
93 delete_options.wal_dir = alternative_wal_dir_;
94 EXPECT_OK(DestroyDB(dbname_, delete_options));
95 // Destroy it for not alternative WAL dir is used.
96 EXPECT_OK(DestroyDB(dbname_, options));
97 db_ = nullptr;
98 Reopen(options);
99 Random::GetTLSInstance()->Reset(0xdeadbeef);
100 }
101
~DBTestBase()102 DBTestBase::~DBTestBase() {
103 #ifndef NDEBUG
104 ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->DisableProcessing();
105 ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->LoadDependency({});
106 ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearAllCallBacks();
107 #endif
108 Close();
109 Options options;
110 options.db_paths.emplace_back(dbname_, 0);
111 options.db_paths.emplace_back(dbname_ + "_2", 0);
112 options.db_paths.emplace_back(dbname_ + "_3", 0);
113 options.db_paths.emplace_back(dbname_ + "_4", 0);
114 options.env = env_;
115
116 if (getenv("KEEP_DB")) {
117 printf("DB is still at %s\n", dbname_.c_str());
118 } else {
119 EXPECT_OK(DestroyDB(dbname_, options));
120 }
121 delete env_;
122 }
123
ShouldSkipOptions(int option_config,int skip_mask)124 bool DBTestBase::ShouldSkipOptions(int option_config, int skip_mask) {
125 #ifdef ROCKSDB_LITE
126 // These options are not supported in ROCKSDB_LITE
127 if (option_config == kHashSkipList ||
128 option_config == kPlainTableFirstBytePrefix ||
129 option_config == kPlainTableCappedPrefix ||
130 option_config == kPlainTableCappedPrefixNonMmap ||
131 option_config == kPlainTableAllBytesPrefix ||
132 option_config == kVectorRep || option_config == kHashLinkList ||
133 option_config == kUniversalCompaction ||
134 option_config == kUniversalCompactionMultiLevel ||
135 option_config == kUniversalSubcompactions ||
136 option_config == kFIFOCompaction ||
137 option_config == kConcurrentSkipList) {
138 return true;
139 }
140 #endif
141
142 if ((skip_mask & kSkipUniversalCompaction) &&
143 (option_config == kUniversalCompaction ||
144 option_config == kUniversalCompactionMultiLevel ||
145 option_config == kUniversalSubcompactions)) {
146 return true;
147 }
148 if ((skip_mask & kSkipMergePut) && option_config == kMergePut) {
149 return true;
150 }
151 if ((skip_mask & kSkipNoSeekToLast) &&
152 (option_config == kHashLinkList || option_config == kHashSkipList)) {
153 return true;
154 }
155 if ((skip_mask & kSkipPlainTable) &&
156 (option_config == kPlainTableAllBytesPrefix ||
157 option_config == kPlainTableFirstBytePrefix ||
158 option_config == kPlainTableCappedPrefix ||
159 option_config == kPlainTableCappedPrefixNonMmap)) {
160 return true;
161 }
162 if ((skip_mask & kSkipHashIndex) &&
163 (option_config == kBlockBasedTableWithPrefixHashIndex ||
164 option_config == kBlockBasedTableWithWholeKeyHashIndex)) {
165 return true;
166 }
167 if ((skip_mask & kSkipFIFOCompaction) && option_config == kFIFOCompaction) {
168 return true;
169 }
170 if ((skip_mask & kSkipMmapReads) && option_config == kWalDirAndMmapReads) {
171 return true;
172 }
173 return false;
174 }
175
176 // Switch to a fresh database with the next option configuration to
177 // test. Return false if there are no more configurations to test.
ChangeOptions(int skip_mask)178 bool DBTestBase::ChangeOptions(int skip_mask) {
179 for (option_config_++; option_config_ < kEnd; option_config_++) {
180 if (ShouldSkipOptions(option_config_, skip_mask)) {
181 continue;
182 }
183 break;
184 }
185
186 if (option_config_ >= kEnd) {
187 Destroy(last_options_);
188 return false;
189 } else {
190 auto options = CurrentOptions();
191 options.create_if_missing = true;
192 DestroyAndReopen(options);
193 return true;
194 }
195 }
196
197 // Switch between different compaction styles.
ChangeCompactOptions()198 bool DBTestBase::ChangeCompactOptions() {
199 if (option_config_ == kDefault) {
200 option_config_ = kUniversalCompaction;
201 Destroy(last_options_);
202 auto options = CurrentOptions();
203 options.create_if_missing = true;
204 Reopen(options);
205 return true;
206 } else if (option_config_ == kUniversalCompaction) {
207 option_config_ = kUniversalCompactionMultiLevel;
208 Destroy(last_options_);
209 auto options = CurrentOptions();
210 options.create_if_missing = true;
211 Reopen(options);
212 return true;
213 } else if (option_config_ == kUniversalCompactionMultiLevel) {
214 option_config_ = kLevelSubcompactions;
215 Destroy(last_options_);
216 auto options = CurrentOptions();
217 assert(options.max_subcompactions > 1);
218 Reopen(options);
219 return true;
220 } else if (option_config_ == kLevelSubcompactions) {
221 option_config_ = kUniversalSubcompactions;
222 Destroy(last_options_);
223 auto options = CurrentOptions();
224 assert(options.max_subcompactions > 1);
225 Reopen(options);
226 return true;
227 } else {
228 return false;
229 }
230 }
231
232 // Switch between different WAL settings
ChangeWalOptions()233 bool DBTestBase::ChangeWalOptions() {
234 if (option_config_ == kDefault) {
235 option_config_ = kDBLogDir;
236 Destroy(last_options_);
237 auto options = CurrentOptions();
238 Destroy(options);
239 options.create_if_missing = true;
240 Reopen(options);
241 return true;
242 } else if (option_config_ == kDBLogDir) {
243 option_config_ = kWalDirAndMmapReads;
244 Destroy(last_options_);
245 auto options = CurrentOptions();
246 Destroy(options);
247 options.create_if_missing = true;
248 Reopen(options);
249 return true;
250 } else if (option_config_ == kWalDirAndMmapReads) {
251 option_config_ = kRecycleLogFiles;
252 Destroy(last_options_);
253 auto options = CurrentOptions();
254 Destroy(options);
255 Reopen(options);
256 return true;
257 } else {
258 return false;
259 }
260 }
261
262 // Switch between different filter policy
263 // Jump from kDefault to kFilter to kFullFilter
ChangeFilterOptions()264 bool DBTestBase::ChangeFilterOptions() {
265 if (option_config_ == kDefault) {
266 option_config_ = kFilter;
267 } else if (option_config_ == kFilter) {
268 option_config_ = kFullFilterWithNewTableReaderForCompactions;
269 } else if (option_config_ == kFullFilterWithNewTableReaderForCompactions) {
270 option_config_ = kPartitionedFilterWithNewTableReaderForCompactions;
271 } else {
272 return false;
273 }
274 Destroy(last_options_);
275
276 auto options = CurrentOptions();
277 options.create_if_missing = true;
278 TryReopen(options);
279 return true;
280 }
281
282 // Switch between different DB options for file ingestion tests.
ChangeOptionsForFileIngestionTest()283 bool DBTestBase::ChangeOptionsForFileIngestionTest() {
284 if (option_config_ == kDefault) {
285 option_config_ = kUniversalCompaction;
286 Destroy(last_options_);
287 auto options = CurrentOptions();
288 options.create_if_missing = true;
289 TryReopen(options);
290 return true;
291 } else if (option_config_ == kUniversalCompaction) {
292 option_config_ = kUniversalCompactionMultiLevel;
293 Destroy(last_options_);
294 auto options = CurrentOptions();
295 options.create_if_missing = true;
296 TryReopen(options);
297 return true;
298 } else if (option_config_ == kUniversalCompactionMultiLevel) {
299 option_config_ = kLevelSubcompactions;
300 Destroy(last_options_);
301 auto options = CurrentOptions();
302 assert(options.max_subcompactions > 1);
303 TryReopen(options);
304 return true;
305 } else if (option_config_ == kLevelSubcompactions) {
306 option_config_ = kUniversalSubcompactions;
307 Destroy(last_options_);
308 auto options = CurrentOptions();
309 assert(options.max_subcompactions > 1);
310 TryReopen(options);
311 return true;
312 } else if (option_config_ == kUniversalSubcompactions) {
313 option_config_ = kDirectIO;
314 Destroy(last_options_);
315 auto options = CurrentOptions();
316 TryReopen(options);
317 return true;
318 } else {
319 return false;
320 }
321 }
322
323 // Return the current option configuration.
CurrentOptions(const anon::OptionsOverride & options_override) const324 Options DBTestBase::CurrentOptions(
325 const anon::OptionsOverride& options_override) const {
326 return GetOptions(option_config_, GetDefaultOptions(), options_override);
327 }
328
CurrentOptions(const Options & default_options,const anon::OptionsOverride & options_override) const329 Options DBTestBase::CurrentOptions(
330 const Options& default_options,
331 const anon::OptionsOverride& options_override) const {
332 return GetOptions(option_config_, default_options, options_override);
333 }
334
GetDefaultOptions() const335 Options DBTestBase::GetDefaultOptions() const {
336 Options options;
337 options.write_buffer_size = 4090 * 4096;
338 options.target_file_size_base = 2 * 1024 * 1024;
339 options.max_bytes_for_level_base = 10 * 1024 * 1024;
340 options.max_open_files = 5000;
341 options.wal_recovery_mode = WALRecoveryMode::kTolerateCorruptedTailRecords;
342 options.compaction_pri = CompactionPri::kByCompensatedSize;
343 options.env = env_;
344 if (!env_->skip_fsync_) {
345 options.track_and_verify_wals_in_manifest = true;
346 }
347 return options;
348 }
349
GetOptions(int option_config,const Options & default_options,const anon::OptionsOverride & options_override) const350 Options DBTestBase::GetOptions(
351 int option_config, const Options& default_options,
352 const anon::OptionsOverride& options_override) const {
353 // this redundant copy is to minimize code change w/o having lint error.
354 Options options = default_options;
355 BlockBasedTableOptions table_options;
356 bool set_block_based_table_factory = true;
357 #ifndef NDEBUG
358 #if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && \
359 !defined(OS_AIX)
360 ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearCallBack(
361 "NewRandomAccessFile:O_DIRECT");
362 ROCKSDB_NAMESPACE::SyncPoint::GetInstance()->ClearCallBack(
363 "NewWritableFile:O_DIRECT");
364 #endif
365 #endif
366
367 bool can_allow_mmap = IsMemoryMappedAccessSupported();
368 switch (option_config) {
369 #ifndef ROCKSDB_LITE
370 case kHashSkipList:
371 options.prefix_extractor.reset(NewFixedPrefixTransform(1));
372 options.memtable_factory.reset(NewHashSkipListRepFactory(16));
373 options.allow_concurrent_memtable_write = false;
374 options.unordered_write = false;
375 break;
376 case kPlainTableFirstBytePrefix:
377 options.table_factory.reset(NewPlainTableFactory());
378 options.prefix_extractor.reset(NewFixedPrefixTransform(1));
379 options.allow_mmap_reads = can_allow_mmap;
380 options.max_sequential_skip_in_iterations = 999999;
381 set_block_based_table_factory = false;
382 break;
383 case kPlainTableCappedPrefix:
384 options.table_factory.reset(NewPlainTableFactory());
385 options.prefix_extractor.reset(NewCappedPrefixTransform(8));
386 options.allow_mmap_reads = can_allow_mmap;
387 options.max_sequential_skip_in_iterations = 999999;
388 set_block_based_table_factory = false;
389 break;
390 case kPlainTableCappedPrefixNonMmap:
391 options.table_factory.reset(NewPlainTableFactory());
392 options.prefix_extractor.reset(NewCappedPrefixTransform(8));
393 options.allow_mmap_reads = false;
394 options.max_sequential_skip_in_iterations = 999999;
395 set_block_based_table_factory = false;
396 break;
397 case kPlainTableAllBytesPrefix:
398 options.table_factory.reset(NewPlainTableFactory());
399 options.prefix_extractor.reset(NewNoopTransform());
400 options.allow_mmap_reads = can_allow_mmap;
401 options.max_sequential_skip_in_iterations = 999999;
402 set_block_based_table_factory = false;
403 break;
404 case kVectorRep:
405 options.memtable_factory.reset(new VectorRepFactory(100));
406 options.allow_concurrent_memtable_write = false;
407 options.unordered_write = false;
408 break;
409 case kHashLinkList:
410 options.prefix_extractor.reset(NewFixedPrefixTransform(1));
411 options.memtable_factory.reset(
412 NewHashLinkListRepFactory(4, 0, 3, true, 4));
413 options.allow_concurrent_memtable_write = false;
414 options.unordered_write = false;
415 break;
416 case kDirectIO: {
417 options.use_direct_reads = true;
418 options.use_direct_io_for_flush_and_compaction = true;
419 options.compaction_readahead_size = 2 * 1024 * 1024;
420 #ifndef NDEBUG
421 SetupSyncPointsToMockDirectIO();
422 #endif
423 break;
424 }
425 #endif // ROCKSDB_LITE
426 case kMergePut:
427 options.merge_operator = MergeOperators::CreatePutOperator();
428 break;
429 case kFilter:
430 table_options.filter_policy.reset(NewBloomFilterPolicy(10, true));
431 break;
432 case kFullFilterWithNewTableReaderForCompactions:
433 table_options.filter_policy.reset(NewBloomFilterPolicy(10, false));
434 options.new_table_reader_for_compaction_inputs = true;
435 options.compaction_readahead_size = 10 * 1024 * 1024;
436 break;
437 case kPartitionedFilterWithNewTableReaderForCompactions:
438 table_options.filter_policy.reset(NewBloomFilterPolicy(10, false));
439 table_options.partition_filters = true;
440 table_options.index_type =
441 BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch;
442 options.new_table_reader_for_compaction_inputs = true;
443 options.compaction_readahead_size = 10 * 1024 * 1024;
444 break;
445 case kUncompressed:
446 options.compression = kNoCompression;
447 break;
448 case kNumLevel_3:
449 options.num_levels = 3;
450 break;
451 case kDBLogDir:
452 options.db_log_dir = alternative_db_log_dir_;
453 break;
454 case kWalDirAndMmapReads:
455 options.wal_dir = alternative_wal_dir_;
456 // mmap reads should be orthogonal to WalDir setting, so we piggyback to
457 // this option config to test mmap reads as well
458 options.allow_mmap_reads = can_allow_mmap;
459 break;
460 case kManifestFileSize:
461 options.max_manifest_file_size = 50; // 50 bytes
462 break;
463 case kPerfOptions:
464 options.soft_rate_limit = 2.0;
465 options.delayed_write_rate = 8 * 1024 * 1024;
466 options.report_bg_io_stats = true;
467 // TODO(3.13) -- test more options
468 break;
469 case kUniversalCompaction:
470 options.compaction_style = kCompactionStyleUniversal;
471 options.num_levels = 1;
472 break;
473 case kUniversalCompactionMultiLevel:
474 options.compaction_style = kCompactionStyleUniversal;
475 options.num_levels = 8;
476 break;
477 case kCompressedBlockCache:
478 options.allow_mmap_writes = can_allow_mmap;
479 table_options.block_cache_compressed = NewLRUCache(8 * 1024 * 1024);
480 break;
481 case kInfiniteMaxOpenFiles:
482 options.max_open_files = -1;
483 break;
484 case kxxHashChecksum: {
485 table_options.checksum = kxxHash;
486 break;
487 }
488 case kxxHash64Checksum: {
489 table_options.checksum = kxxHash64;
490 break;
491 }
492 case kFIFOCompaction: {
493 options.compaction_style = kCompactionStyleFIFO;
494 options.max_open_files = -1;
495 break;
496 }
497 case kBlockBasedTableWithPrefixHashIndex: {
498 table_options.index_type = BlockBasedTableOptions::kHashSearch;
499 options.prefix_extractor.reset(NewFixedPrefixTransform(1));
500 break;
501 }
502 case kBlockBasedTableWithWholeKeyHashIndex: {
503 table_options.index_type = BlockBasedTableOptions::kHashSearch;
504 options.prefix_extractor.reset(NewNoopTransform());
505 break;
506 }
507 case kBlockBasedTableWithPartitionedIndex: {
508 table_options.format_version = 3;
509 table_options.index_type = BlockBasedTableOptions::kTwoLevelIndexSearch;
510 options.prefix_extractor.reset(NewNoopTransform());
511 break;
512 }
513 case kBlockBasedTableWithPartitionedIndexFormat4: {
514 table_options.format_version = 4;
515 // Format 4 changes the binary index format. Since partitioned index is a
516 // super-set of simple indexes, we are also using kTwoLevelIndexSearch to
517 // test this format.
518 table_options.index_type = BlockBasedTableOptions::kTwoLevelIndexSearch;
519 // The top-level index in partition filters are also affected by format 4.
520 table_options.filter_policy.reset(NewBloomFilterPolicy(10, false));
521 table_options.partition_filters = true;
522 table_options.index_block_restart_interval = 8;
523 break;
524 }
525 case kBlockBasedTableWithIndexRestartInterval: {
526 table_options.index_block_restart_interval = 8;
527 break;
528 }
529 case kOptimizeFiltersForHits: {
530 options.optimize_filters_for_hits = true;
531 set_block_based_table_factory = true;
532 break;
533 }
534 case kRowCache: {
535 options.row_cache = NewLRUCache(1024 * 1024);
536 break;
537 }
538 case kRecycleLogFiles: {
539 options.recycle_log_file_num = 2;
540 break;
541 }
542 case kLevelSubcompactions: {
543 options.max_subcompactions = 4;
544 break;
545 }
546 case kUniversalSubcompactions: {
547 options.compaction_style = kCompactionStyleUniversal;
548 options.num_levels = 8;
549 options.max_subcompactions = 4;
550 break;
551 }
552 case kConcurrentSkipList: {
553 options.allow_concurrent_memtable_write = true;
554 options.enable_write_thread_adaptive_yield = true;
555 break;
556 }
557 case kPipelinedWrite: {
558 options.enable_pipelined_write = true;
559 break;
560 }
561 case kConcurrentWALWrites: {
562 // This options optimize 2PC commit path
563 options.two_write_queues = true;
564 options.manual_wal_flush = true;
565 break;
566 }
567 case kUnorderedWrite: {
568 options.allow_concurrent_memtable_write = false;
569 options.unordered_write = false;
570 break;
571 }
572
573 default:
574 break;
575 }
576
577 if (options_override.filter_policy) {
578 table_options.filter_policy = options_override.filter_policy;
579 table_options.partition_filters = options_override.partition_filters;
580 table_options.metadata_block_size = options_override.metadata_block_size;
581 }
582 if (set_block_based_table_factory) {
583 options.table_factory.reset(NewBlockBasedTableFactory(table_options));
584 }
585 options.env = env_;
586 options.create_if_missing = true;
587 options.fail_if_options_file_error = true;
588 return options;
589 }
590
CreateColumnFamilies(const std::vector<std::string> & cfs,const Options & options)591 void DBTestBase::CreateColumnFamilies(const std::vector<std::string>& cfs,
592 const Options& options) {
593 ColumnFamilyOptions cf_opts(options);
594 size_t cfi = handles_.size();
595 handles_.resize(cfi + cfs.size());
596 for (auto cf : cfs) {
597 Status s = db_->CreateColumnFamily(cf_opts, cf, &handles_[cfi++]);
598 ASSERT_OK(s);
599 }
600 }
601
CreateAndReopenWithCF(const std::vector<std::string> & cfs,const Options & options)602 void DBTestBase::CreateAndReopenWithCF(const std::vector<std::string>& cfs,
603 const Options& options) {
604 CreateColumnFamilies(cfs, options);
605 std::vector<std::string> cfs_plus_default = cfs;
606 cfs_plus_default.insert(cfs_plus_default.begin(), kDefaultColumnFamilyName);
607 ReopenWithColumnFamilies(cfs_plus_default, options);
608 }
609
ReopenWithColumnFamilies(const std::vector<std::string> & cfs,const std::vector<Options> & options)610 void DBTestBase::ReopenWithColumnFamilies(const std::vector<std::string>& cfs,
611 const std::vector<Options>& options) {
612 ASSERT_OK(TryReopenWithColumnFamilies(cfs, options));
613 }
614
ReopenWithColumnFamilies(const std::vector<std::string> & cfs,const Options & options)615 void DBTestBase::ReopenWithColumnFamilies(const std::vector<std::string>& cfs,
616 const Options& options) {
617 ASSERT_OK(TryReopenWithColumnFamilies(cfs, options));
618 }
619
SetTimeElapseOnlySleepOnReopen(DBOptions * options)620 void DBTestBase::SetTimeElapseOnlySleepOnReopen(DBOptions* options) {
621 time_elapse_only_sleep_on_reopen_ = true;
622
623 // Need to disable stats dumping and persisting which also use
624 // RepeatableThread, which uses InstrumentedCondVar::TimedWaitInternal.
625 // With time_elapse_only_sleep_, this can hang on some platforms (MacOS)
626 // because (a) on some platforms, pthread_cond_timedwait does not appear
627 // to release the lock for other threads to operate if the deadline time
628 // is already passed, and (b) TimedWait calls are currently a bad abstraction
629 // because the deadline parameter is usually computed from Env time,
630 // but is interpreted in real clock time.
631 options->stats_dump_period_sec = 0;
632 options->stats_persist_period_sec = 0;
633 }
634
MaybeInstallTimeElapseOnlySleep(const DBOptions & options)635 void DBTestBase::MaybeInstallTimeElapseOnlySleep(const DBOptions& options) {
636 if (time_elapse_only_sleep_on_reopen_) {
637 assert(options.env == env_ ||
638 static_cast_with_check<CompositeEnvWrapper>(options.env)
639 ->env_target() == env_);
640 assert(options.stats_dump_period_sec == 0);
641 assert(options.stats_persist_period_sec == 0);
642 // We cannot set these before destroying the last DB because they might
643 // cause a deadlock or similar without the appropriate options set in
644 // the DB.
645 env_->time_elapse_only_sleep_ = true;
646 env_->no_slowdown_ = true;
647 } else {
648 // Going back in same test run is not yet supported, so no
649 // reset in this case.
650 }
651 }
652
TryReopenWithColumnFamilies(const std::vector<std::string> & cfs,const std::vector<Options> & options)653 Status DBTestBase::TryReopenWithColumnFamilies(
654 const std::vector<std::string>& cfs, const std::vector<Options>& options) {
655 Close();
656 EXPECT_EQ(cfs.size(), options.size());
657 std::vector<ColumnFamilyDescriptor> column_families;
658 for (size_t i = 0; i < cfs.size(); ++i) {
659 column_families.push_back(ColumnFamilyDescriptor(cfs[i], options[i]));
660 }
661 DBOptions db_opts = DBOptions(options[0]);
662 last_options_ = options[0];
663 MaybeInstallTimeElapseOnlySleep(db_opts);
664 return DB::Open(db_opts, dbname_, column_families, &handles_, &db_);
665 }
666
TryReopenWithColumnFamilies(const std::vector<std::string> & cfs,const Options & options)667 Status DBTestBase::TryReopenWithColumnFamilies(
668 const std::vector<std::string>& cfs, const Options& options) {
669 Close();
670 std::vector<Options> v_opts(cfs.size(), options);
671 return TryReopenWithColumnFamilies(cfs, v_opts);
672 }
673
Reopen(const Options & options)674 void DBTestBase::Reopen(const Options& options) {
675 ASSERT_OK(TryReopen(options));
676 }
677
Close()678 void DBTestBase::Close() {
679 for (auto h : handles_) {
680 EXPECT_OK(db_->DestroyColumnFamilyHandle(h));
681 }
682 handles_.clear();
683 delete db_;
684 db_ = nullptr;
685 }
686
DestroyAndReopen(const Options & options)687 void DBTestBase::DestroyAndReopen(const Options& options) {
688 // Destroy using last options
689 Destroy(last_options_);
690 Reopen(options);
691 }
692
Destroy(const Options & options,bool delete_cf_paths)693 void DBTestBase::Destroy(const Options& options, bool delete_cf_paths) {
694 std::vector<ColumnFamilyDescriptor> column_families;
695 if (delete_cf_paths) {
696 for (size_t i = 0; i < handles_.size(); ++i) {
697 ColumnFamilyDescriptor cfdescriptor;
698 // GetDescriptor is not implemented for ROCKSDB_LITE
699 handles_[i]->GetDescriptor(&cfdescriptor).PermitUncheckedError();
700 column_families.push_back(cfdescriptor);
701 }
702 }
703 Close();
704 ASSERT_OK(DestroyDB(dbname_, options, column_families));
705 }
706
ReadOnlyReopen(const Options & options)707 Status DBTestBase::ReadOnlyReopen(const Options& options) {
708 MaybeInstallTimeElapseOnlySleep(options);
709 return DB::OpenForReadOnly(options, dbname_, &db_);
710 }
711
TryReopen(const Options & options)712 Status DBTestBase::TryReopen(const Options& options) {
713 Close();
714 last_options_.table_factory.reset();
715 // Note: operator= is an unsafe approach here since it destructs
716 // std::shared_ptr in the same order of their creation, in contrast to
717 // destructors which destructs them in the opposite order of creation. One
718 // particular problem is that the cache destructor might invoke callback
719 // functions that use Option members such as statistics. To work around this
720 // problem, we manually call destructor of table_factory which eventually
721 // clears the block cache.
722 last_options_ = options;
723 MaybeInstallTimeElapseOnlySleep(options);
724 return DB::Open(options, dbname_, &db_);
725 }
726
IsDirectIOSupported()727 bool DBTestBase::IsDirectIOSupported() {
728 return test::IsDirectIOSupported(env_, dbname_);
729 }
730
IsMemoryMappedAccessSupported() const731 bool DBTestBase::IsMemoryMappedAccessSupported() const {
732 return (!encrypted_env_);
733 }
734
Flush(int cf)735 Status DBTestBase::Flush(int cf) {
736 if (cf == 0) {
737 return db_->Flush(FlushOptions());
738 } else {
739 return db_->Flush(FlushOptions(), handles_[cf]);
740 }
741 }
742
Flush(const std::vector<int> & cf_ids)743 Status DBTestBase::Flush(const std::vector<int>& cf_ids) {
744 std::vector<ColumnFamilyHandle*> cfhs;
745 std::for_each(cf_ids.begin(), cf_ids.end(),
746 [&cfhs, this](int id) { cfhs.emplace_back(handles_[id]); });
747 return db_->Flush(FlushOptions(), cfhs);
748 }
749
Put(const Slice & k,const Slice & v,WriteOptions wo)750 Status DBTestBase::Put(const Slice& k, const Slice& v, WriteOptions wo) {
751 if (kMergePut == option_config_) {
752 return db_->Merge(wo, k, v);
753 } else {
754 return db_->Put(wo, k, v);
755 }
756 }
757
Put(int cf,const Slice & k,const Slice & v,WriteOptions wo)758 Status DBTestBase::Put(int cf, const Slice& k, const Slice& v,
759 WriteOptions wo) {
760 if (kMergePut == option_config_) {
761 return db_->Merge(wo, handles_[cf], k, v);
762 } else {
763 return db_->Put(wo, handles_[cf], k, v);
764 }
765 }
766
Merge(const Slice & k,const Slice & v,WriteOptions wo)767 Status DBTestBase::Merge(const Slice& k, const Slice& v, WriteOptions wo) {
768 return db_->Merge(wo, k, v);
769 }
770
Merge(int cf,const Slice & k,const Slice & v,WriteOptions wo)771 Status DBTestBase::Merge(int cf, const Slice& k, const Slice& v,
772 WriteOptions wo) {
773 return db_->Merge(wo, handles_[cf], k, v);
774 }
775
Delete(const std::string & k)776 Status DBTestBase::Delete(const std::string& k) {
777 return db_->Delete(WriteOptions(), k);
778 }
779
Delete(int cf,const std::string & k)780 Status DBTestBase::Delete(int cf, const std::string& k) {
781 return db_->Delete(WriteOptions(), handles_[cf], k);
782 }
783
SingleDelete(const std::string & k)784 Status DBTestBase::SingleDelete(const std::string& k) {
785 return db_->SingleDelete(WriteOptions(), k);
786 }
787
SingleDelete(int cf,const std::string & k)788 Status DBTestBase::SingleDelete(int cf, const std::string& k) {
789 return db_->SingleDelete(WriteOptions(), handles_[cf], k);
790 }
791
SetPreserveDeletesSequenceNumber(SequenceNumber sn)792 bool DBTestBase::SetPreserveDeletesSequenceNumber(SequenceNumber sn) {
793 return db_->SetPreserveDeletesSequenceNumber(sn);
794 }
795
Get(const std::string & k,const Snapshot * snapshot)796 std::string DBTestBase::Get(const std::string& k, const Snapshot* snapshot) {
797 ReadOptions options;
798 options.verify_checksums = true;
799 options.snapshot = snapshot;
800 std::string result;
801 Status s = db_->Get(options, k, &result);
802 if (s.IsNotFound()) {
803 result = "NOT_FOUND";
804 } else if (!s.ok()) {
805 result = s.ToString();
806 }
807 return result;
808 }
809
Get(int cf,const std::string & k,const Snapshot * snapshot)810 std::string DBTestBase::Get(int cf, const std::string& k,
811 const Snapshot* snapshot) {
812 ReadOptions options;
813 options.verify_checksums = true;
814 options.snapshot = snapshot;
815 std::string result;
816 Status s = db_->Get(options, handles_[cf], k, &result);
817 if (s.IsNotFound()) {
818 result = "NOT_FOUND";
819 } else if (!s.ok()) {
820 result = s.ToString();
821 }
822 return result;
823 }
824
MultiGet(std::vector<int> cfs,const std::vector<std::string> & k,const Snapshot * snapshot,const bool batched)825 std::vector<std::string> DBTestBase::MultiGet(std::vector<int> cfs,
826 const std::vector<std::string>& k,
827 const Snapshot* snapshot,
828 const bool batched) {
829 ReadOptions options;
830 options.verify_checksums = true;
831 options.snapshot = snapshot;
832 std::vector<ColumnFamilyHandle*> handles;
833 std::vector<Slice> keys;
834 std::vector<std::string> result;
835
836 for (unsigned int i = 0; i < cfs.size(); ++i) {
837 handles.push_back(handles_[cfs[i]]);
838 keys.push_back(k[i]);
839 }
840 std::vector<Status> s;
841 if (!batched) {
842 s = db_->MultiGet(options, handles, keys, &result);
843 for (unsigned int i = 0; i < s.size(); ++i) {
844 if (s[i].IsNotFound()) {
845 result[i] = "NOT_FOUND";
846 } else if (!s[i].ok()) {
847 result[i] = s[i].ToString();
848 }
849 }
850 } else {
851 std::vector<PinnableSlice> pin_values(cfs.size());
852 result.resize(cfs.size());
853 s.resize(cfs.size());
854 db_->MultiGet(options, cfs.size(), handles.data(), keys.data(),
855 pin_values.data(), s.data());
856 for (unsigned int i = 0; i < s.size(); ++i) {
857 if (s[i].IsNotFound()) {
858 result[i] = "NOT_FOUND";
859 } else if (!s[i].ok()) {
860 result[i] = s[i].ToString();
861 } else {
862 result[i].assign(pin_values[i].data(), pin_values[i].size());
863 }
864 }
865 }
866 return result;
867 }
868
MultiGet(const std::vector<std::string> & k,const Snapshot * snapshot)869 std::vector<std::string> DBTestBase::MultiGet(const std::vector<std::string>& k,
870 const Snapshot* snapshot) {
871 ReadOptions options;
872 options.verify_checksums = true;
873 options.snapshot = snapshot;
874 std::vector<Slice> keys;
875 std::vector<std::string> result;
876 std::vector<Status> statuses(k.size());
877 std::vector<PinnableSlice> pin_values(k.size());
878
879 for (unsigned int i = 0; i < k.size(); ++i) {
880 keys.push_back(k[i]);
881 }
882 db_->MultiGet(options, dbfull()->DefaultColumnFamily(), keys.size(),
883 keys.data(), pin_values.data(), statuses.data());
884 result.resize(k.size());
885 for (auto iter = result.begin(); iter != result.end(); ++iter) {
886 iter->assign(pin_values[iter - result.begin()].data(),
887 pin_values[iter - result.begin()].size());
888 }
889 for (unsigned int i = 0; i < statuses.size(); ++i) {
890 if (statuses[i].IsNotFound()) {
891 result[i] = "NOT_FOUND";
892 }
893 }
894 return result;
895 }
896
Get(const std::string & k,PinnableSlice * v)897 Status DBTestBase::Get(const std::string& k, PinnableSlice* v) {
898 ReadOptions options;
899 options.verify_checksums = true;
900 Status s = dbfull()->Get(options, dbfull()->DefaultColumnFamily(), k, v);
901 return s;
902 }
903
GetNumSnapshots()904 uint64_t DBTestBase::GetNumSnapshots() {
905 uint64_t int_num;
906 EXPECT_TRUE(dbfull()->GetIntProperty("rocksdb.num-snapshots", &int_num));
907 return int_num;
908 }
909
GetTimeOldestSnapshots()910 uint64_t DBTestBase::GetTimeOldestSnapshots() {
911 uint64_t int_num;
912 EXPECT_TRUE(
913 dbfull()->GetIntProperty("rocksdb.oldest-snapshot-time", &int_num));
914 return int_num;
915 }
916
GetSequenceOldestSnapshots()917 uint64_t DBTestBase::GetSequenceOldestSnapshots() {
918 uint64_t int_num;
919 EXPECT_TRUE(
920 dbfull()->GetIntProperty("rocksdb.oldest-snapshot-sequence", &int_num));
921 return int_num;
922 }
923
924 // Return a string that contains all key,value pairs in order,
925 // formatted like "(k1->v1)(k2->v2)".
Contents(int cf)926 std::string DBTestBase::Contents(int cf) {
927 std::vector<std::string> forward;
928 std::string result;
929 Iterator* iter = (cf == 0) ? db_->NewIterator(ReadOptions())
930 : db_->NewIterator(ReadOptions(), handles_[cf]);
931 for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
932 std::string s = IterStatus(iter);
933 result.push_back('(');
934 result.append(s);
935 result.push_back(')');
936 forward.push_back(s);
937 }
938
939 // Check reverse iteration results are the reverse of forward results
940 unsigned int matched = 0;
941 for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
942 EXPECT_LT(matched, forward.size());
943 EXPECT_EQ(IterStatus(iter), forward[forward.size() - matched - 1]);
944 matched++;
945 }
946 EXPECT_EQ(matched, forward.size());
947
948 delete iter;
949 return result;
950 }
951
AllEntriesFor(const Slice & user_key,int cf)952 std::string DBTestBase::AllEntriesFor(const Slice& user_key, int cf) {
953 Arena arena;
954 auto options = CurrentOptions();
955 InternalKeyComparator icmp(options.comparator);
956 ReadRangeDelAggregator range_del_agg(&icmp,
957 kMaxSequenceNumber /* upper_bound */);
958 ReadOptions read_options;
959 ScopedArenaIterator iter;
960 if (cf == 0) {
961 iter.set(dbfull()->NewInternalIterator(read_options, &arena, &range_del_agg,
962 kMaxSequenceNumber));
963 } else {
964 iter.set(dbfull()->NewInternalIterator(read_options, &arena, &range_del_agg,
965 kMaxSequenceNumber, handles_[cf]));
966 }
967 InternalKey target(user_key, kMaxSequenceNumber, kTypeValue);
968 iter->Seek(target.Encode());
969 std::string result;
970 if (!iter->status().ok()) {
971 result = iter->status().ToString();
972 } else {
973 result = "[ ";
974 bool first = true;
975 while (iter->Valid()) {
976 ParsedInternalKey ikey(Slice(), 0, kTypeValue);
977 if (ParseInternalKey(iter->key(), &ikey, true /* log_err_key */) !=
978 Status::OK()) {
979 result += "CORRUPTED";
980 } else {
981 if (!last_options_.comparator->Equal(ikey.user_key, user_key)) {
982 break;
983 }
984 if (!first) {
985 result += ", ";
986 }
987 first = false;
988 switch (ikey.type) {
989 case kTypeValue:
990 result += iter->value().ToString();
991 break;
992 case kTypeMerge:
993 // keep it the same as kTypeValue for testing kMergePut
994 result += iter->value().ToString();
995 break;
996 case kTypeDeletion:
997 result += "DEL";
998 break;
999 case kTypeSingleDeletion:
1000 result += "SDEL";
1001 break;
1002 default:
1003 assert(false);
1004 break;
1005 }
1006 }
1007 iter->Next();
1008 }
1009 if (!first) {
1010 result += " ";
1011 }
1012 result += "]";
1013 }
1014 return result;
1015 }
1016
1017 #ifndef ROCKSDB_LITE
NumSortedRuns(int cf)1018 int DBTestBase::NumSortedRuns(int cf) {
1019 ColumnFamilyMetaData cf_meta;
1020 if (cf == 0) {
1021 db_->GetColumnFamilyMetaData(&cf_meta);
1022 } else {
1023 db_->GetColumnFamilyMetaData(handles_[cf], &cf_meta);
1024 }
1025 int num_sr = static_cast<int>(cf_meta.levels[0].files.size());
1026 for (size_t i = 1U; i < cf_meta.levels.size(); i++) {
1027 if (cf_meta.levels[i].files.size() > 0) {
1028 num_sr++;
1029 }
1030 }
1031 return num_sr;
1032 }
1033
TotalSize(int cf)1034 uint64_t DBTestBase::TotalSize(int cf) {
1035 ColumnFamilyMetaData cf_meta;
1036 if (cf == 0) {
1037 db_->GetColumnFamilyMetaData(&cf_meta);
1038 } else {
1039 db_->GetColumnFamilyMetaData(handles_[cf], &cf_meta);
1040 }
1041 return cf_meta.size;
1042 }
1043
SizeAtLevel(int level)1044 uint64_t DBTestBase::SizeAtLevel(int level) {
1045 std::vector<LiveFileMetaData> metadata;
1046 db_->GetLiveFilesMetaData(&metadata);
1047 uint64_t sum = 0;
1048 for (const auto& m : metadata) {
1049 if (m.level == level) {
1050 sum += m.size;
1051 }
1052 }
1053 return sum;
1054 }
1055
TotalLiveFiles(int cf)1056 size_t DBTestBase::TotalLiveFiles(int cf) {
1057 ColumnFamilyMetaData cf_meta;
1058 if (cf == 0) {
1059 db_->GetColumnFamilyMetaData(&cf_meta);
1060 } else {
1061 db_->GetColumnFamilyMetaData(handles_[cf], &cf_meta);
1062 }
1063 size_t num_files = 0;
1064 for (auto& level : cf_meta.levels) {
1065 num_files += level.files.size();
1066 }
1067 return num_files;
1068 }
1069
CountLiveFiles()1070 size_t DBTestBase::CountLiveFiles() {
1071 std::vector<LiveFileMetaData> metadata;
1072 db_->GetLiveFilesMetaData(&metadata);
1073 return metadata.size();
1074 }
1075
NumTableFilesAtLevel(int level,int cf)1076 int DBTestBase::NumTableFilesAtLevel(int level, int cf) {
1077 std::string property;
1078 if (cf == 0) {
1079 // default cfd
1080 EXPECT_TRUE(db_->GetProperty("rocksdb.num-files-at-level" + ToString(level),
1081 &property));
1082 } else {
1083 EXPECT_TRUE(db_->GetProperty(handles_[cf],
1084 "rocksdb.num-files-at-level" + ToString(level),
1085 &property));
1086 }
1087 return atoi(property.c_str());
1088 }
1089
CompressionRatioAtLevel(int level,int cf)1090 double DBTestBase::CompressionRatioAtLevel(int level, int cf) {
1091 std::string property;
1092 if (cf == 0) {
1093 // default cfd
1094 EXPECT_TRUE(db_->GetProperty(
1095 "rocksdb.compression-ratio-at-level" + ToString(level), &property));
1096 } else {
1097 EXPECT_TRUE(db_->GetProperty(
1098 handles_[cf], "rocksdb.compression-ratio-at-level" + ToString(level),
1099 &property));
1100 }
1101 return std::stod(property);
1102 }
1103
TotalTableFiles(int cf,int levels)1104 int DBTestBase::TotalTableFiles(int cf, int levels) {
1105 if (levels == -1) {
1106 levels = (cf == 0) ? db_->NumberLevels() : db_->NumberLevels(handles_[1]);
1107 }
1108 int result = 0;
1109 for (int level = 0; level < levels; level++) {
1110 result += NumTableFilesAtLevel(level, cf);
1111 }
1112 return result;
1113 }
1114
1115 // Return spread of files per level
FilesPerLevel(int cf)1116 std::string DBTestBase::FilesPerLevel(int cf) {
1117 int num_levels =
1118 (cf == 0) ? db_->NumberLevels() : db_->NumberLevels(handles_[1]);
1119 std::string result;
1120 size_t last_non_zero_offset = 0;
1121 for (int level = 0; level < num_levels; level++) {
1122 int f = NumTableFilesAtLevel(level, cf);
1123 char buf[100];
1124 snprintf(buf, sizeof(buf), "%s%d", (level ? "," : ""), f);
1125 result += buf;
1126 if (f > 0) {
1127 last_non_zero_offset = result.size();
1128 }
1129 }
1130 result.resize(last_non_zero_offset);
1131 return result;
1132 }
1133
1134 #endif // !ROCKSDB_LITE
1135
1136 #ifndef NDEBUG
GetBlobFileNumbers()1137 std::vector<uint64_t> DBTestBase::GetBlobFileNumbers() {
1138 VersionSet* const versions = dbfull()->TEST_GetVersionSet();
1139 assert(versions);
1140
1141 ColumnFamilyData* const cfd = versions->GetColumnFamilySet()->GetDefault();
1142 assert(cfd);
1143
1144 Version* const current = cfd->current();
1145 assert(current);
1146
1147 const VersionStorageInfo* const storage_info = current->storage_info();
1148 assert(storage_info);
1149
1150 const auto& blob_files = storage_info->GetBlobFiles();
1151
1152 std::vector<uint64_t> result;
1153 result.reserve(blob_files.size());
1154
1155 for (const auto& blob_file : blob_files) {
1156 result.emplace_back(blob_file.first);
1157 }
1158
1159 return result;
1160 }
1161 #endif
1162
CountFiles()1163 size_t DBTestBase::CountFiles() {
1164 size_t count = 0;
1165 std::vector<std::string> files;
1166 if (env_->GetChildren(dbname_, &files).ok()) {
1167 count += files.size();
1168 }
1169
1170 if (dbname_ != last_options_.wal_dir) {
1171 if (env_->GetChildren(last_options_.wal_dir, &files).ok()) {
1172 count += files.size();
1173 }
1174 }
1175
1176 return count;
1177 };
1178
CountFiles(size_t * count)1179 Status DBTestBase::CountFiles(size_t* count) {
1180 std::vector<std::string> files;
1181 Status s = env_->GetChildren(dbname_, &files);
1182 if (!s.ok()) {
1183 return s;
1184 }
1185 size_t files_count = files.size();
1186
1187 if (dbname_ != last_options_.wal_dir) {
1188 s = env_->GetChildren(last_options_.wal_dir, &files);
1189 if (!s.ok()) {
1190 return s;
1191 }
1192 *count = files_count + files.size();
1193 }
1194
1195 return Status::OK();
1196 }
1197
Size(const Slice & start,const Slice & limit,int cf,uint64_t * size)1198 Status DBTestBase::Size(const Slice& start, const Slice& limit, int cf,
1199 uint64_t* size) {
1200 Range r(start, limit);
1201 if (cf == 0) {
1202 return db_->GetApproximateSizes(&r, 1, size);
1203 } else {
1204 return db_->GetApproximateSizes(handles_[1], &r, 1, size);
1205 }
1206 }
1207
Compact(int cf,const Slice & start,const Slice & limit,uint32_t target_path_id)1208 void DBTestBase::Compact(int cf, const Slice& start, const Slice& limit,
1209 uint32_t target_path_id) {
1210 CompactRangeOptions compact_options;
1211 compact_options.target_path_id = target_path_id;
1212 ASSERT_OK(db_->CompactRange(compact_options, handles_[cf], &start, &limit));
1213 }
1214
Compact(int cf,const Slice & start,const Slice & limit)1215 void DBTestBase::Compact(int cf, const Slice& start, const Slice& limit) {
1216 ASSERT_OK(
1217 db_->CompactRange(CompactRangeOptions(), handles_[cf], &start, &limit));
1218 }
1219
Compact(const Slice & start,const Slice & limit)1220 void DBTestBase::Compact(const Slice& start, const Slice& limit) {
1221 ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &limit));
1222 }
1223
1224 // Do n memtable compactions, each of which produces an sstable
1225 // covering the range [small,large].
MakeTables(int n,const std::string & small,const std::string & large,int cf)1226 void DBTestBase::MakeTables(int n, const std::string& small,
1227 const std::string& large, int cf) {
1228 for (int i = 0; i < n; i++) {
1229 ASSERT_OK(Put(cf, small, "begin"));
1230 ASSERT_OK(Put(cf, large, "end"));
1231 ASSERT_OK(Flush(cf));
1232 MoveFilesToLevel(n - i - 1, cf);
1233 }
1234 }
1235
1236 // Prevent pushing of new sstables into deeper levels by adding
1237 // tables that cover a specified range to all levels.
FillLevels(const std::string & smallest,const std::string & largest,int cf)1238 void DBTestBase::FillLevels(const std::string& smallest,
1239 const std::string& largest, int cf) {
1240 MakeTables(db_->NumberLevels(handles_[cf]), smallest, largest, cf);
1241 }
1242
MoveFilesToLevel(int level,int cf)1243 void DBTestBase::MoveFilesToLevel(int level, int cf) {
1244 #ifndef NDEBUG
1245 for (int l = 0; l < level; ++l) {
1246 if (cf > 0) {
1247 EXPECT_OK(dbfull()->TEST_CompactRange(l, nullptr, nullptr, handles_[cf]));
1248 } else {
1249 EXPECT_OK(dbfull()->TEST_CompactRange(l, nullptr, nullptr));
1250 }
1251 }
1252 #endif
1253 }
1254
1255 #ifndef ROCKSDB_LITE
DumpFileCounts(const char * label)1256 void DBTestBase::DumpFileCounts(const char* label) {
1257 fprintf(stderr, "---\n%s:\n", label);
1258 #ifndef NDEBUG
1259 fprintf(stderr, "maxoverlap: %" PRIu64 "\n",
1260 dbfull()->TEST_MaxNextLevelOverlappingBytes());
1261 #endif
1262 for (int level = 0; level < db_->NumberLevels(); level++) {
1263 int num = NumTableFilesAtLevel(level);
1264 if (num > 0) {
1265 fprintf(stderr, " level %3d : %d files\n", level, num);
1266 }
1267 }
1268 }
1269 #endif // !ROCKSDB_LITE
1270
DumpSSTableList()1271 std::string DBTestBase::DumpSSTableList() {
1272 std::string property;
1273 db_->GetProperty("rocksdb.sstables", &property);
1274 return property;
1275 }
1276
GetSstFiles(Env * env,std::string path,std::vector<std::string> * files)1277 void DBTestBase::GetSstFiles(Env* env, std::string path,
1278 std::vector<std::string>* files) {
1279 EXPECT_OK(env->GetChildren(path, files));
1280
1281 files->erase(
1282 std::remove_if(files->begin(), files->end(), [](std::string name) {
1283 uint64_t number;
1284 FileType type;
1285 return !(ParseFileName(name, &number, &type) && type == kTableFile);
1286 }), files->end());
1287 }
1288
GetSstFileCount(std::string path)1289 int DBTestBase::GetSstFileCount(std::string path) {
1290 std::vector<std::string> files;
1291 DBTestBase::GetSstFiles(env_, path, &files);
1292 return static_cast<int>(files.size());
1293 }
1294
1295 // this will generate non-overlapping files since it keeps increasing key_idx
GenerateNewFile(int cf,Random * rnd,int * key_idx,bool nowait)1296 void DBTestBase::GenerateNewFile(int cf, Random* rnd, int* key_idx,
1297 bool nowait) {
1298 for (int i = 0; i < KNumKeysByGenerateNewFile; i++) {
1299 ASSERT_OK(Put(cf, Key(*key_idx), rnd->RandomString((i == 99) ? 1 : 990)));
1300 (*key_idx)++;
1301 }
1302 #ifndef NDEBUG
1303 if (!nowait) {
1304 ASSERT_OK(dbfull()->TEST_WaitForFlushMemTable());
1305 ASSERT_OK(dbfull()->TEST_WaitForCompact());
1306 }
1307 #endif
1308 }
1309
1310 // this will generate non-overlapping files since it keeps increasing key_idx
GenerateNewFile(Random * rnd,int * key_idx,bool nowait)1311 void DBTestBase::GenerateNewFile(Random* rnd, int* key_idx, bool nowait) {
1312 for (int i = 0; i < KNumKeysByGenerateNewFile; i++) {
1313 ASSERT_OK(Put(Key(*key_idx), rnd->RandomString((i == 99) ? 1 : 990)));
1314 (*key_idx)++;
1315 }
1316 #ifndef NDEBUG
1317 if (!nowait) {
1318 ASSERT_OK(dbfull()->TEST_WaitForFlushMemTable());
1319 ASSERT_OK(dbfull()->TEST_WaitForCompact());
1320 }
1321 #endif
1322 }
1323
1324 const int DBTestBase::kNumKeysByGenerateNewRandomFile = 51;
1325
GenerateNewRandomFile(Random * rnd,bool nowait)1326 void DBTestBase::GenerateNewRandomFile(Random* rnd, bool nowait) {
1327 for (int i = 0; i < kNumKeysByGenerateNewRandomFile; i++) {
1328 ASSERT_OK(Put("key" + rnd->RandomString(7), rnd->RandomString(2000)));
1329 }
1330 ASSERT_OK(Put("key" + rnd->RandomString(7), rnd->RandomString(200)));
1331 #ifndef NDEBUG
1332 if (!nowait) {
1333 ASSERT_OK(dbfull()->TEST_WaitForFlushMemTable());
1334 ASSERT_OK(dbfull()->TEST_WaitForCompact());
1335 }
1336 #endif
1337 }
1338
IterStatus(Iterator * iter)1339 std::string DBTestBase::IterStatus(Iterator* iter) {
1340 std::string result;
1341 if (iter->Valid()) {
1342 result = iter->key().ToString() + "->" + iter->value().ToString();
1343 } else {
1344 result = "(invalid)";
1345 }
1346 return result;
1347 }
1348
OptionsForLogIterTest()1349 Options DBTestBase::OptionsForLogIterTest() {
1350 Options options = CurrentOptions();
1351 options.create_if_missing = true;
1352 options.WAL_ttl_seconds = 1000;
1353 return options;
1354 }
1355
DummyString(size_t len,char c)1356 std::string DBTestBase::DummyString(size_t len, char c) {
1357 return std::string(len, c);
1358 }
1359
VerifyIterLast(std::string expected_key,int cf)1360 void DBTestBase::VerifyIterLast(std::string expected_key, int cf) {
1361 Iterator* iter;
1362 ReadOptions ro;
1363 if (cf == 0) {
1364 iter = db_->NewIterator(ro);
1365 } else {
1366 iter = db_->NewIterator(ro, handles_[cf]);
1367 }
1368 iter->SeekToLast();
1369 ASSERT_EQ(IterStatus(iter), expected_key);
1370 delete iter;
1371 }
1372
1373 // Used to test InplaceUpdate
1374
1375 // If previous value is nullptr or delta is > than previous value,
1376 // sets newValue with delta
1377 // If previous value is not empty,
1378 // updates previous value with 'b' string of previous value size - 1.
updateInPlaceSmallerSize(char * prevValue,uint32_t * prevSize,Slice delta,std::string * newValue)1379 UpdateStatus DBTestBase::updateInPlaceSmallerSize(char* prevValue,
1380 uint32_t* prevSize,
1381 Slice delta,
1382 std::string* newValue) {
1383 if (prevValue == nullptr) {
1384 *newValue = std::string(delta.size(), 'c');
1385 return UpdateStatus::UPDATED;
1386 } else {
1387 *prevSize = *prevSize - 1;
1388 std::string str_b = std::string(*prevSize, 'b');
1389 memcpy(prevValue, str_b.c_str(), str_b.size());
1390 return UpdateStatus::UPDATED_INPLACE;
1391 }
1392 }
1393
updateInPlaceSmallerVarintSize(char * prevValue,uint32_t * prevSize,Slice delta,std::string * newValue)1394 UpdateStatus DBTestBase::updateInPlaceSmallerVarintSize(char* prevValue,
1395 uint32_t* prevSize,
1396 Slice delta,
1397 std::string* newValue) {
1398 if (prevValue == nullptr) {
1399 *newValue = std::string(delta.size(), 'c');
1400 return UpdateStatus::UPDATED;
1401 } else {
1402 *prevSize = 1;
1403 std::string str_b = std::string(*prevSize, 'b');
1404 memcpy(prevValue, str_b.c_str(), str_b.size());
1405 return UpdateStatus::UPDATED_INPLACE;
1406 }
1407 }
1408
updateInPlaceLargerSize(char *,uint32_t *,Slice delta,std::string * newValue)1409 UpdateStatus DBTestBase::updateInPlaceLargerSize(char* /*prevValue*/,
1410 uint32_t* /*prevSize*/,
1411 Slice delta,
1412 std::string* newValue) {
1413 *newValue = std::string(delta.size(), 'c');
1414 return UpdateStatus::UPDATED;
1415 }
1416
updateInPlaceNoAction(char *,uint32_t *,Slice,std::string *)1417 UpdateStatus DBTestBase::updateInPlaceNoAction(char* /*prevValue*/,
1418 uint32_t* /*prevSize*/,
1419 Slice /*delta*/,
1420 std::string* /*newValue*/) {
1421 return UpdateStatus::UPDATE_FAILED;
1422 }
1423
1424 // Utility method to test InplaceUpdate
validateNumberOfEntries(int numValues,int cf)1425 void DBTestBase::validateNumberOfEntries(int numValues, int cf) {
1426 Arena arena;
1427 auto options = CurrentOptions();
1428 InternalKeyComparator icmp(options.comparator);
1429 ReadRangeDelAggregator range_del_agg(&icmp,
1430 kMaxSequenceNumber /* upper_bound */);
1431 // This should be defined after range_del_agg so that it destructs the
1432 // assigned iterator before it range_del_agg is already destructed.
1433 ReadOptions read_options;
1434 ScopedArenaIterator iter;
1435 if (cf != 0) {
1436 iter.set(dbfull()->NewInternalIterator(read_options, &arena, &range_del_agg,
1437 kMaxSequenceNumber, handles_[cf]));
1438 } else {
1439 iter.set(dbfull()->NewInternalIterator(read_options, &arena, &range_del_agg,
1440 kMaxSequenceNumber));
1441 }
1442 iter->SeekToFirst();
1443 ASSERT_OK(iter->status());
1444 int seq = numValues;
1445 while (iter->Valid()) {
1446 ParsedInternalKey ikey;
1447 ikey.clear();
1448 ASSERT_OK(ParseInternalKey(iter->key(), &ikey, true /* log_err_key */));
1449
1450 // checks sequence number for updates
1451 ASSERT_EQ(ikey.sequence, (unsigned)seq--);
1452 iter->Next();
1453 }
1454 ASSERT_EQ(0, seq);
1455 }
1456
CopyFile(const std::string & source,const std::string & destination,uint64_t size)1457 void DBTestBase::CopyFile(const std::string& source,
1458 const std::string& destination, uint64_t size) {
1459 const EnvOptions soptions;
1460 std::unique_ptr<SequentialFile> srcfile;
1461 ASSERT_OK(env_->NewSequentialFile(source, &srcfile, soptions));
1462 std::unique_ptr<WritableFile> destfile;
1463 ASSERT_OK(env_->NewWritableFile(destination, &destfile, soptions));
1464
1465 if (size == 0) {
1466 // default argument means copy everything
1467 ASSERT_OK(env_->GetFileSize(source, &size));
1468 }
1469
1470 char buffer[4096];
1471 Slice slice;
1472 while (size > 0) {
1473 uint64_t one = std::min(uint64_t(sizeof(buffer)), size);
1474 ASSERT_OK(srcfile->Read(one, &slice, buffer));
1475 ASSERT_OK(destfile->Append(slice));
1476 size -= slice.size();
1477 }
1478 ASSERT_OK(destfile->Close());
1479 }
1480
GetAllDataFiles(const FileType file_type,std::unordered_map<std::string,uint64_t> * files,uint64_t * total_size)1481 Status DBTestBase::GetAllDataFiles(
1482 const FileType file_type, std::unordered_map<std::string, uint64_t>* files,
1483 uint64_t* total_size /* = nullptr */) {
1484 if (total_size) {
1485 *total_size = 0;
1486 }
1487 std::vector<std::string> children;
1488 Status s = env_->GetChildren(dbname_, &children);
1489 if (s.ok()) {
1490 for (auto& file_name : children) {
1491 uint64_t number;
1492 FileType type;
1493 if (ParseFileName(file_name, &number, &type) && type == file_type) {
1494 std::string file_path = dbname_ + "/" + file_name;
1495 uint64_t file_size = 0;
1496 s = env_->GetFileSize(file_path, &file_size);
1497 if (!s.ok()) {
1498 break;
1499 }
1500 (*files)[file_path] = file_size;
1501 if (total_size) {
1502 *total_size += file_size;
1503 }
1504 }
1505 }
1506 }
1507 return s;
1508 }
1509
ListTableFiles(Env * env,const std::string & path)1510 std::vector<std::uint64_t> DBTestBase::ListTableFiles(Env* env,
1511 const std::string& path) {
1512 std::vector<std::string> files;
1513 std::vector<uint64_t> file_numbers;
1514 EXPECT_OK(env->GetChildren(path, &files));
1515 uint64_t number;
1516 FileType type;
1517 for (size_t i = 0; i < files.size(); ++i) {
1518 if (ParseFileName(files[i], &number, &type)) {
1519 if (type == kTableFile) {
1520 file_numbers.push_back(number);
1521 }
1522 }
1523 }
1524 return file_numbers;
1525 }
1526
VerifyDBFromMap(std::map<std::string,std::string> true_data,size_t * total_reads_res,bool tailing_iter,std::map<std::string,Status> status)1527 void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
1528 size_t* total_reads_res, bool tailing_iter,
1529 std::map<std::string, Status> status) {
1530 size_t total_reads = 0;
1531
1532 for (auto& kv : true_data) {
1533 Status s = status[kv.first];
1534 if (s.ok()) {
1535 ASSERT_EQ(Get(kv.first), kv.second);
1536 } else {
1537 std::string value;
1538 ASSERT_EQ(s, db_->Get(ReadOptions(), kv.first, &value));
1539 }
1540 total_reads++;
1541 }
1542
1543 // Normal Iterator
1544 {
1545 int iter_cnt = 0;
1546 ReadOptions ro;
1547 ro.total_order_seek = true;
1548 Iterator* iter = db_->NewIterator(ro);
1549 // Verify Iterator::Next()
1550 iter_cnt = 0;
1551 auto data_iter = true_data.begin();
1552 Status s;
1553 for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) {
1554 ASSERT_EQ(iter->key().ToString(), data_iter->first);
1555 Status current_status = status[data_iter->first];
1556 if (!current_status.ok()) {
1557 s = current_status;
1558 }
1559 ASSERT_EQ(iter->status(), s);
1560 if (current_status.ok()) {
1561 ASSERT_EQ(iter->value().ToString(), data_iter->second);
1562 }
1563 iter_cnt++;
1564 total_reads++;
1565 }
1566 ASSERT_EQ(data_iter, true_data.end()) << iter_cnt << " / "
1567 << true_data.size();
1568 delete iter;
1569
1570 // Verify Iterator::Prev()
1571 // Use a new iterator to make sure its status is clean.
1572 iter = db_->NewIterator(ro);
1573 iter_cnt = 0;
1574 s = Status::OK();
1575 auto data_rev = true_data.rbegin();
1576 for (iter->SeekToLast(); iter->Valid(); iter->Prev(), data_rev++) {
1577 ASSERT_EQ(iter->key().ToString(), data_rev->first);
1578 Status current_status = status[data_rev->first];
1579 if (!current_status.ok()) {
1580 s = current_status;
1581 }
1582 ASSERT_EQ(iter->status(), s);
1583 if (current_status.ok()) {
1584 ASSERT_EQ(iter->value().ToString(), data_rev->second);
1585 }
1586 iter_cnt++;
1587 total_reads++;
1588 }
1589 ASSERT_EQ(data_rev, true_data.rend()) << iter_cnt << " / "
1590 << true_data.size();
1591
1592 // Verify Iterator::Seek()
1593 for (auto kv : true_data) {
1594 iter->Seek(kv.first);
1595 ASSERT_EQ(kv.first, iter->key().ToString());
1596 ASSERT_EQ(kv.second, iter->value().ToString());
1597 total_reads++;
1598 }
1599 delete iter;
1600 }
1601
1602 if (tailing_iter) {
1603 #ifndef ROCKSDB_LITE
1604 // Tailing iterator
1605 int iter_cnt = 0;
1606 ReadOptions ro;
1607 ro.tailing = true;
1608 ro.total_order_seek = true;
1609 Iterator* iter = db_->NewIterator(ro);
1610
1611 // Verify ForwardIterator::Next()
1612 iter_cnt = 0;
1613 auto data_iter = true_data.begin();
1614 for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) {
1615 ASSERT_EQ(iter->key().ToString(), data_iter->first);
1616 ASSERT_EQ(iter->value().ToString(), data_iter->second);
1617 iter_cnt++;
1618 total_reads++;
1619 }
1620 ASSERT_EQ(data_iter, true_data.end()) << iter_cnt << " / "
1621 << true_data.size();
1622
1623 // Verify ForwardIterator::Seek()
1624 for (auto kv : true_data) {
1625 iter->Seek(kv.first);
1626 ASSERT_EQ(kv.first, iter->key().ToString());
1627 ASSERT_EQ(kv.second, iter->value().ToString());
1628 total_reads++;
1629 }
1630
1631 delete iter;
1632 #endif // ROCKSDB_LITE
1633 }
1634
1635 if (total_reads_res) {
1636 *total_reads_res = total_reads;
1637 }
1638 }
1639
VerifyDBInternal(std::vector<std::pair<std::string,std::string>> true_data)1640 void DBTestBase::VerifyDBInternal(
1641 std::vector<std::pair<std::string, std::string>> true_data) {
1642 Arena arena;
1643 InternalKeyComparator icmp(last_options_.comparator);
1644 ReadRangeDelAggregator range_del_agg(&icmp,
1645 kMaxSequenceNumber /* upper_bound */);
1646 ReadOptions read_options;
1647 auto iter = dbfull()->NewInternalIterator(read_options, &arena,
1648 &range_del_agg, kMaxSequenceNumber);
1649 iter->SeekToFirst();
1650 for (auto p : true_data) {
1651 ASSERT_TRUE(iter->Valid());
1652 ParsedInternalKey ikey;
1653 ASSERT_OK(ParseInternalKey(iter->key(), &ikey, true /* log_err_key */));
1654 ASSERT_EQ(p.first, ikey.user_key);
1655 ASSERT_EQ(p.second, iter->value());
1656 iter->Next();
1657 };
1658 ASSERT_FALSE(iter->Valid());
1659 iter->~InternalIterator();
1660 }
1661
1662 #ifndef ROCKSDB_LITE
1663
GetNumberOfSstFilesForColumnFamily(DB * db,std::string column_family_name)1664 uint64_t DBTestBase::GetNumberOfSstFilesForColumnFamily(
1665 DB* db, std::string column_family_name) {
1666 std::vector<LiveFileMetaData> metadata;
1667 db->GetLiveFilesMetaData(&metadata);
1668 uint64_t result = 0;
1669 for (auto& fileMetadata : metadata) {
1670 result += (fileMetadata.column_family_name == column_family_name);
1671 }
1672 return result;
1673 }
1674 #endif // ROCKSDB_LITE
1675
VerifySstUniqueIds(const TablePropertiesCollection & props)1676 void VerifySstUniqueIds(const TablePropertiesCollection& props) {
1677 ASSERT_FALSE(props.empty()); // suspicious test if empty
1678 std::unordered_set<std::string> seen;
1679 for (auto& pair : props) {
1680 std::string id;
1681 ASSERT_OK(GetUniqueIdFromTableProperties(*pair.second, &id));
1682 ASSERT_TRUE(seen.insert(id).second);
1683 }
1684 }
1685
1686 } // namespace ROCKSDB_NAMESPACE
1687