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 "RelAlgExecutor.h"
  #include "DataMgr/ForeignStorage/ForeignStorageException.h"
  #include "DataMgr/ForeignStorage/MetadataPlaceholder.h"
  #include "Parser/ParserNode.h"
  #include "QueryEngine/CalciteDeserializerUtils.h"
  #include "QueryEngine/CardinalityEstimator.h"
  #include "QueryEngine/ColumnFetcher.h"
  #include "QueryEngine/EquiJoinCondition.h"
  #include "QueryEngine/ErrorHandling.h"
  #include "QueryEngine/ExpressionRewrite.h"
  #include "QueryEngine/ExtensionFunctionsBinding.h"
  #include "QueryEngine/ExternalExecutor.h"
  #include "QueryEngine/FromTableReordering.h"
  #include "QueryEngine/QueryPhysicalInputsCollector.h"
  #include "QueryEngine/RangeTableIndexVisitor.h"
  #include "QueryEngine/RelAlgDagBuilder.h"
  #include "QueryEngine/RelAlgTranslator.h"
  #include "QueryEngine/ResultSetBuilder.h"
  #include "QueryEngine/RexVisitor.h"
  #include "QueryEngine/TableOptimizer.h"
  #include "QueryEngine/WindowContext.h"
  #include "Shared/TypedDataAccessors.h"
  #include "Shared/measure.h"
  #include "Shared/misc.h"
  #include "Shared/shard_key.h"
  
  #include <boost/algorithm/cxx11/any_of.hpp>
  #include <boost/range/adaptor/reversed.hpp>
  
  #include <algorithm>
  #include <functional>
  #include <numeric>
  
  bool g_skip_intermediate_count{true};
  bool g_enable_interop{false};
  bool g_enable_union{false};
  
  extern bool g_enable_bump_allocator;
  
  namespace {
  
  bool node_is_aggregate(const RelAlgNode* ra) {
    const auto compound = dynamic_cast<const RelCompound*>(ra);
    const auto aggregate = dynamic_cast<const RelAggregate*>(ra);
    return ((compound && compound->isAggregate()) || aggregate);
  }
  
  std::unordered_set<PhysicalInput> get_physical_inputs(
      const Catalog_Namespace::Catalog& cat,
      const RelAlgNode* ra) {
    auto phys_inputs = get_physical_inputs(ra);
    std::unordered_set<PhysicalInput> phys_inputs2;
    for (auto& phi : phys_inputs) {
      phys_inputs2.insert(
          PhysicalInput{cat.getColumnIdBySpi(phi.table_id, phi.col_id), phi.table_id});
    }
    return phys_inputs2;
  }
  
  void set_parallelism_hints(const RelAlgNode& ra_node,
                             const Catalog_Namespace::Catalog& catalog) {
    std::map<ChunkKey, std::set<foreign_storage::ForeignStorageMgr::ParallelismHint>>
        parallelism_hints_per_table;
    for (const auto& physical_input : get_physical_inputs(&ra_node)) {
      int table_id = physical_input.table_id;
      auto table = catalog.getMetadataForTable(table_id, false);
      if (table && table->storageType == StorageType::FOREIGN_TABLE) {
        int col_id = catalog.getColumnIdBySpi(table_id, physical_input.col_id);
        const auto col_desc = catalog.getMetadataForColumn(table_id, col_id);
        auto foreign_table = catalog.getForeignTable(table_id);
        for (const auto& fragment :
             foreign_table->fragmenter->getFragmentsForQuery().fragments) {
          Chunk_NS::Chunk chunk{col_desc};
          ChunkKey chunk_key = {
              catalog.getDatabaseId(), table_id, col_id, fragment.fragmentId};
          // do not include chunk hints that are in CPU memory
          if (!chunk.isChunkOnDevice(
                  &catalog.getDataMgr(), chunk_key, Data_Namespace::CPU_LEVEL, 0)) {
            parallelism_hints_per_table[{catalog.getDatabaseId(), table_id}].insert(
                foreign_storage::ForeignStorageMgr::ParallelismHint{col_id,
                                                                    fragment.fragmentId});
          }
        }
      }
    }
    if (!parallelism_hints_per_table.empty()) {
      auto foreign_storage_mgr =
          catalog.getDataMgr().getPersistentStorageMgr()->getForeignStorageMgr();
      CHECK(foreign_storage_mgr);
      foreign_storage_mgr->setParallelismHints(parallelism_hints_per_table);
    }
  }
  
  void prepare_string_dictionaries(const RelAlgNode& ra_node,
                                   const Catalog_Namespace::Catalog& catalog) {
    for (const auto& physical_input : get_physical_inputs(&ra_node)) {
      int table_id = physical_input.table_id;
      auto table = catalog.getMetadataForTable(table_id, false);
      if (table && table->storageType == StorageType::FOREIGN_TABLE) {
        int col_id = catalog.getColumnIdBySpi(table_id, physical_input.col_id);
        const auto col_desc = catalog.getMetadataForColumn(table_id, col_id);
        auto foreign_table = catalog.getForeignTable(table_id);
        if (col_desc->columnType.is_dict_encoded_type()) {
          CHECK(foreign_table->fragmenter != nullptr);
          for (const auto& fragment :
               foreign_table->fragmenter->getFragmentsForQuery().fragments) {
            ChunkKey chunk_key = {
                catalog.getDatabaseId(), table_id, col_id, fragment.fragmentId};
            const ChunkMetadataMap& metadata_map = fragment.getChunkMetadataMap();
            CHECK(metadata_map.find(col_id) != metadata_map.end());
            if (foreign_storage::is_metadata_placeholder(*(metadata_map.at(col_id)))) {
              // When this goes out of scope it will stay in CPU cache but become
              // evictable
              std::shared_ptr<Chunk_NS::Chunk> chunk =
                  Chunk_NS::Chunk::getChunk(col_desc,
                                            &(catalog.getDataMgr()),
                                            chunk_key,
                                            Data_Namespace::CPU_LEVEL,
                                            0,
                                            0,
                                            0);
            }
          }
        }
      }
    }
  }
  
  void prepare_foreign_table_for_execution(const RelAlgNode& ra_node,
                                           const Catalog_Namespace::Catalog& catalog) {
    // Iterate through ra_node inputs for types that need to be loaded pre-execution
    // If they do not have valid metadata, load them into CPU memory to generate
    // the metadata and leave them ready to be used by the query
    set_parallelism_hints(ra_node, catalog);
    prepare_string_dictionaries(ra_node, catalog);
  }
  
  }  // namespace
  
  size_t RelAlgExecutor::getOuterFragmentCount(const CompilationOptions& co,
                                               const ExecutionOptions& eo) {
    if (eo.find_push_down_candidates) {
      return 0;
    }
  
    if (eo.just_explain) {
      return 0;
    }
  
    CHECK(query_dag_);
  
    query_dag_->resetQueryExecutionState();
    const auto& ra = query_dag_->getRootNode();
  
    auto lock = executor_->acquireExecuteMutex();
    ScopeGuard row_set_holder = [this] { cleanupPostExecution(); };
    const auto phys_inputs = get_physical_inputs(cat_, &ra);
    const auto phys_table_ids = get_physical_table_inputs(&ra);
    executor_->setCatalog(&cat_);
    executor_->setupCaching(phys_inputs, phys_table_ids);
  
    ScopeGuard restore_metainfo_cache = [this] { executor_->clearMetaInfoCache(); };
    auto ed_seq = RaExecutionSequence(&ra);
  
    if (!getSubqueries().empty()) {
      return 0;
    }
  
    CHECK(!ed_seq.empty());
    if (ed_seq.size() > 1) {
      return 0;
    }
  
    decltype(temporary_tables_)().swap(temporary_tables_);
    decltype(target_exprs_owned_)().swap(target_exprs_owned_);
    executor_->catalog_ = &cat_;
    executor_->temporary_tables_ = &temporary_tables_;
  
    WindowProjectNodeContext::reset(executor_);
    auto exec_desc_ptr = ed_seq.getDescriptor(0);
    CHECK(exec_desc_ptr);
    auto& exec_desc = *exec_desc_ptr;
    const auto body = exec_desc.getBody();
    if (body->isNop()) {
      return 0;
    }
  
    const auto project = dynamic_cast<const RelProject*>(body);
    if (project) {
      auto work_unit =
          createProjectWorkUnit(project, {{}, SortAlgorithm::Default, 0, 0}, eo);
  
      return get_frag_count_of_table(work_unit.exe_unit.input_descs[0].getTableId(),
                                     executor_);
    }
  
    const auto compound = dynamic_cast<const RelCompound*>(body);
    if (compound) {
      if (compound->isDeleteViaSelect()) {
        return 0;
      } else if (compound->isUpdateViaSelect()) {
        return 0;
      } else {
        if (compound->isAggregate()) {
          return 0;
        }
  
        const auto work_unit =
            createCompoundWorkUnit(compound, {{}, SortAlgorithm::Default, 0, 0}, eo);
  
        return get_frag_count_of_table(work_unit.exe_unit.input_descs[0].getTableId(),
                                       executor_);
      }
    }
  
    return 0;
  }
  
  ExecutionResult RelAlgExecutor::executeRelAlgQuery(const CompilationOptions& co,
                                                     const ExecutionOptions& eo,
                                                     const bool just_explain_plan,
                                                     RenderInfo* render_info) {
    CHECK(query_dag_);
    auto timer = DEBUG_TIMER(__func__);
    INJECT_TIMER(executeRelAlgQuery);
  
    auto run_query = [&](const CompilationOptions& co_in) {
      auto execution_result =
          executeRelAlgQueryNoRetry(co_in, eo, just_explain_plan, render_info);
      if (post_execution_callback_) {
        VLOG(1) << "Running post execution callback.";
        (*post_execution_callback_)();
      }
      return execution_result;
    };
  
    try {
      return run_query(co);
    } catch (const QueryMustRunOnCpu&) {
      if (!g_allow_cpu_retry) {
        throw;
      }
    }
    LOG(INFO) << "Query unable to run in GPU mode, retrying on CPU";
    auto co_cpu = CompilationOptions::makeCpuOnly(co);
  
    if (render_info) {
      render_info->setForceNonInSituData();
    }
    return run_query(co_cpu);
  }
  
  ExecutionResult RelAlgExecutor::executeRelAlgQueryNoRetry(const CompilationOptions& co,
                                                            const ExecutionOptions& eo,
                                                            const bool just_explain_plan,
                                                            RenderInfo* render_info) {
    INJECT_TIMER(executeRelAlgQueryNoRetry);
    auto timer = DEBUG_TIMER(__func__);
    auto timer_setup = DEBUG_TIMER("Query pre-execution steps");
  
    query_dag_->resetQueryExecutionState();
    const auto& ra = query_dag_->getRootNode();
  
    // capture the lock acquistion time
    auto clock_begin = timer_start();
    if (g_enable_dynamic_watchdog) {
      executor_->resetInterrupt();
    }
    std::string query_session{""};
    std::string query_str{"N/A"};
    std::string query_submitted_time{""};
    // gather necessary query's info
    if (query_state_ != nullptr && query_state_->getConstSessionInfo() != nullptr) {
      query_session = query_state_->getConstSessionInfo()->get_session_id();
      query_str = query_state_->getQueryStr();
      query_submitted_time = query_state_->getQuerySubmittedTime();
    }
  
    bool acquire_spin_lock = false;
    auto validate_or_explain_query =
        just_explain_plan || eo.just_validate || eo.just_explain || eo.just_calcite_explain;
    ScopeGuard clearRuntimeInterruptStatus = [this,
                                              &query_session,
                                              &render_info,
                                              &eo,
                                              &acquire_spin_lock,
                                              &query_submitted_time,
                                              &validate_or_explain_query] {
      // reset the runtime query interrupt status after the end of query execution
      if (!render_info && !query_session.empty() && eo.allow_runtime_query_interrupt &&
          !validate_or_explain_query) {
        executor_->clearQuerySessionStatus(
            query_session, query_submitted_time, acquire_spin_lock);
      }
    };
  
    if (!render_info && eo.allow_runtime_query_interrupt && !validate_or_explain_query) {
      // if we reach here, the current query which was waiting an idle executor
      // within the dispatch queue is now scheduled to the specific executor
      // (not UNITARY_EXECUTOR)
      // so we update the query session's status with the executor that takes this query
      std::tie(query_session, query_str) = executor_->attachExecutorToQuerySession(
          query_session, query_str, query_submitted_time);
  
      // For now we can run a single query at a time, so if other query is already
      // executed by different executor (i.e., when we turn on parallel executor),
      // we can check the possibility of running this query by using the spinlock.
      // If it fails to acquire a lock, it sleeps {g_pending_query_interrupt_freq} ms.
      // And then try to get the lock (and again and again...)
      if (!query_session.empty()) {
        while (executor_->execute_spin_lock_.test_and_set(std::memory_order_acquire)) {
          try {
            executor_->checkPendingQueryStatus(query_session);
          } catch (QueryExecutionError& e) {
            if (e.getErrorCode() == Executor::ERR_INTERRUPTED) {
              throw std::runtime_error(
                  "Query execution has been interrupted (pending query)");
            }
            throw e;
          } catch (...) {
            throw std::runtime_error("Checking pending query status failed: unknown error");
          }
          // here it fails to acquire the lock, so sleep...
          std::this_thread::sleep_for(
              std::chrono::milliseconds(g_pending_query_interrupt_freq));
        }
        acquire_spin_lock = true;
        // now the query is going to be executed, so update the status as "RUNNING"
        executor_->updateQuerySessionStatus(query_state_,
                                            QuerySessionStatus::QueryStatus::RUNNING);
      }
    }
    auto acquire_execute_mutex = [](Executor * executor) -> auto {
      auto ret = executor->acquireExecuteMutex();
      return ret;
    };
    // now we get the spinlock that means this query is ready to run by the executor
    // so we acquire executor lock in here to make sure that this executor holds
    // all necessary resources and at the same time protect them against other executor
    auto lock = acquire_execute_mutex(executor_);
  
    if (!render_info && !query_session.empty() && eo.allow_runtime_query_interrupt &&
        !validate_or_explain_query) {
      // check whether this query session is "already" interrupted
      // this case occurs when there is very short gap between being interrupted and
      // taking the execute lock
      // for instance, the session can fire multiple queries that other queries are waiting
      // in the spinlock but the user can request the query interruption
      // if so we have to remove "all" queries initiated by this session and we do in here
      // without running the query
      try {
        executor_->checkPendingQueryStatus(query_session);
      } catch (QueryExecutionError& e) {
        if (e.getErrorCode() == Executor::ERR_INTERRUPTED) {
          throw std::runtime_error("Query execution has been interrupted (pending query)");
        }
        throw e;
      } catch (...) {
        throw std::runtime_error("Checking pending query status failed: unknown error");
      }
    }
  
    // Notify foreign tables to load prior to caching
    prepare_foreign_table_for_execution(ra, cat_);
  
    int64_t queue_time_ms = timer_stop(clock_begin);
    ScopeGuard row_set_holder = [this] { cleanupPostExecution(); };
    const auto phys_inputs = get_physical_inputs(cat_, &ra);
    const auto phys_table_ids = get_physical_table_inputs(&ra);
    executor_->setCatalog(&cat_);
    executor_->setupCaching(phys_inputs, phys_table_ids);
  
    ScopeGuard restore_metainfo_cache = [this] { executor_->clearMetaInfoCache(); };
    auto ed_seq = RaExecutionSequence(&ra);
  
    if (just_explain_plan) {
      std::stringstream ss;
      std::vector<const RelAlgNode*> nodes;
      for (size_t i = 0; i < ed_seq.size(); i++) {
        nodes.emplace_back(ed_seq.getDescriptor(i)->getBody());
      }
      size_t ctr = nodes.size();
      size_t tab_ctr = 0;
      for (auto& body : boost::adaptors::reverse(nodes)) {
        const auto index = ctr--;
        const auto tabs = std::string(tab_ctr++, '\t');
        CHECK(body);
        ss << tabs << std::to_string(index) << " : " << body->toString() << "\n";
        if (auto sort = dynamic_cast<const RelSort*>(body)) {
          ss << tabs << "  : " << sort->getInput(0)->toString() << "\n";
        }
        if (dynamic_cast<const RelProject*>(body) ||
            dynamic_cast<const RelCompound*>(body)) {
          if (auto join = dynamic_cast<const RelLeftDeepInnerJoin*>(body->getInput(0))) {
            ss << tabs << "  : " << join->toString() << "\n";
          }
        }
      }
      const auto& subqueries = getSubqueries();
      if (!subqueries.empty()) {
        ss << "Subqueries: "
           << "\n";
        for (const auto& subquery : subqueries) {
          const auto ra = subquery->getRelAlg();
          ss << "\t" << ra->toString() << "\n";
        }
      }
      auto rs = std::make_shared<ResultSet>(ss.str());
      return {rs, {}};
    }
  
    if (render_info) {
      // set render to be non-insitu in certain situations.
      if (!render_info->disallow_in_situ_only_if_final_ED_is_aggregate &&
          ed_seq.size() > 1) {
        // old logic
        // disallow if more than one ED
        render_info->setInSituDataIfUnset(false);
      }
    }
  
    if (eo.find_push_down_candidates) {
      // this extra logic is mainly due to current limitations on multi-step queries
      // and/or subqueries.
      return executeRelAlgQueryWithFilterPushDown(
          ed_seq, co, eo, render_info, queue_time_ms);
    }
    timer_setup.stop();
  
    // Dispatch the subqueries first
    for (auto subquery : getSubqueries()) {
      const auto subquery_ra = subquery->getRelAlg();
      CHECK(subquery_ra);
      if (subquery_ra->hasContextData()) {
        continue;
      }
      // Execute the subquery and cache the result.
      RelAlgExecutor ra_executor(executor_, cat_, query_state_);
      RaExecutionSequence subquery_seq(subquery_ra);
      auto result = ra_executor.executeRelAlgSeq(subquery_seq, co, eo, nullptr, 0);
      subquery->setExecutionResult(std::make_shared<ExecutionResult>(result));
    }
    return executeRelAlgSeq(ed_seq, co, eo, render_info, queue_time_ms);
  }
  
  AggregatedColRange RelAlgExecutor::computeColRangesCache() {
    AggregatedColRange agg_col_range_cache;
    const auto phys_inputs = get_physical_inputs(cat_, &getRootRelAlgNode());
    return executor_->computeColRangesCache(phys_inputs);
  }
  
  StringDictionaryGenerations RelAlgExecutor::computeStringDictionaryGenerations() {
    const auto phys_inputs = get_physical_inputs(cat_, &getRootRelAlgNode());
    return executor_->computeStringDictionaryGenerations(phys_inputs);
  }
  
  TableGenerations RelAlgExecutor::computeTableGenerations() {
    const auto phys_table_ids = get_physical_table_inputs(&getRootRelAlgNode());
    return executor_->computeTableGenerations(phys_table_ids);
  }
  
  Executor* RelAlgExecutor::getExecutor() const {
    return executor_;
  }
  
  void RelAlgExecutor::cleanupPostExecution() {
    CHECK(executor_);
    executor_->row_set_mem_owner_ = nullptr;
  }
  
  namespace {
  
  inline void check_sort_node_source_constraint(const RelSort* sort) {
    CHECK_EQ(size_t(1), sort->inputCount());
    const auto source = sort->getInput(0);
    if (dynamic_cast<const RelSort*>(source)) {
      throw std::runtime_error("Sort node not supported as input to another sort");
    }
  }
  
  }  // namespace
  
  QueryStepExecutionResult RelAlgExecutor::executeRelAlgQuerySingleStep(
      const RaExecutionSequence& seq,
      const size_t step_idx,
      const CompilationOptions& co,
      const ExecutionOptions& eo,
      RenderInfo* render_info) {
    INJECT_TIMER(executeRelAlgQueryStep);
  
    auto exe_desc_ptr = seq.getDescriptor(step_idx);
    CHECK(exe_desc_ptr);
    const auto sort = dynamic_cast<const RelSort*>(exe_desc_ptr->getBody());
  
    size_t shard_count{0};
    auto merge_type = [&shard_count](const RelAlgNode* body) -> MergeType {
      return node_is_aggregate(body) && !shard_count ? MergeType::Reduce : MergeType::Union;
    };
  
    if (sort) {
      check_sort_node_source_constraint(sort);
      const auto source_work_unit = createSortInputWorkUnit(sort, eo);
      shard_count = GroupByAndAggregate::shard_count_for_top_groups(
          source_work_unit.exe_unit, *executor_->getCatalog());
      if (!shard_count) {
        // No point in sorting on the leaf, only execute the input to the sort node.
        CHECK_EQ(size_t(1), sort->inputCount());
        const auto source = sort->getInput(0);
        if (sort->collationCount() || node_is_aggregate(source)) {
          auto temp_seq = RaExecutionSequence(std::make_unique<RaExecutionDesc>(source));
          CHECK_EQ(temp_seq.size(), size_t(1));
          ExecutionOptions eo_copy = {
              eo.output_columnar_hint,
              eo.allow_multifrag,
              eo.just_explain,
              eo.allow_loop_joins,
              eo.with_watchdog,
              eo.jit_debug,
              eo.just_validate || sort->isEmptyResult(),
              eo.with_dynamic_watchdog,
              eo.dynamic_watchdog_time_limit,
              eo.find_push_down_candidates,
              eo.just_calcite_explain,
              eo.gpu_input_mem_limit_percent,
              eo.allow_runtime_query_interrupt,
              eo.running_query_interrupt_freq,
              eo.pending_query_interrupt_freq,
              eo.executor_type,
          };
          // Use subseq to avoid clearing existing temporary tables
          return {
              executeRelAlgSubSeq(temp_seq, std::make_pair(0, 1), co, eo_copy, nullptr, 0),
              merge_type(source),
              source->getId(),
              false};
        }
      }
    }
    QueryStepExecutionResult result{
        executeRelAlgSubSeq(seq,
                            std::make_pair(step_idx, step_idx + 1),
                            co,
                            eo,
                            render_info,
                            queue_time_ms_),
        merge_type(exe_desc_ptr->getBody()),
        exe_desc_ptr->getBody()->getId(),
        false};
    if (post_execution_callback_) {
      VLOG(1) << "Running post execution callback.";
      (*post_execution_callback_)();
    }
    return result;
  }
  
  void RelAlgExecutor::prepareLeafExecution(
      const AggregatedColRange& agg_col_range,
      const StringDictionaryGenerations& string_dictionary_generations,
      const TableGenerations& table_generations) {
    // capture the lock acquistion time
    auto clock_begin = timer_start();
    if (g_enable_dynamic_watchdog) {
      executor_->resetInterrupt();
    }
    queue_time_ms_ = timer_stop(clock_begin);
    executor_->row_set_mem_owner_ =
        std::make_shared<RowSetMemoryOwner>(Executor::getArenaBlockSize(), cpu_threads());
    executor_->row_set_mem_owner_->setDictionaryGenerations(string_dictionary_generations);
    executor_->table_generations_ = table_generations;
    executor_->agg_col_range_cache_ = agg_col_range;
  }
  
  ExecutionResult RelAlgExecutor::executeRelAlgSeq(const RaExecutionSequence& seq,
                                                   const CompilationOptions& co,
                                                   const ExecutionOptions& eo,
                                                   RenderInfo* render_info,
                                                   const int64_t queue_time_ms,
                                                   const bool with_existing_temp_tables) {
    INJECT_TIMER(executeRelAlgSeq);
    auto timer = DEBUG_TIMER(__func__);
    if (!with_existing_temp_tables) {
      decltype(temporary_tables_)().swap(temporary_tables_);
    }
    decltype(target_exprs_owned_)().swap(target_exprs_owned_);
    executor_->catalog_ = &cat_;
    executor_->temporary_tables_ = &temporary_tables_;
  
    time(&now_);
    CHECK(!seq.empty());
  
    auto get_descriptor_count = [&seq, &eo]() -> size_t {
      if (eo.just_explain) {
        if (dynamic_cast<const RelLogicalValues*>(seq.getDescriptor(0)->getBody())) {
          // run the logical values descriptor to generate the result set, then the next
          // descriptor to generate the explain
          CHECK_GE(seq.size(), size_t(2));
          return 2;
        } else {
          return 1;
        }
      } else {
        return seq.size();
      }
    };
  
    const auto exec_desc_count = get_descriptor_count();
  
    for (size_t i = 0; i < exec_desc_count; i++) {
      VLOG(1) << "Executing query step " << i;
      // only render on the last step
      try {
        executeRelAlgStep(seq,
                          i,
                          co,
                          eo,
                          (i == exec_desc_count - 1) ? render_info : nullptr,
                          queue_time_ms);
      } catch (const NativeExecutionError&) {
        if (!g_enable_interop) {
          throw;
        }
        auto eo_extern = eo;
        eo_extern.executor_type = ::ExecutorType::Extern;
        auto exec_desc_ptr = seq.getDescriptor(i);
        const auto body = exec_desc_ptr->getBody();
        const auto compound = dynamic_cast<const RelCompound*>(body);
        if (compound && (compound->getGroupByCount() || compound->isAggregate())) {
          LOG(INFO) << "Also failed to run the query using interoperability";
          throw;
        }
        executeRelAlgStep(seq,
                          i,
                          co,
                          eo_extern,
                          (i == exec_desc_count - 1) ? render_info : nullptr,
                          queue_time_ms);
      }
    }
  
    return seq.getDescriptor(exec_desc_count - 1)->getResult();
  }
  
  ExecutionResult RelAlgExecutor::executeRelAlgSubSeq(
      const RaExecutionSequence& seq,
      const std::pair<size_t, size_t> interval,
      const CompilationOptions& co,
      const ExecutionOptions& eo,
      RenderInfo* render_info,
      const int64_t queue_time_ms) {
    INJECT_TIMER(executeRelAlgSubSeq);
    executor_->catalog_ = &cat_;
    executor_->temporary_tables_ = &temporary_tables_;
  
    time(&now_);
    for (size_t i = interval.first; i < interval.second; i++) {
      // only render on the last step
      executeRelAlgStep(seq,
                        i,
                        co,
                        eo,
                        (i == interval.second - 1) ? render_info : nullptr,
                        queue_time_ms);
    }
  
    return seq.getDescriptor(interval.second - 1)->getResult();
  }
  
  void RelAlgExecutor::executeRelAlgStep(const RaExecutionSequence& seq,
                                         const size_t step_idx,
                                         const CompilationOptions& co,
                                         const ExecutionOptions& eo,
                                         RenderInfo* render_info,
                                         const int64_t queue_time_ms) {
    INJECT_TIMER(executeRelAlgStep);
    auto timer = DEBUG_TIMER(__func__);
    WindowProjectNodeContext::reset(executor_);
    auto exec_desc_ptr = seq.getDescriptor(step_idx);
    CHECK(exec_desc_ptr);
    auto& exec_desc = *exec_desc_ptr;
    const auto body = exec_desc.getBody();
    if (body->isNop()) {
      handleNop(exec_desc);
      return;
    }
    const ExecutionOptions eo_work_unit{
        eo.output_columnar_hint,
        eo.allow_multifrag,
        eo.just_explain,
        eo.allow_loop_joins,
        eo.with_watchdog && (step_idx == 0 || dynamic_cast<const RelProject*>(body)),
        eo.jit_debug,
        eo.just_validate,
        eo.with_dynamic_watchdog,
        eo.dynamic_watchdog_time_limit,
        eo.find_push_down_candidates,
        eo.just_calcite_explain,
        eo.gpu_input_mem_limit_percent,
        eo.allow_runtime_query_interrupt,
        eo.running_query_interrupt_freq,
        eo.pending_query_interrupt_freq,
        eo.executor_type,
        step_idx == 0 ? eo.outer_fragment_indices : std::vector<size_t>()};
  
    // Notify foreign tables to load prior to execution
    prepare_foreign_table_for_execution(*body, cat_);
  
    const auto compound = dynamic_cast<const RelCompound*>(body);
    if (compound) {
      if (compound->isDeleteViaSelect()) {
        executeDelete(compound, co, eo_work_unit, queue_time_ms);
      } else if (compound->isUpdateViaSelect()) {
        executeUpdate(compound, co, eo_work_unit, queue_time_ms);
      } else {
        exec_desc.setResult(
            executeCompound(compound, co, eo_work_unit, render_info, queue_time_ms));
        VLOG(3) << "Returned from executeCompound(), addTemporaryTable("
                << static_cast<int>(-compound->getId()) << ", ...)"
                << " exec_desc.getResult().getDataPtr()->rowCount()="
                << exec_desc.getResult().getDataPtr()->rowCount();
        if (exec_desc.getResult().isFilterPushDownEnabled()) {
          return;
        }
        addTemporaryTable(-compound->getId(), exec_desc.getResult().getDataPtr());
      }
      return;
    }
    const auto project = dynamic_cast<const RelProject*>(body);
    if (project) {
      if (project->isDeleteViaSelect()) {
        executeDelete(project, co, eo_work_unit, queue_time_ms);
      } else if (project->isUpdateViaSelect()) {
        executeUpdate(project, co, eo_work_unit, queue_time_ms);
      } else {
        std::optional<size_t> prev_count;
        // Disabling the intermediate count optimization in distributed, as the previous
        // execution descriptor will likely not hold the aggregated result.
        if (g_skip_intermediate_count && step_idx > 0 && !g_cluster) {
          auto prev_exec_desc = seq.getDescriptor(step_idx - 1);
          CHECK(prev_exec_desc);
          RelAlgNode const* prev_body = prev_exec_desc->getBody();
          // This optimization needs to be restricted in its application for UNION, which
          // can have 2 input nodes in which neither should restrict the count of the other.
          // However some non-UNION queries are measurably slower with this restriction, so
          // it is only applied when g_enable_union is true.
          bool const parent_check =
              !g_enable_union || project->getInput(0)->getId() == prev_body->getId();
          // If the previous node produced a reliable count, skip the pre-flight count
          if (parent_check && (dynamic_cast<const RelCompound*>(prev_body) ||
                               dynamic_cast<const RelLogicalValues*>(prev_body))) {
            const auto& prev_exe_result = prev_exec_desc->getResult();
            const auto prev_result = prev_exe_result.getRows();
            if (prev_result) {
              prev_count = prev_result->rowCount();
              VLOG(3) << "Setting output row count for projection node to previous node ("
                      << prev_exec_desc->getBody()->toString() << ") to " << *prev_count;
            }
          }
        }
        // For intermediate results we want to keep the result fragmented
        // to have higher parallelism on next steps.
        bool multifrag_result = g_enable_multifrag_rs && (step_idx != seq.size() - 1);
        exec_desc.setResult(
            executeProject(project,
                           co,
                           eo_work_unit.with_multifrag_result(multifrag_result),
                           render_info,
                           queue_time_ms,
                           prev_count));
        if (exec_desc.getResult().isFilterPushDownEnabled()) {
          return;
        }
        addTemporaryTable(-project->getId(), exec_desc.getResult().getTable());
      }
      return;
    }
    const auto aggregate = dynamic_cast<const RelAggregate*>(body);
    if (aggregate) {
      exec_desc.setResult(
          executeAggregate(aggregate, co, eo_work_unit, render_info, queue_time_ms));
      addTemporaryTable(-aggregate->getId(), exec_desc.getResult().getDataPtr());
      return;
    }
    const auto filter = dynamic_cast<const RelFilter*>(body);
    if (filter) {
      exec_desc.setResult(
          executeFilter(filter, co, eo_work_unit, render_info, queue_time_ms));
      addTemporaryTable(-filter->getId(), exec_desc.getResult().getDataPtr());
      return;
    }
    const auto sort = dynamic_cast<const RelSort*>(body);
    if (sort) {
      exec_desc.setResult(executeSort(sort, co, eo_work_unit, render_info, queue_time_ms));
      if (exec_desc.getResult().isFilterPushDownEnabled()) {
        return;
      }
      addTemporaryTable(-sort->getId(), exec_desc.getResult().getDataPtr());
      return;
    }
    const auto logical_values = dynamic_cast<const RelLogicalValues*>(body);
    if (logical_values) {
      exec_desc.setResult(executeLogicalValues(logical_values, eo_work_unit));
      addTemporaryTable(-logical_values->getId(), exec_desc.getResult().getDataPtr());
      return;
    }
    const auto modify = dynamic_cast<const RelModify*>(body);
    if (modify) {
      exec_desc.setResult(executeModify(modify, eo_work_unit));
      return;
    }
    const auto logical_union = dynamic_cast<const RelLogicalUnion*>(body);
    if (logical_union) {
      exec_desc.setResult(executeUnion(logical_union,
                                       seq,
                                       co,
                                       eo_work_unit.with_preserve_order(true),
                                       render_info,
                                       queue_time_ms));
      addTemporaryTable(-logical_union->getId(), exec_desc.getResult().getDataPtr());
      return;
    }
    const auto table_func = dynamic_cast<const RelTableFunction*>(body);
    if (table_func) {
      exec_desc.setResult(
          executeTableFunction(table_func, co, eo_work_unit, queue_time_ms));
      addTemporaryTable(-table_func->getId(), exec_desc.getResult().getDataPtr());
      return;
    }
    LOG(FATAL) << "Unhandled body type: " << body->toString();
  }
  
  void RelAlgExecutor::handleNop(RaExecutionDesc& ed) {
    // just set the result of the previous node as the result of no op
    auto body = ed.getBody();
    CHECK(dynamic_cast<const RelAggregate*>(body));
    CHECK_EQ(size_t(1), body->inputCount());
    const auto input = body->getInput(0);
    body->setOutputMetainfo(input->getOutputMetainfo());
    const auto it = temporary_tables_.find(-input->getId());
    CHECK(it != temporary_tables_.end());
  
    CHECK_EQ(it->second.getFragCount(), 1);
    ed.setResult({it->second.getResultSet(0), input->getOutputMetainfo()});
  
    // set up temp table as it could be used by the outer query or next step
    addTemporaryTable(-body->getId(), it->second);
  }
  
  namespace {
  
  class RexUsedInputsVisitor : public RexVisitor<std::unordered_set<const RexInput*>> {
   public:
    RexUsedInputsVisitor(const Catalog_Namespace::Catalog& cat) : RexVisitor(), cat_(cat) {}
  
    const std::vector<std::shared_ptr<RexInput>>& get_inputs_owned() const {
      return synthesized_physical_inputs_owned;
    }
  
    std::unordered_set<const RexInput*> visitInput(
        const RexInput* rex_input) const override {
      const auto input_ra = rex_input->getSourceNode();
      CHECK(input_ra);
      const auto scan_ra = dynamic_cast<const RelScan*>(input_ra);
      if (scan_ra) {
        const auto td = scan_ra->getTableDescriptor();
        if (td) {
          const auto col_id = rex_input->getIndex();
          const auto cd = cat_.getMetadataForColumnBySpi(td->tableId, col_id + 1);
          if (cd && cd->columnType.get_physical_cols() > 0) {
            CHECK(IS_GEO(cd->columnType.get_type()));
            std::unordered_set<const RexInput*> synthesized_physical_inputs;
            for (auto i = 0; i < cd->columnType.get_physical_cols(); i++) {
              auto physical_input =
                  new RexInput(scan_ra, SPIMAP_GEO_PHYSICAL_INPUT(col_id, i));
              synthesized_physical_inputs_owned.emplace_back(physical_input);
              synthesized_physical_inputs.insert(physical_input);
            }
            return synthesized_physical_inputs;
          }
        }
      }
      return {rex_input};
    }
  
   protected:
    std::unordered_set<const RexInput*> aggregateResult(
        const std::unordered_set<const RexInput*>& aggregate,
        const std::unordered_set<const RexInput*>& next_result) const override {
      auto result = aggregate;
      result.insert(next_result.begin(), next_result.end());
      return result;
    }
  
   private:
    mutable std::vector<std::shared_ptr<RexInput>> synthesized_physical_inputs_owned;
    const Catalog_Namespace::Catalog& cat_;
  };
  
  const RelAlgNode* get_data_sink(const RelAlgNode* ra_node) {
    if (auto table_func = dynamic_cast<const RelTableFunction*>(ra_node)) {
      return table_func;
    }
    if (auto join = dynamic_cast<const RelJoin*>(ra_node)) {
      CHECK_EQ(size_t(2), join->inputCount());
      return join;
    }
    if (!dynamic_cast<const RelLogicalUnion*>(ra_node)) {
      CHECK_EQ(size_t(1), ra_node->inputCount());
    }
    auto only_src = ra_node->getInput(0);
    const bool is_join = dynamic_cast<const RelJoin*>(only_src) ||
                         dynamic_cast<const RelLeftDeepInnerJoin*>(only_src);
    return is_join ? only_src : ra_node;
  }
  
  std::pair<std::unordered_set<const RexInput*>, std::vector<std::shared_ptr<RexInput>>>
  get_used_inputs(const RelCompound* compound, const Catalog_Namespace::Catalog& cat) {
    RexUsedInputsVisitor visitor(cat);
    const auto filter_expr = compound->getFilterExpr();
    std::unordered_set<const RexInput*> used_inputs =
        filter_expr ? visitor.visit(filter_expr) : std::unordered_set<const RexInput*>{};
    const auto sources_size = compound->getScalarSourcesSize();
    for (size_t i = 0; i < sources_size; ++i) {
      const auto source_inputs = visitor.visit(compound->getScalarSource(i));
      used_inputs.insert(source_inputs.begin(), source_inputs.end());
    }
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned(visitor.get_inputs_owned());
    return std::make_pair(used_inputs, used_inputs_owned);
  }
  
  std::pair<std::unordered_set<const RexInput*>, std::vector<std::shared_ptr<RexInput>>>
  get_used_inputs(const RelAggregate* aggregate, const Catalog_Namespace::Catalog& cat) {
    CHECK_EQ(size_t(1), aggregate->inputCount());
    std::unordered_set<const RexInput*> used_inputs;
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
    const auto source = aggregate->getInput(0);
    const auto& in_metainfo = source->getOutputMetainfo();
    const auto group_count = aggregate->getGroupByCount();
    CHECK_GE(in_metainfo.size(), group_count);
    for (size_t i = 0; i < group_count; ++i) {
      auto synthesized_used_input = new RexInput(source, i);
      used_inputs_owned.emplace_back(synthesized_used_input);
      used_inputs.insert(synthesized_used_input);
    }
    for (const auto& agg_expr : aggregate->getAggExprs()) {
      for (size_t i = 0; i < agg_expr->size(); ++i) {
        const auto operand_idx = agg_expr->getOperand(i);
        CHECK_GE(in_metainfo.size(), static_cast<size_t>(operand_idx));
        auto synthesized_used_input = new RexInput(source, operand_idx);
        used_inputs_owned.emplace_back(synthesized_used_input);
        used_inputs.insert(synthesized_used_input);
      }
    }
    return std::make_pair(used_inputs, used_inputs_owned);
  }
  
  std::pair<std::unordered_set<const RexInput*>, std::vector<std::shared_ptr<RexInput>>>
  get_used_inputs(const RelProject* project, const Catalog_Namespace::Catalog& cat) {
    RexUsedInputsVisitor visitor(cat);
    std::unordered_set<const RexInput*> used_inputs;
    for (size_t i = 0; i < project->size(); ++i) {
      const auto proj_inputs = visitor.visit(project->getProjectAt(i));
      used_inputs.insert(proj_inputs.begin(), proj_inputs.end());
    }
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned(visitor.get_inputs_owned());
    return std::make_pair(used_inputs, used_inputs_owned);
  }
  
  std::pair<std::unordered_set<const RexInput*>, std::vector<std::shared_ptr<RexInput>>>
  get_used_inputs(const RelTableFunction* table_func,
                  const Catalog_Namespace::Catalog& cat) {
    RexUsedInputsVisitor visitor(cat);
    std::unordered_set<const RexInput*> used_inputs;
    for (size_t i = 0; i < table_func->getTableFuncInputsSize(); ++i) {
      const auto table_func_inputs = visitor.visit(table_func->getTableFuncInputAt(i));
      used_inputs.insert(table_func_inputs.begin(), table_func_inputs.end());
    }
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned(visitor.get_inputs_owned());
    return std::make_pair(used_inputs, used_inputs_owned);
  }
  
  std::pair<std::unordered_set<const RexInput*>, std::vector<std::shared_ptr<RexInput>>>
  get_used_inputs(const RelFilter* filter, const Catalog_Namespace::Catalog& cat) {
    std::unordered_set<const RexInput*> used_inputs;
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
    const auto data_sink_node = get_data_sink(filter);
    for (size_t nest_level = 0; nest_level < data_sink_node->inputCount(); ++nest_level) {
      const auto source = data_sink_node->getInput(nest_level);
      const auto scan_source = dynamic_cast<const RelScan*>(source);
      if (scan_source) {
        CHECK(source->getOutputMetainfo().empty());
        for (size_t i = 0; i < scan_source->size(); ++i) {
          auto synthesized_used_input = new RexInput(scan_source, i);
          used_inputs_owned.emplace_back(synthesized_used_input);
          used_inputs.insert(synthesized_used_input);
        }
      } else {
        const auto& partial_in_metadata = source->getOutputMetainfo();
        for (size_t i = 0; i < partial_in_metadata.size(); ++i) {
          auto synthesized_used_input = new RexInput(source, i);
          used_inputs_owned.emplace_back(synthesized_used_input);
          used_inputs.insert(synthesized_used_input);
        }
      }
    }
    return std::make_pair(used_inputs, used_inputs_owned);
  }
  
  std::pair<std::unordered_set<const RexInput*>, std::vector<std::shared_ptr<RexInput>>>
  get_used_inputs(const RelLogicalUnion* logical_union, const Catalog_Namespace::Catalog&) {
    std::unordered_set<const RexInput*> used_inputs(logical_union->inputCount());
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
    used_inputs_owned.reserve(logical_union->inputCount());
    VLOG(3) << "logical_union->inputCount()=" << logical_union->inputCount();
    auto const n_inputs = logical_union->inputCount();
    for (size_t nest_level = 0; nest_level < n_inputs; ++nest_level) {
      auto input = logical_union->getInput(nest_level);
      for (size_t i = 0; i < input->size(); ++i) {
        used_inputs_owned.emplace_back(std::make_shared<RexInput>(input, i));
        used_inputs.insert(used_inputs_owned.back().get());
      }
    }
    return std::make_pair(std::move(used_inputs), std::move(used_inputs_owned));
  }
  
  int table_id_from_ra(const RelAlgNode* ra_node) {
    const auto scan_ra = dynamic_cast<const RelScan*>(ra_node);
    if (scan_ra) {
      const auto td = scan_ra->getTableDescriptor();
      CHECK(td);
      return td->tableId;
    }
    return -ra_node->getId();
  }
  
  std::unordered_map<const RelAlgNode*, int> get_input_nest_levels(
      const RelAlgNode* ra_node,
      const std::vector<size_t>& input_permutation) {
    const auto data_sink_node = get_data_sink(ra_node);
    std::unordered_map<const RelAlgNode*, int> input_to_nest_level;
    for (size_t input_idx = 0; input_idx < data_sink_node->inputCount(); ++input_idx) {
      const auto input_node_idx =
          input_permutation.empty() ? input_idx : input_permutation[input_idx];
      const auto input_ra = data_sink_node->getInput(input_node_idx);
      // Having a non-zero mapped value (input_idx) results in the query being interpretted
      // as a JOIN within CodeGenerator::codegenColVar() due to rte_idx being set to the
      // mapped value (input_idx) which originates here. This would be incorrect for UNION.
      size_t const idx = dynamic_cast<const RelLogicalUnion*>(ra_node) ? 0 : input_idx;
      const auto it_ok = input_to_nest_level.emplace(input_ra, idx);
      CHECK(it_ok.second);
      LOG_IF(INFO, !input_permutation.empty())
          << "Assigned input " << input_ra->toString() << " to nest level " << input_idx;
    }
    return input_to_nest_level;
  }
  
  std::pair<std::unordered_set<const RexInput*>, std::vector<std::shared_ptr<RexInput>>>
  get_join_source_used_inputs(const RelAlgNode* ra_node,
                              const Catalog_Namespace::Catalog& cat) {
    const auto data_sink_node = get_data_sink(ra_node);
    if (auto join = dynamic_cast<const RelJoin*>(data_sink_node)) {
      CHECK_EQ(join->inputCount(), 2u);
      const auto condition = join->getCondition();
      RexUsedInputsVisitor visitor(cat);
      auto condition_inputs = visitor.visit(condition);
      std::vector<std::shared_ptr<RexInput>> condition_inputs_owned(
          visitor.get_inputs_owned());
      return std::make_pair(condition_inputs, condition_inputs_owned);
    }
  
    if (auto left_deep_join = dynamic_cast<const RelLeftDeepInnerJoin*>(data_sink_node)) {
      CHECK_GE(left_deep_join->inputCount(), 2u);
      const auto condition = left_deep_join->getInnerCondition();
      RexUsedInputsVisitor visitor(cat);
      auto result = visitor.visit(condition);
      for (size_t nesting_level = 1; nesting_level <= left_deep_join->inputCount() - 1;
           ++nesting_level) {
        const auto outer_condition = left_deep_join->getOuterCondition(nesting_level);
        if (outer_condition) {
          const auto outer_result = visitor.visit(outer_condition);
          result.insert(outer_result.begin(), outer_result.end());
        }
      }
      std::vector<std::shared_ptr<RexInput>> used_inputs_owned(visitor.get_inputs_owned());
      return std::make_pair(result, used_inputs_owned);
    }
  
    if (dynamic_cast<const RelLogicalUnion*>(ra_node)) {
      CHECK_GT(ra_node->inputCount(), 1u) << ra_node->toString();
    } else if (dynamic_cast<const RelTableFunction*>(ra_node)) {
      CHECK_GT(ra_node->inputCount(), 0u) << ra_node->toString();
    } else {
      CHECK_EQ(ra_node->inputCount(), 1u) << ra_node->toString();
    }
    return std::make_pair(std::unordered_set<const RexInput*>{},
                          std::vector<std::shared_ptr<RexInput>>{});
  }
  
  void collect_used_input_desc(
      std::vector<InputDescriptor>& input_descs,
      const Catalog_Namespace::Catalog& cat,
      std::unordered_set<std::shared_ptr<const InputColDescriptor>>& input_col_descs_unique,
      const RelAlgNode* ra_node,
      const std::unordered_set<const RexInput*>& source_used_inputs,
      const std::unordered_map<const RelAlgNode*, int>& input_to_nest_level) {
    VLOG(3) << "ra_node=" << ra_node->toString()
            << " input_col_descs_unique.size()=" << input_col_descs_unique.size()
            << " source_used_inputs.size()=" << source_used_inputs.size();
    for (const auto used_input : source_used_inputs) {
      const auto input_ra = used_input->getSourceNode();
      const int table_id = table_id_from_ra(input_ra);
      const auto col_id = used_input->getIndex();
      auto it = input_to_nest_level.find(input_ra);
      if (it != input_to_nest_level.end()) {
        const int input_desc = it->second;
        input_col_descs_unique.insert(std::make_shared<const InputColDescriptor>(
            dynamic_cast<const RelScan*>(input_ra)
                ? cat.getColumnIdBySpi(table_id, col_id + 1)
                : col_id,
            table_id,
            input_desc));
      } else if (!dynamic_cast<const RelLogicalUnion*>(ra_node)) {
        throw std::runtime_error("Bushy joins not supported");
      }
    }
  }
  
  template <class RA>
  std::pair<std::vector<InputDescriptor>,
            std::list<std::shared_ptr<const InputColDescriptor>>>
  get_input_desc_impl(const RA* ra_node,
                      const std::unordered_set<const RexInput*>& used_inputs,
                      const std::unordered_map<const RelAlgNode*, int>& input_to_nest_level,
                      const std::vector<size_t>& input_permutation,
                      const Catalog_Namespace::Catalog& cat) {
    std::vector<InputDescriptor> input_descs;
    const auto data_sink_node = get_data_sink(ra_node);
    for (size_t input_idx = 0; input_idx < data_sink_node->inputCount(); ++input_idx) {
      const auto input_node_idx =
          input_permutation.empty() ? input_idx : input_permutation[input_idx];
      auto input_ra = data_sink_node->getInput(input_node_idx);
      const int table_id = table_id_from_ra(input_ra);
      input_descs.emplace_back(table_id, input_idx);
    }
    std::sort(input_descs.begin(),
              input_descs.end(),
              [](const InputDescriptor& lhs, const InputDescriptor& rhs) {
                return lhs.getNestLevel() < rhs.getNestLevel();
              });
    std::unordered_set<std::shared_ptr<const InputColDescriptor>> input_col_descs_unique;
    collect_used_input_desc(input_descs,
                            cat,
                            input_col_descs_unique,  // modified
                            ra_node,
                            used_inputs,
                            input_to_nest_level);
    std::unordered_set<const RexInput*> join_source_used_inputs;
    std::vector<std::shared_ptr<RexInput>> join_source_used_inputs_owned;
    std::tie(join_source_used_inputs, join_source_used_inputs_owned) =
        get_join_source_used_inputs(ra_node, cat);
    collect_used_input_desc(input_descs,
                            cat,
                            input_col_descs_unique,  // modified
                            ra_node,
                            join_source_used_inputs,
                            input_to_nest_level);
    std::vector<std::shared_ptr<const InputColDescriptor>> input_col_descs(
        input_col_descs_unique.begin(), input_col_descs_unique.end());
  
    std::sort(input_col_descs.begin(),
              input_col_descs.end(),
              [](std::shared_ptr<const InputColDescriptor> const& lhs,
                 std::shared_ptr<const InputColDescriptor> const& rhs) {
                return std::make_tuple(lhs->getScanDesc().getNestLevel(),
                                       lhs->getColId(),
                                       lhs->getScanDesc().getTableId()) <
                       std::make_tuple(rhs->getScanDesc().getNestLevel(),
                                       rhs->getColId(),
                                       rhs->getScanDesc().getTableId());
              });
    return {input_descs,
            std::list<std::shared_ptr<const InputColDescriptor>>(input_col_descs.begin(),
                                                                 input_col_descs.end())};
  }
  
  template <class RA>
  std::tuple<std::vector<InputDescriptor>,
             std::list<std::shared_ptr<const InputColDescriptor>>,
             std::vector<std::shared_ptr<RexInput>>>
  get_input_desc(const RA* ra_node,
                 const std::unordered_map<const RelAlgNode*, int>& input_to_nest_level,
                 const std::vector<size_t>& input_permutation,
                 const Catalog_Namespace::Catalog& cat) {
    std::unordered_set<const RexInput*> used_inputs;
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
    std::tie(used_inputs, used_inputs_owned) = get_used_inputs(ra_node, cat);
    VLOG(3) << "used_inputs.size() = " << used_inputs.size();
    auto input_desc_pair = get_input_desc_impl(
        ra_node, used_inputs, input_to_nest_level, input_permutation, cat);
    return std::make_tuple(
        input_desc_pair.first, input_desc_pair.second, used_inputs_owned);
  }
  
  size_t get_scalar_sources_size(const RelCompound* compound) {
    return compound->getScalarSourcesSize();
  }
  
  size_t get_scalar_sources_size(const RelProject* project) {
    return project->size();
  }
  
  size_t get_scalar_sources_size(const RelTableFunction* table_func) {
    return table_func->getTableFuncInputsSize();
  }
  
  const RexScalar* scalar_at(const size_t i, const RelCompound* compound) {
    return compound->getScalarSource(i);
  }
  
  const RexScalar* scalar_at(const size_t i, const RelProject* project) {
    return project->getProjectAt(i);
  }
  
  const RexScalar* scalar_at(const size_t i, const RelTableFunction* table_func) {
    return table_func->getTableFuncInputAt(i);
  }
  
  std::shared_ptr<Analyzer::Expr> set_transient_dict(
      const std::shared_ptr<Analyzer::Expr> expr) {
    const auto& ti = expr->get_type_info();
    if (!ti.is_string() || ti.get_compression() != kENCODING_NONE) {
      return expr;
    }
    auto transient_dict_ti = ti;
    transient_dict_ti.set_compression(kENCODING_DICT);
    transient_dict_ti.set_comp_param(TRANSIENT_DICT_ID);
    transient_dict_ti.set_fixed_size();
    return expr->add_cast(transient_dict_ti);
  }
  
  void set_transient_dict_maybe(
      std::vector<std::shared_ptr<Analyzer::Expr>>& scalar_sources,
      const std::shared_ptr<Analyzer::Expr>& expr) {
    try {
      scalar_sources.push_back(set_transient_dict(fold_expr(expr.get())));
    } catch (...) {
      scalar_sources.push_back(fold_expr(expr.get()));
    }
  }
  
  std::shared_ptr<Analyzer::Expr> cast_dict_to_none(
      const std::shared_ptr<Analyzer::Expr>& input) {
    const auto& input_ti = input->get_type_info();
    if (input_ti.is_string() && input_ti.get_compression() == kENCODING_DICT) {
      return input->add_cast(SQLTypeInfo(kTEXT, input_ti.get_notnull()));
    }
    return input;
  }
  
  template <class RA>
  std::vector<std::shared_ptr<Analyzer::Expr>> translate_scalar_sources(
      const RA* ra_node,
      const RelAlgTranslator& translator,
      const ::ExecutorType executor_type) {
    std::vector<std::shared_ptr<Analyzer::Expr>> scalar_sources;
    const size_t scalar_sources_size = get_scalar_sources_size(ra_node);
    VLOG(3) << "get_scalar_sources_size(" << ra_node->toString()
            << ") = " << scalar_sources_size;
    for (size_t i = 0; i < scalar_sources_size; ++i) {
      const auto scalar_rex = scalar_at(i, ra_node);
      if (dynamic_cast<const RexRef*>(scalar_rex)) {
        // RexRef are synthetic scalars we append at the end of the real ones
        // for the sake of taking memory ownership, no real work needed here.
        continue;
      }
  
      const auto scalar_expr =
          rewrite_array_elements(translator.translateScalarRex(scalar_rex).get());
      const auto rewritten_expr = rewrite_expr(scalar_expr.get());
      if (executor_type == ExecutorType::Native) {
        set_transient_dict_maybe(scalar_sources, rewritten_expr);
      } else {
        scalar_sources.push_back(cast_dict_to_none(fold_expr(rewritten_expr.get())));
      }
    }
  
    return scalar_sources;
  }
  
  template <class RA>
  std::vector<std::shared_ptr<Analyzer::Expr>> translate_scalar_sources_for_update(
      const RA* ra_node,
      const RelAlgTranslator& translator,
      int32_t tableId,
      const Catalog_Namespace::Catalog& cat,
      const ColumnNameList& colNames,
      size_t starting_projection_column_idx) {
    std::vector<std::shared_ptr<Analyzer::Expr>> scalar_sources;
    for (size_t i = 0; i < get_scalar_sources_size(ra_node); ++i) {
      const auto scalar_rex = scalar_at(i, ra_node);
      if (dynamic_cast<const RexRef*>(scalar_rex)) {
        // RexRef are synthetic scalars we append at the end of the real ones
        // for the sake of taking memory ownership, no real work needed here.
        continue;
      }
  
      std::shared_ptr<Analyzer::Expr> translated_expr;
      if (i >= starting_projection_column_idx && i < get_scalar_sources_size(ra_node) - 1) {
        translated_expr = cast_to_column_type(translator.translateScalarRex(scalar_rex),
                                              tableId,
                                              cat,
                                              colNames[i - starting_projection_column_idx]);
      } else {
        translated_expr = translator.translateScalarRex(scalar_rex);
      }
      const auto scalar_expr = rewrite_array_elements(translated_expr.get());
      const auto rewritten_expr = rewrite_expr(scalar_expr.get());
      set_transient_dict_maybe(scalar_sources, rewritten_expr);
    }
  
    return scalar_sources;
  }
  
  std::list<std::shared_ptr<Analyzer::Expr>> translate_groupby_exprs(
      const RelCompound* compound,
      const std::vector<std::shared_ptr<Analyzer::Expr>>& scalar_sources) {
    if (!compound->isAggregate()) {
      return {nullptr};
    }
    std::list<std::shared_ptr<Analyzer::Expr>> groupby_exprs;
    for (size_t group_idx = 0; group_idx < compound->getGroupByCount(); ++group_idx) {
      groupby_exprs.push_back(set_transient_dict(scalar_sources[group_idx]));
    }
    return groupby_exprs;
  }
  
  std::list<std::shared_ptr<Analyzer::Expr>> translate_groupby_exprs(
      const RelAggregate* aggregate,
      const std::vector<std::shared_ptr<Analyzer::Expr>>& scalar_sources) {
    std::list<std::shared_ptr<Analyzer::Expr>> groupby_exprs;
    for (size_t group_idx = 0; group_idx < aggregate->getGroupByCount(); ++group_idx) {
      groupby_exprs.push_back(set_transient_dict(scalar_sources[group_idx]));
    }
    return groupby_exprs;
  }
  
  QualsConjunctiveForm translate_quals(const RelCompound* compound,
                                       const RelAlgTranslator& translator) {
    const auto filter_rex = compound->getFilterExpr();
    const auto filter_expr =
        filter_rex ? translator.translateScalarRex(filter_rex) : nullptr;
    return filter_expr ? qual_to_conjunctive_form(fold_expr(filter_expr.get()))
                       : QualsConjunctiveForm{};
  }
  
  std::vector<Analyzer::Expr*> translate_targets(
      std::vector<std::shared_ptr<Analyzer::Expr>>& target_exprs_owned,
      const std::vector<std::shared_ptr<Analyzer::Expr>>& scalar_sources,
      const std::list<std::shared_ptr<Analyzer::Expr>>& groupby_exprs,
      const RelCompound* compound,
      const RelAlgTranslator& translator,
      const ExecutorType executor_type) {
    std::vector<Analyzer::Expr*> target_exprs;
    for (size_t i = 0; i < compound->size(); ++i) {
      const auto target_rex = compound->getTargetExpr(i);
      const auto target_rex_agg = dynamic_cast<const RexAgg*>(target_rex);
      std::shared_ptr<Analyzer::Expr> target_expr;
      if (target_rex_agg) {
        target_expr =
            RelAlgTranslator::translateAggregateRex(target_rex_agg, scalar_sources);
      } else {
        const auto target_rex_scalar = dynamic_cast<const RexScalar*>(target_rex);
        const auto target_rex_ref = dynamic_cast<const RexRef*>(target_rex_scalar);
        if (target_rex_ref) {
          const auto ref_idx = target_rex_ref->getIndex();
          CHECK_GE(ref_idx, size_t(1));
          CHECK_LE(ref_idx, groupby_exprs.size());
          const auto groupby_expr = *std::next(groupby_exprs.begin(), ref_idx - 1);
          target_expr = var_ref(groupby_expr.get(), Analyzer::Var::kGROUPBY, ref_idx);
        } else {
          target_expr = translator.translateScalarRex(target_rex_scalar);
          auto rewritten_expr = rewrite_expr(target_expr.get());
          target_expr = fold_expr(rewritten_expr.get());
          if (executor_type == ExecutorType::Native) {
            try {
              target_expr = set_transient_dict(target_expr);
            } catch (...) {
              // noop
            }
          } else {
            target_expr = cast_dict_to_none(target_expr);
          }
        }
      }
      CHECK(target_expr);
      target_exprs_owned.push_back(target_expr);
      target_exprs.push_back(target_expr.get());
    }
    return target_exprs;
  }
  
  std::vector<Analyzer::Expr*> translate_targets(
      std::vector<std::shared_ptr<Analyzer::Expr>>& target_exprs_owned,
      const std::vector<std::shared_ptr<Analyzer::Expr>>& scalar_sources,
      const std::list<std::shared_ptr<Analyzer::Expr>>& groupby_exprs,
      const RelAggregate* aggregate,
      const RelAlgTranslator& translator) {
    std::vector<Analyzer::Expr*> target_exprs;
    size_t group_key_idx = 1;
    for (const auto& groupby_expr : groupby_exprs) {
      auto target_expr =
          var_ref(groupby_expr.get(), Analyzer::Var::kGROUPBY, group_key_idx++);
      target_exprs_owned.push_back(target_expr);
      target_exprs.push_back(target_expr.get());
    }
  
    for (const auto& target_rex_agg : aggregate->getAggExprs()) {
      auto target_expr =
          RelAlgTranslator::translateAggregateRex(target_rex_agg.get(), scalar_sources);
      CHECK(target_expr);
      target_expr = fold_expr(target_expr.get());
      target_exprs_owned.push_back(target_expr);
      target_exprs.push_back(target_expr.get());
    }
    return target_exprs;
  }
  
  bool is_count_distinct(const Analyzer::Expr* expr) {
    const auto agg_expr = dynamic_cast<const Analyzer::AggExpr*>(expr);
    return agg_expr && agg_expr->get_is_distinct();
  }
  
  bool is_agg(const Analyzer::Expr* expr) {
    const auto agg_expr = dynamic_cast<const Analyzer::AggExpr*>(expr);
    if (agg_expr && agg_expr->get_contains_agg()) {
      auto agg_type = agg_expr->get_aggtype();
      if (agg_type == SQLAgg::kMIN || agg_type == SQLAgg::kMAX ||
          agg_type == SQLAgg::kSUM || agg_type == SQLAgg::kAVG) {
        return true;
      }
    }
    return false;
  }
  
  inline SQLTypeInfo get_logical_type_for_expr(const Analyzer::Expr& expr) {
    if (is_count_distinct(&expr)) {
      return SQLTypeInfo(kBIGINT, false);
    } else if (is_agg(&expr)) {
      return get_nullable_logical_type_info(expr.get_type_info());
    }
    return get_logical_type_info(expr.get_type_info());
  }
  
  template <class RA>
  std::vector<TargetMetaInfo> get_targets_meta(
      const RA* ra_node,
      const std::vector<Analyzer::Expr*>& target_exprs) {
    std::vector<TargetMetaInfo> targets_meta;
    CHECK_EQ(ra_node->size(), target_exprs.size());
    for (size_t i = 0; i < ra_node->size(); ++i) {
      CHECK(target_exprs[i]);
      // TODO(alex): remove the count distinct type fixup.
      targets_meta.emplace_back(ra_node->getFieldName(i),
                                get_logical_type_for_expr(*target_exprs[i]),
                                target_exprs[i]->get_type_info());
    }
    return targets_meta;
  }
  
  template <>
  std::vector<TargetMetaInfo> get_targets_meta(
      const RelFilter* filter,
      const std::vector<Analyzer::Expr*>& target_exprs) {
    RelAlgNode const* input0 = filter->getInput(0);
    if (auto const* input = dynamic_cast<RelCompound const*>(input0)) {
      return get_targets_meta(input, target_exprs);
    } else if (auto const* input = dynamic_cast<RelProject const*>(input0)) {
      return get_targets_meta(input, target_exprs);
    } else if (auto const* input = dynamic_cast<RelLogicalUnion const*>(input0)) {
      return get_targets_meta(input, target_exprs);
    } else if (auto const* input = dynamic_cast<RelAggregate const*>(input0)) {
      return get_targets_meta(input, target_exprs);
    } else if (auto const* input = dynamic_cast<RelScan const*>(input0)) {
      return get_targets_meta(input, target_exprs);
    }
    UNREACHABLE() << "Unhandled node type: " << input0->toString();
    return {};
  }
  
  }  // namespace
  
  void RelAlgExecutor::executeUpdate(const RelAlgNode* node,
                                     const CompilationOptions& co_in,
                                     const ExecutionOptions& eo_in,
                                     const int64_t queue_time_ms) {
    CHECK(node);
    auto timer = DEBUG_TIMER(__func__);
  
    UpdateTriggeredCacheInvalidator::invalidateCaches();
  
    auto co = co_in;
    co.hoist_literals = false;  // disable literal hoisting as it interferes with dict
                                // encoded string updates
  
    auto execute_update_for_node = [this, &co, &eo_in](const auto node,
                                                       auto& work_unit,
                                                       const bool is_aggregate) {
      dml_transaction_parameters_ =
          std::make_unique<UpdateTransactionParameters>(node->getModifiedTableDescriptor(),
                                                        node->getTargetColumns(),
                                                        node->getOutputMetainfo(),
                                                        node->isVarlenUpdateRequired());
  
      const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
  
      auto execute_update_ra_exe_unit = [this, &co, &eo_in, &table_infos](
                                            const RelAlgExecutionUnit& ra_exe_unit,
                                            const bool is_aggregate) {
        CompilationOptions co_project = CompilationOptions::makeCpuOnly(co);
  
        auto eo = eo_in;
        if (dml_transaction_parameters_->tableIsTemporary()) {
          eo.output_columnar_hint = true;
          co_project.allow_lazy_fetch = false;
          co_project.filter_on_deleted_column =
              false;  // project the entire delete column for columnar update
        }
  
        auto update_transaction_parameters =
            dynamic_cast<UpdateTransactionParameters*>(dml_transaction_parameters_.get());
        CHECK(update_transaction_parameters);
        auto update_callback = yieldUpdateCallback(*update_transaction_parameters);
        try {
          auto table_update_metadata =
              executor_->executeUpdate(ra_exe_unit,
                                       table_infos,
                                       co_project,
                                       eo,
                                       cat_,
                                       executor_->row_set_mem_owner_,
                                       update_callback,
                                       is_aggregate);
          post_execution_callback_ = [table_update_metadata, this]() {
            dml_transaction_parameters_->finalizeTransaction(cat_);
            TableOptimizer table_optimizer{
                dml_transaction_parameters_->getTableDescriptor(), executor_, cat_};
            table_optimizer.vacuumFragmentsAboveMinSelectivity(table_update_metadata);
          };
        } catch (const QueryExecutionError& e) {
          throw std::runtime_error(getErrorMessageFromCode(e.getErrorCode()));
        }
      };
  
      if (dml_transaction_parameters_->tableIsTemporary()) {
        // hold owned target exprs during execution if rewriting
        auto query_rewrite = std::make_unique<QueryRewriter>(table_infos, executor_);
        // rewrite temp table updates to generate the full column by moving the where
        // clause into a case if such a rewrite is not possible, bail on the update
        // operation build an expr for the update target
        auto update_transaction_params =
            dynamic_cast<UpdateTransactionParameters*>(dml_transaction_parameters_.get());
        CHECK(update_transaction_params);
        const auto td = update_transaction_params->getTableDescriptor();
        CHECK(td);
        const auto update_column_names = update_transaction_params->getUpdateColumnNames();
        if (update_column_names.size() > 1) {
          throw std::runtime_error(
              "Multi-column update is not yet supported for temporary tables.");
        }
  
        auto cd = cat_.getMetadataForColumn(td->tableId, update_column_names.front());
        CHECK(cd);
        auto projected_column_to_update =
            makeExpr<Analyzer::ColumnVar>(cd->columnType, td->tableId, cd->columnId, 0);
        const auto rewritten_exe_unit = query_rewrite->rewriteColumnarUpdate(
            work_unit.exe_unit, projected_column_to_update);
        if (rewritten_exe_unit.target_exprs.front()->get_type_info().is_varlen()) {
          throw std::runtime_error(
              "Variable length updates not yet supported on temporary tables.");
        }
        execute_update_ra_exe_unit(rewritten_exe_unit, is_aggregate);
      } else {
        execute_update_ra_exe_unit(work_unit.exe_unit, is_aggregate);
      }
    };
  
    if (auto compound = dynamic_cast<const RelCompound*>(node)) {
      auto work_unit =
          createCompoundWorkUnit(compound, {{}, SortAlgorithm::Default, 0, 0}, eo_in);
  
      execute_update_for_node(compound, work_unit, compound->isAggregate());
    } else if (auto project = dynamic_cast<const RelProject*>(node)) {
      auto work_unit =
          createProjectWorkUnit(project, {{}, SortAlgorithm::Default, 0, 0}, eo_in);
  
      if (project->isSimple()) {
        CHECK_EQ(size_t(1), project->inputCount());
        const auto input_ra = project->getInput(0);
        if (dynamic_cast<const RelSort*>(input_ra)) {
          const auto& input_table =
              get_temporary_table(&temporary_tables_, -input_ra->getId());
          work_unit.exe_unit.scan_limit = input_table.rowCount();
        }
      }
  
      execute_update_for_node(project, work_unit, false);
    } else {
      throw std::runtime_error("Unsupported parent node for update: " + node->toString());
    }
  }
  
  void RelAlgExecutor::executeDelete(const RelAlgNode* node,
                                     const CompilationOptions& co,
                                     const ExecutionOptions& eo_in,
                                     const int64_t queue_time_ms) {
    CHECK(node);
    auto timer = DEBUG_TIMER(__func__);
  
    DeleteTriggeredCacheInvalidator::invalidateCaches();
  
    auto execute_delete_for_node = [this, &co, &eo_in](const auto node,
                                                       auto& work_unit,
                                                       const bool is_aggregate) {
      auto* table_descriptor = node->getModifiedTableDescriptor();
      CHECK(table_descriptor);
      if (!table_descriptor->hasDeletedCol) {
        throw std::runtime_error(
            "DELETE only supported on tables with the vacuum attribute set to 'delayed'");
      }
  
      const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
  
      auto execute_delete_ra_exe_unit =
          [this, &table_infos, &table_descriptor, &eo_in, &co](const auto& exe_unit,
                                                               const bool is_aggregate) {
            dml_transaction_parameters_ =
                std::make_unique<DeleteTransactionParameters>(table_descriptor);
            auto delete_params = dynamic_cast<DeleteTransactionParameters*>(
                dml_transaction_parameters_.get());
            CHECK(delete_params);
            auto delete_callback = yieldDeleteCallback(*delete_params);
            CompilationOptions co_delete = CompilationOptions::makeCpuOnly(co);
  
            auto eo = eo_in;
            if (dml_transaction_parameters_->tableIsTemporary()) {
              eo.output_columnar_hint = true;
              co_delete.filter_on_deleted_column =
                  false;  // project the entire delete column for columnar update
            } else {
              CHECK_EQ(exe_unit.target_exprs.size(), size_t(1));
            }
  
            try {
              auto table_update_metadata =
                  executor_->executeUpdate(exe_unit,
                                           table_infos,
                                           co_delete,
                                           eo,
                                           cat_,
                                           executor_->row_set_mem_owner_,
                                           delete_callback,
                                           is_aggregate);
              post_execution_callback_ = [table_update_metadata, this]() {
                dml_transaction_parameters_->finalizeTransaction(cat_);
                TableOptimizer table_optimizer{
                    dml_transaction_parameters_->getTableDescriptor(), executor_, cat_};
                table_optimizer.vacuumFragmentsAboveMinSelectivity(table_update_metadata);
              };
            } catch (const QueryExecutionError& e) {
              throw std::runtime_error(getErrorMessageFromCode(e.getErrorCode()));
            }
          };
  
      if (table_is_temporary(table_descriptor)) {
        auto query_rewrite = std::make_unique<QueryRewriter>(table_infos, executor_);
        auto cd = cat_.getDeletedColumn(table_descriptor);
        CHECK(cd);
        auto delete_column_expr = makeExpr<Analyzer::ColumnVar>(
            cd->columnType, table_descriptor->tableId, cd->columnId, 0);
        const auto rewritten_exe_unit =
            query_rewrite->rewriteColumnarDelete(work_unit.exe_unit, delete_column_expr);
        execute_delete_ra_exe_unit(rewritten_exe_unit, is_aggregate);
      } else {
        execute_delete_ra_exe_unit(work_unit.exe_unit, is_aggregate);
      }
    };
  
    if (auto compound = dynamic_cast<const RelCompound*>(node)) {
      const auto work_unit =
          createCompoundWorkUnit(compound, {{}, SortAlgorithm::Default, 0, 0}, eo_in);
      execute_delete_for_node(compound, work_unit, compound->isAggregate());
    } else if (auto project = dynamic_cast<const RelProject*>(node)) {
      auto work_unit =
          createProjectWorkUnit(project, {{}, SortAlgorithm::Default, 0, 0}, eo_in);
      if (project->isSimple()) {
        CHECK_EQ(size_t(1), project->inputCount());
        const auto input_ra = project->getInput(0);
        if (dynamic_cast<const RelSort*>(input_ra)) {
          const auto& input_table =
              get_temporary_table(&temporary_tables_, -input_ra->getId());
          work_unit.exe_unit.scan_limit = input_table.rowCount();
        }
      }
      execute_delete_for_node(project, work_unit, false);
    } else {
      throw std::runtime_error("Unsupported parent node for delete: " + node->toString());
    }
  }
  
  ExecutionResult RelAlgExecutor::executeCompound(const RelCompound* compound,
                                                  const CompilationOptions& co,
                                                  const ExecutionOptions& eo,
                                                  RenderInfo* render_info,
                                                  const int64_t queue_time_ms) {
    auto timer = DEBUG_TIMER(__func__);
    const auto work_unit =
        createCompoundWorkUnit(compound, {{}, SortAlgorithm::Default, 0, 0}, eo);
    CompilationOptions co_compound = co;
    return executeWorkUnit(work_unit,
                           compound->getOutputMetainfo(),
                           compound->isAggregate(),
                           co_compound,
                           eo,
                           render_info,
                           queue_time_ms);
  }
  
  ExecutionResult RelAlgExecutor::executeAggregate(const RelAggregate* aggregate,
                                                   const CompilationOptions& co,
                                                   const ExecutionOptions& eo,
                                                   RenderInfo* render_info,
                                                   const int64_t queue_time_ms) {
    auto timer = DEBUG_TIMER(__func__);
    const auto work_unit = createAggregateWorkUnit(
        aggregate, {{}, SortAlgorithm::Default, 0, 0}, eo.just_explain);
    return executeWorkUnit(work_unit,
                           aggregate->getOutputMetainfo(),
                           true,
                           co,
                           eo,
                           render_info,
                           queue_time_ms);
  }
  
  namespace {
  
  // Returns true iff the execution unit contains window functions.
  bool is_window_execution_unit(const RelAlgExecutionUnit& ra_exe_unit) {
    return std::any_of(ra_exe_unit.target_exprs.begin(),
                       ra_exe_unit.target_exprs.end(),
                       [](const Analyzer::Expr* expr) {
                         return dynamic_cast<const Analyzer::WindowFunction*>(expr);
                       });
  }
  
  }  // namespace
  
  ExecutionResult RelAlgExecutor::executeProject(
      const RelProject* project,
      const CompilationOptions& co,
      const ExecutionOptions& eo,
      RenderInfo* render_info,
      const int64_t queue_time_ms,
      const std::optional<size_t> previous_count) {
    auto timer = DEBUG_TIMER(__func__);
    auto work_unit = createProjectWorkUnit(project, {{}, SortAlgorithm::Default, 0, 0}, eo);
    CompilationOptions co_project = co;
    if (project->isSimple()) {
      CHECK_EQ(size_t(1), project->inputCount());
      const auto input_ra = project->getInput(0);
      if (dynamic_cast<const RelSort*>(input_ra)) {
        co_project.device_type = ExecutorDeviceType::CPU;
        const auto& input_table =
            get_temporary_table(&temporary_tables_, -input_ra->getId());
        work_unit.exe_unit.scan_limit =
            std::min(input_table.getLimit(), input_table.rowCount());
      }
    }
    return executeWorkUnit(work_unit,
                           project->getOutputMetainfo(),
                           false,
                           co_project,
                           eo,
                           render_info,
                           queue_time_ms,
                           previous_count);
  }
  
  ExecutionResult RelAlgExecutor::executeTableFunction(const RelTableFunction* table_func,
                                                       const CompilationOptions& co_in,
                                                       const ExecutionOptions& eo,
                                                       const int64_t queue_time_ms) {
    INJECT_TIMER(executeTableFunction);
    auto timer = DEBUG_TIMER(__func__);
  
    auto co = co_in;
  
    if (g_cluster) {
      throw std::runtime_error("Table functions not supported in distributed mode yet");
    }
    if (!g_enable_table_functions) {
      throw std::runtime_error("Table function support is disabled");
    }
    auto table_func_work_unit = createTableFunctionWorkUnit(
        table_func,
        eo.just_explain,
        /*is_gpu = */ co.device_type == ExecutorDeviceType::GPU);
    const auto body = table_func_work_unit.body;
    CHECK(body);
  
    const auto table_infos =
        get_table_infos(table_func_work_unit.exe_unit.input_descs, executor_);
  
    ExecutionResult result{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
                                                       co.device_type,
                                                       QueryMemoryDescriptor(),
                                                       nullptr,
                                                       executor_->getCatalog(),
                                                       executor_->blockSize(),
                                                       executor_->gridSize()),
                           {}};
  
    try {
      result = {executor_->executeTableFunction(
                    table_func_work_unit.exe_unit, table_infos, co, eo, cat_),
                body->getOutputMetainfo()};
    } catch (const QueryExecutionError& e) {
      handlePersistentError(e.getErrorCode());
      CHECK(e.getErrorCode() == Executor::ERR_OUT_OF_GPU_MEM);
      throw std::runtime_error("Table function ran out of memory during execution");
    }
    result.setQueueTime(queue_time_ms);
    return result;
  }
  
  namespace {
  
  // Creates a new expression which has the range table index set to 1. This is needed to
  // reuse the hash join construction helpers to generate a hash table for the window
  // function partition: create an equals expression with left and right sides identical
  // except for the range table index.
  std::shared_ptr<Analyzer::Expr> transform_to_inner(const Analyzer::Expr* expr) {
    const auto tuple = dynamic_cast<const Analyzer::ExpressionTuple*>(expr);
    if (tuple) {
      std::vector<std::shared_ptr<Analyzer::Expr>> transformed_tuple;
      for (const auto& element : tuple->getTuple()) {
        transformed_tuple.push_back(transform_to_inner(element.get()));
      }
      return makeExpr<Analyzer::ExpressionTuple>(transformed_tuple);
    }
    const auto col = dynamic_cast<const Analyzer::ColumnVar*>(expr);
    if (!col) {
      throw std::runtime_error("Only columns supported in the window partition for now");
    }
    return makeExpr<Analyzer::ColumnVar>(
        col->get_type_info(), col->get_table_id(), col->get_column_id(), 1);
  }
  
  }  // namespace
  
  void RelAlgExecutor::computeWindow(const RelAlgExecutionUnit& ra_exe_unit,
                                     const CompilationOptions& co,
                                     const ExecutionOptions& eo,
                                     ColumnCacheMap& column_cache_map,
                                     const int64_t queue_time_ms) {
    auto query_infos = get_table_infos(ra_exe_unit.input_descs, executor_);
    CHECK_EQ(query_infos.size(), size_t(1));
    if (query_infos.front().info.fragments.size() != 1) {
      throw std::runtime_error(
          "Only single fragment tables supported for window functions for now");
    }
    if (eo.executor_type == ::ExecutorType::Extern) {
      return;
    }
    query_infos.push_back(query_infos.front());
    auto window_project_node_context = WindowProjectNodeContext::create(executor_);
    for (size_t target_index = 0; target_index < ra_exe_unit.target_exprs.size();
         ++target_index) {
      const auto& target_expr = ra_exe_unit.target_exprs[target_index];
      const auto window_func = dynamic_cast<const Analyzer::WindowFunction*>(target_expr);
      if (!window_func) {
        continue;
      }
      // Always use baseline layout hash tables for now, make the expression a tuple.
      const auto& partition_keys = window_func->getPartitionKeys();
      std::shared_ptr<Analyzer::Expr> partition_key_tuple;
      if (partition_keys.size() > 1) {
        partition_key_tuple = makeExpr<Analyzer::ExpressionTuple>(partition_keys);
      } else {
        if (partition_keys.empty()) {
          throw std::runtime_error(
              "Empty window function partitions are not supported yet");
        }
        CHECK_EQ(partition_keys.size(), size_t(1));
        partition_key_tuple = partition_keys.front();
      }
      // Creates a tautology equality with the partition expression on both sides.
      const auto partition_key_cond =
          makeExpr<Analyzer::BinOper>(kBOOLEAN,
                                      kBW_EQ,
                                      kONE,
                                      partition_key_tuple,
                                      transform_to_inner(partition_key_tuple.get()));
      auto context = createWindowFunctionContext(window_func,
                                                 partition_key_cond,
                                                 ra_exe_unit,
                                                 query_infos,
                                                 co,
                                                 column_cache_map,
                                                 executor_->getRowSetMemoryOwner());
      context->compute();
      window_project_node_context->addWindowFunctionContext(std::move(context),
                                                            target_index);
    }
  }
  
  std::unique_ptr<WindowFunctionContext> RelAlgExecutor::createWindowFunctionContext(
      const Analyzer::WindowFunction* window_func,
      const std::shared_ptr<Analyzer::BinOper>& partition_key_cond,
      const RelAlgExecutionUnit& ra_exe_unit,
      const std::vector<InputTableInfo>& query_infos,
      const CompilationOptions& co,
      ColumnCacheMap& column_cache_map,
      std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner) {
    const auto memory_level = co.device_type == ExecutorDeviceType::GPU
                                  ? MemoryLevel::GPU_LEVEL
                                  : MemoryLevel::CPU_LEVEL;
    const auto join_table_or_err =
        executor_->buildHashTableForQualifier(partition_key_cond,
                                              query_infos,
                                              memory_level,
                                              HashType::OneToMany,
                                              column_cache_map,
                                              ra_exe_unit.query_hint);
    if (!join_table_or_err.fail_reason.empty()) {
      throw std::runtime_error(join_table_or_err.fail_reason);
    }
    CHECK(join_table_or_err.hash_table->getHashType() == HashType::OneToMany);
    const auto& order_keys = window_func->getOrderKeys();
    std::vector<std::shared_ptr<Chunk_NS::Chunk>> chunks_owner;
    const size_t elem_count = query_infos.front().info.fragments.front().getNumTuples();
    auto context = std::make_unique<WindowFunctionContext>(window_func,
                                                           join_table_or_err.hash_table,
                                                           elem_count,
                                                           co.device_type,
                                                           row_set_mem_owner);
    for (const auto& order_key : order_keys) {
      const auto order_col =
          std::dynamic_pointer_cast<const Analyzer::ColumnVar>(order_key);
      if (!order_col) {
        throw std::runtime_error("Only order by columns supported for now");
      }
      const int8_t* column;
      size_t join_col_elem_count;
      std::tie(column, join_col_elem_count) =
          ColumnFetcher::getOneColumnFragment(executor_,
                                              *order_col,
                                              query_infos.front().info.fragments.front(),
                                              memory_level,
                                              0,
                                              nullptr,
                                              /*thread_idx=*/0,
                                              chunks_owner,
                                              column_cache_map);
      CHECK_EQ(join_col_elem_count, elem_count);
      context->addOrderColumn(column, order_col.get(), chunks_owner);
    }
    return context;
  }
  
  ExecutionResult RelAlgExecutor::executeFilter(const RelFilter* filter,
                                                const CompilationOptions& co,
                                                const ExecutionOptions& eo,
                                                RenderInfo* render_info,
                                                const int64_t queue_time_ms) {
    auto timer = DEBUG_TIMER(__func__);
    const auto work_unit =
        createFilterWorkUnit(filter, {{}, SortAlgorithm::Default, 0, 0}, eo.just_explain);
    return executeWorkUnit(
        work_unit, filter->getOutputMetainfo(), false, co, eo, render_info, queue_time_ms);
  }
  
  bool sameTypeInfo(std::vector<TargetMetaInfo> const& lhs,
                    std::vector<TargetMetaInfo> const& rhs) {
    if (lhs.size() == rhs.size()) {
      for (size_t i = 0; i < lhs.size(); ++i) {
        if (lhs[i].get_type_info() != rhs[i].get_type_info()) {
          return false;
        }
      }
      return true;
    }
    return false;
  }
  
  bool isGeometry(TargetMetaInfo const& target_meta_info) {
    return target_meta_info.get_type_info().is_geometry();
  }
  
  ExecutionResult RelAlgExecutor::executeUnion(const RelLogicalUnion* logical_union,
                                               const RaExecutionSequence& seq,
                                               const CompilationOptions& co,
                                               const ExecutionOptions& eo,
                                               RenderInfo* render_info,
                                               const int64_t queue_time_ms) {
    auto timer = DEBUG_TIMER(__func__);
    if (!logical_union->isAll()) {
      throw std::runtime_error("UNION without ALL is not supported yet.");
    }
    // Will throw a std::runtime_error if types don't match.
    logical_union->checkForMatchingMetaInfoTypes();
    logical_union->setOutputMetainfo(logical_union->getInput(0)->getOutputMetainfo());
    if (boost::algorithm::any_of(logical_union->getOutputMetainfo(), isGeometry)) {
      throw std::runtime_error("UNION does not support subqueries with geo-columns.");
    }
  #if 0  // We relax this for modin-omnisci mode
    // Only Projections and Aggregates from a UNION are supported for now.
    query_dag_->eachNode([logical_union](RelAlgNode const* node) {
     if (node->hasInput(logical_union) &&
        !shared::dynamic_castable_to_any<RelProject, RelLogicalUnion, RelAggregate>(
            node)) {
      throw std::runtime_error("UNION ALL not yet supported in this context.");
      }
    });
  #endif
  
    auto work_unit =
        createUnionWorkUnit(logical_union, {{}, SortAlgorithm::Default, 0, 0}, eo);
    return executeWorkUnit(work_unit,
                           logical_union->getOutputMetainfo(),
                           false,
                           CompilationOptions::makeCpuOnly(co),
                           eo,
                           render_info,
                           queue_time_ms);
  }
  
  ExecutionResult RelAlgExecutor::executeLogicalValues(
      const RelLogicalValues* logical_values,
      const ExecutionOptions& eo) {
    auto timer = DEBUG_TIMER(__func__);
    QueryMemoryDescriptor query_mem_desc(executor_,
                                         logical_values->getNumRows(),
                                         QueryDescriptionType::Projection,
                                         /*is_table_function=*/false);
  
    auto tuple_type = logical_values->getTupleType();
    for (size_t i = 0; i < tuple_type.size(); ++i) {
      auto& target_meta_info = tuple_type[i];
      if (target_meta_info.get_type_info().is_varlen()) {
        throw std::runtime_error("Variable length types not supported in VALUES yet.");
      }
      if (target_meta_info.get_type_info().get_type() == kNULLT) {
        // replace w/ bigint
        tuple_type[i] =
            TargetMetaInfo(target_meta_info.get_resname(), SQLTypeInfo(kBIGINT, false));
      }
      query_mem_desc.addColSlotInfo(
          {std::make_tuple(tuple_type[i].get_type_info().get_size(), 8)});
    }
    logical_values->setOutputMetainfo(tuple_type);
  
    std::vector<TargetInfo> target_infos;
    for (const auto& tuple_type_component : tuple_type) {
      target_infos.emplace_back(TargetInfo{false,
                                           kCOUNT,
                                           tuple_type_component.get_type_info(),
                                           SQLTypeInfo(kNULLT, false),
                                           false,
                                           false});
    }
  
    std::shared_ptr<ResultSet> rs{
        ResultSetLogicalValuesBuilder{logical_values,
                                      target_infos,
                                      ExecutorDeviceType::CPU,
                                      query_mem_desc,
                                      executor_->getRowSetMemoryOwner(),
                                      executor_}
            .build()};
  
    return {rs, tuple_type};
  }
  
  namespace {
  
  template <class T>
  int64_t insert_one_dict_str(T* col_data,
                              const std::string& columnName,
                              const SQLTypeInfo& columnType,
                              const Analyzer::Constant* col_cv,
                              const Catalog_Namespace::Catalog& catalog) {
    if (col_cv->get_is_null()) {
      *col_data = inline_fixed_encoding_null_val(columnType);
    } else {
      const int dict_id = columnType.get_comp_param();
      const auto col_datum = col_cv->get_constval();
      const auto& str = *col_datum.stringval;
      const auto dd = catalog.getMetadataForDict(dict_id);
      CHECK(dd && dd->stringDict);
      int32_t str_id = dd->stringDict->getOrAdd(str);
      if (!dd->dictIsTemp) {
        const auto checkpoint_ok = dd->stringDict->checkpoint();
        if (!checkpoint_ok) {
          throw std::runtime_error("Failed to checkpoint dictionary for column " +
                                   columnName);
        }
      }
      const bool invalid = str_id > max_valid_int_value<T>();
      if (invalid || str_id == inline_int_null_value<int32_t>()) {
        if (invalid) {
          LOG(ERROR) << "Could not encode string: " << str
                     << ", the encoded value doesn't fit in " << sizeof(T) * 8
                     << " bits. Will store NULL instead.";
        }
        str_id = inline_fixed_encoding_null_val(columnType);
      }
      *col_data = str_id;
    }
    return *col_data;
  }
  
  template <class T>
  int64_t insert_one_dict_str(T* col_data,
                              const ColumnDescriptor* cd,
                              const Analyzer::Constant* col_cv,
                              const Catalog_Namespace::Catalog& catalog) {
    return insert_one_dict_str(col_data, cd->columnName, cd->columnType, col_cv, catalog);
  }
  
  }  // namespace
  
  ExecutionResult RelAlgExecutor::executeModify(const RelModify* modify,
                                                const ExecutionOptions& eo) {
    auto timer = DEBUG_TIMER(__func__);
    if (eo.just_explain) {
      throw std::runtime_error("EXPLAIN not supported for ModifyTable");
    }
  
    auto rs = std::make_shared<ResultSet>(TargetInfoList{},
                                          ExecutorDeviceType::CPU,
                                          QueryMemoryDescriptor(),
                                          executor_->getRowSetMemoryOwner(),
                                          executor_->getCatalog(),
                                          executor_->blockSize(),
                                          executor_->gridSize());
  
    std::vector<TargetMetaInfo> empty_targets;
    return {rs, empty_targets};
  }
  
  namespace {
  int64_t int_value_from_numbers_ptr(const SQLTypeInfo& type_info, const int8_t* data) {
    size_t sz = 0;
    switch (type_info.get_type()) {
      case kTINYINT:
      case kSMALLINT:
      case kINT:
      case kBIGINT:
      case kTIMESTAMP:
      case kTIME:
      case kDATE:
        sz = type_info.get_logical_size();
        break;
      case kTEXT:
      case kVARCHAR:
      case kCHAR:
        CHECK(type_info.get_compression() == kENCODING_DICT);
        sz = type_info.get_size();
        break;
      default:
        CHECK(false) << "Unexpected sharding key datatype";
    }
  
    switch (sz) {
      case 1:
        return *(reinterpret_cast<const int8_t*>(data));
      case 2:
        return *(reinterpret_cast<const int16_t*>(data));
      case 4:
        return *(reinterpret_cast<const int32_t*>(data));
      case 8:
        return *(reinterpret_cast<const int64_t*>(data));
      default:
        CHECK(false);
        return 0;
    }
  }
  
  const TableDescriptor* get_shard_for_key(const TableDescriptor* td,
                                           const Catalog_Namespace::Catalog& cat,
                                           const Fragmenter_Namespace::InsertData& data) {
    auto shard_column_md = cat.getShardColumnMetadataForTable(td);
    CHECK(shard_column_md);
    auto sharded_column_id = shard_column_md->columnId;
    const TableDescriptor* shard{nullptr};
    for (size_t i = 0; i < data.columnIds.size(); ++i) {
      if (data.columnIds[i] == sharded_column_id) {
        const auto shard_tables = cat.getPhysicalTablesDescriptors(td);
        const auto shard_count = shard_tables.size();
        CHECK(data.data[i].numbersPtr);
        auto value = int_value_from_numbers_ptr(shard_column_md->columnType,
                                                data.data[i].numbersPtr);
        const size_t shard_idx = SHARD_FOR_KEY(value, shard_count);
        shard = shard_tables[shard_idx];
        break;
      }
    }
    return shard;
  }
  
  }  // namespace
  
  ExecutionResult RelAlgExecutor::executeSimpleInsert(const Analyzer::Query& query) {
    // Note: We currently obtain an executor for this method, but we do not need it.
    // Therefore, we skip the executor state setup in the regular execution path. In the
    // future, we will likely want to use the executor to evaluate expressions in the insert
    // statement.
  
    const auto& targets = query.get_targetlist();
    const int table_id = query.get_result_table_id();
    const auto& col_id_list = query.get_result_col_list();
  
    std::vector<const ColumnDescriptor*> col_descriptors;
    std::vector<int> col_ids;
    std::unordered_map<int, std::unique_ptr<uint8_t[]>> col_buffers;
    std::unordered_map<int, std::vector<std::string>> str_col_buffers;
    std::unordered_map<int, std::vector<ArrayDatum>> arr_col_buffers;
  
    for (const int col_id : col_id_list) {
      const auto cd = get_column_descriptor(col_id, table_id, cat_);
      const auto col_enc = cd->columnType.get_compression();
      if (cd->columnType.is_string()) {
        switch (col_enc) {
          case kENCODING_NONE: {
            auto it_ok =
                str_col_buffers.insert(std::make_pair(col_id, std::vector<std::string>{}));
            CHECK(it_ok.second);
            break;
          }
          case kENCODING_DICT: {
            const auto dd = cat_.getMetadataForDict(cd->columnType.get_comp_param());
            CHECK(dd);
            const auto it_ok = col_buffers.emplace(
                col_id, std::make_unique<uint8_t[]>(cd->columnType.get_size()));
            CHECK(it_ok.second);
            break;
          }
          default:
            CHECK(false);
        }
      } else if (cd->columnType.is_geometry()) {
        auto it_ok =
            str_col_buffers.insert(std::make_pair(col_id, std::vector<std::string>{}));
        CHECK(it_ok.second);
      } else if (cd->columnType.is_array()) {
        auto it_ok =
            arr_col_buffers.insert(std::make_pair(col_id, std::vector<ArrayDatum>{}));
        CHECK(it_ok.second);
      } else {
        const auto it_ok = col_buffers.emplace(
            col_id,
            std::unique_ptr<uint8_t[]>(
                new uint8_t[cd->columnType.get_logical_size()]()));  // changed to zero-init
                                                                     // the buffer
        CHECK(it_ok.second);
      }
      col_descriptors.push_back(cd);
      col_ids.push_back(col_id);
    }
    size_t col_idx = 0;
    Fragmenter_Namespace::InsertData insert_data;
    insert_data.databaseId = cat_.getCurrentDB().dbId;
    insert_data.tableId = table_id;
    for (auto target_entry : targets) {
      auto col_cv = dynamic_cast<const Analyzer::Constant*>(target_entry->get_expr());
      if (!col_cv) {
        auto col_cast = dynamic_cast<const Analyzer::UOper*>(target_entry->get_expr());
        CHECK(col_cast);
        CHECK_EQ(kCAST, col_cast->get_optype());
        col_cv = dynamic_cast<const Analyzer::Constant*>(col_cast->get_operand());
      }
      CHECK(col_cv);
      const auto cd = col_descriptors[col_idx];
      auto col_datum = col_cv->get_constval();
      auto col_type = cd->columnType.get_type();
      uint8_t* col_data_bytes{nullptr};
      if (!cd->columnType.is_array() && !cd->columnType.is_geometry() &&
          (!cd->columnType.is_string() ||
           cd->columnType.get_compression() == kENCODING_DICT)) {
        const auto col_data_bytes_it = col_buffers.find(col_ids[col_idx]);
        CHECK(col_data_bytes_it != col_buffers.end());
        col_data_bytes = col_data_bytes_it->second.get();
      }
      switch (col_type) {
        case kBOOLEAN: {
          auto col_data = col_data_bytes;
          *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
                                            : (col_datum.boolval ? 1 : 0);
          break;
        }
        case kTINYINT: {
          auto col_data = reinterpret_cast<int8_t*>(col_data_bytes);
          *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
                                            : col_datum.tinyintval;
          break;
        }
        case kSMALLINT: {
          auto col_data = reinterpret_cast<int16_t*>(col_data_bytes);
          *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
                                            : col_datum.smallintval;
          break;
        }
        case kINT: {
          auto col_data = reinterpret_cast<int32_t*>(col_data_bytes);
          *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
                                            : col_datum.intval;
          break;
        }
        case kBIGINT:
        case kDECIMAL:
        case kNUMERIC: {
          auto col_data = reinterpret_cast<int64_t*>(col_data_bytes);
          *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
                                            : col_datum.bigintval;
          break;
        }
        case kFLOAT: {
          auto col_data = reinterpret_cast<float*>(col_data_bytes);
          *col_data = col_datum.floatval;
          break;
        }
        case kDOUBLE: {
          auto col_data = reinterpret_cast<double*>(col_data_bytes);
          *col_data = col_datum.doubleval;
          break;
        }
        case kTEXT:
        case kVARCHAR:
        case kCHAR: {
          switch (cd->columnType.get_compression()) {
            case kENCODING_NONE:
              str_col_buffers[col_ids[col_idx]].push_back(
                  col_datum.stringval ? *col_datum.stringval : "");
              break;
            case kENCODING_DICT: {
              switch (cd->columnType.get_size()) {
                case 1:
                  insert_one_dict_str(
                      reinterpret_cast<uint8_t*>(col_data_bytes), cd, col_cv, cat_);
                  break;
                case 2:
                  insert_one_dict_str(
                      reinterpret_cast<uint16_t*>(col_data_bytes), cd, col_cv, cat_);
                  break;
                case 4:
                  insert_one_dict_str(
                      reinterpret_cast<int32_t*>(col_data_bytes), cd, col_cv, cat_);
                  break;
                default:
                  CHECK(false);
              }
              break;
            }
            default:
              CHECK(false);
          }
          break;
        }
        case kTIME:
        case kTIMESTAMP:
        case kDATE: {
          auto col_data = reinterpret_cast<int64_t*>(col_data_bytes);
          *col_data = col_cv->get_is_null() ? inline_fixed_encoding_null_val(cd->columnType)
                                            : col_datum.bigintval;
          break;
        }
        case kARRAY: {
          const auto is_null = col_cv->get_is_null();
          const auto size = cd->columnType.get_size();
          const SQLTypeInfo elem_ti = cd->columnType.get_elem_type();
          // POINT coords: [un]compressed coords always need to be encoded, even if NULL
          const auto is_point_coords = (cd->isGeoPhyCol && elem_ti.get_type() == kTINYINT);
          if (is_null && !is_point_coords) {
            if (size > 0) {
              // NULL fixlen array: NULL_ARRAY sentinel followed by NULL sentinels
              if (elem_ti.is_string() && elem_ti.get_compression() == kENCODING_DICT) {
                throw std::runtime_error("Column " + cd->columnName +
                                         " doesn't accept NULL values");
              }
              int8_t* buf = (int8_t*)checked_malloc(size);
              put_null_array(static_cast<void*>(buf), elem_ti, "");
              for (int8_t* p = buf + elem_ti.get_size(); (p - buf) < size;
                   p += elem_ti.get_size()) {
                put_null(static_cast<void*>(p), elem_ti, "");
              }
              arr_col_buffers[col_ids[col_idx]].emplace_back(size, buf, is_null);
            } else {
              arr_col_buffers[col_ids[col_idx]].emplace_back(0, nullptr, is_null);
            }
            break;
          }
          const auto l = col_cv->get_value_list();
          size_t len = l.size() * elem_ti.get_size();
          if (size > 0 && static_cast<size_t>(size) != len) {
            throw std::runtime_error("Array column " + cd->columnName + " expects " +
                                     std::to_string(size / elem_ti.get_size()) +
                                     " values, " + "received " + std::to_string(l.size()));
          }
          if (elem_ti.is_string()) {
            CHECK(kENCODING_DICT == elem_ti.get_compression());
            CHECK(4 == elem_ti.get_size());
  
            int8_t* buf = (int8_t*)checked_malloc(len);
            int32_t* p = reinterpret_cast<int32_t*>(buf);
  
            int elemIndex = 0;
            for (auto& e : l) {
              auto c = std::dynamic_pointer_cast<Analyzer::Constant>(e);
              CHECK(c);
              insert_one_dict_str(&p[elemIndex], cd->columnName, elem_ti, c.get(), cat_);
              elemIndex++;
            }
            arr_col_buffers[col_ids[col_idx]].push_back(ArrayDatum(len, buf, is_null));
  
          } else {
            int8_t* buf = (int8_t*)checked_malloc(len);
            int8_t* p = buf;
            for (auto& e : l) {
              auto c = std::dynamic_pointer_cast<Analyzer::Constant>(e);
              CHECK(c);
              p = appendDatum(p, c->get_constval(), elem_ti);
            }
            arr_col_buffers[col_ids[col_idx]].push_back(ArrayDatum(len, buf, is_null));
          }
          break;
        }
        case kPOINT:
        case kLINESTRING:
        case kPOLYGON:
        case kMULTIPOLYGON:
          str_col_buffers[col_ids[col_idx]].push_back(
              col_datum.stringval ? *col_datum.stringval : "");
          break;
        default:
          CHECK(false);
      }
      ++col_idx;
    }
    for (const auto& kv : col_buffers) {
      insert_data.columnIds.push_back(kv.first);
      DataBlockPtr p;
      p.numbersPtr = reinterpret_cast<int8_t*>(kv.second.get());
      insert_data.data.push_back(p);
    }
    for (auto& kv : str_col_buffers) {
      insert_data.columnIds.push_back(kv.first);
      DataBlockPtr p;
      p.stringsPtr = &kv.second;
      insert_data.data.push_back(p);
    }
    for (auto& kv : arr_col_buffers) {
      insert_data.columnIds.push_back(kv.first);
      DataBlockPtr p;
      p.arraysPtr = &kv.second;
      insert_data.data.push_back(p);
    }
    insert_data.numRows = 1;
    auto data_memory_holder = import_export::fill_missing_columns(&cat_, insert_data);
    const auto table_descriptor = cat_.getMetadataForTable(table_id);
    CHECK(table_descriptor);
    if (table_descriptor->nShards > 0) {
      auto shard = get_shard_for_key(table_descriptor, cat_, insert_data);
      CHECK(shard);
      shard->fragmenter->insertDataNoCheckpoint(insert_data);
    } else {
      table_descriptor->fragmenter->insertDataNoCheckpoint(insert_data);
    }
  
    // Ensure checkpoint happens across all shards, if not in distributed
    // mode (aggregator handles checkpointing in distributed mode)
    if (!g_cluster &&
        table_descriptor->persistenceLevel == Data_Namespace::MemoryLevel::DISK_LEVEL) {
      const_cast<Catalog_Namespace::Catalog&>(cat_).checkpointWithAutoRollback(table_id);
    }
  
    auto rs = std::make_shared<ResultSet>(TargetInfoList{},
                                          ExecutorDeviceType::CPU,
                                          QueryMemoryDescriptor(),
                                          executor_->getRowSetMemoryOwner(),
                                          nullptr,
                                          0,
                                          0);
    std::vector<TargetMetaInfo> empty_targets;
    return {rs, empty_targets};
  }
  
  namespace {
  
  // TODO(alex): Once we're fully migrated to the relational algebra model, change
  // the executor interface to use the collation directly and remove this conversion.
  std::list<Analyzer::OrderEntry> get_order_entries(const RelSort* sort) {
    std::list<Analyzer::OrderEntry> result;
    for (size_t i = 0; i < sort->collationCount(); ++i) {
      const auto sort_field = sort->getCollation(i);
      result.emplace_back(sort_field.getField() + 1,
                          sort_field.getSortDir() == SortDirection::Descending,
                          sort_field.getNullsPosition() == NullSortedPosition::First);
    }
    return result;
  }
  
  size_t get_scan_limit(const RelAlgNode* ra, const size_t limit) {
    const auto aggregate = dynamic_cast<const RelAggregate*>(ra);
    if (aggregate) {
      return 0;
    }
    const auto compound = dynamic_cast<const RelCompound*>(ra);
    return (compound && compound->isAggregate()) ? 0 : limit;
  }
  
  bool first_oe_is_desc(const std::list<Analyzer::OrderEntry>& order_entries) {
    return !order_entries.empty() && order_entries.front().is_desc;
  }
  
  }  // namespace
  
  ExecutionResult RelAlgExecutor::executeSort(const RelSort* sort,
                                              const CompilationOptions& co,
                                              const ExecutionOptions& eo,
                                              RenderInfo* render_info,
                                              const int64_t queue_time_ms) {
    auto timer = DEBUG_TIMER(__func__);
    check_sort_node_source_constraint(sort);
    const auto source = sort->getInput(0);
    const bool is_aggregate = node_is_aggregate(source);
    auto it = leaf_results_.find(sort->getId());
    if (it != leaf_results_.end()) {
      // Add any transient string literals to the sdp on the agg
      const auto source_work_unit = createSortInputWorkUnit(sort, eo);
      GroupByAndAggregate::addTransientStringLiterals(
          source_work_unit.exe_unit, executor_, executor_->row_set_mem_owner_);
      // Handle push-down for LIMIT for multi-node
      auto& aggregated_result = it->second;
      auto& result_rows = aggregated_result.rs;
      const size_t limit = sort->getLimit();
      const size_t offset = sort->getOffset();
      const auto order_entries = get_order_entries(sort);
      if (limit || offset) {
        if (!order_entries.empty()) {
          result_rows->sort(order_entries, limit + offset, executor_);
        }
        result_rows->dropFirstN(offset);
        if (limit) {
          result_rows->keepFirstN(limit);
        }
      }
      ExecutionResult result(result_rows, aggregated_result.targets_meta);
      sort->setOutputMetainfo(aggregated_result.targets_meta);
      return result;
    }
  
    std::list<std::shared_ptr<Analyzer::Expr>> groupby_exprs;
    bool is_desc{false};
  
    auto execute_sort_query = [this,
                               sort,
                               &source,
                               &is_aggregate,
                               &eo,
                               &co,
                               render_info,
                               queue_time_ms,
                               &groupby_exprs,
                               &is_desc]() -> ExecutionResult {
      const auto source_work_unit = createSortInputWorkUnit(sort, eo);
      is_desc = first_oe_is_desc(source_work_unit.exe_unit.sort_info.order_entries);
      ExecutionOptions eo_copy = {
          eo.output_columnar_hint,
          eo.allow_multifrag,
          eo.just_explain,
          eo.allow_loop_joins,
          eo.with_watchdog,
          eo.jit_debug,
          eo.just_validate || sort->isEmptyResult(),
          eo.with_dynamic_watchdog,
          eo.dynamic_watchdog_time_limit,
          eo.find_push_down_candidates,
          eo.just_calcite_explain,
          eo.gpu_input_mem_limit_percent,
          eo.allow_runtime_query_interrupt,
          eo.running_query_interrupt_freq,
          eo.pending_query_interrupt_freq,
          eo.executor_type,
      };
  
      groupby_exprs = source_work_unit.exe_unit.groupby_exprs;
      auto source_result = executeWorkUnit(source_work_unit,
                                           source->getOutputMetainfo(),
                                           is_aggregate,
                                           co,
                                           eo_copy,
                                           render_info,
                                           queue_time_ms);
      if (render_info && render_info->isPotentialInSituRender()) {
        return source_result;
      }
      if (source_result.isFilterPushDownEnabled()) {
        return source_result;
      }
      auto rows_to_sort = source_result.getRows();
      if (eo.just_explain) {
        return {rows_to_sort, {}};
      }
      const size_t limit = sort->getLimit();
      const size_t offset = sort->getOffset();
      if (sort->collationCount() != 0 && !rows_to_sort->definitelyHasNoRows() &&
          !use_speculative_top_n(source_work_unit.exe_unit,
                                 rows_to_sort->getQueryMemDesc())) {
        const size_t top_n = limit == 0 ? 0 : limit + offset;
        rows_to_sort->sort(
            source_work_unit.exe_unit.sort_info.order_entries, top_n, executor_);
      }
      if (limit || offset) {
        if (g_cluster && sort->collationCount() == 0) {
          if (offset >= rows_to_sort->rowCount()) {
            rows_to_sort->dropFirstN(offset);
          } else {
            rows_to_sort->keepFirstN(limit + offset);
          }
        } else {
          rows_to_sort->dropFirstN(offset);
          if (limit) {
            rows_to_sort->keepFirstN(limit);
          }
        }
      }
      return {rows_to_sort, source_result.getTargetsMeta()};
    };
  
    try {
      return execute_sort_query();
    } catch (const SpeculativeTopNFailed& e) {
      CHECK_EQ(size_t(1), groupby_exprs.size());
      CHECK(groupby_exprs.front());
      speculative_topn_blacklist_.add(groupby_exprs.front(), is_desc);
      return execute_sort_query();
    }
  }
  
  RelAlgExecutor::WorkUnit RelAlgExecutor::createSortInputWorkUnit(
      const RelSort* sort,
      const ExecutionOptions& eo) {
    const auto source = sort->getInput(0);
    const size_t limit = sort->getLimit();
    const size_t offset = sort->getOffset();
    const size_t scan_limit = sort->collationCount() ? 0 : get_scan_limit(source, limit);
    const size_t scan_total_limit =
        scan_limit ? get_scan_limit(source, scan_limit + offset) : 0;
    size_t max_groups_buffer_entry_guess{
        scan_total_limit ? scan_total_limit : max_groups_buffer_entry_default_guess};
    SortAlgorithm sort_algorithm{SortAlgorithm::SpeculativeTopN};
    const auto order_entries = get_order_entries(sort);
    SortInfo sort_info{order_entries, sort_algorithm, limit, offset};
    auto source_work_unit = createWorkUnit(source, sort_info, eo);
    const auto& source_exe_unit = source_work_unit.exe_unit;
  
    // we do not allow sorting geometry or array types
    for (auto order_entry : order_entries) {
      CHECK_GT(order_entry.tle_no, 0);  // tle_no is a 1-base index
      const auto& te = source_exe_unit.target_exprs[order_entry.tle_no - 1];
      const auto& ti = get_target_info(te, false);
      if (ti.sql_type.is_geometry() || ti.sql_type.is_array()) {
        throw std::runtime_error(
            "Columns with geometry or array types cannot be used in an ORDER BY clause.");
      }
    }
  
    if (source_exe_unit.groupby_exprs.size() == 1) {
      if (!source_exe_unit.groupby_exprs.front()) {
        sort_algorithm = SortAlgorithm::StreamingTopN;
      } else {
        if (speculative_topn_blacklist_.contains(source_exe_unit.groupby_exprs.front(),
                                                 first_oe_is_desc(order_entries))) {
          sort_algorithm = SortAlgorithm::Default;
        }
      }
    }
  
    sort->setOutputMetainfo(source->getOutputMetainfo());
    // NB: the `body` field of the returned `WorkUnit` needs to be the `source` node,
    // not the `sort`. The aggregator needs the pre-sorted result from leaves.
    return {RelAlgExecutionUnit{source_exe_unit.input_descs,
                                std::move(source_exe_unit.input_col_descs),
                                source_exe_unit.simple_quals,
                                source_exe_unit.quals,
                                source_exe_unit.join_quals,
                                source_exe_unit.groupby_exprs,
                                source_exe_unit.target_exprs,
                                nullptr,
                                {sort_info.order_entries, sort_algorithm, limit, offset},
                                scan_total_limit,
                                source_exe_unit.query_hint,
                                source_exe_unit.use_bump_allocator,
                                source_exe_unit.union_all,
                                source_exe_unit.query_state},
            source,
            max_groups_buffer_entry_guess,
            std::move(source_work_unit.query_rewriter),
            source_work_unit.input_permutation,
            source_work_unit.left_deep_join_input_sizes};
  }
  
  namespace {
  
  /**
   *  Upper bound estimation for the number of groups. Not strictly correct and not tight,
   * but if the tables involved are really small we shouldn't waste time doing the NDV
   * estimation. We don't account for cross-joins and / or group by unnested array, which
   * is the reason this estimation isn't entirely reliable.
   */
  size_t groups_approx_upper_bound(const std::vector<InputTableInfo>& table_infos) {
    CHECK(!table_infos.empty());
    const auto& first_table = table_infos.front();
    size_t max_num_groups = first_table.info.getNumTuplesUpperBound();
    for (const auto& table_info : table_infos) {
      if (table_info.info.getNumTuplesUpperBound() > max_num_groups) {
        max_num_groups = table_info.info.getNumTuplesUpperBound();
      }
    }
    return std::max(max_num_groups, size_t(1));
  }
  
  /**
   * Determines whether a query needs to compute the size of its output buffer. Returns
   * true for projection queries with no LIMIT or a LIMIT that exceeds the high scan limit
   * threshold (meaning it would be cheaper to compute the number of rows passing or use
   * the bump allocator than allocate the current scan limit per GPU)
   */
  bool compute_output_buffer_size(const RelAlgExecutionUnit& ra_exe_unit) {
    for (const auto target_expr : ra_exe_unit.target_exprs) {
      if (dynamic_cast<const Analyzer::AggExpr*>(target_expr)) {
        return false;
      }
    }
    if (ra_exe_unit.groupby_exprs.size() == 1 && !ra_exe_unit.groupby_exprs.front() &&
        (!ra_exe_unit.scan_limit || ra_exe_unit.scan_limit > Executor::high_scan_limit)) {
      return true;
    }
    return false;
  }
  
  inline bool exe_unit_has_quals(const RelAlgExecutionUnit ra_exe_unit) {
    return !(ra_exe_unit.quals.empty() && ra_exe_unit.join_quals.empty() &&
             ra_exe_unit.simple_quals.empty());
  }
  
  RelAlgExecutionUnit decide_approx_count_distinct_implementation(
      const RelAlgExecutionUnit& ra_exe_unit_in,
      const std::vector<InputTableInfo>& table_infos,
      const Executor* executor,
      const ExecutorDeviceType device_type_in,
      std::vector<std::shared_ptr<Analyzer::Expr>>& target_exprs_owned) {
    RelAlgExecutionUnit ra_exe_unit = ra_exe_unit_in;
    for (size_t i = 0; i < ra_exe_unit.target_exprs.size(); ++i) {
      const auto target_expr = ra_exe_unit.target_exprs[i];
      const auto agg_info = get_target_info(target_expr, g_bigint_count);
      if (agg_info.agg_kind != kAPPROX_COUNT_DISTINCT) {
        continue;
      }
      CHECK(dynamic_cast<const Analyzer::AggExpr*>(target_expr));
      const auto arg = static_cast<Analyzer::AggExpr*>(target_expr)->get_own_arg();
      CHECK(arg);
      const auto& arg_ti = arg->get_type_info();
      // Avoid calling getExpressionRange for variable length types (string and array),
      // it'd trigger an assertion since that API expects to be called only for types
      // for which the notion of range is well-defined. A bit of a kludge, but the
      // logic to reject these types anyway is at lower levels in the stack and not
      // really worth pulling into a separate function for now.
      if (!(arg_ti.is_number() || arg_ti.is_boolean() || arg_ti.is_time() ||
            (arg_ti.is_string() && arg_ti.get_compression() == kENCODING_DICT))) {
        continue;
      }
      const auto arg_range = getExpressionRange(arg.get(), table_infos, executor);
      if (arg_range.getType() != ExpressionRangeType::Integer) {
        continue;
      }
      // When running distributed, the threshold for using the precise implementation
      // must be consistent across all leaves, otherwise we could have a mix of precise
      // and approximate bitmaps and we cannot aggregate them.
      const auto device_type = g_cluster ? ExecutorDeviceType::GPU : device_type_in;
      const auto bitmap_sz_bits = arg_range.getIntMax() - arg_range.getIntMin() + 1;
      const auto sub_bitmap_count =
          get_count_distinct_sub_bitmap_count(bitmap_sz_bits, ra_exe_unit, device_type);
      int64_t approx_bitmap_sz_bits{0};
      const auto error_rate =
          static_cast<Analyzer::AggExpr*>(target_expr)->get_error_rate();
      if (error_rate) {
        CHECK(error_rate->get_type_info().get_type() == kINT);
        CHECK_GE(error_rate->get_constval().intval, 1);
        approx_bitmap_sz_bits = hll_size_for_rate(error_rate->get_constval().intval);
      } else {
        approx_bitmap_sz_bits = g_hll_precision_bits;
      }
      CountDistinctDescriptor approx_count_distinct_desc{CountDistinctImplType::Bitmap,
                                                         arg_range.getIntMin(),
                                                         approx_bitmap_sz_bits,
                                                         true,
                                                         device_type,
                                                         sub_bitmap_count};
      CountDistinctDescriptor precise_count_distinct_desc{CountDistinctImplType::Bitmap,
                                                          arg_range.getIntMin(),
                                                          bitmap_sz_bits,
                                                          false,
                                                          device_type,
                                                          sub_bitmap_count};
      if (approx_count_distinct_desc.bitmapPaddedSizeBytes() >=
          precise_count_distinct_desc.bitmapPaddedSizeBytes()) {
        auto precise_count_distinct = makeExpr<Analyzer::AggExpr>(
            get_agg_type(kCOUNT, arg.get()), kCOUNT, arg, true, nullptr);
        target_exprs_owned.push_back(precise_count_distinct);
        ra_exe_unit.target_exprs[i] = precise_count_distinct.get();
      }
    }
    return ra_exe_unit;
  }
  
  void build_render_targets(RenderInfo& render_info,
                            const std::vector<Analyzer::Expr*>& work_unit_target_exprs,
                            const std::vector<TargetMetaInfo>& targets_meta) {
    CHECK_EQ(work_unit_target_exprs.size(), targets_meta.size());
    render_info.targets.clear();
    for (size_t i = 0; i < targets_meta.size(); ++i) {
      render_info.targets.emplace_back(std::make_shared<Analyzer::TargetEntry>(
          targets_meta[i].get_resname(),
          work_unit_target_exprs[i]->get_shared_ptr(),
          false));
    }
  }
  
  inline bool can_use_bump_allocator(const RelAlgExecutionUnit& ra_exe_unit,
                                     const CompilationOptions& co,
                                     const ExecutionOptions& eo) {
    return g_enable_bump_allocator && (co.device_type == ExecutorDeviceType::GPU) &&
           !eo.output_columnar_hint && ra_exe_unit.sort_info.order_entries.empty();
  }
  
  }  // namespace
  
  ExecutionResult RelAlgExecutor::executeWorkUnit(
      const RelAlgExecutor::WorkUnit& work_unit,
      const std::vector<TargetMetaInfo>& targets_meta,
      const bool is_agg,
      const CompilationOptions& co_in,
      const ExecutionOptions& eo,
      RenderInfo* render_info,
      const int64_t queue_time_ms,
      const std::optional<size_t> previous_count) {
    INJECT_TIMER(executeWorkUnit);
    auto timer = DEBUG_TIMER(__func__);
  
    auto co = co_in;
    ColumnCacheMap column_cache;
    if (is_window_execution_unit(work_unit.exe_unit)) {
      if (!g_enable_window_functions) {
        throw std::runtime_error("Window functions support is disabled");
      }
      co.device_type = ExecutorDeviceType::CPU;
      co.allow_lazy_fetch = false;
      computeWindow(work_unit.exe_unit, co, eo, column_cache, queue_time_ms);
    }
    if (!eo.just_explain && eo.find_push_down_candidates) {
      // find potential candidates:
      auto selected_filters = selectFiltersToBePushedDown(work_unit, co, eo);
      if (!selected_filters.empty() || eo.just_calcite_explain) {
        return ExecutionResult(selected_filters, eo.find_push_down_candidates);
      }
    }
    if (render_info && render_info->isPotentialInSituRender()) {
      co.allow_lazy_fetch = false;
    }
    const auto body = work_unit.body;
    CHECK(body);
    auto it = leaf_results_.find(body->getId());
    VLOG(3) << "body->getId()=" << body->getId() << " body->toString()=" << body->toString()
            << " it==leaf_results_.end()=" << (it == leaf_results_.end());
    if (it != leaf_results_.end()) {
      GroupByAndAggregate::addTransientStringLiterals(
          work_unit.exe_unit, executor_, executor_->row_set_mem_owner_);
      auto& aggregated_result = it->second;
      auto& result_rows = aggregated_result.rs;
      ExecutionResult result(result_rows, aggregated_result.targets_meta);
      body->setOutputMetainfo(aggregated_result.targets_meta);
      if (render_info) {
        build_render_targets(*render_info, work_unit.exe_unit.target_exprs, targets_meta);
      }
      return result;
    }
    const auto table_infos = get_table_infos(work_unit.exe_unit, executor_);
  
    auto ra_exe_unit = decide_approx_count_distinct_implementation(
        work_unit.exe_unit, table_infos, executor_, co.device_type, target_exprs_owned_);
  
    // register query hint if query_dag_ is valid
    ra_exe_unit.query_hint =
        query_dag_ ? query_dag_->getQueryHints() : RegisteredQueryHint::defaults();
  
    auto max_groups_buffer_entry_guess = work_unit.max_groups_buffer_entry_guess;
    if (is_window_execution_unit(ra_exe_unit)) {
      CHECK_EQ(table_infos.size(), size_t(1));
      CHECK_EQ(table_infos.front().info.fragments.size(), size_t(1));
      max_groups_buffer_entry_guess =
          table_infos.front().info.fragments.front().getNumTuples();
      ra_exe_unit.scan_limit = max_groups_buffer_entry_guess;
    } else if (compute_output_buffer_size(ra_exe_unit) && !isRowidLookup(work_unit)) {
      if (previous_count && !exe_unit_has_quals(ra_exe_unit)) {
        ra_exe_unit.scan_limit = *previous_count;
      } else {
        // TODO(adb): enable bump allocator path for render queries
        if (can_use_bump_allocator(ra_exe_unit, co, eo) && !render_info) {
          ra_exe_unit.scan_limit = 0;
          ra_exe_unit.use_bump_allocator = true;
        } else if (eo.executor_type == ::ExecutorType::Extern) {
          ra_exe_unit.scan_limit = 0;
        } else if (!eo.just_explain) {
          const auto filter_count_all = getFilteredCountAll(work_unit, true, co, eo);
          if (filter_count_all) {
            ra_exe_unit.scan_limit = std::max(*filter_count_all, size_t(1));
          }
        }
      }
    }
    ExecutionResult result{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
                                                       co.device_type,
                                                       QueryMemoryDescriptor(),
                                                       nullptr,
                                                       executor_->getCatalog(),
                                                       executor_->blockSize(),
                                                       executor_->gridSize()),
                           {}};
  
    auto execute_and_handle_errors =
        [&](const auto max_groups_buffer_entry_guess_in,
            const bool has_cardinality_estimation) -> ExecutionResult {
      // Note that the groups buffer entry guess may be modified during query execution.
      // Create a local copy so we can track those changes if we need to attempt a retry
      // due to OOM
      auto local_groups_buffer_entry_guess = max_groups_buffer_entry_guess_in;
      try {
        return {executor_->executeWorkUnit(local_groups_buffer_entry_guess,
                                           is_agg,
                                           table_infos,
                                           ra_exe_unit,
                                           co,
                                           eo,
                                           cat_,
                                           render_info,
                                           has_cardinality_estimation,
                                           column_cache),
                targets_meta};
      } catch (const QueryExecutionError& e) {
        handlePersistentError(e.getErrorCode());
        return handleOutOfMemoryRetry(
            {ra_exe_unit, work_unit.body, local_groups_buffer_entry_guess},
            targets_meta,
            is_agg,
            co,
            eo,
            render_info,
            e.wasMultifragKernelLaunch(),
            queue_time_ms);
      }
    };
  
    auto cache_key = ra_exec_unit_desc_for_caching(ra_exe_unit);
    try {
      auto cached_cardinality = executor_->getCachedCardinality(cache_key);
      auto card = cached_cardinality.second;
      if (cached_cardinality.first && card >= 0) {
        result = execute_and_handle_errors(card, true);
      } else {
        result = execute_and_handle_errors(
            max_groups_buffer_entry_guess,
            groups_approx_upper_bound(table_infos) <= g_big_group_threshold);
      }
    } catch (const CardinalityEstimationRequired& e) {
      // check the cardinality cache
      auto cached_cardinality = executor_->getCachedCardinality(cache_key);
      auto card = cached_cardinality.second;
      if (cached_cardinality.first && card >= 0) {
        result = execute_and_handle_errors(card, true);
      } else {
        const auto estimated_groups_buffer_entry_guess =
            2 * std::min(groups_approx_upper_bound(table_infos),
                         getNDVEstimation(work_unit, e.range(), is_agg, co, eo));
        CHECK_GT(estimated_groups_buffer_entry_guess, size_t(0));
        result = execute_and_handle_errors(estimated_groups_buffer_entry_guess, true);
        if (!(eo.just_validate || eo.just_explain)) {
          executor_->addToCardinalityCache(cache_key, estimated_groups_buffer_entry_guess);
        }
      }
    }
  
    result.setQueueTime(queue_time_ms);
    if (render_info) {
      build_render_targets(*render_info, work_unit.exe_unit.target_exprs, targets_meta);
      if (render_info->isPotentialInSituRender()) {
        // return an empty result (with the same queue time, and zero render time)
        return {std::make_shared<ResultSet>(
                    queue_time_ms,
                    0,
                    executor_->row_set_mem_owner_
                        ? executor_->row_set_mem_owner_->cloneStrDictDataOnly()
                        : nullptr),
                {}};
      }
    }
    return result;
  }
  
  std::optional<size_t> RelAlgExecutor::getFilteredCountAll(const WorkUnit& work_unit,
                                                            const bool is_agg,
                                                            const CompilationOptions& co,
                                                            const ExecutionOptions& eo) {
    const auto count =
        makeExpr<Analyzer::AggExpr>(SQLTypeInfo(g_bigint_count ? kBIGINT : kINT, false),
                                    kCOUNT,
                                    nullptr,
                                    false,
                                    nullptr);
    const auto count_all_exe_unit =
        create_count_all_execution_unit(work_unit.exe_unit, count);
    size_t one{1};
    TemporaryTable count_all_result;
    try {
      ColumnCacheMap column_cache;
      count_all_result =
          executor_->executeWorkUnit(one,
                                     is_agg,
                                     get_table_infos(work_unit.exe_unit, executor_),
                                     count_all_exe_unit,
                                     co,
                                     eo,
                                     cat_,
                                     nullptr,
                                     false,
                                     column_cache);
    } catch (const foreign_storage::ForeignStorageException& error) {
      throw error;
    } catch (const QueryMustRunOnCpu&) {
      // force a retry of the top level query on CPU
      throw;
    } catch (const std::exception& e) {
      LOG(WARNING) << "Failed to run pre-flight filtered count with error " << e.what();
      return std::nullopt;
    }
    CHECK_EQ(count_all_result.getFragCount(), 1);
    const auto count_row = count_all_result[0]->getNextRow(false, false);
    CHECK_EQ(size_t(1), count_row.size());
    const auto& count_tv = count_row.front();
    const auto count_scalar_tv = boost::get<ScalarTargetValue>(&count_tv);
    CHECK(count_scalar_tv);
    const auto count_ptr = boost::get<int64_t>(count_scalar_tv);
    CHECK(count_ptr);
    CHECK_GE(*count_ptr, 0);
    auto count_upper_bound = static_cast<size_t>(*count_ptr);
    return std::max(count_upper_bound, size_t(1));
  }
  
  bool RelAlgExecutor::isRowidLookup(const WorkUnit& work_unit) {
    const auto& ra_exe_unit = work_unit.exe_unit;
    if (ra_exe_unit.input_descs.size() != 1) {
      return false;
    }
    const auto& table_desc = ra_exe_unit.input_descs.front();
    if (table_desc.getSourceType() != InputSourceType::TABLE) {
      return false;
    }
    const int table_id = table_desc.getTableId();
    for (const auto& simple_qual : ra_exe_unit.simple_quals) {
      const auto comp_expr =
          std::dynamic_pointer_cast<const Analyzer::BinOper>(simple_qual);
      if (!comp_expr || comp_expr->get_optype() != kEQ) {
        return false;
      }
      const auto lhs = comp_expr->get_left_operand();
      const auto lhs_col = dynamic_cast<const Analyzer::ColumnVar*>(lhs);
      if (!lhs_col || !lhs_col->get_table_id() || lhs_col->get_rte_idx()) {
        return false;
      }
      const auto rhs = comp_expr->get_right_operand();
      const auto rhs_const = dynamic_cast<const Analyzer::Constant*>(rhs);
      if (!rhs_const) {
        return false;
      }
      auto cd = get_column_descriptor(lhs_col->get_column_id(), table_id, cat_);
      if (cd->isVirtualCol) {
        CHECK_EQ("rowid", cd->columnName);
        return true;
      }
    }
    return false;
  }
  
  ExecutionResult RelAlgExecutor::handleOutOfMemoryRetry(
      const RelAlgExecutor::WorkUnit& work_unit,
      const std::vector<TargetMetaInfo>& targets_meta,
      const bool is_agg,
      const CompilationOptions& co,
      const ExecutionOptions& eo,
      RenderInfo* render_info,
      const bool was_multifrag_kernel_launch,
      const int64_t queue_time_ms) {
    // Disable the bump allocator
    // Note that this will have basically the same affect as using the bump allocator for
    // the kernel per fragment path. Need to unify the max_groups_buffer_entry_guess = 0
    // path and the bump allocator path for kernel per fragment execution.
    auto ra_exe_unit_in = work_unit.exe_unit;
    ra_exe_unit_in.use_bump_allocator = false;
  
    auto result = ExecutionResult{std::make_shared<ResultSet>(std::vector<TargetInfo>{},
                                                              co.device_type,
                                                              QueryMemoryDescriptor(),
                                                              nullptr,
                                                              executor_->getCatalog(),
                                                              executor_->blockSize(),
                                                              executor_->gridSize()),
                                  {}};
  
    const auto table_infos = get_table_infos(ra_exe_unit_in, executor_);
    auto max_groups_buffer_entry_guess = work_unit.max_groups_buffer_entry_guess;
    ExecutionOptions eo_no_multifrag{eo.output_columnar_hint,
                                     false,
                                     false,
                                     eo.allow_loop_joins,
                                     eo.with_watchdog,
                                     eo.jit_debug,
                                     false,
                                     eo.with_dynamic_watchdog,
                                     eo.dynamic_watchdog_time_limit,
                                     false,
                                     false,
                                     eo.gpu_input_mem_limit_percent,
                                     eo.allow_runtime_query_interrupt,
                                     eo.running_query_interrupt_freq,
                                     eo.pending_query_interrupt_freq,
                                     eo.executor_type,
                                     eo.outer_fragment_indices};
  
    if (was_multifrag_kernel_launch) {
      try {
        // Attempt to retry using the kernel per fragment path. The smaller input size
        // required may allow the entire kernel to execute in GPU memory.
        LOG(WARNING) << "Multifrag query ran out of memory, retrying with multifragment "
                        "kernels disabled.";
        const auto ra_exe_unit = decide_approx_count_distinct_implementation(
            ra_exe_unit_in, table_infos, executor_, co.device_type, target_exprs_owned_);
        ColumnCacheMap column_cache;
        result = {executor_->executeWorkUnit(max_groups_buffer_entry_guess,
                                             is_agg,
                                             table_infos,
                                             ra_exe_unit,
                                             co,
                                             eo_no_multifrag,
                                             cat_,
                                             nullptr,
                                             true,
                                             column_cache),
                  targets_meta};
        result.setQueueTime(queue_time_ms);
      } catch (const QueryExecutionError& e) {
        handlePersistentError(e.getErrorCode());
        LOG(WARNING) << "Kernel per fragment query ran out of memory, retrying on CPU.";
      }
    }
  
    if (render_info) {
      render_info->setForceNonInSituData();
    }
  
    const auto co_cpu = CompilationOptions::makeCpuOnly(co);
    // Only reset the group buffer entry guess if we ran out of slots, which
    // suggests a
    // highly pathological input which prevented a good estimation of distinct tuple
    // count. For projection queries, this will force a per-fragment scan limit, which is
    // compatible with the CPU path
    VLOG(1) << "Resetting max groups buffer entry guess.";
    max_groups_buffer_entry_guess = 0;
  
    int iteration_ctr = -1;
    while (true) {
      iteration_ctr++;
      auto ra_exe_unit = decide_approx_count_distinct_implementation(
          ra_exe_unit_in, table_infos, executor_, co_cpu.device_type, target_exprs_owned_);
      ColumnCacheMap column_cache;
      try {
        result = {executor_->executeWorkUnit(max_groups_buffer_entry_guess,
                                             is_agg,
                                             table_infos,
                                             ra_exe_unit,
                                             co_cpu,
                                             eo_no_multifrag,
                                             cat_,
                                             nullptr,
                                             true,
                                             column_cache),
                  targets_meta};
      } catch (const QueryExecutionError& e) {
        // Ran out of slots
        if (e.getErrorCode() < 0) {
          // Even the conservative guess failed; it should only happen when we group
          // by a huge cardinality array. Maybe we should throw an exception instead?
          // Such a heavy query is entirely capable of exhausting all the host memory.
          CHECK(max_groups_buffer_entry_guess);
          // Only allow two iterations of increasingly large entry guesses up to a maximum
          // of 512MB per column per kernel
          if (g_enable_watchdog || iteration_ctr > 1) {
            throw std::runtime_error("Query ran out of output slots in the result");
          }
          max_groups_buffer_entry_guess *= 2;
          LOG(WARNING) << "Query ran out of slots in the output buffer, retrying with max "
                          "groups buffer entry "
                          "guess equal to "
                       << max_groups_buffer_entry_guess;
        } else {
          handlePersistentError(e.getErrorCode());
        }
        continue;
      }
      result.setQueueTime(queue_time_ms);
      return result;
    }
    return result;
  }
  
  void RelAlgExecutor::handlePersistentError(const int32_t error_code) {
    LOG(ERROR) << "Query execution failed with error "
               << getErrorMessageFromCode(error_code);
    if (error_code == Executor::ERR_OUT_OF_GPU_MEM) {
      // We ran out of GPU memory, this doesn't count as an error if the query is
      // allowed to continue on CPU because retry on CPU is explicitly allowed through
      // --allow-cpu-retry.
      LOG(INFO) << "Query ran out of GPU memory, attempting punt to CPU";
      if (!g_allow_cpu_retry) {
        throw std::runtime_error(
            "Query ran out of GPU memory, unable to automatically retry on CPU");
      }
      return;
    }
    throw std::runtime_error(getErrorMessageFromCode(error_code));
  }
  
  namespace {
  struct ErrorInfo {
    const char* code{nullptr};
    const char* description{nullptr};
  };
  ErrorInfo getErrorDescription(const int32_t error_code) {
    switch (error_code) {
      case Executor::ERR_DIV_BY_ZERO:
        return {.code = "ERR_DIV_BY_ZERO", .description = "Division by zero"};
      case Executor::ERR_OUT_OF_GPU_MEM:
        return {.code = "ERR_OUT_OF_GPU_MEM",
                .description =
                    "Query couldn't keep the entire working set of columns in GPU memory"};
      case Executor::ERR_UNSUPPORTED_SELF_JOIN:
        return {.code = "ERR_UNSUPPORTED_SELF_JOIN",
                .description = "Self joins not supported yet"};
      case Executor::ERR_OUT_OF_CPU_MEM:
        return {.code = "ERR_OUT_OF_CPU_MEM",
                .description = "Not enough host memory to execute the query"};
      case Executor::ERR_OVERFLOW_OR_UNDERFLOW:
        return {.code = "ERR_OVERFLOW_OR_UNDERFLOW",
                .description = "Overflow or underflow"};
      case Executor::ERR_OUT_OF_TIME:
        return {.code = "ERR_OUT_OF_TIME",
                .description = "Query execution has exceeded the time limit"};
      case Executor::ERR_INTERRUPTED:
        return {.code = "ERR_INTERRUPTED",
                .description = "Query execution has been interrupted"};
      case Executor::ERR_COLUMNAR_CONVERSION_NOT_SUPPORTED:
        return {
            .code = "ERR_COLUMNAR_CONVERSION_NOT_SUPPORTED",
            .description = "Columnar conversion not supported for variable length types"};
      case Executor::ERR_TOO_MANY_LITERALS:
        return {.code = "ERR_TOO_MANY_LITERALS",
                .description = "Too many literals in the query"};
      case Executor::ERR_STRING_CONST_IN_RESULTSET:
        return {.code = "ERR_STRING_CONST_IN_RESULTSET",
                .description =
                    "NONE ENCODED String types are not supported as input result set."};
      case Executor::ERR_OUT_OF_RENDER_MEM:
        return {.code = "ERR_OUT_OF_RENDER_MEM",
                .description = "Not enough OpenGL memory to render the query results"};
      case Executor::ERR_STREAMING_TOP_N_NOT_SUPPORTED_IN_RENDER_QUERY:
        return {.code = "ERR_STREAMING_TOP_N_NOT_SUPPORTED_IN_RENDER_QUERY",
                .description = "Streaming-Top-N not supported in Render Query"};
      case Executor::ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES:
        return {.code = "ERR_SINGLE_VALUE_FOUND_MULTIPLE_VALUES",
                .description = "Multiple distinct values encountered"};
      case Executor::ERR_GEOS:
        return {.code = "ERR_GEOS", .description = "ERR_GEOS"};
      default:
        return {.code = nullptr, .description = nullptr};
    }
  }
  
  }  // namespace
  
  std::string RelAlgExecutor::getErrorMessageFromCode(const int32_t error_code) {
    if (error_code < 0) {
      return "Ran out of slots in the query output buffer";
    }
    const auto errorInfo = getErrorDescription(error_code);
  
    if (errorInfo.code) {
      return errorInfo.code + ": "s + errorInfo.description;
    } else {
      return "Other error: code "s + std::to_string(error_code);
    }
  }
  
  void RelAlgExecutor::executePostExecutionCallback() {
    if (post_execution_callback_) {
      VLOG(1) << "Running post execution callback.";
      (*post_execution_callback_)();
    }
  }
  
  RelAlgExecutor::WorkUnit RelAlgExecutor::createWorkUnit(const RelAlgNode* node,
                                                          const SortInfo& sort_info,
                                                          const ExecutionOptions& eo) {
    const auto compound = dynamic_cast<const RelCompound*>(node);
    if (compound) {
      return createCompoundWorkUnit(compound, sort_info, eo);
    }
    const auto project = dynamic_cast<const RelProject*>(node);
    if (project) {
      return createProjectWorkUnit(project, sort_info, eo);
    }
    const auto aggregate = dynamic_cast<const RelAggregate*>(node);
    if (aggregate) {
      return createAggregateWorkUnit(aggregate, sort_info, eo.just_explain);
    }
    const auto filter = dynamic_cast<const RelFilter*>(node);
    if (filter) {
      return createFilterWorkUnit(filter, sort_info, eo.just_explain);
    }
    LOG(FATAL) << "Unhandled node type: " << node->toString();
    return {};
  }
  
  namespace {
  
  JoinType get_join_type(const RelAlgNode* ra) {
    auto sink = get_data_sink(ra);
    if (auto join = dynamic_cast<const RelJoin*>(sink)) {
      return join->getJoinType();
    }
    if (dynamic_cast<const RelLeftDeepInnerJoin*>(sink)) {
      return JoinType::INNER;
    }
  
    return JoinType::INVALID;
  }
  
  std::unique_ptr<const RexOperator> get_bitwise_equals(const RexScalar* scalar) {
    const auto condition = dynamic_cast<const RexOperator*>(scalar);
    if (!condition || condition->getOperator() != kOR || condition->size() != 2) {
      return nullptr;
    }
    const auto equi_join_condition =
        dynamic_cast<const RexOperator*>(condition->getOperand(0));
    if (!equi_join_condition || equi_join_condition->getOperator() != kEQ) {
      return nullptr;
    }
    const auto both_are_null_condition =
        dynamic_cast<const RexOperator*>(condition->getOperand(1));
    if (!both_are_null_condition || both_are_null_condition->getOperator() != kAND ||
        both_are_null_condition->size() != 2) {
      return nullptr;
    }
    const auto lhs_is_null =
        dynamic_cast<const RexOperator*>(both_are_null_condition->getOperand(0));
    const auto rhs_is_null =
        dynamic_cast<const RexOperator*>(both_are_null_condition->getOperand(1));
    if (!lhs_is_null || !rhs_is_null || lhs_is_null->getOperator() != kISNULL ||
        rhs_is_null->getOperator() != kISNULL) {
      return nullptr;
    }
    CHECK_EQ(size_t(1), lhs_is_null->size());
    CHECK_EQ(size_t(1), rhs_is_null->size());
    CHECK_EQ(size_t(2), equi_join_condition->size());
    const auto eq_lhs = dynamic_cast<const RexInput*>(equi_join_condition->getOperand(0));
    const auto eq_rhs = dynamic_cast<const RexInput*>(equi_join_condition->getOperand(1));
    const auto is_null_lhs = dynamic_cast<const RexInput*>(lhs_is_null->getOperand(0));
    const auto is_null_rhs = dynamic_cast<const RexInput*>(rhs_is_null->getOperand(0));
    if (!eq_lhs || !eq_rhs || !is_null_lhs || !is_null_rhs) {
      return nullptr;
    }
    std::vector<std::unique_ptr<const RexScalar>> eq_operands;
    if (*eq_lhs == *is_null_lhs && *eq_rhs == *is_null_rhs) {
      RexDeepCopyVisitor deep_copy_visitor;
      auto lhs_op_copy = deep_copy_visitor.visit(equi_join_condition->getOperand(0));
      auto rhs_op_copy = deep_copy_visitor.visit(equi_join_condition->getOperand(1));
      eq_operands.emplace_back(lhs_op_copy.release());
      eq_operands.emplace_back(rhs_op_copy.release());
      return boost::make_unique<const RexOperator>(
          kBW_EQ, eq_operands, equi_join_condition->getType());
    }
    return nullptr;
  }
  
  std::unique_ptr<const RexOperator> get_bitwise_equals_conjunction(
      const RexScalar* scalar) {
    const auto condition = dynamic_cast<const RexOperator*>(scalar);
    if (condition && condition->getOperator() == kAND) {
      CHECK_GE(condition->size(), size_t(2));
      auto acc = get_bitwise_equals(condition->getOperand(0));
      if (!acc) {
        return nullptr;
      }
      for (size_t i = 1; i < condition->size(); ++i) {
        std::vector<std::unique_ptr<const RexScalar>> and_operands;
        and_operands.emplace_back(std::move(acc));
        and_operands.emplace_back(get_bitwise_equals_conjunction(condition->getOperand(i)));
        acc =
            boost::make_unique<const RexOperator>(kAND, and_operands, condition->getType());
      }
      return acc;
    }
    return get_bitwise_equals(scalar);
  }
  
  std::vector<JoinType> left_deep_join_types(const RelLeftDeepInnerJoin* left_deep_join) {
    CHECK_GE(left_deep_join->inputCount(), size_t(2));
    std::vector<JoinType> join_types(left_deep_join->inputCount() - 1, JoinType::INNER);
    for (size_t nesting_level = 1; nesting_level <= left_deep_join->inputCount() - 1;
         ++nesting_level) {
      if (left_deep_join->getOuterCondition(nesting_level)) {
        join_types[nesting_level - 1] = JoinType::LEFT;
      }
    }
    return join_types;
  }
  
  template <class RA>
  std::vector<size_t> do_table_reordering(
      std::vector<InputDescriptor>& input_descs,
      std::list<std::shared_ptr<const InputColDescriptor>>& input_col_descs,
      const JoinQualsPerNestingLevel& left_deep_join_quals,
      std::unordered_map<const RelAlgNode*, int>& input_to_nest_level,
      const RA* node,
      const std::vector<InputTableInfo>& query_infos,
      const Executor* executor) {
    if (g_cluster) {
      // Disable table reordering in distributed mode. The aggregator does not have enough
      // information to break ties
      return {};
    }
    const auto& cat = *executor->getCatalog();
    for (const auto& table_info : query_infos) {
      if (table_info.table_id < 0) {
        continue;
      }
      const auto td = cat.getMetadataForTable(table_info.table_id);
      CHECK(td);
      if (table_is_replicated(td)) {
        return {};
      }
    }
    const auto input_permutation =
        get_node_input_permutation(left_deep_join_quals, query_infos, executor);
    input_to_nest_level = get_input_nest_levels(node, input_permutation);
    std::tie(input_descs, input_col_descs, std::ignore) =
        get_input_desc(node, input_to_nest_level, input_permutation, cat);
    return input_permutation;
  }
  
  std::vector<size_t> get_left_deep_join_input_sizes(
      const RelLeftDeepInnerJoin* left_deep_join) {
    std::vector<size_t> input_sizes;
    for (size_t i = 0; i < left_deep_join->inputCount(); ++i) {
      const auto inputs = get_node_output(left_deep_join->getInput(i));
      input_sizes.push_back(inputs.size());
    }
    return input_sizes;
  }
  
  std::list<std::shared_ptr<Analyzer::Expr>> rewrite_quals(
      const std::list<std::shared_ptr<Analyzer::Expr>>& quals) {
    std::list<std::shared_ptr<Analyzer::Expr>> rewritten_quals;
    for (const auto& qual : quals) {
      const auto rewritten_qual = rewrite_expr(qual.get());
      rewritten_quals.push_back(rewritten_qual ? rewritten_qual : qual);
    }
    return rewritten_quals;
  }
  
  }  // namespace
  
  RelAlgExecutor::WorkUnit RelAlgExecutor::createCompoundWorkUnit(
      const RelCompound* compound,
      const SortInfo& sort_info,
      const ExecutionOptions& eo) {
    std::vector<InputDescriptor> input_descs;
    std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
    auto input_to_nest_level = get_input_nest_levels(compound, {});
    std::tie(input_descs, input_col_descs, std::ignore) =
        get_input_desc(compound, input_to_nest_level, {}, cat_);
    VLOG(3) << "input_descs=" << shared::printContainer(input_descs);
    const auto query_infos = get_table_infos(input_descs, executor_);
    CHECK_EQ(size_t(1), compound->inputCount());
    const auto left_deep_join =
        dynamic_cast<const RelLeftDeepInnerJoin*>(compound->getInput(0));
    JoinQualsPerNestingLevel left_deep_join_quals;
    const auto join_types = left_deep_join ? left_deep_join_types(left_deep_join)
                                           : std::vector<JoinType>{get_join_type(compound)};
    std::vector<size_t> input_permutation;
    std::vector<size_t> left_deep_join_input_sizes;
    if (left_deep_join) {
      left_deep_join_input_sizes = get_left_deep_join_input_sizes(left_deep_join);
      left_deep_join_quals = translateLeftDeepJoinFilter(
          left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
      if (g_from_table_reordering &&
          std::find(join_types.begin(), join_types.end(), JoinType::LEFT) ==
              join_types.end()) {
        input_permutation = do_table_reordering(input_descs,
                                                input_col_descs,
                                                left_deep_join_quals,
                                                input_to_nest_level,
                                                compound,
                                                query_infos,
                                                executor_);
        input_to_nest_level = get_input_nest_levels(compound, input_permutation);
        std::tie(input_descs, input_col_descs, std::ignore) =
            get_input_desc(compound, input_to_nest_level, input_permutation, cat_);
        left_deep_join_quals = translateLeftDeepJoinFilter(
            left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
      }
    }
    RelAlgTranslator translator(cat_,
                                query_state_,
                                executor_,
                                input_to_nest_level,
                                join_types,
                                now_,
                                eo.just_explain);
    const auto scalar_sources =
        translate_scalar_sources(compound, translator, eo.executor_type);
    const auto groupby_exprs = translate_groupby_exprs(compound, scalar_sources);
    const auto quals_cf = translate_quals(compound, translator);
    const auto target_exprs = translate_targets(target_exprs_owned_,
                                                scalar_sources,
                                                groupby_exprs,
                                                compound,
                                                translator,
                                                eo.executor_type);
    CHECK_EQ(compound->size(), target_exprs.size());
    const RelAlgExecutionUnit exe_unit = {
        input_descs,
        input_col_descs,
        quals_cf.simple_quals,
        rewrite_quals(quals_cf.quals),
        left_deep_join_quals,
        groupby_exprs,
        target_exprs,
        nullptr,
        sort_info,
        0,
        query_dag_ ? query_dag_->getQueryHints() : RegisteredQueryHint::defaults(),
        false,
        std::nullopt,
        query_state_};
    auto query_rewriter = std::make_unique<QueryRewriter>(query_infos, executor_);
    const auto rewritten_exe_unit = query_rewriter->rewrite(exe_unit);
    const auto targets_meta = get_targets_meta(compound, rewritten_exe_unit.target_exprs);
    compound->setOutputMetainfo(targets_meta);
    return {rewritten_exe_unit,
            compound,
            max_groups_buffer_entry_default_guess,
            std::move(query_rewriter),
            input_permutation,
            left_deep_join_input_sizes};
  }
  
  namespace {
  
  std::vector<const RexScalar*> rex_to_conjunctive_form(const RexScalar* qual_expr) {
    CHECK(qual_expr);
    const auto bin_oper = dynamic_cast<const RexOperator*>(qual_expr);
    if (!bin_oper || bin_oper->getOperator() != kAND) {
      return {qual_expr};
    }
    CHECK_GE(bin_oper->size(), size_t(2));
    auto lhs_cf = rex_to_conjunctive_form(bin_oper->getOperand(0));
    for (size_t i = 1; i < bin_oper->size(); ++i) {
      const auto rhs_cf = rex_to_conjunctive_form(bin_oper->getOperand(i));
      lhs_cf.insert(lhs_cf.end(), rhs_cf.begin(), rhs_cf.end());
    }
    return lhs_cf;
  }
  
  std::shared_ptr<Analyzer::Expr> build_logical_expression(
      const std::vector<std::shared_ptr<Analyzer::Expr>>& factors,
      const SQLOps sql_op) {
    CHECK(!factors.empty());
    auto acc = factors.front();
    for (size_t i = 1; i < factors.size(); ++i) {
      acc = Parser::OperExpr::normalize(sql_op, kONE, acc, factors[i]);
    }
    return acc;
  }
  
  template <class QualsList>
  bool list_contains_expression(const QualsList& haystack,
                                const std::shared_ptr<Analyzer::Expr>& needle) {
    for (const auto& qual : haystack) {
      if (*qual == *needle) {
        return true;
      }
    }
    return false;
  }
  
  // Transform `(p AND q) OR (p AND r)` to `p AND (q OR r)`. Avoids redundant
  // evaluations of `p` and allows use of the original form in joins if `p`
  // can be used for hash joins.
  std::shared_ptr<Analyzer::Expr> reverse_logical_distribution(
      const std::shared_ptr<Analyzer::Expr>& expr) {
    const auto expr_terms = qual_to_disjunctive_form(expr);
    CHECK_GE(expr_terms.size(), size_t(1));
    const auto& first_term = expr_terms.front();
    const auto first_term_factors = qual_to_conjunctive_form(first_term);
    std::vector<std::shared_ptr<Analyzer::Expr>> common_factors;
    // First, collect the conjunctive components common to all the disjunctive components.
    // Don't do it for simple qualifiers, we only care about expensive or join qualifiers.
    for (const auto& first_term_factor : first_term_factors.quals) {
      bool is_common =
          expr_terms.size() > 1;  // Only report common factors for disjunction.
      for (size_t i = 1; i < expr_terms.size(); ++i) {
        const auto crt_term_factors = qual_to_conjunctive_form(expr_terms[i]);
        if (!list_contains_expression(crt_term_factors.quals, first_term_factor)) {
          is_common = false;
          break;
        }
      }
      if (is_common) {
        common_factors.push_back(first_term_factor);
      }
    }
    if (common_factors.empty()) {
      return expr;
    }
    // Now that the common expressions are known, collect the remaining expressions.
    std::vector<std::shared_ptr<Analyzer::Expr>> remaining_terms;
    for (const auto& term : expr_terms) {
      const auto term_cf = qual_to_conjunctive_form(term);
      std::vector<std::shared_ptr<Analyzer::Expr>> remaining_quals(
          term_cf.simple_quals.begin(), term_cf.simple_quals.end());
      for (const auto& qual : term_cf.quals) {
        if (!list_contains_expression(common_factors, qual)) {
          remaining_quals.push_back(qual);
        }
      }
      if (!remaining_quals.empty()) {
        remaining_terms.push_back(build_logical_expression(remaining_quals, kAND));
      }
    }
    // Reconstruct the expression with the transformation applied.
    const auto common_expr = build_logical_expression(common_factors, kAND);
    if (remaining_terms.empty()) {
      return common_expr;
    }
    const auto remaining_expr = build_logical_expression(remaining_terms, kOR);
    return Parser::OperExpr::normalize(kAND, kONE, common_expr, remaining_expr);
  }
  
  }  // namespace
  
  std::list<std::shared_ptr<Analyzer::Expr>> RelAlgExecutor::makeJoinQuals(
      const RexScalar* join_condition,
      const std::vector<JoinType>& join_types,
      const std::unordered_map<const RelAlgNode*, int>& input_to_nest_level,
      const bool just_explain) const {
    RelAlgTranslator translator(
        cat_, query_state_, executor_, input_to_nest_level, join_types, now_, just_explain);
    const auto rex_condition_cf = rex_to_conjunctive_form(join_condition);
    std::list<std::shared_ptr<Analyzer::Expr>> join_condition_quals;
    for (const auto rex_condition_component : rex_condition_cf) {
      const auto bw_equals = get_bitwise_equals_conjunction(rex_condition_component);
      const auto join_condition =
          reverse_logical_distribution(translator.translateScalarRex(
              bw_equals ? bw_equals.get() : rex_condition_component));
      auto join_condition_cf = qual_to_conjunctive_form(join_condition);
      join_condition_quals.insert(join_condition_quals.end(),
                                  join_condition_cf.quals.begin(),
                                  join_condition_cf.quals.end());
      join_condition_quals.insert(join_condition_quals.end(),
                                  join_condition_cf.simple_quals.begin(),
                                  join_condition_cf.simple_quals.end());
    }
    return combine_equi_join_conditions(join_condition_quals);
  }
  
  // Translate left deep join filter and separate the conjunctive form qualifiers
  // per nesting level. The code generated for hash table lookups on each level
  // must dominate its uses in deeper nesting levels.
  JoinQualsPerNestingLevel RelAlgExecutor::translateLeftDeepJoinFilter(
      const RelLeftDeepInnerJoin* join,
      const std::vector<InputDescriptor>& input_descs,
      const std::unordered_map<const RelAlgNode*, int>& input_to_nest_level,
      const bool just_explain) {
    const auto join_types = left_deep_join_types(join);
    const auto join_condition_quals = makeJoinQuals(
        join->getInnerCondition(), join_types, input_to_nest_level, just_explain);
    MaxRangeTableIndexVisitor rte_idx_visitor;
    JoinQualsPerNestingLevel result(input_descs.size() - 1);
    std::unordered_set<std::shared_ptr<Analyzer::Expr>> visited_quals;
    for (size_t rte_idx = 1; rte_idx < input_descs.size(); ++rte_idx) {
      const auto outer_condition = join->getOuterCondition(rte_idx);
      if (outer_condition) {
        result[rte_idx - 1].quals =
            makeJoinQuals(outer_condition, join_types, input_to_nest_level, just_explain);
        CHECK_LE(rte_idx, join_types.size());
        CHECK(join_types[rte_idx - 1] == JoinType::LEFT);
        result[rte_idx - 1].type = JoinType::LEFT;
        continue;
      }
      for (const auto& qual : join_condition_quals) {
        if (visited_quals.count(qual)) {
          continue;
        }
        const auto qual_rte_idx = rte_idx_visitor.visit(qual.get());
        if (static_cast<size_t>(qual_rte_idx) <= rte_idx) {
          const auto it_ok = visited_quals.emplace(qual);
          CHECK(it_ok.second);
          result[rte_idx - 1].quals.push_back(qual);
        }
      }
      CHECK_LE(rte_idx, join_types.size());
      CHECK(join_types[rte_idx - 1] == JoinType::INNER);
      result[rte_idx - 1].type = JoinType::INNER;
    }
    return result;
  }
  
  namespace {
  
  std::vector<std::shared_ptr<Analyzer::Expr>> synthesize_inputs(
      const RelAlgNode* ra_node,
      const size_t nest_level,
      const std::vector<TargetMetaInfo>& in_metainfo,
      const std::unordered_map<const RelAlgNode*, int>& input_to_nest_level) {
    CHECK_LE(size_t(1), ra_node->inputCount());
    CHECK_GE(size_t(2), ra_node->inputCount());
    const auto input = ra_node->getInput(nest_level);
    const auto it_rte_idx = input_to_nest_level.find(input);
    CHECK(it_rte_idx != input_to_nest_level.end());
    const int rte_idx = it_rte_idx->second;
    const int table_id = table_id_from_ra(input);
    std::vector<std::shared_ptr<Analyzer::Expr>> inputs;
    const auto scan_ra = dynamic_cast<const RelScan*>(input);
    int input_idx = 0;
    for (const auto& input_meta : in_metainfo) {
      inputs.push_back(
          std::make_shared<Analyzer::ColumnVar>(input_meta.get_type_info(),
                                                table_id,
                                                scan_ra ? input_idx + 1 : input_idx,
                                                rte_idx));
      ++input_idx;
    }
    return inputs;
  }
  
  }  // namespace
  
  RelAlgExecutor::WorkUnit RelAlgExecutor::createAggregateWorkUnit(
      const RelAggregate* aggregate,
      const SortInfo& sort_info,
      const bool just_explain) {
    std::vector<InputDescriptor> input_descs;
    std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
    const auto input_to_nest_level = get_input_nest_levels(aggregate, {});
    std::tie(input_descs, input_col_descs, used_inputs_owned) =
        get_input_desc(aggregate, input_to_nest_level, {}, cat_);
    const auto join_type = get_join_type(aggregate);
  
    RelAlgTranslator translator(cat_,
                                query_state_,
                                executor_,
                                input_to_nest_level,
                                {join_type},
                                now_,
                                just_explain);
    CHECK_EQ(size_t(1), aggregate->inputCount());
    const auto source = aggregate->getInput(0);
    const auto& in_metainfo = source->getOutputMetainfo();
    const auto scalar_sources =
        synthesize_inputs(aggregate, size_t(0), in_metainfo, input_to_nest_level);
    const auto groupby_exprs = translate_groupby_exprs(aggregate, scalar_sources);
    const auto target_exprs = translate_targets(
        target_exprs_owned_, scalar_sources, groupby_exprs, aggregate, translator);
    const auto targets_meta = get_targets_meta(aggregate, target_exprs);
    aggregate->setOutputMetainfo(targets_meta);
    return {RelAlgExecutionUnit{
                input_descs,
                input_col_descs,
                {},
                {},
                {},
                groupby_exprs,
                target_exprs,
                nullptr,
                sort_info,
                0,
                query_dag_ ? query_dag_->getQueryHints() : RegisteredQueryHint::defaults(),
                false,
                std::nullopt,
                query_state_},
            aggregate,
            max_groups_buffer_entry_default_guess,
            nullptr};
  }
  
  RelAlgExecutor::WorkUnit RelAlgExecutor::createProjectWorkUnit(
      const RelProject* project,
      const SortInfo& sort_info,
      const ExecutionOptions& eo) {
    std::vector<InputDescriptor> input_descs;
    std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
    auto input_to_nest_level = get_input_nest_levels(project, {});
    std::tie(input_descs, input_col_descs, std::ignore) =
        get_input_desc(project, input_to_nest_level, {}, cat_);
    const auto query_infos = get_table_infos(input_descs, executor_);
  
    const auto left_deep_join =
        dynamic_cast<const RelLeftDeepInnerJoin*>(project->getInput(0));
    JoinQualsPerNestingLevel left_deep_join_quals;
    const auto join_types = left_deep_join ? left_deep_join_types(left_deep_join)
                                           : std::vector<JoinType>{get_join_type(project)};
    std::vector<size_t> input_permutation;
    std::vector<size_t> left_deep_join_input_sizes;
    if (left_deep_join) {
      left_deep_join_input_sizes = get_left_deep_join_input_sizes(left_deep_join);
      const auto query_infos = get_table_infos(input_descs, executor_);
      left_deep_join_quals = translateLeftDeepJoinFilter(
          left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
      if (g_from_table_reordering) {
        input_permutation = do_table_reordering(input_descs,
                                                input_col_descs,
                                                left_deep_join_quals,
                                                input_to_nest_level,
                                                project,
                                                query_infos,
                                                executor_);
        input_to_nest_level = get_input_nest_levels(project, input_permutation);
        std::tie(input_descs, input_col_descs, std::ignore) =
            get_input_desc(project, input_to_nest_level, input_permutation, cat_);
        left_deep_join_quals = translateLeftDeepJoinFilter(
            left_deep_join, input_descs, input_to_nest_level, eo.just_explain);
      }
    }
  
    RelAlgTranslator translator(cat_,
                                query_state_,
                                executor_,
                                input_to_nest_level,
                                join_types,
                                now_,
                                eo.just_explain);
    const auto target_exprs_owned =
        translate_scalar_sources(project, translator, eo.executor_type);
    target_exprs_owned_.insert(
        target_exprs_owned_.end(), target_exprs_owned.begin(), target_exprs_owned.end());
    const auto target_exprs = get_exprs_not_owned(target_exprs_owned);
  
    const RelAlgExecutionUnit exe_unit = {
        input_descs,
        input_col_descs,
        {},
        {},
        left_deep_join_quals,
        {nullptr},
        target_exprs,
        nullptr,
        sort_info,
        0,
        query_dag_ ? query_dag_->getQueryHints() : RegisteredQueryHint::defaults(),
        false,
        std::nullopt,
        query_state_};
    auto query_rewriter = std::make_unique<QueryRewriter>(query_infos, executor_);
    const auto rewritten_exe_unit = query_rewriter->rewrite(exe_unit);
    const auto targets_meta = get_targets_meta(project, rewritten_exe_unit.target_exprs);
    project->setOutputMetainfo(targets_meta);
    return {rewritten_exe_unit,
            project,
            max_groups_buffer_entry_default_guess,
            std::move(query_rewriter),
            input_permutation,
            left_deep_join_input_sizes};
  }
  
  namespace {
  
  std::vector<std::shared_ptr<Analyzer::Expr>> target_exprs_for_union(
      RelAlgNode const* input_node) {
    std::vector<TargetMetaInfo> const& tmis = input_node->getOutputMetainfo();
    VLOG(3) << "input_node->getOutputMetainfo()=" << shared::printContainer(tmis);
    const int negative_node_id = -input_node->getId();
    std::vector<std::shared_ptr<Analyzer::Expr>> target_exprs;
    target_exprs.reserve(tmis.size());
    for (size_t i = 0; i < tmis.size(); ++i) {
      target_exprs.push_back(std::make_shared<Analyzer::ColumnVar>(
          tmis[i].get_type_info(), negative_node_id, i, 0));
    }
    return target_exprs;
  }
  
  }  // namespace
  
  RelAlgExecutor::WorkUnit RelAlgExecutor::createUnionWorkUnit(
      const RelLogicalUnion* logical_union,
      const SortInfo& sort_info,
      const ExecutionOptions& eo) {
    std::vector<InputDescriptor> input_descs;
    std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
    // Map ra input ptr to index (0, 1).
    auto input_to_nest_level = get_input_nest_levels(logical_union, {});
    std::tie(input_descs, input_col_descs, std::ignore) =
        get_input_desc(logical_union, input_to_nest_level, {}, cat_);
    const auto query_infos = get_table_infos(input_descs, executor_);
    auto const max_num_tuples =
        std::accumulate(query_infos.cbegin(),
                        query_infos.cend(),
                        size_t(0),
                        [](auto max, auto const& query_info) {
                          return std::max(max, query_info.info.getNumTuples());
                        });
  
    VLOG(3) << "input_to_nest_level.size()=" << input_to_nest_level.size() << " Pairs are:";
    for (auto& pair : input_to_nest_level) {
      VLOG(3) << "  (" << pair.first->toString() << ", " << pair.second << ')';
    }
  
    RelAlgTranslator translator(
        cat_, query_state_, executor_, input_to_nest_level, {}, now_, eo.just_explain);
  
    auto const input_exprs_owned = target_exprs_for_union(logical_union->getInput(0));
    CHECK(!input_exprs_owned.empty())
        << "No metainfo found for input node " << logical_union->getInput(0)->toString();
    VLOG(3) << "input_exprs_owned.size()=" << input_exprs_owned.size();
    for (auto& input_expr : input_exprs_owned) {
      VLOG(3) << "  " << input_expr->toString();
    }
    target_exprs_owned_.insert(
        target_exprs_owned_.end(), input_exprs_owned.begin(), input_exprs_owned.end());
    const auto target_exprs = get_exprs_not_owned(input_exprs_owned);
  
    VLOG(3) << "input_descs=" << shared::printContainer(input_descs)
            << " input_col_descs=" << shared::printContainer(input_col_descs)
            << " target_exprs.size()=" << target_exprs.size()
            << " max_num_tuples=" << max_num_tuples;
  
    const RelAlgExecutionUnit exe_unit = {
        input_descs,
        input_col_descs,
        {},  // quals_cf.simple_quals,
        {},  // rewrite_quals(quals_cf.quals),
        {},
        {nullptr},
        target_exprs,
        nullptr,
        sort_info,
        max_num_tuples,
        query_dag_ ? query_dag_->getQueryHints() : RegisteredQueryHint::defaults(),
        false,
        logical_union->isAll(),
        query_state_};
    auto query_rewriter = std::make_unique<QueryRewriter>(query_infos, executor_);
    const auto rewritten_exe_unit = query_rewriter->rewrite(exe_unit);
  
    RelAlgNode const* input0 = logical_union->getInput(0);
    if (auto const* node = dynamic_cast<const RelCompound*>(input0)) {
      logical_union->setOutputMetainfo(
          get_targets_meta(node, rewritten_exe_unit.target_exprs));
    } else if (auto const* node = dynamic_cast<const RelProject*>(input0)) {
      logical_union->setOutputMetainfo(
          get_targets_meta(node, rewritten_exe_unit.target_exprs));
    } else if (auto const* node = dynamic_cast<const RelLogicalUnion*>(input0)) {
      logical_union->setOutputMetainfo(
          get_targets_meta(node, rewritten_exe_unit.target_exprs));
    } else if (auto const* node = dynamic_cast<const RelAggregate*>(input0)) {
      logical_union->setOutputMetainfo(
          get_targets_meta(node, rewritten_exe_unit.target_exprs));
    } else if (auto const* node = dynamic_cast<const RelScan*>(input0)) {
      logical_union->setOutputMetainfo(
          get_targets_meta(node, rewritten_exe_unit.target_exprs));
    } else if (auto const* node = dynamic_cast<const RelFilter*>(input0)) {
      logical_union->setOutputMetainfo(
          get_targets_meta(node, rewritten_exe_unit.target_exprs));
    } else if (dynamic_cast<const RelSort*>(input0)) {
      throw QueryNotSupported("LIMIT and OFFSET are not currently supported with UNION.");
    } else {
      throw QueryNotSupported("Unsupported input type: " + input0->toString());
    }
    VLOG(3) << "logical_union->getOutputMetainfo()="
            << shared::printContainer(logical_union->getOutputMetainfo())
            << " rewritten_exe_unit.input_col_descs.front()->getScanDesc().getTableId()="
            << rewritten_exe_unit.input_col_descs.front()->getScanDesc().getTableId();
  
    return {rewritten_exe_unit,
            logical_union,
            max_groups_buffer_entry_default_guess,
            std::move(query_rewriter)};
  }
  
  RelAlgExecutor::TableFunctionWorkUnit RelAlgExecutor::createTableFunctionWorkUnit(
      const RelTableFunction* table_func,
      const bool just_explain,
      const bool is_gpu) {
    std::vector<InputDescriptor> input_descs;
    std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
    auto input_to_nest_level = get_input_nest_levels(table_func, {});
    std::tie(input_descs, input_col_descs, std::ignore) =
        get_input_desc(table_func, input_to_nest_level, {}, cat_);
    const auto query_infos = get_table_infos(input_descs, executor_);
    RelAlgTranslator translator(
        cat_, query_state_, executor_, input_to_nest_level, {}, now_, just_explain);
    const auto input_exprs_owned =
        translate_scalar_sources(table_func, translator, ::ExecutorType::Native);
    target_exprs_owned_.insert(
        target_exprs_owned_.end(), input_exprs_owned.begin(), input_exprs_owned.end());
    const auto input_exprs = get_exprs_not_owned(input_exprs_owned);
  
    const auto table_function_impl_and_type_infos = [=]() {
      if (is_gpu) {
        try {
          return bind_table_function(
              table_func->getFunctionName(), input_exprs_owned, is_gpu);
        } catch (ExtensionFunctionBindingError& e) {
          LOG(WARNING) << "createTableFunctionWorkUnit[GPU]: " << e.what()
                       << " Redirecting " << table_func->getFunctionName()
                       << " to run on CPU.";
          throw QueryMustRunOnCpu();
        }
      } else {
        try {
          return bind_table_function(
              table_func->getFunctionName(), input_exprs_owned, is_gpu);
        } catch (ExtensionFunctionBindingError& e) {
          LOG(WARNING) << "createTableFunctionWorkUnit[CPU]: " << e.what();
          throw;
        }
      }
    }();
    const auto& table_function_impl = std::get<0>(table_function_impl_and_type_infos);
    const auto& table_function_type_infos = std::get<1>(table_function_impl_and_type_infos);
  
    size_t output_row_sizing_param = 0;
    if (table_function_impl.hasUserSpecifiedOutputSizeMultiplier() ||
        table_function_impl.hasUserSpecifiedOutputSizeConstant()) {
      const auto parameter_index =
          table_function_impl.getOutputRowSizeParameter(table_function_type_infos);
      CHECK_GT(parameter_index, size_t(0));
      const auto parameter_expr = table_func->getTableFuncInputAt(parameter_index - 1);
      const auto parameter_expr_literal = dynamic_cast<const RexLiteral*>(parameter_expr);
      if (!parameter_expr_literal) {
        throw std::runtime_error(
            "Provided output buffer sizing parameter is not a literal. Only literal "
            "values are supported with output buffer sizing configured table "
            "functions.");
      }
      int64_t literal_val = parameter_expr_literal->getVal<int64_t>();
      if (literal_val < 0) {
        throw std::runtime_error("Provided output sizing parameter " +
                                 std::to_string(literal_val) +
                                 " must be positive integer.");
      }
      output_row_sizing_param = static_cast<size_t>(literal_val);
    } else if (table_function_impl.hasNonUserSpecifiedOutputSizeConstant()) {
      output_row_sizing_param = table_function_impl.getOutputRowSizeParameter();
    } else {
      UNREACHABLE();
    }
  
    std::vector<Analyzer::ColumnVar*> input_col_exprs;
    size_t input_index = 0;
    for (const auto& ti : table_function_type_infos) {
      if (ti.is_column_list()) {
        for (int i = 0; i < ti.get_dimension(); i++) {
          auto& input_expr = input_exprs[input_index];
          auto col_var = dynamic_cast<Analyzer::ColumnVar*>(input_expr);
          CHECK(col_var);
          input_expr->set_type_info(
              ti);  // ti is shared in between all columns in the same column_list
          input_col_exprs.push_back(col_var);
          input_index++;
        }
      } else if (ti.is_column()) {
        auto& input_expr = input_exprs[input_index];
        auto col_var = dynamic_cast<Analyzer::ColumnVar*>(input_expr);
        CHECK(col_var);
        input_expr->set_type_info(ti);
        input_col_exprs.push_back(col_var);
        input_index++;
      } else {
        input_index++;
      }
    }
    CHECK_EQ(input_col_exprs.size(), table_func->getColInputsSize());
    std::vector<Analyzer::Expr*> table_func_outputs;
    for (size_t i = 0; i < table_function_impl.getOutputsSize(); i++) {
      const auto ti = table_function_impl.getOutputSQLType(i);
      target_exprs_owned_.push_back(std::make_shared<Analyzer::ColumnVar>(ti, 0, i, -1));
      table_func_outputs.push_back(target_exprs_owned_.back().get());
    }
    const TableFunctionExecutionUnit exe_unit = {
        input_descs,
        input_col_descs,
        input_exprs,              // table function inputs
        input_col_exprs,          // table function column inputs (duplicates w/ above)
        table_func_outputs,       // table function projected exprs
        output_row_sizing_param,  // output buffer sizing param
        table_function_impl};
    const auto targets_meta = get_targets_meta(table_func, exe_unit.target_exprs);
    table_func->setOutputMetainfo(targets_meta);
    return {exe_unit, table_func};
  }
  
  namespace {
  
  std::pair<std::vector<TargetMetaInfo>, std::vector<std::shared_ptr<Analyzer::Expr>>>
  get_inputs_meta(const RelFilter* filter,
                  const RelAlgTranslator& translator,
                  const std::vector<std::shared_ptr<RexInput>>& inputs_owned,
                  const std::unordered_map<const RelAlgNode*, int>& input_to_nest_level) {
    std::vector<TargetMetaInfo> in_metainfo;
    std::vector<std::shared_ptr<Analyzer::Expr>> exprs_owned;
    const auto data_sink_node = get_data_sink(filter);
    auto input_it = inputs_owned.begin();
    for (size_t nest_level = 0; nest_level < data_sink_node->inputCount(); ++nest_level) {
      const auto source = data_sink_node->getInput(nest_level);
      const auto scan_source = dynamic_cast<const RelScan*>(source);
      if (scan_source) {
        CHECK(source->getOutputMetainfo().empty());
        std::vector<std::shared_ptr<Analyzer::Expr>> scalar_sources_owned;
        for (size_t i = 0; i < scan_source->size(); ++i, ++input_it) {
          scalar_sources_owned.push_back(translator.translateScalarRex(input_it->get()));
        }
        const auto source_metadata =
            get_targets_meta(scan_source, get_exprs_not_owned(scalar_sources_owned));
        in_metainfo.insert(
            in_metainfo.end(), source_metadata.begin(), source_metadata.end());
        exprs_owned.insert(
            exprs_owned.end(), scalar_sources_owned.begin(), scalar_sources_owned.end());
      } else {
        const auto& source_metadata = source->getOutputMetainfo();
        input_it += source_metadata.size();
        in_metainfo.insert(
            in_metainfo.end(), source_metadata.begin(), source_metadata.end());
        const auto scalar_sources_owned = synthesize_inputs(
            data_sink_node, nest_level, source_metadata, input_to_nest_level);
        exprs_owned.insert(
            exprs_owned.end(), scalar_sources_owned.begin(), scalar_sources_owned.end());
      }
    }
    return std::make_pair(in_metainfo, exprs_owned);
  }
  
  }  // namespace
  
  RelAlgExecutor::WorkUnit RelAlgExecutor::createFilterWorkUnit(const RelFilter* filter,
                                                                const SortInfo& sort_info,
                                                                const bool just_explain) {
    CHECK_EQ(size_t(1), filter->inputCount());
    std::vector<InputDescriptor> input_descs;
    std::list<std::shared_ptr<const InputColDescriptor>> input_col_descs;
    std::vector<TargetMetaInfo> in_metainfo;
    std::vector<std::shared_ptr<RexInput>> used_inputs_owned;
    std::vector<std::shared_ptr<Analyzer::Expr>> target_exprs_owned;
  
    const auto input_to_nest_level = get_input_nest_levels(filter, {});
    std::tie(input_descs, input_col_descs, used_inputs_owned) =
        get_input_desc(filter, input_to_nest_level, {}, cat_);
    const auto join_type = get_join_type(filter);
    RelAlgTranslator translator(cat_,
                                query_state_,
                                executor_,
                                input_to_nest_level,
                                {join_type},
                                now_,
                                just_explain);
    std::tie(in_metainfo, target_exprs_owned) =
        get_inputs_meta(filter, translator, used_inputs_owned, input_to_nest_level);
    const auto filter_expr = translator.translateScalarRex(filter->getCondition());
    const auto qual = fold_expr(filter_expr.get());
    target_exprs_owned_.insert(
        target_exprs_owned_.end(), target_exprs_owned.begin(), target_exprs_owned.end());
    const auto target_exprs = get_exprs_not_owned(target_exprs_owned);
    filter->setOutputMetainfo(in_metainfo);
    const auto rewritten_qual = rewrite_expr(qual.get());
    return {{input_descs,
             input_col_descs,
             {},
             {rewritten_qual ? rewritten_qual : qual},
             {},
             {nullptr},
             target_exprs,
             nullptr,
             sort_info,
             0,
             query_dag_ ? query_dag_->getQueryHints() : RegisteredQueryHint::defaults()},
            filter,
            max_groups_buffer_entry_default_guess,
            nullptr};
  }
  
  SpeculativeTopNBlacklist RelAlgExecutor::speculative_topn_blacklist_;
  