From beaaf00facbac05abf1d61d740d92bdde5b29b07 Mon Sep 17 00:00:00 2001
From: "Raphael S. Carvalho" <raphaelsc@scylladb.com>
Date: Mon, 23 Jun 2025 16:31:13 -0300
Subject: [PATCH] test: Add test that compaction doesn't cross logical group
 boundary

Signed-off-by: Raphael S. Carvalho <raphaelsc@scylladb.com>
---
 replica/compaction_group.hh         |  4 ++
 test/boost/compaction_group_test.cc | 67 +++++++++++++++++++++++++++++
 test/lib/test_services.cc           |  6 +++
 test/lib/test_services.hh           |  1 +
 4 files changed, 78 insertions(+)

diff --git a/replica/compaction_group.hh b/replica/compaction_group.hh
index 7112ab9ba8..1502e39a18 100644
--- a/replica/compaction_group.hh
+++ b/replica/compaction_group.hh
@@ -242,6 +242,10 @@ public:
 
     future<> split(sstables::compaction_type_options::split opt, tasks::task_info tablet_split_task_info);
 
+    void set_repair_sstable_classifier(repair_classifier_func repair_sstable_classifier) {
+        _repair_sstable_classifier = std::move(repair_sstable_classifier);
+    }
+
     friend class storage_group;
 };
 
diff --git a/test/boost/compaction_group_test.cc b/test/boost/compaction_group_test.cc
index 568902c0e7..97e84e6dd1 100644
--- a/test/boost/compaction_group_test.cc
+++ b/test/boost/compaction_group_test.cc
@@ -202,3 +202,70 @@ SEASTAR_TEST_CASE(basic_compaction_group_splitting_test) {
         }
     });
 }
+
+static mutation_reader sstable_reader(shared_sstable sst, schema_ptr s, reader_permit permit) {
+    return sst->as_mutation_source().make_mutation_reader(s, std::move(permit), query::full_partition_range, s->full_slice());
+}
+
+SEASTAR_TEST_CASE(compactions_dont_cross_group_boundary_test) {
+    return test_env::do_with_async([] (test_env& env) {
+        auto builder = schema_builder("tests", "compactions_dont_cross_group_boundary")
+                .with_column("id", utf8_type, column_kind::partition_key)
+                .with_column("cl", int32_type, column_kind::clustering_key)
+                .with_column("value", int32_type);
+        auto s = builder.build();
+
+        auto t = env.make_table_for_tests(s);
+        auto close_table = deferred_stop(t);
+        t->start();
+
+        // Disable auto compaction to allow us to trigger compaction manually later.
+        t->disable_auto_compaction().get();
+
+        auto is_unrepaired = [] (dht::token t) { return t.raw() % 3 == 0; };
+        auto is_repairing = [] (dht::token t) { return t.raw() % 3 == 1; };
+        auto is_repaired = [] (dht::token t) { return t.raw() % 3 == 2; };
+
+        auto sst_factory = env.make_sst_factory(s);
+        auto generate_sstables = [&] (std::function<bool(dht::token)> filter) {
+            for (int i = 0; i < 4; i++) {
+                t->add_sstable_and_update_cache(generate_sstable(s, sst_factory, filter)).get();
+            }
+        };
+        generate_sstables(is_unrepaired);
+        generate_sstables(is_repairing);
+        generate_sstables(is_repaired);
+
+        auto repair_token_classifier = [&] (dht::token t) -> replica::repair_sstable_classification {
+            if (is_unrepaired(t)) {
+                return replica::repair_sstable_classification::unrepaired;
+            } else if (is_repairing(t)) {
+                return replica::repair_sstable_classification::repairing;
+            }
+            return replica::repair_sstable_classification::repaired;
+        };
+        auto repair_sstable_classifier = [&] (const sstables::shared_sstable& sst) -> replica::repair_sstable_classification {
+            return repair_token_classifier(sst->get_first_decorated_key().token());
+        };
+        t.set_repair_sstable_classifier(repair_sstable_classifier);
+
+        for (int i = 0; i < 4; i++) {
+            t->compact_all_sstables({}).get();
+        }
+
+        auto validate_sstable = [&] (const sstables::shared_sstable& sst) {
+            auto reader = sstable_reader(sst, s, env.make_reader_permit()); // reader holds sst and s alive.
+            auto close_reader = deferred_close(reader);
+
+            auto expected_classification = repair_sstable_classifier(sst);
+
+            while (auto m = read_mutation_from_mutation_reader(reader).get()) {
+                BOOST_REQUIRE(repair_token_classifier(m->decorated_key().token()) == expected_classification);
+            }
+        };
+        auto all_sstables = t->get_sstables();
+        for (auto& sst : *all_sstables) {
+            validate_sstable(sst);
+        }
+    });
+}
diff --git a/test/lib/test_services.cc b/test/lib/test_services.cc
index 37a067581a..88c6bb582f 100644
--- a/test/lib/test_services.cc
+++ b/test/lib/test_services.cc
@@ -170,6 +170,12 @@ void table_for_tests::set_tombstone_gc_enabled(bool tombstone_gc_enabled) noexce
     _data->cf->set_tombstone_gc_enabled(tombstone_gc_enabled);
 }
 
+void table_for_tests::set_repair_sstable_classifier(replica::repair_classifier_func repair_sstable_classifier) {
+    _data->cf->for_each_compaction_group([&] (replica::compaction_group& cg) {
+        cg.set_repair_sstable_classifier(repair_sstable_classifier);
+    });
+}
+
 namespace sstables {
 
 std::vector<db::object_storage_endpoint_param> make_storage_options_config(const data_dictionary::storage_options& so) {
diff --git a/test/lib/test_services.hh b/test/lib/test_services.hh
index 836dffbdfb..c8327d29a2 100644
--- a/test/lib/test_services.hh
+++ b/test/lib/test_services.hh
@@ -67,4 +67,5 @@ struct table_for_tests {
     future<> stop();
 
     void set_tombstone_gc_enabled(bool tombstone_gc_enabled) noexcept;
+    void set_repair_sstable_classifier(replica::repair_classifier_func repair_sstable_classifier);
 };