// Licensed to the Apache Software Foundation (ASF) under one

// or more contributor license agreements.  See the NOTICE file

// distributed with this work for additional information

// regarding copyright ownership.  The ASF licenses this file

// to you under the Apache License, Version 2.0 (the

// "License"); you may not use this file except in compliance

// with the License.  You may obtain a copy of the License at

//

//   http://www.apache.org/licenses/LICENSE-2.0

//

// Unless required by applicable law or agreed to in writing,

// software distributed under the License is distributed on an

// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY

// KIND, either express or implied.  See the License for the

// specific language governing permissions and limitations

// under the License.

#pragma once

#include <roaring/roaring.hh>

#include <string>

#include <vector>

#include "absl/strings/substitute.h"

#include "common/cast_set.h"

#include "common/logging.h"

#include "storage/segment/common.h"

namespace doris {

namespace segment_v2 {

// RowRange stands for range[From, To), From is inclusive,

// To is exclusive. It is used for row id range calculation.

class RowRange {

public:

    // Returns true if two ranges are overlapped or false.

    // The union range will be returned through range.

    static bool range_union(const RowRange& left, const RowRange& right, RowRange* range) {

        if (left._from <= right._from) {

            if (left._to >= right._from) {

                range->_from = left._from;

                range->_to = std::max(left._to, right._to);

                return true;

            }

        } else if (right._to >= left._from) {

            range->_from = right._from;

            range->_to = std::max(left._to, right._to);

            return true;

        }

        // return a invalid range

        range->_from = 0;

        range->_to = 0;

        return false;

    }

    // Returns true if the two ranges are intersected or false.

    // The intersection of the two ranges is returned through range.

    static bool range_intersection(const RowRange& left, const RowRange& right, RowRange* range) {

        if (left._from <= right._from) {

            if (left._to > right._from) {

                range->_from = right._from;

                range->_to = std::min(left._to, right._to);

                return true;

            }

        } else if (right._to > left._from) {

            range->_from = left._from;

            range->_to = std::min(left._to, right._to);

            return true;

        }

        // return a invalid range

        range->_from = 0;

        range->_to = 0;

        return false;

    }

    RowRange() : _from(0), _to(0) {}

    // Creates a range of [from, to) (from inclusive and to exclusive; empty ranges are invalid)

    RowRange(int64_t from, int64_t to) : _from(from), _to(to) {}

    bool is_valid() const { return _from < _to; }

    size_t count() const { return _to - _from; }

    bool is_before(const RowRange& other) const { return _to <= other._from; }

    bool is_after(const RowRange& other) const { return _from >= other._to; }

    int64_t from() const { return _from; }

    int64_t to() const { return _to; }

    std::string to_string() const { return absl::Substitute("[$0-$1)", _from, _to); }

    uint64_t get_digest(uint64_t seed) const {

        uint64_t hash = seed;

        hash = hash * 31 + _from;

        hash = hash * 31 + _to;

        return hash;

    }

private:

    int64_t _from;

    int64_t _to;

};

class RowRanges {

public:

    RowRanges() : _count(0) {}

    void clear() {

        _ranges.clear();

        _count = 0;

    }

    // Creates a new RowRanges object with the single range [0, row_count).

    static RowRanges create_single(uint64_t row_count) {

        RowRanges ranges;

        ranges.add(RowRange(0, row_count));

        return ranges;

    }

    // Creates a new RowRanges object with the single range [from, to).

    static RowRanges create_single(int64_t from, int64_t to) {

        DCHECK(from <= to);

        RowRanges ranges;

        ranges.add(RowRange(from, to));

        return ranges;

    }

    // Calculates the union of the two specified RowRanges object. The union of two range is calculated if there are

    // elements between them. Otherwise, the two disjunct ranges are stored separately.

    // For example:

    // [113, 241) ∪ [221, 340) = [113, 340)

    // [113, 230) ∪ [230, 340) = [113, 340]

    // while

    // [113, 230) ∪ [231, 340) = [113, 230), [231, 340)

    static void ranges_union(const RowRanges& left, const RowRanges& right, RowRanges* result) {

        RowRanges tmp_range;

        auto it1 = left._ranges.begin();

        auto it2 = right._ranges.begin();

        // merge and add

        while (it1 != left._ranges.end() && it2 != right._ranges.end()) {

            if (it1->is_after(*it2)) {

                tmp_range.add(*it2);

                ++it2;

            } else {

                tmp_range.add(*it1);

                ++it1;

            }

        }

        while (it1 != left._ranges.end()) {

            tmp_range.add(*it1);

            ++it1;

        }

        while (it2 != right._ranges.end()) {

            tmp_range.add(*it2);

            ++it2;

        }

        *result = std::move(tmp_range);

    }

    // Calculates the intersection of the two specified RowRanges object. Two ranges intersect if they have common

    // elements otherwise the result is empty.

    // For example:

    // [113, 241) ∩ [221, 340) = [221, 241)

    // while

    // [113, 230) ∩ [230, 340) = <EMPTY>

    //

    // The result RowRanges object will contain all the row indexes there were contained in both of the specified objects

    static void ranges_intersection(const RowRanges& left, const RowRanges& right,

                                    RowRanges* result) {

        RowRanges tmp_range;

        int right_index = 0;

        for (auto it1 = left._ranges.begin(); it1 != left._ranges.end(); ++it1) {

            const RowRange& range1 = *it1;

            for (int i = right_index; i < right._ranges.size(); ++i) {

                const RowRange& range2 = right._ranges[i];

                if (range1.is_before(range2)) {

                    break;

                } else if (range1.is_after(range2)) {

                    right_index = i + 1;

                    continue;

                }

                RowRange merge_range;

                bool ret = RowRange::range_intersection(range1, range2, &merge_range);

                DCHECK(ret);

                tmp_range.add(merge_range);

            }

        }

        *result = std::move(tmp_range);

    }

    // Calculates the exception (set difference) of the two specified RowRanges objects: left \ right.

    // The result contains all row indexes that are in the left ranges but NOT in the right ranges.

    // For example:

    // [100, 300) \ [150, 200) = [100, 150), [200, 300)

    // [100, 300) \ [0, 150) = [150, 300)

    // [100, 300) \ [250, 400) = [100, 250)

    // [100, 200) \ [200, 300) = [100, 200)

    // [100, 300) \ [0, 400) = <EMPTY>

    // [100, 200), [300, 400) \ [150, 350) = [100, 150), [350, 400)

    static void ranges_exception(const RowRanges& left, const RowRanges& right, RowRanges* result) {

        RowRanges tmp_range;

        int right_index = 0;

        for (auto it1 = left._ranges.begin(); it1 != left._ranges.end(); ++it1) {

            int64_t current_from = it1->from();

            int64_t current_to = it1->to();

            for (int i = right_index; i < right._ranges.size(); ++i) {

                const RowRange& range2 = right._ranges[i];

                if (current_from >= current_to) {

                    // Current range fully consumed

                    break;

                }

                if (current_to <= range2.from()) {

                    // Current remaining range is entirely before range2, no more subtraction needed

                    break;

                }

                if (current_from >= range2.to()) {

                    // range2 is entirely before the current remaining range, advance right_index

                    right_index = i + 1;

                    continue;

                }

                // There is overlap between [current_from, current_to) and range2

                if (current_from < range2.from()) {

                    // Left portion before the overlap: [current_from, range2.from())

                    tmp_range.add(RowRange(current_from, range2.from()));

                }

                // Advance current_from past the overlap

                current_from = range2.to();

            }

            // Add whatever remains of the current left range

            if (current_from < current_to) {

                tmp_range.add(RowRange(current_from, current_to));

            }

        }

        *result = std::move(tmp_range);

    }

    static roaring::Roaring ranges_to_roaring(const RowRanges& ranges) {

        roaring::Roaring result;

        for (auto it = ranges._ranges.begin(); it != ranges._ranges.end(); ++it) {

            result.addRange(it->from(), it->to());

        }

        return result;

    }

    size_t count() { return _count; }

    bool is_empty() { return _count == 0; }

    bool contain(rowid_t from, rowid_t to) {

        // binary search

        RowRange tmp_range = RowRange(from, to);

        size_t start = 0;

        size_t end = _ranges.size();

        while (start <= end) {

            size_t mid = (start + end) / 2;

            if (_ranges[mid].is_before(tmp_range)) {

                start = mid;

            } else if (_ranges[mid].is_after(tmp_range)) {

                end = mid - 1;

            } else {

                return true;

            }

        }

        return false;

    }

    int64_t from() {

        DCHECK(!is_empty());

        return _ranges[0].from();

    }

    int64_t to() {

        DCHECK(!is_empty());

        return _ranges[_ranges.size() - 1].to();

    }

    size_t range_size() const { return _ranges.size(); }

    RowRange get_range(size_t index) const { return _ranges[index]; }

    int64_t get_range_from(size_t range_index) const { return _ranges[range_index].from(); }

    int64_t get_range_to(size_t range_index) const { return _ranges[range_index].to(); }

    size_t get_range_count(size_t range_index) const { return _ranges[range_index].count(); }

    std::string to_string() {

        std::string result;

        for (auto range : _ranges) {

            result += range.to_string() + " ";

        }

        return result;

    }

    // Adds a range to the end of the list of ranges. It maintains the disjunct ascending order(*) of the ranges by

    // trying to union the specified range to the last ranges in the list. The specified range shall be larger(*) than

    // the last one or might be overlapped with some of the last ones.

    void add(const RowRange& range) {

        if (range.count() == 0) {

            return;

        }

        RowRange range_to_add = range;

        for (int i = cast_set<int>(_ranges.size()) - 1; i >= 0; --i) {

            const RowRange last = _ranges[i];

            DCHECK(!last.is_after(range));

            RowRange u;

            bool ret = RowRange::range_union(last, range_to_add, &u);

            if (!ret) {

                // range do not intersect with the last

                break;

            }

            range_to_add = u;

            _ranges.erase(_ranges.begin() + i);

            _count -= last.count();

        }

        _ranges.emplace_back(range_to_add);

        _count += range_to_add.count();

    }

    // Returns the row index (within the original row space) of the pos-th element

    // across all ranges. For example, if ranges are [0,3000) and [8000,11000),

    // pos=0 returns 0, pos=2999 returns 2999, pos=3000 returns 8000.

    int64_t get_row_index_by_pos(int64_t pos) const {

        DORIS_CHECK(pos < _count);

        size_t remaining = pos;

        for (const auto& range : _ranges) {

            size_t range_len = range.count();

            if (remaining < range_len) {

                return range.from() + remaining;

            }

            remaining -= range_len;

        }

        // pos is out of bounds; return -1 to indicate invalid

        DCHECK(false) << "pos " << pos << " is out of bounds for RowRanges with count " << _count;

        return -1;

    }

    uint64_t get_digest(uint64_t seed) const {

        for (auto range : _ranges) {

            seed = range.get_digest(seed);

        }

        return seed;

    }

private:

    std::vector<RowRange> _ranges;

    size_t _count;

};

} // namespace segment_v2

} // namespace doris