diff --git a/apps/desktop/src/components/export/ExportProgressPopover.vue b/apps/desktop/src/components/export/ExportProgressPopover.vue index 57f8421da6..4410436277 100644 --- a/apps/desktop/src/components/export/ExportProgressPopover.vue +++ b/apps/desktop/src/components/export/ExportProgressPopover.vue @@ -6,6 +6,7 @@ import { Button } from "@/components/ui/button"; import { Loader2, Check, CheckCircle2, XCircle, AlertCircle, X, FileDown, DatabaseBackup, FileCode2, ArrowRightLeft, Layers3, GitCompareArrows, ChevronRight, FolderOpen, Copy } from "@lucide/vue"; import { formatDataTransferDuration, useExportTracker, type ExportTask } from "@/composables/useExportTracker"; import { dataTransferFailureCopyText, sqlFileFailureCopyText } from "@/components/export/failureDetailCopyText"; +import SqlFileProgressIndicator from "@/components/sql-file/SqlFileProgressIndicator.vue"; import { translateBackendError } from "@/i18n/backend-errors"; import { useToast } from "@/composables/useToast"; import { isTauriRuntime } from "@/lib/backend/tauriRuntime"; @@ -65,12 +66,6 @@ const progressValue = (task: ExportTask) => { if (!task.totalObjects || task.totalObjects <= 0) return 0; return Math.min(100, Math.round(((task.objectIndex ?? 0) / task.totalObjects) * 100)); } - if (task.kind === "sql-file") { - const total = task.totalRows ?? task.statementIndex ?? 0; - if (task.status === "Done") return 100; - if (total <= 0) return 0; - return Math.min(95, Math.round((task.rowsExported / total) * 100)); - } if (task.kind === "data-transfer") { if (!task.totalTables || task.totalTables <= 0) return 0; if (task.status === "Done") return 100; @@ -358,7 +353,8 @@ function openTask(task: ExportTask): void { -
+ +
{{ rowsText(task) }} + {{ t("exportProgress.elapsed", { duration: formatDataTransferDuration(task.elapsedMs) }) }} {{ elapsedText(task) }} {{ taskStatusText(task) }} diff --git a/apps/desktop/src/components/export/__tests__/ExportProgressPopover.spec.ts b/apps/desktop/src/components/export/__tests__/ExportProgressPopover.spec.ts index d32e452b29..2c0838250b 100644 --- a/apps/desktop/src/components/export/__tests__/ExportProgressPopover.spec.ts +++ b/apps/desktop/src/components/export/__tests__/ExportProgressPopover.spec.ts @@ -94,6 +94,61 @@ describe("failure detail copy text", () => { }); }); +describe("SQL file byte progress", () => { + it("keeps legacy progress indeterminate instead of using successful statements as the total", async () => { + const tracker = useExportTracker(); + const task = tracker.addSqlFileTask("legacy-sql-progress", "large.sql", "/tmp/large.sql"); + tracker.updateSqlFileTask(task.exportId, { + executionId: task.exportId, + status: "running", + statementIndex: 772205, + successCount: 772205, + failureCount: 0, + affectedRows: 0, + elapsedMs: 1000, + statementSummary: "INSERT INTO users VALUES (1)", + }); + await mountPopover(); + expect(document.querySelector('[data-testid="sql-file-progress"] [role="progressbar"]')?.hasAttribute("aria-valuenow")).toBe(false); + expect(task.totalRows).toBeNull(); + }); + + it.each(["done", "error", "cancelled"] as const)("shows byte progress and preserves it on %s", async (status) => { + const tracker = useExportTracker(); + const task = tracker.addSqlFileTask(`byte-progress-${status}`, "large.sql", "/tmp/large.sql"); + const progress = { + executionId: task.exportId, + status: "running" as const, + statementIndex: 772205, + successCount: 772205, + failureCount: 0, + affectedRows: 0, + elapsedMs: 1000, + statementSummary: "INSERT INTO users VALUES (1)", + bytesRead: 1024, + totalBytes: 4096, + phase: "executing" as const, + }; + tracker.updateSqlFileTask(task.exportId, progress); + await mountPopover(); + const indicator = document.querySelector('[data-testid="sql-file-progress"]')!; + expect(indicator.textContent).toContain("Executing SQL"); + expect(indicator.textContent).toContain("Read 1.0 KB / 4.0 KB"); + expect(indicator.querySelector('[role="progressbar"]')?.getAttribute("aria-valuenow")).toBe("25"); + + tracker.updateSqlFileTask(task.exportId, { ...progress, status: "statementDone", fileIndex: 1, statementIndex: 2, successCount: 2, bytesRead: 2048 }); + await nextTick(); + expect(indicator.querySelector('[role="progressbar"]')?.getAttribute("aria-valuenow")).toBe("50"); + + tracker.updateSqlFileTask(task.exportId, { ...progress, status, bytesRead: undefined, totalBytes: undefined, phase: undefined }); + await nextTick(); + expect(indicator.querySelector('[role="progressbar"]')?.getAttribute("aria-valuenow")).toBe(status === "done" ? "100" : "50"); + expect(task.bytesRead).toBe(2048); + expect(task.totalBytes).toBe(4096); + expect(task.elapsedMs).toBe(1000); + }); +}); + describe("ExportProgressPopover task duration", () => { it("distinguishes manual exports from scheduled backups and uses a compact failure dot", async () => { const tracker = useExportTracker(); diff --git a/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.retry.spec.ts b/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.retry.spec.ts index c083fa55ba..1863481961 100644 --- a/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.retry.spec.ts +++ b/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.retry.spec.ts @@ -223,6 +223,21 @@ afterEach(() => { }); describe("SqlFileExecutionDialog retries", () => { + it("shows byte-based progress while SQL executes and only completes on a terminal event", async () => { + await mountReadyDialog(); + const executionGate = deferred(); + mocks.executeSqlFiles.mockImplementationOnce(async (request: { executionId: string }) => { + mocks.progressHandler?.(progress(request.executionId, "running", { statementIndex: 500, successCount: 499, bytesRead: 1024, totalBytes: 4096, phase: "executing" })); + await executionGate.promise; + mocks.progressHandler?.(progress(request.executionId, "done", { bytesRead: 4096, totalBytes: 4096 })); + }); + findButton("sqlFile.execute").click(); + await vi.waitFor(() => expect(root!.querySelector('[role="progressbar"]')?.getAttribute("aria-valuenow")).toBe("25")); + expect(root!.textContent).toContain("sqlFile.progressPhase.executing"); + executionGate.resolve(); + await vi.waitFor(() => expect(root!.querySelector('[role="progressbar"]')?.getAttribute("aria-valuenow")).toBe("100")); + }); + it("does not auto-select the first connection for an unassociated external file", async () => { root = document.createElement("div"); document.body.append(root); diff --git a/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.vue b/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.vue index ee9ff9507a..bff651096f 100644 --- a/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.vue +++ b/apps/desktop/src/components/sql-file/SqlFileExecutionDialog.vue @@ -21,6 +21,7 @@ import { fetchSqlFileTargetOptions } from "@/composables/useDatabaseOptions"; import { requiresSqlFileTargetDatabaseSelection, supportsConnectionLevelDatabaseBootstrap } from "@/lib/connection/connectionLevelDatabaseBootstrap"; import { cancelSqlFileExecution, executeSqlFiles, inspectSqlFileTables, listenSqlFileProgress, previewSqlFile, type SqlFilePreview, type SqlFileProgress, type SqlFileStatus, type SqlFileTable } from "@/lib/backend/api"; import { buildDisplayFileNames, tooltipText as computeTooltipText } from "./sqlFilePreviewLabel"; +import SqlFileProgressIndicator from "./SqlFileProgressIndicator.vue"; import { useExportTracker, type ExportTask } from "@/composables/useExportTracker"; import { translateBackendError } from "@/i18n/backend-errors"; import { Check, CheckSquare, ChevronRight, FileCode, FolderOpen, Loader2, Play, Square, X } from "@lucide/vue"; @@ -222,14 +223,6 @@ const statusIcon = computed(() => { return FileCode; }); -const progressPercent = computed(() => { - if (!progress.value) return 0; - if (terminalStatus.value === "done") return 100; - const attempted = progress.value.successCount + progress.value.failureCount; - const current = Math.max(progress.value.statementIndex, attempted); - if (current <= 0) return running.value ? 8 : 0; - return Math.min(95, Math.max(8, Math.round((attempted / current) * 100))); -}); const sqlFileFailures = computed(() => activeExecutionTask.value?.sqlFileFailures ?? []); const sqlFileFailureCount = computed(() => sqlFileFailures.value.length + (activeExecutionTask.value?.sqlFileFailuresOmitted ?? 0)); const unlistedTerminalError = computed(() => { @@ -826,9 +819,7 @@ watch(
-
-
-
+
diff --git a/apps/desktop/src/components/sql-file/SqlFileProgressIndicator.vue b/apps/desktop/src/components/sql-file/SqlFileProgressIndicator.vue new file mode 100644 index 0000000000..d1d561f71b --- /dev/null +++ b/apps/desktop/src/components/sql-file/SqlFileProgressIndicator.vue @@ -0,0 +1,57 @@ + + + + + diff --git a/apps/desktop/src/composables/useExportTracker.ts b/apps/desktop/src/composables/useExportTracker.ts index 99e2937f7e..27d5aceb43 100644 --- a/apps/desktop/src/composables/useExportTracker.ts +++ b/apps/desktop/src/composables/useExportTracker.ts @@ -48,6 +48,9 @@ export interface ExportTask { failureCount?: number; affectedRows?: number; elapsedMs?: number; + bytesRead?: number; + totalBytes?: number; + sqlFilePhase?: api.SqlFileProgress["phase"]; startedAt?: number; finishedAt?: number; statementSummary?: string; @@ -661,9 +664,12 @@ export function useExportTracker() { task.failureCount = progress.failureCount; task.affectedRows = progress.affectedRows; task.elapsedMs = progress.elapsedMs; + task.bytesRead = progress.bytesRead ?? task.bytesRead; + task.totalBytes = progress.totalBytes ?? task.totalBytes; + task.sqlFilePhase = progress.phase ?? task.sqlFilePhase; task.statementSummary = progress.statementSummary; task.rowsExported = progress.successCount + progress.failureCount; - task.totalRows = Math.max(progress.statementIndex, progress.successCount + progress.failureCount) || null; + task.totalRows = null; } function updateDataTransferTask(transferId: string, progress: api.TransferProgress) { diff --git a/apps/desktop/src/i18n/locales/en.ts b/apps/desktop/src/i18n/locales/en.ts index 4ab505b81c..9db5943b29 100644 --- a/apps/desktop/src/i18n/locales/en.ts +++ b/apps/desktop/src/i18n/locales/en.ts @@ -6474,6 +6474,15 @@ export default { cancelled: "Cancelled", }, fileProgress: "File {current} of {total}", + byteProgress: "File progress", + byteProgressHint: "Based on file bytes read, not remaining execution time. Reaches 100% only when execution finishes. Compressed files use compressed bytes.", + bytesRead: "Read {read}", + bytesReadOfTotal: "Read {read} / {total}", + progressPhase: { + preparing: "Preparing SQL file", + reading: "Reading SQL file", + executing: "Executing SQL", + }, fileColumn: "File", totalFiles: "{count} files", }, diff --git a/apps/desktop/src/i18n/locales/zh-CN.ts b/apps/desktop/src/i18n/locales/zh-CN.ts index de861e161a..e62d2f96d7 100644 --- a/apps/desktop/src/i18n/locales/zh-CN.ts +++ b/apps/desktop/src/i18n/locales/zh-CN.ts @@ -6449,6 +6449,15 @@ export default withEnglishFallback({ cancelled: "已取消", }, fileProgress: "第 {current}/{total} 个文件", + byteProgress: "文件进度", + byteProgressHint: "按文件读取字节数计算，不代表剩余执行时间；执行结束后才显示 100%。压缩文件按压缩后的字节数计算。", + bytesRead: "已读取 {read}", + bytesReadOfTotal: "已读取 {read} / {total}", + progressPhase: { + preparing: "正在准备 SQL 文件", + reading: "正在读取 SQL 文件", + executing: "正在执行 SQL", + }, fileColumn: "文件", totalFiles: "{count} 个文件", }, diff --git a/apps/desktop/src/i18n/locales/zh-TW.ts b/apps/desktop/src/i18n/locales/zh-TW.ts index 6dfbee75eb..50489a6cfd 100644 --- a/apps/desktop/src/i18n/locales/zh-TW.ts +++ b/apps/desktop/src/i18n/locales/zh-TW.ts @@ -5410,6 +5410,15 @@ export default withEnglishFallback({ cancelled: "已取消", }, fileProgress: "第 {current}/{total} 個檔案", + byteProgress: "檔案進度", + byteProgressHint: "按檔案讀取位元組數計算，不代表剩餘執行時間；執行結束後才顯示 100%。壓縮檔案按壓縮後的位元組數計算。", + bytesRead: "已讀取 {read}", + bytesReadOfTotal: "已讀取 {read} / {total}", + progressPhase: { + preparing: "正在準備 SQL 檔案", + reading: "正在讀取 SQL 檔案", + executing: "正在執行 SQL", + }, fileColumn: "檔案", totalFiles: "{count} 個檔案", runInBackground: "背景執行", diff --git a/apps/desktop/src/lib/backend/tauri.ts b/apps/desktop/src/lib/backend/tauri.ts index f02ab450b1..291d98564f 100644 --- a/apps/desktop/src/lib/backend/tauri.ts +++ b/apps/desktop/src/lib/backend/tauri.ts @@ -4819,6 +4819,9 @@ export interface SqlFileProgress { elapsedMs: number; statementSummary: string; error?: string | null; + bytesRead?: number; + totalBytes?: number; + phase?: "preparing" | "reading" | "executing"; fileIndex?: number; fileName?: string; } diff --git a/apps/desktop/src/lib/sql/sqlFileProgress.spec.ts b/apps/desktop/src/lib/sql/sqlFileProgress.spec.ts new file mode 100644 index 0000000000..92ab82d074 --- /dev/null +++ b/apps/desktop/src/lib/sql/sqlFileProgress.spec.ts @@ -0,0 +1,34 @@ +import { describe, expect, it } from "vitest"; +import { formatSqlFileBytes, sqlFileProgressPercent } from "./sqlFileProgress"; + +describe("SQL file byte progress", () => { + it.each(["running", "statementDone", "statementFailed", "error", "cancelled", "Running", "Error", "Cancelled"])("uses bytes without reporting completion for %s", (status) => { + expect(sqlFileProgressPercent(status, 250, 1000)).toBe(25); + expect(sqlFileProgressPercent(status, 1000, 1000)).toBe(99); + expect(sqlFileProgressPercent(status, 1200, 1000)).toBe(99); + }); + + it.each(["done", "Done"])("completes empty files and legacy events only after %s", (status) => { + expect(sqlFileProgressPercent(status, 0, 0)).toBe(100); + expect(sqlFileProgressPercent(status)).toBe(100); + }); + + it("does not invent a percentage for missing or invalid totals", () => { + expect(sqlFileProgressPercent("running")).toBeNull(); + expect(sqlFileProgressPercent("running", 50)).toBeNull(); + expect(sqlFileProgressPercent("running", undefined, 100)).toBeNull(); + expect(sqlFileProgressPercent("running", 0, 0)).toBeNull(); + expect(sqlFileProgressPercent("running", -1, 100)).toBeNull(); + expect(sqlFileProgressPercent("running", 1, Number.NaN)).toBeNull(); + expect(sqlFileProgressPercent("running", Number.POSITIVE_INFINITY, 100)).toBeNull(); + expect(sqlFileProgressPercent("running", 0, 100)).toBe(0); + }); + + it("formats byte counts for small and large files", () => { + expect(formatSqlFileBytes(0)).toBe("0 B"); + expect(formatSqlFileBytes(512)).toBe("512 B"); + expect(formatSqlFileBytes(1536)).toBe("1.5 KB"); + expect(formatSqlFileBytes(5 * 1024 ** 3)).toBe("5.0 GB"); + expect(formatSqlFileBytes(1024 ** 4)).toBe("1.0 TB"); + }); +}); diff --git a/apps/desktop/src/lib/sql/sqlFileProgress.ts b/apps/desktop/src/lib/sql/sqlFileProgress.ts new file mode 100644 index 0000000000..e8370fae3c --- /dev/null +++ b/apps/desktop/src/lib/sql/sqlFileProgress.ts @@ -0,0 +1,16 @@ +export function sqlFileProgressPercent(status: string, bytesRead?: number, totalBytes?: number): number | null { + if (status.toLowerCase() === "done") return 100; + if (bytesRead === undefined || totalBytes === undefined || !Number.isFinite(bytesRead) || !Number.isFinite(totalBytes) || bytesRead < 0 || totalBytes <= 0) return null; + return Math.min(99, Math.floor((bytesRead / totalBytes) * 100)); +} + +export function formatSqlFileBytes(bytes: number): string { + const units = ["B", "KB", "MB", "GB", "TB"]; + let value = Math.max(0, bytes); + let unitIndex = 0; + while (value >= 1024 && unitIndex < units.length - 1) { + value /= 1024; + unitIndex += 1; + } + return `${unitIndex === 0 ? value : value.toFixed(1)} ${units[unitIndex]}`; +} diff --git a/crates/dbx-core/src/sql.rs b/crates/dbx-core/src/sql.rs index fd4fd15775..923864d150 100644 --- a/crates/dbx-core/src/sql.rs +++ b/crates/dbx-core/src/sql.rs @@ -298,6 +298,12 @@ pub struct SqlFileProgress { pub elapsed_ms: u128, pub statement_summary: String, pub error: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub bytes_read: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub total_bytes: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub phase: Option, /// When processing multiple files, the 0-based index of the current file. #[serde(skip_serializing_if = "Option::is_none")] pub file_index: Option, @@ -306,6 +312,14 @@ pub struct SqlFileProgress { pub file_name: Option, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase")] +pub enum SqlFilePhase { + Preparing, + Reading, + Executing, +} + pub fn decode_sql_file_bytes(bytes: &[u8]) -> Result { if bytes.starts_with(&[0xEF, 0xBB, 0xBF]) { return std::str::from_utf8(&bytes[3..]).map(|text| text.to_string()).map_err(|_| sql_file_encoding_error()); diff --git a/crates/dbx-core/src/sql_file_import.rs b/crates/dbx-core/src/sql_file_import.rs index f89cdc24ed..3c07b5292f 100644 --- a/crates/dbx-core/src/sql_file_import.rs +++ b/crates/dbx-core/src/sql_file_import.rs @@ -1,5 +1,6 @@ use std::io::Read as StdRead; use std::path::Path; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; @@ -23,8 +24,8 @@ use crate::query::{ }; use crate::sql::{ optimize_sql_file_import_statements, prepare_sql_file_statement, split_sql_batches, statement_summary, - SqlFileImportStatement, SqlFileImportStatementKind, SqlFileProgress, SqlFileRequest, SqlFileStatementAction, - SqlFileStatus, SqlParsingOptions, SqlStatementSplitter, SqlStatementWithControl, + SqlFileImportStatement, SqlFileImportStatementKind, SqlFilePhase, SqlFileProgress, SqlFileRequest, + SqlFileStatementAction, SqlFileStatus, SqlParsingOptions, SqlStatementSplitter, SqlStatementWithControl, }; use crate::types::QueryResult; @@ -89,6 +90,53 @@ const SQL_FILE_STATEMENT_BATCH_SIZE: usize = 256; const SQL_FILE_PREVIEW_ENCODING_SAMPLE_BYTES: usize = 1024 * 1024; const SQL_FILE_PROGRESS_EMIT_INTERVAL: Duration = Duration::from_millis(100); +struct SqlFileReadProgress { + bytes_read: Arc, + total_bytes: Option, +} + +impl SqlFileReadProgress { + async fn new(file_paths: &[&Path]) -> Self { + let mut total_bytes = Some(0u64); + for file_path in file_paths { + let size = tokio::fs::metadata(file_path).await.ok().filter(|metadata| metadata.is_file()); + total_bytes = total_bytes.and_then(|total| total.checked_add(size?.len())); + if total_bytes.is_none() { + break; + } + } + Self { bytes_read: Arc::new(AtomicU64::new(0)), total_bytes } + } + + fn attach(&self, mut progress: SqlFileProgress) -> SqlFileProgress { + progress.bytes_read = Some(self.bytes_read.load(Ordering::Relaxed)); + progress.total_bytes = self.total_bytes; + progress.phase = match progress.status { + SqlFileStatus::Done | SqlFileStatus::Error | SqlFileStatus::Cancelled => None, + SqlFileStatus::Started => Some(SqlFilePhase::Preparing), + SqlFileStatus::Running if progress.file_index.is_some() => Some(SqlFilePhase::Preparing), + _ if progress.statement_summary.is_empty() => Some(SqlFilePhase::Reading), + _ => Some(SqlFilePhase::Executing), + }; + progress + } +} + +struct SqlFileCountingReader { + reader: Reader, + bytes_read: Option>, +} + +impl StdRead for SqlFileCountingReader { + fn read(&mut self, buffer: &mut [u8]) -> std::io::Result { + let read = self.reader.read(buffer)?; + if let Some(bytes_read) = &self.bytes_read { + bytes_read.fetch_add(read as u64, Ordering::Relaxed); + } + Ok(read) + } +} + pub struct SqlFileProgressEmitter Instant> { emit: F, now: C, @@ -468,6 +516,10 @@ pub async fn execute_sql_file_paths( return Err(error); } + let read_progress = SqlFileReadProgress::new(file_paths).await; + let mut emit = |progress| emit(read_progress.attach(progress)); + emit(sql_file_progress(&request.execution_id, SqlFileStatus::Started, 0, 0, 0, 0, started_at, "", None)); + let import_target = sql_file_import_target(state, &request.connection_id, &request.database).await; let options = import_target .as_ref() @@ -541,6 +593,9 @@ pub async fn execute_sql_file_paths( elapsed_ms: started_at.elapsed().as_millis(), statement_summary: String::new(), error: None, + bytes_read: None, + total_bytes: None, + phase: None, file_index: Some(file_index), file_name: Some(file_name.clone()), }); @@ -551,19 +606,24 @@ pub async fn execute_sql_file_paths( let normalize_mysql_binary_literals = import_target.as_ref().is_some_and(|target| { crate::sql::is_mysql_compatible_import_target(&target.db_type, target.driver_profile.as_deref()) }); - let mut decoder = - match SqlFileStreamDecoder::open_for_target(file_path, normalize_mysql_binary_literals).await { - Ok(decoder) => decoder, - Err(error) => { - emit(sql_file_execution_error_progress( - &request.execution_id, - started_at, - &progress, - error.clone(), - )); - return Err(error); - } - }; + let mut decoder = match SqlFileStreamDecoder::open_for_target_with_progress( + file_path, + normalize_mysql_binary_literals, + read_progress.bytes_read.clone(), + ) + .await + { + Ok(decoder) => decoder, + Err(error) => { + emit(sql_file_execution_error_progress( + &request.execution_id, + started_at, + &progress, + error.clone(), + )); + return Err(error); + } + }; loop { let chunk = match decoder.next_chunk().await { @@ -590,6 +650,17 @@ pub async fn execute_sql_file_paths( emit_sql_file_terminal_progress(request, &token, started_at, &progress, &mut emit); return Ok(()); } + emit(sql_file_progress( + &request.execution_id, + SqlFileStatus::Running, + progress.statement_index, + progress.success_count, + progress.failure_count, + progress.affected_rows, + started_at, + "", + None, + )); let mut next_statements = splitter.push_chunk(&chunk); if let Some(filter) = restore_filter.as_mut() { if let Err(error) = filter_restore_statements( @@ -667,6 +738,9 @@ pub async fn execute_sql_file_paths( elapsed_ms: started_at.elapsed().as_millis(), statement_summary: String::new(), error: None, + bytes_read: None, + total_bytes: None, + phase: None, file_index: Some(file_index), file_name: Some(file_name), }); @@ -794,17 +868,24 @@ struct SqlFileStreamDecoder { } enum SqlFileByteReader { - Plain(BufReader), - Gzip(Arc>>>), + Plain(BufReader, Option>), + Gzip(Arc>>>>), } impl SqlFileByteReader { async fn open(file_path: &Path) -> Result { + Self::open_with_progress(file_path, None).await + } + + async fn open_with_progress(file_path: &Path, bytes_read: Option>) -> Result { if is_gzip_sql_file_path(file_path) { let path = file_path.to_path_buf(); let reader = tokio::task::spawn_blocking(move || { let file = std::fs::File::open(&path).map_err(|error| error.to_string())?; - Ok::<_, String>(flate2::read::GzDecoder::new(std::io::BufReader::new(file))) + Ok::<_, String>(flate2::read::GzDecoder::new(SqlFileCountingReader { + reader: std::io::BufReader::new(file), + bytes_read, + })) }) .await .map_err(|error| format!("Failed to open compressed SQL file: {error}"))??; @@ -812,12 +893,18 @@ impl SqlFileByteReader { } let file = tokio::fs::File::open(file_path).await.map_err(|error| error.to_string())?; - Ok(Self::Plain(BufReader::with_capacity(SQL_FILE_READ_CHUNK_BYTES, file))) + Ok(Self::Plain(BufReader::with_capacity(SQL_FILE_READ_CHUNK_BYTES, file), bytes_read)) } async fn read(&mut self, buffer: &mut [u8]) -> Result { match self { - Self::Plain(reader) => reader.read(buffer).await.map_err(|error| error.to_string()), + Self::Plain(reader, bytes_read) => { + let read = reader.read(buffer).await.map_err(|error| error.to_string())?; + if let Some(bytes_read) = bytes_read { + bytes_read.fetch_add(read as u64, Ordering::Relaxed); + } + Ok(read) + } Self::Gzip(reader) => { let reader = reader.clone(); let capacity = buffer.len(); @@ -852,25 +939,34 @@ fn is_gzip_sql_file_path(path: &Path) -> bool { impl SqlFileStreamDecoder { #[cfg(test)] async fn open(file_path: &Path) -> Result { - Self::open_with_options(file_path, None, false).await + Self::open_with_options(file_path, None, false, None).await } async fn open_with_detection_limit(file_path: &Path, detection_limit: Option) -> Result { - Self::open_with_options(file_path, detection_limit, false).await + Self::open_with_options(file_path, detection_limit, false, None).await } async fn open_for_target(file_path: &Path, normalize_mysql_binary_literals: bool) -> Result { - Self::open_with_options(file_path, None, normalize_mysql_binary_literals).await + Self::open_with_options(file_path, None, normalize_mysql_binary_literals, None).await + } + + async fn open_for_target_with_progress( + file_path: &Path, + normalize_mysql_binary_literals: bool, + bytes_read: Arc, + ) -> Result { + Self::open_with_options(file_path, None, normalize_mysql_binary_literals, Some(bytes_read)).await } async fn open_with_options( file_path: &Path, detection_limit: Option, normalize_mysql_binary_literals: bool, + bytes_read: Option>, ) -> Result { let (encoding, bom_len, detected_mysql_binary_literals) = detect_sql_file_encoding(file_path, detection_limit).await?; - let mut reader = SqlFileByteReader::open(file_path).await?; + let mut reader = SqlFileByteReader::open_with_progress(file_path, bytes_read).await?; let mut prefix = [0u8; 3]; let prefix_len = reader.read(&mut prefix).await?; let prefix = &prefix[..prefix_len]; @@ -1474,6 +1570,9 @@ pub fn sql_file_progress( elapsed_ms: started_at.elapsed().as_millis(), statement_summary: statement_summary.to_string(), error, + bytes_read: None, + total_bytes: None, + phase: None, file_index: None, file_name: None, } @@ -2271,6 +2370,9 @@ mod tests { elapsed_ms: statement_index as u128, statement_summary: format!("statement {statement_index}"), error: None, + bytes_read: None, + total_bytes: None, + phase: None, file_index: None, file_name: None, } @@ -2315,6 +2417,143 @@ mod tests { assert_eq!(emitted[emitted.len() - 2].statement_index, 1_000); } + #[tokio::test] + async fn file_progress_counts_source_bytes_without_counting_encoding_detection() { + let sql = "SELECT '中文';\n".repeat(40_000); + let mut utf16 = vec![0xff, 0xfe]; + utf16.extend(sql.encode_utf16().flat_map(u16::to_le_bytes)); + let (gbk, _, _) = encoding_rs::GBK.encode(&sql); + for bytes in [sql.as_bytes().to_vec(), utf16, gbk.into_owned()] { + let path = temporary_sql_file(&bytes).await; + let read_progress = SqlFileReadProgress::new(&[&path]).await; + let mut decoder = + SqlFileStreamDecoder::open_for_target_with_progress(&path, false, read_progress.bytes_read.clone()) + .await + .unwrap(); + assert_eq!(read_progress.bytes_read.load(Ordering::Relaxed), 3); + let mut decoded = String::new(); + let mut previous = 0; + while let Some(chunk) = decoder.next_chunk().await.unwrap() { + decoded.push_str(&chunk); + let current = read_progress.bytes_read.load(Ordering::Relaxed); + assert!(current >= previous && current <= bytes.len() as u64); + previous = current; + } + assert_eq!(decoded, sql); + assert_eq!(read_progress.bytes_read.load(Ordering::Relaxed), bytes.len() as u64); + assert_eq!(read_progress.total_bytes, Some(bytes.len() as u64)); + tokio::fs::remove_file(path).await.unwrap(); + } + } + + #[tokio::test] + async fn file_progress_aggregates_plain_compressed_and_empty_files() { + let sql = "SELECT 1;\n".repeat(80_000); + let paths = [ + temporary_sql_file(sql.as_bytes()).await, + temporary_gzip_sql_file(sql.as_bytes()).await, + temporary_sql_file(b"").await, + ]; + let path_refs = paths.iter().map(|path| path.as_path()).collect::>(); + let read_progress = SqlFileReadProgress::new(&path_refs).await; + let mut expected_bytes = 0; + for path in &paths { + let mut decoder = + SqlFileStreamDecoder::open_for_target_with_progress(path, false, read_progress.bytes_read.clone()) + .await + .unwrap(); + while decoder.next_chunk().await.unwrap().is_some() {} + expected_bytes += tokio::fs::metadata(path).await.unwrap().len(); + assert_eq!(read_progress.bytes_read.load(Ordering::Relaxed), expected_bytes); + } + assert_eq!(read_progress.total_bytes, Some(expected_bytes)); + assert!(expected_bytes < 2 * sql.len() as u64); + for path in paths { + tokio::fs::remove_file(path).await.unwrap(); + } + } + + #[tokio::test] + async fn file_progress_does_not_invent_missing_file_sizes() { + let path = std::env::temp_dir().join(format!("dbx-missing-sql-{}", uuid::Uuid::new_v4())); + assert_eq!(SqlFileReadProgress::new(&[&path]).await.total_bytes, None); + } + + #[test] + fn file_progress_distinguishes_reading_execution_and_terminal_states() { + let read_progress = SqlFileReadProgress { bytes_read: Arc::new(AtomicU64::new(25)), total_bytes: Some(100) }; + let started = read_progress.attach(test_progress(SqlFileStatus::Started, 0)); + assert_eq!(started.phase, Some(SqlFilePhase::Preparing)); + let mut reading = test_progress(SqlFileStatus::Running, 0); + reading.statement_summary.clear(); + assert_eq!(read_progress.attach(reading).phase, Some(SqlFilePhase::Reading)); + assert_eq!(read_progress.attach(test_progress(SqlFileStatus::Running, 1)).phase, Some(SqlFilePhase::Executing)); + for status in [SqlFileStatus::Done, SqlFileStatus::Error, SqlFileStatus::Cancelled] { + let event = read_progress.attach(test_progress(status, 1)); + assert_eq!(event.status, status); + assert_eq!(event.phase, None); + assert_eq!(event.bytes_read, Some(25)); + assert_eq!(event.total_bytes, Some(100)); + } + let old_payload = serde_json::to_value(test_progress(SqlFileStatus::Running, 1)).unwrap(); + assert!(old_payload.get("bytesRead").is_none()); + assert_eq!(serde_json::from_value::(old_payload).unwrap().bytes_read, None); + } + + #[tokio::test] + async fn file_progress_executor_preserves_success_cancellation_and_failure() { + let directory = std::env::temp_dir().join(format!("dbx-file-progress-{}", uuid::Uuid::new_v4())); + tokio::fs::create_dir_all(&directory).await.unwrap(); + let storage = crate::storage::Storage::open(&directory.join("storage.db")).await.unwrap(); + let state = AppState::new(storage); + let comments = "-- no database operation\n".repeat(40_000).into_bytes(); + for (bytes, cancel, terminal) in [ + (Vec::new(), false, SqlFileStatus::Done), + (comments.clone(), false, SqlFileStatus::Done), + (comments, true, SqlFileStatus::Cancelled), + (vec![0xef, 0xbb, 0xbf, 0xff], false, SqlFileStatus::Error), + (b"SELECT 1;".to_vec(), false, SqlFileStatus::Error), + ] { + let path = temporary_sql_file(&bytes).await; + let request = SqlFileRequest { + execution_id: "file-progress".to_string(), + connection_id: "unconfigured".to_string(), + database: String::new(), + file_path: path.to_string_lossy().to_string(), + continue_on_error: false, + selected_tables: None, + part_cooldown_ms: 0, + skip_relational_constraints: false, + }; + let token = CancellationToken::new(); + let mut events = Vec::new(); + let result = execute_sql_file_path(&state, &request, &path, token.clone(), Instant::now(), |event| { + if cancel && event.phase == Some(SqlFilePhase::Reading) { + token.cancel(); + } + events.push(event); + }) + .await; + assert_eq!(result.is_err(), terminal == SqlFileStatus::Error); + assert_eq!(events.last().unwrap().status, terminal); + assert!(events.iter().all(|event| event.total_bytes == Some(bytes.len() as u64))); + assert!(events.windows(2).all(|pair| pair[0].bytes_read <= pair[1].bytes_read)); + assert_eq!( + events + .iter() + .filter(|event| matches!( + event.status, + SqlFileStatus::Done | SqlFileStatus::Error | SqlFileStatus::Cancelled + )) + .count(), + 1 + ); + tokio::fs::remove_file(path).await.unwrap(); + } + drop(state); + tokio::fs::remove_dir_all(directory).await.unwrap(); + } + #[test] fn progress_emitter_sends_key_events_immediately() { let base = Instant::now();