Skip to main content

rudb_vector/
buffer.rs

1//! The run of values behind a flat vector, and the seam the buffer manager arrives through.
2//!
3//! `spec/engine/03-data-plane.md` section 3.8. Today every vector owns its payload and every scan
4//! allocates, which is the right place to start and is not where this ends. At layer three the scan
5//! reads a page out of the buffer manager and the vector wants to point into that page rather than
6//! copy out of it, and the copy it avoids is the largest single copy in the system, because it is
7//! every byte of every column every query reads.
8//!
9//! The type that supports both is one enum holding either an owned run or a borrowed one, and the
10//! part that has to be decided correctly the first time is how the borrow is expressed, because that
11//! shows up in every signature that mentions a vector.
12//!
13//! # Why the pin is a handle and not a lifetime
14//!
15//! The obvious way to express a borrow in Rust is a lifetime parameter, and it is the wrong one
16//! here. A lifetime on [`Buffer`] is a lifetime on [`Data`](crate::Data), which is a lifetime on
17//! [`Vector`](crate::Vector), which is a lifetime on [`Chunk`](crate::Chunk), which is a lifetime on
18//! every operator's state, on every trait object in the pipeline, and on every queue a chunk is put
19//! into for another thread to pick up. The scheduler is exactly that last thing, so the borrow would
20//! have to outlive a hand off between threads that the compiler has no way to see the end of. The
21//! two ways out of that are unsafe code and a copy at the boundary, and the copy at the boundary is
22//! the thing the borrow existed to avoid.
23//!
24//! So the pin is a [`Pin`], a reference counted handle the buffer holds, and the page stays alive
25//! because the handle is alive rather than because a region ends. It costs one atomic increment per
26//! vector construction, which is not measurable next to reading the page it is protecting, and the
27//! ownership story stays uniform: a chunk is `Send`, always, whatever its columns are pointing at.
28//!
29//! # Why it lands now with one variant
30//!
31//! There is no buffer manager, so there is nothing to borrow from, and writing the borrowed variant
32//! now would be writing an interface against an imaginary caller. What lands now is the enum, with
33//! only the owned variant in it, so that adding the second variant at layer three is a change inside
34//! this crate rather than a change to every signature in the workspace. The reader side of that
35//! migration is already done by the [`Deref`] below: everything outside this crate reads a slice, and
36//! a slice is what both variants will hand back.
37//!
38//! The writer side is [`Buffer::to_mut`], which is the one function that has to grow a case. A write
39//! through a borrowed buffer has to copy the page into an owned run first, which is what `Cow` does
40//! and for the same reason, and having the call site named now means that day is a change to one
41//! function rather than a search for every `push`.
42
43use std::any::Any;
44use std::ops::Deref;
45use std::sync::Arc;
46
47/// What keeps a page alive for as long as a buffer points into it.
48///
49/// Opaque on purpose. The vector does not know what a page is and has no business looking inside
50/// one, it only has to hold the thing that stops the page being evicted, and the buffer manager at
51/// layer three decides what that thing is. The bounds are the load bearing part and they are here
52/// now: `Send` and `Sync`, because a chunk carrying one of these crosses a thread boundary every
53/// time the scheduler moves a pipeline, and `'static`, because a handle with a lifetime on it would
54/// have put the lifetime back on [`Vector`](crate::Vector) by another route.
55pub type Pin = Arc<dyn Any + Send + Sync>;
56
57/// A run of values of one physical type.
58///
59/// Derefs to a slice, which is how every reader in the workspace gets at it, so a reader does not
60/// know or care which variant it is holding. Writers go through [`Self::push`] and
61/// [`Self::to_mut`].
62#[derive(Debug, Clone, PartialEq, Eq)]
63pub struct Buffer<T> {
64    store: Store<T>,
65}
66
67/// Where the values actually are.
68///
69/// One variant today. The second is a run inside a page, carrying the [`Pin`] that keeps the page
70/// where it is, and it arrives with the buffer manager because that is the only thing that can hand
71/// one out. It is an enum with one arm rather than a newtype around `Vec<T>` because the shape is
72/// what the rest of the workspace is being compiled against, and a newtype that turns into an enum
73/// later is the flag day this exists to avoid.
74#[derive(Debug, Clone, PartialEq, Eq)]
75enum Store<T> {
76    /// The vector owns the values.
77    Owned(Vec<T>),
78}
79
80impl<T> Buffer<T> {
81    /// An empty buffer.
82    #[must_use]
83    pub fn new() -> Self {
84        Self { store: Store::Owned(Vec::new()) }
85    }
86
87    /// An empty buffer with room for `capacity` values.
88    #[must_use]
89    pub fn with_capacity(capacity: usize) -> Self {
90        Self { store: Store::Owned(Vec::with_capacity(capacity)) }
91    }
92
93    /// A buffer owning `values`.
94    #[must_use]
95    pub fn from_vec(values: Vec<T>) -> Self {
96        Self { store: Store::Owned(values) }
97    }
98
99    /// How many bytes of memory this buffer is holding.
100    ///
101    /// Capacity rather than length, because capacity is what was taken from the allocator. The day
102    /// a buffer can be a run inside a pinned page this stops being the whole story, since the page
103    /// is charged once by whoever pinned it and a hundred buffers over it are charged nothing, and
104    /// this is the one function that has to know the difference.
105    #[must_use]
106    pub fn footprint(&self) -> usize {
107        match &self.store {
108            Store::Owned(values) => values.capacity() * size_of::<T>(),
109        }
110    }
111
112    /// The values.
113    #[must_use]
114    #[inline]
115    pub fn as_slice(&self) -> &[T] {
116        match &self.store {
117            Store::Owned(values) => values,
118        }
119    }
120
121    /// The values as an owned run, copying only if they were not owned already.
122    ///
123    /// Free today because everything is owned. The day it is not, this is one of the two places that
124    /// can copy a page, and it is spelled as a method rather than a field access so that it is
125    /// greppable when that day comes.
126    #[must_use]
127    pub fn into_vec(self) -> Vec<T> {
128        match self.store {
129            Store::Owned(values) => values,
130        }
131    }
132
133    /// The values, writable, copying them out of the page first if they are not owned.
134    ///
135    /// The copy on write point, and the only one. Everything that mutates a buffer goes through
136    /// here, so the borrowed variant needs a case in this function and in nothing else.
137    #[inline]
138    pub fn to_mut(&mut self) -> &mut Vec<T> {
139        match &mut self.store {
140            Store::Owned(values) => values,
141        }
142    }
143
144    /// Appends one value.
145    #[inline]
146    pub fn push(&mut self, value: T) {
147        self.to_mut().push(value);
148    }
149
150    /// Room for `additional` more values, taken in one allocation.
151    pub fn reserve(&mut self, additional: usize) {
152        self.to_mut().reserve(additional);
153    }
154}
155
156impl<T: Clone> Buffer<T> {
157    /// Appends a run of values.
158    pub fn extend_from_slice(&mut self, values: &[T]) {
159        self.to_mut().extend_from_slice(values);
160    }
161}
162
163impl<T> Default for Buffer<T> {
164    fn default() -> Self {
165        Self::new()
166    }
167}
168
169impl<T> Deref for Buffer<T> {
170    type Target = [T];
171
172    #[inline]
173    fn deref(&self) -> &[T] {
174        self.as_slice()
175    }
176}
177
178impl<T> From<Vec<T>> for Buffer<T> {
179    fn from(values: Vec<T>) -> Self {
180        Self::from_vec(values)
181    }
182}
183
184impl<T> FromIterator<T> for Buffer<T> {
185    fn from_iter<I: IntoIterator<Item = T>>(iter: I) -> Self {
186        Self::from_vec(iter.into_iter().collect())
187    }
188}
189
190impl<T> Extend<T> for Buffer<T> {
191    fn extend<I: IntoIterator<Item = T>>(&mut self, iter: I) {
192        self.to_mut().extend(iter);
193    }
194}
195
196impl<T> IntoIterator for Buffer<T> {
197    type Item = T;
198    type IntoIter = std::vec::IntoIter<T>;
199
200    fn into_iter(self) -> Self::IntoIter {
201        self.into_vec().into_iter()
202    }
203}
204
205#[cfg(test)]
206mod tests {
207    use std::sync::Arc;
208
209    use super::{Buffer, Pin};
210
211    #[test]
212    fn a_buffer_reads_back_as_a_slice() {
213        let buffer: Buffer<i32> = vec![1, 2, 3].into();
214        assert_eq!(buffer.as_slice(), &[1, 2, 3]);
215        assert_eq!(buffer.len(), 3);
216        assert_eq!(buffer[1], 2);
217        assert_eq!(buffer.iter().sum::<i32>(), 6);
218        assert_eq!(buffer.clone().into_vec(), vec![1, 2, 3]);
219    }
220
221    #[test]
222    fn writing_goes_through_one_function() {
223        let mut buffer = Buffer::with_capacity(4);
224        buffer.push(1u8);
225        buffer.extend_from_slice(&[2, 3]);
226        buffer.extend([4u8]);
227        buffer.to_mut().sort_unstable_by(|a, b| b.cmp(a));
228        assert_eq!(buffer.as_slice(), &[4, 3, 2, 1]);
229    }
230
231    #[test]
232    fn an_empty_buffer_is_the_default_and_collects_like_a_vector() {
233        assert!(Buffer::<u64>::default().is_empty());
234        assert!(Buffer::<u64>::new().is_empty());
235        let collected: Buffer<u64> = (0..4).collect();
236        assert_eq!(collected.as_slice(), &[0, 1, 2, 3]);
237        assert_eq!(collected.into_iter().count(), 4);
238    }
239
240    /// The property the whole section 3.8 decision is about. A buffer of any payload can be sent to
241    /// another thread without a lifetime being involved, and so can a pin, which is what makes a
242    /// chunk `Send` once the borrowed variant exists. Asserted rather than assumed, because a pin
243    /// that was an `Rc` would compile everywhere else and fail here.
244    #[test]
245    fn a_buffer_and_a_pin_both_cross_a_thread_boundary() {
246        const fn assert_send<T: Send>() {}
247        assert_send::<Buffer<i64>>();
248        assert_send::<Pin>();
249        let pin: Pin = Arc::new(vec![0u8; 8]);
250        let buffer: Buffer<i64> = vec![7; 2].into();
251        let handle = std::thread::spawn(move || (buffer.len(), Arc::strong_count(&pin)));
252        assert_eq!(handle.join().expect("the thread"), (2, 1));
253    }
254}