Import: choose source format (SQL/JSON) and batch JSON inserts (#1548)

* feat(import): let Import choose source format (SQL or JSON)

* perf(plugin-json): batch row inserts and report the exact import total

Author: datlechin

CHANGELOG.md

@@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Import a JSON file into a table. A dedicated sheet accepts an array of objects, newline-delimited JSON, or TablePro's own JSON export, and lets you map each field to a column in an existing table or in a new table with inferred, editable columns.
+- Import a JSON file into a table. The Import menu now lets you pick the source, SQL or JSON, instead of going straight to SQL. The JSON flow accepts an array of objects, newline-delimited JSON, or TablePro's own JSON export, and lets you map each field to a column in an existing table or in a new table with inferred, editable columns. Rows insert in batched multi-row statements, so large imports over a remote connection finish in seconds instead of minutes.
 - The window title bar shows the open table's name, with its database and schema below, so you can tell which table you're viewing without checking the sidebar. (#1475)
 - iOS: open DuckDB database files and in-memory DuckDB databases, matching the Mac app. (#1526)
 - Save the current query as a favorite from a star button in the SQL editor toolbar.

Packages/TableProCore/Sources/TableProPluginKit/PluginImportDataSink.swift

@@ -10,6 +10,7 @@ public protocol PluginImportDataSink: AnyObject, Sendable {
     var targetTable: String? { get }
     func execute(statement: String) async throws
     func insertRow(_ values: [String: PluginCellValue]) async throws
+    func insertRows(_ rows: [[String: PluginCellValue]]) async throws
     func deleteAllRowsFromTargetTable() async throws
     func beginTransaction() async throws
     func commitTransaction() async throws
@@ -23,6 +24,11 @@ public extension PluginImportDataSink {
     func insertRow(_ values: [String: PluginCellValue]) async throws {
         throw PluginImportError.importFailed("Row-based import is not supported by this connection")
     }
+    func insertRows(_ rows: [[String: PluginCellValue]]) async throws {
+        for row in rows {
+            try await insertRow(row)
+        }
+    }
     func deleteAllRowsFromTargetTable() async throws {
         throw PluginImportError.importFailed("Clearing the target table is not supported by this connection")
     }

Packages/TableProCore/Sources/TableProPluginKit/PluginImportProgress.swift

@@ -27,6 +27,18 @@ public final class PluginImportProgress: @unchecked Sendable {
         }
     }
 
+    public func incrementStatement(by amount: Int) {
+        guard amount > 0 else { return }
+        lock.lock()
+        let previous = internalCount
+        internalCount += amount
+        let count = internalCount
+        lock.unlock()
+        if count / updateInterval != previous / updateInterval {
+            progress.completedUnitCount = Int64(count)
+        }
+    }
+
     public func setStatus(_ message: String) {
         progress.localizedAdditionalDescription = message
     }

Plugins/JSONImportPlugin/JSONImportPlugin.swift

@@ -43,12 +43,12 @@ final class JSONImportPlugin: ImportFormatPlugin, SettablePlugin {
         let url = source.fileURL()
         let useTransaction = settings.wrapInTransaction && settings.errorHandling != .skipAndContinue
 
-        progress.setEstimatedTotal(max(1, Int(source.fileSizeBytes() / 256)))
-
+        let batchSize = 500
         var inserted = 0
         var skipped = 0
         var errors: [PluginImportResult.ImportStatementError] = []
         let maxErrors = 1_000
+        var batch: [(line: Int, row: [String: PluginCellValue])] = []
 
         do {
             if settings.deleteExistingRows {
@@ -59,25 +59,35 @@ final class JSONImportPlugin: ImportFormatPlugin, SettablePlugin {
             }
 
             if JSONImportParsing.isLineDelimited(url) {
+                progress.setEstimatedTotal(max(1, Int(source.fileSizeBytes() / 256)))
                 var lineNumber = 0
                 for try await line in url.lines {
                     try progress.checkCancellation()
                     lineNumber += 1
                     let trimmed = line.trimmingCharacters(in: .whitespacesAndNewlines)
                     guard !trimmed.isEmpty else { continue }
-                    let row = try JSONImportParsing.parseRow(fromLine: trimmed)
-                    try await insert(row, into: sink, at: lineNumber, progress: progress,
-                                     inserted: &inserted, skipped: &skipped, errors: &errors, maxErrors: maxErrors)
+                    batch.append((lineNumber, try JSONImportParsing.parseRow(fromLine: trimmed)))
+                    if batch.count >= batchSize {
+                        try await flush(&batch, into: sink, progress: progress,
+                                        inserted: &inserted, skipped: &skipped, errors: &errors, maxErrors: maxErrors)
+                    }
                 }
             } else {
                 let rawRows = try JSONImportParsing.parseRows(at: url, targetTable: sink.targetTable)
+                progress.setEstimatedTotal(rawRows.count)
                 for (index, rawRow) in rawRows.enumerated() {
                     try progress.checkCancellation()
-                    try await insert(JSONImportParsing.convertRow(rawRow), into: sink, at: index + 1, progress: progress,
-                                     inserted: &inserted, skipped: &skipped, errors: &errors, maxErrors: maxErrors)
+                    batch.append((index + 1, JSONImportParsing.convertRow(rawRow)))
+                    if batch.count >= batchSize {
+                        try await flush(&batch, into: sink, progress: progress,
+                                        inserted: &inserted, skipped: &skipped, errors: &errors, maxErrors: maxErrors)
+                    }
                 }
             }
 
+            try await flush(&batch, into: sink, progress: progress,
+                            inserted: &inserted, skipped: &skipped, errors: &errors, maxErrors: maxErrors)
+
             if useTransaction {
                 try await sink.commitTransaction()
             }
@@ -131,6 +141,41 @@ final class JSONImportPlugin: ImportFormatPlugin, SettablePlugin {
         }
     }
 
+    private func flush(
+        _ batch: inout [(line: Int, row: [String: PluginCellValue])],
+        into sink: any PluginImportDataSink,
+        progress: PluginImportProgress,
+        inserted: inout Int,
+        skipped: inout Int,
+        errors: inout [PluginImportResult.ImportStatementError],
+        maxErrors: Int
+    ) async throws {
+        guard !batch.isEmpty else { return }
+        let entries = batch
+        batch.removeAll(keepingCapacity: true)
+
+        do {
+            try await sink.insertRows(entries.map(\.row))
+            inserted += entries.count
+            progress.incrementStatement(by: entries.count)
+        } catch {
+            switch settings.errorHandling {
+            case .stopAndRollback, .stopAndCommit:
+                let firstLine = entries.first?.line ?? 0
+                throw PluginImportError.statementFailed(
+                    statement: "rows \(firstLine)-\(entries.last?.line ?? firstLine)",
+                    line: firstLine,
+                    underlyingError: error
+                )
+            case .skipAndContinue:
+                for entry in entries {
+                    try await insert(entry.row, into: sink, at: entry.line, progress: progress,
+                                     inserted: &inserted, skipped: &skipped, errors: &errors, maxErrors: maxErrors)
+                }
+            }
+        }
+    }
+
     // MARK: - Source introspection
 
     func detectSourceFields(at url: URL, targetTable: String?) throws -> [PluginImportField] {

Plugins/TableProPluginKit/PluginImportDataSink.swift

@@ -10,6 +10,7 @@ public protocol PluginImportDataSink: AnyObject, Sendable {
     var targetTable: String? { get }
     func execute(statement: String) async throws
     func insertRow(_ values: [String: PluginCellValue]) async throws
+    func insertRows(_ rows: [[String: PluginCellValue]]) async throws
     func deleteAllRowsFromTargetTable() async throws
     func beginTransaction() async throws
     func commitTransaction() async throws
@@ -23,6 +24,11 @@ public extension PluginImportDataSink {
     func insertRow(_ values: [String: PluginCellValue]) async throws {
         throw PluginImportError.importFailed("Row-based import is not supported by this connection")
     }
+    func insertRows(_ rows: [[String: PluginCellValue]]) async throws {
+        for row in rows {
+            try await insertRow(row)
+        }
+    }
     func deleteAllRowsFromTargetTable() async throws {
         throw PluginImportError.importFailed("Clearing the target table is not supported by this connection")
     }

Plugins/TableProPluginKit/PluginImportProgress.swift

@@ -27,6 +27,18 @@ public final class PluginImportProgress: @unchecked Sendable {
         }
     }
 
+    public func incrementStatement(by amount: Int) {
+        guard amount > 0 else { return }
+        lock.lock()
+        let previous = internalCount
+        internalCount += amount
+        let count = internalCount
+        lock.unlock()
+        if count / updateInterval != previous / updateInterval {
+            progress.completedUnitCount = Int64(count)
+        }
+    }
+
     public func setStatus(_ message: String) {
         progress.localizedAdditionalDescription = message
     }

TablePro/Core/ChangeTracking/SQLStatementGenerator.swift

@@ -196,6 +196,36 @@ struct SQLStatementGenerator {
         return ParameterizedStatement(sql: sql, parameters: bindParameters)
     }
 
+    func insertStatement(columns insertColumns: [String], rows: [[PluginCellValue]])
+        -> ParameterizedStatement?
+    {
+        guard !insertColumns.isEmpty, !rows.isEmpty,
+              rows.allSatisfy({ $0.count == insertColumns.count }) else { return nil }
+
+        var bindParameters: [Any?] = []
+        let columnList = insertColumns.map(quoteIdentifierFn).joined(separator: ", ")
+        let rowTuples = rows.map { values -> String in
+            let placeholders = values.map { value -> String in
+                bindParameters.append(value.asAny)
+                return placeholder(at: bindParameters.count - 1)
+            }.joined(separator: ", ")
+            return "(\(placeholders))"
+        }.joined(separator: ", ")
+
+        let sql =
+            "INSERT INTO \(quoteIdentifierFn(tableName)) (\(columnList)) VALUES \(rowTuples)"
+
+        return ParameterizedStatement(sql: sql, parameters: bindParameters)
+    }
+
+    var maxBindParameters: Int {
+        switch databaseType {
+        case .sqlite: 32_766
+        case .mssql: 2_100
+        default: 65_535
+        }
+    }
+
     func deleteAllRowsStatement() -> String {
         "DELETE FROM \(quoteIdentifierFn(tableName))"
     }

TablePro/Core/Plugins/ImportDataSinkAdapter.swift

@@ -72,6 +72,55 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
         _ = try await driver.executeParameterized(query: statement.sql, parameters: statement.parameters)
     }
 
+    func insertRows(_ rows: [[String: PluginCellValue]]) async throws {
+        guard targetTable != nil else {
+            throw PluginImportError.importFailed("No target table configured for row import")
+        }
+        guard let rowGenerator else {
+            throw PluginImportError.importFailed("Could not resolve SQL dialect for row import")
+        }
+
+        var index = 0
+        while index < rows.count {
+            let (columns, values) = mappedColumnsAndValues(rows[index])
+            guard !columns.isEmpty else {
+                index += 1
+                continue
+            }
+
+            var groupValues: [[PluginCellValue]] = [values]
+            var next = index + 1
+            while next < rows.count {
+                let (nextColumns, nextValues) = mappedColumnsAndValues(rows[next])
+                guard nextColumns == columns else { break }
+                groupValues.append(nextValues)
+                next += 1
+            }
+            index = next
+
+            let chunkSize = max(1, rowGenerator.maxBindParameters / columns.count)
+            var offset = 0
+            while offset < groupValues.count {
+                let end = min(offset + chunkSize, groupValues.count)
+                let chunk = Array(groupValues[offset..<end])
+                if let statement = rowGenerator.insertStatement(columns: columns, rows: chunk) {
+                    _ = try await driver.executeParameterized(query: statement.sql, parameters: statement.parameters)
+                }
+                offset = end
+            }
+        }
+    }
+
+    private func mappedColumnsAndValues(_ values: [String: PluginCellValue]) -> ([String], [PluginCellValue]) {
+        var pairs: [(column: String, value: PluginCellValue)] = []
+        for (field, value) in values {
+            guard let column = columnMapping[field.lowercased()] else { continue }
+            pairs.append((column, value))
+        }
+        pairs.sort { $0.column < $1.column }
+        return (pairs.map(\.column), pairs.map(\.value))
+    }
+
     func deleteAllRowsFromTargetTable() async throws {
         guard targetTable != nil, let rowGenerator else {
             throw PluginImportError.importFailed("No target table configured for row import")

TablePro/Core/Plugins/PluginManager+Registration.swift

@@ -262,6 +262,31 @@ extension PluginManager {
         return Array(importPlugins.values)
     }
 
+    func importPlugins(for databaseType: DatabaseType) -> [any ImportFormatPlugin] {
+        guard supportsImport(for: databaseType) else { return [] }
+        let typeId = databaseType.rawValue
+        return allImportPlugins()
+            .filter { plugin in
+                let supported = type(of: plugin).supportedDatabaseTypeIds
+                let excluded = type(of: plugin).excludedDatabaseTypeIds
+                if !supported.isEmpty && !supported.contains(typeId) { return false }
+                if excluded.contains(typeId) { return false }
+                return true
+            }
+            .sorted { lhs, rhs in
+                let lhsRowBased = type(of: lhs).requiresTargetTable
+                let rhsRowBased = type(of: rhs).requiresTargetTable
+                if lhsRowBased != rhsRowBased { return !lhsRowBased }
+                return type(of: lhs).formatDisplayName < type(of: rhs).formatDisplayName
+            }
+    }
+
+    func importFormatOptions(for databaseType: DatabaseType) -> [ImportFormatOption] {
+        importPlugins(for: databaseType).map {
+            ImportFormatOption(id: type(of: $0).formatId, name: type(of: $0).formatDisplayName)
+        }
+    }
+
     /// Returns a temporary plugin driver for query building (buildBrowseQuery), or nil
     /// if the plugin doesn't implement custom query building (NoSQL hooks).
     func queryBuildingDriver(for databaseType: DatabaseType) -> (any PluginDatabaseDriver)? {

TablePro/Core/Services/Export/ImportRouting.swift

@@ -0,0 +1,30 @@
+//
+//  ImportRouting.swift
+//  TablePro
+//
+
+import Foundation
+
+struct ImportFormatOption: Identifiable, Equatable {
+    let id: String
+    let name: String
+
+    var submenuLabel: String {
+        String(format: String(localized: "From %@\u{2026}"), name)
+    }
+
+    var standaloneLabel: String {
+        String(format: String(localized: "Import %@\u{2026}"), name)
+    }
+}
+
+enum ImportSheetRoute: Equatable {
+    case statement(formatId: String)
+    case rowMapping(formatId: String)
+}
+
+enum ImportRouting {
+    static func route(formatId: String, requiresTargetTable: Bool) -> ImportSheetRoute {
+        requiresTargetTable ? .rowMapping(formatId: formatId) : .statement(formatId: formatId)
+    }
+}