You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by je...@apache.org on 2017/10/09 16:24:57 UTC
[17/51] [partial] nifi-minifi-cpp git commit: MINIFI-372: Replace
leveldb with RocksDB
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/48867732/thirdparty/rocksdb/db/db_test_util.cc
----------------------------------------------------------------------
diff --git a/thirdparty/rocksdb/db/db_test_util.cc b/thirdparty/rocksdb/db/db_test_util.cc
new file mode 100644
index 0000000..c4d465b
--- /dev/null
+++ b/thirdparty/rocksdb/db/db_test_util.cc
@@ -0,0 +1,1395 @@
+// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
+// This source code is licensed under both the GPLv2 (found in the
+// COPYING file in the root directory) and Apache 2.0 License
+// (found in the LICENSE.Apache file in the root directory).
+//
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#include "db/db_test_util.h"
+#include "db/forward_iterator.h"
+#include "rocksdb/env_encryption.h"
+
+namespace rocksdb {
+
+// Special Env used to delay background operations
+
+SpecialEnv::SpecialEnv(Env* base)
+ : EnvWrapper(base),
+ rnd_(301),
+ sleep_counter_(this),
+ addon_time_(0),
+ time_elapse_only_sleep_(false),
+ no_slowdown_(false) {
+ delay_sstable_sync_.store(false, std::memory_order_release);
+ drop_writes_.store(false, std::memory_order_release);
+ no_space_.store(false, std::memory_order_release);
+ non_writable_.store(false, std::memory_order_release);
+ count_random_reads_ = false;
+ count_sequential_reads_ = false;
+ manifest_sync_error_.store(false, std::memory_order_release);
+ manifest_write_error_.store(false, std::memory_order_release);
+ log_write_error_.store(false, std::memory_order_release);
+ random_file_open_counter_.store(0, std::memory_order_relaxed);
+ delete_count_.store(0, std::memory_order_relaxed);
+ num_open_wal_file_.store(0);
+ log_write_slowdown_ = 0;
+ bytes_written_ = 0;
+ sync_counter_ = 0;
+ non_writeable_rate_ = 0;
+ new_writable_count_ = 0;
+ non_writable_count_ = 0;
+ table_write_callback_ = nullptr;
+}
+#ifndef ROCKSDB_LITE
+ROT13BlockCipher rot13Cipher_(16);
+#endif // ROCKSDB_LITE
+
+DBTestBase::DBTestBase(const std::string path)
+ : mem_env_(!getenv("MEM_ENV") ? nullptr : new MockEnv(Env::Default())),
+#ifndef ROCKSDB_LITE
+ encrypted_env_(
+ !getenv("ENCRYPTED_ENV")
+ ? nullptr
+ : NewEncryptedEnv(mem_env_ ? mem_env_ : Env::Default(),
+ new CTREncryptionProvider(rot13Cipher_))),
+#else
+ encrypted_env_(nullptr),
+#endif // ROCKSDB_LITE
+ env_(new SpecialEnv(encrypted_env_
+ ? encrypted_env_
+ : (mem_env_ ? mem_env_ : Env::Default()))),
+ option_config_(kDefault) {
+ env_->SetBackgroundThreads(1, Env::LOW);
+ env_->SetBackgroundThreads(1, Env::HIGH);
+ dbname_ = test::TmpDir(env_) + path;
+ alternative_wal_dir_ = dbname_ + "/wal";
+ alternative_db_log_dir_ = dbname_ + "/db_log_dir";
+ auto options = CurrentOptions();
+ options.env = env_;
+ auto delete_options = options;
+ delete_options.wal_dir = alternative_wal_dir_;
+ EXPECT_OK(DestroyDB(dbname_, delete_options));
+ // Destroy it for not alternative WAL dir is used.
+ EXPECT_OK(DestroyDB(dbname_, options));
+ db_ = nullptr;
+ Reopen(options);
+ Random::GetTLSInstance()->Reset(0xdeadbeef);
+}
+
+DBTestBase::~DBTestBase() {
+ rocksdb::SyncPoint::GetInstance()->DisableProcessing();
+ rocksdb::SyncPoint::GetInstance()->LoadDependency({});
+ rocksdb::SyncPoint::GetInstance()->ClearAllCallBacks();
+ Close();
+ Options options;
+ options.db_paths.emplace_back(dbname_, 0);
+ options.db_paths.emplace_back(dbname_ + "_2", 0);
+ options.db_paths.emplace_back(dbname_ + "_3", 0);
+ options.db_paths.emplace_back(dbname_ + "_4", 0);
+ options.env = env_;
+
+ if (getenv("KEEP_DB")) {
+ printf("DB is still at %s\n", dbname_.c_str());
+ } else {
+ EXPECT_OK(DestroyDB(dbname_, options));
+ }
+ delete env_;
+}
+
+bool DBTestBase::ShouldSkipOptions(int option_config, int skip_mask) {
+#ifdef ROCKSDB_LITE
+ // These options are not supported in ROCKSDB_LITE
+ if (option_config == kHashSkipList ||
+ option_config == kPlainTableFirstBytePrefix ||
+ option_config == kPlainTableCappedPrefix ||
+ option_config == kPlainTableCappedPrefixNonMmap ||
+ option_config == kPlainTableAllBytesPrefix ||
+ option_config == kVectorRep || option_config == kHashLinkList ||
+ option_config == kHashCuckoo || option_config == kUniversalCompaction ||
+ option_config == kUniversalCompactionMultiLevel ||
+ option_config == kUniversalSubcompactions ||
+ option_config == kFIFOCompaction ||
+ option_config == kConcurrentSkipList) {
+ return true;
+ }
+#endif
+
+ if ((skip_mask & kSkipUniversalCompaction) &&
+ (option_config == kUniversalCompaction ||
+ option_config == kUniversalCompactionMultiLevel)) {
+ return true;
+ }
+ if ((skip_mask & kSkipMergePut) && option_config == kMergePut) {
+ return true;
+ }
+ if ((skip_mask & kSkipNoSeekToLast) &&
+ (option_config == kHashLinkList || option_config == kHashSkipList)) {
+ return true;
+ }
+ if ((skip_mask & kSkipPlainTable) &&
+ (option_config == kPlainTableAllBytesPrefix ||
+ option_config == kPlainTableFirstBytePrefix ||
+ option_config == kPlainTableCappedPrefix ||
+ option_config == kPlainTableCappedPrefixNonMmap)) {
+ return true;
+ }
+ if ((skip_mask & kSkipHashIndex) &&
+ (option_config == kBlockBasedTableWithPrefixHashIndex ||
+ option_config == kBlockBasedTableWithWholeKeyHashIndex)) {
+ return true;
+ }
+ if ((skip_mask & kSkipHashCuckoo) && (option_config == kHashCuckoo)) {
+ return true;
+ }
+ if ((skip_mask & kSkipFIFOCompaction) && option_config == kFIFOCompaction) {
+ return true;
+ }
+ if ((skip_mask & kSkipMmapReads) && option_config == kWalDirAndMmapReads) {
+ return true;
+ }
+ return false;
+}
+
+// Switch to a fresh database with the next option configuration to
+// test. Return false if there are no more configurations to test.
+bool DBTestBase::ChangeOptions(int skip_mask) {
+ for (option_config_++; option_config_ < kEnd; option_config_++) {
+ if (ShouldSkipOptions(option_config_, skip_mask)) {
+ continue;
+ }
+ break;
+ }
+
+ if (option_config_ >= kEnd) {
+ Destroy(last_options_);
+ return false;
+ } else {
+ auto options = CurrentOptions();
+ options.create_if_missing = true;
+ DestroyAndReopen(options);
+ return true;
+ }
+}
+
+// Switch between different compaction styles.
+bool DBTestBase::ChangeCompactOptions() {
+ if (option_config_ == kDefault) {
+ option_config_ = kUniversalCompaction;
+ Destroy(last_options_);
+ auto options = CurrentOptions();
+ options.create_if_missing = true;
+ TryReopen(options);
+ return true;
+ } else if (option_config_ == kUniversalCompaction) {
+ option_config_ = kUniversalCompactionMultiLevel;
+ Destroy(last_options_);
+ auto options = CurrentOptions();
+ options.create_if_missing = true;
+ TryReopen(options);
+ return true;
+ } else if (option_config_ == kUniversalCompactionMultiLevel) {
+ option_config_ = kLevelSubcompactions;
+ Destroy(last_options_);
+ auto options = CurrentOptions();
+ assert(options.max_subcompactions > 1);
+ TryReopen(options);
+ return true;
+ } else if (option_config_ == kLevelSubcompactions) {
+ option_config_ = kUniversalSubcompactions;
+ Destroy(last_options_);
+ auto options = CurrentOptions();
+ assert(options.max_subcompactions > 1);
+ TryReopen(options);
+ return true;
+ } else {
+ return false;
+ }
+}
+
+// Switch between different WAL settings
+bool DBTestBase::ChangeWalOptions() {
+ if (option_config_ == kDefault) {
+ option_config_ = kDBLogDir;
+ Destroy(last_options_);
+ auto options = CurrentOptions();
+ Destroy(options);
+ options.create_if_missing = true;
+ TryReopen(options);
+ return true;
+ } else if (option_config_ == kDBLogDir) {
+ option_config_ = kWalDirAndMmapReads;
+ Destroy(last_options_);
+ auto options = CurrentOptions();
+ Destroy(options);
+ options.create_if_missing = true;
+ TryReopen(options);
+ return true;
+ } else if (option_config_ == kWalDirAndMmapReads) {
+ option_config_ = kRecycleLogFiles;
+ Destroy(last_options_);
+ auto options = CurrentOptions();
+ Destroy(options);
+ TryReopen(options);
+ return true;
+ } else {
+ return false;
+ }
+}
+
+// Switch between different filter policy
+// Jump from kDefault to kFilter to kFullFilter
+bool DBTestBase::ChangeFilterOptions() {
+ if (option_config_ == kDefault) {
+ option_config_ = kFilter;
+ } else if (option_config_ == kFilter) {
+ option_config_ = kFullFilterWithNewTableReaderForCompactions;
+ } else if (option_config_ == kFullFilterWithNewTableReaderForCompactions) {
+ option_config_ = kPartitionedFilterWithNewTableReaderForCompactions;
+ } else {
+ return false;
+ }
+ Destroy(last_options_);
+
+ auto options = CurrentOptions();
+ options.create_if_missing = true;
+ TryReopen(options);
+ return true;
+}
+
+// Return the current option configuration.
+Options DBTestBase::CurrentOptions(
+ const anon::OptionsOverride& options_override) const {
+ return GetOptions(option_config_, GetDefaultOptions(), options_override);
+}
+
+Options DBTestBase::CurrentOptions(
+ const Options& default_options,
+ const anon::OptionsOverride& options_override) const {
+ return GetOptions(option_config_, default_options, options_override);
+}
+
+Options DBTestBase::GetDefaultOptions() {
+ Options options;
+ options.write_buffer_size = 4090 * 4096;
+ options.target_file_size_base = 2 * 1024 * 1024;
+ options.max_bytes_for_level_base = 10 * 1024 * 1024;
+ options.max_open_files = 5000;
+ options.wal_recovery_mode = WALRecoveryMode::kTolerateCorruptedTailRecords;
+ options.compaction_pri = CompactionPri::kByCompensatedSize;
+ return options;
+}
+
+Options DBTestBase::GetOptions(
+ int option_config, const Options& default_options,
+ const anon::OptionsOverride& options_override) const {
+ // this redundant copy is to minimize code change w/o having lint error.
+ Options options = default_options;
+ BlockBasedTableOptions table_options;
+ bool set_block_based_table_factory = true;
+#if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && \
+ !defined(OS_AIX)
+ rocksdb::SyncPoint::GetInstance()->ClearCallBack(
+ "NewRandomAccessFile:O_DIRECT");
+ rocksdb::SyncPoint::GetInstance()->ClearCallBack(
+ "NewWritableFile:O_DIRECT");
+#endif
+
+ bool can_allow_mmap = IsMemoryMappedAccessSupported();
+ switch (option_config) {
+#ifndef ROCKSDB_LITE
+ case kHashSkipList:
+ options.prefix_extractor.reset(NewFixedPrefixTransform(1));
+ options.memtable_factory.reset(NewHashSkipListRepFactory(16));
+ options.allow_concurrent_memtable_write = false;
+ break;
+ case kPlainTableFirstBytePrefix:
+ options.table_factory.reset(new PlainTableFactory());
+ options.prefix_extractor.reset(NewFixedPrefixTransform(1));
+ options.allow_mmap_reads = can_allow_mmap;
+ options.max_sequential_skip_in_iterations = 999999;
+ set_block_based_table_factory = false;
+ break;
+ case kPlainTableCappedPrefix:
+ options.table_factory.reset(new PlainTableFactory());
+ options.prefix_extractor.reset(NewCappedPrefixTransform(8));
+ options.allow_mmap_reads = can_allow_mmap;
+ options.max_sequential_skip_in_iterations = 999999;
+ set_block_based_table_factory = false;
+ break;
+ case kPlainTableCappedPrefixNonMmap:
+ options.table_factory.reset(new PlainTableFactory());
+ options.prefix_extractor.reset(NewCappedPrefixTransform(8));
+ options.allow_mmap_reads = false;
+ options.max_sequential_skip_in_iterations = 999999;
+ set_block_based_table_factory = false;
+ break;
+ case kPlainTableAllBytesPrefix:
+ options.table_factory.reset(new PlainTableFactory());
+ options.prefix_extractor.reset(NewNoopTransform());
+ options.allow_mmap_reads = can_allow_mmap;
+ options.max_sequential_skip_in_iterations = 999999;
+ set_block_based_table_factory = false;
+ break;
+ case kVectorRep:
+ options.memtable_factory.reset(new VectorRepFactory(100));
+ options.allow_concurrent_memtable_write = false;
+ break;
+ case kHashLinkList:
+ options.prefix_extractor.reset(NewFixedPrefixTransform(1));
+ options.memtable_factory.reset(
+ NewHashLinkListRepFactory(4, 0, 3, true, 4));
+ options.allow_concurrent_memtable_write = false;
+ break;
+ case kHashCuckoo:
+ options.memtable_factory.reset(
+ NewHashCuckooRepFactory(options.write_buffer_size));
+ options.allow_concurrent_memtable_write = false;
+ break;
+#endif // ROCKSDB_LITE
+ case kMergePut:
+ options.merge_operator = MergeOperators::CreatePutOperator();
+ break;
+ case kFilter:
+ table_options.filter_policy.reset(NewBloomFilterPolicy(10, true));
+ break;
+ case kFullFilterWithNewTableReaderForCompactions:
+ table_options.filter_policy.reset(NewBloomFilterPolicy(10, false));
+ options.new_table_reader_for_compaction_inputs = true;
+ options.compaction_readahead_size = 10 * 1024 * 1024;
+ break;
+ case kPartitionedFilterWithNewTableReaderForCompactions:
+ table_options.filter_policy.reset(NewBloomFilterPolicy(10, false));
+ table_options.partition_filters = true;
+ table_options.index_type =
+ BlockBasedTableOptions::IndexType::kTwoLevelIndexSearch;
+ options.new_table_reader_for_compaction_inputs = true;
+ options.compaction_readahead_size = 10 * 1024 * 1024;
+ break;
+ case kUncompressed:
+ options.compression = kNoCompression;
+ break;
+ case kNumLevel_3:
+ options.num_levels = 3;
+ break;
+ case kDBLogDir:
+ options.db_log_dir = alternative_db_log_dir_;
+ break;
+ case kWalDirAndMmapReads:
+ options.wal_dir = alternative_wal_dir_;
+ // mmap reads should be orthogonal to WalDir setting, so we piggyback to
+ // this option config to test mmap reads as well
+ options.allow_mmap_reads = can_allow_mmap;
+ break;
+ case kManifestFileSize:
+ options.max_manifest_file_size = 50; // 50 bytes
+ break;
+ case kPerfOptions:
+ options.soft_rate_limit = 2.0;
+ options.delayed_write_rate = 8 * 1024 * 1024;
+ options.report_bg_io_stats = true;
+ // TODO(3.13) -- test more options
+ break;
+ case kUniversalCompaction:
+ options.compaction_style = kCompactionStyleUniversal;
+ options.num_levels = 1;
+ break;
+ case kUniversalCompactionMultiLevel:
+ options.compaction_style = kCompactionStyleUniversal;
+ options.num_levels = 8;
+ break;
+ case kCompressedBlockCache:
+ options.allow_mmap_writes = can_allow_mmap;
+ table_options.block_cache_compressed = NewLRUCache(8 * 1024 * 1024);
+ break;
+ case kInfiniteMaxOpenFiles:
+ options.max_open_files = -1;
+ break;
+ case kxxHashChecksum: {
+ table_options.checksum = kxxHash;
+ break;
+ }
+ case kFIFOCompaction: {
+ options.compaction_style = kCompactionStyleFIFO;
+ break;
+ }
+ case kBlockBasedTableWithPrefixHashIndex: {
+ table_options.index_type = BlockBasedTableOptions::kHashSearch;
+ options.prefix_extractor.reset(NewFixedPrefixTransform(1));
+ break;
+ }
+ case kBlockBasedTableWithWholeKeyHashIndex: {
+ table_options.index_type = BlockBasedTableOptions::kHashSearch;
+ options.prefix_extractor.reset(NewNoopTransform());
+ break;
+ }
+ case kBlockBasedTableWithPartitionedIndex: {
+ table_options.index_type = BlockBasedTableOptions::kTwoLevelIndexSearch;
+ options.prefix_extractor.reset(NewNoopTransform());
+ break;
+ }
+ case kBlockBasedTableWithIndexRestartInterval: {
+ table_options.index_block_restart_interval = 8;
+ break;
+ }
+ case kOptimizeFiltersForHits: {
+ options.optimize_filters_for_hits = true;
+ set_block_based_table_factory = true;
+ break;
+ }
+ case kRowCache: {
+ options.row_cache = NewLRUCache(1024 * 1024);
+ break;
+ }
+ case kRecycleLogFiles: {
+ options.recycle_log_file_num = 2;
+ break;
+ }
+ case kLevelSubcompactions: {
+ options.max_subcompactions = 4;
+ break;
+ }
+ case kUniversalSubcompactions: {
+ options.compaction_style = kCompactionStyleUniversal;
+ options.num_levels = 8;
+ options.max_subcompactions = 4;
+ break;
+ }
+ case kConcurrentSkipList: {
+ options.allow_concurrent_memtable_write = true;
+ options.enable_write_thread_adaptive_yield = true;
+ break;
+ }
+ case kDirectIO: {
+ options.use_direct_reads = true;
+ options.use_direct_io_for_flush_and_compaction = true;
+ options.compaction_readahead_size = 2 * 1024 * 1024;
+#if !defined(OS_MACOSX) && !defined(OS_WIN) && !defined(OS_SOLARIS) && \
+ !defined(OS_AIX)
+ rocksdb::SyncPoint::GetInstance()->SetCallBack(
+ "NewWritableFile:O_DIRECT", [&](void* arg) {
+ int* val = static_cast<int*>(arg);
+ *val &= ~O_DIRECT;
+ });
+ rocksdb::SyncPoint::GetInstance()->SetCallBack(
+ "NewRandomAccessFile:O_DIRECT", [&](void* arg) {
+ int* val = static_cast<int*>(arg);
+ *val &= ~O_DIRECT;
+ });
+ rocksdb::SyncPoint::GetInstance()->EnableProcessing();
+#endif
+ break;
+ }
+ case kPipelinedWrite: {
+ options.enable_pipelined_write = true;
+ break;
+ }
+ case kConcurrentWALWrites: {
+ // This options optimize 2PC commit path
+ options.concurrent_prepare = true;
+ options.manual_wal_flush = true;
+ break;
+ }
+
+ default:
+ break;
+ }
+
+ if (options_override.filter_policy) {
+ table_options.filter_policy = options_override.filter_policy;
+ table_options.partition_filters = options_override.partition_filters;
+ table_options.metadata_block_size = options_override.metadata_block_size;
+ }
+ if (set_block_based_table_factory) {
+ options.table_factory.reset(NewBlockBasedTableFactory(table_options));
+ }
+ options.env = env_;
+ options.create_if_missing = true;
+ options.fail_if_options_file_error = true;
+ return options;
+}
+
+void DBTestBase::CreateColumnFamilies(const std::vector<std::string>& cfs,
+ const Options& options) {
+ ColumnFamilyOptions cf_opts(options);
+ size_t cfi = handles_.size();
+ handles_.resize(cfi + cfs.size());
+ for (auto cf : cfs) {
+ ASSERT_OK(db_->CreateColumnFamily(cf_opts, cf, &handles_[cfi++]));
+ }
+}
+
+void DBTestBase::CreateAndReopenWithCF(const std::vector<std::string>& cfs,
+ const Options& options) {
+ CreateColumnFamilies(cfs, options);
+ std::vector<std::string> cfs_plus_default = cfs;
+ cfs_plus_default.insert(cfs_plus_default.begin(), kDefaultColumnFamilyName);
+ ReopenWithColumnFamilies(cfs_plus_default, options);
+}
+
+void DBTestBase::ReopenWithColumnFamilies(const std::vector<std::string>& cfs,
+ const std::vector<Options>& options) {
+ ASSERT_OK(TryReopenWithColumnFamilies(cfs, options));
+}
+
+void DBTestBase::ReopenWithColumnFamilies(const std::vector<std::string>& cfs,
+ const Options& options) {
+ ASSERT_OK(TryReopenWithColumnFamilies(cfs, options));
+}
+
+Status DBTestBase::TryReopenWithColumnFamilies(
+ const std::vector<std::string>& cfs, const std::vector<Options>& options) {
+ Close();
+ EXPECT_EQ(cfs.size(), options.size());
+ std::vector<ColumnFamilyDescriptor> column_families;
+ for (size_t i = 0; i < cfs.size(); ++i) {
+ column_families.push_back(ColumnFamilyDescriptor(cfs[i], options[i]));
+ }
+ DBOptions db_opts = DBOptions(options[0]);
+ return DB::Open(db_opts, dbname_, column_families, &handles_, &db_);
+}
+
+Status DBTestBase::TryReopenWithColumnFamilies(
+ const std::vector<std::string>& cfs, const Options& options) {
+ Close();
+ std::vector<Options> v_opts(cfs.size(), options);
+ return TryReopenWithColumnFamilies(cfs, v_opts);
+}
+
+void DBTestBase::Reopen(const Options& options) {
+ ASSERT_OK(TryReopen(options));
+}
+
+void DBTestBase::Close() {
+ for (auto h : handles_) {
+ db_->DestroyColumnFamilyHandle(h);
+ }
+ handles_.clear();
+ delete db_;
+ db_ = nullptr;
+}
+
+void DBTestBase::DestroyAndReopen(const Options& options) {
+ // Destroy using last options
+ Destroy(last_options_);
+ ASSERT_OK(TryReopen(options));
+}
+
+void DBTestBase::Destroy(const Options& options) {
+ Close();
+ ASSERT_OK(DestroyDB(dbname_, options));
+}
+
+Status DBTestBase::ReadOnlyReopen(const Options& options) {
+ return DB::OpenForReadOnly(options, dbname_, &db_);
+}
+
+Status DBTestBase::TryReopen(const Options& options) {
+ Close();
+ last_options_.table_factory.reset();
+ // Note: operator= is an unsafe approach here since it destructs shared_ptr in
+ // the same order of their creation, in contrast to destructors which
+ // destructs them in the opposite order of creation. One particular problme is
+ // that the cache destructor might invoke callback functions that use Option
+ // members such as statistics. To work around this problem, we manually call
+ // destructor of table_facotry which eventually clears the block cache.
+ last_options_ = options;
+ return DB::Open(options, dbname_, &db_);
+}
+
+bool DBTestBase::IsDirectIOSupported() {
+ EnvOptions env_options;
+ env_options.use_mmap_writes = false;
+ env_options.use_direct_writes = true;
+ std::string tmp = TempFileName(dbname_, 999);
+ Status s;
+ {
+ unique_ptr<WritableFile> file;
+ s = env_->NewWritableFile(tmp, &file, env_options);
+ }
+ if (s.ok()) {
+ s = env_->DeleteFile(tmp);
+ }
+ return s.ok();
+}
+
+bool DBTestBase::IsMemoryMappedAccessSupported() const {
+ return (!encrypted_env_);
+}
+
+Status DBTestBase::Flush(int cf) {
+ if (cf == 0) {
+ return db_->Flush(FlushOptions());
+ } else {
+ return db_->Flush(FlushOptions(), handles_[cf]);
+ }
+}
+
+Status DBTestBase::Put(const Slice& k, const Slice& v, WriteOptions wo) {
+ if (kMergePut == option_config_) {
+ return db_->Merge(wo, k, v);
+ } else {
+ return db_->Put(wo, k, v);
+ }
+}
+
+Status DBTestBase::Put(int cf, const Slice& k, const Slice& v,
+ WriteOptions wo) {
+ if (kMergePut == option_config_) {
+ return db_->Merge(wo, handles_[cf], k, v);
+ } else {
+ return db_->Put(wo, handles_[cf], k, v);
+ }
+}
+
+Status DBTestBase::Merge(const Slice& k, const Slice& v, WriteOptions wo) {
+ return db_->Merge(wo, k, v);
+}
+
+Status DBTestBase::Merge(int cf, const Slice& k, const Slice& v,
+ WriteOptions wo) {
+ return db_->Merge(wo, handles_[cf], k, v);
+}
+
+Status DBTestBase::Delete(const std::string& k) {
+ return db_->Delete(WriteOptions(), k);
+}
+
+Status DBTestBase::Delete(int cf, const std::string& k) {
+ return db_->Delete(WriteOptions(), handles_[cf], k);
+}
+
+Status DBTestBase::SingleDelete(const std::string& k) {
+ return db_->SingleDelete(WriteOptions(), k);
+}
+
+Status DBTestBase::SingleDelete(int cf, const std::string& k) {
+ return db_->SingleDelete(WriteOptions(), handles_[cf], k);
+}
+
+std::string DBTestBase::Get(const std::string& k, const Snapshot* snapshot) {
+ ReadOptions options;
+ options.verify_checksums = true;
+ options.snapshot = snapshot;
+ std::string result;
+ Status s = db_->Get(options, k, &result);
+ if (s.IsNotFound()) {
+ result = "NOT_FOUND";
+ } else if (!s.ok()) {
+ result = s.ToString();
+ }
+ return result;
+}
+
+std::string DBTestBase::Get(int cf, const std::string& k,
+ const Snapshot* snapshot) {
+ ReadOptions options;
+ options.verify_checksums = true;
+ options.snapshot = snapshot;
+ std::string result;
+ Status s = db_->Get(options, handles_[cf], k, &result);
+ if (s.IsNotFound()) {
+ result = "NOT_FOUND";
+ } else if (!s.ok()) {
+ result = s.ToString();
+ }
+ return result;
+}
+
+Status DBTestBase::Get(const std::string& k, PinnableSlice* v) {
+ ReadOptions options;
+ options.verify_checksums = true;
+ Status s = dbfull()->Get(options, dbfull()->DefaultColumnFamily(), k, v);
+ return s;
+}
+
+uint64_t DBTestBase::GetNumSnapshots() {
+ uint64_t int_num;
+ EXPECT_TRUE(dbfull()->GetIntProperty("rocksdb.num-snapshots", &int_num));
+ return int_num;
+}
+
+uint64_t DBTestBase::GetTimeOldestSnapshots() {
+ uint64_t int_num;
+ EXPECT_TRUE(
+ dbfull()->GetIntProperty("rocksdb.oldest-snapshot-time", &int_num));
+ return int_num;
+}
+
+// Return a string that contains all key,value pairs in order,
+// formatted like "(k1->v1)(k2->v2)".
+std::string DBTestBase::Contents(int cf) {
+ std::vector<std::string> forward;
+ std::string result;
+ Iterator* iter = (cf == 0) ? db_->NewIterator(ReadOptions())
+ : db_->NewIterator(ReadOptions(), handles_[cf]);
+ for (iter->SeekToFirst(); iter->Valid(); iter->Next()) {
+ std::string s = IterStatus(iter);
+ result.push_back('(');
+ result.append(s);
+ result.push_back(')');
+ forward.push_back(s);
+ }
+
+ // Check reverse iteration results are the reverse of forward results
+ unsigned int matched = 0;
+ for (iter->SeekToLast(); iter->Valid(); iter->Prev()) {
+ EXPECT_LT(matched, forward.size());
+ EXPECT_EQ(IterStatus(iter), forward[forward.size() - matched - 1]);
+ matched++;
+ }
+ EXPECT_EQ(matched, forward.size());
+
+ delete iter;
+ return result;
+}
+
+std::string DBTestBase::AllEntriesFor(const Slice& user_key, int cf) {
+ Arena arena;
+ auto options = CurrentOptions();
+ InternalKeyComparator icmp(options.comparator);
+ RangeDelAggregator range_del_agg(icmp, {} /* snapshots */);
+ ScopedArenaIterator iter;
+ if (cf == 0) {
+ iter.set(dbfull()->NewInternalIterator(&arena, &range_del_agg));
+ } else {
+ iter.set(
+ dbfull()->NewInternalIterator(&arena, &range_del_agg, handles_[cf]));
+ }
+ InternalKey target(user_key, kMaxSequenceNumber, kTypeValue);
+ iter->Seek(target.Encode());
+ std::string result;
+ if (!iter->status().ok()) {
+ result = iter->status().ToString();
+ } else {
+ result = "[ ";
+ bool first = true;
+ while (iter->Valid()) {
+ ParsedInternalKey ikey(Slice(), 0, kTypeValue);
+ if (!ParseInternalKey(iter->key(), &ikey)) {
+ result += "CORRUPTED";
+ } else {
+ if (!last_options_.comparator->Equal(ikey.user_key, user_key)) {
+ break;
+ }
+ if (!first) {
+ result += ", ";
+ }
+ first = false;
+ switch (ikey.type) {
+ case kTypeValue:
+ result += iter->value().ToString();
+ break;
+ case kTypeMerge:
+ // keep it the same as kTypeValue for testing kMergePut
+ result += iter->value().ToString();
+ break;
+ case kTypeDeletion:
+ result += "DEL";
+ break;
+ case kTypeSingleDeletion:
+ result += "SDEL";
+ break;
+ default:
+ assert(false);
+ break;
+ }
+ }
+ iter->Next();
+ }
+ if (!first) {
+ result += " ";
+ }
+ result += "]";
+ }
+ return result;
+}
+
+#ifndef ROCKSDB_LITE
+int DBTestBase::NumSortedRuns(int cf) {
+ ColumnFamilyMetaData cf_meta;
+ if (cf == 0) {
+ db_->GetColumnFamilyMetaData(&cf_meta);
+ } else {
+ db_->GetColumnFamilyMetaData(handles_[cf], &cf_meta);
+ }
+ int num_sr = static_cast<int>(cf_meta.levels[0].files.size());
+ for (size_t i = 1U; i < cf_meta.levels.size(); i++) {
+ if (cf_meta.levels[i].files.size() > 0) {
+ num_sr++;
+ }
+ }
+ return num_sr;
+}
+
+uint64_t DBTestBase::TotalSize(int cf) {
+ ColumnFamilyMetaData cf_meta;
+ if (cf == 0) {
+ db_->GetColumnFamilyMetaData(&cf_meta);
+ } else {
+ db_->GetColumnFamilyMetaData(handles_[cf], &cf_meta);
+ }
+ return cf_meta.size;
+}
+
+uint64_t DBTestBase::SizeAtLevel(int level) {
+ std::vector<LiveFileMetaData> metadata;
+ db_->GetLiveFilesMetaData(&metadata);
+ uint64_t sum = 0;
+ for (const auto& m : metadata) {
+ if (m.level == level) {
+ sum += m.size;
+ }
+ }
+ return sum;
+}
+
+size_t DBTestBase::TotalLiveFiles(int cf) {
+ ColumnFamilyMetaData cf_meta;
+ if (cf == 0) {
+ db_->GetColumnFamilyMetaData(&cf_meta);
+ } else {
+ db_->GetColumnFamilyMetaData(handles_[cf], &cf_meta);
+ }
+ size_t num_files = 0;
+ for (auto& level : cf_meta.levels) {
+ num_files += level.files.size();
+ }
+ return num_files;
+}
+
+size_t DBTestBase::CountLiveFiles() {
+ std::vector<LiveFileMetaData> metadata;
+ db_->GetLiveFilesMetaData(&metadata);
+ return metadata.size();
+}
+#endif // ROCKSDB_LITE
+
+int DBTestBase::NumTableFilesAtLevel(int level, int cf) {
+ std::string property;
+ if (cf == 0) {
+ // default cfd
+ EXPECT_TRUE(db_->GetProperty(
+ "rocksdb.num-files-at-level" + NumberToString(level), &property));
+ } else {
+ EXPECT_TRUE(db_->GetProperty(
+ handles_[cf], "rocksdb.num-files-at-level" + NumberToString(level),
+ &property));
+ }
+ return atoi(property.c_str());
+}
+
+double DBTestBase::CompressionRatioAtLevel(int level, int cf) {
+ std::string property;
+ if (cf == 0) {
+ // default cfd
+ EXPECT_TRUE(db_->GetProperty(
+ "rocksdb.compression-ratio-at-level" + NumberToString(level),
+ &property));
+ } else {
+ EXPECT_TRUE(db_->GetProperty(
+ handles_[cf],
+ "rocksdb.compression-ratio-at-level" + NumberToString(level),
+ &property));
+ }
+ return std::stod(property);
+}
+
+int DBTestBase::TotalTableFiles(int cf, int levels) {
+ if (levels == -1) {
+ levels = (cf == 0) ? db_->NumberLevels() : db_->NumberLevels(handles_[1]);
+ }
+ int result = 0;
+ for (int level = 0; level < levels; level++) {
+ result += NumTableFilesAtLevel(level, cf);
+ }
+ return result;
+}
+
+// Return spread of files per level
+std::string DBTestBase::FilesPerLevel(int cf) {
+ int num_levels =
+ (cf == 0) ? db_->NumberLevels() : db_->NumberLevels(handles_[1]);
+ std::string result;
+ size_t last_non_zero_offset = 0;
+ for (int level = 0; level < num_levels; level++) {
+ int f = NumTableFilesAtLevel(level, cf);
+ char buf[100];
+ snprintf(buf, sizeof(buf), "%s%d", (level ? "," : ""), f);
+ result += buf;
+ if (f > 0) {
+ last_non_zero_offset = result.size();
+ }
+ }
+ result.resize(last_non_zero_offset);
+ return result;
+}
+
+size_t DBTestBase::CountFiles() {
+ std::vector<std::string> files;
+ env_->GetChildren(dbname_, &files);
+
+ std::vector<std::string> logfiles;
+ if (dbname_ != last_options_.wal_dir) {
+ env_->GetChildren(last_options_.wal_dir, &logfiles);
+ }
+
+ return files.size() + logfiles.size();
+}
+
+uint64_t DBTestBase::Size(const Slice& start, const Slice& limit, int cf) {
+ Range r(start, limit);
+ uint64_t size;
+ if (cf == 0) {
+ db_->GetApproximateSizes(&r, 1, &size);
+ } else {
+ db_->GetApproximateSizes(handles_[1], &r, 1, &size);
+ }
+ return size;
+}
+
+void DBTestBase::Compact(int cf, const Slice& start, const Slice& limit,
+ uint32_t target_path_id) {
+ CompactRangeOptions compact_options;
+ compact_options.target_path_id = target_path_id;
+ ASSERT_OK(db_->CompactRange(compact_options, handles_[cf], &start, &limit));
+}
+
+void DBTestBase::Compact(int cf, const Slice& start, const Slice& limit) {
+ ASSERT_OK(
+ db_->CompactRange(CompactRangeOptions(), handles_[cf], &start, &limit));
+}
+
+void DBTestBase::Compact(const Slice& start, const Slice& limit) {
+ ASSERT_OK(db_->CompactRange(CompactRangeOptions(), &start, &limit));
+}
+
+// Do n memtable compactions, each of which produces an sstable
+// covering the range [small,large].
+void DBTestBase::MakeTables(int n, const std::string& small,
+ const std::string& large, int cf) {
+ for (int i = 0; i < n; i++) {
+ ASSERT_OK(Put(cf, small, "begin"));
+ ASSERT_OK(Put(cf, large, "end"));
+ ASSERT_OK(Flush(cf));
+ MoveFilesToLevel(n - i - 1, cf);
+ }
+}
+
+// Prevent pushing of new sstables into deeper levels by adding
+// tables that cover a specified range to all levels.
+void DBTestBase::FillLevels(const std::string& smallest,
+ const std::string& largest, int cf) {
+ MakeTables(db_->NumberLevels(handles_[cf]), smallest, largest, cf);
+}
+
+void DBTestBase::MoveFilesToLevel(int level, int cf) {
+ for (int l = 0; l < level; ++l) {
+ if (cf > 0) {
+ dbfull()->TEST_CompactRange(l, nullptr, nullptr, handles_[cf]);
+ } else {
+ dbfull()->TEST_CompactRange(l, nullptr, nullptr);
+ }
+ }
+}
+
+void DBTestBase::DumpFileCounts(const char* label) {
+ fprintf(stderr, "---\n%s:\n", label);
+ fprintf(stderr, "maxoverlap: %" PRIu64 "\n",
+ dbfull()->TEST_MaxNextLevelOverlappingBytes());
+ for (int level = 0; level < db_->NumberLevels(); level++) {
+ int num = NumTableFilesAtLevel(level);
+ if (num > 0) {
+ fprintf(stderr, " level %3d : %d files\n", level, num);
+ }
+ }
+}
+
+std::string DBTestBase::DumpSSTableList() {
+ std::string property;
+ db_->GetProperty("rocksdb.sstables", &property);
+ return property;
+}
+
+void DBTestBase::GetSstFiles(std::string path,
+ std::vector<std::string>* files) {
+ env_->GetChildren(path, files);
+
+ files->erase(
+ std::remove_if(files->begin(), files->end(), [](std::string name) {
+ uint64_t number;
+ FileType type;
+ return !(ParseFileName(name, &number, &type) && type == kTableFile);
+ }), files->end());
+}
+
+int DBTestBase::GetSstFileCount(std::string path) {
+ std::vector<std::string> files;
+ GetSstFiles(path, &files);
+ return static_cast<int>(files.size());
+}
+
+// this will generate non-overlapping files since it keeps increasing key_idx
+void DBTestBase::GenerateNewFile(int cf, Random* rnd, int* key_idx,
+ bool nowait) {
+ for (int i = 0; i < KNumKeysByGenerateNewFile; i++) {
+ ASSERT_OK(Put(cf, Key(*key_idx), RandomString(rnd, (i == 99) ? 1 : 990)));
+ (*key_idx)++;
+ }
+ if (!nowait) {
+ dbfull()->TEST_WaitForFlushMemTable();
+ dbfull()->TEST_WaitForCompact();
+ }
+}
+
+// this will generate non-overlapping files since it keeps increasing key_idx
+void DBTestBase::GenerateNewFile(Random* rnd, int* key_idx, bool nowait) {
+ for (int i = 0; i < KNumKeysByGenerateNewFile; i++) {
+ ASSERT_OK(Put(Key(*key_idx), RandomString(rnd, (i == 99) ? 1 : 990)));
+ (*key_idx)++;
+ }
+ if (!nowait) {
+ dbfull()->TEST_WaitForFlushMemTable();
+ dbfull()->TEST_WaitForCompact();
+ }
+}
+
+const int DBTestBase::kNumKeysByGenerateNewRandomFile = 51;
+
+void DBTestBase::GenerateNewRandomFile(Random* rnd, bool nowait) {
+ for (int i = 0; i < kNumKeysByGenerateNewRandomFile; i++) {
+ ASSERT_OK(Put("key" + RandomString(rnd, 7), RandomString(rnd, 2000)));
+ }
+ ASSERT_OK(Put("key" + RandomString(rnd, 7), RandomString(rnd, 200)));
+ if (!nowait) {
+ dbfull()->TEST_WaitForFlushMemTable();
+ dbfull()->TEST_WaitForCompact();
+ }
+}
+
+std::string DBTestBase::IterStatus(Iterator* iter) {
+ std::string result;
+ if (iter->Valid()) {
+ result = iter->key().ToString() + "->" + iter->value().ToString();
+ } else {
+ result = "(invalid)";
+ }
+ return result;
+}
+
+Options DBTestBase::OptionsForLogIterTest() {
+ Options options = CurrentOptions();
+ options.create_if_missing = true;
+ options.WAL_ttl_seconds = 1000;
+ return options;
+}
+
+std::string DBTestBase::DummyString(size_t len, char c) {
+ return std::string(len, c);
+}
+
+void DBTestBase::VerifyIterLast(std::string expected_key, int cf) {
+ Iterator* iter;
+ ReadOptions ro;
+ if (cf == 0) {
+ iter = db_->NewIterator(ro);
+ } else {
+ iter = db_->NewIterator(ro, handles_[cf]);
+ }
+ iter->SeekToLast();
+ ASSERT_EQ(IterStatus(iter), expected_key);
+ delete iter;
+}
+
+// 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.
+UpdateStatus DBTestBase::updateInPlaceSmallerSize(char* prevValue,
+ uint32_t* prevSize,
+ Slice delta,
+ std::string* newValue) {
+ if (prevValue == nullptr) {
+ *newValue = std::string(delta.size(), 'c');
+ return UpdateStatus::UPDATED;
+ } else {
+ *prevSize = *prevSize - 1;
+ std::string str_b = std::string(*prevSize, 'b');
+ memcpy(prevValue, str_b.c_str(), str_b.size());
+ return UpdateStatus::UPDATED_INPLACE;
+ }
+}
+
+UpdateStatus DBTestBase::updateInPlaceSmallerVarintSize(char* prevValue,
+ uint32_t* prevSize,
+ Slice delta,
+ std::string* newValue) {
+ if (prevValue == nullptr) {
+ *newValue = std::string(delta.size(), 'c');
+ return UpdateStatus::UPDATED;
+ } else {
+ *prevSize = 1;
+ std::string str_b = std::string(*prevSize, 'b');
+ memcpy(prevValue, str_b.c_str(), str_b.size());
+ return UpdateStatus::UPDATED_INPLACE;
+ }
+}
+
+UpdateStatus DBTestBase::updateInPlaceLargerSize(char* prevValue,
+ uint32_t* prevSize,
+ Slice delta,
+ std::string* newValue) {
+ *newValue = std::string(delta.size(), 'c');
+ return UpdateStatus::UPDATED;
+}
+
+UpdateStatus DBTestBase::updateInPlaceNoAction(char* prevValue,
+ uint32_t* prevSize, Slice delta,
+ std::string* newValue) {
+ return UpdateStatus::UPDATE_FAILED;
+}
+
+// Utility method to test InplaceUpdate
+void DBTestBase::validateNumberOfEntries(int numValues, int cf) {
+ ScopedArenaIterator iter;
+ Arena arena;
+ auto options = CurrentOptions();
+ InternalKeyComparator icmp(options.comparator);
+ RangeDelAggregator range_del_agg(icmp, {} /* snapshots */);
+ if (cf != 0) {
+ iter.set(
+ dbfull()->NewInternalIterator(&arena, &range_del_agg, handles_[cf]));
+ } else {
+ iter.set(dbfull()->NewInternalIterator(&arena, &range_del_agg));
+ }
+ iter->SeekToFirst();
+ ASSERT_EQ(iter->status().ok(), true);
+ int seq = numValues;
+ while (iter->Valid()) {
+ ParsedInternalKey ikey;
+ ikey.sequence = -1;
+ ASSERT_EQ(ParseInternalKey(iter->key(), &ikey), true);
+
+ // checks sequence number for updates
+ ASSERT_EQ(ikey.sequence, (unsigned)seq--);
+ iter->Next();
+ }
+ ASSERT_EQ(0, seq);
+}
+
+void DBTestBase::CopyFile(const std::string& source,
+ const std::string& destination, uint64_t size) {
+ const EnvOptions soptions;
+ unique_ptr<SequentialFile> srcfile;
+ ASSERT_OK(env_->NewSequentialFile(source, &srcfile, soptions));
+ unique_ptr<WritableFile> destfile;
+ ASSERT_OK(env_->NewWritableFile(destination, &destfile, soptions));
+
+ if (size == 0) {
+ // default argument means copy everything
+ ASSERT_OK(env_->GetFileSize(source, &size));
+ }
+
+ char buffer[4096];
+ Slice slice;
+ while (size > 0) {
+ uint64_t one = std::min(uint64_t(sizeof(buffer)), size);
+ ASSERT_OK(srcfile->Read(one, &slice, buffer));
+ ASSERT_OK(destfile->Append(slice));
+ size -= slice.size();
+ }
+ ASSERT_OK(destfile->Close());
+}
+
+std::unordered_map<std::string, uint64_t> DBTestBase::GetAllSSTFiles(
+ uint64_t* total_size) {
+ std::unordered_map<std::string, uint64_t> res;
+
+ if (total_size) {
+ *total_size = 0;
+ }
+ std::vector<std::string> files;
+ env_->GetChildren(dbname_, &files);
+ for (auto& file_name : files) {
+ uint64_t number;
+ FileType type;
+ std::string file_path = dbname_ + "/" + file_name;
+ if (ParseFileName(file_name, &number, &type) && type == kTableFile) {
+ uint64_t file_size = 0;
+ env_->GetFileSize(file_path, &file_size);
+ res[file_path] = file_size;
+ if (total_size) {
+ *total_size += file_size;
+ }
+ }
+ }
+ return res;
+}
+
+std::vector<std::uint64_t> DBTestBase::ListTableFiles(Env* env,
+ const std::string& path) {
+ std::vector<std::string> files;
+ std::vector<uint64_t> file_numbers;
+ env->GetChildren(path, &files);
+ uint64_t number;
+ FileType type;
+ for (size_t i = 0; i < files.size(); ++i) {
+ if (ParseFileName(files[i], &number, &type)) {
+ if (type == kTableFile) {
+ file_numbers.push_back(number);
+ }
+ }
+ }
+ return file_numbers;
+}
+
+void DBTestBase::VerifyDBFromMap(std::map<std::string, std::string> true_data,
+ size_t* total_reads_res, bool tailing_iter,
+ std::map<std::string, Status> status) {
+ size_t total_reads = 0;
+
+ for (auto& kv : true_data) {
+ Status s = status[kv.first];
+ if (s.ok()) {
+ ASSERT_EQ(Get(kv.first), kv.second);
+ } else {
+ std::string value;
+ ASSERT_EQ(s, db_->Get(ReadOptions(), kv.first, &value));
+ }
+ total_reads++;
+ }
+
+ // Normal Iterator
+ {
+ int iter_cnt = 0;
+ ReadOptions ro;
+ ro.total_order_seek = true;
+ Iterator* iter = db_->NewIterator(ro);
+ // Verify Iterator::Next()
+ iter_cnt = 0;
+ auto data_iter = true_data.begin();
+ Status s;
+ for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) {
+ ASSERT_EQ(iter->key().ToString(), data_iter->first);
+ Status current_status = status[data_iter->first];
+ if (!current_status.ok()) {
+ s = current_status;
+ }
+ ASSERT_EQ(iter->status(), s);
+ if (current_status.ok()) {
+ ASSERT_EQ(iter->value().ToString(), data_iter->second);
+ }
+ iter_cnt++;
+ total_reads++;
+ }
+ ASSERT_EQ(data_iter, true_data.end()) << iter_cnt << " / "
+ << true_data.size();
+ delete iter;
+
+ // Verify Iterator::Prev()
+ // Use a new iterator to make sure its status is clean.
+ iter = db_->NewIterator(ro);
+ iter_cnt = 0;
+ s = Status::OK();
+ auto data_rev = true_data.rbegin();
+ for (iter->SeekToLast(); iter->Valid(); iter->Prev(), data_rev++) {
+ ASSERT_EQ(iter->key().ToString(), data_rev->first);
+ Status current_status = status[data_rev->first];
+ if (!current_status.ok()) {
+ s = current_status;
+ }
+ ASSERT_EQ(iter->status(), s);
+ if (current_status.ok()) {
+ ASSERT_EQ(iter->value().ToString(), data_rev->second);
+ }
+ iter_cnt++;
+ total_reads++;
+ }
+ ASSERT_EQ(data_rev, true_data.rend()) << iter_cnt << " / "
+ << true_data.size();
+
+ // Verify Iterator::Seek()
+ for (auto kv : true_data) {
+ iter->Seek(kv.first);
+ ASSERT_EQ(kv.first, iter->key().ToString());
+ ASSERT_EQ(kv.second, iter->value().ToString());
+ total_reads++;
+ }
+ delete iter;
+ }
+
+ if (tailing_iter) {
+#ifndef ROCKSDB_LITE
+ // Tailing iterator
+ int iter_cnt = 0;
+ ReadOptions ro;
+ ro.tailing = true;
+ ro.total_order_seek = true;
+ Iterator* iter = db_->NewIterator(ro);
+
+ // Verify ForwardIterator::Next()
+ iter_cnt = 0;
+ auto data_iter = true_data.begin();
+ for (iter->SeekToFirst(); iter->Valid(); iter->Next(), data_iter++) {
+ ASSERT_EQ(iter->key().ToString(), data_iter->first);
+ ASSERT_EQ(iter->value().ToString(), data_iter->second);
+ iter_cnt++;
+ total_reads++;
+ }
+ ASSERT_EQ(data_iter, true_data.end()) << iter_cnt << " / "
+ << true_data.size();
+
+ // Verify ForwardIterator::Seek()
+ for (auto kv : true_data) {
+ iter->Seek(kv.first);
+ ASSERT_EQ(kv.first, iter->key().ToString());
+ ASSERT_EQ(kv.second, iter->value().ToString());
+ total_reads++;
+ }
+
+ delete iter;
+#endif // ROCKSDB_LITE
+ }
+
+ if (total_reads_res) {
+ *total_reads_res = total_reads;
+ }
+}
+
+void DBTestBase::VerifyDBInternal(
+ std::vector<std::pair<std::string, std::string>> true_data) {
+ Arena arena;
+ InternalKeyComparator icmp(last_options_.comparator);
+ RangeDelAggregator range_del_agg(icmp, {});
+ auto iter = dbfull()->NewInternalIterator(&arena, &range_del_agg);
+ iter->SeekToFirst();
+ for (auto p : true_data) {
+ ASSERT_TRUE(iter->Valid());
+ ParsedInternalKey ikey;
+ ASSERT_TRUE(ParseInternalKey(iter->key(), &ikey));
+ ASSERT_EQ(p.first, ikey.user_key);
+ ASSERT_EQ(p.second, iter->value());
+ iter->Next();
+ };
+ ASSERT_FALSE(iter->Valid());
+ iter->~InternalIterator();
+}
+
+#ifndef ROCKSDB_LITE
+
+uint64_t DBTestBase::GetNumberOfSstFilesForColumnFamily(
+ DB* db, std::string column_family_name) {
+ std::vector<LiveFileMetaData> metadata;
+ db->GetLiveFilesMetaData(&metadata);
+ uint64_t result = 0;
+ for (auto& fileMetadata : metadata) {
+ result += (fileMetadata.column_family_name == column_family_name);
+ }
+ return result;
+}
+#endif // ROCKSDB_LITE
+
+} // namespace rocksdb
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/48867732/thirdparty/rocksdb/db/db_test_util.h
----------------------------------------------------------------------
diff --git a/thirdparty/rocksdb/db/db_test_util.h b/thirdparty/rocksdb/db/db_test_util.h
new file mode 100644
index 0000000..cd1265e
--- /dev/null
+++ b/thirdparty/rocksdb/db/db_test_util.h
@@ -0,0 +1,939 @@
+// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
+// This source code is licensed under both the GPLv2 (found in the
+// COPYING file in the root directory) and Apache 2.0 License
+// (found in the LICENSE.Apache file in the root directory).
+//
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+
+#pragma once
+#ifndef __STDC_FORMAT_MACROS
+#define __STDC_FORMAT_MACROS
+#endif
+
+#include <fcntl.h>
+#include <inttypes.h>
+
+#include <algorithm>
+#include <map>
+#include <set>
+#include <string>
+#include <thread>
+#include <unordered_set>
+#include <utility>
+#include <vector>
+
+#include "db/db_impl.h"
+#include "db/dbformat.h"
+#include "env/mock_env.h"
+#include "memtable/hash_linklist_rep.h"
+#include "rocksdb/cache.h"
+#include "rocksdb/compaction_filter.h"
+#include "rocksdb/convenience.h"
+#include "rocksdb/db.h"
+#include "rocksdb/env.h"
+#include "rocksdb/filter_policy.h"
+#include "rocksdb/options.h"
+#include "rocksdb/slice.h"
+#include "rocksdb/sst_file_writer.h"
+#include "rocksdb/statistics.h"
+#include "rocksdb/table.h"
+#include "rocksdb/utilities/checkpoint.h"
+#include "table/block_based_table_factory.h"
+#include "table/mock_table.h"
+#include "table/plain_table_factory.h"
+#include "table/scoped_arena_iterator.h"
+#include "util/compression.h"
+#include "util/filename.h"
+#include "util/mutexlock.h"
+
+#include "util/string_util.h"
+#include "util/sync_point.h"
+#include "util/testharness.h"
+#include "util/testutil.h"
+#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();
+ cond_count_.TimedWait(now + /*1s*/ 1 * 1000 * 1000);
+ if (env_->NowMicros() - start > /*10s*/ 10 * 1000 * 1000) {
+ 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;
+ // These will be used only if filter_policy is set
+ bool partition_filters = false;
+ uint64_t metadata_block_size = 1024;
+ BlockBasedTableOptions::IndexType index_type =
+ BlockBasedTableOptions::IndexType::kBinarySearch;
+
+ // Used as a bit mask of individual enums in which to skip an XF test point
+ int skip_policy = 0;
+};
+
+} // namespace anon
+
+enum SkipPolicy { kSkipNone = 0, kSkipNoSnapshot = 1, kSkipNoPrefix = 2 };
+
+// A hacky skip list mem table that triggers flush after number of entries.
+class SpecialMemTableRep : public MemTableRep {
+ public:
+ explicit SpecialMemTableRep(Allocator* allocator, MemTableRep* memtable,
+ int num_entries_flush)
+ : 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.
+ virtual void Insert(KeyHandle handle) override {
+ memtable_->Insert(handle);
+ num_entries_++;
+ }
+
+ // 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:
+ unique_ptr<MemTableRep> memtable_;
+ 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) {}
+
+ using MemTableRepFactory::CreateMemTableRep;
+ virtual MemTableRep* CreateMemTableRep(
+ const MemTableRep::KeyComparator& compare, Allocator* allocator,
+ const SliceTransform* transform, Logger* logger) override {
+ return new SpecialMemTableRep(
+ allocator, factory_.CreateMemTableRep(compare, allocator, transform, 0),
+ num_entries_flush_);
+ }
+ virtual const char* Name() const override { return "SkipListFactory"; }
+
+ bool IsInsertConcurrentlySupported() const override {
+ return factory_.IsInsertConcurrentlySupported();
+ }
+
+ private:
+ SkipListFactory factory_;
+ int num_entries_flush_;
+};
+
+// Special Env used to delay background operations
+class SpecialEnv : public EnvWrapper {
+ public:
+ explicit SpecialEnv(Env* base);
+
+ Status NewWritableFile(const std::string& f, unique_ptr<WritableFile>* r,
+ const EnvOptions& soptions) override {
+ class SSTableFile : public WritableFile {
+ private:
+ SpecialEnv* env_;
+ unique_ptr<WritableFile> base_;
+
+ public:
+ SSTableFile(SpecialEnv* env, unique_ptr<WritableFile>&& base)
+ : env_(env), base_(std::move(base)) {}
+ 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)) {
+ return Status::NoSpace("No space left on device");
+ } else {
+ env_->bytes_written_ += data.size();
+ return base_->Append(data);
+ }
+ }
+ Status PositionedAppend(const Slice& data, uint64_t offset) 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)) {
+ return Status::NoSpace("No space left on device");
+ } else {
+ env_->bytes_written_ += data.size();
+ return base_->PositionedAppend(data, offset);
+ }
+ }
+ Status Truncate(uint64_t size) override { return base_->Truncate(size); }
+ 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;
+ }
+ 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("DBTestWritableFile.GetPreallocationStatus",
+ &preallocation_size);
+#endif // !(defined NDEBUG) || !defined(OS_WIN)
+ 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;
+ }
+ 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);
+ }
+ 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;
+ }
+ void SetIOPriority(Env::IOPriority pri) override {
+ base_->SetIOPriority(pri);
+ }
+ Env::IOPriority GetIOPriority() override {
+ return base_->GetIOPriority();
+ }
+ bool use_direct_io() const override {
+ return base_->use_direct_io();
+ }
+ Status Allocate(uint64_t offset, uint64_t len) override {
+ return base_->Allocate(offset, len);
+ }
+ };
+ class ManifestFile : public WritableFile {
+ public:
+ ManifestFile(SpecialEnv* env, unique_ptr<WritableFile>&& b)
+ : env_(env), base_(std::move(b)) {}
+ 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);
+ }
+ }
+ Status Truncate(uint64_t size) override { return base_->Truncate(size); }
+ 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(); }
+
+ private:
+ SpecialEnv* env_;
+ unique_ptr<WritableFile> base_;
+ };
+ class WalFile : public WritableFile {
+ public:
+ WalFile(SpecialEnv* env, unique_ptr<WritableFile>&& b)
+ : env_(env), base_(std::move(b)) {
+ env_->num_open_wal_file_.fetch_add(1);
+ }
+ virtual ~WalFile() { env_->num_open_wal_file_.fetch_add(-1); }
+ Status Append(const Slice& data) override {
+#if !(defined NDEBUG) || !defined(OS_WIN)
+ TEST_SYNC_POINT("SpecialEnv::WalFile::Append:1");
+#endif
+ Status s;
+ if (env_->log_write_error_.load(std::memory_order_acquire)) {
+ s = Status::IOError("simulated writer error");
+ } else {
+ int slowdown =
+ env_->log_write_slowdown_.load(std::memory_order_acquire);
+ if (slowdown > 0) {
+ env_->SleepForMicroseconds(slowdown);
+ }
+ s = base_->Append(data);
+ }
+#if !(defined NDEBUG) || !defined(OS_WIN)
+ TEST_SYNC_POINT("SpecialEnv::WalFile::Append:2");
+#endif
+ return s;
+ }
+ Status Truncate(uint64_t size) override { return base_->Truncate(size); }
+ 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();
+ }
+ Status Flush() override { return base_->Flush(); }
+ Status Sync() override {
+ ++env_->sync_counter_;
+ return base_->Sync();
+ }
+ bool IsSyncThreadSafe() const override {
+ return env_->is_wal_sync_thread_safe_.load();
+ }
+
+ private:
+ SpecialEnv* env_;
+ unique_ptr<WritableFile> base_;
+ };
+
+ 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");
+ }
+
+ 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);
+ 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,
+ unique_ptr<RandomAccessFile>* r,
+ const EnvOptions& soptions) override {
+ class CountingFile : public RandomAccessFile {
+ public:
+ CountingFile(unique_ptr<RandomAccessFile>&& target,
+ anon::AtomicCounter* counter,
+ std::atomic<size_t>* bytes_read)
+ : target_(std::move(target)),
+ counter_(counter),
+ bytes_read_(bytes_read) {}
+ virtual Status Read(uint64_t offset, size_t n, Slice* result,
+ char* scratch) const override {
+ counter_->Increment();
+ Status s = target_->Read(offset, n, result, scratch);
+ *bytes_read_ += result->size();
+ return s;
+ }
+
+ private:
+ unique_ptr<RandomAccessFile> target_;
+ anon::AtomicCounter* counter_;
+ std::atomic<size_t>* bytes_read_;
+ };
+
+ Status s = target()->NewRandomAccessFile(f, r, soptions);
+ random_file_open_counter_++;
+ if (s.ok() && count_random_reads_) {
+ r->reset(new CountingFile(std::move(*r), &random_read_counter_,
+ &random_read_bytes_counter_));
+ }
+ return s;
+ }
+
+ Status NewSequentialFile(const std::string& f, unique_ptr<SequentialFile>* r,
+ const EnvOptions& soptions) override {
+ class CountingFile : public SequentialFile {
+ public:
+ CountingFile(unique_ptr<SequentialFile>&& target,
+ 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:
+ unique_ptr<SequentialFile> target_;
+ 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();
+ if (no_slowdown_ || time_elapse_only_sleep_) {
+ addon_time_.fetch_add(micros);
+ }
+ if (!no_slowdown_) {
+ target()->SleepForMicroseconds(micros);
+ }
+ }
+
+ virtual Status GetCurrentTime(int64_t* unix_time) override {
+ Status s;
+ if (!time_elapse_only_sleep_) {
+ s = target()->GetCurrentTime(unix_time);
+ }
+ if (s.ok()) {
+ *unix_time += addon_time_.load();
+ }
+ return s;
+ }
+
+ virtual uint64_t NowNanos() override {
+ return (time_elapse_only_sleep_ ? 0 : target()->NowNanos()) +
+ addon_time_.load() * 1000;
+ }
+
+ virtual uint64_t NowMicros() override {
+ return (time_elapse_only_sleep_ ? 0 : target()->NowMicros()) +
+ addon_time_.load();
+ }
+
+ virtual Status DeleteFile(const std::string& fname) override {
+ delete_count_.fetch_add(1);
+ return target()->DeleteFile(fname);
+ }
+
+ 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_;
+
+ // Number of WAL files that are still open for write.
+ std::atomic<int> num_open_wal_file_;
+
+ bool count_random_reads_;
+ anon::AtomicCounter random_read_counter_;
+ std::atomic<size_t> random_read_bytes_counter_;
+ std::atomic<int> random_file_open_counter_;
+
+ 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_;
+
+ std::atomic<int> delete_count_;
+
+ bool time_elapse_only_sleep_;
+
+ bool no_slowdown_;
+
+ std::atomic<bool> is_wal_sync_thread_safe_{true};
+};
+
+#ifndef ROCKSDB_LITE
+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_;
+};
+#endif
+
+// 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"; }
+};
+
+class DBTestBase : public testing::Test {
+ public:
+ // Sequence of option configurations to try
+ enum OptionConfig : int {
+ kDefault = 0,
+ kBlockBasedTableWithPrefixHashIndex = 1,
+ kBlockBasedTableWithWholeKeyHashIndex = 2,
+ kPlainTableFirstBytePrefix = 3,
+ kPlainTableCappedPrefix = 4,
+ kPlainTableCappedPrefixNonMmap = 5,
+ kPlainTableAllBytesPrefix = 6,
+ kVectorRep = 7,
+ kHashLinkList = 8,
+ kHashCuckoo = 9,
+ kMergePut = 10,
+ kFilter = 11,
+ kFullFilterWithNewTableReaderForCompactions = 12,
+ kUncompressed = 13,
+ kNumLevel_3 = 14,
+ kDBLogDir = 15,
+ kWalDirAndMmapReads = 16,
+ kManifestFileSize = 17,
+ kPerfOptions = 18,
+ kHashSkipList = 19,
+ kUniversalCompaction = 20,
+ kUniversalCompactionMultiLevel = 21,
+ kCompressedBlockCache = 22,
+ kInfiniteMaxOpenFiles = 23,
+ kxxHashChecksum = 24,
+ kFIFOCompaction = 25,
+ kOptimizeFiltersForHits = 26,
+ kRowCache = 27,
+ kRecycleLogFiles = 28,
+ kConcurrentSkipList = 29,
+ kPipelinedWrite = 30,
+ kConcurrentWALWrites = 31,
+ kEnd = 32,
+ kDirectIO = 33,
+ kLevelSubcompactions = 34,
+ kUniversalSubcompactions = 35,
+ kBlockBasedTableWithIndexRestartInterval = 36,
+ kBlockBasedTableWithPartitionedIndex = 37,
+ kPartitionedFilterWithNewTableReaderForCompactions = 38,
+ };
+
+ public:
+ std::string dbname_;
+ std::string alternative_wal_dir_;
+ std::string alternative_db_log_dir_;
+ MockEnv* mem_env_;
+ Env* encrypted_env_;
+ SpecialEnv* env_;
+ DB* db_;
+ std::vector<ColumnFamilyHandle*> handles_;
+
+ int option_config_;
+ 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,
+ kSkipHashCuckoo = 64,
+ kSkipFIFOCompaction = 128,
+ kSkipMmapReads = 256,
+ };
+
+ explicit DBTestBase(const std::string path);
+
+ ~DBTestBase();
+
+ 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);
+ }
+
+ static bool ShouldSkipOptions(int option_config, int skip_mask = kNoSkip);
+
+ // 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);
+
+ // Switch between different compaction styles.
+ bool ChangeCompactOptions();
+
+ // Switch between different WAL-realted options.
+ bool ChangeWalOptions();
+
+ // Switch between different filter policy
+ // Jump from kDefault to kFilter to kFullFilter
+ bool ChangeFilterOptions();
+
+ // Return the current option configuration.
+ 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();
+
+ Options GetOptions(int option_config,
+ const Options& default_options = GetDefaultOptions(),
+ const anon::OptionsOverride& options_override =
+ anon::OptionsOverride()) const;
+
+ DBImpl* dbfull() { return reinterpret_cast<DBImpl*>(db_); }
+
+ 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);
+
+ Status TryReopenWithColumnFamilies(const std::vector<std::string>& cfs,
+ const std::vector<Options>& options);
+
+ Status TryReopenWithColumnFamilies(const std::vector<std::string>& cfs,
+ const Options& options);
+
+ void Reopen(const Options& options);
+
+ void Close();
+
+ void DestroyAndReopen(const Options& options);
+
+ void Destroy(const Options& options);
+
+ Status ReadOnlyReopen(const Options& options);
+
+ Status TryReopen(const Options& options);
+
+ bool IsDirectIOSupported();
+
+ bool IsMemoryMappedAccessSupported() const;
+
+ Status Flush(int cf = 0);
+
+ Status Put(const Slice& k, const Slice& v, WriteOptions wo = WriteOptions());
+
+ Status Put(int cf, const Slice& k, const Slice& v,
+ WriteOptions wo = WriteOptions());
+
+ Status Merge(const Slice& k, const Slice& v,
+ WriteOptions wo = WriteOptions());
+
+ Status Merge(int cf, const Slice& k, const Slice& v,
+ WriteOptions wo = WriteOptions());
+
+ Status Delete(const std::string& k);
+
+ Status Delete(int cf, const std::string& k);
+
+ Status SingleDelete(const std::string& k);
+
+ Status SingleDelete(int cf, const std::string& k);
+
+ std::string Get(const std::string& k, const Snapshot* snapshot = nullptr);
+
+ std::string Get(int cf, const std::string& k,
+ const Snapshot* snapshot = nullptr);
+
+ Status Get(const std::string& k, PinnableSlice* v);
+
+ 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);
+
+#ifndef ROCKSDB_LITE
+ int NumSortedRuns(int cf = 0);
+
+ uint64_t TotalSize(int cf = 0);
+
+ uint64_t SizeAtLevel(int level);
+
+ size_t TotalLiveFiles(int cf = 0);
+
+ size_t CountLiveFiles();
+#endif // ROCKSDB_LITE
+
+ int NumTableFilesAtLevel(int level, int cf = 0);
+
+ double CompressionRatioAtLevel(int level, int cf = 0);
+
+ int TotalTableFiles(int cf = 0, int levels = -1);
+
+ // 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);
+
+ void MoveFilesToLevel(int level, int cf = 0);
+
+ void DumpFileCounts(const char* label);
+
+ std::string DumpSSTableList();
+
+ void GetSstFiles(std::string path, std::vector<std::string>* files);
+
+ 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);
+
+ void GenerateNewFile(int fd, Random* rnd, int* key_idx, bool nowait = false);
+
+ static const int kNumKeysByGenerateNewRandomFile;
+ static const int KNumKeysByGenerateNewFile = 100;
+
+ 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.
+ static UpdateStatus updateInPlaceSmallerSize(char* prevValue,
+ uint32_t* prevSize, Slice delta,
+ std::string* newValue);
+
+ static UpdateStatus updateInPlaceSmallerVarintSize(char* prevValue,
+ uint32_t* prevSize,
+ Slice delta,
+ std::string* newValue);
+
+ static UpdateStatus updateInPlaceLargerSize(char* prevValue,
+ uint32_t* prevSize, Slice delta,
+ std::string* newValue);
+
+ static UpdateStatus updateInPlaceNoAction(char* prevValue, uint32_t* prevSize,
+ Slice delta, std::string* newValue);
+
+ // 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);
+
+ std::unordered_map<std::string, uint64_t> GetAllSSTFiles(
+ uint64_t* total_size = nullptr);
+
+ std::vector<std::uint64_t> ListTableFiles(Env* env, const std::string& path);
+
+ 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>());
+
+ void VerifyDBInternal(
+ std::vector<std::pair<std::string, std::string>> true_data);
+
+#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);
+ }
+};
+
+} // namespace rocksdb
http://git-wip-us.apache.org/repos/asf/nifi-minifi-cpp/blob/48867732/thirdparty/rocksdb/db/dbformat.cc
----------------------------------------------------------------------
diff --git a/thirdparty/rocksdb/db/dbformat.cc b/thirdparty/rocksdb/db/dbformat.cc
new file mode 100644
index 0000000..20c5449
--- /dev/null
+++ b/thirdparty/rocksdb/db/dbformat.cc
@@ -0,0 +1,177 @@
+// Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
+// This source code is licensed under both the GPLv2 (found in the
+// COPYING file in the root directory) and Apache 2.0 License
+// (found in the LICENSE.Apache file in the root directory).
+//
+// Copyright (c) 2011 The LevelDB Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file. See the AUTHORS file for names of contributors.
+#include "db/dbformat.h"
+
+#ifndef __STDC_FORMAT_MACROS
+#define __STDC_FORMAT_MACROS
+#endif
+
+#include <inttypes.h>
+#include <stdio.h>
+#include "monitoring/perf_context_imp.h"
+#include "port/port.h"
+#include "util/coding.h"
+#include "util/string_util.h"
+
+namespace rocksdb {
+
+// kValueTypeForSeek defines the ValueType that should be passed when
+// constructing a ParsedInternalKey object for seeking to a particular
+// sequence number (since we sort sequence numbers in decreasing order
+// and the value type is embedded as the low 8 bits in the sequence
+// number in internal keys, we need to use the highest-numbered
+// ValueType, not the lowest).
+const ValueType kValueTypeForSeek = kTypeSingleDeletion;
+const ValueType kValueTypeForSeekForPrev = kTypeDeletion;
+
+uint64_t PackSequenceAndType(uint64_t seq, ValueType t) {
+ assert(seq <= kMaxSequenceNumber);
+ assert(IsExtendedValueType(t));
+ return (seq << 8) | t;
+}
+
+void UnPackSequenceAndType(uint64_t packed, uint64_t* seq, ValueType* t) {
+ *seq = packed >> 8;
+ *t = static_cast<ValueType>(packed & 0xff);
+
+ assert(*seq <= kMaxSequenceNumber);
+ assert(IsExtendedValueType(*t));
+}
+
+void AppendInternalKey(std::string* result, const ParsedInternalKey& key) {
+ result->append(key.user_key.data(), key.user_key.size());
+ PutFixed64(result, PackSequenceAndType(key.sequence, key.type));
+}
+
+void AppendInternalKeyFooter(std::string* result, SequenceNumber s,
+ ValueType t) {
+ PutFixed64(result, PackSequenceAndType(s, t));
+}
+
+std::string ParsedInternalKey::DebugString(bool hex) const {
+ char buf[50];
+ snprintf(buf, sizeof(buf), "' seq:%" PRIu64 ", type:%d", sequence,
+ static_cast<int>(type));
+ std::string result = "'";
+ result += user_key.ToString(hex);
+ result += buf;
+ return result;
+}
+
+std::string InternalKey::DebugString(bool hex) const {
+ std::string result;
+ ParsedInternalKey parsed;
+ if (ParseInternalKey(rep_, &parsed)) {
+ result = parsed.DebugString(hex);
+ } else {
+ result = "(bad)";
+ result.append(EscapeString(rep_));
+ }
+ return result;
+}
+
+const char* InternalKeyComparator::Name() const {
+ return name_.c_str();
+}
+
+int InternalKeyComparator::Compare(const Slice& akey, const Slice& bkey) const {
+ // Order by:
+ // increasing user key (according to user-supplied comparator)
+ // decreasing sequence number
+ // decreasing type (though sequence# should be enough to disambiguate)
+ int r = user_comparator_->Compare(ExtractUserKey(akey), ExtractUserKey(bkey));
+ PERF_COUNTER_ADD(user_key_comparison_count, 1);
+ if (r == 0) {
+ const uint64_t anum = DecodeFixed64(akey.data() + akey.size() - 8);
+ const uint64_t bnum = DecodeFixed64(bkey.data() + bkey.size() - 8);
+ if (anum > bnum) {
+ r = -1;
+ } else if (anum < bnum) {
+ r = +1;
+ }
+ }
+ return r;
+}
+
+int InternalKeyComparator::Compare(const ParsedInternalKey& a,
+ const ParsedInternalKey& b) const {
+ // Order by:
+ // increasing user key (according to user-supplied comparator)
+ // decreasing sequence number
+ // decreasing type (though sequence# should be enough to disambiguate)
+ int r = user_comparator_->Compare(a.user_key, b.user_key);
+ PERF_COUNTER_ADD(user_key_comparison_count, 1);
+ if (r == 0) {
+ if (a.sequence > b.sequence) {
+ r = -1;
+ } else if (a.sequence < b.sequence) {
+ r = +1;
+ } else if (a.type > b.type) {
+ r = -1;
+ } else if (a.type < b.type) {
+ r = +1;
+ }
+ }
+ return r;
+}
+
+void InternalKeyComparator::FindShortestSeparator(
+ std::string* start,
+ const Slice& limit) const {
+ // Attempt to shorten the user portion of the key
+ Slice user_start = ExtractUserKey(*start);
+ Slice user_limit = ExtractUserKey(limit);
+ std::string tmp(user_start.data(), user_start.size());
+ user_comparator_->FindShortestSeparator(&tmp, user_limit);
+ if (tmp.size() <= user_start.size() &&
+ user_comparator_->Compare(user_start, tmp) < 0) {
+ // User key has become shorter physically, but larger logically.
+ // Tack on the earliest possible number to the shortened user key.
+ PutFixed64(&tmp, PackSequenceAndType(kMaxSequenceNumber,kValueTypeForSeek));
+ assert(this->Compare(*start, tmp) < 0);
+ assert(this->Compare(tmp, limit) < 0);
+ start->swap(tmp);
+ }
+}
+
+void InternalKeyComparator::FindShortSuccessor(std::string* key) const {
+ Slice user_key = ExtractUserKey(*key);
+ std::string tmp(user_key.data(), user_key.size());
+ user_comparator_->FindShortSuccessor(&tmp);
+ if (tmp.size() <= user_key.size() &&
+ user_comparator_->Compare(user_key, tmp) < 0) {
+ // User key has become shorter physically, but larger logically.
+ // Tack on the earliest possible number to the shortened user key.
+ PutFixed64(&tmp, PackSequenceAndType(kMaxSequenceNumber,kValueTypeForSeek));
+ assert(this->Compare(*key, tmp) < 0);
+ key->swap(tmp);
+ }
+}
+
+LookupKey::LookupKey(const Slice& _user_key, SequenceNumber s) {
+ size_t usize = _user_key.size();
+ size_t needed = usize + 13; // A conservative estimate
+ char* dst;
+ if (needed <= sizeof(space_)) {
+ dst = space_;
+ } else {
+ dst = new char[needed];
+ }
+ start_ = dst;
+ // NOTE: We don't support users keys of more than 2GB :)
+ dst = EncodeVarint32(dst, static_cast<uint32_t>(usize + 8));
+ kstart_ = dst;
+ memcpy(dst, _user_key.data(), usize);
+ dst += usize;
+ EncodeFixed64(dst, PackSequenceAndType(s, kValueTypeForSeek));
+ dst += 8;
+ end_ = dst;
+}
+
+} // namespace rocksdb