From d21c7e5c44d8e190f9c7e19de36ff55dea635c05 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nguy=E1=BB=85n=20Nam=20Long?= <26674308+J2TeamNNL@users.noreply.github.com> Date: Sun, 1 Nov 2026 10:02:21 +0700 Subject: [PATCH 1/4] feat(plugin-influxdb): add an InfluxDB 2 and 3 driver over HTTP SQL and InfluxQL on 3.x, Flux and InfluxQL on 2.x, chosen from the server version and the statement. Read-only grid with filters, paging and counts. Flux is tiered for Safe Mode. --- .github/plugin-registry.json | 3 + apple/Packages/TableProCore/Package.swift | 11 + .../TableProInfluxDBCore/InfluxCSV.swift | 79 +++++ .../TableProInfluxDBCore/InfluxError.swift | 43 +++ .../InfluxQueryLanguage.swift | 94 ++++++ .../TableProInfluxDBCore/InfluxRequest.swift | 219 ++++++++++++++ .../InfluxResponseDecoder.swift | 217 ++++++++++++++ .../InfluxResultTable.swift | 54 ++++ .../TableProInfluxDBCore/InfluxSchema.swift | 107 +++++++ .../InfluxSelectBuilder.swift | 221 ++++++++++++++ .../TableProInfluxDBCore/InfluxServer.swift | 49 ++++ .../InfluxQueryTests.swift | 208 +++++++++++++ .../InfluxResponseDecoderTests.swift | 93 ++++++ .../InfluxDBConnection.swift | 152 ++++++++++ .../InfluxDBDriverPlugin/InfluxDBPlugin.swift | 130 +++++++++ .../InfluxDBPluginDriver+Metadata.swift | 214 ++++++++++++++ .../InfluxDBPluginDriver.swift | 221 ++++++++++++++ apple/Plugins/InfluxDBDriverPlugin/Info.plist | 12 + .../influxdb-icon.imageset/Contents.json | 16 + .../influxdb-icon.imageset/influxdb.svg | 1 + ...ginMetadataRegistry+InfluxDBDefaults.swift | 95 ++++++ ...ginMetadataRegistry+RegistryDefaults.swift | 2 +- .../Utilities/SQL/QueryClassifier+Flux.swift | 96 ++++++ .../Core/Utilities/SQL/QueryClassifier.swift | 2 + .../Models/Connection/DatabaseType.swift | 1 + .../TablePro/Resources/Localizable.xcstrings | 40 +++ ...PluginMetadataRegistryTypeCountTests.swift | 3 +- .../SQL/QueryClassifierFluxTests.swift | 59 ++++ apple/project.yml | 10 + apple/scripts/probes/check-influxdb.sh | 274 ++++++++++++++++++ 30 files changed, 2724 insertions(+), 2 deletions(-) create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxCSV.swift create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxError.swift create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxQueryLanguage.swift create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxRequest.swift create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxResponseDecoder.swift create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxResultTable.swift create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxSchema.swift create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxSelectBuilder.swift create mode 100644 apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxServer.swift create mode 100644 apple/Packages/TableProCore/Tests/TableProInfluxDBCoreTests/InfluxQueryTests.swift create mode 100644 apple/Packages/TableProCore/Tests/TableProInfluxDBCoreTests/InfluxResponseDecoderTests.swift create mode 100644 apple/Plugins/InfluxDBDriverPlugin/InfluxDBConnection.swift create mode 100644 apple/Plugins/InfluxDBDriverPlugin/InfluxDBPlugin.swift create mode 100644 apple/Plugins/InfluxDBDriverPlugin/InfluxDBPluginDriver+Metadata.swift create mode 100644 apple/Plugins/InfluxDBDriverPlugin/InfluxDBPluginDriver.swift create mode 100644 apple/Plugins/InfluxDBDriverPlugin/Info.plist create mode 100644 apple/TablePro/Assets.xcassets/influxdb-icon.imageset/Contents.json create mode 100644 apple/TablePro/Assets.xcassets/influxdb-icon.imageset/influxdb.svg create mode 100644 apple/TablePro/Core/Plugins/PluginMetadataRegistry+InfluxDBDefaults.swift create mode 100644 apple/TablePro/Core/Utilities/SQL/QueryClassifier+Flux.swift create mode 100644 apple/TableProTests/Core/Utilities/SQL/QueryClassifierFluxTests.swift create mode 100755 apple/scripts/probes/check-influxdb.sh diff --git a/.github/plugin-registry.json b/.github/plugin-registry.json index f6bf29a547..d7edbab8e8 100644 --- a/.github/plugin-registry.json +++ b/.github/plugin-registry.json @@ -40,6 +40,9 @@ "HanaDriver": { "name": "SAP HANA Driver" }, + "InfluxDBDriverPlugin": { + "name": "InfluxDB Driver" + }, "KafkaDriverPlugin": { "name": "Kafka Driver" }, diff --git a/apple/Packages/TableProCore/Package.swift b/apple/Packages/TableProCore/Package.swift index 94fe92c270..e8796d7cb4 100644 --- a/apple/Packages/TableProCore/Package.swift +++ b/apple/Packages/TableProCore/Package.swift @@ -37,6 +37,7 @@ let package = Package( .library(name: "TableProGoogleCloud", targets: ["TableProGoogleCloud"]), .library(name: "TableProSpannerCore", targets: ["TableProSpannerCore"]), .library(name: "TableProWeaviateCore", targets: ["TableProWeaviateCore"]), + .library(name: "TableProInfluxDBCore", targets: ["TableProInfluxDBCore"]), .library(name: "TableProFirebaseCore", targets: ["TableProFirebaseCore"]), .library(name: "TableProBigtableCore", targets: ["TableProBigtableCore"]), .library(name: "TableProNumberFormatting", targets: ["TableProNumberFormatting"]), @@ -192,6 +193,11 @@ let package = Package( dependencies: [], path: "Sources/TableProWeaviateCore" ), + .target( + name: "TableProInfluxDBCore", + dependencies: [], + path: "Sources/TableProInfluxDBCore" + ), .target( name: "TableProBigtableCore", dependencies: ["TableProGoogleCloud"], @@ -295,6 +301,11 @@ let package = Package( dependencies: ["TableProTeradataCore"], path: "Tests/TableProTeradataCoreTests" ), + .testTarget( + name: "TableProInfluxDBCoreTests", + dependencies: ["TableProInfluxDBCore"], + path: "Tests/TableProInfluxDBCoreTests" + ), .testTarget( name: "TableProTrinoCoreTests", dependencies: ["TableProTrinoCore", "TableProSQLGrammar"], diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxCSV.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxCSV.swift new file mode 100644 index 0000000000..938f6b2202 --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxCSV.swift @@ -0,0 +1,79 @@ +import Foundation + +/// RFC 4180 records. A blank line comes back as an empty record because the Flux annotated CSV +/// uses it to separate one table from the next. +enum InfluxCSV { + static func records(_ text: String) -> [[String]] { + var records: [[String]] = [] + var record: [String] = [] + var field = "" + var inQuotes = false + var fieldStarted = false + var iterator = Array(text.unicodeScalars).makeIterator() + var pending: Unicode.Scalar? + + func next() -> Unicode.Scalar? { + if let scalar = pending { + pending = nil + return scalar + } + return iterator.next() + } + + func endRecord() { + if fieldStarted || !record.isEmpty { + record.append(field) + records.append(record) + } else { + records.append([]) + } + record = [] + field = "" + fieldStarted = false + } + + while let scalar = next() { + if inQuotes { + if scalar == "\"" { + if let following = next() { + if following == "\"" { + field.unicodeScalars.append("\"") + } else { + inQuotes = false + pending = following + } + } else { + inQuotes = false + } + } else { + field.unicodeScalars.append(scalar) + } + continue + } + switch scalar { + case "\"": + inQuotes = true + fieldStarted = true + case ",": + record.append(field) + field = "" + fieldStarted = true + case "\r": + if let following = next(), following != "\n" { + pending = following + } + endRecord() + case "\n": + endRecord() + default: + field.unicodeScalars.append(scalar) + fieldStarted = true + } + } + if fieldStarted || !record.isEmpty { + record.append(field) + records.append(record) + } + return records + } +} diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxError.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxError.swift new file mode 100644 index 0000000000..2d3681778d --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxError.swift @@ -0,0 +1,43 @@ +import Foundation + +public enum InfluxError: Error, Equatable, LocalizedError { + case notConnected + case invalidURL(String) + case unauthorized(String) + case server(status: Int, message: String) + case query(String) + case malformedResponse(String) + case missingDatabase + case missingOrganization + case unsupportedLanguage(InfluxQueryLanguage, generation: InfluxGeneration) + + public var errorDescription: String? { + switch self { + case .notConnected: + return String(localized: "Not connected to InfluxDB.") + case .invalidURL(let text): + return String(format: String(localized: "Invalid InfluxDB address: %@"), text) + case .unauthorized(let message): + return message.isEmpty + ? String(localized: "InfluxDB refused the token.") + : String(format: String(localized: "InfluxDB refused the token: %@"), message) + case .server(let status, let message): + return message.isEmpty ? "HTTP \(status)" : "HTTP \(status): \(message)" + case .query(let message): + return message + case .malformedResponse(let detail): + return String(format: String(localized: "Unexpected InfluxDB response: %@"), detail) + case .missingDatabase: + return String(localized: "Choose a database or bucket first.") + case .missingOrganization: + return String(localized: "InfluxDB 2 needs an organization. Set it in the connection settings.") + case .unsupportedLanguage(let language, let generation): + let available = generation.languages.map(\.displayName).joined(separator: ", ") + return String( + format: String(localized: "%@ is not available on this server. It accepts %@."), + language.displayName, + available + ) + } + } +} diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxQueryLanguage.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxQueryLanguage.swift new file mode 100644 index 0000000000..069220c585 --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxQueryLanguage.swift @@ -0,0 +1,94 @@ +import Foundation + +public enum InfluxQueryLanguage: String, Sendable, Equatable, CaseIterable { + case sql + case influxQL = "influxql" + case flux + + public var displayName: String { + switch self { + case .sql: return "SQL" + case .influxQL: return "InfluxQL" + case .flux: return "Flux" + } + } +} + +/// Picks the language a statement is sent in. The server version decides which languages exist; +/// within them the statement's own text decides, so the grid's `SELECT` and a typed Flux pipeline +/// both run without a setting to flip. A first-line comment naming a language (`-- influxql`, +/// `// flux`) overrides the guess. +public enum InfluxLanguageRouter { + public static func language( + for statement: String, + generation: InfluxGeneration + ) throws -> InfluxQueryLanguage { + let language = hint(in: statement) ?? guess(statement, generation: generation) + guard generation.languages.contains(language) else { + throw InfluxError.unsupportedLanguage(language, generation: generation) + } + return language + } + + static func hint(in statement: String) -> InfluxQueryLanguage? { + let firstLine = statement + .trimmingCharacters(in: .whitespacesAndNewlines) + .split(separator: "\n", maxSplits: 1, omittingEmptySubsequences: false) + .first + .map(String.init) ?? "" + let body: Substring + if firstLine.hasPrefix("--") { + body = firstLine.dropFirst(2) + } else if firstLine.hasPrefix("//") { + body = firstLine.dropFirst(2) + } else { + return nil + } + return InfluxQueryLanguage(rawValue: body.trimmingCharacters(in: .whitespaces).lowercased()) + } + + static func guess(_ statement: String, generation: InfluxGeneration) -> InfluxQueryLanguage { + let code = strippingLeadingComments(statement) + if looksLikeFlux(code) { + return .flux + } + switch generation { + case .v1, .v2: + return .influxQL + case .v3: + return looksLikeInfluxQLOnly(code) ? .influxQL : .sql + } + } + + static func looksLikeFlux(_ code: String) -> Bool { + if code.contains("|>") { return true } + let lowered = code.lowercased() + return ["from(", "import ", "buckets(", "option "].contains { lowered.hasPrefix($0) } + } + + private static let influxQLOnlyPrefixes = [ + "SHOW MEASUREMENT", "SHOW TAG", "SHOW FIELD", "SHOW SERIES", "SHOW RETENTION", "SHOW DATABASES", + ] + + /// Statements DataFusion SQL rejects but InfluxQL reads: its `SHOW` catalog commands, + /// `GROUP BY time(...)` buckets and duration literals such as `now() - 1h`. + static func looksLikeInfluxQLOnly(_ code: String) -> Bool { + let normalized = code.uppercased() + .components(separatedBy: .whitespacesAndNewlines) + .filter { !$0.isEmpty } + .joined(separator: " ") + if influxQLOnlyPrefixes.contains(where: normalized.hasPrefix) { return true } + if normalized.range(of: #"GROUP BY .*\bTIME\s*\("#, options: .regularExpression) != nil { return true } + return normalized.range(of: #"NOW\(\)\s*[-+]\s*\d+(NS|U|µ|MS|S|M|H|D|W)\b"#, options: .regularExpression) != nil + } + + static func strippingLeadingComments(_ statement: String) -> String { + var lines = statement.components(separatedBy: .newlines)[...] + while let first = lines.first { + let trimmed = first.trimmingCharacters(in: .whitespaces) + guard trimmed.isEmpty || trimmed.hasPrefix("--") || trimmed.hasPrefix("//") else { break } + lines = lines.dropFirst() + } + return lines.joined(separator: "\n").trimmingCharacters(in: .whitespacesAndNewlines) + } +} diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxRequest.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxRequest.swift new file mode 100644 index 0000000000..f807a0b24e --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxRequest.swift @@ -0,0 +1,219 @@ +import Foundation + +public struct InfluxRequest: Sendable, Equatable { + public let method: String + public let path: String + public let queryItems: [URLQueryItem] + public let headers: [String: String] + public let body: Data? + + public init( + method: String, + path: String, + queryItems: [URLQueryItem] = [], + headers: [String: String] = [:], + body: Data? = nil + ) { + self.method = method + self.path = path + self.queryItems = queryItems + self.headers = headers + self.body = body + } + + public func url(base: URL) -> URL? { + guard var components = URLComponents(url: base, resolvingAgainstBaseURL: false) else { return nil } + let basePath = components.path.hasSuffix("/") ? String(components.path.dropLast()) : components.path + components.path = basePath + path + components.queryItems = queryItems.isEmpty ? nil : queryItems + // `+` is legal in a query, but every InfluxDB line reads it as a space. + components.percentEncodedQuery = components.percentEncodedQuery? + .replacingOccurrences(of: "+", with: "%2B") + return components.url + } +} + +public enum InfluxCredentials: Sendable, Equatable { + case none + case token(String) + + public init(token: String?) { + let token = (token ?? "").trimmingCharacters(in: .whitespacesAndNewlines) + self = token.isEmpty ? .none : .token(token) + } + + /// 2.x reads `Token`, 3.x reads `Bearer`; both accept the other's word today, but the + /// documented one is what a proxy in front of them is most likely to pass through. + func authorization(for generation: InfluxGeneration?) -> String? { + switch self { + case .none: + return nil + case .token(let token): + return generation == .v3 ? "Bearer \(token)" : "Token \(token)" + } + } +} + +public enum InfluxRequests { + /// 3.x with auth on refuses an anonymous `/ping`; before the version is known the token goes + /// out with the `Token` word, which every line accepts. + public static func ping(credentials: InfluxCredentials) -> InfluxRequest { + InfluxRequest(method: "GET", path: "/ping", headers: authHeaders(credentials, generation: nil)) + } + + public static func listDatabases( + generation: InfluxGeneration, + organization: String, + credentials: InfluxCredentials, + offset: Int = 0 + ) -> InfluxRequest { + let headers = authHeaders(credentials, generation: generation) + switch generation { + case .v3: + return InfluxRequest( + method: "GET", + path: "/api/v3/configure/database", + queryItems: [URLQueryItem(name: "format", value: "json")], + headers: headers + ) + case .v2: + return InfluxRequest( + method: "GET", + path: "/api/v2/buckets", + queryItems: [ + URLQueryItem(name: "org", value: organization), + URLQueryItem(name: "limit", value: String(bucketPageSize)), + URLQueryItem(name: "offset", value: String(offset)), + ], + headers: headers + ) + case .v1: + return query("SHOW DATABASES", language: .influxQL, database: "", organization: organization, + generation: generation, credentials: credentials) + } + } + + public static let bucketPageSize = 100 + + public static func listDBRPs(organization: String, credentials: InfluxCredentials) -> InfluxRequest { + InfluxRequest( + method: "GET", + path: "/api/v2/dbrps", + queryItems: [URLQueryItem(name: "org", value: organization)], + headers: authHeaders(credentials, generation: .v2) + ) + } + + public static func listOrganizations(credentials: InfluxCredentials) -> InfluxRequest { + InfluxRequest(method: "GET", path: "/api/v2/orgs", headers: authHeaders(credentials, generation: .v2)) + } + + public static func query( + _ statement: String, + language: InfluxQueryLanguage, + database: String, + organization: String, + generation: InfluxGeneration, + credentials: InfluxCredentials + ) -> InfluxRequest { + var headers = authHeaders(credentials, generation: generation) + switch (generation, language) { + case (.v3, .sql), (.v3, .influxQL), (.v3, .flux): + headers["Content-Type"] = "application/json" + headers["Accept"] = "text/csv" + let payload: [String: String] = ["db": database, "q": statement, "format": "csv"] + return InfluxRequest( + method: "POST", + path: language == .sql ? "/api/v3/query_sql" : "/api/v3/query_influxql", + headers: headers, + body: try? JSONSerialization.data(withJSONObject: payload, options: [.sortedKeys]) + ) + case (_, .flux): + headers["Content-Type"] = "application/json" + headers["Accept"] = "application/csv" + let payload: [String: Any] = [ + "query": statement, + "type": "flux", + "dialect": ["annotations": ["datatype"], "header": true], + ] + return InfluxRequest( + method: "POST", + path: "/api/v2/query", + queryItems: [URLQueryItem(name: "org", value: organization)], + headers: headers, + body: try? JSONSerialization.data(withJSONObject: payload, options: [.sortedKeys]) + ) + case (_, .sql), (_, .influxQL): + headers["Content-Type"] = "application/x-www-form-urlencoded" + headers["Accept"] = "application/json" + var form = URLComponents() + form.queryItems = [URLQueryItem(name: "q", value: statement)] + if !database.isEmpty { + form.queryItems?.append(URLQueryItem(name: "db", value: database)) + } + let body = (form.percentEncodedQuery ?? "").replacingOccurrences(of: "+", with: "%2B") + return InfluxRequest(method: "POST", path: "/query", headers: headers, body: Data(body.utf8)) + } + } + + static func authHeaders(_ credentials: InfluxCredentials, generation: InfluxGeneration?) -> [String: String] { + guard let value = credentials.authorization(for: generation) else { return [:] } + return ["Authorization": value] + } +} + +public enum InfluxCatalogDecoder { + public static let systemDatabaseNames = ["_internal", "_monitoring", "_tasks"] + + /// `/api/v3/configure/database?format=json`: `[{"iox::database": "metrics"}]`. + public static func v3Databases(_ data: Data) throws -> [String] { + guard let rows = try? JSONSerialization.jsonObject(with: data) as? [[String: Any]] else { + throw InfluxError.malformedResponse(String(bytes: data.prefix(200), encoding: .utf8) ?? "") + } + return rows.compactMap { $0["iox::database"] as? String } + } + + /// `/api/v2/orgs`: `{"orgs": [{"name": "tp"}, ...]}`. + public static func v2Organizations(_ data: Data) throws -> [String] { + guard let root = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + let orgs = root["orgs"] as? [[String: Any]] + else { + throw InfluxError.malformedResponse(String(bytes: data.prefix(200), encoding: .utf8) ?? "") + } + return orgs.compactMap { $0["name"] as? String } + } + + /// `/api/v2/buckets`: `{"buckets": [{"id": "1ef8...", "name": "metrics"}, ...]}`, as id and name. + public static func v2Buckets(_ data: Data) throws -> [(id: String, name: String)] { + guard let root = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + let buckets = root["buckets"] as? [[String: Any]] + else { + throw InfluxError.malformedResponse(String(bytes: data.prefix(200), encoding: .utf8) ?? "") + } + return buckets.compactMap { bucket in + guard let name = bucket["name"] as? String else { return nil } + return (bucket["id"] as? String ?? "", name) + } + } + + /// `/api/v2/dbrps`: the InfluxQL database name each bucket answers to, keyed by bucket id. + /// InfluxQL resolves `db` through these mappings, not through bucket names. A mapping someone + /// made wins over the virtual one 2.x makes per bucket, and a default one over the rest. + public static func v2InfluxQLDatabases(_ data: Data) throws -> [String: String] { + guard let root = try? JSONSerialization.jsonObject(with: data) as? [String: Any], + let mappings = root["content"] as? [[String: Any]] + else { + throw InfluxError.malformedResponse(String(bytes: data.prefix(200), encoding: .utf8) ?? "") + } + func rank(_ mapping: [String: Any]) -> Int { + (mapping["default"] as? Bool == true ? 2 : 0) + (mapping["virtual"] as? Bool == true ? 0 : 1) + } + var best: [String: [String: Any]] = [:] + for mapping in mappings { + guard let bucket = mapping["bucketID"] as? String, mapping["database"] is String else { continue } + if let current = best[bucket], rank(current) >= rank(mapping) { continue } + best[bucket] = mapping + } + return best.compactMapValues { $0["database"] as? String } + } +} diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxResponseDecoder.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxResponseDecoder.swift new file mode 100644 index 0000000000..ece22ca03e --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxResponseDecoder.swift @@ -0,0 +1,217 @@ +import Foundation + +public enum InfluxResponseDecoder { + // MARK: - 3.x CSV + + /// `/api/v3/query_sql` and `/api/v3/query_influxql` with `format=csv`. CSV is the one 3.x format + /// that keeps the column order of the statement; JSON returns objects and drops null columns. + /// An empty cell reads as null, since the format has no other way to write one. + public static func decodeV3CSV(_ text: String) -> InfluxResultTable { + var records = InfluxCSV.records(text).filter { !$0.isEmpty } + guard !records.isEmpty else { return .empty } + let columns = records.removeFirst() + let rows: [[String?]] = records.map { record in + columns.indices.map { index in + guard index < record.count, !record[index].isEmpty else { return nil } + return record[index] + } + } + let typeNames = columns.enumerated().map { index, column in + InfluxTypeName.inferred(column: column, values: rows.map { $0[index] }) + } + let normalized = rows.map { row in + row.enumerated().map { index, value in + typeNames[index] == InfluxTypeName.timestamp + ? value.map(InfluxTypeName.normalizedTimestamp) + : value + } + } + return InfluxResultTable(columns: columns, typeNames: typeNames, rows: normalized) + } + + // MARK: - InfluxQL over /query (1.x and the 2.x compatibility API) + + /// Flattens every series of every statement into one grid. A statement that groups by tag + /// returns one series per tag set; the tags become leading columns, and when the series name + /// varies (`SHOW FIELD KEYS` over several measurements) a `name` column comes first. + public static func decodeV1JSON(_ data: Data) throws -> InfluxResultTable { + guard let root = try? JSONSerialization.jsonObject(with: data) as? [String: Any] else { + throw InfluxError.malformedResponse(String(bytes: data.prefix(200), encoding: .utf8) ?? "") + } + if let error = root["error"] as? String { + throw InfluxError.query(error) + } + let results = root["results"] as? [[String: Any]] ?? [] + if let error = results.compactMap({ $0["error"] as? String }).first { + throw InfluxError.query(error) + } + let series = results.flatMap { $0["series"] as? [[String: Any]] ?? [] } + guard !series.isEmpty else { return .empty } + + let names = Set(series.compactMap { $0["name"] as? String }) + let includesName = names.count > 1 + var tagKeys: [String] = [] + var valueColumns: [String] = [] + for item in series { + for key in (item["tags"] as? [String: Any] ?? [:]).keys.sorted() where !tagKeys.contains(key) { + tagKeys.append(key) + } + for column in item["columns"] as? [String] ?? [] where !valueColumns.contains(column) { + valueColumns.append(column) + } + } + valueColumns.removeAll { tagKeys.contains($0) } + let columns = (includesName ? ["name"] : []) + tagKeys + valueColumns + + var rows: [[String?]] = [] + var jsonTypes = [String?](repeating: nil, count: columns.count) + for item in series { + let seriesColumns = item["columns"] as? [String] ?? [] + let tags = item["tags"] as? [String: Any] ?? [:] + for values in item["values"] as? [[Any]] ?? [] { + var row = [String?](repeating: nil, count: columns.count) + var offset = 0 + if includesName { + row[0] = item["name"] as? String + offset = 1 + } + for (index, key) in tagKeys.enumerated() { + row[offset + index] = text(tags[key]) + } + for (index, column) in seriesColumns.enumerated() where index < values.count { + guard let target = columns.firstIndex(of: column) else { continue } + row[target] = text(values[index]) + if jsonTypes[target] == nil { + jsonTypes[target] = typeName(of: values[index]) + } + } + rows.append(row) + } + } + let typeNames = columns.enumerated().map { index, column -> String in + if column == InfluxTypeName.timeColumn { return InfluxTypeName.timestamp } + if tagKeys.contains(column) { return InfluxTypeName.tag } + return jsonTypes[index] ?? InfluxTypeName.string + } + return InfluxResultTable(columns: columns, typeNames: typeNames, rows: rows) + } + + // MARK: - Flux annotated CSV (2.x) + + /// Each Flux table arrives as its own block with its own header. The grid gets the union of + /// the headers in first-seen order, without the empty annotation column and `result`. + public static func decodeFluxCSV(_ text: String) throws -> InfluxResultTable { + var columns: [String] = [] + var typeNames: [String] = [] + var rows: [[String?]] = [] + + var datatypes: [String] = [] + var header: [String]? + for record in InfluxCSV.records(text) { + if record.isEmpty || record == [""] { + datatypes = [] + header = nil + continue + } + if let first = record.first, first.hasPrefix("#") { + if first == "#datatype" { datatypes = record } + continue + } + guard let currentHeader = header else { + header = record + if isFluxErrorHeader(record) { continue } + for (index, column) in record.enumerated() where keepsFluxColumn(column, index: index) { + if !columns.contains(column) { + columns.append(column) + typeNames.append(fluxTypeName(index < datatypes.count ? datatypes[index] : "")) + } + } + continue + } + if isFluxErrorHeader(currentHeader) { + throw InfluxError.query(record.count > 1 ? record[1] : "Flux error") + } + var row = [String?](repeating: nil, count: columns.count) + for (index, column) in currentHeader.enumerated() where keepsFluxColumn(column, index: index) { + guard index < record.count, let target = columns.firstIndex(of: column) else { continue } + row[target] = record[index].isEmpty ? nil : record[index] + } + rows.append(row) + } + for index in rows.indices where rows[index].count < columns.count { + rows[index] += [String?](repeating: nil, count: columns.count - rows[index].count) + } + return InfluxResultTable(columns: columns, typeNames: typeNames, rows: rows) + } + + private static func isFluxErrorHeader(_ header: [String]) -> Bool { + header.dropFirst().elementsEqual(["error", "reference"]) + } + + private static func keepsFluxColumn(_ column: String, index: Int) -> Bool { + !(index == 0 && column.isEmpty) && column != "result" + } + + static func fluxTypeName(_ annotation: String) -> String { + switch annotation { + case "dateTime:RFC3339", "dateTime:RFC3339Nano": return InfluxTypeName.timestamp + case "long": return InfluxTypeName.integer + case "unsignedLong": return InfluxTypeName.unsigned + case "double": return InfluxTypeName.float + case "boolean": return InfluxTypeName.boolean + case "duration": return "duration" + case "base64Binary": return "binary" + default: return InfluxTypeName.string + } + } + + // MARK: - Errors + + /// The message out of an error body: `{"message": ...}` on 2.x, `{"error": ...}` on 1.x and some + /// 3.x paths, plain text on others. + public static func errorMessage(from data: Data) -> String { + if let object = try? JSONSerialization.jsonObject(with: data) as? [String: Any] { + for key in ["message", "error", "data"] { + if let message = object[key] as? String, !message.isEmpty { return message } + } + } + return (String(bytes: data.prefix(2_000), encoding: .utf8) ?? "") + .trimmingCharacters(in: .whitespacesAndNewlines) + } + + // MARK: - Values + + static func text(_ value: Any?) -> String? { + switch value { + case nil, is NSNull: + return nil + case let string as String: + return string + case let number as NSNumber: + if CFGetTypeID(number) == CFBooleanGetTypeID() { + return number.boolValue ? "true" : "false" + } + return number.stringValue + case let other?: + if JSONSerialization.isValidJSONObject(other), + let data = try? JSONSerialization.data(withJSONObject: other, options: [.sortedKeys]) { + return String(bytes: data, encoding: .utf8) ?? "" + } + return "\(other)" + } + } + + static func typeName(of value: Any) -> String? { + switch value { + case is NSNull: + return nil + case is String: + return InfluxTypeName.string + case let number as NSNumber: + if CFGetTypeID(number) == CFBooleanGetTypeID() { return InfluxTypeName.boolean } + return CFNumberIsFloatType(number) ? InfluxTypeName.float : InfluxTypeName.integer + default: + return InfluxTypeName.string + } + } +} diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxResultTable.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxResultTable.swift new file mode 100644 index 0000000000..a7ce8d00d5 --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxResultTable.swift @@ -0,0 +1,54 @@ +import Foundation + +public struct InfluxResultTable: Sendable, Equatable { + public var columns: [String] + public var typeNames: [String] + public var rows: [[String?]] + + public init(columns: [String], typeNames: [String], rows: [[String?]]) { + self.columns = columns + self.typeNames = typeNames + self.rows = rows + } + + public static let empty = InfluxResultTable(columns: [], typeNames: [], rows: []) + + public func column(named name: String) -> [String?] { + guard let index = columns.firstIndex(of: name) else { return [] } + return rows.map { index < $0.count ? $0[index] : nil } + } +} + +public enum InfluxTypeName { + public static let timestamp = "timestamp" + public static let tag = "tag" + public static let float = "float" + public static let integer = "integer" + public static let unsigned = "unsigned" + public static let string = "string" + public static let boolean = "boolean" + + public static let timeColumn = "time" + + /// Names the text a column holds when the response carries no types (the 3.x CSV format): + /// integer, float or boolean when every value reads as one, else string. + static func inferred(column: String, values: [String?]) -> String { + if column == timeColumn { return timestamp } + let present = values.compactMap { $0 } + guard !present.isEmpty else { return string } + if present.allSatisfy({ Int64($0) != nil }) { return integer } + if present.allSatisfy({ UInt64($0) != nil }) { return unsigned } + if present.allSatisfy({ Double($0) != nil }) { return float } + if present.allSatisfy({ $0 == "true" || $0 == "false" }) { return boolean } + return string + } + + /// 3.x returns `2024-10-27T03:33:20` for a UTC timestamp. Adding the zone makes it read the + /// same as the 2.x value and parse as RFC 3339 when pasted back into a filter. + static func normalizedTimestamp(_ value: String) -> String { + guard value.count >= 19, value.dropFirst(10).first == "T" else { return value } + let time = value.dropFirst(11) + if time.contains("Z") || time.contains("+") || time.contains("-") { return value } + return value + "Z" + } +} diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxSchema.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxSchema.swift new file mode 100644 index 0000000000..e2eeedc8c3 --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxSchema.swift @@ -0,0 +1,107 @@ +import Foundation + +public struct InfluxColumn: Sendable, Equatable { + public enum Role: Sendable, Equatable { + case time + case tag + case field + } + + public let name: String + public let typeName: String + public let role: Role + + public init(name: String, typeName: String, role: Role) { + self.name = name + self.typeName = typeName + self.role = role + } + + public var isText: Bool { + role == .tag || typeName == InfluxTypeName.string + } +} + +/// The catalog statements behind the sidebar, and the mapping of what they return into columns. +/// Columns come back with `time` first and the rest by name, the order InfluxQL's `SELECT *` uses. +public enum InfluxSchema { + public static func measurementsStatement(generation: InfluxGeneration) -> (String, InfluxQueryLanguage) { + switch generation { + case .v3: + return ( + "SELECT table_name FROM information_schema.tables WHERE table_schema = 'iox' ORDER BY table_name", + .sql + ) + case .v1, .v2: + return ("SHOW MEASUREMENTS", .influxQL) + } + } + + public static func measurements(from table: InfluxResultTable, generation: InfluxGeneration) -> [String] { + let column = generation == .v3 ? "table_name" : "name" + return table.column(named: column).compactMap { $0 } + } + + public static func v3ColumnsStatement(measurement: String) -> String { + let literal = InfluxSelectBuilder.stringLiteral(measurement, language: .sql) + return "SELECT column_name, data_type FROM information_schema.columns " + + "WHERE table_schema = 'iox' AND table_name = \(literal)" + } + + public static func v3Columns(from table: InfluxResultTable) -> [InfluxColumn] { + let names = table.column(named: "column_name") + let types = table.column(named: "data_type") + let columns = zip(names, types).compactMap { name, arrowType -> InfluxColumn? in + guard let name else { return nil } + return column(name: name, arrowType: arrowType ?? "") + } + return ordered(columns) + } + + public static func v1TagKeysStatement(measurement: String) -> String { + "SHOW TAG KEYS FROM \(InfluxSelectBuilder.quoteIdentifier(measurement, language: .influxQL))" + } + + public static func v1FieldKeysStatement(measurement: String) -> String { + "SHOW FIELD KEYS FROM \(InfluxSelectBuilder.quoteIdentifier(measurement, language: .influxQL))" + } + + public static func v1Columns(tagKeys: InfluxResultTable, fieldKeys: InfluxResultTable) -> [InfluxColumn] { + var columns = [InfluxColumn(name: InfluxTypeName.timeColumn, typeName: InfluxTypeName.timestamp, role: .time)] + for tag in tagKeys.column(named: "tagKey").compactMap({ $0 }) { + columns.append(InfluxColumn(name: tag, typeName: InfluxTypeName.tag, role: .tag)) + } + let names = fieldKeys.column(named: "fieldKey") + let types = fieldKeys.column(named: "fieldType") + for (name, type) in zip(names, types) { + guard let name, !columns.contains(where: { $0.name == name }) else { continue } + columns.append(InfluxColumn(name: name, typeName: type ?? InfluxTypeName.string, role: .field)) + } + return ordered(columns) + } + + static func column(name: String, arrowType: String) -> InfluxColumn { + if arrowType.hasPrefix("Timestamp") { + return InfluxColumn(name: name, typeName: InfluxTypeName.timestamp, role: .time) + } + if arrowType.hasPrefix("Dictionary") { + return InfluxColumn(name: name, typeName: InfluxTypeName.tag, role: .tag) + } + let typeName: String + switch arrowType { + case "Float64", "Float32": typeName = InfluxTypeName.float + case "Int64", "Int32": typeName = InfluxTypeName.integer + case "UInt64", "UInt32": typeName = InfluxTypeName.unsigned + case "Boolean": typeName = InfluxTypeName.boolean + default: typeName = InfluxTypeName.string + } + return InfluxColumn(name: name, typeName: typeName, role: .field) + } + + static func ordered(_ columns: [InfluxColumn]) -> [InfluxColumn] { + let time = columns.filter { $0.name == InfluxTypeName.timeColumn } + let rest = columns.filter { $0.name != InfluxTypeName.timeColumn } + .sorted { $0.name < $1.name } + return time + rest + } +} diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxSelectBuilder.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxSelectBuilder.swift new file mode 100644 index 0000000000..3cc09b7336 --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxSelectBuilder.swift @@ -0,0 +1,221 @@ +import Foundation + +public struct InfluxFilter: Sendable, Equatable { + public let column: String + public let op: String + public let value: String + public let secondValue: String? + public let isCaseSensitive: Bool + public let isText: Bool + + public init( + column: String, + op: String, + value: String, + secondValue: String? = nil, + isCaseSensitive: Bool = true, + isText: Bool = false + ) { + self.column = column + self.op = op + self.value = value + self.secondValue = secondValue + self.isCaseSensitive = isCaseSensitive + self.isText = isText + } +} + +public struct InfluxSort: Sendable, Equatable { + public let column: String + public let ascending: Bool + + public init(column: String, ascending: Bool) { + self.column = column + self.ascending = ascending + } +} + +/// Writes the grid's `SELECT` in SQL (3.x) or InfluxQL (1.x, 2.x). The two share the shape +/// `SELECT ... FROM "m" WHERE ... ORDER BY time DESC LIMIT n OFFSET m`, so a query built before +/// the server version is known runs on either. They differ in literals, regex syntax and in what +/// InfluxQL leaves out: no `IN`, no `IS NULL`, and `ORDER BY time` only. +public enum InfluxSelectBuilder { + public static func select( + measurement: String, + columns: [String], + filters: [InfluxFilter] = [], + logicMode: String = "and", + sorts: [InfluxSort] = [], + limit: Int, + offset: Int, + language: InfluxQueryLanguage + ) -> String { + let projection = language == .sql && !columns.isEmpty + ? columns.map { quoteIdentifier($0, language: language) }.joined(separator: ", ") + : "*" + var statement = "SELECT \(projection) FROM \(quoteIdentifier(measurement, language: language))" + statement += whereClause(filters, logicMode: logicMode, language: language) + statement += " ORDER BY " + orderBy(sorts, language: language) + statement += " LIMIT \(max(limit, 0)) OFFSET \(max(offset, 0))" + return statement + } + + /// InfluxQL answers `COUNT(*)` with one `count_` column per field, which is the point + /// count only for a measurement with one field; the caller decides whether to ask. + public static func count( + measurement: String, + filters: [InfluxFilter] = [], + logicMode: String = "and", + language: InfluxQueryLanguage + ) -> String { + "SELECT COUNT(*) FROM \(quoteIdentifier(measurement, language: language))" + + whereClause(filters, logicMode: logicMode, language: language) + } + + public static func rowCount(from table: InfluxResultTable) -> Int? { + guard let row = table.rows.first else { return 0 } + let counts = zip(table.columns, row).compactMap { column, value -> Int? in + guard column.lowercased().hasPrefix("count"), let value else { return nil } + return Int(value) + } + return counts.max() + } + + static func whereClause(_ filters: [InfluxFilter], logicMode: String, language: InfluxQueryLanguage) -> String { + let conditions = filters.compactMap { condition(for: $0, language: language) } + guard !conditions.isEmpty else { return "" } + let joiner = logicMode.lowercased() == "or" ? " OR " : " AND " + return " WHERE " + conditions.joined(separator: joiner) + } + + static func orderBy(_ sorts: [InfluxSort], language: InfluxQueryLanguage) -> String { + let timeFirst = "time DESC" + switch language { + case .sql: + guard !sorts.isEmpty else { return timeFirst } + return sorts + .map { "\(quoteIdentifier($0.column, language: language)) \($0.ascending ? "ASC" : "DESC")" } + .joined(separator: ", ") + case .influxQL, .flux: + guard let time = sorts.first(where: { $0.column == InfluxTypeName.timeColumn }) else { + return timeFirst + } + return time.ascending ? "time ASC" : timeFirst + } + } + + // MARK: - Conditions + + static func condition(for filter: InfluxFilter, language: InfluxQueryLanguage) -> String? { + let column = quoteIdentifier(filter.column, language: language) + let op = filter.op.uppercased() + switch op { + case "=", "!=", "<>", "<", ">", "<=", ">=": + let sqlOp = op == "<>" ? "!=" : op + return "\(column) \(sqlOp) \(literal(filter.value, filter: filter, language: language))" + case "CONTAINS": + return regex(column, escapedRegex(filter.value), filter, language) + case "NOT CONTAINS": + return regex(column, escapedRegex(filter.value), filter, language, negated: true) + case "STARTS WITH": + return regex(column, "^" + escapedRegex(filter.value), filter, language) + case "ENDS WITH": + return regex(column, escapedRegex(filter.value) + "$", filter, language) + case "REGEX": + return regex(column, filter.value, filter, language) + case "IN", "NOT IN": + let values = filter.value + .split(separator: ",") + .map { $0.trimmingCharacters(in: .whitespaces) } + .filter { !$0.isEmpty } + guard !values.isEmpty else { return nil } + let negated = op == "NOT IN" + let parts = values.map { + "\(column) \(negated ? "!=" : "=") \(literal($0, filter: filter, language: language))" + } + return "(" + parts.joined(separator: negated ? " AND " : " OR ") + ")" + case "BETWEEN": + guard let upper = filter.secondValue else { return nil } + let low = literal(filter.value, filter: filter, language: language) + let high = literal(upper, filter: filter, language: language) + return "(\(column) >= \(low) AND \(column) <= \(high))" + case "IS NULL": + return language == .sql ? "\(column) IS NULL" : "\(column) = ''" + case "IS NOT NULL": + return language == .sql ? "\(column) IS NOT NULL" : "\(column) != ''" + case "IS EMPTY": + return "\(column) = ''" + case "IS NOT EMPTY": + return "\(column) != ''" + default: + return nil + } + } + + /// Escapes the metacharacters RE2 and the Rust regex crate share. `NSRegularExpression`'s + /// escaper also escapes `/`, which InfluxQL's `/.../` literal would then escape twice. + static func escapedRegex(_ text: String) -> String { + var escaped = "" + for character in text { + if "\\^$.|?*+()[]{}".contains(character) { escaped.append("\\") } + escaped.append(character) + } + return escaped + } + + private static func regex( + _ column: String, + _ pattern: String, + _ filter: InfluxFilter, + _ language: InfluxQueryLanguage, + negated: Bool = false + ) -> String { + let flagged = filter.isCaseSensitive ? pattern : "(?i)" + pattern + switch language { + case .sql: + return "\(column) \(negated ? "!~" : "~") \(stringLiteral(flagged, language: language))" + case .influxQL, .flux: + let delimited = flagged.replacingOccurrences(of: "/", with: "\\/") + return "\(column) \(negated ? "!~" : "=~") /\(delimited)/" + } + } + + static func literal(_ value: String, filter: InfluxFilter, language: InfluxQueryLanguage) -> String { + if filter.column == InfluxTypeName.timeColumn || filter.isText { + return stringLiteral(value, language: language) + } + if Double(value) != nil, value.rangeOfCharacter(from: .letters) == nil { + return value + } + if value == "true" || value == "false" { + return value + } + return stringLiteral(value, language: language) + } + + // MARK: - Quoting + + public static func quoteIdentifier(_ name: String, language: InfluxQueryLanguage) -> String { + switch language { + case .sql: + return "\"" + name.replacingOccurrences(of: "\"", with: "\"\"") + "\"" + case .influxQL, .flux: + let escaped = name + .replacingOccurrences(of: "\\", with: "\\\\") + .replacingOccurrences(of: "\"", with: "\\\"") + return "\"" + escaped + "\"" + } + } + + public static func stringLiteral(_ value: String, language: InfluxQueryLanguage) -> String { + switch language { + case .sql: + return "'" + value.replacingOccurrences(of: "'", with: "''") + "'" + case .influxQL, .flux: + let escaped = value + .replacingOccurrences(of: "\\", with: "\\\\") + .replacingOccurrences(of: "'", with: "\\'") + return "'" + escaped + "'" + } + } +} diff --git a/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxServer.swift b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxServer.swift new file mode 100644 index 0000000000..9be3276b4f --- /dev/null +++ b/apple/Packages/TableProCore/Sources/TableProInfluxDBCore/InfluxServer.swift @@ -0,0 +1,49 @@ +import Foundation + +public enum InfluxGeneration: Sendable, Equatable { + case v1 + case v2 + case v3 + + /// Reads the `X-Influxdb-Version` header `/ping` returns: `v2.9.1` on 2.x, `3.12.2` on 3.x, + /// `1.11.8` on 1.x. A server that sends none is taken as 3.x, the only line whose `/ping` + /// body also names the version. + public init(versionHeader: String?) { + let trimmed = (versionHeader ?? "").trimmingCharacters(in: .whitespaces) + let digits = trimmed.hasPrefix("v") ? String(trimmed.dropFirst()) : trimmed + switch digits.split(separator: ".").first.flatMap({ Int($0) }) { + case 1: self = .v1 + case 2: self = .v2 + default: self = .v3 + } + } + + public var languages: [InfluxQueryLanguage] { + switch self { + case .v1: return [.influxQL] + case .v2: return [.flux, .influxQL] + case .v3: return [.sql, .influxQL] + } + } + + /// The language the data grid's queries are written in: both read the same + /// `SELECT ... FROM "m" ORDER BY time DESC LIMIT n OFFSET m`. + public var gridLanguage: InfluxQueryLanguage { + self == .v3 ? .sql : .influxQL + } +} + +public struct InfluxServerInfo: Sendable, Equatable { + public let generation: InfluxGeneration + public let version: String + + public init(generation: InfluxGeneration, version: String) { + self.generation = generation + self.version = version + } + + public init(versionHeader: String?) { + let version = (versionHeader ?? "").trimmingCharacters(in: .whitespaces) + self.init(generation: InfluxGeneration(versionHeader: versionHeader), version: version) + } +} diff --git a/apple/Packages/TableProCore/Tests/TableProInfluxDBCoreTests/InfluxQueryTests.swift b/apple/Packages/TableProCore/Tests/TableProInfluxDBCoreTests/InfluxQueryTests.swift new file mode 100644 index 0000000000..ca4ed68175 --- /dev/null +++ b/apple/Packages/TableProCore/Tests/TableProInfluxDBCoreTests/InfluxQueryTests.swift @@ -0,0 +1,208 @@ +import Foundation +@testable import TableProInfluxDBCore +import Testing + +struct InfluxQueryTests { + @Test(arguments: [ + ("v2.9.1", InfluxGeneration.v2), + ("3.12.2", .v3), + ("1.11.8", .v1), + ("", .v3), + ]) + func generationComesFromTheVersionHeader(header: String, expected: InfluxGeneration) { + #expect(InfluxGeneration(versionHeader: header) == expected) + } + + @Test func v3RoutesSQLByDefaultAndInfluxQLOnlyStatementsToInfluxQL() throws { + #expect(try InfluxLanguageRouter.language(for: "SELECT * FROM cpu", generation: .v3) == .sql) + #expect(try InfluxLanguageRouter.language(for: "show measurements", generation: .v3) == .influxQL) + #expect(try InfluxLanguageRouter.language( + for: "SELECT mean(usage) FROM cpu WHERE time > now() - 1h GROUP BY time(5m)", generation: .v3 + ) == .influxQL) + #expect(try InfluxLanguageRouter.language(for: "SHOW TABLES", generation: .v3) == .sql) + } + + @Test func v2RoutesFluxPipelinesToFluxAndTheRestToInfluxQL() throws { + let flux = "// last hour\nfrom(bucket: \"metrics\")\n |> range(start: -1h)" + #expect(try InfluxLanguageRouter.language(for: flux, generation: .v2) == .flux) + #expect(try InfluxLanguageRouter.language(for: "SELECT * FROM cpu", generation: .v2) == .influxQL) + } + + @Test func aFirstLineCommentNamesTheLanguage() throws { + #expect(try InfluxLanguageRouter.language(for: "-- influxql\nSELECT * FROM cpu", generation: .v3) == .influxQL) + } + + @Test func aLanguageTheServerLacksIsRefused() { + #expect(throws: InfluxError.unsupportedLanguage(.flux, generation: .v3)) { + try InfluxLanguageRouter.language(for: "from(bucket: \"x\") |> range(start: -1h)", generation: .v3) + } + #expect(throws: InfluxError.unsupportedLanguage(.sql, generation: .v2)) { + try InfluxLanguageRouter.language(for: "-- sql\nSELECT 1", generation: .v2) + } + } + + @Test func browseQueryIsTheSameShapeInBothLanguages() { + let sql = InfluxSelectBuilder.select(measurement: "cpu", columns: [], limit: 100, offset: 200, language: .sql) + let influxQL = InfluxSelectBuilder.select( + measurement: "cpu", columns: [], limit: 100, offset: 200, language: .influxQL + ) + #expect(sql == #"SELECT * FROM "cpu" ORDER BY time DESC LIMIT 100 OFFSET 200"#) + #expect(influxQL == sql) + } + + @Test func sqlSelectsNamedColumnsAndSortsByAnyColumn() { + let query = InfluxSelectBuilder.select( + measurement: #"we"ird"#, columns: ["time", "host"], sorts: [InfluxSort(column: "host", ascending: true)], + limit: 10, offset: 0, language: .sql + ) + #expect(query == #"SELECT "time", "host" FROM "we""ird" ORDER BY "host" ASC LIMIT 10 OFFSET 0"#) + } + + @Test func influxQLSortsByTimeOnly() { + let byHost = InfluxSelectBuilder.select( + measurement: "cpu", columns: ["time", "host"], sorts: [InfluxSort(column: "host", ascending: true)], + limit: 10, offset: 0, language: .influxQL + ) + #expect(byHost.hasSuffix("ORDER BY time DESC LIMIT 10 OFFSET 0")) + let byTime = InfluxSelectBuilder.select( + measurement: "cpu", columns: [], sorts: [InfluxSort(column: "time", ascending: true)], + limit: 10, offset: 0, language: .influxQL + ) + #expect(byTime.contains("ORDER BY time ASC")) + } + + @Test func filtersRenderPerLanguage() { + let filters = [ + InfluxFilter(column: "host", op: "=", value: "a'b", isText: true), + InfluxFilter(column: "usage", op: ">=", value: "12.5"), + InfluxFilter(column: "note", op: "CONTAINS", value: "a/b", isCaseSensitive: false, isText: true), + InfluxFilter(column: "region", op: "IN", value: "eu, us", isText: true), + InfluxFilter(column: "time", op: "BETWEEN", value: "2024-01-01T00:00:00Z", secondValue: "2025-01-01T00:00:00Z"), + ] + let sql = InfluxSelectBuilder.select( + measurement: "cpu", columns: [], filters: filters, limit: 5, offset: 0, language: .sql + ) + #expect(sql.contains(#""host" = 'a''b'"#)) + #expect(sql.contains(#""usage" >= 12.5"#)) + #expect(sql.contains(#""note" ~ '(?i)a/b'"#)) + #expect(sql.contains(#"("region" = 'eu' OR "region" = 'us')"#)) + #expect(sql.contains(#"("time" >= '2024-01-01T00:00:00Z' AND "time" <= '2025-01-01T00:00:00Z')"#)) + + let influxQL = InfluxSelectBuilder.select( + measurement: "cpu", columns: [], filters: filters, logicMode: "or", limit: 5, offset: 0, language: .influxQL + ) + #expect(influxQL.contains(#""host" = 'a\'b'"#)) + #expect(influxQL.contains(#""note" =~ /(?i)a\/b/"#)) + #expect(influxQL.contains(" OR ")) + } + + @Test func countReadsTheLargestPerFieldCount() { + let influxQL = InfluxResultTable( + columns: ["time", "count_idle", "count_usage"], typeNames: [], rows: [["1970-01-01T00:00:00Z", "2", "3"]] + ) + #expect(InfluxSelectBuilder.rowCount(from: influxQL) == 3) + let sql = InfluxResultTable(columns: ["count(*)"], typeNames: [], rows: [["7"]]) + #expect(InfluxSelectBuilder.rowCount(from: sql) == 7) + #expect(InfluxSelectBuilder.rowCount(from: .empty) == 0) + let filter = InfluxFilter(column: "host", op: "=", value: "a", isText: true) + #expect(InfluxSelectBuilder.count(measurement: "cpu", filters: [filter], language: .sql) + == #"SELECT COUNT(*) FROM "cpu" WHERE "host" = 'a'"#) + } + + @Test func nullChecksUseEmptyTagsOnInfluxQL() { + let filter = InfluxFilter(column: "host", op: "IS NULL", value: "") + #expect(InfluxSelectBuilder.condition(for: filter, language: .sql) == #""host" IS NULL"#) + #expect(InfluxSelectBuilder.condition(for: filter, language: .influxQL) == #""host" = ''"#) + } + + @Test func v3QueryPostsJSONAskingForCSV() throws { + let request = InfluxRequests.query( + "SELECT 1", language: .sql, database: "metrics", organization: "", + generation: .v3, credentials: .token("t") + ) + #expect(request.path == "/api/v3/query_sql") + #expect(request.headers["Authorization"] == "Bearer t") + let body = try #require(request.body) + let payload = try JSONSerialization.jsonObject(with: body) as? [String: String] + #expect(payload == ["db": "metrics", "q": "SELECT 1", "format": "csv"]) + } + + @Test func v2InfluxQLPostsAFormWithAnEncodedPlus() throws { + let request = InfluxRequests.query( + "SELECT a+b FROM x", language: .influxQL, database: "metrics", organization: "tp", + generation: .v2, credentials: .token("t") + ) + #expect(request.path == "/query") + #expect(request.headers["Authorization"] == "Token t") + let body = String(bytes: try #require(request.body), encoding: .utf8) ?? "" + #expect(body == "q=SELECT%20a%2Bb%20FROM%20x&db=metrics") + } + + @Test func v2FluxNamesTheOrganization() throws { + let request = InfluxRequests.query( + "buckets()", language: .flux, database: "", organization: "my org", + generation: .v2, credentials: .token("t") + ) + let url = try #require(request.url(base: URL(string: "http://h:8086")!)) + #expect(url.absoluteString == "http://h:8086/api/v2/query?org=my%20org") + } + + @Test func pingCarriesTheTokenForAServerWithAuthOn() { + #expect(InfluxRequests.ping(credentials: .token("t")).headers["Authorization"] == "Token t") + #expect(InfluxRequests.ping(credentials: .none).headers.isEmpty) + } + + @Test func containsEscapesRegexMetacharacters() { + let filter = InfluxFilter(column: "host", op: "CONTAINS", value: "a.b(c)", isText: true) + #expect(InfluxSelectBuilder.condition(for: filter, language: .sql) == #""host" ~ 'a\.b\(c\)'"#) + } + + @Test func aBlankTokenSendsNoAuthorization() { + #expect(InfluxCredentials(token: " \n") == .none) + #expect(InfluxCredentials(token: "t").authorization(for: .v3) == "Bearer t") + } + + @Test func catalogReadsDatabasesAndBuckets() throws { + #expect(try InfluxCatalogDecoder.v3Databases(Data(#"[{"iox::database":"_internal"},{"iox::database":"metrics"}]"#.utf8)) + == ["_internal", "metrics"]) + let buckets = try InfluxCatalogDecoder.v2Buckets(Data(#"{"buckets":[{"id":"b1","name":"metrics"}]}"#.utf8)) + #expect(buckets.map(\.name) == ["metrics"]) + #expect(buckets.map(\.id) == ["b1"]) + } + + @Test func aMadeDefaultDBRPWinsOverTheVirtualOne() throws { + let json = """ + {"content":[\ + {"bucketID":"b1","database":"metrics","default":true,"virtual":true},\ + {"bucketID":"b1","database":"legacy","default":true,"virtual":false},\ + {"bucketID":"b1","database":"other","default":false,"virtual":false},\ + {"bucketID":"b2","database":"logs","default":true,"virtual":true}]} + """ + #expect(try InfluxCatalogDecoder.v2InfluxQLDatabases(Data(json.utf8)) == ["b1": "legacy", "b2": "logs"]) + } + + @Test func v3ColumnsMapArrowTypesAndPutTimeFirst() { + let table = InfluxResultTable( + columns: ["column_name", "data_type"], + typeNames: ["string", "string"], + rows: [ + ["host", "Dictionary(Int32, Utf8)"], ["idle", "Int64"], ["time", "Timestamp(ns)"], + ["usage", "Float64"], ["ok", "Boolean"], + ] + ) + let columns = InfluxSchema.v3Columns(from: table) + #expect(columns.map(\.name) == ["time", "host", "idle", "ok", "usage"]) + #expect(columns.map(\.typeName) == ["timestamp", "tag", "integer", "boolean", "float"]) + } + + @Test func v1ColumnsJoinTagAndFieldKeys() { + let tags = InfluxResultTable(columns: ["tagKey"], typeNames: ["string"], rows: [["region"], ["host"]]) + let fields = InfluxResultTable( + columns: ["fieldKey", "fieldType"], typeNames: ["string", "string"], + rows: [["usage", "float"], ["idle", "integer"]] + ) + let columns = InfluxSchema.v1Columns(tagKeys: tags, fieldKeys: fields) + #expect(columns.map(\.name) == ["time", "host", "idle", "region", "usage"]) + #expect(columns.first { $0.name == "host" }?.isText == true) + } +} diff --git a/apple/Packages/TableProCore/Tests/TableProInfluxDBCoreTests/InfluxResponseDecoderTests.swift b/apple/Packages/TableProCore/Tests/TableProInfluxDBCoreTests/InfluxResponseDecoderTests.swift new file mode 100644 index 0000000000..6a435818bb --- /dev/null +++ b/apple/Packages/TableProCore/Tests/TableProInfluxDBCoreTests/InfluxResponseDecoderTests.swift @@ -0,0 +1,93 @@ +import Foundation +@testable import TableProInfluxDBCore +import Testing + +struct InfluxResponseDecoderTests { + @Test func v3CSVKeepsStatementColumnOrderAndReadsEmptyAsNull() { + let csv = "usage,host,note,time\n12.5,a,,2024-10-27T03:33:20\n50.1,b,hot,2024-10-27T03:34:20\n" + let table = InfluxResponseDecoder.decodeV3CSV(csv) + #expect(table.columns == ["usage", "host", "note", "time"]) + #expect(table.typeNames == ["float", "string", "string", "timestamp"]) + #expect(table.rows[0] == ["12.5", "a", nil, "2024-10-27T03:33:20Z"]) + #expect(table.rows[1][2] == "hot") + } + + @Test func v3CSVReadsQuotedFieldsWithCommasQuotesAndNewlines() { + let csv = "name,note\r\nx,\"a, \"\"b\"\"\nc\"\r\n" + let table = InfluxResponseDecoder.decodeV3CSV(csv) + #expect(table.rows == [["x", "a, \"b\"\nc"]]) + } + + @Test func v3CSVInfersIntegerAndKeepsZonedTimestamps() { + let table = InfluxResponseDecoder.decodeV3CSV("idle,time\n80,2024-10-27T03:33:20.5+07:00\n") + #expect(table.typeNames == ["integer", "timestamp"]) + #expect(table.rows[0][1] == "2024-10-27T03:33:20.5+07:00") + } + + @Test func emptyV3BodyIsAnEmptyTable() { + #expect(InfluxResponseDecoder.decodeV3CSV("") == .empty) + } + + @Test func v1JSONFlattensOneSeries() throws { + let json = """ + {"results":[{"statement_id":0,"series":[{"name":"cpu","columns":["time","host","idle","note","usage"],\ + "values":[["2024-10-27T03:34:20Z","b",40,"hot",50.1],["2024-10-27T03:33:20Z","a",80,null,12.5]]}]}]} + """ + let table = try InfluxResponseDecoder.decodeV1JSON(Data(json.utf8)) + #expect(table.columns == ["time", "host", "idle", "note", "usage"]) + #expect(table.typeNames == ["timestamp", "string", "integer", "string", "float"]) + #expect(table.rows[1] == ["2024-10-27T03:33:20Z", "a", "80", nil, "12.5"]) + } + + @Test func v1JSONPutsGroupTagsAndVaryingNamesFirst() throws { + let json = """ + {"results":[{"statement_id":0,"series":[\ + {"name":"cpu","tags":{"host":"a"},"columns":["time","mean"],"values":[["t1",1.5]]},\ + {"name":"mem","tags":{"host":"b"},"columns":["time","mean"],"values":[["t1",true]]}]}]} + """ + let table = try InfluxResponseDecoder.decodeV1JSON(Data(json.utf8)) + #expect(table.columns == ["name", "host", "time", "mean"]) + #expect(table.typeNames[1] == "tag") + #expect(table.rows == [["cpu", "a", "t1", "1.5"], ["mem", "b", "t1", "true"]]) + } + + @Test func v1JSONStatementErrorThrows() { + let json = #"{"results":[{"statement_id":0,"error":"measurement not found"}]}"# + #expect(throws: InfluxError.query("measurement not found")) { + try InfluxResponseDecoder.decodeV1JSON(Data(json.utf8)) + } + } + + @Test func fluxCSVUnionsTableHeadersAndDropsAnnotationColumns() throws { + let csv = """ + #datatype,string,long,dateTime:RFC3339,string,long,double + ,result,table,_time,host,idle,usage + ,_result,0,2024-10-27T03:33:20Z,a,80,12.5 + + #datatype,string,long,dateTime:RFC3339,string,long,string + ,result,table,_time,host,idle,note + ,_result,1,2024-10-27T03:34:20Z,b,40,hot + + """ + let table = try InfluxResponseDecoder.decodeFluxCSV(csv) + #expect(table.columns == ["table", "_time", "host", "idle", "usage", "note"]) + #expect(table.typeNames == ["integer", "timestamp", "string", "integer", "float", "string"]) + #expect(table.rows == [ + ["0", "2024-10-27T03:33:20Z", "a", "80", "12.5", nil], + ["1", "2024-10-27T03:34:20Z", "b", "40", nil, "hot"], + ]) + } + + @Test func fluxErrorTableThrows() { + let csv = "#datatype,string,string\n,error,reference\n,\"runtime error: boom\",\n" + #expect(throws: InfluxError.query("runtime error: boom")) { + try InfluxResponseDecoder.decodeFluxCSV(csv) + } + } + + @Test func errorMessagePrefersMessageThenErrorThenText() { + #expect(InfluxResponseDecoder.errorMessage(from: Data(#"{"code":"invalid","message":"bad"}"#.utf8)) == "bad") + #expect(InfluxResponseDecoder.errorMessage(from: Data(#"{"error":"nope"}"#.utf8)) == "nope") + #expect(InfluxResponseDecoder.errorMessage(from: Data("SQL error: x\n".utf8)) == "SQL error: x") + } +} diff --git a/apple/Plugins/InfluxDBDriverPlugin/InfluxDBConnection.swift b/apple/Plugins/InfluxDBDriverPlugin/InfluxDBConnection.swift new file mode 100644 index 0000000000..4ee66b24b9 --- /dev/null +++ b/apple/Plugins/InfluxDBDriverPlugin/InfluxDBConnection.swift @@ -0,0 +1,152 @@ +import Foundation +import os +import TableProInfluxDBCore +import TableProPluginKit + +internal struct InfluxDBResponse { + let statusCode: Int + let headers: [AnyHashable: Any] + let data: Data + + var isSuccess: Bool { (200 ..< 300).contains(statusCode) } + + func header(_ name: String) -> String? { + headers.first { ($0.key as? String)?.caseInsensitiveCompare(name) == .orderedSame }?.value as? String + } +} + +internal final class InfluxDBConnection: NSObject, @unchecked Sendable { + private static let logger = Logger(subsystem: "com.TablePro", category: "InfluxDBConnection") + + private let lock = NSLock() + private var session: URLSession? + private var inFlight: [ObjectIdentifier: PluginURLSessionTaskBox] = [:] + + let baseURL: URL + let credentials: InfluxCredentials + private let skipTLSVerify: Bool + private let queryTimeout: HttpQueryTimeoutBox + + init(baseURL: URL, credentials: InfluxCredentials, skipTLSVerify: Bool, queryTimeout: HttpQueryTimeoutBox) { + self.baseURL = baseURL + self.credentials = credentials + self.skipTLSVerify = skipTLSVerify + self.queryTimeout = queryTimeout + } + + func open(connectTimeout: TimeInterval) { + let configuration = URLSessionConfiguration.ephemeral + configuration.timeoutIntervalForRequest = connectTimeout + configuration.timeoutIntervalForResource = HttpQueryTimeout.sessionResourceTimeout + let session = URLSession(configuration: configuration, delegate: self, delegateQueue: nil) + lock.withLock { self.session = session } + } + + func close() { + cancelAll() + lock.withLock { + session?.invalidateAndCancel() + session = nil + } + } + + func cancelAll() { + let boxes = lock.withLock { + let boxes = Array(inFlight.values) + inFlight.removeAll() + return boxes + } + boxes.forEach { $0.cancel() } + } + + func send(_ request: InfluxRequest, timeout: TimeInterval? = nil) async throws -> InfluxDBResponse { + guard let session = lock.withLock({ session }) else { throw InfluxError.notConnected } + guard let url = request.url(base: baseURL) else { + throw InfluxError.invalidURL(baseURL.absoluteString + request.path) + } + var urlRequest = URLRequest(url: url) + urlRequest.httpMethod = request.method + urlRequest.httpBody = request.body + urlRequest.timeoutInterval = timeout ?? queryTimeout.requestTimeoutInterval + for (name, value) in request.headers { + urlRequest.setValue(value, forHTTPHeaderField: name) + } + + let taskBox = PluginURLSessionTaskBox() + let key = ObjectIdentifier(taskBox) + lock.withLock { inFlight[key] = taskBox } + defer { lock.withLock { _ = inFlight.removeValue(forKey: key) } } + return try await withTaskCancellationHandler { + try await perform(urlRequest, session: session, taskBox: taskBox) + } onCancel: { + taskBox.cancel() + } + } + + private func perform( + _ request: URLRequest, + session: URLSession, + taskBox: PluginURLSessionTaskBox + ) async throws -> InfluxDBResponse { + try await withCheckedThrowingContinuation { continuation in + let task = session.dataTask(with: request) { data, response, error in + taskBox.finish() + if let error { + if (error as? URLError)?.code == .cancelled { + continuation.resume(throwing: CancellationError()) + } else { + continuation.resume(throwing: error) + } + return + } + guard let http = response as? HTTPURLResponse else { + continuation.resume(throwing: InfluxError.malformedResponse("Not an HTTP response")) + return + } + continuation.resume(returning: InfluxDBResponse( + statusCode: http.statusCode, + headers: http.allHeaderFields, + data: data ?? Data() + )) + } + taskBox.set(task) + task.resume() + } + } + + static func error(for response: InfluxDBResponse) -> InfluxError { + let message = InfluxResponseDecoder.errorMessage(from: response.data) + if response.statusCode == 401 || response.statusCode == 403 { + return .unauthorized(message) + } + return .server(status: response.statusCode, message: message) + } +} + +extension InfluxDBConnection: URLSessionTaskDelegate { + func urlSession( + _ session: URLSession, + didReceive challenge: URLAuthenticationChallenge, + completionHandler: @escaping (URLSession.AuthChallengeDisposition, URLCredential?) -> Void + ) { + guard skipTLSVerify, + challenge.protectionSpace.authenticationMethod == NSURLAuthenticationMethodServerTrust, + let trust = challenge.protectionSpace.serverTrust + else { + completionHandler(.performDefaultHandling, nil) + return + } + completionHandler(.useCredential, URLCredential(trust: trust)) + } + + // A redirect would carry the token to wherever it points. + func urlSession( + _ session: URLSession, + task: URLSessionTask, + willPerformHTTPRedirection response: HTTPURLResponse, + newRequest request: URLRequest, + completionHandler: @escaping (URLRequest?) -> Void + ) { + completionHandler(nil) + } +} diff --git a/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPlugin.swift b/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPlugin.swift new file mode 100644 index 0000000000..0e79c198c4 --- /dev/null +++ b/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPlugin.swift @@ -0,0 +1,130 @@ +import Foundation +import TableProInfluxDBCore +import TableProPluginKit + +enum InfluxDBFieldID { + static let token = "influxToken" + static let organization = "influxOrg" + static let skipTLSVerify = "influxSkipTLSVerify" +} + +final class InfluxDBPlugin: NSObject, TableProPlugin, DriverPlugin { + static let pluginName = "InfluxDB Driver" + static let pluginVersion = "1.0.0" + static let pluginDescription = "InfluxDB 2 and 3 over the HTTP API, with SQL, InfluxQL and Flux" + static let capabilities: [PluginCapability] = [.databaseDriver] + + static let databaseTypeId = "InfluxDB" + static let databaseDisplayName = "InfluxDB" + static let iconName = "influxdb-icon" + static let defaultPort = 8_086 + static let isDownloadable = true + + static let navigationModel: NavigationModel = .standard + static let pathFieldRole: PathFieldRole = .database + static let requiresAuthentication = false + static let brandColorHex = "#22ADF6" + static let queryLanguageName = "SQL" + static let editorLanguage: EditorLanguage = .sql + static let supportsForeignKeys = false + static let supportsSchemaEditing = false + static let supportsDatabaseSwitching = true + static let requiresReconnectForDatabaseSwitch = false + static let supportsImport = false + static let supportsExport = true + static let supportsSSH = true + static let supportsSSL = true + static let supportsReadOnlyMode = true + static let supportsHealthMonitor = true + static let supportsForeignKeyDisable = false + static let supportsAddColumn = false + static let supportsModifyColumn = false + static let supportsDropColumn = false + static let supportsAddIndex = false + static let supportsDropIndex = false + static let supportsModifyPrimaryKey = false + static let databaseGroupingStrategy: GroupingStrategy = .flat + static let defaultGroupName = "default" + static let tableEntityName = "Measurements" + static let containerEntityName = "Database" + static let systemDatabaseNames: [String] = InfluxCatalogDecoder.systemDatabaseNames + static let immutableColumns: [String] = [InfluxTypeName.timeColumn] + static let structureColumnFields: [StructureColumnField] = [.name, .type, .nullable, .comment] + static let sqlDialect: SQLDialectDescriptor? = nil + + static let columnTypesByCategory: [String: [String]] = influxDBColumnTypes + static let additionalConnectionFields: [ConnectionField] = influxDBConnectionFields() + static var statementCompletions: [CompletionEntry] { influxDBCompletions } + + func createDriver(config: DriverConnectionConfig) -> any PluginDatabaseDriver { + InfluxDBPluginDriver(config: config) + } +} + +func influxDBConnectionFields() -> [ConnectionField] { + [ + ConnectionField( + id: InfluxDBFieldID.token, + label: String(localized: "Token"), + placeholder: "API token", + fieldType: .secure, + section: .authentication, + hidesPassword: true + ).withHidesUsername(true), + ConnectionField( + id: InfluxDBFieldID.organization, + label: String(localized: "Organization"), + placeholder: "InfluxDB 2 only, optional with one organization", + section: .connection + ), + ConnectionField( + id: InfluxDBFieldID.skipTLSVerify, + label: String(localized: "Skip TLS Verification"), + defaultValue: "false", + fieldType: .toggle, + section: .advanced + ), + ] +} + +let influxDBCompletions: [CompletionEntry] = [ + CompletionEntry(label: "SHOW MEASUREMENTS", insertText: "SHOW MEASUREMENTS"), + CompletionEntry(label: "SHOW TAG KEYS", insertText: "SHOW TAG KEYS FROM \"measurement\""), + CompletionEntry(label: "SHOW FIELD KEYS", insertText: "SHOW FIELD KEYS FROM \"measurement\""), + CompletionEntry( + label: "SQL: last hour", + insertText: "SELECT * FROM \"measurement\" WHERE time >= now() - INTERVAL '1 hour' ORDER BY time DESC" + ), + CompletionEntry( + label: "SQL: 5 minute averages", + insertText: """ + SELECT date_bin(INTERVAL '5 minutes', time) AS bucket, avg("field") + FROM "measurement" + WHERE time >= now() - INTERVAL '1 hour' + GROUP BY bucket + ORDER BY bucket + """ + ), + CompletionEntry( + label: "InfluxQL: 5 minute averages", + insertText: "SELECT MEAN(\"field\") FROM \"measurement\" WHERE time > now() - 1h GROUP BY time(5m)" + ), + CompletionEntry( + label: "Flux: last hour", + insertText: """ + from(bucket: "bucket") + |> range(start: -1h) + |> filter(fn: (r) => r._measurement == "measurement") + |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value") + """ + ), +] + +let influxDBColumnTypes: [String: [String]] = [ + "Time": [InfluxTypeName.timestamp], + "Tag": [InfluxTypeName.tag], + "Field": [ + InfluxTypeName.float, InfluxTypeName.integer, InfluxTypeName.unsigned, + InfluxTypeName.string, InfluxTypeName.boolean, + ], +] diff --git a/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPluginDriver+Metadata.swift b/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPluginDriver+Metadata.swift new file mode 100644 index 0000000000..095761b056 --- /dev/null +++ b/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPluginDriver+Metadata.swift @@ -0,0 +1,214 @@ +import Foundation +import TableProInfluxDBCore +import TableProPluginKit + +extension InfluxDBPluginDriver { + func fetchDatabases() async throws -> [String] { + let (connection, server) = try requireConnection() + let organization = currentOrganization + switch server.generation { + case .v3: + let response = try await connection.send(InfluxRequests.listDatabases( + generation: .v3, organization: organization, credentials: connection.credentials + )) + guard response.isSuccess else { throw InfluxDBConnection.error(for: response) } + return try InfluxCatalogDecoder.v3Databases(response.data) + case .v2: + var buckets: [(id: String, name: String)] = [] + while true { + let response = try await connection.send(InfluxRequests.listDatabases( + generation: .v2, organization: organization, credentials: connection.credentials, + offset: buckets.count + )) + guard response.isSuccess else { throw InfluxDBConnection.error(for: response) } + let page = try InfluxCatalogDecoder.v2Buckets(response.data) + buckets += page + if page.count < InfluxRequests.bucketPageSize { break } + } + let mappings = try await connection.send( + InfluxRequests.listDBRPs(organization: organization, credentials: connection.credentials) + ) + guard mappings.isSuccess else { throw InfluxDBConnection.error(for: mappings) } + let databaseByBucketID = try InfluxCatalogDecoder.v2InfluxQLDatabases(mappings.data) + remember(influxQLDatabases: Dictionary( + buckets.compactMap { bucket in databaseByBucketID[bucket.id].map { (bucket.name, $0) } }, + uniquingKeysWith: { first, _ in first } + )) + return buckets.map(\.name).sorted() + case .v1: + return try await run("SHOW DATABASES", language: .influxQL).column(named: "name").compactMap { $0 } + } + } + + func fetchDatabaseMetadata(_ database: String) async throws -> PluginDatabaseMetadata { + PluginDatabaseMetadata(name: database) + } + + func fetchTables(schema: String?) async throws -> [PluginTableInfo] { + guard currentDatabase != nil else { return [] } + let (_, server) = try requireConnection() + let (statement, language) = InfluxSchema.measurementsStatement(generation: server.generation) + let table = try await run(statement, language: language) + return InfluxSchema.measurements(from: table, generation: server.generation) + .map { PluginTableInfo(name: $0, type: "TABLE") } + } + + func fetchColumns(table: String, schema: String?) async throws -> [PluginColumnInfo] { + try await columns(of: table).map { column in + PluginColumnInfo( + name: column.name, + dataType: column.typeName, + isNullable: column.role != .time, + isPrimaryKey: column.role == .time, + comment: column.role == .tag ? "tag" : nil + ) + } + } + + func columns(of measurement: String) async throws -> [InfluxColumn] { + if let cached = cachedColumns(measurement) { return cached } + let (_, server) = try requireConnection() + let columns: [InfluxColumn] + switch server.generation { + case .v3: + let table = try await run(InfluxSchema.v3ColumnsStatement(measurement: measurement), language: .sql) + columns = InfluxSchema.v3Columns(from: table) + case .v1, .v2: + async let tags = run(InfluxSchema.v1TagKeysStatement(measurement: measurement), language: .influxQL) + async let fields = run(InfluxSchema.v1FieldKeysStatement(measurement: measurement), language: .influxQL) + columns = InfluxSchema.v1Columns(tagKeys: try await tags, fieldKeys: try await fields) + } + cache(columns, for: measurement) + return columns + } + + func fetchIndexes(table: String, schema: String?) async throws -> [PluginIndexInfo] { + [] + } + + func fetchForeignKeys(table: String, schema: String?) async throws -> [PluginForeignKeyInfo] { + [] + } + + /// InfluxDB has no DDL: a measurement exists once a point is written to it. The definition + /// tab shows its columns and their roles instead. + func fetchTableDDL(table: String, schema: String?) async throws -> String { + let columns = try await columns(of: table) + let width = columns.map(\.name.count).max() ?? 0 + let lines = columns.map { column -> String in + let role: String + switch column.role { + case .time: role = "time" + case .tag: role = "tag" + case .field: role = "field" + } + let padding = String(repeating: " ", count: width - column.name.count) + return "-- \(column.name)\(padding) \(role) \(column.typeName)" + } + return (["-- Measurement \(table)"] + lines).joined(separator: "\n") + } + + func fetchViewDefinition(view: String, schema: String?) async throws -> String { + throw InfluxError.query(String(localized: "InfluxDB has no views.")) + } + + func fetchTableMetadata(table: String, schema: String?) async throws -> PluginTableMetadata { + PluginTableMetadata(tableName: table, engine: "InfluxDB") + } + + // MARK: - Grid + + func buildBrowseQuery( + table: String, + sortColumns: [(columnIndex: Int, ascending: Bool)], + columns: [String], + limit: Int, + offset: Int + ) -> String? { + InfluxSelectBuilder.select( + measurement: table, + columns: columns, + sorts: sorts(sortColumns, columns: columns), + limit: limit, + offset: offset, + language: gridLanguage + ) + } + + func buildFilteredQuery( + table: String, + schema: String?, + queryFilters: [PluginQueryFilter], + logicMode: String, + sortColumns: [(columnIndex: Int, ascending: Bool)], + columns: [String], + limit: Int, + offset: Int, + columnKinds: [String: PluginColumnKind] + ) -> String? { + InfluxSelectBuilder.select( + measurement: table, + columns: columns, + filters: filters(queryFilters, table: table, columnKinds: columnKinds), + logicMode: logicMode, + sorts: sorts(sortColumns, columns: columns), + limit: limit, + offset: offset, + language: gridLanguage + ) + } + + func fetchExactRowCount( + table: String, + schema: String?, + queryFilters: [PluginQueryFilter], + logicMode: String + ) async throws -> Int? { + let language = gridLanguage + // InfluxQL counts per field, and a point may hold any subset of them, so only a measurement + // with one field has a count that is the number of points. + if language == .influxQL { + let fields = try await columns(of: table).filter { $0.role == .field } + guard fields.count == 1 else { return nil } + } + let statement = InfluxSelectBuilder.count( + measurement: table, + filters: filters(queryFilters, table: table, columnKinds: [:]), + logicMode: logicMode, + language: language + ) + return InfluxSelectBuilder.rowCount(from: try await run(statement, language: language)) + } + + func defaultExportQuery(table: String) -> String? { + let language = gridLanguage + return "SELECT * FROM \(InfluxSelectBuilder.quoteIdentifier(table, language: language))" + } + + private func sorts(_ sortColumns: [(columnIndex: Int, ascending: Bool)], columns: [String]) -> [InfluxSort] { + sortColumns.compactMap { sort in + guard columns.indices.contains(sort.columnIndex) else { return nil } + return InfluxSort(column: columns[sort.columnIndex], ascending: sort.ascending) + } + } + + private func filters( + _ queryFilters: [PluginQueryFilter], + table: String, + columnKinds: [String: PluginColumnKind] + ) -> [InfluxFilter] { + let known = cachedColumns(table) ?? [] + return queryFilters.map { filter in + let isText = known.first { $0.name == filter.column }?.isText + ?? (columnKinds[filter.column] == .text) + return InfluxFilter( + column: filter.column, + op: filter.op, + value: filter.value, + secondValue: filter.secondValue, + isCaseSensitive: filter.isCaseSensitive, + isText: isText + ) + } + } +} diff --git a/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPluginDriver.swift b/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPluginDriver.swift new file mode 100644 index 0000000000..275542f93f --- /dev/null +++ b/apple/Plugins/InfluxDBDriverPlugin/InfluxDBPluginDriver.swift @@ -0,0 +1,221 @@ +import Foundation +import os +import TableProInfluxDBCore +import TableProPluginKit + +internal final class InfluxDBPluginDriver: PluginDatabaseDriver, @unchecked Sendable { + static let logger = Logger(subsystem: "com.TablePro", category: "InfluxDBPluginDriver") + + private let config: DriverConnectionConfig + private let lock = NSLock() + private var connection: InfluxDBConnection? + private var server: InfluxServerInfo? + private var organization: String + private var database: String + private var columnCache: [String: [InfluxColumn]] = [:] + private var influxQLDatabases: [String: String] = [:] + let queryTimeout = HttpQueryTimeoutBox() + + init(config: DriverConnectionConfig) { + self.config = config + self.organization = (config.additionalFields[InfluxDBFieldID.organization] ?? "") + .trimmingCharacters(in: .whitespaces) + self.database = config.database.trimmingCharacters(in: .whitespaces) + } + + var serverVersion: String? { lock.withLock { server?.version } } + var supportsTransactions: Bool { false } + var capabilities: PluginCapabilities { [.cancelQuery] } + var currentDatabase: String? { lock.withLock { database.isEmpty ? nil : database } } + + func beginTransaction() async throws {} + func commitTransaction() async throws {} + func rollbackTransaction() async throws {} + + // MARK: - Lifecycle + + func connect() async throws { + let scheme = config.ssl.isEnabled ? "https" : "http" + let host = config.host.isEmpty ? "localhost" : config.host + let port = config.port > 0 ? config.port : InfluxDBPlugin.defaultPort + let address = "\(scheme)://\(host.contains(":") ? "[\(host)]" : host):\(port)" + guard let baseURL = URL(string: address) else { throw InfluxError.invalidURL(address) } + + let credentials = InfluxCredentials(token: config.additionalFields[InfluxDBFieldID.token]) + let skipTLS = config.additionalFields[InfluxDBFieldID.skipTLSVerify] == "true" + || (config.ssl.isEnabled && !config.ssl.verifiesCertificate) + let connectMilliseconds = PluginConnectTimeout.milliseconds( + in: config.additionalFields, + default: Int(HttpQueryTimeout.sessionBootstrapRequestTimeout * 1_000) + ) + let deadline = PluginConnectDeadline(milliseconds: connectMilliseconds) + let connection = InfluxDBConnection( + baseURL: baseURL, + credentials: credentials, + skipTLSVerify: skipTLS, + queryTimeout: queryTimeout + ) + connection.open(connectTimeout: TimeInterval(connectMilliseconds) / 1_000) + + do { + let ping = try await connection.send( + InfluxRequests.ping(credentials: credentials), + timeout: deadline.remainingSeconds() + ) + guard ping.isSuccess else { throw InfluxDBConnection.error(for: ping) } + let server = InfluxServerInfo(versionHeader: ping.header("X-Influxdb-Version")) + lock.withLock { + self.connection = connection + self.server = server + columnCache.removeAll() + } + if server.generation == .v2, lock.withLock({ organization.isEmpty }) { + try await adoptOnlyOrganization(connection, timeout: deadline.remainingSeconds()) + } + // `/ping` answers without a token, so the catalog call is what proves the token works. + _ = try await fetchDatabases() + } catch { + lock.withLock { + self.connection = nil + self.server = nil + } + connection.close() + throw error + } + } + + func disconnect() { + let current = lock.withLock { + let current = connection + connection = nil + server = nil + columnCache.removeAll() + return current + } + current?.close() + } + + func ping() async throws { + let (connection, server) = try requireConnection() + if server.generation == .v1 { + let response = try await connection.send(InfluxRequests.ping(credentials: connection.credentials)) + guard response.isSuccess else { throw InfluxDBConnection.error(for: response) } + return + } + _ = try await fetchDatabases() + } + + func cancelQuery() throws { + lock.withLock { connection }?.cancelAll() + } + + func applyQueryTimeout(_ seconds: Int) async throws { + queryTimeout.set(serverTimeoutSeconds: seconds) + } + + func switchDatabase(to database: String) async throws { + lock.withLock { + self.database = database + columnCache.removeAll() + } + } + + // MARK: - Queries + + func execute(query: String) async throws -> PluginQueryResult { + let started = Date() + let trimmed = query.trimmingCharacters(in: .whitespacesAndNewlines) + if trimmed.lowercased() == "select 1" { + try await ping() + return PluginQueryResult( + columns: ["ok"], columnTypeNames: [InfluxTypeName.integer], rows: [[.text("1")]], + rowsAffected: 0, executionTime: Date().timeIntervalSince(started) + ) + } + let (_, server) = try requireConnection() + let language = try InfluxLanguageRouter.language(for: trimmed, generation: server.generation) + let table = try await run(trimmed, language: language) + return PluginQueryResult( + columns: table.columns, + columnTypeNames: table.typeNames, + rows: table.rows.map { row in row.map { $0.map(PluginCellValue.text) ?? .null } }, + rowsAffected: 0, + executionTime: Date().timeIntervalSince(started) + ) + } + + func run(_ statement: String, language: InfluxQueryLanguage) async throws -> InfluxResultTable { + let (connection, server) = try requireConnection() + let (database, organization) = lock.withLock { (self.database, self.organization) } + if server.generation == .v3, database.isEmpty { + throw InfluxError.missingDatabase + } + if language == .flux, organization.isEmpty { + throw InfluxError.missingOrganization + } + // On 2.x InfluxQL names a DBRP mapping, which may differ from the bucket the sidebar shows. + let target = server.generation == .v2 && language == .influxQL + ? lock.withLock { influxQLDatabases[database] } ?? database + : database + let request = InfluxRequests.query( + statement, + language: language, + database: target, + organization: organization, + generation: server.generation, + credentials: connection.credentials + ) + let response = try await connection.send(request) + guard response.isSuccess else { throw InfluxDBConnection.error(for: response) } + switch (server.generation, language) { + case (.v3, _): + return InfluxResponseDecoder.decodeV3CSV(String(bytes: response.data, encoding: .utf8) ?? "") + case (_, .flux): + return try InfluxResponseDecoder.decodeFluxCSV(String(bytes: response.data, encoding: .utf8) ?? "") + default: + return try InfluxResponseDecoder.decodeV1JSON(response.data) + } + } + + func requireConnection() throws -> (InfluxDBConnection, InfluxServerInfo) { + guard let pair = lock.withLock({ connection.flatMap { conn in server.map { (conn, $0) } } }) else { + throw InfluxError.notConnected + } + return pair + } + + var currentOrganization: String { + lock.withLock { organization } + } + + var gridLanguage: InfluxQueryLanguage { + lock.withLock { server?.generation.gridLanguage } ?? .sql + } + + func remember(influxQLDatabases mapping: [String: String]) { + lock.withLock { influxQLDatabases = mapping } + } + + func cachedColumns(_ measurement: String) -> [InfluxColumn]? { + lock.withLock { columnCache[measurement] } + } + + func cache(_ columns: [InfluxColumn], for measurement: String) { + lock.withLock { columnCache[measurement] = columns } + } + + // MARK: - Organization + + /// 2.x scopes buckets and Flux to an organization. A token that sees exactly one needs no + /// setting, which is the common case for a self-hosted server. + private func adoptOnlyOrganization(_ connection: InfluxDBConnection, timeout: TimeInterval) async throws { + let response = try await connection.send( + InfluxRequests.listOrganizations(credentials: connection.credentials), + timeout: timeout + ) + guard response.isSuccess else { throw InfluxDBConnection.error(for: response) } + let names = try InfluxCatalogDecoder.v2Organizations(response.data) + guard names.count == 1, let only = names.first else { throw InfluxError.missingOrganization } + lock.withLock { organization = only } + } +} diff --git a/apple/Plugins/InfluxDBDriverPlugin/Info.plist b/apple/Plugins/InfluxDBDriverPlugin/Info.plist new file mode 100644 index 0000000000..a472fa3961 --- /dev/null +++ b/apple/Plugins/InfluxDBDriverPlugin/Info.plist @@ -0,0 +1,12 @@ + + + + + TableProPluginKitVersion + TABLEPRO_PLUGINKIT_VERSION + TableProProvidesDatabaseTypeIds + + InfluxDB + + + diff --git a/apple/TablePro/Assets.xcassets/influxdb-icon.imageset/Contents.json b/apple/TablePro/Assets.xcassets/influxdb-icon.imageset/Contents.json new file mode 100644 index 0000000000..178aadb462 --- /dev/null +++ b/apple/TablePro/Assets.xcassets/influxdb-icon.imageset/Contents.json @@ -0,0 +1,16 @@ +{ + "images": [ + { + "filename": "influxdb.svg", + "idiom": "universal" + } + ], + "info": { + "author": "xcode", + "version": 1 + }, + "properties": { + "preserves-vector-representation": true, + "template-rendering-intent": "template" + } +} diff --git a/apple/TablePro/Assets.xcassets/influxdb-icon.imageset/influxdb.svg b/apple/TablePro/Assets.xcassets/influxdb-icon.imageset/influxdb.svg new file mode 100644 index 0000000000..152eaa8286 --- /dev/null +++ b/apple/TablePro/Assets.xcassets/influxdb-icon.imageset/influxdb.svg @@ -0,0 +1 @@ +InfluxDB \ No newline at end of file diff --git a/apple/TablePro/Core/Plugins/PluginMetadataRegistry+InfluxDBDefaults.swift b/apple/TablePro/Core/Plugins/PluginMetadataRegistry+InfluxDBDefaults.swift new file mode 100644 index 0000000000..3a951f614d --- /dev/null +++ b/apple/TablePro/Core/Plugins/PluginMetadataRegistry+InfluxDBDefaults.swift @@ -0,0 +1,95 @@ +import Foundation +import TableProPluginKit + +extension PluginMetadataRegistry { + func influxDBPluginDefaults() -> [(typeId: String, snapshot: PluginMetadataSnapshot)] { + [ + ("InfluxDB", PluginMetadataSnapshot( + displayName: "InfluxDB", iconName: "influxdb-icon", defaultPort: 8_086, + requiresAuthentication: false, supportsForeignKeys: false, supportsSchemaEditing: false, + isDownloadable: true, primaryUrlScheme: "", parameterStyle: .questionMark, + navigationModel: .standard, explainVariants: [], pathFieldRole: .database, + supportsHealthMonitor: true, urlSchemes: [], postConnectActions: [], + brandColorHex: "#22ADF6", + queryLanguageName: "SQL", editorLanguage: .sql, + connectionMode: .network, supportsDatabaseSwitching: true, + capabilities: PluginMetadataSnapshot.CapabilityFlags( + supportsSchemaSwitching: false, + supportsImport: false, + supportsExport: true, + supportsSSH: true, + supportsSSL: true, + supportsCascadeDrop: false, + supportsForeignKeyDisable: false, + supportsReadOnlyMode: true, + supportsQueryProgress: false, + requiresReconnectForDatabaseSwitch: false, + supportsDropDatabase: false, + supportsAddColumn: false, + supportsModifyColumn: false, + supportsDropColumn: false, + supportsAddIndex: false, + supportsDropIndex: false, + supportsModifyPrimaryKey: false, + supportsOpportunisticTLS: false, + supportsPrincipalConnectionLimit: false + ), + schema: PluginMetadataSnapshot.SchemaInfo( + defaultSchemaName: "", + defaultGroupName: "default", + tableEntityName: "Measurements", + containerEntityName: "Database", + defaultPrimaryKeyColumn: nil, + immutableColumns: ["time"], + systemDatabaseNames: ["_internal", "_monitoring", "_tasks"], + systemSchemaNames: [], + fileExtensions: [], + databaseGroupingStrategy: .flat, + structureColumnFields: [.name, .type, .nullable, .comment] + ), + editor: PluginMetadataSnapshot.EditorConfig( + sqlDialect: nil, + statementCompletions: [], + columnTypesByCategory: [ + "Time": ["timestamp"], + "Tag": ["tag"], + "Field": ["float", "integer", "unsigned", "string", "boolean"] + ] + ), + connection: PluginMetadataSnapshot.ConnectionConfig( + additionalConnectionFields: influxDBConnectionFields(), + category: .analytical, + tagline: String(localized: "Time series database"), + hidesBuiltInPassword: true, + hidesBuiltInDatabase: false + ) + )) + ] + } +} + +func influxDBConnectionFields() -> [ConnectionField] { + [ + ConnectionField( + id: "influxToken", + label: String(localized: "Token"), + placeholder: "API token", + fieldType: .secure, + section: .authentication, + hidesPassword: true + ).withHidesUsername(true), + ConnectionField( + id: "influxOrg", + label: String(localized: "Organization"), + placeholder: "InfluxDB 2 only, optional with one organization", + section: .connection + ), + ConnectionField( + id: "influxSkipTLSVerify", + label: String(localized: "Skip TLS Verification"), + defaultValue: "false", + fieldType: .toggle, + section: .advanced + ) + ] +} diff --git a/apple/TablePro/Core/Plugins/PluginMetadataRegistry+RegistryDefaults.swift b/apple/TablePro/Core/Plugins/PluginMetadataRegistry+RegistryDefaults.swift index 1df518f949..5629cf0adf 100644 --- a/apple/TablePro/Core/Plugins/PluginMetadataRegistry+RegistryDefaults.swift +++ b/apple/TablePro/Core/Plugins/PluginMetadataRegistry+RegistryDefaults.swift @@ -1046,7 +1046,7 @@ extension PluginMetadataRegistry { + cloudPluginDefaults() + elasticsearchPluginDefaults() + surrealDBPluginDefaults() + kafkaPluginDefaults() + typesensePluginDefaults() + r2SQLPluginDefaults() + weaviatePluginDefaults() + hanaPluginDefaults() + firebasePluginDefaults() + bigtablePluginDefaults() - + trinoPluginDefaults() + + trinoPluginDefaults() + influxDBPluginDefaults() } // swiftlint:enable function_body_length } diff --git a/apple/TablePro/Core/Utilities/SQL/QueryClassifier+Flux.swift b/apple/TablePro/Core/Utilities/SQL/QueryClassifier+Flux.swift new file mode 100644 index 0000000000..5123e4f91e --- /dev/null +++ b/apple/TablePro/Core/Utilities/SQL/QueryClassifier+Flux.swift @@ -0,0 +1,96 @@ +import Foundation + +/// InfluxDB 2 reads Flux as well as InfluxQL. The SQL reading tiers `from(bucket: ...)` as an unknown +/// statement, so a read-only connection would refuse every Flux read. A Flux script only reads unless +/// it calls a function that writes points or sends data out; InfluxQL and SQL fall through to the SQL +/// reading. +extension QueryClassifier { + static func influxDBClassification(_ sql: String) -> QueryClassification? { + let code = fluxCode(sql) + switch languageHint(sql) { + case "flux": break + case "influxql", "sql": return nil + default: guard looksLikeFlux(code) else { return nil } + } + let writes = fluxWritingCalls.contains { code.range(of: $0, options: .regularExpression) != nil } + let withStrings = fluxCode(sql, keepingStrings: true) + let reachesOutside = fluxOutsideCalls.contains { code.range(of: $0, options: .regularExpression) != nil } + || withStrings.range(of: fluxOutsideImport, options: .regularExpression) != nil + return QueryClassification(tier: writes ? .write : .safe, reachesFilesystemOrExecutesCode: reachesOutside) + } + + /// `to()` and `wideTo()` write points, `experimental.to()` too; a `sql.to()` writes to another database. + /// Any mention counts, not only a call, so `w = to` followed by `w(bucket: ...)` is a write too. + /// A record field or parameter named `to:` is not one. + private static let fluxWritingCalls = [#"\b(?:to|wideTo)\b(?!\s*:)"#] + + /// Packages that reach a network service or the server's files rather than the bucket. The import + /// path is matched as well as the member, since `import h "http"` renames the package. + private static let fluxOutsideCalls = [ + #"\b(?:http|requests|sql|slack|pagerduty|mqtt|socket|sensu|telegram|opsgenie|victorops|discord|teams)\."#, + #"\bcsv\.from\s*\(\s*file\s*:"#, + ] + + private static let fluxOutsideImport = + #"\bimport\s+(?:\w+\s+)?"(?:[^"]*/)?(?:http|requests|sql|slack|pagerduty|mqtt|socket|sensu|telegram|opsgenie|victorops|discord|teams)""# + + /// The driver sends a statement whose first line is `-- influxql`, `-- sql` or `// flux` in that + /// language whatever the rest looks like, so the classifier reads it the same way. + private static func languageHint(_ sql: String) -> String? { + let firstLine = sql.trimmingCharacters(in: .whitespacesAndNewlines) + .split(separator: "\n", maxSplits: 1, omittingEmptySubsequences: false).first ?? "" + guard firstLine.hasPrefix("--") || firstLine.hasPrefix("//") else { return nil } + return firstLine.dropFirst(2).trimmingCharacters(in: .whitespaces).lowercased() + } + + private static func looksLikeFlux(_ code: String) -> Bool { + if code.contains("|>") { return true } + let lowered = code.lowercased() + return ["from(", "import ", "buckets(", "option "].contains { lowered.hasPrefix($0) } + } + + /// The text without `//` line comments and, unless `keepingStrings`, with string contents blanked, + /// so a comment or a string that spells `to(` neither writes nor hides a call. + static func fluxCode(_ sql: String, keepingStrings: Bool = false) -> String { + var code = "" + var inString = false + var inComment = false + var escaped = false + var previous: Character? + for character in sql { + if inComment { + if character == "\n" { + inComment = false + code.append(character) + } + previous = character + continue + } + if inString { + if escaped { + escaped = false + } else if character == "\\" { + escaped = true + } else if character == "\"" { + inString = false + } + if keepingStrings || !inString { + code.append(character) + } + previous = character + continue + } + if character == "/", previous == "/" { + code.removeLast() + inComment = true + } else if character == "\"" { + inString = true + code.append(character) + } else { + code.append(character) + } + previous = character + } + return code.trimmingCharacters(in: .whitespacesAndNewlines) + } +} diff --git a/apple/TablePro/Core/Utilities/SQL/QueryClassifier.swift b/apple/TablePro/Core/Utilities/SQL/QueryClassifier.swift index 16b070d1cb..73ba53e45d 100644 --- a/apple/TablePro/Core/Utilities/SQL/QueryClassifier.swift +++ b/apple/TablePro/Core/Utilities/SQL/QueryClassifier.swift @@ -1017,6 +1017,8 @@ private extension QueryClassifier { return worstRequest { firebaseClassification($0, databaseType: databaseType) } case .bigtable: return worstRequest(bigtableClassification) + case .influxdb: + return influxDBClassification(sql) default: return nil } diff --git a/apple/TablePro/Models/Connection/DatabaseType.swift b/apple/TablePro/Models/Connection/DatabaseType.swift index bce99781a3..6dfe6fdef3 100644 --- a/apple/TablePro/Models/Connection/DatabaseType.swift +++ b/apple/TablePro/Models/Connection/DatabaseType.swift @@ -53,6 +53,7 @@ extension DatabaseType { static let trino = DatabaseType(rawValue: "Trino") static let presto = DatabaseType(rawValue: "Presto") static let weaviate = DatabaseType(rawValue: "Weaviate") + static let influxdb = DatabaseType(rawValue: "InfluxDB") static let bigtable = DatabaseType(rawValue: "Bigtable") static let firestore = DatabaseType(rawValue: "Firestore") static let firebaseRealtimeDatabase = DatabaseType(rawValue: "Firebase Realtime Database") diff --git a/apple/TablePro/Resources/Localizable.xcstrings b/apple/TablePro/Resources/Localizable.xcstrings index 4b999ea5d9..e05bc4cec1 100644 --- a/apple/TablePro/Resources/Localizable.xcstrings +++ b/apple/TablePro/Resources/Localizable.xcstrings @@ -4014,6 +4014,10 @@ }, "%@ is not a Vault address. Use a URL such as https://vault.example.com:8200." : { + }, + "%@ is not available on this server. It accepts %@." : { + "extractionState" : "manual", + "generatesSymbol" : false }, "%@ is read-only, so nothing can be written to it." : { "localizations" : { @@ -37322,6 +37326,10 @@ } } }, + "Choose a database or bucket first." : { + "extractionState" : "manual", + "generatesSymbol" : false + }, "Choose a folder containing .sql files" : { "localizations" : { "fr" : { @@ -93665,6 +93673,22 @@ "extractionState" : "manual", "generatesSymbol" : false }, + "InfluxDB 2 needs an organization. Set it in the connection settings." : { + "extractionState" : "manual", + "generatesSymbol" : false + }, + "InfluxDB has no views." : { + "extractionState" : "manual", + "generatesSymbol" : false + }, + "InfluxDB refused the token: %@" : { + "extractionState" : "manual", + "generatesSymbol" : false + }, + "InfluxDB refused the token." : { + "extractionState" : "manual", + "generatesSymbol" : false + }, "Info" : { "extractionState" : "stale", "localizations" : { @@ -95835,6 +95859,10 @@ } } }, + "Invalid InfluxDB address: %@" : { + "extractionState" : "manual", + "generatesSymbol" : false + }, "Invalid JSON" : { "localizations" : { "fr" : { @@ -121453,6 +121481,10 @@ "extractionState" : "manual", "generatesSymbol" : false }, + "Not connected to InfluxDB." : { + "extractionState" : "manual", + "generatesSymbol" : false + }, "Not connected to MongoDB" : { "extractionState" : "manual", "generatesSymbol" : false, @@ -129062,6 +129094,10 @@ } } }, + "Organization" : { + "extractionState" : "manual", + "generatesSymbol" : false + }, "Other" : { "localizations" : { "fr" : { @@ -204595,6 +204631,10 @@ } } }, + "Unexpected InfluxDB response: %@" : { + "extractionState" : "manual", + "generatesSymbol" : false + }, "Unexpected response from the SSO portal for profile \"%@\"." : { "extractionState" : "manual", "generatesSymbol" : false, diff --git a/apple/TableProTests/Core/Plugins/PluginMetadataRegistryTypeCountTests.swift b/apple/TableProTests/Core/Plugins/PluginMetadataRegistryTypeCountTests.swift index e5d6ddaacd..1a7a8794c6 100644 --- a/apple/TableProTests/Core/Plugins/PluginMetadataRegistryTypeCountTests.swift +++ b/apple/TableProTests/Core/Plugins/PluginMetadataRegistryTypeCountTests.swift @@ -10,7 +10,8 @@ struct PluginMetadataRegistryTypeCountTests { "CockroachDB", "Dameng", "Databend", "DuckDB", "DynamoDB", "Elasticsearch", "etcd", "Kafka", "KingbaseES", "libSQL", "MariaDB", "MongoDB", "MySQL", "Oracle", "PGlite", "PostgreSQL", "Redis", "Redshift", "SAP HANA", "ScyllaDB", "Snowflake", "Spanner", "SQL Server", "SQLite", "SurrealDB", "Teradata", "TiDB", - "Trino", "Presto", "Turso", "Typesense", "Weaviate", "OceanBase", "Firestore", "Firebase Realtime Database", "Bigtable" + "Trino", "Presto", "Turso", "Typesense", "Weaviate", "OceanBase", "Firestore", "Firebase Realtime Database", "Bigtable", + "InfluxDB" ] private static func builtInTypeIds() -> Set { diff --git a/apple/TableProTests/Core/Utilities/SQL/QueryClassifierFluxTests.swift b/apple/TableProTests/Core/Utilities/SQL/QueryClassifierFluxTests.swift new file mode 100644 index 0000000000..a5a6da3544 --- /dev/null +++ b/apple/TableProTests/Core/Utilities/SQL/QueryClassifierFluxTests.swift @@ -0,0 +1,59 @@ +import Foundation +@testable import TablePro +import Testing + +struct QueryClassifierFluxTests { + @Test func aFluxReadIsSafe() { + let flux = """ + // last hour + from(bucket: "metrics") + |> range(start: -1h) + |> filter(fn: (r) => r._measurement == "cpu") + """ + #expect(QueryClassifier.classifyTier(flux, databaseType: .influxdb) == .safe) + #expect(!QueryClassifier.isWriteQuery("buckets()", databaseType: .influxdb)) + } + + @Test func toAndWideToWrite() { + let copy = "from(bucket: \"a\") |> range(start: -1h) |> to(bucket: \"b\")" + #expect(QueryClassifier.classifyTier(copy, databaseType: .influxdb) == .write) + let wide = "import \"influxdata/influxdb\"\nfrom(bucket: \"a\") |> range(start: -1h) |> influxdb.wideTo(bucket: \"b\")" + #expect(QueryClassifier.isWriteQuery(wide, databaseType: .influxdb)) + } + + @Test func aLanguageHintDecidesTheReading() { + let influxQL = "-- influxql\nDELETE FROM \"cpu\" WHERE \"host\" = 'a |> b'" + #expect(QueryClassifier.classifyTier(influxQL, databaseType: .influxdb) != .safe) + let flux = "// flux\nfrom(bucket: \"a\") |> range(start: -1h) |> to(bucket: \"b\")" + #expect(QueryClassifier.classifyTier(flux, databaseType: .influxdb) == .write) + } + + @Test func anAliasedWriteStillWrites() { + let flux = "write = to\nfrom(bucket: \"a\") |> range(start: -1h) |> write(bucket: \"b\")" + #expect(QueryClassifier.classifyTier(flux, databaseType: .influxdb) == .write) + let field = "from(bucket: \"a\") |> range(start: -1h) |> map(fn: (r) => ({r with to: 1}))" + #expect(QueryClassifier.classifyTier(field, databaseType: .influxdb) == .safe) + } + + @Test func aRenamedNetworkImportReachesOutside() { + let flux = "import h \"http\"\nfrom(bucket: \"a\") |> range(start: -1h) |> map(fn: (r) => ({r with s: h.get(url: \"x\")}))" + #expect(QueryClassifier.reachesFilesystemOrExecutesCode(flux, databaseType: .influxdb)) + } + + @Test func aStringOrCommentThatSpellsToDoesNotWrite() { + let flux = "from(bucket: \"to(\") // |> to(bucket: \"x\")\n |> range(start: -1h)" + #expect(QueryClassifier.classifyTier(flux, databaseType: .influxdb) == .safe) + } + + @Test func aNetworkPackageReachesOutside() { + let flux = "import \"http\"\nfrom(bucket: \"a\") |> range(start: -1h) |> map(fn: (r) => ({r with s: http.post(url: \"x\")}))" + #expect(QueryClassifier.reachesFilesystemOrExecutesCode(flux, databaseType: .influxdb)) + } + + @Test func influxQLAndSQLKeepTheSQLReading() { + #expect(QueryClassifier.classifyTier("SELECT * FROM \"cpu\" LIMIT 10", databaseType: .influxdb) == .safe) + #expect(QueryClassifier.classifyTier("SHOW MEASUREMENTS", databaseType: .influxdb) == .safe) + #expect(QueryClassifier.classifyTier("DROP MEASUREMENT \"cpu\"", databaseType: .influxdb) == .destructive) + #expect(QueryClassifier.isWriteQuery("SELECT * INTO \"copy\" FROM \"cpu\"", databaseType: .influxdb)) + } +} diff --git a/apple/project.yml b/apple/project.yml index cbf751b9d2..dae93dc036 100644 --- a/apple/project.yml +++ b/apple/project.yml @@ -1262,6 +1262,15 @@ targets: - package: TableProCore product: TableProWeaviateCore + InfluxDBDriverPlugin: + templates: [DriverPlugin] + templateAttributes: + folder: InfluxDBDriverPlugin + principalClass: InfluxDBPlugin + dependencies: + - package: TableProCore + product: TableProInfluxDBCore + # Every plugin is in exactly one of BundledPlugins (embedded in the app) and RegistryPlugins (each # release publishes it); scripts/ci/check-plugin-manifest.py enforces it. AllPlugins compiles both. aggregateTargets: @@ -1299,6 +1308,7 @@ aggregateTargets: - EtcdDriverPlugin - FirebaseDriverPlugin - HanaDriver + - InfluxDBDriverPlugin - KafkaDriverPlugin - KingbaseESDriverPlugin - LibSQLDriverPlugin diff --git a/apple/scripts/probes/check-influxdb.sh b/apple/scripts/probes/check-influxdb.sh new file mode 100755 index 0000000000..d33ddcb0d0 --- /dev/null +++ b/apple/scripts/probes/check-influxdb.sh @@ -0,0 +1,274 @@ +#!/usr/bin/env bash +# +# Check the InfluxDB driver against real 2.x and 3.x servers. +# +# The plugin is registry-only, so no test target compiles it. This builds PluginKit, TableProInfluxDBCore, Plugins/Shared +# and the plugin's own sources with a harness and drives the driver the way the app does: connect, the sidebar catalog, +# the grid's browse, filter and count queries, and typed SQL, InfluxQL and Flux. Both servers must hold the points the +# usage below writes; the probe adds a 2,500-point `load_