Skip to content

Commit

Permalink
feat: add page data serialize with compression
Browse files Browse the repository at this point in the history
  • Loading branch information
peeeuzin committed Dec 5, 2024
1 parent 2bc0fef commit 1b5364d
Show file tree
Hide file tree
Showing 4 changed files with 71 additions and 8 deletions.
14 changes: 13 additions & 1 deletion src/config.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,8 @@ pub struct WALConfig {
pub struct CompressionConfig {
/// The compression level.
pub level: u32,
/// Enable compression
pub enabled: bool,
}

impl Default for CompressionConfig {
Expand All @@ -38,7 +40,10 @@ impl Default for CompressionConfig {

impl CompressionConfig {
pub fn new() -> Self {
Self { level: 6 }
Self {
level: 6,
enabled: false,
}
}

/// The compression level.
Expand All @@ -47,6 +52,13 @@ impl CompressionConfig {
self.level = level;
self
}

/// Enable compression
/// Default: true
pub fn enabled(&mut self, enabled: bool) -> &mut Self {
self.enabled = enabled;
self
}
}

impl Default for DustDataConfig {
Expand Down
6 changes: 3 additions & 3 deletions src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -46,9 +46,12 @@
//! let user = collection.get("user:1").unwrap();
//! ```
pub mod btree;
pub mod collection;
pub mod config;
pub mod error;
pub mod page;
mod ser_de;

pub use collection::Collection;
pub use config::*;
Expand All @@ -61,9 +64,6 @@ use std::fmt::Debug;
use std::fs;
use std::sync::Arc;

pub mod btree;
pub mod page;

#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)]
pub enum Either<L, R> {
Left(L),
Expand Down
9 changes: 5 additions & 4 deletions src/page.rs
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@ use std::{

use crate::{
error::{CorruptedDataError, CorruptedDataKind, Error, Result},
ser_de::{deserialize, serialize},
Either,
};
use crc32fast::Hasher;
Expand Down Expand Up @@ -78,7 +79,7 @@ impl<T: Serialize + DeserializeOwned + PartialOrd + Ord + Clone> Page<T> {
}

pub fn write(&mut self, data: T) -> Result<(LocationOffset, LocationOffset)> {
let data = bincode::serialize(&data).map_err(Error::SerializeError)?;
let data = serialize(&data)?;

// cell_addr is the position of the cell in the page
let cell_addr: LocationOffset = self.header.upper - data.len() as LocationOffset;
Expand Down Expand Up @@ -130,7 +131,7 @@ impl<T: Serialize + DeserializeOwned + PartialOrd + Ord + Clone> Page<T> {
index: LocationOffset,
data: T,
) -> Result<(LocationOffset, LocationOffset)> {
let data = bincode::serialize(&data).map_err(Error::SerializeError)?;
let data = serialize(&data)?;

let offset = self.index_to_offset(index);

Expand Down Expand Up @@ -188,7 +189,7 @@ impl<T: Serialize + DeserializeOwned + PartialOrd + Ord + Clone> Page<T> {
}

pub fn replace(&mut self, index: LocationOffset, data: T) -> Result<T> {
let data = bincode::serialize(&data).map_err(Error::SerializeError)?;
let data = serialize(&data)?;

let offset = self.index_to_offset(index);

Expand Down Expand Up @@ -259,7 +260,7 @@ impl<T: Serialize + DeserializeOwned + PartialOrd + Ord + Clone> Page<T> {
.map_err(Error::IoError)?;
buffer.read_exact(&mut data).unwrap();

let data = bincode::deserialize(&data).map_err(Error::SerializeError)?;
let data = deserialize(&data)?;

Ok(Some(data))
}
Expand Down
50 changes: 50 additions & 0 deletions src/ser_de.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,50 @@
use crate::{
error::{Error, Result},
CompressionConfig,
};
use flate2::{read::GzDecoder, write::GzEncoder, Compression};
use serde::{de::DeserializeOwned, Serialize};
use std::{
io::{Read, Write},
sync::OnceLock,
};

fn compression_config() -> &'static CompressionConfig {
static COMPRESSION_CONFIG: OnceLock<CompressionConfig> = OnceLock::new();
COMPRESSION_CONFIG.get_or_init(CompressionConfig::default)
}

pub fn serialize<T>(data: &T) -> Result<Vec<u8>>
where
T: Serialize,
{
let mut bytes = bincode::serialize(data).map_err(Error::SerializeError)?;

let compression_config = compression_config();

if compression_config.enabled {
let mut encoder = GzEncoder::new(Vec::new(), Compression::new(compression_config.level));
encoder.write_all(&bytes).map_err(Error::IoError)?;

bytes = encoder.finish().map_err(Error::IoError)?;
}

Ok(bytes)
}

pub fn deserialize<T>(bytes: &[u8]) -> Result<T>
where
T: DeserializeOwned,
{
let compression_config = compression_config();
let mut decoder = GzDecoder::new(bytes);

if compression_config.enabled && decoder.header().is_some() {
let mut buffer = Vec::new();
decoder.read_to_end(&mut buffer).unwrap();

bincode::deserialize(&buffer).map_err(Error::SerializeError)
} else {
bincode::deserialize(bytes).map_err(Error::SerializeError)
}
}

0 comments on commit 1b5364d

Please sign in to comment.