CPP-802 - Handle prepared id mismatch when repreparing on the fly

Michael Fero · Michael Fero · commit cf9dddb07a41 · 2019-10-10T15:02:29.000-04:00
diff --git a/cpp-driver/gtests/src/integration/integration.cpp b/cpp-driver/gtests/src/integration/integration.cpp
@@ -56,6 +56,7 @@ Integration::Integration()
     , is_ccm_start_requested_(true)
     , is_ccm_start_node_individually_(false)
     , is_session_requested_(true)
+    , is_keyspace_change_requested_(true)
     , is_test_chaotic_(false)
     , is_beta_protocol_(Options::is_beta_protocol())
     , protocol_version_(CASS_HIGHEST_SUPPORTED_PROTOCOL_VERSION)
@@ -300,6 +301,16 @@ void Integration::drop_type(const std::string& type_name) {
   session_.execute(drop_type_query.str(), CASS_CONSISTENCY_ANY, false, false);
 }
 
+bool Integration::use_keyspace(const std::string& keyspace_name) {
+  std::stringstream use_keyspace_query;
+  use_keyspace_query << "USE " << keyspace_name;
+  session_.execute(use_keyspace_query.str());
+  if (this->HasFailure()) {
+    return false;
+  }
+  return true;
+}
+
 void Integration::connect(Cluster cluster) {
   // Establish the session connection
   cluster_ = cluster;
@@ -322,10 +333,9 @@ void Integration::connect(Cluster cluster) {
   CHECK_FAILURE;
 
   // Update the session to use the new keyspace by default
-  std::stringstream use_keyspace_query;
-  use_keyspace_query << "USE " << keyspace_name_;
-  session_.execute(use_keyspace_query.str());
-  CHECK_FAILURE;
+  if (is_keyspace_change_requested_) {
+    use_keyspace(keyspace_name_);
+  }
 }
 
 void Integration::connect() {
diff --git a/cpp-driver/gtests/src/integration/integration.hpp b/cpp-driver/gtests/src/integration/integration.hpp
@@ -109,7 +109,10 @@
 
 #define CASSANDRA_KEY_VALUE_TABLE_FORMAT \
   "CREATE TABLE IF NOT EXISTS %s (key %s PRIMARY KEY, value %s)"
+#define CASSANDRA_KEY_VALUE_QUALIFIED_TABLE_FORMAT \
+  "CREATE TABLE IF NOT EXISTS %s.%s (key %s PRIMARY KEY, value %s)"
 #define CASSANDRA_KEY_VALUE_INSERT_FORMAT "INSERT INTO %s (key, value) VALUES(%s, %s)"
+#define CASSANDRA_KEY_VALUE_QUALIFIED_INSERT_FORMAT "INSERT INTO %s.%s (key, value) VALUES(%s, %s)"
 #define CASSANDRA_SELECT_VALUE_FORMAT "SELECT value FROM %s WHERE key=%s"
 #define CASSANDRA_DELETE_ROW_FORMAT "DELETE FROM %s WHERE key=%s"
 #define CASSANDRA_UPDATE_VALUE_FORMAT "UPDATE %s SET value=%s WHERE key=%s"
@@ -279,6 +282,11 @@ class Integration : public testing::Test {
    * (DEFAULT: true)
    */
   bool is_session_requested_;
+  /**
+   * Flag to indicate if the newly created keyspace should be set for the session connection.
+   * (DEFAULT: true)
+   */
+  bool is_keyspace_change_requested_;
   /**
    * Flag to indicate if a test is chaotic and should have its CCM cluster
    * destroyed
@@ -376,6 +384,14 @@ class Integration : public testing::Test {
    */
   virtual void drop_type(const std::string& type_name);
 
+  /**
+   * Update the current keyspace used by the session
+   *
+   * @param keyspace_name Keyspace to use
+   * @return True if keyspace was changed; false otherwise
+   */
+  virtual bool use_keyspace(const std::string& keyspace_name);
+
   /**
    * Establish the session connection using provided cluster object.
    *
diff --git a/cpp-driver/gtests/src/integration/tests/test_prepared.cpp b/cpp-driver/gtests/src/integration/tests/test_prepared.cpp
@@ -0,0 +1,110 @@
+/*
+  Copyright (c) DataStax, Inc.
+
+  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 "integration.hpp"
+
+/**
+ * Prepared integration tests; common operations
+ */
+class PreparedTests : public Integration {
+  void SetUp() {
+    is_keyspace_change_requested_ = false;
+    Integration::SetUp();
+  }
+};
+
+/**
+ * Execute a statement that forces a re-prepare resulting in a new prepared ID that fails fast and
+ * returns an error.
+ *
+ * This test will create a new table, prepare a statement using a fully qualified query, update the
+ * default keyspace, then drop and re-create the table to force the server to invalidate the
+ * prepared ID. After the table is dropped the prepared statement will be used to execute an insert
+ * query that will result in an error being returned when re-using the original prepared statement.
+ *
+ * @see: https://issues.apache.org/jira/browse/CASSANDRA-15252 (Server version restriction may need
+ *       to be added if/when Apache Cassandra issue is addressed.
+ *
+ * @test_category error
+ * @test_category queries:prepared
+ * @since core:2.14.0
+ * @expected_result Re-prepare will fail fast and return error.
+ */
+CASSANDRA_INTEGRATION_TEST_F(PreparedTests, FailFastWhenPreparedIDChangesDuringReprepare) {
+  CHECK_FAILURE;
+
+  // Create the table and initial prepared statement
+  session_.execute(format_string(CASSANDRA_KEY_VALUE_QUALIFIED_TABLE_FORMAT, keyspace_name_.c_str(),
+                                 table_name_.c_str(), "int", "int"));
+  Prepared insert_prepared =
+      session_.prepare(format_string(CASSANDRA_KEY_VALUE_QUALIFIED_INSERT_FORMAT,
+                                     keyspace_name_.c_str(), table_name_.c_str(), "?", "?"));
+
+  // Update the current keyspace for the session
+  ASSERT_TRUE(use_keyspace(keyspace_name_));
+
+  // Drop and re-create the table to invalidate the prepared statement on the server
+  drop_table(table_name_);
+  session_.execute(format_string(CASSANDRA_KEY_VALUE_QUALIFIED_TABLE_FORMAT, keyspace_name_.c_str(),
+                                 table_name_.c_str(), "int", "int"));
+
+  // Execute the insert statement and validate the error code
+  logger_.add_critera("ID mismatch while trying to prepare query");
+  Statement insert_statement = insert_prepared.bind();
+  insert_statement.bind<Integer>(0, Integer(0));
+  insert_statement.bind<Integer>(1, Integer(1));
+  Result result = session_.execute(insert_statement, false);
+  EXPECT_TRUE(contains(result.error_message(), "ID mismatch while trying to prepare query"));
+}
+
+/**
+ * Execute a statement that forces a re-prepare resulting in a same prepared ID.
+ *
+ * This test will connect to a cluster and use a keyspace, prepare a statement using a unqualified
+ * query, then drop and re-create the table to force the server to invalidate the
+ * prepared ID. After the table is dropped the prepared statement will be used to execute an insert
+ * query that will result the statement being re-prepared and the insert statement succeeding.
+ *
+ * @test_category queries:prepared
+ * @since core:1.0.0
+ * @expected_result Re-prepare will correctly execute the insert statement.
+ */
+CASSANDRA_INTEGRATION_TEST_F(PreparedTests, PreparedIDUnchangedDuringReprepare) {
+  CHECK_FAILURE;
+
+  // Allow for unqualified queries
+  use_keyspace(keyspace_name_);
+
+  // Create the table and initial prepared statement
+  session_.execute(
+      format_string(CASSANDRA_KEY_VALUE_TABLE_FORMAT, table_name_.c_str(), "int", "int"));
+  Prepared insert_prepared = session_.prepare(
+      format_string(CASSANDRA_KEY_VALUE_INSERT_FORMAT, table_name_.c_str(), "?", "?"));
+
+  // Drop and re-create the table to invalidate the prepared statement on the server
+  drop_table(table_name_);
+  session_.execute(
+      format_string(CASSANDRA_KEY_VALUE_TABLE_FORMAT, table_name_.c_str(), "int", "int"));
+
+  // Execute the insert statement and validate success
+  logger_.add_critera("Prepared query with ID");
+  Statement insert_statement = insert_prepared.bind();
+  insert_statement.bind<Integer>(0, Integer(0));
+  insert_statement.bind<Integer>(1, Integer(1));
+  Result result = session_.execute(insert_statement, false);
+  EXPECT_EQ(CASS_OK, result.error_code());
+  EXPECT_EQ(1u, logger_.count());
+}
diff --git a/cpp-driver/src/request_handler.cpp b/cpp-driver/src/request_handler.cpp
@@ -36,6 +36,20 @@ using namespace datastax;
 using namespace datastax::internal;
 using namespace datastax::internal::core;
 
+static String to_hex(const String& byte_id) {
+  static const char half_byte_to_hex[] = { '0', '1', '2', '3', '4', '5', '6', '7',
+                                           '8', '9', 'a', 'b', 'c', 'd', 'e', 'f' };
+  OStringStream ss;
+
+  const char* data = byte_id.data();
+  for (size_t i = 0; i < byte_id.length(); ++i) {
+    uint8_t byte = static_cast<uint8_t>(data[i]);
+    ss << half_byte_to_hex[(byte >> 4) & 0x0F];
+    ss << half_byte_to_hex[byte & 0x0F];
+  }
+  return ss.str();
+}
+
 class SingleHostQueryPlan : public QueryPlan {
 public:
   SingleHostQueryPlan(const Address& address)
@@ -53,7 +67,7 @@ class SingleHostQueryPlan : public QueryPlan {
 
 class PrepareCallback : public SimpleRequestCallback {
 public:
-  PrepareCallback(const String& query, RequestExecution* request_execution);
+  PrepareCallback(const String& query, const String& id, RequestExecution* request_execution);
 
 private:
   class PrepareRequest : public core::PrepareRequest {
@@ -72,21 +86,29 @@ class PrepareCallback : public SimpleRequestCallback {
 
 private:
   RequestExecution::Ptr request_execution_;
+  String id_;
 };
 
-PrepareCallback::PrepareCallback(const String& query, RequestExecution* request_execution)
+PrepareCallback::PrepareCallback(const String& query, const String& id,
+                                 RequestExecution* request_execution)
     : SimpleRequestCallback(
           Request::ConstPtr(new PrepareRequest(query, request_execution->request()->keyspace(),
                                                request_execution->request_timeout_ms())))
-    , request_execution_(request_execution) {}
+    , request_execution_(request_execution)
+    , id_(id) {}
 
 void PrepareCallback::on_internal_set(ResponseMessage* response) {
   switch (response->opcode()) {
     case CQL_OPCODE_RESULT: {
       ResultResponse* result = static_cast<ResultResponse*>(response->response_body().get());
       if (result->kind() == CASS_RESULT_KIND_PREPARED) {
-        request_execution_->notify_result_metadata_changed(request(), result);
-        request_execution_->on_retry_current_host();
+        String result_id = result->prepared_id().to_string();
+        if (id_ != result_id) {
+          request_execution_->notify_prepared_id_mismatch(id_, result_id);
+        } else {
+          request_execution_->notify_result_metadata_changed(request(), result);
+          request_execution_->on_retry_current_host();
+        }
       } else {
         request_execution_->on_retry_next_host();
       }
@@ -461,6 +483,17 @@ void RequestExecution::notify_result_metadata_changed(const Request* request,
   }
 }
 
+void RequestExecution::notify_prepared_id_mismatch(const String& expected_id,
+                                                   const String& received_id) {
+  OStringStream ss;
+  ss << "ID mismatch while trying to prepare query (expected ID " << to_hex(expected_id)
+     << ", received ID " << to_hex(received_id)
+     << "). This prepared statement won't work anymore. This usually happens when you run a "
+        "'USE...' query after the statement was prepared.";
+  String message = ss.str();
+  request_handler_->set_error(CASS_ERROR_LIB_UNEXPECTED_RESPONSE, message);
+}
+
 void RequestExecution::on_result_response(Connection* connection, ResponseMessage* response) {
   ResultResponse* result = static_cast<ResultResponse*>(response->response_body().get());
 
@@ -602,14 +635,17 @@ void RequestExecution::on_error_response(Connection* connection, ResponseMessage
 }
 
 void RequestExecution::on_error_unprepared(Connection* connection, ErrorResponse* error) {
-  String query;
+  LOG_DEBUG("Unprepared error response returned for request: %s",
+            error->message().to_string().c_str());
 
+  String query;
+  String id = error->prepared_id().to_string();
   if (request()->opcode() == CQL_OPCODE_EXECUTE) {
     const ExecuteRequest* execute = static_cast<const ExecuteRequest*>(request());
     query = execute->prepared()->query();
   } else if (request()->opcode() == CQL_OPCODE_BATCH) {
     const BatchRequest* batch = static_cast<const BatchRequest*>(request());
-    if (!batch->find_prepared_query(error->prepared_id().to_string(), &query)) {
+    if (!batch->find_prepared_query(id, &query)) {
       set_error(CASS_ERROR_LIB_UNEXPECTED_RESPONSE,
                 "Unable to find prepared statement in batch statement");
       return;
@@ -621,7 +657,7 @@ void RequestExecution::on_error_unprepared(Connection* connection, ErrorResponse
     return;
   }
 
-  RequestCallback::Ptr callback(new PrepareCallback(query, this));
+  RequestCallback::Ptr callback(new PrepareCallback(query, id, this));
   if (connection->write_and_flush(callback) < 0) {
     // Try to prepare on the same host but on a different connection
     retry_current_host();
diff --git a/cpp-driver/src/request_handler.hpp b/cpp-driver/src/request_handler.hpp
@@ -275,6 +275,7 @@ class RequestExecution : public RequestCallback {
   void next_host() { current_host_ = request_handler_->next_host(RequestHandler::Protected()); }
 
   void notify_result_metadata_changed(const Request* request, ResultResponse* result_response);
+  void notify_prepared_id_mismatch(const String& expected_id, const String& received_id);
 
   virtual void on_retry_current_host();
   virtual void on_retry_next_host();
