Disable the dropCollection filtering

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>

internal/core/src/query/ExprImpl.h

@@ -13,11 +13,12 @@
 #include "Expr.h"
 #include <tuple>
 #include <vector>
+#include <boost/container/vector.hpp>
 
 namespace milvus::query {
 template <typename T>
 struct TermExprImpl : TermExpr {
-    std::vector<T> terms_;
+    boost::container::vector<T> terms_;
 };
 
 template <typename T>

internal/core/src/query/visitors/ShowExprVisitor.cpp

@@ -115,6 +115,8 @@ ShowExprVisitor::visit(TermExpr& expr) {
     Assert(field_is_vector(expr.data_type_) == false);
     auto terms = [&] {
         switch (expr.data_type_) {
+            case DataType::BOOL:
+                return TermExtract<bool>(expr);
             case DataType::INT8:
                 return TermExtract<int8_t>(expr);
             case DataType::INT16:

internal/core/src/segcore/ConcurrentVector.h

@@ -25,33 +25,6 @@
 
 namespace milvus::segcore {
 
-// we don't use std::array because capacity of concurrent_vector wastes too much memory
-// template <typename Type>
-// class FixedVector : public std::vector<Type> {
-// public:
-//    // This is a stupid workaround for tbb API to avoid memory copy
-//    explicit FixedVector(int64_t size) : placeholder_size_(size) {
-//    }
-//    FixedVector(const FixedVector<Type>& placeholder_vec)
-//        : std::vector<Type>(placeholder_vec.placeholder_size_), is_placeholder_(false) {
-//        // Assert(placeholder_vec.is_placeholder_);
-//    }
-//    FixedVector(FixedVector<Type>&&) = delete;
-//
-//    FixedVector&
-//    operator=(FixedVector<Type>&&) = delete;
-//
-//    FixedVector&
-//    operator=(const FixedVector<Type>&) = delete;
-//
-//    bool is_placeholder() {
-//        return is_placeholder_;
-//    }
-// private:
-//    bool is_placeholder_ = true;
-//    int placeholder_size_ = 0;
-//};
-
 template <typename Type>
 using FixedVector = boost::container::vector<Type>;
 

internal/core/src/segcore/IndexingEntry.cpp

@@ -111,6 +111,8 @@ CreateIndex(const FieldMeta& field_meta, int64_t chunk_size) {
         }
     }
     switch (field_meta.get_data_type()) {
+        case DataType::BOOL:
+            return std::make_unique<ScalarIndexingEntry<bool>>(field_meta, chunk_size);
         case DataType::INT8:
             return std::make_unique<ScalarIndexingEntry<int8_t>>(field_meta, chunk_size);
         case DataType::INT16:

internal/core/src/segcore/InsertRecord.cpp

@@ -27,6 +27,10 @@ InsertRecord::InsertRecord(const Schema& schema, int64_t chunk_size) : uids_(1),
             }
         }
         switch (field.get_data_type()) {
+            case DataType::BOOL: {
+                entity_vec_.emplace_back(std::make_shared<ConcurrentVector<bool>>(chunk_size));
+                break;
+            }
             case DataType::INT8: {
                 entity_vec_.emplace_back(std::make_shared<ConcurrentVector<int8_t>>(chunk_size));
                 break;

internal/kv/etcd/etcd_kv.go

@@ -68,18 +68,6 @@ func (kv *EtcdKV) Load(key string) (string, error) {
 	return string(resp.Kvs[0].Value), nil
 }
 
-func (kv *EtcdKV) GetCount(key string) (int64, error) {
-	key = path.Join(kv.rootPath, key)
-	ctx, cancel := context.WithTimeout(context.TODO(), RequestTimeout)
-	defer cancel()
-	resp, err := kv.client.Get(ctx, key)
-	if err != nil {
-		return -1, err
-	}
-
-	return resp.Count, nil
-}
-
 func (kv *EtcdKV) MultiLoad(keys []string) ([]string, error) {
 	ops := make([]clientv3.Op, 0, len(keys))
 	for _, keyLoad := range keys {

internal/querynode/flow_graph_filter_dm_node.go

@@ -70,10 +70,10 @@ func (fdmNode *filterDmNode) filterInvalidInsertMessage(msg *msgstream.InsertMsg
 		return msg
 	}
 
-	// If the last record is drop type, all insert requests are invalid.
-	if !records[len(records)-1].createOrDrop {
-		return nil
-	}
+	// TODO: If the last record is drop type, all insert requests are invalid.
+	//if !records[len(records)-1].createOrDrop {
+	//	return nil
+	//}
 
 	// Filter insert requests before last record.
 	if len(msg.RowIDs) != len(msg.Timestamps) || len(msg.RowIDs) != len(msg.RowData) {

internal/writenode/client/client.go

@@ -6,7 +6,6 @@ import (
 	"github.com/golang/protobuf/proto"
 	"go.etcd.io/etcd/clientv3"
 
-	"github.com/zilliztech/milvus-distributed/internal/errors"
 	"github.com/zilliztech/milvus-distributed/internal/kv"
 	etcdkv "github.com/zilliztech/milvus-distributed/internal/kv/etcd"
 	"github.com/zilliztech/milvus-distributed/internal/msgstream"
@@ -80,21 +79,6 @@ func (c *Client) DescribeSegment(segmentID UniqueID) (*SegmentDescription, error
 	}
 
 	key := c.kvPrefix + strconv.FormatInt(segmentID, 10)
-
-	etcdKV, ok := c.kvClient.(*etcdkv.EtcdKV)
-	if !ok {
-		return nil, errors.New("type assertion failed for etcd kv")
-	}
-	count, err := etcdKV.GetCount(key)
-	if err != nil {
-		return nil, err
-	}
-
-	if count <= 0 {
-		ret.IsClosed = false
-		return ret, nil
-	}
-
 	value, err := c.kvClient.Load(key)
 	if err != nil {
 		return ret, err

internal/writenode/flow_graph_filter_dm_node.go

@@ -81,10 +81,10 @@ func (fdmNode *filterDmNode) filterInvalidInsertMessage(msg *msgstream.InsertMsg
 		return msg
 	}
 
-	// If the last record is drop type, all insert requests are invalid.
-	if !records[len(records)-1].createOrDrop {
-		return nil
-	}
+	// TODO: If the last record is drop type, all insert requests are invalid.
+	//if !records[len(records)-1].createOrDrop {
+	//	return nil
+	//}
 
 	// Filter insert requests before last record.
 	if len(msg.RowIDs) != len(msg.Timestamps) || len(msg.RowIDs) != len(msg.RowData) {