// Copyright(C) 2019 - 2020 Zilliz.All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License

#include <boost/container/vector.hpp>
#include <boost/filesystem/operations.hpp>
#include <fmt/core.h>
#include <folly/FBVector.h>
#include <folly/ScopeGuard.h>
#include <gtest/gtest.h>
#include <nlohmann/json.hpp>
#include <stdint.h>
#include <stdlib.h>
#include <algorithm>
#include <chrono>
#include <cstddef>
#include <functional>
#include <limits>
#include <map>
#include <memory>
#include <string>
#include <thread>
#include <unordered_set>
#include <variant>
#include <vector>

#include "common/Consts.h"
#include "common/Tracer.h"
#include "common/TracerBase.h"
#include "common/Types.h"
#include "common/protobuf_utils.h"
#include "cachinglayer/LoadingOverheadTracker.h"
#include "gtest/gtest.h"
#include "index/HybridScalarIndex.h"
#include "index/Index.h"
#include "index/IndexFactory.h"
#include "index/IndexInfo.h"
#include "index/IndexStats.h"
#include "index/Meta.h"
#include "indexbuilder/IndexCreatorBase.h"
#include "indexbuilder/IndexFactory.h"
#include "milvus-storage/filesystem/fs.h"
#include "pb/common.pb.h"
#include "pb/schema.pb.h"
#include "segcore/memory_planner.h"
#include "segcore/storagev1translator/SealedIndexTranslator.h"
#include "storage/ChunkManager.h"
#include "storage/FileManager.h"
#include "storage/InsertData.h"
#include "storage/PayloadReader.h"
#include "storage/EntryStreamUtils.h"
#include "storage/ThreadPools.h"
#include "storage/Types.h"
#include "storage/Util.h"
#include "test_utils/Constants.h"

using namespace milvus::index;
using namespace milvus::indexbuilder;
using namespace milvus;
using namespace milvus::index;

TEST(HybridScalarIndexPlannerPolicy, ShouldUseOpDelegatesToInternalIndex) {
    HybridScalarIndex<int64_t> int_index(7);
    std::vector<int64_t> int_data{1, 2, 3, 4};
    int_index.Build(int_data.size(), int_data.data());
    EXPECT_FALSE(int_index.ShouldUseOp(proto::plan::OpType::Match));
    EXPECT_TRUE(int_index.ShouldUseOp(proto::plan::OpType::Equal));

    HybridScalarIndex<std::string> string_index(7);
    std::vector<std::string> string_data{"alpha", "beta", "alphabet"};
    string_index.Build(string_data.size(), string_data.data());
    EXPECT_TRUE(string_index.ShouldUseOp(proto::plan::OpType::Match));
    EXPECT_TRUE(string_index.ShouldUseOp(proto::plan::OpType::PrefixMatch));
    EXPECT_TRUE(string_index.ShouldUseOp(proto::plan::OpType::RegexMatch));
    EXPECT_TRUE(string_index.ShouldUseOp(proto::plan::OpType::Equal));
}

template <typename T>
static std::vector<T>
GenerateData(const size_t size, const size_t cardinality) {
    std::vector<T> result;
    for (size_t i = 0; i < size; ++i) {
        result.push_back(rand() % cardinality);
    }
    return result;
}

template <>
std::vector<std::string>
GenerateData<std::string>(const size_t size, const size_t cardinality) {
    std::vector<std::string> result;
    for (size_t i = 0; i < size; ++i) {
        result.push_back(std::to_string(rand() % cardinality));
    }
    return result;
}

template <typename T>
class HybridIndexTestV1 : public testing::Test {
 protected:
    void
    Init(int64_t collection_id,
         int64_t partition_id,
         int64_t segment_id,
         int64_t field_id,
         int64_t index_build_id,
         int64_t index_version) {
        proto::schema::FieldSchema field_schema;
        field_schema.set_nullable(nullable_);
        if (has_default_value_) {
            auto default_value = field_schema.mutable_default_value();
            if constexpr (std::is_same_v<int8_t, T> ||
                          std::is_same_v<int16_t, T> ||
                          std::is_same_v<int32_t, T>) {
                default_value->set_int_data(10);
            } else if constexpr (std::is_same_v<int64_t, T>) {
                default_value->set_long_data(10);
            } else if constexpr (std::is_same_v<float, T>) {
                default_value->set_float_data(10);
            } else if constexpr (std::is_same_v<double, T>) {
                default_value->set_double_data(10);
            } else if constexpr (std::is_same_v<std::string, T>) {
                default_value->set_string_data("10");
            }
        }
        if constexpr (std::is_same_v<int8_t, T>) {
            field_schema.set_data_type(proto::schema::DataType::Int8);
        } else if constexpr (std::is_same_v<int16_t, T>) {
            field_schema.set_data_type(proto::schema::DataType::Int16);
        } else if constexpr (std::is_same_v<int32_t, T>) {
            field_schema.set_data_type(proto::schema::DataType::Int32);
        } else if constexpr (std::is_same_v<int64_t, T>) {
            field_schema.set_data_type(proto::schema::DataType::Int64);
        } else if constexpr (std::is_same_v<float, T>) {
            field_schema.set_data_type(proto::schema::DataType::Float);
        } else if constexpr (std::is_same_v<double, T>) {
            field_schema.set_data_type(proto::schema::DataType::Double);
        } else if constexpr (std::is_same_v<std::string, T>) {
            field_schema.set_data_type(proto::schema::DataType::String);
        }
        auto field_meta = storage::FieldDataMeta{
            collection_id, partition_id, segment_id, field_id, field_schema};
        auto index_meta = storage::IndexMeta{
            segment_id, field_id, index_build_id, index_version};
        field_meta_ = field_meta;
        index_meta_ = index_meta;

        std::vector<T> data_gen;
        data_gen = GenerateData<T>(nb_, cardinality_);
        for (auto x : data_gen) {
            data_.push_back(x);
        }

        auto field_data =
            storage::CreateFieldData(type_, DataType::NONE, nullable_);
        if (nullable_) {
            valid_data_.reserve(nb_);
            uint8_t* ptr = new uint8_t[(nb_ + 7) / 8];
            for (int i = 0; i < nb_; i++) {
                int byteIndex = i / 8;
                int bitIndex = i % 8;
                if (i % 2 == 0) {
                    valid_data_.push_back(true);
                    ptr[byteIndex] |= (1 << bitIndex);
                } else {
                    valid_data_.push_back(false);
                    ptr[byteIndex] &= ~(1 << bitIndex);
                }
            }
            field_data->FillFieldData(data_.data(), ptr, data_.size(), 0);
            delete[] ptr;
        } else {
            field_data->FillFieldData(data_.data(), data_.size());
        }
        auto payload_reader =
            std::make_shared<milvus::storage::PayloadReader>(field_data);
        storage::InsertData insert_data(payload_reader);
        insert_data.SetFieldDataMeta(field_meta);
        insert_data.SetTimestamps(0, 100);

        auto serialized_bytes = insert_data.Serialize(storage::Remote);

        auto log_path = fmt::format("/{}/{}/{}/{}/{}/{}",
                                    TestLocalPath,
                                    collection_id,
                                    partition_id,
                                    segment_id,
                                    field_id,
                                    0);
        chunk_manager_->Write(
            log_path, serialized_bytes.data(), serialized_bytes.size());

        storage::FileManagerContext ctx(
            field_meta, index_meta, chunk_manager_, fs_);
        std::vector<std::string> index_files;

        Config config;
        config["index_type"] = milvus::index::HYBRID_INDEX_TYPE;
        config[INSERT_FILES_KEY] = std::vector<std::string>{log_path};
        config["bitmap_cardinality_limit"] = "1000";
        if (!hybrid_high_cardinality_index_type_.empty()) {
            config[milvus::index::HYBRID_HIGH_CARDINALITY_INDEX_TYPE] =
                hybrid_high_cardinality_index_type_;
        }
        config[INDEX_NUM_ROWS_KEY] = nb_;
        config[milvus::index::SCALAR_INDEX_ENGINE_VERSION] = 3;
        if (has_lack_binlog_row_) {
            config[INDEX_NUM_ROWS_KEY] = nb_ + lack_binlog_row_;
        }

        {
            auto build_index =
                indexbuilder::IndexFactory::GetInstance().CreateIndex(
                    type_, config, ctx);
            build_index->Build();

            auto create_index_result = build_index->Upload();
            auto memSize = create_index_result->GetMemSize();
            auto serializedSize = create_index_result->GetSerializedSize();
            ASSERT_GT(memSize, 0);
            ASSERT_GT(serializedSize, 0);
            index_files = create_index_result->GetIndexFiles();
            index_files_ = index_files;
        }

        index::CreateIndexInfo index_info{};
        index_info.index_type = milvus::index::HYBRID_INDEX_TYPE;
        index_info.field_type = type_;

        config["index_files"] = index_files;
        config[milvus::LOAD_PRIORITY] =
            milvus::proto::common::LoadPriority::HIGH;
        ctx.set_for_loading_index(true);
        index_ =
            index::IndexFactory::GetInstance().CreateIndex(index_info, ctx);
        index_->LoadUnified(config);
    }

    virtual void
    SetParam() {
        nb_ = 10000;
        cardinality_ = 30;
        nullable_ = false;
        index_version_ = 1001;
        index_build_id_ = 1001;
    }
    void
    SetUp() override {
        SetParam();

        if constexpr (std::is_same_v<T, int8_t>) {
            type_ = DataType::INT8;
        } else if constexpr (std::is_same_v<T, int16_t>) {
            type_ = DataType::INT16;
        } else if constexpr (std::is_same_v<T, int32_t>) {
            type_ = DataType::INT32;
        } else if constexpr (std::is_same_v<T, int64_t>) {
            type_ = DataType::INT64;
        } else if constexpr (std::is_same_v<T, std::string>) {
            type_ = DataType::VARCHAR;
        }
        int64_t collection_id = 1;
        int64_t partition_id = 2;
        int64_t segment_id = 3;
        int64_t field_id = 101;
        std::string root_path = TestLocalPath;

        storage::StorageConfig storage_config;
        storage_config.storage_type = "local";
        storage_config.root_path = root_path;
        chunk_manager_ = storage::CreateChunkManager(storage_config);
        fs_ = storage::InitArrowFileSystem(storage_config);

        Init(collection_id,
             partition_id,
             segment_id,
             field_id,
             index_build_id_,
             index_version_);
    }

    void
    TearDown() override {
        CleanupLocalRoot(/*report_error=*/true);
    }

    virtual ~HybridIndexTestV1() override {
        CleanupLocalRoot(/*report_error=*/false);
    }

 public:
    void
    TestInFunc() {
        boost::container::vector<T> test_data;
        std::unordered_set<T> s;
        size_t nq = 10;
        for (size_t i = 0; i < nq; i++) {
            test_data.push_back(data_[i]);
            s.insert(data_[i]);
        }
        auto index_ptr =
            dynamic_cast<index::HybridScalarIndex<T>*>(index_.get());
        auto bitset = index_ptr->In(test_data.size(), test_data.data());
        size_t start = 0;
        if (has_lack_binlog_row_) {
            for (int i = 0; i < lack_binlog_row_; i++) {
                if (!has_default_value_) {
                    ASSERT_EQ(bitset[i], false);
                } else {
                    if constexpr (std::is_same_v<std::string, T>) {
                        ASSERT_EQ(bitset[i], s.find("10") != s.end());
                    } else {
                        ASSERT_EQ(bitset[i], s.find(10) != s.end());
                    }
                }
            }
            start += lack_binlog_row_;
        }
        for (size_t i = start; i < bitset.size(); i++) {
            if (nullable_ && !valid_data_[i - start]) {
                ASSERT_EQ(bitset[i], false);
            } else {
                ASSERT_EQ(bitset[i], s.find(data_[i - start]) != s.end());
            }
        }
    }

    void
    TestNotInFunc() {
        boost::container::vector<T> test_data;
        std::unordered_set<T> s;
        size_t nq = 10;
        for (size_t i = 0; i < nq; i++) {
            test_data.push_back(data_[i]);
            s.insert(data_[i]);
        }
        auto index_ptr =
            dynamic_cast<index::HybridScalarIndex<T>*>(index_.get());
        auto bitset = index_ptr->NotIn(test_data.size(), test_data.data());
        size_t start = 0;
        if (has_lack_binlog_row_) {
            for (int i = 0; i < lack_binlog_row_; i++) {
                if (!has_default_value_) {
                    ASSERT_EQ(bitset[i], false);
                } else {
                    if constexpr (std::is_same_v<std::string, T>) {
                        ASSERT_EQ(bitset[i], s.find("10") == s.end());
                    } else {
                        ASSERT_EQ(bitset[i], s.find(10) == s.end());
                    }
                }
            }
            start += lack_binlog_row_;
        }
        for (size_t i = start; i < bitset.size(); i++) {
            if (nullable_ && !valid_data_[i - start]) {
                ASSERT_EQ(bitset[i], false);
            } else {
                ASSERT_NE(bitset[i], s.find(data_[i - start]) != s.end());
            }
        }
    }

    void
    TestIsNullFunc() {
        auto index_ptr =
            dynamic_cast<index::HybridScalarIndex<T>*>(index_.get());
        auto bitset = index_ptr->IsNull();
        size_t start = 0;
        if (has_lack_binlog_row_) {
            for (int i = 0; i < lack_binlog_row_; i++) {
                if (has_default_value_) {
                    ASSERT_EQ(bitset[i], false);
                } else {
                    ASSERT_EQ(bitset[i], true);
                }
            }
            start += lack_binlog_row_;
        }
        for (size_t i = start; i < bitset.size(); i++) {
            if (nullable_ && !valid_data_[i - start]) {
                ASSERT_EQ(bitset[i], true);
            } else {
                ASSERT_EQ(bitset[i], false);
            }
        }
    }

    void
    TestIsNotNullFunc() {
        auto index_ptr =
            dynamic_cast<index::HybridScalarIndex<T>*>(index_.get());
        auto bitset = index_ptr->IsNotNull();
        size_t start = 0;
        if (has_lack_binlog_row_) {
            for (int i = 0; i < lack_binlog_row_; i++) {
                if (has_default_value_) {
                    ASSERT_EQ(bitset[i], true);
                } else {
                    ASSERT_EQ(bitset[i], false);
                }
            }
            start += lack_binlog_row_;
        }
        for (size_t i = start; i < bitset.size(); i++) {
            if (nullable_ && !valid_data_[i - start]) {
                ASSERT_EQ(bitset[i], false);
            } else {
                ASSERT_EQ(bitset[i], true);
            }
        }
    }

    void
    TestCompareValueFunc() {
        if constexpr (!std::is_same_v<T, std::string>) {
            using RefFunc = std::function<bool(int64_t)>;
            std::vector<std::tuple<T, OpType, RefFunc, bool>> test_cases{
                {10,
                 OpType::GreaterThan,
                 [&](int64_t i) -> bool { return data_[i] > 10; },
                 false},
                {10,
                 OpType::GreaterEqual,
                 [&](int64_t i) -> bool { return data_[i] >= 10; },
                 true},
                {10,
                 OpType::LessThan,
                 [&](int64_t i) -> bool { return data_[i] < 10; },
                 false},
                {10,
                 OpType::LessEqual,
                 [&](int64_t i) -> bool { return data_[i] <= 10; },
                 true},
            };
            for (const auto& [test_value, op, ref, default_value_res] :
                 test_cases) {
                auto index_ptr =
                    dynamic_cast<index::HybridScalarIndex<T>*>(index_.get());
                auto bitset = index_ptr->Range(test_value, op);
                size_t start = 0;
                if (has_lack_binlog_row_) {
                    for (int i = 0; i < lack_binlog_row_; i++) {
                        if (has_default_value_) {
                            ASSERT_EQ(bitset[i], default_value_res);
                        } else {
                            ASSERT_EQ(bitset[i], false);
                        }
                    }
                    start += lack_binlog_row_;
                }
                for (size_t i = start; i < bitset.size(); i++) {
                    auto ans = bitset[i];
                    auto should = ref(i - start);
                    if (nullable_ && !valid_data_[i - start]) {
                        ASSERT_EQ(ans, false)
                            << "op: " << op << ", @" << i << ", ans: " << ans
                            << ", ref: " << should;
                    } else {
                        ASSERT_EQ(ans, should)
                            << "op: " << op << ", @" << i << ", ans: " << ans
                            << ", ref: " << should;
                    }
                }
            }
        }
    }

    void
    TestRangeCompareFunc() {
        if constexpr (!std::is_same_v<T, std::string>) {
            using RefFunc = std::function<bool(int64_t)>;
            struct TestParam {
                int64_t lower_val;
                int64_t upper_val;
                bool lower_inclusive;
                bool upper_inclusive;
                RefFunc ref;
                bool default_value_res;
            };
            std::vector<TestParam> test_cases = {
                {
                    10,
                    30,
                    false,
                    false,
                    [&](int64_t i) { return 10 < data_[i] && data_[i] < 30; },
                    false,
                },
                {
                    10,
                    30,
                    true,
                    false,
                    [&](int64_t i) { return 10 <= data_[i] && data_[i] < 30; },
                    true,
                },
                {
                    10,
                    30,
                    true,
                    true,
                    [&](int64_t i) { return 10 <= data_[i] && data_[i] <= 30; },
                    true,
                },
                {
                    10,
                    30,
                    false,
                    true,
                    [&](int64_t i) { return 10 < data_[i] && data_[i] <= 30; },
                    false,
                }};

            for (const auto& test_case : test_cases) {
                auto index_ptr =
                    dynamic_cast<index::HybridScalarIndex<T>*>(index_.get());
                auto bitset = index_ptr->Range(test_case.lower_val,
                                               test_case.lower_inclusive,
                                               test_case.upper_val,
                                               test_case.upper_inclusive);
                size_t start = 0;
                if (has_lack_binlog_row_) {
                    for (int i = 0; i < lack_binlog_row_; i++) {
                        if (has_default_value_) {
                            ASSERT_EQ(bitset[i], test_case.default_value_res);
                        } else {
                            ASSERT_EQ(bitset[i], false);
                        }
                    }
                    start += lack_binlog_row_;
                }
                for (size_t i = start; i < bitset.size(); i++) {
                    auto ans = bitset[i];
                    auto should = test_case.ref(i - start);
                    if (nullable_ && !valid_data_[i - start]) {
                        ASSERT_EQ(ans, false)
                            << "lower:" << test_case.lower_val
                            << "upper:" << test_case.upper_val << ", @" << i
                            << ", ans: " << ans << ", ref: " << false;
                    } else {
                        ASSERT_EQ(ans, should)
                            << "lower:" << test_case.lower_val
                            << "upper:" << test_case.upper_val << ", @" << i
                            << ", ans: " << ans << ", ref: " << should;
                    }
                }
            }
        }
    }

 public:
    IndexBasePtr index_;
    DataType type_;
    size_t nb_;
    size_t cardinality_;
    boost::container::vector<T> data_;
    std::shared_ptr<storage::ChunkManager> chunk_manager_;
    milvus_storage::ArrowFileSystemPtr fs_;
    storage::FieldDataMeta field_meta_;
    storage::IndexMeta index_meta_;
    std::vector<std::string> index_files_;
    bool nullable_;
    FixedVector<bool> valid_data_;
    int index_build_id_;
    int index_version_;
    bool has_default_value_{false};
    bool has_lack_binlog_row_{false};
    size_t lack_binlog_row_{100};
    std::string hybrid_high_cardinality_index_type_;

 private:
    void
    CleanupLocalRoot(bool report_error) {
        index_.reset();
        if (chunk_manager_ == nullptr) {
            return;
        }

        const auto root_path = chunk_manager_->GetRootPath();
        boost::system::error_code ec;
        for (int attempt = 0; attempt < 3; ++attempt) {
            ec.clear();
            boost::filesystem::remove_all(root_path, ec);
            if (!ec) {
                return;
            }

            boost::system::error_code exists_ec;
            if (!boost::filesystem::exists(root_path, exists_ec) &&
                !exists_ec) {
                return;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(50));
        }

        if (report_error) {
            ADD_FAILURE() << "failed to remove hybrid index test root "
                          << root_path << ": " << ec.message();
        }
    }
};

TYPED_TEST_SUITE_P(HybridIndexTestV1);

TYPED_TEST_P(HybridIndexTestV1, CountFuncTest) {
    auto count = this->index_->Count();
    EXPECT_EQ(count, this->nb_);
}

TYPED_TEST_P(HybridIndexTestV1, ResourceEstimateUsesInternalIndexType) {
    constexpr uint64_t index_size = 1024;
    std::map<std::string, std::string> index_params{
        {"index_type", milvus::index::HYBRID_INDEX_TYPE},
        {milvus::index::SCALAR_INDEX_ENGINE_VERSION, "3"}};

    storage::FileManagerContext ctx(
        this->field_meta_, this->index_meta_, this->chunk_manager_, this->fs_);
    ctx.set_for_loading_index(true);

    auto request = index::IndexFactory::GetInstance().ScalarIndexLoadResource(
        this->type_,
        0,
        index_size,
        index_params,
        false,
        this->nb_,
        this->index_files_,
        ctx);

    EXPECT_EQ(request.final_memory_cost, index_size);
    EXPECT_EQ(request.final_disk_cost, 0);
    EXPECT_EQ(request.max_memory_cost, 2 * index_size);
    EXPECT_EQ(request.max_disk_cost, 0);
    EXPECT_FALSE(request.has_raw_data);
}

TYPED_TEST_P(HybridIndexTestV1, INFuncTest) {
    this->TestInFunc();
}

TYPED_TEST_P(HybridIndexTestV1, NotINFuncTest) {
    this->TestNotInFunc();
}

TYPED_TEST_P(HybridIndexTestV1, IsNullFuncTest) {
    this->TestIsNullFunc();
}

TYPED_TEST_P(HybridIndexTestV1, IsNotNullFuncTest) {
    this->TestIsNotNullFunc();
}

TYPED_TEST_P(HybridIndexTestV1, CompareValFuncTest) {
    this->TestCompareValueFunc();
}

TYPED_TEST_P(HybridIndexTestV1, TestRangeCompareFuncTest) {
    this->TestRangeCompareFunc();
}

using BitmapType =
    testing::Types<int8_t, int16_t, int32_t, int64_t, std::string>;
using InvertedType = testing::Types<int16_t, int32_t, int64_t, std::string>;

REGISTER_TYPED_TEST_SUITE_P(HybridIndexTestV1,
                            CountFuncTest,
                            ResourceEstimateUsesInternalIndexType,
                            INFuncTest,
                            IsNullFuncTest,
                            IsNotNullFuncTest,
                            NotINFuncTest,
                            CompareValFuncTest,
                            TestRangeCompareFuncTest);

INSTANTIATE_TYPED_TEST_SUITE_P(HybridIndexE2ECheck_LowCardinality,
                               HybridIndexTestV1,
                               BitmapType);

template <typename T>
class HybridIndexTestV2 : public HybridIndexTestV1<T> {
 public:
    virtual void
    SetParam() override {
        this->nb_ = 10000;
        this->cardinality_ = 2000;
        this->nullable_ = false;
        this->index_version_ = 1002;
        this->index_build_id_ = 1002;
    }

    virtual ~HybridIndexTestV2() {
    }
};

TYPED_TEST_SUITE_P(HybridIndexTestV2);

TYPED_TEST_P(HybridIndexTestV2, CountFuncTest) {
    auto count = this->index_->Count();
    EXPECT_EQ(count, this->nb_);
}

TYPED_TEST_P(HybridIndexTestV2, INFuncTest) {
    this->TestInFunc();
}

TYPED_TEST_P(HybridIndexTestV2, NotINFuncTest) {
    this->TestNotInFunc();
}

TYPED_TEST_P(HybridIndexTestV2, IsNullFuncTest) {
    this->TestIsNullFunc();
}

TYPED_TEST_P(HybridIndexTestV2, IsNotNullFuncTest) {
    this->TestIsNotNullFunc();
}

TYPED_TEST_P(HybridIndexTestV2, CompareValFuncTest) {
    this->TestCompareValueFunc();
}

TYPED_TEST_P(HybridIndexTestV2, TestRangeCompareFuncTest) {
    this->TestRangeCompareFunc();
}

template <typename T>
class HybridIndexTestInverted : public HybridIndexTestV1<T> {
 public:
    virtual void
    SetParam() override {
        this->nb_ = 10000;
        this->cardinality_ = 2000;
        this->nullable_ = false;
        this->index_version_ = 1005;
        this->index_build_id_ = 1005;
        this->hybrid_high_cardinality_index_type_ = "INVERTED";
    }

    virtual ~HybridIndexTestInverted() {
    }
};

TYPED_TEST_SUITE_P(HybridIndexTestInverted);

TYPED_TEST_P(HybridIndexTestInverted,
             ResourceEstimateUsesInternalInvertedIndexType) {
    auto stream_budget = storage::EntryStreamMaxTransientBytes();
    auto index_size = static_cast<uint64_t>(stream_budget);
    if (stream_budget == std::numeric_limits<size_t>::max() ||
        index_size > std::numeric_limits<uint64_t>::max() -
                         storage::kTailMergeGrace - 1024) {
        index_size = storage::kTailMergeGrace + 1024;
    } else {
        index_size += 1024;
    }
    auto stream_overhead = static_cast<uint64_t>(
        std::min<size_t>(index_size, storage::EntryStreamMaxTransientBytes()));
    std::map<std::string, std::string> index_params{
        {"index_type", milvus::index::HYBRID_INDEX_TYPE},
        {milvus::index::SCALAR_INDEX_ENGINE_VERSION, "3"}};

    storage::FileManagerContext ctx(
        this->field_meta_, this->index_meta_, this->chunk_manager_, this->fs_);
    ctx.set_for_loading_index(true);

    auto request = index::IndexFactory::GetInstance().ScalarIndexLoadResource(
        this->type_,
        0,
        index_size,
        index_params,
        false,
        this->nb_,
        this->index_files_,
        ctx);

    EXPECT_EQ(request.final_memory_cost, 0);
    EXPECT_EQ(request.final_disk_cost, index_size);
    EXPECT_EQ(request.max_memory_cost, stream_overhead);
    EXPECT_EQ(request.max_disk_cost, index_size);
    EXPECT_FALSE(request.has_raw_data);
}

TYPED_TEST_P(HybridIndexTestInverted,
             ScalarIndexLoadingOverheadDoesNotCapFileDimension) {
    auto& budget = storage::TransientMemoryBudget::GetLoadTransientBudget();
    auto old_capacity = budget.CapacityBytes();
    auto cleanup = folly::makeGuard(
        [&budget, old_capacity]() { budget.SetCapacityBytes(old_capacity); });
    budget.SetCapacityBytes(0);

    std::map<std::string, std::string> index_params{
        {"index_type", milvus::index::HYBRID_INDEX_TYPE},
        {milvus::index::SCALAR_INDEX_ENGINE_VERSION, "3"}};
    milvus::segcore::LoadIndexInfo load_info{};
    load_info.collection_id = 1;
    load_info.partition_id = 2;
    load_info.segment_id = 3;
    load_info.field_id = 101;
    load_info.field_type = this->type_;
    load_info.element_type = DataType::NONE;
    load_info.enable_mmap = false;
    load_info.index_id = this->index_build_id_;
    load_info.index_build_id = this->index_build_id_;
    load_info.index_version = this->index_version_;
    load_info.index_params = index_params;
    load_info.index_files = this->index_files_;
    load_info.index_engine_version = this->index_version_;
    load_info.index_size = 1024;
    load_info.num_rows = this->nb_;
    load_info.dim = 0;

    index::CreateIndexInfo index_info{};
    index_info.index_type = milvus::index::HYBRID_INDEX_TYPE;
    index_info.field_type = this->type_;
    index_info.index_engine_version = this->index_version_;

    storage::FileManagerContext ctx(
        this->field_meta_, this->index_meta_, this->chunk_manager_, this->fs_);
    ctx.set_for_loading_index(true);

    Config config = index_params;
    milvus::segcore::storagev1translator::SealedIndexTranslator translator(
        index_info,
        &load_info,
        milvus::tracer::TraceContext{},
        ctx,
        std::move(config));

    ASSERT_TRUE(translator.meta()->loading_overhead.has_value());
    EXPECT_EQ(translator.meta()->loading_overhead->group,
              milvus::segcore::kLoadTransientOverheadGroup);
    EXPECT_EQ(
        translator.meta()->loading_overhead->upper_bound.memory_bytes,
        milvus::cachinglayer::LoadingOverheadTracker::kUnlimited.memory_bytes);
    EXPECT_EQ(translator.meta()->loading_overhead->upper_bound.file_bytes, 0);
}

template <typename T>
class HybridIndexTestNullable : public HybridIndexTestV1<T> {
 public:
    virtual void
    SetParam() override {
        this->nb_ = 10000;
        this->cardinality_ = 2000;
        this->nullable_ = true;
        this->index_version_ = 1003;
        this->index_build_id_ = 1003;
    }

    virtual ~HybridIndexTestNullable() {
    }
};

TYPED_TEST_SUITE_P(HybridIndexTestNullable);

TYPED_TEST_P(HybridIndexTestNullable, CountFuncTest) {
    auto count = this->index_->Count();
    EXPECT_EQ(count, this->nb_);
}

TYPED_TEST_P(HybridIndexTestNullable, INFuncTest) {
    this->TestInFunc();
}

TYPED_TEST_P(HybridIndexTestNullable, NotINFuncTest) {
    this->TestNotInFunc();
}

TYPED_TEST_P(HybridIndexTestNullable, IsNullFuncTest) {
    this->TestIsNullFunc();
}

TYPED_TEST_P(HybridIndexTestNullable, IsNotNullFuncTest) {
    this->TestIsNotNullFunc();
}

TYPED_TEST_P(HybridIndexTestNullable, CompareValFuncTest) {
    this->TestCompareValueFunc();
}

TYPED_TEST_P(HybridIndexTestNullable, TestRangeCompareFuncTest) {
    this->TestRangeCompareFunc();
}

template <typename T>
class HybridIndexTestV3 : public HybridIndexTestV1<T> {
 public:
    virtual void
    SetParam() override {
        this->nb_ = 10000;
        this->cardinality_ = 2000;
        this->nullable_ = true;
        this->index_version_ = 1003;
        this->index_build_id_ = 1003;
        this->has_default_value_ = false;
        this->has_lack_binlog_row_ = true;
        this->lack_binlog_row_ = 100;
    }

    virtual ~HybridIndexTestV3() {
    }
};

TYPED_TEST_SUITE_P(HybridIndexTestV3);

TYPED_TEST_P(HybridIndexTestV3, CountFuncTest) {
    auto count = this->index_->Count();
    if (this->has_lack_binlog_row_) {
        EXPECT_EQ(count, this->nb_ + this->lack_binlog_row_);
    } else {
        EXPECT_EQ(count, this->nb_);
    }
}

TYPED_TEST_P(HybridIndexTestV3, INFuncTest) {
    this->TestInFunc();
}

TYPED_TEST_P(HybridIndexTestV3, NotINFuncTest) {
    this->TestNotInFunc();
}

TYPED_TEST_P(HybridIndexTestV3, IsNullFuncTest) {
    this->TestIsNullFunc();
}

TYPED_TEST_P(HybridIndexTestV3, IsNotNullFuncTest) {
    this->TestIsNotNullFunc();
}

TYPED_TEST_P(HybridIndexTestV3, CompareValFuncTest) {
    this->TestCompareValueFunc();
}

TYPED_TEST_P(HybridIndexTestV3, TestRangeCompareFuncTest) {
    this->TestRangeCompareFunc();
}

template <typename T>
class HybridIndexTestV4 : public HybridIndexTestV1<T> {
 public:
    virtual void
    SetParam() override {
        this->nb_ = 10000;
        this->cardinality_ = 2000;
        this->nullable_ = true;
        this->index_version_ = 1003;
        this->index_build_id_ = 1003;
        this->has_default_value_ = true;
        this->has_lack_binlog_row_ = true;
        this->lack_binlog_row_ = 100;
    }

    virtual ~HybridIndexTestV4() {
    }
};

TYPED_TEST_SUITE_P(HybridIndexTestV4);

TYPED_TEST_P(HybridIndexTestV4, CountFuncTest) {
    auto count = this->index_->Count();
    if (this->has_lack_binlog_row_) {
        EXPECT_EQ(count, this->nb_ + this->lack_binlog_row_);
    } else {
        EXPECT_EQ(count, this->nb_);
    }
}

TYPED_TEST_P(HybridIndexTestV4, INFuncTest) {
    this->TestInFunc();
}

TYPED_TEST_P(HybridIndexTestV4, NotINFuncTest) {
    this->TestNotInFunc();
}

TYPED_TEST_P(HybridIndexTestV4, IsNullFuncTest) {
    this->TestIsNullFunc();
}

TYPED_TEST_P(HybridIndexTestV4, IsNotNullFuncTest) {
    this->TestIsNotNullFunc();
}

TYPED_TEST_P(HybridIndexTestV4, CompareValFuncTest) {
    this->TestCompareValueFunc();
}

TYPED_TEST_P(HybridIndexTestV4, TestRangeCompareFuncTest) {
    this->TestRangeCompareFunc();
}

using BitmapType =
    testing::Types<int8_t, int16_t, int32_t, int64_t, std::string>;

REGISTER_TYPED_TEST_SUITE_P(HybridIndexTestV2,
                            CountFuncTest,
                            INFuncTest,
                            IsNullFuncTest,
                            IsNotNullFuncTest,
                            NotINFuncTest,
                            CompareValFuncTest,
                            TestRangeCompareFuncTest);

REGISTER_TYPED_TEST_SUITE_P(HybridIndexTestNullable,
                            CountFuncTest,
                            INFuncTest,
                            IsNullFuncTest,
                            IsNotNullFuncTest,
                            NotINFuncTest,
                            CompareValFuncTest,
                            TestRangeCompareFuncTest);

REGISTER_TYPED_TEST_SUITE_P(HybridIndexTestV3,
                            CountFuncTest,
                            INFuncTest,
                            IsNullFuncTest,
                            IsNotNullFuncTest,
                            NotINFuncTest,
                            CompareValFuncTest,
                            TestRangeCompareFuncTest);

REGISTER_TYPED_TEST_SUITE_P(HybridIndexTestV4,
                            CountFuncTest,
                            INFuncTest,
                            IsNullFuncTest,
                            IsNotNullFuncTest,
                            NotINFuncTest,
                            CompareValFuncTest,
                            TestRangeCompareFuncTest);

REGISTER_TYPED_TEST_SUITE_P(HybridIndexTestInverted,
                            ResourceEstimateUsesInternalInvertedIndexType,
                            ScalarIndexLoadingOverheadDoesNotCapFileDimension);

INSTANTIATE_TYPED_TEST_SUITE_P(HybridIndexE2ECheck_HighCardinality,
                               HybridIndexTestV2,
                               BitmapType);

INSTANTIATE_TYPED_TEST_SUITE_P(HybridIndexE2ECheck_Inverted,
                               HybridIndexTestInverted,
                               InvertedType);

INSTANTIATE_TYPED_TEST_SUITE_P(HybridIndexE2ECheck_Nullable,
                               HybridIndexTestNullable,
                               BitmapType);

INSTANTIATE_TYPED_TEST_SUITE_P(HybridIndexE2ECheck_HasLackNullBinlog,
                               HybridIndexTestV3,
                               BitmapType);

INSTANTIATE_TYPED_TEST_SUITE_P(HybridIndexE2ECheck_HasLackDefaultValueBinlog,
                               HybridIndexTestV4,
                               BitmapType);
