oxirs_tdb/compression/
unified.rs

1//! Unified compression module bringing together all compression capabilities
2//!
3//! This module provides a unified interface to all compression algorithms
4//! supported by oxirs-tdb, with automatic algorithm selection based on
5//! data characteristics.
6//!
7//! ## Supported Algorithms
8//! - **LZ4**: Fast compression/decompression, moderate ratio
9//! - **Zstandard**: High compression ratio, good speed
10//! - **Brotli**: Web-optimized, excellent ratio for text
11//! - **Snappy**: Extremely fast, moderate ratio
12//! - **Prefix**: RDF URI prefix compression
13//! - **Delta**: Delta encoding for sorted numeric data
14//! - **Run-Length**: Run-length encoding for repeated values
15//! - **Bitmap**: Bitmap compression for sparse data
16
17use crate::error::{Result, TdbError};
18use serde::{Deserialize, Serialize};
19
20/// Compression strategy
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub enum CompressionStrategy {
23    /// Optimize for speed
24    Speed,
25    /// Balance speed and compression ratio
26    Balanced,
27    /// Optimize for compression ratio
28    Ratio,
29}
30
31/// Compression algorithm identifier
32#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
33#[repr(u8)]
34pub enum CompressionAlgorithm {
35    /// No compression
36    None = 0,
37    /// LZ4 - fast compression
38    Lz4 = 1,
39    /// Zstandard - high compression ratio
40    Zstd = 2,
41    /// Brotli - web-optimized
42    Brotli = 3,
43    /// Snappy - extremely fast
44    Snappy = 4,
45    /// Prefix compression for URIs
46    Prefix = 5,
47    /// Delta encoding
48    Delta = 6,
49    /// Run-length encoding
50    RunLength = 7,
51    /// Bitmap compression
52    Bitmap = 8,
53}
54
55/// Compression level (0-9, where 9 is maximum)
56#[derive(Debug, Clone, Copy)]
57pub struct CompressionLevel(u8);
58
59impl CompressionLevel {
60    /// Fastest compression (level 1)
61    pub const FAST: Self = Self(1);
62    /// Default compression (level 5)
63    pub const DEFAULT: Self = Self(5);
64    /// Best compression (level 9)
65    pub const BEST: Self = Self(9);
66
67    /// Create a new compression level
68    pub fn new(level: u8) -> Self {
69        Self(level.min(9))
70    }
71
72    /// Get the numeric level
73    pub fn value(&self) -> u8 {
74        self.0
75    }
76}
77
78/// Compression statistics
79#[derive(Debug, Clone, Serialize, Deserialize)]
80pub struct CompressionStats {
81    /// Algorithm used
82    pub algorithm: CompressionAlgorithm,
83    /// Original size in bytes
84    pub original_size: usize,
85    /// Compressed size in bytes
86    pub compressed_size: usize,
87    /// Compression ratio (compressed / original)
88    pub ratio: f64,
89    /// Compression time in microseconds
90    pub compression_time_us: u64,
91    /// Decompression time in microseconds (if measured)
92    pub decompression_time_us: Option<u64>,
93}
94
95impl CompressionStats {
96    /// Calculate compression ratio
97    pub fn new(
98        algorithm: CompressionAlgorithm,
99        original_size: usize,
100        compressed_size: usize,
101        compression_time_us: u64,
102    ) -> Self {
103        let ratio = if original_size > 0 {
104            compressed_size as f64 / original_size as f64
105        } else {
106            1.0
107        };
108
109        Self {
110            algorithm,
111            original_size,
112            compressed_size,
113            ratio,
114            compression_time_us,
115            decompression_time_us: None,
116        }
117    }
118
119    /// Calculate space savings percentage
120    pub fn savings_percent(&self) -> f64 {
121        (1.0 - self.ratio) * 100.0
122    }
123}
124
125/// Unified compression engine
126pub struct UnifiedCompression {
127    /// Default algorithm
128    default_algorithm: CompressionAlgorithm,
129    /// Default compression level
130    default_level: CompressionLevel,
131    /// Adaptive strategy
132    strategy: CompressionStrategy,
133}
134
135impl UnifiedCompression {
136    /// Create a new unified compression engine
137    pub fn new() -> Self {
138        Self {
139            default_algorithm: CompressionAlgorithm::Zstd,
140            default_level: CompressionLevel::DEFAULT,
141            strategy: CompressionStrategy::Balanced,
142        }
143    }
144
145    /// Set default algorithm
146    pub fn with_algorithm(mut self, algorithm: CompressionAlgorithm) -> Self {
147        self.default_algorithm = algorithm;
148        self
149    }
150
151    /// Set default compression level
152    pub fn with_level(mut self, level: CompressionLevel) -> Self {
153        self.default_level = level;
154        self
155    }
156
157    /// Set compression strategy
158    pub fn with_strategy(mut self, strategy: CompressionStrategy) -> Self {
159        self.strategy = strategy;
160        self
161    }
162
163    /// Compress data using the default algorithm
164    pub fn compress(&self, data: &[u8]) -> Result<Vec<u8>> {
165        self.compress_with(data, self.default_algorithm, self.default_level)
166    }
167
168    /// Compress data with a specific algorithm
169    pub fn compress_with(
170        &self,
171        data: &[u8],
172        algorithm: CompressionAlgorithm,
173        level: CompressionLevel,
174    ) -> Result<Vec<u8>> {
175        use std::time::Instant;
176
177        let start = Instant::now();
178
179        let compressed = match algorithm {
180            CompressionAlgorithm::None => data.to_vec(),
181            CompressionAlgorithm::Lz4 => self.compress_lz4(data)?,
182            CompressionAlgorithm::Zstd => self.compress_zstd(data, level.value() as i32)?,
183            CompressionAlgorithm::Brotli => self.compress_brotli(data, level.value() as u32)?,
184            CompressionAlgorithm::Snappy => self.compress_snappy(data)?,
185            CompressionAlgorithm::Prefix => self.compress_prefix(data)?,
186            CompressionAlgorithm::Delta => self.compress_delta(data)?,
187            CompressionAlgorithm::RunLength => self.compress_run_length(data)?,
188            CompressionAlgorithm::Bitmap => self.compress_bitmap(data)?,
189        };
190
191        Ok(compressed)
192    }
193
194    /// Decompress data
195    pub fn decompress(&self, data: &[u8], algorithm: CompressionAlgorithm) -> Result<Vec<u8>> {
196        match algorithm {
197            CompressionAlgorithm::None => Ok(data.to_vec()),
198            CompressionAlgorithm::Lz4 => self.decompress_lz4(data),
199            CompressionAlgorithm::Zstd => self.decompress_zstd(data),
200            CompressionAlgorithm::Brotli => self.decompress_brotli(data),
201            CompressionAlgorithm::Snappy => self.decompress_snappy(data),
202            CompressionAlgorithm::Prefix => self.decompress_prefix(data),
203            CompressionAlgorithm::Delta => self.decompress_delta(data),
204            CompressionAlgorithm::RunLength => self.decompress_run_length(data),
205            CompressionAlgorithm::Bitmap => self.decompress_bitmap(data),
206        }
207    }
208
209    /// Select best algorithm for given data
210    pub fn select_algorithm(&self, data: &[u8]) -> CompressionAlgorithm {
211        // Heuristic-based selection
212        match self.strategy {
213            CompressionStrategy::Speed => CompressionAlgorithm::Snappy,
214            CompressionStrategy::Balanced => {
215                if data.len() < 1024 {
216                    CompressionAlgorithm::Lz4
217                } else {
218                    CompressionAlgorithm::Zstd
219                }
220            }
221            CompressionStrategy::Ratio => CompressionAlgorithm::Brotli,
222        }
223    }
224
225    // Algorithm-specific implementations
226
227    fn compress_lz4(&self, data: &[u8]) -> Result<Vec<u8>> {
228        oxiarc_lz4::compress(data)
229            .map_err(|e| TdbError::Other(format!("LZ4 compression failed: {:?}", e)))
230    }
231
232    fn decompress_lz4(&self, data: &[u8]) -> Result<Vec<u8>> {
233        oxiarc_lz4::decompress(data, 100 * 1024 * 1024)
234            .map_err(|e| TdbError::Other(format!("LZ4 decompression failed: {:?}", e)))
235    }
236
237    fn compress_zstd(&self, data: &[u8], level: i32) -> Result<Vec<u8>> {
238        oxiarc_zstd::encode_all(data, level)
239            .map_err(|e| TdbError::Other(format!("Zstd compression failed: {}", e)))
240    }
241
242    fn decompress_zstd(&self, data: &[u8]) -> Result<Vec<u8>> {
243        oxiarc_zstd::decode_all(data)
244            .map_err(|e| TdbError::Other(format!("Zstd decompression failed: {}", e)))
245    }
246
247    fn compress_brotli(&self, data: &[u8], level: u32) -> Result<Vec<u8>> {
248        // Quality = `level`, default lgwin (22) matches the previous streaming
249        // window size. The on-disk payload is a standard Brotli stream.
250        oxiarc_brotli::compress(data, level)
251            .map_err(|e| TdbError::Other(format!("Brotli compression failed: {}", e)))
252    }
253
254    fn decompress_brotli(&self, data: &[u8]) -> Result<Vec<u8>> {
255        oxiarc_brotli::decompress(data)
256            .map_err(|e| TdbError::Other(format!("Brotli decompression failed: {}", e)))
257    }
258
259    fn compress_snappy(&self, data: &[u8]) -> Result<Vec<u8>> {
260        // Raw Snappy block format (no framing), matching the previous behavior.
261        Ok(oxiarc_snappy::compress(data))
262    }
263
264    fn decompress_snappy(&self, data: &[u8]) -> Result<Vec<u8>> {
265        oxiarc_snappy::decompress(data)
266            .map_err(|e| TdbError::Other(format!("Snappy decompression failed: {}", e)))
267    }
268
269    /// Prefix compression, routed to the dedicated
270    /// [`crate::compression::prefix::PrefixCompressor`] (RDF URI namespace
271    /// prefix extraction, splitting at the last `#` or `/`).
272    ///
273    /// `PrefixCompressor` is designed to amortize a *shared* prefix
274    /// dictionary across many strings compressed with the same `&mut self`
275    /// instance; `UnifiedCompression::compress_with` takes `&self` and
276    /// compresses one opaque byte buffer per call, so there is no dictionary
277    /// to share across calls here. A fresh compressor is used for each call
278    /// and the discovered prefix text (not just its call-local numeric id)
279    /// is written into the output so `decompress_prefix` is fully
280    /// self-contained: `[prefix_len: u32 LE][prefix bytes][suffix bytes]`.
281    fn compress_prefix(&self, data: &[u8]) -> Result<Vec<u8>> {
282        let s = std::str::from_utf8(data).map_err(|e| {
283            TdbError::Other(format!(
284                "Prefix compression requires valid UTF-8 (URI/string) input: {e}"
285            ))
286        })?;
287
288        let mut compressor = crate::compression::prefix::PrefixCompressor::new(4);
289        let compressed = compressor
290            .compress(s)
291            .map_err(|e| TdbError::Other(format!("Prefix compression failed: {e}")))?;
292
293        let prefix_text: &str = if compressed.prefix_id == 0 {
294            ""
295        } else {
296            compressor
297                .get_prefix(compressed.prefix_id)
298                .map(|s| s.as_str())
299                .ok_or_else(|| {
300                    TdbError::Other(
301                        "Prefix compression produced an unresolvable prefix id".to_string(),
302                    )
303                })?
304        };
305
306        let prefix_bytes = prefix_text.as_bytes();
307        let suffix_bytes = compressed.suffix.as_bytes();
308        let mut out = Vec::with_capacity(4 + prefix_bytes.len() + suffix_bytes.len());
309        out.extend_from_slice(&(prefix_bytes.len() as u32).to_le_bytes());
310        out.extend_from_slice(prefix_bytes);
311        out.extend_from_slice(suffix_bytes);
312        Ok(out)
313    }
314
315    fn decompress_prefix(&self, data: &[u8]) -> Result<Vec<u8>> {
316        if data.len() < 4 {
317            return Err(TdbError::Other(
318                "Prefix-compressed data truncated: missing length header".to_string(),
319            ));
320        }
321        let prefix_len = u32::from_le_bytes([data[0], data[1], data[2], data[3]]) as usize;
322        let suffix_start = 4usize.checked_add(prefix_len).ok_or_else(|| {
323            TdbError::Other("Prefix-compressed data: length overflow".to_string())
324        })?;
325        if data.len() < suffix_start {
326            return Err(TdbError::Other(
327                "Prefix-compressed data truncated: prefix shorter than declared length".to_string(),
328            ));
329        }
330
331        let mut out = Vec::with_capacity(data.len() - 4);
332        out.extend_from_slice(&data[4..suffix_start]);
333        out.extend_from_slice(&data[suffix_start..]);
334        Ok(out)
335    }
336
337    /// Delta encoding, routed to the dedicated
338    /// [`crate::compression::delta::DeltaEncoder`] byte-sequence codec
339    /// (good for slowly-varying sorted numeric/byte data).
340    fn compress_delta(&self, data: &[u8]) -> Result<Vec<u8>> {
341        crate::compression::delta::DeltaEncoder::encode_byte_sequence(data)
342            .map_err(|e| TdbError::Other(format!("Delta compression failed: {e}")))
343    }
344
345    fn decompress_delta(&self, data: &[u8]) -> Result<Vec<u8>> {
346        crate::compression::delta::DeltaEncoder::decode_byte_sequence(data)
347            .map_err(|e| TdbError::Other(format!("Delta decompression failed: {e}")))
348    }
349
350    /// Run-length encoding, routed to the dedicated
351    /// [`crate::compression::run_length::RunLengthEncoder`] (good for data
352    /// with long runs of repeated bytes).
353    fn compress_run_length(&self, data: &[u8]) -> Result<Vec<u8>> {
354        crate::compression::run_length::RunLengthEncoder::encode(data)
355            .map_err(|e| TdbError::Other(format!("Run-length compression failed: {e}")))
356    }
357
358    fn decompress_run_length(&self, data: &[u8]) -> Result<Vec<u8>> {
359        crate::compression::run_length::RunLengthEncoder::decode(data)
360            .map_err(|e| TdbError::Other(format!("Run-length decompression failed: {e}")))
361    }
362
363    /// Bitmap encoding, routed to the dedicated
364    /// [`crate::compression::bitmap::BitmapWAHEncoder`] (Word-Aligned
365    /// Hybrid; good for sparse boolean/bitmap data).
366    fn compress_bitmap(&self, data: &[u8]) -> Result<Vec<u8>> {
367        crate::compression::bitmap::BitmapWAHEncoder::encode(data)
368            .map_err(|e| TdbError::Other(format!("Bitmap compression failed: {e}")))
369    }
370
371    fn decompress_bitmap(&self, data: &[u8]) -> Result<Vec<u8>> {
372        crate::compression::bitmap::BitmapWAHEncoder::decode(data)
373            .map_err(|e| TdbError::Other(format!("Bitmap decompression failed: {e}")))
374    }
375
376    /// Benchmark all algorithms on sample data
377    pub fn benchmark(&self, data: &[u8]) -> Vec<CompressionStats> {
378        let mut results = Vec::new();
379
380        let algorithms = vec![
381            CompressionAlgorithm::Lz4,
382            CompressionAlgorithm::Zstd,
383            CompressionAlgorithm::Brotli,
384            CompressionAlgorithm::Snappy,
385        ];
386
387        for algo in algorithms {
388            if let Ok(compressed) = self.compress_with(data, algo, CompressionLevel::DEFAULT) {
389                let stats = CompressionStats::new(
390                    algo,
391                    data.len(),
392                    compressed.len(),
393                    0, // Time not measured in simple benchmark
394                );
395                results.push(stats);
396            }
397        }
398
399        results
400    }
401}
402
403impl Default for UnifiedCompression {
404    fn default() -> Self {
405        Self::new()
406    }
407}
408
409#[cfg(test)]
410mod tests {
411    use super::*;
412
413    #[test]
414    fn test_unified_compression_creation() {
415        let _compression = UnifiedCompression::new();
416    }
417
418    #[test]
419    fn test_compression_level() {
420        assert_eq!(CompressionLevel::FAST.value(), 1);
421        assert_eq!(CompressionLevel::DEFAULT.value(), 5);
422        assert_eq!(CompressionLevel::BEST.value(), 9);
423
424        let custom = CompressionLevel::new(7);
425        assert_eq!(custom.value(), 7);
426
427        // Test clamping
428        let clamped = CompressionLevel::new(15);
429        assert_eq!(clamped.value(), 9);
430    }
431
432    #[test]
433    fn test_lz4_roundtrip() {
434        let compression = UnifiedCompression::new();
435        let data = b"Hello, World! This is a test of LZ4 compression.";
436
437        let compressed = compression
438            .compress_with(data, CompressionAlgorithm::Lz4, CompressionLevel::DEFAULT)
439            .unwrap();
440
441        // Note: For small data, compression may add overhead
442        // The important thing is correctness of roundtrip
443
444        let decompressed = compression
445            .decompress(&compressed, CompressionAlgorithm::Lz4)
446            .unwrap();
447
448        assert_eq!(decompressed, data);
449    }
450
451    #[test]
452    fn test_zstd_roundtrip() {
453        let compression = UnifiedCompression::new();
454        let data = b"Zstandard compression test data with some repetition repetition repetition";
455
456        let compressed = compression
457            .compress_with(data, CompressionAlgorithm::Zstd, CompressionLevel::DEFAULT)
458            .unwrap();
459
460        let decompressed = compression
461            .decompress(&compressed, CompressionAlgorithm::Zstd)
462            .unwrap();
463
464        assert_eq!(decompressed, data);
465    }
466
467    #[test]
468    fn test_snappy_roundtrip() {
469        let compression = UnifiedCompression::new();
470        let data = b"Snappy is designed for speed rather than maximum compression";
471
472        let compressed = compression
473            .compress_with(
474                data,
475                CompressionAlgorithm::Snappy,
476                CompressionLevel::DEFAULT,
477            )
478            .unwrap();
479
480        let decompressed = compression
481            .decompress(&compressed, CompressionAlgorithm::Snappy)
482            .unwrap();
483
484        assert_eq!(decompressed, data);
485    }
486
487    #[test]
488    fn test_brotli_roundtrip() {
489        let compression = UnifiedCompression::new();
490        let data = b"Brotli excels at compressing text and structured web content. ".repeat(20);
491
492        let compressed = compression
493            .compress_with(
494                &data,
495                CompressionAlgorithm::Brotli,
496                CompressionLevel::DEFAULT,
497            )
498            .unwrap();
499
500        let decompressed = compression
501            .decompress(&compressed, CompressionAlgorithm::Brotli)
502            .unwrap();
503        assert_eq!(decompressed, data);
504    }
505
506    #[test]
507    fn test_brotli_roundtrip_incompressible() {
508        // Random/incompressible data must round-trip losslessly through the
509        // standard Brotli on-disk format (no stored fallback).
510        use scirs2_core::random::rng;
511        use scirs2_core::RngExt;
512        let mut r = rng();
513        let data: Vec<u8> = (0..1500).map(|_| r.random_range(0..256) as u8).collect();
514
515        let compression = UnifiedCompression::new();
516        let compressed = compression
517            .compress_with(
518                &data,
519                CompressionAlgorithm::Brotli,
520                CompressionLevel::DEFAULT,
521            )
522            .unwrap();
523        let decompressed = compression
524            .decompress(&compressed, CompressionAlgorithm::Brotli)
525            .unwrap();
526        assert_eq!(decompressed, data);
527    }
528
529    #[test]
530    fn test_algorithm_selection() {
531        let compression = UnifiedCompression::new().with_strategy(CompressionStrategy::Speed);
532        assert_eq!(
533            compression.select_algorithm(b"test"),
534            CompressionAlgorithm::Snappy
535        );
536
537        let compression = UnifiedCompression::new().with_strategy(CompressionStrategy::Ratio);
538        assert_eq!(
539            compression.select_algorithm(b"test"),
540            CompressionAlgorithm::Brotli
541        );
542    }
543
544    #[test]
545    fn test_compression_stats() {
546        let stats = CompressionStats::new(CompressionAlgorithm::Zstd, 1000, 500, 100);
547
548        assert_eq!(stats.ratio, 0.5);
549        assert_eq!(stats.savings_percent(), 50.0);
550    }
551
552    #[test]
553    fn test_benchmark() {
554        let compression = UnifiedCompression::new();
555        let data = b"Sample data for benchmarking compression algorithms. ".repeat(10);
556
557        let results = compression.benchmark(&data);
558
559        assert!(results.len() >= 4); // At least 4 algorithms
560        for result in &results {
561            assert!(result.compressed_size > 0);
562            assert!(result.compressed_size <= result.original_size);
563        }
564    }
565
566    #[test]
567    fn test_none_algorithm() {
568        let compression = UnifiedCompression::new();
569        let data = b"Test data";
570
571        let compressed = compression
572            .compress_with(data, CompressionAlgorithm::None, CompressionLevel::DEFAULT)
573            .unwrap();
574
575        assert_eq!(compressed, data);
576
577        let decompressed = compression
578            .decompress(&compressed, CompressionAlgorithm::None)
579            .unwrap();
580
581        assert_eq!(decompressed, data);
582    }
583
584    /// Regression test: `compress_with`/`decompress` used to return
585    /// `TdbError::Other("Algorithm ... not yet implemented")` for
586    /// Prefix/Delta/RunLength/Bitmap even though dedicated codecs exist for
587    /// all four. They must now actually round-trip through those codecs.
588    #[test]
589    fn test_prefix_algorithm_round_trips() {
590        let compression = UnifiedCompression::new();
591        let data = b"http://example.org/ontology#Person";
592
593        let compressed = compression
594            .compress_with(
595                data,
596                CompressionAlgorithm::Prefix,
597                CompressionLevel::DEFAULT,
598            )
599            .expect("prefix compression should now be implemented");
600        let decompressed = compression
601            .decompress(&compressed, CompressionAlgorithm::Prefix)
602            .expect("prefix decompression should now be implemented");
603        assert_eq!(decompressed, data);
604    }
605
606    #[test]
607    fn test_prefix_algorithm_rejects_non_utf8() {
608        let compression = UnifiedCompression::new();
609        let data: &[u8] = &[0xFF, 0xFE, 0xFD];
610
611        let result = compression.compress_with(
612            data,
613            CompressionAlgorithm::Prefix,
614            CompressionLevel::DEFAULT,
615        );
616        assert!(
617            result.is_err(),
618            "non-UTF-8 input must be a clear error, not silently mangled"
619        );
620    }
621
622    #[test]
623    fn test_delta_algorithm_round_trips() {
624        let compression = UnifiedCompression::new();
625        let data: Vec<u8> = (0..=255u8).collect();
626
627        let compressed = compression
628            .compress_with(
629                &data,
630                CompressionAlgorithm::Delta,
631                CompressionLevel::DEFAULT,
632            )
633            .expect("delta compression should now be implemented");
634        let decompressed = compression
635            .decompress(&compressed, CompressionAlgorithm::Delta)
636            .expect("delta decompression should now be implemented");
637        assert_eq!(decompressed, data);
638    }
639
640    #[test]
641    fn test_run_length_algorithm_round_trips() {
642        let compression = UnifiedCompression::new();
643        let data: Vec<u8> = b"aaaaaaaaaabbbbbbbbbbccccccccccdddddddddd".to_vec();
644
645        let compressed = compression
646            .compress_with(
647                &data,
648                CompressionAlgorithm::RunLength,
649                CompressionLevel::DEFAULT,
650            )
651            .expect("run-length compression should now be implemented");
652        let decompressed = compression
653            .decompress(&compressed, CompressionAlgorithm::RunLength)
654            .expect("run-length decompression should now be implemented");
655        assert_eq!(decompressed, data);
656    }
657
658    #[test]
659    fn test_bitmap_algorithm_round_trips() {
660        let compression = UnifiedCompression::new();
661        let data: Vec<u8> = vec![0u8, 0, 0, 0xFF, 0, 0, 0, 0xFF, 0, 0];
662
663        let compressed = compression
664            .compress_with(
665                &data,
666                CompressionAlgorithm::Bitmap,
667                CompressionLevel::DEFAULT,
668            )
669            .expect("bitmap compression should now be implemented");
670        let decompressed = compression
671            .decompress(&compressed, CompressionAlgorithm::Bitmap)
672            .expect("bitmap decompression should now be implemented");
673        assert_eq!(decompressed, data);
674    }
675}
oxirs_tdb/compression/unified.rs

oxirs_tdb/compression/
unified.rs
