You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
rocksdb/table/block_based/data_block_hash_index_test.cc

720 lines
23 KiB

// 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 <cstdlib>
#include <string>
#include <unordered_map>
Use only "local" range tombstones during Get (#4449) Summary: Previously, range tombstones were accumulated from every level, which was necessary if a range tombstone in a higher level covered a key in a lower level. However, RangeDelAggregator::AddTombstones's complexity is based on the number of tombstones that are currently stored in it, which is wasteful in the Get case, where we only need to know the highest sequence number of range tombstones that cover the key from higher levels, and compute the highest covering sequence number at the current level. This change introduces this optimization, and removes the use of RangeDelAggregator from the Get path. In the benchmark results, the following command was used to initialize the database: ``` ./db_bench -db=/dev/shm/5k-rts -use_existing_db=false -benchmarks=filluniquerandom -write_buffer_size=1048576 -compression_type=lz4 -target_file_size_base=1048576 -max_bytes_for_level_base=4194304 -value_size=112 -key_size=16 -block_size=4096 -level_compaction_dynamic_level_bytes=true -num=5000000 -max_background_jobs=12 -benchmark_write_rate_limit=20971520 -range_tombstone_width=100 -writes_per_range_tombstone=100 -max_num_range_tombstones=50000 -bloom_bits=8 ``` ...and the following command was used to measure read throughput: ``` ./db_bench -db=/dev/shm/5k-rts/ -use_existing_db=true -benchmarks=readrandom -disable_auto_compactions=true -num=5000000 -reads=100000 -threads=32 ``` The filluniquerandom command was only run once, and the resulting database was used to measure read performance before and after the PR. Both binaries were compiled with `DEBUG_LEVEL=0`. Readrandom results before PR: ``` readrandom : 4.544 micros/op 220090 ops/sec; 16.9 MB/s (63103 of 100000 found) ``` Readrandom results after PR: ``` readrandom : 11.147 micros/op 89707 ops/sec; 6.9 MB/s (63103 of 100000 found) ``` So it's actually slower right now, but this PR paves the way for future optimizations (see #4493). ---- Pull Request resolved: https://github.com/facebook/rocksdb/pull/4449 Differential Revision: D10370575 Pulled By: abhimadan fbshipit-source-id: 9a2e152be1ef36969055c0e9eb4beb0d96c11f4d
6 years ago
#include "db/table_properties_collector.h"
#include "rocksdb/slice.h"
#include "table/block_based/block.h"
#include "table/block_based/block_based_table_reader.h"
#include "table/block_based/block_builder.h"
#include "table/block_based/data_block_hash_index.h"
#include "table/get_context.h"
Use only "local" range tombstones during Get (#4449) Summary: Previously, range tombstones were accumulated from every level, which was necessary if a range tombstone in a higher level covered a key in a lower level. However, RangeDelAggregator::AddTombstones's complexity is based on the number of tombstones that are currently stored in it, which is wasteful in the Get case, where we only need to know the highest sequence number of range tombstones that cover the key from higher levels, and compute the highest covering sequence number at the current level. This change introduces this optimization, and removes the use of RangeDelAggregator from the Get path. In the benchmark results, the following command was used to initialize the database: ``` ./db_bench -db=/dev/shm/5k-rts -use_existing_db=false -benchmarks=filluniquerandom -write_buffer_size=1048576 -compression_type=lz4 -target_file_size_base=1048576 -max_bytes_for_level_base=4194304 -value_size=112 -key_size=16 -block_size=4096 -level_compaction_dynamic_level_bytes=true -num=5000000 -max_background_jobs=12 -benchmark_write_rate_limit=20971520 -range_tombstone_width=100 -writes_per_range_tombstone=100 -max_num_range_tombstones=50000 -bloom_bits=8 ``` ...and the following command was used to measure read throughput: ``` ./db_bench -db=/dev/shm/5k-rts/ -use_existing_db=true -benchmarks=readrandom -disable_auto_compactions=true -num=5000000 -reads=100000 -threads=32 ``` The filluniquerandom command was only run once, and the resulting database was used to measure read performance before and after the PR. Both binaries were compiled with `DEBUG_LEVEL=0`. Readrandom results before PR: ``` readrandom : 4.544 micros/op 220090 ops/sec; 16.9 MB/s (63103 of 100000 found) ``` Readrandom results after PR: ``` readrandom : 11.147 micros/op 89707 ops/sec; 6.9 MB/s (63103 of 100000 found) ``` So it's actually slower right now, but this PR paves the way for future optimizations (see #4493). ---- Pull Request resolved: https://github.com/facebook/rocksdb/pull/4449 Differential Revision: D10370575 Pulled By: abhimadan fbshipit-source-id: 9a2e152be1ef36969055c0e9eb4beb0d96c11f4d
6 years ago
#include "table/table_builder.h"
#include "test_util/testharness.h"
#include "test_util/testutil.h"
namespace ROCKSDB_NAMESPACE {
bool SearchForOffset(DataBlockHashIndex& index, const char* data,
uint16_t map_offset, const Slice& key,
uint8_t& restart_point) {
uint8_t entry = index.Lookup(data, map_offset, key);
if (entry == kCollision) {
return true;
}
if (entry == kNoEntry) {
return false;
}
return entry == restart_point;
}
// Random KV generator similer to block_test
static std::string RandomString(Random* rnd, int len) {
std::string r;
test::RandomString(rnd, len, &r);
return r;
}
std::string GenerateKey(int primary_key, int secondary_key, int padding_size,
Random* rnd) {
char buf[50];
char* p = &buf[0];
snprintf(buf, sizeof(buf), "%6d%4d", primary_key, secondary_key);
std::string k(p);
if (padding_size) {
k += RandomString(rnd, padding_size);
}
return k;
}
// Generate random key value pairs.
// The generated key will be sorted. You can tune the parameters to generated
// different kinds of test key/value pairs for different scenario.
void GenerateRandomKVs(std::vector<std::string>* keys,
std::vector<std::string>* values, const int from,
const int len, const int step = 1,
const int padding_size = 0,
const int keys_share_prefix = 1) {
Random rnd(302);
// generate different prefix
for (int i = from; i < from + len; i += step) {
// generating keys that shares the prefix
for (int j = 0; j < keys_share_prefix; ++j) {
keys->emplace_back(GenerateKey(i, j, padding_size, &rnd));
// 100 bytes values
values->emplace_back(RandomString(&rnd, 100));
}
}
}
TEST(DataBlockHashIndex, DataBlockHashTestSmall) {
DataBlockHashIndexBuilder builder;
builder.Initialize(0.75 /*util_ratio*/);
for (int j = 0; j < 5; j++) {
for (uint8_t i = 0; i < 2 + j; i++) {
std::string key("key" + std::to_string(i));
uint8_t restart_point = i;
builder.Add(key, restart_point);
}
size_t estimated_size = builder.EstimateSize();
std::string buffer("fake"), buffer2;
size_t original_size = buffer.size();
estimated_size += original_size;
builder.Finish(buffer);
ASSERT_EQ(buffer.size(), estimated_size);
buffer2 = buffer; // test for the correctness of relative offset
Slice s(buffer2);
DataBlockHashIndex index;
uint16_t map_offset;
index.Initialize(s.data(), static_cast<uint16_t>(s.size()), &map_offset);
// the additional hash map should start at the end of the buffer
ASSERT_EQ(original_size, map_offset);
for (uint8_t i = 0; i < 2; i++) {
std::string key("key" + std::to_string(i));
uint8_t restart_point = i;
ASSERT_TRUE(
SearchForOffset(index, s.data(), map_offset, key, restart_point));
}
builder.Reset();
}
}
TEST(DataBlockHashIndex, DataBlockHashTest) {
// bucket_num = 200, #keys = 100. 50% utilization
DataBlockHashIndexBuilder builder;
builder.Initialize(0.75 /*util_ratio*/);
for (uint8_t i = 0; i < 100; i++) {
std::string key("key" + std::to_string(i));
uint8_t restart_point = i;
builder.Add(key, restart_point);
}
size_t estimated_size = builder.EstimateSize();
std::string buffer("fake content"), buffer2;
size_t original_size = buffer.size();
estimated_size += original_size;
builder.Finish(buffer);
ASSERT_EQ(buffer.size(), estimated_size);
buffer2 = buffer; // test for the correctness of relative offset
Slice s(buffer2);
DataBlockHashIndex index;
uint16_t map_offset;
index.Initialize(s.data(), static_cast<uint16_t>(s.size()), &map_offset);
// the additional hash map should start at the end of the buffer
ASSERT_EQ(original_size, map_offset);
for (uint8_t i = 0; i < 100; i++) {
std::string key("key" + std::to_string(i));
uint8_t restart_point = i;
ASSERT_TRUE(
SearchForOffset(index, s.data(), map_offset, key, restart_point));
}
}
TEST(DataBlockHashIndex, DataBlockHashTestCollision) {
// bucket_num = 2. There will be intense hash collisions
DataBlockHashIndexBuilder builder;
builder.Initialize(0.75 /*util_ratio*/);
for (uint8_t i = 0; i < 100; i++) {
std::string key("key" + std::to_string(i));
uint8_t restart_point = i;
builder.Add(key, restart_point);
}
size_t estimated_size = builder.EstimateSize();
std::string buffer("some other fake content to take up space"), buffer2;
size_t original_size = buffer.size();
estimated_size += original_size;
builder.Finish(buffer);
ASSERT_EQ(buffer.size(), estimated_size);
buffer2 = buffer; // test for the correctness of relative offset
Slice s(buffer2);
DataBlockHashIndex index;
uint16_t map_offset;
index.Initialize(s.data(), static_cast<uint16_t>(s.size()), &map_offset);
// the additional hash map should start at the end of the buffer
ASSERT_EQ(original_size, map_offset);
for (uint8_t i = 0; i < 100; i++) {
std::string key("key" + std::to_string(i));
uint8_t restart_point = i;
ASSERT_TRUE(
SearchForOffset(index, s.data(), map_offset, key, restart_point));
}
}
TEST(DataBlockHashIndex, DataBlockHashTestLarge) {
DataBlockHashIndexBuilder builder;
builder.Initialize(0.75 /*util_ratio*/);
std::unordered_map<std::string, uint8_t> m;
for (uint8_t i = 0; i < 100; i++) {
if (i % 2) {
continue; // leave half of the keys out
}
std::string key = "key" + std::to_string(i);
uint8_t restart_point = i;
builder.Add(key, restart_point);
m[key] = restart_point;
}
size_t estimated_size = builder.EstimateSize();
std::string buffer("filling stuff"), buffer2;
size_t original_size = buffer.size();
estimated_size += original_size;
builder.Finish(buffer);
ASSERT_EQ(buffer.size(), estimated_size);
buffer2 = buffer; // test for the correctness of relative offset
Slice s(buffer2);
DataBlockHashIndex index;
uint16_t map_offset;
index.Initialize(s.data(), static_cast<uint16_t>(s.size()), &map_offset);
// the additional hash map should start at the end of the buffer
ASSERT_EQ(original_size, map_offset);
for (uint8_t i = 0; i < 100; i++) {
std::string key = "key" + std::to_string(i);
uint8_t restart_point = i;
if (m.count(key)) {
ASSERT_TRUE(m[key] == restart_point);
ASSERT_TRUE(
SearchForOffset(index, s.data(), map_offset, key, restart_point));
} else {
// we allow false positve, so don't test the nonexisting keys.
// when false positive happens, the search will continue to the
// restart intervals to see if the key really exist.
}
}
}
TEST(DataBlockHashIndex, RestartIndexExceedMax) {
DataBlockHashIndexBuilder builder;
builder.Initialize(0.75 /*util_ratio*/);
std::unordered_map<std::string, uint8_t> m;
for (uint8_t i = 0; i <= 253; i++) {
std::string key = "key" + std::to_string(i);
uint8_t restart_point = i;
builder.Add(key, restart_point);
}
ASSERT_TRUE(builder.Valid());
builder.Reset();
for (uint8_t i = 0; i <= 254; i++) {
std::string key = "key" + std::to_string(i);
uint8_t restart_point = i;
builder.Add(key, restart_point);
}
ASSERT_FALSE(builder.Valid());
builder.Reset();
ASSERT_TRUE(builder.Valid());
}
TEST(DataBlockHashIndex, BlockRestartIndexExceedMax) {
Options options = Options();
BlockBuilder builder(1 /* block_restart_interval */,
true /* use_delta_encoding */,
false /* use_value_delta_encoding */,
BlockBasedTableOptions::kDataBlockBinaryAndHash);
// #restarts <= 253. HashIndex is valid
for (int i = 0; i <= 253; i++) {
std::string ukey = "key" + std::to_string(i);
InternalKey ikey(ukey, 0, kTypeValue);
builder.Add(ikey.Encode().ToString(), "value");
}
{
// read serialized contents of the block
Slice rawblock = builder.Finish();
// create block reader
BlockContents contents;
contents.data = rawblock;
Block reader(std::move(contents), kDisableGlobalSequenceNumber);
ASSERT_EQ(reader.IndexType(),
BlockBasedTableOptions::kDataBlockBinaryAndHash);
}
builder.Reset();
// #restarts > 253. HashIndex is not used
for (int i = 0; i <= 254; i++) {
std::string ukey = "key" + std::to_string(i);
InternalKey ikey(ukey, 0, kTypeValue);
builder.Add(ikey.Encode().ToString(), "value");
}
{
// read serialized contents of the block
Slice rawblock = builder.Finish();
// create block reader
BlockContents contents;
contents.data = rawblock;
Block reader(std::move(contents), kDisableGlobalSequenceNumber);
ASSERT_EQ(reader.IndexType(),
BlockBasedTableOptions::kDataBlockBinarySearch);
}
}
TEST(DataBlockHashIndex, BlockSizeExceedMax) {
Options options = Options();
std::string ukey(10, 'k');
InternalKey ikey(ukey, 0, kTypeValue);
BlockBuilder builder(1 /* block_restart_interval */,
false /* use_delta_encoding */,
false /* use_value_delta_encoding */,
BlockBasedTableOptions::kDataBlockBinaryAndHash);
{
// insert a large value. The block size plus HashIndex is 65536.
std::string value(65502, 'v');
builder.Add(ikey.Encode().ToString(), value);
// read serialized contents of the block
Slice rawblock = builder.Finish();
ASSERT_LE(rawblock.size(), kMaxBlockSizeSupportedByHashIndex);
std::cerr << "block size: " << rawblock.size() << std::endl;
// create block reader
BlockContents contents;
contents.data = rawblock;
Block reader(std::move(contents), kDisableGlobalSequenceNumber);
ASSERT_EQ(reader.IndexType(),
BlockBasedTableOptions::kDataBlockBinaryAndHash);
}
builder.Reset();
{
// insert a large value. The block size plus HashIndex would be 65537.
// This excceed the max block size supported by HashIndex (65536).
// So when build finishes HashIndex will not be created for the block.
std::string value(65503, 'v');
builder.Add(ikey.Encode().ToString(), value);
// read serialized contents of the block
Slice rawblock = builder.Finish();
ASSERT_LE(rawblock.size(), kMaxBlockSizeSupportedByHashIndex);
std::cerr << "block size: " << rawblock.size() << std::endl;
// create block reader
BlockContents contents;
contents.data = rawblock;
Block reader(std::move(contents), kDisableGlobalSequenceNumber);
// the index type have fallen back to binary when build finish.
ASSERT_EQ(reader.IndexType(),
BlockBasedTableOptions::kDataBlockBinarySearch);
}
}
TEST(DataBlockHashIndex, BlockTestSingleKey) {
Options options = Options();
BlockBuilder builder(16 /* block_restart_interval */,
true /* use_delta_encoding */,
false /* use_value_delta_encoding */,
BlockBasedTableOptions::kDataBlockBinaryAndHash);
std::string ukey("gopher");
std::string value("gold");
InternalKey ikey(ukey, 10, kTypeValue);
builder.Add(ikey.Encode().ToString(), value /*value*/);
// read serialized contents of the block
Slice rawblock = builder.Finish();
// create block reader
BlockContents contents;
contents.data = rawblock;
Block reader(std::move(contents), kDisableGlobalSequenceNumber);
const InternalKeyComparator icmp(BytewiseComparator());
Add an option to put first key of each sst block in the index (#5289) Summary: The first key is used to defer reading the data block until this file gets to the top of merging iterator's heap. For short range scans, most files never make it to the top of the heap, so this change can reduce read amplification by a lot sometimes. Consider the following workload. There are a few data streams (we'll be calling them "logs"), each stream consisting of a sequence of blobs (we'll be calling them "records"). Each record is identified by log ID and a sequence number within the log. RocksDB key is concatenation of log ID and sequence number (big endian). Reads are mostly relatively short range scans, each within a single log. Writes are mostly sequential for each log, but writes to different logs are randomly interleaved. Compactions are disabled; instead, when we accumulate a few tens of sst files, we create a new column family and start writing to it. So, a typical sst file consists of a few ranges of blocks, each range corresponding to one log ID (we use FlushBlockPolicy to cut blocks at log boundaries). A typical read would go like this. First, iterator Seek() reads one block from each sst file. Then a series of Next()s move through one sst file (since writes to each log are mostly sequential) until the subiterator reaches the end of this log in this sst file; then Next() switches to the next sst file and reads sequentially from that, and so on. Often a range scan will only return records from a small number of blocks in small number of sst files; in this case, the cost of initial Seek() reading one block from each file may be bigger than the cost of reading the actually useful blocks. Neither iterate_upper_bound nor bloom filters can prevent reading one block from each file in Seek(). But this PR can: if the index contains first key from each block, we don't have to read the block until this block actually makes it to the top of merging iterator's heap, so for short range scans we won't read any blocks from most of the sst files. This PR does the deferred block loading inside value() call. This is not ideal: there's no good way to report an IO error from inside value(). As discussed with siying offline, it would probably be better to change InternalIterator's interface to explicitly fetch deferred value and get status. I'll do it in a separate PR. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5289 Differential Revision: D15256423 Pulled By: al13n321 fbshipit-source-id: 750e4c39ce88e8d41662f701cf6275d9388ba46a
5 years ago
auto iter = reader.NewDataIterator(&icmp, icmp.user_comparator());
bool may_exist;
// search in block for the key just inserted
{
InternalKey seek_ikey(ukey, 10, kValueTypeForSeek);
may_exist = iter->SeekForGet(seek_ikey.Encode().ToString());
ASSERT_TRUE(may_exist);
ASSERT_TRUE(iter->Valid());
ASSERT_EQ(
options.comparator->Compare(iter->key(), ikey.Encode().ToString()), 0);
ASSERT_EQ(iter->value(), value);
}
// search in block for the existing ukey, but with higher seqno
{
InternalKey seek_ikey(ukey, 20, kValueTypeForSeek);
// HashIndex should be able to set the iter correctly
may_exist = iter->SeekForGet(seek_ikey.Encode().ToString());
ASSERT_TRUE(may_exist);
ASSERT_TRUE(iter->Valid());
// user key should match
ASSERT_EQ(options.comparator->Compare(ExtractUserKey(iter->key()), ukey),
0);
// seek_key seqno number should be greater than that of iter result
ASSERT_GT(GetInternalKeySeqno(seek_ikey.Encode()),
GetInternalKeySeqno(iter->key()));
ASSERT_EQ(iter->value(), value);
}
// Search in block for the existing ukey, but with lower seqno
// in this case, hash can find the only occurrence of the user_key, but
// ParseNextDataKey() will skip it as it does not have a older seqno.
// In this case, GetForSeek() is effective to locate the user_key, and
// iter->Valid() == false indicates that we've reached to the end of
// the block and the caller should continue searching the next block.
{
InternalKey seek_ikey(ukey, 5, kValueTypeForSeek);
may_exist = iter->SeekForGet(seek_ikey.Encode().ToString());
ASSERT_TRUE(may_exist);
ASSERT_FALSE(iter->Valid()); // should have reached to the end of block
}
delete iter;
}
TEST(DataBlockHashIndex, BlockTestLarge) {
Random rnd(1019);
Options options = Options();
std::vector<std::string> keys;
std::vector<std::string> values;
BlockBuilder builder(16 /* block_restart_interval */,
true /* use_delta_encoding */,
false /* use_value_delta_encoding */,
BlockBasedTableOptions::kDataBlockBinaryAndHash);
int num_records = 500;
GenerateRandomKVs(&keys, &values, 0, num_records);
// Generate keys. Adding a trailing "1" to indicate existent keys.
// Later will Seeking for keys with a trailing "0" to test seeking
// non-existent keys.
for (int i = 0; i < num_records; i++) {
std::string ukey(keys[i] + "1" /* existing key marker */);
InternalKey ikey(ukey, 0, kTypeValue);
builder.Add(ikey.Encode().ToString(), values[i]);
}
// read serialized contents of the block
Slice rawblock = builder.Finish();
// create block reader
BlockContents contents;
contents.data = rawblock;
Block reader(std::move(contents), kDisableGlobalSequenceNumber);
const InternalKeyComparator icmp(BytewiseComparator());
// random seek existent keys
for (int i = 0; i < num_records; i++) {
Add an option to put first key of each sst block in the index (#5289) Summary: The first key is used to defer reading the data block until this file gets to the top of merging iterator's heap. For short range scans, most files never make it to the top of the heap, so this change can reduce read amplification by a lot sometimes. Consider the following workload. There are a few data streams (we'll be calling them "logs"), each stream consisting of a sequence of blobs (we'll be calling them "records"). Each record is identified by log ID and a sequence number within the log. RocksDB key is concatenation of log ID and sequence number (big endian). Reads are mostly relatively short range scans, each within a single log. Writes are mostly sequential for each log, but writes to different logs are randomly interleaved. Compactions are disabled; instead, when we accumulate a few tens of sst files, we create a new column family and start writing to it. So, a typical sst file consists of a few ranges of blocks, each range corresponding to one log ID (we use FlushBlockPolicy to cut blocks at log boundaries). A typical read would go like this. First, iterator Seek() reads one block from each sst file. Then a series of Next()s move through one sst file (since writes to each log are mostly sequential) until the subiterator reaches the end of this log in this sst file; then Next() switches to the next sst file and reads sequentially from that, and so on. Often a range scan will only return records from a small number of blocks in small number of sst files; in this case, the cost of initial Seek() reading one block from each file may be bigger than the cost of reading the actually useful blocks. Neither iterate_upper_bound nor bloom filters can prevent reading one block from each file in Seek(). But this PR can: if the index contains first key from each block, we don't have to read the block until this block actually makes it to the top of merging iterator's heap, so for short range scans we won't read any blocks from most of the sst files. This PR does the deferred block loading inside value() call. This is not ideal: there's no good way to report an IO error from inside value(). As discussed with siying offline, it would probably be better to change InternalIterator's interface to explicitly fetch deferred value and get status. I'll do it in a separate PR. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5289 Differential Revision: D15256423 Pulled By: al13n321 fbshipit-source-id: 750e4c39ce88e8d41662f701cf6275d9388ba46a
5 years ago
auto iter = reader.NewDataIterator(&icmp, icmp.user_comparator());
// find a random key in the lookaside array
int index = rnd.Uniform(num_records);
std::string ukey(keys[index] + "1" /* existing key marker */);
InternalKey ikey(ukey, 0, kTypeValue);
// search in block for this key
bool may_exist = iter->SeekForGet(ikey.Encode().ToString());
ASSERT_TRUE(may_exist);
ASSERT_TRUE(iter->Valid());
ASSERT_EQ(values[index], iter->value());
delete iter;
}
// random seek non-existent user keys
// In this case A), the user_key cannot be found in HashIndex. The key may
// exist in the next block. So the iter is set invalidated to tell the
// caller to search the next block. This test case belongs to this case A).
//
// Note that for non-existent keys, there is possibility of false positive,
// i.e. the key is still hashed into some restart interval.
// Two additional possible outcome:
// B) linear seek the restart interval and not found, the iter stops at the
// starting of the next restart interval. The key does not exist
// anywhere.
// C) linear seek the restart interval and not found, the iter stops at the
// the end of the block, i.e. restarts_. The key may exist in the next
// block.
// So these combinations are possible when searching non-existent user_key:
//
// case# may_exist iter->Valid()
// A true false
// B false true
// C true false
for (int i = 0; i < num_records; i++) {
Add an option to put first key of each sst block in the index (#5289) Summary: The first key is used to defer reading the data block until this file gets to the top of merging iterator's heap. For short range scans, most files never make it to the top of the heap, so this change can reduce read amplification by a lot sometimes. Consider the following workload. There are a few data streams (we'll be calling them "logs"), each stream consisting of a sequence of blobs (we'll be calling them "records"). Each record is identified by log ID and a sequence number within the log. RocksDB key is concatenation of log ID and sequence number (big endian). Reads are mostly relatively short range scans, each within a single log. Writes are mostly sequential for each log, but writes to different logs are randomly interleaved. Compactions are disabled; instead, when we accumulate a few tens of sst files, we create a new column family and start writing to it. So, a typical sst file consists of a few ranges of blocks, each range corresponding to one log ID (we use FlushBlockPolicy to cut blocks at log boundaries). A typical read would go like this. First, iterator Seek() reads one block from each sst file. Then a series of Next()s move through one sst file (since writes to each log are mostly sequential) until the subiterator reaches the end of this log in this sst file; then Next() switches to the next sst file and reads sequentially from that, and so on. Often a range scan will only return records from a small number of blocks in small number of sst files; in this case, the cost of initial Seek() reading one block from each file may be bigger than the cost of reading the actually useful blocks. Neither iterate_upper_bound nor bloom filters can prevent reading one block from each file in Seek(). But this PR can: if the index contains first key from each block, we don't have to read the block until this block actually makes it to the top of merging iterator's heap, so for short range scans we won't read any blocks from most of the sst files. This PR does the deferred block loading inside value() call. This is not ideal: there's no good way to report an IO error from inside value(). As discussed with siying offline, it would probably be better to change InternalIterator's interface to explicitly fetch deferred value and get status. I'll do it in a separate PR. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5289 Differential Revision: D15256423 Pulled By: al13n321 fbshipit-source-id: 750e4c39ce88e8d41662f701cf6275d9388ba46a
5 years ago
auto iter = reader.NewDataIterator(&icmp, icmp.user_comparator());
// find a random key in the lookaside array
int index = rnd.Uniform(num_records);
std::string ukey(keys[index] + "0" /* non-existing key marker */);
InternalKey ikey(ukey, 0, kTypeValue);
// search in block for this key
bool may_exist = iter->SeekForGet(ikey.Encode().ToString());
if (!may_exist) {
ASSERT_TRUE(iter->Valid());
}
if (!iter->Valid()) {
ASSERT_TRUE(may_exist);
}
delete iter;
}
}
// helper routine for DataBlockHashIndex.BlockBoundary
void TestBoundary(InternalKey& ik1, std::string& v1, InternalKey& ik2,
std::string& v2, InternalKey& seek_ikey,
GetContext& get_context, Options& options) {
std::unique_ptr<WritableFileWriter> file_writer;
std::unique_ptr<RandomAccessFileReader> file_reader;
std::unique_ptr<TableReader> table_reader;
int level_ = -1;
std::vector<std::string> keys;
const ImmutableCFOptions ioptions(options);
const MutableCFOptions moptions(options);
const InternalKeyComparator internal_comparator(options.comparator);
EnvOptions soptions;
soptions.use_mmap_reads = ioptions.allow_mmap_reads;
file_writer.reset(
test::GetWritableFileWriter(new test::StringSink(), "" /* don't care */));
std::unique_ptr<TableBuilder> builder;
std::vector<std::unique_ptr<IntTblPropCollectorFactory>>
int_tbl_prop_collector_factories;
std::string column_family_name;
builder.reset(ioptions.table_factory->NewTableBuilder(
TableBuilderOptions(ioptions, moptions, internal_comparator,
&int_tbl_prop_collector_factories,
options.compression, options.sample_for_compression,
CompressionOptions(), false /* skip_filters */,
column_family_name, level_),
TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
file_writer.get()));
builder->Add(ik1.Encode().ToString(), v1);
builder->Add(ik2.Encode().ToString(), v2);
EXPECT_TRUE(builder->status().ok());
Status s = builder->Finish();
file_writer->Flush();
EXPECT_TRUE(s.ok()) << s.ToString();
Introduce a new storage specific Env API (#5761) Summary: The current Env API encompasses both storage/file operations, as well as OS related operations. Most of the APIs return a Status, which does not have enough metadata about an error, such as whether its retry-able or not, scope (i.e fault domain) of the error etc., that may be required in order to properly handle a storage error. The file APIs also do not provide enough control over the IO SLA, such as timeout, prioritization, hinting about placement and redundancy etc. This PR separates out the file/storage APIs from Env into a new FileSystem class. The APIs are updated to return an IOStatus with metadata about the error, as well as to take an IOOptions structure as input in order to allow more control over the IO. The user can set both ```options.env``` and ```options.file_system``` to specify that RocksDB should use the former for OS related operations and the latter for storage operations. Internally, a ```CompositeEnvWrapper``` has been introduced that inherits from ```Env``` and redirects individual methods to either an ```Env``` implementation or the ```FileSystem``` as appropriate. When options are sanitized during ```DB::Open```, ```options.env``` is replaced with a newly allocated ```CompositeEnvWrapper``` instance if both env and file_system have been specified. This way, the rest of the RocksDB code can continue to function as before. This PR also ports PosixEnv to the new API by splitting it into two - PosixEnv and PosixFileSystem. PosixEnv is defined as a sub-class of CompositeEnvWrapper, and threading/time functions are overridden with Posix specific implementations in order to avoid an extra level of indirection. The ```CompositeEnvWrapper``` translates ```IOStatus``` return code to ```Status```, and sets the severity to ```kSoftError``` if the io_status is retryable. The error handling code in RocksDB can then recover the DB automatically. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5761 Differential Revision: D18868376 Pulled By: anand1976 fbshipit-source-id: 39efe18a162ea746fabac6360ff529baba48486f
5 years ago
EXPECT_EQ(
test::GetStringSinkFromLegacyWriter(file_writer.get())->contents().size(),
builder->FileSize());
// Open the table
file_reader.reset(test::GetRandomAccessFileReader(new test::StringSource(
Introduce a new storage specific Env API (#5761) Summary: The current Env API encompasses both storage/file operations, as well as OS related operations. Most of the APIs return a Status, which does not have enough metadata about an error, such as whether its retry-able or not, scope (i.e fault domain) of the error etc., that may be required in order to properly handle a storage error. The file APIs also do not provide enough control over the IO SLA, such as timeout, prioritization, hinting about placement and redundancy etc. This PR separates out the file/storage APIs from Env into a new FileSystem class. The APIs are updated to return an IOStatus with metadata about the error, as well as to take an IOOptions structure as input in order to allow more control over the IO. The user can set both ```options.env``` and ```options.file_system``` to specify that RocksDB should use the former for OS related operations and the latter for storage operations. Internally, a ```CompositeEnvWrapper``` has been introduced that inherits from ```Env``` and redirects individual methods to either an ```Env``` implementation or the ```FileSystem``` as appropriate. When options are sanitized during ```DB::Open```, ```options.env``` is replaced with a newly allocated ```CompositeEnvWrapper``` instance if both env and file_system have been specified. This way, the rest of the RocksDB code can continue to function as before. This PR also ports PosixEnv to the new API by splitting it into two - PosixEnv and PosixFileSystem. PosixEnv is defined as a sub-class of CompositeEnvWrapper, and threading/time functions are overridden with Posix specific implementations in order to avoid an extra level of indirection. The ```CompositeEnvWrapper``` translates ```IOStatus``` return code to ```Status```, and sets the severity to ```kSoftError``` if the io_status is retryable. The error handling code in RocksDB can then recover the DB automatically. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5761 Differential Revision: D18868376 Pulled By: anand1976 fbshipit-source-id: 39efe18a162ea746fabac6360ff529baba48486f
5 years ago
test::GetStringSinkFromLegacyWriter(file_writer.get())->contents(),
0 /*uniq_id*/, ioptions.allow_mmap_reads)));
const bool kSkipFilters = true;
const bool kImmortal = true;
ioptions.table_factory->NewTableReader(
TableReaderOptions(ioptions, moptions.prefix_extractor.get(), soptions,
internal_comparator, !kSkipFilters, !kImmortal,
level_),
std::move(file_reader),
Introduce a new storage specific Env API (#5761) Summary: The current Env API encompasses both storage/file operations, as well as OS related operations. Most of the APIs return a Status, which does not have enough metadata about an error, such as whether its retry-able or not, scope (i.e fault domain) of the error etc., that may be required in order to properly handle a storage error. The file APIs also do not provide enough control over the IO SLA, such as timeout, prioritization, hinting about placement and redundancy etc. This PR separates out the file/storage APIs from Env into a new FileSystem class. The APIs are updated to return an IOStatus with metadata about the error, as well as to take an IOOptions structure as input in order to allow more control over the IO. The user can set both ```options.env``` and ```options.file_system``` to specify that RocksDB should use the former for OS related operations and the latter for storage operations. Internally, a ```CompositeEnvWrapper``` has been introduced that inherits from ```Env``` and redirects individual methods to either an ```Env``` implementation or the ```FileSystem``` as appropriate. When options are sanitized during ```DB::Open```, ```options.env``` is replaced with a newly allocated ```CompositeEnvWrapper``` instance if both env and file_system have been specified. This way, the rest of the RocksDB code can continue to function as before. This PR also ports PosixEnv to the new API by splitting it into two - PosixEnv and PosixFileSystem. PosixEnv is defined as a sub-class of CompositeEnvWrapper, and threading/time functions are overridden with Posix specific implementations in order to avoid an extra level of indirection. The ```CompositeEnvWrapper``` translates ```IOStatus``` return code to ```Status```, and sets the severity to ```kSoftError``` if the io_status is retryable. The error handling code in RocksDB can then recover the DB automatically. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5761 Differential Revision: D18868376 Pulled By: anand1976 fbshipit-source-id: 39efe18a162ea746fabac6360ff529baba48486f
5 years ago
test::GetStringSinkFromLegacyWriter(file_writer.get())->contents().size(),
&table_reader);
// Search using Get()
ReadOptions ro;
ASSERT_OK(table_reader->Get(ro, seek_ikey.Encode().ToString(), &get_context,
moptions.prefix_extractor.get()));
}
TEST(DataBlockHashIndex, BlockBoundary) {
BlockBasedTableOptions table_options;
table_options.data_block_index_type =
BlockBasedTableOptions::kDataBlockBinaryAndHash;
table_options.block_restart_interval = 1;
table_options.block_size = 4096;
Options options;
options.comparator = BytewiseComparator();
options.table_factory.reset(NewBlockBasedTableFactory(table_options));
// insert two large k/v pair. Given that the block_size is 4096, one k/v
// pair will take up one block.
// [ k1/v1 ][ k2/v2 ]
// [ Block N ][ Block N+1 ]
{
// [ "aab"@100 ][ "axy"@10 ]
// | Block N ][ Block N+1 ]
// seek for "axy"@60
std::string uk1("aab");
InternalKey ik1(uk1, 100, kTypeValue);
std::string v1(4100, '1'); // large value
std::string uk2("axy");
InternalKey ik2(uk2, 10, kTypeValue);
std::string v2(4100, '2'); // large value
PinnableSlice value;
std::string seek_ukey("axy");
InternalKey seek_ikey(seek_ukey, 60, kTypeValue);
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
GetContext::kNotFound, seek_ukey, &value, nullptr,
New API to get all merge operands for a Key (#5604) Summary: This is a new API added to db.h to allow for fetching all merge operands associated with a Key. The main motivation for this API is to support use cases where doing a full online merge is not necessary as it is performance sensitive. Example use-cases: 1. Update subset of columns and read subset of columns - Imagine a SQL Table, a row is encoded as a K/V pair (as it is done in MyRocks). If there are many columns and users only updated one of them, we can use merge operator to reduce write amplification. While users only read one or two columns in the read query, this feature can avoid a full merging of the whole row, and save some CPU. 2. Updating very few attributes in a value which is a JSON-like document - Updating one attribute can be done efficiently using merge operator, while reading back one attribute can be done more efficiently if we don't need to do a full merge. ---------------------------------------------------------------------------------------------------- API : Status GetMergeOperands( const ReadOptions& options, ColumnFamilyHandle* column_family, const Slice& key, PinnableSlice* merge_operands, GetMergeOperandsOptions* get_merge_operands_options, int* number_of_operands) Example usage : int size = 100; int number_of_operands = 0; std::vector<PinnableSlice> values(size); GetMergeOperandsOptions merge_operands_info; db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), "k1", values.data(), merge_operands_info, &number_of_operands); Description : Returns all the merge operands corresponding to the key. If the number of merge operands in DB is greater than merge_operands_options.expected_max_number_of_operands no merge operands are returned and status is Incomplete. Merge operands returned are in the order of insertion. merge_operands-> Points to an array of at-least merge_operands_options.expected_max_number_of_operands and the caller is responsible for allocating it. If the status returned is Incomplete then number_of_operands will contain the total number of merge operands found in DB for key. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5604 Test Plan: Added unit test and perf test in db_bench that can be run using the command: ./db_bench -benchmarks=getmergeoperands --merge_operator=sortlist Differential Revision: D16657366 Pulled By: vjnadimpalli fbshipit-source-id: 0faadd752351745224ee12d4ae9ef3cb529951bf
5 years ago
nullptr, true, nullptr, nullptr);
TestBoundary(ik1, v1, ik2, v2, seek_ikey, get_context, options);
ASSERT_EQ(get_context.State(), GetContext::kFound);
ASSERT_EQ(value, v2);
value.Reset();
}
{
// [ "axy"@100 ][ "axy"@10 ]
// | Block N ][ Block N+1 ]
// seek for "axy"@60
std::string uk1("axy");
InternalKey ik1(uk1, 100, kTypeValue);
std::string v1(4100, '1'); // large value
std::string uk2("axy");
InternalKey ik2(uk2, 10, kTypeValue);
std::string v2(4100, '2'); // large value
PinnableSlice value;
std::string seek_ukey("axy");
InternalKey seek_ikey(seek_ukey, 60, kTypeValue);
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
GetContext::kNotFound, seek_ukey, &value, nullptr,
New API to get all merge operands for a Key (#5604) Summary: This is a new API added to db.h to allow for fetching all merge operands associated with a Key. The main motivation for this API is to support use cases where doing a full online merge is not necessary as it is performance sensitive. Example use-cases: 1. Update subset of columns and read subset of columns - Imagine a SQL Table, a row is encoded as a K/V pair (as it is done in MyRocks). If there are many columns and users only updated one of them, we can use merge operator to reduce write amplification. While users only read one or two columns in the read query, this feature can avoid a full merging of the whole row, and save some CPU. 2. Updating very few attributes in a value which is a JSON-like document - Updating one attribute can be done efficiently using merge operator, while reading back one attribute can be done more efficiently if we don't need to do a full merge. ---------------------------------------------------------------------------------------------------- API : Status GetMergeOperands( const ReadOptions& options, ColumnFamilyHandle* column_family, const Slice& key, PinnableSlice* merge_operands, GetMergeOperandsOptions* get_merge_operands_options, int* number_of_operands) Example usage : int size = 100; int number_of_operands = 0; std::vector<PinnableSlice> values(size); GetMergeOperandsOptions merge_operands_info; db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), "k1", values.data(), merge_operands_info, &number_of_operands); Description : Returns all the merge operands corresponding to the key. If the number of merge operands in DB is greater than merge_operands_options.expected_max_number_of_operands no merge operands are returned and status is Incomplete. Merge operands returned are in the order of insertion. merge_operands-> Points to an array of at-least merge_operands_options.expected_max_number_of_operands and the caller is responsible for allocating it. If the status returned is Incomplete then number_of_operands will contain the total number of merge operands found in DB for key. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5604 Test Plan: Added unit test and perf test in db_bench that can be run using the command: ./db_bench -benchmarks=getmergeoperands --merge_operator=sortlist Differential Revision: D16657366 Pulled By: vjnadimpalli fbshipit-source-id: 0faadd752351745224ee12d4ae9ef3cb529951bf
5 years ago
nullptr, true, nullptr, nullptr);
TestBoundary(ik1, v1, ik2, v2, seek_ikey, get_context, options);
ASSERT_EQ(get_context.State(), GetContext::kFound);
ASSERT_EQ(value, v2);
value.Reset();
}
{
// [ "axy"@100 ][ "axy"@10 ]
// | Block N ][ Block N+1 ]
// seek for "axy"@120
std::string uk1("axy");
InternalKey ik1(uk1, 100, kTypeValue);
std::string v1(4100, '1'); // large value
std::string uk2("axy");
InternalKey ik2(uk2, 10, kTypeValue);
std::string v2(4100, '2'); // large value
PinnableSlice value;
std::string seek_ukey("axy");
InternalKey seek_ikey(seek_ukey, 120, kTypeValue);
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
GetContext::kNotFound, seek_ukey, &value, nullptr,
New API to get all merge operands for a Key (#5604) Summary: This is a new API added to db.h to allow for fetching all merge operands associated with a Key. The main motivation for this API is to support use cases where doing a full online merge is not necessary as it is performance sensitive. Example use-cases: 1. Update subset of columns and read subset of columns - Imagine a SQL Table, a row is encoded as a K/V pair (as it is done in MyRocks). If there are many columns and users only updated one of them, we can use merge operator to reduce write amplification. While users only read one or two columns in the read query, this feature can avoid a full merging of the whole row, and save some CPU. 2. Updating very few attributes in a value which is a JSON-like document - Updating one attribute can be done efficiently using merge operator, while reading back one attribute can be done more efficiently if we don't need to do a full merge. ---------------------------------------------------------------------------------------------------- API : Status GetMergeOperands( const ReadOptions& options, ColumnFamilyHandle* column_family, const Slice& key, PinnableSlice* merge_operands, GetMergeOperandsOptions* get_merge_operands_options, int* number_of_operands) Example usage : int size = 100; int number_of_operands = 0; std::vector<PinnableSlice> values(size); GetMergeOperandsOptions merge_operands_info; db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), "k1", values.data(), merge_operands_info, &number_of_operands); Description : Returns all the merge operands corresponding to the key. If the number of merge operands in DB is greater than merge_operands_options.expected_max_number_of_operands no merge operands are returned and status is Incomplete. Merge operands returned are in the order of insertion. merge_operands-> Points to an array of at-least merge_operands_options.expected_max_number_of_operands and the caller is responsible for allocating it. If the status returned is Incomplete then number_of_operands will contain the total number of merge operands found in DB for key. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5604 Test Plan: Added unit test and perf test in db_bench that can be run using the command: ./db_bench -benchmarks=getmergeoperands --merge_operator=sortlist Differential Revision: D16657366 Pulled By: vjnadimpalli fbshipit-source-id: 0faadd752351745224ee12d4ae9ef3cb529951bf
5 years ago
nullptr, true, nullptr, nullptr);
TestBoundary(ik1, v1, ik2, v2, seek_ikey, get_context, options);
ASSERT_EQ(get_context.State(), GetContext::kFound);
ASSERT_EQ(value, v1);
value.Reset();
}
{
// [ "axy"@100 ][ "axy"@10 ]
// | Block N ][ Block N+1 ]
// seek for "axy"@5
std::string uk1("axy");
InternalKey ik1(uk1, 100, kTypeValue);
std::string v1(4100, '1'); // large value
std::string uk2("axy");
InternalKey ik2(uk2, 10, kTypeValue);
std::string v2(4100, '2'); // large value
PinnableSlice value;
std::string seek_ukey("axy");
InternalKey seek_ikey(seek_ukey, 5, kTypeValue);
GetContext get_context(options.comparator, nullptr, nullptr, nullptr,
GetContext::kNotFound, seek_ukey, &value, nullptr,
New API to get all merge operands for a Key (#5604) Summary: This is a new API added to db.h to allow for fetching all merge operands associated with a Key. The main motivation for this API is to support use cases where doing a full online merge is not necessary as it is performance sensitive. Example use-cases: 1. Update subset of columns and read subset of columns - Imagine a SQL Table, a row is encoded as a K/V pair (as it is done in MyRocks). If there are many columns and users only updated one of them, we can use merge operator to reduce write amplification. While users only read one or two columns in the read query, this feature can avoid a full merging of the whole row, and save some CPU. 2. Updating very few attributes in a value which is a JSON-like document - Updating one attribute can be done efficiently using merge operator, while reading back one attribute can be done more efficiently if we don't need to do a full merge. ---------------------------------------------------------------------------------------------------- API : Status GetMergeOperands( const ReadOptions& options, ColumnFamilyHandle* column_family, const Slice& key, PinnableSlice* merge_operands, GetMergeOperandsOptions* get_merge_operands_options, int* number_of_operands) Example usage : int size = 100; int number_of_operands = 0; std::vector<PinnableSlice> values(size); GetMergeOperandsOptions merge_operands_info; db_->GetMergeOperands(ReadOptions(), db_->DefaultColumnFamily(), "k1", values.data(), merge_operands_info, &number_of_operands); Description : Returns all the merge operands corresponding to the key. If the number of merge operands in DB is greater than merge_operands_options.expected_max_number_of_operands no merge operands are returned and status is Incomplete. Merge operands returned are in the order of insertion. merge_operands-> Points to an array of at-least merge_operands_options.expected_max_number_of_operands and the caller is responsible for allocating it. If the status returned is Incomplete then number_of_operands will contain the total number of merge operands found in DB for key. Pull Request resolved: https://github.com/facebook/rocksdb/pull/5604 Test Plan: Added unit test and perf test in db_bench that can be run using the command: ./db_bench -benchmarks=getmergeoperands --merge_operator=sortlist Differential Revision: D16657366 Pulled By: vjnadimpalli fbshipit-source-id: 0faadd752351745224ee12d4ae9ef3cb529951bf
5 years ago
nullptr, true, nullptr, nullptr);
TestBoundary(ik1, v1, ik2, v2, seek_ikey, get_context, options);
ASSERT_EQ(get_context.State(), GetContext::kNotFound);
value.Reset();
}
}
} // namespace ROCKSDB_NAMESPACE
int main(int argc, char** argv) {
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}