Merge pull request #2743 from fishpenguin/scalar-field-filtering

Scalar field filtering

CHANGELOG.md

@@ -21,7 +21,10 @@ Please mark all changes in change log and use the issue from GitHub
 -   \#2696 Check the validity of the parameters of creating collection: segment_size
 -   \#2697 Index can not be created
 -   \#2698 Count entities got wrong result with binary vectors
+-   \#2728 Index type name should returned if index type is not supported
 -   \#2731 No entity returned with `get_entity_by_id`
+-   \#2732 Server destroyed after `delete by id`
+-   \#2733 The max value of top-k should be limited
 
 ## Feature
 -   \#2319 Redo metadata to support MVCC

core/src/db/DB.h

@@ -103,9 +103,15 @@ class DB {
     virtual Status
     DeleteVector(const std::string& collection_id, IDNumber vector_id) = 0;
 
+    virtual Status
+    DeleteEntity(const std::string& collection_id, IDNumber entity_id) = 0;
+
     virtual Status
     DeleteVectors(const std::string& collection_id, IDNumbers vector_ids) = 0;
 
+    virtual Status
+    DeleteEntities(const std::string& collection_id, IDNumbers entity_ids) = 0;
+
     virtual Status
     Flush(const std::string& collection_id) = 0;
 

core/src/db/DBImpl.cpp

@@ -1001,6 +1001,12 @@ DBImpl::DeleteVector(const std::string& collection_id, IDNumber vector_id) {
     return DeleteVectors(collection_id, ids);
 }
 
+Status
+DBImpl::DeleteEntity(const std::string& collection_id, milvus::engine::IDNumber entity_id) {
+    IDNumbers ids;
+    ids.push_back(entity_id);
+}
+
 Status
 DBImpl::DeleteVectors(const std::string& collection_id, IDNumbers vector_ids) {
     if (!initialized_.load(std::memory_order_acquire)) {
@@ -1025,6 +1031,30 @@ DBImpl::DeleteVectors(const std::string& collection_id, IDNumbers vector_ids) {
     return status;
 }
 
+Status
+DBImpl::DeleteEntities(const std::string& collection_id, milvus::engine::IDNumbers entity_ids) {
+    if (!initialized_.load(std::memory_order_acquire)) {
+        return SHUTDOWN_ERROR;
+    }
+
+    Status status;
+    if (options_.wal_enable_) {
+        wal_mgr_->DeleteById(collection_id, entity_ids);
+        swn_wal_.Notify();
+    } else {
+        wal::MXLogRecord record;
+        record.lsn = 0;  // need to get from meta ?
+        record.type = wal::MXLogType::Delete;
+        record.collection_id = collection_id;
+        record.ids = entity_ids.data();
+        record.length = entity_ids.size();
+
+        status = ExecWalRecord(record);
+    }
+
+    return status;
+}
+
 Status
 DBImpl::Flush(const std::string& collection_id) {
     if (!initialized_.load(std::memory_order_acquire)) {

core/src/db/DBImpl.h

@@ -113,9 +113,15 @@ class DBImpl : public DB, public server::CacheConfigHandler, public server::Engi
     Status
     DeleteVector(const std::string& collection_id, IDNumber vector_id) override;
 
+    Status
+    DeleteEntity(const std::string& collection_id, IDNumber entity_id) override;
+
     Status
     DeleteVectors(const std::string& collection_id, IDNumbers vector_ids) override;
 
+    Status
+    DeleteEntities(const std::string& collection_id, IDNumbers entity_ids) override;
+
     Status
     Flush(const std::string& collection_id) override;
 

core/src/db/wal/WalManager.cpp

@@ -575,9 +575,9 @@ WalManager::InsertEntities(const std::string& collection_id, const std::string&
 }
 
 bool
-WalManager::DeleteById(const std::string& collection_id, const IDNumbers& vector_ids) {
-    size_t vector_num = vector_ids.size();
-    if (vector_num == 0) {
+WalManager::DeleteById(const std::string& collection_id, const IDNumbers& entity_ids) {
+    size_t entity_num = entity_ids.size();
+    if (entity_num == 0) {
         LOG_WAL_ERROR_ << "The ids is empty.";
         return false;
     }
@@ -591,7 +591,7 @@ WalManager::DeleteById(const std::string& collection_id, const IDNumbers& vector
     record.partition_tag = "";
 
     uint64_t new_lsn = 0;
-    for (size_t i = 0; i < vector_num; i += record.length) {
+    for (size_t i = 0; i < entity_num; i += record.length) {
         size_t surplus_space = p_buffer_->SurplusSpace();
         size_t max_rcd_num = 0;
         if (surplus_space >= head_size + unit_size) {
@@ -600,8 +600,8 @@ WalManager::DeleteById(const std::string& collection_id, const IDNumbers& vector
             max_rcd_num = (mxlog_config_.buffer_size - head_size) / unit_size;
         }
 
-        record.length = std::min(vector_num - i, max_rcd_num);
-        record.ids = vector_ids.data() + i;
+        record.length = std::min(entity_num - i, max_rcd_num);
+        record.ids = entity_ids.data() + i;
         record.data_size = 0;
         record.data = nullptr;
 

core/src/server/ValidationUtil.cpp

@@ -193,8 +193,14 @@ Status
 ValidateCollectionIndexType(int32_t index_type) {
     int engine_type = static_cast<int>(engine::EngineType(index_type));
     if (engine_type <= 0 || engine_type > static_cast<int>(engine::EngineType::MAX_VALUE)) {
-        std::string msg = "Invalid index type: " + std::to_string(index_type) + ". " +
-                          "Make sure the index type is in IndexType list.";
+        std::string index_type_str;
+        for (auto it = engine::s_map_engine_type.begin(); it != engine::s_map_engine_type.end(); it++) {
+            if (it->second == (engine::EngineType)index_type) {
+                index_type_str = it->first;
+            }
+        }
+        std::string msg =
+            "Invalid index type: " + index_type_str + ". " + "Make sure the index type is in IndexType list.";
         LOG_SERVER_ERROR_ << msg;
         return Status(SERVER_INVALID_INDEX_TYPE, msg);
     }

core/src/server/delivery/request/DeleteByIDRequest.cpp

@@ -78,7 +78,7 @@ DeleteByIDRequest::OnExecute() {
 
         rc.RecordSection("check validation");
 
-        status = DBWrapper::DB()->DeleteVectors(collection_name_, vector_ids_);
+        status = DBWrapper::DB()->DeleteEntities(collection_name_, vector_ids_);
         if (!status.ok()) {
             return status;
         }

core/src/server/grpc_impl/GrpcRequestHandler.cpp

@@ -21,6 +21,7 @@
 
 #include "context/HybridSearchContext.h"
 #include "query/BinaryQuery.h"
+#include "server/ValidationUtil.h"
 #include "server/context/ConnectionContext.h"
 #include "tracing/TextMapCarrier.h"
 #include "tracing/TracerUtil.h"
@@ -1543,7 +1544,12 @@ GrpcRequestHandler::DeserializeJsonToBoolQuery(
             json::iterator vector_param_it = it.value().begin();
             if (vector_param_it != it.value().end()) {
                 vector_query->field_name = vector_param_it.key();
-                vector_query->topk = vector_param_it.value()["topk"];
+                int64_t topk = vector_param_it.value()["topk"];
+                status = server::ValidateSearchTopk(topk);
+                if (!status.ok()) {
+                    return status;
+                }
+                vector_query->topk = topk;
                 vector_query->extra_params = vector_param_it.value()["params"];
             }