idb/binlog/
row_image.rs

1//! Binlog row image parsing — extract PK values from row event data.
2//!
3//! MySQL binlog row events (WRITE/UPDATE/DELETE_ROWS) encode row data as
4//! packed binary images: a null bitmap followed by column values in column
5//! order. This module parses those images to extract primary key column
6//! values, enabling correlation between binlog events and InnoDB tablespace
7//! pages.
8//!
9//! All integer fields use **little-endian** byte order and standard 2's
10//! complement encoding (NOT InnoDB's big-endian XOR'd sign-bit format).
11
12use byteorder::{ByteOrder, LittleEndian};
13use serde::Serialize;
14
15// -----------------------------------------------------------------------
16// MySQL protocol type codes (from mysql_com.h / field_types.h)
17// -----------------------------------------------------------------------
18
19/// TINY (1-byte integer).
20const MYSQL_TYPE_TINY: u8 = 1;
21/// SHORT (2-byte integer).
22const MYSQL_TYPE_SHORT: u8 = 2;
23/// LONG (4-byte integer).
24const MYSQL_TYPE_LONG: u8 = 3;
25/// FLOAT (4-byte IEEE 754).
26const MYSQL_TYPE_FLOAT: u8 = 4;
27/// DOUBLE (8-byte IEEE 754).
28const MYSQL_TYPE_DOUBLE: u8 = 5;
29/// LONGLONG (8-byte integer).
30const MYSQL_TYPE_LONGLONG: u8 = 8;
31/// INT24 / MEDIUMINT (3-byte integer).
32const MYSQL_TYPE_INT24: u8 = 9;
33/// DATETIME (legacy 8-byte format, pre-5.6.4).
34const MYSQL_TYPE_DATETIME: u8 = 12;
35/// VARCHAR (variable-length string).
36const MYSQL_TYPE_VARCHAR: u8 = 15;
37/// BIT.
38const MYSQL_TYPE_BIT: u8 = 16;
39/// TIMESTAMP2 (temporal with fractional seconds, 5.6.4+).
40const MYSQL_TYPE_TIMESTAMP2: u8 = 17;
41/// DATETIME2 (temporal with fractional seconds, 5.6.4+).
42const MYSQL_TYPE_DATETIME2: u8 = 18;
43/// TIME2 (temporal with fractional seconds, 5.6.4+).
44const MYSQL_TYPE_TIME2: u8 = 19;
45/// NEWDECIMAL (packed decimal).
46const MYSQL_TYPE_NEWDECIMAL: u8 = 246;
47/// ENUM.
48const MYSQL_TYPE_ENUM: u8 = 247;
49/// SET.
50const MYSQL_TYPE_SET: u8 = 248;
51/// BLOB / TEXT (variable-length binary).
52const MYSQL_TYPE_BLOB: u8 = 252;
53/// VAR_STRING.
54const MYSQL_TYPE_VAR_STRING: u8 = 253;
55/// STRING (fixed-length).
56const MYSQL_TYPE_STRING: u8 = 254;
57
58// -----------------------------------------------------------------------
59// Public types
60// -----------------------------------------------------------------------
61
62/// Column metadata for a binlog row image column.
63#[derive(Debug, Clone, Serialize)]
64pub struct BinlogColumnMeta {
65    /// MySQL protocol type code (1=TINY, 2=SHORT, 3=LONG, 8=LONGLONG, 15=VARCHAR, etc.).
66    pub column_type: u8,
67    /// Whether the column is unsigned.
68    pub is_unsigned: bool,
69    /// Type-specific metadata (max length for VARCHAR, display size for INT, etc.).
70    pub type_metadata: u16,
71    /// Whether this column is part of the primary key.
72    pub is_pk: bool,
73    /// Ordinal position within the PK (0-based), if `is_pk` is true.
74    pub pk_ordinal: Option<usize>,
75}
76
77/// A decoded primary key value from a binlog row image.
78#[derive(Debug, Clone, PartialEq, Serialize)]
79pub enum BinlogPkValue {
80    /// Signed integer value.
81    Int(i64),
82    /// Unsigned integer value.
83    Uint(u64),
84    /// String value (VARCHAR, CHAR, etc.).
85    Str(String),
86    /// Raw bytes (BLOB, BINARY, etc.).
87    Bytes(Vec<u8>),
88}
89
90impl std::fmt::Display for BinlogPkValue {
91    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
92        match self {
93            BinlogPkValue::Int(v) => write!(f, "{}", v),
94            BinlogPkValue::Uint(v) => write!(f, "{}", v),
95            BinlogPkValue::Str(s) => write!(f, "{}", s),
96            BinlogPkValue::Bytes(b) => {
97                write!(f, "0x")?;
98                for byte in b {
99                    write!(f, "{:02x}", byte)?;
100                }
101                Ok(())
102            }
103        }
104    }
105}
106
107// -----------------------------------------------------------------------
108// Metadata parsing
109// -----------------------------------------------------------------------
110
111/// Parse per-column metadata from a TABLE_MAP event's metadata section.
112///
113/// Each MySQL column type consumes a different number of bytes from the
114/// metadata block. This function walks the metadata bytes and returns
115/// one `u16` metadata value per column.
116///
117/// # Arguments
118///
119/// * `column_types` — Column type codes from the TABLE_MAP event.
120/// * `metadata_bytes` — Raw metadata section bytes.
121///
122/// # Returns
123///
124/// A vector of per-column metadata values (same length as `column_types`).
125pub fn parse_column_metadata(column_types: &[u8], metadata_bytes: &[u8]) -> Vec<u16> {
126    let mut result = Vec::with_capacity(column_types.len());
127    let mut offset = 0;
128
129    for &col_type in column_types {
130        let (value, consumed) = read_type_metadata(col_type, metadata_bytes, offset);
131        result.push(value);
132        offset += consumed;
133    }
134
135    result
136}
137
138/// Read one column's metadata from the metadata byte stream.
139///
140/// Returns `(metadata_value, bytes_consumed)`.
141fn read_type_metadata(column_type: u8, data: &[u8], offset: usize) -> (u16, usize) {
142    match column_type {
143        // 0-byte metadata types
144        MYSQL_TYPE_TINY | MYSQL_TYPE_SHORT | MYSQL_TYPE_LONG | MYSQL_TYPE_LONGLONG
145        | MYSQL_TYPE_INT24 | MYSQL_TYPE_DATETIME => (0, 0),
146
147        // 1-byte metadata types
148        MYSQL_TYPE_FLOAT | MYSQL_TYPE_DOUBLE => {
149            let v = data.get(offset).copied().unwrap_or(0) as u16;
150            (v, 1)
151        }
152        MYSQL_TYPE_DATETIME2 | MYSQL_TYPE_TIMESTAMP2 | MYSQL_TYPE_TIME2 => {
153            let v = data.get(offset).copied().unwrap_or(0) as u16;
154            (v, 1)
155        }
156        MYSQL_TYPE_BLOB => {
157            // Pack length (1-4), stored as 1 byte.
158            let v = data.get(offset).copied().unwrap_or(0) as u16;
159            (v, 1)
160        }
161
162        // 2-byte metadata types (little-endian u16)
163        MYSQL_TYPE_VARCHAR | MYSQL_TYPE_VAR_STRING => {
164            if offset + 2 <= data.len() {
165                (LittleEndian::read_u16(&data[offset..]), 2)
166            } else {
167                (0, 2)
168            }
169        }
170        MYSQL_TYPE_NEWDECIMAL => {
171            // precision (1 byte) + scale (1 byte), packed as (precision << 8 | scale).
172            let precision = data.get(offset).copied().unwrap_or(0) as u16;
173            let scale = data.get(offset + 1).copied().unwrap_or(0) as u16;
174            ((precision << 8) | scale, 2)
175        }
176        MYSQL_TYPE_STRING | MYSQL_TYPE_ENUM | MYSQL_TYPE_SET => {
177            // 2 bytes: real type + length
178            if offset + 2 <= data.len() {
179                let b0 = data[offset] as u16;
180                let b1 = data[offset + 1] as u16;
181                ((b0 << 8) | b1, 2)
182            } else {
183                (0, 2)
184            }
185        }
186        MYSQL_TYPE_BIT => {
187            // bits (1 byte) + bytes (1 byte)
188            if offset + 2 <= data.len() {
189                let bits = data[offset] as u16;
190                let bytes = data[offset + 1] as u16;
191                ((bytes << 8) | bits, 2)
192            } else {
193                (0, 2)
194            }
195        }
196
197        // Unknown type — consume 0 bytes (best effort).
198        _ => (0, 0),
199    }
200}
201
202// -----------------------------------------------------------------------
203// Row image PK extraction
204// -----------------------------------------------------------------------
205
206/// Extract primary key values from a binlog row image.
207///
208/// Parses the first row from the packed row image data, decoding column
209/// values based on the column metadata, and returns only the PK columns
210/// sorted by their ordinal position within the key.
211///
212/// # Arguments
213///
214/// * `row_data` — Raw row image bytes (null bitmap + column values).
215/// * `columns` — Column metadata (type, signedness, PK membership).
216///
217/// # Returns
218///
219/// `Some(Vec<BinlogPkValue>)` with PK values in key order, or `None` if
220/// any PK column is NULL or cannot be decoded.
221pub fn extract_pk_from_row_image(
222    row_data: &[u8],
223    columns: &[BinlogColumnMeta],
224) -> Option<Vec<BinlogPkValue>> {
225    let col_count = columns.len();
226    if col_count == 0 {
227        return None;
228    }
229
230    // Null bitmap: ceil(col_count / 8) bytes.
231    let null_bitmap_len = col_count.div_ceil(8);
232    if row_data.len() < null_bitmap_len {
233        return None;
234    }
235    let null_bitmap = &row_data[..null_bitmap_len];
236    let mut offset = null_bitmap_len;
237
238    // Collect PK values keyed by pk_ordinal.
239    let pk_count = columns.iter().filter(|c| c.is_pk).count();
240    if pk_count == 0 {
241        return None;
242    }
243    let mut pk_values: Vec<(usize, BinlogPkValue)> = Vec::with_capacity(pk_count);
244
245    for (i, col) in columns.iter().enumerate() {
246        // Check null bitmap.
247        let is_null = (null_bitmap[i / 8] >> (i % 8)) & 1 == 1;
248
249        if is_null {
250            if col.is_pk {
251                // PK column is NULL — cannot extract PK.
252                return None;
253            }
254            continue;
255        }
256
257        // Decode or skip this column's value.
258        let remaining = &row_data[offset..];
259        let size = match column_value_size(col.column_type, col.type_metadata, remaining) {
260            Some(s) => s,
261            None => {
262                if col.is_pk {
263                    return None;
264                }
265                // Cannot determine size of non-PK column — abort.
266                return None;
267            }
268        };
269
270        if col.is_pk {
271            if remaining.len() < size {
272                return None;
273            }
274            let value = decode_column_value(
275                col.column_type,
276                col.is_unsigned,
277                col.type_metadata,
278                &remaining[..size],
279            )?;
280            let ordinal = col.pk_ordinal.unwrap_or(0);
281            pk_values.push((ordinal, value));
282        }
283
284        offset += size;
285    }
286
287    // Sort by PK ordinal.
288    pk_values.sort_by_key(|(ord, _)| *ord);
289    Some(pk_values.into_iter().map(|(_, v)| v).collect())
290}
291
292// -----------------------------------------------------------------------
293// Column value size computation
294// -----------------------------------------------------------------------
295
296/// Compute the byte size of a column value in the row image.
297///
298/// This is needed to skip over non-PK columns when parsing the row.
299/// Returns `None` for column types whose size cannot be determined.
300fn column_value_size(column_type: u8, metadata: u16, data: &[u8]) -> Option<usize> {
301    match column_type {
302        MYSQL_TYPE_TINY => Some(1),
303        MYSQL_TYPE_SHORT => Some(2),
304        MYSQL_TYPE_INT24 => Some(3),
305        MYSQL_TYPE_LONG => Some(4),
306        MYSQL_TYPE_LONGLONG => Some(8),
307        MYSQL_TYPE_FLOAT => Some(4),
308        MYSQL_TYPE_DOUBLE => Some(8),
309        MYSQL_TYPE_DATETIME => Some(8), // Legacy fixed 8-byte format (pre-5.6.4)
310
311        MYSQL_TYPE_VARCHAR | MYSQL_TYPE_VAR_STRING => {
312            // Length prefix: 1 byte if max_len < 256, else 2 bytes LE.
313            let max_len = metadata as usize;
314            if max_len < 256 {
315                let len = *data.first()? as usize;
316                Some(1 + len)
317            } else {
318                if data.len() < 2 {
319                    return None;
320                }
321                let len = LittleEndian::read_u16(data) as usize;
322                Some(2 + len)
323            }
324        }
325
326        MYSQL_TYPE_STRING => {
327            // Fixed-length string. Metadata high byte is real type, low byte is length.
328            // For types that map to STRING in binlog (CHAR, ENUM, SET), the
329            // metadata encodes the real type and max byte length.
330            let real_type = (metadata >> 8) as u8;
331            let max_len = metadata & 0xFF;
332
333            match real_type {
334                // ENUM: size is 1 or 2 bytes depending on element count.
335                MYSQL_TYPE_ENUM => {
336                    let size = max_len as usize;
337                    Some(size)
338                }
339                // SET: size in bytes.
340                MYSQL_TYPE_SET => {
341                    let size = max_len as usize;
342                    Some(size)
343                }
344                // Default CHAR: if max_len > 255, use 2-byte length prefix.
345                _ => {
346                    if max_len > 255 {
347                        if data.len() < 2 {
348                            return None;
349                        }
350                        let len = LittleEndian::read_u16(data) as usize;
351                        Some(2 + len)
352                    } else {
353                        let len = *data.first()? as usize;
354                        Some(1 + len)
355                    }
356                }
357            }
358        }
359
360        MYSQL_TYPE_BLOB => {
361            // Pack length from metadata (1-4 bytes).
362            let pack_len = metadata as usize;
363            if pack_len == 0 || data.len() < pack_len {
364                return None;
365            }
366            let blob_len = match pack_len {
367                1 => data[0] as usize,
368                2 => LittleEndian::read_u16(data) as usize,
369                3 => data[0] as usize | (data[1] as usize) << 8 | (data[2] as usize) << 16,
370                4 => LittleEndian::read_u32(data) as usize,
371                _ => return None,
372            };
373            Some(pack_len + blob_len)
374        }
375
376        MYSQL_TYPE_NEWDECIMAL => {
377            let precision = (metadata >> 8) as usize;
378            let scale = (metadata & 0xFF) as usize;
379            Some(decimal_binary_size(precision, scale))
380        }
381
382        MYSQL_TYPE_BIT => {
383            let bits = (metadata & 0xFF) as usize;
384            let bytes = (metadata >> 8) as usize;
385            Some(bytes + if bits > 0 { 1 } else { 0 })
386        }
387
388        MYSQL_TYPE_DATETIME2 => {
389            // 5 bytes + fractional seconds part.
390            let fsp = metadata as usize;
391            Some(5 + fsp_storage_size(fsp))
392        }
393
394        MYSQL_TYPE_TIMESTAMP2 => {
395            // 4 bytes + fractional seconds part.
396            let fsp = metadata as usize;
397            Some(4 + fsp_storage_size(fsp))
398        }
399
400        MYSQL_TYPE_TIME2 => {
401            // 3 bytes + fractional seconds part.
402            let fsp = metadata as usize;
403            Some(3 + fsp_storage_size(fsp))
404        }
405
406        // Unsupported type — cannot determine size.
407        _ => None,
408    }
409}
410
411/// Compute storage size for fractional seconds precision.
412///
413/// MySQL temporal types store fractional seconds in ceil(fsp/2) bytes.
414fn fsp_storage_size(fsp: usize) -> usize {
415    fsp.div_ceil(2)
416}
417
418/// Compute binary storage size for a DECIMAL(precision, scale).
419///
420/// MySQL packs 9 digits per 4 bytes, with leftover digits using fewer bytes.
421fn decimal_binary_size(precision: usize, scale: usize) -> usize {
422    // Digits-to-bytes mapping for leftover digits (0-8).
423    const DIG2BYTES: [usize; 10] = [0, 1, 1, 2, 2, 3, 3, 4, 4, 4];
424
425    let intg = precision - scale;
426    let intg_full = intg / 9;
427    let intg_leftover = intg % 9;
428    let frac_full = scale / 9;
429    let frac_leftover = scale % 9;
430
431    intg_full * 4 + DIG2BYTES[intg_leftover] + frac_full * 4 + DIG2BYTES[frac_leftover]
432}
433
434// -----------------------------------------------------------------------
435// Column value decoding
436// -----------------------------------------------------------------------
437
438/// Decode a single column value from its raw bytes.
439///
440/// Returns `None` for unsupported types (the caller should treat this as
441/// an inability to extract the PK).
442fn decode_column_value(
443    column_type: u8,
444    is_unsigned: bool,
445    metadata: u16,
446    data: &[u8],
447) -> Option<BinlogPkValue> {
448    match column_type {
449        MYSQL_TYPE_TINY => {
450            if data.is_empty() {
451                return None;
452            }
453            if is_unsigned {
454                Some(BinlogPkValue::Uint(data[0] as u64))
455            } else {
456                Some(BinlogPkValue::Int(data[0] as i8 as i64))
457            }
458        }
459
460        MYSQL_TYPE_SHORT => {
461            if data.len() < 2 {
462                return None;
463            }
464            if is_unsigned {
465                Some(BinlogPkValue::Uint(LittleEndian::read_u16(data) as u64))
466            } else {
467                Some(BinlogPkValue::Int(LittleEndian::read_i16(data) as i64))
468            }
469        }
470
471        MYSQL_TYPE_INT24 => {
472            if data.len() < 3 {
473                return None;
474            }
475            let raw = data[0] as u32 | (data[1] as u32) << 8 | (data[2] as u32) << 16;
476            if is_unsigned {
477                Some(BinlogPkValue::Uint(raw as u64))
478            } else {
479                // Sign-extend from 24 bits.
480                let signed = if raw & 0x800000 != 0 {
481                    (raw | 0xFF000000) as i32 as i64
482                } else {
483                    raw as i64
484                };
485                Some(BinlogPkValue::Int(signed))
486            }
487        }
488
489        MYSQL_TYPE_LONG => {
490            if data.len() < 4 {
491                return None;
492            }
493            if is_unsigned {
494                Some(BinlogPkValue::Uint(LittleEndian::read_u32(data) as u64))
495            } else {
496                Some(BinlogPkValue::Int(LittleEndian::read_i32(data) as i64))
497            }
498        }
499
500        MYSQL_TYPE_LONGLONG => {
501            if data.len() < 8 {
502                return None;
503            }
504            if is_unsigned {
505                Some(BinlogPkValue::Uint(LittleEndian::read_u64(data)))
506            } else {
507                Some(BinlogPkValue::Int(LittleEndian::read_i64(data)))
508            }
509        }
510
511        MYSQL_TYPE_FLOAT => {
512            if data.len() < 4 {
513                return None;
514            }
515            // Store float as its integer bit pattern for PK purposes.
516            let bits = LittleEndian::read_u32(data);
517            let val = f32::from_bits(bits);
518            Some(BinlogPkValue::Str(format!("{}", val)))
519        }
520
521        MYSQL_TYPE_DOUBLE => {
522            if data.len() < 8 {
523                return None;
524            }
525            let bits = LittleEndian::read_u64(data);
526            let val = f64::from_bits(bits);
527            Some(BinlogPkValue::Str(format!("{}", val)))
528        }
529
530        MYSQL_TYPE_VARCHAR | MYSQL_TYPE_VAR_STRING => {
531            let max_len = metadata as usize;
532            if max_len < 256 {
533                let len = *data.first()? as usize;
534                if data.len() < 1 + len {
535                    return None;
536                }
537                let s = String::from_utf8_lossy(&data[1..1 + len]).into_owned();
538                Some(BinlogPkValue::Str(s))
539            } else {
540                if data.len() < 2 {
541                    return None;
542                }
543                let len = LittleEndian::read_u16(data) as usize;
544                if data.len() < 2 + len {
545                    return None;
546                }
547                let s = String::from_utf8_lossy(&data[2..2 + len]).into_owned();
548                Some(BinlogPkValue::Str(s))
549            }
550        }
551
552        MYSQL_TYPE_STRING => {
553            let real_type = (metadata >> 8) as u8;
554            let max_len = metadata & 0xFF;
555
556            match real_type {
557                MYSQL_TYPE_ENUM => {
558                    let size = max_len as usize;
559                    if data.len() < size {
560                        return None;
561                    }
562                    let val = match size {
563                        1 => data[0] as u64,
564                        2 => LittleEndian::read_u16(data) as u64,
565                        _ => return None,
566                    };
567                    Some(BinlogPkValue::Uint(val))
568                }
569                MYSQL_TYPE_SET => {
570                    let size = max_len as usize;
571                    if data.len() < size {
572                        return None;
573                    }
574                    Some(BinlogPkValue::Bytes(data[..size].to_vec()))
575                }
576                _ => {
577                    // Fixed-length CHAR: length-prefixed.
578                    if max_len > 255 {
579                        if data.len() < 2 {
580                            return None;
581                        }
582                        let len = LittleEndian::read_u16(data) as usize;
583                        if data.len() < 2 + len {
584                            return None;
585                        }
586                        let s = String::from_utf8_lossy(&data[2..2 + len]).into_owned();
587                        Some(BinlogPkValue::Str(s))
588                    } else {
589                        let len = *data.first()? as usize;
590                        if data.len() < 1 + len {
591                            return None;
592                        }
593                        let s = String::from_utf8_lossy(&data[1..1 + len]).into_owned();
594                        Some(BinlogPkValue::Str(s))
595                    }
596                }
597            }
598        }
599
600        MYSQL_TYPE_BLOB => {
601            let pack_len = metadata as usize;
602            if pack_len == 0 || data.len() < pack_len {
603                return None;
604            }
605            let blob_len = match pack_len {
606                1 => data[0] as usize,
607                2 => LittleEndian::read_u16(data) as usize,
608                3 => data[0] as usize | (data[1] as usize) << 8 | (data[2] as usize) << 16,
609                4 => LittleEndian::read_u32(data) as usize,
610                _ => return None,
611            };
612            if data.len() < pack_len + blob_len {
613                return None;
614            }
615            Some(BinlogPkValue::Bytes(
616                data[pack_len..pack_len + blob_len].to_vec(),
617            ))
618        }
619
620        // Other types: return None (cannot decode for PK purposes).
621        _ => None,
622    }
623}
624
625// -----------------------------------------------------------------------
626// Tests
627// -----------------------------------------------------------------------
628
629#[cfg(test)]
630mod tests {
631    use super::*;
632
633    // -- parse_column_metadata tests --
634
635    #[test]
636    fn parse_metadata_int_types() {
637        // INT types consume 0 metadata bytes.
638        let types = vec![MYSQL_TYPE_TINY, MYSQL_TYPE_LONG, MYSQL_TYPE_LONGLONG];
639        let meta = parse_column_metadata(&types, &[]);
640        assert_eq!(meta, vec![0, 0, 0]);
641    }
642
643    #[test]
644    fn parse_metadata_varchar() {
645        let types = vec![MYSQL_TYPE_VARCHAR];
646        // Max length = 200 (0xC8, 0x00 LE).
647        let meta_bytes = [0xC8, 0x00];
648        let meta = parse_column_metadata(&types, &meta_bytes);
649        assert_eq!(meta, vec![200]);
650    }
651
652    #[test]
653    fn parse_metadata_blob() {
654        let types = vec![MYSQL_TYPE_BLOB];
655        let meta_bytes = [2]; // pack length = 2
656        let meta = parse_column_metadata(&types, &meta_bytes);
657        assert_eq!(meta, vec![2]);
658    }
659
660    #[test]
661    fn parse_metadata_mixed() {
662        // LONG (0 bytes) + VARCHAR (2 bytes) + BLOB (1 byte) + SHORT (0 bytes)
663        let types = vec![
664            MYSQL_TYPE_LONG,
665            MYSQL_TYPE_VARCHAR,
666            MYSQL_TYPE_BLOB,
667            MYSQL_TYPE_SHORT,
668        ];
669        let meta_bytes = [0x00, 0x01, 3]; // VARCHAR max_len=256 (0x0100 LE), BLOB pack=3
670        let meta = parse_column_metadata(&types, &meta_bytes);
671        assert_eq!(meta, vec![0, 256, 3, 0]);
672    }
673
674    #[test]
675    fn parse_metadata_newdecimal() {
676        let types = vec![MYSQL_TYPE_NEWDECIMAL];
677        // precision=10, scale=2
678        let meta_bytes = [10, 2];
679        let meta = parse_column_metadata(&types, &meta_bytes);
680        // Packed as (10 << 8) | 2 = 2562.
681        assert_eq!(meta, vec![2562]);
682    }
683
684    #[test]
685    fn parse_metadata_datetime2() {
686        let types = vec![MYSQL_TYPE_DATETIME2];
687        let meta_bytes = [3]; // fsp=3
688        let meta = parse_column_metadata(&types, &meta_bytes);
689        assert_eq!(meta, vec![3]);
690    }
691
692    // -- INT decoding tests --
693
694    #[test]
695    fn decode_tiny_signed() {
696        let cols = vec![make_pk_col(MYSQL_TYPE_TINY, false, 0, 0)];
697        let mut data = vec![0u8; 1]; // null bitmap: no nulls
698        data.push(0xFE); // -2 as i8
699        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
700        assert_eq!(pk, vec![BinlogPkValue::Int(-2)]);
701    }
702
703    #[test]
704    fn decode_tiny_unsigned() {
705        let cols = vec![make_pk_col(MYSQL_TYPE_TINY, true, 0, 0)];
706        let mut data = vec![0u8; 1];
707        data.push(0xFE); // 254 unsigned
708        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
709        assert_eq!(pk, vec![BinlogPkValue::Uint(254)]);
710    }
711
712    #[test]
713    fn decode_long_signed() {
714        let cols = vec![make_pk_col(MYSQL_TYPE_LONG, false, 0, 0)];
715        let mut data = vec![0u8; 1];
716        let mut buf = [0u8; 4];
717        LittleEndian::write_i32(&mut buf, -42);
718        data.extend_from_slice(&buf);
719        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
720        assert_eq!(pk, vec![BinlogPkValue::Int(-42)]);
721    }
722
723    #[test]
724    fn decode_long_unsigned() {
725        let cols = vec![make_pk_col(MYSQL_TYPE_LONG, true, 0, 0)];
726        let mut data = vec![0u8; 1];
727        let mut buf = [0u8; 4];
728        LittleEndian::write_u32(&mut buf, 3_000_000_000);
729        data.extend_from_slice(&buf);
730        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
731        assert_eq!(pk, vec![BinlogPkValue::Uint(3_000_000_000)]);
732    }
733
734    #[test]
735    fn decode_longlong_signed() {
736        let cols = vec![make_pk_col(MYSQL_TYPE_LONGLONG, false, 0, 0)];
737        let mut data = vec![0u8; 1];
738        let mut buf = [0u8; 8];
739        LittleEndian::write_i64(&mut buf, -9_000_000_000);
740        data.extend_from_slice(&buf);
741        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
742        assert_eq!(pk, vec![BinlogPkValue::Int(-9_000_000_000)]);
743    }
744
745    #[test]
746    fn decode_longlong_unsigned() {
747        let cols = vec![make_pk_col(MYSQL_TYPE_LONGLONG, true, 0, 0)];
748        let mut data = vec![0u8; 1];
749        let mut buf = [0u8; 8];
750        LittleEndian::write_u64(&mut buf, 18_000_000_000_000_000_000);
751        data.extend_from_slice(&buf);
752        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
753        assert_eq!(pk, vec![BinlogPkValue::Uint(18_000_000_000_000_000_000)]);
754    }
755
756    #[test]
757    fn decode_short_signed() {
758        let cols = vec![make_pk_col(MYSQL_TYPE_SHORT, false, 0, 0)];
759        let mut data = vec![0u8; 1];
760        let mut buf = [0u8; 2];
761        LittleEndian::write_i16(&mut buf, -1000);
762        data.extend_from_slice(&buf);
763        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
764        assert_eq!(pk, vec![BinlogPkValue::Int(-1000)]);
765    }
766
767    #[test]
768    fn decode_int24_signed() {
769        let cols = vec![make_pk_col(MYSQL_TYPE_INT24, false, 0, 0)];
770        let mut data = vec![0u8; 1];
771        // -100 as 3-byte LE: 0xFFFF9C in LE = [0x9C, 0xFF, 0xFF]
772        data.extend_from_slice(&[0x9C, 0xFF, 0xFF]);
773        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
774        assert_eq!(pk, vec![BinlogPkValue::Int(-100)]);
775    }
776
777    #[test]
778    fn decode_int24_unsigned() {
779        let cols = vec![make_pk_col(MYSQL_TYPE_INT24, true, 0, 0)];
780        let mut data = vec![0u8; 1];
781        // 100000 = 0x0186A0, LE = [0xA0, 0x86, 0x01]
782        data.extend_from_slice(&[0xA0, 0x86, 0x01]);
783        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
784        assert_eq!(pk, vec![BinlogPkValue::Uint(100_000)]);
785    }
786
787    // -- VARCHAR decoding tests --
788
789    #[test]
790    fn decode_varchar_short_prefix() {
791        // VARCHAR with max_len < 256 → 1-byte length prefix.
792        let cols = vec![make_pk_col(MYSQL_TYPE_VARCHAR, false, 200, 0)];
793        let mut data = vec![0u8; 1]; // null bitmap
794        data.push(5); // length = 5
795        data.extend_from_slice(b"hello");
796        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
797        assert_eq!(pk, vec![BinlogPkValue::Str("hello".to_string())]);
798    }
799
800    #[test]
801    fn decode_varchar_long_prefix() {
802        // VARCHAR with max_len >= 256 → 2-byte length prefix.
803        let cols = vec![make_pk_col(MYSQL_TYPE_VARCHAR, false, 500, 0)];
804        let mut data = vec![0u8; 1]; // null bitmap
805        let mut len_buf = [0u8; 2];
806        LittleEndian::write_u16(&mut len_buf, 11);
807        data.extend_from_slice(&len_buf);
808        data.extend_from_slice(b"hello world");
809        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
810        assert_eq!(pk, vec![BinlogPkValue::Str("hello world".to_string())]);
811    }
812
813    // -- Null bitmap tests --
814
815    #[test]
816    fn null_pk_returns_none() {
817        let cols = vec![make_pk_col(MYSQL_TYPE_LONG, false, 0, 0)];
818        // Null bitmap: bit 0 set → column 0 is NULL.
819        let data = vec![0x01];
820        assert!(extract_pk_from_row_image(&data, &cols).is_none());
821    }
822
823    #[test]
824    fn null_non_pk_column_skipped() {
825        // Two columns: non-PK INT (NULL) + PK INT (42).
826        let cols = vec![
827            BinlogColumnMeta {
828                column_type: MYSQL_TYPE_LONG,
829                is_unsigned: true,
830                type_metadata: 0,
831                is_pk: false,
832                pk_ordinal: None,
833            },
834            make_pk_col(MYSQL_TYPE_LONG, true, 0, 0),
835        ];
836        let mut data = vec![0x01]; // bit 0 set → col 0 is NULL, col 1 is not
837        let mut buf = [0u8; 4];
838        LittleEndian::write_u32(&mut buf, 42);
839        data.extend_from_slice(&buf);
840        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
841        assert_eq!(pk, vec![BinlogPkValue::Uint(42)]);
842    }
843
844    // -- Multi-column PK tests --
845
846    #[test]
847    fn multi_column_pk_sorted_by_ordinal() {
848        // 3 columns: PK ord=1 (SHORT), non-PK (LONG), PK ord=0 (TINY).
849        let cols = vec![
850            BinlogColumnMeta {
851                column_type: MYSQL_TYPE_SHORT,
852                is_unsigned: true,
853                type_metadata: 0,
854                is_pk: true,
855                pk_ordinal: Some(1),
856            },
857            BinlogColumnMeta {
858                column_type: MYSQL_TYPE_LONG,
859                is_unsigned: true,
860                type_metadata: 0,
861                is_pk: false,
862                pk_ordinal: None,
863            },
864            BinlogColumnMeta {
865                column_type: MYSQL_TYPE_TINY,
866                is_unsigned: true,
867                type_metadata: 0,
868                is_pk: true,
869                pk_ordinal: Some(0),
870            },
871        ];
872        let mut data = vec![0u8; 1]; // null bitmap: all non-null
873                                     // Col 0 (SHORT): 1000
874        let mut buf = [0u8; 2];
875        LittleEndian::write_u16(&mut buf, 1000);
876        data.extend_from_slice(&buf);
877        // Col 1 (LONG): 99999
878        let mut buf4 = [0u8; 4];
879        LittleEndian::write_u32(&mut buf4, 99999);
880        data.extend_from_slice(&buf4);
881        // Col 2 (TINY): 7
882        data.push(7);
883
884        let pk = extract_pk_from_row_image(&data, &cols).unwrap();
885        // Sorted by ordinal: ord=0 (TINY 7) first, ord=1 (SHORT 1000) second.
886        assert_eq!(pk, vec![BinlogPkValue::Uint(7), BinlogPkValue::Uint(1000)]);
887    }
888
889    // -- column_value_size tests --
890
891    #[test]
892    fn size_varchar_short() {
893        let data = [5, b'h', b'e', b'l', b'l', b'o'];
894        assert_eq!(column_value_size(MYSQL_TYPE_VARCHAR, 200, &data), Some(6));
895    }
896
897    #[test]
898    fn size_varchar_long() {
899        let mut data = vec![0u8; 2];
900        LittleEndian::write_u16(&mut data, 3);
901        data.extend_from_slice(b"abc");
902        assert_eq!(column_value_size(MYSQL_TYPE_VARCHAR, 500, &data), Some(5));
903    }
904
905    #[test]
906    fn size_blob_pack2() {
907        let mut data = vec![0u8; 2];
908        LittleEndian::write_u16(&mut data, 10);
909        data.extend_from_slice(&[0u8; 10]);
910        assert_eq!(column_value_size(MYSQL_TYPE_BLOB, 2, &data), Some(12));
911    }
912
913    #[test]
914    fn size_datetime2_fsp3() {
915        // 5 base + ceil(3/2) = 5 + 2 = 7
916        assert_eq!(
917            column_value_size(MYSQL_TYPE_DATETIME2, 3, &[0u8; 10]),
918            Some(7)
919        );
920    }
921
922    // -- Display tests --
923
924    #[test]
925    fn pk_value_display() {
926        assert_eq!(BinlogPkValue::Int(-42).to_string(), "-42");
927        assert_eq!(BinlogPkValue::Uint(100).to_string(), "100");
928        assert_eq!(BinlogPkValue::Str("hello".into()).to_string(), "hello");
929        assert_eq!(BinlogPkValue::Bytes(vec![0xDE, 0xAD]).to_string(), "0xdead");
930    }
931
932    // -- decimal_binary_size tests --
933
934    #[test]
935    fn decimal_size_10_2() {
936        // DECIMAL(10,2): intg=8 → 0 full + 8 leftover (4 bytes), frac=2 → 0 full + 2 leftover (1 byte)
937        assert_eq!(decimal_binary_size(10, 2), 5);
938    }
939
940    #[test]
941    fn decimal_size_18_0() {
942        // DECIMAL(18,0): intg=18 → 2 full (8 bytes) + 0 leftover
943        assert_eq!(decimal_binary_size(18, 0), 8);
944    }
945
946    // -- Empty / edge cases --
947
948    #[test]
949    fn empty_columns_returns_none() {
950        assert!(extract_pk_from_row_image(&[0], &[]).is_none());
951    }
952
953    #[test]
954    fn no_pk_columns_returns_none() {
955        let cols = vec![BinlogColumnMeta {
956            column_type: MYSQL_TYPE_LONG,
957            is_unsigned: true,
958            type_metadata: 0,
959            is_pk: false,
960            pk_ordinal: None,
961        }];
962        let mut data = vec![0u8; 1];
963        data.extend_from_slice(&[0u8; 4]);
964        assert!(extract_pk_from_row_image(&data, &cols).is_none());
965    }
966
967    #[test]
968    fn truncated_data_returns_none() {
969        let cols = vec![make_pk_col(MYSQL_TYPE_LONGLONG, false, 0, 0)];
970        let data = vec![0u8; 1]; // null bitmap only, no value bytes
971        assert!(extract_pk_from_row_image(&data, &cols).is_none());
972    }
973
974    // -- Helper --
975
976    fn make_pk_col(col_type: u8, unsigned: bool, meta: u16, ordinal: usize) -> BinlogColumnMeta {
977        BinlogColumnMeta {
978            column_type: col_type,
979            is_unsigned: unsigned,
980            type_metadata: meta,
981            is_pk: true,
982            pk_ordinal: Some(ordinal),
983        }
984    }
985}
idb/binlog/row_image.rs

idb/binlog/
row_image.rs
