// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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 <algorithm>
#include <atomic>
#include <chrono>
#include <memory>
#include <string>
#include <thread>
#include <vector>

#include "core/block/block.h"
#include "core/column/column_array.h"
#include "core/column/column_nullable.h"
#include "core/column/column_string.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_string.h"
#include "exprs/function/match.h"
#include "runtime/runtime_state.h"
#include "storage/index/index_query_context.h"
#include "storage/index/inverted/analyzer/analyzer.h"
#include "storage/index/inverted/analyzer/custom_analyzer.h"
#include "storage/index/inverted/inverted_index_iterator.h"
#include "storage/index/inverted/inverted_index_reader.h"
#include "storage/olap_common.h"
#include "storage/tablet/tablet_schema.h"

namespace doris {

// Helper structure to manage analyzer lifetime
struct TestInvertedIndexCtx {
    std::unique_ptr<InvertedIndexAnalyzerCtx> ctx;
    std::shared_ptr<lucene::analysis::Analyzer> analyzer_holder;
};

// Helper function to create inverted index context
TestInvertedIndexCtx create_inverted_index_ctx(InvertedIndexParserType parser_type) {
    TestInvertedIndexCtx test_ctx;
    test_ctx.ctx = std::make_unique<InvertedIndexAnalyzerCtx>();
    test_ctx.ctx->parser_type = parser_type;
    if (parser_type != InvertedIndexParserType::PARSER_NONE) {
        InvertedIndexAnalyzerConfig config;
        config.parser_type = parser_type;
        test_ctx.analyzer_holder =
                doris::segment_v2::inverted_index::InvertedIndexAnalyzer::create_analyzer(&config);
        test_ctx.ctx->analyzer = test_ctx.analyzer_holder;
    }
    return test_ctx;
}

TEST(FunctionMatchTest, analyse_query_str) {
    FunctionMatchPhrase func_match_phrase;

    {
        auto inverted_index_ctx = nullptr;
        auto query_tokens =
                func_match_phrase.analyse_query_str_token(inverted_index_ctx, "a b c", "name");
        ASSERT_EQ(query_tokens.size(), 0);
    }

    {
        auto inverted_index_ctx = std::make_unique<InvertedIndexAnalyzerCtx>();
        inverted_index_ctx->parser_type = InvertedIndexParserType::PARSER_NONE;
        auto query_tokens = func_match_phrase.analyse_query_str_token(inverted_index_ctx.get(),
                                                                      "a b c", "name");
        ASSERT_EQ(query_tokens.size(), 1);
    }

    {
        auto inverted_index_ctx = std::make_unique<InvertedIndexAnalyzerCtx>();
        inverted_index_ctx->parser_type = InvertedIndexParserType::PARSER_ENGLISH;
        InvertedIndexAnalyzerConfig config;
        config.parser_type = InvertedIndexParserType::PARSER_ENGLISH;
        auto analyzer =
                doris::segment_v2::inverted_index::InvertedIndexAnalyzer::create_analyzer(&config);
        inverted_index_ctx->analyzer = analyzer;
        auto query_tokens = func_match_phrase.analyse_query_str_token(inverted_index_ctx.get(),
                                                                      "a b c", "name");
        ASSERT_EQ(query_tokens.size(), 3);
    }
}

TEST(FunctionMatchTest, regexp_rejects_expensive_bounded_repeat) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});

    auto string_col = ColumnString::create();
    string_col->insert_data("abcd", 4);
    ColumnUInt8::Container result(1, 0);

    FunctionMatchRegexp function;
    for (const char* pattern : {"(ab?c?d){1000,5000}", "(?# [)(ab?c?d){1000,5000}"}) {
        SCOPED_TRACE(pattern);
        Status status = function.execute_match(context.get(), "test_column", pattern, 1,
                                               string_col.get(), nullptr, nullptr, result);
        EXPECT_FALSE(status.ok());
        EXPECT_NE(status.to_string().find("bounded repetition exceeds 50"), std::string::npos);
    }
}

// Test FunctionMatchAny::execute_match
TEST(FunctionMatchTest, match_any_execute) {
    FunctionMatchAny func_match_any;

    // Create test columns
    auto string_col = ColumnString::create();
    string_col->insert_data("apple banana cherry", 19);
    string_col->insert_data("dog cat bird", 12);
    string_col->insert_data("red blue green", 14);
    string_col->insert_data("hello world", 11);
    string_col->insert_data("", 0);

    ColumnUInt8::Container result(5, 0);
    auto inverted_index_ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    // Test various query scenarios
    struct TestCase {
        std::string query;
        std::vector<uint8_t> expected;
    };

    std::vector<TestCase> test_cases = {
            {"apple", {1, 0, 0, 0, 0}},       // Match first row only
            {"dog bird", {0, 1, 0, 0, 0}},    // Match second row (has both)
            {"red yellow", {0, 0, 1, 0, 0}},  // Match third row (has red)
            {"hello", {0, 0, 0, 1, 0}},       // Match fourth row
            {"nonexistent", {0, 0, 0, 0, 0}}, // No matches
            {"apple dog", {1, 1, 0, 0, 0}},   // Match first and second
            {"", {0, 0, 0, 0, 0}}             // Empty query
    };

    for (const auto& test_case : test_cases) {
        std::fill(result.begin(), result.end(), 0);

        // Create a mock FunctionContext - this would normally be provided by the execution engine
        FunctionContext context;

        // Note: This is a simplified test. In reality, execute_match requires proper setup
        // including enabling the allow_execute_match option
        // For now, we test the basic structure and expect the method to handle the setup gracefully

        // Basic validation that test case structure is correct
        EXPECT_EQ(test_case.expected.size(), 5);
    }
}

// Test FunctionMatchAll::execute_match
TEST(FunctionMatchTest, match_all_execute) {
    FunctionMatchAll func_match_all;

    auto string_col = ColumnString::create();
    string_col->insert_data("apple banana cherry", 19);
    string_col->insert_data("dog cat bird", 12);
    string_col->insert_data("red blue green", 14);
    string_col->insert_data("quick brown fox", 15);
    string_col->insert_data("", 0);

    ColumnUInt8::Container result(5, 0);
    auto inverted_index_ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    // Test match all scenarios
    struct TestCase {
        std::string query;
        std::vector<uint8_t> expected;
    };

    std::vector<TestCase> test_cases = {
            {"apple banana", {1, 0, 0, 0, 0}}, // First row has both
            {"dog cat", {0, 1, 0, 0, 0}},      // Second row has both
            {"red yellow", {0, 0, 0, 0, 0}},   // No row has both
            {"quick fox", {0, 0, 0, 1, 0}},    // Fourth row has both
            {"nonexistent", {0, 0, 0, 0, 0}},  // No matches
            {"", {0, 0, 0, 0, 0}}              // Empty query
    };

    for (const auto& test_case : test_cases) {
        std::fill(result.begin(), result.end(), 0);
        // Similar to match_any, this requires proper FunctionContext setup

        // Basic validation that test case structure is correct
        EXPECT_EQ(test_case.expected.size(), 5);
    }
}

// Test FunctionMatchPhrase::execute_match
TEST(FunctionMatchTest, match_phrase_execute) {
    FunctionMatchPhrase func_match_phrase;

    auto string_col = ColumnString::create();
    string_col->insert_data("quick brown fox", 15);
    string_col->insert_data("brown quick fox", 15);
    string_col->insert_data("fox brown quick", 15);
    string_col->insert_data("quick fox brown", 15);
    string_col->insert_data("the quick brown fox jumps", 25);
    string_col->insert_data("", 0);

    ColumnUInt8::Container result(6, 0);
    auto inverted_index_ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    struct TestCase {
        std::string query;
        std::vector<uint8_t> expected;
    };

    std::vector<TestCase> test_cases = {
            {"quick brown", {1, 0, 0, 0, 1, 0}}, // Match rows with consecutive "quick brown"
            {"brown fox", {1, 0, 1, 0, 1, 0}},   // Match rows with consecutive "brown fox"
            {"quick fox", {0, 0, 0, 1, 0, 0}},   // Only fourth row has consecutive "quick fox"
            {"fox quick", {0, 0, 0, 0, 0, 0}},   // No consecutive "fox quick"
            {"", {0, 0, 0, 0, 0, 0}}             // Empty query
    };

    for (const auto& test_case : test_cases) {
        std::fill(result.begin(), result.end(), 0);
        // Test structure - requires proper setup for actual execution

        // Basic validation that test case structure is correct
        EXPECT_EQ(test_case.expected.size(), 6);
    }
}

// Test FunctionMatchPhrasePrefix::execute_match
TEST(FunctionMatchTest, match_phrase_prefix_execute) {
    FunctionMatchPhrasePrefix func_match_phrase_prefix;

    auto string_col = ColumnString::create();
    string_col->insert_data("programming language", 20);
    string_col->insert_data("program files", 13);
    string_col->insert_data("language programming", 20);
    string_col->insert_data("computer program", 16);
    string_col->insert_data("", 0);

    ColumnUInt8::Container result(5, 0);
    auto inverted_index_ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    struct TestCase {
        std::string query;
        std::vector<uint8_t> expected;
    };

    std::vector<TestCase> test_cases = {
            {"prog", {1, 1, 0, 1, 0}},          // Prefix match for "program*"
            {"programming", {1, 0, 1, 0, 0}},   // Exact word match
            {"progra", {1, 1, 0, 1, 0}},        // Prefix "progra*"
            {"language prog", {0, 0, 1, 0, 0}}, // Phrase with prefix
            {"nonexist", {0, 0, 0, 0, 0}},      // No matches
            {"", {0, 0, 0, 0, 0}}               // Empty query
    };

    for (const auto& test_case : test_cases) {
        std::fill(result.begin(), result.end(), 0);
        // Test structure

        // Basic validation that test case structure is correct
        EXPECT_EQ(test_case.expected.size(), 5);
    }
}

// Test FunctionMatchRegexp::execute_match
TEST(FunctionMatchTest, match_regexp_execute) {
    FunctionMatchRegexp func_match_regexp;

    auto string_col = ColumnString::create();
    string_col->insert_data("test123data", 11);
    string_col->insert_data("data456test", 11);
    string_col->insert_data("abc789xyz", 9);
    string_col->insert_data("nodigits", 8);
    string_col->insert_data("", 0);

    ColumnUInt8::Container result(5, 0);
    auto inverted_index_ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_NONE);

    struct TestCase {
        std::string pattern;
        std::vector<uint8_t> expected;
    };

    std::vector<TestCase> test_cases = {
            {"\\d+", {1, 1, 1, 0, 0}},       // Match digits
            {"test", {1, 1, 0, 0, 0}},       // Match "test"
            {"^test", {1, 0, 0, 0, 0}},      // Start with "test"
            {"test$", {0, 1, 0, 0, 0}},      // End with "test"
            {"[a-z]+\\d+", {1, 1, 1, 0, 0}}, // Letters followed by digits
            {"xyz$", {0, 0, 1, 0, 0}},       // End with "xyz"
            {"invalid[", {0, 0, 0, 0, 0}}    // Invalid regex (should not crash)
    };

    for (const auto& test_case : test_cases) {
        std::fill(result.begin(), result.end(), 0);
        // Test structure

        // Basic validation that test case structure is correct
        EXPECT_EQ(test_case.expected.size(), 5);
    }
}

// Test FunctionMatchPhraseEdge::execute_match
TEST(FunctionMatchTest, match_phrase_edge_execute) {
    FunctionMatchPhraseEdge func_match_phrase_edge;

    auto string_col = ColumnString::create();
    string_col->insert_data("database management system", 26);
    string_col->insert_data("data management", 15);
    string_col->insert_data("system database", 15);
    string_col->insert_data("manage databases", 16);
    string_col->insert_data("", 0);

    ColumnUInt8::Container result(5, 0);
    auto inverted_index_ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    struct TestCase {
        std::string query;
        std::vector<uint8_t> expected;
    };

    std::vector<TestCase> test_cases = {
            {"data", {1, 1, 0, 0, 0}},            // Single word edge match
            {"manage", {1, 1, 0, 1, 0}},          // Edge match for "manage*"
            {"database system", {1, 0, 1, 0, 0}}, // Phrase edge match
            {"nonexistent", {0, 0, 0, 0, 0}},     // No matches
            {"", {0, 0, 0, 0, 0}}                 // Empty query
    };

    for (const auto& test_case : test_cases) {
        std::fill(result.begin(), result.end(), 0);
        // Test structure

        // Basic validation that test case structure is correct
        EXPECT_EQ(test_case.expected.size(), 5);
    }
}

// Test get_query_type_from_fn_name
TEST(FunctionMatchTest, get_query_type_from_fn_name) {
    FunctionMatchAny match_any;
    FunctionMatchAll match_all;
    FunctionMatchPhrase match_phrase;
    FunctionMatchPhrasePrefix match_phrase_prefix;
    FunctionMatchRegexp match_regexp;
    FunctionMatchPhraseEdge match_phrase_edge;

    // Test query type identification
    EXPECT_EQ(match_any.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_ANY_QUERY);
    EXPECT_EQ(match_all.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_ALL_QUERY);
    EXPECT_EQ(match_phrase.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_QUERY);
    EXPECT_EQ(match_phrase_prefix.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_PREFIX_QUERY);
    EXPECT_EQ(match_regexp.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_REGEXP_QUERY);
    EXPECT_EQ(match_phrase_edge.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_EDGE_QUERY);
}

// Test analyse_data_token with different parser types
TEST(FunctionMatchTest, analyse_data_token) {
    FunctionMatchAny match_any;

    auto string_col = ColumnString::create();
    string_col->insert_data("Hello World! This is a test.", 29);
    string_col->insert_data("Multiple words here", 19);

    // Test with PARSER_NONE
    {
        auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_NONE);
        int32_t offset = 0;
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                                   nullptr, offset);
        EXPECT_EQ(tokens.size(), 1);
        std::string actual_term = tokens[0].get_single_term();
        std::string expected_term = "Hello World! This is a test.";
        // Remove null terminator if present
        if (!actual_term.empty() && actual_term.back() == '\0') {
            actual_term.pop_back();
        }
        EXPECT_EQ(actual_term, expected_term);
    }

    // Test with PARSER_ENGLISH
    {
        auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);
        int32_t offset = 0;
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                                   nullptr, offset);
        // English parser should split into multiple tokens
        EXPECT_GT(tokens.size(), 1);
    }
}

// Test error handling and edge cases
TEST(FunctionMatchTest, error_handling_and_edge_cases) {
    FunctionMatchAny match_any;

    // Test with null inverted index context
    {
        auto query_tokens = match_any.analyse_query_str_token(nullptr, "test query", "test_col");
        EXPECT_EQ(query_tokens.size(), 0);
    }

    // Test with empty query string
    {
        auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);
        auto query_tokens = match_any.analyse_query_str_token(ctx.ctx.get(), "", "test_col");
        EXPECT_EQ(query_tokens.size(), 0);
    }

    // Test with empty data
    {
        auto string_col = ColumnString::create();
        string_col->insert_data("", 0);

        auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);
        int32_t offset = 0;
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                                   nullptr, offset);
        EXPECT_EQ(tokens.size(), 0);
    }
}

// Test with array offsets (for array column types)
TEST(FunctionMatchTest, array_offset_handling) {
    FunctionMatchAny match_any;

    auto string_col = ColumnString::create();
    string_col->insert_data("first", 5);
    string_col->insert_data("second", 6);
    string_col->insert_data("third", 5);
    string_col->insert_data("fourth", 6);

    // Simulate array offsets: [0,2] and [2,4] representing two arrays
    ColumnArray::Offsets64 array_offsets = {2, 4};

    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    // Test first array [first, second]
    {
        int32_t offset = 0;
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                                   &array_offsets, offset);
        EXPECT_GT(tokens.size(), 0);
        // offset should be updated to 2
        EXPECT_EQ(offset, 2);
    }

    // Test second array [third, fourth]
    {
        int32_t offset = 2; // Start from where previous ended
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 1,
                                                   &array_offsets, offset);
        EXPECT_GT(tokens.size(), 0);
        // offset should be updated to 4
        EXPECT_EQ(offset, 4);
    }
}

TEST(FunctionMatchTest, array_match_uses_tokens_from_all_elements) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    auto string_col = ColumnString::create();
    for (const std::string value : {"alpha token", "tail only", "head only", "omega token",
                                    "unrelated head", "unrelated tail", "alpha token", ""}) {
        string_col->insert_data(value.data(), value.size());
    }
    ColumnArray::Offsets64 array_offsets = {2, 4, 6, 8, 8};

    FunctionMatchAny match_any;
    FunctionMatchAll match_all;
    FunctionMatchRegexp match_regexp;
    FunctionMatchPhrase match_phrase;
    FunctionMatchPhrasePrefix match_phrase_prefix;
    FunctionMatchPhraseEdge match_phrase_edge;
    auto check_match = [&](const FunctionMatchBase& function, const std::string& query,
                           const std::vector<uint8_t>& expected) {
        SCOPED_TRACE(function.get_name() + ": " + query);
        ColumnUInt8::Container result(array_offsets.size(), 0);
        ASSERT_TRUE(function.execute_match(context.get(), "tags", query, array_offsets.size(),
                                           string_col.get(), ctx.ctx.get(), &array_offsets, result)
                            .ok());
        for (size_t row = 0; row < expected.size(); ++row) {
            EXPECT_EQ(result[row], expected[row]) << "row: " << row;
        }
    };
    check_match(match_any, "alpha", {1, 0, 0, 1, 0});
    check_match(match_any, "omega", {0, 1, 0, 0, 0});
    check_match(match_all, "alpha tail", {1, 0, 0, 0, 0});
    check_match(match_all, "alpha alpha", {1, 0, 0, 1, 0});
    check_match(match_regexp, "^alpha$", {1, 0, 0, 1, 0});
    check_match(match_phrase, "alpha token", {1, 0, 0, 1, 0});
    check_match(match_phrase, "token tail", {1, 0, 0, 0, 0});
    check_match(match_phrase_prefix, "alpha tok", {1, 0, 0, 1, 0});
    check_match(match_phrase_prefix, "token tai", {1, 0, 0, 0, 0});
    check_match(match_phrase_edge, "pha tok", {1, 0, 0, 1, 0});
    check_match(match_phrase_edge, "ken tai", {1, 0, 0, 0, 0});

    segment_v2::inverted_index::CustomAnalyzerConfig::Builder builder;
    builder.with_tokenizer_config("keyword", {});
    builder.add_token_filter_config("lowercase", {});
    auto provider =
            std::make_shared<segment_v2::inverted_index::CustomAnalyzerProvider>(builder.build());
    ctx.ctx->parser_type = InvertedIndexParserType::PARSER_NONE;
    ctx.ctx->analyzer_name = "custom_keyword_lowercase";
    ctx.ctx->analyzer_provider = provider;
    ctx.ctx->analyzer = provider->get_analyzer();
    check_match(match_any, "ALPHA TOKEN", {1, 0, 0, 1, 0});
    check_match(match_phrase, "ALPHA TOKEN", {1, 0, 0, 1, 0});

    ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_NONE);
    check_match(match_any, "alpha token", {1, 0, 0, 1, 0});
    check_match(match_all, "alpha token", {1, 0, 0, 1, 0});
    check_match(match_phrase, "alpha token", {1, 0, 0, 1, 0});
    check_match(match_phrase_prefix, "alpha", {1, 0, 0, 1, 0});
    check_match(match_phrase_edge, "pha", {1, 0, 0, 1, 0});
    check_match(match_any, "", {0, 0, 0, 1, 0});
}

TEST(FunctionMatchTest, array_phrase_spans_non_null_elements) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    auto string_col = ColumnString::create();
    for (const std::string value :
         {"hello", "world", "hello", "hidden", "world", "hello", "there", "world"}) {
        string_col->insert_data(value.data(), value.size());
    }
    ColumnArray::Offsets64 offsets = {2, 5, 8};
    ColumnUInt8::Container null_map = {0, 0, 0, 1, 0, 0, 0, 0};

    auto check_match = [&](const FunctionMatchBase& function, const std::string& query,
                           const std::vector<uint8_t>& expected) {
        SCOPED_TRACE(function.get_name() + ": " + query);
        ColumnUInt8::Container result(expected.size(), 0);
        ASSERT_TRUE(function.execute_match(context.get(), "tags", query, expected.size(),
                                           string_col.get(), ctx.ctx.get(), &offsets, result,
                                           &null_map)
                            .ok());
        for (size_t row = 0; row < expected.size(); ++row) {
            EXPECT_EQ(result[row], expected[row]) << "row: " << row;
        }
    };
    check_match(FunctionMatchPhrase {}, "hello world", {1, 1, 0});
    check_match(FunctionMatchPhrasePrefix {}, "hello wor", {1, 1, 0});
    check_match(FunctionMatchPhraseEdge {}, "llo wor", {1, 1, 0});
    check_match(FunctionMatchPhrase {}, "hello hidden world", {0, 0, 0});
}

TEST(FunctionMatchTest, array_phrase_respects_analyzer_positions) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});

    segment_v2::inverted_index::Settings tokenizer_settings;
    tokenizer_settings.set("tokenize_on_chars", "[whitespace]");
    segment_v2::inverted_index::Settings delimiter_settings;
    delimiter_settings.set("preserve_original", "true");
    segment_v2::inverted_index::CustomAnalyzerConfig::Builder builder;
    builder.with_tokenizer_config("char_group", tokenizer_settings);
    builder.add_token_filter_config("word_delimiter", delimiter_settings);
    auto provider =
            std::make_shared<segment_v2::inverted_index::CustomAnalyzerProvider>(builder.build());
    auto ctx = std::make_unique<InvertedIndexAnalyzerCtx>();
    ctx->analyzer_name = "word_delimiter_with_original";
    ctx->analyzer_provider = provider;
    ctx->analyzer = provider->get_analyzer();

    auto string_col = ColumnString::create();
    string_col->insert_data("foo-bar", 7);
    string_col->insert_data("baz", 3);
    ColumnArray::Offsets64 offsets = {2};

    FunctionMatchPhrase match_phrase;
    int32_t unused_offset = 0;
    auto data_tokens = match_phrase.analyse_data_token("tags", ctx.get(), string_col.get(), 0,
                                                       nullptr, unused_offset);
    ASSERT_EQ(data_tokens.size(), 3);
    EXPECT_EQ(data_tokens[0].position, data_tokens[1].position);
    auto query_tokens = match_phrase.analyse_query_str_token(ctx.get(), "foo-bar baz", "tags");
    ASSERT_EQ(query_tokens.size(), 4);

    ColumnUInt8::Container result(1, 0);
    ASSERT_TRUE(match_phrase
                        .execute_match(context.get(), "tags", "foo-bar baz", 1, string_col.get(),
                                       ctx.get(), &offsets, result)
                        .ok());
    EXPECT_EQ(result[0], 0);
    result[0] = 0;
    ASSERT_TRUE(match_phrase
                        .execute_match(context.get(), "tags", "foo bar baz", 1, string_col.get(),
                                       ctx.get(), &offsets, result)
                        .ok());
    EXPECT_EQ(result[0], 1);
}

TEST(FunctionMatchTest, array_phrase_preserves_leading_analyzer_gap) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});

    segment_v2::inverted_index::Settings tokenizer_settings;
    tokenizer_settings.set("tokenize_on_chars", "[whitespace]");
    segment_v2::inverted_index::CustomAnalyzerConfig::Builder builder;
    builder.with_tokenizer_config("char_group", tokenizer_settings);
    builder.add_token_filter_config("word_delimiter", {});
    auto provider =
            std::make_shared<segment_v2::inverted_index::CustomAnalyzerProvider>(builder.build());
    auto ctx = std::make_unique<InvertedIndexAnalyzerCtx>();
    ctx->analyzer_name = "word_delimiter";
    ctx->analyzer_provider = provider;
    ctx->analyzer = provider->get_analyzer();

    auto string_col = ColumnString::create();
    string_col->insert_data("hello", 5);
    string_col->insert_data("--- world", 9);
    ColumnArray::Offsets64 offsets = {2};

    FunctionMatchPhrase match_phrase;
    int32_t unused_offset = 0;
    auto later_tokens = match_phrase.analyse_data_token("tags", ctx.get(), string_col.get(), 1,
                                                        nullptr, unused_offset);
    ASSERT_EQ(later_tokens.size(), 1);
    EXPECT_EQ(later_tokens[0].position, 2);

    ColumnUInt8::Container result(1, 0);
    ASSERT_TRUE(match_phrase
                        .execute_match(context.get(), "tags", "hello world", 1, string_col.get(),
                                       ctx.get(), &offsets, result)
                        .ok());
    EXPECT_EQ(result[0], 0);
}

TEST(FunctionMatchTest, long_unrelated_phrase_miss) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    std::string unrelated;
    std::string matching;
    for (int i = 0; i < 1000; ++i) {
        unrelated += "gamma ";
        matching += "alpha ";
    }
    auto string_col = ColumnString::create();
    string_col->insert_data(unrelated.data(), unrelated.size());
    string_col->insert_data(matching.data(), matching.size());
    string_col->insert_data(unrelated.data(), unrelated.size());
    ColumnUInt8::Container result(3, 0);
    FunctionMatchPhrase match_phrase;
    ASSERT_TRUE(match_phrase
                        .execute_match(context.get(), "tags", matching, 3, string_col.get(),
                                       ctx.ctx.get(), nullptr, result)
                        .ok());
    EXPECT_EQ(result, (ColumnUInt8::Container {0, 1, 0}));
}

TEST(FunctionMatchTest, long_phrase_with_many_short_rows) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    std::string query;
    for (int i = 0; i < 1000; ++i) {
        query += "alpha ";
    }
    auto string_col = ColumnString::create();
    for (int i = 0; i < 2000; ++i) {
        string_col->insert_data("alpha", 5);
    }
    ColumnUInt8::Container result(2000, 0);
    FunctionMatchPhrase match_phrase;
    ASSERT_TRUE(match_phrase
                        .execute_match(context.get(), "tags", query, 2000, string_col.get(),
                                       ctx.ctx.get(), nullptr, result)
                        .ok());
    EXPECT_EQ(std::count(result.begin(), result.end(), 1), 0);
}

TEST(FunctionMatchTest, long_phrase_with_unique_start_miss) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    std::string value = "beta ";
    for (int i = 0; i < 1000; ++i) {
        value += "alpha ";
    }
    auto string_col = ColumnString::create();
    string_col->insert_data(value.data(), value.size());
    ColumnUInt8::Container result(1, 0);
    FunctionMatchPhrase match_phrase;
    ASSERT_TRUE(match_phrase
                        .execute_match(context.get(), "tags", value + "gamma", 1, string_col.get(),
                                       ctx.ctx.get(), nullptr, result)
                        .ok());
    EXPECT_EQ(result[0], 0);
}

TEST(FunctionMatchTest, keyword_array_many_elements) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_NONE);

    auto string_col = ColumnString::create();
    for (int i = 0; i < 1000; ++i) {
        string_col->insert_data("alpha", 5);
    }
    string_col->insert_data("omega", 5);
    ColumnArray::Offsets64 offsets = {1000, 1001};
    ColumnUInt8::Container result(2, 0);
    FunctionMatchAny match_any;
    ASSERT_TRUE(match_any
                        .execute_match(context.get(), "tags", "omega", 2, string_col.get(),
                                       ctx.ctx.get(), &offsets, result)
                        .ok());
    EXPECT_EQ(result, (ColumnUInt8::Container {0, 1}));
}

TEST(FunctionMatchTest, long_repeated_phrase_miss) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    std::string value;
    for (int i = 0; i < 10000; ++i) {
        value += "alpha ";
    }
    std::string query;
    for (int i = 0; i < 999; ++i) {
        query += "alpha ";
    }
    query += "beta";

    auto string_col = ColumnString::create();
    string_col->insert_data(value.data(), value.size());
    ColumnUInt8::Container result(1, 0);
    FunctionMatchPhrase match_phrase;
    ASSERT_TRUE(match_phrase
                        .execute_match(context.get(), "tags", query, 1, string_col.get(),
                                       ctx.ctx.get(), nullptr, result)
                        .ok());
    EXPECT_EQ(result[0], 0);
}

TEST(FunctionMatchTest, long_phrase_modes_cross_bitset_word) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto context = FunctionContext::create_context(&runtime_state, {}, {});
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    std::string value;
    std::string phrase;
    std::string edge = "pha ";
    for (int i = 0; i < 64; ++i) {
        value += "alpha ";
        phrase += "alpha ";
        if (i > 0) {
            edge += "alpha ";
        }
    }
    value += "beta";
    phrase += "beta";
    edge += "bet";

    auto string_col = ColumnString::create();
    string_col->insert_data(value.data(), value.size());
    auto check_match = [&](const FunctionMatchBase& function, const std::string& query) {
        SCOPED_TRACE(function.get_name());
        ColumnUInt8::Container result(1, 0);
        ASSERT_TRUE(function.execute_match(context.get(), "tags", query, 1, string_col.get(),
                                           ctx.ctx.get(), nullptr, result)
                            .ok());
        EXPECT_EQ(result[0], 1);
    };
    check_match(FunctionMatchPhrase {}, phrase);
    check_match(FunctionMatchPhrasePrefix {}, phrase.substr(0, phrase.size() - 1));
    check_match(FunctionMatchPhraseEdge {}, edge);
}

TEST(FunctionMatchTest, null_array_elements_do_not_match_hidden_payload) {
    TQueryOptions query_options;
    query_options.__set_enable_match_without_inverted_index(true);
    RuntimeState runtime_state(query_options, TQueryGlobals {});
    auto array_type =
            std::make_shared<DataTypeArray>(make_nullable(std::make_shared<DataTypeString>()));
    auto string_type = std::make_shared<DataTypeString>();
    auto context = FunctionContext::create_context(
            &runtime_state, std::make_shared<DataTypeUInt8>(), {array_type, string_type});
    auto analyzer_ctx = std::make_shared<InvertedIndexAnalyzerCtx>();
    analyzer_ctx->parser_type = InvertedIndexParserType::PARSER_NONE;
    context->set_function_state(FunctionContext::FRAGMENT_LOCAL, analyzer_ctx);

    auto check_match = [&](const FunctionMatchBase& function) {
        auto values = ColumnString::create();
        for (const std::string value : {"alpha", "tail", "alpha", "alpha", "tail", "alpha"}) {
            values->insert_data(value.data(), value.size());
        }
        auto null_map = ColumnUInt8::create();
        const uint8_t nulls[] = {1, 0, 1, 0, 0, 1};
        for (uint8_t is_null : nulls) {
            null_map->insert_value(is_null);
        }
        auto offsets = ColumnArray::ColumnOffsets::create();
        for (uint64_t offset : {2, 4, 6}) {
            offsets->insert_value(offset);
        }
        auto query = ColumnString::create();
        for (size_t row = 0; row < 3; ++row) {
            query->insert_data("alpha", 5);
        }

        Block block;
        block.insert(
                {ColumnArray::create(ColumnNullable::create(std::move(values), std::move(null_map)),
                                     std::move(offsets)),
                 array_type, "tags"});
        block.insert({std::move(query), string_type, "query"});
        block.insert({nullptr, std::make_shared<DataTypeUInt8>(), "result"});
        ASSERT_TRUE(function.execute_impl(context.get(), block, {0, 1}, 2, 3).ok());

        const auto& result = assert_cast<const ColumnUInt8&>(*block.get_by_position(2).column);
        SCOPED_TRACE(function.get_name());
        EXPECT_EQ(result.get_element(0), 0);
        EXPECT_EQ(result.get_element(1), 1);
        EXPECT_EQ(result.get_element(2), 0);
    };

    check_match(FunctionMatchAny {});
    check_match(FunctionMatchAll {});
    check_match(FunctionMatchRegexp {});
    check_match(FunctionMatchPhrase {});
    check_match(FunctionMatchPhrasePrefix {});
    check_match(FunctionMatchPhraseEdge {});
}

// Test Unicode and special character handling
TEST(FunctionMatchTest, unicode_and_special_chars) {
    FunctionMatchAny match_any;

    auto string_col = ColumnString::create();
    string_col->insert_data("测试文本", 12);        // Chinese text
    string_col->insert_data("café résumé", 12);     // French accents
    string_col->insert_data("🎵🎶🎸", 12);          // Emojis
    string_col->insert_data("user@domain.com", 15); // Email
    string_col->insert_data("C++ programming", 15); // Special chars

    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    for (int i = 0; i < 5; ++i) {
        int32_t offset = 0;
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), i,
                                                   nullptr, offset);
        // Should handle all text types without crashing
        EXPECT_GE(tokens.size(), 0);
    }
}

// Test performance with large data
TEST(FunctionMatchTest, performance_large_data) {
    FunctionMatchAny match_any;

    auto string_col = ColumnString::create();

    // Insert large text data
    std::string large_text(10000, 'x');
    large_text += " test keyword ";
    large_text += std::string(10000, 'y');

    string_col->insert_data(large_text.c_str(), large_text.length());

    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    // Test should complete without timeout or crash
    int32_t offset = 0;
    auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                               nullptr, offset);
    EXPECT_GT(tokens.size(), 0);
}

// Test different analyzer types
TEST(FunctionMatchTest, different_analyzer_types) {
    constexpr static uint32_t MAX_PATH_LEN = 1024;
    char buffer[MAX_PATH_LEN];
    EXPECT_NE(getcwd(buffer, MAX_PATH_LEN), nullptr);
    std::string _current_dir = std::string(buffer);
    config::inverted_index_dict_path =
            _current_dir + "/contrib/clucene/src/contribs-lib/CLucene/analysis/jieba/dict";
    FunctionMatchAny match_any;

    auto string_col = ColumnString::create();
    string_col->insert_data("The Quick Brown Fox Jumps", 25);

    // Test different parser types
    std::vector<InvertedIndexParserType> parser_types = {
            InvertedIndexParserType::PARSER_NONE, InvertedIndexParserType::PARSER_ENGLISH,
            InvertedIndexParserType::PARSER_CHINESE, InvertedIndexParserType::PARSER_STANDARD};

    for (auto parser_type : parser_types) {
        auto ctx = create_inverted_index_ctx(parser_type);
        int32_t offset = 0;
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                                   nullptr, offset);

        if (parser_type == InvertedIndexParserType::PARSER_NONE) {
            EXPECT_EQ(tokens.size(), 1); // Should be one token
        } else {
            EXPECT_GT(tokens.size(), 1); // Should be multiple tokens
        }
    }
}

// Test inverted index evaluation (simulate basic functionality)
TEST(FunctionMatchTest, evaluate_inverted_index_basic) {
    FunctionMatchAny match_any;

    // Test basic inverted index query type mapping
    auto query_type = match_any.get_query_type_from_fn_name();
    EXPECT_EQ(query_type, doris::segment_v2::InvertedIndexQueryType::MATCH_ANY_QUERY);

    // Test with different match functions
    FunctionMatchAll match_all;
    FunctionMatchPhrase match_phrase;
    FunctionMatchPhrasePrefix match_phrase_prefix;
    FunctionMatchRegexp match_regexp;
    FunctionMatchPhraseEdge match_phrase_edge;

    EXPECT_EQ(match_all.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_ALL_QUERY);
    EXPECT_EQ(match_phrase.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_QUERY);
    EXPECT_EQ(match_phrase_prefix.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_PREFIX_QUERY);
    EXPECT_EQ(match_regexp.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_REGEXP_QUERY);
    EXPECT_EQ(match_phrase_edge.get_query_type_from_fn_name(),
              doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_EDGE_QUERY);
}

// Test check function with different error conditions
TEST(FunctionMatchTest, check_function_error_handling) {
    FunctionMatchAny match_any;

    // Note: The actual check function requires proper runtime state setup
    // This test verifies the function exists and can be called
    // In real scenarios, it would test enable_match_without_inverted_index option

    // Test that the check function is implemented
    EXPECT_TRUE(true); // Placeholder - actual implementation would test error scenarios
}

// Test execute_impl basic structure
TEST(FunctionMatchTest, execute_impl_structure) {
    FunctionMatchAny match_any;

    // Test that execute_impl method exists and has the correct signature
    // Note: Full testing would require proper Block and FunctionContext setup

    // Create basic block structure
    Block block;
    ColumnNumbers arguments = {0, 1}; // column indices
    uint32_t result_col = 2;          // result column index
    size_t input_rows_count = 5;

    // This test verifies the method signature exists
    // Actual execution would require full runtime context
    (void)arguments;        // Mark as used
    (void)result_col;       // Mark as used
    (void)input_rows_count; // Mark as used
    EXPECT_TRUE(true);      // Placeholder for structure verification
}

// Test custom analyzer support
TEST(FunctionMatchTest, custom_analyzer_handling) {
    FunctionMatchAny match_any;

    segment_v2::inverted_index::CustomAnalyzerConfig::Builder builder;
    builder.with_tokenizer_config("keyword", {});
    builder.add_token_filter_config("lowercase", {});
    auto provider =
            std::make_shared<segment_v2::inverted_index::CustomAnalyzerProvider>(builder.build());

    InvertedIndexAnalyzerCtx analyzer_ctx;
    analyzer_ctx.analyzer_name = "custom_keyword_lowercase";
    analyzer_ctx.parser_type = InvertedIndexParserType::PARSER_NONE;
    analyzer_ctx.analyzer_provider = provider;
    analyzer_ctx.analyzer = provider->get_analyzer();

    auto query_tokens = match_any.analyse_query_str_token(&analyzer_ctx, "TEST QUERY", "test_col");
    ASSERT_EQ(query_tokens.size(), 1);
    ASSERT_TRUE(query_tokens[0].is_single_term());
    EXPECT_EQ(query_tokens[0].get_single_term(), "test query");

    auto string_col = ColumnString::create();
    string_col->insert_data("TEST QUERY", 10);
    int32_t offset = 0;
    auto data_tokens = match_any.analyse_data_token("test_col", &analyzer_ctx, string_col.get(), 0,
                                                    nullptr, offset);
    ASSERT_EQ(data_tokens.size(), 1);
    ASSERT_TRUE(data_tokens[0].is_single_term());
    EXPECT_EQ(data_tokens[0].get_single_term(), "test query");
}

// Test column type validation
TEST(FunctionMatchTest, column_type_validation) {
    FunctionMatchAny match_any;

    // Test with different column types
    auto string_col = ColumnString::create();
    string_col->insert_data("test data", 9);

    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    // Test with valid string column
    int32_t offset = 0;
    auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                               nullptr, offset);
    EXPECT_GT(tokens.size(), 0);

    // Additional column type tests would go here
    // (testing with non-string columns, nullable columns, etc.)
}

// Test phrase query validation
TEST(FunctionMatchTest, phrase_query_validation) {
    FunctionMatchPhrase match_phrase;
    FunctionMatchPhrasePrefix match_phrase_prefix;
    FunctionMatchPhraseEdge match_phrase_edge;

    // These functions require phrase support in the index
    // Test that they have proper validation logic

    auto phrase_query_type = match_phrase.get_query_type_from_fn_name();
    auto prefix_query_type = match_phrase_prefix.get_query_type_from_fn_name();
    auto edge_query_type = match_phrase_edge.get_query_type_from_fn_name();

    EXPECT_EQ(phrase_query_type, doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_QUERY);
    EXPECT_EQ(prefix_query_type,
              doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_PREFIX_QUERY);
    EXPECT_EQ(edge_query_type, doris::segment_v2::InvertedIndexQueryType::MATCH_PHRASE_EDGE_QUERY);
}

// Test regex compilation and error handling
TEST(FunctionMatchTest, regex_compilation_handling) {
    FunctionMatchRegexp match_regexp;

    auto string_col = ColumnString::create();
    string_col->insert_data("test123", 7);
    string_col->insert_data("abc456", 6);

    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_NONE);

    // Test data analysis (basic setup)
    int32_t offset = 0;
    auto tokens = match_regexp.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                                  nullptr, offset);
    EXPECT_GE(tokens.size(), 0);

    // Note: Full regex testing would require proper execute_match context
    // This tests the basic token analysis functionality
}

// Test memory management and cleanup
TEST(FunctionMatchTest, memory_management) {
    // Test that contexts are properly created and destroyed
    {
        auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);
        EXPECT_NE(ctx.ctx.get(), nullptr);
        EXPECT_NE(ctx.ctx->analyzer, nullptr);
    }

    {
        auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_NONE);
        EXPECT_NE(ctx.ctx.get(), nullptr);
        // analyzer should be nullptr for PARSER_NONE
    }

    // Test with multiple contexts
    std::vector<TestInvertedIndexCtx> contexts;
    for (int i = 0; i < 10; ++i) {
        contexts.push_back(create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH));
    }

    // Cleanup should happen automatically
    contexts.clear();
    EXPECT_TRUE(true); // No crashes expected
}

// Test concurrent access (basic thread safety)
TEST(FunctionMatchTest, basic_thread_safety) {
    FunctionMatchAny match_any;

    auto string_col = ColumnString::create();
    string_col->insert_data("concurrent test data", 20);

    // Test that multiple threads can use different contexts safely
    std::vector<std::thread> threads;
    std::atomic<int> success_count {0};

    for (int i = 0; i < 5; ++i) {
        threads.emplace_back([&match_any, &string_col, &success_count]() {
            try {
                auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);
                int32_t offset = 0;
                auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(),
                                                           string_col.get(), 0, nullptr, offset);
                if (!tokens.empty()) {
                    success_count++;
                }
            } catch (...) {
                // Should not throw
            }
        });
    }

    for (auto& thread : threads) {
        thread.join();
    }

    EXPECT_EQ(success_count.load(), 5);
}

// Test boundary conditions and edge cases
TEST(FunctionMatchTest, boundary_conditions) {
    FunctionMatchAny match_any;

    auto string_col = ColumnString::create();

    // Test with various boundary data
    string_col->insert_data("", 0);                                // Empty string
    string_col->insert_data("a", 1);                               // Single character
    string_col->insert_data(std::string(1000, 'x').c_str(), 1000); // Very long string
    string_col->insert_data("special chars: !@#$%^&*()", 25);      // Special characters
    string_col->insert_data("unicode: 测试 🎵", 15);               // Unicode

    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    // Test each boundary condition
    for (int i = 0; i < 5; ++i) {
        int32_t offset = 0;
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), i,
                                                   nullptr, offset);
        // Should handle all cases without crashing
        EXPECT_GE(tokens.size(), 0);
    }
}

// Test with nullable columns (if supported)
TEST(FunctionMatchTest, nullable_column_handling) {
    FunctionMatchAny match_any;

    // Create a regular string column
    auto string_col = ColumnString::create();
    string_col->insert_data("test data", 9);
    string_col->insert_data("more test", 9);

    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    // Test normal processing
    int32_t offset = 0;
    auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), 0,
                                               nullptr, offset);
    EXPECT_GT(tokens.size(), 0);

    // Note: Nullable column testing would require ColumnNullable setup
    // This test verifies basic column handling
}

// Test integration with query planner (mock)
TEST(FunctionMatchTest, query_planner_integration) {
    // Test that match functions can be properly identified and used

    FunctionMatchAny match_any;
    FunctionMatchAll match_all;
    FunctionMatchPhrase match_phrase;
    FunctionMatchPhrasePrefix match_phrase_prefix;
    FunctionMatchRegexp match_regexp;
    FunctionMatchPhraseEdge match_phrase_edge;

    // Verify function names are correct
    EXPECT_EQ(match_any.get_name(), "match_any");
    EXPECT_EQ(match_all.get_name(), "match_all");
    EXPECT_EQ(match_phrase.get_name(), "match_phrase");
    EXPECT_EQ(match_phrase_prefix.get_name(), "match_phrase_prefix");
    EXPECT_EQ(match_regexp.get_name(), "match_regexp");
    EXPECT_EQ(match_phrase_edge.get_name(), "match_phrase_edge");
}

// Test error propagation
TEST(FunctionMatchTest, error_propagation) {
    FunctionMatchAny match_any;

    // Test with invalid context
    auto tokens = match_any.analyse_query_str_token(nullptr, "test", "col");
    EXPECT_EQ(tokens.size(), 0); // Should handle gracefully

    // Test with invalid column data
    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);
    int32_t offset = 0;

    // Note: Full error testing would require actual invalid column scenarios
    // This tests basic error handling structure
    (void)ctx;    // Mark as used
    (void)offset; // Mark as used
    EXPECT_TRUE(true);
}

// Test performance characteristics
TEST(FunctionMatchTest, performance_characteristics) {
    FunctionMatchAny match_any;

    auto string_col = ColumnString::create();

    // Add many rows of data
    for (int i = 0; i < 1000; ++i) {
        std::string data = "test data row " + std::to_string(i);
        string_col->insert_data(data.c_str(), data.length());
    }

    auto ctx = create_inverted_index_ctx(InvertedIndexParserType::PARSER_ENGLISH);

    // Test processing time for large datasets
    auto start = std::chrono::high_resolution_clock::now();

    for (int i = 0; i < 100; ++i) { // Sample some rows
        int32_t offset = 0;
        auto tokens = match_any.analyse_data_token("test_col", ctx.ctx.get(), string_col.get(), i,
                                                   nullptr, offset);
        EXPECT_GT(tokens.size(), 0);
    }

    auto end = std::chrono::high_resolution_clock::now();
    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(end - start);

    // Should complete in reasonable time (less than 1 second for this test)
    EXPECT_LT(duration.count(), 1000);
}

// Test function registration and factory
TEST(FunctionMatchTest, function_registration) {
    // This test would verify that match functions are properly registered
    // in the function factory and can be retrieved by name

    // Note: Full testing would require access to SimpleFunctionFactory
    // This test verifies the concept exists
    EXPECT_TRUE(true);
}

namespace {

// A FULLTEXT reader that answers every query with an empty result and counts the queries it
// served, so a test can put indexes with different capabilities side by side on one column.
class CountingFulltextReader final : public segment_v2::InvertedIndexReader {
public:
    static std::shared_ptr<CountingFulltextReader> create(int64_t index_id,
                                                          const std::string& analyzer,
                                                          bool support_phrase) {
        TabletIndexPB pb;
        pb.set_index_id(index_id);
        pb.set_index_name("idx_" + analyzer);
        pb.set_index_type(IndexType::INVERTED);
        pb.add_col_unique_id(0);
        (*pb.mutable_properties())["analyzer"] = analyzer;
        (*pb.mutable_properties())["support_phrase"] = support_phrase ? "true" : "false";
        TabletIndex index;
        index.init_from_pb(pb);
        return std::shared_ptr<CountingFulltextReader>(new CountingFulltextReader(index));
    }

    segment_v2::InvertedIndexReaderType type() override {
        return segment_v2::InvertedIndexReaderType::FULLTEXT;
    }
    Status query(const segment_v2::IndexQueryContextPtr& /*context*/,
                 const std::string& /*column_name*/, const Field& /*query_value*/,
                 segment_v2::InvertedIndexQueryType /*query_type*/,
                 std::shared_ptr<roaring::Roaring>& /*bit_map*/,
                 const InvertedIndexAnalyzerCtx* /*analyzer_ctx*/) override {
        ++queries;
        return Status::OK();
    }
    Status try_query(const segment_v2::IndexQueryContextPtr& /*context*/,
                     const std::string& /*column_name*/, const Field& /*query_value*/,
                     segment_v2::InvertedIndexQueryType /*query_type*/, size_t* count) override {
        *count = 0;
        return Status::OK();
    }
    Status new_iterator(std::unique_ptr<segment_v2::IndexIterator>* /*iterator*/) override {
        return Status::OK();
    }

    int queries = 0;

private:
    explicit CountingFulltextReader(const TabletIndex& index)
            : segment_v2::InvertedIndexReader(&index, nullptr) {
        // No index file stands behind this reader, so there is no null bitmap to read either.
        set_has_null(false);
    }
};

} // namespace

// A column can carry a docs-only index -- a gram index is docs-only by default -- ahead of a
// positional one. MATCH_PHRASE ... USING ANALYZER names the positional index and the query runs
// on it, so phrase support has to be judged on that index. Judging it on the first FULLTEXT
// reader rejected a valid query purely because of the order the indexes were created in.
TEST(FunctionMatchTest, phrase_support_is_checked_on_the_index_the_analyzer_selects) {
    auto docs_only = CountingFulltextReader::create(1, "docs_only_analyzer", false);
    auto positional = CountingFulltextReader::create(2, "positional_analyzer", true);
    segment_v2::InvertedIndexIterator iterator;
    iterator.add_reader(segment_v2::InvertedIndexReaderType::FULLTEXT, docs_only);
    iterator.add_reader(segment_v2::InvertedIndexReaderType::FULLTEXT, positional);
    OlapReaderStatistics stats;
    auto context = std::make_shared<segment_v2::IndexQueryContext>();
    context->stats = &stats;
    iterator.set_context(context);

    const auto string_type = std::make_shared<DataTypeString>();
    const ColumnsWithTypeAndName arguments {
            {string_type->create_column_const(
                     1, Field::create_field<TYPE_STRING>(std::string("request ok"))),
             string_type, "query"}};
    const std::vector<IndexFieldNameAndTypePair> columns {{"msg", string_type}};
    InvertedIndexAnalyzerCtx analyzer_ctx;
    analyzer_ctx.analyzer_key = "positional_analyzer";

    FunctionMatchPhrase match_phrase;
    FunctionMatchPhrasePrefix match_phrase_prefix;
    for (const FunctionMatchBase* function :
         std::vector<const FunctionMatchBase*> {&match_phrase, &match_phrase_prefix}) {
        segment_v2::InvertedIndexResultBitmap bitmap;
        const auto status = function->evaluate_inverted_index(arguments, columns, {&iterator}, 4,
                                                              &analyzer_ctx, bitmap);
        EXPECT_TRUE(status.ok()) << function->get_name() << ": " << status;
    }
    EXPECT_EQ(positional->queries, 2);
    EXPECT_EQ(docs_only->queries, 0);

    // Naming the docs-only index still gets the phrase query refused.
    analyzer_ctx.analyzer_key = "docs_only_analyzer";
    segment_v2::InvertedIndexResultBitmap bitmap;
    const auto refused = match_phrase.evaluate_inverted_index(arguments, columns, {&iterator}, 4,
                                                              &analyzer_ctx, bitmap);
    EXPECT_TRUE(refused.is<ErrorCode::INDEX_INVALID_PARAMETERS>()) << refused;
    EXPECT_EQ(docs_only->queries, 0);
}

} // namespace doris