Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : */
4 :
5 : #include <boost/tuple/tuple.hpp>
6 : #include <boost/foreach.hpp>
7 : #include "query.h"
8 : #include "stats_query.h"
9 : #include "base/work_pipeline.h"
10 :
11 : /*
12 : * This function performs GetRowAsync for each row key
13 : * Input: It takes rowkeys for which we have to get col
14 : values.Multiple instances execute this call
15 : simultaneously and possibly multiple times(steps)
16 : The instances store intermediate result in exts
17 : res stores consolidated result/instance
18 : * Ouput: Keeps calling itself for as many rowkeys involved
19 : Returns NULL if no more row key to be queried
20 : */
21 215579 : ExternalBase::Efn DbQueryUnit::QueryExec(uint32_t inst,
22 : const vector<q_result *> & exts,
23 : const Input & inp, Stage0Out & res) {
24 :
25 215579 : uint32_t step = exts.size();
26 215869 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
27 215869 : Input & cinp = const_cast<Input &> (inp);
28 : // Previous row fetch resulted in error, dont issue any more requests
29 215869 : if (query_fetch_error) {
30 360 : return NULL;
31 : }
32 216778 : if (step) {
33 214943 : res.query_result.insert(res.query_result.end(),
34 107465 : exts[step-1]->begin(),
35 107496 : exts[step-1]->end());
36 : }
37 216612 : res.current_row = cinp.row_count.fetch_add(1);
38 216612 : const uint32_t current_row = res.current_row;
39 : // continue as long as we have not reached end of rows
40 216612 : if (current_row < inp.total_rows) {
41 : // Frame the getrow context
42 108272 : GetRowInput *ip_ctx = new GetRowInput();
43 108063 : ip_ctx->rowkey = cinp.keys[current_row];
44 107812 : ip_ctx->cfname = cinp.cf_name;
45 107848 : ip_ctx->crange = cinp.cr;
46 107822 : ip_ctx->where_vec = cinp.where_vec;
47 107846 : ip_ctx->chunk_no = m_query->parallel_batch_num;
48 107846 : ip_ctx->qid = m_query->query_id;
49 107923 : ip_ctx->sub_qid = sub_query_id;
50 107923 : ip_ctx->row_no = current_row;
51 107923 : ip_ctx->inst = inst;
52 215732 : return boost::bind(&DbQueryUnit::PipelineCb, this, cinp.cf_name,
53 215214 : cinp.keys[current_row], cinp.cr, cinp.where_vec, ip_ctx, _1);
54 : } else {
55 : // done processing fetching all rows
56 107946 : return NULL;
57 : }
58 : }
59 :
60 108080 : bool DbQueryUnit::PipelineCb(std::string &cfname, GenDb::DbDataValueVec &rowkey,
61 : const GenDb::ColumnNameRange &cr,
62 : GenDb::WhereIndexInfoVec &where_vec,
63 : GetRowInput * ip_ctx, void *privdata) {
64 :
65 : // prepend T2: to value in each tuple in where_vec
66 108080 : std::string T2_string = GenDb::DbDataValueToString(rowkey.at(0));
67 321255 : BOOST_FOREACH(GenDb::WhereIndexInfo &where_info, where_vec) {
68 106396 : std::string value = GenDb::DbDataValueToString(where_info.get<2>());
69 106375 : if (boost::starts_with(value, "%")) {
70 1594 : continue;
71 : }
72 104734 : std::ostringstream where_oss;
73 105124 : where_oss << T2_string << ":" << value;
74 105126 : where_info.get<2>() = where_oss.str();
75 106416 : }
76 : /*
77 : * Call GetRowAsync, with args prepopulated
78 : */
79 108096 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
80 215836 : return m_query->dbif_->Db_GetRowAsync(cfname, rowkey, cr, where_vec,
81 : GenDb::DbConsistency::LOCAL_ONE,
82 215842 : boost::bind(&DbQueryUnit::cb, this, _1, _2, ip_ctx, privdata));
83 108171 : }
84 :
85 : /*
86 : * This function merges all the instance's results into the res
87 : */
88 1809 : bool DbQueryUnit::QueryMerge(const std::vector<boost::shared_ptr<Stage0Out> >
89 : & subs,
90 : const boost::shared_ptr<Input> & inp,
91 : Output & res) {
92 : // Merge all the instances result. Each instance is a vector of
93 : // query_results from steps
94 3617 : res.query_result = boost::shared_ptr<std::vector<query_result_unit_t> >
95 3617 : (new std::vector<query_result_unit_t>());
96 1807 : for (vector<boost::shared_ptr<Stage0Out> >::const_iterator it =
97 111627 : subs.begin(); it!=subs.end(); it++) {
98 324260 : res.query_result->insert(res.query_result->end(),
99 107642 : (*it)->query_result.begin(),
100 108083 : (*it)->query_result.end());
101 : }
102 1810 : return true;
103 : }
104 :
105 : /*
106 : * This function populates the rowkeys that have to be used
107 : * in querying.It populates it based on the cftype and the t2
108 : * and t1 values
109 : */
110 1810 : std::vector<GenDb::DbDataValueVec> DbQueryUnit::populate_row_keys() {
111 1810 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
112 1810 : uint32_t t2_start = m_query->from_time() >> g_viz_constants.RowTimeInBits;
113 1810 : uint32_t t2_end = m_query->end_time() >> g_viz_constants.RowTimeInBits;
114 :
115 1810 : if (m_query->is_object_table_query(m_query->table()))
116 : {
117 : GenDb::DbDataValue timestamp_start =
118 1056 : (uint32_t)std::numeric_limits<int32_t>::min();
119 1056 : cr.start_.push_back(timestamp_start);
120 1056 : }
121 : GenDb::DbDataValue timestamp_end =
122 1810 : (uint32_t)std::numeric_limits<int32_t>::max();
123 1810 : cr.finish_.push_back(timestamp_end);
124 :
125 1810 : std::vector<GenDb::DbDataValueVec> keys; // vector of keys for multi-row get
126 1810 : GenDb::ColListVec mget_res; // vector of result for each row
127 15671 : for (uint32_t t2 = t2_start; t2 <= t2_end; t2++)
128 : {
129 13861 : GenDb::ColList result;
130 13857 : GenDb::DbDataValueVec rowkey;
131 :
132 13856 : rowkey.push_back(t2);
133 13846 : if (m_query->is_stat_table_query(m_query->table())) {
134 1728 : uint8_t partition_no = 0;
135 1728 : rowkey.push_back(partition_no);
136 : }
137 41519 : if (m_query->is_flow_query(m_query->table()) ||
138 27653 : m_query->is_session_query(m_query->table())) {
139 3150 : for (uint8_t part_no = (uint8_t)g_viz_constants.PARTITION_MIN;
140 3150 : part_no < (uint8_t)g_viz_constants.PARTITION_MAX + 1;
141 : part_no++) {
142 2836 : GenDb::DbDataValueVec tmp_rowkey(rowkey);
143 2827 : tmp_rowkey.push_back(part_no);
144 2824 : if (m_query->is_flow_query(m_query->table())) {
145 1619 : for (uint8_t is_si = 0; is_si < 2; is_si++) {
146 1079 : GenDb::DbDataValueVec tmp_rowkey2(tmp_rowkey);
147 1078 : tmp_rowkey2.push_back(is_si);
148 1077 : for (GenDb::DbDataValueVec::iterator it =
149 3234 : row_key_suffix.begin(); it!=row_key_suffix.end();
150 1080 : it++) {
151 1077 : GenDb::DbDataValueVec tmp_rowkey3(tmp_rowkey2);
152 1079 : tmp_rowkey3.push_back(*it);
153 1077 : keys.push_back(tmp_rowkey3);
154 1080 : }
155 1080 : }
156 : } else {
157 2287 : for (GenDb::DbDataValueVec::iterator it =
158 9128 : row_key_suffix.begin(); it!=row_key_suffix.end();
159 4557 : it++) {
160 4556 : tmp_rowkey.push_back(*it);
161 : }
162 2280 : keys.push_back(tmp_rowkey);
163 : }
164 2827 : }
165 : } else {
166 13538 : if (!t_only_row)
167 : {
168 2083 : for (GenDb::DbDataValueVec::iterator it =
169 7124 : row_key_suffix.begin(); it!=row_key_suffix.end();
170 2959 : it++) {
171 2963 : rowkey.push_back(*it);
172 : }
173 : }
174 :
175 : // If querying message_index_tables, partition_no is an
176 : // additional row_key
177 : // It spans values 0..15
178 13535 : if (t_only_col) {
179 113297 : for (uint8_t part_no = (uint8_t)g_viz_constants.PARTITION_MIN;
180 113297 : part_no < (uint8_t)g_viz_constants.PARTITION_MAX + 1;
181 : part_no++) {
182 102084 : GenDb::DbDataValueVec tmp_rowkey(rowkey);
183 101539 : tmp_rowkey.push_back(part_no);
184 101540 : keys.push_back(tmp_rowkey);
185 101815 : }
186 : } else {
187 2083 : keys.push_back(rowkey);
188 : }
189 : }
190 13613 : }
191 3620 : return keys;
192 1810 : }
193 :
194 : /*
195 : * This function calls the GetRowAsync based on the input
196 : * passed to it. It creates a pipeline which executes
197 : * the multiple GetRow calls asynchronously. It also
198 : * provides a callback which is used to collect the
199 : * results from the GetRow operation
200 : */
201 1810 : query_status_t DbQueryUnit::process_query()
202 : {
203 1810 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
204 1810 : uint32_t t2_start = m_query->from_time() >> g_viz_constants.RowTimeInBits;
205 1810 : uint32_t t2_end = m_query->end_time() >> g_viz_constants.RowTimeInBits;
206 :
207 1810 : QE_TRACE(DEBUG, " Async Database query for " <<
208 : (t2_end - t2_start + 1) << " rows");
209 1810 : QE_TRACE(DEBUG, " Async Database query for T2_start:"
210 : << t2_start
211 : << " T2_end:" << t2_end
212 : << " cf:" << cfname
213 : << " where_vec size:" << where_vec.size()
214 : << " column_start size:" << cr.start_.size()
215 : << " column_end size:" << cr.finish_.size()
216 : << " where_vec size:" << where_vec.size());
217 1810 : std::vector<GenDb::DbDataValueVec> keys = populate_row_keys();
218 :
219 : /* Create a pipeline to fetch all rows corresponding to keys */
220 1810 : int max_tasks = keys.size();
221 1810 : std::vector<std::pair<int,int> > tinfo;
222 110536 : for (uint idx=0; idx<(uint)max_tasks; idx++) {
223 108726 : tinfo.push_back(make_pair(0, -1));
224 : }
225 :
226 : QEPipeT * wp = new QEPipeT(
227 3620 : new WorkStage<Input, Output, q_result, Stage0Out>(
228 : tinfo,
229 : boost::bind(&DbQueryUnit::QueryExec, this, _1, _2, _3, _4),
230 1810 : boost::bind(&DbQueryUnit::QueryMerge, this, _1, _2, _3)));
231 :
232 : // Populate the input to the pipeline
233 1810 : boost::shared_ptr<Input> inp(new Input());
234 1810 : inp.get()->row_count = 0;
235 1810 : inp.get()->total_rows = keys.size();
236 1810 : inp.get()->cf_name = cfname;
237 1810 : inp.get()->cr = cr;
238 1809 : inp.get()->where_vec = where_vec;
239 1809 : inp.get()->keys = keys;
240 : // Start the pipeline with callback and input
241 1810 : wp->Start(boost::bind(&DbQueryUnit::WPCompleteCb, this, wp, _1), inp);
242 1810 : return QUERY_IN_PROGRESS;
243 1810 : }
244 :
245 : /*
246 : * This is called after getting all the results from
247 : * all the row keys. It copies result to the query_result
248 : * and deletes the pipeline
249 : */
250 1809 : void DbQueryUnit::WPCompleteCb(QEPipeT *wp, bool ret_code) {
251 1809 : boost::shared_ptr<Output> res = wp->Result();
252 : //copy pipeline output to DbQueryUnit query_output
253 1810 : query_result = (res->query_result);
254 1810 : int size = res->query_result->size();
255 1809 : QE_TRACE(DEBUG, " Database query completed with Async "
256 : << size << " rows");
257 : // Have the result ready and processing is done
258 : // sort the result before returning
259 1810 : std::sort(query_result->begin(), query_result->end());
260 1809 : if (IS_TRACE_ENABLED(WHERE_RESULT_TRACE)) {
261 0 : std::stringstream ss;
262 0 : for (std::vector<query_result_unit_t>::const_iterator it =
263 0 : query_result->begin(); it != query_result->end(); it++) {
264 0 : const query_result_unit_t &result_unit(*it);
265 0 : ss << "T: " << result_unit.timestamp << ": ";
266 0 : for (GenDb::DbDataValueVec::const_iterator rt =
267 0 : result_unit.info.begin(); rt != result_unit.info.end();
268 0 : rt++) {
269 0 : ss << " " << *rt;
270 : }
271 0 : ss << std::endl;
272 : }
273 0 : QE_TRACE(DEBUG, "Result: " << cfname << ": " << ss.str());
274 0 : }
275 1810 : delete wp;
276 1810 : if (ret_code && !query_fetch_error) {
277 1807 : status_details = 0;
278 1807 : query_status = QUERY_SUCCESS;
279 : } else {
280 3 : query_status = QUERY_FAILURE;
281 : }
282 1810 : parent_query->subquery_processed(this);
283 1810 : }
284 :
285 108370 : void DbQueryUnit::cb(GenDb::DbOpResult::type dresult,
286 : std::auto_ptr<GenDb::ColList> column_list,
287 : GetRowInput * get_row_ctx, void * privdata) {
288 108370 : std::unique_ptr<q_result> q_result_ptr(new q_result);
289 108390 : std::unique_ptr<GetRowInput> gri(get_row_ctx);
290 : uint32_t t2;
291 108388 : uint8_t session_type = 0;
292 108388 : uint8_t is_si = 0;
293 108388 : AnalyticsQuery *m_query = (AnalyticsQuery *)main_query;
294 108388 : QueryEngine *qe = m_query->qe_;
295 : try {
296 108388 : GenDb::DbDataValueVec val = gri.get()->rowkey;
297 108358 : t2 = boost::get<uint32_t>(val.at(0));
298 108362 : if (m_query->is_session_query(m_query->table())
299 108359 : || m_query->is_flow_query(m_query->table())) {
300 3383 : session_type = boost::get<uint8_t>(val.at(3));
301 3383 : is_si = boost::get<uint8_t>(val.at(2));
302 : }
303 108363 : } catch (boost::bad_get& ex) {
304 0 : assert(0);
305 : }
306 108347 : if (dresult != GenDb::DbOpResult::OK) {
307 : // Dont issue any more requests
308 12 : query_fetch_error = true;
309 12 : ExternalProcIf<q_result> * rpi(
310 : reinterpret_cast<ExternalProcIf<q_result> *>(privdata));
311 12 : if (m_query->is_stat_table_query(m_query->table())) {
312 4 : std::scoped_lock lock(qe->smutex_);
313 4 : qe->stable_stats_.Update(m_query->stat_name_attr, false, true,
314 : false, 1);
315 4 : }
316 12 : rpi->Response(std::move(q_result_ptr));
317 12 : return;
318 : }
319 : // Update the reads against the stat
320 108335 : if (m_query->is_stat_table_query(m_query->table())) {
321 1611 : std::scoped_lock lock(qe->smutex_);
322 1611 : qe->stable_stats_.Update(m_query->stat_name_attr, false, false,
323 : false, 1);
324 1611 : }
325 108367 : GenDb::NewColVec::iterator i;
326 :
327 113137 : for (i = column_list->columns_.begin(); i != column_list->columns_.end();
328 4777 : i++) {
329 : {
330 4777 : query_result_unit_t result_unit;
331 : uint32_t t1;
332 4777 : if (m_query->is_stat_table_query(m_query->table())) {
333 : try {
334 221 : t1 = boost::get<uint32_t>(i->name->at(i->name->size()-2));
335 0 : } catch (boost::bad_get& ex) {
336 0 : assert(0);
337 : }
338 4556 : } else if (m_query->is_session_query(m_query->table())
339 4556 : || m_query->is_flow_query(m_query->table())) {
340 408 : int ts_at = 2;
341 : try {
342 408 : t1 = boost::get<uint32_t>(i->name->at(ts_at));
343 0 : } catch (boost::bad_get& ex) {
344 0 : assert(0);
345 : }
346 : } else {
347 : // For MessageIndex tables t1 is stored in the first column
348 : // except for timestamp table
349 4148 : int ts_at = 0;
350 4148 : if (t_only_col) {
351 4028 : ts_at = i->name->size() - 2;
352 : } else {
353 120 : ts_at = i->name->size() - 1;
354 : }
355 4148 : assert(ts_at >= 0);
356 : try {
357 4148 : t1 = boost::get<uint32_t>(i->name->at(ts_at));
358 0 : } catch (boost::bad_get& ex) {
359 0 : assert(0);
360 : }
361 : }
362 4777 : result_unit.timestamp = TIMESTAMP_FROM_T2T1(t2, t1);
363 :
364 5288 : if
365 9399 : ((result_unit.timestamp < m_query->from_time()) ||
366 4622 : (result_unit.timestamp > m_query->end_time()))
367 : {
368 : //QE_TRACE(DEBUG, "Discarding timestamp "
369 : // << result_unit.timestamp);
370 : // got a result outside of the time range
371 511 : continue;
372 : }
373 :
374 : // Add to result vector
375 4266 : if (m_query->is_stat_table_query(m_query->table())) {
376 220 : std::string attribstr;
377 : boost::uuids::uuid uuid;
378 :
379 : try {
380 220 : uuid = boost::get<boost::uuids::uuid>(i->name->at(i->name->size()-1));
381 0 : } catch (boost::bad_get& ex) {
382 0 : QE_ASSERT(0);
383 0 : } catch (const std::out_of_range& oor) {
384 0 : QE_ASSERT(0);
385 : }
386 :
387 : try {
388 220 : attribstr = boost::get<std::string>(i->value->at(i->value->size()-1));
389 0 : } catch (boost::bad_get& ex) {
390 0 : QE_ASSERT(0);
391 0 : } catch (const std::out_of_range& oor) {
392 0 : QE_ASSERT(0);
393 : }
394 :
395 220 : result_unit.set_stattable_info(
396 : attribstr,
397 : uuid);
398 4266 : } else if (m_query->is_session_query(m_query->table())
399 4046 : || m_query->is_flow_query(m_query->table())) {
400 243 : result_unit.info.clear();
401 243 : result_unit.info.push_back(is_si);
402 243 : result_unit.info.push_back(session_type);
403 243 : result_unit.info.push_back(i->name->at(0));
404 243 : result_unit.info.push_back(i->name->at(1));
405 243 : result_unit.info.push_back(i->name->at(2));
406 243 : result_unit.info.push_back(i->name->at(3));
407 243 : GenDb::DbDataValueVec::const_iterator itr;
408 : GenDb::DbDataValueVec::const_iterator end =
409 243 : (m_query->selectquery_->unroll_needed?
410 243 : (i->value->end()):(i->value->end() - 1));
411 7151 : for (itr = i->value->begin(); itr != end;
412 6909 : itr++) {
413 6909 : result_unit.info.push_back(*itr);
414 : }
415 : } else {
416 : // If message index table uuid is not the value, but
417 : // column name
418 3803 : if (t_only_col) {
419 3684 : GenDb::DbDataValueVec val = gri.get()->rowkey;
420 3684 : message_table_query_get_row(val, i, result_unit);
421 3684 : } else {
422 119 : result_unit.info = *i->value;
423 : }
424 : }
425 4266 : q_result_ptr->push_back(result_unit);
426 4777 : }
427 :
428 : }
429 108341 : if (privdata) {
430 108345 : ExternalProcIf<q_result> * rpi(
431 : reinterpret_cast<ExternalProcIf<q_result> *>(privdata));
432 108345 : rpi->Response(std::move(q_result_ptr));
433 : }
434 :
435 108393 : }
436 :
437 3684 : void DbQueryUnit::message_table_query_get_row(
438 : GenDb::DbDataValueVec const &val,
439 : GenDb::NewColVec::iterator const &res_it,
440 : query_result_unit_t &result_unit) {
441 : // cassandra returns fields in the ascending order by column-name.
442 : // pushing fields in order as per schema.
443 : // key = rowkey[0] T2
444 : // key2 = rowkey[1] partition
445 : // column1 = name[0] T1
446 : // column2 = name[1] UUID
447 : // column3 = value[10] T2:Source
448 : // column4 = value[11] T2:Messagetype
449 : // column5 = value[12] T2:ModuleId
450 : // column6 = value[13] T2:<object-type1>:<object-value1>)
451 : // column7 = value[14]
452 : // column8 = value[15]
453 : // column9 = value[16]
454 : // column10 = value[0]
455 : // column11 = value[1]
456 : // column12 = value[2]
457 : // column13 = value[3]
458 : // column14 = value[4]
459 : // column15 = value[5]
460 : // column16 = value[6]
461 : // column17 = value[7]
462 : // column18 = value[8]
463 : // column19 = value[9]
464 : // DATA = value[17]
465 3684 : result_unit.info.push_back(val.at(0));
466 3683 : result_unit.info.push_back(val.at(1));
467 3684 : result_unit.info.push_back(res_it->name->at(0));
468 3684 : result_unit.info.push_back(res_it->name->at(1));
469 3684 : result_unit.info.push_back(res_it->value->at(10));
470 3684 : result_unit.info.push_back(res_it->value->at(11));
471 3684 : result_unit.info.push_back(res_it->value->at(12));
472 3684 : result_unit.info.push_back(res_it->value->at(13));
473 3684 : result_unit.info.push_back(res_it->value->at(14));
474 3684 : result_unit.info.push_back(res_it->value->at(15));
475 3684 : result_unit.info.push_back(res_it->value->at(16));
476 3684 : result_unit.info.push_back(res_it->value->at(0));
477 3684 : result_unit.info.push_back(res_it->value->at(1));
478 3684 : result_unit.info.push_back(res_it->value->at(2));
479 3684 : result_unit.info.push_back(res_it->value->at(3));
480 3684 : result_unit.info.push_back(res_it->value->at(4));
481 3684 : result_unit.info.push_back(res_it->value->at(5));
482 3684 : result_unit.info.push_back(res_it->value->at(6));
483 3684 : result_unit.info.push_back(res_it->value->at(7));
484 3684 : result_unit.info.push_back(res_it->value->at(8));
485 3684 : result_unit.info.push_back(res_it->value->at(9));
486 3684 : result_unit.info.push_back(res_it->value->at(17));
487 3684 : }
|
