query: Add cql_format property to partition_slice

It will specify in which format CQL values should be serialized. Will
allow for rolling out new CQL binary protocol versions without
stalling reads.

idl/read_command.idl.hh

@@ -19,7 +19,12 @@
  * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
  */
 
+class cql_serialization_format final {
+    uint8_t protocol_version();
+};
+
 namespace query {
+
 class specific_ranges {
     partition_key pk();
     std::vector<range<clustering_key_prefix>> ranges();
@@ -31,6 +36,7 @@ class partition_slice {
     std::vector<uint32_t> regular_columns;
     query::partition_slice::option_set options;
     std::unique_ptr<query::specific_ranges> get_specific_ranges();
+    cql_serialization_format cql_format();
 };
 
 class read_command {
@@ -40,4 +46,5 @@ class read_command {
     uint32_t row_limit;
     std::chrono::time_point<gc_clock, gc_clock::duration> timestamp;
 };
+
 }

query-request.hh

@@ -114,10 +114,12 @@ public:
     option_set options;
 private:
     std::unique_ptr<specific_ranges> _specific_ranges;
+    cql_serialization_format _cql_format;
 public:
     partition_slice(clustering_row_ranges row_ranges, std::vector<column_id> static_columns,
         std::vector<column_id> regular_columns, option_set options,
-        std::unique_ptr<specific_ranges> specific_ranges = nullptr);
+        std::unique_ptr<specific_ranges> specific_ranges = nullptr,
+        cql_serialization_format = cql_serialization_format::internal());
     partition_slice(const partition_slice&);
     partition_slice(partition_slice&&);
     ~partition_slice();
@@ -132,6 +134,9 @@ public:
     const std::unique_ptr<specific_ranges>& get_specific_ranges() const {
         return _specific_ranges;
     }
+    const cql_serialization_format& cql_format() const {
+        return _cql_format;
+    }
 
     friend std::ostream& operator<<(std::ostream& out, const partition_slice& ps);
     friend std::ostream& operator<<(std::ostream& out, const specific_ranges& ps);

query-result-set.cc

@@ -172,7 +172,7 @@ result_set_builder::deserialize(const result_row_view& row, bool is_static)
             if (cell) {
                 auto ctype = static_pointer_cast<const collection_type_impl>(col.type);
                 auto view = cell.value();
-                auto normal_form = ctype->to_value(ctype->deserialize_mutation_form(view), cql_serialization_format::internal());
+                auto normal_form = ctype->to_value(ctype->deserialize_mutation_form(view), _slice.cql_format());
                 cells.emplace(col.name_as_text(), col.type->deserialize_value(normal_form));
             }
         }

query.cc

@@ -44,8 +44,9 @@ std::ostream& operator<<(std::ostream& out, const partition_slice& ps) {
     if (ps._specific_ranges) {
         out << ", specific=[" << *ps._specific_ranges << "]";
     }
-    return out << ", options=" << sprint("%x", ps.options.mask()) // FIXME: pretty print options
-        << "}";
+    out << ", options=" << sprint("%x", ps.options.mask()); // FIXME: pretty print options
+    out << ", cql_format=" << ps.cql_format();
+    return out << "}";
 }
 
 std::ostream& operator<<(std::ostream& out, const read_command& r) {
@@ -61,13 +62,18 @@ std::ostream& operator<<(std::ostream& out, const specific_ranges& s) {
     return out << "{" << s._pk << " : " << join(", ", s._ranges) << "}";
 }
 
-partition_slice::partition_slice(clustering_row_ranges row_ranges, std::vector<column_id> static_columns,
-    std::vector<column_id> regular_columns, option_set options, std::unique_ptr<specific_ranges> specific_ranges)
+partition_slice::partition_slice(clustering_row_ranges row_ranges,
+    std::vector<column_id> static_columns,
+    std::vector<column_id> regular_columns,
+    option_set options,
+    std::unique_ptr<specific_ranges> specific_ranges,
+    cql_serialization_format cql_format)
     : _row_ranges(std::move(row_ranges))
     , static_columns(std::move(static_columns))
     , regular_columns(std::move(regular_columns))
     , options(options)
     , _specific_ranges(std::move(specific_ranges))
+    , _cql_format(std::move(cql_format))
 {}
 
 partition_slice::partition_slice(partition_slice&&) = default;
@@ -80,6 +86,7 @@ partition_slice::partition_slice(const partition_slice& s)
     , regular_columns(s.regular_columns)
     , options(s.options)
     , _specific_ranges(s._specific_ranges ? std::make_unique<specific_ranges>(*s._specific_ranges) : nullptr)
+    , _cql_format(s._cql_format)
 {}
 
 partition_slice::~partition_slice()