db: prevent accidental copies of result_set_row by making it move-only

result_set_row is a heavyweight object containing multiple cell types:
regular columns, partition keys, and static values. To prevent expensive
accidental copies, delete the copy constructor and replace it with:

1. A move constructor for efficient vector reallocation
2. An explicit copy() method when copies are actually needed

This change reduces overhead in some non-hot paths by eliminating implicit
deep copies. Please note, previously, in `create_view_from_mutation()`,
we kept a copy of `result_set_row`, and then reused `table_rs` for
holding the mutation for `scylla_tables`. Because we don't copy
the `result_set_row` in this change, in order to avoid invalidating
the `row` after reusing `table_rs` in the outer scope, we define a
new `table_rs` shadowing the one in the out scope.

Signed-off-by: Kefu Chai <[email protected]>

Closes scylladb#22741

Author: tchaikov

db/schema_applier.cc

@@ -121,7 +121,7 @@ static std::optional<table_id> table_id_from_mutations(const schema_mutations& s
     if (table_rs.empty()) {
         return std::nullopt;
     }
-    query::result_set_row table_row = table_rs.row(0);
+    const query::result_set_row& table_row = table_rs.row(0);
     return table_id(table_row.get_nonnull<utils::UUID>("id"));
 }
 

db/schema_tables.cc

@@ -1298,7 +1298,7 @@ future<lw_shared_ptr<keyspace_metadata>> create_keyspace_from_schema_partition(d
     // Scylla-specific row will only be present if SCYLLA_KEYSPACES schema feature is available in the cluster
     if (scylla_specific_rs) {
         if (!scylla_specific_rs->empty()) {
-            auto row = scylla_specific_rs->row(0);
+            const auto& row = scylla_specific_rs->row(0);
             auto storage_type = row.get<sstring>("storage_type");
             auto options = row.get<map_type_impl::native_type>("storage_options");
             if (storage_type && options) {
@@ -2188,7 +2188,7 @@ schema_ptr create_table_from_mutations(const schema_ctxt& ctxt, schema_mutations
     slogger.trace("create_table_from_mutations: version={}, {}", version, sm);
 
     auto table_rs = query::result_set(sm.columnfamilies_mutation());
-    query::result_set_row table_row = table_rs.row(0);
+    const query::result_set_row& table_row = table_rs.row(0);
 
     auto ks_name = table_row.get_nonnull<sstring>("keyspace_name");
     auto cf_name = table_row.get_nonnull<sstring>("table_name");
@@ -2459,7 +2459,7 @@ static index_metadata create_index_from_index_row(const query::result_set_row& r
 
 view_ptr create_view_from_mutations(const schema_ctxt& ctxt, schema_mutations sm, std::optional<table_schema_version> version)  {
     auto table_rs = query::result_set(sm.columnfamilies_mutation());
-    query::result_set_row row = table_rs.row(0);
+    const query::result_set_row& row = table_rs.row(0);
 
     auto ks_name = row.get_nonnull<sstring>("keyspace_name");
     auto cf_name = row.get_nonnull<sstring>("view_name");
@@ -2469,7 +2469,7 @@ view_ptr create_view_from_mutations(const schema_ctxt& ctxt, schema_mutations sm
     prepare_builder_from_table_row(ctxt, builder, row);
 
     if (sm.scylla_tables()) {
-        table_rs = query::result_set(*sm.scylla_tables());
+        auto table_rs = query::result_set(*sm.scylla_tables());
         if (!table_rs.empty()) {
             prepare_builder_from_scylla_tables_row(ctxt, builder, table_rs.row(0));
         }

query-result-set.hh

@@ -52,6 +52,12 @@ public:
         : _schema{schema}
         , _cells{std::move(cells)}
     { }
+    result_set_row(result_set_row&&) = default;
+    result_set_row(const result_set_row&) = delete;
+    result_set_row& operator=(const result_set_row&) = delete;
+    result_set_row copy() const {
+        return {_schema, std::unordered_map{_cells}};
+    }
     // Look up a deserialized row cell value by column name
     const data_value*
     get_data_value(const sstring& column_name) const {

test/boost/multishard_mutation_query_test.cc

@@ -484,7 +484,7 @@ class data_result_builder {
             return false;
         }
         for (const auto& row : res.rows()) {
-            _rows.push_back(row);
+            _rows.emplace_back(row.copy());
             _last_ckey = extract_ckey(row);
             auto last_pkey = extract_pkey(row);
             if (_last_pkey && last_pkey.equal(*_s, *_last_pkey)) {
@@ -799,7 +799,7 @@ SEASTAR_THREAD_TEST_CASE(test_read_reversed) {
             std::vector<query::result_set_row> expected_rows;
             for (const auto& mut : expected_results) {
                 auto rs = query::result_set(mut);
-                std::copy(rs.rows().begin(), rs.rows().end(), std::back_inserter(expected_rows));
+                std::ranges::copy(rs.rows() | std::views::transform([](const auto& row) { return row.copy(); }), std::back_inserter(expected_rows));
             }
             auto expected_data_results = query::result_set(s, std::move(expected_rows));