Merge branch 'jinhai' of 192.168.1.105:jinhai/vecwise_engine into jinhai

Former-commit-id: 6ef6cb1953310c4e01b1787e1a041c25f7dfd951

cpp/src/db/DB.h

@@ -30,6 +30,8 @@ public:
     virtual Status search(const std::string& group_id, size_t k, size_t nq,
             const float* vectors, QueryResults& results) = 0;
 
+    virtual Status drop_all() = 0;
+
     DB() = default;
     DB(const DB&) = delete;
     DB& operator=(const DB&) = delete;

cpp/src/db/DBImpl.cpp

@@ -246,9 +246,9 @@ Status DBImpl::background_merge_files(const std::string& group_id) {
         return status;
     }
 
-    if (raw_files.size() == 0) {
-        return Status::OK();
-    }
+    /* if (raw_files.size() == 0) { */
+    /*     return Status::OK(); */
+    /* } */
 
     bool has_merge = false;
 
@@ -261,9 +261,7 @@ Status DBImpl::background_merge_files(const std::string& group_id) {
         merge_files(group_id, kv.first, kv.second);
     }
 
-    if (has_merge) {
-        try_build_index();
-    }
+    try_build_index();
 
     _pMeta->cleanup_ttl_files(1);
 
@@ -346,6 +344,10 @@ void DBImpl::background_compaction() {
     }
 }
 
+Status DBImpl::drop_all() {
+    return _pMeta->drop_all();
+}
+
 DBImpl::~DBImpl() {
     {
         std::unique_lock<std::mutex> lock(_mutex);

cpp/src/db/DBImpl.h

@@ -36,6 +36,8 @@ public:
     virtual Status search(const std::string& group_id, size_t k, size_t nq,
             const float* vectors, QueryResults& results) override;
 
+    virtual Status drop_all() override;
+
     virtual ~DBImpl();
 
 private:

cpp/src/db/DBMetaImpl.cpp

@@ -427,6 +427,13 @@ Status DBMetaImpl::cleanup() {
     return Status::OK();
 }
 
+Status DBMetaImpl::drop_all() {
+    if (boost::filesystem::is_directory(_options.path)) {
+        boost::filesystem::remove_all(_options.path);
+    }
+    return Status::OK();
+}
+
 DBMetaImpl::~DBMetaImpl() {
     cleanup();
 }

cpp/src/db/DBMetaImpl.h

@@ -48,6 +48,8 @@ public:
 
     virtual Status cleanup_ttl_files(uint16_t seconds) override;
 
+    virtual Status drop_all() override;
+
     virtual ~DBMetaImpl();
 
 private:

cpp/src/db/LocalMetaImpl.cpp

@@ -246,6 +246,11 @@ Status LocalMetaImpl::cleanup_ttl_files(uint16_t seconds) {
     return Status::OK();
 }
 
+Status LocalMetaImpl::drop_all() {
+    // PXU TODO
+    return Status::OK();
+}
+
 } // namespace meta
 } // namespace engine
 } // namespace vecwise

cpp/src/db/LocalMetaImpl.h

@@ -43,6 +43,8 @@ public:
 
     virtual Status cleanup_ttl_files(uint16_t seconds) override;
 
+    virtual Status drop_all() override;
+
 private:
 
     Status GetGroupMetaInfoByPath(const std::string& path, GroupSchema& group_info);

cpp/src/db/Meta.h

@@ -83,6 +83,8 @@ public:
     virtual Status cleanup() = 0;
     virtual Status cleanup_ttl_files(uint16_t) = 0;
 
+    virtual Status drop_all() = 0;
+
     static DateT GetDate(const std::time_t& t);
     static DateT GetDate();
 

cpp/unittest/db/db_tests.cpp

@@ -7,6 +7,8 @@
 #include <faiss/IndexFlat.h>
 #include <faiss/MetaIndexes.h>
 #include <faiss/AutoTune.h>
+#include <thread>
+#include <easylogging++.h>
 
 #include "db/DB.h"
 #include "faiss/Index.h"
@@ -27,6 +29,7 @@ TEST(DBTest, DB_TEST) {
     static const int group_dim = 256;
 
     engine::Options opt;
+    opt.memory_sync_interval = 1;
     opt.meta.backend_uri = "http://127.0.0.1";
     opt.meta.path = "/tmp/vecwise_test/db_test";
 
@@ -38,7 +41,6 @@ TEST(DBTest, DB_TEST) {
     group_info.dimension = group_dim;
     group_info.group_id = group_name;
     engine::Status stat = db->add_group(group_info);
-    //ASSERT_STATS(stat);
 
     engine::meta::GroupSchema group_info_get;
     group_info_get.group_id = group_name;
@@ -47,21 +49,48 @@ TEST(DBTest, DB_TEST) {
     ASSERT_EQ(group_info_get.dimension, group_dim);
 
     engine::IDNumbers vector_ids;
-    std::vector<float> vec_f;
-    for(int i = 0; i < group_dim; i++){
-        vec_f.push_back((float)i);
-        vector_ids.push_back(i+1);
+    engine::IDNumbers target_ids;
+
+    int d = 256;
+    int nb = 10;
+    float *xb = new float[d * nb];
+    for(int i = 0; i < nb; i++) {
+        for(int j = 0; j < d; j++) xb[d * i + j] = drand48();
+        xb[d * i] += i / 2000.;
     }
-    stat = db->add_vectors(group_name, 1, vec_f.data(), vector_ids);
+
+    int qb = 1;
+    float *qxb = new float[d * qb];
+    for(int i = 0; i < qb; i++) {
+        for(int j = 0; j < d; j++) qxb[d * i + j] = drand48();
+        qxb[d * i] += i / 2000.;
+    }
+
+    int loop = 500000;
+
+    for (auto i=0; i<loop; ++i) {
+        if (i==40) {
+            db->add_vectors(group_name, qb, qxb, target_ids);
+        } else {
+            db->add_vectors(group_name, nb, xb, vector_ids);
+        }
+    }
+
+    std::this_thread::sleep_for(std::chrono::seconds(2));
+
+    engine::QueryResults results;
+    int k = 10;
+    LOG(DEBUG) << "PRE";
+    stat = db->search(group_name, k, qb, qxb, results);
+    LOG(DEBUG) << "POST";
     ASSERT_STATS(stat);
+    ASSERT_EQ(results[0][0], target_ids[0]);
 
-    //engine::QueryResults results;
-    //std::vector<float> vec_s = vec_f;
-    //stat = db->search(group_name, 1, 1, vec_f.data(), results);
-    //ASSERT_STATS(stat);
-    //ASSERT_EQ(results.size(), 1);
-    //ASSERT_EQ(results[0][0], vector_ids[0]);
-
+    delete [] xb;
+    delete [] qxb;
+    delete db;
+    engine::DB::Open(opt, &db);
+    db->drop_all();
     delete db;
 }
 
@@ -141,4 +170,7 @@ TEST(SearchTest, DB_TEST) {
     // TODO(linxj): add groundTruth assert
 
     delete db;
-}
+    engine::DB::Open(opt, &db);
+    db->drop_all();
+    delete db;
+}