use std::collections::HashMap;
use std::io::Write;
use serde::Serialize;
use crate::cli::{csv_escape, wprintln};
use crate::IdbError;
pub struct BinlogOptions {
pub file: String,
pub limit: Option<usize>,
pub filter_type: Option<String>,
pub verbose: bool,
pub json: bool,
pub csv: bool,
pub correlate: Option<String>,
}
#[derive(Serialize)]
struct CorrelatedBinlogAnalysis {
#[serde(flatten)]
analysis: crate::binlog::BinlogAnalysis,
correlated_events: Vec<crate::binlog::CorrelatedEvent>,
}
pub fn execute(opts: &BinlogOptions, writer: &mut dyn Write) -> Result<(), IdbError> {
if opts.correlate.is_some() {
return execute_correlated(opts, writer);
}
let file = std::fs::File::open(&opts.file)
.map_err(|e| IdbError::Io(format!("{}: {}", opts.file, e)))?;
let reader = std::io::BufReader::new(file);
let analysis = crate::binlog::analyze_binlog(reader)?;
if opts.json {
let json =
serde_json::to_string_pretty(&analysis).map_err(|e| IdbError::Parse(e.to_string()))?;
wprintln!(writer, "{}", json)?;
return Ok(());
}
if opts.csv {
return write_csv(&analysis, opts, writer);
}
write_text(&analysis, opts, writer)
}
fn execute_correlated(opts: &BinlogOptions, writer: &mut dyn Write) -> Result<(), IdbError> {
let ts_path = opts.correlate.as_ref().unwrap();
let mut binlog = crate::binlog::BinlogFile::open(&opts.file)?;
let mut ts = crate::cli::open_tablespace(ts_path, None, false)?;
let correlated = crate::binlog::correlate_events(&mut binlog, &mut ts)?;
let file = std::fs::File::open(&opts.file)
.map_err(|e| IdbError::Io(format!("{}: {}", opts.file, e)))?;
let reader = std::io::BufReader::new(file);
let analysis = crate::binlog::analyze_binlog(reader)?;
if opts.json {
let combined = CorrelatedBinlogAnalysis {
analysis,
correlated_events: correlated,
};
let json =
serde_json::to_string_pretty(&combined).map_err(|e| IdbError::Parse(e.to_string()))?;
wprintln!(writer, "{}", json)?;
return Ok(());
}
let correlated_map: HashMap<u64, &crate::binlog::CorrelatedEvent> =
correlated.iter().map(|e| (e.binlog_pos, e)).collect();
if opts.csv {
return write_correlated_csv(&analysis, &correlated_map, opts, writer);
}
write_correlated_text(&analysis, &correlated_map, opts, writer)
}
fn write_correlated_text(
analysis: &crate::binlog::BinlogAnalysis,
correlated: &HashMap<u64, &crate::binlog::CorrelatedEvent>,
opts: &BinlogOptions,
writer: &mut dyn Write,
) -> Result<(), IdbError> {
wprintln!(writer, "Binary Log: {}", opts.file)?;
wprintln!(
writer,
" Server Version: {}",
analysis.format_description.server_version
)?;
wprintln!(
writer,
" Binlog Version: {}",
analysis.format_description.binlog_version
)?;
wprintln!(
writer,
" Checksum Algorithm: {}",
analysis.format_description.checksum_alg
)?;
wprintln!(
writer,
" Correlated Events: {} (tablespace: {})",
correlated.len(),
opts.correlate.as_deref().unwrap_or("--")
)?;
wprintln!(writer)?;
wprintln!(
writer,
"Event Type Summary ({} total):",
analysis.event_count
)?;
let mut type_counts: Vec<_> = analysis.event_type_counts.iter().collect();
type_counts.sort_by(|a, b| b.1.cmp(a.1));
for (name, count) in &type_counts {
wprintln!(writer, " {:<30} {:>6}", name, count)?;
}
wprintln!(writer)?;
if !analysis.table_maps.is_empty() {
wprintln!(writer, "Table Maps ({}):", analysis.table_maps.len())?;
for tm in &analysis.table_maps {
wprintln!(
writer,
" table_id={} {}.{} ({} columns)",
tm.table_id,
tm.database_name,
tm.table_name,
tm.column_count
)?;
if opts.verbose {
wprintln!(writer, " Column types: {:?}", &tm.column_types)?;
}
}
wprintln!(writer)?;
}
let events = filter_events(&analysis.events, opts);
let limit = opts.limit.unwrap_or(events.len());
let display_events = &events[..limit.min(events.len())];
if !display_events.is_empty() {
wprintln!(
writer,
"{:<12} {:<30} {:<10} {:<12} {:<8} {}",
"Position",
"Type",
"Size",
"Timestamp",
"Page",
"PK"
)?;
wprintln!(writer, "{}", "-".repeat(90))?;
for evt in display_events {
if let Some(ce) = correlated.get(&evt.offset) {
let pk_display = if ce.pk_values.is_empty() {
"--".to_string()
} else {
format!("({})", ce.pk_values.join(", "))
};
wprintln!(
writer,
"{:<12} {:<30} {:<10} {:<12} {:<8} {}",
evt.offset,
evt.event_type,
evt.event_length,
evt.timestamp,
ce.page_no,
pk_display
)?;
} else {
wprintln!(
writer,
"{:<12} {:<30} {:<10} {:<12} {:<8} {}",
evt.offset,
evt.event_type,
evt.event_length,
evt.timestamp,
"--",
"--"
)?;
}
}
}
if events.len() > limit {
wprintln!(
writer,
"\n... {} more events (use --limit to show more)",
events.len() - limit
)?;
}
Ok(())
}
fn write_correlated_csv(
analysis: &crate::binlog::BinlogAnalysis,
correlated: &HashMap<u64, &crate::binlog::CorrelatedEvent>,
opts: &BinlogOptions,
writer: &mut dyn Write,
) -> Result<(), IdbError> {
wprintln!(
writer,
"position,type,size,timestamp,server_id,page_no,space_id,pk_values"
)?;
let events = filter_events(&analysis.events, opts);
let limit = opts.limit.unwrap_or(events.len());
let display_events = &events[..limit.min(events.len())];
for evt in display_events {
if let Some(ce) = correlated.get(&evt.offset) {
let pk_str = ce.pk_values.join(";");
wprintln!(
writer,
"{},{},{},{},{},{},{},{}",
evt.offset,
csv_escape(&evt.event_type),
evt.event_length,
evt.timestamp,
evt.server_id,
ce.page_no,
ce.space_id,
csv_escape(&pk_str)
)?;
} else {
wprintln!(
writer,
"{},{},{},{},{},,,",
evt.offset,
csv_escape(&evt.event_type),
evt.event_length,
evt.timestamp,
evt.server_id
)?;
}
}
Ok(())
}
fn write_text(
analysis: &crate::binlog::BinlogAnalysis,
opts: &BinlogOptions,
writer: &mut dyn Write,
) -> Result<(), IdbError> {
wprintln!(writer, "Binary Log: {}", opts.file)?;
wprintln!(
writer,
" Server Version: {}",
analysis.format_description.server_version
)?;
wprintln!(
writer,
" Binlog Version: {}",
analysis.format_description.binlog_version
)?;
wprintln!(
writer,
" Checksum Algorithm: {}",
analysis.format_description.checksum_alg
)?;
wprintln!(writer)?;
wprintln!(
writer,
"Event Type Summary ({} total):",
analysis.event_count
)?;
let mut type_counts: Vec<_> = analysis.event_type_counts.iter().collect();
type_counts.sort_by(|a, b| b.1.cmp(a.1));
for (name, count) in &type_counts {
wprintln!(writer, " {:<30} {:>6}", name, count)?;
}
wprintln!(writer)?;
if !analysis.table_maps.is_empty() {
wprintln!(writer, "Table Maps ({}):", analysis.table_maps.len())?;
for tm in &analysis.table_maps {
wprintln!(
writer,
" table_id={} {}.{} ({} columns)",
tm.table_id,
tm.database_name,
tm.table_name,
tm.column_count
)?;
if opts.verbose {
wprintln!(writer, " Column types: {:?}", &tm.column_types)?;
}
}
wprintln!(writer)?;
}
let events = filter_events(&analysis.events, opts);
let limit = opts.limit.unwrap_or(events.len());
let display_events = &events[..limit.min(events.len())];
if !display_events.is_empty() {
wprintln!(
writer,
"{:<12} {:<30} {:<10} {:<12}",
"Position",
"Type",
"Size",
"Timestamp"
)?;
wprintln!(writer, "{}", "-".repeat(66))?;
for evt in display_events {
wprintln!(
writer,
"{:<12} {:<30} {:<10} {:<12}",
evt.offset,
evt.event_type,
evt.event_length,
evt.timestamp
)?;
}
}
if events.len() > limit {
wprintln!(
writer,
"\n... {} more events (use --limit to show more)",
events.len() - limit
)?;
}
Ok(())
}
fn write_csv(
analysis: &crate::binlog::BinlogAnalysis,
opts: &BinlogOptions,
writer: &mut dyn Write,
) -> Result<(), IdbError> {
wprintln!(writer, "position,type,size,timestamp,server_id")?;
let events = filter_events(&analysis.events, opts);
let limit = opts.limit.unwrap_or(events.len());
let display_events = &events[..limit.min(events.len())];
for evt in display_events {
wprintln!(
writer,
"{},{},{},{},{}",
evt.offset,
csv_escape(&evt.event_type),
evt.event_length,
evt.timestamp,
evt.server_id
)?;
}
Ok(())
}
fn filter_events<'a>(
events: &'a [crate::binlog::BinlogEventSummary],
opts: &BinlogOptions,
) -> Vec<&'a crate::binlog::BinlogEventSummary> {
match &opts.filter_type {
Some(filter) => {
let filter_upper = filter.to_uppercase();
events
.iter()
.filter(|e| e.event_type.to_uppercase().contains(&filter_upper))
.collect()
}
None => events.iter().collect(),
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::binlog::{BinlogAnalysis, BinlogEventSummary, FormatDescriptionEvent};
fn sample_analysis() -> BinlogAnalysis {
let mut event_type_counts = HashMap::new();
event_type_counts.insert("QUERY_EVENT".to_string(), 5);
event_type_counts.insert("TABLE_MAP_EVENT".to_string(), 2);
BinlogAnalysis {
format_description: FormatDescriptionEvent {
binlog_version: 4,
server_version: "8.0.35".to_string(),
create_timestamp: 0,
header_length: 19,
checksum_alg: 1,
},
event_count: 7,
event_type_counts,
table_maps: Vec::new(),
events: vec![
BinlogEventSummary {
offset: 4,
event_type: "FORMAT_DESCRIPTION_EVENT".to_string(),
type_code: 15,
event_length: 119,
timestamp: 1700000000,
server_id: 1,
},
BinlogEventSummary {
offset: 123,
event_type: "QUERY_EVENT".to_string(),
type_code: 2,
event_length: 50,
timestamp: 1700000001,
server_id: 1,
},
],
}
}
#[test]
fn test_write_text_output() {
let analysis = sample_analysis();
let opts = BinlogOptions {
file: "test-bin.000001".to_string(),
limit: None,
filter_type: None,
verbose: false,
json: false,
csv: false,
correlate: None,
};
let mut buf = Vec::new();
write_text(&analysis, &opts, &mut buf).unwrap();
let output = String::from_utf8(buf).unwrap();
assert!(output.contains("Binary Log: test-bin.000001"));
assert!(output.contains("Server Version: 8.0.35"));
assert!(output.contains("7 total"));
assert!(output.contains("FORMAT_DESCRIPTION_EVENT"));
}
#[test]
fn test_write_csv_output() {
let analysis = sample_analysis();
let opts = BinlogOptions {
file: "test-bin.000001".to_string(),
limit: None,
filter_type: None,
verbose: false,
json: false,
csv: true,
correlate: None,
};
let mut buf = Vec::new();
write_csv(&analysis, &opts, &mut buf).unwrap();
let output = String::from_utf8(buf).unwrap();
assert!(output.starts_with("position,type,size,timestamp,server_id"));
assert!(output.contains("FORMAT_DESCRIPTION_EVENT"));
}
#[test]
fn test_filter_events_by_type() {
let analysis = sample_analysis();
let opts = BinlogOptions {
file: "test".to_string(),
limit: None,
filter_type: Some("query".to_string()),
verbose: false,
json: false,
csv: false,
correlate: None,
};
let filtered = filter_events(&analysis.events, &opts);
assert_eq!(filtered.len(), 1);
assert_eq!(filtered[0].event_type, "QUERY_EVENT");
}
#[test]
fn test_json_output() {
let analysis = sample_analysis();
let json = serde_json::to_string_pretty(&analysis).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed["event_count"], 7);
}
#[test]
fn test_correlated_text_output() {
let analysis = sample_analysis();
let ce = crate::binlog::CorrelatedEvent {
binlog_pos: 123,
event_type: crate::binlog::RowEventType::Insert,
database: "test".to_string(),
table: "users".to_string(),
page_no: 42,
space_id: 5,
page_lsn: 999,
pk_values: vec!["1".to_string(), "alice".to_string()],
timestamp: 1700000001,
};
let correlated: HashMap<u64, &crate::binlog::CorrelatedEvent> =
[(123, &ce)].into_iter().collect();
let opts = BinlogOptions {
file: "test-bin.000001".to_string(),
limit: None,
filter_type: None,
verbose: false,
json: false,
csv: false,
correlate: Some("/tmp/users.ibd".to_string()),
};
let mut buf = Vec::new();
write_correlated_text(&analysis, &correlated, &opts, &mut buf).unwrap();
let output = String::from_utf8(buf).unwrap();
assert!(output.contains("Correlated Events: 1"));
assert!(output.contains("Page"));
assert!(output.contains("PK"));
assert!(output.contains("42"));
assert!(output.contains("(1, alice)"));
let lines: Vec<&str> = output.lines().collect();
let fde_line = lines
.iter()
.find(|l| l.contains("FORMAT_DESCRIPTION"))
.unwrap();
assert!(fde_line.contains("--"));
}
#[test]
fn test_correlated_csv_output() {
let analysis = sample_analysis();
let ce = crate::binlog::CorrelatedEvent {
binlog_pos: 123,
event_type: crate::binlog::RowEventType::Insert,
database: "test".to_string(),
table: "users".to_string(),
page_no: 42,
space_id: 5,
page_lsn: 999,
pk_values: vec!["1".to_string()],
timestamp: 1700000001,
};
let correlated: HashMap<u64, &crate::binlog::CorrelatedEvent> =
[(123, &ce)].into_iter().collect();
let opts = BinlogOptions {
file: "test-bin.000001".to_string(),
limit: None,
filter_type: None,
verbose: false,
json: false,
csv: true,
correlate: Some("/tmp/users.ibd".to_string()),
};
let mut buf = Vec::new();
write_correlated_csv(&analysis, &correlated, &opts, &mut buf).unwrap();
let output = String::from_utf8(buf).unwrap();
assert!(
output.starts_with("position,type,size,timestamp,server_id,page_no,space_id,pk_values")
);
assert!(output.contains(",42,5,"));
let fde_line = output
.lines()
.find(|l| l.contains("FORMAT_DESCRIPTION"))
.unwrap();
assert!(fde_line.ends_with(",,,"));
}
#[test]
fn test_correlated_json_output() {
let analysis = sample_analysis();
let correlated = vec![crate::binlog::CorrelatedEvent {
binlog_pos: 123,
event_type: crate::binlog::RowEventType::Insert,
database: "test".to_string(),
table: "users".to_string(),
page_no: 42,
space_id: 5,
page_lsn: 999,
pk_values: vec!["1".to_string()],
timestamp: 1700000001,
}];
let combined = CorrelatedBinlogAnalysis {
analysis,
correlated_events: correlated,
};
let json = serde_json::to_string_pretty(&combined).unwrap();
let parsed: serde_json::Value = serde_json::from_str(&json).unwrap();
assert_eq!(parsed["event_count"], 7);
assert_eq!(parsed["correlated_events"][0]["page_no"], 42);
assert_eq!(parsed["correlated_events"][0]["space_id"], 5);
assert_eq!(parsed["correlated_events"][0]["event_type"], "Insert");
}
}