diff --git a/api/kv_operation.py b/api/kv_operation.py index f18c846018..11da991a1d 100644 --- a/api/kv_operation.py +++ b/api/kv_operation.py @@ -24,6 +24,8 @@ parent_dir = os.path.dirname(current_dir) new_path_dir = os.path.join(parent_dir, "bazel-out", "k8-fastbuild", "bin", "api") sys.path.insert(0, new_path_dir) +# bazel-bin points at the build output for whatever platform was built. +sys.path.insert(0, os.path.join(parent_dir, "bazel-bin", "api")) import pybind_kv @@ -44,3 +46,71 @@ def get_value(key: str or int or float, config_path: str = current_dir + "/ip_ad :return: A string of the key's corresponding value. """ return pybind_kv.get(str(key), os.path.abspath(config_path)) + + +def create_index_entry(index_name: str, attributes, primary_key: str, + config_path: str = current_dir + "/ip_address.config") -> int: + """ + Add a secondary index entry, e.g. create_index_entry("by_city", "Davis", "user:1"). + + :param index_name: The index the entry belongs to, chosen by you. + :param attributes: One attribute value, or a list of them for a multi-attribute index. + Mixing the two forms in one call is not supported. + :param primary_key: The key of the record being indexed. It must already exist. + :param config_path: Path to the client config. + :return: 0 on success, -3 if the servers rejected it, -1 or -2 if it could not be sent. + """ + return pybind_kv.create_index_entry(index_name, attributes, str(primary_key), + os.path.abspath(config_path)) + + +def delete_index_entry(index_name: str, attributes, primary_key: str, + config_path: str = current_dir + "/ip_address.config") -> int: + """ + Remove a secondary index entry. The record itself is not deleted. + + :return: 0 on success, -3 if the servers rejected it (no such entry, empty index + name or primary key, or a zero byte in any part), -1 or -2 if it could not be sent. + """ + return pybind_kv.delete_index_entry(index_name, attributes, str(primary_key), + os.path.abspath(config_path)) + + +def update_index_entry(index_name: str, old_attributes, new_attributes, primary_key: str, + config_path: str = current_dir + "/ip_address.config") -> int: + """ + Move an entry to new attributes, e.g. when a record's city changes. The old entry is + removed and the new one added in a single write. + + :return: 0 on success, -3 if the servers rejected it (the old entry does not exist, + or the input was invalid), -1 or -2 if it could not be sent. + """ + return pybind_kv.update_index_entry(index_name, old_attributes, new_attributes, + str(primary_key), os.path.abspath(config_path)) + + +def query_by_index(index_name: str, attributes, + config_path: str = current_dir + "/ip_address.config") -> list: + """ + Primary keys whose entry starts with the given attributes, each once and sorted. + + Attributes match from the left: in an index built as (owner, format), you can query + by owner, or by owner and format, but not by format alone. An empty list returns + every entry in the index. + + :return: A list of primary keys, empty if nothing matched or the request failed. + """ + return pybind_kv.query_by_index(index_name, attributes, os.path.abspath(config_path)) + + +def query_records_by_index(index_name: str, attributes, + config_path: str = current_dir + "/ip_address.config") -> list: + """ + Like query_by_index, but returns (key, value) pairs. The records are read on the + server and come back in the same reply, instead of one request per match. + + :return: A list of (primary_key, value) tuples, empty if nothing matched or the + request failed. + """ + return pybind_kv.query_records_by_index(index_name, attributes, + os.path.abspath(config_path)) diff --git a/api/pybind_kv_service.cpp b/api/pybind_kv_service.cpp index 013683fbd7..cf305c5376 100644 --- a/api/pybind_kv_service.cpp +++ b/api/pybind_kv_service.cpp @@ -20,11 +20,14 @@ #include #include #include +#include #include #include #include #include +#include +#include #include "common/proto/signature_info.pb.h" #include "interface/kv/kv_client.h" @@ -60,7 +63,126 @@ bool set(std::string key, std::string value, std::string config_path) { } } +namespace { + +KVClient MakeClient(const std::string& config_path) { + ResDBConfig config = GenerateResDBConfig(config_path); + config.SetClientTimeoutMs(100000); + return KVClient(config); +} + +} // namespace + +// Index commands. Return 0 on success, -3 if the servers rejected the +// request, or a send error (-1, -2). Attributes may be a list or, for a +// single-attribute index, a plain string. +int create_index_entry(std::string index_name, + std::vector attributes, + std::string primary_key, std::string config_path) { + return MakeClient(config_path) + .CreateIndexEntry(index_name, attributes, primary_key); +} + +int create_index_entry_one(std::string index_name, std::string attribute, + std::string primary_key, std::string config_path) { + return create_index_entry(index_name, {attribute}, primary_key, config_path); +} + +int delete_index_entry(std::string index_name, + std::vector attributes, + std::string primary_key, std::string config_path) { + return MakeClient(config_path) + .DeleteIndexEntry(index_name, attributes, primary_key); +} + +int delete_index_entry_one(std::string index_name, std::string attribute, + std::string primary_key, std::string config_path) { + return delete_index_entry(index_name, {attribute}, primary_key, config_path); +} + +int update_index_entry(std::string index_name, + std::vector old_attributes, + std::vector new_attributes, + std::string primary_key, std::string config_path) { + return MakeClient(config_path) + .UpdateIndexEntry(index_name, old_attributes, new_attributes, + primary_key); +} + +int update_index_entry_one(std::string index_name, std::string old_attribute, + std::string new_attribute, std::string primary_key, + std::string config_path) { + return update_index_entry(index_name, {old_attribute}, {new_attribute}, + primary_key, config_path); +} + +// Primary keys whose entry starts with `attributes`, each once and sorted. +// An empty list means the whole index. +std::vector query_by_index(std::string index_name, + std::vector attributes, + std::string config_path) { + std::vector keys; + auto items = MakeClient(config_path).QueryByIndex(index_name, attributes); + if (items == nullptr) { + return keys; + } + for (const auto& item : items->item()) { + keys.push_back(item.key()); + } + return keys; +} + +std::vector query_by_index_one(std::string index_name, + std::string attribute, + std::string config_path) { + return query_by_index(index_name, {attribute}, config_path); +} + +// Same query, but each record's value comes back in the same reply instead of +// one request per match. +std::vector> query_records_by_index( + std::string index_name, std::vector attributes, + std::string config_path) { + std::vector> records; + auto items = + MakeClient(config_path).QueryByIndex(index_name, attributes, true); + if (items == nullptr) { + return records; + } + for (const auto& item : items->item()) { + records.emplace_back(item.key(), item.value_info().value()); + } + return records; +} + +std::vector> query_records_by_index_one( + std::string index_name, std::string attribute, std::string config_path) { + return query_records_by_index(index_name, {attribute}, config_path); +} + PYBIND11_MODULE(pybind_kv, m) { m.def("get", &get, "A function that gets a value from the key-value store"); m.def("set", &set, "A function that sets a value in the key-value store"); + // Each command has a single-string and a list overload: pybind never turns a + // str into a list of strings, so "Davis" needs its own signature. + m.def("create_index_entry", &create_index_entry_one, + "Add an index entry with one attribute"); + m.def("create_index_entry", &create_index_entry, + "Add an index entry with a list of attributes"); + m.def("delete_index_entry", &delete_index_entry_one, + "Remove an index entry with one attribute"); + m.def("delete_index_entry", &delete_index_entry, + "Remove an index entry with a list of attributes"); + m.def("update_index_entry", &update_index_entry_one, + "Move an index entry to a new attribute"); + m.def("update_index_entry", &update_index_entry, + "Move an index entry to new attributes"); + m.def("query_by_index", &query_by_index_one, + "Primary keys matching one attribute"); + m.def("query_by_index", &query_by_index, + "Primary keys matching a list of attributes"); + m.def("query_records_by_index", &query_records_by_index_one, + "(key, value) pairs matching one attribute"); + m.def("query_records_by_index", &query_records_by_index, + "(key, value) pairs matching a list of attributes"); } diff --git a/benchmark/e2e/BUILD b/benchmark/e2e/BUILD new file mode 100644 index 0000000000..cc29b3ac45 --- /dev/null +++ b/benchmark/e2e/BUILD @@ -0,0 +1,29 @@ +# +# 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. +# + +package(default_visibility = ["//visibility:private"]) + +cc_binary( + name = "composite_key_e2e_benchmark", + srcs = ["composite_key_e2e_benchmark.cpp"], + deps = [ + "//interface/kv:kv_client", + "//platform/config:resdb_config_utils", + ], +) diff --git a/benchmark/e2e/composite_key_e2e_benchmark.cpp b/benchmark/e2e/composite_key_e2e_benchmark.cpp new file mode 100644 index 0000000000..70af721097 --- /dev/null +++ b/benchmark/e2e/composite_key_e2e_benchmark.cpp @@ -0,0 +1,328 @@ +/* + * 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. + */ + +// End-to-end benchmark for secondary indexes (issue #180), measured against a +// running cluster rather than the storage layer. Four ways to answer the same +// question, "which records have city = Davis?": +// +// 1. scan every record is fetched and filtered by the application +// 2. index one query, primary keys only +// 3. index+get one query, then one get per match +// 4. index+values one query that returns the records (--with_values) +// +// Usage (from the repository root, with a cluster already running): +// bazel run //benchmark/e2e:composite_key_e2e_benchmark -- \ +// --config $PWD/service/tools/config/interface/service.config \ +// --records 1000 --selectivity 10 --repeat 3 +// +// --config takes any client config, so the same binary measures a local +// cluster or a multi-machine deployment. + +#include + +#include +#include +#include +#include +#include +#include + +#include "interface/kv/kv_client.h" +#include "platform/config/resdb_config_utils.h" + +namespace { + +using resdb::GenerateResDBConfig; +using resdb::Items; +using resdb::KVClient; +using resdb::ResDBConfig; + +// Each run uses its own index name, so entries from earlier runs on the +// same cluster are not counted. +std::string IndexName(const std::string& run_id) { + return "bench_by_city_" + run_id; +} +constexpr char kTargetCity[] = "Davis"; +constexpr char kOtherCity[] = "Other"; + +double MillisSince(std::chrono::steady_clock::time_point start) { + auto elapsed = std::chrono::steady_clock::now() - start; + return std::chrono::duration(elapsed).count(); +} + +std::string RecordKey(const std::string& run_id, int i) { + return "bench:" + run_id + ":" + std::to_string(i); +} + +// Every record carries its city, so the scan has something to filter on. +// `padding` makes the record the size a real document would be: the scan has +// to transfer every record, the index query only the matching ones. +std::string RecordValue(const std::string& city, int i, int value_size) { + std::string value = + "{\"city\":\"" + city + "\",\"n\":" + std::to_string(i) + ",\"data\":\""; + if (static_cast(value.size()) + 2 < value_size) { + value.append(value_size - value.size() - 2, 'x'); + } + value.append("\"}"); + return value; +} + +// The proxy batches client requests (100 requests or 100ms, whichever comes +// first), so a single client sending one request at a time pays the full wait +// every time. Loading uses several clients at once so the batches fill. +int Load(const ResDBConfig& config, KVClient* client, const std::string& run_id, + int records, int selectivity, int threads, int value_size) { + int matching = 0; + for (int i = 0; i < records; ++i) { + bool is_match = (i % 100) < selectivity; + const std::string city = is_match ? kTargetCity : kOtherCity; + if (is_match) { + ++matching; + } + client->Set(RecordKey(run_id, i), RecordValue(city, i, value_size)); + } + // Set does not wait for consensus, so confirm the last write landed before + // indexing: CreateIndexEntry refuses records that don't exist yet. + for (int attempt = 0; attempt < 100; ++attempt) { + auto value = client->Get(RecordKey(run_id, records - 1)); + if (value != nullptr && !value->empty()) { + break; + } + } + std::atomic failed(0); + std::vector workers; + for (int t = 0; t < threads; ++t) { + workers.emplace_back([&, t]() { + ResDBConfig thread_config = config; + KVClient thread_client(thread_config); + for (int i = t; i < records; i += threads) { + bool is_match = (i % 100) < selectivity; + const std::string city = is_match ? kTargetCity : kOtherCity; + if (thread_client.CreateIndexEntry(IndexName(run_id), {city}, + RecordKey(run_id, i)) != 0) { + ++failed; + } + } + }); + } + for (std::thread& worker : workers) { + worker.join(); + } + if (failed > 0) { + printf(" WARNING: %d index entries could not be created\n", failed.load()); + } + return matching; +} + +// 1. What an application has to do today: fetch everything, filter locally. +double ScanAndFilter(KVClient* client, const std::string& run_id, int* found) { + auto start = std::chrono::steady_clock::now(); + std::unique_ptr all = + client->GetKeyRange("bench:" + run_id + ":", "bench:" + run_id + ";"); + int matches = 0; + if (all != nullptr) { + const std::string needle = std::string("\"city\":\"") + kTargetCity + "\""; + for (const auto& item : all->item()) { + if (item.value_info().value().find(needle) != std::string::npos) { + ++matches; + } + } + } + double ms = MillisSince(start); + *found = matches; + return ms; +} + +// 2. One query, primary keys only. +double QueryKeysOnly(KVClient* client, const std::string& run_id, + int* found) { + auto start = std::chrono::steady_clock::now(); + std::unique_ptr keys = + client->QueryByIndex(IndexName(run_id), {kTargetCity}); + double ms = MillisSince(start); + *found = keys == nullptr ? -1 : keys->item_size(); + return ms; +} + +// 3. One query, then one get per match. Each get is its own consensus round. +double QueryThenGet(KVClient* client, const std::string& run_id, + int* found) { + auto start = std::chrono::steady_clock::now(); + std::unique_ptr keys = + client->QueryByIndex(IndexName(run_id), {kTargetCity}); + int fetched = 0; + if (keys != nullptr) { + for (const auto& item : keys->item()) { + auto value = client->Get(item.key()); + if (value != nullptr && !value->empty()) { + ++fetched; + } + } + } + double ms = MillisSince(start); + *found = fetched; + return ms; +} + +// 4. One query that returns the records themselves. +double QueryWithValues(KVClient* client, const std::string& run_id, + int* found) { + auto start = std::chrono::steady_clock::now(); + std::unique_ptr records = + client->QueryByIndex(IndexName(run_id), {kTargetCity}, true); + int with_value = 0; + if (records != nullptr) { + for (const auto& item : records->item()) { + if (!item.value_info().value().empty()) { + ++with_value; + } + } + } + double ms = MillisSince(start); + *found = with_value; + return ms; +} + +void ShowUsage() { + printf( + "--config path to the client config (required)\n" + "--records how many records to load (default 1000)\n" + "--selectivity percent of records that match, 1-100 (default 10)\n" + "--repeat how many times to time each approach (default 3)\n" + "--threads clients used while loading (default 16)\n" + "--value_size bytes per record, like a real document (default 64)\n"); +} + +} // namespace + +int main(int argc, char** argv) { + std::string config_file; + int records = 1000; + int selectivity = 10; + int repeat = 3; + int threads = 16; + int value_size = 64; + + static struct option long_options[] = { + {"config", required_argument, NULL, 'c'}, + {"records", required_argument, NULL, 'r'}, + {"selectivity", required_argument, NULL, 's'}, + {"repeat", required_argument, NULL, 'n'}, + {"threads", required_argument, NULL, 't'}, + {"value_size", required_argument, NULL, 'v'}, + {NULL, 0, NULL, 0}, + }; + int option_index = 0; + int c; + while ((c = getopt_long(argc, argv, "", long_options, &option_index)) != -1) { + switch (c) { + case 'c': + config_file = optarg; + break; + case 'r': + records = atoi(optarg); + break; + case 's': + selectivity = atoi(optarg); + break; + case 'n': + repeat = atoi(optarg); + break; + case 't': + threads = atoi(optarg); + break; + case 'v': + value_size = atoi(optarg); + break; + default: + ShowUsage(); + return 1; + } + } + if (config_file.empty() || records <= 0 || selectivity <= 0 || + selectivity > 100 || repeat <= 0) { + ShowUsage(); + return 1; + } + + ResDBConfig config = GenerateResDBConfig(config_file); + config.SetClientTimeoutMs(100000); + KVClient client(config); + + // A fresh key range per run, so repeated runs don't read each other's data. + const std::string run_id = + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count() % + 100000); + + printf("\nrecords=%d selectivity=%d%% value_size=%dB repeat=%d run_id=%s\n", + records, selectivity, value_size, repeat, run_id.c_str()); + printf("loading...\n"); + auto load_start = std::chrono::steady_clock::now(); + int expected = + Load(config, &client, run_id, records, selectivity, threads, value_size); + printf("loaded %d records (%d matching) in %.0f ms\n\n", records, expected, + MillisSince(load_start)); + + struct Result { + const char* name; + double best = 1e18; + int found = 0; + }; + Result results[4] = {{"scan + filter in the app"}, + {"index query (keys only)"}, + {"index query + one get per match"}, + {"index query with values"}}; + + for (int i = 0; i < repeat; ++i) { + int found = 0; + double ms = ScanAndFilter(&client, run_id, &found); + results[0].best = std::min(results[0].best, ms); + results[0].found = found; + + ms = QueryKeysOnly(&client, run_id, &found); + results[1].best = std::min(results[1].best, ms); + results[1].found = found; + + ms = QueryThenGet(&client, run_id, &found); + results[2].best = std::min(results[2].best, ms); + results[2].found = found; + + ms = QueryWithValues(&client, run_id, &found); + results[3].best = std::min(results[3].best, ms); + results[3].found = found; + } + + printf("%-34s %10s %10s %8s\n", "approach", "time (ms)", "speedup", "found"); + printf("%-34s %10s %10s %8s\n", "----------------------------------", + "----------", "----------", "--------"); + for (const Result& r : results) { + printf("%-34s %10.1f %9.1fx %8d\n", r.name, r.best, + results[0].best / r.best, r.found); + } + printf("\nfastest of %d runs; %d records, %d expected matches\n\n", repeat, + records, expected); + + bool ok = results[0].found == expected && results[1].found == expected && + results[2].found == expected && results[3].found == expected; + if (!ok) { + printf("WARNING: the four approaches did not agree on the match count\n"); + return 1; + } + return 0; +} diff --git a/chain/storage/composite_key_codec.cpp b/chain/storage/composite_key_codec.cpp index 781fa99543..3df108941c 100644 --- a/chain/storage/composite_key_codec.cpp +++ b/chain/storage/composite_key_codec.cpp @@ -75,8 +75,11 @@ std::string EncodeCompositeKeyPrefix( result.append(attribute_prefix[i]); } - // Trailing delimiter makes this a strict byte prefix of the full key. - result.push_back(kCompositeKeyDelim); + // End on a delimiter so "Davis" can't match "Davisville". With no + // attributes, the delimiter after index_name already does this. + if (!attribute_prefix.empty()) { + result.push_back(kCompositeKeyDelim); + } return result; } @@ -90,7 +93,8 @@ bool DecodeCompositeKey(const std::string& encoded, size_t ns_len = std::strlen(kCompositeKeyNamespace); if (encoded.size() < ns_len + 1 || - encoded.compare(0, ns_len, kCompositeKeyNamespace) != 0) { + encoded.compare(0, ns_len, kCompositeKeyNamespace) != 0 || + encoded[ns_len] != kCompositeKeyDelim) { return false; } diff --git a/chain/storage/composite_key_codec_test.cpp b/chain/storage/composite_key_codec_test.cpp index 30509f5f98..306dc21446 100644 --- a/chain/storage/composite_key_codec_test.cpp +++ b/chain/storage/composite_key_codec_test.cpp @@ -82,3 +82,19 @@ TEST(CompositeKeyCodecTest, EmptyAttributes) { EXPECT_EQ(attrs.size(), 0u); EXPECT_EQ(pk, "user:1"); } + +TEST(CompositeKeyCodecTest, PrefixWithNoAttributesMatchesWholeIndex) { + std::string prefix = EncodeCompositeKeyPrefix("byCity", {}); + EXPECT_EQ(EncodeCompositeKey("byCity", {"Davis"}, "user:1").find(prefix), 0u); + EXPECT_EQ(EncodeCompositeKey("byCity", {}, "user:2").find(prefix), 0u); + // Must not reach into another index whose name starts the same way. + EXPECT_NE(EncodeCompositeKey("byCity2", {"Davis"}, "user:3").find(prefix), + 0u); +} + +TEST(CompositeKeyCodecTest, DecodeRejectsMissingDelimiterAfterNamespace) { + std::string idx, pk; + std::vector attrs; + std::string bad = std::string("ckX") + "idx" + kCompositeKeyDelim + "pk"; + EXPECT_FALSE(DecodeCompositeKey(bad, &idx, &attrs, &pk)); +} diff --git a/chain/storage/duckdb.h b/chain/storage/duckdb.h index 153c19c61b..8573669c29 100644 --- a/chain/storage/duckdb.h +++ b/chain/storage/duckdb.h @@ -101,12 +101,11 @@ class DuckDB : public Storage { bool Flush(bool should_sync = false) override { return true; } - // No-op overrides so DuckDB stays concrete: Storage declares these - // composite-key methods pure virtual, and without them DuckDB would be - // abstract and fail to compile. The SQL backend doesn't use them. - int CreateCompositeKey(const std::string&) override { return 0; } + // Composite keys aren't supported on DuckDB. Return -1 (failure) so callers + // don't think an index entry was written. + int CreateCompositeKey(const std::string&) override { return -1; } - int DeleteCompositeKey(const std::string&) override { return 0; } + int DeleteCompositeKey(const std::string&) override { return -1; } std::vector GetByCompositeKeyPrefix( const std::string&) override { @@ -114,7 +113,7 @@ class DuckDB : public Storage { } int UpdateCompositeKey(const std::string&, const std::string&) override { - return 0; + return -1; } private: diff --git a/chain/storage/kv_storage_test.cpp b/chain/storage/kv_storage_test.cpp index b6719072c1..f3ebbac872 100644 --- a/chain/storage/kv_storage_test.cpp +++ b/chain/storage/kv_storage_test.cpp @@ -418,6 +418,20 @@ TEST_P(KVStorageTest, GetAllItemsExcludesCompositeKeys) { EXPECT_EQ(davis.size(), 1u); } +// GetRange must not return index entries either. "ck\0..." sorts between +// "apple" and "dog", so an unfiltered scan would include it. +TEST_P(KVStorageTest, GetRangeExcludesCompositeKeys) { + EXPECT_EQ(storage->SetValue("apple", "1"), 0); + EXPECT_EQ(storage->SetValue("dog", "2"), 0); + EXPECT_EQ(storage->CreateCompositeKey( + EncodeCompositeKey("byCity", {"Davis"}, "apple")), + 0); + + // MemoryDB's GetRange is unordered, so accept either order. + std::string range = storage->GetRange("a", "z"); + EXPECT_TRUE(range == "[1,2]" || range == "[2,1]") << range; +} + INSTANTIATE_TEST_CASE_P(KVStorageTest, KVStorageTest, ::testing::Values(MEM, LEVELDB, LEVELDB_WITH_BLOCK_CACHE)); diff --git a/chain/storage/leveldb.cpp b/chain/storage/leveldb.cpp index 15b1a85115..71c253dfa9 100644 --- a/chain/storage/leveldb.cpp +++ b/chain/storage/leveldb.cpp @@ -203,10 +203,18 @@ std::string ResLevelDB::GetValue(const std::string& key) { std::string ResLevelDB::GetRange(const std::string& min_key, const std::string& max_key) { std::string values = "["; + // Hide composite-key index entries from user-facing scans. + const std::string ck_prefix = + std::string(kCompositeKeyNamespace) + kCompositeKeyDelim; + leveldb::Iterator* it = db_->NewIterator(leveldb::ReadOptions()); bool first_iteration = true; for (it->Seek(min_key); it->Valid() && it->key().ToString() <= max_key; it->Next()) { + if (it->key().size() >= ck_prefix.size() && + memcmp(it->key().data(), ck_prefix.data(), ck_prefix.size()) == 0) { + continue; + } if (!first_iteration) values.append(","); first_iteration = false; values.append(it->value().ToString()); diff --git a/executor/kv/BUILD b/executor/kv/BUILD index 89bb67c7fd..5790e016fd 100644 --- a/executor/kv/BUILD +++ b/executor/kv/BUILD @@ -26,6 +26,7 @@ cc_library( hdrs = ["kv_executor.h"], deps = [ "//chain/storage", + "//chain/storage:composite_key_codec", "//common:comm", "//executor/common:transaction_manager", "//executor/contract/executor:contract_executor", @@ -34,6 +35,17 @@ cc_library( ], ) +cc_test( + name = "kv_executor_index_test", + srcs = ["kv_executor_index_test.cpp"], + deps = [ + ":kv_executor", + "//chain/storage:leveldb", + "//chain/storage:memory_db", + "//common/test:test_main", + ], +) + cc_test( name = "kv_executor_test", srcs = ["kv_executor_test.cpp"], diff --git a/executor/kv/kv_executor.cpp b/executor/kv/kv_executor.cpp index 9751e64533..19f53600e8 100644 --- a/executor/kv/kv_executor.cpp +++ b/executor/kv/kv_executor.cpp @@ -21,9 +21,23 @@ #include +#include + +#include "chain/storage/composite_key_codec.h" #include "executor/contract/executor/contract_executor.h" namespace resdb { +namespace { + +// Index entry keys start with "ck\0". SET must not write them, or a client +// could plant fake index entries. +bool IsCompositeKeyRange(const std::string& key) { + static const std::string kPrefix = std::string( + storage::kCompositeKeyNamespace) + storage::kCompositeKeyDelim; + return key.compare(0, kPrefix.size(), kPrefix) == 0; +} + +} // namespace KVExecutor::KVExecutor(std::unique_ptr storage) { storage_ = std::move(storage); @@ -75,6 +89,11 @@ std::unique_ptr KVExecutor::ExecuteRequest( std::string result = ExecuteSQL(kv_request.sql_query()); kv_response.set_sql_response(result); kv_response.set_value(result); // keep legacy field populated + } else if (kv_request.cmd() == KVRequest::CREATE_INDEX_ENTRY || + kv_request.cmd() == KVRequest::DELETE_INDEX_ENTRY || + kv_request.cmd() == KVRequest::UPDATE_INDEX_ENTRY || + kv_request.cmd() == KVRequest::QUERY_BY_INDEX) { + ExecuteIndexCommand(kv_request, &kv_response); } else if(!kv_request.smart_contract_request().empty()){ std::unique_ptr resp = contract_manager_->ExecuteData(kv_request.smart_contract_request()); @@ -129,6 +148,11 @@ std::unique_ptr KVExecutor::ExecuteData( std::string result = ExecuteSQL(kv_request.sql_query()); kv_response.set_sql_response(result); kv_response.set_value(result); // keep legacy field populated + } else if (kv_request.cmd() == KVRequest::CREATE_INDEX_ENTRY || + kv_request.cmd() == KVRequest::DELETE_INDEX_ENTRY || + kv_request.cmd() == KVRequest::UPDATE_INDEX_ENTRY || + kv_request.cmd() == KVRequest::QUERY_BY_INDEX) { + ExecuteIndexCommand(kv_request, &kv_response); } else if(!kv_request.smart_contract_request().empty()){ std::unique_ptr resp = contract_manager_->ExecuteData(kv_request.smart_contract_request()); @@ -145,6 +169,11 @@ std::unique_ptr KVExecutor::ExecuteData( } void KVExecutor::Set(const std::string& key, const std::string& value) { + if (IsCompositeKeyRange(key)) { + LOG(ERROR) << "SET rejected: keys starting with \"ck\\0\" are reserved " + "for secondary indexes"; + return; + } storage_->SetValueWithSeq(key, value, seq_); } @@ -162,6 +191,11 @@ std::string KVExecutor::GetRange(const std::string& min_key, void KVExecutor::SetWithVersion(const std::string& key, const std::string& value, int version) { + if (IsCompositeKeyRange(key)) { + LOG(ERROR) << "SET_WITH_VERSION rejected: keys starting with \"ck\\0\" " + "are reserved for secondary indexes"; + return; + } storage_->SetValueWithVersion(key, value, version); } @@ -220,5 +254,151 @@ std::string KVExecutor::ExecuteSQL(const std::string& sql_query) { return storage_->ExecuteSQL(sql_query); } +void KVExecutor::ExecuteIndexCommand(const KVRequest& request, + KVResponse* response) { + std::vector attributes(request.attributes().begin(), + request.attributes().end()); + switch (request.cmd()) { + case KVRequest::CREATE_INDEX_ENTRY: + response->set_success(CreateIndexEntry(request.index_name(), attributes, + request.primary_key())); + break; + case KVRequest::DELETE_INDEX_ENTRY: + response->set_success(DeleteIndexEntry(request.index_name(), attributes, + request.primary_key())); + break; + case KVRequest::UPDATE_INDEX_ENTRY: { + std::vector new_attributes(request.new_attributes().begin(), + request.new_attributes().end()); + response->set_success(UpdateIndexEntry(request.index_name(), attributes, + new_attributes, + request.primary_key())); + break; + } + case KVRequest::QUERY_BY_INDEX: + QueryByIndex(request.index_name(), attributes, request.with_values(), + response->mutable_items()); + break; + default: + break; + } +} + +// MemoryDB stores SET and SET_WITH_VERSION records separately, so check both. +bool KVExecutor::PrimaryKeyExists(const std::string& key) { + return !Get(key).empty() || + !storage_->GetValueWithVersion(key, 0).first.empty(); +} + +// Exact match: a prefix search for "user:1" also returns "user:10". +bool KVExecutor::IndexEntryExists(const std::string& composite_key) { + for (const std::string& key : + storage_->GetByCompositeKeyPrefix(composite_key)) { + if (key == composite_key) { + return true; + } + } + return false; +} + +bool KVExecutor::CreateIndexEntry(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key) { + if (index_name.empty() || primary_key.empty()) { + LOG(ERROR) << "CreateIndexEntry: index_name and primary_key are required"; + return false; + } + if (!PrimaryKeyExists(primary_key)) { + LOG(ERROR) << "CreateIndexEntry: primary key not found: " << primary_key; + return false; + } + std::string composite_key = + storage::EncodeCompositeKey(index_name, attributes, primary_key); + if (composite_key.empty()) { + LOG(ERROR) << "CreateIndexEntry: input contains a \\0 byte"; + return false; + } + return storage_->CreateCompositeKey(composite_key) == 0; +} + +bool KVExecutor::DeleteIndexEntry(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key) { + if (index_name.empty() || primary_key.empty()) { + LOG(ERROR) << "DeleteIndexEntry: index_name and primary_key are required"; + return false; + } + std::string composite_key = + storage::EncodeCompositeKey(index_name, attributes, primary_key); + if (composite_key.empty() || !IndexEntryExists(composite_key)) { + LOG(ERROR) << "DeleteIndexEntry: no such index entry"; + return false; + } + return storage_->DeleteCompositeKey(composite_key) == 0; +} + +bool KVExecutor::UpdateIndexEntry( + const std::string& index_name, + const std::vector& old_attributes, + const std::vector& new_attributes, + const std::string& primary_key) { + if (index_name.empty() || primary_key.empty()) { + LOG(ERROR) << "UpdateIndexEntry: index_name and primary_key are required"; + return false; + } + std::string old_key = + storage::EncodeCompositeKey(index_name, old_attributes, primary_key); + std::string new_key = + storage::EncodeCompositeKey(index_name, new_attributes, primary_key); + if (old_key.empty() || new_key.empty()) { + LOG(ERROR) << "UpdateIndexEntry: input contains a \\0 byte"; + return false; + } + if (!IndexEntryExists(old_key)) { + LOG(ERROR) << "UpdateIndexEntry: no existing entry to update"; + return false; + } + return storage_->UpdateCompositeKey(old_key, new_key) == 0; +} + +void KVExecutor::QueryByIndex(const std::string& index_name, + const std::vector& attribute_prefix, + bool with_values, Items* items) { + if (index_name.empty()) { + LOG(ERROR) << "QueryByIndex: index_name is required"; + return; + } + std::string prefix = + storage::EncodeCompositeKeyPrefix(index_name, attribute_prefix); + if (prefix.empty()) { + LOG(ERROR) << "QueryByIndex: input contains a \\0 byte"; + return; + } + // A record can have several matching entries (e.g. {"Davis", "active"} and + // {"Davis", "inactive"}), so collect into a set: each key once, sorted. + std::set primary_keys; + for (const std::string& composite_key : + storage_->GetByCompositeKeyPrefix(prefix)) { + std::string decoded_index; + std::vector decoded_attributes; + std::string primary_key; + if (storage::DecodeCompositeKey(composite_key, &decoded_index, + &decoded_attributes, &primary_key)) { + primary_keys.insert(primary_key); + } + } + for (const std::string& primary_key : primary_keys) { + Item* item = items->add_item(); + item->set_key(primary_key); + if (with_values) { + // Read here, so the caller doesn't send one request per match. + std::string value = Get(primary_key); + if (value.empty()) { + value = storage_->GetValueWithVersion(primary_key, 0).first; + } + item->mutable_value_info()->set_value(value); + } + } +} } // namespace resdb diff --git a/executor/kv/kv_executor.h b/executor/kv/kv_executor.h index f0199b21f5..93e70ed69f 100644 --- a/executor/kv/kv_executor.h +++ b/executor/kv/kv_executor.h @@ -21,7 +21,9 @@ #include #include +#include #include +#include #include "chain/storage/storage.h" #include "executor/common/transaction_manager.h" @@ -58,7 +60,35 @@ class KVExecutor : public TransactionManager { void GetTopHistory(const std::string& key, int top_number, Items* items); std::string ExecuteSQL(const std::string& sql_query); + // Secondary indexes (issue #180). An index entry says which attributes a + // record has, e.g. ("by_city", {"Davis"}) -> "user:1", and is stored as the + // key "ck\0by_city\0Davis\0user:1". Only these functions build such keys. + + // Fails if the record (primary_key) doesn't exist. + bool CreateIndexEntry(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key); + // Fails if the entry doesn't exist. + bool DeleteIndexEntry(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key); + // Moves an entry to new attributes in one write. Fails if it doesn't exist. + bool UpdateIndexEntry(const std::string& index_name, + const std::vector& old_attributes, + const std::vector& new_attributes, + const std::string& primary_key); + // Adds the primary keys of matching entries to `items`, each once, sorted. + // With `with_values`, each record's value is read and returned too. + void QueryByIndex(const std::string& index_name, + const std::vector& attribute_prefix, + bool with_values, Items* items); + private: + // Shared by ExecuteRequest and ExecuteData. + void ExecuteIndexCommand(const KVRequest& request, KVResponse* response); + bool PrimaryKeyExists(const std::string& key); + bool IndexEntryExists(const std::string& composite_key); + std::unique_ptr contract_manager_; }; diff --git a/executor/kv/kv_executor_index_test.cpp b/executor/kv/kv_executor_index_test.cpp new file mode 100644 index 0000000000..88fd7cff13 --- /dev/null +++ b/executor/kv/kv_executor_index_test.cpp @@ -0,0 +1,421 @@ +/* + * 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 +#include + +#include +#include +#include +#include +#include + +#include "chain/storage/leveldb.h" +#include "chain/storage/memory_db.h" +#include "executor/kv/kv_executor.h" +#include "proto/kv/kv.pb.h" + +namespace resdb { +namespace { + +using ::testing::Combine; +using ::testing::ElementsAre; +using ::testing::IsEmpty; +using ::testing::Values; + +enum Backend { MEMORY_DB, LEVEL_DB }; +// ExecuteData and ExecuteRequest each dispatch commands; test both. +enum EntryPoint { EXECUTE_DATA, EXECUTE_REQUEST }; + +class KVExecutorIndexTest + : public ::testing::TestWithParam> { + protected: + KVExecutorIndexTest() { + std::unique_ptr storage; + if (std::get<0>(GetParam()) == LEVEL_DB) { + db_path_ = ::testing::TempDir() + "kv_executor_index_test_db"; + std::filesystem::remove_all(db_path_); + storage = storage::NewResLevelDB(db_path_); + } else { + storage = std::make_unique(); + } + executor_ = std::make_unique(std::move(storage)); + } + + ~KVExecutorIndexTest() override { + executor_.reset(); // close LevelDB before deleting its folder + if (!db_path_.empty()) { + std::filesystem::remove_all(db_path_); + } + } + + KVResponse Execute(const KVRequest& request) { + std::unique_ptr raw; + if (std::get<1>(GetParam()) == EXECUTE_DATA) { + std::string data; + request.SerializeToString(&data); + raw = executor_->ExecuteData(data); + } else { + raw = executor_->ExecuteRequest(request); + } + KVResponse response; + if (raw != nullptr) { + response.ParseFromString(*raw); + } + return response; + } + + void Set(const std::string& key, const std::string& value) { + KVRequest request; + request.set_cmd(KVRequest::SET); + request.set_key(key); + request.set_value(value); + Execute(request); + } + + void SetWithVersion(const std::string& key, const std::string& value) { + KVRequest request; + request.set_cmd(KVRequest::SET_WITH_VERSION); + request.set_key(key); + request.set_value(value); + request.set_version(0); + Execute(request); + } + + std::string Get(const std::string& key) { + KVRequest request; + request.set_cmd(KVRequest::GET); + request.set_key(key); + return Execute(request).value(); + } + + bool IndexCommand(KVRequest::CMD cmd, const std::string& index, + const std::vector& attributes, + const std::string& primary_key, + const std::vector& new_attributes = {}) { + KVRequest request; + request.set_cmd(cmd); + request.set_index_name(index); + for (const auto& a : attributes) request.add_attributes(a); + for (const auto& a : new_attributes) request.add_new_attributes(a); + request.set_primary_key(primary_key); + return Execute(request).success(); + } + + bool Create(const std::string& index, const std::vector& attrs, + const std::string& pk) { + return IndexCommand(KVRequest::CREATE_INDEX_ENTRY, index, attrs, pk); + } + + bool Delete(const std::string& index, const std::vector& attrs, + const std::string& pk) { + return IndexCommand(KVRequest::DELETE_INDEX_ENTRY, index, attrs, pk); + } + + bool Update(const std::string& index, const std::vector& from, + const std::vector& to, const std::string& pk) { + return IndexCommand(KVRequest::UPDATE_INDEX_ENTRY, index, from, pk, to); + } + + std::vector> QueryWithValues( + const std::string& index, const std::vector& attrs) { + KVRequest request; + request.set_cmd(KVRequest::QUERY_BY_INDEX); + request.set_index_name(index); + for (const auto& a : attrs) request.add_attributes(a); + request.set_with_values(true); + KVResponse response = Execute(request); + std::vector> records; + for (const Item& item : response.items().item()) { + records.push_back({item.key(), item.value_info().value()}); + } + return records; + } + + std::vector Query(const std::string& index, + const std::vector& attrs) { + KVRequest request; + request.set_cmd(KVRequest::QUERY_BY_INDEX); + request.set_index_name(index); + for (const auto& a : attrs) request.add_attributes(a); + // Keep the response in a variable: looping over Execute(...).items() + // directly would read from a temporary that's already been destroyed. + KVResponse response = Execute(request); + std::vector keys; + for (const Item& item : response.items().item()) { + keys.push_back(item.key()); + } + return keys; + } + + std::string db_path_; + std::unique_ptr executor_; +}; + +// A string holding a real zero byte (a plain "a\0b" literal stops at the \0). +std::string WithNul(const std::string& a, const std::string& b) { + return a + std::string(1, '\0') + b; +} + +TEST_P(KVExecutorIndexTest, CreateThenQuery) { + Set("user:1", "Ana"); + Set("user:2", "Bo"); + Set("user:3", "Cy"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:2")); + EXPECT_TRUE(Create("by_city", {"Sacramento"}, "user:3")); + + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:1", "user:2")); + EXPECT_THAT(Query("by_city", {"Sacramento"}), ElementsAre("user:3")); + EXPECT_THAT(Query("by_city", {"Chico"}), IsEmpty()); +} + +TEST_P(KVExecutorIndexTest, CreateFailsWhenRecordDoesNotExist) { + EXPECT_FALSE(Create("by_city", {"Davis"}, "user:99")); + EXPECT_THAT(Query("by_city", {"Davis"}), IsEmpty()); +} + +// MemoryDB keeps SET_WITH_VERSION records in a separate map from SET records. +TEST_P(KVExecutorIndexTest, CreateWorksForRecordWrittenWithVersion) { + SetWithVersion("user:5", "Eve"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:5")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:5")); +} + +TEST_P(KVExecutorIndexTest, CreateRejectsBadInput) { + Set("user:1", "Ana"); + EXPECT_FALSE(Create("", {"Davis"}, "user:1")); + EXPECT_FALSE(Create("by_city", {"Davis"}, "")); + EXPECT_FALSE(Create("by_city", {WithNul("Da", "vis")}, "user:1")); + EXPECT_FALSE(Create(WithNul("by", "city"), {"Davis"}, "user:1")); + EXPECT_THAT(Query("by_city", {"Davis"}), IsEmpty()); +} + +TEST_P(KVExecutorIndexTest, CreatingTwiceKeepsOneEntry) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:1")); +} + +// "Davis" must not match "Davisville". +TEST_P(KVExecutorIndexTest, QueryMatchesWholeAttribute) { + Set("user:1", "Ana"); + Set("user:2", "Bo"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Create("by_city", {"Davisville"}, "user:2")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:1")); +} + +// Same attribute, different index: queries must not mix them. +TEST_P(KVExecutorIndexTest, QueryOnlyReadsItsOwnIndex) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_THAT(Query("by_birthplace", {"Davis"}), IsEmpty()); + EXPECT_THAT(Query("by_cit", {"Davis"}), IsEmpty()); +} + +TEST_P(KVExecutorIndexTest, QueryWithLeadingAttributes) { + Set("user:1", "Ana"); + Set("user:2", "Bo"); + EXPECT_TRUE(Create("by_city_status", {"Davis", "active"}, "user:1")); + EXPECT_TRUE(Create("by_city_status", {"Davis", "inactive"}, "user:2")); + EXPECT_THAT(Query("by_city_status", {"Davis"}), + ElementsAre("user:1", "user:2")); + EXPECT_THAT(Query("by_city_status", {"Davis", "active"}), + ElementsAre("user:1")); +} + +// No attributes = every entry in the index. +TEST_P(KVExecutorIndexTest, QueryWithNoAttributesReturnsWholeIndex) { + Set("user:1", "Ana"); + Set("user:2", "Bo"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Create("by_city", {"Sacramento"}, "user:2")); + EXPECT_THAT(Query("by_city", {}), ElementsAre("user:1", "user:2")); +} + +TEST_P(KVExecutorIndexTest, QueryRejectsBadInput) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_THAT(Query("", {"Davis"}), IsEmpty()); + EXPECT_THAT(Query("by_city", {WithNul("Da", "vis")}), IsEmpty()); +} + +TEST_P(KVExecutorIndexTest, DeleteRemovesOnlyThatEntry) { + Set("user:1", "Ana"); + Set("user:10", "Jo"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:10")); + + EXPECT_TRUE(Delete("by_city", {"Davis"}, "user:1")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:10")); + // user:1's entry is gone; "user:10" must not count as a match for it. + EXPECT_FALSE(Delete("by_city", {"Davis"}, "user:1")); +} + +TEST_P(KVExecutorIndexTest, DeleteFailsWhenEntryDoesNotExist) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_FALSE(Delete("by_city", {"Sacramento"}, "user:1")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:1")); +} + +TEST_P(KVExecutorIndexTest, UpdateMovesEntry) { + Set("user:1", "Ana"); + Set("user:2", "Bo"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:2")); + + EXPECT_TRUE(Update("by_city", {"Davis"}, {"Sacramento"}, "user:2")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:1")); + EXPECT_THAT(Query("by_city", {"Sacramento"}), ElementsAre("user:2")); +} + +// A wrong "from" must change nothing, and must not create the new entry. +TEST_P(KVExecutorIndexTest, UpdateFailsWhenEntryDoesNotExist) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_FALSE(Update("by_city", {"Chico"}, {"Sacramento"}, "user:1")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:1")); + EXPECT_THAT(Query("by_city", {"Sacramento"}), IsEmpty()); +} + +TEST_P(KVExecutorIndexTest, UpdateRejectsBadInput) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_FALSE(Update("by_city", {"Davis"}, {WithNul("Sac", "ramento")}, + "user:1")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:1")); +} + +// Values are opaque bytes, so a record holding an image indexes like any other. +TEST_P(KVExecutorIndexTest, IndexRecordWithBinaryValue) { + std::string image("\x89PNG\r\n\x1a\n", 8); // PNG signature + image += std::string(3, '\0'); // zero bytes + image += std::string("\xff\xfe\xfd", 3); // not valid text + + Set("img:1", image); + EXPECT_EQ(Get("img:1"), image); + EXPECT_TRUE(Create("by_type", {"png"}, "img:1")); + EXPECT_THAT(Query("by_type", {"png"}), ElementsAre("img:1")); +} + +// A record with an empty value can't be told apart from a missing one. +TEST_P(KVExecutorIndexTest, CannotIndexRecordWithEmptyValue) { + Set("user:1", ""); + EXPECT_FALSE(Create("by_city", {"Davis"}, "user:1")); +} + +// SET and SET_WITH_VERSION must not be able to plant a fake index entry. +TEST_P(KVExecutorIndexTest, SetCannotWriteIndexEntries) { + const std::string forged = std::string("ck") + '\0' + "by_city" + '\0' + + "Davis" + '\0' + "user:666"; + Set(forged, "x"); + SetWithVersion(forged, "x"); + EXPECT_THAT(Query("by_city", {"Davis"}), IsEmpty()); + EXPECT_EQ(Get(forged), ""); +} + +// Keys that merely start with "ck" (no \0) are ordinary user keys. +TEST_P(KVExecutorIndexTest, SetStillAllowsKeysStartingWithCk) { + Set("ck_balance", "100"); + EXPECT_EQ(Get("ck_balance"), "100"); +} + +// user:1 has two matching entries; it must come back once. +TEST_P(KVExecutorIndexTest, QueryReturnsEachRecordOnce) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city_status", {"Davis", "active"}, "user:1")); + EXPECT_TRUE(Create("by_city_status", {"Davis", "inactive"}, "user:1")); + EXPECT_THAT(Query("by_city_status", {"Davis"}), ElementsAre("user:1")); +} + +// Stored order is by attributes ("active" < "inactive"), so user:2 comes +// first internally; results must still be sorted by primary key. +TEST_P(KVExecutorIndexTest, QueryResultsAreSortedByPrimaryKey) { + Set("user:1", "Ana"); + Set("user:2", "Bo"); + EXPECT_TRUE(Create("by_city_status", {"Davis", "inactive"}, "user:1")); + EXPECT_TRUE(Create("by_city_status", {"Davis", "active"}, "user:2")); + EXPECT_THAT(Query("by_city_status", {"Davis"}), + ElementsAre("user:1", "user:2")); +} + +TEST_P(KVExecutorIndexTest, UpdateToSameAttributesKeepsEntry) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Update("by_city", {"Davis"}, {"Davis"}, "user:1")); + EXPECT_THAT(Query("by_city", {"Davis"}), ElementsAre("user:1")); +} + +TEST_P(KVExecutorIndexTest, UpdateOntoExistingEntry) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Create("by_city", {"Sacramento"}, "user:1")); + EXPECT_TRUE(Update("by_city", {"Davis"}, {"Sacramento"}, "user:1")); + EXPECT_THAT(Query("by_city", {"Davis"}), IsEmpty()); + EXPECT_THAT(Query("by_city", {"Sacramento"}), ElementsAre("user:1")); +} + +TEST_P(KVExecutorIndexTest, QueryWithValuesReturnsRecords) { + Set("user:1", "Ana"); + Set("user:2", "Bo"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:2")); + + EXPECT_THAT(QueryWithValues("by_city", {"Davis"}), + ElementsAre(std::make_pair("user:1", "Ana"), + std::make_pair("user:2", "Bo"))); +} + +// Without the flag the values must stay out of the reply. +TEST_P(KVExecutorIndexTest, QueryWithoutValuesReturnsKeysOnly) { + Set("user:1", "Ana"); + EXPECT_TRUE(Create("by_city", {"Davis"}, "user:1")); + + KVRequest request; + request.set_cmd(KVRequest::QUERY_BY_INDEX); + request.set_index_name("by_city"); + request.add_attributes("Davis"); + KVResponse response = Execute(request); + ASSERT_EQ(response.items().item_size(), 1); + EXPECT_EQ(response.items().item(0).key(), "user:1"); + EXPECT_EQ(response.items().item(0).value_info().value(), ""); +} + +TEST_P(KVExecutorIndexTest, QueryWithValuesHandlesBinaryAndVersionedRecords) { + std::string image("\x89PNG\r\n\x1a\n", 8); + image += std::string(3, '\0'); + Set("img:1", image); + SetWithVersion("user:5", "Eve"); + EXPECT_TRUE(Create("by_type", {"png"}, "img:1")); + EXPECT_TRUE(Create("by_type", {"png"}, "user:5")); + + EXPECT_THAT(QueryWithValues("by_type", {"png"}), + ElementsAre(std::make_pair("img:1", image), + std::make_pair("user:5", "Eve"))); +} + +INSTANTIATE_TEST_SUITE_P(AllBackendsAndEntryPoints, KVExecutorIndexTest, + Combine(Values(MEMORY_DB, LEVEL_DB), + Values(EXECUTE_DATA, EXECUTE_REQUEST))); + +} // namespace +} // namespace resdb diff --git a/interface/kv/kv_client.cpp b/interface/kv/kv_client.cpp index e9075dbdd0..ef44d3f1b9 100644 --- a/interface/kv/kv_client.cpp +++ b/interface/kv/kv_client.cpp @@ -22,6 +22,20 @@ #include namespace resdb { +namespace { + +// Fills the parts every index command sends. +void SetIndexFields(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key, KVRequest* request) { + request->set_index_name(index_name); + for (const std::string& attribute : attributes) { + request->add_attributes(attribute); + } + request->set_primary_key(primary_key); +} + +} // namespace KVClient::KVClient(const ResDBConfig& config) : TransactionConstructor(config) {} @@ -159,4 +173,73 @@ std::unique_ptr KVClient::QueryResQL(const std::string& sql_query) return std::make_unique(); } +int KVClient::CreateIndexEntry(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key) { + KVRequest request; + request.set_cmd(KVRequest::CREATE_INDEX_ENTRY); + SetIndexFields(index_name, attributes, primary_key, &request); + + KVResponse response; + int ret = SendRequest(request, &response); + if (ret != 0) { + LOG(ERROR) << "send request fail, ret:" << ret; + return ret; + } + return response.success() ? 0 : -3; +} + +int KVClient::DeleteIndexEntry(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key) { + KVRequest request; + request.set_cmd(KVRequest::DELETE_INDEX_ENTRY); + SetIndexFields(index_name, attributes, primary_key, &request); + + KVResponse response; + int ret = SendRequest(request, &response); + if (ret != 0) { + LOG(ERROR) << "send request fail, ret:" << ret; + return ret; + } + return response.success() ? 0 : -3; +} + +int KVClient::UpdateIndexEntry(const std::string& index_name, + const std::vector& old_attributes, + const std::vector& new_attributes, + const std::string& primary_key) { + KVRequest request; + request.set_cmd(KVRequest::UPDATE_INDEX_ENTRY); + SetIndexFields(index_name, old_attributes, primary_key, &request); + for (const std::string& attribute : new_attributes) { + request.add_new_attributes(attribute); + } + + KVResponse response; + int ret = SendRequest(request, &response); + if (ret != 0) { + LOG(ERROR) << "send request fail, ret:" << ret; + return ret; + } + return response.success() ? 0 : -3; +} + +std::unique_ptr KVClient::QueryByIndex( + const std::string& index_name, + const std::vector& attribute_prefix, bool with_values) { + KVRequest request; + request.set_cmd(KVRequest::QUERY_BY_INDEX); + SetIndexFields(index_name, attribute_prefix, "", &request); + request.set_with_values(with_values); + + KVResponse response; + int ret = SendRequest(request, &response); + if (ret != 0) { + LOG(ERROR) << "send request fail, ret:" << ret; + return nullptr; + } + return std::make_unique(response.items()); +} + } // namespace resdb diff --git a/interface/kv/kv_client.h b/interface/kv/kv_client.h index 9a90d09033..8cdd6fac39 100644 --- a/interface/kv/kv_client.h +++ b/interface/kv/kv_client.h @@ -19,6 +19,10 @@ #pragma once +#include +#include +#include + #include "interface/rdbc/transaction_constructor.h" #include "proto/kv/kv.pb.h" @@ -63,6 +67,27 @@ class KVClient : public TransactionConstructor { // Execute an arbitrary SQL query with the ReSQL query service, based on DuckDB. std::unique_ptr QueryResQL(const std::string& sql_query); + + // Secondary indexes (issue #180). The server builds the stored key from + // these parts. Return 0 on success, -3 if the servers rejected the request, + // or the send error from TransactionConstructor::SendRequest (-1, -2). + int CreateIndexEntry(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key); + int DeleteIndexEntry(const std::string& index_name, + const std::vector& attributes, + const std::string& primary_key); + int UpdateIndexEntry(const std::string& index_name, + const std::vector& old_attributes, + const std::vector& new_attributes, + const std::string& primary_key); + // Primary keys whose entry starts with `attribute_prefix`, each once and + // sorted. Empty prefix means the whole index. nullptr if the send failed. + // With `with_values`, each record's value comes back in the same reply. + std::unique_ptr QueryByIndex( + const std::string& index_name, + const std::vector& attribute_prefix, + bool with_values = false); }; } // namespace resdb diff --git a/proto/kv/kv.proto b/proto/kv/kv.proto index 6134030480..e1a4f2d810 100644 --- a/proto/kv/kv.proto +++ b/proto/kv/kv.proto @@ -36,6 +36,11 @@ message KVRequest { GET_TOP = 10; // NEW: SQL command type SQL = 11; + // Secondary index (composite key) commands, see issue #180. + CREATE_INDEX_ENTRY = 12; + DELETE_INDEX_ENTRY = 13; + UPDATE_INDEX_ENTRY = 14; + QUERY_BY_INDEX = 15; } CMD cmd = 1; string key = 2; @@ -52,6 +57,15 @@ message KVRequest { bytes smart_contract_request = 10; // NEW: raw SQL text for SQL commands (cmd == SQL) string sql_query = 11; + // NEW: For index commands. The server builds the stored key from these. + string index_name = 12; + repeated string attributes = 13; + string primary_key = 14; + // For UPDATE_INDEX_ENTRY: the attributes to move the entry to. + repeated string new_attributes = 15; + // For QUERY_BY_INDEX: also return each record's value, saving the caller + // one request per match. + bool with_values = 16; } message ValueInfo { @@ -76,4 +90,6 @@ message KVResponse { bytes smart_contract_response = 10; // NEW: response payload for SQL commands string sql_response = 11; + // Whether CREATE/DELETE/UPDATE_INDEX_ENTRY made the change. + bool success = 12; } diff --git a/service/tools/kv/api_tools/composite_key_e2e_test.sh b/service/tools/kv/api_tools/composite_key_e2e_test.sh new file mode 100755 index 0000000000..83d6f07a77 --- /dev/null +++ b/service/tools/kv/api_tools/composite_key_e2e_test.sh @@ -0,0 +1,179 @@ +#!/bin/bash +# +# 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. +# +# End-to-end test for secondary indexes (composite keys), issue #180. +# Starts a local cluster with LevelDB, runs every index command against it and +# checks the results. Run from the repository root: +# +# ./service/tools/kv/api_tools/composite_key_e2e_test.sh +# +# Exits 0 and prints PASS if everything matched, non-zero otherwise. + +set -u + +if [ ! -f WORKSPACE ]; then + echo "run this from the repository root" + exit 1 +fi + +# Leave no servers behind, even on Ctrl-C. +trap 'killall -9 kv_service 2>/dev/null' EXIT INT TERM + +CONFIG=service/tools/config/interface/service.config +CLI=bazel-bin/service/tools/kv/api_tools/kv_service_tools +CERT_DIR=service/tools/data/cert +FAILURES=0 + +run() { + $CLI --config $CONFIG "$@" 2>/dev/null +} + +# Prints just the primary keys a query returned, sorted, space separated. +# A request that never reached the servers must not look like "no matches", +# so the header line has to be there before we strip it. +query() { + local index=$1 + local attrs=${2:-} + local out + if [ -z "$attrs" ]; then + out=$(run --cmd query_by_index --index "$index") + else + out=$(run --cmd query_by_index --index "$index" --attrs "$attrs") + fi + case "$out" in + *"result(s)"*) ;; + *) + echo "QUERY FAILED" + return + ;; + esac + echo "$out" | tail -n +2 | sort | tr '\n' ' ' +} + +check() { + local what=$1 got=$2 want=$3 + if [ "$got" = "$want" ]; then + echo " ok $what" + else + echo " FAIL $what" + echo " expected: '$want'" + echo " got: '$got'" + FAILURES=$((FAILURES + 1)) + fi +} + +start_cluster() { + killall -9 kv_service 2>/dev/null + sleep 1 + ./service/tools/kv/server_tools/start_kv_service.sh \ + --define enable_leveldb=True > /dev/null 2>&1 + sleep 5 +} + +echo "== setup ==" +if [ ! -f "$CERT_DIR/cert_1.cert" ]; then + echo " generating keys and certificates" + ./service/tools/kv/server_tools/generate_keys_and_certs.sh > /dev/null 2>&1 +fi +bazel build //service/tools/kv/api_tools:kv_service_tools > /dev/null 2>&1 || + { echo " FAIL could not build the CLI"; exit 1; } +rm -rf ./*_db +start_cluster +echo " cluster started with a fresh database" + +echo "== store records ==" +# set is fire-and-forget, so every write is confirmed with a read. +run --cmd set --key photo:1 --value jpeg-1 > /dev/null +run --cmd set --key photo:2 --value png-2 > /dev/null +run --cmd set --key photo:3 --value jpeg-3 > /dev/null +check "photo:1 stored" "$(run --cmd get --key photo:1)" \ + "get key = photo:1 value = jpeg-1" +check "photo:2 stored" "$(run --cmd get --key photo:2)" \ + "get key = photo:2 value = png-2" +check "photo:3 stored" "$(run --cmd get --key photo:3)" \ + "get key = photo:3 value = jpeg-3" + +echo "== create index entries ==" +run --cmd create_index_entry --index by_owner_format --attrs shray,jpeg \ + --pk photo:1 > /dev/null +run --cmd create_index_entry --index by_owner_format --attrs shray,png \ + --pk photo:2 > /dev/null +run --cmd create_index_entry --index by_owner_format --attrs ana,jpeg \ + --pk photo:3 > /dev/null +check "query by owner" "$(query by_owner_format shray)" "photo:1 photo:2 " +check "query by owner+format" "$(query by_owner_format shray,jpeg)" "photo:1 " +check "whole index" "$(query by_owner_format)" "photo:1 photo:2 photo:3 " +check "no match" "$(query by_owner_format nobody)" "" +# Attributes match from the left only: "jpeg" is the second one. +check "second attribute alone finds nothing" \ + "$(query by_owner_format jpeg)" "" + +echo "== rejected requests ==" +check "indexing a record that does not exist" \ + "$(run --cmd create_index_entry --index by_owner_format --attrs x,y \ + --pk photo:999 | grep -c 'ret = -3')" "1" +check "deleting an entry that does not exist" \ + "$(run --cmd delete_index_entry --index by_owner_format --attrs x,y \ + --pk photo:1 | grep -c 'ret = -3')" "1" +check "updating an entry that does not exist" \ + "$(run --cmd update_index_entry --index by_owner_format --attrs x,y \ + --new_attrs a,b --pk photo:1 | grep -c 'ret = -3')" "1" +check "rejected create left the index unchanged" \ + "$(query by_owner_format)" "photo:1 photo:2 photo:3 " + +echo "== update ==" +run --cmd update_index_entry --index by_owner_format --attrs shray,png \ + --new_attrs shray,jpeg --pk photo:2 > /dev/null +check "moved out of the old attributes" "$(query by_owner_format shray,png)" "" +check "moved into the new attributes" \ + "$(query by_owner_format shray,jpeg)" "photo:1 photo:2 " + +echo "== delete ==" +run --cmd delete_index_entry --index by_owner_format --attrs ana,jpeg \ + --pk photo:3 > /dev/null +check "entry removed" "$(query by_owner_format ana)" "" +check "the record itself is untouched" "$(run --cmd get --key photo:3)" \ + "get key = photo:3 value = jpeg-3" + +echo "== index entries stay out of normal reads ==" +check "get on a record is unaffected" "$(run --cmd get --key photo:1)" \ + "get key = photo:1 value = jpeg-1" +# "ck\0..." sorts between the photo keys, and an index entry's value is empty, +# so an unfiltered range scan would splice empty elements into this list. +check "range scan hides index entries" \ + "$(run --cmd get_key_range --min_key ' ' --max_key '~' | tr -d '\000' | + grep -acE '\[,|,,|,\]')" "0" + +echo "== restart: the index is on disk ==" +start_cluster +check "index survived the restart" \ + "$(query by_owner_format shray,jpeg)" "photo:1 photo:2 " +check "records survived the restart" "$(run --cmd get --key photo:1)" \ + "get key = photo:1 value = jpeg-1" + +echo "== teardown ==" +killall -9 kv_service 2>/dev/null +echo " cluster stopped" + +if [ $FAILURES -eq 0 ]; then + echo "PASS" + exit 0 +fi +echo "FAIL: $FAILURES check(s) failed" +exit 1 diff --git a/service/tools/kv/api_tools/kv_service_tools.cpp b/service/tools/kv/api_tools/kv_service_tools.cpp index 75c8780d07..3931c74e18 100644 --- a/service/tools/kv/api_tools/kv_service_tools.cpp +++ b/service/tools/kv/api_tools/kv_service_tools.cpp @@ -40,7 +40,9 @@ void ShowUsage() { "--config: config path\n" "--cmd " "set/get/set_with_version/get_with_version/get_key_range/" - "get_key_range_with_version/get_top/get_history/sql\n" + "get_key_range_with_version/get_top/get_history/sql/" + "create_index_entry/delete_index_entry/update_index_entry/" + "query_by_index\n" "--key key\n" "--value value, if cmd is a get operation\n" "--sql SQL text, if cmd is sql\n" @@ -50,6 +52,12 @@ void ShowUsage() { "--min_version, if cmd is get_history\n" "--max_version, if cmd is get_history\n" "--top, if cmd is get_top\n" + "--index index name, if cmd is an index command\n" + "--attrs comma-separated attribute values, e.g. Davis or shray,jpeg\n" + "--new_attrs the attributes to move the entry to, if cmd is " + "update_index_entry\n" + "--pk primary key of the record being indexed\n" + "--with_values return each record's value with query_by_index\n" "\n" "More examples can be found from README.\n"); } @@ -67,8 +75,34 @@ static struct option long_options[] = { {"max_key", required_argument, NULL, 'Y'}, {"top", required_argument, NULL, 't'}, {"sql", required_argument, NULL, 'q'}, + {"index", required_argument, NULL, 'i'}, + {"attrs", required_argument, NULL, 'a'}, + {"new_attrs", required_argument, NULL, 'n'}, + {"pk", required_argument, NULL, 'p'}, + {"with_values", no_argument, NULL, 'w'}, + // getopt_long reads until it finds this terminator. + {NULL, 0, NULL, 0}, }; +// "shray,jpeg" -> {"shray", "jpeg"}; "" -> {} (meaning the whole index). +std::vector SplitAttributes(const std::string& text) { + std::vector attributes; + if (text.empty()) { + return attributes; + } + size_t start = 0; + while (true) { + size_t comma = text.find(',', start); + if (comma == std::string::npos) { + attributes.push_back(text.substr(start)); + break; + } + attributes.push_back(text.substr(start, comma - start)); + start = comma + 1; + } + return attributes; +} + void OldAPI(char** argv) { std::string client_config_file = argv[1]; std::string cmd = argv[2]; @@ -122,6 +156,11 @@ int main(int argc, char** argv) { std::string min_key, max_key; std::string value; std::string sql; + std::string index_name; + std::string attrs; + std::string new_attrs; + std::string pk; + bool with_values = false; std::string client_config_file; int top = 0; // getopt_long returns int and signals the end of the options with -1. Storing @@ -174,6 +213,21 @@ int main(int argc, char** argv) { case 'q': sql = optarg; break; + case 'i': + index_name = optarg; + break; + case 'a': + attrs = optarg; + break; + case 'n': + new_attrs = optarg; + break; + case 'p': + pk = optarg; + break; + case 'w': + with_values = true; + break; case 'h': ShowUsage(); break; @@ -288,6 +342,55 @@ int main(int argc, char** argv) { } else { printf("SQL query failed\n"); } + } else if (cmd == "create_index_entry" || cmd == "delete_index_entry") { + if (index_name.empty() || pk.empty()) { + ShowUsage(); + return 0; + } + int ret = cmd == "create_index_entry" + ? client.CreateIndexEntry(index_name, SplitAttributes(attrs), + pk) + : client.DeleteIndexEntry(index_name, SplitAttributes(attrs), + pk); + printf("%s index = %s, attrs = %s, pk = %s, ret = %d%s\n", cmd.c_str(), + index_name.c_str(), attrs.c_str(), pk.c_str(), ret, + ret == -3 ? " (rejected by the servers)" : ""); + return ret == 0 ? 0 : 1; + } else if (cmd == "update_index_entry") { + if (index_name.empty() || pk.empty()) { + ShowUsage(); + return 0; + } + int ret = client.UpdateIndexEntry(index_name, SplitAttributes(attrs), + SplitAttributes(new_attrs), pk); + printf( + "update_index_entry index = %s, attrs = %s, new_attrs = %s, pk = %s, " + "ret = %d%s\n", + index_name.c_str(), attrs.c_str(), new_attrs.c_str(), pk.c_str(), ret, + ret == -3 ? " (rejected by the servers)" : ""); + return ret == 0 ? 0 : 1; + } else if (cmd == "query_by_index") { + if (index_name.empty()) { + ShowUsage(); + return 0; + } + auto res = client.QueryByIndex(index_name, SplitAttributes(attrs), + with_values); + if (res == nullptr) { + printf("query_by_index index = %s, attrs = %s failed\n", + index_name.c_str(), attrs.c_str()); + return 1; + } + printf("query_by_index index = %s, attrs = %s, %d result(s)\n", + index_name.c_str(), attrs.c_str(), res->item_size()); + for (const auto& item : res->item()) { + if (with_values) { + printf("%s = %s\n", item.key().c_str(), + item.value_info().value().c_str()); + } else { + printf("%s\n", item.key().c_str()); + } + } } else { ShowUsage(); }