diff --git a/.changeset/text-index-dictionary-block-size.md b/.changeset/text-index-dictionary-block-size.md new file mode 100644 index 0000000000..e6331b58b2 --- /dev/null +++ b/.changeset/text-index-dictionary-block-size.md @@ -0,0 +1,21 @@ +--- +'@hyperdx/otel-collector': patch +--- + +perf: use 8,192-token text index dictionary blocks on the seeded tables + +On ClickHouse 26.6 and later, the collector's schema migration now sets +`text_index_dictionary_block_size = 8192` (the default is 512) on `otel_logs`, +`otel_traces` and `hyperdx_sessions`, with `ALTER TABLE … MODIFY SETTING` after +the seed. Each part's text index sparse index is 16x smaller. Reading filter +values with `mergeTreeTextIndex()` loads that index for every part on every +read, and on a 537M-row table 4,096-token blocks already made those reads about +10x cheaper per part, made searches for a rare attribute value 3-7x faster, and +made searches for a common value 10-30% faster. + +Because it's an `ALTER`, existing installs get it too: new parts use 8,192-token +blocks, and existing parts keep theirs until they're merged. A table that +already sets `text_index_dictionary_block_size` is left alone, and operators who +manage these tables themselves can turn the change off with +`HYPERDX_OTEL_EXPORTER_SET_TEXT_INDEX_BLOCK_SIZE=false`. ClickHouse 26.2–26.5 +doesn't have the setting, so those servers keep 512-token blocks. diff --git a/.changeset/topk-text-index-values.md b/.changeset/topk-text-index-values.md new file mode 100644 index 0000000000..3565b3d1c8 --- /dev/null +++ b/.changeset/topk-text-index-values.md @@ -0,0 +1,27 @@ +--- +'@hyperdx/common-utils': patch +--- + +perf: read large-block Map text index values per key, most frequent first + +"All values" lookups on Map key=value text indexes with large dictionary blocks +(2048 or more, e.g. `text_index_dictionary_block_size = 8192`, which the +collector's migration now sets on ClickHouse 26.6+) +now read each key separately, in parallel queries of at most 10 keys, and stop +after 20,000 tokens per key, so a key with a value per row (request IDs, UUIDs) +can't hold up the others. Values come from `topKWeighted(limit)` weighted by +each token's row count, so a capped list holds the most common values, most +common first. These reads no longer apply `max_rows_to_read`: the index is read +in sorted token order, so the cap left every key that sorts after a +high-cardinality key with no values. + +Indexes with smaller blocks, including ClickHouse's default of 512, keep the +previous query shape (one read for all of a query's keys, the alphabetically +first values, and the `max_rows_to_read` cap), now in parallel queries of at +most 10 keys, so the cap's cut-off affects at most 9 other keys. Every read loads +each part's sparse index, which is large with small blocks, so one read per key +would multiply that cost. The block size comes from the index definition, then +the table's `text_index_dictionary_block_size`, then the server's. + +Text indexes on plain columns now return the most frequent values with +`topKWeighted`, in parallel queries of at most 10 columns, without the row cap. diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index 335054d479..b1bba4f792 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -45,6 +45,8 @@ services: # HYPERDX_OTEL_EXPORTER_LOGS_TTL / _TRACES_TTL / _METRICS_TTL / _SESSIONS_TTL # Apply TTL changes to already-existing tables too (default false): # HYPERDX_OTEL_EXPORTER_RECONCILE_TABLE_TTL: 'true' + # Leave text index dictionary block sizes alone (default: set 8192 on 26.6+): + # HYPERDX_OTEL_EXPORTER_SET_TEXT_INDEX_BLOCK_SIZE: 'false' ENABLE_PROMQL: 'true' volumes: - ./docker/otel-collector/config.yaml:/etc/otelcol-contrib/config.yaml diff --git a/docker/clickhouse/local/init-db-e2e.sh b/docker/clickhouse/local/init-db-e2e.sh index c5242dc598..8465661103 100755 --- a/docker/clickhouse/local/init-db-e2e.sh +++ b/docker/clickhouse/local/init-db-e2e.sh @@ -53,7 +53,7 @@ ENGINE = MergeTree PARTITION BY toDate(Timestamp) ORDER BY (toStartOfFiveMinutes(Timestamp), ServiceName, Timestamp) TTL toDateTime(Timestamp) + toIntervalDay(30) -SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1; +SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1, text_index_dictionary_block_size = 8192; -- Distributed table over the e2e logs table whose DDL declares a column -- (\`ColumnMissingFromShard\`) that its underlying MergeTree table does NOT have. @@ -129,7 +129,7 @@ ENGINE = MergeTree PARTITION BY toDate(Timestamp) ORDER BY (ServiceName, SpanName, toDateTime(Timestamp)) TTL toDate(Timestamp) + toIntervalDay(30) -SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1; +SETTINGS index_granularity = 8192, ttl_only_drop_parts = 1, text_index_dictionary_block_size = 8192; CREATE TABLE ${DATABASE}.e2e_hyperdx_sessions ( diff --git a/packages/app/src/components/DBRowTable.tsx b/packages/app/src/components/DBRowTable.tsx index 2c90cbf204..a7a251cccd 100644 --- a/packages/app/src/components/DBRowTable.tsx +++ b/packages/app/src/components/DBRowTable.tsx @@ -28,7 +28,10 @@ import { isJSDataTypeJSONStringifiable, JSDataType, } from '@hyperdx/common-utils/dist/clickhouse'; -import { splitAndTrimWithBracket } from '@hyperdx/common-utils/dist/core/utils'; +import { + getEngineSetting, + splitAndTrimWithBracket, +} from '@hyperdx/common-utils/dist/core/utils'; import { BuilderChartConfigWithDateRange, SelectList, @@ -1541,14 +1544,6 @@ const BLOCK_COLUMN_SETTINGS = [ { column: '_block_offset', setting: 'enable_block_offset_column' }, ]; -// engine_full ends with the SETTINGS clause. -function getEngineSetting(engineFull: string, name: string) { - const settingsClause = engineFull.match(/ SETTINGS (.*)$/)?.[1] ?? ''; - return splitAndTrimWithBracket(settingsClause) - .map(setting => setting.split(' = ')) - .find(([key]) => key === name)?.[1]; -} - function omitsBlockColumnSetting(engineFull: string) { return BLOCK_COLUMN_SETTINGS.some( ({ setting }) => getEngineSetting(engineFull, setting) == null, diff --git a/packages/common-utils/src/__tests__/kvItemsRewrite.int.test.ts b/packages/common-utils/src/__tests__/kvItemsRewrite.int.test.ts index 3748c8bd24..7a59955b65 100644 --- a/packages/common-utils/src/__tests__/kvItemsRewrite.int.test.ts +++ b/packages/common-utils/src/__tests__/kvItemsRewrite.int.test.ts @@ -70,7 +70,7 @@ describe('KV items rewrite integration', () => { LogAttributeItems Array(String) MATERIALIZED arrayMap((x) -> concat(x.1, '=', x.2), LogAttributes::Array(Tuple(String, String))) CODEC(ZSTD(1)), - INDEX idx_log_attr_items LogAttributeItems TYPE text(tokenizer = 'array') GRANULARITY 1 + INDEX idx_log_attr_items LogAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192) GRANULARITY 1 ) ENGINE = MergeTree ORDER BY (Timestamp)`, diff --git a/packages/common-utils/src/__tests__/metadata.int.test.ts b/packages/common-utils/src/__tests__/metadata.int.test.ts index 0e9133bc87..5d782bc871 100644 --- a/packages/common-utils/src/__tests__/metadata.int.test.ts +++ b/packages/common-utils/src/__tests__/metadata.int.test.ts @@ -697,7 +697,7 @@ describe('Metadata Integration Tests', () => { ResourceAttributeItems Array(String) MATERIALIZED arrayMap(x -> concat(x.1, '=', x.2), CAST(ResourceAttributes, 'Array(Tuple(String, String))')) CODEC(ZSTD(1)), - INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array') GRANULARITY 1 + INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192) GRANULARITY 1 ) ENGINE = MergeTree() ORDER BY (Timestamp) @@ -712,7 +712,7 @@ describe('Metadata Integration Tests', () => { LogAttributeItems Array(String) MATERIALIZED arrayMap((x) -> concat(x.1, '=', x.2), LogAttributes::Array(Tuple(String, String))) CODEC(ZSTD(1)), - INDEX idx_log_attr_items LogAttributeItems TYPE text(tokenizer = 'array') GRANULARITY 1 + INDEX idx_log_attr_items LogAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192) GRANULARITY 1 ) ENGINE = MergeTree() ORDER BY (Timestamp) @@ -907,6 +907,41 @@ 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}']`, + `ResourceAttributes['${mapKey}']`, + ], + maxValuesPerKey: 1, + }); + + // UNION ALL branches can arrive in any order. + expect(result).toHaveLength(2); + expect(result).toEqual( + expect.arrayContaining([ + { key: `ResourceAttributes['${freqKey}']`, value: ['z-common'] }, + { key: `ResourceAttributes['${mapKey}']`, value: [podNameA] }, + ]), + ); + }); + it('routes TraceId through getTextIndexKeyValues', async () => { if (!textIndexSupported) { console.warn( @@ -1622,14 +1657,14 @@ describe('Metadata Integration Tests', () => { { path: 'key', tableName: 'test_map_keys_key_text_index_order', - indexDefinition: `INDEX idx_log_attr_key mapKeys(LogAttributes) TYPE text(tokenizer = 'array') GRANULARITY 1`, + indexDefinition: `INDEX idx_log_attr_key mapKeys(LogAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192) GRANULARITY 1`, }, { path: 'kv', tableName: 'test_map_keys_kv_text_index_order', indexDefinition: `LogAttributeItems Array(String) MATERIALIZED arrayMap(x -> concat(x.1, '=', x.2), CAST(LogAttributes, 'Array(Tuple(String, String))')), - INDEX idx_log_attr_items LogAttributeItems TYPE text(tokenizer = 'array') GRANULARITY 1`, + INDEX idx_log_attr_items LogAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192) GRANULARITY 1`, }, ])( 'getMapKeys - $path text index ordering', @@ -1719,4 +1754,92 @@ describe('Metadata Integration Tests', () => { }); }, ); + + describe('getAllKeyValues - values repeated across parts', () => { + // mergeTreeTextIndex returns one row per part and token, so a token cap + // stops before the later parts' values. Each part holds the same SHARED + // values plus OWN values no other part has. + const tableName = 'test_text_index_values_across_parts'; + const PARTS = 25; + const SHARED = 1000; + const OWN = 100; + const dateRange = (): [Date, Date] => [ + new Date(Date.now() - 60 * 60 * 1000), + new Date(Date.now() + 60 * 1000), + ]; + let textIndexSupported = false; + + beforeAll(async () => { + const probe = new Metadata(hdxClient, new MetadataCache()); + textIndexSupported = supportsMergeTreeTextIndex( + await probe.getServerVersion({ connectionId: 'test_connection' }), + ); + if (!textIndexSupported) return; + + await client.command({ + query: `CREATE OR REPLACE TABLE default.${tableName} ( + Timestamp DateTime64(9), + ResourceAttributes Map(LowCardinality(String), String), + ResourceAttributeItems Array(String) MATERIALIZED + arrayMap(x -> concat(x.1, '=', x.2), CAST(ResourceAttributes, 'Array(Tuple(String, String))')), + Tags Array(String), + INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192) GRANULARITY 1, + INDEX idx_tags Tags TYPE text(tokenizer = 'array', dictionary_block_size = 8192) GRANULARITY 1 + ) + ENGINE = MergeTree() + PARTITION BY toDate(Timestamp) + ORDER BY Timestamp + `, + }); + await client.command({ + query: `SYSTEM STOP MERGES default.${tableName}`, + }); + for (let i = 0; i < PARTS; i++) { + await client.command({ + query: `INSERT INTO default.${tableName} (Timestamp, ResourceAttributes, Tags) + SELECT now64(9) - INTERVAL 10 MINUTE, map('pod', v), [v] + FROM ( + SELECT if(number < ${SHARED}, concat('shared-', toString(number)), concat('own-${i}-', toString(number))) AS v + FROM numbers(${SHARED + OWN}) + )`, + }); + } + }); + + afterAll(async () => { + await client.command({ + query: `DROP TABLE IF EXISTS default.${tableName}`, + }); + }); + + it.each([ + ['Map key=value', "ResourceAttributes['pod']"], + ['plain-column', 'Tags'], + ])( + 'returns every distinct value from the %s text index for "Load more"', + async (_, keyExpression) => { + if (!textIndexSupported) { + console.warn( + 'Skipping: ClickHouse < 26.3 does not support mergeTreeTextIndex()', + ); + return; + } + + // "Load more" passes null: every part, no token cap. + const metadata = new Metadata(hdxClient, new MetadataCache()); + const result = await metadata.getAllKeyValues({ + databaseName: 'default', + tableName, + keyExpressions: [keyExpression], + maxValuesPerKey: 10000, + mapKeyIndexMaxParts: null, + connectionId: 'test_connection', + dateRange: dateRange(), + timestampValueExpression: 'Timestamp', + }); + expect(result).toHaveLength(1); + expect(result[0].value).toHaveLength(SHARED + OWN * PARTS); + }, + ); + }); }); diff --git a/packages/common-utils/src/__tests__/metadata.test.ts b/packages/common-utils/src/__tests__/metadata.test.ts index 8ea6193746..08e640d148 100644 --- a/packages/common-utils/src/__tests__/metadata.test.ts +++ b/packages/common-utils/src/__tests__/metadata.test.ts @@ -2,6 +2,7 @@ import { ColumnMeta } from '@/clickhouse'; import { ClickhouseClient } from '@/clickhouse/node'; import { COLUMNS_CACHE_TTL_MS, + DEFAULT_METADATA_MAX_ROWS_TO_READ, GET_ALL_KEY_VALUES_CHUNK_SIZE, Metadata, MetadataCache, @@ -1380,8 +1381,8 @@ describe('Metadata', () => { const sql = calls[calls.length - 1][0].query as string; expect(sql).toContain('mergeTreeTextIndex('); - expect(sql).toContain('groupUniqArray('); - expect(sql).toContain('GROUP BY key'); + expect(sql).toContain('topKWeighted('); + expect(sql).toContain('LIMIT 20000'); expect(sql).not.toContain('startsWith(token,'); }); @@ -1400,9 +1401,128 @@ describe('Metadata', () => { expect(sql).toContain('mergeTreeTextIndex('); expect(sql).toContain('startsWith(token,'); expect(sql).toContain('substring(token, position(token,'); + expect(sql).toContain("startsWith(token, 'requestId=')"); + }); + + const withDictionaryBlockSize = (size: number) => + jest.spyOn(metadata, 'getSkipIndices').mockResolvedValue([ + { + name: 'idx_log_attr_items', + type: 'text', + typeFull: `text(tokenizer = 'array', dictionary_block_size = ${size})`, + expression: 'LogAttributes', + granularity: 100000000, + }, + ]); + + it('reads each Map key in its own capped branch when the index has large dictionary blocks', async () => { + setupDefaultLogsSchema(); + withDictionaryBlockSize(8192); + + await metadata.getAllKeyValues({ + ...baseArgs, + keyExpressions: [ + "LogAttributes['requestId']", + "LogAttributes['region']", + ], + }); + + const call = (mockClickhouseClient.query as jest.Mock).mock.calls.find( + ([opts]) => (opts.query as string).includes('startsWith(token,'), + )![0]; + const branches = (call.query as string).split('UNION ALL'); + expect(branches).toHaveLength(2); + expect(branches[0]).toContain("startsWith(token, 'requestId=')"); + expect(branches[0]).toContain("'requestId' AS key"); + expect(branches[1]).toContain("startsWith(token, 'region=')"); + for (const branch of branches) { + expect(branch).toContain('LIMIT 20000'); + } + expect(call.clickhouse_settings.max_rows_to_read).toBe('0'); + expect(call.clickhouse_settings.read_overflow_mode).toBeUndefined(); + }); + + it.each([ + ['map KV', "LogAttributes['requestId']"], + ['native', 'TraceId'], + ])( + 'caps the %s text index read per key, and drops the cap when every part is read', + async (_, keyExpression) => { + setupDefaultLogsSchema(); + withDictionaryBlockSize(8192); + + for (const mapKeyIndexMaxParts of [5, null]) { + await metadata.getAllKeyValues({ + ...baseArgs, + keyExpressions: [keyExpression], + mapKeyIndexMaxParts, + }); + } + + const tokenLimits = (mockClickhouseClient.query as jest.Mock).mock.calls + .map(([opts]) => opts.query as string) + .filter(q => q.includes('mergeTreeTextIndex(')) + .map(q => q.match(/LIMIT (\d+)/)?.[1]); + expect(tokenLimits).toEqual(['20000', undefined]); + }, + ); + + it('reads all Map keys in one pass when the index has small dictionary blocks', async () => { + setupDefaultLogsSchema(); + withDictionaryBlockSize(512); + + await metadata.getAllKeyValues({ + ...baseArgs, + keyExpressions: [ + "LogAttributes['requestId']", + "LogAttributes['region']", + ], + }); + + const call = (mockClickhouseClient.query as jest.Mock).mock.calls.find( + ([opts]) => (opts.query as string).includes('startsWith(token,'), + )![0]; + const sql = call.query as string; + expect(sql).not.toContain('UNION ALL'); + expect(sql).toContain( + "startsWith(token, 'requestId=') OR startsWith(token, 'region=')", + ); + expect(sql).toContain('groupUniqArray('); expect(sql).toContain('GROUP BY column, key'); + expect(call.clickhouse_settings).toMatchObject({ + max_rows_to_read: String(DEFAULT_METADATA_MAX_ROWS_TO_READ), + read_overflow_mode: 'break', + }); }); + it.each([ + ['map KV', "LogAttributes['requestId']"], + ['native', 'TraceId'], + ])( + 'reads the %s text index without a row cap, even when one is configured', + async (_, keyExpression) => { + setupDefaultLogsSchema(); + // Small-block Map indexes keep the cap; see the one-pass test. + withDictionaryBlockSize(8192); + jest + .spyOn(metadata, 'getClickHouseSettings') + .mockReturnValue({ max_rows_to_read: '1000' }); + + await metadata.getAllKeyValues({ + ...baseArgs, + keyExpressions: [keyExpression], + }); + + const calls = (mockClickhouseClient.query as jest.Mock).mock.calls; + const textIndexCall = calls.find(([opts]) => + opts.query.includes('mergeTreeTextIndex('), + ); + expect(textIndexCall[0].clickhouse_settings).toMatchObject({ + max_rows_to_read: '0', + }); + }, + ); + it.each([ ['map KV', "LogAttributes['requestId']"], ['native', 'TraceId'], @@ -1485,6 +1605,36 @@ describe('Metadata', () => { }, ); + it.each([ + [8192, [10, 10, 5]], + // The per-key threshold is 2048. + [2048, [10, 10, 5]], + [1024, [1, 1, 1]], + [512, [1, 1, 1]], + ])( + 'batches Map text index values by dictionary block size (%d)', + async (blockSize, expectedBranchCounts) => { + setupDefaultLogsSchema(); + withDictionaryBlockSize(blockSize); + + await metadata.getAllKeyValues({ + ...baseArgs, + keyExpressions: Array.from( + { length: 25 }, + (_, i) => `LogAttributes['key${i}']`, + ), + }); + + const branchCounts = ( + mockClickhouseClient.query as jest.Mock + ).mock.calls + .map(([opts]) => opts.query as string) + .filter(q => q.includes('startsWith(token,')) + .map(q => q.split('UNION ALL').length); + expect(branchCounts).toEqual(expectedBranchCounts); + }, + ); + it('routes ServiceName and SeverityText through getMetadataMVKeyValues (KV rollup MV path)', async () => { setupDefaultLogsSchema(); @@ -1805,34 +1955,38 @@ 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 () => { - setupDefaultLogsSchema(); - - await metadata.getAllKeyValues({ - ...baseArgs, - keyExpressions: ["LogAttributes['requestId']"], - maxValuesPerKey: MAX_VALUES, - }); + it.each([ + [8192, 'topKWeighted'], + [512, 'groupUniqArray'], + ])( + 'threads maxValuesPerKey into the map text-index query (dictionary blocks of %d)', + async (blockSize, aggregate) => { + setupDefaultLogsSchema(); + withDictionaryBlockSize(blockSize); + + await metadata.getAllKeyValues({ + ...baseArgs, + keyExpressions: ["LogAttributes['requestId']"], + maxValuesPerKey: MAX_VALUES, + }); - const mapTextIndexCall = ( - mockClickhouseClient.query as jest.Mock - ).mock.calls.find( - (c: any[]) => - typeof c[0].query === 'string' && - 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); - }); + const mapTextIndexCall = ( + mockClickhouseClient.query as jest.Mock + ).mock.calls.find( + (c: any[]) => + typeof c[0].query === 'string' && + c[0].query.includes('startsWith(token,'), + ); + expect(mapTextIndexCall).toBeDefined(); + expect(mapTextIndexCall![0].query).toContain( + `${aggregate}(${MAX_VALUES})(`, + ); + }, + ); it('threads maxValuesPerKey into the native text-index query', async () => { setupDefaultLogsSchema(); @@ -1852,11 +2006,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 () => { @@ -2065,6 +2217,52 @@ describe('Metadata', () => { value: ['r1'], }; + it("looks up every Map column's dictionary block size at once", async () => { + setupDefaultLogsSchema(); + const kvIndex = (mapColumn: string, indexName: string) => ({ + kv: { + columnName: mapColumn, + mapColumn, + indexName, + separator: '=', + useHasAny: false, + }, + }); + jest.spyOn(metadata, 'getMapColumnTextIndexes').mockResolvedValue( + new Map([ + ['LogAttributes', kvIndex('LogAttributes', 'idx_log_attr_items')], + [ + 'ResourceAttributes', + kvIndex('ResourceAttributes', 'idx_res_attr_items'), + ], + ]), + ); + let releaseSettings: () => void = () => {}; + const settings = jest + .spyOn(metadata, 'getMergeTreeSettings') + .mockReturnValue( + new Promise(resolve => { + releaseSettings = () => resolve(undefined); + }), + ); + + const result = metadata.getAllKeyValues({ + ...baseArgs, + keyExpressions: [ + "LogAttributes['requestId']", + "ResourceAttributes['service.name']", + ], + }); + // Both lookups start before either finishes. + for (let i = 0; i < 50 && settings.mock.calls.length < 2; i++) { + await new Promise(resolve => setImmediate(resolve)); + } + expect(settings).toHaveBeenCalledTimes(2); + + releaseSettings(); + await result; + }); + it('yields each query as it finishes, without waiting for the slowest', async () => { setupDefaultLogsSchema(); let releaseMapTextIndex: () => void = () => {}; diff --git a/packages/common-utils/src/__tests__/utils.test.ts b/packages/common-utils/src/__tests__/utils.test.ts index 327585152b..452e3f5890 100644 --- a/packages/common-utils/src/__tests__/utils.test.ts +++ b/packages/common-utils/src/__tests__/utils.test.ts @@ -15,6 +15,7 @@ import { getDistributedTableArgs, getFirstOrderingItem, getPercentileAggFnLevel, + getTextIndexDictionaryBlockSize, hasNonEmptyOrderBy, hasPositiveSeriesLimit, isFirstOrderByAscending, @@ -3620,4 +3621,53 @@ describe('utils', () => { expect(resolvePercentileAggFn(select)).toBe(select); }); }); + + describe('getTextIndexDictionaryBlockSize', () => { + const engineFull = + 'MergeTree ORDER BY Timestamp SETTINGS index_granularity = 8192, text_index_dictionary_block_size = 4096'; + + it.each([ + [ + 'the index argument first', + { + typeFull: "text(tokenizer = 'array', dictionary_block_size = 8192)", + engineFull, + serverMergeTreeSettings: new Map([ + ['text_index_dictionary_block_size', '1024'], + ]), + }, + 8192, + ], + [ + 'the table setting next', + { + typeFull: "text(tokenizer = 'array')", + engineFull, + serverMergeTreeSettings: new Map([ + ['text_index_dictionary_block_size', '1024'], + ]), + }, + 4096, + ], + [ + 'the server setting next', + { + typeFull: "text(tokenizer = 'array')", + engineFull: 'MergeTree ORDER BY Timestamp', + serverMergeTreeSettings: new Map([ + ['text_index_dictionary_block_size', '1024'], + ]), + }, + 1024, + ], + ['512 when nothing is known', {}, 512], + [ + '512 for a value that is not a positive integer', + { typeFull: "text(tokenizer = 'array', dictionary_block_size = 0)" }, + 512, + ], + ])('uses %s', (_, input, expected) => { + expect(getTextIndexDictionaryBlockSize(input)).toBe(expected); + }); + }); }); diff --git a/packages/common-utils/src/core/metadata.ts b/packages/common-utils/src/core/metadata.ts index 81a93eedad..b77e667d35 100644 --- a/packages/common-utils/src/core/metadata.ts +++ b/packages/common-utils/src/core/metadata.ts @@ -59,6 +59,7 @@ import { import { getAlignedDateRange, getDistributedTableArgs, + getTextIndexDictionaryBlockSize, MetadataMVQueryOptions, objectHash, splitAndTrimWithBracket, @@ -78,6 +79,25 @@ export const DEFAULT_MAP_KEY_INDEX_MAX_PARTS = 20; // 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; +// Index rows a text index values lookup reads per key. The dictionary is +// sorted, so a key with more than this gets the most frequent among its +// alphabetically first ones. +const TEXT_INDEX_TOKENS_PER_KEY = 20_000; + +// mergeTreeTextIndex returns one row per part and token, so the cap counts a +// value once per part it's in, and values only in parts read after it are +// missed. That sample is enough for the top values of a filter, but "Load +// more" (no part limit) wants every value, so its read isn't capped. +const textIndexTokenLimit = (maxParts: number | null): ChSql => + maxParts == null + ? chSql`` + : chSql`LIMIT ${{ UNSAFE_RAW_SQL: String(TEXT_INDEX_TOKENS_PER_KEY) }}`; +// Keys or columns per text index values query; see dispatchKeyValueQueries. +const TEXT_INDEX_KEYS_PER_QUERY = 10; +// Smallest dictionary block size that gets one read per key; see +// getMapTextIndexKeyValues. +const PER_KEY_TEXT_INDEX_MIN_DICTIONARY_BLOCK_SIZE = 2048; + /** * 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 @@ -1751,11 +1771,37 @@ export class Metadata { }); } + // shortcut: parts written before a table's text_index_dictionary_block_size + // changed keep their old block size; good enough to pick a query shape. + private async getTextIndexDictionaryBlockSize({ + databaseName, + tableName, + connectionId, + indexName, + }: TableConnection & { indexName: string }): Promise { + const [skipIndices, tableMetadata, serverMergeTreeSettings] = + await Promise.all([ + this.getSkipIndices({ databaseName, tableName, connectionId }).catch( + () => [] as SkipIndexMetadata[], + ), + this.getTableMetadata({ databaseName, tableName, connectionId }).catch( + () => undefined, + ), + this.getMergeTreeSettings({ connectionId }), + ]); + return getTextIndexDictionaryBlockSize({ + typeFull: skipIndices.find(idx => idx.name === indexName)?.typeFull, + engineFull: tableMetadata?.engine_full, + serverMergeTreeSettings, + }); + } + private async getMapTextIndexKeyValues({ databaseName, tableName, connectionId, queryOptions, + perKey, maxParts, partSelection, dateRange, @@ -1763,6 +1809,8 @@ export class Metadata { signal, }: TableConnection & { queryOptions: TextIndexMapColumnQueryOptions; + /** One read per key instead of one per column; see the comment below. */ + perKey: boolean; /** null reads every part in the time range. */ maxParts: number | null; partSelection: MapKeyIndexPartSelection; @@ -1771,24 +1819,11 @@ export class Metadata { signal?: AbortSignal; }): Promise { const queryOptionsHash = objectHash(queryOptions); - const cacheKey = `${databaseName}.${tableName}.${connectionId}.${dateRange[0].toString()}.${dateRange[1].toString()}.${queryOptionsHash}.${maxParts}.${partSelection}.${timestampValueExpression}.getMapTextIndexKeyValues`; + const cacheKey = `${databaseName}.${tableName}.${connectionId}.${dateRange[0].toString()}.${dateRange[1].toString()}.${queryOptionsHash}.${perKey}.${maxParts}.${partSelection}.${timestampValueExpression}.getMapTextIndexKeyValues`; return this.cache.getOrFetch(cacheKey, async () => { try { const sqlBranches: Array = []; for (const [columnName, info] of queryOptions.entries()) { - const orChain = concatChSql( - ' OR ', - // Inline keys as SQL-escaped literals, not bind params: ~100 - // per-key params exceed the web client's URL param budget and - // silently switch the request to a multipart body that proxies - // may reject. Keys are ingest-controlled, hence SqlString.escape. - info.keys.map( - k => - chSql`startsWith(token, ${{ - UNSAFE_RAW_SQL: SqlString.escape(`${k}${info.separator}`), - }})`, - ), - ); const partsFilter = await this.partsOverlapFilter({ databaseName, tableName, @@ -1798,18 +1833,50 @@ export class Metadata { partSelection, }); const valueSql = chSql`substring(token, position(token, ${{ String: info.separator }}) + ${{ Int32: info.separator.length }})`; - const sql = chSql` - SELECT * FROM ( - SELECT ${{ String: columnName }} as column, + // Inline keys as SQL-escaped literals, not bind params: ~100 + // per-key params exceed the web client's URL param budget and + // silently switch the request to a multipart body that proxies + // may reject. Keys are ingest-controlled, hence SqlString.escape. + const keyPrefix = (k: string) => + chSql`${{ UNSAFE_RAW_SQL: SqlString.escape(`${k}${info.separator}`) }}`; + // Every read of the index loads each part's sparse index, which has + // one entry per dictionary block. With large blocks that is cheap, so + // each key gets its own read whose LIMIT stops it early, and a key + // with a value per row can't slow down the others; `cardinality` is + // the token's row count in the part, so topKWeighted keeps the most + // frequent values among the tokens read. With small blocks + // (ClickHouse's default 512) the sparse index is large, so the + // batch's keys share one read. + if (!perKey) { + sqlBranches.push(chSql` + SELECT ${{ String: columnName }} AS column, substring(token, 1, position(token, ${{ String: info.separator }}) - 1) AS key, - groupUniqArray(${{ Int32: info.limit }})(${valueSql}) AS value + groupUniqArray(${{ UNSAFE_RAW_SQL: inlineNonNegativeInt(info.limit, 'limit') }})(${valueSql}) AS value FROM mergeTreeTextIndex(${{ String: databaseName }}, ${{ String: tableName }}, ${{ String: info.indexName }}) WHERE ${partsFilter} - AND (${orChain}) + AND (${concatChSql( + ' OR ', + info.keys.map(k => chSql`startsWith(token, ${keyPrefix(k)})`), + )}) AND ${valueSql} != '' GROUP BY column, key - )`; - sqlBranches.push(sql); + `); + continue; + } + for (const k of info.keys) { + sqlBranches.push(chSql` + SELECT ${{ String: columnName }} AS column, + ${{ UNSAFE_RAW_SQL: SqlString.escape(k) }} AS key, + topKWeighted(${{ UNSAFE_RAW_SQL: inlineNonNegativeInt(info.limit, 'limit') }})(v, cardinality) AS value + FROM ( + SELECT ${valueSql} AS v, cardinality + FROM mergeTreeTextIndex(${{ String: databaseName }}, ${{ String: tableName }}, ${{ String: info.indexName }}) + WHERE ${partsFilter} + AND startsWith(token, ${keyPrefix(k)}) + AND v != '' + ${textIndexTokenLimit(maxParts)} + )`); + } } const sql = concatChSql(' UNION ALL ', sqlBranches); @@ -1818,14 +1885,25 @@ export class Metadata { query: sql.sql, query_params: sql.params, connectionId, - clickhouse_settings: { - max_rows_to_read: String( - this.getClickHouseSettings().max_rows_to_read ?? - DEFAULT_METADATA_MAX_ROWS_TO_READ, - ), - read_overflow_mode: 'break', - ...this.getClickHouseSettings(), - }, + clickhouse_settings: perKey + ? { + ...this.getClickHouseSettings(), + // The per-key LIMIT bounds each read. A row cap would stop + // the sorted dictionary read partway and leave the keys that + // sort after it with no values. + max_rows_to_read: '0', + } + : { + // shortcut: bounds the shared read, but keys that sort after + // a high-cardinality key can come back with no values; drop + // it if that matters more than the read's cost. + max_rows_to_read: String( + this.getClickHouseSettings().max_rows_to_read ?? + DEFAULT_METADATA_MAX_ROWS_TO_READ, + ), + read_overflow_mode: 'break', + ...this.getClickHouseSettings(), + }, abort_signal: signal, }) .then(res => @@ -1883,14 +1961,16 @@ export class Metadata { maxParts: maxParts ?? undefined, partSelection, }); + // See getMapTextIndexKeyValues for the per-column LIMIT. const sql = chSql` - SELECT * FROM ( - SELECT ${{ String: columnName }} AS key, - groupUniqArray(${{ Int32: info.limit }})(token) AS value + SELECT ${{ String: columnName }} AS key, + topKWeighted(${{ UNSAFE_RAW_SQL: inlineNonNegativeInt(info.limit, 'limit') }})(token, cardinality) AS value + FROM ( + SELECT token, cardinality FROM mergeTreeTextIndex(${{ String: databaseName }}, ${{ String: tableName }}, ${{ String: info.indexName }}) WHERE ${partsFilter} AND token != '' - GROUP BY key + ${textIndexTokenLimit(maxParts)} )`; sqlBranches.push(sql); } @@ -1902,12 +1982,9 @@ export class Metadata { query_params: sql.params, connectionId, clickhouse_settings: { - max_rows_to_read: String( - this.getClickHouseSettings().max_rows_to_read ?? - DEFAULT_METADATA_MAX_ROWS_TO_READ, - ), - read_overflow_mode: 'break', ...this.getClickHouseSettings(), + // See getMapTextIndexKeyValues. + max_rows_to_read: '0', }, abort_signal: signal, }) @@ -3060,28 +3137,52 @@ export class Metadata { // fire all the kv fetch queries const promises: Array> = []; - if (mapTextIndexQueryOptions.size > 0) { - promises.push( - this.getMapTextIndexKeyValues({ - databaseName, - tableName, - connectionId, - queryOptions: mapTextIndexQueryOptions, - maxParts: mapKeyIndexMaxParts, - partSelection: mapKeyIndexPartSelection, - 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. Within a + // batch, large dictionary blocks get one read per key; see + // getMapTextIndexKeyValues. + // Every Map column's block size is looked up at once, not one at a time. + const mapColumns = await Promise.all( + [...mapTextIndexQueryOptions].map(async ([column, options]) => ({ + column, + options, + perKey: + (await this.getTextIndexDictionaryBlockSize({ + databaseName, + tableName, + connectionId, + indexName: options.indexName, + })) >= PER_KEY_TEXT_INDEX_MIN_DICTIONARY_BLOCK_SIZE, + })), + ); + for (const { column, options, perKey } of mapColumns) { + 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 }]]), + perKey, + maxParts: mapKeyIndexMaxParts, + partSelection: mapKeyIndexPartSelection, + 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), maxParts: mapKeyIndexMaxParts, partSelection: mapKeyIndexPartSelection, dateRange, diff --git a/packages/common-utils/src/core/utils.ts b/packages/common-utils/src/core/utils.ts index 7c93078458..a79601e5ca 100644 --- a/packages/common-utils/src/core/utils.ts +++ b/packages/common-utils/src/core/utils.ts @@ -1443,6 +1443,41 @@ export type TextIndexTokenizer = } | { type: 'array' }; +/** Reads a setting from `system.tables.engine_full`, which ends with the SETTINGS clause. */ +export function getEngineSetting(engineFull: string, name: string) { + const settingsClause = engineFull.match(/ SETTINGS (.*)$/)?.[1] ?? ''; + return splitAndTrimWithBracket(settingsClause) + .map(setting => setting.split(' = ')) + .find(([key]) => key === name)?.[1]; +} + +const DEFAULT_TEXT_INDEX_DICTIONARY_BLOCK_SIZE = 512; + +/** + * A text index's dictionary block size: the index argument, else the table's + * `text_index_dictionary_block_size`, else the server's, else ClickHouse's 512. + */ +export function getTextIndexDictionaryBlockSize({ + typeFull, + engineFull, + serverMergeTreeSettings, +}: { + typeFull?: string; + engineFull?: string; + serverMergeTreeSettings?: Map; +}): number { + const value = + typeFull?.match(/\bdictionary_block_size\s*=\s*(\d+)/)?.[1] ?? + (engineFull + ? getEngineSetting(engineFull, 'text_index_dictionary_block_size') + : undefined) ?? + serverMergeTreeSettings?.get('text_index_dictionary_block_size'); + const size = Number(value); + return Number.isInteger(size) && size > 0 + ? size + : DEFAULT_TEXT_INDEX_DICTIONARY_BLOCK_SIZE; +} + /** * Parses the tokenizer and any associated tokenizer parameters from a text index type definition. * diff --git a/packages/otel-collector/cmd/migrate/main.go b/packages/otel-collector/cmd/migrate/main.go index b67e373099..6d680adf69 100644 --- a/packages/otel-collector/cmd/migrate/main.go +++ b/packages/otel-collector/cmd/migrate/main.go @@ -9,6 +9,7 @@ import ( "crypto/tls" "crypto/x509" "database/sql" + "errors" "fmt" "log" "net/url" @@ -42,6 +43,9 @@ type Config struct { // Reconcile TTL on already-existing tables, not just newly created ones. ReconcileTableTTL bool + // Set 8,192-token text index dictionary blocks on the seeded tables. + SetTextIndexBlockSize bool + // TLS settings TLSCAFile string TLSCertFile string @@ -68,6 +72,7 @@ func loadConfig() (*Config, error) { MetricsTTL: getEnv("HYPERDX_OTEL_EXPORTER_METRICS_TTL", tablesTTL), SessionsTTL: getEnv("HYPERDX_OTEL_EXPORTER_SESSIONS_TTL", tablesTTL), ReconcileTableTTL: getEnv("HYPERDX_OTEL_EXPORTER_RECONCILE_TABLE_TTL", "false") == "true", + SetTextIndexBlockSize: getEnv("HYPERDX_OTEL_EXPORTER_SET_TEXT_INDEX_BLOCK_SIZE", "true") != "false", TLSCAFile: getEnv("CLICKHOUSE_TLS_CA_FILE", ""), TLSCertFile: getEnv("CLICKHOUSE_TLS_CERT_FILE", ""), TLSKeyFile: getEnv("CLICKHOUSE_TLS_KEY_FILE", ""), @@ -644,6 +649,49 @@ func supportsFullTextSearch(major, minor int) bool { return major > 26 || (major == 26 && minor >= 2) } +// textIndexTables are the seeded tables with text indexes. +var textIndexTables = []string{"otel_logs", "otel_traces", "hyperdx_sessions"} + +// supportsTextIndexTableSettings returns true if the ClickHouse version has the +// text_index_dictionary_block_size MergeTree setting. This requires ClickHouse +// >= 26.6. +func supportsTextIndexTableSettings(major, minor int) bool { + return major > 26 || (major == 26 && minor >= 6) +} + +// setTextIndexBlockSize gives the seeded tables 8,192-token text index +// dictionary blocks (the default is 512), so filter value reads load a 16x +// smaller sparse index per part. It runs after the seed so existing tables +// get it too; new parts use it, existing parts keep theirs until merged. A +// table that already sets the value is left alone. +func setTextIndexBlockSize(ctx context.Context, db *sql.DB, database string) error { + var firstErr error + for _, table := range textIndexTables { + var engineFull string + err := db.QueryRowContext(ctx, + "SELECT engine_full FROM system.tables WHERE database = ? AND name = ?", + database, table).Scan(&engineFull) + if errors.Is(err, sql.ErrNoRows) { + continue + } + if err == nil && strings.Contains(engineFull, "text_index_dictionary_block_size") { + continue + } + if err == nil { + _, err = db.ExecContext(ctx, fmt.Sprintf( + "ALTER TABLE `%s`.`%s` MODIFY SETTING text_index_dictionary_block_size = 8192", + database, table)) + } + if err != nil { + log.Printf("WARNING: could not set the text index block size on %s: %v", table, err) + if firstErr == nil { + firstErr = err + } + } + } + return firstErr +} + // swapLogsSchemaForCompat replaces the full-text-search logs schema with the // compatibility variant (bloom_filter indexes) in the processed temp directory. // It removes 00002_otel_logs.sql and renames 00002_otel_logs_compat.sql to @@ -875,6 +923,15 @@ func main() { os.Exit(1) } + // On by default; HYPERDX_OTEL_EXPORTER_SET_TEXT_INDEX_BLOCK_SIZE=false lets + // operators who manage these tables opt out. Non-fatal so it can't block + // startup; older servers keep 512-token blocks. + if cfg.SetTextIndexBlockSize && supportsTextIndexTableSettings(chMajor, chMinor) { + if err := setTextIndexBlockSize(ctx, db, cfg.Database); err != nil { + log.Printf("WARNING: text index block size: %v", err) + } + } + // CREATE TABLE IF NOT EXISTS won't update TTL on existing tables; reconcile // (opt-in) applies it to them too. Non-fatal so it can't block startup. if cfg.ReconcileTableTTL { diff --git a/packages/otel-collector/cmd/migrate/main_test.go b/packages/otel-collector/cmd/migrate/main_test.go index 84e42687d5..6e2fea4620 100644 --- a/packages/otel-collector/cmd/migrate/main_test.go +++ b/packages/otel-collector/cmd/migrate/main_test.go @@ -111,6 +111,9 @@ func TestLoadConfig(t *testing.T) { if cfg.ReconcileTableTTL { t.Error("ReconcileTableTTL should default to false") } + if !cfg.SetTextIndexBlockSize { + t.Error("SetTextIndexBlockSize should default to true") + } if cfg.SchemaDir != dir { t.Errorf("SchemaDir: got %q, want %q", cfg.SchemaDir, dir) } @@ -629,6 +632,20 @@ func TestProcessSchemaDir_NonexistentDir(t *testing.T) { // per-signal TTL resolution + reconcile helpers // --------------------------------------------------------------------------- +func TestLoadConfig_TextIndexBlockSizeOptOut(t *testing.T) { + dir := t.TempDir() + os.Args = []string{"migrate", dir} + t.Setenv("HYPERDX_OTEL_EXPORTER_SET_TEXT_INDEX_BLOCK_SIZE", "false") + + cfg, err := loadConfig() + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if cfg.SetTextIndexBlockSize { + t.Error("SetTextIndexBlockSize should be false") + } +} + func TestLoadConfig_PerSignalTTL(t *testing.T) { dir := t.TempDir() os.Args = []string{"migrate", dir} @@ -958,6 +975,35 @@ func TestSupportsFullTextSearch(t *testing.T) { } } +// --------------------------------------------------------------------------- +// supportsTextIndexTableSettings +// --------------------------------------------------------------------------- + +func TestSupportsTextIndexTableSettings(t *testing.T) { + tests := []struct { + name string + major int + minor int + expected bool + }{ + {"26.6 exact threshold", 26, 6, true}, + {"26.10 above minor", 26, 10, true}, + {"27.0 above major", 27, 0, true}, + {"26.5 below minor", 26, 5, false}, + {"26.2 text indexes without the setting", 26, 2, false}, + {"25.9 below major", 25, 9, false}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := supportsTextIndexTableSettings(tt.major, tt.minor) + if got != tt.expected { + t.Errorf("supportsTextIndexTableSettings(%d, %d) = %v, want %v", tt.major, tt.minor, got, tt.expected) + } + }) + } +} + // --------------------------------------------------------------------------- // swapLogsSchemaForCompat // ---------------------------------------------------------------------------