feat(es): paginate past 10k documents via pit and search_after

apps/desktop/src/components/document/DocumentBrowser.vue

@@ -22,7 +22,7 @@ import type { DynamoDbIndexInfo, DynamoDbTableDescription } from "@/lib/backend/
 import { useConnectionStore } from "@/stores/connectionStore";
 import { clampSearchSplitWidth } from "@/lib/dataGrid/dataGridSearchSplit";
 import { documentViewerFontStyle } from "@/lib/document/documentViewerFontStyle";
-import { clampDocumentPage, clampElasticsearchRequestLimit, documentPageRequestLimit, resetElasticsearchDocumentTotals, resolveElasticsearchDocumentTotals } from "@/lib/document/elasticsearchDocumentTotals";
+import { ELASTICSEARCH_DEFAULT_MAX_RESULT_WINDOW, clampDocumentPage, resetElasticsearchDocumentTotals, resolveElasticsearchDocumentTotals } from "@/lib/document/elasticsearchDocumentTotals";
 import { canGoNextDocumentPage, isSameDocumentQueryTotalCountRequest, resolveDocumentQueryTotals, type DocumentQueryTotalCountRequest } from "@/lib/document/documentQueryTotals";
 import {
   arrayObjectAncestorPathForDocumentField,
@@ -238,6 +238,7 @@ const pageTotal = computed(() => paginationTotal.value);
 const documentPageCount = computed(() => (pageTotal.value === undefined ? undefined : Math.max(1, Math.ceil(pageTotal.value / pageSize.value))));
 const canGoNextPage = computed(() => {
   if (documentStoreProvider.value.kind === "dynamodb") return dynamodbHasNextCursor.value;
+  if (documentStoreProvider.value.kind === "elasticsearch") return elasticsearchHasNextCursor.value;
   return canGoNextDocumentPage({
     page: page.value,
     pageSize: pageSize.value,
@@ -247,7 +248,7 @@ const canGoNextPage = computed(() => {
 });
 const documentRequestLimit = computed(() => {
   if (documentStoreProvider.value.kind !== "elasticsearch") return pageSize.value;
-  return documentPageRequestLimit(page.value, pageSize.value, paginationTotal.value);
+  return Math.min(pageSize.value, ELASTICSEARCH_DEFAULT_MAX_RESULT_WINDOW);
 });
 
 const tableFindPaneStyle = computed(() => {
@@ -289,6 +290,8 @@ const dynamodbIndexName = ref("__table__");
 const dynamodbPageCursors = ref<Array<string | undefined>>([undefined]);
 const dynamodbHasNextCursor = ref(false);
 const dynamodbExactTotal = ref<number | undefined>();
+const elasticsearchPageCursors = ref<Array<string | undefined>>([undefined]);
+const elasticsearchHasNextCursor = ref(false);
 let dynamodbExactCountKey: string | null = null;
 
 const dynamodbIndexOptions = computed<Array<{ value: string; label: string; index?: DynamoDbIndexInfo }>>(() => [
@@ -440,7 +443,7 @@ const gridResult = computed<QueryResult>(() => {
 
 async function exportAllDocumentStoreDocuments(onProgress?: (info: { rowsExported: number; totalRows: number | null }) => void): Promise<QueryResult | undefined> {
   const kind = documentStoreProvider.value.kind;
-  if (kind !== "mongodb" && kind !== "dynamodb") return undefined;
+  if (kind !== "mongodb" && kind !== "dynamodb" && kind !== "elasticsearch") return undefined;
 
   const connectionId = props.connectionId;
   const database = props.database;
@@ -456,44 +459,53 @@ async function exportAllDocumentStoreDocuments(onProgress?: (info: { rowsExporte
   let exportedCopyDocuments: JsonRecord[] | undefined = kind === "mongodb" ? [] : undefined;
   let totalRows: number | null = null;
   let cursor: string | undefined;
+  let lastCursor: string | undefined;
   const seenCursors = new Set<string>();
 
-  while (exportedDocuments.length < rowLimit) {
-    const requestLimit = Math.min(batchSize, kind === "dynamodb" ? 1000 : Number.POSITIVE_INFINITY, rowLimit - exportedDocuments.length);
-    if (requestLimit <= 0) break;
-    const result = await api.documentFindDocuments(connectionId, database, collection, kind === "dynamodb" ? 0 : exportedDocuments.length, requestLimit, filter, undefined, sort, undefined, exportExecutionId, cursor);
-    const pageDocuments = result.documents.slice(0, requestLimit).map(asRecord);
-    exportedDocuments.push(...pageDocuments);
+  try {
+    while (exportedDocuments.length < rowLimit) {
+      const requestLimit = Math.min(batchSize, kind === "dynamodb" ? 1000 : kind === "elasticsearch" ? ELASTICSEARCH_DEFAULT_MAX_RESULT_WINDOW : Number.POSITIVE_INFINITY, rowLimit - exportedDocuments.length);
+      if (requestLimit <= 0) break;
+      if (cursor) lastCursor = cursor;
+      const result = await api.documentFindDocuments(connectionId, database, collection, kind === "dynamodb" || kind === "elasticsearch" ? 0 : exportedDocuments.length, requestLimit, filter, undefined, sort, undefined, exportExecutionId, cursor, kind === "elasticsearch");
+      const pageDocuments = result.documents.slice(0, requestLimit).map(asRecord);
+      exportedDocuments.push(...pageDocuments);
 
-    if (kind === "mongodb" && exportedCopyDocuments) {
-      if (result.extended_documents?.length === result.documents.length) {
-        exportedCopyDocuments.push(...result.extended_documents.slice(0, pageDocuments.length).map(asRecord));
-      } else {
-        exportedCopyDocuments = undefined;
+      if (kind === "mongodb" && exportedCopyDocuments) {
+        if (result.extended_documents?.length === result.documents.length) {
+          exportedCopyDocuments.push(...result.extended_documents.slice(0, pageDocuments.length).map(asRecord));
+        } else {
+          exportedCopyDocuments = undefined;
+        }
       }
+
+      if ((kind === "mongodb" || kind === "elasticsearch") && result.total_is_exact !== false) totalRows = Math.min(result.total, rowLimit);
+      onProgress?.({ rowsExported: exportedDocuments.length, totalRows });
+
+      if (kind === "dynamodb" || kind === "elasticsearch") {
+        cursor = result.next_cursor;
+        if (!cursor) break;
+        if (seenCursors.has(cursor)) throw new Error(kind === "dynamodb" ? t("dynamodb.repeatedCursor") : "Elasticsearch cursor repeated during export");
+        seenCursors.add(cursor);
+        continue;
+      }
+
+      const reachedExactTotal = result.total_is_exact !== false && exportedDocuments.length >= result.total;
+      if (pageDocuments.length === 0 || pageDocuments.length < requestLimit || reachedExactTotal) break;
     }
-
-    if (kind === "mongodb" && result.total_is_exact !== false) totalRows = Math.min(result.total, rowLimit);
-    onProgress?.({ rowsExported: exportedDocuments.length, totalRows });
-
-    if (kind === "dynamodb") {
-      cursor = result.next_cursor;
-      if (!cursor) break;
-      if (seenCursors.has(cursor)) throw new Error(t("dynamodb.repeatedCursor"));
-      seenCursors.add(cursor);
-      continue;
+  } finally {
+    if (kind === "elasticsearch") {
+      const cursorToClose = lastCursor ?? cursor;
+      if (cursorToClose) void closeElasticsearchCursor(cursorToClose);
     }
-
-    const reachedExactTotal = result.total_is_exact !== false && exportedDocuments.length >= result.total;
-    if (pageDocuments.length === 0 || pageDocuments.length < requestLimit || reachedExactTotal) break;
   }
 
-  if (kind === "dynamodb") {
+  if (kind === "dynamodb" || kind === "elasticsearch") {
     const truncatedByLimit = !!cursor && exportedDocuments.length >= rowLimit;
     totalRows = truncatedByLimit ? null : exportedDocuments.length;
     onProgress?.({ rowsExported: exportedDocuments.length, totalRows });
     if (truncatedByLimit) {
-      toast(t("dynamodb.exportLimitReached", { count: rowLimit }), 6000);
+      toast(kind === "dynamodb" ? t("dynamodb.exportLimitReached", { count: rowLimit }) : `Elasticsearch export limit reached (${rowLimit})`, 6000);
     }
   }
 
@@ -501,7 +513,7 @@ async function exportAllDocumentStoreDocuments(onProgress?: (info: { rowsExporte
   if (result.columns.length === 0) result.columns = gridResult.value.columns;
   result.column_types = kind === "mongodb" ? mongoDocumentGridColumnTypes(exportedDocuments, result.columns) : undefined;
   result.affected_rows = exportedDocuments.length;
-  result.truncated = kind === "dynamodb" && !!cursor && exportedDocuments.length >= rowLimit;
+  result.truncated = (kind === "dynamodb" || kind === "elasticsearch") && !!cursor && exportedDocuments.length >= rowLimit;
   result.has_more = result.truncated;
   return result;
 }
@@ -744,6 +756,22 @@ function resetDynamoDbPagination(options: { preserveExactCount?: boolean } = {})
   if (!options.preserveExactCount) resetDynamoDbExactCount();
 }
 
+async function closeElasticsearchCursor(cursor?: string) {
+  if (!cursor) return;
+  try {
+    await api.closeQuerySession(props.connectionId, props.database, cursor);
+  } catch (error) {
+    console.warn("[DBX] failed to close Elasticsearch cursor", error);
+  }
+}
+
+function resetElasticsearchPagination() {
+  const cursor = [...elasticsearchPageCursors.value].reverse().find((candidate): candidate is string => !!candidate);
+  if (cursor) void closeElasticsearchCursor(cursor);
+  elasticsearchPageCursors.value = [undefined];
+  elasticsearchHasNextCursor.value = false;
+}
+
 function currentDocumentFilter(): string | undefined {
   const filter = currentDocumentFilterJson(filterInput.value, appliedDocumentFilter.value, documentStoreProvider.value.kind);
   if (documentStoreProvider.value.kind !== "dynamodb" || dynamodbIndexName.value === "__table__") return filter;
@@ -808,7 +836,7 @@ const documentQueryPreview = computed(() => {
     collection: props.collection,
     filterJson: filter,
     sortJson: sortInput.value.trim(),
-    skip: page.value * pageSize.value,
+    skip: documentStoreProvider.value.kind === "elasticsearch" ? 0 : page.value * pageSize.value,
     limit: documentRequestLimit.value,
   });
 });
@@ -1248,6 +1276,7 @@ function resetElasticsearchTotals(options: { preservePaginationTotal?: boolean }
   paginationTotal.value = nextTotals.paginationTotal;
   total.value = nextTotals.total;
   totalIsExact.value = nextTotals.totalIsExact;
+  resetElasticsearchPagination();
 }
 
 function clampPageToPaginationTotal(): number | undefined {
@@ -1352,16 +1381,18 @@ async function load(options: { page?: number; append?: boolean; offset?: number;
       resetElasticsearchTotals();
     }
     const sort = currentDocumentSortJson(sortInput.value);
-    const cursor = storeKind === "dynamodb" ? dynamodbPageCursors.value[requestPage] : undefined;
-    if (storeKind === "dynamodb" && requestPage > 0 && !cursor) {
-      throw new Error(t("dynamodb.pageCursorUnavailable"));
+    const cursor = storeKind === "dynamodb" ? dynamodbPageCursors.value[requestPage] : storeKind === "elasticsearch" ? elasticsearchPageCursors.value[requestPage] : undefined;
+    if ((storeKind === "dynamodb" || storeKind === "elasticsearch") && requestPage > 0 && !cursor) {
+      throw new Error(storeKind === "dynamodb" ? t("dynamodb.pageCursorUnavailable") : "Elasticsearch page cursor unavailable; go back to the first page and page forward again");
     }
-    const skip = storeKind === "dynamodb" ? 0 : (options.offset ?? requestPage * pageSize.value);
-    const requestedLimit = options.limit ?? (storeKind === "elasticsearch" ? documentPageRequestLimit(requestPage, pageSize.value, paginationTotal.value) : pageSize.value);
-    // Elasticsearch rejects from+size beyond index.max_result_window even when an explicit limit
-    // (e.g. from the per-tab page size control) bypassed the default-path clamp above.
-    const requestLimit = storeKind === "elasticsearch" ? clampElasticsearchRequestLimit(skip, requestedLimit) : requestedLimit;
-    const result = await api.documentFindDocuments(connectionId, database, collection, skip, requestLimit, filter, undefined, sort, undefined, executionId, cursor);
+    // Starting a fresh ES first page invalidates any previous PIT cursor stack.
+    if (storeKind === "elasticsearch" && !cursor && requestPage === 0) {
+      resetElasticsearchPagination();
+    }
+    const skip = storeKind === "dynamodb" || storeKind === "elasticsearch" ? 0 : (options.offset ?? requestPage * pageSize.value);
+    const requestedLimit = options.limit ?? pageSize.value;
+    const requestLimit = storeKind === "elasticsearch" ? Math.min(requestedLimit, ELASTICSEARCH_DEFAULT_MAX_RESULT_WINDOW) : requestedLimit;
+    const result = await api.documentFindDocuments(connectionId, database, collection, skip, requestLimit, filter, undefined, sort, undefined, executionId, cursor, storeKind === "elasticsearch");
     if (documentLoadExecutionId.value !== executionId) return;
     if (connectionId !== props.connectionId || database !== props.database || collection !== props.collection || storeKind !== documentStoreProvider.value.kind) return;
     const nextDocuments =
@@ -1387,6 +1418,10 @@ async function load(options: { page?: number; append?: boolean; offset?: number;
       dynamodbHasNextCursor.value = !!result.next_cursor;
     }
     if (storeKind === "elasticsearch") {
+      const nextCursors = elasticsearchPageCursors.value.slice(0, requestPage + 1);
+      nextCursors[requestPage + 1] = result.next_cursor ?? undefined;
+      elasticsearchPageCursors.value = nextCursors;
+      elasticsearchHasNextCursor.value = !!result.next_cursor;
       applyElasticsearchSearchTotal(result.total, result.total_is_exact !== false, filter);
     } else if (storeKind === "dynamodb") {
       cancelElasticsearchCount();
@@ -1501,28 +1536,34 @@ async function paginate(offset: number, limit: number) {
   }
   const pageSizeChanged = normalizedLimit !== pageSize.value;
   pageSize.value = normalizedLimit;
-  if (documentStoreProvider.value.kind === "dynamodb" && pageSizeChanged) {
+  if (pageSizeChanged && (documentStoreProvider.value.kind === "dynamodb" || documentStoreProvider.value.kind === "elasticsearch")) {
     page.value = 0;
-    resetDynamoDbPagination({ preserveExactCount: true });
+    if (documentStoreProvider.value.kind === "dynamodb") {
+      resetDynamoDbPagination({ preserveExactCount: true });
+    } else {
+      resetElasticsearchPagination();
+    }
     await load({ page: 0 });
     return;
   }
   const requestedPage = Math.floor(normalizedOffset / normalizedLimit);
   const nextPage = clampDocumentPage(requestedPage, normalizedLimit, paginationTotal.value);
-  if (documentStoreProvider.value.kind !== "dynamodb") {
+  if (documentStoreProvider.value.kind !== "dynamodb" && documentStoreProvider.value.kind !== "elasticsearch") {
     await load({ page: nextPage, offset: nextPage * normalizedLimit, limit: normalizedLimit });
     return;
   }
   for (let cursorPage = 0; cursorPage <= nextPage; cursorPage += 1) {
-    if (cursorPage > 0 && !dynamodbPageCursors.value[cursorPage]) {
-      error.value = t("dynamodb.pageCursorUnavailable");
+    const pageCursor = documentStoreProvider.value.kind === "dynamodb" ? dynamodbPageCursors.value[cursorPage] : elasticsearchPageCursors.value[cursorPage];
+    if (cursorPage > 0 && !pageCursor) {
+      error.value = documentStoreProvider.value.kind === "dynamodb" ? t("dynamodb.pageCursorUnavailable") : "Elasticsearch page cursor unavailable; go back to the first page and page forward again";
       return;
     }
     if (cursorPage === nextPage) {
       await load({ page: cursorPage });
       return;
     }
-    if (!dynamodbPageCursors.value[cursorPage + 1]) {
+    const nextCursor = documentStoreProvider.value.kind === "dynamodb" ? dynamodbPageCursors.value[cursorPage + 1] : elasticsearchPageCursors.value[cursorPage + 1];
+    if (!nextCursor) {
       await load({ page: cursorPage });
     }
   }
@@ -2041,6 +2082,7 @@ onBeforeUnmount(() => {
   documentRequestGeneration++;
   loadedDocumentQueryTotalCountRequest = undefined;
   cancelElasticsearchCount();
+  resetElasticsearchPagination();
   stopDocumentLoadingTimer();
   endTableSearchSplitResize();
 });
@@ -2317,7 +2359,7 @@ defineExpose({ focusSearch });
       :inexact-total-row-count-mode="documentStoreProvider.kind === 'mongodb' ? 'estimated' : 'at-least'"
       :pagination-total-row-count="pageTotal"
       :count-total-rows="countExactDocumentTotal"
-      :full-export-result="documentStoreProvider.kind === 'mongodb' || documentStoreProvider.kind === 'dynamodb' ? exportAllDocumentStoreDocuments : undefined"
+      :full-export-result="documentStoreProvider.kind === 'mongodb' || documentStoreProvider.kind === 'dynamodb' || documentStoreProvider.kind === 'elasticsearch' ? exportAllDocumentStoreDocuments : undefined"
       @sort="onSort"
       @reload="refreshDocuments"
       @paginate="(offset: number, limit: number) => paginate(offset, limit)"

apps/desktop/src/components/document/__tests__/DocumentBrowserFieldSearch.spec.ts

@@ -14,6 +14,7 @@ const backend = vi.hoisted(() => ({
   documentUpdateDocument: vi.fn(),
   documentDeleteDocument: vi.fn(),
   documentSaveMeilisearchBatch: vi.fn(),
+  closeQuerySession: vi.fn(),
 }));
 
 const documentJsonEditor = vi.hoisted(() => ({
@@ -79,6 +80,7 @@ vi.mock("@/lib/backend/api", () => ({
   documentUpdateDocument: backend.documentUpdateDocument,
   documentDeleteDocument: backend.documentDeleteDocument,
   documentSaveMeilisearchBatch: backend.documentSaveMeilisearchBatch,
+  closeQuerySession: backend.closeQuerySession,
 }));
 
 vi.mock("@/stores/connectionStore", () => ({
@@ -315,6 +317,7 @@ beforeEach(async () => {
   backend.documentUpdateDocument.mockReset();
   backend.documentDeleteDocument.mockReset();
   backend.documentSaveMeilisearchBatch.mockReset();
+  backend.closeQuerySession.mockReset();
   dataGrid.fullExportResult = undefined;
   dataGrid.customSaveHandler = undefined;
   dataGrid.countTotalRows = undefined;
@@ -889,8 +892,39 @@ describe("DocumentBrowser MongoDB filter value types", () => {
     expect(progress).toHaveBeenLastCalledWith({ rowsExported: 1, totalRows: null });
   });
 
-  it("does not offer the MongoDB full export callback to Elasticsearch viewers", () => {
-    expect(dataGrid.fullExportResult).toBeUndefined();
+  it("exports Elasticsearch documents with cursor pagination", async () => {
+    settings.editorSettings.exportBatchSize = 2;
+    backend.documentFindDocuments.mockReset();
+    backend.documentFindDocuments
+      .mockResolvedValueOnce({
+        documents: [
+          { _id: "one", title: "First" },
+          { _id: "two", title: "Second" },
+        ],
+        total: 3,
+        total_is_exact: true,
+        next_cursor: "cursor-1",
+      })
+      .mockResolvedValueOnce({
+        documents: [{ _id: "three", title: "Third" }],
+        total: 3,
+        total_is_exact: true,
+      });
+
+    expect(dataGrid.fullExportResult).toBeTypeOf("function");
+    const result = await dataGrid.fullExportResult!();
+    await flushUi();
+
+    expect(backend.documentFindDocuments.mock.calls.map((call) => [call[3], call[4], call[10], call[11]])).toEqual([
+      [0, 2, undefined, true],
+      [0, 2, "cursor-1", true],
+    ]);
+    expect(result?.rows).toEqual([
+      ["one", "First"],
+      ["two", "Second"],
+      ["three", "Third"],
+    ]);
+    expect(backend.closeQuerySession).toHaveBeenCalledWith("connection-1", "", "cursor-1");
   });
 
   it("identifies consistently numeric MongoDB columns for shared grid alignment", async () => {

apps/desktop/src/components/document/__tests__/DocumentBrowserInfiniteScroll.spec.ts

@@ -260,6 +260,10 @@ describe("DocumentBrowser infinite scroll (issue #6455)", () => {
     await mountBrowser(databaseType);
 
     expect(backend.documentFindDocuments.mock.calls[0]!.slice(0, 5)).toEqual(["mongo-1", "test", "docs", 0, 5]);
+    if (databaseType === "elasticsearch") {
+      expect(backend.documentFindDocuments.mock.calls[0]![10]).toBeUndefined();
+      expect(backend.documentFindDocuments.mock.calls[0]![11]).toBe(true);
+    }
     expect(dataGrid.pageLimit).toBe(5);
     expect(dataGrid.pageSizePreference).toBe("table-open");
   });