reifydb_core/value/column/buffer/
pool.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::collections::HashMap;
5
6use reifydb_runtime::sync::mutex::Mutex;
7use reifydb_value::value::value_type::ValueType;
8
9use crate::value::column::buffer::ColumnBuffer;
10
11const CAP_PER_TYPE: usize = 64;
12
13pub struct ColumnBufferPool {
14	inner: Mutex<HashMap<ValueType, Vec<ColumnBuffer>>>,
15}
16
17impl Default for ColumnBufferPool {
18	fn default() -> Self {
19		Self::new()
20	}
21}
22
23impl ColumnBufferPool {
24	pub fn new() -> Self {
25		Self {
26			inner: Mutex::new(HashMap::new()),
27		}
28	}
29
30	pub fn acquire(&self, target: &ValueType, min_capacity: usize) -> ColumnBuffer {
31		if is_poolable(target) {
32			let mut pool = self.inner.lock();
33			if let Some(bucket) = pool.get_mut(target) {
34				let mut best_idx: Option<usize> = None;
35				let mut best_cap: usize = usize::MAX;
36				for (i, buf) in bucket.iter().enumerate() {
37					let cap = buf.capacity();
38					if cap >= min_capacity && cap < best_cap {
39						best_cap = cap;
40						best_idx = Some(i);
41					}
42				}
43				if let Some(i) = best_idx {
44					return bucket.swap_remove(i);
45				}
46			}
47		}
48		ColumnBuffer::with_capacity(target.clone(), min_capacity)
49	}
50
51	pub fn release(&self, mut buffer: ColumnBuffer) {
52		let buffer_type = buffer.get_type();
53		if !is_poolable(&buffer_type) {
54			return;
55		}
56		buffer.clear();
57		let mut pool = self.inner.lock();
58		let bucket = pool.entry(buffer_type).or_default();
59		if bucket.len() < CAP_PER_TYPE {
60			bucket.push(buffer);
61		}
62	}
63
64	pub fn len(&self) -> usize {
65		self.inner.lock().values().map(|v| v.len()).sum()
66	}
67
68	pub fn is_empty(&self) -> bool {
69		self.len() == 0
70	}
71}
72
73fn is_poolable(t: &ValueType) -> bool {
74	matches!(
75		t,
76		ValueType::Boolean
77			| ValueType::Float4 | ValueType::Float8
78			| ValueType::Int1 | ValueType::Int2
79			| ValueType::Int4 | ValueType::Int8
80			| ValueType::Int16 | ValueType::Uint1
81			| ValueType::Uint2 | ValueType::Uint4
82			| ValueType::Uint8 | ValueType::Uint16
83			| ValueType::Utf8 | ValueType::Date
84			| ValueType::DateTime | ValueType::Time
85			| ValueType::Duration | ValueType::IdentityId
86			| ValueType::Uuid4 | ValueType::Uuid7
87			| ValueType::Blob | ValueType::Int
88			| ValueType::Uint | ValueType::Decimal
89			| ValueType::DictionaryId
90	)
91}
92
93#[cfg(test)]
94mod tests {
95	use reifydb_value::value::{Value, value_type::ValueType};
96
97	use super::{ColumnBufferPool, is_poolable};
98	use crate::value::column::buffer::ColumnBuffer;
99
100	#[test]
101	fn acquire_from_empty_pool_allocates_fresh() {
102		let pool = ColumnBufferPool::new();
103		let buf = pool.acquire(&ValueType::Int8, 4);
104		assert_eq!(buf.get_type(), ValueType::Int8);
105		assert!(buf.capacity() >= 4);
106		assert!(pool.is_empty());
107	}
108
109	#[test]
110	fn release_then_acquire_reuses_same_allocation() {
111		let pool = ColumnBufferPool::new();
112		let mut buf = ColumnBuffer::with_capacity(ValueType::Int8, 16);
113		// The buffer must be non-empty at release, or the capacity assertion below would pass on a
114		// freshly allocated buffer instead of proving reuse.
115		for i in 0..8i64 {
116			buf.push_value(Value::Int8(i));
117		}
118		let original_capacity = buf.capacity();
119		pool.release(buf);
120		assert_eq!(pool.len(), 1);
121
122		let reused = pool.acquire(&ValueType::Int8, 1);
123		assert_eq!(reused.get_type(), ValueType::Int8);
124		assert_eq!(reused.capacity(), original_capacity);
125		assert_eq!(reused.len(), 0);
126		assert!(pool.is_empty());
127	}
128
129	#[test]
130	fn best_fit_prefers_smallest_qualifying_buffer() {
131		let pool = ColumnBufferPool::new();
132		pool.release(ColumnBuffer::with_capacity(ValueType::Int8, 4));
133		pool.release(ColumnBuffer::with_capacity(ValueType::Int8, 32));
134		pool.release(ColumnBuffer::with_capacity(ValueType::Int8, 16));
135		pool.release(ColumnBuffer::with_capacity(ValueType::Int8, 64));
136		assert_eq!(pool.len(), 4);
137
138		// The 16-capacity buffer is the smallest that satisfies a request for 10; capacity alignment
139		// may round it up, so the upper bound is the next pooled size rather than an exact value.
140		let pick = pool.acquire(&ValueType::Int8, 10);
141		assert!(pick.capacity() >= 10);
142		assert!(pick.capacity() < 32);
143		assert_eq!(pool.len(), 3);
144	}
145
146	#[test]
147	fn release_at_cap_drops_overflow() {
148		let pool = ColumnBufferPool::new();
149		// CAP_PER_TYPE is 64; release 65 to overflow by one.
150		for _ in 0..65 {
151			pool.release(ColumnBuffer::with_capacity(ValueType::Int8, 1));
152		}
153		assert_eq!(pool.len(), 64);
154	}
155
156	#[test]
157	fn buffers_do_not_cross_pollute_across_types() {
158		let pool = ColumnBufferPool::new();
159		pool.release(ColumnBuffer::with_capacity(ValueType::Int8, 16));
160		pool.release(ColumnBuffer::with_capacity(ValueType::Utf8, 16));
161		assert_eq!(pool.len(), 2);
162
163		let int8 = pool.acquire(&ValueType::Int8, 1);
164		assert_eq!(int8.get_type(), ValueType::Int8);
165		assert_eq!(pool.len(), 1);
166
167		let utf8 = pool.acquire(&ValueType::Utf8, 1);
168		assert_eq!(utf8.get_type(), ValueType::Utf8);
169		assert!(pool.is_empty());
170	}
171
172	#[test]
173	fn non_poolable_types_bypass_pool() {
174		let pool = ColumnBufferPool::new();
175		// Option(Int8) is not poolable.
176		let opt_ty = ValueType::Option(Box::new(ValueType::Int8));
177		let opt_buf = ColumnBuffer::with_capacity(opt_ty.clone(), 8);
178		pool.release(opt_buf);
179		assert!(pool.is_empty(), "Option-wrapped buffers must not enter the pool");
180
181		let acquired = pool.acquire(&opt_ty, 4);
182		assert!(acquired.capacity() >= 4);
183		assert!(pool.is_empty());
184	}
185
186	#[test]
187	fn is_poolable_matrix() {
188		assert!(is_poolable(&ValueType::Boolean));
189		assert!(is_poolable(&ValueType::Float4));
190		assert!(is_poolable(&ValueType::Float8));
191		assert!(is_poolable(&ValueType::Int1));
192		assert!(is_poolable(&ValueType::Int2));
193		assert!(is_poolable(&ValueType::Int4));
194		assert!(is_poolable(&ValueType::Int8));
195		assert!(is_poolable(&ValueType::Int16));
196		assert!(is_poolable(&ValueType::Uint1));
197		assert!(is_poolable(&ValueType::Uint2));
198		assert!(is_poolable(&ValueType::Uint4));
199		assert!(is_poolable(&ValueType::Uint8));
200		assert!(is_poolable(&ValueType::Uint16));
201		assert!(is_poolable(&ValueType::Utf8));
202		assert!(is_poolable(&ValueType::Date));
203		assert!(is_poolable(&ValueType::DateTime));
204		assert!(is_poolable(&ValueType::Time));
205		assert!(is_poolable(&ValueType::Duration));
206		assert!(is_poolable(&ValueType::IdentityId));
207		assert!(is_poolable(&ValueType::Uuid4));
208		assert!(is_poolable(&ValueType::Uuid7));
209		assert!(is_poolable(&ValueType::Blob));
210		assert!(is_poolable(&ValueType::Int));
211		assert!(is_poolable(&ValueType::Uint));
212		assert!(is_poolable(&ValueType::Decimal));
213		assert!(is_poolable(&ValueType::DictionaryId));
214
215		assert!(!is_poolable(&ValueType::Option(Box::new(ValueType::Int8))));
216		assert!(!is_poolable(&ValueType::Any));
217		assert!(!is_poolable(&ValueType::List(Box::new(ValueType::Int8))));
218		assert!(!is_poolable(&ValueType::Record(vec![])));
219		assert!(!is_poolable(&ValueType::Tuple(vec![])));
220	}
221}
reifydb_core/value/column/buffer/pool.rs

reifydb_core/value/column/buffer/
pool.rs
