[Query-Frontend] Add dynamic query vertical sharding #6678

afhassan · 2025-03-28T17:34:37Z

What this PR does:

Refactor dynamic query splitting to reduce complexity
- The main source of complexity in the previous implementation was the variability of the first split duration compared to other subsequent splits when calculating the total duration of data that will be fetched from storage by the query.
- The solution is to ignore the first interval when calculating the duration fetched. This means that the duration fetched becomes an estimate, but the significance of ignoring the first interval is small compared to the reduced complexity.
Add dynamic query vertical sharding
- The goal of dynamic query splitting is to protect store-gateways by limiting the number of read requests sent per query, while providing enough parallelism for the query to execute efficiently
- In some scenarios, a smaller vertical shard size allows more horizontal splitting while keeping store-gateways protected. This allows more efficient query execution.
- The new change calculates the most optimal combination of vertical sharding and horizontal spliting to result in the largest number of total shards (without exceeding any of the configured limits)

Example query sum(rate(metric[1h])) by (pod) with 100 day range and the following configurations:

max_shards_per_query: 100
max_fetched_data_duration_per_query: 6000h // 250 day
query_vertical_shard_size: 3

Option one: 1 vertical shard (no vertical sharding)

1 day split interval would result in 100 splits (100 shards) and 200 days total duration fetched

Option two: 2 vertical shards

5 day split interval would result in 20 splits (40 shards) and 240 days total duration fetched

Option three: 3 vertical shards

100 day split interval would result in no horizontal splitting (3 shards) and 303 days total duration fetched

Previously, we would always choose option three. After the change, in this scenario we would choose option one to not sharding vertically to allow 100 horizontal splits.

Checklist

Tests updated
Documentation added
CHANGELOG.md updated - the order of entries should be [CHANGE], [FEATURE], [ENHANCEMENT], [BUGFIX]

Signed-off-by: Ahmed Hassan <[email protected]>

justinjung04 · 2025-04-14T14:55:31Z

pkg/querier/tripperware/queryrange/split_by_interval_test.go

-			expectedDurationFetchedBySelectors:     4 * day,
-			expectedDurationFetchedByLookbackDelta: 0 * day,
+			expectedDurationFetchedByRange:     40 * day,
+			expectedDurationFetchedBySelectors: 4 * day,
 		},


Should we add a test about the edge case you described in the PR description?

I added separate test cases for dynamic splitting + vertical sharding scenarios. I also added an example in the PR description to explain the change better.

Signed-off-by: Ahmed Hassan <[email protected]>

afhassan · 2025-04-17T22:59:01Z

pkg/querier/tripperware/queryrange/split_by_interval.go

+		for candidateVerticalShardSize := 1; candidateVerticalShardSize <= maxVerticalShardSize; candidateVerticalShardSize++ {
+			maxSplitsFromMaxShards := getMaxSplitsFromMaxQueryShards(dynamicSplitCfg.MaxShardsPerQuery, candidateVerticalShardSize)
+			maxSplitsFromDurationFetched := getMaxSplitsFromDurationFetched(dynamicSplitCfg.MaxFetchedDataDurationPerQuery, candidateVerticalShardSize, queryExpr, r.GetStart(), r.GetEnd(), r.GetStep(), baseInterval, lookbackDelta)
+
+			// Use the more restrictive max splits limit
+			var maxSplits int
+			switch {
+			case dynamicSplitCfg.MaxShardsPerQuery > 0 && dynamicSplitCfg.MaxFetchedDataDurationPerQuery > 0:
+				maxSplits = min(maxSplitsFromMaxShards, maxSplitsFromDurationFetched)
+			case dynamicSplitCfg.MaxShardsPerQuery > 0:
+				maxSplits = maxSplitsFromMaxShards
+			case dynamicSplitCfg.MaxFetchedDataDurationPerQuery > 0:
+				maxSplits = maxSplitsFromDurationFetched
+			}
+
+			candidateInterval := getIntervalFromMaxSplits(r, baseInterval, maxSplits)
+			if candidateTotalShards := getExpectedTotalShards(r.GetStart(), r.GetEnd(), candidateInterval, candidateVerticalShardSize); candidateTotalShards > totalShards {
+				interval = candidateInterval
+				verticalShardSize = candidateVerticalShardSize
+				totalShards = candidateTotalShards
+			}
+		}


This is the main change to introduce dynamic vertical sharding. We do the same calculations to determine the split interval, but now inside a loop that goes over vertical sharding from 1 to max.

The best vertical sharding value is the one that is expected to result in the most number of total shards, which is checked by getExpectedTotalShards().

afhassan · 2025-04-17T23:00:24Z

pkg/querier/tripperware/queryrange/split_by_interval.go

+
+		// Set number of vertical shards to be used in shard_by middleware
+		if isShardable && maxVerticalShardSize > 1 {
+			ctx = tripperware.InjectVerticalShardSizeToContext(ctx, verticalShardSize)


The chosen value for vertical sharding is injected to context here, and then used in shard_by.go middlware

afhassan · 2025-04-17T23:01:40Z

pkg/querier/tripperware/queryrange/split_by_interval.go

-// Return the fixed base duration fetched by the query regardless of the number of splits, and the duration that is fetched once for every split
-func getDurationFetchedByQuerySplitting(expr parser.Expr, queryStart int64, queryEnd int64, queryStep int64, baseInterval time.Duration, lookbackDelta time.Duration) (fixedDurationFetched time.Duration, perSplitDurationFetched time.Duration) {
-	// First analyze the query using original start-end time. Duration fetched by lookbackDelta here only reflects the start time of first split
-	durationFetchedByRange, durationFetchedBySelectors, durationFetchedByLookbackDeltaFirstSplit := analyzeDurationFetchedByQueryExpr(expr, queryStart, queryEnd, baseInterval, lookbackDelta)
-
-	fixedDurationFetched += durationFetchedByRange        // Duration fetched by the query range is constant regardless of how many splits the query has
-	perSplitDurationFetched += durationFetchedBySelectors // Duration fetched by selectors is fetched once for every query split
-
-	// Next analyze the query using the next split start time to find the duration fetched by lookbackDelta for splits other than first one
-	nextIntervalStart := nextIntervalBoundary(queryStart, queryStep, baseInterval) + queryStep
-	_, _, durationFetchedByLookbackDeltaOtherSplits := analyzeDurationFetchedByQueryExpr(expr, nextIntervalStart, queryEnd, baseInterval, lookbackDelta)
-
-	// Handle different cases for lookbackDelta
-	if durationFetchedByLookbackDeltaFirstSplit > 0 && durationFetchedByLookbackDeltaOtherSplits > 0 {
-		// lookbackDelta is fetching additional duration for all splits
-		perSplitDurationFetched += durationFetchedByLookbackDeltaOtherSplits
-	} else if durationFetchedByLookbackDeltaOtherSplits > 0 {
-		// lookbackDelta is fetching additional duration for all splits except first one
-		perSplitDurationFetched += durationFetchedByLookbackDeltaOtherSplits
-		fixedDurationFetched -= durationFetchedByLookbackDeltaOtherSplits
-	} else if durationFetchedByLookbackDeltaFirstSplit > 0 {
-		// lookbackDelta is fetching additional duration for first split only
-		fixedDurationFetched += durationFetchedByLookbackDeltaFirstSplit
-	}
-
-	return fixedDurationFetched, perSplitDurationFetched
-}
-


To simplify the calculation for expected duration fetched by query, we no longer calculate lookbackDelta separate from matrix selectors, which makes this function obsolete.

afhassan · 2025-04-17T23:05:50Z

pkg/querier/tripperware/queryrange/split_by_interval.go

-			// Adjust start and end time based on matrix selectors and/or subquery selector and increment total duration fetched, this excludes lookbackDelta
-			start, end := util.GetTimeRangesForSelector(queryStart, queryEnd, 0, n, path, evalRange)
-			startIntervalIndex := floorDiv(start, baseIntervalMillis)
-			endIntervalIndex := floorDiv(end, baseIntervalMillis)
-			totalDurationFetchedCount += int(endIntervalIndex-startIntervalIndex) + 1
-
-			// Increment duration fetched by lookbackDelta
-			startLookbackDelta := start - util.DurationMilliseconds(lookbackDelta)
-			startLookbackDeltaIntervalIndex := floorDiv(startLookbackDelta, baseIntervalMillis)
-			if evalRange == 0 && startLookbackDeltaIntervalIndex < startIntervalIndex {
-				durationFetchedByLookbackDeltaCount += int(startIntervalIndex - startLookbackDeltaIntervalIndex)
-			}
+			// Adjust start time based on matrix selectors and/or subquery selectors and calculate additional lookback duration fetched
+			start, end := util.GetTimeRangesForSelector(queryStart, queryEnd, lookbackDelta, n, path, evalRange)
+			durationFetchedBySelectors := (end - start) - (queryEnd - queryStart)
+			durationFetchedBySelectorsCount += int(ceilDiv(durationFetchedBySelectors, baseIntervalMillis))
+


Also simplified because we no longer need to calculate duration fetched by lookbackDelta separately. It is all included in durationFetchedBySelectors.

justinjung04

Nice, thank you

refactor dynamic query splitting to reduce complexity

981393b

Signed-off-by: Ahmed Hassan <[email protected]>

pull-request-size bot added the size/L label Mar 28, 2025

add query dynamic vertical sharding

a98c452

Signed-off-by: Ahmed Hassan <[email protected]>

afhassan changed the title ~~[Query-Frontend] Refactor dynamic query splitting to reduce complexity~~ [Query-Frontend] Add dynamic query vertical sharding Apr 3, 2025

afhassan and others added 4 commits April 3, 2025 17:46

fix unit tests for dynamic vertical sharding

2ea7f0b

Signed-off-by: Ahmed Hassan <[email protected]>

Merge branch 'cortexproject:master' into dynamic-query-vertical-sharding

7ea1b6f

store dynamic vertical shard size in context

447f6d3

Signed-off-by: Ahmed Hassan <[email protected]>

refactoring dynamicIntervalFn

df26e9a

Signed-off-by: Ahmed Hassan <[email protected]>

afhassan marked this pull request as ready for review April 10, 2025 01:53

dosubot bot added the component/query-frontend label Apr 10, 2025

justinjung04 reviewed Apr 14, 2025

View reviewed changes

add separate vertical sharding tests

a97c205

Signed-off-by: Ahmed Hassan <[email protected]>

pull-request-size bot added size/XL and removed size/L labels Apr 17, 2025

fix formatting

cda7751

Signed-off-by: Ahmed Hassan <[email protected]>

afhassan commented Apr 17, 2025

View reviewed changes

justinjung04 approved these changes Apr 22, 2025

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Query-Frontend] Add dynamic query vertical sharding #6678

[Query-Frontend] Add dynamic query vertical sharding #6678

afhassan commented Mar 28, 2025 •

edited

Loading

justinjung04 Apr 14, 2025

afhassan Apr 17, 2025

afhassan Apr 17, 2025

afhassan Apr 17, 2025

afhassan Apr 17, 2025 •

edited

Loading

afhassan Apr 17, 2025

justinjung04 left a comment

[Query-Frontend] Add dynamic query vertical sharding #6678

Are you sure you want to change the base?

[Query-Frontend] Add dynamic query vertical sharding #6678

Conversation

afhassan commented Mar 28, 2025 • edited Loading

justinjung04 Apr 14, 2025

Choose a reason for hiding this comment

afhassan Apr 17, 2025

Choose a reason for hiding this comment

afhassan Apr 17, 2025

Choose a reason for hiding this comment

afhassan Apr 17, 2025

Choose a reason for hiding this comment

afhassan Apr 17, 2025 • edited Loading

Choose a reason for hiding this comment

afhassan Apr 17, 2025

Choose a reason for hiding this comment

justinjung04 left a comment

Choose a reason for hiding this comment

afhassan commented Mar 28, 2025 •

edited

Loading

afhassan Apr 17, 2025 •

edited

Loading