diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt new file mode 100644 index 0000000000..c33a40b9f6 --- /dev/null +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryExecutor.kt @@ -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 { + val authors = filter.authors!! + val kinds = filter.kinds?.takeIf { it.isNotEmpty() } + val since = filter.since + val until = filter.until + + val stmts = ArrayList((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(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() + } + } +} diff --git a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt index fb723bced0..46ca1963ce 100644 --- a/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt +++ b/quartz/src/commonMain/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/QueryBuilder.kt @@ -44,24 +44,52 @@ class QueryBuilder( fun query( filter: Filter, db: SQLiteConnection, - ): List = db.runQuery(toSql(filter, hasher(db))) + ): List { + val merge = filter.toFilterWithDTags() + if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) { + val out = ArrayList(merge.limit!!) + MergeQueryExecutor.run(db, merge) { out.add(it.toEvent()) } + return out + } + return db.runQuery(toSql(filter, hasher(db))) + } fun 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 query( filters: List, db: SQLiteConnection, - ): List = db.runQuery(toSql(filters, hasher(db))) + ): List = + if (filters.size == 1) { + query(filters[0], db) + } else { + db.runQuery(toSql(filters, hasher(db))) + } fun query( filters: List, 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 = db.runRawQuery(toSql(filter, hasher(db))) + ): List { + val merge = filter.toFilterWithDTags() + if (MergeQueryExecutor.streamCount(merge, indexStrategy) > 0) { + val out = ArrayList(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, db: SQLiteConnection, - ): List = db.runRawQuery(toSql(filters, hasher(db))) + ): List = + if (filters.size == 1) { + rawQuery(filters[0], db) + } else { + db.runRawQuery(toSql(filters, hasher(db))) + } fun rawQuery( filters: List, 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 diff --git a/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryCorrectnessTest.kt b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryCorrectnessTest.kt new file mode 100644 index 0000000000..794911a50e --- /dev/null +++ b/quartz/src/commonTest/kotlin/com/vitorpamplona/quartz/nip01Core/store/sqlite/MergeQueryCorrectnessTest.kt @@ -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 { a, b -> + if (a.createdAt != b.createdAt) { + b.createdAt.compareTo(a.createdAt) + } else { + a.id.compareTo(b.id) + } + } + + private fun reference( + all: List, + filter: Filter, + limit: Int, + ): List = + 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 = + 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() + 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() + 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(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() + 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(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() + 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(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() + // 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(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() + 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(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() + 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(filter).map { it.id } + + val streamedIds = ArrayList() + store.query(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() + 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(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() + 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(listOf(filter)).map { it.id } + val single = store.query(filter).map { it.id } + + assertEquals(single, listOfOne, "single-element filter list must match the single-filter path") + store.close() + } +} diff --git a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/FollowFeedReadBenchmark.kt b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/FollowFeedReadBenchmark.kt index 4a54e5242e..ec37154f1e 100644 --- a/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/FollowFeedReadBenchmark.kt +++ b/quartz/src/jvmTest/kotlin/com/vitorpamplona/quartz/nip01Core/relay/prodbench/FollowFeedReadBenchmark.kt @@ -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, + ): Pair { + 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,