diff --git a/apple/Plugins/CSVImportPlugin/CSVImportPlugin.swift b/apple/Plugins/CSVImportPlugin/CSVImportPlugin.swift
index 817d03e083..b52fcf2174 100644
--- a/apple/Plugins/CSVImportPlugin/CSVImportPlugin.swift
+++ b/apple/Plugins/CSVImportPlugin/CSVImportPlugin.swift
@@ -29,6 +29,16 @@ final class CSVImportPlugin: ObservableObject, ImportFormatPlugin, SettablePlugi
 
     var fieldDetectionSignature: String { settings.detectionSignature }
 
+    var rowImportConfiguration: RowImportRunner.Configuration? { runConfiguration }
+
+    private var runConfiguration: RowImportRunner.Configuration {
+        RowImportRunner.Configuration(
+            errorHandling: settings.errorHandling,
+            wrapInTransaction: settings.wrapInTransaction,
+            deleteExistingRows: settings.deleteExistingRows
+        )
+    }
+
     required init() { loadSettings() }
 
     @MainActor
@@ -82,11 +92,7 @@ final class CSVImportPlugin: ObservableObject, ImportFormatPlugin, SettablePlugi
         let lineOffset = hasHeader ? 2 : 1
         var cursor = 0
         let outcome = try await RowImportRunner.run(
-            configuration: RowImportRunner.Configuration(
-                errorHandling: settings.errorHandling,
-                wrapInTransaction: settings.wrapInTransaction,
-                deleteExistingRows: settings.deleteExistingRows
-            ),
+            configuration: runConfiguration,
             sink: sink,
             progress: progress
         ) {
diff --git a/apple/Plugins/FormatPluginTests/RowImportConfigurationTests.swift b/apple/Plugins/FormatPluginTests/RowImportConfigurationTests.swift
new file mode 100644
index 0000000000..a6024b9e38
--- /dev/null
+++ b/apple/Plugins/FormatPluginTests/RowImportConfigurationTests.swift
@@ -0,0 +1,54 @@
+//
+//  RowImportConfigurationTests.swift
+//  TableProTests
+//
+
+import Foundation
+import TableProPluginKit
+import Testing
+
+/// A run of several files has one transaction, so it reads each format's error handling and transaction options
+/// before it starts, and they have to be the ones the format's own import runs with.
+struct RowImportConfigurationTests {
+    @Test("CSV reports the options its import runs with")
+    func csvReportsItsOptions() throws {
+        let plugin = CSVImportPlugin()
+        let saved = plugin.settings
+        defer { plugin.settings = saved }
+        plugin.settings.errorHandling = .stopAndCommit
+        plugin.settings.wrapInTransaction = false
+        plugin.settings.deleteExistingRows = true
+
+        let configuration = try #require(plugin.rowImportConfiguration)
+
+        #expect(configuration.errorHandling == .stopAndCommit)
+        #expect(configuration.wrapInTransaction == false)
+        #expect(configuration.deleteExistingRows == true)
+    }
+
+    @Test("JSON reports the options its import runs with")
+    func jsonReportsItsOptions() throws {
+        let plugin = JSONImportPlugin()
+        let saved = plugin.settings
+        defer { plugin.settings = saved }
+        plugin.settings.errorHandling = .skipAndContinue
+
+        let configuration = try #require(plugin.rowImportConfiguration)
+
+        #expect(configuration.errorHandling == .skipAndContinue)
+        #expect(configuration.wrapInTransaction == saved.wrapInTransaction)
+    }
+
+    @Test("XLSX reports the options its import runs with")
+    func xlsxReportsItsOptions() throws {
+        let plugin = XLSXImportPlugin()
+        let saved = plugin.settings
+        defer { plugin.settings = saved }
+        plugin.settings.deleteExistingRows = true
+
+        let configuration = try #require(plugin.rowImportConfiguration)
+
+        #expect(configuration.deleteExistingRows == true)
+        #expect(configuration.errorHandling == saved.errorHandling)
+    }
+}
diff --git a/apple/Plugins/JSONImportPlugin/JSONImportPlugin.swift b/apple/Plugins/JSONImportPlugin/JSONImportPlugin.swift
index 862385b769..a765985859 100644
--- a/apple/Plugins/JSONImportPlugin/JSONImportPlugin.swift
+++ b/apple/Plugins/JSONImportPlugin/JSONImportPlugin.swift
@@ -27,6 +27,16 @@ final class JSONImportPlugin: ObservableObject, ImportFormatPlugin, SettablePlug
 
     required init() { loadSettings() }
 
+    var rowImportConfiguration: RowImportRunner.Configuration? { runConfiguration }
+
+    private var runConfiguration: RowImportRunner.Configuration {
+        RowImportRunner.Configuration(
+            errorHandling: settings.errorHandling,
+            wrapInTransaction: settings.wrapInTransaction,
+            deleteExistingRows: settings.deleteExistingRows
+        )
+    }
+
     @MainActor
     func settingsView() -> AnyView? {
         AnyView(JSONImportOptionsView(plugin: self))
@@ -51,11 +61,7 @@ final class JSONImportPlugin: ObservableObject, ImportFormatPlugin, SettablePlug
     ) async throws -> PluginImportResult {
         let startTime = Date()
         let url = source.fileURL()
-        let configuration = RowImportRunner.Configuration(
-            errorHandling: settings.errorHandling,
-            wrapInTransaction: settings.wrapInTransaction,
-            deleteExistingRows: settings.deleteExistingRows
-        )
+        let configuration = runConfiguration
 
         let outcome: RowImportRunner.Outcome
         var unreadableLines: [PluginImportResult.ImportStatementError] = []
diff --git a/apple/Plugins/TableProPluginKit/ImportFormatPlugin.swift b/apple/Plugins/TableProPluginKit/ImportFormatPlugin.swift
index 7d7ebe9cbb..da8c861b35 100644
--- a/apple/Plugins/TableProPluginKit/ImportFormatPlugin.swift
+++ b/apple/Plugins/TableProPluginKit/ImportFormatPlugin.swift
@@ -18,6 +18,10 @@ public protocol ImportFormatPlugin: TableProPlugin, Sendable {
 
     var fieldDetectionSignature: String { get }
 
+    /// The error handling and transaction options the next import runs with. A host that runs several files in one
+    /// transaction reads them first, because the files have to agree. Nil for a format that runs statements.
+    var rowImportConfiguration: RowImportRunner.Configuration? { get }
+
     func performImport(
         source: any PluginImportSource,
         sink: any PluginImportDataSink,
@@ -36,5 +40,7 @@ public extension ImportFormatPlugin {
 
     var fieldDetectionSignature: String { "" }
 
+    var rowImportConfiguration: RowImportRunner.Configuration? { nil }
+
     func detectSourceFields(at url: URL, targetTable: String?) throws -> [PluginImportField] { [] }
 }
diff --git a/apple/Plugins/XLSXImportPlugin/XLSXImportPlugin.swift b/apple/Plugins/XLSXImportPlugin/XLSXImportPlugin.swift
index 510182d0ac..b9cf0455fe 100644
--- a/apple/Plugins/XLSXImportPlugin/XLSXImportPlugin.swift
+++ b/apple/Plugins/XLSXImportPlugin/XLSXImportPlugin.swift
@@ -40,6 +40,16 @@ final class XLSXImportPlugin: ObservableObject, ImportFormatPlugin, SettablePlug
 
     var fieldDetectionSignature: String { settings.detectionSignature }
 
+    var rowImportConfiguration: RowImportRunner.Configuration? { runConfiguration }
+
+    private var runConfiguration: RowImportRunner.Configuration {
+        RowImportRunner.Configuration(
+            errorHandling: settings.errorHandling,
+            wrapInTransaction: settings.wrapInTransaction,
+            deleteExistingRows: settings.deleteExistingRows
+        )
+    }
+
     @MainActor
     func settingsView() -> AnyView? {
         AnyView(XLSXImportOptionsView(plugin: self))
@@ -78,11 +88,7 @@ final class XLSXImportPlugin: ObservableObject, ImportFormatPlugin, SettablePlug
         progress.setEstimatedTotal(max(1, sheet.rows.count))
         var cursor = 0
         let outcome = try await RowImportRunner.run(
-            configuration: RowImportRunner.Configuration(
-                errorHandling: settings.errorHandling,
-                wrapInTransaction: settings.wrapInTransaction,
-                deleteExistingRows: settings.deleteExistingRows
-            ),
+            configuration: runConfiguration,
             sink: sink,
             progress: progress
         ) {
diff --git a/apple/TablePro/Core/Plugins/ImportDataSinkAdapter.swift b/apple/TablePro/Core/Plugins/ImportDataSinkAdapter.swift
index 4134d7c08a..9651ef5ad8 100644
--- a/apple/TablePro/Core/Plugins/ImportDataSinkAdapter.swift
+++ b/apple/TablePro/Core/Plugins/ImportDataSinkAdapter.swift
@@ -49,6 +49,10 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
     /// the user's own: a `BEGIN` there would join it, and the import's `COMMIT` would commit work the
     /// user never asked to commit, so the import then runs inside it and leaves ending it to them.
     private let ownsTransaction = OSAllocatedUnfairLock(initialState: false)
+    /// One file of several imported in one run. The run opened the transaction and ends it, so this sink sends no
+    /// `BEGIN`, `COMMIT` or `ROLLBACK`: reading the session would not stop it, since `.unknown` permits a `BEGIN`
+    /// and MySQL commits the open transaction on one.
+    private let joinsTransaction: Bool
     private let transactionRecord = OSAllocatedUnfairLock(initialState: TransactionRecord())
 
     private struct TransactionRecord {
@@ -78,9 +82,11 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
         targetTable: String? = nil,
         columnMapping: [String: String] = [:],
         sourceFields: Set<String> = [],
+        joinsTransaction: Bool = false,
         isCancelled: @escaping @Sendable () -> Bool = { false }
     ) {
         self.isCancelled = isCancelled
+        self.joinsTransaction = joinsTransaction
         self.driver = driver
         self.databaseType = databaseType
         self.grammar = databaseType.lexicalGrammar
@@ -310,6 +316,7 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
     }
 
     func beginTransaction() async throws {
+        guard !joinsTransaction else { return }
         let owner = WriteTransactionOwner.resolve(
             supportsTransactions: driver.supportsTransactions,
             sessionState: await driver.heldSessionTransactionState()
diff --git a/apple/TablePro/Core/Services/Export/ImportBatchFiles.swift b/apple/TablePro/Core/Services/Export/ImportBatchFiles.swift
new file mode 100644
index 0000000000..fb69c23ea6
--- /dev/null
+++ b/apple/TablePro/Core/Services/Export/ImportBatchFiles.swift
@@ -0,0 +1,62 @@
+//
+//  ImportBatchFiles.swift
+//  TablePro
+//
+
+import Foundation
+
+/// The files a choice of several files or folders imports as rows, each with the format its extension names.
+///
+/// A folder gives the visible files directly inside it, not those of its subfolders, in Finder's order. A file no
+/// row format reads, a SQL dump among them, is left out by name rather than dropped silently: a dump runs its own
+/// statements and is imported on its own.
+internal enum ImportBatchFiles {
+    internal struct File: Equatable, Sendable {
+        internal let url: URL
+        internal let formatId: String
+    }
+
+    internal struct Resolution: Equatable, Sendable {
+        internal var files: [File] = []
+        internal var leftOut: [URL] = []
+    }
+
+    internal static func resolve(
+        _ urls: [URL],
+        among options: [ImportFormatOption],
+        readsRows: (String) -> Bool
+    ) -> Resolution {
+        var resolution = Resolution()
+        var seen = Set<String>()
+        for url in urls {
+            let candidates = isDirectory(url) ? contents(of: url) : [url]
+            for candidate in candidates {
+                let key = candidate.standardizedFileURL.path(percentEncoded: false)
+                guard seen.insert(key).inserted else { continue }
+                if case .format(let formatId) = ImportFileFormatResolver.match(candidate, among: options),
+                   readsRows(formatId) {
+                    resolution.files.append(File(url: candidate, formatId: formatId))
+                } else {
+                    resolution.leftOut.append(candidate)
+                }
+            }
+        }
+        return resolution
+    }
+
+    internal static func isDirectory(_ url: URL) -> Bool {
+        (try? url.resourceValues(forKeys: [.isDirectoryKey]))?.isDirectory == true
+    }
+
+    private static func contents(of folder: URL) -> [URL] {
+        let keys: [URLResourceKey] = [.isRegularFileKey]
+        guard let entries = try? FileManager.default.contentsOfDirectory(
+            at: folder,
+            includingPropertiesForKeys: keys,
+            options: [.skipsHiddenFiles]
+        ) else { return [] }
+        return entries
+            .filter { (try? $0.resourceValues(forKeys: Set(keys)))?.isRegularFile == true }
+            .sorted { $0.lastPathComponent.localizedStandardCompare($1.lastPathComponent) == .orderedAscending }
+    }
+}
diff --git a/apple/TablePro/Core/Services/Export/ImportService.swift b/apple/TablePro/Core/Services/Export/ImportService.swift
index 501436f233..2403f95dac 100644
--- a/apple/TablePro/Core/Services/Export/ImportService.swift
+++ b/apple/TablePro/Core/Services/Export/ImportService.swift
@@ -19,6 +19,9 @@ struct ImportState {
     var statusMessage: String = ""
     var errorMessage: String?
     var transactionOutcome: ImportTransactionOutcome = .untracked
+    var fileCount: Int = 0
+    var fileIndex: Int = 0
+    var fileName: String = ""
 }
 
 // MARK: - Import Service
@@ -33,6 +36,9 @@ final class ImportService: ObservableObject {
     private let historyRecorder: QueryHistoryRecording
     private let gate: ExecutionGate
     @Published private var currentProgress: PluginImportProgress?
+    /// A Stop pressed before a file's progress exists, while a table is created or between two files, which
+    /// cancelling the progress alone would drop. A service runs one import, so it is never reset.
+    private var isStopRequested = false
 
     init(
         connection: DatabaseConnection,
@@ -47,6 +53,7 @@ final class ImportService: ObservableObject {
     // MARK: - Cancellation
 
     func cancelImport() {
+        isStopRequested = true
         currentProgress?.cancel()
     }
 
@@ -95,6 +102,7 @@ final class ImportService: ObservableObject {
             )
         }
         defer { source.cleanup() }
+        guard !isStopRequested else { throw PluginImportCancellationError() }
 
         let historyName = sourceName ?? url.lastPathComponent
         /// After the source exists, so a refusal still removes the decompressed copy it was handed.
@@ -106,29 +114,8 @@ final class ImportService: ObservableObject {
             state.estimatedTotalStatements = Int(initialTotal)
         }
         currentProgress = progress
-
-        let observation = nsProgress.observe(\.completedUnitCount) { [weak self] observed, _ in
-            Task { @MainActor [weak self] in
-                guard let self else { return }
-                let processed = Int(observed.completedUnitCount)
-                let total = Int(observed.totalUnitCount)
-                self.state.processedStatements = processed
-                self.state.estimatedTotalStatements = total
-                if total > 0 {
-                    self.state.progress = min(1.0, Double(processed) / Double(total))
-                }
-            }
-        }
-        defer { observation.invalidate() }
-
-        let statusObservation = nsProgress.observe(\.localizedAdditionalDescription) { [weak self] observed, _ in
-            let status = observed.localizedAdditionalDescription ?? ""
-            Task { @MainActor [weak self] in
-                guard let self, !status.isEmpty else { return }
-                self.state.statusMessage = status
-            }
-        }
-        defer { statusObservation.invalidate() }
+        let observations = track(nsProgress)
+        defer { observations.forEach { $0.invalidate() } }
 
         let result: PluginImportResult
         let startedAt = Date()
@@ -214,6 +201,276 @@ final class ImportService: ObservableObject {
         return result
     }
 
+    private func track(_ nsProgress: Progress) -> [NSKeyValueObservation] {
+        let counts = nsProgress.observe(\.completedUnitCount) { [weak self] observed, _ in
+            Task { @MainActor [weak self] in
+                guard let self else { return }
+                let processed = Int(observed.completedUnitCount)
+                let total = Int(observed.totalUnitCount)
+                self.state.processedStatements = processed
+                self.state.estimatedTotalStatements = total
+                if total > 0 {
+                    self.state.progress = min(1.0, Double(processed) / Double(total))
+                }
+            }
+        }
+        let status = nsProgress.observe(\.localizedAdditionalDescription) { [weak self] observed, _ in
+            let status = observed.localizedAdditionalDescription ?? ""
+            Task { @MainActor [weak self] in
+                guard let self, !status.isEmpty else { return }
+                self.state.statusMessage = status
+            }
+        }
+        return [counts, status]
+    }
+
+    // MARK: - Several files
+
+    /// Imports each file into its own table in one lease, one session scope and, unless the policy turns it off,
+    /// one transaction, so a failure in any file takes back every file under Stop and Rollback. A failure or a Stop
+    /// is in the report rather than thrown, because the report says what became of the files before it.
+    func importFiles(
+        _ files: [RowImportRunItem],
+        scope: DatabaseScope,
+        policy: RowImportBatchPolicy,
+        authorization: OperationReceipt? = nil
+    ) async throws -> RowImportBatchReport {
+        let plugins: [any ImportFormatPlugin] = try files.map { file in
+            guard let plugin = PluginManager.shared.importPlugin(forFormat: file.formatId) else {
+                throw PluginImportError.importFailed("Import format '\(file.formatId)' not found")
+            }
+            return plugin
+        }
+
+        let route = DatabaseManager.shared.executionRoute(for: scope)
+        state = ImportState(isImporting: true)
+        state.fileCount = files.count
+        defer {
+            state.isImporting = false
+            currentProgress = nil
+        }
+        guard !isStopRequested else {
+            return RowImportBatchReport(
+                files: files,
+                outcomes: files.map { _ in .notRun },
+                ending: .stopped(file: 0),
+                transaction: .untracked,
+                executionTime: 0
+            )
+        }
+
+        try await authorizeRun(named: Self.batchName(fileCount: files.count), given: authorization)
+        let operationStart = ContinuousClock.Instant.now
+        let report = try await DatabaseManager.shared.withScopedDriver(
+            scope: scope,
+            route: route,
+            workload: .bulk,
+            cancellation: .protectedWrite
+        ) { driver in
+            try await self.runFiles(files, plugins: plugins, driver: driver, policy: policy)
+        }
+        state.transactionOutcome = report.transaction
+
+        await recordHistory(of: report, in: scope)
+        switch report.ending {
+        case .completed:
+            reportImportFinished(
+                .succeeded(OperationSummary(statementCount: report.importedRowCount)),
+                in: scope,
+                startedAt: operationStart
+            )
+        case .failed:
+            let reason = report.failure?.localizedDescription ?? String(localized: "Unknown error")
+            reportImportFinished(.failed(reason: reason), in: scope, startedAt: operationStart)
+        case .stopped:
+            break
+        }
+        return report
+    }
+
+    internal static func batchName(fileCount: Int) -> String {
+        fileCount == 1
+            ? String(localized: "1 file")
+            : String(format: String(localized: "%lld files"), Int64(fileCount))
+    }
+
+    @MainActor
+    private func runFiles(
+        _ files: [RowImportRunItem],
+        plugins: [any ImportFormatPlugin],
+        driver: DatabaseDriver,
+        policy: RowImportBatchPolicy
+    ) async throws -> RowImportBatchReport {
+        let started = Date()
+        let owner = ImportDataSinkAdapter(driver: driver, databaseType: connection.type)
+        let sessionScope = await openPreservingScope(on: driver)
+        if policy.usesTransaction {
+            do {
+                try await owner.beginTransaction()
+            } catch {
+                await closePreservingScope(sessionScope, on: driver)
+                throw error
+            }
+        }
+
+        var outcomes = [RowImportBatchReport.FileOutcome](repeating: .notRun, count: files.count)
+        var ending = RowImportBatchReport.Ending.completed
+        for (index, file) in files.enumerated() {
+            guard !isStopRequested, !Task.isCancelled else {
+                outcomes[index] = .stopped(rowsBefore: 0)
+                ending = .stopped(file: index)
+                break
+            }
+            switch await runFile(file, plugin: plugins[index], index: index, driver: driver, policy: policy) {
+            case .imported(let result):
+                outcomes[index] = .imported(result)
+            case .stopped(let rows):
+                outcomes[index] = .stopped(rowsBefore: rows)
+                ending = .stopped(file: index)
+            case .failed(let error, let rows):
+                outcomes[index] = .failed(error, rowsBefore: rows)
+                if policy.errorHandling != .skipAndContinue {
+                    ending = .failed(file: index)
+                }
+            case .notRun:
+                break
+            }
+            guard ending == .completed else { break }
+        }
+
+        if policy.usesTransaction {
+            do {
+                try await endTransaction(owner, after: ending, policy: policy)
+            } catch {
+                await closePreservingScope(sessionScope, on: driver)
+                throw error
+            }
+        }
+        await closePreservingScope(sessionScope, on: driver)
+        return RowImportBatchReport(
+            files: files,
+            outcomes: outcomes,
+            ending: ending,
+            transaction: owner.transactionOutcome,
+            executionTime: Date().timeIntervalSince(started)
+        )
+    }
+
+    @MainActor
+    private func runFile(
+        _ file: RowImportRunItem,
+        plugin: any ImportFormatPlugin,
+        index: Int,
+        driver: DatabaseDriver,
+        policy: RowImportBatchPolicy
+    ) async -> RowImportBatchReport.FileOutcome {
+        state.fileIndex = index
+        state.fileName = file.sourceName
+        state.processedStatements = 0
+        state.estimatedTotalStatements = 0
+        state.progress = 0
+        state.statusMessage = ""
+
+        let nsProgress = Progress(totalUnitCount: 0)
+        let progress = PluginImportProgress(progress: nsProgress)
+        currentProgress = progress
+        let observations = track(nsProgress)
+        defer { observations.forEach { $0.invalidate() } }
+
+        let source = PlainFileImportSource(url: file.url)
+        defer { source.cleanup() }
+        let sink = ImportDataSinkAdapter(
+            driver: driver,
+            databaseType: connection.type,
+            targetTable: file.targetTable,
+            columnMapping: file.columnMapping,
+            sourceFields: file.sourceFields,
+            joinsTransaction: policy.usesTransaction,
+            isCancelled: { progress.isCancelled }
+        )
+        do {
+            return .imported(try await plugin.performImport(source: source, sink: sink, progress: progress))
+        } catch is PluginImportCancellationError {
+            return .stopped(rowsBefore: progress.processedStatements)
+        } catch is CancellationError {
+            return .stopped(rowsBefore: progress.processedStatements)
+        } catch {
+            return .failed(error, rowsBefore: progress.processedStatements)
+        }
+    }
+
+    /// Stop and Commit keeps what ran before the failure, as one file's import does. A failure under Stop and
+    /// Rollback, and a Stop under any policy, take back every file.
+    @MainActor
+    private func endTransaction(
+        _ owner: ImportDataSinkAdapter,
+        after ending: RowImportBatchReport.Ending,
+        policy: RowImportBatchPolicy
+    ) async throws {
+        switch ending {
+        case .completed:
+            do {
+                try await owner.commitTransaction()
+            } catch {
+                try? await owner.rollbackTransaction()
+                throw error
+            }
+        case .failed where policy.errorHandling == .stopAndCommit:
+            do {
+                try await owner.commitTransaction()
+            } catch {
+                Self.logger.warning("Commit of a partial import failed: \(error.publicLogShape, privacy: .public)")
+            }
+        case .failed, .stopped:
+            do {
+                try await owner.rollbackTransaction()
+            } catch {
+                Self.logger.warning("Rollback of a failed import failed: \(error.publicLogShape, privacy: .public)")
+            }
+        }
+    }
+
+    /// A file's row is recorded only while its rows are still there, so a rolled-back run records the failure alone.
+    private func recordHistory(of report: RowImportBatchReport, in scope: DatabaseScope) async {
+        for (file, outcome) in zip(report.files, report.outcomes) {
+            switch outcome {
+            case .imported(let result) where report.rowsKept:
+                await recordImport(of: file.sourceName, statements: result.executedStatements,
+                                   executionTime: result.executionTime, error: nil, in: scope)
+            case .failed(let error, let rows):
+                await recordImport(of: file.sourceName, statements: rows,
+                                   executionTime: 0, error: error, in: scope)
+            case .imported, .stopped, .notRun:
+                break
+            }
+        }
+    }
+
+    private func recordImport(
+        of fileName: String,
+        statements: Int,
+        executionTime: TimeInterval,
+        error: (any Error)?,
+        in scope: DatabaseScope
+    ) async {
+        let query = error == nil
+            ? "-- Import from \(fileName) (\(statements) statements)"
+            : "-- Import from \(fileName) (\(statements) statements before failure)"
+        await historyRecorder.record(
+            QueryHistoryRecordRequest(
+                query: query,
+                connectionId: connection.id,
+                databaseName: scope.database,
+                databaseType: connection.type,
+                source: .dataImport,
+                executionTime: executionTime,
+                rowCount: -1,
+                wasSuccessful: error == nil,
+                errorMessage: error?.localizedDescription
+            )
+        )
+    }
+
     /// Before the lease, because a confirmation or a Touch ID prompt answered while it is held keeps
     /// every other tab on this connection waiting on the person.
     private func authorizeRun(named fileName: String, given authorization: OperationReceipt?) async throws {
diff --git a/apple/TablePro/Core/Services/Export/RowImportBatchRun.swift b/apple/TablePro/Core/Services/Export/RowImportBatchRun.swift
new file mode 100644
index 0000000000..2df72439da
--- /dev/null
+++ b/apple/TablePro/Core/Services/Export/RowImportBatchRun.swift
@@ -0,0 +1,91 @@
+//
+//  RowImportBatchRun.swift
+//  TablePro
+//
+
+import Foundation
+import TableProPluginKit
+
+/// One file of a run that imports several, with the table and mapping it was given.
+internal struct RowImportRunItem: Sendable, Equatable {
+    internal let url: URL
+    internal let sourceName: String
+    internal let formatId: String
+    internal let targetTable: String
+    internal let columnMapping: [String: String]
+    internal let sourceFields: Set<String>
+}
+
+/// What a run of several files does when one of them fails. Every file shares it, because they share one
+/// transaction.
+internal struct RowImportBatchPolicy: Sendable, Equatable {
+    internal let errorHandling: ImportErrorHandling
+    internal let wrapInTransaction: Bool
+
+    internal init(errorHandling: ImportErrorHandling, wrapInTransaction: Bool) {
+        self.errorHandling = errorHandling
+        self.wrapInTransaction = wrapInTransaction
+    }
+
+    internal init(_ configuration: RowImportRunner.Configuration) {
+        self.init(errorHandling: configuration.errorHandling, wrapInTransaction: configuration.wrapInTransaction)
+    }
+
+    /// Skip and Continue keeps every row that went in, so it opens no transaction, as one file's import does not.
+    internal var usesTransaction: Bool {
+        wrapInTransaction && errorHandling != .skipAndContinue
+    }
+}
+
+internal struct RowImportBatchReport {
+    internal enum FileOutcome {
+        case imported(PluginImportResult)
+        case failed(any Error, rowsBefore: Int)
+        case stopped(rowsBefore: Int)
+        case notRun
+    }
+
+    internal enum Ending: Equatable {
+        case completed
+        case failed(file: Int)
+        case stopped(file: Int)
+    }
+
+    internal let files: [RowImportRunItem]
+    internal let outcomes: [FileOutcome]
+    internal let ending: Ending
+    internal let transaction: ImportTransactionOutcome
+    internal let executionTime: TimeInterval
+
+    /// Whether the rows the run wrote are still there. A rollback takes back every file, the ones that finished
+    /// included.
+    internal var rowsKept: Bool {
+        switch transaction {
+        case .rolledBack, .endedByServer:
+            return false
+        case .untracked, .committed, .committedBeforeRollback, .endedByFile:
+            return true
+        }
+    }
+
+    internal var importedRowCount: Int {
+        outcomes.reduce(0) { total, outcome in
+            guard case .imported(let result) = outcome else { return total }
+            return total + result.executedStatements
+        }
+    }
+
+    internal var failure: (any Error)? {
+        guard case .failed(let index) = ending, case .failed(let error, _) = outcomes[index] else { return nil }
+        return error
+    }
+
+    /// A row or a whole file that Skip and Continue left out.
+    internal var hasFailures: Bool {
+        outcomes.contains { outcome in
+            if case .imported(let result) = outcome { return result.skippedStatements > 0 }
+            if case .failed = outcome { return true }
+            return false
+        }
+    }
+}
diff --git a/apple/TablePro/Core/Services/Export/RowImportTableMaker.swift b/apple/TablePro/Core/Services/Export/RowImportTableMaker.swift
new file mode 100644
index 0000000000..d639395b28
--- /dev/null
+++ b/apple/TablePro/Core/Services/Export/RowImportTableMaker.swift
@@ -0,0 +1,160 @@
+//
+//  RowImportTableMaker.swift
+//  TablePro
+//
+
+import Foundation
+import TableProPluginKit
+import TableProSQLGrammar
+
+/// Creates the tables a row import fills, or clears one it created on an earlier attempt, behind one gate grant
+/// that the rows reuse. One prompt covers every table of a run, so importing twenty files asks once.
+@MainActor
+internal struct RowImportTableMaker {
+    private enum Step {
+        case create(TableScope, statements: [String], sql: String)
+        case clear(String)
+    }
+
+    internal let connection: DatabaseConnection
+    private let gate: any ExecutionGate
+
+    internal init(connection: DatabaseConnection, gate: any ExecutionGate = ExecutionGateProvider.shared) {
+        self.connection = connection
+        self.gate = gate
+    }
+
+    internal static var statementError: NSError {
+        NSError(
+            domain: "RowImport", code: -1,
+            userInfo: [NSLocalizedDescriptionKey: String(localized: "Could not build the CREATE TABLE statement")]
+        )
+    }
+
+    /// `onCreated` hears of each table as soon as it exists, so a failure on a later one still leaves the planner
+    /// knowing the earlier ones are this sheet's to reuse.
+    internal func prepare(
+        _ definitions: [PluginCreateTableDefinition],
+        scope: DatabaseScope,
+        planner: NewTableImportPlanner,
+        description: String? = nil,
+        onCreated: (TableScope, String) -> Void
+    ) async throws -> OperationReceipt {
+        guard ImportDataSinkAdapter.canWriteRows(into: connection.type) else {
+            throw PluginImportError.importFailed(String(
+                format: String(localized: "%@ cannot take rows from this file, so no table was created."),
+                connection.type.rawValue
+            ))
+        }
+        var steps: [Step] = []
+        for definition in definitions {
+            steps.append(try await step(for: definition, scope: scope, planner: planner))
+        }
+
+        let receipt = try await authorize(steps, description: description)
+        var createdAny = false
+        /// Also when a later table fails, since the ones before it are in the database.
+        defer {
+            if createdAny {
+                CatalogChangeService.post(
+                    .changed(CatalogChange(connectionId: connection.id, database: scope.database, kinds: .tables))
+                )
+            }
+        }
+        for step in steps {
+            switch step {
+            case .create(let table, let statements, let sql):
+                for statement in statements {
+                    try await runOnLeasedDriver(statement, scope: scope)
+                    createdAny = true
+                }
+                onCreated(table, sql)
+            case .clear(let sql):
+                try await runOnLeasedDriver(sql, scope: scope)
+            }
+        }
+        return receipt
+    }
+
+    private func step(
+        for definition: PluginCreateTableDefinition,
+        scope: DatabaseScope,
+        planner: NewTableImportPlanner
+    ) async throws -> Step {
+        let tableName = definition.tableName
+        let table = TableScope(table: tableName, in: scope)
+        let statements = try await DatabaseManager.shared.createTableStatements(
+            definition: definition,
+            scope: scope,
+            route: DatabaseManager.shared.executionRoute(for: scope)
+        )
+        guard !statements.isEmpty else { throw Self.statementError }
+        let sql = statements.joined(separator: "\n")
+        switch planner.plan(forTable: table, createTableSQL: sql) {
+        case .create:
+            return .create(table, statements: statements, sql: sql)
+        case .reuseAfterClearing:
+            let generator = try SQLStatementGenerator(
+                tableName: tableName,
+                columns: [],
+                primaryKeyColumns: [],
+                databaseType: connection.type
+            )
+            return .clear(generator.deleteAllRowsStatement())
+        case .nameTakenWithDifferentColumns:
+            throw PluginImportError.importFailed(
+                String(
+                    format: String(localized: "The table %@ was already created with different columns. Choose another name."),
+                    tableName
+                )
+            )
+        }
+    }
+
+    /// Every statement the sheet issues on its own account goes through the gate. The retry path's `DELETE FROM`
+    /// used to skip it while the `CREATE TABLE` beside it did not, so a connection set to confirm destructive
+    /// statements emptied a table without asking.
+    private func authorize(_ steps: [Step], description: String?) async throws -> OperationReceipt {
+        var statements: [String] = []
+        var clears = false
+        for step in steps {
+            switch step {
+            case .create(_, let created, _):
+                statements += created
+            case .clear(let sql):
+                statements.append(sql)
+                clears = true
+            }
+        }
+        let decision = await gate.authorize(
+            OperationRequest(
+                connectionId: connection.id,
+                databaseType: connection.type,
+                sql: SQLScriptText(databaseType: connection.type).script(statements),
+                kind: clears ? .destructiveQuery : .schemaMutation,
+                caller: .userInterface,
+                capabilities: .interactiveUser,
+                operationDescription: description
+                    ?? (clears ? String(localized: "Clear Table") : String(localized: "Create Table"))
+            )
+        )
+        guard case .authorized(let receipt) = decision else {
+            throw decision.denialError ?? ExecutionGateError.denied(String(localized: "Operation not permitted"))
+        }
+        return receipt
+    }
+
+    /// One statement at a time on the import's own kind of lease, always after `authorize` has returned. Taking it
+    /// earlier would hold the connection's gate open across a confirmation the user has not answered yet, and the
+    /// gate is not reentrant, so the import that follows would then wait on a sheet waiting on the user.
+    private func runOnLeasedDriver(_ sql: String, scope: DatabaseScope) async throws {
+        let route = DatabaseManager.shared.executionRoute(for: scope)
+        _ = try await DatabaseManager.shared.withScopedDriver(
+            scope: scope,
+            route: route,
+            cancellation: .protectedWrite
+        ) { driver in
+            try await driver.execute(query: sql)
+        }
+    }
+}
diff --git a/apple/TablePro/Core/Services/Infrastructure/ImportFilePanel.swift b/apple/TablePro/Core/Services/Infrastructure/ImportFilePanel.swift
index 99dbb2bb68..fa5ac278d5 100644
--- a/apple/TablePro/Core/Services/Infrastructure/ImportFilePanel.swift
+++ b/apple/TablePro/Core/Services/Infrastructure/ImportFilePanel.swift
@@ -35,6 +35,24 @@ internal enum ImportFilePanel {
         )
     }
 
+    /// Import Data…, which also takes several files or a folder. One file goes where its format sends it; several, or
+    /// a folder, import as rows into a table each, so a folder passes only while it holds a file a row format reads.
+    internal static func presentFiles(
+        matching options: [ImportFormatOption],
+        readsRows: @escaping (String) -> Bool,
+        message: String,
+        in window: NSWindow?
+    ) async -> [URL] {
+        guard let panel = await run(
+            contentTypes: ImportFileFormatResolver.contentTypes(for: options),
+            gate: ImportFileGate(options: options, readsRows: readsRows),
+            allowsSeveral: true,
+            message: message,
+            in: window
+        ) else { return [] }
+        return panel.urls
+    }
+
     /// Import Data From > a named format, and the object browser's own import items. The user has
     /// already said what the file holds, so every file is enabled and nothing is read off the
     /// extension. That is the route for a CSV called `orders.txt` or carrying no extension at all.
@@ -51,6 +69,16 @@ internal enum ImportFilePanel {
         message: String,
         in window: NSWindow?
     ) async -> URL? {
+        await run(contentTypes: contentTypes, gate: gate, allowsSeveral: false, message: message, in: window)?.url
+    }
+
+    private static func run(
+        contentTypes: [UTType],
+        gate: ImportFileGate?,
+        allowsSeveral: Bool,
+        message: String,
+        in window: NSWindow?
+    ) async -> NSOpenPanel? {
         guard let window else {
             logger.warning("No host window, cannot present the import file panel")
             return nil
@@ -61,8 +89,8 @@ internal enum ImportFilePanel {
         panel.allowedContentTypes = contentTypes
         /// After `allowedContentTypes`: macOS 27 rewrites both from it.
         panel.canChooseFiles = true
-        panel.canChooseDirectories = false
-        panel.allowsMultipleSelection = false
+        panel.canChooseDirectories = allowsSeveral
+        panel.allowsMultipleSelection = allowsSeveral
         panel.message = message
 
         panel.delegate = gate
@@ -70,7 +98,7 @@ internal enum ImportFilePanel {
         withExtendedLifetime(gate) {}
 
         guard response == .OK else { return nil }
-        return panel.url
+        return panel
     }
 }
 
@@ -80,11 +108,25 @@ internal enum ImportFilePanel {
 private final class ImportFileGate: NSObject, NSOpenSavePanelDelegate {
     private let options: [ImportFormatOption]
 
-    init(options: [ImportFormatOption]) {
+    /// Set when the panel takes folders: one passes only while it holds a file a row format reads.
+    private let readsRows: ((String) -> Bool)?
+
+    init(options: [ImportFormatOption], readsRows: ((String) -> Bool)? = nil) {
         self.options = options
+        self.readsRows = readsRows
     }
 
+    /// AppKit sends this once for each selected file or folder.
     func panel(_ sender: Any, validate url: URL) throws {
+        if let readsRows, ImportBatchFiles.isDirectory(url) {
+            guard ImportBatchFiles.resolve([url], among: options, readsRows: readsRows).files.isEmpty else { return }
+            throw ImportFileGateError.emptyFolder(
+                folderName: url.lastPathComponent,
+                extensions: ImportFileFormatResolver.acceptedExtensions(
+                    for: options.filter { readsRows($0.id) }
+                )
+            )
+        }
         switch ImportFileFormatResolver.match(url, among: options) {
         case .format:
             return
@@ -115,9 +157,12 @@ private enum ImportFileGateError: LocalizedError {
     case unrecognized(fileName: String, extensions: [String])
     case compressedRowFormat(fileName: String, formatName: String)
     case ambiguous(fileName: String, formatNames: [String])
+    case emptyFolder(folderName: String, extensions: [String])
 
     var errorDescription: String? {
         switch self {
+        case .emptyFolder(let folderName, _):
+            return String(format: String(localized: "“%@” holds no file to import."), folderName)
         case .unrecognized(let fileName, _):
             return String(format: String(localized: "There is no import format for “%@”."), fileName)
         case .compressedRowFormat(let fileName, _):
@@ -129,6 +174,9 @@ private enum ImportFileGateError: LocalizedError {
 
     var recoverySuggestion: String? {
         switch self {
+        case .emptyFolder(_, let extensions):
+            let list = extensions.map { ".\($0)" }.joined(separator: ", ")
+            return String(format: String(localized: "A folder imports the %@ files directly inside it."), list)
         case .unrecognized(_, let extensions):
             let list = extensions.map { ".\($0)" }.joined(separator: ", ")
             return String(format: String(localized: "This connection imports %@."), list)
diff --git a/apple/TablePro/Resources/Localizable.xcstrings b/apple/TablePro/Resources/Localizable.xcstrings
index 08afbdda02..4de666133d 100644
--- a/apple/TablePro/Resources/Localizable.xcstrings
+++ b/apple/TablePro/Resources/Localizable.xcstrings
@@ -844,6 +844,9 @@
     },
     "“%@” Changed on Disk" : {
 
+    },
+    "“%@” holds no file to import." : {
+
     },
     "\"%@\" is a reserved theme identifier." : {
       "extractionState" : "manual",
@@ -3513,6 +3516,9 @@
           }
         }
       }
+    },
+    "%@ files cannot be imported together with other files." : {
+
     },
     "%@ had no rows to transfer." : {
       "localizations" : {
@@ -3549,6 +3555,9 @@
           }
         }
       }
+    },
+    "%@ has no column to create." : {
+
     },
     "%@ holds a document or an array, and this text is not valid JSON." : {
       "extractionState" : "manual",
@@ -5379,6 +5388,9 @@
     "%@ takes one level. End it with a semicolon before the next statement." : {
       "extractionState" : "manual",
       "generatesSymbol" : false
+    },
+    "%@ takes rows from two files, so Delete existing rows before import would remove the first file's rows." : {
+
     },
     "%@ to %@ characters" : {
       "localizations" : {
@@ -6670,6 +6682,9 @@
           }
         }
       }
+    },
+    "%1$@ cannot be read. %2$@" : {
+
     },
     "%1$@ chart of %2$@ by %3$@ with %4$d points" : {
       "localizations" : {
@@ -6901,6 +6916,12 @@
           }
         }
       }
+    },
+    "%1$@ failed, and the files imported before it were kept. %2$@" : {
+
+    },
+    "%1$@ failed, so every file was rolled back. %2$@" : {
+
     },
     "%1$@ for %2$@" : {
       "localizations" : {
@@ -7030,6 +7051,12 @@
           }
         }
       }
+    },
+    "%1$@ maps no field to a column of %2$@." : {
+
+    },
+    "%1$@ maps two fields to one column of %2$@." : {
+
     },
     "%1$@ must be at least %2$lld." : {
       "localizations" : {
@@ -7229,6 +7256,9 @@
     },
     "%1$@ takes %2$lld dimensions, and the value has %3$lld." : {
 
+    },
+    "%1$@ to %2$@: %3$@" : {
+
     },
     "%1$@ wants to run a query on \"%2$@\"" : {
       "localizations" : {
@@ -7932,6 +7962,9 @@
           }
         }
       }
+    },
+    "%1$lld of %2$lld fields mapped" : {
+
     },
     "%1$lld of %2$lld members" : {
       "localizations" : {
@@ -8072,6 +8105,9 @@
           }
         }
       }
+    },
+    "%1$lld rows from %2$lld files in %3$@ seconds." : {
+
     },
     "%1$lld rows from %3$lld tables written to %2$@." : {
       "localizations" : {
@@ -10384,6 +10420,12 @@
     },
     "%lld dims" : {
 
+    },
+    "%lld failed" : {
+
+    },
+    "%lld files" : {
+
     },
     "%lld in, %lld out" : {
       "localizations" : {
@@ -10605,6 +10647,9 @@
           }
         }
       }
+    },
+    "%lld new columns" : {
+
     },
     "%lld objects" : {
       "localizations" : {
@@ -11276,6 +11321,12 @@
           }
         }
       }
+    },
+    "%lld rows before it were kept." : {
+
+    },
+    "%lld rows before it were rolled back." : {
+
     },
     "%lld rows do not match the foreign keys this change asks for, so nothing was changed. Correct or remove those rows, then try again." : {
       "extractionState" : "manual",
@@ -13118,6 +13169,9 @@
           }
         }
       }
+    },
+    "1 file" : {
+
     },
     "1 item" : {
       "localizations" : {
@@ -13151,6 +13205,9 @@
           }
         }
       }
+    },
+    "1 new column" : {
+
     },
     "1 of %lld conflicts" : {
       "extractionState" : "stale",
@@ -13291,6 +13348,12 @@
           }
         }
       }
+    },
+    "1 row before it was kept." : {
+
+    },
+    "1 row before it was rolled back." : {
+
     },
     "1 row does not fit a %1$@ INSERT on its own, so its statement passes that size." : {
       "extractionState" : "manual",
@@ -15493,6 +15556,9 @@
     "A Firestore document is a JSON object of fields." : {
       "extractionState" : "manual",
       "generatesSymbol" : false
+    },
+    "A folder imports the %@ files directly inside it." : {
+
     },
     "A foreign key change recreates the table, and these cannot travel with it:\n\n%@\n\nSave them on their own first, then change the foreign key." : {
       "localizations" : {
@@ -33623,6 +33689,9 @@
           }
         }
       }
+    },
+    "Cannot read the file" : {
+
     },
     "Cannot Read the Source" : {
       "localizations" : {
@@ -37394,6 +37463,9 @@
           }
         }
       }
+    },
+    "Choose a table for %@." : {
+
     },
     "Choose a target to compare against." : {
       "localizations" : {
@@ -37616,6 +37688,9 @@
           }
         }
       }
+    },
+    "Choose at least one file to import." : {
+
     },
     "Choose at least one object to copy." : {
       "localizations" : {
@@ -68803,6 +68878,9 @@
           }
         }
       }
+    },
+    "Edit the import of %@" : {
+
     },
     "Edit Trigger" : {
       "localizations" : {
@@ -90794,6 +90872,9 @@
           }
         }
       }
+    },
+    "Import the rows of each file into its own table" : {
+
     },
     "Import Tunnel Commands?" : {
       "extractionState" : "stale",
@@ -99379,6 +99460,9 @@
           }
         }
       }
+    },
+    "Left out, because they do not import into a table: %@" : {
+
     },
     "left parenthesis" : {
       "extractionState" : "manual",
@@ -112734,6 +112818,9 @@
           }
         }
       }
+    },
+    "New table %@" : {
+
     },
     "New table:" : {
       "localizations" : {
@@ -119501,6 +119588,9 @@
           }
         }
       }
+    },
+    "None of these files import into a table: %@." : {
+
     },
     "None selected" : {
       "localizations" : {
@@ -120459,6 +120549,9 @@
           }
         }
       }
+    },
+    "not imported" : {
+
     },
     "not in list" : {
       "localizations" : {
@@ -121599,6 +121692,9 @@
           }
         }
       }
+    },
+    "Nothing to Import" : {
+
     },
     "Nothing to Restore" : {
       "localizations" : {
@@ -127392,6 +127488,9 @@
           }
         }
       }
+    },
+    "Options for" : {
+
     },
     "Options the engine offers" : {
       "localizations" : {
@@ -141067,6 +141166,9 @@
           }
         }
       }
+    },
+    "Reading the files…" : {
+
     },
     "Reading the source…" : {
       "localizations" : {
@@ -141090,6 +141192,9 @@
           }
         }
       }
+    },
+    "Reading…" : {
+
     },
     "Reads every table inside one transaction, so a dump of several tables is consistent with itself. The transaction stays open for the whole export." : {
       "extractionState" : "manual",
@@ -149745,6 +149850,9 @@
           }
         }
       }
+    },
+    "rolled back" : {
+
     },
     "root" : {
       "extractionState" : "stale",
@@ -158476,6 +158584,9 @@
           }
         }
       }
+    },
+    "Select files or a folder to import" : {
+
     },
     "Select files to open" : {
       "localizations" : {
@@ -161001,6 +161112,9 @@
           }
         }
       }
+    },
+    "Set On error and Wrap in transaction the same for %@." : {
+
     },
     "Set Password" : {
       "localizations" : {
@@ -172157,6 +172271,12 @@
           }
         }
       }
+    },
+    "Stopped at %@. Every file was rolled back." : {
+
+    },
+    "Stopped at %@. The files imported before it were kept." : {
+
     },
     "Stopped at %lld rows, the most R2 SQL returns from one query." : {
       "extractionState" : "manual",
@@ -172169,6 +172289,9 @@
           }
         }
       }
+    },
+    "Stopped." : {
+
     },
     "Stopping…" : {
       "extractionState" : "stale",
@@ -201222,6 +201345,9 @@
           }
         }
       }
+    },
+    "Two files create a table named %@." : {
+
     },
     "Two People" : {
       "extractionState" : "manual",
diff --git a/apple/TablePro/Views/Components/TransferResultAlert.swift b/apple/TablePro/Views/Components/TransferResultAlert.swift
index 8d3a82c75c..19f85c6ceb 100644
--- a/apple/TablePro/Views/Components/TransferResultAlert.swift
+++ b/apple/TablePro/Views/Components/TransferResultAlert.swift
@@ -211,6 +211,126 @@ internal enum TransferResultAlert {
         }
     }
 
+    /// A run of several files says what became of each, with its row count, and whether a failure took back the files
+    /// before it. Copy Details carries every skipped row's error as well.
+    internal static func presentBatchImportResult(
+        _ report: RowImportBatchReport,
+        window: NSWindow?,
+        completion: @escaping @MainActor () -> Void
+    ) {
+        let alert = NSAlert()
+        switch report.ending {
+        case .completed:
+            alert.messageText = report.hasFailures
+                ? String(localized: "Import completed with errors")
+                : String(localized: "Import completed")
+            alert.alertStyle = report.hasFailures ? .warning : .informational
+        case .failed:
+            alert.messageText = String(localized: "Import failed")
+            alert.alertStyle = .critical
+        case .stopped:
+            alert.messageText = String(localized: "Import stopped")
+            alert.alertStyle = .warning
+        }
+        alert.informativeText = batchImportSummary(report)
+        AlertHelper.addCancelButton(to: alert, title: String(localized: "Done"))
+        let lines = batchImportLines(report)
+        alert.accessoryView = TransferReportView(
+            shown: lines.joined(separator: "\n"),
+            copied: ([lines.joined(separator: "\n")] + batchImportErrors(report)).joined(separator: "\n\n")
+        )
+        alert.layout()
+        AlertHelper.present(alert, in: window) { _ in completion() }
+    }
+
+    internal static func batchImportSummary(_ report: RowImportBatchReport) -> String {
+        let seconds = String(format: "%.2f", report.executionTime)
+        switch report.ending {
+        case .completed:
+            return String(
+                format: String(localized: "%1$lld rows from %2$lld files in %3$@ seconds."),
+                Int64(report.importedRowCount),
+                Int64(report.files.count),
+                seconds
+            )
+        case .failed(let index):
+            let failure = importFailureText(for: report.failure)
+            let template = report.rowsKept
+                ? String(localized: "%1$@ failed, and the files imported before it were kept. %2$@")
+                : String(localized: "%1$@ failed, so every file was rolled back. %2$@")
+            return String(format: template, report.files[index].sourceName, failure)
+        case .stopped(let index):
+            let template = report.rowsKept
+                ? String(localized: "Stopped at %@. The files imported before it were kept.")
+                : String(localized: "Stopped at %@. Every file was rolled back.")
+            return String(format: template, report.files[index].sourceName)
+        }
+    }
+
+    /// One line per file: where its rows went and how many, or why none did.
+    internal static func batchImportLines(_ report: RowImportBatchReport) -> [String] {
+        zip(report.files, report.outcomes).map { file, outcome in
+            let status: String
+            switch outcome {
+            case .imported(let result):
+                status = importedRows(result, kept: report.rowsKept)
+            case .failed(let error, let rows):
+                status = [importFailureText(for: error), rowsBefore(rows, kept: report.rowsKept)]
+                    .compactMap { $0 }
+                    .joined(separator: " ")
+            case .stopped(let rows):
+                status = [String(localized: "Stopped."), rowsBefore(rows, kept: report.rowsKept)]
+                    .compactMap { $0 }
+                    .joined(separator: " ")
+            case .notRun:
+                status = String(localized: "not imported")
+            }
+            return String(format: String(localized: "%1$@ to %2$@: %3$@"), file.sourceName, file.targetTable, status)
+        }
+    }
+
+    /// A file that stopped part way can already have written batches, which stay or go with the run.
+    private static func rowsBefore(_ rows: Int, kept: Bool) -> String? {
+        guard rows > 0 else { return nil }
+        if rows == 1 {
+            return kept
+                ? String(localized: "1 row before it was kept.")
+                : String(localized: "1 row before it was rolled back.")
+        }
+        let template = kept
+            ? String(localized: "%lld rows before it were kept.")
+            : String(localized: "%lld rows before it were rolled back.")
+        return String(format: template, Int64(rows))
+    }
+
+    private static func importedRows(_ result: PluginImportResult, kept: Bool) -> String {
+        let rows = result.executedStatements
+        var text = rows == 1
+            ? String(localized: "1 row")
+            : String(format: String(localized: "%lld rows"), Int64(rows))
+        if result.skippedStatements > 0 {
+            text += ", " + String(format: String(localized: "%lld failed"), Int64(result.skippedStatements))
+        }
+        if !kept {
+            text += ", " + String(localized: "rolled back")
+        }
+        return text
+    }
+
+    /// Everything the lines cut short: each skipped row's error, and a failed file's whole error and statement.
+    private static func batchImportErrors(_ report: RowImportBatchReport) -> [String] {
+        zip(report.files, report.outcomes).compactMap { file, outcome in
+            switch outcome {
+            case .imported(let result) where !result.errors.isEmpty:
+                return "\(file.sourceName)\n\(failureReport(for: result.errors))"
+            case .failed(let error, _):
+                return "\(file.sourceName)\n\(failureReport(for: error) ?? error.localizedDescription)"
+            case .imported, .stopped, .notRun:
+                return nil
+            }
+        }
+    }
+
     internal static func presentImportFailure(
         error: (any Error)?,
         window: NSWindow?,
diff --git a/apple/TablePro/Views/Import/ImportProgressView.swift b/apple/TablePro/Views/Import/ImportProgressView.swift
index 0d90b34d0d..db754ad3b8 100644
--- a/apple/TablePro/Views/Import/ImportProgressView.swift
+++ b/apple/TablePro/Views/Import/ImportProgressView.swift
@@ -23,6 +23,15 @@ struct ImportProgressView: View {
                 .font(.title3.weight(.semibold))
 
             VStack(spacing: 8) {
+                if service.state.fileCount > 1 {
+                    HStack {
+                        Text(fileLine)
+                            .font(.body)
+                            .lineLimit(1)
+                            .truncationMode(.middle)
+                        Spacer()
+                    }
+                }
                 HStack {
                     if service.state.statusMessage.isEmpty {
                         Text("Executed ^[\(service.state.processedStatements) statement](inflect: true)")
@@ -60,6 +69,15 @@ struct ImportProgressView: View {
         }
     }
 
+    private var fileLine: String {
+        String(
+            format: String(localized: "%1$@ (%2$lld of %3$lld)"),
+            service.state.fileName,
+            Int64(service.state.fileIndex + 1),
+            Int64(service.state.fileCount)
+        )
+    }
+
     private var progressValue: Double {
         guard service.state.estimatedTotalStatements > 0 else { return 0 }
         return min(
diff --git a/apple/TablePro/Views/Import/RowImportBatch.swift b/apple/TablePro/Views/Import/RowImportBatch.swift
new file mode 100644
index 0000000000..355ba8a4d8
--- /dev/null
+++ b/apple/TablePro/Views/Import/RowImportBatch.swift
@@ -0,0 +1,653 @@
+//
+//  RowImportBatch.swift
+//  TablePro
+//
+
+import Combine
+import Foundation
+import os
+import TableProPluginKit
+
+/// Several files imported as rows in one run, each into its own table: what each file is read as, where it goes,
+/// and whether the run as a whole can start. Held apart from `RowImportBatchSheet` so its rules are testable.
+@MainActor
+internal final class RowImportBatch: ObservableObject {
+    internal enum ReadState: Equatable {
+        case pending
+        case reading
+        case ready
+        case failed(String)
+    }
+
+    /// The identity of a read: the file's destination, its table and the format's options. A file reads again
+    /// when any of them changes.
+    internal struct ReadKey: Hashable {
+        let destination: RowImportDestination
+        let table: String?
+        let signature: String
+    }
+
+    internal enum Choice: Hashable {
+        case newTable
+        case existingTable(String)
+    }
+
+    internal struct File: Identifiable {
+        internal let id = UUID()
+        internal let url: URL
+        internal let formatId: String
+        internal var isIncluded = true
+        internal var target: RowImportTarget
+
+        /// The target is still the run's own proposal, which a table list arriving later may replace.
+        internal var targetIsProposed = true
+        internal var read: ReadState = .pending
+        internal var loadedRead: ReadKey?
+
+        internal var sourceName: String {
+            url.lastPathComponent
+        }
+    }
+
+    /// What the sheet needs from the database and the plugins, so a test can stand in for both.
+    internal struct Services {
+        internal var fetchTables: @MainActor (DatabaseScope) async throws -> [TableInfo]
+        internal var fetchColumns: @MainActor (String, DatabaseScope) async throws -> [String]
+        internal var detectFields: @MainActor (String, URL, String?) async throws -> [PluginImportField]
+        internal var configuration: @MainActor (String) -> RowImportRunner.Configuration?
+        internal var formatName: @MainActor (String) -> String
+        internal var sqlType: @MainActor (PluginImportFieldType) -> String
+
+        internal static func live(for connection: DatabaseConnection) -> Services {
+            Services(
+                fetchTables: { scope in
+                    try await DatabaseManager.shared.withMetadataDriver(scope: scope) { driver in
+                        try await driver.fetchTables()
+                    }
+                },
+                fetchColumns: { table, scope in
+                    try await DatabaseManager.shared.withMetadataDriver(scope: scope) { driver in
+                        try await driver.fetchColumns(table: table)
+                    }.map(\.name)
+                },
+                detectFields: { formatId, url, table in
+                    guard let plugin = PluginManager.shared.importPlugin(forFormat: formatId) else {
+                        throw PluginImportError.importFailed("Import format '\(formatId)' not found")
+                    }
+                    return try await ImportFieldDetection.detectFields(plugin: plugin, at: url, targetTable: table)
+                },
+                configuration: { formatId in
+                    PluginManager.shared.importPlugin(forFormat: formatId)?.rowImportConfiguration
+                },
+                formatName: { formatId in
+                    PluginManager.shared.importPlugin(forFormat: formatId).map { type(of: $0).formatDisplayName }
+                        ?? formatId.uppercased()
+                },
+                sqlType: { fieldType in
+                    let serverVersion = DatabaseManager.shared.driver(for: connection.id)?.serverVersion
+                    return ImportTypeMapper.sqlType(
+                        for: fieldType,
+                        databaseType: connection.type,
+                        serverVersion: serverVersion
+                    )
+                }
+            )
+        }
+    }
+
+    private static let logger = Logger(subsystem: "com.TablePro", category: "RowImportBatch")
+
+    @Published internal private(set) var files: [File]
+    @Published internal private(set) var tables: [TableInfo] = []
+
+    /// Folded names of every object the database holds, nil while the list is unknown, as in the one-file sheet.
+    @Published internal private(set) var catalogNameKeys: Set<String>?
+    @Published internal private(set) var tableListError: String?
+    @Published internal private(set) var isLoadingTables = false
+
+    /// Reads wait for the first table list, which can turn a file's proposed new table into the table it matches.
+    @Published internal private(set) var isCatalogSettled = false
+    @Published internal private(set) var signatures: [String: String] = [:]
+
+    /// Each format's error handling and transaction options as its options view last left them. They change no
+    /// read, so they are kept apart from the signature, and the readiness they decide follows them here.
+    @Published internal private(set) var configurations: [String: RowImportRunner.Configuration] = [:]
+    @Published internal private(set) var readRetries = 0
+
+    /// The tables this run created, so a retry after a failed run reuses them rather than tripping on their names.
+    @Published internal var planner = NewTableImportPlanner()
+
+    @Published internal private(set) var leftOut: [URL]
+    internal let scope: DatabaseScope?
+    private let databaseType: DatabaseType
+    private let style: NewTableNameStyle
+    private let services: Services
+    private let mappingStore: ImportColumnMappingStore
+    private var columnsByTable: [String: [String]] = [:]
+
+    internal init(
+        files: [ImportBatchFiles.File],
+        leftOut: [URL],
+        connection: DatabaseConnection,
+        scope: DatabaseScope?,
+        services: Services,
+        mappingStore: ImportColumnMappingStore = .shared
+    ) {
+        self.mappingStore = mappingStore
+        self.leftOut = leftOut
+        self.scope = scope
+        self.databaseType = connection.type
+        self.style = NewTableNameStyle.forDatabaseType(connection.type)
+        self.services = services
+        self.files = files.map { File(url: $0.url, formatId: $0.formatId, target: .newTable(named: "")) }
+        proposeTargets()
+    }
+
+    // MARK: - Formats
+
+    /// Every format the files are in, in the order the files first name them.
+    internal var formatIds: [String] {
+        var seen = Set<String>()
+        return files.map(\.formatId).filter { seen.insert($0).inserted }
+    }
+
+    internal func formatName(_ formatId: String) -> String {
+        services.formatName(formatId)
+    }
+
+    internal func setSignature(_ signature: String, for formatId: String) {
+        guard signatures[formatId] != signature else { return }
+        signatures[formatId] = signature
+    }
+
+    internal func setConfiguration(_ configuration: RowImportRunner.Configuration?, for formatId: String) {
+        configurations[formatId] = configuration
+    }
+
+    private func configuration(for formatId: String) -> RowImportRunner.Configuration? {
+        configurations[formatId] ?? services.configuration(formatId)
+    }
+
+    // MARK: - Catalog
+
+    /// Tables alone, because they are the only objects a file's rows can be inserted into.
+    internal var availableTables: [TableInfo] {
+        tables.filter(\.type.acceptsImportedRows)
+    }
+
+    internal func loadTables() async {
+        guard !isLoadingTables else { return }
+        defer { isCatalogSettled = true }
+        guard let scope else {
+            catalogNameKeys = nil
+            tableListError = String(localized: "This connection is not open.")
+            return
+        }
+        isLoadingTables = true
+        defer { isLoadingTables = false }
+        do {
+            let objects = try await services.fetchTables(scope)
+            tables = objects
+            catalogNameKeys = NewTableNaming.comparisonKeys(for: objects.map(\.name))
+            tableListError = nil
+            columnsByTable = [:]
+        } catch {
+            catalogNameKeys = nil
+            tableListError = error.localizedDescription
+            Self.logger.warning("Failed to load tables: \(error.publicLogShape, privacy: .public)")
+        }
+        proposeTargets()
+    }
+
+    /// Names a new table must not take: the catalog's, less the tables this run created, which a retry lands on.
+    private var catalogKeysOutsideRun: Set<String> {
+        let ours = NewTableNaming.comparisonKeys(for: scope.map(planner.createdTableNames(in:)) ?? [])
+        return (catalogNameKeys ?? []).subtracting(ours)
+    }
+
+    /// A file goes to the table its name matches, as the one-file sheet would name it, or to a new table under that
+    /// name. Two files never get one new name. Only a target the user has not changed is replaced, and a table this
+    /// run created is not a match: the retry lands on it as the new table it was.
+    private func proposeTargets() {
+        let ours = NewTableNaming.comparisonKeys(for: scope.map(planner.createdTableNames(in:)) ?? [])
+        let tablesByKey = Dictionary(
+            availableTables.map { ($0.name.lowercased(), $0.name) }.filter { !ours.contains($0.0) },
+            uniquingKeysWith: { first, _ in first }
+        )
+        var taken = catalogKeysOutsideRun
+        for file in files where !file.targetIsProposed && file.target.destination == .newTable {
+            taken.insert(file.target.newTableName.trimmingCharacters(in: .whitespaces).lowercased())
+        }
+        for index in files.indices where files[index].targetIsProposed {
+            let base = NewTableNaming.suggestion(forFileNamed: files[index].sourceName, style: style, avoiding: [])
+            if let table = tablesByKey[base.lowercased()] {
+                if files[index].target.destination != .existingTable || files[index].target.existingTable != table {
+                    files[index].target = .existingTable(table, proposingNewTable: "")
+                }
+                continue
+            }
+            let name = NewTableNaming.suggestion(forFileNamed: files[index].sourceName, style: style, avoiding: taken)
+            taken.insert(name.lowercased())
+            if files[index].target.destination == .newTable {
+                files[index].target.newTableName = name
+                files[index].target.proposedTableName = name
+            } else {
+                files[index].target = .newTable(named: name)
+            }
+        }
+    }
+
+    // MARK: - Choices
+
+    /// Files and folders dropped on the list join the run after the files already in it.
+    internal func add(_ resolution: ImportBatchFiles.Resolution) {
+        let known = Set(files.map { $0.url.standardizedFileURL })
+        let added = resolution.files.filter { !known.contains($0.url.standardizedFileURL) }
+        files += added.map { File(url: $0.url, formatId: $0.formatId, target: .newTable(named: "")) }
+        leftOut += resolution.leftOut.filter { !leftOut.contains($0) }
+        proposeTargets()
+    }
+
+    internal func file(_ id: File.ID) -> File? {
+        files.first { $0.id == id }
+    }
+
+    private func index(of id: File.ID) -> Int? {
+        files.firstIndex { $0.id == id }
+    }
+
+    internal func setIncluded(_ isIncluded: Bool, for id: File.ID) {
+        guard let index = index(of: id) else { return }
+        files[index].isIncluded = isIncluded
+    }
+
+    internal func choice(for file: File) -> Choice {
+        switch file.target.destination {
+        case .newTable:
+            return .newTable
+        case .existingTable:
+            return file.target.existingTable.map(Choice.existingTable) ?? .newTable
+        }
+    }
+
+    internal func choose(_ choice: Choice, for id: File.ID) {
+        guard let index = index(of: id) else { return }
+        files[index].targetIsProposed = false
+        switch choice {
+        case .newTable:
+            files[index].target.destination = .newTable
+            if files[index].target.newTableName.trimmingCharacters(in: .whitespaces).isEmpty {
+                let name = NewTableNaming.suggestion(
+                    forFileNamed: files[index].sourceName,
+                    style: style,
+                    avoiding: catalogKeysOutsideRun.union(newTableKeys(excluding: id))
+                )
+                files[index].target.newTableName = name
+                files[index].target.proposedTableName = name
+            }
+        case .existingTable(let table):
+            files[index].target.destination = .existingTable
+            files[index].target.existingTable = table
+        }
+    }
+
+    /// What the sheet that edits one file handed back, which answers a read of the current options.
+    internal func apply(_ target: RowImportTarget, to id: File.ID) {
+        guard let index = index(of: id) else { return }
+        files[index].target = target
+        files[index].targetIsProposed = false
+        files[index].read = .ready
+        files[index].loadedRead = readKey(for: files[index])
+    }
+
+    internal func edit(for id: File.ID, onDone: @escaping @MainActor (RowImportTarget) -> Void) -> RowImportSheetEdit? {
+        guard let scope, let file = file(id) else { return nil }
+        return RowImportSheetEdit(
+            scope: scope,
+            target: file.target,
+            planner: planner,
+            takenTableNames: newTableKeys(excluding: id),
+            onDone: onDone
+        )
+    }
+
+    /// Folded names of the new tables the other included files create.
+    private func newTableKeys(excluding id: File.ID) -> Set<String> {
+        Set(files.compactMap { file in
+            guard file.id != id, file.isIncluded, file.target.destination == .newTable else { return nil }
+            let name = file.target.newTableName.trimmingCharacters(in: .whitespaces)
+            return name.isEmpty ? nil : name.lowercased()
+        })
+    }
+
+    // MARK: - Reading
+
+    internal func readKey(for file: File) -> ReadKey {
+        let isExisting = file.target.destination == .existingTable
+        return ReadKey(
+            destination: file.target.destination,
+            table: isExisting ? file.target.existingTable : nil,
+            signature: signatures[file.formatId] ?? ""
+        )
+    }
+
+    internal struct ReadRequest: Hashable {
+        let keys: [ReadKey?]
+        let retries: Int
+        let isCatalogSettled: Bool
+    }
+
+    /// Changes whenever a file needs a read, so `.task(id:)` cancels the reads in flight and starts again.
+    internal var readRequest: ReadRequest {
+        ReadRequest(
+            keys: files.map { $0.isIncluded ? readKey(for: $0) : nil },
+            retries: readRetries,
+            isCatalogSettled: isCatalogSettled
+        )
+    }
+
+    internal func needsRead(_ file: File) -> Bool {
+        guard file.isIncluded else { return false }
+        if file.target.destination == .existingTable, file.target.existingTable == nil { return false }
+        return file.loadedRead != readKey(for: file)
+    }
+
+    /// One file at a time, so twenty files do not open twenty reads at once.
+    internal func readFiles() async {
+        guard isCatalogSettled else { return }
+        let pending = files.filter(needsRead).map(\.id)
+        for id in pending {
+            guard !Task.isCancelled else { return }
+            await read(id)
+        }
+    }
+
+    internal func retryRead(of id: File.ID) {
+        guard let index = index(of: id) else { return }
+        files[index].loadedRead = nil
+        files[index].read = .pending
+        readRetries += 1
+    }
+
+    private func read(_ id: File.ID) async {
+        guard let index = index(of: id) else { return }
+        let file = files[index]
+        let key = readKey(for: file)
+        files[index].read = .reading
+        do {
+            switch key.destination {
+            case .newTable:
+                let fields = try await services.detectFields(file.formatId, file.url, nil)
+                guard let index = landing(id, key) else { return }
+                files[index].target.newTable.load(fields: fields) { services.sqlType($0) }
+            case .existingTable:
+                guard let table = key.table, let scope else { return }
+                let columns = try await columns(of: table, in: scope)
+                let fields = try await services.detectFields(file.formatId, file.url, table)
+                guard let index = landing(id, key) else { return }
+                files[index].target.mapping = mapping(
+                    fields: fields,
+                    columns: columns,
+                    table: table,
+                    scope: scope,
+                    prior: files[index].target.mapping
+                )
+            }
+            guard let index = landing(id, key) else { return }
+            files[index].read = .ready
+            files[index].loadedRead = key
+        } catch {
+            guard let index = landing(id, key) else { return }
+            files[index].read = .failed(error.localizedDescription)
+            files[index].loadedRead = key
+            Self.logger.warning("Failed to read an import file: \(error.publicLogShape, privacy: .public)")
+        }
+    }
+
+    /// The file's slot when the read still answers what the file asks for now.
+    private func landing(_ id: File.ID, _ key: ReadKey) -> Int? {
+        guard !Task.isCancelled, let index = index(of: id), readKey(for: files[index]) == key else { return nil }
+        return index
+    }
+
+    private func columns(of table: String, in scope: DatabaseScope) async throws -> [String] {
+        if let cached = columnsByTable[table] { return cached }
+        let columns = try await services.fetchColumns(table, scope)
+        columnsByTable[table] = columns
+        return columns
+    }
+
+    /// The choices already made for this table first, then the mapping saved for it, then the name match: the order
+    /// the one-file sheet restores them in.
+    private func mapping(
+        fields: [PluginImportField],
+        columns: [String],
+        table: String,
+        scope: DatabaseScope,
+        prior: RowImportTarget.Mapping?
+    ) -> RowImportTarget.Mapping {
+        let saved = mappingStore.overrides(for: TableScope(table: table, in: scope))
+        var tiers = [saved.mapValues(\.choice)]
+        if let prior, prior.table == table {
+            tiers.insert(prior.choicesByField, at: 0)
+        }
+        let choices = ImportColumnMatcher.resolve(fields: fields.map(\.name), columns: columns, preferring: tiers)
+        return RowImportTarget.Mapping(table: table, fields: fields, columns: columns, choices: choices)
+    }
+
+    // MARK: - Readiness
+
+    internal var includedFiles: [File] {
+        files.filter(\.isIncluded)
+    }
+
+    internal var isReading: Bool {
+        includedFiles.contains { $0.read == .reading || needsRead($0) }
+    }
+
+    /// What stands between the run and Import, the first of it, naming the file it is about.
+    internal var problem: String? {
+        let included = includedFiles
+        guard !included.isEmpty else {
+            return String(localized: "Choose at least one file to import.")
+        }
+        for file in included {
+            if let problem = problem(of: file) {
+                return problem
+            }
+        }
+        if let name = duplicateNewTableName(in: included) {
+            return String(format: String(localized: "Two files create a table named %@."), name)
+        }
+        if let table = clearedTwice(in: included) {
+            return String(
+                format: String(localized: """
+                    %@ takes rows from two files, so Delete existing rows before import would remove the first \
+                    file's rows.
+                    """),
+                table
+            )
+        }
+        return policyProblem(in: included)
+    }
+
+    internal func problem(of file: File) -> String? {
+        if case .failed(let reason) = file.read {
+            return String(format: String(localized: "%1$@ cannot be read. %2$@"), file.sourceName, reason)
+        }
+        switch file.target.destination {
+        case .existingTable:
+            guard let table = file.target.existingTable else {
+                return String(format: String(localized: "Choose a table for %@."), file.sourceName)
+            }
+            guard file.read == .ready, let mapping = file.target.mapping else { return nil }
+            if mapping.mappedFieldCount == 0 {
+                return String(format: String(localized: "%1$@ maps no field to a column of %2$@."), file.sourceName, table)
+            }
+            if mapping.mapsOneColumnTwice {
+                return String(
+                    format: String(localized: "%1$@ maps two fields to one column of %2$@."), file.sourceName, table
+                )
+            }
+            return nil
+        case .newTable:
+            if let problem = nameProblem(of: file) {
+                return String(
+                    format: String(localized: "%1$@: %2$@"),
+                    file.sourceName,
+                    problem.message(for: file.target.newTableName)
+                )
+            }
+            guard file.read == .ready else { return nil }
+            if let problem = file.target.newTable.problem {
+                return String(format: String(localized: "%1$@: %2$@"), file.sourceName, problem.message)
+            }
+            if !file.target.newTable.hasNamedColumn {
+                return String(format: String(localized: "%@ has no column to create."), file.sourceName)
+            }
+            return nil
+        }
+    }
+
+    private func nameProblem(of file: File) -> NewTableNameProblem? {
+        let name = file.target.newTableName
+        let problem = NewTableNaming.problem(with: name, style: style, existingNames: catalogNameKeys)
+        let trimmed = name.trimmingCharacters(in: .whitespaces)
+        guard problem == .nameTaken, let scope, planner.created(TableScope(table: trimmed, in: scope)) else {
+            return problem
+        }
+        return nil
+    }
+
+    private func duplicateNewTableName(in included: [File]) -> String? {
+        var seen = Set<String>()
+        for file in included where file.target.destination == .newTable {
+            let name = file.target.newTableName.trimmingCharacters(in: .whitespaces)
+            guard seen.insert(name.lowercased()).inserted else { return name }
+        }
+        return nil
+    }
+
+    /// The first file into a table clears it, and the second clears it again, taking the first file's rows. Grouped by
+    /// the table each file ends in, new or existing, since a table a failed run created is offered as existing too.
+    private func clearedTwice(in included: [File]) -> String? {
+        let grouped = Dictionary(grouping: included.filter { $0.target.tableName != nil }) {
+            $0.target.tableName?.lowercased() ?? ""
+        }
+        for (_, group) in grouped.sorted(by: { $0.key < $1.key }) where group.count > 1 {
+            let clears = group.contains { configuration(for: $0.formatId)?.deleteExistingRows == true }
+            if clears { return group.first?.target.tableName }
+        }
+        return nil
+    }
+
+    private func policyProblem(in included: [File]) -> String? {
+        var formats: [String] = []
+        for file in included where !formats.contains(file.formatId) {
+            formats.append(file.formatId)
+        }
+        let policies = formats.map { configuration(for: $0).map(RowImportBatchPolicy.init) }
+        if let index = policies.firstIndex(where: { $0 == nil }) {
+            return String(
+                format: String(localized: "%@ files cannot be imported together with other files."),
+                formatName(formats[index])
+            )
+        }
+        let distinct = Set(policies.compactMap { $0.map { "\($0.errorHandling.rawValue)|\($0.usesTransaction)" } })
+        guard distinct.count > 1 else { return nil }
+        return String(
+            format: String(localized: "Set On error and Wrap in transaction the same for %@."),
+            formats.map(formatName).formatted(.list(type: .and))
+        )
+    }
+
+    internal var policy: RowImportBatchPolicy? {
+        includedFiles.first.flatMap { configuration(for: $0.formatId) }.map(RowImportBatchPolicy.init)
+    }
+
+    internal var canImport: Bool {
+        !isReading && problem == nil && includedFiles.allSatisfy { $0.read == .ready } && policy != nil
+    }
+
+    // MARK: - Plan
+
+    internal struct PlannedMapping {
+        internal let fields: [String]
+        internal let columns: [String]
+        internal let columnMapping: [String: String]
+        internal let table: String
+    }
+
+    internal struct Plan {
+        internal let items: [RowImportRunItem]
+        internal let newTables: [PluginCreateTableDefinition]
+        internal let mappings: [PlannedMapping]
+    }
+
+    /// The run as the included files stand, or nil while any of them is not ready.
+    internal func plan() -> Plan? {
+        guard canImport else { return nil }
+        var items: [RowImportRunItem] = []
+        var newTables: [PluginCreateTableDefinition] = []
+        var mappings: [PlannedMapping] = []
+        for file in includedFiles {
+            let fields: [String]
+            let columns: [String]
+            let columnMapping: [String: String]
+            let table: String
+            switch file.target.destination {
+            case .existingTable:
+                guard let name = file.target.existingTable, let mapping = file.target.mapping else { return nil }
+                table = name
+                fields = mapping.fieldNames
+                columns = mapping.columns
+                columnMapping = mapping.columnMapping
+            case .newTable:
+                guard let name = file.target.tableName,
+                      let definition = file.target.newTable.definition(tableName: name) else { return nil }
+                table = name
+                newTables.append(definition)
+                columnMapping = file.target.newTable.columnMapping
+                fields = file.target.newTable.fields
+                columns = fields.compactMap { columnMapping[$0] }
+            }
+            items.append(RowImportRunItem(
+                url: file.url,
+                sourceName: file.sourceName,
+                formatId: file.formatId,
+                targetTable: table,
+                columnMapping: columnMapping,
+                sourceFields: Set(fields)
+            ))
+            mappings.append(PlannedMapping(fields: fields, columns: columns, columnMapping: columnMapping, table: table))
+        }
+        return Plan(items: items, newTables: newTables, mappings: mappings)
+    }
+
+    // MARK: - Summary
+
+    internal func summary(of file: File) -> String {
+        switch file.read {
+        case .pending, .reading:
+            return String(localized: "Reading…")
+        case .failed:
+            return String(localized: "Cannot read the file")
+        case .ready:
+            break
+        }
+        switch file.target.destination {
+        case .existingTable:
+            guard let mapping = file.target.mapping else { return String(localized: "Reading…") }
+            return String(
+                format: String(localized: "%1$lld of %2$lld fields mapped"),
+                Int64(mapping.mappedFieldCount),
+                Int64(mapping.fields.count)
+            )
+        case .newTable:
+            let count = file.target.newTable.columnMapping.count
+            return count == 1
+                ? String(localized: "1 new column")
+                : String(format: String(localized: "%lld new columns"), Int64(count))
+        }
+    }
+}
diff --git a/apple/TablePro/Views/Import/RowImportBatchSheet.swift b/apple/TablePro/Views/Import/RowImportBatchSheet.swift
new file mode 100644
index 0000000000..c043952aa7
--- /dev/null
+++ b/apple/TablePro/Views/Import/RowImportBatchSheet.swift
@@ -0,0 +1,531 @@
+//
+//  RowImportBatchSheet.swift
+//  TablePro
+//
+
+import AppKit
+import Combine
+import SwiftUI
+import TableProPluginKit
+import UniformTypeIdentifiers
+
+/// Several files, or a folder of them, imported as rows into a table each in one run. Every file starts on the table
+/// its name matches or a new table named after it, Edit… opens the one-file sheet on it, and the options of each
+/// format are set once for every file in that format.
+struct RowImportBatchSheet: View {
+    private struct EditRequest: Identifiable {
+        let id: RowImportBatch.File.ID
+    }
+
+    @ObservedObject private var pluginManager = PluginManager.shared
+    @Binding var isPresented: Bool
+    let connection: DatabaseConnection
+
+    @StateObject private var batch: RowImportBatch
+    @State private var shownFormat: String?
+    @State private var editing: EditRequest?
+    @State private var isDropTargeted = false
+
+    /// The plugins' options are shared and persistent, and this sheet edits them in place, so Cancel puts them back.
+    @State private var settingsSnapshot: PluginSettingsSnapshot?
+    @State private var importSucceeded = false
+    @State private var importService: ImportService?
+    @State private var showProgressDialog = false
+    @State private var importTask: Task<Void, Never>?
+
+    /// The sheet's own window for its alerts, never `NSApp.keyWindow`, which is the progress sheet while it closes.
+    @State private var hostWindow: NSWindow?
+
+    init(
+        isPresented: Binding<Bool>,
+        connection: DatabaseConnection,
+        files: [ImportBatchFiles.File],
+        leftOut: [URL]
+    ) {
+        _isPresented = isPresented
+        self.connection = connection
+        _batch = StateObject(wrappedValue: RowImportBatch(
+            files: files,
+            leftOut: leftOut,
+            connection: connection,
+            scope: DatabaseManager.shared.browseScope(for: connection.id),
+            services: .live(for: connection)
+        ))
+    }
+
+    var body: some View {
+        VStack(spacing: 0) {
+            headerView
+                .padding()
+            Divider()
+
+            fileTable
+                .frame(maxWidth: .infinity, minHeight: 200, maxHeight: .infinity)
+                .sheet(item: $editing) { request in
+                    editSheet(for: request.id)
+                }
+                .onDrop(of: [.fileURL], isTargeted: $isDropTargeted) { providers in
+                    accept(providers)
+                }
+                .overlay {
+                    if isDropTargeted {
+                        RoundedRectangle(cornerRadius: 4)
+                            .strokeBorder(Color.accentColor, lineWidth: 2)
+                            .allowsHitTesting(false)
+                    }
+                }
+            if !batch.leftOut.isEmpty || batch.tableListError != nil {
+                notes
+                    .padding(.horizontal)
+                    .padding(.vertical, 8)
+            }
+            Divider()
+
+            optionsForm
+                .padding(.horizontal)
+                .padding(.vertical, 10)
+            Divider()
+
+            footerView
+                .padding()
+        }
+        .frame(minWidth: 780, minHeight: 600, idealHeight: 680, maxHeight: .infinity)
+        .background {
+            WindowAccessor { window in
+                hostWindow = window
+            }
+        }
+        .background { signatureObservers }
+        .task {
+            settingsSnapshot = PluginSettingsSnapshot(plugins: batch.formatIds.compactMap {
+                plugin(for: $0) as? any SettablePluginDiscoverable
+            })
+            for formatId in batch.formatIds {
+                batch.setSignature(plugin(for: formatId)?.fieldDetectionSignature ?? "", for: formatId)
+            }
+            await batch.loadTables()
+        }
+        .task(id: batch.readRequest) {
+            await batch.readFiles()
+        }
+        .onDisappear {
+            importTask?.cancel()
+            if !importSucceeded { settingsSnapshot?.restore() }
+            settingsSnapshot = nil
+        }
+        .sheet(isPresented: $showProgressDialog) {
+            if let service = importService {
+                ImportProgressView(service: service) { service.cancelImport() }
+                    .interactiveDismissDisabled()
+            }
+        }
+    }
+
+    // MARK: - Header
+
+    private var headerView: some View {
+        HStack(alignment: .top, spacing: 12) {
+            Image(systemName: "tablecells")
+                .font(.title)
+                .foregroundStyle(.blue)
+            VStack(alignment: .leading, spacing: 2) {
+                Text(ImportService.batchName(fileCount: batch.files.count))
+                    .font(.headline)
+                Text("Import the rows of each file into its own table")
+                    .font(.subheadline)
+                    .foregroundStyle(.secondary)
+            }
+            Spacer()
+            if batch.isLoadingTables || batch.isReading {
+                ProgressView().controlSize(.small)
+            }
+        }
+    }
+
+    // MARK: - Files
+
+    private var fileTable: some View {
+        Table(batch.files) {
+            TableColumn("") { file in
+                Toggle(file.sourceName, isOn: includedBinding(file.id))
+                    .labelsHidden()
+                    .accessibilityLabel(Text(String(format: String(localized: "Import %@"), file.sourceName)))
+            }
+            .width(20)
+            TableColumn("File") { file in
+                Text(file.sourceName)
+                    .lineLimit(1)
+                    .truncationMode(.middle)
+                    .help(file.url.path(percentEncoded: false))
+                    .foregroundStyle(file.isIncluded ? .primary : .secondary)
+            }
+            .width(min: 120, ideal: 190)
+            TableColumn("Format") { file in
+                Text(batch.formatName(file.formatId))
+                    .foregroundStyle(.secondary)
+            }
+            .width(min: 50, ideal: 60)
+            TableColumn("Destination") { file in
+                destinationPicker(file)
+                    .disabled(!file.isIncluded)
+            }
+            .width(min: 170, ideal: 230)
+            TableColumn("Columns") { file in
+                columnsCell(file)
+                    .disabled(!file.isIncluded)
+            }
+            .width(min: 170, ideal: 230)
+        }
+    }
+
+    private func destinationPicker(_ file: RowImportBatch.File) -> some View {
+        Picker(
+            String(format: String(localized: "Destination for %@"), file.sourceName),
+            selection: choiceBinding(file.id)
+        ) {
+            Text(newTableLabel(file)).tag(RowImportBatch.Choice.newTable)
+            Divider()
+            ForEach(tableChoices(for: file), id: \.self) { table in
+                Text(table).tag(RowImportBatch.Choice.existingTable(table))
+            }
+        }
+        .labelsHidden()
+    }
+
+    private func newTableLabel(_ file: RowImportBatch.File) -> String {
+        let name = file.target.newTableName.trimmingCharacters(in: .whitespaces)
+        guard !name.isEmpty else { return String(localized: "New table") }
+        return String(format: String(localized: "New table %@"), name)
+    }
+
+    /// The table a file already points at stays listed when the table list could not be read, or the menu would
+    /// draw blank.
+    private func tableChoices(for file: RowImportBatch.File) -> [String] {
+        var names = batch.availableTables.map(\.name)
+        if let current = file.target.existingTable, !names.contains(current) {
+            names.insert(current, at: 0)
+        }
+        return names
+    }
+
+    private func columnsCell(_ file: RowImportBatch.File) -> some View {
+        HStack(spacing: 6) {
+            if let problem = batch.problem(of: file), file.isIncluded {
+                Image(systemName: "exclamationmark.triangle.fill")
+                    .foregroundStyle(.yellow)
+                    .help(problem)
+                    .accessibilityLabel(Text(problem))
+            }
+            Text(batch.summary(of: file))
+                .foregroundStyle(.secondary)
+                .lineLimit(1)
+            Spacer(minLength: 4)
+            if case .failed = file.read {
+                Button(String(localized: "Try Again")) { batch.retryRead(of: file.id) }
+                    .buttonStyle(.link)
+            } else {
+                Button(String(localized: "Edit…")) { editing = EditRequest(id: file.id) }
+                    .buttonStyle(.link)
+                    .disabled(batch.scope == nil)
+                    .accessibilityLabel(Text(String(format: String(localized: "Edit the import of %@"), file.sourceName)))
+            }
+        }
+    }
+
+    @ViewBuilder
+    private var notes: some View {
+        VStack(alignment: .leading, spacing: 4) {
+            if !batch.leftOut.isEmpty {
+                Label(
+                    String(
+                        format: String(localized: "Left out, because they do not import into a table: %@"),
+                        batch.leftOut.map(\.lastPathComponent).formatted(.list(type: .and))
+                    ),
+                    systemImage: "info.circle"
+                )
+                .lineLimit(2)
+            }
+            if let error = batch.tableListError {
+                HStack(spacing: 6) {
+                    Label(
+                        String(format: String(localized: "Could not read the table list. %@"), error),
+                        systemImage: "exclamationmark.triangle"
+                    )
+                    .lineLimit(2)
+                    Button(String(localized: "Try Again")) {
+                        Task { await batch.loadTables() }
+                    }
+                    .buttonStyle(.link)
+                    .disabled(batch.isLoadingTables)
+                }
+            }
+        }
+        .font(.caption)
+        .foregroundStyle(.secondary)
+        .frame(maxWidth: .infinity, alignment: .leading)
+    }
+
+    /// A dropped file or folder joins the run, read the same way as a folder chosen in the panel.
+    private func accept(_ providers: [NSItemProvider]) -> Bool {
+        let lookup = ImportFormatLookup.live
+        let options = lookup.offeredFormats(connection.type)
+        let rowFormats = Set(options.map(\.id).filter { lookup.requiresTargetTable($0) == true })
+        guard !rowFormats.isEmpty else { return false }
+        for provider in providers where provider.hasItemConformingToTypeIdentifier(UTType.fileURL.identifier) {
+            provider.loadItem(forTypeIdentifier: UTType.fileURL.identifier) { data, _ in
+                guard let data = data as? Data, let url = URL(dataRepresentation: data, relativeTo: nil) else { return }
+                Task { @MainActor in
+                    adopt(ImportBatchFiles.resolve([url], among: options) { rowFormats.contains($0) })
+                }
+            }
+        }
+        return true
+    }
+
+    private func adopt(_ resolution: ImportBatchFiles.Resolution) {
+        let newFormats = Set(resolution.files.map(\.formatId)).subtracting(batch.formatIds)
+        batch.add(resolution)
+        for formatId in newFormats {
+            if let settable = plugin(for: formatId) as? any SettablePluginDiscoverable {
+                settingsSnapshot?.include(settable)
+            }
+            batch.setSignature(plugin(for: formatId)?.fieldDetectionSignature ?? "", for: formatId)
+        }
+    }
+
+    @ViewBuilder
+    private func editSheet(for id: RowImportBatch.File.ID) -> some View {
+        if let file = batch.file(id), let edit = batch.edit(for: id, onDone: { batch.apply($0, to: id) }) {
+            RowImportSheet(
+                isPresented: Binding(
+                    get: { editing?.id == id },
+                    set: { if !$0 { editing = nil } }
+                ),
+                connection: connection,
+                fileURL: file.url,
+                sourceName: file.sourceName,
+                formatId: file.formatId,
+                edit: edit
+            )
+        }
+    }
+
+    // MARK: - Options
+
+    private var currentFormat: String? {
+        let formats = batch.formatIds
+        if let shownFormat, formats.contains(shownFormat) { return shownFormat }
+        return formats.first
+    }
+
+    private var optionsForm: some View {
+        VStack(alignment: .leading, spacing: 8) {
+            HStack(spacing: 12) {
+                Text("Options").font(.callout.weight(.semibold))
+                if batch.formatIds.count > 1 {
+                    Picker(String(localized: "Options for"), selection: formatBinding) {
+                        ForEach(batch.formatIds, id: \.self) { formatId in
+                            Text(batch.formatName(formatId)).tag(formatId)
+                        }
+                    }
+                    .pickerStyle(.segmented)
+                    .labelsHidden()
+                    .fixedSize()
+                }
+            }
+            if let formatId = currentFormat,
+               let settable = plugin(for: formatId) as? any SettablePluginDiscoverable,
+               let optionsView = settable.settingsView() {
+                optionsView
+            }
+        }
+        .frame(maxWidth: .infinity, alignment: .leading)
+    }
+
+    @ViewBuilder
+    private var signatureObservers: some View {
+        ForEach(batch.formatIds, id: \.self) { formatId in
+            if let observer = RowImportOptionsObservation.observer(
+                for: plugin(for: formatId),
+                onChange: { signature, configuration in
+                    batch.setSignature(signature, for: formatId)
+                    batch.setConfiguration(configuration, for: formatId)
+                }
+            ) {
+                observer
+            }
+        }
+    }
+
+    private func plugin(for formatId: String) -> (any ImportFormatPlugin)? {
+        pluginManager.importPlugin(forFormat: formatId)
+    }
+
+    // MARK: - Footer
+
+    private var footerView: some View {
+        DialogFooter {
+            if let message = batch.problem, !batch.isReading || batch.includedFiles.isEmpty {
+                Text(message)
+                    .font(.caption)
+                    .foregroundStyle(.red)
+                    .lineLimit(2)
+            } else if batch.isReading {
+                Text("Reading the files…")
+                    .font(.caption)
+                    .foregroundStyle(.secondary)
+            }
+        } actions: {
+            Button("Cancel") { isPresented = false }
+                .keyboardShortcut(.cancelAction)
+            Button("Import") { performImport() }
+                .buttonStyle(.borderedProminent)
+                .disabled(!batch.canImport || importService?.state.isImporting == true)
+                .keyboardShortcut(.defaultAction)
+        }
+    }
+
+    // MARK: - Bindings
+
+    private func includedBinding(_ id: RowImportBatch.File.ID) -> Binding<Bool> {
+        Binding(
+            get: { batch.file(id)?.isIncluded ?? false },
+            set: { batch.setIncluded($0, for: id) }
+        )
+    }
+
+    private func choiceBinding(_ id: RowImportBatch.File.ID) -> Binding<RowImportBatch.Choice> {
+        Binding(
+            get: { batch.file(id).map(batch.choice(for:)) ?? .newTable },
+            set: { batch.choose($0, for: id) }
+        )
+    }
+
+    private var formatBinding: Binding<String> {
+        Binding(
+            get: { currentFormat ?? "" },
+            set: { shownFormat = $0 }
+        )
+    }
+
+    // MARK: - Import
+
+    private func performImport() {
+        guard let scope = batch.scope else {
+            TransferResultAlert.presentImportFailure(error: DatabaseError.notConnected, window: hostWindow) {}
+            return
+        }
+        guard let plan = batch.plan(), let policy = batch.policy else { return }
+        let service = ImportService(connection: connection)
+        importService = service
+        showProgressDialog = true
+
+        importTask = Task {
+            do {
+                var authorization: OperationReceipt?
+                if !plan.newTables.isEmpty {
+                    authorization = try await RowImportTableMaker(connection: connection).prepare(
+                        plan.newTables,
+                        scope: scope,
+                        planner: batch.planner,
+                        description: String(
+                            format: String(localized: "Import %@"),
+                            ImportService.batchName(fileCount: plan.items.count)
+                        )
+                    ) { table, sql in
+                        batch.planner.recordCreated(table, createTableSQL: sql)
+                    }
+                }
+                for mapping in plan.mappings {
+                    RowImportMapping.remember(
+                        fields: mapping.fields,
+                        columns: mapping.columns,
+                        columnMapping: mapping.columnMapping,
+                        in: TableScope(table: mapping.table, in: scope)
+                    )
+                }
+                let report = try await service.importFiles(
+                    plan.items,
+                    scope: scope,
+                    policy: policy,
+                    authorization: authorization
+                )
+                showProgressDialog = false
+                present(report, scope: scope)
+            } catch ExecutionGateError.cancelledByUser {
+                showProgressDialog = false
+            } catch is CancellationError {
+                showProgressDialog = false
+            } catch {
+                showProgressDialog = false
+                TransferResultAlert.presentImportFailure(error: error, window: hostWindow) {}
+            }
+        }
+    }
+
+    /// A run that finished closes the sheet. One that stopped stays open, so a changed setting can run it again.
+    private func present(_ report: RowImportBatchReport, scope: DatabaseScope) {
+        let completed = report.ending == .completed
+        if completed {
+            importSucceeded = true
+        }
+        TransferResultAlert.presentBatchImportResult(report, window: hostWindow) {
+            if report.rowsKept {
+                for table in Set(report.files.map(\.targetTable)) {
+                    AppCommands.shared.objectChanged.send(
+                        DatabaseObjectChange(connectionId: connection.id, scope: scope, name: table, kind: .rows)
+                    )
+                }
+            }
+            if completed {
+                isPresented = false
+            }
+        }
+    }
+}
+
+/// The options a format's own view edits, followed for every format of a run: the signature decides the reads, and
+/// the error handling and transaction options decide whether the files can run together.
+private struct RowImportOptionsObserver<Plugin: ImportFormatPlugin & ObservableObject>: View {
+    @ObservedObject var plugin: Plugin
+    let onChange: (String, RowImportRunner.Configuration?) -> Void
+
+    private var key: String {
+        let configuration = plugin.rowImportConfiguration
+        return [
+            plugin.fieldDetectionSignature,
+            configuration?.errorHandling.rawValue ?? "",
+            configuration.map { String($0.wrapInTransaction) } ?? "",
+            configuration.map { String($0.deleteExistingRows) } ?? "",
+        ].joined(separator: "|")
+    }
+
+    var body: some View {
+        Color.clear
+            .onAppear { report() }
+            .onChange(of: key) { _ in report() }
+    }
+
+    private func report() {
+        onChange(plugin.fieldDetectionSignature, plugin.rowImportConfiguration)
+    }
+}
+
+private enum RowImportOptionsObservation {
+    @MainActor
+    static func observer(
+        for plugin: (any ImportFormatPlugin)?,
+        onChange: @escaping (String, RowImportRunner.Configuration?) -> Void
+    ) -> AnyView? {
+        guard let observable = plugin as? any ImportFormatPlugin & ObservableObject else { return nil }
+        return makeObserver(for: observable, onChange: onChange)
+    }
+
+    @MainActor
+    private static func makeObserver<Plugin: ImportFormatPlugin & ObservableObject>(
+        for plugin: Plugin,
+        onChange: @escaping (String, RowImportRunner.Configuration?) -> Void
+    ) -> AnyView {
+        AnyView(RowImportOptionsObserver(plugin: plugin, onChange: onChange))
+    }
+}
diff --git a/apple/TablePro/Views/Import/RowImportMapping.swift b/apple/TablePro/Views/Import/RowImportMapping.swift
index 9ea8630370..50f45334db 100644
--- a/apple/TablePro/Views/Import/RowImportMapping.swift
+++ b/apple/TablePro/Views/Import/RowImportMapping.swift
@@ -31,8 +31,16 @@ internal final class RowImportMapping: ObservableObject {
 
     private let store: ImportColumnMappingStore
 
-    internal init(store: ImportColumnMappingStore = .shared) {
+    /// Choices made for this table before the sheet opened, by a run of several files, put ahead of the saved ones
+    /// on the first read of that table.
+    private var seed: (table: TableScope, choices: [String: ImportFieldChoice])?
+
+    internal init(
+        store: ImportColumnMappingStore = .shared,
+        seed: (table: TableScope, choices: [String: ImportFieldChoice])? = nil
+    ) {
         self.store = store
+        self.seed = seed
     }
 
     internal var fields: [String] {
@@ -91,6 +99,9 @@ internal final class RowImportMapping: ObservableObject {
         if reloading {
             let session = Dictionary(rows.map { ($0.field.name, $0.choice) }, uniquingKeysWith: { first, _ in first })
             tiers.insert(session, at: 0)
+        } else if let seed, seed.table == scope {
+            tiers.insert(seed.choices, at: 0)
+            self.seed = nil
         }
         table = scope
         saved = remembered
@@ -129,15 +140,25 @@ internal final class RowImportMapping: ObservableObject {
         columns: [String],
         columnMapping: [String: String],
         in scope: TableScope
+    ) {
+        Self.remember(fields: fields, columns: columns, columnMapping: columnMapping, in: scope, store: store)
+        if table == scope {
+            saved = store.overrides(for: scope)
+        }
+    }
+
+    internal static func remember(
+        fields: [String],
+        columns: [String],
+        columnMapping: [String: String],
+        in scope: TableScope,
+        store: ImportColumnMappingStore = .shared
     ) {
         let choices = fields.map { field in
             columnMapping[field].map { ImportFieldChoice(include: true, column: $0) } ?? .skipped
         }
         let overrides = ImportColumnMatcher.overrides(fields: fields, columns: columns, choices: choices)
         store.remember(overrides, forFields: fields, in: scope)
-        if table == scope {
-            saved = store.overrides(for: scope)
-        }
     }
 
     private var appliesSavedChoice: Bool {
diff --git a/apple/TablePro/Views/Import/RowImportSheet.swift b/apple/TablePro/Views/Import/RowImportSheet.swift
index 881591ffae..6d3f3c0a57 100644
--- a/apple/TablePro/Views/Import/RowImportSheet.swift
+++ b/apple/TablePro/Views/Import/RowImportSheet.swift
@@ -27,10 +27,10 @@ struct RowImportSheet: View {
     let sourceName: String
     let formatId: String
 
-    private enum Destination: Hashable {
-        case existingTable
-        case newTable
-    }
+    /// Set when the sheet edits one file of several: it hands its choices back on Done and imports nothing.
+    let edit: RowImportSheetEdit?
+
+    private typealias Destination = RowImportDestination
 
     /// The identity of a read. `.task(id:)` cancels the read in flight when it changes.
     private struct SourceRead: Hashable {
@@ -127,14 +127,30 @@ struct RowImportSheet: View {
         connection: DatabaseConnection,
         fileURL: URL,
         sourceName: String,
-        formatId: String
+        formatId: String,
+        edit: RowImportSheetEdit? = nil
     ) {
         _isPresented = isPresented
         self.connection = connection
         self.fileURL = fileURL
         self.sourceName = sourceName
         self.formatId = formatId
-        _scope = State(initialValue: DatabaseManager.shared.browseScope(for: connection.id))
+        self.edit = edit
+        guard let edit else {
+            _scope = State(initialValue: DatabaseManager.shared.browseScope(for: connection.id))
+            return
+        }
+        let target = edit.target
+        _scope = State(initialValue: edit.scope)
+        _destination = State(initialValue: target.destination)
+        _selectedTargetTable = State(initialValue: target.existingTable)
+        _newTableName = State(initialValue: target.newTableName)
+        _proposedTableName = State(initialValue: target.proposedTableName)
+        _newTable = State(initialValue: target.newTable)
+        _newTablePlanner = State(initialValue: edit.planner)
+        _mapping = StateObject(wrappedValue: RowImportMapping(seed: target.mapping.map { mapping in
+            (TableScope(table: mapping.table, in: edit.scope), mapping.choicesByField)
+        }))
     }
 
     // MARK: - Derived catalog state
@@ -151,6 +167,9 @@ struct RowImportSheet: View {
     /// planner rather than folded into `catalogNameKeys`, which is rebuilt only when the catalog is.
     private var newTableNameProblem: NewTableNameProblem? {
         let trimmed = newTableName.trimmingCharacters(in: .whitespaces)
+        if let edit, !trimmed.isEmpty, edit.takenTableNames.contains(trimmed.lowercased()) {
+            return .nameTaken
+        }
         let problem = NewTableNaming.problem(
             with: newTableName,
             style: NewTableNameStyle.forDatabaseType(connection.type),
@@ -177,10 +196,12 @@ struct RowImportSheet: View {
                 .frame(maxWidth: .infinity, maxHeight: .infinity)
             Divider()
 
-            optionsForm
-                .padding(.horizontal)
-                .padding(.vertical, 10)
-            Divider()
+            if edit == nil {
+                optionsForm
+                    .padding(.horizontal)
+                    .padding(.vertical, 10)
+                Divider()
+            }
 
             footerView
                 .padding()
@@ -193,8 +214,10 @@ struct RowImportSheet: View {
         }
         .background { detectionSignatureObserver }
         .task {
-            settingsSnapshot = PluginSettingsSnapshot(
-                plugins: [currentPlugin as? any SettablePluginDiscoverable].compactMap { $0 })
+            if edit == nil {
+                settingsSnapshot = PluginSettingsSnapshot(
+                    plugins: [currentPlugin as? any SettablePluginDiscoverable].compactMap { $0 })
+            }
             suggestNewTableName()
             await loadTables()
         }
@@ -358,7 +381,7 @@ struct RowImportSheet: View {
         } actions: {
             Button("Cancel") { isPresented = false }
                 .keyboardShortcut(.cancelAction)
-            Button("Import") { performImport() }
+            Button(edit == nil ? String(localized: "Import") : String(localized: "Done")) { performImport() }
                 .buttonStyle(.borderedProminent)
                 .disabled(!canImport)
                 .keyboardShortcut(.defaultAction)
@@ -637,34 +660,9 @@ struct RowImportSheet: View {
             return nil
         case .newTable:
             if let problem = newTableNameProblem {
-                switch problem {
-                case .blank:
-                    return String(localized: "Enter a name for the new table.")
-                case .nameTaken:
-                    return String(
-                        format: String(localized: "A table named %@ already exists."),
-                        newTableName.trimmingCharacters(in: .whitespaces)
-                    )
-                case .reservedPrefix(let prefix):
-                    return String(
-                        format: String(localized: "This database keeps names beginning with %@ for itself."),
-                        prefix
-                    )
-                case .tooLong(let maximumBytes):
-                    return String(
-                        format: String(localized: "This database allows at most %lld bytes in a table name."),
-                        Int64(maximumBytes)
-                    )
-                }
-            }
-            switch newTable.problem {
-            case .unnamedColumn:
-                return String(localized: "Every included column needs a name.")
-            case .duplicateName:
-                return String(localized: "Column names must be unique.")
-            case nil:
-                return nil
+                return problem.message(for: newTableName)
             }
+            return newTable.problem?.message
         }
     }
 
@@ -782,7 +780,7 @@ struct RowImportSheet: View {
         let suggestion = NewTableNaming.suggestion(
             forFileNamed: sourceName,
             style: NewTableNameStyle.forDatabaseType(connection.type),
-            avoiding: (catalogNameKeys ?? []).subtracting(ours)
+            avoiding: (catalogNameKeys ?? []).subtracting(ours).union(edit?.takenTableNames ?? [])
         )
         proposedTableName = suggestion
         newTableName = suggestion
@@ -879,6 +877,10 @@ struct RowImportSheet: View {
     // MARK: - Import
 
     private func performImport() {
+        if let edit {
+            finishEdit(edit)
+            return
+        }
         guard let scope else {
             importError = DatabaseError.notConnected
             showErrorDialog = true
@@ -906,7 +908,7 @@ struct RowImportSheet: View {
         case .newTable:
             let name = newTableName.trimmingCharacters(in: .whitespaces)
             guard !name.isEmpty, let definition = newTable.definition(tableName: name) else {
-                importError = Self.createTableStatementError
+                importError = RowImportTableMaker.statementError
                 showErrorDialog = true
                 return
             }
@@ -925,11 +927,28 @@ struct RowImportSheet: View {
         }
     }
 
-    private static var createTableStatementError: NSError {
-        NSError(
-            domain: "RowImport", code: -1,
-            userInfo: [NSLocalizedDescriptionKey: String(localized: "Could not build the CREATE TABLE statement")]
+    /// Hands the choices back only when they answer a read of the current table and options, the same readiness
+    /// Import asks for.
+    private func finishEdit(_ edit: RowImportSheetEdit) {
+        guard currentReadIsReady else { return }
+        var target = RowImportTarget(
+            destination: destination,
+            existingTable: selectedTargetTable,
+            newTableName: newTableName,
+            proposedTableName: proposedTableName,
+            mapping: nil,
+            newTable: newTable
         )
+        if destination == .existingTable, let table = selectedTargetTable {
+            target.mapping = RowImportTarget.Mapping(
+                table: table,
+                fields: mapping.rows.map(\.field),
+                columns: mapping.columns,
+                choices: mapping.choices
+            )
+        }
+        edit.onDone(target)
+        isPresented = false
     }
 
     private func runImport(_ plan: ImportPlan, scope: DatabaseScope) {
@@ -942,7 +961,13 @@ struct RowImportSheet: View {
             do {
                 var authorization: OperationReceipt?
                 if let newTable = plan.newTable {
-                    authorization = try await prepareTable(newTable, scope: scope)
+                    authorization = try await RowImportTableMaker(connection: connection).prepare(
+                        [newTable],
+                        scope: scope,
+                        planner: newTablePlanner
+                    ) { table, sql in
+                        newTablePlanner.recordCreated(table, createTableSQL: sql)
+                    }
                 }
                 mapping.remember(
                     fields: plan.fields,
@@ -993,111 +1018,4 @@ struct RowImportSheet: View {
             }
         }
     }
-
-    /// Returns the gate's grant for the table's own statements, which the rows that follow reuse.
-    @MainActor
-    private func prepareTable(
-        _ definition: PluginCreateTableDefinition,
-        scope: DatabaseScope
-    ) async throws -> OperationReceipt {
-        guard ImportDataSinkAdapter.canWriteRows(into: connection.type) else {
-            throw PluginImportError.importFailed(String(
-                format: String(localized: "%@ cannot take rows from this file, so no table was created."),
-                connection.type.rawValue
-            ))
-        }
-        let tableName = definition.tableName
-        let table = TableScope(table: tableName, in: scope)
-        let statements = try await DatabaseManager.shared.createTableStatements(
-            definition: definition,
-            scope: scope,
-            route: DatabaseManager.shared.executionRoute(for: scope)
-        )
-        guard !statements.isEmpty else { throw Self.createTableStatementError }
-        let sql = statements.joined(separator: "\n")
-        switch newTablePlanner.plan(forTable: table, createTableSQL: sql) {
-        case .create:
-            let receipt = try await createTable(statements: statements, scope: scope)
-            newTablePlanner.recordCreated(table, createTableSQL: sql)
-            return receipt
-        case .reuseAfterClearing:
-            return try await clearRows(of: tableName, scope: scope)
-        case .nameTakenWithDifferentColumns:
-            throw PluginImportError.importFailed(
-                String(
-                    format: String(localized: "The table %@ was already created with different columns. Choose another name."),
-                    tableName
-                )
-            )
-        }
-    }
-
-    @MainActor
-    private func clearRows(of tableName: String, scope: DatabaseScope) async throws -> OperationReceipt {
-        let generator = try SQLStatementGenerator(
-            tableName: tableName,
-            columns: [],
-            primaryKeyColumns: [],
-            databaseType: connection.type
-        )
-        let sql = generator.deleteAllRowsStatement()
-        let receipt = try await authorize(
-            sql: sql, kind: .destructiveQuery, description: String(localized: "Clear Table")
-        )
-        try await runOnLeasedDriver(sql, scope: scope)
-        return receipt
-    }
-
-    /// One call per statement the driver wrote, because an engine that runs one statement per call refuses a table
-    /// and its indexes sent together.
-    private func createTable(statements: [String], scope: DatabaseScope) async throws -> OperationReceipt {
-        let script = SQLScriptText(databaseType: connection.type).script(statements)
-        let receipt = try await authorize(
-            sql: script, kind: .schemaMutation, description: String(localized: "Create Table")
-        )
-        for statement in statements {
-            try await runOnLeasedDriver(statement, scope: scope)
-        }
-        CatalogChangeService.post(
-            .changed(CatalogChange(connectionId: connection.id, database: scope.database, kinds: .tables))
-        )
-        return receipt
-    }
-
-    /// The sheet's own statements take the same lease the import does, one at a time and always
-    /// after `authorize` has returned. Taking it earlier would hold the connection's gate open
-    /// across a safe-mode confirmation the user has not answered yet, and the gate is not
-    /// reentrant, so the import that follows would then wait on a sheet waiting on the user.
-    @MainActor
-    private func runOnLeasedDriver(_ sql: String, scope: DatabaseScope) async throws {
-        let route = DatabaseManager.shared.executionRoute(for: scope)
-        _ = try await DatabaseManager.shared.withScopedDriver(
-            scope: scope,
-            route: route,
-            cancellation: .protectedWrite
-        ) { driver in
-            try await driver.execute(query: sql)
-        }
-    }
-
-    /// Every statement this sheet issues on its own account goes through the gate. The retry path's
-    /// `DELETE FROM` used to skip it while the `CREATE TABLE` beside it did not, so a connection
-    /// set to confirm destructive statements emptied a table without asking.
-    private func authorize(sql: String, kind: OperationKind, description: String) async throws -> OperationReceipt {
-        let decision = await ExecutionGateProvider.shared.authorize(
-            OperationRequest(
-                connectionId: connection.id,
-                databaseType: connection.type,
-                sql: sql,
-                kind: kind,
-                caller: .userInterface,
-                capabilities: .interactiveUser,
-                operationDescription: description
-            )
-        )
-        guard case .authorized(let receipt) = decision else {
-            throw decision.denialError ?? ExecutionGateError.denied(String(localized: "Operation not permitted"))
-        }
-        return receipt
-    }
 }
diff --git a/apple/TablePro/Views/Import/RowImportTarget.swift b/apple/TablePro/Views/Import/RowImportTarget.swift
new file mode 100644
index 0000000000..f6220239ff
--- /dev/null
+++ b/apple/TablePro/Views/Import/RowImportTarget.swift
@@ -0,0 +1,138 @@
+//
+//  RowImportTarget.swift
+//  TablePro
+//
+
+import Foundation
+import TableProPluginKit
+
+internal enum RowImportDestination: Hashable {
+    case existingTable
+    case newTable
+}
+
+/// Where one file's rows go and how its fields reach the columns, as a run of several files holds it for each file
+/// and as `RowImportSheet` edits it for one of them.
+internal struct RowImportTarget {
+    /// The fields of the file against the columns of an existing table, as one read found them.
+    internal struct Mapping {
+        internal let table: String
+        internal let fields: [PluginImportField]
+        internal let columns: [String]
+        internal var choices: [ImportFieldChoice]
+
+        internal var fieldNames: [String] {
+            fields.map(\.name)
+        }
+
+        internal var columnMapping: [String: String] {
+            var mapping: [String: String] = [:]
+            for (field, choice) in zip(fields, choices) {
+                guard let column = choice.mappedColumn else { continue }
+                mapping[field.name] = column
+            }
+            return mapping
+        }
+
+        internal var mappedFieldCount: Int {
+            choices.filter { $0.mappedColumn != nil }.count
+        }
+
+        internal var mapsOneColumnTwice: Bool {
+            let targets = choices.compactMap { $0.mappedColumn?.lowercased() }
+            return Set(targets).count != targets.count
+        }
+
+        internal var choicesByField: [String: ImportFieldChoice] {
+            Dictionary(zip(fieldNames, choices), uniquingKeysWith: { first, _ in first })
+        }
+    }
+
+    internal var destination: RowImportDestination
+    internal var existingTable: String?
+    internal var newTableName: String
+
+    /// The last name proposed for the new table, so a later proposal can tell its own guess from a typed name.
+    internal var proposedTableName: String
+    internal var mapping: Mapping?
+    internal var newTable: NewTableDraft
+
+    internal static func newTable(named name: String) -> RowImportTarget {
+        RowImportTarget(
+            destination: .newTable,
+            existingTable: nil,
+            newTableName: name,
+            proposedTableName: name,
+            mapping: nil,
+            newTable: NewTableDraft()
+        )
+    }
+
+    internal static func existingTable(_ table: String, proposingNewTable name: String) -> RowImportTarget {
+        RowImportTarget(
+            destination: .existingTable,
+            existingTable: table,
+            newTableName: name,
+            proposedTableName: name,
+            mapping: nil,
+            newTable: NewTableDraft()
+        )
+    }
+
+    /// The table the rows go to, trimmed the way the import names it.
+    internal var tableName: String? {
+        switch destination {
+        case .existingTable:
+            return existingTable
+        case .newTable:
+            let trimmed = newTableName.trimmingCharacters(in: .whitespaces)
+            return trimmed.isEmpty ? nil : trimmed
+        }
+    }
+}
+
+/// What `RowImportSheet` needs to edit one file of several instead of importing it.
+internal struct RowImportSheetEdit {
+    internal let scope: DatabaseScope
+    internal let target: RowImportTarget
+    internal let planner: NewTableImportPlanner
+
+    /// Folded names the other files of the run create, which this file's new table must not take.
+    internal let takenTableNames: Set<String>
+    internal let onDone: @MainActor (RowImportTarget) -> Void
+}
+
+internal extension NewTableNameProblem {
+    func message(for name: String) -> String {
+        switch self {
+        case .blank:
+            return String(localized: "Enter a name for the new table.")
+        case .nameTaken:
+            return String(
+                format: String(localized: "A table named %@ already exists."),
+                name.trimmingCharacters(in: .whitespaces)
+            )
+        case .reservedPrefix(let prefix):
+            return String(
+                format: String(localized: "This database keeps names beginning with %@ for itself."),
+                prefix
+            )
+        case .tooLong(let maximumBytes):
+            return String(
+                format: String(localized: "This database allows at most %lld bytes in a table name."),
+                Int64(maximumBytes)
+            )
+        }
+    }
+}
+
+internal extension NewTableColumnProblem {
+    var message: String {
+        switch self {
+        case .unnamedColumn:
+            return String(localized: "Every included column needs a name.")
+        case .duplicateName:
+            return String(localized: "Column names must be unique.")
+        }
+    }
+}
diff --git a/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+RowImport.swift b/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+RowImport.swift
index 8dd9321536..abfc0a5604 100644
--- a/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+RowImport.swift
+++ b/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+RowImport.swift
@@ -50,6 +50,30 @@ internal extension MainContentCoordinator {
         return nil
     }
 
+    /// A folder holding a single file a row format reads opens the sheet for one file. Files no row format reads
+    /// are named in the batch sheet, or in an alert when nothing else is left.
+    func presentRowImportBatch(_ resolution: ImportBatchFiles.Resolution) {
+        guard let first = resolution.files.first else {
+            presentError(
+                String(localized: "Nothing to Import"),
+                String(
+                    format: String(localized: "None of these files import into a table: %@."),
+                    resolution.leftOut.map(\.lastPathComponent).formatted(.list(type: .and))
+                ),
+                workspaceWindow
+            )
+            return
+        }
+        releaseImportFile()
+        if resolution.files.count == 1, resolution.leftOut.isEmpty {
+            importFile = ImportFileHandoff(url: first.url, ownsFile: false)
+            activeSheet = .rowImport(formatId: first.formatId)
+            return
+        }
+        importBatch = resolution
+        activeSheet = .rowImportBatch
+    }
+
     func reportImportRefusal(_ refusal: RowImportRefusal) {
         guard case .importNotSupported = refusal else { return }
         presentError(
diff --git a/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+SidebarActions.swift b/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+SidebarActions.swift
index 40bcac00ce..af7fb21f22 100644
--- a/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+SidebarActions.swift
+++ b/apple/TablePro/Views/Main/Extensions/MainContentCoordinator+SidebarActions.swift
@@ -233,18 +233,26 @@ extension MainContentCoordinator {
     /// The menu bar's Import Data…, the Actions pull-down's leaf of the same name, and ⇧⌘I. The
     /// file decides the format, so every format this connection imports is offered and the sheet
     /// follows what the user picked. Resolving a format first is what made the command a permanent
-    /// alias for SQL (#3047).
+    /// alias for SQL (#3047). Several files, or a folder, import as rows into a table each.
     func openImportPanel() {
         guard let options = offeredImportFormats() else { return }
+        let lookup = importFormatLookup
+        let readsRows: (String) -> Bool = { lookup.requiresTargetTable($0) == true }
         Task { @MainActor [weak self] in
             guard let self else { return }
-            let url = await ImportFilePanel.present(
+            let urls = await ImportFilePanel.presentFiles(
                 matching: options,
-                message: String(localized: "Select a file to import"),
+                readsRows: readsRows,
+                message: String(localized: "Select files or a folder to import"),
                 in: self.contentWindow
             )
-            guard let url, case .format(let formatId) = ImportFileFormatResolver.match(url, among: options) else { return }
-            self.presentImportSheet(fileURL: url, formatId: formatId)
+            if urls.count == 1, let url = urls.first, !ImportBatchFiles.isDirectory(url) {
+                guard case .format(let formatId) = ImportFileFormatResolver.match(url, among: options) else { return }
+                self.presentImportSheet(fileURL: url, formatId: formatId)
+                return
+            }
+            guard !urls.isEmpty else { return }
+            self.presentRowImportBatch(ImportBatchFiles.resolve(urls, among: options, readsRows: readsRows))
         }
     }
 
diff --git a/apple/TablePro/Views/Main/Extensions/MainContentView+TransferSheets.swift b/apple/TablePro/Views/Main/Extensions/MainContentView+TransferSheets.swift
index 438e264f9d..797759eba1 100644
--- a/apple/TablePro/Views/Main/Extensions/MainContentView+TransferSheets.swift
+++ b/apple/TablePro/Views/Main/Extensions/MainContentView+TransferSheets.swift
@@ -67,6 +67,15 @@ extension MainContentView {
                     formatId: formatId
                 )
             }
+        case .rowImportBatch:
+            if let batch = coordinator.importBatch {
+                RowImportBatchSheet(
+                    isPresented: dismissBinding,
+                    connection: connection,
+                    files: batch.files,
+                    leftOut: batch.leftOut
+                )
+            }
         case .transferTables(let tables, let schema):
             transferSheet(tables: tables, schema: schema, dismiss: dismissBinding)
         case .backupDatabase(let databases):
diff --git a/apple/TablePro/Views/Main/MainContentCoordinator.swift b/apple/TablePro/Views/Main/MainContentCoordinator.swift
index 4cfe29f328..752a9447eb 100644
--- a/apple/TablePro/Views/Main/MainContentCoordinator.swift
+++ b/apple/TablePro/Views/Main/MainContentCoordinator.swift
@@ -53,6 +53,8 @@ enum ActiveSheet: Identifiable {
     case exportDialog
     case importDialog(formatId: String)
     case rowImport(formatId: String)
+    /// Several files, or a folder, each imported as rows into its own table. The files are in `importBatch`.
+    case rowImportBatch
     case exportQueryResults
     /// The tables the user right-clicked travel with the request, because the object browser may be
     /// pointed somewhere else by the time the sheet appears.
@@ -99,6 +101,7 @@ enum ActiveSheet: Identifiable {
         case .exportDialog: "exportDialog"
         case .importDialog(let formatId): "importDialog-\(formatId)"
         case .rowImport(let formatId): "rowImport-\(formatId)"
+        case .rowImportBatch: "rowImportBatch"
         case .exportQueryResults: "exportQueryResults"
         case .transferTables(let tables, let schema):
             "transferTables-\(schema ?? "")-\(tables.sorted().joined(separator: ","))"
@@ -310,6 +313,7 @@ final class MainContentCoordinator: ObservableObject {
     @Published var sessionContexts: [PluginSessionContext] = []
     @Published var containerDropRequest: DatabaseDropRequest?
     @Published var importFile: ImportFileHandoff?
+    @Published var importBatch: ImportBatchFiles.Resolution?
     @Published var exportPreselection: ExportPreselection?
     @Published var exportFormatId: String?
     // Kept after the sheet closes, so reopening Search Database after opening a result shows the results again.
diff --git a/apple/TablePro/Views/Main/MainContentView.swift b/apple/TablePro/Views/Main/MainContentView.swift
index d110403180..019269ae9c 100644
--- a/apple/TablePro/Views/Main/MainContentView.swift
+++ b/apple/TablePro/Views/Main/MainContentView.swift
@@ -243,7 +243,7 @@ struct MainContentView: View {
             DatabaseSearchSheet(session: session) { result in
                 coordinator.openDatabaseSearchResult(result, in: session)
             }
-        case .exportDialog, .exportQueryResults, .importDialog, .rowImport,
+        case .exportDialog, .exportQueryResults, .importDialog, .rowImport, .rowImportBatch,
              .transferTables, .backupDatabase, .restoreDatabase:
             transferSheetContent(for: sheet, dismiss: dismissBinding)
         case .maintenance(let operation, let tableName, let database, let schema):
diff --git a/apple/TableProTests/Core/Plugins/ImportDataSinkAdapterRollbackTests.swift b/apple/TableProTests/Core/Plugins/ImportDataSinkAdapterRollbackTests.swift
index a83fde3f4b..cb8bf27edd 100644
--- a/apple/TableProTests/Core/Plugins/ImportDataSinkAdapterRollbackTests.swift
+++ b/apple/TableProTests/Core/Plugins/ImportDataSinkAdapterRollbackTests.swift
@@ -94,6 +94,24 @@ struct ImportDataSinkAdapterRollbackTests {
         }
     }
 
+    /// One file of several runs inside the run's transaction. On an engine that cannot report its session, a `BEGIN`
+    /// of the file's own would open a second one, and MySQL commits the run's on it.
+    @Test("A sink that joins a run's transaction sends no BEGIN, COMMIT or ROLLBACK")
+    func joinedSinkSendsNoTransactionStatements() async throws {
+        let driver = StatusFlagDriver(reportsState: false)
+        let connection = TestFixtures.makeConnection(type: .mysql)
+        let adapter = PluginDriverAdapter(connection: connection, pluginDriver: driver)
+        let sink = ImportDataSinkAdapter(driver: adapter, databaseType: .mysql, joinsTransaction: true)
+
+        try await sink.beginTransaction()
+        try await sink.execute(statement: "INSERT INTO courses VALUES (1)", line: 1)
+        try await sink.commitTransaction()
+        try await sink.rollbackTransaction()
+
+        #expect(driver.executed == ["INSERT INTO courses VALUES (1)"])
+        #expect(sink.transactionOutcome == .untracked)
+    }
+
     @Test("A transaction a DROP already committed is not rolled back, and the sink says so")
     func committedTransactionIsNotRolledBack() async throws {
         let driver = StatusFlagDriver()
diff --git a/apple/TableProTests/Core/Services/ImportBatchFilesTests.swift b/apple/TableProTests/Core/Services/ImportBatchFilesTests.swift
new file mode 100644
index 0000000000..b8c0a0b34f
--- /dev/null
+++ b/apple/TableProTests/Core/Services/ImportBatchFilesTests.swift
@@ -0,0 +1,71 @@
+//
+//  ImportBatchFilesTests.swift
+//  TableProTests
+//
+
+import Foundation
+@testable import TablePro
+import Testing
+
+/// A folder chosen in Import Data… gives the files directly inside it that a row format reads, in Finder's order,
+/// and names what it leaves out.
+struct ImportBatchFilesTests {
+    private let options = [
+        ImportFormatOption(id: "sql", name: "SQL", acceptedFileExtensions: ["sql", "gz"]),
+        ImportFormatOption(id: "csv", name: "CSV", acceptedFileExtensions: ["csv", "tsv"]),
+        ImportFormatOption(id: "json", name: "JSON", acceptedFileExtensions: ["json", "jsonl"]),
+    ]
+
+    private func readsRows(_ formatId: String) -> Bool {
+        formatId != "sql"
+    }
+
+    private func makeFolder(_ names: [String], subfolders: [String] = []) throws -> URL {
+        let folder = FileManager.default.temporaryDirectory.appendingPathComponent("batch-files-\(UUID().uuidString)")
+        try FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true)
+        for name in names {
+            try Data("id\n1\n".utf8).write(to: folder.appendingPathComponent(name))
+        }
+        for name in subfolders {
+            let sub = folder.appendingPathComponent(name)
+            try FileManager.default.createDirectory(at: sub, withIntermediateDirectories: true)
+            try Data("id\n1\n".utf8).write(to: sub.appendingPathComponent("nested.csv"))
+        }
+        return folder
+    }
+
+    @Test("A folder gives its row files in Finder's order, with the format each extension names")
+    func folderGivesRowFilesInOrder() throws {
+        let folder = try makeFolder(["orders10.csv", "orders2.csv", "people.json", "notes.tsv"])
+        defer { try? FileManager.default.removeItem(at: folder) }
+
+        let resolution = ImportBatchFiles.resolve([folder], among: options, readsRows: readsRows)
+
+        #expect(resolution.files.map(\.url.lastPathComponent) == ["notes.tsv", "orders2.csv", "orders10.csv", "people.json"])
+        #expect(resolution.files.map(\.formatId) == ["csv", "csv", "csv", "json"])
+        #expect(resolution.leftOut.isEmpty)
+    }
+
+    @Test("A dump and a file no format reads are left out by name, hidden files and subfolders are skipped")
+    func leavesOutWhatNoRowFormatReads() throws {
+        let folder = try makeFolder(["schema.sql", "readme.md", ".hidden.csv", "orders.csv"], subfolders: ["archive"])
+        defer { try? FileManager.default.removeItem(at: folder) }
+
+        let resolution = ImportBatchFiles.resolve([folder], among: options, readsRows: readsRows)
+
+        #expect(resolution.files.map(\.url.lastPathComponent) == ["orders.csv"])
+        #expect(resolution.leftOut.map(\.lastPathComponent) == ["readme.md", "schema.sql"])
+    }
+
+    @Test("Files chosen together keep their order, and a file named twice imports once")
+    func filesKeepTheirOrder() throws {
+        let folder = try makeFolder(["b.csv", "a.json"])
+        defer { try? FileManager.default.removeItem(at: folder) }
+        let first = folder.appendingPathComponent("b.csv")
+        let second = folder.appendingPathComponent("a.json")
+
+        let resolution = ImportBatchFiles.resolve([first, second, first], among: options, readsRows: readsRows)
+
+        #expect(resolution.files.map(\.url.lastPathComponent) == ["b.csv", "a.json"])
+    }
+}
diff --git a/apple/TableProTests/Core/Services/ImportServiceBatchTests.swift b/apple/TableProTests/Core/Services/ImportServiceBatchTests.swift
new file mode 100644
index 0000000000..ed342f9883
--- /dev/null
+++ b/apple/TableProTests/Core/Services/ImportServiceBatchTests.swift
@@ -0,0 +1,264 @@
+//
+//  ImportServiceBatchTests.swift
+//  TableProTests
+//
+
+import Foundation
+@testable import TablePro
+import TableProPluginKit
+import Testing
+
+/// Records every statement and answers the session state an engine that cannot report it would.
+private final class RecordingDriver: PluginDriverStub, @unchecked Sendable {
+    var transactionState: PluginSessionTransactionState = .unknown
+    var failingValue: String?
+    private(set) var events: [String] = []
+
+    func execute(query: String) async throws -> PluginQueryResult {
+        events.append(query)
+        if let failingValue, query.contains("'\(failingValue)'") {
+            throw DatabaseError.queryFailed("Duplicate entry '\(failingValue)'")
+        }
+        return .empty
+    }
+
+    func switchDatabase(to database: String) async throws {}
+
+    func sessionTransactionState() async -> PluginSessionTransactionState {
+        transactionState
+    }
+
+    var inserts: [String] {
+        events.filter { $0.hasPrefix("INSERT") }
+    }
+}
+
+/// A row format the way the CSV, JSON and XLSX importers run: every line of the file is one row, written through
+/// `RowImportRunner` with the configuration the format reports.
+private final class LineRowsPlugin: ImportFormatPlugin {
+    static let pluginName = "Line Rows"
+    static let pluginVersion = "1.0"
+    static let pluginDescription = "One row per line"
+    static let formatId = "tablepro-tests-line-rows"
+    static let formatDisplayName = "Lines"
+    static let acceptedFileExtensions = ["lines"]
+    static let iconName = "doc"
+    static let requiresTargetTable = true
+
+    private let configuration: RowImportRunner.Configuration
+
+    required init() {
+        configuration = RowImportRunner.Configuration(
+            errorHandling: .stopAndRollback, wrapInTransaction: true, deleteExistingRows: false
+        )
+    }
+
+    init(_ configuration: RowImportRunner.Configuration) {
+        self.configuration = configuration
+    }
+
+    var rowImportConfiguration: RowImportRunner.Configuration? { configuration }
+
+    func performImport(
+        source: any PluginImportSource,
+        sink: any PluginImportDataSink,
+        progress: PluginImportProgress
+    ) async throws -> PluginImportResult {
+        let text = try String(contentsOf: source.fileURL(), encoding: .utf8)
+        let lines = text.split(separator: "\n").map(String.init)
+        var delivered = false
+        let outcome = try await RowImportRunner.run(configuration: configuration, sink: sink, progress: progress) {
+            guard !delivered else { return nil }
+            delivered = true
+            return lines.enumerated().map { index, line in (line: index + 1, row: ["id": .text(line)]) }
+        }
+        return PluginImportResult(
+            executedStatements: outcome.inserted,
+            executionTime: 0,
+            skippedStatements: outcome.skipped,
+            errors: outcome.errors
+        )
+    }
+}
+
+private actor RecordingHistory: QueryHistoryRecording {
+    private(set) var queries: [String] = []
+
+    func record(_ request: QueryHistoryRecordRequest) async -> Bool {
+        queries.append(request.query)
+        return true
+    }
+}
+
+/// Twenty files of a folder import as one run: one transaction, so Stop and Rollback takes back every file when one
+/// fails, and a file's sink never sends a `BEGIN` of its own inside it.
+@MainActor
+struct ImportServiceBatchTests {
+    private struct Run {
+        let report: RowImportBatchReport
+        let driver: RecordingDriver
+        let history: [String]
+    }
+
+    private func runFiles(
+        _ contents: [String],
+        policy: RowImportBatchPolicy,
+        configuration: RowImportRunner.Configuration? = nil,
+        failingValue: String? = nil,
+        sessionState: PluginSessionTransactionState = .unknown,
+        stopsFirst: Bool = false
+    ) async throws -> Run {
+        let driver = RecordingDriver()
+        driver.failingValue = failingValue
+        driver.transactionState = sessionState
+        let connection = TestFixtures.makeConnection(database: "shop", type: .mysql)
+        var session = ConnectionSession(
+            connection: connection,
+            driver: PluginDriverAdapter(connection: connection, pluginDriver: driver)
+        )
+        session.browseDatabase = "shop"
+        session.status = .connected
+        DatabaseManager.shared.injectSession(session, for: connection.id)
+        defer { DatabaseManager.shared.removeSession(for: connection.id) }
+
+        let formatId = "\(LineRowsPlugin.formatId)-\(UUID().uuidString)"
+        PluginManager.shared.importPlugins[formatId] = LineRowsPlugin(configuration ?? RowImportRunner.Configuration(
+            errorHandling: policy.errorHandling,
+            wrapInTransaction: policy.wrapInTransaction,
+            deleteExistingRows: false
+        ))
+        defer { PluginManager.shared.importPlugins[formatId] = nil }
+
+        let folder = FileManager.default.temporaryDirectory.appendingPathComponent("batch-import-\(UUID().uuidString)")
+        try FileManager.default.createDirectory(at: folder, withIntermediateDirectories: true)
+        defer { try? FileManager.default.removeItem(at: folder) }
+        var items: [RowImportRunItem] = []
+        for (index, content) in contents.enumerated() {
+            let url = folder.appendingPathComponent("file\(index + 1).lines")
+            try Data(content.utf8).write(to: url)
+            items.append(RowImportRunItem(
+                url: url,
+                sourceName: url.lastPathComponent,
+                formatId: formatId,
+                targetTable: "table\(index + 1)",
+                columnMapping: ["id": "id"],
+                sourceFields: ["id"]
+            ))
+        }
+
+        let history = RecordingHistory()
+        let service = ImportService(connection: connection, historyRecorder: history, gate: AlwaysAllowGate())
+        if stopsFirst {
+            service.cancelImport()
+        }
+        let report = try await service.importFiles(
+            items,
+            scope: DatabaseScope(connectionId: connection.id, database: "shop", schema: nil),
+            policy: policy
+        )
+        return Run(report: report, driver: driver, history: await history.queries)
+    }
+
+    private static let rollback = RowImportBatchPolicy(errorHandling: .stopAndRollback, wrapInTransaction: true)
+
+    @Test("Every file runs inside one transaction, with no BEGIN of its own")
+    func oneTransactionForEveryFile() async throws {
+        let run = try await runFiles(["a1\na2", "b1", "c1"], policy: Self.rollback)
+
+        #expect(run.report.ending == .completed)
+        #expect(run.driver.events.filter { $0 == "BEGIN" }.count == 1)
+        #expect(run.driver.events.filter { $0 == "COMMIT" }.count == 1)
+        #expect(run.driver.events.first(where: { $0 == "BEGIN" || $0.hasPrefix("INSERT") }) == "BEGIN")
+        #expect(run.driver.events.last(where: { $0 == "COMMIT" || $0.hasPrefix("INSERT") }) == "COMMIT")
+        #expect(run.driver.inserts.count == 3)
+        #expect(run.report.importedRowCount == 4)
+        #expect(run.report.transaction == .committed)
+    }
+
+    @Test("A failure in a later file rolls back the files before it under Stop and Rollback")
+    func failureRollsBackEveryFile() async throws {
+        let contents = (1...20).map { "row\($0)" }
+        let run = try await runFiles(contents, policy: Self.rollback, failingValue: "row12")
+
+        #expect(run.report.ending == .failed(file: 11))
+        #expect(run.driver.inserts.count == 12)
+        #expect(run.driver.events.contains("ROLLBACK"))
+        #expect(!run.driver.events.contains("COMMIT"))
+        #expect(run.report.transaction == .rolledBack)
+        #expect(!run.report.rowsKept)
+        guard case .notRun = run.report.outcomes[12] else {
+            Issue.record("The file after the failure must not run")
+            return
+        }
+    }
+
+    @Test("A rolled-back run records only the failure in history")
+    func rolledBackRunRecordsOnlyTheFailure() async throws {
+        let run = try await runFiles(["a1", "b1", "bad"], policy: Self.rollback, failingValue: "bad")
+
+        #expect(run.history == ["-- Import from file3.lines (0 statements before failure)"])
+    }
+
+    @Test("Stop and Commit keeps the files before the failure")
+    func stopAndCommitKeepsEarlierFiles() async throws {
+        let run = try await runFiles(
+            ["a1", "b1", "bad"],
+            policy: RowImportBatchPolicy(errorHandling: .stopAndCommit, wrapInTransaction: true),
+            failingValue: "bad"
+        )
+
+        #expect(run.report.ending == .failed(file: 2))
+        #expect(run.driver.events.last == "COMMIT")
+        #expect(!run.driver.events.contains("ROLLBACK"))
+        #expect(run.report.rowsKept)
+        #expect(run.history.contains("-- Import from file1.lines (1 statements)"))
+    }
+
+    @Test("Skip and Continue opens no transaction and goes on past a failing row")
+    func skipAndContinueGoesOn() async throws {
+        let run = try await runFiles(
+            ["a1\nbad\na3", "b1"],
+            policy: RowImportBatchPolicy(errorHandling: .skipAndContinue, wrapInTransaction: true),
+            failingValue: "bad"
+        )
+
+        #expect(run.report.ending == .completed)
+        #expect(!run.driver.events.contains("BEGIN"))
+        #expect(run.report.importedRowCount == 3)
+        #expect(run.report.hasFailures)
+    }
+
+    @Test("A run without a transaction stops at the failure and keeps the files before it")
+    func noTransactionStopsAndKeeps() async throws {
+        let run = try await runFiles(
+            ["a1", "bad", "c1"],
+            policy: RowImportBatchPolicy(errorHandling: .stopAndRollback, wrapInTransaction: false),
+            failingValue: "bad"
+        )
+
+        #expect(run.report.ending == .failed(file: 1))
+        #expect(!run.driver.events.contains("BEGIN"))
+        #expect(!run.driver.events.contains("ROLLBACK"))
+        #expect(run.report.rowsKept)
+        #expect(run.driver.inserts.count == 2)
+    }
+
+    @Test("A session holding the user's transaction is joined, never committed or rolled back")
+    func userTransactionIsJoined() async throws {
+        let run = try await runFiles(["a1", "bad"], policy: Self.rollback, failingValue: "bad", sessionState: .inTransaction)
+
+        #expect(!run.driver.events.contains("BEGIN"))
+        #expect(!run.driver.events.contains("COMMIT"))
+        #expect(!run.driver.events.contains("ROLLBACK"))
+    }
+
+    /// The progress sheet is up while the new tables are created, before any file's progress exists to cancel.
+    @Test("A Stop pressed before the first file runs imports nothing")
+    func stopBeforeTheRunImportsNothing() async throws {
+        let run = try await runFiles(["a1", "b1"], policy: Self.rollback, stopsFirst: true)
+
+        #expect(run.report.ending == .stopped(file: 0))
+        #expect(run.driver.events.isEmpty)
+        #expect(run.history.isEmpty)
+    }
+}
diff --git a/apple/TableProTests/Views/Import/RowImportBatchTests.swift b/apple/TableProTests/Views/Import/RowImportBatchTests.swift
new file mode 100644
index 0000000000..6bdbe91290
--- /dev/null
+++ b/apple/TableProTests/Views/Import/RowImportBatchTests.swift
@@ -0,0 +1,250 @@
+//
+//  RowImportBatchTests.swift
+//  TableProTests
+//
+
+import Foundation
+import TableProPluginKit
+import Testing
+
+@testable import TablePro
+
+/// Each file of a folder starts on the table its name matches, or a new table named the way the one-file sheet
+/// names it, and the run starts only when every file is ready and the files agree on one transaction.
+@MainActor
+struct RowImportBatchTests {
+    private let suite: EphemeralDefaults
+    private let connection = TestFixtures.makeConnection(database: "shop", type: .postgresql)
+
+    init() throws {
+        suite = try EphemeralDefaults()
+    }
+
+    private var scope: DatabaseScope {
+        DatabaseScope(connectionId: connection.id, database: "shop", schema: "public")
+    }
+
+    private final class Catalog {
+        var tables: [String] = []
+        var columns: [String: [String]] = [:]
+        var fields: [String: [String]] = [:]
+        var configurations: [String: RowImportRunner.Configuration] = [:]
+        var unreadable: Set<String> = []
+        var tableListFails = false
+    }
+
+    private struct Unreadable: LocalizedError {
+        var errorDescription: String? { "Line 3 is not valid UTF-8 text." }
+    }
+
+    private static func configuration(
+        _ errorHandling: ImportErrorHandling = .stopAndRollback,
+        wrap: Bool = true,
+        deleteExisting: Bool = false
+    ) -> RowImportRunner.Configuration {
+        RowImportRunner.Configuration(errorHandling: errorHandling, wrapInTransaction: wrap, deleteExistingRows: deleteExisting)
+    }
+
+    private func services(_ catalog: Catalog) -> RowImportBatch.Services {
+        RowImportBatch.Services(
+            fetchTables: { _ in
+                if catalog.tableListFails { throw DatabaseError.notConnected }
+                return catalog.tables.map { TableInfo(name: $0, type: .table, rowCount: nil, schema: "public") }
+            },
+            fetchColumns: { table, _ in catalog.columns[table] ?? [] },
+            detectFields: { _, url, _ in
+                if catalog.unreadable.contains(url.lastPathComponent) { throw Unreadable() }
+                return (catalog.fields[url.lastPathComponent] ?? ["id", "name"]).map {
+                    PluginImportField(name: $0, sampleValue: nil, inferredType: .text)
+                }
+            },
+            configuration: { catalog.configurations[$0] ?? Self.configuration() },
+            formatName: { $0.uppercased() },
+            sqlType: { _ in "TEXT" }
+        )
+    }
+
+    private func makeBatch(_ names: [String], catalog: Catalog) async -> RowImportBatch {
+        let files = names.map { name in
+            ImportBatchFiles.File(
+                url: URL(fileURLWithPath: "/tmp/exports/\(name)"),
+                formatId: URL(fileURLWithPath: name).pathExtension
+            )
+        }
+        let batch = RowImportBatch(
+            files: files,
+            leftOut: [],
+            connection: connection,
+            scope: scope,
+            services: services(catalog),
+            mappingStore: ImportColumnMappingStore(defaults: suite.defaults)
+        )
+        await batch.loadTables()
+        await batch.readFiles()
+        return batch
+    }
+
+    private func targets(_ batch: RowImportBatch) -> [String] {
+        batch.files.map { file in
+            switch file.target.destination {
+            case .existingTable: return "existing:\(file.target.existingTable ?? "")"
+            case .newTable: return "new:\(file.target.newTableName)"
+            }
+        }
+    }
+
+    @Test("A file whose name matches a table goes into it, the rest into new tables named after them")
+    func proposesMatchingAndNewTables() async {
+        let catalog = Catalog()
+        catalog.tables = ["orders"]
+        catalog.columns["orders"] = ["id", "name"]
+
+        let batch = await makeBatch(["Orders.csv", "Customer List.csv"], catalog: catalog)
+
+        #expect(targets(batch) == ["existing:orders", "new:customer_list"])
+        #expect(batch.files.allSatisfy { $0.read == .ready })
+        #expect(batch.canImport)
+    }
+
+    @Test("Two files that would get one new name get two, and neither takes a name the database has")
+    func newNamesAreUniqueAcrossTheRun() async {
+        let catalog = Catalog()
+        catalog.tables = ["sales_2024_view"]
+
+        let batch = await makeBatch(["sales 2024.csv", "sales-2024.json", "sales_2024.csv"], catalog: catalog)
+
+        #expect(targets(batch) == ["new:sales_2024", "new:sales_2024_2", "new:sales_2024_3"])
+        #expect(batch.problem == nil)
+    }
+
+    @Test("Twenty files of a folder are ready to import into twenty new tables")
+    func twentyFilesAreReady() async throws {
+        let names = (1...20).map { "export_\($0).csv" }
+        let batch = await makeBatch(names, catalog: Catalog())
+
+        let plan = try #require(batch.plan())
+        #expect(plan.items.count == 20)
+        #expect(plan.newTables.count == 20)
+        #expect(Set(plan.items.map(\.targetTable)).count == 20)
+    }
+
+    @Test("Two files given one new table name stop the run")
+    func duplicateNewTableNameIsAProblem() async {
+        let batch = await makeBatch(["a.csv", "b.csv"], catalog: Catalog())
+        let second = batch.files[1].id
+        var target = batch.files[1].target
+        target.newTableName = "a"
+        batch.apply(target, to: second)
+
+        #expect(batch.problem == String(format: String(localized: "Two files create a table named %@."), "a"))
+        #expect(batch.plan() == nil)
+    }
+
+    @Test("Formats that handle errors differently cannot share one transaction")
+    func formatsMustAgreeOnThePolicy() async {
+        let catalog = Catalog()
+        catalog.configurations["json"] = Self.configuration(.skipAndContinue)
+
+        let batch = await makeBatch(["a.csv", "b.json"], catalog: catalog)
+
+        #expect(batch.problem == String(
+            format: String(localized: "Set On error and Wrap in transaction the same for %@."),
+            ["CSV", "JSON"].formatted(.list(type: .and))
+        ))
+        #expect(!batch.canImport)
+    }
+
+    @Test("Wrap in transaction does not split formats that all skip and continue")
+    func skipAndContinueIgnoresWrap() async {
+        let catalog = Catalog()
+        catalog.configurations["csv"] = Self.configuration(.skipAndContinue, wrap: true)
+        catalog.configurations["json"] = Self.configuration(.skipAndContinue, wrap: false)
+
+        let batch = await makeBatch(["a.csv", "b.json"], catalog: catalog)
+
+        #expect(batch.problem == nil)
+        #expect(batch.policy?.usesTransaction == false)
+    }
+
+    @Test("Delete existing rows refuses a table two files fill")
+    func clearingATableTwiceIsAProblem() async {
+        let catalog = Catalog()
+        catalog.tables = ["orders"]
+        catalog.columns["orders"] = ["id", "name"]
+        catalog.configurations["csv"] = Self.configuration(deleteExisting: true)
+
+        let batch = await makeBatch(["orders.csv", "march.csv"], catalog: catalog)
+        batch.choose(.existingTable("orders"), for: batch.files[1].id)
+        await batch.readFiles()
+
+        #expect(batch.problem?.hasPrefix("orders takes rows from two files") == true)
+    }
+
+    @Test("Delete existing rows refuses a table one file creates and another fills")
+    func clearingANewTableTwiceIsAProblem() async {
+        let catalog = Catalog()
+        catalog.configurations["csv"] = Self.configuration(deleteExisting: true)
+        let batch = await makeBatch(["orders.csv", "march.csv"], catalog: catalog)
+        batch.planner.recordCreated(TableScope(table: "orders", in: scope), createTableSQL: "CREATE TABLE orders (id TEXT)")
+        catalog.tables = ["orders"]
+        catalog.columns["orders"] = ["id", "name"]
+        await batch.loadTables()
+        batch.choose(.existingTable("orders"), for: batch.files[1].id)
+        await batch.readFiles()
+
+        #expect(batch.problem?.hasPrefix("orders takes rows from two files") == true)
+    }
+
+    @Test("A file that cannot be read names itself, and leaving it out lets the rest import")
+    func unreadableFileCanBeLeftOut() async {
+        let catalog = Catalog()
+        catalog.unreadable = ["broken.csv"]
+
+        let batch = await makeBatch(["good.csv", "broken.csv"], catalog: catalog)
+
+        #expect(batch.problem == String(
+            format: String(localized: "%1$@ cannot be read. %2$@"), "broken.csv", "Line 3 is not valid UTF-8 text."
+        ))
+        batch.setIncluded(false, for: batch.files[1].id)
+        #expect(batch.canImport)
+        #expect(batch.plan()?.items.map(\.sourceName) == ["good.csv"])
+    }
+
+    @Test("A file into an existing table maps its fields by name and imports only mapped ones")
+    func existingTableMapsByName() async throws {
+        let catalog = Catalog()
+        catalog.tables = ["people"]
+        catalog.columns["people"] = ["id", "full_name"]
+        catalog.fields["people.csv"] = ["id", "full_name", "notes"]
+
+        let batch = await makeBatch(["people.csv"], catalog: catalog)
+
+        let plan = try #require(batch.plan())
+        #expect(plan.items.first?.columnMapping == ["id": "id", "full_name": "full_name"])
+        #expect(plan.newTables.isEmpty)
+    }
+
+    @Test("Without a table list every file still gets a new table, and the list error is shown")
+    func failedTableListStillProposes() async {
+        let catalog = Catalog()
+        catalog.tableListFails = true
+
+        let batch = await makeBatch(["orders.csv"], catalog: catalog)
+
+        #expect(targets(batch) == ["new:orders"])
+        #expect(batch.tableListError != nil)
+        #expect(batch.catalogNameKeys == nil)
+    }
+
+    @Test("A table the run created on a failed attempt is not reported as taken on the retry")
+    func ownTablesAreNotTaken() async {
+        let catalog = Catalog()
+        let batch = await makeBatch(["orders.csv"], catalog: catalog)
+        batch.planner.recordCreated(TableScope(table: "orders", in: scope), createTableSQL: "CREATE TABLE orders (id TEXT)")
+        catalog.tables = ["orders"]
+        await batch.loadTables()
+
+        #expect(targets(batch) == ["new:orders"])
+        #expect(batch.problem == nil)
+    }
+}
diff --git a/apple/changelog.d/3745.added.md b/apple/changelog.d/3745.added.md
new file mode 100644
index 0000000000..90ecb34b87
--- /dev/null
+++ b/apple/changelog.d/3745.added.md
@@ -0,0 +1 @@
+Import Data… takes several files or a folder and imports each into its own table in one run.