perf(store): k-way merge for the home-feed REQ shape

The home-feed REQ (`authors=[…] (+ kinds=[…]) [+ since/until] limit=N`, newest-first) is one of the most common relay queries. SQLite serves it 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) — on a cold on-disk 1M corpus that was the `follow-feed` regression (relayBench: 97.7 ms vs strfry 17.6 ms). Add `MergeQueryExecutor`, an app-level k-way merge that opens one lazy newest-first cursor per stream off the existing composite indexes (`query_by_kind_pubkey_created`, or `query_by_pubkey_created` for authors-only), merges their heads `(created_at DESC, id ASC)` and stops at the limit — reading only O(limit + streams) rows regardless of how much history the authors have. It reuses indexes that already exist, so write throughput and on-disk size are untouched. Eligibility is narrow (2..2048 streams, simple filter, explicit limit, no ids/d-tags); everything else falls through to the single-SQL plan. Wired into both `query` and the zero-decode `rawQuery` paths (the relay REQ hot path) and the single-element filter-list variants, so `LiveEventStore` REQs go through it. `MergeQueryCorrectnessTest` proves the merge returns exactly the single-SQL top-N — vs an independent Kotlin reference and vs the SQL path — across distinct/tied created_at, since/until windows, authors-only, fewer-than-limit, streaming onEach, and the raw path. `FollowFeedReadBenchmark` gains a `merge` variant: at 1.05M events it's flat ~10-12 ms across both prolific-recent and sparse-old, where `scan` is catastrophic on sparse follows (1995 ms) and `current` is disk-bound on prolific ones. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012EZeWww5TJnzBZKPoc6mvU
2026-08-09 08:04:45 +00:00 · 2026-07-04 16:49:13 +00:00
parent f29196f0fc
commit 8d09671218
4 changed files with 628 additions and 8 deletions
@@ -0,0 +1,199 @@
+/*
+ * 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.SQLiteStatement
+
+/**
+ * k-way merge executor for the **home-feed** query shape:
+ * `authors=[…] (+ kinds=[…]) [+ since/until] limit=N` ordered newest-first.
+ *
+ * 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`.
+ *
+ * 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).
+ *
+ * Merge order is `(created_at DESC, id ASC)`: a deterministic, correct
+ * top-N. NIP-01 leaves same-`created_at` ties unspecified, so this only
+ * pins down (and makes reproducible) which events sit exactly at a
+ * same-second boundary — the returned set is a valid newest-N either way.
+ */
+internal object MergeQueryExecutor {
+    const val COLS = "id, pubkey, created_at, kind, tags, content, 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`.
+     */
+    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).
+     */
+    fun streamCount(
+        filter: QueryBuilder.FilterWithDTags,
+        indexStrategy: IndexingStrategy,
+    ): Int {
+        if (!filter.isSimpleQuery()) return -1
+        if (filter.ids != null) return -1
+        if (filter.dTags != null) return -1
+        if (filter.limit == null || filter.limit <= 0) return -1
+        val authors = filter.authors ?: return -1
+        if (authors.isEmpty()) return -1
+        val kinds = filter.kinds
+        val streams =
+            if (kinds != null && kinds.isNotEmpty()) {
+                authors.size * kinds.size
+            } else {
+                // authors-only needs the (pubkey, created_at) index to stream.
+                if (!indexStrategy.indexEventsByPubkeyAlone) return -1
+                authors.size
+            }
+        // 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
+    }
+
+    /** Prepares one bound, newest-first cursor per stream. */
+    private fun prepareStreams(
+        db: SQLiteConnection,
+        filter: QueryBuilder.FilterWithDTags,
+    ): List<SQLiteStatement> {
+        val authors = filter.authors!!
+        val kinds = filter.kinds?.takeIf { it.isNotEmpty() }
+        val since = filter.since
+        val until = filter.until
+
+        val stmts = ArrayList<SQLiteStatement>((kinds?.size ?: 1) * authors.size)
+        if (kinds != null) {
+            val sql =
+                buildString {
+                    append("SELECT ").append(COLS)
+                    append(" FROM event_headers INDEXED BY query_by_kind_pubkey_created")
+                    append(" WHERE kind = ? AND pubkey = ?")
+                    if (until != null) append(" AND created_at <= ?")
+                    if (since != null) append(" AND created_at >= ?")
+                    append(" ORDER BY created_at DESC")
+                }
+            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)
+                }
+            }
+        } else {
+            val sql =
+                buildString {
+                    append("SELECT ").append(COLS)
+                    append(" FROM event_headers INDEXED BY query_by_pubkey_created")
+                    append(" WHERE pubkey = ?")
+                    if (until != null) append(" AND created_at <= ?")
+                    if (since != null) append(" AND created_at >= ?")
+                    append(" ORDER BY created_at DESC")
+                }
+            for (author in authors) {
+                val stmt = db.prepare(sql)
+                var p = 1
+                stmt.bindText(p++, author)
+                if (until != null) stmt.bindLong(p++, until)
+                if (since != null) stmt.bindLong(p++, since)
+                stmts.add(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).
+     */
+    fun run(
+        db: SQLiteConnection,
+        filter: QueryBuilder.FilterWithDTags,
+        onRow: (SQLiteStatement) -> Unit,
+    ) {
+        val stmts = prepareStreams(db, filter)
+        try {
+            val k = stmts.size
+            val headCreatedAt = LongArray(k)
+            val headId = arrayOfNulls<String>(k)
+            val live = BooleanArray(k)
+
+            // Position each cursor on its newest row.
+            for (i in 0 until k) {
+                if (stmts[i].step()) {
+                    headId[i] = stmts[i].getText(0)
+                    headCreatedAt[i] = stmts[i].getLong(2)
+                    live[i] = true
+                }
+            }
+
+            var emitted = 0
+            val limit = filter.limit!!
+            while (emitted < limit) {
+                // Pick the newest live head: created_at DESC, then id ASC.
+                var best = -1
+                for (i in 0 until k) {
+                    if (!live[i]) continue
+                    if (best == -1 ||
+                        headCreatedAt[i] > headCreatedAt[best] ||
+                        (headCreatedAt[i] == headCreatedAt[best] && headId[i]!! < headId[best]!!)
+                    ) {
+                        best = i
+                    }
+                }
+                if (best == -1) break
+
+                onRow(stmts[best]) // cursor is still on the head row
+                emitted++
+
+                // Advance the winner to its next row.
+                if (stmts[best].step()) {
+                    headId[best] = stmts[best].getText(0)
+                    headCreatedAt[best] = stmts[best].getLong(2)
+                } else {
+                    live[best] = false
+                }
+            }
+        } finally {
+            for (s in stmts) s.close()
+        }
+    }
+}
@@ -44,24 +44,52 @@ class QueryBuilder(
    fun <T : Event> query(
        filter: Filter,
        db: SQLiteConnection,
-    ): List<T> = db.runQuery(toSql(filter, hasher(db)))
+    ): List<T> {
+        val merge = filter.toFilterWithDTags()
+        if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
+            val out = ArrayList<T>(merge.limit!!)
+            MergeQueryExecutor.run(db, merge) { out.add(it.toEvent()) }
+            return out
+        }
+        return db.runQuery(toSql(filter, hasher(db)))
+    }

    fun <T : Event> query(
        filter: Filter,
        db: SQLiteConnection,
        onEach: (T) -> Unit,
-    ) = db.runQuery(toSql(filter, hasher(db)), onEach)
+    ) {
+        val merge = filter.toFilterWithDTags()
+        if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
+            MergeQueryExecutor.run(db, merge) { onEach(it.toEvent()) }
+            return
+        }
+        db.runQuery(toSql(filter, hasher(db)), onEach)
+    }

+    // A single-filter list is the home-feed REQ shape — route it through the
+    // single-filter path so the k-way merge (MergeQueryExecutor) applies.
    fun <T : Event> query(
        filters: List<Filter>,
        db: SQLiteConnection,
-    ): List<T> = db.runQuery(toSql(filters, hasher(db)))
+    ): List<T> =
+        if (filters.size == 1) {
+            query(filters[0], db)
+        } else {
+            db.runQuery(toSql(filters, hasher(db)))
+        }

    fun <T : Event> query(
        filters: List<Filter>,
        db: SQLiteConnection,
        onEach: (T) -> Unit,
-    ) = db.runQuery(toSql(filters, hasher(db)), onEach)
+    ) {
+        if (filters.size == 1) {
+            query(filters[0], db, onEach)
+        } else {
+            db.runQuery(toSql(filters, hasher(db)), onEach)
+        }
+    }

    // ---------------------------
    // Raw methods for performance
@@ -69,24 +97,50 @@ class QueryBuilder(
    fun rawQuery(
        filter: Filter,
        db: SQLiteConnection,
-    ): List<RawEvent> = db.runRawQuery(toSql(filter, hasher(db)))
+    ): List<RawEvent> {
+        val merge = filter.toFilterWithDTags()
+        if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
+            val out = ArrayList<RawEvent>(merge.limit!!)
+            MergeQueryExecutor.run(db, merge) { out.add(it.toRawEvent()) }
+            return out
+        }
+        return db.runRawQuery(toSql(filter, hasher(db)))
+    }

    fun rawQuery(
        filter: Filter,
        db: SQLiteConnection,
        onEach: (RawEvent) -> Unit,
-    ) = db.runRawQuery(toSql(filter, hasher(db)), onEach)
+    ) {
+        val merge = filter.toFilterWithDTags()
+        if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) {
+            MergeQueryExecutor.run(db, merge) { onEach(it.toRawEvent()) }
+            return
+        }
+        db.runRawQuery(toSql(filter, hasher(db)), onEach)
+    }

    fun rawQuery(
        filters: List<Filter>,
        db: SQLiteConnection,
-    ): List<RawEvent> = db.runRawQuery(toSql(filters, hasher(db)))
+    ): List<RawEvent> =
+        if (filters.size == 1) {
+            rawQuery(filters[0], db)
+        } else {
+            db.runRawQuery(toSql(filters, hasher(db)))
+        }

    fun rawQuery(
        filters: List<Filter>,
        db: SQLiteConnection,
        onEach: (RawEvent) -> Unit,
-    ) = db.runRawQuery(toSql(filters, hasher(db)), onEach)
+    ) {
+        if (filters.size == 1) {
+            rawQuery(filters[0], db, onEach)
+        } else {
+            db.runRawQuery(toSql(filters, hasher(db)), onEach)
+        }
+    }

    // -----------
    // Debug Tools
@@ -0,0 +1,330 @@
+/*
+ * 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.assertTrue
+
+/**
+ * Correctness guard for [MergeQueryExecutor]: the app-level k-way merge that
+ * serves the home-feed shape (`authors=[…] (+ kinds=[…]) [+ since/until]
+ * limit=N`, newest-first). For every eligible filter, the merge must return
+ * exactly the same top-N the single-SQL plan does.
+ *
+ * Two independent oracles per case:
+ *  - a **reference** computed in Kotlin (filter, sort `created_at DESC, id
+ *    ASC`, take limit) — the definition of a correct newest-N;
+ *  - the **SQL path** (`QueryBuilder.toSql` executed directly), which the
+ *    merge is replacing. With `useAndIndexIdOnOrderBy` on, both tie-break by
+ *    `id ASC`, so on distinct-and-tied `created_at` alike the two paths must
+ *    agree row-for-row.
+ */
+class MergeQueryCorrectnessTest {
+    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 String(out)
+    }
+
+    private val sig = "0".repeat(128)
+    private var idSeq = 0
+
+    private fun ev(
+        pubkey: String,
+        createdAt: Long,
+        kind: Int,
+    ): Event = EventFactory.create(hex64(7, idSeq++), pubkey, createdAt, kind, emptyArray(), "", sig)
+
+    /** newest-first, tie-break id ASC — the merge's contract. */
+    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>,
+        filter: Filter,
+        limit: Int,
+    ): List<String> =
+        all
+            .asSequence()
+            .filter { filter.authors == null || it.pubKey in filter.authors!! }
+            .filter { filter.kinds == null || it.kind in filter.kinds!! }
+            .filter { filter.since == null || it.createdAt >= filter.since!! }
+            .filter { filter.until == null || it.createdAt <= filter.until!! }
+            .sortedWith(newestFirst)
+            .take(limit)
+            .map { it.id }
+            .toList()
+
+    /** Runs the raw single-SQL plan (bypassing the merge) and returns its ids. */
+    private fun sqlIds(
+        store: EventStore,
+        filter: Filter,
+    ): List<String> =
+        runBlocking {
+            store.store.pool.useReader { conn ->
+                val hasher = store.store.seedModule.hasher(conn)
+                val spec = store.store.queryBuilder.toSql(filter, hasher)
+                conn.prepare(spec.sql).use { stmt ->
+                    spec.args.forEachIndexed { i, a -> stmt.bindText(i + 1, a) }
+                    val ids = ArrayList<String>()
+                    while (stmt.step()) ids.add(stmt.getText(0))
+                    ids
+                }
+            }
+        }
+
+    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(
+                    indexEventsByCreatedAtAlone = true,
+                    indexEventsByPubkeyAlone = true,
+                    useAndIndexIdOnOrderBy = true,
+                    indexFullTextSearch = false,
+                ),
+        )
+
+    @Test
+    fun distinctCreatedAt_multiKind() =
+        runBlocking {
+            val store = newStore()
+            val authors = (0 until 8).map { hex64(2, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            // Each event a distinct created_at: ordering is fully determined.
+            for (round in 0 until 40) {
+                for ((i, pk) in authors.withIndex()) {
+                    val kind = if (round % 3 == 0) 6 else 1
+                    all.add(ev(pk, t++, kind))
+                }
+            }
+            // Background noise from other authors/kinds the filter must exclude.
+            for (i in 0 until 200) all.add(ev(hex64(9, i), t++, if (i % 2 == 0) 1 else 7))
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(1, 6), authors = authors, limit = 50)
+            assertTrue(mergeEligible(store, filter), "home-feed filter must be merge-eligible")
+
+            val merged = store.query<Event>(filter).map { it.id }
+            assertEquals(reference(all, filter, 50), merged, "merge must equal the Kotlin reference")
+            assertEquals(sqlIds(store, filter), merged, "merge must equal the single-SQL plan")
+            store.close()
+        }
+
+    @Test
+    fun authorsOnly_noKinds() =
+        runBlocking {
+            val store = newStore()
+            val authors = (0 until 6).map { hex64(3, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            for (round in 0 until 30) {
+                for (pk in authors) all.add(ev(pk, t++, (round % 4) + 1))
+            }
+            for (i in 0 until 100) all.add(ev(hex64(8, i), t++, 1))
+            store.batchInsert(all)
+
+            val filter = Filter(authors = authors, limit = 40)
+            assertTrue(mergeEligible(store, filter), "authors-only filter must be merge-eligible")
+
+            val merged = store.query<Event>(filter).map { it.id }
+            assertEquals(reference(all, filter, 40), merged)
+            assertEquals(sqlIds(store, filter), merged)
+            store.close()
+        }
+
+    @Test
+    fun withSinceAndUntil() =
+        runBlocking {
+            val store = newStore()
+            val authors = (0 until 5).map { hex64(4, it) }
+            val all = ArrayList<Event>()
+            val base = 1_700_000_000L
+            for (i in 0 until 300) {
+                val pk = authors[i % authors.size]
+                all.add(ev(pk, base + i.toLong(), if (i % 5 == 0) 6 else 1))
+            }
+            store.batchInsert(all)
+
+            val since = base + 50
+            val until = base + 250
+            val filter = Filter(kinds = listOf(1, 6), authors = authors, since = since, until = until, limit = 500)
+            assertTrue(mergeEligible(store, filter))
+
+            val merged = store.query<Event>(filter).map { it.id }
+            val ref = reference(all, filter, 500)
+            assertEquals(ref, merged)
+            assertEquals(sqlIds(store, filter), merged)
+            // Sanity: the window really did bound the result.
+            assertTrue(merged.isNotEmpty() && ref.size < 300)
+            store.close()
+        }
+
+    @Test
+    fun tiedCreatedAt_setMatchesReferenceMultiset() =
+        runBlocking {
+            val store = newStore()
+            val authors = (0 until 6).map { hex64(5, it) }
+            val all = ArrayList<Event>()
+            // Many events share the same created_at across authors — exercises
+            // the same-second tie-break (id ASC) on both paths.
+            var t = 1_700_000_000L
+            for (block in 0 until 20) {
+                val ts = t
+                for (pk in authors) {
+                    all.add(ev(pk, ts, 1))
+                    all.add(ev(pk, ts, 6))
+                }
+                t += 1
+            }
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(1, 6), authors = authors, limit = 30)
+            assertTrue(mergeEligible(store, filter))
+
+            val merged = store.query<Event>(filter).map { it.id }
+            // With useAndIndexIdOnOrderBy both paths pin id ASC, so even under
+            // ties the ordered result is deterministic and must match exactly.
+            assertEquals(reference(all, filter, 30), merged)
+            assertEquals(sqlIds(store, filter), merged)
+            store.close()
+        }
+
+    @Test
+    fun fewerMatchesThanLimit() =
+        runBlocking {
+            val store = newStore()
+            val authors = (0 until 4).map { hex64(6, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            for (pk in authors) repeat(3) { all.add(ev(pk, t++, 1)) }
+            // Unrelated authors so the store isn't tiny.
+            for (i in 0 until 50) all.add(ev(hex64(1, i), t++, 1))
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(1), authors = authors, limit = 500)
+            assertTrue(mergeEligible(store, filter))
+
+            val merged = store.query<Event>(filter).map { it.id }
+            assertEquals(12, merged.size)
+            assertEquals(reference(all, filter, 500), merged)
+            assertEquals(sqlIds(store, filter), merged)
+            store.close()
+        }
+
+    @Test
+    fun onEachCallbackMatchesListQuery() =
+        runBlocking {
+            val store = newStore()
+            val authors = (0 until 5).map { hex64(2, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            for (round in 0 until 20) for (pk in authors) all.add(ev(pk, t++, 1))
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(1), authors = authors, limit = 25)
+            val listIds = store.query<Event>(filter).map { it.id }
+
+            val streamedIds = ArrayList<String>()
+            store.query<Event>(filter) { streamedIds.add(it.id) }
+
+            assertEquals(listIds, streamedIds, "streaming onEach must match the list query")
+            assertEquals(reference(all, filter, 25), listIds)
+            store.close()
+        }
+
+    @Test
+    fun rawQueryPathMatchesDecodedQuery() =
+        runBlocking {
+            val store = newStore()
+            val authors = (0 until 6).map { hex64(4, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            for (round in 0 until 25) for (pk in authors) all.add(ev(pk, t++, if (round % 4 == 0) 6 else 1))
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(1, 6), authors = authors, limit = 30)
+            val decoded = store.query<Event>(filter).map { it.id }
+            val raw = store.rawQuery(filter).map { it.id }
+
+            assertEquals(decoded, raw, "the zero-decode raw path must match the decoded query")
+            assertEquals(reference(all, filter, 30), raw)
+            store.close()
+        }
+
+    @Test
+    fun singleElementFilterListRoutesThroughMerge() =
+        runBlocking {
+            val store = newStore()
+            val authors = (0 until 5).map { hex64(3, it) }
+            val all = ArrayList<Event>()
+            var t = 1_700_000_000L
+            for (round in 0 until 20) for (pk in authors) all.add(ev(pk, t++, 1))
+            store.batchInsert(all)
+
+            val filter = Filter(kinds = listOf(1), authors = authors, limit = 25)
+            val listOfOne = store.query<Event>(listOf(filter)).map { it.id }
+            val single = store.query<Event>(filter).map { it.id }
+
+            assertEquals(single, listOfOne, "single-element filter list must match the single-filter path")
+            store.close()
+        }
+}
@@ -24,6 +24,8 @@ 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.nip01Core.store.sqlite.MergeQueryExecutor
+import com.vitorpamplona.quartz.nip01Core.store.sqlite.QueryBuilder
 import com.vitorpamplona.quartz.nip01Core.store.sqlite.explainQuery
 import com.vitorpamplona.quartz.utils.EventFactory
 import kotlinx.coroutines.runBlocking
@@ -173,10 +175,45 @@ class FollowFeedReadBenchmark {
                        println("    %-8s ERROR: %s".format(name, e.message?.take(80)))
                    }
                }
+                // The app-level k-way merge is not a SQL string, so time it directly.
+                try {
+                    val (n, ms) = timeMerge(store, authors)
+                    val streams = 2 * authors.size
+                    println("    %-8s %6.1f ms  (%d rows)  k-way merge, %d streams".format("merge", ms, n, streams))
+                } catch (e: Exception) {
+                    println("    %-8s ERROR: %s".format("merge", e.message?.take(80)))
+                }
            }
            store.close()
        }

+    private fun timeMerge(
+        store: EventStore,
+        authors: List<String>,
+    ): Pair<Int, Double> {
+        val filter =
+            QueryBuilder.FilterWithDTags(
+                authors = authors,
+                kinds = listOf(1, 6),
+                limit = 500,
+            )
+
+        fun run() =
+            runBlocking {
+                store.store.pool.useReader { c ->
+                    var n = 0
+                    MergeQueryExecutor.run(c, filter) { n++ }
+                    n
+                }
+            }
+        repeat(3) { run() }
+        val runs = 10
+        val start = System.nanoTime()
+        var got = 0
+        repeat(runs) { got = run() }
+        return got to (System.nanoTime() - start) / 1e6 / runs
+    }
+
    private fun timeRaw(
        store: EventStore,
        sql: String,