feat(plugins): native delete and export for the engines with no SQL

CHANGELOG.md

@@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Delete for a Kafka topic, and `DROP TOPIC` in the Kafka query editor.
+- Delete for a DynamoDB table.
 - Credential profiles, one username and password shared by any number of connections. (#2853)
 - **Profiles** pane in Settings, listing credential profiles and SSH servers with how many connections use each.
 - **Credentials** picker on a connection's Authentication section, with **Save These as a Profile…**.
@@ -70,6 +72,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Stale error banner over a pinned result after clearing the results of a failed query.
 - `DROP TABLE` and `TRUNCATE TABLE` generated for Elasticsearch, Kafka, Weaviate and etcd, which have no SQL. (#2884)
+- Empty Elasticsearch and Weaviate exports, which asked the engine for `SELECT * FROM`.
+- Drop Table and Truncate Table offered on every iOS engine, including Redis keys.
 - Delete and Truncate offered on engines that have no statement for them.
 - Truncate on a Redis database emptying whichever database the connection was on.
 - Base64 text instead of the request in the Typesense drop and truncate confirmation.

Packages/TableProCore/Sources/TableProConnectionLibrary/SQLDDLFallbackPolicy.swift

@@ -0,0 +1,52 @@
+//
+//  SQLDDLFallbackPolicy.swift
+//  TableProConnectionLibrary
+//
+
+import Foundation
+
+/// Whether an app may build `DROP <kind> <name>` or `TRUNCATE TABLE <name>` itself for an engine.
+///
+/// `PluginDatabaseDriver` defaults both table operations to nil, and nil says two different things.
+/// On MySQL it means the generic DDL is right, which is why eleven SQL plugins never implement
+/// either hook. On Elasticsearch it means the engine has no such statement at all, and there the
+/// generic DDL is text the driver rejects: `DROP TABLE "test_index"` reached a cluster as a console
+/// request and came back "Enter a request like: GET /my-index/_search" (#2884).
+///
+/// Neither of the engine's other language facts can answer it. DynamoDB writes PartiQL and so
+/// declares an editor language of `.sql`, yet PartiQL has no DDL and its driver returns nil from
+/// both hooks deliberately. The SQL dialect descriptor cannot answer it either, because it defaults
+/// to nil and several engines with ordinary `DROP TABLE` never curate one. So the answer is stated
+/// here per engine.
+///
+/// Keyed by the raw database type id rather than by a `DatabaseType`, because the two apps do not
+/// share that type: the Mac app has `TablePro/Models/Connection/DatabaseType.swift` and iOS has
+/// `TableProCoreTypes.DatabaseType`, two separate structs whose constant lists have already drifted
+/// apart. This target is one of the five both apps link, and it carries no plugin ABI, so a string
+/// key here is what lets one list serve both. It follows `SqlDialect.from(databaseTypeId:)`, which
+/// is keyed the same way for the same reason.
+public enum SQLDDLFallbackPolicy {
+    /// Engines with no SQL DDL to fall back on.
+    ///
+    /// Most of these have a driver that answers both hooks, so an app never reaches the fallback for
+    /// them. They are listed anyway, because a driver answers per object kind: Typesense returns nil
+    /// for anything that is not a collection, and without this an app would answer that with
+    /// `DROP VIEW`.
+    public static let engineIdsWithoutSQLDDL: Set<String> = [
+        "Elasticsearch",
+        "Typesense",
+        "Weaviate",
+        "Kafka",
+        "etcd",
+        "Redis",
+        "MongoDB",
+        "SurrealDB",
+        "DynamoDB",
+        "Beancount",
+    ]
+
+    /// True when `DROP`/`TRUNCATE` built by the app is something this engine could run.
+    public static func allowsGeneratedDDL(databaseTypeId: String) -> Bool {
+        !engineIdsWithoutSQLDDL.contains(databaseTypeId)
+    }
+}

Packages/TableProCore/Sources/TableProWeaviateCore/WeaviateOperations.swift

@@ -16,6 +16,27 @@ public enum WeaviateOperations {
         return "DELETE /v1/schema/\(name)"
     }
 
+    public static let exportTag = "WEAVIATE_EXPORT:"
+
+    /// The statement Export asks the driver for, naming one collection and nothing else.
+    ///
+    /// A tag rather than a browse query, because a browse query carries a `limit` that would cap
+    /// the export. `streamRows` decodes this and walks the collection a page at a time, yielding
+    /// each one, so a large collection never has to fit in memory at once. Without it the app
+    /// fabricates `SELECT * FROM "<collection>"`, which this driver has no parser for.
+    public static func encodeExport(collection: String) -> String {
+        "\(exportTag)\(Data(collection.utf8).base64EncodedString())"
+    }
+
+    public static func decodeExport(_ query: String) -> String? {
+        guard query.hasPrefix(exportTag),
+              let data = Data(base64Encoded: String(query.dropFirst(exportTag.count))),
+              let collection = String(data: data, encoding: .utf8),
+              !collection.isEmpty
+        else { return nil }
+        return collection
+    }
+
     /// Weaviate has only collections, so any other kind the app asks about is not something this
     /// engine drops, and answering anyway would delete the collection of that name instead.
     public static func isCollectionObject(_ objectType: String) -> Bool {

Plugins/DynamoDBDriverPlugin/DynamoDBConnection.swift

@@ -151,6 +151,10 @@ internal struct DescribeTableResponse: Decodable {
     let Table: TableDescription
 }
 
+/// DeleteTable answers with the table's description as it enters DELETING. Nothing here reads it:
+/// the sidebar refreshes from ListTables, and the useful part of the reply is that it succeeded.
+internal struct DeleteTableResponse: Decodable {}
+
 internal struct TableDescription: Decodable {
     let TableName: String
     let KeySchema: [KeySchemaElement]?
@@ -338,6 +342,13 @@ internal final class DynamoDBConnection: @unchecked Sendable {
         return try await request(target: "DynamoDB_20120810.DescribeTable", body: body)
     }
 
+    /// DeleteTable answers as soon as the table enters DELETING, not once it is gone, so a listing
+    /// taken straight afterwards can still show it.
+    func deleteTable(tableName: String) async throws -> DeleteTableResponse {
+        let body: [String: Any] = ["TableName": tableName]
+        return try await request(target: "DynamoDB_20120810.DeleteTable", body: body)
+    }
+
     func scan(
         tableName: String,
         limit: Int? = nil,

Plugins/DynamoDBDriverPlugin/DynamoDBOperations.swift

@@ -0,0 +1,52 @@
+//
+//  DynamoDBOperations.swift
+//  DynamoDBDriverPlugin
+//
+//  The table operations the app offers, as statements this driver also executes.
+//
+
+import Foundation
+
+enum DynamoDBOperations {
+    /// The statement that deletes one table.
+    ///
+    /// PartiQL has no DDL, so this is the driver's own vocabulary rather than something the
+    /// `ExecuteStatement` API would take: `execute(query:)` recognises it and issues `DeleteTable`.
+    /// Spelled `DROP TABLE` because the confirmation dialog shows the statement verbatim and
+    /// `QueryClassifier` reads the leading verb to tier it destructive, and because it is what a
+    /// person reading it would expect. Returning nil instead let the app invent the same text and
+    /// send it to `ExecuteStatement`, which rejects it (#2884).
+    static func dropTable(named name: String, objectType: String) -> String? {
+        guard isTableObject(objectType), let quoted = quotedName(name) else { return nil }
+        return "DROP TABLE \(quoted)"
+    }
+
+    /// DynamoDB has only tables, so any other kind the app asks about is not something this engine
+    /// drops, and answering anyway would delete the table of that name instead.
+    static func isTableObject(_ objectType: String) -> Bool {
+        objectType.uppercased() == "TABLE"
+    }
+
+    /// The table named by a drop statement, or nil when the text is not one.
+    static func droppedTableName(in statement: String) -> String? {
+        let trimmed = statement.trimmingCharacters(in: .whitespacesAndNewlines)
+        guard trimmed.uppercased().hasPrefix("DROP TABLE ") else { return nil }
+        let rest = trimmed.dropFirst("DROP TABLE ".count).trimmingCharacters(in: .whitespaces)
+        guard rest.hasPrefix("\""), rest.hasSuffix("\""), rest.count > 2 else { return nil }
+        let unquoted = String(rest.dropFirst().dropLast()).replacingOccurrences(of: "\"\"", with: "\"")
+        return isValidTableName(unquoted) ? unquoted : nil
+    }
+
+    /// A DynamoDB table name is 3 to 255 characters of `a-z A-Z 0-9 _ - .` and nothing else, so a
+    /// name carrying anything more did not come from the table listing and is refused rather than
+    /// sent as a delete.
+    static func isValidTableName(_ name: String) -> Bool {
+        guard (3...255).contains(name.count) else { return false }
+        return name.allSatisfy { $0.isLetter || $0.isNumber || $0 == "_" || $0 == "-" || $0 == "." }
+    }
+
+    private static func quotedName(_ name: String) -> String? {
+        guard isValidTableName(name) else { return nil }
+        return "\"\(name)\""
+    }
+}

Plugins/DynamoDBDriverPlugin/DynamoDBPluginDriver.swift

@@ -56,13 +56,16 @@ internal final class DynamoDBPluginDriver: PluginDatabaseDriver, @unchecked Send
         "SELECT * FROM \(quoteIdentifier(table))"
     }
 
+    /// DynamoDB has no truncate. Emptying a table means scanning it and deleting every item in
+    /// batches, which is a long billed job rather than a statement, and DeleteTable plus
+    /// CreateTable loses the table's settings. Neither is what Truncate promises, so it stays
+    /// unoffered rather than offered as something else.
     func truncateTableStatements(table: String, schema: String?, cascade: Bool) -> [String]? {
-        // DynamoDB does not support TRUNCATE; scan and delete all items
         nil
     }
 
     func dropObjectStatement(name: String, objectType: String, schema: String?, cascade: Bool) -> String? {
-        nil
+        DynamoDBOperations.dropTable(named: name, objectType: objectType)
     }
 
     init(config: DriverConnectionConfig) {
@@ -123,6 +126,10 @@ internal final class DynamoDBPluginDriver: PluginDatabaseDriver, @unchecked Send
             return try await executeTaggedQuery(trimmed, conn: conn, startTime: startTime)
         }
 
+        if let table = DynamoDBOperations.droppedTableName(in: trimmed) {
+            return try await executeDropTable(table, conn: conn, startTime: startTime)
+        }
+
         return try await executePartiQL(trimmed, conn: conn, startTime: startTime)
     }
 
@@ -779,6 +786,25 @@ internal final class DynamoDBPluginDriver: PluginDatabaseDriver, @unchecked Send
 
     // MARK: - Tagged Query Execution
 
+    /// Issues DeleteTable for the driver's own `DROP TABLE "x"` statement.
+    ///
+    /// The cached description goes with it, or a table recreated under the same name would be read
+    /// through the old key schema. DeleteTable returns once the table is DELETING rather than gone,
+    /// so the row count is reported as the one table the request named, not as work completed.
+    private func executeDropTable(
+        _ table: String, conn: DynamoDBConnection, startTime: Date
+    ) async throws -> PluginQueryResult {
+        _ = try await conn.deleteTable(tableName: table)
+        lock.withLock { _tableDescriptionCache.removeValue(forKey: table) }
+        return PluginQueryResult(
+            columns: ["result"],
+            columnTypeNames: ["String"],
+            rows: [[.text("DELETING")]],
+            rowsAffected: 1,
+            executionTime: Date().timeIntervalSince(startTime)
+        )
+    }
+
     private func executeTaggedQuery(
         _ query: String, conn: DynamoDBConnection, startTime: Date
     ) async throws -> PluginQueryResult {

Plugins/ElasticsearchDriverPlugin/ElasticsearchOperations.swift

@@ -20,6 +20,28 @@ enum ElasticsearchOperations {
         return "DELETE \(path)"
     }
 
+    static let exportTag = "ELASTICSEARCH_EXPORT:"
+
+    /// The statement Export asks the driver for, naming one index and nothing else.
+    ///
+    /// A tag rather than a console request, because Export must read the whole index and a console
+    /// request carries a `size` that would cap it. `streamRows` decodes this and pages with a
+    /// point-in-time and `search_after`, yielding each batch, so a large index never has to fit in
+    /// memory at once. Without it the app fabricates `SELECT * FROM "<index>"`, which this driver
+    /// answers with "Enter a request like: GET /my-index/_search" and the file comes out empty.
+    static func encodeExport(index: String) -> String {
+        "\(exportTag)\(Data(index.utf8).base64EncodedString())"
+    }
+
+    static func decodeExport(_ query: String) -> String? {
+        guard query.hasPrefix(exportTag),
+              let data = Data(base64Encoded: String(query.dropFirst(exportTag.count))),
+              let index = String(data: data, encoding: .utf8),
+              !index.isEmpty
+        else { return nil }
+        return index
+    }
+
     /// Only an index. Elasticsearch has no views or materialized views, so anything else the app
     /// asks about is not something this engine can drop.
     static func isIndexObject(_ objectType: String) -> Bool {

Plugins/ElasticsearchDriverPlugin/ElasticsearchPluginDriver+Execution.swift

@@ -37,6 +37,120 @@ extension ElasticsearchPluginDriver {
         return try await executeConsole(trimmed, conn: conn, startTime: startTime)
     }
 
+    // MARK: - Export
+
+    /// Exports one index by paging it, yielding each batch instead of holding the whole index.
+    ///
+    /// The protocol default runs `execute` once and buffers the entire result, which is fine for a
+    /// grid page and not for an export. Anything that is not an export tag keeps that default.
+    func streamRows(query: String) -> AsyncThrowingStream<PluginStreamElement, Error> {
+        guard let index = ElasticsearchOperations.decodeExport(query) else {
+            return defaultStreamRows(query: query)
+        }
+        return AsyncThrowingStream(bufferingPolicy: .unbounded) { continuation in
+            let task = Task {
+                do {
+                    try await self.streamIndex(index, continuation: continuation)
+                    continuation.finish()
+                } catch {
+                    continuation.finish(throwing: error)
+                }
+            }
+            continuation.onTermination = { @Sendable _ in task.cancel() }
+        }
+    }
+
+    /// A point-in-time pins the index for the whole export, so a document written while it runs
+    /// cannot shift the sort and be exported twice or skipped. `search_after` walks it, which is the
+    /// only way past the 10,000 `max_result_window` ceiling.
+    private func streamIndex(
+        _ index: String,
+        continuation: AsyncThrowingStream<PluginStreamElement, Error>.Continuation
+    ) async throws {
+        guard let conn = connection else { throw ElasticsearchError.notConnected }
+        let mappingColumns = try await cachedMappingColumns(index)
+        let fields = ElasticsearchMappingFlattener.fieldInfo(from: mappingColumns)
+        let parsed = ElasticsearchParsedSearch(
+            index: index, from: 0, size: Self.deepPageBatchSize, sorts: [], filters: [], logicMode: "AND"
+        )
+
+        let pit = try await conn.openPointInTime(index: index, keepAlive: Self.pitKeepAlive)
+        defer { Task { await conn.closePointInTime(id: pit) } }
+
+        var columns: [String]?
+        var searchAfter: [Any]?
+        while true {
+            try Task.checkCancellation()
+            var body = ElasticsearchQueryBuilder.searchBody(
+                for: parsed, fields: fields, size: Self.deepPageBatchSize,
+                tiebreaker: true, searchAfter: searchAfter, supportsCaseInsensitive: supportsCaseInsensitiveSearch
+            )
+            body["pit"] = ["id": pit, "keep_alive": Self.pitKeepAlive]
+            let response = try await conn.search(index: nil, body: body)
+            let hits = extractHits(response)
+            guard !hits.isEmpty else { break }
+
+            /// The columns are settled once, from the mapping plus the first batch, because a
+            /// stream has one header and a later batch holding a field the first did not must not
+            /// widen it.
+            if columns == nil {
+                let resolved = ElasticsearchMappingFlattener.columns(forHits: hits, mappingColumns: mappingColumns)
+                columns = resolved
+                continuation.yield(.header(PluginStreamHeader(
+                    columns: resolved,
+                    columnTypeNames: resolved.map { exportTypeName($0, fields: fields) },
+                    estimatedRowCount: nil
+                )))
+            }
+            if let resolved = columns {
+                continuation.yield(.rows(ElasticsearchMappingFlattener.rows(forHits: hits, columns: resolved)))
+            }
+
+            searchAfter = hits.last?["sort"] as? [Any]
+            if hits.count < Self.deepPageBatchSize || searchAfter == nil { break }
+        }
+
+        if columns == nil {
+            let resolved = ElasticsearchMappingFlattener.columns(forHits: [], mappingColumns: mappingColumns)
+            continuation.yield(.header(PluginStreamHeader(
+                columns: resolved,
+                columnTypeNames: resolved.map { exportTypeName($0, fields: fields) },
+                estimatedRowCount: nil
+            )))
+        }
+    }
+
+    private func exportTypeName(_ column: String, fields: [String: ElasticsearchFieldInfo]) -> String {
+        switch column {
+        case ElasticsearchMappingFlattener.idColumn, ElasticsearchMappingFlattener.indexColumn:
+            return "keyword"
+        case ElasticsearchMappingFlattener.scoreColumn:
+            return "float"
+        default:
+            return fields[column]?.type ?? ""
+        }
+    }
+
+    private func defaultStreamRows(query: String) -> AsyncThrowingStream<PluginStreamElement, Error> {
+        AsyncThrowingStream(bufferingPolicy: .unbounded) { continuation in
+            let task = Task {
+                do {
+                    let result = try await self.execute(query: query)
+                    continuation.yield(.header(PluginStreamHeader(
+                        columns: result.columns,
+                        columnTypeNames: result.columnTypeNames,
+                        estimatedRowCount: nil
+                    )))
+                    if !result.rows.isEmpty { continuation.yield(.rows(result.rows)) }
+                    continuation.finish()
+                } catch {
+                    continuation.finish(throwing: error)
+                }
+            }
+            continuation.onTermination = { @Sendable _ in task.cancel() }
+        }
+    }
+
     // MARK: - Search
 
     private func executeSearch(

Plugins/ElasticsearchDriverPlugin/ElasticsearchPluginDriver.swift

@@ -202,6 +202,12 @@ internal final class ElasticsearchPluginDriver: PluginDatabaseDriver, @unchecked
         )
     }
 
+    /// Export reads the index through the driver's own paging rather than through a fabricated
+    /// `SELECT * FROM "<index>"`, which this driver rejects.
+    func defaultExportQuery(table: String) -> String? {
+        ElasticsearchOperations.encodeExport(index: table)
+    }
+
     // MARK: - Table Operations
 
     func dropObjectStatement(name: String, objectType: String, schema: String?, cascade: Bool) -> String? {