This is an automated email from the ASF dual-hosted git repository.
PragmaTwice pushed a commit to branch unstable
in repository https://gitbox.apache.org/repos/asf/kvrocks.git
The following commit(s) were added to refs/heads/unstable by this push:
new 65518a761 feat(kvrocks2redis): support HFE in full export and
migration WAL (#3644)
65518a761 is described below
commit 65518a761dc384e8dd1956b87704df2229ae5ebf
Author: Twice <[email protected]>
AuthorDate: Sat Sep 26 15:44:09 2026 +0800
feat(kvrocks2redis): support HFE in full export and migration WAL (#3644)
Adds HFE-aware full export, skipping expired fields and restoring live
field expirations with `HPEXPIREAT`.
Also preserves per-key hash encoding in slot-migration snapshot WAL
using the existing `2` / `2 13 1` LogData format; generic rebatching
(#3645) and the existing RedisWriter startup race remain out of scope.
Refs #3436.
Assisted-by: Codex
---
CMakeLists.txt | 12 +-
src/cluster/slot_migrate.cc | 5 +
tests/cppunit/kvrocks2redis_parser_test.cc | 135 +++++++++++++++++++++
.../slotmigrate/hash_field_expiration_test.go | 70 +++++++++++
utils/kvrocks2redis/README.md | 8 +-
utils/kvrocks2redis/parser.cc | 44 +++++--
utils/kvrocks2redis/parser.h | 2 +-
7 files changed, 257 insertions(+), 19 deletions(-)
diff --git a/CMakeLists.txt b/CMakeLists.txt
index b074b52a8..bbcfe7dc5 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -312,13 +312,17 @@ target_link_libraries(kvrocks PRIVATE kvrocks_objs
${EXTERNAL_LIBS})
# kvrocks2redis sync tool
file(GLOB KVROCKS2REDIS_SRCS utils/kvrocks2redis/*.cc)
-add_executable(kvrocks2redis ${KVROCKS2REDIS_SRCS})
+list(FILTER KVROCKS2REDIS_SRCS EXCLUDE REGEX utils/kvrocks2redis/main.cc)
-target_link_libraries(kvrocks2redis PRIVATE kvrocks_objs ${EXTERNAL_LIBS})
+add_library(kvrocks2redis_objs OBJECT ${KVROCKS2REDIS_SRCS})
+target_link_libraries(kvrocks2redis_objs PUBLIC kvrocks_objs ${EXTERNAL_LIBS})
+
+add_executable(kvrocks2redis utils/kvrocks2redis/main.cc)
+target_link_libraries(kvrocks2redis PRIVATE kvrocks_objs kvrocks2redis_objs
${EXTERNAL_LIBS})
# kvrocks unit tests
file(GLOB_RECURSE TESTS_SRCS tests/cppunit/*.cc)
add_executable(unittest ${TESTS_SRCS})
-target_include_directories(unittest PRIVATE tests/cppunit)
+target_include_directories(unittest PRIVATE tests/cppunit utils)
-target_link_libraries(unittest PRIVATE kvrocks_objs gtest_main gmock
${EXTERNAL_LIBS})
+target_link_libraries(unittest PRIVATE kvrocks_objs kvrocks2redis_objs
gtest_main gmock ${EXTERNAL_LIBS})
diff --git a/src/cluster/slot_migrate.cc b/src/cluster/slot_migrate.cc
index 28907306b..66228058b 100644
--- a/src/cluster/slot_migrate.cc
+++ b/src/cluster/slot_migrate.cc
@@ -595,6 +595,11 @@ Status SlotMigrator::sendSnapshot() {
if (redis_type == RedisType::kRedisList) {
redis::WriteBatchLogData batch_log_data(redis_type,
{std::to_string(RedisCommand::kRedisCmdRPush)});
log_data = batch_log_data.Encode();
+ } else if (redis_type == RedisType::kRedisHash) {
+ HashMetadata metadata(false);
+ auto s = metadata.Decode(iter.Value());
+ if (!s.ok()) return {Status::NotOK, s.ToString()};
+ log_data = redis::WriteBatchLogData(metadata.mode).Encode();
} else {
redis::WriteBatchLogData batch_log_data(redis_type);
log_data = batch_log_data.Encode();
diff --git a/tests/cppunit/kvrocks2redis_parser_test.cc
b/tests/cppunit/kvrocks2redis_parser_test.cc
new file mode 100644
index 000000000..3183ae9e5
--- /dev/null
+++ b/tests/cppunit/kvrocks2redis_parser_test.cc
@@ -0,0 +1,135 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#include <algorithm>
+#include <tuple>
+
+#include "kvrocks2redis/parser.h"
+#include "server/redis_reply.h"
+#include "test_base.h"
+
+class RecordingWriter : public Writer {
+ public:
+ RecordingWriter() : Writer(nullptr) {}
+ Status Write(const std::string &ns, const std::vector<std::string> &aofs)
override {
+ if (fail) return {Status::NotOK, "test writer failure"};
+ auto &output = commands[ns];
+ output.insert(output.end(), aofs.begin(), aofs.end());
+ return Status::OK();
+ }
+ std::map<std::string, std::vector<std::string>> commands;
+ bool fail = false;
+};
+
+class Kvrocks2RedisParserTest : public TestFixture, public
::testing::TestWithParam<std::tuple<bool, bool>> {
+ protected:
+ void SetUp() override { config_.slot_id_encoded = std::get<0>(GetParam()); }
+
+ using Field = std::tuple<std::string, std::string, uint64_t>;
+ void putHash(const std::string &ns, const std::string &key,
HashSubkeyEncodingMode mode,
+ const std::vector<Field> &fields, uint64_t key_expire = 0) {
+ HashMetadata metadata(false, mode);
+ metadata.flags = kRedisHash | (std::get<1>(GetParam()) ?
METADATA_64BIT_ENCODING_MASK : 0);
+ metadata.version = 42;
+ metadata.expire = key_expire;
+ metadata.size = fields.size();
+ auto ns_key = ComposeNamespaceKey(ns, key, storage_->IsSlotIdEncoded());
+ rocksdb::WriteBatch batch;
+ for (const auto &[field, value, expire] : fields) {
+ if (expire == 0) {
+ ++metadata.persist;
+ } else {
+ metadata.lower = metadata.lower == 0 ? expire :
std::min(metadata.lower, expire);
+ metadata.upper = std::max(metadata.upper, expire);
+ }
+ auto subkey = InternalKey(ns_key, field, metadata.version,
storage_->IsSlotIdEncoded()).Encode();
+ ASSERT_TRUE(batch.Put(subkey, metadata.EncodeSubkeyValue(value,
expire)).ok());
+ }
+ std::string bytes;
+ metadata.Encode(&bytes);
+ ASSERT_TRUE(batch.Put(storage_->GetCFHandle(ColumnFamilyID::Metadata),
ns_key, bytes).ok());
+ ASSERT_TRUE(storage_->GetDB()->Write(storage_->DefaultWriteOptions(),
&batch).ok());
+ }
+};
+
+TEST_P(Kvrocks2RedisParserTest,
FullExportUsesPerKeyEncodingAndAbsoluteExpiration) {
+ const auto hfe = HashSubkeyEncodingMode::kFieldExpiration;
+ const auto legacy = HashSubkeyEncodingMode::kLegacy;
+ const uint64_t field_expire = 4000000000123ULL;
+ const uint64_t key_expire = 4000000060456ULL;
+ const std::string binary("value\0\xff", 7);
+ auto legacy_value = HashMetadata(false, hfe).EncodeSubkeyValue(binary,
field_expire);
+ putHash("ns", "a{hash}", legacy, {{"field", legacy_value, 0}});
+ putHash("ns", "b{hash}", hfe,
+ {{"a-binary", binary, 0}, {"b-empty", "", 0}, {"c-expired", "gone",
1}, {"d-live", "live", field_expire}},
+ key_expire);
+ putHash("ns", "c{hash}", hfe, {{"field", "expired-key", 0}}, 1);
+ putHash("ns", "d{hash}", hfe, {{"field", "expired-field", 1}});
+ putHash("other", "b{hash}", legacy, {{"field", legacy_value, 0}});
+
+ // Ignore subkeys from an obsolete version, even when their payload is not
valid HFE encoding.
+ auto stale = InternalKey(ComposeNamespaceKey("ns", "b{hash}",
storage_->IsSlotIdEncoded()), "old", 41,
+ storage_->IsSlotIdEncoded())
+ .Encode();
+ ASSERT_TRUE(storage_->GetDB()->Put(storage_->DefaultWriteOptions(), stale,
"stale").ok());
+
+ for (auto default_mode : {legacy, hfe}) {
+ config_.hash_encoding_mode = default_mode;
+ RecordingWriter writer;
+ Parser parser(storage_.get(), &writer);
+ ASSERT_TRUE(parser.ParseFullDB().IsOK());
+ auto stored_key_expire = std::get<1>(GetParam()) ? key_expire :
Metadata::ExpireMsToS(key_expire) * 1000;
+ std::vector<std::string> expected = {
+ redis::ArrayOfBulkStrings({"HSET", "a{hash}", "field", legacy_value}),
+ redis::ArrayOfBulkStrings({"HSET", "b{hash}", "a-binary", binary}),
+ redis::ArrayOfBulkStrings({"HSET", "b{hash}", "b-empty", ""}),
+ redis::ArrayOfBulkStrings({"HSET", "b{hash}", "d-live", "live"}),
+ redis::ArrayOfBulkStrings({"HPEXPIREAT", "b{hash}",
std::to_string(field_expire), "FIELDS", "1", "d-live"}),
+ redis::ArrayOfBulkStrings({"EXPIREAT", "b{hash}",
std::to_string(stored_key_expire / 1000)})};
+ ASSERT_EQ(writer.commands.size(), 2);
+ EXPECT_EQ(writer.commands.at("ns"), expected);
+ EXPECT_EQ(writer.commands.at("other"),
+ std::vector<std::string>{redis::ArrayOfBulkStrings({"HSET",
"b{hash}", "field", legacy_value})});
+ }
+}
+
+TEST_P(Kvrocks2RedisParserTest, InvalidHashDataAndWriterErrorsAreNotIgnored) {
+ putHash("ns", "hash", HashSubkeyEncodingMode::kFieldExpiration, {{"field",
"value", 0}});
+ RecordingWriter writer;
+ Parser parser(storage_.get(), &writer);
+ writer.fail = true;
+ EXPECT_FALSE(parser.ParseFullDB().IsOK());
+ writer.fail = false;
+
+ auto ns_key = ComposeNamespaceKey("ns", "hash", storage_->IsSlotIdEncoded());
+ auto key = InternalKey(ns_key, "field", 42,
storage_->IsSlotIdEncoded()).Encode();
+ ASSERT_TRUE(storage_->GetDB()->Put(storage_->DefaultWriteOptions(), key,
std::string(7, '\0')).ok());
+ EXPECT_FALSE(parser.ParseFullDB().IsOK());
+ EXPECT_TRUE(writer.commands.empty());
+
+ std::string metadata;
+ auto *cf = storage_->GetCFHandle(ColumnFamilyID::Metadata);
+ ASSERT_TRUE(storage_->GetDB()->Get(storage_->DefaultScanOptions(), cf,
ns_key, &metadata).ok());
+ metadata.resize(metadata.size() - 1);
+ ASSERT_TRUE(storage_->GetDB()->Put(storage_->DefaultWriteOptions(), cf,
ns_key, metadata).ok());
+ EXPECT_FALSE(parser.ParseFullDB().IsOK());
+ EXPECT_TRUE(writer.commands.empty());
+}
+
+INSTANTIATE_TEST_SUITE_P(Encodings, Kvrocks2RedisParserTest,
::testing::Combine(::testing::Bool(), ::testing::Bool()));
diff --git a/tests/gocase/integration/slotmigrate/hash_field_expiration_test.go
b/tests/gocase/integration/slotmigrate/hash_field_expiration_test.go
index 1cf489ae3..a2299b71d 100644
--- a/tests/gocase/integration/slotmigrate/hash_field_expiration_test.go
+++ b/tests/gocase/integration/slotmigrate/hash_field_expiration_test.go
@@ -23,6 +23,7 @@ import (
"context"
"errors"
"fmt"
+ "strconv"
"strings"
"testing"
"time"
@@ -205,3 +206,72 @@ func TestSlotMigrateHashFieldExpiration(t *testing.T) {
})
}
}
+
+func TestSlotMigrateSnapshotHashEncodingInWAL(t *testing.T) {
+ for _, mode := range []string{"legacy", "field-expiration"} {
+ ctx := context.Background()
+ source := util.StartServer(t, util.KvrocksServerConfigs{
+ "cluster-enabled": "yes",
+ "hash-encoding-mode": mode,
+ "migrate-batch-size-kb": "16",
+ })
+ defer source.Close()
+ src := source.NewClient()
+ defer func() { require.NoError(t, src.Close()) }()
+ destination := util.StartServer(t, util.KvrocksServerConfigs{
+ "cluster-enabled": "yes",
+ "hash-encoding-mode": "legacy",
+ })
+ defer destination.Close()
+ dst := destination.NewClient()
+ defer func() { require.NoError(t, dst.Close()) }()
+
+ sourceID, destinationID := strings.Repeat("a", 40),
strings.Repeat("b", 40)
+ require.NoError(t, src.Do(ctx, "CLUSTERX", "SETNODEID",
sourceID).Err())
+ require.NoError(t, dst.Do(ctx, "CLUSTERX", "SETNODEID",
destinationID).Err())
+ nodes := fmt.Sprintf("%s %s %d master - 0-10000\n%s %s %d
master - 10001-16383",
+ sourceID, source.Host(), source.Port(), destinationID,
destination.Host(), destination.Port())
+ require.NoError(t, src.Do(ctx, "CLUSTERX", "SETNODES", nodes,
"1").Err())
+ require.NoError(t, dst.Do(ctx, "CLUSTERX", "SETNODES", nodes,
"1").Err())
+
+ t.Run(mode, func(t *testing.T) {
+ // A single small hash keeps this test independent of
migration batch splitting and merging.
+ key := fmt.Sprintf("hash{%s}", util.SlotTable[0])
+ binaryValue := strings.Repeat("\x00", 8) +
"value\x00\xff"
+ values := map[string]string{"live": "v", "persistent":
binaryValue}
+ require.NoError(t, src.HSet(ctx, key, values).Err())
+ wantCommands := []any{[]any{"HSET", key, "live", "v"}}
+ var expireAt int64 = -1
+ if mode == "field-expiration" {
+ expireAt = time.Now().Add(10 *
time.Minute).UnixMilli()
+ result, err := src.Do(ctx, "HPEXPIREAT", key,
expireAt, "FIELDS", 1, "live").Int64Slice()
+ require.NoError(t, err)
+ require.Equal(t, []int64{1}, result)
+ wantCommands = append(wantCommands,
[]any{"HPEXPIREAT", key, strconv.FormatInt(expireAt, 10), "FIELDS", "1",
"live"})
+ require.NoError(t, src.ConfigSet(ctx,
"hash-encoding-mode", "legacy").Err())
+ } else {
+ require.NoError(t, src.ConfigSet(ctx,
"hash-encoding-mode", "field-expiration").Err())
+ }
+ wantCommands = append(wantCommands, []any{"HSET", key,
"persistent", binaryValue})
+
+ require.NoError(t, src.Do(ctx, "CLUSTERX", "MIGRATE",
0, destinationID).Err())
+ waitForMigrateState(t, src, 0,
SlotMigrationStateSuccess)
+ waitForImportState(t, dst, 0, SlotImportStateSuccess)
+ got, err := dst.HGetAll(ctx, key).Result()
+ require.NoError(t, err)
+ require.Equal(t, values, got)
+ if mode == "field-expiration" {
+ expires, err := dst.Do(ctx, "HPEXPIRETIME",
key, "FIELDS", 2, "live", "persistent").Int64Slice()
+ require.NoError(t, err)
+ require.Equal(t, []int64{expireAt, -1}, expires)
+ }
+
+ // Raw KV correctness alone does not ensure that
downstream WAL consumers decode the fields correctly.
+ result, err := dst.Do(ctx, "POLLUPDATES", 0, "MAX",
1000, "FORMAT", "RESP").Result()
+ require.NoError(t, err)
+ batch := result.(map[any]any)
+ require.Equal(t, batch["latest_sequence"],
batch["next_sequence"])
+ require.Equal(t, []any{"default", wantCommands},
batch["updates"])
+ })
+ }
+}
diff --git a/utils/kvrocks2redis/README.md b/utils/kvrocks2redis/README.md
index a7047265f..4507fa3d4 100644
--- a/utils/kvrocks2redis/README.md
+++ b/utils/kvrocks2redis/README.md
@@ -10,4 +10,10 @@ Another thread (named `redis-writer`) reads the AOF
continuously and sends the c
When the program runs, the following files are generated:
1. xxx_appendonly.aof: parsed data will be saved in this file.
2. xxx_last_next_offset.txt: indicates the position of the AOF file read by
the `redis-writer` thread.
-3. last_next_seq.txt: indicates the sequence number parsed by `kvrocks2redis`
to record the synchronization location and check whether incremental
synchronization can be performed.
\ No newline at end of file
+3. last_next_seq.txt: indicates the sequence number parsed by `kvrocks2redis`
to record the synchronization location and check whether incremental
synchronization can be performed.
+
+## Hash field expiration
+
+Full synchronization decodes each hash using its stored encoding mode, skips
expired fields, and preserves live field expirations using `HSET` followed by
`HPEXPIREAT` with the original absolute timestamp.
+Incremental synchronization uses the same commands to replay HFE writes from
the WAL.
+The target Redis server must support `HPEXPIREAT` when synchronizing hashes
with field expirations.
diff --git a/utils/kvrocks2redis/parser.cc b/utils/kvrocks2redis/parser.cc
index a80f349ab..25ba98174 100644
--- a/utils/kvrocks2redis/parser.cc
+++ b/utils/kvrocks2redis/parser.cc
@@ -29,6 +29,7 @@
#include "logging.h"
#include "server/redis_reply.h"
#include "storage/redis_metadata.h"
+#include "time_util.h"
#include "types/redis_string.h"
Status Parser::ParseFullDB() {
@@ -55,7 +56,7 @@ Status Parser::ParseFullDB() {
if (metadata.Type() == kRedisString) {
s = parseSimpleKV(iter->key(), iter->value(), metadata.expire);
} else {
- s = parseComplexKV(iter->key(), metadata);
+ s = parseComplexKV(iter->key(), metadata, iter->value());
}
if (!s.IsOK()) return s;
}
@@ -79,12 +80,18 @@ Status Parser::parseSimpleKV(const Slice &ns_key, const
Slice &value, uint64_t e
return s;
}
-Status Parser::parseComplexKV(const Slice &ns_key, const Metadata &metadata) {
+Status Parser::parseComplexKV(const Slice &ns_key, const Metadata &metadata,
const Slice &metadata_bytes) {
RedisType type = metadata.Type();
if (type < kRedisHash || type > kRedisSortedint) {
return {Status::NotOK, "unknown metadata type: " + std::to_string(type)};
}
+ HashMetadata hash_metadata(false);
+ if (type == kRedisHash) {
+ auto ds = hash_metadata.Decode(metadata_bytes);
+ if (!ds.ok()) return {Status::NotOK, ds.ToString()};
+ }
+
auto [ns, user_key] = ExtractNamespaceKey<std::string>(ns_key,
slot_id_encoded_);
std::string prefix_key = InternalKey(ns_key, "", metadata.version,
slot_id_encoded_).Encode();
std::string next_version_prefix_key = InternalKey(ns_key, "",
metadata.version + 1, slot_id_encoded_).Encode();
@@ -93,7 +100,6 @@ Status Parser::parseComplexKV(const Slice &ns_key, const
Metadata &metadata) {
rocksdb::Slice upper_bound(next_version_prefix_key);
read_options.iterate_upper_bound = &upper_bound;
- std::string output;
auto no_txn_ctx = engine::Context::NoTransactionContext(storage_);
auto iter = util::UniqueIterator(no_txn_ctx, read_options);
for (iter->Seek(prefix_key); iter->Valid(); iter->Next()) {
@@ -104,19 +110,31 @@ Status Parser::parseComplexKV(const Slice &ns_key, const
Metadata &metadata) {
InternalKey ikey(iter->key(), slot_id_encoded_);
std::string sub_key = ikey.GetSubKey().ToString();
std::string value = iter->value().ToString();
+ std::vector<std::string> commands;
switch (type) {
- case kRedisHash:
- output = redis::ArrayOfBulkStrings({"HSET", user_key, sub_key, value});
+ case kRedisHash: {
+ Slice field_value(value);
+ uint64_t expire = 0;
+ auto ds = hash_metadata.DecodeSubkeyValue(&field_value, &expire);
+ if (!ds.ok()) return {Status::NotOK, ds.ToString()};
+ // Full sync starts with an empty target; expired fields need not be
recreated.
+ if (expire > 0 && expire <= util::GetTimeStampMS()) continue;
+ commands.emplace_back(redis::ArrayOfBulkStrings({"HSET", user_key,
sub_key, field_value.ToString()}));
+ if (expire > 0) {
+ commands.emplace_back(
+ redis::ArrayOfBulkStrings({"HPEXPIREAT", user_key,
std::to_string(expire), "FIELDS", "1", sub_key}));
+ }
break;
+ }
case kRedisSet:
- output = redis::ArrayOfBulkStrings({"SADD", user_key, sub_key});
+ commands.emplace_back(redis::ArrayOfBulkStrings({"SADD", user_key,
sub_key}));
break;
case kRedisList:
- output = redis::ArrayOfBulkStrings({"RPUSH", user_key, value});
+ commands.emplace_back(redis::ArrayOfBulkStrings({"RPUSH", user_key,
value}));
break;
case kRedisZSet: {
double score = DecodeDouble(value.data());
- output = redis::ArrayOfBulkStrings({"ZADD", user_key,
util::Float2String(score), sub_key});
+ commands.emplace_back(redis::ArrayOfBulkStrings({"ZADD", user_key,
util::Float2String(score), sub_key}));
break;
}
case kRedisBitmap: {
@@ -127,21 +145,21 @@ Status Parser::parseComplexKV(const Slice &ns_key, const
Metadata &metadata) {
}
case kRedisSortedint: {
std::string val =
std::to_string(DecodeFixed64(ikey.GetSubKey().data()));
- output = redis::ArrayOfBulkStrings({"ZADD", user_key, val, val});
+ commands.emplace_back(redis::ArrayOfBulkStrings({"ZADD", user_key,
val, val}));
break;
}
default:
break; // should never get here
}
- if (type != kRedisBitmap) {
- auto s = writer_->Write(ns, {output});
- if (!s.IsOK()) return s.Prefixed(fmt::format("failed to write the '{}'
command to AOF", output));
+ if (!commands.empty()) {
+ auto s = writer_->Write(ns, commands);
+ if (!s.IsOK()) return s.Prefixed("failed to write commands to AOF");
}
}
if (metadata.expire > 0) {
- output = redis::ArrayOfBulkStrings({"EXPIREAT", user_key,
std::to_string(metadata.expire / 1000)});
+ auto output = redis::ArrayOfBulkStrings({"EXPIREAT", user_key,
std::to_string(metadata.expire / 1000)});
Status s = writer_->Write(ns, {output});
if (!s.IsOK()) return s.Prefixed("failed to write the EXPIREAT command to
AOF");
}
diff --git a/utils/kvrocks2redis/parser.h b/utils/kvrocks2redis/parser.h
index 09ba2f17c..690bed37d 100644
--- a/utils/kvrocks2redis/parser.h
+++ b/utils/kvrocks2redis/parser.h
@@ -48,6 +48,6 @@ class Parser {
bool slot_id_encoded_ = false;
Status parseSimpleKV(const Slice &ns_key, const Slice &value, uint64_t
expire);
- Status parseComplexKV(const Slice &ns_key, const Metadata &metadata);
+ Status parseComplexKV(const Slice &ns_key, const Metadata &metadata, const
Slice &metadata_bytes);
Status parseBitmapSegment(const Slice &ns, const Slice &user_key, int index,
const Slice &bitmap);
};
