feat(import): import rows as Update, Insert or Update, or Insert, Skip Existing

apple/TablePro/Core/ChangeTracking/SQLStatementGenerator.swift

@@ -273,6 +273,46 @@ struct SQLStatementGenerator {
         return ParameterizedStatement(sql: sql, parameters: bindParameters)
     }
 
+    // The server compares, so its casting and collation decide a match, as they do for the UPDATE that follows.
+    func keyMatchCountStatement(keyColumns: [String], keys: [[PluginCellValue]]) -> ParameterizedStatement? {
+        guard !keyColumns.isEmpty, !keys.isEmpty, keys.allSatisfy({ $0.count == keyColumns.count }) else { return nil }
+
+        var bindParameters: [Any?] = []
+        let quotedKey = keyColumns.map(quoteIdentifierFn)
+        let selects = keys.enumerated().map { position, values -> String in
+            let conditions = zip(quotedKey, values).map { column, value -> String in
+                bindParameters.append(value.asAny)
+                return "\(column) = \(placeholder(at: bindParameters.count - 1))"
+            }.joined(separator: " AND ")
+            return "SELECT \(position), COUNT(*) FROM \(qualifiedTableName) WHERE \(conditions)"
+        }
+
+        return ParameterizedStatement(sql: selects.joined(separator: " UNION ALL "), parameters: bindParameters)
+    }
+
+    func keyedUpdateStatement(
+        columns setColumns: [String],
+        values: [PluginCellValue],
+        keyColumns: [String],
+        keyValues: [PluginCellValue]
+    ) -> ParameterizedStatement? {
+        guard !setColumns.isEmpty, setColumns.count == values.count,
+              !keyColumns.isEmpty, keyColumns.count == keyValues.count else { return nil }
+
+        var bindParameters: [Any?] = []
+        func bound(_ column: String, _ value: PluginCellValue) -> String {
+            bindParameters.append(value.asAny)
+            return "\(quoteIdentifierFn(column)) = \(placeholder(at: bindParameters.count - 1))"
+        }
+        let assignments = zip(setColumns, values).map(bound).joined(separator: ", ")
+        let conditions = zip(keyColumns, keyValues).map(bound).joined(separator: " AND ")
+
+        return ParameterizedStatement(
+            sql: "UPDATE \(qualifiedTableName) SET \(assignments) WHERE \(conditions)",
+            parameters: bindParameters
+        )
+    }
+
     var maxBindParameters: Int {
         switch databaseType {
         case .sqlite: 32_766

apple/TablePro/Core/Plugins/ImportDataSinkAdapter.swift

@@ -36,6 +36,8 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
     private let sourceFields: Set<String>
     private let sourceFieldsPerFoldedName: [String: Int]
     private let rowGenerator: SQLStatementGenerator?
+    private let rowWriting: ImportRowWriting
+    private let tally: OSAllocatedUnfairLock<ImportRowTally>
 
     /// Asked before every statement this sink sends, because one `insertRows` call is no longer one
     /// statement: a byte-heavy batch, or an Oracle target that takes one row per statement, splits
@@ -82,10 +84,13 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
         targetTable: String? = nil,
         columnMapping: [String: String] = [:],
         sourceFields: Set<String> = [],
+        rowWriting: ImportRowWriting = .insert,
         joinsTransaction: Bool = false,
         isCancelled: @escaping @Sendable () -> Bool = { false }
     ) {
         self.isCancelled = isCancelled
+        self.rowWriting = rowWriting
+        self.tally = OSAllocatedUnfairLock(initialState: ImportRowTally(mode: rowWriting.mode))
         self.joinsTransaction = joinsTransaction
         self.driver = driver
         self.databaseType = databaseType
@@ -114,6 +119,10 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
         try await execute(statement: statement, line: 1)
     }
 
+    var rowTally: ImportRowTally {
+        tally.withLock { $0 }
+    }
+
     var transactionOutcome: ImportTransactionOutcome {
         transactionRecord.withLock { record in
             switch record.ending {
@@ -190,20 +199,25 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
             throw PluginImportError.importFailed("Could not resolve SQL dialect for \(targetTable)")
         }
 
-        let (columns, bindValues) = mappedColumnsAndValues(values)
-        guard !columns.isEmpty else {
+        let row = mappedRow(values)
+        guard !row.columns.isEmpty else {
             guard values.isEmpty else {
                 throw PluginImportError.importFailed(Self.unmappedRowMessage)
             }
             return
         }
-        guard let statement = rowGenerator.insertStatement(columns: columns, values: bindValues) else {
+        guard !rowWriting.mode.matchesExistingRows else {
+            try await writeMatching([row], generator: rowGenerator)
+            return
+        }
+        guard let statement = rowGenerator.insertStatement(columns: row.columns, values: row.values) else {
             throw PluginImportError.importFailed(
                 String(format: String(localized: "Could not build an INSERT for %@"), targetTable)
             )
         }
 
         _ = try await driver.executeParameterized(query: statement.sql, parameters: statement.parameters)
+        tally.withLock { $0.inserted += 1 }
     }
 
     func insertRows(_ rows: [[String: PluginCellValue]]) async throws {
@@ -214,29 +228,42 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
             throw PluginImportError.importFailed("Could not resolve SQL dialect for row import")
         }
 
-        if rows.contains(where: { !$0.isEmpty && mappedColumnsAndValues($0).0.isEmpty }) {
+        let mapped = rows.map(mappedRow)
+        if zip(rows, mapped).contains(where: { !$0.0.isEmpty && $0.1.columns.isEmpty }) {
             throw PluginImportError.importFailed(Self.unmappedRowMessage)
         }
+        let writable = mapped.filter { !$0.columns.isEmpty }
+
+        if rowWriting.mode.matchesExistingRows {
+            try await writeMatching(writable, generator: rowGenerator)
+        } else {
+            try await insertGrouped(writable, generator: rowGenerator)
+        }
+    }
 
+    private struct MappedRow {
+        let columns: [String]
+        let values: [PluginCellValue]
+    }
+
+    private func mappedRow(_ values: [String: PluginCellValue]) -> MappedRow {
+        let (columns, mappedValues) = mappedColumnsAndValues(values)
+        return MappedRow(columns: columns, values: mappedValues)
+    }
+
+    private func insertGrouped(_ rows: [MappedRow], generator: SQLStatementGenerator) async throws {
         var index = 0
         while index < rows.count {
-            let (columns, values) = mappedColumnsAndValues(rows[index])
-            guard !columns.isEmpty else {
-                index += 1
-                continue
-            }
-
-            var groupValues: [[PluginCellValue]] = [values]
+            let columns = rows[index].columns
+            var groupValues: [[PluginCellValue]] = [rows[index].values]
             var next = index + 1
-            while next < rows.count {
-                let (nextColumns, nextValues) = mappedColumnsAndValues(rows[next])
-                guard nextColumns == columns else { break }
-                groupValues.append(nextValues)
+            while next < rows.count, rows[next].columns == columns {
+                groupValues.append(rows[next].values)
                 next += 1
             }
             index = next
 
-            try await insertGroup(groupValues, columns: columns, generator: rowGenerator)
+            try await insertGroup(groupValues, columns: columns, generator: generator)
         }
     }
 
@@ -275,6 +302,170 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
             )
         }
         _ = try await driver.executeParameterized(query: statement.sql, parameters: statement.parameters)
+        tally.withLock { $0.inserted += batch.count }
+    }
+
+    // MARK: - Matching existing rows
+
+    private struct KeyedRow {
+        let row: MappedRow
+        let key: [PluginCellValue]
+    }
+
+    // Not the engines' upsert forms: they refuse a NOT NULL column the file leaves out even when the row exists.
+    private func writeMatching(_ rows: [MappedRow], generator: SQLStatementGenerator) async throws {
+        var run: [KeyedRow] = []
+        var keysInRun = Set<[String]>()
+        for row in rows {
+            let key = try keyValues(of: row)
+            // A repeated key has to see the write of its first occurrence, so the run ends before it.
+            let folded = key.contains(where: \.isNull) ? nil : key.map(Self.foldedKeyValue)
+            if let folded, keysInRun.contains(folded) {
+                try await writeMatchedRun(run, generator: generator)
+                run.removeAll(keepingCapacity: true)
+                keysInRun.removeAll(keepingCapacity: true)
+            }
+            if let folded { keysInRun.insert(folded) }
+            run.append(KeyedRow(row: row, key: key))
+        }
+        try await writeMatchedRun(run, generator: generator)
+    }
+
+    // Folds more than a collation or a numeric comparison does: two keys the database takes as one must not share a run.
+    static func foldedKeyValue(_ value: PluginCellValue) -> String {
+        switch value {
+        case .null:
+            return "null"
+        case .bytes(let data):
+            return "bytes:" + data.base64EncodedString()
+        case .text(let text):
+            let folded = text.trimmingCharacters(in: .whitespacesAndNewlines)
+                .folding(options: [.caseInsensitive, .diacriticInsensitive, .widthInsensitive], locale: nil)
+            // A Double, not a Decimal: a REAL column holds 0.1 and 0.100000000000000001 as one value.
+            guard folded.contains(where: { $0.isASCII && $0.isNumber }),
+                  folded.allSatisfy({ numberCharacters.contains($0) }),
+                  let number = Double(folded) else {
+                return "text:" + folded
+            }
+            return "number:\(number == 0 ? 0 : number)"
+        }
+    }
+
+    private static let numberCharacters = Set("0123456789+-.e")
+
+    private func keyValues(of row: MappedRow) throws -> [PluginCellValue] {
+        try rowWriting.keyColumns.map { column in
+            guard let index = row.columns.firstIndex(of: column) else {
+                throw PluginImportError.importFailed(String(
+                    format: String(localized: "This row has no value for the key column %@, so it cannot be matched."),
+                    column
+                ))
+            }
+            return row.values[index]
+        }
+    }
+
+    private func writeMatchedRun(_ run: [KeyedRow], generator: SQLStatementGenerator) async throws {
+        guard !run.isEmpty else { return }
+        let counts = try await matchCounts(of: run.map(\.key), generator: generator)
+        if let ambiguous = zip(run, counts).first(where: { $0.1 > 1 }) {
+            throw PluginImportError.importFailed(String(
+                format: String(localized: "The key %1$@ matched %2$@ rows, so the row was not written."),
+                keyDescription(ambiguous.0.key),
+                ambiguous.1.formatted()
+            ))
+        }
+
+        // Writes go out in file order, so a new parent row lands before an update that points at it.
+        var inserts: [MappedRow] = []
+        for (entry, count) in zip(run, counts) {
+            switch (count, rowWriting.mode) {
+            case (0, .update):
+                tally.withLock { $0.unmatched += 1 }
+            case (0, _):
+                inserts.append(entry.row)
+            case (_, .update), (_, .upsert):
+                try await insertGrouped(inserts, generator: generator)
+                inserts.removeAll(keepingCapacity: true)
+                try await update(entry, generator: generator)
+            default:
+                tally.withLock { $0.existing += 1 }
+            }
+        }
+        try await insertGrouped(inserts, generator: generator)
+    }
+
+    private func update(_ entry: KeyedRow, generator: SQLStatementGenerator) async throws {
+        let keyColumns = Set(rowWriting.keyColumns)
+        let assignments = zip(entry.row.columns, entry.row.values).filter { !keyColumns.contains($0.0) }
+        guard !assignments.isEmpty else {
+            tally.withLock { $0.existing += 1 }
+            return
+        }
+        guard !isCancelled() else { throw PluginImportCancellationError() }
+        guard let statement = generator.keyedUpdateStatement(
+            columns: assignments.map(\.0),
+            values: assignments.map(\.1),
+            keyColumns: rowWriting.keyColumns,
+            keyValues: entry.key
+        ) else {
+            throw PluginImportError.importFailed(String(localized: "Could not build an UPDATE for the mapped columns"))
+        }
+        _ = try await driver.executeParameterized(query: statement.sql, parameters: statement.parameters)
+        tally.withLock { $0.updated += 1 }
+    }
+
+    private static let keysPerProbe = 100
+
+    private func matchCounts(of keys: [[PluginCellValue]], generator: SQLStatementGenerator) async throws -> [Int] {
+        let keyWidth = max(1, rowWriting.keyColumns.count)
+        let perStatement = max(1, min(Self.keysPerProbe, generator.maxBindParameters / keyWidth))
+        var counts: [Int] = []
+        counts.reserveCapacity(keys.count)
+        var start = 0
+        while start < keys.count {
+            let chunk = Array(keys[start..<min(start + perStatement, keys.count)])
+            guard !isCancelled() else { throw PluginImportCancellationError() }
+            guard let statement = generator.keyMatchCountStatement(keyColumns: rowWriting.keyColumns, keys: chunk) else {
+                throw PluginImportError.importFailed(String(localized: "Could not build the query that matches rows"))
+            }
+            let result = try await driver.executeParameterized(query: statement.sql, parameters: statement.parameters)
+            counts += try Self.counts(in: result, expected: chunk.count)
+            start += chunk.count
+        }
+        return counts
+    }
+
+    private static func counts(in result: QueryResult, expected: Int) throws -> [Int] {
+        var counts = [Int?](repeating: nil, count: expected)
+        for row in result.rows {
+            guard row.count >= 2, let position = integer(row[0]), counts.indices.contains(position),
+                  let count = integer(row[1]) else { continue }
+            counts[position] = count
+        }
+        let read = counts.compactMap { $0 }
+        guard read.count == expected else {
+            throw PluginImportError.importFailed(String(localized: "Could not read which rows are already in the table"))
+        }
+        return read
+    }
+
+    private static func integer(_ value: PluginCellValue) -> Int? {
+        guard let text = value.asText?.trimmingCharacters(in: .whitespaces) else { return nil }
+        return Int(text) ?? Double(text).flatMap { $0.rounded() == $0 ? Int(exactly: $0) : nil }
+    }
+
+    private func keyDescription(_ key: [PluginCellValue]) -> String {
+        zip(rowWriting.keyColumns, key)
+            .map { "\($0.0) = \($0.1.isNull ? "NULL" : $0.1.sortKey)" }
+            .joined(separator: ", ")
+    }
+
+    private static var matchingRefusesClearingMessage: String {
+        String(localized: """
+            Delete existing rows before import empties the table, so no row would match. Turn it off, or import as \
+            Insert.
+            """)
     }
 
     /// A row carrying values none of which reach a mapped column writes nothing. Reporting it as
@@ -312,6 +503,9 @@ final class ImportDataSinkAdapter: PluginImportDataSink, @unchecked Sendable {
         guard targetTable != nil, let rowGenerator else {
             throw PluginImportError.importFailed("No target table configured for row import")
         }
+        guard !rowWriting.mode.matchesExistingRows else {
+            throw PluginImportError.importFailed(Self.matchingRefusesClearingMessage)
+        }
         _ = try await driver.execute(query: rowGenerator.deleteAllRowsStatement())
     }