Repository: man-group/ArcticDB (language: python)

Issue:
KeyType::LOG and KeyType::LOG_COMPACTED share the same prefix in S3 and Azure
`KeyType::LOG`'s long name is `log` and `KeyType::LOG_COMPACTED`'s long name is `logc`.
And currently there is no "end delimiter" for key prefix. So if user is trying to iterate `KeyType::LOG`, the prefix for iteration will be `.../log`. As prefix of keys `KeyType::LOG_COMPACTED` is `.../logc`, the iteration will return both keys.

The issue currently has no impact but could be hazardous for future development.


Discussion / hints:

Fix patch:
diff --git a/cpp/arcticdb/storage/azure/azure_storage.cpp b/cpp/arcticdb/storage/azure/azure_storage.cpp
index 8b381c34a4..498cbe351d 100644
--- a/cpp/arcticdb/storage/azure/azure_storage.cpp
+++ b/cpp/arcticdb/storage/azure/azure_storage.cpp
@@ -229,32 +229,33 @@ void do_remove_impl(Composite<VariantKey>&& ks,
         }
 }
 
-auto default_prefix_handler() {
-    return [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor& key_descriptor, KeyType) {
-        return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
-    };
+std::string prefix_handler(const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor& key_descriptor, KeyType) {
+    return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
 }
 
-template<class KeyBucketizer, class PrefixHandler>
+template<class KeyBucketizer>
 void do_iterate_type_impl(KeyType key_type,
     const IterateTypeVisitor& visitor,
     const std::string& root_folder,
     AzureClientWrapper& azure_client,
     KeyBucketizer&& bucketizer,
-    PrefixHandler&& prefix_handler = default_prefix_handler(),
     const std::string& prefix = std::string{}) {
         ARCTICDB_SAMPLE(AzureStorageIterateType, 0)
         auto key_type_dir = key_type_folder(root_folder, key_type);
+        const auto path_to_key_size = key_type_dir.size() + 1 + bucketizer.bucketize_length(key_type);
+        // if prefix is empty, add / to avoid matching both log and logc when key_type_dir is {root_folder}/log
+        if (prefix.empty()) {
+            key_type_dir += "/";
+        }
 
         KeyDescriptor key_descriptor(prefix,
             is_ref_key_class(key_type) ? IndexDescriptor::UNKNOWN : IndexDescriptor::TIMESTAMP, FormatType::TOKENIZED);
         auto key_prefix = prefix_handler(prefix, key_type_dir, key_descriptor, key_type);
-        const auto root_folder_size = key_type_dir.size() + 1 + bucketizer.bucketize_length(key_type);
 
         try {
             for (auto page = azure_client.list_blobs(key_prefix); page.HasPage(); page.MoveToNextPage()) {
                 for (const auto& blob : page.Blobs) {
-                    auto key = blob.Name.substr(root_folder_size);
+                    auto key = blob.Name.substr(path_to_key_size);
                     ARCTICDB_TRACE(log::version(), "Got object_list: {}, key: {}", blob.Name, key);
                     auto k = variant_key_from_bytes(
                                 reinterpret_cast<uint8_t*>(key.data()),
@@ -316,11 +317,7 @@ void AzureStorage::do_remove(Composite<VariantKey>&& ks, RemoveOpts) {
 }
 
 void AzureStorage::do_iterate_type(KeyType key_type, const IterateTypeVisitor& visitor, const std::string &prefix) {
-    auto prefix_handler = [] (const std::string& prefix, const std::string& key_type_dir, const KeyDescriptor key_descriptor, KeyType) {
-        return !prefix.empty() ? fmt::format("{}/{}*{}", key_type_dir, key_descriptor, prefix) : key_type_dir;
-    };
-
-    detail::do_iterate_type_impl(key_type, visitor, root_folder_, *azure_client_, FlatBucketizer{}, std::move(prefix_handler), prefix);
+    detail::do_iterate_type_impl(key_type, visitor, root_folder_, *azure_client_, FlatBucketizer{}, prefix);
 }
 
 bool AzureStorage::do_key_exists(const VariantKey& key) {
diff --git a/cpp/arcticdb/storage/s3/detail-inl.hpp b/cpp/arcticdb/storage/s3/detail-inl.hpp
index 04881c2e62..228f886229 100644
--- a/cpp/arcticdb/storage/s3/detail-inl.hpp
+++ b/cpp/arcticdb/storage/s3/detail-inl.hpp
@@ -284,6 +284,11 @@ namespace s3 {
         ) {
             ARCTICDB_SAMPLE(S3StorageIterateType, 0)
             auto key_type_dir = key_type_folder(root_folder, key_type);
+            const auto path_to_key_size = key_type_dir.size() + 1 + bucketizer.bucketize_length(key_type);
+            // if prefix is empty, add / to avoid matching both log and logc when key_type_dir is {root_folder}/log
+            if (prefix.empty()) {
+                key_type_dir += "/";
+            }
 
             // Generally we get the key descriptor from the AtomKey, but in the case of iterating version journals
             // where we want to have a narrower prefix, we can use the info that it's a version journal and derive
@@ -304,11 +309,10 @@ namespace s3 {
                 if (list_objects_result.is_success()) {
                     auto& output = list_objects_result.get_output();
 
-                    const auto root_folder_size = key_type_dir.size() + 1 + bucketizer.bucketize_length(key_type);
                     ARCTICDB_RUNTIME_DEBUG(log::storage(), "Received object list");
 
                     for (auto& s3_object_name: output.s3_object_names) {
-                        auto key = s3_object_name.substr(root_folder_size);
+                        auto key = s3_object_name.substr(path_to_key_size);
                         ARCTICDB_TRACE(log::version(), "Got object_list: {}, key: {}", s3_object_name, key);
                         auto k = variant_key_from_bytes(
                                 reinterpret_cast<uint8_t *>(key.data()),


Test patch:
diff --git a/cpp/arcticdb/storage/test/common.hpp b/cpp/arcticdb/storage/test/common.hpp
index ac94e0bee5..907cfd9d05 100644
--- a/cpp/arcticdb/storage/test/common.hpp
+++ b/cpp/arcticdb/storage/test/common.hpp
@@ -14,9 +14,9 @@
 
 namespace arcticdb::storage {
 
-inline VariantKey get_test_key(std::string name) {
+inline VariantKey get_test_key(std::string name, entity::KeyType key_type = entity::KeyType::TABLE_DATA) {
     auto builder = arcticdb::atom_key_builder();
-    return builder.build<arcticdb::entity::KeyType::TABLE_DATA>(name);
+    return builder.build(name, key_type);
 }
 
 inline Segment get_test_segment() {
@@ -25,44 +25,55 @@ inline Segment get_test_segment() {
     return encode_dispatch(std::move(segment_in_memory), codec_opts, arcticdb::EncodingVersion::V2);
 }
 
-inline void write_in_store(Storage &store, std::string symbol) {
-    auto variant_key = get_test_key(symbol);
+inline void write_in_store(Storage &store, std::string symbol, entity::KeyType key_type = entity::KeyType::TABLE_DATA) {
+    auto variant_key = get_test_key(symbol, key_type);
     store.write(KeySegmentPair(std::move(variant_key), get_test_segment()));
 }
 
-inline void update_in_store(Storage &store, std::string symbol) {
-    auto variant_key = get_test_key(symbol);
+inline void update_in_store(Storage &store, std::string symbol, entity::KeyType key_type = entity::KeyType::TABLE_DATA) {
+    auto variant_key = get_test_key(symbol, key_type);
     store.update(KeySegmentPair(std::move(variant_key), get_test_segment()), arcticdb::storage::UpdateOpts{});
 }
 
-inline bool exists_in_store(Storage &store, std::string symbol) {
-    auto variant_key = get_test_key(symbol);
+inline bool exists_in_store(Storage &store, std::string symbol, entity::KeyType key_type = entity::KeyType::TABLE_DATA) {
+    auto variant_key = get_test_key(symbol, key_type);
     return store.key_exists(variant_key);
 }
 
-inline std::string read_in_store(Storage &store, std::string symbol) {
-    auto variant_key = get_test_key(symbol);
+inline std::string read_in_store(Storage &store, std::string symbol, entity::KeyType key_type = entity::KeyType::TABLE_DATA) {
+    auto variant_key = get_test_key(symbol, key_type);
     auto opts = ReadKeyOpts{};
     auto result = store.read(std::move(variant_key), opts);
     return std::get<StringId>(result.atom_key().id());
 }
 
-inline void remove_in_store(Storage &store, std::vector<std::string> symbols) {
+inline void remove_in_store(Storage &store, std::vector<std::string> symbols, entity::KeyType key_type = entity::KeyType::TABLE_DATA) {
     auto to_remove = std::vector<VariantKey>();
     for (auto &symbol: symbols) {
-        to_remove.emplace_back(get_test_key(symbol));
+        to_remove.emplace_back(get_test_key(symbol, key_type));
     }
     auto opts = RemoveOpts();
     store.remove(Composite(std::move(to_remove)), opts);
 }
 
-inline std::set<std::string> list_in_store(Storage &store) {
+inline std::set<std::string> list_in_store(Storage &store, entity::KeyType key_type = entity::KeyType::TABLE_DATA) {
     auto keys = std::set<std::string>();
-    store.iterate_type(KeyType::TABLE_DATA, [&keys](VariantKey &&key) {
+    store.iterate_type(key_type, [&keys](VariantKey &&key) {
         auto atom_key = std::get<AtomKey>(key);
         keys.emplace(std::get<StringId>(atom_key.id()));
     });
     return keys;
 }
 
+inline std::set<std::string> populate_store(Storage &store, std::string_view symbol_prefix, int start, int end,
+                                            entity::KeyType key_type = entity::KeyType::TABLE_DATA) {
+    auto symbols = std::set<std::string>();
+    for (int i = start; i < end; ++i) {
+        auto symbol = fmt::format("{}_{}", symbol_prefix, i);
+        write_in_store(store, symbol, key_type);
+        symbols.emplace(symbol);
+    }
+    return symbols;
+}
+
 }
diff --git a/cpp/arcticdb/storage/test/test_azure_storage.cpp b/cpp/arcticdb/storage/test/test_azure_storage.cpp
index c4896e1d4a..92e56ee2a6 100644
--- a/cpp/arcticdb/storage/test/test_azure_storage.cpp
+++ b/cpp/arcticdb/storage/test/test_azure_storage.cpp
@@ -74,3 +74,13 @@ TEST_F(AzureMockStorageFixture, test_list) {
 
     ASSERT_EQ(list_in_store(store), symbols);
 }
+
+TEST_F(AzureMockStorageFixture, test_matching_key_type_prefix_list) {
+    auto log_symbols = populate_store(store, "symbol_log", 0, 5, entity::KeyType::LOG);
+    ASSERT_EQ(list_in_store(store, entity::KeyType::LOG), log_symbols);
+
+    auto log_compacted_symbols = populate_store(store, "symbol_logc", 0, 5, entity::KeyType::LOG_COMPACTED);
+    ASSERT_EQ(list_in_store(store, entity::KeyType::LOG_COMPACTED), log_compacted_symbols);
+
+    ASSERT_EQ(list_in_store(store, entity::KeyType::LOG), log_symbols);
+}
diff --git a/cpp/arcticdb/storage/test/test_s3_storage.cpp b/cpp/arcticdb/storage/test/test_s3_storage.cpp
index 8a0c9e3f59..4365f4c748 100644
--- a/cpp/arcticdb/storage/test/test_s3_storage.cpp
+++ b/cpp/arcticdb/storage/test/test_s3_storage.cpp
@@ -289,3 +289,13 @@ TEST_F(S3StorageFixture, test_list) {
 
     ASSERT_THROW(list_in_store(store), UnexpectedS3ErrorException);
 }
+
+TEST_F(S3StorageFixture, test_matching_key_type_prefix_list) {
+    auto log_symbols = populate_store(store, "symbol_log", 0, 5, entity::KeyType::LOG);
+    ASSERT_EQ(list_in_store(store, entity::KeyType::LOG), log_symbols);
+
+    auto log_compacted_symbols = populate_store(store, "symbol_logc", 0, 5, entity::KeyType::LOG_COMPACTED);
+    ASSERT_EQ(list_in_store(store, entity::KeyType::LOG_COMPACTED), log_compacted_symbols);
+
+    ASSERT_EQ(list_in_store(store, entity::KeyType::LOG), log_symbols);
+}