Server-side pagination for large result sets

- ResultStore: execute query, store batches server-side, serve pages on demand
- POST /query/paged → returns query_id + total_rows + page count (no rows)
- GET /query/page/{id}/{page}?size=100 → returns one page of rows
- RecordBatch slicing for efficient page extraction from Arrow batches
- LRU eviction: keeps 50 most recent query results in memory
- Tested: 100K rows → 1,000 pages of 100, any page fetchable by number
- Supervisor pattern: chunk results, serve on demand, retry-safe (idempotent GET)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

crates/queryd/src/service.rs

@@ -2,7 +2,7 @@ use arrow::array::RecordBatch;
 use arrow::json::writer::{JsonArray, Writer as JsonWriter};
 use axum::{
     Json, Router,
-    extract::State,
+    extract::{Path, Query, State},
     http::StatusCode,
     response::IntoResponse,
     routing::{get, post},
@@ -12,16 +12,29 @@ use serde::{Deserialize, Serialize};
 use crate::cache::CacheStats;
 use crate::context::QueryEngine;
 use crate::delta;
+use crate::paged::ResultStore;
+
+#[derive(Clone)]
+pub struct QueryState {
+    pub engine: QueryEngine,
+    pub result_store: ResultStore,
+}
 
 pub fn router(engine: QueryEngine) -> Router {
+    let state = QueryState {
+        engine: engine.clone(),
+        result_store: ResultStore::new(100, 50), // 100 rows/page, keep 50 results
+    };
     Router::new()
         .route("/health", get(health))
         .route("/sql", post(execute_query))
+        .route("/paged", post(paged_query))
+        .route("/page/{query_id}/{page}", get(get_page))
         .route("/cache/pin", post(pin_dataset))
         .route("/cache/evict", post(evict_dataset))
         .route("/cache/stats", get(cache_stats))
         .route("/compact", post(compact_dataset))
-        .with_state(engine)
+        .with_state(state)
 }
 
 async fn health() -> &'static str {
@@ -60,12 +73,12 @@ fn batches_to_json(batches: &[RecordBatch]) -> Result<serde_json::Value, String>
 }
 
 async fn execute_query(
-    State(engine): State<QueryEngine>,
+    State(state): State<QueryState>,
     Json(req): Json<QueryRequest>,
 ) -> impl IntoResponse {
     tracing::info!("executing query: {}", req.sql);
 
-    match engine.query(&req.sql).await {
+    match state.engine.query(&req.sql).await {
         Ok(batches) => {
             if batches.is_empty() {
                 return Ok(Json(QueryResponse {
@@ -96,6 +109,35 @@ async fn execute_query(
     }
 }
 
+// --- Paged Queries (large result sets) ---
+
+async fn paged_query(
+    State(state): State<QueryState>,
+    Json(req): Json<QueryRequest>,
+) -> impl IntoResponse {
+    tracing::info!("paged query: {}", req.sql);
+    match state.result_store.execute_and_store(&state.engine, &req.sql).await {
+        Ok(handle) => Ok(Json(handle)),
+        Err(e) => Err((StatusCode::BAD_REQUEST, e)),
+    }
+}
+
+#[derive(Deserialize)]
+struct PageQuery {
+    size: Option<usize>,
+}
+
+async fn get_page(
+    State(state): State<QueryState>,
+    Path((query_id, page)): Path<(String, usize)>,
+    Query(q): Query<PageQuery>,
+) -> impl IntoResponse {
+    match state.result_store.get_page(&query_id, page, q.size).await {
+        Ok(result) => Ok(Json(result)),
+        Err(e) => Err((StatusCode::NOT_FOUND, e)),
+    }
+}
+
 // --- Cache Management ---
 
 #[derive(Deserialize)]
@@ -104,28 +146,28 @@ struct CacheRequest {
 }
 
 async fn pin_dataset(
-    State(engine): State<QueryEngine>,
+    State(state): State<QueryState>,
     Json(req): Json<CacheRequest>,
 ) -> impl IntoResponse {
-    match engine.pin_dataset(&req.dataset).await {
+    match state.engine.pin_dataset(&req.dataset).await {
         Ok(()) => Ok((StatusCode::OK, format!("pinned: {}", req.dataset))),
         Err(e) => Err((StatusCode::INTERNAL_SERVER_ERROR, e)),
     }
 }
 
 async fn evict_dataset(
-    State(engine): State<QueryEngine>,
+    State(state): State<QueryState>,
     Json(req): Json<CacheRequest>,
 ) -> impl IntoResponse {
-    if engine.cache().evict(&req.dataset).await {
+    if state.engine.cache().evict(&req.dataset).await {
         (StatusCode::OK, format!("evicted: {}", req.dataset))
     } else {
         (StatusCode::NOT_FOUND, format!("not cached: {}", req.dataset))
     }
 }
 
-async fn cache_stats(State(engine): State<QueryEngine>) -> impl IntoResponse {
-    let stats = engine.cache().stats().await;
+async fn cache_stats(State(state): State<QueryState>) -> impl IntoResponse {
+    let stats = state.engine.cache().stats().await;
     Json(stats)
 }
 
@@ -139,11 +181,11 @@ struct CompactRequest {
 }
 
 async fn compact_dataset(
-    State(engine): State<QueryEngine>,
+    State(state): State<QueryState>,
     Json(req): Json<CompactRequest>,
 ) -> impl IntoResponse {
     match delta::compact(
-        engine.store(),
+        state.engine.store(),
         &req.dataset,
         &req.base_key,
         req.primary_key.as_deref(),