Implementasi paginasi standar di Django menggunakan django.core.paginator.Paginator menghasilkan query berbasis LIMIT dan OFFSET. Pola ini mengalami degradasi performa linear seiring bertambahnya halaman data (deep offset) pada tabel berskala jutaan baris. Solusi deterministik untuk mengeliminasi latensi ini adalah migrasi ke keyset pagination (cursor-based pagination).

Akar Masalah Degradasi I/O pada Deep Offset

Saat mengeksekusi query seperti SELECT * FROM audit_log ORDER BY created_at DESC LIMIT 20 OFFSET 500000;, storage engine PostgreSQL tidak langsung melompat ke baris nomor 500.001. Database engine harus membaca 500.020 tuple dari disk/buffer cache, mengurutkannya sesuai kriteria, lalu membuang (discard) 500.000 baris pertama.

Proses ini memicu beban tinggi pada I/O disk dan CPU buffer hits. Kompleksitas waktunya adalah O(N) terhadap nilai offset. Keyset pagination menyelesaikan masalah ini dengan menggunakan nilai penanda baris terakhir (cursor) sebagai predikat filter kondisi: WHERE (created_at, id) < (last_created_at, last_id). Kompleksitas terpangkas menjadi O(log N) karena database langsung melompat ke posisi leaf node indeks B-Tree yang tepat.

Perbandingan Query Plan: EXPLAIN ANALYZE

Berikut representasi logis dari EXPLAIN ANALYZE pada PostgreSQL 15 dengan tabel berisi 5.000.000 baris data:

-- Pola OFFSET: Membaca dan membuang 500.000 baris
EXPLAIN ANALYZE SELECT * FROM audit_log ORDER BY created_at DESC, id DESC LIMIT 20 OFFSET 500000;

Limit (cost=48210.12..48212.05 rows=20 width=84) (actual time=285.312..285.324 rows=20 loops=1)
  -> Index Scan using idx_created_at_id on audit_log (cost=0.43..482101.40 rows=5000000 width=84) 
        (actual time=0.041..254.120 rows=500020 loops=1)
Execution Time: 285.360 ms

-- Pola KEYSET: Direct seek menggunakan B-Tree index scan
EXPLAIN ANALYZE SELECT * FROM audit_log 
WHERE (created_at, id) < ('2023-10-15 08:30:00+00', 4500000) 
ORDER BY created_at DESC, id DESC LIMIT 20;

Limit (cost=0.43..2.35 rows=20 width=84) (actual time=0.038..0.049 rows=20 loops=1)
  -> Index Scan using idx_created_at_id on audit_log (cost=0.43..482101.40 rows=5000000 width=84) 
        (actual time=0.036..0.046 rows=20 loops=1)
        Index Cond: (ROW(created_at, id) < ROW('2023-10-15 08:30:00+00'::timestamptz, 4500000))
Execution Time: 0.075 ms

Deklarasi Model dan Indeks Komposit Django

Keyset pagination wajib didukung indeks database yang mencakup kolom sorting utama dan kolom penentu keunikan (tie-breaker). Kolom penanda waktu seperti created_at tidak unik secara mandiri, sehingga id (primary key) harus digabungkan.

from django.db import models

class AuditLog(models.Model):
    created_at = models.DateTimeField(db_index=True)
    message = models.TextField()
    level = models.CharField(max_length=20)

    class Meta:
        # ponytail: descending index compound for direct cursor seeks.
        # Upgrade path: add partial index if queries filter by a fixed partition/tenant.
        indexes = [
            models.Index(
                fields=['-created_at', '-id'],
                name='idx_audit_created_id_desc'
            ),
        ]
        ordering = ['-created_at', '-id']

    def __str__(self):
        return f"{self.id} - {self.created_at}"

Implementasi Query Pattern ORM

Django ORM tidak memiliki sintaks tuple comparison bawaan seperti (a, b) < (x, y) secara agnostik lintas driver. Gunakan operator logika kombinasi Q object untuk mengekspresikan relasi ekuivalen tanpa library pihak ketiga:

(created_at < cursor_created_at) OR (created_at == cursor_created_at AND id < cursor_id)

from datetime import datetime
from typing import Optional, Tuple, List
from django.db.models import Q, QuerySet

def paginate_keyset(
    queryset: QuerySet,
    cursor_created_at: Optional[datetime] = None,
    cursor_id: Optional[int] = None,
    page_size: int = 20
) -> Tuple[List[models.Model], Optional[dict]]:
    """
    Query keyset pagination murni menggunakan Django ORM.
    Mendukung tie-breaker id jika created_at bernilai identik.
    """
    qs = queryset
    if cursor_created_at is not None and cursor_id is not None:
        # Evaluasi deterministik tie-breaker
        condition = Q(created_at__lt=cursor_created_at) | Q(
            created_at=cursor_created_at, id__lt=cursor_id
        )
        qs = qs.filter(condition)

    # Ambil page_size + 1 untuk mengecek ketersediaan halaman berikutnya (has_next)
    records = list(qs.order_by('-created_at', '-id')[:page_size + 1])
    
    has_next = len(records) > page_size
    results = records[:page_size]

    next_cursor = None
    if has_next and results:
        last_item = results[-1]
        next_cursor = {
            'cursor_created_at': last_item.created_at.isoformat(),
            'cursor_id': last_item.id
        }

    return results, next_cursor

Edge Case: Timestamp Identik

Pada lingkungan konkurensi tinggi, batch write menghasilkan ribuan baris dengan created_at identik mikrodetiknya. Jika query hanya menggunakan created_at < cursor_created_at, baris-baris ber-timestamp sama yang terpotong pada batas limit halaman akan terlewati selamanya (skipped records).

Indeks komposit (created_at, id) bersama predikat Q(created_at=cursor_created_at, id__lt=cursor_id) menjamin pagination tetap deterministik tanpa ada data yang hilang atau diduplikasi di antara pergantian cursor.

Trade-offs dan Batasan

  • Tidak ada lompatan halaman acak (No Random Page Jumping): Klien tidak bisa langsung beralih ke halaman 45. Navigasi harus sequential (Next/Previous).
  • Arah balik query (Bidirectional Traversal): Untuk pagination mundur, filter operator harus dibalik (__gt) dan urutan sorting dibalik sementara, lalu hasilnya dibalik kembali di memori.
  • Keterikatan Indeks: Skema query terikat ketat dengan urutan kolom pada composite index. Mengubah arah sort (misal ascending) memerlukan composite index yang mendukung urutan tersebut secara simetris.

Unit Test Integritas Cursor

Pengujian memastikan konsistensi navigasi ketika data memiliki timestamp identik dan memastikan tidak ada data yang terduplikasi di batas halaman.

from django.test import TestCase
from django.utils import timezone
from datetime import timedelta
from .models import AuditLog
from .services import paginate_keyset

class KeysetPaginationTestCase(TestCase):
    def setUp(self):
        fixed_time = timezone.now()
        # Buat 5 item dengan timestamp identik untuk validasi tie-breaker
        for i in range(1, 6):
            AuditLog.objects.create(
                id=i,
                created_at=fixed_time,
                message=f"Log identik {i}",
                level="INFO"
            )
        # Buat 5 item dengan timestamp lebih baru
        for i in range(6, 11):
            AuditLog.objects.create(
                id=i,
                created_at=fixed_time + timedelta(seconds=i),
                message=f"Log beda {i}",
                level="INFO"
            )

    def test_keyset_traversal_integrity(self):
        page_size = 4
        all_retrieved_ids = []

        # Halaman 1
        results, next_cursor = paginate_keyset(AuditLog.objects.all(), page_size=page_size)
        self.assertEqual(len(results), 4)
        all_retrieved_ids.extend([item.id for item in results])
        self.assertIsNotNone(next_cursor)

        # Halaman 2
        results_page2, next_cursor = paginate_keyset(
            AuditLog.objects.all(),
            cursor_created_at=next_cursor['cursor_created_at'],
            cursor_id=next_cursor['cursor_id'],
            page_size=page_size
        )
        self.assertEqual(len(results_page2), 4)
        all_retrieved_ids.extend([item.id for item in results_page2])

        # Halaman 3 (Sisa data)
        results_page3, next_cursor = paginate_keyset(
            AuditLog.objects.all(),
            cursor_created_at=next_cursor['cursor_created_at'],
            cursor_id=next_cursor['cursor_id'],
            page_size=page_size
        )
        self.assertEqual(len(results_page3), 2)
        all_retrieved_ids.extend([item.id for item in results_page3])
        self.assertIsNone(next_cursor)

        # Pastikan seluruh 10 data terambil tepat satu kali tanpa duplikasi
        self.assertEqual(len(all_retrieved_ids), 10)
        self.assertEqual(len(set(all_retrieved_ids)), 10)
        # Pastikan data terurut menurun
        self.assertEqual(all_retrieved_ids, sorted(all_retrieved_ids, reverse=True))