cqlite_core/storage/sstable/bti/
encoder.rs

1//! CEP-25 Compliant Byte-comparable key encoding for BTI format
2//!
3//! Converts CQL keys to byte sequences where lexicographic comparison
4//! of unsigned bytes produces the same result as typed comparison.
5//!
6//! This implementation follows the CEP-25 specification for Cassandra 5.0
7//! byte-comparable key encoding used in trie-indexed SSTables.
8//!
9//! Key Requirements:
10//! - Lexicographic byte comparison must match typed comparison
11//! - Support for all Cassandra 5.0 data types  
12//! - Proper null value handling with type prefixes
13//! - Variable-length encoding for efficiency
14//! - Nested type support (collections, UDTs, tuples)
15//! - Escape sequences for proper ordering
16
17use super::BtiError;
18use crate::error::Result;
19use crate::types::{UdtValue, Value};
20
21/// CEP-25 compliant byte-comparable key encoder
22///
23/// Implements the byte-comparable key encoding as specified in CEP-25.
24/// The encoding ensures that lexicographic comparison of the encoded bytes
25/// produces the same ordering as the typed comparison of the original values.
26pub struct ByteComparableEncoder {
27    /// Buffer for building encoded keys
28    buffer: Vec<u8>,
29    /// Configuration for encoding behavior
30    config: EncoderConfig,
31}
32
33/// Configuration for the byte-comparable encoder
34#[derive(Debug, Clone)]
35pub struct EncoderConfig {
36    /// Enable variable-length integer encoding
37    pub use_varint_encoding: bool,
38    /// Maximum depth for nested types to prevent infinite recursion
39    pub max_nesting_depth: usize,
40    /// Enable prefix compression for collections
41    pub enable_prefix_compression: bool,
42    /// Strict CEP-25 compliance mode
43    pub strict_compliance: bool,
44}
45
46impl Default for EncoderConfig {
47    fn default() -> Self {
48        Self {
49            use_varint_encoding: true,
50            max_nesting_depth: 32,
51            enable_prefix_compression: true,
52            strict_compliance: true,
53        }
54    }
55}
56
57/// Type prefixes for byte-comparable encoding as per CEP-25
58mod type_prefixes {
59    pub const NULL: u8 = 0x00;
60    pub const BOOLEAN_FALSE: u8 = 0x01;
61    pub const BOOLEAN_TRUE: u8 = 0x02;
62    pub const TINYINT: u8 = 0x10;
63    pub const SMALLINT: u8 = 0x11;
64    pub const INTEGER: u8 = 0x12;
65    pub const BIGINT: u8 = 0x13;
66    pub const FLOAT: u8 = 0x20;
67    pub const DOUBLE: u8 = 0x21;
68    #[allow(dead_code)]
69    pub const DECIMAL: u8 = 0x22;
70    #[allow(dead_code)]
71    pub const VARINT: u8 = 0x23;
72    pub const TEXT: u8 = 0x30;
73    pub const BLOB: u8 = 0x31;
74    pub const UUID: u8 = 0x40;
75    pub const TIMESTAMP: u8 = 0x41;
76    #[allow(dead_code)]
77    pub const DATE: u8 = 0x42;
78    #[allow(dead_code)]
79    pub const TIME: u8 = 0x43;
80    #[allow(dead_code)]
81    pub const DURATION: u8 = 0x44;
82    pub const LIST: u8 = 0x50;
83    pub const SET: u8 = 0x51;
84    pub const MAP: u8 = 0x52;
85    pub const TUPLE: u8 = 0x60;
86    pub const UDT: u8 = 0x61;
87    pub const FROZEN: u8 = 0x70;
88    #[allow(dead_code)]
89    pub const TOMBSTONE: u8 = 0x80;
90    #[allow(dead_code)]
91    pub const ESCAPE: u8 = 0xFF;
92    pub const SEPARATOR: u8 = 0x00;
93    pub const TERMINATOR: u8 = 0x01;
94}
95
96/// Escape sequences for special bytes in values
97mod escape_sequences {
98    #[allow(dead_code)]
99    pub const ESCAPE_BYTE: u8 = 0xFF;
100    #[allow(dead_code)]
101    pub const ESCAPED_NULL: &[u8] = &[0xFF, 0x00];
102    #[allow(dead_code)]
103    pub const ESCAPED_ESCAPE: &[u8] = &[0xFF, 0xFF];
104    #[allow(dead_code)]
105    pub const ESCAPED_SEPARATOR: &[u8] = &[0xFF, 0x01];
106}
107
108impl Default for ByteComparableEncoder {
109    fn default() -> Self {
110        Self::new()
111    }
112}
113
114impl ByteComparableEncoder {
115    /// Create new encoder with default configuration
116    pub fn new() -> Self {
117        Self {
118            buffer: Vec::new(),
119            config: EncoderConfig::default(),
120        }
121    }
122
123    /// Create new encoder with custom configuration
124    pub fn with_config(config: EncoderConfig) -> Self {
125        Self {
126            buffer: Vec::new(),
127            config,
128        }
129    }
130
131    /// Get current configuration
132    pub fn config(&self) -> &EncoderConfig {
133        &self.config
134    }
135
136    /// Set new configuration
137    pub fn set_config(&mut self, config: EncoderConfig) {
138        self.config = config;
139    }
140
141    /// Encode a single value to byte-comparable format
142    pub fn encode_value(&mut self, value: &Value) -> Result<Vec<u8>> {
143        self.buffer.clear();
144        self.encode_value_to_buffer(value)?;
145        Ok(self.buffer.clone())
146    }
147
148    /// Encode a composite key (multiple values) to byte-comparable format
149    pub fn encode_composite_key(&mut self, values: &[Value]) -> Result<Vec<u8>> {
150        self.buffer.clear();
151
152        for (i, value) in values.iter().enumerate() {
153            if i > 0 {
154                // Add separator byte between key components
155                self.buffer.push(0x00);
156            }
157            self.encode_value_to_buffer(value)?;
158        }
159
160        Ok(self.buffer.clone())
161    }
162
163    /// Encode value directly to internal buffer with depth tracking
164    fn encode_value_to_buffer(&mut self, value: &Value) -> Result<()> {
165        self.encode_value_to_buffer_with_depth(value, 0)
166    }
167
168    /// Encode value with nesting depth tracking
169    fn encode_value_to_buffer_with_depth(&mut self, value: &Value, depth: usize) -> Result<()> {
170        if depth > self.config.max_nesting_depth {
171            return Err(BtiError::InvalidByteComparableKey(format!(
172                "Maximum nesting depth {} exceeded",
173                self.config.max_nesting_depth
174            ))
175            .into());
176        }
177
178        match value {
179            Value::Null => self.encode_null(),
180            Value::Boolean(b) => self.encode_boolean(*b),
181            Value::TinyInt(i) => self.encode_tinyint(*i),
182            Value::SmallInt(i) => self.encode_smallint(*i),
183            Value::Integer(i) => self.encode_int(*i),
184            Value::BigInt(i) => self.encode_bigint(*i),
185            Value::Counter(c) => self.encode_bigint(*c), // Counter encoded as bigint
186            Value::Float32(f) => self.encode_float32(*f),
187            Value::Float(f) => self.encode_double(*f),
188            Value::Text(s) => self
189                .encode_text(std::str::from_utf8(s).map_err(|e| {
190                    crate::Error::corruption(format!("invalid UTF-8 in text: {e}"))
191                })?),
192            Value::Blob(bytes) => self.encode_blob(bytes),
193            Value::Uuid(uuid) => self.encode_uuid_bytes(uuid),
194            Value::Timestamp(ts) => self.encode_timestamp(*ts),
195            Value::Json(json) => self.encode_json(json),
196            Value::List(items) => self.encode_list_with_depth(items, depth + 1),
197            Value::Set(items) => self.encode_set_with_depth(items, depth + 1),
198            Value::Map(map) => self.encode_map_with_depth(map, depth + 1),
199            Value::Tuple(items) => self.encode_tuple_with_depth(items, depth + 1),
200            Value::Udt(udt) => self.encode_udt_with_depth(udt, depth + 1),
201            Value::Frozen(inner) => {
202                self.buffer.push(type_prefixes::FROZEN);
203                self.encode_value_to_buffer_with_depth(inner, depth + 1)
204            }
205            Value::Varint(data) => {
206                // For BTI encoding, treat varint as blob since we don't have the original value
207                self.encode_blob(data)
208            }
209            Value::Decimal { scale: _, unscaled } => self.encode_blob(unscaled), // Treat decimal as blob for BTI
210            Value::Duration {
211                months,
212                days,
213                nanos,
214            } => {
215                // Encode duration as a composite key
216                self.buffer.push(type_prefixes::DURATION);
217                self.encode_int(*months)?;
218                self.encode_int(*days)?;
219                self.encode_bigint(*nanos)?;
220                Ok(())
221            }
222            Value::Tombstone(_) => {
223                // Tombstones are encoded as null with special marker
224                self.buffer.push(type_prefixes::NULL);
225                self.buffer.push(0xFF); // Special tombstone marker
226                Ok(())
227            }
228            Value::Date(d) => {
229                self.buffer.push(type_prefixes::DATE);
230                self.encode_int(*d)
231            }
232            Value::Time(t) => {
233                self.buffer.push(type_prefixes::TIME);
234                self.encode_bigint(*t)
235            }
236            Value::Inet(bytes) => self.encode_blob(bytes),
237        }
238    }
239
240    /// Encode null value with proper type prefix
241    fn encode_null(&mut self) -> Result<()> {
242        self.buffer.push(type_prefixes::NULL);
243        Ok(())
244    }
245
246    /// Encode text/varchar with proper UTF-8 ordering and escape sequences
247    fn encode_text(&mut self, text: &str) -> Result<()> {
248        self.buffer.push(type_prefixes::TEXT);
249
250        // Encode UTF-8 bytes with proper escaping
251        for &byte in text.as_bytes() {
252            match byte {
253                0x00 => self
254                    .buffer
255                    .extend_from_slice(escape_sequences::ESCAPED_NULL),
256                0xFF => self
257                    .buffer
258                    .extend_from_slice(escape_sequences::ESCAPED_ESCAPE),
259                _ => self.buffer.push(byte),
260            }
261        }
262
263        // Add terminator for proper ordering
264        self.buffer.push(type_prefixes::TERMINATOR);
265        Ok(())
266    }
267
268    /// Encode JSON as escaped text
269    fn encode_json(&mut self, json: &serde_json::Value) -> Result<()> {
270        let json_str = json.to_string();
271        self.encode_text(&json_str)
272    }
273
274    /// Encode tinyint (i8) with proper ordering
275    fn encode_tinyint(&mut self, value: i8) -> Result<()> {
276        self.buffer.push(type_prefixes::TINYINT);
277        // Transform to unsigned for proper lexicographic ordering
278        let unsigned = (value as i16 + 128) as u8;
279        self.buffer.push(unsigned);
280        Ok(())
281    }
282
283    /// Encode smallint (i16) with proper ordering
284    fn encode_smallint(&mut self, value: i16) -> Result<()> {
285        self.buffer.push(type_prefixes::SMALLINT);
286        // Transform to unsigned for proper lexicographic ordering
287        let unsigned = (value as i32 + 32768) as u16;
288        self.buffer.extend_from_slice(&unsigned.to_be_bytes());
289        Ok(())
290    }
291
292    /// Encode integer (i32) with sign-magnitude encoding
293    fn encode_int(&mut self, value: i32) -> Result<()> {
294        self.buffer.push(type_prefixes::INTEGER);
295
296        // Use two's complement transformation for proper ordering
297        // Transform signed to unsigned preserving order: flip sign bit
298        let unsigned = (value as u32) ^ 0x8000_0000;
299
300        self.buffer.extend_from_slice(&unsigned.to_be_bytes());
301        Ok(())
302    }
303
304    /// Encode bigint (i64) with sign-magnitude encoding
305    fn encode_bigint(&mut self, value: i64) -> Result<()> {
306        self.buffer.push(type_prefixes::BIGINT);
307
308        // Use two's complement transformation for proper ordering
309        let unsigned = if value >= 0 {
310            (value as u64) + 0x8000_0000_0000_0000
311        } else {
312            (value as u64) ^ 0xFFFF_FFFF_FFFF_FFFF
313        };
314
315        self.buffer.extend_from_slice(&unsigned.to_be_bytes());
316        Ok(())
317    }
318
319    /// Encode UUID bytes with proper byte ordering
320    fn encode_uuid_bytes(&mut self, uuid: &[u8; 16]) -> Result<()> {
321        self.buffer.push(type_prefixes::UUID);
322        // UUID bytes in network byte order are naturally comparable
323        self.buffer.extend_from_slice(uuid);
324        Ok(())
325    }
326
327    /// Encode timestamp (microseconds since epoch)
328    fn encode_timestamp(&mut self, timestamp: i64) -> Result<()> {
329        self.buffer.push(type_prefixes::TIMESTAMP);
330
331        // Transform for proper ordering (timestamps can be negative)
332        let unsigned = if timestamp >= 0 {
333            (timestamp as u64) + 0x8000_0000_0000_0000
334        } else {
335            (timestamp as u64) ^ 0xFFFF_FFFF_FFFF_FFFF
336        };
337
338        self.buffer.extend_from_slice(&unsigned.to_be_bytes());
339        Ok(())
340    }
341
342    /// Encode boolean with proper type prefixes
343    fn encode_boolean(&mut self, value: bool) -> Result<()> {
344        if value {
345            self.buffer.push(type_prefixes::BOOLEAN_TRUE);
346        } else {
347            self.buffer.push(type_prefixes::BOOLEAN_FALSE);
348        }
349        Ok(())
350    }
351
352    /// Encode float32 with IEEE 754 ordering adjustment
353    fn encode_float32(&mut self, value: f32) -> Result<()> {
354        self.buffer.push(type_prefixes::FLOAT);
355
356        // Handle special values first
357        if value.is_nan() {
358            // NaN sorts after all other values
359            self.buffer.extend_from_slice(&[0xFF, 0xFF, 0xFF, 0xFF]);
360            return Ok(());
361        }
362
363        let bits = value.to_bits();
364
365        // Adjust for proper ordering of IEEE 754 floats
366        let adjusted = if (bits & 0x8000_0000) == 0 {
367            // Positive: add sign bit offset
368            bits | 0x8000_0000
369        } else {
370            // Negative: flip all bits
371            !bits
372        };
373
374        self.buffer.extend_from_slice(&adjusted.to_be_bytes());
375        Ok(())
376    }
377
378    /// Encode double (f64) with IEEE 754 ordering adjustment
379    fn encode_double(&mut self, value: f64) -> Result<()> {
380        self.buffer.push(type_prefixes::DOUBLE);
381
382        // Handle special values first
383        if value.is_nan() {
384            // NaN sorts after all other values
385            self.buffer
386                .extend_from_slice(&[0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF]);
387            return Ok(());
388        }
389
390        let bits = value.to_bits();
391
392        // Adjust for proper ordering of IEEE 754 floats
393        let adjusted = if (bits & 0x8000_0000_0000_0000) == 0 {
394            // Positive: add sign bit offset
395            bits | 0x8000_0000_0000_0000
396        } else {
397            // Negative: flip all bits
398            !bits
399        };
400
401        self.buffer.extend_from_slice(&adjusted.to_be_bytes());
402        Ok(())
403    }
404
405    /// Encode blob (binary data) with proper escaping
406    fn encode_blob(&mut self, bytes: &[u8]) -> Result<()> {
407        self.buffer.push(type_prefixes::BLOB);
408
409        // Encode raw bytes with proper escaping
410        for &byte in bytes {
411            match byte {
412                0x00 => self
413                    .buffer
414                    .extend_from_slice(escape_sequences::ESCAPED_NULL),
415                0xFF => self
416                    .buffer
417                    .extend_from_slice(escape_sequences::ESCAPED_ESCAPE),
418                _ => self.buffer.push(byte),
419            }
420        }
421
422        // Add terminator
423        self.buffer.push(type_prefixes::TERMINATOR);
424        Ok(())
425    }
426
427    /// Encode list with element-by-element encoding and depth tracking
428    fn encode_list_with_depth(&mut self, items: &[Value], depth: usize) -> Result<()> {
429        self.buffer.push(type_prefixes::LIST);
430
431        // Encode length as varint if configured
432        if self.config.use_varint_encoding {
433            self.encode_varint(items.len() as u64)?;
434        } else {
435            self.buffer
436                .extend_from_slice(&(items.len() as u32).to_be_bytes());
437        }
438
439        // Encode each element with separator
440        for (i, item) in items.iter().enumerate() {
441            if i > 0 {
442                self.buffer.push(type_prefixes::SEPARATOR);
443            }
444            self.encode_value_to_buffer_with_depth(item, depth)?;
445        }
446
447        // Add terminator
448        self.buffer.push(type_prefixes::TERMINATOR);
449        Ok(())
450    }
451
452    /// Backward compatibility wrapper
453    #[allow(dead_code)]
454    fn encode_list(&mut self, items: &[Value]) -> Result<()> {
455        self.encode_list_with_depth(items, 1)
456    }
457
458    /// Encode set (sorted for deterministic ordering) with depth tracking
459    fn encode_set_with_depth(&mut self, items: &[Value], _depth: usize) -> Result<()> {
460        self.buffer.push(type_prefixes::SET);
461
462        // For byte-comparable encoding, we need to sort the encoded items
463        let mut encoded_items = Vec::new();
464
465        for item in items {
466            let mut encoder = ByteComparableEncoder::with_config(self.config.clone());
467            let encoded = encoder.encode_value(item)?;
468            encoded_items.push(encoded);
469        }
470
471        // Sort encoded items lexicographically for deterministic ordering
472        encoded_items.sort();
473
474        // Encode length
475        if self.config.use_varint_encoding {
476            self.encode_varint(encoded_items.len() as u64)?;
477        } else {
478            self.buffer
479                .extend_from_slice(&(encoded_items.len() as u32).to_be_bytes());
480        }
481
482        // Add sorted encoded items with separators
483        for (i, encoded_item) in encoded_items.iter().enumerate() {
484            if i > 0 {
485                self.buffer.push(type_prefixes::SEPARATOR);
486            }
487            self.buffer.extend_from_slice(encoded_item);
488        }
489
490        // Add terminator
491        self.buffer.push(type_prefixes::TERMINATOR);
492        Ok(())
493    }
494
495    /// Backward compatibility wrapper
496    #[allow(dead_code)]
497    fn encode_set(&mut self, items: &[Value]) -> Result<()> {
498        self.encode_set_with_depth(items, 1)
499    }
500
501    /// Encode map from Vec of tuples with sorted key-value pairs and depth tracking
502    fn encode_map_with_depth(&mut self, map: &Vec<(Value, Value)>, _depth: usize) -> Result<()> {
503        self.buffer.push(type_prefixes::MAP);
504
505        // Encode key-value pairs and sort by encoded keys
506        let mut encoded_pairs = Vec::new();
507
508        for (key, value) in map {
509            let mut key_encoder = ByteComparableEncoder::with_config(self.config.clone());
510            let encoded_key = key_encoder.encode_value(key)?;
511
512            let mut value_encoder = ByteComparableEncoder::with_config(self.config.clone());
513            let encoded_value = value_encoder.encode_value(value)?;
514
515            encoded_pairs.push((encoded_key, encoded_value));
516        }
517
518        // Sort by encoded keys for deterministic ordering
519        encoded_pairs.sort_by(|a, b| a.0.cmp(&b.0));
520
521        // Encode length
522        if self.config.use_varint_encoding {
523            self.encode_varint(encoded_pairs.len() as u64)?;
524        } else {
525            self.buffer
526                .extend_from_slice(&(encoded_pairs.len() as u32).to_be_bytes());
527        }
528
529        // Add sorted pairs with separators
530        for (i, (encoded_key, encoded_value)) in encoded_pairs.iter().enumerate() {
531            if i > 0 {
532                self.buffer.push(type_prefixes::SEPARATOR);
533            }
534
535            // Encode key-value pair
536            self.buffer.extend_from_slice(encoded_key);
537            self.buffer.push(type_prefixes::SEPARATOR);
538            self.buffer.extend_from_slice(encoded_value);
539        }
540
541        // Add terminator
542        self.buffer.push(type_prefixes::TERMINATOR);
543        Ok(())
544    }
545
546    /// Backward compatibility wrapper
547    #[allow(dead_code)]
548    fn encode_map_vec(&mut self, map: &Vec<(Value, Value)>) -> Result<()> {
549        self.encode_map_with_depth(map, 1)
550    }
551
552    /// Encode tuple with positional fields and depth tracking
553    fn encode_tuple_with_depth(&mut self, items: &[Value], depth: usize) -> Result<()> {
554        self.buffer.push(type_prefixes::TUPLE);
555
556        // Encode length
557        if self.config.use_varint_encoding {
558            self.encode_varint(items.len() as u64)?;
559        } else {
560            self.buffer
561                .extend_from_slice(&(items.len() as u32).to_be_bytes());
562        }
563
564        // Encode each field with separator (order is significant for tuples)
565        for (i, item) in items.iter().enumerate() {
566            if i > 0 {
567                self.buffer.push(type_prefixes::SEPARATOR);
568            }
569            self.encode_value_to_buffer_with_depth(item, depth)?;
570        }
571
572        // Add terminator
573        self.buffer.push(type_prefixes::TERMINATOR);
574        Ok(())
575    }
576
577    /// Encode UDT (User Defined Type) with field ordering and depth tracking
578    fn encode_udt_with_depth(&mut self, udt: &UdtValue, depth: usize) -> Result<()> {
579        self.buffer.push(type_prefixes::UDT);
580
581        // Encode type name and keyspace for disambiguation
582        self.encode_text(&udt.keyspace)?;
583        self.encode_text(&udt.type_name)?;
584
585        // Encode field count
586        if self.config.use_varint_encoding {
587            self.encode_varint(udt.fields.len() as u64)?;
588        } else {
589            self.buffer
590                .extend_from_slice(&(udt.fields.len() as u32).to_be_bytes());
591        }
592
593        // Encode fields in schema order (important for UDTs)
594        for (i, field) in udt.fields.iter().enumerate() {
595            if i > 0 {
596                self.buffer.push(type_prefixes::SEPARATOR);
597            }
598
599            // Encode field name
600            self.encode_text(&field.name)?;
601            self.buffer.push(type_prefixes::SEPARATOR);
602
603            // Encode field value (null if None)
604            match &field.value {
605                Some(value) => self.encode_value_to_buffer_with_depth(value, depth)?,
606                None => self.encode_null()?,
607            }
608        }
609
610        // Add terminator
611        self.buffer.push(type_prefixes::TERMINATOR);
612        Ok(())
613    }
614
615    /// Encode variable-length integer (varint)
616    fn encode_varint(&mut self, mut value: u64) -> Result<()> {
617        while value >= 0x80 {
618            self.buffer.push((value & 0xFF) as u8 | 0x80);
619            value >>= 7;
620        }
621        self.buffer.push(value as u8);
622        Ok(())
623    }
624}
625
626/// Byte-comparable key decoder (for debugging/testing)
627pub struct ByteComparableDecoder;
628
629impl ByteComparableDecoder {
630    /// Decode a byte-comparable key back to readable format (best effort)
631    pub fn decode_key_debug(encoded: &[u8]) -> String {
632        if encoded.is_empty() {
633            return "<empty>".to_string();
634        }
635
636        // Simple hex representation for debugging
637        let hex: String = encoded
638            .iter()
639            .map(|b| format!("{:02x}", b))
640            .collect::<Vec<_>>()
641            .join(" ");
642
643        // Try to detect if it looks like text (allow null bytes)
644        if let Ok(text) = std::str::from_utf8(encoded) {
645            // Check if most characters are printable (allow some control chars like null)
646            let printable_count = text
647                .chars()
648                .filter(|c| c.is_ascii_graphic() || c.is_ascii_whitespace())
649                .count();
650            let total_count = text.chars().count();
651
652            if total_count > 0 && (printable_count as f32 / total_count as f32) >= 0.7 {
653                let clean_text = text.trim_end_matches('\0').trim();
654                if !clean_text.is_empty() {
655                    return format!("\"{}\" ({})", clean_text, hex);
656                }
657            }
658        }
659
660        // Try to extract text even from binary data that might contain readable strings
661        let mut text_parts = Vec::new();
662        let mut current_text = String::new();
663
664        for &byte in encoded {
665            if byte.is_ascii_graphic() || byte == b' ' {
666                current_text.push(byte as char);
667            } else if !current_text.is_empty() {
668                text_parts.push(current_text.clone());
669                current_text.clear();
670            }
671        }
672
673        if !current_text.is_empty() {
674            text_parts.push(current_text);
675        }
676
677        // If we found any readable text parts, include them
678        if !text_parts.is_empty() {
679            let text_content = text_parts.join(" ");
680            return format!("\"{}\" ({})", text_content, hex);
681        }
682
683        format!("0x{}", hex)
684    }
685}
686
687/// Batch encoder for efficient encoding of multiple values
688pub struct BatchEncoder {
689    encoder: ByteComparableEncoder,
690    batch_buffer: Vec<Vec<u8>>,
691}
692
693impl Default for BatchEncoder {
694    fn default() -> Self {
695        Self::new()
696    }
697}
698
699impl BatchEncoder {
700    /// Create new batch encoder
701    pub fn new() -> Self {
702        Self {
703            encoder: ByteComparableEncoder::new(),
704            batch_buffer: Vec::new(),
705        }
706    }
707
708    /// Encode a batch of values efficiently
709    pub fn encode_batch(&mut self, values: &[Value]) -> Result<Vec<Vec<u8>>> {
710        self.batch_buffer.clear();
711        self.batch_buffer.reserve(values.len());
712
713        for value in values {
714            let encoded = self.encoder.encode_value(value)?;
715            self.batch_buffer.push(encoded);
716        }
717
718        Ok(self.batch_buffer.clone())
719    }
720
721    /// Clear the batch buffer
722    pub fn clear(&mut self) {
723        self.batch_buffer.clear();
724    }
725}
726
727/// Performance statistics for the encoder
728#[derive(Debug, Clone, Default)]
729pub struct EncoderStats {
730    /// Current buffer capacity
731    pub buffer_capacity: usize,
732    /// Current buffer size
733    pub buffer_size: usize,
734    /// Number of encodings performed
735    pub encodings_performed: u64,
736    /// Total bytes encoded
737    pub total_bytes_encoded: u64,
738}
739
740impl ByteComparableEncoder {
741    /// Reserve capacity in the internal buffer
742    pub fn reserve(&mut self, additional: usize) {
743        self.buffer.reserve(additional);
744    }
745
746    /// Get performance statistics
747    pub fn get_stats(&self) -> EncoderStats {
748        EncoderStats {
749            buffer_capacity: self.buffer.capacity(),
750            buffer_size: self.buffer.len(),
751            encodings_performed: 0, // Would need to track this in practice
752            total_bytes_encoded: self.buffer.len() as u64,
753        }
754    }
755
756    /// Validate an encoded key for correctness
757    pub fn validate_encoded_key(&self, encoded: &[u8]) -> Result<()> {
758        if encoded.is_empty() {
759            return Err(BtiError::InvalidByteComparableKey("Empty encoded key".to_string()).into());
760        }
761
762        let type_prefix = encoded[0];
763
764        // Validate type prefix
765        match type_prefix {
766            type_prefixes::NULL => {
767                if encoded.len() > 2 {
768                    return Err(BtiError::InvalidByteComparableKey(
769                        "Null value too long".to_string(),
770                    )
771                    .into());
772                }
773            }
774            type_prefixes::BOOLEAN_FALSE | type_prefixes::BOOLEAN_TRUE => {
775                if encoded.len() != 1 {
776                    return Err(BtiError::InvalidByteComparableKey(
777                        "Boolean value should be exactly 1 byte".to_string(),
778                    )
779                    .into());
780                }
781            }
782            type_prefixes::TINYINT => {
783                if encoded.len() != 2 {
784                    return Err(BtiError::InvalidByteComparableKey(
785                        "TinyInt should be exactly 2 bytes".to_string(),
786                    )
787                    .into());
788                }
789            }
790            type_prefixes::SMALLINT => {
791                if encoded.len() != 3 {
792                    return Err(BtiError::InvalidByteComparableKey(
793                        "SmallInt should be exactly 3 bytes".to_string(),
794                    )
795                    .into());
796                }
797            }
798            type_prefixes::INTEGER => {
799                if encoded.len() != 5 {
800                    return Err(BtiError::InvalidByteComparableKey(
801                        "Integer should be exactly 5 bytes".to_string(),
802                    )
803                    .into());
804                }
805            }
806            type_prefixes::BIGINT | type_prefixes::TIMESTAMP => {
807                if encoded.len() != 9 {
808                    return Err(BtiError::InvalidByteComparableKey(
809                        "BigInt/Timestamp should be exactly 9 bytes".to_string(),
810                    )
811                    .into());
812                }
813            }
814            type_prefixes::FLOAT => {
815                if encoded.len() != 5 {
816                    return Err(BtiError::InvalidByteComparableKey(
817                        "Float should be exactly 5 bytes".to_string(),
818                    )
819                    .into());
820                }
821            }
822            type_prefixes::DOUBLE => {
823                if encoded.len() != 9 {
824                    return Err(BtiError::InvalidByteComparableKey(
825                        "Double should be exactly 9 bytes".to_string(),
826                    )
827                    .into());
828                }
829            }
830            type_prefixes::UUID => {
831                if encoded.len() != 17 {
832                    return Err(BtiError::InvalidByteComparableKey(
833                        "UUID should be exactly 17 bytes".to_string(),
834                    )
835                    .into());
836                }
837            }
838            type_prefixes::TEXT | type_prefixes::BLOB => {
839                if encoded.len() < 2 || encoded[encoded.len() - 1] != type_prefixes::TERMINATOR {
840                    return Err(BtiError::InvalidByteComparableKey(
841                        "Text/Blob should end with terminator".to_string(),
842                    )
843                    .into());
844                }
845            }
846            type_prefixes::LIST
847            | type_prefixes::SET
848            | type_prefixes::MAP
849            | type_prefixes::TUPLE
850            | type_prefixes::UDT => {
851                if encoded.len() < 2 || encoded[encoded.len() - 1] != type_prefixes::TERMINATOR {
852                    return Err(BtiError::InvalidByteComparableKey(
853                        "Collection/Complex type should end with terminator".to_string(),
854                    )
855                    .into());
856                }
857            }
858            _ => {
859                return Err(BtiError::InvalidByteComparableKey(format!(
860                    "Unknown type prefix: 0x{:02x}",
861                    type_prefix
862                ))
863                .into());
864            }
865        }
866
867        Ok(())
868    }
869}
870
871#[cfg(test)]
872mod tests {
873    use super::*;
874    use uuid::Uuid;
875
876    #[test]
877    fn test_text_encoding() {
878        let mut encoder = ByteComparableEncoder::new();
879
880        let encoded_a = encoder.encode_value(&Value::text("a".to_string())).unwrap();
881        let encoded_b = encoder.encode_value(&Value::text("b".to_string())).unwrap();
882        let encoded_aa = encoder
883            .encode_value(&Value::text("aa".to_string()))
884            .unwrap();
885
886        // Lexicographic comparison should match string comparison
887        assert!(encoded_a < encoded_b);
888        assert!(encoded_a < encoded_aa);
889        assert!(encoded_aa < encoded_b);
890    }
891
892    #[test]
893    fn test_integer_encoding() {
894        let mut encoder = ByteComparableEncoder::new();
895
896        let encoded_neg = encoder.encode_value(&Value::Integer(-100)).unwrap();
897        let encoded_zero = encoder.encode_value(&Value::Integer(0)).unwrap();
898        let encoded_pos = encoder.encode_value(&Value::Integer(100)).unwrap();
899
900        // Proper numeric ordering
901        assert!(encoded_neg < encoded_zero);
902        assert!(encoded_zero < encoded_pos);
903    }
904
905    #[test]
906    fn test_boolean_encoding() {
907        let mut encoder = ByteComparableEncoder::new();
908
909        let encoded_false = encoder.encode_value(&Value::Boolean(false)).unwrap();
910        let encoded_true = encoder.encode_value(&Value::Boolean(true)).unwrap();
911
912        // false < true
913        assert!(encoded_false < encoded_true);
914    }
915
916    #[test]
917    fn test_uuid_encoding() {
918        let mut encoder = ByteComparableEncoder::new();
919
920        let uuid1 = Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap();
921        let uuid2 = Uuid::parse_str("00000000-0000-0000-0000-000000000002").unwrap();
922
923        let encoded1 = encoder
924            .encode_value(&Value::Uuid(*uuid1.as_bytes()))
925            .unwrap();
926        let encoded2 = encoder
927            .encode_value(&Value::Uuid(*uuid2.as_bytes()))
928            .unwrap();
929
930        assert!(encoded1 < encoded2);
931    }
932
933    #[test]
934    fn test_composite_key_encoding() {
935        let mut encoder = ByteComparableEncoder::new();
936
937        let key1 = vec![Value::text("partition1".to_string()), Value::Integer(1)];
938        let key2 = vec![Value::text("partition1".to_string()), Value::Integer(2)];
939        let key3 = vec![Value::text("partition2".to_string()), Value::Integer(1)];
940
941        let encoded1 = encoder.encode_composite_key(&key1).unwrap();
942        let encoded2 = encoder.encode_composite_key(&key2).unwrap();
943        let encoded3 = encoder.encode_composite_key(&key3).unwrap();
944
945        // Proper composite key ordering
946        assert!(encoded1 < encoded2); // Same partition, different clustering
947        assert!(encoded2 < encoded3); // Different partition
948    }
949
950    #[test]
951    fn test_list_encoding() {
952        let mut encoder = ByteComparableEncoder::new();
953
954        let list1 = Value::List(vec![Value::Integer(1), Value::Integer(2)]);
955        let list2 = Value::List(vec![
956            Value::Integer(1),
957            Value::Integer(2),
958            Value::Integer(3),
959        ]);
960
961        let encoded1 = encoder.encode_value(&list1).unwrap();
962        let encoded2 = encoder.encode_value(&list2).unwrap();
963
964        // Shorter list should come first
965        assert!(encoded1 < encoded2);
966    }
967
968    #[test]
969    fn test_float_special_values() {
970        let mut encoder = ByteComparableEncoder::new();
971
972        let neg_inf = encoder
973            .encode_value(&Value::Float(f64::NEG_INFINITY))
974            .unwrap();
975        let neg_one = encoder.encode_value(&Value::Float(-1.0)).unwrap();
976        let zero = encoder.encode_value(&Value::Float(0.0)).unwrap();
977        let one = encoder.encode_value(&Value::Float(1.0)).unwrap();
978        let pos_inf = encoder.encode_value(&Value::Float(f64::INFINITY)).unwrap();
979
980        // Proper float ordering
981        assert!(neg_inf < neg_one);
982        assert!(neg_one < zero);
983        assert!(zero < one);
984        assert!(one < pos_inf);
985    }
986
987    #[test]
988    fn test_decode_key_debug() {
989        let text_bytes = b"hello\0";
990        let decoded = ByteComparableDecoder::decode_key_debug(text_bytes);
991        assert!(decoded.contains("hello"));
992
993        let binary_bytes = &[0xFF, 0xFE, 0xFD];
994        let decoded = ByteComparableDecoder::decode_key_debug(binary_bytes);
995        assert!(decoded.starts_with("0x"));
996    }
997
998    #[test]
999    fn test_encoder_reuse() {
1000        let mut encoder = ByteComparableEncoder::new();
1001
1002        let encoded1 = encoder
1003            .encode_value(&Value::text("test1".to_string()))
1004            .unwrap();
1005        let encoded2 = encoder
1006            .encode_value(&Value::text("test2".to_string()))
1007            .unwrap();
1008
1009        // Each encoding should be independent
1010        assert_ne!(encoded1, encoded2);
1011        assert!(encoded1 < encoded2);
1012    }
1013
1014    #[test]
1015    fn test_encoder_config() {
1016        let config = EncoderConfig {
1017            use_varint_encoding: false,
1018            max_nesting_depth: 16,
1019            enable_prefix_compression: false,
1020            strict_compliance: false,
1021        };
1022
1023        let encoder = ByteComparableEncoder::with_config(config);
1024        assert!(!encoder.config().use_varint_encoding);
1025        assert_eq!(encoder.config().max_nesting_depth, 16);
1026    }
1027
1028    #[test]
1029    fn test_max_nesting_depth() {
1030        let config = EncoderConfig {
1031            max_nesting_depth: 2,
1032            ..Default::default()
1033        };
1034
1035        let mut encoder = ByteComparableEncoder::with_config(config);
1036
1037        // Create deeply nested structure
1038        let deep_nested = Value::List(vec![Value::List(vec![Value::List(vec![Value::Integer(
1039            1,
1040        )])])]);
1041
1042        // Should fail due to depth limit
1043        let result = encoder.encode_value(&deep_nested);
1044        assert!(result.is_err());
1045    }
1046
1047    #[test]
1048    fn test_batch_encoder() {
1049        let mut batch_encoder = BatchEncoder::new();
1050
1051        let values = vec![
1052            Value::Integer(1),
1053            Value::text("hello".to_string()),
1054            Value::Boolean(true),
1055        ];
1056
1057        let encoded_batch = batch_encoder.encode_batch(&values).unwrap();
1058        assert_eq!(encoded_batch.len(), 3);
1059
1060        // Verify individual encodings
1061        let mut single_encoder = ByteComparableEncoder::new();
1062        for (i, value) in values.iter().enumerate() {
1063            let single_encoded = single_encoder.encode_value(value).unwrap();
1064            assert_eq!(encoded_batch[i], single_encoded);
1065        }
1066    }
1067
1068    #[test]
1069    fn test_ordering_across_types() {
1070        let mut encoder = ByteComparableEncoder::new();
1071
1072        // Different types should have deterministic ordering based on type prefixes
1073        let null_val = encoder.encode_value(&Value::Null).unwrap();
1074        let bool_val = encoder.encode_value(&Value::Boolean(false)).unwrap();
1075        let int_val = encoder.encode_value(&Value::Integer(0)).unwrap();
1076        let text_val = encoder.encode_value(&Value::text("".to_string())).unwrap();
1077
1078        // Null should come first, then booleans, then numbers, then text
1079        assert!(null_val < bool_val);
1080        assert!(bool_val < int_val);
1081        assert!(int_val < text_val);
1082    }
1083
1084    #[test]
1085    fn test_validation() {
1086        let encoder = ByteComparableEncoder::new();
1087
1088        // Test valid encodings
1089        assert!(encoder.validate_encoded_key(&[type_prefixes::NULL]).is_ok());
1090        assert!(encoder
1091            .validate_encoded_key(&[type_prefixes::BOOLEAN_TRUE])
1092            .is_ok());
1093
1094        // Test invalid encodings
1095        assert!(encoder.validate_encoded_key(&[]).is_err()); // Empty
1096        assert!(encoder
1097            .validate_encoded_key(&[type_prefixes::NULL, 0x00, 0x00, 0x00])
1098            .is_err()); // Null too long
1099    }
1100
1101    #[test]
1102    fn test_performance_stats() {
1103        let mut encoder = ByteComparableEncoder::new();
1104        encoder.reserve(1024);
1105
1106        let stats = encoder.get_stats();
1107        assert!(stats.buffer_capacity >= 1024);
1108        assert_eq!(stats.buffer_size, 0);
1109
1110        // Encode something
1111        encoder
1112            .encode_value(&Value::text("test".to_string()))
1113            .unwrap();
1114        let stats_after = encoder.get_stats();
1115        assert!(stats_after.buffer_size > 0);
1116    }
1117
1118    #[test]
1119    fn test_timestamp_encoding() {
1120        let mut encoder = ByteComparableEncoder::new();
1121
1122        let past = encoder.encode_value(&Value::Timestamp(-1000)).unwrap();
1123        let epoch = encoder.encode_value(&Value::Timestamp(0)).unwrap();
1124        let future = encoder.encode_value(&Value::Timestamp(1000)).unwrap();
1125
1126        // All should start with timestamp prefix
1127        assert_eq!(past[0], type_prefixes::TIMESTAMP);
1128        assert_eq!(epoch[0], type_prefixes::TIMESTAMP);
1129        assert_eq!(future[0], type_prefixes::TIMESTAMP);
1130
1131        // Proper temporal ordering
1132        assert!(past < epoch);
1133        assert!(epoch < future);
1134    }
1135
1136    #[test]
1137    fn test_blob_encoding() {
1138        let mut encoder = ByteComparableEncoder::new();
1139
1140        let blob1 = encoder
1141            .encode_value(&Value::blob(vec![0x01, 0x02]))
1142            .unwrap();
1143        let blob2 = encoder
1144            .encode_value(&Value::blob(vec![0x01, 0x03]))
1145            .unwrap();
1146        let blob_with_null = encoder
1147            .encode_value(&Value::blob(vec![0x01, 0x00, 0x02]))
1148            .unwrap();
1149
1150        // Should start with blob prefix
1151        assert_eq!(blob1[0], type_prefixes::BLOB);
1152        assert_eq!(blob2[0], type_prefixes::BLOB);
1153
1154        // Proper lexicographic ordering
1155        assert!(blob1 < blob2);
1156
1157        // Should contain escaped null sequence
1158        assert!(blob_with_null
1159            .windows(2)
1160            .any(|w| w == escape_sequences::ESCAPED_NULL));
1161    }
1162
1163    #[test]
1164    fn test_comprehensive_ordering() {
1165        let mut encoder = ByteComparableEncoder::new();
1166
1167        // Test a comprehensive set of values for ordering consistency
1168        let values = vec![
1169            Value::Null,
1170            Value::Boolean(false),
1171            Value::Boolean(true),
1172            Value::TinyInt(-1),
1173            Value::TinyInt(0),
1174            Value::TinyInt(1),
1175            Value::SmallInt(-100),
1176            Value::SmallInt(100),
1177            Value::Integer(-1000),
1178            Value::Integer(1000),
1179            Value::BigInt(-10000),
1180            Value::BigInt(10000),
1181            Value::Float32(-1.0),
1182            Value::Float32(1.0),
1183            Value::Float(-1.0),
1184            Value::Float(1.0),
1185            Value::text("a".to_string()),
1186            Value::text("z".to_string()),
1187            Value::blob(vec![0x01]),
1188            Value::blob(vec![0xFF]),
1189            Value::Uuid([0u8; 16]),
1190            Value::Uuid([0xFFu8; 16]),
1191            Value::Timestamp(-1000),
1192            Value::Timestamp(1000),
1193        ];
1194
1195        let encoded_values: Vec<_> = values
1196            .iter()
1197            .map(|v| encoder.encode_value(v).unwrap())
1198            .collect();
1199
1200        // Verify that lexicographic ordering of encoded values matches the input ordering
1201        for i in 0..encoded_values.len() - 1 {
1202            assert!(
1203                encoded_values[i] <= encoded_values[i + 1],
1204                "Ordering violation at index {} and {}",
1205                i,
1206                i + 1
1207            );
1208        }
1209    }
1210}
cqlite_core/storage/sstable/bti/encoder.rs

cqlite_core/storage/sstable/bti/
encoder.rs
