Merge 'Fix hang in multishard_writer' from Asias

"
This series fix hang in multishard_writer when error happens. It contains
- multishard_writer: Abort the queue attached to consumers when producer fails
- repair: Fix hang when the writer is dead

Fixes #6241
Refs: #6248
"

* asias-stream_fix_multishard_writer_hang:
  repair: Fix hang when the writer is dead
  mutation_writer_test: Add test_multishard_writer_producer_aborts
  multishard_writer: Abort the queue attached to consumers when producer fails

(cherry picked from commit 8925e00e96)

SCYLLA-VERSION-GEN

@@ -1,7 +1,7 @@
 #!/bin/sh
 
 PRODUCT=scylla
-VERSION=3.2.3
+VERSION=3.2.5
 
 if test -f version
 then

api/storage_service.cc

@@ -254,6 +254,9 @@ void set_storage_service(http_context& ctx, routes& r) {
         if (column_family.empty()) {
             resp = service::get_local_storage_service().take_snapshot(tag, keynames);
         } else {
+            if (keynames.empty()) {
+                throw httpd::bad_param_exception("The keyspace of column families must be specified");
+            }
             if (keynames.size() > 1) {
                 throw httpd::bad_param_exception("Only one keyspace allowed when specifying a column family");
             }

cql3/restrictions/statement_restrictions.cc

@@ -390,28 +390,45 @@ std::vector<const column_definition*> statement_restrictions::get_column_defs_fo
     if (need_filtering()) {
         auto& sim = db.find_column_family(_schema).get_index_manager();
         auto [opt_idx, _] = find_idx(sim);
-        auto column_uses_indexing = [&opt_idx] (const column_definition* cdef) {
-            return opt_idx && opt_idx->depends_on(*cdef);
+        auto column_uses_indexing = [&opt_idx] (const column_definition* cdef, ::shared_ptr<single_column_restriction> restr) {
+            return opt_idx && restr && restr->is_supported_by(*opt_idx);
         };
+        auto single_pk_restrs = dynamic_pointer_cast<single_column_partition_key_restrictions>(_partition_key_restrictions);
         if (_partition_key_restrictions->needs_filtering(*_schema)) {
             for (auto&& cdef : _partition_key_restrictions->get_column_defs()) {
-                if (!column_uses_indexing(cdef)) {
+                ::shared_ptr<single_column_restriction> restr;
+                if (single_pk_restrs) {
+                    auto it = single_pk_restrs->restrictions().find(cdef);
+                    if (it != single_pk_restrs->restrictions().end()) {
+                        restr = dynamic_pointer_cast<single_column_restriction>(it->second);
+                    }
+                }
+                if (!column_uses_indexing(cdef, restr)) {
                     column_defs_for_filtering.emplace_back(cdef);
                 }
             }
         }
+        auto single_ck_restrs = dynamic_pointer_cast<single_column_clustering_key_restrictions>(_clustering_columns_restrictions);
         const bool pk_has_unrestricted_components = _partition_key_restrictions->has_unrestricted_components(*_schema);
         if (pk_has_unrestricted_components || _clustering_columns_restrictions->needs_filtering(*_schema)) {
             column_id first_filtering_id = pk_has_unrestricted_components ? 0 : _schema->clustering_key_columns().begin()->id +
                     _clustering_columns_restrictions->num_prefix_columns_that_need_not_be_filtered();
             for (auto&& cdef : _clustering_columns_restrictions->get_column_defs()) {
-                if (cdef->id >= first_filtering_id && !column_uses_indexing(cdef)) {
+                ::shared_ptr<single_column_restriction> restr;
+                if (single_pk_restrs) {
+                    auto it = single_ck_restrs->restrictions().find(cdef);
+                    if (it != single_ck_restrs->restrictions().end()) {
+                        restr = dynamic_pointer_cast<single_column_restriction>(it->second);
+                    }
+                }
+                if (cdef->id >= first_filtering_id && !column_uses_indexing(cdef, restr)) {
                     column_defs_for_filtering.emplace_back(cdef);
                 }
             }
         }
         for (auto&& cdef : _nonprimary_key_restrictions->get_column_defs()) {
-            if (!column_uses_indexing(cdef)) {
+            auto restr = dynamic_pointer_cast<single_column_restriction>(_nonprimary_key_restrictions->get_restriction(*cdef));
+            if (!column_uses_indexing(cdef, restr)) {
                 column_defs_for_filtering.emplace_back(cdef);
             }
         }

db/config.cc

@@ -691,6 +691,7 @@ db::config::config(std::shared_ptr<db::extensions> exts)
     , shutdown_announce_in_ms(this, "shutdown_announce_in_ms", value_status::Used, 2 * 1000, "Time a node waits after sending gossip shutdown message in milliseconds. Same as -Dcassandra.shutdown_announce_in_ms in cassandra.")
     , developer_mode(this, "developer_mode", value_status::Used, false, "Relax environment checks. Setting to true can reduce performance and reliability significantly.")
     , skip_wait_for_gossip_to_settle(this, "skip_wait_for_gossip_to_settle", value_status::Used, -1, "An integer to configure the wait for gossip to settle. -1: wait normally, 0: do not wait at all, n: wait for at most n polls. Same as -Dcassandra.skip_wait_for_gossip_to_settle in cassandra.")
+    , force_gossip_generation(this, "force_gossip_generation", liveness::LiveUpdate, value_status::Used, -1 , "Force gossip to use the generation number provided by user")
     , experimental(this, "experimental", value_status::Used, false, "Set to true to unlock all experimental features.")
     , experimental_features(this, "experimental_features", value_status::Used, {}, "Unlock experimental features provided as the option arguments (possible values: 'lwt' and 'cdc'). Can be repeated.")
     , lsa_reclamation_step(this, "lsa_reclamation_step", value_status::Used, 1, "Minimum number of segments to reclaim in a single step")

db/config.hh

@@ -269,6 +269,7 @@ public:
     named_value<uint32_t> shutdown_announce_in_ms;
     named_value<bool> developer_mode;
     named_value<int32_t> skip_wait_for_gossip_to_settle;
+    named_value<int32_t> force_gossip_generation;
     named_value<bool> experimental;
     named_value<std::vector<enum_option<experimental_features_t>>> experimental_features;
     named_value<size_t> lsa_reclamation_step;

dist/redhat/scylla.spec.mustache

@@ -17,6 +17,10 @@ Obsoletes:	scylla-server < 1.1
 
 %undefine _find_debuginfo_dwz_opts
 
+# Prevent find-debuginfo.sh from tempering with scylla's build-id (#5881)
+%undefine _unique_build_ids
+%global _no_recompute_build_ids 1
+
 %description
 Scylla is a highly scalable, eventually consistent, distributed,
 partitioned row DB.

docs/sstable-scylla-format.md

@@ -76,6 +76,9 @@ Scylla with issue #4139 fixed)
 bit 4: CorrectEmptyCounters (if set, indicates the sstable was generated by
 Scylla with issue #4363 fixed)
 
+bit 5: CorrectUDTsInCollections (if set, indicates that the sstable was generated
+by Scylla with issue #6130 fixed)
+
 ## extension_attributes subcomponent
 
     extension_attributes = extension_attribute_count extension_attribute*

gms/gossiper.cc

@@ -1622,11 +1622,15 @@ future<> gossiper::start_gossiping(int generation_nbr, std::map<application_stat
     // message on all cpus and forard them to cpu0 to process.
     return get_gossiper().invoke_on_all([do_bind] (gossiper& g) {
         g.init_messaging_service_handler(do_bind);
-    }).then([this, generation_nbr, preload_local_states] {
+    }).then([this, generation_nbr, preload_local_states] () mutable {
         build_seeds_list();
-        /* initialize the heartbeat state for this localEndpoint */
-        maybe_initialize_local_state(generation_nbr);
+        if (_cfg.force_gossip_generation() > 0) {
+            generation_nbr = _cfg.force_gossip_generation();
+            logger.warn("Use the generation number provided by user: generation = {}", generation_nbr);
+        }
         endpoint_state& local_state = endpoint_state_map[get_broadcast_address()];
+        local_state.set_heart_beat_state_and_update_timestamp(heart_beat_state(generation_nbr));
+        local_state.mark_alive();
         for (auto& entry : preload_local_states) {
             local_state.add_application_state(entry.first, entry.second);
         }

locator/simple_strategy.cc

@@ -53,13 +53,13 @@ std::vector<inet_address> simple_strategy::calculate_natural_endpoints(const tok
     endpoints.reserve(replicas);
 
     for (auto& token : tm.ring_range(t)) {
+        if (endpoints.size() == replicas) {
+           break;
+        }
         auto ep = tm.get_endpoint(token);
         assert(ep);
 
         endpoints.push_back(*ep);
-        if (endpoints.size() == replicas) {
-           break;
-        }
     }
 
     return std::move(endpoints.get_vector());

mutation_writer/multishard_writer.cc

@@ -177,6 +177,13 @@ future<> multishard_writer::distribute_mutation_fragments() {
                 return handle_end_of_stream();
             }
         });
+    }).handle_exception([this] (std::exception_ptr ep) {
+        for (auto& q : _queue_reader_handles) {
+            if (q) {
+                q->abort(ep);
+            }
+        }
+        return make_exception_future<>(std::move(ep));
     });
 }
 

repair/row_level.cc

@@ -444,7 +444,7 @@ class repair_writer {
     uint64_t _estimated_partitions;
     size_t _nr_peer_nodes;
     // Needs more than one for repair master
-    std::vector<std::optional<future<uint64_t>>> _writer_done;
+    std::vector<std::optional<future<>>> _writer_done;
     std::vector<std::optional<seastar::queue<mutation_fragment_opt>>> _mq;
     // Current partition written to disk
     std::vector<lw_shared_ptr<const decorated_key_with_hash>> _current_dk_written_to_sstable;
@@ -524,7 +524,15 @@ public:
                 return consumer(std::move(reader));
             });
         },
-        t.stream_in_progress());
+        t.stream_in_progress()).then([this, node_idx] (uint64_t partitions) {
+            rlogger.debug("repair_writer: keyspace={}, table={}, managed to write partitions={} to sstable",
+                _schema->ks_name(), _schema->cf_name(), partitions);
+        }).handle_exception([this, node_idx] (std::exception_ptr ep) {
+            rlogger.warn("repair_writer: keyspace={}, table={}, multishard_writer failed: {}",
+                    _schema->ks_name(), _schema->cf_name(), ep);
+            _mq[node_idx]->abort(ep);
+            return make_exception_future<>(std::move(ep));
+        });
     }
 
     future<> write_partition_end(unsigned node_idx) {
@@ -551,21 +559,33 @@ public:
         }
     }
 
+    future<> write_end_of_stream(unsigned node_idx) {
+        if (_mq[node_idx]) {
+            // Partition_end is never sent on wire, so we have to write one ourselves.
+            return write_partition_end(node_idx).then([this, node_idx] () mutable {
+                // Empty mutation_fragment_opt means no more data, so the writer can seal the sstables.
+                return _mq[node_idx]->push_eventually(mutation_fragment_opt());
+            });
+        } else {
+            return make_ready_future<>();
+        }
+    }
+
+    future<> do_wait_for_writer_done(unsigned node_idx) {
+        if (_writer_done[node_idx]) {
+            return std::move(*(_writer_done[node_idx]));
+        } else {
+            return make_ready_future<>();
+        }
+    }
+
     future<> wait_for_writer_done() {
         return parallel_for_each(boost::irange(unsigned(0), unsigned(_nr_peer_nodes)), [this] (unsigned node_idx) {
-            if (_writer_done[node_idx] && _mq[node_idx]) {
-                // Partition_end is never sent on wire, so we have to write one ourselves.
-                return write_partition_end(node_idx).then([this, node_idx] () mutable {
-                    // Empty mutation_fragment_opt means no more data, so the writer can seal the sstables.
-                    return _mq[node_idx]->push_eventually(mutation_fragment_opt()).then([this, node_idx] () mutable {
-                        return (*_writer_done[node_idx]).then([] (uint64_t partitions) {
-                            rlogger.debug("Managed to write partitions={} to sstable", partitions);
-                            return make_ready_future<>();
-                        });
-                    });
-                });
-            }
-            return make_ready_future<>();
+            return when_all_succeed(write_end_of_stream(node_idx), do_wait_for_writer_done(node_idx));
+        }).handle_exception([this] (std::exception_ptr ep) {
+            rlogger.warn("repair_writer: keyspace={}, table={}, wait_for_writer_done failed: {}",
+                    _schema->ks_name(), _schema->cf_name(), ep);
+            return make_exception_future<>(std::move(ep));
         });
     }
 };

schema.cc

@@ -288,10 +288,10 @@ schema::schema(const raw_schema& raw, std::optional<raw_view_info> raw_view_info
                     + column_offset(column_kind::regular_column),
             _raw._columns.end(), column_definition::name_comparator(regular_column_name_type()));
 
-    std::sort(_raw._columns.begin(),
+    std::stable_sort(_raw._columns.begin(),
               _raw._columns.begin() + column_offset(column_kind::clustering_key),
               [] (auto x, auto y) { return x.id < y.id; });
-    std::sort(_raw._columns.begin() + column_offset(column_kind::clustering_key),
+    std::stable_sort(_raw._columns.begin() + column_offset(column_kind::clustering_key),
               _raw._columns.begin() + column_offset(column_kind::static_column),
               [] (auto x, auto y) { return x.id < y.id; });
 

service/query_state.hh

@@ -38,7 +38,12 @@ private:
 public:
     query_state(client_state& client_state, service_permit permit)
         : _client_state(client_state)
-        , _trace_state_ptr(_client_state.get_trace_state())
+        , _trace_state_ptr(tracing::trace_state_ptr())
+        , _permit(std::move(permit))
+    { }
+    query_state(client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit)
+        : _client_state(client_state)
+        , _trace_state_ptr(std::move(trace_state))
         , _permit(std::move(permit))
     { }
 

sstables/column_translation.hh

@@ -72,47 +72,8 @@ private:
         static std::vector<column_info> build(
                 const schema& s,
                 const utils::chunked_vector<serialization_header::column_desc>& src,
-                bool is_static) {
-            std::vector<column_info> cols;
-            if (s.is_dense()) {
-                const column_definition& col = is_static ? *s.static_begin() : *s.regular_begin();
-                cols.push_back(column_info{
-                    &col.name(),
-                    col.type,
-                    col.id,
-                    col.type->value_length_if_fixed(),
-                    col.is_multi_cell(),
-                    col.is_counter(),
-                    false
-                });
-            } else {
-                cols.reserve(src.size());
-                for (auto&& desc : src) {
-                    const bytes& type_name = desc.type_name.value;
-                    data_type type = db::marshal::type_parser::parse(to_sstring_view(type_name));
-                    const column_definition* def = s.get_column_definition(desc.name.value);
-                    std::optional<column_id> id;
-                    bool schema_mismatch = false;
-                    if (def) {
-                        id = def->id;
-                        schema_mismatch = def->is_multi_cell() != type->is_multi_cell() ||
-                                          def->is_counter() != type->is_counter() ||
-                                          !def->type->is_value_compatible_with(*type);
-                    }
-                    cols.push_back(column_info{
-                        &desc.name.value,
-                        type,
-                        id,
-                        type->value_length_if_fixed(),
-                        type->is_multi_cell(),
-                        type->is_counter(),
-                        schema_mismatch
-                    });
-                }
-                boost::range::stable_partition(cols, [](const column_info& column) { return !column.is_collection; });
-            }
-            return cols;
-        }
+                const sstable_enabled_features& features,
+                bool is_static);
 
         utils::UUID schema_uuid;
         std::vector<column_info> regular_schema_columns_from_sstable;
@@ -125,10 +86,10 @@ private:
         state(state&&) = default;
         state& operator=(state&&) = default;
 
-        state(const schema& s, const serialization_header& header)
+        state(const schema& s, const serialization_header& header, const sstable_enabled_features& features)
             : schema_uuid(s.version())
-            , regular_schema_columns_from_sstable(build(s, header.regular_columns.elements, false))
-            , static_schema_columns_from_sstable(build(s, header.static_columns.elements, true))
+            , regular_schema_columns_from_sstable(build(s, header.regular_columns.elements, features, false))
+            , static_schema_columns_from_sstable(build(s, header.static_columns.elements, features, true))
             , clustering_column_value_fix_lengths (get_clustering_values_fixed_lengths(header))
         {}
     };
@@ -136,9 +97,10 @@ private:
     lw_shared_ptr<const state> _state = make_lw_shared<const state>();
 
 public:
-    column_translation get_for_schema(const schema& s, const serialization_header& header) {
+    column_translation get_for_schema(
+            const schema& s, const serialization_header& header, const sstable_enabled_features& features) {
         if (s.version() != _state->schema_uuid) {
-            _state = make_lw_shared(state(s, header));
+            _state = make_lw_shared(state(s, header, features));
         }
         return *this;
     }

sstables/mp_row_consumer.cc

@@ -38,6 +38,8 @@
  */
 
 #include "mp_row_consumer.hh"
+#include "column_translation.hh"
+#include "concrete_types.hh"
 
 namespace sstables {
 
@@ -79,4 +81,86 @@ atomic_cell make_counter_cell(api::timestamp_type timestamp, bytes_view value) {
     return ccb.build(timestamp);
 }
 
+// See #6130.
+static data_type freeze_types_in_collections(data_type t) {
+    return ::visit(*t, make_visitor(
+    [] (const map_type_impl& typ) -> data_type {
+        return map_type_impl::get_instance(
+                freeze_types_in_collections(typ.get_keys_type()->freeze()),
+                freeze_types_in_collections(typ.get_values_type()->freeze()),
+                typ.is_multi_cell());
+    },
+    [] (const set_type_impl& typ) -> data_type {
+        return set_type_impl::get_instance(
+                freeze_types_in_collections(typ.get_elements_type()->freeze()),
+                typ.is_multi_cell());
+    },
+    [] (const list_type_impl& typ) -> data_type {
+        return list_type_impl::get_instance(
+                freeze_types_in_collections(typ.get_elements_type()->freeze()),
+                typ.is_multi_cell());
+    },
+    [&] (const abstract_type& typ) -> data_type {
+        return std::move(t);
+    }
+    ));
+}
+
+/* If this function returns false, the caller cannot assume that the SSTable comes from Scylla.
+ * It might, if for some reason a table was created using Scylla that didn't contain any feature bit,
+ * but that should never happen. */
+static bool is_certainly_scylla_sstable(const sstable_enabled_features& features) {
+    return features.enabled_features;
+}
+
+std::vector<column_translation::column_info> column_translation::state::build(
+        const schema& s,
+        const utils::chunked_vector<serialization_header::column_desc>& src,
+        const sstable_enabled_features& features,
+        bool is_static) {
+    std::vector<column_info> cols;
+    if (s.is_dense()) {
+        const column_definition& col = is_static ? *s.static_begin() : *s.regular_begin();
+        cols.push_back(column_info{
+            &col.name(),
+            col.type,
+            col.id,
+            col.type->value_length_if_fixed(),
+            col.is_multi_cell(),
+            col.is_counter(),
+            false
+        });
+    } else {
+        cols.reserve(src.size());
+        for (auto&& desc : src) {
+            const bytes& type_name = desc.type_name.value;
+            data_type type = db::marshal::type_parser::parse(to_sstring_view(type_name));
+            if (!features.is_enabled(CorrectUDTsInCollections) && is_certainly_scylla_sstable(features)) {
+                // See #6130.
+                type = freeze_types_in_collections(std::move(type));
+            }
+            const column_definition* def = s.get_column_definition(desc.name.value);
+            std::optional<column_id> id;
+            bool schema_mismatch = false;
+            if (def) {
+                id = def->id;
+                schema_mismatch = def->is_multi_cell() != type->is_multi_cell() ||
+                                  def->is_counter() != type->is_counter() ||
+                                  !def->type->is_value_compatible_with(*type);
+            }
+            cols.push_back(column_info{
+                &desc.name.value,
+                type,
+                id,
+                type->value_length_if_fixed(),
+                type->is_multi_cell(),
+                type->is_counter(),
+                schema_mismatch
+            });
+        }
+        boost::range::stable_partition(cols, [](const column_info& column) { return !column.is_collection; });
+    }
+    return cols;
+}
+
 }

sstables/row.hh

@@ -1344,7 +1344,7 @@ public:
         , _consumer(consumer)
         , _sst(sst)
         , _header(sst->get_serialization_header())
-        , _column_translation(sst->get_column_translation(s, _header))
+        , _column_translation(sst->get_column_translation(s, _header, sst->features()))
         , _has_shadowable_tombstones(sst->has_shadowable_tombstones())
     {
         setup_columns(_regular_row, _column_translation.regular_columns());

sstables/sstables.hh

@@ -780,8 +780,9 @@ public:
     const serialization_header& get_serialization_header() const {
         return get_mutable_serialization_header(*_components);
     }
-    column_translation get_column_translation(const schema& s, const serialization_header& h) {
-        return _column_translation.get_for_schema(s, h);
+    column_translation get_column_translation(
+            const schema& s, const serialization_header& h, const sstable_enabled_features& f) {
+        return _column_translation.get_for_schema(s, h, f);
     }
     const std::vector<unsigned>& get_shards_for_this_sstable() const {
         return _shards;

sstables/types.hh

@@ -459,7 +459,8 @@ enum sstable_feature : uint8_t {
     ShadowableTombstones = 2, // See #3885
     CorrectStaticCompact = 3, // See #4139
     CorrectEmptyCounters = 4, // See #4363
-    End = 5,
+    CorrectUDTsInCollections = 5, // See #6130
+    End = 6,
 };
 
 // Scylla-specific features enabled for a particular sstable.

tests/mutation_writer_test.cc

@@ -118,6 +118,53 @@ SEASTAR_TEST_CASE(test_multishard_writer) {
     });
 }
 
+SEASTAR_TEST_CASE(test_multishard_writer_producer_aborts) {
+    return do_with_cql_env_thread([] (cql_test_env& e) {
+        auto test_random_streams = [] (random_mutation_generator&& gen, size_t partition_nr, generate_error error = generate_error::no) {
+            auto muts = gen(partition_nr);
+            schema_ptr s = gen.schema();
+            auto source_reader = partition_nr > 0 ? flat_mutation_reader_from_mutations(muts) : make_empty_flat_reader(s);
+            int mf_produced = 0;
+            auto get_next_mutation_fragment = [&source_reader, &mf_produced] () mutable {
+                if (mf_produced++ > 800) {
+                    return make_exception_future<mutation_fragment_opt>(std::runtime_error("the producer failed"));
+                } else {
+                    return source_reader(db::no_timeout);
+                }
+            };
+            auto& partitioner = dht::global_partitioner();
+            try {
+                distribute_reader_and_consume_on_shards(s, partitioner,
+                    make_generating_reader(s, std::move(get_next_mutation_fragment)),
+                    [&partitioner, error] (flat_mutation_reader reader) mutable {
+                        if (error) {
+                            return make_exception_future<>(std::runtime_error("Failed to write"));
+                        }
+                        return repeat([&partitioner, reader = std::move(reader), error] () mutable {
+                            return reader(db::no_timeout).then([&partitioner,  error] (mutation_fragment_opt mf_opt) mutable {
+                                if (mf_opt) {
+                                    if (mf_opt->is_partition_start()) {
+                                        auto shard = partitioner.shard_of(mf_opt->as_partition_start().key().token());
+                                        BOOST_REQUIRE_EQUAL(shard, this_shard_id());
+                                    }
+                                    return make_ready_future<stop_iteration>(stop_iteration::no);
+                                } else {
+                                    return make_ready_future<stop_iteration>(stop_iteration::yes);
+                                }
+                            });
+                        });
+                    }
+                ).get0();
+            } catch (...) {
+                // The distribute_reader_and_consume_on_shards is expected to fail and not block forever
+            }
+        };
+
+        test_random_streams(random_mutation_generator(random_mutation_generator::generate_counters::no, local_shard_only::yes), 1000, generate_error::no);
+        test_random_streams(random_mutation_generator(random_mutation_generator::generate_counters::no, local_shard_only::yes), 1000, generate_error::yes);
+    });
+}
+
 namespace {
 
 class bucket_writer {

tests/sstable_3_x_test.cc

@@ -5262,3 +5262,131 @@ SEASTAR_THREAD_TEST_CASE(test_sstable_log_too_many_rows) {
     test_sstable_log_too_many_rows_f(random, (random + 1), false);
     test_sstable_log_too_many_rows_f((random + 1), random, true);
 }
+
+// The following test runs on tests/sstables/3.x/uncompressed/legacy_udt_in_collection
+// It was created using Scylla 3.0.x using the following CQL statements:
+//
+// CREATE KEYSPACE ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};
+// CREATE TYPE ks.ut (a int, b int);
+// CREATE TABLE ks.t ( pk int PRIMARY KEY,
+//                     m map<int, frozen<ut>>,
+//                     fm frozen<map<int, frozen<ut>>>,
+//                     mm map<int, frozen<map<int, frozen<ut>>>>,
+//                     fmm frozen<map<int, frozen<map<int, frozen<ut>>>>>,
+//                     s set<frozen<ut>>,
+//                     fs frozen<set<frozen<ut>>>,
+//                     l list<frozen<ut>>,
+//                     fl frozen<list<frozen<ut>>>
+//                   ) WITH compression = {};
+// UPDATE ks.t USING TIMESTAMP 1525385507816568 SET
+//         m[0] = {a: 0, b: 0},
+//         fm = {0: {a: 0, b: 0}},
+//         mm[0] = {0: {a: 0, b: 0}},
+//         fmm = {0: {0: {a: 0, b: 0}}},
+//         s = s + {{a: 0, b: 0}},
+//         fs = {{a: 0, b: 0}},
+//         l[scylla_timeuuid_list_index(7fb27e80-7b12-11ea-9fad-f4d108a9e4a3)] = {a: 0, b: 0},
+//         fl = [{a: 0, b: 0}]
+//     WHERE pk = 0;
+//
+// It checks whether a SSTable containing UDTs nested in collections, which contains incorrect serialization headers
+// (doesn't wrap nested UDTs in the FrozenType<...> tag) can be loaded by new versions of Scylla.
+
+static const sstring LEGACY_UDT_IN_COLLECTION_PATH =
+    "tests/sstables/3.x/uncompressed/legacy_udt_in_collection";
+
+SEASTAR_THREAD_TEST_CASE(test_legacy_udt_in_collection_table) {
+    auto abj = defer([] { await_background_jobs().get(); });
+
+    auto ut = user_type_impl::get_instance("ks", to_bytes("ut"),
+            {to_bytes("a"), to_bytes("b")},
+            {int32_type, int32_type}, false);
+    auto m_type = map_type_impl::get_instance(int32_type, ut, true);
+    auto fm_type = map_type_impl::get_instance(int32_type, ut, false);
+    auto mm_type = map_type_impl::get_instance(int32_type, fm_type, true);
+    auto fmm_type = map_type_impl::get_instance(int32_type, fm_type, false);
+    auto s_type = set_type_impl::get_instance(ut, true);
+    auto fs_type = set_type_impl::get_instance(ut, false);
+    auto l_type = list_type_impl::get_instance(ut, true);
+    auto fl_type = list_type_impl::get_instance(ut, false);
+
+    auto s = schema_builder("ks", "t")
+        .with_column("pk", int32_type, column_kind::partition_key)
+        .with_column("m", m_type)
+        .with_column("fm", fm_type)
+        .with_column("mm", mm_type)
+        .with_column("fmm", fmm_type)
+        .with_column("s", s_type)
+        .with_column("fs", fs_type)
+        .with_column("l", l_type)
+        .with_column("fl", fl_type)
+        .set_compressor_params(compression_parameters::no_compression())
+        .build();
+
+    auto m_cdef = s->get_column_definition(to_bytes("m"));
+    auto fm_cdef = s->get_column_definition(to_bytes("fm"));
+    auto mm_cdef = s->get_column_definition(to_bytes("mm"));
+    auto fmm_cdef = s->get_column_definition(to_bytes("fmm"));
+    auto s_cdef = s->get_column_definition(to_bytes("s"));
+    auto fs_cdef = s->get_column_definition(to_bytes("fs"));
+    auto l_cdef = s->get_column_definition(to_bytes("l"));
+    auto fl_cdef = s->get_column_definition(to_bytes("fl"));
+    BOOST_REQUIRE(m_cdef && fm_cdef && mm_cdef && fmm_cdef && s_cdef && fs_cdef && l_cdef && fl_cdef);
+
+    auto ut_val = make_user_value(ut, {int32_t(0), int32_t(0)});
+    auto fm_val = make_map_value(fm_type, {{int32_t(0), ut_val}});
+    auto fmm_val = make_map_value(fmm_type, {{int32_t(0), fm_val}});
+    auto fs_val = make_set_value(fs_type, {ut_val});
+    auto fl_val = make_list_value(fl_type, {ut_val});
+
+    mutation mut{s, partition_key::from_deeply_exploded(*s, {0})};
+    auto ckey = clustering_key::make_empty();
+
+    // m[0] = {a: 0, b: 0}
+    {
+        collection_mutation_description desc;
+        desc.cells.emplace_back(int32_type->decompose(0),
+            atomic_cell::make_live(*ut, write_timestamp, ut->decompose(ut_val), atomic_cell::collection_member::yes));
+        mut.set_clustered_cell(ckey, *m_cdef, desc.serialize(*m_type));
+    }
+
+    // fm = {0: {a: 0, b: 0}}
+    mut.set_clustered_cell(ckey, *fm_cdef, atomic_cell::make_live(*fm_type, write_timestamp, fm_type->decompose(fm_val)));
+
+    // mm[0] = {0: {a: 0, b: 0}},
+    {
+        collection_mutation_description desc;
+        desc.cells.emplace_back(int32_type->decompose(0),
+            atomic_cell::make_live(*fm_type, write_timestamp, fm_type->decompose(fm_val), atomic_cell::collection_member::yes));
+        mut.set_clustered_cell(ckey, *mm_cdef, desc.serialize(*mm_type));
+    }
+
+    // fmm = {0: {0: {a: 0, b: 0}}},
+    mut.set_clustered_cell(ckey, *fmm_cdef, atomic_cell::make_live(*fmm_type, write_timestamp, fmm_type->decompose(fmm_val)));
+
+    // s = s + {{a: 0, b: 0}},
+    {
+        collection_mutation_description desc;
+        desc.cells.emplace_back(ut->decompose(ut_val),
+            atomic_cell::make_live(*bytes_type, write_timestamp, bytes{}, atomic_cell::collection_member::yes));
+        mut.set_clustered_cell(ckey, *s_cdef, desc.serialize(*s_type));
+    }
+
+    // fs = {{a: 0, b: 0}},
+    mut.set_clustered_cell(ckey, *fs_cdef, atomic_cell::make_live(*fs_type, write_timestamp, fs_type->decompose(fs_val)));
+
+    // l[scylla_timeuuid_list_index(7fb27e80-7b12-11ea-9fad-f4d108a9e4a3)] = {a: 0, b: 0},
+    {
+        collection_mutation_description desc;
+        desc.cells.emplace_back(timeuuid_type->decompose(utils::UUID("7fb27e80-7b12-11ea-9fad-f4d108a9e4a3")),
+            atomic_cell::make_live(*ut, write_timestamp, ut->decompose(ut_val), atomic_cell::collection_member::yes));
+        mut.set_clustered_cell(ckey, *l_cdef, desc.serialize(*l_type));
+    }
+
+    // fl = [{a: 0, b: 0}]
+    mut.set_clustered_cell(ckey, *fl_cdef, atomic_cell::make_live(*fl_type, write_timestamp, fl_type->decompose(fl_val)));
+
+    sstable_assertions sst(s, LEGACY_UDT_IN_COLLECTION_PATH);
+    sst.load();
+    assert_that(sst.read_rows_flat()).produces(mut).produces_end_of_stream();
+}

tests/sstables/3.x/uncompressed/legacy_udt_in_collection/mc-1-big-Digest.crc32

@@ -0,0 +1 @@
+3519784297

tests/sstables/3.x/uncompressed/legacy_udt_in_collection/mc-1-big-TOC.txt

@@ -0,0 +1,9 @@
+Scylla.db
+CRC.db
+Filter.db
+Statistics.db
+TOC.txt
+Digest.crc32
+Index.db
+Summary.db
+Data.db

transport/server.cc

@@ -347,6 +347,7 @@ future<std::unique_ptr<cql_server::response>>
 
     trace_props.set_if<tracing::trace_state_props::log_slow_query>(tracing::tracing::get_local_tracing_instance().slow_query_tracing_enabled());
     trace_props.set_if<tracing::trace_state_props::full_tracing>(tracing_request != tracing_request_type::not_requested);
+    tracing::trace_state_ptr trace_state;
 
     if (trace_props) {
         if (cqlop == cql_binary_opcode::QUERY ||
@@ -354,15 +355,15 @@ future<std::unique_ptr<cql_server::response>>
             cqlop == cql_binary_opcode::EXECUTE ||
             cqlop == cql_binary_opcode::BATCH) {
             trace_props.set_if<tracing::trace_state_props::write_on_close>(tracing_request == tracing_request_type::write_on_close);
-            client_state.create_tracing_session(tracing::trace_type::QUERY, trace_props);
+            trace_state = tracing::tracing::get_local_tracing_instance().create_session(tracing::trace_type::QUERY, trace_props);
         }
     }
 
-    tracing::set_request_size(client_state.get_trace_state(), fbuf.bytes_left());
+    tracing::set_request_size(trace_state, fbuf.bytes_left());
 
     auto linearization_buffer = std::make_unique<bytes_ostream>();
     auto linearization_buffer_ptr = linearization_buffer.get();
-    return futurize_apply([this, cqlop, stream, &fbuf, &client_state, linearization_buffer_ptr, permit = std::move(permit)] () mutable {
+    return futurize_apply([this, cqlop, stream, &fbuf, &client_state, linearization_buffer_ptr, permit = std::move(permit), trace_state] () mutable {
         // When using authentication, we need to ensure we are doing proper state transitions,
         // i.e. we cannot simply accept any query/exec ops unless auth is complete
         switch (client_state.get_auth_state()) {
@@ -393,23 +394,23 @@ future<std::unique_ptr<cql_server::response>>
             return *user;
         }();
 
-        tracing::set_username(client_state.get_trace_state(), user);
+        tracing::set_username(trace_state, user);
 
         auto in = request_reader(std::move(fbuf), *linearization_buffer_ptr);
         switch (cqlop) {
-        case cql_binary_opcode::STARTUP:       return process_startup(stream, std::move(in), client_state);
-        case cql_binary_opcode::AUTH_RESPONSE: return process_auth_response(stream, std::move(in), client_state);
-        case cql_binary_opcode::OPTIONS:       return process_options(stream, std::move(in), client_state);
-        case cql_binary_opcode::QUERY:         return process_query(stream, std::move(in), client_state, std::move(permit));
-        case cql_binary_opcode::PREPARE:       return process_prepare(stream, std::move(in), client_state);
-        case cql_binary_opcode::EXECUTE:       return process_execute(stream, std::move(in), client_state, std::move(permit));
-        case cql_binary_opcode::BATCH:         return process_batch(stream, std::move(in), client_state, std::move(permit));
-        case cql_binary_opcode::REGISTER:      return process_register(stream, std::move(in), client_state);
+        case cql_binary_opcode::STARTUP:       return process_startup(stream, std::move(in), client_state, trace_state);
+        case cql_binary_opcode::AUTH_RESPONSE: return process_auth_response(stream, std::move(in), client_state, trace_state);
+        case cql_binary_opcode::OPTIONS:       return process_options(stream, std::move(in), client_state, trace_state);
+        case cql_binary_opcode::QUERY:         return process_query(stream, std::move(in), client_state, std::move(permit), trace_state);
+        case cql_binary_opcode::PREPARE:       return process_prepare(stream, std::move(in), client_state, trace_state);
+        case cql_binary_opcode::EXECUTE:       return process_execute(stream, std::move(in), client_state, std::move(permit), trace_state);
+        case cql_binary_opcode::BATCH:         return process_batch(stream, std::move(in), client_state, std::move(permit), trace_state);
+        case cql_binary_opcode::REGISTER:      return process_register(stream, std::move(in), client_state, trace_state);
         default:                               throw exceptions::protocol_exception(format("Unknown opcode {:d}", int(cqlop)));
         }
-    }).then_wrapped([this, cqlop, stream, &client_state, linearization_buffer = std::move(linearization_buffer)] (future<std::unique_ptr<cql_server::response>> f) -> std::unique_ptr<cql_server::response> {
+    }).then_wrapped([this, cqlop, stream, &client_state, linearization_buffer = std::move(linearization_buffer), trace_state] (future<std::unique_ptr<cql_server::response>> f) -> std::unique_ptr<cql_server::response> {
         auto stop_trace = defer([&] {
-            tracing::stop_foreground(client_state.get_trace_state());
+            tracing::stop_foreground(trace_state);
         });
         --_server._requests_serving;
         try {
@@ -440,28 +441,28 @@ future<std::unique_ptr<cql_server::response>>
                 break;
             }
 
-            tracing::set_response_size(client_state.get_trace_state(), response->size());
+            tracing::set_response_size(trace_state, response->size());
             return response;
         } catch (const exceptions::unavailable_exception& ex) {
-            return make_unavailable_error(stream, ex.code(), ex.what(), ex.consistency, ex.required, ex.alive, client_state.get_trace_state());
+            return make_unavailable_error(stream, ex.code(), ex.what(), ex.consistency, ex.required, ex.alive, trace_state);
         } catch (const exceptions::read_timeout_exception& ex) {
-            return make_read_timeout_error(stream, ex.code(), ex.what(), ex.consistency, ex.received, ex.block_for, ex.data_present, client_state.get_trace_state());
+            return make_read_timeout_error(stream, ex.code(), ex.what(), ex.consistency, ex.received, ex.block_for, ex.data_present, trace_state);
         } catch (const exceptions::read_failure_exception& ex) {
-            return make_read_failure_error(stream, ex.code(), ex.what(), ex.consistency, ex.received, ex.failures, ex.block_for, ex.data_present, client_state.get_trace_state());
+            return make_read_failure_error(stream, ex.code(), ex.what(), ex.consistency, ex.received, ex.failures, ex.block_for, ex.data_present, trace_state);
         } catch (const exceptions::mutation_write_timeout_exception& ex) {
-            return make_mutation_write_timeout_error(stream, ex.code(), ex.what(), ex.consistency, ex.received, ex.block_for, ex.type, client_state.get_trace_state());
+            return make_mutation_write_timeout_error(stream, ex.code(), ex.what(), ex.consistency, ex.received, ex.block_for, ex.type, trace_state);
         } catch (const exceptions::mutation_write_failure_exception& ex) {
-            return make_mutation_write_failure_error(stream, ex.code(), ex.what(), ex.consistency, ex.received, ex.failures, ex.block_for, ex.type, client_state.get_trace_state());
+            return make_mutation_write_failure_error(stream, ex.code(), ex.what(), ex.consistency, ex.received, ex.failures, ex.block_for, ex.type, trace_state);
         } catch (const exceptions::already_exists_exception& ex) {
-            return make_already_exists_error(stream, ex.code(), ex.what(), ex.ks_name, ex.cf_name, client_state.get_trace_state());
+            return make_already_exists_error(stream, ex.code(), ex.what(), ex.ks_name, ex.cf_name, trace_state);
         } catch (const exceptions::prepared_query_not_found_exception& ex) {
-            return make_unprepared_error(stream, ex.code(), ex.what(), ex.id, client_state.get_trace_state());
+            return make_unprepared_error(stream, ex.code(), ex.what(), ex.id, trace_state);
         } catch (const exceptions::cassandra_exception& ex) {
-            return make_error(stream, ex.code(), ex.what(), client_state.get_trace_state());
+            return make_error(stream, ex.code(), ex.what(), trace_state);
         } catch (std::exception& ex) {
-            return make_error(stream, exceptions::exception_code::SERVER_ERROR, ex.what(), client_state.get_trace_state());
+            return make_error(stream, exceptions::exception_code::SERVER_ERROR, ex.what(), trace_state);
         } catch (...) {
-            return make_error(stream, exceptions::exception_code::SERVER_ERROR, "unknown error", client_state.get_trace_state());
+            return make_error(stream, exceptions::exception_code::SERVER_ERROR, "unknown error", trace_state);
         }
     });
 }
@@ -661,8 +662,8 @@ future<fragmented_temporary_buffer> cql_server::connection::read_and_decompress_
     return _buffer_reader.read_exactly(_read_buf, length);
 }
 
-future<std::unique_ptr<cql_server::response>> cql_server::connection::process_startup(uint16_t stream, request_reader in, service::client_state& client_state)
-{
+future<std::unique_ptr<cql_server::response>> cql_server::connection::process_startup(uint16_t stream, request_reader in, service::client_state& client_state,
+        tracing::trace_state_ptr trace_state) {
     auto options = in.read_string_map();
     auto compression_opt = options.find("COMPRESSION");
     if (compression_opt != options.end()) {
@@ -678,33 +679,31 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_st
     }
     auto& a = client_state.get_auth_service()->underlying_authenticator();
     if (a.require_authentication()) {
-        return make_ready_future<std::unique_ptr<cql_server::response>>(make_autheticate(stream, a.qualified_java_name(), client_state.get_trace_state()));
+        return make_ready_future<std::unique_ptr<cql_server::response>>(make_autheticate(stream, a.qualified_java_name(), trace_state));
     }
-    return make_ready_future<std::unique_ptr<cql_server::response>>(make_ready(stream, client_state.get_trace_state()));
+    return make_ready_future<std::unique_ptr<cql_server::response>>(make_ready(stream, trace_state));
 }
 
-future<std::unique_ptr<cql_server::response>> cql_server::connection::process_auth_response(uint16_t stream, request_reader in, service::client_state& client_state)
-{
+future<std::unique_ptr<cql_server::response>> cql_server::connection::process_auth_response(uint16_t stream, request_reader in, service::client_state& client_state,
+        tracing::trace_state_ptr trace_state) {
     auto sasl_challenge = client_state.get_auth_service()->underlying_authenticator().new_sasl_challenge();
     auto buf = in.read_raw_bytes_view(in.bytes_left());
     auto challenge = sasl_challenge->evaluate_response(buf);
     if (sasl_challenge->is_complete()) {
-        return sasl_challenge->get_authenticated_user().then([this, sasl_challenge, stream, &client_state, challenge = std::move(challenge)](auth::authenticated_user user) mutable {
+        return sasl_challenge->get_authenticated_user().then([this, sasl_challenge, stream, &client_state, challenge = std::move(challenge), trace_state](auth::authenticated_user user) mutable {
             client_state.set_login(::make_shared<auth::authenticated_user>(std::move(user)));
             auto f = client_state.check_user_can_login();
-            return f.then([this, stream, &client_state, challenge = std::move(challenge)]() mutable {
-                auto tr_state = client_state.get_trace_state();
-                return make_ready_future<std::unique_ptr<cql_server::response>>(make_auth_success(stream, std::move(challenge), tr_state));
+            return f.then([this, stream, &client_state, challenge = std::move(challenge), trace_state]() mutable {
+                return make_ready_future<std::unique_ptr<cql_server::response>>(make_auth_success(stream, std::move(challenge), trace_state));
             });
         });
     }
-    auto tr_state = client_state.get_trace_state();
-    return make_ready_future<std::unique_ptr<cql_server::response>>(make_auth_challenge(stream, std::move(challenge), tr_state));
+    return make_ready_future<std::unique_ptr<cql_server::response>>(make_auth_challenge(stream, std::move(challenge), trace_state));
 }
 
-future<std::unique_ptr<cql_server::response>> cql_server::connection::process_options(uint16_t stream, request_reader in, service::client_state& client_state)
-{
-    return make_ready_future<std::unique_ptr<cql_server::response>>(make_supported(stream, client_state.get_trace_state()));
+future<std::unique_ptr<cql_server::response>> cql_server::connection::process_options(uint16_t stream, request_reader in, service::client_state& client_state,
+        tracing::trace_state_ptr trace_state) {
+    return make_ready_future<std::unique_ptr<cql_server::response>>(make_supported(stream, std::move(trace_state)));
 }
 
 void
@@ -712,10 +711,10 @@ cql_server::connection::init_cql_serialization_format() {
     _cql_serialization_format = cql_serialization_format(_version);
 }
 
-future<std::unique_ptr<cql_server::response>> cql_server::connection::process_query(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit)
+future<std::unique_ptr<cql_server::response>> cql_server::connection::process_query(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit, tracing::trace_state_ptr trace_state)
 {
     auto query = in.read_long_string_view();
-    auto q_state = std::make_unique<cql_query_state>(client_state, std::move(permit));
+    auto q_state = std::make_unique<cql_query_state>(client_state, trace_state, std::move(permit));
     auto& query_state = q_state->query_state;
     q_state->options = in.read_options(_version, _cql_serialization_format, this->timeout_config(), _server._cql_config);
     auto& options = *q_state->options;
@@ -735,12 +734,12 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_qu
     }).finally([q_state = std::move(q_state)] {});
 }
 
-future<std::unique_ptr<cql_server::response>> cql_server::connection::process_prepare(uint16_t stream, request_reader in, service::client_state& client_state)
-{
+future<std::unique_ptr<cql_server::response>> cql_server::connection::process_prepare(uint16_t stream, request_reader in, service::client_state& client_state,
+        tracing::trace_state_ptr trace_state) {
     auto query = sstring(in.read_long_string_view());
 
-    tracing::add_query(client_state.get_trace_state(), query);
-    tracing::begin(client_state.get_trace_state(), "Preparing CQL3 query", client_state.get_client_address());
+    tracing::add_query(trace_state, query);
+    tracing::begin(trace_state, "Preparing CQL3 query", client_state.get_client_address());
 
     auto cpu_id = engine().cpu_id();
     auto cpus = boost::irange(0u, smp::count);
@@ -752,19 +751,19 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_pr
         } else {
             return make_ready_future<>();
         }
-    }).then([this, query, stream, &client_state] () mutable {
-        tracing::trace(client_state.get_trace_state(), "Done preparing on remote shards");
-        return _server._query_processor.local().prepare(std::move(query), client_state, false).then([this, stream, &client_state] (auto msg) {
-            tracing::trace(client_state.get_trace_state(), "Done preparing on a local shard - preparing a result. ID is [{}]", seastar::value_of([&msg] {
+    }).then([this, query, stream, &client_state, trace_state] () mutable {
+        tracing::trace(trace_state, "Done preparing on remote shards");
+        return _server._query_processor.local().prepare(std::move(query), client_state, false).then([this, stream, &client_state, trace_state] (auto msg) {
+            tracing::trace(trace_state, "Done preparing on a local shard - preparing a result. ID is [{}]", seastar::value_of([&msg] {
                 return messages::result_message::prepared::cql::get_id(msg);
             }));
-            return this->make_result(stream, msg, client_state.get_trace_state());
+            return this->make_result(stream, msg, trace_state);
         });
     });
 }
 
-future<std::unique_ptr<cql_server::response>> cql_server::connection::process_execute(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit)
-{
+future<std::unique_ptr<cql_server::response>> cql_server::connection::process_execute(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit,
+        tracing::trace_state_ptr trace_state) {
     cql3::prepared_cache_key_type cache_key(in.read_short_bytes());
     auto& id = cql3::prepared_cache_key_type::cql_id(cache_key);
     bool needs_authorization = false;
@@ -781,7 +780,7 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_ex
         throw exceptions::prepared_query_not_found_exception(id);
     }
 
-    auto q_state = std::make_unique<cql_query_state>(client_state, std::move(permit));
+    auto q_state = std::make_unique<cql_query_state>(client_state, trace_state, std::move(permit));
     auto& query_state = q_state->query_state;
     if (_version == 1) {
         std::vector<cql3::raw_value_view> values;
@@ -795,22 +794,22 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_ex
     auto& options = *q_state->options;
     auto skip_metadata = options.skip_metadata();
 
-    tracing::set_page_size(client_state.get_trace_state(), options.get_page_size());
-    tracing::set_consistency_level(client_state.get_trace_state(), options.get_consistency());
-    tracing::set_optional_serial_consistency_level(client_state.get_trace_state(), options.get_serial_consistency());
-    tracing::add_query(client_state.get_trace_state(), prepared->raw_cql_statement);
-    tracing::add_prepared_statement(client_state.get_trace_state(), prepared);
+    tracing::set_page_size(trace_state, options.get_page_size());
+    tracing::set_consistency_level(trace_state, options.get_consistency());
+    tracing::set_optional_serial_consistency_level(trace_state, options.get_serial_consistency());
+    tracing::add_query(trace_state, prepared->raw_cql_statement);
+    tracing::add_prepared_statement(trace_state, prepared);
 
-    tracing::begin(client_state.get_trace_state(), seastar::value_of([&id] { return seastar::format("Execute CQL3 prepared query [{}]", id); }),
+    tracing::begin(trace_state, seastar::value_of([&id] { return seastar::format("Execute CQL3 prepared query [{}]", id); }),
                    client_state.get_client_address());
 
     auto stmt = prepared->statement;
-    tracing::trace(query_state.get_trace_state(), "Checking bounds");
+    tracing::trace(trace_state, "Checking bounds");
     if (stmt->get_bound_terms() != options.get_values_count()) {
         const auto msg = format("Invalid amount of bind variables: expected {:d} received {:d}",
                 stmt->get_bound_terms(),
                 options.get_values_count());
-        tracing::trace(query_state.get_trace_state(), msg);
+        tracing::trace(trace_state, msg);
         throw exceptions::invalid_request_exception(msg);
     }
 
@@ -828,7 +827,7 @@ future<std::unique_ptr<cql_server::response>> cql_server::connection::process_ex
 }
 
 future<std::unique_ptr<cql_server::response>>
-cql_server::connection::process_batch(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit)
+cql_server::connection::process_batch(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit, tracing::trace_state_ptr trace_state)
 {
     if (_version == 1) {
         throw exceptions::protocol_exception("BATCH messages are not support in version 1 of the protocol");
@@ -844,7 +843,7 @@ cql_server::connection::process_batch(uint16_t stream, request_reader in, servic
     modifications.reserve(n);
     values.reserve(n);
 
-    tracing::begin(client_state.get_trace_state(), "Execute batch of CQL3 queries", client_state.get_client_address());
+    tracing::begin(trace_state, "Execute batch of CQL3 queries", client_state.get_client_address());
 
     for ([[gnu::unused]] auto i : boost::irange(0u, n)) {
         const auto kind = in.read_byte();
@@ -858,7 +857,7 @@ cql_server::connection::process_batch(uint16_t stream, request_reader in, servic
             auto query = in.read_long_string_view();
             stmt_ptr = _server._query_processor.local().get_statement(query, client_state);
             ps = stmt_ptr->checked_weak_from_this();
-            tracing::add_query(client_state.get_trace_state(), query);
+            tracing::add_query(trace_state, query);
             break;
         }
         case 1: {
@@ -877,7 +876,7 @@ cql_server::connection::process_batch(uint16_t stream, request_reader in, servic
                 needs_authorization = pending_authorization_entries.emplace(std::move(cache_key), ps->checked_weak_from_this()).second;
             }
 
-            tracing::add_query(client_state.get_trace_state(), ps->raw_cql_statement);
+            tracing::add_query(trace_state, ps->raw_cql_statement);
             break;
         }
         default:
@@ -891,8 +890,8 @@ cql_server::connection::process_batch(uint16_t stream, request_reader in, servic
         }
 
         ::shared_ptr<cql3::statements::modification_statement> modif_statement_ptr = static_pointer_cast<cql3::statements::modification_statement>(ps->statement);
-        tracing::add_table_name(client_state.get_trace_state(), modif_statement_ptr->keyspace(), modif_statement_ptr->column_family());
-        tracing::add_prepared_statement(client_state.get_trace_state(), ps);
+        tracing::add_table_name(trace_state, modif_statement_ptr->keyspace(), modif_statement_ptr->column_family());
+        tracing::add_prepared_statement(trace_state, ps);
 
         modifications.emplace_back(std::move(modif_statement_ptr), needs_authorization);
 
@@ -907,15 +906,15 @@ cql_server::connection::process_batch(uint16_t stream, request_reader in, servic
         values.emplace_back(std::move(tmp));
     }
 
-    auto q_state = std::make_unique<cql_query_state>(client_state, std::move(permit));
+    auto q_state = std::make_unique<cql_query_state>(client_state, trace_state, std::move(permit));
     auto& query_state = q_state->query_state;
     // #563. CQL v2 encodes query_options in v1 format for batch requests.
     q_state->options = std::make_unique<cql3::query_options>(cql3::query_options::make_batch_options(std::move(*in.read_options(_version < 3 ? 1 : _version, _cql_serialization_format, this->timeout_config(), _server._cql_config)), std::move(values)));
     auto& options = *q_state->options;
 
-    tracing::set_consistency_level(client_state.get_trace_state(), options.get_consistency());
-    tracing::set_optional_serial_consistency_level(client_state.get_trace_state(), options.get_serial_consistency());
-    tracing::trace(client_state.get_trace_state(), "Creating a batch statement");
+    tracing::set_consistency_level(trace_state, options.get_consistency());
+    tracing::set_optional_serial_consistency_level(trace_state, options.get_serial_consistency());
+    tracing::trace(trace_state, "Creating a batch statement");
 
     auto batch = ::make_shared<cql3::statements::batch_statement>(cql3::statements::batch_statement::type(type), std::move(modifications), cql3::attributes::none(), _server._query_processor.local().get_cql_stats());
     return _server._query_processor.local().process_batch(batch, query_state, options, std::move(pending_authorization_entries)).then([this, stream, batch, &query_state] (auto msg) {
@@ -928,15 +927,15 @@ cql_server::connection::process_batch(uint16_t stream, request_reader in, servic
 }
 
 future<std::unique_ptr<cql_server::response>>
-cql_server::connection::process_register(uint16_t stream, request_reader in, service::client_state& client_state)
-{
+cql_server::connection::process_register(uint16_t stream, request_reader in, service::client_state& client_state,
+        tracing::trace_state_ptr trace_state) {
     std::vector<sstring> event_types;
     in.read_string_list(event_types);
     for (auto&& event_type : event_types) {
         auto et = parse_event_type(event_type);
         _server._notifier->register_event(et, this);
     }
-    return make_ready_future<std::unique_ptr<cql_server::response>>(make_ready(stream, client_state.get_trace_state()));
+    return make_ready_future<std::unique_ptr<cql_server::response>>(make_ready(stream, std::move(trace_state)));
 }
 
 std::unique_ptr<cql_server::response> cql_server::connection::make_unavailable_error(int16_t stream, exceptions::exception_code err, sstring msg, db::consistency_level cl, int32_t required, int32_t alive, const tracing::trace_state_ptr& tr_state)

transport/server.hh

@@ -94,8 +94,8 @@ struct cql_query_state {
     service::query_state query_state;
     std::unique_ptr<cql3::query_options> options;
 
-    cql_query_state(service::client_state& client_state, service_permit permit)
-        : query_state(client_state, std::move(permit))
+    cql_query_state(service::client_state& client_state, tracing::trace_state_ptr trace_state, service_permit permit)
+        : query_state(client_state, std::move(trace_state), std::move(permit))
     { }
 };
 
@@ -186,14 +186,14 @@ private:
         cql_binary_frame_v3 parse_frame(temporary_buffer<char> buf);
         future<fragmented_temporary_buffer> read_and_decompress_frame(size_t length, uint8_t flags);
         future<std::optional<cql_binary_frame_v3>> read_frame();
-        future<std::unique_ptr<cql_server::response>> process_startup(uint16_t stream, request_reader in, service::client_state& client_state);
-        future<std::unique_ptr<cql_server::response>> process_auth_response(uint16_t stream, request_reader in, service::client_state& client_state);
-        future<std::unique_ptr<cql_server::response>> process_options(uint16_t stream, request_reader in, service::client_state& client_state);
-        future<std::unique_ptr<cql_server::response>> process_query(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit);
-        future<std::unique_ptr<cql_server::response>> process_prepare(uint16_t stream, request_reader in, service::client_state& client_state);
-        future<std::unique_ptr<cql_server::response>> process_execute(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit);
-        future<std::unique_ptr<cql_server::response>> process_batch(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit);
-        future<std::unique_ptr<cql_server::response>> process_register(uint16_t stream, request_reader in, service::client_state& client_state);
+        future<std::unique_ptr<cql_server::response>> process_startup(uint16_t stream, request_reader in, service::client_state& client_state, tracing::trace_state_ptr trace_state);
+        future<std::unique_ptr<cql_server::response>> process_auth_response(uint16_t stream, request_reader in, service::client_state& client_state, tracing::trace_state_ptr trace_state);
+        future<std::unique_ptr<cql_server::response>> process_options(uint16_t stream, request_reader in, service::client_state& client_state, tracing::trace_state_ptr trace_state);
+        future<std::unique_ptr<cql_server::response>> process_query(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit, tracing::trace_state_ptr trace_state);
+        future<std::unique_ptr<cql_server::response>> process_prepare(uint16_t stream, request_reader in, service::client_state& client_state, tracing::trace_state_ptr trace_state);
+        future<std::unique_ptr<cql_server::response>> process_execute(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit, tracing::trace_state_ptr trace_state);
+        future<std::unique_ptr<cql_server::response>> process_batch(uint16_t stream, request_reader in, service::client_state& client_state, service_permit permit, tracing::trace_state_ptr trace_state);
+        future<std::unique_ptr<cql_server::response>> process_register(uint16_t stream, request_reader in, service::client_state& client_state, tracing::trace_state_ptr trace_state);
 
         std::unique_ptr<cql_server::response> make_unavailable_error(int16_t stream, exceptions::exception_code err, sstring msg, db::consistency_level cl, int32_t required, int32_t alive, const tracing::trace_state_ptr& tr_state);
         std::unique_ptr<cql_server::response> make_read_timeout_error(int16_t stream, exceptions::exception_code err, sstring msg, db::consistency_level cl, int32_t received, int32_t blockfor, bool data_present, const tracing::trace_state_ptr& tr_state);

utils/logalloc.cc

@@ -2064,6 +2064,17 @@ bool segment_pool::migrate_segment(segment* src, segment* dst)
 }
 
 void tracker::impl::register_region(region::impl* r) {
+    // If needed, increase capacity of regions before taking the reclaim lock,
+    // to avoid failing an allocation when push_back() tries to increase
+    // capacity.
+    //
+    // The capacity increase is atomic (wrt _regions) so it cannot be
+    // observed
+    if (_regions.size() == _regions.capacity()) {
+        auto copy = _regions;
+        copy.reserve(copy.capacity() * 2);
+        _regions = std::move(copy);
+    }
     reclaiming_lock _(*this);
     _regions.push_back(r);
     llogger.debug("Registered region @{} with id={}", r, r->id());