Merge pull request #52 from tma1-ai/greptime-read-time-bound

fix(greptime): pre-filter base tables in subqueries so JOINs don't defeat index pushdown

Author: killme2008

packages/shared/src/server/repositories/greptime/dashboards.ts

@@ -1,4 +1,5 @@
 import { type FilterState } from "../../../types";
+import { InvalidRequestError } from "../../../errors";
 import { greptimeQuery } from "../../greptime/client";
 import { createGreptimeFilterFromFilterState } from "../../greptime/sql/factory";
 import { USAGE_COST_KNOWN_KEYS } from "../../greptime/sql/fragments";
@@ -111,7 +112,20 @@ export const getScoreAggregateGreptime = async (
       dashboardGreptimeColumnDefinitions,
     ),
   );
-  const restRes = restList.apply();
+  // This aggregation reads `scores s` (optionally JOIN traces); it has no `observations o` alias.
+  // Observation-scoped dashboard columns (level / token / cost / tool-name filters) compile to
+  // predicates correlated to `o`, so they cannot be expressed here -- reject them loudly instead of
+  // emitting SQL that references a non-existent alias.
+  if (restList.some((f) => f.table === "observations")) {
+    throw new InvalidRequestError(
+      "Observation-scoped filters are not supported for score aggregation on GreptimeDB",
+    );
+  }
+  // Split by physical table so each side pre-filters inside its own subquery: a flat `scores JOIN
+  // traces` defeats GreptimeDB pushdown of the scores TIME INDEX / trace_id bloom and the traces TIME
+  // INDEX to the region scans.
+  const scoreRestRes = restList.filter((f) => f.table !== "traces").apply();
+  const traceRestRes = restList.filter((f) => f.table === "traces").apply();
 
   const hasTraceFilter = restList.some((f) => f.table === "traces");
   const timeFilter = restList.find(
@@ -122,7 +136,8 @@ export const getScoreAggregateGreptime = async (
   const useLookback = Boolean(timeFilter && hasTraceFilter);
   const params: Record<string, unknown> = {
     projectId,
-    ...restRes.params,
+    ...scoreRestRes.params,
+    ...traceRestRes.params,
     ...env.params,
   };
   if (useLookback && timeFilter) {
@@ -143,12 +158,22 @@ export const getScoreAggregateGreptime = async (
     query: `
       SELECT s.name AS name, count(*) AS count, avg(s.value) AS avg_value,
         s.source AS source, s.data_type AS data_type
-      FROM scores s
-      ${hasTraceFilter ? "JOIN traces t ON t.id = s.trace_id AND t.project_id = s.project_id AND " + notDeleted("t") : ""}
-      WHERE s.project_id = :projectId AND ${notDeleted("s")}
-        ${restRes.query ? `AND ${restRes.query}` : ""}
-        ${env.query ? `AND ${env.query}` : ""}
-        ${useLookback ? "AND t.timestamp >= :tracesTimestamp" : ""}
+      FROM (
+        SELECT * FROM scores s
+        WHERE s.project_id = :projectId AND ${notDeleted("s")}
+          ${scoreRestRes.query ? `AND ${scoreRestRes.query}` : ""}
+          ${env.query ? `AND ${env.query}` : ""}
+      ) s
+      ${
+        hasTraceFilter
+          ? `JOIN (
+        SELECT * FROM traces t
+        WHERE t.project_id = :projectId AND ${notDeleted("t")}
+          ${traceRestRes.query ? `AND ${traceRestRes.query}` : ""}
+          ${useLookback ? "AND t.timestamp >= :tracesTimestamp" : ""}
+      ) t ON t.id = s.trace_id AND t.project_id = s.project_id`
+          : ""
+      }
       GROUP BY s.name, s.source, s.data_type
       ORDER BY count(*) DESC`,
     params,
@@ -188,7 +213,6 @@ const getObservationDetailByTypeByTime = async (opts: {
       dashboardGreptimeColumnDefinitions,
     ),
   );
-  const restRes = restList.apply();
 
   const hasTraceFilter = restList.some((f) => f.table === "traces");
   // CH derived the trace lookback from an observation start_time lower bound, only when a trace
@@ -212,11 +236,20 @@ const getObservationDetailByTypeByTime = async (opts: {
     Boolean(env.query) ||
     restList.some((f) => f.table === "observations");
 
+  // Split predicates by physical table so each base table pre-filters inside its own subquery: a flat
+  // `uc JOIN observations JOIN traces ... WHERE` defeats GreptimeDB pushdown of the uc/observations TIME
+  // INDEX window to the region scans. Score-grain predicates correlate to `o`, so they stay obs-side.
+  const obsRestRes = restList.filter((f) => f.table !== "traces").apply();
+  const traceRestRes = restList.filter((f) => f.table === "traces").apply();
+  const obsRestClause = obsRestRes.query ? `AND ${obsRestRes.query}` : "";
+  const traceRestClause = traceRestRes.query ? `AND ${traceRestRes.query}` : "";
+
   const params: Record<string, unknown> = {
     projectId,
     winFrom: greptimeTsParam(new Date(fromTime)),
     winTo: greptimeTsParam(new Date(toTime)),
-    ...restRes.params,
+    ...obsRestRes.params,
+    ...traceRestRes.params,
     ...env.params,
   };
   if (useLookback && obsStartLowerBound) {
@@ -231,20 +264,30 @@ const getObservationDetailByTypeByTime = async (opts: {
   const kind = jsonColumn === "cost_details" ? "cost" : "usage";
   params.kind = kind;
 
-  const traceJoin = hasTraceFilter
-    ? `LEFT JOIN traces t ON o.trace_id = t.id AND o.project_id = t.project_id AND ${notDeleted("t")}`
-    : "";
-  // Conditional in Q2 (see needsObservationJoin); Q1 always reads observations directly so it keeps
-  // its own FROM.
-  const obsJoin = needsObservationJoin
-    ? `JOIN observations o ON uc.entity_id = o.id AND uc.project_id = o.project_id AND ${notDeleted("o")}`
-    : "";
-  const restClause = restRes.query ? `AND ${restRes.query}` : "";
   const envClause = env.query ? `AND ${env.query}` : "";
   const lookbackClause = useLookback
     ? "AND t.timestamp >= :traceTimestamp"
     : "";
 
+  // Pre-filtered traces subquery. INNER join: a trace filter means the row must match a passing trace
+  // -- the same effect the old `LEFT JOIN traces ... WHERE t.<pred>` produced.
+  const tracesJoinSql = hasTraceFilter
+    ? `JOIN (
+        SELECT * FROM traces t
+        WHERE t.project_id = :projectId AND ${notDeleted("t")} ${traceRestClause} ${lookbackClause}
+      ) t ON o.trace_id = t.id AND o.project_id = t.project_id`
+    : "";
+  // Pre-filtered observations subquery for Q2, bounded by the same window (the EAV row's timestamp is
+  // the observation start_time) so the scan prunes via the TIME INDEX before the EAV join.
+  const obsJoinSql = needsObservationJoin
+    ? `JOIN (
+        SELECT * FROM observations o
+        WHERE o.project_id = :projectId AND ${notDeleted("o")}
+          AND o.start_time >= :winFrom AND o.start_time < :winTo
+          ${obsRestClause} ${envClause}
+      ) o ON uc.entity_id = o.id AND uc.project_id = o.project_id`
+    : "";
+
   const byBucket = new Map<number, Record<string, number>>();
   const allKeys = new Set<string>();
   const setSum = (bucketMsKey: number, key: string, value: number) => {
@@ -269,13 +312,14 @@ const getObservationDetailByTypeByTime = async (opts: {
       query: `
       SELECT date_bin(INTERVAL '${bucketSizeSeconds}' second, o.start_time) AS bucket,
         ${knownSums}
-      FROM observations o
-      ${traceJoin}
-      WHERE o.project_id = :projectId AND ${notDeleted("o")}
-        AND o.start_time >= :winFrom AND o.start_time < :winTo
-        ${restClause}
-        ${envClause}
-        ${lookbackClause}
+      FROM (
+        SELECT * FROM observations o
+        WHERE o.project_id = :projectId AND ${notDeleted("o")}
+          AND o.start_time >= :winFrom AND o.start_time < :winTo
+          ${obsRestClause}
+          ${envClause}
+      ) o
+      ${tracesJoinSql}
       GROUP BY bucket
       ORDER BY bucket ASC`,
       params,
@@ -295,16 +339,15 @@ const getObservationDetailByTypeByTime = async (opts: {
       SELECT date_bin(INTERVAL '${bucketSizeSeconds}' second, uc.${quoteIdent("timestamp")}) AS bucket,
         uc.${quoteIdent("key")} AS detail_key,
         sum(uc.${quoteIdent("value")}) AS sum
-      FROM observations_usage_cost uc
-      ${obsJoin}
-      ${traceJoin}
-      WHERE uc.${quoteIdent("kind")} = :kind
-        AND uc.project_id = :projectId AND ${notDeleted("uc")}
-        AND uc.${quoteIdent("timestamp")} >= :winFrom AND uc.${quoteIdent("timestamp")} < :winTo
-        AND uc.${quoteIdent("key")} NOT IN (${KNOWN_DETAIL_KEYS_SQL})
-        ${restClause}
-        ${envClause}
-        ${lookbackClause}
+      FROM (
+        SELECT * FROM observations_usage_cost uc
+        WHERE uc.${quoteIdent("kind")} = :kind
+          AND uc.project_id = :projectId AND ${notDeleted("uc")}
+          AND uc.${quoteIdent("timestamp")} >= :winFrom AND uc.${quoteIdent("timestamp")} < :winTo
+          AND uc.${quoteIdent("key")} NOT IN (${KNOWN_DETAIL_KEYS_SQL})
+      ) uc
+      ${obsJoinSql}
+      ${tracesJoinSql}
       GROUP BY bucket, uc.${quoteIdent("key")}
       ORDER BY bucket ASC`,
       params,

packages/shared/src/server/repositories/greptime/eventsObservations.ts

@@ -15,8 +15,8 @@ import {
   buildObservationsTableQuery,
   getObservationsTableCountGreptime,
   getObservationsTableRowsGreptime,
+  observationsScopedFrom,
 } from "./observationsTable";
-import { notDeleted } from "./queryHelpers";
 
 /**
  * GreptimeDB obs-from-events collapse (04-read-path.md, P5 Piece D).
@@ -78,11 +78,7 @@ const resolvePositionQualifyingIds = async (
             PARTITION BY o.trace_id
             ORDER BY o.start_time ${direction}, o.id ${direction}
           ) AS rn
-        FROM observations o
-        ${compiled.traceJoin ? "LEFT JOIN traces t ON t.id = o.trace_id AND t.project_id = o.project_id AND " + notDeleted("t") : ""}
-        WHERE ${compiled.whereSql} AND ${notDeleted("o")}
-          ${compiled.lookback ? "AND t.timestamp >= :obsTraceLookback" : ""}
-          ${search.query}
+        FROM ${observationsScopedFrom(compiled, search)}
       )
       SELECT id FROM qualifying WHERE rn = :rank`,
     params: {

packages/shared/src/server/repositories/greptime/observationsTable.ts

@@ -71,10 +71,16 @@ export type GreptimeObservationsTableProps = {
   searchType?: Parameters<typeof greptimeSearchCondition>[0]["searchType"];
 };
 
-type CompiledObs = {
-  whereSql: string;
+export type CompiledObs = {
+  // observations-scoped predicates (o.* + score-grain EXISTS, which correlate to o.id).
+  obsWhereSql: string;
+  // trace-scoped predicates (t.*), applied inside the traces subquery.
+  traceWhereSql: string;
   params: Record<string, unknown>;
   traceJoin: boolean;
+  // A trace FILTER (or the trace lookback) means an observation must match a passing/recent trace, so
+  // the join must be INNER. A join that exists only to order by a trace column stays LEFT.
+  innerTraceJoin: boolean;
   lookback?: string;
 };
 
@@ -97,8 +103,15 @@ export const buildObservationsTableQuery = (
     ),
   ]);
 
+  // Split by physical table so each side can be pre-filtered inside its own subquery (a flat
+  // `observations o LEFT JOIN traces t ... WHERE` defeats GreptimeDB index pushdown on the driving
+  // observations scan). Score-grain filters carry table "scores" but render as EXISTS correlated to
+  // `o.id`, so they stay on the observations side.
+  const obsFilters = filters.filter((f) => f.table !== "traces");
+  const traceFilters = filters.filter((f) => f.table === "traces");
+
   const traceJoin =
-    filters.some((f) => f.table === "traces") ||
+    traceFilters.length() > 0 ||
     findUiColumnMapping(observationsTableMapping, orderBy?.column)
       ?.greptimeTableName === "traces";
 
@@ -117,15 +130,39 @@ export const buildObservationsTableQuery = (
         )
       : undefined;
 
-  const applied = filters.apply();
+  const obsRes = obsFilters.apply();
+  const traceRes = traceFilters.apply();
   return {
-    whereSql: applied.query,
-    params: applied.params,
+    obsWhereSql: obsRes.query,
+    traceWhereSql: traceRes.query,
+    // `projectId` binds the `t.project_id = :projectId` scope in the traces subquery (the o-side
+    // project filter above carries its own auto-generated placeholder, not :projectId).
+    params: { projectId, ...obsRes.params, ...traceRes.params },
     traceJoin,
+    innerTraceJoin: traceFilters.length() > 0 || lookback !== undefined,
     lookback,
   };
 };
 
+// Build the pre-filtered FROM clause (observations subquery [join traces subquery]) so each base table
+// scan prunes before the join. `SELECT *` is an intentional intermediate relation (projection pushdown
+// drops unreferenced columns; the outer query wraps JSON columns via its explicit select list).
+export const observationsScopedFrom = (
+  compiled: CompiledObs,
+  search: { query: string },
+): string => {
+  const obsSub = `(SELECT * FROM observations o
+      WHERE ${compiled.obsWhereSql} AND ${notDeleted("o")} ${search.query}) o`;
+  if (!compiled.traceJoin) return obsSub;
+  const traceSub = `(SELECT * FROM traces t
+      WHERE t.project_id = :projectId AND ${notDeleted("t")}
+        ${compiled.traceWhereSql ? `AND ${compiled.traceWhereSql}` : ""}
+        ${compiled.lookback ? "AND t.timestamp >= :obsTraceLookback" : ""}) t`;
+  const joinKind = compiled.innerTraceJoin ? "JOIN" : "LEFT JOIN";
+  return `${obsSub}
+    ${joinKind} ${traceSub} ON t.id = o.trace_id AND t.project_id = o.project_id`;
+};
+
 const observationsOrderBy = (orderBy?: OrderByState): string => {
   const primary: OrderByState = orderBy ?? {
     column: "startTime",
@@ -149,11 +186,7 @@ export const getObservationsTableCountGreptime = async (
   const rows = await greptimeQuery<{ count: string | number }>({
     query: `
       SELECT count(*) AS count
-      FROM observations o
-      ${compiled.traceJoin ? "LEFT JOIN traces t ON t.id = o.trace_id AND t.project_id = o.project_id AND " + notDeleted("t") : ""}
-      WHERE ${compiled.whereSql} AND ${notDeleted("o")}
-        ${compiled.lookback ? "AND t.timestamp >= :obsTraceLookback" : ""}
-        ${search.query}`,
+      FROM ${observationsScopedFrom(compiled, search)}`,
     params: {
       ...compiled.params,
       ...search.params,
@@ -178,11 +211,7 @@ export const getObservationsTableRowsGreptime = async (
   const rows = await greptimeQuery<Record<string, unknown>>({
     query: `
       SELECT ${greptimeObservationSelect({ prefix: "o", excludeIo: exclude, excludeMetadata: exclude })}
-      FROM observations o
-      ${compiled.traceJoin ? "LEFT JOIN traces t ON t.id = o.trace_id AND t.project_id = o.project_id AND " + notDeleted("t") : ""}
-      WHERE ${compiled.whereSql} AND ${notDeleted("o")}
-        ${compiled.lookback ? "AND t.timestamp >= :obsTraceLookback" : ""}
-        ${search.query}
+      FROM ${observationsScopedFrom(compiled, search)}
       ${observationsOrderBy(props.orderBy)}
       ${props.limit !== undefined && props.offset !== undefined ? "LIMIT :limit OFFSET :offset" : ""}`,
     params: {

packages/shared/src/server/repositories/greptime/queryHelpers.ts

@@ -1,5 +1,6 @@
 import { quoteIdent } from "../../greptime/schemaUtils";
 import { greptimeTimestampLiteral } from "../../greptime/sql/greptime-filter";
+import { greptimeQuery } from "../../greptime/client";
 
 /**
  * Shared SQL fragments for the GreptimeDB read repositories (04-read-path.md, P1).
@@ -48,6 +49,41 @@ const quoteColumnRef = (ref: string): string => {
 /** Bind a Date as a ms-precision GreptimeDB timestamp literal (string -> TIMESTAMP coercion). */
 export const greptimeTsParam = (d: Date): string => greptimeTimestampLiteral(d);
 
+/**
+ * Earliest trace `timestamp` for a finite set of session/user ids, scoped by `project_id` and pruned
+ * by the bloom skipping index on `scopeColumn` (04-read-path.md, migration 0006). Returns null when the
+ * set has no live traces.
+ *
+ * Why: the all-time metrics reads (`getSessionsWithMetricsGreptime` with an id-only filter,
+ * `getUserMetrics` with an empty filter) pass no UI timestamp bound. After the join-pushdown fix the
+ * `traces` side prunes via the `session_id`/`user_id` bloom index, but the `observations` side then has
+ * no index-eligible predicate. Feeding `min(timestamp) - INTERVAL` as an `observations.start_time` lower
+ * bound restores TIME-INDEX pruning there.
+ *
+ * IMPORTANT — this is a DELIBERATE lookback-bounded narrowing, NOT strict all-time equivalence: it drops
+ * observations whose `start_time` precedes the group's earliest trace by more than the caller's INTERVAL
+ * (pathological clock skew > INTERVAL or a back-dated import; for sane data the dropped set is empty).
+ * It mirrors the exact heuristic the windowed metrics path already applies (`obsLookback = tsFilter -
+ * INTERVAL`). Callers subtract their own INTERVAL and must document the trade-off.
+ */
+export const deriveTraceMinTimestamp = async (
+  projectId: string,
+  scopeColumn: "session_id" | "user_id",
+  ids: readonly string[],
+): Promise<Date | null> => {
+  if (ids.length === 0) return null;
+  const inClause = greptimeInClause(scopeColumn, ids, "scope");
+  const rows = await greptimeQuery<{ min_ts: Date | string | null }>({
+    query: `SELECT min(${quoteIdent("timestamp")}) AS min_ts FROM traces
+      WHERE ${quoteIdent("project_id")} = :projectId AND ${inClause.sql} AND ${notDeleted()}`,
+    params: { projectId, ...inClause.params },
+    readOnly: true,
+  });
+  const v = rows[0]?.min_ts;
+  if (v == null) return null;
+  return v instanceof Date ? v : new Date(v);
+};
+
 /** UTC calendar-day bounds [start, end) for a same-day match, as ms-precision literals. */
 export const greptimeDayBounds = (d: Date): { start: string; end: string } => {
   const start = new Date(