DBHandler.cpp

Go to the documentation of this file.

/*
 * Copyright 2022 HEAVY.AI, 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 "DBHandler.h"
#include "DistributedLoader.h"
#include "TokenCompletionHints.h"

#ifdef HAVE_PROFILER
#include <gperftools/heap-profiler.h>
#endif  // HAVE_PROFILER

#include "MapDRelease.h"

#include "Calcite/Calcite.h"
#include "gen-cpp/CalciteServer.h"

#include "QueryEngine/ErrorHandling.h"
#include "QueryEngine/RelAlgExecutor.h"

#include "Catalog/Catalog.h"
#include "Catalog/DdlCommandExecutor.h"
#include "DataMgr/ForeignStorage/ArrowForeignStorage.h"
#include "DataMgr/ForeignStorage/DummyForeignStorage.h"
#include "DataMgr/ForeignStorage/PassThroughBuffer.h"
#include "DistributedHandler.h"
#include "Fragmenter/InsertOrderFragmenter.h"
#include "Geospatial/ColumnNames.h"
#include "Geospatial/Compression.h"
#include "Geospatial/GDAL.h"
#include "Geospatial/Types.h"
#include "ImportExport/Importer.h"
#include "LockMgr/LockMgr.h"
#include "OSDependent/heavyai_hostname.h"
#include "Parser/ParserWrapper.h"
#include "Parser/ReservedKeywords.h"
#include "QueryEngine/ArrowResultSet.h"
#include "QueryEngine/CalciteAdapter.h"
#include "QueryEngine/Execute.h"
#include "QueryEngine/ExtensionFunctionsWhitelist.h"
#include "QueryEngine/GpuMemUtils.h"
#include "QueryEngine/JoinFilterPushDown.h"
#include "QueryEngine/JsonAccessors.h"
#include "QueryEngine/QueryDispatchQueue.h"
#include "QueryEngine/ResultSetBuilder.h"
#include "QueryEngine/TableFunctions/TableFunctionsFactory.h"
#include "QueryEngine/TableOptimizer.h"
#include "QueryEngine/ThriftSerializers.h"
#include "RequestInfo.h"
#ifdef HAVE_RUNTIME_LIBS
#include "RuntimeLibManager/RuntimeLibManager.h"
#endif
#include "Shared/ArrowUtil.h"
#include "Shared/DateTimeParser.h"
#include "Shared/StringTransform.h"
#include "Shared/SysDefinitions.h"
#include "Shared/file_path_util.h"
#include "Shared/heavyai_shared_mutex.h"
#include "Shared/import_helpers.h"
#include "Shared/measure.h"
#include "Shared/misc.h"
#include "Shared/scope.h"
#include "UdfCompiler/UdfCompiler.h"

#ifdef HAVE_AWS_S3
#include <aws/core/auth/AWSCredentialsProviderChain.h>
#endif
#include <fcntl.h>
#include <picosha2.h>
#include <sys/types.h>
#include <algorithm>
#include <boost/algorithm/string.hpp>
#include <boost/filesystem.hpp>
#include <boost/make_shared.hpp>
#include <boost/process/search_path.hpp>
#include <boost/program_options.hpp>
#include <boost/tokenizer.hpp>
#include <chrono>
#include <cmath>
#include <csignal>
#include <fstream>
#include <future>
#include <map>
#include <memory>
#include <random>
#include <string>
#include <thread>
#include <typeinfo>

#include <arrow/api.h>
#include <arrow/io/api.h>
#include <arrow/ipc/api.h>

#include "Shared/ArrowUtil.h"
#include "Shared/distributed.h"

#ifdef ENABLE_IMPORT_PARQUET
extern bool g_enable_parquet_import_fsi;
#endif

#ifdef HAVE_AWS_S3
extern bool g_allow_s3_server_privileges;
#endif

extern bool g_enable_system_tables;
bool g_allow_system_dashboard_update{false};
bool g_uniform_request_ids_per_thrift_call{true};
extern bool g_allow_memory_status_log;

using Catalog_Namespace::Catalog;
using Catalog_Namespace::SysCatalog;

#define INVALID_SESSION_ID ""

#define SET_REQUEST_ID(parent_request_id)                         \
  if (g_uniform_request_ids_per_thrift_call && parent_request_id) \
    logger::set_request_id(parent_request_id);                    \
  else if (logger::set_new_request_id(); parent_request_id)       \
  LOG(INFO) << "This request has parent request_id(" << parent_request_id << ')'

#define THROW_DB_EXCEPTION(errstr) \
  {                                \
    TDBException ex;               \
    ex.error_msg = errstr;         \
    LOG(ERROR) << ex.error_msg;    \
    throw ex;                      \
  }

thread_local std::string TrackingProcessor::client_address;
thread_local ClientProtocol TrackingProcessor::client_protocol;

namespace {

bool dashboard_exists(const Catalog_Namespace::Catalog& cat,
                      const int32_t user_id,
                      const std::string& dashboard_name) {
  return (cat.getMetadataForDashboard(std::to_string(user_id), dashboard_name));
}

struct ForceDisconnect : public std::runtime_error {
  ForceDisconnect(const std::string& cause) : std::runtime_error(cause) {}
};

}  // namespace

#ifdef ENABLE_GEOS
// from Geospatial/GeosValidation.cpp
extern std::unique_ptr<std::string> g_libgeos_so_filename;
#endif

DBHandler::DBHandler(const std::vector<LeafHostInfo>& db_leaves,
                     const std::vector<LeafHostInfo>& string_leaves,
                     const std::string& base_data_path,
                     const bool allow_multifrag,
                     const bool jit_debug,
                     const bool intel_jit_profile,
                     const bool read_only,
                     const bool allow_loop_joins,
                     const bool enable_rendering,
                     const bool renderer_prefer_igpu,
                     const unsigned renderer_vulkan_timeout_ms,
                     const bool renderer_use_parallel_executors,
                     const bool enable_auto_clear_render_mem,
                     const int render_oom_retry_threshold,
                     const size_t render_mem_bytes,
                     const size_t max_concurrent_render_sessions,
                     const size_t reserved_gpu_mem,
                     const bool render_compositor_use_last_gpu,
                     const bool renderer_enable_slab_allocation,
                     const size_t num_reader_threads,
                     const AuthMetadata& authMetadata,
                     SystemParameters& system_parameters,
                     const bool legacy_syntax,
                     const int idle_session_duration,
                     const int max_session_duration,
                     const std::string& udf_filename,
                     const std::string& clang_path,
                     const std::vector<std::string>& clang_options,
#ifdef ENABLE_GEOS
                     const std::string& libgeos_so_filename,
#endif
#ifdef HAVE_TORCH_TFS
                     const std::string& torch_lib_path,
#endif
                     const File_Namespace::DiskCacheConfig& disk_cache_config,
                     const bool is_new_db)
    : leaf_aggregator_(db_leaves)
    , db_leaves_(db_leaves)
    , string_leaves_(string_leaves)
    , base_data_path_(base_data_path)
    , random_gen_(std::random_device{}())
    , session_id_dist_(0, INT32_MAX)
    , jit_debug_(jit_debug)
    , intel_jit_profile_(intel_jit_profile)
    , allow_multifrag_(allow_multifrag)
    , read_only_(read_only)
    , allow_loop_joins_(allow_loop_joins)
    , authMetadata_(authMetadata)
    , system_parameters_(system_parameters)
    , legacy_syntax_(legacy_syntax)
    , dispatch_queue_(
          std::make_unique<QueryDispatchQueue>(system_parameters.num_executors))
    , super_user_rights_(false)
    , idle_session_duration_(idle_session_duration * 60)
    , max_session_duration_(max_session_duration * 60)
    , enable_rendering_(enable_rendering)
    , renderer_prefer_igpu_(renderer_prefer_igpu)
    , renderer_vulkan_timeout_(renderer_vulkan_timeout_ms)
    , renderer_use_parallel_executors_(renderer_use_parallel_executors)
    , enable_auto_clear_render_mem_(enable_auto_clear_render_mem)
    , render_oom_retry_threshold_(render_oom_retry_threshold)
    , render_mem_bytes_(render_mem_bytes)
    , max_concurrent_render_sessions_(max_concurrent_render_sessions)
    , reserved_gpu_mem_(reserved_gpu_mem)
    , render_compositor_use_last_gpu_(render_compositor_use_last_gpu)
    , renderer_enable_slab_allocation_{renderer_enable_slab_allocation}
    , num_reader_threads_(num_reader_threads)
#ifdef ENABLE_GEOS
    , libgeos_so_filename_(libgeos_so_filename)
#endif
#ifdef HAVE_TORCH_TFS
    , torch_lib_path_(torch_lib_path)
#endif
    , disk_cache_config_(disk_cache_config)
    , udf_filename_(udf_filename)
    , clang_path_(clang_path)
    , clang_options_(clang_options)
    , max_num_sessions_(-1) {
  LOG(INFO) << "HeavyDB Server " << MAPD_RELEASE;
  initialize(is_new_db);
  resetSessionsStore();
}

void DBHandler::init_executor_resource_mgr() {
  size_t num_cpu_slots{0};
  size_t num_gpu_slots{0};
  size_t cpu_result_mem{0};
  size_t cpu_buffer_pool_mem{0};
  size_t gpu_buffer_pool_mem{0};
  LOG(INFO) << "Initializing Executor Resource Manager";

  if (g_cpu_threads_override != 0) {
    LOG(INFO) << "\tSetting Executor resource pool avaiable CPU threads/slots to "
                 "user-specified value of "
              << g_cpu_threads_override << ".";
    num_cpu_slots = g_cpu_threads_override;
  } else {
    LOG(INFO) << "\tSetting Executor resource pool avaiable CPU threads/slots to default "
                 "value of "
              << cpu_threads() << ".";
    // Setting the number of CPU slots to cpu_threads() will cause the ExecutorResourceMgr
    // to set the logical number of available cpu slots to mirror the number of threads in
    // the tbb thread pool and used elsewhere in the system, but we may want to consider a
    // capability to allow the executor resource pool number of threads to be set
    // independently as some fraction of the what cpu_threads() will return, to give some
    // breathing room for all the other processes in the system that use CPU threadds
    num_cpu_slots = cpu_threads();
  }
  LOG(INFO) << "\tSetting max per-query CPU threads to ratio of "
            << g_executor_resource_mgr_per_query_max_cpu_slots_ratio << " of "
            << num_cpu_slots << " available threads, or "
            << static_cast<size_t>(g_executor_resource_mgr_per_query_max_cpu_slots_ratio *
                                   num_cpu_slots)
            << " threads.";

  // system_parameters_.num_gpus will be -1 if there are no GPUs enabled so we need to
  // guard against this
  num_gpu_slots = system_parameters_.num_gpus < 0 ? static_cast<size_t>(0)
                                                  : system_parameters_.num_gpus;

  cpu_buffer_pool_mem = data_mgr_->getCpuBufferPoolSize();
  if (g_executor_resource_mgr_cpu_result_mem_bytes != Executor::auto_cpu_mem_bytes) {
    cpu_result_mem = g_executor_resource_mgr_cpu_result_mem_bytes;
  } else {
    const size_t system_mem_bytes = DataMgr::getTotalSystemMemory();
    CHECK_GT(system_mem_bytes, size_t(0));
    const size_t remaining_cpu_mem_bytes = system_mem_bytes >= cpu_buffer_pool_mem
                                               ? system_mem_bytes - cpu_buffer_pool_mem
                                               : 0UL;
    cpu_result_mem =
        std::max(static_cast<size_t>(remaining_cpu_mem_bytes *
                                     g_executor_resource_mgr_cpu_result_mem_ratio),
                 static_cast<size_t>(1UL << 32));
  }
  // Below gets total combined size of all gpu buffer pools
  // Likely will move to per device pool resource management,
  // but keeping simple for now
  gpu_buffer_pool_mem = data_mgr_->getGpuBufferPoolSize();

  // When we move to using the BufferMgrs directly in
  // ExecutorResourcePool, there won't be a need for
  // the buffer_pool_max_occupancy variable - a
  // safety "fudge" factor as what the resource pool sees
  // and what the BufferMgrs see will be exactly the same.

  // However we need to ensure we can quickly access
  // chunk state of BufferMgrs without going through coarse lock
  // before we do this, so use this fudge ratio for now

  // Note that if we are not conservative enough with the below and
  // overshoot, the error will still be caught and if on GPU, the query
  // can be re-run on CPU

  constexpr double buffer_pool_max_occupancy{0.95};
  const size_t conservative_cpu_buffer_pool_mem =
      static_cast<size_t>(cpu_buffer_pool_mem * buffer_pool_max_occupancy);
  const size_t conservative_gpu_buffer_pool_mem =
      static_cast<size_t>(gpu_buffer_pool_mem * buffer_pool_max_occupancy);

  LOG(INFO)
      << "\tSetting Executor resource pool reserved space for CPU buffer pool memory to "
      << format_num_bytes(conservative_cpu_buffer_pool_mem) << ".";
  if (gpu_buffer_pool_mem > 0UL) {
    LOG(INFO) << "\tSetting Executor resource pool reserved space for GPU buffer pool "
                 "memory to "
              << format_num_bytes(conservative_gpu_buffer_pool_mem) << ".";
  }
  LOG(INFO) << "\tSetting Executor resource pool reserved space for CPU result memory to "
            << format_num_bytes(cpu_result_mem) << ".";

  Executor::init_resource_mgr(
      num_cpu_slots,
      num_gpu_slots,
      cpu_result_mem,
      conservative_cpu_buffer_pool_mem,
      conservative_gpu_buffer_pool_mem,
      g_executor_resource_mgr_per_query_max_cpu_slots_ratio,
      g_executor_resource_mgr_per_query_max_cpu_result_mem_ratio,
      g_executor_resource_mgr_allow_cpu_kernel_concurrency,
      g_executor_resource_mgr_allow_cpu_gpu_kernel_concurrency,
      g_executor_resource_mgr_allow_cpu_slot_oversubscription_concurrency,
      g_executor_resource_mgr_allow_cpu_result_mem_oversubscription_concurrency,
      g_executor_resource_mgr_max_available_resource_use_ratio);
}

void DBHandler::validate_configurations() {
#ifndef _WIN32
  size_t temp;
  CHECK(!__builtin_mul_overflow(g_num_tuple_threshold_switch_to_baseline,
                                g_ratio_num_hash_entry_to_num_tuple_switch_to_baseline,
                                &temp))
      << "The product of g_num_tuple_threshold_switch_to_baseline and "
         "g_ratio_num_hash_entry_to_num_tuple_switch_to_baseline exceeds 64 bits.";
#endif
}

void DBHandler::resetSessionsStore() {
  if (sessions_store_) {
    // Disconnect any existing sessions.
    auto sessions = sessions_store_->getAllSessions();
    for (auto session : sessions) {
      sessions_store_->disconnect(session->get_session_id());
    }
  }
  sessions_store_ = Catalog_Namespace::SessionsStore::create(
      base_data_path_,
      1,
      idle_session_duration_,
      max_session_duration_,
      max_num_sessions_,
      [this](auto& session_ptr) { disconnect_impl(session_ptr); });
}

void DBHandler::initialize(const bool is_new_db) {
  if (!initialized_) {
    initialized_ = true;
  } else {
    THROW_DB_EXCEPTION(
        "Server already initialized; service restart required to activate any new "
        "entitlements.");
    return;
  }

  if (system_parameters_.cpu_only || system_parameters_.num_gpus == 0) {
    executor_device_type_ = ExecutorDeviceType::CPU;
    cpu_mode_only_ = true;
  } else {
#ifdef HAVE_CUDA
    executor_device_type_ = ExecutorDeviceType::GPU;
    cpu_mode_only_ = false;
#else
    executor_device_type_ = ExecutorDeviceType::CPU;
    LOG(WARNING) << "This build isn't CUDA enabled, will run on CPU";
    cpu_mode_only_ = true;
#endif
  }

  bool is_rendering_enabled = enable_rendering_;
  if (system_parameters_.num_gpus == 0) {
    is_rendering_enabled = false;
  }

  const auto data_path =
      boost::filesystem::path(base_data_path_) / shared::kDataDirectoryName;
  // calculate the total amount of memory we need to reserve from each gpu that the Buffer
  // manage cannot ask for
  size_t total_reserved = reserved_gpu_mem_;
  if (is_rendering_enabled) {
    total_reserved += render_mem_bytes_;
  }

  std::unique_ptr<CudaMgr_Namespace::CudaMgr> cuda_mgr;
#ifdef HAVE_CUDA
  if (!cpu_mode_only_ || is_rendering_enabled) {
    try {
      cuda_mgr = std::make_unique<CudaMgr_Namespace::CudaMgr>(
          system_parameters_.num_gpus, system_parameters_.start_gpu);
      if (system_parameters_.num_gpus < 0) {
        system_parameters_.num_gpus = cuda_mgr->getDeviceCount();
      } else {
        system_parameters_.num_gpus =
            std::min(system_parameters_.num_gpus, cuda_mgr->getDeviceCount());
      }
    } catch (const std::exception& e) {
      LOG(ERROR) << "Unable to instantiate CudaMgr, falling back to CPU-only mode. "
                 << e.what();
      executor_device_type_ = ExecutorDeviceType::CPU;
      cpu_mode_only_ = true;
      is_rendering_enabled = false;
    }
  }
#endif  // HAVE_CUDA

  validate_configurations();

  try {
    data_mgr_.reset(new Data_Namespace::DataMgr(data_path.string(),
                                                system_parameters_,
                                                std::move(cuda_mgr),
                                                !cpu_mode_only_,
                                                total_reserved,
                                                num_reader_threads_,
                                                disk_cache_config_));
  } catch (const std::exception& e) {
    LOG(FATAL) << "Failed to initialize data manager: " << e.what();
  }
  if (g_enable_executor_resource_mgr) {
    init_executor_resource_mgr();
  }

  std::string udf_ast_filename("");

  try {
    if (!udf_filename_.empty()) {
      const auto cuda_mgr = data_mgr_->getCudaMgr();
      const CudaMgr_Namespace::NvidiaDeviceArch device_arch =
          cuda_mgr ? cuda_mgr->getDeviceArch()
                   : CudaMgr_Namespace::NvidiaDeviceArch::Kepler;
      UdfCompiler compiler(device_arch, clang_path_, clang_options_);

      const auto [cpu_udf_ir_file, cuda_udf_ir_file] = compiler.compileUdf(udf_filename_);
      Executor::addUdfIrToModule(cpu_udf_ir_file, /*is_cuda_ir=*/false);
      if (!cuda_udf_ir_file.empty()) {
        Executor::addUdfIrToModule(cuda_udf_ir_file, /*is_cuda_ir=*/true);
      }
      udf_ast_filename = compiler.getAstFileName(udf_filename_);
    }
  } catch (const std::exception& e) {
    LOG(FATAL) << "Failed to initialize UDF compiler: " << e.what();
  }

  try {
    calcite_ =
        std::make_shared<Calcite>(system_parameters_, base_data_path_, udf_ast_filename);
  } catch (const std::exception& e) {
    LOG(FATAL) << "Failed to initialize Calcite server: " << e.what();
  }

  try {
    ExtensionFunctionsWhitelist::add(calcite_->getExtensionFunctionWhitelist());
    if (!udf_filename_.empty()) {
      ExtensionFunctionsWhitelist::addUdfs(calcite_->getUserDefinedFunctionWhitelist());
    }
  } catch (const std::exception& e) {
    LOG(FATAL) << "Failed to initialize extension functions: " << e.what();
  }

  try {
    table_functions::init_table_functions();
  } catch (const std::exception& e) {
    LOG(FATAL) << "Failed to initialize table functions factory: " << e.what();
  }

#ifdef HAVE_RUNTIME_LIBS
  try {
#ifdef HAVE_TORCH_TFS
    RuntimeLibManager::loadRuntimeLibs(torch_lib_path_);
#else
    RuntimeLibManager::loadRuntimeLibs();
#endif
  } catch (const std::exception& e) {
    LOG(ERROR) << "Failed to load runtime libraries: " << e.what();
    LOG(ERROR) << "Support for runtime library table functions is disabled.";
  }
#endif

  try {
    auto udtfs = ThriftSerializers::to_thrift(
        table_functions::TableFunctionsFactory::get_table_funcs(/*is_runtime=*/false));
    std::vector<TUserDefinedFunction> udfs = {};
    calcite_->setRuntimeExtensionFunctions(udfs, udtfs, /*is_runtime=*/false);
  } catch (const std::exception& e) {
    LOG(FATAL) << "Failed to register compile-time table functions: " << e.what();
  }

  if (!data_mgr_->gpusPresent() && !cpu_mode_only_) {
    executor_device_type_ = ExecutorDeviceType::CPU;
    LOG(ERROR) << "No GPUs detected, falling back to CPU mode";
    cpu_mode_only_ = true;
  }

  LOG(INFO) << "Started in " << executor_device_type_ << " mode.";

  try {
    g_base_path = base_data_path_;
    SysCatalog::instance().init(base_data_path_,
                                data_mgr_,
                                authMetadata_,
                                calcite_,
                                is_new_db,
                                !db_leaves_.empty(),
                                string_leaves_);
  } catch (const std::exception& e) {
    LOG(FATAL) << "Failed to initialize system catalog: " << e.what();
  }

  import_path_ = boost::filesystem::path(base_data_path_) / shared::kDefaultImportDirName;
  start_time_ = std::time(nullptr);

  if (is_rendering_enabled) {
    try {
      render_handler_.reset(new RenderHandler(this,
                                              render_mem_bytes_,
                                              max_concurrent_render_sessions_,
                                              render_compositor_use_last_gpu_,
                                              false,
                                              false,
                                              renderer_prefer_igpu_,
                                              renderer_vulkan_timeout_,
                                              renderer_use_parallel_executors_,
                                              system_parameters_,
                                              renderer_enable_slab_allocation_));
    } catch (const std::exception& e) {
      LOG(ERROR) << "Backend rendering disabled: " << e.what();
    }
  }

  query_engine_ = QueryEngine::createInstance(data_mgr_->getCudaMgr(), cpu_mode_only_);

#ifdef ENABLE_GEOS
  if (!libgeos_so_filename_.empty()) {
    g_libgeos_so_filename.reset(new std::string(libgeos_so_filename_));
    LOG(INFO) << "Overriding default geos library with '" + *g_libgeos_so_filename + "'";
  }
#endif
}

DBHandler::~DBHandler() {
  shutdown();
}

void DBHandler::check_read_only(const std::string& str) {
  if (DBHandler::read_only_) {
    THROW_DB_EXCEPTION(str + " disabled: server running in read-only mode.");
  }
}

std::string const DBHandler::createInMemoryCalciteSession(
    const std::shared_ptr<Catalog_Namespace::Catalog>& catalog_ptr) {
  // We would create an in memory session for calcite with super user privileges which
  // would be used for getting all tables metadata when a user runs the query. The
  // session would be under the name of a proxy user/password which would only persist
  // till server's lifetime or execution of calcite query(in memory) whichever is the
  // earliest.
  heavyai::lock_guard<heavyai::shared_mutex> lg(calcite_sessions_mtx_);
  std::string session_id;
  do {
    session_id = generate_random_string(Catalog_Namespace::CALCITE_SESSION_ID_LENGTH);
  } while (calcite_sessions_.find(session_id) != calcite_sessions_.end());
  Catalog_Namespace::UserMetadata user_meta(-1,
                                            calcite_->getInternalSessionProxyUserName(),
                                            calcite_->getInternalSessionProxyPassword(),
                                            true,
                                            -1,
                                            true,
                                            false);
  const auto emplace_ret = calcite_sessions_.emplace(
      session_id,
      std::make_shared<Catalog_Namespace::SessionInfo>(
          catalog_ptr, user_meta, executor_device_type_, session_id));
  CHECK(emplace_ret.second);
  return session_id;
}

void DBHandler::removeInMemoryCalciteSession(const std::string& session_id) {
  // Remove InMemory calcite Session.
  heavyai::lock_guard<heavyai::shared_mutex> lg(calcite_sessions_mtx_);
  CHECK(calcite_sessions_.erase(session_id)) << session_id;
}

// internal connection for connections with no password
void DBHandler::internal_connect(TSessionId& session_id,
                                 const std::string& username,
                                 const std::string& dbname) {
  logger::set_new_request_id();
  auto stdlog = STDLOG();            // session_id set by connect_impl()
  std::string username2 = username;  // login() may reset username given as argument
  std::string dbname2 = dbname;      // login() may reset dbname given as argument
  Catalog_Namespace::UserMetadata user_meta;
  std::shared_ptr<Catalog> cat = nullptr;
  try {
    cat =
        SysCatalog::instance().login(dbname2, username2, std::string(), user_meta, false);
  } catch (std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }

  DBObject dbObject(dbname2, DatabaseDBObjectType);
  dbObject.loadKey(*cat);
  dbObject.setPrivileges(AccessPrivileges::ACCESS);
  std::vector<DBObject> dbObjects;
  dbObjects.push_back(dbObject);
  if (!SysCatalog::instance().checkPrivileges(user_meta, dbObjects)) {
    THROW_DB_EXCEPTION("Unauthorized Access: user " + user_meta.userLoggable() +
                       " is not allowed to access database " + dbname2 + ".");
  }
  connect_impl(session_id, std::string(), dbname2, user_meta, cat, stdlog);
}

bool DBHandler::isAggregator() const {
  return leaf_aggregator_.leafCount() > 0;
}

void DBHandler::krb5_connect(TKrb5Session& session,
                             const std::string& inputToken,
                             const std::string& dbname) {
  THROW_DB_EXCEPTION("Unauthrorized Access. Kerberos login not supported");
}

void DBHandler::connect(TSessionId& session_id,
                        const std::string& username,
                        const std::string& passwd,
                        const std::string& dbname) {
  logger::set_new_request_id();
  auto stdlog = STDLOG();  // session_info set by connect_impl()
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  std::string username2 = username;  // login() may reset username given as argument
  std::string dbname2 = dbname;      // login() may reset dbname given as argument
  Catalog_Namespace::UserMetadata user_meta;
  std::shared_ptr<Catalog> cat = nullptr;
  try {
    cat = SysCatalog::instance().login(
        dbname2, username2, passwd, user_meta, !super_user_rights_);
  } catch (std::exception& e) {
    stdlog.appendNameValuePairs("user", username, "db", dbname, "exception", e.what());
    THROW_DB_EXCEPTION(e.what());
  }

  DBObject dbObject(dbname2, DatabaseDBObjectType);
  dbObject.loadKey(*cat);
  dbObject.setPrivileges(AccessPrivileges::ACCESS);
  std::vector<DBObject> dbObjects;
  dbObjects.push_back(dbObject);
  if (!SysCatalog::instance().checkPrivileges(user_meta, dbObjects)) {
    stdlog.appendNameValuePairs(
        "user", username, "db", dbname, "exception", "Missing Privileges");
    THROW_DB_EXCEPTION("Unauthorized Access: user " + user_meta.userLoggable() +
                       " is not allowed to access database " + dbname2 + ".");
  }
  connect_impl(session_id, passwd, dbname2, user_meta, cat, stdlog);

  // if pki auth session_id will come back encrypted with user pubkey
  SysCatalog::instance().check_for_session_encryption(passwd, session_id);
}

void DBHandler::connect_impl(TSessionId& session_id,
                             const std::string& passwd,
                             const std::string& dbname,
                             const Catalog_Namespace::UserMetadata& user_meta,
                             std::shared_ptr<Catalog> cat,
                             query_state::StdLog& stdlog) {
  // TODO(sy): Is there any reason to have dbname as a parameter
  // here when the cat parameter already provides cat->name()?
  // Should dbname and cat->name() ever differ?
  auto session_ptr = sessions_store_->add(user_meta, cat, executor_device_type_);
  session_id = session_ptr->get_session_id();
  LOG(INFO) << "User " << user_meta.userLoggable() << " connected to database " << dbname;
  stdlog.setSessionInfo(session_ptr);
  session_ptr->set_connection_info(getConnectionInfo().toString());
  if (!super_user_rights_) {  // no need to connect to leaf_aggregator_ at this time
    // while doing warmup
  }
  auto const roles =
      stdlog.getConstSessionInfo()->get_currentUser().isSuper
          ? std::vector<std::string>{{"super"}}
          : SysCatalog::instance().getRoles(
                false, false, stdlog.getConstSessionInfo()->get_currentUser().userName);
  stdlog.appendNameValuePairs("roles", boost::algorithm::join(roles, ","));
}

void DBHandler::disconnect(const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session_ptr, "client", getConnectionInfo().toString());
  sessions_store_->disconnect(request_info.sessionId());
}

void DBHandler::disconnect_impl(Catalog_Namespace::SessionInfoPtr& session_ptr) {
  const auto session_id = session_ptr->get_session_id();
  std::exception_ptr leaf_exception = nullptr;
  try {
    if (leaf_aggregator_.leafCount() > 0) {
      leaf_aggregator_.disconnect(session_id);
    }
  } catch (...) {
    leaf_exception = std::current_exception();
  }

  if (render_handler_) {
    render_handler_->disconnect(session_id);
  }

  if (leaf_exception) {
    std::rethrow_exception(leaf_exception);
  }
}

void DBHandler::switch_database(const TSessionId& session_id_or_json,
                                const std::string& dbname) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session_ptr);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  std::string dbname2 = dbname;  // switchDatabase() may reset dbname given as argument
  try {
    std::shared_ptr<Catalog> cat = SysCatalog::instance().switchDatabase(
        dbname2, session_ptr->get_currentUser().userName);
    session_ptr->set_catalog_ptr(cat);
    if (leaf_aggregator_.leafCount() > 0) {
      leaf_aggregator_.switch_database(request_info.sessionId(), dbname);
      return;
    }
  } catch (std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::clone_session(TSessionId& session2_id,
                              const TSessionId& session1_id_or_json) {
  heavyai::RequestInfo const request_info(session1_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session1_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session1_ptr);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());

  try {
    const Catalog_Namespace::UserMetadata& user_meta = session1_ptr->get_currentUser();
    std::shared_ptr<Catalog> cat = session1_ptr->get_catalog_ptr();
    auto session2_ptr = sessions_store_->add(user_meta, cat, executor_device_type_);
    session2_id = session2_ptr->get_session_id();
    LOG(INFO) << "User " << user_meta.userLoggable() << " connected to database "
              << cat->name();
    if (leaf_aggregator_.leafCount() > 0) {
      leaf_aggregator_.clone_session(request_info.sessionId(), session2_id);
      return;
    }
  } catch (std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::interrupt(const TSessionId& query_session_id_or_json,
                          const TSessionId& interrupt_session_id_or_json) {
  // if this is for distributed setting, query_session becomes a parent session (agg)
  // and the interrupt session is one of existing session in the leaf node (leaf)
  // so we can think there exists a logical mapping
  // between query_session (agg) and interrupt_session (leaf)
  heavyai::RequestInfo const query_request_info(query_session_id_or_json);
  heavyai::RequestInfo const interrupt_request_info(interrupt_session_id_or_json);
  SET_REQUEST_ID(interrupt_request_info.requestId());
  auto session_ptr = get_session_ptr(interrupt_request_info.sessionId());
  auto& cat = session_ptr->getCatalog();
  auto stdlog = STDLOG(session_ptr);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  const auto allow_query_interrupt =
      g_enable_runtime_query_interrupt || g_enable_non_kernel_time_query_interrupt;
  if (g_enable_dynamic_watchdog || allow_query_interrupt) {
    const auto dbname = cat.getCurrentDB().dbName;
    auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID,
                                          jit_debug_ ? "/tmp" : "",
                                          jit_debug_ ? "mapdquery" : "",
                                          system_parameters_);
    CHECK(executor);

    if (leaf_aggregator_.leafCount() > 0) {
      leaf_aggregator_.interrupt(query_request_info.sessionId(),
                                 interrupt_request_info.sessionId());
    }
    auto target_executor_ids =
        executor->getExecutorIdsRunningQuery(query_request_info.sessionId());
    if (target_executor_ids.empty()) {
      heavyai::shared_lock<heavyai::shared_mutex> session_read_lock(
          executor->getSessionLock());
      if (executor->checkIsQuerySessionEnrolled(query_request_info.sessionId(),
                                                session_read_lock)) {
        session_read_lock.unlock();
        VLOG(1) << "Received interrupt: "
                << "User " << session_ptr->get_currentUser().userLoggable()
                << ", Database " << dbname << std::endl;
        executor->interrupt(query_request_info.sessionId(),
                            interrupt_request_info.sessionId());
      }
    } else {
      for (auto& executor_id : target_executor_ids) {
        VLOG(1) << "Received interrupt: "
                << "Executor " << executor_id << ", User "
                << session_ptr->get_currentUser().userLoggable() << ", Database "
                << dbname << std::endl;
        auto target_executor = Executor::getExecutor(executor_id);
        target_executor->interrupt(query_request_info.sessionId(),
                                   interrupt_request_info.sessionId());
      }
    }

    LOG(INFO) << "User " << session_ptr->get_currentUser().userName
              << " interrupted session with database " << dbname << std::endl;
  }
}

TRole::type DBHandler::getServerRole() const {
  if (g_cluster) {
    if (leaf_aggregator_.leafCount() > 0) {
      return TRole::type::AGGREGATOR;
    }
    return TRole::type::LEAF;
  }
  return TRole::type::SERVER;
}
void DBHandler::get_server_status(TServerStatus& _return,
                                  const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  const auto rendering_enabled = bool(render_handler_);
  _return.read_only = read_only_;
  _return.version = MAPD_RELEASE;
  _return.rendering_enabled = rendering_enabled;
  _return.start_time = start_time_;
  _return.edition = MAPD_EDITION;
  _return.host_name = heavyai::get_hostname();
  _return.poly_rendering_enabled = rendering_enabled;
  _return.role = getServerRole();
  _return.renderer_status_json =
      render_handler_ ? render_handler_->get_renderer_status_json() : "";
}

void DBHandler::get_status(std::vector<TServerStatus>& _return,
                           const TSessionId& session_id_or_json) {
  //
  // get_status() is now called locally at startup on the aggregator
  // in order to validate that all nodes of a cluster are running the
  // same software version and the same renderer status
  //
  // In that context, it is called with the InvalidSessionID, and
  // with the local super-user flag set.
  //
  // Hence, we allow this session-less mode only in distributed mode, and
  // then on a leaf (always), or on the aggregator (only in super-user mode)
  //
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto const allow_invalid_session = g_cluster && (!isAggregator() || super_user_rights_);

  if (!allow_invalid_session || request_info.sessionId() != getInvalidSessionId()) {
    auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
    stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  } else {
    LOG(INFO) << "get_status() called in session-less mode";
  }
  const auto rendering_enabled = bool(render_handler_);
  TServerStatus ret;
  ret.read_only = read_only_;
  ret.version = MAPD_RELEASE;
  ret.rendering_enabled = rendering_enabled;
  ret.start_time = start_time_;
  ret.edition = MAPD_EDITION;
  ret.host_name = heavyai::get_hostname();
  ret.poly_rendering_enabled = rendering_enabled;
  ret.role = getServerRole();
  ret.renderer_status_json =
      render_handler_ ? render_handler_->get_renderer_status_json() : "";
  ret.host_id = "";

  _return.push_back(ret);
  if (leaf_aggregator_.leafCount() > 0) {
    std::vector<TServerStatus> leaf_status =
        leaf_aggregator_.getLeafStatus(request_info.sessionId());
    _return.insert(_return.end(), leaf_status.begin(), leaf_status.end());
  }
}

void DBHandler::get_hardware_info(TClusterHardwareInfo& _return,
                                  const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  THardwareInfo ret;
  const auto cuda_mgr = data_mgr_->getCudaMgr();
  if (cuda_mgr) {
    ret.num_gpu_hw = cuda_mgr->getDeviceCount();
    ret.start_gpu = cuda_mgr->getStartGpu();
    if (ret.start_gpu >= 0) {
      ret.num_gpu_allocated = cuda_mgr->getDeviceCount() - cuda_mgr->getStartGpu();
      // ^ This will break as soon as we allow non contiguous GPU allocations to MapD
    }
    for (int16_t device_id = 0; device_id < ret.num_gpu_hw; device_id++) {
      TGpuSpecification gpu_spec;
      auto deviceProperties = cuda_mgr->getDeviceProperties(device_id);
      gpu_spec.num_sm = deviceProperties->numMPs;
      gpu_spec.clock_frequency_kHz = deviceProperties->clockKhz;
      gpu_spec.memory = deviceProperties->globalMem;
      gpu_spec.compute_capability_major = deviceProperties->computeMajor;
      gpu_spec.compute_capability_minor = deviceProperties->computeMinor;
      ret.gpu_info.push_back(gpu_spec);
    }
  }

  // start  hardware/OS dependent code
  ret.num_cpu_hw = std::thread::hardware_concurrency();
  // ^ This might return diffrent results in case of hyper threading
  // end hardware/OS dependent code

  _return.hardware_info.push_back(ret);
}

void DBHandler::get_session_info(TSessionInfo& _return,
                                 const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  CHECK(session_ptr);
  auto stdlog = STDLOG(session_ptr);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto user_metadata = session_ptr->get_currentUser();
  _return.user = user_metadata.userName;
  _return.database = session_ptr->getCatalog().getCurrentDB().dbName;
  _return.start_time = session_ptr->get_start_time();
  _return.is_super = user_metadata.isSuper;
}

void DBHandler::set_leaf_info(const TSessionId& session, const TLeafInfo& info) {
  g_distributed_leaf_idx = info.leaf_id;
  g_distributed_num_leaves = info.num_leaves;
}

void DBHandler::value_to_thrift_column(const TargetValue& tv,
                                       const SQLTypeInfo& ti,
                                       TColumn& column) {
  if (ti.is_array()) {
    CHECK(!ti.get_elem_type().get_notnull())
        << "element types of arrays should always be nullable";
    TColumn tColumn;
    const auto array_tv = boost::get<ArrayTargetValue>(&tv);
    CHECK(array_tv);
    bool is_null = !array_tv->is_initialized();
    if (!is_null) {
      const auto& vec = array_tv->get();
      for (const auto& elem_tv : vec) {
        value_to_thrift_column(elem_tv, ti.get_elem_type(), tColumn);
      }
    }
    column.data.arr_col.push_back(tColumn);
    column.nulls.push_back(is_null && !ti.get_notnull());
  } else if (ti.is_geometry()) {
    const auto scalar_tv = boost::get<ScalarTargetValue>(&tv);
    if (scalar_tv) {
      auto s_n = boost::get<NullableString>(scalar_tv);
      auto s = boost::get<std::string>(s_n);
      if (s) {
        column.data.str_col.push_back(*s);
      } else {
        column.data.str_col.emplace_back("");  // null string
        auto null_p = boost::get<void*>(s_n);
        CHECK(null_p && !*null_p);
      }
      column.nulls.push_back(!s && !ti.get_notnull());
    } else {
      const auto array_tv = boost::get<ArrayTargetValue>(&tv);
      CHECK(array_tv);
      bool is_null = !array_tv->is_initialized();
      if (!is_null) {
        auto elem_type = SQLTypeInfo(kDOUBLE, false);
        TColumn tColumn;
        const auto& vec = array_tv->get();
        for (const auto& elem_tv : vec) {
          value_to_thrift_column(elem_tv, elem_type, tColumn);
        }
        column.data.arr_col.push_back(tColumn);
        column.nulls.push_back(false);
      } else {
        TColumn tColumn;
        column.data.arr_col.push_back(tColumn);
        column.nulls.push_back(is_null && !ti.get_notnull());
      }
    }
  } else {
    CHECK(!ti.is_column());
    const auto scalar_tv = boost::get<ScalarTargetValue>(&tv);
    CHECK(scalar_tv);
    if (boost::get<int64_t>(scalar_tv)) {
      int64_t data = *(boost::get<int64_t>(scalar_tv));

      if (ti.is_decimal()) {
        double val = static_cast<double>(data);
        if (ti.get_scale() > 0) {
          val /= pow(10.0, std::abs(ti.get_scale()));
        }
        column.data.real_col.push_back(val);
      } else {
        column.data.int_col.push_back(data);
      }

      switch (ti.get_type()) {
        case kBOOLEAN:
          column.nulls.push_back(data == NULL_BOOLEAN && !ti.get_notnull());
          break;
        case kTINYINT:
          column.nulls.push_back(data == NULL_TINYINT && !ti.get_notnull());
          break;
        case kSMALLINT:
          column.nulls.push_back(data == NULL_SMALLINT && !ti.get_notnull());
          break;
        case kINT:
          column.nulls.push_back(data == NULL_INT && !ti.get_notnull());
          break;
        case kNUMERIC:
        case kDECIMAL:
        case kBIGINT:
          column.nulls.push_back(data == NULL_BIGINT && !ti.get_notnull());
          break;
        case kTIME:
        case kTIMESTAMP:
        case kDATE:
        case kINTERVAL_DAY_TIME:
        case kINTERVAL_YEAR_MONTH:
          column.nulls.push_back(data == NULL_BIGINT && !ti.get_notnull());
          break;
        default:
          column.nulls.push_back(false);
      }
    } else if (boost::get<double>(scalar_tv)) {
      double data = *(boost::get<double>(scalar_tv));
      column.data.real_col.push_back(data);
      if (ti.get_type() == kFLOAT) {
        column.nulls.push_back(data == NULL_FLOAT && !ti.get_notnull());
      } else {
        column.nulls.push_back(data == NULL_DOUBLE && !ti.get_notnull());
      }
    } else if (boost::get<float>(scalar_tv)) {
      CHECK_EQ(kFLOAT, ti.get_type());
      float data = *(boost::get<float>(scalar_tv));
      column.data.real_col.push_back(data);
      column.nulls.push_back(data == NULL_FLOAT && !ti.get_notnull());
    } else if (boost::get<NullableString>(scalar_tv)) {
      auto s_n = boost::get<NullableString>(scalar_tv);
      auto s = boost::get<std::string>(s_n);
      if (s) {
        column.data.str_col.push_back(*s);
      } else {
        column.data.str_col.emplace_back("");  // null string
        auto null_p = boost::get<void*>(s_n);
        CHECK(null_p && !*null_p);
      }
      column.nulls.push_back(!s && !ti.get_notnull());
    } else {
      CHECK(false);
    }
  }
}

TDatum DBHandler::value_to_thrift(const TargetValue& tv, const SQLTypeInfo& ti) {
  TDatum datum;
  const auto scalar_tv = boost::get<ScalarTargetValue>(&tv);
  if (!scalar_tv) {
    CHECK(ti.is_array());
    CHECK(!ti.get_elem_type().get_notnull())
        << "element types of arrays should always be nullable";
    const auto array_tv = boost::get<ArrayTargetValue>(&tv);
    CHECK(array_tv);
    if (array_tv->is_initialized()) {
      const auto& vec = array_tv->get();
      for (const auto& elem_tv : vec) {
        const auto scalar_col_val = value_to_thrift(elem_tv, ti.get_elem_type());
        datum.val.arr_val.push_back(scalar_col_val);
      }
      // Datum is not null, at worst it's an empty array Datum
      datum.is_null = false;
    } else {
      datum.is_null = true;
    }
    return datum;
  }
  if (boost::get<int64_t>(scalar_tv)) {
    int64_t data = *(boost::get<int64_t>(scalar_tv));

    if (ti.is_decimal()) {
      double val = static_cast<double>(data);
      if (ti.get_scale() > 0) {
        val /= pow(10.0, std::abs(ti.get_scale()));
      }
      datum.val.real_val = val;
    } else {
      datum.val.int_val = data;
    }

    switch (ti.get_type()) {
      case kBOOLEAN:
        datum.is_null = (datum.val.int_val == NULL_BOOLEAN);
        break;
      case kTINYINT:
        datum.is_null = (datum.val.int_val == NULL_TINYINT);
        break;
      case kSMALLINT:
        datum.is_null = (datum.val.int_val == NULL_SMALLINT);
        break;
      case kINT:
        datum.is_null = (datum.val.int_val == NULL_INT);
        break;
      case kDECIMAL:
      case kNUMERIC:
      case kBIGINT:
        datum.is_null = (datum.val.int_val == NULL_BIGINT);
        break;
      case kTIME:
      case kTIMESTAMP:
      case kDATE:
      case kINTERVAL_DAY_TIME:
      case kINTERVAL_YEAR_MONTH:
        datum.is_null = (datum.val.int_val == NULL_BIGINT);
        break;
      default:
        datum.is_null = false;
    }
  } else if (boost::get<double>(scalar_tv)) {
    datum.val.real_val = *(boost::get<double>(scalar_tv));
    if (ti.get_type() == kFLOAT) {
      datum.is_null = (datum.val.real_val == NULL_FLOAT);
    } else {
      datum.is_null = (datum.val.real_val == NULL_DOUBLE);
    }
  } else if (boost::get<float>(scalar_tv)) {
    CHECK_EQ(kFLOAT, ti.get_type());
    datum.val.real_val = *(boost::get<float>(scalar_tv));
    datum.is_null = (datum.val.real_val == NULL_FLOAT);
  } else if (boost::get<NullableString>(scalar_tv)) {
    auto s_n = boost::get<NullableString>(scalar_tv);
    auto s = boost::get<std::string>(s_n);
    if (s) {
      datum.val.str_val = *s;
    } else {
      auto null_p = boost::get<void*>(s_n);
      CHECK(null_p && !*null_p);
    }
    datum.is_null = !s;
  } else {
    CHECK(false);
  }
  return datum;
}

void DBHandler::sql_execute_local(
    TQueryResult& _return,
    const QueryStateProxy& query_state_proxy,
    const std::shared_ptr<Catalog_Namespace::SessionInfo> session_ptr,
    const std::string& query_str,
    const bool column_format,
    const std::string& nonce,
    const int32_t first_n,
    const int32_t at_most_n,
    const bool use_calcite) {
  _return.total_time_ms = 0;
  _return.nonce = nonce;
  ParserWrapper pw{query_str};
  switch (pw.getQueryType()) {
    case ParserWrapper::QueryType::Read: {
      _return.query_type = TQueryType::READ;
      VLOG(1) << "query type: READ";
      break;
    }
    case ParserWrapper::QueryType::Write: {
      _return.query_type = TQueryType::WRITE;
      VLOG(1) << "query type: WRITE";
      break;
    }
    case ParserWrapper::QueryType::SchemaRead: {
      _return.query_type = TQueryType::SCHEMA_READ;
      VLOG(1) << "query type: SCHEMA READ";
      break;
    }
    case ParserWrapper::QueryType::SchemaWrite: {
      _return.query_type = TQueryType::SCHEMA_WRITE;
      VLOG(1) << "query type: SCHEMA WRITE";
      break;
    }
    default: {
      _return.query_type = TQueryType::UNKNOWN;
      LOG(WARNING) << "query type: UNKNOWN";
      break;
    }
  }

  ExecutionResult result;
  lockmgr::LockedTableDescriptors locks;
  _return.total_time_ms += measure<>::execution([&]() {
    DBHandler::sql_execute_impl(result,
                                query_state_proxy,
                                column_format,
                                session_ptr->get_executor_device_type(),
                                first_n,
                                at_most_n,
                                use_calcite,
                                locks);
    DBHandler::convertData(
        _return, result, query_state_proxy, column_format, first_n, at_most_n);
  });
}

void DBHandler::convertData(TQueryResult& _return,
                            ExecutionResult& result,
                            const QueryStateProxy& query_state_proxy,
                            const bool column_format,
                            const int32_t first_n,
                            const int32_t at_most_n) {
  _return.execution_time_ms += result.getExecutionTime();
  if (result.empty()) {
    return;
  }

  switch (result.getResultType()) {
    case ExecutionResult::QueryResult:
      convertRows(_return,
                  query_state_proxy,
                  result.getTargetsMeta(),
                  *result.getRows(),
                  column_format,
                  first_n,
                  at_most_n);
      break;
    case ExecutionResult::SimpleResult:
      convertResult(_return, *result.getRows(), true);
      break;
    case ExecutionResult::Explanation:
      convertExplain(_return, *result.getRows(), true);
      break;
    case ExecutionResult::CalciteDdl:
      convertRows(_return,
                  query_state_proxy,
                  result.getTargetsMeta(),
                  *result.getRows(),
                  column_format,
                  -1,
                  -1);
      break;
  }
}

void DBHandler::sql_execute(TQueryResult& _return,
                            const TSessionId& session_id_or_json,
                            const std::string& query_str,
                            const bool column_format,
                            const std::string& nonce,
                            const int32_t first_n,
                            const int32_t at_most_n) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  const std::string exec_ra_prefix = "execute relalg";
  const bool use_calcite = !boost::starts_with(query_str, exec_ra_prefix);
  auto actual_query =
      use_calcite ? query_str : boost::trim_copy(query_str.substr(exec_ra_prefix.size()));
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto query_state = create_query_state(session_ptr, actual_query);
  auto stdlog = STDLOG(session_ptr, query_state);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  stdlog.appendNameValuePairs("nonce", nonce);
  auto timer = DEBUG_TIMER(__func__);
  try {
    ScopeGuard reset_was_deferred_copy_from = [this, &session_ptr] {
      deferred_copy_from_sessions.remove(session_ptr->get_session_id());
    };

    if (first_n >= 0 && at_most_n >= 0) {
      THROW_DB_EXCEPTION(std::string("At most one of first_n and at_most_n can be set"));
    }

    if (leaf_aggregator_.leafCount() > 0) {
      if (!agg_handler_) {
        THROW_DB_EXCEPTION("Distributed support is disabled.");
      }
      _return.total_time_ms = measure<>::execution([&]() {
        agg_handler_->cluster_execute(_return,
                                      query_state->createQueryStateProxy(),
                                      query_state->getQueryStr(),
                                      column_format,
                                      nonce,
                                      first_n,
                                      at_most_n,
                                      system_parameters_);
      });
      _return.nonce = nonce;
    } else {
      sql_execute_local(_return,
                        query_state->createQueryStateProxy(),
                        session_ptr,
                        actual_query,
                        column_format,
                        nonce,
                        first_n,
                        at_most_n,
                        use_calcite);
    }
    _return.total_time_ms += process_deferred_copy_from(request_info.sessionId());
    std::string debug_json = timer.stopAndGetJson();
    if (!debug_json.empty()) {
      _return.__set_debug(std::move(debug_json));
    }
    stdlog.appendNameValuePairs(
        "execution_time_ms",
        _return.execution_time_ms,
        "total_time_ms",  // BE-3420 - Redundant with duration field
        stdlog.duration<std::chrono::milliseconds>());
    VLOG(1) << "Table Schema Locks:\n" << lockmgr::TableSchemaLockMgr::instance();
    VLOG(1) << "Table Data Locks:\n" << lockmgr::TableDataLockMgr::instance();
  } catch (const std::exception& e) {
    if (strstr(e.what(), "java.lang.NullPointerException")) {
      THROW_DB_EXCEPTION("query failed from broken view or other schema related issue");
    } else if (strstr(e.what(), "SQL Error: Encountered \";\"")) {
      THROW_DB_EXCEPTION("multiple SQL statements not allowed");
    } else if (strstr(e.what(), "SQL Error: Encountered \"<EOF>\" at line 0, column 0")) {
      THROW_DB_EXCEPTION("empty SQL statment not allowed");
    } else {
      THROW_DB_EXCEPTION(e.what());
    }
  }
}

void DBHandler::sql_execute(ExecutionResult& _return,
                            const TSessionId& session_id_or_json,
                            const std::string& query_str,
                            const bool column_format,
                            const int32_t first_n,
                            const int32_t at_most_n,
                            lockmgr::LockedTableDescriptors& locks) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  const std::string exec_ra_prefix = "execute relalg";
  const bool use_calcite = !boost::starts_with(query_str, exec_ra_prefix);
  auto actual_query =
      use_calcite ? query_str : boost::trim_copy(query_str.substr(exec_ra_prefix.size()));

  auto session_ptr = get_session_ptr(request_info.sessionId());
  CHECK(session_ptr);
  auto query_state = create_query_state(session_ptr, actual_query);
  auto stdlog = STDLOG(session_ptr, query_state);
  auto timer = DEBUG_TIMER(__func__);

  try {
    ScopeGuard reset_was_deferred_copy_from = [this, &session_ptr] {
      deferred_copy_from_sessions.remove(session_ptr->get_session_id());
    };

    if (first_n >= 0 && at_most_n >= 0) {
      THROW_DB_EXCEPTION(std::string("At most one of first_n and at_most_n can be set"));
    }
    auto total_time_ms = measure<>::execution([&]() {
      DBHandler::sql_execute_impl(_return,
                                  query_state->createQueryStateProxy(),
                                  column_format,
                                  session_ptr->get_executor_device_type(),
                                  first_n,
                                  at_most_n,
                                  use_calcite,
                                  locks);
    });

    _return.setExecutionTime(total_time_ms +
                             process_deferred_copy_from(request_info.sessionId()));

    stdlog.appendNameValuePairs(
        "execution_time_ms",
        _return.getExecutionTime(),
        "total_time_ms",  // BE-3420 - Redundant with duration field
        stdlog.duration<std::chrono::milliseconds>());
    VLOG(1) << "Table Schema Locks:\n" << lockmgr::TableSchemaLockMgr::instance();
    VLOG(1) << "Table Data Locks:\n" << lockmgr::TableDataLockMgr::instance();
  } catch (const std::exception& e) {
    if (strstr(e.what(), "java.lang.NullPointerException")) {
      THROW_DB_EXCEPTION("query failed from broken view or other schema related issue");
    } else if (strstr(e.what(), "SQL Error: Encountered \";\"")) {
      THROW_DB_EXCEPTION("multiple SQL statements not allowed");
    } else if (strstr(e.what(), "SQL Error: Encountered \"<EOF>\" at line 0, column 0")) {
      THROW_DB_EXCEPTION("empty SQL statment not allowed");
    } else {
      THROW_DB_EXCEPTION(e.what());
    }
  }
}

int64_t DBHandler::process_deferred_copy_from(const TSessionId& session_id) {
  int64_t total_time_ms(0);
  // if the SQL statement we just executed was a geo COPY FROM, the import
  // parameters were captured, and this flag set, so we do the actual import here
  if (auto deferred_copy_from_state = deferred_copy_from_sessions(session_id)) {
    // import_geo_table() calls create_table() which calls this function to
    // do the work, so reset the flag now to avoid executing this part a
    // second time at the end of that, which would fail as the table was
    // already created! Also reset the flag with a ScopeGuard on exiting
    // this function any other way, such as an exception from the code above!
    deferred_copy_from_sessions.remove(session_id);

    // create table as replicated?
    TCreateParams create_params;
    if (deferred_copy_from_state->partitions == "REPLICATED") {
      create_params.is_replicated = true;
    }

    // now do (and time) the import
    total_time_ms = measure<>::execution([&]() {
      importGeoTableGlobFilterSort(session_id,
                                   deferred_copy_from_state->table,
                                   deferred_copy_from_state->file_name,
                                   deferred_copy_from_state->copy_params,
                                   TRowDescriptor(),
                                   create_params);
    });
  }
  return total_time_ms;
}

void DBHandler::sql_execute_df(TDataFrame& _return,
                               const TSessionId& session_id_or_json,
                               const std::string& query_str,
                               const TDeviceType::type results_device_type,
                               const int32_t device_id,
                               const int32_t first_n,
                               const TArrowTransport::type transport_method) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  CHECK(session_ptr);
  auto query_state = create_query_state(session_ptr, query_str);
  auto stdlog = STDLOG(session_ptr, query_state);

  const auto executor_device_type = session_ptr->get_executor_device_type();

  if (results_device_type == TDeviceType::GPU) {
    if (executor_device_type != ExecutorDeviceType::GPU) {
      THROW_DB_EXCEPTION(std::string("GPU mode is not allowed in this session"));
    }
    if (!data_mgr_->gpusPresent()) {
      THROW_DB_EXCEPTION(std::string("No GPU is available in this server"));
    }
    if (device_id < 0 || device_id >= data_mgr_->getCudaMgr()->getDeviceCount()) {
      THROW_DB_EXCEPTION(
          std::string("Invalid device_id or unavailable GPU with this ID"));
    }
  }
  ParserWrapper pw{query_str};
  if (pw.getQueryType() != ParserWrapper::QueryType::Read) {
    THROW_DB_EXCEPTION(std::string(
        "Only read queries supported for the Arrow sql_execute_df endpoint."));
  }
  if (ExplainInfo(query_str).isCalciteExplain()) {
    THROW_DB_EXCEPTION(std::string(
        "Explain is currently unsupported by the Arrow sql_execute_df endpoint."));
  }

  ExecutionResult execution_result;
  lockmgr::LockedTableDescriptors locks;
  sql_execute_impl(execution_result,
                   query_state->createQueryStateProxy(),
                   true, /* column_format - does this do anything? */
                   executor_device_type,
                   first_n,
                   -1, /* at_most_n */
                   true,
                   locks);

  const auto result_set = execution_result.getRows();
  const auto executor_results_device_type = results_device_type == TDeviceType::CPU
                                                ? ExecutorDeviceType::CPU
                                                : ExecutorDeviceType::GPU;
  _return.execution_time_ms =
      execution_result.getExecutionTime() - result_set->getQueueTime();
  const auto converter = std::make_unique<ArrowResultSetConverter>(
      result_set,
      data_mgr_,
      executor_results_device_type,
      device_id,
      getTargetNames(execution_result.getTargetsMeta()),
      first_n,
      ArrowTransport(transport_method));
  ArrowResult arrow_result;
  _return.arrow_conversion_time_ms +=
      measure<>::execution([&] { arrow_result = converter->getArrowResult(); });
  _return.sm_handle =
      std::string(arrow_result.sm_handle.begin(), arrow_result.sm_handle.end());
  _return.sm_size = arrow_result.sm_size;
  _return.df_handle =
      std::string(arrow_result.df_handle.begin(), arrow_result.df_handle.end());
  _return.df_buffer =
      std::string(arrow_result.df_buffer.begin(), arrow_result.df_buffer.end());
  if (executor_results_device_type == ExecutorDeviceType::GPU) {
    std::lock_guard<std::mutex> map_lock(handle_to_dev_ptr_mutex_);
    CHECK(!ipc_handle_to_dev_ptr_.count(_return.df_handle));
    ipc_handle_to_dev_ptr_.insert(
        std::make_pair(_return.df_handle, arrow_result.serialized_cuda_handle));
  }
  _return.df_size = arrow_result.df_size;
}

void DBHandler::sql_execute_gdf(TDataFrame& _return,
                                const TSessionId& session_id_or_json,
                                const std::string& query_str,
                                const int32_t device_id,
                                const int32_t first_n) {
  heavyai::RequestInfo request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  request_info.setRequestId(logger::request_id());
  sql_execute_df(_return,
                 request_info.json(),
                 query_str,
                 TDeviceType::GPU,
                 device_id,
                 first_n,
                 TArrowTransport::SHARED_MEMORY);
}

// For now we have only one user of a data frame in all cases.
void DBHandler::deallocate_df(const TSessionId& session_id_or_json,
                              const TDataFrame& df,
                              const TDeviceType::type device_type,
                              const int32_t device_id) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  std::string serialized_cuda_handle = "";
  if (device_type == TDeviceType::GPU) {
    std::lock_guard<std::mutex> map_lock(handle_to_dev_ptr_mutex_);
    if (ipc_handle_to_dev_ptr_.count(df.df_handle) != size_t(1)) {
      TDBException ex;
      ex.error_msg = std::string(
          "Current data frame handle is not bookkept or been inserted "
          "twice");
      LOG(ERROR) << ex.error_msg;
      throw ex;
    }
    serialized_cuda_handle = ipc_handle_to_dev_ptr_[df.df_handle];
    ipc_handle_to_dev_ptr_.erase(df.df_handle);
  }
  std::vector<char> sm_handle(df.sm_handle.begin(), df.sm_handle.end());
  std::vector<char> df_handle(df.df_handle.begin(), df.df_handle.end());
  ArrowResult result{
      sm_handle, df.sm_size, df_handle, df.df_size, serialized_cuda_handle};
  ArrowResultSet::deallocateArrowResultBuffer(
      result,
      device_type == TDeviceType::CPU ? ExecutorDeviceType::CPU : ExecutorDeviceType::GPU,
      device_id,
      data_mgr_);
}

void DBHandler::sql_validate(TRowDescriptor& _return,
                             const TSessionId& session_id_or_json,
                             const std::string& query_str) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  try {
    auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
    stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
    auto query_state = create_query_state(stdlog.getSessionInfo(), query_str);
    stdlog.setQueryState(query_state);

    ParserWrapper pw{query_str};
    if (ExplainInfo(query_str).isExplain() || pw.is_ddl || pw.is_update_dml) {
      throw std::runtime_error("Can only validate SELECT statements.");
    }

    const auto execute_read_lock = legacylockmgr::getExecuteReadLock();

    TPlanResult parse_result;
    lockmgr::LockedTableDescriptors locks;
    std::tie(parse_result, locks) = parse_to_ra(query_state->createQueryStateProxy(),
                                                query_state->getQueryStr(),
                                                {},
                                                true,
                                                system_parameters_,
                                                /*check_privileges=*/true);
    const auto query_ra = parse_result.plan_result;
    _return = validateRelAlg(query_ra, query_state->createQueryStateProxy());
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(std::string(e.what()));
  }
}

namespace {

struct ProjectionTokensForCompletion {
  std::unordered_set<std::string> uc_column_names;
  std::unordered_set<std::string> uc_column_table_qualifiers;
};

// Extract what looks like a (qualified) identifier from the partial query.
// The results will be used to rank the auto-completion results: tables which
// contain at least one of the identifiers first.
ProjectionTokensForCompletion extract_projection_tokens_for_completion(
    const std::string& sql) {
  boost::regex id_regex{R"(([[:alnum:]]|_|\.)+)",
                        boost::regex::extended | boost::regex::icase};
  boost::sregex_token_iterator tok_it(sql.begin(), sql.end(), id_regex, 0);
  boost::sregex_token_iterator end;
  std::unordered_set<std::string> uc_column_names;
  std::unordered_set<std::string> uc_column_table_qualifiers;
  for (; tok_it != end; ++tok_it) {
    std::string column_name = *tok_it;
    std::vector<std::string> column_tokens;
    boost::split(column_tokens, column_name, boost::is_any_of("."));
    if (column_tokens.size() == 2) {
      // If the column name is qualified, take user's word.
      uc_column_table_qualifiers.insert(to_upper(column_tokens.front()));
    } else {
      uc_column_names.insert(to_upper(column_name));
    }
  }
  return {uc_column_names, uc_column_table_qualifiers};
}

}  // namespace

void DBHandler::get_completion_hints(std::vector<TCompletionHint>& hints,
                                     const TSessionId& session_id_or_json,
                                     const std::string& sql,
                                     const int cursor) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  std::vector<std::string> visible_tables;  // Tables allowed for the given session.
  get_completion_hints_unsorted(hints, visible_tables, stdlog, sql, cursor);
  const auto proj_tokens = extract_projection_tokens_for_completion(sql);
  auto compatible_table_names = get_uc_compatible_table_names_by_column(
      proj_tokens.uc_column_names, visible_tables, stdlog);
  // Add the table qualifiers explicitly specified by the user.
  compatible_table_names.insert(proj_tokens.uc_column_table_qualifiers.begin(),
                                proj_tokens.uc_column_table_qualifiers.end());
  // Sort the hints by category, from COLUMN (most specific) to KEYWORD.
  std::sort(
      hints.begin(),
      hints.end(),
      [&compatible_table_names](const TCompletionHint& lhs, const TCompletionHint& rhs) {
        if (lhs.type == TCompletionHintType::TABLE &&
            rhs.type == TCompletionHintType::TABLE) {
          // Between two tables, one which is compatible with the specified
          // projections and one which isn't, pick the one which is compatible.
          if (compatible_table_names.find(to_upper(lhs.hints.back())) !=
                  compatible_table_names.end() &&
              compatible_table_names.find(to_upper(rhs.hints.back())) ==
                  compatible_table_names.end()) {
            return true;
          }
        }
        return lhs.type < rhs.type;
      });
}

void DBHandler::get_completion_hints_unsorted(std::vector<TCompletionHint>& hints,
                                              std::vector<std::string>& visible_tables,
                                              query_state::StdLog& stdlog,
                                              const std::string& sql,
                                              const int cursor) {
  const auto& session_info = *stdlog.getConstSessionInfo();
  try {
    get_tables_impl(visible_tables, session_info, GET_PHYSICAL_TABLES_AND_VIEWS);

    // Filter out keywords suggested by Calcite which we don't support.
    hints = just_whitelisted_keyword_hints(
        calcite_->getCompletionHints(session_info, visible_tables, sql, cursor));
  } catch (const std::exception& e) {
    TDBException ex;
    ex.error_msg = std::string(e.what());
    LOG(ERROR) << ex.error_msg;
    throw ex;
  }
  boost::regex from_expr{R"(\s+from\s+)", boost::regex::extended | boost::regex::icase};
  const size_t length_to_cursor =
      cursor < 0 ? sql.size() : std::min(sql.size(), static_cast<size_t>(cursor));
  // Trust hints from Calcite after the FROM keyword.
  if (boost::regex_search(sql.cbegin(), sql.cbegin() + length_to_cursor, from_expr)) {
    return;
  }
  // Before FROM, the query is too incomplete for context-sensitive completions.
  get_token_based_completions(hints, stdlog, visible_tables, sql, cursor);
}

void DBHandler::get_token_based_completions(std::vector<TCompletionHint>& hints,
                                            query_state::StdLog& stdlog,
                                            std::vector<std::string>& visible_tables,
                                            const std::string& sql,
                                            const int cursor) {
  const auto last_word =
      find_last_word_from_cursor(sql, cursor < 0 ? sql.size() : cursor);
  boost::regex select_expr{R"(\s*select\s+)",
                           boost::regex::extended | boost::regex::icase};
  const size_t length_to_cursor =
      cursor < 0 ? sql.size() : std::min(sql.size(), static_cast<size_t>(cursor));
  // After SELECT but before FROM, look for all columns in all tables which match the
  // prefix.
  if (boost::regex_search(sql.cbegin(), sql.cbegin() + length_to_cursor, select_expr)) {
    const auto column_names_by_table = fill_column_names_by_table(visible_tables, stdlog);
    // Trust the fully qualified columns the most.
    if (get_qualified_column_hints(hints, last_word, column_names_by_table)) {
      return;
    }
    // Not much information to use, just retrieve column names which match the prefix.
    if (should_suggest_column_hints(sql)) {
      get_column_hints(hints, last_word, column_names_by_table);
      return;
    }
    const std::string kFromKeyword{"FROM"};
    if (boost::istarts_with(kFromKeyword, last_word)) {
      TCompletionHint keyword_hint;
      keyword_hint.type = TCompletionHintType::KEYWORD;
      keyword_hint.replaced = last_word;
      keyword_hint.hints.emplace_back(kFromKeyword);
      hints.push_back(keyword_hint);
    }
  } else {
    const std::string kSelectKeyword{"SELECT"};
    if (boost::istarts_with(kSelectKeyword, last_word)) {
      TCompletionHint keyword_hint;
      keyword_hint.type = TCompletionHintType::KEYWORD;
      keyword_hint.replaced = last_word;
      keyword_hint.hints.emplace_back(kSelectKeyword);
      hints.push_back(keyword_hint);
    }
  }
}

std::unordered_map<std::string, std::unordered_set<std::string>>
DBHandler::fill_column_names_by_table(std::vector<std::string>& table_names,
                                      query_state::StdLog& stdlog) {
  std::unordered_map<std::string, std::unordered_set<std::string>> column_names_by_table;
  for (auto it = table_names.begin(); it != table_names.end();) {
    TTableDetails table_details;
    try {
      get_table_details_impl(table_details, stdlog, *it, false, false);
    } catch (const TDBException& e) {
      // Remove the corrupted Table/View name from the list for further processing.
      it = table_names.erase(it);
      continue;
    }
    for (const auto& column_type : table_details.row_desc) {
      column_names_by_table[*it].emplace(column_type.col_name);
    }
    ++it;
  }
  return column_names_by_table;
}

ConnectionInfo DBHandler::getConnectionInfo() const {
  return ConnectionInfo{TrackingProcessor::client_address,
                        TrackingProcessor::client_protocol};
}

std::unordered_set<std::string> DBHandler::get_uc_compatible_table_names_by_column(
    const std::unordered_set<std::string>& uc_column_names,
    std::vector<std::string>& table_names,
    query_state::StdLog& stdlog) {
  std::unordered_set<std::string> compatible_table_names_by_column;
  for (auto it = table_names.begin(); it != table_names.end();) {
    TTableDetails table_details;
    try {
      get_table_details_impl(table_details, stdlog, *it, false, false);
    } catch (const TDBException& e) {
      // Remove the corrupted Table/View name from the list for further processing.
      it = table_names.erase(it);
      continue;
    }
    for (const auto& column_type : table_details.row_desc) {
      if (uc_column_names.find(to_upper(column_type.col_name)) != uc_column_names.end()) {
        compatible_table_names_by_column.emplace(to_upper(*it));
        break;
      }
    }
    ++it;
  }
  return compatible_table_names_by_column;
}

void DBHandler::dispatch_query_task(std::shared_ptr<QueryDispatchQueue::Task> query_task,
                                    const bool is_update_delete) {
  CHECK(dispatch_queue_);
  dispatch_queue_->submit(std::move(query_task), is_update_delete);
}

TRowDescriptor DBHandler::validateRelAlg(const std::string& query_ra,
                                         QueryStateProxy query_state_proxy) {
  TQueryResult query_result;
  ExecutionResult execution_result;
  auto execute_rel_alg_task = std::make_shared<QueryDispatchQueue::Task>(
      [this,
       &execution_result,
       query_state_proxy,
       &query_ra,
       parent_thread_local_ids =
           logger::thread_local_ids()](const size_t executor_index) {
        logger::LocalIdsScopeGuard lisg = parent_thread_local_ids.setNewThreadId();
        execute_rel_alg(execution_result,
                        query_state_proxy,
                        query_ra,
                        true,
                        ExecutorDeviceType::CPU,
                        -1,
                        -1,
                        /*just_validate=*/true,
                        /*find_filter_push_down_candidates=*/false,
                        ExplainInfo(),
                        executor_index);
      });
  dispatch_query_task(execute_rel_alg_task, /*is_update_delete=*/false);
  auto result_future = execute_rel_alg_task->get_future();
  result_future.get();
  DBHandler::convertData(query_result, execution_result, query_state_proxy, true, -1, -1);

  const auto& row_desc = query_result.row_set.row_desc;
  const auto& targets_meta = execution_result.getTargetsMeta();
  CHECK_EQ(row_desc.size(), targets_meta.size());

  // TODO: Below fixup logic should no longer be needed after the comp_param refactor
  TRowDescriptor fixedup_row_desc;
  for (size_t i = 0; i < row_desc.size(); i++) {
    const auto& col_desc = row_desc[i];
    auto fixedup_col_desc = col_desc;
    if (col_desc.col_type.encoding == TEncodingType::DICT &&
        col_desc.col_type.comp_param > 0) {
      const auto& type_info = targets_meta[i].get_type_info();
      CHECK_EQ(type_info.get_compression(), kENCODING_DICT);
      const auto cat = Catalog_Namespace::SysCatalog::instance().getCatalog(
          type_info.getStringDictKey().db_id);
      const auto dd = cat->getMetadataForDict(col_desc.col_type.comp_param, false);
      CHECK(dd);
      fixedup_col_desc.col_type.comp_param = dd->dictNBits;
    }
    fixedup_row_desc.push_back(fixedup_col_desc);
  }
  return fixedup_row_desc;
}

void DBHandler::get_roles(std::vector<std::string>& roles,
                          const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    // WARNING: This appears to not include roles a user is a member of,
    // if the role has no permissions granted to it.
    roles =
        SysCatalog::instance().getRoles(session_ptr->get_currentUser().userName,
                                        session_ptr->getCatalog().getCurrentDB().dbId);
  } else {
    roles = SysCatalog::instance().getRoles(
        false, true, session_ptr->get_currentUser().userName);
  }
}

bool DBHandler::has_role(const TSessionId& session_id_or_json,
                         const std::string& granteeName,
                         const std::string& roleName) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  const auto session_ptr = get_session_ptr(request_info.sessionId());
  const auto stdlog = STDLOG(session_ptr);
  const auto current_user = session_ptr->get_currentUser();
  if (!current_user.isSuper) {
    if (const auto* user = SysCatalog::instance().getUserGrantee(granteeName);
        user && current_user.userName != granteeName) {
      THROW_DB_EXCEPTION("Only super users can check other user's roles.");
    } else if (!SysCatalog::instance().isRoleGrantedToGrantee(
                   current_user.userName, granteeName, true)) {
      THROW_DB_EXCEPTION(
          "Only super users can check roles assignment that have not been directly "
          "granted to a user.");
    }
  }
  return SysCatalog::instance().isRoleGrantedToGrantee(granteeName, roleName, false);
}

static TDBObject serialize_db_object(const std::string& roleName,
                                     const DBObject& inObject) {
  TDBObject outObject;
  outObject.objectName = inObject.getName();
  outObject.grantee = roleName;
  outObject.objectId = inObject.getObjectKey().objectId;
  const auto ap = inObject.getPrivileges();
  switch (inObject.getObjectKey().permissionType) {
    case DatabaseDBObjectType:
      outObject.privilegeObjectType = TDBObjectType::DatabaseDBObjectType;
      outObject.privs.push_back(ap.hasPermission(DatabasePrivileges::CREATE_DATABASE));
      outObject.privs.push_back(ap.hasPermission(DatabasePrivileges::DROP_DATABASE));
      outObject.privs.push_back(ap.hasPermission(DatabasePrivileges::VIEW_SQL_EDITOR));
      outObject.privs.push_back(ap.hasPermission(DatabasePrivileges::ACCESS));

      break;
    case TableDBObjectType:
      outObject.privilegeObjectType = TDBObjectType::TableDBObjectType;
      outObject.privs.push_back(ap.hasPermission(TablePrivileges::CREATE_TABLE));
      outObject.privs.push_back(ap.hasPermission(TablePrivileges::DROP_TABLE));
      outObject.privs.push_back(ap.hasPermission(TablePrivileges::SELECT_FROM_TABLE));
      outObject.privs.push_back(ap.hasPermission(TablePrivileges::INSERT_INTO_TABLE));
      outObject.privs.push_back(ap.hasPermission(TablePrivileges::UPDATE_IN_TABLE));
      outObject.privs.push_back(ap.hasPermission(TablePrivileges::DELETE_FROM_TABLE));
      outObject.privs.push_back(ap.hasPermission(TablePrivileges::TRUNCATE_TABLE));
      outObject.privs.push_back(ap.hasPermission(TablePrivileges::ALTER_TABLE));

      break;
    case DashboardDBObjectType:
      outObject.privilegeObjectType = TDBObjectType::DashboardDBObjectType;
      outObject.privs.push_back(ap.hasPermission(DashboardPrivileges::CREATE_DASHBOARD));
      outObject.privs.push_back(ap.hasPermission(DashboardPrivileges::DELETE_DASHBOARD));
      outObject.privs.push_back(ap.hasPermission(DashboardPrivileges::VIEW_DASHBOARD));
      outObject.privs.push_back(ap.hasPermission(DashboardPrivileges::EDIT_DASHBOARD));

      break;
    case ViewDBObjectType:
      outObject.privilegeObjectType = TDBObjectType::ViewDBObjectType;
      outObject.privs.push_back(ap.hasPermission(ViewPrivileges::CREATE_VIEW));
      outObject.privs.push_back(ap.hasPermission(ViewPrivileges::DROP_VIEW));
      outObject.privs.push_back(ap.hasPermission(ViewPrivileges::SELECT_FROM_VIEW));
      outObject.privs.push_back(ap.hasPermission(ViewPrivileges::INSERT_INTO_VIEW));
      outObject.privs.push_back(ap.hasPermission(ViewPrivileges::UPDATE_IN_VIEW));
      outObject.privs.push_back(ap.hasPermission(ViewPrivileges::DELETE_FROM_VIEW));

      break;
    case ServerDBObjectType:
      outObject.privilegeObjectType = TDBObjectType::ServerDBObjectType;
      outObject.privs.push_back(ap.hasPermission(ServerPrivileges::CREATE_SERVER));
      outObject.privs.push_back(ap.hasPermission(ServerPrivileges::DROP_SERVER));
      outObject.privs.push_back(ap.hasPermission(ServerPrivileges::ALTER_SERVER));
      outObject.privs.push_back(ap.hasPermission(ServerPrivileges::SERVER_USAGE));

      break;
    default:
      CHECK(false);
  }
  const int type_val = static_cast<int>(inObject.getType());
  CHECK(type_val >= 0 && type_val < 6);
  outObject.objectType = static_cast<TDBObjectType::type>(type_val);
  return outObject;
}

bool DBHandler::has_database_permission(const AccessPrivileges& privs,
                                        const TDBObjectPermissions& permissions) {
  if (!permissions.__isset.database_permissions_) {
    THROW_DB_EXCEPTION("Database permissions not set for check.")
  }
  auto perms = permissions.database_permissions_;
  if ((perms.create_ && !privs.hasPermission(DatabasePrivileges::CREATE_DATABASE)) ||
      (perms.delete_ && !privs.hasPermission(DatabasePrivileges::DROP_DATABASE)) ||
      (perms.view_sql_editor_ &&
       !privs.hasPermission(DatabasePrivileges::VIEW_SQL_EDITOR)) ||
      (perms.access_ && !privs.hasPermission(DatabasePrivileges::ACCESS))) {
    return false;
  } else {
    return true;
  }
}

bool DBHandler::has_table_permission(const AccessPrivileges& privs,
                                     const TDBObjectPermissions& permissions) {
  if (!permissions.__isset.table_permissions_) {
    THROW_DB_EXCEPTION("Table permissions not set for check.")
  }
  auto perms = permissions.table_permissions_;
  if ((perms.create_ && !privs.hasPermission(TablePrivileges::CREATE_TABLE)) ||
      (perms.drop_ && !privs.hasPermission(TablePrivileges::DROP_TABLE)) ||
      (perms.select_ && !privs.hasPermission(TablePrivileges::SELECT_FROM_TABLE)) ||
      (perms.insert_ && !privs.hasPermission(TablePrivileges::INSERT_INTO_TABLE)) ||
      (perms.update_ && !privs.hasPermission(TablePrivileges::UPDATE_IN_TABLE)) ||
      (perms.delete_ && !privs.hasPermission(TablePrivileges::DELETE_FROM_TABLE)) ||
      (perms.truncate_ && !privs.hasPermission(TablePrivileges::TRUNCATE_TABLE)) ||
      (perms.alter_ && !privs.hasPermission(TablePrivileges::ALTER_TABLE))) {
    return false;
  } else {
    return true;
  }
}

bool DBHandler::has_dashboard_permission(const AccessPrivileges& privs,
                                         const TDBObjectPermissions& permissions) {
  if (!permissions.__isset.dashboard_permissions_) {
    THROW_DB_EXCEPTION("Dashboard permissions not set for check.")
  }
  auto perms = permissions.dashboard_permissions_;
  if ((perms.create_ && !privs.hasPermission(DashboardPrivileges::CREATE_DASHBOARD)) ||
      (perms.delete_ && !privs.hasPermission(DashboardPrivileges::DELETE_DASHBOARD)) ||
      (perms.view_ && !privs.hasPermission(DashboardPrivileges::VIEW_DASHBOARD)) ||
      (perms.edit_ && !privs.hasPermission(DashboardPrivileges::EDIT_DASHBOARD))) {
    return false;
  } else {
    return true;
  }
}

bool DBHandler::has_view_permission(const AccessPrivileges& privs,
                                    const TDBObjectPermissions& permissions) {
  if (!permissions.__isset.view_permissions_) {
    THROW_DB_EXCEPTION("View permissions not set for check.")
  }
  auto perms = permissions.view_permissions_;
  if ((perms.create_ && !privs.hasPermission(ViewPrivileges::CREATE_VIEW)) ||
      (perms.drop_ && !privs.hasPermission(ViewPrivileges::DROP_VIEW)) ||
      (perms.select_ && !privs.hasPermission(ViewPrivileges::SELECT_FROM_VIEW)) ||
      (perms.insert_ && !privs.hasPermission(ViewPrivileges::INSERT_INTO_VIEW)) ||
      (perms.update_ && !privs.hasPermission(ViewPrivileges::UPDATE_IN_VIEW)) ||
      (perms.delete_ && !privs.hasPermission(ViewPrivileges::DELETE_FROM_VIEW))) {
    return false;
  } else {
    return true;
  }
}

bool DBHandler::has_server_permission(const AccessPrivileges& privs,
                                      const TDBObjectPermissions& permissions) {
  CHECK(permissions.__isset.server_permissions_);
  auto perms = permissions.server_permissions_;
  if ((perms.create_ && !privs.hasPermission(ServerPrivileges::CREATE_SERVER)) ||
      (perms.drop_ && !privs.hasPermission(ServerPrivileges::DROP_SERVER)) ||
      (perms.alter_ && !privs.hasPermission(ServerPrivileges::ALTER_SERVER)) ||
      (perms.usage_ && !privs.hasPermission(ServerPrivileges::SERVER_USAGE))) {
    return false;
  } else {
    return true;
  }
}

bool DBHandler::has_object_privilege(const TSessionId& session_id_or_json,
                                     const std::string& granteeName,
                                     const std::string& objectName,
                                     const TDBObjectType::type objectType,
                                     const TDBObjectPermissions& permissions) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session_ptr);
  auto const& cat = session_ptr->getCatalog();
  auto const& current_user = session_ptr->get_currentUser();
  if (!current_user.isSuper && !SysCatalog::instance().isRoleGrantedToGrantee(
                                   current_user.userName, granteeName, false)) {
    THROW_DB_EXCEPTION(
        "Users except superusers can only check privileges for self or roles granted "
        "to "
        "them.")
  }
  Catalog_Namespace::UserMetadata user_meta;
  if (SysCatalog::instance().getMetadataForUser(granteeName, user_meta) &&
      user_meta.isSuper) {
    return true;
  }
  Grantee* grnt = SysCatalog::instance().getGrantee(granteeName);
  if (!grnt) {
    THROW_DB_EXCEPTION("User or Role " + granteeName + " does not exist.")
  }
  DBObjectType type;
  std::string func_name;
  switch (objectType) {
    case TDBObjectType::DatabaseDBObjectType:
      type = DBObjectType::DatabaseDBObjectType;
      func_name = "database";
      break;
    case TDBObjectType::TableDBObjectType:
      type = DBObjectType::TableDBObjectType;
      func_name = "table";
      break;
    case TDBObjectType::DashboardDBObjectType:
      type = DBObjectType::DashboardDBObjectType;
      func_name = "dashboard";
      break;
    case TDBObjectType::ViewDBObjectType:
      type = DBObjectType::ViewDBObjectType;
      func_name = "view";
      break;
    case TDBObjectType::ServerDBObjectType:
      type = DBObjectType::ServerDBObjectType;
      func_name = "server";
      break;
    default:
      THROW_DB_EXCEPTION("Invalid object type (" + std::to_string(objectType) + ").");
  }
  DBObject req_object(objectName, type);
  req_object.loadKey(cat);

  auto grantee_object = grnt->findDbObject(req_object.getObjectKey(), false);
  if (grantee_object) {
    // if grantee has privs on the object
    return permissionFuncMap_[func_name](grantee_object->getPrivileges(), permissions);
  } else {
    // no privileges on that object
    return false;
  }
}

void DBHandler::get_db_objects_for_grantee(std::vector<TDBObject>& TDBObjectsForRole,
                                           const TSessionId& session_id_or_json,
                                           const std::string& roleName) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session_ptr);
  auto const& user = session_ptr->get_currentUser();
  if (!user.isSuper &&
      !SysCatalog::instance().isRoleGrantedToGrantee(user.userName, roleName, false)) {
    return;
  }
  auto* rl = SysCatalog::instance().getGrantee(roleName);
  if (rl) {
    auto dbId = session_ptr->getCatalog().getCurrentDB().dbId;
    for (auto& dbObject : *rl->getDbObjects(true)) {
      if (dbObject.first.dbId != dbId) {
        // TODO (max): it doesn't scale well in case we have many DBs (not a typical
        // usecase for now, though)
        continue;
      }
      TDBObject tdbObject = serialize_db_object(roleName, *dbObject.second);
      TDBObjectsForRole.push_back(tdbObject);
    }
  } else {
    THROW_DB_EXCEPTION("User or role " + roleName + " does not exist.");
  }
}

void DBHandler::get_db_object_privs(std::vector<TDBObject>& TDBObjects,
                                    const TSessionId& session_id_or_json,
                                    const std::string& objectName,
                                    const TDBObjectType::type type) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session_ptr);
  const auto& cat = session_ptr->getCatalog();
  DBObjectType object_type;
  switch (type) {
    case TDBObjectType::DatabaseDBObjectType:
      object_type = DBObjectType::DatabaseDBObjectType;
      break;
    case TDBObjectType::TableDBObjectType:
      object_type = DBObjectType::TableDBObjectType;
      break;
    case TDBObjectType::DashboardDBObjectType:
      object_type = DBObjectType::DashboardDBObjectType;
      break;
    case TDBObjectType::ViewDBObjectType:
      object_type = DBObjectType::ViewDBObjectType;
      break;
    case TDBObjectType::ServerDBObjectType:
      object_type = DBObjectType::ServerDBObjectType;
      break;
    default:
      THROW_DB_EXCEPTION("Failed to get object privileges for " + objectName +
                         ": unknown object type (" + std::to_string(type) + ").");
  }
  DBObject object_to_find(objectName, object_type);

  // TODO(adb): Use DatabaseLock to protect method
  try {
    if (object_type == DashboardDBObjectType) {
      if (objectName == "") {
        object_to_find = DBObject(-1, object_type);
      } else {
        object_to_find = DBObject(std::stoi(objectName), object_type);
      }
    } else if ((object_type == TableDBObjectType || object_type == ViewDBObjectType) &&
               !objectName.empty()) {
      // special handling for view / table
      auto td = cat.getMetadataForTable(objectName, false);
      if (td) {
        object_type = td->isView ? ViewDBObjectType : TableDBObjectType;
        object_to_find = DBObject(objectName, object_type);
      }
    }
    object_to_find.loadKey(cat);
  } catch (const std::exception&) {
    THROW_DB_EXCEPTION("Object with name " + objectName + " does not exist.");
  }

  // object type on database level
  DBObject object_to_find_dblevel("", object_type);
  object_to_find_dblevel.loadKey(cat);
  // if user is superuser respond with a full priv
  if (session_ptr->get_currentUser().isSuper) {
    // using ALL_TABLE here to set max permissions
    DBObject dbObj{object_to_find.getObjectKey(),
                   AccessPrivileges::ALL_TABLE,
                   session_ptr->get_currentUser().userId};
    dbObj.setName("super");
    TDBObjects.push_back(
        serialize_db_object(session_ptr->get_currentUser().userName, dbObj));
  };

  std::vector<std::string> grantees =
      SysCatalog::instance().getRoles(true,
                                      session_ptr->get_currentUser().isSuper,
                                      session_ptr->get_currentUser().userName);
  for (const auto& grantee : grantees) {
    DBObject* object_found;
    auto* gr = SysCatalog::instance().getGrantee(grantee);
    if (gr && (object_found = gr->findDbObject(object_to_find.getObjectKey(), true))) {
      TDBObjects.push_back(serialize_db_object(grantee, *object_found));
    }
    // check object permissions on Database level
    if (gr &&
        (object_found = gr->findDbObject(object_to_find_dblevel.getObjectKey(), true))) {
      TDBObjects.push_back(serialize_db_object(grantee, *object_found));
    }
  }
}

void DBHandler::getAllRolesForUserImpl(
    std::shared_ptr<Catalog_Namespace::SessionInfo const> session_ptr,
    std::vector<std::string>& roles,
    const std::string& granteeName,
    bool effective) {
  auto* grantee = SysCatalog::instance().getGrantee(granteeName);
  if (grantee) {
    if (session_ptr->get_currentUser().isSuper) {
      roles = grantee->getRoles(/*only_direct=*/!effective);
    } else if (grantee->isUser()) {
      if (session_ptr->get_currentUser().userName == granteeName) {
        roles = grantee->getRoles(/*only_direct=*/!effective);
      } else {
        THROW_DB_EXCEPTION(
            "Only a superuser is authorized to request list of roles granted to another "
            "user.");
      }
    } else {
      CHECK(!grantee->isUser());
      // granteeName is actually a roleName here and we can check a role
      // only if it is granted to us
      if (SysCatalog::instance().isRoleGrantedToGrantee(
              session_ptr->get_currentUser().userName, granteeName, false)) {
        roles = grantee->getRoles(/*only_direct=*/!effective);
      } else {
        THROW_DB_EXCEPTION("A user can check only roles granted to him.");
      }
    }
  } else {
    THROW_DB_EXCEPTION("Grantee " + granteeName + " does not exist.");
  }
}

void DBHandler::get_all_roles_for_user(std::vector<std::string>& roles,
                                       const TSessionId& session_id_or_json,
                                       const std::string& granteeName) {
  // WARNING: This function only returns directly granted roles.
  // See also: get_all_effective_roles_for_user() for all of a user's roles.
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  auto session_ptr = stdlog.getConstSessionInfo();
  getAllRolesForUserImpl(session_ptr, roles, granteeName, /*effective=*/false);
}

void DBHandler::get_all_effective_roles_for_user(std::vector<std::string>& roles,
                                                 const TSessionId& session_id_or_json,
                                                 const std::string& granteeName) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  auto session_ptr = stdlog.getConstSessionInfo();
  getAllRolesForUserImpl(session_ptr, roles, granteeName, /*effective=*/true);
}

namespace {
std::string dump_table_col_names(
    const std::map<std::string, std::vector<std::string>>& table_col_names) {
  std::ostringstream oss;
  for (const auto& [table_name, col_names] : table_col_names) {
    oss << ":" << table_name;
    for (const auto& col_name : col_names) {
      oss << "," << col_name;
    }
  }
  return oss.str();
}
}  // namespace

void DBHandler::get_result_row_for_pixel(
    TPixelTableRowResult& _return,
    const TSessionId& session_id_or_json,
    const int64_t widget_id,
    const TPixel& pixel,
    const std::map<std::string, std::vector<std::string>>& table_col_names,
    const bool column_format,
    const int32_t pixel_radius,
    const std::string& nonce) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session_ptr,
                       "widget_id",
                       widget_id,
                       "pixel.x",
                       pixel.x,
                       "pixel.y",
                       pixel.y,
                       "column_format",
                       column_format,
                       "pixel_radius",
                       pixel_radius,
                       "table_col_names",
                       dump_table_col_names(table_col_names),
                       "nonce",
                       nonce);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  if (!render_handler_) {
    THROW_DB_EXCEPTION("Backend rendering is disabled.");
  }

  try {
    render_handler_->get_result_row_for_pixel(_return,
                                              session_ptr,
                                              widget_id,
                                              pixel,
                                              table_col_names,
                                              column_format,
                                              pixel_radius,
                                              nonce);
  } catch (std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

TColumnType DBHandler::populateThriftColumnType(const Catalog* cat,
                                                const ColumnDescriptor* cd) {
  TColumnType col_type;
  col_type.col_name = cd->columnName;
  col_type.src_name = cd->sourceName;
  col_type.col_id = cd->columnId;
  col_type.col_type.type = type_to_thrift(cd->columnType);
  col_type.col_type.encoding = encoding_to_thrift(cd->columnType);
  col_type.col_type.nullable = !cd->columnType.get_notnull();
  col_type.col_type.is_array = cd->columnType.get_type() == kARRAY;
  if (col_type.col_type.is_array || cd->columnType.get_type() == kDATE) {
    col_type.col_type.size = cd->columnType.get_size();  // only for arrays and dates
  }
  if (IS_GEO(cd->columnType.get_type())) {
    ThriftSerializers::fixup_geo_column_descriptor(
        col_type, cd->columnType.get_subtype(), cd->columnType.get_output_srid());
  } else {
    col_type.col_type.precision = cd->columnType.get_precision();
    col_type.col_type.scale = cd->columnType.get_scale();
  }
  col_type.is_system = cd->isSystemCol;
  if (cd->columnType.get_compression() == EncodingType::kENCODING_DICT &&
      cat != nullptr) {
    // have to get the actual size of the encoding from the dictionary definition
    const int dict_id = cd->columnType.get_comp_param();
    if (!cat->getMetadataForDict(dict_id, false)) {
      col_type.col_type.comp_param = 0;
      return col_type;
    }
    auto dd = cat->getMetadataForDict(dict_id, false);
    if (!dd) {
      THROW_DB_EXCEPTION("Dictionary doesn't exist");
    }
    col_type.col_type.comp_param = dd->dictNBits;
  } else {
    col_type.col_type.comp_param =
        (cd->columnType.is_date_in_days() && cd->columnType.get_comp_param() == 0)
            ? 32
            : cd->columnType.get_comp_param();
  }
  col_type.is_reserved_keyword = ImportHelpers::is_reserved_name(col_type.col_name);
  if (cd->default_value.has_value()) {
    col_type.__set_default_value(cd->getDefaultValueLiteral());
  }
  return col_type;
}

void DBHandler::get_internal_table_details(TTableDetails& _return,
                                           const TSessionId& session_id_or_json,
                                           const std::string& table_name,
                                           const bool include_system_columns) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog =
      STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  get_table_details_impl(_return, stdlog, table_name, include_system_columns, false);
}

void DBHandler::get_internal_table_details_for_database(
    TTableDetails& _return,
    const TSessionId& session_id_or_json,
    const std::string& table_name,
    const std::string& database_name) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog =
      STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  get_table_details_impl(_return, stdlog, table_name, true, false, database_name);
}

void DBHandler::get_table_details(TTableDetails& _return,
                                  const TSessionId& session_id_or_json,
                                  const std::string& table_name) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog =
      STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());

  auto execute_read_lock = legacylockmgr::getExecuteReadLock();
  get_table_details_impl(_return, stdlog, table_name, false, false);
}

void DBHandler::get_table_details_for_database(TTableDetails& _return,
                                               const TSessionId& session_id_or_json,
                                               const std::string& table_name,
                                               const std::string& database_name) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog =
      STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());

  auto execute_read_lock = legacylockmgr::getExecuteReadLock();
  get_table_details_impl(_return, stdlog, table_name, false, false, database_name);
}

namespace {
TTableRefreshInfo get_refresh_info(const TableDescriptor* td) {
  CHECK(td->isForeignTable());
  auto foreign_table = dynamic_cast<const foreign_storage::ForeignTable*>(td);
  CHECK(foreign_table);
  TTableRefreshInfo refresh_info;
  const auto& update_type =
      foreign_table->getOption(foreign_storage::ForeignTable::REFRESH_UPDATE_TYPE_KEY);
  CHECK(update_type.has_value());
  if (update_type.value() == foreign_storage::ForeignTable::ALL_REFRESH_UPDATE_TYPE) {
    refresh_info.update_type = TTableRefreshUpdateType::ALL;
  } else if (update_type.value() ==
             foreign_storage::ForeignTable::APPEND_REFRESH_UPDATE_TYPE) {
    refresh_info.update_type = TTableRefreshUpdateType::APPEND;
  } else {
    UNREACHABLE() << "Unexpected refresh update type: " << update_type.value();
  }

  const auto& timing_type =
      foreign_table->getOption(foreign_storage::ForeignTable::REFRESH_TIMING_TYPE_KEY);
  CHECK(timing_type.has_value());
  if (timing_type.value() == foreign_storage::ForeignTable::MANUAL_REFRESH_TIMING_TYPE) {
    refresh_info.timing_type = TTableRefreshTimingType::MANUAL;
    refresh_info.interval_count = -1;
  } else if (timing_type.value() ==
             foreign_storage::ForeignTable::SCHEDULE_REFRESH_TIMING_TYPE) {
    refresh_info.timing_type = TTableRefreshTimingType::SCHEDULED;
    const auto& start_date_time = foreign_table->getOption(
        foreign_storage::ForeignTable::REFRESH_START_DATE_TIME_KEY);
    CHECK(start_date_time.has_value());
    auto start_date_time_epoch = dateTimeParse<kTIMESTAMP>(start_date_time.value(), 0);
    refresh_info.start_date_time =
        shared::convert_temporal_to_iso_format({kTIMESTAMP}, start_date_time_epoch);
    const auto& interval =
        foreign_table->getOption(foreign_storage::ForeignTable::REFRESH_INTERVAL_KEY);
    CHECK(interval.has_value());
    const auto& interval_str = interval.value();
    refresh_info.interval_count =
        std::stoi(interval_str.substr(0, interval_str.length() - 1));
    auto interval_type = std::toupper(interval_str[interval_str.length() - 1]);
    if (interval_type == 'H') {
      refresh_info.interval_type = TTableRefreshIntervalType::HOUR;
    } else if (interval_type == 'D') {
      refresh_info.interval_type = TTableRefreshIntervalType::DAY;
    } else if (interval_type == 'S') {
      // This use case is for development only.
      refresh_info.interval_type = TTableRefreshIntervalType::NONE;
    } else {
      UNREACHABLE() << "Unexpected interval type: " << interval_str;
    }
  } else {
    UNREACHABLE() << "Unexpected refresh timing type: " << timing_type.value();
  }
  if (foreign_table->last_refresh_time !=
      foreign_storage::ForeignTable::NULL_REFRESH_TIME) {
    refresh_info.last_refresh_time = shared::convert_temporal_to_iso_format(
        {kTIMESTAMP}, foreign_table->last_refresh_time);
  }
  if (foreign_table->next_refresh_time !=
      foreign_storage::ForeignTable::NULL_REFRESH_TIME) {
    refresh_info.next_refresh_time = shared::convert_temporal_to_iso_format(
        {kTIMESTAMP}, foreign_table->next_refresh_time);
  }
  return refresh_info;
}
}  // namespace

void DBHandler::get_table_details_impl(TTableDetails& _return,
                                       query_state::StdLog& stdlog,
                                       const std::string& table_name,
                                       const bool get_system,
                                       const bool get_physical,
                                       const std::string& database_name) {
  try {
    auto session_info = stdlog.getSessionInfo();
    auto cat = (database_name.empty())
                   ? &session_info->getCatalog()
                   : SysCatalog::instance().getCatalog(database_name).get();
    if (!cat) {
      THROW_DB_EXCEPTION("Database " + database_name + " does not exist.");
    }
    const auto td_with_lock =
        lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>::acquireTableDescriptor(
            *cat, table_name, false);
    const auto td = td_with_lock();
    CHECK(td);

    bool have_privileges_on_view_sources = true;
    if (td->isView) {
      auto query_state = create_query_state(session_info, td->viewSQL);
      stdlog.setQueryState(query_state);
      try {
        if (hasTableAccessPrivileges(td, *session_info)) {
          const auto [query_ra, locks] = parse_to_ra(query_state->createQueryStateProxy(),
                                                     query_state->getQueryStr(),
                                                     {},
                                                     true,
                                                     system_parameters_,
                                                     false);
          try {
            calcite_->checkAccessedObjectsPrivileges(query_state->createQueryStateProxy(),
                                                     query_ra);
          } catch (const std::runtime_error&) {
            have_privileges_on_view_sources = false;
          }

          _return.row_desc =
              validateRelAlg(query_ra.plan_result, query_state->createQueryStateProxy());
        } else {
          throw std::runtime_error(
              "Unable to access view " + table_name +
              ". The view may not exist, or the logged in user may not "
              "have permission to access the view.");
        }
      } catch (const std::exception& e) {
        throw std::runtime_error("View '" + table_name +
                                 "' query has failed with an error: '" +
                                 std::string(e.what()) +
                                 "'.\nThe view must be dropped and re-created to "
                                 "resolve the error. \nQuery:\n" +
                                 query_state->getQueryStr());
      }
    } else {
      if (hasTableAccessPrivileges(td, *session_info)) {
        const auto col_descriptors = cat->getAllColumnMetadataForTable(
            td->tableId, get_system, true, get_physical);
        const auto deleted_cd = cat->getDeletedColumn(td);
        for (const auto cd : col_descriptors) {
          if (cd == deleted_cd) {
            continue;
          }
          _return.row_desc.push_back(populateThriftColumnType(cat, cd));
        }
      } else {
        throw std::runtime_error(
            "Unable to access table " + table_name +
            ". The table may not exist, or the logged in user may not "
            "have permission to access the table.");
      }
    }
    _return.fragment_size = td->maxFragRows;
    _return.page_size = td->fragPageSize;
    _return.max_rows = td->maxRows;
    _return.view_sql =
        (have_privileges_on_view_sources ? td->viewSQL
                                         : "[Not enough privileges to see the view SQL]");
    _return.shard_count = td->nShards * std::max(g_leaf_count, size_t(1));
    if (td->nShards > 0) {
      auto cd = cat->getMetadataForColumn(td->tableId, td->shardedColumnId);
      CHECK(cd);
      _return.sharded_column_name = cd->columnName;
    }
    _return.key_metainfo = td->keyMetainfo;
    _return.is_temporary = td->persistenceLevel == Data_Namespace::MemoryLevel::CPU_LEVEL;
    _return.partition_detail =
        td->partitions.empty()
            ? TPartitionDetail::DEFAULT
            : (table_is_replicated(td)
                   ? TPartitionDetail::REPLICATED
                   : (td->partitions == "SHARDED" ? TPartitionDetail::SHARDED
                                                  : TPartitionDetail::OTHER));
    if (td->isView) {
      _return.table_type = TTableType::VIEW;
    } else if (td->isTemporaryTable()) {
      _return.table_type = TTableType::TEMPORARY;
    } else if (td->isForeignTable()) {
      _return.table_type = TTableType::FOREIGN;
      _return.refresh_info = get_refresh_info(td);
    } else {
      _return.table_type = TTableType::DEFAULT;
    }

  } catch (const std::runtime_error& e) {
    THROW_DB_EXCEPTION(std::string(e.what()));
  }
}

void DBHandler::get_link_view(TFrontendView& _return,
                              const TSessionId& session_id_or_json,
                              const std::string& link) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session_ptr);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto const& cat = session_ptr->getCatalog();
  auto ld = cat.getMetadataForLink(std::to_string(cat.getCurrentDB().dbId) + link);
  if (!ld) {
    THROW_DB_EXCEPTION("Link " + link + " is not valid.");
  }
  _return.view_state = ld->viewState;
  _return.view_name = ld->link;
  _return.update_time = ld->updateTime;
  _return.view_metadata = ld->viewMetadata;
}

bool DBHandler::hasTableAccessPrivileges(
    const TableDescriptor* td,
    const Catalog_Namespace::SessionInfo& session_info) {
  auto& cat = session_info.getCatalog();
  auto user_metadata = session_info.get_currentUser();

  if (user_metadata.isSuper) {
    return true;
  }

  DBObject dbObject(td->tableName, td->isView ? ViewDBObjectType : TableDBObjectType);
  dbObject.loadKey(cat);
  std::vector<DBObject> privObjects = {dbObject};

  return SysCatalog::instance().hasAnyPrivileges(user_metadata, privObjects);
}

void DBHandler::get_tables_impl(std::vector<std::string>& table_names,
                                const Catalog_Namespace::SessionInfo& session_info,
                                const GetTablesType get_tables_type,
                                const std::string& database_name) {
  if (database_name.empty()) {
    table_names = session_info.getCatalog().getTableNamesForUser(
        session_info.get_currentUser(), get_tables_type);
  } else {
    auto request_cat = SysCatalog::instance().getCatalog(database_name);
    if (!request_cat) {
      THROW_DB_EXCEPTION("Database " + database_name + " does not exist.");
    }
    table_names = request_cat->getTableNamesForUser(session_info.get_currentUser(),
                                                    get_tables_type);
  }
}

void DBHandler::get_tables(std::vector<std::string>& table_names,
                           const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  get_tables_impl(
      table_names, *stdlog.getConstSessionInfo(), GET_PHYSICAL_TABLES_AND_VIEWS);
}

void DBHandler::get_tables_for_database(std::vector<std::string>& table_names,
                                        const TSessionId& session_id_or_json,
                                        const std::string& database_name) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());

  get_tables_impl(table_names,
                  *stdlog.getConstSessionInfo(),
                  GET_PHYSICAL_TABLES_AND_VIEWS,
                  database_name);
}

void DBHandler::get_physical_tables(std::vector<std::string>& table_names,
                                    const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  get_tables_impl(table_names, *stdlog.getConstSessionInfo(), GET_PHYSICAL_TABLES);
}

void DBHandler::get_views(std::vector<std::string>& table_names,
                          const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  get_tables_impl(table_names, *stdlog.getConstSessionInfo(), GET_VIEWS);
}

void DBHandler::get_tables_meta_impl(std::vector<TTableMeta>& _return,
                                     QueryStateProxy query_state_proxy,
                                     const Catalog_Namespace::SessionInfo& session_info,
                                     const bool with_table_locks) {
  const auto& cat = session_info.getCatalog();
  // Get copies of table descriptors here in order to avoid possible use of dangling
  // pointers, if tables are concurrently dropped.
  const auto tables = cat.getAllTableMetadataCopy();
  _return.reserve(tables.size());

  for (const auto& td : tables) {
    if (td.shard >= 0) {
      // skip shards, they're not standalone tables
      continue;
    }
    if (!hasTableAccessPrivileges(&td, session_info)) {
      // skip table, as there are no privileges to access it
      continue;
    }

    TTableMeta ret;
    ret.table_name = td.tableName;
    ret.is_view = td.isView;
    ret.is_replicated = table_is_replicated(&td);
    ret.shard_count = td.nShards;
    ret.max_rows = td.maxRows;
    ret.table_id = td.tableId;

    std::vector<TTypeInfo> col_types;
    std::vector<std::string> col_names;
    size_t num_cols = 0;
    if (td.isView) {
      try {
        TPlanResult parse_result;
        lockmgr::LockedTableDescriptors locks;
        std::tie(parse_result, locks) = parse_to_ra(
            query_state_proxy, td.viewSQL, {}, with_table_locks, system_parameters_);
        const auto query_ra = parse_result.plan_result;

        ExecutionResult ex_result;
        execute_rel_alg(ex_result,
                        query_state_proxy,
                        query_ra,
                        true,
                        ExecutorDeviceType::CPU,
                        -1,
                        -1,
                        /*just_validate=*/true,
                        /*find_push_down_candidates=*/false,
                        ExplainInfo());
        TQueryResult result;
        DBHandler::convertData(result, ex_result, query_state_proxy, true, -1, -1);
        num_cols = result.row_set.row_desc.size();
        for (const auto& col : result.row_set.row_desc) {
          if (col.is_physical) {
            num_cols--;
            continue;
          }
          col_types.push_back(col.col_type);
          col_names.push_back(col.col_name);
        }
      } catch (std::exception& e) {
        LOG(WARNING) << "get_tables_meta: Ignoring broken view: " << td.tableName;
      }
    } else {
      try {
        if (hasTableAccessPrivileges(&td, session_info)) {
          const auto col_descriptors =
              cat.getAllColumnMetadataForTable(td.tableId, false, true, false);
          const auto deleted_cd = cat.getDeletedColumn(&td);
          for (const auto cd : col_descriptors) {
            if (cd == deleted_cd) {
              continue;
            }
            col_types.push_back(ThriftSerializers::type_info_to_thrift(cd->columnType));
            col_names.push_back(cd->columnName);
          }
          num_cols = col_descriptors.size();
        } else {
          continue;
        }
      } catch (const std::runtime_error& e) {
        THROW_DB_EXCEPTION(e.what());
      }
    }

    ret.num_cols = num_cols;
    std::copy(col_types.begin(), col_types.end(), std::back_inserter(ret.col_types));
    std::copy(col_names.begin(), col_names.end(), std::back_inserter(ret.col_names));

    _return.push_back(ret);
  }
}

void DBHandler::get_tables_meta(std::vector<TTableMeta>& _return,
                                const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  auto query_state = create_query_state(session_ptr, "");
  stdlog.setQueryState(query_state);

  auto execute_read_lock = legacylockmgr::getExecuteReadLock();

  try {
    get_tables_meta_impl(_return, query_state->createQueryStateProxy(), *session_ptr);
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::get_users(std::vector<std::string>& user_names,
                          const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  std::list<Catalog_Namespace::UserMetadata> user_list;

  if (!session_ptr->get_currentUser().isSuper) {
    user_list = SysCatalog::instance().getAllUserMetadata(
        session_ptr->getCatalog().getCurrentDB().dbId);
  } else {
    user_list = SysCatalog::instance().getAllUserMetadata();
  }
  for (auto u : user_list) {
    user_names.push_back(u.userName);
  }
}

void DBHandler::get_version(std::string& version) {
  version = MAPD_RELEASE;
}

namespace {

ScopeGuard pause_and_resume_executor_queue() {
  if (g_enable_executor_resource_mgr) {
    Executor::pause_executor_queue();
    return [] {
      // we need to resume erm queue if we throw any exception
      // that heavydb server can handle w/o shutting it down
      Executor::resume_executor_queue();
    };
  }
  return [] {};
}

}  // namespace

void DBHandler::clear_gpu_memory(const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION("Superuser privilege is required to run clear_gpu_memory");
  }
  auto resume_executor_queue = pause_and_resume_executor_queue();
  // clear renderer memory first
  // this will block until any running render finishes
  if (render_handler_) {
    render_handler_->clear_gpu_memory();
  }
  // then clear the QE memory
  // the renderer will have disconnected from any QE memory
  try {
    Executor::clearMemory(Data_Namespace::MemoryLevel::GPU_LEVEL);
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::clear_cpu_memory(const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION("Superuser privilege is required to run clear_cpu_memory");
  }
  auto resume_executor_queue = pause_and_resume_executor_queue();
  // clear renderer memory first
  // this will block until any running render finishes
  if (render_handler_) {
    render_handler_->clear_cpu_memory();
  }
  // then clear the QE memory
  // the renderer will have disconnected from any QE memory
  try {
    Executor::clearMemory(Data_Namespace::MemoryLevel::CPU_LEVEL);
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::clearRenderMemory(const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION("Superuser privilege is required to run clear_render_memory");
  }
  if (render_handler_) {
    auto resume_executor_queue = pause_and_resume_executor_queue();
    render_handler_->clear_cpu_memory();
    render_handler_->clear_gpu_memory();
  }
}

void DBHandler::pause_executor_queue(const TSessionId& session) {
  auto stdlog = STDLOG(get_session_ptr(session));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION("Superuser privilege is required to run PAUSE EXECUTOR QUEUE");
  }
  try {
    Executor::pause_executor_queue();
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::resume_executor_queue(const TSessionId& session) {
  auto stdlog = STDLOG(get_session_ptr(session));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION("Superuser privilege is required to run RESUME EXECUTOR QUEUE");
  }
  try {
    Executor::resume_executor_queue();
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::set_cur_session(const TSessionId& parent_session_id_or_json,
                                const TSessionId& leaf_session_id_or_json,
                                const std::string& start_time_str,
                                const std::string& label,
                                bool for_running_query_kernel) {
  // internal API to manage query interruption in distributed mode
  heavyai::RequestInfo const parent_request_info(parent_session_id_or_json);
  heavyai::RequestInfo const leaf_request_info(leaf_session_id_or_json);
  SET_REQUEST_ID(leaf_request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(leaf_request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();

  auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID).get();
  executor->enrollQuerySession(parent_request_info.sessionId(),
                               label,
                               start_time_str,
                               Executor::UNITARY_EXECUTOR_ID,
                               for_running_query_kernel
                                   ? QuerySessionStatus::QueryStatus::RUNNING_QUERY_KERNEL
                                   : QuerySessionStatus::QueryStatus::RUNNING_IMPORTER);
}

void DBHandler::invalidate_cur_session(const TSessionId& parent_session_id_or_json,
                                       const TSessionId& leaf_session_id_or_json,
                                       const std::string& start_time_str,
                                       const std::string& label,
                                       bool for_running_query_kernel) {
  // internal API to manage query interruption in distributed mode
  heavyai::RequestInfo const parent_request_info(parent_session_id_or_json);
  heavyai::RequestInfo const leaf_request_info(leaf_session_id_or_json);
  SET_REQUEST_ID(leaf_request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(leaf_request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID).get();
  executor->clearQuerySessionStatus(parent_request_info.sessionId(), start_time_str);
}

TSessionId DBHandler::getInvalidSessionId() const {
  return INVALID_SESSION_ID;
}

void DBHandler::get_memory(std::vector<TNodeMemoryInfo>& _return,
                           const TSessionId& session_id_or_json,
                           const std::string& memory_level) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  std::vector<Data_Namespace::MemoryInfo> internal_memory;
  if (!memory_level.compare("gpu")) {
    internal_memory =
        SysCatalog::instance().getDataMgr().getMemoryInfo(MemoryLevel::GPU_LEVEL);
  } else {
    internal_memory =
        SysCatalog::instance().getDataMgr().getMemoryInfo(MemoryLevel::CPU_LEVEL);
  }

  for (auto memInfo : internal_memory) {
    TNodeMemoryInfo nodeInfo;
    nodeInfo.page_size = memInfo.pageSize;
    nodeInfo.max_num_pages = memInfo.maxNumPages;
    nodeInfo.num_pages_allocated = memInfo.numPageAllocated;
    nodeInfo.is_allocation_capped = memInfo.isAllocationCapped;
    for (auto gpu : memInfo.nodeMemoryData) {
      TMemoryData md;
      md.slab = gpu.slabNum;
      md.start_page = gpu.startPage;
      md.num_pages = gpu.numPages;
      md.touch = gpu.touch;
      md.chunk_key.insert(md.chunk_key.end(), gpu.chunk_key.begin(), gpu.chunk_key.end());
      md.is_free = gpu.memStatus == Buffer_Namespace::MemStatus::FREE;
      nodeInfo.node_memory_data.push_back(md);
    }
    _return.push_back(nodeInfo);
  }
}

void DBHandler::get_databases(std::vector<TDBInfo>& dbinfos,
                              const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  const auto& user = session_ptr->get_currentUser();
  Catalog_Namespace::DBSummaryList dbs =
      SysCatalog::instance().getDatabaseListForUser(user);
  for (auto& db : dbs) {
    TDBInfo dbinfo;
    dbinfo.db_name = std::move(db.dbName);
    dbinfo.db_owner = std::move(db.dbOwnerName);
    dbinfos.push_back(std::move(dbinfo));
  }
}

TExecuteMode::type DBHandler::getExecutionMode(const TSessionId& session_id) {
  auto executor = get_session_ptr(session_id)->get_executor_device_type();
  switch (executor) {
    case ExecutorDeviceType::CPU:
      return TExecuteMode::CPU;
    case ExecutorDeviceType::GPU:
      return TExecuteMode::GPU;
    default:
      UNREACHABLE();
  }
  UNREACHABLE();
  return TExecuteMode::CPU;
}
void DBHandler::set_execution_mode(const TSessionId& session_id_or_json,
                                   const TExecuteMode::type mode) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto session_ptr = get_session_ptr(request_info.sessionId());
  auto stdlog = STDLOG(session_ptr);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  DBHandler::set_execution_mode_nolock(session_ptr.get(), mode);
}

namespace {

void check_table_not_sharded(const TableDescriptor* td) {
  if (td && td->nShards) {
    throw std::runtime_error("Cannot import a sharded table directly to a leaf");
  }
}

void check_valid_column_names(const std::list<const ColumnDescriptor*>& descs,
                              const std::vector<std::string>& column_names) {
  std::unordered_set<std::string> unique_names;
  for (const auto& name : column_names) {
    auto lower_name = to_lower(name);
    if (unique_names.find(lower_name) != unique_names.end()) {
      THROW_DB_EXCEPTION("Column " + name + " is mentioned multiple times");
    } else {
      unique_names.insert(lower_name);
    }
  }
  for (const auto& cd : descs) {
    auto iter = unique_names.find(to_lower(cd->columnName));
    if (iter != unique_names.end()) {
      unique_names.erase(iter);
    }
  }
  if (!unique_names.empty()) {
    THROW_DB_EXCEPTION("Column " + *unique_names.begin() + " does not exist");
  }
}

// Return vector of IDs mapping column descriptors to the list of comumn names.
// The size of the vector is the number of actual columns (geophisical columns excluded).
// ID is either a position in column_names matching the descriptor, or -1 if the column
// is missing from the column_names
std::vector<int> column_ids_by_names(const std::list<const ColumnDescriptor*>& descs,
                                     const std::vector<std::string>& column_names) {
  std::vector<int> desc_to_column_ids;
  if (column_names.empty()) {
    int col_idx = 0;
    for (const auto& cd : descs) {
      if (!cd->isGeoPhyCol) {
        desc_to_column_ids.push_back(col_idx);
        ++col_idx;
      }
    }
  } else {
    for (const auto& cd : descs) {
      if (!cd->isGeoPhyCol) {
        bool found = false;
        for (size_t j = 0; j < column_names.size(); ++j) {
          if (to_lower(cd->columnName) == to_lower(column_names[j])) {
            found = true;
            desc_to_column_ids.push_back(j);
            break;
          }
        }
        if (!found) {
          if (!cd->columnType.get_notnull()) {
            desc_to_column_ids.push_back(-1);
          } else {
            THROW_DB_EXCEPTION("Column '" + cd->columnName +
                               "' cannot be omitted due to NOT NULL constraint");
          }
        }
      }
    }
  }
  return desc_to_column_ids;
}

void log_cache_size(const Catalog_Namespace::Catalog& cat) {
  std::ostringstream oss;
  oss << "Cache size information {";
  auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID);
  // 1. Data recycler
  // 1.a Resultset Recycler
  auto resultset_cache_size =
      executor->getResultSetRecyclerHolder()
          .getResultSetRecycler()
          ->getResultSetRecyclerMetricTracker()
          .getCurrentCacheSize(DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
  if (resultset_cache_size) {
    oss << "\"query_resultset\": " << *resultset_cache_size << " bytes, ";
  }

  // 1.b Join Hash Table Recycler
  auto perfect_join_ht_cache_size =
      PerfectJoinHashTable::getHashTableCache()->getCurrentCacheSizeForDevice(
          CacheItemType::PERFECT_HT, DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
  auto baseline_join_ht_cache_size =
      BaselineJoinHashTable::getHashTableCache()->getCurrentCacheSizeForDevice(
          CacheItemType::BASELINE_HT, DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
  auto bbox_intersect_ht_cache_size =
      BoundingBoxIntersectJoinHashTable::getHashTableCache()
          ->getCurrentCacheSizeForDevice(CacheItemType::BBOX_INTERSECT_HT,
                                         DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
  auto bbox_intersect_ht_tuner_cache_size =
      BoundingBoxIntersectJoinHashTable::getBoundingBoxIntersectTuningParamCache()
          ->getCurrentCacheSizeForDevice(CacheItemType::BBOX_INTERSECT_AUTO_TUNER_PARAM,
                                         DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
  auto sum_hash_table_cache_size =
      perfect_join_ht_cache_size + baseline_join_ht_cache_size +
      bbox_intersect_ht_cache_size + bbox_intersect_ht_tuner_cache_size;
  oss << "\"hash_tables\": " << sum_hash_table_cache_size << " bytes, ";

  // 1.c Chunk Metadata Recycler
  auto chunk_metadata_cache_size =
      executor->getResultSetRecyclerHolder()
          .getChunkMetadataRecycler()
          ->getCurrentCacheSizeForDevice(CacheItemType::CHUNK_METADATA,
                                         DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
  oss << "\"chunk_metadata\": " << chunk_metadata_cache_size << " bytes, ";

  // 2. Query Plan Dag
  auto query_plan_dag_cache_size =
      executor->getQueryPlanDagCache().getCurrentNodeMapSize();
  oss << "\"query_plan_dag\": " << query_plan_dag_cache_size << " bytes, ";

  // 3. Compiled (GPU) Code
  oss << "\"compiled_GPU code\": "
      << QueryEngine::getInstance()->gpu_code_accessor->getCacheSize() << " bytes, ";

  // 4. String Dictionary
  oss << "\"string_dictionary\": " << cat.getTotalMemorySizeForDictionariesForDatabase()
      << " bytes";
  oss << "}";
  LOG(INFO) << oss.str();
}

void log_system_cpu_memory_status(std::string const& query,
                                  const Catalog_Namespace::Catalog& cat) {
  if (g_allow_memory_status_log) {
    std::ostringstream oss;
    oss << query << "\n" << cat.getDataMgr().getSystemMemoryUsage();
    LOG(INFO) << oss.str();
    log_cache_size(cat);
  }
}
}  // namespace

void DBHandler::fillGeoColumns(
    const TSessionId& session_id,
    const Catalog& catalog,
    std::vector<std::unique_ptr<import_export::TypedImportBuffer>>& import_buffers,
    const ColumnDescriptor* cd,
    size_t& col_idx,
    size_t num_rows,
    const std::string& table_name) {
  auto geo_col_idx = col_idx - 1;
  const auto wkt_or_wkb_hex_column = import_buffers[geo_col_idx]->getGeoStringBuffer();
  std::vector<std::vector<double>> coords_column, bounds_column;
  std::vector<std::vector<int>> ring_sizes_column, poly_rings_column;
  SQLTypeInfo ti = cd->columnType;
  const bool validate_with_geos_if_available = false;
  if (num_rows != wkt_or_wkb_hex_column->size() ||
      !Geospatial::GeoTypesFactory::getGeoColumns(wkt_or_wkb_hex_column,
                                                  ti,
                                                  coords_column,
                                                  bounds_column,
                                                  ring_sizes_column,
                                                  poly_rings_column,
                                                  validate_with_geos_if_available)) {
    std::ostringstream oss;
    oss << "Invalid geometry in column " << cd->columnName;
    THROW_DB_EXCEPTION(oss.str());
  }

  // Populate physical columns, advance col_idx
  import_export::Importer::set_geo_physical_import_buffer_columnar(catalog,
                                                                   cd,
                                                                   import_buffers,
                                                                   col_idx,
                                                                   coords_column,
                                                                   bounds_column,
                                                                   ring_sizes_column,
                                                                   poly_rings_column);
}

void DBHandler::fillMissingBuffers(
    const TSessionId& session_id,
    const Catalog& catalog,
    std::vector<std::unique_ptr<import_export::TypedImportBuffer>>& import_buffers,
    const std::list<const ColumnDescriptor*>& cds,
    const std::vector<int>& desc_id_to_column_id,
    size_t num_rows,
    const std::string& table_name) {
  size_t skip_physical_cols = 0;
  size_t col_idx = 0, import_idx = 0;
  for (const auto& cd : cds) {
    if (skip_physical_cols > 0) {
      CHECK(cd->isGeoPhyCol);
      skip_physical_cols--;
      continue;
    } else if (cd->columnType.is_geometry()) {
      skip_physical_cols = cd->columnType.get_physical_cols();
    }
    if (desc_id_to_column_id[import_idx] == -1) {
      import_buffers[col_idx]->addDefaultValues(cd, num_rows);
      col_idx++;
      if (cd->columnType.is_geometry()) {
        fillGeoColumns(
            session_id, catalog, import_buffers, cd, col_idx, num_rows, table_name);
      }
    } else {
      col_idx++;
      col_idx += skip_physical_cols;
    }
    import_idx++;
  }
}

namespace {
std::string get_load_tag(const std::string& load_tag, const std::string& table_name) {
  std::ostringstream oss;
  oss << load_tag << "(" << table_name << ")";
  return oss.str();
}

std::string get_import_tag(const std::string& import_tag,
                           const std::string& table_name,
                           const std::string& file_path) {
  std::ostringstream oss;
  oss << import_tag << "(" << table_name << ", file_path:" << file_path << ")";
  return oss.str();
}
}  // namespace

void DBHandler::load_table_binary(const TSessionId& session_id_or_json,
                                  const std::string& table_name,
                                  const std::vector<TRow>& rows,
                                  const std::vector<std::string>& column_names) {
  try {
    heavyai::RequestInfo const request_info(session_id_or_json);
    SET_REQUEST_ID(request_info.requestId());
    auto stdlog =
        STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
    stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
    auto session_ptr = stdlog.getConstSessionInfo();

    if (rows.empty()) {
      THROW_DB_EXCEPTION("No rows to insert");
    }

    const auto execute_read_lock = legacylockmgr::getExecuteReadLock();
    std::unique_ptr<import_export::Loader> loader;
    std::vector<std::unique_ptr<import_export::TypedImportBuffer>> import_buffers;
    auto schema_read_lock = prepare_loader_generic(*session_ptr,
                                                   table_name,
                                                   rows.front().cols.size(),
                                                   &loader,
                                                   &import_buffers,
                                                   column_names,
                                                   "load_table_binary");

    auto col_descs = loader->get_column_descs();
    auto desc_id_to_column_id = column_ids_by_names(col_descs, column_names);

    size_t rows_completed = 0;
    auto const load_tag = get_load_tag("load_table_binary", table_name);
    log_system_cpu_memory_status("start_" + load_tag, session_ptr->getCatalog());
    ScopeGuard cleanup = [&load_tag, &session_ptr]() {
      log_system_cpu_memory_status("finish_" + load_tag, session_ptr->getCatalog());
    };
    for (auto const& row : rows) {
      size_t col_idx = 0;
      try {
        for (auto cd : col_descs) {
          auto mapped_idx = desc_id_to_column_id[col_idx];
          if (mapped_idx != -1) {
            import_buffers[col_idx]->add_value(
                cd, row.cols[mapped_idx], row.cols[mapped_idx].is_null);
          }
          col_idx++;
        }
        rows_completed++;
      } catch (const std::exception& e) {
        for (size_t col_idx_to_pop = 0; col_idx_to_pop < col_idx; ++col_idx_to_pop) {
          import_buffers[col_idx_to_pop]->pop_value();
        }
        LOG(ERROR) << "Input exception thrown: " << e.what()
                   << ". Row discarded, issue at column : " << (col_idx + 1)
                   << " data :" << row;
      }
    }
    fillMissingBuffers(request_info.sessionId(),
                       session_ptr->getCatalog(),
                       import_buffers,
                       col_descs,
                       desc_id_to_column_id,
                       rows_completed,
                       table_name);
    auto insert_data_lock = lockmgr::InsertDataLockMgr::getWriteLockForTable(
        session_ptr->getCatalog(), table_name);
    if (!loader->load(import_buffers, rows.size(), session_ptr.get())) {
      THROW_DB_EXCEPTION(loader->getErrorMessage());
    }
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(std::string(e.what()));
  }
}

std::unique_ptr<lockmgr::AbstractLockContainer<const TableDescriptor*>>
DBHandler::prepare_loader_generic(
    const Catalog_Namespace::SessionInfo& session_info,
    const std::string& table_name,
    size_t num_cols,
    std::unique_ptr<import_export::Loader>* loader,
    std::vector<std::unique_ptr<import_export::TypedImportBuffer>>* import_buffers,
    const std::vector<std::string>& column_names,
    std::string load_type) {
  if (num_cols == 0) {
    THROW_DB_EXCEPTION("No columns to insert");
  }
  check_read_only(load_type);
  auto& cat = session_info.getCatalog();
  auto td_with_lock =
      std::make_unique<lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>>(
          lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>::acquireTableDescriptor(
              cat, table_name, true));
  const auto td = (*td_with_lock)();
  CHECK(td);

  if (g_cluster && !leaf_aggregator_.leafCount()) {
    // Sharded table rows need to be routed to the leaf by an aggregator.
    check_table_not_sharded(td);
  }
  check_table_load_privileges(session_info, table_name);

  loader->reset(new import_export::Loader(cat, td));

  auto col_descs = (*loader)->get_column_descs();
  check_valid_column_names(col_descs, column_names);
  if (column_names.empty()) {
    // TODO(andrew): nColumns should be number of non-virtual/non-system columns.
    //               Subtracting 1 (rowid) until TableDescriptor is updated.
    auto geo_physical_cols = std::count_if(
        col_descs.begin(), col_descs.end(), [](auto cd) { return cd->isGeoPhyCol; });
    const auto num_table_cols = static_cast<size_t>(td->nColumns) - geo_physical_cols -
                                (td->hasDeletedCol ? 2 : 1);
    if (num_cols != num_table_cols) {
      throw std::runtime_error("Number of columns to load (" + std::to_string(num_cols) +
                               ") does not match number of columns in table " +
                               td->tableName + " (" + std::to_string(num_table_cols) +
                               ")");
    }
  } else if (num_cols != column_names.size()) {
    THROW_DB_EXCEPTION(
        "Number of columns specified does not match the "
        "number of columns given (" +
        std::to_string(num_cols) + " vs " + std::to_string(column_names.size()) + ")");
  }

  *import_buffers = import_export::setup_column_loaders(td, loader->get());
  return std::move(td_with_lock);
}
namespace {

size_t get_column_size(const TColumn& column) {
  if (!column.nulls.empty()) {
    return column.nulls.size();
  } else {
    // it is a very bold estimate but later we check it against REAL data
    // and if this function returns a wrong result (e.g. both int and string
    // vectors are filled with values), we get an error
    return column.data.int_col.size() + column.data.arr_col.size() +
           column.data.real_col.size() + column.data.str_col.size();
  }
}

}  // namespace

void DBHandler::load_table_binary_columnar(const TSessionId& session_id_or_json,
                                           const std::string& table_name,
                                           const std::vector<TColumn>& cols,
                                           const std::vector<std::string>& column_names) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog =
      STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();

  const auto execute_read_lock = legacylockmgr::getExecuteReadLock();
  std::unique_ptr<import_export::Loader> loader;
  std::vector<std::unique_ptr<import_export::TypedImportBuffer>> import_buffers;
  auto schema_read_lock = prepare_loader_generic(*session_ptr,
                                                 table_name,
                                                 cols.size(),
                                                 &loader,
                                                 &import_buffers,
                                                 column_names,
                                                 "load_table_binary_columnar");

  auto desc_id_to_column_id =
      column_ids_by_names(loader->get_column_descs(), column_names);
  size_t num_rows = get_column_size(cols.front());
  size_t import_idx = 0;  // index into the TColumn vector being loaded
  size_t col_idx = 0;     // index into column description vector
  auto const load_tag = get_load_tag("load_table_binary_columnar", table_name);
  log_system_cpu_memory_status("start_" + load_tag, session_ptr->getCatalog());
  ScopeGuard cleanup = [&load_tag, &session_ptr]() {
    log_system_cpu_memory_status("finish_" + load_tag, session_ptr->getCatalog());
  };
  try {
    size_t skip_physical_cols = 0;
    for (auto cd : loader->get_column_descs()) {
      if (skip_physical_cols > 0) {
        CHECK(cd->isGeoPhyCol);
        skip_physical_cols--;
        continue;
      }
      auto mapped_idx = desc_id_to_column_id[import_idx];
      if (mapped_idx != -1) {
        size_t col_rows = import_buffers[col_idx]->add_values(cd, cols[mapped_idx]);
        if (col_rows != num_rows) {
          std::ostringstream oss;
          oss << "load_table_binary_columnar: Inconsistent number of rows in column "
              << cd->columnName << " ,  expecting " << num_rows << " rows, column "
              << col_idx << " has " << col_rows << " rows";
          THROW_DB_EXCEPTION(oss.str());
        }
        // Advance to the next column in the table
        col_idx++;
        // For geometry columns: process WKT strings and fill physical columns
        if (cd->columnType.is_geometry()) {
          fillGeoColumns(request_info.sessionId(),
                         session_ptr->getCatalog(),
                         import_buffers,
                         cd,
                         col_idx,
                         num_rows,
                         table_name);
          skip_physical_cols = cd->columnType.get_physical_cols();
        }
      } else {
        col_idx++;
        if (cd->columnType.is_geometry()) {
          skip_physical_cols = cd->columnType.get_physical_cols();
          col_idx += skip_physical_cols;
        }
      }
      // Advance to the next column of values being loaded
      import_idx++;
    }
  } catch (const std::exception& e) {
    std::ostringstream oss;
    oss << "load_table_binary_columnar: Input exception thrown: " << e.what()
        << ". Issue at column : " << (col_idx + 1) << ". Import aborted";
    THROW_DB_EXCEPTION(oss.str());
  }
  fillMissingBuffers(request_info.sessionId(),
                     session_ptr->getCatalog(),
                     import_buffers,
                     loader->get_column_descs(),
                     desc_id_to_column_id,
                     num_rows,
                     table_name);
  auto insert_data_lock = lockmgr::InsertDataLockMgr::getWriteLockForTable(
      session_ptr->getCatalog(), table_name);
  if (!loader->load(import_buffers, num_rows, session_ptr.get())) {
    THROW_DB_EXCEPTION(loader->getErrorMessage());
  }
}

using RecordBatchVector = std::vector<std::shared_ptr<arrow::RecordBatch>>;

#define ARROW_THRIFT_THROW_NOT_OK(s) \
  do {                               \
    ::arrow::Status _s = (s);        \
    if (UNLIKELY(!_s.ok())) {        \
      TDBException ex;               \
      ex.error_msg = _s.ToString();  \
      LOG(ERROR) << s.ToString();    \
      throw ex;                      \
    }                                \
  } while (0)

namespace {

RecordBatchVector loadArrowStream(const std::string& stream) {
  RecordBatchVector batches;
  try {
    // TODO(wesm): Make this simpler in general, see ARROW-1600
    auto stream_buffer =
        std::make_shared<arrow::Buffer>(reinterpret_cast<const uint8_t*>(stream.c_str()),
                                        static_cast<int64_t>(stream.size()));

    arrow::io::BufferReader buf_reader(stream_buffer);
    std::shared_ptr<arrow::RecordBatchReader> batch_reader;
    ARROW_ASSIGN_OR_THROW(batch_reader,
                          arrow::ipc::RecordBatchStreamReader::Open(&buf_reader));

    while (true) {
      std::shared_ptr<arrow::RecordBatch> batch;
      // Read batch (zero-copy) from the stream
      ARROW_THRIFT_THROW_NOT_OK(batch_reader->ReadNext(&batch));
      if (batch == nullptr) {
        break;
      }
      batches.emplace_back(std::move(batch));
    }
  } catch (const std::exception& e) {
    LOG(ERROR) << "Error parsing Arrow stream: " << e.what() << ". Import aborted";
  }
  return batches;
}

}  // namespace

void DBHandler::load_table_binary_arrow(const TSessionId& session_id_or_json,
                                        const std::string& table_name,
                                        const std::string& arrow_stream,
                                        const bool use_column_names) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog =
      STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
  auto session_ptr = stdlog.getConstSessionInfo();

  RecordBatchVector batches = loadArrowStream(arrow_stream);
  // Assuming have one batch for now
  if (batches.size() != 1) {
    THROW_DB_EXCEPTION("Expected a single Arrow record batch. Import aborted");
  }

  std::shared_ptr<arrow::RecordBatch> batch = batches[0];
  std::unique_ptr<import_export::Loader> loader;
  std::vector<std::unique_ptr<import_export::TypedImportBuffer>> import_buffers;
  std::vector<std::string> column_names;
  if (use_column_names) {
    column_names = batch->schema()->field_names();
  }
  const auto execute_read_lock = legacylockmgr::getExecuteReadLock();
  auto schema_read_lock =
      prepare_loader_generic(*session_ptr,
                             table_name,
                             static_cast<size_t>(batch->num_columns()),
                             &loader,
                             &import_buffers,
                             column_names,
                             "load_table_binary_arrow");

  auto desc_id_to_column_id =
      column_ids_by_names(loader->get_column_descs(), column_names);
  size_t num_rows = 0;

  // col_idx indexes "desc_id_to_column_id"
  size_t col_idx = 0;
  auto const load_tag = get_load_tag("load_table_binary_arrow", table_name);
  log_system_cpu_memory_status("start_" + load_tag, session_ptr->getCatalog());
  ScopeGuard cleanup = [&load_tag, &session_ptr]() {
    log_system_cpu_memory_status("finish_" + load_tag, session_ptr->getCatalog());
  };
  try {
    for (auto cd : loader->get_column_descs()) {
      if (cd->isGeoPhyCol) {
        // Skip in the case of "cd" being a physical cols, as they are generated
        // in fillGeoColumns:
        //  * Point: coords col
        //  * MultiPoint/LineString: coords/bounds cols
        //  etc...
        continue;
      }
      auto mapped_idx = desc_id_to_column_id[col_idx];
      if (mapped_idx != -1) {
        auto& array = *batch->column(mapped_idx);
        import_export::ArraySliceRange row_slice(0, array.length());

        // col_id indexes "import_buffers"
        size_t col_id = cd->columnId;

        // When importing a buffer with "add_arrow_values", the index in
        // "importing_buffers" is given by the "columnId" attribute of a ColumnDescriptor.
        // This index will differ from "col_idx" if any of the importing columns is a
        // geometry column as they have physical columns for other properties (i.e. a
        // LineString also has "coords" and "bounds").
        num_rows = import_buffers[col_id - 1]->add_arrow_values(
            cd, array, true, row_slice, nullptr);
        // For geometry columns: process WKT strings and fill physical columns
        if (cd->columnType.is_geometry()) {
          fillGeoColumns(request_info.sessionId(),
                         session_ptr->getCatalog(),
                         import_buffers,
                         cd,
                         col_id,
                         num_rows,
                         table_name);
        }
      }
      // Advance to the next column in the table
      col_idx++;
    }
  } catch (const std::exception& e) {
    LOG(ERROR) << "Input exception thrown: " << e.what()
               << ". Issue at column : " << (col_idx + 1) << ". Import aborted";
    // TODO(tmostak): Go row-wise on binary columnar import to be consistent with our
    // other import paths
    THROW_DB_EXCEPTION(e.what());
  }
  fillMissingBuffers(request_info.sessionId(),
                     session_ptr->getCatalog(),
                     import_buffers,
                     loader->get_column_descs(),
                     desc_id_to_column_id,
                     num_rows,
                     table_name);
  auto insert_data_lock = lockmgr::InsertDataLockMgr::getWriteLockForTable(
      session_ptr->getCatalog(), table_name);
  if (!loader->load(import_buffers, num_rows, session_ptr.get())) {
    THROW_DB_EXCEPTION(loader->getErrorMessage());
  }
}

void DBHandler::load_table(const TSessionId& session_id_or_json,
                           const std::string& table_name,
                           const std::vector<TStringRow>& rows,
                           const std::vector<std::string>& column_names) {
  try {
    heavyai::RequestInfo const request_info(session_id_or_json);
    SET_REQUEST_ID(request_info.requestId());
    auto stdlog =
        STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
    stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
    auto session_ptr = stdlog.getConstSessionInfo();

    if (rows.empty()) {
      THROW_DB_EXCEPTION("No rows to insert");
    }
    auto const load_tag = get_load_tag("load_table", table_name);
    log_system_cpu_memory_status("start_" + load_tag, session_ptr->getCatalog());
    ScopeGuard cleanup = [&load_tag, &session_ptr]() {
      log_system_cpu_memory_status("finish_" + load_tag, session_ptr->getCatalog());
    };
    const auto execute_read_lock = legacylockmgr::getExecuteReadLock();
    std::unique_ptr<import_export::Loader> loader;
    std::vector<std::unique_ptr<import_export::TypedImportBuffer>> import_buffers;
    auto schema_read_lock =
        prepare_loader_generic(*session_ptr,
                               table_name,
                               static_cast<size_t>(rows.front().cols.size()),
                               &loader,
                               &import_buffers,
                               column_names,
                               "load_table");

    auto col_descs = loader->get_column_descs();
    auto desc_id_to_column_id = column_ids_by_names(col_descs, column_names);
    import_export::CopyParams copy_params;
    size_t rows_completed = 0;
    for (auto const& row : rows) {
      size_t import_idx = 0;  // index into the TStringRow being loaded
      size_t col_idx = 0;     // index into column description vector
      try {
        size_t skip_physical_cols = 0;
        for (auto cd : col_descs) {
          if (skip_physical_cols > 0) {
            CHECK(cd->isGeoPhyCol);
            skip_physical_cols--;
            continue;
          }
          auto mapped_idx = desc_id_to_column_id[import_idx];
          if (mapped_idx != -1) {
            import_buffers[col_idx]->add_value(cd,
                                               row.cols[mapped_idx].str_val,
                                               row.cols[mapped_idx].is_null,
                                               copy_params);
          }
          col_idx++;
          if (cd->columnType.is_geometry()) {
            // physical geo columns will be filled separately lately
            skip_physical_cols = cd->columnType.get_physical_cols();
            col_idx += skip_physical_cols;
          }
          // Advance to the next field within the row
          import_idx++;
        }
        rows_completed++;
      } catch (const std::exception& e) {
        LOG(ERROR) << "Input exception thrown: " << e.what()
                   << ". Row discarded, issue at column : " << (col_idx + 1)
                   << " data :" << row;
        THROW_DB_EXCEPTION(std::string("Exception: ") + e.what());
      }
    }
    // do batch filling of geo columns separately
    if (rows.size() != 0) {
      const auto& row = rows[0];
      size_t col_idx = 0;  // index into column description vector
      try {
        size_t import_idx = 0;
        size_t skip_physical_cols = 0;
        for (auto cd : col_descs) {
          if (skip_physical_cols > 0) {
            skip_physical_cols--;
            continue;
          }
          auto mapped_idx = desc_id_to_column_id[import_idx];
          col_idx++;
          if (cd->columnType.is_geometry()) {
            skip_physical_cols = cd->columnType.get_physical_cols();
            if (mapped_idx != -1) {
              fillGeoColumns(request_info.sessionId(),
                             session_ptr->getCatalog(),
                             import_buffers,
                             cd,
                             col_idx,
                             rows_completed,
                             table_name);
            } else {
              col_idx += skip_physical_cols;
            }
          }
          import_idx++;
        }
      } catch (const std::exception& e) {
        LOG(ERROR) << "Input exception thrown: " << e.what()
                   << ". Row discarded, issue at column : " << (col_idx + 1)
                   << " data :" << row;
        THROW_DB_EXCEPTION(e.what());
      }
    }
    fillMissingBuffers(request_info.sessionId(),
                       session_ptr->getCatalog(),
                       import_buffers,
                       col_descs,
                       desc_id_to_column_id,
                       rows_completed,
                       table_name);
    auto insert_data_lock = lockmgr::InsertDataLockMgr::getWriteLockForTable(
        session_ptr->getCatalog(), table_name);
    if (!loader->load(import_buffers, rows_completed, session_ptr.get())) {
      THROW_DB_EXCEPTION(loader->getErrorMessage());
    }

  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(std::string(e.what()));
  }
}

char DBHandler::unescape_char(std::string str) {
  char out = str[0];
  if (str.size() == 2 && str[0] == '\\') {
    if (str[1] == 't') {
      out = '\t';
    } else if (str[1] == 'n') {
      out = '\n';
    } else if (str[1] == '0') {
      out = '\0';
    } else if (str[1] == '\'') {
      out = '\'';
    } else if (str[1] == '\\') {
      out = '\\';
    }
  }
  return out;
}

import_export::CopyParams DBHandler::thrift_to_copyparams(const TCopyParams& cp) {
  import_export::CopyParams copy_params;
  switch (cp.has_header) {
    case TImportHeaderRow::AUTODETECT:
      copy_params.has_header = import_export::ImportHeaderRow::kAutoDetect;
      break;
    case TImportHeaderRow::NO_HEADER:
      copy_params.has_header = import_export::ImportHeaderRow::kNoHeader;
      break;
    case TImportHeaderRow::HAS_HEADER:
      copy_params.has_header = import_export::ImportHeaderRow::kHasHeader;
      break;
    default:
      CHECK(false);
  }
  copy_params.quoted = cp.quoted;
  if (cp.delimiter.length() > 0) {
    copy_params.delimiter = unescape_char(cp.delimiter);
  } else {
    copy_params.delimiter = '\0';
  }
  if (cp.null_str.length() > 0) {
    copy_params.null_str = cp.null_str;
  }
  if (cp.quote.length() > 0) {
    copy_params.quote = unescape_char(cp.quote);
  }
  if (cp.escape.length() > 0) {
    copy_params.escape = unescape_char(cp.escape);
  }
  if (cp.line_delim.length() > 0) {
    copy_params.line_delim = unescape_char(cp.line_delim);
  }
  if (cp.array_delim.length() > 0) {
    copy_params.array_delim = unescape_char(cp.array_delim);
  }
  if (cp.array_begin.length() > 0) {
    copy_params.array_begin = unescape_char(cp.array_begin);
  }
  if (cp.array_end.length() > 0) {
    copy_params.array_end = unescape_char(cp.array_end);
  }
  if (cp.threads != 0) {
    copy_params.threads = cp.threads;
  }
  if (cp.s3_access_key.length() > 0) {
    copy_params.s3_access_key = cp.s3_access_key;
  }
  if (cp.s3_secret_key.length() > 0) {
    copy_params.s3_secret_key = cp.s3_secret_key;
  }
  if (cp.s3_session_token.length() > 0) {
    copy_params.s3_session_token = cp.s3_session_token;
  }
  if (cp.s3_region.length() > 0) {
    copy_params.s3_region = cp.s3_region;
  }
  if (cp.s3_endpoint.length() > 0) {
    copy_params.s3_endpoint = cp.s3_endpoint;
  }
#ifdef HAVE_AWS_S3
  if (g_allow_s3_server_privileges && cp.s3_access_key.length() == 0 &&
      cp.s3_secret_key.length() == 0 && cp.s3_session_token.length() == 0) {
    const auto& server_credentials =
        Aws::Auth::DefaultAWSCredentialsProviderChain().GetAWSCredentials();
    copy_params.s3_access_key = server_credentials.GetAWSAccessKeyId();
    copy_params.s3_secret_key = server_credentials.GetAWSSecretKey();
    copy_params.s3_session_token = server_credentials.GetSessionToken();
  }
#endif

  switch (cp.source_type) {
    case TSourceType::DELIMITED_FILE:
      copy_params.source_type = import_export::SourceType::kDelimitedFile;
      break;
    case TSourceType::GEO_FILE:
      copy_params.source_type = import_export::SourceType::kGeoFile;
      break;
    case TSourceType::PARQUET_FILE:
#ifdef ENABLE_IMPORT_PARQUET
      copy_params.source_type = import_export::SourceType::kParquetFile;
      break;
#else
      THROW_DB_EXCEPTION("Parquet not supported");
#endif
    case TSourceType::ODBC:
      THROW_DB_EXCEPTION("ODBC source not supported");
    case TSourceType::RASTER_FILE:
      copy_params.source_type = import_export::SourceType::kRasterFile;
      break;
    default:
      CHECK(false);
  }

  switch (cp.geo_coords_encoding) {
    case TEncodingType::GEOINT:
      copy_params.geo_coords_encoding = kENCODING_GEOINT;
      break;
    case TEncodingType::NONE:
      copy_params.geo_coords_encoding = kENCODING_NONE;
      break;
    default:
      THROW_DB_EXCEPTION("Invalid geo_coords_encoding in TCopyParams: " +
                         std::to_string((int)cp.geo_coords_encoding));
  }
  copy_params.geo_coords_comp_param = cp.geo_coords_comp_param;
  switch (cp.geo_coords_type) {
    case TDatumType::GEOGRAPHY:
      copy_params.geo_coords_type = kGEOGRAPHY;
      break;
    case TDatumType::GEOMETRY:
      copy_params.geo_coords_type = kGEOMETRY;
      break;
    default:
      THROW_DB_EXCEPTION("Invalid geo_coords_type in TCopyParams: " +
                         std::to_string((int)cp.geo_coords_type));
  }
  switch (cp.geo_coords_srid) {
    case 4326:
    case 3857:
    case 900913:
      copy_params.geo_coords_srid = cp.geo_coords_srid;
      break;
    default:
      THROW_DB_EXCEPTION("Invalid geo_coords_srid in TCopyParams (" +
                         std::to_string((int)cp.geo_coords_srid));
  }
  copy_params.sanitize_column_names = cp.sanitize_column_names;
  copy_params.geo_layer_name = cp.geo_layer_name;
  copy_params.geo_explode_collections = cp.geo_explode_collections;
  copy_params.source_srid = cp.source_srid;
  switch (cp.raster_point_type) {
    case TRasterPointType::NONE:
      copy_params.raster_point_type = import_export::RasterPointType::kNone;
      break;
    case TRasterPointType::AUTO:
      copy_params.raster_point_type = import_export::RasterPointType::kAuto;
      break;
    case TRasterPointType::SMALLINT:
      copy_params.raster_point_type = import_export::RasterPointType::kSmallInt;
      break;
    case TRasterPointType::INT:
      copy_params.raster_point_type = import_export::RasterPointType::kInt;
      break;
    case TRasterPointType::FLOAT:
      copy_params.raster_point_type = import_export::RasterPointType::kFloat;
      break;
    case TRasterPointType::DOUBLE:
      copy_params.raster_point_type = import_export::RasterPointType::kDouble;
      break;
    case TRasterPointType::POINT:
      copy_params.raster_point_type = import_export::RasterPointType::kPoint;
      break;
    default:
      CHECK(false);
  }
  copy_params.raster_import_bands = cp.raster_import_bands;
  if (cp.raster_scanlines_per_thread < 0) {
    THROW_DB_EXCEPTION("Invalid raster_scanlines_per_thread in TCopyParams (" +
                       std::to_string((int)cp.raster_scanlines_per_thread));
  } else {
    copy_params.raster_scanlines_per_thread = cp.raster_scanlines_per_thread;
  }
  switch (cp.raster_point_transform) {
    case TRasterPointTransform::NONE:
      copy_params.raster_point_transform = import_export::RasterPointTransform::kNone;
      break;
    case TRasterPointTransform::AUTO:
      copy_params.raster_point_transform = import_export::RasterPointTransform::kAuto;
      break;
    case TRasterPointTransform::FILE:
      copy_params.raster_point_transform = import_export::RasterPointTransform::kFile;
      break;
    case TRasterPointTransform::WORLD:
      copy_params.raster_point_transform = import_export::RasterPointTransform::kWorld;
      break;
    default:
      CHECK(false);
  }
  copy_params.raster_point_compute_angle = cp.raster_point_compute_angle;
  copy_params.raster_import_dimensions = cp.raster_import_dimensions;
  copy_params.dsn = cp.odbc_dsn;
  copy_params.connection_string = cp.odbc_connection_string;
  copy_params.sql_select = cp.odbc_sql_select;
  copy_params.sql_order_by = cp.odbc_sql_order_by;
  copy_params.username = cp.odbc_username;
  copy_params.password = cp.odbc_password;
  copy_params.credential_string = cp.odbc_credential_string;
  copy_params.add_metadata_columns = cp.add_metadata_columns;
  copy_params.trim_spaces = cp.trim_spaces;
  copy_params.geo_validate_geometry = cp.geo_validate_geometry;
  copy_params.raster_drop_if_all_null = cp.raster_drop_if_all_null;
  return copy_params;
}

TCopyParams DBHandler::copyparams_to_thrift(const import_export::CopyParams& cp) {
  TCopyParams copy_params;
  copy_params.delimiter = cp.delimiter;
  copy_params.null_str = cp.null_str;
  switch (cp.has_header) {
    case import_export::ImportHeaderRow::kAutoDetect:
      copy_params.has_header = TImportHeaderRow::AUTODETECT;
      break;
    case import_export::ImportHeaderRow::kNoHeader:
      copy_params.has_header = TImportHeaderRow::NO_HEADER;
      break;
    case import_export::ImportHeaderRow::kHasHeader:
      copy_params.has_header = TImportHeaderRow::HAS_HEADER;
      break;
    default:
      CHECK(false);
  }
  copy_params.quoted = cp.quoted;
  copy_params.quote = cp.quote;
  copy_params.escape = cp.escape;
  copy_params.line_delim = cp.line_delim;
  copy_params.array_delim = cp.array_delim;
  copy_params.array_begin = cp.array_begin;
  copy_params.array_end = cp.array_end;
  copy_params.threads = cp.threads;
  copy_params.s3_access_key = cp.s3_access_key;
  copy_params.s3_secret_key = cp.s3_secret_key;
  copy_params.s3_session_token = cp.s3_session_token;
  copy_params.s3_region = cp.s3_region;
  copy_params.s3_endpoint = cp.s3_endpoint;
  switch (cp.source_type) {
    case import_export::SourceType::kDelimitedFile:
      copy_params.source_type = TSourceType::DELIMITED_FILE;
      break;
    case import_export::SourceType::kGeoFile:
      copy_params.source_type = TSourceType::GEO_FILE;
      break;
    case import_export::SourceType::kParquetFile:
      copy_params.source_type = TSourceType::PARQUET_FILE;
      break;
    case import_export::SourceType::kRasterFile:
      copy_params.source_type = TSourceType::RASTER_FILE;
      break;
    case import_export::SourceType::kOdbc:
      copy_params.source_type = TSourceType::ODBC;
      break;
    default:
      CHECK(false);
  }
  switch (cp.geo_coords_encoding) {
    case kENCODING_GEOINT:
      copy_params.geo_coords_encoding = TEncodingType::GEOINT;
      break;
    default:
      copy_params.geo_coords_encoding = TEncodingType::NONE;
      break;
  }
  copy_params.geo_coords_comp_param = cp.geo_coords_comp_param;
  switch (cp.geo_coords_type) {
    case kGEOGRAPHY:
      copy_params.geo_coords_type = TDatumType::GEOGRAPHY;
      break;
    case kGEOMETRY:
      copy_params.geo_coords_type = TDatumType::GEOMETRY;
      break;
    default:
      CHECK(false);
  }
  copy_params.geo_coords_srid = cp.geo_coords_srid;
  copy_params.sanitize_column_names = cp.sanitize_column_names;
  copy_params.geo_layer_name = cp.geo_layer_name;
  copy_params.geo_assign_render_groups = false;
  copy_params.geo_explode_collections = cp.geo_explode_collections;
  copy_params.source_srid = cp.source_srid;
  switch (cp.raster_point_type) {
    case import_export::RasterPointType::kNone:
      copy_params.raster_point_type = TRasterPointType::NONE;
      break;
    case import_export::RasterPointType::kAuto:
      copy_params.raster_point_type = TRasterPointType::AUTO;
      break;
    case import_export::RasterPointType::kSmallInt:
      copy_params.raster_point_type = TRasterPointType::SMALLINT;
      break;
    case import_export::RasterPointType::kInt:
      copy_params.raster_point_type = TRasterPointType::INT;
      break;
    case import_export::RasterPointType::kFloat:
      copy_params.raster_point_type = TRasterPointType::FLOAT;
      break;
    case import_export::RasterPointType::kDouble:
      copy_params.raster_point_type = TRasterPointType::DOUBLE;
      break;
    case import_export::RasterPointType::kPoint:
      copy_params.raster_point_type = TRasterPointType::POINT;
      break;
    default:
      CHECK(false);
  }
  copy_params.raster_import_bands = cp.raster_import_bands;
  copy_params.raster_scanlines_per_thread = cp.raster_scanlines_per_thread;
  switch (cp.raster_point_transform) {
    case import_export::RasterPointTransform::kNone:
      copy_params.raster_point_transform = TRasterPointTransform::NONE;
      break;
    case import_export::RasterPointTransform::kAuto:
      copy_params.raster_point_transform = TRasterPointTransform::AUTO;
      break;
    case import_export::RasterPointTransform::kFile:
      copy_params.raster_point_transform = TRasterPointTransform::FILE;
      break;
    case import_export::RasterPointTransform::kWorld:
      copy_params.raster_point_transform = TRasterPointTransform::WORLD;
      break;
    default:
      CHECK(false);
  }
  copy_params.raster_point_compute_angle = cp.raster_point_compute_angle;
  copy_params.raster_import_dimensions = cp.raster_import_dimensions;
  copy_params.odbc_dsn = cp.dsn;
  copy_params.odbc_connection_string = cp.connection_string;
  copy_params.odbc_sql_select = cp.sql_select;
  copy_params.odbc_sql_order_by = cp.sql_order_by;
  copy_params.odbc_username = cp.username;
  copy_params.odbc_password = cp.password;
  copy_params.odbc_credential_string = cp.credential_string;
  copy_params.add_metadata_columns = cp.add_metadata_columns;
  copy_params.trim_spaces = cp.trim_spaces;
  copy_params.geo_validate_geometry = cp.geo_validate_geometry;
  copy_params.raster_drop_if_all_null = cp.raster_drop_if_all_null;
  return copy_params;
}

namespace {
void add_vsi_network_prefix(std::string& path) {
  // do we support network file access?
  bool gdal_network = Geospatial::GDAL::supportsNetworkFileAccess();

  // modify head of filename based on source location
  if (boost::istarts_with(path, "http://") || boost::istarts_with(path, "https://")) {
    if (!gdal_network) {
      THROW_DB_EXCEPTION(
          "HTTP geo file import not supported! Update to GDAL 2.2 or later!");
    }
    // invoke GDAL CURL virtual file reader
    path = "/vsicurl/" + path;
  } else if (boost::istarts_with(path, "s3://")) {
    if (!gdal_network) {
      THROW_DB_EXCEPTION(
          "S3 geo file import not supported! Update to GDAL 2.2 or later!");
    }
    // invoke GDAL S3 virtual file reader
    boost::replace_first(path, "s3://", "/vsis3/");
  }
}

void add_vsi_geo_prefix(std::string& path) {
  // single gzip'd file (not an archive)?
  if (boost::iends_with(path, ".gz") && !boost::iends_with(path, ".tar.gz")) {
    path = "/vsigzip/" + path;
  }
}

void add_vsi_archive_prefix(std::string& path) {
  // check for compressed file or file bundle
  if (boost::iends_with(path, ".zip")) {
    // zip archive
    path = "/vsizip/" + path;
  } else if (boost::iends_with(path, ".tar") || boost::iends_with(path, ".tgz") ||
             boost::iends_with(path, ".tar.gz")) {
    // tar archive (compressed or uncompressed)
    path = "/vsitar/" + path;
  }
}

std::string remove_vsi_prefixes(const std::string& path_in) {
  std::string path(path_in);

  // these will be first
  if (boost::istarts_with(path, "/vsizip/")) {
    boost::replace_first(path, "/vsizip/", "");
  } else if (boost::istarts_with(path, "/vsitar/")) {
    boost::replace_first(path, "/vsitar/", "");
  } else if (boost::istarts_with(path, "/vsigzip/")) {
    boost::replace_first(path, "/vsigzip/", "");
  }

  // then these
  if (boost::istarts_with(path, "/vsicurl/")) {
    boost::replace_first(path, "/vsicurl/", "");
  } else if (boost::istarts_with(path, "/vsis3/")) {
    boost::replace_first(path, "/vsis3/", "s3://");
  }

  return path;
}

bool path_is_relative(const std::string& path) {
  if (boost::istarts_with(path, "s3://") || boost::istarts_with(path, "http://") ||
      boost::istarts_with(path, "https://")) {
    return false;
  }
  return !boost::filesystem::path(path).is_absolute();
}

bool path_has_valid_filename(const std::string& path) {
  auto filename = boost::filesystem::path(path).filename().string();
  if (filename.size() == 0 || filename[0] == '.' || filename[0] == '/') {
    return false;
  }
  return true;
}

bool is_a_supported_geo_file(const std::string& path) {
  if (!path_has_valid_filename(path)) {
    return false;
  }
  // this is now just for files that we want to recognize
  // as geo when inside an archive (see below)
  // @TODO(se) make this more flexible?
  if (boost::iends_with(path, ".shp") || boost::iends_with(path, ".geojson") ||
      boost::iends_with(path, ".json") || boost::iends_with(path, ".kml") ||
      boost::iends_with(path, ".kmz") || boost::iends_with(path, ".gdb") ||
      boost::iends_with(path, ".gdb.zip") || boost::iends_with(path, ".fgb")) {
    return true;
  }
  return false;
}

bool is_a_supported_archive_file(const std::string& path) {
  if (!path_has_valid_filename(path)) {
    return false;
  }
  if (boost::iends_with(path, ".zip") && !boost::iends_with(path, ".gdb.zip")) {
    return true;
  } else if (boost::iends_with(path, ".tar") || boost::iends_with(path, ".tgz") ||
             boost::iends_with(path, ".tar.gz")) {
    return true;
  }
  return false;
}

std::string find_first_geo_file_in_archive(const std::string& archive_path,
                                           const import_export::CopyParams& copy_params) {
  // get the recursive list of all files in the archive
  std::vector<std::string> files =
      import_export::Importer::gdalGetAllFilesInArchive(archive_path, copy_params);

  // report the list
  LOG(INFO) << "Found " << files.size() << " files in Archive "
            << remove_vsi_prefixes(archive_path);
  for (const auto& file : files) {
    LOG(INFO) << "  " << file;
  }

  // scan the list for the first candidate file
  bool found_suitable_file = false;
  std::string file_name;
  for (const auto& file : files) {
    if (is_a_supported_geo_file(file)) {
      file_name = file;
      found_suitable_file = true;
      break;
    }
  }

  // if we didn't find anything
  if (!found_suitable_file) {
    LOG(INFO) << "Failed to find any supported geo files in Archive: " +
                     remove_vsi_prefixes(archive_path);
    file_name.clear();
  }

  // done
  return file_name;
}

bool is_local_file(const std::string& file_path) {
  return (!boost::istarts_with(file_path, "s3://") &&
          !boost::istarts_with(file_path, "http://") &&
          !boost::istarts_with(file_path, "https://"));
}

void validate_import_file_path_if_local(const std::string& file_path) {
  if (is_local_file(file_path)) {
    ddl_utils::validate_allowed_file_path(
        file_path, ddl_utils::DataTransferType::IMPORT, true);
  }
}
}  // namespace

void DBHandler::detect_column_types(TDetectResult& _return,
                                    const TSessionId& session_id_or_json,
                                    const std::string& file_name_in,
                                    const TCopyParams& cp) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  check_read_only("detect_column_types");

  bool is_raster = false;
  boost::filesystem::path file_path;
  import_export::CopyParams copy_params = thrift_to_copyparams(cp);
  if (copy_params.source_type != import_export::SourceType::kOdbc) {
    std::string file_name{file_name_in};
    if (path_is_relative(file_name)) {
      // assume relative paths are relative to data_path / import / <session>
      auto temp_file_path = import_path_ /
                            picosha2::hash256_hex_string(request_info.sessionId()) /
                            boost::filesystem::path(file_name).filename();
      file_name = temp_file_path.string();
    }
    validate_import_file_path_if_local(file_name);

    if ((copy_params.source_type == import_export::SourceType::kGeoFile ||
         copy_params.source_type == import_export::SourceType::kRasterFile) &&
        is_local_file(file_name)) {
      const shared::FilePathOptions options{copy_params.regex_path_filter,
                                            copy_params.file_sort_order_by,
                                            copy_params.file_sort_regex};
      auto file_paths = shared::local_glob_filter_sort_files(file_name, options, false);
      // For geo and raster detect, pick the first file, if multiple files are provided
      // (e.g. through file globbing).
      CHECK(!file_paths.empty());
      file_name = file_paths[0];
    }

    // if it's a geo or raster import, handle alternative paths (S3, HTTP, archive etc.)
    if (copy_params.source_type == import_export::SourceType::kGeoFile) {
      if (is_a_supported_archive_file(file_name)) {
        // find the archive file
        add_vsi_network_prefix(file_name);
        if (!import_export::Importer::gdalFileExists(file_name, copy_params)) {
          THROW_DB_EXCEPTION("Archive does not exist: " + file_name_in);
        }
        // find geo file in archive
        add_vsi_archive_prefix(file_name);
        std::string geo_file = find_first_geo_file_in_archive(file_name, copy_params);
        // prepare to detect that geo file
        if (geo_file.size()) {
          file_name = file_name + std::string("/") + geo_file;
        }
      } else {
        // prepare to detect geo file directly
        add_vsi_network_prefix(file_name);
        add_vsi_geo_prefix(file_name);
      }
    } else if (copy_params.source_type == import_export::SourceType::kRasterFile) {
      // prepare to detect raster file directly
      add_vsi_network_prefix(file_name);
      add_vsi_geo_prefix(file_name);
      is_raster = true;
    }

    file_path = boost::filesystem::path(file_name);
    // can be a s3 url
    if (!boost::istarts_with(file_name, "s3://")) {
      if (!boost::filesystem::path(file_name).is_absolute()) {
        file_path = import_path_ /
                    picosha2::hash256_hex_string(request_info.sessionId()) /
                    boost::filesystem::path(file_name).filename();
        file_name = file_path.string();
      }

      if (copy_params.source_type == import_export::SourceType::kGeoFile ||
          copy_params.source_type == import_export::SourceType::kRasterFile) {
        // check for geo or raster file
        if (!import_export::Importer::gdalFileOrDirectoryExists(file_name, copy_params)) {
          THROW_DB_EXCEPTION("File or directory \"" + file_path.string() +
                             "\" does not exist.")
        }
      } else {
        // check for regular file
        if (!shared::file_or_glob_path_exists(file_path.string())) {
          THROW_DB_EXCEPTION("File or directory \"" + file_path.string() +
                             "\" does not exist.");
        }
      }
    }
  }

  try {
    if (copy_params.source_type == import_export::SourceType::kDelimitedFile
#ifdef ENABLE_IMPORT_PARQUET
        || (copy_params.source_type == import_export::SourceType::kParquetFile)
#endif
    ) {
      import_export::Detector detector(file_path, copy_params);
      auto best_types = detector.getBestColumnTypes();
      std::vector<std::string> headers = detector.get_headers();
      copy_params = detector.get_copy_params();

      _return.copy_params = copyparams_to_thrift(copy_params);
      _return.row_set.row_desc.resize(best_types.size());
      for (size_t col_idx = 0; col_idx < best_types.size(); col_idx++) {
        TColumnType col;
        auto& ti = best_types[col_idx];
        col.col_type.precision = ti.get_precision();
        col.col_type.scale = ti.get_scale();
        col.col_type.comp_param = ti.get_comp_param();
        if (ti.is_geometry()) {
          // set this so encoding_to_thrift does the right thing
          ti.set_compression(copy_params.geo_coords_encoding);
          // fill in these directly
          col.col_type.precision = static_cast<int>(copy_params.geo_coords_type);
          col.col_type.scale = copy_params.geo_coords_srid;
          col.col_type.comp_param = copy_params.geo_coords_comp_param;
        }
        col.col_type.type = type_to_thrift(ti);
        col.col_type.encoding = encoding_to_thrift(ti);
        if (ti.is_array()) {
          col.col_type.is_array = true;
        }
        if (copy_params.sanitize_column_names) {
          col.col_name = ImportHelpers::sanitize_name(headers[col_idx]);
        } else {
          col.col_name = headers[col_idx];
        }
        col.is_reserved_keyword = ImportHelpers::is_reserved_name(col.col_name);
        _return.row_set.row_desc[col_idx] = col;
      }
      auto sample_data = detector.get_sample_rows(shared::kDefaultSampleRowsCount);

      TRow sample_row;
      for (auto row : sample_data) {
        sample_row.cols.clear();
        for (const auto& s : row) {
          TDatum td;
          td.val.str_val = s;
          td.is_null = s.empty();
          sample_row.cols.push_back(td);
        }
        _return.row_set.rows.push_back(sample_row);
      }
    } else if (copy_params.source_type == import_export::SourceType::kGeoFile ||
               copy_params.source_type == import_export::SourceType::kRasterFile) {
      check_geospatial_files(file_path, copy_params);
      std::list<ColumnDescriptor> cds = import_export::Importer::gdalToColumnDescriptors(
          file_path.string(), is_raster, Geospatial::kGeoColumnName, copy_params);
      for (auto cd : cds) {
        if (copy_params.sanitize_column_names) {
          cd.columnName = ImportHelpers::sanitize_name(cd.columnName);
        }
        _return.row_set.row_desc.push_back(populateThriftColumnType(nullptr, &cd));
      }
      if (!is_raster) {
        // @TODO(se) support for raster?
        std::map<std::string, std::vector<std::string>> sample_data;
        import_export::Importer::readMetadataSampleGDAL(file_path.string(),
                                                        Geospatial::kGeoColumnName,
                                                        sample_data,
                                                        shared::kDefaultSampleRowsCount,
                                                        copy_params);
        if (sample_data.size() > 0) {
          for (size_t i = 0; i < sample_data.begin()->second.size(); i++) {
            TRow sample_row;
            for (auto cd : cds) {
              TDatum td;
              td.val.str_val = sample_data[cd.sourceName].at(i);
              td.is_null = td.val.str_val.empty();
              sample_row.cols.push_back(td);
            }
            _return.row_set.rows.push_back(sample_row);
          }
        }
      }
      _return.copy_params = copyparams_to_thrift(copy_params);
    }
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION("detect_column_types error: " + std::string(e.what()));
  }
}

void DBHandler::render_vega(TRenderResult& _return,
                            const TSessionId& session_id_or_json,
                            const int64_t widget_id,
                            const std::string& vega_json,
                            const int compression_level,
                            const std::string& nonce) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()),
                       "widget_id",
                       widget_id,
                       "compression_level",
                       compression_level,
                       "vega_json",
                       vega_json,
                       "nonce",
                       nonce);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  stdlog.appendNameValuePairs("nonce", nonce);
  if (!render_handler_) {
    THROW_DB_EXCEPTION("Backend rendering is disabled.");
  }

  // cast away const-ness of incoming Thrift string ref
  // to allow it to be passed down as an r-value and
  // ultimately std::moved into the RenderSession
  auto& non_const_vega_json = const_cast<std::string&>(vega_json);

  _return.total_time_ms = measure<>::execution([&]() {
    try {
      render_handler_->render_vega(_return,
                                   stdlog.getSessionInfo(),
                                   widget_id,
                                   std::move(non_const_vega_json),
                                   compression_level,
                                   nonce);
    } catch (std::exception& e) {
      THROW_DB_EXCEPTION(e.what());
    }
  });
}

static bool is_allowed_on_dashboard(const Catalog_Namespace::SessionInfo& session_info,
                                    int32_t dashboard_id,
                                    AccessPrivileges requestedPermissions) {
  DBObject object(dashboard_id, DashboardDBObjectType);
  auto& catalog = session_info.getCatalog();
  auto& user = session_info.get_currentUser();
  object.loadKey(catalog);
  object.setPrivileges(requestedPermissions);
  std::vector<DBObject> privs = {object};
  return SysCatalog::instance().checkPrivileges(user, privs);
}

// custom expressions
namespace {
using Catalog_Namespace::CustomExpression;
using Catalog_Namespace::DataSourceType;

std::unique_ptr<Catalog_Namespace::CustomExpression> create_custom_expr_from_thrift_obj(
    const TCustomExpression& t_custom_expr,
    const Catalog& catalog) {
  if (t_custom_expr.data_source_name.empty()) {
    THROW_DB_EXCEPTION("Custom expression data source name cannot be empty.")
  }
  CHECK(t_custom_expr.data_source_type == TDataSourceType::type::TABLE)
      << "Unexpected data source type: "
      << static_cast<int>(t_custom_expr.data_source_type);
  auto td = catalog.getMetadataForTable(t_custom_expr.data_source_name, false);
  if (!td) {
    THROW_DB_EXCEPTION("Custom expression references a table \"" +
                       t_custom_expr.data_source_name + "\" that does not exist.")
  }
  DataSourceType data_source_type = DataSourceType::TABLE;
  return std::make_unique<CustomExpression>(
      t_custom_expr.name, t_custom_expr.expression_json, data_source_type, td->tableId);
}

TCustomExpression create_thrift_obj_from_custom_expr(const CustomExpression& custom_expr,
                                                     const Catalog& catalog) {
  TCustomExpression t_custom_expr;
  t_custom_expr.id = custom_expr.id;
  t_custom_expr.name = custom_expr.name;
  t_custom_expr.expression_json = custom_expr.expression_json;
  t_custom_expr.data_source_id = custom_expr.data_source_id;
  t_custom_expr.is_deleted = custom_expr.is_deleted;
  CHECK(custom_expr.data_source_type == DataSourceType::TABLE)
      << "Unexpected data source type: "
      << static_cast<int>(custom_expr.data_source_type);
  t_custom_expr.data_source_type = TDataSourceType::type::TABLE;
  auto td = catalog.getMetadataForTable(custom_expr.data_source_id, false);
  if (td) {
    t_custom_expr.data_source_name = td->tableName;
  } else {
    LOG(WARNING)
        << "Custom expression references a deleted data source. Custom expression id: "
        << custom_expr.id << ", name: " << custom_expr.name;
  }
  return t_custom_expr;
}
}  // namespace

int32_t DBHandler::create_custom_expression(const TSessionId& session_id_or_json,
                                            const TCustomExpression& t_custom_expr) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  check_read_only("create_custom_expression");

  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION("Custom expressions can only be created by super users.")
  }
  auto& catalog = session_ptr->getCatalog();
  heavyai::unique_lock<heavyai::shared_mutex> write_lock(custom_expressions_mutex_);
  return catalog.createCustomExpression(
      create_custom_expr_from_thrift_obj(t_custom_expr, catalog));
}

void DBHandler::get_custom_expressions(std::vector<TCustomExpression>& _return,
                                       const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());

  auto session_ptr = stdlog.getConstSessionInfo();
  auto& catalog = session_ptr->getCatalog();
  heavyai::shared_lock<heavyai::shared_mutex> read_lock(custom_expressions_mutex_);
  auto custom_expressions =
      catalog.getCustomExpressionsForUser(session_ptr->get_currentUser());
  for (const auto& custom_expression : custom_expressions) {
    _return.emplace_back(create_thrift_obj_from_custom_expr(*custom_expression, catalog));
  }
}

void DBHandler::update_custom_expression(const TSessionId& session_id_or_json,
                                         const int32_t id,
                                         const std::string& expression_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  check_read_only("update_custom_expression");

  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION("Custom expressions can only be updated by super users.")
  }
  auto& catalog = session_ptr->getCatalog();
  heavyai::unique_lock<heavyai::shared_mutex> write_lock(custom_expressions_mutex_);
  catalog.updateCustomExpression(id, expression_json);
}

void DBHandler::delete_custom_expressions(
    const TSessionId& session_id_or_json,
    const std::vector<int32_t>& custom_expression_ids,
    const bool do_soft_delete) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  check_read_only("delete_custom_expressions");

  auto session_ptr = stdlog.getConstSessionInfo();
  if (!session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION("Custom expressions can only be deleted by super users.")
  }
  auto& catalog = session_ptr->getCatalog();
  heavyai::unique_lock<heavyai::shared_mutex> write_lock(custom_expressions_mutex_);
  catalog.deleteCustomExpressions(custom_expression_ids, do_soft_delete);
}

// dashboards
void DBHandler::get_dashboard(TDashboard& dashboard,
                              const TSessionId& session_id_or_json,
                              const int32_t dashboard_id) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  auto const& cat = session_ptr->getCatalog();
  Catalog_Namespace::UserMetadata user_meta;
  auto dash = cat.getMetadataForDashboard(dashboard_id);
  if (!dash) {
    THROW_DB_EXCEPTION("Dashboard with dashboard id " + std::to_string(dashboard_id) +
                       " doesn't exist");
  }
  if (!is_allowed_on_dashboard(
          *session_ptr, dash->dashboardId, AccessPrivileges::VIEW_DASHBOARD)) {
    THROW_DB_EXCEPTION("User has no view privileges for the dashboard with id " +
                       std::to_string(dashboard_id));
  }
  user_meta.userName = "";
  SysCatalog::instance().getMetadataForUserById(dash->userId, user_meta);
  dashboard = get_dashboard_impl(session_ptr, user_meta, dash);
}

void DBHandler::get_dashboards(std::vector<TDashboard>& dashboards,
                               const TSessionId& session_id_or_json) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  auto const& cat = session_ptr->getCatalog();
  Catalog_Namespace::UserMetadata user_meta;
  const auto dashes = cat.getAllDashboardsMetadata();
  user_meta.userName = "";
  for (const auto dash : dashes) {
    if (is_allowed_on_dashboard(
            *session_ptr, dash->dashboardId, AccessPrivileges::VIEW_DASHBOARD)) {
      // dashboardState is intentionally not populated here
      // for payload reasons
      // use get_dashboard call to get state
      dashboards.push_back(get_dashboard_impl(session_ptr, user_meta, dash, false));
    }
  }
}

TDashboard DBHandler::get_dashboard_impl(
    const std::shared_ptr<Catalog_Namespace::SessionInfo const>& session_ptr,
    Catalog_Namespace::UserMetadata& user_meta,
    const DashboardDescriptor* dash,
    const bool populate_state) {
  auto const& cat = session_ptr->getCatalog();
  SysCatalog::instance().getMetadataForUserById(dash->userId, user_meta);
  auto objects_list = SysCatalog::instance().getMetadataForObject(
      cat.getCurrentDB().dbId,
      static_cast<int>(DBObjectType::DashboardDBObjectType),
      dash->dashboardId);
  TDashboard dashboard;
  dashboard.dashboard_name = dash->dashboardName;
  if (populate_state) {
    dashboard.dashboard_state = dash->dashboardState;
  }
  dashboard.image_hash = dash->imageHash;
  dashboard.update_time = dash->updateTime;
  dashboard.dashboard_metadata = dash->dashboardMetadata;
  dashboard.dashboard_id = dash->dashboardId;
  dashboard.dashboard_owner = dash->user;
  TDashboardPermissions perms;
  // Super user has all permissions.
  if (session_ptr->get_currentUser().isSuper) {
    perms.create_ = true;
    perms.delete_ = true;
    perms.edit_ = true;
    perms.view_ = true;
  } else {
    // Collect all grants on current user
    // add them to the permissions.
    auto obj_to_find =
        DBObject(dashboard.dashboard_id, DBObjectType::DashboardDBObjectType);
    obj_to_find.loadKey(cat);
    std::vector<std::string> grantees =
        SysCatalog::instance().getRoles(true,
                                        session_ptr->get_currentUser().isSuper,
                                        session_ptr->get_currentUser().userName);
    for (const auto& grantee : grantees) {
      DBObject* object_found;
      auto* gr = SysCatalog::instance().getGrantee(grantee);
      if (gr && (object_found = gr->findDbObject(obj_to_find.getObjectKey(), true))) {
        const auto obj_privs = object_found->getPrivileges();
        perms.create_ |= obj_privs.hasPermission(DashboardPrivileges::CREATE_DASHBOARD);
        perms.delete_ |= obj_privs.hasPermission(DashboardPrivileges::DELETE_DASHBOARD);
        perms.edit_ |= obj_privs.hasPermission(DashboardPrivileges::EDIT_DASHBOARD);
        perms.view_ |= obj_privs.hasPermission(DashboardPrivileges::VIEW_DASHBOARD);
      }
    }
  }
  dashboard.dashboard_permissions = perms;
  if (objects_list.empty() ||
      (objects_list.size() == 1 && objects_list[0]->roleName == user_meta.userName)) {
    dashboard.is_dash_shared = false;
  } else {
    dashboard.is_dash_shared = true;
  }
  return dashboard;
}

namespace dbhandler {
bool is_info_schema_db(const std::string& db_name) {
  return (db_name == shared::kInfoSchemaDbName &&
          SysCatalog::instance().hasExecutedMigration(shared::kInfoSchemaMigrationName));
}

void check_not_info_schema_db(const std::string& db_name, bool throw_db_exception) {
  if (is_info_schema_db(db_name)) {
    std::string error_message{"Write requests/queries are not allowed in the " +
                              shared::kInfoSchemaDbName + " database."};
    if (throw_db_exception) {
      THROW_DB_EXCEPTION(error_message)
    } else {
      throw std::runtime_error(error_message);
    }
  }
}
}  // namespace dbhandler

int32_t DBHandler::create_dashboard(const TSessionId& session_id_or_json,
                                    const std::string& dashboard_name,
                                    const std::string& dashboard_state,
                                    const std::string& image_hash,
                                    const std::string& dashboard_metadata) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  CHECK(session_ptr);
  check_read_only("create_dashboard");
  auto& cat = session_ptr->getCatalog();
  if (!g_allow_system_dashboard_update) {
    dbhandler::check_not_info_schema_db(cat.name(), true);
  }

  if (!session_ptr->checkDBAccessPrivileges(DBObjectType::DashboardDBObjectType,
                                            AccessPrivileges::CREATE_DASHBOARD)) {
    THROW_DB_EXCEPTION("Not enough privileges to create a dashboard.");
  }

  if (dashboard_exists(cat, session_ptr->get_currentUser().userId, dashboard_name)) {
    THROW_DB_EXCEPTION("Dashboard with name: " + dashboard_name + " already exists.");
  }

  DashboardDescriptor dd;
  dd.dashboardName = dashboard_name;
  dd.dashboardState = dashboard_state;
  dd.imageHash = image_hash;
  dd.dashboardMetadata = dashboard_metadata;
  dd.userId = session_ptr->get_currentUser().userId;
  dd.user = session_ptr->get_currentUser().userName;

  try {
    auto id = cat.createDashboard(dd);
    // TODO: transactionally unsafe
    SysCatalog::instance().createDBObject(
        session_ptr->get_currentUser(), dashboard_name, DashboardDBObjectType, cat, id);
    return id;
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::replace_dashboard(const TSessionId& session_id_or_json,
                                  const int32_t dashboard_id,
                                  const std::string& dashboard_name,
                                  const std::string& dashboard_owner,
                                  const std::string& dashboard_state,
                                  const std::string& image_hash,
                                  const std::string& dashboard_metadata) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  CHECK(session_ptr);
  check_read_only("replace_dashboard");
  auto& cat = session_ptr->getCatalog();
  if (!g_allow_system_dashboard_update) {
    dbhandler::check_not_info_schema_db(cat.name(), true);
  }

  if (!is_allowed_on_dashboard(
          *session_ptr, dashboard_id, AccessPrivileges::EDIT_DASHBOARD)) {
    THROW_DB_EXCEPTION("Not enough privileges to replace a dashboard.");
  }

  if (auto dash = cat.getMetadataForDashboard(
          std::to_string(session_ptr->get_currentUser().userId), dashboard_name)) {
    if (dash->dashboardId != dashboard_id) {
      THROW_DB_EXCEPTION("Dashboard with name: " + dashboard_name + " already exists.");
    }
  }

  DashboardDescriptor dd;
  dd.dashboardName = dashboard_name;
  dd.dashboardState = dashboard_state;
  dd.imageHash = image_hash;
  dd.dashboardMetadata = dashboard_metadata;
  Catalog_Namespace::UserMetadata user;
  if (!SysCatalog::instance().getMetadataForUser(dashboard_owner, user)) {
    THROW_DB_EXCEPTION(std::string("Dashboard owner ") + dashboard_owner +
                       " does not exist");
  }
  dd.userId = user.userId;
  dd.user = dashboard_owner;
  dd.dashboardId = dashboard_id;

  try {
    cat.replaceDashboard(dd);
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

void DBHandler::delete_dashboard(const TSessionId& session_id_or_json,
                                 const int32_t dashboard_id) {
  delete_dashboards(session_id_or_json, {dashboard_id});
}

void DBHandler::delete_dashboards(const TSessionId& session_id_or_json,
                                  const std::vector<int32_t>& dashboard_ids) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  check_read_only("delete_dashboards");
  auto& cat = session_ptr->getCatalog();
  if (!g_allow_system_dashboard_update) {
    dbhandler::check_not_info_schema_db(cat.name(), true);
  }
  // Checks will be performed in catalog
  try {
    cat.deleteMetadataForDashboards(dashboard_ids, session_ptr->get_currentUser());
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

std::vector<std::string> DBHandler::get_valid_groups(const TSessionId& session_id_or_json,
                                                     int32_t dashboard_id,
                                                     std::vector<std::string> groups) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  const auto session_info = get_session_copy(request_info.sessionId());
  auto& cat = session_info.getCatalog();
  auto dash = cat.getMetadataForDashboard(dashboard_id);
  if (!dash) {
    THROW_DB_EXCEPTION("Dashboard id " + std::to_string(dashboard_id) +
                       " does not exist");
  } else if (session_info.get_currentUser().userId != dash->userId &&
             !session_info.get_currentUser().isSuper) {
    throw std::runtime_error(
        "User should be either owner of dashboard or super user to share/unshare it");
  }
  std::vector<std::string> valid_groups;
  Catalog_Namespace::UserMetadata user_meta;
  for (auto& group : groups) {
    user_meta.isSuper = false;  // initialize default flag
    if (!SysCatalog::instance().getGrantee(group)) {
      THROW_DB_EXCEPTION("User/Role " + group + " does not exist");
    } else if (!user_meta.isSuper) {
      valid_groups.push_back(group);
    }
  }
  return valid_groups;
}

void DBHandler::validateGroups(const std::vector<std::string>& groups) {
  for (auto const& group : groups) {
    if (!SysCatalog::instance().getGrantee(group)) {
      THROW_DB_EXCEPTION("User/Role '" + group + "' does not exist");
    }
  }
}

void DBHandler::validateDashboardIdsForSharing(
    const Catalog_Namespace::SessionInfo& session_info,
    const std::vector<int32_t>& dashboard_ids) {
  auto& cat = session_info.getCatalog();
  std::map<std::string, std::list<int32_t>> errors;
  for (auto const& dashboard_id : dashboard_ids) {
    auto dashboard = cat.getMetadataForDashboard(dashboard_id);
    if (!dashboard) {
      errors["Dashboard id does not exist"].push_back(dashboard_id);
    } else if (session_info.get_currentUser().userId != dashboard->userId &&
               !session_info.get_currentUser().isSuper) {
      errors["User should be either owner of dashboard or super user to share/unshare it"]
          .push_back(dashboard_id);
    }
  }
  if (!errors.empty()) {
    std::stringstream error_stream;
    error_stream << "Share/Unshare dashboard(s) failed with error(s)\n";
    for (const auto& [error, id_list] : errors) {
      error_stream << "Dashboard ids " << join(id_list, ", ") << ": " << error << "\n";
    }
    THROW_DB_EXCEPTION(error_stream.str());
  }
}

void DBHandler::shareOrUnshareDashboards(const TSessionId& session_id,
                                         const std::vector<int32_t>& dashboard_ids,
                                         const std::vector<std::string>& groups,
                                         const TDashboardPermissions& permissions,
                                         const bool do_share) {
  auto stdlog = STDLOG(get_session_ptr(session_id));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  check_read_only(do_share ? "share_dashboards" : "unshare_dashboards");
  if (!permissions.create_ && !permissions.delete_ && !permissions.edit_ &&
      !permissions.view_) {
    THROW_DB_EXCEPTION("At least one privilege should be assigned for " +
                       std::string(do_share ? "grants" : "revokes"));
  }
  auto session_ptr = stdlog.getConstSessionInfo();
  auto const& catalog = session_ptr->getCatalog();
  auto& sys_catalog = SysCatalog::instance();
  validateGroups(groups);
  validateDashboardIdsForSharing(*session_ptr, dashboard_ids);
  std::vector<DBObject> batch_objects;
  for (auto const& dashboard_id : dashboard_ids) {
    DBObject object(dashboard_id, DBObjectType::DashboardDBObjectType);
    AccessPrivileges privs;
    if (permissions.delete_) {
      privs.add(AccessPrivileges::DELETE_DASHBOARD);
    }
    if (permissions.create_) {
      privs.add(AccessPrivileges::CREATE_DASHBOARD);
    }
    if (permissions.edit_) {
      privs.add(AccessPrivileges::EDIT_DASHBOARD);
    }
    if (permissions.view_) {
      privs.add(AccessPrivileges::VIEW_DASHBOARD);
    }
    object.setPrivileges(privs);
    batch_objects.push_back(object);
  }
  if (do_share) {
    sys_catalog.grantDBObjectPrivilegesBatch(groups, batch_objects, catalog);
  } else {
    sys_catalog.revokeDBObjectPrivilegesBatch(groups, batch_objects, catalog);
  }
}

void DBHandler::share_dashboards(const TSessionId& session_id_or_json,
                                 const std::vector<int32_t>& dashboard_ids,
                                 const std::vector<std::string>& groups,
                                 const TDashboardPermissions& permissions) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  shareOrUnshareDashboards(
      request_info.sessionId(), dashboard_ids, groups, permissions, true);
}

// NOOP: Grants not available for objects as of now
void DBHandler::share_dashboard(const TSessionId& session_id_or_json,
                                const int32_t dashboard_id,
                                const std::vector<std::string>& groups,
                                const std::vector<std::string>& objects,
                                const TDashboardPermissions& permissions,
                                const bool grant_role = false) {
  share_dashboards(session_id_or_json, {dashboard_id}, groups, permissions);
}

void DBHandler::unshare_dashboards(const TSessionId& session_id_or_json,
                                   const std::vector<int32_t>& dashboard_ids,
                                   const std::vector<std::string>& groups,
                                   const TDashboardPermissions& permissions) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  shareOrUnshareDashboards(
      request_info.sessionId(), dashboard_ids, groups, permissions, false);
}

void DBHandler::unshare_dashboard(const TSessionId& session_id_or_json,
                                  const int32_t dashboard_id,
                                  const std::vector<std::string>& groups,
                                  const std::vector<std::string>& objects,
                                  const TDashboardPermissions& permissions) {
  unshare_dashboards(session_id_or_json, {dashboard_id}, groups, permissions);
}

void DBHandler::get_dashboard_grantees(
    std::vector<TDashboardGrantees>& dashboard_grantees,
    const TSessionId& session_id_or_json,
    const int32_t dashboard_id) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  auto const& cat = session_ptr->getCatalog();
  Catalog_Namespace::UserMetadata user_meta;
  auto dash = cat.getMetadataForDashboard(dashboard_id);
  if (!dash) {
    THROW_DB_EXCEPTION("Dashboard id " + std::to_string(dashboard_id) +
                       " does not exist");
  } else if (session_ptr->get_currentUser().userId != dash->userId &&
             !session_ptr->get_currentUser().isSuper) {
    THROW_DB_EXCEPTION(
        "User should be either owner of dashboard or super user to access grantees");
  }
  std::vector<ObjectRoleDescriptor*> objectsList;
  objectsList = SysCatalog::instance().getMetadataForObject(
      cat.getCurrentDB().dbId,
      static_cast<int>(DBObjectType::DashboardDBObjectType),
      dashboard_id);  // By default objecttypecan be only dashabaords
  user_meta.userId = -1;
  user_meta.userName = "";
  SysCatalog::instance().getMetadataForUserById(dash->userId, user_meta);
  for (auto object : objectsList) {
    if (user_meta.userName == object->roleName) {
      // Mask owner
      continue;
    }
    TDashboardGrantees grantee;
    TDashboardPermissions perm;
    grantee.name = object->roleName;
    grantee.is_user = object->roleType;
    perm.create_ = object->privs.hasPermission(DashboardPrivileges::CREATE_DASHBOARD);
    perm.delete_ = object->privs.hasPermission(DashboardPrivileges::DELETE_DASHBOARD);
    perm.edit_ = object->privs.hasPermission(DashboardPrivileges::EDIT_DASHBOARD);
    perm.view_ = object->privs.hasPermission(DashboardPrivileges::VIEW_DASHBOARD);
    grantee.permissions = perm;
    dashboard_grantees.push_back(grantee);
  }
}

void DBHandler::create_link(std::string& _return,
                            const TSessionId& session_id_or_json,
                            const std::string& view_state,
                            const std::string& view_metadata) {
  heavyai::RequestInfo const request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG(get_session_ptr(request_info.sessionId()));
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  auto session_ptr = stdlog.getConstSessionInfo();
  // check_read_only("create_link");
  auto& cat = session_ptr->getCatalog();

  LinkDescriptor ld;
  ld.userId = session_ptr->get_currentUser().userId;
  ld.viewState = view_state;
  ld.viewMetadata = view_metadata;

  try {
    _return = cat.createLink(ld, 6);
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(e.what());
  }
}

TColumnType DBHandler::create_geo_column(const TDatumType::type type,
                                         const std::string& name,
                                         const bool is_array) {
  TColumnType ct;
  ct.col_name = name;
  ct.col_type.type = type;
  ct.col_type.is_array = is_array;
  return ct;
}

void DBHandler::check_geospatial_files(const boost::filesystem::path file_path,
                                       const import_export::CopyParams& copy_params) {
  const std::list<std::string> shp_ext{".shp", ".shx", ".dbf"};
  if (std::find(shp_ext.begin(),
                shp_ext.end(),
                boost::algorithm::to_lower_copy(file_path.extension().string())) !=
      shp_ext.end()) {
    for (auto ext : shp_ext) {
      auto aux_file = file_path;
      if (!import_export::Importer::gdalFileExists(
              aux_file.replace_extension(boost::algorithm::to_upper_copy(ext)).string(),
              copy_params) &&
          !import_export::Importer::gdalFileExists(
              aux_file.replace_extension(ext).string(), copy_params)) {
        throw std::runtime_error("required file for shapefile does not exist: " +
                                 aux_file.filename().string());
      }
    }
  }
}

void DBHandler::create_table(const TSessionId& session_id_or_json,
                             const std::string& table_name,
                             const TRowDescriptor& rd,
                             const TCreateParams& create_params) {
  heavyai::RequestInfo request_info(session_id_or_json);
  SET_REQUEST_ID(request_info.requestId());
  auto stdlog = STDLOG("table_name", table_name);
  stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
  check_read_only("create_table");

  if (ImportHelpers::is_reserved_name(table_name)) {
    THROW_DB_EXCEPTION("Invalid table name (reserved keyword): " + table_name);
  } else if (table_name != ImportHelpers::sanitize_name(table_name)) {
    THROW_DB_EXCEPTION("Invalid characters in table name: " + table_name);
  }

  auto rds = rd;

  std::string stmt{"CREATE TABLE " + table_name};
  std::vector<std::string> col_stmts;

  for (auto col : rds) {
    if (ImportHelpers::is_reserved_name(col.col_name)) {
      THROW_DB_EXCEPTION("Invalid column name (reserved keyword): " + col.col_name);
    } else if (col.col_name != ImportHelpers::sanitize_name(col.col_name)) {
      THROW_DB_EXCEPTION("Invalid characters in column name: " + col.col_name);
    }
    if (col.col_type.type == TDatumType::INTERVAL_DAY_TIME ||
        col.col_type.type == TDatumType::INTERVAL_YEAR_MONTH) {
      THROW_DB_EXCEPTION("Unsupported type: " + thrift_to_name(col.col_type) +
                         " for column: " + col.col_name);
    }

    if (col.col_type.type == TDatumType::DECIMAL) {
      // if no precision or scale passed in set to default 14,7
      if (col.col_type.precision == 0 && col.col_type.scale == 0) {
        col.col_type.precision = 14;
        col.col_type.scale = 7;
      }
    }

    std::string col_stmt;
    col_stmt.append(col.col_name + " " + thrift_to_name(col.col_type));
    if (col.__isset.default_value) {
      col_stmt.append(" DEFAULT " + col.default_value);
    }

    // As of 2016-06-27 the Immerse v1 frontend does not explicitly set the
    // `nullable` argument, leading this to default to false. Uncomment for v2.
    // if (!col.col_type.nullable) col_stmt.append(" NOT NULL");

    if (thrift_to_encoding(col.col_type.encoding) != kENCODING_NONE) {
      col_stmt.append(" ENCODING " + thrift_to_encoding_name(col.col_type));
      if (thrift_to_encoding(col.col_type.encoding) == kENCODING_DICT ||
          thrift_to_encoding(col.col_type.encoding) == kENCODING_FIXED ||
          thrift_to_encoding(col.col_type.encoding) == kENCODING_GEOINT ||
          thrift_to_encoding(col.col_type.encoding) == kENCODING_DATE_IN_DAYS) {
        col_stmt.append("(" + std::to_string(col.col_type.comp_param) + ")");
      }
    } else if (col.col_type.type == TDatumType::STR) {
      // non DICT encoded strings
      col_stmt.append(" ENCODING NONE");
    } else if (col.col_type.type == TDatumType::POINT ||
               col.col_type.type == TDatumType::MULTIPOINT ||
               col.col_type.type == TDatumType::LINESTRING ||
               col.col_type.type == TDatumType::MULTILINESTRING ||
               col.col_type.type == TDatumType::POLYGON ||
               col.col_type.type == TDatumType::MULTIPOLYGON) {
      // non encoded compressable geo
      if (col.col_type.scale == 4326) {
        col_stmt.append(" ENCODING NONE");
      }
    }
    col_stmts.push_back(col_stmt);
  }

  stmt.append(" (" + boost::algorithm::join(col_stmts, ", ") + ")");

  if (create_params.is_replicated) {
    stmt.append(" WITH (PARTITIONS = 'REPLICATED')");
  }

  stmt.append(";");

  TQueryResult ret;
  request_info.setRequestId(logger::request_id());
  sql_execute(ret, request_info.json(), stmt, true, "", -1, -1);
}

void DBHandler::import_table(const TSessionId& session_id_or_json,
                             const std::string& table_name,
                             const std::string& file_name_in,
                             const TCopyParams& cp) {
  try {
    heavyai::RequestInfo const request_info(session_id_or_json);
    SET_REQUEST_ID(request_info.requestId());
    auto stdlog =
        STDLOG(get_session_ptr(request_info.sessionId()), "table_name", table_name);
    stdlog.appendNameValuePairs("client", getConnectionInfo().toString());
    auto session_ptr = stdlog.getConstSessionInfo();
    check_read_only("import_table");
    LOG(INFO) << "import_table " << table_name << " from " << file_name_in;

    const auto execute_read_lock = legacylockmgr::getExecuteReadLock();
    auto& cat = session_ptr->getCatalog();
    auto executor = Executor::getExecutor(Executor::UNITARY_EXECUTOR_ID);
    auto start_time = ::toString(std::chrono::system_clock::now());
    if (g_enable_non_kernel_time_query_interrupt) {
      executor->enrollQuerySession(request_info.sessionId(),
                                   "IMPORT_TABLE",
                                   start_time,
                                   Executor::UNITARY_EXECUTOR_ID,
                                   QuerySessionStatus::QueryStatus::RUNNING_IMPORTER);
    }

    ScopeGuard clearInterruptStatus = [executor, &request_info, &start_time] {
      // reset the runtime query interrupt status
      if (g_enable_non_kernel_time_query_interrupt) {
        executor->clearQuerySessionStatus(request_info.sessionId(), start_time);
      }
    };
    const auto td_with_lock =
        lockmgr::TableSchemaLockContainer<lockmgr::ReadLock>::acquireTableDescriptor(
            cat, table_name);
    const auto td = td_with_lock();
    CHECK(td);
    check_table_load_privileges(*session_ptr, table_name);

    std::string copy_from_source;
    import_export::CopyParams copy_params = thrift_to_copyparams(cp);
    if (copy_params.source_type == import_export::SourceType::kOdbc) {
      copy_from_source = copy_params.sql_select;
    } else {
      std::string file_name{file_name_in};
      auto file_path = boost::filesystem::path(file_name);
      if (!boost::istarts_with(file_name, "s3://")) {
        if (!boost::filesystem::path(file_name).is_absolute()) {
          file_path = import_path_ /
                      picosha2::hash256_hex_string(request_info.sessionId()) /
                      boost::filesystem::path(file_name).filename();
          file_name = file_path.string();
        }
        if (!shared::file_or_glob_path_exists(file_path.string())) {
          THROW_DB_EXCEPTION("File or directory \"" + file_path.string() +
                             "\" does not exist.");
        }
      }
      validate_import_file_path_if_local(file_name);

      // TODO(andrew): add delimiter detection to Importer
      if (copy_params.delimiter == '\0') {
        copy_params.delimiter = ',';
        if (boost::filesystem::path(file_path).extension() == ".tsv") {
          copy_params.delimiter = '\t';
        }
      }
      copy_from_source = file_path.string();
    }
    auto const load_tag = get_import_tag("import_table", table_name, copy_from_source);
    log_system_cpu_memory_status("start_" + load_tag, session_ptr->getCatalog());
    ScopeGuard cleanup = [&load_tag, &session_ptr]() {
      log_system_cpu_memory_status("finish_" + load_tag, session_ptr->getCatalog());
    };
    const auto insert_data_lock = lockmgr::InsertDataLockMgr::getWriteLockForTable(
        session_ptr->getCatalog(), table_name);
    std::unique_ptr<import_export::AbstractImporter> importer;
    importer = import_export::create_importer(cat, td, copy_from_source, copy_params);
    auto ms = measure<>::execution([&]() { importer->import(session_ptr.get()); });
    LOG(INFO) << "Total Import Time: " << (double)ms / 1000.0 << " Seconds.";
  } catch (const TDBException& e) {
    throw;
  } catch (const std::exception& e) {
    THROW_DB_EXCEPTION(std::string(e.what()));
  }
}

namespace {

// helper functions for error checking below
// these would usefully be added as methods of TDatumType
// but that's not possible as it's auto-generated by Thrift

bool TTypeInfo_IsGeo</