fix(elasticsearch): correct totals and CAT JSON output
This commit is contained in:
onenewcode
GitHub
parent
a8e5e8d8f9
commit
f07509b5ad
|
|
@ -111,7 +111,7 @@ onMounted(() => {
|
|||
</Button>
|
||||
</header>
|
||||
<div class="min-h-0 flex-1 overflow-hidden bg-background p-4">
|
||||
<pre v-show="responseView === 'raw' || !parsedBody.valid" class="m-0 h-full overflow-auto bg-transparent p-0 font-mono text-sm leading-6 whitespace-pre-wrap break-words">{{ body }}</pre>
|
||||
<pre v-show="responseView === 'raw' || !parsedBody.valid" class="m-0 h-full overflow-auto bg-transparent p-0 font-mono text-sm leading-6 whitespace-pre">{{ body }}</pre>
|
||||
<div v-if="parsedBody.valid" v-show="responseView === 'json'" class="h-full min-h-0">
|
||||
<JsonTree ref="jsonTreeRef" :value="parsedBody.value" :highlight-json="highlightJson" :virtualized="true" class="dbx-editor-font-family text-sm leading-6" />
|
||||
</div>
|
||||
|
|
|
|||
|
|
@ -16,6 +16,7 @@ import * as api from "@/lib/backend/api";
|
|||
import { useConnectionStore } from "@/stores/connectionStore";
|
||||
import { clampSearchSplitWidth } from "@/lib/dataGrid/dataGridSearchSplit";
|
||||
import { documentViewerFontStyle } from "@/lib/document/documentViewerFontStyle";
|
||||
import { clampDocumentPage, documentPageRequestLimit, resetElasticsearchDocumentTotals, resolveElasticsearchDocumentTotals } from "@/lib/document/elasticsearchDocumentTotals";
|
||||
import {
|
||||
arrayObjectAncestorPathForDocumentField,
|
||||
buildDocumentFilterCondition,
|
||||
|
|
@ -84,7 +85,9 @@ type ViewMode = "document" | "table";
|
|||
const documents = ref<JsonRecord[]>([]);
|
||||
const copyDocuments = ref<JsonRecord[]>([]);
|
||||
const lastGridColumns = ref<string[]>([]);
|
||||
const total = ref(0);
|
||||
const total = ref<number | undefined>(undefined);
|
||||
const totalIsExact = ref(true);
|
||||
const paginationTotal = ref<number | undefined>(undefined);
|
||||
const loading = ref(false);
|
||||
const documentLoadExecutionId = ref("");
|
||||
const documentLoadCancelling = ref(false);
|
||||
|
|
@ -123,15 +126,26 @@ const tableFindPaneWidth = ref<number | null>(null);
|
|||
const isResizingTableSearchSplit = ref(false);
|
||||
let tableSearchSplitStartX = 0;
|
||||
let tableSearchSplitStartWidth = 0;
|
||||
let elasticsearchCountKey: string | null = null;
|
||||
let elasticsearchExactTotal: number | undefined;
|
||||
let elasticsearchPaginationLowerBound: number | undefined;
|
||||
let elasticsearchCountExecutionId = "";
|
||||
let elasticsearchCountGeneration = 0;
|
||||
const documentStoreProvider = computed(() => documentStoreProviderFor(props.databaseType));
|
||||
|
||||
const pageTotal = computed(() => paginationTotal.value ?? total.value ?? 0);
|
||||
const documentRequestLimit = computed(() => {
|
||||
if (documentStoreProvider.value.kind !== "elasticsearch" || paginationTotal.value === undefined) return pageSize.value;
|
||||
return documentPageRequestLimit(page.value, pageSize.value, paginationTotal.value);
|
||||
});
|
||||
|
||||
const tableFindPaneStyle = computed(() => {
|
||||
if (tableFindPaneWidth.value == null) return {};
|
||||
return { flex: `0 0 ${tableFindPaneWidth.value}px` };
|
||||
});
|
||||
const documentFontStyle = computed(() => documentViewerFontStyle(settingsStore.editorSettings));
|
||||
const documentStoreLabels = computed(() => ({
|
||||
documentsLabel: documentStoreProvider.value.documentsLabel({ total: total.value, t }),
|
||||
documentsLabel: documentStoreProvider.value.documentsLabel({ total: total.value ?? 0, t }),
|
||||
queryPreview: documentQueryPreview.value,
|
||||
}));
|
||||
|
||||
|
|
@ -436,7 +450,7 @@ const documentQueryPreview = computed(() => {
|
|||
filterJson: filter,
|
||||
sortJson: sortInput.value.trim(),
|
||||
skip: page.value * pageSize.value,
|
||||
limit: pageSize.value,
|
||||
limit: documentRequestLimit.value,
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -580,6 +594,7 @@ async function gridSave(changes: DocumentGridChanges) {
|
|||
await api.documentInsertDocument(props.connectionId, props.database, props.collection, JSON.stringify(doc));
|
||||
}
|
||||
|
||||
if (isEs) resetElasticsearchTotals({ preservePaginationTotal: true });
|
||||
await load();
|
||||
}
|
||||
|
||||
|
|
@ -684,6 +699,100 @@ function startDocumentLoadingTimer() {
|
|||
}, 100);
|
||||
}
|
||||
|
||||
function elasticsearchCountFilterKey(filter: string | undefined): string {
|
||||
return JSON.stringify([props.connectionId, props.database, props.collection, filter ?? ""]);
|
||||
}
|
||||
|
||||
function cancelElasticsearchCount() {
|
||||
elasticsearchCountGeneration++;
|
||||
const executionId = elasticsearchCountExecutionId;
|
||||
elasticsearchCountExecutionId = "";
|
||||
if (executionId) void api.cancelQuery(executionId);
|
||||
}
|
||||
|
||||
function resetElasticsearchTotals(options: { preservePaginationTotal?: boolean } = {}) {
|
||||
const nextTotals = resetElasticsearchDocumentTotals(paginationTotal.value, options.preservePaginationTotal);
|
||||
cancelElasticsearchCount();
|
||||
elasticsearchCountKey = null;
|
||||
elasticsearchExactTotal = undefined;
|
||||
elasticsearchPaginationLowerBound = undefined;
|
||||
paginationTotal.value = nextTotals.paginationTotal;
|
||||
total.value = nextTotals.total;
|
||||
totalIsExact.value = nextTotals.totalIsExact;
|
||||
}
|
||||
|
||||
function clampPageToPaginationTotal(): boolean {
|
||||
const cap = paginationTotal.value;
|
||||
if (cap === undefined) return false;
|
||||
const nextPage = clampDocumentPage(page.value, pageSize.value, cap);
|
||||
if (page.value === nextPage) return false;
|
||||
page.value = nextPage;
|
||||
return true;
|
||||
}
|
||||
|
||||
function startElasticsearchExactCount(filter: string | undefined) {
|
||||
if (elasticsearchCountExecutionId || elasticsearchExactTotal !== undefined || !elasticsearchCountKey) return;
|
||||
const key = elasticsearchCountKey;
|
||||
const executionId = uuid();
|
||||
const generation = elasticsearchCountGeneration;
|
||||
elasticsearchCountExecutionId = executionId;
|
||||
|
||||
void api
|
||||
.elasticsearchCountDocuments(props.connectionId, props.collection, filter, executionId)
|
||||
.then((exactCount) => {
|
||||
if (generation !== elasticsearchCountGeneration || key !== elasticsearchCountKey || executionId !== elasticsearchCountExecutionId || !Number.isFinite(exactCount) || exactCount < 0) {
|
||||
return;
|
||||
}
|
||||
elasticsearchExactTotal = exactCount;
|
||||
const totals = resolveElasticsearchDocumentTotals(elasticsearchPaginationLowerBound ?? exactCount, false, exactCount);
|
||||
total.value = totals.total;
|
||||
totalIsExact.value = totals.totalIsExact;
|
||||
paginationTotal.value = totals.paginationTotal;
|
||||
if (clampPageToPaginationTotal()) void load();
|
||||
})
|
||||
.catch(() => {
|
||||
// The lower-bound result remains truthful when a background count fails.
|
||||
})
|
||||
.finally(() => {
|
||||
if (generation === elasticsearchCountGeneration && executionId === elasticsearchCountExecutionId) {
|
||||
elasticsearchCountExecutionId = "";
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
function applyElasticsearchSearchTotal(searchTotal: number, isExact: boolean, filter: string | undefined) {
|
||||
const key = elasticsearchCountFilterKey(filter);
|
||||
if (key !== elasticsearchCountKey) {
|
||||
cancelElasticsearchCount();
|
||||
elasticsearchCountKey = key;
|
||||
elasticsearchExactTotal = undefined;
|
||||
elasticsearchPaginationLowerBound = undefined;
|
||||
}
|
||||
|
||||
elasticsearchPaginationLowerBound = searchTotal;
|
||||
const totals = resolveElasticsearchDocumentTotals(searchTotal, isExact, elasticsearchExactTotal);
|
||||
if (isExact) {
|
||||
cancelElasticsearchCount();
|
||||
elasticsearchExactTotal = searchTotal;
|
||||
total.value = totals.total;
|
||||
totalIsExact.value = totals.totalIsExact;
|
||||
paginationTotal.value = totals.paginationTotal;
|
||||
return;
|
||||
}
|
||||
|
||||
if (elasticsearchExactTotal !== undefined) {
|
||||
total.value = totals.total;
|
||||
totalIsExact.value = totals.totalIsExact;
|
||||
paginationTotal.value = totals.paginationTotal;
|
||||
return;
|
||||
}
|
||||
|
||||
total.value = totals.total;
|
||||
totalIsExact.value = totals.totalIsExact;
|
||||
paginationTotal.value = totals.paginationTotal;
|
||||
startElasticsearchExactCount(filter);
|
||||
}
|
||||
|
||||
async function load() {
|
||||
if (documentLoadExecutionId.value) void api.cancelQuery(documentLoadExecutionId.value);
|
||||
const executionId = uuid();
|
||||
|
|
@ -696,8 +805,12 @@ async function load() {
|
|||
const previousSelectedId = previousSelectedIdx === null ? null : documentIdentity(documents.value[previousSelectedIdx]);
|
||||
try {
|
||||
const filter = currentDocumentFilter();
|
||||
if (documentStoreProvider.value.kind === "elasticsearch" && elasticsearchCountKey !== null && elasticsearchCountKey !== elasticsearchCountFilterKey(filter)) {
|
||||
resetElasticsearchTotals();
|
||||
}
|
||||
const sort = currentDocumentSortJson(sortInput.value);
|
||||
const result = await api.documentFindDocuments(props.connectionId, props.database, props.collection, page.value * pageSize.value, pageSize.value, filter, undefined, sort, executionId);
|
||||
const skip = page.value * pageSize.value;
|
||||
const result = await api.documentFindDocuments(props.connectionId, props.database, props.collection, skip, documentRequestLimit.value, filter, undefined, sort, executionId);
|
||||
if (documentLoadExecutionId.value !== executionId) return;
|
||||
const nextDocuments =
|
||||
documentStoreProvider.value.kind === "elasticsearch" && result.raw_documents?.length === result.documents.length
|
||||
|
|
@ -722,7 +835,14 @@ async function load() {
|
|||
}
|
||||
lastGridColumns.value = [...keySet];
|
||||
}
|
||||
total.value = result.total;
|
||||
if (documentStoreProvider.value.kind === "elasticsearch") {
|
||||
applyElasticsearchSearchTotal(result.total, result.total_is_exact !== false, filter);
|
||||
} else {
|
||||
cancelElasticsearchCount();
|
||||
total.value = result.total;
|
||||
totalIsExact.value = true;
|
||||
paginationTotal.value = result.total;
|
||||
}
|
||||
syncSelectedDocumentAfterLoad(previousSelectedIdx, previousSelectedId);
|
||||
} catch (e: unknown) {
|
||||
if (documentLoadExecutionId.value === executionId) error.value = e instanceof Error ? e.message : String(e);
|
||||
|
|
@ -736,6 +856,11 @@ async function load() {
|
|||
}
|
||||
}
|
||||
|
||||
async function refreshDocuments() {
|
||||
if (documentStoreProvider.value.kind === "elasticsearch") resetElasticsearchTotals({ preservePaginationTotal: true });
|
||||
await load();
|
||||
}
|
||||
|
||||
async function cancelDocumentLoad() {
|
||||
const executionId = documentLoadExecutionId.value;
|
||||
if (!executionId || documentLoadCancelling.value) return;
|
||||
|
|
@ -754,20 +879,22 @@ async function cancelDocumentLoad() {
|
|||
|
||||
function applyFilter() {
|
||||
page.value = 0;
|
||||
load();
|
||||
if (documentStoreProvider.value.kind === "elasticsearch") resetElasticsearchTotals();
|
||||
void load();
|
||||
}
|
||||
|
||||
function paginate(offset: number, limit: number) {
|
||||
const normalizedLimit = normalizeResultPageSize(limit, pageSize.value);
|
||||
pageSize.value = normalizedLimit;
|
||||
page.value = Math.floor(Math.max(0, offset) / normalizedLimit);
|
||||
load();
|
||||
const requestedPage = Math.floor(Math.max(0, offset) / normalizedLimit);
|
||||
page.value = clampDocumentPage(requestedPage, normalizedLimit, paginationTotal.value);
|
||||
void load();
|
||||
}
|
||||
|
||||
function onSort(column: string, _columnIndex: number, direction: "asc" | "desc" | null) {
|
||||
sortInput.value = documentStoreProvider.value.sortInputForColumn(column, direction);
|
||||
page.value = 0;
|
||||
load();
|
||||
void load();
|
||||
}
|
||||
|
||||
function asRecord(value: unknown): JsonRecord {
|
||||
|
|
@ -1129,6 +1256,7 @@ async function saveDoc() {
|
|||
isNew.value = false;
|
||||
documentEditMode.value = "fields";
|
||||
editFields.value = [];
|
||||
if (kind === "elasticsearch") resetElasticsearchTotals({ preservePaginationTotal: true });
|
||||
await load();
|
||||
if (selectedIdx.value !== null && documents.value[selectedIdx.value]) {
|
||||
editJson.value = stringifyDocumentStoreValue(documents.value[selectedIdx.value], documentStoreProvider.value.kind, 2);
|
||||
|
|
@ -1151,6 +1279,7 @@ async function applyDeleteDoc(idx: number) {
|
|||
selectedIdx.value = null;
|
||||
editJson.value = "";
|
||||
}
|
||||
if (documentStoreProvider.value.kind === "elasticsearch") resetElasticsearchTotals({ preservePaginationTotal: true });
|
||||
await load();
|
||||
} catch (e: unknown) {
|
||||
error.value = e instanceof Error ? e.message : String(e);
|
||||
|
|
@ -1176,13 +1305,13 @@ async function confirmDelete() {
|
|||
function prevPage() {
|
||||
if (page.value <= 0) return;
|
||||
page.value--;
|
||||
load();
|
||||
void load();
|
||||
}
|
||||
|
||||
function nextPage() {
|
||||
if ((page.value + 1) * pageSize.value >= total.value) return;
|
||||
if ((page.value + 1) * pageSize.value >= pageTotal.value) return;
|
||||
page.value++;
|
||||
load();
|
||||
void load();
|
||||
}
|
||||
|
||||
function docPreview(doc: JsonRecord): string {
|
||||
|
|
@ -1274,6 +1403,7 @@ onMounted(async () => {
|
|||
onBeforeUnmount(() => {
|
||||
window.removeEventListener("pointerdown", handleDocumentBrowserPointerDown, true);
|
||||
if (documentLoadExecutionId.value) void api.cancelQuery(documentLoadExecutionId.value);
|
||||
cancelElasticsearchCount();
|
||||
stopDocumentLoadingTimer();
|
||||
endTableSearchSplitResize();
|
||||
});
|
||||
|
|
@ -1339,14 +1469,14 @@ defineExpose({ focusSearch });
|
|||
<span class="shrink-0 ml-1">{{ documentStoreLabels.documentsLabel }}</span>
|
||||
|
||||
<Button v-if="viewMode === 'document'" variant="ghost" size="icon" class="h-5 w-5" @click="startNew"><Plus class="h-3 w-3" /></Button>
|
||||
<Button v-if="viewMode === 'document'" variant="ghost" size="icon" class="h-5 w-5" @click="load"><RefreshCw class="h-3 w-3" :class="{ 'animate-spin': loading }" /></Button>
|
||||
<Button v-if="viewMode === 'document'" variant="ghost" size="icon" class="h-5 w-5" @click="refreshDocuments"><RefreshCw class="h-3 w-3" :class="{ 'animate-spin': loading }" /></Button>
|
||||
|
||||
<div v-if="viewMode === 'document'" class="flex items-center gap-1 ml-1">
|
||||
<Button variant="ghost" size="icon" class="h-5 w-5" :disabled="page <= 0" @click="prevPage">
|
||||
<ChevronLeft class="h-3 w-3" />
|
||||
</Button>
|
||||
<span>{{ page + 1 }} / {{ Math.max(1, Math.ceil(total / pageSize)) }}</span>
|
||||
<Button variant="ghost" size="icon" class="h-5 w-5" :disabled="(page + 1) * pageSize >= total" @click="nextPage">
|
||||
<span>{{ page + 1 }} / {{ Math.max(1, Math.ceil(pageTotal / pageSize)) }}</span>
|
||||
<Button variant="ghost" size="icon" class="h-5 w-5" :disabled="(page + 1) * pageSize >= pageTotal" @click="nextPage">
|
||||
<ChevronRight class="h-3 w-3" />
|
||||
</Button>
|
||||
</div>
|
||||
|
|
@ -1446,8 +1576,10 @@ defineExpose({ focusSearch });
|
|||
:page-offset="page * pageSize"
|
||||
:page-limit="pageSize"
|
||||
:total-row-count="total"
|
||||
:total-row-count-is-exact="totalIsExact"
|
||||
:pagination-total-row-count="pageTotal"
|
||||
@sort="onSort"
|
||||
@reload="load"
|
||||
@reload="refreshDocuments"
|
||||
@paginate="(offset: number, limit: number) => paginate(offset, limit)"
|
||||
>
|
||||
<template #search-bar="{ localFilterCount, hasLocalColumnFilters, localFilterSummaries, clearLocalFilter }: { localFilterCount: number; hasLocalColumnFilters: boolean; localFilterSummaries: LocalFilterSummary[]; clearLocalFilter: (columnIndex?: number) => void }">
|
||||
|
|
|
|||
|
|
@ -287,6 +287,8 @@ interface DataGridProps {
|
|||
pageLimit?: number;
|
||||
countSql?: string;
|
||||
totalRowCount?: number;
|
||||
totalRowCountIsExact?: boolean;
|
||||
paginationTotalRowCount?: number;
|
||||
totalRowCountLoading?: boolean;
|
||||
loading?: boolean;
|
||||
cacheKey?: string;
|
||||
|
|
@ -2363,9 +2365,13 @@ const inferredBackendTotalRowCount = computed(() => {
|
|||
});
|
||||
const serverKnownTotalRowCount = computed(() => props.totalRowCount ?? manualTotalRowCount.value);
|
||||
const displayedTotalRowCount = computed(() => serverKnownTotalRowCount.value ?? inferredBackendTotalRowCount.value);
|
||||
const totalRowCountIsExact = computed(() => props.totalRowCountIsExact !== false);
|
||||
// A backend can expose an exact display total while deliberately restricting
|
||||
// offset pagination to a smaller safe range.
|
||||
const paginationTotalRowCount = computed(() => props.paginationTotalRowCount ?? serverKnownTotalRowCount.value);
|
||||
// Only a server-confirmed total drives pagination — an inferred total means
|
||||
// rows exist that we never fetched, so navigation must stay inside rows.length.
|
||||
const hasKnownTotalRowCount = computed(() => typeof serverKnownTotalRowCount.value === "number" && serverKnownTotalRowCount.value >= 0);
|
||||
const hasKnownPaginationTotalRowCount = computed(() => typeof paginationTotalRowCount.value === "number" && paginationTotalRowCount.value >= 0);
|
||||
// When context=results and the caller hasn't configured server-side
|
||||
// pagination (no pageLimit), the backend handed us every row up-front and
|
||||
// rowCount IS the total. Without this hint, the "page is full → assume more"
|
||||
|
|
@ -2391,11 +2397,11 @@ const canGoNextPage = computed(() => {
|
|||
pageSize: pageSize.value,
|
||||
pageOffset: props.pageOffset,
|
||||
currentPage: currentPage.value,
|
||||
totalRowCount: hasKnownTotalRowCount.value ? displayedTotalRowCount.value : undefined,
|
||||
totalRowCount: hasKnownPaginationTotalRowCount.value ? paginationTotalRowCount.value : undefined,
|
||||
allRowsLoaded: allRowsLoaded.value,
|
||||
});
|
||||
});
|
||||
const canJumpLastPage = computed(() => canGoNextPage.value && (hasKnownTotalRowCount.value || allRowsLoaded.value || !!props.tableMeta || !!props.countSql));
|
||||
const canJumpLastPage = computed(() => canGoNextPage.value && (hasKnownPaginationTotalRowCount.value || allRowsLoaded.value || !!props.tableMeta || !!props.countSql));
|
||||
const totalRowCountBusy = computed(() => props.totalRowCountLoading === true || manualTotalRowCountLoading.value);
|
||||
const canCalculateTotalRowCount = computed(() => !!props.connectionId && (!!props.tableMeta || !!props.countSql));
|
||||
// When a refresh/rollback completes and the current page exceeds the last
|
||||
|
|
@ -2408,7 +2414,7 @@ watch(
|
|||
// and the completion was triggered by a refresh/rollback.
|
||||
if (!loading && prevLoading && isRefreshingData.value) {
|
||||
isRefreshingData.value = false;
|
||||
const total = displayedTotalRowCount.value;
|
||||
const total = paginationTotalRowCount.value;
|
||||
if (!total || total <= 0) return;
|
||||
const lastPageNum = Math.max(1, Math.ceil(total / pageSize.value));
|
||||
if (currentPage.value <= lastPageNum) return;
|
||||
|
|
@ -2571,8 +2577,8 @@ function applyCustomPageSize() {
|
|||
|
||||
async function lastPage() {
|
||||
if (infiniteScrollEnabled.value) return;
|
||||
if (hasKnownTotalRowCount.value) {
|
||||
const total = displayedTotalRowCount.value ?? 0;
|
||||
if (hasKnownPaginationTotalRowCount.value) {
|
||||
const total = paginationTotalRowCount.value ?? 0;
|
||||
if (total <= 0) return;
|
||||
const lastPageNum = Math.ceil(total / pageSize.value);
|
||||
if (lastPageNum <= currentPage.value) return;
|
||||
|
|
@ -8916,7 +8922,7 @@ const gridContextMenuItems = computed<ContextMenuItem[]>(() => {
|
|||
<div class="flex min-w-0 items-center gap-2 overflow-hidden">
|
||||
<span v-if="hasData" class="shrink-0">
|
||||
{{ t(showTruncationWarning ? "grid.loadedRows" : "grid.totalRows", { count: result.rows.length }) }}
|
||||
<span v-if="typeof displayedTotalRowCount === 'number' && displayedTotalRowCount >= 0" class="text-muted-foreground/70">{{ t("grid.totalRowCount", { count: displayedTotalRowCount }) }}</span>
|
||||
<span v-if="typeof displayedTotalRowCount === 'number' && displayedTotalRowCount >= 0" class="text-muted-foreground/70">{{ t(totalRowCountIsExact === false ? "grid.totalRowCountAtLeast" : "grid.totalRowCount", { count: displayedTotalRowCount }) }}</span>
|
||||
<span v-else-if="totalRowCountBusy" class="text-muted-foreground/70">
|
||||
{{ t("grid.totalRowCountLoading") }}
|
||||
</span>
|
||||
|
|
|
|||
|
|
@ -873,6 +873,7 @@ export default {
|
|||
totalRows: "Total {count} rows",
|
||||
loadedRows: "Loaded {count} rows",
|
||||
totalRowCount: "({count} total)",
|
||||
totalRowCountAtLeast: "(at least {count} total)",
|
||||
totalRowCountLoading: "(counting...)",
|
||||
loadingMore: "Loading more data...",
|
||||
allLoaded: "all loaded",
|
||||
|
|
|
|||
|
|
@ -822,6 +822,7 @@ export default withEnglishFallback({
|
|||
totalRows: "Total {count} filas",
|
||||
loadedRows: "{count} filas cargadas",
|
||||
totalRowCount: "({count} en total)",
|
||||
totalRowCountAtLeast: "(al menos {count} en total)",
|
||||
totalRowCountLoading: "(contando...)",
|
||||
loadingMore: "Cargando más datos...",
|
||||
allLoaded: "todo cargado",
|
||||
|
|
|
|||
|
|
@ -820,6 +820,7 @@ export default withEnglishFallback({
|
|||
totalRows: "Totale {count} righe",
|
||||
loadedRows: "{count} righe caricate",
|
||||
totalRowCount: "({count} in totale)",
|
||||
totalRowCountAtLeast: "(almeno {count} in totale)",
|
||||
totalRowCountLoading: "(conteggio...)",
|
||||
loadingMore: "Caricamento altri dati...",
|
||||
allLoaded: "tutto caricato",
|
||||
|
|
|
|||
|
|
@ -821,6 +821,7 @@ export default withEnglishFallback({
|
|||
totalRows: "{count}件表示",
|
||||
loadedRows: "{count}件読み込み済み",
|
||||
totalRowCount: "（全{count}件）",
|
||||
totalRowCountAtLeast: "（少なくとも{count}件）",
|
||||
totalRowCountLoading: "（カウント中...）",
|
||||
calculateTotalRows: "総行数をカウント",
|
||||
calculateTotalRowsInline: "（総行数をカウント）",
|
||||
|
|
|
|||
|
|
@ -822,6 +822,7 @@ export default withEnglishFallback({
|
|||
totalRows: "Total de {count} linhas",
|
||||
loadedRows: "{count} linhas carregadas",
|
||||
totalRowCount: "({count} no total)",
|
||||
totalRowCountAtLeast: "(pelo menos {count} no total)",
|
||||
totalRowCountLoading: "(contando...)",
|
||||
loadingMore: "Carregando mais dados...",
|
||||
allLoaded: "tudo carregado",
|
||||
|
|
|
|||
|
|
@ -875,6 +875,7 @@ export default withEnglishFallback({
|
|||
totalRows: "共 {count} 行",
|
||||
loadedRows: "已加载 {count} 行",
|
||||
totalRowCount: "（总计 {count} 行）",
|
||||
totalRowCountAtLeast: "（至少 {count} 行）",
|
||||
totalRowCountLoading: "（统计中...）",
|
||||
loadingMore: "加载更多数据...",
|
||||
allLoaded: "已全部加载",
|
||||
|
|
|
|||
|
|
@ -822,6 +822,7 @@ export default withEnglishFallback({
|
|||
totalRows: "共 {count} 筆",
|
||||
loadedRows: "已載入 {count} 筆",
|
||||
totalRowCount: "（總計 {count} 筆）",
|
||||
totalRowCountAtLeast: "（至少 {count} 筆）",
|
||||
totalRowCountLoading: "（統計中...）",
|
||||
loadingMore: "載入更多資料...",
|
||||
allLoaded: "已全部載入",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,45 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { clampDocumentPage, documentPageRequestLimit, resetElasticsearchDocumentTotals, resolveElasticsearchDocumentTotals } from "@/lib/document/elasticsearchDocumentTotals";
|
||||
|
||||
describe("Elasticsearch document totals", () => {
|
||||
it("keeps a lower-bound search total separate from an exact background count", () => {
|
||||
expect(resolveElasticsearchDocumentTotals(10_000, false)).toEqual({
|
||||
total: 10_000,
|
||||
totalIsExact: false,
|
||||
paginationTotal: 10_000,
|
||||
});
|
||||
expect(resolveElasticsearchDocumentTotals(10_000, false, 552_033)).toEqual({
|
||||
total: 552_033,
|
||||
totalIsExact: true,
|
||||
paginationTotal: 10_000,
|
||||
});
|
||||
});
|
||||
|
||||
it("uses exact search totals for both display and pagination", () => {
|
||||
expect(resolveElasticsearchDocumentTotals(42, true, 100)).toEqual({
|
||||
total: 42,
|
||||
totalIsExact: true,
|
||||
paginationTotal: 42,
|
||||
});
|
||||
});
|
||||
|
||||
it("clears a stale display total without losing the safe page cap during a refresh", () => {
|
||||
expect(resetElasticsearchDocumentTotals(10_000, true)).toEqual({
|
||||
total: undefined,
|
||||
totalIsExact: true,
|
||||
paginationTotal: 10_000,
|
||||
});
|
||||
expect(resetElasticsearchDocumentTotals(10_000)).toEqual({
|
||||
total: undefined,
|
||||
totalIsExact: true,
|
||||
paginationTotal: undefined,
|
||||
});
|
||||
});
|
||||
|
||||
it("clamps the final request without exceeding the conservative page cap", () => {
|
||||
expect(clampDocumentPage(30, 333, 10_000)).toBe(30);
|
||||
expect(clampDocumentPage(31, 333, 10_000)).toBe(30);
|
||||
expect(documentPageRequestLimit(30, 333, 10_000)).toBe(10);
|
||||
expect(documentPageRequestLimit(0, 100, undefined)).toBe(100);
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,19 @@
|
|||
import { describe, expect, it } from "vitest";
|
||||
import { elasticsearchJsonResponseForResult } from "@/lib/elasticsearch/elasticsearchJsonResponse";
|
||||
import type { QueryResult } from "@/types/database";
|
||||
|
||||
const catJsonResult: QueryResult = {
|
||||
columns: ["status", "response"],
|
||||
rows: [[200, '[{"index":"data_pack_and_box_index_v1","docs.count":"42"}]']],
|
||||
affected_rows: 0,
|
||||
execution_time_ms: 1,
|
||||
};
|
||||
|
||||
describe("Elasticsearch JSON response detection", () => {
|
||||
it("routes an unformatted CAT response to the JSON renderer", () => {
|
||||
expect(elasticsearchJsonResponseForResult("elasticsearch", "GET /_cat/indices/data_pack_and_box_index_v1", catJsonResult)).toEqual({
|
||||
status: 200,
|
||||
body: '[{"index":"data_pack_and_box_index_v1","docs.count":"42"}]',
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -491,6 +491,7 @@ export const mongoDropDatabase = forward("mongoDropDatabase");
|
|||
export const mongoDropCollection = forward("mongoDropCollection");
|
||||
export const mongoRenameCollection = forward("mongoRenameCollection");
|
||||
export const documentFindDocuments = forward("documentFindDocuments");
|
||||
export const elasticsearchCountDocuments = forward("elasticsearchCountDocuments");
|
||||
export const mongoFindDocuments = forward("mongoFindDocuments");
|
||||
export const mongoParseShellCommand = forward("mongoParseShellCommand");
|
||||
export const mongoFindOne = forward("mongoFindOne");
|
||||
|
|
@ -611,6 +612,7 @@ export type {
|
|||
KvPutOptions,
|
||||
KvPutResponse,
|
||||
KvDeleteResponse,
|
||||
DocumentQueryResult,
|
||||
MongoDocumentResult,
|
||||
HistoryEntry,
|
||||
HistoryConnectionFilter,
|
||||
|
|
|
|||
|
|
@ -75,6 +75,7 @@ import type {
|
|||
KvPutOptions,
|
||||
KvPutResponse,
|
||||
KvDeleteResponse,
|
||||
DocumentQueryResult,
|
||||
MongoDocumentResult,
|
||||
MongoCollectionStatsResult,
|
||||
MongoGridFsBucketInfo,
|
||||
|
|
@ -2178,10 +2179,14 @@ export async function mongoFindOne(connectionId: string, database: string, colle
|
|||
return post("/api/mongo/find-one", { connectionId, database, collection, filter, projection, options, executionId });
|
||||
}
|
||||
|
||||
export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, projection?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, projection?: string, sort?: string, executionId?: string): Promise<DocumentQueryResult> {
|
||||
return post("/api/document-store/find-documents", { connectionId, database, collection, skip, limit, filter, projection, sort, executionId });
|
||||
}
|
||||
|
||||
export async function elasticsearchCountDocuments(connectionId: string, index: string, filter?: string, executionId?: string): Promise<number> {
|
||||
return post("/api/document-store/elasticsearch-count-documents", { connectionId, index, filter, executionId });
|
||||
}
|
||||
|
||||
export async function mongoCountDocuments(connectionId: string, database: string, collection: string, filter?: string, mode?: "accurate" | "legacy", executionId?: string): Promise<number> {
|
||||
return post("/api/mongo/count-documents", { connectionId, database, collection, filter, mode, executionId });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1849,14 +1849,18 @@ export async function zookeeperDelete(connectionId: string, key: string): Promis
|
|||
return invoke("zookeeper_delete", { connectionId, key });
|
||||
}
|
||||
|
||||
// --- MongoDB ---
|
||||
export interface MongoDocumentResult {
|
||||
// --- Document stores ---
|
||||
export interface DocumentQueryResult {
|
||||
documents: any[];
|
||||
raw_documents?: string[];
|
||||
extended_documents?: any[];
|
||||
total: number;
|
||||
total_is_exact?: boolean;
|
||||
}
|
||||
|
||||
// Kept for callers that are specifically using MongoDB APIs.
|
||||
export type MongoDocumentResult = DocumentQueryResult;
|
||||
|
||||
export interface MongoCollectionStatsResult {
|
||||
count: unknown;
|
||||
size: unknown;
|
||||
|
|
@ -1942,10 +1946,14 @@ export async function mongoParseShellCommand(source: string): Promise<MongoComma
|
|||
return normalizeRustMongoCommand(raw);
|
||||
}
|
||||
|
||||
export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, projection?: string, sort?: string, executionId?: string): Promise<MongoDocumentResult> {
|
||||
export async function documentFindDocuments(connectionId: string, database: string, collection: string, skip: number, limit: number, filter?: string, projection?: string, sort?: string, executionId?: string): Promise<DocumentQueryResult> {
|
||||
return invoke("document_find_documents", { connectionId, database, collection, skip, limit, filter, projection, sort, executionId });
|
||||
}
|
||||
|
||||
export async function elasticsearchCountDocuments(connectionId: string, index: string, filter?: string, executionId?: string): Promise<number> {
|
||||
return invoke("elasticsearch_count_documents", { connectionId, index, filter, executionId });
|
||||
}
|
||||
|
||||
export async function mongoCountDocuments(connectionId: string, database: string, collection: string, filter?: string, mode?: "accurate" | "legacy", executionId?: string): Promise<number> {
|
||||
return invoke("mongo_count_documents", { connectionId, database, collection, filter, mode, executionId });
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,49 @@
|
|||
export interface ElasticsearchDocumentTotals {
|
||||
total: number;
|
||||
totalIsExact: boolean;
|
||||
paginationTotal: number;
|
||||
}
|
||||
|
||||
export interface ResetElasticsearchDocumentTotals {
|
||||
total: undefined;
|
||||
totalIsExact: boolean;
|
||||
paginationTotal?: number;
|
||||
}
|
||||
|
||||
export function resolveElasticsearchDocumentTotals(searchTotal: number, searchTotalIsExact: boolean, exactCount?: number): ElasticsearchDocumentTotals {
|
||||
if (searchTotalIsExact || exactCount === undefined) {
|
||||
return {
|
||||
total: searchTotal,
|
||||
totalIsExact: searchTotalIsExact,
|
||||
paginationTotal: searchTotal,
|
||||
};
|
||||
}
|
||||
return {
|
||||
total: exactCount,
|
||||
totalIsExact: true,
|
||||
paginationTotal: Math.min(searchTotal, exactCount),
|
||||
};
|
||||
}
|
||||
|
||||
export function resetElasticsearchDocumentTotals(paginationTotal: number | undefined, preservePaginationTotal = false): ResetElasticsearchDocumentTotals {
|
||||
return {
|
||||
total: undefined,
|
||||
totalIsExact: true,
|
||||
paginationTotal: preservePaginationTotal ? paginationTotal : undefined,
|
||||
};
|
||||
}
|
||||
|
||||
export function clampDocumentPage(page: number, pageSize: number, paginationTotal?: number): number {
|
||||
const normalizedPage = Math.max(0, Math.floor(page));
|
||||
if (paginationTotal === undefined) return normalizedPage;
|
||||
if (paginationTotal <= 0) return 0;
|
||||
const lastPage = Math.max(0, Math.ceil(paginationTotal / Math.max(1, pageSize)) - 1);
|
||||
return Math.min(normalizedPage, lastPage);
|
||||
}
|
||||
|
||||
export function documentPageRequestLimit(page: number, pageSize: number, paginationTotal?: number): number {
|
||||
const normalizedPageSize = Math.max(1, Math.floor(pageSize));
|
||||
if (paginationTotal === undefined) return normalizedPageSize;
|
||||
const remaining = paginationTotal - Math.max(0, Math.floor(page)) * normalizedPageSize;
|
||||
return remaining > 0 ? Math.min(normalizedPageSize, remaining) : normalizedPageSize;
|
||||
}
|
||||
|
|
@ -9,9 +9,8 @@ export interface ElasticsearchJsonResponse {
|
|||
const ELASTICSEARCH_REST_STATEMENT = /^(?:GET|POST|PUT|DELETE|HEAD)\s+\S+/i;
|
||||
|
||||
/**
|
||||
* Detect the result shape emitted for a JSON response to an explicit
|
||||
* Elasticsearch REST request. SQL and text (such as CAT) results keep using
|
||||
* the normal data-grid path.
|
||||
* Detect the raw HTTP result emitted for an Elasticsearch REST request.
|
||||
* DBX asks unformatted CAT requests for JSON so they use this response panel.
|
||||
*/
|
||||
export function elasticsearchJsonResponseForResult(databaseType: DatabaseType | undefined, sourceStatement: string | undefined, result: QueryResult | undefined): ElasticsearchJsonResponse | undefined {
|
||||
if (databaseType !== "elasticsearch" || !result || typeof sourceStatement !== "string") return undefined;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,50 @@
|
|||
use serde::{Deserialize, Serialize};
|
||||
|
||||
/// Common result shape returned by document-store queries.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct DocumentQueryResult {
|
||||
pub documents: Vec<serde_json::Value>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub raw_documents: Option<Vec<String>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub extended_documents: Option<Vec<serde_json::Value>>,
|
||||
pub total: u64,
|
||||
// Older document stores always return an exact total. Keep that wire shape
|
||||
// unchanged and only send the flag when Elasticsearch reports a lower bound.
|
||||
#[serde(default = "default_total_is_exact", skip_serializing_if = "is_true")]
|
||||
pub total_is_exact: bool,
|
||||
}
|
||||
|
||||
fn default_total_is_exact() -> bool {
|
||||
true
|
||||
}
|
||||
|
||||
fn is_true(value: &bool) -> bool {
|
||||
*value
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::DocumentQueryResult;
|
||||
|
||||
#[test]
|
||||
fn exact_results_keep_the_existing_wire_shape() {
|
||||
let result = DocumentQueryResult {
|
||||
documents: Vec::new(),
|
||||
raw_documents: None,
|
||||
extended_documents: None,
|
||||
total: 1,
|
||||
total_is_exact: true,
|
||||
};
|
||||
|
||||
let serialized = serde_json::to_value(result).unwrap();
|
||||
assert!(serialized.get("total_is_exact").is_none());
|
||||
|
||||
let deserialized: DocumentQueryResult = serde_json::from_value(serde_json::json!({
|
||||
"documents": [],
|
||||
"total": 1,
|
||||
}))
|
||||
.unwrap();
|
||||
assert!(deserialized.total_is_exact);
|
||||
}
|
||||
}
|
||||
|
|
@ -1,4 +1,4 @@
|
|||
use percent_encoding::{utf8_percent_encode, AsciiSet, CONTROLS};
|
||||
use percent_encoding::{percent_decode_str, utf8_percent_encode, AsciiSet, CONTROLS};
|
||||
use reqwest::{Client as HttpClient, Method, StatusCode};
|
||||
use serde::Deserialize;
|
||||
use serde_json::Value;
|
||||
|
|
@ -7,7 +7,7 @@ use std::error::Error;
|
|||
use std::time::Duration;
|
||||
|
||||
use super::{http_client_builder, with_connection_timeout};
|
||||
use crate::db::mongo_driver::MongoDocumentResult;
|
||||
use crate::db::document_result::DocumentQueryResult;
|
||||
|
||||
const ELASTICSEARCH_PATH_SEGMENT_ENCODE_SET: &AsciiSet = &CONTROLS
|
||||
.add(b' ')
|
||||
|
|
@ -421,6 +421,12 @@ fn push_mapping_column(
|
|||
#[derive(Deserialize)]
|
||||
struct SearchResponse {
|
||||
hits: SearchHits,
|
||||
#[serde(rename = "_shards")]
|
||||
shards: Option<ElasticsearchShards>,
|
||||
#[serde(default)]
|
||||
timed_out: bool,
|
||||
#[serde(default)]
|
||||
terminated_early: bool,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
|
|
@ -431,13 +437,20 @@ struct SearchHits {
|
|||
|
||||
enum HitsTotal {
|
||||
Count(u64),
|
||||
Value { value: u64 },
|
||||
Value { value: u64, is_exact: bool },
|
||||
}
|
||||
|
||||
impl HitsTotal {
|
||||
fn value(&self) -> u64 {
|
||||
match self {
|
||||
Self::Count(value) | Self::Value { value } => *value,
|
||||
Self::Count(value) | Self::Value { value, .. } => *value,
|
||||
}
|
||||
}
|
||||
|
||||
fn is_exact(&self) -> bool {
|
||||
match self {
|
||||
Self::Count(_) => true,
|
||||
Self::Value { is_exact, .. } => *is_exact,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -452,7 +465,10 @@ impl<'de> Deserialize<'de> for HitsTotal {
|
|||
return Ok(Self::Count(count));
|
||||
}
|
||||
if let Some(count) = value.get("value").and_then(serde_json::Value::as_u64) {
|
||||
return Ok(Self::Value { value: count });
|
||||
// Object totals are exact only when Elasticsearch explicitly says
|
||||
// so. Treat an absent or unknown relation conservatively.
|
||||
let is_exact = value.get("relation").and_then(serde_json::Value::as_str) == Some("eq");
|
||||
return Ok(Self::Value { value: count, is_exact });
|
||||
}
|
||||
Err(serde::de::Error::custom("expected hits.total as a number or an object with value"))
|
||||
}
|
||||
|
|
@ -475,7 +491,7 @@ pub async fn find_documents(
|
|||
limit: i64,
|
||||
filter: Option<&str>,
|
||||
sort: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
) -> Result<DocumentQueryResult, String> {
|
||||
let body = build_find_documents_body(skip, limit, filter, sort)?;
|
||||
|
||||
let path = elasticsearch_index_path(index, "_search");
|
||||
|
|
@ -491,7 +507,58 @@ pub async fn find_documents(
|
|||
search_response_to_document_result(result)
|
||||
}
|
||||
|
||||
fn search_response_to_document_result(result: SearchResponse) -> Result<MongoDocumentResult, String> {
|
||||
#[derive(Deserialize)]
|
||||
struct CountResponse {
|
||||
count: u64,
|
||||
#[serde(rename = "_shards")]
|
||||
shards: ElasticsearchShards,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct ElasticsearchShards {
|
||||
total: u64,
|
||||
successful: u64,
|
||||
#[serde(default)]
|
||||
skipped: u64,
|
||||
failed: u64,
|
||||
}
|
||||
|
||||
impl ElasticsearchShards {
|
||||
fn is_complete(&self) -> bool {
|
||||
// Elasticsearch only skips shards that cannot match, so successful and
|
||||
// skipped shards together must account for every requested shard.
|
||||
self.failed == 0 && self.successful.checked_add(self.skipped) == Some(self.total)
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn count_documents(client: &EsClient, index: &str, filter: Option<&str>) -> Result<u64, String> {
|
||||
let body = build_count_documents_body(filter)?;
|
||||
let path = elasticsearch_index_path(index, "_count");
|
||||
let resp = client.post(&path).json(&body).send().await.map_err(|e| format!("Elasticsearch request failed: {e}"))?;
|
||||
|
||||
if !client.response_status(&resp).is_success() {
|
||||
let body = resp.text().await.unwrap_or_default();
|
||||
return Err(format!("Elasticsearch error: {body}"));
|
||||
}
|
||||
|
||||
let result: CountResponse = resp.json().await.map_err(|e| format!("Elasticsearch count parse error: {e}"))?;
|
||||
if !result.shards.is_complete() {
|
||||
return Err(format!(
|
||||
"Elasticsearch count returned an incomplete shard response: {} successful, {} skipped, {} failed of {} shards",
|
||||
result.shards.successful, result.shards.skipped, result.shards.failed, result.shards.total,
|
||||
));
|
||||
}
|
||||
Ok(result.count)
|
||||
}
|
||||
|
||||
fn search_response_to_document_result(result: SearchResponse) -> Result<DocumentQueryResult, String> {
|
||||
// A 200 search response can still omit failed shards. Only expose an
|
||||
// exact total when both the total relation and shard metadata agree.
|
||||
let total_is_exact = !result.timed_out
|
||||
&& !result.terminated_early
|
||||
&& result.hits.total.is_exact()
|
||||
&& result.shards.as_ref().is_some_and(ElasticsearchShards::is_complete);
|
||||
let total = result.hits.total.value();
|
||||
let documents: Vec<serde_json::Value> = result
|
||||
.hits
|
||||
.hits
|
||||
|
|
@ -518,11 +585,12 @@ fn search_response_to_document_result(result: SearchResponse) -> Result<MongoDoc
|
|||
.collect::<Result<Vec<_>, _>>()
|
||||
.map_err(|e| format!("Elasticsearch document serialization failed: {e}"))?;
|
||||
|
||||
Ok(MongoDocumentResult {
|
||||
Ok(DocumentQueryResult {
|
||||
documents,
|
||||
raw_documents: Some(raw_documents),
|
||||
extended_documents: None,
|
||||
total: result.hits.total.value(),
|
||||
total,
|
||||
total_is_exact,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
@ -544,6 +612,14 @@ fn build_find_documents_body(
|
|||
Ok(serde_json::Value::Object(body))
|
||||
}
|
||||
|
||||
fn build_count_documents_body(filter: Option<&str>) -> Result<serde_json::Value, String> {
|
||||
let mut body = serde_json::Map::new();
|
||||
if let Some(query) = elasticsearch_query_from_document_filter(filter)? {
|
||||
body.insert("query".to_string(), query);
|
||||
}
|
||||
Ok(serde_json::Value::Object(body))
|
||||
}
|
||||
|
||||
fn elasticsearch_query_from_document_filter(filter: Option<&str>) -> Result<Option<serde_json::Value>, String> {
|
||||
let Some(filter) = filter.map(str::trim).filter(|value| !value.is_empty()) else {
|
||||
return Ok(None);
|
||||
|
|
@ -865,6 +941,32 @@ fn is_elasticsearch_ndjson_path(path: &str) -> bool {
|
|||
|| path.ends_with("/_msearch/template")
|
||||
}
|
||||
|
||||
fn is_elasticsearch_cat_path(path: &str) -> bool {
|
||||
let path = path.split('?').next().unwrap_or(path).trim_end_matches('/');
|
||||
path == "/_cat" || path.starts_with("/_cat/")
|
||||
}
|
||||
|
||||
fn elasticsearch_query_parameter(path: &str, name: &str) -> Option<String> {
|
||||
let query = path.split_once('?')?.1;
|
||||
query.split('&').find_map(|part| {
|
||||
let (key, value) = part.split_once('=').unwrap_or((part, ""));
|
||||
let key = percent_decode_str(key).decode_utf8_lossy();
|
||||
if key.eq_ignore_ascii_case(name) {
|
||||
Some(percent_decode_str(value).decode_utf8_lossy().into_owned())
|
||||
} else {
|
||||
None
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn add_default_cat_json_format(mut request: ElasticsearchRestRequest) -> ElasticsearchRestRequest {
|
||||
if is_elasticsearch_cat_path(&request.path) && elasticsearch_query_parameter(&request.path, "format").is_none() {
|
||||
request.path.push(if request.path.contains('?') { '&' } else { '?' });
|
||||
request.path.push_str("format=json");
|
||||
}
|
||||
request
|
||||
}
|
||||
|
||||
fn normalize_elasticsearch_rest_path(path: &str) -> String {
|
||||
let (path_part, query) = path.split_once('?').map_or((path, None), |(path, query)| (path, Some(query)));
|
||||
let mut normalized = String::with_capacity(path.len());
|
||||
|
|
@ -998,7 +1100,9 @@ pub async fn execute_rest_query(client: &EsClient, input: &str) -> Result<crate:
|
|||
return execute_sql_query(client, input, start).await;
|
||||
}
|
||||
|
||||
let request = parse_elasticsearch_rest_request(input)?;
|
||||
// CAT APIs default to text, so request JSON for an unformatted CAT call.
|
||||
// The frontend renders the returned HTTP body in its JSON response panel.
|
||||
let request = add_default_cat_json_format(parse_elasticsearch_rest_request(input)?);
|
||||
let mut builder = client.request(request.method, &request.path);
|
||||
if let Some(body) = request.body {
|
||||
builder = match request.body_kind {
|
||||
|
|
@ -1721,8 +1825,8 @@ fn parse_aggregations(aggs: &serde_json::Map<String, serde_json::Value>) -> (Vec
|
|||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::{
|
||||
build_find_documents_body, elasticsearch_accept_invalid_certs, elasticsearch_base_url_fallbacks,
|
||||
redact_elasticsearch_url, EsClient, SearchResponse,
|
||||
build_count_documents_body, build_find_documents_body, elasticsearch_accept_invalid_certs,
|
||||
elasticsearch_base_url_fallbacks, redact_elasticsearch_url, EsClient, SearchResponse,
|
||||
};
|
||||
use serde_json::json;
|
||||
use std::time::Duration;
|
||||
|
|
@ -1947,6 +2051,7 @@ mod tests {
|
|||
"sort": [{ "created_at": { "order": "desc" } }]
|
||||
})
|
||||
);
|
||||
assert!(body.get("track_total_hits").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -1981,6 +2086,95 @@ mod tests {
|
|||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builds_elasticsearch_count_body_with_the_same_native_query_filter() {
|
||||
let body = build_count_documents_body(Some(r#"{"$esQuery":{"term":{"status":"active"}}}"#)).unwrap();
|
||||
|
||||
assert_eq!(body, json!({ "query": { "term": { "status": "active" } } }));
|
||||
assert!(body.get("from").is_none());
|
||||
assert!(body.get("size").is_none());
|
||||
assert!(body.get("sort").is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builds_elasticsearch_count_body_with_structured_filter_operators() {
|
||||
let filter = r#"{"$and":[{"city":{"$ne":"上海"}},{"age":{"$gte":18}}]}"#;
|
||||
assert_eq!(
|
||||
build_count_documents_body(Some(filter)).unwrap(),
|
||||
json!({
|
||||
"query": {
|
||||
"bool": {
|
||||
"filter": [
|
||||
{ "bool": { "must_not": [{ "term": { "city": "上海" } }] } },
|
||||
{ "range": { "age": { "gte": 18 } } }
|
||||
]
|
||||
}
|
||||
}
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_only_complete_elasticsearch_count_shards() {
|
||||
assert!(super::ElasticsearchShards { total: 3, successful: 2, skipped: 1, failed: 0 }.is_complete());
|
||||
assert!(!super::ElasticsearchShards { total: 3, successful: 2, skipped: 0, failed: 0 }.is_complete());
|
||||
assert!(!super::ElasticsearchShards { total: 3, successful: 2, skipped: 0, failed: 1 }.is_complete());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn counts_documents_with_the_translated_filter() {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let response_body = r#"{"count":552033,"_shards":{"total":3,"successful":3,"skipped":0,"failed":0}}"#;
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (mut socket, _) = listener.accept().await.unwrap();
|
||||
let request = read_http_request(&mut socket).await;
|
||||
assert!(request.starts_with("POST /orders/_count "));
|
||||
let body = request.split_once("\r\n\r\n").unwrap().1;
|
||||
assert_eq!(
|
||||
serde_json::from_str::<serde_json::Value>(body).unwrap(),
|
||||
json!({ "query": { "term": { "status": "active" } } })
|
||||
);
|
||||
let response = format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
||||
response_body.len(),
|
||||
response_body,
|
||||
);
|
||||
socket.write_all(response.as_bytes()).await.unwrap();
|
||||
});
|
||||
|
||||
let client = EsClient::new(&format!("http://{addr}"), None, None, false, Duration::from_secs(1));
|
||||
assert_eq!(super::count_documents(&client, "orders", Some(r#"{"status":"active"}"#)).await.unwrap(), 552_033);
|
||||
server.await.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn rejects_partial_elasticsearch_document_count() {
|
||||
use tokio::io::AsyncWriteExt;
|
||||
|
||||
let response_body = r#"{"count":4,"_shards":{"total":3,"successful":2,"skipped":0,"failed":1}}"#;
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (mut socket, _) = listener.accept().await.unwrap();
|
||||
let request = read_http_request(&mut socket).await;
|
||||
assert!(request.starts_with("POST /orders/_count "));
|
||||
let response = format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
||||
response_body.len(),
|
||||
response_body,
|
||||
);
|
||||
socket.write_all(response.as_bytes()).await.unwrap();
|
||||
});
|
||||
|
||||
let client = EsClient::new(&format!("http://{addr}"), None, None, false, Duration::from_secs(1));
|
||||
let error = super::count_documents(&client, "orders", None).await.unwrap_err();
|
||||
assert!(error.contains("incomplete shard response: 2 successful, 0 skipped, 1 failed of 3 shards"));
|
||||
server.await.unwrap();
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn builds_elasticsearch_find_body_with_structured_filter_operators() {
|
||||
let body = build_find_documents_body(
|
||||
|
|
@ -2053,6 +2247,7 @@ mod tests {
|
|||
#[test]
|
||||
fn parses_search_total_from_elasticsearch_6_number_shape() {
|
||||
let response: SearchResponse = serde_json::from_value(json!({
|
||||
"_shards": { "total": 1, "successful": 1, "skipped": 0, "failed": 0 },
|
||||
"hits": {
|
||||
"total": 5,
|
||||
"hits": []
|
||||
|
|
@ -2061,11 +2256,13 @@ mod tests {
|
|||
.unwrap();
|
||||
|
||||
assert_eq!(response.hits.total.value(), 5);
|
||||
assert!(response.hits.total.is_exact());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_search_total_from_elasticsearch_7_object_shape() {
|
||||
let response: SearchResponse = serde_json::from_value(json!({
|
||||
"_shards": { "total": 1, "successful": 1, "skipped": 0, "failed": 0 },
|
||||
"hits": {
|
||||
"total": { "value": 5, "relation": "eq" },
|
||||
"hits": []
|
||||
|
|
@ -2074,6 +2271,78 @@ mod tests {
|
|||
.unwrap();
|
||||
|
||||
assert_eq!(response.hits.total.value(), 5);
|
||||
assert!(response.hits.total.is_exact());
|
||||
let result = super::search_response_to_document_result(response).unwrap();
|
||||
assert!(result.total_is_exact);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn preserves_elasticsearch_lower_bound_total_relation() {
|
||||
let response: SearchResponse = serde_json::from_value(json!({
|
||||
"hits": {
|
||||
"total": { "value": 10_000, "relation": "gte" },
|
||||
"hits": []
|
||||
}
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(response.hits.total.value(), 10_000);
|
||||
assert!(!response.hits.total.is_exact());
|
||||
let result = super::search_response_to_document_result(response).unwrap();
|
||||
assert_eq!(result.total, 10_000);
|
||||
assert!(!result.total_is_exact);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn treats_search_total_as_a_lower_bound_when_a_shard_failed() {
|
||||
let response: SearchResponse = serde_json::from_value(json!({
|
||||
"_shards": { "total": 3, "successful": 2, "skipped": 0, "failed": 1 },
|
||||
"hits": {
|
||||
"total": { "value": 5, "relation": "eq" },
|
||||
"hits": []
|
||||
}
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
let result = super::search_response_to_document_result(response).unwrap();
|
||||
assert_eq!(result.total, 5);
|
||||
assert!(!result.total_is_exact);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn treats_timed_out_or_terminated_searches_as_lower_bounds() {
|
||||
for response in [
|
||||
json!({
|
||||
"timed_out": true,
|
||||
"_shards": { "total": 1, "successful": 1, "skipped": 0, "failed": 0 },
|
||||
"hits": { "total": { "value": 5, "relation": "eq" }, "hits": [] }
|
||||
}),
|
||||
json!({
|
||||
"terminated_early": true,
|
||||
"_shards": { "total": 1, "successful": 1, "skipped": 0, "failed": 0 },
|
||||
"hits": { "total": { "value": 5, "relation": "eq" }, "hits": [] }
|
||||
}),
|
||||
] {
|
||||
let response: SearchResponse = serde_json::from_value(response).unwrap();
|
||||
let result = super::search_response_to_document_result(response).unwrap();
|
||||
assert_eq!(result.total, 5);
|
||||
assert!(!result.total_is_exact);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn treats_search_total_without_shard_metadata_as_a_lower_bound() {
|
||||
let response: SearchResponse = serde_json::from_value(json!({
|
||||
"hits": {
|
||||
"total": { "value": 5, "relation": "eq" },
|
||||
"hits": []
|
||||
}
|
||||
}))
|
||||
.unwrap();
|
||||
|
||||
let result = super::search_response_to_document_result(response).unwrap();
|
||||
assert_eq!(result.total, 5);
|
||||
assert!(!result.total_is_exact);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
@ -2191,6 +2460,22 @@ mod tests {
|
|||
assert_eq!(result.affected_rows, 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn adds_json_format_only_when_cat_format_is_not_explicit() {
|
||||
let plain =
|
||||
super::add_default_cat_json_format(super::parse_elasticsearch_rest_request("GET /_cat/indices").unwrap());
|
||||
assert_eq!(plain.path, "/_cat/indices?format=json");
|
||||
|
||||
let defaulted =
|
||||
super::add_default_cat_json_format(super::parse_elasticsearch_rest_request("GET /_cat/indices?v").unwrap());
|
||||
assert_eq!(defaulted.path, "/_cat/indices?v&format=json");
|
||||
|
||||
let explicit_text = super::add_default_cat_json_format(
|
||||
super::parse_elasticsearch_rest_request("GET /_cat/indices?format=txt").unwrap(),
|
||||
);
|
||||
assert_eq!(explicit_text.path, "/_cat/indices?format=txt");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn preserves_http_status_for_plain_text_rest_errors() {
|
||||
let result =
|
||||
|
|
@ -2223,11 +2508,10 @@ mod tests {
|
|||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execute_rest_query_keeps_plain_text_response_body() {
|
||||
async fn execute_rest_query_preserves_index_specific_cat_json_response() {
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
let body =
|
||||
"health status index docs.count store.size\ngreen open app-log-2026-07 42 10mb\n";
|
||||
let body = r#"[{"health":"green","index":"app-log-2026-07","docs.count":"42"}]"#;
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
|
|
@ -2235,7 +2519,67 @@ mod tests {
|
|||
let mut request = [0_u8; 1024];
|
||||
let read = socket.read(&mut request).await.unwrap();
|
||||
let request = String::from_utf8_lossy(&request[..read]);
|
||||
assert!(request.starts_with("GET /_cat/indices "));
|
||||
assert!(request.starts_with("GET /_cat/indices/data_pack_and_box_index_v1?format=json "));
|
||||
let response = format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
||||
body.len(),
|
||||
body
|
||||
);
|
||||
socket.write_all(response.as_bytes()).await.unwrap();
|
||||
});
|
||||
|
||||
let client = EsClient::new(&format!("http://{addr}"), None, None, false, Duration::from_secs(1));
|
||||
let result = super::execute_rest_query(&client, "GET /_cat/indices/data_pack_and_box_index_v1").await.unwrap();
|
||||
server.await.unwrap();
|
||||
|
||||
assert_eq!(result.columns, vec!["status", "response"]);
|
||||
assert_eq!(result.rows, vec![vec![json!(200), json!(body)]]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execute_rest_query_preserves_explicit_cat_json_response() {
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
let body = r#"[{"index":"data_pack_and_box_index_v1","docs.count":"42"}]"#;
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (mut socket, _) = listener.accept().await.unwrap();
|
||||
let mut request = [0_u8; 1024];
|
||||
let read = socket.read(&mut request).await.unwrap();
|
||||
let request = String::from_utf8_lossy(&request[..read]);
|
||||
assert!(request.starts_with("GET /_cat/indices/data_pack_and_box_index_v1?format=json "));
|
||||
let response = format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
||||
body.len(),
|
||||
body
|
||||
);
|
||||
socket.write_all(response.as_bytes()).await.unwrap();
|
||||
});
|
||||
|
||||
let client = EsClient::new(&format!("http://{addr}"), None, None, false, Duration::from_secs(1));
|
||||
let result = super::execute_rest_query(&client, "GET /_cat/indices/data_pack_and_box_index_v1?format=json")
|
||||
.await
|
||||
.unwrap();
|
||||
server.await.unwrap();
|
||||
|
||||
assert_eq!(result.columns, vec!["status", "response"]);
|
||||
assert_eq!(result.rows, vec![vec![json!(200), json!(body)]])
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execute_rest_query_keeps_default_cat_text_when_server_does_not_return_json() {
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
let body = "health status\ngreen open\n";
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (mut socket, _) = listener.accept().await.unwrap();
|
||||
let mut request = [0_u8; 1024];
|
||||
let read = socket.read(&mut request).await.unwrap();
|
||||
let request = String::from_utf8_lossy(&request[..read]);
|
||||
assert!(request.starts_with("GET /_cat/indices?format=json "));
|
||||
let response = format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
||||
body.len(),
|
||||
|
|
@ -2249,8 +2593,36 @@ mod tests {
|
|||
server.await.unwrap();
|
||||
|
||||
assert_eq!(result.columns, vec!["response"]);
|
||||
assert_eq!(result.rows.len(), 2);
|
||||
assert_eq!(result.rows[1][0], json!("green open app-log-2026-07 42 10mb"));
|
||||
assert_eq!(result.rows, vec![vec![json!("health status")], vec![json!("green open")]]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn execute_rest_query_keeps_explicit_text_cat_response() {
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
|
||||
let body = "health status\ngreen open\n";
|
||||
let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
|
||||
let addr = listener.local_addr().unwrap();
|
||||
let server = tokio::spawn(async move {
|
||||
let (mut socket, _) = listener.accept().await.unwrap();
|
||||
let mut request = [0_u8; 1024];
|
||||
let read = socket.read(&mut request).await.unwrap();
|
||||
let request = String::from_utf8_lossy(&request[..read]);
|
||||
assert!(request.starts_with("GET /_cat/indices?format=txt "));
|
||||
let response = format!(
|
||||
"HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
|
||||
body.len(),
|
||||
body
|
||||
);
|
||||
socket.write_all(response.as_bytes()).await.unwrap();
|
||||
});
|
||||
|
||||
let client = EsClient::new(&format!("http://{addr}"), None, None, false, Duration::from_secs(1));
|
||||
let result = super::execute_rest_query(&client, "GET /_cat/indices?format=txt").await.unwrap();
|
||||
server.await.unwrap();
|
||||
|
||||
assert_eq!(result.columns, vec!["response"]);
|
||||
assert_eq!(result.rows, vec![vec![json!("health status")], vec![json!("green open")]]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ pub mod agent_driver;
|
|||
pub mod clickhouse_driver;
|
||||
pub mod cloudflare_d1;
|
||||
pub use cloudflare_d1 as cloudflare_d1_driver;
|
||||
pub mod document_result;
|
||||
pub mod duckdb_driver;
|
||||
pub mod duckdb_sql;
|
||||
#[cfg(feature = "duckdb-bundled")]
|
||||
|
|
|
|||
|
|
@ -12,15 +12,9 @@ use futures::{io::AsyncReadExt, io::AsyncWriteExt, TryStreamExt};
|
|||
use percent_encoding::percent_decode_str;
|
||||
use std::{collections::HashSet, time::Duration};
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct MongoDocumentResult {
|
||||
pub documents: Vec<serde_json::Value>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub raw_documents: Option<Vec<String>>,
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub extended_documents: Option<Vec<serde_json::Value>>,
|
||||
pub total: u64,
|
||||
}
|
||||
pub use super::document_result::DocumentQueryResult;
|
||||
/// Backward-compatible name for callers of Mongo-specific APIs.
|
||||
pub type MongoDocumentResult = DocumentQueryResult;
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct MongoDropIndexesResult {
|
||||
|
|
@ -676,7 +670,13 @@ pub async fn find_documents(
|
|||
extended_documents.push(Bson::Document(doc).into_relaxed_extjson());
|
||||
}
|
||||
|
||||
Ok(MongoDocumentResult { documents, raw_documents: None, extended_documents: Some(extended_documents), total })
|
||||
Ok(MongoDocumentResult {
|
||||
documents,
|
||||
raw_documents: None,
|
||||
extended_documents: Some(extended_documents),
|
||||
total,
|
||||
total_is_exact: true,
|
||||
})
|
||||
}
|
||||
|
||||
#[derive(Debug, Default, Deserialize)]
|
||||
|
|
@ -812,7 +812,13 @@ pub async fn find_documents_extended_json(
|
|||
documents.push(bson_to_browser_json(&Bson::Document(doc)));
|
||||
}
|
||||
|
||||
Ok(MongoDocumentResult { extended_documents: Some(documents.clone()), documents, raw_documents: None, total })
|
||||
Ok(MongoDocumentResult {
|
||||
extended_documents: Some(documents.clone()),
|
||||
documents,
|
||||
raw_documents: None,
|
||||
total,
|
||||
total_is_exact: true,
|
||||
})
|
||||
}
|
||||
|
||||
/// Run `db.collection.aggregate(pipeline, options)`.
|
||||
|
|
@ -858,6 +864,7 @@ pub async fn aggregate_documents(
|
|||
raw_documents: None,
|
||||
extended_documents: Some(vec![extended]),
|
||||
total: 1,
|
||||
total_is_exact: true,
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -909,7 +916,13 @@ async fn drain_document_cursor(
|
|||
documents.truncate(max_rows);
|
||||
extended_documents.truncate(max_rows);
|
||||
}
|
||||
Ok(MongoDocumentResult { documents, raw_documents: None, extended_documents: Some(extended_documents), total })
|
||||
Ok(MongoDocumentResult {
|
||||
documents,
|
||||
raw_documents: None,
|
||||
extended_documents: Some(extended_documents),
|
||||
total,
|
||||
total_is_exact: true,
|
||||
})
|
||||
}
|
||||
|
||||
fn parse_aggregate_options_document(options_json: Option<&str>) -> Result<Document, String> {
|
||||
|
|
@ -963,7 +976,13 @@ pub async fn distinct(
|
|||
let extended_documents = values.into_iter().map(|value| value.into_relaxed_extjson()).collect::<Vec<_>>();
|
||||
let total = documents.len() as u64;
|
||||
|
||||
Ok(MongoDocumentResult { documents, raw_documents: None, extended_documents: Some(extended_documents), total })
|
||||
Ok(MongoDocumentResult {
|
||||
documents,
|
||||
raw_documents: None,
|
||||
extended_documents: Some(extended_documents),
|
||||
total,
|
||||
total_is_exact: true,
|
||||
})
|
||||
}
|
||||
|
||||
pub async fn create_index(
|
||||
|
|
@ -1329,12 +1348,14 @@ fn single_document_result(document: Option<Document>) -> MongoDocumentResult {
|
|||
raw_documents: None,
|
||||
extended_documents: Some(vec![Bson::Document(document).into_relaxed_extjson()]),
|
||||
total: 1,
|
||||
total_is_exact: true,
|
||||
},
|
||||
None => MongoDocumentResult {
|
||||
documents: Vec::new(),
|
||||
raw_documents: None,
|
||||
extended_documents: Some(Vec::new()),
|
||||
total: 0,
|
||||
total_is_exact: true,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -483,7 +483,7 @@ pub async fn find_documents(
|
|||
collection: &str,
|
||||
skip: u64,
|
||||
limit: i64,
|
||||
) -> Result<crate::db::mongo_driver::MongoDocumentResult, String> {
|
||||
) -> Result<crate::db::document_result::DocumentQueryResult, String> {
|
||||
if client.kind == VectorDbKind::ChromaDb {
|
||||
let start = std::time::Instant::now();
|
||||
let url = format!(
|
||||
|
|
@ -520,11 +520,12 @@ pub async fn find_documents(
|
|||
Value::Object(map)
|
||||
})
|
||||
.collect();
|
||||
return Ok(crate::db::mongo_driver::MongoDocumentResult {
|
||||
return Ok(crate::db::document_result::DocumentQueryResult {
|
||||
documents,
|
||||
raw_documents: None,
|
||||
extended_documents: None,
|
||||
total: result.affected_rows,
|
||||
total_is_exact: true,
|
||||
});
|
||||
}
|
||||
|
||||
|
|
@ -567,11 +568,12 @@ pub async fn find_documents(
|
|||
Value::Object(map)
|
||||
})
|
||||
.collect();
|
||||
Ok(crate::db::mongo_driver::MongoDocumentResult {
|
||||
Ok(crate::db::document_result::DocumentQueryResult {
|
||||
documents,
|
||||
raw_documents: None,
|
||||
extended_documents: None,
|
||||
total: result.affected_rows,
|
||||
total_is_exact: true,
|
||||
})
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
use crate::connection::{AppState, PoolKind};
|
||||
use crate::db::agent_driver::mongo_document_id_params;
|
||||
use crate::db::mongo_driver::MongoDocumentResult;
|
||||
use crate::db::document_result::DocumentQueryResult;
|
||||
use crate::db::{elasticsearch_driver, mongo_driver, vector_driver};
|
||||
|
||||
pub use crate::db::vector_driver::CollectionInfo;
|
||||
|
|
@ -377,7 +377,7 @@ pub async fn find_documents_core(
|
|||
filter: Option<&str>,
|
||||
projection: Option<&str>,
|
||||
sort: Option<&str>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
) -> Result<DocumentQueryResult, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
|
|
@ -425,6 +425,24 @@ pub async fn find_documents_core(
|
|||
}
|
||||
}
|
||||
|
||||
pub async fn count_elasticsearch_documents_core(
|
||||
state: &AppState,
|
||||
connection_id: &str,
|
||||
index: &str,
|
||||
filter: Option<&str>,
|
||||
) -> Result<u64, String> {
|
||||
ensure_document_pool(state, connection_id).await?;
|
||||
let connections = state.connections.read().await;
|
||||
match connections.get(connection_id).ok_or("Not found")? {
|
||||
PoolKind::Elasticsearch(client) => {
|
||||
let client = client.clone();
|
||||
drop(connections);
|
||||
elasticsearch_driver::count_documents(&client, index, filter).await
|
||||
}
|
||||
_ => Err("Not an Elasticsearch connection".to_string()),
|
||||
}
|
||||
}
|
||||
|
||||
fn is_unknown_agent_method_error(error: &str, method: &str) -> bool {
|
||||
let lower = error.to_ascii_lowercase();
|
||||
lower.contains(method) && (lower.contains("unknown method") || lower.contains("method not found"))
|
||||
|
|
|
|||
|
|
@ -476,6 +476,10 @@ async fn main() {
|
|||
.route("/document-store/list-databases", post(routes::document_store::list_databases))
|
||||
.route("/document-store/list-collections", post(routes::document_store::list_collections))
|
||||
.route("/document-store/find-documents", post(routes::document_store::find_documents))
|
||||
.route(
|
||||
"/document-store/elasticsearch-count-documents",
|
||||
post(routes::document_store::elasticsearch_count_documents),
|
||||
)
|
||||
.route("/document-store/list-gridfs-buckets", post(routes::document_store::list_gridfs_buckets))
|
||||
.route("/document-store/create-gridfs-bucket", post(routes::document_store::create_gridfs_bucket))
|
||||
.route("/document-store/delete-gridfs-bucket", post(routes::document_store::delete_gridfs_bucket))
|
||||
|
|
|
|||
|
|
@ -69,6 +69,15 @@ pub struct DocumentFindRequest {
|
|||
pub execution_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct ElasticsearchCountDocumentsRequest {
|
||||
pub connection_id: String,
|
||||
pub index: String,
|
||||
pub filter: Option<String>,
|
||||
pub execution_id: Option<String>,
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct DocumentInsertRequest {
|
||||
|
|
@ -167,7 +176,7 @@ pub async fn list_collections(
|
|||
pub async fn find_documents(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<DocumentFindRequest>,
|
||||
) -> Result<Json<dbx_core::db::mongo_driver::MongoDocumentResult>, AppError> {
|
||||
) -> Result<Json<dbx_core::db::document_result::DocumentQueryResult>, AppError> {
|
||||
let result = run_cancellable(
|
||||
&state,
|
||||
req.execution_id,
|
||||
|
|
@ -187,6 +196,24 @@ pub async fn find_documents(
|
|||
Ok(Json(result))
|
||||
}
|
||||
|
||||
pub async fn elasticsearch_count_documents(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<ElasticsearchCountDocumentsRequest>,
|
||||
) -> Result<Json<u64>, AppError> {
|
||||
let result = run_cancellable(
|
||||
&state,
|
||||
req.execution_id,
|
||||
dbx_core::document_ops::count_elasticsearch_documents_core(
|
||||
&state.app,
|
||||
&req.connection_id,
|
||||
&req.index,
|
||||
req.filter.as_deref(),
|
||||
),
|
||||
)
|
||||
.await?;
|
||||
Ok(Json(result))
|
||||
}
|
||||
|
||||
pub async fn insert_document(
|
||||
State(state): State<Arc<WebState>>,
|
||||
Json(req): Json<DocumentInsertRequest>,
|
||||
|
|
|
|||
|
|
@ -3,7 +3,7 @@ use std::sync::Arc;
|
|||
use tauri::State;
|
||||
|
||||
use crate::commands::connection::{ensure_connection_writable, AppState};
|
||||
use dbx_core::db::mongo_driver::MongoDocumentResult;
|
||||
use dbx_core::db::document_result::DocumentQueryResult;
|
||||
use dbx_core::document_ops::CollectionInfo;
|
||||
|
||||
pub(crate) async fn run_cancellable<T, F>(
|
||||
|
|
@ -58,7 +58,7 @@ pub async fn document_find_documents(
|
|||
projection: Option<String>,
|
||||
sort: Option<String>,
|
||||
execution_id: Option<String>,
|
||||
) -> Result<MongoDocumentResult, String> {
|
||||
) -> Result<DocumentQueryResult, String> {
|
||||
let app = state.inner().clone();
|
||||
run_cancellable(
|
||||
&app,
|
||||
|
|
@ -78,6 +78,23 @@ pub async fn document_find_documents(
|
|||
.await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn elasticsearch_count_documents(
|
||||
state: State<'_, Arc<AppState>>,
|
||||
connection_id: String,
|
||||
index: String,
|
||||
filter: Option<String>,
|
||||
execution_id: Option<String>,
|
||||
) -> Result<u64, String> {
|
||||
let app = state.inner().clone();
|
||||
run_cancellable(
|
||||
&app,
|
||||
execution_id,
|
||||
dbx_core::document_ops::count_elasticsearch_documents_core(&app, &connection_id, &index, filter.as_deref()),
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub async fn document_insert_document(
|
||||
state: State<'_, Arc<AppState>>,
|
||||
|
|
|
|||
|
|
@ -1388,6 +1388,7 @@ pub fn run() {
|
|||
commands::document_cmd::document_list_databases,
|
||||
commands::document_cmd::document_list_collections,
|
||||
commands::document_cmd::document_find_documents,
|
||||
commands::document_cmd::elasticsearch_count_documents,
|
||||
commands::document_cmd::document_list_gridfs_buckets,
|
||||
commands::document_cmd::document_create_gridfs_bucket,
|
||||
commands::document_cmd::document_delete_gridfs_bucket,
|
||||
|
|
|
|||
Loading…
Reference in New Issue