From 3afc206d712d4180ef0afd2993869dbc55ad2f6f Mon Sep 17 00:00:00 2001 From: Ngo Quoc Dat Date: Tue, 15 Sep 2026 19:17:55 +0700 Subject: [PATCH] feat(plugins): native delete and export for the engines with no SQL --- CHANGELOG.md | 4 + .../SQLDDLFallbackPolicy.swift | 52 ++++++++ .../WeaviateOperations.swift | 21 ++++ .../DynamoDBConnection.swift | 11 ++ .../DynamoDBOperations.swift | 52 ++++++++ .../DynamoDBPluginDriver.swift | 30 ++++- .../ElasticsearchOperations.swift | 22 ++++ .../ElasticsearchPluginDriver+Execution.swift | 114 ++++++++++++++++++ .../ElasticsearchPluginDriver.swift | 6 + Plugins/KafkaDriverPlugin/KafkaApiKey.swift | 5 + Plugins/KafkaDriverPlugin/KafkaCluster.swift | 12 ++ .../KafkaDeleteTopicsRequest.swift | 63 ++++++++++ Plugins/KafkaDriverPlugin/KafkaError.swift | 2 + .../KafkaDriverPlugin/KafkaPluginDriver.swift | 33 +++++ Plugins/KafkaDriverPlugin/KafkaQL.swift | 25 +++- .../WeaviatePluginDriver+Execution.swift | 74 ++++++++++++ .../WeaviatePluginDriver.swift | 6 + .../Core/Plugins/PluginDriverAdapter.swift | 1 + .../Database/SQLDDLFallbackPolicy.swift | 48 ++------ .../TableProMobile/Views/TableListView.swift | 16 ++- .../Helpers/SQLDDLFallbackPolicyTests.swift | 41 +++++++ .../Database/SQLDDLFallbackPolicyTests.swift | 1 + .../Plugins/DynamoDBOperationsTests.swift | 57 +++++++++ .../Plugins/KafkaIntegrationTests.swift | 40 ++++++ TableProTests/Plugins/KafkaQLTests.swift | 20 +++ docs/databases/dynamodb.mdx | 1 + docs/databases/elasticsearch.mdx | 2 + docs/databases/kafka.mdx | 5 + project.yml | 2 + 29 files changed, 724 insertions(+), 42 deletions(-) create mode 100644 Packages/TableProCore/Sources/TableProConnectionLibrary/SQLDDLFallbackPolicy.swift create mode 100644 Plugins/DynamoDBDriverPlugin/DynamoDBOperations.swift create mode 100644 Plugins/KafkaDriverPlugin/KafkaDeleteTopicsRequest.swift create mode 100644 TableProMobile/TableProMobileTests/Helpers/SQLDDLFallbackPolicyTests.swift create mode 100644 TableProTests/Plugins/DynamoDBOperationsTests.swift diff --git a/CHANGELOG.md b/CHANGELOG.md index ba027b98e5..4062745d03 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/Packages/TableProCore/Sources/TableProConnectionLibrary/SQLDDLFallbackPolicy.swift b/Packages/TableProCore/Sources/TableProConnectionLibrary/SQLDDLFallbackPolicy.swift new file mode 100644 index 0000000000..64f29e5f9e --- /dev/null +++ b/Packages/TableProCore/Sources/TableProConnectionLibrary/SQLDDLFallbackPolicy.swift @@ -0,0 +1,52 @@ +// +// SQLDDLFallbackPolicy.swift +// TableProConnectionLibrary +// + +import Foundation + +/// Whether an app may build `DROP ` or `TRUNCATE TABLE ` 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 = [ + "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) + } +} diff --git a/Packages/TableProCore/Sources/TableProWeaviateCore/WeaviateOperations.swift b/Packages/TableProCore/Sources/TableProWeaviateCore/WeaviateOperations.swift index e8867215fc..2c83f47e0b 100644 --- a/Packages/TableProCore/Sources/TableProWeaviateCore/WeaviateOperations.swift +++ b/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 ""`, 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 { diff --git a/Plugins/DynamoDBDriverPlugin/DynamoDBConnection.swift b/Plugins/DynamoDBDriverPlugin/DynamoDBConnection.swift index 5adebeb62d..b29ef7803f 100644 --- a/Plugins/DynamoDBDriverPlugin/DynamoDBConnection.swift +++ b/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, diff --git a/Plugins/DynamoDBDriverPlugin/DynamoDBOperations.swift b/Plugins/DynamoDBDriverPlugin/DynamoDBOperations.swift new file mode 100644 index 0000000000..fa7873ff94 --- /dev/null +++ b/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)\"" + } +} diff --git a/Plugins/DynamoDBDriverPlugin/DynamoDBPluginDriver.swift b/Plugins/DynamoDBDriverPlugin/DynamoDBPluginDriver.swift index 43d0e33cf6..fc394a6a75 100644 --- a/Plugins/DynamoDBDriverPlugin/DynamoDBPluginDriver.swift +++ b/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 { diff --git a/Plugins/ElasticsearchDriverPlugin/ElasticsearchOperations.swift b/Plugins/ElasticsearchDriverPlugin/ElasticsearchOperations.swift index 7fcb6afde6..d6b50da5b6 100644 --- a/Plugins/ElasticsearchDriverPlugin/ElasticsearchOperations.swift +++ b/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 ""`, 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 { diff --git a/Plugins/ElasticsearchDriverPlugin/ElasticsearchPluginDriver+Execution.swift b/Plugins/ElasticsearchDriverPlugin/ElasticsearchPluginDriver+Execution.swift index 066905155b..4ed925005d 100644 --- a/Plugins/ElasticsearchDriverPlugin/ElasticsearchPluginDriver+Execution.swift +++ b/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 { + 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.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 { + 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( diff --git a/Plugins/ElasticsearchDriverPlugin/ElasticsearchPluginDriver.swift b/Plugins/ElasticsearchDriverPlugin/ElasticsearchPluginDriver.swift index d3d923e014..3b8d119707 100644 --- a/Plugins/ElasticsearchDriverPlugin/ElasticsearchPluginDriver.swift +++ b/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 ""`, 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? { diff --git a/Plugins/KafkaDriverPlugin/KafkaApiKey.swift b/Plugins/KafkaDriverPlugin/KafkaApiKey.swift index 1813a86d80..02fdd4dd52 100644 --- a/Plugins/KafkaDriverPlugin/KafkaApiKey.swift +++ b/Plugins/KafkaDriverPlugin/KafkaApiKey.swift @@ -31,6 +31,7 @@ enum KafkaApiKey: Int16, CaseIterable, Sendable { case listGroups = 16 case saslHandshake = 17 case apiVersions = 18 + case deleteTopics = 20 case deleteRecords = 21 case describeConfigs = 32 case saslAuthenticate = 36 @@ -47,6 +48,7 @@ enum KafkaApiKey: Int16, CaseIterable, Sendable { case .listGroups: return "ListGroups" case .saslHandshake: return "SaslHandshake" case .apiVersions: return "ApiVersions" + case .deleteTopics: return "DeleteTopics" case .deleteRecords: return "DeleteRecords" case .describeConfigs: return "DescribeConfigs" case .saslAuthenticate: return "SaslAuthenticate" @@ -70,6 +72,7 @@ enum KafkaApiKey: Int16, CaseIterable, Sendable { case .describeGroups: return 5 case .listGroups: return 3 case .apiVersions: return 3 + case .deleteTopics: return 4 case .deleteRecords: return 2 case .describeConfigs: return 4 case .saslAuthenticate: return 2 @@ -92,6 +95,7 @@ enum KafkaApiKey: Int16, CaseIterable, Sendable { case .listGroups: return 4 case .saslHandshake: return 1 case .apiVersions: return 3 + case .deleteTopics: return 5 case .deleteRecords: return 2 case .describeConfigs: return 4 case .saslAuthenticate: return 2 @@ -112,6 +116,7 @@ enum KafkaApiKey: Int16, CaseIterable, Sendable { case .listGroups: return 0 case .saslHandshake: return 0 case .apiVersions: return 0 + case .deleteTopics: return 1 case .deleteRecords: return 0 case .describeConfigs: return 1 case .saslAuthenticate: return 0 diff --git a/Plugins/KafkaDriverPlugin/KafkaCluster.swift b/Plugins/KafkaDriverPlugin/KafkaCluster.swift index dc2bd7aba2..027cafca88 100644 --- a/Plugins/KafkaDriverPlugin/KafkaCluster.swift +++ b/Plugins/KafkaDriverPlugin/KafkaCluster.swift @@ -86,6 +86,18 @@ actor KafkaCluster { return bootstrapConnection } + /// The connection to the broker that is currently the controller. + /// + /// Only the controller accepts an admin request such as DeleteTopics; any other broker answers + /// NOT_CONTROLLER. On a single-broker cluster this is the bootstrap connection. Under + /// `bootstrapOnly` it has to be, and a cluster whose controller is elsewhere will say so + /// rather than have the client dial an address it cannot reach. + func controllerConnection() async throws -> KafkaConnection { + let metadata = try await metadata() + guard metadata.controllerId >= 0 else { return try controlConnection() } + return try await connection(forLeader: metadata.controllerId) + } + /// The connection to use for a partition whose leader is `nodeId`. /// /// Under `bootstrapOnly` this always returns the bootstrap connection. That can mean diff --git a/Plugins/KafkaDriverPlugin/KafkaDeleteTopicsRequest.swift b/Plugins/KafkaDriverPlugin/KafkaDeleteTopicsRequest.swift new file mode 100644 index 0000000000..f9dc96c708 --- /dev/null +++ b/Plugins/KafkaDriverPlugin/KafkaDeleteTopicsRequest.swift @@ -0,0 +1,63 @@ +import Foundation + +enum KafkaDeleteTopicsRequest { + /// How long the broker may spend on the deletion before answering REQUEST_TIMED_OUT. + /// + /// The broker marks the topic for deletion and answers; the log directories go later. A + /// timeout here therefore means "not accepted", not "half deleted". + static let timeoutMs: Int32 = 30_000 + + /// Deletes one topic, by name. + /// + /// Only the controller may accept this, so the request goes to the broker Metadata names as + /// controller rather than to whichever broker the client happens to hold. On a single-broker + /// cluster those are the same connection; on a real one they are not, and sending it anywhere + /// else answers NOT_CONTROLLER. + /// + /// Capped at v5 deliberately: v6 addresses topics by 16-byte UUID instead of by name, which is + /// a different request shape rather than a bigger one, the same line `KafkaApiKey` draws for + /// Fetch and Metadata. + static func deleteTopic(_ topic: String, cluster: KafkaCluster) async throws { + let connection = try await cluster.controllerConnection() + let version = try await connection.negotiatedVersion(for: .deleteTopics) + let flexible = KafkaApiKey.deleteTopics.isFlexible(version: version) + + let request = KafkaRequest(api: .deleteTopics, version: version) { writer, _ in + if flexible { + /// v4 and v5 still name the topic, as a compact string inside a compact array. + writer.compactArrayCount(1) + writer.compactString(topic) + } else { + writer.legacyArrayCount(1) + writer.legacyString(topic) + } + writer.int32(timeoutMs) + if flexible { writer.emptyTaggedFields() } + } + + var body = try await connection.send(request) + if version >= 1 { _ = try body.int32() } // throttleTimeMs + + let results = try body.array(compact: flexible) { reader -> Int16 in + _ = flexible ? try reader.nullableCompactString() : try reader.nullableLegacyString() + let errorCode = try reader.int16() + /// v5 added a broker-supplied message. It is read so the reader stays in step with the + /// wire, and discarded: `KafkaErrorCode.describe` is the app's own wording. + if version >= 5 { + _ = flexible ? try reader.nullableCompactString() : try reader.nullableLegacyString() + } + if flexible { try reader.taggedFields() } + return errorCode + } + if flexible { try body.taggedFields() } + + guard let errorCode = results.first else { + throw KafkaError.malformedResponse( + String(localized: "The broker answered a topic deletion with no result.") + ) + } + guard errorCode == KafkaErrorCode.none else { + throw KafkaError.broker(code: errorCode, api: KafkaApiKey.deleteTopics.name) + } + } +} diff --git a/Plugins/KafkaDriverPlugin/KafkaError.swift b/Plugins/KafkaDriverPlugin/KafkaError.swift index 9319cd12d1..b46baa6200 100644 --- a/Plugins/KafkaDriverPlugin/KafkaError.swift +++ b/Plugins/KafkaDriverPlugin/KafkaError.swift @@ -91,6 +91,8 @@ enum KafkaErrorCode { static let notCoordinator: Int16 = 16 static let illegalSaslState: Int16 = 34 static let unsupportedVersion: Int16 = 35 + static let notController: Int16 = 41 + static let topicDeletionDisabled: Int16 = 72 static let topicAuthorizationFailed: Int16 = 29 static let groupAuthorizationFailed: Int16 = 30 static let clusterAuthorizationFailed: Int16 = 31 diff --git a/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift b/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift index 8d70530152..f92bd3440d 100644 --- a/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift +++ b/Plugins/KafkaDriverPlugin/KafkaPluginDriver.swift @@ -218,6 +218,20 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { func quoteIdentifier(_ name: String) -> String { KafkaQL.quote(name) } + /// A topic delete, in KafkaQL, which is what the confirmation shows and what `execute` runs. + /// Kafka has no views, so any other object kind is not something this engine drops. + func dropObjectStatement(name: String, objectType: String, schema: String?, cascade: Bool) -> String? { + guard objectType.uppercased() == "TABLE", !name.isEmpty else { return nil } + return "DROP TOPIC \(KafkaQL.quote(name))" + } + + /// Kafka has no truncate. Deleting records up to an offset is per partition and leaves the + /// topic's retention to remove them, which is not what Truncate promises, so it stays + /// unoffered rather than offered as something else. + func truncateTableStatements(table: String, schema: String?, cascade: Bool) -> [String]? { + nil + } + func escapeStringLiteral(_ value: String) -> String { value.replacingOccurrences(of: "\"", with: "\\\"") } @@ -334,6 +348,8 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { return try await runDescribeGroup(group) case .describeTopic(let topic): return try await runDescribeTopic(topic) + case .dropTopic(let topic): + return try await runDropTopic(topic) case .showCluster: return try await runShowCluster() } @@ -502,6 +518,23 @@ final class KafkaPluginDriver: PluginDatabaseDriver, @unchecked Sendable { ) } + /// Deletes one topic through the controller. + /// + /// The broker accepts the deletion and removes the log directories afterwards, so a Metadata + /// read taken straight after this can still carry the topic. The row says what was asked for + /// rather than claiming the data is already gone. + private func runDropTopic(_ topic: String) async throws -> PluginQueryResult { + let started = Date() + try await KafkaDeleteTopicsRequest.deleteTopic(topic, cluster: cluster) + return PluginQueryResult( + columns: ["topic", "result"], + columnTypeNames: ["string", "string"], + rows: [[.text(topic), .text("deleting")]], + rowsAffected: 1, + executionTime: Date().timeIntervalSince(started) + ) + } + private func runDescribeTopic(_ topic: String) async throws -> PluginQueryResult { let metadata = try await cluster.metadata(topics: [topic]) let found = try metadata.requireTopic(named: topic) diff --git a/Plugins/KafkaDriverPlugin/KafkaQL.swift b/Plugins/KafkaDriverPlugin/KafkaQL.swift index 2f30f13846..04731d2a2f 100644 --- a/Plugins/KafkaDriverPlugin/KafkaQL.swift +++ b/Plugins/KafkaDriverPlugin/KafkaQL.swift @@ -42,6 +42,7 @@ enum KafkaStatement: Sendable { case describeGroup(String) case describeTopic(String) case showCluster + case dropTopic(String) } /// A small command language for the query editor. @@ -70,14 +71,36 @@ enum KafkaQL { return try parseShow(&tokens) case "DESCRIBE", "DESC": return try parseDescribe(&tokens) + case "DROP": + return try parseDrop(&tokens) default: throw KafkaError.syntax(String( - format: String(localized: "%@ is not a Kafka command. Try CONSUME, PRODUCE, SHOW or DESCRIBE."), + format: String(localized: "%@ is not a Kafka command. Try CONSUME, PRODUCE, SHOW, DESCRIBE or DROP."), head )) } } + // MARK: - DROP + + /// `DROP TOPIC `, and deliberately not `DROP TABLE`. + /// + /// `SELECT` is accepted as sugar for CONSUME because the habit transfers harmlessly, but a + /// Kafka topic is not a table and `DROP TABLE` is the text the app used to invent for engines + /// with no SQL (#2884). Accepting it here would make that mistake look supported. + private static func parseDrop(_ tokens: inout Tokenizer) throws -> KafkaStatement { + guard let kind = tokens.next()?.uppercased(), kind == "TOPIC" else { + throw KafkaError.syntax(String(localized: "DROP needs TOPIC and a topic name.")) + } + guard let topic = tokens.next() else { + throw KafkaError.syntax(String(localized: "DROP TOPIC needs a topic name.")) + } + guard tokens.next() == nil else { + throw KafkaError.syntax(String(localized: "DROP TOPIC takes one topic name.")) + } + return .dropTopic(unquote(topic)) + } + // MARK: - CONSUME private static func parseConsume(_ tokens: inout Tokenizer) throws -> KafkaConsumeQuery { diff --git a/Plugins/WeaviateDriverPlugin/WeaviatePluginDriver+Execution.swift b/Plugins/WeaviateDriverPlugin/WeaviatePluginDriver+Execution.swift index 55632941d2..c6e983719e 100644 --- a/Plugins/WeaviateDriverPlugin/WeaviatePluginDriver+Execution.swift +++ b/Plugins/WeaviateDriverPlugin/WeaviatePluginDriver+Execution.swift @@ -37,6 +37,80 @@ extension WeaviatePluginDriver { ) } + // MARK: - Export + + static let exportPageSize = 500 + + /// Exports one collection by walking it a page at a time, yielding each page instead of + /// holding the whole collection. The protocol default runs `execute` once and buffers + /// everything, which is fine for a grid page and not for an export. + func streamRows(query: String) -> AsyncThrowingStream { + guard let collectionName = WeaviateOperations.decodeExport(query) else { + return defaultStreamRows(query: query) + } + return AsyncThrowingStream(bufferingPolicy: .unbounded) { continuation in + let task = Task { + do { + try await self.streamCollection(collectionName, continuation: continuation) + continuation.finish() + } catch { + continuation.finish(throwing: error) + } + } + continuation.onTermination = { @Sendable _ in task.cancel() } + } + } + + private func streamCollection( + _ name: String, + continuation: AsyncThrowingStream.Continuation + ) async throws { + let client = try requireClient() + let collection = try await cachedCollection(name) + let columns = WeaviateSchema.columns(for: collection).map(\.name) + continuation.yield(.header(PluginStreamHeader( + columns: columns, + columnTypeNames: columns.map { typeName(for: $0, collection: collection) }, + estimatedRowCount: nil + ))) + + var offset = 0 + while true { + try Task.checkCancellation() + let objects = try await client.objects( + collection: name, limit: Self.exportPageSize, offset: offset, includeVector: true + ) + guard !objects.isEmpty else { break } + continuation.yield(.rows(objects.map { object in + WeaviateObjectCodec.row(for: object, columns: columns).map { value in + value.map(PluginCellValue.text) ?? .null + } + })) + if objects.count < Self.exportPageSize { break } + offset += objects.count + } + } + + private func defaultStreamRows(query: String) -> AsyncThrowingStream { + 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() } + } + } + private func executeSearch( _ query: String, client: WeaviateClient, diff --git a/Plugins/WeaviateDriverPlugin/WeaviatePluginDriver.swift b/Plugins/WeaviateDriverPlugin/WeaviatePluginDriver.swift index 2706908e0e..6a86ae47a7 100644 --- a/Plugins/WeaviateDriverPlugin/WeaviatePluginDriver.swift +++ b/Plugins/WeaviateDriverPlugin/WeaviatePluginDriver.swift @@ -109,6 +109,12 @@ internal final class WeaviatePluginDriver: PluginDatabaseDriver, @unchecked Send ) } + /// Export reads the collection through the driver's own paging rather than through a + /// fabricated `SELECT * FROM ""`, which this driver has no parser for. + func defaultExportQuery(table: String) -> String? { + WeaviateOperations.encodeExport(collection: table) + } + // MARK: - Table Operations func dropObjectStatement(name: String, objectType: String, schema: String?, cascade: Bool) -> String? { diff --git a/TablePro/Core/Plugins/PluginDriverAdapter.swift b/TablePro/Core/Plugins/PluginDriverAdapter.swift index 09e916a6eb..8d394a9a38 100644 --- a/TablePro/Core/Plugins/PluginDriverAdapter.swift +++ b/TablePro/Core/Plugins/PluginDriverAdapter.swift @@ -5,6 +5,7 @@ import Foundation import os +import TableProConnectionLibrary import TableProNumberFormatting import TableProPluginKit diff --git a/TablePro/Models/Database/SQLDDLFallbackPolicy.swift b/TablePro/Models/Database/SQLDDLFallbackPolicy.swift index d066285521..6016948689 100644 --- a/TablePro/Models/Database/SQLDDLFallbackPolicy.swift +++ b/TablePro/Models/Database/SQLDDLFallbackPolicy.swift @@ -4,45 +4,23 @@ // import Foundation +import TableProConnectionLibrary -/// Whether the app may build `DROP ` or `TRUNCATE TABLE ` itself when an -/// engine's plugin returned no statement of its own. +/// The Mac app's view of `TableProConnectionLibrary.SQLDDLFallbackPolicy`, in terms of its own +/// `DatabaseType`. /// -/// `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, and two tests hold it: one walks `DatabaseType.allKnownTypes` so a new engine -/// cannot inherit an answer by accident, and one reads the plugin sources so an engine whose plugin -/// declares a non-SQL editor language can never be missing from this list. -enum SQLDDLFallbackPolicy { - /// Engines with no SQL DDL to fall back on. - /// - /// Most of these have a plugin that answers both hooks, so the app never reaches the fallback - /// for them. They are listed anyway, because a plugin answers per object kind: Typesense - /// returns nil for anything that is not a collection, and without this the app would answer - /// that with `DROP VIEW`. - static let enginesWithoutSQLDDL: Set = [ - .elasticsearch, - .typesense, - .weaviate, - .kafka, - .etcd, - .redis, - .mongodb, - .surrealdb, - .dynamodb, - .beancount, - ] +/// The list itself lives in `TableProConnectionLibrary` because iOS hardcoded the same DDL in +/// `TableProMobile/Views/TableListView.swift` and needs the same answer, and that is one of the +/// five package targets both apps link. The two apps do not share `DatabaseType`, so the shared +/// list keys on the raw id and this maps it back. +extension SQLDDLFallbackPolicy { + /// Engines with no SQL DDL to fall back on, as this app's `DatabaseType`. + static var enginesWithoutSQLDDL: Set { + Set(engineIdsWithoutSQLDDL.map { DatabaseType(rawValue: $0) }) + } /// True when `DROP`/`TRUNCATE` built by the app is something this engine could run. static func allowsGeneratedDDL(for databaseType: DatabaseType) -> Bool { - !enginesWithoutSQLDDL.contains(databaseType) + allowsGeneratedDDL(databaseTypeId: databaseType.rawValue) } } diff --git a/TableProMobile/TableProMobile/Views/TableListView.swift b/TableProMobile/TableProMobile/Views/TableListView.swift index 04ae337137..8d36366bf3 100644 --- a/TableProMobile/TableProMobile/Views/TableListView.swift +++ b/TableProMobile/TableProMobile/Views/TableListView.swift @@ -1,4 +1,5 @@ import SwiftUI +import TableProConnectionLibrary import TableProDatabase import TableProModels @@ -13,6 +14,14 @@ struct TableListView: View { coordinator.supportsSchemas ? coordinator.activeSchema : nil } + /// Truncate and Drop write literal `TRUNCATE TABLE` / `DROP TABLE` below, so they are only + /// offered where that is a statement the engine could run. On Redis the rows are keys and the + /// driver tokenises the text as a Redis command, so `DROP TABLE "session:42"` came back as an + /// unknown command after promising a delete. + private var engineSpeaksSQLDDL: Bool { + SQLDDLFallbackPolicy.allowsGeneratedDDL(databaseTypeId: connection.type.rawValue) + } + /// Scoped to the connection: one shared key leaves a filter from another connection applied /// to a list that never shows it. @SceneStorage private var searchText: String @@ -78,7 +87,7 @@ struct TableListView: View { } let isView = table.type == .view || table.type == .materializedView - if !isView && !connection.safeModeLevel.blocksWrites { + if !isView && !connection.safeModeLevel.blocksWrites && engineSpeaksSQLDDL { Divider() Button(role: .destructive) { @@ -192,7 +201,6 @@ struct TableListView: View { Text(errorMessage) } } - } private struct TableRow: View { @@ -226,8 +234,8 @@ private struct TableRow: View { private func formatRowCount(_ count: Int) -> String { if count >= 1_000_000 { return String(format: "%.1fM", Double(count) / 1_000_000) - } else if count >= 1000 { - return String(format: "%.1fK", Double(count) / 1000) + } else if count >= 1_000 { + return String(format: "%.1fK", Double(count) / 1_000) } return "\(count)" } diff --git a/TableProMobile/TableProMobileTests/Helpers/SQLDDLFallbackPolicyTests.swift b/TableProMobile/TableProMobileTests/Helpers/SQLDDLFallbackPolicyTests.swift new file mode 100644 index 0000000000..2a46653d89 --- /dev/null +++ b/TableProMobile/TableProMobileTests/Helpers/SQLDDLFallbackPolicyTests.swift @@ -0,0 +1,41 @@ +import Foundation +import TableProConnectionLibrary +import TableProCoreTypes +import Testing + +@testable import TableProMobile + +@Suite("SQL DDL fallback policy on iOS") +struct SQLDDLFallbackPolicyIOSTests { + /// Redis rows are keys and its driver tokenises the query text as a Redis command, so a + /// `DROP TABLE "session:42"` came back as an unknown command after the menu had promised a + /// delete. It is the only engine iOS ships that has no SQL DDL. + @Test("Redis is the only iOS engine with no SQL DDL") + func redisIsTheOnlyNonSQLEngine() { + let withoutDDL = IOSDriverFactory().supportedTypes() + .filter { !SQLDDLFallbackPolicy.allowsGeneratedDDL(databaseTypeId: $0.rawValue) } + .map(\.rawValue) + .sorted() + #expect(withoutDDL == ["Redis"]) + } + + @Test("Every SQL engine iOS ships keeps Drop and Truncate") + func sqlEnginesKeepTheirOperations() { + for type in IOSDriverFactory().supportedTypes() where type != .redis { + #expect( + SQLDDLFallbackPolicy.allowsGeneratedDDL(databaseTypeId: type.rawValue), + "\(type.rawValue) should keep Drop and Truncate" + ) + } + } + + /// The list is shared through the kit precisely so the two apps cannot drift. This asserts the + /// ids are the raw values iOS actually uses, which is the only thing tying the string set to + /// this app's `DatabaseType`. + @Test("The shared ids match this app's own type constants") + func sharedIdsMatchLocalConstants() { + #expect(SQLDDLFallbackPolicy.engineIdsWithoutSQLDDL.contains(DatabaseType.redis.rawValue)) + #expect(!SQLDDLFallbackPolicy.engineIdsWithoutSQLDDL.contains(DatabaseType.postgresql.rawValue)) + #expect(!SQLDDLFallbackPolicy.engineIdsWithoutSQLDDL.contains(DatabaseType.sqlite.rawValue)) + } +} diff --git a/TableProTests/Models/Database/SQLDDLFallbackPolicyTests.swift b/TableProTests/Models/Database/SQLDDLFallbackPolicyTests.swift index 6d31399280..521d431d0b 100644 --- a/TableProTests/Models/Database/SQLDDLFallbackPolicyTests.swift +++ b/TableProTests/Models/Database/SQLDDLFallbackPolicyTests.swift @@ -5,6 +5,7 @@ import Foundation @testable import TablePro +import TableProConnectionLibrary import TableProPluginKit import Testing diff --git a/TableProTests/Plugins/DynamoDBOperationsTests.swift b/TableProTests/Plugins/DynamoDBOperationsTests.swift new file mode 100644 index 0000000000..2e67d2f36d --- /dev/null +++ b/TableProTests/Plugins/DynamoDBOperationsTests.swift @@ -0,0 +1,57 @@ +// +// DynamoDBOperationsTests.swift +// TableProTests +// + +import Foundation +@testable import TablePro +import Testing + +@Suite("DynamoDB table operations") +struct DynamoDBOperationsTests { + @Test("Dropping a table produces a statement the driver recognises") + func dropIsRecognised() throws { + let statement = try #require(DynamoDBOperations.dropTable(named: "orders", objectType: "TABLE")) + #expect(statement == "DROP TABLE \"orders\"") + #expect(DynamoDBOperations.droppedTableName(in: statement) == "orders") + } + + /// The statement is shown in the confirmation and then run, so a statement the driver's own + /// dispatch does not route would be #2884 in a third engine. + @Test("Every generated drop round-trips back to its table name", arguments: [ + "orders", "my-table", "my_table", "my.table", "Orders2024", + ]) + func dropRoundTrips(name: String) throws { + let statement = try #require(DynamoDBOperations.dropTable(named: name, objectType: "TABLE")) + #expect(DynamoDBOperations.droppedTableName(in: statement) == name) + } + + /// A DynamoDB table name is 3 to 255 characters of letters, digits, underscore, hyphen and dot. + /// A name holding anything else did not come from the table listing. + @Test("A name DynamoDB could not have is refused", arguments: [ + "", "ab", "has space", "has\"quote", "semi;colon", "star*", "slash/y", + ]) + func invalidNamesRefused(name: String) { + #expect(DynamoDBOperations.dropTable(named: name, objectType: "TABLE") == nil) + } + + @Test("Only a table is droppable", arguments: ["VIEW", "MATERIALIZED VIEW", "FOREIGN TABLE"]) + func onlyTablesAreDroppable(objectType: String) { + #expect(DynamoDBOperations.dropTable(named: "orders", objectType: objectType) == nil) + } + + @Test("Text that is not a drop statement routes nowhere", arguments: [ + "SELECT * FROM \"orders\"", "DROP TABLE orders", "DROP TABLE \"\"", "DELETE FROM \"orders\"", "", + ]) + func nonDropStatementsAreNotRouted(statement: String) { + #expect(DynamoDBOperations.droppedTableName(in: statement) == nil) + } + + /// PartiQL has no DDL, so this text is the driver's own vocabulary. It is spelled `DROP TABLE` + /// partly so the generic SQL classifier tiers it destructive without a DynamoDB arm. + @Test("A drop reads as destructive") + func dropClassifiesDestructive() throws { + let statement = try #require(DynamoDBOperations.dropTable(named: "orders", objectType: "TABLE")) + #expect(QueryClassifier.classify(statement, databaseType: .dynamodb).tier == .destructive) + } +} diff --git a/TableProTests/Plugins/KafkaIntegrationTests.swift b/TableProTests/Plugins/KafkaIntegrationTests.swift index a842ea66c0..114eee422f 100644 --- a/TableProTests/Plugins/KafkaIntegrationTests.swift +++ b/TableProTests/Plugins/KafkaIntegrationTests.swift @@ -367,6 +367,46 @@ struct KafkaIntegrationTests { #expect(ddl.contains("in-sync")) } + // MARK: - Topic deletion + + /// DeleteTopics is hand-written wire protocol, so the only way to trust it is to ask a real + /// broker. The broker accepts the request and removes the log directories afterwards, so this + /// waits for the topic to leave the listing rather than asserting it is gone immediately. + @Test("Dropping a topic removes it from the cluster") + func dropTopicRemovesIt() async throws { + let harness = try await KafkaTestBroker.harness(topic: "tp-it-drop", partitions: 2) + defer { harness.tearDown() } + + let before = try await harness.driver.fetchTables(schema: nil) + #expect(before.contains { $0.name == harness.topic }) + + let statement = try #require(harness.driver.dropObjectStatement( + name: harness.topic, objectType: "TABLE", schema: nil, cascade: false + )) + #expect(statement == "DROP TOPIC \(harness.quoted)") + + let result = try await harness.driver.execute(query: statement) + #expect(result.rowsAffected == 1) + + var listed = true + for _ in 0 ..< 40 where listed { + try await Task.sleep(nanoseconds: 250_000_000) + let tables = try await harness.driver.fetchTables(schema: nil) + listed = tables.contains { $0.name == harness.topic } + } + #expect(!listed, "the topic was still listed after the broker accepted the deletion") + } + + @Test("Dropping a topic that does not exist reports the broker's own answer") + func dropUnknownTopicReports() async throws { + let harness = try await KafkaTestBroker.harness(topic: "tp-it-drop-missing", partitions: 1) + defer { harness.tearDown() } + + await #expect(throws: (any Error).self) { + try await harness.driver.execute(query: "DROP TOPIC \"tp-it-no-such-topic-9e3f\"") + } + } + /// The lag report is what a Kafka debugging session is usually after. @Test("Consumer group lag is reported per partition") func consumerGroupLag() async throws { diff --git a/TableProTests/Plugins/KafkaQLTests.swift b/TableProTests/Plugins/KafkaQLTests.swift index e384a91096..77e403f8a2 100644 --- a/TableProTests/Plugins/KafkaQLTests.swift +++ b/TableProTests/Plugins/KafkaQLTests.swift @@ -158,6 +158,26 @@ struct KafkaQLTests { #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("PRODUCE INTO t") } #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("CONSUME \"unterminated") } #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("SHOW EVERYTHING") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("DROP") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("DROP TOPIC") } + #expect(throws: KafkaError.self) { _ = try KafkaQL.parse("DROP TOPIC a b") } + } + + /// A topic delete is KafkaQL's own verb. `DROP TABLE` stays a syntax error above, because a + /// topic is not a table and that text is what the app used to invent for engines with no SQL. + @Test("DROP TOPIC names the topic to delete") + func parsesDropTopic() throws { + guard case .dropTopic(let topic) = try KafkaQL.parse("DROP TOPIC orders") else { + Issue.record("expected a dropTopic statement") + return + } + #expect(topic == "orders") + + guard case .dropTopic(let quoted) = try KafkaQL.parse("DROP TOPIC \"my topic\"") else { + Issue.record("expected a dropTopic statement") + return + } + #expect(quoted == "my topic") } private func isShowTopics(_ statement: KafkaStatement) -> Bool { diff --git a/docs/databases/dynamodb.mdx b/docs/databases/dynamodb.mdx index c46eb8380a..c8b90a2ac7 100644 --- a/docs/databases/dynamodb.mdx +++ b/docs/databases/dynamodb.mdx @@ -100,6 +100,7 @@ For read-only access, drop `PartiQLInsert`, `PartiQLUpdate` and `PartiQLDelete`. - Paging is cursor-based, so page 40 re-scans everything before it. - A list or map cell is cut at 10,000 characters and ends in `...`. Saving an edit to a cut cell stores the fragment; change long nested values with PartiQL instead. - Table structure is fixed at creation: no structure editing, no transactions, no import. +- Truncate is not offered. DynamoDB empties a table by scanning it and deleting every item, which is a long billed job rather than a statement. Deleting the table is available, from the sidebar or with `DROP TABLE ""`, which issues `DeleteTable`. - Item counts come from DynamoDB and refresh roughly every six hours, so they lag. - DAX endpoints are not supported. Leave **Custom Endpoint** empty or point it at a standard endpoint. diff --git a/docs/databases/elasticsearch.mdx b/docs/databases/elasticsearch.mdx index 7467ade365..466b3b7316 100644 --- a/docs/databases/elasticsearch.mdx +++ b/docs/databases/elasticsearch.mdx @@ -47,6 +47,8 @@ Grid edits become REST calls keyed by `_id`: `POST /index/_update/{id}`, `PUT /i Deleting an index from the sidebar sends `DELETE /`, the request the confirmation shows. +Export reads the whole index with a point-in-time and `search_after`, so it is not capped at the 10,000-document result window. + ## Query DSL console diff --git a/docs/databases/kafka.mdx b/docs/databases/kafka.mdx index 35502ef869..7ebf28de2f 100644 --- a/docs/databases/kafka.mdx +++ b/docs/databases/kafka.mdx @@ -127,6 +127,11 @@ Editing a cell, deleting a row, and sorting a column are all unavailable, and th rather than failing at save time. Kafka has no update primitive, no per-message delete, and no server-side sort. To change what a consumer sees, produce a new message. +Deleting a topic is available, from the sidebar or with `DROP TOPIC `. The request goes to +the cluster controller, which accepts it and removes the log directories afterwards, so the topic +can still appear in the list for a moment. Truncate is not offered: Kafka removes records by +retention or by an offset per partition, neither of which empties a topic the way Truncate means. + Paging jumps are unavailable. Page two continues from where page one stopped, so a topic being written to while you read it will not repeat or skip a message. diff --git a/project.yml b/project.yml index 39c9494bf4..1f74cda891 100644 --- a/project.yml +++ b/project.yml @@ -422,6 +422,7 @@ targets: - Plugins/DuckDBDriverPlugin/DuckDBTypeRendering.swift - Plugins/DuckDBDriverPlugin/DuckDBViewDefinition.swift - Plugins/DuckDBDriverPlugin/QuackConnectBuilder.swift + - Plugins/DynamoDBDriverPlugin/DynamoDBOperations.swift - Plugins/DynamoDBDriverPlugin/DynamoDBQueryBuilder.swift - Plugins/DynamoDBDriverPlugin/DynamoDBStatementGenerator.swift - Plugins/ElasticsearchDriverPlugin/ElasticsearchConsoleParser.swift @@ -555,6 +556,7 @@ targets: - Plugins/KafkaDriverPlugin/KafkaBrowseEngine.swift - Plugins/KafkaDriverPlugin/KafkaCluster.swift - Plugins/KafkaDriverPlugin/KafkaConnection.swift + - Plugins/KafkaDriverPlugin/KafkaDeleteTopicsRequest.swift - Plugins/KafkaDriverPlugin/KafkaFetchRequest.swift - Plugins/KafkaDriverPlugin/KafkaGroupsRequest.swift - Plugins/KafkaDriverPlugin/KafkaMetadataRequest.swift