rupg_common/
ids.rs

1//! The identifiers of spec/04 section 4.7. The file format stores them, so their widths are fixed from M1.
2//!
3//! Each identifier has a text form that parses back (spec/04 section 4.8).
4
5use std::fmt;
6use std::str::FromStr;
7
8use crate::error::Error;
9use crate::sqlstate::SqlState;
10
11fn parse_error(what: &str, text: &str) -> Error {
12    Error::new(SqlState::INVALID_PARAMETER_VALUE, format!("invalid {what}: {text:?}"))
13}
14
15/// An object id, 32 bits, as in PostgreSQL.
16#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
17pub struct Oid(pub u32);
18
19impl Oid {
20    /// The invalid OID, 0.
21    pub const INVALID: Oid = Oid(0);
22    /// The first OID for a user object. Built-in objects have the OIDs of the pin, below this value.
23    pub const FIRST_USER: Oid = Oid(16384);
24}
25
26impl fmt::Display for Oid {
27    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
28        write!(f, "{}", self.0)
29    }
30}
31
32impl FromStr for Oid {
33    type Err = Error;
34
35    fn from_str(s: &str) -> Result<Oid, Error> {
36        s.parse().map(Oid).map_err(|_| parse_error("OID", s))
37    }
38}
39
40/// A shard number, 16 bits. A single node has shard 0.
41#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
42pub struct ShardId(pub u16);
43
44impl fmt::Display for ShardId {
45    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
46        write!(f, "{}", self.0)
47    }
48}
49
50/// A row id, 64 bits: the shard in the high 16 bits and a local number in the low 48 bits.
51///
52/// A row keeps its row id for its life. An update keeps it, and a move from the hot store to the cold store keeps it.
53#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
54pub struct RowId(u64);
55
56impl RowId {
57    /// The largest local number, 2^48 - 1.
58    pub const MAX_LOCAL: u64 = (1 << 48) - 1;
59
60    /// Makes a row id. Returns `None` if `local` does not fit in 48 bits.
61    pub const fn new(shard: ShardId, local: u64) -> Option<RowId> {
62        if local > RowId::MAX_LOCAL {
63            return None;
64        }
65        Some(RowId(((shard.0 as u64) << 48) | local))
66    }
67
68    /// The row id with this stored value.
69    pub const fn from_bits(bits: u64) -> RowId {
70        RowId(bits)
71    }
72
73    /// The stored value.
74    pub const fn bits(self) -> u64 {
75        self.0
76    }
77
78    /// The shard.
79    pub const fn shard(self) -> ShardId {
80        ShardId((self.0 >> 48) as u16)
81    }
82
83    /// The local number.
84    pub const fn local(self) -> u64 {
85        self.0 & RowId::MAX_LOCAL
86    }
87}
88
89impl fmt::Display for RowId {
90    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
91        write!(f, "{}:{}", self.shard(), self.local())
92    }
93}
94
95impl FromStr for RowId {
96    type Err = Error;
97
98    fn from_str(s: &str) -> Result<RowId, Error> {
99        let (shard, local) = s.split_once(':').ok_or_else(|| parse_error("row id", s))?;
100        let shard = shard.parse().map_err(|_| parse_error("row id", s))?;
101        let local = local.parse().map_err(|_| parse_error("row id", s))?;
102        RowId::new(ShardId(shard), local).ok_or_else(|| parse_error("row id", s))
103    }
104}
105
106/// A hybrid logical clock value, 64 bits: 48 bits of milliseconds since 1 January 2020 UTC in the high bits and a 16-bit logical counter in the low bits.
107///
108/// Commit timestamps, snapshots and page timestamps are `Hlc` values. The order of the `u64` is the order of the clock.
109#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
110pub struct Hlc(u64);
111
112impl Hlc {
113    /// The Unix time of the epoch, 2020-01-01 00:00:00 UTC, in milliseconds.
114    pub const EPOCH_UNIX_MS: u64 = 1_577_836_800_000;
115    /// The largest physical part, 2^48 - 1 milliseconds, about 8,900 years.
116    pub const MAX_PHYSICAL: u64 = (1 << 48) - 1;
117    /// The zero value. No commit has it.
118    pub const ZERO: Hlc = Hlc(0);
119    /// The largest value.
120    pub const MAX: Hlc = Hlc(u64::MAX);
121
122    /// Makes a value from its two parts. Returns `None` if `physical` does not fit in 48 bits.
123    pub const fn new(physical: u64, logical: u16) -> Option<Hlc> {
124        if physical > Hlc::MAX_PHYSICAL {
125            return None;
126        }
127        Some(Hlc((physical << 16) | logical as u64))
128    }
129
130    /// The value with these stored bits.
131    pub const fn from_bits(bits: u64) -> Hlc {
132        Hlc(bits)
133    }
134
135    /// The stored bits.
136    pub const fn bits(self) -> u64 {
137        self.0
138    }
139
140    /// Milliseconds since the epoch of 2020.
141    pub const fn physical(self) -> u64 {
142        self.0 >> 16
143    }
144
145    /// The logical counter.
146    pub const fn logical(self) -> u16 {
147        self.0 as u16
148    }
149
150    /// The physical part as a Unix time in milliseconds.
151    pub const fn unix_ms(self) -> u64 {
152        self.physical() + Hlc::EPOCH_UNIX_MS
153    }
154
155    /// Converts a Unix time in milliseconds to the physical part. Times before the epoch become 0.
156    pub const fn physical_from_unix_ms(unix_ms: u64) -> u64 {
157        unix_ms.saturating_sub(Hlc::EPOCH_UNIX_MS)
158    }
159
160    /// The next local value after `self`, for a local event at wall time `wall` (milliseconds since the epoch).
161    ///
162    /// This is the send rule of Kulkarni and others (2014). If the wall clock is ahead, the value takes it with logical 0. Otherwise the logical counter goes up by one, and when it overflows the physical part goes forward by one millisecond.
163    pub fn tick(self, wall: u64) -> Hlc {
164        let wall = wall.min(Hlc::MAX_PHYSICAL);
165        if wall > self.physical() {
166            return Hlc(wall << 16);
167        }
168        Hlc(self.0 + 1)
169    }
170
171    /// The next local value after `self` that is also above `remote`, a value received from another node or ring.
172    pub fn merge(self, remote: Hlc, wall: u64) -> Hlc {
173        self.max(remote).tick(wall)
174    }
175}
176
177impl fmt::Display for Hlc {
178    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
179        write!(f, "{}.{}", self.physical(), self.logical())
180    }
181}
182
183impl FromStr for Hlc {
184    type Err = Error;
185
186    fn from_str(s: &str) -> Result<Hlc, Error> {
187        let (physical, logical) = s.split_once('.').ok_or_else(|| parse_error("HLC value", s))?;
188        let physical = physical.parse().map_err(|_| parse_error("HLC value", s))?;
189        let logical = logical.parse().map_err(|_| parse_error("HLC value", s))?;
190        Hlc::new(physical, logical).ok_or_else(|| parse_error("HLC value", s))
191    }
192}
193
194/// A transaction id, 64 bits, local to the node, assigned at the first write.
195///
196/// PostgreSQL clients see the low 32 bits as `xid` and the full value as `xid8`. PostgreSQL reserves the 32-bit values 0, 1 and 2, so the counter skips each value whose low 32 bits are one of them (spec/11 section 11.9).
197#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
198pub struct Xid(u64);
199
200impl Xid {
201    /// The first id.
202    pub const FIRST: Xid = Xid(3);
203    /// The 32-bit id that PostgreSQL uses for a frozen row.
204    pub const FROZEN_XID: u32 = 2;
205
206    /// The id with this stored value. Returns `None` for a value that the counter skips.
207    pub const fn from_bits(bits: u64) -> Option<Xid> {
208        if (bits as u32) < 3 {
209            return None;
210        }
211        Some(Xid(bits))
212    }
213
214    /// The stored value, which is also `xid8`.
215    pub const fn bits(self) -> u64 {
216        self.0
217    }
218
219    /// The 32-bit `xid`.
220    pub const fn xid32(self) -> u32 {
221        self.0 as u32
222    }
223
224    /// The next id.
225    pub const fn next(self) -> Xid {
226        let mut bits = self.0 + 1;
227        if (bits as u32) < 3 {
228            bits = (bits & !0xffff_ffff) | 3;
229        }
230        Xid(bits)
231    }
232}
233
234impl fmt::Display for Xid {
235    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
236        write!(f, "{}", self.0)
237    }
238}
239
240impl FromStr for Xid {
241    type Err = Error;
242
243    fn from_str(s: &str) -> Result<Xid, Error> {
244        let bits = s.parse().map_err(|_| parse_error("transaction id", s))?;
245        Xid::from_bits(bits).ok_or_else(|| parse_error("transaction id", s))
246    }
247}
248
249/// A `pg_lsn` value. rupg has no single log, so the LSN that clients see is the commit timestamp of the newest durable commit (spec/04 section 4.7). It is monotonic and 64 bits wide.
250#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash)]
251pub struct Lsn(pub u64);
252
253impl From<Hlc> for Lsn {
254    fn from(hlc: Hlc) -> Lsn {
255        Lsn(hlc.bits())
256    }
257}
258
259impl fmt::Display for Lsn {
260    /// The `pg_lsn` text form: the high and low 32 bits in upper case hex, separated by a slash.
261    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
262        write!(f, "{:X}/{:X}", self.0 >> 32, self.0 as u32)
263    }
264}
265
266impl FromStr for Lsn {
267    type Err = Error;
268
269    fn from_str(s: &str) -> Result<Lsn, Error> {
270        let (high, low) = s.split_once('/').ok_or_else(|| parse_error("LSN", s))?;
271        let ok =
272            |p: &str| !p.is_empty() && p.len() <= 8 && p.bytes().all(|b| b.is_ascii_hexdigit());
273        if !ok(high) || !ok(low) {
274            return Err(parse_error("LSN", s));
275        }
276        let high = u64::from_str_radix(high, 16).map_err(|_| parse_error("LSN", s))?;
277        let low = u64::from_str_radix(low, 16).map_err(|_| parse_error("LSN", s))?;
278        Ok(Lsn((high << 32) | low))
279    }
280}
281
282#[cfg(test)]
283mod tests {
284    use super::*;
285
286    #[test]
287    fn row_id_fields() {
288        let id = RowId::new(ShardId(7), 42).unwrap();
289        assert_eq!(id.bits(), (7 << 48) | 42);
290        assert_eq!((id.shard(), id.local()), (ShardId(7), 42));
291        assert_eq!(id.to_string(), "7:42");
292        assert_eq!("7:42".parse::<RowId>().unwrap(), id);
293        assert!(RowId::new(ShardId(0), 1 << 48).is_none());
294        assert!("0:281474976710656".parse::<RowId>().is_err());
295        assert!("65536:1".parse::<RowId>().is_err());
296    }
297
298    #[test]
299    fn hlc_parts_and_order() {
300        let a = Hlc::new(1000, 5).unwrap();
301        assert_eq!((a.physical(), a.logical()), (1000, 5));
302        assert!(Hlc::new(1001, 0).unwrap() > Hlc::new(1000, u16::MAX).unwrap());
303        assert_eq!(a.to_string(), "1000.5");
304        assert_eq!("1000.5".parse::<Hlc>().unwrap(), a);
305        assert!(Hlc::new(1 << 48, 0).is_none());
306        // 2026-10-07 00:00:00 UTC.
307        let unix = 1_791_331_200_000;
308        assert_eq!(Hlc::new(Hlc::physical_from_unix_ms(unix), 0).unwrap().unix_ms(), unix);
309    }
310
311    #[test]
312    fn hlc_tick() {
313        let a = Hlc::new(1000, 5).unwrap();
314        // The wall clock is ahead: take it.
315        assert_eq!(a.tick(2000), Hlc::new(2000, 0).unwrap());
316        // The wall clock is behind or equal: count.
317        assert_eq!(a.tick(1000), Hlc::new(1000, 6).unwrap());
318        assert_eq!(a.tick(10), Hlc::new(1000, 6).unwrap());
319        // The counter overflows into the next millisecond.
320        let full = Hlc::new(1000, u16::MAX).unwrap();
321        assert_eq!(full.tick(1000), Hlc::new(1001, 0).unwrap());
322        // A remote value ahead of both.
323        assert_eq!(a.merge(Hlc::new(5000, 9).unwrap(), 2000), Hlc::new(5000, 10).unwrap());
324    }
325
326    #[test]
327    fn hlc_is_monotonic_under_a_clock_that_jumps() {
328        let mut h = Hlc::ZERO;
329        for wall in [10, 20, 5, 5, 0, 30, 29, 1_000_000, 3] {
330            let next = h.tick(wall);
331            assert!(next > h);
332            h = next;
333        }
334    }
335
336    #[test]
337    fn xid_skips_the_reserved_values() {
338        assert_eq!(Xid::FIRST.xid32(), 3);
339        let last = Xid::from_bits(0xffff_ffff).unwrap();
340        let next = last.next();
341        assert_eq!(next.bits(), (1 << 32) | 3);
342        assert_eq!(next.xid32(), 3);
343        assert!(Xid::from_bits(1 << 32).is_none());
344        assert!(Xid::from_bits(2).is_none());
345        assert_eq!("4294967299".parse::<Xid>().unwrap(), next);
346    }
347
348    #[test]
349    fn lsn_text() {
350        let lsn = Lsn(0x16_B374_D848);
351        assert_eq!(lsn.to_string(), "16/B374D848");
352        assert_eq!("16/B374D848".parse::<Lsn>().unwrap(), lsn);
353        assert_eq!("0/0".parse::<Lsn>().unwrap(), Lsn(0));
354        assert!("16".parse::<Lsn>().is_err());
355        assert!("1/123456789".parse::<Lsn>().is_err());
356        assert_eq!(Lsn::from(Hlc::new(1, 2).unwrap()), Lsn(0x1_0002));
357    }
358}
rupg_common/ids.rs

rupg_common/
ids.rs
