Enhancement: Implement for NFS and Azure storages (#2172)

#### Reference Issues/PRs
Expose `Library.get_key_path` for NFS and Azure storages.

#### What does this implement or fix?
get_key_path will be used to determine where each key is located. This
can help to set hard quotas for the platforms that support path based
quotas, like VAST.

#### Any other comments?

#### Checklist

<details>
  <summary>
   Checklist for code changes...
  </summary>
 
- [ ] Have you updated the relevant docstrings, documentation and
copyright notice?
- [ ] Is this contribution tested against [all ArcticDB's
features](../docs/mkdocs/docs/technical/contributing.md)?
- [ ] Do all exceptions introduced raise appropriate [error
messages](https://docs.arcticdb.io/error_messages/)?
 - [x] Are API changes highlighted in the PR description?
- [x] Is the PR labelled as enhancement or bug so it appears in
autogenerated release notes?
</details>

<!--
Thanks for contributing a Pull Request to ArcticDB! Please ensure you
have taken a look at:
- ArcticDB's Code of Conduct:
https://github.com/man-group/ArcticDB/blob/master/CODE_OF_CONDUCT.md
- ArcticDB's Contribution Licensing:
https://github.com/man-group/ArcticDB/blob/master/docs/mkdocs/docs/technical/contributing.md#contribution-licensing
-->

---------

Co-authored-by: Maxim Morozov <[email protected]>

cpp/arcticdb/storage/azure/azure_storage.cpp

@@ -391,6 +391,12 @@ bool AzureStorage::do_key_exists(const VariantKey& key) {
     return detail::do_key_exists_impl(key, root_folder_, *azure_client_);
 }
 
+std::string AzureStorage::do_key_path(const VariantKey& key) const {
+    auto b = FlatBucketizer{};
+    auto key_type_dir = key_type_folder(root_folder_, variant_key_type(key));
+    return object_path(b.bucketize(key_type_dir, key), key);
+}
+
 } // namespace azure
 
 } // namespace arcticdb::storage

cpp/arcticdb/storage/azure/azure_storage.hpp

@@ -67,7 +67,7 @@ class AzureStorage final : public Storage {
         return false;
     }
 
-    std::string do_key_path(const VariantKey&) const final { return {}; };
+    std::string do_key_path(const VariantKey&) const final;
 
   private:
     std::unique_ptr<AzureClientWrapper> azure_client_;

cpp/arcticdb/storage/s3/nfs_backed_storage.cpp

@@ -154,6 +154,12 @@ std::string NfsBackedStorage::name() const {
     return fmt::format("nfs_backed_storage-{}/{}/{}", region_, bucket_name_, root_folder_);
 }
 
+std::string NfsBackedStorage::do_key_path(const VariantKey& key) const {
+    auto b = NfsBucketizer{};
+    auto key_type_dir = key_type_folder(root_folder_, variant_key_type(key));
+    return object_path(b.bucketize(key_type_dir, key), key);
+}
+
 void NfsBackedStorage::do_write(KeySegmentPair& key_seg) {
     auto enc = KeySegmentPair{encode_object_id(key_seg.variant_key()), key_seg.segment_ptr()};
     s3::detail::do_write_impl(enc, root_folder_, bucket_name_, *s3_client_, NfsBucketizer{});

cpp/arcticdb/storage/s3/nfs_backed_storage.hpp

@@ -65,7 +65,7 @@ class NfsBackedStorage final : public Storage {
         return false;
     }
 
-    std::string do_key_path(const VariantKey&) const final { return {}; };
+    std::string do_key_path(const VariantKey&) const final;
 
     auto& client() { return s3_client_; }
     const std::string& bucket_name() const { return bucket_name_; }

cpp/arcticdb/storage/test/test_azure_storage.cpp

@@ -84,3 +84,17 @@ TEST_F(AzureMockStorageFixture, test_matching_key_type_prefix_list) {
 
     ASSERT_EQ(list_in_store(store, entity::KeyType::LOG), log_symbols);
 }
+
+TEST_F(AzureMockStorageFixture, test_key_path) {
+    std::vector<VariantKey> res;
+
+    store.iterate_type(KeyType::TABLE_DATA, [&](VariantKey &&found_key) {
+        res.emplace_back(found_key);
+    }, "");
+
+    for(auto vk: res) {
+        auto key_path = store.key_path(vk);
+        ASSERT_TRUE(key_path.size() > 0);
+        ASSERT_TRUE(key_path.starts_with(store.library_path().to_delim_path('/')));
+    }
+}

cpp/arcticdb/storage/test/test_s3_storage.cpp

@@ -300,8 +300,24 @@ TEST_F(S3StorageFixture, test_key_exists) {
     ASSERT_TRUE(exists_in_store(store, "symbol"));
     ASSERT_FALSE(exists_in_store(store, "symbol-not-present"));
     ASSERT_THROW(
-        exists_in_store(store, MockS3Client::get_failure_trigger("symbol", StorageOperation::EXISTS, Aws::S3::S3Errors::NETWORK_CONNECTION, false)),
-        UnexpectedS3ErrorException);
+            exists_in_store(store, MockS3Client::get_failure_trigger("symbol", StorageOperation::EXISTS,
+                                                                     Aws::S3::S3Errors::NETWORK_CONNECTION, false)),
+            UnexpectedS3ErrorException);
+}
+
+TEST_P(S3AndNfsStorageFixture, test_key_path) {
+    std::vector<VariantKey> res;
+
+    auto store = get_storage();
+    store->iterate_type(KeyType::TABLE_DATA, [&](VariantKey &&found_key) {
+        res.emplace_back(found_key);
+    }, "");
+
+    for(auto vk : res) {
+        auto key_path = store->key_path(vk);
+        ASSERT_TRUE(key_path.size() > 0);
+        ASSERT_TRUE(key_path.starts_with(get_root_folder(store->library_path())));
+    }
 }
 
 TEST_F(S3StorageFixture, test_read){

python/tests/integration/arcticdb/test_s3.py

@@ -18,6 +18,7 @@
 from arcticdb_ext.storage import NoDataFoundException
 
 from arcticdb_ext import set_config_string
+from arcticdb_ext.storage import KeyType
 from arcticdb.util.test import create_df, assert_frame_equal
 
 from arcticdb.storage_fixtures.s3 import MotoNfsBackedS3StorageFixtureFactory
@@ -190,3 +191,40 @@ def test_wrapped_s3_storage(lib_name, wrapped_s3_storage_bucket):
     # There should be no problems after the failure simulation has been turned off
     lib.read("s")
     lib.write("s", data=create_df())
+
+
+@pytest.fixture(scope="session")
+def test_prefix():
+    return "test_bucket_prefix"
+
+
+@pytest.fixture(scope="function",
+                params=[MotoNfsBackedS3StorageFixtureFactory,
+                        MotoS3StorageFixtureFactory])
+def storage_bucket(test_prefix, request):
+    with request.param(
+            use_ssl=False,
+            ssl_test_support=False,
+            bucket_versioning=False,
+            default_prefix=test_prefix
+    ) as factory:
+        with factory.create_fixture() as bucket:
+            yield bucket
+
+
+def test_library_get_key_path(lib_name, storage_bucket, test_prefix):
+    lib = storage_bucket.create_version_store_factory(lib_name)()
+    lib.write("s", data=create_df())
+    lib_tool = lib.library_tool()
+
+    keys_count = 0
+    for key_type in KeyType.__members__.values():
+        keys = lib_tool.find_keys(key_type)
+        keys_count += len(keys)
+        for key in keys:
+            path = lib_tool.get_key_path(key)
+            assert path != ""
+            assert path.startswith(test_prefix)
+
+    assert keys_count > 0
+