Sending not needed CQL Query "USE <keyspace>" to a new host before PREPARE CQL message.

[OlegLoginov]

If a new CQL host was added to the driver (including the host just down & up) - the drivers resends all Prepared statement to the host for the session context restoring.
If the Prepared Query is stored with 'keyspace', the driver sends 'USE ' (in a special temporary connection) before the PREPARE CQL request.
The 'USE' query is needed if the statement uses short non-qualified table names.
But if the prepared statement uses fully qualified table names the additional 'USE ' query is useless. And only affects the time of the new connection establishing, which can affect the system startup time if we have many new connections at the same time.

The issue fix must prevent these not needed 'USE ' queries from the driver.

Implementation notes:
ResultResponse as result of Prepare request has metadata including table name & keyspace related to the prepared statement.
The keyspace is stored in Session together with prepared_id & prepared_query. The keyspace should be cleaned or ignored when the prepared statement uses only fully qualified table names to prevent the discussed 'USE ' queries when a new host is added.
For non-qualified table names in prepared statement the 'USE' query makes sense and the driver behavior should still unchanged.

[OlegLoginov]

Fixed by the commit above.

[OlegLoginov]

Notes for the code changes:

    // Attempt to use the per-query keyspace first (v5+/DSEv2+ only) then
    // the keyspace in the result metadata.
    std::string keyspace;
    bool keyspace_from_response = false;
    if (supports_set_keyspace(result_response->protocol_version()) &&
        !request->keyspace().empty()) {
      keyspace = request->keyspace();             <<<< [1]
    } else {
      keyspace = result_response->keyspace().to_string();   <<< [2]
      keyspace_from_response = true;
    }

The 'keyspace' can be taken
[1] from the original request
For example if the user calls the API functions:

CassStatement*  cass_statement_new(const char* query, size_t parameter_count);
CassError  cass_statement_set_keyspace(CassStatement* statement, const char* keyspace);
CassFuture*  cass_session_prepare_from_existing(CassSession* session, CassStatement* statement);

But currently (with protocol_version==4) this branch does not work because:

inline bool supports_set_keyspace(int version) {
  return version >= CASS_PROTOCOL_VERSION_V5;
}

[2] from the ResultResponse metadata - that's our case, and only for this branch we can clean the 'keyspace' value.

    } else if (request->opcode()  == CQL_OPCODE_PREPARE &&
               result_response->kind() == CASS_RESULT_KIND_PREPARED) {
      if (keyspace_from_response && connection() && connection()->keyspace().empty()) {
        // Clear keyspace if keyspace was not set in connection.
        // This prevents sending useless 'USE <ks>' to new CQL hosts.
        keyspace.clear();

The new code works for a response to PREPARE request (CASS_RESULT_KIND_PREPARED).
Checking of keyspace_from_response guaranties that the driver behavior for case [1] is unchanged.
If the user has done "USE " on the session level - the driver behavior should not be changed too. For this we check the current keyspace in the connection.

[OlegLoginov]

If the user sends "USE " - and the request was successfully processed - the driver goes through the code path:

void Connection::consume(char* input, size_t size) {
...
          switch (callback->state()) {
            case RequestCallback::REQUEST_STATE_READING:
              pending_reads_.remove(callback.get());
              callback->set_state(RequestCallback::REQUEST_STATE_FINISHED);
              maybe_set_keyspace(response.get());    <<<<< UPDATING KEYSPACE IN THIS CONNECTION


void RequestExecution::on_result_response(Connection* connection, ResponseMessage* response) {
...
    case CASS_RESULT_KIND_SET_KEYSPACE:
      request_handler_->io_worker()->broadcast_keyspace_change(result->keyspace().to_string());     <<<<<<<<<
...

void IOWorker::broadcast_keyspace_change(const std::string& keyspace) {
  set_keyspace(keyspace);
  session_->broadcast_keyspace_change(keyspace, this);   <<<<<<<<

void Session::broadcast_keyspace_change(const std::string& keyspace,
                                        const IOWorker* calling_io_worker) {
  // This can run on an IO worker thread. This is thread-safe because the IO workers
  // vector never changes after initialization and IOWorker::set_keyspace() uses
  // a lock. This also means that calling "USE <keyspace>" frequently is an
  // anti-pattern.
  for (IOWorkerVec::iterator it = io_workers_.begin(),
       end = io_workers_.end(); it != end; ++it) {
    if (*it == calling_io_worker) continue;
      (*it)->set_keyspace(keyspace);   <<<<<<<<<<<<  UPDATING ALL WORKERS
  }

  set_keyspace(keyspace);   <<<<< UPDATING SESSION KEYSPACE
...