make sure prepared result metadata is replaced on reprepare

PYTHON-621

cassandra/cluster.py

@@ -3549,6 +3549,10 @@ class ResponseFuture(object):
 
         if isinstance(response, ResultMessage):
             if response.kind == RESULT_KIND_PREPARED:
+                # result metadata is the only thing that could have changed from an alter
+                _, _, _, result_metadata = response.results
+                self.prepared_statement.result_metadata = result_metadata
+
                 # use self._query to re-use the same host and
                 # at the same time properly borrow the connection
                 request_id = self._query(self._current_host)

tests/integration/standard/test_prepared_statements.py

@@ -23,7 +23,7 @@ from cassandra import InvalidRequest
 
 from cassandra import ConsistencyLevel
 from cassandra.cluster import Cluster
-from cassandra.query import PreparedStatement, UNSET_VALUE
+from cassandra.query import PreparedStatement, UNSET_VALUE, tuple_factory
 from tests.integration import get_server_versions
 
 
@@ -385,3 +385,39 @@ class PreparedStatementTests(unittest.TestCase):
 
         with self.assertRaises(InvalidRequest):
             self.session.execute(prepared, [0])
+
+    def test_invalidated_result_metadata(self):
+        """
+        Tests to make sure cached metadata is updated when an invalidated prepared statement is reprepared.
+
+        @since 2.7.0
+        @jira_ticket PYTHON-621
+
+        Prior to this fix, the request would blow up with a protocol error when the result was decoded expecting a different
+        number of columns.
+        """
+        s = self.session
+        s.result_factory = tuple_factory
+
+        table = "test1rf.%s" % self._testMethodName.lower()
+
+        s.execute("DROP TABLE IF EXISTS %s" % table)
+        s.execute("CREATE TABLE %s (k int PRIMARY KEY, a int, b int, c int)" % table)
+        s.execute("INSERT INTO %s (k, a, b, c) VALUES (0, 0, 0, 0)" % table)
+
+        wildcard_prepared = s.prepare("SELECT * FROM %s" % table)
+        original_result_metadata = wildcard_prepared.result_metadata
+        self.assertEqual(len(original_result_metadata), 4)
+
+        r = s.execute(wildcard_prepared)
+        self.assertEqual(r[0], (0, 0, 0, 0))
+
+        s.execute("ALTER TABLE %s DROP c" % table)
+
+        # Get a bunch of requests in the pipeline with varying states of result_meta, reprepare, resolved
+        futures = set(s.execute_async(wildcard_prepared.bind(None)) for _ in range(200))
+        for f in futures:
+            self.assertEqual(f.result()[0], (0, 0, 0))
+        self.assertIsNot(wildcard_prepared.result_metadata, original_result_metadata)
+        s.execute("DROP TABLE %s" % table)
+