// 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 <gtest/gtest.h>
#include <nlohmann/detail/iterators/iteration_proxy.hpp>
#include <nlohmann/json.hpp>
#include <nlohmann/json_fwd.hpp>
#include <stdint.h>
#include <boost/filesystem.hpp>
#include <chrono>
#include <initializer_list>
#include <iostream>
#include <limits>
#include <map>
#include <memory>
#include <optional>
#include <ratio>
#include <string>
#include <thread>
#include <utility>
#include <vector>

#include "bitset/bitset.h"
#include "bitset/detail/element_vectorized.h"
#include "common/Consts.h"
#include "common/EasyAssert.h"
#include "common/FieldMeta.h"
#include "common/IndexMeta.h"
#include "common/Schema.h"
#include "common/Types.h"
#include "common/common_type_c.h"
#include "common/protobuf_utils.h"
#include "exec/expression/ExprCache.h"
#include "exec/expression/function/FunctionFactory.h"
#include "expr/ITypeExpr.h"
#include "filemanager/InputStream.h"
#include "gtest/gtest.h"
#include "index/Meta.h"
#include "index/TextMatchIndex.h"
#include "index/Utils.h"
#include "knowhere/comp/index_param.h"
#include "milvus-storage/lob_column/lob_column_manager.h"
#include "milvus-storage/lob_column/lob_column_writer.h"
#include "milvus-storage/lob_column/lob_reference.h"
#include "pb/plan.pb.h"
#include "pb/schema.pb.h"
#include "plan/PlanNode.h"
#include "query/ExecPlanNodeVisitor.h"
#include "query/PlanNode.h"
#include "query/PlanProto.h"
#include "segcore/ChunkedSegmentSealedImpl.h"
#include "segcore/SegcoreConfig.h"
#include "segcore/SegmentGrowing.h"
#include "segcore/SegmentGrowingImpl.h"
#include "segcore/SegmentSealed.h"
#include "segcore/default_fs.h"
#include "segcore/segment_c.h"
#include "storage/FileManager.h"
#include "storage/Util.h"
#include "storage/loon_ffi/property_singleton.h"
#include "test_utils/DataGen.h"
#include "test_utils/GenExprProto.h"
#include "test_utils/storage_test_utils.h"

using namespace milvus;
using namespace milvus::query;
using namespace milvus::segcore;

namespace {
SchemaPtr
GenTestSchema(std::map<std::string, std::string> params = {},
              bool nullable = false) {
    auto schema = std::make_shared<Schema>();
    {
        FieldMeta f(FieldName("pk"),
                    FieldId(100),
                    DataType::INT64,
                    false,
                    std::nullopt);
        schema->AddField(std::move(f));
        schema->set_primary_field_id(FieldId(100));
    }
    {
        FieldMeta f(FieldName("str"),
                    FieldId(101),
                    DataType::VARCHAR,
                    65536,
                    nullable,
                    true,
                    true,
                    params,
                    std::nullopt);
        schema->AddField(std::move(f));
    }
    {
        FieldMeta f(FieldName("fvec"),
                    FieldId(102),
                    DataType::VECTOR_FLOAT,
                    16,
                    knowhere::metric::L2,
                    false,
                    std::nullopt);
        schema->AddField(std::move(f));
    }
    return schema;
}

storage::FileManagerContext
CreateTextMatchTestFileManagerContext(
    int64_t build_id,
    proto::schema::DataType data_type = proto::schema::DataType::VarChar) {
    auto storage_config = get_default_local_storage_config();
    auto chunk_manager = storage::CreateChunkManager(storage_config);
    auto fs = storage::InitArrowFileSystem(storage_config);

    storage::FieldDataMeta field_meta{1, 2, 3, 101};
    field_meta.field_schema.set_data_type(data_type);
    storage::IndexMeta index_meta{3, 101, build_id, 10000};
    return storage::FileManagerContext(
        field_meta, index_meta, chunk_manager, fs);
}

std::unique_ptr<index::TextMatchIndex>
BuildTextMatchIndexForUpload(const storage::FileManagerContext& ctx) {
    auto index = std::make_unique<index::TextMatchIndex>(
        ctx, index::TANTIVY_INDEX_LATEST_VERSION, "milvus_tokenizer", "{}", "");

    std::vector<std::string> texts = {
        "football basketball", "swimming football", "table tennis"};
    auto field_data =
        storage::CreateFieldData(DataType::VARCHAR, DataType::NONE, false);
    field_data->FillFieldData(texts.data(), texts.size());

    index->BuildIndexFromFieldData({field_data}, false);
    return index;
}

void
AssertTextMatchUploadReturnsRelativePaths(
    const std::vector<index::SerializedIndexFileInfo>& files) {
    ASSERT_FALSE(files.empty());
    for (const auto& file : files) {
        ASSERT_FALSE(file.file_name.empty());
        ASSERT_EQ(file.file_name.find(TestRemotePath), std::string::npos)
            << file.file_name;
        ASSERT_EQ(file.file_name.find(TEXT_LOG_ROOT_PATH), std::string::npos)
            << file.file_name;
        ASSERT_GT(file.file_size, 0);
    }
}

std::shared_ptr<milvus::plan::FilterBitsNode>
GetMatchExpr(SchemaPtr schema,
             const std::string& query,
             proto::plan::OpType op,
             int64_t slop = 0) {
    const auto& str_meta = schema->operator[](FieldName("str"));
    auto column_info = test::GenColumnInfo(str_meta.get_id().get(),
                                           proto::schema::DataType::VarChar,
                                           false,
                                           false);

    auto unary_range_expr = test::GenUnaryRangeExpr(op, query);
    unary_range_expr->set_allocated_column_info(column_info);
    auto generic_for_slop = milvus::test::GenGenericValue(slop);
    unary_range_expr->add_extra_values()->CopyFrom(*generic_for_slop);
    delete generic_for_slop;
    auto expr = test::GenExpr();
    expr->set_allocated_unary_range_expr(unary_range_expr);

    auto parser = ProtoParser(schema);
    auto typed_expr = parser.ParseExprs(*expr);
    auto parsed =
        std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, typed_expr);
    return parsed;
};

std::shared_ptr<milvus::plan::FilterBitsNode>
GetNotMatchExpr(SchemaPtr schema,
                const std::string& query,
                proto::plan::OpType op,
                int64_t slop = 0) {
    const auto& str_meta = schema->operator[](FieldName("str"));
    proto::plan::GenericValue val;
    val.set_string_val(query);
    std::vector<proto::plan::GenericValue> extra_values;
    auto generic_for_slop = milvus::test::GenGenericValue(slop);
    extra_values.push_back(*generic_for_slop);
    delete generic_for_slop;
    auto child_expr = std::make_shared<expr::UnaryRangeFilterExpr>(
        milvus::expr::ColumnInfo(str_meta.get_id(), DataType::VARCHAR),
        op,
        val,
        extra_values);
    auto expr = std::make_shared<expr::LogicalUnaryExpr>(
        expr::LogicalUnaryExpr::OpType::LogicalNot, child_expr);
    auto parsed =
        std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
    return parsed;
};

std::shared_ptr<milvus::plan::FilterBitsNode>
GetFuzzyMatchExpr(SchemaPtr schema,
                  const std::string& query,
                  int64_t max_edit_distance) {
    // For TextMatchFuzzy the single extra value carries the max edit distance,
    // reusing the slot phrase match uses for slop.
    return GetMatchExpr(
        schema, query, proto::plan::OpType::TextMatchFuzzy, max_edit_distance);
};

std::shared_ptr<milvus::plan::FilterBitsNode>
GetFuzzyMatchExprNoDistance(SchemaPtr schema, const std::string& query) {
    // A fuzzy plan with no extra value at all, to reach the executor's
    // "max_edit_distance is required" guard.
    const auto& str_meta = schema->operator[](FieldName("str"));
    auto column_info = test::GenColumnInfo(str_meta.get_id().get(),
                                           proto::schema::DataType::VarChar,
                                           false,
                                           false);
    auto unary_range_expr =
        test::GenUnaryRangeExpr(proto::plan::OpType::TextMatchFuzzy, query);
    unary_range_expr->set_allocated_column_info(column_info);
    auto expr = test::GenExpr();
    expr->set_allocated_unary_range_expr(unary_range_expr);
    auto parser = ProtoParser(schema);
    auto typed_expr = parser.ParseExprs(*expr);
    return std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
                                                  typed_expr);
};
}  // namespace

TEST(ParseJson, Naive) {
    {
        std::string s(R"({"tokenizer": "jieba"})");
        nlohmann::json j = nlohmann::json::parse(s);
        auto m = j.get<std::map<std::string, std::string>>();
        for (const auto& [k, v] : m) {
            std::cout << k << ": " << v << std::endl;
        }
    }

    {
        std::string s(
            R"({"analyzer":"stop","stop_words":["an","the"],"case_insensitive":false})");
        nlohmann::json j = nlohmann::json::parse(s);
        for (const auto& [key, value] : j.items()) {
            std::cout << key << ": " << value.dump() << std::endl;
        }
    }
}

TEST(ParseTokenizerParams, NoTokenizerParams) {
    TypeParams params{{"k", "v"}};
    auto p = ParseTokenizerParams(params);
    ASSERT_EQ("{}", std::string(p));
}

TEST(ParseTokenizerParams, Default) {
    TypeParams params{{"analyzer_params", R"({"tokenizer": "standard"})"}};
    auto p = ParseTokenizerParams(params);
    ASSERT_EQ(params.at("analyzer_params"), p);
}

TEST(TextMatch, Index) {
    using Index = index::TextMatchIndex;
    auto index = std::make_unique<Index>(std::numeric_limits<int64_t>::max(),
                                         "unique_id",
                                         "milvus_tokenizer",
                                         "{}",
                                         /*enable_background_merge=*/false);
    index->CreateReader(milvus::index::SetBitsetSealed);
    index->AddTextSealed("football, basketball, pingpang", true, 0);
    index->AddTextSealed("", false, 1);
    index->AddTextSealed("swimming, football", true, 2);
    index->Commit();
    index->Reload();

    {
        auto res = index->MatchQuery("football", 1);
        ASSERT_EQ(res.size(), 3);
        ASSERT_TRUE(res[0]);
        ASSERT_FALSE(res[1]);
        ASSERT_TRUE(res[2]);
        auto res1 = index->IsNull();
        ASSERT_FALSE(res1[0]);
        ASSERT_TRUE(res1[1]);
        ASSERT_FALSE(res1[2]);
        auto res2 = index->IsNotNull();
        ASSERT_TRUE(res2[0]);
        ASSERT_FALSE(res2[1]);
        ASSERT_TRUE(res2[2]);
        res = index->MatchQuery("nothing", 1);
        ASSERT_EQ(res.size(), 3);
        ASSERT_FALSE(res[0]);
        ASSERT_FALSE(res[1]);
        ASSERT_FALSE(res[2]);
        auto res3 = index->MatchQuery("football pingpang cricket", 2);
        ASSERT_EQ(res3.size(), 3);
        ASSERT_TRUE(res3[0]);
        ASSERT_FALSE(res3[1]);
        ASSERT_FALSE(res3[2]);
    }

    {
        auto res = index->PhraseMatchQuery("football", 0);
        ASSERT_EQ(res.size(), 3);
        ASSERT_TRUE(res[0]);
        ASSERT_FALSE(res[1]);
        ASSERT_TRUE(res[2]);

        auto res1 = index->PhraseMatchQuery("swimming football", 0);
        ASSERT_EQ(res1.size(), 3);
        ASSERT_FALSE(res1[0]);
        ASSERT_FALSE(res1[1]);
        ASSERT_TRUE(res1[2]);

        auto res2 = index->PhraseMatchQuery("football swimming", 0);
        ASSERT_EQ(res2.size(), 3);
        ASSERT_FALSE(res2[0]);
        ASSERT_FALSE(res2[1]);
        ASSERT_FALSE(res2[2]);

        auto res3 = index->PhraseMatchQuery("football swimming", 1);
        ASSERT_EQ(res3.size(), 3);
        ASSERT_FALSE(res3[0]);
        ASSERT_FALSE(res3[1]);
        ASSERT_FALSE(res3[2]);

        auto res4 = index->PhraseMatchQuery("football swimming", 2);
        ASSERT_EQ(res4.size(), 3);
        ASSERT_FALSE(res4[0]);
        ASSERT_FALSE(res4[1]);
        ASSERT_TRUE(res4[2]);
    }
}

TEST(TextMatch, FuzzyIndex) {
    using Index = index::TextMatchIndex;
    auto index = std::make_unique<Index>(std::numeric_limits<int64_t>::max(),
                                         "unique_id",
                                         "milvus_tokenizer",
                                         "{}",
                                         /*enable_background_merge=*/false);
    index->CreateReader(milvus::index::SetBitsetSealed);
    index->AddTextSealed("football, basketball, pingpang", true, 0);
    index->AddTextSealed("", false, 1);
    index->AddTextSealed("swimming, football", true, 2);
    index->Commit();
    index->Reload();

    {
        // "footbal" is one edit from "football", so distance 1 matches rows 0 and 2.
        auto res = index->FuzzyMatchQuery("footbal", 1);
        ASSERT_EQ(res.size(), 3);
        ASSERT_TRUE(res[0]);
        ASSERT_FALSE(res[1]);
        ASSERT_TRUE(res[2]);

        // the same typo under exact text match finds nothing: fuzzy really differs.
        auto exact = index->MatchQuery("footbal", 1);
        ASSERT_EQ(exact.size(), 3);
        ASSERT_FALSE(exact[0]);
        ASSERT_FALSE(exact[2]);

        // distance 0 is exact, so the typo still finds nothing.
        auto res0 = index->FuzzyMatchQuery("footbal", 0);
        ASSERT_EQ(res0.size(), 3);
        ASSERT_FALSE(res0[0]);
        ASSERT_FALSE(res0[2]);

        // "fotbal" is two edits from "football": excluded at distance 1, matched at distance 2.
        auto miss = index->FuzzyMatchQuery("fotbal", 1);
        ASSERT_EQ(miss.size(), 3);
        ASSERT_FALSE(miss[0]);
        ASSERT_FALSE(miss[2]);
        auto hit = index->FuzzyMatchQuery("fotbal", 2);
        ASSERT_EQ(hit.size(), 3);
        ASSERT_TRUE(hit[0]);
        ASSERT_FALSE(hit[1]);
        ASSERT_TRUE(hit[2]);
    }
}

TEST(TextMatch, UploadReturnsRelativeTextLogPaths) {
    auto ctx = CreateTextMatchTestFileManagerContext(1000);
    auto index = BuildTextMatchIndexForUpload(ctx);

    auto stats = index->Upload({});

    AssertTextMatchUploadReturnsRelativePaths(
        stats->GetSerializedIndexFileInfo());
}

TEST(TextMatch, UploadUnifiedReturnsRelativeTextLogPaths) {
    auto ctx = CreateTextMatchTestFileManagerContext(1001);
    auto index = BuildTextMatchIndexForUpload(ctx);

    auto stats = index->UploadUnified({});

    AssertTextMatchUploadReturnsRelativePaths(
        stats->GetSerializedIndexFileInfo());
    ASSERT_EQ(stats->GetSerializedIndexFileInfo().size(), 1);
    ASSERT_NE(stats->GetSerializedIndexFileInfo()[0].file_name.find(".v3"),
              std::string::npos);
}

// Regression test: BuildIndexFromFieldData with multiple FieldData batches
// and nullable fields must record correct global offsets in null_offset_.
// Previously, null_offset_.push_back(i) used the batch-local index instead
// of the global accumulated offset, causing wrong null tracking for all
// batches after the first.
TEST(TextMatch, BuildIndexFromFieldDataMultiBatchNullable) {
    using Index = index::TextMatchIndex;

    // Helper to create a nullable FieldData<std::string> batch.
    auto make_batch =
        [](const std::vector<std::string>& texts,
           const std::vector<bool>& valids) -> milvus::FieldDataPtr {
        auto fd = storage::CreateFieldData(
            DataType::VARCHAR, DataType::NONE, true, 1, texts.size());
        // Pack valid bits into uint8_t array (LSB first).
        size_t byte_count = (texts.size() + 7) / 8;
        std::vector<uint8_t> valid_bytes(byte_count, 0);
        for (size_t i = 0; i < valids.size(); i++) {
            if (valids[i]) {
                valid_bytes[i >> 3] |= (1u << (i & 7));
            }
        }
        fd->FillFieldData(texts.data(), valid_bytes.data(), texts.size(), 0);
        return fd;
    };

    // Three batches, nulls scattered across all batches.
    // Batch 0 (3 rows): "hello", NULL,  "world"
    // Batch 1 (3 rows): NULL,   "foo",  NULL
    // Batch 2 (2 rows): "bar",  NULL
    //
    // Global layout (8 rows total):
    //   offset 0: "hello" (valid)
    //   offset 1: ""      (null)
    //   offset 2: "world" (valid)
    //   offset 3: ""      (null)
    //   offset 4: "foo"   (valid)
    //   offset 5: ""      (null)
    //   offset 6: "bar"   (valid)
    //   offset 7: ""      (null)
    auto batch0 = make_batch({"hello", "", "world"}, {true, false, true});
    auto batch1 = make_batch({"", "foo", ""}, {false, true, false});
    auto batch2 = make_batch({"bar", ""}, {true, false});

    std::vector<milvus::FieldDataPtr> field_datas = {batch0, batch1, batch2};

    auto index = std::make_unique<Index>(200,
                                         "test_multi_batch",
                                         "milvus_tokenizer",
                                         "{}",
                                         /*enable_background_merge=*/true);
    index->CreateReader(milvus::index::SetBitsetGrowing);
    index->RegisterAnalyzer("milvus_tokenizer", "{}");

    index->BuildIndexFromFieldData(field_datas, true /* nullable */);
    index->Commit();
    index->Reload();

    // Verify null tracking: offsets 1, 3, 5, 7 should be null.
    {
        auto null_bits = index->IsNull();
        ASSERT_EQ(null_bits.size(), 8);
        ASSERT_FALSE(null_bits[0]);  // "hello"
        ASSERT_TRUE(null_bits[1]);   // null
        ASSERT_FALSE(null_bits[2]);  // "world"
        ASSERT_TRUE(null_bits[3]);   // null (batch 1, row 0)
        ASSERT_FALSE(null_bits[4]);  // "foo"
        ASSERT_TRUE(null_bits[5]);   // null (batch 1, row 2)
        ASSERT_FALSE(null_bits[6]);  // "bar"
        ASSERT_TRUE(null_bits[7]);   // null (batch 2, row 1)
    }

    // Verify not-null tracking.
    {
        auto not_null_bits = index->IsNotNull();
        ASSERT_EQ(not_null_bits.size(), 8);
        ASSERT_TRUE(not_null_bits[0]);
        ASSERT_FALSE(not_null_bits[1]);
        ASSERT_TRUE(not_null_bits[2]);
        ASSERT_FALSE(not_null_bits[3]);
        ASSERT_TRUE(not_null_bits[4]);
        ASSERT_FALSE(not_null_bits[5]);
        ASSERT_TRUE(not_null_bits[6]);
        ASSERT_FALSE(not_null_bits[7]);
    }

    // Verify text search still works on valid rows.
    {
        auto res = index->MatchQuery("hello", 1);
        ASSERT_EQ(res.size(), 8);
        ASSERT_TRUE(res[0]);
        for (int i = 1; i < 8; i++) {
            ASSERT_FALSE(res[i]) << "unexpected match at offset " << i;
        }
    }
    {
        auto res = index->MatchQuery("foo", 1);
        ASSERT_EQ(res.size(), 8);
        ASSERT_TRUE(res[4]);
        for (int i = 0; i < 8; i++) {
            if (i != 4) {
                ASSERT_FALSE(res[i]) << "unexpected match at offset " << i;
            }
        }
    }
    {
        auto res = index->MatchQuery("bar", 1);
        ASSERT_EQ(res.size(), 8);
        ASSERT_TRUE(res[6]);
        for (int i = 0; i < 8; i++) {
            if (i != 6) {
                ASSERT_FALSE(res[i]) << "unexpected match at offset " << i;
            }
        }
    }
}

TEST(TextMatch, BuildIndexFromTextFieldData) {
    auto ctx = CreateTextMatchTestFileManagerContext(
        1002, proto::schema::DataType::Text);
    auto index = std::make_unique<index::TextMatchIndex>(
        ctx, index::TANTIVY_INDEX_LATEST_VERSION, "milvus_tokenizer", "{}", "");

    std::vector<std::string> texts = {
        "football basketball", "swimming football", "table tennis"};
    auto field_data =
        storage::CreateFieldData(DataType::TEXT, DataType::NONE, false);
    field_data->FillFieldData(texts.data(), texts.size());

    ASSERT_NO_THROW(index->BuildIndexFromFieldData({field_data}, false));
}

TEST(TextMatch, SealedCreateTextIndexDecodesTextLobRefs) {
    constexpr int64_t collection_id = 10001;
    constexpr int64_t partition_id = 10002;
    constexpr int64_t segment_id = 10003;
    const FieldId text_field_id(101);
    const std::string unique_token = "zzlobuniqueterm";

    auto test_dir =
        "sealed_text_lob_index_" +
        std::to_string(
            std::chrono::steady_clock::now().time_since_epoch().count());
    boost::filesystem::remove_all(TestLocalPath + test_dir);

    auto schema = std::make_shared<Schema>();
    schema->AddField(FieldMeta(
        FieldName("pk"), FieldId(100), DataType::INT64, false, std::nullopt));
    schema->set_primary_field_id(FieldId(100));
    std::map<std::string, std::string> text_params = {
        {"enable_match", "true"},
        {"enable_analyzer", "true"},
        {"analyzer_params", R"({"tokenizer": "standard"})"},
    };
    schema->AddField(FieldMeta(FieldName("str"),
                               text_field_id,
                               DataType::TEXT,
                               65536,
                               false,
                               true,
                               true,
                               text_params,
                               std::nullopt));

    auto lob_base_path =
        test_dir + "/lobs/" + std::to_string(text_field_id.get());
    milvus_storage::lob_column::LobColumnConfig lob_config;
    lob_config.lob_base_path = lob_base_path;
    lob_config.field_id = text_field_id.get();
    lob_config.inline_threshold = 1;
    lob_config.max_lob_file_bytes = 256 * 1024;
    lob_config.flush_threshold_bytes = 64 * 1024;
    auto properties = milvus::storage::LoonFFIPropertiesSingleton::GetInstance()
                          .GetProperties();
    ASSERT_NE(properties, nullptr);
    lob_config.properties = *properties;

    auto fs = milvus::segcore::GetDefaultArrowFileSystem();
    auto manager_result =
        milvus_storage::lob_column::LobColumnManager::Create(fs, lob_config);
    ASSERT_TRUE(manager_result.ok()) << manager_result.status().ToString();
    auto manager = std::move(manager_result).ValueOrDie();
    auto writer_result = manager->CreateWriter();
    ASSERT_TRUE(writer_result.ok()) << writer_result.status().ToString();
    auto writer = std::move(writer_result).ValueOrDie();

    std::string large_text(72 * 1024, 'x');
    large_text += " " + unique_token;
    std::vector<std::string> texts = {
        "plain text without target token",
        large_text,
        "another row without target token",
    };

    std::vector<std::string> encoded_refs;
    encoded_refs.reserve(texts.size());
    for (const auto& text : texts) {
        auto ref_result = writer->WriteText(text);
        ASSERT_TRUE(ref_result.ok()) << ref_result.status().ToString();
        auto ref = std::move(ref_result).ValueOrDie();
        ASSERT_EQ(ref.size(), milvus_storage::lob_column::LOB_REFERENCE_SIZE);
        ASSERT_TRUE(milvus_storage::lob_column::IsLOBReference(ref.data()));
        encoded_refs.emplace_back(reinterpret_cast<const char*>(ref.data()),
                                  ref.size());
    }
    auto close_result = writer->Close();
    ASSERT_TRUE(close_result.ok()) << close_result.status().ToString();
    auto lob_files = std::move(close_result).ValueOrDie();
    ASSERT_FALSE(lob_files.empty());

    auto field_data =
        storage::CreateFieldData(DataType::TEXT, DataType::NONE, false);
    field_data->FillFieldData(encoded_refs.data(), encoded_refs.size());

    auto cm = storage::CreateChunkManager(get_default_local_storage_config());
    auto load_info = PrepareSingleFieldInsertBinlog(collection_id,
                                                    partition_id,
                                                    segment_id,
                                                    text_field_id.get(),
                                                    {field_data},
                                                    cm);

    auto segment = CreateSealedSegment(schema, empty_index_meta, segment_id);
    auto* sealed = dynamic_cast<ChunkedSegmentSealedImpl*>(segment.get());
    ASSERT_NE(sealed, nullptr);
    sealed->SetTextLobPathForTesting(text_field_id, lob_base_path);

    auto status = LoadFieldData(segment.get(), &load_info);
    ASSERT_EQ(status.error_code, Success) << status.error_msg;

    sealed->CreateTextIndex(text_field_id);
    auto index_pin = sealed->GetTextIndex(nullptr, text_field_id);
    auto hits = index_pin.get()->MatchQuery(unique_token, 1);
    ASSERT_EQ(hits.size(), texts.size());
    EXPECT_FALSE(hits[0]);
    EXPECT_TRUE(hits[1]);
    EXPECT_FALSE(hits[2]);

    boost::filesystem::remove_all(TestLocalPath + test_dir);
}

TEST(TextMatch, GrowingBuildTextIndexFromTextLobRefsDecodesText) {
    const FieldId text_field_id(101);
    const std::string large_token = "zzgrowingloblarge";
    const std::string second_batch_token = "zzgrowinglobbatch";
    const std::string null_token = "zzgrowinglobnull";
    constexpr int64_t row_count = 1030;

    auto test_dir =
        "growing_text_lob_index_" +
        std::to_string(
            std::chrono::steady_clock::now().time_since_epoch().count());
    boost::filesystem::remove_all(TestLocalPath + test_dir);

    auto schema = std::make_shared<Schema>();
    schema->AddField(FieldMeta(
        FieldName("pk"), FieldId(100), DataType::INT64, false, std::nullopt));
    schema->set_primary_field_id(FieldId(100));
    std::map<std::string, std::string> text_params = {
        {"enable_match", "true"},
        {"enable_analyzer", "true"},
        {"analyzer_params", R"({"tokenizer": "standard"})"},
    };
    schema->AddField(FieldMeta(FieldName("str"),
                               text_field_id,
                               DataType::TEXT,
                               65536,
                               true,
                               true,
                               true,
                               text_params,
                               std::nullopt));

    auto lob_base_path =
        test_dir + "/lobs/" + std::to_string(text_field_id.get());
    milvus_storage::lob_column::LobColumnConfig lob_config;
    lob_config.lob_base_path = lob_base_path;
    lob_config.field_id = text_field_id.get();
    lob_config.inline_threshold = 1;
    lob_config.max_lob_file_bytes = 256 * 1024;
    lob_config.flush_threshold_bytes = 64 * 1024;
    auto properties = milvus::storage::LoonFFIPropertiesSingleton::GetInstance()
                          .GetProperties();
    ASSERT_NE(properties, nullptr);
    lob_config.properties = *properties;

    auto fs = milvus::segcore::GetDefaultArrowFileSystem();
    auto manager_result =
        milvus_storage::lob_column::LobColumnManager::Create(fs, lob_config);
    ASSERT_TRUE(manager_result.ok()) << manager_result.status().ToString();
    auto manager = std::move(manager_result).ValueOrDie();
    auto writer_result = manager->CreateWriter();
    ASSERT_TRUE(writer_result.ok()) << writer_result.status().ToString();
    auto writer = std::move(writer_result).ValueOrDie();

    std::vector<std::string> encoded_refs;
    encoded_refs.reserve(row_count);
    for (int64_t i = 0; i < row_count; ++i) {
        std::string text = "plain growing text row " + std::to_string(i);
        if (i == 3) {
            text = std::string(72 * 1024, 'x') + " " + large_token;
        } else if (i == 1025) {
            text = "second batch decoded text " + second_batch_token;
        } else if (i == 7) {
            text = "invalid nullable row " + null_token;
        }

        auto ref_result = writer->WriteText(text);
        ASSERT_TRUE(ref_result.ok()) << ref_result.status().ToString();
        auto ref = std::move(ref_result).ValueOrDie();
        ASSERT_EQ(ref.size(), milvus_storage::lob_column::LOB_REFERENCE_SIZE);
        ASSERT_TRUE(milvus_storage::lob_column::IsLOBReference(ref.data()));
        encoded_refs.emplace_back(reinterpret_cast<const char*>(ref.data()),
                                  ref.size());
    }
    auto close_result = writer->Close();
    ASSERT_TRUE(close_result.ok()) << close_result.status().ToString();
    auto lob_files = std::move(close_result).ValueOrDie();
    ASSERT_FALSE(lob_files.empty());

    auto field_data =
        storage::CreateFieldData(DataType::TEXT, DataType::NONE, true);
    std::vector<bool> valids(row_count, true);
    valids[7] = false;
    std::vector<uint8_t> valid_bytes((row_count + 7) / 8, 0);
    for (int64_t i = 0; i < row_count; ++i) {
        if (valids[i]) {
            valid_bytes[i >> 3] |= (1u << (i & 7));
        }
    }
    field_data->FillFieldData(
        encoded_refs.data(), valid_bytes.data(), row_count, 0);

    auto segment = CreateGrowingSegment(schema, empty_index_meta);
    auto* growing = dynamic_cast<SegmentGrowingImpl*>(segment.get());
    ASSERT_NE(growing, nullptr);
    growing->SetTextLobPathForTesting(text_field_id, lob_base_path);
    growing->BuildTextIndexFromTextLobRefs(
        text_field_id, {field_data}, 0, (*schema)[text_field_id]);

    auto index_pin = growing->GetTextIndex(nullptr, text_field_id);
    auto large_hits = index_pin.get()->MatchQuery(large_token, 1);
    ASSERT_EQ(large_hits.size(), row_count);
    EXPECT_TRUE(large_hits[3]);

    auto second_batch_hits = index_pin.get()->MatchQuery(second_batch_token, 1);
    ASSERT_EQ(second_batch_hits.size(), row_count);
    EXPECT_TRUE(second_batch_hits[1025]);

    auto null_hits = index_pin.get()->MatchQuery(null_token, 1);
    ASSERT_EQ(null_hits.size(), row_count);
    EXPECT_FALSE(null_hits[7]);

    boost::filesystem::remove_all(TestLocalPath + test_dir);
}

// Regression test: BuildIndexFromFieldData with a single batch should still
// work correctly (i == offset in this case, so the old bug was hidden).
TEST(TextMatch, BuildIndexFromFieldDataSingleBatchNullable) {
    using Index = index::TextMatchIndex;

    auto fd =
        storage::CreateFieldData(DataType::VARCHAR, DataType::NONE, true, 1, 4);
    std::vector<std::string> texts = {"alpha", "", "beta", ""};
    std::vector<bool> valids = {true, false, true, false};
    size_t byte_count = (texts.size() + 7) / 8;
    std::vector<uint8_t> valid_bytes(byte_count, 0);
    for (size_t i = 0; i < valids.size(); i++) {
        if (valids[i]) {
            valid_bytes[i >> 3] |= (1u << (i & 7));
        }
    }
    fd->FillFieldData(texts.data(), valid_bytes.data(), texts.size(), 0);

    std::vector<milvus::FieldDataPtr> field_datas = {fd};

    auto index = std::make_unique<Index>(200,
                                         "test_single_batch",
                                         "milvus_tokenizer",
                                         "{}",
                                         /*enable_background_merge=*/true);
    index->CreateReader(milvus::index::SetBitsetGrowing);
    index->RegisterAnalyzer("milvus_tokenizer", "{}");

    index->BuildIndexFromFieldData(field_datas, true);
    index->Commit();
    index->Reload();

    auto null_bits = index->IsNull();
    ASSERT_EQ(null_bits.size(), 4);
    ASSERT_FALSE(null_bits[0]);
    ASSERT_TRUE(null_bits[1]);
    ASSERT_FALSE(null_bits[2]);
    ASSERT_TRUE(null_bits[3]);

    auto res = index->MatchQuery("alpha", 1);
    ASSERT_EQ(res.size(), 4);
    ASSERT_TRUE(res[0]);
    ASSERT_FALSE(res[1]);
    ASSERT_FALSE(res[2]);
    ASSERT_FALSE(res[3]);

    auto res2 = index->MatchQuery("beta", 1);
    ASSERT_EQ(res2.size(), 4);
    ASSERT_FALSE(res2[0]);
    ASSERT_FALSE(res2[1]);
    ASSERT_TRUE(res2[2]);
    ASSERT_FALSE(res2[3]);
}

TEST(TextMatch, GrowingNaive) {
    auto schema = GenTestSchema();
    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    std::vector<std::string> raw_str = {"football, basketball, pingpang",
                                        "swimming, football"};

    int64_t N = 2;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }

    seg->PreInsert(N);
    seg->Insert(0,
                N,
                raw_data.row_ids_.data(),
                raw_data.timestamps_.data(),
                raw_data.raw_);

    std::this_thread::sleep_for(std::chrono::milliseconds(200) * 2);

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "football", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_TRUE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "football", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_FALSE(final[1]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "swimming", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "swimming", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
        }
    }

    {
        auto expr =
            GetMatchExpr(schema, "basketball, swimming", OpType::TextMatch);
        BitsetType final;
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_TRUE(final[1]);
        auto expr1 =
            GetNotMatchExpr(schema, "basketball, swimming", OpType::TextMatch);
        final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);

        auto expr2 =
            GetMatchExpr(schema, "football, pingpang", OpType::PhraseMatch);
        final = ExecuteQueryExpr(expr2, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);
        auto expr3 =
            GetMatchExpr(schema, "football, pingpang", OpType::PhraseMatch, 1);
        final = ExecuteQueryExpr(expr3, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_FALSE(final[1]);
    }

    {
        // A typo (edit distance 1) matches the same rows as the exact term and
        // exercises the executor + growing commit/reload path for fuzzy match.
        BitsetType final;
        auto expr = GetFuzzyMatchExpr(schema, "footbal", 1);
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_TRUE(final[1]);
        auto expr1 =
            GetNotMatchExpr(schema, "footbal", OpType::TextMatchFuzzy, 1);
        final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);
    }

    {
        // The executor rejects an out-of-range or missing max_edit_distance
        // (the parser guarantees [0, 2], but a raw proto may not).
        auto bad = GetFuzzyMatchExpr(schema, "footbal", 3);
        ASSERT_ANY_THROW(ExecuteQueryExpr(bad, seg.get(), N, MAX_TIMESTAMP));
        auto missing = GetFuzzyMatchExprNoDistance(schema, "footbal");
        ASSERT_ANY_THROW(
            ExecuteQueryExpr(missing, seg.get(), N, MAX_TIMESTAMP));
    }
}

// Regression test for https://github.com/milvus-io/milvus/issues/48388
// On growing segments, the TextIndex may index rows beyond the query timestamp.
// ExecTextMatch must truncate the result bitmap to active_count to avoid
// FilterBitsNode assertion failure: col_vec_size != need_process_rows_.
TEST(TextMatch, GrowingIndexAheadOfActiveCount) {
    auto schema = GenTestSchema();
    auto seg = CreateGrowingSegment(schema, empty_index_meta);

    // Batch 1: 3 rows with timestamps 0,1,2
    int64_t N1 = 3;
    uint64_t seed = 19190504;
    auto data1 = DataGen(schema, N1, seed);
    auto str_col1 = data1.raw_->mutable_fields_data()
                        ->at(1)
                        .mutable_scalars()
                        ->mutable_string_data()
                        ->mutable_data();
    str_col1->at(0) = "football, basketball";
    str_col1->at(1) = "swimming, tennis";
    str_col1->at(2) = "football, swimming";
    seg->PreInsert(N1);
    seg->Insert(
        0, N1, data1.row_ids_.data(), data1.timestamps_.data(), data1.raw_);

    // Batch 2: 2 rows with timestamps 3,4 (later than batch 1)
    int64_t N2 = 2;
    auto data2 = DataGen(schema, N2, seed, N1);  // ts_offset=N1 → ts 3,4
    // Fix row_ids to continue from batch 1
    for (int i = 0; i < N2; i++) {
        data2.row_ids_[i] = N1 + i;
    }
    auto str_col2 = data2.raw_->mutable_fields_data()
                        ->at(1)
                        .mutable_scalars()
                        ->mutable_string_data()
                        ->mutable_data();
    str_col2->at(0) = "football, rugby";
    str_col2->at(1) = "football, cricket";
    seg->PreInsert(N2);
    seg->Insert(
        N1, N2, data2.row_ids_.data(), data2.timestamps_.data(), data2.raw_);

    // Wait for TextIndex to index all 5 rows
    std::this_thread::sleep_for(std::chrono::milliseconds(200) * 2);

    // Query with timestamp=2 so only batch 1 (3 rows) is visible,
    // but TextIndex has indexed all 5 rows.
    // Before fix: assertion failure (bitset size 5 != need_process_rows 3)
    // After fix: result is correctly truncated to 3 rows
    int64_t query_ts = N1 - 1;  // timestamp=2, active_count should be N1
    for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
        auto expr = GetMatchExpr(schema, "football", op);
        BitsetType result = ExecuteQueryExpr(expr, seg.get(), N1, query_ts);
        ASSERT_EQ(result.size(), N1)
            << "result bitmap size should equal active_count (N1=" << N1
            << "), not total indexed rows (" << N1 + N2 << ")";
        // Row 0 ("football, basketball") and row 2 ("football, swimming") match
        ASSERT_TRUE(result[0]);
        ASSERT_FALSE(result[1]);
        ASSERT_TRUE(result[2]);
    }

    // Also verify full-range query still works correctly
    for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
        auto expr = GetMatchExpr(schema, "football", op);
        BitsetType result =
            ExecuteQueryExpr(expr, seg.get(), N1 + N2, MAX_TIMESTAMP);
        ASSERT_EQ(result.size(), N1 + N2);
        ASSERT_TRUE(result[0]);   // "football, basketball"
        ASSERT_FALSE(result[1]);  // "swimming, tennis"
        ASSERT_TRUE(result[2]);   // "football, swimming"
        ASSERT_TRUE(result[3]);   // "football, rugby"
        ASSERT_TRUE(result[4]);   // "football, cricket"
    }
}

TEST(TextMatch, GrowingNaiveNullable) {
    auto schema = GenTestSchema({}, true);
    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    std::vector<std::string> raw_str = {
        "football, basketball, pingpang", "swimming, football", ""};
    std::vector<bool> raw_str_valid = {true, true, false};

    int64_t N = 3;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    auto str_col_valid =
        raw_data.raw_->mutable_fields_data()->at(1).mutable_valid_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }
    for (int64_t i = 0; i < N; i++) {
        str_col_valid->at(i) = raw_str_valid[i];
    }

    seg->PreInsert(N);
    seg->Insert(0,
                N,
                raw_data.row_ids_.data(),
                raw_data.timestamps_.data(),
                raw_data.raw_);

    std::this_thread::sleep_for(std::chrono::milliseconds(200) * 2);
    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "football", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "football", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "swimming", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "swimming", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        auto expr =
            GetMatchExpr(schema, "basketball, swimming", OpType::TextMatch);
        BitsetType final;
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_TRUE(final[1]);
        ASSERT_FALSE(final[2]);
        auto expr1 =
            GetNotMatchExpr(schema, "basketball, swimming", OpType::TextMatch);
        final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);
        ASSERT_FALSE(final[2]);

        auto expr2 =
            GetMatchExpr(schema, "football, pingpang", OpType::PhraseMatch);
        final = ExecuteQueryExpr(expr2, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);
        ASSERT_FALSE(final[2]);
        auto expr3 =
            GetMatchExpr(schema, "football, pingpang", OpType::PhraseMatch, 1);
        final = ExecuteQueryExpr(expr3, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_FALSE(final[1]);
        ASSERT_FALSE(final[2]);
    }
}

TEST(TextMatch, SealedNaive) {
    auto schema = GenTestSchema();
    std::vector<std::string> raw_str = {"football, basketball, pingpang",
                                        "swimming, football"};

    int64_t N = 2;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }

    auto seg = CreateSealedWithFieldDataLoaded(schema, raw_data);
    seg->CreateTextIndex(FieldId(101));

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "football", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_TRUE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "football", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_FALSE(final[1]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "swimming", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "swimming", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
        }
    }

    {
        auto expr =
            GetMatchExpr(schema, "basketball, swimming", OpType::TextMatch);
        BitsetType final;
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_TRUE(final[1]);
        auto expr1 =
            GetNotMatchExpr(schema, "basketball, swimming", OpType::TextMatch);
        final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);

        auto expr2 =
            GetMatchExpr(schema, "football, pingpang", OpType::PhraseMatch);
        final = ExecuteQueryExpr(expr2, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);
        auto expr3 =
            GetMatchExpr(schema, "football, pingpang", OpType::PhraseMatch, 1);
        final = ExecuteQueryExpr(expr3, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_FALSE(final[1]);
    }

    {
        // A typo (edit distance 1) matches the same rows as the exact term,
        // exercising the executor dispatch for fuzzy on a sealed segment.
        BitsetType final;
        auto expr = GetFuzzyMatchExpr(schema, "footbal", 1);
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_TRUE(final[1]);
        auto expr1 =
            GetNotMatchExpr(schema, "footbal", OpType::TextMatchFuzzy, 1);
        final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);
    }
}

TEST(TextMatch, SealedNaiveNullable) {
    auto schema = GenTestSchema({}, true);
    std::vector<std::string> raw_str = {
        "football, basketball, pingpang", "swimming, football", ""};
    std::vector<bool> raw_str_valid = {true, true, false};

    int64_t N = 3;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }
    auto str_col_valid =
        raw_data.raw_->mutable_fields_data()->at(1).mutable_valid_data();
    for (int64_t i = 0; i < N; i++) {
        str_col_valid->at(i) = raw_str_valid[i];
    }

    auto seg = CreateSealedWithFieldDataLoaded(schema, raw_data);
    seg->CreateTextIndex(FieldId(101));
    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "football", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "football", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "swimming", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "swimming", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        auto expr =
            GetMatchExpr(schema, "basketball, swimming", OpType::TextMatch);
        BitsetType final;
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_TRUE(final[1]);
        ASSERT_FALSE(final[2]);
        auto expr1 =
            GetNotMatchExpr(schema, "basketball, swimming", OpType::TextMatch);
        final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);
        ASSERT_FALSE(final[2]);

        auto expr2 =
            GetMatchExpr(schema, "football, pingpang", OpType::PhraseMatch);
        final = ExecuteQueryExpr(expr2, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_FALSE(final[1]);
        ASSERT_FALSE(final[2]);
        auto expr3 =
            GetMatchExpr(schema, "football, pingpang", OpType::PhraseMatch, 1);
        final = ExecuteQueryExpr(expr3, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_TRUE(final[0]);
        ASSERT_FALSE(final[1]);
        ASSERT_FALSE(final[2]);
    }
}

TEST(TextMatch, GrowingJieBa) {
    auto schema = GenTestSchema({
        {"enable_match", "true"},
        {"enable_analyzer", "true"},
        {"analyzer_params", R"({"tokenizer": "jieba"})"},
    });
    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    std::vector<std::string> raw_str = {"青铜时代", "黄金时代"};

    int64_t N = 2;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }

    seg->PreInsert(N);
    seg->Insert(0,
                N,
                raw_data.row_ids_.data(),
                raw_data.timestamps_.data(),
                raw_data.raw_);

    std::this_thread::sleep_for(std::chrono::milliseconds(200) * 2);

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "青铜", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "青铜", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "黄金", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "黄金", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "时代", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_TRUE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "时代", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_FALSE(final[1]);
        }
    }

    {
        BitsetType final;
        auto expr = GetMatchExpr(schema, "黄金时代", OpType::PhraseMatch);
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_TRUE(final[1]);
    }
}

TEST(TextMatch, GrowingJieBaNullable) {
    auto schema = GenTestSchema(
        {
            {"enable_match", "true"},
            {"enable_tokenizer", "true"},
            {"analyzer_params", R"({"tokenizer": "jieba"})"},
        },
        true);
    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    std::vector<std::string> raw_str = {"青铜时代", "黄金时代", ""};
    std::vector<bool> raw_str_valid = {true, true, false};

    int64_t N = 3;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }
    auto str_col_valid =
        raw_data.raw_->mutable_fields_data()->at(1).mutable_valid_data();
    for (int64_t i = 0; i < N; i++) {
        str_col_valid->at(i) = raw_str_valid[i];
    }

    seg->PreInsert(N);
    seg->Insert(0,
                N,
                raw_data.row_ids_.data(),
                raw_data.timestamps_.data(),
                raw_data.raw_);

    std::this_thread::sleep_for(std::chrono::milliseconds(200) * 2);
    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "青铜", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "青铜", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "黄金", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "黄金", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "时代", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "时代", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        BitsetType final;
        auto expr = GetMatchExpr(schema, "黄金时代", OpType::PhraseMatch);
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_TRUE(final[1]);
        ASSERT_FALSE(final[2]);
    }
}

TEST(TextMatch, SealedJieBa) {
    auto schema = GenTestSchema({
        {"enable_match", "true"},
        {"enable_analyzer", "true"},
        {"analyzer_params", R"({"tokenizer": "jieba"})"},
    });
    std::vector<std::string> raw_str = {"青铜时代", "黄金时代"};

    int64_t N = 2;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }

    auto seg = CreateSealedWithFieldDataLoaded(schema, raw_data);
    seg->CreateTextIndex(FieldId(101));

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "青铜", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "青铜", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "黄金", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "黄金", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "时代", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_TRUE(final[1]);
            auto expr1 = GetNotMatchExpr(schema, "时代", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_FALSE(final[1]);
        }
    }

    {
        BitsetType final;
        auto expr = GetMatchExpr(schema, "黄金时代", OpType::PhraseMatch);
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_TRUE(final[1]);
    }
}

TEST(TextMatch, SealedJieBaNullable) {
    auto schema = GenTestSchema(
        {
            {"enable_match", "true"},
            {"enable_tokenizer", "true"},
            {"analyzer_params", R"({"tokenizer": "jieba"})"},
        },
        true);
    std::vector<std::string> raw_str = {"青铜时代", "黄金时代", ""};
    std::vector<bool> raw_str_valid = {true, true, false};

    int64_t N = 3;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }
    auto str_col_valid =
        raw_data.raw_->mutable_fields_data()->at(1).mutable_valid_data();
    for (int64_t i = 0; i < N; i++) {
        str_col_valid->at(i) = raw_str_valid[i];
    }

    auto seg = CreateSealedWithFieldDataLoaded(schema, raw_data);
    seg->CreateTextIndex(FieldId(101));

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "青铜", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "青铜", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "黄金", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "黄金", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        BitsetType final;
        for (auto op : {OpType::TextMatch, OpType::PhraseMatch}) {
            auto expr = GetMatchExpr(schema, "时代", op);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_TRUE(final[0]);
            ASSERT_TRUE(final[1]);
            ASSERT_FALSE(final[2]);
            auto expr1 = GetNotMatchExpr(schema, "时代", op);
            final = ExecuteQueryExpr(expr1, seg.get(), N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            ASSERT_FALSE(final[0]);
            ASSERT_FALSE(final[1]);
            ASSERT_FALSE(final[2]);
        }
    }

    {
        BitsetType final;
        auto expr = GetMatchExpr(schema, "黄金时代", OpType::PhraseMatch);
        final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
        ASSERT_EQ(final.size(), N);
        ASSERT_FALSE(final[0]);
        ASSERT_TRUE(final[1]);
        ASSERT_FALSE(final[2]);
    }
}

// Test that growing segment loading flushed binlogs will build text match index.
TEST(TextMatch, GrowingLoadData) {
    milvus::exec::expression::FunctionFactory& factory =
        milvus::exec::expression::FunctionFactory::Instance();
    factory.Initialize();
    int64_t N = 7;
    auto schema = GenTestSchema({}, true);
    schema->AddField(
        FieldName("RowID"), FieldId(0), DataType::INT64, false, std::nullopt);
    schema->AddField(FieldName("Timestamp"),
                     FieldId(1),
                     DataType::INT64,
                     false,
                     std::nullopt);
    std::vector<std::string> raw_str = {"football, basketball, pingpang",
                                        "swimming, football",
                                        "golf",
                                        "",
                                        "baseball",
                                        "kungfu, football",
                                        ""};
    auto raw_data = DataGen(schema, N);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }
    auto str_col_valid =
        raw_data.raw_->mutable_fields_data()->at(1).mutable_valid_data();
    for (int64_t i = 0; i < N; i++) {
        str_col_valid->at(i) = true;
    }
    // so we cannot match the second row
    str_col_valid->at(1) = false;

    auto storage_config = get_default_local_storage_config();
    auto cm = storage::CreateChunkManager(storage_config);
    auto load_info = PrepareInsertBinlog(1, 2, 3, raw_data, cm);

    auto segment = CreateGrowingSegment(schema, empty_index_meta);
    auto status = LoadFieldData(segment.get(), &load_info);
    ASSERT_EQ(status.error_code, Success);
    ASSERT_EQ(segment->get_real_count(), N);
    ASSERT_NE(segment->get_field_avg_size(FieldId(101)), 0);

    // Check whether the text index has been built.
    auto expr = GetMatchExpr(schema, "football", OpType::TextMatch);
    BitsetType final;
    final = ExecuteQueryExpr(expr, segment.get(), N, MAX_TIMESTAMP);
    ASSERT_EQ(final.size(), N);
    ASSERT_TRUE(final[0]);
    ASSERT_FALSE(final[1]);
    ASSERT_FALSE(final[2]);
    ASSERT_FALSE(final[3]);
    ASSERT_FALSE(final[4]);
    ASSERT_TRUE(final[5]);
    ASSERT_FALSE(final[6]);
}

TEST(TextMatch, ConcurrentReadWriteWithNull) {
    auto schema = GenTestSchema({}, true);
    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int64_t N = 1000;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col_valid =
        raw_data.raw_->mutable_fields_data()->at(1).mutable_valid_data();
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N - 1; i++) {
        str_col->at(i) = "";
    }
    str_col->at(N - 1) = "football";
    for (int64_t i = 0; i < N - 1; i++) {
        str_col_valid->at(i) = false;
    }
    str_col_valid->at(N - 1) = true;

    std::thread writer([&seg, &raw_data, N]() {
        seg->PreInsert(N);
        seg->Insert(0,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    });

    std::thread reader([&seg, N]() {
        auto start = std::chrono::high_resolution_clock::now();
        ;
        const std::chrono::seconds timeout_duration{2};
        while (true) {
            if (std::chrono::high_resolution_clock::now() - start >
                timeout_duration) {
                ASSERT_TRUE(false)
                    << "Failed to get valid results within timeout";
                break;
            }
            BitsetType final;
            auto expr =
                GetMatchExpr(GenTestSchema(), "football", OpType::TextMatch);
            final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
            if (final.size() != N || !final[N - 1]) {
                continue;
            }
            for (int64_t i = 0; i < N - 1; i++) {
                ASSERT_FALSE(final[i]);
            }
            ASSERT_TRUE(final[N - 1]);
            break;
        }
    });

    writer.join();
    reader.join();
}

TEST(TextMatch, ExprResCacheSealed) {
    using milvus::exec::ExprResCacheManager;
    auto& mgr = ExprResCacheManager::Instance();
    ExprResCacheManager::SetEnabled(true);
    mgr.Clear();
    mgr.SetCapacityBytes(1ULL << 20);

    auto schema = GenTestSchema();
    std::vector<std::string> raw_str = {"football, basketball, pingpang",
                                        "swimming, football"};

    int64_t N = 2;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }

    auto seg = CreateSealedWithFieldDataLoaded(schema, raw_data);
    seg->CreateTextIndex(FieldId(101));

    ASSERT_EQ(mgr.GetEntryCount(), 0);

    BitsetType final;
    auto expr = GetMatchExpr(schema, "football", OpType::TextMatch);
    final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
    ASSERT_EQ(final.size(), N);
    ASSERT_TRUE(final[0]);
    ASSERT_TRUE(final[1]);

    // Expect one cache entry inserted
    ASSERT_EQ(mgr.GetEntryCount(), 1);

    // Run again; should hit cache and not increase entries
    final = ExecuteQueryExpr(expr, seg.get(), N, MAX_TIMESTAMP);
    ASSERT_EQ(final.size(), N);
    ASSERT_TRUE(final[0]);
    ASSERT_TRUE(final[1]);
    ASSERT_EQ(mgr.GetEntryCount(), 1);

    // Cleanup
    mgr.Clear();
    ExprResCacheManager::SetEnabled(false);
}

TEST(TextMatch, ExprResCacheFilterBitsDoesNotDuplicateTextMatchEntry) {
    using milvus::exec::ExprResCacheManager;
    auto& mgr = ExprResCacheManager::Instance();
    ExprResCacheManager::SetEnabled(true);
    mgr.Clear();
    mgr.SetCapacityBytes(1ULL << 20);

    auto schema = GenTestSchema();
    std::vector<std::string> raw_str = {"football, basketball, pingpang",
                                        "swimming, football"};

    int64_t N = 2;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }

    auto seg = CreateSealedWithFieldDataLoaded(schema, raw_data);
    seg->CreateTextIndex(FieldId(101));

    auto expr = GetMatchExpr(schema, "football", OpType::TextMatch);
    auto plan_fragment = plan::PlanFragment(expr);
    auto query_context = std::make_shared<milvus::exec::QueryContext>(
        DEAFULT_QUERY_ID, seg.get(), N, MAX_TIMESTAMP);
    query_context->set_enable_expr_cache(true);
    query_context->set_enable_sub_expr_cache_write(false);

    auto row = ExecPlanNodeVisitor::ExecuteTask(plan_fragment, query_context);
    ASSERT_NE(row, nullptr);
    ASSERT_EQ(mgr.GetEntryCount(), 1);

    ExprResCacheManager::Key filter_key{seg->get_segment_id(),
                                        expr->ToString()};
    ExprResCacheManager::Value filter_value;
    filter_value.active_count = N;
    ASSERT_TRUE(mgr.Get(filter_key, filter_value));

    ExprResCacheManager::Key text_match_key{seg->get_segment_id(),
                                            expr->filter()->ToString()};
    ExprResCacheManager::Value text_match_value;
    text_match_value.active_count = N;
    ASSERT_FALSE(mgr.Get(text_match_key, text_match_value));

    mgr.Clear();
    ExprResCacheManager::SetEnabled(false);
}

namespace {

BitsetType
ExecuteFilterBitsWithFullCache(
    const std::shared_ptr<plan::FilterBitsNode>& filter_plan,
    const segcore::SegmentInternalInterface* segment,
    int64_t active_count,
    Timestamp timestamp,
    int64_t entity_ttl_physical_time_us) {
    auto plan_fragment = plan::PlanFragment(filter_plan);
    auto query_context = std::make_shared<milvus::exec::QueryContext>(
        DEAFULT_QUERY_ID,
        segment,
        active_count,
        timestamp,
        0,
        0,
        milvus::query::PlanOptions(),
        std::make_shared<milvus::exec::QueryConfig>(),
        nullptr,
        std::unordered_map<std::string,
                           std::shared_ptr<milvus::exec::BaseConfig>>(),
        entity_ttl_physical_time_us);
    query_context->set_enable_expr_cache(true);
    query_context->set_enable_sub_expr_cache_write(false);

    auto row = ExecPlanNodeVisitor::ExecuteTask(plan_fragment, query_context);
    AssertInfo(row != nullptr,
               "ExecuteTask returned null row vector for query expression");
    auto col_vec = std::dynamic_pointer_cast<ColumnVector>(row->childrens()[0]);
    AssertInfo(col_vec != nullptr, "failed to cast to ColumnVector");
    BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
    BitsetType query_view(view);
    query_view.flip();
    return query_view;
}

}  // namespace

TEST(TextMatch, ExprResCacheFilterBitsUsesCurrentFilterNodeKey) {
    using milvus::exec::ExprResCacheManager;
    auto& mgr = ExprResCacheManager::Instance();
    ExprResCacheManager::SetEnabled(true);
    mgr.Clear();
    mgr.SetCapacityBytes(1ULL << 20);

    auto schema = GenTestSchema();
    std::vector<std::string> raw_str = {"football, basketball, pingpang",
                                        "swimming, football"};

    int64_t N = 2;
    uint64_t seed = 19190504;
    auto raw_data = DataGen(schema, N, seed);
    auto str_col = raw_data.raw_->mutable_fields_data()
                       ->at(1)
                       .mutable_scalars()
                       ->mutable_string_data()
                       ->mutable_data();
    for (int64_t i = 0; i < N; i++) {
        str_col->at(i) = raw_str[i];
    }

    auto seg = CreateSealedWithFieldDataLoaded(schema, raw_data);
    seg->CreateTextIndex(FieldId(101));

    auto football = GetMatchExpr(schema, "football", OpType::TextMatch);
    auto swimming = GetMatchExpr(schema, "swimming", OpType::TextMatch);

    auto football_result = ExecuteFilterBitsWithFullCache(
        football, seg.get(), N, MAX_TIMESTAMP, 0);
    ASSERT_TRUE(football_result[0]);
    ASSERT_TRUE(football_result[1]);

    auto swimming_result = ExecuteFilterBitsWithFullCache(
        swimming, seg.get(), N, MAX_TIMESTAMP, 0);
    ASSERT_FALSE(swimming_result[0]);
    ASSERT_TRUE(swimming_result[1]);

    mgr.Clear();
    ExprResCacheManager::SetEnabled(false);
}

TEST(TextMatch, ExprResCacheFilterBitsIncludesEntityTTLPhysicalTime) {
    using milvus::exec::ExprResCacheManager;
    auto& mgr = ExprResCacheManager::Instance();
    ExprResCacheManager::SetEnabled(true);
    mgr.Clear();
    mgr.SetCapacityBytes(1ULL << 20);

    auto schema = std::make_shared<Schema>();
    auto pk_fid = schema->AddDebugField("pk", DataType::INT64);
    auto ttl_fid = schema->AddDebugField("ttl_field", DataType::TIMESTAMPTZ);
    schema->set_primary_field_id(pk_fid);
    schema->set_ttl_field_id(ttl_fid);

    int64_t N = 2;
    auto raw_data = DataGen(schema, N, 19190504);
    for (auto& field_data : *raw_data.raw_->mutable_fields_data()) {
        if (field_data.field_id() == ttl_fid.get()) {
            auto* data =
                field_data.mutable_scalars()->mutable_timestamptz_data();
            data->set_data(0, 150);
            data->set_data(1, 250);
        }
    }

    auto seg = CreateSealedWithFieldDataLoaded(schema, raw_data);

    auto always_true_expr = std::make_shared<expr::AlwaysTrueExpr>();
    auto plan = std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
                                                       always_true_expr);

    auto before_expire =
        ExecuteFilterBitsWithFullCache(plan, seg.get(), N, MAX_TIMESTAMP, 100);
    ASSERT_TRUE(before_expire[0]);
    ASSERT_TRUE(before_expire[1]);

    auto after_expire =
        ExecuteFilterBitsWithFullCache(plan, seg.get(), N, MAX_TIMESTAMP, 200);
    ASSERT_FALSE(after_expire[0]);
    ASSERT_TRUE(after_expire[1]);

    mgr.Clear();
    ExprResCacheManager::SetEnabled(false);
}
