starred/openobserve
1
0
Fork 0
mirror of https://github.com/openobserve/openobserve.git synced 2026-11-05 08:09:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: http2 streaming pagination fix (#6864)

Browse Source
This commit is contained in:
Omkar Kesarkhane
2025-05-31 21:20:26 +08:00
committed by GitHub
parent 64e3d7d101
commit 63d6db94fe
3 changed files with 64 additions and 54 deletions
Download Patch File Download Diff File Expand all files Collapse all files
web/src/composables/useLogs.ts
+61 -53
View File
@@ -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 >
web/src/ts/interfaces/query.ts
+1 -1
View File
@@ -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;
web/src/workers/streamWorker.js
+2
View File
@@ -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);