kodiakhq · Sep 1, 2026 · Aug 31, 2026 · Sep 1, 2026 · Sep 1, 2026
@@ -0,0 +1,6 @@
+---
+'@hyperdx/api': patch
+'@hyperdx/app': patch
+---
+
+feat: Accept optional time bounds on Prometheus label values endpoint
@@ -0,0 +1,244 @@
+import { ClickhouseClient } from '@/clickhouse';
+import * as config from '@/config';
+import { queryLabelValues } from '@/controllers/timeseriesEngine';
+import {
+  closeTestFixtureClickHouseClient,
+  DEFAULT_DATABASE,
+  dropTimeSeriesTable,
+  executeSqlCommand,
+  seedTimeSeriesTagsTable,
+} from '@/fixtures';
+
+const CONNECTION_ID = 'timeseries-engine-int-test';
+const BOUNDED_TABLE = 'test_ts_label_values';
+const UNBOUNDED_TABLE = 'test_ts_label_values_no_bounds';
+
+const HOUR_SEC = 3600;
+const OLD_START_SEC = 1600000000;
+const RECENT_START_SEC = 1700000000;
+const sec = (s: number) => s * 1000;
+
+// Three series: one only in the distant past, two only in the recent window
+// sharing a metric name (so DISTINCT has something to collapse).
+const SERIES = [
+  {
+    metricName: 'old_metric',
+    tags: { job: 'batch' },
+    startSec: OLD_START_SEC,
+    endSec: OLD_START_SEC + HOUR_SEC,
+  },
+  {
+    metricName: 'recent_metric',
+    tags: { job: 'api' },
+    startSec: RECENT_START_SEC,
+    endSec: RECENT_START_SEC + HOUR_SEC,
+  },
+  {
+    metricName: 'recent_metric',
+    tags: { job: 'web' },
+    startSec: RECENT_START_SEC,
+    endSec: RECENT_START_SEC + HOUR_SEC,
+  },
+];
+
+describe('timeseriesEngine controller', () => {
+  let client: ClickhouseClient;
+
+  const labelValues = (
+    args: Partial<Parameters<typeof queryLabelValues>[0]> & {
+      labelName: string;
+    },
+  ) =>
+    queryLabelValues({
+      client,
+      connectionId: CONNECTION_ID,
+      databaseName: DEFAULT_DATABASE,
+      tableName: BOUNDED_TABLE,
+      ...args,
+    });
+
+  beforeAll(async () => {
+    client = new ClickhouseClient({
+      host: config.CLICKHOUSE_HOST,
+      username: config.CLICKHOUSE_USER,
+      password: config.CLICKHOUSE_PASSWORD,
+    });
+    await seedTimeSeriesTagsTable({ table: BOUNDED_TABLE, series: SERIES });
+    await seedTimeSeriesTagsTable({
+      table: UNBOUNDED_TABLE,
+      series: SERIES,
+      storeTimeBounds: false,
+    });
+  });
+
+  afterAll(async () => {
+    for (const table of [BOUNDED_TABLE, UNBOUNDED_TABLE]) {
+      await dropTimeSeriesTable({ table });
+    }
+    await client.close();
+    await closeTestFixtureClickHouseClient();
+  });
+
+  describe('queryLabelValues', () => {
+    describe('__name__', () => {
+      it('returns every metric name, deduplicated and sorted', async () => {
+        expect(await labelValues({ labelName: '__name__' })).toEqual([
+          'old_metric',
+          'recent_metric',
+        ]);
+      });
+
+      it('keeps only metrics whose series overlap the window', async () => {
+        expect(
+          await labelValues({
+            labelName: '__name__',
+            startMs: sec(RECENT_START_SEC),
+            endMs: sec(RECENT_START_SEC + HOUR_SEC),
+          }),
+        ).toEqual(['recent_metric']);
+
+        expect(
+          await labelValues({
+            labelName: '__name__',
+            startMs: sec(OLD_START_SEC),
+            endMs: sec(OLD_START_SEC + HOUR_SEC),
+          }),
+        ).toEqual(['old_metric']);
+      });
+
+      // A lone `end` bounds min_time (nothing may start after the window) and a
+      // lone `start` bounds max_time (nothing may end before it). Getting the
+      // pairing backwards still filters, just wrongly, so both directions are
+      // pinned.
+      it('applies a lone end bound against min_time', async () => {
+        expect(
+          await labelValues({
+            labelName: '__name__',
+            endMs: sec(OLD_START_SEC + HOUR_SEC),
+          }),
+        ).toEqual(['old_metric']);
+      });
+
+      it('applies a lone start bound against max_time', async () => {
+        expect(
+          await labelValues({
+            labelName: '__name__',
+            startMs: sec(RECENT_START_SEC),
+          }),
+        ).toEqual(['recent_metric']);
+      });
+
+      it('returns nothing for a window no series covers', async () => {
+        expect(
+          await labelValues({
+            labelName: '__name__',
+            startMs: sec(OLD_START_SEC + 2 * HOUR_SEC),
+            endMs: sec(OLD_START_SEC + 3 * HOUR_SEC),
+          }),
+        ).toEqual([]);
+      });
+    });
+
+    describe('tag labels', () => {
+      it('returns the label values, deduplicated and sorted', async () => {
+        expect(await labelValues({ labelName: 'job' })).toEqual([
+          'api',
+          'batch',
+          'web',
+        ]);
+      });
+
+      it('scopes label values to the window', async () => {
+        expect(
+          await labelValues({
+            labelName: 'job',
+            startMs: sec(RECENT_START_SEC),
+            endMs: sec(RECENT_START_SEC + HOUR_SEC),
+          }),
+        ).toEqual(['api', 'web']);
+      });
+
+      it('returns nothing for a label no series carries', async () => {
+        expect(await labelValues({ labelName: 'instance' })).toEqual([]);
+      });
+
+      // The label name reaches both the WHERE clause and the value expression.
+      // Parameterized, this is an ordinary miss; interpolated, it is a syntax
+      // error — so the assertion is "empty", not "throws".
+      it('parameterizes the label name rather than interpolating it', async () => {
+        expect(await labelValues({ labelName: `job') OR 1=1 --` })).toEqual([]);
+      });
+    });
+
+    describe('limit', () => {
+      it('caps the number of values returned, keeping the ordering', async () => {
+        expect(await labelValues({ labelName: '__name__', limit: 1 })).toEqual([
+          'old_metric',
+        ]);
+      });
+
+      // Prometheus reads limit=0 as unlimited, so it must not become `LIMIT 0`.
+      it('treats a zero limit as unlimited', async () => {
+        expect(await labelValues({ labelName: '__name__', limit: 0 })).toEqual([
+          'old_metric',
+          'recent_metric',
+        ]);
+      });
+
+      it('returns everything when the limit exceeds the result set', async () => {
+        expect(
+          await labelValues({ labelName: '__name__', limit: 100 }),
+        ).toEqual(['old_metric', 'recent_metric']);
+      });
+    });
+
+    // store_min_time_and_max_time = 0 leaves the tags table without the columns
+    // the predicate reads, and referencing a missing column is a hard error.
+    // Time bounds are best-effort narrowing, so the request widens instead.
+    describe('table without min_time/max_time', () => {
+      it('ignores the bounds instead of failing', async () => {
+        expect(
+          await queryLabelValues({
+            client,
+            connectionId: CONNECTION_ID,
+            databaseName: DEFAULT_DATABASE,
+            tableName: UNBOUNDED_TABLE,
+            labelName: '__name__',
+            startMs: sec(RECENT_START_SEC),
+            endMs: sec(RECENT_START_SEC + HOUR_SEC),
+          }),
+        ).toEqual(['old_metric', 'recent_metric']);
+      });
+
+      it('still filters on the label itself', async () => {
+        expect(
+          await queryLabelValues({
+            client,
+            connectionId: CONNECTION_ID,
+            databaseName: DEFAULT_DATABASE,
+            tableName: UNBOUNDED_TABLE,
+            labelName: 'job',
+          }),
+        ).toEqual(['api', 'batch', 'web']);
+      });
+    });
+
+    it('rejects when the table is not a TimeSeries table', async () => {
+      await executeSqlCommand(
+        `CREATE OR REPLACE TABLE ${DEFAULT_DATABASE}.test_ts_plain (ts DateTime) ENGINE = MergeTree ORDER BY ts`,
+      );
+      try {
+        await expect(
+          labelValues({
+            labelName: '__name__',
+            tableName: 'test_ts_plain',
+          }),
+        ).rejects.toThrow(/TimeSeries table only/);
+      } finally {
+        await executeSqlCommand(
+          `DROP TABLE IF EXISTS ${DEFAULT_DATABASE}.test_ts_plain`,
+        );
+      }
+    });
+  });
+});
@@ -0,0 +1,137 @@
+import {
+  ChSql,
+  chSql,
+  concatChSql,
+} from '@hyperdx/common-utils/dist/clickhouse';
+import {
+  Metadata,
+  MetadataCache,
+} from '@hyperdx/common-utils/dist/core/metadata';
+
+import { ClickhouseClient } from '@/clickhouse';
+import logger from '@/utils/logger';
+
+export const PROMETHEUS_MAX_EXECUTION_SEC = 30;
+export const PROMETHEUS_MAX_RESULT_ROWS = 100000;
+
+/**
+ * Indicates whether the tags inner table associated with the given TimeSeries table
+ * includes the optional `min_time` and `max_time` columns.
+ */
+async function timeSeriesTagsTableHasTimeBounds({
+  metadata,
+  connectionId,
+  database,
+  table,
+}: {
+  metadata: Metadata;
+  connectionId: string;
+  database: string;
+  table: string;
+}): Promise<boolean> {
+  try {
+    const columns = await metadata.getTimeSeriesTableColumns({
+      connectionId,
+      databaseName: database,
+      tableName: table,
+      innerTableType: 'Tags',
+    });
+
+    const columnNames = new Set(columns.map(column => column.name));
+    return columnNames.has('min_time') && columnNames.has('max_time');
+  } catch (e) {
+    logger.warn(
+      { err: e, database, table },
+      'Failed to check if TimeSeries tags table has time bounds columns',
+    );
+    return false;
+  }
+}
+
+/**
+ * Queries distinct values for the given label name from the given TimeSeries table,
+ * optionally filtering by a time range (if the table has the optional time range columns).
+ */
+export async function queryLabelValues({
+  client,
+  databaseName,
+  connectionId,
+  tableName,
+  labelName,
+  startMs,
+  endMs,
+  limit,
+}: {
+  client: ClickhouseClient;
+  connectionId: string;
+  databaseName: string;
+  tableName: string;
+  labelName: string;
+  startMs?: number;
+  endMs?: number;
+  limit?: number;
+}): Promise<string[]> {
+  const isMetricName = labelName === '__name__';
+  const value = isMetricName
+    ? chSql`${{ Identifier: 'metric_name' }} `
+    : chSql`${{ Identifier: 'tags' }}[${{ String: labelName }}]`;
+
+  const conditions: ChSql[] = [];
+  if (!isMetricName)
+    conditions.push(
+      chSql`mapContains(${{ Identifier: 'tags' }}, ${{ String: labelName }})`,
+    );
+
+  const metadata = new Metadata(client, new MetadataCache());
+  const hasStartTime = startMs != null;
+  const hasEndTime = endMs != null;
+
+  // min and max time columns are optional in the tags inner table,
+  // so we check if they exist before adding time conditions
+  const tableHasTimeBounds =
+    (hasStartTime || hasEndTime) && // Short-circuit if no time conditions are needed
+    (await timeSeriesTagsTableHasTimeBounds({
+      metadata,
+      connectionId,
+      database: databaseName,
+      table: tableName,
+    }));
+
+  if (tableHasTimeBounds && endMs != null)
+    conditions.push(
+      chSql`(min_time IS NULL OR min_time <= fromUnixTimestamp64Milli(${{ Int64: endMs }}))`,
+    );
+
+  if (tableHasTimeBounds && startMs != null)
+    conditions.push(
+      chSql`(max_time IS NULL OR max_time >= fromUnixTimestamp64Milli(${{ Int64: startMs }}))`,
+    );
+
+  const where = conditions.length
+    ? chSql`WHERE ${concatChSql(' AND ', ...conditions)}`
+    : chSql``;
+  const limitSql = limit
+    ? chSql`LIMIT ${{ Int32: Math.min(limit, PROMETHEUS_MAX_RESULT_ROWS) }}`
+    : chSql``;
+  const query = chSql`
+    SELECT DISTINCT ${value} AS val
+    FROM timeSeriesTags(${{ Identifier: databaseName }}, ${{ Identifier: tableName }})
+    ${where}
+    ORDER BY val
+    ${limitSql}
+  `;
+
+  const resp = await client.query({
+    query: query.sql,
+    query_params: query.params,
+    format: 'JSON',
+    clickhouse_settings: {
+      allow_experimental_time_series_table: 1,
+      max_execution_time: PROMETHEUS_MAX_EXECUTION_SEC,
+      max_result_rows: String(PROMETHEUS_MAX_RESULT_ROWS),
+    },
+  });
+
+  const json = await resp.json<{ val: string }>();
+  return json.data.map(r => r.val);
+}