// Licensed to the LF AI & Data foundation 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 <boost/container/vector.hpp>
#include <folly/FBVector.h>
#include <gtest/gtest.h>
#include <algorithm>
#include <array>
#include <cstdint>
#include <functional>
#include <map>
#include <memory>
#include <string>
#include <string_view>
#include <tuple>
#include <utility>
#include <vector>

#include "bitset/bitset.h"
#include "common/Array.h"
#include "common/Consts.h"
#include "common/IndexMeta.h"
#include "common/Schema.h"
#include "common/Types.h"
#include "common/Utils.h"
#include "common/Vector.h"
#include "common/protobuf_utils.h"
#include "exec/expression/EvalCtx.h"
#include "expr/ITypeExpr.h"
#include "filemanager/InputStream.h"
#include "gtest/gtest.h"
#include "index/BitmapIndex.h"
#include "knowhere/comp/index_param.h"
#include "pb/plan.pb.h"
#include "plan/PlanNode.h"
#include "query/ExecPlanNodeVisitor.h"
#include "query/Plan.h"
#include "query/PlanImpl.h"
#include "query/PlanNode.h"
#include "query/Utils.h"
#include "segcore/SegcoreConfig.h"
#include "segcore/SegmentGrowing.h"
#include "segcore/SegmentGrowingImpl.h"
#include "test_utils/DataGen.h"
#include "test_utils/GenExprProto.h"
#include "test_utils/storage_test_utils.h"
#include "test_utils/cachinglayer_test_utils.h"

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

namespace {

proto::plan::GenericValue
Int64Value(int64_t value) {
    proto::plan::GenericValue val;
    val.set_int64_val(value);
    return val;
}

void
SetInt64ArrayFieldData(GeneratedData& raw_data,
                       FieldId field_id,
                       const std::vector<std::vector<int64_t>>& rows) {
    ASSERT_EQ(raw_data.raw_->num_rows(), static_cast<int64_t>(rows.size()));
    for (int i = 0; i < raw_data.raw_->fields_data_size(); i++) {
        auto* field_data = raw_data.raw_->mutable_fields_data(i);
        if (field_data->field_id() != field_id.get()) {
            continue;
        }

        auto* arrays =
            field_data->mutable_scalars()->mutable_array_data()->mutable_data();
        arrays->Clear();
        for (const auto& row : rows) {
            auto* array_data = arrays->Add();
            array_data->mutable_long_data()->mutable_data()->Add(
                row.data(), row.data() + row.size());
        }
        return;
    }
    FAIL() << "field id not found: " << field_id.get();
}

void
AssertColumnVector(const ColumnVectorPtr& vec,
                   const std::vector<bool>& expected_bits,
                   const std::vector<bool>& expected_valid,
                   const std::string& label) {
    ASSERT_EQ(vec->size(), expected_bits.size()) << label;
    ASSERT_EQ(expected_bits.size(), expected_valid.size()) << label;
    BitsetTypeView bits(vec->GetRawData(), vec->size());
    BitsetTypeView valid(vec->GetValidRawData(), vec->size());
    for (size_t i = 0; i < expected_bits.size(); ++i) {
        EXPECT_EQ(bits[i], expected_bits[i]) << label << ", row " << i;
        EXPECT_EQ(valid[i], expected_valid[i]) << label << ", row " << i;
    }
}

}  // namespace

TEST(Expr, TestArraySubscriptMissingElementIsUnknown) {
    auto schema = std::make_shared<Schema>();
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    auto struct_array_fid = schema->AddDebugArrayField(
        "structA[price_array]", DataType::INT64, false);
    schema->set_primary_field_id(i64_fid);

    constexpr int N = 3;
    auto raw_data = DataGen(schema, N, 42, 0, 1, 2);
    std::vector<std::vector<int64_t>> rows = {{1, 2}, {}, {3}};
    SetInt64ArrayFieldData(raw_data, long_array_fid, rows);
    SetInt64ArrayFieldData(raw_data, struct_array_fid, rows);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    auto offset = seg->PreInsert(N);
    seg->Insert(offset,
                N,
                raw_data.row_ids_.data(),
                raw_data.timestamps_.data(),
                raw_data.raw_);
    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());
    ASSERT_NE(seg_promote, nullptr);

    struct Case {
        std::string name;
        milvus::expr::TypedExprPtr expr;
        std::vector<bool> bits;
        std::vector<bool> valid;
        std::vector<bool> not_bits;
        std::vector<bool> not_valid;
    };

    auto make_cases = [](const expr::ColumnInfo& column) {
        std::vector<proto::plan::GenericValue> term_values{Int64Value(2)};
        return std::vector<Case>{
            {"equal",
             std::make_shared<expr::UnaryRangeFilterExpr>(
                 column, proto::plan::OpType::Equal, Int64Value(2)),
             {true, false, false},
             {true, false, false},
             {false, false, false},
             {true, false, false}},
            {"not_equal",
             std::make_shared<expr::UnaryRangeFilterExpr>(
                 column, proto::plan::OpType::NotEqual, Int64Value(2)),
             {false, false, false},
             {true, false, false},
             {true, false, false},
             {true, false, false}},
            {"binary_range",
             std::make_shared<expr::BinaryRangeFilterExpr>(
                 column, Int64Value(1), Int64Value(3), false, false),
             {true, false, false},
             {true, false, false},
             {false, false, false},
             {true, false, false}},
            {"term",
             std::make_shared<expr::TermFilterExpr>(column, term_values),
             {true, false, false},
             {true, false, false},
             {false, false, false},
             {true, false, false}},
            {"arith",
             std::make_shared<expr::BinaryArithOpEvalRangeExpr>(
                 column,
                 proto::plan::OpType::Equal,
                 proto::plan::ArithOpType::Add,
                 Int64Value(3),
                 Int64Value(1)),
             {true, false, false},
             {true, false, false},
             {false, false, false},
             {true, false, false}},
        };
    };

    std::vector<std::pair<std::string, FieldId>> fields = {
        {"array", long_array_fid},
        {"struct_array", struct_array_fid},
    };
    for (const auto& [field_name, field_id] : fields) {
        auto column =
            expr::ColumnInfo(field_id, DataType::ARRAY, DataType::INT64, {"1"});
        for (const auto& testcase : make_cases(column)) {
            auto plan = std::make_shared<plan::FilterBitsNode>(
                DEFAULT_PLANNODE_ID, testcase.expr);
            auto vec = milvus::test::gen_filter_res(
                plan.get(), seg_promote, N, MAX_TIMESTAMP);
            AssertColumnVector(vec,
                               testcase.bits,
                               testcase.valid,
                               field_name + " " + testcase.name);

            exec::OffsetVector offsets = {1, 0, 2};
            auto offset_vec = milvus::test::gen_filter_res(
                plan.get(), seg_promote, N, MAX_TIMESTAMP, &offsets);
            AssertColumnVector(
                offset_vec,
                {testcase.bits[1], testcase.bits[0], testcase.bits[2]},
                {testcase.valid[1], testcase.valid[0], testcase.valid[2]},
                field_name + " " + testcase.name + " offsets");

            auto not_expr = std::make_shared<expr::LogicalUnaryExpr>(
                expr::LogicalUnaryExpr::OpType::LogicalNot, testcase.expr);
            auto not_plan = std::make_shared<plan::FilterBitsNode>(
                DEFAULT_PLANNODE_ID, not_expr);
            auto not_vec = milvus::test::gen_filter_res(
                not_plan.get(), seg_promote, N, MAX_TIMESTAMP);
            AssertColumnVector(not_vec,
                               testcase.not_bits,
                               testcase.not_valid,
                               "not " + field_name + " " + testcase.name);

            auto final =
                ExecuteQueryExpr(not_plan, seg_promote, N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            for (int i = 0; i < N; ++i) {
                EXPECT_EQ(final[i],
                          testcase.not_bits[i] && testcase.not_valid[i])
                    << "not " << field_name << " " << testcase.name << ", row "
                    << i;
            }
        }
    }
}

TEST(Expr, TestArrayElementPredicateWithoutNestedPathThrows) {
    auto schema = std::make_shared<Schema>();
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    schema->set_primary_field_id(i64_fid);

    constexpr int N = 2;
    auto raw_data = DataGen(schema, N, 42, 0, 1, 2);
    SetInt64ArrayFieldData(raw_data, long_array_fid, {{1, 2}, {3, 4}});

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    auto offset = seg->PreInsert(N);
    seg->Insert(offset,
                N,
                raw_data.row_ids_.data(),
                raw_data.timestamps_.data(),
                raw_data.raw_);
    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());
    ASSERT_NE(seg_promote, nullptr);

    auto whole_array_column =
        expr::ColumnInfo(long_array_fid, DataType::ARRAY, DataType::INT64);
    std::vector<proto::plan::GenericValue> term_values{Int64Value(2)};
    std::vector<std::pair<std::string, expr::TypedExprPtr>> cases = {
        {"unary",
         std::make_shared<expr::UnaryRangeFilterExpr>(
             whole_array_column, proto::plan::OpType::Equal, Int64Value(2))},
        {"binary_range",
         std::make_shared<expr::BinaryRangeFilterExpr>(
             whole_array_column, Int64Value(1), Int64Value(3), false, false)},
        {"term",
         std::make_shared<expr::TermFilterExpr>(whole_array_column,
                                                term_values)},
        {"arith",
         std::make_shared<expr::BinaryArithOpEvalRangeExpr>(
             whole_array_column,
             proto::plan::OpType::Equal,
             proto::plan::ArithOpType::Add,
             Int64Value(3),
             Int64Value(1))},
    };

    for (const auto& [name, typed_expr] : cases) {
        auto plan = std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
                                                           typed_expr);
        EXPECT_ANY_THROW({
            auto vec = milvus::test::gen_filter_res(
                plan.get(), seg_promote, N, MAX_TIMESTAMP);
            (void)vec;
        }) << name;
    }
}

TEST(Expr, TestArrayRange) {
    std::vector<std::tuple<std::string,
                           std::string,
                           std::function<bool(milvus::Array & array)>>>
        testcases = {
            // binary_range_expr: 1 < long_array[0] < 10000
            {"1 < long_array[0] < 10000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return 1 < val && val < 10000;
             }},
            // binary_range_expr: 1 < long_array[1024] < 10000
            {"1 < long_array[1024] < 10000",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return 1 < val && val < 10000;
             }},
            // binary_range_expr: 1 <= long_array[0] < 10000
            {"1 <= long_array[0] < 10000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return 1 <= val && val < 10000;
             }},
            // binary_range_expr: 1 <= long_array[1024] < 10000
            {"1 <= long_array[1024] < 10000",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return 1 <= val && val < 10000;
             }},
            // binary_range_expr: 1 < long_array[0] <= 10000
            {"1 < long_array[0] <= 10000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return 1 < val && val <= 10000;
             }},
            // binary_range_expr: 1 < long_array[1024] <= 10000
            {"1 < long_array[1024] <= 10000",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return 1 < val && val <= 10000;
             }},
            // binary_range_expr: 1 <= long_array[0] <= 10000
            {"1 <= long_array[0] <= 10000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return 1 <= val && val <= 10000;
             }},
            // binary_range_expr: 1 <= long_array[1024] <= 10000
            {"1 <= long_array[1024] <= 10000",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return 1 <= val && val <= 10000;
             }},
            // binary_range_expr: "aaa" <= string_array[0] <= "zzz"
            {R"("aaa" <= string_array[0] <= "zzz")",
             "string",
             [](milvus::Array& array) {
                 auto val = array.get_data<std::string_view>(0);
                 return "aaa" <= val && val <= "zzz";
             }},
            // binary_range_expr: 1.1 <= double_array[0] <= 2048.12
            {"1.1 <= double_array[0] <= 2048.12",
             "float",
             [](milvus::Array& array) {
                 auto val = array.get_data<double>(0);
                 return 1.1 <= val && val <= 2048.12;
             }},
            // unary_range_expr: long_array[0] >= 10000
            {"long_array[0] >= 10000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val >= 10000;
             }},
            // unary_range_expr: long_array[0] > 2000
            {"long_array[0] > 2000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val > 2000;
             }},
            // unary_range_expr: long_array[0] <= 2000
            {"long_array[0] <= 2000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val <= 2000;
             }},
            // unary_range_expr: long_array[0] < 2000
            {"long_array[0] < 2000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val < 2000;
             }},
            // unary_range_expr: long_array[0] == 2000
            {"long_array[0] == 2000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val == 2000;
             }},
            // unary_range_expr: long_array[0] != 2000
            {"long_array[0] != 2000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val != 2000;
             }},
            // unary_range_expr: bool_array[0] == false
            {"bool_array[0] == false",
             "bool",
             [](milvus::Array& array) {
                 auto val = array.get_data<bool>(0);
                 return !val;
             }},
            // unary_range_expr: string_array[0] == "abc"
            {R"(string_array[0] == "abc")",
             "string",
             [](milvus::Array& array) {
                 auto val = array.get_data<std::string_view>(0);
                 return val == "abc";
             }},
            // unary_range_expr: double_array[0] == 2.2
            {"double_array[0] == 2.2",
             "float",
             [](milvus::Array& array) {
                 auto val = array.get_data<double>(0);
                 return val == 2.2;
             }},
            // unary_range_expr: double_array[1024] == 2.2
            {"double_array[1024] == 2.2",
             "float",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val == 2.2;
             }},
            // unary_range_expr: double_array[1024] != 2.2
            {"double_array[1024] != 2.2",
             "float",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val != 2.2;
             }},
            // unary_range_expr: double_array[1024] >= 2.2
            {"double_array[1024] >= 2.2",
             "float",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val >= 2.2;
             }},
            // unary_range_expr: double_array[1024] > 2.2
            {"double_array[1024] > 2.2",
             "float",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val > 2.2;
             }},
            // unary_range_expr: double_array[1024] <= 2.2
            {"double_array[1024] <= 2.2",
             "float",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val <= 2.2;
             }},
            // unary_range_expr: double_array[1024] < 2.2
            {"double_array[1024] < 2.2",
             "float",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val < 2.2;
             }},

        };
    auto schema = std::make_shared<Schema>();
    schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    auto bool_array_fid =
        schema->AddDebugField("bool_array", DataType::ARRAY, DataType::BOOL);
    auto string_array_fid = schema->AddDebugField(
        "string_array", DataType::ARRAY, DataType::VARCHAR);
    auto float_array_fid =
        schema->AddDebugField("double_array", DataType::ARRAY, DataType::FLOAT);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::map<std::string, std::vector<ScalarFieldProto>> array_cols;
    int num_iters = 1;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter);
        auto new_long_array_col =
            raw_data.get_col<ScalarFieldProto>(long_array_fid);
        auto new_bool_array_col =
            raw_data.get_col<ScalarFieldProto>(bool_array_fid);
        auto new_string_array_col =
            raw_data.get_col<ScalarFieldProto>(string_array_fid);
        auto new_float_array_col =
            raw_data.get_col<ScalarFieldProto>(float_array_fid);

        array_cols["long"].insert(array_cols["long"].end(),
                                  new_long_array_col.begin(),
                                  new_long_array_col.end());
        array_cols["bool"].insert(array_cols["bool"].end(),
                                  new_bool_array_col.begin(),
                                  new_bool_array_col.end());
        array_cols["string"].insert(array_cols["string"].end(),
                                    new_string_array_col.begin(),
                                    new_string_array_col.end());
        array_cols["float"].insert(array_cols["float"].end(),
                                   new_float_array_col.begin(),
                                   new_float_array_col.end());

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

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());
    ScopedSchemaHandle schema_handle(*schema);
    for (auto [expr, array_type, ref_func] : testcases) {
        auto plan_str = schema_handle.ParseSearch(
            expr, "fakevec", 10, "L2", R"({"nprobe": 10})", 3);
        auto plan =
            CreateSearchPlanByExpr(schema, plan_str.data(), plan_str.size());
        BitsetType final;
        final = ExecuteQueryExpr(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0],
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0].get(),
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP,
            &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols[array_type][i]);
            auto ref = ref_func(array);
            ASSERT_EQ(ans, ref);
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], ref);
            }
        }
    }
}

TEST(Expr, TestArrayEqual) {
    std::vector<
        std::tuple<std::string, std::function<bool(std::vector<int64_t>)>>>
        testcases = {
            // unary_range_expr: long_array == [1, 2, 3]
            {"long_array == [1, 2, 3]",
             [](std::vector<int64_t> v) {
                 if (v.size() != 3) {
                     return false;
                 }
                 for (int i = 0; i < 3; ++i) {
                     if (v[i] != i + 1) {
                         return false;
                     }
                 }
                 return true;
             }},
            // unary_range_expr: long_array != [1, 2, 3]
            {"long_array != [1, 2, 3]",
             [](std::vector<int64_t> v) {
                 if (v.size() != 3) {
                     return true;
                 }
                 for (int i = 0; i < 3; ++i) {
                     if (v[i] != i + 1) {
                         return true;
                     }
                 }
                 return false;
             }},
        };
    auto schema = std::make_shared<Schema>();
    schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::vector<ScalarFieldProto> long_array_col;
    int num_iters = 1;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter, 0, 1, 3);
        auto new_long_array_col =
            raw_data.get_col<ScalarFieldProto>(long_array_fid);
        long_array_col.insert(long_array_col.end(),
                              new_long_array_col.begin(),
                              new_long_array_col.end());
        seg->PreInsert(N);
        seg->Insert(iter * N,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());
    ScopedSchemaHandle schema_handle(*schema);
    for (auto [expr, ref_func] : testcases) {
        auto plan_str = schema_handle.ParseSearch(
            expr, "fakevec", 10, "L2", R"({"nprobe": 10})", 3);
        auto plan =
            CreateSearchPlanByExpr(schema, plan_str.data(), plan_str.size());
        BitsetType final;
        final = ExecuteQueryExpr(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0],
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0].get(),
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP,
            &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(long_array_col[i]);
            std::vector<int64_t> array_values(array.length());
            for (int j = 0; j < array.length(); ++j) {
                array_values.push_back(array.get_data<int64_t>(j));
            }
            auto ref = ref_func(array_values);
            ASSERT_EQ(ans, ref);
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], ref);
            }
        }
    }
}

TEST(Expr, TestArrayNullExpr) {
    std::vector<std::tuple<std::string, std::function<bool(bool)>>> testcases =
        {
            // null_expr: long_array is null
            {"long_array is null", [](bool v) { return !v; }},
        };
    auto schema = std::make_shared<Schema>();
    schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid = schema->AddDebugField(
        "long_array", DataType::ARRAY, DataType::INT64, true);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::vector<ScalarFieldProto> long_array_col;
    int num_iters = 1;
    FixedVector<bool> valid_data;

    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter, 0, 1, 3);
        auto new_long_array_col =
            raw_data.get_col<ScalarFieldProto>(long_array_fid);
        long_array_col.insert(long_array_col.end(),
                              new_long_array_col.begin(),
                              new_long_array_col.end());
        auto new_valid_col = raw_data.get_col_valid(long_array_fid);
        valid_data.insert(
            valid_data.end(), new_valid_col.begin(), new_valid_col.end());
        seg->PreInsert(N);
        seg->Insert(iter * N,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());
    ScopedSchemaHandle schema_handle(*schema);
    for (auto [expr, ref_func] : testcases) {
        auto plan_str = schema_handle.ParseSearch(
            expr, "fakevec", 10, "L2", R"({"nprobe": 10})", 3);
        auto plan =
            CreateSearchPlanByExpr(schema, plan_str.data(), plan_str.size());
        BitsetType final;
        final = ExecuteQueryExpr(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0],
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0].get(),
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP,
            &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);
        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto valid = valid_data[i];
            auto ref = ref_func(valid);
            ASSERT_EQ(ans, ref);
        }
    }
}

TEST(Expr, TestArrayNullExprWithBitmapIndex) {
    auto schema = std::make_shared<Schema>();
    schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid = schema->AddDebugField(
        "long_array", DataType::ARRAY, DataType::INT64, true);
    schema->set_primary_field_id(i64_fid);

    constexpr int N = 128;
    auto raw_data = DataGen(schema, N, 44, 0, 1, 3);
    auto valid_data = raw_data.get_col_valid(long_array_fid);
    auto long_array_col = raw_data.get_col<ScalarFieldProto>(long_array_fid);

    auto segment = CreateSealedWithFieldDataLoaded(schema, raw_data);

    FixedVector<Array> arrays;
    arrays.reserve(N);
    std::vector<uint8_t> valid_bitmap((N + 7) / 8, 0);
    for (int i = 0; i < N; ++i) {
        arrays.emplace_back(long_array_col[i]);
        if (valid_data[i]) {
            valid_bitmap[i >> 3] |= 1 << (i & 0x07);
        }
    }

    auto field_data =
        storage::CreateFieldData(DataType::ARRAY, DataType::INT64, true);
    field_data->FillFieldData(arrays.data(), valid_bitmap.data(), N, 0);

    proto::schema::FieldSchema field_schema;
    field_schema.set_name("long_array");
    field_schema.set_fieldid(long_array_fid.get());
    field_schema.set_data_type(proto::schema::DataType::Array);
    field_schema.set_element_type(proto::schema::DataType::Int64);
    field_schema.set_nullable(true);
    storage::FileManagerContext ctx;
    ctx.fieldDataMeta = storage::FieldDataMeta{
        kCollectionID,
        kPartitionID,
        kSegmentID,
        long_array_fid.get(),
        field_schema,
    };
    ctx.indexMeta =
        storage::IndexMeta{kSegmentID, long_array_fid.get(), 4000, 4000};

    auto bitmap_index =
        std::make_unique<index::BitmapIndex<int64_t>>(ctx, false);
    bitmap_index->BuildWithFieldData({field_data});
    ASSERT_FALSE(bitmap_index->IsNestedIndex());
    ASSERT_EQ(bitmap_index->Count(), N);

    LoadIndexInfo load_index_info;
    load_index_info.field_id = long_array_fid.get();
    load_index_info.field_type = DataType::ARRAY;
    load_index_info.element_type = DataType::INT64;
    load_index_info.index_params = GenIndexParams(bitmap_index.get());
    load_index_info.cache_index =
        CreateTestCacheIndex("array_bitmap", std::move(bitmap_index));
    segment->LoadIndex(load_index_info);

    auto null_expr = std::make_shared<expr::NullExpr>(
        expr::ColumnInfo(
            long_array_fid, DataType::ARRAY, DataType::INT64, {}, true),
        proto::plan::NullExpr_NullOp_IsNull);
    auto plan =
        std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, null_expr);

    auto final = ExecuteQueryExpr(plan, segment.get(), N, MAX_TIMESTAMP);
    ASSERT_EQ(final.size(), N);
    for (int i = 0; i < N; ++i) {
        ASSERT_EQ(final[i], !valid_data[i]) << "row " << i;
    }

    milvus::exec::OffsetVector offsets;
    offsets.reserve(N / 2);
    for (int i = 0; i < N; ++i) {
        if (i % 2 == 0) {
            offsets.emplace_back(i);
        }
    }
    auto col_vec = milvus::test::gen_filter_res(
        plan.get(), segment.get(), N, MAX_TIMESTAMP, &offsets);
    BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
    ASSERT_EQ(view.size(), offsets.size());
    for (int i = 0; i < offsets.size(); ++i) {
        ASSERT_EQ(view[i], !valid_data[offsets[i]])
            << "offset row " << offsets[i];
    }
}

TEST(Expr, TestStructArrayParentNullExprUsesRepresentativeSubField) {
    auto schema = std::make_shared<Schema>();
    auto fakevec_fid = schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto profile_history_fid = schema->AddDebugField(
        "profile[history]", DataType::ARRAY, DataType::INT64, true);
    schema->set_primary_field_id(i64_fid);

    constexpr int N = 128;
    auto raw_data = DataGen(schema, N, 43, 0, 1, 2);
    auto valid_data = raw_data.get_col_valid(profile_history_fid);
    auto profile_history_col =
        raw_data.get_col<ScalarFieldProto>(profile_history_fid);

    auto growing = CreateGrowingSegment(schema, empty_index_meta);
    auto offset = growing->PreInsert(N);
    growing->Insert(offset,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    auto growing_segment = dynamic_cast<SegmentGrowingImpl*>(growing.get());
    ASSERT_NE(growing_segment, nullptr);

    FixedVector<Array> arrays;
    arrays.reserve(N);
    std::vector<uint8_t> valid_bitmap((N + 7) / 8, 0);
    for (int i = 0; i < N; ++i) {
        arrays.emplace_back(profile_history_col[i]);
        if (valid_data[i]) {
            valid_bitmap[i >> 3] |= 1 << (i & 0x07);
        }
    }

    auto field_data =
        storage::CreateFieldData(DataType::ARRAY, DataType::INT64, true);
    field_data->FillFieldData(arrays.data(), valid_bitmap.data(), N, 0);

    auto cm = storage::RemoteChunkManagerSingleton::GetInstance()
                  .GetRemoteChunkManager();
    auto sealed_segment = CreateSealedSegment(schema);
    auto field_data_info =
        PrepareSingleFieldInsertBinlog(kCollectionID,
                                       kPartitionID,
                                       kSegmentID,
                                       profile_history_fid.get(),
                                       {field_data},
                                       cm);
    sealed_segment->LoadFieldData(field_data_info);

    auto make_plan = [&](proto::plan::NullExpr_NullOp op) {
        auto null_expr = std::make_shared<expr::NullExpr>(
            expr::ColumnInfo(profile_history_fid,
                             DataType::ARRAY,
                             DataType::INT64,
                             {},
                             true),
            op);
        return std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
                                                      null_expr);
    };

    std::vector<
        std::pair<proto::plan::NullExpr_NullOp, std::function<bool(bool)>>>
        testcases = {
            {proto::plan::NullExpr_NullOp_IsNull,
             [](bool valid) { return !valid; }},
            {proto::plan::NullExpr_NullOp_IsNotNull,
             [](bool valid) { return valid; }},
        };

    for (auto [op, ref_func] : testcases) {
        std::array<const SegmentInternalInterface*, 2> segments = {
            static_cast<const SegmentInternalInterface*>(growing_segment),
            static_cast<const SegmentInternalInterface*>(sealed_segment.get())};
        for (auto* segment : segments) {
            auto plan = make_plan(op);
            auto final = ExecuteQueryExpr(plan, segment, N, MAX_TIMESTAMP);
            ASSERT_EQ(final.size(), N);
            for (int i = 0; i < N; ++i) {
                ASSERT_EQ(final[i], ref_func(valid_data[i]))
                    << "segment type " << segment->type() << ", row " << i;
            }

            milvus::exec::OffsetVector offsets;
            offsets.reserve(N / 2);
            for (int i = 0; i < N; ++i) {
                if (i % 2 == 0) {
                    offsets.emplace_back(i);
                }
            }
            auto col_vec = milvus::test::gen_filter_res(
                plan.get(), segment, N, MAX_TIMESTAMP, &offsets);
            BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
            ASSERT_EQ(view.size(), offsets.size());
            for (int i = 0; i < offsets.size(); ++i) {
                ASSERT_EQ(view[i], ref_func(valid_data[offsets[i]]))
                    << "segment type " << segment->type() << ", offset row "
                    << offsets[i];
            }
        }
    }

    (void)fakevec_fid;
}

TEST(Expr, TestVectorArrayNullExpr) {
    auto schema = std::make_shared<Schema>();
    auto fakevec_fid = schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto vector_array_fid =
        schema->AddDebugVectorArrayField("array_float_vector",
                                         DataType::VECTOR_FLOAT,
                                         4,
                                         knowhere::metric::L2,
                                         true);
    schema->set_primary_field_id(i64_fid);

    constexpr int N = 128;
    auto raw_data = DataGen(schema, N, 42, 0, 1, 2);
    auto valid_data = raw_data.get_col_valid(vector_array_fid);
    auto vector_array_col =
        raw_data.get_col<VectorFieldProto>(vector_array_fid);

    auto growing = CreateGrowingSegment(schema, empty_index_meta);
    auto offset = growing->PreInsert(N);
    growing->Insert(offset,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    auto growing_segment = dynamic_cast<SegmentGrowingImpl*>(growing.get());
    ASSERT_NE(growing_segment, nullptr);

    std::vector<uint8_t> valid_bitmap((N + 7) / 8, 0);
    std::vector<milvus::VectorArray> vector_arrays;
    vector_arrays.reserve(N);
    for (int i = 0; i < N; ++i) {
        if (valid_data[i]) {
            valid_bitmap[i >> 3] |= 1 << (i & 0x07);
            vector_arrays.emplace_back(vector_array_col[i]);
        }
    }

    auto field_data = storage::CreateFieldData(
        DataType::VECTOR_ARRAY, DataType::VECTOR_FLOAT, true, 4);
    field_data->FillFieldData(vector_arrays.data(), valid_bitmap.data(), N, 0);

    auto cm = storage::RemoteChunkManagerSingleton::GetInstance()
                  .GetRemoteChunkManager();
    auto sealed_segment = CreateSealedSegment(schema);
    auto field_data_info =
        PrepareSingleFieldInsertBinlog(kCollectionID,
                                       kPartitionID,
                                       kSegmentID,
                                       vector_array_fid.get(),
                                       {field_data},
                                       cm);
    sealed_segment->LoadFieldData(field_data_info);

    auto make_plan = [&](proto::plan::NullExpr_NullOp op) {
        auto null_expr = std::make_shared<expr::NullExpr>(
            expr::ColumnInfo(vector_array_fid,
                             DataType::VECTOR_ARRAY,
                             DataType::VECTOR_FLOAT,
                             {},
                             true),
            op);
        return std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
                                                      null_expr);
    };

    std::vector<
        std::pair<proto::plan::NullExpr_NullOp, std::function<bool(bool)>>>
        testcases = {
            {proto::plan::NullExpr_NullOp_IsNull,
             [](bool valid) { return !valid; }},
            {proto::plan::NullExpr_NullOp_IsNotNull,
             [](bool valid) { return valid; }},
        };

    for (auto [op, ref_func] : testcases) {
        std::array<const SegmentInternalInterface*, 2> segments = {
            static_cast<const SegmentInternalInterface*>(growing_segment),
            static_cast<const SegmentInternalInterface*>(sealed_segment.get())};
        for (auto* segment : segments) {
            auto plan = make_plan(op);
            auto final = ExecuteQueryExpr(plan, segment, N, MAX_TIMESTAMP);
            EXPECT_EQ(final.size(), N);
            for (int i = 0; i < N; ++i) {
                ASSERT_EQ(final[i], ref_func(valid_data[i]))
                    << "segment type " << segment->type() << ", row " << i;
            }

            milvus::exec::OffsetVector offsets;
            offsets.reserve(N / 2);
            for (int i = 0; i < N; ++i) {
                if (i % 2 == 0) {
                    offsets.emplace_back(i);
                }
            }
            auto col_vec = milvus::test::gen_filter_res(
                plan.get(), segment, N, MAX_TIMESTAMP, &offsets);
            BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
            ASSERT_EQ(view.size(), offsets.size());
            for (int i = 0; i < offsets.size(); ++i) {
                ASSERT_EQ(view[i], ref_func(valid_data[offsets[i]]))
                    << "segment type " << segment->type() << ", offset row "
                    << offsets[i];
            }
        }
    }

    (void)fakevec_fid;
}

TEST(Expr, TestVectorArrayLengthExpr) {
    auto schema = std::make_shared<Schema>();
    auto fakevec_fid = schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto vector_array_fid =
        schema->AddDebugVectorArrayField("array_float_vector",
                                         DataType::VECTOR_FLOAT,
                                         4,
                                         knowhere::metric::L2,
                                         true);
    schema->set_primary_field_id(i64_fid);

    constexpr int N = 128;
    constexpr int kArrayLen = 2;
    auto raw_data = DataGen(schema, N, 42, 0, 1, kArrayLen);
    auto valid_data = raw_data.get_col_valid(vector_array_fid);
    auto vector_array_col =
        raw_data.get_col<VectorFieldProto>(vector_array_fid);

    auto growing = CreateGrowingSegment(schema, empty_index_meta);
    auto offset = growing->PreInsert(N);
    growing->Insert(offset,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    auto growing_segment = dynamic_cast<SegmentGrowingImpl*>(growing.get());
    ASSERT_NE(growing_segment, nullptr);

    std::vector<uint8_t> valid_bitmap((N + 7) / 8, 0);
    std::vector<milvus::VectorArray> vector_arrays;
    vector_arrays.reserve(N);
    for (int i = 0; i < N; ++i) {
        if (valid_data[i]) {
            valid_bitmap[i >> 3] |= 1 << (i & 0x07);
            vector_arrays.emplace_back(vector_array_col[i]);
        }
    }

    auto field_data = storage::CreateFieldData(
        DataType::VECTOR_ARRAY, DataType::VECTOR_FLOAT, true, 4);
    field_data->FillFieldData(vector_arrays.data(), valid_bitmap.data(), N, 0);

    auto cm = storage::RemoteChunkManagerSingleton::GetInstance()
                  .GetRemoteChunkManager();
    auto sealed_segment = CreateSealedSegment(schema);
    auto field_data_info =
        PrepareSingleFieldInsertBinlog(kCollectionID,
                                       kPartitionID,
                                       kSegmentID,
                                       vector_array_fid.get(),
                                       {field_data},
                                       cm);
    sealed_segment->LoadFieldData(field_data_info);

    auto make_plan = [&](proto::plan::OpType op, int64_t target) {
        proto::plan::GenericValue value;
        value.set_int64_val(target);
        proto::plan::GenericValue right_operand;
        right_operand.set_int64_val(0);
        auto expr = std::make_shared<milvus::expr::BinaryArithOpEvalRangeExpr>(
            milvus::expr::ColumnInfo(vector_array_fid,
                                     DataType::VECTOR_ARRAY,
                                     DataType::VECTOR_FLOAT,
                                     {},
                                     true),
            op,
            proto::plan::ArithOpType::ArrayLength,
            value,
            right_operand);
        return std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID,
                                                      expr);
    };

    struct Testcase {
        proto::plan::OpType op;
        int64_t target;
        std::function<bool(bool)> ref_func;
    };
    std::vector<Testcase> testcases = {
        {proto::plan::OpType::Equal,
         kArrayLen,
         [](bool valid) { return valid; }},
        {proto::plan::OpType::NotEqual,
         kArrayLen + 1,
         [](bool valid) { return valid; }},
        {proto::plan::OpType::GreaterThan,
         kArrayLen - 1,
         [](bool valid) { return valid; }},
        {proto::plan::OpType::GreaterEqual,
         kArrayLen,
         [](bool valid) { return valid; }},
        {proto::plan::OpType::LessThan,
         kArrayLen,
         [](bool valid) { return false; }},
        {proto::plan::OpType::LessEqual,
         kArrayLen,
         [](bool valid) { return valid; }},
    };

    for (const auto& testcase : testcases) {
        std::array<const SegmentInternalInterface*, 2> segments = {
            static_cast<const SegmentInternalInterface*>(growing_segment),
            static_cast<const SegmentInternalInterface*>(sealed_segment.get())};
        for (auto* segment : segments) {
            auto plan = make_plan(testcase.op, testcase.target);
            auto final = ExecuteQueryExpr(plan, segment, N, MAX_TIMESTAMP);
            EXPECT_EQ(final.size(), N);
            for (int i = 0; i < N; ++i) {
                ASSERT_EQ(final[i], testcase.ref_func(valid_data[i]))
                    << "segment type " << segment->type() << ", row " << i;
            }

            milvus::exec::OffsetVector offsets;
            offsets.reserve(N / 2);
            for (int i = 0; i < N; ++i) {
                if (i % 2 == 0) {
                    offsets.emplace_back(i);
                }
            }
            auto col_vec = milvus::test::gen_filter_res(
                plan.get(), segment, N, MAX_TIMESTAMP, &offsets);
            BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
            ASSERT_EQ(view.size(), offsets.size());
            for (int i = 0; i < offsets.size(); ++i) {
                ASSERT_EQ(view[i], testcase.ref_func(valid_data[offsets[i]]))
                    << "segment type " << segment->type() << ", offset row "
                    << offsets[i];
            }
        }
    }

    (void)fakevec_fid;
}

TEST(Expr, PraseArrayContainsExpr) {
    // Test expressions for array_contains operations
    std::vector<const char*> exprs{
        // json_contains_expr: array_contains(array, 1)
        "array_contains(array, 1)",
        // json_contains_expr: array_contains_all(array, [1, 2, 3])
        "array_contains_all(array, [1, 2, 3])",
        // json_contains_expr: array_contains_any(array, [1, 2, 3])
        "array_contains_any(array, [1, 2, 3])",
    };

    auto schema = std::make_shared<Schema>();
    schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    schema->AddField(FieldName("array"),
                     FieldId(101),
                     DataType::ARRAY,
                     DataType::INT64,
                     false);
    ScopedSchemaHandle schema_handle(*schema);

    for (auto& expr : exprs) {
        auto plan_str = schema_handle.ParseSearch(
            expr, "fakevec", 10, "L2", R"({"nprobe": 10})", 3);
        auto plan =
            CreateSearchPlanByExpr(schema, plan_str.data(), plan_str.size());
    }
}

template <typename T>
struct ArrayTestcase {
    std::vector<T> term;
    std::vector<std::string> nested_path;
};

TEST(Expr, TestArrayContains) {
    auto schema = std::make_shared<Schema>();
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto int_array_fid =
        schema->AddDebugField("int_array", DataType::ARRAY, DataType::INT8);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    auto bool_array_fid =
        schema->AddDebugField("bool_array", DataType::ARRAY, DataType::BOOL);
    auto float_array_fid =
        schema->AddDebugField("float_array", DataType::ARRAY, DataType::FLOAT);
    auto double_array_fid = schema->AddDebugField(
        "double_array", DataType::ARRAY, DataType::DOUBLE);
    auto string_array_fid = schema->AddDebugField(
        "string_array", DataType::ARRAY, DataType::VARCHAR);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::map<std::string, std::vector<ScalarFieldProto>> array_cols;
    int num_iters = 1;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter);
        auto new_int_array_col =
            raw_data.get_col<ScalarFieldProto>(int_array_fid);
        auto new_long_array_col =
            raw_data.get_col<ScalarFieldProto>(long_array_fid);
        auto new_bool_array_col =
            raw_data.get_col<ScalarFieldProto>(bool_array_fid);
        auto new_float_array_col =
            raw_data.get_col<ScalarFieldProto>(float_array_fid);
        auto new_double_array_col =
            raw_data.get_col<ScalarFieldProto>(double_array_fid);
        auto new_string_array_col =
            raw_data.get_col<ScalarFieldProto>(string_array_fid);

        array_cols["int"].insert(array_cols["int"].end(),
                                 new_int_array_col.begin(),
                                 new_int_array_col.end());
        array_cols["long"].insert(array_cols["long"].end(),
                                  new_long_array_col.begin(),
                                  new_long_array_col.end());
        array_cols["bool"].insert(array_cols["bool"].end(),
                                  new_bool_array_col.begin(),
                                  new_bool_array_col.end());
        array_cols["float"].insert(array_cols["float"].end(),
                                   new_float_array_col.begin(),
                                   new_float_array_col.end());
        array_cols["double"].insert(array_cols["double"].end(),
                                    new_double_array_col.begin(),
                                    new_double_array_col.end());
        array_cols["string"].insert(array_cols["string"].end(),
                                    new_string_array_col.begin(),
                                    new_string_array_col.end());
        seg->PreInsert(N);
        seg->Insert(iter * N,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());

    std::vector<ArrayTestcase<bool>> bool_testcases{{{true, true}, {}},
                                                    {{false, false}, {}}};

    for (auto testcase : bool_testcases) {
        auto check = [&](const std::vector<bool>& values) {
            for (auto const& e : testcase.term) {
                if (std::find(values.begin(), values.end(), e) !=
                    values.end()) {
                    return true;
                }
            }
            return false;
        };
        std::vector<proto::plan::GenericValue> values;
        for (const auto& val : testcase.term) {
            proto::plan::GenericValue gen_val;
            gen_val.set_bool_val(val);
            values.push_back(gen_val);
        }
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(bool_array_fid, DataType::ARRAY, DataType::BOOL),
            proto::plan::JSONContainsExpr_JSONOp_Contains,
            true,
            values);
        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan.get(), seg_promote, N * num_iters, MAX_TIMESTAMP, &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols["bool"][i]);
            std::vector<bool> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<bool>(j));
            }
            ASSERT_EQ(ans, check(res)) << "@" << i;
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], check(res)) << "@" << i;
            }
        }
    }

    std::vector<ArrayTestcase<double>> double_testcases{
        {{1.123, 10.34}, {"double"}},
        {{10.34, 100.234}, {"double"}},
        {{100.234, 1000.4546}, {"double"}},
        {{1000.4546, 1.123}, {"double"}},
        {{1000.4546, 10.34}, {"double"}},
        {{1.123, 100.234}, {"double"}},
    };

    for (auto testcase : double_testcases) {
        auto check = [&](const std::vector<double>& values) {
            for (auto const& e : testcase.term) {
                if (std::find(values.begin(), values.end(), e) !=
                    values.end()) {
                    return true;
                }
            }
            return false;
        };

        std::vector<proto::plan::GenericValue> values;
        for (const auto& val : testcase.term) {
            proto::plan::GenericValue gen_val;
            gen_val.set_float_val(val);
            values.push_back(gen_val);
        }
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(
                double_array_fid, DataType::ARRAY, DataType::DOUBLE),
            proto::plan::JSONContainsExpr_JSONOp_Contains,
            true,
            values);
        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan.get(), seg_promote, N * num_iters, MAX_TIMESTAMP, &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols["double"][i]);
            std::vector<double> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<double>(j));
            }
            ASSERT_EQ(ans, check(res));
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], check(res));
            }
        }
    }

    for (auto testcase : double_testcases) {
        auto check = [&](const std::vector<float>& values) {
            for (auto const& e : testcase.term) {
                if (std::find(values.begin(), values.end(), e) !=
                    values.end()) {
                    return true;
                }
            }
            return false;
        };
        std::vector<proto::plan::GenericValue> values;
        for (const auto& val : testcase.term) {
            proto::plan::GenericValue gen_val;
            gen_val.set_float_val(val);
            values.push_back(gen_val);
        }
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(float_array_fid, DataType::ARRAY, DataType::FLOAT),
            proto::plan::JSONContainsExpr_JSONOp_Contains,
            true,
            values);
        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan.get(), seg_promote, N * num_iters, MAX_TIMESTAMP, &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols["float"][i]);
            std::vector<float> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<float>(j));
            }
            ASSERT_EQ(ans, check(res));
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], check(res));
            }
        }
    }

    std::vector<ArrayTestcase<int64_t>> testcases{
        {{1, 10}, {"int"}},
        {{10, 100}, {"int"}},
        {{100, 1000}, {"int"}},
        {{1000, 10}, {"int"}},
        {{2, 4, 6, 8, 10}, {"int"}},
        {{1, 2, 3, 4, 5}, {"int"}},
    };

    for (auto testcase : testcases) {
        auto check = [&](const std::vector<int64_t>& values) {
            for (auto const& e : testcase.term) {
                if (std::find(values.begin(), values.end(), e) ==
                    values.end()) {
                    return false;
                }
            }
            return true;
        };

        std::vector<proto::plan::GenericValue> values;
        for (const auto& val : testcase.term) {
            proto::plan::GenericValue gen_val;
            gen_val.set_int64_val(val);
            values.push_back(gen_val);
        }
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(int_array_fid, DataType::ARRAY, DataType::INT8),
            proto::plan::JSONContainsExpr_JSONOp_ContainsAll,
            true,
            values);
        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan.get(), seg_promote, N * num_iters, MAX_TIMESTAMP, &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols["int"][i]);
            std::vector<int64_t> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<int64_t>(j));
            }
            ASSERT_EQ(ans, check(res));
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], check(res));
            }
        }
    }

    for (auto testcase : testcases) {
        auto check = [&](const std::vector<int64_t>& values) {
            for (auto const& e : testcase.term) {
                if (std::find(values.begin(), values.end(), e) ==
                    values.end()) {
                    return false;
                }
            }
            return true;
        };

        std::vector<proto::plan::GenericValue> values;
        for (const auto& val : testcase.term) {
            proto::plan::GenericValue gen_val;
            gen_val.set_int64_val(val);
            values.push_back(gen_val);
        }
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(long_array_fid, DataType::ARRAY, DataType::INT64),
            proto::plan::JSONContainsExpr_JSONOp_ContainsAll,
            true,
            values);
        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan.get(), seg_promote, N * num_iters, MAX_TIMESTAMP, &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols["long"][i]);
            std::vector<int64_t> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<int64_t>(j));
            }
            ASSERT_EQ(ans, check(res));
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], check(res));
            }
        }
    }

    std::vector<ArrayTestcase<std::string>> testcases_string = {
        {{"1sads", "10dsf"}, {"string"}},
        {{"10dsf", "100"}, {"string"}},
        {{"100", "10dsf", "1sads"}, {"string"}},
        {{"100ddfdsssdfdsfsd0", "100"}, {"string"}},
    };

    for (auto testcase : testcases_string) {
        auto check = [&](const std::vector<std::string_view>& values) {
            for (auto const& e : testcase.term) {
                if (std::find(values.begin(), values.end(), e) ==
                    values.end()) {
                    return false;
                }
            }
            return true;
        };

        std::vector<proto::plan::GenericValue> values;
        for (const auto& val : testcase.term) {
            proto::plan::GenericValue gen_val;
            gen_val.set_string_val(val);
            values.push_back(gen_val);
        }
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(
                string_array_fid, DataType::ARRAY, DataType::VARCHAR),
            proto::plan::JSONContainsExpr_JSONOp_ContainsAll,
            true,
            values);
        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan.get(), seg_promote, N * num_iters, MAX_TIMESTAMP, &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols["string"][i]);
            std::vector<std::string_view> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<std::string_view>(j));
            }
            ASSERT_EQ(ans, check(res));
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], check(res));
            }
        }
    }
}

// Behavior coverage for ExecArrayContains across single- and multi-target
// invocations and across INT64 / VARCHAR element types. Exercises the typed
// cached set path (no MultiElement / variant round-trip).
TEST(Expr, TestArrayContainsTargetCoverage) {
    auto schema = std::make_shared<Schema>();
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    auto string_array_fid = schema->AddDebugField(
        "string_array", DataType::ARRAY, DataType::VARCHAR);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::map<std::string, std::vector<ScalarFieldProto>> array_cols;
    int num_iters = 1;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter);
        auto new_long_array_col =
            raw_data.get_col<ScalarFieldProto>(long_array_fid);
        auto new_string_array_col =
            raw_data.get_col<ScalarFieldProto>(string_array_fid);
        array_cols["long"].insert(array_cols["long"].end(),
                                  new_long_array_col.begin(),
                                  new_long_array_col.end());
        array_cols["string"].insert(array_cols["string"].end(),
                                    new_string_array_col.begin(),
                                    new_string_array_col.end());
        seg->PreInsert(N);
        seg->Insert(iter * N,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());

    // --- Single-target Contains on INT64 ---
    {
        int64_t target = 10;
        auto check = [target](const std::vector<int64_t>& values) {
            return std::find(values.begin(), values.end(), target) !=
                   values.end();
        };
        proto::plan::GenericValue gen_val;
        gen_val.set_int64_val(target);
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(long_array_fid, DataType::ARRAY, DataType::INT64),
            proto::plan::JSONContainsExpr_JSONOp_Contains,
            true,
            std::vector<proto::plan::GenericValue>{gen_val});
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        BitsetType final_bits =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final_bits.size(), N * num_iters);

        for (int i = 0; i < N * num_iters; ++i) {
            auto array = milvus::Array(array_cols["long"][i]);
            std::vector<int64_t> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<int64_t>(j));
            }
            ASSERT_EQ(final_bits[i], check(res)) << "@" << i;
        }
    }

    // --- Single-target Contains on VARCHAR ---
    {
        std::string target = "1sads";
        auto check = [&target](const std::vector<std::string_view>& values) {
            return std::find(values.begin(), values.end(), target) !=
                   values.end();
        };
        proto::plan::GenericValue gen_val;
        gen_val.set_string_val(target);
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(
                string_array_fid, DataType::ARRAY, DataType::VARCHAR),
            proto::plan::JSONContainsExpr_JSONOp_Contains,
            true,
            std::vector<proto::plan::GenericValue>{gen_val});
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        BitsetType final_bits =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final_bits.size(), N * num_iters);

        for (int i = 0; i < N * num_iters; ++i) {
            auto array = milvus::Array(array_cols["string"][i]);
            std::vector<std::string_view> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<std::string_view>(j));
            }
            ASSERT_EQ(final_bits[i], check(res)) << "@" << i;
        }
    }

    // --- ContainsAny with a one-element target list ---
    {
        int64_t target = 100;
        auto check = [target](const std::vector<int64_t>& values) {
            return std::find(values.begin(), values.end(), target) !=
                   values.end();
        };
        proto::plan::GenericValue gen_val;
        gen_val.set_int64_val(target);
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(long_array_fid, DataType::ARRAY, DataType::INT64),
            proto::plan::JSONContainsExpr_JSONOp_ContainsAny,
            true,
            std::vector<proto::plan::GenericValue>{gen_val});
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        BitsetType final_bits =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final_bits.size(), N * num_iters);

        for (int i = 0; i < N * num_iters; ++i) {
            auto array = milvus::Array(array_cols["long"][i]);
            std::vector<int64_t> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<int64_t>(j));
            }
            ASSERT_EQ(final_bits[i], check(res)) << "@" << i;
        }
    }

    // --- Multi-target ContainsAny on INT64 (no element should match all
    //     three targets, but each row may match any of them). ---
    {
        std::vector<int64_t> targets = {10, 100, 1000};
        auto check = [&targets](const std::vector<int64_t>& values) {
            for (auto t : targets) {
                if (std::find(values.begin(), values.end(), t) !=
                    values.end()) {
                    return true;
                }
            }
            return false;
        };
        std::vector<proto::plan::GenericValue> vals;
        for (auto t : targets) {
            proto::plan::GenericValue gv;
            gv.set_int64_val(t);
            vals.push_back(gv);
        }
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(long_array_fid, DataType::ARRAY, DataType::INT64),
            proto::plan::JSONContainsExpr_JSONOp_ContainsAny,
            true,
            vals);
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        BitsetType final_bits =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final_bits.size(), N * num_iters);

        for (int i = 0; i < N * num_iters; ++i) {
            auto array = milvus::Array(array_cols["long"][i]);
            std::vector<int64_t> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<int64_t>(j));
            }
            ASSERT_EQ(final_bits[i], check(res)) << "@" << i;
        }
    }

    // --- Multi-target ContainsAny on VARCHAR. ---
    {
        std::vector<std::string> targets = {"1sads", "10dsf", "100"};
        auto check = [&targets](const std::vector<std::string_view>& values) {
            for (auto const& t : targets) {
                if (std::find(values.begin(), values.end(), t) !=
                    values.end()) {
                    return true;
                }
            }
            return false;
        };
        std::vector<proto::plan::GenericValue> vals;
        for (const auto& t : targets) {
            proto::plan::GenericValue gv;
            gv.set_string_val(t);
            vals.push_back(gv);
        }
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(
                string_array_fid, DataType::ARRAY, DataType::VARCHAR),
            proto::plan::JSONContainsExpr_JSONOp_ContainsAny,
            true,
            vals);
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        BitsetType final_bits =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final_bits.size(), N * num_iters);

        for (int i = 0; i < N * num_iters; ++i) {
            auto array = milvus::Array(array_cols["string"][i]);
            std::vector<std::string_view> res;
            for (int j = 0; j < array.length(); ++j) {
                res.push_back(array.get_data<std::string_view>(j));
            }
            ASSERT_EQ(final_bits[i], check(res)) << "@" << i;
        }
    }
}

TEST(Expr, TestArrayContainsFloatLiteralCastsToElementType) {
    auto schema = std::make_shared<Schema>();
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto float_array_fid =
        schema->AddDebugField("float_array", DataType::ARRAY, DataType::FLOAT);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    constexpr int N = 3;
    auto raw_data = DataGen(schema, N, 42, 0, 1, 3);

    constexpr double target_1 = 492710.03520180425;
    constexpr double target_2 = 38241.68284883746;
    bool replaced_array_field = false;
    for (int i = 0; i < raw_data.raw_->fields_data_size(); ++i) {
        auto* field_data = raw_data.raw_->mutable_fields_data(i);
        if (field_data->field_id() != float_array_fid.get()) {
            continue;
        }

        auto* arrays =
            field_data->mutable_scalars()->mutable_array_data()->mutable_data();
        arrays->Clear();

        auto* matched = arrays->Add();
        matched->mutable_float_data()->add_data(static_cast<float>(target_1));
        matched->mutable_float_data()->add_data(static_cast<float>(target_2));

        auto* close_but_different = arrays->Add();
        close_but_different->mutable_float_data()->add_data(
            static_cast<float>(target_1) + 1.0F);
        close_but_different->mutable_float_data()->add_data(
            static_cast<float>(target_2) + 1.0F);

        arrays->Add();
        replaced_array_field = true;
        break;
    }
    ASSERT_TRUE(replaced_array_field);

    seg->PreInsert(N);
    seg->Insert(0,
                N,
                raw_data.row_ids_.data(),
                raw_data.timestamps_.data(),
                raw_data.raw_);
    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());
    ASSERT_NE(seg_promote, nullptr);

    auto make_float_value = [](double target) {
        proto::plan::GenericValue value;
        value.set_float_val(target);
        return value;
    };

    auto make_values = [&](std::vector<double> targets) {
        std::vector<proto::plan::GenericValue> values;
        values.reserve(targets.size());
        for (auto target : targets) {
            values.push_back(make_float_value(target));
        }
        return values;
    };

    auto expect_only_first_match =
        [&](const std::shared_ptr<plan::FilterBitsNode>& plan) {
            auto result = ExecuteQueryExpr(plan, seg_promote, N, MAX_TIMESTAMP);
            ASSERT_EQ(result.size(), N);
            EXPECT_TRUE(result[0]);
            EXPECT_FALSE(result[1]);
            EXPECT_FALSE(result[2]);
        };

    auto check = [&](proto::plan::JSONContainsExpr_JSONOp op,
                     std::vector<double> targets) {
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(float_array_fid, DataType::ARRAY, DataType::FLOAT),
            op,
            true,
            make_values(std::move(targets)));
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        expect_only_first_match(plan);
    };

    check(proto::plan::JSONContainsExpr_JSONOp_Contains, {target_1});
    check(proto::plan::JSONContainsExpr_JSONOp_ContainsAny,
          {target_1, target_2});
    check(proto::plan::JSONContainsExpr_JSONOp_ContainsAll,
          {target_1, target_2});

    auto element_expr = std::make_shared<milvus::expr::UnaryRangeFilterExpr>(
        expr::ColumnInfo(
            float_array_fid, DataType::ARRAY, DataType::FLOAT, {"0"}),
        proto::plan::OpType::Equal,
        make_float_value(target_1));
    expect_only_first_match(std::make_shared<plan::FilterBitsNode>(
        DEFAULT_PLANNODE_ID, element_expr));

    auto array_value = make_float_value(0);
    auto* array = array_value.mutable_array_val();
    array->set_same_type(true);
    array->set_element_type(proto::schema::DataType::Float);
    array->add_array()->set_float_val(target_1);
    array->add_array()->set_float_val(target_2);

    auto array_expr = std::make_shared<milvus::expr::UnaryRangeFilterExpr>(
        expr::ColumnInfo(float_array_fid, DataType::ARRAY, DataType::FLOAT),
        proto::plan::OpType::Equal,
        array_value);
    expect_only_first_match(std::make_shared<plan::FilterBitsNode>(
        DEFAULT_PLANNODE_ID, array_expr));
}

TEST(Expr, TestArrayContainsEmptyValues) {
    auto schema = std::make_shared<Schema>();
    auto int_array_fid =
        schema->AddDebugField("int_array", DataType::ARRAY, DataType::INT8);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    auto bool_array_fid =
        schema->AddDebugField("bool_array", DataType::ARRAY, DataType::BOOL);
    auto float_array_fid =
        schema->AddDebugField("float_array", DataType::ARRAY, DataType::FLOAT);
    auto double_array_fid = schema->AddDebugField(
        "double_array", DataType::ARRAY, DataType::DOUBLE);
    auto string_array_fid = schema->AddDebugField(
        "string_array", DataType::ARRAY, DataType::VARCHAR);
    schema->set_primary_field_id(schema->AddDebugField("id", DataType::INT64));
    std::vector<FieldId> fields = {
        int_array_fid,
        long_array_fid,
        bool_array_fid,
        float_array_fid,
        double_array_fid,
        string_array_fid,
    };

    auto dummy_seg = CreateGrowingSegment(schema, empty_index_meta);

    int N = 1000;
    std::vector<int> age_col;
    int num_iters = 100;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter);
        dummy_seg->PreInsert(N);
        dummy_seg->Insert(iter * N,
                          N,
                          raw_data.row_ids_.data(),
                          raw_data.timestamps_.data(),
                          raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(dummy_seg.get());
    std::vector<proto::plan::GenericValue> empty_values;

    for (auto field_id : fields) {
        auto expr = std::make_shared<milvus::expr::JsonContainsExpr>(
            expr::ColumnInfo(field_id, DataType::ARRAY),
            proto::plan::JSONContainsExpr_JSONOp_ContainsAny,
            true,
            empty_values);

        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);
        for (int i = 0; i < N * num_iters; ++i) {
            ASSERT_EQ(final[i], false);
        }
    }
}

TEST(Expr, TestArrayBinaryArith) {
    auto schema = std::make_shared<Schema>();
    schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto int_array_fid =
        schema->AddDebugField("int_array", DataType::ARRAY, DataType::INT8);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    auto float_array_fid =
        schema->AddDebugField("float_array", DataType::ARRAY, DataType::FLOAT);
    auto double_array_fid = schema->AddDebugField(
        "double_array", DataType::ARRAY, DataType::DOUBLE);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::map<std::string, std::vector<ScalarFieldProto>> array_cols;
    int num_iters = 1;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter);
        auto new_int_array_col =
            raw_data.get_col<ScalarFieldProto>(int_array_fid);
        auto new_long_array_col =
            raw_data.get_col<ScalarFieldProto>(long_array_fid);
        auto new_float_array_col =
            raw_data.get_col<ScalarFieldProto>(float_array_fid);
        auto new_double_array_col =
            raw_data.get_col<ScalarFieldProto>(double_array_fid);

        array_cols["int"].insert(array_cols["int"].end(),
                                 new_int_array_col.begin(),
                                 new_int_array_col.end());
        array_cols["long"].insert(array_cols["long"].end(),
                                  new_long_array_col.begin(),
                                  new_long_array_col.end());
        array_cols["float"].insert(array_cols["float"].end(),
                                   new_float_array_col.begin(),
                                   new_float_array_col.end());
        array_cols["double"].insert(array_cols["double"].end(),
                                    new_double_array_col.begin(),
                                    new_double_array_col.end());
        seg->PreInsert(N);
        seg->Insert(iter * N,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());
    ScopedSchemaHandle schema_handle(*schema);

    std::vector<std::tuple<std::string,
                           std::string,
                           std::function<bool(milvus::Array & array)>>>
        testcases = {
            // int_array[0] + 2 == 5
            {"int_array[0] + 2 == 5",
             "int",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val + 2 == 5;
             }},
            // int_array[0] + 2 != 5
            {"int_array[0] + 2 != 5",
             "int",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val + 2 != 5;
             }},
            // int_array[0] + 2 > 5
            {"int_array[0] + 2 > 5",
             "int",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val + 2 > 5;
             }},
            // int_array[0] + 2 >= 5
            {"int_array[0] + 2 >= 5",
             "int",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val + 2 >= 5;
             }},
            // int_array[0] + 2 < 5
            {"int_array[0] + 2 < 5",
             "int",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val + 2 < 5;
             }},
            // int_array[0] + 2 <= 5
            {"int_array[0] + 2 <= 5",
             "int",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val + 2 <= 5;
             }},
            // long_array[0] - 1 == 144
            {"long_array[0] - 1 == 144",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val - 1 == 144;
             }},
            // long_array[0] - 1 != 144
            {"long_array[0] - 1 != 144",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val - 1 != 144;
             }},
            // long_array[0] - 1 > 144
            {"long_array[0] - 1 > 144",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val - 1 > 144;
             }},
            // long_array[0] - 1 >= 144
            {"long_array[0] - 1 >= 144",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val - 1 >= 144;
             }},
            // long_array[0] - 1 < 144
            {"long_array[0] - 1 < 144",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val - 1 < 144;
             }},
            // long_array[0] - 1 <= 144
            {"long_array[0] - 1 <= 144",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val - 1 <= 144;
             }},
            // float_array[0] + 2.2 == 133.2
            {"float_array[0] + 2.2 == 133.2",
             "float",
             [](milvus::Array& array) {
                 auto val = array.get_data<double>(0);
                 return val + 2.2 == 133.2;
             }},
            // float_array[0] + 2.2 != 133.2
            {"float_array[0] + 2.2 != 133.2",
             "float",
             [](milvus::Array& array) {
                 auto val = array.get_data<double>(0);
                 return val + 2.2 != 133.2;
             }},
            // double_array[0] - 11.1 == 125.7
            {"double_array[0] - 11.1 == 125.7",
             "double",
             [](milvus::Array& array) {
                 auto val = array.get_data<double>(0);
                 return val - 11.1 == 125.7;
             }},
            // double_array[0] - 11.1 != 125.7
            {"double_array[0] - 11.1 != 125.7",
             "double",
             [](milvus::Array& array) {
                 auto val = array.get_data<double>(0);
                 return val - 11.1 != 125.7;
             }},
            // long_array[0] * 2 == 8
            {"long_array[0] * 2 == 8",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val * 2 == 8;
             }},
            // long_array[0] * 2 != 20
            {"long_array[0] * 2 != 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val * 2 != 20;
             }},
            // long_array[0] * 2 > 20
            {"long_array[0] * 2 > 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val * 2 > 20;
             }},
            // long_array[0] * 2 >= 20
            {"long_array[0] * 2 >= 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val * 2 >= 20;
             }},
            // long_array[0] * 2 < 20
            {"long_array[0] * 2 < 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val * 2 < 20;
             }},
            // long_array[0] * 2 <= 20
            {"long_array[0] * 2 <= 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val * 2 <= 20;
             }},
            // long_array[0] / 2 == 8
            {"long_array[0] / 2 == 8",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val / 2 == 8;
             }},
            // long_array[0] / 2 != 20
            {"long_array[0] / 2 != 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val / 2 != 20;
             }},
            // long_array[0] / 2 > 20
            {"long_array[0] / 2 > 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val / 2 > 20;
             }},
            // long_array[0] / 2 >= 20
            {"long_array[0] / 2 >= 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val / 2 >= 20;
             }},
            // long_array[0] / 2 < 20
            {"long_array[0] / 2 < 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val / 2 < 20;
             }},
            // long_array[0] / 2 <= 20
            {"long_array[0] / 2 <= 20",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val / 2 <= 20;
             }},
            // long_array[0] % 3 == 0
            {"long_array[0] % 3 == 0",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val % 3 == 0;
             }},
            // long_array[0] % 3 != 2
            {"long_array[0] % 3 != 2",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val % 3 != 2;
             }},
            // long_array[0] % 3 > 2
            {"long_array[0] % 3 > 2",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val % 3 > 2;
             }},
            // long_array[0] % 3 >= 2
            {"long_array[0] % 3 >= 2",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val % 3 >= 2;
             }},
            // long_array[0] % 3 < 2
            {"long_array[0] % 3 < 2",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val % 3 < 2;
             }},
            // long_array[0] % 3 <= 2
            {"long_array[0] % 3 <= 2",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val % 3 <= 2;
             }},
            // Bitwise AND/OR/XOR over an array integer element. long_array
            // values span a wide range, so OR/XOR clauses use a mid-range
            // threshold to keep the result mixed; AND clauses test low bits.
            {"(long_array[0] & 1) == 0",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return (val & 1) == 0;
             }},
            {"(long_array[0] & 1) != 0",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return (val & 1) != 0;
             }},
            {"(long_array[0] & 8) == 8",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return (val & 8) == 8;
             }},
            {"(long_array[0] | 4) < 5000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return (val | 4) < 5000;
             }},
            {"(long_array[0] ^ 15) < 5000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return (val ^ 15) < 5000;
             }},
            // Shift / bitwise-NOT over an array integer element. ~ is rewritten
            // to (x ^ -1), exercising the same BitXor array path.
            {"(long_array[0] >> 1) < 5000",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return (int64_t(val) >> 1) < 5000;
             }},
            {"(long_array[0] >> 2) >= 0",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return (int64_t(val) >> 2) >= 0;
             }},
            {"~long_array[0] != 0",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return (~int64_t(val)) != 0;
             }},
            // float_array[1024] + 2.2 == 133.2
            {"float_array[1024] + 2.2 == 133.2",
             "float",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val + 2.2 == 133.2;
             }},
            // float_array[1024] + 2.2 != 133.2
            {"float_array[1024] + 2.2 != 133.2",
             "float",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val + 2.2 != 133.2;
             }},
            // double_array[1024] - 11.1 == 125.7
            {"double_array[1024] - 11.1 == 125.7",
             "double",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val - 11.1 == 125.7;
             }},
            // double_array[1024] - 11.1 != 125.7
            {"double_array[1024] - 11.1 != 125.7",
             "double",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<double>(1024);
                 return val - 11.1 != 125.7;
             }},
            // long_array[1024] * 2 == 8
            {"long_array[1024] * 2 == 8",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return val * 2 == 8;
             }},
            // long_array[1024] * 2 != 20
            {"long_array[1024] * 2 != 20",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return val * 2 != 20;
             }},
            // long_array[1024] / 2 == 8
            {"long_array[1024] / 2 == 8",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return val / 2 == 8;
             }},
            // long_array[1024] / 2 != 20
            {"long_array[1024] / 2 != 20",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return val / 2 != 20;
             }},
            // long_array[1024] % 3 == 0
            {"long_array[1024] % 3 == 0",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return val % 3 == 0;
             }},
            // long_array[1024] % 3 != 2
            {"long_array[1024] % 3 != 2",
             "long",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<int64_t>(1024);
                 return val % 3 != 2;
             }},
            // array_length(int_array) == 10
            {"array_length(int_array) == 10",
             "int",
             [](milvus::Array& array) { return array.length() == 10; }},
            // array_length(int_array) != 8
            {"array_length(int_array) != 8",
             "int",
             [](milvus::Array& array) { return array.length() != 8; }},
            // array_length(int_array) > 8
            {"array_length(int_array) > 8",
             "int",
             [](milvus::Array& array) { return array.length() > 8; }},
            // array_length(int_array) >= 8
            {"array_length(int_array) >= 8",
             "int",
             [](milvus::Array& array) { return array.length() >= 8; }},
            // array_length(int_array) < 8
            {"array_length(int_array) < 8",
             "int",
             [](milvus::Array& array) { return array.length() < 8; }},
            // array_length(int_array) <= 8
            {"array_length(int_array) <= 8",
             "int",
             [](milvus::Array& array) { return array.length() <= 8; }},
        };

    for (auto [expr, array_type, ref_func] : testcases) {
        auto plan_str = schema_handle.ParseSearch(
            expr, "fakevec", 10, "L2", R"({"nprobe": 10})", 3);
        auto plan =
            CreateSearchPlanByExpr(schema, plan_str.data(), plan_str.size());
        BitsetType final;
        final = ExecuteQueryExpr(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0],
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0].get(),
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP,
            &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols[array_type][i]);
            auto ref = ref_func(array);
            ASSERT_EQ(ans, ref);
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], ref);
            }
        }
    }
}

template <typename T>
struct UnaryRangeTestcase {
    milvus::OpType op_type;
    T value;
    std::vector<std::string> nested_path;
    std::function<bool(milvus::Array&)> check_func;
};

TEST(Expr, TestArrayStringMatch) {
    auto schema = std::make_shared<Schema>();
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto string_array_fid = schema->AddDebugField(
        "string_array", DataType::ARRAY, DataType::VARCHAR);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::map<std::string, std::vector<ScalarFieldProto>> array_cols;
    int num_iters = 1;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter);
        auto new_string_array_col =
            raw_data.get_col<ScalarFieldProto>(string_array_fid);
        array_cols["string"].insert(array_cols["string"].end(),
                                    new_string_array_col.begin(),
                                    new_string_array_col.end());
        seg->PreInsert(N);
        seg->Insert(iter * N,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());

    std::vector<UnaryRangeTestcase<std::string>> prefix_testcases{
        {OpType::PrefixMatch,
         "abc",
         {"0"},
         [](milvus::Array& array) {
             return PrefixMatch(array.get_data<std::string_view>(0), "abc");
         }},
        {OpType::PrefixMatch,
         "def",
         {"1"},
         [](milvus::Array& array) {
             return PrefixMatch(array.get_data<std::string_view>(1), "def");
         }},
        {OpType::PrefixMatch,
         "def",
         {"1024"},
         [](milvus::Array& array) {
             if (array.length() <= 1024) {
                 return false;
             }
             return PrefixMatch(array.get_data<std::string_view>(1024), "def");
         }},
    };
    //vector_anns:<field_id:201 predicates:<unary_range_expr:<column_info:<field_id:131 data_type:Array nested_path:"0" element_type:VarChar > op:PrefixMatch value:<string_val:"abc" > > > query_info:<> placeholder_tag:"$0" >
    for (auto& testcase : prefix_testcases) {
        proto::plan::GenericValue value;
        value.set_string_val(testcase.value);
        auto expr = std::make_shared<milvus::expr::UnaryRangeFilterExpr>(
            milvus::expr::ColumnInfo(
                string_array_fid, DataType::ARRAY, testcase.nested_path),
            testcase.op_type,
            value,
            std::vector<proto::plan::GenericValue>{});
        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan.get(), seg_promote, N * num_iters, MAX_TIMESTAMP, &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols["string"][i]);
            ASSERT_EQ(ans, testcase.check_func(array));
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], testcase.check_func(array));
            }
        }
    }
}

TEST(Expr, TestArrayInTerm) {
    auto schema = std::make_shared<Schema>();
    schema->AddDebugField(
        "fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    auto bool_array_fid =
        schema->AddDebugField("bool_array", DataType::ARRAY, DataType::BOOL);
    auto float_array_fid =
        schema->AddDebugField("float_array", DataType::ARRAY, DataType::FLOAT);
    auto string_array_fid = schema->AddDebugField(
        "string_array", DataType::ARRAY, DataType::VARCHAR);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::map<std::string, std::vector<ScalarFieldProto>> array_cols;
    int num_iters = 1;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter);
        auto new_long_array_col =
            raw_data.get_col<ScalarFieldProto>(long_array_fid);
        auto new_bool_array_col =
            raw_data.get_col<ScalarFieldProto>(bool_array_fid);
        auto new_float_array_col =
            raw_data.get_col<ScalarFieldProto>(float_array_fid);
        auto new_string_array_col =
            raw_data.get_col<ScalarFieldProto>(string_array_fid);
        array_cols["long"].insert(array_cols["long"].end(),
                                  new_long_array_col.begin(),
                                  new_long_array_col.end());
        array_cols["bool"].insert(array_cols["bool"].end(),
                                  new_bool_array_col.begin(),
                                  new_bool_array_col.end());
        array_cols["float"].insert(array_cols["float"].end(),
                                   new_float_array_col.begin(),
                                   new_float_array_col.end());
        array_cols["string"].insert(array_cols["string"].end(),
                                    new_string_array_col.begin(),
                                    new_string_array_col.end());
        seg->PreInsert(N);
        seg->Insert(iter * N,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());
    ScopedSchemaHandle schema_handle(*schema);

    std::vector<std::tuple<std::string,
                           std::string,
                           std::function<bool(milvus::Array & array)>>>
        testcases = {
            // term_expr: long_array[0] in [1, 2, 3]
            {"long_array[0] in [1, 2, 3]",
             "long",
             [](milvus::Array& array) {
                 auto val = array.get_data<int64_t>(0);
                 return val == 1 || val == 2 || val == 3;
             }},
            // term_expr: long_array[0] in [] (empty list)
            {"long_array[0] in []",
             "long",
             [](milvus::Array& array) { return false; }},
            // term_expr: bool_array[0] in [false, false]
            {"bool_array[0] in [false, false]",
             "bool",
             [](milvus::Array& array) {
                 auto val = array.get_data<bool>(0);
                 return !val;
             }},
            // term_expr: bool_array[0] in [] (empty list)
            {"bool_array[0] in []",
             "bool",
             [](milvus::Array& array) { return false; }},
            // term_expr: float_array[0] in [1.23, 124.31]
            {"float_array[0] in [1.23, 124.31]",
             "float",
             [](milvus::Array& array) {
                 auto val = array.get_data<double>(0);
                 return val == 1.23 || val == 124.31;
             }},
            // term_expr: float_array[0] in [] (empty list)
            {"float_array[0] in []",
             "float",
             [](milvus::Array& array) { return false; }},
            // term_expr: string_array[0] in ["abc", "idhgf1s"]
            {R"(string_array[0] in ["abc", "idhgf1s"])",
             "string",
             [](milvus::Array& array) {
                 auto val = array.get_data<std::string_view>(0);
                 return val == "abc" || val == "idhgf1s";
             }},
            // term_expr: string_array[0] in [] (empty list)
            {R"(string_array[0] in [])",
             "string",
             [](milvus::Array& array) { return false; }},
            // term_expr: string_array[1024] in ["abc", "idhgf1s"]
            {R"(string_array[1024] in ["abc", "idhgf1s"])",
             "string",
             [](milvus::Array& array) {
                 if (array.length() <= 1024) {
                     return false;
                 }
                 auto val = array.get_data<std::string_view>(1024);
                 return val == "abc" || val == "idhgf1s";
             }},
        };

    for (auto [expr, array_type, ref_func] : testcases) {
        auto plan_str = schema_handle.ParseSearch(
            expr, "fakevec", 10, "L2", R"({"nprobe": 10})", 3);
        auto plan =
            CreateSearchPlanByExpr(schema, plan_str.data(), plan_str.size());
        BitsetType final;
        final = ExecuteQueryExpr(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0],
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan->plan_node_->plannodes_->sources()[0]->sources()[0].get(),
            seg_promote,
            N * num_iters,
            MAX_TIMESTAMP,
            &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols[array_type][i]);
            ASSERT_EQ(ans, ref_func(array));
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], ref_func(array));
            }
        }
    }
}

TEST(Expr, TestTermInArray) {
    auto schema = std::make_shared<Schema>();
    auto i64_fid = schema->AddDebugField("id", DataType::INT64);
    auto long_array_fid =
        schema->AddDebugField("long_array", DataType::ARRAY, DataType::INT64);
    schema->set_primary_field_id(i64_fid);

    auto seg = CreateGrowingSegment(schema, empty_index_meta);
    int N = 1000;
    std::map<std::string, std::vector<ScalarFieldProto>> array_cols;
    int num_iters = 1;
    for (int iter = 0; iter < num_iters; ++iter) {
        auto raw_data = DataGen(schema, N, iter);
        auto new_long_array_col =
            raw_data.get_col<ScalarFieldProto>(long_array_fid);
        array_cols["long"].insert(array_cols["long"].end(),
                                  new_long_array_col.begin(),
                                  new_long_array_col.end());
        seg->PreInsert(N);
        seg->Insert(iter * N,
                    N,
                    raw_data.row_ids_.data(),
                    raw_data.timestamps_.data(),
                    raw_data.raw_);
    }

    auto seg_promote = dynamic_cast<SegmentGrowingImpl*>(seg.get());

    struct TermTestCases {
        std::vector<int64_t> values;
        std::vector<std::string> nested_path;
        std::function<bool(milvus::Array&)> check_func;
    };
    std::vector<TermTestCases> testcases = {
        {{100},
         {},
         [](milvus::Array& array) {
             for (int i = 0; i < array.length(); ++i) {
                 auto val = array.get_data<int64_t>(i);
                 if (val == 100) {
                     return true;
                 }
             }
             return false;
         }},
        {{1024},
         {},
         [](milvus::Array& array) {
             for (int i = 0; i < array.length(); ++i) {
                 auto val = array.get_data<int64_t>(i);
                 if (val == 1024) {
                     return true;
                 }
             }
             return false;
         }},
    };

    for (auto& testcase : testcases) {
        std::vector<proto::plan::GenericValue> values;
        for (auto& v : testcase.values) {
            proto::plan::GenericValue val;
            val.set_int64_val(v);
            values.emplace_back(val);
        }
        auto expr = std::make_shared<milvus::expr::TermFilterExpr>(
            milvus::expr::ColumnInfo(
                long_array_fid, DataType::ARRAY, testcase.nested_path),
            values,
            true);
        BitsetType final;
        auto plan =
            std::make_shared<plan::FilterBitsNode>(DEFAULT_PLANNODE_ID, expr);
        final =
            ExecuteQueryExpr(plan, seg_promote, N * num_iters, MAX_TIMESTAMP);
        EXPECT_EQ(final.size(), N * num_iters);

        // specify some offsets and do scalar filtering on these offsets
        milvus::exec::OffsetVector offsets;
        offsets.reserve(N * num_iters / 2);
        for (auto i = 0; i < N * num_iters; ++i) {
            if (i % 2 == 0) {
                offsets.emplace_back(i);
            }
        }
        auto col_vec = milvus::test::gen_filter_res(
            plan.get(), seg_promote, N * num_iters, MAX_TIMESTAMP, &offsets);
        BitsetTypeView view(col_vec->GetRawData(), col_vec->size());
        EXPECT_EQ(view.size(), N * num_iters / 2);

        for (int i = 0; i < N * num_iters; ++i) {
            auto ans = final[i];
            auto array = milvus::Array(array_cols["long"][i]);
            ASSERT_EQ(ans, testcase.check_func(array));
            if (i % 2 == 0) {
                ASSERT_EQ(view[int(i / 2)], testcase.check_func(array));
            }
        }
    }
}

TEST(Expr, TestArrayContainsForStruct) {
    // Step 1: Prepare schema with array field
    int dim = 4;
    auto schema = std::make_shared<Schema>();
    auto vec_fid = schema->AddDebugVectorArrayField("structA[array_float_vec]",
                                                    DataType::VECTOR_FLOAT,
                                                    dim,
                                                    knowhere::metric::L2);
    auto int_array_fid = schema->AddDebugArrayField(
        "structA[price_array]", DataType::INT32, false);

    auto int64_fid = schema->AddDebugField("id", DataType::INT64);
    schema->set_primary_field_id(int64_fid);

    size_t N = 500;
    int array_len = 3;

    // Step 2: Generate test data
    auto raw_data = DataGen(schema, N, 42, 0, 1, array_len);

    // Use random values between 1-20 for array elements
    std::mt19937 rng(42);  // Fixed seed for reproducibility
    std::uniform_int_distribution<int> dist(1, 20);

    // Track which rows contain value 5 for verification
    std::set<int> rows_containing_5;

    for (int i = 0; i < raw_data.raw_->fields_data_size(); i++) {
        auto* field_data = raw_data.raw_->mutable_fields_data(i);
        if (field_data->field_id() == int_array_fid.get()) {
            field_data->mutable_scalars()
                ->mutable_array_data()
                ->mutable_data()
                ->Clear();

            for (int row = 0; row < N; row++) {
                auto* array_data = field_data->mutable_scalars()
                                       ->mutable_array_data()
                                       ->mutable_data()
                                       ->Add();

                for (int elem = 0; elem < array_len; elem++) {
                    int value = dist(rng);  // Random value between 1-20
                    array_data->mutable_int_data()->mutable_data()->Add(value);
                    if (value == 5) {
                        rows_containing_5.insert(row);
                    }
                }
            }
            break;
        }
    }

    // Step 3: Create sealed segment with field data
    auto segment = CreateSealedWithFieldDataLoaded(schema, raw_data);

    // Step 4: Load vector index for element-level search
    auto array_vec_values = raw_data.get_col<VectorFieldProto>(vec_fid);

    // DataGen generates VECTOR_ARRAY with data in float_vector (flattened),
    // not in vector_array (nested structure)
    std::vector<float> vector_data(dim * N * array_len);
    for (int i = 0; i < N; i++) {
        const auto& float_vec = array_vec_values[i].float_vector().data();
        // float_vec contains array_len * dim floats
        for (int j = 0; j < array_len * dim; j++) {
            vector_data[i * array_len * dim + j] = float_vec[j];
        }
    }

    // For element-level search, index all elements (N * array_len vectors)
    auto indexing = GenVecIndexing(N * array_len,
                                   dim,
                                   vector_data.data(),
                                   knowhere::IndexEnum::INDEX_HNSW);
    LoadIndexInfo load_index_info;
    load_index_info.field_id = vec_fid.get();
    load_index_info.index_params = GenIndexParams(indexing.get());
    load_index_info.cache_index =
        CreateTestCacheIndex("test", std::move(indexing));
    load_index_info.index_params["metric_type"] = knowhere::metric::L2;
    load_index_info.field_type = DataType::VECTOR_ARRAY;
    load_index_info.element_type = DataType::VECTOR_FLOAT;
    segment->LoadIndex(load_index_info);

    int topK = 5;

    ScopedSchemaHandle schema_handle(*schema);

    // Step 5a: Test with array_contains_any filter
    {
        std::string expr = "array_contains_any(structA[price_array], [5])";

        auto plan_bytes = schema_handle.ParseSearch(
            expr, "structA[array_float_vec]", topK, "L2", R"({"ef": 50})", 3);
        auto plan = CreateSearchPlanByExpr(
            schema, plan_bytes.data(), plan_bytes.size());
        ASSERT_NE(plan, nullptr);

        auto num_queries = 1;
        auto seed = 1024;
        auto ph_group_raw =
            CreatePlaceholderGroup(num_queries, dim, seed, true);
        auto ph_group =
            ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());

        auto search_result =
            segment->Search(plan.get(), ph_group.get(), 1L << 63);

        // Verify results
        ASSERT_NE(search_result, nullptr);

        // Array contains is a regular filter, not element-level search
        ASSERT_FALSE(search_result->seg_offsets_.empty());

        // Should have topK results per query
        ASSERT_LE(search_result->seg_offsets_.size(),
                  static_cast<size_t>(topK * num_queries));

        for (size_t i = 0; i < search_result->seg_offsets_.size(); i++) {
            int64_t doc_id = search_result->seg_offsets_[i];
            float distance = search_result->distances_[i];

            // Verify the doc's array contains value 5 using our tracked set
            ASSERT_TRUE(rows_containing_5.count(doc_id) > 0)
                << "Result doc_id " << doc_id << " should contain value 5";
        }

        // Verify distances are sorted (ascending for L2)
        for (size_t i = 1; i < search_result->distances_.size(); ++i) {
            ASSERT_LE(search_result->distances_[i - 1],
                      search_result->distances_[i])
                << "Distances should be sorted in ascending order";
        }
    }

    // Step 5b: Test with array_contains_all filter
    // contains_all requires the array to contain ALL given elements
    {
        std::string expr = "array_contains_all(structA[price_array], [5, 10])";

        auto plan_bytes = schema_handle.ParseSearch(
            expr, "structA[array_float_vec]", topK, "L2", R"({"ef": 50})", 3);
        auto plan = CreateSearchPlanByExpr(
            schema, plan_bytes.data(), plan_bytes.size());
        ASSERT_NE(plan, nullptr);

        auto num_queries = 1;
        auto seed = 1024;
        auto ph_group_raw =
            CreatePlaceholderGroup(num_queries, dim, seed, true);
        auto ph_group =
            ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());

        auto search_result =
            segment->Search(plan.get(), ph_group.get(), 1L << 63);

        ASSERT_NE(search_result, nullptr);

        // Verify each result's array contains BOTH 5 and 10
        auto array_col =
            raw_data.get_col(int_array_fid)->scalars().array_data().data();
        for (size_t i = 0; i < search_result->seg_offsets_.size(); i++) {
            int64_t doc_id = search_result->seg_offsets_[i];
            float distance = search_result->distances_[i];

            auto& arr = array_col[doc_id].int_data().data();
            bool has_5 = std::find(arr.begin(), arr.end(), 5) != arr.end();
            bool has_10 = std::find(arr.begin(), arr.end(), 10) != arr.end();
            ASSERT_TRUE(has_5 && has_10) << "Result doc_id " << doc_id
                                         << " should contain both 5 and 10";
        }

        for (size_t i = 1; i < search_result->distances_.size(); ++i) {
            ASSERT_LE(search_result->distances_[i - 1],
                      search_result->distances_[i])
                << "Distances should be sorted in ascending order";
        }
    }

    // Step 6: Test with scalar index on price_array field
    {
        // Get array data from raw_data and convert to boost::container::vector format
        // (required by InvertedIndexTantivy::BuildWithRawDataForUT)
        auto array_col =
            raw_data.get_col(int_array_fid)->scalars().array_data().data();
        std::vector<boost::container::vector<int32_t>> vec_of_array;
        vec_of_array.reserve(N);
        for (size_t i = 0; i < N; i++) {
            boost::container::vector<int32_t> arr;
            for (size_t j = 0; j < array_col[i].int_data().data_size(); j++) {
                arr.push_back(array_col[i].int_data().data(j));
            }
            vec_of_array.push_back(arr);
        }

        // Build inverted index using simplified API
        auto arr_index =
            std::make_unique<index::InvertedIndexTantivy<int32_t>>();
        Config cfg;
        cfg["is_array"] = true;
        cfg["is_nested_index"] = true;
        arr_index->BuildWithRawDataForUT(N, vec_of_array.data(), cfg);

        // Load index into segment
        LoadIndexInfo arr_index_info;
        arr_index_info.field_id = int_array_fid.get();
        arr_index_info.index_params = GenIndexParams(arr_index.get());
        arr_index_info.cache_index =
            CreateTestCacheIndex("test_array", std::move(arr_index));
        segment->LoadIndex(arr_index_info);

        // Now search with index
        std::string expr = "array_contains_any(structA[price_array], [5])";

        auto plan_bytes = schema_handle.ParseSearch(
            expr, "structA[array_float_vec]", topK, "L2", R"({"ef": 50})", 3);
        auto plan = CreateSearchPlanByExpr(
            schema, plan_bytes.data(), plan_bytes.size());
        ASSERT_NE(plan, nullptr);

        auto num_queries = 1;
        auto seed = 1024;
        auto ph_group_raw =
            CreatePlaceholderGroup(num_queries, dim, seed, true);
        auto ph_group =
            ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());

        auto search_result =
            segment->Search(plan.get(), ph_group.get(), 1L << 63);

        // Verify results with index
        ASSERT_NE(search_result, nullptr);
        ASSERT_FALSE(search_result->seg_offsets_.empty());
        ASSERT_LE(search_result->seg_offsets_.size(),
                  static_cast<size_t>(topK * num_queries));

        for (size_t i = 0; i < search_result->seg_offsets_.size(); i++) {
            int64_t doc_id = search_result->seg_offsets_[i];
            float distance = search_result->distances_[i];

            // Verify the doc's array contains value 5
            ASSERT_TRUE(rows_containing_5.count(doc_id) > 0)
                << "Result doc_id " << doc_id
                << " should contain value 5 (with index)";
        }

        // Verify distances are sorted (ascending for L2)
        for (size_t i = 1; i < search_result->distances_.size(); ++i) {
            ASSERT_LE(search_result->distances_[i - 1],
                      search_result->distances_[i])
                << "Distances should be sorted in ascending order (with index)";
        }
    }
}
