CPP-578 - Add support for NO_COMPACT startup option

Starting with Apache Cassandra v4.0, Thrift and COMPACT STORAGE is no
longer supported. For uninterrupted cluster upgrades, the C/C++ driver
supports the 'NO_COMPACT' startup option. Enabling this feature using
`cass_cluster_set_no_compact()` will have same effect as
'DROP COMPACT STORAGE', but only for the connected session.

include/cassandra.h

@@ -1865,6 +1865,28 @@ CASS_EXPORT CassError
 cass_cluster_set_prepare_on_up_or_add_host(CassCluster* cluster,
                                            cass_bool_t enabled);
 
+/**
+ * Enable the <b>NO_COMPACT</b> startup option.
+ *
+ * This can help facilitate uninterrupted cluster upgrades where tables using
+ * <b>COMPACT_STORAGE</b> will operate in "compatibility mode" for
+ * <b>BATCH</b>, <b>DELETE</b>, <b>SELECT</b>, and <b>UPDATE</b> CQL operations.
+ *
+ * <b>Default:</b> cass_false
+ *
+ * @cassandra{3.0.16+}
+ * @cassandra{3.11.2+}
+ * @cassandra{4.0+}
+ *
+ * @public @memberof CassCluster
+ *
+ * @param[in] cluster
+ * @param[in] enabled
+ */
+CASS_EXPORT CassError
+cass_cluster_set_no_compact(CassCluster* cluster,
+                            cass_bool_t enabled);
+
 /***********************************************************************************
  *
  * Session

src/cluster.cpp

@@ -465,6 +465,13 @@ CassError cass_cluster_set_prepare_on_up_or_add_host(CassCluster* cluster,
   return CASS_OK;
 }
 
+CassError cass_cluster_set_no_compact(CassCluster* cluster,
+                                 cass_bool_t enabled) {
+  cluster->config().set_no_compact(enabled == cass_true);
+  return CASS_OK;
+}
+
+
 void cass_cluster_free(CassCluster* cluster) {
   delete cluster->from();
 }

src/config.hpp

@@ -80,7 +80,8 @@ class Config {
       , use_schema_(true)
       , use_randomized_contact_points_(true)
       , prepare_on_all_hosts_(true)
-      , prepare_on_up_or_add_host_(true) { }
+      , prepare_on_up_or_add_host_(true)
+      , no_compact_(false) { }
 
   Config new_instance() const {
     Config config = *this;
@@ -394,6 +395,12 @@ class Config {
     prepare_on_up_or_add_host_ = enabled;
   }
 
+  bool no_compact() const { return no_compact_; }
+
+  void set_no_compact(bool enabled) {
+    no_compact_ = enabled;
+  }
+
 private:
   int port_;
   int protocol_version_;
@@ -441,6 +448,7 @@ class Config {
   bool use_randomized_contact_points_;
   bool prepare_on_all_hosts_;
   bool prepare_on_up_or_add_host_;
+  bool no_compact_;
 };
 
 } // namespace cass

src/connection.cpp

@@ -794,7 +794,7 @@ void Connection::on_supported(ResponseMessage* response) {
 
   internal_write(RequestCallback::Ptr(
                    new StartupCallback(Request::ConstPtr(
-                                         new StartupRequest()))));
+                                         new StartupRequest(config().no_compact())))));
 }
 
 void Connection::on_pending_schema_agreement(Timer* timer) {

src/startup_request.cpp

@@ -39,6 +39,14 @@ int StartupRequest::encode(int version, RequestCallback* callback, BufferVec* bu
     options[key] = version_;
   }
 
+  if (no_compact_enabled_) {
+    const char* key = "NO_COMPACT";
+    const char* value = "true";
+    length += 2 + strlen(key);
+    length += 2 + strlen(value);
+    options[key] = value;
+  }
+
   bufs->push_back(Buffer(length));
   bufs->back().encode_string_map(0, options);
 

src/startup_request.hpp

@@ -28,15 +28,15 @@ namespace cass {
 
 class StartupRequest : public Request {
 public:
-  StartupRequest()
+  StartupRequest(bool no_compact_enabled)
       : Request(CQL_OPCODE_STARTUP)
       , version_("3.0.0")
-      , compression_("") {}
-
-  bool encode(size_t reserved, char** output, size_t& size);
+      , compression_("")
+      , no_compact_enabled_(no_compact_enabled) { }
 
   const std::string version() const { return version_; }
   const std::string compression() const { return compression_; }
+  bool no_compact_enabled() const { return no_compact_enabled_; }
 
 private:
   int encode(int version, RequestCallback* callback, BufferVec* bufs) const;
@@ -46,6 +46,7 @@ class StartupRequest : public Request {
 
   std::string version_;
   std::string compression_;
+  bool no_compact_enabled_;
 };
 
 } // namespace cass