1use serde::Serialize;
7
8#[cfg(feature = "postgres")]
9use tokio_postgres::Client;
10
11use crate::commands::info::{self, MigrationState};
12use crate::config::WaypointConfig;
13use crate::db::DbClient;
14use crate::dialect::DialectKind;
15use crate::error::Result;
16#[cfg(any(not(feature = "postgres"), not(feature = "mysql")))]
17use crate::error::WaypointError;
18use crate::placeholder::{build_placeholders, replace_placeholders};
19#[cfg(feature = "postgres")]
20use crate::sql_parser::split_statements;
21
22#[derive(Debug, Serialize)]
24pub struct ExplainReport {
25 pub migrations: Vec<MigrationExplain>,
27}
28
29#[derive(Debug, Serialize)]
31pub struct MigrationExplain {
32 pub script: String,
34 pub version: Option<String>,
36 pub statements: Vec<StatementExplain>,
38}
39
40#[derive(Debug, Serialize)]
42pub struct StatementExplain {
43 pub statement_preview: String,
45 pub plan: String,
47 pub estimated_rows: Option<f64>,
49 pub estimated_cost: Option<f64>,
51 pub warnings: Vec<String>,
53 pub is_ddl: bool,
55}
56
57#[cfg(feature = "postgres")]
59pub async fn execute(client: &Client, config: &WaypointConfig) -> Result<ExplainReport> {
60 let infos = info::execute(client, config).await?;
61
62 let pending: Vec<_> = infos
63 .iter()
64 .filter(|i| matches!(i.state, MigrationState::Pending | MigrationState::Outdated))
65 .collect();
66
67 let schema = &config.migrations.schema;
68 let db_user = crate::db::get_current_user(client)
69 .await
70 .unwrap_or_else(|_| "unknown".to_string());
71 let db_name = crate::db::get_current_database(client)
72 .await
73 .unwrap_or_else(|_| "unknown".to_string());
74
75 let resolved = crate::migration::scan_migrations(&config.migrations.locations)?;
77
78 let mut migrations = Vec::new();
79
80 for info in &pending {
81 let migration = resolved.iter().find(|m| m.script == info.script);
83 let sql = match migration {
84 Some(m) => {
85 let placeholders =
86 build_placeholders(&config.placeholders, schema, &db_user, &db_name, &m.script);
87 replace_placeholders(&m.sql, &placeholders)?
88 }
89 None => continue,
90 };
91
92 let statements_raw = split_statements(&sql);
93 let mut statements = Vec::new();
94
95 client.batch_execute("BEGIN").await?;
97
98 for stmt_str in &statements_raw {
99 let trimmed = stmt_str.trim();
100 if trimmed.is_empty() || trimmed.starts_with("--") {
101 continue;
102 }
103
104 let preview: String = trimmed.chars().take(80).collect();
105 let preview = if trimmed.len() > 80 {
106 format!("{}...", preview)
107 } else {
108 preview
109 };
110
111 let upper = trimmed.to_uppercase();
112 let is_ddl = upper.starts_with("CREATE")
113 || upper.starts_with("ALTER")
114 || upper.starts_with("DROP")
115 || upper.starts_with("TRUNCATE");
116
117 if is_ddl {
118 match client.batch_execute(trimmed).await {
120 Ok(()) => {}
121 Err(e) => {
122 log::debug!("DDL statement failed during explain: {}", e);
123 }
124 }
125 statements.push(StatementExplain {
126 statement_preview: preview,
127 plan: "DDL statement — not explainable".to_string(),
128 estimated_rows: None,
129 estimated_cost: None,
130 warnings: vec![],
131 is_ddl: true,
132 });
133 } else {
134 let explain_sql = format!("EXPLAIN (FORMAT TEXT) {}", trimmed);
136 match client.query(&explain_sql, &[]).await {
137 Ok(rows_result) => {
138 let plan_lines: Vec<String> =
139 rows_result.iter().map(|r| r.get::<_, String>(0)).collect();
140 let plan_str = plan_lines.join("\n");
141
142 let (rows, cost, warnings) = extract_plan_info_text(&plan_str);
143
144 statements.push(StatementExplain {
145 statement_preview: preview,
146 plan: plan_str,
147 estimated_rows: rows,
148 estimated_cost: cost,
149 warnings,
150 is_ddl: false,
151 });
152 }
153 Err(e) => {
154 statements.push(StatementExplain {
155 statement_preview: preview,
156 plan: format!("EXPLAIN failed: {}", e),
157 estimated_rows: None,
158 estimated_cost: None,
159 warnings: vec![],
160 is_ddl: false,
161 });
162 }
163 }
164 }
165 }
166
167 let _ = client.batch_execute("ROLLBACK").await;
169
170 migrations.push(MigrationExplain {
171 script: info.script.clone(),
172 version: info.version.clone(),
173 statements,
174 });
175 }
176
177 Ok(ExplainReport { migrations })
178}
179
180pub async fn execute_db(client: &DbClient, config: &WaypointConfig) -> Result<ExplainReport> {
182 match client.dialect_kind() {
183 #[cfg(feature = "postgres")]
184 DialectKind::Postgres => execute(client.as_postgres()?, config).await,
185 #[cfg(not(feature = "postgres"))]
186 DialectKind::Postgres => Err(WaypointError::ConfigError(
187 "PostgreSQL support is not compiled in (enable the `postgres` feature)".into(),
188 )),
189 #[cfg(feature = "mysql")]
190 DialectKind::Mysql => execute_mysql(client, config).await,
191 #[cfg(not(feature = "mysql"))]
192 DialectKind::Mysql => Err(WaypointError::ConfigError(
193 "MySQL support is not compiled in (enable the `mysql` feature)".into(),
194 )),
195 }
196}
197
198#[cfg(feature = "mysql")]
207async fn execute_mysql(client: &DbClient, config: &WaypointConfig) -> Result<ExplainReport> {
208 use mysql_async::prelude::*;
209 let pool = client.as_mysql()?;
210 let infos = info::execute_db(client, config).await?;
211
212 let pending: Vec<_> = infos
213 .iter()
214 .filter(|i| matches!(i.state, MigrationState::Pending | MigrationState::Outdated))
215 .collect();
216
217 let schema = client.resolve_schema(&config.migrations.schema).await?;
218 let db_user = client
219 .current_user()
220 .await
221 .unwrap_or_else(|_| "unknown".into());
222 let db_name = client
223 .current_database()
224 .await
225 .unwrap_or_else(|_| "unknown".into());
226
227 let resolved = crate::migration::scan_migrations(&config.migrations.locations)?;
228 let mut migrations = Vec::new();
229
230 for info in &pending {
231 let migration = resolved.iter().find(|m| m.script == info.script);
232 let sql = match migration {
233 Some(m) => {
234 let placeholders = build_placeholders(
235 &config.placeholders,
236 &schema,
237 &db_user,
238 &db_name,
239 &m.script,
240 );
241 replace_placeholders(&m.sql, &placeholders)?
242 }
243 None => continue,
244 };
245
246 let mut statements = Vec::new();
247 let mut conn = pool.get_conn().await?;
248
249 for stmt_str in crate::sql_parser::split_mysql_statements(&sql) {
250 let trimmed = stmt_str.trim();
251 if trimmed.is_empty() {
252 continue;
253 }
254
255 let preview: String = trimmed.chars().take(80).collect();
256 let preview = if trimmed.len() > 80 {
257 format!("{}...", preview)
258 } else {
259 preview
260 };
261
262 let upper = trimmed.to_uppercase();
263 let is_ddl = upper.starts_with("CREATE")
264 || upper.starts_with("ALTER")
265 || upper.starts_with("DROP")
266 || upper.starts_with("TRUNCATE")
267 || upper.starts_with("RENAME");
268
269 if is_ddl {
270 statements.push(StatementExplain {
271 statement_preview: preview,
272 plan: "DDL statement — not explainable".to_string(),
273 estimated_rows: None,
274 estimated_cost: None,
275 warnings: vec![],
276 is_ddl: true,
277 });
278 } else {
279 let explain_sql = format!("EXPLAIN FORMAT=JSON {}", trimmed);
280 match conn.query_first::<String, _>(&explain_sql).await {
281 Ok(Some(plan_json)) => {
282 let (rows, warnings) = extract_plan_info_mysql(&plan_json);
283 statements.push(StatementExplain {
284 statement_preview: preview,
285 plan: plan_json,
286 estimated_rows: rows,
287 estimated_cost: None, warnings,
289 is_ddl: false,
290 });
291 }
292 Ok(None) => statements.push(StatementExplain {
293 statement_preview: preview,
294 plan: "EXPLAIN produced no rows".to_string(),
295 estimated_rows: None,
296 estimated_cost: None,
297 warnings: vec![],
298 is_ddl: false,
299 }),
300 Err(e) => statements.push(StatementExplain {
301 statement_preview: preview,
302 plan: format!("EXPLAIN failed: {}", e),
303 estimated_rows: None,
304 estimated_cost: None,
305 warnings: vec![],
306 is_ddl: false,
307 }),
308 }
309 }
310 }
311
312 migrations.push(MigrationExplain {
313 script: info.script.clone(),
314 version: info.version.clone(),
315 statements,
316 });
317 }
318
319 Ok(ExplainReport { migrations })
320}
321
322#[cfg(feature = "mysql")]
327fn extract_plan_info_mysql(plan: &str) -> (Option<f64>, Vec<String>) {
328 let mut warnings = Vec::new();
329 let mut rows = None;
330
331 if let Some(idx) = plan.find("\"rows_examined_per_scan\":") {
333 let after = &plan[idx + "\"rows_examined_per_scan\":".len()..];
334 let after = after.trim_start();
335 let end = after
336 .find(|c: char| !c.is_ascii_digit())
337 .unwrap_or(after.len());
338 if let Ok(r) = after[..end].parse::<f64>() {
339 rows = Some(r);
340 }
341 }
342
343 if plan.contains("\"access_type\": \"ALL\"") || plan.contains("\"access_type\":\"ALL\"") {
345 let big_table = rows.map(|r| r > 10_000.0).unwrap_or(false);
346 if big_table {
347 warnings.push(format!(
348 "Full table scan (~{:.0} rows) — consider adding an index",
349 rows.unwrap_or(0.0)
350 ));
351 } else {
352 warnings.push("Full table scan detected — consider adding an index".to_string());
353 }
354 }
355
356 (rows, warnings)
357}
358
359#[cfg(feature = "postgres")]
360fn extract_plan_info_text(plan_text: &str) -> (Option<f64>, Option<f64>, Vec<String>) {
361 let mut warnings = Vec::new();
362 let mut total_rows = None;
363 let mut total_cost = None;
364
365 for line in plan_text.lines() {
367 let trimmed = line.trim();
368 if let Some(cost_start) = trimmed.find("cost=") {
369 let rest = &trimmed[cost_start + 5..];
370 if let Some(dot_dot) = rest.find("..") {
371 let after_dots = &rest[dot_dot + 2..];
372 if let Some(space_pos) = after_dots.find(' ') {
373 if let Ok(cost) = after_dots[..space_pos].parse::<f64>() {
374 if total_cost.is_none() {
375 total_cost = Some(cost);
376 }
377 }
378 }
379 }
380 }
381 if let Some(rows_start) = trimmed.find("rows=") {
382 let rest = &trimmed[rows_start + 5..];
383 let end = rest
384 .find(|c: char| !c.is_ascii_digit())
385 .unwrap_or(rest.len());
386 if let Ok(rows) = rest[..end].parse::<f64>() {
387 if total_rows.is_none() {
388 total_rows = Some(rows);
389 }
390 }
391 }
392
393 if trimmed.contains("Seq Scan") {
395 if let Some(rows) = total_rows {
396 if rows > 10000.0 {
397 let table = trimmed
399 .find("on ")
400 .map(|i| {
401 let after = &trimmed[i + 3..];
402 after.split_whitespace().next().unwrap_or("unknown")
403 })
404 .unwrap_or("unknown");
405 warnings.push(format!(
406 "Sequential Scan on '{}' (~{:.0} rows) — consider adding an index",
407 table, rows
408 ));
409 }
410 }
411 }
412 }
413
414 (total_rows, total_cost, warnings)
415}