|
| 1 | +// |
| 2 | +// CatalogTableListing.swift |
| 3 | +// TablePro |
| 4 | +// |
| 5 | + |
| 6 | +import Foundation |
| 7 | +import os |
| 8 | + |
| 9 | +/// Every table one database holds, across all of its schemas. |
| 10 | +/// |
| 11 | +/// An engine that answers `fetchTablesInAllSchemas()` is asked once. Any other is asked schema by |
| 12 | +/// schema, and each of those reads queues on the metadata lane by itself, so a sidebar expansion |
| 13 | +/// that arrives in the middle waits behind one schema rather than behind all of them. Every read |
| 14 | +/// goes through the one scope the caller names: a scope per schema would open a pooled connection |
| 15 | +/// per schema. |
| 16 | +@MainActor |
| 17 | +internal enum CatalogTableListing { |
| 18 | + /// A schema whose own read failed is named rather than dropped. Read as empty, it would tell a |
| 19 | + /// search that nothing in it matches, and hide exactly the table the search was looking for. |
| 20 | + internal struct Result: Sendable, Equatable { |
| 21 | + internal let tables: [TableInfo] |
| 22 | + internal let unlistedSchemas: Set<String> |
| 23 | + |
| 24 | + /// This listing with another read of some of its unlisted schemas folded in. A schema the |
| 25 | + /// read listed replaces what was known of it; one it still could not list keeps its rows |
| 26 | + /// and stays unlisted. |
| 27 | + internal func merging(_ retry: Result, retried schemas: Set<String>) -> Result { |
| 28 | + let listedNow = schemas.subtracting(retry.unlistedSchemas) |
| 29 | + let kept = tables.filter { table in |
| 30 | + guard let schema = table.schema else { return true } |
| 31 | + return !listedNow.contains(schema) |
| 32 | + } |
| 33 | + return Result( |
| 34 | + tables: kept + retry.tables, |
| 35 | + unlistedSchemas: unlistedSchemas.subtracting(schemas).union(retry.unlistedSchemas) |
| 36 | + ) |
| 37 | + } |
| 38 | + |
| 39 | + /// A refresh that could not read a schema says nothing new about it, so the rows an earlier |
| 40 | + /// listing had for that schema are carried over rather than dropped. |
| 41 | + internal func keepingRows(from previous: Result?) -> Result { |
| 42 | + guard let previous, !unlistedSchemas.isEmpty else { return self } |
| 43 | + let carried = previous.tables.filter { table in |
| 44 | + guard let schema = table.schema else { return false } |
| 45 | + return unlistedSchemas.contains(schema) |
| 46 | + } |
| 47 | + return Result(tables: tables + carried, unlistedSchemas: unlistedSchemas) |
| 48 | + } |
| 49 | + } |
| 50 | + |
| 51 | + private static let logger = Logger(subsystem: "com.TablePro", category: "CatalogTableListing") |
| 52 | + |
| 53 | + internal static func tables( |
| 54 | + in scope: DatabaseScope, |
| 55 | + excludingSchemas excluded: Set<String>, |
| 56 | + metadata: ScopedMetadataProviding = DatabaseManager.shared |
| 57 | + ) async throws -> Result { |
| 58 | + let listed = try await metadata.withMetadataDriver(scope: scope, workload: .bulk) { driver in |
| 59 | + try await driver.fetchTablesInAllSchemas() |
| 60 | + } |
| 61 | + if let listed { |
| 62 | + let tables = listed.filter { table in |
| 63 | + guard let schema = table.schema else { return true } |
| 64 | + return !excluded.contains(schema) |
| 65 | + } |
| 66 | + return Result(tables: tables, unlistedSchemas: []) |
| 67 | + } |
| 68 | + let schemas = try await metadata.withMetadataDriver(scope: scope, workload: .bulk) { driver in |
| 69 | + try await driver.fetchSchemas() |
| 70 | + } |
| 71 | + return try await tables(inSchemas: schemas.filter { !excluded.contains($0) }, scope: scope, metadata: metadata) |
| 72 | + } |
| 73 | + |
| 74 | + /// The named schemas one by one, which is also how a listing asks again for the schemas it |
| 75 | + /// could not read the first time. |
| 76 | + /// |
| 77 | + /// Only a failure that belongs to one schema is recorded against it. A lost connection fails |
| 78 | + /// every schema the same way, and recording that as a listing of nothing would read as a |
| 79 | + /// database with no tables, so it fails the whole read instead, as does every schema failing. |
| 80 | + internal static func tables( |
| 81 | + inSchemas schemas: [String], |
| 82 | + scope: DatabaseScope, |
| 83 | + metadata: ScopedMetadataProviding = DatabaseManager.shared |
| 84 | + ) async throws -> Result { |
| 85 | + var tables: [TableInfo] = [] |
| 86 | + var unlisted: Set<String> = [] |
| 87 | + var lastError: Error? |
| 88 | + for schema in schemas { |
| 89 | + try Task.checkCancellation() |
| 90 | + do { |
| 91 | + tables += try await metadata.withMetadataDriver(scope: scope, workload: .bulk) { driver in |
| 92 | + try await driver.fetchTables(schema: schema) |
| 93 | + } |
| 94 | + } catch is CancellationError { |
| 95 | + throw CancellationError() |
| 96 | + } catch let error as DatabaseError { |
| 97 | + throw error |
| 98 | + } catch { |
| 99 | + logger.warning( |
| 100 | + "[catalog] schema not listed schema=\(schema, privacy: .private(mask: .hash)) error=\(error.publicLogShape, privacy: .public)" |
| 101 | + ) |
| 102 | + unlisted.insert(schema) |
| 103 | + lastError = error |
| 104 | + } |
| 105 | + } |
| 106 | + if let lastError, !schemas.isEmpty, unlisted.count == schemas.count { |
| 107 | + throw lastError |
| 108 | + } |
| 109 | + return Result(tables: tables, unlistedSchemas: unlisted) |
| 110 | + } |
| 111 | +} |
0 commit comments