use crate::compression::{
AdvancedCompressionType, CompressedData, CompressionAlgorithm, CompressionMetadata,
};
use anyhow::{anyhow, Result};
use scirs2_core::metrics::{Counter, Gauge, Histogram};
use std::collections::HashMap;
use std::time::Instant;
pub struct ColumnStoreCompressor {
columns: Vec<ColumnDefinition>,
stats: ColumnStoreStats,
analytics_mode: bool,
}
#[derive(Debug, Clone, Default)]
pub struct ColumnStoreStats {
pub rows_compressed: u64,
pub bytes_compressed: u64,
pub compression_time_us: u64,
pub decompression_time_us: u64,
pub avg_compression_ratio: f64,
pub column_stats: HashMap<String, ColumnStats>,
}
#[derive(Debug, Clone, Default)]
pub struct ColumnStats {
pub value_count: u64,
pub unique_values: u64,
pub null_count: u64,
pub min_value: Option<f64>,
pub max_value: Option<f64>,
pub avg_value_length: f64,
pub compression_ratio: f64,
}
#[derive(Debug, Clone)]
pub struct ColumnDefinition {
pub name: String,
pub data_type: ColumnDataType,
pub compression: ColumnCompressionType,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ColumnDataType {
Int8,
Int16,
Int32,
Int64,
Float32,
Float64,
String,
Binary,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ColumnCompressionType {
None,
RunLength,
Delta,
Dictionary,
FrameOfReference,
Bitmap,
Lz4,
}
#[derive(Debug, Clone)]
pub struct RdfTripleLayout {
pub subject: ColumnDefinition,
pub predicate: ColumnDefinition,
pub object: ColumnDefinition,
}
impl RdfTripleLayout {
pub fn new() -> Self {
Self {
subject: ColumnDefinition {
name: "subject".to_string(),
data_type: ColumnDataType::Int64, compression: ColumnCompressionType::Delta,
},
predicate: ColumnDefinition {
name: "predicate".to_string(),
data_type: ColumnDataType::Int64, compression: ColumnCompressionType::Dictionary, },
object: ColumnDefinition {
name: "object".to_string(),
data_type: ColumnDataType::Int64, compression: ColumnCompressionType::Delta,
},
}
}
pub fn columns(&self) -> Vec<ColumnDefinition> {
vec![
self.subject.clone(),
self.predicate.clone(),
self.object.clone(),
]
}
}
impl Default for RdfTripleLayout {
fn default() -> Self {
Self::new()
}
}
#[derive(Debug, Clone)]
pub struct BatchCompressionConfig {
pub batch_size: usize,
pub parallel: bool,
pub target_ratio: f64,
}
impl ColumnStoreCompressor {
pub fn new() -> Self {
Self {
columns: Vec::new(),
stats: ColumnStoreStats::default(),
analytics_mode: false,
}
}
pub fn with_analytics() -> Self {
Self {
columns: Vec::new(),
stats: ColumnStoreStats::default(),
analytics_mode: true,
}
}
pub fn for_rdf_triples() -> Self {
let layout = RdfTripleLayout::new();
let mut compressor = Self::with_analytics();
for column in layout.columns() {
compressor.add_column(column);
}
compressor
}
pub fn add_column(&mut self, column: ColumnDefinition) {
self.columns.push(column);
}
pub fn get_stats(&self) -> &ColumnStoreStats {
&self.stats
}
pub fn reset_stats(&mut self) {
self.stats = ColumnStoreStats::default();
}
pub fn analyze_column(&mut self, column_name: &str, values: &[Vec<u8>]) {
if values.is_empty() {
return;
}
let mut unique_values = std::collections::HashSet::new();
let mut total_size = 0;
let mut is_sequential = true;
let mut last_value: Option<u64> = None;
for value in values {
unique_values.insert(value.clone());
total_size += value.len();
if value.len() >= 8 {
let current_value = u64::from_le_bytes([
value[0], value[1], value[2], value[3], value[4], value[5], value[6], value[7],
]);
if let Some(last) = last_value {
if current_value != last + 1 {
is_sequential = false;
}
}
last_value = Some(current_value);
} else {
is_sequential = false;
}
}
let uniqueness_ratio = unique_values.len() as f64 / values.len() as f64;
let avg_size = total_size / values.len();
let compression_type = if is_sequential {
ColumnCompressionType::Delta
} else if uniqueness_ratio < 0.1 {
ColumnCompressionType::RunLength
} else if avg_size > 8 && uniqueness_ratio < 0.5 {
ColumnCompressionType::Dictionary
} else {
ColumnCompressionType::None
};
if let Some(column) = self.columns.iter_mut().find(|c| c.name == column_name) {
column.compression = compression_type;
} else {
let data_type = if avg_size <= 1 {
ColumnDataType::Int8
} else if avg_size <= 2 {
ColumnDataType::Int16
} else if avg_size <= 4 {
ColumnDataType::Int32
} else if avg_size <= 8 {
ColumnDataType::Int64
} else {
ColumnDataType::Binary
};
self.add_column(ColumnDefinition {
name: column_name.to_string(),
data_type,
compression: compression_type,
});
}
}
pub fn compress_batch(
&mut self,
rows: &[Vec<u8>],
config: &BatchCompressionConfig,
) -> Result<Vec<u8>> {
if rows.is_empty() {
return Ok(Vec::new());
}
let start = Instant::now();
let mut all_compressed = Vec::new();
for chunk in rows.chunks(config.batch_size) {
let compressed_chunk = self.compress_rows(chunk)?;
all_compressed.extend_from_slice(&(compressed_chunk.len() as u32).to_le_bytes());
all_compressed.extend_from_slice(&compressed_chunk);
}
self.stats.rows_compressed += rows.len() as u64;
self.stats.bytes_compressed += rows.iter().map(|r| r.len()).sum::<usize>() as u64;
self.stats.compression_time_us += start.elapsed().as_micros() as u64;
let original_size: usize = rows.iter().map(|r| r.len()).sum();
let compressed_size = all_compressed.len();
if original_size > 0 {
let ratio = compressed_size as f64 / original_size as f64;
let total_compressions = self.stats.rows_compressed / config.batch_size as u64;
if total_compressions > 0 {
self.stats.avg_compression_ratio =
(self.stats.avg_compression_ratio * (total_compressions - 1) as f64 + ratio)
/ total_compressions as f64;
} else {
self.stats.avg_compression_ratio = ratio;
}
}
Ok(all_compressed)
}
pub fn decompress_batch(&mut self, compressed: &[u8]) -> Result<Vec<Vec<u8>>> {
let start = Instant::now();
let mut all_rows = Vec::new();
let mut offset = 0;
while offset < compressed.len() {
if offset + 4 > compressed.len() {
break;
}
let chunk_size = u32::from_le_bytes([
compressed[offset],
compressed[offset + 1],
compressed[offset + 2],
compressed[offset + 3],
]) as usize;
offset += 4;
if offset + chunk_size > compressed.len() {
return Err(anyhow!("Truncated batch data"));
}
let chunk = &compressed[offset..offset + chunk_size];
let rows = self.decompress_to_rows(chunk)?;
all_rows.extend(rows);
offset += chunk_size;
}
self.stats.decompression_time_us += start.elapsed().as_micros() as u64;
Ok(all_rows)
}
pub fn collect_column_stats(&mut self, column_name: &str, values: &[Vec<u8>]) {
if values.is_empty() {
return;
}
let mut stats = ColumnStats {
value_count: values.len() as u64,
..Default::default()
};
let mut unique_values = std::collections::HashSet::new();
let mut total_length = 0;
let mut min_val = f64::MAX;
let mut max_val = f64::MIN;
for value in values {
unique_values.insert(value.clone());
total_length += value.len();
if value.len() == 8 {
let num_val = f64::from_le_bytes([
value[0], value[1], value[2], value[3], value[4], value[5], value[6], value[7],
]);
if num_val.is_finite() {
min_val = min_val.min(num_val);
max_val = max_val.max(num_val);
}
}
}
stats.unique_values = unique_values.len() as u64;
stats.avg_value_length = total_length as f64 / values.len() as f64;
if min_val != f64::MAX && max_val != f64::MIN {
stats.min_value = Some(min_val);
stats.max_value = Some(max_val);
}
self.stats
.column_stats
.insert(column_name.to_string(), stats);
}
pub fn get_optimal_compression(&self, column_name: &str) -> ColumnCompressionType {
if let Some(stats) = self.stats.column_stats.get(column_name) {
let uniqueness_ratio = stats.unique_values as f64 / stats.value_count as f64;
if uniqueness_ratio < 0.01 {
ColumnCompressionType::Bitmap
} else if uniqueness_ratio < 0.1 {
ColumnCompressionType::Dictionary
} else if let (Some(min), Some(max)) = (stats.min_value, stats.max_value) {
let range = max - min;
if range < 1000.0 {
ColumnCompressionType::FrameOfReference
} else {
ColumnCompressionType::Delta
}
} else {
ColumnCompressionType::Lz4
}
} else {
ColumnCompressionType::None
}
}
pub fn compress_rows(&self, rows: &[Vec<u8>]) -> Result<Vec<u8>> {
if rows.is_empty() {
return Ok(Vec::new());
}
if self.columns.is_empty() {
return Err(anyhow!("No column definitions provided"));
}
let row_count = rows.len();
let mut compressed = Vec::new();
compressed.extend_from_slice(&(row_count as u32).to_le_bytes());
compressed.extend_from_slice(&(self.columns.len() as u32).to_le_bytes());
for (col_idx, column) in self.columns.iter().enumerate() {
let mut column_data = Vec::new();
for row in rows {
let cell_data = self.extract_cell_data(row, col_idx, &column.data_type)?;
column_data.extend_from_slice(&cell_data);
}
let compressed_column = self.compress_column_data(&column_data, column.compression)?;
compressed.extend_from_slice(&(compressed_column.len() as u32).to_le_bytes());
compressed.extend_from_slice(&compressed_column);
}
Ok(compressed)
}
pub fn decompress_to_rows(&self, compressed: &[u8]) -> Result<Vec<Vec<u8>>> {
if compressed.is_empty() {
return Ok(Vec::new());
}
if compressed.len() < 8 {
return Err(anyhow!("Invalid compressed data"));
}
let row_count =
u32::from_le_bytes([compressed[0], compressed[1], compressed[2], compressed[3]])
as usize;
let col_count =
u32::from_le_bytes([compressed[4], compressed[5], compressed[6], compressed[7]])
as usize;
if col_count != self.columns.len() {
return Err(anyhow!("Column count mismatch"));
}
let mut offset = 8;
let mut column_data = Vec::new();
for column in &self.columns {
if offset + 4 > compressed.len() {
return Err(anyhow!("Truncated column data"));
}
let column_size = u32::from_le_bytes([
compressed[offset],
compressed[offset + 1],
compressed[offset + 2],
compressed[offset + 3],
]) as usize;
offset += 4;
if offset + column_size > compressed.len() {
return Err(anyhow!("Truncated column data"));
}
let compressed_column = &compressed[offset..offset + column_size];
let decompressed_column =
self.decompress_column_data(compressed_column, column.compression)?;
column_data.push(decompressed_column);
offset += column_size;
}
let mut rows = Vec::new();
for row_idx in 0..row_count {
let mut row = Vec::new();
for (col_idx, column) in self.columns.iter().enumerate() {
let cell_data =
self.extract_row_cell(&column_data[col_idx], row_idx, &column.data_type)?;
row.extend_from_slice(&cell_data);
}
rows.push(row);
}
Ok(rows)
}
fn extract_cell_data(
&self,
row: &[u8],
col_idx: usize,
data_type: &ColumnDataType,
) -> Result<Vec<u8>> {
let cell_size = match data_type {
ColumnDataType::Int8 => 1,
ColumnDataType::Int16 => 2,
ColumnDataType::Int32 | ColumnDataType::Float32 => 4,
ColumnDataType::Int64 | ColumnDataType::Float64 => 8,
ColumnDataType::String | ColumnDataType::Binary => {
8
}
};
let start = col_idx * cell_size;
if start + cell_size > row.len() {
return Err(anyhow!("Row data too short for column {}", col_idx));
}
Ok(row[start..start + cell_size].to_vec())
}
fn extract_row_cell(
&self,
column_data: &[u8],
row_idx: usize,
data_type: &ColumnDataType,
) -> Result<Vec<u8>> {
let cell_size = match data_type {
ColumnDataType::Int8 => 1,
ColumnDataType::Int16 => 2,
ColumnDataType::Int32 | ColumnDataType::Float32 => 4,
ColumnDataType::Int64 | ColumnDataType::Float64 => 8,
ColumnDataType::String | ColumnDataType::Binary => 8,
};
let start = row_idx * cell_size;
if start + cell_size > column_data.len() {
return Err(anyhow!("Column data too short for row {}", row_idx));
}
Ok(column_data[start..start + cell_size].to_vec())
}
fn compress_column_data(
&self,
data: &[u8],
compression: ColumnCompressionType,
) -> Result<Vec<u8>> {
match compression {
ColumnCompressionType::None => Ok(data.to_vec()),
ColumnCompressionType::RunLength => {
crate::compression::run_length::RunLengthEncoder::encode(data)
}
ColumnCompressionType::Delta => {
crate::compression::delta::DeltaEncoder::encode_byte_sequence(data)
}
_ => {
Ok(data.to_vec())
}
}
}
fn decompress_column_data(
&self,
compressed: &[u8],
compression: ColumnCompressionType,
) -> Result<Vec<u8>> {
match compression {
ColumnCompressionType::None => Ok(compressed.to_vec()),
ColumnCompressionType::RunLength => {
crate::compression::run_length::RunLengthEncoder::decode(compressed)
}
ColumnCompressionType::Delta => {
crate::compression::delta::DeltaEncoder::decode_byte_sequence(compressed)
}
_ => {
Ok(compressed.to_vec())
}
}
}
}
impl Default for ColumnStoreCompressor {
fn default() -> Self {
Self::new()
}
}
impl CompressionAlgorithm for ColumnStoreCompressor {
fn compress(&self, data: &[u8]) -> Result<CompressedData> {
if self.columns.is_empty() {
return Err(anyhow!(
"No column definitions for column store compression"
));
}
let rows = vec![data.to_vec()];
let start = Instant::now();
let compressed_bytes = self.compress_rows(&rows)?;
let compression_time = start.elapsed();
let mut metadata_map = HashMap::new();
metadata_map.insert("columns".to_string(), self.columns.len().to_string());
metadata_map.insert("rows".to_string(), "1".to_string());
let metadata = CompressionMetadata {
algorithm: AdvancedCompressionType::ColumnStore,
original_size: data.len() as u64,
compressed_size: compressed_bytes.len() as u64,
compression_time_us: compression_time.as_micros() as u64,
metadata: metadata_map,
};
Ok(CompressedData {
data: compressed_bytes,
metadata,
})
}
fn decompress(&self, compressed: &CompressedData) -> Result<Vec<u8>> {
if compressed.metadata.algorithm != AdvancedCompressionType::ColumnStore {
return Err(anyhow!(
"Invalid compression algorithm: expected ColumnStore, got {}",
compressed.metadata.algorithm
));
}
let rows = self.decompress_to_rows(&compressed.data)?;
if rows.is_empty() {
Ok(Vec::new())
} else {
Ok(rows[0].clone())
}
}
fn algorithm_type(&self) -> AdvancedCompressionType {
AdvancedCompressionType::ColumnStore
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_column_store_basic() {
let mut compressor = ColumnStoreCompressor::new();
compressor.add_column(ColumnDefinition {
name: "id".to_string(),
data_type: ColumnDataType::Int64,
compression: ColumnCompressionType::None,
});
compressor.add_column(ColumnDefinition {
name: "value".to_string(),
data_type: ColumnDataType::Int64,
compression: ColumnCompressionType::Delta,
});
let test_data = vec![0u8; 16];
let compressed = compressor.compress(&test_data).unwrap();
assert_eq!(
compressed.metadata.algorithm,
AdvancedCompressionType::ColumnStore
);
let decompressed = compressor.decompress(&compressed).unwrap();
assert_eq!(test_data, decompressed);
}
#[test]
fn test_column_definitions() {
let mut compressor = ColumnStoreCompressor::new();
compressor.add_column(ColumnDefinition {
name: "test".to_string(),
data_type: ColumnDataType::Int32,
compression: ColumnCompressionType::RunLength,
});
assert_eq!(compressor.columns.len(), 1);
assert_eq!(compressor.columns[0].name, "test");
}
#[test]
fn test_analytics_mode() {
let compressor = ColumnStoreCompressor::with_analytics();
assert!(compressor.analytics_mode);
assert_eq!(compressor.columns.len(), 0);
}
#[test]
fn test_rdf_triple_layout() {
let layout = RdfTripleLayout::new();
assert_eq!(layout.subject.name, "subject");
assert_eq!(layout.predicate.name, "predicate");
assert_eq!(layout.object.name, "object");
let columns = layout.columns();
assert_eq!(columns.len(), 3);
}
#[test]
fn test_rdf_triple_compressor() {
let compressor = ColumnStoreCompressor::for_rdf_triples();
assert_eq!(compressor.columns.len(), 3);
assert!(compressor.analytics_mode);
assert_eq!(compressor.columns[0].name, "subject");
assert_eq!(compressor.columns[1].name, "predicate");
assert_eq!(compressor.columns[2].name, "object");
}
#[test]
fn test_batch_compression() {
let mut compressor = ColumnStoreCompressor::new();
compressor.add_column(ColumnDefinition {
name: "col1".to_string(),
data_type: ColumnDataType::Int64,
compression: ColumnCompressionType::None,
});
let config = BatchCompressionConfig {
batch_size: 10,
parallel: false,
target_ratio: 0.5,
};
let rows: Vec<Vec<u8>> = (0..20).map(|_| vec![0u8; 8]).collect();
let compressed = compressor.compress_batch(&rows, &config).unwrap();
assert!(!compressed.is_empty());
let decompressed = compressor.decompress_batch(&compressed).unwrap();
assert_eq!(decompressed.len(), 20);
}
#[test]
fn test_column_statistics() {
let mut compressor = ColumnStoreCompressor::new();
let values: Vec<Vec<u8>> = vec![
vec![1, 0, 0, 0, 0, 0, 0, 0],
vec![2, 0, 0, 0, 0, 0, 0, 0],
vec![1, 0, 0, 0, 0, 0, 0, 0], vec![3, 0, 0, 0, 0, 0, 0, 0],
];
compressor.collect_column_stats("test_col", &values);
let stats = compressor.get_stats();
assert!(stats.column_stats.contains_key("test_col"));
let col_stats = &stats.column_stats["test_col"];
assert_eq!(col_stats.value_count, 4);
assert_eq!(col_stats.unique_values, 3); assert_eq!(col_stats.avg_value_length, 8.0);
}
#[test]
fn test_optimal_compression_selection() {
let mut compressor = ColumnStoreCompressor::new();
let low_card_values: Vec<Vec<u8>> = (0..100)
.map(|i| {
let val = (i % 5) as u64; val.to_le_bytes().to_vec()
})
.collect();
compressor.collect_column_stats("low_card", &low_card_values);
let compression = compressor.get_optimal_compression("low_card");
assert!(matches!(
compression,
ColumnCompressionType::Dictionary | ColumnCompressionType::Bitmap
));
let high_card_values: Vec<Vec<u8>> = (0..100)
.map(|i| {
let val = i as u64;
val.to_le_bytes().to_vec()
})
.collect();
compressor.collect_column_stats("high_card", &high_card_values);
let compression = compressor.get_optimal_compression("high_card");
assert!(matches!(
compression,
ColumnCompressionType::Delta | ColumnCompressionType::FrameOfReference
));
}
#[test]
fn test_compression_stats_tracking() {
let mut compressor = ColumnStoreCompressor::new();
compressor.add_column(ColumnDefinition {
name: "col1".to_string(),
data_type: ColumnDataType::Int64,
compression: ColumnCompressionType::None,
});
let config = BatchCompressionConfig {
batch_size: 10,
parallel: false,
target_ratio: 0.5,
};
let rows: Vec<Vec<u8>> = (0..10).map(|_| vec![0u8; 8]).collect();
compressor.compress_batch(&rows, &config).unwrap();
let stats = compressor.get_stats();
assert_eq!(stats.rows_compressed, 10);
assert!(stats.compression_time_us > 0);
}
#[test]
fn test_reset_stats() {
let mut compressor = ColumnStoreCompressor::new();
compressor.add_column(ColumnDefinition {
name: "col1".to_string(),
data_type: ColumnDataType::Int64,
compression: ColumnCompressionType::None,
});
let config = BatchCompressionConfig {
batch_size: 5,
parallel: false,
target_ratio: 0.5,
};
let rows: Vec<Vec<u8>> = (0..5).map(|_| vec![0u8; 8]).collect();
compressor.compress_batch(&rows, &config).unwrap();
assert!(compressor.get_stats().rows_compressed > 0);
compressor.reset_stats();
assert_eq!(compressor.get_stats().rows_compressed, 0);
assert_eq!(compressor.get_stats().compression_time_us, 0);
}
#[test]
fn test_batch_decompress_empty() {
let mut compressor = ColumnStoreCompressor::new();
let result = compressor.decompress_batch(&[]).unwrap();
assert!(result.is_empty());
}
#[test]
fn test_numeric_range_detection() {
let mut compressor = ColumnStoreCompressor::new();
let small_range: Vec<Vec<u8>> = (100..200)
.map(|i| {
let val = i as f64;
val.to_le_bytes().to_vec()
})
.collect();
compressor.collect_column_stats("small_range", &small_range);
let stats = &compressor.get_stats().column_stats["small_range"];
assert!(stats.min_value.is_some());
assert!(stats.max_value.is_some());
assert!((stats.min_value.unwrap() - 100.0).abs() < 0.01);
assert!((stats.max_value.unwrap() - 199.0).abs() < 0.01);
let compression = compressor.get_optimal_compression("small_range");
assert_eq!(compression, ColumnCompressionType::FrameOfReference);
}
#[test]
fn test_column_compression_types() {
assert_ne!(ColumnCompressionType::None, ColumnCompressionType::Delta);
assert_ne!(
ColumnCompressionType::RunLength,
ColumnCompressionType::Dictionary
);
assert_ne!(ColumnCompressionType::Bitmap, ColumnCompressionType::Lz4);
}
#[test]
fn test_batch_config_defaults() {
let config = BatchCompressionConfig {
batch_size: 1000,
parallel: true,
target_ratio: 0.7,
};
assert_eq!(config.batch_size, 1000);
assert!(config.parallel);
assert_eq!(config.target_ratio, 0.7);
}
}