db_test_util.h 31.4 KB
Newer Older
1
// Copyright (c) 2011-present, Facebook, Inc.  All rights reserved.
S
Siying Dong 已提交
2 3 4
//  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).
5 6 7 8 9 10
//
// 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.

#pragma once
11

12
#include <fcntl.h>
13
#include <cinttypes>
14 15

#include <algorithm>
16
#include <map>
17 18 19 20 21 22 23
#include <set>
#include <string>
#include <thread>
#include <unordered_set>
#include <utility>
#include <vector>

24
#include "db/db_impl/db_impl.h"
25
#include "db/dbformat.h"
26
#include "env/mock_env.h"
27
#include "file/filename.h"
28
#include "memtable/hash_linklist_rep.h"
29 30
#include "rocksdb/cache.h"
#include "rocksdb/compaction_filter.h"
A
agiardullo 已提交
31
#include "rocksdb/convenience.h"
32 33 34 35 36
#include "rocksdb/db.h"
#include "rocksdb/env.h"
#include "rocksdb/filter_policy.h"
#include "rocksdb/options.h"
#include "rocksdb/slice.h"
37
#include "rocksdb/sst_file_writer.h"
Y
Yi Wu 已提交
38
#include "rocksdb/statistics.h"
39 40
#include "rocksdb/table.h"
#include "rocksdb/utilities/checkpoint.h"
41
#include "table/block_based/block_based_table_factory.h"
42
#include "table/mock_table.h"
43
#include "table/plain/plain_table_factory.h"
S
sdong 已提交
44
#include "table/scoped_arena_iterator.h"
45
#include "test_util/mock_time_env.h"
46
#include "util/compression.h"
47
#include "util/mutexlock.h"
S
sdong 已提交
48

49 50 51
#include "test_util/sync_point.h"
#include "test_util/testharness.h"
#include "test_util/testutil.h"
52
#include "util/string_util.h"
53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79
#include "utilities/merge_operators.h"

namespace rocksdb {

namespace anon {
class AtomicCounter {
 public:
  explicit AtomicCounter(Env* env = NULL)
      : env_(env), cond_count_(&mu_), count_(0) {}

  void Increment() {
    MutexLock l(&mu_);
    count_++;
    cond_count_.SignalAll();
  }

  int Read() {
    MutexLock l(&mu_);
    return count_;
  }

  bool WaitFor(int count) {
    MutexLock l(&mu_);

    uint64_t start = env_->NowMicros();
    while (count_ < count) {
      uint64_t now = env_->NowMicros();
80 81
      cond_count_.TimedWait(now + /*1s*/ 1 * 1000 * 1000);
      if (env_->NowMicros() - start > /*10s*/ 10 * 1000 * 1000) {
82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106
        return false;
      }
      if (count_ < count) {
        GTEST_LOG_(WARNING) << "WaitFor is taking more time than usual";
      }
    }

    return true;
  }

  void Reset() {
    MutexLock l(&mu_);
    count_ = 0;
    cond_count_.SignalAll();
  }

 private:
  Env* env_;
  port::Mutex mu_;
  port::CondVar cond_count_;
  int count_;
};

struct OptionsOverride {
  std::shared_ptr<const FilterPolicy> filter_policy = nullptr;
M
Maysam Yabandeh 已提交
107 108
  // These will be used only if filter_policy is set
  bool partition_filters = false;
M
Maysam Yabandeh 已提交
109
  uint64_t metadata_block_size = 1024;
110 111 112 113 114 115 116

  // Used as a bit mask of individual enums in which to skip an XF test point
  int skip_policy = 0;
};

}  // namespace anon

S
Siying Dong 已提交
117 118
enum SkipPolicy { kSkipNone = 0, kSkipNoSnapshot = 1, kSkipNoPrefix = 2 };

119 120 121
// A hacky skip list mem table that triggers flush after number of entries.
class SpecialMemTableRep : public MemTableRep {
 public:
122 123
  explicit SpecialMemTableRep(Allocator* allocator, MemTableRep* memtable,
                              int num_entries_flush)
124 125 126 127 128 129 130 131 132 133 134
      : MemTableRep(allocator),
        memtable_(memtable),
        num_entries_flush_(num_entries_flush),
        num_entries_(0) {}

  virtual KeyHandle Allocate(const size_t len, char** buf) override {
    return memtable_->Allocate(len, buf);
  }

  // Insert key into the list.
  // REQUIRES: nothing that compares equal to key is currently in the list.
M
Maysam Yabandeh 已提交
135
  virtual void Insert(KeyHandle handle) override {
136
    num_entries_++;
M
Maysam Yabandeh 已提交
137
    memtable_->Insert(handle);
138 139
  }

M
Maysam Yabandeh 已提交
140 141 142 143 144
  void InsertConcurrently(KeyHandle handle) override {
    num_entries_++;
    memtable_->Insert(handle);
  }

145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173
  // Returns true iff an entry that compares equal to key is in the list.
  virtual bool Contains(const char* key) const override {
    return memtable_->Contains(key);
  }

  virtual size_t ApproximateMemoryUsage() override {
    // Return a high memory usage when number of entries exceeds the threshold
    // to trigger a flush.
    return (num_entries_ < num_entries_flush_) ? 0 : 1024 * 1024 * 1024;
  }

  virtual void Get(const LookupKey& k, void* callback_args,
                   bool (*callback_func)(void* arg,
                                         const char* entry)) override {
    memtable_->Get(k, callback_args, callback_func);
  }

  uint64_t ApproximateNumEntries(const Slice& start_ikey,
                                 const Slice& end_ikey) override {
    return memtable_->ApproximateNumEntries(start_ikey, end_ikey);
  }

  virtual MemTableRep::Iterator* GetIterator(Arena* arena = nullptr) override {
    return memtable_->GetIterator(arena);
  }

  virtual ~SpecialMemTableRep() override {}

 private:
174
  std::unique_ptr<MemTableRep> memtable_;
175 176 177 178 179 180 181 182 183 184 185 186 187
  int num_entries_flush_;
  int num_entries_;
};

// The factory for the hacky skip list mem table that triggers flush after
// number of entries exceeds a threshold.
class SpecialSkipListFactory : public MemTableRepFactory {
 public:
  // After number of inserts exceeds `num_entries_flush` in a mem table, trigger
  // flush.
  explicit SpecialSkipListFactory(int num_entries_flush)
      : num_entries_flush_(num_entries_flush) {}

188
  using MemTableRepFactory::CreateMemTableRep;
189
  virtual MemTableRep* CreateMemTableRep(
190
      const MemTableRep::KeyComparator& compare, Allocator* allocator,
A
Andrew Kryczka 已提交
191
      const SliceTransform* transform, Logger* /*logger*/) override {
192 193 194 195 196 197
    return new SpecialMemTableRep(
        allocator, factory_.CreateMemTableRep(compare, allocator, transform, 0),
        num_entries_flush_);
  }
  virtual const char* Name() const override { return "SkipListFactory"; }

198 199 200 201
  bool IsInsertConcurrentlySupported() const override {
    return factory_.IsInsertConcurrentlySupported();
  }

202 203 204 205 206
 private:
  SkipListFactory factory_;
  int num_entries_flush_;
};

207 208 209 210 211
// Special Env used to delay background operations
class SpecialEnv : public EnvWrapper {
 public:
  explicit SpecialEnv(Env* base);

212
  Status NewWritableFile(const std::string& f, std::unique_ptr<WritableFile>* r,
213 214 215 216
                         const EnvOptions& soptions) override {
    class SSTableFile : public WritableFile {
     private:
      SpecialEnv* env_;
217
      std::unique_ptr<WritableFile> base_;
218 219

     public:
220
      SSTableFile(SpecialEnv* env, std::unique_ptr<WritableFile>&& base)
S
sdong 已提交
221
          : env_(env), base_(std::move(base)) {}
222 223 224 225 226 227 228 229
      Status Append(const Slice& data) override {
        if (env_->table_write_callback_) {
          (*env_->table_write_callback_)();
        }
        if (env_->drop_writes_.load(std::memory_order_acquire)) {
          // Drop writes on the floor
          return Status::OK();
        } else if (env_->no_space_.load(std::memory_order_acquire)) {
230
          return Status::NoSpace("No space left on device");
231 232 233 234 235
        } else {
          env_->bytes_written_ += data.size();
          return base_->Append(data);
        }
      }
A
Aaron Gao 已提交
236
      Status PositionedAppend(const Slice& data, uint64_t offset) override {
A
Aaron Gao 已提交
237 238 239 240 241 242 243 244 245 246 247 248 249
        if (env_->table_write_callback_) {
          (*env_->table_write_callback_)();
        }
        if (env_->drop_writes_.load(std::memory_order_acquire)) {
          // Drop writes on the floor
          return Status::OK();
        } else if (env_->no_space_.load(std::memory_order_acquire)) {
          return Status::NoSpace("No space left on device");
        } else {
          env_->bytes_written_ += data.size();
          return base_->PositionedAppend(data, offset);
        }
      }
S
sdong 已提交
250
      Status Truncate(uint64_t size) override { return base_->Truncate(size); }
251 252 253 254 255 256 257
      Status RangeSync(uint64_t offset, uint64_t nbytes) override {
        Status s = base_->RangeSync(offset, nbytes);
#if !(defined NDEBUG) || !defined(OS_WIN)
        TEST_SYNC_POINT_CALLBACK("SpecialEnv::SStableFile::RangeSync", &s);
#endif  // !(defined NDEBUG) || !defined(OS_WIN)
        return s;
      }
258
      Status Close() override {
259 260
// SyncPoint is not supported in Released Windows Mode.
#if !(defined NDEBUG) || !defined(OS_WIN)
261 262 263 264 265
        // Check preallocation size
        // preallocation size is never passed to base file.
        size_t preallocation_size = preallocation_block_size();
        TEST_SYNC_POINT_CALLBACK("DBTestWritableFile.GetPreallocationStatus",
                                 &preallocation_size);
266
#endif  // !(defined NDEBUG) || !defined(OS_WIN)
267 268 269 270 271
        Status s = base_->Close();
#if !(defined NDEBUG) || !defined(OS_WIN)
        TEST_SYNC_POINT_CALLBACK("SpecialEnv::SStableFile::Close", &s);
#endif  // !(defined NDEBUG) || !defined(OS_WIN)
        return s;
272 273 274 275 276 277 278
      }
      Status Flush() override { return base_->Flush(); }
      Status Sync() override {
        ++env_->sync_counter_;
        while (env_->delay_sstable_sync_.load(std::memory_order_acquire)) {
          env_->SleepForMicroseconds(100000);
        }
279 280 281 282 283
        Status s = base_->Sync();
#if !(defined NDEBUG) || !defined(OS_WIN)
        TEST_SYNC_POINT_CALLBACK("SpecialEnv::SStableFile::Sync", &s);
#endif  // !(defined NDEBUG) || !defined(OS_WIN)
        return s;
284 285 286 287
      }
      void SetIOPriority(Env::IOPriority pri) override {
        base_->SetIOPriority(pri);
      }
288 289 290
      Env::IOPriority GetIOPriority() override {
        return base_->GetIOPriority();
      }
A
Aaron Gao 已提交
291 292 293
      bool use_direct_io() const override {
        return base_->use_direct_io();
      }
294 295 296
      Status Allocate(uint64_t offset, uint64_t len) override {
        return base_->Allocate(offset, len);
      }
297 298 299
    };
    class ManifestFile : public WritableFile {
     public:
300
      ManifestFile(SpecialEnv* env, std::unique_ptr<WritableFile>&& b)
S
sdong 已提交
301
          : env_(env), base_(std::move(b)) {}
302 303 304 305 306 307 308
      Status Append(const Slice& data) override {
        if (env_->manifest_write_error_.load(std::memory_order_acquire)) {
          return Status::IOError("simulated writer error");
        } else {
          return base_->Append(data);
        }
      }
309
      Status Truncate(uint64_t size) override { return base_->Truncate(size); }
310 311 312 313 314 315 316 317 318 319 320
      Status Close() override { return base_->Close(); }
      Status Flush() override { return base_->Flush(); }
      Status Sync() override {
        ++env_->sync_counter_;
        if (env_->manifest_sync_error_.load(std::memory_order_acquire)) {
          return Status::IOError("simulated sync error");
        } else {
          return base_->Sync();
        }
      }
      uint64_t GetFileSize() override { return base_->GetFileSize(); }
321 322 323
      Status Allocate(uint64_t offset, uint64_t len) override {
        return base_->Allocate(offset, len);
      }
324 325 326

     private:
      SpecialEnv* env_;
327
      std::unique_ptr<WritableFile> base_;
328 329 330
    };
    class WalFile : public WritableFile {
     public:
331
      WalFile(SpecialEnv* env, std::unique_ptr<WritableFile>&& b)
332 333 334 335
          : env_(env), base_(std::move(b)) {
        env_->num_open_wal_file_.fetch_add(1);
      }
      virtual ~WalFile() { env_->num_open_wal_file_.fetch_add(-1); }
336
      Status Append(const Slice& data) override {
S
sdong 已提交
337 338 339 340
#if !(defined NDEBUG) || !defined(OS_WIN)
        TEST_SYNC_POINT("SpecialEnv::WalFile::Append:1");
#endif
        Status s;
341
        if (env_->log_write_error_.load(std::memory_order_acquire)) {
S
sdong 已提交
342
          s = Status::IOError("simulated writer error");
343 344 345 346 347 348
        } else {
          int slowdown =
              env_->log_write_slowdown_.load(std::memory_order_acquire);
          if (slowdown > 0) {
            env_->SleepForMicroseconds(slowdown);
          }
S
sdong 已提交
349
          s = base_->Append(data);
350
        }
S
sdong 已提交
351 352 353 354
#if !(defined NDEBUG) || !defined(OS_WIN)
        TEST_SYNC_POINT("SpecialEnv::WalFile::Append:2");
#endif
        return s;
355
      }
356
      Status Truncate(uint64_t size) override { return base_->Truncate(size); }
357 358 359 360 361 362 363 364 365 366 367 368
      Status Close() override {
// SyncPoint is not supported in Released Windows Mode.
#if !(defined NDEBUG) || !defined(OS_WIN)
        // Check preallocation size
        // preallocation size is never passed to base file.
        size_t preallocation_size = preallocation_block_size();
        TEST_SYNC_POINT_CALLBACK("DBTestWalFile.GetPreallocationStatus",
                                 &preallocation_size);
#endif  // !(defined NDEBUG) || !defined(OS_WIN)

        return base_->Close();
      }
369 370 371 372 373
      Status Flush() override { return base_->Flush(); }
      Status Sync() override {
        ++env_->sync_counter_;
        return base_->Sync();
      }
374 375 376
      bool IsSyncThreadSafe() const override {
        return env_->is_wal_sync_thread_safe_.load();
      }
377 378 379
      Status Allocate(uint64_t offset, uint64_t len) override {
        return base_->Allocate(offset, len);
      }
380 381 382

     private:
      SpecialEnv* env_;
383
      std::unique_ptr<WritableFile> base_;
384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403
    };

    if (non_writeable_rate_.load(std::memory_order_acquire) > 0) {
      uint32_t random_number;
      {
        MutexLock l(&rnd_mutex_);
        random_number = rnd_.Uniform(100);
      }
      if (random_number < non_writeable_rate_.load()) {
        return Status::IOError("simulated random write error");
      }
    }

    new_writable_count_++;

    if (non_writable_count_.load() > 0) {
      non_writable_count_--;
      return Status::IOError("simulated write error");
    }

A
Aaron Gao 已提交
404 405 406 407 408 409 410 411
    EnvOptions optimized = soptions;
    if (strstr(f.c_str(), "MANIFEST") != nullptr ||
        strstr(f.c_str(), "log") != nullptr) {
      optimized.use_mmap_writes = false;
      optimized.use_direct_writes = false;
    }

    Status s = target()->NewWritableFile(f, r, optimized);
412 413 414 415 416 417 418 419 420 421 422 423 424
    if (s.ok()) {
      if (strstr(f.c_str(), ".sst") != nullptr) {
        r->reset(new SSTableFile(this, std::move(*r)));
      } else if (strstr(f.c_str(), "MANIFEST") != nullptr) {
        r->reset(new ManifestFile(this, std::move(*r)));
      } else if (strstr(f.c_str(), "log") != nullptr) {
        r->reset(new WalFile(this, std::move(*r)));
      }
    }
    return s;
  }

  Status NewRandomAccessFile(const std::string& f,
425
                             std::unique_ptr<RandomAccessFile>* r,
426 427 428
                             const EnvOptions& soptions) override {
    class CountingFile : public RandomAccessFile {
     public:
429
      CountingFile(std::unique_ptr<RandomAccessFile>&& target,
430
                   anon::AtomicCounter* counter,
Y
Yi Wu 已提交
431
                   std::atomic<size_t>* bytes_read)
432 433 434
          : target_(std::move(target)),
            counter_(counter),
            bytes_read_(bytes_read) {}
435 436 437
      virtual Status Read(uint64_t offset, size_t n, Slice* result,
                          char* scratch) const override {
        counter_->Increment();
438 439 440
        Status s = target_->Read(offset, n, result, scratch);
        *bytes_read_ += result->size();
        return s;
441 442
      }

443 444 445 446 447 448
      virtual Status Prefetch(uint64_t offset, size_t n) override {
        Status s = target_->Prefetch(offset, n);
        *bytes_read_ += n;
        return s;
      }

449
     private:
450
      std::unique_ptr<RandomAccessFile> target_;
451
      anon::AtomicCounter* counter_;
Y
Yi Wu 已提交
452
      std::atomic<size_t>* bytes_read_;
453 454 455
    };

    Status s = target()->NewRandomAccessFile(f, r, soptions);
456
    random_file_open_counter_++;
457
    if (s.ok() && count_random_reads_) {
458 459
      r->reset(new CountingFile(std::move(*r), &random_read_counter_,
                                &random_read_bytes_counter_));
460
    }
461 462 463
    if (s.ok() && soptions.compaction_readahead_size > 0) {
      compaction_readahead_size_ = soptions.compaction_readahead_size;
    }
464 465 466
    return s;
  }

S
Siying Dong 已提交
467
  virtual Status NewSequentialFile(const std::string& f,
468
                                   std::unique_ptr<SequentialFile>* r,
S
Siying Dong 已提交
469
                                   const EnvOptions& soptions) override {
470 471
    class CountingFile : public SequentialFile {
     public:
472
      CountingFile(std::unique_ptr<SequentialFile>&& target,
473 474 475 476 477 478 479 480 481
                   anon::AtomicCounter* counter)
          : target_(std::move(target)), counter_(counter) {}
      virtual Status Read(size_t n, Slice* result, char* scratch) override {
        counter_->Increment();
        return target_->Read(n, result, scratch);
      }
      virtual Status Skip(uint64_t n) override { return target_->Skip(n); }

     private:
482
      std::unique_ptr<SequentialFile> target_;
483 484 485 486 487 488 489 490 491 492 493 494
      anon::AtomicCounter* counter_;
    };

    Status s = target()->NewSequentialFile(f, r, soptions);
    if (s.ok() && count_sequential_reads_) {
      r->reset(new CountingFile(std::move(*r), &sequential_read_counter_));
    }
    return s;
  }

  virtual void SleepForMicroseconds(int micros) override {
    sleep_counter_.Increment();
M
Maysam Yabandeh 已提交
495
    if (no_slowdown_ || time_elapse_only_sleep_) {
496
      addon_time_.fetch_add(micros);
497
    }
M
Maysam Yabandeh 已提交
498
    if (!no_slowdown_) {
499 500 501 502 503
      target()->SleepForMicroseconds(micros);
    }
  }

  virtual Status GetCurrentTime(int64_t* unix_time) override {
504 505 506 507
    Status s;
    if (!time_elapse_only_sleep_) {
      s = target()->GetCurrentTime(unix_time);
    }
508 509 510 511 512 513
    if (s.ok()) {
      *unix_time += addon_time_.load();
    }
    return s;
  }

514 515 516 517 518
  virtual uint64_t NowCPUNanos() override {
    now_cpu_count_.fetch_add(1);
    return target()->NowCPUNanos();
  }

519
  virtual uint64_t NowNanos() override {
520 521
    return (time_elapse_only_sleep_ ? 0 : target()->NowNanos()) +
           addon_time_.load() * 1000;
522 523 524
  }

  virtual uint64_t NowMicros() override {
525 526
    return (time_elapse_only_sleep_ ? 0 : target()->NowMicros()) +
           addon_time_.load();
527 528
  }

529 530 531 532 533
  virtual Status DeleteFile(const std::string& fname) override {
    delete_count_.fetch_add(1);
    return target()->DeleteFile(fname);
  }

534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560
  Random rnd_;
  port::Mutex rnd_mutex_;  // Lock to pretect rnd_

  // sstable Sync() calls are blocked while this pointer is non-nullptr.
  std::atomic<bool> delay_sstable_sync_;

  // Drop writes on the floor while this pointer is non-nullptr.
  std::atomic<bool> drop_writes_;

  // Simulate no-space errors while this pointer is non-nullptr.
  std::atomic<bool> no_space_;

  // Simulate non-writable file system while this pointer is non-nullptr
  std::atomic<bool> non_writable_;

  // Force sync of manifest files to fail while this pointer is non-nullptr
  std::atomic<bool> manifest_sync_error_;

  // Force write to manifest files to fail while this pointer is non-nullptr
  std::atomic<bool> manifest_write_error_;

  // Force write to log files to fail while this pointer is non-nullptr
  std::atomic<bool> log_write_error_;

  // Slow down every log write, in micro-seconds.
  std::atomic<int> log_write_slowdown_;

561 562 563
  // Number of WAL files that are still open for write.
  std::atomic<int> num_open_wal_file_;

564 565
  bool count_random_reads_;
  anon::AtomicCounter random_read_counter_;
Y
Yi Wu 已提交
566
  std::atomic<size_t> random_read_bytes_counter_;
567
  std::atomic<int> random_file_open_counter_;
568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586

  bool count_sequential_reads_;
  anon::AtomicCounter sequential_read_counter_;

  anon::AtomicCounter sleep_counter_;

  std::atomic<int64_t> bytes_written_;

  std::atomic<int> sync_counter_;

  std::atomic<uint32_t> non_writeable_rate_;

  std::atomic<uint32_t> new_writable_count_;

  std::atomic<uint32_t> non_writable_count_;

  std::function<void()>* table_write_callback_;

  std::atomic<int64_t> addon_time_;
587

588 589
  std::atomic<int> now_cpu_count_;

590 591
  std::atomic<int> delete_count_;

592
  std::atomic<bool> time_elapse_only_sleep_;
593

M
Maysam Yabandeh 已提交
594
  bool no_slowdown_;
595

S
sdong 已提交
596
  std::atomic<bool> is_wal_sync_thread_safe_{true};
597

598
  std::atomic<size_t> compaction_readahead_size_{};
599 600
};

Y
Yi Wu 已提交
601
#ifndef ROCKSDB_LITE
Y
Yi Wu 已提交
602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625
class OnFileDeletionListener : public EventListener {
 public:
  OnFileDeletionListener() : matched_count_(0), expected_file_name_("") {}

  void SetExpectedFileName(const std::string file_name) {
    expected_file_name_ = file_name;
  }

  void VerifyMatchedCount(size_t expected_value) {
    ASSERT_EQ(matched_count_, expected_value);
  }

  void OnTableFileDeleted(const TableFileDeletionInfo& info) override {
    if (expected_file_name_ != "") {
      ASSERT_EQ(expected_file_name_, info.file_path);
      expected_file_name_ = "";
      matched_count_++;
    }
  }

 private:
  size_t matched_count_;
  std::string expected_file_name_;
};
Y
Yi Wu 已提交
626
#endif
Y
Yi Wu 已提交
627

628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649
// A test merge operator mimics put but also fails if one of merge operands is
// "corrupted".
class TestPutOperator : public MergeOperator {
 public:
  virtual bool FullMergeV2(const MergeOperationInput& merge_in,
                           MergeOperationOutput* merge_out) const override {
    if (merge_in.existing_value != nullptr &&
        *(merge_in.existing_value) == "corrupted") {
      return false;
    }
    for (auto value : merge_in.operand_list) {
      if (value == "corrupted") {
        return false;
      }
    }
    merge_out->existing_operand = merge_in.operand_list.back();
    return true;
  }

  virtual const char* Name() const override { return "TestPutOperator"; }
};

650
class DBTestBase : public testing::Test {
Y
Yi Wu 已提交
651
 public:
652
  // Sequence of option configurations to try
Y
Yi Wu 已提交
653
  enum OptionConfig : int {
654 655 656 657 658
    kDefault = 0,
    kBlockBasedTableWithPrefixHashIndex = 1,
    kBlockBasedTableWithWholeKeyHashIndex = 2,
    kPlainTableFirstBytePrefix = 3,
    kPlainTableCappedPrefix = 4,
659 660 661 662
    kPlainTableCappedPrefixNonMmap = 5,
    kPlainTableAllBytesPrefix = 6,
    kVectorRep = 7,
    kHashLinkList = 8,
663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684
    kMergePut = 9,
    kFilter = 10,
    kFullFilterWithNewTableReaderForCompactions = 11,
    kUncompressed = 12,
    kNumLevel_3 = 13,
    kDBLogDir = 14,
    kWalDirAndMmapReads = 15,
    kManifestFileSize = 16,
    kPerfOptions = 17,
    kHashSkipList = 18,
    kUniversalCompaction = 19,
    kUniversalCompactionMultiLevel = 20,
    kCompressedBlockCache = 21,
    kInfiniteMaxOpenFiles = 22,
    kxxHashChecksum = 23,
    kFIFOCompaction = 24,
    kOptimizeFiltersForHits = 25,
    kRowCache = 26,
    kRecycleLogFiles = 27,
    kConcurrentSkipList = 28,
    kPipelinedWrite = 29,
    kConcurrentWALWrites = 30,
685 686 687 688
    kDirectIO,
    kLevelSubcompactions,
    kBlockBasedTableWithIndexRestartInterval,
    kBlockBasedTableWithPartitionedIndex,
689
    kBlockBasedTableWithPartitionedIndexFormat4,
690
    kPartitionedFilterWithNewTableReaderForCompactions,
691
    kUniversalSubcompactions,
B
Bo Hou 已提交
692
    kxxHash64Checksum,
M
Maysam Yabandeh 已提交
693
    kUnorderedWrite,
694
    // This must be the last line
695
    kEnd,
696 697 698 699 700
  };

 public:
  std::string dbname_;
  std::string alternative_wal_dir_;
S
sdong 已提交
701
  std::string alternative_db_log_dir_;
702
  MockEnv* mem_env_;
E
Ewout Prangsma 已提交
703
  Env* encrypted_env_;
704
  SpecialEnv* env_;
705
  std::shared_ptr<Env> env_guard_;
706 707 708
  DB* db_;
  std::vector<ColumnFamilyHandle*> handles_;

Y
Yi Wu 已提交
709
  int option_config_;
710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725
  Options last_options_;

  // Skip some options, as they may not be applicable to a specific test.
  // To add more skip constants, use values 4, 8, 16, etc.
  enum OptionSkip {
    kNoSkip = 0,
    kSkipDeletesFilterFirst = 1,
    kSkipUniversalCompaction = 2,
    kSkipMergePut = 4,
    kSkipPlainTable = 8,
    kSkipHashIndex = 16,
    kSkipNoSeekToLast = 32,
    kSkipFIFOCompaction = 128,
    kSkipMmapReads = 256,
  };

726 727 728 729 730 731 732
  const int kRangeDelSkipConfigs =
      // Plain tables do not support range deletions.
      kSkipPlainTable |
      // MmapReads disables the iterator pinning that RangeDelAggregator
      // requires.
      kSkipMmapReads;

733 734 735 736
  explicit DBTestBase(const std::string path);

  ~DBTestBase();

737 738 739 740 741 742 743 744 745 746 747 748
  static std::string RandomString(Random* rnd, int len) {
    std::string r;
    test::RandomString(rnd, len, &r);
    return r;
  }

  static std::string Key(int i) {
    char buf[100];
    snprintf(buf, sizeof(buf), "key%06d", i);
    return std::string(buf);
  }

S
sdong 已提交
749 750
  static bool ShouldSkipOptions(int option_config, int skip_mask = kNoSkip);

751 752 753 754
  // Switch to a fresh database with the next option configuration to
  // test.  Return false if there are no more configurations to test.
  bool ChangeOptions(int skip_mask = kNoSkip);

S
Siying Dong 已提交
755
  // Switch between different compaction styles.
756 757
  bool ChangeCompactOptions();

S
Siying Dong 已提交
758 759 760
  // Switch between different WAL-realted options.
  bool ChangeWalOptions();

761 762 763 764
  // Switch between different filter policy
  // Jump from kDefault to kFilter to kFullFilter
  bool ChangeFilterOptions();

765 766 767
  // Switch between different DB options for file ingestion tests.
  bool ChangeOptionsForFileIngestionTest();

768
  // Return the current option configuration.
Y
Yi Wu 已提交
769 770 771 772 773 774 775 776
  Options CurrentOptions(const anon::OptionsOverride& options_override =
                             anon::OptionsOverride()) const;

  Options CurrentOptions(const Options& default_options,
                         const anon::OptionsOverride& options_override =
                             anon::OptionsOverride()) const;

  static Options GetDefaultOptions();
777

Y
Yi Wu 已提交
778 779 780 781
  Options GetOptions(int option_config,
                     const Options& default_options = GetDefaultOptions(),
                     const anon::OptionsOverride& options_override =
                         anon::OptionsOverride()) const;
782

S
sdong 已提交
783
  DBImpl* dbfull() { return reinterpret_cast<DBImpl*>(db_); }
784 785 786 787 788 789 790 791 792 793 794 795 796

  void CreateColumnFamilies(const std::vector<std::string>& cfs,
                            const Options& options);

  void CreateAndReopenWithCF(const std::vector<std::string>& cfs,
                             const Options& options);

  void ReopenWithColumnFamilies(const std::vector<std::string>& cfs,
                                const std::vector<Options>& options);

  void ReopenWithColumnFamilies(const std::vector<std::string>& cfs,
                                const Options& options);

S
sdong 已提交
797 798
  Status TryReopenWithColumnFamilies(const std::vector<std::string>& cfs,
                                     const std::vector<Options>& options);
799 800 801 802 803 804 805 806 807 808

  Status TryReopenWithColumnFamilies(const std::vector<std::string>& cfs,
                                     const Options& options);

  void Reopen(const Options& options);

  void Close();

  void DestroyAndReopen(const Options& options);

809
  void Destroy(const Options& options, bool delete_cf_paths = false);
810 811 812 813 814

  Status ReadOnlyReopen(const Options& options);

  Status TryReopen(const Options& options);

A
Aaron Gao 已提交
815 816
  bool IsDirectIOSupported();

E
Ewout Prangsma 已提交
817 818
  bool IsMemoryMappedAccessSupported() const;

819 820
  Status Flush(int cf = 0);

Y
Yanqin Jin 已提交
821 822
  Status Flush(const std::vector<int>& cf_ids);

823 824 825 826 827
  Status Put(const Slice& k, const Slice& v, WriteOptions wo = WriteOptions());

  Status Put(int cf, const Slice& k, const Slice& v,
             WriteOptions wo = WriteOptions());

828 829 830 831 832 833
  Status Merge(const Slice& k, const Slice& v,
               WriteOptions wo = WriteOptions());

  Status Merge(int cf, const Slice& k, const Slice& v,
               WriteOptions wo = WriteOptions());

834 835 836 837
  Status Delete(const std::string& k);

  Status Delete(int cf, const std::string& k);

A
Andres Noetzli 已提交
838 839 840 841
  Status SingleDelete(const std::string& k);

  Status SingleDelete(int cf, const std::string& k);

842 843
  bool SetPreserveDeletesSequenceNumber(SequenceNumber sn);

844 845 846 847 848
  std::string Get(const std::string& k, const Snapshot* snapshot = nullptr);

  std::string Get(int cf, const std::string& k,
                  const Snapshot* snapshot = nullptr);

849 850
  Status Get(const std::string& k, PinnableSlice* v);

A
Anand Ananthabhotla 已提交
851 852
  std::vector<std::string> MultiGet(std::vector<int> cfs,
                                    const std::vector<std::string>& k,
853 854
                                    const Snapshot* snapshot,
                                    const bool batched);
A
Anand Ananthabhotla 已提交
855

856 857 858
  std::vector<std::string> MultiGet(const std::vector<std::string>& k,
                                    const Snapshot* snapshot = nullptr);

859 860 861 862 863 864 865 866 867 868
  uint64_t GetNumSnapshots();

  uint64_t GetTimeOldestSnapshots();

  // Return a string that contains all key,value pairs in order,
  // formatted like "(k1->v1)(k2->v2)".
  std::string Contents(int cf = 0);

  std::string AllEntriesFor(const Slice& user_key, int cf = 0);

869
#ifndef ROCKSDB_LITE
870 871 872 873 874 875
  int NumSortedRuns(int cf = 0);

  uint64_t TotalSize(int cf = 0);

  uint64_t SizeAtLevel(int level);

V
Vasili Svirski 已提交
876
  size_t TotalLiveFiles(int cf = 0);
877

878 879 880 881
  size_t CountLiveFiles();

  int NumTableFilesAtLevel(int level, int cf = 0);

882 883
  double CompressionRatioAtLevel(int level, int cf = 0);

884
  int TotalTableFiles(int cf = 0, int levels = -1);
Y
Yi Wu 已提交
885
#endif  // ROCKSDB_LITE
886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910

  // Return spread of files per level
  std::string FilesPerLevel(int cf = 0);

  size_t CountFiles();

  uint64_t Size(const Slice& start, const Slice& limit, int cf = 0);

  void Compact(int cf, const Slice& start, const Slice& limit,
               uint32_t target_path_id);

  void Compact(int cf, const Slice& start, const Slice& limit);

  void Compact(const Slice& start, const Slice& limit);

  // Do n memtable compactions, each of which produces an sstable
  // covering the range [small,large].
  void MakeTables(int n, const std::string& small, const std::string& large,
                  int cf = 0);

  // Prevent pushing of new sstables into deeper levels by adding
  // tables that cover a specified range to all levels.
  void FillLevels(const std::string& smallest, const std::string& largest,
                  int cf);

911 912
  void MoveFilesToLevel(int level, int cf = 0);

Y
Yi Wu 已提交
913
#ifndef ROCKSDB_LITE
914
  void DumpFileCounts(const char* label);
Y
Yi Wu 已提交
915
#endif  // ROCKSDB_LITE
916 917 918

  std::string DumpSSTableList();

919 920
  static void GetSstFiles(Env* env, std::string path,
                          std::vector<std::string>* files);
D
dyniusz 已提交
921

922 923 924 925 926
  int GetSstFileCount(std::string path);

  // this will generate non-overlapping files since it keeps increasing key_idx
  void GenerateNewFile(Random* rnd, int* key_idx, bool nowait = false);

927 928
  void GenerateNewFile(int fd, Random* rnd, int* key_idx, bool nowait = false);

929
  static const int kNumKeysByGenerateNewRandomFile;
K
krad 已提交
930
  static const int KNumKeysByGenerateNewFile = 100;
931

932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947
  void GenerateNewRandomFile(Random* rnd, bool nowait = false);

  std::string IterStatus(Iterator* iter);

  Options OptionsForLogIterTest();

  std::string DummyString(size_t len, char c = 'a');

  void VerifyIterLast(std::string expected_key, int cf = 0);

  // Used to test InplaceUpdate

  // If previous value is nullptr or delta is > than previous value,
  //   sets newValue with delta
  // If previous value is not empty,
  //   updates previous value with 'b' string of previous value size - 1.
S
sdong 已提交
948 949 950
  static UpdateStatus updateInPlaceSmallerSize(char* prevValue,
                                               uint32_t* prevSize, Slice delta,
                                               std::string* newValue);
951

S
sdong 已提交
952 953 954 955
  static UpdateStatus updateInPlaceSmallerVarintSize(char* prevValue,
                                                     uint32_t* prevSize,
                                                     Slice delta,
                                                     std::string* newValue);
956

S
sdong 已提交
957 958 959
  static UpdateStatus updateInPlaceLargerSize(char* prevValue,
                                              uint32_t* prevSize, Slice delta,
                                              std::string* newValue);
960

S
sdong 已提交
961 962
  static UpdateStatus updateInPlaceNoAction(char* prevValue, uint32_t* prevSize,
                                            Slice delta, std::string* newValue);
963 964 965 966 967 968

  // Utility method to test InplaceUpdate
  void validateNumberOfEntries(int numValues, int cf = 0);

  void CopyFile(const std::string& source, const std::string& destination,
                uint64_t size = 0);
969

970 971
  std::unordered_map<std::string, uint64_t> GetAllSSTFiles(
      uint64_t* total_size = nullptr);
Y
Yi Wu 已提交
972 973 974

  std::vector<std::uint64_t> ListTableFiles(Env* env, const std::string& path);

I
Islam AbdelRahman 已提交
975 976 977 978
  void VerifyDBFromMap(
      std::map<std::string, std::string> true_data,
      size_t* total_reads_res = nullptr, bool tailing_iter = false,
      std::map<std::string, Status> status = std::map<std::string, Status>());
979 980 981

  void VerifyDBInternal(
      std::vector<std::pair<std::string, std::string>> true_data);
982

Y
Yi Wu 已提交
983 984 985 986 987 988 989 990
#ifndef ROCKSDB_LITE
  uint64_t GetNumberOfSstFilesForColumnFamily(DB* db,
                                              std::string column_family_name);
#endif  // ROCKSDB_LITE

  uint64_t TestGetTickerCount(const Options& options, Tickers ticker_type) {
    return options.statistics->getTickerCount(ticker_type);
  }
991 992 993 994 995

  uint64_t TestGetAndResetTickerCount(const Options& options,
                                      Tickers ticker_type) {
    return options.statistics->getAndResetTickerCount(ticker_type);
  }
996 997 998
};

}  // namespace rocksdb