compress.hh: switch compressor::name() from an instance member to a virtual call

Before this patch, `compressor` is designed to be a proper abstract
class, where the creator of a compressor doesn't even know
what he's creating -- he passes a name, and it gets turned into a
`compressor` behind a scenes.

But later, when creation of compressors will involve looking up
dictionaries, this abstraction will only get in the way.
So we give up on keeping `compressor` abstract, and instead of
using "opaque" names we turn to an explicit enum of possible compressor types.

The main point of this patch is to add the `algorithm` enum and the `algorithm_to_name()`
function. The rest of the patch switches the `compressor::name()` function
to use `algorithm_to_name()` instead of the passed-by-constructor
`compressor::_name`, to keep a single source of truth for the names.

api/storage_service.cc

@@ -1491,7 +1491,7 @@ rest_sstable_info(http_context& ctx, std::unique_ptr<http::request> req) {
                                 if (!cp->options().contains(compression_parameters::SSTABLE_COMPRESSION)) {
                                     ss::mapper e;
                                     e.key = compression_parameters::SSTABLE_COMPRESSION;
-                                    e.value = cp->name();
+                                    e.value = sstring(cp->name());
                                     nm.attributes.push(std::move(e));
                                 }
                                 info.extended_properties.push(std::move(nm));

compress.cc

@@ -10,51 +10,43 @@
 #include <zlib.h>
 #include <snappy-c.h>
 
+#include <seastar/util/log.hh>
 #include "compress.hh"
 #include "exceptions/exceptions.hh"
 #include "utils/class_registrator.hh"
 
-sstring compressor::make_name(std::string_view short_name) {
-    return seastar::format("org.apache.cassandra.io.compress.{}", short_name);
-}
+static seastar::logger compressor_factory_logger("compressor_factory");
 
 class lz4_processor: public compressor {
 public:
-    using compressor::compressor;
-
     size_t uncompress(const char* input, size_t input_len, char* output,
                     size_t output_len) const override;
     size_t compress(const char* input, size_t input_len, char* output,
                     size_t output_len) const override;
     size_t compress_max_size(size_t input_len) const override;
+    algorithm get_algorithm() const override;
 };
 
 class snappy_processor: public compressor {
 public:
-    using compressor::compressor;
-
     size_t uncompress(const char* input, size_t input_len, char* output,
                     size_t output_len) const override;
     size_t compress(const char* input, size_t input_len, char* output,
                     size_t output_len) const override;
     size_t compress_max_size(size_t input_len) const override;
+    algorithm get_algorithm() const override { return algorithm::snappy; }
 };
 
 class deflate_processor: public compressor {
 public:
-    using compressor::compressor;
-
     size_t uncompress(const char* input, size_t input_len, char* output,
                     size_t output_len) const override;
     size_t compress(const char* input, size_t input_len, char* output,
                     size_t output_len) const override;
     size_t compress_max_size(size_t input_len) const override;
+    algorithm get_algorithm() const override { return algorithm::deflate; }
 };
 
-compressor::compressor(sstring name)
-    : _name(std::move(name))
-{}
-
 std::set<sstring> compressor::option_names() const {
     return {};
 }
@@ -63,12 +55,18 @@ std::map<sstring, sstring> compressor::options() const {
     return {};
 }
 
+std::string compressor::name() const {
+    auto result = std::string(compression_parameters::name_prefix);
+    result.append(compression_parameters::algorithm_to_name(get_algorithm()));
+    return result;
+}
+
 compressor::ptr_type compressor::create(const sstring& name, const opt_getter& opts) {
     if (name.empty()) {
         return {};
     }
 
-    qualified_name qn(make_name(""), name);
+    qualified_name qn(compression_parameters::name_prefix, name);
 
     for (auto& c : { lz4, snappy, deflate }) {
         if (c->name() == static_cast<const sstring&>(qn)) {
@@ -93,9 +91,9 @@ shared_ptr<compressor> compressor::create(const std::map<sstring, sstring>& opti
     return {};
 }
 
-thread_local const shared_ptr<compressor> compressor::lz4 = ::make_shared<lz4_processor>(make_name("LZ4Compressor"));
-thread_local const shared_ptr<compressor> compressor::snappy = ::make_shared<snappy_processor>(make_name("SnappyCompressor"));
-thread_local const shared_ptr<compressor> compressor::deflate = ::make_shared<deflate_processor>(make_name("DeflateCompressor"));
+thread_local const shared_ptr<compressor> compressor::lz4 = ::make_shared<lz4_processor>();
+thread_local const shared_ptr<compressor> compressor::snappy = ::make_shared<snappy_processor>();
+thread_local const shared_ptr<compressor> compressor::deflate = ::make_shared<deflate_processor>();
 
 const sstring compression_parameters::SSTABLE_COMPRESSION = "sstable_compression";
 const sstring compression_parameters::CHUNK_LENGTH_KB = "chunk_length_in_kb";
@@ -113,6 +111,23 @@ compression_parameters::compression_parameters(compressor_ptr c)
     : _compressor(std::move(c))
 {}
 
+std::string_view compression_parameters::algorithm_to_name(algorithm alg) {
+    switch (alg) {
+        case algorithm::lz4: return "LZ4Compressor";
+        case algorithm::deflate: return "DeflateCompressor";
+        case algorithm::snappy: return "SnappyCompressor";
+        case algorithm::zstd: return "ZstdCompressor";
+        case algorithm::none: on_internal_error(compressor_factory_logger, "algorithm_to_name(): called with algorithm::none");
+    }
+    abort();
+}
+
+std::string compression_parameters::algorithm_to_qualified_name(algorithm alg) {
+    auto result = std::string(name_prefix);
+    result.append(algorithm_to_name(alg));
+    return result;
+}
+
 compression_parameters::compression_parameters(const std::map<sstring, sstring>& options) {
     _compressor = compressor::create(options);
 
@@ -250,6 +265,10 @@ size_t lz4_processor::compress_max_size(size_t input_len) const {
     return LZ4_COMPRESSBOUND(input_len) + 4;
 }
 
+auto lz4_processor::get_algorithm() const -> algorithm {
+    return algorithm::lz4;
+}
+
 size_t deflate_processor::uncompress(const char* input,
                 size_t input_len, char* output, size_t output_len) const {
     z_stream zs;

compress.hh

@@ -18,9 +18,14 @@
 #include "seastarx.hh"
 
 class compressor {
-    sstring _name;
 public:
-    compressor(sstring);
+    enum class algorithm {
+        lz4,
+        zstd,
+        snappy,
+        deflate,
+        none,
+    };
 
     virtual ~compressor() {}
 
@@ -50,12 +55,9 @@ public:
      */
     virtual std::map<sstring, sstring> options() const;
 
-    /**
-     * Compressor class name.
-     */
-    const sstring& name() const {
-        return _name;
-    }
+    std::string name() const;
+
+    virtual algorithm get_algorithm() const = 0;
 
     // to cheaply bridge sstable compression options / maps
     using opt_string = std::optional<sstring>;
@@ -68,8 +70,6 @@ public:
     static thread_local const ptr_type lz4;
     static thread_local const ptr_type snappy;
     static thread_local const ptr_type deflate;
-
-    static sstring make_name(std::string_view short_name);
 };
 
 template<typename BaseType, typename... Args>
@@ -80,6 +80,9 @@ using compressor_registry = class_registry<compressor, const typename compressor
 
 class compression_parameters {
 public:
+    using algorithm = compressor::algorithm;
+    static constexpr std::string_view name_prefix = "org.apache.cassandra.io.compress.";
+
     static constexpr int32_t DEFAULT_CHUNK_LENGTH = 4 * 1024;
     static constexpr double DEFAULT_CRC_CHECK_CHANCE = 1.0;
 
@@ -108,6 +111,8 @@ public:
     static compression_parameters no_compression() {
         return compression_parameters(nullptr);
     }
+    static std::string_view algorithm_to_name(algorithm);
+    static std::string algorithm_to_qualified_name(algorithm);
 private:
     void validate_options(const std::map<sstring, sstring>&);
 };

sstables/compress.cc

@@ -295,7 +295,7 @@ size_t local_compression::compress_max_size(size_t input_len) const {
 
 void compression::set_compressor(compressor_ptr c) {
     if (c) {
-        unqualified_name uqn(compressor::make_name(""), c->name());
+        unqualified_name uqn(compression_parameters::name_prefix, c->name());
         const sstring& cn = uqn;
         name.value = bytes(cn.begin(), cn.end());
         for (auto& [k, v] : c->options()) {

zstd.cc

@@ -20,7 +20,6 @@
 #include <concepts>
 
 static const sstring COMPRESSION_LEVEL = "compression_level";
-static const sstring COMPRESSOR_NAME = compressor::make_name("ZstdCompressor");
 static const size_t DCTX_SIZE = ZSTD_estimateDCtxSize();
 
 class zstd_processor : public compressor {
@@ -83,10 +82,10 @@ public:
 
     std::set<sstring> option_names() const override;
     std::map<sstring, sstring> options() const override;
+    algorithm get_algorithm() const override;
 };
 
-zstd_processor::zstd_processor(const opt_getter& opts)
-    : compressor(COMPRESSOR_NAME) {
+zstd_processor::zstd_processor(const opt_getter& opts) {
     auto level = opts(COMPRESSION_LEVEL);
     if (level) {
         try {
@@ -152,5 +151,9 @@ std::map<sstring, sstring> zstd_processor::options() const {
     return {{COMPRESSION_LEVEL, std::to_string(_compression_level)}};
 }
 
+auto zstd_processor::get_algorithm() const -> algorithm {
+    return algorithm::zstd;
+}
+
 static const class_registrator<compressor, zstd_processor, const compressor::opt_getter&>
-    registrator(COMPRESSOR_NAME);
+    registrator(sstring(compression_parameters::algorithm_to_name(compressor::algorithm::zstd)));