use std::cell::RefCell;
use vibesql_types::SqlValue;
use crate::Row;
const DEFAULT_ROW_CAPACITY: usize = 128;
const DEFAULT_VALUE_CAPACITY: usize = 16;
const MAX_POOLED_BUFFERS: usize = 32;
thread_local! {
static ROW_POOL: RefCell<Vec<Vec<Row>>> = const { RefCell::new(Vec::new()) };
static VALUE_POOL: RefCell<Vec<Vec<SqlValue>>> = const { RefCell::new(Vec::new()) };
}
#[derive(Debug, Clone, Copy, Default)]
pub struct QueryBufferPool;
impl QueryBufferPool {
pub fn new() -> Self {
Self
}
pub fn get_row_buffer(&self, min_capacity: usize) -> Vec<Row> {
ROW_POOL.with(|pool| {
let mut pool = pool.borrow_mut();
if let Some(pos) = pool.iter().position(|buf| buf.capacity() >= min_capacity) {
let mut buffer = pool.swap_remove(pos);
buffer.clear();
buffer
} else {
Vec::with_capacity(min_capacity.max(DEFAULT_ROW_CAPACITY))
}
})
}
pub fn return_row_buffer(&self, mut buffer: Vec<Row>) {
buffer.clear();
ROW_POOL.with(|pool| {
let mut pool = pool.borrow_mut();
if pool.len() < MAX_POOLED_BUFFERS {
pool.push(buffer);
}
});
}
pub fn get_value_buffer(&self, min_capacity: usize) -> Vec<SqlValue> {
VALUE_POOL.with(|pool| {
let mut pool = pool.borrow_mut();
if let Some(pos) = pool.iter().position(|buf| buf.capacity() >= min_capacity) {
let mut buffer = pool.swap_remove(pos);
buffer.clear();
buffer
} else {
Vec::with_capacity(min_capacity.max(DEFAULT_VALUE_CAPACITY))
}
})
}
pub fn return_value_buffer(&self, mut buffer: Vec<SqlValue>) {
buffer.clear();
VALUE_POOL.with(|pool| {
let mut pool = pool.borrow_mut();
if pool.len() < MAX_POOLED_BUFFERS {
pool.push(buffer);
}
});
}
pub fn stats(&self) -> QueryBufferPoolStats {
let row_buffers_pooled = ROW_POOL.with(|pool| pool.borrow().len());
let value_buffers_pooled = VALUE_POOL.with(|pool| pool.borrow().len());
QueryBufferPoolStats { row_buffers_pooled, value_buffers_pooled }
}
pub fn clear_thread_local_pools() {
ROW_POOL.with(|pool| {
pool.borrow_mut().clear();
});
VALUE_POOL.with(|pool| {
pool.borrow_mut().clear();
});
}
}
#[derive(Debug, Clone, Copy)]
pub struct QueryBufferPoolStats {
pub row_buffers_pooled: usize,
pub value_buffers_pooled: usize,
}
pub struct RowBufferGuard {
buffer: Option<Vec<Row>>,
pool: QueryBufferPool,
}
impl RowBufferGuard {
pub fn new(buffer: Vec<Row>, pool: QueryBufferPool) -> Self {
Self { buffer: Some(buffer), pool }
}
pub fn take(mut self) -> Vec<Row> {
self.buffer.take().expect("buffer already taken")
}
#[allow(clippy::should_implement_trait)]
pub fn as_ref(&self) -> &Vec<Row> {
self.buffer.as_ref().expect("buffer already taken")
}
#[allow(clippy::should_implement_trait)]
pub fn as_mut(&mut self) -> &mut Vec<Row> {
self.buffer.as_mut().expect("buffer already taken")
}
}
impl Drop for RowBufferGuard {
fn drop(&mut self) {
if let Some(buffer) = self.buffer.take() {
self.pool.return_row_buffer(buffer);
}
}
}
pub struct ValueBufferGuard {
buffer: Option<Vec<SqlValue>>,
pool: QueryBufferPool,
}
impl ValueBufferGuard {
pub fn new(buffer: Vec<SqlValue>, pool: QueryBufferPool) -> Self {
Self { buffer: Some(buffer), pool }
}
pub fn take(mut self) -> Vec<SqlValue> {
self.buffer.take().expect("buffer already taken")
}
#[allow(clippy::should_implement_trait)]
pub fn as_ref(&self) -> &Vec<SqlValue> {
self.buffer.as_ref().expect("buffer already taken")
}
#[allow(clippy::should_implement_trait)]
pub fn as_mut(&mut self) -> &mut Vec<SqlValue> {
self.buffer.as_mut().expect("buffer already taken")
}
}
impl Drop for ValueBufferGuard {
fn drop(&mut self) {
if let Some(buffer) = self.buffer.take() {
self.pool.return_value_buffer(buffer);
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_row_buffer_pool_reuse() {
let pool = QueryBufferPool::new();
let buffer = pool.get_row_buffer(10);
assert!(buffer.capacity() >= 10);
pool.return_row_buffer(buffer);
let buffer2 = pool.get_row_buffer(10);
assert!(buffer2.capacity() >= 10);
}
#[test]
fn test_value_buffer_pool_reuse() {
let pool = QueryBufferPool::new();
let buffer = pool.get_value_buffer(5);
assert!(buffer.capacity() >= 5);
pool.return_value_buffer(buffer);
let buffer2 = pool.get_value_buffer(5);
assert!(buffer2.capacity() >= 5);
}
#[test]
fn test_buffer_guard_auto_return() {
let pool = QueryBufferPool::new();
{
let buffer = pool.get_row_buffer(10);
let _guard = RowBufferGuard::new(buffer, pool);
}
let stats = pool.stats();
assert_eq!(stats.row_buffers_pooled, 1);
}
#[test]
fn test_pool_max_size() {
let pool = QueryBufferPool::new();
for _ in 0..(MAX_POOLED_BUFFERS + 10) {
let buffer = Vec::with_capacity(10);
pool.return_row_buffer(buffer);
}
let stats = pool.stats();
assert_eq!(stats.row_buffers_pooled, MAX_POOLED_BUFFERS);
}
#[test]
fn test_concurrent_access_thread_safety() {
use std::{sync::Arc, thread};
let pool = Arc::new(QueryBufferPool::new());
let mut handles = vec![];
for _ in 0..4 {
let pool = Arc::clone(&pool);
let handle = thread::spawn(move || {
for _ in 0..100 {
let buffer = pool.get_row_buffer(10);
assert!(buffer.capacity() >= 10);
pool.return_row_buffer(buffer);
}
});
handles.push(handle);
}
for handle in handles {
handle.join().unwrap();
}
let stats = pool.stats();
assert!(stats.row_buffers_pooled <= MAX_POOLED_BUFFERS);
}
}