test: Verify partitioned set store split and unsplit correctly

Signed-off-by: Raphael S. Carvalho <raphaelsc@scylladb.com>

sstables/sstable_set.cc

@@ -270,7 +270,21 @@ bool partitioned_sstable_set::store_as_unleveled(const shared_sstable& sst) cons
     // since many of such sstables would have presence on almost all intervals.
     static constexpr float unleveled_threshold = 0.85f;
     auto sst_tr = dht::token_range(sst->get_first_decorated_key().token(), sst->get_last_decorated_key().token());
-    return dht::overlap_ratio(_token_range, sst_tr) >= unleveled_threshold;
+    bool as_unleveled = dht::overlap_ratio(_token_range, sst_tr) >= unleveled_threshold;
+
+    utils::get_local_injector().inject("sstable_set_insertion_verification", [&] () {
+        auto& i = utils::get_local_injector();
+        auto table_name = i.inject_parameter<std::string_view>("sstable_set_insertion_verification", "table").value();
+        bool expect_unleveled = i.inject_parameter<int>("sstable_set_insertion_verification", "expect_unleveled").value();
+        if (_schema->cf_name() != table_name) {
+            return;
+        }
+        sstlog.info("SSTable {}, as_unleveled={}, expect_unleveled={}, sst_tr={}, overlap_ratio={}",
+            sst->generation(), as_unleveled, expect_unleveled, sst_tr, dht::overlap_ratio(_token_range, sst_tr));
+        SCYLLA_ASSERT(as_unleveled == expect_unleveled);
+    });
+
+    return as_unleveled;
 }
 
 dht::ring_position partitioned_sstable_set::to_ring_position(const dht::compatible_ring_position_or_view& crp) {

test/cluster/test_sstable_set.py

@@ -0,0 +1,70 @@
+# Copyright (C) 2025-present ScyllaDB
+#
+# SPDX-License-Identifier: LicenseRef-ScyllaDB-Source-Available-1.0
+#
+
+import asyncio
+import pytest
+import time
+import logging
+from test.pylib.manager_client import ManagerClient
+from test.pylib.util import wait_for_cql_and_get_hosts
+from test.cluster.util import create_new_test_keyspace
+from test.cluster.conftest import skip_mode
+
+logger = logging.getLogger(__name__)
+
+@pytest.mark.parametrize("mode", ['vnode', 'tablet'])
+@skip_mode('release', 'error injections are not supported in release mode')
+@pytest.mark.asyncio
+async def test_partitioned_sstable_set(manager: ManagerClient, mode):
+    cfg = {
+        'tablets_mode_for_new_keyspaces': 'enabled',
+    }
+
+    cmdline = ['--smp=1']
+    server = await manager.server_add(config=cfg, cmdline=cmdline)
+    await manager.api.disable_tablet_balancing(server.ip_addr)
+
+    cql = manager.get_cql()
+    await wait_for_cql_and_get_hosts(cql, [server], time.time() + 60)
+
+    if mode == 'tablet':
+        ks = await create_new_test_keyspace(cql, "WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 1} AND tablets = {'initial': 4};")
+    else:
+        ks = await create_new_test_keyspace(cql, "WITH replication = {'class': 'NetworkTopologyStrategy', 'replication_factor': 1} AND tablets = {'enabled': 'false'};")
+
+    cql.execute(f"""CREATE TABLE {ks}.test (pk int PRIMARY KEY, c int) WITH compaction = {{
+        'class' : 'IncrementalCompactionStrategy',
+        'sstable_size_in_mb' : '0'
+    }}""")
+
+    await manager.api.disable_autocompaction(server.ip_addr, ks)
+
+    keys = range(100)
+    await asyncio.gather(*[cql.run_async(f"INSERT INTO {ks}.test (pk, c) VALUES ({k}, {k});") for k in keys])
+
+    # Expect flushed sstables to be stored as unleveled
+    await manager.api.enable_injection(server.ip_addr, 'sstable_set_insertion_verification',
+                                       one_shot=False,
+                                       parameters={'table': 'test', 'expect_unleveled': '1'})
+
+    logger.info("Verifying unsplit sstables are stored correctly in the set")
+
+    await manager.api.keyspace_flush(server.ip_addr, ks, "test")
+
+    await manager.api.disable_injection(server.ip_addr, 'sstable_set_insertion_verification')
+
+    # Split all sstables without verification since they're compacted incrementally
+    await manager.api.keyspace_compaction(server.ip_addr, ks, "test")
+
+    # Expect all split sstables are stored as leveled
+    await manager.api.enable_injection(server.ip_addr, 'sstable_set_insertion_verification',
+                                       one_shot=False,
+                                       parameters={'table': 'test', 'expect_unleveled': '0'})
+
+    logger.info("Verifying split sstables are stored correctly in the set")
+
+    await manager.api.keyspace_compaction(server.ip_addr, ks, "test")
+
+    await cql.run_async(f"DROP KEYSPACE {ks}")

utils/error_injection.hh

@@ -472,12 +472,18 @@ public:
     }
 
     template <typename T = std::string_view>
-    std::optional<T> inject_parameter(const std::string_view& name) {
+    std::optional<T> inject_parameter(const std::string_view& name, const std::string_view param_name) {
         auto* data = get_data(name);
         if (!data) {
             return std::nullopt;
         }
-        return data->shared_data->template get<T>("value");
+        return data->shared_data->template get<T>(std::string(param_name));
+    }
+
+    template <typename T = std::string_view>
+    [[gnu::always_inline]]
+    std::optional<T> inject_parameter(const std::string_view& name) {
+         return inject_parameter<T>(name, "value");
     }
 
     // \brief Export the value of the parameter with the given name
@@ -620,6 +626,12 @@ public:
         return make_ready_future<>();
     }
 
+    template <typename T>
+    [[gnu::always_inline]]
+    std::optional<T> inject_parameter(const std::string_view& name, const std::string_view param_name) {
+        return std::nullopt;
+    }
+
     template <typename T>
     [[gnu::always_inline]]
     std::optional<T> inject_parameter(const std::string_view& name) {