// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

#include "exprs/vexpr.h"

#include <fmt/format.h>
#include <gen_cpp/Exprs_types.h>
#include <gen_cpp/FrontendService_types.h>
#include <thrift/protocol/TDebugProtocol.h>

#include <algorithm>
#include <boost/algorithm/string/split.hpp>
#include <boost/iterator/iterator_facade.hpp>
#include <cstdint>
#include <memory>
#include <stack>
#include <string_view>
#include <utility>

#include "common/config.h"
#include "common/exception.h"
#include "common/status.h"
#include "core/column/column_nothing.h"
#include "core/column/column_vector.h"
#include "core/data_type/data_type_array.h"
#include "core/data_type/data_type_decimal.h"
#include "core/data_type/data_type_factory.hpp"
#include "core/data_type/data_type_nullable.h"
#include "core/data_type/data_type_number.h"
#include "core/data_type/define_primitive_type.h"
#include "core/field.h"
#include "core/value/timestamptz_value.h"
#include "exec/common/util.hpp"
#include "exec/pipeline/pipeline_task.h"
#include "exprs/short_circuit_evaluation_expr.h"
#include "exprs/varray_literal.h"
#include "exprs/vcase_expr.h"
#include "exprs/vcast_expr.h"
#include "exprs/vcolumn_ref.h"
#include "exprs/vcompound_pred.h"
#include "exprs/vcondition_expr.h"
#include "exprs/vectorized_fn_call.h"
#include "exprs/vexpr_context.h"
#include "exprs/vexpr_fwd.h"
#include "exprs/vin_predicate.h"
#include "exprs/vinfo_func.h"
#include "exprs/virtual_slot_ref.h"
#include "exprs/vlambda_function_call_expr.h"
#include "exprs/vlambda_function_expr.h"
#include "exprs/vliteral.h"
#include "exprs/vmap_literal.h"
#include "exprs/vmatch_predicate.h"
#include "exprs/vsearch.h"
#include "exprs/vslot_ref.h"
#include "exprs/vstruct_literal.h"
#include "storage/index/ann/ann_search_params.h"
#include "storage/index/ann/ann_topn_runtime.h"
#include "storage/index/inverted/inverted_index_parser.h"
#include "storage/segment/column_reader.h"

namespace doris {

class RowDescriptor;
class RuntimeState;

// NOLINTBEGIN(readability-function-cognitive-complexity)
// NOLINTBEGIN(readability-function-size)
TExprNode create_texpr_node_from(const void* data, const PrimitiveType& type, int precision,
                                 int scale) {
    TExprNode node;

    switch (type) {
    case TYPE_BOOLEAN: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_BOOLEAN>(data, &node));
        break;
    }
    case TYPE_TINYINT: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_TINYINT>(data, &node));
        break;
    }
    case TYPE_SMALLINT: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_SMALLINT>(data, &node));
        break;
    }
    case TYPE_INT: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_INT>(data, &node));
        break;
    }
    case TYPE_BIGINT: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_BIGINT>(data, &node));
        break;
    }
    case TYPE_LARGEINT: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_LARGEINT>(data, &node));
        break;
    }
    case TYPE_FLOAT: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_FLOAT>(data, &node));
        break;
    }
    case TYPE_DOUBLE: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DOUBLE>(data, &node));
        break;
    }
    case TYPE_DATEV2: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DATEV2>(data, &node));
        break;
    }
    case TYPE_DATETIMEV2: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DATETIMEV2>(data, &node, precision, scale));
        break;
    }
    case TYPE_DATE: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DATE>(data, &node));
        break;
    }
    case TYPE_DATETIME: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DATETIME>(data, &node));
        break;
    }
    case TYPE_DECIMALV2: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DECIMALV2>(data, &node, precision, scale));
        break;
    }
    case TYPE_DECIMAL32: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DECIMAL32>(data, &node, precision, scale));
        break;
    }
    case TYPE_DECIMAL64: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DECIMAL64>(data, &node, precision, scale));
        break;
    }
    case TYPE_DECIMAL128I: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DECIMAL128I>(data, &node, precision, scale));
        break;
    }
    case TYPE_DECIMAL256: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DECIMAL256>(data, &node, precision, scale));
        break;
    }
    case TYPE_CHAR: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_CHAR>(data, &node));
        break;
    }
    case TYPE_VARCHAR: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_VARCHAR>(data, &node));
        break;
    }
    case TYPE_STRING: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_STRING>(data, &node));
        break;
    }
    case TYPE_VARBINARY: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_VARBINARY>(data, &node));
        break;
    }
    case TYPE_IPV4: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_IPV4>(data, &node));
        break;
    }
    case TYPE_IPV6: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_IPV6>(data, &node));
        break;
    }
    case TYPE_TIMEV2: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_TIMEV2>(data, &node, precision, scale));
        break;
    }
    case TYPE_TIMESTAMPTZ: {
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_TIMESTAMPTZ>(data, &node, precision, scale));
        break;
    }
    default:
        throw Exception(ErrorCode::INTERNAL_ERROR, "runtime filter meet invalid type {}",
                        int(type));
    }
    return node;
}

TExprNode create_texpr_node_from(const Field& field, const PrimitiveType& type, int precision,
                                 int scale) {
    TExprNode node;
    switch (type) {
    case TYPE_BOOLEAN: {
        const auto& storage = static_cast<bool>(field.get<TYPE_BOOLEAN>());
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_BOOLEAN>(&storage, &node));
        break;
    }
    case TYPE_TINYINT: {
        const auto& storage = static_cast<int8_t>(field.get<TYPE_TINYINT>());
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_TINYINT>(&storage, &node));
        break;
    }
    case TYPE_SMALLINT: {
        const auto& storage = static_cast<int16_t>(field.get<TYPE_SMALLINT>());
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_SMALLINT>(&storage, &node));
        break;
    }
    case TYPE_INT: {
        const auto& storage = static_cast<int32_t>(field.get<TYPE_INT>());
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_INT>(&storage, &node));
        break;
    }
    case TYPE_BIGINT: {
        const auto& storage = static_cast<int64_t>(field.get<TYPE_BIGINT>());
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_BIGINT>(&storage, &node));
        break;
    }
    case TYPE_LARGEINT: {
        const auto& storage = static_cast<int128_t>(field.get<TYPE_LARGEINT>());
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_LARGEINT>(&storage, &node));
        break;
    }
    case TYPE_FLOAT: {
        const auto& storage = static_cast<float>(field.get<TYPE_FLOAT>());
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_FLOAT>(&storage, &node));
        break;
    }
    case TYPE_DOUBLE: {
        const auto& storage = static_cast<double>(field.get<TYPE_DOUBLE>());
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DOUBLE>(&storage, &node));
        break;
    }
    case TYPE_DATEV2: {
        const auto& storage = field.get<TYPE_DATEV2>();

        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DATEV2>(&storage, &node));
        break;
    }
    case TYPE_DATETIMEV2: {
        const auto& storage = field.get<TYPE_DATETIMEV2>();
        THROW_IF_ERROR(
                create_texpr_literal_node<TYPE_DATETIMEV2>(&storage, &node, precision, scale));
        break;
    }
    case TYPE_TIMESTAMPTZ: {
        const auto& storage = field.get<TYPE_TIMESTAMPTZ>();

        THROW_IF_ERROR(
                create_texpr_literal_node<TYPE_TIMESTAMPTZ>(&storage, &node, precision, scale));
        break;
    }
    case TYPE_DATE: {
        const auto& storage = field.get<TYPE_DATE>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DATE>(&storage, &node));
        break;
    }
    case TYPE_DATETIME: {
        const auto& storage = field.get<TYPE_DATETIME>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_DATETIME>(&storage, &node));
        break;
    }
    case TYPE_DECIMALV2: {
        const auto& storage = field.get<TYPE_DECIMALV2>();

        THROW_IF_ERROR(
                create_texpr_literal_node<TYPE_DECIMALV2>(&storage, &node, precision, scale));
        break;
    }
    case TYPE_DECIMAL32: {
        const auto& storage = field.get<TYPE_DECIMAL32>();
        THROW_IF_ERROR(
                create_texpr_literal_node<TYPE_DECIMAL32>(&storage, &node, precision, scale));
        break;
    }
    case TYPE_DECIMAL64: {
        const auto& storage = field.get<TYPE_DECIMAL64>();
        THROW_IF_ERROR(
                create_texpr_literal_node<TYPE_DECIMAL64>(&storage, &node, precision, scale));
        break;
    }
    case TYPE_DECIMAL128I: {
        const auto& storage = field.get<TYPE_DECIMAL128I>();
        THROW_IF_ERROR(
                create_texpr_literal_node<TYPE_DECIMAL128I>(&storage, &node, precision, scale));
        break;
    }
    case TYPE_DECIMAL256: {
        const auto& storage = field.get<TYPE_DECIMAL256>();
        THROW_IF_ERROR(
                create_texpr_literal_node<TYPE_DECIMAL256>(&storage, &node, precision, scale));
        break;
    }
    case TYPE_CHAR: {
        const auto& storage = field.get<TYPE_CHAR>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_CHAR>(&storage, &node));
        break;
    }
    case TYPE_VARCHAR: {
        const auto& storage = field.get<TYPE_VARCHAR>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_VARCHAR>(&storage, &node));
        break;
    }
    case TYPE_STRING: {
        const auto& storage = field.get<TYPE_STRING>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_STRING>(&storage, &node));
        break;
    }
    case TYPE_IPV4: {
        const auto& storage = field.get<TYPE_IPV4>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_IPV4>(&storage, &node));
        break;
    }
    case TYPE_IPV6: {
        const auto& storage = field.get<TYPE_IPV6>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_IPV6>(&storage, &node));
        break;
    }
    case TYPE_TIMEV2: {
        const auto& storage = field.get<TYPE_TIMEV2>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_TIMEV2>(&storage, &node));
        break;
    }
    case TYPE_VARBINARY: {
        const auto& svf = field.get<TYPE_VARBINARY>();
        THROW_IF_ERROR(create_texpr_literal_node<TYPE_VARBINARY>(&svf, &node));
        break;
    }
    default:
        throw Exception(ErrorCode::INTERNAL_ERROR, "runtime filter meet invalid type {}",
                        int(type));
    }
    return node;
}

// NOLINTEND(readability-function-size)
// NOLINTEND(readability-function-cognitive-complexity)
} // namespace doris

namespace doris {

VExpr::VExpr(const TExprNode& node)
        : _node_type(node.node_type),
          _opcode(node.__isset.opcode ? node.opcode : TExprOpcode::INVALID_OPCODE) {
    if (node.__isset.fn) {
        _fn = node.fn;
    }

    bool is_nullable = true;
    if (node.__isset.is_nullable) {
        is_nullable = node.is_nullable;
    }
    // If we define null literal ,should make nullable data type to get correct field instead of undefined ptr
    if (node.node_type == TExprNodeType::NULL_LITERAL) {
        CHECK(is_nullable);
    }
    _data_type = get_data_type_with_default_argument(
            DataTypeFactory::instance().create_data_type(node.type, is_nullable));
}

VExpr::VExpr(const VExpr& vexpr) = default;

VExpr::VExpr(DataTypePtr type, bool is_slotref)
        : _opcode(TExprOpcode::INVALID_OPCODE),
          _data_type(get_data_type_with_default_argument(type)) {
    if (is_slotref) {
        _node_type = TExprNodeType::SLOT_REF;
    }
}

Status VExpr::prepare(RuntimeState* state, const RowDescriptor& row_desc, VExprContext* context) {
    ++context->_depth_num;
    if (context->_depth_num > config::max_depth_of_expr_tree) {
        return Status::Error<ErrorCode::EXCEEDED_LIMIT>(
                "The depth of the expression tree is too big, make it less than {}",
                config::max_depth_of_expr_tree);
    }

    for (auto& i : _children) {
        RETURN_IF_ERROR(i->prepare(state, row_desc, context));
    }
    --context->_depth_num;
#ifndef BE_TEST
    _enable_inverted_index_query = state->query_options().enable_inverted_index_query;
#endif
    return Status::OK();
}

Status VExpr::open(RuntimeState* state, VExprContext* context,
                   FunctionContext::FunctionStateScope scope) {
    for (auto& i : _children) {
        RETURN_IF_ERROR(i->open(state, context, scope));
    }
    if (scope == FunctionContext::FRAGMENT_LOCAL) {
        RETURN_IF_ERROR(VExpr::get_const_col(context, nullptr));
    }
    return Status::OK();
}

void VExpr::close(VExprContext* context, FunctionContext::FunctionStateScope scope) {
    for (auto& i : _children) {
        i->close(context, scope);
    }
}

// NOLINTBEGIN(readability-function-size)
Status VExpr::create_expr(const TExprNode& expr_node, VExprSPtr& expr) {
    try {
        switch (expr_node.node_type) {
        case TExprNodeType::BOOL_LITERAL:
        case TExprNodeType::INT_LITERAL:
        case TExprNodeType::LARGE_INT_LITERAL:
        case TExprNodeType::IPV4_LITERAL:
        case TExprNodeType::IPV6_LITERAL:
        case TExprNodeType::FLOAT_LITERAL:
        case TExprNodeType::DECIMAL_LITERAL:
        case TExprNodeType::DATE_LITERAL:
        case TExprNodeType::TIMEV2_LITERAL:
        case TExprNodeType::STRING_LITERAL:
        case TExprNodeType::JSON_LITERAL:
        case TExprNodeType::VARBINARY_LITERAL:
        case TExprNodeType::NULL_LITERAL: {
            expr = VLiteral::create_shared(expr_node);
            break;
        }
        case TExprNodeType::ARRAY_LITERAL: {
            expr = VArrayLiteral::create_shared(expr_node);
            break;
        }
        case TExprNodeType::MAP_LITERAL: {
            expr = VMapLiteral::create_shared(expr_node);
            break;
        }
        case TExprNodeType::STRUCT_LITERAL: {
            expr = VStructLiteral::create_shared(expr_node);
            break;
        }
        case TExprNodeType::SLOT_REF: {
            if (expr_node.slot_ref.__isset.is_virtual_slot && expr_node.slot_ref.is_virtual_slot) {
                expr = VirtualSlotRef::create_shared(expr_node);
                expr->_node_type = TExprNodeType::VIRTUAL_SLOT_REF;
            } else {
                expr = VSlotRef::create_shared(expr_node);
            }
            break;
        }
        case TExprNodeType::COLUMN_REF: {
            expr = VColumnRef::create_shared(expr_node);
            break;
        }
        case TExprNodeType::COMPOUND_PRED: {
            expr = VCompoundPred::create_shared(expr_node);
            break;
        }
        case TExprNodeType::LAMBDA_FUNCTION_EXPR: {
            expr = VLambdaFunctionExpr::create_shared(expr_node);
            break;
        }
        case TExprNodeType::LAMBDA_FUNCTION_CALL_EXPR: {
            expr = VLambdaFunctionCallExpr::create_shared(expr_node);
            break;
        }
        case TExprNodeType::ARITHMETIC_EXPR:
        case TExprNodeType::BINARY_PRED:
        case TExprNodeType::NULL_AWARE_BINARY_PRED:
        case TExprNodeType::COMPUTE_FUNCTION_CALL: {
            expr = VectorizedFnCall::create_shared(expr_node);
            break;
        }
        case TExprNodeType::FUNCTION_CALL: {
            if (expr_node.fn.name.function_name == "if") {
                if (expr_node.__isset.short_circuit_evaluation &&
                    expr_node.short_circuit_evaluation) {
                    expr = ShortCircuitIfExpr::create_shared(expr_node);
                } else {
                    expr = VectorizedIfExpr::create_shared(expr_node);
                }
                break;
            } else if (expr_node.fn.name.function_name == "ifnull" ||
                       expr_node.fn.name.function_name == "nvl") {
                if (expr_node.__isset.short_circuit_evaluation &&
                    expr_node.short_circuit_evaluation) {
                    expr = ShortCircuitIfNullExpr::create_shared(expr_node);
                } else {
                    expr = VectorizedIfNullExpr::create_shared(expr_node);
                }
                break;
            } else if (expr_node.fn.name.function_name == "coalesce") {
                if (expr_node.__isset.short_circuit_evaluation &&
                    expr_node.short_circuit_evaluation) {
                    expr = ShortCircuitCoalesceExpr::create_shared(expr_node);
                } else {
                    expr = VectorizedCoalesceExpr::create_shared(expr_node);
                }
                break;
            }
            expr = VectorizedFnCall::create_shared(expr_node);
            break;
        }
        case TExprNodeType::MATCH_PRED: {
            expr = VMatchPredicate::create_shared(expr_node);
            break;
        }
        case TExprNodeType::CAST_EXPR: {
            expr = VCastExpr::create_shared(expr_node);
            break;
        }
        case TExprNodeType::TRY_CAST_EXPR: {
            expr = TryCastExpr::create_shared(expr_node);
            break;
        }
        case TExprNodeType::IN_PRED: {
            expr = VInPredicate::create_shared(expr_node);
            break;
        }
        case TExprNodeType::CASE_EXPR: {
            if (!expr_node.__isset.case_expr) {
                return Status::InternalError("Case expression not set in thrift node");
            }
            if (expr_node.__isset.short_circuit_evaluation && expr_node.short_circuit_evaluation) {
                expr = ShortCircuitCaseExpr::create_shared(expr_node);
            } else {
                expr = VCaseExpr::create_shared(expr_node);
            }
            break;
        }
        case TExprNodeType::INFO_FUNC: {
            expr = VInfoFunc::create_shared(expr_node);
            break;
        }
        case TExprNodeType::SEARCH_EXPR: {
            expr = VSearchExpr::create_shared(expr_node);
            break;
        }
        default:
            return Status::InternalError("Unknown expr node type: {}", expr_node.node_type);
        }
    } catch (const Exception& e) {
        if (e.code() == ErrorCode::INTERNAL_ERROR) {
            return Status::InternalError("Create Expr failed because {}\nTExprNode={}", e.what(),
                                         apache::thrift::ThriftDebugString(expr_node));
        }
        return Status::Error<false>(e.code(), "Create Expr failed because {}", e.what());
        LOG(WARNING) << "create expr failed, TExprNode={}, reason={}"
                     << apache::thrift::ThriftDebugString(expr_node) << e.what();
    }
    if (!expr->data_type()) {
        return Status::InvalidArgument("Unknown expr type: {}", expr_node.node_type);
    }
    return Status::OK();
}
// NOLINTEND(readability-function-size)

Status VExpr::create_tree_from_thrift(const std::vector<TExprNode>& nodes, int* node_idx,
                                      VExprSPtr& root_expr, VExprContextSPtr& ctx) {
    // propagate error case
    if (*node_idx >= nodes.size()) {
        return Status::InternalError("Failed to reconstruct expression tree from thrift.");
    }

    // create root expr
    int root_children = nodes[*node_idx].num_children;
    VExprSPtr root;
    RETURN_IF_ERROR(create_expr(nodes[*node_idx], root));
    DCHECK(root != nullptr);
    root_expr = root;
    ctx = std::make_shared<VExprContext>(root);
    // short path for leaf node
    if (root_children <= 0) {
        return Status::OK();
    }

    // non-recursive traversal
    using VExprSPtrCountPair = std::pair<VExprSPtr, int>;
    std::stack<std::shared_ptr<VExprSPtrCountPair>> s;
    s.emplace(std::make_shared<VExprSPtrCountPair>(root, root_children));
    while (!s.empty()) {
        // copy the shared ptr resource to avoid dangling reference
        auto parent = s.top();
        // Decrement or pop
        if (parent->second > 1) {
            parent->second -= 1;
        } else {
            s.pop();
        }

        DCHECK(parent->first != nullptr);
        if (++*node_idx >= nodes.size()) {
            return Status::InternalError("Failed to reconstruct expression tree from thrift.");
        }

        VExprSPtr expr;
        RETURN_IF_ERROR(create_expr(nodes[*node_idx], expr));
        DCHECK(expr != nullptr);
        parent->first->add_child(expr);
        // push to stack if has children
        int num_children = nodes[*node_idx].num_children;
        if (num_children > 0) {
            s.emplace(std::make_shared<VExprSPtrCountPair>(expr, num_children));
        }
    }
    return Status::OK();
}

Status VExpr::create_expr_tree(const TExpr& texpr, VExprContextSPtr& ctx) {
    if (texpr.nodes.empty()) {
        ctx = nullptr;
        return Status::OK();
    }
    int node_idx = 0;
    VExprSPtr e;
    Status status = create_tree_from_thrift(texpr.nodes, &node_idx, e, ctx);
    if (status.ok() && node_idx + 1 != texpr.nodes.size()) {
        status = Status::InternalError(
                "Expression tree only partially reconstructed. Not all thrift nodes were "
                "used.");
    }
    if (!status.ok()) {
        LOG(ERROR) << "Could not construct expr tree.\n"
                   << status << "\n"
                   << apache::thrift::ThriftDebugString(texpr);
    }
    return status;
}

Status VExpr::create_expr_trees(const std::vector<TExpr>& texprs, VExprContextSPtrs& ctxs) {
    ctxs.clear();
    for (const auto& texpr : texprs) {
        VExprContextSPtr ctx;
        RETURN_IF_ERROR(create_expr_tree(texpr, ctx));
        ctxs.push_back(ctx);
    }
    return Status::OK();
}

Status VExpr::check_expr_output_type(const VExprContextSPtrs& ctxs,
                                     const RowDescriptor& output_row_desc) {
    if (ctxs.empty()) {
        return Status::OK();
    }
    auto name_and_types = VectorizedUtils::create_name_and_data_types(output_row_desc);
    if (ctxs.size() != name_and_types.size()) {
        return Status::InternalError(
                "output type size not match expr size {} , expected output size {} ", ctxs.size(),
                name_and_types.size());
    }
    auto check_type_can_be_converted = [](DataTypePtr& from, DataTypePtr& to) -> bool {
        if (to->equals(*from)) {
            return true;
        }
        if (to->is_nullable() && !from->is_nullable()) {
            return remove_nullable(to)->equals(*from);
        }
        return false;
    };
    for (int i = 0; i < ctxs.size(); i++) {
        auto real_expr_type = get_data_type_with_default_argument(ctxs[i]->root()->data_type());
        auto&& [name, expected_type] = name_and_types[i];
        if (!check_type_can_be_converted(real_expr_type, expected_type)) {
            return Status::InternalError(
                    "output type not match expr type, col name {} , expected type {} , real type "
                    "{}",
                    name, expected_type->get_name(), real_expr_type->get_name());
        }
    }
    return Status::OK();
}

Status VExpr::prepare(const VExprContextSPtrs& ctxs, RuntimeState* state,
                      const RowDescriptor& row_desc) {
    for (auto ctx : ctxs) {
        RETURN_IF_ERROR(ctx->prepare(state, row_desc));
    }
    return Status::OK();
}

Status VExpr::open(const VExprContextSPtrs& ctxs, RuntimeState* state) {
    for (const auto& ctx : ctxs) {
        RETURN_IF_ERROR(ctx->open(state));
    }
    return Status::OK();
}

bool VExpr::contains_blockable_function(const VExprContextSPtrs& ctxs) {
    return std::any_of(ctxs.begin(), ctxs.end(), [](const VExprContextSPtr& ctx) {
        return ctx != nullptr && ctx->root() != nullptr && ctx->root()->is_blockable();
    });
}

Status VExpr::clone_if_not_exists(const VExprContextSPtrs& ctxs, RuntimeState* state,
                                  VExprContextSPtrs& new_ctxs) {
    if (!new_ctxs.empty()) {
        // 'ctxs' was already cloned into '*new_ctxs', nothing to do.
        DCHECK_EQ(new_ctxs.size(), ctxs.size());
        for (auto& new_ctx : new_ctxs) {
            DCHECK(new_ctx->_is_clone);
        }
        return Status::OK();
    }
    new_ctxs.resize(ctxs.size());
    for (int i = 0; i < ctxs.size(); ++i) {
        RETURN_IF_ERROR(ctxs[i]->clone(state, new_ctxs[i]));
    }
    return Status::OK();
}

std::string VExpr::debug_string() const {
    // TODO: implement partial debug string for member vars
    std::stringstream out;
    out << " type=" << _data_type->get_name();

    if (!_children.empty()) {
        out << " children=" << debug_string(_children);
    }

    return out.str();
}

std::string VExpr::debug_string(const VExprSPtrs& exprs) {
    std::stringstream out;
    out << "[";

    for (int i = 0; i < exprs.size(); ++i) {
        out << (i == 0 ? "" : " ") << exprs[i]->debug_string();
    }

    out << "]";
    return out.str();
}

std::string VExpr::debug_string(const VExprContextSPtrs& ctxs) {
    VExprSPtrs exprs;
    for (const auto& ctx : ctxs) {
        exprs.push_back(ctx->root());
    }
    return debug_string(exprs);
}

bool VExpr::is_constant() const {
    return std::all_of(_children.begin(), _children.end(),
                       [](const VExprSPtr& expr) { return expr->is_constant(); });
}

Status VExpr::get_const_col(VExprContext* context,
                            std::shared_ptr<ColumnPtrWrapper>* column_wrapper) {
    if (!is_constant()) {
        return Status::OK();
    }

    if (_constant_col != nullptr) {
        DCHECK(column_wrapper != nullptr);
        *column_wrapper = _constant_col;
        return Status::OK();
    }

    ColumnPtr result;
    RETURN_IF_ERROR(execute_column(context, nullptr, nullptr, 1, result));
    _constant_col = std::make_shared<ColumnPtrWrapper>(result);
    if (column_wrapper != nullptr) {
        *column_wrapper = _constant_col;
    }

    return Status::OK();
}

void VExpr::register_function_context(RuntimeState* state, VExprContext* context) {
    std::vector<DataTypePtr> arg_types;
    for (auto& i : _children) {
        arg_types.push_back(i->data_type());
    }

    _fn_context_index = context->register_function_context(state, _data_type, arg_types);
}

Status VExpr::init_function_context(RuntimeState* state, VExprContext* context,
                                    FunctionContext::FunctionStateScope scope,
                                    const FunctionBasePtr& function) const {
    FunctionContext* fn_ctx = context->fn_context(_fn_context_index);
    if (scope == FunctionContext::FRAGMENT_LOCAL) {
        std::vector<std::shared_ptr<ColumnPtrWrapper>> constant_cols;
        for (auto c : _children) {
            std::shared_ptr<ColumnPtrWrapper> const_col;
            RETURN_IF_ERROR(c->get_const_col(context, &const_col));
            constant_cols.push_back(const_col);
        }
        fn_ctx->set_constant_cols(constant_cols);
    }

    if (scope == FunctionContext::FRAGMENT_LOCAL) {
        RETURN_IF_ERROR(function->open(fn_ctx, FunctionContext::FRAGMENT_LOCAL));
    }
    RETURN_IF_ERROR(function->open(fn_ctx, FunctionContext::THREAD_LOCAL));
    return Status::OK();
}

void VExpr::close_function_context(VExprContext* context, FunctionContext::FunctionStateScope scope,
                                   const FunctionBasePtr& function) const {
    if (_fn_context_index != -1) {
        FunctionContext* fn_ctx = context->fn_context(_fn_context_index);
        // `close_function_context` is called in VExprContext's destructor so do not throw exceptions here.
        static_cast<void>(function->close(fn_ctx, FunctionContext::THREAD_LOCAL));
        if (scope == FunctionContext::FRAGMENT_LOCAL) {
            static_cast<void>(function->close(fn_ctx, FunctionContext::FRAGMENT_LOCAL));
        }
    }
}

Status VExpr::check_constant(const Block& block, ColumnNumbers arguments) const {
    if (is_constant() && !VectorizedUtils::all_arguments_are_constant(block, arguments)) {
        return Status::InternalError("const check failed, expr={}", debug_string());
    }
    return Status::OK();
}

uint64_t VExpr::get_digest(uint64_t seed) const {
    auto digest = seed;
    for (auto child : _children) {
        digest = child->get_digest(digest);
        if (digest == 0) {
            return 0;
        }
    }

    auto& fn_name = _fn.name.function_name;
    if (!fn_name.empty()) {
        digest = HashUtil::hash64(fn_name.c_str(), fn_name.size(), digest);
    } else {
        digest = HashUtil::hash64((const char*)&_node_type, sizeof(_node_type), digest);
        digest = HashUtil::hash64((const char*)&_opcode, sizeof(_opcode), digest);
    }
    return digest;
}

ColumnPtr VExpr::get_result_from_const(size_t count) const {
    return ColumnConst::create(_constant_col->column_ptr, count);
}

Status VExpr::_evaluate_inverted_index(VExprContext* context, const FunctionBasePtr& function,
                                       uint32_t segment_num_rows) {
    // Pre-allocate vectors based on an estimated or known size
    std::vector<segment_v2::IndexIterator*> iterators;
    std::vector<IndexFieldNameAndTypePair> data_type_with_names;
    std::vector<int> column_ids;
    ColumnsWithTypeAndName arguments;
    VExprSPtrs children_exprs;

    // Reserve space to avoid multiple reallocations
    const size_t estimated_size = get_num_children();
    iterators.reserve(estimated_size);
    data_type_with_names.reserve(estimated_size);
    column_ids.reserve(estimated_size);
    children_exprs.reserve(estimated_size);

    auto index_context = context->get_index_context();

    // if child is cast expr, we need to ensure target data type is the same with storage data type.
    // or they are all string type
    // and if data type is array, we need to get the nested data type to ensure that.
    for (const auto& child : children()) {
        if (child->node_type() == TExprNodeType::CAST_EXPR) {
            auto* cast_expr = assert_cast<VCastExpr*>(child.get());
            DCHECK_EQ(cast_expr->get_num_children(), 1);
            if (cast_expr->get_child(0)->is_slot_ref()) {
                auto* column_slot_ref = assert_cast<VSlotRef*>(cast_expr->get_child(0).get());
                auto column_id = column_slot_ref->column_id();
                const auto* storage_name_type =
                        context->get_index_context()->get_storage_name_and_type_by_column_id(
                                column_id);
                auto storage_type = remove_nullable(storage_name_type->second);
                auto target_type = remove_nullable(cast_expr->get_target_type());
                auto origin_primitive_type = storage_type->get_primitive_type();
                auto target_primitive_type = target_type->get_primitive_type();
                if (is_complex_type(storage_type->get_primitive_type())) {
                    if (storage_type->get_primitive_type() == TYPE_ARRAY &&
                        target_type->get_primitive_type() == TYPE_ARRAY) {
                        auto nested_storage_type =
                                (assert_cast<const DataTypeArray*>(storage_type.get()))
                                        ->get_nested_type();
                        origin_primitive_type = nested_storage_type->get_primitive_type();
                        auto nested_target_type =
                                (assert_cast<const DataTypeArray*>(target_type.get()))
                                        ->get_nested_type();
                        target_primitive_type = nested_target_type->get_primitive_type();
                    } else {
                        continue;
                    }
                }
                if (origin_primitive_type != TYPE_VARIANT &&
                    (storage_type->equals(*target_type) ||
                     (is_string_type(target_primitive_type) &&
                      is_string_type(origin_primitive_type)))) {
                    children_exprs.emplace_back(expr_without_cast(child));
                }
            } else {
                return Status::OK(); // for example: cast("abc") as ipv4 case
            }
        } else {
            children_exprs.emplace_back(child);
        }
    }

    if (children_exprs.empty()) {
        return Status::OK(); // Early exit if no children to process
    }

    for (const auto& child : children_exprs) {
        if (child->is_slot_ref()) {
            auto* column_slot_ref = assert_cast<VSlotRef*>(child.get());
            auto column_id = column_slot_ref->column_id();
            auto* iter = context->get_index_context()->get_inverted_index_iterator_by_column_id(
                    column_id);
            //column does not have inverted index
            if (iter == nullptr) {
                continue;
            }
            const auto* storage_name_type =
                    context->get_index_context()->get_storage_name_and_type_by_column_id(column_id);
            if (storage_name_type == nullptr) {
                auto err_msg = fmt::format(
                        "storage_name_type cannot be found for column {} while in {} "
                        "evaluate_inverted_index",
                        column_id, expr_name());
                LOG(ERROR) << err_msg;
                return Status::InternalError(err_msg);
            }
            iterators.emplace_back(iter);
            data_type_with_names.emplace_back(*storage_name_type);
            column_ids.emplace_back(column_id);
        } else if (child->is_literal()) {
            auto* column_literal = assert_cast<VLiteral*>(child.get());
            arguments.emplace_back(column_literal->get_column_ptr(),
                                   column_literal->get_data_type(), column_literal->expr_name());
        } else if (child->can_push_down_to_index()) {
            RETURN_IF_ERROR(child->evaluate_inverted_index(context, segment_num_rows));
        } else {
            return Status::OK(); // others cases
        }
    }

    // is null or is not null has no arguments
    if (iterators.empty() || (arguments.empty() && !(function->get_name() == "is_not_null_pred" ||
                                                     function->get_name() == "is_null_pred"))) {
        return Status::OK(); // Nothing to evaluate or no literals to compare against
    }

    const InvertedIndexAnalyzerCtx* analyzer_ctx = nullptr;
    if (auto index_ctx = context->get_index_context(); index_ctx != nullptr) {
        analyzer_ctx = index_ctx->get_analyzer_ctx_for_expr(this);
    }

    auto result_bitmap = segment_v2::InvertedIndexResultBitmap();
    // Pass analyzer_key to function (used by match predicates for multi-analyzer index selection)
    auto res = function->evaluate_inverted_index(arguments, data_type_with_names, iterators,
                                                 segment_num_rows, analyzer_ctx, result_bitmap);
    if (!res.ok()) {
        return res;
    }
    if (!result_bitmap.is_empty()) {
        index_context->set_index_result_for_expr(this, result_bitmap);
        for (int column_id : column_ids) {
            index_context->set_true_for_index_status(this, column_id);
        }
    }
    return Status::OK();
}

size_t VExpr::estimate_memory(const size_t rows) {
    if (is_const_and_have_executed()) {
        return 0;
    }

    size_t estimate_size = 0;
    for (auto& child : _children) {
        estimate_size += child->estimate_memory(rows);
    }

    if (_data_type->have_maximum_size_of_value()) {
        estimate_size += rows * _data_type->get_size_of_value_in_memory();
    } else {
        estimate_size += rows * 64; /// TODO: need a more reasonable value
    }
    return estimate_size;
}

Status VExpr::execute_column(VExprContext* context, const Block* block, const Selector* selector,
                             size_t count, ColumnPtr& result_column) const {
    RETURN_IF_ERROR(execute_column_impl(context, block, selector, count, result_column));
    if (result_column->size() != count) {
        return Status::InternalError("Expr {} return column size {} not equal to expected size {}",
                                     expr_name(), result_column->size(), count);
    }
    DCHECK(selector == nullptr || selector->size() == count);
    // Validate type match. ColumnNothing is exempt (used as a placeholder in tests/stubs).
    if (!check_and_get_column<ColumnNothing>(result_column.get())) {
        auto result_type = execute_type(block);
        if (result_type != nullptr) {
            Status st = result_type->check_column(*result_column);
            if (!st.ok()) {
                // Nullable(T) may legitimately produce a non-nullable T column when all rows are
                // non-null (use_default_implementation_for_nulls optimization). Allow this.
                const auto* nullable_type =
                        check_and_get_data_type<DataTypeNullable>(result_type.get());
                if (nullable_type && !check_and_get_column<ColumnNullable>(result_column.get())) {
                    st = nullable_type->get_nested_type()->check_column(*result_column);
                }
            }
            if (!st.ok()) {
                return Status::InternalError(
                        "Expr {} return column type mismatch: declared={}, actual={}", expr_name(),
                        result_type->get_name(), result_column->get_name());
            }
        }
    }
    return Status::OK();
}

bool VExpr::fast_execute(VExprContext* context, const Selector* selector, size_t count,
                         ColumnPtr& result_column) const {
    if (context->get_index_context() &&
        context->get_index_context()->get_index_result_column().contains(this)) {
        // prepare a column to save result
        result_column = filter_column_with_selector(
                context->get_index_context()->get_index_result_column()[this], selector, count);
        if (_data_type->is_nullable()) {
            result_column = make_nullable(result_column);
        }
        return true;
    }
    return false;
}

bool VExpr::equals(const VExpr& other) {
    return false;
}

Status VExpr::evaluate_ann_range_search(
        const segment_v2::AnnRangeSearchRuntime& runtime,
        const std::vector<std::unique_ptr<segment_v2::IndexIterator>>& index_iterators,
        const std::vector<ColumnId>& idx_to_cid,
        const std::vector<std::unique_ptr<segment_v2::ColumnIterator>>& column_iterators,
        roaring::Roaring& row_bitmap, AnnIndexStats& ann_index_stats, bool enable_result_cache,
        AnnRangeSearchEvaluationResult& result) {
    result = {};
    return Status::OK();
}

void VExpr::prepare_ann_range_search(const doris::VectorSearchUserParams& params,
                                     segment_v2::AnnRangeSearchRuntime& range_search_runtime,
                                     bool& suitable_for_ann_index) {
    if (!suitable_for_ann_index) {
        return;
    }
    for (auto& child : _children) {
        child->prepare_ann_range_search(params, range_search_runtime, suitable_for_ann_index);
        if (!suitable_for_ann_index) {
            return;
        }
    }
}

Status VExpr::execute_filter(VExprContext* context, const Block* block,
                             uint8_t* __restrict result_filter_data, size_t rows, bool accept_null,
                             bool* can_filter_all) const {
    ColumnPtr filter_column;
    RETURN_IF_ERROR(execute_column(context, block, nullptr, rows, filter_column));
    if (const auto* const_column = check_and_get_column<ColumnConst>(*filter_column)) {
        // const(nullable) or const(bool)
        const bool result = accept_null
                                    ? (const_column->is_null_at(0) || const_column->get_bool(0))
                                    : (!const_column->is_null_at(0) && const_column->get_bool(0));
        if (!result) {
            // filter all
            *can_filter_all = true;
            memset(result_filter_data, 0, rows);
            return Status::OK();
        }
    } else if (const auto* nullable_column = check_and_get_column<ColumnNullable>(*filter_column)) {
        // nullable(bool)
        const ColumnPtr& nested_column = nullable_column->get_nested_column_ptr();
        const IColumn::Filter& filter = assert_cast<const ColumnUInt8&>(*nested_column).get_data();
        const auto* __restrict filter_data = filter.data();
        const auto* __restrict null_map_data = nullable_column->get_null_map_data().data();

        if (accept_null) {
            for (size_t i = 0; i < rows; ++i) {
                result_filter_data[i] &= (null_map_data[i]) || filter_data[i];
            }
        } else {
            for (size_t i = 0; i < rows; ++i) {
                result_filter_data[i] &= (!null_map_data[i]) & filter_data[i];
            }
        }

        if (!simd::contain_one(result_filter_data, rows)) {
            *can_filter_all = true;
            return Status::OK();
        }
    } else {
        // bool
        const IColumn::Filter& filter = assert_cast<const ColumnUInt8&>(*filter_column).get_data();
        const auto* __restrict filter_data = filter.data();

        for (size_t i = 0; i < rows; ++i) {
            result_filter_data[i] &= filter_data[i];
        }

        if (!simd::contain_one(result_filter_data, rows)) {
            *can_filter_all = true;
            return Status::OK();
        }
    }

    return Status::OK();
}

} // namespace doris