diff --git a/CHANGELOG.md b/CHANGELOG.md index 56fdef005..9cad8805e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -86,6 +86,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Literal backticks in cloudflared, cloud-sql-proxy, SSH config, remote command and dump tool install messages. - Tunnel command preview showing port 0 or the wrong host when Port is blank or the connection uses a host list. - SSH tab host-list warning naming replica set failover for Redis and Kafka, and implying Sentinel works through a tunnel. +- MongoDB exports dropping fields first seen after the 200th document, and exporting a field null in the first 200 as text. - iPhone and iPad reading a Safe Mode level they do not recognize from iCloud as Off. ### Security diff --git a/Packages/TableProCore/Tests/TableProSyncTests/SyncRecordMapperTests.swift b/Packages/TableProCore/Tests/TableProSyncTests/SyncRecordMapperTests.swift index 0463cdcad..2fa2afdfb 100644 --- a/Packages/TableProCore/Tests/TableProSyncTests/SyncRecordMapperTests.swift +++ b/Packages/TableProCore/Tests/TableProSyncTests/SyncRecordMapperTests.swift @@ -86,4 +86,25 @@ struct SyncRecordMapperTests { let decoded = try #require(SyncRecordMapper.toConnection(record)) #expect(decoded.safeModeLevel == .confirmWrites) } + + @Test("An unrecognized wire value keeps the legacy read-only restriction") + func unknownWireValuePreservesReadOnly() throws { + let record = makeRawRecord(safeModeLevelRaw: "someFutureLevel", isReadOnly: true) + let decoded = try #require(SyncRecordMapper.toConnection(record)) + #expect(decoded.safeModeLevel == .readOnly) + } + + @Test("A rename preserves an unrecognized wire value and requires confirmation") + func renamePreservesUnknownWireValue() throws { + let record = makeRawRecord(safeModeLevelRaw: "someFutureLevel") + var connection = try #require(SyncRecordMapper.toConnection(record)) + connection.name = "Renamed" + + SyncRecordMapper.updateRecord(record, with: connection) + + #expect(record["safeModeLevel"] as? String == "someFutureLevel") + let decoded = try #require(SyncRecordMapper.toConnection(record)) + #expect(decoded.name == "Renamed") + #expect(decoded.safeModeLevel == .confirmWrites) + } } diff --git a/Plugins/MongoDBDriverPlugin/BsonDocumentFlattener.swift b/Plugins/MongoDBDriverPlugin/BsonDocumentFlattener.swift index be250cea9..a1b150e69 100644 --- a/Plugins/MongoDBDriverPlugin/BsonDocumentFlattener.swift +++ b/Plugins/MongoDBDriverPlugin/BsonDocumentFlattener.swift @@ -461,7 +461,7 @@ struct BsonDocumentFlattener { return counts.max(by: { $0.value < $1.value })?.key ?? .string } - private static func valueKind(for value: Any, representation: MongoDBUuidRepresentation) -> BsonValueKind { + static func valueKind(for value: Any, representation: MongoDBUuidRepresentation) -> BsonValueKind { if value is NSNull { return .null } switch value { diff --git a/Plugins/MongoDBDriverPlugin/MongoDBConnection+SyncHelpers.swift b/Plugins/MongoDBDriverPlugin/MongoDBConnection+SyncHelpers.swift index 09992e0d7..981b02d6b 100644 --- a/Plugins/MongoDBDriverPlugin/MongoDBConnection+SyncHelpers.swift +++ b/Plugins/MongoDBDriverPlugin/MongoDBConnection+SyncHelpers.swift @@ -153,7 +153,8 @@ extension MongoDBConnection { } func aggregateSync( - client: OpaquePointer, database: String, collection: String, pipeline: String + client: OpaquePointer, database: String, collection: String, pipeline: String, + optionsJson: String? = nil ) throws -> (docs: [[String: Any]], isTruncated: Bool) { try checkCancelled() @@ -167,7 +168,9 @@ extension MongoDBConnection { let timeoutMS = queryTimeoutMS var optsBson: OpaquePointer? - if timeoutMS > 0 { + if let optionsJson { + optsBson = jsonToBson(optionsJson) + } else if timeoutMS > 0 { optsBson = jsonToBson("{\"maxTimeMS\": \(timeoutMS)}") } defer { if let opts = optsBson { bson_destroy(opts) } } @@ -415,12 +418,14 @@ extension MongoDBConnection { func iterateCursorStreaming( cursor: OpaquePointer, continuation: AsyncThrowingStream.Continuation, - streamState: MongoStreamState + streamState: MongoStreamState, + census: () throws -> MongoFieldCensus? ) { var docPtr: OpaquePointer? var sample: [[String: Any]] = [] var projection: MongoStreamProjection? var emitted = 0 + var unannounced = Set() while mongoc_cursor_next(cursor, &docPtr) { if Task.isCancelled { @@ -432,24 +437,34 @@ extension MongoDBConnection { guard let doc = docPtr else { continue } let dict = bsonToDict(doc) - if let projection { - continuation.yield(.rows([projection.row(for: dict, convert: streamCellValue)])) - emitted += 1 - if emitted >= PluginRowLimits.emergencyMax { - logger.warning("Streamed result truncated at \(PluginRowLimits.emergencyMax) documents") - break + if projection == nil, sample.count >= MongoStreamProjection.sampleSize { + do { + projection = openStream(sample: sample, census: try census(), continuation: continuation) + } catch { + cleanup(streamState) + continuation.finish(throwing: error) + return } + emitted += sample.count + sample = [] + } + + guard let projection else { + sample.append(dict) continue } - sample.append(dict) - if sample.count >= MongoStreamProjection.sampleSize { - projection = openStream(sample: sample, continuation: continuation) - emitted += sample.count - sample = [] + continuation.yield(.rows([projection.row(for: dict, convert: streamCellValue)])) + unannounced.formUnion(projection.unannouncedFields(in: dict)) + emitted += 1 + if emitted >= PluginRowLimits.emergencyMax { + logger.warning("Streamed result truncated at \(PluginRowLimits.emergencyMax) documents") + break } } + logUnannouncedFields(unannounced) + var error = bson_error_t() if mongoc_cursor_error(cursor, &error) { cleanup(streamState) @@ -458,27 +473,48 @@ extension MongoDBConnection { } if projection == nil { - _ = openStream(sample: sample, continuation: continuation) + _ = openStream(sample: sample, census: nil, continuation: continuation) } cleanup(streamState) continuation.finish() } + func fieldCensus( + _ request: MongoFieldCensus.Request?, + client: OpaquePointer, + database: String, + collection: String + ) throws -> MongoFieldCensus? { + guard let request else { return nil } + do { + let groups = try aggregateSync( + client: client, database: database, collection: collection, + pipeline: request.pipeline, optionsJson: request.optionsJson + ) + return MongoFieldCensus(groups: groups.docs) + } catch let error as MongoDBError { + logger.warning( + "Field census failed, the stream header holds the sampled fields only: \(error.message)" + ) + return nil + } + } + + private func logUnannouncedFields(_ fields: Set) { + guard !fields.isEmpty else { return } + let names = fields.sorted().joined(separator: ", ") + logger.warning( + "Stream header left out \(fields.count) fields first seen after the sample: \(names, privacy: .private)" + ) + } + private func openStream( sample: [[String: Any]], + census: MongoFieldCensus?, continuation: AsyncThrowingStream.Continuation ) -> MongoStreamProjection { - let columns = BsonDocumentFlattener.unionColumns(from: sample) - let kinds = BsonDocumentFlattener.columnKinds( - for: columns, documents: sample, representation: uuidRepresentation - ) - let columnTypeNames = kinds.map { - BsonDocumentFlattener.typeName(for: $0, representation: uuidRepresentation) - } - let projection = MongoStreamProjection( - columns: columns, columnTypeNames: columnTypeNames, kinds: kinds - ) + let projection = MongoStreamProjection(sample: sample, census: census, representation: uuidRepresentation) continuation.yield(.header(projection.header)) diff --git a/Plugins/MongoDBDriverPlugin/MongoDBConnection.swift b/Plugins/MongoDBDriverPlugin/MongoDBConnection.swift index 608eb7c8d..b25848c2f 100644 --- a/Plugins/MongoDBDriverPlugin/MongoDBConnection.swift +++ b/Plugins/MongoDBDriverPlugin/MongoDBConnection.swift @@ -795,7 +795,8 @@ final class MongoDBConnection: @unchecked Sendable { database: String, collection: String, filter: String, - optionsJson: String + optionsJson: String, + census: MongoFieldCensus.Request? ) -> AsyncThrowingStream { #if canImport(CLibMongoc) let queue = self.queue @@ -844,7 +845,11 @@ final class MongoDBConnection: @unchecked Sendable { streamState.collection = col streamState.lock.unlock() - iterateCursorStreaming(cursor: cursor, continuation: continuation, streamState: streamState) + iterateCursorStreaming( + cursor: cursor, continuation: continuation, streamState: streamState + ) { + try self.fieldCensus(census, client: client, database: database, collection: collection) + } } catch { continuation.finish(throwing: error) } @@ -859,7 +864,8 @@ final class MongoDBConnection: @unchecked Sendable { database: String, collection: String, pipeline: String, - optionsJson: String? = nil + optionsJson: String?, + census: MongoFieldCensus.Request? ) -> AsyncThrowingStream { #if canImport(CLibMongoc) let queue = self.queue @@ -919,7 +925,11 @@ final class MongoDBConnection: @unchecked Sendable { streamState.collection = col streamState.lock.unlock() - iterateCursorStreaming(cursor: cursor, continuation: continuation, streamState: streamState) + iterateCursorStreaming( + cursor: cursor, continuation: continuation, streamState: streamState + ) { + try self.fieldCensus(census, client: client, database: database, collection: collection) + } } catch { continuation.finish(throwing: error) } diff --git a/Plugins/MongoDBDriverPlugin/MongoDBPluginDriver.swift b/Plugins/MongoDBDriverPlugin/MongoDBPluginDriver.swift index f61d9e3f3..c10c73cb8 100644 --- a/Plugins/MongoDBDriverPlugin/MongoDBPluginDriver.swift +++ b/Plugins/MongoDBDriverPlugin/MongoDBPluginDriver.swift @@ -835,18 +835,23 @@ final class MongoDBPluginDriver: PluginDatabaseDriver, @unchecked Sendable { switch try await runtime.exportPlan(for: trimmed, database: db) { case .cursor(let plan, let databaseSwitch, let writes): if let databaseSwitch { self.currentDb = databaseSwitch } + let census = MongoFieldCensus.request( + for: plan, limit: PluginRowLimits.emergencyMax, timeoutMS: timeout + ) let inner = plan.isFind ? conn.streamFind( database: plan.database, collection: plan.collection, filter: plan.filter, optionsJson: plan.options.findOptionsJson( limit: PluginRowLimits.emergencyMax, timeoutMS: timeout - ) + ), + census: census ) : conn.streamAggregate( database: plan.database, collection: plan.collection, pipeline: plan.pipeline, - optionsJson: plan.options.aggregateOptionsJson(timeoutMS: timeout) + optionsJson: plan.options.aggregateOptionsJson(timeoutMS: timeout), + census: census ) do { for try await element in inner { diff --git a/Plugins/MongoDBDriverPlugin/MongoFieldCensus.swift b/Plugins/MongoDBDriverPlugin/MongoFieldCensus.swift new file mode 100644 index 000000000..2bb3c2fa5 --- /dev/null +++ b/Plugins/MongoDBDriverPlugin/MongoFieldCensus.swift @@ -0,0 +1,101 @@ +import Foundation +import TableProPluginKit + +struct MongoFieldCensus { + struct Tally { + let field: String + let documents: Int64 + let example: Any + } + + struct Request: Sendable { + let pipeline: String + let optionsJson: String? + } + + private static let tallyStages = [ + "{\"$project\": {\"_id\": 0, \"pair\": {\"$objectToArray\": \"$$ROOT\"}}}", + "{\"$unwind\": \"$pair\"}", + "{\"$group\": {\"_id\": {\"$arrayToObject\": [[[\"$pair.k\", {\"$type\": \"$pair.v\"}]]]}, " + + "\"name\": {\"$first\": \"$pair.k\"}, " + + "\"documents\": {\"$sum\": 1}, \"example\": {\"$first\": \"$pair.v\"}}}" + ] + + private static let writingStages: Set = ["$out", "$merge"] + + let tallies: [Tally] + + init(tallies: [Tally]) { + self.tallies = tallies + } + + init(groups: [[String: Any]]) { + self.init(tallies: groups.compactMap(Self.tally(from:))) + } + + var fields: [String] { + Set(tallies.map(\.field)).sorted() + } + + func kinds(representation: MongoDBUuidRepresentation) -> [String: BsonValueKind] { + var votes: [String: [BsonValueKind: Int64]] = [:] + for tally in tallies where !(tally.example is NSNull) { + let kind = BsonDocumentFlattener.valueKind(for: tally.example, representation: representation) + votes[tally.field, default: [:]][kind, default: 0] += tally.documents + } + return votes.compactMapValues { fieldVotes in + fieldVotes.max { $0.value < $1.value }?.key + } + } + + static func request(for plan: MongoScriptCursorPlan, limit ceiling: Int, timeoutMS: Int32) -> Request? { + let sourceStages = plan.isFind + ? findStages(of: plan, limit: ceiling) + : aggregateStages(of: plan.pipeline, limit: ceiling) + guard let sourceStages else { return nil } + return Request( + pipeline: "[\((sourceStages + tallyStages).joined(separator: ", "))]", + optionsJson: plan.options.aggregateOptionsJson(timeoutMS: timeoutMS) + ) + } + + private static func findStages(of plan: MongoScriptCursorPlan, limit ceiling: Int) -> [String] { + var stages = ["{\"$match\": \(plan.filter)}"] + let isPaged = plan.options.skip != nil || plan.options.limit != nil + if isPaged, let sort = document(plan.sort) { stages.append("{\"$sort\": \(sort)}") } + if plan.skip > 0 { stages.append("{\"$skip\": \(plan.skip)}") } + stages += limitStage(plan.options.effectiveLimit(ceiling: ceiling)) + if let projection = document(plan.projection) { stages.append("{\"$project\": \(projection)}") } + return stages + } + + private static func aggregateStages(of pipeline: String, limit ceiling: Int) -> [String]? { + guard let data = pipeline.data(using: .utf8), + let stages = (try? JSONSerialization.jsonObject(with: data)) as? [[String: Any]], + !stages.contains(where: { !writingStages.isDisjoint(with: $0.keys) }) else { + return nil + } + let trimmed = pipeline.trimmingCharacters(in: .whitespacesAndNewlines) + let inner = String(trimmed.dropFirst().dropLast()).trimmingCharacters(in: .whitespacesAndNewlines) + return (inner.isEmpty ? [] : [inner]) + limitStage(ceiling) + } + + private static func limitStage(_ limit: Int) -> [String] { + limit > 0 ? ["{\"$limit\": \(limit)}"] : [] + } + + private static func document(_ text: String?) -> String? { + guard let text else { return nil } + let compact = text.filter { !$0.isWhitespace } + guard !compact.isEmpty, compact != "null", compact != "{}" else { return nil } + return text + } + + private static func tally(from group: [String: Any]) -> Tally? { + guard let field = group["name"] as? String, + let documents = MongoScriptJson.numeric(group["documents"]) else { + return nil + } + return Tally(field: field, documents: documents, example: group["example"] ?? NSNull()) + } +} diff --git a/Plugins/MongoDBDriverPlugin/MongoStreamProjection.swift b/Plugins/MongoDBDriverPlugin/MongoStreamProjection.swift index 0f88647b5..5e1bfb20d 100644 --- a/Plugins/MongoDBDriverPlugin/MongoStreamProjection.swift +++ b/Plugins/MongoDBDriverPlugin/MongoStreamProjection.swift @@ -14,16 +14,19 @@ struct MongoStreamProjection { let columns: [String] let columnTypeNames: [String] let kinds: [BsonValueKind] + private let announced: Set init(columns: [String], columnTypeNames: [String], kinds: [BsonValueKind] = []) { guard !columns.isEmpty else { self.columns = ["_id"] self.columnTypeNames = ["VARCHAR"] self.kinds = [.string] + self.announced = ["_id"] return } self.columns = columns + self.announced = Set(columns) self.columnTypeNames = columns.indices.map { index in index < columnTypeNames.count ? columnTypeNames[index] : "VARCHAR" } @@ -32,6 +35,29 @@ struct MongoStreamProjection { } } + init(sample: [[String: Any]], census: MongoFieldCensus?, representation: MongoDBUuidRepresentation) { + let sampled = BsonDocumentFlattener.unionColumns(from: sample) + let sampledKinds = BsonDocumentFlattener.columnKinds( + for: sampled, documents: sample, representation: representation + ) + let typedInSample = BsonDocumentFlattener.heldKinds(in: sample, representation: representation) + let censusKinds = census?.kinds(representation: representation) ?? [:] + let known = Set(sampled) + let unsampled = (census?.fields ?? []).filter { !known.contains($0) } + + let resolvedSampledKinds: [BsonValueKind] = zip(sampled, sampledKinds).map { field, sampledKind in + guard typedInSample[field] == nil else { return sampledKind } + return censusKinds[field] ?? sampledKind + } + let unsampledKinds: [BsonValueKind] = unsampled.map { censusKinds[$0] ?? .string } + let kinds = resolvedSampledKinds + unsampledKinds + self.init( + columns: sampled + unsampled, + columnTypeNames: kinds.map { BsonDocumentFlattener.typeName(for: $0, representation: representation) }, + kinds: kinds + ) + } + var header: PluginStreamHeader { PluginStreamHeader(columns: columns, columnTypeNames: columnTypeNames) } @@ -45,4 +71,8 @@ struct MongoStreamProjection { return convert(value, kinds[index]) } } + + func unannouncedFields(in document: [String: Any]) -> [String] { + document.keys.filter { !announced.contains($0) } + } } diff --git a/TableProTests/Plugins/MongoFieldCensusTests.swift b/TableProTests/Plugins/MongoFieldCensusTests.swift new file mode 100644 index 000000000..cc469078e --- /dev/null +++ b/TableProTests/Plugins/MongoFieldCensusTests.swift @@ -0,0 +1,168 @@ +import Foundation +import TableProPluginKit +import Testing + +struct MongoFieldCensusTests { + private func findPlan(filter: String = "{}", options: MongoScriptCursorOptions = .none) -> MongoScriptCursorPlan { + MongoScriptCursorPlan( + database: "shop", collection: "orders", isFind: true, filter: filter, pipeline: "[]", options: options + ) + } + + private func aggregatePlan(_ pipeline: String, options: MongoScriptCursorOptions = .none) -> MongoScriptCursorPlan { + MongoScriptCursorPlan( + database: "shop", collection: "orders", isFind: false, filter: "{}", pipeline: pipeline, options: options + ) + } + + private func stages(of request: MongoFieldCensus.Request?) throws -> [[String: Any]] { + let pipeline = try #require(request?.pipeline) + let data = try #require(pipeline.data(using: .utf8)) + return try #require(try JSONSerialization.jsonObject(with: data) as? [[String: Any]]) + } + + private func stageNames(of request: MongoFieldCensus.Request?) throws -> [String] { + try stages(of: request).compactMap { $0.keys.first } + } + + private let tallyStageNames = ["$project", "$unwind", "$group"] + + @Test("A find is tallied over the documents its filter matches, up to the stream's ceiling") + func findTalliesItsFilter() throws { + let request = MongoFieldCensus.request( + for: findPlan(filter: "{\"status\": \"open\"}"), limit: 5_000_000, timeoutMS: 0 + ) + #expect(try stageNames(of: request) == ["$match", "$limit"] + tallyStageNames) + let all = try stages(of: request) + let match = try #require(all.first?["$match"] as? [String: Any]) + #expect(match["status"] as? String == "open") + #expect(all[1]["$limit"] as? Int == 5_000_000) + } + + @Test("A paged find keeps its sort, skip and limit so the tally covers the same documents") + func pagedFindKeepsItsPaging() throws { + var options = MongoScriptCursorOptions.none + options.sort = "{\"createdAt\": -1}" + options.skip = 20 + options.limit = 50 + let request = MongoFieldCensus.request(for: findPlan(options: options), limit: 5_000_000, timeoutMS: 0) + + #expect(try stageNames(of: request) == ["$match", "$sort", "$skip", "$limit"] + tallyStageNames) + let all = try stages(of: request) + #expect(all[2]["$skip"] as? Int == 20) + #expect(all[3]["$limit"] as? Int == 50) + } + + @Test("A sort without paging is left out, since it cannot change which fields exist") + func unpagedSortIsDropped() throws { + var options = MongoScriptCursorOptions.none + options.sort = "{\"createdAt\": -1}" + let request = MongoFieldCensus.request(for: findPlan(options: options), limit: 5_000_000, timeoutMS: 0) + #expect(try stageNames(of: request) == ["$match", "$limit"] + tallyStageNames) + } + + @Test("A find limit above the ceiling is held to the ceiling, as the stream's own find is") + func findLimitIsHeldToTheCeiling() throws { + var options = MongoScriptCursorOptions.none + options.limit = 9_000_000 + let request = MongoFieldCensus.request(for: findPlan(options: options), limit: 5_000_000, timeoutMS: 0) + #expect(try stageNames(of: request) == ["$match", "$limit"] + tallyStageNames) + #expect(try stages(of: request)[1]["$limit"] as? Int == 5_000_000) + } + + @Test("A find projection is applied before the tally, and an empty one is ignored") + func findProjectionShapesTheTally() throws { + var options = MongoScriptCursorOptions.none + options.projection = "{\"name\": 1}" + let projected = MongoFieldCensus.request(for: findPlan(options: options), limit: 5_000_000, timeoutMS: 0) + #expect(try stageNames(of: projected) == ["$match", "$limit", "$project"] + tallyStageNames) + let userProjection = try #require(try stages(of: projected)[2]["$project"] as? [String: Any]) + #expect(userProjection["name"] as? Int == 1) + + options.projection = "{ }" + let empty = MongoFieldCensus.request(for: findPlan(options: options), limit: 5_000_000, timeoutMS: 0) + #expect(try stageNames(of: empty) == ["$match", "$limit"] + tallyStageNames) + } + + @Test("An aggregation is tallied over its own output, up to the stream's ceiling") + func aggregateTalliesItsOutput() throws { + let request = MongoFieldCensus.request( + for: aggregatePlan("[{\"$match\": {}}, {\"$addFields\": {\"total\": 1}}]"), + limit: 5_000_000, timeoutMS: 0 + ) + #expect(try stageNames(of: request) == ["$match", "$addFields", "$limit"] + tallyStageNames) + #expect(try stages(of: request)[2]["$limit"] as? Int == 5_000_000) + } + + @Test("An empty pipeline is tallied over the collection, up to the stream's ceiling") + func emptyPipelineTalliesTheCollection() throws { + let request = MongoFieldCensus.request(for: aggregatePlan("[]"), limit: 5_000_000, timeoutMS: 0) + #expect(try stageNames(of: request) == ["$limit"] + tallyStageNames) + } + + @Test("A pipeline that writes is never run a second time for a tally", arguments: [ + "[{\"$match\": {}}, {\"$out\": \"archive\"}]", + "[{\"$merge\": {\"into\": \"archive\"}}]" + ]) + func writingPipelineHasNoTally(pipeline: String) { + #expect(MongoFieldCensus.request(for: aggregatePlan(pipeline), limit: 5_000_000, timeoutMS: 0) == nil) + } + + @Test("The tally runs with the statement's hint, collation and time limit") + func tallyKeepsTheStatementOptions() throws { + var options = MongoScriptCursorOptions.none + options.hint = "{\"status\": 1}" + options.collation = "{\"locale\": \"fr\"}" + let request = MongoFieldCensus.request(for: findPlan(options: options), limit: 5_000_000, timeoutMS: 3_000) + let optionsJson = try #require(request?.optionsJson) + let data = try #require(optionsJson.data(using: .utf8)) + let parsed = try #require(try JSONSerialization.jsonObject(with: data) as? [String: Any]) + #expect((parsed["hint"] as? [String: Any])?["status"] as? Int == 1) + #expect((parsed["collation"] as? [String: Any])?["locale"] as? String == "fr") + #expect(parsed["maxTimeMS"] as? Int == 3_000) + } + + @Test("Field names are grouped as object keys, which the server compares byte for byte under any collation") + func fieldNamesAreGroupedAsObjectKeys() throws { + let request = MongoFieldCensus.request(for: findPlan(), limit: 5_000_000, timeoutMS: 0) + let group = try #require(try stages(of: request).last?["$group"] as? [String: Any]) + let key = try #require(group["_id"] as? [String: Any]) + #expect(Array(key.keys) == ["$arrayToObject"]) + let argument = try #require(key["$arrayToObject"] as? [Any]) + let pairs = try #require(argument.first as? [Any]) + let pair = try #require(pairs.first as? [Any]) + #expect(pair.first as? String == "$pair.k") + #expect((pair.last as? [String: Any])?["$type"] as? String == "$pair.v") + #expect((group["name"] as? [String: Any])?["$first"] as? String == "$pair.k") + } + + @Test("Group replies become tallies, and one without a field name is skipped") + func groupRepliesBecomeTallies() { + let seen = Date(timeIntervalSince1970: 1_700_000_000) + let census = MongoFieldCensus(groups: [ + ["_id": ["late": "date"], "name": "late", "documents": Int32(3), "example": seen], + ["_id": ["big": "long"], "name": "big", "documents": Int64(4), "example": Int64(1)], + ["_id": ["lost": "int"], "documents": Int32(1)] + ]) + #expect(census.tallies.map(\.field) == ["late", "big"]) + #expect(census.tallies.map(\.documents) == [3, 4]) + #expect(census.tallies.first?.example as? Date == seen) + } + + @Test("A group whose example is missing counts as null") + func missingExampleIsNull() { + let census = MongoFieldCensus(groups: [["_id": ["gone": "null"], "name": "gone", "documents": Int32(2)]]) + #expect(census.fields == ["gone"]) + #expect(census.kinds(representation: .unspecified).isEmpty) + } + + @Test("Every tallied field is listed once, in name order") + func fieldsAreListedOnceInNameOrder() { + let census = MongoFieldCensus(tallies: [ + MongoFieldCensus.Tally(field: "zeta", documents: 1, example: "z"), + MongoFieldCensus.Tally(field: "alpha", documents: 2, example: Int32(1)), + MongoFieldCensus.Tally(field: "zeta", documents: 3, example: NSNull()) + ]) + #expect(census.fields == ["alpha", "zeta"]) + } +} diff --git a/TableProTests/Plugins/MongoStreamProjectionTests.swift b/TableProTests/Plugins/MongoStreamProjectionTests.swift index bd16a9450..e623acd31 100644 --- a/TableProTests/Plugins/MongoStreamProjectionTests.swift +++ b/TableProTests/Plugins/MongoStreamProjectionTests.swift @@ -72,4 +72,115 @@ struct MongoStreamProjectionTests { let row = projection.row(for: ["a": "first", "z": "last"], convert: text) #expect(row == [.text("last"), .text("first")]) } + + private var fullSample: [[String: Any]] { + (1...MongoStreamProjection.sampleSize).map { ["_id": Int32($0), "name": "user \($0)"] } + } + + private func census(_ tallies: [(String, Int64, Any)]) -> MongoFieldCensus { + MongoFieldCensus(tallies: tallies.map { MongoFieldCensus.Tally(field: $0.0, documents: $0.1, example: $0.2) }) + } + + private func cell(_ value: Any, _ kind: BsonValueKind) -> PluginCellValue { + BsonDocumentFlattener.cellValue(for: value, kind: kind, representation: .unspecified) + } + + @Test("A field the census reports after the sample joins the header with its own type") + func censusFieldJoinsTheHeader() throws { + let seen = Date(timeIntervalSince1970: 1_700_000_000) + let projection = MongoStreamProjection( + sample: fullSample, + census: census([ + ("_id", 201, Int32(1)), + ("name", 201, "user 1"), + ("late", 1, seen) + ]), + representation: .unspecified + ) + + #expect(projection.columns == ["_id", "name", "late"]) + let lateIndex = try #require(projection.columns.firstIndex(of: "late")) + #expect(projection.columnTypeNames[lateIndex] == "TIMESTAMP") + + let row = projection.row(for: ["_id": Int32(201), "name": "user 201", "late": seen], convert: cell) + #expect(row[lateIndex] != .null) + } + + @Test("Without a census the header is the sample's columns") + func missingCensusKeepsTheSampledColumns() { + let projection = MongoStreamProjection(sample: fullSample, census: nil, representation: .unspecified) + #expect(projection.columns == ["_id", "name"]) + #expect(projection.columnTypeNames == ["INTEGER", "VARCHAR"]) + } + + @Test("Sampled columns keep their order and census fields follow in name order") + func censusFieldsFollowTheSampledColumns() { + let projection = MongoStreamProjection( + sample: fullSample, + census: census([("zeta", 4, "z"), ("name", 201, "user 1"), ("alpha", 2, true)]), + representation: .unspecified + ) + #expect(projection.columns == ["_id", "name", "alpha", "zeta"]) + #expect(projection.columnTypeNames == ["INTEGER", "VARCHAR", "BOOLEAN", "VARCHAR"]) + } + + @Test("A census field takes the type most of its documents hold, and nulls do not vote") + func censusFieldTakesTheMajorityType() { + let projection = MongoStreamProjection( + sample: fullSample, + census: census([ + ("mixed", 3, "text"), + ("mixed", 10, Int64(5_000_000_000)), + ("dated", 500, NSNull()), + ("dated", 1, Date(timeIntervalSince1970: 0)), + ("empty", 7, NSNull()) + ]), + representation: .unspecified + ) + #expect(projection.columns == ["_id", "name", "dated", "empty", "mixed"]) + #expect(projection.columnTypeNames == ["INTEGER", "VARCHAR", "TIMESTAMP", "VARCHAR", "BIGINT"]) + } + + @Test("A sampled column that held only nulls takes the type the census found for it") + func censusTypesAColumnTheSampleHeldOnlyNullsIn() { + let sample: [[String: Any]] = (1...MongoStreamProjection.sampleSize).map { + ["_id": Int32($0), "deletedAt": NSNull()] + } + let deleted = Date(timeIntervalSince1970: 1_700_000_000) + let projection = MongoStreamProjection( + sample: sample, + census: census([("deletedAt", 200, NSNull()), ("deletedAt", 40, deleted)]), + representation: .unspecified + ) + + #expect(projection.columns == ["_id", "deletedAt"]) + #expect(projection.columnTypeNames == ["INTEGER", "TIMESTAMP"]) + let row = projection.row(for: ["_id": Int32(201), "deletedAt": deleted], convert: cell) + #expect(row[1] != .null) + } + + @Test("The census never changes the type the sample chose for a sampled column") + func censusLeavesSampledTypesAlone() { + let projection = MongoStreamProjection( + sample: fullSample, + census: census([("name", 10_000, Int32(3))]), + representation: .unspecified + ) + #expect(projection.columns == ["_id", "name"]) + #expect(projection.columnTypeNames == ["INTEGER", "VARCHAR"]) + } + + @Test("A field the header left out is reported, and an announced one is not") + func fieldsOutsideTheHeaderAreReported() { + let projection = MongoStreamProjection(sample: fullSample, census: nil, representation: .unspecified) + let late = projection.unannouncedFields(in: ["_id": Int32(201), "name": "user 201", "late": true]) + #expect(late == ["late"]) + #expect(projection.unannouncedFields(in: ["_id": Int32(202), "name": "user 202"]).isEmpty) + } + + @Test("The fallback header announces only _id") + func fallbackHeaderAnnouncesOnlyTheId() { + let projection = MongoStreamProjection(columns: [], columnTypeNames: []) + #expect(projection.unannouncedFields(in: ["_id": 1, "name": "Ada"]) == ["name"]) + } } diff --git a/project.yml b/project.yml index b70891ebb..b7b994fd1 100644 --- a/project.yml +++ b/project.yml @@ -620,6 +620,8 @@ targets: - Plugins/MongoDBDriverPlugin/MongoWriteFailure.swift - Plugins/MongoDBDriverPlugin/MongoWriteLedger.swift - Plugins/MongoDBDriverPlugin/MongoStreamProjection.swift + - Plugins/MongoDBDriverPlugin/MongoFieldCensus.swift + - Plugins/MongoDBDriverPlugin/MongoScriptCursorPlan.swift - Plugins/OracleDriverPlugin/OracleObjectQueries.swift - Plugins/MySQLDriverPlugin/GeometryWKBParser.swift - Plugins/MySQLDriverPlugin/MariaDBFieldClassifier.swift