cassandra_statement.cpp

/*
 *  Copyright 2011 DataStax
 *
 *  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 "php_pdo_cassandra.hpp"
#include "php_pdo_cassandra_int.hpp"

#include <map>
#include <sstream>

template <class T>
T pdo_cassandra_marshal_numeric(pdo_stmt_t *stmt, const std::string &binary TSRMLS_DC);
static pdo_cassandra_type pdo_cassandra_get_cassandra_type(const std::string &type);

static zend_bool pdo_cassandra_describe_keyspace(pdo_stmt_t *stmt TSRMLS_DC)
{
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
    pdo_cassandra_db_handle *H = static_cast <pdo_cassandra_db_handle *>(S->H);

    if (H->has_description) {
        return 1;
    }

    try {
        H->client->describe_keyspace(H->description, H->active_keyspace);
        H->has_description = 1;
        return 1;
    } catch (NotFoundException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_NOT_FOUND, "%s", e.what());
    } catch (InvalidRequestException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_INVALID_REQUEST, "%s", e.why.c_str());
    } catch (UnavailableException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_UNAVAILABLE, "%s", e.what());
    } catch (TimedOutException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_TIMED_OUT, "%s", e.what());
    } catch (AuthenticationException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_AUTHENTICATION_ERROR, "%s", e.why.c_str());
    } catch (AuthorizationException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_AUTHORIZATION_ERROR, "%s", e.why.c_str());
    } catch (SchemaDisagreementException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_SCHEMA_DISAGREEMENT, "%s", e.what());
    } catch (TSSLException &e) {
        pdo_cassandra_error_exception(stmt->dbh, PDO_CASSANDRA_SSL_ERROR, "%s", e.what());
    } catch (TTransportException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_TRANSPORT_ERROR, "%s", e.what());
    } catch (TException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_GENERAL_ERROR, "%s", e.what());
    } catch (std::exception &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_GENERAL_ERROR, "%s", e.what());
    }
    return 0;
}

/** {{{ static void pdo_cassandra_stmt_undescribe(pdo_stmt_t *stmt TSRMLS_DC)
*/
static void pdo_cassandra_stmt_undescribe(pdo_stmt_t *stmt TSRMLS_DC)
{
    if (stmt->columns) {
        int i;
        struct pdo_column_data *cols = stmt->columns;

        for (i = 0; i < stmt->column_count; i++) {
#if PHP_MAJOR_VERSION >= 7
            zend_string_release(cols[i].name);
#else
            efree(cols[i].name);
#endif
        }
        efree(stmt->columns);
        stmt->columns = NULL;
        stmt->column_count = 0;
    }

    // Clear data
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
    S->original_column_names.clear();
    S->column_name_labels.clear();

    stmt->executed = 0;
}
/* }}} */

/** {{{ static int pdo_cassandra_stmt_execute(pdo_stmt_t *stmt TSRMLS_DC)
*/
static int pdo_cassandra_stmt_execute(pdo_stmt_t *stmt TSRMLS_DC)
{
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
    pdo_cassandra_db_handle *H = static_cast <pdo_cassandra_db_handle *>(S->H);

    try {
        if (!H->transport->isOpen()) {
            H->transport->open();
        }
        std::string query(stmt->active_query_string);

        S->result.reset(new CqlResult);
        H->client->execute_cql3_query(*S->result.get(), query, (H->compression ? Compression::GZIP : Compression::NONE), H->consistency);
        S->has_iterator = 0;
        stmt->row_count = S->result.get()->rows.size();
        pdo_cassandra_set_active_keyspace(H, query TSRMLS_CC);
        pdo_cassandra_set_active_columnfamily(H, query TSRMLS_CC);

        // Undescribe the result set because next time there might be different amount of columns
        pdo_cassandra_stmt_undescribe(stmt TSRMLS_CC);
        return 1;
    } catch (NotFoundException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_NOT_FOUND, "%s", e.what());
    } catch (InvalidRequestException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_INVALID_REQUEST, "%s", e.why.c_str());
    } catch (UnavailableException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_UNAVAILABLE, "%s", e.what());
    } catch (TimedOutException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_TIMED_OUT, "%s", e.what());
    } catch (AuthenticationException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_AUTHENTICATION_ERROR, "%s", e.why.c_str());
    } catch (AuthorizationException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_AUTHORIZATION_ERROR, "%s", e.why.c_str());
    } catch (SchemaDisagreementException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_SCHEMA_DISAGREEMENT, "%s", e.what());
    } catch (TSSLException &e) {
        pdo_cassandra_error_exception(stmt->dbh, PDO_CASSANDRA_SSL_ERROR, "%s", e.what());
    } catch (TTransportException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_TRANSPORT_ERROR, "%s", e.what());
    } catch (TException &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_GENERAL_ERROR, "%s", e.what());
    } catch (std::exception &e) {
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_GENERAL_ERROR, "%s", e.what());
    }
    return 0;
}
/* }}} */

static zend_bool pdo_cassandra_add_column(pdo_stmt_t *stmt, const std::string &name, int order)
{
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);

    try {
        S->original_column_names.left.at(name);
        return 0;
    } catch (std::out_of_range &ex) {
        S->original_column_names.insert(ColumnMap::value_type(name, order));
        return 1;
    }
}

/** {{{ static int pdo_cassandra_stmt_fetch(pdo_stmt_t *stmt, enum pdo_fetch_orientation ori, long offset TSRMLS_DC)
*/
static int pdo_cassandra_stmt_fetch(pdo_stmt_t *stmt, enum pdo_fetch_orientation ori, long offset TSRMLS_DC)
{
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
    int order = 0;

    if (!stmt->executed || !S->result.get()->rows.size()) {
        return 0;
    }

    if (!S->has_iterator) {
        S->it = S->result.get()->rows.begin();
        S->has_iterator = 1;

        // Set column names and labels if we don't have them already
        if (!S->original_column_names.size()) {

            // Get unique column names
            for (std::vector<CqlRow>::iterator it = S->result.get()->rows.begin(); it < S->result.get()->rows.end(); it++) {
                for (std::vector<Column>::iterator col_it = (*it).columns.begin(); col_it < (*it).columns.end(); col_it++) {
                    if (pdo_cassandra_add_column(stmt, (*col_it).name, order)) {
                        order++;
                    }
                }
            }
            stmt->column_count = order;
        }
        stmt->column_count = S->original_column_names.size();
    } else {
        S->it++;
    }

    if (S->it == S->result.get()->rows.end()) {
        // Iterated all rows, reset the iterator
        S->has_iterator = 0;
        S->it = S->result.get()->rows.begin();
        return 0;
    }
    return 1;
}
/* }}} */

static std::vector<pdo_cassandra_type> pdo_cassandra_get_element_type_in_collection(pdo_cassandra_type col_type, const std::string &type)
{
    std::vector<pdo_cassandra_type> ret;
    std::string real_type;
    int st = type.find('(') + 1;
    std::string rtype = type.substr(st, type.size() - st - 1);
    if (col_type == PDO_CASSANDRA_TYPE_MAP) {
        st = rtype.find(',');
        ret.push_back(pdo_cassandra_get_cassandra_type(rtype.substr(0, st)));
        ret.push_back(pdo_cassandra_get_cassandra_type(rtype.substr(st + 1)));
    }
    else
        ret.push_back(pdo_cassandra_get_cassandra_type(rtype));
    return ret;
}

/** {{{ pdo_cassandra_type pdo_cassandra_get_cassandra_type(const std::string &type)
*/
static pdo_cassandra_type pdo_cassandra_get_cassandra_type(const std::string &type)
{
    std::string real_type;
    if (type.find("org.apache.cassandra.db.marshal.") != std::string::npos)
        real_type = type.substr(::strlen("org.apache.cassandra.db.marshal."));
    else
        real_type = type;

    if (!real_type.compare("IntegerType"))
        return PDO_CASSANDRA_TYPE_VARINT;
    if (!real_type.compare("Int32Type"))
        return PDO_CASSANDRA_TYPE_INTEGER;
    if (!real_type.compare("LongType") ||
        !real_type.compare("DateType") ||
        !real_type.compare("CounterColumnType") ||
	!real_type.compare("TimestampType"))
        return PDO_CASSANDRA_TYPE_LONG;
    if (!real_type.compare("BooleanType"))
        return PDO_CASSANDRA_TYPE_BOOLEAN;
    if (!real_type.compare("DecimalType"))
        return PDO_CASSANDRA_TYPE_DECIMAL;
    if (!real_type.compare("FloatType"))
        return PDO_CASSANDRA_TYPE_FLOAT;
    if (!real_type.compare("DoubleType"))
        return PDO_CASSANDRA_TYPE_DOUBLE;
    if (!real_type.compare("UUIDType"))
        return PDO_CASSANDRA_TYPE_UUID;
    if (!real_type.compare("TimeUUIDType"))
        return PDO_CASSANDRA_TYPE_TIMEUUID;
    if (!real_type.compare(0, 7, "SetType"))
        return PDO_CASSANDRA_TYPE_SET;
    if (!real_type.compare(0, 7, "MapType"))
        return PDO_CASSANDRA_TYPE_MAP;
    if (!real_type.compare(0, 8, "ListType"))
        return PDO_CASSANDRA_TYPE_LIST;
    return PDO_CASSANDRA_TYPE_UTF8;
}
/* }}} */

namespace StreamExtraction {

    /**
     * Creates a zval reading bina
     */
    zval* extract_zval(zval *ret, const unsigned char *binary, pdo_cassandra_type type, int size);



    /**
     * Read n bytes to a primitive type
     */
    template <class T>
    T extract(const unsigned char *bytes, unsigned int stream_size = sizeof(T))
    {
        T val = T();

        unsigned char *pval = reinterpret_cast<unsigned char *>(&val) + stream_size;
        --pval;
        bool neg = bytes[0] & 0x80;

        for (size_t i = 0; i < stream_size; i++) {
            *pval = bytes[i];
            --pval;
        }

        // Negative number found
        if (neg) {
            pval = reinterpret_cast<unsigned char *>(&val) + sizeof(T);
            for (size_t i = stream_size; i < sizeof(T); i++) {
                --pval;
                *pval = 0xff;
            }
        }

        return val;
    }

    // Zval creators signature
    typedef zval *(*EvaluatorType)(zval *ret, const unsigned char *binary, int size);

    /**
     * Evaluator: Ascii extraction
     */
    zval *evaluate_string(zval * ret, const unsigned char *binary, int size) {
        //char *str = (char *) emalloc(sizeof(*str) * (size + 1));
        //memcpy(str, binary, size);

        Y_ZVAL_STRINGL(ret, (const char *)binary, size);

        return ret;
    }

    /**
     * Evaluator: UUID extraction
     */
    char to_hex(int x)
    {
        return x < 10 ? x + '0' : (x - 10) + 'a';
    }

    char *raw_uuid_to_str(const unsigned char *binary, int size) {

        int size_str = size * 2 + 4;

        char *str = (char *) emalloc(sizeof(*str) * (size_str + 1));

        int writepos = 0;
        int readpos = 0;
        for (int i = 0; i < size + 4; ++i) {
            if (i == 4 || i == 7 || i == 10 || i == 13) {
                str[writepos++] = '-';
            } else {
                str[writepos++] = to_hex((binary[readpos] >> 4) & 0x0F);
                str[writepos++] = to_hex(binary[readpos] & 0x0F);
                ++readpos;
            }
        }

        str[size_str] = '\0';
		return str;
    }

    zval *evaluate_uuid(zval *ret, const unsigned char *binary, int size) {

		char *str = StreamExtraction::raw_uuid_to_str(binary, size);

		Y_ZVAL_STRING(ret, str);
		efree(str);

        return ret;
    }

    /**
     * Evaluator: Binary to char *
     */
    zval *evaluate_bytes_to_zval(zval *ret, const unsigned char *binary, int size) {

        Y_ZVAL_STRINGL(ret, (const char *)binary, size);

        return ret;
    }

    /**
     * Evaluator: integer and long extractions
     */
    template <class T>
    zval *evaluate_integer_type(zval *ret, const unsigned char *binary, int stream_size = sizeof(T)) {

        ZVAL_LONG(ret, StreamExtraction::extract<T>(binary, stream_size));

        return ret;
    }

    template <>
    zval *evaluate_integer_type<bool>(zval *ret, const unsigned char *binary, int stream_size) {

        ZVAL_BOOL(ret, StreamExtraction::extract<bool>(binary, sizeof(bool)));

        return ret;
    }


    /**
     * Decimal number extraction
     * Php lacks such a type therefore we return an array containing the unscaled value and the scale
     */
    zval *evaluate_decimal_type(zval *ret, const unsigned char *binary, int size) {

        array_init(ret);
        // Scale extraction

        zval *scale;

        MAKE_STD_ZVAL(scale);

        extract_zval(scale, binary, PDO_CASSANDRA_TYPE_INTEGER, sizeof(int));
        add_next_index_zval(ret, scale);

        long unscaled_val_size = size - sizeof(int);

        if (unscaled_val_size < 0) {
            unscaled_val_size = 0;
        }

        binary += sizeof(int);

        // Unscaled value extraction
        zval *unscaled_value, *tmp;

        int repeats = unscaled_val_size / sizeof(long);
        int first_bytes = unscaled_val_size % sizeof(long);

        int i;
        for (i = 0; i < repeats ; i ++) {
            MAKE_STD_ZVAL(tmp);

            unscaled_value = evaluate_integer_type<long>(tmp, binary + first_bytes + sizeof(long) * (repeats - i - 1), sizeof(long));
            add_next_index_zval(ret, unscaled_value);
        }

        if (first_bytes) {
            MAKE_STD_ZVAL(tmp);

            if (first_bytes <= (int)sizeof(int)) {
                unscaled_value = evaluate_integer_type<int>(tmp, binary, first_bytes);
            } else {
                unscaled_value = evaluate_integer_type<long>(tmp, binary, first_bytes);
            }
            add_next_index_zval(ret, unscaled_value);
        }

        return ret;
    }

    /**
     * Evaluator: float and double extraction
     */
    template <class T>
    zval *evaluate_float_type(zval *ret, const unsigned char *binary, int size) {

        ZVAL_DOUBLE(ret, StreamExtraction::extract<T>(binary));

        return ret;
    }



        /**
     * Pointers on evaluators indexed by the pdo type
     */
    static std::map<pdo_cassandra_type, EvaluatorType> m_evaluation_map;

    static EvaluatorType evaluate(pdo_cassandra_type type) {
        static bool init = false;
        if (!init) {
            m_evaluation_map[PDO_CASSANDRA_TYPE_BYTES]          = evaluate_bytes_to_zval;
            m_evaluation_map[PDO_CASSANDRA_TYPE_ASCII]          = evaluate_string;
            m_evaluation_map[PDO_CASSANDRA_TYPE_UTF8]           = evaluate_string;
            m_evaluation_map[PDO_CASSANDRA_TYPE_INTEGER]        = evaluate_integer_type<int>;
            m_evaluation_map[PDO_CASSANDRA_TYPE_LONG]           = evaluate_integer_type<long>;
            m_evaluation_map[PDO_CASSANDRA_TYPE_DECIMAL]        = evaluate_decimal_type;
            m_evaluation_map[PDO_CASSANDRA_TYPE_UUID]           = evaluate_uuid;
            m_evaluation_map[PDO_CASSANDRA_TYPE_LEXICAL]        = 0;
            m_evaluation_map[PDO_CASSANDRA_TYPE_TIMEUUID]       = evaluate_uuid;
            m_evaluation_map[PDO_CASSANDRA_TYPE_BOOLEAN]        = evaluate_integer_type<bool>;
            m_evaluation_map[PDO_CASSANDRA_TYPE_VARINT]         = evaluate_bytes_to_zval;
            m_evaluation_map[PDO_CASSANDRA_TYPE_FLOAT]          = evaluate_float_type<float>;
            m_evaluation_map[PDO_CASSANDRA_TYPE_DOUBLE]         = evaluate_float_type<double>;
            m_evaluation_map[PDO_CASSANDRA_TYPE_SET]            = 0;
            m_evaluation_map[PDO_CASSANDRA_TYPE_MAP]            = 0;
            m_evaluation_map[PDO_CASSANDRA_TYPE_LIST]           = 0;
            m_evaluation_map[PDO_CASSANDRA_TYPE_UNKNOWN]        = 0;
            init = true;
        }
        return m_evaluation_map[type];
    }


    /**
     * Allocates a zval, and reads a value according to the specified type
     */
    zval* extract_zval(zval *ret, const unsigned char *binary, pdo_cassandra_type type, int size) {
        if (!size) {
            ZVAL_NULL(ret);
            return ret;
        }

        EvaluatorType pe = evaluate(type);
        if (!pe) {
            std::cout << "Zval extraction for type: " << (int)type << " not handled yet" << std::endl;
            //TODO raise exception
        }

        return (*pe)(ret, binary, size);
    }

};

template <class T>
T pdo_cassandra_marshal_numeric(pdo_stmt_t *stmt, const std::string &binary TSRMLS_DC)
{
    if (sizeof(T) != binary.size()) {
        // Binary is null
        if (!binary.size())
            return T();
        pdo_cassandra_error(stmt->dbh, PDO_CASSANDRA_INTEGER_CONVERSION_ERROR,
                            "pdo_cassandra_marshal_numeric: Binary stream and receiver size doesn't match", "");
        return T();
    }
    return StreamExtraction::extract<T>(reinterpret_cast <const unsigned char *>(binary.c_str()));
}


/** {{{ static int pdo_cassandra_stmt_describe(pdo_stmt_t *stmt, int colno TSRMLS_DC)
*/
static int pdo_cassandra_stmt_describe(pdo_stmt_t *stmt, int colno TSRMLS_DC)
{
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
    pdo_cassandra_db_handle *H = static_cast <pdo_cassandra_db_handle *>(S->H);

    std::string current_column;
    try {
        current_column = S->original_column_names.right.at(colno);
    } catch (std::out_of_range &ex) {
        return 0;
    }

    try {
        std::string column_label = S->column_name_labels.right.at(colno);

#if PHP_MAJOR_VERSION >= 7
        stmt->columns[colno].name    = zend_string_init(column_label.c_str(), column_label.size(), 0);
#else
        stmt->columns[colno].name    = estrdup(stmt->columns[colno].name);
        stmt->columns[colno].namelen = column_label.size();
#endif
    } catch (std::exception &e) {
#if PHP_MAJOR_VERSION >= 7
        stmt->columns[colno].name    = zend_string_init(current_column.c_str(), current_column.size(), 0);
#else
        stmt->columns[colno].name    = estrdup(current_column.c_str());
        stmt->columns[colno].namelen = current_column.size();
#endif
    }

    stmt->columns[colno].param_type = PDO_PARAM_ZVAL;
    stmt->columns[colno].precision  = 0;
    stmt->columns[colno].maxlen     = -1;
    return 1;
}
/* }}} */

zval* parse_collection(const std::string &type, const std::string &data) {

    // Extract Collection and data type
    std::vector<pdo_cassandra_type> elt_types = pdo_cassandra_get_element_type_in_collection(PDO_CASSANDRA_TYPE_LIST, type);
    //  assert((elt_types.size() == 1) && "parse_collection elt_types.size() == 1");
    unsigned short nbElements = StreamExtraction::extract<unsigned short>(reinterpret_cast <const unsigned char *>(data.c_str()));

    // ZVAl initialisation
    zval *collection;

    ALLOC_INIT_ZVAL(collection);

    array_init(collection);

    // Iterating trough the collection
    const unsigned char *datap = reinterpret_cast<const unsigned char *>(data.c_str() + 2);
    while (nbElements--)
        {
            unsigned short elem_size = StreamExtraction::extract<unsigned short>(datap);

            zval *zv;
            MAKE_STD_ZVAL(zv);

            datap += 2;

            add_next_index_zval(collection, StreamExtraction::extract_zval(zv, datap, elt_types[0], elem_size));

            datap += elem_size;
        }
    return collection;
}

zval* parse_collection_map(const std::string &type, const std::string &data) {

    // Extract Collection and data type
    std::vector<pdo_cassandra_type> elt_types = pdo_cassandra_get_element_type_in_collection(PDO_CASSANDRA_TYPE_MAP, type);
    unsigned short nbElements = StreamExtraction::extract<unsigned short>(reinterpret_cast <const unsigned char *>(data.c_str()));

    // ZVAl initialisation
    zval *collection;

    ALLOC_INIT_ZVAL(collection);

    array_init(collection);
    // Iterating trough the collection
    const unsigned char *datap = reinterpret_cast<const unsigned char *>(data.c_str() + 2);
    while (nbElements--)
        {
            // Reading key size
            unsigned short key_size = StreamExtraction::extract<unsigned short>(datap);
            datap += 2;

            // Extracting value
            const unsigned char *valuep = datap + key_size;
            unsigned short value_size = StreamExtraction::extract<unsigned short>(valuep);
            valuep += 2;

            zval *value;

            MAKE_STD_ZVAL(value);

            StreamExtraction::extract_zval(value, valuep, elt_types[1], value_size);

            // Extracting key and pushing the zval in the collection
            if (elt_types[0] == PDO_CASSANDRA_TYPE_ASCII ||
                elt_types[0] == PDO_CASSANDRA_TYPE_UTF8) {
                add_assoc_zval_ex(collection, (const char *)datap, key_size(key_size), value);
			}
			else if (elt_types[0] == PDO_CASSANDRA_TYPE_UUID ||
                elt_types[0] == PDO_CASSANDRA_TYPE_TIMEUUID) {

                // UUID key case, for now it gets no special treatment
                unsigned char *raw_uuid = (unsigned char *) emalloc(sizeof(*raw_uuid) * key_size);
                memcpy(raw_uuid, datap, key_size);
				char *str_uuid = StreamExtraction::raw_uuid_to_str(raw_uuid, key_size);
                add_assoc_zval(collection, str_uuid, value);

                efree(str_uuid);
            } else {
                // Numeric keys case
                assert(key_size == sizeof(int) && "parse_collection_map key_size assert");
                long key = StreamExtraction::extract<int>(datap);
                add_index_zval(collection, key, value);
            }

            datap += value_size + 2 + key_size;
        }

    return collection;
}

/** {{{ static int pdo_cassandra_stmt_get_column(pdo_stmt_t *stmt, int colno, char **ptr, unsigned long *len, int *caller_frees TSRMLS_DC)
*/
static int pdo_cassandra_stmt_get_column(pdo_stmt_t *stmt, int colno, char **ptr, unsigned long *len, int *caller_frees TSRMLS_DC)
{
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
    pdo_cassandra_db_handle *H = static_cast <pdo_cassandra_db_handle *>(S->H);

    std::string current_column;
    try {
        current_column = S->original_column_names.right.at(colno);
    } catch (std::out_of_range &ex) {
        return 0;
    }

    *ptr          = NULL;
    *len          = 0;
    *caller_frees = 0;

    // Do we have data for this column?
    for (std::vector<Column>::iterator col_it = (*S->it).columns.begin(); col_it < (*S->it).columns.end(); col_it++) {
        if (!current_column.compare(0, current_column.size(), (*col_it).name.c_str(), (*col_it).name.size())) {

            pdo_cassandra_type lparam_type;
            if (H->preserve_values) {
                lparam_type = PDO_CASSANDRA_TYPE_BYTES;
            } else {
                lparam_type = pdo_cassandra_get_cassandra_type(S->result.get ()->schema.value_types [current_column]);
            }

            switch (lparam_type)
                {
                case PDO_CASSANDRA_TYPE_LIST:
                case PDO_CASSANDRA_TYPE_SET:
                    {
#if PHP_MAJOR_VERSION >= 7
                        zval *ref = parse_collection(S->result.get ()->schema.value_types [current_column], (*col_it).value);
#else
                        zval **ref = (zval **) emalloc(sizeof(*ref));
                        // The return value of the parse collection must be the zval stuff
                        *ref = parse_collection(S->result.get ()->schema.value_types [current_column], (*col_it).value);
#endif
                        *ptr = (char *)ref;
                        *len = sizeof(zval);
                        *caller_frees = 1;
                        break;
                    }

                case PDO_CASSANDRA_TYPE_MAP:
                    {
#if PHP_MAJOR_VERSION >= 7
                        zval *ref = parse_collection_map(S->result.get ()->schema.value_types [current_column], (*col_it).value);
#else
                        zval **ref = (zval **) emalloc(sizeof(*ref));
                        // The return value of the parse collection must be the zval stuff
                        *ref = parse_collection_map(S->result.get ()->schema.value_types [current_column], (*col_it).value);
#endif
                        *ptr = (char *)ref;
                        *len = sizeof(zval);
                        *caller_frees = 1;
                        break;
                    }

                default:
#if PHP_MAJOR_VERSION >= 7
                    zval *ref = (zval *)malloc(sizeof (zval));
                    ZVAL_UNDEF(ref);

                    StreamExtraction::extract_zval(ref, reinterpret_cast<const unsigned char*>((*col_it).value.c_str()),
                                                                              lparam_type,
                                                                              (*col_it).value.size());

#else
                    zval **ref = (zval **)malloc(sizeof (*ref));
                    zval *pref;
                    MAKE_STD_ZVAL(pref);

                    *ref = StreamExtraction::extract_zval(pref, reinterpret_cast<const unsigned char*>((*col_it).value.c_str()),
                                                                              lparam_type,
                                                                              (*col_it).value.size());
#endif

                    *ptr = (char *)ref;
                    *len = sizeof(zval);
                    break;
                }
            return 1;
        }
    }
    return 0;
}
/* }}} */

/** {{{ static int pdo_cassandra_stmt_get_column_meta(pdo_stmt_t *stmt, long colno, zval *return_value TSRMLS_DC)
 */
static int pdo_cassandra_stmt_get_column_meta(pdo_stmt_t *stmt, long colno, zval *return_value TSRMLS_DC)
{
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
    pdo_cassandra_db_handle *H = static_cast <pdo_cassandra_db_handle *>(S->H);

    // If the stmt is not in executed state, don't even try
    if (!stmt->executed) {
        return FAILURE;
    }

    std::string current_column;
    try {
        current_column = S->original_column_names.right.at(colno);
    } catch (std::out_of_range &ex) {
        return FAILURE;
    }
    array_init(return_value);

    if (!pdo_cassandra_describe_keyspace(stmt TSRMLS_CC)) {
        return 0;
    }

    y_add_assoc_stringl(return_value,
                      "keyspace",
                      const_cast <char *> (H->active_keyspace.c_str()),
                      H->active_keyspace.size());

    y_add_assoc_stringl(return_value,
                      "columnfamily",
                      const_cast <char *> (H->active_columnfamily.c_str()),
                      H->active_columnfamily.size());

    y_add_assoc_string(return_value,
                     "native_type",
                     const_cast <char *>(S->result.get()->schema.value_types[current_column].c_str()));

#if 0
// This code is not fonctionnal. However I kept it because this is how this feature should be implemented
bool found = false;
    for (std::vector<CfDef>::iterator cfdef_it = H->description.cf_defs.begin(); cfdef_it < H->description.cf_defs.end(); cfdef_it++) {
        for (std::vector<ColumnDef>::iterator columndef_it = (*cfdef_it).column_metadata.begin(); columndef_it < (*cfdef_it).column_metadata.end(); columndef_it++) {


            // Only interested in the currently active CF
            if ((*cfdef_it).name.compare(H->active_columnfamily)) {
                continue;
            }
            else if (!(*cfdef_it).key_alias.compare(current_column)) {
                found = true;
                add_assoc_string(return_value,
                                 "native_type",
                                 "key_alias",
                                 1);
                break;
            }
            else if (!current_column.compare(0, current_column.size(), (*columndef_it).name.c_str(), (*columndef_it).name.size())) {
                found = true;
                add_assoc_string(return_value,
                                 "native_type",
                                 const_cast <char *> ((*columndef_it).validation_class.c_str()),
                                 1);
                add_assoc_string(return_value,
                                 "comparator",
                                 const_cast <char *> ((*cfdef_it).comparator_type.c_str()),
                                 1);
                add_assoc_string(return_value,
                                 "default_validation_class",
                                 const_cast <char *> ((*cfdef_it).default_validation_class.c_str()),
                                 1);
                add_assoc_string(return_value,
                                 "key_validation_class",
                                 const_cast <char *> ((*cfdef_it).key_validation_class.c_str()),
                                 1);
                add_assoc_stringl(return_value,
                                  "key_alias",
                                  const_cast <char *> ((*cfdef_it).key_alias.c_str()),
                                  (*cfdef_it).key_alias.size(),
                                  1);
                add_assoc_stringl(return_value,
                                  "original_column_name",
                                  const_cast <char *> (current_column.c_str()),
                                  current_column.size(),
                                  1);
                break;
            }
        }
    }
    if (!found) {
        add_assoc_string(return_value,
                         "native_type",
                         "unknown",
                         1);
    }
#endif
    return SUCCESS;
}
/* }}} */

/** {{{ static int pdo_cassandra_stmt_cursor_closer(pdo_stmt_t *stmt TSRMLS_DC)
*/
static int pdo_cassandra_stmt_cursor_close(pdo_stmt_t *stmt TSRMLS_DC)
{
    pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
    S->has_iterator       = 0;
    return 1;
}
/* }}} */

/** {{{ static int pdo_cassandra_stmt_dtor(pdo_stmt_t *stmt TSRMLS_DC)
*/
static int pdo_cassandra_stmt_dtor(pdo_stmt_t *stmt TSRMLS_DC)
{
    if (stmt->driver_data) {
        pdo_cassandra_stmt *S = static_cast <pdo_cassandra_stmt *>(stmt->driver_data);
        S->result.reset();
        delete S;
        stmt->driver_data = NULL;
    }
    return 1;
}
/* }}} */

struct pdo_stmt_methods cassandra_stmt_methods = {
    pdo_cassandra_stmt_dtor,
    pdo_cassandra_stmt_execute,
    pdo_cassandra_stmt_fetch,
    pdo_cassandra_stmt_describe,
    pdo_cassandra_stmt_get_column,
    NULL, /* param_hook */
    NULL, /* set_attr */
    NULL, /* get_attr */
    pdo_cassandra_stmt_get_column_meta,
    NULL, /* next rowset */
    pdo_cassandra_stmt_cursor_close
};

/*
 * Local variables:
 * tab-width: 4
 * c-basic-offset: 4
 * End:
 * vim600: noet sw=4 ts=4 fdm=marker
 * vim<600: noet sw=4 ts=4
 */