<commit_before>/*
 * Copyright (C) 2017-present ScyllaDB
 */

/*
 * SPDX-License-Identifier: AGPL-3.0-or-later
 */

#pragma once

#include <boost/test/unit_test.hpp>
#include <seastar/util/backtrace.hh>
#include "flat_mutation_reader_v2.hh"
#include "mutation_assertions.hh"
#include "schema.hh"
#include "test/lib/log.hh"

// Intended to be called in a seastar thread
class flat_reader_assertions {
    flat_mutation_reader _reader;
    dht::partition_range _pr;
    range_tombstone_list _tombstones;
    range_tombstone_list _expected_tombstones;
    bool _check_rts = false;
    query::clustering_row_ranges _rt_ck_ranges = {};
    bool _ignore_deletion_time = false;
private:
    mutation_fragment_opt read_next() {
        return _reader().get0();
    }
    range_tombstone maybe_drop_deletion_time(const range_tombstone& rt) const {
        if (!_ignore_deletion_time) {
            return rt;
        } else {
            return {rt.start, rt.start_kind, rt.end, rt.end_kind, {rt.tomb.timestamp, {}}};
        }
    }
    void check_rts() {
        if (_check_rts) {
            // If any split ends of range tombstones remain, consume
            // them.  For finer range restriction the test should call
            // may_produce_tombstones() before this happens.
            may_produce_tombstones(position_range::full(), _rt_ck_ranges);
            testlog.trace("Comparing normalized range tombstones");
            if (!_tombstones.equal(*_reader.schema(), _expected_tombstones)) {
                BOOST_FAIL(format("Expected {}, but got {}", _expected_tombstones, _tombstones));
            }
        }
        _tombstones.clear();
        _expected_tombstones.clear();
        _rt_ck_ranges.clear();
        _check_rts = false;
    }
    range_tombstone trim(const range_tombstone& rt, const query::clustering_row_ranges& ck_ranges) const {
        bound_view::compare less(*_reader.schema());
        auto ret{rt};
        for (auto& range : ck_ranges) {
            ret.trim(*_reader.schema(), position_in_partition::for_range_start(range), position_in_partition::for_range_end(range));
        }
        return ret;
    }
    void apply_rt_unchecked(const range_tombstone& rt_) {
        auto rt = maybe_drop_deletion_time(rt_);
        _tombstones.apply(*_reader.schema(), rt);
        _expected_tombstones.apply(*_reader.schema(), rt);
    }
public:
    flat_reader_assertions(flat_mutation_reader reader)
        : _reader(std::move(reader))
        , _tombstones(*_reader.schema())
        , _expected_tombstones(*_reader.schema())
    { }

    ~flat_reader_assertions() {
        // make sure to close the reader no matter what, to prevent
        // spurios logs about it not being closed if check_rts()
        // throws.
        try {
            check_rts();
        } catch (...) {
            _reader.close().get();
            throw;
        }
        _reader.close().get();
    }

    flat_reader_assertions(const flat_reader_assertions&) = delete;
    flat_reader_assertions(flat_reader_assertions&&) = default;

    flat_reader_assertions& operator=(flat_reader_assertions&& o) {
        if (this != &o) {
            _reader.close().get();
            _reader = std::move(o._reader);
            _pr = std::move(o._pr);
            _tombstones = std::move(o._tombstones);
            _expected_tombstones = std::move(o._expected_tombstones);
            _check_rts = std::move(o._check_rts);
            _ignore_deletion_time = std::move(o._ignore_deletion_time);
        }
        return *this;
    }

    flat_reader_assertions& ignore_deletion_time(bool ignore = true) {
        _ignore_deletion_time = ignore;
        return *this;
    }

    flat_reader_assertions& produces_partition_start(const dht::decorated_key& dk,
                                                     std::optional<tombstone> tomb = std::nullopt) {
        testlog.trace("Expecting partition start with key {}", dk);
        auto mfopt = read_next();
        if (!mfopt) {
            BOOST_FAIL(format("Expected: partition start with key {}, got end of stream", dk));
        }
        if (!mfopt->is_partition_start()) {
            BOOST_FAIL(format("Expected: partition start with key {}, got: {}", dk, mutation_fragment::printer(*_reader.schema(), *mfopt)));
        }
        if (!mfopt->as_partition_start().key().equal(*_reader.schema(), dk)) {
            BOOST_FAIL(format("Expected: partition start with key {}, got: {}", dk, mutation_fragment::printer(*_reader.schema(), *mfopt)));
        }
        if (tomb && mfopt->as_partition_start().partition_tombstone() != *tomb) {
            BOOST_FAIL(format("Expected: partition start with tombstone {}, got: {}", *tomb, mutation_fragment::printer(*_reader.schema(), *mfopt)));
        }
        _tombstones.clear();
        _expected_tombstones.clear();
        _check_rts = false;
        return *this;
    }
<commit_msg>flat_reader_assertions: refactor resetting accumulated tombstone lists<commit_after>/*
 * Copyright (C) 2017-present ScyllaDB
 */

/*
 * SPDX-License-Identifier: AGPL-3.0-or-later
 */

#pragma once

#include <boost/test/unit_test.hpp>
#include <seastar/util/backtrace.hh>
#include "flat_mutation_reader_v2.hh"
#include "mutation_assertions.hh"
#include "schema.hh"
#include "test/lib/log.hh"

// Intended to be called in a seastar thread
class flat_reader_assertions {
    flat_mutation_reader _reader;
    dht::partition_range _pr;
    range_tombstone_list _tombstones;
    range_tombstone_list _expected_tombstones;
    bool _check_rts = false;
    query::clustering_row_ranges _rt_ck_ranges = {};
    bool _ignore_deletion_time = false;
private:
    mutation_fragment_opt read_next() {
        return _reader().get0();
    }
    range_tombstone maybe_drop_deletion_time(const range_tombstone& rt) const {
        if (!_ignore_deletion_time) {
            return rt;
        } else {
            return {rt.start, rt.start_kind, rt.end, rt.end_kind, {rt.tomb.timestamp, {}}};
        }
    }
    void reset_rts() {
        _tombstones.clear();
        _expected_tombstones.clear();
        _rt_ck_ranges.clear();
        _check_rts = false;
    }
    void check_rts() {
        if (_check_rts) {
            // If any split ends of range tombstones remain, consume
            // them.  For finer range restriction the test should call
            // may_produce_tombstones() before this happens.
            may_produce_tombstones(position_range::full(), _rt_ck_ranges);
            testlog.trace("Comparing normalized range tombstones");
            if (!_tombstones.equal(*_reader.schema(), _expected_tombstones)) {
                BOOST_FAIL(format("Expected {}, but got {}", _expected_tombstones, _tombstones));
            }
        }
        reset_rts();
    }
    range_tombstone trim(const range_tombstone& rt, const query::clustering_row_ranges& ck_ranges) const {
        bound_view::compare less(*_reader.schema());
        auto ret{rt};
        for (auto& range : ck_ranges) {
            ret.trim(*_reader.schema(), position_in_partition::for_range_start(range), position_in_partition::for_range_end(range));
        }
        return ret;
    }
    void apply_rt_unchecked(const range_tombstone& rt_) {
        auto rt = maybe_drop_deletion_time(rt_);
        _tombstones.apply(*_reader.schema(), rt);
        _expected_tombstones.apply(*_reader.schema(), rt);
    }
public:
    flat_reader_assertions(flat_mutation_reader reader)
        : _reader(std::move(reader))
        , _tombstones(*_reader.schema())
        , _expected_tombstones(*_reader.schema())
    { }

    ~flat_reader_assertions() {
        // make sure to close the reader no matter what, to prevent
        // spurios logs about it not being closed if check_rts()
        // throws.
        try {
            check_rts();
        } catch (...) {
            _reader.close().get();
            throw;
        }
        _reader.close().get();
    }

    flat_reader_assertions(const flat_reader_assertions&) = delete;
    flat_reader_assertions(flat_reader_assertions&&) = default;

    flat_reader_assertions& operator=(flat_reader_assertions&& o) {
        if (this != &o) {
            _reader.close().get();
            _reader = std::move(o._reader);
            _pr = std::move(o._pr);
            _tombstones = std::move(o._tombstones);
            _expected_tombstones = std::move(o._expected_tombstones);
            _check_rts = std::move(o._check_rts);
            _ignore_deletion_time = std::move(o._ignore_deletion_time);
        }
        return *this;
    }

    flat_reader_assertions& ignore_deletion_time(bool ignore = true) {
        _ignore_deletion_time = ignore;
        return *this;
    }

    flat_reader_assertions& produces_partition_start(const dht::decorated_key& dk,
                                                     std::optional<tombstone> tomb = std::nullopt) {
        testlog.trace("Expecting partition start with key {}", dk);
        auto mfopt = read_next();
        if (!mfopt) {
            BOOST_FAIL(format("Expected: partition start with key {}, got end of stream", dk));
        }
        if (!mfopt->is_partition_start()) {
            BOOST_FAIL(format("Expected: partition start with key {}, got: {}", dk, mutation_fragment::printer(*_reader.schema(), *mfopt)));
        }
        if (!mfopt->as_partition_start().key().equal(*_reader.schema(), dk)) {
            BOOST_FAIL(format("Expected: partition start with key {}, got: {}", dk, mutation_fragment::printer(*_reader.schema(), *mfopt)));
        }
        if (tomb && mfopt->as_partition_start().partition_tombstone() != *tomb) {
            BOOST_FAIL(format("Expected: partition start with tombstone {}, got: {}", *tomb, mutation_fragment::printer(*_reader.schema(), *mfopt)));
        }
        reset_rts();
        return *this;
    }

{'fraction_non_alphanumeric': 0.12206047032474804, 'fraction_numerical': 0.0016288302962435101, 'mean_word_length': 2.6835395575553056, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2129784', 'n_tokens_mistral': 3124, 'n_tokens_neox': 3028, 'n_words': 571}