fix(duckdb): stabilize query pages and bound queued cancellation

agents/drivers/duckdb/tests/duckdb_worker_process.rs

@@ -19,6 +19,12 @@ static TEMP_DB_COUNTER: AtomicU64 = AtomicU64::new(0);
 #[path = "../../../../crates/dbx-core/tests/support/duckdb_export.rs"]
 mod export_regressions;
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+async fn worker_query_pages_preserve_one_execution() {
+    let _guard = duckdb_worker_process_test_guard().await;
+    export_regressions::run_query_pages(PathBuf::from(env!("CARGO_BIN_EXE_dbx-duckdb-driver")));
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn worker_parquet_exports_preserve_rows_and_project_totals() {
     let _guard = duckdb_worker_process_test_guard().await;

apps/desktop/src/components/export/ExportProgressDialog.vue

@@ -60,7 +60,7 @@ const displayName = computed(() => {
   }
   return `${props.tableName} (.${props.format})`;
 });
-const isActive = computed(() => props.status === "Queued" || props.status === "Running" || props.status === "Writing" || props.status === "Cancelling");
+const isActive = computed(() => props.status === "Queued" || props.status === "PreparingSnapshot" || props.status === "Running" || props.status === "Writing" || props.status === "Cancelling");
 const isFinished = computed(() => props.status === "Done" || props.status === "Error" || props.status === "Cancelled");
 const canChangeQueryTimeout = computed(() => props.status === "Error" && !!props.connectionId && !!props.errorMessage && isQueryTimeoutErrorMessage(props.errorMessage));
 const canRevealFile = computed(() => props.status === "Done" && !!props.filePath && isTauriRuntime());
@@ -124,8 +124,8 @@ watch(
 
         <!-- Progress bar -->
         <div class="w-full bg-muted rounded-full h-2 overflow-hidden">
-          <template v-if="status === 'Running' || status === 'Writing'">
-            <div v-if="totalRows" class="h-full bg-primary rounded-full transition-[width] duration-300" :style="{ width: `${progressPercent}%` }" />
+          <template v-if="status === 'PreparingSnapshot' || status === 'Running' || status === 'Writing'">
+            <div v-if="totalRows && status !== 'PreparingSnapshot'" class="h-full bg-primary rounded-full transition-[width] duration-300" :style="{ width: `${progressPercent}%` }" />
             <div v-else class="h-full w-full overflow-hidden rounded-full">
               <div class="export-progress-indeterminate h-full rounded-full bg-primary" />
             </div>
@@ -147,6 +147,10 @@ watch(
             <Loader2 class="h-4 w-4 animate-spin text-primary" />
             <span>{{ t("databaseBackup.cancelling") }}</span>
           </template>
+          <template v-else-if="status === 'PreparingSnapshot'">
+            <Loader2 class="h-4 w-4 shrink-0 animate-spin text-primary" />
+            <span>{{ t("exportProgress.preparingSnapshot") }}</span>
+          </template>
           <template v-else-if="status === 'Running'">
             <Loader2 class="h-4 w-4 animate-spin text-primary" />
             <span>{{ t("exportProgress.fetching") }}</span>
@@ -169,6 +173,8 @@ watch(
           </template>
         </div>
 
+        <p v-if="status === 'PreparingSnapshot'" class="text-xs text-muted-foreground">{{ t("exportProgress.snapshotResources") }}</p>
+
         <!-- Row count -->
         <div v-if="status !== 'Queued'" class="text-xs text-muted-foreground tabular-nums">
           {{ rowsText }}

apps/desktop/src/components/export/ExportProgressPopover.vue

@@ -49,7 +49,7 @@ const visibleTasks = computed(() => {
 
 const hasMore = computed(() => tasks.value.length > MAX_VISIBLE);
 
-const isActive = (status: string) => status === "Running" || status === "Writing" || status === "Cancelling";
+const isActive = (status: string) => status === "PreparingSnapshot" || status === "Running" || status === "Writing" || status === "Cancelling";
 const isFinished = (status: string) => status === "Done" || status === "Error" || status === "Cancelled";
 
 const finishedCount = computed(() => tasks.value.filter((t) => isFinished(t.status)).length);
@@ -212,6 +212,7 @@ const rowsText = (task: ExportTask) => {
 };
 
 const taskStatusText = (task: ExportTask) => {
+  if (task.status === "PreparingSnapshot") return t("exportProgress.preparingSnapshot");
   if (task.status === "Queued") return t("exportProgress.queued");
   if (task.status === "Unknown") return t("exportProgress.unknown");
   if (task.status === "Cancelling") return t("databaseBackup.cancelling");
@@ -272,6 +273,7 @@ const statusIcon = (task: ExportTask) => {
 
 const statusColor = (status: string) => {
   switch (status) {
+    case "PreparingSnapshot":
     case "Running":
     case "Writing":
     case "Cancelling":
@@ -385,7 +387,7 @@ function openTask(task: ExportTask): void {
             <div v-else-if="isActive(task.status)" class="w-full bg-muted rounded-full h-1.5 overflow-hidden">
               <div
                 v-if="
-                  task.totalRows ||
+                  (task.totalRows && task.status !== 'PreparingSnapshot') ||
                   (task.kind === 'database-export' && (task.totalObjects || task.overallPercent !== undefined)) ||
                   (task.kind === 'data-dictionary' && task.dictionaryPhase === 'collecting' && task.dictionaryProgressKnown && task.dictionaryTotal) ||
                   (task.kind === 'data-transfer' && task.totalTables) ||

apps/desktop/src/components/export/__tests__/ExportProgressDialog.spec.ts

@@ -82,6 +82,18 @@ describe("ExportProgressDialog display name", () => {
 });
 
 describe("ExportProgressDialog query timeout action", () => {
+  it("explains snapshot preparation and keeps cancellation available", async () => {
+    i18n.global.locale.value = "en";
+    const onCancel = vi.fn();
+    await mountDialog({ ...baseProps, status: "PreparingSnapshot", totalRows: 20000, onCancel });
+    expect(document.body.textContent).toContain("Preparing a consistent export snapshot");
+    expect(document.body.textContent).toContain("temporary disk space");
+    const cancel = [...document.querySelectorAll("button")].find((button) => button.textContent?.trim() === "Cancel");
+    expect(cancel).toBeTruthy();
+    cancel!.click();
+    expect(onCancel).toHaveBeenCalledOnce();
+  });
+
   it("offers connection settings for a query timeout", async () => {
     i18n.global.locale.value = "en";
     const onChangeQueryTimeout = vi.fn();

apps/desktop/src/components/export/__tests__/ExportProgressPopover.spec.ts

@@ -95,6 +95,26 @@ describe("failure detail copy text", () => {
 });
 
 describe("data dictionary background task", () => {
+  it("keeps a snapshot export active and preserves cancellation across preparation updates", async () => {
+    const tracker = useExportTracker();
+    const task = tracker.addTask("settlements", "csv", "/tmp/settlements.csv", "snapshot-task");
+    const progress = { exportId: task.exportId, tableName: "settlements", rowsExported: 0, totalRows: 20000, status: "PreparingSnapshot" as const };
+    tracker.updateTableExportTask(task.exportId, progress);
+    const cancel = vi.fn();
+    tracker.registerTaskCancelHandler(task.exportId, cancel);
+    await mountPopover();
+    expect(tracker.activeCount.value).toBe(1);
+    expect(document.body.textContent).toContain("Preparing a consistent export snapshot");
+    expect(document.querySelector('button[title="Cancel"]')).toBeTruthy();
+    await tracker.cancelTask(task.exportId);
+    expect(cancel).toHaveBeenCalledOnce();
+    // A delayed preparation event must not turn a cancelling task back into running.
+    tracker.unregisterTaskCancelHandler(task.exportId);
+    task.status = "Cancelling";
+    tracker.updateTableExportTask(task.exportId, progress);
+    expect(task.status).toBe("Cancelling");
+  });
+
   it("shows actual collected-object progress and retains the finished output after the wizard closes", async () => {
     const tracker = useExportTracker();
     tracker.addDataDictionaryTask("dictionary-task", "shop", 2);

apps/desktop/src/composables/useExportTracker.ts

@@ -8,7 +8,7 @@ import { releaseManualExport, submitManualExport, type ManualExportMode } from "
 import { acceptOraclePackageTransferApproval, acceptOraclePackageTransferPreflight, acceptOracleRoutineDeploymentOperation, type OracleRoutineDeployment } from "@/lib/schema/oracleRoutineDeploy";
 
 export type BackgroundTaskKind = "table-export" | "database-export" | "data-dictionary" | "sql-file" | "data-transfer" | "data-generation" | "multi-db-execution" | "schema-diff" | "data-compare";
-export type BackgroundTaskStatus = "Queued" | "Unknown" | "Running" | "Writing" | "Cancelling" | "Done" | "Error" | "Cancelled";
+export type BackgroundTaskStatus = "Queued" | "Unknown" | "PreparingSnapshot" | "Running" | "Writing" | "Cancelling" | "Done" | "Error" | "Cancelled";
 export type DatabaseExportSource = "manual" | "scheduled";
 
 export interface DataTransferFailure {
@@ -372,7 +372,7 @@ function recordSqlFileFailure(task: ExportTask, progress: api.SqlFileProgress, c
 }
 
 function normalizeExportStatus(status: string): BackgroundTaskStatus {
-  if (status === "Writing" || status === "Done" || status === "Error" || status === "Cancelled") return status;
+  if (status === "PreparingSnapshot" || status === "Writing" || status === "Done" || status === "Error" || status === "Cancelled") return status;
   return "Running";
 }
 
@@ -431,7 +431,7 @@ function findActiveOverlappingTransfer(request: api.TransferRequest): string[] {
   for (const task of taskMap.values()) {
     if (task.exportId === request.transferId) continue;
     if (task.kind !== "data-transfer") continue;
-    if (task.status !== "Running" && task.status !== "Writing") continue;
+    if (task.status !== "PreparingSnapshot" && task.status !== "Running" && task.status !== "Writing") continue;
     if (task.targetConnectionId !== request.targetConnectionId) continue;
     if ((task.targetCatalog ?? "") !== (request.targetCatalog ?? "")) continue;
     if (task.targetDatabase !== request.targetDatabase) continue;
@@ -446,7 +446,7 @@ function findActiveOverlappingTransfer(request: api.TransferRequest): string[] {
 export function useExportTracker() {
   const tasks = computed(() => Array.from(taskMap.values()));
 
-  const activeCount = computed(() => tasks.value.filter((t) => t.status === "Queued" || t.status === "Unknown" || t.status === "Running" || t.status === "Writing" || t.status === "Cancelling").length);
+  const activeCount = computed(() => tasks.value.filter((t) => t.status === "Queued" || t.status === "Unknown" || t.status === "PreparingSnapshot" || t.status === "Running" || t.status === "Writing" || t.status === "Cancelling").length);
 
   const hasActive = computed(() => activeCount.value > 0);
 
@@ -787,7 +787,7 @@ export function useExportTracker() {
     task.rowsExported = progress.rowsExported;
     task.totalRows = progress.totalRows;
     const nextStatus = normalizeExportStatus(progress.status);
-    task.status = task.status === "Cancelling" && (nextStatus === "Running" || nextStatus === "Writing") ? "Cancelling" : nextStatus;
+    task.status = task.status === "Cancelling" && (nextStatus === "PreparingSnapshot" || nextStatus === "Running" || nextStatus === "Writing") ? "Cancelling" : nextStatus;
     task.errorMessage = progress.errorMessage || null;
     if (task.status === "Done" || task.status === "Error" || task.status === "Cancelled") finishExportTask(task);
   }

apps/desktop/src/i18n/locales/en.ts

@@ -3949,6 +3949,8 @@ export default {
     agentSessionMissing: "Streaming export needs a result-set session, but this driver returned no session_id.",
     title: "Exporting Table Data",
     fetching: "Fetching data from database...",
+    preparingSnapshot: "Preparing a consistent export snapshot...",
+    snapshotResources: "Large results may use temporary disk space. Other queries on this connection wait until the snapshot is ready.",
     writing: "Writing to file...",
     done: "Export completed!",
     error: "Export failed",

apps/desktop/src/i18n/locales/es.ts

@@ -3737,6 +3737,8 @@ export default withEnglishFallback({
     copyFailureDetail: "Copiar detalles del fallo",
     failureDetailCopied: "Detalles del fallo copiados",
     failureDetailCopyFailed: "No se pudieron copiar los detalles del fallo: {message}",
+    preparingSnapshot: "Generando una instantánea de exportación coherente...",
+    snapshotResources: "Los grandes volúmenes de datos pueden utilizar espacio temporal en disco. Las demás consultas de esta conexión deben esperar a que finalice la generación de la instantánea.",
   },
   welcome: {
     introEyebrow: "Espacio de trabajo de bases de datos",

apps/desktop/src/i18n/locales/it.ts

@@ -3592,6 +3592,8 @@ export default withEnglishFallback({
     copyFailureDetail: "Copia dettagli dell'errore",
     failureDetailCopied: "Dettagli dell'errore copiati",
     failureDetailCopyFailed: "Impossibile copiare i dettagli dell'errore: {message}",
+    preparingSnapshot: "Generazione di uno snapshot di esportazione coerente in corso...",
+    snapshotResources: "Con grandi volumi di dati potrebbe essere utilizzato spazio su disco temporaneo. Le altre query su questa connessione devono attendere il completamento della generazione dello snapshot.",
   },
   welcome: {
     introEyebrow: "Spazio di lavoro del database",

apps/desktop/src/i18n/locales/ja.ts

@@ -3616,6 +3616,8 @@ export default withEnglishFallback({
     multiDbExecutionTitle: "複数データベースで実行：{name}",
     multiDbTargets: "ターゲット {completed}/{total}、成功 {success}、失敗 {failed}",
     openTask: "タスクを開く",
+    preparingSnapshot: "一貫したエクスポートスナップショットを生成しています...",
+    snapshotResources: "大量のデータの場合、一時ディスク領域を使用する可能性があります。この接続上の他のクエリは、スナップショットの生成が完了するまで待機する必要があります。",
   },
   welcome: {
     introEyebrow: "データベースワークスペース",

apps/desktop/src/i18n/locales/ko.ts

@@ -3546,6 +3546,8 @@ export default withEnglishFallback({
     multiDbExecutionTitle: "다중 데이터베이스 실행: {name}",
     multiDbTargets: "대상 {completed}/{total}, 성공 {success}, 실패 {failed}",
     openTask: "작업 열기",
+    preparingSnapshot: "일관된 내보내기 스냅샷을 생성하는 중...",
+    snapshotResources: "대용량 데이터는 임시 디스크 공간을 사용할 수 있습니다. 이 연결의 다른 쿼리는 스냅샷 생성이 완료될 때까지 기다려야 합니다.",
   },
   welcome: {
     introEyebrow: "데이터베이스 작업 공간",

apps/desktop/src/i18n/locales/pt-BR.ts

@@ -3592,6 +3592,8 @@ export default withEnglishFallback({
     multiDbExecutionTitle: "Execução em vários bancos de dados: {name}",
     multiDbTargets: "Destinos {completed}/{total}, sucesso {success}, falha {failed}",
     openTask: "Abrir tarefa",
+    preparingSnapshot: "Gerando um snapshot de exportação consistente...",
+    snapshotResources: "Grandes volumes de dados podem usar espaço temporário em disco. Outras consultas nesta conexão precisam aguardar a conclusão da geração do snapshot.",
   },
   welcome: {
     introEyebrow: "Espaço de trabalho de banco de dados",

apps/desktop/src/i18n/locales/ru.ts

@@ -3693,6 +3693,8 @@ export default withEnglishFallback({
     copyFailureDetail: "Копировать детали ошибок",
     failureDetailCopied: "Детали ошибок скопированы",
     failureDetailCopyFailed: "Не удалось скопировать детали ошибок: {message}",
+    preparingSnapshot: "Создание согласованного снимка для экспорта...",
+    snapshotResources: "Большие объёмы данных могут занимать временное дисковое пространство. Другим запросам в этом подключении придётся ждать завершения создания снимка.",
   },
   welcome: {
     introEyebrow: "Рабочее пространство базы данных",

apps/desktop/src/i18n/locales/th.ts

@@ -3869,6 +3869,8 @@ export default withEnglishFallback({
     copyFailureDetail: "คัดลอกรายละเอียดความล้มเหลว",
     failureDetailCopied: "คัดลอกรายละเอียดความล้มเหลวแล้ว",
     failureDetailCopyFailed: "คัดลอกรายละเอียดความล้มเหลวไม่สำเร็จ: {message}",
+    preparingSnapshot: "กำลังสร้างสแนปช็อตการส่งออกที่สอดคล้องกัน...",
+    snapshotResources: "ข้อมูลปริมาณมากอาจใช้พื้นที่ดิสก์ชั่วคราว คิวรีอื่นบนการเชื่อมต่อนี้ต้องรอจนกว่าการสร้างสแนปช็อตจะเสร็จสิ้น",
   },
   welcome: {
     introEyebrow: "Workspace ฐานข้อมูล",

apps/desktop/src/i18n/locales/zh-CN.ts

@@ -3879,6 +3879,8 @@ export default withEnglishFallback({
     agentSessionMissing: "流式导出需要结果集会话，但当前驱动未返回 session_id。",
     title: "导出表数据",
     fetching: "正在从数据库获取数据...",
+    preparingSnapshot: "正在生成一致的导出快照...",
+    snapshotResources: "大数据量可能使用临时磁盘空间。此连接上的其他查询需等待快照生成完成。",
     writing: "正在写入文件...",
     done: "导出完成！",
     error: "导出失败",

apps/desktop/src/i18n/locales/zh-TW.ts

@@ -3514,6 +3514,8 @@ export default withEnglishFallback({
     agentSessionMissing: "串流匯出需要結果集工作階段，但目前驅動程式未回傳 session_id。",
     title: "匯出資料表資料",
     fetching: "正在從資料庫擷取資料...",
+    preparingSnapshot: "正在產生一致的匯出快照...",
+    snapshotResources: "大量資料可能使用暫存磁碟空間。此連線上的其他查詢需等候快照產生完成。",
     writing: "正在寫入檔案...",
     done: "匯出完成！",
     error: "匯出失敗",

apps/desktop/src/lib/__tests__/database/databaseDriverManifest.spec.ts

@@ -37,7 +37,7 @@ describe("databaseDriverManifest", () => {
     }
   });
 
-  it("uses agent cursor only for agent or external runtimes", () => {
+  it("uses result cursors for agent, external, and native DuckDB queries", () => {
     expect(databaseRuntimeMode("gaussdb")).toBe("native");
     expect(databaseRuntimeMode("meilisearch")).toBe("native");
     expect(databaseRuntimeMode("dynamodb")).toBe("native");
@@ -45,6 +45,7 @@ describe("databaseDriverManifest", () => {
     expect(usesAgentCursorForQuery("meilisearch")).toBe(false);
     expect(usesAgentCursorForQuery("dynamodb")).toBe(false);
     expect(usesAgentCursorForQuery("mysql")).toBe(false);
+    expect(usesAgentCursorForQuery("duckdb")).toBe(true);
     expect(usesAgentCursorForQuery("cassandra")).toBe(true);
     expect(usesAgentCursorForQuery("jdbc")).toBe(true);
     expect(usesAgentCursorForQuery("prestosql")).toBe(true);

apps/desktop/src/lib/backend/tauri.ts

@@ -6659,7 +6659,7 @@ export interface ExportProgress {
 }
 
 // --- Table Export ---
-export type TableExportStatus = "Running" | "Writing" | "Done" | "Error" | "Cancelled" | "Unknown";
+export type TableExportStatus = "PreparingSnapshot" | "Running" | "Writing" | "Done" | "Error" | "Cancelled" | "Unknown";
 
 export interface TableExportRequest {
   exportId: string;

apps/desktop/src/lib/database/databaseDriverManifest.ts

@@ -172,6 +172,8 @@ export function manifestDatabaseTypes(): DatabaseType[] {
 }
 
 export function usesAgentCursorForQuery(dbType?: DatabaseType, driverProfile?: string): boolean {
+  // DuckDB's native worker captures one bounded result for all subsequent pages.
+  if (dbType === "duckdb") return true;
   if (dbType === "sqlserver" && driverProfile?.trim().toLowerCase() === "sqlserver-legacy") return true;
   const runtimeMode = databaseRuntimeMode(dbType);
   return runtimeMode === "agent" || runtimeMode === "external";

crates/dbx-core/src/data/database_export.rs

@@ -472,6 +472,7 @@ impl LenientExportErrors {
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum ExportStatus {
+    PreparingSnapshot,
     Running,
     Writing,
     Done,

crates/dbx-core/src/data/query_result_export.rs

@@ -1185,6 +1185,7 @@ async fn export_query_result_core_inner(
     let mut offset: usize = 0;
     let mut wrote_text_header = false;
     if request.database_type == DatabaseType::DuckDb && txn_session_id.is_none() {
+        on_progress(progress(request, 0, ExportStatus::PreparingSnapshot, None));
         let snapshot = duckdb_snapshot.insert(DuckDbExportSnapshot::new(
             &request.connection_id,
             &request.database,

crates/dbx-core/src/data/table_export.rs

@@ -1879,6 +1879,14 @@ async fn export_table_data_core_inner(
     }
 
     if db_type == DatabaseType::DuckDb {
+        on_progress(TableExportProgress {
+            export_id: request.export_id.clone(),
+            table_name: request.table_name.clone(),
+            rows_exported: 0,
+            total_rows,
+            status: ExportStatus::PreparingSnapshot,
+            error_message: None,
+        });
         let snapshot = duckdb_snapshot.insert(DuckDbExportSnapshot::new(
             &request.connection_id,
             &request.database,

crates/dbx-core/src/query/mod.rs

@@ -1798,21 +1798,34 @@ async fn do_execute_typed(
         #[cfg(feature = "duckdb-sidecar")]
         PoolKind::DuckDbWorker(client) => {
             let client = client.clone();
+            let operation_cancel = cancel_token.unwrap_or_default();
             if let Some(ref execution_id) = options.execution_id {
-                let cancel_client = client.clone();
+                let operation_cancel = operation_cancel.clone();
                 state.running_queries.register_interrupt(execution_id, move || {
-                    let cancel_client = cancel_client.clone();
-                    tokio::spawn(async move {
-                        if let Err(error) = cancel_client.cancel().await {
-                            log::warn!("Failed to cancel DuckDB worker query: {error}");
-                        }
-                    });
+                    operation_cancel.cancel();
                 });
             }
+            let cancel_token = Some(operation_cancel);
             let sql = sql.to_string();
             let database = database.map(str::to_string);
             let max_rows = options.max_rows;
-            match client.execute_typed(database, sql, max_rows, cancel_token, query_timeout).await {
+            let result = if let Some(page_size) = options.page_size.filter(|_| !is_write_sql(&sql)) {
+                client
+                    .execute_query_page(
+                        database,
+                        sql,
+                        max_rows,
+                        page_size,
+                        options.result_session_id.as_deref(),
+                        cancel_token,
+                        query_timeout,
+                    )
+                    .await
+                    .map_err(db::duckdb_worker_protocol::DuckDbWorkerError::from)
+            } else {
+                client.execute_typed(database, sql, max_rows, cancel_token, query_timeout).await
+            };
+            match result {
                 Ok(result) => Ok(result),
                 Err(error) => {
                     let is_control_error = error.message == QUERY_CANCELED
@@ -3643,6 +3656,8 @@ pub async fn close_query_session(
     let pool_handle = state.pool_handle(&pool_key).await;
     let pool = pool_handle.as_ref().ok_or("Connection not found")?;
     match pool {
+        #[cfg(feature = "duckdb-sidecar")]
+        PoolKind::DuckDbWorker(client) => client.close_query_session(session_id).await,
         PoolKind::Agent(client) => {
             let client = client.clone();
             let mut client = client.lock().await;

crates/dbx-core/tests/live_duckdb_export.rs

@@ -3,6 +3,12 @@
 #[path = "support/duckdb_export.rs"]
 mod support;
 
+#[test]
+#[ignore = "requires DBX_DUCKDB_DRIVER_PATH pointing at a DuckDB worker"]
+fn duckdb_query_pages_preserve_one_execution() {
+    support::run_query_pages(std::env::var_os("DBX_DUCKDB_DRIVER_PATH").unwrap().into());
+}
+
 #[test]
 #[ignore = "requires DBX_DUCKDB_DRIVER_PATH pointing at a DuckDB worker"]
 fn duckdb_parquet_exports_are_consistent() {

crates/dbx-core/tests/support/duckdb_export.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 use dbx_core::connection::{AppState, PoolKind};
 use dbx_core::db::duckdb_worker_process::DuckDbWorkerClient;
 use dbx_core::models::connection::ConnectionConfig;
+use dbx_core::query::{close_query_session, execute_sql_statement_with_options, QueryExecutionOptions};
 use dbx_core::query_result_export::{export_query_result_core, QueryResultExportRequest};
 use dbx_core::table_export::{export_table_data_core, TableExportRequest};
 use dbx_core::table_import::{parse_csv_bytes, parse_xlsx_file};
@@ -30,6 +31,10 @@ pub fn run_snapshot_cleanup(driver: PathBuf) {
     run_with_stack(move || snapshots_preserve_query_values_and_cleanup(driver));
 }
 
+pub fn run_query_pages(driver: PathBuf) {
+    run_with_stack(move || query_pages_preserve_one_execution(driver));
+}
+
 struct Fixture {
     state: AppState,
     client: Arc<DuckDbWorkerClient>,
@@ -74,6 +79,10 @@ impl Fixture {
             "SELECT count(*) FROM duckdb_tables() WHERE database_name = 'temp' AND starts_with(table_name, '__dbx_export_')",
         ).await;
         assert_eq!(cell_number(&result.rows[0][0]), 0);
+        let result = self.sql(
+            "SELECT count(*) FROM duckdb_tables() WHERE database_name = 'temp' AND starts_with(table_name, '__dbx_result_')",
+        ).await;
+        assert_eq!(cell_number(&result.rows[0][0]), 0);
         assert_eq!(cell_number(&self.sql("SELECT * FROM keep_me").await.rows[0][0]), 42);
     }
 
@@ -206,6 +215,33 @@ async fn snapshots_preserve_query_values_and_cleanup(driver: PathBuf) {
     assert_eq!(cell_number(&fixture.sql("SELECT nextval('calls')").await.rows[0][0]), 2051);
     fixture.assert_clean().await;
 
+    // Native DuckDB exposes SQL BEGIN/ROLLBACK on its shared session, rather
+    // than DBX manual-transaction handles. Export must see pending writes and
+    // must not commit or discard the user's transaction.
+    assert!(dbx_core::query::begin_manual_transaction(&fixture.state, &fixture.id, "main", None, None)
+        .await
+        .unwrap_err()
+        .contains("not supported"));
+    fixture.sql("CREATE TABLE transaction_values (id INT)").await;
+    fixture.sql("BEGIN").await;
+    fixture.sql("INSERT INTO transaction_values VALUES (42)").await;
+    let request = fixture.query_request("SELECT * FROM transaction_values", &path, "csv");
+    let preparing = std::sync::atomic::AtomicBool::new(false);
+    export_query_result_core(&fixture.state, &request, None, |progress| {
+        if matches!(progress.status, dbx_core::table_export::ExportStatus::PreparingSnapshot) {
+            preparing.store(true, std::sync::atomic::Ordering::Relaxed);
+        }
+    })
+    .await
+    .unwrap();
+    assert!(preparing.load(std::sync::atomic::Ordering::Relaxed));
+    let parsed = parse_csv_bytes(&std::fs::read(&path).unwrap(), 10).unwrap();
+    assert_eq!(parsed.rows.len(), 1);
+    assert_eq!(cell_number(&parsed.rows[0][0]), 42);
+    fixture.sql("ROLLBACK").await;
+    assert!(fixture.sql("SELECT * FROM transaction_values").await.rows.is_empty());
+    fixture.assert_clean().await;
+
     let mut request = fixture.query_request("SELECT * FROM range(2050) t(id) ORDER BY id DESC", &path, "csv");
     request.row_limit = Some(513);
     export_query_result_core(&fixture.state, &request, None, |_| {}).await.unwrap();
@@ -242,3 +278,114 @@ async fn snapshots_preserve_query_values_and_cleanup(driver: PathBuf) {
     fixture.assert_clean().await;
     fixture.finish().await;
 }
+
+async fn query_pages_preserve_one_execution(driver: PathBuf) {
+    let fixture = Fixture::new(driver).await;
+    fixture.sql("CREATE TEMP TABLE keep_me AS SELECT 42").await;
+    fixture.sql("CREATE SEQUENCE page_calls START 1").await;
+    fixture.sql("CREATE TABLE page_source AS SELECT i AS id, 'same' AS period FROM range(20000) t(i)").await;
+    for sql in ["SHOW TABLES", "EXPLAIN SELECT 42", "DESCRIBE page_source"] {
+        let result = execute_sql_statement_with_options(
+            &fixture.state,
+            &fixture.id,
+            "main",
+            sql,
+            None,
+            None,
+            QueryExecutionOptions { page_size: Some(100), max_rows: Some(1000), ..Default::default() },
+        )
+        .await
+        .unwrap();
+        assert!(!result.rows.is_empty());
+        assert!(!result.has_more);
+    }
+    for order in ["", " ORDER BY period"] {
+        let sql = format!("SELECT id, period FROM page_source{order}; -- trailing comment");
+        let mut session_id = None;
+        let mut ids = BTreeSet::new();
+        loop {
+            let page = execute_sql_statement_with_options(
+                &fixture.state,
+                &fixture.id,
+                "main",
+                &sql,
+                None,
+                None,
+                QueryExecutionOptions {
+                    page_size: Some(512),
+                    max_rows: Some(20000),
+                    result_session_id: session_id.clone(),
+                    ..Default::default()
+                },
+            )
+            .await
+            .unwrap();
+            assert!(page.rows.len() <= 512);
+            for row in &page.rows {
+                assert!(ids.insert(cell_number(&row[0])), "duplicate grid row");
+            }
+            session_id = page.session_id;
+            if !page.has_more {
+                break;
+            }
+            assert!(session_id.is_some());
+        }
+        assert_eq!(ids, (0..20000).collect());
+        fixture.assert_clean().await;
+    }
+    let sql = "SELECT nextval('page_calls') AS dup, i AS dup, i AS rowid FROM range(2050) t(i)";
+    let mut session_id = None;
+    let mut values = Vec::new();
+    loop {
+        let page = execute_sql_statement_with_options(
+            &fixture.state,
+            &fixture.id,
+            "main",
+            sql,
+            None,
+            None,
+            QueryExecutionOptions {
+                page_size: Some(100),
+                max_rows: Some(2050),
+                result_session_id: session_id.clone(),
+                ..Default::default()
+            },
+        )
+        .await
+        .unwrap();
+        assert_eq!(page.columns, vec!["dup", "dup", "rowid"]);
+        values.extend(page.rows.iter().map(|row| cell_number(&row[0])));
+        session_id = page.session_id;
+        if !page.has_more {
+            break;
+        }
+    }
+    assert_eq!(values, (1..=2050).collect::<Vec<_>>());
+    assert_eq!(cell_number(&fixture.sql("SELECT nextval('page_calls')").await.rows[0][0]), 2051);
+    fixture.assert_clean().await;
+
+    let sql = "SELECT * FROM range(2050) t(id) ORDER BY id";
+    let first = fixture.client.execute_query_page(None, sql.into(), Some(513), 500, None, None, None).await.unwrap();
+    assert!(first.has_more);
+    let last = fixture
+        .client
+        .execute_query_page(None, sql.into(), Some(513), 500, first.session_id.as_deref(), None, None)
+        .await
+        .unwrap();
+    assert_eq!(last.rows.len(), 13);
+    assert!(!last.has_more);
+    assert!(last.truncated);
+    fixture.assert_clean().await;
+
+    let first = fixture.client.execute_query_page(None, sql.into(), Some(2050), 100, None, None, None).await.unwrap();
+    let id = first.session_id.unwrap();
+    assert!(close_query_session(&fixture.state, &fixture.id, "main", &id, None, None).await.unwrap());
+    assert!(fixture
+        .client
+        .execute_query_page(None, sql.into(), Some(2050), 100, Some(&id), None, None)
+        .await
+        .unwrap_err()
+        .contains("expired"));
+    fixture.assert_clean().await;
+    fixture.finish().await;
+}

crates/dbx-drivers/src/db/duckdb_query_sessions.rs

@@ -0,0 +1,283 @@
+//! Stable result pages on existing workers, without requiring a new worker protocol.
+//! A temporary table lets DuckDB spill large results instead of keeping JSON rows
+//! in the application. Only the configured result limit (plus one probe) is captured.
+
+use super::*;
+use std::time::Instant;
+
+const IDLE_TIMEOUT: Duration = Duration::from_secs(300);
+const CLEANUP_TIMEOUT: Duration = Duration::from_secs(5);
+
+pub(super) struct QuerySession {
+    table: String,
+    database: Option<String>,
+    columns: Vec<String>,
+    offset: usize,
+    row_limit: usize,
+    generation: u64,
+    last_activity: Instant,
+    expiry: CancellationToken,
+}
+
+impl QuerySession {
+    fn qualified_table(&self) -> String {
+        format!("\"temp\".\"main\".\"{}\"", self.table)
+    }
+}
+
+impl DuckDbWorkerClient {
+    /// All pages refer to one execution, even for an unordered query or an
+    /// ORDER BY with ties. Never silently rerun an expired session's SQL.
+    #[allow(clippy::too_many_arguments)]
+    pub async fn execute_query_page(
+        &self,
+        database: Option<String>,
+        sql: String,
+        max_rows: Option<usize>,
+        page_size: usize,
+        session_id: Option<&str>,
+        cancel: Option<CancellationToken>,
+        timeout: Option<Duration>,
+    ) -> Result<db::QueryResult, String> {
+        let id = session_id.map(str::to_string).unwrap_or_else(|| uuid::Uuid::new_v4().simple().to_string());
+        let mut session = if session_id.is_some() {
+            self.inner
+                .query_sessions
+                .lock()
+                .await
+                .remove(&id)
+                .ok_or("DuckDB result session expired; rerun the query")?
+        } else {
+            let statements = dbx_sql_core::sql::split_sql_statements_for_database(
+                &sql,
+                crate::models::connection::DatabaseType::DuckDb,
+            );
+            if statements.len() != 1 {
+                return Err("DuckDB result sessions require a single query".into());
+            }
+            let sql = &statements[0];
+            // SHOW/DESCRIBE/EXPLAIN return rows, but are not legal subqueries.
+            // Execute such metadata statements once using the normal row cap.
+            let inspected = dbx_sql_core::query_execution_sql::strip_sql_comments_and_literals(sql);
+            let keyword = inspected.split_whitespace().next().unwrap_or("").to_ascii_uppercase();
+            if !matches!(keyword.as_str(), "SELECT" | "WITH" | "FROM" | "VALUES" | "TABLE" | "PIVOT" | "UNPIVOT") {
+                return self.execute(database, sql.to_string(), max_rows, cancel, timeout).await;
+            }
+            let metadata = self
+                .execute(
+                    database.clone(),
+                    format!("DESCRIBE (\n{sql}\n)"),
+                    Some(i32::MAX as usize),
+                    cancel.clone(),
+                    timeout,
+                )
+                .await?;
+            let columns = metadata
+                .rows
+                .iter()
+                .map(|row| {
+                    row.first()
+                        .and_then(|cell| cell.as_str())
+                        .map(str::to_string)
+                        .ok_or_else(|| "DuckDB returned invalid result metadata".to_string())
+                })
+                .collect::<Result<Vec<_>, _>>()?;
+            if metadata.truncated || columns.is_empty() {
+                return Err("DuckDB returned incomplete result metadata".into());
+            }
+            let table = format!("__dbx_result_{id}");
+            let row_limit = max_rows.unwrap_or(10_000).max(1);
+            let aliases = (0..columns.len()).map(|i| format!("\"__dbx_column_{i}\"")).collect::<Vec<_>>().join(", ");
+            let create = format!(
+                "CREATE TEMP TABLE \"{table}\" (\"__dbx_row_number\", {aliases}) AS \
+                 SELECT row_number() OVER (), * FROM (\n{sql}\n) AS \"__dbx_source\" LIMIT {}",
+                row_limit.saturating_add(1),
+            );
+            self.execute(database.clone(), create, Some(1), cancel.clone(), timeout).await?;
+            QuerySession {
+                table,
+                database: database.clone(),
+                columns,
+                offset: 0,
+                row_limit,
+                generation: self.inner.state.lock().await.generation,
+                last_activity: Instant::now(),
+                expiry: CancellationToken::new(),
+            }
+        };
+        // Retire the previous idle timer before reading a page. It must not
+        // expire a session while that page is temporarily outside the map.
+        session.expiry.cancel();
+        let process = self.inner.state.lock().await;
+        let alive = process.connected && process.generation == session.generation;
+        drop(process);
+        if !alive || session.database != database || session.last_activity.elapsed() >= IDLE_TIMEOUT {
+            self.drop_query_snapshot(&session).await;
+            return Err("DuckDB result session expired; rerun the query".into());
+        }
+        let page_size = page_size.max(1).min(session.row_limit.saturating_sub(session.offset));
+        let projection = session
+            .columns
+            .iter()
+            .enumerate()
+            .map(|(i, name)| format!("\"__dbx_column_{i}\" AS \"{}\"", name.replace('"', "\"\"")))
+            .collect::<Vec<_>>()
+            .join(", ");
+        let page_sql = format!(
+            "SELECT {projection} FROM {} WHERE \"__dbx_row_number\" > {} \
+             AND \"__dbx_row_number\" <= {} ORDER BY \"__dbx_row_number\"",
+            session.qualified_table(),
+            session.offset,
+            session.offset.saturating_add(page_size).saturating_add(1),
+        );
+        let result = self.execute(database, page_sql, Some(page_size.saturating_add(1)), cancel, timeout).await;
+        let mut result = match result {
+            Ok(result) => result,
+            Err(error) => {
+                self.drop_query_snapshot(&session).await;
+                return Err(error);
+            }
+        };
+        let extra = result.rows.len() > page_size;
+        result.rows.truncate(page_size);
+        session.offset = session.offset.saturating_add(result.rows.len());
+        result.has_more = extra && session.offset < session.row_limit;
+        result.truncated = extra && !result.has_more;
+        if result.has_more {
+            session.last_activity = Instant::now();
+            session.expiry = CancellationToken::new();
+            let expiry = session.expiry.clone();
+            self.inner.query_sessions.lock().await.insert(id.clone(), session);
+            result.session_id = Some(id.clone());
+            self.expire_query_session(id, expiry);
+        } else {
+            self.drop_query_snapshot(&session).await;
+            result.session_id = None;
+        }
+        Ok(result)
+    }
+
+    pub async fn close_query_session(&self, id: &str) -> Result<bool, String> {
+        let session = self.inner.query_sessions.lock().await.remove(id);
+        if let Some(session) = session {
+            self.drop_query_snapshot(&session).await;
+            Ok(true)
+        } else {
+            Ok(false)
+        }
+    }
+
+    async fn drop_query_snapshot(&self, session: &QuerySession) {
+        session.expiry.cancel();
+        let process = self.inner.state.lock().await;
+        let alive = process.connected && process.generation == session.generation;
+        drop(process);
+        if !alive {
+            return;
+        }
+        let sql = format!("DROP TABLE IF EXISTS {}", session.qualified_table());
+        if let Err(error) = self.execute_interrupt_only(session.database.clone(), sql, Some(1), CLEANUP_TIMEOUT).await {
+            log::error!("Failed to remove DuckDB result snapshot {}: {error}", session.table);
+        }
+    }
+
+    fn expire_query_session(&self, id: String, expiry: CancellationToken) {
+        let weak = Arc::downgrade(&self.inner);
+        tokio::spawn(async move {
+            tokio::select! {
+                biased;
+                _ = expiry.cancelled() => return,
+                _ = tokio::time::sleep(IDLE_TIMEOUT) => {},
+            }
+            let Some(inner) = weak.upgrade() else {
+                return;
+            };
+            let client = DuckDbWorkerClient { inner };
+            let expired = {
+                let mut sessions = client.inner.query_sessions.lock().await;
+                if expiry.is_cancelled() {
+                    return;
+                }
+                sessions.remove(&id)
+            };
+            if let Some(session) = expired {
+                client.drop_query_snapshot(&session).await;
+            }
+        });
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn queued_client() -> DuckDbWorkerClient {
+        DuckDbWorkerClient::new_unconnected_with_timeouts(
+            PathBuf::from("must-not-start"),
+            vec![],
+            ":memory:".into(),
+            vec![],
+            None,
+            1,
+            Duration::from_secs(60),
+            Duration::from_secs(5),
+        )
+    }
+
+    #[tokio::test]
+    async fn cancelling_a_queued_read_does_not_interrupt_the_lock_owner() {
+        let client = queued_client();
+        let _owner = client.inner.query_lock.lock().await;
+        let cancel = CancellationToken::new();
+        let read = client.execute(None, "SELECT 1".into(), Some(1), Some(cancel.clone()), None);
+        tokio::pin!(read);
+        tokio::select! {
+            _ = &mut read => panic!("read must wait for the owner"),
+            _ = tokio::task::yield_now() => {},
+        }
+        cancel.cancel();
+        let error = tokio::time::timeout(Duration::from_secs(1), read).await.unwrap().unwrap_err();
+        assert_eq!(error, crate::execution::canceled_error());
+        assert_eq!(client.inner.state.lock().await.generation, 0);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn abandoned_result_sessions_expire_without_starting_a_worker() {
+        let client = queued_client();
+        let expiry = CancellationToken::new();
+        client.inner.query_sessions.lock().await.insert(
+            "idle".into(),
+            QuerySession {
+                table: "__dbx_result_idle".into(),
+                database: None,
+                columns: vec!["id".into()],
+                offset: 1,
+                row_limit: 100,
+                generation: 0,
+                last_activity: Instant::now(),
+                expiry: expiry.clone(),
+            },
+        );
+        client.expire_query_session("idle".into(), expiry);
+        tokio::task::yield_now().await;
+        tokio::time::advance(IDLE_TIMEOUT + Duration::from_secs(1)).await;
+        tokio::task::yield_now().await;
+        assert!(client.inner.query_sessions.lock().await.is_empty());
+        assert_eq!(client.inner.state.lock().await.generation, 0);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn cleanup_wait_is_bounded_even_while_the_worker_is_busy() {
+        let client = queued_client();
+        let _owner = client.inner.query_lock.lock().await;
+        let error = tokio::time::timeout(
+            CLEANUP_TIMEOUT + Duration::from_secs(1),
+            client.execute_interrupt_only(None, "DROP TABLE unused".into(), Some(1), CLEANUP_TIMEOUT),
+        )
+        .await
+        .expect("queued cleanup must stop waiting")
+        .unwrap_err();
+        assert_eq!(error, crate::execution::timeout_error_for(CLEANUP_TIMEOUT));
+        assert_eq!(client.inner.state.lock().await.generation, 0);
+    }
+}

crates/dbx-drivers/src/db/duckdb_worker_process.rs

@@ -40,6 +40,10 @@ const WORKER_INTERRUPT_ACK_TIMEOUT: Duration = Duration::from_secs(2);
 pub const DUCKDB_DRIVER_PATH_ENV: &str = "DBX_DUCKDB_DRIVER_PATH";
 type PendingRequests = Arc<Mutex<HashMap<String, PendingRequest>>>;
 
+#[path = "duckdb_query_sessions.rs"]
+mod query_sessions;
+use query_sessions::QuerySession;
+
 /// How a request reacts to cancellation or a timeout. Interactive queries
 /// restart the worker because a DuckDB connection stuck inside a statement
 /// cannot be reused. Statements whose caller shares the session (`:memory:`
@@ -65,6 +69,7 @@ struct DuckDbWorkerClientInner {
     pending: PendingRequests,
     connect_lock: Mutex<()>,
     query_lock: Mutex<()>,
+    query_sessions: Mutex<HashMap<String, QuerySession>>,
     process_limiter: Arc<Semaphore>,
     process_limit: usize,
     executable: PathBuf,
@@ -193,6 +198,7 @@ impl DuckDbWorkerClient {
                 pending: Arc::new(Mutex::new(HashMap::new())),
                 connect_lock: Mutex::new(()),
                 query_lock: Mutex::new(()),
+                query_sessions: Mutex::new(HashMap::new()),
                 process_limiter,
                 process_limit,
                 executable,
@@ -288,7 +294,28 @@ impl DuckDbWorkerClient {
         preserve_insertion_order: bool,
         abort_policy: AbortPolicy,
     ) -> Result<db::QueryResult, DuckDbWorkerError> {
-        let _query_guard = self.inner.query_lock.lock().await;
+        // A queued request must remain cancellable. In particular, cancelling
+        // a grid read waiting behind an export must not kill that export.
+        let lock = self.inner.query_lock.lock();
+        tokio::pin!(lock);
+        let _query_guard = match (cancel_token.as_ref(), query_timeout) {
+            (Some(token), Some(duration)) => tokio::select! {
+                biased;
+                _ = token.cancelled() => return Err(crate::execution::canceled_error().into()),
+                guard = &mut lock => guard,
+                _ = tokio::time::sleep(duration) => return Err(crate::execution::timeout_error_for(duration).into()),
+            },
+            (Some(token), None) => tokio::select! {
+                biased;
+                _ = token.cancelled() => return Err(crate::execution::canceled_error().into()),
+                guard = &mut lock => guard,
+            },
+            (None, Some(duration)) => tokio::select! {
+                guard = &mut lock => guard,
+                _ = tokio::time::sleep(duration) => return Err(crate::execution::timeout_error_for(duration).into()),
+            },
+            (None, None) => lock.await,
+        };
         let client = self.clone();
         // Cancellation and timeout apply `abort_policy` below. An ordinary
         // query error (parser/binder/catalog/runtime) leaves the worker healthy, so we keep it

crates/dbx-sql-data/src/query_result_sql.rs

@@ -214,7 +214,9 @@ pub fn build_query_pagination_execution_plan(
     // example Db2 for z/OS before V12). DBX connections use the Agent/JDBC
     // cursor for DB2, so keep every page on that cursor path instead of
     // generating OFFSET ... FETCH FIRST SQL that fails with SQLCODE -104.
-    if options.database_type == Some(DatabaseType::Db2) && options.use_agent_cursor {
+    // DuckDB must also keep offset replay on one result session: independent
+    // LIMIT/OFFSET executions can return overlapping rows even with ORDER BY.
+    if matches!(options.database_type, Some(DatabaseType::Db2 | DatabaseType::DuckDb)) && options.use_agent_cursor {
         if !options.first_page_uses_actual_sql && options.sql == options.query_base_sql {
             plan.sql_to_execute = options.query_base_sql;
         }
@@ -2967,6 +2969,29 @@ fn fallback_alias(index: usize) -> String {
 
 #[cfg(test)]
 mod tests {
+    #[test]
+    fn duckdb_pages_and_offset_replays_keep_a_single_result_session() {
+        for offset in [0, 100, 500] {
+            for session_id in [None, Some("duckdb-result".to_string())] {
+                let sql = "SELECT * FROM settlements ORDER BY period";
+                let plan = build_query_pagination_execution_plan(QueryPaginationExecutionPlanOptions {
+                    sql: sql.into(),
+                    query_base_sql: sql.into(),
+                    database_type: Some(DatabaseType::DuckDb),
+                    pagination: QueryPagination { limit: 100, offset, session_id },
+                    use_agent_cursor: true,
+                    first_page_uses_actual_sql: true,
+                    execution_hint: None,
+                });
+                assert!(plan.use_agent_result_session);
+                assert_eq!(plan.sql_to_execute, sql);
+                assert_eq!(plan.page_offset, Some(offset));
+                assert_eq!(plan.page_limit, Some(100));
+                assert!(plan.pagination_error.is_none());
+            }
+        }
+    }
+
     #[test]
     fn dynamodb_partiql_uses_api_bounds_without_sql_rewrites() {
         for sql in [