// 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/string_ref.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/index/zone_map/zonemap_eval_context.h"

#include "storage/segment/column_reader.h"

namespace doris {

class RowDescriptor;

class RuntimeState;

ZoneMapFilterResult VExpr::evaluate_zonemap_filter(const ZoneMapEvalContext& ctx) const {

return unsupported_zonemap_filter(ctx);

}

ZoneMapFilterResult VExpr::evaluate_dictionary_filter(const DictionaryEvalContext& ctx) const {

return ZoneMapFilterResult::kUnsupported;

}

ZoneMapFilterResult VExpr::evaluate_bloom_filter(const BloomFilterEvalContext& ctx) const {

return ZoneMapFilterResult::kUnsupported;

}

// 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;

}

}

TExprNode VExpr::clone_texpr_node() const {

TExprNode node;

node.__set_node_type(_node_type);

node.__set_opcode(_opcode);

node.__set_type(create_type_desc(remove_nullable(_data_type)->get_primitive_type(),

static_cast<int>(_data_type->get_precision()),

static_cast<int>(_data_type->get_scale())));

node.__set_is_nullable(_data_type->is_nullable());

node.__set_num_children(get_num_children());

node.__set_fn(_fn);

return node;

}

Status VExpr::clone_node(VExprSPtr* cloned_expr) const {

DORIS_CHECK(cloned_expr != nullptr);

return Status::NotSupported("Cannot clone expression {} for file-local rewrite", expr_name());

}

Status VExpr::deep_clone(VExprSPtr* cloned_expr,

const VExprCloneNodeOverride& clone_node_override) const {

DORIS_CHECK(cloned_expr != nullptr);

VExprSPtr cloned;

if (clone_node_override) {

RETURN_IF_ERROR(clone_node_override(*this, &cloned));

}

if (cloned == nullptr) {

RETURN_IF_ERROR(clone_node(&cloned));

}

DORIS_CHECK(cloned != nullptr);

VExprSPtrs cloned_children;

cloned_children.reserve(_children.size());

for (const auto& child : _children) {

DORIS_CHECK(child != nullptr);

VExprSPtr cloned_child;

RETURN_IF_ERROR(child->deep_clone(&cloned_child, clone_node_override));

cloned_children.push_back(std::move(cloned_child));

}

cloned->set_children(std::move(cloned_children));

cloned->reset_prepare_state();

*cloned_expr = std::move(cloned);

return Status::OK();

}

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::reset_prepare_state() {

_prepared = false;

_prepare_finished = false;

_open_finished = false;

for (auto& child : _children) {

child->reset_prepare_state();

}

}

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 && column_wrapper == nullptr) {

return Status::OK();

} else if (_constant_col != 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);