expression.cpp

/**
 * @file expression.cpp
 * @brief Expression evaluator implementation
 */

#include "parser/expression.hpp"

#include <cstddef>
#include <memory>
#include <string>
#include <utility>
#include <vector>

#include "common/value.hpp"
#include "executor/types.hpp"
#include "parser/token.hpp"

namespace cloudsql::parser {

/**
 * @brief Evaluate binary expression
 */
common::Value BinaryExpr::evaluate(const executor::Tuple* tuple,
                                   const executor::Schema* schema) const {
    const common::Value left_val = left_->evaluate(tuple, schema);
    const common::Value right_val = right_->evaluate(tuple, schema);

    switch (op_) {
        case TokenType::Plus:
            if (left_val.type() == common::ValueType::TYPE_FLOAT64 ||
                right_val.type() == common::ValueType::TYPE_FLOAT64) {
                return common::Value::make_float64(left_val.to_float64() + right_val.to_float64());
            }
            return common::Value::make_int64(left_val.to_int64() + right_val.to_int64());
        case TokenType::Minus:
            if (left_val.type() == common::ValueType::TYPE_FLOAT64 ||
                right_val.type() == common::ValueType::TYPE_FLOAT64) {
                return common::Value::make_float64(left_val.to_float64() - right_val.to_float64());
            }
            return common::Value::make_int64(left_val.to_int64() - right_val.to_int64());
        case TokenType::Star:
            if (left_val.type() == common::ValueType::TYPE_FLOAT64 ||
                right_val.type() == common::ValueType::TYPE_FLOAT64) {
                return common::Value::make_float64(left_val.to_float64() * right_val.to_float64());
            }
            return common::Value::make_int64(left_val.to_int64() * right_val.to_int64());
        case TokenType::Slash:
            return common::Value::make_float64(left_val.to_float64() / right_val.to_float64());
        case TokenType::Eq:
            return common::Value(left_val == right_val);
        case TokenType::Ne:
            return common::Value(left_val != right_val);
        case TokenType::Lt:
            return common::Value(left_val < right_val);
        case TokenType::Le:
            return common::Value(left_val <= right_val);
        case TokenType::Gt:
            return common::Value(left_val > right_val);
        case TokenType::Ge:
            return common::Value(left_val >= right_val);
        case TokenType::And:
            return common::Value(left_val.as_bool() && right_val.as_bool());
        case TokenType::Or:
            return common::Value(left_val.as_bool() || right_val.as_bool());
        default:
            return common::Value::make_null();
    }
}

void BinaryExpr::evaluate_vectorized(const executor::VectorBatch& batch,
                                     const executor::Schema& schema,
                                     executor::ColumnVector& result) const {
    const size_t row_count = batch.row_count();
    result.clear();

    // Try to optimize: Column op Constant
    if (left_->type() == ExprType::Column && right_->type() == ExprType::Constant) {
        const auto& col_expr = static_cast<const ColumnExpr&>(*left_);
        const auto& const_expr = static_cast<const ConstantExpr&>(*right_);
        const size_t col_idx = schema.find_column(col_expr.name());

        if (col_idx != static_cast<size_t>(-1)) {
            auto& src_col = const_cast<executor::VectorBatch&>(batch).get_column(col_idx);

            // INT64 optimize
            if (src_col.type() == common::ValueType::TYPE_INT64 &&
                const_expr.value().type() == common::ValueType::TYPE_INT64) {
                auto& num_src = dynamic_cast<executor::NumericVector<int64_t>&>(src_col);
                const int64_t* src_data = num_src.raw_data();
                const int64_t const_val = const_expr.value().as_int64();

                if (op_ == TokenType::Gt) {
                    auto& bool_res = dynamic_cast<executor::NumericVector<bool>&>(result);
                    bool_res.resize(row_count);
                    uint8_t* res_data = bool_res.raw_data_mut();
                    for (size_t i = 0; i < row_count; ++i) {
                        if (num_src.is_null(i)) {
                            bool_res.set_null(i, true);
                        } else {
                            res_data[i] = static_cast<uint8_t>(src_data[i] > const_val);
                            bool_res.set_null(i, false);
                        }
                    }
                    return;
                }
                if (op_ == TokenType::Eq) {
                    auto& bool_res = dynamic_cast<executor::NumericVector<bool>&>(result);
                    bool_res.resize(row_count);
                    uint8_t* res_data = bool_res.raw_data_mut();
                    for (size_t i = 0; i < row_count; ++i) {
                        if (num_src.is_null(i)) {
                            bool_res.set_null(i, true);
                        } else {
                            res_data[i] = static_cast<uint8_t>(src_data[i] == const_val);
                            bool_res.set_null(i, false);
                        }
                    }
                    return;
                }
            }
        }
    }

    // Fallback to row-by-row if not optimized
    for (size_t i = 0; i < row_count; ++i) {
        std::vector<common::Value> row_vals;
        for (size_t c = 0; c < batch.column_count(); ++c) {
            row_vals.push_back(const_cast<executor::VectorBatch&>(batch).get_column(c).get(i));
        }
        executor::Tuple t(std::move(row_vals));
        result.append(evaluate(&t, &schema));
    }
}

std::string BinaryExpr::to_string() const {
    std::string op_str;
    switch (op_) {
        case TokenType::Plus:
            op_str = " + ";
            break;
        case TokenType::Minus:
            op_str = " - ";
            break;
        case TokenType::Star:
            op_str = " * ";
            break;
        case TokenType::Slash:
            op_str = " / ";
            break;
        case TokenType::Eq:
            op_str = " = ";
            break;
        case TokenType::Ne:
            op_str = " <> ";
            break;
        case TokenType::Lt:
            op_str = " < ";
            break;
        case TokenType::Le:
            op_str = " <= ";
            break;
        case TokenType::Gt:
            op_str = " > ";
            break;
        case TokenType::Ge:
            op_str = " >= ";
            break;
        case TokenType::And:
            op_str = " AND ";
            break;
        case TokenType::Or:
            op_str = " OR ";
            break;
        default:
            op_str = " ";
            break;
    }
    return left_->to_string() + op_str + right_->to_string();
}

std::unique_ptr<Expression> BinaryExpr::clone() const {
    return std::make_unique<BinaryExpr>(left_->clone(), op_, right_->clone());
}

/**
 * @brief Evaluate unary expression
 */
common::Value UnaryExpr::evaluate(const executor::Tuple* tuple,
                                  const executor::Schema* schema) const {
    const common::Value val = expr_->evaluate(tuple, schema);
    switch (op_) {
        case TokenType::Minus:
            if (val.is_numeric()) {
                return common::Value(-val.to_float64());
            }
            break;
        case TokenType::Not:
            return common::Value(!val.as_bool());
        default:
            break;
    }
    return common::Value::make_null();
}

void UnaryExpr::evaluate_vectorized(const executor::VectorBatch& batch,
                                    const executor::Schema& schema,
                                    executor::ColumnVector& result) const {
    const size_t row_count = batch.row_count();
    result.clear();
    for (size_t i = 0; i < row_count; ++i) {
        std::vector<common::Value> row_vals;
        for (size_t c = 0; c < batch.column_count(); ++c) {
            row_vals.push_back(const_cast<executor::VectorBatch&>(batch).get_column(c).get(i));
        }
        executor::Tuple t(std::move(row_vals));
        result.append(evaluate(&t, &schema));
    }
}

std::string UnaryExpr::to_string() const {
    return (op_ == TokenType::Minus ? "-" : "NOT ") + expr_->to_string();
}

std::unique_ptr<Expression> UnaryExpr::clone() const {
    return std::make_unique<UnaryExpr>(op_, expr_->clone());
}

/**
 * @brief Evaluate column expression using tuple and schema
 */
common::Value ColumnExpr::evaluate(const executor::Tuple* tuple,
                                   const executor::Schema* schema) const {
    if (tuple == nullptr || schema == nullptr) {
        return common::Value::make_null();
    }

    size_t index = static_cast<size_t>(-1);

    /* 1. Try exact match (either fully qualified or just name) */
    index = schema->find_column(this->to_string());

    /* 2. If not found and it's qualified, try just the column name */
    if (index == static_cast<size_t>(-1) && has_table()) {
        index = schema->find_column(name_);
    }

    if (index == static_cast<size_t>(-1)) {
        return common::Value::make_null();
    }

    return tuple->get(index);
}

void ColumnExpr::evaluate_vectorized(const executor::VectorBatch& batch,
                                     const executor::Schema& schema,
                                     executor::ColumnVector& result) const {
    size_t index = static_cast<size_t>(-1);

    /* 1. Try exact match (either fully qualified or just name) */
    index = schema.find_column(this->to_string());

    /* 2. If not found and it's qualified, try just the column name */
    if (index == static_cast<size_t>(-1) && has_table()) {
        index = schema.find_column(name_);
    }

    result.clear();
    if (index == static_cast<size_t>(-1)) {
        for (size_t i = 0; i < batch.row_count(); ++i) {
            result.append(common::Value::make_null());
        }
        return;
    }

    auto& src_col = const_cast<executor::VectorBatch&>(batch).get_column(index);
    for (size_t i = 0; i < batch.row_count(); ++i) {
        result.append(src_col.get(i));
    }
}

std::string ColumnExpr::to_string() const {
    return has_table() ? table_name_ + "." + name_ : name_;
}

std::unique_ptr<Expression> ColumnExpr::clone() const {
    return has_table() ? std::make_unique<ColumnExpr>(table_name_, name_)
                       : std::make_unique<ColumnExpr>(name_);
}

common::Value ConstantExpr::evaluate(const executor::Tuple* tuple,
                                     const executor::Schema* schema) const {
    (void)tuple;
    (void)schema;
    return value_;
}

void ConstantExpr::evaluate_vectorized(const executor::VectorBatch& batch,
                                       const executor::Schema& schema,
                                       executor::ColumnVector& result) const {
    (void)schema;
    result.clear();
    for (size_t i = 0; i < batch.row_count(); ++i) {
        result.append(value_);
    }
}

std::string ConstantExpr::to_string() const {
    if (value_.type() == common::ValueType::TYPE_TEXT) {
        return "'" + value_.to_string() + "'";
    }
    return value_.to_string();
}

std::unique_ptr<Expression> ConstantExpr::clone() const {
    return std::make_unique<ConstantExpr>(value_);
}

/**
 * @brief Evaluate function expression
 */
common::Value FunctionExpr::evaluate(const executor::Tuple* tuple,
                                     const executor::Schema* schema) const {
    if (tuple == nullptr || schema == nullptr) {
        return common::Value::make_null();
    }

    /* Attempt to look up the function result in the schema (e.g. for aggregates) */
    const size_t index = schema->find_column(this->to_string());
    if (index != static_cast<size_t>(-1)) {
        return tuple->get(index);
    }

    return common::Value::make_null();
}

void FunctionExpr::evaluate_vectorized(const executor::VectorBatch& batch,
                                       const executor::Schema& schema,
                                       executor::ColumnVector& result) const {
    const size_t row_count = batch.row_count();
    result.clear();
    for (size_t i = 0; i < row_count; ++i) {
        std::vector<common::Value> row_vals;
        for (size_t c = 0; c < batch.column_count(); ++c) {
            row_vals.push_back(const_cast<executor::VectorBatch&>(batch).get_column(c).get(i));
        }
        executor::Tuple t(std::move(row_vals));
        result.append(evaluate(&t, &schema));
    }
}

std::string FunctionExpr::to_string() const {
    std::string result = func_name_ + "(";
    if (distinct_) {
        result += "DISTINCT ";
    }
    bool first = true;
    for (const auto& arg : args_) {
        if (!first) {
            result += ", ";
        }
        result += arg->to_string();
        first = false;
    }
    if (args_.empty() && func_name_ == "COUNT") {
        result += "*";
    }
    result += ")";
    return result;
}

std::unique_ptr<Expression> FunctionExpr::clone() const {
    auto result = std::make_unique<FunctionExpr>(func_name_);
    result->set_distinct(distinct_);
    for (const auto& arg : args_) {
        result->add_arg(arg->clone());
    }
    return result;
}

/**
 * @brief Evaluate IN expression
 */
common::Value InExpr::evaluate(const executor::Tuple* tuple, const executor::Schema* schema) const {
    const common::Value col_val = column_->evaluate(tuple, schema);
    for (const auto& val : values_) {
        if (col_val == val->evaluate(tuple, schema)) {
            return common::Value(!not_flag_);
        }
    }
    return common::Value(not_flag_);
}

void InExpr::evaluate_vectorized(const executor::VectorBatch& batch, const executor::Schema& schema,
                                 executor::ColumnVector& result) const {
    const size_t row_count = batch.row_count();
    result.clear();
    for (size_t i = 0; i < row_count; ++i) {
        std::vector<common::Value> row_vals;
        for (size_t c = 0; c < batch.column_count(); ++c) {
            row_vals.push_back(const_cast<executor::VectorBatch&>(batch).get_column(c).get(i));
        }
        executor::Tuple t(std::move(row_vals));
        result.append(evaluate(&t, &schema));
    }
}

std::string InExpr::to_string() const {
    std::string result = column_->to_string() + (not_flag_ ? " NOT IN (" : " IN (");
    bool first = true;
    for (const auto& val : values_) {
        if (!first) {
            result += ", ";
        }
        result += val->to_string();
        first = false;
    }
    result += ")";
    return result;
}

std::unique_ptr<Expression> InExpr::clone() const {
    std::vector<std::unique_ptr<Expression>> cloned_vals;
    cloned_vals.reserve(values_.size());
    for (const auto& val : values_) {
        cloned_vals.push_back(val->clone());
    }
    return std::make_unique<InExpr>(column_->clone(), std::move(cloned_vals), not_flag_);
}

/**
 * @brief Evaluate IS NULL expression
 */
common::Value IsNullExpr::evaluate(const executor::Tuple* tuple,
                                   const executor::Schema* schema) const {
    const common::Value val = expr_->evaluate(tuple, schema);
    const bool result = val.is_null();
    return common::Value(not_flag_ ? !result : result);
}

void IsNullExpr::evaluate_vectorized(const executor::VectorBatch& batch,
                                     const executor::Schema& schema,
                                     executor::ColumnVector& result) const {
    const size_t row_count = batch.row_count();
    result.clear();
    for (size_t i = 0; i < row_count; ++i) {
        std::vector<common::Value> row_vals;
        for (size_t c = 0; c < batch.column_count(); ++c) {
            row_vals.push_back(const_cast<executor::VectorBatch&>(batch).get_column(c).get(i));
        }
        executor::Tuple t(std::move(row_vals));
        result.append(evaluate(&t, &schema));
    }
}

std::string IsNullExpr::to_string() const {
    return expr_->to_string() + (not_flag_ ? " IS NOT NULL" : " IS NULL");
}

std::unique_ptr<Expression> IsNullExpr::clone() const {
    return std::make_unique<IsNullExpr>(expr_->clone(), not_flag_);
}

}  // namespace cloudsql::parser