OmniSciDB  cde582ebc3
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Groups Pages
QueryRunner.cpp
Go to the documentation of this file.
1 /*
2  * Copyright 2022 HEAVY.AI, Inc.
3  *
4  * Licensed under the Apache License, Version 2.0 (the "License");
5  * you may not use this file except in compliance with the License.
6  * You may obtain a copy of the License at
7  *
8  * http://www.apache.org/licenses/LICENSE-2.0
9  *
10  * Unless required by applicable law or agreed to in writing, software
11  * distributed under the License is distributed on an "AS IS" BASIS,
12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13  * See the License for the specific language governing permissions and
14  * limitations under the License.
15  */
16 
17 #include "QueryRunner.h"
18 
19 #include "Calcite/Calcite.h"
20 #include "Catalog/Catalog.h"
22 #include "DistributedLoader.h"
23 #include "Geospatial/ColumnNames.h"
25 #include "Logger/Logger.h"
26 #include "Parser/ParserNode.h"
27 #include "Parser/ParserWrapper.h"
36 #include "Shared/StringTransform.h"
37 #include "Shared/SysDefinitions.h"
39 #include "Shared/import_helpers.h"
41 #include "gen-cpp/CalciteServer.h"
42 #include "include/bcrypt.h"
43 
44 #include <boost/filesystem/operations.hpp>
45 #include <csignal>
46 #include <random>
47 
48 #define CALCITEPORT 3279
49 
50 extern size_t g_leaf_count;
51 extern bool g_enable_filter_push_down;
52 
54 
55 extern bool g_serialize_temp_tables;
57 std::mutex calcite_lock;
58 
59 using namespace Catalog_Namespace;
60 namespace {
61 
62 std::shared_ptr<Calcite> g_calcite = nullptr;
63 
64 void calcite_shutdown_handler() noexcept {
65  if (g_calcite) {
66  g_calcite->close_calcite_server();
67  g_calcite.reset();
68  }
69 }
70 
74 }
75 
76 } // namespace
77 
78 namespace QueryRunner {
79 
80 std::unique_ptr<QueryRunner> QueryRunner::qr_instance_ = nullptr;
81 
82 query_state::QueryStates QueryRunner::query_states_;
83 
84 QueryRunner* QueryRunner::init(const char* db_path,
85  const std::string& udf_filename,
86  const size_t max_gpu_mem,
87  const int reserved_gpu_mem) {
88  return QueryRunner::init(db_path,
90  "HyperInteractive",
92  {},
93  {},
94  udf_filename,
95  true,
96  max_gpu_mem,
97  reserved_gpu_mem);
98 }
99 
101  const char* db_path,
102  const std::vector<LeafHostInfo>& string_servers,
103  const std::vector<LeafHostInfo>& leaf_servers) {
104  return QueryRunner::init(db_path,
106  "HyperInteractive",
108  string_servers,
109  leaf_servers,
110  "",
111  true,
112  0,
113  256 << 20,
114  false,
115  false,
116  disk_cache_config);
117 }
118 
119 QueryRunner* QueryRunner::init(const char* db_path,
120  const std::string& user,
121  const std::string& pass,
122  const std::string& db_name,
123  const std::vector<LeafHostInfo>& string_servers,
124  const std::vector<LeafHostInfo>& leaf_servers,
125  const std::string& udf_filename,
126  bool uses_gpus,
127  const size_t max_gpu_mem,
128  const int reserved_gpu_mem,
129  const bool create_user,
130  const bool create_db,
131  const File_Namespace::DiskCacheConfig* disk_cache_config) {
132  // Whitelist root path for tests by default
134  ddl_utils::FilePathWhitelist::initialize(db_path, "[\"/\"]", "[\"/\"]");
135  LOG_IF(FATAL, !leaf_servers.empty()) << "Distributed test runner not supported.";
136  CHECK(leaf_servers.empty());
137  qr_instance_.reset(new QueryRunner(db_path,
138  user,
139  pass,
140  db_name,
141  string_servers,
142  leaf_servers,
143  udf_filename,
144  uses_gpus,
145  max_gpu_mem,
146  reserved_gpu_mem,
147  create_user,
148  create_db,
149  disk_cache_config));
150  return qr_instance_.get();
151 }
152 
153 QueryRunner::QueryRunner(const char* db_path,
154  const std::string& user_name,
155  const std::string& passwd,
156  const std::string& db_name,
157  const std::vector<LeafHostInfo>& string_servers,
158  const std::vector<LeafHostInfo>& leaf_servers,
159  const std::string& udf_filename,
160  bool uses_gpus,
161  const size_t max_gpu_mem,
162  const int reserved_gpu_mem,
163  const bool create_user,
164  const bool create_db,
165  const File_Namespace::DiskCacheConfig* cache_config)
166  : dispatch_queue_(std::make_unique<QueryDispatchQueue>(1)) {
168  boost::filesystem::path base_path{db_path};
169  CHECK(boost::filesystem::exists(base_path));
170  auto system_db_file =
172  CHECK(boost::filesystem::exists(system_db_file));
173  auto data_dir = base_path / shared::kDataDirectoryName;
174  File_Namespace::DiskCacheConfig disk_cache_config{
175  (base_path / shared::kDefaultDiskCacheDirName).string(),
177  if (cache_config) {
178  disk_cache_config = *cache_config;
179  }
181 
184  g_calcite =
185  std::make_shared<Calcite>(-1, CALCITEPORT, db_path, 1024, 5000, true, udf_filename);
186  ExtensionFunctionsWhitelist::add(g_calcite->getExtensionFunctionWhitelist());
187  if (!udf_filename.empty()) {
188  ExtensionFunctionsWhitelist::addUdfs(g_calcite->getUserDefinedFunctionWhitelist());
189  }
190 
192  auto udtfs = ThriftSerializers::to_thrift(
194  std::vector<TUserDefinedFunction> udfs = {};
195  g_calcite->setRuntimeExtensionFunctions(udfs, udtfs, /*is_runtime=*/false);
196 
197  std::unique_ptr<CudaMgr_Namespace::CudaMgr> cuda_mgr;
198 #ifdef HAVE_CUDA
199  if (uses_gpus) {
200  cuda_mgr = std::make_unique<CudaMgr_Namespace::CudaMgr>(-1, 0);
201  }
202 #else
203  uses_gpus = false;
204 #endif
205  SystemParameters mapd_params;
206  mapd_params.gpu_buffer_mem_bytes = max_gpu_mem;
207  mapd_params.aggregator = !leaf_servers.empty();
208 
209  data_mgr_.reset(new Data_Namespace::DataMgr(data_dir.string(),
210  mapd_params,
211  std::move(cuda_mgr),
212  uses_gpus,
213  reserved_gpu_mem,
214  0,
215  disk_cache_config));
216 
217  auto& sys_cat = Catalog_Namespace::SysCatalog::instance();
218 
219  g_base_path = base_path.string();
220  sys_cat.init(g_base_path,
221  data_mgr_,
222  {},
223  g_calcite,
224  false,
225  mapd_params.aggregator,
226  string_servers);
227 
228  query_engine_ = QueryEngine::createInstance(data_mgr_->getCudaMgr(), !uses_gpus);
229 
230  if (create_user) {
231  if (!sys_cat.getMetadataForUser(user_name, user)) {
232  sys_cat.createUser(
233  user_name,
235  passwd, /*is_super=*/false, /*default_db=*/"", /*can_login=*/true},
236  g_read_only);
237  }
238  }
239  CHECK(sys_cat.getMetadataForUser(user_name, user));
240  CHECK(bcrypt_checkpw(passwd.c_str(), user.passwd_hash.c_str()) == 0);
241 
242  if (create_db) {
243  if (!sys_cat.getMetadataForDB(db_name, db_metadata_)) {
244  sys_cat.createDatabase(db_name, user.userId);
245  }
246  }
247  CHECK(sys_cat.getMetadataForDB(db_name, db_metadata_));
248  CHECK(user.isSuper || (user.userId == db_metadata_.dbOwner));
249  auto cat = sys_cat.getCatalog(db_metadata_, create_db);
250  CHECK(cat);
251  session_info_ = std::make_unique<Catalog_Namespace::SessionInfo>(
252  cat, user, ExecutorDeviceType::GPU, "");
253 }
254 
255 void QueryRunner::resizeDispatchQueue(const size_t num_executors) {
256  dispatch_queue_ = std::make_unique<QueryDispatchQueue>(num_executors);
257 }
258 
259 QueryRunner::QueryRunner(std::unique_ptr<Catalog_Namespace::SessionInfo> session)
260  : session_info_(std::move(session))
261  , dispatch_queue_(std::make_unique<QueryDispatchQueue>(1)) {}
262 
263 std::shared_ptr<Catalog_Namespace::Catalog> QueryRunner::getCatalog() const {
265  return session_info_->get_catalog_ptr();
266 }
267 
268 std::shared_ptr<Calcite> QueryRunner::getCalcite() const {
269  // TODO: Embed Calcite shared_ptr ownership in QueryRunner
270  return g_calcite;
271 }
272 
273 bool QueryRunner::gpusPresent() const {
275  return session_info_->getCatalog().getDataMgr().gpusPresent();
276 }
277 
278 void QueryRunner::clearGpuMemory() const {
281 }
282 
283 void QueryRunner::clearCpuMemory() const {
286 }
287 
288 std::vector<MemoryInfo> QueryRunner::getMemoryInfo(
289  const Data_Namespace::MemoryLevel memory_level) const {
291  return session_info_->getCatalog().getDataMgr().getMemoryInfo(memory_level);
292 }
293 
294 BufferPoolStats QueryRunner::getBufferPoolStats(
295  const Data_Namespace::MemoryLevel memory_level,
296  const bool current_db_only) const {
297  // Only works single-node for now
299  const std::vector<MemoryInfo> memory_infos =
300  session_info_->getCatalog().getDataMgr().getMemoryInfo(memory_level);
301  if (memory_level == Data_Namespace::MemoryLevel::CPU_LEVEL) {
302  CHECK_EQ(memory_infos.size(), static_cast<size_t>(1));
303  }
304  std::set<std::vector<int32_t>> chunk_keys;
305  std::set<std::vector<int32_t>> table_keys;
306  std::set<std::vector<int32_t>> column_keys;
307  std::set<std::vector<int32_t>> fragment_keys;
308  size_t total_num_buffers{
309  0}; // can be greater than chunk keys set size due to table replication
310  size_t total_num_bytes{0};
311  for (auto& pool_memory_info : memory_infos) {
312  const std::vector<MemoryData>& memory_data = pool_memory_info.nodeMemoryData;
313  for (auto& memory_datum : memory_data) {
314  total_num_buffers++;
315  const auto& chunk_key = memory_datum.chunk_key;
316  if (memory_datum.memStatus == Buffer_Namespace::MemStatus::FREE ||
317  chunk_key.size() < 4) {
318  continue;
319  }
320  if (current_db_only) {
321  if (chunk_key[0] != db_metadata_.dbId) {
322  continue;
323  }
324  }
325  total_num_bytes += (memory_datum.numPages * pool_memory_info.pageSize);
326  table_keys.insert({chunk_key[0], chunk_key[1]});
327  column_keys.insert({chunk_key[0], chunk_key[1], chunk_key[2]});
328  fragment_keys.insert({chunk_key[0], chunk_key[1], chunk_key[3]});
329  chunk_keys.insert(chunk_key);
330  }
331  }
332  return {total_num_buffers,
333  total_num_bytes,
334  table_keys.size(),
335  column_keys.size(),
336  fragment_keys.size(),
337  chunk_keys.size()};
338 }
339 
340 RegisteredQueryHint QueryRunner::getParsedQueryHint(const std::string& query_str) {
343  auto query_state = create_query_state(session_info_, query_str);
344  auto& cat = session_info_->getCatalog();
346 
347  auto calcite_mgr = cat.getCalciteMgr();
348  const auto calciteQueryParsingOption =
349  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
350  const auto calciteOptimizationOption =
351  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
352  const auto query_ra = calcite_mgr
353  ->process(query_state->createQueryStateProxy(),
354  pg_shim(query_str),
355  calciteQueryParsingOption,
356  calciteOptimizationOption)
357  .plan_result;
358  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra, query_state);
359  auto query_hints =
360  ra_executor.getParsedQueryHint(ra_executor.getRootRelAlgNodeShPtr().get());
361  return query_hints ? *query_hints : RegisteredQueryHint::defaults();
362 }
363 
364 std::shared_ptr<const RelAlgNode> QueryRunner::getRootNodeFromParsedQuery(
365  const std::string& query_str) {
368  auto query_state = create_query_state(session_info_, query_str);
369  auto& cat = session_info_->getCatalog();
371 
372  auto calcite_mgr = cat.getCalciteMgr();
373  const auto calciteQueryParsingOption =
374  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
375  const auto calciteOptimizationOption =
376  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
377  const auto query_ra = calcite_mgr
378  ->process(query_state->createQueryStateProxy(),
379  pg_shim(query_str),
380  calciteQueryParsingOption,
381  calciteOptimizationOption)
382  .plan_result;
383  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra, query_state);
384  return ra_executor.getRootRelAlgNodeShPtr();
385 }
386 
388  std::unordered_map<size_t, std::unordered_map<unsigned, RegisteredQueryHint>>>
389 QueryRunner::getParsedQueryHints(const std::string& query_str) {
392  auto query_state = create_query_state(session_info_, query_str);
393  auto& cat = session_info_->getCatalog();
395  auto calcite_mgr = cat.getCalciteMgr();
396  const auto calciteQueryParsingOption =
397  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
398  const auto calciteOptimizationOption =
399  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
400  const auto query_ra = calcite_mgr
401  ->process(query_state->createQueryStateProxy(),
402  pg_shim(query_str),
403  calciteQueryParsingOption,
404  calciteOptimizationOption)
405  .plan_result;
406  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra, query_state);
407  return ra_executor.getParsedQueryHints();
408 }
409 
410 std::optional<RegisteredQueryHint> QueryRunner::getParsedGlobalQueryHints(
411  const std::string& query_str) {
414  auto query_state = create_query_state(session_info_, query_str);
415  auto& cat = session_info_->getCatalog();
417  auto calcite_mgr = cat.getCalciteMgr();
418  const auto calciteQueryParsingOption =
419  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
420  const auto calciteOptimizationOption =
421  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
422  const auto query_ra = calcite_mgr
423  ->process(query_state->createQueryStateProxy(),
424  pg_shim(query_str),
425  calciteQueryParsingOption,
426  calciteOptimizationOption)
427  .plan_result;
428  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra, query_state);
429  return ra_executor.getGlobalQueryHint();
430 }
431 
432 RaExecutionSequence QueryRunner::getRaExecutionSequence(const std::string& query_str) {
435  auto query_state = create_query_state(session_info_, query_str);
436  auto& cat = session_info_->getCatalog();
438  auto calcite_mgr = cat.getCalciteMgr();
439  const auto calciteQueryParsingOption =
440  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
441  const auto calciteOptimizationOption =
442  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
443  const auto query_ra = calcite_mgr
444  ->process(query_state->createQueryStateProxy(),
445  pg_shim(query_str),
446  calciteQueryParsingOption,
447  calciteOptimizationOption)
448  .plan_result;
449  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra, query_state);
450  return ra_executor.getRaExecutionSequence(ra_executor.getRootRelAlgNodeShPtr().get(),
451  executor.get());
452 }
453 
454 // used to validate calcite ddl statements
455 void QueryRunner::validateDDLStatement(const std::string& stmt_str_in) {
457 
458  std::string stmt_str = stmt_str_in;
459  // First remove special chars
460  boost::algorithm::trim_left_if(stmt_str, boost::algorithm::is_any_of("\n"));
461  // Then remove spaces
462  boost::algorithm::trim_left(stmt_str);
463 
464  auto query_state = create_query_state(session_info_, stmt_str);
465  auto stdlog = STDLOG(query_state);
466 
467  auto& cat = session_info_->getCatalog();
468  auto calcite_mgr = cat.getCalciteMgr();
469  const auto calciteQueryParsingOption =
470  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
471  const auto calciteOptimizationOption =
472  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
473  calcite_mgr->process(query_state->createQueryStateProxy(),
474  pg_shim(stmt_str),
475  calciteQueryParsingOption,
476  calciteOptimizationOption);
477 }
478 
479 std::shared_ptr<RelAlgTranslator> QueryRunner::getRelAlgTranslator(
480  const std::string& query_str,
481  Executor* executor) {
484  auto query_state = create_query_state(session_info_, query_str);
485  auto& cat = session_info_->getCatalog();
486  auto calcite_mgr = cat.getCalciteMgr();
487  const auto calciteQueryParsingOption =
488  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
489  const auto calciteOptimizationOption =
490  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
491  const auto query_ra = calcite_mgr
492  ->process(query_state->createQueryStateProxy(),
493  pg_shim(query_str),
494  calciteQueryParsingOption,
495  calciteOptimizationOption)
496  .plan_result;
497  executor->setCatalog(&cat);
498  auto ra_executor = RelAlgExecutor(executor, cat, query_ra);
499  auto root_node_shared_ptr = ra_executor.getRootRelAlgNodeShPtr();
500  return ra_executor.getRelAlgTranslator(root_node_shared_ptr.get());
501 }
502 
503 QueryPlanDagInfo QueryRunner::getQueryInfoForDataRecyclerTest(
504  const std::string& query_str) {
507  auto query_state = create_query_state(session_info_, query_str);
508  auto& cat = session_info_->getCatalog();
510  auto calcite_mgr = cat.getCalciteMgr();
511  const auto calciteQueryParsingOption =
512  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
513  const auto calciteOptimizationOption =
514  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
515  const auto query_ra = calcite_mgr
516  ->process(query_state->createQueryStateProxy(),
517  pg_shim(query_str),
518  calciteQueryParsingOption,
519  calciteOptimizationOption)
520  .plan_result;
521  executor->setCatalog(&cat);
522  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra);
523  // note that we assume the test for data recycler that needs to have join_info
524  // does not contain any ORDER BY clause; this is necessary to create work_unit
525  // without actually performing the query
526  auto root_node_shared_ptr = ra_executor.getRootRelAlgNodeShPtr();
527  auto join_info = ra_executor.getJoinInfo(root_node_shared_ptr.get());
528  auto relAlgTranslator = ra_executor.getRelAlgTranslator(root_node_shared_ptr.get());
529  return {root_node_shared_ptr, join_info.first, join_info.second, relAlgTranslator};
530 }
531 
532 std::unique_ptr<Parser::Stmt> QueryRunner::createStatement(
533  const std::string& stmt_str_in) {
536 
537  std::string stmt_str = stmt_str_in;
538  // First remove special chars
539  boost::algorithm::trim_left_if(stmt_str, boost::algorithm::is_any_of("\n"));
540  // Then remove spaces
541  boost::algorithm::trim_left(stmt_str);
542 
543  ParserWrapper pw{stmt_str};
544 
545  auto query_state = create_query_state(session_info_, stmt_str);
546  auto stdlog = STDLOG(query_state);
547 
548  if (pw.is_ddl) {
549  const auto& cat = session_info_->getCatalog();
550  auto calcite_mgr = cat.getCalciteMgr();
551  const auto calciteQueryParsingOption =
552  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
553  const auto calciteOptimizationOption =
554  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
555  const auto query_json = calcite_mgr
556  ->process(query_state->createQueryStateProxy(),
557  pg_shim(stmt_str),
558  calciteQueryParsingOption,
559  calciteOptimizationOption)
560  .plan_result;
561  return Parser::create_stmt_for_json(query_json);
562  }
563 
564  // simply fail here as non-Calcite parsing is about to be removed
565  UNREACHABLE();
566  return nullptr;
567 }
568 
569 void QueryRunner::runDDLStatement(const std::string& stmt_str_in) {
572 
573  std::string stmt_str = stmt_str_in;
574  // First remove special chars
575  boost::algorithm::trim_left_if(stmt_str, boost::algorithm::is_any_of("\n"));
576  // Then remove spaces
577  boost::algorithm::trim_left(stmt_str);
578 
579  ParserWrapper pw{stmt_str};
580 
581  auto query_state = create_query_state(session_info_, stmt_str);
582  auto stdlog = STDLOG(query_state);
583 
584  if (pw.is_ddl || pw.getDMLType() == ParserWrapper::DMLType::Insert) {
585  auto& cat = session_info_->getCatalog();
586  auto calcite_mgr = cat.getCalciteMgr();
587  const auto calciteQueryParsingOption =
588  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
589  const auto calciteOptimizationOption =
590  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
591  const auto query_ra = calcite_mgr
592  ->process(query_state->createQueryStateProxy(),
593  pg_shim(stmt_str),
594  calciteQueryParsingOption,
595  calciteOptimizationOption)
596  .plan_result;
597  if (pw.getDMLType() == ParserWrapper::DMLType::Insert) {
598  rapidjson::Document ddl_query;
599  ddl_query.Parse(query_ra);
600  CHECK(ddl_query.HasMember("payload"));
601  CHECK(ddl_query["payload"].IsObject());
602  auto stmt = Parser::InsertValuesStmt(ddl_query["payload"].GetObject());
603  stmt.execute(*session_info_, false /* read only */);
604  return;
605  }
607  executor.execute(false /* read only */);
608  return;
609  }
610 }
611 
612 std::shared_ptr<ResultSet> QueryRunner::runSQL(const std::string& query_str,
614  ExecutionOptions eo) {
617 
618  ParserWrapper pw{query_str};
619  if (pw.getDMLType() == ParserWrapper::DMLType::Insert) {
620  runDDLStatement(query_str);
621  return nullptr;
622  }
623  const auto execution_result = runSelectQuery(query_str, std::move(co), std::move(eo));
624  VLOG(1) << session_info_->getCatalog().getDataMgr().getSystemMemoryUsage();
625  return execution_result->getRows();
626 }
627 
628 std::shared_ptr<ResultSet> QueryRunner::runSQL(const std::string& query_str,
629  const ExecutorDeviceType device_type,
630  const bool hoist_literals,
631  const bool allow_loop_joins) {
632  auto co = CompilationOptions::defaults(device_type);
633  co.hoist_literals = hoist_literals;
634  return runSQL(
635  query_str, std::move(co), defaultExecutionOptionsForRunSQL(allow_loop_joins));
636 }
637 
638 ExecutionOptions QueryRunner::defaultExecutionOptionsForRunSQL(bool allow_loop_joins,
639  bool just_explain) {
640  return {g_enable_columnar_output,
641  false,
642  true,
643  just_explain,
644  allow_loop_joins,
645  false,
646  false,
647  false,
648  false,
649  10000,
650  false,
651  false,
653  false,
654  1000};
655 }
656 
657 std::shared_ptr<Executor> QueryRunner::getExecutor() const {
660  auto query_state = create_query_state(session_info_, "");
661  auto stdlog = STDLOG(query_state);
663  return executor;
664 }
665 
666 std::shared_ptr<ResultSet> QueryRunner::runSQLWithAllowingInterrupt(
667  const std::string& query_str,
668  const std::string& session_id,
669  const ExecutorDeviceType device_type,
670  const double running_query_check_freq,
671  const unsigned pending_query_check_freq) {
674  auto current_user = session_info_->get_currentUser();
675  auto session_info = std::make_shared<Catalog_Namespace::SessionInfo>(
676  session_info_->get_catalog_ptr(), current_user, device_type, session_id);
677  auto query_state = create_query_state(session_info, query_str);
678  auto stdlog = STDLOG(query_state);
679  auto& cat = query_state->getConstSessionInfo()->getCatalog();
680  std::string query_ra{""};
681 
682  std::shared_ptr<ExecutionResult> result;
683  auto query_launch_task = std::make_shared<QueryDispatchQueue::Task>(
684  [&cat,
685  &query_ra,
686  &device_type,
687  &query_state,
688  &result,
689  &running_query_check_freq,
690  &pending_query_check_freq](const size_t worker_id) {
691  auto executor = Executor::getExecutor(worker_id);
693 
695  false,
696  true,
697  false,
698  true,
699  false,
700  false,
701  false,
702  false,
703  10000,
704  false,
705  false,
707  true,
708  running_query_check_freq,
709  pending_query_check_freq};
710  {
711  // async query initiation for interrupt test
712  // incurs data race warning in TSAN since
713  // calcite_mgr is shared across multiple query threads
714  // so here we lock the manager during query parsing
715  std::lock_guard<std::mutex> calcite_lock_guard(calcite_lock);
716  auto calcite_mgr = cat.getCalciteMgr();
717  const auto calciteQueryParsingOption =
718  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
719  const auto calciteOptimizationOption =
720  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
721  query_ra = calcite_mgr
722  ->process(query_state->createQueryStateProxy(),
723  pg_shim(query_state->getQueryStr()),
724  calciteQueryParsingOption,
725  calciteOptimizationOption)
726  .plan_result;
727  }
728  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra, query_state);
729  result = std::make_shared<ExecutionResult>(
730  ra_executor.executeRelAlgQuery(co, eo, false, nullptr));
731  });
733  executor->enrollQuerySession(session_id,
734  query_str,
735  query_state->getQuerySubmittedTime(),
737  QuerySessionStatus::QueryStatus::PENDING_QUEUE);
739  dispatch_queue_->submit(query_launch_task, /*is_update_delete=*/false);
740  auto result_future = query_launch_task->get_future();
741  result_future.get();
742  CHECK(result);
743  return result->getRows();
744 }
745 
746 std::vector<std::shared_ptr<ResultSet>> QueryRunner::runMultipleStatements(
747  const std::string& sql,
748  const ExecutorDeviceType dt) {
749  std::vector<std::shared_ptr<ResultSet>> results;
750  // TODO: Need to properly handle escaped semicolons instead of doing a naive split().
751  auto fields = split(sql, ";");
752  for (const auto& field : fields) {
753  auto text = strip(field) + ";";
754  if (text == ";") {
755  continue;
756  }
757 
758  ParserWrapper pw{text};
759  if (pw.is_ddl || pw.getDMLType() == ParserWrapper::DMLType::Insert) {
760  runDDLStatement(text);
761  results.push_back(nullptr);
762  } else {
763  // is not DDL, then assume it's DML and try to execute
764  results.push_back(runSQL(text, dt, true, true));
765  }
766  }
767  return results;
768 }
769 
770 void QueryRunner::runImport(Parser::CopyTableStmt* import_stmt) {
771  CHECK(import_stmt);
772  import_stmt->execute(*session_info_, false /* read only */);
773 }
774 
775 std::unique_ptr<import_export::Loader> QueryRunner::getLoader(
776  const TableDescriptor* td) const {
777  auto cat = getCatalog();
778  return std::make_unique<import_export::Loader>(*cat, td);
779 }
780 
781 namespace {
782 
783 std::shared_ptr<ExecutionResult> run_select_query_with_filter_push_down(
784  QueryStateProxy query_state_proxy,
785  const ExecutorDeviceType device_type,
786  const bool hoist_literals,
787  const bool allow_loop_joins,
788  const bool just_explain,
789  const ExecutorExplainType explain_type,
790  const bool with_filter_push_down) {
791  auto const& query_state = query_state_proxy.getQueryState();
792  auto& cat = query_state.getConstSessionInfo()->getCatalog();
795  co.explain_type = explain_type;
796 
799  eo.just_explain = just_explain;
800  eo.allow_loop_joins = allow_loop_joins;
801  eo.find_push_down_candidates = with_filter_push_down;
803 
804  auto calcite_mgr = cat.getCalciteMgr();
805  const auto calciteQueryParsingOption =
806  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
807  const auto calciteOptimizationOption =
808  calcite_mgr->getCalciteOptimizationOption(false, g_enable_watchdog, {});
809  const auto query_ra = calcite_mgr
810  ->process(query_state_proxy,
811  pg_shim(query_state.getQueryStr()),
812  calciteQueryParsingOption,
813  calciteOptimizationOption)
814  .plan_result;
815  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra);
816  auto result = std::make_shared<ExecutionResult>(
817  ra_executor.executeRelAlgQuery(co, eo, false, nullptr));
818  const auto& filter_push_down_requests = result->getPushedDownFilterInfo();
819  if (!filter_push_down_requests.empty()) {
820  std::vector<TFilterPushDownInfo> filter_push_down_info;
821  for (const auto& req : filter_push_down_requests) {
822  TFilterPushDownInfo filter_push_down_info_for_request;
823  filter_push_down_info_for_request.input_prev = req.input_prev;
824  filter_push_down_info_for_request.input_start = req.input_start;
825  filter_push_down_info_for_request.input_next = req.input_next;
826  filter_push_down_info.push_back(filter_push_down_info_for_request);
827  }
828  const auto calciteQueryParsingOption =
829  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
830  const auto calciteOptimizationOption = calcite_mgr->getCalciteOptimizationOption(
831  false, g_enable_watchdog, filter_push_down_info);
832  const auto new_query_ra = calcite_mgr
833  ->process(query_state_proxy,
834  pg_shim(query_state.getQueryStr()),
835  calciteQueryParsingOption,
836  calciteOptimizationOption)
837  .plan_result;
838  const ExecutionOptions eo_modified{eo.output_columnar_hint,
839  eo.keep_result,
840  eo.allow_multifrag,
841  eo.just_explain,
842  eo.allow_loop_joins,
843  eo.with_watchdog,
844  eo.jit_debug,
845  eo.just_validate,
848  /*find_push_down_candidates=*/false,
849  /*just_calcite_explain=*/false,
854  auto new_ra_executor = RelAlgExecutor(executor.get(), cat, new_query_ra);
855  return std::make_shared<ExecutionResult>(
856  new_ra_executor.executeRelAlgQuery(co, eo_modified, false, nullptr));
857  } else {
858  return result;
859  }
860 }
861 
862 } // namespace
863 
864 std::shared_ptr<ResultSet> QueryRunner::getCalcitePlan(const std::string& query_str,
865  bool enable_watchdog,
866  bool as_json_str) {
869  const auto& cat = session_info_->getCatalog();
870  auto query_state = create_query_state(session_info_, query_str);
871  auto stdlog = STDLOG(query_state);
872 
873  std::shared_ptr<ResultSet> result;
874  auto query_launch_task = std::make_shared<QueryDispatchQueue::Task>(
875  [&cat, &query_str, &enable_watchdog, &as_json_str, &query_state, &result](
876  const size_t worker_id) {
877  auto executor = Executor::getExecutor(worker_id);
878  auto calcite_mgr = cat.getCalciteMgr();
879  const auto calciteQueryParsingOption =
880  calcite_mgr->getCalciteQueryParsingOption(true, as_json_str, false);
881  const auto calciteOptimizationOption = calcite_mgr->getCalciteOptimizationOption(
882  g_enable_calcite_view_optimize, enable_watchdog, {});
883  const auto query_ra = calcite_mgr
884  ->process(query_state->createQueryStateProxy(),
885  pg_shim(query_str),
886  calciteQueryParsingOption,
887  calciteOptimizationOption)
888  .plan_result;
889  result = std::make_shared<ResultSet>(query_ra);
890  return result;
891  });
893  dispatch_queue_->submit(query_launch_task, /*is_update_delete=*/false);
894  auto result_future = query_launch_task->get_future();
895  result_future.get();
896  CHECK(result);
897  return result;
898 }
899 
900 std::shared_ptr<ExecutionResult> QueryRunner::runSelectQuery(const std::string& query_str,
902  ExecutionOptions eo) {
905  auto query_state = create_query_state(session_info_, query_str);
906  auto stdlog = STDLOG(query_state);
908  return run_select_query_with_filter_push_down(query_state->createQueryStateProxy(),
909  co.device_type,
910  co.hoist_literals,
911  eo.allow_loop_joins,
912  eo.just_explain,
915  }
916 
917  auto& cat = session_info_->getCatalog();
918 
919  std::shared_ptr<ExecutionResult> result;
920  auto query_launch_task =
921  std::make_shared<QueryDispatchQueue::Task>([&cat,
922  &query_str,
923  &co,
924  explain_type = this->explain_type_,
925  &eo,
926  &query_state,
927  &result](const size_t worker_id) {
928  auto executor = Executor::getExecutor(worker_id);
929  // TODO The next line should be deleted since it overwrites co, but then
930  // NycTaxiTest.RunSelectsEncodingDictWhereGreater fails due to co not getting
931  // reset to its default values.
932  co = CompilationOptions::defaults(co.device_type);
933  co.explain_type = explain_type;
934  auto calcite_mgr = cat.getCalciteMgr();
935  const auto calciteQueryParsingOption =
936  calcite_mgr->getCalciteQueryParsingOption(true, false, true);
937  const auto calciteOptimizationOption = calcite_mgr->getCalciteOptimizationOption(
939  const auto query_ra = calcite_mgr
940  ->process(query_state->createQueryStateProxy(),
941  pg_shim(query_str),
942  calciteQueryParsingOption,
943  calciteOptimizationOption)
944  .plan_result;
945  auto ra_executor = RelAlgExecutor(executor.get(), cat, query_ra);
946  result = std::make_shared<ExecutionResult>(
947  ra_executor.executeRelAlgQuery(co, eo, false, nullptr));
948  });
950  dispatch_queue_->submit(query_launch_task, /*is_update_delete=*/false);
951  auto result_future = query_launch_task->get_future();
952  result_future.get();
953  CHECK(result);
954  return result;
955 }
956 
957 std::shared_ptr<ExecutionResult> QueryRunner::runSelectQuery(
958  const std::string& query_str,
959  const ExecutorDeviceType device_type,
960  const bool hoist_literals,
961  const bool allow_loop_joins,
962  const bool just_explain) {
963  auto co = CompilationOptions::defaults(device_type);
964  co.hoist_literals = hoist_literals;
965  return runSelectQuery(query_str,
966  std::move(co),
967  defaultExecutionOptionsForRunSQL(allow_loop_joins, just_explain));
968 }
969 
970 ExtractedQueryPlanDag QueryRunner::extractQueryPlanDag(const std::string& query_str) {
971  auto query_dag_info = getQueryInfoForDataRecyclerTest(query_str);
973  auto extracted_dag_info = QueryPlanDagExtractor::extractQueryPlanDag(
974  query_dag_info.root_node.get(), executor);
975  return extracted_dag_info;
976 }
977 
978 // this function exists to test data recycler
979 // specifically, it is tricky to get a hashtable cache key when we only know
980 // a target query sql in test code
981 // so this function utilizes an incorrect way to manipulate our hashtable recycler
982 // but provides the cached hashtable for performing the test
983 // a set "visited" contains cached hashtable keys that we have retrieved so far
984 // based on that, this function iterates hashtable cache and return a cached one
985 // when its hashtable cache key has not been visited yet
986 // for instance, if we call this funtion with an empty "visited" key, we return
987 // the first hashtable that its iterator visits
988 std::tuple<QueryPlanHash,
989  std::shared_ptr<HashTable>,
990  std::optional<HashtableCacheMetaInfo>>
991 QueryRunner::getCachedHashtableWithoutCacheKey(std::set<size_t>& visited,
992  CacheItemType hash_table_type,
993  DeviceIdentifier device_identifier) {
994  HashtableRecycler* hash_table_cache{nullptr};
995  switch (hash_table_type) {
997  hash_table_cache = PerfectJoinHashTable::getHashTableCache();
998  break;
999  }
1001  hash_table_cache = BaselineJoinHashTable::getHashTableCache();
1002  break;
1003  }
1005  hash_table_cache = OverlapsJoinHashTable::getHashTableCache();
1006  break;
1007  }
1008  default: {
1009  UNREACHABLE();
1010  break;
1011  }
1012  }
1013  CHECK(hash_table_cache);
1014  return hash_table_cache->getCachedHashtableWithoutCacheKey(
1015  visited, hash_table_type, device_identifier);
1016 }
1017 
1018 std::shared_ptr<CacheItemMetric> QueryRunner::getCacheItemMetric(
1019  QueryPlanHash cache_key,
1020  CacheItemType hash_table_type,
1021  DeviceIdentifier device_identifier) {
1022  HashtableRecycler* hash_table_cache{nullptr};
1023  switch (hash_table_type) {
1025  hash_table_cache = PerfectJoinHashTable::getHashTableCache();
1026  break;
1027  }
1029  hash_table_cache = BaselineJoinHashTable::getHashTableCache();
1030  break;
1031  }
1033  hash_table_cache = OverlapsJoinHashTable::getHashTableCache();
1034  break;
1035  }
1036  default: {
1037  UNREACHABLE();
1038  break;
1039  }
1040  }
1041  CHECK(hash_table_cache);
1042  return hash_table_cache->getCachedItemMetric(
1043  hash_table_type, device_identifier, cache_key);
1044 }
1045 
1046 size_t QueryRunner::getNumberOfCachedItem(CacheItemStatus item_status,
1047  CacheItemType hash_table_type,
1048  bool with_overlaps_tuning_param) const {
1049  auto get_num_cached_auto_tuner_param = [&item_status]() {
1050  auto auto_tuner_cache = OverlapsJoinHashTable::getOverlapsTuningParamCache();
1051  CHECK(auto_tuner_cache);
1052  switch (item_status) {
1053  case CacheItemStatus::ALL: {
1054  return auto_tuner_cache->getCurrentNumCachedItems(
1057  }
1059  return auto_tuner_cache->getCurrentNumCleanCachedItems(
1062  }
1064  return auto_tuner_cache->getCurrentNumDirtyCachedItems(
1067  }
1068  default: {
1069  UNREACHABLE();
1070  return static_cast<size_t>(0);
1071  }
1072  }
1073  };
1074 
1075  auto get_num_cached_hashtable =
1076  [&item_status,
1077  &hash_table_type,
1078  &with_overlaps_tuning_param,
1079  &get_num_cached_auto_tuner_param](HashtableRecycler* hash_table_cache) {
1080  switch (item_status) {
1081  case CacheItemStatus::ALL: {
1082  if (with_overlaps_tuning_param) {
1083  // we assume additional consideration of turing param cache is only valid
1084  // for overlaps join hashtable
1085  CHECK_EQ(hash_table_type, CacheItemType::OVERLAPS_HT);
1086  return hash_table_cache->getCurrentNumCachedItems(
1087  hash_table_type, DataRecyclerUtil::CPU_DEVICE_IDENTIFIER) +
1088  get_num_cached_auto_tuner_param();
1089  }
1090  return hash_table_cache->getCurrentNumCachedItems(
1091  hash_table_type, DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
1092  }
1094  if (with_overlaps_tuning_param) {
1095  CHECK_EQ(hash_table_type, CacheItemType::OVERLAPS_HT);
1096  return hash_table_cache->getCurrentNumCleanCachedItems(
1097  hash_table_type, DataRecyclerUtil::CPU_DEVICE_IDENTIFIER) +
1098  get_num_cached_auto_tuner_param();
1099  }
1100  return hash_table_cache->getCurrentNumCleanCachedItems(
1101  hash_table_type, DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
1102  }
1104  if (with_overlaps_tuning_param) {
1105  CHECK_EQ(hash_table_type, CacheItemType::OVERLAPS_HT);
1106  return hash_table_cache->getCurrentNumDirtyCachedItems(
1107  hash_table_type, DataRecyclerUtil::CPU_DEVICE_IDENTIFIER) +
1108  get_num_cached_auto_tuner_param();
1109  }
1110  return hash_table_cache->getCurrentNumDirtyCachedItems(
1111  hash_table_type, DataRecyclerUtil::CPU_DEVICE_IDENTIFIER);
1112  }
1113  default: {
1114  UNREACHABLE();
1115  return static_cast<size_t>(0);
1116  }
1117  }
1118  };
1119 
1120  switch (hash_table_type) {
1122  auto hash_table_cache = PerfectJoinHashTable::getHashTableCache();
1123  CHECK(hash_table_cache);
1124  return get_num_cached_hashtable(hash_table_cache);
1125  }
1127  auto hash_table_cache = BaselineJoinHashTable::getHashTableCache();
1128  CHECK(hash_table_cache);
1129  return get_num_cached_hashtable(hash_table_cache);
1130  }
1132  auto hash_table_cache = OverlapsJoinHashTable::getHashTableCache();
1133  CHECK(hash_table_cache);
1134  return get_num_cached_hashtable(hash_table_cache);
1135  }
1137  return get_num_cached_auto_tuner_param();
1138  }
1139  default: {
1140  UNREACHABLE();
1141  return 0;
1142  }
1143  }
1144  return 0;
1145 }
1146 
1147 void QueryRunner::reset() {
1148  qr_instance_->query_engine_.reset();
1149  qr_instance_.reset(nullptr);
1151 }
1152 
1153 ImportDriver::ImportDriver(std::shared_ptr<Catalog_Namespace::Catalog> cat,
1154  const Catalog_Namespace::UserMetadata& user,
1155  const ExecutorDeviceType dt,
1156  const std::string session_id)
1157  : QueryRunner(
1158  std::make_unique<Catalog_Namespace::SessionInfo>(cat, user, dt, session_id)) {}
1159 
1160 void ImportDriver::importGeoTable(const std::string& file_path,
1161  const std::string& table_name,
1162  const bool compression,
1163  const bool create_table,
1164  const bool explode_collections) {
1165  using namespace import_export;
1166 
1167  static constexpr bool kIsGeoRaster{false};
1168 
1170 
1171  CopyParams copy_params;
1173  if (compression) {
1174  copy_params.geo_coords_encoding = EncodingType::kENCODING_GEOINT;
1175  copy_params.geo_coords_comp_param = 32;
1176  } else {
1177  copy_params.geo_coords_encoding = EncodingType::kENCODING_NONE;
1178  copy_params.geo_coords_comp_param = 0;
1179  }
1180  copy_params.geo_assign_render_groups = true;
1181  copy_params.geo_explode_collections = explode_collections;
1182 
1183  std::map<std::string, std::string> colname_to_src;
1184  auto& cat = session_info_->getCatalog();
1185  auto cds = Importer::gdalToColumnDescriptors(
1186  file_path, kIsGeoRaster, Geospatial::kGeoColumnName, copy_params);
1187 
1188  for (auto& cd : cds) {
1189  const auto col_name_sanitized = ImportHelpers::sanitize_name(cd.columnName);
1190  const auto ret =
1191  colname_to_src.insert(std::make_pair(col_name_sanitized, cd.columnName));
1192  CHECK(ret.second);
1193  cd.columnName = col_name_sanitized;
1194  }
1195 
1196  if (create_table) {
1197  const auto td = cat.getMetadataForTable(table_name);
1198  if (td != nullptr) {
1199  throw std::runtime_error(
1200  "Error: Table " + table_name +
1201  " already exists. Possible failure to correctly re-create " +
1202  shared::kDataDirectoryName + " directory.");
1203  }
1204  if (table_name != ImportHelpers::sanitize_name(table_name)) {
1205  throw std::runtime_error("Invalid characters in table name: " + table_name);
1206  }
1207 
1208  std::string stmt{"CREATE TABLE " + table_name};
1209  std::vector<std::string> col_stmts;
1210 
1211  for (auto& cd : cds) {
1212  if (cd.columnType.get_type() == SQLTypes::kINTERVAL_DAY_TIME ||
1213  cd.columnType.get_type() == SQLTypes::kINTERVAL_YEAR_MONTH) {
1214  throw std::runtime_error(
1215  "Unsupported type: INTERVAL_DAY_TIME or INTERVAL_YEAR_MONTH for col " +
1216  cd.columnName + " (table: " + table_name + ")");
1217  }
1218 
1219  if (cd.columnType.get_type() == SQLTypes::kDECIMAL) {
1220  if (cd.columnType.get_precision() == 0 && cd.columnType.get_scale() == 0) {
1221  cd.columnType.set_precision(14);
1222  cd.columnType.set_scale(7);
1223  }
1224  }
1225 
1226  std::string col_stmt;
1227  col_stmt.append(cd.columnName + " " + cd.columnType.get_type_name() + " ");
1228 
1229  if (cd.columnType.get_compression() != EncodingType::kENCODING_NONE) {
1230  col_stmt.append("ENCODING " + cd.columnType.get_compression_name() + " ");
1231  } else {
1232  if (cd.columnType.is_string()) {
1233  col_stmt.append("ENCODING NONE");
1234  } else if (cd.columnType.is_geometry()) {
1235  if (cd.columnType.get_output_srid() == 4326) {
1236  col_stmt.append("ENCODING NONE");
1237  }
1238  }
1239  }
1240  col_stmts.push_back(col_stmt);
1241  }
1242 
1243  stmt.append(" (" + boost::algorithm::join(col_stmts, ",") + ");");
1244  runDDLStatement(stmt);
1245 
1246  LOG(INFO) << "Created table: " << table_name;
1247  } else {
1248  LOG(INFO) << "Not creating table: " << table_name;
1249  }
1250 
1251  const auto td = cat.getMetadataForTable(table_name);
1252  if (td == nullptr) {
1253  throw std::runtime_error("Error: Failed to create table " + table_name);
1254  }
1255 
1256  import_export::Importer importer(cat, td, file_path, copy_params);
1257  auto ms = measure<>::execution(
1258  [&]() { importer.importGDAL(colname_to_src, session_info_.get(), kIsGeoRaster); });
1259  LOG(INFO) << "Import Time for " << table_name << ": " << (double)ms / 1000.0 << " s";
1260 }
1261 
1262 } // namespace QueryRunner
bool g_enable_calcite_view_optimize
Definition: QueryRunner.cpp:56
Classes used to wrap parser calls for calcite redirection.
static void addUdfs(const std::string &json_func_sigs)
#define CHECK_EQ(x, y)
Definition: Logger.h:230
ImportStatus importGDAL(const std::map< std::string, std::string > &colname_to_src, const Catalog_Namespace::SessionInfo *session_info, const bool is_raster)
Definition: Importer.cpp:5309
#define CALCITEPORT
Definition: QueryRunner.cpp:48
size_t DeviceIdentifier
Definition: DataRecycler.h:129
const std::string kDataDirectoryName
static std::vector< TableFunction > get_table_funcs()
static ExtractedQueryPlanDag extractQueryPlanDag(const RelAlgNode *top_node, Executor *executor)
std::string cat(Ts &&...args)
std::unique_ptr< QueryDispatchQueue > dispatch_queue_
Definition: QueryRunner.h:329
static TimeT::rep execution(F func, Args &&...args)
Definition: sample.cpp:29
ExecutorDeviceType
static void initialize(const std::string &data_dir, const std::string &allowed_import_paths, const std::string &allowed_export_paths)
Definition: DdlUtils.cpp:826
const std::string kDefaultDiskCacheDirName
std::string strip(std::string_view str)
trim any whitespace from the left and right ends of a string
ImportDriver(std::shared_ptr< Catalog_Namespace::Catalog > cat, const Catalog_Namespace::UserMetadata &user, const ExecutorDeviceType dt=ExecutorDeviceType::GPU, const std::string session_id="")
#define LOG(tag)
Definition: Logger.h:216
ExecutorExplainType explain_type_
Definition: QueryRunner.h:325
std::string join(T const &container, std::string const &delim)
static void add(const std::string &json_func_sigs)
#define UNREACHABLE()
Definition: Logger.h:266
static HashtableRecycler * getHashTableCache()
std::mutex calcite_lock
Definition: QueryRunner.cpp:57
std::optional< std::unordered_map< size_t, std::unordered_map< unsigned, RegisteredQueryHint > > > getParsedQueryHints()
std::optional< RegisteredQueryHint > getParsedQueryHint(const RelAlgNode *node)
void set_once_fatal_func(FatalFunc fatal_func)
Definition: Logger.cpp:337
Catalog_Namespace::DBMetadata db_metadata_
Definition: QueryRunner.h:327
static ExecutionOptions defaultExecutionOptionsForRunSQL(bool allow_loop_joins=true, bool just_explain=false)
virtual std::shared_ptr< ResultSet > runSQL(const std::string &query_str, CompilationOptions co, ExecutionOptions eo)
const std::string kGeoColumnName
Definition: ColumnNames.h:23
std::vector< std::string > split(std::string_view str, std::string_view delim, std::optional< size_t > maxsplit)
split apart a string into a vector of substrings
double running_query_interrupt_freq
#define LOG_IF(severity, condition)
Definition: Logger.h:312
static void clearMemory(const Data_Namespace::MemoryLevel memory_level)
Definition: Execute.cpp:492
static void addShutdownCallback(std::function< void()> shutdown_callback)
std::shared_ptr< const RelAlgNode > getRootRelAlgNodeShPtr() const
static std::shared_ptr< Executor > getExecutor(const ExecutorId id, const std::string &debug_dir="", const std::string &debug_file="", const SystemParameters &system_parameters=SystemParameters())
Definition: Execute.cpp:468
std::shared_ptr< QueryEngine > query_engine_
Definition: QueryRunner.h:331
This file contains the class specification and related data structures for Catalog.
RaExecutionSequence getRaExecutionSequence(const RelAlgNode *root_node, Executor *executor)
const rapidjson::Value & field(const rapidjson::Value &obj, const char field[]) noexcept
Definition: JsonAccessors.h:31
bool g_enable_columnar_output
Definition: Execute.cpp:99
Supported runtime functions management and retrieval.
static SysCatalog & instance()
Definition: SysCatalog.h:337
void execute(const Catalog_Namespace::SessionInfo &session, bool read_only_mode) override
Classes representing a parse tree.
CacheItemType
Definition: DataRecycler.h:38
const std::string kDefaultDbName
std::string g_base_path
Definition: SysCatalog.cpp:62
void init(LogOptions const &log_opts)
Definition: Logger.cpp:308
QueryState & getQueryState()
Definition: QueryState.h:181
std::unique_ptr< Parser::Stmt > create_stmt_for_json(const std::string &query_json)
static HashtableRecycler * getHashTableCache()
static std::shared_ptr< QueryEngine > createInstance(CudaMgr_Namespace::CudaMgr *cuda_mgr, bool cpu_only)
Definition: QueryEngine.h:89
A container for relational algebra descriptors defining the execution order for a relational algebra ...
ExecutorExplainType explain_type
bool g_enable_watchdog
virtual void runDDLStatement(const std::string &)
static std::unique_ptr< QueryRunner > qr_instance_
Definition: QueryRunner.h:323
double g_gpu_mem_limit_percent
Definition: QueryRunner.cpp:53
std::string sanitize_name(const std::string &name)
import_export::SourceType source_type
Definition: CopyParams.h:57
static OverlapsTuningParamRecycler * getOverlapsTuningParamCache()
unsigned pending_query_interrupt_freq
bool g_serialize_temp_tables
Definition: Catalog.cpp:108
ExecutorDeviceType device_type
std::optional< RegisteredQueryHint > getGlobalQueryHint()
void importGeoTable(const std::string &file_path, const std::string &table_name, const bool compression, const bool create_table, const bool explode_collections)
const std::string kRootUsername
static RegisteredQueryHint defaults()
Definition: QueryHint.h:247
TExtArgumentType::type to_thrift(const ExtArgumentType &t)
std::shared_ptr< ExecutionResult > run_select_query_with_filter_push_down(QueryStateProxy query_state_proxy, const ExecutorDeviceType device_type, const bool hoist_literals, const bool allow_loop_joins, const bool just_explain, const ExecutorExplainType explain_type, const bool with_filter_push_down)
ExecutionResult execute(bool read_only_mode)
bool g_read_only
Definition: File.cpp:40
std::shared_ptr< Catalog_Namespace::SessionInfo > session_info_
Definition: QueryRunner.h:328
size_t QueryPlanHash
virtual std::shared_ptr< ExecutionResult > runSelectQuery(const std::string &query_str, CompilationOptions co, ExecutionOptions eo)
std::shared_ptr< Calcite > g_calcite
Definition: QueryRunner.cpp:62
const std::string kCatalogDirectoryName
static std::shared_ptr< query_state::QueryState > create_query_state(Ts &&...args)
Definition: QueryRunner.h:301
static CompilationOptions defaults(const ExecutorDeviceType device_type=ExecutorDeviceType::GPU)
bool g_enable_filter_push_down
Definition: Execute.cpp:95
QueryPlanDagInfo getQueryInfoForDataRecyclerTest(const std::string &)
#define CHECK(condition)
Definition: Logger.h:222
static HashtableRecycler * getHashTableCache()
double gpu_input_mem_limit_percent
Serializers for query engine types to/from thrift.
std::shared_ptr< Catalog_Namespace::Catalog > getCatalog() const
unsigned dynamic_watchdog_time_limit
std::shared_ptr< Data_Namespace::DataMgr > data_mgr_
Definition: QueryRunner.h:330
size_t g_leaf_count
Definition: ParserNode.cpp:76
static constexpr DeviceIdentifier CPU_DEVICE_IDENTIFIER
Definition: DataRecycler.h:136
static ExecutionOptions defaults()
ExecutorExplainType
std::string pg_shim(const std::string &query)
#define STDLOG(...)
Definition: QueryState.h:235
#define VLOG(n)
Definition: Logger.h:316
std::atomic< bool > isSuper
Definition: SysCatalog.h:107
static const ExecutorId UNITARY_EXECUTOR_ID
Definition: Execute.h:376