laantungir/amethyst
1
0
Fork 0
mirror of https://github.com/vitorpamplona/amethyst.git synced 2026-07-22 07:48:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #3663 from vitorpamplona/claude/sqlite-query-performance-bh2j27

Browse Source 
SQLite query scaling: FTS contentless, tag-merge, pooled statements
This commit is contained in:
Vitor Pamplona
2026-07-21 19:38:22 -04:00
committed by GitHub
parent 35e61e1d98 d571bee394
commit 72b8df7f6d
14 changed files with 1781 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

168
quartz/plans/2026-07-21-sqlite-query-scaling.md Normal file
View File

@@ -0,0 +1,168 @@
# SQLite query scaling: contentless FTS + rowid search, tag-path merge, pooled statements
**Status: shipped.** Follow-up to a scale-curve report (`SQLite vs Vespa`,
throughput vs corpus size) showing the app-side SQLite store degrading with
corpus size on several query shapes — NIP-50 search ~18× (25k→400k), batch
ingest 16×, author-timeline 13×, follow-feed 2.4× — while point reads stayed
flat. Two of those (author-timeline, follow-feed) were already addressed
(the pubkey-alone index and `MergeQueryExecutor`) and mostly reflect running
the report under the *client* `DefaultIndexingStrategy` rather than
`relayIndexingStrategy()`. This change adds the **large-IN tag watcher** to the
merge executor, fixes the **FTS delete path** (which degraded with corpus
size) and shrinks the FTS index, fixes **NIP-50 search ordering** (it was
sorting by `created_at`, not relevance), and fixes a **statement-cache** miss
the merge paths hit.
It does **not** fix NIP-50 *search* latency: bm25 relevance scoring (like the
old `created_at` sort) must visit every match, so search cost still grows with
the match set (§1). Corpus-independent search is an external-engine job.
Everything here is read/size work; the write path and on-disk index set are
unchanged except the FTS table, which gets *smaller* and deletes faster.
## 1. FTS — contentless index (fast deletes + smaller), NOT a search-scaling fix
The old index was `fts5(event_header_row_id, content)`: it stored a second
copy of the tokenized text, its rowid was auto-assigned (unrelated to the
event), and — the real problem — the `fts_foreign_key` delete trigger deleted
by the `event_header_row_id` *column*, which FTS5 cannot seek, so it **scanned
the whole index per delete**.
Changes (`FullTextSearchModule`, `QueryBuilder`, DB version 4→5):
- **rowid = `event_headers.row_id`.** Set explicitly on every insert; the
delete trigger, reindex, and catch-up paths key off it. Deletes become an
O(log n) primary-key seek instead of an O(n) column scan. This is the win:
every event removal fires the trigger — replaceable rotation, kind-5,
expiration, right-to-vanish — so on the old schema deletion throughput
degraded with corpus size.
- **Contentless** (`fts5(content, content='', contentless_delete=1)`). The
indexed text is *derived* (`SearchableEvent.indexableContent()`, not a raw
column), so FTS5 **external-content** — which re-reads the source column from
the base table — cannot express it; **contentless** is the correct primitive.
It drops the stored content copy (index shrinks) and `contentless_delete=1`
keeps the delete trigger working.
- **Segment compaction.** `reindexAll` finishes with `'optimize'`, and the
periodic `SQLiteEventStore.optimize()` (geode's maintenance tick) folds in a
bounded `'merge'`, so incremental / deferred-catch-up inserts don't leave the
index as many small segments.
Delete cost — `FtsSearchScalingBenchmark.deleteByColumnVsByRowid`, 500 deletes,
in-memory:
| rows | by column (old) | by rowid (new) |
|---|---:|---:|
| 2k | 91.8 ms | 6.6 ms |
| 8k | 362.0 ms | 4.7 ms |
By-column grows ~linearly with the table (O(n)/delete); by-rowid is flat —
~78× at 8k rows and widening.
**Search ordering fixed to relevance (NIP-50), which the store was getting
wrong.** NIP-50 says results are returned "in descending order by quality of
search result ... not by the usual `.created_at`", with the limit applied after
the score — but the store sorted search by `created_at DESC` (pre-existing).
*Every* search filter now orders by FTS5 bm25 (`ORDER BY event_fts.rank`,
`created_at DESC` as a tie-break): the tag-free shape (`makeSimpleSearch`) and
`search + tag` (the `prepareRowIDSubQueries`/`makeQueryIn` path, which carries
the rank column through the row-id subquery via `projectRank` and applies the
LIMIT by rank). Only the negentropy snapshot keeps `created_at` — a sync set,
not a ranked result. Verified against stronger-but-older matches outranking
weaker-but-newer ones, tag-scoped, with the limit cutting by score
(`SearchRelevanceOrderTest`, `Fts5CapabilityProbe`).
An earlier draft of *this* change instead added a `searchOrderByRowId` flag
(order by the FTS rowid for O(limit) search); that is *ingestion* order — wrong
events under a limit once ingestion diverges from time order, and not relevance
either — so it was removed.
This is a **correctness** fix, not a scaling one: bm25 (like the created_at
sort) must score every match, so search latency still grows with the match set
(the report's 18× curve) and `optimize` doesn't change the asymptotics
(measured within noise at 100k/200k). Corpus-independent search is genuinely an
external-engine job (the Vespa side of the report), not this index.
Migration (v4→v5): the old rowids can't be remapped, so `event_fts` is dropped
and rebuilt — synchronous stores rebuild in the upgrade transaction (client
corpora are small), deferred stores reset the catch-up watermark to 0 and let
the background worker repopulate (no long migration transaction).
`ContentlessFtsMigrationTest` fabricates a real v4 DB (old schema + garbage
rows + `user_version=4`) and asserts the reopen rebuilds search, wipes the
stale rows, maps rowid→row_id, and keeps the delete trigger working.
## 2. Tag watcher — extend `MergeQueryExecutor` to the tag path
`kinds=[7] AND #e=[hundreds of note ids] LIMIT n` (reactions/replies) had the
same shape the follow-feed fix already solved for authors: the per-value
streams come sorted off `(tag_hash[, kind], created_at)`, but their union does
not, so SQLite collected every matching row and TEMP-B-TREE-sorted to the limit
— O(matching history), growing with the corpus (`TagAuthorIndexBenchmark`:
12.8 ms cold at 200k → 14.2 ms at 1M).
`MergeQueryExecutor` now opens one lazy newest-first cursor per `(value[,
kind])` off `query_by_tags_hash_kind` (or `query_by_tags_hash` when there is no
kind, gated by `indexTagsByCreatedAtAlone`) and heap-merges to the limit —
**O(limit + streams)**. Unlike the author path, one event can carry several
queried tag values, so the tag merge **dedups by event id** through a `seen`
set (the author path, one pubkey per event, skips it). Eligibility is narrow
(one non-`d` tag key, ≥2 values, a limit, no ids/authors/d-tag/search, no
AND-tags); everything else falls through to the single-SQL plan. Counts and
deletes are unchanged (still single-SQL). `TagMergeCorrectnessTest` pins it
against a Kotlin reference including cross-stream dedup, since/until windows,
the raw path, and tie handling (tag cursors order off `event_tags`, which has
no id column, so same-second ties are a valid newest-N but not id-exact).
## 3. Pooled statement cache — so the merges actually cache
`StatementCachingConnection` kept **one** handle per SQL string. The k-way
merge opens *many* identical-SQL cursors at once (one per author/tag stream),
so every stream past the first missed the cache and prepared uncached, and a
repeated follow-feed / reactions REQ re-prepared all of them each poll. The
cache now keeps a small **pool** per SQL (bounded by a global cap, default
raised 256→512), so concurrent same-SQL checkouts reuse cached handles and
repeated polls reuse their per-stream cursors. `StatementCachingConnectionTest`
covers sequential reuse, concurrent distinctness/independence, freed-handle
reuse, and cap overflow → uncached fallback.
## Correctness / scope
- DB version bump 4→5 with a real-upgrade test.
- New capability test (`Fts5CapabilityProbe`) pins the FTS5 features the
contentless index needs (contentless_delete, absent-rowid delete no-op,
rowid ordering, merge/optimize) to the bundled SQLite (3.50.1), so a driver
downgrade fails loudly.
- All existing `store.sqlite` tests pass unchanged except `QueryAssemblerTest`,
whose asserted EXPLAIN output updated for the new join column
(`event_fts.rowid`) and the contentless table's virtual-index marker
(`0:M2`→`0:M1`).
- Search ordering: bm25 relevance for every search REQ shape — tag-free,
`search + tag`, and multi-filter all-search (e.g. the client's
search-across-kinds, unioned then deduped by event keeping the best score).
Mixed search/non-search multi-filter REQs stay `created_at` (a non-search
branch has no defined relevance).
## Audit follow-ups (post-review hardening)
A two-reviewer adversarial audit of the branch found no correctness/data-loss/
crash bugs; it produced these robustness fixes and one gap-closure:
- **Multi-filter search ordering** (gap): a REQ of several filters that all
carry a search term was `created_at`-ordered (the union path). Now
relevance-ordered via `unionSubqueriesIfNeeded(projectRank)` — each branch
projects `rank`, `UNION ALL` + `GROUP BY row_id MIN(rank)` dedups across
branches keeping the best score. Only when every branch is a search branch
(and FTS is on); count/delete unions stay single-column.
- **Merge stream-prep leak** (F1): `prepareAuthorStreams`/`prepareTagStreams`
now build cursors through `buildStreams`, which closes any already-prepared
statements if a later prepare throws — otherwise a mid-loop failure stranded
checked-out, un-reset handles (read locks) in the pooled connection.
- **Stream-count overflow** (F3): `authors×kinds` / `values×kinds` computed as
`Long` so a pathological product can't wrap `Int` back into the eligible
band and route a huge fan-out into the merge.
- **Finalizer footgun** (F2): the pooled statement's `finalize()` was renamed
`finalizeStatement()` — a no-arg `finalize()` is the JVM's `Object.finalize`,
so the GC could double-close the native handle after explicit close.
Migration cost noted (low/operational, not correctness): the synchronous
v4→v5 rebuild runs in the upgrade transaction; atomic and kill-safe (rolls
back to v4), but a very large client store pays a one-time first-open stall.

165
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/FullTextSearchModule.kt
View File

@@ -31,6 +31,37 @@ import com.vitorpamplona.quartz.utils.EventFactory
/**
* NIP-50 full-text search index over event content.
*
* The index is a **contentless** FTS5 table (`content=''`,
* `contentless_delete=1`) whose `rowid` is the event's
* `event_headers.row_id`. Two consequences:
*
* - **Deletes.** The `fts_foreign_key` trigger deletes by `rowid` (an FTS5
* primary-key seek, O(log n)). The old `fts5(event_header_row_id,
* content)` schema deleted by a *regular* column, which FTS5 cannot seek —
* it scans the whole index per delete (O(n)), so deletion throughput
* degraded with corpus size. Every event removal fires this trigger
* (replaceable rotation, kind-5, expiration, right-to-vanish), so the
* seek matters. Measured `Fts5CapabilityProbe`/`FtsSearchScalingBenchmark`:
* ~78× at 8k rows and widening with the table.
* - **Size.** A contentless table stores only the inverted index, not a
* second copy of the tokenized text. The indexed text is *derived*
* ([SearchableEvent.indexableContent], not any raw column), so FTS5
* external-content — which reads the source column from the base table —
* cannot express it; contentless is the correct primitive.
*
* Search results are ordered by **relevance**, per NIP-50 ("descending order
* by quality of search result ... not by the usual `.created_at`", limit
* applied after the score) — via FTS5 bm25 (`ORDER BY event_fts.rank`), with
* `created_at DESC` only as a tie-break. This holds for *every* search filter:
* the tag-free shape ([QueryBuilder.makeSimpleSearch]), `search + tag` (whose
* row-id subquery carries the rank through via `projectRank`), and a
* multi-filter all-search REQ (unioned, deduped by event keeping the best
* score). Only the negentropy snapshot — and a multi-filter REQ mixing search
* and non-search branches — keep `created_at` (a sync set / a branch with no
* defined relevance). bm25 must score every match, so search latency still grows with the
* match set regardless of ordering; corpus-independent search needs an
* external engine, not this index.
*
* When [enabled] is `false` the module becomes an inert no-op: no
* `event_fts` virtual table and no `fts_foreign_key` delete trigger are
* created, inserts skip the per-event tokenization cost, and the reindex
@@ -52,29 +83,53 @@ class FullTextSearchModule(
) : IModule {
val tableName = "event_fts"
val triggerName = "fts_foreign_key"
val eventHeaderRowIdName = "event_header_row_id"
/**
* The FTS column that links back to `event_headers`. It is the implicit
* `rowid` of the (contentless) FTS table, which we set equal to
* `event_headers.row_id` on every insert — so the join is
* `event_headers.row_id = event_fts.rowid`, with no stored column.
*/
val rowIdColumn = "rowid"
val contentName = "content"
val stateTableName = "fts_catchup_state"
/**
* Whether the on-disk `event_fts` is an FTS5 table (vs the fts4/fts3
* fallback). Only FTS5 supports the contentless schema and the
* `merge`/`optimize` maintenance commands; the fallback stores content
* and skips maintenance. Read only on the (single-threaded) writer.
* Cached lazily from `sqlite_master` so a reopened DB — where [create]
* never runs — still resolves it.
*/
private var isFts5: Boolean? = null
override fun create(db: SQLiteConnection) {
if (!enabled) return
val ftsVersion = versionFinder(db)
db.execSQL(
"""
CREATE VIRTUAL TABLE $tableName
USING fts$ftsVersion($eventHeaderRowIdName, $contentName)
""",
)
isFts5 = ftsVersion >= 5
// FTS5: contentless index (no stored content copy) with delete
// support. fts4/fts3 (bundled driver never selects them) fall back to
// a plain content-storing table — rowid-explicit insert and
// delete-by-rowid work there too, only the size win is FTS5-only.
val columns =
if (ftsVersion >= 5) {
"$contentName, content='', contentless_delete=1"
} else {
contentName
}
db.execSQL("CREATE VIRTUAL TABLE $tableName USING fts$ftsVersion($columns)")
// Foreign key cleanup for full text search
// Foreign key cleanup for full text search. Deletes by the FTS rowid
// (= event_headers.row_id); a header with no FTS row (non-searchable
// kind) deletes nothing, which is a harmless no-op.
db.execSQL(
"""
CREATE TRIGGER $triggerName
AFTER DELETE ON event_headers
FOR EACH ROW
BEGIN
DELETE FROM $tableName
WHERE old.row_id = $tableName.$eventHeaderRowIdName;
DELETE FROM $tableName WHERE $tableName.rowid = old.row_id;
END;
""",
)
@@ -82,6 +137,16 @@ class FullTextSearchModule(
createStateTable(db)
}
private fun resolveIsFts5(db: SQLiteConnection): Boolean {
isFts5?.let { return it }
val sql =
db.prepare("SELECT sql FROM sqlite_master WHERE type = 'table' AND name = ?").use { stmt ->
stmt.bindText(1, tableName)
if (stmt.step()) stmt.getText(0) else ""
}
return sql.contains("fts5", ignoreCase = true).also { isFts5 = it }
}
/**
* Watermark for the deferred path: everything with
* `row_id <= last_row_id` is guaranteed indexed. Idempotent — also
@@ -124,13 +189,13 @@ class FullTextSearchModule(
val insertFTS =
"""
INSERT OR ROLLBACK INTO $tableName ($eventHeaderRowIdName, $contentName)
INSERT OR ROLLBACK INTO $tableName (rowid, $contentName)
VALUES (?, ?)
""".trimIndent()
val deleteFTSByRowId =
"""
DELETE FROM $tableName WHERE $eventHeaderRowIdName = ?
DELETE FROM $tableName WHERE rowid = ?
""".trimIndent()
fun insert(
@@ -200,7 +265,19 @@ class FullTextSearchModule(
dropTrigger(db)
drop(db)
create(db)
populateAll(db)
// The rebuild just wrote every row as its own tiny segment; compact
// them into one so the first search after a reindex isn't a scan over
// hundreds of segments.
optimize(db)
}
/**
* Scan every stored searchable event and insert its derived content into
* the (already created, empty) FTS index, keyed by `event_headers.row_id`.
* The caller owns the transaction and the create/drop lifecycle.
*/
private fun populateAll(db: SQLiteConnection) {
val kinds = searchableKindsPresent(db)
if (kinds.isEmpty()) return
@@ -230,6 +307,58 @@ class FullTextSearchModule(
}
}
/**
* v4 → v5 migration: the pre-v5 index was `fts5(event_header_row_id,
* content)` with an auto-assigned rowid unrelated to `event_headers`, and
* it stored a second copy of the content. v5 is the contentless,
* rowid = row_id schema. The old rowids can't be remapped in place, so the
* table is dropped and repopulated. Runs inside the migration transaction.
*
* - **synchronous** stores rebuild now (client corpora are small).
* - **deferred** stores reset the catch-up watermark to 0 so the relay's
* background worker repopulates without a long migration transaction.
*/
fun migrateV4ToContentless(db: SQLiteConnection) {
if (!enabled) return
dropTrigger(db)
drop(db)
create(db)
if (deferIndexing) {
db.prepare("UPDATE $stateTableName SET last_row_id = 0 WHERE id = 1").use { it.step() }
} else {
populateAll(db)
optimize(db)
}
}
/**
* Full FTS5 segment compaction — merges the b-tree segments left by
* incremental inserts into one, so a `MATCH` touches a single segment
* instead of dozens. Expensive (rewrites the whole index); call it once
* after a rebuild, not per batch. No-op on the fts4/fts3 fallback.
*/
fun optimize(db: SQLiteConnection) {
if (!enabled || !resolveIsFts5(db)) return
db.prepare("INSERT INTO $tableName($tableName) VALUES ('optimize')").use { it.step() }
}
/**
* Bounded incremental segment merge — does at most [pages] pages of merge
* work, so it stays cheap enough to run on a periodic maintenance tick
* while the index keeps growing from deferred catch-up. No-op on the
* fts4/fts3 fallback.
*/
fun mergeSegments(
db: SQLiteConnection,
pages: Int = 16,
) {
if (!enabled || !resolveIsFts5(db)) return
db.prepare("INSERT INTO $tableName($tableName, rank) VALUES ('merge', ?)").use { stmt ->
stmt.bindLong(1, pages.toLong())
stmt.step()
}
}
/**
* Process one batch of a resumable rebuild: re-derive the FTS rows
* for up to [batchSize] events whose `row_id > ` [afterRowId] and
@@ -329,13 +458,11 @@ class FullTextSearchModule(
// Unlike [reindexBatch] there is NO per-row delete here: rows past
// the watermark were never indexed (deferred mode skips insert()),
// and the watermark advances atomically with the FTS rows it
// covers, so a crash replay is impossible. The delete would also
// be ruinous — `event_header_row_id` is a plain FTS5 column, so
// deleting by it scans the whole FTS table per row, which turned
// the first catch-up implementation O(n²). Consequence: switching
// a database back and forth between deferred and synchronous
// strategies requires a [reindexAll] in between (same rule as a
// searchable-kinds change).
// covers, so a crash replay is impossible. (Delete-by-rowid is now
// O(log n) on the contentless index, so this is purely about not
// doing redundant work.) Consequence: switching a database back and
// forth between deferred and synchronous strategies requires a
// [reindexAll] in between (same rule as a searchable-kinds change).
val limit = batchSize.coerceAtLeast(1)
val kinds = searchableKindsPresent(db)
var last = watermark

314
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt
View File

@@ -24,63 +24,94 @@ import androidx.sqlite.SQLiteConnection
import androidx.sqlite.SQLiteStatement
/**
* k-way merge executor for the **home-feed** query shape:
* `authors=[…] (+ kinds=[…]) [+ since/until] limit=N` ordered newest-first.
* k-way merge executor for the two "wide fan-out, newest-N" query shapes
* whose single-SQL plan reads O(matching history) rather than O(limit):
*
* SQLite serves this by seeking every `(kind, pubkey)` combo and feeding
* *all* matching rows through a LIMIT-bounded sorter — so it reads O(the
* followed set's whole matching history). For prolific follows on a cold
* on-disk DB that's the `follow-feed` regression (relayBench: 97 ms vs
* strfry 17 ms). See `quartz/plans/2026-07-04-follow-feed-read-tradeoff.md`.
* 1. **home-feed** — `authors=[…] (+ kinds=[…]) [+ since/until] limit=N`.
* SQLite seeks every `(kind, pubkey)` combo and feeds *all* matching rows
* through a LIMIT-bounded sorter — O(the followed set's whole matching
* history). For prolific follows on a cold on-disk DB that was the
* `follow-feed` regression (relayBench: 97 ms vs strfry 17 ms). See
* `quartz/plans/2026-07-04-follow-feed-read-tradeoff.md`.
* 2. **tag watcher** — `#<x>=[hundreds of values] (+ kinds=[…])
* [+ since/until] limit=N`, the reactions/replies archetype
* (`kinds=[7] AND #e=[note ids]`). The per-value streams come sorted off
* `(tag_hash[, kind], created_at)`, but their union does not, so SQLite
* collects every matching row and TEMP-B-TREE sorts to the limit — the
* tag-index analogue of the follow-feed shape. Measured by
* `TagAuthorIndexBenchmark` (jvmTest prodbench): `#e IN 300, limit 500`
* cost 12.8 ms cold at 200k events and 14.2 ms at 1M, growing with
* matching history.
*
* Each `(kind, pubkey)` is already a newest-first stream off the
* `query_by_kind_pubkey_created (kind, pubkey, created_at DESC)` index
* (or `query_by_pubkey_created` for authors-only). This opens one lazy
* cursor per stream and merges their heads, stopping at the limit — so it
* reads only **O(limit + streams)** rows regardless of how much history the
* authors have, and it reuses the existing indexes (no write/size cost).
* Each stream is already a newest-first cursor off an existing index:
* - authors: `query_by_kind_pubkey_created (kind, pubkey, created_at DESC)`
* (or `query_by_pubkey_created` for authors-only);
* - tags: `query_by_tags_hash_kind (tag_hash, kind, created_at DESC)`
* (or `query_by_tags_hash` for the no-kind case, gated by
* [IndexingStrategy.indexTagsByCreatedAtAlone]).
*
* The merge opens one lazy cursor per stream, merges their heads newest-first,
* and stops at the limit — reading **O(limit + streams)** rows regardless of
* how much history the authors/tags have, and reusing the existing indexes
* (no write/size cost). With the pooled statement cache
* ([StatementCachingConnection]) the per-stream cursors are prepared once and
* reused across repeated polls of the same REQ.
*
* Merge order is `created_at DESC`, tie-broken by `id ASC`. NIP-01 leaves
* same-`created_at` ties unspecified, so the returned set is a valid
* newest-N either way. The `id ASC` tie-break is exact — byte-for-byte the
* same events the single-SQL path returns — only when the store indexes id
* ([IndexingStrategy.useAndIndexIdOnOrderBy]): then each per-stream cursor
* streams in `(created_at DESC, id ASC)` straight off the index, so a
* stream's same-second head really is its id-minimum. Without that index
* the per-stream cursor yields same-second rows in rowid order, so the
* result is still a valid newest-N but may differ from the single-SQL path
* exactly at a same-second boundary.
* same-`created_at` ties unspecified, so the returned set is a valid newest-N
* either way. The `id ASC` tie-break is exact — byte-for-byte the same events
* the single-SQL path returns — only when the store indexes id
* ([IndexingStrategy.useAndIndexIdOnOrderBy]) **and** the stream cursor can
* order by id off the index. The author streams can (id is on
* `event_headers`); the tag streams cannot (the cursor orders off
* `event_tags`, which has no id column), so a tag stream yields same-second
* rows in rowid order — still a valid newest-N, but same-second ties may
* differ from an id-ordered reference.
*
* **Cross-stream duplicates.** An author appears in exactly one author stream
* (one pubkey per event), so the home-feed merge never double-counts. A single
* event can carry several of the queried tag values (or a repeated tag), so it
* can surface in several tag streams — the single-SQL path dedups with
* `SELECT DISTINCT`. The tag merge therefore dedups by event id through a
* `seen` set; the author merge skips that set entirely.
*/
internal object MergeQueryExecutor {
// TODO: the same collect-all + TEMP-B-TREE-sort pattern exists one index
// over, on the tag path: `kinds + tags(#e IN [hundreds]) + limit` (the
// reactions/replies watcher archetype) unions per-value streams that are
// each sorted off `(tag_hash, kind, created_at)` and sorts the union.
// `streamCount` currently rejects any filter with tags, so those queries
// never merge. Measured by `TagAuthorIndexBenchmark`: `#e IN 300,
// limit 500` costs 12.8 ms cold at 200k events and 14.2 ms at 1M
// (6.7 ms with indexTagsWithKindAndPubkey on) — tolerable, but it
// scales with matching history like the follow-feed shape did; extend
// the merge to per-tag-value streams if the relayBench
// `reactions-watch` scenario shows it in the profile vs strfry.
/** Author-stream projection: a single `event_headers` scan, unqualified. */
const val COLS = "id, pubkey, created_at, kind, tags, content, sig"
/** Tag-stream projection: `event_tags` joins `event_headers`, so qualify. */
private const val EH_COLS =
"event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig"
/**
* Above this many streams, fall back to the single-SQL plan: the
* per-stream cursor setup stops paying off, and huge author lists are
* collecting a lot no matter what. `kinds.size × authors.size`.
* per-stream cursor setup stops paying off, and huge fan-outs are
* collecting a lot no matter what. `kinds.size × (authors|values).size`.
*/
const val MAX_STREAMS = 2048
/**
* Stream count if [filter] is merge-eligible, else `-1`. Eligible = a
* simple (no tag/search/id/d-tag) query with authors + a limit, whose
* per-stream index exists. `kinds` optional: with it, one stream per
* `(kind, author)`; without, one per author (needs the pubkey index).
* Stream count if [filter] is merge-eligible under *either* shape, else
* `-1`. Routing check for [QueryBuilder]; [run] re-derives which shape.
*/
fun streamCount(
filter: QueryBuilder.FilterWithDTags,
indexStrategy: IndexingStrategy,
): Int {
val authorStreams = authorStreamCount(filter, indexStrategy)
if (authorStreams > 0) return authorStreams
return tagStreamCount(filter, indexStrategy)
}
/**
* Author-shape stream count, or `-1`. Eligible = a simple (no tag/search/
* id/d-tag) query with authors + a limit, whose per-stream index exists.
* `kinds` optional: with it, one stream per `(kind, author)`; without, one
* per author (needs the pubkey index).
*/
fun authorStreamCount(
filter: QueryBuilder.FilterWithDTags,
indexStrategy: IndexingStrategy,
): Int {
if (!filter.isSimpleQuery()) return -1
if (filter.ids != null) return -1
@@ -94,27 +125,69 @@ internal object MergeQueryExecutor {
// distinct set.
val distinctAuthors = authors.distinct().size
val kinds = filter.kinds
// Long product: a pathological authors×kinds could overflow Int and
// wrap back into the eligible band, routing a huge fan-out here.
val streams =
if (kinds != null && kinds.isNotEmpty()) {
distinctAuthors * kinds.distinct().size
distinctAuthors.toLong() * kinds.distinct().size
} else {
// authors-only needs the (pubkey, created_at) index to stream.
if (!indexStrategy.indexEventsByPubkeyAlone) return -1
distinctAuthors
distinctAuthors.toLong()
}
// A single stream is already the optimal single index seek — let the
// normal path handle it; only merge when there's something to merge.
return if (streams in 2..MAX_STREAMS) streams else -1
return if (streams in 2..MAX_STREAMS.toLong()) streams.toInt() else -1
}
/** Prepares one bound, newest-first cursor per stream. */
private fun prepareStreams(
/**
* Tag-shape stream count, or `-1`. Eligible = a single non-`d` tag key
* with `IN` (any-of) semantics and ≥2 distinct values, plus a limit, no
* ids/authors/d-tag/search, and no `AND`-tags (`tagsAll`) — the large-IN
* watcher shape. `kinds` optional: with it, one stream per
* `(value, kind)` off `query_by_tags_hash_kind`; without, one per value
* off `query_by_tags_hash` (needs [IndexingStrategy.indexTagsByCreatedAtAlone]).
*
* Authors are excluded on purpose: `tag ∩ author ∩ kind` is a covered
* single seek under [IndexingStrategy.indexTagsWithKindAndPubkey], not a
* fan-out, and mixing an author predicate into per-tag streams would not
* reduce the read.
*/
fun tagStreamCount(
filter: QueryBuilder.FilterWithDTags,
indexStrategy: IndexingStrategy,
): Int {
if (filter.ids != null) return -1
if (filter.authors != null) return -1
if (filter.dTags != null) return -1
if (filter.search != null && filter.search.isNotEmpty()) return -1
if (filter.limit == null || filter.limit <= 0) return -1
// AND-tags can't be expressed as a union of per-value streams.
if (filter.nonDTagsAll != null && filter.nonDTagsAll.isNotEmpty()) return -1
val inTags = filter.nonDTagsIn ?: return -1
// A second tag key would AND across keys — not a single union.
if (inTags.size != 1) return -1
val values = inTags.values.first().distinct()
if (values.size < 2) return -1
val kinds = filter.kinds?.distinct()?.takeIf { it.isNotEmpty() }
val streams =
if (kinds != null) {
values.size.toLong() * kinds.size
} else {
if (!indexStrategy.indexTagsByCreatedAtAlone) return -1
values.size.toLong()
}
return if (streams in 2..MAX_STREAMS.toLong()) streams.toInt() else -1
}
/** Prepares one bound, newest-first cursor per author stream. */
private fun prepareAuthorStreams(
db: SQLiteConnection,
filter: QueryBuilder.FilterWithDTags,
indexStrategy: IndexingStrategy,
): List<SQLiteStatement> {
// Dedup so a repeated pubkey/kind can't open two identical cursors and
// double-emit (see streamCount).
// double-emit (see authorStreamCount).
val authors = filter.authors!!.distinct()
val kinds = filter.kinds?.distinct()?.takeIf { it.isNotEmpty() }
val since = filter.since
@@ -128,8 +201,7 @@ internal object MergeQueryExecutor {
val orderBy =
if (indexStrategy.useAndIndexIdOnOrderBy) "created_at DESC, id ASC" else "created_at DESC"
val stmts = ArrayList<SQLiteStatement>((kinds?.size ?: 1) * authors.size)
if (kinds != null) {
return if (kinds != null) {
val sql =
buildString {
append("SELECT ").append(COLS)
@@ -139,16 +211,14 @@ internal object MergeQueryExecutor {
if (since != null) append(" AND created_at >= ?")
append(" ORDER BY ").append(orderBy)
}
for (kind in kinds) {
for (author in authors) {
val stmt = db.prepare(sql)
var p = 1
stmt.bindLong(p++, kind.toLong())
stmt.bindText(p++, author)
if (until != null) stmt.bindLong(p++, until)
if (since != null) stmt.bindLong(p++, since)
stmts.add(stmt)
}
buildStreams(kinds.size * authors.size) { i ->
val stmt = db.prepare(sql)
var p = 1
stmt.bindLong(p++, kinds[i / authors.size].toLong())
stmt.bindText(p++, authors[i % authors.size])
if (until != null) stmt.bindLong(p++, until)
if (since != null) stmt.bindLong(p++, since)
stmt
}
} else {
val sql =
@@ -160,30 +230,135 @@ internal object MergeQueryExecutor {
if (since != null) append(" AND created_at >= ?")
append(" ORDER BY ").append(orderBy)
}
for (author in authors) {
buildStreams(authors.size) { i ->
val stmt = db.prepare(sql)
var p = 1
stmt.bindText(p++, author)
stmt.bindText(p++, authors[i])
if (until != null) stmt.bindLong(p++, until)
if (since != null) stmt.bindLong(p++, since)
stmts.add(stmt)
stmt
}
}
}
/** Prepares one bound, newest-first cursor per tag-value stream. */
private fun prepareTagStreams(
db: SQLiteConnection,
filter: QueryBuilder.FilterWithDTags,
hasher: TagNameValueHasher,
): List<SQLiteStatement> {
val entry = filter.nonDTagsIn!!.entries.first()
val tagName = entry.key
val values = entry.value.distinct()
val kinds = filter.kinds?.distinct()?.takeIf { it.isNotEmpty() }
val since = filter.since
val until = filter.until
// The tag cursors stream off event_tags (which has no id column), so
// the tie order can only be created_at DESC — see the class doc.
return if (kinds != null) {
val sql =
buildString {
append("SELECT ").append(EH_COLS)
append(" FROM event_tags INDEXED BY query_by_tags_hash_kind")
append(" JOIN event_headers ON event_headers.row_id = event_tags.event_header_row_id")
append(" WHERE event_tags.tag_hash = ? AND event_tags.kind = ?")
if (until != null) append(" AND event_tags.created_at <= ?")
if (since != null) append(" AND event_tags.created_at >= ?")
append(" ORDER BY event_tags.created_at DESC")
}
buildStreams(values.size * kinds.size) { i ->
val stmt = db.prepare(sql)
var p = 1
stmt.bindLong(p++, hasher.hash(tagName, values[i / kinds.size]))
stmt.bindLong(p++, kinds[i % kinds.size].toLong())
if (until != null) stmt.bindLong(p++, until)
if (since != null) stmt.bindLong(p++, since)
stmt
}
} else {
val sql =
buildString {
append("SELECT ").append(EH_COLS)
append(" FROM event_tags INDEXED BY query_by_tags_hash")
append(" JOIN event_headers ON event_headers.row_id = event_tags.event_header_row_id")
append(" WHERE event_tags.tag_hash = ?")
if (until != null) append(" AND event_tags.created_at <= ?")
if (since != null) append(" AND event_tags.created_at >= ?")
append(" ORDER BY event_tags.created_at DESC")
}
buildStreams(values.size) { i ->
val stmt = db.prepare(sql)
var p = 1
stmt.bindLong(p++, hasher.hash(tagName, values[i]))
if (until != null) stmt.bindLong(p++, until)
if (since != null) stmt.bindLong(p++, since)
stmt
}
}
return stmts
}
/**
* Runs the merge, calling [onRow] with each winning cursor positioned on
* the row to emit, newest-first, up to `limit`. [onRow] must read the
* current row (it stays valid until the next step).
* Runs the merge for whichever shape [filter] matches, calling [onRow]
* with each winning cursor positioned on the row to emit, newest-first,
* up to `limit`. [onRow] must read the current row (it stays valid until
* the next step). [hasher] is only consulted for the tag shape.
*/
fun run(
db: SQLiteConnection,
filter: QueryBuilder.FilterWithDTags,
indexStrategy: IndexingStrategy,
hasher: (SQLiteConnection) -> TagNameValueHasher,
onRow: (SQLiteStatement) -> Unit,
) {
if (authorStreamCount(filter, indexStrategy) > 0) {
// One pubkey per event ⇒ author streams never overlap: no dedup.
mergeStreams(prepareAuthorStreams(db, filter, indexStrategy), filter.limit!!, dedup = false, onRow)
} else {
// A single event can match several tag values ⇒ dedup by id.
mergeStreams(prepareTagStreams(db, filter, hasher(db)), filter.limit!!, dedup = true, onRow)
}
}
/**
* Prepares [count] cursors via [prepareOne], closing any already-prepared
* statements if a later prepare throws — otherwise a mid-loop failure would
* strand checked-out, un-reset handles in the pooled connection (dead
* slots holding read locks). On success the caller ([mergeStreams]) owns
* closing them.
*/
private inline fun buildStreams(
count: Int,
prepareOne: (Int) -> SQLiteStatement,
): List<SQLiteStatement> {
val stmts = ArrayList<SQLiteStatement>(count)
try {
for (i in 0 until count) stmts.add(prepareOne(i))
} catch (e: Throwable) {
for (s in stmts) {
try {
s.close()
} catch (_: Throwable) {
}
}
throw e
}
return stmts
}
/**
* Heap-free k-way merge over the prepared [stmts]: repeatedly emits the
* newest live head (`created_at DESC`, tie `id ASC`) until [limit] rows
* are emitted or every stream is drained. When [dedup] is set an event id
* already emitted is skipped (its cursor still advances), so a row that
* surfaces in several streams is emitted once.
*/
private fun mergeStreams(
stmts: List<SQLiteStatement>,
limit: Int,
dedup: Boolean,
onRow: (SQLiteStatement) -> Unit,
) {
val stmts = prepareStreams(db, filter, indexStrategy)
try {
val k = stmts.size
val headCreatedAt = LongArray(k)
@@ -199,8 +374,8 @@ internal object MergeQueryExecutor {
}
}
val seen = if (dedup) HashSet<String>() else null
var emitted = 0
val limit = filter.limit!!
while (emitted < limit) {
// Pick the newest live head: created_at DESC, then id ASC.
var best = -1
@@ -215,8 +390,11 @@ internal object MergeQueryExecutor {
}
if (best == -1) break
onRow(stmts[best]) // cursor is still on the head row
emitted++
// Emit unless this id was already emitted by another stream.
if (seen == null || seen.add(headId[best]!!)) {
onRow(stmts[best]) // cursor is still on the head row
emitted++
}
// Advance the winner to its next row.
if (stmts[best].step()) {

111
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt
View File

@@ -49,7 +49,7 @@ class QueryBuilder(
val merge = filter.toFilterWithDTags()
if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
val out = ArrayList<T>(merge.limit!!)
MergeQueryExecutor.run(db, merge, indexStrategy) { out.add(it.toEvent()) }
MergeQueryExecutor.run(db, merge, indexStrategy, hasher) { out.add(it.toEvent()) }
return out
}
return db.runQuery(toSql(filter, hasher(db)))
@@ -62,7 +62,7 @@ class QueryBuilder(
) {
val merge = filter.toFilterWithDTags()
if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
MergeQueryExecutor.run(db, merge, indexStrategy) { onEach(it.toEvent()) }
MergeQueryExecutor.run(db, merge, indexStrategy, hasher) { onEach(it.toEvent()) }
return
}
db.runQuery(toSql(filter, hasher(db)), onEach)
@@ -102,7 +102,7 @@ class QueryBuilder(
val merge = filter.toFilterWithDTags()
if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
val out = ArrayList<RawEvent>(merge.limit!!)
MergeQueryExecutor.run(db, merge, indexStrategy) { out.add(it.toRawEvent()) }
MergeQueryExecutor.run(db, merge, indexStrategy, hasher) { out.add(it.toRawEvent()) }
return out
}
return db.runRawQuery(toSql(filter, hasher(db)))
@@ -115,7 +115,7 @@ class QueryBuilder(
) {
val merge = filter.toFilterWithDTags()
if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
MergeQueryExecutor.run(db, merge, indexStrategy) { onEach(it.toRawEvent()) }
MergeQueryExecutor.run(db, merge, indexStrategy, hasher) { onEach(it.toRawEvent()) }
return
}
db.runRawQuery(toSql(filter, hasher(db)), onEach)
@@ -202,13 +202,17 @@ class QueryBuilder(
)
}
val rowIdSubqueries = prepareRowIDSubQueries(filter, hasher)
// A search term that survived the simple-search branch above always
// combines with tags here (search + #t etc.). NIP-50 still orders by
// relevance, so carry the FTS rank through the subquery and out.
val rankSearch = newFilter.search != null && newFilter.search.isNotEmpty()
val rowIdSubqueries = prepareRowIDSubQueries(filter, hasher, projectRank = rankSearch)
return if (rowIdSubqueries == null) {
QuerySpec(makeEverythingQuery())
} else {
QuerySpec(
makeQueryIn(rowIdSubqueries.sql),
makeQueryIn(rowIdSubqueries.sql, orderByRank = rankSearch),
rowIdSubqueries.args,
)
}
@@ -220,7 +224,13 @@ class QueryBuilder(
): QuerySpec {
if (filters.size == 1) return toSql(filters.first(), hasher)
val rowIdSubqueries = unionSubqueriesIfNeeded(filters, hasher)
// A multi-filter search REQ (e.g. the client's search-across-kinds,
// all filters sharing one term) must still be NIP-50 relevance-ordered.
// Only when EVERY branch is a search branch (and FTS is on, so each has
// a rank column) — a non-search branch has no defined relevance, so a
// mixed REQ falls back to created_at.
val rankSearch = fts.enabled && filters.all { !it.search.isNullOrEmpty() }
val rowIdSubqueries = unionSubqueriesIfNeeded(filters, hasher, projectRank = rankSearch)
return if (rowIdSubqueries == null) {
QuerySpec(
@@ -229,7 +239,7 @@ class QueryBuilder(
)
} else {
QuerySpec(
makeQueryIn(rowIdSubqueries.sql),
makeQueryIn(rowIdSubqueries.sql, orderByRank = rankSearch),
rowIdSubqueries.args,
)
}
@@ -427,7 +437,7 @@ class QueryBuilder(
val sql =
buildString {
append("SELECT event_headers.id, event_headers.created_at FROM event_headers")
append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.${fts.eventHeaderRowIdName}")
append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.rowid")
if (clause.conditions.isNotEmpty()) {
append("\nWHERE ${clause.conditions}")
}
@@ -470,14 +480,20 @@ class QueryBuilder(
private fun makeEverythingQuery() = "SELECT id, pubkey, created_at, kind, tags, content, sig FROM event_headers ORDER BY created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}"
private fun makeQueryIn(rowIdQuery: String) =
"""
// [orderByRank] presents the joined result in NIP-50 relevance order: the
// subquery (built with `projectRank`) exposes the FTS bm25 score as a
// `rank` column, and `created_at DESC` is only a tie-break. Off, it keeps
// the default newest-first ordering for every non-search shape.
private fun makeQueryIn(
rowIdQuery: String,
orderByRank: Boolean = false,
) = """
SELECT id, pubkey, created_at, kind, tags, content, sig FROM event_headers
INNER JOIN (
$rowIdQuery
) AS filtered
ON event_headers.row_id = filtered.row_id
ORDER BY created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}
ORDER BY ${if (orderByRank) "filtered.rank, " else ""}created_at DESC${if (indexStrategy.useAndIndexIdOnOrderBy) ", id ASC" else ""}
""".trimIndent()
private fun <T : Event> SQLiteConnection.runQuery(query: QuerySpec): List<T> =
@@ -713,16 +729,34 @@ class QueryBuilder(
fun unionSubqueriesIfNeeded(
filters: List<Filter>,
hasher: TagNameValueHasher,
// See [prepareRowIDSubQueries]. When set, every branch is a search
// branch that also projects a `rank` column; the union keeps one row
// per event with its BEST (min) bm25 score so the caller can present
// the whole multi-filter search REQ NIP-50-ranked. Callers must only
// pass true when all filters carry a search term (else a branch has no
// rank column). Off for count/delete, which stay single-column.
projectRank: Boolean = false,
): QuerySpec? {
val inner =
filters.mapNotNull { filter ->
prepareRowIDSubQueries(filter, hasher)
prepareRowIDSubQueries(filter, hasher, projectRank)
}
if (inner.isEmpty()) return null
return if (inner.size == 1) {
inner.first()
if (inner.size == 1) return inner.first()
return if (projectRank) {
// UNION ALL keeps every (row_id, rank) so an event matching two
// branches under different terms isn't dropped before MIN; GROUP BY
// then dedups by event keeping the best score.
QuerySpec(
sql =
"SELECT row_id, MIN(rank) as rank FROM (\n " +
inner.joinToString("\n UNION ALL\n ") { "SELECT row_id, rank FROM (${it.sql})" } +
"\n ) GROUP BY row_id",
args = inner.flatMap { it.args },
)
} else {
QuerySpec(
sql = inner.joinToString("\n UNION\n ") { "SELECT row_id FROM (${it.sql})" },
@@ -748,6 +782,13 @@ class QueryBuilder(
fun prepareRowIDSubQueries(
filter: Filter,
hasher: TagNameValueHasher,
// When set on a search filter, the subquery also projects the FTS
// bm25 score as a `rank` column and orders its own LIMIT by relevance,
// so the caller ([makeQueryIn] with `orderByRank`) can present results
// NIP-50-ranked. Off (the default) for count/delete/union/negentropy,
// which never expose a second column (a two-column subquery breaks
// `row_id IN (…)`) and don't rank.
projectRank: Boolean = false,
): QuerySpec? {
if (filter.isEmpty()) return null
@@ -761,6 +802,10 @@ class QueryBuilder(
val mustJoinSearch = filter.search != null && fts.enabled
// Only emit the rank column when there is actually an FTS join to take
// it from; a `projectRank` request on a tag-only filter is ignored.
val emitRank = projectRank && mustJoinSearch
val nonDTagsIn = filter.tags?.filter { it.key != "d" } ?: emptyMap()
val nonDTagsAll = filter.tagsAll?.filter { it.key != "d" } ?: emptyMap()
@@ -789,7 +834,11 @@ class QueryBuilder(
buildString {
// always do tags if there are any
if (reverseLookup) {
append("SELECT DISTINCT(event_tags.event_header_row_id) as row_id FROM event_tags")
append("SELECT DISTINCT(event_tags.event_header_row_id) as row_id")
// rank is functionally determined by the row_id (one FTS
// row per event), so it doesn't change what DISTINCT folds.
if (emitRank) append(", ${fts.tableName}.rank as rank")
append(" FROM event_tags")
// it's quite rare to have 2 tags in the filter, but possible
nonDTagsIn.keys.forEachIndexed { index, tagName ->
@@ -815,13 +864,15 @@ class QueryBuilder(
}
if (mustJoinSearch) {
append(" INNER JOIN ${fts.tableName} ON ${fts.tableName}.${fts.eventHeaderRowIdName} = event_tags.event_header_row_id")
append(" INNER JOIN ${fts.tableName} ON ${fts.tableName}.rowid = event_tags.event_header_row_id")
}
} else if (mustJoinSearch) {
append("SELECT ${fts.tableName}.${fts.eventHeaderRowIdName} as row_id FROM ${fts.tableName}")
append("SELECT ${fts.tableName}.rowid as row_id")
if (emitRank) append(", ${fts.tableName}.rank as rank")
append(" FROM ${fts.tableName}")
if (hasHeaders) {
append(" INNER JOIN event_headers ON event_headers.row_id = ${fts.tableName}.${fts.eventHeaderRowIdName}")
append(" INNER JOIN event_headers ON event_headers.row_id = ${fts.tableName}.rowid")
}
} else {
// no tags and no search.
@@ -934,15 +985,17 @@ class QueryBuilder(
append(" WHERE ${clause.conditions}")
}
if (filter.limit != null) {
if (reverseLookup) {
if (emitRank) {
// NIP-50: the LIMIT keeps the most RELEVANT rows, not
// the newest, so the inner cut is by rank too.
append(" ORDER BY rank")
} else if (reverseLookup) {
append(" ORDER BY event_tags.created_at DESC")
append(" LIMIT ")
append(filter.limit)
} else {
append(" ORDER BY event_headers.created_at DESC")
append(" LIMIT ")
append(filter.limit)
}
append(" LIMIT ")
append(filter.limit)
}
}
@@ -1001,11 +1054,17 @@ class QueryBuilder(
val sql =
buildString {
append("SELECT event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig FROM event_headers")
append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.${fts.eventHeaderRowIdName}")
append("\nINNER JOIN ${fts.tableName} ON event_headers.row_id = ${fts.tableName}.rowid")
if (clause.conditions.isNotEmpty()) {
append("\nWHERE ${clause.conditions}")
}
append("\nORDER BY event_headers.created_at DESC")
// NIP-50: search results are ordered by relevance ("quality of
// search result"), not created_at, and the limit is applied
// after the score. FTS5 exposes bm25 as the `rank` column (more
// negative = more relevant), so ORDER BY rank ascending is
// best-match-first. created_at DESC is only a tie-break so
// equally-relevant matches come newest-first deterministically.
append("\nORDER BY ${fts.tableName}.rank, event_headers.created_at DESC")
if (indexStrategy.useAndIndexIdOnOrderBy) {
append(", event_headers.id ASC")
}

14
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SQLiteEventStore.kt
View File

@@ -60,7 +60,7 @@ class SQLiteEventStore(
val extraPragmas: List<String> = emptyList(),
) {
companion object {
const val DATABASE_VERSION = 4
const val DATABASE_VERSION = 5
}
val seedModule = SeedModule()
@@ -213,6 +213,12 @@ class SQLiteEventStore(
// watermark seeds at the current MAX(row_id).
fullTextSearchModule.createStateTable(db)
}
4 -> {
// Upgrade from version 4 to 5: the FTS index became a
// contentless table keyed by event_headers.row_id. The old
// rowids can't be remapped, so drop and repopulate.
fullTextSearchModule.migrateV4ToContentless(db)
}
}
}
}
@@ -274,6 +280,12 @@ class SQLiteEventStore(
pool.useWriter { db ->
db.execSQL("PRAGMA analysis_limit = 400;")
db.execSQL("PRAGMA optimize;")
// Fold a bounded FTS segment merge into the same periodic
// maintenance tick: incremental (and deferred catch-up) inserts
// leave the NIP-50 index as many small segments, and a MATCH
// queries every one. Bounded so the tick stays cheap; a no-op when
// there is nothing to merge or FTS is off.
fullTextSearchModule.mergeSegments(db)
}
/**

80
quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StatementCachingConnection.kt
View File

@@ -34,46 +34,72 @@ import androidx.sqlite.SQLiteStatement
* real reset + clearBindings happens on the next checkout). Statements are
* only truly finalized when the connection itself closes.
*
* Each SQL string caches a small **pool** of handles rather than a single
* one, so overlapping checkouts of the *same* SQL all reuse cached handles.
* That is exactly the k-way-merge query shape ([MergeQueryExecutor]): it
* opens one identical-SQL cursor per author/tag stream — dozens to hundreds
* live at once — which a single-handle cache could not serve (every stream
* past the first fell back to an uncached prepare). The pool lets a repeated
* follow-feed / reactions-watcher REQ reuse its per-stream cursors instead
* of re-preparing them each poll.
*
* Constraints, by design of the call sites:
* - **Not thread-safe** — same contract as the underlying connection,
* which the pool already serializes (single writer under a mutex).
* - **No overlapping use of the same SQL** — checking out one SQL string
* twice without closing the first use would alias one native handle.
* Insert/query paths never nest the same statement; a checkout while
* the previous one is still open falls back to an uncached statement.
* which the pool already serializes (single writer under a mutex; each
* reader held by one coroutine at a time).
*/
class StatementCachingConnection(
private val delegate: SQLiteConnection,
/**
* Ceiling on retained statements. Query SQL embeds one `?` per filter
* element, so shape variety is client-controlled — without a cap a
* long-lived relay connection would accumulate native handles without
* bound. Once full, unseen SQL just prepares uncached. 256 comfortably
* covers the write path's fixed set plus the recurring filter shapes.
* Ceiling on retained statements across all SQL strings. Query SQL
* embeds one `?` per filter element, so shape variety is
* client-controlled — without a cap a long-lived relay connection would
* accumulate native handles without bound. Once full, unseen SQL (or an
* extra concurrent copy of a cached SQL) just prepares uncached. 512
* covers the write path's fixed set, the recurring single-shot filter
* shapes, and a few hundred concurrent per-stream merge cursors.
*/
private val maxCachedStatements: Int = 256,
private val maxCachedStatements: Int = 512,
) : SQLiteConnection by delegate {
private val cache = HashMap<String, CachedStatement>()
// One reusable pool per SQL string. Several entries of the same pool may
// be checked out simultaneously (the merge path); a `prepare` reuses the
// first free entry, grows the pool while under the global cap, and only
// then falls back to an uncached statement.
private val cache = HashMap<String, ArrayList<CachedStatement>>()
private var cachedCount = 0
override fun prepare(sql: String): SQLiteStatement {
val cached =
cache[sql] ?: run {
if (cache.size >= maxCachedStatements) return delegate.prepare(sql)
CachedStatement(delegate.prepare(sql)).also { cache[sql] = it }
val pool = cache[sql]
if (pool != null) {
for (i in pool.indices) {
val stmt = pool[i]
if (!stmt.checkedOut) {
stmt.checkedOut = true
stmt.clearBindings()
return stmt
}
}
// Every pooled handle for this SQL is in use — grow if the global
// budget allows, else serve an uncached statement.
if (cachedCount >= maxCachedStatements) return delegate.prepare(sql)
return CachedStatement(delegate.prepare(sql)).also {
it.checkedOut = true
pool.add(it)
cachedCount++
}
if (cached.checkedOut) {
// Same SQL prepared while the previous handle is still in use —
// stay correct with a plain uncached statement.
return delegate.prepare(sql)
}
cached.checkedOut = true
cached.clearBindings()
return cached
if (cachedCount >= maxCachedStatements) return delegate.prepare(sql)
return CachedStatement(delegate.prepare(sql)).also {
it.checkedOut = true
cache[sql] = arrayListOf(it)
cachedCount++
}
}
override fun close() {
cache.values.forEach { runCatching { it.finalize() } }
cache.values.forEach { pool -> pool.forEach { runCatching { it.finalizeStatement() } } }
cache.clear()
cachedCount = 0
delegate.close()
}
@@ -94,6 +120,10 @@ class StatementCachingConnection(
checkedOut = false
}
fun finalize() = delegate.close()
// Not named `finalize`: a no-arg `finalize()` is treated by the JVM as
// Object.finalize(), so the GC would call it and double-close the
// native handle after our explicit close(). This is only ever invoked
// explicitly from the connection's close().
fun finalizeStatement() = delegate.close()
}
}

34
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryAssemblerTest.kt
View File

@@ -238,9 +238,9 @@ class QueryAssemblerTest : BaseDBTest() {
INNER JOIN (
SELECT row_id FROM (SELECT event_headers.row_id as row_id FROM event_headers ORDER BY event_headers.created_at DESC LIMIT 10)
UNION
SELECT row_id FROM (SELECT event_fts.event_header_row_id as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.event_header_row_id WHERE (event_headers.kind IN ("1", "1111")) AND (event_headers.pubkey = "7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d") AND (event_fts MATCH "keywords") ORDER BY event_headers.created_at DESC LIMIT 100)
SELECT row_id FROM (SELECT event_fts.rowid as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.rowid WHERE (event_headers.kind IN ("1", "1111")) AND (event_headers.pubkey = "7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d") AND (event_fts MATCH "keywords") ORDER BY event_headers.created_at DESC LIMIT 100)
UNION
SELECT row_id FROM (SELECT event_fts.event_header_row_id as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.event_header_row_id WHERE (event_headers.kind = "20") AND (event_fts MATCH "cats") ORDER BY event_headers.created_at DESC LIMIT 30)
SELECT row_id FROM (SELECT event_fts.rowid as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.rowid WHERE (event_headers.kind = "20") AND (event_fts MATCH "cats") ORDER BY event_headers.created_at DESC LIMIT 30)
) AS filtered
ON event_headers.row_id = filtered.row_id
ORDER BY $orderBy
@@ -252,13 +252,13 @@ class QueryAssemblerTest : BaseDBTest() {
│ │ └── SCAN (subquery-1)
│ ├── UNION USING TEMP B-TREE
│ │ ├── CO-ROUTINE (subquery-3)
│ │ │ ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
│ │ │ ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
│ │ │ ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
│ │ │ └── USE TEMP B-TREE FOR ORDER BY
│ │ └── SCAN (subquery-3)
│ └── UNION USING TEMP B-TREE
│ ├── CO-ROUTINE (subquery-5)
│ │ ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
│ │ ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
│ │ ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
│ │ └── USE TEMP B-TREE FOR ORDER BY
│ └── SCAN (subquery-5)
@@ -275,9 +275,9 @@ class QueryAssemblerTest : BaseDBTest() {
INNER JOIN (
SELECT row_id FROM (SELECT event_headers.row_id as row_id FROM event_headers ORDER BY event_headers.created_at DESC LIMIT 10)
UNION
SELECT row_id FROM (SELECT event_fts.event_header_row_id as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.event_header_row_id WHERE (event_headers.kind IN ("1", "1111")) AND (event_headers.pubkey = "7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d") AND (event_fts MATCH "keywords") ORDER BY event_headers.created_at DESC LIMIT 100)
SELECT row_id FROM (SELECT event_fts.rowid as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.rowid WHERE (event_headers.kind IN ("1", "1111")) AND (event_headers.pubkey = "7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d") AND (event_fts MATCH "keywords") ORDER BY event_headers.created_at DESC LIMIT 100)
UNION
SELECT row_id FROM (SELECT event_fts.event_header_row_id as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.event_header_row_id WHERE (event_headers.kind = "20") AND (event_fts MATCH "cats") ORDER BY event_headers.created_at DESC LIMIT 30)
SELECT row_id FROM (SELECT event_fts.rowid as row_id FROM event_fts INNER JOIN event_headers ON event_headers.row_id = event_fts.rowid WHERE (event_headers.kind = "20") AND (event_fts MATCH "cats") ORDER BY event_headers.created_at DESC LIMIT 30)
) AS filtered
ON event_headers.row_id = filtered.row_id
ORDER BY $orderBy
@@ -290,13 +290,13 @@ class QueryAssemblerTest : BaseDBTest() {
│ │ └── SCAN (subquery-1)
│ ├── UNION USING TEMP B-TREE
│ │ ├── CO-ROUTINE (subquery-3)
│ │ │ ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
│ │ │ ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
│ │ │ ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
│ │ │ └── USE TEMP B-TREE FOR ORDER BY
│ │ └── SCAN (subquery-3)
│ └── UNION USING TEMP B-TREE
│ ├── CO-ROUTINE (subquery-5)
│ │ ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
│ │ ├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
│ │ ├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
│ │ └── USE TEMP B-TREE FOR ORDER BY
│ └── SCAN (subquery-5)
@@ -712,10 +712,10 @@ class QueryAssemblerTest : BaseDBTest() {
assertEquals(
"""
SELECT event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig FROM event_headers
INNER JOIN event_fts ON event_headers.row_id = event_fts.event_header_row_id
INNER JOIN event_fts ON event_headers.row_id = event_fts.rowid
WHERE (event_fts MATCH "keywords") AND (event_headers.pubkey IN ("7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d", "f3ac434d61bc0f491a814782ccfdf9c439dae1f0bde9097ad4a245f4c495cd14", "12ae0fd81c85e1e7d9ed096397dc3129849425fe6f8afce7213ebf38ddfc6ca9"))
ORDER BY event_headers.created_at DESC, event_headers.id ASC
├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
ORDER BY event_fts.rank, event_headers.created_at DESC, event_headers.id ASC
├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
└── USE TEMP B-TREE FOR ORDER BY
""".trimIndent(),
@@ -725,10 +725,10 @@ class QueryAssemblerTest : BaseDBTest() {
assertEquals(
"""
SELECT event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig FROM event_headers
INNER JOIN event_fts ON event_headers.row_id = event_fts.event_header_row_id
INNER JOIN event_fts ON event_headers.row_id = event_fts.rowid
WHERE (event_fts MATCH "keywords") AND (event_headers.pubkey IN ("7c5eb72a4584fdaaeaa145b25c92ea9917704224951219dbd43acef9e91fb88d", "f3ac434d61bc0f491a814782ccfdf9c439dae1f0bde9097ad4a245f4c495cd14", "12ae0fd81c85e1e7d9ed096397dc3129849425fe6f8afce7213ebf38ddfc6ca9"))
ORDER BY event_headers.created_at DESC
├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
ORDER BY event_fts.rank, event_headers.created_at DESC
├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
└── USE TEMP B-TREE FOR ORDER BY
""".trimIndent(),
@@ -741,14 +741,14 @@ class QueryAssemblerTest : BaseDBTest() {
fun testKindAndSearch() =
forEachDB { db ->
val filter = Filter(kinds = listOf(1, 1111, 10000), search = "keywords")
val orderBy = if (db.indexStrategy.useAndIndexIdOnOrderBy) "event_headers.created_at DESC, event_headers.id ASC" else "event_headers.created_at DESC"
val orderBy = if (db.indexStrategy.useAndIndexIdOnOrderBy) "event_fts.rank, event_headers.created_at DESC, event_headers.id ASC" else "event_fts.rank, event_headers.created_at DESC"
assertEquals(
"""
SELECT event_headers.id, event_headers.pubkey, event_headers.created_at, event_headers.kind, event_headers.tags, event_headers.content, event_headers.sig FROM event_headers
INNER JOIN event_fts ON event_headers.row_id = event_fts.event_header_row_id
INNER JOIN event_fts ON event_headers.row_id = event_fts.rowid
WHERE (event_fts MATCH "keywords") AND (event_headers.kind IN ("1", "1111", "10000"))
ORDER BY $orderBy
├── SCAN event_fts VIRTUAL TABLE INDEX 0:M2
├── SCAN event_fts VIRTUAL TABLE INDEX 0:M1
├── SEARCH event_headers USING INTEGER PRIMARY KEY (rowid=?)
└── USE TEMP B-TREE FOR ORDER BY
""".trimIndent(),

202
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/SearchRelevanceOrderTest.kt Normal file
View File

@@ -0,0 +1,202 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.store.sqlite
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
import com.vitorpamplona.quartz.utils.EventFactory
import kotlinx.coroutines.runBlocking
import kotlin.test.Test
import kotlin.test.assertEquals
/**
* NIP-50: search results are ordered by "quality of search result" (relevance),
* **not** by `created_at`, and the limit is applied after the score. The store
* uses FTS5 bm25 (`ORDER BY event_fts.rank`), so a stronger match outranks a
* newer one.
*/
class SearchRelevanceOrderTest {
private val signer = NostrSignerSync()
private fun note(
content: String,
createdAt: Long,
) = signer.sign(TextNoteEvent.build(content, createdAt = createdAt))
@Test
fun strongerMatchOutranksNewer() =
runBlocking {
val store = EventStore(dbName = null)
try {
// Older, but the term appears 3× in a short doc → most relevant.
val strong = note("bitcoin bitcoin bitcoin", createdAt = 1_000)
// Newer, term once buried in a long doc → least relevant.
val weak = note("bitcoin is one topic among many other unrelated words here padding", createdAt = 9_000)
// Newer still, but does not match at all.
val nonMatch = note("completely different subject entirely", createdAt = 9_999)
store.insert(weak)
store.insert(strong)
store.insert(nonMatch)
val results = store.query<Event>(Filter(search = "bitcoin", limit = 10)).map { it.id }
// Relevance order (strong before weak) — the opposite of
// created_at DESC (which would put weak first) — and the
// non-matching note is absent.
assertEquals(listOf(strong.id, weak.id), results)
} finally {
store.close()
}
}
@Test
fun limitAppliesAfterRelevanceScore() =
runBlocking {
val store = EventStore(dbName = null)
try {
// Three docs of decreasing relevance but increasing created_at,
// so created_at order and relevance order are exact opposites.
val best = note("apple apple apple apple", createdAt = 1)
val mid = note("apple apple filler words", createdAt = 2)
val worst = note("apple among lots of other unrelated filler words here", createdAt = 3)
store.insert(best)
store.insert(mid)
store.insert(worst)
// limit=2 after scoring keeps the two MOST RELEVANT, not the
// two newest.
val top2 = store.query<Event>(Filter(search = "apple", limit = 2)).map { it.id }
assertEquals(listOf(best.id, mid.id), top2)
} finally {
store.close()
}
}
/** A `#t` tag makes this a `search + tag` filter — the combined path. */
private var idSeq = 0
private fun hexId(n: Int): String {
val s = n.toString(16)
return "0".repeat(64 - s.length) + s
}
private fun tagged(
content: String,
createdAt: Long,
topic: String,
): Event =
EventFactory.create(
hexId(++idSeq),
"00".repeat(32),
createdAt,
1,
arrayOf(arrayOf("t", topic)),
content,
"0".repeat(128),
)
private fun evk(
content: String,
createdAt: Long,
kind: Int,
): Event = EventFactory.create(hexId(++idSeq), "00".repeat(32), createdAt, kind, arrayOf(), content, "0".repeat(128))
@Test
fun multiFilterSearchIsRelevanceOrderedAcrossBranches() =
runBlocking {
val store = EventStore(dbName = null)
try {
// Relevance A > B > C, created_at A < B < C (reverse), and the
// branches split by kind: A,C are kind 1 (TextNote), B is kind
// 1111 (Comment) — both searchable kinds.
val a = evk("apple apple apple apple", 1, 1)
val b = evk("apple apple apple", 2, 1111)
val c = evk("apple filler filler filler filler filler", 3, 1)
store.batchInsert(listOf(a, b, c))
// The client's search-across-kinds shape: one term, two filters.
val filters =
listOf(
Filter(search = "apple", kinds = listOf(1), limit = 100),
Filter(search = "apple", kinds = listOf(1111), limit = 100),
)
val ids = store.query<Event>(filters).map { it.id }
assertEquals(listOf(a.id, b.id, c.id), ids, "multi-filter search must be relevance-ordered across branches")
} finally {
store.close()
}
}
@Test
fun multiFilterSearchDedupsEventsMatchingSeveralBranches() =
runBlocking {
val store = EventStore(dbName = null)
try {
val x = evk("banana banana banana", 1, 1)
val y = evk("banana one", 2, 1)
store.batchInsert(listOf(x, y))
// Overlapping branches: both kind-1 events match BOTH filters.
// GROUP BY row_id must fold each to a single ranked row.
val filters =
listOf(
Filter(search = "banana", kinds = listOf(1, 6), limit = 100),
Filter(search = "banana", kinds = listOf(1, 2), limit = 100),
)
val ids = store.query<Event>(filters).map { it.id }
assertEquals(ids.size, ids.toSet().size, "no event may appear twice across branches")
assertEquals(listOf(x.id, y.id), ids, "deduped, relevance-ordered")
assertEquals(2, store.count(filters), "count parity across the union")
} finally {
store.close()
}
}
@Test
fun searchWithATagIsAlsoRelevanceOrdered() =
runBlocking {
val store = EventStore(dbName = null)
try {
// All tagged #t=nostr; relevance decreases as created_at rises,
// so created_at order would be the exact reverse of relevance.
val strong = tagged("nostr nostr nostr", createdAt = 1, topic = "nostr")
val weak = tagged("nostr among many other unrelated filler words here padding", createdAt = 2, topic = "nostr")
// Matches the term but wrong tag → excluded by the tag filter.
val wrongTag = tagged("nostr nostr nostr nostr", createdAt = 3, topic = "other")
// Right tag but doesn't match the term → excluded by search.
val noMatch = tagged("bitcoin only", createdAt = 4, topic = "nostr")
store.batchInsert(listOf(strong, weak, wrongTag, noMatch))
val filter = Filter(search = "nostr", tags = mapOf("t" to listOf("nostr")), limit = 10)
val ids = store.query<Event>(filter).map { it.id }
assertEquals(listOf(strong.id, weak.id), ids, "search + tag must be relevance-ordered, tag-scoped")
// limit after score keeps the most relevant one.
val top1 = store.query<Event>(filter.copy(limit = 1)).map { it.id }
assertEquals(listOf(strong.id), top1)
} finally {
store.close()
}
}
}

113
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/StatementCachingConnectionTest.kt Normal file
View File

@@ -0,0 +1,113 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.store.sqlite
import androidx.sqlite.SQLiteStatement
import androidx.sqlite.driver.bundled.BundledSQLiteDriver
import kotlin.test.AfterTest
import kotlin.test.BeforeTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertNotSame
import kotlin.test.assertSame
class StatementCachingConnectionTest {
private lateinit var conn: StatementCachingConnection
private val sql = "SELECT ? AS v"
@BeforeTest
fun setup() {
conn = StatementCachingConnection(BundledSQLiteDriver().open(":memory:"))
}
@AfterTest
fun tearDown() {
conn.close()
}
private fun readOne(stmt: SQLiteStatement): Long {
stmt.step()
return stmt.getLong(0)
}
@Test
fun sequentialSameSqlReusesTheSameHandle() {
var first: SQLiteStatement? = null
conn.prepare(sql).use { stmt ->
stmt.bindLong(1, 7)
assertEquals(7, readOne(stmt))
first = stmt
}
// Closed (returned to pool) — the next prepare of the same SQL must
// hand back the very same cached handle, not a fresh prepare.
conn.prepare(sql).use { stmt ->
assertSame(first, stmt)
stmt.bindLong(1, 9)
assertEquals(9, readOne(stmt))
}
}
@Test
fun concurrentSameSqlHandlesAreDistinctAndIndependent() {
// The k-way-merge shape: many identical-SQL cursors live at once.
val a = conn.prepare(sql)
val b = conn.prepare(sql)
val c = conn.prepare(sql)
assertNotSame(a, b)
assertNotSame(b, c)
assertNotSame(a, c)
a.bindLong(1, 1)
b.bindLong(1, 2)
c.bindLong(1, 3)
// Each cursor keeps its own bindings/position even while the others
// are open — no aliasing of one native handle.
assertEquals(1, readOne(a))
assertEquals(2, readOne(b))
assertEquals(3, readOne(c))
a.close()
b.close()
c.close()
// After release, a fresh concurrent burst reuses the pooled handles.
val reused = conn.prepare(sql)
assertSame(a, reused, "pool should hand back a freed handle before preparing anew")
reused.close()
}
@Test
fun overflowingTheGlobalCapFallsBackToUncached() {
val small = StatementCachingConnection(BundledSQLiteDriver().open(":memory:"), maxCachedStatements = 2)
try {
val live = (0 until 5).map { small.prepare(sql) }
// All five must be usable even though only two can be cached; the
// extra three are plain uncached statements.
live.forEachIndexed { i, stmt ->
stmt.bindLong(1, i.toLong())
assertEquals(i.toLong(), readOne(stmt))
}
live.forEach { it.close() }
} finally {
small.close()
}
}
}

295
quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/TagMergeCorrectnessTest.kt Normal file
View File

@@ -0,0 +1,295 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.store.sqlite
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.utils.EventFactory
import kotlinx.coroutines.runBlocking
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertFalse
import kotlin.test.assertTrue
/**
* Correctness guard for the tag-stream path of [MergeQueryExecutor]: the
* `#<x>=[values] (+ kinds) [+ since/until] limit=N` watcher shape
* (reactions/replies). The merge must return the same newest-N a
* `SELECT DISTINCT … ORDER BY created_at DESC LIMIT N` would, deduping events
* that carry several of the queried tag values.
*
* Where `created_at` is distinct the order is fully determined and asserted
* against a Kotlin reference. Where it ties, the tag cursors can only order by
* `created_at` (event_tags has no id column), so the result is a valid
* newest-N but not id-exact — those cases assert set + size instead.
*/
class TagMergeCorrectnessTest {
private val hex = "0123456789abcdef"
private fun mix(seed: Long): Long {
var z = seed + -0x61c8864680b583ebL
z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L
z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L
return z xor (z ushr 31)
}
private fun hex64(
salt: Long,
index: Int,
): String {
val out = CharArray(64)
for (w in 0 until 4) {
val v = mix(salt * 1_000_003 + index.toLong() * 4 + w)
for (b in 0 until 8) {
val byte = ((v ushr (b * 8)) and 0xFF).toInt()
out[(w * 8 + b) * 2] = hex[byte ushr 4]
out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF]
}
}
return out.concatToString()
}
private val sig = "0".repeat(128)
private var idSeq = 0
private fun ev(
createdAt: Long,
kind: Int,
eTags: List<String>,
): Event =
EventFactory.create(
hex64(7, idSeq++),
hex64(1, idSeq),
createdAt,
kind,
eTags.map { arrayOf("e", it) }.toTypedArray(),
"",
sig,
)
private val newestFirst =
Comparator<Event> { a, b ->
if (a.createdAt != b.createdAt) b.createdAt.compareTo(a.createdAt) else a.id.compareTo(b.id)
}
private fun reference(
all: List<Event>,
values: Set<String>,
kinds: Set<Int>?,
since: Long?,
until: Long?,
limit: Int,
): List<String> =
all
.asSequence()
.filter { e -> e.tags.any { it.size >= 2 && it[0] == "e" && it[1] in values } }
.filter { kinds == null || it.kind in kinds }
.filter { since == null || it.createdAt >= since }
.filter { until == null || it.createdAt <= until }
.sortedWith(newestFirst)
.map { it.id }
.distinct()
.take(limit)
.toList()
private fun mergeEligible(
store: EventStore,
filter: Filter,
): Boolean =
MergeQueryExecutor.streamCount(
with(store.store.queryBuilder) { filter.toFilterWithDTags() },
store.store.queryBuilder.indexStrategy,
) > 0
private fun newStore() =
EventStore(
dbName = null,
indexStrategy =
DefaultIndexingStrategy(
indexTagsByCreatedAtAlone = true,
useAndIndexIdOnOrderBy = true,
indexFullTextSearch = false,
),
)
@Test
fun distinctCreatedAt_withKinds_matchesReference() =
runBlocking {
val store = newStore()
val notes = (0 until 6).map { hex64(2, it) }
val all = ArrayList<Event>()
var t = 1_700_000_000L
for (round in 0 until 40) {
// reactions (kind 7) and replies (kind 1) to a rotating note
all.add(ev(t++, 7, listOf(notes[round % notes.size])))
all.add(ev(t++, 1, listOf(notes[(round + 1) % notes.size])))
}
// Noise: kind-7 to notes NOT in the query set, and other tags.
for (i in 0 until 100) all.add(ev(t++, 7, listOf(hex64(9, i))))
store.batchInsert(all)
val queried = notes.take(3)
val filter = Filter(kinds = listOf(7), tags = mapOf("e" to queried), limit = 25)
assertTrue(mergeEligible(store, filter), "tag watcher must be merge-eligible")
val merged = store.query<Event>(filter).map { it.id }
assertEquals(reference(all, queried.toSet(), setOf(7), null, null, 25), merged)
store.close()
}
@Test
fun crossStreamDuplicate_emittedOnce() =
runBlocking {
val store = newStore()
val a = hex64(3, 0)
val b = hex64(3, 1)
val all = ArrayList<Event>()
var t = 1_700_000_000L
// Events tagging BOTH a and b — they appear in both value streams
// and must be emitted exactly once.
repeat(5) { all.add(ev(t++, 1, listOf(a, b))) }
// Events tagging only one of them.
repeat(5) { all.add(ev(t++, 1, listOf(a))) }
repeat(5) { all.add(ev(t++, 1, listOf(b))) }
store.batchInsert(all)
val filter = Filter(kinds = listOf(1), tags = mapOf("e" to listOf(a, b)), limit = 500)
assertTrue(mergeEligible(store, filter))
val merged = store.query<Event>(filter).map { it.id }
assertEquals(merged.size, merged.toSet().size, "no event may be emitted twice")
assertEquals(15, merged.size, "5 both + 5 a-only + 5 b-only = 15 distinct")
assertEquals(reference(all, setOf(a, b), setOf(1), null, null, 500), merged)
store.close()
}
@Test
fun noKinds_usesTagCreatedAtIndex() =
runBlocking {
val store = newStore()
val notes = (0 until 5).map { hex64(4, it) }
val all = ArrayList<Event>()
var t = 1_700_000_000L
for (round in 0 until 30) all.add(ev(t++, (round % 3) + 1, listOf(notes[round % notes.size])))
for (i in 0 until 60) all.add(ev(t++, 1, listOf(hex64(8, i))))
store.batchInsert(all)
val queried = notes.take(3)
val filter = Filter(tags = mapOf("e" to queried), limit = 20)
assertTrue(mergeEligible(store, filter), "no-kind tag watcher must be merge-eligible with the tag index")
val merged = store.query<Event>(filter).map { it.id }
assertEquals(reference(all, queried.toSet(), null, null, null, 20), merged)
store.close()
}
@Test
fun withSinceAndUntil_boundsTheWindow() =
runBlocking {
val store = newStore()
val notes = (0 until 4).map { hex64(5, it) }
val all = ArrayList<Event>()
val base = 1_700_000_000L
for (i in 0 until 300) all.add(ev(base + i.toLong(), 7, listOf(notes[i % notes.size])))
store.batchInsert(all)
val since = base + 50
val until = base + 250
val filter = Filter(kinds = listOf(7), tags = mapOf("e" to notes), since = since, until = until, limit = 500)
assertTrue(mergeEligible(store, filter))
val merged = store.query<Event>(filter).map { it.id }
val ref = reference(all, notes.toSet(), setOf(7), since, until, 500)
assertEquals(ref, merged)
assertTrue(merged.isNotEmpty() && ref.size < 300)
store.close()
}
@Test
fun rawPathMatchesDecoded() =
runBlocking {
val store = newStore()
val notes = (0 until 6).map { hex64(6, it) }
val all = ArrayList<Event>()
var t = 1_700_000_000L
for (round in 0 until 25) all.add(ev(t++, 7, listOf(notes[round % notes.size])))
store.batchInsert(all)
val filter = Filter(kinds = listOf(7), tags = mapOf("e" to notes.take(3)), limit = 15)
val decoded = store.query<Event>(filter).map { it.id }
val raw = store.store.rawQuery(filter).map { it.id }
assertEquals(decoded, raw, "the zero-decode raw path must match the decoded query")
store.close()
}
@Test
fun tiedCreatedAt_matchesReferenceAsSet() =
runBlocking {
val store = newStore()
val notes = (0 until 4).map { hex64(10, it) }
val all = ArrayList<Event>()
// Many events share a created_at — the tag cursor can't id-order
// within a second, so assert a valid newest-N by set + size.
var t = 1_700_000_000L
for (block in 0 until 15) {
val ts = t
for (n in notes) all.add(ev(ts, 7, listOf(n)))
t += 1
}
store.batchInsert(all)
val filter = Filter(kinds = listOf(7), tags = mapOf("e" to notes), limit = 22)
assertTrue(mergeEligible(store, filter))
val merged = store.query<Event>(filter).map { it.id }
assertEquals(22, merged.size)
assertEquals(merged.size, merged.toSet().size)
// The whole result must sit within the newest slice the reference
// would return once ties are resolved either way: everything in
// `merged` must be at or above the created_at cutoff.
val ids = merged.toSet()
val chosen = all.filter { it.id in ids }
val cutoff = chosen.minOf { it.createdAt }
val eligibleAboveCutoff = all.filter { it.createdAt > cutoff }.map { it.id }.toSet()
assertTrue(eligibleAboveCutoff.all { it in ids }, "every event newer than the cutoff must be included")
store.close()
}
@Test
fun ineligibleShapesFallThrough() =
runBlocking {
val store = newStore()
store.batchInsert(listOf(ev(1_700_000_000L, 7, listOf(hex64(2, 0)))))
// Single value → single seek, not a merge.
assertFalse(mergeEligible(store, Filter(kinds = listOf(7), tags = mapOf("e" to listOf(hex64(2, 0))), limit = 10)))
// No limit → not merge-eligible.
assertFalse(mergeEligible(store, Filter(kinds = listOf(7), tags = mapOf("e" to listOf(hex64(2, 0), hex64(2, 1))))))
// Authors present → covered-index seek shape, not a tag merge.
assertFalse(
mergeEligible(
store,
Filter(kinds = listOf(7), authors = listOf(hex64(1, 1)), tags = mapOf("e" to listOf(hex64(2, 0), hex64(2, 1))), limit = 10),
),
)
store.close()
}
}

2
quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/FollowFeedReadBenchmark.kt
View File

@@ -202,7 +202,7 @@ class FollowFeedReadBenchmark {
runBlocking {
store.store.pool.useReader { c ->
var n = 0
MergeQueryExecutor.run(c, filter, store.store.queryBuilder.indexStrategy) { n++ }
MergeQueryExecutor.run(c, filter, store.store.queryBuilder.indexStrategy, store.store.seedModule::hasher) { n++ }
n
}
}

185
quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/FtsSearchScalingBenchmark.kt Normal file
View File

@@ -0,0 +1,185 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.relay.prodbench
import androidx.sqlite.SQLiteConnection
import androidx.sqlite.driver.bundled.BundledSQLiteDriver
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.store.sqlite.DefaultIndexingStrategy
import com.vitorpamplona.quartz.nip01Core.store.sqlite.EventStore
import com.vitorpamplona.quartz.utils.EventFactory
import kotlinx.coroutines.runBlocking
import kotlin.test.Test
/**
* Two measurements around the contentless FTS index, motivating the v4→v5
* schema change and being honest about what it does and does not fix.
*
* 1. **Delete scaling — the win.** The old `fts5(event_header_row_id,
* content)` schema had the `fts_foreign_key` trigger delete by a regular
* FTS column, which FTS5 cannot seek (it scans, O(n) per delete). The
* contentless schema keys deletes off the rowid (= event_headers.row_id),
* an O(log n) primary-key seek. Every event removal fires this trigger
* (replaceable rotation, kind-5, expiration, right-to-vanish).
* 2. **Search scaling — the limit.** `MATCH … ORDER BY rank LIMIT n` (NIP-50
* relevance ordering, bm25) must score *every* matching document, so
* search cost grows with the match set regardless of ordering (created_at
* has the same shape). Segment `optimize` compacts the index but does not
* change that; corpus-independent search needs an external engine. Shown
* fragmented vs optimized to size the (secondary) compaction effect.
*
* Size search with `-DftsBenchScale=N` (default 1). Not an assertion test.
*/
class FtsSearchScalingBenchmark {
companion object {
val SCALE = System.getProperty("ftsBenchScale")?.toInt() ?: 1
val SIZES = listOf(50_000, 100_000, 200_000).map { it * SCALE }
const val NEEDLE = "zzneedle"
}
private val hex = "0123456789abcdef"
private fun mix(seed: Long): Long {
var z = seed + -0x61c8864680b583ebL
z = (z xor (z ushr 30)) * -0x40a7b892e31b1a47L
z = (z xor (z ushr 27)) * -0x6b2fb644ecceee15L
return z xor (z ushr 31)
}
private fun hex64(index: Int): String {
val out = CharArray(64)
for (w in 0 until 4) {
val v = mix(index.toLong() * 4 + w + 99)
for (b in 0 until 8) {
val byte = ((v ushr (b * 8)) and 0xFF).toInt()
out[(w * 8 + b) * 2] = hex[byte ushr 4]
out[(w * 8 + b) * 2 + 1] = hex[byte and 0xF]
}
}
return String(out)
}
private val vocab = (0 until 400).map { "word$it" }
private val sig = "0".repeat(128)
private fun seed(n: Int): List<Event> {
val base = 1_700_000_000L
val events = ArrayList<Event>(n)
for (i in 0 until n) {
val r = mix(i.toLong())
val content =
buildString {
for (w in 0 until 8) append(vocab[((r ushr (w * 3)) and 0x1FF).toInt() % vocab.size]).append(' ')
// ~1% carry the searched term.
if (i % 100 == 0) append(NEEDLE)
}
events.add(EventFactory.create(hex64(i), hex64(i % 5000), base + i.toLong(), 1, emptyArray(), content, sig))
}
return events
}
private inline fun timeMs(block: () -> Unit): Double {
val start = System.nanoTime()
block()
return (System.nanoTime() - start) / 1e6
}
private fun SQLiteConnection.exec(sql: String) = prepare(sql).use { it.step() }
@Test
fun deleteByColumnVsByRowid() {
// Old-schema (delete by FTS column) vs contentless (delete by rowid),
// 500 deletes at two table sizes. By-column should grow with the
// table; by-rowid should stay flat.
println("─ FtsSearchScalingBenchmark.delete (500 deletes) ─")
println(" %-9s %14s %14s".format("rows", "byColumn", "byRowid"))
for (n in listOf(2_000 * SCALE, 8_000 * SCALE)) {
val db = BundledSQLiteDriver().open(":memory:")
try {
db.exec("CREATE VIRTUAL TABLE col USING fts5(event_header_row_id, content)")
db.exec("CREATE VIRTUAL TABLE row USING fts5(content, content='', contentless_delete=1)")
for (i in 1..n) {
db.prepare("INSERT INTO col(event_header_row_id, content) VALUES (?, 'alpha beta gamma')").use {
it.bindLong(1, i.toLong())
it.step()
}
db.prepare("INSERT INTO row(rowid, content) VALUES (?, 'alpha beta gamma')").use {
it.bindLong(1, i.toLong())
it.step()
}
}
val byColumn =
timeMs {
for (i in 1..500) {
db.prepare("DELETE FROM col WHERE event_header_row_id = ?").use {
it.bindLong(1, i.toLong())
it.step()
}
}
}
val byRowid =
timeMs {
for (i in 1..500) {
db.prepare("DELETE FROM row WHERE rowid = ?").use {
it.bindLong(1, i.toLong())
it.step()
}
}
}
println(" %-9s %11.2f ms %11.2f ms".format("${n / 1000}k", byColumn, byRowid))
} finally {
db.close()
}
}
}
@Test
fun searchScaling() =
runBlocking {
println("─ FtsSearchScalingBenchmark.search (created_at DESC, limit=50) ─")
println(" %-9s %16s %16s".format("corpus", "fragmented", "optimized"))
for (size in SIZES) {
val store = EventStore(dbName = null, indexStrategy = DefaultIndexingStrategy())
try {
seed(size).chunked(10_000).forEach { store.batchInsert(it) }
val f = Filter(search = NEEDLE, limit = 50)
val frag = time(store, f)
store.store.reindexFullTextSearch() // rebuild + optimize()
val opt = time(store, f)
println(" %-9s %13.2f ms %13.2f ms".format("${size / 1000}k", frag, opt))
} finally {
store.close()
}
}
}
private suspend fun time(
store: EventStore,
filter: Filter,
): Double {
repeat(3) { store.query<Event>(filter) }
val runs = 20
val start = System.nanoTime()
repeat(runs) { store.query<Event>(filter) }
return (System.nanoTime() - start) / 1e6 / runs
}
}

132
quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/ContentlessFtsMigrationTest.kt Normal file
View File

@@ -0,0 +1,132 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.store.sqlite
import androidx.sqlite.SQLiteConnection
import androidx.sqlite.driver.bundled.BundledSQLiteDriver
import com.vitorpamplona.quartz.nip01Core.core.Event
import com.vitorpamplona.quartz.nip01Core.relay.filters.Filter
import com.vitorpamplona.quartz.nip01Core.signers.NostrSignerSync
import com.vitorpamplona.quartz.nip10Notes.TextNoteEvent
import com.vitorpamplona.quartz.utils.Secp256k1Instance
import kotlinx.coroutines.runBlocking
import java.nio.file.Files
import java.nio.file.Path
import kotlin.io.path.deleteIfExists
import kotlin.test.AfterTest
import kotlin.test.BeforeTest
import kotlin.test.Test
import kotlin.test.assertEquals
import kotlin.test.assertTrue
/**
* Verifies the real v4 → v5 FTS upgrade path: a database written with the old
* `fts5(event_header_row_id, content)` index (auto-assigned rowid, content
* stored) must, on next open, drop that table and rebuild a contentless index
* keyed by `event_headers.row_id`, with search intact.
*
* The v4 state is fabricated by opening a fresh v5 store, then rewriting its
* `event_fts` to the old schema (with deliberately wrong content, to prove the
* rebuild wipes it) and stamping `user_version = 4`.
*/
class ContentlessFtsMigrationTest {
private val signer = NostrSignerSync()
private lateinit var dbFile: Path
private fun path() = dbFile.toAbsolutePath().toString()
@BeforeTest
fun setup() {
Secp256k1Instance
dbFile = Files.createTempFile("contentless-fts-migration-", ".db")
Files.deleteIfExists(dbFile)
}
@AfterTest
fun tearDown() {
listOf("", "-wal", "-shm", "-journal").forEach { Path.of(dbFile.toString() + it).deleteIfExists() }
}
@Test
fun upgradesOldFtsSchemaAndRebuildsSearch() =
runBlocking {
val alpha = signer.sign(TextNoteEvent.build("uniqalpha searchable body", createdAt = 1_700_000_000L))
val beta = signer.sign(TextNoteEvent.build("uniqbeta searchable body", createdAt = 1_700_000_100L))
// 1. Fresh v5 store, seed events, close.
EventStore(dbName = path(), relay = null).also {
it.insert(alpha)
it.insert(beta)
it.close()
}
// 2. Rewrite event_fts to the pre-v5 schema with WRONG content and
// downgrade user_version to 4 — the state a v4 database is in.
BundledSQLiteDriver().open(path()).use { db ->
db.exec("DROP TRIGGER IF EXISTS fts_foreign_key")
db.exec("DROP TABLE IF EXISTS event_fts")
db.exec("CREATE VIRTUAL TABLE event_fts USING fts5(event_header_row_id, content)")
db.exec(
"""
CREATE TRIGGER fts_foreign_key AFTER DELETE ON event_headers FOR EACH ROW
BEGIN DELETE FROM event_fts WHERE old.row_id = event_fts.event_header_row_id; END
""".trimIndent(),
)
// Stale/garbage rows: a real v4 index would hold correct data,
// but seeding garbage proves the migration rebuilds from
// event_headers rather than trusting the old table.
db.exec("INSERT INTO event_fts(event_header_row_id, content) VALUES (1, 'uniqstale garbage')")
db.exec("PRAGMA user_version = 4")
}
// 3. Reopen with current code → onUpgrade(4→5) → migrateV4ToContentless.
val store = EventStore(dbName = path(), relay = null)
try {
// Rebuilt from event_headers: real content is searchable...
assertEquals(alpha.id, store.query<Event>(Filter(search = "uniqalpha")).single().id)
assertEquals(beta.id, store.query<Event>(Filter(search = "uniqbeta")).single().id)
// ...and the old garbage is gone.
assertTrue(store.query<Event>(Filter(search = "uniqstale")).isEmpty())
// The new rowid IS event_headers.row_id: the join returns the
// right event for each FTS rowid.
store.store.pool.useReader { db ->
db
.prepare(
"SELECT h.id FROM event_fts f JOIN event_headers h ON h.row_id = f.rowid ORDER BY f.rowid",
).use { stmt ->
val ids = ArrayList<String>()
while (stmt.step()) ids.add(stmt.getText(0))
assertEquals(listOf(alpha.id, beta.id), ids, "FTS rowid must map to event_headers.row_id")
}
}
// The rebuilt delete trigger still cleans up FTS on delete.
store.store.delete(beta.id)
assertTrue(store.query<Event>(Filter(search = "uniqbeta")).isEmpty())
assertEquals(alpha.id, store.query<Event>(Filter(search = "uniqalpha")).single().id)
} finally {
store.close()
}
}
private fun SQLiteConnection.exec(sql: String) = prepare(sql).use { it.step() }
}

123
quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/Fts5CapabilityProbe.kt Normal file
View File

@@ -0,0 +1,123 @@
/*
* Copyright (c) 2025 Vitor Pamplona
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of
* this software and associated documentation files (the "Software"), to deal in
* the Software without restriction, including without limitation the rights to use,
* copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the
* Software, and to permit persons to whom the Software is furnished to do so,
* subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all
* copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
* FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
* COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN
* AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package com.vitorpamplona.quartz.nip01Core.store.sqlite
import androidx.sqlite.driver.bundled.BundledSQLiteDriver
import kotlin.test.Test
import kotlin.test.assertEquals
/**
* Pins the FTS5 features the [FullTextSearchModule] contentless index depends
* on to the bundled SQLite. All three shipped between SQLite 3.43 and 3.44
* (2023); if the bundled driver is ever downgraded below that, this fails
* loudly instead of the store silently breaking search on delete.
*
* - `content=''` **contentless** table with `contentless_delete=1`: lets the
* index drop the duplicated content column yet still delete rows by rowid
* (the `fts_foreign_key` trigger needs it).
* - explicit `rowid` on insert (= `event_headers.row_id`): the join key and
* the O(log n) delete key.
* - bm25 `rank` on a contentless table, through a join: NIP-50 relevance order.
* - `'merge'` / `'optimize'` maintenance commands: segment compaction.
*/
class Fts5CapabilityProbe {
@Test
fun contentlessDeleteAndRowidOrderingAreSupported() {
val db = BundledSQLiteDriver().open(":memory:")
try {
db.execSQL("CREATE VIRTUAL TABLE cl USING fts5(content, content='', contentless_delete=1)")
db.execSQL("INSERT INTO cl(rowid, content) VALUES (100, 'hello world')")
db.execSQL("INSERT INTO cl(rowid, content) VALUES (50, 'hello there')")
db.execSQL("INSERT INTO cl(rowid, content) VALUES (200, 'hello again')")
db.execSQL("DELETE FROM cl WHERE rowid = 50")
// Deleting an absent rowid must be a harmless no-op: the store's
// fts_foreign_key trigger fires on EVERY event_headers delete, but
// only searchable events ever got an FTS row.
db.execSQL("DELETE FROM cl WHERE rowid = 999999")
val order = ArrayList<Long>()
db.prepare("SELECT rowid FROM cl WHERE cl MATCH 'hello' ORDER BY rowid DESC LIMIT 5").use {
while (it.step()) order.add(it.getLong(0))
}
// Deleted 50 is gone; the rest come back newest-rowid first.
assertEquals(listOf(200L, 100L), order)
} finally {
db.close()
}
}
@Test
fun bm25RankWorksOnContentlessTableInAJoin() {
// NIP-50 orders by relevance, not created_at. Verify FTS5 bm25 `rank`
// works on a contentless table and is reachable through the same
// join-back-to-base-table shape the store's search query uses.
val db = BundledSQLiteDriver().open(":memory:")
try {
db.execSQL("CREATE TABLE headers (row_id INTEGER PRIMARY KEY, created_at INTEGER, tag TEXT)")
db.execSQL("CREATE VIRTUAL TABLE fts USING fts5(content, content='', contentless_delete=1)")
// row 10: term appears 3× in a short doc (most relevant).
// row 20: term once in a long doc (least relevant) but NEWER.
db.execSQL("INSERT INTO headers VALUES (10, 100, 'A')")
db.execSQL("INSERT INTO fts(rowid, content) VALUES (10, 'needle needle needle')")
db.execSQL("INSERT INTO headers VALUES (20, 999, 'B')")
db.execSQL("INSERT INTO fts(rowid, content) VALUES (20, 'needle alpha beta gamma delta epsilon zeta eta')")
// created_at DESC would return B (999) first; relevance returns A.
val byRank = ArrayList<String>()
db
.prepare(
"""
SELECT headers.tag FROM headers
INNER JOIN fts ON headers.row_id = fts.rowid
WHERE fts MATCH 'needle'
ORDER BY fts.rank
LIMIT 10
""".trimIndent(),
).use { while (it.step()) byRank.add(it.getText(0)) }
assertEquals(listOf("A", "B"), byRank, "bm25 rank must put the more relevant (shorter, higher-tf) doc first")
} finally {
db.close()
}
}
@Test
fun segmentMergeAndOptimizeAreSupported() {
val db = BundledSQLiteDriver().open(":memory:")
try {
db.execSQL("CREATE VIRTUAL TABLE m USING fts5(content)")
db.execSQL("INSERT INTO m(rowid, content) VALUES (1, 'a b c')")
db.execSQL("INSERT INTO m(rowid, content) VALUES (2, 'd e f')")
// Bounded incremental merge, then a full optimize — both must parse
// and run without error on the bundled build.
db.execSQL("INSERT INTO m(m, rank) VALUES ('merge', -16)")
db.execSQL("INSERT INTO m(m) VALUES ('optimize')")
val hits = ArrayList<Long>()
db.prepare("SELECT rowid FROM m WHERE m MATCH 'e' ORDER BY rowid").use {
while (it.step()) hits.add(it.getLong(0))
}
assertEquals(listOf(2L), hits)
} finally {
db.close()
}
}
}