ClickHouse

1
#include <Core/PlainRanges.h>
2

3
namespace DB
4
{
5

6
PlainRanges::PlainRanges(const Range & range)
7
{
8
    ranges.push_back(range);
9
}
10

11

12
PlainRanges::PlainRanges(const Ranges & ranges_, bool may_have_intersection, bool ordered)
13
{
14
    if (may_have_intersection)
15
        ranges = ordered ? makePlainFromOrdered(ranges_) : makePlainFromUnordered(ranges_);
16
    else
17
        ranges = ranges_;
18
}
19

20
Ranges PlainRanges::makePlainFromOrdered(const Ranges & ranges_)
21
{
22
    if (ranges_.size() <= 1)
23
        return ranges_;
24

25
    Ranges ret{ranges_.front()};
26

27
    for (size_t i = 1; i < ranges_.size(); ++i)
28
    {
29
        const auto & cur = ranges_[i];
30
        if (ret.back().intersectsRange(cur))
31
            ret.back() = *ret.back().unionWith(cur);
32
        else
33
            ret.push_back(cur);
34
    }
35

36
    return ret;
37
}
38

39
Ranges PlainRanges::makePlainFromUnordered(Ranges ranges_)
40
{
41
    if (ranges_.size() <= 1)
42
        return ranges_;
43

44
    std::sort(ranges_.begin(), ranges_.end(), compareByLeftBound);
45
    return makePlainFromOrdered(ranges_);
46
}
47

48
PlainRanges PlainRanges::unionWith(const PlainRanges & other)
49
{
50
    auto left_itr = ranges.begin();
51
    auto right_itr = other.ranges.begin();
52

53
    Ranges new_range;
54
    for (; left_itr != ranges.end() && right_itr != other.ranges.end();)
55
    {
56
        if (left_itr->leftThan(*right_itr))
57
        {
58
            new_range.push_back(*left_itr);
59
            left_itr++;
60
        }
61
        else if (left_itr->rightThan(*right_itr))
62
        {
63
            new_range.push_back(*right_itr);
64
            right_itr++;
65
        }
66
        else /// union
67
        {
68
            new_range.emplace_back(*(left_itr->unionWith(*right_itr)));
69
            if (compareByRightBound(*left_itr, *right_itr))
70
                left_itr++;
71
            else
72
                right_itr++;
73
        }
74
    }
75

76
    while (left_itr != ranges.end())
77
    {
78
        new_range.push_back(*left_itr);
79
        left_itr++;
80
    }
81

82
    while (right_itr != other.ranges.end())
83
    {
84
        new_range.push_back(*right_itr);
85
        right_itr++;
86
    }
87

88
    /// After union two PlainRanges, new ranges may like: [1, 4], [2, 5]
89
    /// We must make them plain.
90

91
    return PlainRanges(makePlainFromOrdered(new_range));
92
}
93

94
PlainRanges PlainRanges::intersectWith(const PlainRanges & other)
95
{
96
    auto left_itr = ranges.begin();
97
    auto right_itr = other.ranges.begin();
98

99
    Ranges new_ranges;
100
    for (; left_itr != ranges.end() && right_itr != other.ranges.end();)
101
    {
102
        if (left_itr->leftThan(*right_itr))
103
        {
104
            left_itr++;
105
        }
106
        else if (left_itr->rightThan(*right_itr))
107
        {
108
            right_itr++;
109
        }
110
        else /// intersection
111
        {
112
            auto intersected = left_itr->intersectWith(*right_itr);
113

114
            if (intersected) /// skip blank range
115
                new_ranges.emplace_back(*intersected);
116

117
            if (compareByRightBound(*left_itr, *right_itr))
118
                left_itr++;
119
            else
120
                right_itr++;
121
        }
122
    }
123
    return PlainRanges(new_ranges);
124
}
125

126
bool PlainRanges::compareByLeftBound(const Range & lhs, const Range & rhs)
127
{
128
    if (lhs.left == NEGATIVE_INFINITY && rhs.left == NEGATIVE_INFINITY)
129
        return false;
130
    return Range::less(lhs.left, rhs.left) || ((!lhs.left_included && rhs.left_included) && Range::equals(lhs.left, rhs.left));
131
};
132

133
bool PlainRanges::compareByRightBound(const Range & lhs, const Range & rhs)
134
{
135
    if (lhs.right == POSITIVE_INFINITY && rhs.right == POSITIVE_INFINITY)
136
        return false;
137
    return Range::less(lhs.right, rhs.right) || ((!lhs.right_included && rhs.right_included) && Range::equals(lhs.right, rhs.right));
138
};
139

140

141
std::vector<Ranges> PlainRanges::invert(const Ranges & to_invert_ranges)
142
{
143
    /// invert a blank ranges
144
    if (to_invert_ranges.empty())
145
        return {makeUniverse().ranges};
146

147
    std::vector<Ranges> reverted_ranges;
148
    for (const auto & range : to_invert_ranges)
149
    {
150
        if (range.isInfinite())
151
            /// return a blank ranges
152
            return {{}};
153
        reverted_ranges.push_back(range.invertRange());
154
    }
155
    return reverted_ranges;
156
};
157
}
158