query.cpp 10.7 KB
Newer Older
Axel Auweter's avatar
Axel Auweter committed
1
2
3
4
5
6
7
8
/*
 * query.cpp
 *
 *  Created on: Feb 19, 2015
 *      Author: Axel Auweter
 */

#include <iostream>
9
10
#include <iomanip>
#include <sstream>
Axel Auweter's avatar
Axel Auweter committed
11
12
13
14
15
#include <list>
#include <string>
#include <algorithm>

#include <cstdlib>
16
#include <cinttypes>
Axel Auweter's avatar
Axel Auweter committed
17

18
19
20
#include <boost/algorithm/string.hpp>

#include "dcdbendian.h"
Axel Auweter's avatar
Axel Auweter committed
21
22
23
24
#include "query.h"
#include "casshelper.h"

/* Lovely spaghetti code coming up next. Be aware... */
25
void DCDBQuery::doQuery(const char* hostname, std::list<std::string> sensors, DCDBTimeStamp start, DCDBTimeStamp end, bool raw, bool localtime)
Axel Auweter's avatar
Axel Auweter committed
26
27
28
29
30
31
32
33
34
35
36
37
{
  /* Connect to db */
  CassCluster *cluster = CassHelper::create_cluster(hostname);
  CassSession* session = cass_session_new();

  if (CassHelper::connect_session(session, cluster) != CASS_OK) {
     cass_session_free(session);
     cass_cluster_free(cluster);
     std::cout << "Cannot connect to Cassandra database." << std::endl;
     exit(EXIT_FAILURE);
  }

38
39
40
41
  /* Print the CSV header */
  std::cout << "Sensor,Time,Value" << std::endl;

  /* Iterate over list of sensors requested by the user */
Axel Auweter's avatar
Axel Auweter committed
42
  for (std::list<std::string>::iterator it = sensors.begin(); it != sensors.end(); it++) {
43
44
45
      /* Lookup the sensor in the published sensors table */
      std::string sensorPattern;
      lookupPublishedSensorPattern(session, *it,sensorPattern);
Axel Auweter's avatar
Axel Auweter committed
46
47

      std::list<SensorId> sidList;
48
      expandSensorPattern(session, sensorPattern, start, end, sidList);
49
50
51
52
53
54
55

#if 0
      for (std::list<SensorId>::iterator sit = sidList.begin(); sit != sidList.end(); sit++) {
          std::cout << std::hex << std::setfill('0') << std::setw(16) << (*sit).raw[0]
                    << " " << std::hex << std::setfill('0') << std::setw(16) << (*sit).raw[1] << std::endl;
      }
#endif
Axel Auweter's avatar
Axel Auweter committed
56

57
      querySensorsCSV(session, *it, sidList, start, end, raw, localtime);
Axel Auweter's avatar
Axel Auweter committed
58
59
60
61
62
63
64
65
  }

  /* Clean up */
  cass_session_free(session);
  cass_cluster_free(cluster);

}

66
void DCDBQuery::lookupPublishedSensorPattern(CassSession* session, std::string name, std::string& pattern)
Axel Auweter's avatar
Axel Auweter committed
67
68
69
70
71
72
{
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const CassPrepared* prepared = nullptr;

73
  CassString query = cass_string_init("SELECT pattern FROM dcdb_config.publishedsensors WHERE name = ? ;");
Axel Auweter's avatar
Axel Auweter committed
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
  CassString pattern_cstr;

  future = cass_session_prepare(session, query);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    CassHelper::print_error(future);
  } else {
    prepared = cass_future_get_prepared(future);
  }

  statement = cass_prepared_bind(prepared);
  cass_statement_bind_string_by_name(statement, "name", cass_string_init(name.c_str()));

  future = cass_session_execute(session, statement);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
      CassHelper::print_error(future);
  } else {
      const CassResult* result = cass_future_get_result(future);
      CassIterator* iterator = cass_iterator_from_result(result);

      if (cass_iterator_next(iterator)) {
          const CassRow* row = cass_iterator_get_row(iterator);
          cass_value_get_string(cass_row_get_column_by_name(row, "pattern"), &pattern_cstr);
          pattern = std::string(pattern_cstr.data, pattern_cstr.length);
      }
      else {
          std::cout << "Unknown sensor: " << name << std::endl;
          exit(EXIT_FAILURE);
      }

      cass_result_free(result);
      cass_iterator_free(iterator);
  }
}

114
void DCDBQuery::expandSensorPattern(CassSession* session, std::string sensorPattern, DCDBTimeStamp start, DCDBTimeStamp end, std::list<SensorId>& sensorIds)
Axel Auweter's avatar
Axel Auweter committed
115
116
117
118
{
  /* Clear the list of sensorIds */
  sensorIds.clear();

119
120
  /* Strip all slashes from publishedSensorName */
  sensorPattern.erase(std::remove(sensorPattern.begin(), sensorPattern.end(), '/'), sensorPattern.end());
Axel Auweter's avatar
Axel Auweter committed
121

122
123
124
125
126
127
  /* Calculate lower and upper boundaries for the expansion of the pattern */
  std::string low  = sensorPattern;
  std::string high = sensorPattern;
  if (sensorPattern.find("*") != std::string::npos) {
      low.replace(sensorPattern.find("*"), 1, 33-sensorPattern.length(), '0');
      high.replace(sensorPattern.find("*"), 1, 33-sensorPattern.length(), 'F');
Axel Auweter's avatar
Axel Auweter committed
128
129
  }

130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
  SensorId lowId, highId;
  if (!topicToSid(lowId, low)) {
      std::cout << "Internal error" << std::endl;
      exit(EXIT_FAILURE);
  }
  if (!topicToSid(highId, high)) {
      std::cout << "Internal error" << std::endl;
      exit(EXIT_FAILURE);
  }
  lowId.dsid.rsvd = start.getWeekstamp();
  highId.dsid.rsvd = end.getWeekstamp();

  low = sidConvert(lowId);
  high = sidConvert(highId);

//  std::cout << "Lower boundary for sensor scan: " << std::hex << std::setfill('0') << std::setw(16) << lowId.raw[0] << " " << std::hex << std::setfill('0') << std::setw(16) << lowId.raw[1] << std::endl;
//  std::cout << "Upper boundary for sensor scan: " << std::hex << std::setfill('0') << std::setw(16) << highId.raw[0] << " " << std::hex << std::setfill('0') << std::setw(16) << highId.raw[1] << std::endl;

Axel Auweter's avatar
Axel Auweter committed
148
149
150
151
152
153
  /* Query the database to see which raw sensors actually exist in the interval between low and high */
  CassError rc = CASS_OK;
  CassStatement* statement = nullptr;
  CassFuture* future = nullptr;
  const CassPrepared* prepared = nullptr;

154
  CassString query = cass_string_init("SELECT DISTINCT sid FROM dcdb.sensordata WHERE TOKEN(sid) >= ? and TOKEN(sid) <= ?;");
Axel Auweter's avatar
Axel Auweter committed
155
156
157
158
159
160
161
162
163
164
165
166

  future = cass_session_prepare(session, query);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    CassHelper::print_error(future);
  } else {
    prepared = cass_future_get_prepared(future);
  }

  statement = cass_prepared_bind(prepared);
167
168
169
170
171
172

  CassBytes lowCb = cass_bytes_init((cass_byte_t*)(low.c_str()), 16);
  CassBytes highCb = cass_bytes_init((cass_byte_t*)(high.c_str()), 16);

  cass_statement_bind_bytes(statement, 0, lowCb);
  cass_statement_bind_bytes(statement, 1, highCb);
Axel Auweter's avatar
Axel Auweter committed
173
174
175
176
177
178
179
180
181
182
183

  future = cass_session_execute(session, statement);
  cass_future_wait(future);

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
      CassHelper::print_error(future);
  } else {
      const CassResult* result = cass_future_get_result(future);
      CassIterator* iterator = cass_iterator_from_result(result);

184
      while (cass_iterator_next(iterator)) {
Axel Auweter's avatar
Axel Auweter committed
185
          const CassRow* row = cass_iterator_get_row(iterator);
186
187
188
189
190
191
192
          CassBytes res;
          cass_value_get_bytes(cass_row_get_column_by_name(row, "sid"), &res);
          SensorId sensor;
          sensor.raw[0] = Endian::BEToHost(((uint64_t*)res.data)[0]);
          sensor.raw[1] = Endian::BEToHost(((uint64_t*)res.data)[1]);

          /* Check if the sensorId matches the pattern */
193
          if (sidPatternMatch(sensor, sensorPattern)) {
194
195
            sensorIds.push_back(sensor);
          }
Axel Auweter's avatar
Axel Auweter committed
196
197
198
199
200
      }

      cass_result_free(result);
      cass_iterator_free(iterator);
  }
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281

  cass_future_free(future);
  cass_statement_free(statement);
  cass_prepared_free(prepared);
}

/* Ugly copy of DCDBLib's topicToSid - It's really time to clean up the API */
bool DCDBQuery::topicToSid(SensorId& sid, std::string topic)
{
  uint64_t pos = 0;
  const char* buf = topic.c_str();
  sid.raw[0] = 0;
  sid.raw[1] = 0;
  while (*buf && pos < 128) {
      if (*buf >= '0' && *buf <= '9') {
          sid.raw[pos / 64] |= (((uint64_t)(*buf - '0')) << (60-(pos%64)));
          pos += 4;
      }
      else if (*buf >= 'A' && *buf <= 'F') {
          sid.raw[pos / 64] |= (((uint64_t)(*buf - 'A' + 0xa)) << (60-(pos%64)));
          pos += 4;
      }
      else if (*buf >= 'a' && *buf <= 'f') {
          sid.raw[pos / 64] |= (((uint64_t)(*buf - 'a' + 0xa)) << (60-(pos%64)));
          pos += 4;
      }
      buf++;
  }
  return pos == 128;
}

/* Ugly copy of DCDBLib's sidConvert - It's really time to clean up the API */
std::string DCDBQuery::sidConvert(SensorId& sid)
{
    uint64_t ll[2];
    ll[0] = Endian::hostToBE(sid.raw[0]);
    ll[1] = Endian::hostToBE(sid.raw[1]);
    return std::string((char*)ll, 16);
}

bool DCDBQuery::sidPatternMatch(SensorId& sid, std::string pattern)
{
  /* Strip all slashes from pattern */
  pattern.erase(std::remove(pattern.begin(), pattern.end(), '/'), pattern.end());

  /* Convert to lower case */
  boost::algorithm::to_lower(pattern);

  /* Calculate the wildcard length */
  int wl = (pattern.find("*") != std::string::npos) ? 33 - pattern.length() : 0;

  /* Do a character by character comparison */
  int posP = 0;
  int posS = 0;
  while (posS < 32) {
      char p, cs[2];
      uint64_t s;
      p = pattern.c_str()[posP];
      s = sid.raw[posS / 16];
      s >>= (60-((4*posS)%64));
      s &= 0xf;
      snprintf(cs, 2, "%" PRIx64, s);

      if (p == '*') {
          /* Jump over the wildcard */
          posS += wl;
          posP++;
          continue;
      }
      else if ((posS >= 24 && posS <= 27) || (p==cs[0])) {
          posS++;
          posP++;
          continue;
      }
      else {
          return false;
      }
  }
  return true;
}

282
void DCDBQuery::querySensorsCSV(CassSession* session, std::string sensorName, std::list<SensorId>& sidList, DCDBTimeStamp& start, DCDBTimeStamp& end, bool raw, bool localtime)
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
{
  /* Since everything has been mangled until here, we assume parameter safety and do a non-prepared statement... Oh oh... */
  std::stringstream query;
  query << "SELECT * FROM dcdb.sensordata WHERE sid IN (";
  for (std::list<SensorId>::iterator it = sidList.begin(); it != sidList.end(); ++it) {
      if (it != sidList.begin()) {
          query << ", ";
      }
      query << "0x";
      query << std::hex << std::setfill('0') << std::setw(16) << (*it).raw[0]
            << std::hex << std::setfill('0') << std::setw(16) << (*it).raw[1];
  }
  query << ") AND ts >= ";
  query << std::dec << start.getRaw();
  query << " AND ts <= ";
  query << std::dec << end.getRaw();
  query << ";";

  CassString csQuery = cass_string_init(query.str().c_str());
  CassStatement* statement = cass_statement_new(csQuery, 0);
  CassFuture* future = cass_session_execute(session, statement);

  if (cass_future_error_code(future) == CASS_OK) {
      const CassResult* result = cass_future_get_result(future);
      CassIterator* rows = cass_iterator_from_result(result);

      while (cass_iterator_next(rows)) {
          const CassRow* row = cass_iterator_get_row(rows);

          cass_int64_t ts, value;
          cass_value_get_int64(cass_row_get_column_by_name(row, "ts"), &ts);
          cass_value_get_int64(cass_row_get_column_by_name(row, "value"), &value);

316
317
318
319
          DCDBTimeStamp t((uint64_t)ts);
          if (localtime) {
              t.convertToLocal();
          }
320
          if (raw) {
321
              std::cout << sensorName << "," << std::dec << t.getRaw() << "," << std::dec << value << std::endl;
322
323
324
325
          }
          else {
              std::cout << sensorName << "," << t.getString() << "," << std::dec << value << std::endl;
          }
326
327
328
329
      }
  }
  cass_statement_free(statement);
  cass_future_free(future);
Axel Auweter's avatar
Axel Auweter committed
330
}