kacrab_protocol/compression.rs
1//! Record-batch compression codecs.
2//!
3//! The [`Compression`] enum is always available so [`crate::record::RecordBatch`]
4//! can read the compression type from `attributes` regardless of which codec
5//! features are enabled. Actual compress/decompress operations require the
6//! corresponding Cargo feature:
7//!
8//! | Feature | Codec | Backend | Pure Rust | HC mode |
9//! |----------|--------|---------------------|-----------|---------|
10//! | `gzip` | Gzip | `flate2` | yes | n/a |
11//! | `snappy` | Snappy | `snap` | yes | n/a |
12//! | `lz4` | Lz4 | `lz4_flex` | yes | no |
13//! | `lz4-hc` | Lz4 | `lz4` (C-FFI) | no | yes |
14//! | `zstd` | Zstd | `zstd` (C-FFI) | no | n/a |
15//!
16//! `compression` is a meta-feature that enables `gzip + snappy + lz4 + zstd`
17//! (pure-Rust LZ4). Opt into `lz4-hc` explicitly if you need
18//! High-Compression mode — see the LZ4 section below.
19//!
20//! ## LZ4 backend selection
21//!
22//! * **`lz4`** — `lz4_flex` block API behind a Kafka-compatible custom frame. Fast mode only; the
23//! `level` argument is ignored. Pure Rust, no C toolchain at build time, cross-compile clean.
24//! * **`lz4-hc`** — `lz4` crate (FFI to liblz4) behind the same frame format. Levels 3..=12 use HC
25//! mode; levels 0..=2 fall back to fast mode. Requires a C compiler at build time.
26//!
27//! Both features are independent — enabling `lz4-hc` alone is sufficient
28//! (the C lib handles fast mode too). When **both** are enabled, `lz4-hc`
29//! wins at runtime; `lz4_flex` is linked but unused. Most users should
30//! pick one.
31//!
32//! For high compression ratio in Kafka workloads, prefer **`zstd`** over
33//! `lz4-hc` — modern Kafka clients (Java, librdkafka) default to fast LZ4
34//! when LZ4 is selected, and switch to zstd when ratio matters.
35
36#[cfg(feature = "gzip")]
37pub mod gzip;
38#[cfg(any(feature = "lz4", feature = "lz4-hc"))]
39pub mod lz4;
40#[cfg(feature = "snappy")]
41pub mod snappy;
42#[cfg(feature = "zstd")]
43pub mod zstd;
44
45pub mod error;
46
47pub use self::error::{CompressionError, CompressionErrorKind};
48
49/// Result alias for compression operations.
50pub type Result<T> = core::result::Result<T, CompressionError>;
51
52/// Kafka record-batch compression codec, encoded in bits 0–2 of the batch
53/// `attributes` field.
54#[derive(Debug, Clone, Copy, PartialEq, Eq)]
55#[repr(i16)]
56#[non_exhaustive]
57pub enum Compression {
58 /// No compression.
59 None = 0,
60 /// Gzip (`flate2`).
61 Gzip = 1,
62 /// Snappy (`snap`).
63 Snappy = 2,
64 /// LZ4 frame format (`lz4_flex`).
65 Lz4 = 3,
66 /// Zstandard (`zstd` crate).
67 Zstd = 4,
68}
69
70impl Compression {
71 /// Decode the codec from a record-batch `attributes` field.
72 /// Only bits 0–2 are examined; higher bits are masked off.
73 pub const fn from_attributes(attributes: i16) -> Result<Self> {
74 match attributes & 0x07 {
75 0 => Ok(Self::None),
76 1 => Ok(Self::Gzip),
77 2 => Ok(Self::Snappy),
78 3 => Ok(Self::Lz4),
79 4 => Ok(Self::Zstd),
80 n => Err(CompressionError::new(
81 Self::None,
82 CompressionErrorKind::UnknownCodec(n),
83 )),
84 }
85 }
86
87 /// Compress `payload` at the codec default level.
88 pub fn compress(self, payload: &[u8]) -> Result<Vec<u8>> {
89 self.compress_with_level(payload, None)
90 }
91
92 /// Compress `payload` at the given level (codec-specific).
93 ///
94 /// * `Gzip` accepts `0..=9`, default `6`.
95 /// * `Zstd` accepts `1..=22`, default `3`.
96 /// * `Snappy` ignores the level.
97 /// * `Lz4` level handling depends on the active backend feature:
98 /// * with `lz4` only — level ignored, always fast mode.
99 /// * with `lz4-hc` — level `0..=2` → fast, `3..=12` → HC mode, values above `12` clamp to
100 /// `12`, negative/zero levels clamp to fast level `1`.
101 pub fn compress_with_level(self, payload: &[u8], level: Option<i32>) -> Result<Vec<u8>> {
102 match self {
103 Self::None => {
104 let _ = level;
105 Ok(payload.to_vec())
106 },
107 #[cfg(feature = "gzip")]
108 Self::Gzip => gzip::compress_with_level(payload, level),
109 #[cfg(feature = "snappy")]
110 Self::Snappy => snappy::compress_with_level(payload, level),
111 #[cfg(any(feature = "lz4", feature = "lz4-hc"))]
112 Self::Lz4 => lz4::compress_with_level(payload, level),
113 #[cfg(feature = "zstd")]
114 Self::Zstd => zstd::compress_with_level(payload, level),
115 #[cfg(not(feature = "gzip"))]
116 Self::Gzip => Err(CompressionError::new(
117 self,
118 CompressionErrorKind::CodecDisabled,
119 )),
120 #[cfg(not(feature = "snappy"))]
121 Self::Snappy => Err(CompressionError::new(
122 self,
123 CompressionErrorKind::CodecDisabled,
124 )),
125 #[cfg(not(any(feature = "lz4", feature = "lz4-hc")))]
126 Self::Lz4 => Err(CompressionError::new(
127 self,
128 CompressionErrorKind::CodecDisabled,
129 )),
130 #[cfg(not(feature = "zstd"))]
131 Self::Zstd => Err(CompressionError::new(
132 self,
133 CompressionErrorKind::CodecDisabled,
134 )),
135 }
136 }
137
138 /// Decompress `payload`. Returns the decompressed payload, or an error if
139 /// the required Cargo feature is not enabled.
140 pub fn decompress(self, payload: &[u8]) -> Result<Vec<u8>> {
141 match self {
142 Self::None => Ok(payload.to_vec()),
143 #[cfg(feature = "gzip")]
144 Self::Gzip => gzip::decompress(payload),
145 #[cfg(feature = "snappy")]
146 Self::Snappy => snappy::decompress(payload),
147 #[cfg(any(feature = "lz4", feature = "lz4-hc"))]
148 Self::Lz4 => lz4::decompress(payload),
149 #[cfg(feature = "zstd")]
150 Self::Zstd => zstd::decompress(payload),
151 #[cfg(not(feature = "gzip"))]
152 Self::Gzip => Err(CompressionError::new(
153 self,
154 CompressionErrorKind::CodecDisabled,
155 )),
156 #[cfg(not(feature = "snappy"))]
157 Self::Snappy => Err(CompressionError::new(
158 self,
159 CompressionErrorKind::CodecDisabled,
160 )),
161 #[cfg(not(any(feature = "lz4", feature = "lz4-hc")))]
162 Self::Lz4 => Err(CompressionError::new(
163 self,
164 CompressionErrorKind::CodecDisabled,
165 )),
166 #[cfg(not(feature = "zstd"))]
167 Self::Zstd => Err(CompressionError::new(
168 self,
169 CompressionErrorKind::CodecDisabled,
170 )),
171 }
172 }
173}