sensordatastore.cpp

/*
 * sensordatastore.cpp
 *
 *  Created on: Jul 24, 2013
 *      Author: Axel Auweter
 */

/**
 * @mainpage
 * The DCDBLib library is a dynamic runtime library providing
 * functions to initialize and access the DCDB data store. It
 * is being used by the CollectAgent to handle insertion of
 * data and can be used by tools responsible for data analysis.
 *
 * Its main class is the SensorDataStore class which provides
 * functions to connect to the data store, initialize an empty
 * data base and to (TODO) retrieve data.
 *
 * For its internal handling, SensorDataStore relies on the
 * SensorDataStoreImpl class (which hides all private member
 * functions belonging to the SensorDataStore class from the
 * header that is used by programmers who link against this
 * library). Raw database functionality is abstracted into the
 * CassandraBackend class (to easy switching to other
 * key-value style databases in the future).
 *
 * To use the library in your client application, simply
 * include the sensordatastore.h header file and initialize
 * an object of the SensorDataStore class.
 */

#include <string>
#include <iostream>
#include <cstdint>
#include <cinttypes>

#include "cassandra.h"

#include "sensordatastore.h"
#include "sensordatastore_internal.h"
#include "connection.h"
#include "dcdbglobals.h"

#include "dcdbendian.h"

/**
 * @details
 * Since we want high-performance inserts, we prepare the
 * insert CQL query in advance and only bind it on the actual
 * insert.
 */
void SensorDataStoreImpl::prepareInsert(uint64_t ttl)
{
  CassError rc = CASS_OK;
  CassFuture* future = NULL;
  const char* query;

  /*
   * Free the old prepared if necessary.
   */
  if (preparedInsert) {
      cass_prepared_free(preparedInsert);
  }

  char *queryBuf = NULL;
  if (ttl == 0) {
      query = "INSERT INTO dcdb.sensordata (sid, ts, value) VALUES (?, ?, ?);";
  }
  else {
      queryBuf = (char*)malloc(256);
      snprintf(queryBuf, 256, "INSERT INTO dcdb.sensordata (sid, ts, value) VALUES (?, ?, ?) USING TTL %" PRIu64 " ;", ttl);
      query = queryBuf;
  }

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

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
      connection->printError(future);
  } else {
      preparedInsert = cass_future_get_prepared(future);
  }

  cass_future_free(future);
  if (queryBuf) {
      free(queryBuf);
  }
}

/**
 * @details
 * To insert a sensor reading, the Rsvd field of the SensorId must
 * be filled with a time component that ensures that the maximum
 * number of 2^32 columns per key is not exceeded while still
 * allowing relatively easy retrieval of data.
 *
 * We achieve this by using a "week-stamp" (i.e. number of weeks
 * since Unix epoch) within the Rsvd field of the SensorId before
 * calling the Cassandra Backend to do the raw insert.
 *
 * Applications should not call this function directly, but
 * use the insert function provided by the SensorDataStore class.
 */
void SensorDataStoreImpl::insert(SensorId* sid, uint64_t ts, uint64_t value)
{
#if 0
  std::cout << "Inserting@SensorDataStoreImpl (" << sid->raw[0] << " " << sid->raw[1] << ", " << ts << ", " << value << ")" << std::endl;
#endif

  /* Calculate and insert week number */
  uint16_t week = ts / 604800000000000;
  sid->setRsvd(week);

  /* Insert into Cassandra */
  std::string key = sid->serialize();

  CassError rc = CASS_OK;
  CassStatement* statement = NULL;
  CassFuture *future = NULL;

  statement = cass_prepared_bind(preparedInsert);

  cass_statement_bind_bytes_by_name(statement, "sid", (cass_byte_t*)(key.c_str()), 16);
  cass_statement_bind_int64_by_name(statement, "ts", ts);
  cass_statement_bind_int64_by_name(statement, "value", value);

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

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
      connection->printError(future);
  }

  cass_future_free(future);
  cass_statement_free(statement);
}

/**
 * @details
 * This function updates the prepared statement for inserts
 * with the new TTL value.
 */
void SensorDataStoreImpl::setTTL(uint64_t ttl)
{
  prepareInsert(ttl);
}

/**
 * @details
 * This function issues a regular query to the data store
 * and creates a SensorDataStoreReading object for each
 * entry which is stored in the result list.
 */
void SensorDataStoreImpl::query(std::list<SensorDataStoreReading>& result, SensorId sid, DCDBTimeStamp start, DCDBTimeStamp end)
{
  CassError rc = CASS_OK;
  CassStatement* statement = NULL;
  CassFuture *future = NULL;
  const CassPrepared* prepared = nullptr;
  const char* query = "SELECT * FROM " KEYSPACE_NAME "." CF_SENSORDATA " WHERE sid = ? AND ts >= ? AND ts <= ? ;";

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

  rc = cass_future_error_code(future);
  if (rc != CASS_OK) {
    connection->printError(future);
    cass_future_free(future);
    return;
  }

  prepared = cass_future_get_prepared(future);
  cass_future_free(future);

  std::string key = sid.serialize();

#if 0
  std::cout << "Query: " << query << std::endl << "sid: " << key << " ts1: " << start.getRaw() << " ts2: " << end.getRaw() << std::endl;
#endif

  statement = cass_prepared_bind(prepared);
  cass_statement_bind_bytes(statement, 0, (const cass_byte_t*)(key.c_str()), 16);
  cass_statement_bind_int64(statement, 1, start.getRaw());
  cass_statement_bind_int64(statement, 2, end.getRaw());

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

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

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

          SensorDataStoreReading entry;

          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);

          entry.sensorId = sid;
          entry.timeStamp = DCDBTimeStamp((uint64_t)ts);
          entry.value = (uint64_t)value;

          result.push_back(entry);
#if 0
          if (localtime) {
              t.convertToLocal();
          }
          if (raw) {
              std::cout << sensorName << "," << std::dec << t.getRaw() << "," << std::dec << value << std::endl;
          }
          else {
              std::cout << sensorName << "," << t.getString() << "," << std::dec << value << std::endl;
          }
#endif
      }
      cass_iterator_free(rows);
      cass_result_free(cresult);
  }

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

/**
 * @details
 * This function generates an integrated value of the time series
 * by first querying for the result set list using query() and then
 * summing up the result.
 */
SDSQueryResult SensorDataStoreImpl::querySum(int64_t& result, SensorId sid, DCDBTimeStamp start, DCDBTimeStamp end)
{
  std::list<SensorDataStoreReading> queryResult;

  /* Issue a standard query */
  query(queryResult, sid, start, end);

  /* Check if at least 2 readings in result */
  if (queryResult.size() < 2)
    return SDS_EMPTYSET;

  /* Integrate the result */
  result = 0;

  SensorDataStoreReading prev;
  for (std::list<SensorDataStoreReading>::iterator it = queryResult.begin(); it != queryResult.end(); it++) {
      if (!(it == queryResult.begin())) {
          SensorDataStoreReading cur = *it;

          /* Calculate average between two readings */
          int64_t avg = (cur.value + prev.value) / 2;

          /* Calculate time difference */
          uint64_t dt = cur.timeStamp.getRaw() - prev.timeStamp.getRaw();

          /* Sum up (with lousy attempt to keep it numerically stable - should probably use double instead) */
          if (dt > 10000000000) {
              /* dt > 10s => convert dt to s first */
              dt /= 1000000000;
              result += avg * dt;
          }
          else {
              /* dt < 10s => multiply first */
              avg *= dt;
              result += avg / 1000000000;
          }
      }
      prev = *it;
  }
  return SDS_OK;
}

/**
 * @details
 * This constructor sets the internal connection variable to
 * the externally provided DCDBConnection object and also
 * retrieves the CassSession pointer of the connection.
 */
SensorDataStoreImpl::SensorDataStoreImpl(DCDBConnection* conn)
{
  connection = conn;
  session = connection->getSessionHandle();

  preparedInsert = nullptr;
  prepareInsert(0);
}

/**
 * @details
 * Due to the simplicity of the class, the destructor is left empty.
 */
SensorDataStoreImpl::~SensorDataStoreImpl()
{
  connection = nullptr;
  session = nullptr;
  if (preparedInsert) {
      cass_prepared_free(preparedInsert);
  }
}

/**
 * @details
 * Instead of doing the actual work, this function simply
 * forwards to the insert function of the SensorDataStoreImpl
 * class.
 */
void SensorDataStore::insert(SensorId* sid, uint64_t ts, uint64_t value)
{
    impl->insert(sid, ts, value);
}

/**
 * @details
 * Instead of doing the actual work, this function simply
 * forwards to the insert function of the SensorDataStoreImpl
 * class.
 */
void SensorDataStore::setTTL(uint64_t ttl)
{
    impl->setTTL(ttl);
}

/**
 * @details
 * Instead of doing the actual work, this function simply
 * forwards to the insert function of the SensorDataStoreImpl
 * class.
 */
void SensorDataStore::query(std::list<SensorDataStoreReading>& result, SensorId sid, DCDBTimeStamp start, DCDBTimeStamp end)
{
    impl->query(result, sid, start, end);
}

/**
 * @details
 * Instead of doing the actual work, this function simply
 * forwards to the insert function of the SensorDataStoreImpl
 * class.
 */
SDSQueryResult SensorDataStore::querySum(int64_t& result, SensorId sid, DCDBTimeStamp start, DCDBTimeStamp end)
{
    return impl->querySum(result, sid, start, end);
}

/**
 * @details
 * This constructor allocates the implementation class which
 * holds the actual implementation of the class functionality.
 */
SensorDataStore::SensorDataStore(DCDBConnection* conn)
{
    impl = new SensorDataStoreImpl(conn);
}

/**
 * @details
 * The SensorDataStore desctructor deallocates the
 * SensorDataStoreImpl and CassandraBackend objects.
 */
SensorDataStore::~SensorDataStore()
{
  /* Clean up... */
  if (impl)
    delete impl;
}