rocksdb/table/block_based/data_block_hash_index_test.cc
Peter Dillinger 9d490593d0 Preliminary support for custom compression algorithms (#13659)
Summary:
This change builds on https://github.com/facebook/rocksdb/issues/13540 and https://github.com/facebook/rocksdb/issues/13626 in allowing a CompressionManager / Compressor / Decompressor to use a custom compression algorithm, with a distinct CompressionType. For background, review the API comments on CompressionManager and its CompatibilityName() function.

Highlights:
* Reserve and name 127 new CompressionTypes that can be used for custom compression algorithms / schemas. In many or most cases I expect the enumerators such as `kCustomCompression8F` to be used in user code rather than casting between integers and CompressionTypes, as I expect the supported custom compression algorithms to be identifiable / enumerable at compile time.
* When using these custom compression types, a CompressionManager must use a CompatibilityName() other than the built-in one AND new format_version=7 (see below).
* When building new SST files, track the full set of CompressionTypes actually used (usually just one aside from kNoCompression), using our efficient bitset SmallEnumSet, which supports fast iteration over the bits set to 1. Ideally, to support mixed or non-mixed compression algorithms in a file as efficiently as possible, we would know the set of CompressionTypes as SST file open time.
* New schema for `TableProperties::compression_name` in format_version=7 to represent the CompressionManager's CompatibilityName(), the set of CompressionTypes used, and potentially more in the future, while keeping the data relatively human-readable.
  * It would be possible to do this without a new format_version, but then the only way to ensure incompatible versions fail is with an unsupported CompressionType tag, not with a compression_name property. Therefore, (a) I prefer not to put something misleading in the `compression_name` property (a built-in compression name) when there is nuance because of a CompressionManager, and (b) I prefer better, more consistent error messages that refer to either format_version or the CompressionManager's CompatibilityName(), rather than an unrecognized custom CompressionType value (which could have come from various CompressionManagers).
* The current configured CompressionManager is passed in to TableReaders so that it (or one it knows about) can be used if it matches the CompatibilityName() used for compression in the SST file. Until the connection with ObjectRegistry is implemented, the only way to read files generated with a particular CompressionManager using custom compression algorithms is to configure it (or a known relative; see FindCompatibleCompressionManager()) in the ColumnFamilyOptions.
* Optimized snappy compression with BuiltinDecompressorV2SnappyOnly, to offset some small added overheads with the new tracking. This is essentially an early part of the planned refactoring that will get rid of the old internal compression APIs.
* Another small optimization in eliminating an unnecessary key copy in flush (builder.cc).
* Fix some handling of named CompressionManagers in CompressionManager::CreateFromString() (problem seen in https://github.com/facebook/rocksdb/issues/13647)

Smaller things:
* Adds Name() and GetId() functions to Compressor for debugging/logging purposes. (Compressor and Decompressor are not expected to be Customizable because they are only instantiated by a CompressionManager.)
* When using an explicit compression_manager, the GetId() of the CompressionManager and the Compressor used to build the file are stored as bonus entries in the compression_options table property. This table property is not parsed anywhere, so it is currently for human reading, but still could be parsed with the new underscore-prefixed bonus entries. IMHO, this is preferable to additional table properties, which would increase memory fragmentation in the TableProperties objects and likely take slightly more CPU on SST open and slightly more storage.
* ReleaseWorkingArea() function from protected to public to make wrappers work, because of a quirk in C++ (vs. Java) in which you cannot access protected members of another instance of the same class (sigh)
* Added `CompressionManager:: SupportsCompressionType()` for early options sanity checking.

Follow-up before release:
* Make format_version=7 official / supported
* Stress test coverage

Sooner than later:
* Update tests for RoundRobinManager and SimpleMixedCompressionManager to take advantage of e.g. set of compression types in compression_name property
* ObjectRegistry stuff
* Refactor away old internal compression APIs

Pull Request resolved: https://github.com/facebook/rocksdb/pull/13659

Test Plan:
Basic unit test added.

## Performance

### SST write performance
```
SUFFIX=`tty | sed 's|/|_|g'`; for ARGS in "-compression_type=none" "-compression_type=snappy" "-compression_type=zstd" "-compression_type=snappy -verify_compression=1" "-compression_type=zstd -verify_compression=1" "-compression_type=zstd -compression_max_dict_bytes=8180"; do echo $ARGS; (for I in `seq 1 20`; do BIN=/dev/shm/dbbench${SUFFIX}.bin; rm -f $BIN; cp db_bench $BIN; $BIN -db=/dev/shm/dbbench$SUFFIX --benchmarks=fillseq -num=10000000 -compaction_style=2 -fifo_compaction_max_table_files_size_mb=1000 -fifo_compaction_allow_compaction=0 -disable_wal -write_buffer_size=12000000 -format_version=7 $ARGS 2>&1 | grep micros/op; done) | awk '{n++; sum += $5;} END { print int(sum / n); }'; done
```

Ops/sec, Before -> After, both fv=6:
-compression_type=none
1894386 -> 1858403 (-2.0%)
-compression_type=snappy
1859131 -> 1807469 (-2.8%)
-compression_type=zstd
1191428 -> 1214374 (+1.9%)
-compression_type=snappy -verify_compression=1
1861819 -> 1858342 (+0.2%)
-compression_type=zstd -verify_compression=1
979435 -> 995870 (+1.6%)
-compression_type=zstd -compression_max_dict_bytes=8180
905349 -> 940563 (+3.9%)

Ops/sec, Before fv=6 -> After fv=7:
-compression_type=none
1879365 -> 1836159 (-2.3%)
-compression_type=snappy
1865460 -> 1830916 (-1.9%)
-compression_type=zstd
1191428 -> 1210260 (+1.6%)
-compression_type=snappy -verify_compression=1
1866756 -> 1818989 (-2.6%)
-compression_type=zstd -verify_compression=1
982640 -> 997129 (+1.5%)
-compression_type=zstd -compression_max_dict_bytes=8180
912608 -> 937248 (+2.7%)

### SST read performance
Create DBs
```
for COMP in none snappy zstd; do echo $ARGS; ./db_bench -db=/dev/shm/dbbench-7-$COMP --benchmarks=fillseq,flush -num=10000000 -compaction_style=2 -fifo_compaction_max_table_files_size_mb=1000 -fifo_compaction_allow_compaction=0 -disable_wal -write_buffer_size=12000000 -compression_type=$COMP -format_version=7; done
```
And test
```
for COMP in none
snappy zstd none; do echo $COMP; (for I in `seq 1 8`; do ./db_bench -readonly -db=/dev/shm/dbbench
-7-$COMP --benchmarks=readrandom -num=10000000 -duration=20 -threads=8 2>&1 | grep micros/op; done
) | awk '{n++; sum += $5;} END { print int(sum / n); }'; done
```

Ops/sec, Before -> After (both fv=6)
none
1491732 -> 1500209 (+0.6%)
snappy
1157216 -> 1169202 (+1.0%)
zstd
695414 -> 703719 (+1.2%)
none (again)
1491787 -> 1528789 (+2.4%)

Ops/sec, Before fv=6 -> After fv=7:
none
1492278 -> 1508668 (+1.1%)
snappy
1140769 -> 1152613 (+1.0%)
zstd
696437 -> 696511 (+0.0%)
none (again)
1500585 -> 1512037 (+0.7%)

Overall, I think we can take the read CPU improvement in exchange for the hit (in some cases) on background write CPU

Reviewed By: hx235

Differential Revision: D76520739

Pulled By: pdillinger

fbshipit-source-id: e73bd72502ff85c8779cba313f26f7d1fd50be3a
2025-06-16 14:19:03 -07:00

721 lines
24 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 "table/block_based/data_block_hash_index.h"
#include <cstdlib>
#include <string>
#include <unordered_map>
#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/get_context.h"
#include "table/table_builder.h"
#include "test_util/testharness.h"
#include "test_util/testutil.h"
#include "util/random.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;
}
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 += rnd->RandomString(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(rnd.RandomString(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));
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));
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));
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));
// 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));
const InternalKeyComparator icmp(BytewiseComparator());
auto iter = reader.NewDataIterator(icmp.user_comparator(),
kDisableGlobalSequenceNumber);
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));
const InternalKeyComparator icmp(BytewiseComparator());
// random seek existent keys
for (int i = 0; i < num_records; i++) {
auto iter = reader.NewDataIterator(icmp.user_comparator(),
kDisableGlobalSequenceNumber);
// 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++) {
auto iter = reader.NewDataIterator(icmp.user_comparator(),
kDisableGlobalSequenceNumber);
// 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 ImmutableOptions ioptions(options);
const MutableCFOptions moptions(options);
const InternalKeyComparator internal_comparator(options.comparator);
EnvOptions soptions;
soptions.use_mmap_reads = ioptions.allow_mmap_reads;
test::StringSink* sink = new test::StringSink();
std::unique_ptr<FSWritableFile> f(sink);
file_writer.reset(
new WritableFileWriter(std::move(f), "" /* don't care */, FileOptions()));
std::unique_ptr<TableBuilder> builder;
InternalTblPropCollFactories internal_tbl_prop_coll_factories;
std::string column_family_name;
const ReadOptions read_options;
const WriteOptions write_options;
builder.reset(moptions.table_factory->NewTableBuilder(
TableBuilderOptions(
ioptions, moptions, read_options, write_options, internal_comparator,
&internal_tbl_prop_coll_factories, options.compression,
CompressionOptions(),
TablePropertiesCollectorFactory::Context::kUnknownColumnFamily,
column_family_name, level_, kUnknownNewestKeyTime),
file_writer.get()));
builder->Add(ik1.Encode().ToString(), v1);
builder->Add(ik2.Encode().ToString(), v2);
EXPECT_TRUE(builder->status().ok());
Status s = builder->Finish();
ASSERT_OK(file_writer->Flush(IOOptions()));
EXPECT_TRUE(s.ok()) << s.ToString();
EXPECT_EQ(sink->contents().size(), builder->FileSize());
// Open the table
test::StringSource* source = new test::StringSource(
sink->contents(), 0 /*uniq_id*/, ioptions.allow_mmap_reads);
std::unique_ptr<FSRandomAccessFile> file(source);
file_reader.reset(new RandomAccessFileReader(std::move(file), "test"));
const bool kSkipFilters = true;
const bool kImmortal = true;
ASSERT_OK(moptions.table_factory->NewTableReader(
TableReaderOptions(ioptions, moptions.prefix_extractor,
nullptr /* compression_manager */, soptions,
internal_comparator,
0 /* block_protection_bytes_per_key */, !kSkipFilters,
!kImmortal, level_),
std::move(file_reader), sink->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,
nullptr, 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,
nullptr, 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,
nullptr, 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,
nullptr, 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) {
ROCKSDB_NAMESPACE::port::InstallStackTraceHandler();
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}