cql_exchange.cc

Go to the documentation of this file.

// Copyright (C) 2018-2020 Internet Systems Consortium, Inc. ("ISC")
// Copyright (C) 2016-2017 Deutsche Telekom AG.
//
// Authors: Razvan Becheriu <razvan.becheriu@qualitance.com>
//          Andrei Pavel <andrei.pavel@qualitance.com>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//           http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

#include <config.h>

#include <cql/cql_connection.h>
#include <cql/cql_exchange.h>
#include <cql/sql_common.h>
#include <database/db_exceptions.h>

#include <boost/multi_index/hashed_index.hpp>
#include <boost/multi_index/member.hpp>
#include <boost/multi_index/sequenced_index.hpp>
#include <boost/multi_index_container.hpp>
#include <boost/noncopyable.hpp>
#include <boost/shared_ptr.hpp>

#include <fstream>
#include <iostream>
#include <map>
#include <string>
#include <utility>
#include <vector>

namespace isc {
namespace db {

#define KEA_CASS_CHECK(cass_error)   \
    {                                \
        if (cass_error != CASS_OK) { \
            return cass_error;       \
        }                            \
    }

struct ExchangeDataTypeHash {
public:
    size_t operator()(const ExchangeDataType& key) const {
        return std::hash<size_t>{}(static_cast<size_t>(key));
    }
};

typedef std::unordered_map<ExchangeDataType, CqlFunction, ExchangeDataTypeHash>
    CqlFunctionMap;
extern CqlFunctionMap CQL_FUNCTIONS;

std::size_t
hash_value(const CassValueType& key) {
    return key;
}


typedef std::unordered_map<std::type_index, ExchangeDataType> AnyTypeMap;

// Declare uint8_t as key here for compatibility with g++ version 5. Ideally,
// it would be CassValueType
typedef std::unordered_map<uint8_t, ExchangeDataType> CassTypeMap;

static AnyTypeMap ANY_TYPE_MAP = {
    {typeid(NULL), EXCHANGE_DATA_TYPE_NONE},
    {typeid(cass_bool_t*), EXCHANGE_DATA_TYPE_BOOL},
    {typeid(cass_int8_t*), EXCHANGE_DATA_TYPE_INT8},
    {typeid(cass_int16_t*), EXCHANGE_DATA_TYPE_INT16},
    {typeid(cass_int32_t*), EXCHANGE_DATA_TYPE_INT32},
    {typeid(cass_int64_t*), EXCHANGE_DATA_TYPE_INT64},
    {typeid(std::string*), EXCHANGE_DATA_TYPE_STRING},
    {typeid(CassBlob*), EXCHANGE_DATA_TYPE_BYTES},
    {typeid(CassUuid*), EXCHANGE_DATA_TYPE_UUID},
    {typeid(Udt*), EXCHANGE_DATA_TYPE_UDT}, // user data type
    {typeid(AnyCollection*), EXCHANGE_DATA_TYPE_COLLECTION}};

static CassTypeMap CASS_TYPE_MAP = {
    {CASS_VALUE_TYPE_CUSTOM, EXCHANGE_DATA_TYPE_UDT},
    {CASS_VALUE_TYPE_ASCII, EXCHANGE_DATA_TYPE_STRING},
    {CASS_VALUE_TYPE_BIGINT, EXCHANGE_DATA_TYPE_INT64},
    {CASS_VALUE_TYPE_BLOB, EXCHANGE_DATA_TYPE_BYTES},
    {CASS_VALUE_TYPE_BOOLEAN, EXCHANGE_DATA_TYPE_BOOL},
    {CASS_VALUE_TYPE_COUNTER, EXCHANGE_DATA_TYPE_INT32},
    {CASS_VALUE_TYPE_DECIMAL, EXCHANGE_DATA_TYPE_INT32},
    {CASS_VALUE_TYPE_DOUBLE, EXCHANGE_DATA_TYPE_INT64},
    {CASS_VALUE_TYPE_FLOAT, EXCHANGE_DATA_TYPE_INT32},
    {CASS_VALUE_TYPE_INT, EXCHANGE_DATA_TYPE_INT32},
    {CASS_VALUE_TYPE_TEXT, EXCHANGE_DATA_TYPE_STRING},
    {CASS_VALUE_TYPE_TIMESTAMP, EXCHANGE_DATA_TYPE_INT64},
    {CASS_VALUE_TYPE_UUID, EXCHANGE_DATA_TYPE_UUID},
    {CASS_VALUE_TYPE_VARCHAR, EXCHANGE_DATA_TYPE_STRING},
    {CASS_VALUE_TYPE_VARINT, EXCHANGE_DATA_TYPE_INT32},
    {CASS_VALUE_TYPE_TIMEUUID, EXCHANGE_DATA_TYPE_INT64},
    {CASS_VALUE_TYPE_INET, EXCHANGE_DATA_TYPE_NONE},
    {CASS_VALUE_TYPE_DATE, EXCHANGE_DATA_TYPE_INT64},
    {CASS_VALUE_TYPE_TIME, EXCHANGE_DATA_TYPE_INT64},
    {CASS_VALUE_TYPE_SMALL_INT, EXCHANGE_DATA_TYPE_INT16},
    {CASS_VALUE_TYPE_TINY_INT, EXCHANGE_DATA_TYPE_INT8},
    {CASS_VALUE_TYPE_LIST, EXCHANGE_DATA_TYPE_COLLECTION},
    {CASS_VALUE_TYPE_MAP, EXCHANGE_DATA_TYPE_COLLECTION},
    {CASS_VALUE_TYPE_SET, EXCHANGE_DATA_TYPE_COLLECTION},
    {CASS_VALUE_TYPE_UDT, EXCHANGE_DATA_TYPE_UDT},
    {CASS_VALUE_TYPE_TUPLE, EXCHANGE_DATA_TYPE_UDT}};

Udt::Udt(const CqlConnection& connection, const std::string& name)
    : AnyArray(), connection_(connection), name_(name) {
    // Create type.
    cass_data_type_ = cass_keyspace_meta_user_type_by_name(
        connection_.keyspace_meta_, name_.c_str());
    if (!cass_data_type_) {
        isc_throw(DbOperationError,
                  "Udt::Udt(): UDT " << name_ << " does not exist ");
    }
    // Create container.
    cass_user_type_ = cass_user_type_new_from_data_type(cass_data_type_);
    if (!cass_user_type_) {
        isc_throw(DbOperationError,
                  "Udt::Udt(): Type " << name_
                                      << " is not a UDT as expected. ");
    }
}

Udt::~Udt() {
    //
    // Bug: it seems that if there is no call to
    //      cass_user_type_set_*(cass_user_type_), then
    //      cass_user_type_free(cass_user_type_) might SIGSEGV, so we never
    //      free. Udt objects should have application scope though.
    // cass_user_type_free(cass_user_type_);
}

void
AnyArray::add(const boost::any& value) {
    push_back(value);
}

void
AnyArray::remove(const size_t& index) {
    if (size() <= index) {
        isc_throw(BadValue,
                  "AnyArray::remove(): index "
                      << index << " out of bounds: [0, " << (size() - 1)
                      << "]");
    }
    erase(begin() + index);
}

static CassError
CqlBindNone(const boost::any& /* value */,
            const size_t& index,
            CassStatement* statement) {
    return cass_statement_bind_null(statement, index);
}

static CassError
CqlBindBool(const boost::any& value,
            const size_t& index,
            CassStatement* statement) {
    return cass_statement_bind_bool(statement, index,
                                    *boost::any_cast<cass_bool_t*>(value));
}

static CassError
CqlBindInt8(const boost::any& value,
            const size_t& index,
            CassStatement* statement) {
    return cass_statement_bind_int8(statement, index,
                                    *boost::any_cast<cass_int8_t*>(value));
}

static CassError
CqlBindInt16(const boost::any& value,
             const size_t& index,
             CassStatement* statement) {
    return cass_statement_bind_int16(statement, index,
                                     *boost::any_cast<cass_int16_t*>(value));
}

static CassError
CqlBindInt32(const boost::any& value,
             const size_t& index,
             CassStatement* statement) {
    return cass_statement_bind_int32(statement, index,
                                     *boost::any_cast<cass_int32_t*>(value));
}

static CassError
CqlBindInt64(const boost::any& value,
             const size_t& index,
             CassStatement* statement) {
    return cass_statement_bind_int64(statement, index,
                                     *boost::any_cast<cass_int64_t*>(value));
}

static CassError
CqlBindString(const boost::any& value,
              const size_t& index,
              CassStatement* statement) {
    return cass_statement_bind_string(
        statement, index, boost::any_cast<std::string*>(value)->c_str());
}

static CassError
CqlBindBytes(const boost::any& value,
             const size_t& index,
             CassStatement* statement) {
    CassBlob* blob_value = boost::any_cast<CassBlob*>(value);
    return cass_statement_bind_bytes(statement, index, blob_value->data(),
                                     blob_value->size());
}

static CassError
CqlBindUuid(const boost::any& value,
            const size_t& index,
            CassStatement* statement) {
    return cass_statement_bind_uuid(statement, index,
                                    *boost::any_cast<CassUuid*>(value));
}

static CassError
CqlBindUdt(const boost::any& value,
           const size_t& index,
           CassStatement* statement) {
    Udt* udt = boost::any_cast<Udt*>(value);

    if (!udt) {
        isc_throw(BadValue, "Invalid value specified, not an Udt object");
    }

    size_t i = 0u;

    // Let's iterate over all elements in udt and check that we indeed
    // can assign the set function for each specified type.
    for (boost::any& element : *udt) {
        try {
            KEA_CASS_CHECK(
                CQL_FUNCTIONS[exchangeType(element)].cqlUdtSetFunction_(
                    element, i, udt->cass_user_type_));
        } catch (const boost::bad_any_cast& exception) {
            isc_throw(DbOperationError,
                      "CqlCommon::udtSetData(): "
                          << exception.what()
                          << " when binding parameter of type "
                          << element.type().name()
                          << "in UDT with function CQL_FUNCTIONS["
                          << exchangeType(element) << "].cqlUdtSetFunction_");
        }
        ++i;
    }

    return cass_statement_bind_user_type(statement, index,
                                         udt->cass_user_type_);
}

static CassError
CqlBindCollection(const boost::any& value,
                  const size_t& index,
                  CassStatement* statement) {
    AnyCollection* elements = boost::any_cast<AnyCollection*>(value);

    CassCollection* collection =
        cass_collection_new(CASS_COLLECTION_TYPE_SET, elements->size());

    // Iterate over all elements and assign appropriate append function
    // for each.
    for (boost::any& element : *elements) {
        ExchangeDataType type = exchangeType(element);
        KEA_CASS_CHECK(CQL_FUNCTIONS[type].cqlCollectionAppendFunction_(
            element, collection));
    }

    const CassError cass_error =
        cass_statement_bind_collection(statement, index, collection);
    cass_collection_free(collection);

    return cass_error;
}

static CassError
CqlUdtSetNone(const boost::any& /* udt_member */,
              const size_t& position,
              CassUserType* cass_user_type) {
    return cass_user_type_set_null(cass_user_type, position);
}

static CassError
CqlUdtSetBool(const boost::any& udt_member,
              const size_t& position,
              CassUserType* cass_user_type) {
    return cass_user_type_set_bool(cass_user_type, position,
                                   *boost::any_cast<cass_bool_t*>(udt_member));
}

static CassError
CqlUdtSetInt8(const boost::any& udt_member,
              const size_t& position,
              CassUserType* cass_user_type) {
    return cass_user_type_set_int8(cass_user_type, position,
                                   *boost::any_cast<cass_int8_t*>(udt_member));
}

static CassError
CqlUdtSetInt16(const boost::any& udt_member,
               const size_t& position,
               CassUserType* cass_user_type) {
    return cass_user_type_set_int16(
        cass_user_type, position, *boost::any_cast<cass_int16_t*>(udt_member));
}

static CassError
CqlUdtSetInt32(const boost::any& udt_member,
               const size_t& position,
               CassUserType* cass_user_type) {
    return cass_user_type_set_int32(
        cass_user_type, position, *boost::any_cast<cass_int32_t*>(udt_member));
}

static CassError
CqlUdtSetInt64(const boost::any& udt_member,
               const size_t& position,
               CassUserType* cass_user_type) {
    return cass_user_type_set_int64(
        cass_user_type, position, *boost::any_cast<cass_int64_t*>(udt_member));
}

static CassError
CqlUdtSetString(const boost::any& udt_member,
                const size_t& position,
                CassUserType* cass_user_type) {
    return cass_user_type_set_string(
        cass_user_type, position,
        boost::any_cast<std::string*>(udt_member)->c_str());
}

static CassError
CqlUdtSetBytes(const boost::any& udt_member,
               const size_t& position,
               CassUserType* cass_user_type) {
    CassBlob* blob_value = boost::any_cast<CassBlob*>(udt_member);
    return cass_user_type_set_bytes(cass_user_type, position,
                                    blob_value->data(), blob_value->size());
}

static CassError
CqlUdtSetUuid(const boost::any& udt_member,
              const size_t& position,
              CassUserType* cass_user_type) {
    return cass_user_type_set_uuid(cass_user_type, position,
                                   *boost::any_cast<CassUuid*>(udt_member));
}

static CassError
CqlUdtSetUdt(const boost::any& udt_member,
             const size_t& position,
             CassUserType* cass_user_type) {
    return cass_user_type_set_user_type(
        cass_user_type, position,
        boost::any_cast<Udt*>(udt_member)->cass_user_type_);
}

static CassError
CqlUdtSetCollection(const boost::any& udt_member,
                    const size_t& position,
                    CassUserType* cass_user_type) {
    return cass_user_type_set_collection(
        cass_user_type, position, boost::any_cast<CassCollection*>(udt_member));
}

static CassError
CqlCollectionAppendNone(const boost::any& /* value */,
                        CassCollection* /* collection */) {
    return CASS_OK;
}

static CassError
CqlCollectionAppendBool(const boost::any& value, CassCollection* collection) {
    return cass_collection_append_bool(collection,
                                       *boost::any_cast<cass_bool_t*>(value));
}

static CassError
CqlCollectionAppendInt8(const boost::any& value, CassCollection* collection) {
    return cass_collection_append_int8(collection,
                                       *boost::any_cast<cass_int8_t*>(value));
}

static CassError
CqlCollectionAppendInt16(const boost::any& value, CassCollection* collection) {
    return cass_collection_append_int16(collection,
                                        *boost::any_cast<cass_int16_t*>(value));
}

static CassError
CqlCollectionAppendInt32(const boost::any& value, CassCollection* collection) {
    return cass_collection_append_int32(collection,
                                        *boost::any_cast<cass_int32_t*>(value));
}

static CassError
CqlCollectionAppendInt64(const boost::any& value, CassCollection* collection) {
    return cass_collection_append_int64(collection,
                                        *boost::any_cast<cass_int64_t*>(value));
}

static CassError
CqlCollectionAppendString(const boost::any& value, CassCollection* collection) {
    return cass_collection_append_string(
        collection, boost::any_cast<std::string*>(value)->c_str());
}

static CassError
CqlCollectionAppendBytes(const boost::any& value, CassCollection* collection) {
    CassBlob* blob_value = boost::any_cast<CassBlob*>(value);
    return cass_collection_append_bytes(collection, blob_value->data(),
                                        blob_value->size());
}

static CassError
CqlCollectionAppendUuid(const boost::any& value, CassCollection* collection) {
    return cass_collection_append_uuid(collection,
                                       *boost::any_cast<CassUuid*>(value));
}

static CassError
CqlCollectionAppendUdt(const boost::any& value, CassCollection* collection) {
    Udt* udt = boost::any_cast<Udt*>(value);
    size_t i = 0u;
    for (boost::any& element : *udt) {
        KEA_CASS_CHECK(CQL_FUNCTIONS[exchangeType(element)].cqlUdtSetFunction_(
            element, i, udt->cass_user_type_));
        ++i;
    }
    return cass_collection_append_user_type(collection, udt->cass_user_type_);
}

static CassError
CqlCollectionAppendCollection(const boost::any& value,
                              CassCollection* collection) {
    return cass_collection_append_collection(
        collection, boost::any_cast<CassCollection*>(value));
}
// @}

static CassError
CqlGetNone(const boost::any& /* data */, const CassValue* /* value */) {
    return CASS_OK;
}

static CassError
CqlGetBool(const boost::any& data, const CassValue* value) {
    return cass_value_get_bool(value, boost::any_cast<cass_bool_t*>(data));
}

static CassError
CqlGetInt8(const boost::any& data, const CassValue* value) {
    return cass_value_get_int8(value, boost::any_cast<cass_int8_t*>(data));
}

static CassError
CqlGetInt16(const boost::any& data, const CassValue* value) {
    return cass_value_get_int16(value, boost::any_cast<cass_int16_t*>(data));
}

static CassError
CqlGetInt32(const boost::any& data, const CassValue* value) {
    return cass_value_get_int32(value, boost::any_cast<cass_int32_t*>(data));
}

static CassError
CqlGetInt64(const boost::any& data, const CassValue* value) {
    return cass_value_get_int64(value, boost::any_cast<cass_int64_t*>(data));
}

static CassError
CqlGetString(const boost::any& data, const CassValue* value) {
    char const* data_value;
    size_t size_value;
    CassError cass_error = cass_value_get_string(
        value, static_cast<char const**>(&data_value), &size_value);
    boost::any_cast<std::string*>(data)->assign(data_value,
                                                data_value + size_value);
    return cass_error;
}

static CassError
CqlGetBytes(const boost::any& data, const CassValue* value) {
    const cass_byte_t* data_value;
    size_t size_value;
    CassError cass_error = cass_value_get_bytes(
        value, static_cast<const cass_byte_t**>(&data_value), &size_value);
    boost::any_cast<CassBlob*>(data)->assign(data_value,
                                             data_value + size_value);
    return cass_error;
}

static CassError
CqlGetUuid(const boost::any& data, const CassValue* value) {
    return cass_value_get_uuid(value, boost::any_cast<CassUuid*>(data));
}

static CassError
CqlGetUdt(const boost::any& data, const CassValue* value) {
    Udt* udt = boost::any_cast<Udt*>(data);

    CassIterator* fields = cass_iterator_fields_from_user_type(value);
    if (!fields) {
        isc_throw(DbOperationError, "CqlGetUdt(): column is not a UDT");
    }
    Udt::const_iterator it = udt->begin();
    while (cass_iterator_next(fields)) {
        const CassValue* field_value =
            cass_iterator_get_user_type_field_value(fields);
        if (cass_value_is_null(field_value)) {
            isc_throw(DbOperationError,
                      "CqlGetUdt(): null value returned in UDT");
        }
        const CassValueType& type = cass_value_type(field_value);
        KEA_CASS_CHECK(CQL_FUNCTIONS[exchangeType(type)].cqlGetFunction_(
            *it, field_value));
        ++it;
        // If cqlGetFunction_() returns != CASS_OK, don't
        // cass_iterator_free(items_iterator) because we're returning from this
        // function and throwing from the callee.
    }
    cass_iterator_free(fields);
    return CASS_OK;
}

static CassError
CqlGetCollection(const boost::any& data, const CassValue* value) {
    AnyCollection* collection = boost::any_cast<AnyCollection*>(data);
    if (!collection) {
        isc_throw(DbOperationError, "CqlGetCollection(): column is not a collection");
    }

    BOOST_ASSERT(collection->size() == 1);

    boost::any underlying_object = *collection->begin();

    collection->clear();

    CassIterator* items = cass_iterator_from_collection(value);
    if (!items) {
        isc_throw(DbOperationError,
                  "CqlGetCollection(): column is not a collection");
    }
    while (cass_iterator_next(items)) {
        const CassValue* item_value = cass_iterator_get_value(items);
        if (cass_value_is_null(item_value)) {
            isc_throw(DbOperationError,
                      "CqlGetCollection(): null value returned in collection");
        }
        const CassValueType& type = cass_value_type(item_value);

        collection->push_back(underlying_object);
        KEA_CASS_CHECK(CQL_FUNCTIONS[exchangeType(type)].cqlGetFunction_(
            *collection->rbegin(), item_value));
        // If cqlGetFunction_() returns != CASS_OK, don't call
        // cass_iterator_free(items_iterator) because we're returning from this
        // function and throwing from the callee.
    }
    cass_iterator_free(items);
    return CASS_OK;
}

CqlFunctionMap CQL_FUNCTIONS =  //
    {{EXCHANGE_DATA_TYPE_NONE,
      {CqlBindNone, CqlUdtSetNone, CqlCollectionAppendNone, CqlGetNone}},
     {EXCHANGE_DATA_TYPE_BOOL,
      {CqlBindBool, CqlUdtSetBool, CqlCollectionAppendBool, CqlGetBool}},
     {EXCHANGE_DATA_TYPE_INT8,
      {CqlBindInt8, CqlUdtSetInt8, CqlCollectionAppendInt8, CqlGetInt8}},
     {EXCHANGE_DATA_TYPE_INT16,
      {CqlBindInt16, CqlUdtSetInt16, CqlCollectionAppendInt16, CqlGetInt16}},
     {EXCHANGE_DATA_TYPE_INT32,
      {CqlBindInt32, CqlUdtSetInt32, CqlCollectionAppendInt32, CqlGetInt32}},
     {EXCHANGE_DATA_TYPE_INT64,
      {CqlBindInt64, CqlUdtSetInt64, CqlCollectionAppendInt64, CqlGetInt64}},
     {EXCHANGE_DATA_TYPE_STRING,
      {CqlBindString, CqlUdtSetString, CqlCollectionAppendString,
       CqlGetString}},
     {EXCHANGE_DATA_TYPE_BYTES,
      {CqlBindBytes, CqlUdtSetBytes, CqlCollectionAppendBytes, CqlGetBytes}},
     {EXCHANGE_DATA_TYPE_UUID,
      {CqlBindUuid, CqlUdtSetUuid, CqlCollectionAppendUuid, CqlGetUuid}},
     {EXCHANGE_DATA_TYPE_UDT,
      {CqlBindUdt, CqlUdtSetUdt, CqlCollectionAppendUdt, CqlGetUdt}},
     {EXCHANGE_DATA_TYPE_COLLECTION,
      {CqlBindCollection, CqlUdtSetCollection, CqlCollectionAppendCollection,
       CqlGetCollection}}};

ExchangeDataType
exchangeType(const boost::any& object) {
    const std::type_index type = object.type();
    AnyTypeMap::const_iterator exchange_type_it = ANY_TYPE_MAP.find(type);
    if (exchange_type_it == ANY_TYPE_MAP.end()) {
        isc_throw(DbOperationError,
                  "exchangeType(): boost::any type "
                      << type.name() << " does not map to any exchange type");
    }
    const ExchangeDataType exchange_type = exchange_type_it->second;
    if (exchange_type >= CQL_FUNCTIONS.size()) {
        isc_throw(BadValue,
                  "exchangeType(): index " << exchange_type << " out of bounds "
                                           << 0 << " - "
                                           << (CQL_FUNCTIONS.size() - 1));
    }
    return exchange_type;
}

ExchangeDataType
exchangeType(const CassValueType& type) {
    CassTypeMap::const_iterator exchange_type_it = CASS_TYPE_MAP.find(type);
    if (exchange_type_it == CASS_TYPE_MAP.end()) {
        isc_throw(DbOperationError,
                  "exchangeType(): Cassandra value type "
                      << type << " does not map to any exchange type");
    }
    const ExchangeDataType exchange_type = exchange_type_it->second;
    if (exchange_type >= CQL_FUNCTIONS.size()) {
        isc_throw(BadValue,
                  "exchangeType(): index " << exchange_type << " out of bounds "
                                           << 0 << " - "
                                           << CQL_FUNCTIONS.size() - 1);
    }
    return exchange_type;
}

void
CqlCommon::bindData(const AnyArray& data, CassStatement* statement) {
    size_t i = 0u;
    for (const boost::any& element : data) {
        CassError cass_error;
        try {
            cass_error = CQL_FUNCTIONS[exchangeType(element)].cqlBindFunction_(
                element, i, statement);
        } catch (const boost::bad_any_cast& exception) {
            isc_throw(DbOperationError,
                      "CqlCommon::bindData(): "
                          << exception.what() << " when binding parameter " << i
                          << " which is of type " << element.type().name()
                          << " with function CQL_FUNCTIONS["
                          << exchangeType(element) << "].cqlBindFunction_()");
        }
        if (cass_error != CASS_OK) {
            isc_throw(DbOperationError,
                      "CqlCommon::bindData(): unable to bind parameter "
                          << i << " which is of type " << element.type().name()
                          << " with function CQL_FUNCTIONS["
                          << exchangeType(element)
                          << "].cqlBindFunction_(), Cassandra error code: "
                          << cass_error_desc(cass_error));
        }
        ++i;
    }
}

void
CqlCommon::getData(const CassRow* row, AnyArray& data) {
    size_t i = 0u;
    for (boost::any& element : data) {
        const CassValue* value = cass_row_get_column(row, i);
        CassError cass_error;
        try {
            cass_error = CQL_FUNCTIONS[exchangeType(element)].cqlGetFunction_(
                element, value);
        } catch (const boost::bad_any_cast& exception) {
            isc_throw(DbOperationError,
                      "CqlCommon::getData(): "
                          << exception.what() << " when retrieving parameter "
                          << i << " which is of type " << element.type().name()
                          << " with function CQL_FUNCTIONS["
                          << exchangeType(element) << "].cqlGetFunction_()");
        }
        if (cass_error != CASS_OK) {
            isc_throw(
                DbOperationError,
                "CqlCommon::getData(): Cassandra error when retrieving column "
                    << i << ", Cassandra error code: "
                    << cass_error_desc(cass_error));
        }
        ++i;
    }
}

CqlExchange::CqlExchange() {
}

CqlExchange::~CqlExchange() {
}

void
CqlExchange::convertToDatabaseTime(const time_t& cltt,
                                   const uint32_t& valid_lifetime,
                                   cass_int64_t& expire) {
    // Calculate expire time.
    cass_int64_t expire_time = static_cast<cass_int64_t>(cltt) +
            static_cast<cass_int64_t>(valid_lifetime);

    expire = expire_time;
}

void
CqlExchange::convertFromDatabaseTime(const cass_int64_t& expire,
                                     const cass_int64_t& valid_lifetime,
                                     time_t& cltt) {
    // Convert to local time.
    cltt = static_cast<time_t>(expire - valid_lifetime);
}

AnyArray
CqlExchange::executeSelect(const CqlConnection& connection, const AnyArray& data,
                           StatementTag statement_tag, const bool& single /* = false */) {
    CassError rc;
    CassStatement* statement = NULL;
    CassFuture* future = NULL;
    AnyArray local_data = data;

    // Find the query statement first.
    StatementMap::const_iterator it = connection.statements_.find(statement_tag);
    if (it == connection.statements_.end()) {
        isc_throw(DbOperationError,
                  "CqlExchange::executeSelect(): Statement "
                      << statement_tag << "has not been prepared.");
    }

    // Bind the data before the query is executed.
    CqlTaggedStatement tagged_statement = it->second;
    if (tagged_statement.is_raw_) {
        // The entire query is the first element in data.
        std::string* query = boost::any_cast<std::string*>(local_data.back());
        local_data.pop_back();
        statement = cass_statement_new(query->c_str(), local_data.size());
    } else {
        statement = cass_prepared_bind(tagged_statement.prepared_statement_);
        if (!statement) {
            isc_throw(DbOperationError,
                      "CqlExchange::executeSelect(): unable to bind statement "
                          << tagged_statement.name_);
        }
    }

    // Set specific level of consistency if we're told to do so.
    if (connection.force_consistency_) {
        rc = cass_statement_set_consistency(statement, connection.consistency_);
        if (rc != CASS_OK) {
            cass_statement_free(statement);
            isc_throw(DbOperationError,
                      "CqlExchange::executeSelect(): unable to set statement "
                      "consistency for statement "
                          << tagged_statement.name_
                          << ", Cassandra error code: " << cass_error_desc(rc));
        }
        if (connection.serial_consistency_ != CASS_CONSISTENCY_UNKNOWN) {
            rc = cass_statement_set_serial_consistency(statement, connection.serial_consistency_);
            if (rc != CASS_OK) {
                cass_statement_free(statement);
                isc_throw(DbOperationError,
                          "CqlExchange::executeSelect(): unable to set statement "
                          "serial consistency for statement "
                              << tagged_statement.name_
                              << ", Cassandra error code: " << cass_error_desc(rc));
            }
        }
    }

    CqlCommon::bindData(local_data, statement);

    // Everything's ready. Call the actual statement.
    future = cass_session_execute(connection.session_, statement);
    if (!future) {
        cass_statement_free(statement);
        isc_throw(DbOperationError,
                  "CqlExchange::executeSelect(): no CassFuture for statement "
                      << tagged_statement.name_);
    }

    // Wait for the statement execution to complete.
    cass_future_wait(future);
    const std::string error = connection.checkFutureError(
        "CqlExchange::executeSelect(): cass_session_execute() != CASS_OK",
        future, statement_tag);
    rc = cass_future_error_code(future);
    if (rc != CASS_OK) {
        cass_future_free(future);
        cass_statement_free(statement);
        isc_throw(DbOperationError, error);
    }

    // Get column values.
    const CassResult* result_collection = cass_future_get_result(future);
    if (single && cass_result_row_count(result_collection) > 1) {
        cass_result_free(result_collection);
        cass_future_free(future);
        cass_statement_free(statement);
        isc_throw(
            MultipleRecords,
            "CqlExchange::executeSelect(): multiple records were found in "
            "the database where only one was expected for statement "
                << tagged_statement.name_);
    }

    // Get results.
    AnyArray return_values;
    AnyArray collection;
    CassIterator* rows = cass_iterator_from_result(result_collection);
    while (cass_iterator_next(rows)) {
        const CassRow* row = cass_iterator_get_row(rows);
        createBindForSelect(return_values, statement_tag);
        CqlCommon::getData(row, return_values);
        collection.add(retrieve());
    }

    // Free resources.
    cass_iterator_free(rows);
    cass_result_free(result_collection);
    cass_future_free(future);
    cass_statement_free(statement);

    return collection;
}

void
CqlExchange::executeMutation(const CqlConnection& connection, const AnyArray& data,
                             StatementTag statement_tag) {
    CassError rc;
    CassStatement* statement = NULL;
    CassFuture* future = NULL;

    // Find the statement on a list of prepared statements.
    StatementMap::const_iterator it =
        connection.statements_.find(statement_tag);
    if (it == connection.statements_.end()) {
        isc_throw(DbOperationError, "CqlExchange::executeSelect(): Statement "
                  << statement_tag << "has not been prepared.");
    }
    // Bind the statement.
    CqlTaggedStatement tagged_statement = it->second;
    statement = cass_prepared_bind(tagged_statement.prepared_statement_);
    if (!statement) {
        isc_throw(DbOperationError,
                  "CqlExchange::executeMutation(): unable to bind statement "
                      << tagged_statement.name_);
    }

    // Set specific level of consistency, if told to do so.
    if (connection.force_consistency_) {
        rc = cass_statement_set_consistency(statement, connection.consistency_);
        if (rc != CASS_OK) {
            cass_statement_free(statement);
            isc_throw(DbOperationError, "CqlExchange::executeMutation(): unable to set"
                      " statement consistency for statement " << tagged_statement.name_
                      << ", Cassandra error code: " << cass_error_desc(rc));
        }
        if (connection.serial_consistency_ != CASS_CONSISTENCY_UNKNOWN) {
            rc = cass_statement_set_serial_consistency(statement, connection.serial_consistency_);
            if (rc != CASS_OK) {
                cass_statement_free(statement);
                isc_throw(DbOperationError,
                          "CqlExchange::executeMutation(): unable to set statement "
                          "serial consistency for statement "
                              << tagged_statement.name_
                              << ", Cassandra error code: " << cass_error_desc(rc));
            }
        }
    }

    CqlCommon::bindData(data, statement);

    future = cass_session_execute(connection.session_, statement);
    if (!future) {
        cass_statement_free(statement);
        isc_throw(DbOperationError,
                  "CqlExchange::executeMutation(): unable to execute statement "
                      << tagged_statement.name_);
    }
    cass_future_wait(future);
    const std::string error = connection.checkFutureError("CqlExchange::executeMutation():"
                              "cass_session_execute() != CASS_OK", future, statement_tag);
    rc = cass_future_error_code(future);
    if (rc != CASS_OK) {
        cass_future_free(future);
        cass_statement_free(statement);
        isc_throw(DbOperationError, error);
    }

    // Check if statement has been applied.
    bool applied = statementApplied(future);

    // Free resources.
    cass_future_free(future);
    cass_statement_free(statement);

    if (!applied) {
        isc_throw(
            StatementNotApplied,
            "CqlExchange::executeMutation(): [applied] is false for statement "
                << tagged_statement.name_);
    }
}

bool
CqlExchange::statementApplied(CassFuture* future,
                                     size_t* row_count,
                                     size_t* column_count) {
    const CassResult* result_collection = cass_future_get_result(future);
    if (!result_collection) {
        isc_throw(DbOperationError, "CqlExchange::statementApplied(): unable to get"
                      " results collection");
    }
    if (row_count) {
        *row_count = cass_result_row_count(result_collection);
    }
    if (column_count) {
        *column_count = cass_result_column_count(result_collection);
    }
    CassIterator* rows = cass_iterator_from_result(result_collection);
    AnyArray data;
    cass_bool_t applied = cass_true;
    while (cass_iterator_next(rows)) {
        const CassRow* row = cass_iterator_get_row(rows);
        // [applied]: bool
        data.add(&applied);
        CqlCommon::getData(row, data);
    }
    cass_iterator_free(rows);
    cass_result_free(result_collection);
    return applied == cass_true;
}

constexpr StatementTag CqlVersionExchange::GET_VERSION;

StatementMap CqlVersionExchange::tagged_statements_ = {
    {GET_VERSION, {GET_VERSION, "SELECT version, minor FROM schema_version "}}
};

CqlVersionExchange::CqlVersionExchange() {
}

CqlVersionExchange::~CqlVersionExchange() {
}

void
CqlVersionExchange::createBindForSelect(AnyArray& data, StatementTag) {
    data.clear(); // Start with a fresh array.
    data.add(&version_); // first column is a major version
    data.add(&minor_); // second column is a minor version
}

boost::any
CqlVersionExchange::retrieve() {
    pair_ = VersionPair(version_, minor_);
    return &pair_;
}

VersionPair
CqlVersionExchange::retrieveVersion(const CqlConnection& connection) {
    // Run statement.
    const AnyArray where_values;
    AnyArray version_collection =
        executeSelect(connection, where_values, GET_VERSION, true);

    if (!version_collection.empty()) {
        return *boost::any_cast<VersionPair*>(*version_collection.begin());
    }

    return VersionPair();
}

}  // namespace db
}  // namespace isc