Decoupled token-aware routing from enabling schema metadata

Author: Michael Penick

include/cassandra.h

@@ -1329,9 +1329,9 @@ cass_cluster_set_load_balance_dc_aware_n(CassCluster* cluster,
 /**
  * Configures the cluster to use token-aware request routing or not.
  *
- * <b>Important:</b> Token-aware routing depends on keyspace information.
- * For this reason enabling token-aware routing will also enable the usage
- * of schema metadata.
+ * <b>Important:</b> Token-aware routing depends on keyspace metadata.
+ * For this reason enabling token-aware routing will also enable retrieving
+ * and updating keyspace schema metadata.
  *
  * <b>Default:</b> cass_true (enabled).
  *
@@ -1343,8 +1343,6 @@ cass_cluster_set_load_balance_dc_aware_n(CassCluster* cluster,
  *
  * @param[in] cluster
  * @param[in] enabled
- *
- * @see cass_cluster_set_use_schema();
  */
 CASS_EXPORT void
 cass_cluster_set_token_aware_routing(CassCluster* cluster,
@@ -1665,9 +1663,8 @@ cass_cluster_set_retry_policy(CassCluster* cluster,
 /**
  * Enable/Disable retrieving and updating schema metadata. If disabled
  * this is allows the driver to skip over retrieving and updating schema
- * metadata, but it also disables the usage of token-aware routing and
- * cass_session_get_schema_meta() will always return an empty object. This can
- * be useful for reducing the startup overhead of short-lived sessions.
+ * metadata and cass_session_get_schema_meta() will always return an empty object.
+ * This can be useful for reducing the startup overhead of short-lived sessions.
  *
  * <b>Default:</b> cass_true (enabled).
  *
@@ -1677,7 +1674,6 @@ cass_cluster_set_retry_policy(CassCluster* cluster,
  * @param[in] enabled
  *
  * @see cass_session_get_schema_meta()
- * @see cass_cluster_set_token_aware_routing()
  */
 CASS_EXPORT void
 cass_cluster_set_use_schema(CassCluster* cluster,

src/cluster.cpp

@@ -259,10 +259,6 @@ CassError cass_cluster_set_load_balance_dc_aware_n(CassCluster* cluster,
 void cass_cluster_set_token_aware_routing(CassCluster* cluster,
                                           cass_bool_t enabled) {
   cluster->config().set_token_aware_routing(enabled == cass_true);
-  // Token-aware routing relies on up-to-date schema information
-  if (enabled == cass_true) {
-    cluster->config().set_use_schema(true);
-  }
 }
 
 void cass_cluster_set_latency_aware_routing(CassCluster* cluster,
@@ -407,10 +403,6 @@ void cass_cluster_set_timestamp_gen(CassCluster* cluster,
 void cass_cluster_set_use_schema(CassCluster* cluster,
                                  cass_bool_t enabled) {
   cluster->config().set_use_schema(enabled == cass_true);
-  // Token-aware routing relies on up-to-date schema information
-  if (enabled == cass_false) {
-    cluster->config().set_token_aware_routing(false);
-  }
 }
 
 CassError cass_cluster_set_use_hostname_resolution(CassCluster* cluster,

src/control_connection.cpp

@@ -112,7 +112,8 @@ ControlConnection::ControlConnection()
   , session_(NULL)
   , connection_(NULL)
   , protocol_version_(0)
-  , should_query_tokens_(false) {}
+  , use_schema_(false)
+  , token_aware_routing_(false) { }
 
 const SharedRefPtr<Host>& ControlConnection::connected_host() const {
   return current_host_;
@@ -126,19 +127,21 @@ void ControlConnection::clear() {
   query_plan_.reset();
   protocol_version_ = 0;
   last_connection_error_.clear();
-  should_query_tokens_ = false;
+  use_schema_ = false;
+  token_aware_routing_ = false;
 }
 
 void ControlConnection::connect(Session* session) {
   session_ = session;
   query_plan_.reset(new ControlStartupQueryPlan(session_->hosts_)); // No hosts lock necessary (read-only)
   protocol_version_ = session_->config().protocol_version();
-  should_query_tokens_ = session_->config().token_aware_routing();
+  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 (session_->config().use_schema()) {
+  if (use_schema_ || token_aware_routing_) {
     set_event_types(CASS_EVENT_TOPOLOGY_CHANGE | CASS_EVENT_STATUS_CHANGE |
                     CASS_EVENT_SCHEMA_CHANGE);
   } else {
@@ -325,10 +328,17 @@ void ControlConnection::on_event(EventResponse* response) {
     }
 
     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:
@@ -390,8 +400,8 @@ void ControlConnection::on_event(EventResponse* response) {
 void ControlConnection::query_meta_hosts() {
   ScopedRefPtr<ControlMultipleRequestHandler<UnusedData> > handler(
         new ControlMultipleRequestHandler<UnusedData>(this, ControlConnection::on_query_hosts, UnusedData()));
-  handler->execute_query("local", SELECT_LOCAL_TOKENS);
-  handler->execute_query("peers", SELECT_PEERS_TOKENS);
+  handler->execute_query("local", token_aware_routing_ ? SELECT_LOCAL_TOKENS : SELECT_LOCAL);
+  handler->execute_query("peers", token_aware_routing_ ? SELECT_PEERS_TOKENS : SELECT_PEERS);
 }
 
 void ControlConnection::on_query_hosts(ControlConnection* control_connection,
@@ -472,7 +482,8 @@ void ControlConnection::on_query_hosts(ControlConnection* control_connection,
 
   session->purge_hosts(is_initial_connection);
 
-  if (session->config().use_schema()) {
+  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;
@@ -490,24 +501,32 @@ void ControlConnection::query_meta_schema() {
         new ControlMultipleRequestHandler<UnusedData>(this, ControlConnection::on_query_meta_schema, UnusedData()));
 
   if (cassandra_version_ >= VersionNumber(3, 0, 0)) {
-    handler->execute_query("keyspaces", SELECT_KEYSPACES_30);
-    handler->execute_query("tables", SELECT_TABLES_30);
-    handler->execute_query("views", SELECT_VIEWS_30);
-    handler->execute_query("columns", SELECT_COLUMNS_30);
-    handler->execute_query("indexes", SELECT_INDEXES_30);
-    handler->execute_query("user_types", SELECT_USERTYPES_30);
-    handler->execute_query("functions", SELECT_FUNCTIONS_30);
-    handler->execute_query("aggregates", SELECT_AGGREGATES_30);
+    if (use_schema_ || token_aware_routing_) {
+      handler->execute_query("keyspaces", SELECT_KEYSPACES_30);
+    }
+    if (use_schema_) {
+      handler->execute_query("tables", SELECT_TABLES_30);
+      handler->execute_query("views", SELECT_VIEWS_30);
+      handler->execute_query("columns", SELECT_COLUMNS_30);
+      handler->execute_query("indexes", SELECT_INDEXES_30);
+      handler->execute_query("user_types", SELECT_USERTYPES_30);
+      handler->execute_query("functions", SELECT_FUNCTIONS_30);
+      handler->execute_query("aggregates", SELECT_AGGREGATES_30);
+    }
   } else {
-    handler->execute_query("keyspaces", SELECT_KEYSPACES_20);
-    handler->execute_query("tables", SELECT_COLUMN_FAMILIES_20);
-    handler->execute_query("columns", SELECT_COLUMNS_20);
-    if (cassandra_version_ >= VersionNumber(2, 1, 0)) {
-      handler->execute_query("user_types", SELECT_USERTYPES_21);
+    if (use_schema_ || token_aware_routing_) {
+      handler->execute_query("keyspaces", SELECT_KEYSPACES_20);
     }
-    if (cassandra_version_ >= VersionNumber(2, 2, 0)) {
-      handler->execute_query("functions", SELECT_FUNCTIONS_22);
-      handler->execute_query("aggregates", SELECT_AGGREGATES_22);
+    if (use_schema_) {
+      handler->execute_query("tables", SELECT_COLUMN_FAMILIES_20);
+      handler->execute_query("columns", SELECT_COLUMNS_20);
+      if (cassandra_version_ >= VersionNumber(2, 1, 0)) {
+        handler->execute_query("user_types", SELECT_USERTYPES_21);
+      }
+      if (cassandra_version_ >= VersionNumber(2, 2, 0)) {
+        handler->execute_query("functions", SELECT_FUNCTIONS_22);
+        handler->execute_query("aggregates", SELECT_AGGREGATES_22);
+      }
     }
   }
 }
@@ -524,58 +543,63 @@ void ControlConnection::on_query_meta_schema(ControlConnection* control_connecti
   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_) {
     session->token_map_->clear_keyspaces();
+
+    ResultResponse* keyspaces_result;
+    if (MultipleRequestHandler::get_result_response(responses, "keyspaces", &keyspaces_result)) {
+      session->token_map_->add_keyspaces(cassandra_version, keyspaces_result);
+    }
   }
 
-  session->metadata().clear_and_update_back(cassandra_version);
+  if (control_connection->use_schema_) {
+    session->metadata().clear_and_update_back(cassandra_version);
 
-  bool is_initial_connection = (control_connection->state_ == CONTROL_STATE_NEW);
+    ResultResponse* keyspaces_result;
+    if (MultipleRequestHandler::get_result_response(responses, "keyspaces", &keyspaces_result)) {
+      session->metadata().update_keyspaces(protocol_version, cassandra_version, keyspaces_result);
+    }
 
-  ResultResponse* keyspaces_result;
-  if (MultipleRequestHandler::get_result_response(responses, "keyspaces", &keyspaces_result)) {
-    if (session->token_map_) {
-      session->token_map_->add_keyspaces(cassandra_version, keyspaces_result);
+    ResultResponse* tables_result;
+    if (MultipleRequestHandler::get_result_response(responses, "tables", &tables_result)) {
+      session->metadata().update_tables(protocol_version, cassandra_version, tables_result);
     }
-    session->metadata().update_keyspaces(protocol_version, cassandra_version, keyspaces_result);
-  }
 
-  ResultResponse* tables_result;
-  if (MultipleRequestHandler::get_result_response(responses, "tables", &tables_result)) {
-    session->metadata().update_tables(protocol_version, cassandra_version, tables_result);
-  }
+    ResultResponse* views_result;
+    if (MultipleRequestHandler::get_result_response(responses, "views", &views_result)) {
+      session->metadata().update_views(protocol_version, cassandra_version, views_result);
+    }
 
-  ResultResponse* views_result;
-  if (MultipleRequestHandler::get_result_response(responses, "views", &views_result)) {
-    session->metadata().update_views(protocol_version, cassandra_version, views_result);
-  }
+    ResultResponse* columns_result = NULL;
+    if (MultipleRequestHandler::get_result_response(responses, "columns", &columns_result)) {
+      session->metadata().update_columns(protocol_version, cassandra_version, columns_result);
+    }
 
-  ResultResponse* columns_result = NULL;
-  if (MultipleRequestHandler::get_result_response(responses, "columns", &columns_result)) {
-    session->metadata().update_columns(protocol_version, cassandra_version, columns_result);
-  }
+    ResultResponse* indexes_result;
+    if (MultipleRequestHandler::get_result_response(responses, "indexes", &indexes_result)) {
+      session->metadata().update_indexes(protocol_version, cassandra_version, indexes_result);
+    }
 
-  ResultResponse* indexes_result;
-  if (MultipleRequestHandler::get_result_response(responses, "indexes", &indexes_result)) {
-    session->metadata().update_indexes(protocol_version, cassandra_version, indexes_result);
-  }
+    ResultResponse* user_types_result;
+    if (MultipleRequestHandler::get_result_response(responses, "user_types", &user_types_result)) {
+      session->metadata().update_user_types(protocol_version, cassandra_version, user_types_result);
+    }
 
-  ResultResponse* user_types_result;
-  if (MultipleRequestHandler::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 (MultipleRequestHandler::get_result_response(responses, "functions", &functions_result)) {
+      session->metadata().update_functions(protocol_version, cassandra_version, functions_result);
+    }
 
-  ResultResponse* functions_result;
-  if (MultipleRequestHandler::get_result_response(responses, "functions", &functions_result)) {
-    session->metadata().update_functions(protocol_version, cassandra_version, functions_result);
-  }
+    ResultResponse* aggregates_result;
+    if (MultipleRequestHandler::get_result_response(responses, "aggregates", &aggregates_result)) {
+      session->metadata().update_aggregates(protocol_version, cassandra_version, aggregates_result);
+    }
 
-  ResultResponse* aggregates_result;
-  if (MultipleRequestHandler::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();
   }
 
-  session->metadata().swap_to_back_and_update_front();
   if (session->token_map_) {
     session->token_map_->build();
   }
@@ -601,7 +625,7 @@ void ControlConnection::refresh_node_info(SharedRefPtr<Host> host,
   std::string query;
   ControlHandler<RefreshNodeData>::ResponseCallback response_callback;
 
-  bool token_query = should_query_tokens_ && (host->was_just_added() || query_tokens);
+  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);
@@ -739,7 +763,7 @@ void ControlConnection::update_node_info(SharedRefPtr<Host> host, const Row* row
              host->address().to_string().c_str());
   }
 
-  if (should_query_tokens_) {
+  if (token_aware_routing_) {
     bool is_connected_host = connection_ != NULL && host->address().compare(connection_->address()) == 0;
     std::string partitioner;
     if (is_connected_host && row->get_string_by_name("partitioner", &partitioner)) {
@@ -798,7 +822,10 @@ void ControlConnection::on_refresh_keyspace(ControlConnection* control_connectio
   if (session->token_map_) {
     session->token_map_->update_keyspaces_and_build(cassandra_version, result);
   }
-  session->metadata().update_keyspaces(protocol_version, cassandra_version, result);
+
+  if (control_connection->use_schema_) {
+    session->metadata().update_keyspaces(protocol_version, cassandra_version, result);
+  }
 }
 
 void ControlConnection::refresh_table_or_view(const StringRef& keyspace_name,

src/control_connection.hpp

@@ -252,7 +252,8 @@ class ControlConnection : public Connection::Listener {
   int protocol_version_;
   VersionNumber cassandra_version_;
   std::string last_connection_error_;
-  bool should_query_tokens_;
+  bool use_schema_;
+  bool token_aware_routing_;
 
   static Address bind_any_ipv4_;
   static Address bind_any_ipv6_;