diff --git a/.changeset/text-index-dictionary-block-size.md b/.changeset/text-index-dictionary-block-size.md new file mode 100644 index 0000000000..da6a7e61b6 --- /dev/null +++ b/.changeset/text-index-dictionary-block-size.md @@ -0,0 +1,18 @@ +--- +'@hyperdx/otel-collector': patch +--- + +perf: use 8,192-token dictionary blocks for every text index in the seed +schemas + +Every text index on `otel_logs`, `otel_traces` and `hyperdx_sessions` now sets +`dictionary_block_size = 8192` (the default is 512). Each part's 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. The setting is a per-index argument rather than the table setting +`text_index_dictionary_block_size`, which ClickHouse only accepts from 26.6. + +Only new installs get the new schema. Existing tables keep their indexes until +the indexes are cleared and materialized again. diff --git a/.changeset/topk-text-index-values.md b/.changeset/topk-text-index-values.md new file mode 100644 index 0000000000..86398f1f36 --- /dev/null +++ b/.changeset/topk-text-index-values.md @@ -0,0 +1,26 @@ +--- +'@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 +(4096 or more, e.g. `dictionary_block_size = 8192` as the seed schemas now use) +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/clickhouse/local/init-db-e2e.sh b/docker/clickhouse/local/init-db-e2e.sh index c5242dc598..4ce993493b 100755 --- a/docker/clickhouse/local/init-db-e2e.sh +++ b/docker/clickhouse/local/init-db-e2e.sh @@ -40,14 +40,14 @@ CREATE TABLE IF NOT EXISTS ${DATABASE}.e2e_otel_logs \`__hdx_materialized_k8s.pod.name\` LowCardinality(String) MATERIALIZED ResourceAttributes['k8s.pod.name'] CODEC(ZSTD(3)), \`__hdx_materialized_k8s.pod.uid\` LowCardinality(String) MATERIALIZED ResourceAttributes['k8s.pod.uid'] CODEC(ZSTD(3)), \`__hdx_materialized_deployment.environment.name\` LowCardinality(String) MATERIALIZED ResourceAttributes['deployment.environment.name'] CODEC(ZSTD(3)), - INDEX idx_trace_id TraceId TYPE text(tokenizer = 'array'), - INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_scope_attr_value mapValues(ScopeAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_log_attr_key mapKeys(LogAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_log_attr_value mapValues(LogAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_lower_body lower(Body) TYPE text(tokenizer = 'splitByNonAlpha') + INDEX idx_trace_id TraceId TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_scope_attr_value mapValues(ScopeAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_log_attr_key mapKeys(LogAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_log_attr_value mapValues(LogAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_lower_body lower(Body) TYPE text(tokenizer = 'splitByNonAlpha', dictionary_block_size = 8192) ) ENGINE = MergeTree PARTITION BY toDate(Timestamp) @@ -118,10 +118,10 @@ CREATE TABLE IF NOT EXISTS ${DATABASE}.e2e_otel_traces INDEX idx_rum_session_id __hdx_materialized_rum.sessionId TYPE bloom_filter(0.001) GRANULARITY 1, INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1, - INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array'), + INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192), INDEX idx_span_attr_key mapKeys(SpanAttributes) TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_span_attr_value mapValues(SpanAttributes) TYPE bloom_filter(0.01) GRANULARITY 1, - INDEX idx_span_attr_items SpanAttributeItems TYPE text(tokenizer = 'array'), + INDEX idx_span_attr_items SpanAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192), INDEX idx_duration Duration TYPE minmax GRANULARITY 1, INDEX idx_lower_span_name lower(SpanName) TYPE tokenbf_v1(32768, 3, 0) GRANULARITY 8 ) diff --git a/docker/otel-collector/schema/seed/00002_otel_logs.sql b/docker/otel-collector/schema/seed/00002_otel_logs.sql index 12ec3fbe37..fc046fdcb7 100644 --- a/docker/otel-collector/schema/seed/00002_otel_logs.sql +++ b/docker/otel-collector/schema/seed/00002_otel_logs.sql @@ -29,15 +29,18 @@ CREATE TABLE IF NOT EXISTS ${DATABASE}.otel_logs `ResourceAttributeItems` Array(String) ALIAS arrayMap((arr) -> concat(arr.1, '=', arr.2), ResourceAttributes::Array(Tuple(String, String))), `ScopeAttributeItems` Array(String) ALIAS arrayMap((arr) -> concat(arr.1, '=', arr.2), ScopeAttributes::Array(Tuple(String, String))), `LogAttributeItems` Array(String) ALIAS arrayMap((arr) -> concat(arr.1, '=', arr.2), LogAttributes::Array(Tuple(String, String))), - INDEX idx_trace_id TraceId TYPE text(tokenizer = 'array'), - INDEX idx_session_id __hdx_materialized_session.id TYPE text(tokenizer = 'array'), - INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array'), - INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_scope_attr_items ScopeAttributeItems TYPE text(tokenizer = 'array'), - INDEX idx_log_attr_key mapKeys(LogAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_log_attr_items LogAttributeItems TYPE text(tokenizer = 'array'), - INDEX idx_lower_body lower(Body) TYPE text(tokenizer = 'splitByNonAlpha') + -- Text indexes use 8,192-token dictionary blocks (the default is 512). Each + -- part's sparse index is 16x smaller, which makes reading filter values + -- with mergeTreeTextIndex() and looking up tokens much cheaper on large parts. + INDEX idx_trace_id TraceId TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_session_id __hdx_materialized_session.id TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_scope_attr_items ScopeAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_log_attr_key mapKeys(LogAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_log_attr_items LogAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_lower_body lower(Body) TYPE text(tokenizer = 'splitByNonAlpha', dictionary_block_size = 8192) ) ENGINE = MergeTree PARTITION BY toDate(Timestamp) diff --git a/docker/otel-collector/schema/seed/00004_hyperdx_sessions.sql b/docker/otel-collector/schema/seed/00004_hyperdx_sessions.sql index e14258ce0d..9f106e9fed 100644 --- a/docker/otel-collector/schema/seed/00004_hyperdx_sessions.sql +++ b/docker/otel-collector/schema/seed/00004_hyperdx_sessions.sql @@ -20,8 +20,11 @@ CREATE TABLE IF NOT EXISTS ${DATABASE}.hyperdx_sessions `__hdx_materialized_session.id` String MATERIALIZED LogAttributes['session.id'] CODEC(ZSTD(3)), `__hdx_materialized_rum.sessionId` String MATERIALIZED ResourceAttributes['rum.sessionId'] CODEC(ZSTD(3)), INDEX idx_trace_id TraceId TYPE bloom_filter(0.001) GRANULARITY 1, - INDEX idx_session_id __hdx_materialized_session.id TYPE text(tokenizer = 'array'), - INDEX idx_rum_session_id __hdx_materialized_rum.sessionId TYPE text(tokenizer = 'array'), + -- Text indexes use 8,192-token dictionary blocks (the default is 512). Each + -- part's sparse index is 16x smaller, which makes reading filter values + -- with mergeTreeTextIndex() and looking up tokens much cheaper on large parts. + INDEX idx_session_id __hdx_materialized_session.id TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_rum_session_id __hdx_materialized_rum.sessionId TYPE text(tokenizer = 'array', dictionary_block_size = 8192), INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_res_attr_value mapValues(ResourceAttributes) TYPE bloom_filter(0.01) GRANULARITY 1, INDEX idx_scope_attr_key mapKeys(ScopeAttributes) TYPE bloom_filter(0.01) GRANULARITY 1, diff --git a/docker/otel-collector/schema/seed/00005_otel_traces.sql b/docker/otel-collector/schema/seed/00005_otel_traces.sql index b3ad5b3a9e..17d0404347 100644 --- a/docker/otel-collector/schema/seed/00005_otel_traces.sql +++ b/docker/otel-collector/schema/seed/00005_otel_traces.sql @@ -27,14 +27,17 @@ CREATE TABLE IF NOT EXISTS ${DATABASE}.otel_traces `SampleRate` UInt64 MATERIALIZED greatest(toUInt64OrZero(SpanAttributes['SampleRate']), 1) CODEC(T64, ZSTD(3)), `ResourceAttributeItems` Array(String) ALIAS arrayMap((arr) -> concat(arr.1, '=', arr.2), ResourceAttributes::Array(Tuple(String, String))), `SpanAttributeItems` Array(String) ALIAS arrayMap((arr) -> concat(arr.1, '=', arr.2), SpanAttributes::Array(Tuple(String, String))), - INDEX idx_trace_id TraceId TYPE text(tokenizer = 'array'), - INDEX idx_rum_session_id __hdx_materialized_rum.sessionId TYPE text(tokenizer = 'array'), - INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array'), - INDEX idx_span_attr_key mapKeys(SpanAttributes) TYPE text(tokenizer = 'array'), - INDEX idx_span_attr_items SpanAttributeItems TYPE text(tokenizer = 'array'), + -- Text indexes use 8,192-token dictionary blocks (the default is 512). Each + -- part's sparse index is 16x smaller, which makes reading filter values + -- with mergeTreeTextIndex() and looking up tokens much cheaper on large parts. + INDEX idx_trace_id TraceId TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_rum_session_id __hdx_materialized_rum.sessionId TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_res_attr_key mapKeys(ResourceAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_res_attr_items ResourceAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_span_attr_key mapKeys(SpanAttributes) TYPE text(tokenizer = 'array', dictionary_block_size = 8192), + INDEX idx_span_attr_items SpanAttributeItems TYPE text(tokenizer = 'array', dictionary_block_size = 8192), INDEX idx_duration Duration TYPE minmax GRANULARITY 1, - INDEX idx_lower_span_name SpanName TYPE text(tokenizer = 'splitByNonAlpha', preprocessor=lower(SpanName)) + INDEX idx_lower_span_name SpanName TYPE text(tokenizer = 'splitByNonAlpha', preprocessor=lower(SpanName), dictionary_block_size = 8192) ) ENGINE = MergeTree PARTITION BY toDate(Timestamp) diff --git a/packages/app/src/components/DBRowTable.tsx b/packages/app/src/components/DBRowTable.tsx index 51ed633ac2..55ea7676c8 100644 --- a/packages/app/src/components/DBRowTable.tsx +++ b/packages/app/src/components/DBRowTable.tsx @@ -29,7 +29,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 { DENOISE_NOISE_THRESHOLD, DENOISE_SAMPLE_SIZE, @@ -1545,14 +1548,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 55853a7421..3735d787c6 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( @@ -1575,14 +1610,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', diff --git a/packages/common-utils/src/__tests__/metadata.test.ts b/packages/common-utils/src/__tests__/metadata.test.ts index 646957efc6..197d484ff3 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,103 @@ 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('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 +1580,33 @@ describe('Metadata', () => { }, ); + it.each([ + [8192, [10, 10, 5]], + [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 +1927,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 +1978,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 () => { 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 6409a4b7d2..dca6609b70 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,16 @@ 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; +// Tokens a text index values lookup reads per key. The dictionary is sorted, +// so a key with more distinct values than this gets the most frequent among +// its alphabetically first ones. Keep it above "Load more"'s 10,000 values. +const TEXT_INDEX_TOKENS_PER_KEY = 20_000; +// 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 = 4096; + /** * 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 @@ -1745,11 +1756,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, @@ -1757,6 +1794,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; @@ -1765,24 +1804,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, @@ -1792,18 +1818,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 != '' + LIMIT ${{ UNSAFE_RAW_SQL: String(TEXT_INDEX_TOKENS_PER_KEY) }} + )`); + } } const sql = concatChSql(' UNION ALL ', sqlBranches); @@ -1812,14 +1870,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 => @@ -1877,14 +1946,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 + LIMIT ${{ UNSAFE_RAW_SQL: String(TEXT_INDEX_TOKENS_PER_KEY) }} )`; sqlBranches.push(sql); } @@ -1896,12 +1967,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, }) @@ -3054,28 +3122,45 @@ export class Metadata { // fire all the kv fetch queries const promises: Array> = []; - if (mapTextIndexQueryOptions.size > 0) { - promises.push( - this.getMapTextIndexKeyValues({ + // 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. + for (const [column, options] of mapTextIndexQueryOptions) { + const perKey = + (await this.getTextIndexDictionaryBlockSize({ databaseName, tableName, connectionId, - queryOptions: mapTextIndexQueryOptions, - maxParts: mapKeyIndexMaxParts, - partSelection: mapKeyIndexPartSelection, - dateRange, - timestampValueExpression, - signal, - }), - ); + indexName: options.indexName, + })) >= PER_KEY_TEXT_INDEX_MIN_DICTIONARY_BLOCK_SIZE; + 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. *