diff --git a/auth/resource.cc b/auth/resource.cc
index 84f7859a90..e088ebdab4 100644
--- a/auth/resource.cc
+++ b/auth/resource.cc
@@ -210,9 +210,9 @@ std::ostream &operator<<(std::ostream &os, const service_level_resource_view &v)
 
 sstring encode_signature(std::string_view name, std::vector<data_type> args) {
     return format("{}[{}]", name,
-            ::join("^", args | boost::adaptors::transformed([] (const data_type t) {
+            fmt::join(args | boost::adaptors::transformed([] (const data_type t) {
                 return t->name();
-            })));
+            }), "^"));
 }
 
 std::pair<sstring, std::vector<data_type>> decode_signature(std::string_view encoded_signature) {
diff --git a/compound_compat.hh b/compound_compat.hh
index edbfc9063c..cbe97575c9 100644
--- a/compound_compat.hh
+++ b/compound_compat.hh
@@ -626,7 +626,8 @@ public:
     bool operator!=(const composite_view& k) const { return !(k == *this); }
 
     friend inline std::ostream& operator<<(std::ostream& os, composite_view v) {
-        return os << "{" << ::join(", ", v.components()) << ", compound=" << v._is_compound << ", static=" << v.is_static() << "}";
+        fmt::print(os, "{{{}, compound={}, static={}}}", fmt::join(v.components(), ", "), v._is_compound, v.is_static());
+        return os;
     }
 };
 
diff --git a/counters.cc b/counters.cc
index d5b41e76ba..8d54b3ab14 100644
--- a/counters.cc
+++ b/counters.cc
@@ -18,7 +18,8 @@ std::ostream& operator<<(std::ostream& os, counter_shard_view csv) {
 }
 
 std::ostream& operator<<(std::ostream& os, counter_cell_view ccv) {
-    return os << "{counter_cell timestamp: " << ccv.timestamp() << " shards: {" << ::join(", ", ccv.shards()) << "}}";
+    fmt::print(os, "{{counter_cell timestamp: {} shards: {{{}}}}}", ccv.timestamp(), fmt::join(ccv.shards(), ", "));
+    return os;
 }
 
 void counter_cell_builder::do_sort_and_remove_duplicates()
diff --git a/cql3/cql3_type.cc b/cql3/cql3_type.cc
index fbf3fba873..69a16dcd5e 100644
--- a/cql3/cql3_type.cc
+++ b/cql3/cql3_type.cc
@@ -262,7 +262,7 @@ class cql3_type::raw_tuple : public raw {
     std::vector<shared_ptr<raw>> _types;
 
     virtual sstring to_string() const override {
-        return format("tuple<{}>", join(", ", _types));
+        return format("tuple<{}>", fmt::join(_types, ", "));
     }
 public:
     raw_tuple(std::vector<shared_ptr<raw>> types)
diff --git a/cql3/functions/abstract_function.hh b/cql3/functions/abstract_function.hh
index f629f8405e..1c72f80942 100644
--- a/cql3/functions/abstract_function.hh
+++ b/cql3/functions/abstract_function.hh
@@ -63,7 +63,7 @@ public:
     }
 
     virtual sstring column_name(const std::vector<sstring>& column_names) const override {
-        return format("{}({})", _name, join(", ", column_names));
+        return format("{}({})", _name, fmt::join(column_names, ", "));
     }
 
     virtual void print(std::ostream& os) const override;
diff --git a/cql3/functions/functions.cc b/cql3/functions/functions.cc
index 9756680342..a4784e982e 100644
--- a/cql3/functions/functions.cc
+++ b/cql3/functions/functions.cc
@@ -417,13 +417,13 @@ functions::get(data_dictionary::database db,
     if (compatibles.empty()) {
         throw exceptions::invalid_request_exception(
                 format("Invalid call to function {}, none of its type signatures match (known type signatures: {})",
-                                                        name, join(", ", candidates)));
+                                                        name, fmt::join(candidates, ", ")));
     }
 
     if (compatibles.size() > 1) {
         throw exceptions::invalid_request_exception(
                 format("Ambiguous call to function {} (can be matched by following signatures: {}): use type casts to disambiguate",
-                    name, join(", ", compatibles)));
+                    name, fmt::join(compatibles, ", ")));
     }
 
     return std::move(compatibles[0]);
diff --git a/cql3/query_processor.cc b/cql3/query_processor.cc
index e1dae6afe9..10a0f5ad46 100644
--- a/cql3/query_processor.cc
+++ b/cql3/query_processor.cc
@@ -836,7 +836,7 @@ query_processor::execute_internal(
         cache_internal cache) {
 
     if (log.is_enabled(logging::log_level::trace)) {
-        log.trace("execute_internal: {}\"{}\" ({})", cache ? "(cached) " : "", query_string, ::join(", ", values));
+        log.trace("execute_internal: {}\"{}\" ({})", cache ? "(cached) " : "", query_string, fmt::join(values, ", "));
     }
     if (cache) {
         return execute_with_params(prepare_internal(query_string), cl, query_state, values);
diff --git a/cql3/restrictions/statement_restrictions.cc b/cql3/restrictions/statement_restrictions.cc
index b0fa0743c6..ad62ca7c63 100644
--- a/cql3/restrictions/statement_restrictions.cc
+++ b/cql3/restrictions/statement_restrictions.cc
@@ -31,6 +31,24 @@
 #include "types/map.hh"
 #include "types/set.hh"
 
+namespace {
+struct maybe_column_definition {
+    const column_definition* value;
+};
+}
+
+template<>
+struct fmt::formatter<maybe_column_definition> : fmt::formatter<std::string_view> {
+    template <typename FormatContext>
+    auto format(const maybe_column_definition& cd, FormatContext& ctx) const {
+        if (cd.value != nullptr) {
+            return fmt::format_to(ctx.out(), "{}", *cd.value);
+        } else {
+            return fmt::format_to(ctx.out(), "(null)");
+        }
+    }
+};
+
 namespace cql3 {
 namespace restrictions {
 
@@ -595,7 +613,11 @@ void statement_restrictions::add_single_column_parition_key_restriction(const ex
     if (has_token(_partition_key_restrictions)) {
         throw exceptions::invalid_request_exception(
                 format("Columns \"{}\" cannot be restricted by both a normal relation and a token relation",
-                        join(", ", expr::get_sorted_column_defs(_partition_key_restrictions))));
+                       fmt::join(expr::get_sorted_column_defs(_partition_key_restrictions) |
+                                 boost::adaptors::transformed([](auto* p) {
+                                   return maybe_column_definition{p};
+                                 }),
+                                 ", ")));
     }
 
     _partition_key_restrictions = expr::make_conjunction(_partition_key_restrictions, restr);
@@ -606,7 +628,11 @@ void statement_restrictions::add_token_partition_key_restriction(const expr::bin
     if (!partition_key_restrictions_is_empty() && !has_token(_partition_key_restrictions)) {
         throw exceptions::invalid_request_exception(
                 format("Columns \"{}\" cannot be restricted by both a normal relation and a token relation",
-                        join(", ", expr::get_sorted_column_defs(_partition_key_restrictions))));
+                        fmt::join(expr::get_sorted_column_defs(_partition_key_restrictions) |
+                                  boost::adaptors::transformed([](auto* p) {
+                                    return maybe_column_definition{p};
+                                  }),
+                                  ", ")));
     }
 
     _partition_key_restrictions = expr::make_conjunction(_partition_key_restrictions, restr);
diff --git a/cql3/selection/selectable.cc b/cql3/selection/selectable.cc
index 3a75138f22..336a6f597f 100644
--- a/cql3/selection/selectable.cc
+++ b/cql3/selection/selectable.cc
@@ -94,7 +94,7 @@ selectable::with_function::new_selector_factory(data_dictionary::database db, sc
 
 sstring
 selectable::with_function::to_string() const {
-    return format("{}({})", _function_name.name, join(", ", _args));
+    return format("{}({})", _function_name.name, fmt::join(_args, ", "));
 }
 
 expr::expression
@@ -112,7 +112,7 @@ selectable::with_anonymous_function::new_selector_factory(data_dictionary::datab
 
 sstring
 selectable::with_anonymous_function::to_string() const {
-    return format("{}({})", _function->name().name, join(", ", _args));
+    return format("{}({})", _function->name().name, fmt::join(_args, ", "));
 }
 
 shared_ptr<selector::factory>
diff --git a/cql3/statements/alter_table_statement.cc b/cql3/statements/alter_table_statement.cc
index 3254b1979f..f9ce6ec573 100644
--- a/cql3/statements/alter_table_statement.cc
+++ b/cql3/statements/alter_table_statement.cc
@@ -131,9 +131,9 @@ static void validate_column_rename(data_dictionary::database db, const schema& s
     if (!schema.indices().empty()) {
         auto dependent_indices = db.find_column_family(schema.id()).get_index_manager().get_dependent_indices(*def);
         if (!dependent_indices.empty()) {
-            auto index_names = ::join(", ", dependent_indices | boost::adaptors::transformed([](const index_metadata& im) {
+            auto index_names = fmt::join(dependent_indices | boost::adaptors::transformed([](const index_metadata& im) {
                 return im.name();
-            }));
+            }), ", ");
             throw exceptions::invalid_request_exception(
                     format("Cannot rename column {} because it has dependent secondary indexes ({})", from, index_names));
         }
diff --git a/cql3/statements/batch_statement.cc b/cql3/statements/batch_statement.cc
index 382ff48712..4ff1446fe4 100644
--- a/cql3/statements/batch_statement.cc
+++ b/cql3/statements/batch_statement.cc
@@ -210,7 +210,7 @@ void batch_statement::verify_batch_size(query_processor& qp, const std::vector<m
                 ks_cf_pairs.insert(m.schema()->ks_name() + "." + m.schema()->cf_name());
             }
             return format("Batch modifying {:d} partitions in {} is of size {:d} bytes, exceeding specified {} threshold of {:d} by {:d}.",
-                    mutations.size(), join(", ", ks_cf_pairs), size, type, threshold, size - threshold);
+                    mutations.size(), fmt::join(ks_cf_pairs, ", "), size, type, threshold, size - threshold);
         };
         if (size > fail_threshold) {
             _logger.error(error("FAIL", fail_threshold).c_str());
diff --git a/cql3/statements/create_table_statement.cc b/cql3/statements/create_table_statement.cc
index 590c14f477..0db7c166f8 100644
--- a/cql3/statements/create_table_statement.cc
+++ b/cql3/statements/create_table_statement.cc
@@ -333,7 +333,7 @@ std::unique_ptr<prepared_statement> create_table_statement::raw_statement::prepa
         } else {
             if (stmt->_columns.size() > 1) {
                 throw exceptions::invalid_request_exception(format("COMPACT STORAGE with composite PRIMARY KEY allows no more than one column not part of the PRIMARY KEY (got: {})",
-                    ::join( ", ", stmt->_columns | boost::adaptors::map_keys)));
+                    fmt::join(stmt->_columns | boost::adaptors::map_keys, ", ")));
             }
 #if 0
             Map.Entry<ColumnIdentifier, AbstractType> lastEntry = stmt.columns.entrySet().iterator().next();
diff --git a/cql3/statements/create_view_statement.cc b/cql3/statements/create_view_statement.cc
index 8e91a0a289..fdf530e031 100644
--- a/cql3/statements/create_view_statement.cc
+++ b/cql3/statements/create_view_statement.cc
@@ -251,7 +251,7 @@ view_ptr create_view_statement::prepare_view(data_dictionary::database db) const
     }
 
     if (!missing_pk_columns.empty()) {
-        auto column_names = ::join(", ", missing_pk_columns | boost::adaptors::transformed(std::mem_fn(&column_definition::name_as_text)));
+        auto column_names = fmt::join(missing_pk_columns | boost::adaptors::transformed(std::mem_fn(&column_definition::name_as_text)), ", ");
         throw exceptions::invalid_request_exception(format("Cannot create Materialized View {} without primary key columns from base {} ({})",
                         column_family(), _base_name.get_column_family(), column_names));
     }
@@ -294,7 +294,7 @@ view_ptr create_view_statement::prepare_view(data_dictionary::database db) const
             target_primary_keys.contains(non_pk_restrictions.cbegin()->first)) {
         // This case (filter by new PK column of the view) works, as explained above
     } else if (!non_pk_restrictions.empty()) {
-        auto column_names = ::join(", ", non_pk_restrictions | boost::adaptors::map_keys | boost::adaptors::transformed(std::mem_fn(&column_definition::name_as_text)));
+        auto column_names = fmt::join(non_pk_restrictions | boost::adaptors::map_keys | boost::adaptors::transformed(std::mem_fn(&column_definition::name_as_text)), ", ");
         throw exceptions::invalid_request_exception(format("Non-primary key columns cannot be restricted in the SELECT statement used for materialized view {} creation (got restrictions on: {})",
                 column_family(), column_names));
     }
diff --git a/cql3/statements/modification_statement.cc b/cql3/statements/modification_statement.cc
index 60dfea4794..d6c017ba86 100644
--- a/cql3/statements/modification_statement.cc
+++ b/cql3/statements/modification_statement.cc
@@ -398,10 +398,10 @@ modification_statement::process_where_clause(data_dictionary::database db, expr:
                 to_string(_restrictions->get_partition_key_restrictions())));
     }
     if (!_restrictions->get_non_pk_restriction().empty()) {
-        auto column_names = ::join(", ", _restrictions->get_non_pk_restriction()
+        auto column_names = fmt::join(_restrictions->get_non_pk_restriction()
                                          | boost::adaptors::map_keys
                                          | boost::adaptors::indirected
-                                         | boost::adaptors::transformed(std::mem_fn(&column_definition::name_as_text)));
+                                         | boost::adaptors::transformed(std::mem_fn(&column_definition::name_as_text)), ", ");
         throw exceptions::invalid_request_exception(format("Invalid where clause contains non PRIMARY KEY columns: {}", column_names));
     }
     const expr::expression& ck_restrictions = _restrictions->get_clustering_columns_restrictions();
diff --git a/cql3/statements/select_statement.cc b/cql3/statements/select_statement.cc
index 69d3abec2c..a3ddf3b4d4 100644
--- a/cql3/statements/select_statement.cc
+++ b/cql3/statements/select_statement.cc
@@ -2195,7 +2195,7 @@ std::unique_ptr<cql3::statements::raw::select_statement> build_select_statement(
         // as non alphanumeric characters.
         auto cols = boost::copy_range<std::vector<sstring>>(selected_columns
                 | boost::adaptors::transformed(std::mem_fn(&column_definition::name_as_cql_string)));
-        out << join(", ", cols);
+        fmt::print(out, "{}", fmt::join(cols, ", "));
     }
     // Note that cf_name may need to be quoted, just like column names above.
     out << " FROM " << util::maybe_quote(sstring(cf_name));
diff --git a/db/commitlog/commitlog_replayer.cc b/db/commitlog/commitlog_replayer.cc
index ebede1ad4f..019d885ad7 100644
--- a/db/commitlog/commitlog_replayer.cc
+++ b/db/commitlog/commitlog_replayer.cc
@@ -324,7 +324,7 @@ future<db::commitlog_replayer> db::commitlog_replayer::create_replayer(seastar::
 future<> db::commitlog_replayer::recover(std::vector<sstring> files, sstring fname_prefix) {
     typedef std::unordered_multimap<unsigned, sstring> shard_file_map;
 
-    rlogger.info("Replaying {}", join(", ", files));
+    rlogger.info("Replaying {}", fmt::join(files, ", "));
 
     // pre-compute work per shard already.
     auto map = ::make_lw_shared<shard_file_map>();
diff --git a/db/hints/host_filter.cc b/db/hints/host_filter.cc
index 7aa776a689..3b84a5a74a 100644
--- a/db/hints/host_filter.cc
+++ b/db/hints/host_filter.cc
@@ -78,7 +78,7 @@ sstring host_filter::to_configuration_string() const {
     case enabled_kind::enabled_for_all:
         return "true";
     case enabled_kind::enabled_selectively:
-        return ::join(",", _dcs);
+        return fmt::to_string(fmt::join(_dcs, ","));
     case enabled_kind::disabled_for_all:
         return "false";
     }
@@ -99,12 +99,12 @@ std::string_view host_filter::enabled_kind_to_string(host_filter::enabled_kind e
 }
 
 std::ostream& operator<<(std::ostream& os, const host_filter& f) {
-    os << "host_filter{enabled_kind="
-        << host_filter::enabled_kind_to_string(f._enabled_kind);
+    fmt::print(os, "host_filter{{enabled_kind={}",
+               host_filter::enabled_kind_to_string(f._enabled_kind));
     if (f._enabled_kind == host_filter::enabled_kind::enabled_selectively) {
-        os << ", dcs={" << ::join(",", f._dcs);
+        fmt::print(os, ", dcs={{{}}}", fmt::join(f._dcs, ","));
     }
-    os << "}";
+    fmt::print(os, "}}");
     return os;
 }
 
diff --git a/db/per_partition_rate_limit_options.cc b/db/per_partition_rate_limit_options.cc
index 56a2dca4c0..38e2b3351f 100644
--- a/db/per_partition_rate_limit_options.cc
+++ b/db/per_partition_rate_limit_options.cc
@@ -45,7 +45,7 @@ per_partition_rate_limit_options::per_partition_rate_limit_options(std::map<sstr
     if (!map.empty()) {
         throw exceptions::configuration_exception(format(
                 "Unknown keys in map for per_partition_rate_limit extension: {}",
-                ::join(", ", map | boost::adaptors::map_keys)));
+                fmt::join(map | boost::adaptors::map_keys, ", ")));
     }
 }
 
diff --git a/db/system_keyspace.cc b/db/system_keyspace.cc
index bdf94ad088..b5aab51622 100644
--- a/db/system_keyspace.cc
+++ b/db/system_keyspace.cc
@@ -1287,7 +1287,7 @@ future<> system_keyspace::save_local_supported_features(const std::set<std::stri
     static const auto req = format("INSERT INTO system.{} (key, supported_features) VALUES (?, ?)", LOCAL);
     return qctx->execute_cql(req,
         sstring(db::system_keyspace::LOCAL),
-        ::join(",", feats)).discard_result();
+        fmt::to_string(fmt::join(feats, ","))).discard_result();
 }
 
 // The cache must be distributed, because the values themselves may not update atomically, so a shard reading that
diff --git a/dht/i_partitioner.cc b/dht/i_partitioner.cc
index a09c99eeec..e0b1dd4c40 100644
--- a/dht/i_partitioner.cc
+++ b/dht/i_partitioner.cc
@@ -75,8 +75,10 @@ std::unique_ptr<dht::i_partitioner> make_partitioner(sstring partitioner_name) {
     try {
         return create_object<i_partitioner>(partitioner_name);
     } catch (std::exception& e) {
-        auto supported_partitioners = ::join(", ", class_registry<i_partitioner>::classes() |
-                boost::adaptors::map_keys);
+        auto supported_partitioners = fmt::join(
+            class_registry<i_partitioner>::classes() |
+            boost::adaptors::map_keys,
+            ", ");
         throw std::runtime_error(format("Partitioner {} is not supported, supported partitioners = {{ {} }} : {}",
                 partitioner_name, supported_partitioners, e.what()));
     }
diff --git a/gms/feature_service.cc b/gms/feature_service.cc
index c4a737bb1f..f076a74c11 100644
--- a/gms/feature_service.cc
+++ b/gms/feature_service.cc
@@ -200,7 +200,7 @@ void feature_service::persist_enabled_feature_info(const gms::feature& f) const
     }
     auto feats_set = to_feature_set(*raw_old_value);
     feats_set.emplace(f.name());
-    db::system_keyspace::set_scylla_local_param(ENABLED_FEATURES_KEY, ::join(",", feats_set)).get0();
+    db::system_keyspace::set_scylla_local_param(ENABLED_FEATURES_KEY, fmt::to_string(fmt::join(feats_set, ","))).get0();
 }
 
 void feature_service::enable(const std::set<std::string_view>& list) {
diff --git a/gms/versioned_value.cc b/gms/versioned_value.cc
index 243d981a26..23fec70b56 100644
--- a/gms/versioned_value.cc
+++ b/gms/versioned_value.cc
@@ -38,9 +38,8 @@ versioned_value versioned_value::network_version() {
 }
 
 sstring versioned_value::make_full_token_string(const std::unordered_set<dht::token>& tokens) {
-    return ::join(";", tokens | boost::adaptors::transformed([] (const dht::token& t) {
-        return t.to_sstring(); })
-    );
+    return fmt::to_string(fmt::join(tokens | boost::adaptors::transformed([] (const dht::token& t) {
+        return t.to_sstring(); }), ";"));
 }
 
 sstring versioned_value::make_token_string(const std::unordered_set<dht::token>& tokens) {
diff --git a/gms/versioned_value.hh b/gms/versioned_value.hh
index dedfa3a15e..0b44aec760 100644
--- a/gms/versioned_value.hh
+++ b/gms/versioned_value.hh
@@ -95,7 +95,7 @@ public:
     }
 
     static sstring version_string(const std::initializer_list<sstring>& args) {
-        return ::join(sstring(versioned_value::DELIMITER_STR), args);
+        return fmt::to_string(fmt::join(args, std::string_view(versioned_value::DELIMITER_STR)));
     }
 
     static sstring make_full_token_string(const std::unordered_set<dht::token>& tokens);
@@ -217,7 +217,7 @@ public:
     }
 
     static versioned_value supported_features(const std::set<std::string_view>& features) {
-        return versioned_value(::join(",", features));
+        return versioned_value(fmt::to_string(fmt::join(features, ",")));
     }
 
     static versioned_value cache_hitrates(const sstring& hitrates) {
diff --git a/mutation/atomic_cell.cc b/mutation/atomic_cell.cc
index 561a0dd26d..185182cf00 100644
--- a/mutation/atomic_cell.cc
+++ b/mutation/atomic_cell.cc
@@ -194,14 +194,13 @@ operator<<(std::ostream& os, const atomic_cell_view::printer& acvp) {
         std::ostringstream cell_value_string_builder;
         if (type.is_counter()) {
             if (acv.is_counter_update()) {
-                cell_value_string_builder << "counter_update_value=" << acv.counter_update_value();
+                fmt::print(cell_value_string_builder, "counter_update_value={}", acv.counter_update_value());
             } else {
-                cell_value_string_builder << "shards: ";
                 auto ccv = counter_cell_view(acv);
-                cell_value_string_builder << ::join(", ", ccv.shards());
+                fmt::print(cell_value_string_builder, "shards: {}", fmt::join(ccv.shards(), ", "));
             }
         } else {
-            cell_value_string_builder << type.to_string(to_bytes(acv.value()));
+            fmt::print(cell_value_string_builder, "{}", type.to_string(to_bytes(acv.value())));
         }
         fmt::print(os, "atomic_cell{{{},ts={:d},expiry={:d},ttl={:d}}}",
             cell_value_string_builder.str(),
diff --git a/mutation/mutation_fragment.cc b/mutation/mutation_fragment.cc
index abcc8cfac5..c3970d2109 100644
--- a/mutation/mutation_fragment.cc
+++ b/mutation/mutation_fragment.cc
@@ -476,7 +476,8 @@ std::ostream& operator<<(std::ostream& out, const range_tombstone_stream& rtl) {
 }
 
 std::ostream& operator<<(std::ostream& out, const clustering_interval_set& set) {
-    return out << "{" << ::join(",\n  ", set) << "}";
+    fmt::print(out, "{{{}}}", fmt::join(set, ",\n  "));
+    return out;
 }
 
 template<typename Hasher>
diff --git a/mutation/mutation_partition.cc b/mutation/mutation_partition.cc
index 01180c786b..9f11d19bf7 100644
--- a/mutation/mutation_partition.cc
+++ b/mutation/mutation_partition.cc
@@ -1058,7 +1058,7 @@ operator<<(std::ostream& os, const mutation_partition::printer& p) {
         os << indent << "tombstone: " << mp._tombstone << ",\n";
     }
     if (!mp._row_tombstones.empty()) {
-        os << indent << "range_tombstones: {" << ::join(",", prefixed("\n    ", mp._row_tombstones)) << "},\n";
+        fmt::print(os, "{}range_tombstones: {{{}}},\n", indent, fmt::join(prefixed("\n    ", mp._row_tombstones), ","));
     }
 
     if (!mp.static_row().empty()) {
diff --git a/mutation/range_tombstone_list.cc b/mutation/range_tombstone_list.cc
index bc90422397..de973d63bd 100644
--- a/mutation/range_tombstone_list.cc
+++ b/mutation/range_tombstone_list.cc
@@ -425,7 +425,8 @@ void range_tombstone_list::update_undo_op::undo(const schema& s, range_tombstone
 }
 
 std::ostream& operator<<(std::ostream& out, const range_tombstone_list& list) {
-    return out << "{" << ::join(", ", list) << "}";
+    fmt::print(out, "{{{}}}", fmt::join(list, ", "));
+    return out;
 }
 
 std::ostream& operator<<(std::ostream& out, const range_tombstone_entry& rt) {
diff --git a/query.cc b/query.cc
index 5bf949d8e4..e4f5a456f2 100644
--- a/query.cc
+++ b/query.cc
@@ -40,17 +40,18 @@ const clustering_range full_clustering_range = clustering_range::make_open_ended
 std::ostream& operator<<(std::ostream& out, const specific_ranges& s);
 
 std::ostream& operator<<(std::ostream& out, const partition_slice& ps) {
-    out << "{"
-        << "regular_cols=[" << join(", ", ps.regular_columns) << "]"
-        << ", static_cols=[" << join(", ", ps.static_columns) << "]"
-        << ", rows=[" << join(", ", ps._row_ranges) << "]"
-        ;
+    fmt::print(out,
+               "{{regular_cols=[{}], static_cols=[{}], rows=[{}]",
+               fmt::join(ps.regular_columns, ", "),
+               fmt::join(ps.static_columns, ", "),
+               ps._row_ranges);
     if (ps._specific_ranges) {
-        out << ", specific=[" << *ps._specific_ranges << "]";
+        fmt::print(out, ", specific=[{}]", *ps._specific_ranges);
     }
-    out << ", options=" << format("{:x}", ps.options.mask()); // FIXME: pretty print options
-    out << ", partition_row_limit=" << ps.partition_row_limit();
-    return out << "}";
+    // FIXME: pretty print options
+    fmt::print(out, ", options={:x}, , partition_row_limit={}}}",
+               ps.options.mask(), ps.partition_row_limit());
+    return out;
 }
 
 std::ostream& operator<<(std::ostream& out, const read_command& r) {
@@ -81,29 +82,28 @@ std::ostream& operator<<(std::ostream& out, const forward_request::reduction_typ
 }
 
 std::ostream& operator<<(std::ostream& out, const forward_request::aggregation_info& a) {
-    return out << "aggregation_info{"
-        << ", name=" << a.name
-        << ", column_names=[" << join(",", a.column_names) << "]"
-        << "}";
+    fmt::print(out, "aggregation_info{{, name={}, column_names=[{}]}}",
+               a.name, fmt::join(a.column_names, ","));;
+    return out;
 }
 
 std::ostream& operator<<(std::ostream& out, const forward_request& r) {
     auto ms = std::chrono::time_point_cast<std::chrono::milliseconds>(r.timeout).time_since_epoch().count();
-
-    out << "forward_request{"
-        << "reductions=[" << join(",", r.reduction_types) << "]";
-    if(r.aggregation_infos) {
-        out << ", aggregation_infos=[" << join(",", r.aggregation_infos.value()) << "]";
+    fmt::print(out, "forward_request{{reductions=[{}]",
+               fmt::join(r.reduction_types, ","));
+    if (r.aggregation_infos) {
+        fmt::print(out, ", aggregation_infos=[{}]",
+                   fmt::join(r.aggregation_infos.value(), ","));
     }
-    return out << ", cmd=" << r.cmd
-        << ", pr=" << r.pr
-        << ", cl=" << r.cl
-        << ", timeout(ms)=" << ms << "}";
+    fmt::print(out, "cmd={}, pr={}, cl={}, timeout(ms)={}}}",
+               r.cmd, r.pr, r.cl, ms);
+    return out;
 }
 
 
 std::ostream& operator<<(std::ostream& out, const specific_ranges& s) {
-    return out << "{" << s._pk << " : " << join(", ", s._ranges) << "}";
+    fmt::print(out, "{{{} : {}}}", s._pk, fmt::join(s._ranges, ", "));
+    return out;
 }
 
 void trim_clustering_row_ranges_to(const schema& s, clustering_row_ranges& ranges, position_in_partition pos, bool reversed) {
diff --git a/replica/database.cc b/replica/database.cc
index 27fe5bdf87..b16533431c 100644
--- a/replica/database.cc
+++ b/replica/database.cc
@@ -2129,7 +2129,7 @@ std::ostream&
 operator<<(std::ostream& os, const exploded_clustering_prefix& ecp) {
     // Can't pass to_hex() to transformed(), since it is overloaded, so wrap:
     auto enhex = [] (auto&& x) { return to_hex(x); };
-    fmt::print(os, "prefix{{{}}}", ::join(":", ecp._v | boost::adaptors::transformed(enhex)));
+    fmt::print(os, "prefix{{{}}}", fmt::join(ecp._v | boost::adaptors::transformed(enhex), ":"));
     return os;
 }
 
diff --git a/replica/memtable.cc b/replica/memtable.cc
index f3d816a831..0f0c51628c 100644
--- a/replica/memtable.cc
+++ b/replica/memtable.cc
@@ -865,7 +865,8 @@ size_t memtable_entry::object_memory_size(allocation_strategy& allocator) {
 
 std::ostream& operator<<(std::ostream& out, memtable& mt) {
     logalloc::reclaim_lock rl(mt);
-    return out << "{memtable: [" << ::join(",\n", mt.partitions) << "]}";
+    fmt::print(out, "{{memtable: [{}]}}", fmt::join(mt.partitions, ",\n"));
+    return out;
 }
 
 std::ostream& operator<<(std::ostream& out, const memtable_entry& mt) {
diff --git a/row_cache.cc b/row_cache.cc
index efbdc278df..aeb8177acf 100644
--- a/row_cache.cc
+++ b/row_cache.cc
@@ -1333,7 +1333,7 @@ void row_cache::upgrade_entry(cache_entry& e) {
 
 std::ostream& operator<<(std::ostream& out, row_cache& rc) {
     rc._read_section(rc._tracker.region(), [&] {
-        out << "{row_cache: " << ::join(", ", rc._partitions.begin(), rc._partitions.end()) << "}";
+        fmt::print(out, "{{row_cache: {}}}", fmt::join(rc._partitions.begin(), rc._partitions.end(), ", "));
     });
     return out;
 }
diff --git a/schema/schema.cc b/schema/schema.cc
index 69fed62f8e..958beb0614 100644
--- a/schema/schema.cc
+++ b/schema/schema.cc
@@ -110,12 +110,12 @@ std::ostream& operator<<(std::ostream& out, const column_mapping& cm) {
         // s->regular_column_name_type()->to_string(e.name()).
         return format("{{id={}, name=0x{}, type={}}}", i, e.name(), e.type()->name());
     };
-
-    return out << "{static=[" << ::join(", ", boost::irange<column_id>(0, n_static) |
-            boost::adaptors::transformed([&] (column_id i) { return pr_entry(i, cm.static_column_at(i)); }))
-        << "], regular=[" << ::join(", ", boost::irange<column_id>(0, n_regular) |
-            boost::adaptors::transformed([&] (column_id i) { return pr_entry(i, cm.regular_column_at(i)); }))
-        << "]}";
+    fmt::print(out, "{{static=[{}], regular=[{}]}}",
+               fmt::join(boost::irange<column_id>(0, n_static) |
+                         boost::adaptors::transformed([&] (column_id i) { return pr_entry(i, cm.static_column_at(i)); }), ", "),
+               fmt::join(boost::irange<column_id>(0, n_regular) |
+                         boost::adaptors::transformed([&] (column_id i) { return pr_entry(i, cm.regular_column_at(i)); }), ", "));
+    return out;
 }
 
 std::ostream& operator<<(std::ostream& os, ordinal_column_id id)
@@ -482,10 +482,11 @@ sstring schema::thrift_key_validator() const {
     if (partition_key_size() == 1) {
         return partition_key_columns().begin()->type->name();
     } else {
-        sstring type_params = ::join(", ", partition_key_columns()
+        auto type_params = fmt::join(partition_key_columns()
                             | boost::adaptors::transformed(std::mem_fn(&column_definition::type))
-                            | boost::adaptors::transformed(std::mem_fn(&abstract_type::name)));
-        return "org.apache.cassandra.db.marshal.CompositeType(" + type_params + ")";
+                            | boost::adaptors::transformed(std::mem_fn(&abstract_type::name)),
+                                        ", ");
+        return format("org.apache.cassandra.db.marshal.CompositeType({})", type_params);
     }
 }
 
diff --git a/schema/schema.hh b/schema/schema.hh
index 425f719bd0..d9e3122589 100644
--- a/schema/schema.hh
+++ b/schema/schema.hh
@@ -358,9 +358,6 @@ public:
     const bytes& name() const;
     sstring name_as_cql_string() const;
     friend std::ostream& operator<<(std::ostream& os, const column_definition& cd);
-    friend std::ostream& operator<<(std::ostream& os, const column_definition* cd) {
-        return cd != nullptr ? os << *cd : os << "(null)";
-    }
     bool has_component_index() const {
         return is_primary_key();
     }
diff --git a/service/migration_manager.cc b/service/migration_manager.cc
index e820e25b98..b43abe5083 100644
--- a/service/migration_manager.cc
+++ b/service/migration_manager.cc
@@ -768,7 +768,7 @@ future<std::vector<mutation>> migration_manager::prepare_column_family_drop_anno
                                     | boost::adaptors::filtered([&old_cfm](const view_ptr& v) { return !old_cfm.get_index_manager().is_index(v); })
                                     | boost::adaptors::transformed([](const view_ptr& v) { return v->cf_name(); });
             co_await coroutine::return_exception(exceptions::invalid_request_exception(format("Cannot drop table when materialized views still depend on it ({}.{{{}}})",
-                        schema->ks_name(), ::join(", ", explicit_view_names))));
+                        schema->ks_name(), fmt::join(explicit_view_names, ", "))));
         }
         mlogger.info("Drop table '{}.{}'", schema->ks_name(), schema->cf_name());
 
diff --git a/service/storage_service.cc b/service/storage_service.cc
index de78c7b141..255856e9d0 100644
--- a/service/storage_service.cc
+++ b/service/storage_service.cc
@@ -1835,7 +1835,7 @@ future<std::unordered_map<sstring, std::vector<sstring>>> storage_service::descr
         // we're done: the results map is ready to return to the client.  the rest is just debug logging:
         auto it_unreachable = results.find(UNREACHABLE);
         if (it_unreachable != results.end()) {
-            slogger.debug("Hosts not in agreement. Didn't get a response from everybody: {}", ::join( ",", it_unreachable->second));
+            slogger.debug("Hosts not in agreement. Didn't get a response from everybody: {}", fmt::join(it_unreachable->second, ","));
         }
         auto my_version = get_schema_version();
         for (auto&& entry : results) {
@@ -2060,7 +2060,7 @@ public:
             errors.emplace_back(format("The {} command failed for nodes={}: the needed nodes are down. It is highly recommended to fix the down nodes and try again", op_desc, nodes_failed));
         }
         if (!errors.empty()) {
-            co_await coroutine::return_exception(std::runtime_error(join("; ", errors)));
+            co_await coroutine::return_exception(std::runtime_error(fmt::to_string(fmt::join(errors, "; "))));
         }
         slogger.info("{}[{}]: Finished {}", desc, uuid(), req);
     }
diff --git a/sstables/mx/writer.cc b/sstables/mx/writer.cc
index dcc0ec903e..0938ce0883 100644
--- a/sstables/mx/writer.cc
+++ b/sstables/mx/writer.cc
@@ -364,10 +364,11 @@ static sstring pk_type_to_string(const schema& s) {
     if (s.partition_key_size() == 1) {
         return s.partition_key_columns().begin()->type->name();
     } else {
-        sstring type_params = ::join(",", s.partition_key_columns()
+        auto type_params = fmt::join(s.partition_key_columns()
                                           | boost::adaptors::transformed(std::mem_fn(&column_definition::type))
-                                          | boost::adaptors::transformed(std::mem_fn(&abstract_type::name)));
-        return "org.apache.cassandra.db.marshal.CompositeType(" + type_params + ")";
+                                          | boost::adaptors::transformed(std::mem_fn(&abstract_type::name)),
+                                        ",");
+        return format("org.apache.cassandra.db.marshal.CompositeType({})", type_params);
     }
 }
 
diff --git a/test/boost/row_cache_test.cc b/test/boost/row_cache_test.cc
index 1a89e3a590..d51e5c71cc 100644
--- a/test/boost/row_cache_test.cc
+++ b/test/boost/row_cache_test.cc
@@ -3435,7 +3435,7 @@ SEASTAR_TEST_CASE(test_concurrent_reads_and_eviction) {
                         return m2 == actual;
                     })) {
                         BOOST_FAIL(format("Mutation read doesn't match any expected version, slice: {}, read: {}\nexpected: [{}]",
-                            slice, actual, ::join(",\n", possible_versions)));
+                            slice, actual, fmt::join(possible_versions, ",\n")));
                     }
                 }
             }).finally([&] {
diff --git a/test/boost/view_schema_test.cc b/test/boost/view_schema_test.cc
index 7c9836a74b..dd12259daf 100644
--- a/test/boost/view_schema_test.cc
+++ b/test/boost/view_schema_test.cc
@@ -9,7 +9,8 @@
 
 #include <boost/test/unit_test.hpp>
 #include <boost/range/adaptor/map.hpp>
-
+#include <fmt/ostream.h>
+#include <fmt/ranges.h>
 #include "replica/database.hh"
 #include "types/user.hh"
 #include "db/view/node_view_update_backlog.hh"
@@ -1501,7 +1502,7 @@ SEASTAR_TEST_CASE(test_filter_with_type_cast) {
 SEASTAR_TEST_CASE(test_restrictions_on_all_types) {
     return do_with_cql_env_thread([] (auto& e) {
         e.execute_cql("create type myType (a int, b uuid, c set<text>)").get();
-        auto column_names = ::join(", ", std::vector<sstring>({
+        const std::string_view column_names[] = {
             "asciival",
             "bigintval",
             "blobval",
@@ -1523,7 +1524,7 @@ SEASTAR_TEST_CASE(test_restrictions_on_all_types) {
             "frozensetval",
             "frozenmapval",
             "tupleval",
-            "udtval"}));
+            "udtval"};
         e.execute_cql(fmt::format("create table cf ("
                     "asciival ascii, "
                     "bigintval bigint, "
@@ -1546,7 +1547,7 @@ SEASTAR_TEST_CASE(test_restrictions_on_all_types) {
                     "frozensetval frozen<set<uuid>>, "
                     "frozenmapval frozen<map<ascii, int>>,"
                     "tupleval frozen<tuple<int, ascii, uuid>>,"
-                    "udtval frozen<myType>, primary key ({}))", column_names)).get();
+                    "udtval frozen<myType>, primary key ({}))", fmt::join(column_names, ", "))).get();
 
         e.execute_cql(fmt::format("create materialized view vcf as select * from cf where "
                 "asciival = 'abc' AND "
@@ -1571,7 +1572,7 @@ SEASTAR_TEST_CASE(test_restrictions_on_all_types) {
                 "frozenmapval = {{'a': 1, 'b': 2}} AND "
                 "tupleval = (1, 'foobar', 6BDDC89A-5644-11E4-97FC-56847AFE9799) AND "
                 "udtval = {{a: 1, b: 6BDDC89A-5644-11E4-97FC-56847AFE9799, c: {{'foo', 'bar'}}}} "
-                "PRIMARY KEY ({})", column_names)).get();
+                "PRIMARY KEY ({})", fmt::join(column_names, ", "))).get();
 
         e.execute_cql(fmt::format("insert into cf ({}) values ( "
                 "'abc',"
@@ -1595,7 +1596,7 @@ SEASTAR_TEST_CASE(test_restrictions_on_all_types) {
                 "{{6BDDC89A-5644-11E4-97FC-56847AFE9799}},"
                 "{{'a': 1, 'b': 2}},"
                 "(1, 'foobar', 6BDDC89A-5644-11E4-97FC-56847AFE9799),"
-                "{{a: 1, b: 6BDDC89A-5644-11E4-97FC-56847AFE9799, c: {{'foo', 'bar'}}}})", column_names)).get();
+                "{{a: 1, b: 6BDDC89A-5644-11E4-97FC-56847AFE9799, c: {{'foo', 'bar'}}}})", fmt::join(column_names, ", "))).get();
 
         eventually([&] {
         auto msg = e.execute_cql("select * from vcf").get0();
diff --git a/test/lib/cql_assertions.cc b/test/lib/cql_assertions.cc
index deb1addc46..ed801063e3 100644
--- a/test/lib/cql_assertions.cc
+++ b/test/lib/cql_assertions.cc
@@ -151,7 +151,7 @@ rows_assertions::with_rows_ignore_order(std::vector<std::vector<bytes_opt>> rows
         });
         if (found == std::end(actual)) {
             fail(format("row {} not found in result set ({})", to_string(expected),
-               ::join(", ", actual | boost::adaptors::transformed([] (auto& r) { return to_string(r); }))));
+               fmt::join(actual | boost::adaptors::transformed([] (auto& r) { return to_string(r); }), ", ")));
         }
     }
     if (rs.size() != rows.size()) {
diff --git a/test/lib/flat_mutation_reader_assertions.hh b/test/lib/flat_mutation_reader_assertions.hh
index 9e5755be32..fc3202c794 100644
--- a/test/lib/flat_mutation_reader_assertions.hh
+++ b/test/lib/flat_mutation_reader_assertions.hh
@@ -363,7 +363,7 @@ public:
     }
 
     flat_reader_assertions_v2& produces(mutation_fragment_v2::kind k, std::vector<int> ck_elements, bool make_full_key = false) {
-        testlog.trace("Expect {} {{{}}}", k, ::join(", ", ck_elements));
+        testlog.trace("Expect {} {{{}}}", k, fmt::join(ck_elements, ", "));
         std::vector<bytes> ck_bytes;
         for (auto&& e : ck_elements) {
             ck_bytes.emplace_back(int32_type->decompose(e));
diff --git a/test/lib/result_set_assertions.cc b/test/lib/result_set_assertions.cc
index 56627c2042..e3888464dc 100644
--- a/test/lib/result_set_assertions.cc
+++ b/test/lib/result_set_assertions.cc
@@ -49,7 +49,7 @@ row_assertion::matches(const query::result_set_row& row) const {
 
 sstring
 row_assertion::describe(schema_ptr schema) const {
-    return "{" + ::join(", ", _expected_values | boost::adaptors::transformed([&schema] (auto&& e) {
+    return format("{{{}}}", fmt::join(_expected_values | boost::adaptors::transformed([&schema] (auto&& e) {
         auto&& name = e.first;
         auto&& value = e.second;
         const column_definition* def = schema->get_column_definition(name);
@@ -61,7 +61,7 @@ row_assertion::describe(schema_ptr schema) const {
         } else {
             return format("{}=\"{}\"", to_sstring(name), def->type->to_string(def->type->decompose(value)));
         }
-    })) + "}";
+    }), ", "));
 }
 
 const result_set_assertions&
diff --git a/types/types.cc b/types/types.cc
index 8e310404e9..d20ab48bf0 100644
--- a/types/types.cc
+++ b/types/types.cc
@@ -1007,8 +1007,8 @@ static sstring cql3_type_name_impl(const abstract_type& t) {
         sstring operator()(const time_type_impl&) { return "time"; }
         sstring operator()(const timeuuid_type_impl&) { return "timeuuid"; }
         sstring operator()(const tuple_type_impl& t) {
-            return format("tuple<{}>", ::join(", ", t.all_types() | boost::adaptors::transformed(std::mem_fn(
-                                                                            &abstract_type::as_cql3_type))));
+            return format("tuple<{}>", fmt::join(t.all_types() | boost::adaptors::transformed(std::mem_fn(
+                                                                            &abstract_type::as_cql3_type)), ", "));
         }
         sstring operator()(const user_type_impl& u) { return u.get_name_as_cql_string(); }
         sstring operator()(const utf8_type_impl&) { return "text"; }
@@ -1176,14 +1176,14 @@ static sstring map_to_string(const std::vector<std::pair<data_value, data_value>
         out << "(";
     }
 
-    out << ::join(", ", v | boost::adaptors::transformed([] (const std::pair<data_value, data_value>& p) {
+    fmt::print(out, "{}", fmt::join(v | boost::adaptors::transformed([] (const std::pair<data_value, data_value>& p) {
         std::ostringstream out;
         const auto& k = p.first;
         const auto& v = p.second;
         out << "{" << k.type()->to_string_impl(k) << " : ";
         out << v.type()->to_string_impl(v) << "}";
         return out.str();
-    }));
+    }), ", "));
 
     if (include_frozen_type) {
         out << ")";
@@ -1489,8 +1489,9 @@ list_type_impl::deserialize(View in) const {
 template data_value list_type_impl::deserialize<>(ser::buffer_view<bytes_ostream::fragment_iterator>) const;
 
 static sstring vector_to_string(const std::vector<data_value>& v, std::string_view sep) {
-    return join(sstring(sep),
-            v | boost::adaptors::transformed([] (const data_value& e) { return e.type()->to_string_impl(e); }));
+    return fmt::to_string(fmt::join(
+            v | boost::adaptors::transformed([] (const data_value& e) { return e.type()->to_string_impl(e); }),
+            sep));
 }
 
 template <typename F>
@@ -2963,7 +2964,7 @@ tuple_type_impl::make_name(const std::vector<data_type>& types) {
     // "org.apache.cassandra.db.marshal.FrozenType(...)".
     // Even when the tuple is frozen.
     // For more details see #4087
-    return format("org.apache.cassandra.db.marshal.TupleType({})", ::join(", ", types | boost::adaptors::transformed(std::mem_fn(&abstract_type::name))));
+    return format("org.apache.cassandra.db.marshal.TupleType({})", fmt::join(types | boost::adaptors::transformed(std::mem_fn(&abstract_type::name)), ", "));
 }
 
 static std::optional<std::vector<data_type>>