#include <optional>

#include "base_test.hpp"
#include "expression/arithmetic_expression.hpp"
#include "expression/binary_predicate_expression.hpp"
#include "expression/case_expression.hpp"
#include "expression/evaluation/expression_evaluator.hpp"
#include "expression/evaluation/expression_result.hpp"
#include "expression/exists_expression.hpp"
#include "expression/expression_functional.hpp"
#include "expression/expression_utils.hpp"
#include "expression/extract_expression.hpp"
#include "expression/function_expression.hpp"
#include "expression/in_expression.hpp"
#include "expression/list_expression.hpp"
#include "expression/pqp_column_expression.hpp"
#include "expression/pqp_subquery_expression.hpp"
#include "expression/value_expression.hpp"
#include "operators/get_table.hpp"
#include "operators/projection.hpp"
#include "operators/table_scan.hpp"
#include "operators/table_wrapper.hpp"
#include "operators/union_all.hpp"
#include "storage/table.hpp"
#include "utils/load_table.hpp"

namespace hyrise {

using namespace expression_functional;  // NOLINT(build/namespaces)

class ExpressionEvaluatorToValuesTest : public BaseTest {
 public:
  void SetUp() override {
    // Load table_a
    table_a = load_table("resources/test_data/tbl/expression_evaluator/input_a.tbl");
    a = PQPColumnExpression::from_table(*table_a, "a");
    b = PQPColumnExpression::from_table(*table_a, "b");
    c = PQPColumnExpression::from_table(*table_a, "c");
    d = PQPColumnExpression::from_table(*table_a, "d");
    e = PQPColumnExpression::from_table(*table_a, "e");
    f = PQPColumnExpression::from_table(*table_a, "f");
    s1 = PQPColumnExpression::from_table(*table_a, "s1");
    s2 = PQPColumnExpression::from_table(*table_a, "s2");
    s3 = PQPColumnExpression::from_table(*table_a, "s3");
    a_plus_b = std::make_shared<ArithmeticExpression>(ArithmeticOperator::Addition, a, b);
    a_plus_c = std::make_shared<ArithmeticExpression>(ArithmeticOperator::Addition, a, c);
    s1_gt_s2 = std::make_shared<BinaryPredicateExpression>(PredicateCondition::GreaterThan, s1, s2);
    s1_lt_s2 = std::make_shared<BinaryPredicateExpression>(PredicateCondition::LessThan, s1, s2);
    a_lt_b = std::make_shared<BinaryPredicateExpression>(PredicateCondition::LessThan, a, b);
    a_lt_c = std::make_shared<BinaryPredicateExpression>(PredicateCondition::LessThan, a, c);

    // Load table_b
    table_b = load_table("resources/test_data/tbl/expression_evaluator/input_b.tbl");
    x = PQPColumnExpression::from_table(*table_b, "x");

    // Load table_bools
    table_bools = load_table("resources/test_data/tbl/expression_evaluator/input_bools.tbl");
    bool_a = PQPColumnExpression::from_table(*table_bools, "a");
    bool_b = PQPColumnExpression::from_table(*table_bools, "b");
    bool_c = PQPColumnExpression::from_table(*table_bools, "c");

    // Load table_date_time
    table_date_time = load_table("resources/test_data/tbl/expression_evaluator/input_date_time.tbl");
    dates = PQPColumnExpression::from_table(*table_date_time, "dates");
    timestamps = PQPColumnExpression::from_table(*table_date_time, "timestamps");

    // Load table_mixed
    table_mixed = load_table("resources/test_data/tbl/expression_evaluator/input_mixed.tbl");
    mixed_a = PQPColumnExpression::from_table(*table_mixed, "a");
    mixed_b = PQPColumnExpression::from_table(*table_mixed, "b");
    mixed_c = PQPColumnExpression::from_table(*table_mixed, "c");
    mixed_d = PQPColumnExpression::from_table(*table_mixed, "d");
    mixed_e = PQPColumnExpression::from_table(*table_mixed, "e");
    mixed_f = PQPColumnExpression::from_table(*table_mixed, "f");

    // Create table_empty
    auto empty_table_columns = TableColumnDefinitions{};
    empty_table_columns.emplace_back("a", DataType::Int, false);
    empty_table_columns.emplace_back("b", DataType::Float, true);
    empty_table_columns.emplace_back("s", DataType::String, false);
    table_empty = std::make_shared<Table>(empty_table_columns, TableType::Data);

    auto segments = Segments{};
    segments.emplace_back(std::make_shared<ValueSegment<int32_t>>(pmr_vector<int32_t>{}));
    segments.emplace_back(std::make_shared<ValueSegment<float>>(pmr_vector<float>{}, pmr_vector<bool>{}));
    segments.emplace_back(std::make_shared<ValueSegment<pmr_string>>(pmr_vector<pmr_string>{}));
    table_empty->append_chunk(segments);

    empty_a = PQPColumnExpression::from_table(*table_empty, "a");
    empty_b = PQPColumnExpression::from_table(*table_empty, "b");
    empty_s = PQPColumnExpression::from_table(*table_empty, "s");
  }

  /**
   * Turn an ExpressionResult<T> into a canonical form "std::vector<std::optional<T>>" to make the writing of tests
   * easier.
   */
  template <typename T>
  std::vector<std::optional<T>> normalize_expression_result(const ExpressionResult<T>& result) {
    std::vector<std::optional<T>> normalized(result.size());

    result.as_view([&](const auto& resolved) {
      for (auto idx = size_t{0}; idx < result.size(); ++idx) {
        if (!resolved.is_null(idx)) {
          normalized[idx] = resolved.value(idx);
        }
      }
    });

    return normalized;
  }

  template <typename R>
  void print(const std::vector<std::optional<R>>& values_or_nulls) {
    for (const auto& value_or_null : values_or_nulls) {
      if (value_or_null) {
        std::cout << *value_or_null << ", ";
      } else {
        std::cout << "NULL, ";
      }
    }
  }

  template <typename R>
  bool test_expression(const std::shared_ptr<Table>& table, const AbstractExpression& expression,
                       const std::vector<std::optional<R>>& expected) {
    const auto actual_result = ExpressionEvaluator{table, ChunkID{0}}.evaluate_expression_to_result<R>(expression);
    const auto actual_normalized = normalize_expression_result(*actual_result);
    if (actual_normalized == expected) {
      return true;
    }

    std::cout << "Actual:\n  ";
    print(actual_normalized);
    std::cout << "\nExpected:\n  ";
    print(expected);
    std::cout << '\n';

    return false;
  }

  template <typename R>
  bool test_expression(const AbstractExpression& expression, const std::vector<std::optional<R>>& expected) {
    const auto actual_result = ExpressionEvaluator{}.evaluate_expression_to_result<R>(expression);
    const auto actual_normalized = normalize_expression_result(*actual_result);
    if (actual_normalized == expected) {
      return true;
    }

    std::cout << "Actual:\n  ";
    print(actual_normalized);
    std::cout << "\nExpected:\n  ";
    print(expected);
    std::cout << '\n';

    return false;
  }

  std::shared_ptr<Table> table_empty, table_a, table_b, table_bools, table_date_time, table_mixed;

  std::shared_ptr<PQPColumnExpression> a, b, c, d, e, f, s1, s2, s3, x, bool_a, bool_b, bool_c, dates, timestamps,
      mixed_a, mixed_b, mixed_c, mixed_d, mixed_e, mixed_f;
  std::shared_ptr<PQPColumnExpression> empty_a, empty_b, empty_s;
  std::shared_ptr<ArithmeticExpression> a_plus_b;
  std::shared_ptr<ArithmeticExpression> a_plus_c;
  std::shared_ptr<BinaryPredicateExpression> a_lt_b;
  std::shared_ptr<BinaryPredicateExpression> a_lt_c;
  std::shared_ptr<BinaryPredicateExpression> s1_gt_s2;
  std::shared_ptr<BinaryPredicateExpression> s1_lt_s2;
};

TEST_F(ExpressionEvaluatorToValuesTest, TernaryOrLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*or_(1, 0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*or_(1, 1), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*or_(0, 1), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*or_(0, 0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*or_(0, NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*or_(1, NullValue{}), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*or_(NullValue{}, NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*or_(NullValue{}, 0), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*or_(NullValue{}, 1), {1}));
}

TEST_F(ExpressionEvaluatorToValuesTest, TernaryOrSeries) {
  // clang-format off
  EXPECT_TRUE(test_expression<int32_t>(table_bools, *or_(bool_a, bool_b), {0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_bools, *or_(bool_a, bool_c), {0, 1, std::nullopt, 0, 1, std::nullopt, 1, 1, 1, 1, 1, 1}));  // NOLINT
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *or_(less_than_(1, empty_a), less_than_(1, empty_a)), {}));
  // clang-format on
}

TEST_F(ExpressionEvaluatorToValuesTest, TernaryAndLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*and_(1, 0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*and_(1, 1), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*and_(0, 1), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*and_(0, 0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*and_(0, NullValue{}), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*and_(1, NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*and_(NullValue{}, NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*and_(NullValue{}, 0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*and_(NullValue{}, 1), {std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, TernaryAndSeries) {
  // clang-format off
  EXPECT_TRUE(test_expression<int32_t>(table_bools, *and_(bool_a, bool_b), {0, 0, 0, 0, 0, 0, 0, 0, 0, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_bools, *and_(bool_a, bool_c), {0, 0, 0, 0, 0, 0, 0, 1, std::nullopt, 0, 1, std::nullopt}));  // NOLINT
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *and_(less_than_(1, empty_a), less_than_(1, empty_a)), {}));
  // clang-format on
}

TEST_F(ExpressionEvaluatorToValuesTest, ValueLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*value_(5), {5}));
  EXPECT_TRUE(test_expression<float>(*value_(5.0f), {5.0f}));
  EXPECT_TRUE(test_expression<int32_t>(*value_(NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<float>(*value_(NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(*value_("Hello"), {"Hello"}));
  EXPECT_TRUE(test_expression<pmr_string>(*value_(NullValue{}), {std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, ArithmeticsLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*mul_(5, 3), {15}));
  EXPECT_TRUE(test_expression<int32_t>(*mul_(5, NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*add_(5, 6), {11}));
  EXPECT_TRUE(test_expression<float>(*add_(5, 6), {11.0}));
  EXPECT_TRUE(test_expression<int32_t>(*sub_(15, 12), {3}));
  EXPECT_TRUE(test_expression<float>(*div_(10.0, 4.0), {2.5f}));
  EXPECT_TRUE(test_expression<float>(*div_(10.0, 0), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*div_(10, 0), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*sub_(NullValue{}, NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*mod_(5, 3), {2}));
  EXPECT_TRUE(test_expression<float>(*mod_(23.25, 3), {2.25}));
  EXPECT_TRUE(test_expression<float>(*mod_(23.25, 0), {std::nullopt}));
  EXPECT_TRUE(test_expression<float>(*mod_(5, 0), {std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, ArithmeticsSeries) {
  // clang-format off
  EXPECT_TRUE(test_expression<int32_t>(table_a, *mul_(a, b), {2, 6, 12, 20}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *mod_(b, a), {0, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *mod_(a, c), {1, std::nullopt, 3, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *add_(a, add_(b, c)), {36, std::nullopt, 41, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *add_(a, NullValue{}), {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));  // NOLINT
  EXPECT_TRUE(test_expression<int32_t>(table_a, *add_(a, add_(b, NullValue{})), {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));  // NOLINT
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *add_(empty_a, empty_b), {}));
  // clang-format on
}

TEST_F(ExpressionEvaluatorToValuesTest, ExpressionReuse) {
  // We can't really test that the reoccuring subexpressions are evaluated only once, but at least we have a test
  // where reoccuring subexpressions are used.
  // clang-format off
  EXPECT_TRUE(test_expression<int32_t>(table_a, *add_(mul_(a, b), mul_(a, b)), {4, 12, 24, 40}));
  // clang-format on
}

TEST_F(ExpressionEvaluatorToValuesTest, PredicatesLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_(5, 3.3), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_(5, 5.0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_(5.1, 5.0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_(null_(), 5.0), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_(5.0, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_(null_(), null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_("Hello", "Wello"), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_("Wello", "Hello"), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_("Wello", null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_equals_(5.3, 3), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_equals_(5.3, 5.3), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_equals_(5.3, 5.4f), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*greater_than_equals_(5.5f, 5.4), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*less_than_(5.2f, 5.4), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*less_than_(5.5f, 5.4), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*less_than_(5.4, 5.4), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*less_than_equals_(5.3, 5.4), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*less_than_equals_(5.4, 5.4), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*less_than_equals_(5.5, 5.4), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*equals_(5.5f, 5.5f), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*equals_(5.5f, 5.7f), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*equals_("Hello", "Hello"), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*equals_("Hello", "hello"), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*equals_("Hello", null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_equals_(5.5f, 5), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_equals_(5.5f, 5.5f), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(4, 3.0, 5.0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(3, 3.0, 5.0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(3, 3.1, 5.0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(5.0f, 3.1, 5), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(5.1f, 3.1, 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(5.1f, 3.1, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(5.1f, null_(), 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(null_(), 3.1, 5), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_inclusive_(null_(), null_(), null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(4, 3.0, 5.0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(3, 3.0, 5.0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(3, 3.1, 5.0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(5.0f, 3.1, 5), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(5.1f, 3.1, 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(3.1f, 3, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(5, null_(), 5.1f), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(3, 3, null_()), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(5.1f, null_(), 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(null_(), 3.1, 5), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_lower_exclusive_(null_(), null_(), null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(4, 3.0, 5.0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(3, 3.0, 5.0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(3, 3.1, 5.0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(5, 3.1, 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(5, 3.1, 5.1), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(3.1, 3, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(3, 3.1, null_()), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(5, null_(), 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(5, null_(), 5.1f), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(null_(), 3.1, 5), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_upper_exclusive_(null_(), null_(), null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(4, 3.0, 5.0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(3, 3, 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(5, 3, 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(5, 3, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(3, 3, null_()), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(5, null_(), 5), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(4, null_(), 5), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(null_(), 3, 5), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*between_exclusive_(null_(), null_(), null_()), {std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, PredicatesSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_a, *greater_than_(b, a), {1, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *greater_than_(s1, s2), {0, 0, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *greater_than_(b, null_()),
                                       {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *greater_than_(c, a), {1, std::nullopt, 1, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *greater_than_equals_(b, mul_(a, 2)), {1, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *equals_(b, mul_(a, 2)), {1, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_equals_(b, mul_(a, 2)), {0, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *less_than_(b, mul_(a, 2)), {0, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *less_than_equals_(b, mul_(a, 2)), {1, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *less_than_equals_(c, f), {1, std::nullopt, 0, std::nullopt}));

  // clang-format off
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_inclusive_(b, a, c), {1, std::nullopt, 1, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_inclusive_(e, a, f), {1, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_inclusive_(3.3, a, b), {0, 0, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_inclusive_(4, a, b), {0, 0, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_inclusive_(b, d, c), {1, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_inclusive_(b, a, d), {1, 1, 1, 1}));

  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_lower_exclusive_(b, a, c), {1, std::nullopt, 1, std::nullopt}));  // NOLINT
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_lower_exclusive_(b, add_(a, 1), c), {0, 0, 0, 0}));  // NOLINT
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_lower_exclusive_(a, a, b), {0, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_lower_exclusive_(a, sub_(a, 1), b), {1, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_lower_exclusive_(2, a, b), {1, 0, 0, 0}));

  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_upper_exclusive_(a, a, b), {1, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_upper_exclusive_(a, a, sub_(b, 1)), {0, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_upper_exclusive_(2, a, b), {0, 1, 0, 0}));

  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_exclusive_(2, a, b), {0, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_exclusive_(2.5, a, b), {0, 1, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_exclusive_(a, a, b), {0, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *between_exclusive_(add_(a, 0.5f), a, b), {1, 1, 1, 1}));
  // clang-format on
}

TEST_F(ExpressionEvaluatorToValuesTest, CaseLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*case_(1, 2, 1), {2}));
  EXPECT_TRUE(test_expression<int32_t>(*case_(1, NullValue{}, 1), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*case_(1, 2.3, 1), {2.3}));
  EXPECT_TRUE(test_expression<int32_t>(*case_(0, 2.3, 1), {1.0}));
  EXPECT_TRUE(test_expression<int32_t>(*case_(0, 2.3, NullValue{}), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*case_(0, 2, 1), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*case_(0, 2, case_(1, 5, 13)), {5}));
  EXPECT_TRUE(test_expression<int32_t>(*case_(NullValue{}, 42, add_(5, 3)), {8}));
  EXPECT_TRUE(test_expression<int32_t>(*case_(1, NullValue{}, 5), {std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, CaseSeries) {
  // clang-format off
  EXPECT_TRUE(test_expression<int32_t>(table_a, *case_(greater_than_(c, a), b, 1337), {2, 1337, 4, 1337}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *case_(greater_than_(c, 0), NullValue{}, c), {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));  // NOLINT
  EXPECT_TRUE(test_expression<int32_t>(table_a, *case_(1, c, a), {33, std::nullopt, 34, std::nullopt}));  // NOLINT
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *case_(greater_than_(empty_a, 3), 1, 2), {}));
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *case_(1, empty_a, empty_a), {}));
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *case_(greater_than_(empty_a, 3), empty_a, empty_a), {}));
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *case_(equals_(add_(NullValue{}, 1), 0), 1, 2), {2}));
  // clang-format on
}

TEST_F(ExpressionEvaluatorToValuesTest, IsNullLiteral) {
  EXPECT_TRUE(test_expression<int32_t>(*is_null_(0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*is_null_(1), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*is_null_(null_()), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*is_not_null_(0), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*is_not_null_(1), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*is_not_null_(null_()), {0}));
}

TEST_F(ExpressionEvaluatorToValuesTest, IsNullSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_a, *is_null_(add_(c, a)), {0, 1, 0, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *is_not_null_(add_(c, a)), {1, 0, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *is_not_null_(empty_a), {}));
}

TEST_F(ExpressionEvaluatorToValuesTest, NegateLiteral) {
  EXPECT_TRUE(test_expression<double>(*unary_minus_(2.5), {-2.5}));
  EXPECT_TRUE(test_expression<int32_t>(*unary_minus_(int32_t{-3}), {int32_t{3}}));
}

TEST_F(ExpressionEvaluatorToValuesTest, NegateSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_a, *unary_minus_(a), {-1, -2, -3, -4}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *unary_minus_(c), {-33, std::nullopt, -34, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *unary_minus_(unary_minus_(c)), {33, std::nullopt, 34, std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, LikeLiteral) {
  EXPECT_TRUE(test_expression<int32_t>(*like_("hello", "hello"), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*like_("hello", "Hello"), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*not_like_("hello", "Hello"), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*like_("hello", "h_ll%o"), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_like_("hello", "h_ll%o"), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*like_("hello", "H_ll_o"), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*not_like_("hello", "H_ll_o"), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*like_("hello", "%h%_l%o"), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_like_("hello", "%h%_l%o"), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*like_(null_(), "%h%_l%o"), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_like_(null_(), "%h%_l%o"), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*like_(null_(), null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_like_(null_(), null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*like_("hello", null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_like_("hello", null_()), {std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, LikeSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_a, *like_(s1, concat_(s1, "%")), {1, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *like_(s1, concat_(s1, "a")), {0, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_like_(s1, "%a%"), {0, 1, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *like_(s1, "%A%"), {0, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *like_(s1, "%H%e%_%l%"), {0, 1, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_like_(s1, "%H%e%_%l%"), {1, 0, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *like_(s3, "%a%"), {std::nullopt, 1, 0, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_like_(s3, "%a%"), {std::nullopt, 0, 1, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *like_(s1, "%a%"), {1, 0, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *like_("Same", s1), {0, 0, 0, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_like_("Same", s1), {1, 1, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *like_(empty_s, "hello"), {}));
  EXPECT_TRUE(test_expression<int32_t>(table_empty, *like_("hello", empty_s), {}));
}

TEST_F(ExpressionEvaluatorToValuesTest, SubstrLiterals) {
  /** Hyrise follows SQLite semantics for negative indices in SUBSTR */

  EXPECT_TRUE(test_expression<pmr_string>(*substr_("", 3, 4), {""}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", 4, 4), {"lo W"}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", -18, 4), {""}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", -12, 1), {""}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", -12, 2), {"H"}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", -12, 12), {"Hello World"}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", -5, 2), {"Wo"}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", -5, -2), {""}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", 4, 40), {"lo World"}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", 20, 1), {""}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", int64_t{4}, 4), {"lo W"}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", 4, int64_t{4}), {"lo W"}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_(null_(), 1, 2), {std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", null_(), 2), {std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(*substr_("Hello World", 2, null_()), {std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, SubstrSeries) {
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *substr_(s1, 2, 3), {"", "ell", "hat", "ame"}));
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *substr_(s3, 4, 1), {std::nullopt, "d", "l", std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *substr_(s1, a, b), {"a", "ell", "at", "e"}));
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *substr_(s3, 2, a), {std::nullopt, "bc", "yzl", std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *substr_("test", 2, c), {"est", std::nullopt, "est", std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(table_empty, *substr_(empty_s, 1, empty_a), {}));
}

TEST_F(ExpressionEvaluatorToValuesTest, ConcatLiterals) {
  EXPECT_TRUE(test_expression<pmr_string>(*concat_(null_(), "world"), {std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(*concat_("hello ", "world"), {"hello world"}));
  EXPECT_TRUE(test_expression<pmr_string>(*concat_("hello", " ", "world"), {"hello world"}));
  EXPECT_TRUE(test_expression<pmr_string>(*concat_("hello", " ", "world", " are you, ", "okay?"),
                                          {"hello world are you, okay?"}));
  EXPECT_TRUE(test_expression<pmr_string>(*concat_("hello", " ", null_(), " are you, ", "okay?"), {std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, ConcatSeries) {
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *concat_(s1, s2), {"ab", "HelloWorld", "whatup", "SameSame"}));
  EXPECT_TRUE(
      test_expression<pmr_string>(table_a, *concat_("yo", s1, s2), {"yoab", "yoHelloWorld", "yowhatup", "yoSameSame"}));
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *concat_(concat_("a", "b", "c"), s1, s2),
                                          {"abcab", "abcHelloWorld", "abcwhatup", "abcSameSame"}));
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *concat_("nope", s1, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *concat_(s1, s2, s3),
                                          {std::nullopt, "HelloWorldabcd", "whatupxyzlol", std::nullopt}));
  EXPECT_TRUE(test_expression<pmr_string>(table_empty, *concat_(empty_s, "hello"), {}));
}

TEST_F(ExpressionEvaluatorToValuesTest, AbsLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*abs_(1), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*abs_(0), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*abs_(-1), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*abs_(null_()), {std::nullopt}));

  EXPECT_TRUE(test_expression<int64_t>(*abs_(int64_t{100}), {100}));
  EXPECT_TRUE(test_expression<int64_t>(*abs_(int64_t{0}), {0}));
  EXPECT_TRUE(test_expression<int64_t>(*abs_(int64_t{-100}), {100}));
  EXPECT_TRUE(test_expression<int64_t>(*abs_(null_()), {std::nullopt}));

  EXPECT_TRUE(test_expression<float>(*abs_(1.23f), {1.23f}));
  EXPECT_TRUE(test_expression<float>(*abs_(0.0f), {0.0f}));
  EXPECT_TRUE(test_expression<float>(*abs_(-1.23f), {1.23f}));
  EXPECT_TRUE(test_expression<float>(*abs_(null_()), {std::nullopt}));

  EXPECT_TRUE(test_expression<double>(*abs_(123.123), {123.123}));
  EXPECT_TRUE(test_expression<double>(*abs_(0.0), {0.0}));
  EXPECT_TRUE(test_expression<double>(*abs_(-123.123), {123.123}));
  EXPECT_TRUE(test_expression<double>(*abs_(null_()), {std::nullopt}));

  EXPECT_THROW(test_expression<pmr_string>(*abs_("hello"), {}), std::logic_error);
}

TEST_F(ExpressionEvaluatorToValuesTest, AbsSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_mixed, *abs_(mixed_a), {1, 2, 3, 4}));
  EXPECT_TRUE(test_expression<int32_t>(table_mixed, *abs_(mixed_b), {1, 2, 3, std::nullopt}));
  EXPECT_TRUE(test_expression<int64_t>(table_mixed, *abs_(mixed_c), {33, std::nullopt, 1234, std::nullopt}));
  EXPECT_TRUE(test_expression<float>(table_mixed, *abs_(mixed_d), {1.23f, 2.34f, 123.4f, std::nullopt}));
  EXPECT_TRUE(test_expression<double>(table_mixed, *abs_(mixed_e), {1234.5, 2345.6, 3456.7, std::nullopt}));
  EXPECT_THROW(test_expression<pmr_string>(table_mixed, *abs_(mixed_f), {"1", "2", "3", "4"}), std::logic_error);
}

TEST_F(ExpressionEvaluatorToValuesTest, Parameter) {
  const auto a_id = ParameterID{0};
  const auto b_id = ParameterID{1};

  auto a_plus_5_times_b = mul_(add_(correlated_parameter_(a_id, a), 5), correlated_parameter_(b_id, b));

  expression_set_parameters(a_plus_5_times_b, {{a_id, 12}, {b_id, 2}});
  EXPECT_TRUE(test_expression<int32_t>(*a_plus_5_times_b, {34}));

  expression_set_parameters(a_plus_5_times_b, {{b_id, 4}});
  EXPECT_TRUE(test_expression<int32_t>(*a_plus_5_times_b, {68}));
}

TEST_F(ExpressionEvaluatorToValuesTest, InListLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*in_(null_(), list_(null_())), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(null_(), list_(null_(), 3)), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(null_(), list_()), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(null_(), list_(1, 2, 3, 4)), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(null_(), list_(null_(), 2, 3, 4)), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_(null_(), 5, null_())), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_()), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_(null_(), add_(2, 3), null_())), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_(null_(), 6, null_())), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_(1, 3)), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_(1.0, 3.0)), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_("Hello", 1.0, "You", 3.0)), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*in_("You", list_("Hello", 1.0, "You", 3.0)), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_(1.0, 5.0)), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_(1.0, add_(1.0, 3.0))), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*in_(5, list_(1.0, add_(2.0, 3.0))), {1}));
}

TEST_F(ExpressionEvaluatorToValuesTest, InListSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(a, list_(1.0, 3.0)), {1, 0, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(a, list_(null_(), 1.0, 3.0)), {1, std::nullopt, 1, std::nullopt}));
  EXPECT_TRUE(
      test_expression<int32_t>(table_a, *in_(sub_(mul_(a, 2), 2), list_(b, 6, null_(), 0)), {1, std::nullopt, 1, 1}));
}

TEST_F(ExpressionEvaluatorToValuesTest, InArbitraryExpression) {
  // We support `<expression_a> IN <expression_b>`, even though it looks weird, because <expression_b> might be a column
  // storing the pre-computed result a of subquery
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(a, div_(b, 2.0f)), {1, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(a, sub_(c, 31)), {0, std::nullopt, 1, std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, InSubqueryUncorrelated) {
  // PQP that returns the column "a"
  const auto table_wrapper_a = std::make_shared<TableWrapper>(table_a);
  const auto pqp_a =
      std::make_shared<Projection>(table_wrapper_a, expression_vector(PQPColumnExpression::from_table(*table_a, "a")));
  const auto subquery_a = pqp_subquery_(pqp_a, DataType::Int, false);

  // PQP that returns the column "c"
  const auto table_wrapper_b = std::make_shared<TableWrapper>(table_a);
  const auto pqp_b =
      std::make_shared<Projection>(table_wrapper_b, expression_vector(PQPColumnExpression::from_table(*table_a, "c")));
  const auto subquery_b = pqp_subquery_(pqp_b, DataType::Int, true);

  pqp_a->never_clear_output();
  pqp_b->never_clear_output();
  execute_all({table_wrapper_a, table_wrapper_b, pqp_a, pqp_b});

  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(6, subquery_a), {0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(a, subquery_a), {1, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(add_(a, 2), subquery_a), {1, 1, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(b, subquery_a), {1, 1, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(34, subquery_b), {1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(34.0, subquery_b), {1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(34.5, subquery_b), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_("hello", subquery_b), {0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(c, subquery_b), {1, std::nullopt, 1, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(6, subquery_a), {1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(a, subquery_a), {0, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(add_(a, 2), subquery_a), {0, 0, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(b, subquery_a), {0, 0, 0, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(34, subquery_b), {0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(34.0, subquery_b), {0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(34.5, subquery_b), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_("hello", subquery_b), {1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(c, subquery_b), {0, std::nullopt, 0, std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, InSubqueryUncorrelatedNotPrecalculated) {
  // Make sure the expression evaluator complains if an uncorrelated subquery was not executed before.
  const auto table_wrapper = std::make_shared<TableWrapper>(table_a);
  table_wrapper->execute();
  const auto table_scan = std::make_shared<TableScan>(table_wrapper, equals_(a, 3));
  table_scan->execute();
  const auto projection = std::make_shared<Projection>(table_scan, expression_vector(b));

  const auto subquery = pqp_subquery_(projection, DataType::Int, true);

  EXPECT_THROW(test_expression<int32_t>(table_a, *in_(3, subquery), {0}), std::logic_error);
  EXPECT_THROW(test_expression<int32_t>(table_a, *in_(4, subquery), {1}), std::logic_error);
  EXPECT_THROW(test_expression<int32_t>(table_a, *in_(4, subquery), {0}), std::logic_error);
}

TEST_F(ExpressionEvaluatorToValuesTest, InSubqueryCorrelated) {
  // PQP that returns the column "b" multiplied with the current value in "a".
  //
  // row   list returned from subquery
  //  0      (1, 2, 3, 4)
  //  1      (2, 4, 6, 8)
  //  2      (3, 6, 9, 12)
  //  3      (4, 8, 12, 16)
  const auto table_wrapper_a = std::make_shared<TableWrapper>(table_a);
  const auto mul_a = mul_(correlated_parameter_(ParameterID{0}, a), PQPColumnExpression::from_table(*table_a, "a"));
  const auto pqp_a = std::make_shared<Projection>(table_wrapper_a, expression_vector(mul_a));
  const auto subquery_a = pqp_subquery_(pqp_a, DataType::Int, false, std::make_pair(ParameterID{0}, ColumnID{0}));

  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(4, subquery_a), {1, 1, 0, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(6, subquery_a), {0, 1, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(16, subquery_a), {0, 0, 0, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(b, subquery_a), {1, 0, 0, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(null_(), subquery_a),
                                       {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(4, subquery_a), {0, 0, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(6, subquery_a), {1, 0, 0, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(16, subquery_a), {1, 1, 1, 0}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(b, subquery_a), {0, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(null_(), subquery_a),
                                       {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));

  // PQP that returns the column "c" added to the current value in "a".
  //
  // row   list returned from subquery
  //  0      (34, NULL, 35, NULL)
  //  1      (35, NULL, 36, NULL)
  //  2      (36, NULL, 37, NULL)
  //  3      (37, NULL, 38, NULL)
  const auto table_wrapper_b = std::make_shared<TableWrapper>(table_a);
  const auto add_b = add_(correlated_parameter_(ParameterID{0}, a), PQPColumnExpression::from_table(*table_a, "c"));
  const auto pqp_b = std::make_shared<Projection>(table_wrapper_b, expression_vector(add_b));
  const auto subquery_b = pqp_subquery_(pqp_b, DataType::Int, true, std::make_pair(ParameterID{0}, ColumnID{0}));

  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(34, subquery_b), {1, std::nullopt, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(35, subquery_b), {1, 1, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(null_(), subquery_b),
                                       {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(36, subquery_b), {std::nullopt, 1, 1, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(36.0, subquery_b), {std::nullopt, 1, 1, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(36.3, subquery_b),
                                       {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));
  EXPECT_TRUE(
      test_expression<int32_t>(table_a, *not_in_(34, subquery_b), {0, std::nullopt, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(35, subquery_b), {0, 0, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(null_(), subquery_b),
                                       {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(36, subquery_b), {std::nullopt, 0, 0, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(36.0, subquery_b), {std::nullopt, 0, 0, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(36.3, subquery_b),
                                       {std::nullopt, std::nullopt, std::nullopt, std::nullopt}));

  // The ExpressionEvaluator must ensure not to reuse correlated subqueries and create a copy for each row. Thus, no
  // subquery operator should have been executed.
  EXPECT_FALSE(pqp_a->executed());
  EXPECT_FALSE(pqp_b->executed());
}

TEST_F(ExpressionEvaluatorToValuesTest, CorrelatedSubqueryPrecalculated) {
  const auto table_wrapper = std::make_shared<TableWrapper>(table_a);
  const auto projection = std::make_shared<Projection>(table_wrapper, expression_vector(a));
  const auto subquery = pqp_subquery_(projection, DataType::Int, false, std::make_pair(ParameterID{0}, ColumnID{0}));

  // Operators for correlated subqueries must not be reused, and the ExpressionEvaluator creates a copy for each row.
  // Thus, the input PQP must not be executed before.
  table_wrapper->execute();
  if constexpr (HYRISE_DEBUG) {
    EXPECT_THROW(test_expression<int32_t>(table_a, *in_(4, subquery), {1, 1, 1, 1}), std::logic_error);
  }

  projection->execute();
  EXPECT_THROW(test_expression<int32_t>(table_a, *in_(4, subquery), {1, 1, 1, 1}), std::logic_error);
}

TEST_F(ExpressionEvaluatorToValuesTest, CorrelatedSubqueryDiamond) {
  const auto table_wrapper = std::make_shared<TableWrapper>(table_a);
  const auto projection = std::make_shared<Projection>(table_wrapper, expression_vector(a));
  const auto union_all = std::make_shared<UnionAll>(projection, projection);
  const auto subquery = pqp_subquery_(union_all, DataType::Int, false, std::make_pair(ParameterID{0}, ColumnID{0}));

  // While executing the subquery LQP, the ExpressionEvaluator must execute each operator once, also for diamonds.
  EXPECT_TRUE(test_expression<int32_t>(table_a, *in_(4, subquery), {1, 1, 1, 1}));
}

TEST_F(ExpressionEvaluatorToValuesTest, NotInListLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(null_(), list_(null_())), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(null_(), list_(null_(), 3)), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(null_(), list_()), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(null_(), list_(1, 2, 3, 4)), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(null_(), list_(null_(), 2, 3, 4)), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_(null_(), 5, null_())), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_()), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_(null_(), add_(2, 3), null_())), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_(null_(), 6, null_())), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_(1, 3)), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_(1.0, 3.0)), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_("Hello", 1.0, "You", 3.0)), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_("You", list_("Hello", 1.0, "You", 3.0)), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_(1.0, 5.0)), {0}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_(1.0, add_(1.0, 3.0))), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*not_in_(5, list_(1.0, add_(2.0, 3.0))), {0}));
}

TEST_F(ExpressionEvaluatorToValuesTest, NotInListSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(a, list_(1.0, 3.0)), {0, 1, 0, 1}));
  EXPECT_TRUE(
      test_expression<int32_t>(table_a, *not_in_(a, list_(null_(), 1.0, 3.0)), {0, std::nullopt, 0, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(sub_(mul_(a, 2), 2), list_(b, 6, null_(), 0)),
                                       {0, std::nullopt, 0, 0}));
}

TEST_F(ExpressionEvaluatorToValuesTest, NotInArbitraryExpression) {
  // We support `<expression_a> IN <expression_b>`, even though it looks weird, because <expression_b> might be a column
  // storing the pre-computed result of a subquery
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(a, div_(b, 2.0f)), {0, 1, 1, 1}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_in_(a, sub_(c, 31)), {1, std::nullopt, 0, std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, Exists) {
  /**
   * Test a co-related EXISTS query
   *
   * SELECT
   *    EXISTS (SELECT a + x, x FROM table_b WHERE a + x = 13)
   * FROM
   *    table_a;
   */
  const auto table_wrapper = std::make_shared<TableWrapper>(table_b);
  const auto parameter_a = correlated_parameter_(ParameterID{0}, x);
  const auto a_plus_x_projection =
      std::make_shared<Projection>(table_wrapper, expression_vector(add_(parameter_a, x), x));
  const auto a_plus_x_column = pqp_column_(ColumnID{0}, DataType::Int, false, "");
  const auto a_plus_x_eq_13_scan = std::make_shared<TableScan>(a_plus_x_projection, equals_(a_plus_x_column, 13));
  const auto pqp_subquery_expression =
      pqp_subquery_(a_plus_x_eq_13_scan, DataType::Int, false, std::make_pair(ParameterID{0}, ColumnID{0}));

  const auto exists_expression = exists_(pqp_subquery_expression);
  EXPECT_TRUE(test_expression<int32_t>(table_a, *exists_expression, {0, 0, 1, 1}));

  const auto not_exists_expression = not_exists_(pqp_subquery_expression);
  EXPECT_TRUE(test_expression<int32_t>(table_a, *not_exists_expression, {1, 1, 0, 0}));

  EXPECT_EQ(exists_expression->data_type(), ExpressionEvaluator::DataTypeBool);
  EXPECT_EQ(not_exists_expression->data_type(), ExpressionEvaluator::DataTypeBool);

  // Correlated subqueries cannot be reused and should be deep copied before execution.
  EXPECT_FALSE(a_plus_x_eq_13_scan->executed());
}

TEST_F(ExpressionEvaluatorToValuesTest, ExtractLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Year, "1992-09-30"), {1992}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Month, "1992-09-30"), {9}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Day, "1992-09-30"), {30}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Hour, "1992-09-30 01:02:03.5"), {1}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Minute, "1992-09-30 01:02:03.5"), {2}));
  EXPECT_TRUE(test_expression<double>(*extract_(DatetimeComponent::Second, "1992-09-30 01:02:03.5"), {3.5}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Year, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Month, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Day, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Hour, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(*extract_(DatetimeComponent::Minute, null_()), {std::nullopt}));
  EXPECT_TRUE(test_expression<double>(*extract_(DatetimeComponent::Second, null_()), {std::nullopt}));

  EXPECT_THROW(test_expression<pmr_string>(*extract_(DatetimeComponent::Hour, "1992-09-30"), {}), std::logic_error);
  EXPECT_THROW(test_expression<float>(*extract_(DatetimeComponent::Minute, "1992-09-30"), {}), std::logic_error);
  EXPECT_THROW(test_expression<int32_t>(*extract_(DatetimeComponent::Second, "1992-09-30"), {}), std::logic_error);
  EXPECT_THROW(test_expression<int32_t>(*extract_(DatetimeComponent::Year, "1999-13-34"), {}), std::logic_error);

  EXPECT_EQ(extract_(DatetimeComponent::Year, "1993-08-01")->data_type(), DataType::Int);
  EXPECT_EQ(extract_(DatetimeComponent::Month, "1993-08-01")->data_type(), DataType::Int);
  EXPECT_EQ(extract_(DatetimeComponent::Day, "1993-08-01")->data_type(), DataType::Int);
  EXPECT_EQ(extract_(DatetimeComponent::Hour, "1993-08-01")->data_type(), DataType::Int);
  EXPECT_EQ(extract_(DatetimeComponent::Minute, "1993-08-01")->data_type(), DataType::Int);
  EXPECT_EQ(extract_(DatetimeComponent::Second, "1993-08-01")->data_type(), DataType::Double);
}

TEST_F(ExpressionEvaluatorToValuesTest, ExtractSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Year, dates),
                                       {2017, 2014, 2011, 2010, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Month, dates),
                                       {12, 8, 9, 1, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Day, dates),
                                       {6, 5, 3, 2, std::nullopt, std::nullopt}));

  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Year, timestamps),
                                       {2017, 2014, 2011, 2010, 2010, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Month, timestamps),
                                       {12, 8, 9, 1, 1, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Day, timestamps),
                                       {6, 5, 3, 3, 2, std::nullopt}));

  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Hour, dates),
                                       {0, 0, 0, 0, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Minute, dates),
                                       {0, 0, 0, 0, std::nullopt, std::nullopt}));
  EXPECT_TRUE(test_expression<double>(table_date_time, *extract_(DatetimeComponent::Second, dates),
                                      {0, 0, 0, 0, std::nullopt, std::nullopt}));

  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Hour, timestamps),
                                       {1, 2, 3, 2, 0, std::nullopt}));
  EXPECT_TRUE(test_expression<int32_t>(table_date_time, *extract_(DatetimeComponent::Minute, timestamps),
                                       {2, 3, 4, 2, 0, std::nullopt}));
  EXPECT_TRUE(test_expression<double>(table_date_time, *extract_(DatetimeComponent::Second, timestamps),
                                      {3, 4, 5.0001, 1.5, 0, std::nullopt}));

  EXPECT_TRUE(test_expression<int32_t>(table_empty, *extract_(DatetimeComponent::Day, empty_s), {}));
  EXPECT_THROW(test_expression<int32_t>(table_a, *extract_(DatetimeComponent::Day, a), {}), std::logic_error);
  EXPECT_THROW(test_expression<int32_t>(table_a, *extract_(DatetimeComponent::Day, s1), {}), std::logic_error);
}

TEST_F(ExpressionEvaluatorToValuesTest, CastLiterals) {
  EXPECT_TRUE(test_expression<int32_t>(*cast_(5.5, DataType::Int), {5}));
  EXPECT_TRUE(test_expression<float>(*cast_(5.5, DataType::Float), {5.5f}));
  EXPECT_TRUE(test_expression<float>(*cast_(5, DataType::Float), {5.0f}));
  EXPECT_TRUE(test_expression<pmr_string>(*cast_(5.5, DataType::String), {"5.5"}));
  EXPECT_TRUE(test_expression<int32_t>(*cast_(null_(), DataType::Int), {std::nullopt}));

  // Ensure requested data type is cast data type
  EXPECT_THROW(test_expression<int32_t>(*cast_("1.2", DataType::Float), {}), std::logic_error);
  // Following SQL standard, CAST("Hello" AS INT) errors
  EXPECT_THROW(test_expression<int32_t>(*cast_("Hello", DataType::Int), {}), std::logic_error);
  EXPECT_THROW(test_expression<float>(*cast_("Hello", DataType::Float), {}), std::logic_error);
  EXPECT_THROW(test_expression<int32_t>(*cast_("1.2", DataType::Int), {}), std::logic_error);
  // Cast as Null is undefined
  EXPECT_THROW(test_expression<pmr_string>(*cast_("Hello", DataType::Null), {}), std::logic_error);
}

TEST_F(ExpressionEvaluatorToValuesTest, CastSeries) {
  EXPECT_TRUE(test_expression<int32_t>(table_a, *cast_(a, DataType::Int), {1, 2, 3, 4}));
  EXPECT_TRUE(test_expression<float>(table_a, *cast_(a, DataType::Float), {1.0f, 2.0f, 3.0f, 4.0f}));
  EXPECT_TRUE(test_expression<pmr_string>(table_a, *cast_(a, DataType::String), {"1", "2", "3", "4"}));
  EXPECT_TRUE(test_expression<int32_t>(table_a, *cast_(f, DataType::Int), {99, 2, 13, 15}));
  EXPECT_TRUE(
      test_expression<pmr_string>(table_a, *cast_(c, DataType::String), {"33", std::nullopt, "34", std::nullopt}));
}

TEST_F(ExpressionEvaluatorToValuesTest, CastNullableStrings) {
  auto table_string_nullable = load_table("resources/test_data/tbl/string_numbers_null.tbl");
  auto string_column = PQPColumnExpression::from_table(*table_string_nullable, "a");

  EXPECT_TRUE(test_expression<int32_t>(table_string_nullable, *cast_(string_column, DataType::Int),
                                       {12, std::nullopt, 1234, std::nullopt}));
}

}  // namespace hyrise
