feat(logs): show and filter the source stream in multi-stream search (#14922)

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Mina Karam
2026-09-29 10:38:35 +00:00
committed by GitHub
co-authored by Claude Opus 5.5
parent 97c0905ace
commit 2281dfd30b
12 changed files with 376 additions and 31 deletions
@@ -209,6 +209,7 @@ vi.mock("@/utils/telemetryCorrelation", async () => {
// Import after all vi.mock() declarations so the mocks are in place
import { useSearchQuery } from "./useSearchQuery";
import { Parser as SqlParser } from "@openobserve/node-sql-parser/build/datafusionsql";
import { gt } from "@/types/i18n";
// ---------------------------------------------------------------------------
@@ -827,6 +828,68 @@ describe("useSearchQuery › handleMultiStream WHERE rewrite", () => {
});
});
describe("useSearchQuery › handleMultiStream _stream_name filter", () => {
let buildSearch: ReturnType<typeof useSearchQuery>["buildSearch"];
const parser = new SqlParser();
beforeEach(() => {
mockState = createMockState();
vi.clearAllMocks();
mockSemanticGroups.value = [];
({ buildSearch } = useSearchQuery(gt));
mockState.searchObj.meta.sqlMode = false;
mockState.searchObj.meta.quickMode = false;
mockState.searchObj.data.stream.selectedStream = ["app", "rum"];
mockState.searchObj.data.stream.selectedStreamFields = [
{ name: "level", streams: ["app", "rum"] },
];
// The real parser, so the arms are exactly what the backend receives.
fnParsedSQLMock.mockImplementation((sql?: string) => (sql ? parser.astify(sql) : {}) as any);
fnUnparsedSQLMock.mockImplementation((ast: any) => parser.sqlify(ast));
});
const armFor = (sql: string, stream: string) =>
sql.split(" UNION ALL BY NAME ").find((arm: string) => arm.includes(`FROM "${stream}"`));
it("accepts a _stream_name filter and resolves it to each stream's own name", () => {
mockState.searchObj.data.query = "_stream_name = 'rum'";
const result = buildSearch(false, false);
expect(result).not.toBeNull();
expect(mockState.searchObj.data.filterErrMsg).toBe("");
const sql = getSql(result);
expect(armFor(sql, "app")).toContain("WHERE 'app' = 'rum'");
expect(armFor(sql, "rum")).toContain("WHERE 'rum' = 'rum'");
expect(armFor(sql, "rum")).toContain("'rum' as _stream_name");
});
it("keeps other conditions and excludes with !=", () => {
mockState.searchObj.data.query = "level = 'error' and _stream_name != 'rum'";
const sql = getSql(buildSearch(false, false));
expect(armFor(sql, "app")).toMatch(/level.* = 'error' AND 'app' != 'rum'/);
expect(armFor(sql, "rum")).toMatch(/level.* = 'error' AND 'rum' != 'rum'/);
});
it("still rewrites an equivalent field alongside _stream_name", () => {
mockState.searchObj.data.stream.selectedStreamFields = [
{ name: "msg", streams: ["app"] },
{ name: "message", streams: ["rum"] },
];
mockSemanticGroups.value = [{ id: "group-1", display: "Message", fields: ["msg", "message"] }];
mockState.searchObj.data.query = "msg = 'boom' and _stream_name = 'rum'";
const sql = getSql(buildSearch(false, false));
expect(armFor(sql, "rum")).toMatch(/message.* = 'boom' AND 'rum' = 'rum'/);
expect(armFor(sql, "app")).toMatch(/msg.* = 'boom' AND 'app' = 'rum'/);
});
});
describe("useSearchQuery › buildSearch › LIMIT in filter mode", () => {
let buildSearch: ReturnType<typeof useSearchQuery>["buildSearch"];
+13 -5
View File
@@ -31,6 +31,11 @@ import { Parser as SqlParser } from "@openobserve/node-sql-parser/build/datafusi
import { buildContextualSqlMessage, isParserLimitation } from "@/utils/query/sqlDiagnostics";
import { maxParenDepth, SQL_PARSE_MAX_DEPTH } from "@/utils/query/sqlComplexity";
import { raw, type TranslateFn } from "@/types/i18n";
import {
STREAM_NAME_FIELD,
referencesStreamName,
replaceStreamNameRefsInWhere,
} from "@/utils/logs/streamNameColumn";
// Walk the WHERE clause AST and replace column references whose name matches
// a key in the fieldMapping (original field → stream-specific field).
@@ -619,10 +624,10 @@ export const useSearchQuery = (t: TranslateFn) => {
let finalQuery: string = preSQLQuery.replace("[INDEX_NAME]", item);
// Per-stream WHERE rewrite: if this stream has equivalent field names
// for any filter fields (reverse semantic group mapping), swap them in.
if (multiStreamFieldMapping?.has(item)) {
const mapping = multiStreamFieldMapping.get(item)!;
// for any filter fields (reverse semantic group mapping), swap them in,
// and turn a _stream_name filter into this stream's name.
const mapping = multiStreamFieldMapping?.get(item);
if (mapping || referencesStreamName(whereClause)) {
// Build a parsable SQL by temporarily replacing template placeholders
const hasFieldListPlaceholder = finalQuery.includes("[FIELD_LIST]");
if (hasFieldListPlaceholder) {
@@ -631,7 +636,8 @@ export const useSearchQuery = (t: TranslateFn) => {
const parsed = fnParsedSQL(finalQuery);
if (parsed?.where) {
replaceColumnRefsInWhere(parsed.where, mapping);
if (mapping) replaceColumnRefsInWhere(parsed.where, mapping);
parsed.where = replaceStreamNameRefsInWhere(parsed.where, item);
finalQuery = fnUnparsedSQL(parsed);
finalQuery = finalQuery.replace(/`/g, '"');
@@ -714,6 +720,8 @@ export const useSearchQuery = (t: TranslateFn) => {
for (const fieldObj of searchObj.data.stream.filteredField) {
const fieldName = fieldObj.expr.value;
// Not a stored field: handleMultiStream resolves it per stream.
if (fieldName === STREAM_NAME_FIELD) continue;
const filteredFields: any = searchObj.data.stream.selectedStreamFields.filter(
(field: any) => field.name === fieldName,
);
@@ -19,6 +19,7 @@ import { mount } from "@vue/test-utils";
import { createI18n } from "vue-i18n";
import store from "@/test/unit/helpers/store";
import useStreamFields from "./useStreamFields";
import { searchState } from "./searchState";
// Create i18n instance
const i18n = createI18n({
@@ -124,4 +125,44 @@ describe("useStreamFields Composable", () => {
expect(loadStreamLists).toHaveBeenCalled;
});
});
describe("updateGridColumns stream name column", () => {
const { searchObj } = searchState();
beforeEach(() => {
searchObj.meta.sqlMode = false;
searchObj.data.stream.selectedFields = [];
searchObj.data.stream.selectedStream = ["app", "rum"];
searchObj.data.queryResults = {
hits: [
{ _timestamp: 2, _stream_name: "rum", message: "b" },
{ _timestamp: 1, _stream_name: "app", message: "a" },
],
} as any;
});
const columnIds = () => searchObj.data.resultGrid.columns.map((column: any) => column.id);
it("adds _stream_name after the timestamp for a multi-stream result", () => {
wrapper.vm.updateGridColumns();
expect(columnIds()).toEqual(["_timestamp", "_stream_name", "source"]);
});
it("adds _stream_name before the selected fields", () => {
searchObj.data.stream.selectedFields = ["message"];
wrapper.vm.updateGridColumns();
expect(columnIds()).toEqual(["_timestamp", "_stream_name", "message"]);
});
it("leaves a single-stream result unchanged", () => {
searchObj.data.stream.selectedStream = ["app"];
wrapper.vm.updateGridColumns();
expect(columnIds()).toEqual(["_timestamp", "source"]);
});
});
});
@@ -44,6 +44,7 @@ import {
CATEGORY,
type FieldObj,
} from "@/utils/fieldCategories";
import { buildStreamNameColumn, shouldShowStreamNameColumn } from "@/utils/logs/streamNameColumn";
import type { KeyFieldsConfig, FieldGroupingConfig } from "@/composables/useServiceCorrelation";
import { useServiceCorrelation } from "@/composables/useServiceCorrelation";
@@ -1043,6 +1044,13 @@ export const useStreamFields = () => {
const parsedSQL: any = fnParsedSQL();
// Multi-stream hits carry _stream_name; show it next to the timestamp.
const showStreamName = shouldShowStreamNameColumn(
searchObj.data.stream.selectedStream,
searchObj.data.queryResults?.hits,
selectedFields,
);
// By default when no fields are selected. Timestamp and Source will be visible. If user selects field, then only selected fields will be visible in table
// In SQL and Quick mode.
// If user adds timestamp manually then only we get it in response.
@@ -1093,6 +1101,10 @@ export const useStreamFields = () => {
});
}
if (showStreamName) {
searchObj.data.resultGrid.columns.push(buildStreamNameColumn());
}
if (selectedFields.length == 0) {
searchObj.data.resultGrid.columns.push({
name: "source",
@@ -1155,6 +1167,10 @@ export const useStreamFields = () => {
});
}
if (showStreamName) {
searchObj.data.resultGrid.columns.push(buildStreamNameColumn());
}
let sizes: any;
// selectedStream array is coerced to its comma-joined string form as key
if (
+8 -4
View File
@@ -184,8 +184,9 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</template>
<ODropdownItem
v-if="
searchObj.data.stream.selectedStreamFields.some((item: any) =>
item.name === row.field ? item.isSchemaField : '',
isFilterableLogField(
row.field,
searchObj.data.stream.selectedStreamFields,
)
"
data-test="log-details-include-field-btn"
@@ -196,8 +197,9 @@ along with this program. If not, see <http://www.gnu.org/licenses/>.
</ODropdownItem>
<ODropdownItem
v-if="
searchObj.data.stream.selectedStreamFields.some((item: any) =>
item.name === row.field ? item.isSchemaField : '',
isFilterableLogField(
row.field,
searchObj.data.stream.selectedStreamFields,
)
"
data-test="log-details-exclude-field-btn"
@@ -515,6 +517,7 @@ import OTable from "@/lib/core/Table/OTable.vue";
import OSearchInput from "@/lib/forms/SearchInput/OSearchInput.vue";
import OSeparator from "@/lib/core/Separator/OSeparator.vue";
import { isSafeNavigableUrl } from "@/utils/safeUrl";
import { isFilterableLogField } from "@/utils/logs/streamNameColumn";
const defaultValue: any = () => {
return {
data: {},
@@ -1106,6 +1109,7 @@ export default defineComponent({
recordSizeOptions,
getImageURL,
tableDropdownOpenMap,
isFilterableLogField,
shouldWrapValues,
toggleWrapLogDetails,
copyContentToClipboard,
+7
View File
@@ -638,6 +638,13 @@ describe("JsonPreview Component", () => {
});
describe("Field Operations", () => {
it("offers include/exclude for schema fields in every stream and for _stream_name", () => {
expect(wrapper.vm.canFilterOnField("field1")).toBe(true);
expect(wrapper.vm.canFilterOnField("field3")).toBe(false);
expect(wrapper.vm.canFilterOnField("nested")).toBe(false);
expect(wrapper.vm.canFilterOnField("_stream_name")).toBe(true);
});
it("should return correct label for adding field to table", () => {
wrapper.vm.searchObj.data.stream.selectedFields = ["field1"];
+10 -14
View File
@@ -102,13 +102,7 @@
</OButton>
</template>
<ODropdownItem
v-if="
!hideSearchTermActions &&
searchObj.data.stream.selectedStreamFields.some((item: any) =>
item.name === key ? item.isSchemaField : '',
) &&
multiStreamFields.includes(key)
"
v-if="!hideSearchTermActions && canFilterOnField(key)"
data-test="log-details-include-field-btn"
@select.stop="addSearchTerm(key, value[key], 'include')"
>
@@ -116,13 +110,7 @@
{{ t("common.includeSearchTerm") }}
</ODropdownItem>
<ODropdownItem
v-if="
!hideSearchTermActions &&
searchObj.data.stream.selectedStreamFields.some((item: any) =>
item.name === key ? item.isSchemaField : '',
) &&
multiStreamFields.includes(key)
"
v-if="!hideSearchTermActions && canFilterOnField(key)"
data-test="log-details-exclude-field-btn"
@select.stop="addSearchTerm(key, value[key], 'exclude')"
>
@@ -283,6 +271,7 @@ import { copyToClipboard } from "@/utils/clipboard";
import { timestampToTimezoneDate } from "@/utils/timezone";
import { toast } from "@/lib/feedback/Toast/useToast";
import { isSafeNavigableUrl } from "@/utils/safeUrl";
import { isFilterableLogField, STREAM_NAME_FIELD } from "@/utils/logs/streamNameColumn";
export default {
name: "JsonPreview",
@@ -678,6 +667,12 @@ export default {
}
};
// `_stream_name` is in no schema; a filter on it is rewritten per stream.
const canFilterOnField = (key: string) =>
key === STREAM_NAME_FIELD ||
(isFilterableLogField(key, searchObj.data.stream.selectedStreamFields) &&
multiStreamFields.value.includes(key));
const updateMultiStreamFields = () => {
searchObj.data.stream.selectedStreamFields.forEach((item: any) => {
if (item.streams?.length == searchObj.data.stream.selectedStream.length) {
@@ -871,6 +866,7 @@ export default {
store,
searchObj,
multiStreamFields,
canFilterOnField,
redirectToTraces,
openCorrelation,
filteredTracesStreamOptions,
+2 -4
View File
@@ -888,6 +888,7 @@ import CellActions from "@/plugins/logs/data-table/CellActions.vue";
import O2AIContextAddBtn from "@/components/common/O2AIContextAddBtn.vue";
import { useLogsHighlighter } from "@/composables/useLogsHighlighter";
import { extractStatusFromLog } from "@/utils/logs/statusParser";
import { isFilterableLogField } from "@/utils/logs/streamNameColumn";
import useBreakpoint from "@/composables/useBreakpoint";
import {
buildPatternVolumeContext,
@@ -2237,10 +2238,7 @@ export default defineComponent({
const contextCellIsStreamField = computed(() => {
const columnId = contextCell.value?.columnId;
if (!columnId) return false;
return (
searchObj.data.stream.selectedStreamFields?.find((field: any) => field.name === columnId)
?.isSchemaField ?? false
);
return isFilterableLogField(columnId, searchObj.data.stream.selectedStreamFields);
});
// Mirrors O2AIContextAddBtn's own gate — the AI actions only exist on
@@ -134,6 +134,17 @@ describe("CellActions", () => {
expect(wrapper.vm.isStreamField).toBe(false);
});
it("returns true for _stream_name, which is in no stream schema", () => {
const wrapper = mountComponent({
column: { id: "_stream_name" },
row: { _stream_name: "app" },
selectedStreamFields: [{ name: "status", isSchemaField: true }],
});
expect(wrapper.vm.isStreamField).toBe(true);
expect(wrapper.find('[data-test="log-details-include-field-app"]').exists()).toBe(true);
expect(wrapper.find('[data-test="log-details-exclude-field-app"]').exists()).toBe(true);
});
it("returns false when selectedStreamFields is empty", () => {
const wrapper = mountComponent({
selectedStreamFields: [],
@@ -58,6 +58,7 @@
import { computed } from "vue";
import type { PropType } from "vue";
import { useI18nTyped } from "@/types/i18n";
import { isFilterableLogField } from "@/utils/logs/streamNameColumn";
import EqualIcon from "@/components/icons/EqualIcon.vue";
import NotEqualIcon from "@/components/icons/NotEqualIcon.vue";
import O2AIContextAddBtn from "@/components/common/O2AIContextAddBtn.vue";
@@ -112,8 +113,7 @@ const sendToAiChat = (value: any) => {
emit("sendToAiChat", value);
};
const isStreamField = computed(() => {
const field: any = props.selectedStreamFields?.find((item: any) => item.name === props.column.id);
return field?.isSchemaField ?? false;
});
const isStreamField = computed(() =>
isFilterableLogField(props.column.id, props.selectedStreamFields),
);
</script>
+116
View File
@@ -0,0 +1,116 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
import { describe, it, expect } from "vitest";
import { Parser } from "@openobserve/node-sql-parser/build/datafusionsql";
import {
STREAM_NAME_FIELD,
buildStreamNameColumn,
isFilterableLogField,
referencesStreamName,
replaceStreamNameRefsInWhere,
shouldShowStreamNameColumn,
} from "./streamNameColumn";
const parser = new Parser();
// Parse, rewrite for one stream and print back the WHERE the backend receives.
const rewriteWhere = (where: string, stream: string): string => {
const ast: any = parser.astify(`select * from "${stream}" where ${where}`);
ast.where = replaceStreamNameRefsInWhere(ast.where, stream);
return parser.sqlify(ast).split(/ WHERE /i)[1];
};
describe("referencesStreamName", () => {
it("matches the column as a whole word only", () => {
expect(referencesStreamName("_stream_name = 'a'")).toBe(true);
expect(referencesStreamName(`"_stream_name" != 'a'`)).toBe(true);
expect(referencesStreamName("my_stream_name_x = 'a'")).toBe(false);
expect(referencesStreamName("level = 'error'")).toBe(false);
});
});
describe("replaceStreamNameRefsInWhere", () => {
it("replaces an equality with the stream's own name", () => {
expect(rewriteWhere("_stream_name = 'app'", "app")).toBe("'app' = 'app'");
expect(rewriteWhere("_stream_name = 'app'", "rum")).toBe("'rum' = 'app'");
});
it("replaces the column inside IN, NOT and combined conditions", () => {
const where = "level = 'error' AND (_stream_name IN ('a', 'b') OR NOT (_stream_name != 'c'))";
const sql = rewriteWhere(where, "b");
expect(sql).not.toContain(STREAM_NAME_FIELD);
expect(sql).toContain("'b' IN ('a', 'b')");
expect(sql).toContain("'b' != 'c'");
expect(sql).toContain("level");
});
it("replaces a quoted column reference", () => {
expect(rewriteWhere(`"_stream_name" = 'app'`, "app")).toBe("'app' = 'app'");
});
it("leaves other columns alone", () => {
const sql = rewriteWhere("stream_name = 'x' AND level = 'error'", "app");
expect(sql).toContain("stream_name");
expect(sql).not.toContain("'app'");
});
});
describe("shouldShowStreamNameColumn", () => {
const hits = [{ _timestamp: 1, _stream_name: "a" }];
it("shows the column for a multi-stream result that carries the stream name", () => {
expect(shouldShowStreamNameColumn(["a", "b"], hits, [])).toBe(true);
});
it("hides it for a single stream, missing tags, or an explicitly selected field", () => {
expect(shouldShowStreamNameColumn(["a"], hits, [])).toBe(false);
expect(shouldShowStreamNameColumn(["a", "b"], [{ _timestamp: 1 }], [])).toBe(false);
expect(shouldShowStreamNameColumn(["a", "b"], undefined, [])).toBe(false);
expect(shouldShowStreamNameColumn(["a", "b"], hits, [STREAM_NAME_FIELD])).toBe(false);
});
});
describe("buildStreamNameColumn", () => {
it("is a closable, unsortable column that reads the stream name", () => {
const column = buildStreamNameColumn();
expect(column.id).toBe(STREAM_NAME_FIELD);
expect(column.meta.closable).toBe(true);
expect(column.sortable).toBe(false);
expect(column.accessorFn({ _stream_name: "app" })).toBe("app");
expect(column.accessorFn({})).toBe("");
});
});
describe("isFilterableLogField", () => {
const fields = [
{ name: "status", isSchemaField: true },
{ name: "message", isSchemaField: false },
];
it("allows schema fields and the stream name, which is in no schema", () => {
expect(isFilterableLogField("status", fields)).toBe(true);
expect(isFilterableLogField(STREAM_NAME_FIELD, fields)).toBe(true);
expect(isFilterableLogField(STREAM_NAME_FIELD, undefined)).toBe(true);
});
it("rejects non-schema and unknown fields", () => {
expect(isFilterableLogField("message", fields)).toBe(false);
expect(isFilterableLogField("unknown", fields)).toBe(false);
expect(isFilterableLogField("status", undefined)).toBe(false);
});
});
+85
View File
@@ -0,0 +1,85 @@
// Copyright 2026 OpenObserve Inc.
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
import { raw } from "@/types/i18n";
// Multi-stream search projects `'<stream>' as _stream_name`, tagging each hit with its stream.
export const STREAM_NAME_FIELD = "_stream_name";
const STREAM_NAME_PATTERN = new RegExp(`\\b${STREAM_NAME_FIELD}\\b`);
export const referencesStreamName = (where: string): boolean => STREAM_NAME_PATTERN.test(where);
const columnName = (node: any): string | null => {
const col = node.column;
if (typeof col === "string") return col.replace(/^"|"$/g, "");
return col?.expr?.value != null ? String(col.expr.value) : null;
};
// A per-stream WHERE cannot use the alias, so stream b's `_stream_name = 'a'` becomes `'b' = 'a'`.
export const replaceStreamNameRefsInWhere = (node: any, stream: string): any => {
if (Array.isArray(node)) {
return node.map((child) => replaceStreamNameRefsInWhere(child, stream));
}
if (!node || typeof node !== "object") return node;
if (node.type === "column_ref" && columnName(node) === STREAM_NAME_FIELD) {
return { type: "single_quote_string", value: stream };
}
// `value` holds the list of an IN (...) or of function arguments such as NOT (...).
for (const key of ["left", "right", "expr", "args", "value"]) {
if (node[key] && typeof node[key] === "object") {
node[key] = replaceStreamNameRefsInWhere(node[key], stream);
}
}
return node;
};
// Include/exclude gate: schema fields, plus `_stream_name`, which is in no schema.
export const isFilterableLogField = (
name: string | number | undefined,
streamFields: Array<{ name: string; isSchemaField?: boolean }> | undefined,
): boolean =>
name === STREAM_NAME_FIELD ||
(streamFields?.find((field) => field.name === name)?.isSchemaField ?? false);
export const shouldShowStreamNameColumn = (
selectedStreams: string[],
hits: any[] | undefined,
selectedFields: string[],
): boolean =>
selectedStreams.length > 1 &&
!selectedFields.includes(STREAM_NAME_FIELD) &&
!!hits?.some((hit: any) => hit?.[STREAM_NAME_FIELD] != null);
export const formatStreamName = (stream: string): string => stream;
// Closable, so the cell gets its actions; include/exclude filter on the raw value.
export const buildStreamNameColumn = () => ({
name: STREAM_NAME_FIELD,
id: STREAM_NAME_FIELD,
accessorFn: (row: any) =>
row?.[STREAM_NAME_FIELD] == null ? "" : formatStreamName(String(row[STREAM_NAME_FIELD])),
header: raw(STREAM_NAME_FIELD),
sortable: false,
enableResizing: true,
meta: {
closable: true,
showWrap: false,
wrapContent: false,
},
size: 180,
});