diff --git a/.changeset/multi-source-filters.md b/.changeset/multi-source-filters.md new file mode 100644 index 0000000000..7bb2603e91 --- /dev/null +++ b/.changeset/multi-source-filters.md @@ -0,0 +1,10 @@ +--- +'@hyperdx/app': minor +--- + +The filters sidebar now works when searching multiple sources. Facet fields +and values merge across the selected sources, and checking a value filters +every source that has the field. A source whose table lacks a filtered column +is excluded from the results with a visible reason on its status chip instead +of silently returning unfiltered rows. Filter pills and add-to-filter from the +row side panel work in multi-source mode too. diff --git a/.changeset/multi-source-search.md b/.changeset/multi-source-search.md new file mode 100644 index 0000000000..9d4868f8f4 --- /dev/null +++ b/.changeset/multi-source-search.md @@ -0,0 +1,15 @@ +--- +'@hyperdx/app': minor +'@hyperdx/common-utils': minor +--- + +Search across multiple sources at once. The search page's source selector can +now expand into a multi-select (up to 3 log/trace sources): results interleave +into one timestamp-ordered timeline with a per-row source badge, normalized +columns (Timestamp, Source, Service, Level, Message, and Duration when traces +are included), a histogram stacked by source, and an add-column picker over the +union of the selected sources' columns. Each source runs its own query +pipeline — sources on different connections work, and a failing source shows a +status chip instead of failing the whole search. Multi-source mode is +Lucene-only and shareable via URL; saved searches and alerts remain +single-source for now. diff --git a/packages/app/src/DBSearchPage.tsx b/packages/app/src/DBSearchPage.tsx index cdae127d1e..0a08d25e9e 100644 --- a/packages/app/src/DBSearchPage.tsx +++ b/packages/app/src/DBSearchPage.tsx @@ -31,7 +31,11 @@ import { ColumnMeta, } from '@hyperdx/common-utils/dist/clickhouse'; import { tcFromSource } from '@hyperdx/common-utils/dist/core/metadata'; -import { buildSearchChartConfig } from '@hyperdx/common-utils/dist/core/searchChartConfig'; +import { + ALERT_COUNT_DEFAULT_SELECT, + buildMultiSourceSearchConfig, + buildSearchChartConfig, +} from '@hyperdx/common-utils/dist/core/searchChartConfig'; import { aliasMapToWithClauses, isBrowser, @@ -95,21 +99,36 @@ import { ErrorBoundary } from '@/components/Error/ErrorBoundary'; import { FavoriteButton } from '@/components/FavoriteButton'; import ResourceTerraformPopover from '@/components/Iac/ResourceTerraformPopover'; import { InputControlled } from '@/components/InputControlled'; +import MultiSourceColumnPicker from '@/components/MultiSourceColumnPicker'; +import MultiSourceSearchFilters from '@/components/MultiSourceSearchFilters'; +import { + MultiSourceTimeChart, + MultiSourceTotalCountChart, +} from '@/components/MultiSourceTimeChart'; import OnboardingModal from '@/components/OnboardingModal'; import SearchWhereInput, { getStoredLanguage, } from '@/components/SearchInput/SearchWhereInput'; import SearchPageActionBar from '@/components/SearchPageActionBar'; +import SearchResultsTable from '@/components/SearchResultsTable'; import SearchTotalCountChart from '@/components/SearchTotalCountChart'; +import { SourceMultiSelectControlled } from '@/components/SourceMultiSelect'; import { TableSourceForm } from '@/components/Sources/SourceForm'; import { SourceSelectControlled } from '@/components/SourceSelect'; import { SQLInlineEditorControlled } from '@/components/SQLEditor/SQLInlineEditor'; import { Tags } from '@/components/Tags'; import { TimePicker } from '@/components/TimePicker'; import { IS_LOCAL_MODE } from '@/config'; +import { MAX_SEARCH_SOURCES } from '@/defaults'; import { useAliasMapFromChartConfig } from '@/hooks/useChartConfig'; import { useExplainQuery } from '@/hooks/useExplainQuery'; +import { + resolveExtraColumnsForSource, + unresolvedFilterColumns, + useMultiSourceColumns, +} from '@/hooks/useMultiSourceSearch'; import { useResolvedSourceParam } from '@/hooks/useResolvedSourceParam'; +import { useResolvedSourcesParam } from '@/hooks/useResolvedSourcesParam'; import { withAppNav } from '@/layout'; import { useCreateSavedSearch, @@ -133,7 +152,6 @@ import { } from '@/utils'; import ChartSQLPreview, { SQLPreview } from './components/ChartSQLPreview'; -import DBSqlRowTableWithSideBar from './components/DBSqlRowTableWithSidebar'; import PatternTable from './components/PatternTable'; import { DBSearchHeatmapChart } from './components/Search/DBSearchHeatmapChart'; import DirectTraceSidePanel from './components/Search/DirectTraceSidePanel'; @@ -181,6 +199,10 @@ const ALLOWED_SOURCE_KINDS = [SourceKind.Log, SourceKind.Trace]; const SearchConfigSchema = z.object({ select: z.string(), source: z.string(), + // Multi-source search: the full selection (2+ engages multi mode). The + // single `source` field stays the primary (= sources[0]) so every + // single-source code path keeps working unchanged. + sources: z.array(z.string()), where: z.string(), whereLanguage: z.enum(['sql', 'lucene']), orderBy: z.string(), @@ -865,6 +887,25 @@ function optimizeDefaultOrderBy( : `${orderByArr[0]} DESC`; } +/** + * Per-source default ORDER BY for multi-source search. Same resolution as + * useDefaultOrderBy minus the table sorting-key optimization (which needs a + * metadata query per source): the source's explicit orderByExpression, else + * its timestamp expression(s) DESC. Time-window pagination requires the first + * term to be the source's timestamp, which this guarantees. + */ +function multiSourceDefaultOrderBy(source: TSource): string { + const isEventSource = + source.kind === SourceKind.Log || source.kind === SourceKind.Trace; + const explicit = isEventSource ? source.orderByExpression?.trim() : undefined; + if (explicit) return explicit; + return optimizeDefaultOrderBy( + source.timestampValueExpression ?? '', + isEventSource ? source.displayedTimestampValueExpression : undefined, + undefined, + ); +} + export function useDefaultOrderBy(sourceID: string | undefined | null) { const { data: source } = useSource({ id: sourceID, @@ -895,6 +936,9 @@ function formatDroppedFiltersMessage(count: number): string { // This is outside as it needs to be a stable reference const queryStateMap = { source: parseAsString, + // JSON-encoded (not comma-separated) because source names may contain + // commas; `source` is always written alongside it as the primary. + sources: parseAsJsonEncoded(), where: parseAsStringEncoded, select: parseAsStringEncoded, whereLanguage: parseAsStringEnum<'sql' | 'lucene'>(['sql', 'lucene']), @@ -1116,6 +1160,7 @@ export function DBSearchPage() { (savedSearchId || directTraceId || rawSearchedConfig.source ? '' : defaultSourceId), + sources: searchedConfig.sources ?? [], filters: searchedConfig.filters ?? [], orderBy: searchedConfig.orderBy ?? '', }, @@ -1184,6 +1229,7 @@ export function DBSearchPage() { whereLanguage: searchedConfig?.whereLanguage ?? getStoredLanguage() ?? 'lucene', source: searchedConfig?.source ?? undefined, + sources: searchedConfig?.sources ?? [], filters: searchedConfig?.filters ?? [], orderBy: searchedConfig?.orderBy ?? '', }); @@ -1198,6 +1244,7 @@ export function DBSearchPage() { // to an existing source. const isSearchConfigEmpty = !rawSearchedConfig.source && + !rawSearchedConfig.sources?.length && !where && !select && !whereLanguage && @@ -1240,6 +1287,7 @@ export function DBSearchPage() { savedSearch, searchedConfig, rawSearchedConfig.source, + rawSearchedConfig.sources, setSearchedConfig, savedSearchId, defaultSourceId, @@ -1268,12 +1316,15 @@ export function DBSearchPage() { const onSubmit = useCallback(() => { onSearch(displayedTimeInputValue); handleSubmit( - ({ select, where, whereLanguage, source, filters, orderBy }) => { + ({ select, where, whereLanguage, source, sources, filters, orderBy }) => { setSearchedConfig({ select, where, whereLanguage, source, + // Writer discipline: only 2+ selections persist the list; a single + // selection clears it so old-style URLs stay canonical. + sources: sources.length > 1 ? sources : null, filters, orderBy, }); @@ -1301,8 +1352,123 @@ export function DBSearchPage() { [debouncedSubmit, setValue], ); + const watchedSource = useWatch({ + control, + name: 'source', + // Watch will reset when changing saved search, so we need to default to the URL + defaultValue: searchedConfig.source ?? undefined, + }); + + // --- Multi-source search: selection & schema state ------------------------ + // 2+ resolved sources in the ?sources= param engage multi mode: one + // independent query pipeline per source, merged client-side. The single + // `source` (primary) keeps every existing code path working; multi mode + // only swaps what gets rendered below. Declared before the filter-state + // hooks so they can work against the union of the selected schemas. + const { sources: searchedMultiSources } = useResolvedSourcesParam( + rawSearchedConfig.sources, + { kinds: ALLOWED_SOURCE_KINDS }, + ); + const isMultiSource = searchedMultiSources.length > 1; + // Delta/pattern analyses are per-source; multi mode pins the results view. + const effectiveAnalysisMode = isMultiSource ? 'results' : analysisMode; + // Raw SQL WHERE names concrete columns of a concrete table — reinterpreting + // it per source risks silently-wrong results, so multi mode requires Lucene. + const isMultiSourceSqlBlocked = + isMultiSource && + (searchedConfig.whereLanguage ?? getStoredLanguage() ?? 'lucene') === + 'sql' && + !!searchedConfig.where; + + // A hand-authored URL may carry only ?sources=; backfill the primary so the + // single-source machinery (form, chart config) has one. + useEffect(() => { + if (!rawSearchedConfig.source && searchedMultiSources.length > 0) { + setSearchedConfig({ source: searchedMultiSources[0].id }); + } + }, [rawSearchedConfig.source, searchedMultiSources, setSearchedConfig]); + + const watchedSources = useWatch({ control, name: 'sources' }); + const formSourceCount = watchedSources?.length ?? 0; + // The multi-select UI stays visible while the user is composing a selection + // (even before a second source is added). + const [multiPickerOpen, setMultiPickerOpen] = useState(false); + const isMultiSelectUI = multiPickerOpen || formSourceCount > 1; + const formIsMulti = formSourceCount > 1; + + const enterMultiSourceSelect = useCallback(() => { + setValue('sources', watchedSource ? [watchedSource] : []); + setMultiPickerOpen(true); + }, [setValue, watchedSource]); + + // Keep the primary `source` field in sync with the selection and re-run the + // search when the selection changes. + const prevWatchedSourcesRef = useRef(null); + useEffect(() => { + const current = watchedSources ?? []; + const prev = prevWatchedSourcesRef.current; + if (prev != null && JSON.stringify(prev) === JSON.stringify(current)) { + return; + } + prevWatchedSourcesRef.current = current; + if (prev == null) { + // Initial hydration from the URL — nothing changed. + return; + } + if (current.length > 0 && current[0] !== watchedSource) { + setValue('source', current[0]); + } + if ((prev?.length ?? 0) <= 1 && current.length > 1) { + // Entering multi mode: the single-source SELECT/ORDER BY strings don't + // translate to the canonical multi-source shape. + setValue('select', ''); + setValue('orderBy', ''); + } + debouncedSubmit(); + }, [watchedSources, watchedSource, setValue, debouncedSubmit]); + + // Collapse the picker back to the single-source select when the user + // reduces a real multi selection to one source. Keyed on the >1 → ≤1 + // transition so it can't fire in the just-opened composing state (picker + // open, one source selected, second not yet picked). + const prevFormSourceCountRef = useRef(formSourceCount); + useEffect(() => { + const prev = prevFormSourceCountRef.current; + prevFormSourceCountRef.current = formSourceCount; + if (prev > 1 && formSourceCount <= 1) { + setMultiPickerOpen(false); + } + }, [formSourceCount]); + + // Per-source top-level columns: powers the add-column picker, the + // per-source `column vs NULL` projection for user-picked extras, and + // per-source filter resolvability. Driven by the form's draft selection + // while composing (so the picker has options before the search is + // submitted), falling back to the searched selection. + const formMultiSources = useMemo( + () => + (watchedSources ?? []) + .map(id => inputSourceObjs?.find(s => s.id === id)) + .filter((s): s is TSource => s != null), + [watchedSources, inputSourceObjs], + ); + const { + columnsBySourceId, + unionColumns, + dateTimeColumns: multiDateTimeColumns, + } = useMultiSourceColumns( + formMultiSources.length > 1 + ? formMultiSources + : isMultiSource + ? searchedMultiSources + : [], + ); + // --- End multi-source selection & schema state ----------------------------- + // Top-level column names for the active source, used to quote - // filter keys that contain special characters. + // filter keys that contain special characters. In multi mode this is the + // union across the selected sources, so filter keys from any of them + // escape correctly. const { data: inputSourceColumns } = useColumns( { databaseName: inputSourceObj?.from?.databaseName ?? '', @@ -1311,20 +1477,19 @@ export function DBSearchPage() { }, { enabled: !!inputSourceObj }, ); - const knownColumns = useMemo( - () => - inputSourceColumns - ? new Set(inputSourceColumns.map(c => c.name)) - : new Set(), - [inputSourceColumns], - ); + const knownColumns = useMemo(() => { + if (isMultiSource) { + const union = new Set(); + for (const names of columnsBySourceId.values()) { + for (const name of names) union.add(name); + } + return union; + } + return inputSourceColumns + ? new Set(inputSourceColumns.map(c => c.name)) + : new Set(); + }, [inputSourceColumns, isMultiSource, columnsBySourceId]); - const watchedSource = useWatch({ - control, - name: 'source', - // Watch will reset when changing saved search, so we need to default to the URL - defaultValue: searchedConfig.source ?? undefined, - }); const prevSourceRef = useRef(watchedSource); // Set when the user switches sources via the dropdown. The follow-up // effect waits for the new source's columns to load and then drops any @@ -1347,11 +1512,21 @@ export function DBSearchPage() { const { dateTimeColumns, onResolvedColumnsChange } = useResolvedDateTimeColumns(inputSourceColumns); + // In multi mode, date/time-typed filter keys may come from any selected + // source's schema. + const effectiveDateTimeColumns = useMemo( + () => + isMultiSource && multiDateTimeColumns.size > 0 + ? new Map([...dateTimeColumns, ...multiDateTimeColumns]) + : dateTimeColumns, + [isMultiSource, dateTimeColumns, multiDateTimeColumns], + ); + const filters = useWatch({ name: 'filters', control }); const searchFilters = useSearchPageFilterState({ searchQuery: filters ?? undefined, onFilterChange: handleSetFilters, - dateTimeColumns, + dateTimeColumns: effectiveDateTimeColumns, knownColumns, }); @@ -1489,6 +1664,174 @@ export function DBSearchPage() { const { data: chartConfig, isLoading: isChartConfigLoading } = useSearchedConfigToChartConfig(chartSearchConfig, defaultSearchConfig); + // --- Multi-source search: query specs ------------------------------------- + // In multi mode the `select` param holds the extra column names picked by + // the user (the canonical columns are always projected). + const multiExtraColumnNames = useMemo( + () => + isMultiSource ? splitAndTrimWithBracket(searchedConfig.select ?? '') : [], + [isMultiSource, searchedConfig.select], + ); + + // The add-column picker edits the form's draft select (like the SELECT + // editor it replaces), then auto-submits. + const inputSelect = useWatch({ name: 'select', control }); + const multiPickerValue = useMemo( + () => (formIsMulti ? splitAndTrimWithBracket(inputSelect ?? '') : []), + [formIsMulti, inputSelect], + ); + const onMultiColumnsChange = useCallback( + (columns: string[]) => { + setValue('select', columns.join(', ')); + debouncedSubmit(); + }, + [setValue, debouncedSubmit], + ); + + // Sidebar filters apply per source. A source whose table lacks a filtered + // column can't answer the filtered search — it's excluded entirely (with a + // visible reason on its status chip) rather than silently returning rows + // that ignore the filter. + const multiSourceFilters = useMemo( + () => (isMultiSource ? (searchedConfig.filters ?? []) : []), + [isMultiSource, searchedConfig.filters], + ); + const multiDisabledReasons = useMemo(() => { + const reasons = new Map(); + if (!isMultiSource || multiSourceFilters.length === 0) return reasons; + for (const source of searchedMultiSources) { + const missing = unresolvedFilterColumns( + multiSourceFilters, + columnsBySourceId.get(source.id), + ); + if (missing.length > 0) { + reasons.set( + source.id, + `${source.name} is excluded: the active filter uses ${missing.join( + ', ', + )}, which it doesn't have`, + ); + } + } + return reasons; + }, [ + isMultiSource, + multiSourceFilters, + searchedMultiSources, + columnsBySourceId, + ]); + + // The single-source chart config, pinned to the searched time range. + const dbSqlRowTableConfig = useMemo(() => { + if (chartConfig == null) { + return undefined; + } + + return { + ...chartConfig, + dateRange: searchedTimeRange, + }; + }, [chartConfig, searchedTimeRange]); + + // The search's query plan: one spec per selected source. A single source is + // just N=1 — its spec carries the user's own SELECT/ORDER BY, so the results + // table renders exactly what the user asked for. The canonical aliases only + // come into play when there is more than one source to reconcile. + const searchStreamSpecs = useMemo(() => { + if (!isMultiSource) { + if (dbSqlRowTableConfig == null || searchedSource == null) return []; + return [{ source: searchedSource, config: dbSqlRowTableConfig }]; + } + if (isMultiSourceSqlBlocked) return []; + // Extra columns and filters both need each source's DESCRIBE (to resolve + // column-vs-NULL and filter resolvability); hold the row queries until + // they've loaded so we don't fire throwaway or erroring queries. + if ( + (multiExtraColumnNames.length > 0 || multiSourceFilters.length > 0) && + columnsBySourceId.size < searchedMultiSources.length + ) { + return []; + } + const includeDuration = searchedMultiSources.some(isTraceSource); + const where = searchedConfig.where ?? ''; + return searchedMultiSources.map(source => ({ + source, + disabledReason: multiDisabledReasons.get(source.id), + config: { + ...buildMultiSourceSearchConfig( + source, + { + where, + whereLanguage: 'lucene', + filters: multiSourceFilters, + orderBy: multiSourceDefaultOrderBy(source), + }, + { + includeDuration, + extraColumns: resolveExtraColumnsForSource( + multiExtraColumnNames, + columnsBySourceId.get(source.id), + ), + }, + ), + dateRange: searchedTimeRange, + }, + })); + }, [ + isMultiSource, + isMultiSourceSqlBlocked, + searchedMultiSources, + searchedSource, + dbSqlRowTableConfig, + searchedConfig.where, + multiExtraColumnNames, + multiSourceFilters, + multiDisabledReasons, + columnsBySourceId, + searchedTimeRange, + ]); + + const multiHistogramSpecs = useMemo(() => { + if (!isMultiSource || isMultiSourceSqlBlocked) return []; + if ( + multiSourceFilters.length > 0 && + columnsBySourceId.size < searchedMultiSources.length + ) { + return []; + } + const where = searchedConfig.where ?? ''; + return searchedMultiSources.map(source => ({ + source, + disabledReason: multiDisabledReasons.get(source.id), + config: { + ...buildMultiSourceSearchConfig(source, { + where, + whereLanguage: 'lucene', + filters: multiSourceFilters, + }), + select: ALERT_COUNT_DEFAULT_SELECT, + orderBy: undefined, + granularity: 'auto' as const, + dateRange: searchedTimeRange, + displayType: DisplayType.StackedBar, + // Match the single-source histogram: reflect the user's exact range + // so chart and table counts agree (see histogramTimeChartConfig). + alignDateRangeToGranularity: false, + dateRangeEndInclusive: true, + }, + })); + }, [ + isMultiSource, + isMultiSourceSqlBlocked, + searchedMultiSources, + searchedConfig.where, + multiSourceFilters, + multiDisabledReasons, + columnsBySourceId, + searchedTimeRange, + ]); + // --- End multi-source search --------------------------------------------- + // query error handling const { hasQueryError, queryError } = useMemo(() => { const hasQueryError = Object.values(_queryErrors).length > 0; @@ -1650,24 +1993,16 @@ export function DBSearchPage() { setTimeout(() => setCollapseAllRows(false), 100); }, [interval, updateRelativeTimeInputValue, setIsLive]); - const dbSqlRowTableConfig = useMemo(() => { - if (chartConfig == null) { - return undefined; - } - - return { - ...chartConfig, - dateRange: searchedTimeRange, - }; - }, [chartConfig, searchedTimeRange]); - // Stable key for persisting column widths in localStorage. Scoped per saved - // search when one is loaded, else per source for ad-hoc searches. + // search when one is loaded, else per source (or source set) for ad-hoc + // searches. const columnSizeTableId = savedSearchId ? `db-search-saved-${savedSearchId}` - : searchedConfig.source - ? `db-search-source-${searchedConfig.source}` - : undefined; + : isMultiSource + ? `db-search-multi-${searchedMultiSources.map(s => s.id).join('-')}` + : searchedConfig.source + ? `db-search-source-${searchedConfig.source}` + : undefined; const displayedColumns = useMemo(() => { // `select` is typed as `string | DerivedColumn[]` upstream, but in the @@ -1969,6 +2304,28 @@ export function DBSearchPage() { ], ); + // Multi-source rows span schemas, so the single-source column toggles are + // omitted — the side panel hides them. Property-add-to-filter IS wired: + // filters resolve per source, and a source that lacks the column is + // excluded with a visible reason. Passing no `source` is required: with a + // null context source, deriveRowSidePanelContextForSource treats every row + // as same-source, which is exactly the cross-source semantics filters now + // have. + const multiRowTableContext = useMemo( + () => ({ + onPropertyAddClick: searchFilters.setFilterValue, + generateSearchUrl, + isChildModalOpen: isDrawerChildModalOpen, + setChildModalOpen: setDrawerChildModalOpen, + }), + [ + searchFilters.setFilterValue, + generateSearchUrl, + isDrawerChildModalOpen, + setDrawerChildModalOpen, + ], + ); + const inputSourceTableConnection = useMemo( () => tcFromSource(inputSourceObj), [inputSourceObj], @@ -2220,22 +2577,49 @@ export function DBSearchPage() { > {/* */} - setIsSourceSchemaPreviewOpen(true)} - isSchemaPreviewEnabled={isSourceSchemaPreviewEnabled( - inputSourceObj, - )} - allowedSourceKinds={ALLOWED_SOURCE_KINDS} - data-testid="source-selector" - style={{ minWidth: 150 }} - /> + {isMultiSelectUI ? ( + + ) : ( + <> + setIsSourceSchemaPreviewOpen(true)} + isSchemaPreviewEnabled={isSourceSchemaPreviewEnabled( + inputSourceObj, + )} + allowedSourceKinds={ALLOWED_SOURCE_KINDS} + data-testid="source-selector" + style={{ minWidth: 150 }} + /> + + + + + + + )} setIsSourceSchemaPreviewOpen(false)} /> - - - - + {formIsMulti ? ( + + ) : ( + + )} + {!formIsMulti && ( + + + + )} <> {!savedSearchId ? ( - + + ) : ( + + )} @@ -2346,7 +2755,7 @@ export function DBSearchPage() { setInputValue={setDisplayedTimeInputValue} onSearch={onTimePickerSearch} onRelativeSearch={onTimePickerRelativeSearch} - showLive={analysisMode === 'results'} + showLive={effectiveAnalysisMode === 'results'} isLiveMode={isLive} // Default to relative time mode if the user has made changes to interval and reloaded. defaultRelativeTimeMode={ @@ -2381,7 +2790,7 @@ export function DBSearchPage() { @@ -2425,7 +2834,21 @@ export function DBSearchPage() { height: '100%', }} > - {!isFilterSidebarCollapsed && ( + {!isFilterSidebarCollapsed && + isMultiSource && + !isMultiSourceSqlBlocked && ( + + setIsFilterSidebarCollapsed(true)} + /> + + )} + {!isFilterSidebarCollapsed && !isMultiSource && ( )} - {analysisMode === 'pattern' && + {effectiveAnalysisMode === 'pattern' && histogramTimeChartConfig != null && ( @@ -2523,7 +2946,7 @@ export function DBSearchPage() { )} - {analysisMode === 'delta' && + {effectiveAnalysisMode === 'delta' && searchedSource != null && isTraceSource(searchedSource) && ( )} - {analysisMode === 'results' && ( + {effectiveAnalysisMode === 'results' && isMultiSource && ( + + {isMultiSourceSqlBlocked ? ( + + + SQL search isn't supported across multiple sources + + + A SQL WHERE clause references the columns of one + specific table. Switch the search language to Lucene to + search across sources, or go back to a single source. + + + ) : ( + <> + + + + {isFilterSidebarCollapsed && ( + + setIsFilterSidebarCollapsed(false) + } + /> + )} + + + {shouldShowLiveModeHint && ( + + )} + + + + + + + + + + )} + + )} + {effectiveAnalysisMode === 'results' && !isMultiSource && ( {chartConfig && histogramTimeChartConfig && ( <> @@ -2725,32 +3230,26 @@ export function DBSearchPage() { px="sm" data-testid="search-results-panel" > - {chartConfig && - searchedConfig.source && - dbSqlRowTableConfig && ( - - )} + )} diff --git a/packages/app/src/__tests__/DBSearchPage.directTrace.test.tsx b/packages/app/src/__tests__/DBSearchPage.directTrace.test.tsx index d3f10d02a5..1f0b9914fa 100644 --- a/packages/app/src/__tests__/DBSearchPage.directTrace.test.tsx +++ b/packages/app/src/__tests__/DBSearchPage.directTrace.test.tsx @@ -211,6 +211,13 @@ jest.mock('../components/ChartSQLPreview', () => ({ SQLPreview: () =>
, })); jest.mock('../components/DBSqlRowTableWithSidebar', () => () =>
); +// Multi-source components pull in DBRowSidePanel (and its deep import graph), +// which this test isolates away just like DBSqlRowTableWithSidebar above. +jest.mock('../components/SearchResultsTable', () => () =>
); +jest.mock('../components/MultiSourceTimeChart', () => ({ + MultiSourceTimeChart: () =>
, + MultiSourceTotalCountChart: () =>
, +})); jest.mock('../components/PatternTable', () => () =>
); jest.mock('../components/Search/DBSearchHeatmapChart', () => ({ DBSearchHeatmapChart: () =>
, diff --git a/packages/app/src/components/DBRowTable.tsx b/packages/app/src/components/DBRowTable.tsx index 19574d849b..59bdf5398d 100644 --- a/packages/app/src/components/DBRowTable.tsx +++ b/packages/app/src/components/DBRowTable.tsx @@ -103,6 +103,7 @@ import { useLocalStorage, usePrevious, } from '@/utils'; +import { MULTI_SOURCE_ROW_FIELDS } from '@/utils/multiSourceMerge'; import ChartErrorState, { ChartErrorStateVariant, @@ -124,6 +125,7 @@ import { useExpandableRows, } from './ExpandableRowTable'; import LogLevel from './LogLevel'; +import { SourceBadge } from './MultiSourceBadge'; import styles from '@styles/LogTable.module.scss'; @@ -167,6 +169,7 @@ function getResolvedColumnSize( const jsType = opts.columnTypeMap.get(column)?._type; if (jsType === JSDataType.Date) return 170; if (column === opts.logLevelColumn) return 115; + if (column === MULTI_SOURCE_ROW_FIELDS.SOURCE_NAME) return 140; return 160; } @@ -615,6 +618,30 @@ export const RawLogTable = memo( const strValue = typeof value === 'string' ? value : `${value}`; + // Multi-source search tags each merged row with its origin + // source; render it as a colored badge (color assigned by the + // merge layer, consistent with the histogram series). + if (column === MULTI_SOURCE_ROW_FIELDS.SOURCE_NAME) { + return ( + + ); + } + + // Multi-source rows project NULL where a source lacks the + // field (e.g. Duration or a picked column for log rows); show a + // quiet dash instead of the literal "null". + if ( + value == null && + info.row.original[MULTI_SOURCE_ROW_FIELDS.SOURCE_ID] != null + ) { + return —; + } + if (column === logLevelColumn) { return ; } @@ -1392,7 +1419,7 @@ export function appendSelectWithAdditionalKeys( } } -function getSelectLength(select: SelectList): number { +export function getSelectLength(select: SelectList): number { if (typeof select === 'string') { return select.split(',').filter(s => s.trim().length > 0).length; } else { @@ -1476,7 +1503,16 @@ export function useConfigWithAdditionalSelect( }, [primaryKey, partitionKey, config, tableMetadata, columns, sourceId]); } -function selectColumnMapWithoutAdditionalKeys( +/** + * The user's SELECT columns, keyed by result-set name, with the row-identity + * columns the query appends (primary/partition/block keys) trimmed off. + * + * Positional rather than name-based because ClickHouse may rewrite column + * names, and the SELECT length can differ from the returned column count + * (e.g. `SELECT *`). Exported for SearchResultsTable, which resolves the same + * columns when rendering a single source's own SELECT. + */ +export function selectColumnMapWithoutAdditionalKeys( selectMeta: ColumnMetaType[] | undefined, additionalKeysLength: number | undefined, ): Map< @@ -1503,6 +1539,144 @@ function selectColumnMapWithoutAdditionalKeys( export type DBRowTableVariant = 'default' | 'muted'; +/** + * Drop rows matching "noisy" event patterns (patterns covering more than + * DENOISE_NOISE_THRESHOLD of a sample) from an already-fetched row set. + * + * Extracted so the search results table and DBSqlRowTable share one + * implementation. Denoising is inherently single-source: it mines patterns + * from one table's body column against that source's severity expression. + */ +export function useDenoisedRows({ + config, + sourceId, + processedRows, + patternColumn, + denoiseResults, + isLive, +}: { + config: BuilderChartConfigWithDateRange; + sourceId?: string; + processedRows: Record[]; + /** Result column the patterns are mined from (the last SELECT column). */ + patternColumn: string | undefined; + denoiseResults: boolean; + isLive?: boolean; +}) { + const { data: source } = useSource({ id: sourceId }); + const groupedPatterns = useGroupedPatterns({ + config, + samples: DENOISE_SAMPLE_SIZE, + bodyValueExpression: patternColumn ?? '', + severityTextExpression: + (source?.kind === SourceKind.Log + ? source.severityTextExpression + : undefined) ?? '', + totalCount: undefined, + enabled: denoiseResults, + }); + const noisyPatterns = useQuery({ + queryKey: ['noisy-patterns', config], + queryFn: async () => { + return Object.values(groupedPatterns.data).filter( + p => + p.count / (groupedPatterns.sampledRowCount ?? 1) > + DENOISE_NOISE_THRESHOLD, + ); + }, + enabled: + denoiseResults && + groupedPatterns.data != null && + Object.values(groupedPatterns.data).length > 0 && + groupedPatterns.miner != null, + }); + const noisyPatternIds = useMemo(() => { + return noisyPatterns.data?.map(p => p.id) ?? []; + }, [noisyPatterns.data]); + + const denoisedRows = useQuery({ + queryKey: [ + 'denoised-rows', + config, + denoiseResults, + // Only include processed rows if denoising is enabled + // This helps prevent the queryKey from getting extremely large + // and causing memory issues, when it's not used. + ...(denoiseResults ? [processedRows] : []), + noisyPatternIds, + patternColumn, + ], + queryFn: async () => { + if (!denoiseResults) { + return []; + } + // No noisy patterns, so no need to denoise + if (noisyPatternIds.length === 0) { + return processedRows; + } + + const matchedLogs = await groupedPatterns.miner?.matchLogs( + processedRows.map(row => row[patternColumn ?? '']), + ); + return processedRows.filter((row, i) => { + const match = matchedLogs?.[i]; + return !noisyPatternIds.includes(`${match}`); + }); + }, + placeholderData: (previousData, previousQuery) => { + // If it's the same search, but new data, return the previous data while we load + if ( + previousQuery?.queryKey?.[0] === 'denoised-rows' && + previousQuery?.queryKey?.[1] === config + ) { + return previousData; + } + return undefined; + }, + gcTime: isLive ? ms('30s') : ms('5m'), // more aggressive gc for live data, since it can end up holding lots of data + enabled: + denoiseResults && + noisyPatterns.isSuccess && + processedRows.length > 0 && + groupedPatterns.miner != null, + }); + + return { + rows: denoiseResults ? (denoisedRows.data ?? []) : processedRows, + noisyPatterns: noisyPatterns.data, + hasNoisyPatterns: noisyPatternIds.length > 0, + isFetching: + denoisedRows.isFetching || + noisyPatterns.isFetching || + groupedPatterns.isLoading, + }; +} + +/** The "Removed Noisy Event Patterns" summary shown above denoised results. */ +export function DenoisedPatternsSummary({ + noisyPatterns, + hasNoisyPatterns, +}: { + noisyPatterns: { id: string; pattern: string }[] | undefined; + hasNoisyPatterns: boolean; +}) { + return ( + + + Removed Noisy Event Patterns + + + {noisyPatterns?.map(p => ( + + {p.pattern} + + ))} + {!hasNoisyPatterns && No noisy patterns found} + + + ); +} + function DBSqlRowTableComponent({ config, sourceId, @@ -1736,88 +1910,17 @@ function DBSqlRowTableComponent({ const { data: source } = useSource({ id: sourceId }); const patternColumn = columns[columns.length - 1]; - const groupedPatterns = useGroupedPatterns({ + const denoise = useDenoisedRows({ config, - samples: DENOISE_SAMPLE_SIZE, - bodyValueExpression: patternColumn ?? '', - severityTextExpression: - (source?.kind === SourceKind.Log - ? source.severityTextExpression - : undefined) ?? '', - totalCount: undefined, - enabled: denoiseResults, - }); - const noisyPatterns = useQuery({ - queryKey: ['noisy-patterns', config], - queryFn: async () => { - return Object.values(groupedPatterns.data).filter( - p => - p.count / (groupedPatterns.sampledRowCount ?? 1) > - DENOISE_NOISE_THRESHOLD, - ); - }, - enabled: - denoiseResults && - groupedPatterns.data != null && - Object.values(groupedPatterns.data).length > 0 && - groupedPatterns.miner != null, - }); - const noisyPatternIds = useMemo(() => { - return noisyPatterns.data?.map(p => p.id) ?? []; - }, [noisyPatterns.data]); - - const denoisedRows = useQuery({ - queryKey: [ - 'denoised-rows', - config, - denoiseResults, - // Only include processed rows if denoising is enabled - // This helps prevent the queryKey from getting extremely large - // and causing memory issues, when it's not used. - ...(denoiseResults ? [processedRows] : []), - noisyPatternIds, - patternColumn, - ], - queryFn: async () => { - if (!denoiseResults) { - return []; - } - // No noisy patterns, so no need to denoise - if (noisyPatternIds.length === 0) { - return processedRows; - } - - const matchedLogs = await groupedPatterns.miner?.matchLogs( - processedRows.map(row => row[patternColumn]), - ); - return processedRows.filter((row, i) => { - const match = matchedLogs?.[i]; - return !noisyPatternIds.includes(`${match}`); - }); - }, - placeholderData: (previousData, previousQuery) => { - // If it's the same search, but new data, return the previous data while we load - if ( - previousQuery?.queryKey?.[0] === 'denoised-rows' && - previousQuery?.queryKey?.[1] === config - ) { - return previousData; - } - return undefined; - }, - gcTime: isLive ? ms('30s') : ms('5m'), // more aggressive gc for live data, since it can end up holding lots of data - enabled: - denoiseResults && - noisyPatterns.isSuccess && - processedRows.length > 0 && - groupedPatterns.miner != null, + sourceId, + processedRows, + patternColumn, + denoiseResults, + isLive, }); const isLoading = denoiseResults - ? isFetching || - denoisedRows.isFetching || - noisyPatterns.isFetching || - groupedPatterns.isLoading + ? isFetching || denoise.isFetching : isFetching; const loadingDate = @@ -1828,28 +1931,17 @@ function DBSqlRowTableComponent({ return ( <> {denoiseResults && ( - - - Removed Noisy Event Patterns - - - {noisyPatterns.data?.map(p => ( - - {p.pattern} - - ))} - {noisyPatternIds.length === 0 && ( - No noisy patterns found - )} - - + )} + + {name} + + ); +} diff --git a/packages/app/src/components/MultiSourceColumnPicker.tsx b/packages/app/src/components/MultiSourceColumnPicker.tsx new file mode 100644 index 0000000000..55230dd479 --- /dev/null +++ b/packages/app/src/components/MultiSourceColumnPicker.tsx @@ -0,0 +1,71 @@ +import { useCallback, useMemo } from 'react'; +import { Group, MultiSelect, Text } from '@mantine/core'; + +import { MultiSourceColumnOption } from '@/hooks/useMultiSourceSearch'; + +/** + * Multi-source replacement for the free-text SELECT editor: pick extra + * columns from the union of the selected sources' top-level columns. Columns + * missing from a source render as blank cells for that source's rows. + */ +export default function MultiSourceColumnPicker({ + unionColumns, + totalSources, + value, + onChange, +}: { + unionColumns: MultiSourceColumnOption[]; + totalSources: number; + /** Currently selected extra column names. */ + value: string[]; + onChange: (columns: string[]) => void; +}) { + const availabilityByName = useMemo( + () => new Map(unionColumns.map(c => [c.name, c.availableCount])), + [unionColumns], + ); + + const data = useMemo( + () => + unionColumns.map(c => ({ + value: c.name, + label: c.name, + })), + [unionColumns], + ); + + const renderOption = useCallback( + ({ option }: { option: { value: string; label: string } }) => { + const available = availabilityByName.get(option.value) ?? 0; + return ( + + + {option.label} + + {available < totalSources && ( + + {available}/{totalSources} sources + + )} + + ); + }, + [availabilityByName, totalSources], + ); + + return ( + + ); +} diff --git a/packages/app/src/components/MultiSourceSearchFilters.tsx b/packages/app/src/components/MultiSourceSearchFilters.tsx new file mode 100644 index 0000000000..a909e28441 --- /dev/null +++ b/packages/app/src/components/MultiSourceSearchFilters.tsx @@ -0,0 +1,315 @@ +import { useMemo } from 'react'; +import { FilterState } from '@hyperdx/common-utils/dist/filters'; +import { + BuilderChartConfigWithDateRange, + TSource, +} from '@hyperdx/common-utils/dist/types'; +import { + ActionIcon, + Box, + Flex, + Group, + ScrollArea, + Stack, + Text, + Tooltip, +} from '@mantine/core'; +import { IconArrowBarToLeft, IconFilterOff } from '@tabler/icons-react'; + +import { + cleanedFacetName, + FilterGroup, +} from '@/components/DBSearchPageFilters'; +import { useFetchFacets } from '@/components/DBSearchPageFilters/hooks'; +import { NestedFilterGroup } from '@/components/DBSearchPageFilters/NestedFilterGroup'; +import { + getFilterStateEntry, + groupFacetsByBaseName, + toQuotedClickHouseKeyExpression, +} from '@/components/DBSearchPageFilters/utils'; +import { useMultiSourceSlots } from '@/hooks/useMultiSourceSearch'; +import useResizable from '@/hooks/useResizable'; +import { FilterStateHook } from '@/searchFilters'; + +import resizeStyles from '@styles/ResizablePanel.module.scss'; +import classes from '@styles/SearchPage.module.scss'; + +export type MultiSourceFilterSpec = { + source: TSource; + /** Per-source config carrying connection/from/where/filters/dateRange. */ + config: BuilderChartConfigWithDateRange; +}; + +// Placeholder for unused hook slots; never fetched (enabled: false). +const STUB_CONFIG: BuilderChartConfigWithDateRange = { + connection: '', + from: { databaseName: '', tableName: '' }, + timestampValueExpression: '', + select: '', + where: '', + whereLanguage: 'sql', + dateRange: [new Date(0), new Date(0)], +}; + +type FacetSlotState = { + facets: { key: string; value: string[] }[] | undefined; + isLoading: boolean; + isFetching: boolean; +}; + +/** Slot hook: the full single-source facet pipeline for one selected source. */ +function useSourceFacetsSlot( + spec: MultiSourceFilterSpec | undefined, + { + dateRange, + filterState, + }: { + dateRange: [Date, Date]; + filterState: FilterStateHook['filters']; + }, +): FacetSlotState { + const { data, isLoading, isFetching } = useFetchFacets({ + chartConfig: spec?.config ?? STUB_CONFIG, + sourceId: spec?.source.id ?? null, + dateRange, + // Value lists show everything in range (not narrowed by the current + // query), matching the sidebar's default "show all values" behavior. + mode: 'all', + filterState, + enabled: spec != null, + }); + + return useMemo( + () => ({ facets: data.keyValues, isLoading, isFetching }), + [data.keyValues, isLoading, isFetching], + ); +} + +const NOOP = () => { + /* pins and load-more are single-source affordances; no-op in multi mode */ +}; +const VALUE_PINS = { onPinClick: NOOP, isPinned: () => false }; + +/** + * Multi-source variant of the search filters sidebar: one facet pipeline per + * selected source, merged by field path with values unioned. Filters apply + * per source; a source that lacks a filtered column is excluded from the + * search (surfaced as a chip on the results table). + * + * Single-source-only affordances (pins, shared filters, value counts, + * load-more, analysis-mode tabs, denoising) are intentionally absent. + */ +export default function MultiSourceSearchFilters({ + specs, + dateRange, + isLive, + knownColumns, + searchFilters, + onCollapse, +}: { + specs: MultiSourceFilterSpec[]; + dateRange: [Date, Date]; + isLive: boolean; + /** Union of the selected sources' top-level column names (for escaping). */ + knownColumns: Set; + searchFilters: FilterStateHook; + onCollapse?: () => void; +}) { + const { size, startResize } = useResizable(16, 'left'); + const { + filters: filterState, + setFilterValue, + clearFilter, + clearAllFilters, + setFilterRange, + } = searchFilters; + + const slots = useMultiSourceSlots(specs, useSourceFacetsSlot, { + dateRange, + filterState, + }); + + const isFetching = slots.some(s => s.isFetching); + const isLoading = slots.some(s => s.isLoading); + + // Merge facets across sources: union values per field path, in first-seen + // order (the first selected source's ordering wins). + const mergedFacets = useMemo(() => { + const byKey = new Map }>(); + for (const slot of slots) { + for (const facet of slot.facets ?? []) { + let entry = byKey.get(facet.key); + if (entry == null) { + entry = { values: [], seen: new Set() }; + byKey.set(facet.key, entry); + } + for (const value of facet.value) { + if (!entry.seen.has(value)) { + entry.seen.add(value); + entry.values.push(value); + } + } + } + } + return [...byKey.entries()].map(([key, entry]) => ({ + key, + value: entry.values, + })); + }, [slots]); + + const hasSelections = Object.keys(filterState).length > 0; + const firstConfig = specs[0]?.config ?? STUB_CONFIG; + const { grouped, nonGrouped } = useMemo( + () => groupFacetsByBaseName(mergedFacets), + [mergedFacets], + ); + + return ( + +
+ + + + + Filters {isFetching && '···'} + + + {hasSelections && ( + + + + + + )} + {onCollapse && ( + + + + + + )} + + + + Values across all selected sources. A filter on a field a source + doesn't have excludes that source from the results. + + {grouped.map(group => ( + ({ + ...child, + sqlKey: toQuotedClickHouseKeyExpression( + child.key, + knownColumns, + ), + }))} + selectedValues={group.children.reduce((acc, child) => { + acc[child.key] = getFilterStateEntry( + filterState, + child.key, + ) ?? { + included: new Set(), + excluded: new Set(), + }; + return acc; + }, {} as FilterState)} + onChange={(key, value) => setFilterValue(key, value)} + onClearClick={key => clearFilter(key)} + onOnlyClick={(key, value) => setFilterValue(key, value, 'only')} + onExcludeClick={(key, value) => + setFilterValue(key, value, 'exclude') + } + onPinClick={NOOP} + isPinned={() => false} + showFilterCounts={false} + onLoadMore={NOOP} + loadMoreLoading={{}} + hasLoadedMore={{}} + isDefaultExpanded={group.children.some(child => { + const entry = getFilterStateEntry(filterState, child.key); + return ( + entry != null && + (entry.included.size > 0 || entry.excluded.size > 0) + ); + })} + chartConfig={firstConfig} + isLive={isLive} + /> + ))} + {nonGrouped.map(facet => { + const facetSqlKey = toQuotedClickHouseKeyExpression( + facet.key, + knownColumns, + ); + const entry = getFilterStateEntry(filterState, facet.key); + return ( + ({ + value, + label: value.toString(), + }))} + optionsLoading={isLoading} + selectedValues={ + entry ?? { included: new Set(), excluded: new Set() } + } + onChange={value => setFilterValue(facet.key, value)} + onClearClick={() => clearFilter(facet.key)} + onOnlyClick={value => setFilterValue(facet.key, value, 'only')} + onExcludeClick={value => + setFilterValue(facet.key, value, 'exclude') + } + valuePins={VALUE_PINS} + onLoadMore={NOOP} + loadMoreLoading={false} + hasLoadedMore={false} + isDefaultExpanded={ + entry != null && + (entry.included.size > 0 || + entry.excluded.size > 0 || + entry.range != null) + } + chartConfig={firstConfig} + isLive={isLive} + onRangeChange={range => setFilterRange(facet.key, range)} + /> + ); + })} + {!isLoading && mergedFacets.length === 0 && ( + + No filterable fields found. + + )} + + + + ); +} diff --git a/packages/app/src/components/MultiSourceTimeChart.tsx b/packages/app/src/components/MultiSourceTimeChart.tsx new file mode 100644 index 0000000000..122c937c49 --- /dev/null +++ b/packages/app/src/components/MultiSourceTimeChart.tsx @@ -0,0 +1,325 @@ +import { useMemo, useState } from 'react'; +import { + ColumnMetaType, + filterColumnMetaByType, + JSDataType, + ResponseJSON, +} from '@hyperdx/common-utils/dist/clickhouse'; +import { + BuilderChartConfigWithDateRange, + DisplayType, + TSource, +} from '@hyperdx/common-utils/dist/types'; +import { Text } from '@mantine/core'; +import { keepPreviousData } from '@tanstack/react-query'; + +import api from '@/api'; +import { + convertToTimeChartConfig, + formatResponseForTimeChart, + useTimeChartSettings, +} from '@/ChartUtils'; +import ChartContainer from '@/components/charts/ChartContainer'; +import ChartErrorState from '@/components/charts/ChartErrorState'; +import { type ActiveClickPayload, MemoChart } from '@/HDXMultiSeriesTimeChart'; +import { useQueriedChartConfig } from '@/hooks/useChartConfig'; +import { useMultiSourceSlots } from '@/hooks/useMultiSourceSearch'; +import type { NumberFormat } from '@/types'; + +import { getMultiSourceColor } from './MultiSourceBadge'; + +/** Synthetic group column tagged onto each source's histogram rows. */ +const SOURCE_GROUP_COLUMN = '__hdx_source'; + +export type MultiSourceChartSpec = { + source: TSource; + /** Per-source count() histogram config (canonical WHERE, no groupBy). */ + config: BuilderChartConfigWithDateRange; + /** When set, the source doesn't run (mirrors MultiSourceStreamSpec). */ + disabledReason?: string; +}; + +// Placeholder for unused hook slots; never queried (enabled: false). +const STUB_CONFIG: BuilderChartConfigWithDateRange = { + connection: '', + from: { databaseName: '', tableName: '' }, + timestampValueExpression: '', + select: '', + where: '', + whereLanguage: 'sql', + dateRange: [new Date(0), new Date(0)], +}; + +type HistogramSlotState = { + data: ReturnType['data']; + isLoading: boolean; + isError: boolean; + error: Error | null; +}; + +function useHistogramSlot( + spec: MultiSourceChartSpec | undefined, + { + enabled, + queryKeyPrefix, + enableParallelQueries, + parallelizeWhenPossible, + }: { + enabled: boolean; + queryKeyPrefix: string; + enableParallelQueries?: boolean; + parallelizeWhenPossible?: boolean; + }, +): HistogramSlotState { + const queriedConfig = useMemo( + () => convertToTimeChartConfig(spec?.config ?? STUB_CONFIG), + [spec?.config], + ); + + const { data, isLoading, isError, error } = useQueriedChartConfig( + queriedConfig, + { + // Key shape mirrors DBTimeChart/SearchTotalCountChart so TanStack can + // de-dupe the histogram and total-count consumers of the same source. + queryKey: [ + queryKeyPrefix, + queriedConfig, + 'chunked', + { + disableQueryChunking: false, + enableParallelQueries, + parallelizeWhenPossible, + }, + ], + placeholderData: keepPreviousData, + enableQueryChunking: true, + enableParallelQueries: enableParallelQueries && parallelizeWhenPossible, + enabled: enabled && spec != null && spec.disabledReason == null, + }, + ); + + // Stable identity per content change so the slots array (and everything + // memoized on it) doesn't churn on unrelated renders. + return useMemo( + () => ({ data, isLoading, isError, error: error ?? null }), + [data, isLoading, isError, error], + ); +} + +/** + * Runs one count() histogram query per selected source (one hook slot per + * source, see useMultiSourceSlots) and merges the responses into a single + * response shape with a synthetic source-name group column — so the standard + * time-chart transform naturally yields one series per source. + */ +function useMultiSourceHistogram( + specs: MultiSourceChartSpec[], + { + enabled = true, + queryKeyPrefix, + enableParallelQueries, + }: { + enabled?: boolean; + queryKeyPrefix: string; + enableParallelQueries?: boolean; + }, +) { + const { data: me, isLoading: isLoadingMe } = api.useMe(); + const slots = useMultiSourceSlots(specs, useHistogramSlot, { + enabled: enabled && !isLoadingMe, + queryKeyPrefix, + enableParallelQueries, + parallelizeWhenPossible: me?.team?.parallelizeWhenPossible, + }); + + const isLoading = slots.some(s => s.isLoading); + const allFailed = slots.length > 0 && slots.every(s => s.isError); + const anyError = slots.some(s => s.isError); + const error = slots.find(s => s.error != null)?.error ?? undefined; + const isComplete = + slots.length > 0 && slots.every(s => s.isError || !!s.data?.isComplete); + + const mergedResponse: ResponseJSON> | undefined = + useMemo(() => { + let meta: ColumnMetaType[] | undefined; + const data: Record[] = []; + for (let i = 0; i < specs.length; i++) { + const response = slots[i]?.data; + if (response?.meta == null || response.meta.length === 0) continue; + if (meta == null) { + meta = [ + ...response.meta, + { name: SOURCE_GROUP_COLUMN, type: 'String' }, + ]; + } + const sourceName = specs[i].source.name; + for (const row of response.data ?? []) { + data.push({ ...row, [SOURCE_GROUP_COLUMN]: sourceName }); + } + } + return meta ? { data, meta, rows: data.length } : undefined; + }, [slots, specs]); + + return { mergedResponse, isLoading, allFailed, anyError, error, isComplete }; +} + +const EMPTY_NUMBER_FORMATS = new Map(); + +/** + * The multi-source search histogram: one stacked count() series per selected + * source, colored consistently with the results-table badges. A thin + * counterpart to DBTimeChart — drag-to-zoom and the legend work; per-series + * drill-down/pinned tooltips are single-source features and are omitted. + */ +export function MultiSourceTimeChart({ + specs, + enabled = true, + queryKeyPrefix, + enableParallelQueries, + onTimeRangeSelect, + showLegend = true, +}: { + specs: MultiSourceChartSpec[]; + enabled?: boolean; + queryKeyPrefix: string; + enableParallelQueries?: boolean; + onTimeRangeSelect?: (start: Date, end: Date) => void; + showLegend?: boolean; +}) { + const { mergedResponse, isLoading, allFailed, error, isComplete } = + useMultiSourceHistogram(specs, { + enabled, + queryKeyPrefix, + enableParallelQueries, + }); + + const firstConfig = specs[0]?.config; + const { dateRange, granularity } = useTimeChartSettings( + firstConfig ?? STUB_CONFIG, + ); + + const [activeClickPayload, setActiveClickPayload] = useState< + ActiveClickPayload | undefined + >(); + + const colorBySourceName = useMemo( + () => + new Map( + specs.map((spec, i) => [spec.source.name, getMultiSourceColor(i)]), + ), + [specs], + ); + + const formatted = useMemo(() => { + if (mergedResponse == null) { + return null; + } + try { + const result = formatResponseForTimeChart({ + currentPeriodResponse: mergedResponse, + dateRange, + granularity, + generateEmptyBuckets: true, + }); + // One series per source: recolor to the shared per-source palette so + // the histogram matches the table badges and status chips. + for (const line of result.lineData) { + const color = colorBySourceName.get(line.dataKey); + if (color != null) { + line.color = color; + } + } + return result; + } catch (e) { + console.error(e); + return null; + } + }, [mergedResponse, dateRange, granularity, colorBySourceName]); + + if (allFailed && error) { + return ; + } + + return ( + + {isLoading && formatted == null ? ( +
+ Loading Chart Data... +
+ ) : formatted == null || formatted.graphResults.length === 0 ? ( +
+ No data found within time range. +
+ ) : ( + + )} + + ); +} + +/** + * Summed "N Results" across every selected source, sharing the histogram's + * per-source queries (identical query keys) so it adds no ClickHouse load. + */ +export function MultiSourceTotalCountChart({ + specs, + enabled = true, + queryKeyPrefix, + enableParallelQueries, +}: { + specs: MultiSourceChartSpec[]; + enabled?: boolean; + queryKeyPrefix: string; + enableParallelQueries?: boolean; +}) { + const { mergedResponse, isLoading, allFailed } = useMultiSourceHistogram( + specs, + { + enabled, + queryKeyPrefix, + enableParallelQueries, + }, + ); + + const totalCount = useMemo(() => { + if (mergedResponse == null) return undefined; + // The count column may be renamed (e.g. via materialized views); fall back + // to the first numeric column, mirroring SearchTotalCountChart. + const countColumn = + mergedResponse.meta?.find(c => c.name === 'count()')?.name ?? + filterColumnMetaByType(mergedResponse.meta ?? [], [ + JSDataType.Number, + ])?.[0]?.name ?? + 'count()'; + return mergedResponse.data.reduce( + (sum: number, row: any) => sum + (Number.parseInt(row[countColumn]) || 0), + 0, + ); + }, [mergedResponse]); + + return ( + + {isLoading && totalCount == null ? ( + ··· Results + ) : totalCount != null && !allFailed ? ( + `${totalCount.toLocaleString()} Results` + ) : ( + '0 Results' + )} + + ); +} diff --git a/packages/app/src/components/SearchResultsTable.tsx b/packages/app/src/components/SearchResultsTable.tsx new file mode 100644 index 0000000000..428f55ba27 --- /dev/null +++ b/packages/app/src/components/SearchResultsTable.tsx @@ -0,0 +1,618 @@ +import { useCallback, useEffect, useMemo, useState } from 'react'; +import { useQueryState } from 'nuqs'; +import { + chSqlToAliasMap, + ClickHouseQueryError, + ColumnMetaType, + convertCHDataTypeToJSType, + isJSDataTypeJSONStringifiable, + JSDataType, +} from '@hyperdx/common-utils/dist/clickhouse'; +import { MULTI_SOURCE_ALIASES } from '@hyperdx/common-utils/dist/core/searchChartConfig'; +import { + BuilderChartConfigWithDateRange, + SourceKind, + TSource, +} from '@hyperdx/common-utils/dist/types'; +import { Flex, Group, Loader, Text, Tooltip } from '@mantine/core'; +import { IconAlertTriangle, IconFilterOff } from '@tabler/icons-react'; +import { SortingState } from '@tanstack/react-table'; + +import api from '@/api'; +import { searchChartConfigDefaults } from '@/defaults'; +import { useMultiSourceSlots } from '@/hooks/useMultiSourceSearch'; +import useOffsetPaginatedQuery from '@/hooks/useOffsetPaginatedQuery'; +import useRowWhere, { RowWhereResult, WithClause } from '@/hooks/useRowWhere'; +import { + mergeStreams, + MULTI_SOURCE_ROW_FIELDS, + StreamSnapshot, +} from '@/utils/multiSourceMerge'; +import { parseAsStringEncoded } from '@/utils/queryParsers'; + +import ChartErrorState from './charts/ChartErrorState'; +import DBRowSidePanel, { + RowSidePanelContext, + RowSidePanelContextProps, +} from './DBRowSidePanel'; +import { + DenoisedPatternsSummary, + getSelectLength, + RawLogTable, + selectColumnMapWithoutAdditionalKeys, + useConfigWithAdditionalSelect, + useDenoisedRows, +} from './DBRowTable'; +import { RowOverviewPanelWrapper } from './DBSqlRowTableWithSidebar'; +import { getMultiSourceColor, SourceBadge } from './MultiSourceBadge'; + +/** + * One selected source plus its fully-built chart config. + * + * With a single source the config carries that source's own SELECT (the user + * authored it); with several, each config projects the canonical + * MULTI_SOURCE_ALIASES so the merged rows share one shape. + */ +export type SearchStreamSpec = { + source: TSource; + config: BuilderChartConfigWithDateRange; + /** + * When set, the source doesn't run at all (e.g. an active filter references + * a column its table lacks); shown on the source's status chip. + */ + disabledReason?: string; +}; + +// Placeholder config for unused hook slots. The metadata hooks inside +// useConfigWithAdditionalSelect self-disable on empty table names, and the +// paginated query slot is explicitly disabled, so this never reaches +// ClickHouse. +const STUB_CONFIG: BuilderChartConfigWithDateRange = { + connection: '', + from: { databaseName: '', tableName: '' }, + timestampValueExpression: '', + select: '', + where: '', + whereLanguage: 'sql', + dateRange: [new Date(0), new Date(0)], +}; + +const EMPTY_CHSQL = { sql: '', params: {} }; +const EMPTY_EXTRA_COLUMNS: string[] = []; + +type SourceStream = { + spec: SearchStreamSpec | undefined; + data: ReturnType['data']; + fetchNextPage: ReturnType['fetchNextPage']; + hasNextPage: boolean; + isFetching: boolean; + isError: boolean; + error: Error | ClickHouseQueryError | null; + getRowWhere: (row: Record) => RowWhereResult; + /** Row-identity columns appended to the SELECT, trimmed off for display. */ + additionalKeysLength: number | undefined; +}; + +/** + * One source's independent query pipeline: the same + * defaults → additional-key SELECT merge → windowed offset pagination → + * row-WHERE machinery as the single-source DBSqlRowTable, packaged as a + * useMultiSourceSlots slot hook. Unused slots get a stub config and stay + * disabled. + */ +function useSourceStream( + spec: SearchStreamSpec | undefined, + { + enabled, + isLive, + enableSmallFirstWindow, + queryKeyPrefix, + }: { + enabled: boolean; + isLive: boolean; + enableSmallFirstWindow?: boolean; + queryKeyPrefix?: string; + }, +): SourceStream { + const { data: me } = api.useMe(); + + const configWithDefaults = useMemo( + () => ({ + ...searchChartConfigDefaults(me?.team), + ...(spec?.config ?? STUB_CONFIG), + }), + [me, spec?.config], + ); + + const mergedConfig = useConfigWithAdditionalSelect( + configWithDefaults, + spec?.source.id, + ); + + const { data, fetchNextPage, hasNextPage, isFetching, isError, error } = + useOffsetPaginatedQuery(mergedConfig ?? configWithDefaults, { + enabled: + enabled && + spec != null && + spec.disabledReason == null && + mergedConfig != null && + // An empty SELECT renders invalid SQL; wait for one to resolve. + getSelectLength(spec.config.select) > 0, + isLive, + queryKeyPrefix, + enableSmallFirstWindow, + }); + + const aliasMap = useMemo(() => { + const map = chSqlToAliasMap(data?.chSql ?? EMPTY_CHSQL); + // NULL-literal projections (`NULL AS "__hdx_duration_ms"` where a source + // lacks the field) are dropped by the SQL alias parser. Backfill them so + // the row-WHERE clause emits `isNull(NULL)` rather than referencing the + // alias as a (nonexistent) table column. ClickHouse reports NULL literals + // as Nullable(Nothing). + for (const col of data?.meta ?? []) { + if (map[col.name] == null && col.type === 'Nullable(Nothing)') { + map[col.name] = 'NULL'; + } + } + return map; + }, [data]); + + const getRowWhere = useRowWhere({ + meta: data?.meta, + aliasMap, + primaryKeyColumns: mergedConfig?.rowKeyColumns, + }); + + // Stable identity per content change, so downstream merge memos don't + // recompute (and re-sort every fetched row) on unrelated parent renders. + return useMemo( + () => ({ + spec, + data, + fetchNextPage, + hasNextPage: hasNextPage ?? false, + isFetching, + isError, + error: error ?? null, + getRowWhere, + additionalKeysLength: mergedConfig?.additionalKeysLength, + }), + [ + spec, + data, + fetchNextPage, + hasNextPage, + isFetching, + isError, + error, + getRowWhere, + mergedConfig?.additionalKeysLength, + ], + ); +} + +const COLUMN_NAME_MAP: Record = { + [MULTI_SOURCE_ALIASES.timestamp]: 'Timestamp', + [MULTI_SOURCE_ROW_FIELDS.SOURCE_NAME]: 'Source', + [MULTI_SOURCE_ALIASES.service]: 'Service', + [MULTI_SOURCE_ALIASES.severity]: 'Level', + [MULTI_SOURCE_ALIASES.durationMs]: 'Duration (ms)', + [MULTI_SOURCE_ALIASES.body]: 'Message', +}; + +function StreamStatusChips({ streams }: { streams: SourceStream[] }) { + return ( + + {streams.map((stream, i) => { + if (stream.spec == null) return null; + const name = stream.spec.source.name; + const disabledReason = stream.spec.disabledReason; + return ( + + + {stream.isFetching && } + {disabledReason != null && ( + + + + + + )} + {stream.isError && ( + + + + + + )} + + ); + })} + + ); +} + +export default function SearchResultsTable({ + sources: specs, + isLive, + enabled = true, + extraColumnNames = EMPTY_EXTRA_COLUMNS, + denoiseResults = false, + sortOrder, + onSortingChange, + onError, + onResolvedColumnsChange, + onScroll, + onSidebarOpen, + onExpandedRowsChange, + collapseAllRows, + enableSmallFirstWindow, + tableId, + context, + keepOpenSelector, + // Row queries are keyed separately from the page's chart/count queries, so + // "is the search fetching?" (live-tail pause, latency telemetry) keeps + // measuring the same thing it always has. + queryKeyPrefix = 'dbSqlRowTable', +}: { + /** 1..MAX_SEARCH_SOURCES selected sources with their built configs. */ + sources: SearchStreamSpec[]; + isLive: boolean; + enabled?: boolean; + /** User-picked extra columns projected into every source's SELECT (N>1). */ + extraColumnNames?: string[]; + /** Drop noisy event patterns from the results (single source only). */ + denoiseResults?: boolean; + /** Current sort, for the single-source case where sorting is supported. */ + sortOrder?: SortingState; + onSortingChange?: (v: SortingState | null) => void; + /** + * Surface a query failure to the page. Only called with a single source — + * with several, a failing source is isolated to its own status chip rather + * than failing the whole search. + */ + onError?: (error: Error | ClickHouseQueryError) => void; + onResolvedColumnsChange?: (meta: ColumnMetaType[]) => void; + onScroll?: (scrollTop: number) => void; + onSidebarOpen?: (rowId: string) => void; + onExpandedRowsChange?: (hasExpandedRows: boolean) => void; + collapseAllRows?: boolean; + enableSmallFirstWindow?: boolean; + tableId?: string; + context?: RowSidePanelContextProps; + keepOpenSelector?: string; + queryKeyPrefix?: string; +}) { + const slots = useMultiSourceSlots(specs, useSourceStream, { + enabled, + isLive, + enableSmallFirstWindow, + queryKeyPrefix, + }); + + const streams = useMemo( + () => + slots.filter( + (s): s is SourceStream & { spec: SearchStreamSpec } => s.spec != null, + ), + [slots], + ); + + // With one source the table shows that source's own SELECT, sorts, and + // denoises — everything the single-source search has always done. The + // canonical aliases, source badges, and cross-source merge only come into + // play once a second source is selected. + const isSingleSource = specs.length === 1; + const singleStream = isSingleSource ? streams[0] : undefined; + + const snapshots: StreamSnapshot[] = useMemo( + () => + streams.map((stream, i) => ({ + sourceId: stream.spec.source.id, + sourceName: stream.spec.source.name, + sourceColor: getMultiSourceColor(i), + rows: stream.data?.data ?? [], + window: stream.data?.window ?? null, + lastPageRowCount: stream.data?.lastPageRowCount ?? null, + hasNextPage: stream.hasNextPage, + isActive: !stream.isError && stream.spec.disabledReason == null, + dateRange: stream.spec.config.dateRange, + })), + [streams], + ); + + // One source needs no merge: its rows already arrive timestamp-ordered from + // its own ORDER BY, and there is no other stream to hold a frontier against. + const merged = useMemo( + () => + isSingleSource + ? null + : mergeStreams(snapshots, 'DESC', MULTI_SOURCE_ALIASES.timestamp), + [isSingleSource, snapshots], + ); + + const columnTypeMap = useMemo(() => { + if (singleStream != null) { + // The user's SELECT columns, positionally trimmed of the row-identity + // columns the query appends (same resolution as DBSqlRowTable). + return selectColumnMapWithoutAdditionalKeys( + singleStream.data?.meta, + singleStream.additionalKeysLength, + ); + } + // Merge column meta across streams by canonical alias name, preferring a + // resolved type over the Nullable(Nothing) a `NULL AS "alias"` projection + // reports. + const map = new Map(); + for (const stream of streams) { + for (const col of stream.data?.meta ?? []) { + const jsType = convertCHDataTypeToJSType(col.type); + const existing = map.get(col.name); + if (existing == null || existing._type == null) { + map.set(col.name, { _type: jsType }); + } + } + } + map.set(MULTI_SOURCE_ROW_FIELDS.SOURCE_NAME, { + _type: JSDataType.String, + }); + return map; + }, [streams, singleStream]); + + const includeDuration = specs.some(s => s.source.kind === SourceKind.Trace); + + const displayedColumns = useMemo(() => { + if (isSingleSource) { + return Array.from(columnTypeMap.keys()); + } + return [ + MULTI_SOURCE_ALIASES.timestamp, + MULTI_SOURCE_ROW_FIELDS.SOURCE_NAME, + MULTI_SOURCE_ALIASES.service, + MULTI_SOURCE_ALIASES.severity, + ...(includeDuration ? [MULTI_SOURCE_ALIASES.durationMs] : []), + ...extraColumnNames, + MULTI_SOURCE_ALIASES.body, + ]; + }, [isSingleSource, columnTypeMap, includeDuration, extraColumnNames]); + + // Stringify object-typed cells (Map/Array/JSON) the same way DBSqlRowTable + // does — both for display and because useRowWhere expects the stringified + // form when rebuilding a row WHERE clause. + const rows = useMemo(() => { + const baseRows = singleStream + ? (singleStream.data?.data ?? []) + : (merged?.rows ?? []); + const objectColumns = [...columnTypeMap.entries()] + .filter(([, v]) => isJSDataTypeJSONStringifiable(v._type)) + .map(([name]) => name); + if (objectColumns.length === 0) { + return baseRows; + } + return baseRows.map(row => { + const newRow = { ...row }; + for (const col of objectColumns) { + if (!(col in newRow) || newRow[col] == null) continue; + if (columnTypeMap.get(col)?._type === JSDataType.JSON) { + newRow[col] = JSON.stringify(newRow[col]).replace(/\//g, '\\/'); + } else { + newRow[col] = JSON.stringify(newRow[col]); + } + } + return newRow; + }); + }, [singleStream, merged?.rows, columnTypeMap]); + + const patternColumn = displayedColumns[displayedColumns.length - 1]; + const denoise = useDenoisedRows({ + config: singleStream?.spec.config ?? STUB_CONFIG, + sourceId: singleStream?.spec.source.id, + processedRows: rows, + patternColumn, + // Denoising mines patterns from one table's body column; it has no + // cross-source meaning, so it only runs with a single source. + denoiseResults: denoiseResults && isSingleSource, + isLive, + }); + + // Row identity dispatches to the row's own stream: each stream has its own + // result meta / alias map / primary-key columns. The client-side source tags + // are stripped first — they aren't real columns. + const generateRowId = useCallback( + (row: Record): RowWhereResult => { + if (singleStream != null) { + return singleStream.getRowWhere(row); + } + const { + [MULTI_SOURCE_ROW_FIELDS.SOURCE_ID]: sourceId, + [MULTI_SOURCE_ROW_FIELDS.SOURCE_NAME]: _name, + [MULTI_SOURCE_ROW_FIELDS.SOURCE_COLOR]: _color, + ...dbRow + } = row; + const stream = streams.find(s => s.spec.source.id === sourceId); + if (stream == null) { + return { where: '', aliasWith: [] }; + } + return stream.getRowWhere(dbRow); + }, + [streams, singleStream], + ); + + // Advance only the stream(s) holding the frontier back; the leaders keep + // their fetched-but-held rows until the laggards catch up. + const fetchNextPage = useCallback(() => { + if (singleStream != null) { + singleStream.fetchNextPage({ cancelRefetch: false }); + return; + } + for (const sourceId of merged?.laggingSourceIds ?? []) { + const stream = streams.find(s => s.spec.source.id === sourceId); + stream?.fetchNextPage({ cancelRefetch: false }); + } + }, [singleStream, merged?.laggingSourceIds, streams]); + + const hasNextPage = streams.some(s => !s.isError && s.hasNextPage); + const isFetching = streams.some(s => s.isFetching); + const isLoading = denoiseResults + ? isFetching || denoise.isFetching + : isFetching; + const allFailed = streams.length > 0 && streams.every(s => s.isError); + const firstError = streams.find(s => s.error != null)?.error ?? undefined; + + // A single source's failure is the whole search's failure, so the page owns + // the error UI (and drops out of live tail), exactly as before. + useEffect(() => { + if (singleStream?.isError && singleStream.error != null) { + onError?.(singleStream.error); + } + }, [singleStream?.isError, singleStream?.error, onError]); + + const singleMeta = singleStream?.data?.meta; + useEffect(() => { + if (singleMeta != null && singleMeta.length > 0) { + onResolvedColumnsChange?.(singleMeta); + } + }, [singleMeta, onResolvedColumnsChange]); + + // Side panel wiring — the same URL-param contract as the legacy table, + // except the panel's source comes from the clicked row rather than being + // fixed for the page. + const [rowId, setRowId] = useQueryState('rowWhere', parseAsStringEncoded); + const [rowSource, setRowSource] = useQueryState('rowSource'); + const [aliasWith, setAliasWith] = useState([]); + + const onRowDetailsClick = useCallback( + (row: Record) => { + const rowWhere = generateRowId(row); + if (!rowWhere.where) return; + setRowId(rowWhere.where); + setAliasWith(rowWhere.aliasWith); + setRowSource( + row[MULTI_SOURCE_ROW_FIELDS.SOURCE_ID] ?? + singleStream?.spec.source.id ?? + null, + ); + onSidebarOpen?.(rowWhere.where); + }, + [generateRowId, setRowId, setRowSource, onSidebarOpen, singleStream], + ); + + const onCloseSidebar = useCallback(() => { + setRowId(null); + setRowSource(null); + }, [setRowId, setRowSource]); + + const sourceForRow = useCallback( + (id: unknown) => + specs.find(s => s.source.id === id)?.source ?? + // Links predating the rowSource param (and every single-source link) + // carry only rowWhere; there is exactly one source it can belong to. + (isSingleSource ? specs[0]?.source : undefined), + [specs, isSingleSource], + ); + + const panelSource = useMemo( + () => sourceForRow(rowSource), + [sourceForRow, rowSource], + ); + + const renderRowDetails = useCallback( + (r: { id: string; aliasWith?: WithClause[]; [key: string]: unknown }) => { + const source = sourceForRow(r[MULTI_SOURCE_ROW_FIELDS.SOURCE_ID]); + if (!source) { + return
Loading...
; + } + return ( + + ); + }, + [sourceForRow], + ); + + const loadingDate = singleStream + ? singleStream.data?.window?.direction === 'ASC' + ? singleStream.data?.window?.endTime + : singleStream.data?.window?.startTime + : merged?.frontier != null && hasNextPage + ? new Date(merged.frontier) + : undefined; + + const firstConfig = streams[0]?.spec.config; + + return ( + + {panelSource != null && ( + + )} + + {/* One source needs no legend: every row came from it. */} + {!isSingleSource && } + {denoiseResults && isSingleSource && ( + + )} + {allFailed && !isSingleSource ? ( + + ) : ( + + )} + + + ); +} diff --git a/packages/app/src/defaults.ts b/packages/app/src/defaults.ts index 5894d8b78d..5b67f68216 100644 --- a/packages/app/src/defaults.ts +++ b/packages/app/src/defaults.ts @@ -2,6 +2,14 @@ import type { BuilderChartConfigWithDateRange } from '@hyperdx/common-utils/dist // Limit defaults export const DEFAULT_SEARCH_ROW_LIMIT = 200; + +// Ceiling on how many sources one search can span. Cost scales linearly with +// the selection: each source runs its own result stream plus histogram/count +// aggregates (~3 ClickHouse queries per source per refresh, re-fired every +// live-tail tick), so 3 keeps the worst case bounded while covering the +// common "app logs + infra logs + traces" setups. Also the hook-slot count in +// useMultiSourceSlots — raising it means adding a slot there too. +export const MAX_SEARCH_SOURCES = 3; export const DEFAULT_QUERY_TIMEOUT = 60; // max_execution_time, seconds export const DEFAULT_FILTER_KEYS_FETCH_LIMIT = 100; export const DEFAULT_SERIES_LIMIT = 100; diff --git a/packages/app/src/hooks/__tests__/useMultiSourceSearch.test.ts b/packages/app/src/hooks/__tests__/useMultiSourceSearch.test.ts new file mode 100644 index 0000000000..58be8d33d2 --- /dev/null +++ b/packages/app/src/hooks/__tests__/useMultiSourceSearch.test.ts @@ -0,0 +1,87 @@ +import { Filter } from '@hyperdx/common-utils/dist/types'; + +import { + filterRootColumn, + resolveExtraColumnsForSource, + unresolvedFilterColumns, +} from '@/hooks/useMultiSourceSearch'; + +describe('filterRootColumn', () => { + it('extracts a plain column reference', () => { + const filter: Filter = { + type: 'sql_ast', + operator: '=', + left: 'ServiceName', + right: "'cart'", + }; + expect(filterRootColumn(filter)).toBe('ServiceName'); + }); + + it('extracts the root of a map subscript', () => { + const filter: Filter = { + type: 'sql_ast', + operator: '=', + left: "LogAttributes['level']", + right: "'error'", + }; + expect(filterRootColumn(filter)).toBe('LogAttributes'); + }); + + it('extracts a backticked identifier', () => { + const filter: Filter = { + type: 'sql_ast', + operator: '=', + left: '`weird-col`', + right: "'x'", + }; + expect(filterRootColumn(filter)).toBe('weird-col'); + }); + + it('returns null for raw sql and lucene filters', () => { + expect( + filterRootColumn({ type: 'sql', condition: "Foo = 'bar'" }), + ).toBeNull(); + expect( + filterRootColumn({ type: 'lucene', condition: 'foo:bar' }), + ).toBeNull(); + }); +}); + +describe('unresolvedFilterColumns', () => { + const filters: Filter[] = [ + { type: 'sql_ast', operator: '=', left: 'ServiceName', right: "'cart'" }, + { type: 'sql_ast', operator: '=', left: 'StatusCode', right: "'Unset'" }, + { type: 'sql', condition: 'anything' }, + ]; + + it('reports columns the source lacks', () => { + expect( + unresolvedFilterColumns(filters, new Set(['ServiceName', 'Body'])), + ).toEqual(['StatusCode']); + }); + + it('is empty when every attributable column resolves', () => { + expect( + unresolvedFilterColumns(filters, new Set(['ServiceName', 'StatusCode'])), + ).toEqual([]); + }); + + it('is empty (not excluding) while columns are still unknown', () => { + expect(unresolvedFilterColumns(filters, undefined)).toEqual([]); + }); +}); + +describe('resolveExtraColumnsForSource', () => { + it('projects the column where present and NULL where missing', () => { + expect( + resolveExtraColumnsForSource( + ['ServiceName', 'StatusCode', 'weird col'], + new Set(['ServiceName', 'weird col']), + ), + ).toEqual([ + { name: 'ServiceName', expression: 'ServiceName' }, + { name: 'StatusCode', expression: null }, + { name: 'weird col', expression: '`weird col`' }, + ]); + }); +}); diff --git a/packages/app/src/hooks/useMultiSourceSearch.ts b/packages/app/src/hooks/useMultiSourceSearch.ts new file mode 100644 index 0000000000..691cadc9bf --- /dev/null +++ b/packages/app/src/hooks/useMultiSourceSearch.ts @@ -0,0 +1,175 @@ +import { useMemo } from 'react'; +import { + ColumnMeta, + filterColumnMetaByType, + JSDataType, +} from '@hyperdx/common-utils/dist/clickhouse'; +import { MultiSourceExtraColumn } from '@hyperdx/common-utils/dist/core/searchChartConfig'; +import { Filter, TSource } from '@hyperdx/common-utils/dist/types'; + +import { MAX_SEARCH_SOURCES } from '@/defaults'; +import { useColumns } from '@/hooks/useMetadata'; + +/** + * Run one instance of a hook per selected source of a multi-source search. + * + * The rules of hooks require a constant hook count per component, but multi + * mode needs one query pipeline per selected source — and `useQueries` can't + * cover these pipelines (the row streams are `useInfiniteQuery`-based, which + * has no plural form, and the chart pipeline composes other hooks). So the + * hook count is pinned at MAX_SEARCH_SOURCES here, in one place: unused + * slots receive `undefined` and every slot hook is expected to self-disable + * for it. + * + * `useSlot` must be a stable, named hook (the rules-of-hooks lint understands + * `use*`-named parameters) and should return a memoized value, so the array + * this returns is referentially stable and safe to use in dependency lists. + */ +export function useMultiSourceSlots( + items: readonly Item[], + useSlot: (item: Item | undefined, opts: Opts) => Result, + opts: Opts, +): Result[] { + const s0 = useSlot(items[0], opts); + const s1 = useSlot(items[1], opts); + const s2 = useSlot(items[2], opts); + const count = Math.min(items.length, MAX_SEARCH_SOURCES); + return useMemo(() => [s0, s1, s2].slice(0, count), [s0, s1, s2, count]); +} + +const EMPTY_SOURCE_PARAMS = { + databaseName: '', + tableName: '', + connectionId: '', +}; + +function columnsParamsFor(source: TSource | undefined) { + if (source == null) return EMPTY_SOURCE_PARAMS; + return { + databaseName: source.from.databaseName, + tableName: source.from.tableName, + connectionId: source.connection, + }; +} + +export type MultiSourceColumnOption = { + name: string; + /** How many of the selected sources have this column. */ + availableCount: number; +}; + +/** Slot hook: DESCRIBE columns for one source. Stable — `.data` is cached. */ +function useSourceColumnsSlot( + source: TSource | undefined, +): ColumnMeta[] | undefined { + return useColumns(columnsParamsFor(source)).data; +} + +/** + * Top-level columns (DESCRIBE) for each selected source of a multi-source + * search, plus the deduped union with per-column availability counts for the + * add-column picker. useColumns self-disables for unused slots. + */ +export function useMultiSourceColumns(sources: TSource[]): { + columnsBySourceId: Map>; + unionColumns: MultiSourceColumnOption[]; + /** Union of Date/DateTime column name → ClickHouse type across sources. */ + dateTimeColumns: Map; +} { + const slotData = useMultiSourceSlots( + sources, + useSourceColumnsSlot, + undefined, + ); + + return useMemo(() => { + const columnsBySourceId = new Map>(); + const availability = new Map(); + const dateTimeColumns = new Map(); + + for (let i = 0; i < sources.length; i++) { + const source = sources[i]; + const columns = slotData[i]; + if (source == null || columns == null) continue; + const names = new Set(columns.map(c => c.name)); + columnsBySourceId.set(source.id, names); + for (const name of names) { + availability.set(name, (availability.get(name) ?? 0) + 1); + } + for (const col of filterColumnMetaByType(columns, [JSDataType.Date]) ?? + []) { + if (!dateTimeColumns.has(col.name)) { + dateTimeColumns.set(col.name, col.type); + } + } + } + + const unionColumns = [...availability.entries()] + .map(([name, availableCount]) => ({ name, availableCount })) + .sort( + (a, b) => + b.availableCount - a.availableCount || a.name.localeCompare(b.name), + ); + + return { columnsBySourceId, unionColumns, dateTimeColumns }; + }, [slotData, sources]); +} + +/** + * Root column a filter references, for per-source resolvability checks. + * sql_ast filters carry the escaped SQL key in `left` (e.g. `ServiceName`, + * a backticked identifier, or `LogAttributes['level']` whose root is + * `LogAttributes`). Other filter types (raw sql/lucene conditions) can't be + * attributed to a single column and return null — callers should apply them + * to every source and rely on per-source error isolation. + */ +export function filterRootColumn(filter: Filter): string | null { + if (filter.type !== 'sql_ast') return null; + const left = filter.left.trim(); + const backticked = left.match(/^`([^`]+)`/); + if (backticked) return backticked[1]; + const plain = left.match(/^[A-Za-z_][A-Za-z0-9_]*/); + return plain ? plain[0] : null; +} + +/** + * For one source: which of the active filters reference a column its table + * doesn't have. A non-empty result means the source can't answer the + * filtered search and should be excluded (with a visible reason). + */ +export function unresolvedFilterColumns( + filters: Filter[], + sourceColumns: Set | undefined, +): string[] { + if (sourceColumns == null) return []; + const missing = new Set(); + for (const filter of filters) { + const root = filterRootColumn(filter); + if (root != null && !sourceColumns.has(root)) { + missing.add(root); + } + } + return [...missing]; +} + +/** Quote a column name as a ClickHouse identifier when it needs it. */ +function quoteIdentifier(name: string): string { + return /^[A-Za-z_][A-Za-z0-9_]*$/.test(name) + ? name + : `\`${name.replace(/`/g, '\\`')}\``; +} + +/** + * Resolve the user-picked extra column names into per-source SELECT + * expressions: the (quoted) column itself where the source's table has it, + * NULL otherwise — so every source still returns the same result shape. + */ +export function resolveExtraColumnsForSource( + extraColumnNames: string[], + sourceColumns: Set | undefined, +): MultiSourceExtraColumn[] { + return extraColumnNames.map(name => ({ + name, + expression: sourceColumns?.has(name) ? quoteIdentifier(name) : null, + })); +} diff --git a/packages/app/src/hooks/useOffsetPaginatedQuery.tsx b/packages/app/src/hooks/useOffsetPaginatedQuery.tsx index 5a27d39b96..2dabb32720 100644 --- a/packages/app/src/hooks/useOffsetPaginatedQuery.tsx +++ b/packages/app/src/hooks/useOffsetPaginatedQuery.tsx @@ -427,7 +427,9 @@ function flattenPages(pages: TQueryFnData[]) { return pages.flatMap(p => p.data); } -function flattenData(data: TData | undefined): TQueryFnData | null { +function flattenData( + data: TData | undefined, +): (TQueryFnData & { lastPageRowCount: number }) | null { if (data == null || data.pages.length === 0) { return null; } @@ -437,6 +439,10 @@ function flattenData(data: TData | undefined): TQueryFnData | null { data: flattenPages(data.pages), chSql: data.pages[0].chSql, window: data.pages[data.pages.length - 1].window, + // Whether the last fetched page hit results distinguishes "still mid-window + // at LIMIT" from "window drained" — multi-source merge uses this to compute + // how far this stream's time coverage safely extends. + lastPageRowCount: data.pages[data.pages.length - 1].data.length, }; } diff --git a/packages/app/src/hooks/useResolvedSourcesParam.ts b/packages/app/src/hooks/useResolvedSourcesParam.ts new file mode 100644 index 0000000000..deb2f449d1 --- /dev/null +++ b/packages/app/src/hooks/useResolvedSourcesParam.ts @@ -0,0 +1,58 @@ +import { useEffect, useMemo } from 'react'; +import { SourceKind, TSource } from '@hyperdx/common-utils/dist/types'; +import { notifications } from '@mantine/notifications'; + +import { MAX_SEARCH_SOURCES } from '@/defaults'; +import { useSources } from '@/source'; +import { resolveSourcesParam } from '@/utils/sourceParams'; + +const EMPTY_SOURCES: TSource[] = []; + +/** + * Resolves the multi-source search param (a list of source IDs or names) to + * the matching sources, deduped and capped at MAX_SEARCH_SOURCES. + * + * Elements that don't match any usable source are dropped from the selection + * and reported once via a Mantine warning, mirroring useResolvedSourceParam. + */ +export function useResolvedSourcesParam( + paramValues: string[] | null | undefined, + { kinds }: { kinds?: SourceKind[] } = {}, +): { sources: TSource[] } { + const { data: allSources } = useSources(); + + // Key the memo on a serialized `kinds` so callers can pass inline arrays + // without breaking memoization. + const kindsKey = kinds?.join(','); + const { sources, unresolvedKey } = useMemo(() => { + const allKinds = new Set(Object.values(SourceKind)); + const resolvedKinds = kindsKey + ? kindsKey.split(',').filter((k): k is SourceKind => allKinds.has(k)) + : undefined; + const resolution = resolveSourcesParam(paramValues, allSources, { + kinds: resolvedKinds, + max: MAX_SEARCH_SOURCES, + }); + if (resolution.status !== 'resolved') { + return { sources: EMPTY_SOURCES, unresolvedKey: undefined }; + } + return { + sources: resolution.sources.length ? resolution.sources : EMPTY_SOURCES, + unresolvedKey: resolution.unresolved.length + ? resolution.unresolved.join(', ') + : undefined, + }; + }, [paramValues, allSources, kindsKey]); + + useEffect(() => { + if (unresolvedKey == null) return; + notifications.show({ + id: 'sources-param-unresolved-' + unresolvedKey, + color: 'yellow', + title: 'Some sources were not found', + message: `No searchable source matches: ${unresolvedKey}. They may have been renamed or deleted.`, + }); + }, [unresolvedKey]); + + return useMemo(() => ({ sources }), [sources]); +} diff --git a/packages/app/src/utils/__tests__/multiSourceMerge.test.ts b/packages/app/src/utils/__tests__/multiSourceMerge.test.ts new file mode 100644 index 0000000000..ae77b18f9e --- /dev/null +++ b/packages/app/src/utils/__tests__/multiSourceMerge.test.ts @@ -0,0 +1,379 @@ +import { + computeFrontier, + coveredUntil, + mergeStreams, + MULTI_SOURCE_ROW_FIELDS, + StreamSnapshot, +} from '@/utils/multiSourceMerge'; + +const TS_KEY = '__hdx_timestamp'; + +const T = (iso: string) => new Date(iso); +const ms = (iso: string) => new Date(iso).getTime(); + +// Search range: 10:00 - 12:00 UTC +const DATE_RANGE: [Date, Date] = [ + T('2026-08-07T10:00:00Z'), + T('2026-08-07T12:00:00Z'), +]; + +const row = (iso: string, extra: Record = {}) => ({ + [TS_KEY]: iso, + ...extra, +}); + +const makeStream = ( + overrides: Partial & { sourceId: string }, +): StreamSnapshot => ({ + sourceName: overrides.sourceId, + rows: [], + window: null, + lastPageRowCount: null, + hasNextPage: true, + isActive: true, + dateRange: DATE_RANGE, + ...overrides, +}); + +const parseTs = (r: Record) => new Date(r[TS_KEY]).getTime(); + +describe('coveredUntil (DESC)', () => { + it('covers the whole range when the stream is fully drained', () => { + const stream = makeStream({ + sourceId: 'a', + hasNextPage: false, + window: { + startTime: T('2026-08-07T10:00:00Z'), + endTime: T('2026-08-07T12:00:00Z'), + }, + lastPageRowCount: 0, + }); + expect(coveredUntil(stream, 'DESC', parseTs)).toBe(DATE_RANGE[0].getTime()); + }); + + it('covers nothing during the initial fetch even though hasNextPage is still false', () => { + // useInfiniteQuery reports hasNextPage=false before the first page lands; + // that must not be mistaken for a drained stream. + const stream = makeStream({ sourceId: 'a', hasNextPage: false }); + expect(coveredUntil(stream, 'DESC', parseTs)).toBe(DATE_RANGE[1].getTime()); + }); + + it('covers nothing before the first page arrives', () => { + const stream = makeStream({ sourceId: 'a' }); + expect(coveredUntil(stream, 'DESC', parseTs)).toBe(DATE_RANGE[1].getTime()); + }); + + it('covers through the window start when the last page was empty', () => { + const stream = makeStream({ + sourceId: 'a', + window: { + startTime: T('2026-08-07T11:45:00Z'), + endTime: T('2026-08-07T12:00:00Z'), + }, + lastPageRowCount: 0, + }); + expect(coveredUntil(stream, 'DESC', parseTs)).toBe( + ms('2026-08-07T11:45:00Z'), + ); + }); + + it('covers only through the oldest fetched row when stopped mid-window at LIMIT', () => { + const stream = makeStream({ + sourceId: 'a', + rows: [row('2026-08-07T11:59:00Z'), row('2026-08-07T11:50:00Z')], + window: { + startTime: T('2026-08-07T11:45:00Z'), + endTime: T('2026-08-07T12:00:00Z'), + }, + lastPageRowCount: 2, + }); + expect(coveredUntil(stream, 'DESC', parseTs)).toBe( + ms('2026-08-07T11:50:00Z'), + ); + }); +}); + +describe('coveredUntil (ASC)', () => { + it('mirrors the DESC semantics from the start of the range', () => { + expect(coveredUntil(makeStream({ sourceId: 'a' }), 'ASC', parseTs)).toBe( + DATE_RANGE[0].getTime(), + ); + expect( + coveredUntil( + makeStream({ + sourceId: 'a', + hasNextPage: false, + window: { + startTime: T('2026-08-07T10:00:00Z'), + endTime: T('2026-08-07T12:00:00Z'), + }, + lastPageRowCount: 0, + }), + 'ASC', + parseTs, + ), + ).toBe(DATE_RANGE[1].getTime()); + expect( + coveredUntil( + makeStream({ + sourceId: 'a', + window: { + startTime: T('2026-08-07T10:00:00Z'), + endTime: T('2026-08-07T10:15:00Z'), + }, + lastPageRowCount: 0, + }), + 'ASC', + parseTs, + ), + ).toBe(ms('2026-08-07T10:15:00Z')); + }); +}); + +const drainedStream = (sourceId: string) => + makeStream({ + sourceId, + hasNextPage: false, + window: { + startTime: T('2026-08-07T10:00:00Z'), + endTime: T('2026-08-07T12:00:00Z'), + }, + lastPageRowCount: 0, + }); + +describe('computeFrontier', () => { + it('is the least-covered active stream (max for DESC)', () => { + const drained = drainedStream('a'); + const midWindow = makeStream({ + sourceId: 'b', + rows: [row('2026-08-07T11:50:00Z')], + window: { + startTime: T('2026-08-07T11:45:00Z'), + endTime: T('2026-08-07T12:00:00Z'), + }, + lastPageRowCount: 1, + }); + expect(computeFrontier([drained, midWindow], 'DESC', parseTs)).toBe( + ms('2026-08-07T11:50:00Z'), + ); + }); + + it('ignores inactive (errored/excluded) streams so they cannot stall the merge', () => { + const drained = drainedStream('a'); + const errored = makeStream({ sourceId: 'b', isActive: false }); + expect(computeFrontier([drained, errored], 'DESC', parseTs)).toBe( + DATE_RANGE[0].getTime(), + ); + }); + + it('is null when no stream is active', () => { + const errored = makeStream({ sourceId: 'a', isActive: false }); + expect(computeFrontier([errored], 'DESC', parseTs)).toBeNull(); + }); +}); + +describe('mergeStreams', () => { + const window0 = { + startTime: T('2026-08-07T11:45:00Z'), + endTime: T('2026-08-07T12:00:00Z'), + }; + + it('interleaves rows across streams newest-first and tags their source', () => { + const a = makeStream({ + sourceId: 'a', + sourceName: 'app logs', + rows: [row('2026-08-07T11:59:00Z'), row('2026-08-07T11:57:00Z')], + window: window0, + lastPageRowCount: 2, + hasNextPage: false, + }); + const b = makeStream({ + sourceId: 'b', + sourceName: 'traces', + rows: [row('2026-08-07T11:58:00Z')], + window: window0, + lastPageRowCount: 1, + hasNextPage: false, + }); + + const { rows } = mergeStreams([a, b], 'DESC', TS_KEY); + + expect(rows.map(r => r[TS_KEY])).toEqual([ + '2026-08-07T11:59:00Z', + '2026-08-07T11:58:00Z', + '2026-08-07T11:57:00Z', + ]); + expect(rows.map(r => r[MULTI_SOURCE_ROW_FIELDS.SOURCE_NAME])).toEqual([ + 'app logs', + 'traces', + 'app logs', + ]); + expect(rows[0][MULTI_SOURCE_ROW_FIELDS.SOURCE_ID]).toBe('a'); + }); + + it('holds back rows older than the frontier until lagging streams catch up', () => { + // Stream a is fully drained down to 10:00; stream b stopped at LIMIT with + // its oldest row at 11:50 — anything older than 11:50 from a must wait. + const a = makeStream({ + sourceId: 'a', + rows: [ + row('2026-08-07T11:55:00Z'), + row('2026-08-07T11:49:00Z'), // older than b's coverage — held back + ], + window: window0, + lastPageRowCount: 2, + hasNextPage: false, + }); + const b = makeStream({ + sourceId: 'b', + rows: [row('2026-08-07T11:50:00Z')], + window: window0, + lastPageRowCount: 1, + hasNextPage: true, + }); + + const { rows, frontier, laggingSourceIds } = mergeStreams( + [a, b], + 'DESC', + TS_KEY, + ); + + expect(frontier).toBe(ms('2026-08-07T11:50:00Z')); + expect(rows.map(r => r[TS_KEY])).toEqual([ + '2026-08-07T11:55:00Z', + '2026-08-07T11:50:00Z', + ]); + expect(laggingSourceIds).toEqual(['b']); + }); + + it('shows everything when all streams are drained', () => { + const a = makeStream({ + sourceId: 'a', + rows: [row('2026-08-07T10:05:00Z')], + window: window0, + lastPageRowCount: 1, + hasNextPage: false, + }); + const b = makeStream({ + sourceId: 'b', + rows: [row('2026-08-07T10:03:00Z')], + window: window0, + lastPageRowCount: 1, + hasNextPage: false, + }); + + const { rows, laggingSourceIds } = mergeStreams([a, b], 'DESC', TS_KEY); + + expect(rows).toHaveLength(2); + expect(laggingSourceIds).toEqual([]); + }); + + it('holds everything back while a stream has no page yet, without marking it lagging', () => { + const a = makeStream({ + sourceId: 'a', + rows: [row('2026-08-07T11:59:00Z')], + window: window0, + lastPageRowCount: 1, + hasNextPage: false, + }); + const pending = makeStream({ sourceId: 'b' }); + + const { rows, laggingSourceIds } = mergeStreams( + [a, pending], + 'DESC', + TS_KEY, + ); + + // Frontier sits at the range end until b's first page lands. + expect(rows).toEqual([]); + // b's initial fetch is already in flight — nothing to advance. + expect(laggingSourceIds).toEqual([]); + }); + + it('still shows rows from errored streams but never waits on them', () => { + const a = makeStream({ + sourceId: 'a', + rows: [row('2026-08-07T11:59:00Z')], + window: window0, + lastPageRowCount: 1, + hasNextPage: false, + }); + const errored = makeStream({ + sourceId: 'b', + rows: [row('2026-08-07T11:58:00Z')], + window: window0, + lastPageRowCount: 1, + isActive: false, + }); + + const { rows, laggingSourceIds } = mergeStreams( + [a, errored], + 'DESC', + TS_KEY, + ); + + expect(rows.map(r => r[TS_KEY])).toEqual([ + '2026-08-07T11:59:00Z', + '2026-08-07T11:58:00Z', + ]); + expect(laggingSourceIds).toEqual([]); + }); + + it('merges oldest-first with a mirrored frontier for ASC', () => { + const window0Asc = { + startTime: T('2026-08-07T10:00:00Z'), + endTime: T('2026-08-07T10:15:00Z'), + }; + const a = makeStream({ + sourceId: 'a', + rows: [ + row('2026-08-07T10:01:00Z'), + row('2026-08-07T10:20:00Z'), // beyond b's coverage — held back + ], + window: window0Asc, + lastPageRowCount: 2, + hasNextPage: false, + }); + const b = makeStream({ + sourceId: 'b', + rows: [row('2026-08-07T10:05:00Z')], + window: window0Asc, + lastPageRowCount: 1, + hasNextPage: true, + }); + + const { rows, frontier, laggingSourceIds } = mergeStreams( + [a, b], + 'ASC', + TS_KEY, + ); + + expect(frontier).toBe(ms('2026-08-07T10:05:00Z')); + expect(rows.map(r => r[TS_KEY])).toEqual([ + '2026-08-07T10:01:00Z', + '2026-08-07T10:05:00Z', + ]); + expect(laggingSourceIds).toEqual(['b']); + }); + + it('advances every stream tied at the frontier', () => { + const a = makeStream({ + sourceId: 'a', + rows: [row('2026-08-07T11:50:00Z')], + window: window0, + lastPageRowCount: 1, + hasNextPage: true, + }); + const b = makeStream({ + sourceId: 'b', + rows: [row('2026-08-07T11:50:00Z')], + window: window0, + lastPageRowCount: 1, + hasNextPage: true, + }); + + const { laggingSourceIds } = mergeStreams([a, b], 'DESC', TS_KEY); + + expect(laggingSourceIds).toEqual(['a', 'b']); + }); +}); diff --git a/packages/app/src/utils/multiSourceMerge.ts b/packages/app/src/utils/multiSourceMerge.ts new file mode 100644 index 0000000000..ae53b730c4 --- /dev/null +++ b/packages/app/src/utils/multiSourceMerge.ts @@ -0,0 +1,239 @@ +/** + * Pure merge logic for multi-source search: k-way merges per-source result + * streams by timestamp, bounded by a "safe frontier" so the interleaved + * timeline never shows a gap another source could still fill. + * + * Every source stream paginates through the same progressive time windows + * (see utils/searchWindows.ts — windows are a pure function of the date + * range), but streams advance at different speeds: one source may be three + * windows deep while another is still mid-window at its row LIMIT. A merged + * DESC timeline is only correct down to the timestamp every stream has + * covered; rows older than that are held back until the lagging streams catch + * up. + */ + +/** Client-side fields tagged onto every merged row. Never sent to ClickHouse. */ +export const MULTI_SOURCE_ROW_FIELDS = { + SOURCE_ID: '__hdx_source_id', + SOURCE_NAME: '__hdx_source_name', + SOURCE_COLOR: '__hdx_source_color', +} as const; + +export type MergeDirection = 'ASC' | 'DESC'; + +export type StreamSnapshot = { + sourceId: string; + sourceName: string; + /** Badge/series color for this source; tagged onto rows for the table cell. */ + sourceColor?: string; + /** + * Rows fetched so far, in stream order (newest-first for DESC, + * oldest-first for ASC) — the order the windowed query produces. + */ + rows: Record[]; + /** The last fetched page's time window; null when no page has completed. */ + window: { startTime: Date; endTime: Date } | null; + /** + * Row count of the last fetched page; 0 means the window was drained, + * >0 means the stream may have stopped mid-window at its LIMIT. + * Null when no page has completed. + */ + lastPageRowCount: number | null; + hasNextPage: boolean; + /** + * Errored/excluded streams don't bound the frontier (they'd stall the merge + * forever); their already-fetched rows are still shown. + */ + isActive: boolean; + /** The full searched range, used when a stream is fully drained. */ + dateRange: [Date, Date]; +}; + +/** + * Epoch-ms timestamp T such that this stream is guaranteed to have produced + * every row it has on the already-covered side of T: + * DESC — all of the stream's rows with ts >= T are fetched; + * ASC — all of the stream's rows with ts <= T are fetched. + * + * Conservative by construction: when the stream stopped mid-window at its + * LIMIT, coverage only extends to the last row it returned, not the window + * boundary. + */ +export function coveredUntil( + stream: StreamSnapshot, + direction: MergeDirection, + parseTs: (row: Record) => number, +): number { + const [start, end] = stream.dateRange; + + if (stream.window == null || stream.lastPageRowCount == null) { + // Nothing fetched yet: no coverage at all. Checked before hasNextPage — + // useInfiniteQuery reports hasNextPage=false during the initial fetch, + // which must not read as "fully drained". + return direction === 'DESC' ? end.getTime() : start.getTime(); + } + + if (!stream.hasNextPage) { + // Fully drained: the stream covered the entire searched range. + return direction === 'DESC' ? start.getTime() : end.getTime(); + } + + if (stream.lastPageRowCount === 0) { + // The last window came back empty, so it is fully covered. + return direction === 'DESC' + ? stream.window.startTime.getTime() + : stream.window.endTime.getTime(); + } + + // Mid-window at LIMIT: covered only through the last row returned. Rows are + // in stream order, so the last row is the furthest-along one. + const lastRow = stream.rows[stream.rows.length - 1]; + if (lastRow == null) { + // Defensive: a non-zero lastPageRowCount implies rows exist. + return direction === 'DESC' + ? stream.window.endTime.getTime() + : stream.window.startTime.getTime(); + } + return parseTs(lastRow); +} + +/** + * The merge frontier: the timestamp every active stream has covered. + * DESC — rows with ts >= frontier are safe to show; ASC — ts <= frontier. + * Null when there are no active streams (nothing bounds the merge). + */ +export function computeFrontier( + streams: StreamSnapshot[], + direction: MergeDirection, + parseTs: (row: Record) => number, +): number | null { + let frontier: number | null = null; + for (const stream of streams) { + if (!stream.isActive) continue; + const covered = coveredUntil(stream, direction, parseTs); + if (frontier == null) { + frontier = covered; + } else { + frontier = + direction === 'DESC' + ? Math.max(frontier, covered) + : Math.min(frontier, covered); + } + } + return frontier; +} + +/** + * The active streams holding the frontier back that can be advanced with + * another page fetch. Streams whose initial fetch hasn't completed are not + * included — their in-flight request IS their advancement. + */ +function laggingStreams( + streams: StreamSnapshot[], + direction: MergeDirection, + parseTs: (row: Record) => number, +): StreamSnapshot[] { + const frontier = computeFrontier(streams, direction, parseTs); + if (frontier == null) return []; + return streams.filter( + stream => + stream.isActive && + stream.hasNextPage && + stream.window != null && + coveredUntil(stream, direction, parseTs) === frontier, + ); +} + +export type MergedRow = Record; + +/** + * Merge all fetched rows across streams into one timestamp-ordered list, + * tagged with their origin source, held back at the frontier. + * + * Rows from inactive (errored/excluded) streams are still included — they are + * valid data — but only active streams bound the frontier, so a dead source + * can't freeze the timeline. + */ +function mergeStreamRows( + streams: StreamSnapshot[], + direction: MergeDirection, + timestampKey: string, +): MergedRow[] { + // Timestamps repeat heavily at second precision; cache the Date parse per + // distinct raw value (same trick as ChartUtils' time-chart transform). + const tsCache = new Map(); + const parseTs = (row: Record): number => { + const raw = row[timestampKey]; + let ts = tsCache.get(raw); + if (ts === undefined) { + ts = new Date(raw).getTime(); + tsCache.set(raw, ts); + } + return ts; + }; + + const frontier = computeFrontier(streams, direction, parseTs); + + const tagged: { row: MergedRow; ts: number }[] = []; + for (const stream of streams) { + for (const row of stream.rows) { + const ts = parseTs(row); + if ( + frontier != null && + (direction === 'DESC' ? ts < frontier : ts > frontier) + ) { + continue; + } + tagged.push({ + row: { + ...row, + [MULTI_SOURCE_ROW_FIELDS.SOURCE_ID]: stream.sourceId, + [MULTI_SOURCE_ROW_FIELDS.SOURCE_NAME]: stream.sourceName, + ...(stream.sourceColor != null + ? { [MULTI_SOURCE_ROW_FIELDS.SOURCE_COLOR]: stream.sourceColor } + : {}), + }, + ts, + }); + } + } + + // Array.prototype.sort is stable, so ties keep (stream order, row order). + tagged.sort((a, b) => (direction === 'DESC' ? b.ts - a.ts : a.ts - b.ts)); + + return tagged.map(t => t.row); +} + +/** + * Convenience wrapper used by the table component: one pass producing the + * merged rows, the frontier (for the "loading up to" indicator), and which + * streams to advance on the next fetch. + */ +export function mergeStreams( + streams: StreamSnapshot[], + direction: MergeDirection, + timestampKey: string, +): { + rows: MergedRow[]; + frontier: number | null; + laggingSourceIds: string[]; +} { + const tsCache = new Map(); + const parseTs = (row: Record): number => { + const raw = row[timestampKey]; + let ts = tsCache.get(raw); + if (ts === undefined) { + ts = new Date(raw).getTime(); + tsCache.set(raw, ts); + } + return ts; + }; + + return { + rows: mergeStreamRows(streams, direction, timestampKey), + frontier: computeFrontier(streams, direction, parseTs), + laggingSourceIds: laggingStreams(streams, direction, parseTs).map( + s => s.sourceId, + ), + }; +} diff --git a/packages/app/src/utils/sourceParams.ts b/packages/app/src/utils/sourceParams.ts index 5874111c99..dc759c11b7 100644 --- a/packages/app/src/utils/sourceParams.ts +++ b/packages/app/src/utils/sourceParams.ts @@ -55,6 +55,51 @@ export type SourceParamResolution = * lowest ID, so the same link always resolves to the same source no matter what * order the API returns them in. */ +/** + * Resolve a list of source params (IDs or names) for multi-source search. + * Each element resolves with the same rules as `resolveSourceParam`; results + * are deduped by ID and capped at `max`. Elements that can't be resolved (or + * resolve to a source of the wrong kind) are reported in `unresolved` so the + * caller can warn without failing the rest of the selection. + */ +export function resolveSourcesParam( + paramValues: string[] | null | undefined, + sources: T[] | undefined, + { kinds, max }: { kinds?: SourceKind[]; max?: number } = {}, +): + | { status: 'pending' } + | { status: 'resolved'; sources: T[]; unresolved: string[] } { + if (paramValues == null || paramValues.length === 0) { + return { status: 'resolved', sources: [], unresolved: [] }; + } + if (sources == null) return { status: 'pending' }; + + const resolved: T[] = []; + const seenIds = new Set(); + const unresolved: string[] = []; + + for (const value of paramValues) { + const resolution = resolveSourceParam(value, sources, { kinds }); + if (resolution.status === 'resolved') { + if (!seenIds.has(resolution.source.id)) { + seenIds.add(resolution.source.id); + resolved.push(resolution.source); + } + } else if ( + resolution.status === 'not-found' || + resolution.status === 'wrong-kind' + ) { + unresolved.push(value); + } + } + + return { + status: 'resolved', + sources: max != null ? resolved.slice(0, max) : resolved, + unresolved, + }; +} + export function resolveSourceParam( paramValue: string | null | undefined, sources: T[] | undefined, diff --git a/packages/common-utils/src/__tests__/clickhouse.test.ts b/packages/common-utils/src/__tests__/clickhouse.test.ts index dcec597f09..f8e2314dc4 100644 --- a/packages/common-utils/src/__tests__/clickhouse.test.ts +++ b/packages/common-utils/src/__tests__/clickhouse.test.ts @@ -185,6 +185,22 @@ describe('chSqlToAliasMap - alias unit test', () => { expect(res).toEqual(aliasMap); }); + it('NULL literal alias (multi-source padding column)', () => { + const chSqlInput: ChSql = { + sql: 'SELECT Timestamp as "__hdx_timestamp", NULL as "__hdx_duration_ms" FROM {HYPERDX_PARAM_1544803905:Identifier}.{HYPERDX_PARAM_129845054:Identifier} ORDER BY Timestamp DESC LIMIT {HYPERDX_PARAM_49586:Int32}', + params: { + HYPERDX_PARAM_1544803905: 'default', + HYPERDX_PARAM_129845054: 'otel_logs', + HYPERDX_PARAM_49586: 200, + }, + }; + const res = chSqlToAliasMap(chSqlInput); + expect(res).toEqual({ + __hdx_timestamp: 'Timestamp', + __hdx_duration_ms: 'NULL', + }); + }); + it('Normal alias, with brackets', () => { const chSqlInput: ChSql = { sql: "SELECT Timestamp as ts,ResourceAttributes['service.name'] as serviceTest,Body,TimestampTime,ServiceName,TimestampTime FROM {HYPERDX_PARAM_1544803905:Identifier}.{HYPERDX_PARAM_129845054:Identifier} WHERE (TimestampTime >= fromUnixTimestamp64Milli({HYPERDX_PARAM_1456399765:Int64}) AND TimestampTime <= fromUnixTimestamp64Milli({HYPERDX_PARAM_1719057412:Int64})) ORDER BY TimestampTime DESC LIMIT {HYPERDX_PARAM_49586:Int32} OFFSET {HYPERDX_PARAM_48:Int32}", diff --git a/packages/common-utils/src/clickhouse/index.ts b/packages/common-utils/src/clickhouse/index.ts index 4bdd8ba097..1583846e2a 100644 --- a/packages/common-utils/src/clickhouse/index.ts +++ b/packages/common-utils/src/clickhouse/index.ts @@ -1003,6 +1003,10 @@ function selectColumnsToAliasMap( `${column.expr.column.expr.value}['${column.expr.array_index[0].index.value}']` : // normal alias column.expr.column.expr.value; + } else if (column.expr.type === 'null') { + // NULL literal projection (multi-source search pads columns a source + // lacks with `NULL AS "alias"`); the parser emits it without a loc. + aliasMap[column.as] = 'NULL'; } else if (column.expr.loc != null) { aliasMap[column.as] = parsedSql.slice( column.expr.loc.start.offset, diff --git a/packages/common-utils/src/core/__tests__/searchChartConfig.test.ts b/packages/common-utils/src/core/__tests__/searchChartConfig.test.ts index 99d28a5c53..32068819d2 100644 --- a/packages/common-utils/src/core/__tests__/searchChartConfig.test.ts +++ b/packages/common-utils/src/core/__tests__/searchChartConfig.test.ts @@ -1,5 +1,7 @@ import { ALERT_COUNT_DEFAULT_SELECT, + buildMultiSourceSearchConfig, + buildMultiSourceSelect, buildSearchChartConfig, } from '@/core/searchChartConfig'; import { DisplayType, Filter, SourceKind, TSource } from '@/types'; @@ -477,3 +479,110 @@ describe('buildSearchChartConfig', () => { }); }); }); + +describe('buildMultiSourceSelect', () => { + it('projects the canonical aliases from a Log source semantic expressions', () => { + const source = makeLogSource({ + displayedTimestampValueExpression: 'Timestamp', + serviceNameExpression: 'ServiceName', + severityTextExpression: 'SeverityText', + bodyExpression: 'Body', + }); + + expect(buildMultiSourceSelect(source)).toBe( + 'Timestamp AS "__hdx_timestamp", ' + + 'ServiceName AS "__hdx_service", ' + + 'SeverityText AS "__hdx_severity", ' + + 'Body AS "__hdx_body"', + ); + }); + + it('falls back to the first timestamp expression and NULL for missing semantics', () => { + const source = makeLogSource({ + timestampValueExpression: 'TimestampTime, Timestamp', + implicitColumnExpression: undefined, + }); + + expect(buildMultiSourceSelect(source)).toBe( + 'TimestampTime AS "__hdx_timestamp", ' + + 'NULL AS "__hdx_service", ' + + 'NULL AS "__hdx_severity", ' + + 'NULL AS "__hdx_body"', + ); + }); + + it('maps Trace sources onto status/span-name and a milliseconds duration', () => { + const source = makeTraceSource({ + serviceNameExpression: 'ServiceName', + statusCodeExpression: 'StatusCode', + spanNameExpression: 'SpanName', + durationExpression: 'Duration', + durationPrecision: 9, + }); + + expect(buildMultiSourceSelect(source, { includeDuration: true })).toBe( + 'Timestamp AS "__hdx_timestamp", ' + + 'ServiceName AS "__hdx_service", ' + + 'StatusCode AS "__hdx_severity", ' + + 'SpanName AS "__hdx_body", ' + + '(Duration)/1e6 AS "__hdx_duration_ms"', + ); + }); + + it('projects NULL duration for Log sources when duration is included', () => { + const source = makeLogSource({ bodyExpression: 'Body' }); + + expect(buildMultiSourceSelect(source, { includeDuration: true })).toContain( + 'NULL AS "__hdx_duration_ms"', + ); + }); + + it('appends extra columns, projecting NULL where a source lacks the column', () => { + const source = makeLogSource({ bodyExpression: 'Body' }); + + const select = buildMultiSourceSelect(source, { + extraColumns: [ + { name: 'ServiceName', expression: 'ServiceName' }, + { name: 'StatusCode', expression: null }, + ], + }); + + expect(select).toContain('ServiceName AS "ServiceName"'); + expect(select).toContain('NULL AS "StatusCode"'); + }); +}); + +describe('buildMultiSourceSearchConfig', () => { + it('keeps the standard search config assembly but swaps in the canonical SELECT', () => { + const source = makeLogSource({ + bodyExpression: 'Body', + tableFilterExpression: "ServiceName != 'noisy'", + }); + + const config = buildMultiSourceSearchConfig(source, { + where: 'error', + whereLanguage: 'lucene', + orderBy: 'TimestampTime DESC', + }); + + expect(config.select).toBe(buildMultiSourceSelect(source)); + expect(config.from).toEqual(source.from); + expect(config.connection).toBe('conn-1'); + expect(config.where).toBe('error'); + expect(config.whereLanguage).toBe('lucene'); + expect(config.orderBy).toBe('TimestampTime DESC'); + // Source-level behaviors (e.g. tableFilterExpression) still apply. + expect(config.filters).toEqual([ + { type: 'sql', condition: "ServiceName != 'noisy'" }, + ]); + }); + + it('never resolves to defaultTableSelectExpression', () => { + const config = buildMultiSourceSearchConfig(makeTraceSource(), { + where: '', + }); + + expect(config.select).not.toContain('SpanName,'); + expect(config.select).toContain('AS "__hdx_timestamp"'); + }); +}); diff --git a/packages/common-utils/src/core/searchChartConfig.ts b/packages/common-utils/src/core/searchChartConfig.ts index 37144db8ab..04d7dcfdca 100644 --- a/packages/common-utils/src/core/searchChartConfig.ts +++ b/packages/common-utils/src/core/searchChartConfig.ts @@ -1,3 +1,4 @@ +import { getFirstTimestampValueExpression } from '@/core/utils'; import { BuilderChartConfig, DateRange, @@ -185,3 +186,143 @@ export function buildSearchChartConfig( return config; } + +/** + * Canonical result-column aliases used when searching across multiple sources + * at once. Every selected source's SELECT is rewritten to this shape, so the + * merged results table can map columns by name regardless of how each source's + * underlying schema names them. + * + * The names are quoted aliases (`expr AS "__hdx_timestamp"`), so ClickHouse + * returns them verbatim — unlike raw expressions, which CH may reformat. + */ +export const MULTI_SOURCE_ALIASES = { + timestamp: '__hdx_timestamp', + service: '__hdx_service', + severity: '__hdx_severity', + body: '__hdx_body', + /** Milliseconds; only projected when a Trace source is in the selection. */ + durationMs: '__hdx_duration_ms', +} as const; + +/** + * An extra user-picked column to project alongside the canonical aliases. + * `expression` is the per-source SQL expression for the column, or null when + * the source has no such column (projected as NULL so every source returns + * the same column set). + */ +export type MultiSourceExtraColumn = { + /** Result column name (used verbatim as the quoted alias). */ + name: string; + expression: string | null; +}; + +const quoteAlias = (name: string) => `"${name.replace(/"/g, '\\"')}"`; + +/** + * Per-source semantic expression for each canonical alias. Mirrors the app's + * display helpers (`getDisplayedTimestampValueExpression`, `getEventBody`, + * `getDurationMsExpression` in packages/app/src/source.ts) — keep in sync. + */ +function multiSourceSemanticExpressions(source: TSource): { + timestamp: string; + service: string; + severity: string; + body: string; + durationMs: string; +} { + const firstTimestamp = getFirstTimestampValueExpression( + source.timestampValueExpression, + ); + + if (isLogSource(source)) { + return { + timestamp: source.displayedTimestampValueExpression || firstTimestamp, + service: source.serviceNameExpression || 'NULL', + severity: source.severityTextExpression || 'NULL', + body: source.bodyExpression || source.implicitColumnExpression || 'NULL', + durationMs: 'NULL', + }; + } + + if (isTraceSource(source)) { + return { + timestamp: source.displayedTimestampValueExpression || firstTimestamp, + service: source.serviceNameExpression || 'NULL', + severity: source.statusCodeExpression || 'NULL', + body: source.spanNameExpression || 'NULL', + // Match getDurationMsExpression: durationPrecision is the sub-second + // digit count (9 = nanoseconds), so /1e(precision-3) yields milliseconds. + durationMs: `(${source.durationExpression})/1e${(source.durationPrecision ?? 9) - 3}`, + }; + } + + // Multi-source search only supports Log and Trace sources today; other kinds + // still get a valid (if minimal) shape so a stray source can't render SQL + // that errors the whole selection. + return { + timestamp: firstTimestamp, + service: 'NULL', + severity: 'NULL', + body: 'NULL', + durationMs: 'NULL', + }; +} + +/** + * Build the canonical aliased SELECT string for one source in a multi-source + * search. Exported for tests. + */ +export function buildMultiSourceSelect( + source: TSource, + { + includeDuration = false, + extraColumns = [], + }: { + /** Project `__hdx_duration_ms` (set when any selected source is a Trace). */ + includeDuration?: boolean; + extraColumns?: MultiSourceExtraColumn[]; + } = {}, +): string { + const exprs = multiSourceSemanticExpressions(source); + + const parts = [ + `${exprs.timestamp} AS ${quoteAlias(MULTI_SOURCE_ALIASES.timestamp)}`, + `${exprs.service} AS ${quoteAlias(MULTI_SOURCE_ALIASES.service)}`, + `${exprs.severity} AS ${quoteAlias(MULTI_SOURCE_ALIASES.severity)}`, + `${exprs.body} AS ${quoteAlias(MULTI_SOURCE_ALIASES.body)}`, + ]; + if (includeDuration) { + parts.push( + `${exprs.durationMs} AS ${quoteAlias(MULTI_SOURCE_ALIASES.durationMs)}`, + ); + } + for (const col of extraColumns) { + parts.push(`${col.expression ?? 'NULL'} AS ${quoteAlias(col.name)}`); + } + + return parts.join(', '); +} + +/** + * Build the chart config for one source of a multi-source search: the standard + * `buildSearchChartConfig` assembly with the SELECT replaced by the canonical + * aliased column set, so every selected source returns the same result shape. + * + * The caller supplies `orderBy` per source (each source's own timestamp-based + * default) — a shared orderBy is meaningless across schemas, and time-window + * pagination requires the first orderBy term to be the source's timestamp. + */ +export function buildMultiSourceSearchConfig( + source: TSource, + input: Omit, + opts: { + includeDuration?: boolean; + extraColumns?: MultiSourceExtraColumn[]; + } = {}, +): SearchChartConfig { + return buildSearchChartConfig(source, { + ...input, + select: buildMultiSourceSelect(source, opts), + }); +}