From 9e2177761a572ad172a1090c173efa35d27ecf58 Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Tue, 22 Sep 2026 01:55:33 -0700 Subject: [PATCH 01/11] chain/storage: fix gaps in composite-key storage Four gaps in the composite-key storage layer: - LevelDB GetRange returned index entries as if they were user data. GetAllItems and GetKeyRange already skipped them; GetRange was missed. - DuckDB's composite-key methods returned 0 (success) while doing nothing. They now return -1. - EncodeCompositeKeyPrefix with no attributes added an extra \0, so a query for a whole index matched nothing. - DecodeCompositeKey didn't check for the \0 after "ck", so "ckXidx..." decoded as a valid entry. Each fix has a test. Nothing changes for keys that don't start with "ck\0". --- chain/storage/composite_key_codec.cpp | 10 +++++++--- chain/storage/composite_key_codec_test.cpp | 16 ++++++++++++++++ chain/storage/duckdb.h | 11 +++++------ chain/storage/kv_storage_test.cpp | 14 ++++++++++++++ chain/storage/leveldb.cpp | 8 ++++++++ 5 files changed, 50 insertions(+), 9 deletions(-) diff --git a/chain/storage/composite_key_codec.cpp b/chain/storage/composite_key_codec.cpp index 781fa9954..3df108941 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 30509f5f9..306dc2144 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 153c19c61..8573669c2 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 b6719072c..f3ebbac87 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 15b1a8511..71c253dfa 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()); From 261d6a4a74d0a759ab921fd225376202dd5b804d Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Tue, 22 Sep 2026 01:55:41 -0700 Subject: [PATCH 02/11] executor/kv: add secondary index commands for composite keys Adds CREATE_INDEX_ENTRY, DELETE_INDEX_ENTRY, UPDATE_INDEX_ENTRY and QUERY_BY_INDEX so clients can use the composite keys added in the storage layer. The client sends the parts (index name, attributes, primary key) and the executor builds the stored key, so a client can't write any key outside the "ck\0" range. SET and SET_WITH_VERSION now refuse keys in that range. - Create fails if the record doesn't exist. - Delete and update fail if the entry doesn't exist. - Query returns each matching primary key once, sorted. - ExecuteRequest and ExecuteData both call one shared helper. Tests: kv_executor_index_test runs 21 cases on MemoryDB and LevelDB, through both ExecuteData and ExecuteRequest. --- executor/kv/BUILD | 12 + executor/kv/kv_executor.cpp | 170 ++++++++++++ executor/kv/kv_executor.h | 29 +++ executor/kv/kv_executor_index_test.cpp | 348 +++++++++++++++++++++++++ proto/kv/kv.proto | 13 + 5 files changed, 572 insertions(+) create mode 100644 executor/kv/kv_executor_index_test.cpp diff --git a/executor/kv/BUILD b/executor/kv/BUILD index 89bb67c7f..5790e016f 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 9751e6453..54b84a4f7 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,141 @@ 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, 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, + 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) { + items->add_item()->set_key(primary_key); + } +} } // namespace resdb diff --git a/executor/kv/kv_executor.h b/executor/kv/kv_executor.h index f0199b21f..b0b78089e 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,34 @@ 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. + void QueryByIndex(const std::string& index_name, + const std::vector& attribute_prefix, + 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 000000000..568fbe9db --- /dev/null +++ b/executor/kv/kv_executor_index_test.cpp @@ -0,0 +1,348 @@ +/* + * 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 "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 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")); +} + +// 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")); +} + +INSTANTIATE_TEST_SUITE_P(AllBackendsAndEntryPoints, KVExecutorIndexTest, + Combine(Values(MEMORY_DB, LEVEL_DB), + Values(EXECUTE_DATA, EXECUTE_REQUEST))); + +} // namespace +} // namespace resdb diff --git a/proto/kv/kv.proto b/proto/kv/kv.proto index 613403048..b4afc8559 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,12 @@ 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; } message ValueInfo { @@ -76,4 +87,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; } From 0004d3a8925c86c1e0424b0f1d8acd588f535c3f Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Tue, 22 Sep 2026 13:45:12 -0700 Subject: [PATCH 03/11] executor/kv: test indexing records with binary and empty values Values are opaque bytes, so a record can hold an image or any other binary data. Adds a test that stores PNG-like bytes (including zero bytes), indexes the record and queries it back. Also documents the one limit: a record whose value is empty can't be indexed, because it looks the same as a missing record. --- executor/kv/kv_executor_index_test.cpp | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/executor/kv/kv_executor_index_test.cpp b/executor/kv/kv_executor_index_test.cpp index 568fbe9db..3ae59558f 100644 --- a/executor/kv/kv_executor_index_test.cpp +++ b/executor/kv/kv_executor_index_test.cpp @@ -289,6 +289,24 @@ TEST_P(KVExecutorIndexTest, UpdateRejectsBadInput) { 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' + From f44f7be5ba0773a340024477c2a8b6a195e15eea Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Thu, 24 Sep 2026 14:42:39 -0700 Subject: [PATCH 04/11] executor/kv: return record values with query_by_index A query returns the primary keys of the matching records, so an application that wants the records has to fetch them one by one, and each of those is its own request through consensus. With with_values set, the executor reads the matching records itself and returns them in the same reply, so a query costs one request whatever the match count. Off by default, so a query still returns keys only unless it is asked for. --- executor/kv/kv_executor.cpp | 16 ++++++-- executor/kv/kv_executor.h | 3 +- executor/kv/kv_executor_index_test.cpp | 55 ++++++++++++++++++++++++++ proto/kv/kv.proto | 3 ++ 4 files changed, 73 insertions(+), 4 deletions(-) diff --git a/executor/kv/kv_executor.cpp b/executor/kv/kv_executor.cpp index 54b84a4f7..19f53600e 100644 --- a/executor/kv/kv_executor.cpp +++ b/executor/kv/kv_executor.cpp @@ -276,7 +276,8 @@ void KVExecutor::ExecuteIndexCommand(const KVRequest& request, break; } case KVRequest::QUERY_BY_INDEX: - QueryByIndex(request.index_name(), attributes, response->mutable_items()); + QueryByIndex(request.index_name(), attributes, request.with_values(), + response->mutable_items()); break; default: break; @@ -362,7 +363,7 @@ bool KVExecutor::UpdateIndexEntry( void KVExecutor::QueryByIndex(const std::string& index_name, const std::vector& attribute_prefix, - Items* items) { + bool with_values, Items* items) { if (index_name.empty()) { LOG(ERROR) << "QueryByIndex: index_name is required"; return; @@ -387,7 +388,16 @@ void KVExecutor::QueryByIndex(const std::string& index_name, } } for (const std::string& primary_key : primary_keys) { - items->add_item()->set_key(primary_key); + 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); + } } } diff --git a/executor/kv/kv_executor.h b/executor/kv/kv_executor.h index b0b78089e..93e70ed69 100644 --- a/executor/kv/kv_executor.h +++ b/executor/kv/kv_executor.h @@ -78,9 +78,10 @@ class KVExecutor : public TransactionManager { 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, - Items* items); + bool with_values, Items* items); private: // Shared by ExecuteRequest and ExecuteData. diff --git a/executor/kv/kv_executor_index_test.cpp b/executor/kv/kv_executor_index_test.cpp index 3ae59558f..88fd7cff1 100644 --- a/executor/kv/kv_executor_index_test.cpp +++ b/executor/kv/kv_executor_index_test.cpp @@ -23,6 +23,7 @@ #include #include #include +#include #include #include "chain/storage/leveldb.h" @@ -132,6 +133,21 @@ class KVExecutorIndexTest 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; @@ -358,6 +374,45 @@ TEST_P(KVExecutorIndexTest, UpdateOntoExistingEntry) { 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))); diff --git a/proto/kv/kv.proto b/proto/kv/kv.proto index b4afc8559..e1a4f2d81 100644 --- a/proto/kv/kv.proto +++ b/proto/kv/kv.proto @@ -63,6 +63,9 @@ message KVRequest { 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 { From ef4c0bf0df32b086dfb65072d6e322f0fbdc6692 Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Tue, 22 Sep 2026 15:11:02 -0700 Subject: [PATCH 05/11] interface/kv: add secondary index methods to KVClient Four methods matching the executor commands: CreateIndexEntry, DeleteIndexEntry, UpdateIndexEntry and QueryByIndex. Callers pass the index name, the attribute values and the primary key; the server builds the stored key. Unlike Set, these wait for the servers' reply, so a rejected request is visible to the caller. They return 0 on success, -3 if the servers rejected the request, or the send error from SendRequest (-1, -2). -3 is new because this is the first KV command the server can refuse; the existing codes keep their meanings. --- interface/kv/kv_client.cpp | 84 ++++++++++++++++++++++++++++++++++++++ interface/kv/kv_client.h | 23 +++++++++++ 2 files changed, 107 insertions(+) diff --git a/interface/kv/kv_client.cpp b/interface/kv/kv_client.cpp index e9075dbdd..2eb738fac 100644 --- a/interface/kv/kv_client.cpp +++ b/interface/kv/kv_client.cpp @@ -22,6 +22,21 @@ #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 +174,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) { + KVRequest request; + request.set_cmd(KVRequest::QUERY_BY_INDEX); + SetIndexFields(index_name, attribute_prefix, "", &request); + + 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 9a90d0903..d03218d45 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,25 @@ 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. + std::unique_ptr QueryByIndex( + const std::string& index_name, + const std::vector& attribute_prefix); }; } // namespace resdb From b68421d32c46448da5c4cf80edf728f2a4940396 Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Tue, 22 Sep 2026 15:11:02 -0700 Subject: [PATCH 06/11] service/tools/kv: add CLI commands for secondary indexes Adds create_index_entry, delete_index_entry, update_index_entry and query_by_index, with --index, --attrs, --new_attrs and --pk. Attributes are comma-separated, e.g. --attrs shray,jpeg. Leaving --attrs off a query returns every entry in the index. Rejected requests print "(rejected by the servers)" instead of a bare return code. --- .../tools/kv/api_tools/kv_service_tools.cpp | 89 ++++++++++++++++++- 1 file changed, 88 insertions(+), 1 deletion(-) diff --git a/service/tools/kv/api_tools/kv_service_tools.cpp b/service/tools/kv/api_tools/kv_service_tools.cpp index 75c8780d0..508c2c004 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,11 @@ 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" "\n" "More examples can be found from README.\n"); } @@ -67,8 +74,31 @@ 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'}, }; +// "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 +152,10 @@ 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; std::string client_config_file; int top = 0; // getopt_long returns int and signals the end of the options with -1. Storing @@ -174,6 +208,18 @@ 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 'h': ShowUsage(); break; @@ -288,6 +334,47 @@ 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)" : ""); + } 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)" : ""); + } else if (cmd == "query_by_index") { + if (index_name.empty()) { + ShowUsage(); + return 0; + } + auto res = client.QueryByIndex(index_name, SplitAttributes(attrs)); + if (res == nullptr) { + printf("query_by_index index = %s, attrs = %s failed\n", + index_name.c_str(), attrs.c_str()); + return 0; + } + 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()) { + printf("%s\n", item.key().c_str()); + } } else { ShowUsage(); } From ff2bb8afd49d0b9a96f774e263c7848c392edfb6 Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Tue, 22 Sep 2026 15:26:30 -0700 Subject: [PATCH 07/11] api: expose secondary index commands to Python Adds create_index_entry, delete_index_entry, update_index_entry and query_by_index to the pybind module and to kv_operation.py, wrapping the KVClient methods. Attributes can be a list, or a plain string for a single-attribute index: kv.create_index_entry("by_city", "Davis", "user:1", config) kv.query_by_index("by_city", "Davis", config) -> ["user:1", ...] kv_operation.py also looks for the module under bazel-bin/api. It only looked in bazel-out/k8-fastbuild, which exists on Linux but not on other platforms. --- api/kv_operation.py | 54 ++++++++++++++++++++++ api/pybind_kv_service.cpp | 96 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 150 insertions(+) diff --git a/api/kv_operation.py b/api/kv_operation.py index f18c84601..70651cbd3 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,55 @@ 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. + :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 no such entry exists, -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 old entry does not exist, -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)) diff --git a/api/pybind_kv_service.cpp b/api/pybind_kv_service.cpp index 013683fbd..0cb8c8226 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,100 @@ 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); +} + 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"); + // The single-string overloads are registered first so "Davis" binds to them + // rather than being read as a list of characters. + 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"); } From 1e13d8840973fb109a868d7ef8da11f79ab54395 Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Tue, 22 Sep 2026 15:28:07 -0700 Subject: [PATCH 08/11] service/tools/kv: add an end-to-end test for secondary indexes Starts a local cluster with LevelDB and runs the index commands against it: create, query by one and by several attributes, query the whole index, the three rejected cases, update, delete, and a restart to check the index is on disk. Every write is confirmed with a read, because set does not wait for consensus. Prints PASS and exits 0, or names the first check that failed and exits non-zero, so CI can run it. --- .../kv/api_tools/composite_key_e2e_test.sh | 156 ++++++++++++++++++ 1 file changed, 156 insertions(+) create mode 100755 service/tools/kv/api_tools/composite_key_e2e_test.sh 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 000000000..6ebe31e2a --- /dev/null +++ b/service/tools/kv/api_tools/composite_key_e2e_test.sh @@ -0,0 +1,156 @@ +#!/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 + +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. +query() { + local index=$1 + local attrs=${2:-} + if [ -z "$attrs" ]; then + run --cmd query_by_index --index "$index" | tail -n +2 | sort | tr '\n' ' ' + else + run --cmd query_by_index --index "$index" --attrs "$attrs" | tail -n +2 | + sort | tr '\n' ' ' + fi +} + +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" + +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 From 5e17940e8094aa6fd92a70a8ad9382950bc0fc0f Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Tue, 22 Sep 2026 15:53:27 -0700 Subject: [PATCH 09/11] service/tools/kv: fix a crash on unknown flags, and harden the e2e test long_options[] had no {NULL, 0, NULL, 0} terminator, so getopt_long read past the end of the array and any unrecognised long option crashed the tool. Adding the four index flags made that easy to hit, because the natural guesses (--index_name, --attributes, --primary_key) are all unknown names. Verified before the fix: exit 139 with no message; after it, the usage text. The index commands now also set the process exit status (0 accepted, 1 rejected or failed), so scripts can check them. End-to-end test: - a query that never reached the servers printed nothing after its header line, which looked exactly like "no matches" and made four checks pass against a dead cluster. The header is now required before the output is used. - added a check that a range scan returns no index entries, which is what the GetRange filter is for. - refuse to run outside the repository root, and stop the cluster on exit or Ctrl-C. Also corrects the reason given in a pybind comment and widens the Python docstrings, which described -3 more narrowly than it behaves. --- api/kv_operation.py | 7 +++-- api/pybind_kv_service.cpp | 4 +-- interface/kv/kv_client.cpp | 2 -- .../kv/api_tools/composite_key_e2e_test.sh | 29 +++++++++++++++++-- .../tools/kv/api_tools/kv_service_tools.cpp | 6 +++- 5 files changed, 38 insertions(+), 10 deletions(-) diff --git a/api/kv_operation.py b/api/kv_operation.py index 70651cbd3..832a45521 100644 --- a/api/kv_operation.py +++ b/api/kv_operation.py @@ -55,6 +55,7 @@ def create_index_entry(index_name: str, attributes, primary_key: str, :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. @@ -68,7 +69,8 @@ def delete_index_entry(index_name: str, attributes, primary_key: str, """ Remove a secondary index entry. The record itself is not deleted. - :return: 0 on success, -3 if no such entry exists, -1 or -2 if it could not be sent. + :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)) @@ -80,7 +82,8 @@ def update_index_entry(index_name: str, old_attributes, new_attributes, primary_ 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 old entry does not exist, -1 or -2 if it could not be sent. + :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)) diff --git a/api/pybind_kv_service.cpp b/api/pybind_kv_service.cpp index 0cb8c8226..235c349bc 100644 --- a/api/pybind_kv_service.cpp +++ b/api/pybind_kv_service.cpp @@ -141,8 +141,8 @@ std::vector query_by_index_one(std::string index_name, 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"); - // The single-string overloads are registered first so "Davis" binds to them - // rather than being read as a list of characters. + // 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, diff --git a/interface/kv/kv_client.cpp b/interface/kv/kv_client.cpp index 2eb738fac..65c22c0e0 100644 --- a/interface/kv/kv_client.cpp +++ b/interface/kv/kv_client.cpp @@ -37,7 +37,6 @@ void SetIndexFields(const std::string& index_name, } // namespace - KVClient::KVClient(const ResDBConfig& config) : TransactionConstructor(config) {} @@ -174,7 +173,6 @@ 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) { diff --git a/service/tools/kv/api_tools/composite_key_e2e_test.sh b/service/tools/kv/api_tools/composite_key_e2e_test.sh index 6ebe31e2a..83d6f07a7 100755 --- a/service/tools/kv/api_tools/composite_key_e2e_test.sh +++ b/service/tools/kv/api_tools/composite_key_e2e_test.sh @@ -27,6 +27,14 @@ 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 @@ -37,15 +45,25 @@ run() { } # 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 - run --cmd query_by_index --index "$index" | tail -n +2 | sort | tr '\n' ' ' + out=$(run --cmd query_by_index --index "$index") else - run --cmd query_by_index --index "$index" --attrs "$attrs" | tail -n +2 | - sort | tr '\n' ' ' + 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() { @@ -136,6 +154,11 @@ check "the record itself is untouched" "$(run --cmd get --key photo: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 diff --git a/service/tools/kv/api_tools/kv_service_tools.cpp b/service/tools/kv/api_tools/kv_service_tools.cpp index 508c2c004..f63e5d5e9 100644 --- a/service/tools/kv/api_tools/kv_service_tools.cpp +++ b/service/tools/kv/api_tools/kv_service_tools.cpp @@ -78,6 +78,8 @@ static struct option long_options[] = { {"attrs", required_argument, NULL, 'a'}, {"new_attrs", required_argument, NULL, 'n'}, {"pk", required_argument, NULL, 'p'}, + // getopt_long reads until it finds this terminator. + {NULL, 0, NULL, 0}, }; // "shray,jpeg" -> {"shray", "jpeg"}; "" -> {} (meaning the whole index). @@ -347,6 +349,7 @@ int main(int argc, char** argv) { 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(); @@ -359,6 +362,7 @@ int main(int argc, char** argv) { "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(); @@ -368,7 +372,7 @@ int main(int argc, char** argv) { if (res == nullptr) { printf("query_by_index index = %s, attrs = %s failed\n", index_name.c_str(), attrs.c_str()); - return 0; + return 1; } printf("query_by_index index = %s, attrs = %s, %d result(s)\n", index_name.c_str(), attrs.c_str(), res->item_size()); From 121e751ab9f923054b51581d0f2033fc16e27a30 Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Thu, 24 Sep 2026 14:43:22 -0700 Subject: [PATCH 10/11] interface/kv, api: let clients ask for record values Adds with_values to KVClient::QueryByIndex, a --with_values flag to the CLI, and query_records_by_index to the Python module, which returns (key, value) pairs. The server does the reading, so an application that wants the matching records gets them in one request instead of one request per match. --- api/kv_operation.py | 13 ++++++++++ api/pybind_kv_service.cpp | 26 +++++++++++++++++++ interface/kv/kv_client.cpp | 3 ++- interface/kv/kv_client.h | 4 ++- .../tools/kv/api_tools/kv_service_tools.cpp | 16 ++++++++++-- 5 files changed, 58 insertions(+), 4 deletions(-) diff --git a/api/kv_operation.py b/api/kv_operation.py index 832a45521..11da991a1 100644 --- a/api/kv_operation.py +++ b/api/kv_operation.py @@ -101,3 +101,16 @@ def query_by_index(index_name: str, attributes, :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 235c349bc..cf305c537 100644 --- a/api/pybind_kv_service.cpp +++ b/api/pybind_kv_service.cpp @@ -138,6 +138,28 @@ std::vector query_by_index_one(std::string index_name, 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"); @@ -159,4 +181,8 @@ PYBIND11_MODULE(pybind_kv, m) { "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/interface/kv/kv_client.cpp b/interface/kv/kv_client.cpp index 65c22c0e0..ef44d3f1b 100644 --- a/interface/kv/kv_client.cpp +++ b/interface/kv/kv_client.cpp @@ -227,10 +227,11 @@ int KVClient::UpdateIndexEntry(const std::string& index_name, std::unique_ptr KVClient::QueryByIndex( const std::string& index_name, - const std::vector& attribute_prefix) { + 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); diff --git a/interface/kv/kv_client.h b/interface/kv/kv_client.h index d03218d45..8cdd6fac3 100644 --- a/interface/kv/kv_client.h +++ b/interface/kv/kv_client.h @@ -83,9 +83,11 @@ class KVClient : public TransactionConstructor { 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); + const std::vector& attribute_prefix, + bool with_values = false); }; } // namespace resdb diff --git a/service/tools/kv/api_tools/kv_service_tools.cpp b/service/tools/kv/api_tools/kv_service_tools.cpp index f63e5d5e9..3931c74e1 100644 --- a/service/tools/kv/api_tools/kv_service_tools.cpp +++ b/service/tools/kv/api_tools/kv_service_tools.cpp @@ -57,6 +57,7 @@ void ShowUsage() { "--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"); } @@ -78,6 +79,7 @@ static struct option long_options[] = { {"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}, }; @@ -158,6 +160,7 @@ int main(int argc, char** argv) { 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 @@ -222,6 +225,9 @@ int main(int argc, char** argv) { case 'p': pk = optarg; break; + case 'w': + with_values = true; + break; case 'h': ShowUsage(); break; @@ -368,7 +374,8 @@ int main(int argc, char** argv) { ShowUsage(); return 0; } - auto res = client.QueryByIndex(index_name, SplitAttributes(attrs)); + 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()); @@ -377,7 +384,12 @@ int main(int argc, char** argv) { 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()) { - printf("%s\n", item.key().c_str()); + 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(); From 1644bd91b359c6d0cfd8b8bbfc3fb4761a039e47 Mon Sep 17 00:00:00 2001 From: Shray Arora Date: Thu, 24 Sep 2026 00:31:56 -0700 Subject: [PATCH 11/11] benchmark/e2e: add an end-to-end benchmark for secondary indexes Measures four ways to answer "which records have city = Davis?" against a running cluster: scan every record and filter in the application, query the index for keys only, query then fetch each match, and query with --with_values. Takes a client config path, so the same binary measures a local cluster or a multi-machine deployment. --records, --selectivity and --value_size set the dataset; --threads loads it with several clients, because the proxy batches 100 requests or 100ms and a single client pays that wait on every request. The four approaches must agree on the match count or the run fails, which already caught two mistakes in the benchmark itself. --- benchmark/e2e/BUILD | 29 ++ benchmark/e2e/composite_key_e2e_benchmark.cpp | 328 ++++++++++++++++++ 2 files changed, 357 insertions(+) create mode 100644 benchmark/e2e/BUILD create mode 100644 benchmark/e2e/composite_key_e2e_benchmark.cpp diff --git a/benchmark/e2e/BUILD b/benchmark/e2e/BUILD new file mode 100644 index 000000000..cc29b3ac4 --- /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 000000000..70af72109 --- /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; +}