diff --git a/apple/Packages/TableProCore/Sources/TableProTrinoCore/TrinoIntrospectionSQL.swift b/apple/Packages/TableProCore/Sources/TableProTrinoCore/TrinoIntrospectionSQL.swift
index b6da4755b9..13b8f67927 100644
--- a/apple/Packages/TableProCore/Sources/TableProTrinoCore/TrinoIntrospectionSQL.swift
+++ b/apple/Packages/TableProCore/Sources/TableProTrinoCore/TrinoIntrospectionSQL.swift
@@ -1,12 +1,19 @@
 import Foundation
 
 public enum TrinoIntrospectionSQL {
+    // `.literal` matches code units; the default skips a quote that a combining mark follows.
     public static func quoteIdentifier(_ name: String) -> String {
-        "\"" + name.replacingOccurrences(of: "\"", with: "\"\"") + "\""
+        "\"" + name.replacingOccurrences(of: "\"", with: "\"\"", options: .literal) + "\""
     }
 
     public static func quoteLiteral(_ value: String) -> String {
-        "'" + value.replacingOccurrences(of: "'", with: "''") + "'"
+        "'" + value.replacingOccurrences(of: "'", with: "''", options: .literal) + "'"
+    }
+
+    // `IN ()` does not parse.
+    private static func membership(_ column: String, in values: [String]) -> String {
+        guard !values.isEmpty else { return "false" }
+        return "\(column) IN (\(values.map(quoteLiteral).joined(separator: ", ")))"
     }
 
     public static func qualifiedName(catalog: String?, schema: String?, table: String) -> String {
@@ -36,6 +43,13 @@ public enum TrinoIntrospectionSQL {
         """
     }
 
+    public static func listTables(catalog: String, inSchemas schemas: [String]) -> String {
+        """
+        SELECT table_name, table_type, table_schema FROM \(quoteIdentifier(catalog)).information_schema.tables \
+        WHERE \(membership("table_schema", in: schemas)) ORDER BY table_schema, table_name
+        """
+    }
+
     public static func listColumns(catalog: String, schema: String, table: String) -> String {
         """
         SELECT column_name, data_type, is_nullable, column_default, comment, ordinal_position \
@@ -83,6 +97,25 @@ public enum TrinoIntrospectionSQL {
         }
     }
 
+    public static func listMaterializedViews(
+        catalog: String,
+        inSchemas schemas: [String],
+        flavor: TrinoServerFlavor
+    ) -> String {
+        switch flavor {
+        case .trino:
+            return """
+                SELECT name, schema_name FROM system.metadata.materialized_views \
+                WHERE catalog_name = \(quoteLiteral(catalog)) AND \(membership("schema_name", in: schemas))
+                """
+        case .presto:
+            return """
+                SELECT table_name, table_schema FROM \(quoteIdentifier(catalog)).information_schema.materialized_views \
+                WHERE \(membership("table_schema", in: schemas))
+                """
+        }
+    }
+
     /// One relation's `table_type`, which says `VIEW` for a view. A materialized view is not told
     /// apart here: ``materializedView(catalog:schema:name:flavor:)`` answers for it.
     public static func relationType(catalog: String, schema: String, table: String) -> String {
diff --git a/apple/Packages/TableProCore/Tests/TableProTrinoCoreTests/TrinoIntrospectionSQLTests.swift b/apple/Packages/TableProCore/Tests/TableProTrinoCoreTests/TrinoIntrospectionSQLTests.swift
new file mode 100644
index 0000000000..7eec294eba
--- /dev/null
+++ b/apple/Packages/TableProCore/Tests/TableProTrinoCoreTests/TrinoIntrospectionSQLTests.swift
@@ -0,0 +1,37 @@
+import TableProTrinoCore
+import Testing
+
+struct TrinoIntrospectionSQLTests {
+    @Test("A schema set is listed in one query that names each schema and projects it")
+    func schemaSetListing() {
+        let statement = TrinoIntrospectionSQL.listTables(catalog: "hive", inSchemas: ["sales", "o'brien"])
+        #expect(statement.contains("FROM \"hive\".information_schema.tables"))
+        #expect(statement.contains("WHERE table_schema IN ('sales', 'o''brien')"))
+        #expect(statement.contains("SELECT table_name, table_type, table_schema"))
+        #expect(TrinoIntrospectionSQL.listTables(catalog: "hive", inSchemas: []).contains("WHERE false"))
+    }
+
+    @Test("A schema set's materialized views come from one query that projects the schema")
+    func schemaSetMaterializedViews() {
+        let statement = TrinoIntrospectionSQL.listMaterializedViews(
+            catalog: "iceberg", inSchemas: ["a", "b"], flavor: .trino
+        )
+        #expect(statement.contains("SELECT name, schema_name FROM system.metadata.materialized_views"))
+        #expect(statement.contains("catalog_name = 'iceberg' AND schema_name IN ('a', 'b')"))
+    }
+
+    @Test("On Presto a schema set's materialized views come from information_schema in one query")
+    func prestoSchemaSetMaterializedViews() {
+        let statement = TrinoIntrospectionSQL.listMaterializedViews(
+            catalog: "hive", inSchemas: ["a", "o'brien"], flavor: .presto
+        )
+        #expect(statement.contains(#"SELECT table_name, table_schema FROM "hive".information_schema.materialized_views"#))
+        #expect(statement.contains("WHERE table_schema IN ('a', 'o''brien')"))
+    }
+
+    @Test("A quote that a combining mark follows is still doubled")
+    func combiningMarkCannotHideAQuote() {
+        #expect(TrinoIntrospectionSQL.quoteLiteral("x'\u{301} OR true--") == "'x''\u{301} OR true--'")
+        #expect(TrinoIntrospectionSQL.quoteIdentifier("a\"\u{301}") == "\"a\"\"\u{301}\"")
+    }
+}
diff --git a/apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleSchemaQueries.swift b/apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleSchemaQueries.swift
index 21a7a03bc7..f190d8ac0d 100644
--- a/apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleSchemaQueries.swift
+++ b/apple/Packages/TableProOracle/Sources/TableProOracleCore/OracleSchemaQueries.swift
@@ -8,12 +8,15 @@ public struct OracleTableRow: Sendable, Equatable {
     /// count.
     public let isPartitioned: Bool
     public let partitionCount: Int?
+    // Nil from a one-schema listing, which does not project the owner.
+    public let schema: String?
 
-    public init(name: String, isView: Bool, isPartitioned: Bool = false, partitionCount: Int? = nil) {
+    public init(name: String, isView: Bool, isPartitioned: Bool = false, partitionCount: Int? = nil, schema: String? = nil) {
         self.name = name
         self.isView = isView
         self.isPartitioned = isPartitioned
         self.partitionCount = partitionCount
+        self.schema = schema
     }
 }
 
@@ -188,6 +191,29 @@ public enum OracleSchemaQueries {
             """
     }
 
+    // ORA-01795: an IN list holds at most 1,000 expressions.
+    public static let maxOwnersPerTableListing = 1_000
+
+    // Names the owners so the read skips SYS and the other Oracle-maintained schemas the host drops.
+    public static func tables(inSchemas schemas: [String]) -> [String] {
+        stride(from: 0, to: schemas.count, by: maxOwnersPerTableListing).map { start in
+            let chunk = schemas[start..<min(start + maxOwnersPerTableListing, schemas.count)]
+            let owners = chunk.map { "'\(escapeLiteral($0))'" }.joined(separator: ", ")
+            return """
+                SELECT t.table_name, 'BASE TABLE' AS table_type,
+                       CASE WHEN pt.table_name IS NULL THEN 'N' ELSE 'Y' END AS is_partitioned,
+                       CASE WHEN pt.interval IS NULL THEN pt.partition_count END AS partition_count,
+                       t.owner
+                FROM \(OracleDictionary.allTables) t
+                LEFT JOIN \(OracleDictionary.allPartTables) pt ON pt.owner = t.owner AND pt.table_name = t.table_name
+                WHERE t.owner IN (\(owners))
+                UNION ALL
+                SELECT view_name, 'VIEW', 'N', NULL, owner FROM \(OracleDictionary.allViews) WHERE owner IN (\(owners))
+                ORDER BY 5, 1
+                """
+        }
+    }
+
     /// One partitioned table's partitions. `HIGH_VALUE` is deliberately absent: it is a LONG
     /// column, the datatype this driver already avoids reading because OracleNIO cannot decode it,
     /// so an Oracle partition states its position rather than its bound.
@@ -476,7 +502,8 @@ public enum OracleSchemaQueries {
             name: name,
             isView: row[safe: 1]?.stringValue == "VIEW",
             isPartitioned: row[safe: 2]?.stringValue == "Y",
-            partitionCount: row[safe: 3]?.stringValue.flatMap(Int.init)
+            partitionCount: row[safe: 3]?.stringValue.flatMap(Int.init),
+            schema: row[safe: 4]?.stringValue
         )
     }
 
diff --git a/apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleDictionaryQualificationTests.swift b/apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleDictionaryQualificationTests.swift
index 83de0c5bbe..e885dc028b 100644
--- a/apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleDictionaryQualificationTests.swift
+++ b/apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleDictionaryQualificationTests.swift
@@ -32,6 +32,7 @@ struct OracleDictionaryQualificationTests {
             ("serverVersion", OracleSchemaQueries.serverVersion),
             ("users", OracleSchemaQueries.users),
             ("tables", OracleSchemaQueries.tables(schema: "HR")),
+            ("tables in schemas", OracleSchemaQueries.tables(inSchemas: ["HR", "SALES"]).joined(separator: "\n")),
             ("partitions", OracleSchemaQueries.partitions(schema: "HR", table: "T")),
             ("subpartitions", OracleSchemaQueries.subpartitions(schema: "HR", table: "T")),
             ("columns 11g", OracleSchemaQueries.columns(schema: "HR", table: "T", release: OracleServerRelease(major: 11))),
diff --git a/apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleSchemaQueriesTests.swift b/apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleSchemaQueriesTests.swift
index 53d1372be5..713700aece 100644
--- a/apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleSchemaQueriesTests.swift
+++ b/apple/Packages/TableProOracle/Tests/TableProOracleCoreTests/OracleSchemaQueriesTests.swift
@@ -14,6 +14,40 @@ struct OracleSchemaQueriesTests {
         #expect(OracleSchemaQueries.foreignKeys(schema: owner, table: "T").contains("ac.OWNER = 'O''BRIEN'"))
     }
 
+    @Test func testQuoteFollowedByACombiningMarkIsEscaped() {
+        let owner = "O'\u{301} OR 1=1 --"
+        #expect(OracleSchemaQueries.escapeLiteral(owner) == "O''\u{301} OR 1=1 --")
+        #expect(OracleSchemaQueries.quoteIdentifier("A\"\u{301}") == "\"A\"\"\u{301}\"")
+    }
+
+    @Test func testSchemaSetListingNamesEveryOwnerAndProjectsIt() throws {
+        let statements = OracleSchemaQueries.tables(inSchemas: ["HR", "O'BRIEN"])
+        let sql = try #require(statements.first)
+        #expect(statements.count == 1)
+        #expect(sql.contains("WHERE t.owner IN ('HR', 'O''BRIEN')"))
+        #expect(sql.contains("FROM SYS.ALL_VIEWS WHERE owner IN ('HR', 'O''BRIEN')"))
+        #expect(sql.contains("SELECT view_name, 'VIEW', 'N', NULL, owner"))
+        #expect(sql.contains("ORDER BY 5, 1"))
+    }
+
+    @Test func testSchemaSetListingSplitsPastTheInListLimit() {
+        let owners = (0..<2_001).map { "U\($0)" }
+        let statements = OracleSchemaQueries.tables(inSchemas: owners)
+        #expect(statements.count == 3)
+        #expect(statements[0].contains("'U999')") && !statements[0].contains("'U1000'"))
+        #expect(statements[1].contains("IN ('U1000', ") && statements[1].contains("'U1999')"))
+        #expect(statements[2].contains("IN ('U2000')"))
+        #expect(OracleSchemaQueries.tables(inSchemas: []).isEmpty)
+    }
+
+    @Test func testParseTableRowReadsTheOwnerOfASchemaSetRow() {
+        let row = OracleSchemaQueries.parseTableRow([
+            .string("ORDERS"), .string("BASE TABLE"), .string("N"), .null, .string("SALES")
+        ])
+        #expect(row == OracleTableRow(name: "ORDERS", isView: false, schema: "SALES"))
+        #expect(OracleSchemaQueries.parseTableRow([.string("ORDERS"), .string("BASE TABLE")])?.schema == nil)
+    }
+
     @Test func testTableNameIsEscaped() {
         let sql = OracleSchemaQueries.columns(schema: "HR", table: "IT'S", release: release23ai)
         #expect(sql.contains("c.TABLE_NAME = 'IT''S'"))
diff --git a/apple/Plugins/DamengDriverPlugin/DamengPluginDriver+Schema.swift b/apple/Plugins/DamengDriverPlugin/DamengPluginDriver+Schema.swift
index 891f70e18b..836c24be74 100644
--- a/apple/Plugins/DamengDriverPlugin/DamengPluginDriver+Schema.swift
+++ b/apple/Plugins/DamengDriverPlugin/DamengPluginDriver+Schema.swift
@@ -22,6 +22,34 @@ extension DamengPluginDriver {
         }
     }
 
+    func fetchTables(inSchemas schemas: [String]) async throws -> [PluginTableInfo]? {
+        // DM8 follows Oracle here, so each statement stays under Oracle's 1,000-expression IN limit.
+        let chunkSize = 1_000
+        var tables: [PluginTableInfo] = []
+        for start in stride(from: 0, to: schemas.count, by: chunkSize) {
+            let owners = Array(schemas[start..<min(start + chunkSize, schemas.count)])
+            let placeholders = Array(repeating: "?", count: owners.count).joined(separator: ", ")
+            let result = try await executeParameterized(
+                query: """
+                    SELECT TABLE_NAME, 'TABLE' AS TABLE_TYPE, OWNER
+                    FROM SYS.ALL_TABLES
+                    WHERE OWNER IN (\(placeholders))
+                    UNION ALL
+                    SELECT VIEW_NAME, 'VIEW', OWNER
+                    FROM SYS.ALL_VIEWS
+                    WHERE OWNER IN (\(placeholders))
+                    ORDER BY 3, 1
+                    """,
+                parameters: (owners + owners).map { .text($0) }
+            )
+            tables += result.rows.compactMap { row in
+                guard let name = row[safe: 0]?.asText, let owner = row[safe: 2]?.asText else { return nil }
+                return PluginTableInfo(name: name, type: row[safe: 1]?.asText ?? "TABLE", schema: owner)
+            }
+        }
+        return tables
+    }
+
     func fetchColumns(table: String, schema: String?) async throws -> [PluginColumnInfo] {
         let owner = effectiveSchema(schema)
         let result = try await executeParameterized(
diff --git a/apple/Plugins/HanaDriverPlugin/HanaCatalogMapping.swift b/apple/Plugins/HanaDriverPlugin/HanaCatalogMapping.swift
index 77d9b2ff4f..c48c8b4c4a 100644
--- a/apple/Plugins/HanaDriverPlugin/HanaCatalogMapping.swift
+++ b/apple/Plugins/HanaDriverPlugin/HanaCatalogMapping.swift
@@ -190,6 +190,13 @@ enum HanaCatalogMapping {
         }
     }
 
+    static func tablesInSchemas(from rows: [[PluginCellValue]]) -> [PluginTableInfo] {
+        rows.compactMap { row in
+            guard let schema = row[safe: 3]?.asText else { return nil }
+            return tables(from: [row], schema: schema).first
+        }
+    }
+
     static func columns(from rows: [[PluginCellValue]]) -> [HanaCatalogColumn] {
         rows.compactMap(HanaCatalogColumn.init(row:))
     }
diff --git a/apple/Plugins/HanaDriverPlugin/HanaCatalogQueries.swift b/apple/Plugins/HanaDriverPlugin/HanaCatalogQueries.swift
index a63811b7e3..f971d59f67 100644
--- a/apple/Plugins/HanaDriverPlugin/HanaCatalogQueries.swift
+++ b/apple/Plugins/HanaDriverPlugin/HanaCatalogQueries.swift
@@ -22,6 +22,21 @@ enum HanaCatalogQueries {
             """
     }
 
+    static func tables(inSchemas schemas: [String]) -> String {
+        let owners = schemas.map(HanaSQL.quoteLiteral).joined(separator: ", ")
+        return """
+            SELECT TABLE_NAME, 'TABLE' AS OBJECT_TYPE, COMMENTS, SCHEMA_NAME
+            FROM SYS.TABLES
+            WHERE SCHEMA_NAME IN (\(owners))
+              AND IS_USER_DEFINED_TYPE = 'FALSE'
+            UNION ALL
+            SELECT VIEW_NAME, 'VIEW', COMMENTS, SCHEMA_NAME
+            FROM SYS.VIEWS
+            WHERE SCHEMA_NAME IN (\(owners))
+            ORDER BY 4, 1
+            """
+    }
+
     static func columns(schema: String, table: String?) -> String {
         let owner = HanaSQL.quoteLiteral(schema)
         let tableFilter = table.map { "\n      AND TABLE_NAME = \(HanaSQL.quoteLiteral($0))" } ?? ""
diff --git a/apple/Plugins/HanaDriverPlugin/HanaPluginDriver+Catalog.swift b/apple/Plugins/HanaDriverPlugin/HanaPluginDriver+Catalog.swift
index bd63157296..90b617b812 100644
--- a/apple/Plugins/HanaDriverPlugin/HanaPluginDriver+Catalog.swift
+++ b/apple/Plugins/HanaDriverPlugin/HanaPluginDriver+Catalog.swift
@@ -26,6 +26,12 @@ extension HanaPluginDriver {
         return HanaCatalogMapping.tables(from: result.rows, schema: owner)
     }
 
+    func fetchTables(inSchemas schemas: [String]) async throws -> [PluginTableInfo]? {
+        guard !schemas.isEmpty else { return [] }
+        let result = try await execute(query: HanaCatalogQueries.tables(inSchemas: schemas))
+        return HanaCatalogMapping.tablesInSchemas(from: result.rows)
+    }
+
     func fetchColumns(table: String, schema: String?) async throws -> [PluginColumnInfo] {
         try await catalogColumns(table: table, schema: effectiveSchema(schema)).map(\.pluginColumn)
     }
diff --git a/apple/Plugins/OracleDriverPlugin/OraclePlugin.swift b/apple/Plugins/OracleDriverPlugin/OraclePlugin.swift
index 020a18bdb2..be84ce0616 100644
--- a/apple/Plugins/OracleDriverPlugin/OraclePlugin.swift
+++ b/apple/Plugins/OracleDriverPlugin/OraclePlugin.swift
@@ -487,14 +487,26 @@ final class OraclePluginDriver: PluginDatabaseDriver, @unchecked Sendable {
 
     func fetchTables(schema: String?) async throws -> [PluginTableInfo] {
         let result = try await rawQuery(OracleSchemaQueries.tables(schema: effectiveSchema(schema)))
-        return result.rows.compactMap(OracleSchemaQueries.parseTableRow).map { row in
-            PluginTableInfo(
-                name: row.name,
-                type: row.isView ? "VIEW" : (row.isPartitioned ? "PARTITIONED TABLE" : "TABLE"),
-                comment: nil,
-                partitionCount: row.partitionCount
-            )
+        return result.rows.compactMap(OracleSchemaQueries.parseTableRow).map(Self.tableInfo)
+    }
+
+    func fetchTables(inSchemas schemas: [String]) async throws -> [PluginTableInfo]? {
+        var tables: [PluginTableInfo] = []
+        for query in OracleSchemaQueries.tables(inSchemas: schemas) {
+            let result = try await rawQuery(query)
+            tables += result.rows.compactMap(OracleSchemaQueries.parseTableRow).map(Self.tableInfo)
         }
+        return tables
+    }
+
+    private static func tableInfo(_ row: OracleTableRow) -> PluginTableInfo {
+        PluginTableInfo(
+            name: row.name,
+            type: row.isView ? "VIEW" : (row.isPartitioned ? "PARTITIONED TABLE" : "TABLE"),
+            schema: row.schema,
+            comment: nil,
+            partitionCount: row.partitionCount
+        )
     }
 
     func fetchColumns(table: String, schema: String?) async throws -> [PluginColumnInfo] {
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/CockroachPluginDriver.swift b/apple/Plugins/PostgreSQLDriverPlugin/CockroachPluginDriver.swift
index 8a9239a530..bc8676133a 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/CockroachPluginDriver.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/CockroachPluginDriver.swift
@@ -61,20 +61,14 @@ final class CockroachPluginDriver: LibPQBackedDriver, @unchecked Sendable {
     // MARK: - Schema
 
     func fetchTables(schema: String?) async throws -> [PluginTableInfo] {
-        let resolvedSchema = schema ?? core.currentSchema
-        let query = """
-            SELECT table_name, table_type
-            FROM information_schema.tables
-            WHERE table_schema = \(PostgreSQLObjectQueries.quoteLiteral(resolvedSchema))
-            ORDER BY table_name
-            """
-        let result = try await execute(query: query)
-        return result.rows.compactMap { row -> PluginTableInfo? in
-            guard let name = row[0].asText else { return nil }
-            let typeStr = (row[1].asText ?? "BASE TABLE").uppercased()
-            let type = typeStr.contains("VIEW") ? "VIEW" : "TABLE"
-            return PluginTableInfo(name: name, type: type)
-        }
+        let result = try await execute(query: CockroachRelationSQL.tablesQuery(schema: schema ?? core.currentSchema))
+        return result.rows.compactMap { CockroachRelationSQL.table(fromRow: $0.map(\.asText)) }
+    }
+
+    func fetchTables(inSchemas schemas: [String]) async throws -> [PluginTableInfo]? {
+        guard !schemas.isEmpty else { return [] }
+        let result = try await execute(query: CockroachRelationSQL.tablesQuery(inSchemas: schemas))
+        return result.rows.compactMap { CockroachRelationSQL.table(fromRow: $0.map(\.asText)) }
     }
 
     func fetchColumns(table: String, schema: String?) async throws -> [PluginColumnInfo] {
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/CockroachRelationSQL.swift b/apple/Plugins/PostgreSQLDriverPlugin/CockroachRelationSQL.swift
index 3ebff383dc..764f0869a9 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/CockroachRelationSQL.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/CockroachRelationSQL.swift
@@ -17,6 +17,30 @@ import TableProPluginKit
 /// `schema` is non-optional on purpose: the `?? currentSchema` fallback stays at the call site,
 /// where the driver knows its own session, so no caller can reach these builders without one.
 public enum CockroachRelationSQL {
+    public static func tablesQuery(schema: String) -> String {
+        """
+        SELECT table_name, table_type
+        FROM information_schema.tables
+        WHERE table_schema = \(PostgreSQLObjectQueries.quoteLiteral(schema))
+        ORDER BY table_name
+        """
+    }
+
+    public static func tablesQuery(inSchemas schemas: [String]) -> String {
+        """
+        SELECT table_name, table_type, table_schema
+        FROM information_schema.tables
+        WHERE \(PostgreSQLObjectQueries.membership("table_schema", in: schemas))
+        ORDER BY table_schema, table_name
+        """
+    }
+
+    public static func table(fromRow row: [String?]) -> PluginTableInfo? {
+        guard let name = row[safe: 0] ?? nil else { return nil }
+        let listedType = ((row[safe: 1] ?? nil) ?? "BASE TABLE").uppercased()
+        return PluginTableInfo(name: name, type: listedType.contains("VIEW") ? "VIEW" : "TABLE", schema: row[safe: 2] ?? nil)
+    }
+
     public static func showCreateTable(table: String, schema: String) -> String {
         "SHOW CREATE TABLE \(PostgreSQLObjectQueries.qualifiedName(schema: schema, name: table))"
     }
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLObjectQueries.swift b/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLObjectQueries.swift
index 6387b57c35..8029f03ef7 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLObjectQueries.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/PostgreSQLObjectQueries.swift
@@ -26,14 +26,15 @@ nonisolated public enum PostgreSQLObjectQueries {
     /// A NUL is dropped rather than escaped. libpq takes a NUL-terminated C string, so a NUL would
     /// truncate the statement, and PostgreSQL rejects `\000` in a literal outright ("invalid byte
     /// sequence for encoding UTF8"). This matches the PluginKit default's own NUL strip.
+    // Every match is by code unit: a grapheme match skips a quote or backslash a combining mark follows.
     public static func quoteLiteral(_ value: String) -> String {
-        let stripped = value.replacingOccurrences(of: "\0", with: "")
-        guard stripped.contains("\\") else { return "'\(escapeQuotes(stripped))'" }
-        return "E'\(escapeQuotes(stripped.replacingOccurrences(of: "\\", with: "\\\\")))'"
+        let stripped = value.replacingOccurrences(of: "\0", with: "", options: .literal)
+        guard stripped.unicodeScalars.contains("\\") else { return "'\(escapeQuotes(stripped))'" }
+        return "E'\(escapeQuotes(stripped.replacingOccurrences(of: "\\", with: "\\\\", options: .literal)))'"
     }
 
     private static func escapeQuotes(_ value: String) -> String {
-        value.replacingOccurrences(of: "'", with: "''")
+        value.replacingOccurrences(of: "'", with: "''", options: .literal)
     }
 
     /// `IN ()` does not parse, so an empty list is `false`.
@@ -43,7 +44,7 @@ nonisolated public enum PostgreSQLObjectQueries {
     }
 
     public static func quoteIdentifier(_ name: String) -> String {
-        "\"\(name.replacingOccurrences(of: "\"", with: "\"\""))\""
+        "\"\(name.replacingOccurrences(of: "\"", with: "\"\"", options: .literal))\""
     }
 
     public static func qualifiedName(schema: String, name: String) -> String {
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/RedshiftPluginDriver.swift b/apple/Plugins/PostgreSQLDriverPlugin/RedshiftPluginDriver.swift
index 1ee4b3e3d8..9d00a179ee 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/RedshiftPluginDriver.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/RedshiftPluginDriver.swift
@@ -91,6 +91,17 @@ final class RedshiftPluginDriver: LibPQBackedDriver, @unchecked Sendable {
         return merged.sorted { $0.name.localizedStandardCompare($1.name) == .orderedAscending }
     }
 
+    func fetchTables(inSchemas schemas: [String]) async throws -> [PluginTableInfo]? {
+        guard !schemas.isEmpty else { return [] }
+        let result = try await execute(query: RedshiftTableCatalog.listingQuery(inSchemas: schemas))
+        var tables = result.rows.compactMap { RedshiftTableCatalog.table(fromListingRow: $0.map(\.asText)) }
+        for schema in schemas where isExternalSchema(schema) {
+            let localNames = Set(tables.lazy.filter { $0.schema == schema }.map(\.name))
+            tables += await fetchExternalTables(schema: schema).filter { !localNames.contains($0.name) }
+        }
+        return tables
+    }
+
     private func fetchExternalTables(schema: String) async -> [PluginTableInfo] {
         do {
             let result = try await execute(
diff --git a/apple/Plugins/PostgreSQLDriverPlugin/RedshiftTableCatalog.swift b/apple/Plugins/PostgreSQLDriverPlugin/RedshiftTableCatalog.swift
index b238eefe7b..cdeb9cdf97 100644
--- a/apple/Plugins/PostgreSQLDriverPlugin/RedshiftTableCatalog.swift
+++ b/apple/Plugins/PostgreSQLDriverPlugin/RedshiftTableCatalog.swift
@@ -12,10 +12,19 @@ nonisolated enum RedshiftTableCatalog {
         """
     }
 
+    static func listingQuery(inSchemas schemas: [String]) -> String {
+        """
+        SELECT table_name, table_type, table_schema
+        FROM information_schema.tables
+        WHERE \(PostgreSQLObjectQueries.membership("table_schema", in: schemas))
+        ORDER BY table_schema, table_name
+        """
+    }
+
     static func table(fromListingRow row: [String?]) -> PluginTableInfo? {
         guard let name = row[safe: 0] ?? nil else { return nil }
         let listedType = (row[safe: 1] ?? nil) ?? "BASE TABLE"
-        return PluginTableInfo(name: name, type: listedType.contains("VIEW") ? "VIEW" : "TABLE")
+        return PluginTableInfo(name: name, type: listedType.contains("VIEW") ? "VIEW" : "TABLE", schema: row[safe: 2] ?? nil)
     }
 
     static func keysQuery(schema: String, table: String) -> String {
diff --git a/apple/Plugins/PostgreSQLDriverTests/CockroachRelationSQLTests.swift b/apple/Plugins/PostgreSQLDriverTests/CockroachRelationSQLTests.swift
index 5f36165c44..a5d5ab9914 100644
--- a/apple/Plugins/PostgreSQLDriverTests/CockroachRelationSQLTests.swift
+++ b/apple/Plugins/PostgreSQLDriverTests/CockroachRelationSQLTests.swift
@@ -9,9 +9,28 @@
 //
 
 import Foundation
+import TableProPluginKit
 import Testing
 
 struct CockroachRelationSQLTests {
+    @Test("A schema set is listed in one query that names each schema and projects it")
+    func schemaSetListing() {
+        let statement = CockroachRelationSQL.tablesQuery(inSchemas: ["public", "o'brien"])
+        #expect(statement.contains("WHERE table_schema IN ('public', 'o''brien')"))
+        #expect(statement.contains("SELECT table_name, table_type, table_schema"))
+        #expect(CockroachRelationSQL.tablesQuery(inSchemas: []).contains("WHERE false"))
+    }
+
+    @Test("A listing row keeps its schema when the query projects one")
+    func listingRowSchema() {
+        let view = CockroachRelationSQL.table(fromRow: ["daily", "VIEW", "analytics"])
+        #expect(view?.name == "daily")
+        #expect(view?.type == "VIEW")
+        #expect(view?.schema == "analytics")
+        #expect(CockroachRelationSQL.table(fromRow: ["orders", "BASE TABLE"])?.schema == nil)
+        #expect(CockroachRelationSQL.table(fromRow: [nil, "VIEW"]) == nil)
+    }
+
     @Test("SHOW CREATE TABLE names the requested schema")
     func showCreateTableNamesRequestedSchema() {
         let statement = CockroachRelationSQL.showCreateTable(table: "orders", schema: "analytics")
diff --git a/apple/Plugins/PostgreSQLDriverTests/PostgreSQLLiteralQuotingTests.swift b/apple/Plugins/PostgreSQLDriverTests/PostgreSQLLiteralQuotingTests.swift
index 13e472686c..d76245531a 100644
--- a/apple/Plugins/PostgreSQLDriverTests/PostgreSQLLiteralQuotingTests.swift
+++ b/apple/Plugins/PostgreSQLDriverTests/PostgreSQLLiteralQuotingTests.swift
@@ -38,6 +38,10 @@ struct PostgreSQLLiteralQuotingTests {
             PostgreSQLSchemaQueries.databaseMetadata(database: schema),
             PostgreSQLObjectQueries.userDefinedTypeList(schemas: [schema], capabilities: caps),
             CockroachRelationSQL.columnsQuery(schema: schema, table: table),
+            CockroachRelationSQL.tablesQuery(schema: schema),
+            CockroachRelationSQL.tablesQuery(inSchemas: [schema, table]),
+            RedshiftTableCatalog.listingQuery(schema: schema),
+            RedshiftTableCatalog.listingQuery(inSchemas: [schema, table]),
             PostgreSQLSchemaQueries.fetchPartitions(schema: schema, table: table),
             PostgreSQLSchemaQueries.columnsQuery(
                 schema: schema, table: table, capabilities: caps, includeMaterializedViews: true
@@ -141,6 +145,13 @@ struct PostgreSQLLiteralQuotingTests {
         )
     }
 
+    @Test("A quote or backslash that a combining mark follows is still escaped")
+    func combiningMarkCannotHideAnEscape() {
+        #expect(PostgreSQLObjectQueries.quoteLiteral("x'\u{301} OR true--") == "'x''\u{301} OR true--'")
+        #expect(PostgreSQLObjectQueries.quoteLiteral("x\\\u{301}") == "E'x\\\\\u{301}'")
+        #expect(PostgreSQLObjectQueries.quoteIdentifier("a\"\u{301}") == "\"a\"\"\u{301}\"")
+    }
+
     @Test("A NUL is dropped on both arms rather than reaching libpq")
     func nulIsStripped() {
         #expect(PostgreSQLObjectQueries.quoteLiteral("a\0b") == "'ab'")
diff --git a/apple/Plugins/PostgreSQLDriverTests/RedshiftTableCatalogTests.swift b/apple/Plugins/PostgreSQLDriverTests/RedshiftTableCatalogTests.swift
index b095207ec3..13b8e7b376 100644
--- a/apple/Plugins/PostgreSQLDriverTests/RedshiftTableCatalogTests.swift
+++ b/apple/Plugins/PostgreSQLDriverTests/RedshiftTableCatalogTests.swift
@@ -14,6 +14,15 @@ struct RedshiftTableCatalogTests {
         #expect(RedshiftTableCatalog.table(fromListingRow: [nil, "VIEW"]) == nil)
     }
 
+    @Test("a schema set is listed in one query that names each schema and projects it")
+    func schemaSetListing() {
+        let statement = RedshiftTableCatalog.listingQuery(inSchemas: ["public", "sales"])
+        #expect(statement.contains("WHERE table_schema IN ('public', 'sales')"))
+        #expect(statement.contains("SELECT table_name, table_type, table_schema"))
+        #expect(RedshiftTableCatalog.table(fromListingRow: ["orders", "BASE TABLE", "sales"])?.schema == "sales")
+        #expect(RedshiftTableCatalog.table(fromListingRow: ["orders", "BASE TABLE"])?.schema == nil)
+    }
+
     @Test("a distribution key and the sort key columns in sort order")
     func distributionAndSortKeys() {
         let rows: [[String?]] = [
diff --git a/apple/Plugins/TrinoDriverPlugin/TrinoPluginDriver+Schema.swift b/apple/Plugins/TrinoDriverPlugin/TrinoPluginDriver+Schema.swift
index 201c296014..6fc6b3733e 100644
--- a/apple/Plugins/TrinoDriverPlugin/TrinoPluginDriver+Schema.swift
+++ b/apple/Plugins/TrinoDriverPlugin/TrinoPluginDriver+Schema.swift
@@ -72,6 +72,26 @@ extension TrinoPluginDriver {
         return tables
     }
 
+    func fetchTables(inSchemas schemas: [String]) async throws -> [PluginTableInfo]? {
+        guard let catalog = currentCatalog, !schemas.isEmpty else { return [] }
+        let result = try await execute(query: TrinoIntrospectionSQL.listTables(catalog: catalog, inSchemas: schemas))
+        let materialized = await materializedViews(catalog: catalog, schemas: schemas)
+
+        var tables: [PluginTableInfo] = []
+        var seen: Set<TrinoRelationKey> = []
+        for row in result.rows {
+            guard let name = firstText(row), let schema = text(row[safe: 2]) else { continue }
+            let key = TrinoRelationKey(schema: schema, name: name)
+            seen.insert(key)
+            let type = materialized.contains(key) ? "MATERIALIZED VIEW" : Self.tableType(text(row[safe: 1]))
+            tables.append(PluginTableInfo(name: name, type: type, schema: schema, comment: nil))
+        }
+        for key in materialized.sorted() where !seen.contains(key) {
+            tables.append(PluginTableInfo(name: key.name, type: "MATERIALIZED VIEW", schema: key.schema, comment: nil))
+        }
+        return tables
+    }
+
     func fetchColumns(table: String, schema: String?) async throws -> [PluginColumnInfo] {
         guard let catalog = currentCatalog, let targetSchema = resolveSchema(schema) else { return [] }
         let result = try await execute(
@@ -187,6 +207,24 @@ extension TrinoPluginDriver {
         return Set(result.rows.compactMap { firstText($0) })
     }
 
+    private func materializedViews(catalog: String, schemas: [String]) async -> Set<TrinoRelationKey> {
+        let sql = TrinoIntrospectionSQL.listMaterializedViews(catalog: catalog, inSchemas: schemas, flavor: flavor)
+        if let result = try? await execute(query: sql) {
+            return Set(result.rows.compactMap { row in
+                guard let name = firstText(row), let schema = text(row[safe: 1]) else { return nil }
+                return TrinoRelationKey(schema: schema, name: name)
+            })
+        }
+        // One schema the connector cannot read fails the whole query; per schema, only that one is lost.
+        var views: Set<TrinoRelationKey> = []
+        for schema in schemas {
+            for name in await materializedViewNames(catalog: catalog, schema: schema) {
+                views.insert(TrinoRelationKey(schema: schema, name: name))
+            }
+        }
+        return views
+    }
+
     private static func tableType(_ kind: String?) -> String {
         (kind ?? "").uppercased().contains("VIEW") ? "VIEW" : "TABLE"
     }
@@ -196,3 +234,12 @@ extension TrinoPluginDriver {
         return value.trimmingCharacters(in: .whitespaces).isEmpty ? nil : value
     }
 }
+
+private struct TrinoRelationKey: Hashable, Comparable {
+    let schema: String
+    let name: String
+
+    static func < (lhs: TrinoRelationKey, rhs: TrinoRelationKey) -> Bool {
+        (lhs.schema, lhs.name) < (rhs.schema, rhs.name)
+    }
+}
diff --git a/apple/TablePro/Core/Database/CatalogTableListing.swift b/apple/TablePro/Core/Database/CatalogTableListing.swift
index 94ab535415..5cea70047e 100644
--- a/apple/TablePro/Core/Database/CatalogTableListing.swift
+++ b/apple/TablePro/Core/Database/CatalogTableListing.swift
@@ -78,7 +78,7 @@ internal enum CatalogTableListing {
         case .all:
             return try await tablesInEverySchema(in: scope, excludingSchemas: excluded, metadata: metadata)
         case .only(let kept):
-            return try await tables(inKeptSchemas: kept.subtracting(excluded), scope: scope, metadata: metadata)
+            return try await tables(inListedSchemas: kept.subtracting(excluded).sorted(), scope: scope, metadata: metadata)
         }
     }
 
@@ -100,22 +100,36 @@ internal enum CatalogTableListing {
         let schemas = try await metadata.withMetadataDriver(scope: scope, workload: .bulk) { driver in
             try await driver.fetchSchemas()
         }
-        return try await tables(inSchemas: schemas.filter { !excluded.contains($0) }, scope: scope, metadata: metadata)
+        // The set read keeps the excluded schemas out of the engine's query.
+        var seen: Set<String> = []
+        let kept = schemas.filter { !excluded.contains($0) && seen.insert($0).inserted }
+        return try await tables(inListedSchemas: kept, scope: scope, metadata: metadata)
     }
 
-    /// The kept set is the schema list, so no schema list is read. An engine without a set read is
+    /// No schema list is read: the caller's names are the list. An engine without a set read is
     /// asked schema by schema, never through the all-schema listing, so a hidden schema is never read.
     private static func tables(
-        inKeptSchemas schemas: Set<String>,
+        inListedSchemas names: [String],
         scope: DatabaseScope,
         metadata: ScopedMetadataProviding
     ) async throws -> Result {
-        guard !schemas.isEmpty else { return Result(tables: [], unlistedSchemas: []) }
-        let names = schemas.sorted()
-        let listed = try await metadata.withMetadataDriver(scope: scope, workload: .bulk) { driver in
-            try await driver.fetchTables(inSchemas: names)
+        guard !names.isEmpty else { return Result(tables: [], unlistedSchemas: []) }
+        let listed: [TableInfo]?
+        do {
+            listed = try await metadata.withMetadataDriver(scope: scope, workload: .bulk) { driver in
+                try await driver.fetchTables(inSchemas: names)
+            }
+        } catch is CancellationError {
+            throw CancellationError()
+        } catch let error as DatabaseError {
+            throw error
+        } catch {
+            // One schema the set read cannot take fails it for all; per schema, only that one goes unlisted.
+            logger.warning("[catalog] set read failed, listing per schema error=\(error.publicLogShape, privacy: .public)")
+            listed = nil
         }
         if let listed {
+            let schemas = Set(names)
             let tables = listed.filter { table in table.schema.map { schemas.contains($0) } ?? true }
             return Result(tables: tables, unlistedSchemas: [])
         }
diff --git a/apple/TableProTests/Core/Autocomplete/SQLSchemaProviderTests.swift b/apple/TableProTests/Core/Autocomplete/SQLSchemaProviderTests.swift
index cd51984f00..2856e120f5 100644
--- a/apple/TableProTests/Core/Autocomplete/SQLSchemaProviderTests.swift
+++ b/apple/TableProTests/Core/Autocomplete/SQLSchemaProviderTests.swift
@@ -247,6 +247,16 @@ final class MockDatabaseDriver: DatabaseDriver, SchemaSwitchable, @unchecked Sen
         return snapshot
     }
 
+    // Nil answers nil, so the host lists each schema itself.
+    var tablesInSchemasError: Error?
+    var fetchTablesInSchemasCallCount = 0
+
+    func fetchTables(inSchemas schemas: [String]) async throws -> [TableInfo]? {
+        fetchTablesInSchemasCallCount += 1
+        if let tablesInSchemasError { throw tablesInSchemasError }
+        return nil
+    }
+
     /// Nil leaves the driver without a single call, so the host lists each schema itself.
     var allSchemaTablesToReturn: [TableInfo]?
     var allSchemaTablesError: Error?
diff --git a/apple/TableProTests/Core/Database/CatalogTableListingTests.swift b/apple/TableProTests/Core/Database/CatalogTableListingTests.swift
index 4d7a9d61dd..60f065dfb9 100644
--- a/apple/TableProTests/Core/Database/CatalogTableListingTests.swift
+++ b/apple/TableProTests/Core/Database/CatalogTableListingTests.swift
@@ -112,6 +112,38 @@ struct CatalogTableListingTests {
         #expect(kept.unlistedSchemas == ["attendance"])
     }
 
+    @Test("A set read that fails falls back to one read per schema, which names only the schema that fails")
+    func failedSetReadFallsBackPerSchema() async throws {
+        let driver = MockDatabaseDriver()
+        driver.schemasToReturn = ["public", "attendance"]
+        driver.tablesInSchemasError = ListingFailed()
+        driver.schemaTablesToReturn = ["public": [table("users", "public")]]
+        driver.schemaTablesErrors = ["attendance": ListingFailed()]
+
+        let listing = try await CatalogTableListing.tables(
+            in: scope, excludingSchemas: [], metadata: RecordingMetadataProvider(driver: driver)
+        )
+
+        #expect(driver.fetchTablesInSchemasCallCount == 1)
+        #expect(driver.fetchSchemaTablesCalls == ["public", "attendance"])
+        #expect(listing.tables.map(\.name) == ["users"])
+        #expect(listing.unlistedSchemas == ["attendance"])
+    }
+
+    @Test("A set read that loses the connection fails the listing without a read per schema")
+    func setReadLostConnectionFailsTheListing() async {
+        let driver = MockDatabaseDriver()
+        driver.schemasToReturn = ["public", "attendance"]
+        driver.tablesInSchemasError = DatabaseError.notConnected
+
+        await #expect(throws: DatabaseError.self) {
+            try await CatalogTableListing.tables(
+                in: scope, excludingSchemas: [], metadata: RecordingMetadataProvider(driver: driver)
+            )
+        }
+        #expect(driver.fetchSchemaTablesCalls.isEmpty)
+    }
+
     @Test("A lost connection fails the whole listing rather than reading as empty")
     func lostConnectionFailsTheListing() async {
         let driver = MockDatabaseDriver()
@@ -202,6 +234,32 @@ struct CatalogTableListingSelectionTests {
         #expect(result.unlistedSchemas.isEmpty)
     }
 
+    @Test("Every schema on an engine with a set read and no all-schema read takes the schema list and one set read")
+    func everySchemaThroughTheSetRead() async throws {
+        let driver = driver()
+        driver.schemasToReturn = ["c", "SYS", "a", "b"]
+        driver.answersTablesInSchemas = true
+
+        let result = try await listing(.all, excluding: ["SYS", "b"], on: driver)
+
+        #expect(driver.reads == ["allSchemaTables", "schemas", "tablesIn(c,a)"])
+        #expect(driver.perSchemaReads.isEmpty)
+        #expect(result.tables.map(\.name) == ["payslips", "orders"])
+        #expect(result.unlistedSchemas.isEmpty)
+    }
+
+    @Test("Every schema being excluded reads no tables at all")
+    func everySchemaExcludedReadsNoTables() async throws {
+        let driver = driver()
+        driver.schemasToReturn = ["SYS"]
+        driver.answersTablesInSchemas = true
+
+        let result = try await listing(.all, excluding: ["SYS"], on: driver)
+
+        #expect(driver.reads == ["allSchemaTables", "schemas"])
+        #expect(result.tables.isEmpty)
+    }
+
     @Test("A filtered listing never calls fetchTablesInAllSchemas, even on an engine that answers it")
     func filteredListingNeverReadsEverySchema() async throws {
         let driver = driver()
diff --git a/apple/changelog.d/3708.changed.md b/apple/changelog.d/3708.changed.md
new file mode 100644
index 0000000000..978a38f953
--- /dev/null
+++ b/apple/changelog.d/3708.changed.md
@@ -0,0 +1 @@
+Open Quickly lists tables in other schemas with one query on Oracle, Dameng, SAP HANA, Trino, CockroachDB and Redshift.
diff --git a/apple/scripts/probes/build-large-catalog-fixtures.sh b/apple/scripts/probes/build-large-catalog-fixtures.sh
new file mode 100755
index 0000000000..391a299ee0
--- /dev/null
+++ b/apple/scripts/probes/build-large-catalog-fixtures.sh
@@ -0,0 +1,191 @@
+#!/usr/bin/env bash
+#
+# Build the large-catalog fixtures from #3661, for timing the sidebar, Open Quickly and autocomplete
+# on a catalog of 50,000 tables.
+#
+#   postgres  database tablepro_large_catalog: 100 schemas lc_s001..lc_s100, 500 tables each
+#   oracle    100 users LC_S001..LC_S100, 500 tables each, in a gvenzl/oracle-free container
+#   mysql     database tablepro_large_catalog: 50,000 tables
+#
+# Every schema holds the same table names (lc_t001..lc_t500), as a tenant-per-schema database does,
+# so a search for lc_t250 matches once per schema.
+#
+# Usage:
+#   scripts/probes/build-large-catalog-fixtures.sh postgres [host] [port] [user]
+#   scripts/probes/build-large-catalog-fixtures.sh mysql [host] [port] [user]
+#   scripts/probes/build-large-catalog-fixtures.sh oracle [container] [system-password]
+#   scripts/probes/build-large-catalog-fixtures.sh drop-postgres|drop-mysql|drop-oracle [same arguments]
+#
+# Oracle runs SQL*Plus inside the container. Start one with:
+#   docker run -d --name oracle-large-catalog -p 1521:1521 -e ORACLE_PASSWORD=probe_pw \
+#     gvenzl/oracle-free:23-slim-faststart
+#
+# SCHEMAS, TABLES_PER_SCHEMA, TABLES (the MySQL table count) and DATABASE override the defaults.
+# Re-running skips a schema that already holds every table. Exits 3 when a client or server is missing.
+
+set -uo pipefail
+
+ENGINE="${1:-}"
+SCHEMAS="${SCHEMAS:-100}"
+TABLES_PER_SCHEMA="${TABLES_PER_SCHEMA:-500}"
+MYSQL_TABLES="${TABLES:-50000}"
+DATABASE="${DATABASE:-tablepro_large_catalog}"
+# Wide enough for the largest number, so an override never truncates two names into one.
+SCHEMA_WIDTH=$(( ${#SCHEMAS} > 3 ? ${#SCHEMAS} : 3 ))
+TABLE_WIDTH=$(( ${#TABLES_PER_SCHEMA} > 3 ? ${#TABLES_PER_SCHEMA} : 3 ))
+MYSQL_WIDTH=$(( ${#MYSQL_TABLES} > 5 ? ${#MYSQL_TABLES} : 5 ))
+
+need() {
+    command -v "$1" > /dev/null || {
+        echo "$1 not found" >&2
+        exit 3
+    }
+}
+
+elapsed_since() {
+    echo "$(( $(date +%s) - $1 ))s"
+}
+
+postgres_fixture() {
+    local host="${1:-127.0.0.1}" port="${2:-5432}" user="${3:-postgres}" started schema i
+    need psql
+    psql_at() {
+        psql -X -q -v ON_ERROR_STOP=1 -h "$host" -p "$port" -U "$user" -d "$1" "${@:2}"
+    }
+    psql_at postgres -Atc "SELECT 1" > /dev/null 2>&1 || {
+        echo "no PostgreSQL at $host:$port as $user" >&2
+        exit 3
+    }
+    if [ -z "$(psql_at postgres -Atc "SELECT 1 FROM pg_database WHERE datname = '$DATABASE'")" ]; then
+        psql_at postgres -c "CREATE DATABASE $DATABASE"
+    fi
+    started=$(date +%s)
+    for i in $(seq 1 "$SCHEMAS"); do
+        schema=$(printf 'lc_s%0*d' "$SCHEMA_WIDTH" "$i")
+        # One transaction per schema: 50,000 tables in one would overrun the shared lock table.
+        psql_at "$DATABASE" -c "
+            CREATE SCHEMA IF NOT EXISTS $schema;
+            DO \$\$
+            BEGIN
+                FOR t IN 1..$TABLES_PER_SCHEMA LOOP
+                    EXECUTE format('CREATE TABLE IF NOT EXISTS $schema.%I (id integer PRIMARY KEY, name text)',
+                                   'lc_t' || lpad(t::text, $TABLE_WIDTH, '0'));
+                END LOOP;
+            END
+            \$\$;" > /dev/null || exit 1
+        [ $((i % 10)) -eq 0 ] && echo "postgres: $i of $SCHEMAS schemas, $(elapsed_since "$started")"
+    done
+    echo "postgres: $(psql_at "$DATABASE" -Atc "SELECT count(*) FROM pg_tables WHERE schemaname LIKE 'lc!_s%' ESCAPE '!'") tables in $DATABASE"
+}
+
+postgres_drop() {
+    local host="${1:-127.0.0.1}" port="${2:-5432}" user="${3:-postgres}"
+    need psql
+    psql -X -q -h "$host" -p "$port" -U "$user" -d postgres -c "DROP DATABASE IF EXISTS $DATABASE"
+}
+
+mysql_fixture() {
+    local host="${1:-127.0.0.1}" port="${2:-3306}" user="${3:-root}" started batch i
+    need mysql
+    mysql_do() {
+        mysql --protocol=TCP -h "$host" -P "$port" -u "$user" "$@"
+    }
+    mysql_do -e "SELECT 1" > /dev/null 2>&1 || {
+        echo "no MySQL at $host:$port as $user (MYSQL_PWD supplies a password)" >&2
+        exit 3
+    }
+    mysql_do -e "CREATE DATABASE IF NOT EXISTS $DATABASE"
+    started=$(date +%s)
+    batch=""
+    for i in $(seq 1 "$MYSQL_TABLES"); do
+        batch+="CREATE TABLE IF NOT EXISTS $(printf 'lc_t%0*d' "$MYSQL_WIDTH" "$i") (id INT PRIMARY KEY, name VARCHAR(64));"
+        if [ $((i % 1000)) -eq 0 ] || [ "$i" -eq "$MYSQL_TABLES" ]; then
+            mysql_do "$DATABASE" -e "$batch" || exit 1
+            batch=""
+            [ $((i % 5000)) -eq 0 ] && echo "mysql: $i of $MYSQL_TABLES tables, $(elapsed_since "$started")"
+        fi
+    done
+    echo "mysql: $(mysql_do -N -e "SELECT count(*) FROM information_schema.tables WHERE table_schema = '$DATABASE'") tables in $DATABASE"
+}
+
+mysql_drop() {
+    local host="${1:-127.0.0.1}" port="${2:-3306}" user="${3:-root}"
+    need mysql
+    mysql --protocol=TCP -h "$host" -P "$port" -u "$user" -e "DROP DATABASE IF EXISTS $DATABASE"
+}
+
+oracle_sqlplus() {
+    docker exec -i "$1" sqlplus -s -L "system/$2@FREEPDB1"
+}
+
+oracle_check() {
+    need docker
+    printf 'SELECT 1 FROM dual;\nEXIT\n' | oracle_sqlplus "$1" "$2" > /dev/null 2>&1 || {
+        echo "no Oracle container $1 answering as system (see the header for docker run)" >&2
+        exit 3
+    }
+}
+
+oracle_fixture() {
+    local container="${1:-oracle-large-catalog}" password="${2:-probe_pw}" started
+    oracle_check "$container" "$password"
+    started=$(date +%s)
+    # Deferred segment creation keeps an empty table to a dictionary row, so 50,000 build in minutes.
+    oracle_sqlplus "$container" "$password" <<SQL || exit 1
+WHENEVER SQLERROR EXIT FAILURE
+DECLARE
+    owner_name VARCHAR2(30);
+    present NUMBER;
+BEGIN
+    FOR s IN 1..$SCHEMAS LOOP
+        owner_name := 'LC_S' || LPAD(s, $SCHEMA_WIDTH, '0');
+        SELECT COUNT(*) INTO present FROM dba_users WHERE username = owner_name;
+        IF present = 0 THEN
+            EXECUTE IMMEDIATE 'CREATE USER ' || owner_name || ' IDENTIFIED BY probe_pw QUOTA UNLIMITED ON USERS';
+        END IF;
+        SELECT COUNT(*) INTO present FROM dba_tables WHERE owner = owner_name;
+        CONTINUE WHEN present >= $TABLES_PER_SCHEMA;
+        FOR t IN 1..$TABLES_PER_SCHEMA LOOP
+            BEGIN
+                EXECUTE IMMEDIATE 'CREATE TABLE ' || owner_name || '.LC_T' || LPAD(t, $TABLE_WIDTH, '0')
+                    || ' (id NUMBER PRIMARY KEY, name VARCHAR2(64)) SEGMENT CREATION DEFERRED';
+            EXCEPTION
+                -- ORA-00955: the table is there from an earlier run.
+                WHEN OTHERS THEN IF SQLCODE != -955 THEN RAISE; END IF;
+            END;
+        END LOOP;
+    END LOOP;
+END;
+/
+SELECT COUNT(*) || ' tables' FROM dba_tables WHERE REGEXP_LIKE(owner, '^LC_S[0-9]+$');
+EXIT
+SQL
+    echo "oracle: built in $(elapsed_since "$started")"
+}
+
+oracle_drop() {
+    local container="${1:-oracle-large-catalog}" password="${2:-probe_pw}"
+    oracle_check "$container" "$password"
+    oracle_sqlplus "$container" "$password" <<SQL
+BEGIN
+    FOR u IN (SELECT username FROM dba_users WHERE REGEXP_LIKE(username, '^LC_S[0-9]+$')) LOOP
+        EXECUTE IMMEDIATE 'DROP USER ' || u.username || ' CASCADE';
+    END LOOP;
+END;
+/
+EXIT
+SQL
+}
+
+case "$ENGINE" in
+    postgres) postgres_fixture "${@:2}" ;;
+    mysql) mysql_fixture "${@:2}" ;;
+    oracle) oracle_fixture "${@:2}" ;;
+    drop-postgres) postgres_drop "${@:2}" ;;
+    drop-mysql) mysql_drop "${@:2}" ;;
+    drop-oracle) oracle_drop "${@:2}" ;;
+    *)
+        sed -n '3,25p' "$0" | sed 's/^# \{0,1\}//'
+        exit 3
+        ;;
+esac