v3.2.7 (#46)

Unified storage

.gitignore

@@ -0,0 +1,13 @@
+release
+debug
+build
+gtest
+flatbuffers
+flatbuffers-1.11.0
+fbs-gen
+.vscode
+.idea
+GPATH
+GRTAGS
+GTAGS
+third_party/faiss

CMakeLists.txt

@@ -2,14 +2,13 @@
 #    GAMMA ENGINE 
 #--------------------------------------------
 project(gamma_engine C CXX)
-cmake_minimum_required(VERSION 3.0)
+cmake_minimum_required(VERSION 3.17)
 
 list(APPEND CMAKE_MODULE_PATH ${PROJECT_SOURCE_DIR}/cmake/Modules)
 
 option(BUILD_TEST "Build tests" off)
 option(BUILD_WITH_GPU "Build gamma with gpu index support" off)
 option(BUILD_TOOLS "Build tools" off)
-option(TABLE_STR_INT64 "table string int64 address" off)
 
 exec_program(
     "sh"
@@ -28,7 +27,10 @@ set(GAMMA_SEARCH ${CMAKE_CURRENT_SOURCE_DIR}/search)
 set(GAMMA_INDEX ${CMAKE_CURRENT_SOURCE_DIR}/index)
 set(GAMMA_INDEX_IMPL ${CMAKE_CURRENT_SOURCE_DIR}/index/impl)
 set(GAMMA_INDEX_IMPL_HNSWLIB ${CMAKE_CURRENT_SOURCE_DIR}/index/impl/hnswlib)
+set(GAMMA_INDEX_IMPL_SSG ${CMAKE_CURRENT_SOURCE_DIR}/index/impl/ssg)
 set(GAMMA_VECTOR ${CMAKE_CURRENT_SOURCE_DIR}/vector)
+set(STORAGE ${CMAKE_CURRENT_SOURCE_DIR}/storage)
+set(STORAGE_COMPRESS ${CMAKE_CURRENT_SOURCE_DIR}/storage/compress)
 set(GAMMA_REALTIME ${CMAKE_CURRENT_SOURCE_DIR}/realtime)
 set(GAMMA_TABLE ${CMAKE_CURRENT_SOURCE_DIR}/table)
 set(GAMMA_C_API ${CMAKE_CURRENT_SOURCE_DIR}/c_api)
@@ -67,11 +69,6 @@ ELSE (DEFINED ENV{ZFP_HOME})
     MESSAGE(STATUS "ZFP home isn't set, so COMPRESS is not supported! ")
 ENDIF (DEFINED ENV{ZFP_HOME})
 
-if(TABLE_STR_INT64)
-    MESSAGE(STATUS "TABLE_STR_INT64 is on")
-    ADD_DEFINITIONS(-DTABLE_STR_INT64)
-endif(TABLE_STR_INT64)
-
 #INCLUDE DIR
 include_directories(
     ${THIRDPARTY}
@@ -86,7 +83,10 @@ include_directories(
     ${GAMMA_INDEX}
     ${GAMMA_INDEX_IMPL}
     ${GAMMA_INDEX_IMPL_HNSWLIB}
+    ${GAMMA_INDEX_IMPL_SSG}
     ${GAMMA_VECTOR}
+    ${STORAGE}
+    ${STORAGE_COMPRESS}
     ${GAMMA_REALTIME}
     ${GAMMA_C_API}
     ${GAMMA_IO}
@@ -108,6 +108,7 @@ set(LIBRARIES
     -lcrypto
     -lzstd
     -lopenblas
+    -ltbb
 )
 
 #ADDITIONAL SOURCE DIR
@@ -119,8 +120,11 @@ aux_source_directory(${UTIL} DIR_SRCS)
 aux_source_directory(${COMMON} DIR_SRCS)
 aux_source_directory(${GAMMA_SEARCH} DIR_SRCS)
 aux_source_directory(${GAMMA_INDEX} DIR_SRCS)
+aux_source_directory(${STORAGE} DIR_SRCS)
+aux_source_directory(${STORAGE_COMPRESS} DIR_SRCS)
 aux_source_directory(${GAMMA_INDEX_IMPL} DIR_SRCS)
 aux_source_directory(${GAMMA_INDEX_IMPL_HNSWLIB} DIR_SRCS)
+aux_source_directory(${GAMMA_INDEX_IMPL_SSG} DIR_SRCS)
 aux_source_directory(${GAMMA_VECTOR} DIR_SRCS)
 aux_source_directory(${GAMMA_REALTIME} DIR_SRCS)
 aux_source_directory(${GAMMA_TABLE} DIR_SRCS)

c_api/api_data/gamma_config.cc

@@ -11,9 +11,22 @@ namespace tig_gamma {
 
 int Config::Serialize(char **out, int *out_len) {
   flatbuffers::FlatBufferBuilder builder;
+
+  std::vector<flatbuffers::Offset<gamma_api::CacheInfo>>
+                                        cache_vector(cache_infos_.size());
+  int i = 0;
+  for (auto &c : cache_infos_) {
+    auto cache = gamma_api::CreateCacheInfo(builder, 
+                                builder.CreateString(c.field_name),
+                                c.cache_size);
+    cache_vector[i++] = cache;
+  }
+  auto cache_vec = builder.CreateVector(cache_vector);
   auto config =
       gamma_api::CreateConfig(builder, builder.CreateString(path_),
-                              builder.CreateString(log_dir_));
+                              builder.CreateString(log_dir_),
+                              cache_vec);
+
   builder.Finish(config);
   *out_len = builder.GetSize();
   *out = (char *)malloc(*out_len * sizeof(char));
@@ -26,6 +39,16 @@ void Config::Deserialize(const char *data, int len) {
 
   path_ = config_->path()->str();
   log_dir_ = config_->log_dir()->str();
+
+  size_t cache_num = config_->cache_infos()->size();
+  cache_infos_.resize(cache_num);
+  for (size_t i = 0; i < cache_num; ++i) {
+    auto c = config_->cache_infos()->Get(i);
+    struct CacheInfo cache_info;
+    cache_info.field_name = c->field_name()->str();
+    cache_info.cache_size = c->cache_size();
+    cache_infos_[i] = cache_info;
+  }
 }
 
 const std::string &Config::Path() {
@@ -42,4 +65,19 @@ const std::string &Config::LogDir() {
 
 void Config::SetLogDir(std::string &log_dir) { log_dir_ = log_dir; }
 
+void Config::AddCacheInfo(const struct CacheInfo &cache) {
+  cache_infos_.push_back(cache);
+}
+
+void Config::AddCacheInfo(struct CacheInfo &&cache) {
+  cache_infos_.emplace_back(std::forward<struct CacheInfo>(cache));
+}
+
+void Config::AddCacheInfo(std::string name, int cache_size) {
+  struct CacheInfo c;
+  c.field_name = name;
+  c.cache_size = cache_size;
+  cache_infos_.push_back(c);
+}
+
 }  // namespace tig_gamma

c_api/api_data/gamma_config.h

@@ -12,8 +12,31 @@
 #include "config_generated.h"
 #include "gamma_raw_data.h"
 
+
 namespace tig_gamma {
 
+struct CacheInfo {
+  std::string field_name;
+  int cache_size;
+  CacheInfo() {}
+
+  CacheInfo(const CacheInfo &other) { *this = other; }
+
+  CacheInfo &operator=(const CacheInfo &other) {
+    field_name = other.field_name;
+    cache_size = other.cache_size;
+    return *this;
+  }
+
+  CacheInfo(CacheInfo &&other) { *this = std::move(other); }
+
+  CacheInfo &operator=(CacheInfo &&other) {
+    field_name = std::move(other.field_name);
+    cache_size = other.cache_size;
+    return *this;
+  }
+};
+
 class Config : public RawData {
  public:
   Config() { config_ = nullptr; }
@@ -30,11 +53,22 @@ class Config : public RawData {
 
   void SetLogDir(std::string &log_dir);
 
+  void AddCacheInfo(const struct CacheInfo &cache);
+
+  void AddCacheInfo(struct CacheInfo &&cache);
+
+  void AddCacheInfo(std::string name, int cache_size);
+
+  std::vector<CacheInfo> &CacheInfos() { return cache_infos_; }
+
+  void ClearCacheInfos() { cache_infos_.resize(0); }
+
  private:
   gamma_api::Config *config_;
 
   std::string path_;
   std::string log_dir_;
+  std::vector<CacheInfo> cache_infos_;
 };
 
 }  // namespace tig_gamma

c_api/api_data/gamma_doc.cc

@@ -53,9 +53,9 @@ void Doc::Deserialize(const char *data, int len) {
   size_t fields_num = doc_->fields()->size();
 
   if (fields_num != table_field_num + vector_field_num) {
-    LOG(ERROR) << "Add Doc fields num [" << fields_num
-               << "], not equal to table_field_num [" << table_field_num
-               << "] + vector_field_num [" << vector_field_num << "]";
+    LOG(WARNING) << "Add Doc fields num [" << fields_num
+                 << "], not equal to table_field_num [" << table_field_num
+                 << "] + vector_field_num [" << vector_field_num << "]";
     return;
   }
 

c_api/gamma_api.cc

@@ -211,3 +211,19 @@ int DelDocByQuery(void *engine, const char *request_str, int len) {
       static_cast<tig_gamma::GammaEngine *>(engine)->DelDocByQuery(request);
   return ret;
 }
+
+int SetConfig(void *engine, const char *config_str, int len) {
+  tig_gamma::Config config;
+  config.Deserialize(config_str, len);
+  int ret =
+      static_cast<tig_gamma::GammaEngine *>(engine)->SetConfig(config);
+  return ret;
+}
+
+int GetConfig(void *engine, char **config_str, int *len) {
+  tig_gamma::Config config;
+  int res = 
+      static_cast<tig_gamma::GammaEngine *>(engine)->GetConfig(config);
+  if (res == 0) { res = config.Serialize(config_str, len); }
+  return res;
+}

c_api/gamma_api.h

@@ -161,6 +161,22 @@ int Search(void *engine, const char *request_str, int req_len,
  */
 int DelDocByQuery(void *engine, const char *request_str, int len);
 
+/** alter all cache size by query
+ *
+ * @param engine  search engine pointer
+ * @param cache_str caches' serialized string
+ * @return 0 successed, 1 failed
+ */
+int SetConfig(void *engine, const char *config_str, int len);
+
+/** get all cache size by query
+ *
+ * @param engine  search engine pointer
+ * @param cache_str caches' serialized string
+ * @return 0 successed, 1 failed
+ */
+int GetConfig(void *engine, char **config_str, int *len);
+
 #ifdef __cplusplus
 }
 #endif

idl/fbs/config.fbs

@@ -1,8 +1,14 @@
 namespace gamma_api;
 
+table CacheInfo {
+  field_name:string;
+  cache_size:int;
+} 
+
 table Config {
   path:string;
   log_dir:string;
+  cache_infos:[CacheInfo];
 }
 
 root_type Config;

index/impl/gamma_index_ivfpq.cc

@@ -1357,7 +1357,8 @@ int GammaIVFPQIndex::Load(const std::string &index_dir) {
     READ1(indexed_vec_count_);
     if (indexed_vec_count_ < 0 ||
         indexed_vec_count_ > vector_->MetaInfo()->size_) {
-      LOG(ERROR) << "invalid indexed count=" << indexed_vec_count_;
+      LOG(ERROR) << "invalid indexed count [" << indexed_vec_count_ 
+                 << "] vector size [" << vector_->MetaInfo()->size_ << "]";
       return INTERNAL_ERR;
     }
     // precomputed table not stored. It is cheaper to recompute it

index/impl/gpu/gamma_index_ivfpq_gpu.cc

@@ -676,7 +676,11 @@ int ParseFilters(GammaSearchCondition *condition,
 template <class T>
 bool IsInRange(Table *table, RangeFilter &range, long docid) {
   T value = 0;
-  table->GetField<T>(docid, range.field, value);
+  std::string field_value;
+  int field_id = table->GetAttrIdx(range.field); 
+  table->GetFieldRawValue(docid, field_id, field_value);
+  memcpy(&value, field_value.c_str(), sizeof(value));
+
   T lower_value, upper_value;
   memcpy(&lower_value, range.lower_value.c_str(), range.lower_value.size());
   memcpy(&upper_value, range.upper_value.c_str(), range.upper_value.size());
@@ -718,11 +722,10 @@ bool FilteredByTermFilter(GammaSearchCondition *condition,
     auto term = condition->term_filters[i];
 
     std::string field_value;
-    table::DecompressStr decompress_str;
-    int len = condition->table->GetFieldString(docid, term.field, field_value,
-                                               decompress_str);
+    int field_id = condition->table->GetAttrIdx(term.field); 
+    condition->table->GetFieldRawValue(docid, field_id, field_value);
     vector<string> field_items;
-    if (len >= 0) field_items = utils::split(field_value, kDelim);
+    if (field_value.size() >= 0) field_items = utils::split(field_value, kDelim);
 
     bool all_in_field_items;
     if (term.is_union == static_cast<int>(FilterOperator::Or))

index/retrieval_model.h

@@ -8,8 +8,10 @@
 #pragma once
 
 #include <vector>
+#include <tbb/concurrent_queue.h>
 
-#include "concurrentqueue/concurrentqueue.h"
+
+// #include "concurrentqueue/concurrentqueue.h"
 #include "reflector.h"
 #include "utils.h"
 
@@ -178,7 +180,6 @@ class ScopeVectors {
 
   size_t Size() { return ptr_.size(); }
 
- private:
   std::vector<const uint8_t *> ptr_;
   std::vector<bool> deletable_;
 };
@@ -292,7 +293,7 @@ class RetrievalModel {
   virtual int Load(const std::string &dir) = 0;
 
   VectorReader *vector_;
-  moodycamel::ConcurrentQueue<int> updated_vids_;
+  tbb::concurrent_bounded_queue<int> updated_vids_;
   // warining: indexed_count_ is only used by framework, sub-class cann't use it
   int indexed_count_;
 };

io/memory_raw_vector_io.h

@@ -1,8 +1,5 @@
 #ifdef WITH_ROCKSDB
 
-#ifndef MEMORY_RAW_VECTOR_IO_H_
-#define MEMORY_RAW_VECTOR_IO_H_
-
 #pragma once
 
 #include <string>
@@ -32,6 +29,4 @@ struct MemoryRawVectorIO : public RawVectorIO, public AsyncFlusher {
 
 }  // namespace tig_gamma
 
-#endif
-
 #endif // WITH_ROCKSDB

io/mmap_raw_vector_io.cc

@@ -7,26 +7,14 @@ namespace tig_gamma {
 int MmapRawVectorIO::Init() { return 0; }
 
 int MmapRawVectorIO::Dump(int start_vid, int end_vid) {
-  for (int i = start_vid / raw_vector->segment_size_;
-       i < end_vid / raw_vector->segment_size_; i++) {
-    int ret = raw_vector->file_mappers_[i]->Sync();
-    if (ret) return ret;
-  }
   return 0;
 }
 
 int MmapRawVectorIO::Load(int vec_num) {
-  int seg_num = vec_num / raw_vector->segment_size_ + 1;
-  int offset = vec_num % raw_vector->segment_size_;
-  for (int i = 1; i < seg_num; ++i) {
-    int ret = raw_vector->Extend();
-    if (ret) {
-      LOG(ERROR) << "load extend error, i=" << i << ", ret=" << ret;
-      return ret;
-    }
+  if (raw_vector->storage_mgr_->Truncate(vec_num)) {
+    LOG(ERROR) << "truncate gamma db error, vec_num=" << vec_num;
+    return INTERNAL_ERR;
   }
-  assert(raw_vector->nsegment_ == seg_num);
-  raw_vector->file_mappers_[seg_num - 1]->SetCurrIdx(offset);
   raw_vector->MetaInfo()->size_ = vec_num;
   LOG(INFO) << "mmap load success! vec num=" << vec_num;
   return 0;

io/mmap_raw_vector_io.h

@@ -1,6 +1,3 @@
-#ifndef MMAP_RAW_VECTOR_IO_H_
-#define MMAP_RAW_VECTOR_IO_H_
-
 #pragma once
 
 #include <string>
@@ -24,4 +21,3 @@ struct MmapRawVectorIO : public RawVectorIO {
 
 }  // namespace tig_gamma
 
-#endif