Line data Source code
1 : /*
2 : * Copyright (c) 2013 Juniper Networks, Inc. All rights reserved.
3 : * Copyright Contributors to the OpenSDN project
4 : */
5 : #include "viz_constants.h"
6 : #include "stats_select.h"
7 : #include "stats_query.h"
8 : #include "query.h"
9 : #include <cstdlib>
10 : #include <boost/assign/list_of.hpp>
11 : #include <boost/functional/hash.hpp>
12 : #include "rapidjson/document.h"
13 : #include "rapidjson/stringbuffer.h"
14 : #include "rapidjson/writer.h"
15 :
16 : using boost::assign::map_list_of;
17 :
18 : using std::string;
19 : using std::map;
20 : using std::vector;
21 : using std::set;
22 : using std::pair;
23 : using std::make_pair;
24 :
25 : struct Centroid {
26 : };
27 :
28 : void
29 6 : StatsSelect::DeleteCentroid (Centroid *t) {
30 6 : free(t);
31 6 : }
32 :
33 : struct TDigest {
34 : };
35 :
36 : void
37 3 : StatsSelect::DeleteTDigest(TDigest *t) {
38 3 : TDigest_destroy(t);
39 3 : }
40 :
41 : bool
42 194 : StatsSelect::Jsonify(const std::string& table,
43 : const std::map<std::string, StatVal>& uniks,
44 : const QEOpServerProxy::AggRowT& aggs, std::string& jstr) {
45 :
46 194 : contrail_rapidjson::Document dd;
47 194 : dd.SetObject();
48 :
49 : try {
50 194 : for (std::map<std::string, StatVal>::const_iterator it = uniks.begin();
51 871 : it!= uniks.end(); it++) {
52 677 : switch (it->second.which()) {
53 372 : case QEOpServerProxy::STRING : {
54 372 : string mapit = boost::get<string>(it->second);
55 372 : contrail_rapidjson::Value val(contrail_rapidjson::kStringType);
56 372 : val.SetString(mapit.c_str(), dd.GetAllocator());
57 372 : contrail_rapidjson::Value vk;
58 372 : dd.AddMember(vk.SetString(it->first.c_str(),
59 : dd.GetAllocator()),
60 : val, dd.GetAllocator());
61 372 : }
62 372 : break;
63 49 : case QEOpServerProxy::UUID : {
64 49 : boost::uuids::uuid mapit = boost::get<boost::uuids::uuid>(it->second);
65 49 : contrail_rapidjson::Value val(contrail_rapidjson::kStringType);
66 49 : std::string ustr = to_string(mapit);
67 49 : val.SetString(ustr.c_str(), dd.GetAllocator());
68 49 : contrail_rapidjson::Value vk;
69 49 : dd.AddMember(vk.SetString(it->first.c_str(),
70 : dd.GetAllocator()),
71 : val, dd.GetAllocator());
72 49 : }
73 49 : break;
74 229 : case QEOpServerProxy::UINT64 : {
75 229 : uint64_t mapit = boost::get<uint64_t>(it->second);
76 229 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
77 229 : val.SetUint64(mapit);
78 229 : contrail_rapidjson::Value vk;
79 229 : dd.AddMember(vk.SetString(it->first.c_str(),
80 : dd.GetAllocator()),
81 : val, dd.GetAllocator());
82 229 : }
83 229 : break;
84 27 : case QEOpServerProxy::DOUBLE : {
85 27 : double mapit = boost::get<double>(it->second);
86 27 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
87 27 : val.SetDouble(mapit);
88 27 : contrail_rapidjson::Value vk;
89 27 : dd.AddMember(vk.SetString(it->first.c_str(),
90 : dd.GetAllocator()),
91 : val, dd.GetAllocator());
92 27 : }
93 27 : break;
94 0 : default:
95 0 : QE_ASSERT(0);
96 : }
97 : }
98 194 : for (QEOpServerProxy::AggRowT::const_iterator it = aggs.begin();
99 301 : it!= aggs.end(); it++) {
100 :
101 107 : string sname("");
102 107 : if (it->first.first == QEOpServerProxy::SUM) {
103 159 : if (it->first.second == g_viz_constants.SESSION_FWD_TEARDOWN_BYTES ||
104 150 : it->first.second == g_viz_constants.SESSION_FWD_TEARDOWN_PKTS ||
105 141 : it->first.second == g_viz_constants.SESSION_REV_TEARDOWN_BYTES ||
106 132 : it->first.second == g_viz_constants.SESSION_REV_TEARDOWN_PKTS ||
107 225 : it->first.second == g_viz_constants.FLOW_TABLE_AGG_PKTS ||
108 66 : it->first.second == g_viz_constants.FLOW_TABLE_AGG_BYTES) {
109 18 : sname = it->first.second;
110 : } else {
111 66 : sname = string("SUM(") + it->first.second + string(")");
112 : }
113 23 : } else if (it->first.first == QEOpServerProxy::COUNT) {
114 7 : if (table == g_viz_constants.SESSION_SERIES_TABLE) {
115 5 : sname = "sample_count";
116 : } else {
117 2 : sname = string("COUNT(") + it->first.second + string(")");
118 : }
119 16 : } else if (it->first.first == QEOpServerProxy::CLASS) {
120 9 : if (table == g_viz_constants.SESSION_SERIES_TABLE) {
121 0 : sname = "session_class_id";
122 9 : } else if (table == g_viz_constants.FLOW_SERIES_TABLE){
123 0 : sname = "flow_class_id";
124 : } else {
125 9 : sname = string("CLASS(") + it->first.second + string(")");
126 : }
127 7 : } else if (it->first.first == QEOpServerProxy::MAX) {
128 1 : sname = string("MAX(") + it->first.second + string(")");
129 6 : } else if (it->first.first == QEOpServerProxy::MIN) {
130 1 : sname = string("MIN(") + it->first.second + string(")");
131 5 : } else if (it->first.first == QEOpServerProxy::PERCENTILES) {
132 1 : sname = string("PERCENTILES(") + it->first.second + string(")");
133 4 : } else if (it->first.first == QEOpServerProxy::AVG) {
134 2 : sname = string("AVG(") + it->first.second + string(")");
135 2 : } else if (it->first.first == QEOpServerProxy::COUNT_DISTINCT) {
136 2 : sname = string("COUNT_DISTINCT(") + it->first.second + string(")");
137 : } else {
138 0 : QE_ASSERT(0);
139 : }
140 107 : switch (it->second.which()) {
141 :
142 103 : case QEOpServerProxy::UINT64 : {
143 103 : uint64_t mapit = boost::get<uint64_t>(it->second);
144 103 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
145 103 : val.SetUint64(mapit);
146 103 : contrail_rapidjson::Value vk;
147 103 : dd.AddMember(vk.SetString(sname.c_str(),
148 : dd.GetAllocator()),
149 : val, dd.GetAllocator());
150 103 : }
151 103 : break;
152 1 : case QEOpServerProxy::DOUBLE : {
153 1 : double mapit = boost::get<double>(it->second);
154 1 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
155 1 : val.SetDouble(mapit);
156 1 : contrail_rapidjson::Value vk;
157 1 : dd.AddMember(vk.SetString(sname.c_str(),
158 : dd.GetAllocator()),
159 : val, dd.GetAllocator());
160 1 : }
161 1 : break;
162 1 : case QEOpServerProxy::TDIGEST : {
163 : boost::shared_ptr<TDigest> mapit =
164 1 : boost::get<boost::shared_ptr<TDigest> >(it->second);
165 :
166 1 : contrail_rapidjson::Value val(contrail_rapidjson::kObjectType);
167 1 : float ptiles[7] = {.01,.05,.25,.5,.75,.95,.99};
168 9 : std::string stiles[7] = {"01","05","25","50","75","95","99"};
169 : #if 0
170 : size_t jj = TDigest_get_ncentroids(mapit.get());
171 : for (size_t kk=0; kk < jj; kk++) {
172 : Centroid *c = TDigest_get_centroid(mapit.get(), kk);
173 : std::cout << "Parse Centroid " <<
174 : Centroid_get_mean(c) << " " <<
175 : Centroid_get_count(c) << " " <<
176 : Centroid_quantile(c, mapit.get()) << std::endl;
177 : }
178 : #endif
179 8 : for (size_t idx=0; idx<7; idx++) {
180 7 : contrail_rapidjson::Value sval(contrail_rapidjson::kNumberType);
181 7 : sval.SetDouble(TDigest_percentile(mapit.get(), ptiles[idx]));
182 7 : contrail_rapidjson::Value vk;
183 7 : val.AddMember(vk.SetString(stiles[idx].c_str(),
184 : dd.GetAllocator()),
185 : sval, dd.GetAllocator());
186 7 : }
187 1 : contrail_rapidjson::Value vk;
188 1 : dd.AddMember(vk.SetString(sname.c_str(),
189 : dd.GetAllocator()),
190 : val, dd.GetAllocator());
191 10 : }
192 1 : break;
193 2 : case QEOpServerProxy::CENTROID : {
194 : boost::shared_ptr<Centroid> mapit =
195 2 : boost::get<boost::shared_ptr<Centroid> >(it->second);
196 2 : contrail_rapidjson::Value val(contrail_rapidjson::kNumberType);
197 2 : val.SetDouble(Centroid_get_mean(mapit.get()));
198 2 : contrail_rapidjson::Value vk;
199 2 : dd.AddMember(vk.SetString(sname.c_str(),
200 : dd.GetAllocator()),
201 : val, dd.GetAllocator());
202 2 : }
203 2 : break;
204 0 : default:
205 0 : QE_ASSERT(0);
206 : }
207 107 : }
208 0 : } catch (boost::bad_get& ex) {
209 0 : QE_ASSERT(0);
210 0 : } catch (const std::out_of_range& oor) {
211 0 : QE_ASSERT(0);
212 : }
213 388 : contrail_rapidjson::StringBuffer sb;
214 194 : contrail_rapidjson::Writer<contrail_rapidjson::StringBuffer> writer(sb);
215 194 : dd.Accept(writer);
216 194 : jstr = sb.GetString();
217 : //QE_LOG_NOQID(INFO, "Jsonify row " << jstr);
218 388 : return true;
219 194 : }
220 :
221 63 : void StatsSelect::MergeFinal(const std::vector<boost::shared_ptr<MapBufT> >& inputs,
222 : MapBufT& output) {
223 63 : MapBufT temp;
224 567 : for (size_t idx = 0; idx < inputs.size(); idx++) {
225 504 : if (agg_sort_cols_.empty())
226 496 : Merge(count_distinct_field_, *inputs[idx], output);
227 : else
228 8 : Merge(count_distinct_field_, *inputs[idx], temp);
229 : }
230 63 : if (!agg_sort_cols_.empty()) {
231 1 : MapBufT::iterator jt = temp.end();
232 4 : for (MapBufT::iterator it = temp.begin(); it!= temp.end(); it++) {
233 3 : if (jt!=temp.end()) {
234 2 : temp.erase(jt);
235 : }
236 3 : std::vector<StatVal> ukey = it->first;
237 3 : const StatMap& uniks = it->second.first;
238 3 : const QEOpServerProxy::AggRowT& narows = it->second.second;
239 3 : for (AggSortT::const_iterator xt = agg_sort_cols_.begin();
240 6 : xt != agg_sort_cols_.end(); xt++) {
241 3 : QE_ASSERT(narows.find(xt->first) != narows.end());
242 3 : ukey[xt->second] = narows.at(xt->first);
243 : }
244 3 : MergeFullRow(ukey, uniks, narows, output);
245 3 : jt = it;
246 3 : }
247 1 : temp.clear();
248 : }
249 63 : }
250 :
251 920 : void StatsSelect::Merge(const std::string& count_distinct_field_, const MapBufT& input, MapBufT& output) {
252 920 : for (MapBufT::const_iterator it = input.begin();
253 1470 : it!=input.end(); it++) {
254 550 : std::vector<StatVal> ukey(it->first);
255 550 : StatMap uniks(it->second.first);
256 550 : QEOpServerProxy::AggRowT narows(it->second.second);
257 :
258 550 : StatMap::iterator itr = uniks.find(count_distinct_field_);
259 550 : if (itr != uniks.end()) {
260 : pair<QEOpServerProxy::AggOper, std::string> aggkey(
261 3 : QEOpServerProxy::COUNT_DISTINCT, count_distinct_field_);
262 3 : narows.insert(make_pair(aggkey, (uint64_t) 1));
263 3 : uniks.erase(itr);
264 3 : size_t hash_slot = ukey.size() - 1;
265 3 : uint64_t hash_val = boost::hash_range(uniks.begin(), uniks.end());
266 3 : ukey[hash_slot] = hash_val;
267 3 : }
268 :
269 550 : MergeFullRow(ukey, uniks, narows, output);
270 :
271 550 : }
272 920 : }
273 :
274 1055 : StatsSelect::StatsSelect(AnalyticsQuery * m_query,
275 1055 : const std::vector<std::string> & select_fields) :
276 1055 : main_query(m_query), select_fields_(select_fields),
277 1056 : ts_period_(0), isT_(false), count_field_() {
278 :
279 1057 : status_ = false;
280 :
281 5109 : for (size_t j=0; j<select_fields_.size(); j++) {
282 :
283 4051 : if (select_fields_[j] == g_viz_constants.STAT_TIME_FIELD) {
284 89 : if (ts_period_) {
285 0 : QE_LOG(INFO, "StatsSelect cannot accept both T and T=");
286 0 : return;
287 : }
288 89 : isT_ = true;
289 89 : sort_cols_.insert(std::make_pair(select_fields_[j], 0));
290 89 : QE_TRACE(DEBUG, "StatsSelect T");
291 :
292 3962 : } else if (select_fields_[j].substr(0, g_viz_constants.STAT_TIMEBIN_FIELD.size()) ==
293 : g_viz_constants.STAT_TIMEBIN_FIELD) {
294 40 : if (isT_) {
295 0 : QE_LOG(INFO, "StatsSelect cannot accept both T and T=");
296 0 : return;
297 : }
298 40 : string tsstr = select_fields_[j].substr(g_viz_constants.STAT_TIMEBIN_FIELD.size());
299 40 : ts_period_ = strtoul(tsstr.c_str(), NULL, 10) * 1000000;
300 40 : if (!ts_period_) {
301 0 : QE_LOG(INFO, "StatsSelect cannot accept T= for " << tsstr);
302 0 : return;
303 : }
304 40 : sort_cols_.insert(std::make_pair(g_viz_constants.STAT_TIMEBIN_FIELD,0));
305 40 : QE_TRACE(DEBUG, "StatsSelect T=");
306 40 : } else {
307 3922 : std::string sfield = select_fields_[j];
308 : QEOpServerProxy::AggOper agg =
309 3921 : StatsQuery::ParseAgg(select_fields_[j], sfield);
310 3922 : if (agg == QEOpServerProxy::INVALID) {
311 3401 : unik_cols_.insert(select_fields_[j]);
312 3402 : QE_TRACE(DEBUG, "StatsSelect unik field " << select_fields_[j]);
313 521 : } else if (agg == QEOpServerProxy::COUNT) {
314 54 : count_field_ = sfield;
315 54 : QE_TRACE(DEBUG, "StatsSelect COUNT " << sfield);
316 467 : } else if (agg == QEOpServerProxy::COUNT_DISTINCT) {
317 18 : count_distinct_field_ = sfield;
318 18 : unik_cols_.insert(sfield);
319 18 : QE_TRACE(DEBUG, "StatsSelect COUNT_DISTINCT " << sfield);
320 449 : } else if (agg == QEOpServerProxy::SUM) {
321 288 : sum_cols_.insert(sfield);
322 288 : QE_TRACE(DEBUG, "StatsSelect SUM " << sfield);
323 161 : } else if (agg == QEOpServerProxy::CLASS) {
324 17 : class_cols_.insert(sfield);
325 17 : QE_TRACE(DEBUG, "StatsSelect CLASS " << sfield);
326 144 : } else if (agg == QEOpServerProxy::MAX) {
327 36 : max_field_.insert(sfield);
328 36 : QE_TRACE(DEBUG, "StatsSelect MAX " << sfield);
329 108 : } else if (agg == QEOpServerProxy::MIN) {
330 18 : min_field_.insert(sfield);
331 18 : QE_TRACE(DEBUG, "StatsSelect MIN " << sfield);
332 90 : } else if (agg == QEOpServerProxy::PERCENTILES) {
333 36 : percentile_cols_.insert(sfield);
334 36 : QE_TRACE(DEBUG, "StatsSelect PERCENTILES " << sfield);
335 54 : } else if (agg == QEOpServerProxy::AVG) {
336 54 : avg_field_.insert(sfield);
337 54 : QE_TRACE(DEBUG, "StatsSelect AVG " << sfield);
338 : } else {
339 0 : QE_ASSERT(0);
340 : }
341 3923 : }
342 : }
343 :
344 1057 : status_ = true;
345 0 : }
346 :
347 1057 : void StatsSelect::SetSortOrder(const std::vector<sort_field_t>& sort_fields) {
348 1057 : if (sort_fields.size()) {
349 24 : sort_cols_.clear();
350 : } else {
351 1033 : QE_TRACE(DEBUG, "StatsSelect Sort By T/T=");
352 : }
353 1081 : for (size_t st = 0; st < sort_fields.size(); st++) {
354 :
355 24 : if (sort_fields[st].name.substr(0, g_viz_constants.STAT_TIMEBIN_FIELD.size()) ==
356 : g_viz_constants.STAT_TIMEBIN_FIELD) {
357 0 : QE_TRACE(DEBUG, "StatsSelect Sort By T=");
358 0 : sort_cols_.insert(std::make_pair(g_viz_constants.STAT_TIMEBIN_FIELD,st));
359 24 : } else if (sort_fields[st].name == g_viz_constants.STAT_TIME_FIELD) {
360 0 : QE_TRACE(DEBUG, "StatsSelect Sort By " << sort_fields[st].name);
361 0 : sort_cols_.insert(std::make_pair(sort_fields[st].name, st));
362 : } else {
363 24 : std::string sfield = sort_fields[st].name;
364 : QEOpServerProxy::AggOper agg =
365 24 : StatsQuery::ParseAgg(sort_fields[st].name, sfield);
366 24 : if (main_query->is_stat_table_query(main_query->table()) &&
367 0 : main_query->stats().is_stat_table_static()) {
368 0 : StatsQuery::column_t c = main_query->stats().get_column_desc(sfield);
369 0 : if (c.datatype == QEOpServerProxy::BLANK) {
370 0 : QE_LOG(INFO, "StatsSelect unknown field " << sort_fields[st].name);
371 0 : return;
372 : }
373 0 : }
374 24 : if (agg == QEOpServerProxy::INVALID) {
375 12 : QE_TRACE(DEBUG, "StatsSelect Sort By " << sort_fields[st].name);
376 12 : sort_cols_.insert(std::make_pair(sort_fields[st].name,st));
377 : } else {
378 12 : agg_sort_cols_.insert(make_pair(make_pair(agg, sfield), st));
379 12 : QE_TRACE(DEBUG, "StatsSelect Sort Agg " << agg << " of " << sfield);
380 : }
381 24 : }
382 : }
383 : }
384 :
385 1088 : void StatsSelect::MergeFullRow(
386 : const std::vector<StatVal>& ukey,
387 : const StatMap& uniks,
388 : const QEOpServerProxy::AggRowT& narows,
389 : MapBufT& output) {
390 :
391 1088 : MapBufT::iterator rt = output.find(ukey);
392 1088 : if (rt!= output.end()) {
393 229 : StatMap& rm = rt->second.first;
394 229 : if (rm!=uniks) {
395 0 : output.insert(make_pair(ukey, make_pair(uniks, narows)));
396 : } else {
397 229 : QEOpServerProxy::AggRowT &arows = rt->second.second;
398 229 : MergeAggRow(arows, narows);
399 : }
400 : } else {
401 859 : QEOpServerProxy::AggRowT arows;
402 859 : output.insert(make_pair(ukey, make_pair(uniks, narows)));
403 859 : }
404 1088 : }
405 :
406 :
407 229 : void StatsSelect::MergeAggRow(QEOpServerProxy::AggRowT &arows,
408 : const QEOpServerProxy::AggRowT &narows) {
409 229 : if (arows.empty()) {
410 119 : arows = narows;
411 119 : return;
412 : }
413 110 : for (QEOpServerProxy::AggRowT::iterator jt = arows.begin();
414 337 : jt!= arows.end(); jt++) {
415 227 : QEOpServerProxy::AggRowT::const_iterator kt = narows.find(jt->first);
416 227 : if (kt!=narows.end()) {
417 : // Attribute name must match for aggregate and for the new value
418 209 : QE_ASSERT(jt->first.second == kt->first.second);
419 :
420 209 : if (jt->first.first == QEOpServerProxy::SUM) {
421 175 : StatsSelect::StatVal & sv = jt->second;
422 : try {
423 175 : if (sv.which() == QEOpServerProxy::UINT64) {
424 175 : sv = boost::get<uint64_t>(sv) + boost::get<uint64_t>(kt->second);
425 : }
426 175 : if (sv.which() == QEOpServerProxy::DOUBLE) {
427 0 : sv = boost::get<double>(sv) + boost::get<double>(kt->second);
428 : }
429 0 : } catch (boost::bad_get& ex) {
430 0 : QE_ASSERT(0);
431 0 : } catch (const std::out_of_range& oor) {
432 0 : QE_ASSERT(0);
433 : }
434 :
435 : }
436 :
437 395 : if (jt->first.first == QEOpServerProxy::COUNT ||
438 186 : jt->first.first == QEOpServerProxy::COUNT_DISTINCT) {
439 : try {
440 24 : uint64_t& sv = boost::get<uint64_t>(jt->second);
441 24 : sv = sv + boost::get<uint64_t>(kt->second);
442 0 : } catch (boost::bad_get& ex) {
443 0 : QE_ASSERT(0);
444 0 : } catch (const std::out_of_range& oor) {
445 0 : QE_ASSERT(0);
446 : }
447 : }
448 209 : if (jt->first.first == QEOpServerProxy::MAX) {
449 2 : StatsSelect::StatVal & sv = jt->second;
450 : try {
451 2 : if (sv.which() == QEOpServerProxy::UINT64) {
452 2 : uint64_t& existing_max = boost::get<uint64_t>(jt->second);
453 2 : existing_max = (existing_max > (boost::get<uint64_t>(kt->second))) ? existing_max:(boost::get<uint64_t>(kt->second));
454 : }
455 2 : if (sv.which() == QEOpServerProxy::DOUBLE) {
456 0 : double& existing_max = boost::get<double>(jt->second);
457 0 : existing_max = (existing_max > (boost::get<double>(kt->second))) ? existing_max:(boost::get<double>(kt->second));
458 : }
459 0 : } catch (boost::bad_get& ex) {
460 0 : QE_ASSERT(0);
461 : }
462 : }
463 209 : if (jt->first.first == QEOpServerProxy::MIN) {
464 2 : StatsSelect::StatVal & sv = jt->second;
465 : try {
466 2 : if (sv.which() == QEOpServerProxy::UINT64) {
467 0 : uint64_t& existing_min = boost::get<uint64_t>(jt->second);
468 0 : existing_min = (existing_min < (boost::get<uint64_t>(kt->second))) ? existing_min:(boost::get<uint64_t>(kt->second));
469 : }
470 2 : if (sv.which() == QEOpServerProxy::DOUBLE) {
471 2 : double& existing_min = boost::get<double>(jt->second);
472 2 : existing_min = (existing_min < (boost::get<double>(kt->second))) ? existing_min:(boost::get<double>(kt->second));
473 : }
474 0 : } catch (boost::bad_get& ex) {
475 0 : QE_ASSERT(0);
476 : }
477 : }
478 209 : if (jt->first.first == QEOpServerProxy::AVG) {
479 4 : StatsSelect::StatVal & sv = jt->second;
480 : try {
481 4 : if (sv.which() == QEOpServerProxy::CENTROID) {
482 : boost::shared_ptr<Centroid>& pt =
483 4 : boost::get<boost::shared_ptr<Centroid> >(jt->second);
484 :
485 : boost::shared_ptr<Centroid> pt2 =
486 4 : boost::get<boost::shared_ptr<Centroid> >(kt->second);
487 4 : Centroid_add(pt.get(),
488 : Centroid_get_mean(pt2.get()),
489 : Centroid_get_count(pt2.get()));
490 4 : }
491 0 : } catch (boost::bad_get& ex) {
492 0 : QE_ASSERT(0);
493 : }
494 : }
495 209 : if (jt->first.first == QEOpServerProxy::PERCENTILES) {
496 2 : StatsSelect::StatVal & sv = jt->second;
497 : try {
498 2 : if (sv.which() == QEOpServerProxy::TDIGEST) {
499 : boost::shared_ptr<TDigest>& pt =
500 2 : boost::get<boost::shared_ptr<TDigest> >(jt->second);
501 :
502 : boost::shared_ptr<TDigest> pt2 =
503 2 : boost::get<boost::shared_ptr<TDigest> >(kt->second);
504 2 : size_t j, ncentroids = TDigest_get_ncentroids(pt2.get());
505 4 : for (j = 0; j < ncentroids; j++) {
506 2 : Centroid * c = TDigest_get_centroid(pt2.get(), j);
507 2 : TDigest* nd = TDigest_add(pt.get(),
508 : Centroid_get_mean(c),
509 : Centroid_get_count(c));
510 2 : if (nd) {
511 0 : pt.reset(nd);
512 : }
513 : }
514 2 : }
515 0 : } catch (boost::bad_get& ex) {
516 0 : QE_ASSERT(0);
517 : }
518 : }
519 : }
520 : }
521 : }
522 :
523 : namespace boost {
524 : std::size_t hash_value(const StatsSelect::StatVal&);
525 : }
526 :
527 1822 : std::size_t boost::hash_value(const StatsSelect::StatVal& sv) {
528 1822 : std::ostringstream ostr;
529 1822 : ostr << sv;
530 3644 : return boost::hash_value(ostr.str());
531 1822 : }
532 :
533 535 : bool StatsSelect::LoadRow(boost::uuids::uuid u,
534 : uint64_t timestamp, const vector<StatEntry>& row, MapBufT& output) {
535 :
536 535 : if (!Status()) return false;
537 535 : uint64_t ts = 0;
538 :
539 : // Build Uniks map
540 535 : StatMap uniks;
541 : set<string>::const_iterator ukit =
542 535 : unik_cols_.find(g_viz_constants.STAT_UUID_FIELD);
543 535 : if (ukit!=unik_cols_.end()) {
544 63 : uniks.insert(make_pair(g_viz_constants.STAT_UUID_FIELD,u));
545 : }
546 :
547 535 : if (isT_) {
548 50 : ts = timestamp;
549 50 : uniks.insert(make_pair(g_viz_constants.STAT_TIME_FIELD,ts));
550 : }
551 :
552 535 : if (ts_period_) {
553 54 : ts = timestamp - (timestamp % ts_period_);
554 54 : uniks.insert(make_pair(g_viz_constants.STAT_TIMEBIN_FIELD,ts));
555 : }
556 :
557 535 : for (vector<StatEntry>::const_iterator it = row.begin();
558 18092 : it != row.end(); it++) {
559 17558 : set<string>::const_iterator uit = unik_cols_.find(it->name);
560 17559 : if (uit!=unik_cols_.end()) {
561 1633 : uniks.insert(make_pair(it->name, it->value));
562 : }
563 : }
564 :
565 : // Build sort vector
566 : // Last slot is reserved for the hash
567 535 : std::vector<StatVal> ukey(sort_cols_.size() + agg_sort_cols_.size() + 1);
568 535 : size_t hash_slot = sort_cols_.size() + agg_sort_cols_.size();
569 535 : uint64_t hash_val = boost::hash_range(uniks.begin(), uniks.end());
570 535 : ukey[hash_slot] = hash_val;
571 :
572 535 : for (map<string, size_t>::const_iterator st = sort_cols_.begin();
573 666 : st!=sort_cols_.end(); st++) {
574 131 : QE_ASSERT(uniks.find(st->first) != uniks.end());
575 131 : ukey[st->second] = uniks.at(st->first);
576 : }
577 :
578 535 : QEOpServerProxy::AggRowT narows;
579 535 : for (vector<StatEntry>::const_iterator it = row.begin();
580 18122 : it != row.end(); it++) {
581 17587 : set<string>::const_iterator uit = sum_cols_.find(it->name);
582 17591 : if (uit!=sum_cols_.end()) {
583 271 : pair<QEOpServerProxy::AggOper,string> aggkey(QEOpServerProxy::SUM,it->name);
584 271 : narows.insert(make_pair(aggkey, it->value));
585 271 : }
586 : }
587 :
588 535 : for (vector<StatEntry>::const_iterator it = row.begin();
589 18115 : it != row.end(); it++) {
590 17580 : set<string>::const_iterator uit = max_field_.find(it->name);
591 17582 : if (uit!=max_field_.end()) {
592 3 : pair<QEOpServerProxy::AggOper,string> aggkey(QEOpServerProxy::MAX,it->name);
593 3 : narows.insert(make_pair(aggkey, it->value));
594 3 : }
595 17582 : uit = min_field_.find(it->name);
596 17591 : if (uit!=min_field_.end()) {
597 3 : pair<QEOpServerProxy::AggOper,string> aggkey(QEOpServerProxy::MIN,it->name);
598 3 : narows.insert(make_pair(aggkey, it->value));
599 3 : }
600 17587 : uit = avg_field_.find(it->name);
601 17594 : if (uit!=avg_field_.end()) {
602 6 : pair<QEOpServerProxy::AggOper,string> aggkey(QEOpServerProxy::AVG,it->name);
603 : try {
604 6 : Centroid *t = Centroid_create(0, 0);
605 6 : StatsSelect::StatVal sv = it->value;
606 : double val;
607 6 : if (sv.which() == QEOpServerProxy::UINT64) {
608 3 : val = boost::get<uint64_t>(sv);
609 : } else {
610 3 : val = boost::get<double>(sv);
611 : }
612 6 : Centroid_add(t, val, 1);
613 6 : boost::shared_ptr<Centroid> pt(t, &StatsSelect::DeleteCentroid);
614 6 : narows.insert(make_pair(aggkey, pt));
615 6 : } catch (boost::bad_get& ex) {
616 0 : QE_ASSERT(0);
617 0 : } catch (const std::out_of_range& oor) {
618 0 : QE_ASSERT(0);
619 : }
620 6 : }
621 17587 : uit = percentile_cols_.find(it->name);
622 17584 : if (uit!=percentile_cols_.end()) {
623 3 : pair<QEOpServerProxy::AggOper,string> aggkey(QEOpServerProxy::PERCENTILES,it->name);
624 :
625 : try {
626 3 : TDigest *t = TDigest_create(0.01, 100);
627 3 : StatsSelect::StatVal sv = it->value;
628 : double val;
629 3 : if (sv.which() == QEOpServerProxy::UINT64) {
630 3 : val = boost::get<uint64_t>(sv);
631 : } else {
632 0 : val = boost::get<double>(sv);
633 : }
634 3 : TDigest_add(t, val, 1);
635 3 : boost::shared_ptr<TDigest> pt(t, &StatsSelect::DeleteTDigest);
636 3 : narows.insert(make_pair(aggkey, pt));
637 3 : } catch (boost::bad_get& ex) {
638 0 : QE_ASSERT(0);
639 0 : } catch (const std::out_of_range& oor) {
640 0 : QE_ASSERT(0);
641 : }
642 3 : }
643 : }
644 :
645 535 : for (std::set<std::string>::const_iterator ct = class_cols_.begin();
646 544 : ct!=class_cols_.end(); ct++) {
647 9 : pair<QEOpServerProxy::AggOper,string> aggkey(QEOpServerProxy::CLASS,*ct);
648 9 : StatMap huniks;
649 9 : for (vector<StatEntry>::const_iterator rit = row.begin();
650 90 : rit != row.end(); rit++) {
651 81 : if (rit->name != *ct) {
652 81 : if (uniks.find(rit->name) != uniks.end()) {
653 : // For generating the hash, consider all attributes that
654 : // are in the row, and that do not match the CLASS attribute,
655 : // and that are in non-aggregate attributes in the SELECT
656 18 : huniks[rit->name] = rit->value;
657 : }
658 : }
659 : }
660 9 : uint64_t hh = boost::hash_range(huniks.begin(), huniks.end());
661 9 : narows.insert(make_pair(aggkey, hh));
662 9 : }
663 :
664 535 : if (!count_field_.empty()) {
665 30 : pair<QEOpServerProxy::AggOper,string> aggkey(QEOpServerProxy::COUNT,count_field_);
666 30 : narows.insert(make_pair(aggkey, (uint64_t) 1));
667 30 : }
668 :
669 535 : MergeFullRow(ukey, uniks, narows, output);
670 :
671 535 : return true;
672 535 : }
673 :
|
