resolve queue rows with keyset cursor seeks instead of offsets

2026-11-10 15:11:58 +01:00 · 2026-10-04 00:30:01 +00:00
parent d710a3185c
commit 4aaeb4871b
7 changed files with 506 additions and 21 deletions
@@ -16,12 +16,15 @@ import com.subtracks.data.model.Song
 import com.subtracks.data.model.SongListItem
 import kotlinx.coroutines.flow.Flow

-internal const val PLAYLIST_SONGS_SQL =
-    "SELECT songs.*, albums.coverArt AS coverArt FROM playlist_songs " +
+internal const val PLAYLIST_SONGS_SELECT =
+    "SELECT songs.*, albums.coverArt AS coverArt, playlist_songs.position AS position FROM playlist_songs " +
        "JOIN songs ON songs.sourceId = playlist_songs.sourceId AND songs.id = playlist_songs.songId " +
        "LEFT JOIN albums ON albums.sourceId = songs.sourceId AND albums.id = songs.albumId " +
-        "WHERE playlist_songs.sourceId = :sourceId AND playlist_songs.playlistId = :playlistId " +
-        "ORDER BY playlist_songs.position"
+        "WHERE playlist_songs.sourceId = :sourceId AND playlist_songs.playlistId = :playlistId"
+
+internal const val PLAYLIST_SONGS_ORDER = " ORDER BY playlist_songs.position"
+
+internal const val PLAYLIST_SONGS_SQL = PLAYLIST_SONGS_SELECT + PLAYLIST_SONGS_ORDER

 internal const val ALBUMS_FILTER =
    "FROM albums WHERE sourceId = :sourceId " +
@@ -259,6 +259,16 @@ val MIGRATION_17_18 =
        }
    }

+val MIGRATION_18_19 =
+    object : Migration(18, 19) {
+        override suspend fun migrate(connection: SQLiteConnection) {
+            connection.execSQL(
+                "CREATE INDEX IF NOT EXISTS `index_songs_sourceId_albumId_order` ON `songs` " +
+                    "(`sourceId`, `albumId`, `disc`, `track`, `id`)",
+            )
+        }
+    }
+
 val MIGRATIONS: Array<Migration> =
    arrayOf(
        MIGRATION_1_2,
@@ -278,4 +288,5 @@ val MIGRATIONS: Array<Migration> =
        MIGRATION_15_16,
        MIGRATION_16_17,
        MIGRATION_17_18,
+        MIGRATION_18_19,
    )
@@ -9,14 +9,17 @@ import com.subtracks.data.model.QueueEntry
 import com.subtracks.data.model.SongListItem
 import com.subtracks.data.model.UpNextEntry

-private const val ALBUM_SONGS_SQL =
-    "SELECT songs.*, albums.coverArt AS coverArt FROM songs " +
+private const val ALBUM_SONGS_SELECT =
+    "SELECT songs.*, albums.coverArt AS coverArt, NULL AS position FROM songs " +
        "LEFT JOIN albums ON albums.sourceId = songs.sourceId AND albums.id = songs.albumId " +
-        "WHERE songs.sourceId = :sourceId AND songs.albumId = :albumId " +
-        "ORDER BY songs.disc, songs.track, songs.id"
+        "WHERE songs.sourceId = :sourceId AND songs.albumId = :albumId"
+
+private const val ALBUM_SONGS_ORDER = " ORDER BY songs.disc, songs.track, songs.id"
+
+private const val ALBUM_SONGS_SQL = ALBUM_SONGS_SELECT + ALBUM_SONGS_ORDER

 private const val SONG_SQL =
-    "SELECT songs.*, albums.coverArt AS coverArt FROM songs " +
+    "SELECT songs.*, albums.coverArt AS coverArt, NULL AS position FROM songs " +
        "LEFT JOIN albums ON albums.sourceId = songs.sourceId AND albums.id = songs.albumId " +
        "WHERE songs.sourceId = :sourceId AND songs.id = :songId"

@@ -76,6 +79,30 @@ interface QueueDao {
        limit: Int,
    ): List<SongListItem>

+    @Query(
+        "$PLAYLIST_SONGS_SELECT AND playlist_songs.position >= :position$PLAYLIST_SONGS_ORDER " +
+            "LIMIT :limit OFFSET :skip",
+    )
+    suspend fun playlistSongsFrom(
+        sourceId: Long,
+        playlistId: String,
+        position: Long,
+        skip: Long,
+        limit: Int,
+    ): List<SongListItem>
+
+    @Query(
+        "$PLAYLIST_SONGS_SELECT AND playlist_songs.position < :position " +
+            "ORDER BY playlist_songs.position DESC LIMIT :limit OFFSET :skip",
+    )
+    suspend fun playlistSongsBefore(
+        sourceId: Long,
+        playlistId: String,
+        position: Long,
+        skip: Long,
+        limit: Int,
+    ): List<SongListItem>
+
    @Query("$ALBUM_SONGS_SQL LIMIT :limit OFFSET :offset")
    suspend fun albumSongs(
        sourceId: Long,
@@ -84,6 +111,34 @@ interface QueueDao {
        limit: Int,
    ): List<SongListItem>

+    @Query(
+        "$ALBUM_SONGS_SELECT AND (songs.disc, songs.track, songs.id) >= (:disc, :track, :id)" +
+            "$ALBUM_SONGS_ORDER LIMIT :limit OFFSET :skip",
+    )
+    suspend fun albumSongsFrom(
+        sourceId: Long,
+        albumId: String,
+        disc: Long,
+        track: Long,
+        id: String,
+        skip: Long,
+        limit: Int,
+    ): List<SongListItem>
+
+    @Query(
+        "$ALBUM_SONGS_SELECT AND (songs.disc, songs.track, songs.id) < (:disc, :track, :id) " +
+            "ORDER BY songs.disc DESC, songs.track DESC, songs.id DESC LIMIT :limit OFFSET :skip",
+    )
+    suspend fun albumSongsBefore(
+        sourceId: Long,
+        albumId: String,
+        disc: Long,
+        track: Long,
+        id: String,
+        skip: Long,
+        limit: Int,
+    ): List<SongListItem>
+
    @Query("$SONG_SQL LIMIT :limit OFFSET :offset")
    suspend fun song(
        sourceId: Long,
@@ -113,7 +168,7 @@ interface QueueDao {
    ): List<String>

    @Query(
-        "SELECT songs.*, albums.coverArt AS coverArt FROM songs " +
+        "SELECT songs.*, albums.coverArt AS coverArt, NULL AS position FROM songs " +
            "LEFT JOIN albums ON albums.sourceId = songs.sourceId AND albums.id = songs.albumId " +
            "WHERE songs.sourceId = :sourceId AND songs.id IN (:ids)",
    )
@@ -35,7 +35,7 @@ import com.subtracks.data.model.UpNextEntry
        ArtworkSeed::class,
        SongDownload::class,
    ],
-    version = 18,
+    version = 19,
    exportSchema = true,
 )
@ColumnTypeConverters(QueueKindConverter::class, DownloadStatusConverter::class)
@@ -209,7 +209,14 @@ data class PlaylistSong(
            onDelete = ForeignKey.CASCADE,
        ),
    ],
-    indices = [Index("sourceId", "albumId"), Index("sourceId", "artistId")],
+    indices = [
+        Index("sourceId", "albumId"),
+        Index("sourceId", "artistId"),
+        Index(
+            name = "index_songs_sourceId_albumId_order",
+            value = ["sourceId", "albumId", "disc", "track", "id"],
+        ),
+    ],
 )
 data class Song(
    val sourceId: Long,
@@ -230,6 +237,7 @@ data class Song(
 data class SongListItem(
    @Embedded val song: Song,
    val coverArt: String?,
+    val position: Long? = null,
 )

 data class DiscKey(
@@ -36,6 +36,24 @@ private data class LoadedSnapshot(
    val anchor: Long,
 )

+private data class QueueRef(
+    val kind: QueueKind,
+    val sourceId: Long,
+    val refId: String,
+)
+
+private data class OrderKey(
+    val id: String,
+    val disc: Long? = null,
+    val track: Long? = null,
+    val position: Long? = null,
+)
+
+private data class KeyCursor(
+    val ordinal: Long,
+    val key: OrderKey,
+)
+
 data class ResolvedQueueEntry(
    val entry: QueueEntry,
    val length: Long,
@@ -206,6 +224,14 @@ class QueueRepository(

    private val lengthCache = ConcurrentHashMap<EntryLengthKey, Long>()

+    private val keyCache = ConcurrentHashMap<QueueRef, KeyCursor>()
+
+    internal var keysetSeeks = 0L
+        private set
+
+    internal var offsetSeeks = 0L
+        private set
+
    private val cursorMutex = Mutex()

    fun playlistEntry(
@@ -327,6 +353,7 @@ class QueueRepository(
    fun invalidateLibraryCache() {
        cachedFlatIds = null
        lengthCache.clear()
+        keyCache.clear()
        shuffleRemoved = null
        queueVersion.incrementAndGet()
    }
@@ -679,19 +706,99 @@ class QueueRepository(
        entry: QueueEntry,
        offset: Long,
        limit: Int,
+    ): List<SongListItem> {
+        if (entry.kind == QueueKind.Song) {
+            offsetSeeks++
+            return dao.song(entry.sourceId, entry.refId, offset, limit)
+        }
+        val ref = QueueRef(entry.kind, entry.sourceId, entry.refId)
+        val anchored = keyCache[ref]?.let { keyedRows(entry, offset, limit, it) }
+        if (anchored != null) {
+            keysetSeeks++
+            if (anchored.isNotEmpty()) remember(ref, offset, anchored)
+            return anchored
+        }
+        offsetSeeks++
+        val fresh =
+            when (entry.kind) {
+                QueueKind.Playlist -> dao.playlistSongs(entry.sourceId, entry.refId, offset, limit)
+                QueueKind.Album -> dao.albumSongs(entry.sourceId, entry.refId, offset, limit)
+                QueueKind.Song -> emptyList()
+            }
+        if (fresh.isNotEmpty()) remember(ref, offset, fresh)
+        return fresh
+    }
+
+    private fun remember(
+        ref: QueueRef,
+        offset: Long,
+        resolved: List<SongListItem>,
+    ) {
+        if (keyCache.size >= LENGTH_CACHE_LIMIT) keyCache.clear()
+        keyCache[ref] = KeyCursor(offset + resolved.size - 1, orderKey(ref.kind, resolved.last()))
+    }
+
+    private fun orderKey(
+        kind: QueueKind,
+        row: SongListItem,
+    ): OrderKey =
+        when (kind) {
+            QueueKind.Album -> OrderKey(row.song.id, disc = row.song.disc, track = row.song.track)
+            QueueKind.Playlist -> OrderKey(row.song.id, position = row.position)
+            QueueKind.Song -> OrderKey(row.song.id)
+        }
+
+    private suspend fun keyedRows(
+        entry: QueueEntry,
+        offset: Long,
+        limit: Int,
+        cursor: KeyCursor,
+    ): List<SongListItem>? {
+        if (!cursor.key.seekable(entry.kind)) return null
+        val from = cursor.ordinal
+        return if (offset >= from) {
+            rowsFrom(entry, cursor.key, offset - from, limit)
+        } else {
+            val end = offset + limit - 1
+            if (end < from) {
+                rowsBefore(entry, cursor.key, from - end - 1, limit).asReversed()
+            } else {
+                val head = rowsBefore(entry, cursor.key, 0, (from - offset).toInt()).asReversed()
+                val tail = rowsFrom(entry, cursor.key, 0, (end - from + 1).toInt())
+                head + tail
+            }
+        }
+    }
+
+    private suspend fun rowsFrom(
+        entry: QueueEntry,
+        key: OrderKey,
+        skip: Long,
+        limit: Int,
    ): List<SongListItem> =
        when (entry.kind) {
-            QueueKind.Playlist -> {
-                dao.playlistSongs(entry.sourceId, entry.refId, offset, limit)
-            }
+            QueueKind.Playlist -> dao.playlistSongsFrom(entry.sourceId, entry.refId, key.position!!, skip, limit)
+            QueueKind.Album -> dao.albumSongsFrom(entry.sourceId, entry.refId, key.disc!!, key.track!!, key.id, skip, limit)
+            QueueKind.Song -> emptyList()
+        }

-            QueueKind.Album -> {
-                dao.albumSongs(entry.sourceId, entry.refId, offset, limit)
-            }
+    private suspend fun rowsBefore(
+        entry: QueueEntry,
+        key: OrderKey,
+        skip: Long,
+        limit: Int,
+    ): List<SongListItem> =
+        when (entry.kind) {
+            QueueKind.Playlist -> dao.playlistSongsBefore(entry.sourceId, entry.refId, key.position!!, skip, limit)
+            QueueKind.Album -> dao.albumSongsBefore(entry.sourceId, entry.refId, key.disc!!, key.track!!, key.id, skip, limit)
+            QueueKind.Song -> emptyList()
+        }

-            QueueKind.Song -> {
-                dao.song(entry.sourceId, entry.refId, offset, limit)
-            }
+    private fun OrderKey.seekable(kind: QueueKind): Boolean =
+        when (kind) {
+            QueueKind.Playlist -> position != null
+            QueueKind.Album -> disc != null && track != null
+            QueueKind.Song -> false
        }

    private suspend fun lengthOf(
@@ -0,0 +1,301 @@
+package com.subtracks.data.repo
+
+import android.content.Context
+import androidx.room3.Room
+import androidx.room3.useReaderConnection
+import androidx.sqlite.driver.bundled.BundledSQLiteDriver
+import androidx.test.core.app.ApplicationProvider
+import androidx.test.ext.junit.runners.AndroidJUnit4
+import com.subtracks.data.db.SubtracksDatabase
+import com.subtracks.data.model.Album
+import com.subtracks.data.model.PlaylistSong
+import com.subtracks.data.model.Song
+import com.subtracks.data.model.Source
+import kotlinx.coroutines.test.runTest
+import org.junit.After
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import org.junit.runner.RunWith
+
+@RunWith(AndroidJUnit4::class)
+class QueueKeysetTest {
+    private lateinit var db: SubtracksDatabase
+    private lateinit var repository: QueueRepository
+
+    @Before
+    fun setUp() {
+        val context = ApplicationProvider.getApplicationContext<Context>()
+        db =
+            Room
+                .inMemoryDatabaseBuilder(context, SubtracksDatabase::class.java)
+                .setDriver(BundledSQLiteDriver())
+                .build()
+        repository = QueueRepository(db)
+    }
+
+    @After
+    fun tearDown() {
+        db.close()
+    }
+
+    @Test
+    fun albumKeysetFromMatchesOffsetWindowAtADeepPosition() =
+        runTest {
+            seedLibrary()
+            val all = allAlbumRows("big")
+            val anchor = all[200]
+
+            val expected = all.drop(200).take(50).map { it.song.id }
+            val actual =
+                db
+                    .queueDao()
+                    .albumSongsFrom(1, "big", anchor.song.disc!!, anchor.song.track!!, anchor.song.id, skip = 0, limit = 50)
+                    .map { it.song.id }
+
+            assertEquals(expected, actual)
+        }
+
+    @Test
+    fun albumKeysetBeforeMatchesOffsetWindowAtADeepPosition() =
+        runTest {
+            seedLibrary()
+            val all = allAlbumRows("big")
+            val anchor = all[200]
+
+            val expected = all.drop(150).take(50).map { it.song.id }
+            val actual =
+                db
+                    .queueDao()
+                    .albumSongsBefore(1, "big", anchor.song.disc!!, anchor.song.track!!, anchor.song.id, skip = 0, limit = 50)
+                    .asReversed()
+                    .map { it.song.id }
+
+            assertEquals(expected, actual)
+        }
+
+    @Test
+    fun albumKeysetRespectsDiscThenTrackThenId() =
+        runTest {
+            seedLibrary()
+            val all = allAlbumRows("big").map { it.song.id }
+
+            assertEquals(300, all.size)
+            assertEquals("b0", all.first())
+            assertEquals("b150", all[150])
+
+            val anchor = allAlbumRows("big")[150]
+            val actual =
+                db
+                    .queueDao()
+                    .albumSongsFrom(1, "big", anchor.song.disc!!, anchor.song.track!!, anchor.song.id, skip = 0, limit = 3)
+                    .map { it.song.id }
+            assertEquals(all.drop(150).take(3), actual)
+        }
+
+    @Test
+    fun albumKeysetBreaksTiesOnId() =
+        runTest {
+            seedLibrary()
+            val all = allAlbumRows("tie").map { it.song.id }
+
+            assertEquals(listOf("t1a", "t1z"), all)
+        }
+
+    @Test
+    fun playlistKeysetMatchesOffsetWindowWithGapsAndGhosts() =
+        runTest {
+            seedLibrary()
+            val all = allPlaylistRows().map { it.song.id }
+            assertTrue(all.size >= 12)
+
+            val anchor = allPlaylistRows()[10]
+            val expected = all.drop(10).take(5)
+            val actual =
+                db
+                    .queueDao()
+                    .playlistSongsFrom(1, "pl1", anchor.position!!, skip = 0, limit = 5)
+                    .map { it.song.id }
+            assertEquals(expected, actual)
+        }
+
+    @Test
+    fun scanningTheWholeAlbumThroughTheRepositoryUsesKeysetSeeks() =
+        runTest {
+            seedLibrary()
+            repository.replace(listOf(repository.albumEntry(1, "big")))
+            val snapshot = repository.snapshot()
+            val expected = allAlbumRows("big").map { it.song.id }
+
+            val resolved = (0 until snapshot.size).map { repository.itemAt(snapshot, it)?.song?.id }
+            assertEquals(expected, resolved)
+            assertTrue("expected keyset seeks, saw ${repository.keysetSeeks}", repository.keysetSeeks >= snapshot.size - 1)
+            assertEquals(1L, repository.offsetSeeks)
+        }
+
+    @Test
+    fun scanningBackwardsThroughTheRepositoryUsesKeysetSeeks() =
+        runTest {
+            seedLibrary()
+            repository.replace(listOf(repository.albumEntry(1, "big")))
+            val snapshot = repository.snapshot()
+            val expected = allAlbumRows("big").map { it.song.id }
+
+            val resolved =
+                (snapshot.size - 1 downTo 0).map { repository.itemAt(snapshot, it)?.song?.id }.asReversed()
+            assertEquals(expected, resolved)
+            assertTrue(repository.keysetSeeks >= snapshot.size - 1)
+        }
+
+    @Test
+    fun aWindowStraddlingTheCachedAnchorResolvesLikeOffset() =
+        runTest {
+            seedLibrary()
+            repository.replace(listOf(repository.albumEntry(1, "big")))
+            val snapshot = repository.snapshot()
+            repository.itemAt(snapshot, 200)
+
+            val range = repository.range(snapshot, first = 190, last = 210)
+
+            val expected = allAlbumRows("big").drop(190).take(21).map { it.song.id }
+            assertEquals(expected, range.map { it.item.song.id })
+            assertEquals((190L..210L).toList(), range.map { it.position })
+        }
+
+    @Test
+    fun aSplitRangeAfterRemovingAMiddleTrackMatchesTheOracle() =
+        runTest {
+            seedLibrary()
+            repository.replace(listOf(repository.albumEntry(1, "big")))
+            val before = repository.snapshot()
+            repository.removeAt(before, 100)
+
+            val after = repository.snapshot()
+            val resolved = (0 until after.size).map { repository.itemAt(after, it)?.song?.id }
+            val all = allAlbumRows("big").map { it.song.id }
+            val expected = all.filterIndexed { index, _ -> index != 100 }
+
+            assertEquals(expected, resolved)
+            assertTrue(repository.keysetSeeks > 0)
+        }
+
+    @Test
+    fun aShuffledWindowResolvesEveryTrackExactlyOnce() =
+        runTest {
+            seedLibrary()
+            repository.replace(listOf(repository.albumEntry(1, "big")))
+            repository.setShuffle(4242L)
+            val snapshot = repository.snapshot()
+
+            val window = repository.window(snapshot, center = snapshot.size / 2, radius = 20)
+            val ids = window.map { it.item.song.id }
+
+            assertEquals(window.size, ids.distinct().size)
+            val expected = allAlbumRows("big").map { it.song.id }
+            assertEquals(expected.size, snapshot.size.toInt())
+            assertTrue(ids.all { it in expected })
+        }
+
+    @Test
+    fun theAlbumKeysetQuerySeeksTheOrderIndex() =
+        runTest {
+            seedLibrary()
+            val plan =
+                queryPlan(
+                    "SELECT songs.*, albums.coverArt AS coverArt, NULL AS position FROM songs " +
+                        "LEFT JOIN albums ON albums.sourceId = songs.sourceId AND albums.id = songs.albumId " +
+                        "WHERE songs.sourceId = 1 AND songs.albumId = 'big' " +
+                        "AND (songs.disc, songs.track, songs.id) >= (1, 1, 'b0') " +
+                        "ORDER BY songs.disc, songs.track, songs.id LIMIT 50 OFFSET 0",
+                )
+
+            assertTrue(plan.joinToString("\n"), plan.any { it.contains("index_songs_sourceId_albumId_order") })
+        }
+
+    private suspend fun queryPlan(sql: String): List<String> =
+        db.useReaderConnection { transactor ->
+            transactor.usePrepared("EXPLAIN QUERY PLAN $sql") { statement ->
+                buildList {
+                    while (statement.step()) add(statement.getText(3))
+                }
+            }
+        }
+
+    private suspend fun allAlbumRows(albumId: String) = db.queueDao().albumSongs(1, albumId, 0, 1000)
+
+    private suspend fun allPlaylistRows() = db.queueDao().playlistSongs(1, "pl1", 0, 1000)
+
+    private suspend fun seedLibrary() {
+        db.sourcesDao().upsertSource(
+            Source(id = 1, name = "test", address = "http://localhost", isActive = true, createdAt = 0),
+        )
+        db.libraryDao().upsertAlbums(
+            listOf(
+                album("big", "Big Album"),
+                album("tie", "Tie Album"),
+            ),
+        )
+        val big =
+            (0 until 300).map { index ->
+                val disc = if (index < 150) 1L else 2L
+                val track = (index % 150 + 1).toLong()
+                song(id = "b$index", albumId = "big", disc = disc, track = track)
+            }
+        val tie =
+            listOf(
+                song(id = "t1z", albumId = "tie", disc = 1, track = 1),
+                song(id = "t1a", albumId = "tie", disc = 1, track = 1),
+            )
+        db.libraryDao().upsertSongs(big + tie)
+
+        val positions = (0 until 20).map { it * 2L }
+        val playlistSongs =
+            positions.flatMapIndexed { index, position ->
+                if (index == 5) {
+                    listOf(PlaylistSong(1, "pl1", "ghost", position))
+                } else {
+                    listOf(PlaylistSong(1, "pl1", "b${index * 7}", position))
+                }
+            }
+        db.libraryDao().upsertPlaylistSongs(playlistSongs)
+    }
+
+    private fun album(
+        id: String,
+        name: String,
+    ) = Album(
+        sourceId = 1,
+        id = id,
+        artistId = "ar1",
+        name = name,
+        albumArtist = "Artist",
+        created = 0,
+        coverArt = null,
+        genre = null,
+        year = null,
+        starred = null,
+        songCount = 1,
+    )
+
+    private fun song(
+        id: String,
+        albumId: String,
+        disc: Long,
+        track: Long,
+    ) = Song(
+        sourceId = 1,
+        id = id,
+        albumId = albumId,
+        artistId = "ar1",
+        title = "Song $id",
+        album = "Album",
+        artist = "Artist",
+        duration = 100,
+        track = track,
+        disc = disc,
+        starred = null,
+        genre = null,
+        created = 0,
+    )
+}