axiom-grdb

GRDB

概述

使用 GRDB.swift 直接访问 SQLite — 这是一个用于 SQLite 数据库的工具包，支持类型安全查询、迁移和响应式观察。

核心原则 在原始 SQL 外封装类型安全的 Swift，同时在需要时保留 SQLite 的全部能力。

要求 iOS 13+，Swift 5.7+ 许可证 MIT（免费且开源）

何时使用 GRDB

需要以下功能时使用原生 GRDB

• ✅ 跨 4 张以上表的复杂 SQL 联接
• ✅ 窗口函数（ROW_NUMBER、RANK、LAG/LEAD）
• ✅ 使用 ValueObservation 的响应式查询
• ✅ 为了性能而完全控制 SQL
• ✅ 超出架构变更范围的高级迁移逻辑

注意： SQLiteData 现在通过查询构建器支持 GROUP BY（.group(by:)）和 HAVING（.having()）— 参见 axiom-sqlitedata-ref 技能。

以下情况改用 SQLiteData

• 类型安全的 @Table 模型已满足需求
• 需要 CloudKit 同步
• 更偏好声明式查询，而不是 SQL

以下情况使用 SwiftData

• 使用 Apple 原生集成进行简单的 CRUD 操作
• 不需要原始 SQL 控制

迁移相关 请参见 axiom-database-migration 技能，了解安全的架构演进模式。

示例提示词

以下是开发者实际提出的问题，本技能旨在解答这些问题：

1. “我需要在一个查询中查询消息及其作者和反应数量。该如何编写 JOIN？”

→ 本技能展示包含多个表和聚合操作的复杂 JOIN 查询

2. “我想观察一个经过筛选的列表，并在带有特定标签的笔记发生变化时更新 UI。”

→ 本技能介绍用于响应式查询更新的 ValueObservation 模式

3. “我正在导入数千条聊天记录，需要自定义迁移逻辑。该如何使用 DatabaseMigrator？”

→ 本技能说明迁移注册、数据转换和安全回滚模式

4. “我的查询很慢（耗时 10 秒以上）。该如何分析和优化？”

→ 本技能介绍 EXPLAIN QUERY PLAN、用于分析的 database.trace 以及索引创建

5. “我需要按到期日期对任务分组并统计完成数量，再按优先级排序。相比类型安全查询，原始 SQL 似乎更简单。”

→ 本技能展示何时使用 GRDB 的原始 SQL 比类型安全封装更清晰

---

数据库设置

DatabaseQueue（单连接）

import GRDB

// File-based database
let dbPath = NSSearchPathForDirectoriesInDomains(.documentDirectory, .userDomainMask, true)[0]
let dbQueue = try DatabaseQueue(path: "\(dbPath)/db.sqlite")

// In-memory database (tests)
let dbQueue = try DatabaseQueue()

DatabasePool（连接池）

// For apps with heavy concurrent access
let dbPool = try DatabasePool(path: dbPath)

使用 Queue 的情况 大多数应用（更简单，也足够用） 使用 Pool 的情况 多个线程进行大量并发写入

Record 类型

使用 Codable

struct Track: Codable {
    var id: String
    var title: String
    var artist: String
    var duration: TimeInterval
}

// Fetch
let tracks = try dbQueue.read { db in
    try Track.fetchAll(db, sql: "SELECT * FROM tracks")
}

// Insert
try dbQueue.write { db in
    try track.insert(db)  // Codable conformance provides insert
}

FetchableRecord（只读）

struct TrackInfo: FetchableRecord {
    var title: String
    var artist: String
    var albumTitle: String

    init(row: Row) {
        title = row["title"]
        artist = row["artist"]
        albumTitle = row["album_title"]
    }
}

let results = try dbQueue.read { db in
    try TrackInfo.fetchAll(db, sql: """
        SELECT tracks.title, tracks.artist, albums.title as album_title
        FROM tracks
        JOIN albums ON tracks.albumId = albums.id
        """)
}

PersistableRecord（写入）

struct Track: Codable, PersistableRecord {
    var id: String
    var title: String

    // Customize table name
    static let databaseTableName = "tracks"
}

try dbQueue.write { db in
    var track = Track(id: "1", title: "Song")
    try track.insert(db)

    track.title = "Updated"
    try track.update(db)

    try track.delete(db)
}

原始 SQL 查询

读取数据

// Fetch all rows
let rows = try dbQueue.read { db in
    try Row.fetchAll(db, sql: "SELECT * FROM tracks WHERE genre = ?", arguments: ["Rock"])
}

// Fetch single value
let count = try dbQueue.read { db in
    try Int.fetchOne(db, sql: "SELECT COUNT(*) FROM tracks")
}

// Fetch into Codable
let tracks = try dbQueue.read { db in
    try Track.fetchAll(db, sql: "SELECT * FROM tracks ORDER BY title")
}

写入数据

try dbQueue.write { db in
    try db.execute(sql: """
        INSERT INTO tracks (id, title, artist, duration)
        VALUES (?, ?, ?, ?)
        """, arguments: ["1", "Song", "Artist", 240])
}

事务

try dbQueue.write { db in
    // Automatic transaction - all or nothing
    for track in tracks {
        try track.insert(db)
    }
    // Commits automatically on success, rolls back on error
}

类型安全查询接口

筛选

let request = Track
    .filter(Column("genre") == "Rock")
    .filter(Column("duration") > 180)

let tracks = try dbQueue.read { db in
    try request.fetchAll(db)
}

排序

let request = Track
    .order(Column("title").asc)
    .limit(10)

联接

struct TrackWithAlbum: FetchableRecord {
    var trackTitle: String
    var albumTitle: String
}

let request = Track
    .joining(required: Track.belongsTo(Album.self))
    .select(Column("title").forKey("trackTitle"), Column("album_title").forKey("albumTitle"))

let results = try dbQueue.read { db in
    try TrackWithAlbum.fetchAll(db, request)
}

复杂联接

let sql = """
    SELECT
        tracks.title as track_title,
        albums.title as album_title,
        artists.name as artist_name,
        COUNT(plays.id) as play_count
    FROM tracks
    JOIN albums ON tracks.albumId = albums.id
    JOIN artists ON albums.artistId = artists.id
    LEFT JOIN plays ON plays.trackId = tracks.id
    WHERE artists.genre = ?
    GROUP BY tracks.id
    HAVING play_count > 10
    ORDER BY play_count DESC
    LIMIT 50
    """

struct TrackStats: FetchableRecord {
    var trackTitle: String
    var albumTitle: String
    var artistName: String
    var playCount: Int

    init(row: Row) {
        trackTitle = row["track_title"]
        albumTitle = row["album_title"]
        artistName = row["artist_name"]
        playCount = row["play_count"]
    }
}

let stats = try dbQueue.read { db in
    try TrackStats.fetchAll(db, sql: sql, arguments: ["Rock"])
}

ValueObservation（响应式查询）

基本观察

import GRDB
import Combine

let observation = ValueObservation.tracking { db in
    try Track.fetchAll(db)
}

// Start observing with Combine
let cancellable = observation.publisher(in: dbQueue)
    .sink(
        receiveCompletion: { _ in },
        receiveValue: { tracks in
            print("Tracks updated: \(tracks.count)")
        }
    )

SwiftUI 集成

import GRDB
import GRDBQuery  // https://github.com/groue/GRDBQuery

@Query(Tracks())
var tracks: [Track]

struct Tracks: Queryable {
    static var defaultValue: [Track] { [] }

    func publisher(in dbQueue: DatabaseQueue) -> AnyPublisher<[Track], Error> {
        ValueObservation
            .tracking { db in try Track.fetchAll(db) }
            .publisher(in: dbQueue)
            .eraseToAnyPublisher()
    }
}

参见 GRDBQuery 文档，了解 SwiftUI 响应式绑定。

筛选观察

func observeGenre(_ genre: String) -> ValueObservation<[Track]> {
    ValueObservation.tracking { db in
        try Track
            .filter(Column("genre") == genre)
            .fetchAll(db)
    }
}

let cancellable = observeGenre("Rock")
    .publisher(in: dbQueue)
    .sink { tracks in
        print("Rock tracks: \(tracks.count)")
    }

迁移

DatabaseMigrator

var migrator = DatabaseMigrator()

// Migration 1: Create tables
migrator.registerMigration("v1") { db in
    try db.create(table: "tracks") { t in
        t.column("id", .text).primaryKey()
        t.column("title", .text).notNull()
        t.column("artist", .text).notNull()
        t.column("duration", .real).notNull()
    }
}

// Migration 2: Add column
migrator.registerMigration("v2_add_genre") { db in
    try db.alter(table: "tracks") { t in
        t.add(column: "genre", .text)
    }
}

// Migration 3: Add index
migrator.registerMigration("v3_add_indexes") { db in
    try db.create(index: "idx_genre", on: "tracks", columns: ["genre"])
}

// Run migrations
try migrator.migrate(dbQueue)

迁移安全模式 请参见 axiom-database-migration 技能。

带数据转换的迁移

migrator.registerMigration("v4_normalize_artists") { db in
    // 1. Create new table
    try db.create(table: "artists") { t in
        t.column("id", .text).primaryKey()
        t.column("name", .text).notNull()
    }

    // 2. Extract unique artists
    try db.execute(sql: """
        INSERT INTO artists (id, name)
        SELECT DISTINCT
            lower(replace(artist, ' ', '_')) as id,
            artist as name
        FROM tracks
        """)

    // 3. Add foreign key to tracks
    try db.alter(table: "tracks") { t in
        t.add(column: "artistId", .text)
            .references("artists", onDelete: .cascade)
    }

    // 4. Populate foreign keys
    try db.execute(sql: """
        UPDATE tracks
        SET artistId = (
            SELECT id FROM artists
            WHERE artists.name = tracks.artist
        )
        """)
}

性能模式

批量写入

try dbQueue.write { db in
    for batch in tracks.chunked(into: 500) {
        for track in batch {
            try track.insert(db)
        }
    }
}

预处理语句

try dbQueue.write { db in
    let statement = try db.makeStatement(sql: """
        INSERT INTO tracks (id, title, artist, duration)
        VALUES (?, ?, ?, ?)
        """)

    for track in tracks {
        try statement.execute(arguments: [track.id, track.title, track.artist, track.duration])
    }
}

索引

try db.create(index: "idx_tracks_artist", on: "tracks", columns: ["artist"])
try db.create(index: "idx_tracks_genre_duration", on: "tracks", columns: ["genre", "duration"])

// Unique index
try db.create(index: "idx_tracks_unique_title", on: "tracks", columns: ["title"], unique: true)

查询规划

// Analyze query performance
let explanation = try dbQueue.read { db in
    try String.fetchOne(db, sql: "EXPLAIN QUERY PLAN SELECT * FROM tracks WHERE artist = ?", arguments: ["Artist"])
}
print(explanation)

从 SQLiteData 切换到底层 GRDB

使用 SQLiteData 时，如果特定操作需要 GRDB：

import SQLiteData
import GRDB

@Dependency(\.database) var database  // SQLiteData Database

// Access underlying GRDB DatabaseQueue
try await database.database.write { db in
    // Full GRDB power here
    try db.execute(sql: "CREATE INDEX idx_genre ON tracks(genre)")
}

常见场景

• 复杂 JOIN 查询
• 自定义迁移
• 批量 SQL 操作
• ValueObservation 设置

快速参考

常见操作

// Read single value
let count = try db.fetchOne(Int.self, sql: "SELECT COUNT(*) FROM tracks")

// Read all rows
let rows = try Row.fetchAll(db, sql: "SELECT * FROM tracks WHERE genre = ?", arguments: ["Rock"])

// Write
try db.execute(sql: "INSERT INTO tracks VALUES (?, ?, ?)", arguments: [id, title, artist])

// Transaction
try dbQueue.write { db in
    // All or nothing
}

// Observe changes
ValueObservation.tracking { db in
    try Track.fetchAll(db)
}.publisher(in: dbQueue)

资源

GitHub：groue/GRDB.swift、groue/GRDBQuery

文档：sqlite.org/docs.html

技能：axiom-database-migration、axiom-sqlitedata、axiom-swiftdata

生产环境性能：高压情况下的查询优化

危险信号 — GRDB 查询变慢时

如果出现以下任一症状：

• ❌ 复杂 JOIN 查询耗时 10 秒以上
• ❌ ValueObservation 在每次变更时都会运行（耗电）
• ❌ 无法解释为什么旧版本上迁移运行了两次

不要

1. 盲目添加索引（不知道哪些列有帮助）
2. 将逻辑移到 Swift 中（过早逃离数据库）
3. 过度设计迁移（不信任系统）

应该

1. 使用 database.trace 进行性能分析
2. 使用 EXPLAIN QUERY PLAN 了解执行过程
3. 信任 GRDB 的迁移版本控制系统

复杂查询的性能分析

查询缓慢时（10 秒以上）

var database = try DatabaseQueue(path: dbPath)

// Enable tracing to see SQL execution
database.trace { print($0) }

// Run the slow query
try database.read { db in
    let results = try Track.fetchAll(db)  // Watch output for execution time
}

// Use EXPLAIN QUERY PLAN to understand execution:
try database.read { db in
    let plan = try String(fetching: db, sql: "EXPLAIN QUERY PLAN SELECT ...")
    print(plan)
    // Look for SCAN (slow, full table) vs SEARCH (fast, indexed)
}

有策略地添加索引

// Add index on frequently queried column
try database.write { db in
    try db.execute(sql: "CREATE INDEX idx_plays_track_id ON plays(track_id)")
}

时间成本

• 性能分析：10 分钟（启用 trace、运行查询、阅读输出）
• 理解：5 分钟（解读 EXPLAIN QUERY PLAN）
• 修复：5 分钟（添加索引）
• 总计：20 分钟（而不是花 30 分钟以上盲目尝试解决方案）

ValueObservation 性能

使用响应式查询时，了解其成本

// Re-evaluates query on ANY write to database
ValueObservation.tracking { db in
    try Track.fetchAll(db)
}.start(in: database, onError: { }, onChange: { tracks in
    // Called for every change — CPU spike!
})

优化模式

// Coalesce rapid updates (recommended)
ValueObservation.tracking { db in
    try Track.fetchAll(db)
}.removeDuplicates()  // Skip duplicate results
 .debounce(for: 0.5, scheduler: DispatchQueue.main)  // Batch updates
 .start(in: database, ...)

决策框架

• 小型数据集（少于 1000 条记录）：使用普通 .tracking
• 中型数据集（1-10k 条记录）：添加 .removeDuplicates() + .debounce()
• 大型数据集（10k 条以上记录）：使用显式表依赖项或谓词

迁移版本控制保证

信任 GRDB 的 DatabaseMigrator——它会避免重复运行迁移

var migrator = DatabaseMigrator()

migrator.registerMigration("v1_initial") { db in
    try db.execute(sql: "CREATE TABLE tracks (...)")
}

migrator.registerMigration("v2_add_plays") { db in
    try db.execute(sql: "CREATE TABLE plays (...)")
}

// GRDB guarantees:
// - Each migration runs exactly ONCE
// - In order (v1, then v2)
// - Safe to call migrate() multiple times
try migrator.migrate(dbQueue)

不需要防御性 SQL（IF NOT EXISTS）

• GRDB 会跟踪已运行的迁移
• 再次运行 migrate() 只会执行新的迁移
• 过度设计只会增加复杂性，没有任何好处

放心信任它。

---

常见错误

❌ 批量写入时不使用事务

for track in 50000Tracks {
    try dbQueue.write { db in try track.insert(db) }  // 50k transactions!
}

修复方法： 使用单个事务分批处理

❌ 在主线程上同步访问数据库

let tracks = try dbQueue.read { db in try Track.fetchAll(db) }  // Blocks UI

修复方法： 使用 async/await 或分派到后台队列

❌ 忘记添加索引

// Slow query without index
try Track.filter(Column("genre") == "Rock").fetchAll(db)

修复方法： 在频繁查询的列上创建索引

❌ N+1 查询

for track in tracks {
    let album = try Album.fetchOne(db, key: track.albumId)  // N queries!
}

修复方法： 使用 JOIN 或批量获取

tvOS

本地 GRDB 数据库在 tvOS 上不会持久保存。 存储空间不足时，系统会删除缓存（包括 Application Support）。仅存储在本地的 GRDB 数据库会在每次启动应用时丢失所有数据。

如果目标平台是 tvOS， 请将 GRDB 与 CloudKit 同步搭配使用（通过 CKSyncEngine 或 SQLiteData 的 SyncEngine），让 iCloud 作为持久化存储，并将本地数据库用作可重建的缓存。有关 tvOS 存储限制的完整说明，请参阅 axiom-tvos。

---

目标平台： iOS 13+，Swift 5.7+ 框架： GRDB.swift 6.0+ 历史记录： 查看 git log 了解变更