// 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 "BinaryRangeExpr.h"

#include <cstdint>
#include <limits>
#include <string>
#include <string_view>
#include <type_traits>
#include <utility>
#include <variant>

#include "common/EasyAssert.h"
#include "common/Tracer.h"
#include "common/bson_view.h"
#include "common/type_c.h"
#include "common/ScopedTimer.h"
#include "exec/expression/JsonNumberComparison.h"
#include "exec/expression/Utils.h"
#include "fmt/core.h"
#include "folly/FBVector.h"
#include "glog/logging.h"
#include "index/SkipIndex.h"
#include "monitor/Monitor.h"
#include "index/json_stats/JsonKeyStats.h"
#include "index/json_stats/utils.h"
#include "log/Log.h"
#include "opentelemetry/trace/span.h"
#include "query/Utils.h"
#include "segcore/SegmentInterface.h"
#include "segcore/SegmentSealed.h"
#include "storage/MmapManager.h"
#include "storage/Types.h"

namespace milvus {

namespace exec {

void
PhyBinaryRangeFilterExpr::Eval(EvalCtx& context, VectorPtr& result) {
    WaitPrefetch();
    tracer::AutoSpan span(
        "PhyBinaryRangeFilterExpr::Eval", tracer::GetRootSpan(), true);
    span.GetSpan()->SetAttribute("data_type",
                                 static_cast<int>(expr_->column_.data_type_));

    auto input = context.get_offset_input();
    SetHasOffsetInput((input != nullptr));

    auto data_type = expr_->column_.data_type_;
    if (expr_->column_.element_level_) {
        data_type = expr_->column_.element_type_;
    }
    switch (data_type) {
        case DataType::BOOL: {
            result = ExecRangeVisitorImpl<bool>(context);
            break;
        }
        case DataType::INT8: {
            result = ExecRangeVisitorImpl<int8_t>(context);
            break;
        }
        case DataType::INT16: {
            result = ExecRangeVisitorImpl<int16_t>(context);
            break;
        }
        case DataType::INT32: {
            result = ExecRangeVisitorImpl<int32_t>(context);
            break;
        }
        case DataType::INT64: {
            result = ExecRangeVisitorImpl<int64_t>(context);
            break;
        }
        case DataType::FLOAT: {
            result = ExecRangeVisitorImpl<float>(context);
            break;
        }
        case DataType::DOUBLE: {
            result = ExecRangeVisitorImpl<double>(context);
            break;
        }
        case DataType::VARCHAR: {
            if (segment_->type() == SegmentType::Growing &&
                !storage::MmapManager::GetInstance()
                     .GetMmapConfig()
                     .growing_enable_mmap) {
                result = ExecRangeVisitorImpl<std::string>(context);
            } else {
                result = ExecRangeVisitorImpl<std::string_view>(context);
            }
            break;
        }
        case DataType::JSON: {
            span.GetSpan()->SetAttribute("json_filter_expr_type",
                                         "binary_range");
            auto lower_type = expr_->lower_val_.val_case();
            auto upper_type = expr_->upper_val_.val_case();
            // For numeric types, if either bound is float, use double for both.
            // This handles mixed int64/float cases properly.
            bool use_double =
                (lower_type == proto::plan::GenericValue::ValCase::kFloatVal ||
                 upper_type == proto::plan::GenericValue::ValCase::kFloatVal);
            bool is_numeric =
                ((lower_type == proto::plan::GenericValue::ValCase::kInt64Val ||
                  lower_type ==
                      proto::plan::GenericValue::ValCase::kFloatVal) &&
                 (upper_type == proto::plan::GenericValue::ValCase::kInt64Val ||
                  upper_type == proto::plan::GenericValue::ValCase::kFloatVal));
            const auto has_unsafe_int_bound =
                is_numeric &&
                ((lower_type == proto::plan::GenericValue::ValCase::kInt64Val &&
                  !IsInt64SafeForJsonDoubleIndex(
                      expr_->lower_val_.int64_val())) ||
                 (upper_type == proto::plan::GenericValue::ValCase::kInt64Val &&
                  !IsInt64SafeForJsonDoubleIndex(
                      expr_->upper_val_.int64_val())));

            if (exec_path_ == ExprExecPath::ScalarIndex && !has_offset_input_) {
                if (is_numeric) {
                    if (has_unsafe_int_bound) {
                        result =
                            ExecRangeVisitorImplForJsonPreciseNumeric(context);
                    } else if (!use_double && PinnedJsonIndexIsFlat()) {
                        result = ExecRangeVisitorImplForIndex<int64_t>();
                    } else {
                        proto::plan::GenericValue double_lower_val;
                        if (lower_type ==
                            proto::plan::GenericValue::ValCase::kInt64Val) {
                            double_lower_val.set_float_val(static_cast<double>(
                                expr_->lower_val_.int64_val()));
                        } else {
                            double_lower_val.set_float_val(
                                expr_->lower_val_.float_val());
                        }
                        proto::plan::GenericValue double_upper_val;
                        if (upper_type ==
                            proto::plan::GenericValue::ValCase::kInt64Val) {
                            double_upper_val.set_float_val(static_cast<double>(
                                expr_->upper_val_.int64_val()));
                        } else {
                            double_upper_val.set_float_val(
                                expr_->upper_val_.float_val());
                        }

                        lower_arg_.SetValue<double>(double_lower_val);
                        upper_arg_.SetValue<double>(double_upper_val);
                        arg_inited_ = true;

                        result = ExecRangeVisitorImplForIndex<double>();
                    }
                } else if (lower_type ==
                           proto::plan::GenericValue::ValCase::kStringVal) {
                    result = ExecRangeVisitorImplForJson<std::string>(context);
                } else {
                    ThrowInfo(
                        UnexpectedError,
                        fmt::format("unsupported value type {} in expression",
                                    lower_type));
                }
            } else {
                if (has_unsafe_int_bound &&
                    (has_offset_input_ ||
                     exec_path_ != ExprExecPath::JsonStats)) {
                    result = ExecRangeVisitorImplForJsonPreciseNumeric(context);
                } else if (is_numeric && use_double) {
                    // Use double when either bound is float
                    result = ExecRangeVisitorImplForJson<double>(context);
                } else if (lower_type ==
                           proto::plan::GenericValue::ValCase::kInt64Val) {
                    result = ExecRangeVisitorImplForJson<int64_t>(context);
                } else if (lower_type ==
                           proto::plan::GenericValue::ValCase::kFloatVal) {
                    result = ExecRangeVisitorImplForJson<double>(context);
                } else if (lower_type ==
                           proto::plan::GenericValue::ValCase::kStringVal) {
                    result = ExecRangeVisitorImplForJson<std::string>(context);
                } else {
                    ThrowInfo(
                        UnexpectedError,
                        fmt::format("unsupported value type {} in expression",
                                    lower_type));
                }
            }
            break;
        }
        case DataType::ARRAY: {
            auto value_type = expr_->lower_val_.val_case();
            switch (value_type) {
                case proto::plan::GenericValue::ValCase::kInt64Val: {
                    result = ExecRangeVisitorImplForArray<int64_t>(context);
                    break;
                }
                case proto::plan::GenericValue::ValCase::kFloatVal: {
                    result = ExecRangeVisitorImplForArray<double>(context);
                    break;
                }
                case proto::plan::GenericValue::ValCase::kStringVal: {
                    result = ExecRangeVisitorImplForArray<std::string>(context);
                    break;
                }
                default: {
                    ThrowInfo(
                        UnexpectedError,
                        fmt::format("unsupported value type {} in expression",
                                    value_type));
                }
            }
            break;
        }
        default:
            ThrowInfo(UnexpectedError,
                      "unsupported data type: {}",
                      expr_->column_.data_type_);
    }
}

VectorPtr
PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForJsonPreciseNumeric(
    EvalCtx& context) {
    const auto& bitmap_input = context.get_bitmap_input();
    auto* input = context.get_offset_input();
    auto real_batch_size =
        has_offset_input_ ? input->size() : GetNextBatchSize();
    if (real_batch_size == 0) {
        return nullptr;
    }

    auto res_vec =
        std::make_shared<ColumnVector>(TargetBitmap(real_batch_size, false),
                                       TargetBitmap(real_batch_size, true));
    TargetBitmapView res(res_vec->GetRawData(), real_batch_size);
    TargetBitmapView valid_res(res_vec->GetValidRawData(), real_batch_size);
    auto pointer = milvus::Json::pointer(expr_->column_.nested_path_);
    auto lower_bound = expr_->lower_val_;
    auto upper_bound = expr_->upper_val_;
    const auto lower_inclusive = expr_->lower_inclusive_;
    const auto upper_inclusive = expr_->upper_inclusive_;

    size_t processed_cursor = 0;
    auto execute_sub_batch =
        [
            pointer,
            lower_bound,
            upper_bound,
            lower_inclusive,
            upper_inclusive,
            &bitmap_input,
            &processed_cursor
        ]<FilterType filter_type = FilterType::sequential>(
            const milvus::Json* data,
            const bool* valid_data,
            const int32_t* offsets,
            const int size,
            TargetBitmapView res,
            TargetBitmapView valid_res) {
        if (data == nullptr) {
            processed_cursor += size;
            return;
        }
        const bool has_bitmap_input = !bitmap_input.empty();
        for (int i = 0; i < size; ++i) {
            auto offset = i;
            if constexpr (filter_type == FilterType::random) {
                offset = offsets ? offsets[i] : i;
            }
            if (valid_data != nullptr && !valid_data[offset]) {
                res[i] = valid_res[i] = false;
                continue;
            }
            if (has_bitmap_input && !bitmap_input[processed_cursor + i]) {
                continue;
            }

            auto number = data[offset].at_numeric(pointer);
            if (number.error()) {
                res[i] = valid_res[i] = false;
                continue;
            }
            auto lower_comparison =
                CompareJsonNumberToBound(number.value(), lower_bound);
            auto upper_comparison =
                CompareJsonNumberToBound(number.value(), upper_bound);
            if (!lower_comparison.has_value() ||
                !upper_comparison.has_value()) {
                res[i] = false;
                continue;
            }
            const auto lower_matches = lower_inclusive ? *lower_comparison >= 0
                                                       : *lower_comparison > 0;
            const auto upper_matches = upper_inclusive ? *upper_comparison <= 0
                                                       : *upper_comparison < 0;
            res[i] = lower_matches && upper_matches;
        }
        processed_cursor += size;
    };

    int64_t processed_size;
    if (has_offset_input_) {
        processed_size = ProcessDataByOffsets<milvus::Json>(
            execute_sub_batch, std::nullptr_t{}, input, res, valid_res);
    } else {
        processed_size = ProcessDataChunks<milvus::Json>(
            execute_sub_batch, std::nullptr_t{}, res, valid_res);
    }
    AssertInfo(processed_size == real_batch_size,
               "internal error: expr processed rows {} not equal "
               "expect batch size {}",
               processed_size,
               real_batch_size);
    return res_vec;
}

template <typename T>
VectorPtr
PhyBinaryRangeFilterExpr::ExecRangeVisitorImpl(EvalCtx& context) {
    if (!has_offset_input_ && exec_path_ == ExprExecPath::PkIndex) {
        if (pk_type_ == DataType::VARCHAR) {
            return ExecRangeVisitorImplForPk<std::string_view>(context);
        } else {
            return ExecRangeVisitorImplForPk<int64_t>(context);
        }
    }

    if (exec_path_ == ExprExecPath::ScalarIndex && !has_offset_input_) {
        return ExecRangeVisitorImplForIndex<T>();
    } else {
        return ExecRangeVisitorImplForData<T>(context);
    }
}

template <typename T, typename IndexInnerType, typename HighPrecisionType>
ColumnVectorPtr
PhyBinaryRangeFilterExpr::PreCheckOverflow(HighPrecisionType& val1,
                                           HighPrecisionType& val2,
                                           bool& lower_inclusive,
                                           bool& upper_inclusive,
                                           OffsetVector* input) {
    lower_inclusive = expr_->lower_inclusive_;
    upper_inclusive = expr_->upper_inclusive_;

    if (!arg_inited_) {
        lower_arg_.SetValue<HighPrecisionType>(expr_->lower_val_);
        upper_arg_.SetValue<HighPrecisionType>(expr_->upper_val_);
        arg_inited_ = true;
    }
    val1 = lower_arg_.GetValue<HighPrecisionType>();
    val2 = upper_arg_.GetValue<HighPrecisionType>();
    auto get_next_overflow_batch =
        [this](OffsetVector* input) -> ColumnVectorPtr {
        TargetBitmap valid_res;
        if (expr_->column_.element_level_) {
            // Element batches are derived from the row cursor so
            // MoveCursor()-based short-circuiting stays aligned.
            // Individual elements cannot be null; their containing row's
            // validity is applied by the element consumer.
            auto batch_size =
                GetNextRealBatchSize(input, /*element_level=*/true);
            valid_res = TargetBitmap(batch_size, true);
            if (input == nullptr) {
                MoveCursor();
            }
        } else if (input != nullptr) {
            valid_res =
                ProcessChunksForValidByOffsets<T>(UseIndexCursor(), *input);
        } else {
            valid_res = ProcessChunksForValid<T>(UseIndexCursor());
        }

        auto batch_size = valid_res.size();
        auto res_vec = std::make_shared<ColumnVector>(TargetBitmap(batch_size),
                                                      std::move(valid_res));
        return res_vec;
    };

    if constexpr (std::is_integral_v<T> && !std::is_same_v<bool, T>) {
        if (milvus::query::gt_ub<T>(val1)) {
            return get_next_overflow_batch(input);
        } else if (milvus::query::lt_lb<T>(val1)) {
            val1 = std::numeric_limits<T>::min();
            lower_inclusive = true;
        }

        if (milvus::query::gt_ub<T>(val2)) {
            val2 = std::numeric_limits<T>::max();
            upper_inclusive = true;
        } else if (milvus::query::lt_lb<T>(val2)) {
            return get_next_overflow_batch(input);
        }
    }
    return nullptr;
}

template <typename T>
VectorPtr
PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForIndex() {
    typedef std::
        conditional_t<std::is_same_v<T, std::string_view>, std::string, T>
            IndexInnerType;
    using Index = index::ScalarIndex<IndexInnerType>;
    typedef std::conditional_t<std::is_integral_v<IndexInnerType> &&
                                   !std::is_same_v<bool, T>,
                               int64_t,
                               IndexInnerType>
        HighPrecisionType;

    HighPrecisionType val1;
    HighPrecisionType val2;
    bool lower_inclusive = false;
    bool upper_inclusive = false;
    if (auto res =
            PreCheckOverflow<T>(val1, val2, lower_inclusive, upper_inclusive)) {
        return res;
    }

    auto real_batch_size =
        GetNextRealBatchSize(nullptr, expr_->column_.element_level_);
    if (real_batch_size == 0) {
        return nullptr;
    }

    auto execute_sub_batch = [lower_inclusive, upper_inclusive](
                                 Index* index_ptr,
                                 HighPrecisionType val1,
                                 HighPrecisionType val2) {
        BinaryRangeIndexFunc<T> func;
        return func(index_ptr, val1, val2, lower_inclusive, upper_inclusive);
    };
    auto res = ProcessIndexChunks<T>(execute_sub_batch, val1, val2);
    AssertInfo(res->size() == real_batch_size,
               "internal error: expr processed rows {} not equal "
               "expect batch size {}",
               res->size(),
               real_batch_size);
    return res;
}

template <typename T>
VectorPtr
PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForData(EvalCtx& context) {
    typedef std::
        conditional_t<std::is_same_v<T, std::string_view>, std::string, T>
            IndexInnerType;
    typedef std::conditional_t<std::is_integral_v<IndexInnerType> &&
                                   !std::is_same_v<bool, T>,
                               int64_t,
                               IndexInnerType>
        HighPrecisionType;

    const auto& bitmap_input = context.get_bitmap_input();
    auto* input = context.get_offset_input();
    HighPrecisionType val1;
    HighPrecisionType val2;
    bool lower_inclusive = false;
    bool upper_inclusive = false;
    if (auto res = PreCheckOverflow<T>(
            val1, val2, lower_inclusive, upper_inclusive, input)) {
        return res;
    }

    auto real_batch_size =
        GetNextRealBatchSize(input, expr_->column_.element_level_);
    if (real_batch_size == 0) {
        return nullptr;
    }
    auto res_vec =
        std::make_shared<ColumnVector>(TargetBitmap(real_batch_size, false),
                                       TargetBitmap(real_batch_size, true));
    TargetBitmapView res(res_vec->GetRawData(), real_batch_size);
    TargetBitmapView valid_res(res_vec->GetValidRawData(), real_batch_size);

    size_t processed_cursor = 0;
    auto execute_sub_batch =
        [ lower_inclusive, upper_inclusive, &processed_cursor, &
          bitmap_input ]<FilterType filter_type = FilterType::sequential>(
            const T* data,
            const bool* valid_data,
            const int32_t* offsets,
            const int size,
            TargetBitmapView res,
            TargetBitmapView valid_res,
            HighPrecisionType val1,
            HighPrecisionType val2) {
        // If data is nullptr, this chunk was skipped by SkipIndex.
        // We only need to update processed_cursor for bitmap_input indexing.
        if (data == nullptr) {
            processed_cursor += size;
            return;
        }
        if (lower_inclusive && upper_inclusive) {
            BinaryRangeElementFunc<T, true, true, filter_type> func;
            func(val1,
                 val2,
                 data,
                 size,
                 res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        } else if (lower_inclusive && !upper_inclusive) {
            BinaryRangeElementFunc<T, true, false, filter_type> func;
            func(val1,
                 val2,
                 data,
                 size,
                 res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        } else if (!lower_inclusive && upper_inclusive) {
            BinaryRangeElementFunc<T, false, true, filter_type> func;
            func(val1,
                 val2,
                 data,
                 size,
                 res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        } else {
            BinaryRangeElementFunc<T, false, false, filter_type> func;
            func(val1,
                 val2,
                 data,
                 size,
                 res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        }
        // there is a batch operation in BinaryRangeElementFunc,
        // so not divide data again for the reason that it may reduce performance if the null distribution is scattered
        // but to mask res with valid_data after the batch operation.
        if constexpr (filter_type == FilterType::sequential) {
            // contiguous rows: reuse the vectorized shared helper
            ApplyValidMask(valid_data, res, valid_res, size);
        } else if (valid_data != nullptr) {
            // scattered by offsets: gather, keep the per-row loop
            for (int i = 0; i < size; i++) {
                auto offset = (offsets) ? offsets[i] : i;
                if (!valid_data[offset]) {
                    res[i] = valid_res[i] = false;
                }
            }
        }
        processed_cursor += size;
    };

    auto skip_index_func =
        [op_ctx = op_ctx_, val1, val2, lower_inclusive, upper_inclusive](
            const SkipIndex& skip_index, FieldId field_id, int64_t chunk_id) {
            if (lower_inclusive && upper_inclusive) {
                return skip_index.CanSkipBinaryRange<T>(
                    op_ctx, field_id, chunk_id, val1, val2, true, true);
            } else if (lower_inclusive && !upper_inclusive) {
                return skip_index.CanSkipBinaryRange<T>(
                    op_ctx, field_id, chunk_id, val1, val2, true, false);
            } else if (!lower_inclusive && upper_inclusive) {
                return skip_index.CanSkipBinaryRange<T>(
                    op_ctx, field_id, chunk_id, val1, val2, false, true);
            } else {
                return skip_index.CanSkipBinaryRange<T>(
                    op_ctx, field_id, chunk_id, val1, val2, false, false);
            }
        };
    int64_t processed_size;
    if (has_offset_input_) {
        if (expr_->column_.element_level_) {
            // For element-level filtering with offset input
            processed_size = ProcessElementLevelByOffsets<T>(execute_sub_batch,
                                                             skip_index_func,
                                                             input,
                                                             res,
                                                             valid_res,
                                                             val1,
                                                             val2);
        } else {
            // For doc-level filtering
            processed_size = ProcessDataByOffsets<T>(execute_sub_batch,
                                                     skip_index_func,
                                                     input,
                                                     res,
                                                     valid_res,
                                                     val1,
                                                     val2);
        }
    } else {
        if (expr_->column_.element_level_) {
            // For element-level filtering without offset input (brute force)
            processed_size = ProcessDataChunksForElementLevel<T>(
                execute_sub_batch, skip_index_func, res, valid_res, val1, val2);
        } else {
            processed_size = ProcessDataChunks<T>(
                execute_sub_batch, skip_index_func, res, valid_res, val1, val2);
        }
    }
    AssertInfo(processed_size == real_batch_size,
               "internal error: expr processed rows {} not equal "
               "expect batch size {}",
               processed_size,
               real_batch_size);
    return res_vec;
}

template <typename ValueType>
VectorPtr
PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForJson(EvalCtx& context) {
    const auto& bitmap_input = context.get_bitmap_input();
    auto* input = context.get_offset_input();
    FieldId field_id = expr_->column_.field_id_;
    if (!has_offset_input_ && exec_path_ == ExprExecPath::JsonStats) {
        milvus::ScopedTimer timer(
            "binary_range_json_by_stats",
            [this](double us) { json_filter_stats_latency_us_ += us; });
        return ExecRangeVisitorImplForJsonStats<ValueType>();
    }

    milvus::ScopedTimer timer(
        "binary_range_json_bruteforce",
        [this](double us) { json_filter_bruteforce_latency_us_ += us; });

    auto real_batch_size =
        has_offset_input_ ? input->size() : GetNextBatchSize();
    if (real_batch_size == 0) {
        return nullptr;
    }
    auto res_vec =
        std::make_shared<ColumnVector>(TargetBitmap(real_batch_size, false),
                                       TargetBitmap(real_batch_size, true));
    TargetBitmapView res(res_vec->GetRawData(), real_batch_size);
    TargetBitmapView valid_res(res_vec->GetValidRawData(), real_batch_size);

    bool lower_inclusive = expr_->lower_inclusive_;
    bool upper_inclusive = expr_->upper_inclusive_;
    if (!arg_inited_) {
        lower_arg_.SetValue<ValueType>(expr_->lower_val_);
        upper_arg_.SetValue<ValueType>(expr_->upper_val_);
        arg_inited_ = true;
    }
    ValueType val1 = lower_arg_.GetValue<ValueType>();
    ValueType val2 = upper_arg_.GetValue<ValueType>();
    auto pointer = milvus::Json::pointer(expr_->column_.nested_path_);

    size_t processed_cursor = 0;
    auto execute_sub_batch =
        [
            lower_inclusive,
            upper_inclusive,
            pointer,
            &bitmap_input,
            &processed_cursor
        ]<FilterType filter_type = FilterType::sequential>(
            const milvus::Json* data,
            const bool* valid_data,
            const int32_t* offsets,
            const int size,
            TargetBitmapView res,
            TargetBitmapView valid_res,
            ValueType val1,
            ValueType val2) {
        // If data is nullptr, this chunk was skipped by SkipIndex.
        // We only need to update processed_cursor for bitmap_input indexing.
        if (data == nullptr) {
            processed_cursor += size;
            return;
        }
        if (lower_inclusive && upper_inclusive) {
            BinaryRangeElementFuncForJson<ValueType, true, true, filter_type>
                func;
            func(val1,
                 val2,
                 pointer,
                 data,
                 valid_data,
                 size,
                 res,
                 valid_res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        } else if (lower_inclusive && !upper_inclusive) {
            BinaryRangeElementFuncForJson<ValueType, true, false, filter_type>
                func;
            func(val1,
                 val2,
                 pointer,
                 data,
                 valid_data,
                 size,
                 res,
                 valid_res,
                 bitmap_input,
                 processed_cursor,
                 offsets);

        } else if (!lower_inclusive && upper_inclusive) {
            BinaryRangeElementFuncForJson<ValueType, false, true, filter_type>
                func;
            func(val1,
                 val2,
                 pointer,
                 data,
                 valid_data,
                 size,
                 res,
                 valid_res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        } else {
            BinaryRangeElementFuncForJson<ValueType, false, false, filter_type>
                func;
            func(val1,
                 val2,
                 pointer,
                 data,
                 valid_data,
                 size,
                 res,
                 valid_res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        }
        processed_cursor += size;
    };
    int64_t processed_size;
    if (has_offset_input_) {
        processed_size = ProcessDataByOffsets<milvus::Json>(execute_sub_batch,
                                                            std::nullptr_t{},
                                                            input,
                                                            res,
                                                            valid_res,
                                                            val1,
                                                            val2);
    } else {
        processed_size = ProcessDataChunks<milvus::Json>(
            execute_sub_batch, std::nullptr_t{}, res, valid_res, val1, val2);
    }
    AssertInfo(processed_size == real_batch_size,
               "internal error: expr processed rows {} not equal "
               "expect batch size {}",
               processed_size,
               real_batch_size);
    return res_vec;
}

template <typename ValueType>
VectorPtr
PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForJsonStats() {
    using GetType = std::conditional_t<std::is_same_v<ValueType, std::string>,
                                       std::string_view,
                                       ValueType>;
    auto real_batch_size = GetNextBatchSize();
    if (real_batch_size == 0) {
        return nullptr;
    }
    auto pointer = milvus::index::JsonPointer(expr_->column_.nested_path_);
    bool lower_inclusive = expr_->lower_inclusive_;
    bool upper_inclusive = expr_->upper_inclusive_;
    std::optional<ValueType> val1;
    std::optional<ValueType> val2;
    if constexpr (!std::is_same_v<GetType, int64_t> &&
                  !std::is_same_v<GetType, double>) {
        val1.emplace(GetValueWithCastNumber<ValueType>(expr_->lower_val_));
        val2.emplace(GetValueWithCastNumber<ValueType>(expr_->upper_val_));
    }

    if (cached_index_chunk_id_ != 0 && TryCacheGet()) {
        // Cache hit — skip Stats computation.
    } else if (cached_index_chunk_id_ != 0 &&
               segment_->type() == SegmentType::Sealed) {
        auto cache_compute_start = CacheClock::now();
        auto* segment = dynamic_cast<const segcore::SegmentSealed*>(segment_);
        auto field_id = expr_->column_.field_id_;
        auto index = segment->GetJsonStats(op_ctx_, field_id);
        Assert(index.get() != nullptr);

        cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
        cached_index_chunk_valid_res_ =
            std::make_shared<TargetBitmap>(active_count_);
        TargetBitmapView res_view(*cached_index_chunk_res_);
        TargetBitmapView valid_res_view(*cached_index_chunk_valid_res_);

        // process shredding data
        const auto& lower_bound = expr_->lower_val_;
        const auto& upper_bound = expr_->upper_val_;
        auto try_execute = [&](milvus::index::JSONType json_type,
                               auto GetType) {
            auto target_field = index->GetShreddingField(pointer, json_type);
            if (!target_field.empty()) {
                using ColType = decltype(GetType);
                TargetBitmap target_res(active_count_, false);
                TargetBitmapView target_res_view(target_res);
                TargetBitmap target_valid(active_count_, true);
                TargetBitmapView target_valid_view(target_valid);
                auto shredding_executor = [val1,
                                           val2,
                                           lower_inclusive,
                                           upper_inclusive,
                                           &lower_bound,
                                           &upper_bound](
                                              const ColType* src,
                                              const bool* valid,
                                              size_t size,
                                              TargetBitmapView res,
                                              TargetBitmapView valid_res) {
                    for (size_t i = 0; i < size; ++i) {
                        if (valid != nullptr && !valid[i]) {
                            res[i] = valid_res[i] = false;
                            continue;
                        }
                        if constexpr (std::is_same_v<ColType, int64_t> ||
                                      std::is_same_v<ColType, double>) {
                            auto lower_comparison =
                                CompareJsonNumberToBound(src[i], lower_bound);
                            auto upper_comparison =
                                CompareJsonNumberToBound(src[i], upper_bound);
                            if (!lower_comparison.has_value() ||
                                !upper_comparison.has_value()) {
                                res[i] = false;
                                continue;
                            }
                            const auto lower_matches =
                                lower_inclusive ? *lower_comparison >= 0
                                                : *lower_comparison > 0;
                            const auto upper_matches =
                                upper_inclusive ? *upper_comparison <= 0
                                                : *upper_comparison < 0;
                            res[i] = lower_matches && upper_matches;
                        } else if (lower_inclusive && upper_inclusive) {
                            res[i] = src[i] >= *val1 && src[i] <= *val2;
                        } else if (lower_inclusive && !upper_inclusive) {
                            res[i] = src[i] >= *val1 && src[i] < *val2;
                        } else if (!lower_inclusive && upper_inclusive) {
                            res[i] = src[i] > *val1 && src[i] <= *val2;
                        } else {
                            res[i] = src[i] > *val1 && src[i] < *val2;
                        }
                    }
                };
                index->ExecutorForShreddingData<ColType>(op_ctx_,
                                                         target_field,
                                                         shredding_executor,
                                                         nullptr,
                                                         target_res_view,
                                                         target_valid_view);
                res_view.inplace_or_with_count(target_res_view, active_count_);
                valid_res_view.inplace_or_with_count(target_valid_view,
                                                     active_count_);
                LOG_DEBUG("using shredding data's field: {} count {}",
                          target_field,
                          res_view.count());
            }
        };

        {
            milvus::ScopedTimer timer(
                "binary_range_json_stats_shredding_data",
                [this](double us) { json_stats_shredding_latency_us_ += us; });

            if constexpr (std::is_same_v<GetType, int64_t>) {
                // int64 compare
                try_execute(milvus::index::JSONType::INT64, int64_t{});
                // and double compare
                try_execute(milvus::index::JSONType::DOUBLE, double{});

            } else if constexpr (std::is_same_v<GetType, double>) {
                try_execute(milvus::index::JSONType::DOUBLE, double{});
                // and int64 compare
                try_execute(milvus::index::JSONType::INT64, int64_t{});
            } else if constexpr (std::is_same_v<GetType, std::string_view> ||
                                 std::is_same_v<GetType, std::string>) {
                try_execute(milvus::index::JSONType::STRING,
                            std::string_view{});
            }
        }

        // process shared data
        auto shared_executor = [val1,
                                val2,
                                lower_inclusive,
                                upper_inclusive,
                                &lower_bound,
                                &upper_bound,
                                &res_view,
                                &valid_res_view](milvus::BsonView bson,
                                                 uint32_t row_id,
                                                 uint32_t value_offset) {
            auto set_known = [&](bool value) {
                res_view[row_id] = value;
                valid_res_view[row_id] = true;
            };
            if constexpr (std::is_same_v<GetType, int64_t> ||
                          std::is_same_v<GetType, double>) {
                auto lower_comparison =
                    CompareBsonNumberToBound(bson, value_offset, lower_bound);
                auto upper_comparison =
                    CompareBsonNumberToBound(bson, value_offset, upper_bound);
                if (!lower_comparison.has_value() ||
                    !upper_comparison.has_value()) {
                    return;
                }
                const auto lower_matches = lower_inclusive
                                               ? *lower_comparison >= 0
                                               : *lower_comparison > 0;
                const auto upper_matches = upper_inclusive
                                               ? *upper_comparison <= 0
                                               : *upper_comparison < 0;
                set_known(lower_matches && upper_matches);
            } else {
                auto val = bson.ParseAsValueAtOffset<GetType>(value_offset);
                if (!val.has_value()) {
                    return;
                }
                if (lower_inclusive && upper_inclusive) {
                    set_known(val.value() >= *val1 && val.value() <= *val2);
                } else if (lower_inclusive && !upper_inclusive) {
                    set_known(val.value() >= *val1 && val.value() < *val2);
                } else if (!lower_inclusive && upper_inclusive) {
                    set_known(val.value() > *val1 && val.value() <= *val2);
                } else {
                    set_known(val.value() > *val1 && val.value() < *val2);
                }
            }
        };
        bool skip_shared_data = false;
        if constexpr (std::is_same_v<GetType, int64_t> ||
                      std::is_same_v<GetType, double>) {
            skip_shared_data =
                index->HasAllShreddingFields(pointer,
                                             {milvus::index::JSONType::INT64,
                                              milvus::index::JSONType::DOUBLE});
        } else if constexpr (std::is_same_v<GetType, std::string_view> ||
                             std::is_same_v<GetType, std::string>) {
            skip_shared_data = index->HasAllShreddingFields(
                pointer, {milvus::index::JSONType::STRING});
        }

        if (!skip_shared_data) {
            milvus::ScopedTimer timer(
                "binary_range_json_stats_shared_data",
                [this](double us) { json_stats_shared_latency_us_ += us; });

            index->ExecuteForSharedData(
                op_ctx_, bson_index_, pointer, shared_executor);
        }
        cached_index_chunk_id_ = 0;
        CachePut(CacheElapsedUs(cache_compute_start));
    }

    auto res = MoveOrSliceBitmap(*cached_index_chunk_res_,
                                 *cached_index_chunk_valid_res_,
                                 current_data_global_pos_,
                                 real_batch_size);
    MoveCursor();
    return res;
}  // namespace exec

template <typename ValueType>
VectorPtr
PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForArray(EvalCtx& context) {
    const auto& bitmap_input = context.get_bitmap_input();
    auto* input = context.get_offset_input();
    auto real_batch_size =
        has_offset_input_ ? input->size() : GetNextBatchSize();
    if (real_batch_size == 0) {
        return nullptr;
    }
    auto res_vec =
        std::make_shared<ColumnVector>(TargetBitmap(real_batch_size, false),
                                       TargetBitmap(real_batch_size, true));
    TargetBitmapView res(res_vec->GetRawData(), real_batch_size);
    TargetBitmapView valid_res(res_vec->GetValidRawData(), real_batch_size);

    bool lower_inclusive = expr_->lower_inclusive_;
    bool upper_inclusive = expr_->upper_inclusive_;

    if (!arg_inited_) {
        lower_arg_.SetValue<ValueType>(expr_->lower_val_);
        upper_arg_.SetValue<ValueType>(expr_->upper_val_);
        arg_inited_ = true;
    }
    ValueType val1 = lower_arg_.GetValue<ValueType>();
    ValueType val2 = upper_arg_.GetValue<ValueType>();

    int index = -1;
    if (expr_->column_.nested_path_.size() > 0) {
        index = std::stoi(expr_->column_.nested_path_[0]);
    }

    size_t processed_cursor = 0;
    auto execute_sub_batch =
        [ lower_inclusive, upper_inclusive, &processed_cursor, &
          bitmap_input ]<FilterType filter_type = FilterType::sequential>(
            const milvus::ArrayView* data,
            const bool* valid_data,
            const int32_t* offsets,
            const int size,
            TargetBitmapView res,
            TargetBitmapView valid_res,
            ValueType val1,
            ValueType val2,
            int index) {
        AssertInfo(index >= 0,
                   "array element range predicate requires nested path");
        // If data is nullptr, this chunk was skipped by SkipIndex.
        // We only need to update processed_cursor for bitmap_input indexing.
        if (data == nullptr) {
            processed_cursor += size;
            return;
        }
        if (lower_inclusive && upper_inclusive) {
            BinaryRangeElementFuncForArray<ValueType, true, true, filter_type>
                func;
            func(val1,
                 val2,
                 index,
                 data,
                 valid_data,
                 size,
                 res,
                 valid_res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        } else if (lower_inclusive && !upper_inclusive) {
            BinaryRangeElementFuncForArray<ValueType, true, false, filter_type>
                func;
            func(val1,
                 val2,
                 index,
                 data,
                 valid_data,
                 size,
                 res,
                 valid_res,
                 bitmap_input,
                 processed_cursor,
                 offsets);

        } else if (!lower_inclusive && upper_inclusive) {
            BinaryRangeElementFuncForArray<ValueType, false, true, filter_type>
                func;
            func(val1,
                 val2,
                 index,
                 data,
                 valid_data,
                 size,
                 res,
                 valid_res,
                 bitmap_input,
                 processed_cursor,
                 offsets);

        } else {
            BinaryRangeElementFuncForArray<ValueType, false, false, filter_type>
                func;
            func(val1,
                 val2,
                 index,
                 data,
                 valid_data,
                 size,
                 res,
                 valid_res,
                 bitmap_input,
                 processed_cursor,
                 offsets);
        }
        processed_cursor += size;
    };

    int64_t processed_size;
    if (has_offset_input_) {
        processed_size =
            ProcessDataByOffsets<milvus::ArrayView>(execute_sub_batch,
                                                    std::nullptr_t{},
                                                    input,
                                                    res,
                                                    valid_res,
                                                    val1,
                                                    val2,
                                                    index);
    } else {
        processed_size = ProcessDataChunks<milvus::ArrayView>(execute_sub_batch,
                                                              std::nullptr_t{},
                                                              res,
                                                              valid_res,
                                                              val1,
                                                              val2,
                                                              index);
    }
    AssertInfo(processed_size == real_batch_size,
               "internal error: expr processed rows {} not equal "
               "expect batch size {}",
               processed_size,
               real_batch_size);
    return res_vec;
}

template <typename T>
VectorPtr
PhyBinaryRangeFilterExpr::ExecRangeVisitorImplForPk(EvalCtx& context) {
    typedef std::
        conditional_t<std::is_same_v<T, std::string_view>, std::string, T>
            PkInnerType;

    if (!arg_inited_) {
        lower_arg_.SetValue<PkInnerType>(expr_->lower_val_);
        upper_arg_.SetValue<PkInnerType>(expr_->upper_val_);
        arg_inited_ = true;
    }

    auto real_batch_size = GetNextBatchSize();
    if (real_batch_size == 0) {
        return nullptr;
    }

    if (cached_index_chunk_id_ != 0) {
        cached_index_chunk_id_ = 0;
        cached_index_chunk_res_ = std::make_shared<TargetBitmap>(active_count_);
        auto cache_view = cached_index_chunk_res_->view();

        PkType lower_pk = lower_arg_.GetValue<PkInnerType>();
        PkType upper_pk = upper_arg_.GetValue<PkInnerType>();
        segment_->pk_binary_range(op_ctx_,
                                  lower_pk,
                                  expr_->lower_inclusive_,
                                  upper_pk,
                                  expr_->upper_inclusive_,
                                  cache_view);
    }

    auto res = MoveOrSliceBitmap(
        *cached_index_chunk_res_, current_data_global_pos_, real_batch_size);
    MoveCursor();
    return res;
}

void
PhyBinaryRangeFilterExpr::DetermineExecPath() {
    // PkIndex (binary range only supports PK on sealed segments)
    if (is_pk_field_ && segment_->type() == SegmentType::Sealed) {
        exec_path_ = ExprExecPath::PkIndex;
        return;
    }

    // JsonStats
    if (CanUseJsonStatsAtInit()) {
        exec_path_ = ExprExecPath::JsonStats;
        return;
    }

    SegmentExpr::DetermineExecPath();
    if (exec_path_ != ExprExecPath::ScalarIndex) {
        return;
    }

    auto data_type = expr_->column_.data_type_;
    if (expr_->column_.element_level_) {
        data_type = expr_->column_.element_type_;
    }

    // ARRAY type cannot use scalar index
    if (data_type == DataType::ARRAY) {
        exec_path_ = ExprExecPath::RawData;
    }
}

void
PhyBinaryRangeFilterExpr::PrefetchRawData() {
    auto datatype = expr_->column_.data_type_;
    if (expr_->column_.element_level_) {
        datatype = expr_->column_.element_type_;
    }

    switch (datatype) {
        case DataType::BOOL:
            PrefetchRawData<bool>();
            break;
        case DataType::INT8:
            PrefetchRawData<int8_t>();
            break;
        case DataType::INT16:
            PrefetchRawData<int16_t>();
            break;
        case DataType::INT32:
            PrefetchRawData<int32_t>();
            break;
        case DataType::INT64:
            PrefetchRawData<int64_t>();
            break;
        case DataType::TIMESTAMPTZ:
            PrefetchRawData<int64_t>();
            break;
        case DataType::FLOAT:
            PrefetchRawData<float>();
            break;
        case DataType::DOUBLE:
            PrefetchRawData<double>();
            break;
        case DataType::VARCHAR:
            if (segment_->type() == SegmentType::Growing &&
                !storage::MmapManager::GetInstance()
                     .GetMmapConfig()
                     .growing_enable_mmap) {
                PrefetchRawData<std::string>();
            } else {
                PrefetchRawData<std::string_view>();
            }
            break;
        default:
            SegmentExpr::PrefetchRawData(expr_->column_.field_id_);
            break;
    }
}

template <typename T>
void
PhyBinaryRangeFilterExpr::PrefetchRawData() {
    using U =
        std::conditional_t<std::is_same_v<T, std::string_view>, std::string, T>;
    using H =
        std::conditional_t<std::is_integral_v<U> && !std::is_same_v<bool, T>,
                           int64_t,
                           U>;
    H lower_val = GetValueWithCastNumber<H>(expr_->lower_val_);
    H upper_val = GetValueWithCastNumber<H>(expr_->upper_val_);
    auto skip_index = segment_->GetSkipIndex();

    std::vector<int64_t> chunks_may_hit;
    for (size_t i = 0; i < num_data_chunk_; ++i) {
        auto skip = skip_index->CanSkipBinaryRange(field_id_,
                                                   i,
                                                   lower_val,
                                                   upper_val,
                                                   expr_->lower_inclusive_,
                                                   expr_->upper_inclusive_);
        if (!skip) {
            chunks_may_hit.push_back(i);
        }
    }

    segment_->prefetch_chunks(op_ctx_, field_id_, chunks_may_hit);
}
}  // namespace exec
}  // namespace milvus
