Skip to content
Merged
2 changes: 1 addition & 1 deletion .mise/tasks/sqlite-compress/test
Original file line number Diff line number Diff line change
@@ -1,3 +1,3 @@
#!/usr/bin/env bash
#MISE description="Run tests"
cargo test -p sqlite-compress
cargo test -p sqlite-compress -- --test-threads=1
12 changes: 3 additions & 9 deletions sqlite-compress/src/cache/connection.rs
Original file line number Diff line number Diff line change
@@ -1,28 +1,22 @@
use std::collections::HashMap;

use zstd::dict::{DecoderDictionary, EncoderDictionary};

use crate::{ColumnKey, DictId, cache::lru::DictLru};
/// Per-connection cache of prepared zstd dictionaries and the current dict id per column.
use crate::{ColumnKey, DictId};
/// Process-wide map of the current dict id per column (shared across all connections via [`crate::Cache`]).
///
/// `encoders` and `decoders` are independent LRUs keyed by [`DictKey`] (schema + dict id).
/// Prepared encoders/decoders live in the separate process-wide pool, not here.
///
/// `current_ids` maps a [`ColumnKey`] to the dict id that `compress` should use. It is
/// filled by `warm_cache` (latest row in `__compress_dicts`) and `insert_into_caches`
/// (after training). `clear_current_id` drops the entry when that column has no dictionary.
/// A missing id is treated as raw zstd (`DictId` 0).
pub(crate) struct DictCache {
current_ids: HashMap<ColumnKey, DictId>,
pub(crate) encoders: DictLru<EncoderDictionary<'static>>,
pub(crate) decoders: DictLru<DecoderDictionary<'static>>,
}

impl DictCache {
pub(crate) fn new() -> Self {
Self {
current_ids: HashMap::new(),
encoders: DictLru::new(),
decoders: DictLru::new(),
}
}

Expand Down
124 changes: 0 additions & 124 deletions sqlite-compress/src/cache/lru.rs

This file was deleted.

2 changes: 1 addition & 1 deletion sqlite-compress/src/cache/mod.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
mod connection;
mod lru;
mod pool;
mod prepared;

pub(crate) use connection::DictCache;
Expand Down
92 changes: 92 additions & 0 deletions sqlite-compress/src/cache/pool.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,92 @@
use std::{
collections::{HashMap, hash_map::Entry},
sync::{Arc, LazyLock},
};

use parking_lot::Mutex;
use zstd::dict::{DecoderDictionary, EncoderDictionary};

use crate::{dict::DictKey, functions::Level};

const MAX_PREPARED_DICTS: usize = 64;

type PreparedMap<T> = HashMap<DictKey, Arc<T>>;
Comment thread
Michelprogram marked this conversation as resolved.

static ENCODERS: LazyLock<Mutex<PreparedMap<EncoderDictionary<'static>>>> =
LazyLock::new(|| Mutex::new(HashMap::with_capacity(MAX_PREPARED_DICTS)));

static DECODERS: LazyLock<Mutex<PreparedMap<DecoderDictionary<'static>>>> =
LazyLock::new(|| Mutex::new(HashMap::with_capacity(MAX_PREPARED_DICTS)));

pub(crate) fn get_encoder(
key: &DictKey,
level: Level,
load: impl FnOnce() -> Result<Vec<u8>, crate::DictError>,
) -> Result<Arc<EncoderDictionary<'static>>, crate::DictError> {
if let Some(dict) = ENCODERS.lock().get(key).cloned() {
return Ok(dict);
}

let raw = load()?;

let mut cache = ENCODERS.lock();
if let Some(dict) = cache.get(key).cloned() {
return Ok(dict);
}

let encoder = Arc::new(EncoderDictionary::copy(&raw, level.get()));
insert_capped(&mut cache, key.clone(), encoder.clone());
Ok(encoder)
}

pub(crate) fn get_decoder(
key: &DictKey,
load: impl FnOnce() -> Result<Vec<u8>, crate::DictError>,
) -> Result<Arc<DecoderDictionary<'static>>, crate::DictError> {
if let Some(dict) = DECODERS.lock().get(key).cloned() {
return Ok(dict);
}

let raw = load()?;

let mut cache = DECODERS.lock();
if let Some(dict) = cache.get(key).cloned() {
return Ok(dict);
}

let decoder = Arc::new(DecoderDictionary::copy(&raw));
insert_capped(&mut cache, key.clone(), decoder.clone());
Ok(decoder)
}

fn insert_capped<T>(map: &mut PreparedMap<T>, key: DictKey, value: Arc<T>) {
let at_cap = map.len() >= MAX_PREPARED_DICTS;

match map.entry(key) {
Entry::Occupied(mut e) => {
e.insert(value);
}
Entry::Vacant(e) => {
if at_cap {
let key = e.into_key();
map.clear();

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This clears the whole map at cap, you sure it is what we want?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yep the downside of this is that you need to fill it again with a lot of cache misses but i don't think we'll be doing clear that much. i mean the cost of having an LRU instead is acceptable

map.insert(key, value);
} else {
e.insert(value);
}
}
}
}

pub(crate) fn insert_prepared(key: DictKey, dictionary: &[u8], level: Level) {
insert_capped(
&mut ENCODERS.lock(),
key.clone(),
Arc::new(EncoderDictionary::copy(dictionary, level.get())),
);
insert_capped(
&mut DECODERS.lock(),
key,
Arc::new(DecoderDictionary::copy(dictionary)),
);
}
51 changes: 5 additions & 46 deletions sqlite-compress/src/cache/prepared.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@ use zstd::dict::{DecoderDictionary, EncoderDictionary};

use crate::{
Cache, ColumnKey, DictError, DictId,
cache::pool,
dict::{DictKey, get_raw_dict, read_current_id},
functions::Level,
state::ExtensionState,
Expand All @@ -18,22 +19,7 @@ pub(crate) fn get_encoder_in_cache(
) -> Result<Arc<EncoderDictionary<'static>>, DictError> {
let key = DictKey::new(schema, dict_id);

if let Some(dict) = state.cache.lock().encoders.peek(&key) {
return Ok(dict);
}

let raw = get_raw_dict(&key, state)?;

let mut cache = state.cache.lock();

if let Some(dict) = cache.encoders.get(&key) {
return Ok(dict);
}

let encoder = Arc::new(EncoderDictionary::copy(&raw, level.get()));
cache.encoders.insert(key, encoder.clone());

Ok(encoder)
pool::get_encoder(&key, level, || get_raw_dict(&key, state))
}

/// Return the prepared zstd decoder for `(schema, dict_id)`, loading it from `__compress_dicts` and caching it on a miss.
Expand All @@ -44,22 +30,7 @@ pub(crate) fn get_decoder_in_cache(
) -> Result<Arc<DecoderDictionary<'static>>, DictError> {
let key = DictKey::new(schema, dict_id);

if let Some(dict) = state.cache.lock().decoders.peek(&key) {
return Ok(dict);
}

let raw = get_raw_dict(&key, state)?;

let mut cache = state.cache.lock();

if let Some(dict) = cache.decoders.get(&key) {
return Ok(dict);
}

let decoder = Arc::new(DecoderDictionary::copy(&raw));
cache.decoders.insert(key, decoder.clone());

Ok(decoder)
pool::get_decoder(&key, || get_raw_dict(&key, state))
}

/// Load this column's current dict id, encoder, and decoder. Call it as early as possible so later `compress` and `decompress` hit the cache.
Expand Down Expand Up @@ -91,20 +62,8 @@ pub(crate) fn insert_into_caches(
dictionary: &[u8],
level: Level,
) {
let mut cache = cache.lock();

cache.set_current_id(column.clone(), dict_id);

let key = DictKey::new(column.schema(), dict_id);

cache.encoders.insert(
key.clone(),
Arc::new(EncoderDictionary::copy(dictionary, level.get())),
);

cache
.decoders
.insert(key, Arc::new(DecoderDictionary::copy(dictionary)));
cache.lock().set_current_id(column.clone(), dict_id);
pool::insert_prepared(DictKey::new(column.schema(), dict_id), dictionary, level);
}

#[cfg(test)]
Expand Down
20 changes: 12 additions & 8 deletions sqlite-compress/src/state.rs
Original file line number Diff line number Diff line change
@@ -1,16 +1,22 @@
use std::sync::Arc;
use std::sync::{Arc, LazyLock};

use libsqlite3_sys::sqlite3;
use parking_lot::Mutex;
use sqlite_ffi::{Connection, Context, SqliteError};

use crate::cache::DictCache;

/// Per-connection state for the extension: the SQLite connection and its dictionary cache.
// Shared cache for all connections.
static SHARED: LazyLock<Cache> = LazyLock::new(|| Cache(Arc::new(Mutex::new(DictCache::new()))));

/// Per-connection handle: this SQLite connection plus a clone of the process-wide dictionary cache.
///
/// `sqlite3_compress_init` calls [`ExtensionState::attach`], which stores an empty [`Cache`] in the connection's client data.
/// `setup`, `train_all`, `compress`, and `decompress` read that same cache back with [`ExtensionState::from_db`] or [`ExtensionState::from_context`].
/// The cache is an `Arc`, so every lookup shares the encoders, decoders, and current dict ids for this connection.
/// `sqlite3_compress_init` calls [`ExtensionState::attach`], which stores a clone of the shared
/// [`Cache`] in the connection's client data (not a fresh empty cache).
/// `setup`, `train_all`, `compress`, and `decompress` read it back with [`ExtensionState::from_db`]
/// or [`ExtensionState::from_context`].
/// All connections therefore share the same `current_ids`; prepared encoders/decoders live in the
/// separate process-wide pool.
pub struct ExtensionState {
pub(crate) connection: Connection,
pub(crate) cache: Cache,
Expand All @@ -35,9 +41,7 @@ impl ExtensionState {
const NAME: &str = "sqlite-compress";

pub fn attach(connection: &Connection) -> Result<(), SqliteError> {
let cache = Cache(Arc::new(Mutex::new(DictCache::new())));

connection.set_client_data(Self::NAME, cache)
connection.set_client_data(Self::NAME, SHARED.clone())
}

pub fn from_db(db: *mut sqlite3) -> Result<Self, SqliteError> {
Expand Down
Loading
Loading