add Macro MILVUS_USE_SNAPSHOT (#2972)

* add Macro MILVUS_USE_SNAPSHOT to control Signed-off-by: yudong.cai <yudong.cai@zilliz.com> * fix clang-format Signed-off-by: yudong.cai <yudong.cai@zilliz.com> * fix merge error Signed-off-by: yudong.cai <yudong.cai@zilliz.com>
2020-07-23 14:15:01 +08:00 · 2020-07-23 14:15:01 +08:00 · f492d80cbb
parent 54df521fce
commit f492d80cbb
5 changed files with 433 additions and 1 deletions
--- a/core/CMakeLists.txt
+++ b/core/CMakeLists.txt
@ -70,6 +70,7 @@ endif ()
 set_milvus_definition(MILVUS_WITH_PROMETHEUS "MILVUS_WITH_PROMETHEUS")
 set_milvus_definition(ENABLE_CPU_PROFILING "ENABLE_CPU_PROFILING")
 set_milvus_definition(MILVUS_WITH_FIU "FIU_ENABLE")
+set_milvus_definition(MILVUS_USE_SNAPSHOT "MILVUS_USE_SNAPSHOT")

 config_summary()

--- a/core/src/db/DB.h
+++ b/core/src/db/DB.h
@ -11,6 +11,7 @@

 #pragma once

+#include <map>
 #include <memory>
 #include <string>
 #include <unordered_map>
@ -19,8 +20,10 @@
 #include "Options.h"
 #include "Types.h"
 #include "context/HybridSearchContext.h"
+#include "db/snapshot/Context.h"
 #include "meta/Meta.h"
 #include "query/GeneralQuery.h"
+#include "segment/Segment.h"
 #include "server/context/Context.h"
 #include "utils/Status.h"

@ -182,6 +185,80 @@ class DB {
                          const std::unordered_map<std::string, std::vector<uint8_t>>& attr_data,
                          std::unordered_map<std::string, int64_t>& attr_size,
                          std::unordered_map<std::string, knowhere::IndexPtr>& attr_indexes) = 0;
+
+    ///////////////////////////////////////////////////////////////////////////////
+    /* mocked SSDBImpl interfaces, will be removed */
+    virtual Status
+    CreateCollection(const snapshot::CreateCollectionContext& context) = 0;
+
+    // Status
+    // DropCollection(const std::string& name);
+
+    virtual Status
+    DescribeCollection(const std::string& collection_name, snapshot::CollectionPtr& collection,
+                       std::map<snapshot::FieldPtr, std::vector<snapshot::FieldElementPtr>>& fields_schema) = 0;
+
+    // Status
+    // HasCollection(const std::string& collection_name, bool& has_or_not);
+
+    // Status
+    // AllCollections(std::vector<std::string>& names);
+
+    // Status
+    // GetCollectionRowCount(const std::string& collection_name, uint64_t& row_count);
+
+    virtual Status
+    LoadCollection(const server::ContextPtr& context, const std::string& collection_name,
+                   const std::vector<std::string>& field_names, bool force = false) = 0;
+
+    virtual Status
+    CreatePartition(const std::string& collection_name, const std::string& partition_name) = 0;
+
+    virtual Status
+    DropPartition(const std::string& collection_name, const std::string& partition_name) = 0;
+
+    virtual Status
+    ShowPartitions(const std::string& collection_name, std::vector<std::string>& partition_names) = 0;
+
+    virtual Status
+    InsertEntities(const std::string& collection_name, const std::string& partition_name, DataChunkPtr& data_chunk) = 0;
+
+    // Status
+    // DeleteEntities(const std::string& collection_name, engine::IDNumbers entity_ids);
+
+    // Status
+    // Flush(const std::string& collection_name);
+
+    // Status
+    // Flush();
+
+    // Status
+    // Compact(const server::ContextPtr& context, const std::string& collection_name, double threshold = 0.0);
+
+    virtual Status
+    GetEntityByID(const std::string& collection_name, const IDNumbers& id_array,
+                  const std::vector<std::string>& field_names, DataChunkPtr& data_chunk) = 0;
+
+    virtual Status
+    GetEntityIDs(const std::string& collection_id, int64_t segment_id, IDNumbers& entity_ids) = 0;
+
+    virtual Status
+    CreateIndex(const server::ContextPtr& context, const std::string& collection_id, const std::string& field_name,
+                const CollectionIndex& index) = 0;
+
+    virtual Status
+    DescribeIndex(const std::string& collection_id, const std::string& field_name, CollectionIndex& index) = 0;
+
+    virtual Status
+    DropIndex(const std::string& collection_name, const std::string& field_name) = 0;
+
+    // Status
+    // DropIndex(const std::string& collection_id);
+
+    virtual Status
+    Query(const server::ContextPtr& context, const std::string& collection_name, const query::QueryPtr& query_ptr,
+          engine::QueryResultPtr& result) = 0;
+    ///////////////////////////////////////////////////////////////////////////////
 };  // DB

 using DBPtr = std::shared_ptr<DB>;
--- a/core/src/db/DBFactory.cpp
+++ b/core/src/db/DBFactory.cpp
@ -10,7 +10,11 @@
 // or implied. See the License for the specific language governing permissions and limitations under the License.

 #include "db/DBFactory.h"
+#ifdef MILVUS_USE_SNAPSHOT
+#include "SSDBImpl.h"
+#else
 #include "DBImpl.h"
+#endif
 #include "meta/MetaFactory.h"
 #include "meta/MySQLMetaImpl.h"
 #include "meta/SqliteMetaImpl.h"
@ -35,7 +39,11 @@ DBFactory::BuildOption() {

 DBPtr
 DBFactory::Build(const DBOptions& options) {
+#ifdef MILVUS_USE_SNAPSHOT
+    return std::make_shared<SSDBImpl>(options);
+#else
    return std::make_shared<DBImpl>(options);
+#endif
 }

 }  // namespace engine
--- a/core/src/db/DBImpl.h
+++ b/core/src/db/DBImpl.h
@ -31,6 +31,7 @@
 #include "db/insert/MemManager.h"
 #include "db/merge/MergeManager.h"
 #include "db/meta/FilesHolder.h"
+#include "db/snapshot/Context.h"
 #include "utils/ThreadPool.h"
 #include "wal/WalManager.h"

@ -308,6 +309,120 @@ class DBImpl : public DB, public server::CacheConfigHandler, public server::Engi
                             const std::unordered_map<std::string, int64_t>& attr_sizes,
                             const std::unordered_map<std::string, meta::hybrid::DataType>& attr_types);

+    ///////////////////////////////////////////////////////////////////////////////
+    /* Mocked SSDBImpl interfaces, will be removed */
+    Status
+    CreateCollection(const snapshot::CreateCollectionContext& context) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // DropCollection(const std::string& name);
+
+    Status
+    DescribeCollection(const std::string& collection_name, snapshot::CollectionPtr& collection,
+                       std::map<snapshot::FieldPtr, std::vector<snapshot::FieldElementPtr>>& fields_schema) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // HasCollection(const std::string& collection_name, bool& has_or_not);
+
+    // Status
+    // AllCollections(std::vector<std::string>& names);
+
+    // Status
+    // GetCollectionRowCount(const std::string& collection_name, uint64_t& row_count);
+
+    Status
+    LoadCollection(const server::ContextPtr& context, const std::string& collection_name,
+                   const std::vector<std::string>& field_names, bool force = false) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    CreatePartition(const std::string& collection_name, const std::string& partition_name) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    DropPartition(const std::string& collection_name, const std::string& partition_name) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    ShowPartitions(const std::string& collection_name, std::vector<std::string>& partition_names) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    InsertEntities(const std::string& collection_name, const std::string& partition_name,
+                   DataChunkPtr& data_chunk) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // DeleteEntities(const std::string& collection_name, engine::IDNumbers entity_ids);
+
+    // Status
+    // Flush(const std::string& collection_name);
+
+    // Status
+    // Flush();
+
+    // Status
+    // Compact(const server::ContextPtr& context, const std::string& collection_name, double threshold = 0.0);
+
+    Status
+    GetEntityByID(const std::string& collection_name, const IDNumbers& id_array,
+                  const std::vector<std::string>& field_names, DataChunkPtr& data_chunk) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    GetEntityIDs(const std::string& collection_id, int64_t segment_id, IDNumbers& entity_ids) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    CreateIndex(const server::ContextPtr& context, const std::string& collection_id, const std::string& field_name,
+                const CollectionIndex& index) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    DescribeIndex(const std::string& collection_id, const std::string& field_name, CollectionIndex& index) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    DropIndex(const std::string& collection_name, const std::string& field_name) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // DropIndex(const std::string& collection_id);
+
+    Status
+    Query(const server::ContextPtr& context, const std::string& collection_name, const query::QueryPtr& query_ptr,
+          engine::QueryResultPtr& result) override {
+        assert(false);
+        return Status::OK();
+    }
+    ///////////////////////////////////////////////////////////////////////////////
+
 private:
    DBOptions options_;

--- a/core/src/db/SSDBImpl.h
+++ b/core/src/db/SSDBImpl.h
@ -22,6 +22,7 @@
 #include <unordered_map>
 #include <vector>

+#include "db/DB.h"
 #include "db/Options.h"
 #include "db/SimpleWaitNotify.h"
 #include "db/SnapshotHandlers.h"
@ -37,7 +38,7 @@
 namespace milvus {
 namespace engine {

-class SSDBImpl {
+class SSDBImpl : public DB {
 public:
    explicit SSDBImpl(const DBOptions& options);

@ -166,6 +167,236 @@ class SSDBImpl {
    void
    ResumeIfLast();

+    ///////////////////////////////////////////////////////////////////////////////
+    /* Mocked DBImpl interfaces, will be removed */
+    Status
+    DropAll() override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    CreateCollection(meta::CollectionSchema& collection_schema) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // DropCollection(const std::string& collection_id) override;
+
+    Status
+    DescribeCollection(meta::CollectionSchema& collection_schema) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // HasCollection(const std::string& collection_id, bool& has_or_not) override;
+
+    Status
+    HasNativeCollection(const std::string& collection_id, bool& has_or_not_) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // AllCollections(std::vector<std::string>& names) override;
+
+    Status
+    GetCollectionInfo(const std::string& collection_id, std::string& collection_info) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    PreloadCollection(const std::shared_ptr<server::Context>& context, const std::string& collection_id,
+                      bool force = false) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    ReLoadSegmentsDeletedDocs(const std::string& collection_id, const std::vector<int64_t>& segment_ids) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    UpdateCollectionFlag(const std::string& collection_id, int64_t flag) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // GetCollectionRowCount(const std::string& collection_id, uint64_t& row_count) override;
+
+    Status
+    CreatePartition(const std::string& collection_id, const std::string& partition_name,
+                    const std::string& partition_tag) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    HasPartition(const std::string& collection_id, const std::string& tag, bool& has_or_not) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    DropPartition(const std::string& partition_name) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    DropPartitionByTag(const std::string& collection_id, const std::string& partition_tag) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    ShowPartitions(const std::string& collection_id,
+                   std::vector<meta::CollectionSchema>& partition_schema_array) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    InsertVectors(const std::string& collection_id, const std::string& partition_tag, VectorsData& vectors) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // Flush(const std::string& collection_id) override;
+
+    // Status
+    // Flush() override;
+
+    // Status
+    // Compact(const std::shared_ptr<server::Context>& context, const std::string& collection_id,
+    //         double threshold = 0.0) override;
+
+    Status
+    GetVectorsByID(const engine::meta::CollectionSchema& collection, const IDNumbers& id_array,
+                   std::vector<engine::VectorsData>& vectors) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    GetEntitiesByID(const std::string& collection_id, const IDNumbers& id_array,
+                    const std::vector<std::string>& field_names, std::vector<engine::VectorsData>& vectors,
+                    std::vector<engine::AttrsData>& attrs) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    GetVectorIDs(const std::string& collection_id, const std::string& segment_id, IDNumbers& vector_ids) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    //    Status
+    //    Merge(const std::set<std::string>& collection_ids) override;
+
+    Status
+    CreateIndex(const std::shared_ptr<server::Context>& context, const std::string& collection_id,
+                const CollectionIndex& index) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    CreateStructuredIndex(const std::string& collection_id, const std::vector<std::string>& field_names,
+                          const std::unordered_map<std::string, meta::hybrid::DataType>& attr_types,
+                          const std::unordered_map<std::string, std::vector<uint8_t>>& attr_data,
+                          std::unordered_map<std::string, int64_t>& attr_size,
+                          std::unordered_map<std::string, knowhere::IndexPtr>& attr_indexes) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    DescribeIndex(const std::string& collection_id, CollectionIndex& index) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    // Status
+    // DropIndex(const std::string& collection_id) override;
+
+    Status
+    CreateHybridCollection(meta::CollectionSchema& collection_schema,
+                           meta::hybrid::FieldsSchema& fields_schema) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    DescribeHybridCollection(meta::CollectionSchema& collection_schema,
+                             meta::hybrid::FieldsSchema& fields_schema) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    InsertEntities(const std::string& collection_name, const std::string& partition_tag,
+                   const std::vector<std::string>& field_names, engine::Entity& entity,
+                   std::unordered_map<std::string, meta::hybrid::DataType>& field_types) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    HybridQuery(const std::shared_ptr<server::Context>& context, const std::string& collection_id,
+                const std::vector<std::string>& partition_tags, query::GeneralQueryPtr general_query,
+                query::QueryPtr query_ptr, std::vector<std::string>& field_names,
+                std::unordered_map<std::string, engine::meta::hybrid::DataType>& attr_type,
+                engine::QueryResult& result) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    QueryByIDs(const std::shared_ptr<server::Context>& context, const std::string& collection_id,
+               const std::vector<std::string>& partition_tags, uint64_t k, const milvus::json& extra_params,
+               const IDNumbers& id_array, ResultIds& result_ids, ResultDistances& result_distances) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    Query(const std::shared_ptr<server::Context>& context, const std::string& collection_id,
+          const std::vector<std::string>& partition_tags, uint64_t k, const milvus::json& extra_params,
+          VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    QueryByFileID(const std::shared_ptr<server::Context>& context, const std::vector<std::string>& file_ids, uint64_t k,
+                  const milvus::json& extra_params, VectorsData& vectors, ResultIds& result_ids,
+                  ResultDistances& result_distances) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    Size(uint64_t& result) override {
+        assert(false);
+        return Status::OK();
+    }
+
+    Status
+    FlushAttrsIndex(const std::string& collection_id) override {
+        assert(false);
+        return Status::OK();
+    }
+    ///////////////////////////////////////////////////////////////////////////////
+
 private:
    DBOptions options_;
    std::atomic<bool> initialized_;