perf: show the most frequent text index values, in batches of 10 keys

.changeset/topk-text-index-values.md

@@ -0,0 +1,13 @@
+---
+'@hyperdx/common-utils': patch
+---
+
+perf: show the most frequent text index values, in batches of 10 keys
+
+"All values" lookups that read a text index (Map key=value indexes and text
+indexes on plain columns) used `groupUniqArray(limit)`, which keeps the
+alphabetically first tokens it reads. They now use `topKWeighted(limit)`
+weighted by each token's row count, so a capped list holds the most common
+values, most common first. These lookups also run in parallel batches of at
+most 10 keys instead of 100, so a key with very many values delays only its own
+batch.

packages/common-utils/src/__tests__/metadata.int.test.ts

@@ -907,6 +907,33 @@ describe('Metadata Integration Tests', () => {
       );
     });
 
+    it('keeps the most frequent text index values when maxValuesPerKey truncates the list', async () => {
+      if (!textIndexSupported) {
+        console.warn(
+          'Skipping: ClickHouse < 26.3 does not support mergeTreeTextIndex()',
+        );
+        return;
+      }
+      // The index dictionary is sorted, so 'a-rare' is read first.
+      const freqKey = `freq.${tag}`;
+      const timestamp = testTime.toISOString().replace('T', ' ').slice(0, 23);
+      await client.command({
+        query: `INSERT INTO default.otel_logs (Timestamp, ResourceAttributes)
+          SELECT '${timestamp}', map('${freqKey}', if(number = 0, 'a-rare', 'z-common'))
+          FROM numbers(4)`,
+      });
+
+      const result = await metadata.getAllKeyValues({
+        ...commonArgs,
+        keyExpressions: [`ResourceAttributes['${freqKey}']`],
+        maxValuesPerKey: 1,
+      });
+
+      expect(result).toEqual([
+        { key: `ResourceAttributes['${freqKey}']`, value: ['z-common'] },
+      ]);
+    });
+
     it('routes TraceId through getTextIndexKeyValues', async () => {
       if (!textIndexSupported) {
         console.warn(

packages/common-utils/src/__tests__/metadata.test.ts

@@ -1375,7 +1375,7 @@ describe('Metadata', () => {
       const sql = calls[calls.length - 1][0].query as string;
 
       expect(sql).toContain('mergeTreeTextIndex(');
-      expect(sql).toContain('groupUniqArray(');
+      expect(sql).toContain('topKWeighted(');
       expect(sql).toContain('GROUP BY key');
       expect(sql).not.toContain('startsWith(token,');
     });
@@ -1398,6 +1398,24 @@ describe('Metadata', () => {
       expect(sql).toContain('GROUP BY column, key');
     });
 
+    it('splits Map text index values into queries of at most 10 keys', async () => {
+      setupDefaultLogsSchema();
+
+      await metadata.getAllKeyValues({
+        ...baseArgs,
+        keyExpressions: Array.from(
+          { length: 25 },
+          (_, i) => `LogAttributes['key${i}']`,
+        ),
+      });
+
+      const keysPerQuery = (mockClickhouseClient.query as jest.Mock).mock.calls
+        .map(([opts]) => opts.query as string)
+        .filter(q => q.includes('startsWith(token,'))
+        .map(q => q.split('startsWith(token,').length - 1);
+      expect(keysPerQuery).toEqual([10, 10, 5]);
+    });
+
     it('routes ServiceName and SeverityText through getMetadataMVKeyValues (KV rollup MV path)', async () => {
       setupDefaultLogsSchema();
 
@@ -1718,9 +1736,8 @@ describe('Metadata', () => {
     // via text index or raw scan.
     describe('maxValuesPerKey threading', () => {
       // Pick a value that (a) isn't ambient in the test setup (like 20,
-      // dates, or granularity numbers) and (b) is easy to spot in a param
-      // dump when a test fails. Int32 chSql params render as numbers, so
-      // assertions compare against the numeric literal.
+      // dates, or granularity numbers) and (b) is easy to spot in the SQL or a
+      // param dump when a test fails.
       const MAX_VALUES = 7777;
 
       it('threads maxValuesPerKey into the map text-index query', async () => {
@@ -1740,11 +1757,9 @@ describe('Metadata', () => {
             c[0].query.includes('startsWith(token,'),
         );
         expect(mapTextIndexCall).toBeDefined();
-        const params = mapTextIndexCall![0].query_params as Record<
-          string,
-          unknown
-        >;
-        expect(Object.values(params)).toContain(MAX_VALUES);
+        expect(mapTextIndexCall![0].query).toContain(
+          `topKWeighted(${MAX_VALUES})(`,
+        );
       });
 
       it('threads maxValuesPerKey into the native text-index query', async () => {
@@ -1765,11 +1780,9 @@ describe('Metadata', () => {
             !c[0].query.includes('startsWith(token,'),
         );
         expect(nativeTextIndexCall).toBeDefined();
-        const params = nativeTextIndexCall![0].query_params as Record<
-          string,
-          unknown
-        >;
-        expect(Object.values(params)).toContain(MAX_VALUES);
+        expect(nativeTextIndexCall![0].query).toContain(
+          `topKWeighted(${MAX_VALUES})(token, cardinality)`,
+        );
       });
 
       it('threads maxValuesPerKey into the raw-table getKeyValues fallback', async () => {

packages/common-utils/src/core/metadata.ts

@@ -69,6 +69,9 @@ const DEFAULT_MAX_KEYS = 1000;
 // Cap keys per dispatched query: each key is another operation for the db to fetch, and simply fetching all keys at once can be too much for the db to handle.
 export const GET_ALL_KEY_VALUES_CHUNK_SIZE = 100;
 
+// Keys per text index values query; see dispatchKeyValueQueries.
+const TEXT_INDEX_KEYS_PER_QUERY = 10;
+
 /**
  * The index read could not be scoped to the requested date range, so it was
  * refused in favour of a path that can. Not a failure — an expected property
@@ -1734,11 +1737,13 @@ export class Metadata {
             timestampValueExpression,
           });
           const valueSql = chSql`substring(token, position(token, ${{ String: info.separator }}) + ${{ Int32: info.separator.length }})`;
+          // `cardinality` is the token's row count in the part, so this keeps
+          // each key's most frequent values rather than the first ones read.
           const sql = chSql`
         SELECT * FROM (
           SELECT ${{ String: columnName }} as column,
             substring(token, 1, position(token, ${{ String: info.separator }}) - 1) AS key,
-            groupUniqArray(${{ Int32: info.limit }})(${valueSql}) AS value
+            topKWeighted(${{ UNSAFE_RAW_SQL: inlineNonNegativeInt(info.limit, 'limit') }})(${valueSql}, cardinality) AS value
           FROM mergeTreeTextIndex(${{ String: databaseName }}, ${{ String: tableName }}, ${{ String: info.indexName }})
           WHERE ${partsFilter}
             AND (${orChain})
@@ -1815,7 +1820,7 @@ export class Metadata {
           const sql = chSql`
         SELECT * FROM (
           SELECT ${{ String: columnName }} AS key,
-            groupUniqArray(${{ Int32: info.limit }})(token) AS value
+            topKWeighted(${{ UNSAFE_RAW_SQL: inlineNonNegativeInt(info.limit, 'limit') }})(token, cardinality) AS value
           FROM mergeTreeTextIndex(${{ String: databaseName }}, ${{ String: tableName }}, ${{ String: info.indexName }})
           WHERE ${partsFilter}
             AND token != ''
@@ -2983,26 +2988,33 @@ export class Metadata {
 
     // fire all the kv fetch queries
     const promises: Array<Promise<KeyValues[] | undefined>> = [];
-    if (mapTextIndexQueryOptions.size > 0) {
-      promises.push(
-        this.getMapTextIndexKeyValues({
-          databaseName,
-          tableName,
-          connectionId,
-          queryOptions: mapTextIndexQueryOptions,
-          dateRange,
-          timestampValueExpression,
-          signal,
-        }),
-      );
+    // Text index lookups run in small parallel batches, so a slow,
+    // high-cardinality key holds up only its own batch's values.
+    for (const [column, options] of mapTextIndexQueryOptions) {
+      for (const keys of chunk(options.keys, TEXT_INDEX_KEYS_PER_QUERY)) {
+        promises.push(
+          this.getMapTextIndexKeyValues({
+            databaseName,
+            tableName,
+            connectionId,
+            queryOptions: new Map([[column, { ...options, keys }]]),
+            dateRange,
+            timestampValueExpression,
+            signal,
+          }),
+        );
+      }
     }
-    if (nativeTextIndexQueryOptions.size > 0) {
+    for (const entries of chunk(
+      [...nativeTextIndexQueryOptions],
+      TEXT_INDEX_KEYS_PER_QUERY,
+    )) {
       promises.push(
         this.getTextIndexKeyValues({
           databaseName,
           tableName,
           connectionId,
-          queryOptions: nativeTextIndexQueryOptions,
+          queryOptions: new Map(entries),
           dateRange,
           timestampValueExpression,
           signal,