rupg_func/
multirange.rs

1//! The multirange types of `multirangetypes.c`: `make_multirange`, which sorts the ranges and merges the ranges that overlap or touch, the order of the multiranges, the constructors, the functions that read the bounds and the flags, `range_merge` of a multirange, `unnest` and the operators of a multirange with a multirange, a range or a value.
2
3use std::cmp::Ordering;
4
5use rupg_common::{Error, Result, SqlState};
6use rupg_pgcatalog::builtin;
7use rupg_types::{RANGE_LB_INC, RANGE_LB_INF, RANGE_UB_INC, RANGE_UB_INF, Range, Value};
8
9use crate::io::base_type;
10use crate::range::{
11    Bound, adjacent, before, bounds, bounds_adjacent, cmp_bounds, contains, contains_elem,
12    intersect, make_range, minus, over_left, overlaps, range_order, split, subtype, union,
13};
14use crate::{Call, Kernel, bad_value};
15
16/// The range type of a multirange type, or 0 for a type that is not a multirange type.
17pub(crate) fn range_of(multirange: u32) -> u32 {
18    builtin::multirange_range(base_type(multirange)).unwrap_or(0)
19}
20
21/// `make_multirange` with `multirange_canonicalize`: the ranges in the order of `range_cmp` with no empty range, and one range for each group of ranges that overlap or touch. `range` is the range type.
22///
23/// # Errors
24///
25/// The errors of the comparison of the bounds.
26pub(crate) fn make_multirange(range: u32, mut ranges: Vec<Range>) -> Result<Vec<Range>> {
27    let sub = subtype(range);
28    let mut failed = None;
29    ranges.sort_by(|a, b| {
30        range_order(sub, a, b).unwrap_or_else(|error| {
31            failed.get_or_insert(error);
32            Ordering::Equal
33        })
34    });
35    if let Some(error) = failed {
36        return Err(error);
37    }
38    let mut out: Vec<Range> = Vec::with_capacity(ranges.len());
39    for current in ranges {
40        if current.is_empty() {
41            continue;
42        }
43        let Some(last) = out.last_mut() else {
44            out.push(current);
45            continue;
46        };
47        if adjacent(range, last, &current)? {
48            *last = union(range, last, &current, false)?;
49        } else if before(range, last, &current)? {
50            out.push(current);
51        } else {
52            *last = union(range, last, &current, true)?;
53        }
54    }
55    Ok(out)
56}
57
58/// `range_agg_finalfn`: the multirange of the type `multirange` that has the values of the ranges. No range gives null.
59///
60/// # Errors
61///
62/// The errors of the comparison of the bounds.
63pub fn range_agg(multirange: u32, ranges: Vec<Range>) -> Result<Value> {
64    if ranges.is_empty() {
65        return Ok(Value::Null);
66    }
67    Ok(Value::Multirange(make_multirange(range_of(multirange), ranges)?))
68}
69
70/// `multirange_cmp`: the order of the ranges one by one, and a multirange with fewer ranges is first when its ranges are the first ranges of the other. `sub` is the subtype.
71pub(crate) fn multirange_order(sub: u32, a: &[Range], b: &[Range]) -> Result<Ordering> {
72    for (x, y) in a.iter().zip(b) {
73        let order = range_order(sub, x, y)?;
74        if order.is_ne() {
75            return Ok(order);
76        }
77    }
78    Ok(a.len().cmp(&b.len()))
79}
80
81/// The ranges of a call.
82fn arg(args: &[Value]) -> Result<&[Range]> {
83    match args.first() {
84        Some(Value::Multirange(ranges)) => Ok(ranges),
85        _ => Err(bad_value()),
86    }
87}
88
89/// `multirange_constructor0`: the empty multirange.
90fn constructor0(_: &Call<'_>, _: &[Value]) -> Result<Value> {
91    Ok(Value::Multirange(Vec::new()))
92}
93
94/// `multirange_constructor1`: the multirange of one range. The cast from a range type to its multirange type calls it.
95fn constructor1(call: &Call<'_>, args: &[Value]) -> Result<Value> {
96    let [Value::Range(range)] = args else { return Err(bad_value()) };
97    Ok(Value::Multirange(make_multirange(range_of(call.ret), vec![(**range).clone()])?))
98}
99
100/// `multirange_constructor2`: the multirange of the ranges of a `VARIADIC` array. The array must have one dimension or none and no null element.
101fn constructor2(call: &Call<'_>, args: &[Value]) -> Result<Value> {
102    let [Value::Array(array)] = args else { return Err(bad_value()) };
103    if array.dims.len() > 1 {
104        return Err(Error::new(
105            SqlState::CARDINALITY_VIOLATION,
106            "multiranges cannot be constructed from multidimensional arrays",
107        ));
108    }
109    let ranges = array
110        .values
111        .iter()
112        .map(|value| match value {
113            Some(Value::Range(range)) => Ok((**range).clone()),
114            Some(_) => Err(bad_value()),
115            None => Err(Error::new(
116                SqlState::NULL_VALUE_NOT_ALLOWED,
117                "multirange values cannot contain null members",
118            )),
119        })
120        .collect::<Result<Vec<_>>>()?;
121    Ok(Value::Multirange(make_multirange(range_of(call.ret), ranges)?))
122}
123
124/// `multirange_lower`: the lower bound of the first range, or null for an empty multirange or an infinite bound.
125fn lower(_: &Call<'_>, args: &[Value]) -> Result<Value> {
126    Ok(arg(args)?.first().map_or(Value::Null, |range| range.lower.clone()))
127}
128
129/// `multirange_upper`: the upper bound of the last range, or null for an empty multirange or an infinite bound.
130fn upper(_: &Call<'_>, args: &[Value]) -> Result<Value> {
131    Ok(arg(args)?.last().map_or(Value::Null, |range| range.upper.clone()))
132}
133
134/// `multirange_empty`: true for a multirange with no range.
135fn is_empty(_: &Call<'_>, args: &[Value]) -> Result<Value> {
136    Ok(Value::Bool(arg(args)?.is_empty()))
137}
138
139/// The kernel of a function that tests a flag of the first or the last range. An empty multirange gives false.
140macro_rules! flag_test {
141    ($name:ident, $end:ident, $flag:expr, $doc:literal) => {
142        #[doc = $doc]
143        fn $name(_: &Call<'_>, args: &[Value]) -> Result<Value> {
144            Ok(Value::Bool(arg(args)?.$end().is_some_and(|range| range.flags & $flag != 0)))
145        }
146    };
147}
148
149flag_test!(
150    lower_inc,
151    first,
152    RANGE_LB_INC,
153    "`multirange_lower_inc`: true when the lower bound is part of the multirange."
154);
155flag_test!(
156    upper_inc,
157    last,
158    RANGE_UB_INC,
159    "`multirange_upper_inc`: true when the upper bound is part of the multirange."
160);
161flag_test!(
162    lower_inf,
163    first,
164    RANGE_LB_INF,
165    "`multirange_lower_inf`: true when the multirange has no lower bound."
166);
167flag_test!(
168    upper_inf,
169    last,
170    RANGE_UB_INF,
171    "`multirange_upper_inf`: true when the multirange has no upper bound."
172);
173
174/// `range_merge_from_multirange`: the smallest range that contains each range of the multirange.
175fn merge(call: &Call<'_>, args: &[Value]) -> Result<Value> {
176    let ranges = arg(args)?;
177    let range = match ranges {
178        [] => Range::empty(),
179        [one] => one.clone(),
180        [first, .., last] => {
181            let flags = (first.flags & (RANGE_LB_INC | RANGE_LB_INF))
182                | (last.flags & (RANGE_UB_INC | RANGE_UB_INF));
183            make_range(call.ret, flags, first.lower.clone(), last.upper.clone())?
184        }
185    };
186    Ok(Value::Range(Box::new(range)))
187}
188
189/// `multirange_unnest`: one row for each range of the multirange, in order.
190pub(crate) fn unnest(_: &Call<'_>, args: &[Value]) -> Result<Vec<Vec<Value>>> {
191    Ok(arg(args)?.iter().map(|range| vec![Value::Range(Box::new(range.clone()))]).collect())
192}
193
194/// The range type of a call: the range type of the first argument or of the result that is a multirange type, or the first that is a range type.
195fn call_range(call: &Call<'_>) -> u32 {
196    let types = || call.args.iter().chain([&call.ret]).copied();
197    types()
198        .map(range_of)
199        .find(|&ty| ty != 0)
200        .or_else(|| types().find(|&ty| subtype(ty) != 0))
201        .unwrap_or(0)
202}
203
204/// The ranges of a range or a multirange argument. An empty range gives no range.
205fn ranges_of(value: &Value) -> Result<&[Range]> {
206    match value {
207        Value::Multirange(ranges) => Ok(ranges),
208        Value::Range(range) if range.is_empty() => Ok(&[]),
209        Value::Range(range) => Ok(std::slice::from_ref(&**range)),
210        _ => Err(bad_value()),
211    }
212}
213
214/// The lower bound of the first range and the upper bound of the last range, or `None` for no range.
215fn span(ranges: &[Range]) -> Option<(Bound<'_>, Bound<'_>)> {
216    let (first, last) = (ranges.first()?, ranges.last()?);
217    Some((bounds(first).0, bounds(last).1))
218}
219
220/// `range_overlaps_multirange_internal` and `multirange_overlaps_multirange_internal`: true when a range of `a` and a range of `b` have a common value. The ranges of each side are in order and do not overlap, so one pass finds the pair.
221fn overlaps_all(range: u32, a: &[Range], b: &[Range]) -> Result<bool> {
222    let sub = subtype(range);
223    let (mut i, mut j) = (0, 0);
224    while let (Some(x), Some(y)) = (a.get(i), b.get(j)) {
225        if overlaps(range, x, y)? {
226            return Ok(true);
227        }
228        if cmp_bounds(sub, &bounds(x).1, &bounds(y).1)?.is_lt() {
229            i += 1;
230        } else {
231            j += 1;
232        }
233    }
234    Ok(false)
235}
236
237/// `multirange_contains_multirange_internal`: true when each range of `b` is in one range of `a`. No range in `b` gives true.
238fn contains_all(range: u32, a: &[Range], b: &[Range]) -> Result<bool> {
239    let sub = subtype(range);
240    let mut i = 0;
241    for y in b {
242        let lower = bounds(y).0;
243        loop {
244            let Some(x) = a.get(i) else { return Ok(false) };
245            if cmp_bounds(sub, &bounds(x).1, &lower)?.is_ge() {
246                break;
247            }
248            i += 1;
249        }
250        if !contains(range, &a[i], y)? {
251            return Ok(false);
252        }
253    }
254    Ok(true)
255}
256
257/// The kernel of an operator that compares the outer bounds of two ranges or multiranges. An empty side gives false.
258macro_rules! position {
259    ($name:ident, $doc:literal, |$sub:pat, $range:ident, $a:ident, $b:ident| $test:expr) => {
260        #[doc = $doc]
261        fn $name(call: &Call<'_>, args: &[Value]) -> Result<Value> {
262            let [a, b] = args else { return Err(bad_value()) };
263            let $range = call_range(call);
264            let $sub = subtype($range);
265            let (Some($a), Some($b)) = (span(ranges_of(a)?), span(ranges_of(b)?)) else {
266                return Ok(Value::Bool(false));
267            };
268            Ok(Value::Bool($test))
269        }
270    };
271}
272
273position!(
274    before_op,
275    "`<<`: true when the upper bound of the left side is before the lower bound of the right side.",
276    |sub, range, a, b| cmp_bounds(sub, &a.1, &b.0)?.is_lt()
277);
278position!(
279    after_op,
280    "`>>`: true when the lower bound of the left side is after the upper bound of the right side.",
281    |sub, range, a, b| cmp_bounds(sub, &a.0, &b.1)?.is_gt()
282);
283position!(
284    over_left_op,
285    "`&<`: true when the left side does not go to the right of the right side.",
286    |sub, range, a, b| cmp_bounds(sub, &a.1, &b.1)?.is_le()
287);
288position!(
289    over_right_op,
290    "`&>`: true when the left side does not go to the left of the right side.",
291    |sub, range, a, b| cmp_bounds(sub, &a.0, &b.0)?.is_ge()
292);
293position!(
294    adjacent_op,
295    "`-|-`: true when the upper bound of one side touches the lower bound of the other side. Only the outer bounds count, as in PostgreSQL.",
296    |_, range, a, b| bounds_adjacent(range, &a.1, &b.0)? || bounds_adjacent(range, &b.1, &a.0)?
297);
298
299/// `&&` of a range or a multirange and a multirange.
300fn overlaps_op(call: &Call<'_>, args: &[Value]) -> Result<Value> {
301    let [a, b] = args else { return Err(bad_value()) };
302    Ok(Value::Bool(overlaps_all(call_range(call), ranges_of(a)?, ranges_of(b)?)?))
303}
304
305/// `@>` of a range or a multirange and a multirange, or of a multirange and a range.
306fn contains_op(call: &Call<'_>, args: &[Value]) -> Result<Value> {
307    let [a, b] = args else { return Err(bad_value()) };
308    Ok(Value::Bool(contains_all(call_range(call), ranges_of(a)?, ranges_of(b)?)?))
309}
310
311/// `<@` of a range or a multirange and a multirange, or of a multirange and a range.
312fn contained_by_op(call: &Call<'_>, args: &[Value]) -> Result<Value> {
313    let [a, b] = args else { return Err(bad_value()) };
314    Ok(Value::Bool(contains_all(call_range(call), ranges_of(b)?, ranges_of(a)?)?))
315}
316
317/// `multirange_contains_elem_internal`: true when a range of the multirange has the value.
318fn contains_value(range: u32, ranges: &[Range], value: &Value) -> Result<bool> {
319    let sub = subtype(range);
320    for x in ranges {
321        if contains_elem(sub, x, value)? {
322            return Ok(true);
323        }
324    }
325    Ok(false)
326}
327
328/// `multirange_contains_elem`: `@>` of a multirange and a value.
329fn contains_elem_op(call: &Call<'_>, args: &[Value]) -> Result<Value> {
330    let [a, value] = args else { return Err(bad_value()) };
331    Ok(Value::Bool(contains_value(call_range(call), ranges_of(a)?, value)?))
332}
333
334/// `elem_contained_by_multirange`: `<@` of a value and a multirange.
335fn elem_contained_op(call: &Call<'_>, args: &[Value]) -> Result<Value> {
336    let [value, a] = args else { return Err(bad_value()) };
337    Ok(Value::Bool(contains_value(call_range(call), ranges_of(a)?, value)?))
338}
339
340/// The ranges of two multirange arguments.
341fn pair(args: &[Value]) -> Result<(&[Range], &[Range])> {
342    match args {
343        [Value::Multirange(a), Value::Multirange(b)] => Ok((a, b)),
344        _ => Err(bad_value()),
345    }
346}
347
348/// `multirange_union`: the ranges of the two multiranges, merged.
349fn union_op(call: &Call<'_>, args: &[Value]) -> Result<Value> {
350    let (a, b) = pair(args)?;
351    if a.is_empty() || b.is_empty() {
352        return Ok(Value::Multirange(if a.is_empty() { b } else { a }.to_vec()));
353    }
354    Ok(Value::Multirange(make_multirange(call_range(call), [a, b].concat())?))
355}
356
357/// `multirange_minus_internal`: the values of `a` that are not in `b`. Each range of `a` loses the ranges of `b` that overlap it, in one pass over the two multiranges.
358fn minus_all(range: u32, a: &[Range], b: &[Range]) -> Result<Vec<Range>> {
359    let mut out = Vec::with_capacity(a.len() + b.len());
360    let mut j = 0;
361    for x in a {
362        let mut x = x.clone();
363        while let Some(y) = b.get(j)
364            && before(range, y, &x)?
365        {
366            j += 1;
367        }
368        while let Some(y) = b.get(j) {
369            if let Some((left, right)) = split(range, &x, y)? {
370                out.push(left);
371                x = right;
372                j += 1;
373            } else if overlaps(range, &x, y)? {
374                x = minus(range, &x, y)?;
375                if x.is_empty() || before(range, &x, y)? {
376                    break;
377                }
378                j += 1;
379            } else {
380                break;
381            }
382        }
383        out.push(x);
384    }
385    make_multirange(range, out)
386}
387
388/// `multirange_minus`: the values of the left multirange that are not in the right one.
389fn minus_op(call: &Call<'_>, args: &[Value]) -> Result<Value> {
390    let (a, b) = pair(args)?;
391    if a.is_empty() || b.is_empty() {
392        return Ok(Value::Multirange(a.to_vec()));
393    }
394    Ok(Value::Multirange(minus_all(call_range(call), a, b)?))
395}
396
397/// `multirange_minus_multi`: the difference as a set function, with no row for an empty result.
398pub(crate) fn minus_multi(call: &Call<'_>, args: &[Value]) -> Result<Vec<Vec<Value>>> {
399    let (a, b) = pair(args)?;
400    let out =
401        if a.is_empty() || b.is_empty() { a.to_vec() } else { minus_all(call_range(call), a, b)? };
402    Ok(if out.is_empty() { Vec::new() } else { vec![vec![Value::Multirange(out)]] })
403}
404
405/// `multirange_intersect_internal`: the values that are in the two multiranges, in one pass over them.
406fn intersect_all(range: u32, a: &[Range], b: &[Range]) -> Result<Vec<Range>> {
407    let mut out = Vec::new();
408    let mut j = 0;
409    for x in a {
410        while let Some(y) = b.get(j)
411            && before(range, y, x)?
412        {
413            j += 1;
414        }
415        while let Some(y) = b.get(j) {
416            if !overlaps(range, x, y)? {
417                break;
418            }
419            out.push(intersect(range, x, y)?);
420            if !over_left(range, y, x)? {
421                break;
422            }
423            j += 1;
424        }
425        if j >= b.len() {
426            break;
427        }
428    }
429    make_multirange(range, out)
430}
431
432/// `multirange_intersect`: the values that are in the two multiranges.
433fn intersect_op(call: &Call<'_>, args: &[Value]) -> Result<Value> {
434    let (a, b) = pair(args)?;
435    if a.is_empty() || b.is_empty() {
436        return Ok(Value::Multirange(Vec::new()));
437    }
438    Ok(Value::Multirange(intersect_all(call_range(call), a, b)?))
439}
440
441/// The kernel of a multirange function by its `prosrc`.
442pub(crate) fn by_src(src: &str) -> Option<Kernel> {
443    Some(match src {
444        "multirange_constructor0" => constructor0,
445        "multirange_constructor1" => constructor1,
446        "multirange_constructor2" => constructor2,
447        "multirange_lower" => lower,
448        "multirange_upper" => upper,
449        "multirange_empty" => is_empty,
450        "multirange_lower_inc" => lower_inc,
451        "multirange_upper_inc" => upper_inc,
452        "multirange_lower_inf" => lower_inf,
453        "multirange_upper_inf" => upper_inf,
454        "range_merge_from_multirange" => merge,
455        "multirange_union" => union_op,
456        "multirange_minus" => minus_op,
457        "multirange_intersect" | "multirange_intersect_agg_transfn" => intersect_op,
458        "range_overlaps_multirange"
459        | "multirange_overlaps_range"
460        | "multirange_overlaps_multirange" => overlaps_op,
461        "multirange_contains_range"
462        | "range_contains_multirange"
463        | "multirange_contains_multirange" => contains_op,
464        "range_contained_by_multirange"
465        | "multirange_contained_by_range"
466        | "multirange_contained_by_multirange" => contained_by_op,
467        "multirange_contains_elem" => contains_elem_op,
468        "elem_contained_by_multirange" => elem_contained_op,
469        "range_before_multirange" | "multirange_before_range" | "multirange_before_multirange" => {
470            before_op
471        }
472        "range_after_multirange" | "multirange_after_range" | "multirange_after_multirange" => {
473            after_op
474        }
475        "range_overleft_multirange"
476        | "multirange_overleft_range"
477        | "multirange_overleft_multirange" => over_left_op,
478        "range_adjacent_multirange"
479        | "multirange_adjacent_range"
480        | "multirange_adjacent_multirange" => adjacent_op,
481        // The names of the `&>` functions have the word "over" and the word "right" side by side.
482        _ if src.contains("multirange") && src.contains(concat!("over", "right")) => over_right_op,
483        _ => return None,
484    })
485}
486
487#[cfg(test)]
488mod tests {
489    use rupg_types::oid;
490
491    use super::*;
492
493    fn int4(lower: i32, upper: i32) -> Range {
494        make_range(oid::INT4RANGE, RANGE_LB_INC, Value::Int4(lower), Value::Int4(upper)).unwrap()
495    }
496
497    #[test]
498    fn canonical() {
499        let ranges = vec![int4(8, 10), int4(1, 5), Range::empty(), int4(3, 7), int4(10, 12)];
500        let out = make_multirange(oid::INT4RANGE, ranges).unwrap();
501        assert_eq!(out, [int4(1, 7), int4(8, 12)]);
502        assert!(make_multirange(oid::INT4RANGE, vec![Range::empty()]).unwrap().is_empty());
503    }
504
505    #[test]
506    fn order() {
507        let one = [int4(1, 2)];
508        let two = [int4(1, 2), int4(3, 4)];
509        assert_eq!(multirange_order(oid::INT4, &one, &two).unwrap(), Ordering::Less);
510        assert_eq!(multirange_order(oid::INT4, &[], &[]).unwrap(), Ordering::Equal);
511        assert_eq!(multirange_order(oid::INT4, &[int4(2, 3)], &two).unwrap(), Ordering::Greater);
512    }
513
514    #[test]
515    fn operators() {
516        let r = oid::INT4RANGE;
517        let a = [int4(1, 5), int4(10, 20)];
518        let b = [int4(0, 2), int4(4, 11), int4(15, 16), int4(19, 30)];
519        assert_eq!(minus_all(r, &a, &b).unwrap(), [int4(2, 4), int4(11, 15), int4(16, 19)]);
520        assert_eq!(
521            intersect_all(r, &a, &b).unwrap(),
522            [int4(1, 2), int4(4, 5), int4(10, 11), int4(15, 16), int4(19, 20)]
523        );
524        assert!(overlaps_all(r, &a, &[int4(7, 12)]).unwrap());
525        assert!(!overlaps_all(r, &a, &[int4(5, 10)]).unwrap());
526        assert!(contains_all(r, &a, &[int4(2, 3), int4(12, 14)]).unwrap());
527        assert!(!contains_all(r, &a, &[int4(2, 11)]).unwrap());
528        assert!(contains_all(r, &[], &[]).unwrap());
529    }
530}
rupg_func/multirange.rs

rupg_func/
multirange.rs
