OmniSciDB  85c2d10cdc
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
ExecuteUpdate.cpp
Go to the documentation of this file.
1 /*
2  * Copyright 2020 OmniSci, 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 "QueryEngine/Execute.h"
18 
25 
27 
29  size_t const fragment_index,
30  const std::shared_ptr<ResultSet>& rs)
31  : fragment_info_(fragment_info), fragment_index_(fragment_index), rs_(rs) {
32  rs->setGeoReturnType(ResultSet::GeoReturnType::GeoTargetValue);
33 }
34 
35 std::vector<TargetValue> UpdateLogForFragment::getEntryAt(const size_t index) const {
36  return rs_->getRowAtNoTranslations(index);
37 }
38 
40  const size_t index) const {
41  return rs_->getRowAt(index);
42 }
43 
44 size_t const UpdateLogForFragment::getRowCount() const {
45  return rs_->rowCount();
46 }
47 
49  const {
50  return fragment_info_;
51 }
52 
54  return rs_->entryCount();
55 }
56 
58  return fragment_index_;
59 }
60 
61 SQLTypeInfo UpdateLogForFragment::getColumnType(const size_t col_idx) const {
62  return rs_->getColType(col_idx);
63 }
64 
66  const RelAlgExecutionUnit& ra_exe_unit_in,
67  const std::vector<InputTableInfo>& table_infos,
68  const CompilationOptions& co,
69  const ExecutionOptions& eo,
71  std::shared_ptr<RowSetMemoryOwner> row_set_mem_owner,
73  const bool is_agg) {
74  CHECK(cb);
75  VLOG(1) << "Executor " << executor_id_
76  << " is executing update/delete work unit:" << ra_exe_unit_in;
77 
78  const auto [ra_exe_unit, deleted_cols_map] = addDeletedColumn(ra_exe_unit_in, co);
79  ColumnCacheMap column_cache;
80 
81  ColumnFetcher column_fetcher(this, column_cache);
82  CHECK_GT(ra_exe_unit.input_descs.size(), size_t(0));
83  const auto table_id = ra_exe_unit.input_descs[0].getTableId();
84  const auto& outer_fragments = table_infos.front().info.fragments;
85 
86  std::vector<FragmentsPerTable> fragments = {{0, {0}}};
87  for (size_t tab_idx = 1; tab_idx < ra_exe_unit.input_descs.size(); tab_idx++) {
88  int table_id = ra_exe_unit.input_descs[tab_idx].getTableId();
89  CHECK_EQ(table_infos[tab_idx].table_id, table_id);
90  const auto& fragmentsPerTable = table_infos[tab_idx].info.fragments;
91  FragmentsPerTable entry = {table_id, {}};
92  for (size_t innerFragId = 0; innerFragId < fragmentsPerTable.size(); innerFragId++) {
93  entry.fragment_ids.push_back(innerFragId);
94  }
95  fragments.push_back(entry);
96  }
97 
98  if (outer_fragments.empty()) {
99  return {};
100  }
101 
102  const auto max_tuple_count_fragment_it = std::max_element(
103  outer_fragments.begin(), outer_fragments.end(), [](const auto& a, const auto& b) {
104  return a.getNumTuples() < b.getNumTuples();
105  });
106  CHECK(max_tuple_count_fragment_it != outer_fragments.end());
107  int64_t global_max_groups_buffer_entry_guess =
108  max_tuple_count_fragment_it->getNumTuples();
109  if (is_agg) {
110  global_max_groups_buffer_entry_guess = std::min(
111  2 * global_max_groups_buffer_entry_guess, static_cast<int64_t>(100'000'000));
112  }
113 
114  auto query_comp_desc = std::make_unique<QueryCompilationDescriptor>();
115  std::unique_ptr<QueryMemoryDescriptor> query_mem_desc;
116  {
117  auto clock_begin = timer_start();
118  std::lock_guard<std::mutex> compilation_lock(compilation_mutex_);
119  compilation_queue_time_ms_ += timer_stop(clock_begin);
120 
121  query_mem_desc = query_comp_desc->compile(global_max_groups_buffer_entry_guess,
122  8,
123  /*has_cardinality_estimation=*/true,
124  ra_exe_unit,
125  table_infos,
126  deleted_cols_map,
127  column_fetcher,
128  co,
129  eo,
130  nullptr,
131  this);
132  }
133  CHECK(query_mem_desc);
134 
135  TableUpdateMetadata table_update_metadata;
136  for (size_t fragment_index = 0; fragment_index < outer_fragments.size();
137  ++fragment_index) {
138  const int64_t crt_fragment_tuple_count =
139  outer_fragments[fragment_index].getNumTuples();
140  if (crt_fragment_tuple_count == 0) {
141  // nothing to update
142  continue;
143  }
144  SharedKernelContext shared_context(table_infos);
145  const auto& frag_offsets = shared_context.getFragOffsets();
146  auto skip_frag = skipFragment(ra_exe_unit.input_descs[0],
147  outer_fragments[fragment_index],
148  ra_exe_unit.simple_quals,
149  frag_offsets,
150  fragment_index);
151  if (skip_frag.first) {
152  VLOG(2) << "Update/delete skipping fragment with table id: "
153  << outer_fragments[fragment_index].physicalTableId
154  << ", fragment id: " << fragment_index;
155  continue;
156  }
157  fragments[0] = {table_id, {fragment_index}};
158 
159  {
160  ExecutionKernel current_fragment_kernel(ra_exe_unit,
162  0,
163  eo,
164  column_fetcher,
165  *query_comp_desc,
166  *query_mem_desc,
167  fragments,
169  /*render_info=*/nullptr,
170  /*rowid_lookup_key=*/-1);
171 
172  auto clock_begin = timer_start();
173  std::lock_guard<std::mutex> kernel_lock(kernel_mutex_);
174  kernel_queue_time_ms_ += timer_stop(clock_begin);
175 
176  current_fragment_kernel.run(this, 0, shared_context);
177  }
178  const auto& proj_fragment_results = shared_context.getFragmentResults();
179  if (proj_fragment_results.empty()) {
180  continue;
181  }
182  const auto& proj_fragment_result = proj_fragment_results[0];
183  const auto proj_result_set = proj_fragment_result.first;
184  CHECK(proj_result_set);
185  cb({outer_fragments[fragment_index], fragment_index, proj_result_set},
186  table_update_metadata);
187  }
188 
190  auto td = cat.getMetadataForTable(table_id);
191  TableOptimizer table_optimizer{td, this, cat};
192  table_optimizer.recomputeMetadataUnlocked(table_update_metadata);
193  }
194  return table_update_metadata;
195 }
SQLTypeInfo getColumnType(const size_t col_idx) const
bool is_agg(const Analyzer::Expr *expr)
#define CHECK_EQ(x, y)
Definition: Logger.h:205
std::string cat(Ts &&...args)
class for a per-database catalog. also includes metadata for the current database and the current use...
Definition: Catalog.h:101
const std::vector< uint64_t > & getFragOffsets()
size_t const getFragmentIndex() const
TableUpdateMetadata executeUpdate(const RelAlgExecutionUnit &ra_exe_unit, const std::vector< InputTableInfo > &table_infos, const CompilationOptions &co, const ExecutionOptions &eo, const Catalog_Namespace::Catalog &cat, std::shared_ptr< RowSetMemoryOwner > row_set_mem_owner, const UpdateLogForFragment::Callback &cb, const bool is_agg)
Driver for running cleanup processes on a table. TableOptimizer provides functions for various cleanu...
TypeR::rep timer_stop(Type clock_begin)
Definition: measure.h:48
bool g_enable_auto_metadata_update
#define CHECK_GT(x, y)
Definition: Logger.h:209
Container for compilation results and assorted options for a single execution unit.
std::function< void(const UpdateLogForFragment &, TableUpdateMetadata &)> Callback
Definition: Execute.h:343
std::vector< TargetValue > getEntryAt(const size_t index) const override
Used by Fragmenter classes to store info about each fragment - the fragment id and number of tuples(r...
Definition: Fragmenter.h:77
FragmentInfoType const & getFragmentInfo() const
size_t fragment_index_
Definition: Execute.h:349
std::vector< TargetValue > getTranslatedEntryAt(const size_t index) const override
std::unordered_map< int, std::unordered_map< int, std::shared_ptr< const ColumnarResults >>> ColumnCacheMap
executor_id_(executor_id)
std::shared_ptr< ResultSet > rs_
Definition: Execute.h:350
void run(Executor *executor, const size_t thread_idx, SharedKernelContext &shared_context)
FragmentInfoType const & fragment_info_
Definition: Execute.h:348
std::vector< std::pair< ResultSetPtr, std::vector< size_t > > > & getFragmentResults()
#define CHECK(condition)
Definition: Logger.h:197
std::vector< size_t > fragment_ids
const TableDescriptor * getMetadataForTable(const std::string &tableName, const bool populateFragmenter=true) const
Returns a pointer to a const TableDescriptor struct matching the provided tableName.
UpdateLogForFragment(FragmentInfoType const &fragment_info, size_t const, const std::shared_ptr< ResultSet > &rs)
size_t const getRowCount() const override
Descriptor for the fragments required for an execution kernel.
#define VLOG(n)
Definition: Logger.h:291
Type timer_start()
Definition: measure.h:42
size_t const getEntryCount() const override