uqa_storage/catalog/
sequence_values.rs

1//
2// Unified Query Algebra
3//
4// Copyright (c) 2023-2026 Cognica, Inc.
5//
6
7//! Sequence value arithmetic shared by every provider, following `PostgreSQL`'s `nextval`.
8
9/// Values a sequence record covers ahead of the ones handed out, as `PostgreSQL`'s `SEQ_LOG_VALS`.
10const LOGGED_AHEAD: i128 = 32;
11
12/// Physical sequence position consumed by one atomic reservation.
13#[derive(Debug, Clone, Copy, PartialEq, Eq)]
14pub struct SequenceValuePosition {
15    pub current: i64,
16    pub called: bool,
17    pub log_count: i64,
18}
19
20/// One atomic sequence reservation. `first_value` is returned immediately, while the remaining values through `last_value` belong to the allocating session.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub struct SequenceValueReservation {
23    pub first_value: i64,
24    pub last_value: i64,
25    pub count: i64,
26    pub log_count: i64,
27}
28
29/// A reservation with the durable position it requires.
30#[derive(Debug, Clone, Copy, PartialEq, Eq)]
31pub struct SequenceValueAllocation {
32    pub reservation: SequenceValueReservation,
33    /// The value the durable record must hold, as called and with no log count, before any value of the reservation is handed out. It is set when the reservation uses up the values an earlier record covered, and lies up to 32 fetches past the reservation, where `PostgreSQL` writes it to its log. A sequence that loses its exact position continues after this value, so no value at or below it is handed out twice.
34    pub logged_value: Option<i64>,
35}
36
37impl SequenceValueAllocation {
38    /// The value the durable record must hold, as called and with no log count, before the reservation is handed out, or `None` when the record covers the reservation already. `position` is the position the reservation continues. `covering` is the value of the record known to cover that position, or `None` when the position was read from a record, which covers nothing past itself.
39    #[must_use]
40    pub fn record_value(
41        &self,
42        position: SequenceValuePosition,
43        covering: Option<i64>,
44        increment: i64,
45        min_value: i64,
46        max_value: i64,
47    ) -> Option<i64> {
48        if self.logged_value.is_some() {
49            return self.logged_value;
50        }
51        let reservation = self.reservation;
52        let ascending = increment > 0;
53        // A reservation that starts over from the opposite bound leaves every earlier record behind it.
54        let wrapped = position.called
55            && if ascending {
56                reservation.first_value <= position.current
57            } else {
58                reservation.first_value >= position.current
59            };
60        let covered = covering.is_some_and(|covering| {
61            !wrapped
62                && if ascending {
63                    reservation.last_value <= covering
64                } else {
65                    reservation.last_value >= covering
66                }
67        });
68        if covered {
69            return None;
70        }
71        // Each reserved value uses up one of the log count, so the count that remains reaches the value `PostgreSQL`'s log holds for this position.
72        let ahead = i128::from(reservation.last_value)
73            + i128::from(increment) * i128::from(reservation.log_count);
74        let bounded = ahead.clamp(i128::from(min_value), i128::from(max_value));
75        Some(i64::try_from(bounded).expect("a value within the sequence bounds"))
76    }
77}
78
79#[derive(Debug, Clone, Copy, PartialEq, Eq)]
80pub enum SequenceReservationResult {
81    Missing,
82    DefinitionChanged,
83    Exhausted,
84    Reserved(SequenceValueReservation),
85}
86
87/// A value update applies only to the definition whose bounds the caller validated.
88#[derive(Debug, Clone, Copy, PartialEq, Eq)]
89pub enum SequenceSetValueResult {
90    Missing,
91    DefinitionChanged,
92    Set(i64),
93}
94
95/// The outcome of moving a sequence's durable position from an expected one.
96#[derive(Debug, Clone, Copy, PartialEq, Eq)]
97pub enum SequenceLogResult {
98    Missing,
99    DefinitionChanged,
100    /// The record holds this position instead of the expected one; nothing was written.
101    Changed(SequenceValuePosition),
102    Logged,
103}
104
105/// Reserve up to `cache_size` values without crossing a sequence bound. Cycling is applied when selecting the first value of a new reservation, matching `PostgreSQL`'s boundary-truncated cache blocks.
106#[must_use]
107pub fn sequence_value_reservation(
108    position: SequenceValuePosition,
109    increment: i64,
110    min_value: i64,
111    max_value: i64,
112    cycle: bool,
113    cache_size: i64,
114) -> Option<SequenceValueReservation> {
115    sequence_value_allocation(position, increment, min_value, max_value, cycle, cache_size)
116        .map(|allocation| allocation.reservation)
117}
118
119/// [`sequence_value_reservation`] with the durable position the reservation requires.
120#[must_use]
121pub fn sequence_value_allocation(
122    position: SequenceValuePosition,
123    increment: i64,
124    min_value: i64,
125    max_value: i64,
126    cycle: bool,
127    cache_size: i64,
128) -> Option<SequenceValueAllocation> {
129    let SequenceValuePosition {
130        current,
131        called,
132        log_count,
133    } = position;
134    debug_assert_ne!(increment, 0);
135    debug_assert!(cache_size > 0);
136    let first_value = if called {
137        match current
138            .checked_add(increment)
139            .filter(|value| (min_value..=max_value).contains(value))
140        {
141            Some(value) => value,
142            None if cycle && increment > 0 => min_value,
143            None if cycle => max_value,
144            None => return None,
145        }
146    } else {
147        current
148    };
149    let distance = if increment > 0 {
150        i128::from(max_value) - i128::from(first_value)
151    } else {
152        i128::from(first_value) - i128::from(min_value)
153    };
154    let step = i128::from(increment).abs();
155    let available = distance / step + 1;
156    let count = available.min(i128::from(cache_size));
157    let last_value = i128::from(first_value) + i128::from(increment) * (count - 1);
158    let initial_count = i128::from(!called);
159    let cache_fetch = i128::from(cache_size) - initial_count;
160    let mut fetch = cache_fetch;
161    let mut next_log_count = i128::from(log_count);
162    let logged = i128::from(log_count) < cache_fetch || !called;
163    if logged {
164        fetch += LOGGED_AHEAD;
165        next_log_count = fetch;
166    }
167    let fetched = fetch.min(available - initial_count);
168    next_log_count -= fetched.min(cache_fetch);
169    next_log_count -= fetch - fetched;
170    // The fetches advance from the first value, which an uncalled sequence returns without advancing.
171    let logged_value = logged.then(|| {
172        let ahead = i128::from(first_value) + i128::from(increment) * (fetched - 1 + initial_count);
173        i64::try_from(ahead).expect("logged sequence value stays in bounds")
174    });
175    Some(SequenceValueAllocation {
176        reservation: SequenceValueReservation {
177            first_value,
178            last_value: i64::try_from(last_value).expect("reserved sequence value stays in bounds"),
179            count: i64::try_from(count).expect("reservation count cannot exceed cache size"),
180            log_count: i64::try_from(next_log_count)
181                .expect("persisted sequence log count cannot exceed the cache request"),
182        },
183        logged_value,
184    })
185}
186
187#[cfg(test)]
188mod tests;
uqa_storage/catalog/sequence_values.rs

uqa_storage/catalog/
sequence_values.rs
