From 63d6db94fe2267fa338182a3554ef97669c44fdb Mon Sep 17 00:00:00 2001 From: Omkar Kesarkhane <127078473+omkarK06@users.noreply.github.com> Date: Sat, 31 May 2025 18:50:26 +0530 Subject: [PATCH] fix: http2 streaming pagination fix (#6864) --- web/src/composables/useLogs.ts | 114 +++++++++++++++++--------------- web/src/ts/interfaces/query.ts | 2 +- web/src/workers/streamWorker.js | 2 + 3 files changed, 64 insertions(+), 54 deletions(-) diff --git a/web/src/composables/useLogs.ts b/web/src/composables/useLogs.ts index 7c627bbabe..85b18301e7 100644 --- a/web/src/composables/useLogs.ts +++ b/web/src/composables/useLogs.ts @@ -313,10 +313,6 @@ const { const { fetchQueryDataWithHttpStream, - cancelStreamQueryBasedOnRequestId, - closeStreamWithError, - closeStream, - resetAuthToken, } = useStreamingSearch(); const searchPartitionMap = reactive<{ [key: string]: number }>({}); @@ -574,7 +570,7 @@ const useLogs = () => { await loadStreamLists(); return; } catch (e: any) { - console.log("Error while getting stream list"); + console.error("Error while getting stream list", e); } }; @@ -5033,6 +5029,11 @@ const useLogs = () => { } const payload = buildWebSocketPayload(queryReq, isPagination, "search"); + + if(shouldGetPageCount(queryReq, fnParsedSQL())) { + queryReq.query.size = queryReq.query.size + 1; + } + const requestId = initializeSearchConnection(payload); if (!requestId) { @@ -5264,12 +5265,12 @@ const useLogs = () => { } else { searchObj.data.queryResults.hits = response.content.results.hits; } - } + - // // We are storing time_offset for the context of pagecount, to get the partial pagecount - // if (searchObj.data.queryResults) { - // searchObj.data.queryResults.time_offset = response.content?.time_offset; - // } + if(shouldGetPageCount(payload.queryReq, fnParsedSQL()) && (searchObj.data.queryResults.hits.length === payload.queryReq.query.size)) { + searchObj.data.queryResults.hits = searchObj.data.queryResults.hits.slice(0, payload.queryReq.query.size - 1); + } + } processPostPaginationData(); } @@ -5333,6 +5334,10 @@ const useLogs = () => { searchObj.data.queryResults = response.content.results; } } + + if(shouldGetPageCount(payload.queryReq, fnParsedSQL()) && (response.content.results.total === payload.queryReq.query.size)) { + searchObj.data.queryResults.pageCountTotal = payload.queryReq.query.size * searchObj.data.resultGrid.currentPage; + } } // We are storing time_offset for the context of pagecount, to get the partial pagecount @@ -5470,7 +5475,6 @@ const useLogs = () => { searchObj.data.queryResults.aggs.push(...response.content.results.hits); } - // if total records in partition is greater than recordsPerPage then we need to update pagination // setting up forceFlag to true to update pagination as we have check for pagination already created more than currentPage + 3 pages. refreshPagination(regeratePaginationFlag); @@ -5704,7 +5708,6 @@ const useLogs = () => { ////// Update results /////// updateResult(queryReq, response, isPagination, appendResult); - // If its a pagination request, then append if (!isPagination) { searchObj.data.queryResults.pagination = []; @@ -5876,19 +5879,20 @@ const useLogs = () => { searchObj.data.histogram.errorDetail = ""; }; + const shouldShowHistogram = (parsedSQL: any) => { + return (isHistogramDataMissing(searchObj) && + isHistogramEnabled(searchObj) && + isSingleStreamSelected(searchObj) && + (!searchObj.meta.sqlMode || + isNonAggregatedSQLMode(searchObj, parsedSQL))) || + (isHistogramEnabled(searchObj) && + isSingleStreamSelected(searchObj) && + !searchObj.meta.sqlMode); + } + const processHistogramRequest = async (queryReq: SearchRequestPayload) => { const parsedSQL: any = fnParsedSQL(); - const shouldShowHistogram = - (isHistogramDataMissing(searchObj) && - isHistogramEnabled(searchObj) && - isSingleStreamSelected(searchObj) && - (!searchObj.meta.sqlMode || - isNonAggregatedSQLMode(searchObj, parsedSQL))) || - (isHistogramEnabled(searchObj) && - isSingleStreamSelected(searchObj) && - !searchObj.meta.sqlMode); - if (searchObj.data.stream.selectedStream.length > 1) { const errMsg = "Histogram is not available for multi stream search."; resetHistogramWithError(errMsg, 0); @@ -5898,9 +5902,13 @@ const useLogs = () => { return; } + const isFromZero = queryReq.query.from == 0 && + searchObj.data.queryResults.hits?.length > 0 + + const _shouldShowHistogram = shouldShowHistogram(parsedSQL); searchObj.data.queryResults.aggs = []; - if (shouldShowHistogram) { + if (_shouldShowHistogram) { searchObj.meta.refreshHistogram = false; if (searchObj.data.queryResults.hits?.length > 0) { searchObjDebug["histogramStartTime"] = performance.now(); @@ -5930,15 +5938,7 @@ const useLogs = () => { resetHistogramWithError("Histogram unavailable for CTEs, DISTINCT and LIMIT queries.", -1); searchObj.meta.histogramDirtyFlag = false; } else if (searchObj.meta.sqlMode && (isDistinctQuery(parsedSQL) || isWithQuery(parsedSQL))) { - let aggFlag = false; - if (parsedSQL) { - aggFlag = hasAggregation(parsedSQL?.columns); - } - if ( - queryReq.query.from == 0 && - searchObj.data.queryResults.hits?.length > 0 && - !aggFlag - ) { + if (shouldGetPageCount(queryReq, parsedSQL) && isFromZero) { setTimeout(async () => { searchObjDebug["pagecountStartTime"] = performance.now(); getPageCountThroughSocket(queryReq); @@ -5960,15 +5960,7 @@ const useLogs = () => { } searchObj.meta.histogramDirtyFlag = false; } else { - let aggFlag = false; - if (parsedSQL) { - aggFlag = hasAggregation(parsedSQL?.columns); - } - if ( - queryReq.query.from == 0 && - searchObj.data.queryResults.hits?.length > 0 && - !aggFlag - ) { + if(shouldGetPageCount(queryReq, parsedSQL) && isFromZero) { setTimeout(async () => { searchObjDebug["pagecountStartTime"] = performance.now(); getPageCountThroughSocket(queryReq); @@ -5998,7 +5990,7 @@ const useLogs = () => { } function isHistogramDataMissing(searchObj: any) { - return searchObj.data.queryResults.aggs === undefined; + return searchObj.data.queryResults.aggs === undefined || searchObj.data.queryResults.aggs.length === 0; } const handleSearchClose = (payload: any, response: any) => { @@ -6116,6 +6108,13 @@ const useLogs = () => { return errorMsg; }; + const getAggsTotal = () => { + return (searchObj.data.queryResults.aggs || []).reduce( + (acc: number, item: { zo_sql_num: number; }) => acc + item.zo_sql_num, + 0, + ); + }; + const refreshPagination = (regenrateFlag: boolean = false) => { try { const { rowsPerPage } = searchObj.meta.resultGrid; @@ -6127,17 +6126,12 @@ const useLogs = () => { let total = 0; let totalPages = 0; - total = (searchObj.data.queryResults.aggs || []).reduce( - ( - acc: number, - item: { - zo_sql_num: number; - }, - ) => { - return acc + item.zo_sql_num; - }, - 0, - ); + total = getAggsTotal(); + + if((searchObj.data.queryResults.pageCountTotal || -1) > total) { + total = searchObj.data.queryResults.pageCountTotal; + } + searchObj.data.queryResults.total = total; searchObj.data.queryResults.pagination = []; @@ -6192,6 +6186,20 @@ const useLogs = () => { } }; + const shouldGetPageCount = (queryReq: any, parsedSQL: any) => { + if(shouldShowHistogram(parsedSQL) || (searchObj.meta.sqlMode && isLimitQuery(parsedSQL))) { + return false; + } + let aggFlag = false; + if (parsedSQL) { + aggFlag = hasAggregation(parsedSQL?.columns); + } + if (!aggFlag) { + return true; + } + return false; + } + const getPageCountThroughSocket = async (queryReq: any) => { if ( searchObj.data.queryResults.total > diff --git a/web/src/ts/interfaces/query.ts b/web/src/ts/interfaces/query.ts index 19a2b07769..846ca63e04 100644 --- a/web/src/ts/interfaces/query.ts +++ b/web/src/ts/interfaces/query.ts @@ -67,7 +67,7 @@ export interface WebSocketSearchResponse { export interface WebSocketSearchPayload { queryReq: SearchRequestPayload; - type: "search" | "histogram" | "pageCount"; + type: "search" | "histogram" | "pageCount" | "values"; isPagination: boolean; traceId: string; org_id: string; diff --git a/web/src/workers/streamWorker.js b/web/src/workers/streamWorker.js index 8264b84c03..69c5b1cd32 100644 --- a/web/src/workers/streamWorker.js +++ b/web/src/workers/streamWorker.js @@ -78,6 +78,8 @@ async function processStream(traceId, reader) { if(msgLines.length > 1){ const eventType = msgLines[0].startsWith('event:') ? msgLines[0].slice(7).trim() : msgLines[0].slice(6).trim(); + //TODO: Logic is duplicated for event:search_response and event:search_response_hits + // Create method to handle this if (msgLines[1]?.startsWith('data:') || msgLines[1]?.startsWith('data: ')) { const data = msgLines[1]?.startsWith('data:') ? msgLines[1]?.slice(6) : msgLines[1]?.slice(5);