text: /*
   * Copyright 2017 MapD Technologies, Inc.
   *
   * Licensed 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 "RelAlgTranslator.h"
  #include "Shared/SqlTypesLayout.h"
  
  #include "CalciteDeserializerUtils.h"
  #include "DateTimePlusRewrite.h"
  #include "DateTimeTranslator.h"
  #include "Descriptors/RelAlgExecutionDescriptor.h"
  #include "ExpressionRewrite.h"
  #include "ExtensionFunctionsBinding.h"
  #include "ExtensionFunctionsWhitelist.h"
  #include "RelAlgAbstractInterpreter.h"
  #include "WindowContext.h"
  
  #include <future>
  #include "../Analyzer/Analyzer.h"
  #include "../Parser/ParserNode.h"
  #include "../Shared/likely.h"
  #include "../Shared/sql_type_to_string.h"
  #include "../Shared/thread_count.h"
  
  extern bool g_enable_watchdog;
  
  bool g_enable_experimental_string_functions = false;
  
  namespace {
  
  SQLTypeInfo build_type_info(const SQLTypes sql_type,
                              const int scale,
                              const int precision) {
    SQLTypeInfo ti(sql_type, 0, 0, true);
    ti.set_scale(scale);
    ti.set_precision(precision);
    return ti;
  }
  
  std::pair<std::shared_ptr<Analyzer::Expr>, SQLQualifier> get_quantified_rhs(
      const RexScalar* rex_scalar,
      const RelAlgTranslator& translator) {
    std::shared_ptr<Analyzer::Expr> rhs;
    SQLQualifier sql_qual{kONE};
    const auto rex_operator = dynamic_cast<const RexOperator*>(rex_scalar);
    if (!rex_operator) {
      return std::make_pair(rhs, sql_qual);
    }
    const auto rex_function = dynamic_cast<const RexFunctionOperator*>(rex_operator);
    const auto qual_str = rex_function ? rex_function->getName() : "";
    if (qual_str == "PG_ANY"sv || qual_str == "PG_ALL"sv) {
      CHECK_EQ(size_t(1), rex_function->size());
      rhs = translator.translateScalarRex(rex_function->getOperand(0));
      sql_qual = (qual_str == "PG_ANY"sv) ? kANY : kALL;
    }
    if (!rhs && rex_operator->getOperator() == kCAST) {
      CHECK_EQ(size_t(1), rex_operator->size());
      std::tie(rhs, sql_qual) = get_quantified_rhs(rex_operator->getOperand(0), translator);
    }
    return std::make_pair(rhs, sql_qual);
  }
  
  std::pair<Datum, bool> datum_from_scalar_tv(const ScalarTargetValue* scalar_tv,
                                              const SQLTypeInfo& ti) noexcept {
    Datum d{0};
    bool is_null_const{false};
    switch (ti.get_type()) {
      case kTINYINT: {
        const auto ival = boost::get<int64_t>(scalar_tv);
        CHECK(ival);
        if (*ival == inline_int_null_val(ti)) {
          is_null_const = true;
        } else {
          d.tinyintval = *ival;
        }
        break;
      }
      case kSMALLINT: {
        const auto ival = boost::get<int64_t>(scalar_tv);
        CHECK(ival);
        if (*ival == inline_int_null_val(ti)) {
          is_null_const = true;
        } else {
          d.smallintval = *ival;
        }
        break;
      }
      case kINT: {
        const auto ival = boost::get<int64_t>(scalar_tv);
        CHECK(ival);
        if (*ival == inline_int_null_val(ti)) {
          is_null_const = true;
        } else {
          d.intval = *ival;
        }
        break;
      }
      case kDECIMAL:
      case kNUMERIC:
      case kBIGINT:
      case kDATE:
      case kTIME:
      case kTIMESTAMP: {
        const auto ival = boost::get<int64_t>(scalar_tv);
        CHECK(ival);
        if (*ival == inline_int_null_val(ti)) {
          is_null_const = true;
        } else {
          d.bigintval = *ival;
        }
        break;
      }
      case kDOUBLE: {
        const auto dval = boost::get<double>(scalar_tv);
        CHECK(dval);
        if (*dval == inline_fp_null_val(ti)) {
          is_null_const = true;
        } else {
          d.doubleval = *dval;
        }
        break;
      }
      case kFLOAT: {
        const auto fval = boost::get<float>(scalar_tv);
        CHECK(fval);
        if (*fval == inline_fp_null_val(ti)) {
          is_null_const = true;
        } else {
          d.floatval = *fval;
        }
        break;
      }
      case kTEXT:
      case kVARCHAR:
      case kCHAR: {
        auto nullable_sptr = boost::get<NullableString>(scalar_tv);
        CHECK(nullable_sptr);
        if (boost::get<void*>(nullable_sptr)) {
          is_null_const = true;
        } else {
          auto sptr = boost::get<std::string>(nullable_sptr);
          d.stringval = new std::string(*sptr);
        }
        break;
      }
      default:
        CHECK(false);
    }
    return {d, is_null_const};
  }
  
  }  // namespace
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateScalarRex(
      const RexScalar* rex) const {
    const auto rex_input = dynamic_cast<const RexInput*>(rex);
    if (rex_input) {
      return translateInput(rex_input);
    }
    const auto rex_literal = dynamic_cast<const RexLiteral*>(rex);
    if (rex_literal) {
      return translateLiteral(rex_literal);
    }
    const auto rex_window_function = dynamic_cast<const RexWindowFunctionOperator*>(rex);
    if (rex_window_function) {
      return translateWindowFunction(rex_window_function);
    }
    const auto rex_function = dynamic_cast<const RexFunctionOperator*>(rex);
    if (rex_function) {
      return translateFunction(rex_function);
    }
    const auto rex_operator = dynamic_cast<const RexOperator*>(rex);
    if (rex_operator) {
      return translateOper(rex_operator);
    }
    const auto rex_case = dynamic_cast<const RexCase*>(rex);
    if (rex_case) {
      return translateCase(rex_case);
    }
    const auto rex_subquery = dynamic_cast<const RexSubQuery*>(rex);
    if (rex_subquery) {
      return translateScalarSubquery(rex_subquery);
    }
    CHECK(false);
    return nullptr;
  }
  
  namespace {
  
  bool is_agg_supported_for_type(const SQLAgg& agg_kind, const SQLTypeInfo& arg_ti) {
    if ((agg_kind == kMIN || agg_kind == kMAX || agg_kind == kSUM || agg_kind == kAVG) &&
        !(arg_ti.is_number() || arg_ti.is_boolean() || arg_ti.is_time())) {
      return false;
    }
  
    return true;
  }
  
  }  // namespace
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateAggregateRex(
      const RexAgg* rex,
      const std::vector<std::shared_ptr<Analyzer::Expr>>& scalar_sources) {
    const auto agg_kind = rex->getKind();
    const bool is_distinct = rex->isDistinct();
    const bool takes_arg{rex->size() > 0};
    std::shared_ptr<Analyzer::Expr> arg_expr;
    std::shared_ptr<Analyzer::Constant> err_rate;
    if (takes_arg) {
      const auto operand = rex->getOperand(0);
      CHECK_LT(operand, scalar_sources.size());
      CHECK_LE(rex->size(), 2u);
      arg_expr = scalar_sources[operand];
      if (agg_kind == kAPPROX_COUNT_DISTINCT && rex->size() == 2) {
        err_rate = std::dynamic_pointer_cast<Analyzer::Constant>(
            scalar_sources[rex->getOperand(1)]);
        if (!err_rate || err_rate->get_type_info().get_type() != kINT ||
            err_rate->get_constval().intval < 1 || err_rate->get_constval().intval > 100) {
          throw std::runtime_error(
              "APPROX_COUNT_DISTINCT's second parameter should be SMALLINT literal between "
              "1 and 100");
        }
      }
      const auto& arg_ti = arg_expr->get_type_info();
      if (!is_agg_supported_for_type(agg_kind, arg_ti)) {
        throw std::runtime_error("Aggregate on " + arg_ti.get_type_name() +
                                 " is not supported yet.");
      }
    }
    const auto agg_ti = get_agg_type(agg_kind, arg_expr.get());
    return makeExpr<Analyzer::AggExpr>(agg_ti, agg_kind, arg_expr, is_distinct, err_rate);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateLiteral(
      const RexLiteral* rex_literal) {
    const auto lit_ti = build_type_info(
        rex_literal->getType(), rex_literal->getScale(), rex_literal->getPrecision());
    const auto target_ti = build_type_info(rex_literal->getTargetType(),
                                           rex_literal->getTypeScale(),
                                           rex_literal->getTypePrecision());
    switch (rex_literal->getType()) {
      case kDECIMAL: {
        const auto val = rex_literal->getVal<int64_t>();
        const int precision = rex_literal->getPrecision();
        const int scale = rex_literal->getScale();
        if (target_ti.is_fp() && !scale) {
          return make_fp_constant(val, target_ti);
        }
        auto lit_expr = scale ? Parser::FixedPtLiteral::analyzeValue(val, scale, precision)
                              : Parser::IntLiteral::analyzeValue(val);
        return lit_ti != target_ti ? lit_expr->add_cast(target_ti) : lit_expr;
      }
      case kTEXT: {
        return Parser::StringLiteral::analyzeValue(rex_literal->getVal<std::string>());
      }
      case kBOOLEAN: {
        Datum d;
        d.boolval = rex_literal->getVal<bool>();
        return makeExpr<Analyzer::Constant>(kBOOLEAN, false, d);
      }
      case kDOUBLE: {
        Datum d;
        d.doubleval = rex_literal->getVal<double>();
        auto lit_expr = makeExpr<Analyzer::Constant>(kDOUBLE, false, d);
        return lit_ti != target_ti ? lit_expr->add_cast(target_ti) : lit_expr;
      }
      case kINTERVAL_DAY_TIME:
      case kINTERVAL_YEAR_MONTH: {
        Datum d;
        d.bigintval = rex_literal->getVal<int64_t>();
        return makeExpr<Analyzer::Constant>(rex_literal->getType(), false, d);
      }
      case kTIME:
      case kTIMESTAMP: {
        Datum d;
        d.bigintval =
            rex_literal->getType() == kTIMESTAMP && rex_literal->getPrecision() > 0
                ? rex_literal->getVal<int64_t>()
                : rex_literal->getVal<int64_t>() / 1000;
        return makeExpr<Analyzer::Constant>(
            SQLTypeInfo(rex_literal->getType(), rex_literal->getPrecision(), 0, false),
            false,
            d);
      }
      case kDATE: {
        Datum d;
        d.bigintval = rex_literal->getVal<int64_t>() * 24 * 3600;
        return makeExpr<Analyzer::Constant>(rex_literal->getType(), false, d);
      }
      case kNULLT: {
        return makeExpr<Analyzer::Constant>(rex_literal->getTargetType(), true, Datum{0});
      }
      default: {
        LOG(FATAL) << "Unexpected literal type " << lit_ti.get_type_name();
      }
    }
    return nullptr;
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateScalarSubquery(
      const RexSubQuery* rex_subquery) const {
    if (just_explain_) {
      throw std::runtime_error("EXPLAIN is not supported with sub-queries");
    }
    CHECK(rex_subquery);
    auto result = rex_subquery->getExecutionResult();
    auto row_set = result->getRows();
    if (row_set->rowCount() > size_t(1)) {
      throw std::runtime_error("Scalar sub-query returned multiple rows");
    }
    if (row_set->rowCount() < size_t(1)) {
      CHECK_EQ(row_set->rowCount(), size_t(0));
      throw std::runtime_error("Scalar sub-query returned no results");
    }
    auto first_row = row_set->getNextRow(false, false);
    auto scalar_tv = boost::get<ScalarTargetValue>(&first_row[0]);
    auto ti = rex_subquery->getType();
    if (ti.is_string()) {
      throw std::runtime_error("Scalar sub-queries which return strings not supported");
    }
    Datum d{0};
    bool is_null_const{false};
    std::tie(d, is_null_const) = datum_from_scalar_tv(scalar_tv, ti);
    return makeExpr<Analyzer::Constant>(ti, is_null_const, d);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateInput(
      const RexInput* rex_input) const {
    const auto source = rex_input->getSourceNode();
    const auto it_rte_idx = input_to_nest_level_.find(source);
    CHECK(it_rte_idx != input_to_nest_level_.end());
    const int rte_idx = it_rte_idx->second;
    const auto scan_source = dynamic_cast<const RelScan*>(source);
    const auto& in_metainfo = source->getOutputMetainfo();
    if (scan_source) {
      // We're at leaf (scan) level and not supposed to have input metadata,
      // the name and type information come directly from the catalog.
      CHECK(in_metainfo.empty());
      const auto table_desc = scan_source->getTableDescriptor();
      const auto cd =
          cat_.getMetadataForColumnBySpi(table_desc->tableId, rex_input->getIndex() + 1);
      CHECK(cd);
      auto col_ti = cd->columnType;
      if (col_ti.is_string()) {
        col_ti.set_type(kTEXT);
      }
      if (cd->isVirtualCol) {
        // TODO(alex): remove at some point, we only need this fixup for backwards
        // compatibility with old imported data
        CHECK_EQ("rowid", cd->columnName);
        col_ti.set_size(8);
      }
      CHECK_LE(static_cast<size_t>(rte_idx), join_types_.size());
      if (rte_idx > 0 && join_types_[rte_idx - 1] == JoinType::LEFT) {
        col_ti.set_notnull(false);
      }
      return std::make_shared<Analyzer::ColumnVar>(
          col_ti, table_desc->tableId, cd->columnId, rte_idx);
    }
    CHECK(!in_metainfo.empty());
    CHECK_GE(rte_idx, 0);
    const size_t col_id = rex_input->getIndex();
    CHECK_LT(col_id, in_metainfo.size());
    auto col_ti = in_metainfo[col_id].get_type_info();
    CHECK_LE(static_cast<size_t>(rte_idx), join_types_.size());
    if (rte_idx > 0 && join_types_[rte_idx - 1] == JoinType::LEFT) {
      col_ti.set_notnull(false);
    }
    return std::make_shared<Analyzer::ColumnVar>(col_ti, -source->getId(), col_id, rte_idx);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateUoper(
      const RexOperator* rex_operator) const {
    CHECK_EQ(size_t(1), rex_operator->size());
    const auto operand_expr = translateScalarRex(rex_operator->getOperand(0));
    const auto sql_op = rex_operator->getOperator();
    switch (sql_op) {
      case kCAST: {
        const auto& target_ti = rex_operator->getType();
        CHECK_NE(kNULLT, target_ti.get_type());
        const auto& operand_ti = operand_expr->get_type_info();
        if (operand_ti.is_string() && target_ti.is_string()) {
          return operand_expr;
        }
        if (target_ti.is_time() ||
            operand_ti
                .is_string()) {  // TODO(alex): check and unify with the rest of the cases
          // Do not propogate encoding on small dates
          return target_ti.is_date_in_days()
                     ? operand_expr->add_cast(SQLTypeInfo(kDATE, false))
                     : operand_expr->add_cast(target_ti);
        }
        if (!operand_ti.is_string() && target_ti.is_string()) {
          return operand_expr->add_cast(target_ti);
        }
  
        return std::make_shared<Analyzer::UOper>(target_ti, false, sql_op, operand_expr);
      }
      case kNOT:
      case kISNULL: {
        return std::make_shared<Analyzer::UOper>(kBOOLEAN, sql_op, operand_expr);
      }
      case kISNOTNULL: {
        auto is_null = std::make_shared<Analyzer::UOper>(kBOOLEAN, kISNULL, operand_expr);
        return std::make_shared<Analyzer::UOper>(kBOOLEAN, kNOT, is_null);
      }
      case kMINUS: {
        const auto& ti = operand_expr->get_type_info();
        return std::make_shared<Analyzer::UOper>(ti, false, kUMINUS, operand_expr);
      }
      case kUNNEST: {
        const auto& ti = operand_expr->get_type_info();
        CHECK(ti.is_array());
        return makeExpr<Analyzer::UOper>(ti.get_elem_type(), false, kUNNEST, operand_expr);
      }
      default:
        CHECK(false);
    }
    return nullptr;
  }
  
  namespace {
  
  std::shared_ptr<Analyzer::Expr> get_in_values_expr(std::shared_ptr<Analyzer::Expr> arg,
                                                     const ResultSet& val_set) {
    if (!can_use_parallel_algorithms(val_set)) {
      return nullptr;
    }
    if (val_set.rowCount() > 5000000 && g_enable_watchdog) {
      throw std::runtime_error(
          "Unable to handle 'expr IN (subquery)', subquery returned 5M+ rows.");
    }
    std::list<std::shared_ptr<Analyzer::Expr>> value_exprs;
    const size_t fetcher_count = cpu_threads();
    std::vector<std::list<std::shared_ptr<Analyzer::Expr>>> expr_set(
        fetcher_count, std::list<std::shared_ptr<Analyzer::Expr>>());
    std::vector<std::future<void>> fetcher_threads;
    const auto& ti = arg->get_type_info();
    const auto entry_count = val_set.entryCount();
    for (size_t i = 0,
                start_entry = 0,
                stride = (entry_count + fetcher_count - 1) / fetcher_count;
         i < fetcher_count && start_entry < entry_count;
         ++i, start_entry += stride) {
      const auto end_entry = std::min(start_entry + stride, entry_count);
      fetcher_threads.push_back(std::async(
          std::launch::async,
          [&](std::list<std::shared_ptr<Analyzer::Expr>>& in_vals,
              const size_t start,
              const size_t end) {
            for (auto index = start; index < end; ++index) {
              auto row = val_set.getRowAt(index);
              if (row.empty()) {
                continue;
              }
              auto scalar_tv = boost::get<ScalarTargetValue>(&row[0]);
              Datum d{0};
              bool is_null_const{false};
              std::tie(d, is_null_const) = datum_from_scalar_tv(scalar_tv, ti);
              if (ti.is_string() && ti.get_compression() != kENCODING_NONE) {
                auto ti_none_encoded = ti;
                ti_none_encoded.set_compression(kENCODING_NONE);
                auto none_encoded_string =
                    makeExpr<Analyzer::Constant>(ti, is_null_const, d);
                auto dict_encoded_string = std::make_shared<Analyzer::UOper>(
                    ti, false, kCAST, none_encoded_string);
                in_vals.push_back(dict_encoded_string);
              } else {
                in_vals.push_back(makeExpr<Analyzer::Constant>(ti, is_null_const, d));
              }
            }
          },
          std::ref(expr_set[i]),
          start_entry,
          end_entry));
    }
    for (auto& child : fetcher_threads) {
      child.get();
    }
  
    val_set.moveToBegin();
    for (auto& exprs : expr_set) {
      value_exprs.splice(value_exprs.end(), exprs);
    }
    return makeExpr<Analyzer::InValues>(arg, value_exprs);
  }
  
  }  // namespace
  
  // Creates an Analyzer expression for an IN subquery which subsequently goes through the
  // regular Executor::codegen() mechanism. The creation of the expression out of subquery's
  // result set is parallelized whenever possible. In addition, take advantage of additional
  // information that elements in the right hand side are constants; see
  // getInIntegerSetExpr().
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateInOper(
      const RexOperator* rex_operator) const {
    if (just_explain_) {
      throw std::runtime_error("EXPLAIN is not supported with sub-queries");
    }
    CHECK(rex_operator->size() == 2);
    const auto lhs = translateScalarRex(rex_operator->getOperand(0));
    const auto rhs = rex_operator->getOperand(1);
    const auto rex_subquery = dynamic_cast<const RexSubQuery*>(rhs);
    CHECK(rex_subquery);
    auto ti = lhs->get_type_info();
    auto result = rex_subquery->getExecutionResult();
    auto& row_set = result->getRows();
    CHECK_EQ(size_t(1), row_set->colCount());
    const auto& rhs_ti = row_set->getColType(0);
    if (rhs_ti.get_type() != ti.get_type()) {
      throw std::runtime_error(
          "The two sides of the IN operator must have the same type; found " +
          ti.get_type_name() + " and " + rhs_ti.get_type_name());
    }
    row_set->moveToBegin();
    if (row_set->entryCount() > 10000) {
      std::shared_ptr<Analyzer::Expr> expr;
      if ((ti.is_integer() || (ti.is_string() && ti.get_compression() == kENCODING_DICT)) &&
          !row_set->getQueryMemDesc().didOutputColumnar()) {
        expr = getInIntegerSetExpr(lhs, *row_set);
        // Handle the highly unlikely case when the InIntegerSet ended up being tiny.
        // Just let it fall through the usual InValues path at the end of this method,
        // its codegen knows to use inline comparisons for few values.
        if (expr && std::static_pointer_cast<Analyzer::InIntegerSet>(expr)
                            ->get_value_list()
                            .size() <= 100) {
          expr = nullptr;
        }
      } else {
        expr = get_in_values_expr(lhs, *row_set);
      }
      if (expr) {
        return expr;
      }
    }
    std::list<std::shared_ptr<Analyzer::Expr>> value_exprs;
    while (true) {
      auto row = row_set->getNextRow(true, false);
      if (row.empty()) {
        break;
      }
      if (g_enable_watchdog && value_exprs.size() >= 10000) {
        throw std::runtime_error(
            "Unable to handle 'expr IN (subquery)', subquery returned 10000+ rows.");
      }
      auto scalar_tv = boost::get<ScalarTargetValue>(&row[0]);
      Datum d{0};
      bool is_null_const{false};
      std::tie(d, is_null_const) = datum_from_scalar_tv(scalar_tv, ti);
      if (ti.is_string() && ti.get_compression() != kENCODING_NONE) {
        auto ti_none_encoded = ti;
        ti_none_encoded.set_compression(kENCODING_NONE);
        auto none_encoded_string = makeExpr<Analyzer::Constant>(ti, is_null_const, d);
        auto dict_encoded_string =
            std::make_shared<Analyzer::UOper>(ti, false, kCAST, none_encoded_string);
        value_exprs.push_back(dict_encoded_string);
      } else {
        value_exprs.push_back(makeExpr<Analyzer::Constant>(ti, is_null_const, d));
      }
    }
    return makeExpr<Analyzer::InValues>(lhs, value_exprs);
  }
  
  namespace {
  
  const size_t g_max_integer_set_size{1 << 25};
  
  void fill_dictionary_encoded_in_vals(
      std::vector<int64_t>& in_vals,
      std::atomic<size_t>& total_in_vals_count,
      const ResultSet* values_rowset,
      const std::pair<int64_t, int64_t> values_rowset_slice,
      const StringDictionaryProxy* source_dict,
      const StringDictionaryProxy* dest_dict,
      const int64_t needle_null_val) {
    CHECK(in_vals.empty());
    bool dicts_are_equal = source_dict == dest_dict;
    for (auto index = values_rowset_slice.first; index < values_rowset_slice.second;
         ++index) {
      const auto row = values_rowset->getOneColRow(index);
      if (UNLIKELY(!row.valid)) {
        continue;
      }
      if (dicts_are_equal) {
        in_vals.push_back(row.value);
      } else {
        const int string_id =
            row.value == needle_null_val
                ? needle_null_val
                : dest_dict->getIdOfString(source_dict->getString(row.value));
        if (string_id != StringDictionary::INVALID_STR_ID) {
          in_vals.push_back(string_id);
        }
      }
      if (UNLIKELY(g_enable_watchdog && (in_vals.size() & 1023) == 0 &&
                   total_in_vals_count.fetch_add(1024) >= g_max_integer_set_size)) {
        throw std::runtime_error(
            "Unable to handle 'expr IN (subquery)', subquery returned 30M+ rows.");
      }
    }
  }
  
  void fill_integer_in_vals(std::vector<int64_t>& in_vals,
                            std::atomic<size_t>& total_in_vals_count,
                            const ResultSet* values_rowset,
                            const std::pair<int64_t, int64_t> values_rowset_slice) {
    CHECK(in_vals.empty());
    for (auto index = values_rowset_slice.first; index < values_rowset_slice.second;
         ++index) {
      const auto row = values_rowset->getOneColRow(index);
      if (row.valid) {
        in_vals.push_back(row.value);
        if (UNLIKELY(g_enable_watchdog && (in_vals.size() & 1023) == 0 &&
                     total_in_vals_count.fetch_add(1024) >= g_max_integer_set_size)) {
          throw std::runtime_error(
              "Unable to handle 'expr IN (subquery)', subquery returned 30M+ rows.");
        }
      }
    }
  }
  
  // Multi-node counterpart of the other version. Saves round-trips, which is crucial
  // for a big right-hand side result. It only handles physical string dictionary ids,
  // therefore it won't be able to handle a right-hand side sub-query with a CASE
  // returning literals on some branches. That case isn't hard too handle either, but
  // it's not clear it's actually important in practice.
  // RelAlgTranslator::getInIntegerSetExpr makes sure, by checking the encodings, that this
  // function isn't called in such cases.
  void fill_dictionary_encoded_in_vals(
      std::vector<int64_t>& in_vals,
      std::atomic<size_t>& total_in_vals_count,
      const ResultSet* values_rowset,
      const std::pair<int64_t, int64_t> values_rowset_slice,
      const std::vector<LeafHostInfo>& leaf_hosts,
      const DictRef source_dict_ref,
      const DictRef dest_dict_ref,
      const int32_t dest_generation,
      const int64_t needle_null_val) {
    CHECK(in_vals.empty());
    std::vector<int32_t> source_ids;
    source_ids.reserve(values_rowset->entryCount());
    bool has_nulls = false;
    if (source_dict_ref == dest_dict_ref) {
      in_vals.reserve(values_rowset_slice.second - values_rowset_slice.first +
                      1);  // Add 1 to cover interval
      for (auto index = values_rowset_slice.first; index < values_rowset_slice.second;
           ++index) {
        const auto row = values_rowset->getOneColRow(index);
        if (!row.valid) {
          continue;
        }
        if (row.value != needle_null_val) {
          in_vals.push_back(row.value);
          if (UNLIKELY(g_enable_watchdog && (in_vals.size() & 1023) == 0 &&
                       total_in_vals_count.fetch_add(1024) >= g_max_integer_set_size)) {
            throw std::runtime_error(
                "Unable to handle 'expr IN (subquery)', subquery returned 30M+ rows.");
          }
        } else {
          has_nulls = true;
        }
      }
      if (has_nulls) {
        in_vals.push_back(
            needle_null_val);  // we've deduped null values as an optimization, although
                               // this is not required by consumer
      }
      return;
    }
    // Code path below is for when dictionaries are not shared
    for (auto index = values_rowset_slice.first; index < values_rowset_slice.second;
         ++index) {
      const auto row = values_rowset->getOneColRow(index);
      if (row.valid) {
        if (row.value != needle_null_val) {
          source_ids.push_back(row.value);
        } else {
          has_nulls = true;
        }
      }
    }
    std::vector<int32_t> dest_ids;
    translate_string_ids(dest_ids,
                         leaf_hosts.front(),
                         dest_dict_ref,
                         source_ids,
                         source_dict_ref,
                         dest_generation);
    CHECK_EQ(dest_ids.size(), source_ids.size());
    in_vals.reserve(dest_ids.size() + (has_nulls ? 1 : 0));
    if (has_nulls) {
      in_vals.push_back(needle_null_val);
    }
    for (const int32_t dest_id : dest_ids) {
      if (dest_id != StringDictionary::INVALID_STR_ID) {
        in_vals.push_back(dest_id);
        if (UNLIKELY(g_enable_watchdog && (in_vals.size() & 1023) == 0 &&
                     total_in_vals_count.fetch_add(1024) >= g_max_integer_set_size)) {
          throw std::runtime_error(
              "Unable to handle 'expr IN (subquery)', subquery returned 30M+ rows.");
        }
      }
    }
  }
  
  }  // namespace
  
  // The typical IN subquery involves either dictionary-encoded strings or integers.
  // Analyzer::InValues is a very heavy representation of the right hand side of such
  // a query since we already know the right hand would be a list of Analyzer::Constant
  // shared pointers. We can avoid the big overhead of each Analyzer::Constant and the
  // refcounting associated with shared pointers by creating an abbreviated InIntegerSet
  // representation of the IN expression which takes advantage of the this information.
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::getInIntegerSetExpr(
      std::shared_ptr<Analyzer::Expr> arg,
      const ResultSet& val_set) const {
    if (!can_use_parallel_algorithms(val_set)) {
      return nullptr;
    }
    std::vector<int64_t> value_exprs;
    const size_t fetcher_count = cpu_threads();
    std::vector<std::vector<int64_t>> expr_set(fetcher_count);
    std::vector<std::future<void>> fetcher_threads;
    const auto& arg_type = arg->get_type_info();
    const auto entry_count = val_set.entryCount();
    CHECK_EQ(size_t(1), val_set.colCount());
    const auto& col_type = val_set.getColType(0);
    if (g_cluster && arg_type.is_string() &&
        (col_type.get_comp_param() <= 0 || arg_type.get_comp_param() <= 0)) {
      // Skip this case for now, see comment for fill_dictionary_encoded_in_vals.
      return nullptr;
    }
    std::atomic<size_t> total_in_vals_count{0};
    for (size_t i = 0,
                start_entry = 0,
                stride = (entry_count + fetcher_count - 1) / fetcher_count;
         i < fetcher_count && start_entry < entry_count;
         ++i, start_entry += stride) {
      expr_set[i].reserve(entry_count / fetcher_count);
      const auto end_entry = std::min(start_entry + stride, entry_count);
      if (arg_type.is_string()) {
        CHECK_EQ(kENCODING_DICT, arg_type.get_compression());
        // const int32_t dest_dict_id = arg_type.get_comp_param();
        // const int32_t source_dict_id = col_type.get_comp_param();
        const DictRef dest_dict_ref(arg_type.get_comp_param(), cat_.getDatabaseId());
        const DictRef source_dict_ref(col_type.get_comp_param(), cat_.getDatabaseId());
        const auto dd = executor_->getStringDictionaryProxy(
            arg_type.get_comp_param(), val_set.getRowSetMemOwner(), true);
        const auto sd = executor_->getStringDictionaryProxy(
            col_type.get_comp_param(), val_set.getRowSetMemOwner(), true);
        CHECK(sd);
        const auto needle_null_val = inline_int_null_val(arg_type);
        fetcher_threads.push_back(std::async(
            std::launch::async,
            [this,
             &val_set,
             &total_in_vals_count,
             sd,
             dd,
             source_dict_ref,
             dest_dict_ref,
             needle_null_val](
                std::vector<int64_t>& in_vals, const size_t start, const size_t end) {
              if (g_cluster) {
                CHECK_GE(dd->getGeneration(), 0);
                fill_dictionary_encoded_in_vals(in_vals,
                                                total_in_vals_count,
                                                &val_set,
                                                {start, end},
                                                cat_.getStringDictionaryHosts(),
                                                source_dict_ref,
                                                dest_dict_ref,
                                                dd->getGeneration(),
                                                needle_null_val);
              } else {
                fill_dictionary_encoded_in_vals(in_vals,
                                                total_in_vals_count,
                                                &val_set,
                                                {start, end},
                                                sd,
                                                dd,
                                                needle_null_val);
              }
            },
            std::ref(expr_set[i]),
            start_entry,
            end_entry));
      } else {
        CHECK(arg_type.is_integer());
        fetcher_threads.push_back(std::async(
            std::launch::async,
            [&val_set, &total_in_vals_count](
                std::vector<int64_t>& in_vals, const size_t start, const size_t end) {
              fill_integer_in_vals(in_vals, total_in_vals_count, &val_set, {start, end});
            },
            std::ref(expr_set[i]),
            start_entry,
            end_entry));
      }
    }
    for (auto& child : fetcher_threads) {
      child.get();
    }
  
    val_set.moveToBegin();
    value_exprs.reserve(entry_count);
    for (auto& exprs : expr_set) {
      value_exprs.insert(value_exprs.end(), exprs.begin(), exprs.end());
    }
    return makeExpr<Analyzer::InIntegerSet>(
        arg, value_exprs, arg_type.get_notnull() && col_type.get_notnull());
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateOper(
      const RexOperator* rex_operator) const {
    CHECK_GT(rex_operator->size(), size_t(0));
    if (rex_operator->size() == 1) {
      return translateUoper(rex_operator);
    }
    const auto sql_op = rex_operator->getOperator();
    if (sql_op == kIN) {
      return translateInOper(rex_operator);
    }
    if (sql_op == kMINUS || sql_op == kPLUS) {
      auto date_plus_minus = translateDatePlusMinus(rex_operator);
      if (date_plus_minus) {
        return date_plus_minus;
      }
    }
    if (sql_op == kOVERLAPS) {
      return translateOverlapsOper(rex_operator);
    } else if (IS_COMPARISON(sql_op)) {
      auto geo_comp = translateGeoComparison(rex_operator);
      if (geo_comp) {
        return geo_comp;
      }
    }
    auto lhs = translateScalarRex(rex_operator->getOperand(0));
    for (size_t i = 1; i < rex_operator->size(); ++i) {
      std::shared_ptr<Analyzer::Expr> rhs;
      SQLQualifier sql_qual{kONE};
      const auto rhs_op = rex_operator->getOperand(i);
      std::tie(rhs, sql_qual) = get_quantified_rhs(rhs_op, *this);
      if (!rhs) {
        rhs = translateScalarRex(rhs_op);
      }
      CHECK(rhs);
      lhs = Parser::OperExpr::normalize(sql_op, sql_qual, lhs, rhs);
    }
    return lhs;
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateOverlapsOper(
      const RexOperator* rex_operator) const {
    const auto sql_op = rex_operator->getOperator();
    CHECK(sql_op == kOVERLAPS);
  
    const auto lhs = translateScalarRex(rex_operator->getOperand(0));
    const auto lhs_ti = lhs->get_type_info();
    if (lhs_ti.is_geometry()) {
      return translateGeoOverlapsOper(rex_operator);
    } else {
      throw std::runtime_error(
          "Overlaps equivalence is currently only supported for geospatial types");
    }
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateCase(
      const RexCase* rex_case) const {
    std::shared_ptr<Analyzer::Expr> else_expr;
    std::list<std::pair<std::shared_ptr<Analyzer::Expr>, std::shared_ptr<Analyzer::Expr>>>
        expr_list;
    for (size_t i = 0; i < rex_case->branchCount(); ++i) {
      const auto when_expr = translateScalarRex(rex_case->getWhen(i));
      const auto then_expr = translateScalarRex(rex_case->getThen(i));
      expr_list.emplace_back(when_expr, then_expr);
    }
    if (rex_case->getElse()) {
      else_expr = translateScalarRex(rex_case->getElse());
    }
    return Parser::CaseExpr::normalize(expr_list, else_expr);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateLike(
      const RexFunctionOperator* rex_function) const {
    CHECK(rex_function->size() == 2 || rex_function->size() == 3);
    const auto arg = translateScalarRex(rex_function->getOperand(0));
    const auto like = translateScalarRex(rex_function->getOperand(1));
    if (!std::dynamic_pointer_cast<const Analyzer::Constant>(like)) {
      throw std::runtime_error("The matching pattern must be a literal.");
    }
    const auto escape = (rex_function->size() == 3)
                            ? translateScalarRex(rex_function->getOperand(2))
                            : nullptr;
    const bool is_ilike = rex_function->getName() == "PG_ILIKE"sv;
    return Parser::LikeExpr::get(arg, like, escape, is_ilike, false);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateRegexp(
      const RexFunctionOperator* rex_function) const {
    CHECK(rex_function->size() == 2 || rex_function->size() == 3);
    const auto arg = translateScalarRex(rex_function->getOperand(0));
    const auto pattern = translateScalarRex(rex_function->getOperand(1));
    if (!std::dynamic_pointer_cast<const Analyzer::Constant>(pattern)) {
      throw std::runtime_error("The matching pattern must be a literal.");
    }
    const auto escape = (rex_function->size() == 3)
                            ? translateScalarRex(rex_function->getOperand(2))
                            : nullptr;
    return Parser::RegexpExpr::get(arg, pattern, escape, false);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateLikely(
      const RexFunctionOperator* rex_function) const {
    CHECK(rex_function->size() == 1);
    const auto arg = translateScalarRex(rex_function->getOperand(0));
    return makeExpr<Analyzer::LikelihoodExpr>(arg, 0.9375);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateUnlikely(
      const RexFunctionOperator* rex_function) const {
    CHECK(rex_function->size() == 1);
    const auto arg = translateScalarRex(rex_function->getOperand(0));
    return makeExpr<Analyzer::LikelihoodExpr>(arg, 0.0625);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateExtract(
      const RexFunctionOperator* rex_function) const {
    CHECK_EQ(size_t(2), rex_function->size());
    const auto timeunit = translateScalarRex(rex_function->getOperand(0));
    const auto timeunit_lit = std::dynamic_pointer_cast<Analyzer::Constant>(timeunit);
    if (!timeunit_lit) {
      throw std::runtime_error("The time unit parameter must be a literal.");
    }
    const auto from_expr = translateScalarRex(rex_function->getOperand(1));
    const bool is_date_trunc = rex_function->getName() == "PG_DATE_TRUNC"sv;
    if (is_date_trunc) {
      return DateTruncExpr::generate(from_expr, *timeunit_lit->get_constval().stringval);
    } else {
      return ExtractExpr::generate(from_expr, *timeunit_lit->get_constval().stringval);
    }
  }
  
  namespace {
  
  std::shared_ptr<Analyzer::Constant> makeNumericConstant(const SQLTypeInfo& ti,
                                                          const long val) {
    CHECK(ti.is_number());
    Datum datum{0};
    switch (ti.get_type()) {
      case kTINYINT: {
        datum.tinyintval = val;
        break;
      }
      case kSMALLINT: {
        datum.smallintval = val;
        break;
      }
      case kINT: {
        datum.intval = val;
        break;
      }
      case kBIGINT: {
        datum.bigintval = val;
        break;
      }
      case kDECIMAL:
      case kNUMERIC: {
        datum.bigintval = val * exp_to_scale(ti.get_scale());
        break;
      }
      case kFLOAT: {
        datum.floatval = val;
        break;
      }
      case kDOUBLE: {
        datum.doubleval = val;
        break;
      }
      default:
        CHECK(false);
    }
    return makeExpr<Analyzer::Constant>(ti, false, datum);
  }
  
  }  // namespace
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateDateadd(
      const RexFunctionOperator* rex_function) const {
    CHECK_EQ(size_t(3), rex_function->size());
    const auto timeunit = translateScalarRex(rex_function->getOperand(0));
    const auto timeunit_lit = std::dynamic_pointer_cast<Analyzer::Constant>(timeunit);
    if (!timeunit_lit) {
      throw std::runtime_error("The time unit parameter must be a literal.");
    }
  
    const auto number_units = translateScalarRex(rex_function->getOperand(1));
    auto cast_number_units = number_units->add_cast(SQLTypeInfo(kBIGINT, false));
    const auto datetime = translateScalarRex(rex_function->getOperand(2));
    const auto& datetime_ti = datetime->get_type_info();
    if (datetime_ti.get_type() == kTIME) {
      throw std::runtime_error("DateAdd operation not supported for TIME.");
    }
    const auto& field = to_dateadd_field(*timeunit_lit->get_constval().stringval);
    if (!datetime_ti.is_high_precision_timestamp() &&
        DateTimeUtils::is_subsecond_dateadd_field(field)) {
      // Scale the number to get value in seconds
      const auto bigint_ti = SQLTypeInfo(kBIGINT, false);
      cast_number_units = makeExpr<Analyzer::BinOper>(
          bigint_ti.get_type(),
          kDIVIDE,
          kONE,
          cast_number_units,
          makeNumericConstant(bigint_ti,
                              DateTimeUtils::get_dateadd_timestamp_precision_scale(field)));
      cast_number_units = fold_expr(cast_number_units.get());
    }
    if (datetime_ti.is_high_precision_timestamp() &&
        DateTimeUtils::is_subsecond_dateadd_field(field)) {
      const auto oper_scale = DateTimeUtils::get_dateadd_high_precision_adjusted_scale(
          field, datetime_ti.get_dimension());
      if (oper_scale.first) {
        // scale number to desired precision
        const auto bigint_ti = SQLTypeInfo(kBIGINT, false);
        cast_number_units =
            makeExpr<Analyzer::BinOper>(bigint_ti.get_type(),
                                        oper_scale.first,
                                        kONE,
                                        cast_number_units,
                                        makeNumericConstant(bigint_ti, oper_scale.second));
        cast_number_units = fold_expr(cast_number_units.get());
      }
    }
    return makeExpr<Analyzer::DateaddExpr>(
        SQLTypeInfo(kTIMESTAMP, datetime_ti.get_dimension(), 0, false),
        to_dateadd_field(*timeunit_lit->get_constval().stringval),
        cast_number_units,
        datetime);
  }
  
  namespace {
  
  std::string get_datetimeplus_rewrite_funcname(const SQLOps& op) {
    CHECK(op == kPLUS);
    return "DATETIME_PLUS"s;
  }
  
  }  // namespace
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateDatePlusMinus(
      const RexOperator* rex_operator) const {
    if (rex_operator->size() != 2) {
      return nullptr;
    }
    const auto datetime = translateScalarRex(rex_operator->getOperand(0));
    const auto datetime_ti = datetime->get_type_info();
    if (!datetime_ti.is_timestamp() && !datetime_ti.is_date()) {
      if (datetime_ti.get_type() == kTIME) {
        throw std::runtime_error("DateTime addition/subtraction not supported for TIME.");
      }
      return nullptr;
    }
    const auto rhs = translateScalarRex(rex_operator->getOperand(1));
    const auto rhs_ti = rhs->get_type_info();
    if (rhs_ti.get_type() == kTIMESTAMP || rhs_ti.get_type() == kDATE) {
      if (datetime_ti.is_high_precision_timestamp() ||
          rhs_ti.is_high_precision_timestamp()) {
        throw std::runtime_error(
            "High Precision timestamps are not supported for TIMESTAMPDIFF operation. Use "
            "DATEDIFF.");
      }
      auto bigint_ti = SQLTypeInfo(kBIGINT, false);
      const auto& rex_operator_ti = rex_operator->getType();
      const auto datediff_field =
          (rex_operator_ti.get_type() == kINTERVAL_DAY_TIME) ? dtSECOND : dtMONTH;
      auto result =
          makeExpr<Analyzer::DatediffExpr>(bigint_ti, datediff_field, rhs, datetime);
      // multiply 1000 to result since expected result should be in millisecond precision.
      if (rex_operator_ti.get_type() == kINTERVAL_DAY_TIME) {
        return makeExpr<Analyzer::BinOper>(bigint_ti.get_type(),
                                           kMULTIPLY,
                                           kONE,
                                           result,
                                           makeNumericConstant(bigint_ti, 1000));
      } else {
        return result;
      }
    }
    const auto op = rex_operator->getOperator();
    if (op == kPLUS) {
      std::vector<std::shared_ptr<Analyzer::Expr>> args = {datetime, rhs};
      auto dt_plus = makeExpr<Analyzer::FunctionOper>(
          datetime_ti, get_datetimeplus_rewrite_funcname(op), args);
      const auto date_trunc = rewrite_to_date_trunc(dt_plus.get());
      if (date_trunc) {
        return date_trunc;
      }
    }
    const auto interval = fold_expr(rhs.get());
    auto interval_ti = interval->get_type_info();
    auto bigint_ti = SQLTypeInfo(kBIGINT, false);
    const auto interval_lit = std::dynamic_pointer_cast<Analyzer::Constant>(interval);
    if (interval_ti.get_type() == kINTERVAL_DAY_TIME) {
      std::shared_ptr<Analyzer::Expr> interval_sec;
      if (interval_lit) {
        interval_sec =
            makeNumericConstant(bigint_ti,
                                (op == kMINUS ? -interval_lit->get_constval().bigintval
                                              : interval_lit->get_constval().bigintval) /
                                    1000);
      } else {
        interval_sec = makeExpr<Analyzer::BinOper>(bigint_ti.get_type(),
                                                   kDIVIDE,
                                                   kONE,
                                                   interval,
                                                   makeNumericConstant(bigint_ti, 1000));
        if (op == kMINUS) {
          interval_sec =
              std::make_shared<Analyzer::UOper>(bigint_ti, false, kUMINUS, interval_sec);
        }
      }
      return makeExpr<Analyzer::DateaddExpr>(datetime_ti, daSECOND, interval_sec, datetime);
    }
    CHECK(interval_ti.get_type() == kINTERVAL_YEAR_MONTH);
    const auto interval_months = op == kMINUS ? std::make_shared<Analyzer::UOper>(
                                                    bigint_ti, false, kUMINUS, interval)
                                              : interval;
    return makeExpr<Analyzer::DateaddExpr>(datetime_ti, daMONTH, interval_months, datetime);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateDatediff(
      const RexFunctionOperator* rex_function) const {
    CHECK_EQ(size_t(3), rex_function->size());
    const auto timeunit = translateScalarRex(rex_function->getOperand(0));
    const auto timeunit_lit = std::dynamic_pointer_cast<Analyzer::Constant>(timeunit);
    if (!timeunit_lit) {
      throw std::runtime_error("The time unit parameter must be a literal.");
    }
    const auto start = translateScalarRex(rex_function->getOperand(1));
    const auto end = translateScalarRex(rex_function->getOperand(2));
    const auto field = to_datediff_field(*timeunit_lit->get_constval().stringval);
    return makeExpr<Analyzer::DatediffExpr>(SQLTypeInfo(kBIGINT, false), field, start, end);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateDatepart(
      const RexFunctionOperator* rex_function) const {
    CHECK_EQ(size_t(2), rex_function->size());
    const auto timeunit = translateScalarRex(rex_function->getOperand(0));
    const auto timeunit_lit = std::dynamic_pointer_cast<Analyzer::Constant>(timeunit);
    if (!timeunit_lit) {
      throw std::runtime_error("The time unit parameter must be a literal.");
    }
    const auto from_expr = translateScalarRex(rex_function->getOperand(1));
    return ExtractExpr::generate(
        from_expr, to_datepart_field(*timeunit_lit->get_constval().stringval));
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateLength(
      const RexFunctionOperator* rex_function) const {
    CHECK_EQ(size_t(1), rex_function->size());
    const auto str_arg = translateScalarRex(rex_function->getOperand(0));
    return makeExpr<Analyzer::CharLengthExpr>(str_arg->decompress(),
                                              rex_function->getName() == "CHAR_LENGTH"sv);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateKeyForString(
      const RexFunctionOperator* rex_function) const {
    const auto& args = translateFunctionArgs(rex_function);
    CHECK_EQ(size_t(1), args.size());
    const auto expr = dynamic_cast<Analyzer::Expr*>(args[0].get());
    if (nullptr == expr || !expr->get_type_info().is_string() ||
        expr->get_type_info().is_varlen()) {
      throw std::runtime_error(rex_function->getName() +
                               " expects a dictionary encoded text column.");
    }
    return makeExpr<Analyzer::KeyForStringExpr>(args[0]);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateLower(
      const RexFunctionOperator* rex_function) const {
    const auto& args = translateFunctionArgs(rex_function);
    CHECK_EQ(size_t(1), args.size());
  
    if (const auto expr = dynamic_cast<Analyzer::Expr*>(args[0].get());
        expr && expr->get_type_info().is_string() && !expr->get_type_info().is_varlen()) {
      return makeExpr<Analyzer::LowerExpr>(args[0]);
    }
  
    throw std::runtime_error(rex_function->getName() +
                             " expects a dictionary encoded text column.");
  }
  
  Analyzer::ExpressionPtr RelAlgTranslator::translateCardinality(
      const RexFunctionOperator* rex_function) const {
    const auto ret_ti = rex_function->getType();
    const auto arg = translateScalarRex(rex_function->getOperand(0));
    const auto arg_ti = arg->get_type_info();
    if (!arg_ti.is_array()) {
      throw std::runtime_error(rex_function->getName() + " expects an array expression.");
    }
    if (arg_ti.get_subtype() == kARRAY) {
      throw std::runtime_error(rex_function->getName() +
                               " expects one-dimension array expression.");
    }
    const auto array_size = arg_ti.get_size();
    const auto array_elem_size = arg_ti.get_elem_type().get_array_context_logical_size();
  
    if (array_size > 0) {
      if (array_elem_size <= 0) {
        throw std::runtime_error(rex_function->getName() +
                                 ": unexpected array element type.");
      }
      // Return cardinality of a fixed length array
      return makeNumericConstant(ret_ti, array_size / array_elem_size);
    }
    // Variable length array cardinality will be calculated at runtime
    return makeExpr<Analyzer::CardinalityExpr>(arg);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateItem(
      const RexFunctionOperator* rex_function) const {
    CHECK_EQ(size_t(2), rex_function->size());
    const auto base = translateScalarRex(rex_function->getOperand(0));
    const auto index = translateScalarRex(rex_function->getOperand(1));
    return makeExpr<Analyzer::BinOper>(
        base->get_type_info().get_elem_type(), false, kARRAY_AT, kONE, base, index);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateNow() const {
    return Parser::TimestampLiteral::get(now_);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateDatetime(
      const RexFunctionOperator* rex_function) const {
    CHECK_EQ(size_t(1), rex_function->size());
    const auto arg = translateScalarRex(rex_function->getOperand(0));
    const auto arg_lit = std::dynamic_pointer_cast<Analyzer::Constant>(arg);
    const std::string datetime_err{R"(Only DATETIME('NOW') supported for now.)"};
    if (!arg_lit) {
      throw std::runtime_error(datetime_err);
    }
    CHECK(arg_lit->get_type_info().is_string());
    if (*arg_lit->get_constval().stringval != "NOW"sv) {
      throw std::runtime_error(datetime_err);
    }
    return translateNow();
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateAbs(
      const RexFunctionOperator* rex_function) const {
    std::list<std::pair<std::shared_ptr<Analyzer::Expr>, std::shared_ptr<Analyzer::Expr>>>
        expr_list;
    CHECK_EQ(size_t(1), rex_function->size());
    const auto operand = translateScalarRex(rex_function->getOperand(0));
    const auto& operand_ti = operand->get_type_info();
    CHECK(operand_ti.is_number());
    const auto zero = makeNumericConstant(operand_ti, 0);
    const auto lt_zero = makeExpr<Analyzer::BinOper>(kBOOLEAN, kLT, kONE, operand, zero);
    const auto uminus_operand =
        makeExpr<Analyzer::UOper>(operand_ti.get_type(), kUMINUS, operand);
    expr_list.emplace_back(lt_zero, uminus_operand);
    return makeExpr<Analyzer::CaseExpr>(operand_ti, false, expr_list, operand);
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateSign(
      const RexFunctionOperator* rex_function) const {
    std::list<std::pair<std::shared_ptr<Analyzer::Expr>, std::shared_ptr<Analyzer::Expr>>>
        expr_list;
    CHECK_EQ(size_t(1), rex_function->size());
    const auto operand = translateScalarRex(rex_function->getOperand(0));
    const auto& operand_ti = operand->get_type_info();
    CHECK(operand_ti.is_number());
    const auto zero = makeNumericConstant(operand_ti, 0);
    const auto lt_zero = makeExpr<Analyzer::BinOper>(kBOOLEAN, kLT, kONE, operand, zero);
    expr_list.emplace_back(lt_zero, makeNumericConstant(operand_ti, -1));
    const auto eq_zero = makeExpr<Analyzer::BinOper>(kBOOLEAN, kEQ, kONE, operand, zero);
    expr_list.emplace_back(eq_zero, makeNumericConstant(operand_ti, 0));
    const auto gt_zero = makeExpr<Analyzer::BinOper>(kBOOLEAN, kGT, kONE, operand, zero);
    expr_list.emplace_back(gt_zero, makeNumericConstant(operand_ti, 1));
    return makeExpr<Analyzer::CaseExpr>(
        operand_ti,
        false,
        expr_list,
        makeExpr<Analyzer::Constant>(operand_ti, true, Datum{0}));
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateOffsetInFragment() const {
    return makeExpr<Analyzer::OffsetInFragment>();
  }
  
  Analyzer::ExpressionPtr RelAlgTranslator::translateArrayFunction(
      const RexFunctionOperator* rex_function) const {
    if (rex_function->getType().get_subtype() == kNULLT) {
      auto sql_type = rex_function->getType();
      CHECK(sql_type.get_type() == kARRAY);
  
      // FIX-ME:  Deal with NULL arrays
      auto translated_function_args(translateFunctionArgs(rex_function));
      if (translated_function_args.size() > 0) {
        auto const& first_element_logical_type(
            get_logical_type_info(translated_function_args[0]->get_type_info()));
  
        on_member_of_typeset<kCHAR, kVARCHAR, kTEXT>(
            first_element_logical_type,
            [&] {
              bool same_type_status = true;
              for (auto const& expr_ptr : translated_function_args) {
                same_type_status =
                    same_type_status && (expr_ptr->get_type_info().is_string());
              }
  
              if (same_type_status == false) {
                throw std::runtime_error(
                    "All elements of the array are not of the same logical subtype; "
                    "consider casting to force this condition.");
              }
  
              sql_type.set_subtype(first_element_logical_type.get_type());
              sql_type.set_compression(kENCODING_FIXED);
              sql_type.set_comp_param(TRANSIENT_DICT_ID);
            },
            [&] {
              // Non string types
              bool same_type_status = true;
              for (auto const& expr_ptr : translated_function_args) {
                same_type_status =
                    same_type_status && (first_element_logical_type ==
                                         get_logical_type_info(expr_ptr->get_type_info()));
              }
  
              if (same_type_status == false) {
                throw std::runtime_error(
                    "All elements of the array are not of the same logical subtype; "
                    "consider casting to force this condition.");
              }
              sql_type.set_subtype(first_element_logical_type.get_type());
              sql_type.set_scale(first_element_logical_type.get_scale());
              sql_type.set_precision(first_element_logical_type.get_precision());
            });
  
        feature_stash_.setCPUOnlyExecutionRequired();
        return makeExpr<Analyzer::ArrayExpr>(
            sql_type, translated_function_args, feature_stash_.getAndBumpArrayExprCount());
      } else {
        throw std::runtime_error("NULL ARRAY[] expressions not supported yet.  FIX-ME.");
      }
    } else {
      feature_stash_.setCPUOnlyExecutionRequired();
      return makeExpr<Analyzer::ArrayExpr>(rex_function->getType(),
                                           translateFunctionArgs(rex_function),
                                           feature_stash_.getAndBumpArrayExprCount());
    }
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateFunction(
      const RexFunctionOperator* rex_function) const {
    if (func_resolve(rex_function->getName(), "LIKE"sv, "PG_ILIKE"sv)) {
      return translateLike(rex_function);
    }
    if (rex_function->getName() == "REGEXP_LIKE"sv) {
      return translateRegexp(rex_function);
    }
    if (rex_function->getName() == "LIKELY"sv) {
      return translateLikely(rex_function);
    }
    if (rex_function->getName() == "UNLIKELY"sv) {
      return translateUnlikely(rex_function);
    }
    if (func_resolve(rex_function->getName(), "PG_EXTRACT"sv, "PG_DATE_TRUNC"sv)) {
      return translateExtract(rex_function);
    }
    if (rex_function->getName() == "DATEADD"sv) {
      return translateDateadd(rex_function);
    }
    if (rex_function->getName() == "DATEDIFF"sv) {
      return translateDatediff(rex_function);
    }
    if (rex_function->getName() == "DATEPART"sv) {
      return translateDatepart(rex_function);
    }
    if (func_resolve(rex_function->getName(), "LENGTH"sv, "CHAR_LENGTH"sv)) {
      return translateLength(rex_function);
    }
    if (rex_function->getName() == "KEY_FOR_STRING"sv) {
      return translateKeyForString(rex_function);
    }
    if (g_enable_experimental_string_functions && rex_function->getName() == "LOWER"sv) {
      return translateLower(rex_function);
    }
    if (func_resolve(rex_function->getName(), "CARDINALITY"sv, "ARRAY_LENGTH"sv)) {
      return translateCardinality(rex_function);
    }
    if (rex_function->getName() == "ITEM"sv) {
      return translateItem(rex_function);
    }
    if (rex_function->getName() == "NOW"sv) {
      return translateNow();
    }
    if (rex_function->getName() == "DATETIME"sv) {
      return translateDatetime(rex_function);
    }
    if (func_resolve(rex_function->getName(), "usTIMESTAMP"sv, "nsTIMESTAMP"sv)) {
      return translateHPTLiteral(rex_function);
    }
    if (rex_function->getName() == "ABS"sv) {
      return translateAbs(rex_function);
    }
    if (rex_function->getName() == "SIGN"sv) {
      return translateSign(rex_function);
    }
    if (func_resolve(rex_function->getName(), "CEIL"sv, "FLOOR"sv)) {
      return makeExpr<Analyzer::FunctionOperWithCustomTypeHandling>(
          rex_function->getType(),
          rex_function->getName(),
          translateFunctionArgs(rex_function));
    } else if (rex_function->getName() == "ROUND"sv) {
      std::vector<std::shared_ptr<Analyzer::Expr>> args =
          translateFunctionArgs(rex_function);
  
      if (rex_function->size() == 1) {
        // push a 0 constant if 2nd operand is missing.
        // this needs to be done as calcite returns
        // only the 1st operand without defaulting the 2nd one
        // when the user did not specify the 2nd operand.
        SQLTypes t = kSMALLINT;
        Datum d;
        d.smallintval = 0;
        args.push_back(makeExpr<Analyzer::Constant>(t, false, d));
      }
  
      // make sure we have only 2 operands
      CHECK(args.size() == 2);
  
      if (!args[0]->get_type_info().is_number()) {
        throw std::runtime_error("Only numeric 1st operands are supported");
      }
  
      // the 2nd operand does not need to be a constant
      // it can happily reference another integer column
      if (!args[1]->get_type_info().is_integer()) {
        throw std::runtime_error("Only integer 2nd operands are supported");
      }
  
      // Calcite may upcast decimals in a way that is
      // incompatible with the extension function input. Play it safe and stick with the
      // argument type instead.
      const SQLTypeInfo ret_ti = args[0]->get_type_info().is_decimal()
                                     ? args[0]->get_type_info()
                                     : rex_function->getType();
  
      return makeExpr<Analyzer::FunctionOperWithCustomTypeHandling>(
          ret_ti, rex_function->getName(), args);
    }
    if (rex_function->getName() == "DATETIME_PLUS"sv) {
      auto dt_plus = makeExpr<Analyzer::FunctionOper>(rex_function->getType(),
                                                      rex_function->getName(),
                                                      translateFunctionArgs(rex_function));
      const auto date_trunc = rewrite_to_date_trunc(dt_plus.get());
      if (date_trunc) {
        return date_trunc;
      }
      return translateDateadd(rex_function);
    }
    if (rex_function->getName() == "/INT"sv) {
      CHECK_EQ(size_t(2), rex_function->size());
      std::shared_ptr<Analyzer::Expr> lhs = translateScalarRex(rex_function->getOperand(0));
      std::shared_ptr<Analyzer::Expr> rhs = translateScalarRex(rex_function->getOperand(1));
      const auto rhs_lit = std::dynamic_pointer_cast<Analyzer::Constant>(rhs);
      return Parser::OperExpr::normalize(kDIVIDE, kONE, lhs, rhs);
    }
    if (rex_function->getName() == "Reinterpret"sv) {
      CHECK_EQ(size_t(1), rex_function->size());
      return translateScalarRex(rex_function->getOperand(0));
    }
    if (func_resolve(rex_function->getName(),
                     "ST_X"sv,
                     "ST_Y"sv,
                     "ST_XMin"sv,
                     "ST_YMin"sv,
                     "ST_XMax"sv,
                     "ST_YMax"sv,
                     "ST_NRings"sv,
                     "ST_NPoints"sv,
                     "ST_Length"sv,
                     "ST_Perimeter"sv,
                     "ST_Area"sv,
                     "ST_SRID"sv,
                     "MapD_GeoPolyBoundsPtr"sv /* deprecated */,
                     "MapD_GeoPolyBoundsPtr"sv /* deprecated */,
                     "OmniSci_Geo_PolyBoundsPtr"sv,
                     "OmniSci_Geo_PolyRenderGroup"sv)) {
      CHECK_EQ(rex_function->size(), size_t(1));
      return translateUnaryGeoFunction(rex_function);
    }
    if (func_resolve(rex_function->getName(),
                     "convert_meters_to_pixel_width"sv,
                     "convert_meters_to_pixel_height"sv,
                     "is_point_in_view"sv,
                     "is_point_size_in_view"sv)) {
      return translateFunctionWithGeoArg(rex_function);
    }
    if (func_resolve(rex_function->getName(),
                     "ST_Distance"sv,
                     "ST_MaxDistance"sv,
                     "ST_Intersects"sv,
                     "ST_Disjoint"sv,
                     "ST_Contains"sv,
                     "ST_Within"sv)) {
      CHECK_EQ(rex_function->size(), size_t(2));
      return translateBinaryGeoFunction(rex_function);
    }
    if (func_resolve(rex_function->getName(), "ST_DWithin"sv, "ST_DFullyWithin"sv)) {
      CHECK_EQ(rex_function->size(), size_t(3));
      return translateTernaryGeoFunction(rex_function);
    }
    if (rex_function->getName() == "OFFSET_IN_FRAGMENT"sv) {
      CHECK_EQ(size_t(0), rex_function->size());
      return translateOffsetInFragment();
    }
    if (rex_function->getName() == "ARRAY"sv) {
      // Var args; currently no check.  Possible fix-me -- can array have 0 elements?
      return translateArrayFunction(rex_function);
    }
    if (func_resolve(rex_function->getName(),
                     "ST_GeomFromText"sv,
                     "ST_GeogFromText"sv,
                     "ST_Point"sv)) {
      return translateGeoConstructor(rex_function);
    }
  
    auto arg_expr_list = translateFunctionArgs(rex_function);
    // Reset possibly wrong return type of rex_function to the return
    // type of the optimal valid implementation. The return type can be
    // wrong in the case of multiple implementations of UDF functions
    // that have different return types but Calcite specifies the return
    // type according to the first implementation.
    auto ext_func_sig = bind_function(rex_function->getName(), arg_expr_list);
    auto ret_ti = ext_arg_type_to_type_info(ext_func_sig.getRet());
    // By defualt, the extension function type will not allow nulls. If one of the arguments
    // is nullable, the extension function must also explicitly allow nulls.
    bool arguments_not_null = true;
    for (const auto& arg_expr : arg_expr_list) {
      if (!arg_expr->get_type_info().get_notnull()) {
        arguments_not_null = false;
        break;
      }
    }
    ret_ti.set_notnull(arguments_not_null);
    return makeExpr<Analyzer::FunctionOper>(ret_ti, rex_function->getName(), arg_expr_list);
  }
  
  namespace {
  
  std::vector<Analyzer::OrderEntry> translate_collation(
      const std::vector<SortField>& sort_fields) {
    std::vector<Analyzer::OrderEntry> collation;
    for (size_t i = 0; i < sort_fields.size(); ++i) {
      const auto& sort_field = sort_fields[i];
      collation.emplace_back(i,
                             sort_field.getSortDir() == SortDirection::Descending,
                             sort_field.getNullsPosition() == NullSortedPosition::First);
    }
    return collation;
  }
  
  bool supported_lower_bound(
      const RexWindowFunctionOperator::RexWindowBound& window_bound) {
    return window_bound.unbounded && window_bound.preceding && !window_bound.following &&
           !window_bound.is_current_row && !window_bound.offset &&
           window_bound.order_key == 0;
  }
  
  bool supported_upper_bound(const RexWindowFunctionOperator* rex_window_function) {
    const auto& window_bound = rex_window_function->getUpperBound();
    const bool to_current_row = !window_bound.unbounded && !window_bound.preceding &&
                                !window_bound.following && window_bound.is_current_row &&
                                !window_bound.offset && window_bound.order_key == 1;
    switch (rex_window_function->getKind()) {
      case SqlWindowFunctionKind::ROW_NUMBER:
      case SqlWindowFunctionKind::RANK:
      case SqlWindowFunctionKind::DENSE_RANK:
      case SqlWindowFunctionKind::CUME_DIST: {
        return to_current_row;
      }
      default: {
        return rex_window_function->getOrderKeys().empty()
                   ? (window_bound.unbounded && !window_bound.preceding &&
                      window_bound.following && !window_bound.is_current_row &&
                      !window_bound.offset && window_bound.order_key == 2)
                   : to_current_row;
      }
    }
  }
  
  }  // namespace
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateWindowFunction(
      const RexWindowFunctionOperator* rex_window_function) const {
    if (!supported_lower_bound(rex_window_function->getLowerBound()) ||
        !supported_upper_bound(rex_window_function) ||
        ((rex_window_function->getKind() == SqlWindowFunctionKind::ROW_NUMBER) !=
         rex_window_function->isRows())) {
      throw std::runtime_error("Frame specification not supported");
    }
    std::vector<std::shared_ptr<Analyzer::Expr>> args;
    for (size_t i = 0; i < rex_window_function->size(); ++i) {
      args.push_back(translateScalarRex(rex_window_function->getOperand(i)));
    }
    std::vector<std::shared_ptr<Analyzer::Expr>> partition_keys;
    for (const auto& partition_key : rex_window_function->getPartitionKeys()) {
      partition_keys.push_back(translateScalarRex(partition_key.get()));
    }
    std::vector<std::shared_ptr<Analyzer::Expr>> order_keys;
    for (const auto& order_key : rex_window_function->getOrderKeys()) {
      order_keys.push_back(translateScalarRex(order_key.get()));
    }
    auto ti = rex_window_function->getType();
    if (window_function_is_value(rex_window_function->getKind())) {
      CHECK_GE(args.size(), 1u);
      ti = args.front()->get_type_info();
    }
    return makeExpr<Analyzer::WindowFunction>(
        ti,
        rex_window_function->getKind(),
        args,
        partition_keys,
        order_keys,
        translate_collation(rex_window_function->getCollation()));
  }
  
  Analyzer::ExpressionPtrVector RelAlgTranslator::translateFunctionArgs(
      const RexFunctionOperator* rex_function) const {
    std::vector<std::shared_ptr<Analyzer::Expr>> args;
    for (size_t i = 0; i < rex_function->size(); ++i) {
      args.push_back(translateScalarRex(rex_function->getOperand(i)));
    }
    return args;
  }
  
  QualsConjunctiveForm qual_to_conjunctive_form(
      const std::shared_ptr<Analyzer::Expr> qual_expr) {
    CHECK(qual_expr);
    auto bin_oper = std::dynamic_pointer_cast<const Analyzer::BinOper>(qual_expr);
    if (!bin_oper) {
      const auto rewritten_qual_expr = rewrite_expr(qual_expr.get());
      return {{}, {rewritten_qual_expr ? rewritten_qual_expr : qual_expr}};
    }
  
    if (bin_oper->get_optype() == kAND) {
      const auto lhs_cf = qual_to_conjunctive_form(bin_oper->get_own_left_operand());
      const auto rhs_cf = qual_to_conjunctive_form(bin_oper->get_own_right_operand());
      auto simple_quals = lhs_cf.simple_quals;
      simple_quals.insert(
          simple_quals.end(), rhs_cf.simple_quals.begin(), rhs_cf.simple_quals.end());
      auto quals = lhs_cf.quals;
      quals.insert(quals.end(), rhs_cf.quals.begin(), rhs_cf.quals.end());
      return {simple_quals, quals};
    }
    int rte_idx{0};
    const auto simple_qual = bin_oper->normalize_simple_predicate(rte_idx);
    return simple_qual ? QualsConjunctiveForm{{simple_qual}, {}}
                       : QualsConjunctiveForm{{}, {qual_expr}};
  }
  
  std::vector<std::shared_ptr<Analyzer::Expr>> qual_to_disjunctive_form(
      const std::shared_ptr<Analyzer::Expr>& qual_expr) {
    CHECK(qual_expr);
    const auto bin_oper = std::dynamic_pointer_cast<const Analyzer::BinOper>(qual_expr);
    if (!bin_oper) {
      const auto rewritten_qual_expr = rewrite_expr(qual_expr.get());
      return {rewritten_qual_expr ? rewritten_qual_expr : qual_expr};
    }
    if (bin_oper->get_optype() == kOR) {
      const auto lhs_df = qual_to_disjunctive_form(bin_oper->get_own_left_operand());
      const auto rhs_df = qual_to_disjunctive_form(bin_oper->get_own_right_operand());
      auto quals = lhs_df;
      quals.insert(quals.end(), rhs_df.begin(), rhs_df.end());
      return quals;
    }
    return {qual_expr};
  }
  
  std::shared_ptr<Analyzer::Expr> RelAlgTranslator::translateHPTLiteral(
      const RexFunctionOperator* rex_function) const {
    /* since calcite uses Avatica package called DateTimeUtils to parse timestamp strings.
       Therefore any string having fractional seconds more 3 places after the decimal
       (milliseconds) will get truncated to 3 decimal places, therefore we lose precision
       (us|ns). Issue: [BE-2461] Here we are hijacking literal cast to Timestamp(6|9) from
       calcite and translating them to generate our own casts.
    */
    CHECK_EQ(size_t(1), rex_function->size());
    const auto operand = translateScalarRex(rex_function->getOperand(0));
    const auto& operand_ti = operand->get_type_info();
    const auto& target_ti = rex_function->getType();
    if (!operand_ti.is_string()) {
      throw std::runtime_error(
          "High precision timestamp cast argument must be a string. Input type is: " +
          operand_ti.get_type_name());
    } else if (!target_ti.is_high_precision_timestamp()) {
      throw std::runtime_error(
          "Cast target type should be high precision timestamp. Input type is: " +
          target_ti.get_type_name());
    } else if (target_ti.get_dimension() != 6 && target_ti.get_dimension() != 9) {
      throw std::runtime_error(
          "Cast target type should be TIMESTAMP(6|9). Input type is: TIMESTAMP(" +
          std::to_string(target_ti.get_dimension()) + ")");
    } else {
      return operand->add_cast(target_ti);
    }
  }
  