storage named collection in metastore (#1022)

Co-authored-by: Alan Yu <alan.yu@timeplus.io>

Author: yokofly

programs/server/Server.cpp

@@ -833,8 +833,6 @@ try
         server_settings.max_io_thread_pool_free_size,
         server_settings.io_thread_pool_queue_size);
 
-    NamedCollectionFactory::instance().loadIfNot();
-
     /// Initialize global local cache for remote filesystem.
     if (config().has("local_cache_for_remote_fs"))
     {
@@ -1206,6 +1204,10 @@ try
 
     auto server_descriptor = global_context->getServerDescriptor();
 
+    /// Start named collection after bootstrap
+    NamedCollectionFactory::instance().loadIfNot();
+    /// proton: ends.
+
     for (auto & server : servers_to_start_before_tables)
     {
         server.start();

src/Cluster/Common/EncodeMetaKey.cpp

@@ -224,6 +224,29 @@ std::string encodeMetaTaskKey(const std::string & ns, const std::string & name)
     return key;
 }
 
+std::string encodeMetaNamedCollectionKey()
+{
+    std::string encoded_key;
+    DB::PrefixTreeEncode::encodeVarUIntAscending(std::to_underlying(MetaKeySpace::NamedCollection), encoded_key);
+    return encoded_key;
+}
+
+std::string encodeMetaNamedCollectionKey(const std::string & name)
+{
+    std::string encoded_key;
+    encoded_key.reserve(1 + name.size());
+    DB::PrefixTreeEncode::encodeVarUIntAscending(std::to_underlying(MetaKeySpace::NamedCollection), encoded_key);
+    DB::PrefixTreeEncode::encodeStringAscending(name, encoded_key);
+    return encoded_key;
+}
+
+std::string decodeMetaNamedCollectionKey(std::string_view data)
+{
+    DB::PrefixTreeEncode::decodeVarUIntAscending(data);
+    std::string decoded;
+    auto result = DB::PrefixTreeEncode::decodeStringAscending(data, decoded);
+    return std::string{result.data(), result.size()};
+}
 
 namespace
 {

src/Cluster/Common/EncodeMetaKey.h

@@ -48,6 +48,10 @@ std::string encodeMetaTaskKey(const std::string & ns, const std::string & name);
 
 std::string encodeMetaAppliedSequenceKey(const StreamIDShard & stream_shard);
 
+std::string encodeMetaNamedCollectionKey();
+std::string encodeMetaNamedCollectionKey(const std::string & name);
+std::string decodeMetaNamedCollectionKey(std::string_view data);
+
 std::string encodeMetaPendingRequestKey(uint64_t sequence_number);
 
 }

src/Cluster/Common/MetaKeySpace.h

@@ -12,6 +12,7 @@ enum class MetaKeySpace : uint8_t
     FormatSchema = 0xfa,
     Alert = 0xf9,
     Task = 0xf8,
+    NamedCollection = 0xf7,
     Database = 0xf4,
     Disk = 0xf3,
     StoragePolicy = 0xf2,

src/Cluster/MetaStore/MetaDB.cpp

@@ -471,6 +471,78 @@ CallResultV<protocol::TaskDescriptorPtrs> MetaDB::listTasks(std::vector<std::str
     return getValues<protocol::TaskDescriptor>(encodeMetaTaskKey(), MetaKeySpace::Task, corrupted_keys);
 }
 
+Error MetaDB::saveNamedCollection(
+    const std::string & name, const protocol::NamedCollectionDescriptor & named_collection, AppliedSequence applied_sn)
+{
+    return mergeKeyValue(
+        encodeMetaNamedCollectionKey(name),
+        cluster::serialize<std::string>(named_collection, /*version=*/1),
+        applied_sn,
+        "named_collection");
+}
+
+Error MetaDB::deleteNamedCollection(const std::string & name, AppliedSequence applied_sn)
+{
+    return deleteKey(encodeMetaNamedCollectionKey(name), applied_sn, "named_collection");
+}
+
+Error MetaDB::deleteNamedCollection(const std::string & name)
+{
+    return deleteKey(encodeMetaNamedCollectionKey(name), std::nullopt, "named_collection");
+}
+
+CallResultV<protocol::NamedCollectionDescriptorPtr> MetaDB::getNamedCollection(const std::string & name) const
+{
+    return getValue<protocol::NamedCollectionDescriptor>(encodeMetaNamedCollectionKey(name), MetaKeySpace::NamedCollection);
+}
+
+CallResultV<protocol::NamedCollectionDescriptorPtrs> MetaDB::getNamedCollection(const std::string & name, size_t versions_requested) const
+{
+    return getValue<protocol::NamedCollectionDescriptor>(
+        encodeMetaNamedCollectionKey(name), MetaKeySpace::NamedCollection, versions_requested);
+}
+
+CallResultV<std::vector<std::string>> MetaDB::listNamedCollections() const
+{
+    std::string prefix_start = encodeMetaNamedCollectionKey();
+    std::string prefix_end = prefix_start;
+    DB::PrefixTreeEncode::keyPrefixEnd(prefix_end);
+
+    rocksdb::Slice prefix_start_slice = prefix_start;
+    rocksdb::Slice prefix_end_slice = prefix_end;
+
+    rocksdb::ReadOptions options;
+    options.auto_prefix_mode = true;
+    options.iterate_upper_bound = &prefix_end_slice;
+
+    std::unique_ptr<rocksdb::Iterator> iter{meta_db->NewIterator(options, default_cf_handle)};
+
+    CallResultV<std::vector<std::string>> result;
+
+    Error last_err;
+
+    iter->Seek(prefix_start_slice);
+    while (iter->Valid())
+    {
+        try
+        {
+            result.result.push_back(decodeMetaNamedCollectionKey(iter->key().ToStringView()));
+        }
+        catch (const Poco::Exception & e)
+        {
+            auto text = fmt::format("Failed to decode value: key={} error={}", iter->key().ToStringView(), e.displayText());
+            last_err = Error(e.code(), text);
+            LOG_ERROR(logger, "{}, will skip it", text);
+        }
+        iter->Next();
+    }
+
+    if (result.result.empty() && last_err.hasError())
+        result.err.swap(last_err);
+
+    return result;
+}
+
 Error MetaDB::deleteMaterializedViewAssignment(const cluster::Stream & mv)
 {
     return deleteKey(encodeMetaMaterializedViewAssignmentKey(mv.ns, mv.name), {}, "MaterializedViewAssignment");

src/Cluster/MetaStore/MetaDB.h

@@ -9,6 +9,7 @@
 #include <Cluster/Protocol/DiskDescriptor.h>
 #include <Cluster/Protocol/FormatSchemaDescriptor.h>
 #include <Cluster/Protocol/MaterializedViewAssignment.h>
+#include <Cluster/Protocol/NamedCollectionDescriptor.h>
 #include <Cluster/Protocol/StoragePolicyDescriptor.h>
 #include <Cluster/Protocol/StreamDescriptor.h>
 #include <Cluster/Protocol/TaskDescriptor.h>
@@ -31,10 +32,7 @@ class MetaDB final
 {
 public:
     MetaDB(
-        const std::filesystem::path & meta_dir,
-        const StreamIDShard & meta_stream_shard_,
-        size_t metadata_keep_versions,
-        LoggerPtr logger_);
+        const std::filesystem::path & meta_dir, const StreamIDShard & meta_stream_shard_, size_t metadata_keep_versions, LoggerPtr logger_);
 
     ~MetaDB();
 
@@ -125,6 +123,15 @@ class MetaDB final
     CallResultV<protocol::TaskDescriptorPtrs> listTasks(const std::string & ns) const;
     CallResultV<protocol::TaskDescriptorPtrs> listTasks(std::vector<std::string> * corrupted_keys = nullptr) const;
 
+    /// Named Collection CRUD
+    Error
+    saveNamedCollection(const std::string & name, const protocol::NamedCollectionDescriptor & named_collection, AppliedSequence applied_sn);
+    Error deleteNamedCollection(const std::string & name, AppliedSequence applied_sn);
+    Error deleteNamedCollection(const std::string & name);
+    CallResultV<protocol::NamedCollectionDescriptorPtr> getNamedCollection(const std::string & name) const;
+    CallResultV<protocol::NamedCollectionDescriptorPtrs> getNamedCollection(const std::string & name, size_t versions_requested) const;
+    CallResultV<std::vector<std::string>> listNamedCollections() const;
+
     /// Internal API, save / delete mv assignment from local metadb
     /// Materialized View Assignments
     Error saveMaterializedViewAssignment(const protocol::MaterializedViewAssignment & assignment, AppliedSequence applied_sn);
@@ -142,13 +149,13 @@ class MetaDB final
     /// Pending request management
     /// Save a pending metadata request with sync write
     Error savePendingRequest(uint64_t sequence_number, const std::string & serialized_data);
-    
+
     /// Delete a pending request after successful apply
     Error deletePendingRequest(uint64_t sequence_number);
-    
+
     /// Iterate pending requests after given sequence number for recovery
     CallResultV<std::vector<std::pair<uint64_t, std::string>>> iteratePendingRequestsAfter(uint64_t after_sn) const;
-    
+
     /// Cleanup old pending requests up to and including the given applied sequence number
     Error cleanupOldPendingRequests(uint64_t applied_sn);
 

src/Cluster/MetaStore/MetaStore.cpp

@@ -507,6 +507,37 @@ GetTaskResponsePtr MetaStore::getTask(GetTaskRequestPtr req) const
         return std::make_shared<GetTaskResponse>(std::move(result.err), /*data_version_=*/req->version());
 }
 
+CreateNamedCollectionResponsePtr MetaStore::createNamedCollection(CreateNamedCollectionRequestPtr req)
+{
+    return processMetaRequestSync<CreateNamedCollectionRequest, CreateNamedCollectionResponse, /*resp_desc=*/false>(std::move(req));
+}
+
+DeleteNamedCollectionResponsePtr MetaStore::deleteNamedCollection(DeleteNamedCollectionRequestPtr req)
+{
+    return processMetaRequestSync<DeleteNamedCollectionRequest, DeleteNamedCollectionResponse, /*resp_desc=*/false>(std::move(req));
+}
+
+ListNamedCollectionsResponsePtr MetaStore::listNamedCollections(ListNamedCollectionsRequestPtr req) const
+{
+    auto result = meta_db->listNamedCollections();
+    if (!result.hasError())
+        return std::make_shared<ListNamedCollectionsResponse>(
+            std::move(result.result), nodeID(), /*sn_=*/0, /*data_version_=*/req->version());
+
+    return std::make_shared<ListNamedCollectionsResponse>(std::move(result.err), /*data_version_=*/req->version());
+}
+
+GetNamedCollectionResponsePtr MetaStore::getNamedCollection(GetNamedCollectionRequestPtr req) const
+{
+    const auto & req_data = req->data();
+    auto result = meta_db->getNamedCollection(req_data.collection_name, req_data.versions_requested);
+    if (!result.hasError())
+        return std::make_shared<GetNamedCollectionResponse>(
+            std::move(result.result), nodeID(), /*sn_=*/0, /*data_version_=*/req->version());
+    else
+        return std::make_shared<GetNamedCollectionResponse>(std::move(result.err), /*data_version_=*/req->version());
+}
+
 Error MetaStore::proposeMetaRequest(cluster::RequestPtr req, int64_t timeout_ms)
 {
     /// Always use local mode

src/Cluster/MetaStore/MetaStore.h

@@ -23,6 +23,8 @@
 #include <Cluster/Requests/CreateDiskResponse.h>
 #include <Cluster/Requests/CreateFormatSchemaRequest.h>
 #include <Cluster/Requests/CreateFormatSchemaResponse.h>
+#include <Cluster/Requests/CreateNamedCollectionRequest.h>
+#include <Cluster/Requests/CreateNamedCollectionResponse.h>
 #include <Cluster/Requests/CreateStoragePolicyRequest.h>
 #include <Cluster/Requests/CreateStoragePolicyResponse.h>
 #include <Cluster/Requests/CreateStreamRequest.h>
@@ -41,6 +43,8 @@
 #include <Cluster/Requests/DeleteDiskResponse.h>
 #include <Cluster/Requests/DeleteFormatSchemaRequest.h>
 #include <Cluster/Requests/DeleteFormatSchemaResponse.h>
+#include <Cluster/Requests/DeleteNamedCollectionRequest.h>
+#include <Cluster/Requests/DeleteNamedCollectionResponse.h>
 #include <Cluster/Requests/DeleteStoragePolicyRequest.h>
 #include <Cluster/Requests/DeleteStoragePolicyResponse.h>
 #include <Cluster/Requests/DeleteStreamRequest.h>
@@ -57,6 +61,8 @@
 #include <Cluster/Requests/GetDatabaseResponse.h>
 #include <Cluster/Requests/GetFormatSchemaRequest.h>
 #include <Cluster/Requests/GetFormatSchemaResponse.h>
+#include <Cluster/Requests/GetNamedCollectionRequest.h>
+#include <Cluster/Requests/GetNamedCollectionResponse.h>
 #include <Cluster/Requests/GetStreamRequest.h>
 #include <Cluster/Requests/GetStreamResponse.h>
 #include <Cluster/Requests/GetTaskRequest.h>
@@ -73,6 +79,8 @@
 #include <Cluster/Requests/ListDisksResponse.h>
 #include <Cluster/Requests/ListFormatSchemasRequest.h>
 #include <Cluster/Requests/ListFormatSchemasResponse.h>
+#include <Cluster/Requests/ListNamedCollectionsRequest.h>
+#include <Cluster/Requests/ListNamedCollectionsResponse.h>
 #include <Cluster/Requests/ListStoragePoliciesRequest.h>
 #include <Cluster/Requests/ListStoragePoliciesResponse.h>
 #include <Cluster/Requests/ListStreamsRequest.h>
@@ -209,6 +217,11 @@ class MetaStore
     ListTasksResponsePtr listTasks(ListTasksRequestPtr req) const;
     GetTaskResponsePtr getTask(GetTaskRequestPtr req) const;
 
+    /// Named Collection
+    CreateNamedCollectionResponsePtr createNamedCollection(CreateNamedCollectionRequestPtr req);
+    DeleteNamedCollectionResponsePtr deleteNamedCollection(DeleteNamedCollectionRequestPtr req);
+    ListNamedCollectionsResponsePtr listNamedCollections(ListNamedCollectionsRequestPtr req) const;
+    GetNamedCollectionResponsePtr getNamedCollection(GetNamedCollectionRequestPtr req) const;
 
     bool isLocalNode(const std::string & host, uint16_t tcp_port, const std::string & fqdn_name) const;
 

src/Cluster/Protocol/CreateNamedCollectionRequestData.cpp

@@ -0,0 +1,50 @@
+#include <Cluster/Protocol/CreateNamedCollectionRequestData.h>
+
+#include <Cluster/Common/serde.h>
+#include <IO/ReadHelpers.h>
+#include <IO/WriteHelpers.h>
+
+
+namespace cluster::protocol
+{
+void CreateNamedCollectionRequestData::serialize(DB::WriteBuffer & wb, uint16_t version) const
+{
+    DB::writeStringBinary(collection_name, wb);
+    chassert(collection);
+    collection->serialize(wb, version);
+
+    DB::writeVarUInt(initiator, wb);
+    DB::writeVarInt(timeout_ms, wb);
+
+    serializeEnum(exists_op, wb);
+    DB::writeStringBinary(requested_by, wb);
+    DB::writeVarInt(requested_ts, wb);
+}
+
+void CreateNamedCollectionRequestData::doDeserialize(DB::ReadBuffer & rb, uint16_t version)
+{
+    DB::readStringBinary(collection_name, rb);
+    collection = std::make_shared<NamedCollectionDescriptor>();
+    collection->deserialize(rb, version);
+
+    DB::readVarUInt(initiator, rb);
+    DB::readVarInt(timeout_ms, rb);
+
+    exists_op = deserializeEnum<ExistsOperation>(rb);
+    DB::readStringBinary(requested_by, rb);
+    DB::readVarInt(requested_ts, rb);
+}
+
+std::string CreateNamedCollectionRequestData::doString() const
+{
+    return fmt::format(
+        "name={} collection={{{}}} initiator=0x{{:x}} timeout_ms={} exists_op={} requested_by={} requested_ts={}",
+        collection_name,
+        collection->string(),
+        initiator,
+        timeout_ms,
+        magic_enum::enum_name(exists_op),
+        requested_by,
+        requested_ts);
+}
+}