12f1137355
Summary: Introduces and uses a SystemClock class to RocksDB. This class contains the time-related functions of an Env and these functions can be redirected from the Env to the SystemClock. Many of the places that used an Env (Timer, PerfStepTimer, RepeatableThread, RateLimiter, WriteController) for time-related functions have been changed to use SystemClock instead. There are likely more places that can be changed, but this is a start to show what can/should be done. Over time it would be nice to migrate most (if not all) of the uses of the time functions from the Env to the SystemClock. There are several Env classes that implement these functions. Most of these have not been converted yet to SystemClock implementations; that will come in a subsequent PR. It would be good to unify many of the Mock Timer implementations, so that they behave similarly and be tested similarly (some override Sleep, some use a MockSleep, etc). Additionally, this change will allow new methods to be introduced to the SystemClock (like https://github.com/facebook/rocksdb/issues/7101 WaitFor) in a consistent manner across a smaller number of classes. Pull Request resolved: https://github.com/facebook/rocksdb/pull/7858 Reviewed By: pdillinger Differential Revision: D26006406 Pulled By: mrambacher fbshipit-source-id: ed10a8abbdab7ff2e23d69d85bd25b3e7e899e90
662 lines
23 KiB
C++
662 lines
23 KiB
C++
// 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).
|
|
|
|
#include "db/blob/blob_file_builder.h"
|
|
|
|
#include <cassert>
|
|
#include <cinttypes>
|
|
#include <string>
|
|
#include <utility>
|
|
#include <vector>
|
|
|
|
#include "db/blob/blob_file_addition.h"
|
|
#include "db/blob/blob_index.h"
|
|
#include "db/blob/blob_log_format.h"
|
|
#include "db/blob/blob_log_sequential_reader.h"
|
|
#include "env/mock_env.h"
|
|
#include "file/filename.h"
|
|
#include "file/random_access_file_reader.h"
|
|
#include "options/cf_options.h"
|
|
#include "rocksdb/env.h"
|
|
#include "rocksdb/file_checksum.h"
|
|
#include "rocksdb/options.h"
|
|
#include "test_util/sync_point.h"
|
|
#include "test_util/testharness.h"
|
|
#include "util/compression.h"
|
|
#include "utilities/fault_injection_env.h"
|
|
|
|
namespace ROCKSDB_NAMESPACE {
|
|
|
|
class TestFileNumberGenerator {
|
|
public:
|
|
uint64_t operator()() { return ++next_file_number_; }
|
|
|
|
private:
|
|
uint64_t next_file_number_ = 1;
|
|
};
|
|
|
|
class BlobFileBuilderTest : public testing::Test {
|
|
protected:
|
|
BlobFileBuilderTest() : mock_env_(Env::Default()) {
|
|
fs_ = mock_env_.GetFileSystem().get();
|
|
clock_ = mock_env_.GetSystemClock();
|
|
}
|
|
|
|
void VerifyBlobFile(uint64_t blob_file_number,
|
|
const std::string& blob_file_path,
|
|
uint32_t column_family_id,
|
|
CompressionType blob_compression_type,
|
|
const std::vector<std::pair<std::string, std::string>>&
|
|
expected_key_value_pairs,
|
|
const std::vector<std::string>& blob_indexes) {
|
|
assert(expected_key_value_pairs.size() == blob_indexes.size());
|
|
|
|
std::unique_ptr<FSRandomAccessFile> file;
|
|
constexpr IODebugContext* dbg = nullptr;
|
|
ASSERT_OK(
|
|
fs_->NewRandomAccessFile(blob_file_path, file_options_, &file, dbg));
|
|
|
|
std::unique_ptr<RandomAccessFileReader> file_reader(
|
|
new RandomAccessFileReader(std::move(file), blob_file_path, clock_));
|
|
|
|
constexpr Statistics* statistics = nullptr;
|
|
BlobLogSequentialReader blob_log_reader(std::move(file_reader), clock_,
|
|
statistics);
|
|
|
|
BlobLogHeader header;
|
|
ASSERT_OK(blob_log_reader.ReadHeader(&header));
|
|
ASSERT_EQ(header.version, kVersion1);
|
|
ASSERT_EQ(header.column_family_id, column_family_id);
|
|
ASSERT_EQ(header.compression, blob_compression_type);
|
|
ASSERT_FALSE(header.has_ttl);
|
|
ASSERT_EQ(header.expiration_range, ExpirationRange());
|
|
|
|
for (size_t i = 0; i < expected_key_value_pairs.size(); ++i) {
|
|
BlobLogRecord record;
|
|
uint64_t blob_offset = 0;
|
|
|
|
ASSERT_OK(blob_log_reader.ReadRecord(
|
|
&record, BlobLogSequentialReader::kReadHeaderKeyBlob, &blob_offset));
|
|
|
|
// Check the contents of the blob file
|
|
const auto& expected_key_value = expected_key_value_pairs[i];
|
|
const auto& key = expected_key_value.first;
|
|
const auto& value = expected_key_value.second;
|
|
|
|
ASSERT_EQ(record.key_size, key.size());
|
|
ASSERT_EQ(record.value_size, value.size());
|
|
ASSERT_EQ(record.expiration, 0);
|
|
ASSERT_EQ(record.key, key);
|
|
ASSERT_EQ(record.value, value);
|
|
|
|
// Make sure the blob reference returned by the builder points to the
|
|
// right place
|
|
BlobIndex blob_index;
|
|
ASSERT_OK(blob_index.DecodeFrom(blob_indexes[i]));
|
|
ASSERT_FALSE(blob_index.IsInlined());
|
|
ASSERT_FALSE(blob_index.HasTTL());
|
|
ASSERT_EQ(blob_index.file_number(), blob_file_number);
|
|
ASSERT_EQ(blob_index.offset(), blob_offset);
|
|
ASSERT_EQ(blob_index.size(), value.size());
|
|
}
|
|
|
|
BlobLogFooter footer;
|
|
ASSERT_OK(blob_log_reader.ReadFooter(&footer));
|
|
ASSERT_EQ(footer.blob_count, expected_key_value_pairs.size());
|
|
ASSERT_EQ(footer.expiration_range, ExpirationRange());
|
|
}
|
|
|
|
MockEnv mock_env_;
|
|
FileSystem* fs_;
|
|
std::shared_ptr<SystemClock> clock_;
|
|
FileOptions file_options_;
|
|
};
|
|
|
|
TEST_F(BlobFileBuilderTest, BuildAndCheckOneFile) {
|
|
// Build a single blob file
|
|
constexpr size_t number_of_blobs = 10;
|
|
constexpr size_t key_size = 1;
|
|
constexpr size_t value_size = 4;
|
|
constexpr size_t value_offset = 1234;
|
|
|
|
Options options;
|
|
options.cf_paths.emplace_back(
|
|
test::PerThreadDBPath(&mock_env_,
|
|
"BlobFileBuilderTest_BuildAndCheckOneFile"),
|
|
0);
|
|
options.enable_blob_files = true;
|
|
|
|
ImmutableCFOptions immutable_cf_options(options);
|
|
MutableCFOptions mutable_cf_options(options);
|
|
|
|
constexpr int job_id = 1;
|
|
constexpr uint32_t column_family_id = 123;
|
|
constexpr char column_family_name[] = "foobar";
|
|
constexpr Env::IOPriority io_priority = Env::IO_HIGH;
|
|
constexpr Env::WriteLifeTimeHint write_hint = Env::WLTH_MEDIUM;
|
|
|
|
std::vector<std::string> blob_file_paths;
|
|
std::vector<BlobFileAddition> blob_file_additions;
|
|
|
|
BlobFileBuilder builder(TestFileNumberGenerator(), &mock_env_, fs_,
|
|
&immutable_cf_options, &mutable_cf_options,
|
|
&file_options_, job_id, column_family_id,
|
|
column_family_name, io_priority, write_hint,
|
|
&blob_file_paths, &blob_file_additions);
|
|
|
|
std::vector<std::pair<std::string, std::string>> expected_key_value_pairs(
|
|
number_of_blobs);
|
|
std::vector<std::string> blob_indexes(number_of_blobs);
|
|
|
|
for (size_t i = 0; i < number_of_blobs; ++i) {
|
|
auto& expected_key_value = expected_key_value_pairs[i];
|
|
|
|
auto& key = expected_key_value.first;
|
|
key = std::to_string(i);
|
|
assert(key.size() == key_size);
|
|
|
|
auto& value = expected_key_value.second;
|
|
value = std::to_string(i + value_offset);
|
|
assert(value.size() == value_size);
|
|
|
|
auto& blob_index = blob_indexes[i];
|
|
|
|
ASSERT_OK(builder.Add(key, value, &blob_index));
|
|
ASSERT_FALSE(blob_index.empty());
|
|
}
|
|
|
|
ASSERT_OK(builder.Finish());
|
|
|
|
// Check the metadata generated
|
|
constexpr uint64_t blob_file_number = 2;
|
|
|
|
ASSERT_EQ(blob_file_paths.size(), 1);
|
|
|
|
const std::string& blob_file_path = blob_file_paths[0];
|
|
|
|
ASSERT_EQ(blob_file_path,
|
|
BlobFileName(immutable_cf_options.cf_paths.front().path,
|
|
blob_file_number));
|
|
|
|
ASSERT_EQ(blob_file_additions.size(), 1);
|
|
|
|
const auto& blob_file_addition = blob_file_additions[0];
|
|
|
|
ASSERT_EQ(blob_file_addition.GetBlobFileNumber(), blob_file_number);
|
|
ASSERT_EQ(blob_file_addition.GetTotalBlobCount(), number_of_blobs);
|
|
ASSERT_EQ(
|
|
blob_file_addition.GetTotalBlobBytes(),
|
|
number_of_blobs * (BlobLogRecord::kHeaderSize + key_size + value_size));
|
|
|
|
// Verify the contents of the new blob file as well as the blob references
|
|
VerifyBlobFile(blob_file_number, blob_file_path, column_family_id,
|
|
kNoCompression, expected_key_value_pairs, blob_indexes);
|
|
}
|
|
|
|
TEST_F(BlobFileBuilderTest, BuildAndCheckMultipleFiles) {
|
|
// Build multiple blob files: file size limit is set to the size of a single
|
|
// value, so each blob ends up in a file of its own
|
|
constexpr size_t number_of_blobs = 10;
|
|
constexpr size_t key_size = 1;
|
|
constexpr size_t value_size = 10;
|
|
constexpr size_t value_offset = 1234567890;
|
|
|
|
Options options;
|
|
options.cf_paths.emplace_back(
|
|
test::PerThreadDBPath(&mock_env_,
|
|
"BlobFileBuilderTest_BuildAndCheckMultipleFiles"),
|
|
0);
|
|
options.enable_blob_files = true;
|
|
options.blob_file_size = value_size;
|
|
|
|
ImmutableCFOptions immutable_cf_options(options);
|
|
MutableCFOptions mutable_cf_options(options);
|
|
|
|
constexpr int job_id = 1;
|
|
constexpr uint32_t column_family_id = 123;
|
|
constexpr char column_family_name[] = "foobar";
|
|
constexpr Env::IOPriority io_priority = Env::IO_HIGH;
|
|
constexpr Env::WriteLifeTimeHint write_hint = Env::WLTH_MEDIUM;
|
|
|
|
std::vector<std::string> blob_file_paths;
|
|
std::vector<BlobFileAddition> blob_file_additions;
|
|
|
|
BlobFileBuilder builder(TestFileNumberGenerator(), &mock_env_, fs_,
|
|
&immutable_cf_options, &mutable_cf_options,
|
|
&file_options_, job_id, column_family_id,
|
|
column_family_name, io_priority, write_hint,
|
|
&blob_file_paths, &blob_file_additions);
|
|
|
|
std::vector<std::pair<std::string, std::string>> expected_key_value_pairs(
|
|
number_of_blobs);
|
|
std::vector<std::string> blob_indexes(number_of_blobs);
|
|
|
|
for (size_t i = 0; i < number_of_blobs; ++i) {
|
|
auto& expected_key_value = expected_key_value_pairs[i];
|
|
|
|
auto& key = expected_key_value.first;
|
|
key = std::to_string(i);
|
|
assert(key.size() == key_size);
|
|
|
|
auto& value = expected_key_value.second;
|
|
value = std::to_string(i + value_offset);
|
|
assert(value.size() == value_size);
|
|
|
|
auto& blob_index = blob_indexes[i];
|
|
|
|
ASSERT_OK(builder.Add(key, value, &blob_index));
|
|
ASSERT_FALSE(blob_index.empty());
|
|
}
|
|
|
|
ASSERT_OK(builder.Finish());
|
|
|
|
// Check the metadata generated
|
|
ASSERT_EQ(blob_file_paths.size(), number_of_blobs);
|
|
ASSERT_EQ(blob_file_additions.size(), number_of_blobs);
|
|
|
|
for (size_t i = 0; i < number_of_blobs; ++i) {
|
|
const uint64_t blob_file_number = i + 2;
|
|
|
|
ASSERT_EQ(blob_file_paths[i],
|
|
BlobFileName(immutable_cf_options.cf_paths.front().path,
|
|
blob_file_number));
|
|
|
|
const auto& blob_file_addition = blob_file_additions[i];
|
|
|
|
ASSERT_EQ(blob_file_addition.GetBlobFileNumber(), blob_file_number);
|
|
ASSERT_EQ(blob_file_addition.GetTotalBlobCount(), 1);
|
|
ASSERT_EQ(blob_file_addition.GetTotalBlobBytes(),
|
|
BlobLogRecord::kHeaderSize + key_size + value_size);
|
|
}
|
|
|
|
// Verify the contents of the new blob files as well as the blob references
|
|
for (size_t i = 0; i < number_of_blobs; ++i) {
|
|
std::vector<std::pair<std::string, std::string>> expected_key_value_pair{
|
|
expected_key_value_pairs[i]};
|
|
std::vector<std::string> blob_index{blob_indexes[i]};
|
|
|
|
VerifyBlobFile(i + 2, blob_file_paths[i], column_family_id, kNoCompression,
|
|
expected_key_value_pair, blob_index);
|
|
}
|
|
}
|
|
|
|
TEST_F(BlobFileBuilderTest, InlinedValues) {
|
|
// All values are below the min_blob_size threshold; no blob files get written
|
|
constexpr size_t number_of_blobs = 10;
|
|
constexpr size_t key_size = 1;
|
|
constexpr size_t value_size = 10;
|
|
constexpr size_t value_offset = 1234567890;
|
|
|
|
Options options;
|
|
options.cf_paths.emplace_back(
|
|
test::PerThreadDBPath(&mock_env_, "BlobFileBuilderTest_InlinedValues"),
|
|
0);
|
|
options.enable_blob_files = true;
|
|
options.min_blob_size = 1024;
|
|
|
|
ImmutableCFOptions immutable_cf_options(options);
|
|
MutableCFOptions mutable_cf_options(options);
|
|
|
|
constexpr int job_id = 1;
|
|
constexpr uint32_t column_family_id = 123;
|
|
constexpr char column_family_name[] = "foobar";
|
|
constexpr Env::IOPriority io_priority = Env::IO_HIGH;
|
|
constexpr Env::WriteLifeTimeHint write_hint = Env::WLTH_MEDIUM;
|
|
|
|
std::vector<std::string> blob_file_paths;
|
|
std::vector<BlobFileAddition> blob_file_additions;
|
|
|
|
BlobFileBuilder builder(TestFileNumberGenerator(), &mock_env_, fs_,
|
|
&immutable_cf_options, &mutable_cf_options,
|
|
&file_options_, job_id, column_family_id,
|
|
column_family_name, io_priority, write_hint,
|
|
&blob_file_paths, &blob_file_additions);
|
|
|
|
for (size_t i = 0; i < number_of_blobs; ++i) {
|
|
const std::string key = std::to_string(i);
|
|
assert(key.size() == key_size);
|
|
|
|
const std::string value = std::to_string(i + value_offset);
|
|
assert(value.size() == value_size);
|
|
|
|
std::string blob_index;
|
|
ASSERT_OK(builder.Add(key, value, &blob_index));
|
|
ASSERT_TRUE(blob_index.empty());
|
|
}
|
|
|
|
ASSERT_OK(builder.Finish());
|
|
|
|
// Check the metadata generated
|
|
ASSERT_TRUE(blob_file_paths.empty());
|
|
ASSERT_TRUE(blob_file_additions.empty());
|
|
}
|
|
|
|
TEST_F(BlobFileBuilderTest, Compression) {
|
|
// Build a blob file with a compressed blob
|
|
if (!Snappy_Supported()) {
|
|
return;
|
|
}
|
|
|
|
constexpr size_t key_size = 1;
|
|
constexpr size_t value_size = 100;
|
|
|
|
Options options;
|
|
options.cf_paths.emplace_back(
|
|
test::PerThreadDBPath(&mock_env_, "BlobFileBuilderTest_Compression"), 0);
|
|
options.enable_blob_files = true;
|
|
options.blob_compression_type = kSnappyCompression;
|
|
|
|
ImmutableCFOptions immutable_cf_options(options);
|
|
MutableCFOptions mutable_cf_options(options);
|
|
|
|
constexpr int job_id = 1;
|
|
constexpr uint32_t column_family_id = 123;
|
|
constexpr char column_family_name[] = "foobar";
|
|
constexpr Env::IOPriority io_priority = Env::IO_HIGH;
|
|
constexpr Env::WriteLifeTimeHint write_hint = Env::WLTH_MEDIUM;
|
|
|
|
std::vector<std::string> blob_file_paths;
|
|
std::vector<BlobFileAddition> blob_file_additions;
|
|
|
|
BlobFileBuilder builder(TestFileNumberGenerator(), &mock_env_, fs_,
|
|
&immutable_cf_options, &mutable_cf_options,
|
|
&file_options_, job_id, column_family_id,
|
|
column_family_name, io_priority, write_hint,
|
|
&blob_file_paths, &blob_file_additions);
|
|
|
|
const std::string key("1");
|
|
const std::string uncompressed_value(value_size, 'x');
|
|
|
|
std::string blob_index;
|
|
|
|
ASSERT_OK(builder.Add(key, uncompressed_value, &blob_index));
|
|
ASSERT_FALSE(blob_index.empty());
|
|
|
|
ASSERT_OK(builder.Finish());
|
|
|
|
// Check the metadata generated
|
|
constexpr uint64_t blob_file_number = 2;
|
|
|
|
ASSERT_EQ(blob_file_paths.size(), 1);
|
|
|
|
const std::string& blob_file_path = blob_file_paths[0];
|
|
|
|
ASSERT_EQ(blob_file_path,
|
|
BlobFileName(immutable_cf_options.cf_paths.front().path,
|
|
blob_file_number));
|
|
|
|
ASSERT_EQ(blob_file_additions.size(), 1);
|
|
|
|
const auto& blob_file_addition = blob_file_additions[0];
|
|
|
|
ASSERT_EQ(blob_file_addition.GetBlobFileNumber(), blob_file_number);
|
|
ASSERT_EQ(blob_file_addition.GetTotalBlobCount(), 1);
|
|
|
|
CompressionOptions opts;
|
|
CompressionContext context(kSnappyCompression);
|
|
constexpr uint64_t sample_for_compression = 0;
|
|
|
|
CompressionInfo info(opts, context, CompressionDict::GetEmptyDict(),
|
|
kSnappyCompression, sample_for_compression);
|
|
|
|
std::string compressed_value;
|
|
ASSERT_TRUE(Snappy_Compress(info, uncompressed_value.data(),
|
|
uncompressed_value.size(), &compressed_value));
|
|
|
|
ASSERT_EQ(blob_file_addition.GetTotalBlobBytes(),
|
|
BlobLogRecord::kHeaderSize + key_size + compressed_value.size());
|
|
|
|
// Verify the contents of the new blob file as well as the blob reference
|
|
std::vector<std::pair<std::string, std::string>> expected_key_value_pairs{
|
|
{key, compressed_value}};
|
|
std::vector<std::string> blob_indexes{blob_index};
|
|
|
|
VerifyBlobFile(blob_file_number, blob_file_path, column_family_id,
|
|
kSnappyCompression, expected_key_value_pairs, blob_indexes);
|
|
}
|
|
|
|
TEST_F(BlobFileBuilderTest, CompressionError) {
|
|
// Simulate an error during compression
|
|
if (!Snappy_Supported()) {
|
|
return;
|
|
}
|
|
|
|
Options options;
|
|
options.cf_paths.emplace_back(
|
|
test::PerThreadDBPath(&mock_env_, "BlobFileBuilderTest_CompressionError"),
|
|
0);
|
|
options.enable_blob_files = true;
|
|
options.blob_compression_type = kSnappyCompression;
|
|
|
|
ImmutableCFOptions immutable_cf_options(options);
|
|
MutableCFOptions mutable_cf_options(options);
|
|
|
|
constexpr int job_id = 1;
|
|
constexpr uint32_t column_family_id = 123;
|
|
constexpr char column_family_name[] = "foobar";
|
|
constexpr Env::IOPriority io_priority = Env::IO_HIGH;
|
|
constexpr Env::WriteLifeTimeHint write_hint = Env::WLTH_MEDIUM;
|
|
|
|
std::vector<std::string> blob_file_paths;
|
|
std::vector<BlobFileAddition> blob_file_additions;
|
|
|
|
BlobFileBuilder builder(TestFileNumberGenerator(), &mock_env_, fs_,
|
|
&immutable_cf_options, &mutable_cf_options,
|
|
&file_options_, job_id, column_family_id,
|
|
column_family_name, io_priority, write_hint,
|
|
&blob_file_paths, &blob_file_additions);
|
|
|
|
SyncPoint::GetInstance()->SetCallBack("CompressData:TamperWithReturnValue",
|
|
[](void* arg) {
|
|
bool* ret = static_cast<bool*>(arg);
|
|
*ret = false;
|
|
});
|
|
SyncPoint::GetInstance()->EnableProcessing();
|
|
|
|
constexpr char key[] = "1";
|
|
constexpr char value[] = "deadbeef";
|
|
|
|
std::string blob_index;
|
|
|
|
ASSERT_TRUE(builder.Add(key, value, &blob_index).IsCorruption());
|
|
|
|
SyncPoint::GetInstance()->DisableProcessing();
|
|
SyncPoint::GetInstance()->ClearAllCallBacks();
|
|
|
|
constexpr uint64_t blob_file_number = 2;
|
|
|
|
ASSERT_EQ(blob_file_paths.size(), 1);
|
|
ASSERT_EQ(blob_file_paths[0],
|
|
BlobFileName(immutable_cf_options.cf_paths.front().path,
|
|
blob_file_number));
|
|
|
|
ASSERT_TRUE(blob_file_additions.empty());
|
|
}
|
|
|
|
TEST_F(BlobFileBuilderTest, Checksum) {
|
|
// Build a blob file with checksum
|
|
|
|
class DummyFileChecksumGenerator : public FileChecksumGenerator {
|
|
public:
|
|
void Update(const char* /* data */, size_t /* n */) override {}
|
|
|
|
void Finalize() override {}
|
|
|
|
std::string GetChecksum() const override { return std::string("dummy"); }
|
|
|
|
const char* Name() const override { return "DummyFileChecksum"; }
|
|
};
|
|
|
|
class DummyFileChecksumGenFactory : public FileChecksumGenFactory {
|
|
public:
|
|
std::unique_ptr<FileChecksumGenerator> CreateFileChecksumGenerator(
|
|
const FileChecksumGenContext& /* context */) override {
|
|
return std::unique_ptr<FileChecksumGenerator>(
|
|
new DummyFileChecksumGenerator);
|
|
}
|
|
|
|
const char* Name() const override { return "DummyFileChecksumGenFactory"; }
|
|
};
|
|
|
|
Options options;
|
|
options.cf_paths.emplace_back(
|
|
test::PerThreadDBPath(&mock_env_, "BlobFileBuilderTest_Checksum"), 0);
|
|
options.enable_blob_files = true;
|
|
options.file_checksum_gen_factory =
|
|
std::make_shared<DummyFileChecksumGenFactory>();
|
|
|
|
ImmutableCFOptions immutable_cf_options(options);
|
|
MutableCFOptions mutable_cf_options(options);
|
|
|
|
constexpr int job_id = 1;
|
|
constexpr uint32_t column_family_id = 123;
|
|
constexpr char column_family_name[] = "foobar";
|
|
constexpr Env::IOPriority io_priority = Env::IO_HIGH;
|
|
constexpr Env::WriteLifeTimeHint write_hint = Env::WLTH_MEDIUM;
|
|
|
|
std::vector<std::string> blob_file_paths;
|
|
std::vector<BlobFileAddition> blob_file_additions;
|
|
|
|
BlobFileBuilder builder(TestFileNumberGenerator(), &mock_env_, fs_,
|
|
&immutable_cf_options, &mutable_cf_options,
|
|
&file_options_, job_id, column_family_id,
|
|
column_family_name, io_priority, write_hint,
|
|
&blob_file_paths, &blob_file_additions);
|
|
|
|
const std::string key("1");
|
|
const std::string value("deadbeef");
|
|
|
|
std::string blob_index;
|
|
|
|
ASSERT_OK(builder.Add(key, value, &blob_index));
|
|
ASSERT_FALSE(blob_index.empty());
|
|
|
|
ASSERT_OK(builder.Finish());
|
|
|
|
// Check the metadata generated
|
|
constexpr uint64_t blob_file_number = 2;
|
|
|
|
ASSERT_EQ(blob_file_paths.size(), 1);
|
|
|
|
const std::string& blob_file_path = blob_file_paths[0];
|
|
|
|
ASSERT_EQ(blob_file_path,
|
|
BlobFileName(immutable_cf_options.cf_paths.front().path,
|
|
blob_file_number));
|
|
|
|
ASSERT_EQ(blob_file_additions.size(), 1);
|
|
|
|
const auto& blob_file_addition = blob_file_additions[0];
|
|
|
|
ASSERT_EQ(blob_file_addition.GetBlobFileNumber(), blob_file_number);
|
|
ASSERT_EQ(blob_file_addition.GetTotalBlobCount(), 1);
|
|
ASSERT_EQ(blob_file_addition.GetTotalBlobBytes(),
|
|
BlobLogRecord::kHeaderSize + key.size() + value.size());
|
|
ASSERT_EQ(blob_file_addition.GetChecksumMethod(), "DummyFileChecksum");
|
|
ASSERT_EQ(blob_file_addition.GetChecksumValue(), "dummy");
|
|
|
|
// Verify the contents of the new blob file as well as the blob reference
|
|
std::vector<std::pair<std::string, std::string>> expected_key_value_pairs{
|
|
{key, value}};
|
|
std::vector<std::string> blob_indexes{blob_index};
|
|
|
|
VerifyBlobFile(blob_file_number, blob_file_path, column_family_id,
|
|
kNoCompression, expected_key_value_pairs, blob_indexes);
|
|
}
|
|
|
|
class BlobFileBuilderIOErrorTest
|
|
: public testing::Test,
|
|
public testing::WithParamInterface<std::string> {
|
|
protected:
|
|
BlobFileBuilderIOErrorTest()
|
|
: mock_env_(Env::Default()),
|
|
fs_(mock_env_.GetFileSystem().get()),
|
|
sync_point_(GetParam()) {}
|
|
|
|
MockEnv mock_env_;
|
|
FileSystem* fs_;
|
|
FileOptions file_options_;
|
|
std::string sync_point_;
|
|
};
|
|
|
|
INSTANTIATE_TEST_CASE_P(
|
|
BlobFileBuilderTest, BlobFileBuilderIOErrorTest,
|
|
::testing::ValuesIn(std::vector<std::string>{
|
|
"BlobFileBuilder::OpenBlobFileIfNeeded:NewWritableFile",
|
|
"BlobFileBuilder::OpenBlobFileIfNeeded:WriteHeader",
|
|
"BlobFileBuilder::WriteBlobToFile:AddRecord",
|
|
"BlobFileBuilder::WriteBlobToFile:AppendFooter"}));
|
|
|
|
TEST_P(BlobFileBuilderIOErrorTest, IOError) {
|
|
// Simulate an I/O error during the specified step of Add()
|
|
// Note: blob_file_size will be set to value_size in order for the first blob
|
|
// to trigger close
|
|
constexpr size_t value_size = 8;
|
|
|
|
Options options;
|
|
options.cf_paths.emplace_back(
|
|
test::PerThreadDBPath(&mock_env_, "BlobFileBuilderIOErrorTest_IOError"),
|
|
0);
|
|
options.enable_blob_files = true;
|
|
options.blob_file_size = value_size;
|
|
options.env = &mock_env_;
|
|
|
|
ImmutableCFOptions immutable_cf_options(options);
|
|
MutableCFOptions mutable_cf_options(options);
|
|
|
|
constexpr int job_id = 1;
|
|
constexpr uint32_t column_family_id = 123;
|
|
constexpr char column_family_name[] = "foobar";
|
|
constexpr Env::IOPriority io_priority = Env::IO_HIGH;
|
|
constexpr Env::WriteLifeTimeHint write_hint = Env::WLTH_MEDIUM;
|
|
|
|
std::vector<std::string> blob_file_paths;
|
|
std::vector<BlobFileAddition> blob_file_additions;
|
|
|
|
BlobFileBuilder builder(TestFileNumberGenerator(), &mock_env_, fs_,
|
|
&immutable_cf_options, &mutable_cf_options,
|
|
&file_options_, job_id, column_family_id,
|
|
column_family_name, io_priority, write_hint,
|
|
&blob_file_paths, &blob_file_additions);
|
|
|
|
SyncPoint::GetInstance()->SetCallBack(sync_point_, [this](void* arg) {
|
|
Status* const s = static_cast<Status*>(arg);
|
|
assert(s);
|
|
|
|
(*s) = Status::IOError(sync_point_);
|
|
});
|
|
SyncPoint::GetInstance()->EnableProcessing();
|
|
|
|
constexpr char key[] = "1";
|
|
constexpr char value[] = "deadbeef";
|
|
|
|
std::string blob_index;
|
|
|
|
ASSERT_TRUE(builder.Add(key, value, &blob_index).IsIOError());
|
|
|
|
SyncPoint::GetInstance()->DisableProcessing();
|
|
SyncPoint::GetInstance()->ClearAllCallBacks();
|
|
|
|
if (sync_point_ == "BlobFileBuilder::OpenBlobFileIfNeeded:NewWritableFile") {
|
|
ASSERT_TRUE(blob_file_paths.empty());
|
|
} else {
|
|
constexpr uint64_t blob_file_number = 2;
|
|
|
|
ASSERT_EQ(blob_file_paths.size(), 1);
|
|
ASSERT_EQ(blob_file_paths[0],
|
|
BlobFileName(immutable_cf_options.cf_paths.front().path,
|
|
blob_file_number));
|
|
}
|
|
|
|
ASSERT_TRUE(blob_file_additions.empty());
|
|
}
|
|
|
|
} // namespace ROCKSDB_NAMESPACE
|
|
|
|
int main(int argc, char** argv) {
|
|
::testing::InitGoogleTest(&argc, argv);
|
|
return RUN_ALL_TESTS();
|
|
}
|