expr.cpp

// This file is made available under Elastic License 2.0.
// This file is based on code available under the Apache license here:
//   https://github.com/apache/incubator-doris/blob/master/be/src/exprs/expr.cpp

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

#include <thrift/protocol/TDebugProtocol.h>

#include <sstream>
#include <utility>
#include <vector>

#include "column/fixed_length_column.h"
#include "common/object_pool.h"
#include "common/status.h"
#include "exprs/vectorized/arithmetic_expr.h"
#include "exprs/vectorized/array_element_expr.h"
#include "exprs/vectorized/array_expr.h"
#include "exprs/vectorized/binary_predicate.h"
#include "exprs/vectorized/case_expr.h"
#include "exprs/vectorized/cast_expr.h"
#include "exprs/vectorized/clone_expr.h"
#include "exprs/vectorized/column_ref.h"
#include "exprs/vectorized/compound_predicate.h"
#include "exprs/vectorized/condition_expr.h"
#include "exprs/vectorized/dictmapping_expr.h"
#include "exprs/vectorized/function_call_expr.h"
#include "exprs/vectorized/in_predicate.h"
#include "exprs/vectorized/info_func.h"
#include "exprs/vectorized/is_null_predicate.h"
#include "exprs/vectorized/java_function_call_expr.h"
#include "exprs/vectorized/literal.h"
#include "exprs/vectorized/placeholder_ref.h"
#include "gen_cpp/Exprs_types.h"
#include "gen_cpp/Types_types.h"
#include "runtime/primitive_type.h"
#include "runtime/raw_value.h"
#include "runtime/runtime_state.h"

#pragma clang diagnostic push
#pragma ide diagnostic ignored "EndlessLoop"
using std::vector;
namespace starrocks {

using vectorized::Int8Column;
using vectorized::Int16Column;
using vectorized::Int32Column;
using vectorized::Int64Column;
using vectorized::Int128Column;
using vectorized::DoubleColumn;
using vectorized::FloatColumn;
using vectorized::BooleanColumn;

// No children here
Expr::Expr(const Expr& expr)
        : _cache_entry(expr._cache_entry),
          _node_type(expr._node_type),
          _opcode(expr._opcode),
          _is_slotref(expr._is_slotref),
          _is_nullable(expr._is_nullable),
          _type(expr._type),
          _output_scale(expr._output_scale),
          _fn(expr._fn),
          _fn_context_index(expr._fn_context_index) {}

Expr::Expr(TypeDescriptor type) : Expr(type, false) {}

Expr::Expr(TypeDescriptor type, bool is_slotref)
        : _opcode(TExprOpcode::INVALID_OPCODE),
          // _vector_opcode(TExprOpcode::INVALID_OPCODE),
          _is_slotref(is_slotref),
          _type(std::move(type)),
          _output_scale(-1),
          _fn_context_index(-1) {
    if (is_slotref) {
        _node_type = (TExprNodeType::SLOT_REF);
    } else {
        switch (_type.type) {
        case TYPE_BOOLEAN:
            _node_type = (TExprNodeType::BOOL_LITERAL);
            break;

        case TYPE_TINYINT:
        case TYPE_SMALLINT:
        case TYPE_INT:
        case TYPE_BIGINT:
            _node_type = (TExprNodeType::INT_LITERAL);
            break;

        case TYPE_LARGEINT:
            _node_type = (TExprNodeType::LARGE_INT_LITERAL);
            break;

        case TYPE_NULL:
            _node_type = (TExprNodeType::NULL_LITERAL);
            break;

        case TYPE_FLOAT:
        case TYPE_DOUBLE:
        case TYPE_TIME:
            _node_type = (TExprNodeType::FLOAT_LITERAL);
            break;

        case TYPE_DECIMAL:
        case TYPE_DECIMALV2:
            _node_type = (TExprNodeType::DECIMAL_LITERAL);
            break;

        case TYPE_DATE:
        case TYPE_DATETIME:
            _node_type = (TExprNodeType::DATE_LITERAL);
            break;

        case TYPE_CHAR:
        case TYPE_VARCHAR:
        case TYPE_HLL:
        case TYPE_OBJECT:
        case TYPE_PERCENTILE:
            _node_type = (TExprNodeType::STRING_LITERAL);
            break;
        case TYPE_ARRAY:
            _node_type = (TExprNodeType::ARRAY_EXPR);
            break;
        case INVALID_TYPE:
        case TYPE_BINARY:
        case TYPE_STRUCT:
        case TYPE_MAP:
        case TYPE_DECIMAL32:
        case TYPE_DECIMAL64:
        case TYPE_DECIMAL128:
        case TYPE_JSON:
            break;

        default:
            DCHECK(false) << "Invalid type." << _type.type;
        }
    }
}

Expr::Expr(const TExprNode& node) : Expr(node, false) {}

Expr::Expr(const TExprNode& node, bool is_slotref)
        : _node_type(node.node_type),
          _opcode(node.__isset.opcode ? node.opcode : TExprOpcode::INVALID_OPCODE),
          // _vector_opcode(
          // node.__isset.vector_opcode ? node.vector_opcode : TExprOpcode::INVALID_OPCODE),
          _is_slotref(is_slotref),
          _is_nullable(node.is_nullable),
          _type(TypeDescriptor::from_thrift(node.type)),
          _output_scale(node.output_scale),
          _fn_context_index(-1) {
    if (node.__isset.fn) {
        _fn = node.fn;
    }
    if (node.__isset.is_monotonic) {
        _is_monotonic = node.is_monotonic;
    }
}

Expr::~Expr() = default;

Status Expr::create_expr_tree(ObjectPool* pool, const TExpr& texpr, ExprContext** ctx) {
    // input is empty
    if (texpr.nodes.empty()) {
        *ctx = nullptr;
        return Status::OK();
    }
    int node_idx = 0;
    Expr* e = nullptr;
    Status status = create_tree_from_thrift(pool, texpr.nodes, nullptr, &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.get_error_msg() << "\n"
                   << apache::thrift::ThriftDebugString(texpr);
    }
    return status;
}

Status Expr::create_expr_trees(ObjectPool* pool, const std::vector<TExpr>& texprs, std::vector<ExprContext*>* ctxs) {
    ctxs->clear();
    for (const auto& texpr : texprs) {
        ExprContext* ctx = nullptr;
        RETURN_IF_ERROR(create_expr_tree(pool, texpr, &ctx));
        ctxs->push_back(ctx);
    }
    return Status::OK();
}

Status Expr::create_tree_from_thrift(ObjectPool* pool, const std::vector<TExprNode>& nodes, Expr* parent, int* node_idx,
                                     Expr** root_expr, ExprContext** ctx) {
    // propagate error case
    if (*node_idx >= nodes.size()) {
        return Status::InternalError("Failed to reconstruct expression tree from thrift.");
    }
    int num_children = nodes[*node_idx].num_children;
    Expr* expr = nullptr;
    RETURN_IF_ERROR(create_vectorized_expr(pool, nodes[*node_idx], &expr));
    DCHECK(expr != nullptr);
    if (parent != nullptr) {
        parent->add_child(expr);
    }
    for (int i = 0; i < num_children; i++) {
        *node_idx += 1;
        RETURN_IF_ERROR(create_tree_from_thrift(pool, nodes, expr, node_idx, nullptr, nullptr));
        // we are expecting a child, but have used all nodes
        // this means we have been given a bad tree and must fail
        if (*node_idx >= nodes.size()) {
            return Status::InternalError("Failed to reconstruct expression tree from thrift.");
        }
    }
    if (parent == nullptr) {
        DCHECK(root_expr != nullptr);
        DCHECK(ctx != nullptr);
        *root_expr = expr;
        *ctx = pool->add(new ExprContext(expr));
    }
    return Status::OK();
}

Status Expr::create_vectorized_expr(starrocks::ObjectPool* pool, const starrocks::TExprNode& texpr_node,
                                    starrocks::Expr** expr) {
    switch (texpr_node.node_type) {
    case TExprNodeType::BOOL_LITERAL:
    case TExprNodeType::INT_LITERAL:
    case TExprNodeType::LARGE_INT_LITERAL:
    case TExprNodeType::FLOAT_LITERAL:
    case TExprNodeType::DECIMAL_LITERAL:
    case TExprNodeType::DATE_LITERAL:
    case TExprNodeType::STRING_LITERAL:
    case TExprNodeType::NULL_LITERAL: {
        *expr = pool->add(new vectorized::VectorizedLiteral(texpr_node));
        break;
    }
    case TExprNodeType::COMPOUND_PRED: {
        *expr = pool->add(vectorized::VectorizedCompoundPredicateFactory::from_thrift(texpr_node));
        break;
    }
    case TExprNodeType::BINARY_PRED: {
        *expr = pool->add(vectorized::VectorizedBinaryPredicateFactory::from_thrift(texpr_node));
        break;
    }
    case TExprNodeType::ARITHMETIC_EXPR: {
        if (texpr_node.opcode != TExprOpcode::INVALID_OPCODE) {
            *expr = pool->add(vectorized::VectorizedArithmeticExprFactory::from_thrift(texpr_node));
            break;
        } else {
            // @TODO: will call FunctionExpr, implement later
            return Status::InternalError("Vectorized engine not support unknown OP arithmetic expr");
        }
    }
    case TExprNodeType::CAST_EXPR: {
        if (texpr_node.__isset.child_type || texpr_node.__isset.child_type_desc) {
            *expr = pool->add(vectorized::VectorizedCastExprFactory::from_thrift(texpr_node));
            break;
        } else {
            // @TODO: will call FunctionExpr, implement later
            return Status::InternalError("Vectorized engine not support unknown child type cast");
        }
    }
    case TExprNodeType::COMPUTE_FUNCTION_CALL:
    case TExprNodeType::FUNCTION_CALL: {
        if (texpr_node.fn.binary_type == TFunctionBinaryType::SRJAR) {
            *expr = pool->add(new vectorized::JavaFunctionCallExpr(texpr_node));
        } else if (texpr_node.fn.name.function_name == "if") {
            *expr = pool->add(vectorized::VectorizedConditionExprFactory::create_if_expr(texpr_node));
        } else if (texpr_node.fn.name.function_name == "nullif") {
            *expr = pool->add(vectorized::VectorizedConditionExprFactory::create_null_if_expr(texpr_node));
        } else if (texpr_node.fn.name.function_name == "ifnull") {
            *expr = pool->add(vectorized::VectorizedConditionExprFactory::create_if_null_expr(texpr_node));
        } else if (texpr_node.fn.name.function_name == "coalesce") {
            *expr = pool->add(vectorized::VectorizedConditionExprFactory::create_coalesce_expr(texpr_node));
        } else if (texpr_node.fn.name.function_name == "is_null_pred" ||
                   texpr_node.fn.name.function_name == "is_not_null_pred") {
            *expr = pool->add(vectorized::VectorizedIsNullPredicateFactory::from_thrift(texpr_node));
        } else {
            *expr = pool->add(new vectorized::VectorizedFunctionCallExpr(texpr_node));
        }
        break;
    }
    case TExprNodeType::IN_PRED: {
        *expr = pool->add(vectorized::VectorizedInPredicateFactory::from_thrift(texpr_node));
        break;
    }
    case TExprNodeType::SLOT_REF: {
        if (!texpr_node.__isset.slot_ref) {
            return Status::InternalError("Slot reference not set in thrift node");
        }
        *expr = pool->add(new vectorized::ColumnRef(texpr_node));
        break;
    }
    case TExprNodeType::CASE_EXPR: {
        if (!texpr_node.__isset.case_expr) {
            return Status::InternalError("Case expression not set in thrift node");
        }

        *expr = pool->add(vectorized::VectorizedCaseExprFactory::from_thrift(texpr_node));
        break;
    }
    case TExprNodeType::ARRAY_EXPR:
        *expr = pool->add(vectorized::ArrayExprFactory::from_thrift(texpr_node));
        break;
    case TExprNodeType::ARRAY_ELEMENT_EXPR:
        *expr = pool->add(vectorized::ArrayElementExprFactory::from_thrift(texpr_node));
        break;
    case TExprNodeType::INFO_FUNC:
        *expr = pool->add(new vectorized::VectorizedInfoFunc(texpr_node));
        break;
    case TExprNodeType::PLACEHOLDER_EXPR:
        *expr = pool->add(new vectorized::PlaceHolderRef(texpr_node));
        break;
    case TExprNodeType::DICT_EXPR:
        *expr = pool->add(new vectorized::DictMappingExpr(texpr_node));
        break;
    case TExprNodeType::CLONE_EXPR:
        *expr = pool->add(new vectorized::CloneExpr(texpr_node));
        break;
    case TExprNodeType::ARRAY_SLICE_EXPR:
    case TExprNodeType::AGG_EXPR:
    case TExprNodeType::TABLE_FUNCTION_EXPR:
    case TExprNodeType::IS_NULL_PRED:
    case TExprNodeType::LIKE_PRED:
    case TExprNodeType::LITERAL_PRED:
    case TExprNodeType::TUPLE_IS_NULL_PRED:
        break;
    }
    if (*expr == nullptr) {
        std::string err_msg =
                fmt::format("Vectorized engine does not support the operator, node_type: {}", texpr_node.node_type);
        LOG(WARNING) << err_msg;
        return Status::InternalError(err_msg);
    }

    return Status::OK();
}

struct MemLayoutData {
    int expr_idx;
    int byte_size;
    bool variable_length;

    // TODO: sort by type as well?  Any reason to do this?
    bool operator<(const MemLayoutData& rhs) const {
        // variable_len go at end
        if (this->variable_length && !rhs.variable_length) {
            return false;
        }

        if (!this->variable_length && rhs.variable_length) {
            return true;
        }

        return this->byte_size < rhs.byte_size;
    }
};

// Returns the byte size of 'type'  Returns 0 for variable length types.
inline static int get_byte_size_of_primitive_type(PrimitiveType type) {
    switch (type) {
    case TYPE_OBJECT:
    case TYPE_HLL:
    case TYPE_PERCENTILE:
    case TYPE_VARCHAR:
        return 0;

    case TYPE_NULL:
    case TYPE_BOOLEAN:
    case TYPE_TINYINT:
        return 1;

    case TYPE_SMALLINT:
        return 2;

    case TYPE_DECIMAL32:
    case TYPE_INT:
    case TYPE_FLOAT:
        return 4;

    case TYPE_DECIMAL64:
    case TYPE_BIGINT:
    case TYPE_TIME:
    case TYPE_DOUBLE:
        return 8;

    case TYPE_DECIMAL128:
    case TYPE_LARGEINT:
    case TYPE_DATETIME:
    case TYPE_DATE:
    case TYPE_DECIMALV2:
        return 16;

    case TYPE_DECIMAL:
        return 40;

    case INVALID_TYPE:
    default:
        DCHECK(false);
    }

    return 0;
}

Status Expr::prepare(const std::vector<ExprContext*>& ctxs, RuntimeState* state) {
    for (auto ctx : ctxs) {
        RETURN_IF_ERROR(ctx->prepare(state));
    }
    return Status::OK();
}

Status Expr::prepare(RuntimeState* state, ExprContext* context) {
    DCHECK(_type.type != INVALID_TYPE);
    for (auto& i : _children) {
        RETURN_IF_ERROR(i->prepare(state, context));
    }
    return Status::OK();
}

Status Expr::open(const std::vector<ExprContext*>& ctxs, RuntimeState* state) {
    for (auto ctx : ctxs) {
        RETURN_IF_ERROR(ctx->open(state));
    }
    return Status::OK();
}

Status Expr::open(RuntimeState* state, ExprContext* context, FunctionContext::FunctionStateScope scope) {
    DCHECK(_type.type != INVALID_TYPE);
    for (auto& i : _children) {
        RETURN_IF_ERROR(i->open(state, context, scope));
    }
    return Status::OK();
}

void Expr::close(const std::vector<ExprContext*>& ctxs, RuntimeState* state) {
    for (auto ctx : ctxs) {
        if (ctx != nullptr) {
            ctx->close(state);
        }
    }
}

void Expr::close(RuntimeState* state, ExprContext* context, FunctionContext::FunctionStateScope scope) {
    for (auto& i : _children) {
        i->close(state, context, scope);
    }
    // TODO(zc)
#if 0
    if (scope == FunctionContext::FRAGMENT_LOCAL) {
        // This is the final, non-cloned context to close. Clean up the whole Expr.
        if (cache_entry_ != NULL) {
            LibCache::instance()->DecrementUseCount(cache_entry_);
            cache_entry_ = NULL;
        }
    }
#endif
}

Status Expr::clone_if_not_exists(const std::vector<ExprContext*>& ctxs, RuntimeState* state,
                                 std::vector<ExprContext*>* new_ctxs) {
    DCHECK(new_ctxs != nullptr);
    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 Expr::debug_string() const {
    // TODO: implement partial debug string for member vars
    std::stringstream out;
    out << " type=" << _type.debug_string();

    if (_opcode != TExprOpcode::INVALID_OPCODE) {
        out << " opcode=" << _opcode;
    }
    out << " node-type=" << to_string(_node_type);
    out << " codegen=false";

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

    return out.str();
}

std::string Expr::debug_string(const std::vector<Expr*>& 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 Expr::debug_string(const std::vector<ExprContext*>& ctxs) {
    std::vector<Expr*> exprs;
    exprs.reserve(ctxs.size());
    for (auto ctx : ctxs) {
        exprs.push_back(ctx->root());
    }
    return debug_string(exprs);
}

bool Expr::is_constant() const {
    for (auto i : _children) {
        if (!i->is_constant()) {
            return false;
        }
    }

    return true;
}

TExprNodeType::type Expr::type_without_cast(const Expr* expr) {
    if (expr->_opcode == TExprOpcode::CAST) {
        return type_without_cast(expr->_children[0]);
    }
    return expr->_node_type;
}

const Expr* Expr::expr_without_cast(const Expr* expr) {
    if (expr->_opcode == TExprOpcode::CAST) {
        return expr_without_cast(expr->_children[0]);
    }
    return expr;
}

bool Expr::is_bound(const std::vector<TupleId>& tuple_ids) const {
    for (auto i : _children) {
        if (!i->is_bound(tuple_ids)) {
            return false;
        }
    }

    return true;
}

int Expr::get_slot_ids(std::vector<SlotId>* slot_ids) const {
    int n = 0;

    for (auto i : _children) {
        n += i->get_slot_ids(slot_ids);
    }

    return n;
}

Expr* Expr::copy(ObjectPool* pool, Expr* old_expr) {
    auto new_expr = old_expr->clone(pool);
    for (auto child : old_expr->_children) {
        auto new_child = copy(pool, child);
        new_expr->_children.push_back(new_child);
    }
    return new_expr;
}

// TODO chenhao
void Expr::close() {
    for (Expr* child : _children) child->close();
    /*if (_cache_entry != nullptr) {
      LibCache::instance()->decrement_use_count(_cache_entry);
      _cache_entry = nullptr;
      }*/
    _cache_entry.reset();
}

void Expr::close(const std::vector<Expr*>& exprs) {
    for (Expr* expr : exprs) expr->close();
}

StatusOr<ColumnPtr> Expr::evaluate_const(ExprContext* context) {
    if (!is_constant()) {
        return nullptr;
    }

    if (_constant_column.ok() && _constant_column.value()) {
        return _constant_column;
    }

    // prevent _constant_column from being assigned by multiple threads in pipeline engine.
    std::call_once(_constant_column_evaluate_once,
                   [this, context] { this->_constant_column = context->evaluate(this, nullptr); });
    return _constant_column;
}

ColumnPtr Expr::evaluate(ExprContext* context, vectorized::Chunk* ptr) {
    return nullptr;
}

vectorized::ColumnRef* Expr::get_column_ref() {
    if (this->is_slotref()) {
        return down_cast<vectorized::ColumnRef*>(this);
    }
    for (auto child : this->children()) {
        vectorized::ColumnRef* ref = nullptr;
        if ((ref = child->get_column_ref()) != nullptr) {
            return ref;
        }
    }
    return nullptr;
}

} // namespace starrocks

#pragma clang diagnostic pop