refactor: rewrite export system with streaming and native macOS patterns (#782)

* refactor: rewrite export system with streaming and native macOS patterns

* fix: export data correctness, atomic writes, and view layer cleanup

* feat: native streaming for all built-in database drivers

* fix: address review findings for export streaming rewrite

* feat: native streaming for all registry database drivers

* fix: progress reporting issues and registry driver streaming

* fix: flush final row count before dismissing export progress dialog

* fix: use unbounded buffer for AsyncThrowingStream to prevent row drops

* fix: address all review findings, batch streaming for performance

Author: datlechin

CHANGELOG.md

@@ -11,13 +11,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Main editor window rewritten on AppKit (`NSWindowController` + `NSToolbar`) for faster tab opens and correct lifecycle
 - Toolbar layout follows Apple HIG (sidebar left, connection center, view actions right)
+- Export engine rewritten to use streaming row fetch instead of offset/limit pagination
+- Export progress integrates with macOS system progress
+- All export file writes use atomic operations for crash safety
 
 ### Fixed
 
 - Cmd+W closing the connection window instead of clearing to empty state
 - ER Diagram and Server Dashboard replacing the current tab instead of opening a new one
 - Welcome window stealing focus on connect, disabling Cmd+T until manual click
 - Toolbar empty on second tab, menu shortcuts disabled after toolbar click
+- JSON export no longer coerces leading-zero strings to integers
+- XLSX export auto-splits tables exceeding 1,048,576 rows into multiple sheets
+- CSV formula injection guard corrected to OWASP-standard prefixes only
+- MQL export validates JSON values before passthrough
+- SQL export gzip compression is now async and cancellable
+- Export progress bar reliably reaches 100%
 
 ### Added
 

Packages/TableProCore/Sources/TableProPluginKit/ExportFormatPlugin.swift

@@ -77,9 +77,35 @@ public struct PluginEnumTypeInfo: Sendable {
     }
 }
 
+public typealias PluginRow = [String?]
+
+public struct PluginStreamHeader: Sendable {
+    public let columns: [String]
+    public let columnTypeNames: [String]
+    public let estimatedRowCount: Int?
+
+    public init(columns: [String], columnTypeNames: [String], estimatedRowCount: Int? = nil) {
+        self.columns = columns
+        self.columnTypeNames = columnTypeNames
+        self.estimatedRowCount = estimatedRowCount
+    }
+}
+
+public enum PluginStreamElement: Sendable {
+    case header(PluginStreamHeader)
+    case rows([PluginRow])
+}
+
+public struct ExportFormatResult: Sendable {
+    public let warnings: [String]
+    public init(warnings: [String] = []) {
+        self.warnings = warnings
+    }
+}
+
 public protocol PluginExportDataSource: AnyObject, Sendable {
     var databaseTypeId: String { get }
-    func fetchRows(table: String, databaseName: String, offset: Int, limit: Int) async throws -> PluginQueryResult
+    func streamRows(table: String, databaseName: String) -> AsyncThrowingStream<PluginStreamElement, Error>
     func fetchTableDDL(table: String, databaseName: String) async throws -> String
     func execute(query: String) async throws -> PluginQueryResult
     func quoteIdentifier(_ identifier: String) -> String
@@ -95,100 +121,65 @@ public extension PluginExportDataSource {
 }
 
 public final class PluginExportProgress: @unchecked Sendable {
-    private let lock = NSLock()
-    private var _currentTable: String = ""
-    private var _currentTableIndex: Int = 0
-    private var _processedRows: Int = 0
-    private var _totalRows: Int = 0
-    private var _statusMessage: String = ""
-    private var _isCancelled: Bool = false
-
+    private let progress: Progress
     private let updateInterval: Int = 1_000
     private var internalRowCount: Int = 0
+    private let lock = NSLock()
 
-    public var onUpdate: (@Sendable (String, Int, Int, Int, String) -> Void)?
-
-    public init() {}
+    public init(progress: Progress) {
+        self.progress = progress
+    }
 
     public func setCurrentTable(_ name: String, index: Int) {
-        lock.lock()
-        _currentTable = name
-        _currentTableIndex = index
-        lock.unlock()
-        notifyUpdate()
+        progress.localizedDescription = name
     }
 
     public func incrementRow() {
         lock.lock()
         internalRowCount += 1
-        _processedRows = internalRowCount
-        let shouldNotify = internalRowCount % updateInterval == 0
+        let count = internalRowCount
+        let shouldNotify = count % updateInterval == 0
         lock.unlock()
         if shouldNotify {
-            notifyUpdate()
+            progress.completedUnitCount = Int64(count)
         }
     }
 
     public func finalizeTable() {
-        notifyUpdate()
-    }
-
-    public func setTotalRows(_ count: Int) {
         lock.lock()
-        _totalRows = count
+        let count = internalRowCount
         lock.unlock()
+        progress.completedUnitCount = Int64(count)
     }
 
     public func setStatus(_ message: String) {
-        lock.lock()
-        _statusMessage = message
-        lock.unlock()
-        notifyUpdate()
+        progress.localizedAdditionalDescription = message
     }
 
     public func checkCancellation() throws {
-        lock.lock()
-        let cancelled = _isCancelled
-        lock.unlock()
-        if cancelled || Task.isCancelled {
+        if progress.isCancelled || Task.isCancelled {
             throw PluginExportCancellationError()
         }
     }
 
     public func cancel() {
-        lock.lock()
-        _isCancelled = true
-        lock.unlock()
+        progress.cancel()
     }
 
     public var isCancelled: Bool {
-        lock.lock()
-        defer { lock.unlock() }
-        return _isCancelled
+        progress.isCancelled || Task.isCancelled
     }
 
     public var processedRows: Int {
         lock.lock()
         defer { lock.unlock() }
-        return _processedRows
+        return internalRowCount
     }
 
     public var totalRows: Int {
-        lock.lock()
-        defer { lock.unlock() }
-        return _totalRows
+        Int(progress.totalUnitCount)
     }
 
-    private func notifyUpdate() {
-        lock.lock()
-        let table = _currentTable
-        let index = _currentTableIndex
-        let rows = _processedRows
-        let total = _totalRows
-        let status = _statusMessage
-        lock.unlock()
-        onUpdate?(table, index, rows, total, status)
-    }
 }
 
 public protocol ExportFormatPlugin: TableProPlugin {
@@ -204,14 +195,13 @@ public protocol ExportFormatPlugin: TableProPlugin {
     func isTableExportable(optionValues: [Bool]) -> Bool
 
     var currentFileExtension: String { get }
-    var warnings: [String] { get }
 
     func export(
         tables: [PluginExportTable],
         dataSource: any PluginExportDataSource,
         destination: URL,
         progress: PluginExportProgress
-    ) async throws
+    ) async throws -> ExportFormatResult
 }
 
 public extension ExportFormatPlugin {
@@ -222,5 +212,4 @@ public extension ExportFormatPlugin {
     func defaultTableOptionValues() -> [Bool] { [] }
     func isTableExportable(optionValues: [Bool]) -> Bool { true }
     var currentFileExtension: String { Self.defaultFileExtension }
-    var warnings: [String] { [] }
 }

Packages/TableProCore/Sources/TableProPluginKit/ImportFormatPlugin.swift

@@ -66,87 +66,63 @@ public extension PluginImportDataSink {
 }
 
 public final class PluginImportProgress: @unchecked Sendable {
-    private let lock = NSLock()
-    private var _processedStatements: Int = 0
-    private var _estimatedTotalStatements: Int = 0
-    private var _statusMessage: String = ""
-    private var _isCancelled: Bool = false
-
+    private let progress: Progress
     private let updateInterval: Int = 500
     private var internalCount: Int = 0
+    private let lock = NSLock()
 
-    public var onUpdate: (@Sendable (Int, Int, String) -> Void)?
-
-    public init() {}
+    public init(progress: Progress) {
+        self.progress = progress
+    }
 
     public func setEstimatedTotal(_ count: Int) {
-        lock.lock()
-        _estimatedTotalStatements = count
-        lock.unlock()
+        progress.totalUnitCount = Int64(count)
     }
 
     public func incrementStatement() {
         lock.lock()
         internalCount += 1
-        _processedStatements = internalCount
-        let shouldNotify = internalCount % updateInterval == 0
+        let count = internalCount
+        let shouldNotify = count % updateInterval == 0
         lock.unlock()
         if shouldNotify {
-            notifyUpdate()
+            progress.completedUnitCount = Int64(count)
         }
     }
 
     public func setStatus(_ message: String) {
-        lock.lock()
-        _statusMessage = message
-        lock.unlock()
-        notifyUpdate()
+        progress.localizedAdditionalDescription = message
     }
 
     public func checkCancellation() throws {
-        lock.lock()
-        let cancelled = _isCancelled
-        lock.unlock()
-        if cancelled || Task.isCancelled {
+        if progress.isCancelled || Task.isCancelled {
             throw PluginImportCancellationError()
         }
     }
 
     public func cancel() {
-        lock.lock()
-        _isCancelled = true
-        lock.unlock()
+        progress.cancel()
     }
 
     public var isCancelled: Bool {
-        lock.lock()
-        defer { lock.unlock() }
-        return _isCancelled
+        progress.isCancelled || Task.isCancelled
     }
 
     public var processedStatements: Int {
         lock.lock()
         defer { lock.unlock() }
-        return _processedStatements
+        return internalCount
     }
 
     public var estimatedTotalStatements: Int {
-        lock.lock()
-        defer { lock.unlock() }
-        return _estimatedTotalStatements
+        Int(progress.totalUnitCount)
     }
 
     public func finalize() {
-        notifyUpdate()
-    }
-
-    private func notifyUpdate() {
         lock.lock()
-        let processed = _processedStatements
-        let total = _estimatedTotalStatements
-        let status = _statusMessage
+        let count = internalCount
         lock.unlock()
-        onUpdate?(processed, total, status)
+        progress.completedUnitCount = Int64(count)
     }
 }
 

Packages/TableProCore/Sources/TableProPluginKit/PluginDatabaseDriver.swift

@@ -170,6 +170,9 @@ public protocol PluginDatabaseDriver: AnyObject, Sendable {
 
     // Default export query (optional)
     func defaultExportQuery(table: String) -> String?
+
+    // Streaming row fetch for export
+    func streamRows(query: String) -> AsyncThrowingStream<PluginStreamElement, Error>
 }
 
 // MARK: - Default Implementations
@@ -530,4 +533,41 @@ public extension PluginDatabaseDriver {
     func fetchRows(query: String, offset: Int, limit: Int) async throws -> PluginQueryResult {
         try await execute(query: "\(query) LIMIT \(limit) OFFSET \(offset)")
     }
+
+    func streamRows(query: String) -> AsyncThrowingStream<PluginStreamElement, Error> {
+        AsyncThrowingStream(bufferingPolicy: .unbounded) { continuation in
+            Task {
+                do {
+                    let batchSize = 1_000
+                    let firstPage = try await fetchRows(query: query, offset: 0, limit: batchSize)
+                    continuation.yield(.header(PluginStreamHeader(
+                        columns: firstPage.columns,
+                        columnTypeNames: firstPage.columnTypeNames,
+                        estimatedRowCount: nil
+                    )))
+                    if !firstPage.rows.isEmpty {
+                        continuation.yield(.rows(firstPage.rows))
+                    }
+                    if firstPage.rows.count < batchSize {
+                        continuation.finish()
+                        return
+                    }
+                    await Task.yield()
+                    var offset = firstPage.rows.count
+                    while true {
+                        try Task.checkCancellation()
+                        let page = try await fetchRows(query: query, offset: offset, limit: batchSize)
+                        if page.rows.isEmpty { break }
+                        continuation.yield(.rows(page.rows))
+                        offset += page.rows.count
+                        if page.rows.count < batchSize { break }
+                        await Task.yield()
+                    }
+                    continuation.finish()
+                } catch {
+                    continuation.finish(throwing: error)
+                }
+            }
+        }
+    }
 }