schema: add per-table tablet options

Unlike with vnodes, each tablet is served only by a single
shard, and it is associated with a memtable that, when
flushed, it creates sstables which token-range is confined
to the tablet owning them.

On one hand, this allows for far better agility and elasticity
since migration of tablets between nodes or shards does not
require rewriting most if not all of the sstables, as required
with vnodes (at the cleanup phase).

Having too few tablets might limit performance due not
being served by all shards or by imbalance between shards
caused by quantization.  The number of tabelts per table has to be
a power of 2 with the current design, and when divided by the
number of shards, some shards will serve N tablets, while others
may serve N+1, and when N is small N+1/N may be significantly
larger than 1. For example, with N=1, some shards will serve
2 tablet replicas and some will serve only 1, causing an imbalance
of 100%.

Now, simply allocating a lot more tablets for each table may
theoretically address this problem, but practically:
a. Each tablet has memory overhead and having too many tablets
in the system with many tables and many tablets for each of them
may overwhelm the system's and cause out-of-memory errors.
b. Too-small tablets cause a proliferation of small sstables
that are less efficient to acces, have higher metadata overhead
(due to per-sstable overhead), and might exhaust the system's
open file-descriptors limitations.

The options introduced in this change can help the user tune
the system in two ways:
1. Sizing the table to prevent unnecessary tablet splits
and migrations.  This can be done when the table is created,
or later on, using ALTER TABLE.
2. Controlling min_per_shard_tablet_count to improve
tablet balancing, for hot tables.

Signed-off-by: Benny Halevy <[email protected]>

Author: bhalevy

configure.py

@@ -1012,6 +1012,7 @@ def find_ninja():
                 'db/view/view_update_generator.cc',
                 'db/virtual_table.cc',
                 'db/virtual_tables.cc',
+                'db/tablet_options.cc',
                 'index/secondary_index_manager.cc',
                 'index/secondary_index.cc',
                 'utils/UUID_gen.cc',

cql3/statements/cf_prop_defs.cc

@@ -9,6 +9,7 @@
  */
 
 #include "cql3/statements/cf_prop_defs.hh"
+#include "cql3/statements/request_validations.hh"
 #include "data_dictionary/data_dictionary.hh"
 #include "db/extensions.hh"
 #include "db/tags/extension.hh"
@@ -20,6 +21,7 @@
 #include "tombstone_gc.hh"
 #include "db/per_partition_rate_limit_extension.hh"
 #include "db/per_partition_rate_limit_options.hh"
+#include "db/tablet_options.hh"
 #include "utils/bloom_calculations.hh"
 
 #include <boost/algorithm/string/predicate.hpp>
@@ -52,6 +54,8 @@ const sstring cf_prop_defs::COMPACTION_STRATEGY_CLASS_KEY = "class";
 
 const sstring cf_prop_defs::COMPACTION_ENABLED_KEY = "enabled";
 
+const sstring cf_prop_defs::KW_TABLETS = "tablets";
+
 schema::extensions_map cf_prop_defs::make_schema_extensions(const db::extensions& exts) const {
     schema::extensions_map er;
     for (auto& p : exts.schema_extensions()) {
@@ -68,20 +72,30 @@ schema::extensions_map cf_prop_defs::make_schema_extensions(const db::extensions
     return er;
 }
 
+data_dictionary::keyspace cf_prop_defs::find_keyspace(const data_dictionary::database db, std::string_view ks_name) {
+    try {
+        return db.find_keyspace(ks_name);
+    } catch (const data_dictionary::no_such_keyspace& e) {
+        throw request_validations::invalid_request("{}", e.what());
+    }
+}
+
 void cf_prop_defs::validate(const data_dictionary::database db, sstring ks_name, const schema::extensions_map& schema_extensions) const {
     // Skip validation if the comapction strategy class is already set as it means we've already
     // prepared (and redoing it would set strategyClass back to null, which we don't want)
     if (_compaction_strategy_class) {
         return;
     }
 
+    const auto& ks = find_keyspace(db, ks_name);
+
     static std::set<sstring> keywords({
         KW_COMMENT,
         KW_GCGRACESECONDS, KW_CACHING, KW_DEFAULT_TIME_TO_LIVE,
         KW_MIN_INDEX_INTERVAL, KW_MAX_INDEX_INTERVAL, KW_SPECULATIVE_RETRY,
         KW_BF_FP_CHANCE, KW_MEMTABLE_FLUSH_PERIOD, KW_COMPACTION,
         KW_COMPRESSION, KW_CRC_CHECK_CHANCE,  KW_ID, KW_PAXOSGRACESECONDS,
-        KW_SYNCHRONOUS_UPDATES
+        KW_SYNCHRONOUS_UPDATES, KW_TABLETS,
     });
     static std::set<sstring> obsolete_keywords({
         sstring("index_interval"),
@@ -162,6 +176,16 @@ void cf_prop_defs::validate(const data_dictionary::database db, sstring ks_name,
     }
 
     speculative_retry::from_sstring(get_string(KW_SPECULATIVE_RETRY, speculative_retry(speculative_retry::type::NONE, 0).to_sstring()));
+
+    if (auto tablet_options_map = get_tablet_options()) {
+        if (!ks.uses_tablets()) {
+            throw exceptions::configuration_exception("tablet options cannot be used when tablets are disabled for the keyspace");
+        }
+        if (!db.features().tablet_options) {
+            throw exceptions::configuration_exception("tablet options cannot be used until all nodes in the cluster enable this feature");
+        }
+        db::tablet_options::validate(*tablet_options_map);
+    }
 }
 
 std::map<sstring, sstring> cf_prop_defs::get_compaction_type_options() const {
@@ -252,6 +276,13 @@ const db::per_partition_rate_limit_options* cf_prop_defs::get_per_partition_rate
     return &ext->get_options();
 }
 
+std::optional<db::tablet_options::map_type> cf_prop_defs::get_tablet_options() const {
+    if (auto tablet_options = get_map(KW_TABLETS)) {
+        return tablet_options.value();
+    }
+    return std::nullopt;
+}
+
 void cf_prop_defs::apply_to_builder(schema_builder& builder, schema::extensions_map schema_extensions, const data_dictionary::database& db, sstring ks_name) const {
     if (has_property(KW_COMMENT)) {
         builder.set_comment(get_string(KW_COMMENT, ""));
@@ -351,6 +382,10 @@ void cf_prop_defs::apply_to_builder(schema_builder& builder, schema::extensions_
 
         builder.add_extension(db::tags_extension::NAME, ::make_shared<db::tags_extension>(tags_map));
     }
+
+    if (auto tablet_options_opt = get_map(KW_TABLETS)) {
+        builder.set_tablet_options(std::move(*tablet_options_opt));
+    }
 }
 
 void cf_prop_defs::validate_minimum_int(const sstring& field, int32_t minimum_value, int32_t default_value) const

cql3/statements/cf_prop_defs.hh

@@ -18,12 +18,14 @@
 
 namespace data_dictionary {
 class database;
+class keyspace;
 }
 
 class tombstone_gc_options;
 
 namespace db {
 class extensions;
+class tablet_options;
 }
 namespace cdc {
 class options;
@@ -60,6 +62,8 @@ public:
     static const sstring COMPACTION_STRATEGY_CLASS_KEY;
     static const sstring COMPACTION_ENABLED_KEY;
 
+    static const sstring KW_TABLETS;
+
     // FIXME: In origin the following consts are in CFMetaData.
     static constexpr int32_t DEFAULT_DEFAULT_TIME_TO_LIVE = 0;
     static constexpr int32_t DEFAULT_MIN_INDEX_INTERVAL = 128;
@@ -70,6 +74,7 @@ public:
 
 private:
     mutable std::optional<sstables::compaction_strategy_type> _compaction_strategy_class;
+    static data_dictionary::keyspace find_keyspace(const data_dictionary::database db, std::string_view ks_name);
 public:
     std::optional<sstables::compaction_strategy_type> get_compaction_strategy_class() const;
 
@@ -103,6 +108,7 @@ public:
     int32_t get_paxos_grace_seconds() const;
     std::optional<table_id> get_id() const;
     bool get_synchronous_updates_flag() const;
+    std::optional<db::tablet_options::map_type> get_tablet_options() const;
 
     void apply_to_builder(schema_builder& builder, schema::extensions_map schema_extensions, const data_dictionary::database& db, sstring ks_name) const;
     void validate_minimum_int(const sstring& field, int32_t minimum_value, int32_t default_value) const;

data_dictionary/data_dictionary.cc

@@ -55,6 +55,11 @@ keyspace::is_internal() const {
     return _ops->is_internal(*this);
 }
 
+bool
+keyspace::uses_tablets() const {
+    return metadata()->uses_tablets();
+}
+
 const locator::abstract_replication_strategy&
 keyspace::get_replication_strategy() const {
     return _ops->get_replication_strategy(*this);

data_dictionary/data_dictionary.hh

@@ -87,6 +87,7 @@ private:
     keyspace(const impl* ops, const void* keyspace);
 public:
     bool is_internal() const;
+    bool uses_tablets() const;
     lw_shared_ptr<keyspace_metadata> metadata() const;
     const user_types_metadata& user_types() const;
     const locator::abstract_replication_strategy& get_replication_strategy() const;

data_dictionary/keyspace_metadata.hh

@@ -65,6 +65,9 @@ public:
     std::optional<unsigned> initial_tablets() const {
         return _initial_tablets;
     }
+    bool uses_tablets() const noexcept {
+        return _initial_tablets.has_value();
+    }
     const std::unordered_map<sstring, schema_ptr>& cf_meta_data() const {
         return _cf_meta_data;
     }

db/CMakeLists.txt

@@ -38,7 +38,8 @@ target_sources(db
     snapshot/backup_task.cc
     rate_limiter.cc
     per_partition_rate_limit_options.cc
-    row_cache.cc)
+    row_cache.cc,
+    tablet_options.cc)
 target_include_directories(db
   PUBLIC
     ${CMAKE_SOURCE_DIR})

db/schema_tables.cc

@@ -336,6 +336,11 @@ schema_ptr scylla_tables(schema_features features) {
             // In this case, for non-system tables, `version` is null and `schema::version()` will be a hash.
             sb.with_column("committed_by_group0", boolean_type);
         }
+
+        // It is safe to add the `tablets` column unconditionally,
+        // since it is written to only after the cluster feature is enabled.
+        sb.with_column("tablets", map_type_impl::get_instance(utf8_type, utf8_type, false));
+
         sb.with_hash_version();
         s = sb.build();
     }
@@ -1733,6 +1738,19 @@ mutation make_scylla_tables_mutation(schema_ptr table, api::timestamp_type times
         auto& cdef = *scylla_tables()->get_column_definition("partitioner");
         m.set_clustered_cell(ckey, cdef, atomic_cell::make_dead(timestamp, gc_clock::now()));
     }
+    // A table will have engaged tablet options
+    // only after they were set by CREATE TABLE or ALTER TABLE,
+    // Meaning the cluster feature is enabled, so it is safe to write
+    // to this columns.
+    if (table->has_tablet_options()) {
+        auto& map = table->raw_tablet_options();
+        auto& cdef = *scylla_tables()->get_column_definition("tablets");
+        if (map.empty()) {
+            m.set_clustered_cell(ckey, cdef, atomic_cell::make_dead(timestamp, gc_clock::now()));
+        } else {
+            m.set_clustered_cell(ckey, cdef, make_map_mutation(map, cdef, timestamp));
+        }
+    }
     // In-memory tables are deprecated since scylla-2024.1.0
     // FIXME: delete the column when there's no live version supporting it anymore.
     // Writing it here breaks upgrade rollback to versions that do not support the in_memory schema_feature
@@ -2154,6 +2172,19 @@ static void prepare_builder_from_table_row(const schema_ctxt& ctxt, schema_build
     }
 }
 
+static void prepare_builder_from_scylla_tables_row(const schema_ctxt& ctxt, schema_builder& builder, const query::result_set_row& table_row) {
+    auto in_mem = table_row.get<bool>("in_memory");
+    auto in_mem_enabled = in_mem.value_or(false);
+    if (in_mem_enabled) {
+        slogger.warn("Support for in_memory tables has been deprecated.");
+    }
+    builder.set_in_memory(in_mem_enabled);
+    if (auto opt_map = get_map<sstring, sstring>(table_row, "tablets")) {
+        auto tablet_options = db::tablet_options(*opt_map);
+        builder.set_tablet_options(tablet_options.to_map());
+    }
+}
+
 schema_ptr create_table_from_mutations(const schema_ctxt& ctxt, schema_mutations sm, std::optional<table_schema_version> version)
 {
     slogger.trace("create_table_from_mutations: version={}, {}", version, sm);
@@ -2208,13 +2239,7 @@ schema_ptr create_table_from_mutations(const schema_ctxt& ctxt, schema_mutations
     if (sm.scylla_tables()) {
         table_rs = query::result_set(*sm.scylla_tables());
         if (!table_rs.empty()) {
-            query::result_set_row table_row = table_rs.row(0);
-            auto in_mem = table_row.get<bool>("in_memory");
-            auto in_mem_enabled = in_mem.value_or(false);
-            if (in_mem_enabled) {
-                slogger.warn("Support for in_memory tables has been deprecated.");
-            }
-            builder.set_in_memory(in_mem_enabled);
+            prepare_builder_from_scylla_tables_row(ctxt, builder, table_rs.row(0));
         }
     }
     v3_columns columns(std::move(column_defs), is_dense, is_compound);
@@ -2445,6 +2470,13 @@ view_ptr create_view_from_mutations(const schema_ctxt& ctxt, schema_mutations sm
     schema_builder builder{ks_name, cf_name, id};
     prepare_builder_from_table_row(ctxt, builder, row);
 
+    if (sm.scylla_tables()) {
+        table_rs = query::result_set(*sm.scylla_tables());
+        if (!table_rs.empty()) {
+            prepare_builder_from_scylla_tables_row(ctxt, builder, table_rs.row(0));
+        }
+    }
+
     auto computed_columns = get_computed_columns(sm);
     auto column_defs = create_columns_from_column_rows(ctxt, query::result_set(sm.columns_mutation()), ks_name, cf_name, false, column_view_virtual::no, computed_columns);
     for (auto&& cdef : column_defs) {

db/tablet_options.cc

@@ -0,0 +1,96 @@
+/*
+ * Copyright 2025-present ScyllaDB
+ */
+/*
+ * SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+ */
+
+#include <cstdlib>
+
+#include "exceptions/exceptions.hh"
+#include "db/tablet_options.hh"
+#include "utils/log.hh"
+
+extern logging::logger dblog;
+
+namespace db {
+
+tablet_options::tablet_options(const map_type& map) {
+    for (auto& [key, value_str] : map) {
+        switch (tablet_options::from_string(key)) {
+        case tablet_option_type::min_tablet_count:
+            if (auto value = std::atol(value_str.c_str())) {
+                min_tablet_count.emplace(value);
+            }
+            break;
+        case tablet_option_type::min_per_shard_tablet_count:
+            if (auto value = std::atof(value_str.c_str())) {
+                min_per_shard_tablet_count.emplace(value);
+            }
+            break;
+        case tablet_option_type::expected_data_size_in_gb:
+            if (auto value = std::atol(value_str.c_str())) {
+                expected_data_size_in_gb.emplace(value);
+            }
+            break;
+        }
+    }
+}
+
+sstring tablet_options::to_string(tablet_option_type hint) {
+    switch (hint) {
+    case tablet_option_type::min_tablet_count: return "min_tablet_count";
+    case tablet_option_type::min_per_shard_tablet_count: return "min_per_shard_tablet_count";
+    case tablet_option_type::expected_data_size_in_gb: return "expected_data_size_in_gb";
+    }
+}
+
+tablet_option_type tablet_options::from_string(sstring hint_desc) {
+    if (hint_desc == "min_tablet_count") {
+        return tablet_option_type::min_tablet_count;
+    } else if (hint_desc == "min_per_shard_tablet_count") {
+        return tablet_option_type::min_per_shard_tablet_count;
+    } else if (hint_desc == "expected_data_size_in_gb") {
+        return tablet_option_type::expected_data_size_in_gb;
+    } else {
+        throw exceptions::syntax_exception(fmt::format("Unknown tablet hint '{}'", hint_desc));
+    }
+}
+
+std::map<sstring, sstring> tablet_options::to_map() const {
+    std::map<sstring, sstring> res;
+    if (min_tablet_count) {
+        res[to_string(tablet_option_type::min_tablet_count)] = fmt::to_string(*min_tablet_count);
+    }
+    if (min_per_shard_tablet_count) {
+        res[to_string(tablet_option_type::min_per_shard_tablet_count)] = fmt::to_string(*min_per_shard_tablet_count);
+    }
+    if (expected_data_size_in_gb) {
+        res[to_string(tablet_option_type::expected_data_size_in_gb)] = fmt::to_string(*expected_data_size_in_gb);
+    }
+    return res;
+}
+
+void tablet_options::validate(const map_type& map) {
+    for (auto& [key, value_str] : map) {
+        switch (tablet_options::from_string(key)) {
+        case tablet_option_type::min_tablet_count:
+            if (auto value = std::atol(value_str.c_str()); value < 0) {
+                throw exceptions::configuration_exception(format("Invalid value '{}' for min_tablet_count", value));
+            }
+            break;
+        case tablet_option_type::min_per_shard_tablet_count:
+            if (auto value = std::atof(value_str.c_str()); value < 0) {
+                throw exceptions::configuration_exception(format("Invalid value '{}' for min_per_shard_tablet_count", value));
+            }
+            break;
+        case tablet_option_type::expected_data_size_in_gb:
+            if (auto value = std::atol(value_str.c_str()); value < 0) {
+                throw exceptions::configuration_exception(format("Invalid value '{}' for expected_data_size_in_gb", value));
+            }
+            break;
+        }
+    }
+}
+
+} // namespace db