Merge "Support reading range tombstones" from Piotr and Vladimir

Implement and test support for reading range tombstones in SSTables 3.

Does not yet support reads which are using slicing or fast forwarding.

From github.com/scylladb/seastar-dev.git haaawk/sstables3/tombstones_v11:

Piotr Jastrzebski (5):
  sstables: Add consumer_m::consume_range_tombstone
  sstables: Support null columns in ck
  sstables: Support reading range_tombstones
  sstables: Test reading range_tombstones
  sstables: Add test for RT with non-full key

Vladimir Krivopalov (2):
  sstables: Add operator<< overload for bound_kind_m.
  keys: Add clustering_key_prefix::make_full helper.

keys.hh

@@ -804,6 +804,23 @@ public:
         return from_exploded(s, prefix.components());
     }
 
+    /* This function makes the passed clustering key full by filling its
+     * missing trailing components with empty values.
+     * This is used to represesent clustering keys of rows in compact tables that may be non-full.
+     * Returns whether a key wasn't full before the call.
+     */
+    static bool make_full(const schema& s, clustering_key_prefix& ck) {
+        if (!ck.is_full(s)) {
+            // TODO: avoid unnecessary copy here
+            auto full_ck_size = s.clustering_key_columns().size();
+            auto exploded = ck.explode(s);
+            exploded.resize(full_ck_size);
+            ck = clustering_key_prefix::from_exploded(std::move(exploded));
+            return true;
+        }
+        return false;
+    }
+
     friend std::ostream& operator<<(std::ostream& out, const clustering_key_prefix& ckp);
 };
 

sstables/m_format_read_helpers.hh

@@ -162,3 +162,35 @@ inline gc_clock::time_point parse_expiry(const serialization_header& header,
 }
 
 };   // namespace sstables
+
+inline std::ostream& operator<<(std::ostream& out, sstables::bound_kind_m kind) {
+    switch (kind) {
+    case sstables::bound_kind_m::excl_end:
+        out << "excl_end";
+        break;
+    case sstables::bound_kind_m::incl_start:
+        out << "incl_start";
+        break;
+    case sstables::bound_kind_m::excl_end_incl_start:
+        out << "excl_end_incl_start";
+        break;
+    case sstables::bound_kind_m::static_clustering:
+        out << "static_clustering";
+        break;
+    case sstables::bound_kind_m::clustering:
+        out << "clustering";
+        break;
+    case sstables::bound_kind_m::incl_end_excl_start:
+        out << "incl_end_excl_start";
+        break;
+    case sstables::bound_kind_m::incl_end:
+        out << "incl_end";
+        break;
+    case sstables::bound_kind_m::excl_start:
+        out << "excl_start";
+        break;
+    default:
+        out << static_cast<std::underlying_type_t<sstables::bound_kind_m>>(kind);
+    }
+    return out;
+}

sstables/mp_row_consumer.hh

@@ -34,6 +34,9 @@
 #include "liveness_info.hh"
 #include "mutation_fragment_filter.hh"
 #include "types.hh"
+#include "keys.hh"
+#include "clustering_bounds_comparator.hh"
+#include "range_tombstone.hh"
 
 namespace sstables {
 
@@ -815,6 +818,53 @@ class mp_row_consumer_m : public consumer_m {
     std::vector<cell> _cells;
     collection_type_impl::mutation _cm;
 
+    struct range_tombstone_start {
+        clustering_key_prefix ck;
+        bound_kind kind;
+        tombstone tomb;
+    };
+
+    inline friend std::ostream& operator<<(std::ostream& o, const sstables::mp_row_consumer_m::range_tombstone_start& rt_start) {
+        o << "{ clustering: " << rt_start.ck
+          << ", kind: " << rt_start.kind
+          << ", tombstone: " << rt_start.tomb << " }";
+        return o;
+    }
+
+    std::optional<range_tombstone_start> _opened_range_tombstone;
+
+    void consume_range_tombstone_start(clustering_key_prefix ck, bound_kind k, tombstone t) {
+        if (_fwd) {
+            // TODO: support fast-forwarding and filtering for range tombstones
+            assert(0 && "Fast forwarding of range tombstones not supported.");
+        }
+        if (_opened_range_tombstone) {
+            throw sstables::malformed_sstable_exception(
+                    format("Range tombstones have to be disjoint: current opened range tombstone {}, new tombstone {}",
+                           *_opened_range_tombstone, t));
+        }
+        _opened_range_tombstone = {std::move(ck), k, std::move(t)};
+    }
+
+    void consume_range_tombstone_end(clustering_key_prefix ck, bound_kind k, tombstone t) {
+        if (!_opened_range_tombstone) {
+            throw sstables::malformed_sstable_exception(
+                    format("Closing range tombstone that wasn't opened: clustering {}, kind {}, tombstone {}",
+                           ck, k, t));
+        }
+        if (_opened_range_tombstone->tomb.compare(t) != 0) {
+            throw sstables::malformed_sstable_exception(
+                    format("Range tombstone with ck {} and two different tombstones at ends: {}, {}",
+                           ck, _opened_range_tombstone->tomb, t));
+        }
+        _reader->push_mutation_fragment(range_tombstone(std::move(_opened_range_tombstone->ck),
+                                                        _opened_range_tombstone->kind,
+                                                        std::move(ck),
+                                                        k,
+                                                        std::move(t)));
+        _opened_range_tombstone = {};
+    }
+
     const column_definition& get_column_definition(std::optional<column_id> column_id) {
         auto column_type = _inside_static_row ? column_kind::static_column : column_kind::regular_column;
         return _schema->column_at(column_type, *column_id);
@@ -825,14 +875,23 @@ class mp_row_consumer_m : public consumer_m {
         auto action = _mf_filter->apply(*_in_progress_row);
         switch (action) {
         case mutation_fragment_filter::result::emit:
+            if (_opened_range_tombstone) {
+                auto ck = _in_progress_row->key();
+                auto end_kind = clustering_key::make_full(*_schema, ck) ? bound_kind::excl_end : bound_kind::incl_end;
+                _reader->push_mutation_fragment(range_tombstone(std::move(_opened_range_tombstone->ck),
+                                                                _opened_range_tombstone->kind,
+                                                                ck,
+                                                                end_kind,
+                                                                _opened_range_tombstone->tomb));
+                _opened_range_tombstone->ck = std::move(ck);
+                _opened_range_tombstone->kind = bound_kind::incl_start;
+            }
             _reader->push_mutation_fragment(*std::exchange(_in_progress_row, {}));
             break;
         case mutation_fragment_filter::result::ignore:
-            {
-                _in_progress_row.reset();
-                if (_mf_filter->is_current_range_changed()) {
-                    return proceed::no;
-                }
+            _in_progress_row.reset();
+            if (_mf_filter->is_current_range_changed()) {
+                return proceed::no;
             }
             break;
         case mutation_fragment_filter::result::store_and_finish:
@@ -1015,6 +1074,42 @@ public:
         return proceed::yes;
     }
 
+    virtual proceed consume_range_tombstone(const std::vector<temporary_buffer<char>>& ecp,
+                                            bound_kind kind,
+                                            tombstone tomb) override {
+        auto ck = clustering_key_prefix::from_range(ecp | boost::adaptors::transformed(
+            [] (const temporary_buffer<char>& b) { return to_bytes_view(b); }));
+        if (kind == bound_kind::incl_start || kind == bound_kind::excl_start) {
+            consume_range_tombstone_start(std::move(ck), kind, std::move(tomb));
+        } else { // *_end kind
+            consume_range_tombstone_end(std::move(ck), kind, std::move(tomb));
+        }
+        return proceed(!_reader->is_buffer_full());
+    }
+
+    virtual proceed consume_range_tombstone(const std::vector<temporary_buffer<char>>& ecp,
+                                            sstables::bound_kind_m kind,
+                                            tombstone end_tombstone,
+                                            tombstone start_tombstone) override {
+        auto ck = clustering_key_prefix::from_range(ecp | boost::adaptors::transformed(
+            [] (const temporary_buffer<char>& b) { return to_bytes_view(b); }));
+        switch (kind) {
+        case bound_kind_m::incl_end_excl_start:
+            consume_range_tombstone_end(ck, bound_kind::incl_end, std::move(end_tombstone));
+            consume_range_tombstone_start(std::move(ck), bound_kind::excl_start, std::move(start_tombstone));
+            break;
+        case bound_kind_m::excl_end_incl_start:
+            consume_range_tombstone_end(ck, bound_kind::excl_end, std::move(end_tombstone));
+            consume_range_tombstone_start(std::move(ck), bound_kind::incl_start, std::move(start_tombstone));
+            break;
+        default:
+            throw sstables::malformed_sstable_exception(
+                    format("Corrupted range tombstone: invalid boundary type {}", kind));
+        }
+
+        return proceed(!_reader->is_buffer_full());
+    }
+
     virtual proceed consume_row_end() override {
         auto fill_cells = [this] (column_kind kind, row& cells) {
             auto max_id = boost::max_element(_cells, [](auto &&a, auto &&b) {
@@ -1054,6 +1149,9 @@ public:
 
     virtual proceed consume_partition_end() override {
         sstlog.trace("mp_row_consumer_m {}: consume_partition_end()", this);
+        if (_opened_range_tombstone) {
+            throw sstables::malformed_sstable_exception("Unclosed range tombstone.");
+        }
         _is_mutation_end = true;
         _in_progress_row.reset();
         _mf_filter.reset();

sstables/mutation_fragment_filter.hh

@@ -134,6 +134,10 @@ public:
         return (_last_lower_bound_counter != _walker.lower_bound_change_counter());
     }
 
+
+    position_in_partition_view lower_bound() const {
+        return _walker.lower_bound();
+    }
 };
 
 };   // namespace sstables

sstables/row.hh

@@ -42,6 +42,7 @@
 
 #include "sstables.hh"
 #include "tombstone.hh"
+#include "m_format_read_helpers.hh"
 
 // sstables::data_consume_row feeds the contents of a single row into a
 // row_consumer object:
@@ -175,6 +176,15 @@ public:
     virtual proceed consume_counter_column(std::optional<column_id> column_id, bytes_view value,
                                            api::timestamp_type timestamp) = 0;
 
+    virtual proceed consume_range_tombstone(const std::vector<temporary_buffer<char>>& ecp,
+                                            bound_kind kind,
+                                            tombstone tomb) = 0;
+
+    virtual proceed consume_range_tombstone(const std::vector<temporary_buffer<char>>& ecp,
+                                            sstables::bound_kind_m,
+                                            tombstone end_tombstone,
+                                            tombstone start_tombstone) = 0;
+
     virtual proceed consume_row_end() = 0;
 
     // Called when the reader is fast forwarded to given element.
@@ -579,6 +589,16 @@ private:
         COLUMN_VALUE,
         COLUMN_END,
         RANGE_TOMBSTONE_MARKER,
+        RANGE_TOMBSTONE_KIND,
+        RANGE_TOMBSTONE_SIZE,
+        RANGE_TOMBSTONE_CONSUME_CK,
+        RANGE_TOMBSTONE_BODY,
+        RANGE_TOMBSTONE_BODY_SIZE,
+        RANGE_TOMBSTONE_BODY_PREV_SIZE,
+        RANGE_TOMBSTONE_BODY_TIMESTAMP,
+        RANGE_TOMBSTONE_BODY_TIMESTAMP2,
+        RANGE_TOMBSTONE_BODY_LOCAL_DELTIME,
+        RANGE_TOMBSTONE_BODY_LOCAL_DELTIME2,
     } _state = state::PARTITION_START;
 
     consumer_m& _consumer;
@@ -614,10 +634,22 @@ private:
     temporary_buffer<char> _cell_path;
     uint64_t _ck_blocks_header;
     uint32_t _ck_blocks_header_offset;
+    bool _null_component_occured;
     uint64_t _subcolumns_to_read = 0;
     api::timestamp_type _complex_column_marked_for_delete;
     tombstone _complex_column_tombstone;
-
+    bool _reading_range_tombstone_ck = false;
+    bound_kind_m _range_tombstone_kind;
+    uint16_t _ck_size;
+    /*
+     * We need two range tombstones because range tombstone marker can be either a single bound
+     * or a double bound that represents end of one range tombstone and start of another at the same time.
+     * If range tombstone marker is a single bound then only _left_range_tombstone is used.
+     * Otherwise, _left_range_tombstone represents tombstone for a range tombstone that's being closed
+     * and _right_range_tombstone represents a tombstone for a range tombstone that's being opened.
+     */
+    tombstone _left_range_tombstone;
+    tombstone _right_range_tombstone;
     void setup_columns(const std::vector<std::optional<column_id>>& column_ids,
                        const std::vector<std::optional<uint32_t>>& column_value_fix_lengths,
                        const std::vector<bool>& column_is_collection,
@@ -662,7 +694,12 @@ private:
     void setup_ck(const std::vector<std::optional<uint32_t>>& column_value_fix_lengths) {
         _row_key.clear();
         _row_key.reserve(column_value_fix_lengths.size());
-        _ck_column_value_fix_lengths = boost::make_iterator_range(column_value_fix_lengths);
+        if (column_value_fix_lengths.empty()) {
+            _ck_column_value_fix_lengths = boost::make_iterator_range(column_value_fix_lengths);
+        } else {
+            _ck_column_value_fix_lengths = boost::make_iterator_range(std::begin(column_value_fix_lengths),
+                                                                      std::begin(column_value_fix_lengths) + _ck_size);
+        }
         _ck_blocks_header_offset = 0u;
     }
     bool no_more_ck_blocks() { return _ck_column_value_fix_lengths.empty(); }
@@ -679,6 +716,9 @@ private:
     bool is_block_empty() {
         return (_ck_blocks_header & (1u << (2 * _ck_blocks_header_offset))) != 0;
     }
+    bool is_block_null() {
+        return (_ck_blocks_header & (1u << (2 * _ck_blocks_header_offset + 1))) != 0;
+    }
     bool should_read_block_header() {
         return _ck_blocks_header_offset == 0u;
     }
@@ -762,6 +802,7 @@ public:
                               _column_translation.regular_column_value_fix_legths(),
                               _column_translation.regular_column_is_collection(),
                               _column_translation.regular_column_is_counter());
+                _ck_size = _column_translation.clustering_column_value_fix_legths().size();
                 goto clustering_row_label;
             }
             if (read_8(data) != read_status::ready) {
@@ -787,14 +828,20 @@ public:
                           _column_translation.regular_column_value_fix_legths(),
                           _column_translation.regular_column_is_collection(),
                           _column_translation.regular_column_is_counter());
+            _ck_size = _column_translation.clustering_column_value_fix_legths().size();
         case state::CLUSTERING_ROW:
         clustering_row_label:
             _is_first_unfiltered = false;
+            _null_component_occured = false;
             setup_ck(_column_translation.clustering_column_value_fix_legths());
         case state::CK_BLOCK:
         ck_block_label:
             if (no_more_ck_blocks()) {
-                goto clustering_row_consume_label;
+                if (_reading_range_tombstone_ck) {
+                    goto range_tombstone_consume_ck_label;
+                } else {
+                    goto clustering_row_consume_label;
+                }
             }
             if (!should_read_block_header()) {
                 _state = state::CK_BLOCK2;
@@ -808,6 +855,14 @@ public:
             _ck_blocks_header = _u64;
         case state::CK_BLOCK2:
         ck_block2_label: {
+            if (is_block_null()) {
+                _null_component_occured = true;
+                move_to_next_ck_block();
+                goto ck_block_label;
+            }
+            if (_null_component_occured) {
+                throw malformed_sstable_exception("non-null component after null component");
+            }
             if (is_block_empty()) {
                 _row_key.push_back({});
                 move_to_next_ck_block();
@@ -1115,7 +1170,91 @@ public:
             goto column_label;
         case state::RANGE_TOMBSTONE_MARKER:
         range_tombstone_marker_label:
-            throw malformed_sstable_exception("unimplemented state");
+            _is_first_unfiltered = false;
+            if (read_8(data) != read_status::ready) {
+                _state = state::RANGE_TOMBSTONE_KIND;
+                break;
+            }
+        case state::RANGE_TOMBSTONE_KIND:
+            _range_tombstone_kind = bound_kind_m(_u8);
+            if (read_16(data) != read_status::ready) {
+                _state = state::RANGE_TOMBSTONE_SIZE;
+                break;
+            }
+        case state::RANGE_TOMBSTONE_SIZE:
+            _ck_size = _u16;
+            if (_ck_size == 0) {
+                _row_key.clear();
+                _range_tombstone_kind = is_start(_range_tombstone_kind)
+                        ? bound_kind_m::incl_start : bound_kind_m::incl_end;
+                goto range_tombstone_body_label;
+            } else {
+                _reading_range_tombstone_ck = true;
+                goto clustering_row_label;
+            }
+            assert(0);
+        case state::RANGE_TOMBSTONE_CONSUME_CK:
+        range_tombstone_consume_ck_label:
+            _reading_range_tombstone_ck = false;
+        case state::RANGE_TOMBSTONE_BODY:
+        range_tombstone_body_label:
+            if (read_unsigned_vint(data) != read_status::ready) {
+                _state = state::RANGE_TOMBSTONE_BODY_SIZE;
+                break;
+            }
+        case state::RANGE_TOMBSTONE_BODY_SIZE:
+            // Ignore result
+            if (read_unsigned_vint(data) != read_status::ready) {
+                _state = state::RANGE_TOMBSTONE_BODY_PREV_SIZE;
+                break;
+            }
+        case state::RANGE_TOMBSTONE_BODY_PREV_SIZE:
+            // Ignore result
+            if (read_unsigned_vint(data) != read_status::ready) {
+                _state = state::RANGE_TOMBSTONE_BODY_TIMESTAMP;
+                break;
+            }
+        case state::RANGE_TOMBSTONE_BODY_TIMESTAMP:
+            _left_range_tombstone.timestamp = parse_timestamp(_header, _u64);
+            if (read_unsigned_vint(data) != read_status::ready) {
+                _state = state::RANGE_TOMBSTONE_BODY_LOCAL_DELTIME;
+                break;
+            }
+        case state::RANGE_TOMBSTONE_BODY_LOCAL_DELTIME:
+            _left_range_tombstone.deletion_time = parse_expiry(_header, _u64);
+            if (!is_boundary(_range_tombstone_kind)) {
+                if (_consumer.consume_range_tombstone(_row_key,
+                                                      to_bound_kind(_range_tombstone_kind),
+                                                      _left_range_tombstone) == consumer_m::proceed::no) {
+                    _row_key.clear();
+                    _state = state::FLAGS;
+                    return consumer_m::proceed::no;
+                }
+                _row_key.clear();
+                goto flags_label;
+            }
+            if (read_unsigned_vint(data) != read_status::ready) {
+                _state = state::RANGE_TOMBSTONE_BODY_TIMESTAMP2;
+                break;
+            }
+        case state::RANGE_TOMBSTONE_BODY_TIMESTAMP2:
+            _right_range_tombstone.timestamp = parse_timestamp(_header, _u64);
+            if (read_unsigned_vint(data) != read_status::ready) {
+                _state = state::RANGE_TOMBSTONE_BODY_LOCAL_DELTIME2;
+                break;
+            }
+        case state::RANGE_TOMBSTONE_BODY_LOCAL_DELTIME2:
+            _right_range_tombstone.deletion_time = parse_expiry(_header, _u64);
+            if (_consumer.consume_range_tombstone(_row_key,
+                                                  _range_tombstone_kind,
+                                                  _left_range_tombstone,
+                                                  _right_range_tombstone) == consumer_m::proceed::no) {
+                _row_key.clear();
+                _state = state::FLAGS;
+                return consumer_m::proceed::no;
+            }
+            _row_key.clear();
+            goto flags_label;
         default:
             throw malformed_sstable_exception("unknown state");
         }

tests/sstable_3_x_test.cc

@@ -1739,7 +1739,155 @@ SEASTAR_THREAD_TEST_CASE(test_uncompressed_deleted_cells_read) {
     .produces_end_of_stream();
 }
 
-// Following tests run on files in tests/sstables/3.x/uncompressed/compound_ck
+// Following tests run on files in tests/sstables/3.x/uncompressed/range_tombstones_simple
+// They were created using following CQL statements:
+//
+// CREATE KEYSPACE test_ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};
+//
+// CREATE TABLE test_ks.test_table ( pk INT, ck INT, val INT, PRIMARY KEY(pk, ck))
+//      WITH compression = { 'enabled' : false };
+//
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 101, 1001);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 102, 1002);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 103, 1003);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 104, 1004);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 105, 1005);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 106, 1006);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 107, 1007);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 108, 1008);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 109, 1009);
+// INSERT INTO test_ks.test_table(pk, ck, val) VALUES(1, 110, 1010);
+// DELETE FROM test_ks.test_table WHERE pk = 1 AND ck > 101 AND ck < 104;
+// DELETE FROM test_ks.test_table WHERE pk = 1 AND ck >= 104 AND ck < 105;
+// DELETE FROM test_ks.test_table WHERE pk = 1 AND ck > 108;
+
+static thread_local const sstring UNCOMPRESSED_RANGE_TOMBSTONES_SIMPLE_PATH = "tests/sstables/3.x/uncompressed/range_tombstones_simple";
+static thread_local const schema_ptr UNCOMPRESSED_RANGE_TOMBSTONES_SIMPLE_SCHEMA =
+    schema_builder("test_ks", "test_table")
+        .with_column("pk", int32_type, column_kind::partition_key)
+        .with_column("ck", int32_type, column_kind::clustering_key)
+        .with_column("val", int32_type)
+        .build();
+
+SEASTAR_THREAD_TEST_CASE(test_uncompressed_range_tombstones_simple_read) {
+    sstable_assertions sst(UNCOMPRESSED_RANGE_TOMBSTONES_SIMPLE_SCHEMA,
+                           UNCOMPRESSED_RANGE_TOMBSTONES_SIMPLE_PATH);
+    sst.load();
+    auto to_key = [] (int key) {
+        auto bytes = int32_type->decompose(int32_t(key));
+        auto pk = partition_key::from_single_value(*UNCOMPRESSED_RANGE_TOMBSTONES_SIMPLE_SCHEMA, bytes);
+        return dht::global_partitioner().decorate_key(*UNCOMPRESSED_RANGE_TOMBSTONES_SIMPLE_SCHEMA, pk);
+    };
+
+    auto to_ck = [] (int ck) {
+        return clustering_key::from_single_value(*UNCOMPRESSED_RANGE_TOMBSTONES_SIMPLE_SCHEMA,
+                                                 int32_type->decompose(ck));
+    };
+
+    auto int_cdef =
+    UNCOMPRESSED_RANGE_TOMBSTONES_SIMPLE_SCHEMA->get_column_definition(to_bytes("val"));
+    BOOST_REQUIRE(int_cdef);
+
+
+    assert_that(sst.read_rows_flat())
+    .produces_partition_start(to_key(1))
+    .produces_row(to_ck(101), {{int_cdef, int32_type->decompose(1001)}})
+    .produces_range_tombstone(range_tombstone(to_ck(101),
+                                              bound_kind::excl_start,
+                                              to_ck(104),
+                                              bound_kind::excl_end,
+                                              tombstone(api::timestamp_type{1529519641211958},
+                                                        gc_clock::time_point(gc_clock::duration(1529519641)))))
+    .produces_range_tombstone(range_tombstone(to_ck(104),
+                                              bound_kind::incl_start,
+                                              to_ck(105),
+                                              bound_kind::excl_end,
+                                              tombstone(api::timestamp_type{1529519641215380},
+                                                        gc_clock::time_point(gc_clock::duration(1529519641)))))
+    .produces_row(to_ck(105), {{int_cdef, int32_type->decompose(1005)}})
+    .produces_row(to_ck(106), {{int_cdef, int32_type->decompose(1006)}})
+    .produces_row(to_ck(107), {{int_cdef, int32_type->decompose(1007)}})
+    .produces_row(to_ck(108), {{int_cdef, int32_type->decompose(1008)}})
+    .produces_range_tombstone(range_tombstone(to_ck(108),
+                                              bound_kind::excl_start,
+                                              clustering_key_prefix::make_empty(),
+                                              bound_kind::incl_end,
+                                              tombstone(api::timestamp_type{1529519643267068},
+                                                        gc_clock::time_point(gc_clock::duration(1529519643)))))
+    .produces_partition_end()
+    .produces_end_of_stream();
+}
+
+// Following tests run on files in tests/sstables/3.x/uncompressed/range_tombstones_partial
+// They were created using following CQL statements:
+//
+// CREATE KEYSPACE test_ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};
+//
+// CREATE TABLE test_ks.test_table ( pk INT, ck1 INT, ck2 INT, PRIMARY KEY(pk, ck1, ck2))
+//      WITH compression = { 'enabled' : false };
+//
+// DELETE FROM test_ks.test_table WHERE pk = 1 AND ck1 > 1 AND ck1 < 3;
+// INSERT INTO test_ks.test_table(pk, ck1, ck2) VALUES(1, 2, 13);
+// DELETE FROM test_ks.test_table WHERE pk = 1 AND ck1 > 3;
+
+static thread_local const sstring UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_PATH = "tests/sstables/3.x/uncompressed/range_tombstones_partial";
+static thread_local const schema_ptr UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_SCHEMA =
+    schema_builder("test_ks", "test_table")
+        .with_column("pk", int32_type, column_kind::partition_key)
+        .with_column("ck1", int32_type, column_kind::clustering_key)
+        .with_column("ck2", int32_type, column_kind::clustering_key)
+        .build();
+
+SEASTAR_THREAD_TEST_CASE(test_uncompressed_range_tombstones_partial_read) {
+    sstable_assertions sst(UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_SCHEMA,
+                           UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_PATH);
+    sst.load();
+    auto to_key = [] (int key) {
+        auto bytes = int32_type->decompose(int32_t(key));
+        auto pk = partition_key::from_single_value(*UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_SCHEMA, bytes);
+        return dht::global_partitioner().decorate_key(*UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_SCHEMA, pk);
+    };
+
+    auto to_ck = [] (int ck) {
+        return clustering_key::from_single_value(*UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_SCHEMA,
+                                                 int32_type->decompose(ck));
+    };
+
+    assert_that(sst.read_rows_flat())
+    .produces_partition_start(to_key(1))
+    .produces_range_tombstone(range_tombstone(to_ck(1),
+                              bound_kind::excl_start,
+                              clustering_key::from_exploded(*UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_SCHEMA, {
+                                  int32_type->decompose(2),
+                                  int32_type->decompose(13)
+                              }),
+                              bound_kind::incl_end,
+                              tombstone(api::timestamp_type{1530543711595401},
+                              gc_clock::time_point(gc_clock::duration(1530543711)))))
+    .produces_row_with_key(clustering_key::from_exploded(*UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_SCHEMA, {
+                                                            int32_type->decompose(2),
+                                                            int32_type->decompose(13)
+                                                        }))
+    .produces_range_tombstone(range_tombstone(clustering_key::from_exploded(*UNCOMPRESSED_RANGE_TOMBSTONES_PARTIAL_SCHEMA, {
+                                  int32_type->decompose(2),
+                                  int32_type->decompose(13)
+                              }),
+                              bound_kind::incl_start,
+                              to_ck(3),
+                              bound_kind::excl_end,
+                              tombstone(api::timestamp_type{1530543711595401},
+                              gc_clock::time_point(gc_clock::duration(1530543711)))))
+    .produces_range_tombstone(range_tombstone(to_ck(3),
+                              bound_kind::excl_start,
+                              clustering_key_prefix::make_empty(),
+                              bound_kind::incl_end,
+                              tombstone(api::timestamp_type{1530543761322213},
+                              gc_clock::time_point(gc_clock::duration(1530543761)))))
+    .produces_partition_end()
+    .produces_end_of_stream();
+}
+
+// Following tests run on files in tests/sstables/3.x/uncompressed/simple
 // They were created using following CQL statements:
 //
 // CREATE KEYSPACE test_ks WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};

tests/sstables/3.x/uncompressed/range_tombstones_partial/mc-1-big-Digest.crc32

@@ -0,0 +1 @@
+32518023

tests/sstables/3.x/uncompressed/range_tombstones_partial/mc-1-big-TOC.txt

@@ -0,0 +1,8 @@
+Index.db
+TOC.txt
+Statistics.db
+Digest.crc32
+CRC.db
+Data.db
+Summary.db
+Filter.db

tests/sstables/3.x/uncompressed/range_tombstones_simple/mc-1-big-Digest.crc32

@@ -0,0 +1 @@
+1409872738

tests/sstables/3.x/uncompressed/range_tombstones_simple/mc-1-big-TOC.txt

@@ -0,0 +1,8 @@
+Statistics.db
+Digest.crc32
+CRC.db
+Data.db
+TOC.txt
+Index.db
+Summary.db
+Filter.db