ControlStartupQueryPlan(const HostMap& hosts, Random* random)

: index_(random != NULL ? random->next(std::max(static_cast<size_t>(1), hosts.size())) : 0)

, count_(0) {

hosts_.reserve(hosts.size());

std::transform(hosts.begin(), hosts.end(), std::back_inserter(hosts_), GetHost());

}

virtual Host::Ptr compute_next() {

const size_t size = hosts_.size();

if (count_ >= size) return Host::Ptr();

size_t index = (index_ + count_) % size;

++count_;

return hosts_[index];

}

HostVec hosts_;

size_t index_;

size_t count_;

const Value* peer_value,

const Value* rpc_value,

Address* output) {

Address peer_address;

if (!Address::from_inet(peer_value->data(), peer_value->size(),

connected_address.port(), &peer_address)) {

LOG_WARN("Invalid address format for peer address");

return false;

}

if (rpc_value->size() > 0) {

if (!Address::from_inet(rpc_value->data(), rpc_value->size(),

connected_address.port(), output)) {

LOG_WARN("Invalid address format for rpc address");

return false;

}

if (connected_address == *output || connected_address == peer_address) {

LOG_DEBUG("system.peers on %s contains a line with rpc_address for itself. "

"This is not normal, but is a known problem for some versions of DSE. "

"Ignoring this entry.", connected_address.to_string(false).c_str());

return false;

}

if (Address::BIND_ANY_IPV4.compare(*output, false) == 0 ||

Address::BIND_ANY_IPV6.compare(*output, false) == 0) {

LOG_WARN("Found host with 'bind any' for rpc_address; using listen_address (%s) to contact instead. "

"If this is incorrect you should configure a specific interface for rpc_address on the server.",

peer_address.to_string(false).c_str());

*output = peer_address;

}

} else {

LOG_WARN("No rpc_address for host %s in system.peers on %s. "

"Ignoring this entry.", peer_address.to_string(false).c_str(),

connected_address.to_string(false).c_str());

return false;

}

return true;

: state_(CONTROL_STATE_NEW)

, session_(NULL)

, connection_(NULL)

, protocol_version_(0)

, use_schema_(false)

state_ = CONTROL_STATE_NEW;

session_ = NULL;

connection_ = NULL;

reconnect_timer_.stop();

query_plan_.reset();

protocol_version_ = 0;

last_connection_error_.clear();

use_schema_ = false;

token_aware_routing_ = false;

session_ = session;

query_plan_.reset(new ControlStartupQueryPlan(session_->hosts_, // No hosts lock necessary (read-only)

session_->random_.get()));

protocol_version_ = session_->config().protocol_version();

use_schema_ = session_->config().use_schema();

token_aware_routing_ = session_->config().token_aware_routing();

if (protocol_version_ < 0) {

protocol_version_ = CASS_HIGHEST_SUPPORTED_PROTOCOL_VERSION;

}

if (use_schema_ || token_aware_routing_) {

set_event_types(CASS_EVENT_TOPOLOGY_CHANGE | CASS_EVENT_STATUS_CHANGE |

CASS_EVENT_SCHEMA_CHANGE);

} else {

set_event_types(CASS_EVENT_TOPOLOGY_CHANGE | CASS_EVENT_STATUS_CHANGE);

}

reconnect(false);

state_ = CONTROL_STATE_CLOSED;

if (connection_ != NULL) {

connection_->close();

}

reconnect_timer_.stop();

reconnect_timer_.start(session_->loop(),

ms,

this,

ControlConnection::on_reconnect);

if (state_ == CONTROL_STATE_CLOSED) {

return;

}

if (!retry_current_host) {

current_host_ = query_plan_->compute_next();

if (!current_host_) {

if (state_ == CONTROL_STATE_READY) {

schedule_reconnect(1000); // TODO(mpenick): Configurable?

} else {

session_->on_control_connection_error(CASS_ERROR_LIB_NO_HOSTS_AVAILABLE,

"No hosts available for the control connection");

}

return;

}

}

if (connection_ != NULL) {

connection_->close();

}

connection_ = new Connection(session_->loop(),

session_->config(),

session_->metrics(),

current_host_,

"", // No keyspace

protocol_version_,

this);

connection_->connect();

LOG_DEBUG("Connection ready on host %s",

connection->address().to_string().c_str());

// The control connection has to refresh meta when there's a reconnect because

// events could have been missed while not connected.

query_meta_hosts();

bool retry_current_host = false;

// This pointer to the connection is no longer valid once it's closed

connection_ = NULL;

if (state_ == CONTROL_STATE_NEW) {

if (connection->is_invalid_protocol()) {

if (protocol_version_ <= 1) {

LOG_ERROR("Host %s does not support any valid protocol version",

connection->address_string().c_str());

session_->on_control_connection_error(CASS_ERROR_LIB_UNABLE_TO_DETERMINE_PROTOCOL,

"Not even protocol version 1 is supported");

return;

}

LOG_WARN("Host %s does not support protocol version %d. "

"Trying protocol version %d...",

connection->address_string().c_str(),

protocol_version_,

protocol_version_ - 1);

protocol_version_--;

retry_current_host = true;

} else if (connection->is_auth_error()) {

session_->on_control_connection_error(CASS_ERROR_SERVER_BAD_CREDENTIALS,

connection->error_message());

return;

} else if (connection->is_ssl_error()) {

session_->on_control_connection_error(connection->ssl_error_code(),

connection->error_message());

return;

}

}

// Don't log if the control connection is closing/closed or retrying because of

// an invalid protocol error.

if (state_ != CONTROL_STATE_CLOSED && !retry_current_host) {

// Log only as an error if it's the initial attempt

if (state_ == CONTROL_STATE_NEW) {

LOG_ERROR("Unable to establish a control connection to host %s because of the following error: %s",

connection->address_string().c_str(),

connection->error_message().c_str());

} else {

LOG_WARN("Lost control connection to host %s with the following error: %s",

connection->address_string().c_str(),

connection->error_message().c_str());

}

}

reconnect(retry_current_host);

// Only process events after an initial set of hosts and schema have been

// established. Adding a host from an UP/NEW_NODE event before the initial

// set will cause the driver to hang waiting for an invalid pending pool

// count.

if (state_ != CONTROL_STATE_READY) return;

switch (response->event_type()) {

case CASS_EVENT_TOPOLOGY_CHANGE: {

std::string address_str = response->affected_node().to_string();

switch (response->topology_change()) {

case EventResponse::NEW_NODE: {

LOG_INFO("New node %s added", address_str.c_str());

Host::Ptr host = session_->get_host(response->affected_node());

if (!host) {

host = session_->add_host(response->affected_node());

refresh_node_info(host, true, true);

}

break;

}

case EventResponse::REMOVED_NODE: {

LOG_INFO("Node %s removed", address_str.c_str());

Host::Ptr host = session_->get_host(response->affected_node());

if (host) {

session_->on_remove(host);

if (session_->token_map_) {

session_->token_map_->remove_host_and_build(host);

}

} else {

LOG_DEBUG("Tried to remove host %s that doesn't exist", address_str.c_str());

}

break;

}

case EventResponse::MOVED_NODE:

LOG_INFO("Node %s moved", address_str.c_str());

Host::Ptr host = session_->get_host(response->affected_node());

if (host) {

refresh_node_info(host, false, true);

} else {

LOG_DEBUG("Move event for host %s that doesn't exist", address_str.c_str());

if (session_->token_map_) {

session_->token_map_->remove_host_and_build(host);

}

}

break;

}

break;

}

case CASS_EVENT_STATUS_CHANGE: {

std::string address_str = response->affected_node().to_string();

switch (response->status_change()) {

case EventResponse::UP: {

LOG_INFO("Node %s is up", address_str.c_str());

on_up(response->affected_node());

break;

}

case EventResponse::DOWN: {

LOG_INFO("Node %s is down", address_str.c_str());

on_down(response->affected_node());

break;

}

}

break;

}

case CASS_EVENT_SCHEMA_CHANGE:

// Only handle keyspace events when using token-aware routing

if (!use_schema_ &&

response->schema_change_target() != EventResponse::KEYSPACE) {

return;

}

LOG_DEBUG("Schema change (%d): %.*s %.*s\n",

response->schema_change(),

(int)response->keyspace().size(), response->keyspace().data(),

(int)response->target().size(), response->target().data());

switch (response->schema_change()) {

case EventResponse::CREATED:

case EventResponse::UPDATED:

switch (response->schema_change_target()) {

case EventResponse::KEYSPACE:

refresh_keyspace(response->keyspace());

break;

case EventResponse::TABLE:

refresh_table_or_view(response->keyspace(), response->target());

break;

case EventResponse::TYPE:

refresh_type(response->keyspace(), response->target());

break;

case EventResponse::FUNCTION:

case EventResponse::AGGREGATE:

refresh_function(response->keyspace(),

response->target(),

response->arg_types(),

response->schema_change_target() == EventResponse::AGGREGATE);

break;

}

break;

case EventResponse::DROPPED:

switch (response->schema_change_target()) {

case EventResponse::KEYSPACE:

session_->metadata().drop_keyspace(response->keyspace().to_string());

break;

case EventResponse::TABLE:

session_->metadata().drop_table_or_view(response->keyspace().to_string(),

response->target().to_string());

break;

case EventResponse::TYPE:

session_->metadata().drop_user_type(response->keyspace().to_string(),

response->target().to_string());

break;

case EventResponse::FUNCTION:

session_->metadata().drop_function(response->keyspace().to_string(),

Metadata::full_function_name(response->target().to_string(),

to_strings(response->arg_types())));

break;

case EventResponse::AGGREGATE:

session_->metadata().drop_aggregate(response->keyspace().to_string(),

Metadata::full_function_name(response->target().to_string(),

to_strings(response->arg_types())));

break;

}

break;

}

break;

default:

assert(false);

break;

}

SharedRefPtr<ControlMultipleRequestCallback<UnusedData> > callback(

new ControlMultipleRequestCallback<UnusedData>(this, ControlConnection::on_query_hosts, UnusedData()));

// This needs to happen before other schema metadata queries so that we have

// a valid Cassandra version because this version determines which follow up

// schema metadata queries are executed.

callback->execute_query("local", token_aware_routing_ ? SELECT_LOCAL_TOKENS : SELECT_LOCAL);

callback->execute_query("peers", token_aware_routing_ ? SELECT_PEERS_TOKENS : SELECT_PEERS);

const UnusedData& data,

const MultipleRequestCallback::ResponseMap& responses) {

Connection* connection = control_connection->connection_;

if (connection == NULL) {

return;

}

Session* session = control_connection->session_;

if (session->token_map_) {

// Clearing token/hosts will not invalidate the replicas

session->token_map_->clear_tokens_and_hosts();

}

bool is_initial_connection = (control_connection->state_ == CONTROL_STATE_NEW);

// If the 'system.local' table is empty the connection isn't used as a control

// connection because at least one node's information is required (itself). An

// empty 'system.local' can happen during the bootstrapping process on some

// versions of Cassandra. If this happens we defunct the connection and move

// to the next node in the query plan.

{

Host::Ptr host = session->get_host(connection->address());

if (host) {

host->set_mark(session->current_host_mark_);

ResultResponse* local_result;

if (MultipleRequestCallback::get_result_response(responses, "local", &local_result) &&

local_result->row_count() > 0) {

control_connection->update_node_info(host, &local_result->first_row(), ADD_HOST);

control_connection->cassandra_version_ = host->cassandra_version();

} else {

LOG_WARN("No row found in %s's local system table",

connection->address_string().c_str());

connection->defunct();

return;

}

} else {

LOG_WARN("Host %s from local system table not found",

connection->address_string().c_str());

connection->defunct();

return;

}

}

{

ResultResponse* peers_result;

if (MultipleRequestCallback::get_result_response(responses, "peers", &peers_result)) {

ResultIterator rows(peers_result);

while (rows.next()) {

Address address;

const Row* row = rows.row();

if (!determine_address_for_peer_host(connection->address(),

row->get_by_name("peer"),

row->get_by_name("rpc_address"),

&address)) {

continue;

}

Host::Ptr host = session->get_host(address);

bool is_new = false;

if (!host) {

is_new = true;

host = session->add_host(address);

}

host->set_mark(session->current_host_mark_);

control_connection->update_node_info(host, rows.row(), ADD_HOST);

if (is_new && !is_initial_connection) {

session->on_add(host, false);

}

}

}

}

session->purge_hosts(is_initial_connection);

if (control_connection->use_schema_ ||

control_connection->token_aware_routing_) {

control_connection->query_meta_schema();

} else if (is_initial_connection) {

control_connection->state_ = CONTROL_STATE_READY;

session->on_control_connection_ready();

// Create a new query plan that considers all the new hosts from the

// "system" tables.

control_connection->query_plan_.reset(session->new_query_plan());

}

SharedRefPtr<ControlMultipleRequestCallback<UnusedData> > callback(

new ControlMultipleRequestCallback<UnusedData>(this, ControlConnection::on_query_meta_schema, UnusedData()));

if (cassandra_version_ >= VersionNumber(3, 0, 0)) {

if (use_schema_ || token_aware_routing_) {

callback->execute_query("keyspaces", SELECT_KEYSPACES_30);

}

if (use_schema_) {

callback->execute_query("tables", SELECT_TABLES_30);

callback->execute_query("views", SELECT_VIEWS_30);

callback->execute_query("columns", SELECT_COLUMNS_30);

callback->execute_query("indexes", SELECT_INDEXES_30);

callback->execute_query("user_types", SELECT_USERTYPES_30);

callback->execute_query("functions", SELECT_FUNCTIONS_30);

callback->execute_query("aggregates", SELECT_AGGREGATES_30);

}

} else {

if (use_schema_ || token_aware_routing_) {

callback->execute_query("keyspaces", SELECT_KEYSPACES_20);

}

if (use_schema_) {

callback->execute_query("tables", SELECT_COLUMN_FAMILIES_20);

callback->execute_query("columns", SELECT_COLUMNS_20);

if (cassandra_version_ >= VersionNumber(2, 1, 0)) {

callback->execute_query("user_types", SELECT_USERTYPES_21);

}

if (cassandra_version_ >= VersionNumber(2, 2, 0)) {

callback->execute_query("functions", SELECT_FUNCTIONS_22);

callback->execute_query("aggregates", SELECT_AGGREGATES_22);

}

}

}

const UnusedData& unused,

const MultipleRequestCallback::ResponseMap& responses) {

Connection* connection = control_connection->connection_;

if (connection == NULL) {

return;

}

Session* session = control_connection->session_;

int protocol_version = control_connection->protocol_version_;

const VersionNumber& cassandra_version = control_connection->cassandra_version_;

bool is_initial_connection = (control_connection->state_ == CONTROL_STATE_NEW);

if (session->token_map_) {

ResultResponse* keyspaces_result;

if (MultipleRequestCallback::get_result_response(responses, "keyspaces", &keyspaces_result)) {

session->token_map_->clear_replicas_and_strategies(); // Only clear replicas once we have the new keyspaces

session->token_map_->add_keyspaces(cassandra_version, keyspaces_result);

}

session->token_map_->build();

}

if (control_connection->use_schema_) {

session->metadata().clear_and_update_back(cassandra_version);

ResultResponse* keyspaces_result;

if (MultipleRequestCallback::get_result_response(responses, "keyspaces", &keyspaces_result)) {

session->metadata().update_keyspaces(protocol_version, cassandra_version, keyspaces_result);

}

ResultResponse* tables_result;

if (MultipleRequestCallback::get_result_response(responses, "tables", &tables_result)) {

session->metadata().update_tables(protocol_version, cassandra_version, tables_result);

}

ResultResponse* views_result;

if (MultipleRequestCallback::get_result_response(responses, "views", &views_result)) {

session->metadata().update_views(protocol_version, cassandra_version, views_result);

}

ResultResponse* columns_result = NULL;

if (MultipleRequestCallback::get_result_response(responses, "columns", &columns_result)) {

session->metadata().update_columns(protocol_version, cassandra_version, columns_result);

}

ResultResponse* indexes_result;

if (MultipleRequestCallback::get_result_response(responses, "indexes", &indexes_result)) {

session->metadata().update_indexes(protocol_version, cassandra_version, indexes_result);

}

ResultResponse* user_types_result;

if (MultipleRequestCallback::get_result_response(responses, "user_types", &user_types_result)) {

session->metadata().update_user_types(protocol_version, cassandra_version, user_types_result);

}

ResultResponse* functions_result;

if (MultipleRequestCallback::get_result_response(responses, "functions", &functions_result)) {

session->metadata().update_functions(protocol_version, cassandra_version, functions_result);

}

ResultResponse* aggregates_result;

if (MultipleRequestCallback::get_result_response(responses, "aggregates", &aggregates_result)) {

session->metadata().update_aggregates(protocol_version, cassandra_version, aggregates_result);

}

session->metadata().swap_to_back_and_update_front();

}

if (is_initial_connection) {

control_connection->state_ = CONTROL_STATE_READY;

session->on_control_connection_ready();

// Create a new query plan that considers all the new hosts from the

// "system" tables.

control_connection->query_plan_.reset(session->new_query_plan());

}

bool is_new_node,

bool query_tokens) {

if (connection_ == NULL || !connection_->is_ready()) {

return;

}

bool is_connected_host = host->address() == connection_->address();

std::string query;

ControlCallback<RefreshNodeData>::ResponseCallback response_callback;

bool token_query = token_aware_routing_ && (host->was_just_added() || query_tokens);

if (is_connected_host || !host->listen_address().empty()) {

if (is_connected_host) {

query.assign(token_query ? SELECT_LOCAL_TOKENS : SELECT_LOCAL);

} else {

query.assign(token_query ? SELECT_PEERS_TOKENS : SELECT_PEERS);

query.append(" WHERE peer = '");

query.append(host->listen_address());

query.append("'");

}

response_callback = ControlConnection::on_refresh_node_info;

} else {

query.assign(token_query ? SELECT_PEERS_TOKENS : SELECT_PEERS);

response_callback = ControlConnection::on_refresh_node_info_all;

}

LOG_DEBUG("refresh_node_info: %s", query.c_str());

RefreshNodeData data(host, is_new_node);

SharedRefPtr<ControlCallback<RefreshNodeData> > callback(

new ControlCallback<RefreshNodeData>(Request::ConstPtr(new QueryRequest(query)),

this,

response_callback,

data));

if (!connection_->write(callback)) {

LOG_ERROR("No more stream available while attempting to refresh node info");

connection_->defunct();

}

const RefreshNodeData& data,

Response* response) {

Connection* connection = control_connection->connection_;

if (connection == NULL) {

return;

}

ResultResponse* result =

static_cast<ResultResponse*>(response);

if (result->row_count() == 0) {

std::string host_address_str = data.host->address().to_string();

LOG_ERROR("No row found for host %s in %s's local/peers system table. "

"%s will be ignored.",

host_address_str.c_str(),

connection->address_string().c_str(),

host_address_str.c_str());

return;

}

control_connection->update_node_info(data.host, &result->first_row(), UPDATE_HOST_AND_BUILD);

if (data.is_new_node) {

control_connection->session_->on_add(data.host, false);

}

const RefreshNodeData& data,

Response* response) {

Connection* connection = control_connection->connection_;

if (connection == NULL) {

return;

}

ResultResponse* result =

static_cast<ResultResponse*>(response);

if (result->row_count() == 0) {

std::string host_address_str = data.host->address().to_string();

LOG_ERROR("No row found for host %s in %s's peers system table. "

"%s will be ignored.",

host_address_str.c_str(),

connection->address_string().c_str(),

host_address_str.c_str());

return;

}

ResultIterator rows(result);

while (rows.next()) {

const Row* row = rows.row();

Address address;

bool is_valid_address

= determine_address_for_peer_host(connection->address(),

row->get_by_name("peer"),

row->get_by_name("rpc_address"),

&address);

if (is_valid_address && data.host->address() == address) {

control_connection->update_node_info(data.host, row, UPDATE_HOST_AND_BUILD);

if (data.is_new_node) {

control_connection->session_->on_add(data.host, false);

}

break;

}

}

const Value* v;

std::string rack;

row->get_string_by_name("rack", &rack);

std::string dc;

row->get_string_by_name("data_center", &dc);

std::string release_version;

row->get_string_by_name("release_version", &release_version);

// This value is not present in the "system.local" query

v = row->get_by_name("peer");

if (v != NULL) {

Address listen_address;

if (Address::from_inet(v->data(), v->size(),

connection_->address().port(),

&listen_address)) {

host->set_listen_address(listen_address.to_string());

} else {

LOG_WARN("Invalid address format for listen address");

}

}

if ((!rack.empty() && rack != host->rack()) ||

(!dc.empty() && dc != host->dc())) {

if (!host->was_just_added()) {

session_->load_balancing_policy_->on_remove(host);

}

host->set_rack_and_dc(rack, dc);

if (!host->was_just_added()) {

session_->load_balancing_policy_->on_add(host);

}

}

VersionNumber cassandra_version;

if (cassandra_version.parse(release_version)) {

host->set_cassaandra_version(cassandra_version);

} else {

LOG_WARN("Invalid release version string \"%s\" on host %s",

release_version.c_str(),

host->address().to_string().c_str());

}

if (token_aware_routing_) {

bool is_connected_host = connection_ != NULL && host->address() == connection_->address();

std::string partitioner;

if (is_connected_host && row->get_string_by_name("partitioner", &partitioner)) {

if (!session_->token_map_) {

session_->token_map_.reset(TokenMap::from_partitioner(partitioner));

}

}

v = row->get_by_name("tokens");

if (v != NULL && v->is_collection()) {

if (session_->token_map_) {

if (type == UPDATE_HOST_AND_BUILD) {

session_->token_map_->update_host_and_build(host, v);

} else {

session_->token_map_->add_host(host, v);

}

}

}

}

std::string query;

if (cassandra_version_ >= VersionNumber(3, 0, 0)) {

query.assign(SELECT_KEYSPACES_30);

} else {

query.assign(SELECT_KEYSPACES_20);

}

query.append(" WHERE keyspace_name='")

.append(keyspace_name.data(), keyspace_name.size())

.append("'");

LOG_DEBUG("Refreshing keyspace %s", query.c_str());

if (!connection_->write(

RequestCallback::Ptr(

new ControlCallback<std::string>(

Request::ConstPtr(new QueryRequest(query)),

this,

ControlConnection::on_refresh_keyspace,

keyspace_name.to_string())))) {

LOG_ERROR("No more stream available while attempting to refresh keyspace info");

connection_->defunct();

}

const std::string& keyspace_name,

Response* response) {

ResultResponse* result = static_cast<ResultResponse*>(response);

if (result->row_count() == 0) {

LOG_ERROR("No row found for keyspace %s in system schema table.",

keyspace_name.c_str());

return;

}

Session* session = control_connection->session_;

int protocol_version = control_connection->protocol_version_;

const VersionNumber& cassandra_version = control_connection->cassandra_version_;

if (session->token_map_) {

session->token_map_->update_keyspaces_and_build(cassandra_version, result);

}

if (control_connection->use_schema_) {

session->metadata().update_keyspaces(protocol_version, cassandra_version, result);

}

const StringRef& table_or_view_name) {

std::string table_query;

std::string view_query;

std::string column_query;

std::string index_query;

if (cassandra_version_ >= VersionNumber(3, 0, 0)) {

table_query.assign(SELECT_TABLES_30);

table_query.append(" WHERE keyspace_name='").append(keyspace_name.data(), keyspace_name.size())

.append("' AND table_name='").append(table_or_view_name.data(), table_or_view_name.size()).append("'");

view_query.assign(SELECT_VIEWS_30);

view_query.append(" WHERE keyspace_name='").append(keyspace_name.data(), keyspace_name.size())

.append("' AND view_name='").append(table_or_view_name.data(), table_or_view_name.size()).append("'");

column_query.assign(SELECT_COLUMNS_30);

column_query.append(" WHERE keyspace_name='").append(keyspace_name.data(), keyspace_name.size())

.append("' AND table_name='").append(table_or_view_name.data(), table_or_view_name.size()).append("'");

index_query.assign(SELECT_INDEXES_30);

index_query.append(" WHERE keyspace_name='").append(keyspace_name.data(), keyspace_name.size())

.append("' AND table_name='").append(table_or_view_name.data(), table_or_view_name.size()).append("'");

LOG_DEBUG("Refreshing table/view %s; %s; %s; %s", table_query.c_str(), view_query.c_str(),

column_query.c_str(), index_query.c_str());

} else {

table_query.assign(SELECT_COLUMN_FAMILIES_20);

table_query.append(" WHERE keyspace_name='").append(keyspace_name.data(), keyspace_name.size())

.append("' AND columnfamily_name='").append(table_or_view_name.data(), table_or_view_name.size()).append("'");

column_query.assign(SELECT_COLUMNS_20);

column_query.append(" WHERE keyspace_name='").append(keyspace_name.data(), keyspace_name.size())

.append("' AND columnfamily_name='").append(table_or_view_name.data(), table_or_view_name.size()).append("'");

LOG_DEBUG("Refreshing table %s; %s", table_query.c_str(), column_query.c_str());

}

SharedRefPtr<ControlMultipleRequestCallback<RefreshTableData> > callback(

new ControlMultipleRequestCallback<RefreshTableData>(this,

ControlConnection::on_refresh_table_or_view,

RefreshTableData(keyspace_name.to_string(), table_or_view_name.to_string())));

callback->execute_query("tables", table_query);

if (!view_query.empty()) {

callback->execute_query("views", view_query);

}

callback->execute_query("columns", column_query);

if (!index_query.empty()) {

callback->execute_query("indexes", index_query);

}

const RefreshTableData& data,

const MultipleRequestCallback::ResponseMap& responses) {

ResultResponse* tables_result;

Session* session = control_connection->session_;

int protocol_version = control_connection->protocol_version_;

const VersionNumber& cassandra_version = control_connection->cassandra_version_;

if (!MultipleRequestCallback::get_result_response(responses, "tables", &tables_result) ||

tables_result->row_count() == 0) {

ResultResponse* views_result;

if (!MultipleRequestCallback::get_result_response(responses, "views", &views_result) ||

views_result->row_count() == 0) {

LOG_ERROR("No row found for table (or view) %s.%s in system schema tables.",

data.keyspace_name.c_str(), data.table_or_view_name.c_str());

return;

}

session->metadata().update_views(protocol_version, cassandra_version, views_result);

} else {

session->metadata().update_tables(protocol_version, cassandra_version, tables_result);

}

ResultResponse* columns_result;

if (MultipleRequestCallback::get_result_response(responses, "columns", &columns_result)) {

session->metadata().update_columns(protocol_version, cassandra_version, columns_result);

}

ResultResponse* indexes_result;

if (MultipleRequestCallback::get_result_response(responses, "indexes", &indexes_result)) {

session->metadata().update_indexes(protocol_version, cassandra_version, indexes_result);

}

const StringRef& type_name) {

std::string query;

if (cassandra_version_ >= VersionNumber(3, 0, 0)) {

query.assign(SELECT_USERTYPES_30);

} else {

query.assign(SELECT_USERTYPES_21);

}

query.append(" WHERE keyspace_name='").append(keyspace_name.data(), keyspace_name.size())

.append("' AND type_name='").append(type_name.data(), type_name.size()).append("'");

LOG_DEBUG("Refreshing type %s", query.c_str());

if (!connection_->write(

RequestCallback::Ptr(

new ControlCallback<std::pair<std::string, std::string> >(

Request::ConstPtr(new QueryRequest(query)),

this,

ControlConnection::on_refresh_type,

std::make_pair(keyspace_name.to_string(), type_name.to_string()))))) {

LOG_ERROR("No more stream available while attempting to refresh type info");

connection_->defunct();

}

const std::pair<std::string, std::string>& keyspace_and_type_names,

Response* response) {

ResultResponse* result = static_cast<ResultResponse*>(response);

if (result->row_count() == 0) {

LOG_ERROR("No row found for keyspace %s and type %s in system schema.",

keyspace_and_type_names.first.c_str(),

keyspace_and_type_names.second.c_str());

return;

}

Session* session = control_connection->session_;

int protocol_version = control_connection->protocol_version_;

const VersionNumber& cassandra_version = control_connection->cassandra_version_;

session->metadata().update_user_types(protocol_version, cassandra_version, result);

const StringRef& function_name,

const StringRefVec& arg_types,

bool is_aggregate) {

std::string query;

if (cassandra_version_ >= VersionNumber(3, 0, 0)) {

if (is_aggregate) {

query.assign(SELECT_AGGREGATES_30);

query.append(" WHERE keyspace_name=? AND aggregate_name=? AND argument_types=?");

} else {

query.assign(SELECT_FUNCTIONS_30);

query.append(" WHERE keyspace_name=? AND function_name=? AND argument_types=?");

}

} else {

if (is_aggregate) {

query.assign(SELECT_AGGREGATES_22);

query.append(" WHERE keyspace_name=? AND aggregate_name=? AND signature=?");

} else {

query.assign(SELECT_FUNCTIONS_22);