range_tombstone_accumulator: drop _reversed flag

db/view/view.hh

@@ -192,8 +192,8 @@ public:
             , _view_updates(std::move(views_to_update))
             , _updates(std::move(updates))
             , _existings(std::move(existings))
-            , _update_tombstone_tracker(*_schema, false)
+            , _update_tombstone_tracker(*_schema)
-            , _existing_tombstone_tracker(*_schema, false)
+            , _existing_tombstone_tracker(*_schema)
             , _now(now) {
     }
     view_update_builder(view_update_builder&& other) noexcept = default;

mutation_compactor.hh

@@ -224,7 +224,7 @@ public:
         , _row_limit(limit)
         , _partition_limit(partition_limit)
         , _partition_row_limit(_slice.options.contains(query::partition_slice::option::distinct) ? 1 : slice.partition_row_limit())
-        , _range_tombstones(s, false)
+        , _range_tombstones(s)
         , _last_dk({dht::token(), partition_key::make_empty()})
     {
         static_assert(!sstable_compaction(), "This constructor cannot be used for sstable compaction.");
@@ -238,7 +238,7 @@ public:
         , _get_max_purgeable(std::move(get_max_purgeable))
         , _can_gc([this] (tombstone t) { return can_gc(t); })
         , _slice(s.full_slice())
-        , _range_tombstones(s, false)
+        , _range_tombstones(s)
         , _last_dk({dht::token(), partition_key::make_empty()})
         , _collector(std::make_unique<mutation_compactor_garbage_collector>(_schema))
     {

range_tombstone.cc

@@ -73,10 +73,6 @@ void range_tombstone_accumulator::update_current_tombstone() {
 
 void range_tombstone_accumulator::drop_unneeded_tombstones(const clustering_key_prefix& ck, int w) {
     auto cmp = [&] (const range_tombstone& rt, const clustering_key_prefix& ck, int w) {
-        if (_reversed) {
-            auto bv = rt.start_bound();
-            return _cmp(ck, w, bv.prefix(), weight(bv.kind()));
-        }
         auto bv = rt.end_bound();
         return _cmp(bv.prefix(), weight(bv.kind()), ck, w);
     };
@@ -91,15 +87,11 @@ void range_tombstone_accumulator::drop_unneeded_tombstones(const clustering_key_
 }
 
 void range_tombstone_accumulator::apply(range_tombstone rt) {
-    if (_reversed) {
+    drop_unneeded_tombstones(rt.start, weight(rt.start_kind));
-        drop_unneeded_tombstones(rt.end, weight(rt.end_kind));
-    } else {
-        drop_unneeded_tombstones(rt.start, weight(rt.start_kind));
-    }
     _current_tombstone.apply(rt.tomb);
 
     auto cmp = [&] (const range_tombstone& rt1, const range_tombstone& rt2) {
-        return _reversed ? _cmp(rt2.start_bound(), rt1.start_bound()) : _cmp(rt1.end_bound(), rt2.end_bound());
+        return _cmp(rt1.end_bound(), rt2.end_bound());
     };
     _range_tombstones.insert(boost::upper_bound(_range_tombstones, rt, cmp), std::move(rt));
 }

range_tombstone.hh

@@ -261,13 +261,12 @@ class range_tombstone_accumulator {
     tombstone _partition_tombstone;
     std::deque<range_tombstone> _range_tombstones;
     tombstone _current_tombstone;
-    bool _reversed;
 private:
     void update_current_tombstone();
     void drop_unneeded_tombstones(const clustering_key_prefix& ck, int w = 0);
 public:
-    range_tombstone_accumulator(const schema& s, bool reversed)
+    explicit range_tombstone_accumulator(const schema& s)
-        : _cmp(s), _reversed(reversed) { }
+        : _cmp(s) { }
 
     void set_partition_tombstone(tombstone t) {
         _partition_tombstone = t;

test/boost/mutation_test.cc

@@ -2889,7 +2889,7 @@ void check_clustering_row_summaries(const schema& schema, const clustering_row_s
 }
 
 void check_clustering_summaries(const schema& schema, const partition_summary& actual, const partition_summary& expected) {
-    range_tombstone_accumulator range_tombstones(schema, false);
+    range_tombstone_accumulator range_tombstones(schema);
     range_tombstones.set_partition_tombstone(expected.tomb);
 
     for (auto [actual_frag, expected_frag] : iterate_over_in_ordered_lockstep(actual.clustering_fragments, expected.clustering_fragments,

test/boost/range_tombstone_list_test.cc

@@ -884,8 +884,7 @@ BOOST_AUTO_TEST_CASE(test_accumulator) {
     auto ts1 = 1;
     auto ts2 = 2;
 
-    testlog.info("Forward");
+    auto acc = range_tombstone_accumulator(*s);
-    auto acc = range_tombstone_accumulator(*s, false);
     acc.apply(rtie(0, 4, ts1));
     BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 0 })), tombstone(ts1, gc_now));
     acc.apply(rtie(1, 2, ts2));
@@ -904,26 +903,4 @@ BOOST_AUTO_TEST_CASE(test_accumulator) {
     BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 13 })), tombstone(ts2, gc_now));
     BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 14 })), tombstone());
     BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 15 })), tombstone());
-
-    testlog.info("Reversed");
-    acc = range_tombstone_accumulator(*s, true);
-
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 15 })), tombstone());
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 14 })), tombstone());
-    acc.apply(rtie(11, 14, ts2));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 13 })), tombstone(ts2, gc_now));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 12 })), tombstone(ts2, gc_now));
-    acc.apply(rtie(10, 12, ts1));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 11 })), tombstone(ts2, gc_now));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 10 })), tombstone(ts1, gc_now));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 9 })), tombstone());
-    acc.apply(rtie(6, 8, ts2));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 5 })), tombstone());
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 4 })), tombstone());
-    acc.apply(rtie(0, 4, ts1));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 3 })), tombstone(ts1, gc_now));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 2 })), tombstone(ts1, gc_now));
-    acc.apply(rtie(1, 2, ts2));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 1 })), tombstone(ts2, gc_now));
-    BOOST_REQUIRE_EQUAL(acc.tombstone_for_row(key({ 0 })), tombstone(ts1, gc_now));
 }