ClickHouse

1
#include <Core/Range.h>
2
#include <Common/FieldVisitorToString.h>
3
#include <IO/WriteBufferFromString.h>
4
#include <IO/Operators.h>
5

6

7
namespace DB
8
{
9

10
Range::Range(const FieldRef & point) /// NOLINT
11
    : left(point), right(point), left_included(true), right_included(true) {}
12

13
/// A bounded two-sided range.
14
Range::Range(const FieldRef & left_, bool left_included_, const FieldRef & right_, bool right_included_)
15
    : left(left_)
16
    , right(right_)
17
    , left_included(left_included_)
18
    , right_included(right_included_)
19
{
20
    shrinkToIncludedIfPossible();
21
}
22

23
Range Range::createWholeUniverse()
24
{
25
    return Range(NEGATIVE_INFINITY, true, POSITIVE_INFINITY, true);
26
}
27

28
Range Range::createWholeUniverseWithoutNull()
29
{
30
    return Range(NEGATIVE_INFINITY, false, POSITIVE_INFINITY, false);
31
}
32

33
Range Range::createRightBounded(const FieldRef & right_point, bool right_included, bool with_null)
34
{
35
    Range r = with_null ? createWholeUniverse() : createWholeUniverseWithoutNull();
36
    r.right = right_point;
37
    r.right_included = right_included;
38
    r.shrinkToIncludedIfPossible();
39
    // Special case for [-Inf, -Inf]
40
    if (r.right.isNegativeInfinity() && right_included)
41
        r.left_included = true;
42
    return r;
43
}
44

45
Range Range::createLeftBounded(const FieldRef & left_point, bool left_included, bool with_null)
46
{
47
    Range r = with_null ? createWholeUniverse() : createWholeUniverseWithoutNull();
48
    r.left = left_point;
49
    r.left_included = left_included;
50
    r.shrinkToIncludedIfPossible();
51
    // Special case for [+Inf, +Inf]
52
    if (r.left.isPositiveInfinity() && left_included)
53
        r.right_included = true;
54
    return r;
55
}
56

57
/** Optimize the range. If it has an open boundary and the Field type is "loose"
58
  * - then convert it to closed, narrowing by one.
59
  * That is, for example, turn (0,2) into [1].
60
  */
61
void Range::shrinkToIncludedIfPossible()
62
{
63
    if (left.isExplicit() && !left_included)
64
    {
65
        if (left.getType() == Field::Types::UInt64 && left.get<UInt64>() != std::numeric_limits<UInt64>::max())
66
        {
67
            ++left.get<UInt64 &>();
68
            left_included = true;
69
        }
70
        if (left.getType() == Field::Types::Int64 && left.get<Int64>() != std::numeric_limits<Int64>::max())
71
        {
72
            ++left.get<Int64 &>();
73
            left_included = true;
74
        }
75
    }
76
    if (right.isExplicit() && !right_included)
77
    {
78
        if (right.getType() == Field::Types::UInt64 && right.get<UInt64>() != std::numeric_limits<UInt64>::min())
79
        {
80
            --right.get<UInt64 &>();
81
            right_included = true;
82
        }
83
        if (right.getType() == Field::Types::Int64 && right.get<Int64>() != std::numeric_limits<Int64>::min())
84
        {
85
            --right.get<Int64 &>();
86
            right_included = true;
87
        }
88
    }
89
}
90

91
bool Range::equals(const Field & lhs, const Field & rhs)
92
{
93
    return applyVisitor(FieldVisitorAccurateEquals(), lhs, rhs);
94
}
95

96
bool Range::less(const Field & lhs, const Field & rhs)
97
{
98
    return applyVisitor(FieldVisitorAccurateLess(), lhs, rhs);
99
}
100

101
bool Range::empty() const
102
{
103
    return less(right, left)
104
        || ((!left_included || !right_included)
105
            && !less(left, right));
106
}
107

108
/// x contained in the range
109
bool Range::contains(const FieldRef & x) const
110
{
111
    return !leftThan(x) && !rightThan(x);
112
}
113

114
/// x is to the left
115
bool Range::rightThan(const FieldRef & x) const
116
{
117
    return less(left, x) || (left_included && equals(x, left));
118
}
119

120
/// x is to the right
121
bool Range::leftThan(const FieldRef & x) const
122
{
123
    return less(x, right) || (right_included && equals(x, right));
124
}
125

126
bool Range::rightThan(const Range & x) const
127
{
128
    return less(x.right, left) || (!(left_included && x.right_included) && equals(left, x.right));
129
}
130

131
bool Range::leftThan(const Range & x) const
132
{
133
    return less(right, x.left) || (!(x.left_included && right_included) && equals(right, x.left));
134
}
135

136
bool Range::fullBounded() const
137
{
138
    return left.getType() != Field::Types::Null && right.getType() != Field::Types::Null;
139
}
140

141
/// (-inf, +inf)
142
bool Range::isInfinite() const
143
{
144
    return left.isNegativeInfinity() && right.isPositiveInfinity();
145
}
146

147
bool Range::intersectsRange(const Range & r) const
148
{
149
    /// r to the left of me.
150
    if (less(r.right, left) || ((!left_included || !r.right_included) && equals(r.right, left)))
151
        return false;
152

153
    /// r to the right of me.
154
    if (less(right, r.left) || ((!right_included || !r.left_included) && equals(r.left, right)))
155
        return false;
156

157
    return true;
158
}
159

160
bool Range::containsRange(const Range & r) const
161
{
162
    /// r starts to the left of me.
163
    if (less(r.left, left) || (r.left_included && !left_included && equals(r.left, left)))
164
        return false;
165

166
    /// r ends right of me.
167
    if (less(right, r.right) || (r.right_included && !right_included && equals(r.right, right)))
168
        return false;
169

170
    return true;
171
}
172

173
void Range::invert()
174
{
175
    std::swap(left, right);
176
    if (left.isPositiveInfinity())
177
        left = NEGATIVE_INFINITY;
178
    if (right.isNegativeInfinity())
179
        right = POSITIVE_INFINITY;
180
    std::swap(left_included, right_included);
181
}
182

183
Ranges Range::invertRange() const
184
{
185
    Ranges ranges;
186
    /// For full bounded range will generate two ranges.
187
    if (fullBounded()) /// case: [1, 3] -> (-inf, 1), (3, +inf)
188
    {
189
        ranges.push_back({NEGATIVE_INFINITY, false, left, !left_included});
190
        ranges.push_back({right, !right_included, POSITIVE_INFINITY, false});
191
    }
192
    else if (isInfinite())
193
    {
194
        /// blank ranges
195
    }
196
    else /// case: (-inf, 1] or [1, +inf)
197
    {
198
        Range r = *this;
199
        std::swap(r.left, r.right);
200
        if (r.left.isPositiveInfinity()) /// [1, +inf)
201
        {
202
            r.left = NEGATIVE_INFINITY;
203
            r.right_included = !r.left_included;
204
            r.left_included = false;
205
        }
206
        else if (r.right.isNegativeInfinity()) /// (-inf, 1]
207
        {
208
            r.right = POSITIVE_INFINITY;
209
            r.left_included = !r.right_included;
210
            r.right_included = false;
211
        }
212
        ranges.push_back(r);
213
    }
214
    return ranges;
215
}
216

217
std::optional<Range> Range::intersectWith(const Range & r) const
218
{
219
    if (!intersectsRange(r))
220
        return {};
221

222
    bool left_bound_use_mine = true;
223
    bool right_bound_use_mine = true;
224

225
    if (less(left, r.left) || ((!left_included && r.left_included) && equals(left, r.left)))
226
        left_bound_use_mine = false;
227

228
    if (less(r.right, right) || ((!r.right_included && right_included) && equals(r.right, right)))
229
        right_bound_use_mine = false;
230

231
    return Range(
232
        left_bound_use_mine ? left : r.left,
233
        left_bound_use_mine ? left_included : r.left_included,
234
        right_bound_use_mine ? right : r.right,
235
        right_bound_use_mine ? right_included : r.right_included);
236
}
237

238
std::optional<Range> Range::unionWith(const Range & r) const
239
{
240
    if (!intersectsRange(r) && !nearByWith(r))
241
        return {};
242

243
    bool left_bound_use_mine = false;
244
    bool right_bound_use_mine = false;
245

246
    if (less(left, r.left) || ((!left_included && r.left_included) && equals(left, r.left)))
247
        left_bound_use_mine = true;
248

249
    if (less(r.right, right) || ((!r.right_included && right_included) && equals(r.right, right)))
250
        right_bound_use_mine = true;
251

252
    return Range(
253
        left_bound_use_mine ? left : r.left,
254
        left_bound_use_mine ? left_included : r.left_included,
255
        right_bound_use_mine ? right : r.right,
256
        right_bound_use_mine ? right_included : r.right_included);
257
}
258

259
bool Range::nearByWith(const Range & r) const
260
{
261
    /// me locates at left
262
    if (((right_included && !r.left_included) || (!right_included && r.left_included)) && equals(right, r.left))
263
        return true;
264

265
    /// r locate left
266
    if (((r.right_included && !left_included) || (r.right_included && !left_included)) && equals(r.right, left))
267
        return true;
268

269
    return false;
270
}
271

272
Range intersect(const Range & a, const Range & b)
273
{
274
    Range res = Range::createWholeUniverse();
275

276
    if (Range::less(a.left, b.left))
277
    {
278
        res.left = b.left;
279
        res.left_included = b.left_included;
280
    }
281
    else if (Range::equals(a.left, b.left))
282
    {
283
        res.left = a.left;
284
        res.left_included = a.left_included && b.left_included;
285
    }
286
    else
287
    {
288
        res.left = a.left;
289
        res.left_included = a.left_included;
290
    }
291

292
    if (Range::less(a.right, b.right))
293
    {
294
        res.right = a.right;
295
        res.right_included = a.right_included;
296
    }
297
    else if (Range::equals(a.right, b.right))
298
    {
299
        res.right = a.right;
300
        res.right_included = a.right_included && b.right_included;
301
    }
302
    else
303
    {
304
        res.right = b.right;
305
        res.right_included = b.right_included;
306
    }
307

308
    if (res.empty())
309
    {
310
        res.right = res.left;
311
        res.right_included = false;
312
        res.left_included = false;
313
    }
314

315
    return res;
316
}
317

318
String Range::toString() const
319
{
320
    WriteBufferFromOwnString str;
321

322
    str << (left_included ? '[' : '(') << applyVisitor(FieldVisitorToString(), left) << ", ";
323
    str << applyVisitor(FieldVisitorToString(), right) << (right_included ? ']' : ')');
324

325
    return str.str();
326
}
327

328
Hyperrectangle intersect(const Hyperrectangle & a, const Hyperrectangle & b)
329
{
330
    size_t result_size = std::min(a.size(), b.size());
331

332
    Hyperrectangle res;
333
    res.reserve(result_size);
334

335
    for (size_t i = 0; i < result_size; ++i)
336
        res.push_back(intersect(a[i], b[i]));
337

338
    return res;
339
}
340

341
String toString(const Hyperrectangle & x)
342
{
343
    WriteBufferFromOwnString str;
344

345
    bool first = true;
346
    for (const auto & range : x)
347
    {
348
        if (!first)
349
            str << " × ";
350
        str << range.toString();
351
        first = false;
352
    }
353

354
    return str.str();
355
}
356

357
}
358