diff --git a/CHANGELOG.md b/CHANGELOG.md index d559ffc244d..8b6bcaa78a8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -75,6 +75,7 @@ * [ENHANCEMENT] Querier: Use non-pointer HistogramBucket slice in response codec. #7809 * [ENHANCEMENT] Update build image and Go version to 1.27.0. #7814 * [ENHANCEMENT] Querier: Reduce merge iterator `BatchSize` from 12 to 8. #7823 +* [ENHANCEMENT] Upgrade promql-engine to latest. #7841 * [BUGFIX] Querier: Fix queryWithRetry and labelsWithRetry returning (nil, nil) on cancelled context by propagating ctx.Err(). #7370 * [BUGFIX] Metrics Helper: Fix non-deterministic bucket order in merged histograms by sorting buckets after map iteration, matching Prometheus client library behavior. #7380 * [BUGFIX] Distributor: Return HTTP 401 Unauthorized when tenant ID resolution fails in the Prometheus Remote Write 2.0 path. #7389 diff --git a/go.mod b/go.mod index a9d47b17a09..a284ed5ffa9 100644 --- a/go.mod +++ b/go.mod @@ -69,7 +69,7 @@ require ( // Pinned to main: promql-engine has never cut a tag, so a pseudo-version is the only // option. We need thanosengine.Opts.SelectorBatchSize (-querier.selector-batch-size, // #7763) and the logicalplan API that pkg/distributed_execution builds on. - github.com/thanos-io/promql-engine v0.0.0-20260817205454-68fb2c2c5d9b + github.com/thanos-io/promql-engine v0.0.0-20260915065525-1b359e941bde // Pinned to main: promql-engine's api.RemoteEndpoints gained mint/maxt parameters on // Engines(), and thanos/pkg/query implements that interface. The newest tag (v0.42.4, // 146 commits behind main) predates the change, so downgrading breaks the build. diff --git a/go.sum b/go.sum index b2bc42c070b..20635af2cab 100644 --- a/go.sum +++ b/go.sum @@ -1001,8 +1001,8 @@ github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e h1:f1 github.com/thanos-community/galaxycache v0.0.0-20211122094458-3a32041a1f1e/go.mod h1:jXcofnrSln/cLI6/dhlBxPQZEEQHVPCcFaH75M+nSzM= github.com/thanos-io/objstore v0.0.0-20250804093838-71d60dfee488 h1:khBsQLLRoF1KzXgTlwFZa6mC32bwYUUAu/AeP49V7UM= github.com/thanos-io/objstore v0.0.0-20250804093838-71d60dfee488/go.mod h1:uDHLkMKOGDAnlN75EAz8VrRzob1+VbgYSuUleatWuF0= -github.com/thanos-io/promql-engine v0.0.0-20260817205454-68fb2c2c5d9b h1:LuoTi193m3MJx1/bjydfVHvq9A2bJ6nqWmdaXGSl0j8= -github.com/thanos-io/promql-engine v0.0.0-20260817205454-68fb2c2c5d9b/go.mod h1:uzn40oZHPXvfdP498h+MiRL2fN7RF519gNaV3LyhChc= +github.com/thanos-io/promql-engine v0.0.0-20260915065525-1b359e941bde h1:ynBM1tJjqwE/maQCs/5Rk9Y8Ebizc33PbZKS+FW2DYg= +github.com/thanos-io/promql-engine v0.0.0-20260915065525-1b359e941bde/go.mod h1:uzn40oZHPXvfdP498h+MiRL2fN7RF519gNaV3LyhChc= github.com/thanos-io/thanos v0.42.5-0.20260817064418-8908023d563b h1:30lUqQ2HfZ2ni0Ogq3TO0kW45K3iEVhinYDkA0npfXw= github.com/thanos-io/thanos v0.42.5-0.20260817064418-8908023d563b/go.mod h1:zGTlIYCS4Nwn64QbcLQKWAzafbuf370q0UMU5nSraDo= github.com/tinylib/msgp v1.6.1 h1:ESRv8eL3u+DNHUoSAAQRE50Hm162zqAnBoGv9PzScPY= diff --git a/vendor/github.com/thanos-io/promql-engine/engine/engine.go b/vendor/github.com/thanos-io/promql-engine/engine/engine.go index f18a93a85ed..c1ec6d00e9a 100644 --- a/vendor/github.com/thanos-io/promql-engine/engine/engine.go +++ b/vendor/github.com/thanos-io/promql-engine/engine/engine.go @@ -556,6 +556,7 @@ func (q *compatibilityQuery) Exec(ctx context.Context) (ret *promql.Result) { } buf := make([]model.StepVector, q.opts.StepsBatch) + var batchSamples int loop: for { select { @@ -578,6 +579,7 @@ loop: for i := range n { vector := &buf[i] + batchSamples += len(vector.SampleIDs) for j, s := range vector.SampleIDs { if series[s].Floats == nil { series[s].Floats = make([]promql.FPoint, 0, totalSteps) @@ -595,8 +597,14 @@ loop: T: vector.T, H: vector.Histograms[j], }) + batchSamples += telemetry.CalculateHistogramSampleCount(vector.Histograms[j]) } } + q.opts.SampleTracker.Add(batchSamples) + batchSamples = 0 + if err := q.opts.SampleTracker.CheckLimit(); err != nil { + return newErrResult(ret, err) + } } } diff --git a/vendor/github.com/thanos-io/promql-engine/logicalplan/distribute.go b/vendor/github.com/thanos-io/promql-engine/logicalplan/distribute.go index b4baf2529ce..225e7d3bb58 100644 --- a/vendor/github.com/thanos-io/promql-engine/logicalplan/distribute.go +++ b/vendor/github.com/thanos-io/promql-engine/logicalplan/distribute.go @@ -157,6 +157,14 @@ type DistributedExecutionOptimizer struct { SkipDedup bool } +// Optimize distributes a plan in three phases: +// +// 1. Classify each subtree as non-distributive, distributable as-is, or +// distributable through an aggregation, avg, or absent rewrite. A parent +// can absorb only children that are distributable as-is. +// 2. Select the roots of maximal distributable subtrees. These are nodes with +// a distribution strategy whose parent has none. +// 3. Replace each selected subtree with the remote plan for its strategy. func (m DistributedExecutionOptimizer) Optimize(plan Node, opts *query.Options) (Node, annotations.Annotations) { engines := m.Endpoints.Engines(MinMaxTime(plan, opts)) sort.Slice(engines, func(i, j int) bool { @@ -181,38 +189,25 @@ func (m DistributedExecutionOptimizer) Optimize(plan Node, opts *query.Options) warns := annotations.New() parents := computeParents(&plan) - distributionPoints := m.computeDistributionPoints(&plan, parents, engineLabels, warns) + distributionPoints := m.computeDistributionPoints(&plan, engineLabels, warns) - TraverseBottomUp(nil, &plan, func(parent, current *Node) (stop bool) { - if _, distributeNow := distributionPoints[current]; !distributeNow { + TraverseBottomUp(nil, &plan, func(_ *Node, current *Node) (stop bool) { + strategy, distributeNow := distributionPoints[current] + if !distributeNow { return false } - if isAvgAggregation(current) && !preservesPartitionLabels(*current, engineLabels) { - // avg without partition labels: rewrite as sum/count. - *current = m.distributeAvg(*current, engines, m.subqueryOpts(parents, current, opts), labelRanges) - return true - } - - if isAbsent(current) { - *current = m.distributeAbsent(*current, engines, calculateStartOffset(current, opts.LookbackDelta), m.subqueryOpts(parents, current, opts)) - return true - } - - if isAggregation(current) { - if preservesPartitionLabels(*current, engineLabels) { - // Partition-preserving aggregation: push as-is since each engine - // computes over disjoint partition values. - *current = m.distributeQuery(current, engines, m.subqueryOpts(parents, current, opts), labelRanges) - } else { - // Distributive aggregation that drops partition labels: use a - // two-level split with local_agg(remote_agg(X)). - *current = m.distributeAggregation((*current).(*Aggregation), engines, m.subqueryOpts(parents, current, opts), labelRanges) - } - return true + subqueryOpts := m.subqueryOpts(parents, current, opts) + switch strategy { + case rewriteAvg: + *current = m.distributeAvg(*current, engines, subqueryOpts, labelRanges) + case rewriteAbsent: + *current = m.distributeAbsent(*current, engines, calculateStartOffset(current, opts.LookbackDelta), subqueryOpts) + case rewriteAggregation: + *current = m.distributeAggregation((*current).(*Aggregation), engines, subqueryOpts, labelRanges) + case distributeAsIs: + *current = m.distributeQuery(current, engines, subqueryOpts, labelRanges) } - - *current = m.distributeQuery(current, engines, m.subqueryOpts(parents, current, opts), labelRanges) return true }) return plan, *warns @@ -243,93 +238,76 @@ func computeParents(plan *Node) map[*Node]*Node { return parents } -func (m DistributedExecutionOptimizer) computeDistributionPoints(plan *Node, parents map[*Node]*Node, engineLabels map[string]struct{}, warns *annotations.Annotations) map[*Node]struct{} { - marks := make(map[*Node]struct{}) +// A rewrite strategy can absorb only children that distribute as-is. +type distributionStrategy uint8 - // First pass: mark distribution points (aggregations, absent functions). - Traverse(plan, func(current *Node) { - // Skip subtrees that are already distributed (e.g. by a previous - // distributed optimizer). This lets multiple distributed optimizers - // be chained: once the plan is distributed, subsequent optimizers - // fall through instead of re-distributing. - if isDistributed(current) { - return - } - if isAbsent(current) { - if m.isDistributive(current, engineLabels, warns) { - marks[current] = struct{}{} - } - return - } - if isAggregation(current) { - // Non-distributive aggregations that don't preserve partition labels - // cannot be distributed, except for avg which gets rewritten as sum/count. - if !m.isDistributive(current, engineLabels, warns) { - if isAvgAggregation(current) { - marks[current] = struct{}{} - } - return - } - // Distributive aggregations (standard or partition-preserving): - // defer to ancestor if possible. - if preservesPartitionLabels(*current, engineLabels) { - if m.hasDistributiveAncestor(parents, current, engineLabels, warns) { - return - } - } - marks[current] = struct{}{} - } - }) +const ( + cannotDistribute distributionStrategy = iota + distributeAsIs + rewriteAggregation + rewriteAvg + rewriteAbsent +) - // Second pass: for nodes whose siblings have marks, mark them too so both - // sides of a binary expression get distributed. - Traverse(plan, func(current *Node) { - if _, ok := marks[current]; ok { - return - } - if isDistributed(current) { - return - } - if subtreeHasMark(current, marks) { - return - } - if !m.isDistributive(current, engineLabels, warns) { - return +func (m DistributedExecutionOptimizer) computeDistributionPoints(plan *Node, engineLabels map[string]struct{}, warns *annotations.Annotations) map[*Node]distributionStrategy { + strategies := make(map[*Node]distributionStrategy) + m.classifyDistribution(plan, strategies, engineLabels, warns) + + // Select roots of maximal distributable subtrees. + points := make(map[*Node]distributionStrategy) + TraverseBottomUp(nil, plan, func(parent, current *Node) bool { + strategy := strategies[current] + if strategy == cannotDistribute || IsConstantExpr(*current) { + return false } - parent := parents[current] - if parent != nil && (m.isDistributive(parent, engineLabels, warns) || isAvgAggregation(parent)) { - if !subtreeHasMark(parent, marks) { - return - } + if parent != nil && strategies[parent] != cannotDistribute { + return false } - marks[current] = struct{}{} + points[current] = strategy + return false }) - - return marks + return points } -// isDistributed reports whether the subtree rooted at node has already been -// processed by a distributed optimizer, i.e. it contains a Deduplicate, -// RemoteExecution or Noop node (Noop is the terminal result of distributing a -// subtree that matched no engines). Such subtrees must not be distributed again. -func isDistributed(node *Node) bool { +// classifyDistribution records how each subtree can be distributed. +func (m DistributedExecutionOptimizer) classifyDistribution(node *Node, strategies map[*Node]distributionStrategy, engineLabels map[string]struct{}, warns *annotations.Annotations) distributionStrategy { switch (*node).(type) { case RemoteMerge, RemoteExecution, Noop: - return true + strategies[node] = cannotDistribute + return cannotDistribute } - return slices.ContainsFunc((*node).Children(), isDistributed) -} -func subtreeHasMark(node *Node, marks map[*Node]struct{}) bool { + childrenCanPushDown := true for _, child := range (*node).Children() { - if _, ok := marks[child]; ok { - return true + if m.classifyDistribution(child, strategies, engineLabels, warns) != distributeAsIs { + childrenCanPushDown = false } - if subtreeHasMark(child, marks) { - return true + } + + distributive := m.isDistributiveOperation(node, engineLabels, warns) + strategy := cannotDistribute + if childrenCanPushDown { + switch { + case isAbsent(node): + if distributive { + strategy = rewriteAbsent + } + case isAvgAggregation(node) && !preservesPartitionLabels(*node, engineLabels): + strategy = rewriteAvg + case isAggregation(node): + if distributive { + strategy = distributeAsIs + if !preservesPartitionLabels(*node, engineLabels) { + strategy = rewriteAggregation + } + } + case distributive: + strategy = distributeAsIs } } - return false + + strategies[node] = strategy + return strategy } func (m DistributedExecutionOptimizer) subqueryOpts(parents map[*Node]*Node, current *Node, opts *query.Options) *query.Options { @@ -722,7 +700,7 @@ func preservesPartitionLabels(expr Node, partitionLabels map[string]struct{}) bo } } -func (m DistributedExecutionOptimizer) isDistributive(expr *Node, engineLabels map[string]struct{}, warns *annotations.Annotations) bool { +func (m DistributedExecutionOptimizer) isDistributiveOperation(expr *Node, engineLabels map[string]struct{}, warns *annotations.Annotations) bool { if expr == nil { return false } @@ -734,10 +712,7 @@ func (m DistributedExecutionOptimizer) isDistributive(expr *Node, engineLabels m if isBinaryExpressionWithOneScalarSide(e) { return true } - return !m.SkipBinaryPushdown && - isBinaryExpressionWithDistributableMatching(e, engineLabels) && - m.isDistributive(&e.LHS, engineLabels, warns) && - m.isDistributive(&e.RHS, engineLabels, warns) + return !m.SkipBinaryPushdown && isBinaryExpressionWithDistributableMatching(e, engineLabels) case *Aggregation: switch e.Op { // Mathematically distributive: can be split into local_agg(remote_agg(X)) @@ -927,23 +902,6 @@ func matchesExternalLabels(ms []*labels.Matcher, externalLabels labels.Labels) b return true } -// hasDistributiveAncestor checks if there's a distributive node somewhere up the -// parent chain from the current node that can handle distribution. -// We must have an unbroken chain of distributive nodes to the ancestor for it to -// be able to handle distribution on our behalf. -func (m DistributedExecutionOptimizer) hasDistributiveAncestor(parents map[*Node]*Node, current *Node, engineLabels map[string]struct{}, warns *annotations.Annotations) bool { - for p := parents[current]; p != nil; p = parents[p] { - if !m.isDistributive(p, engineLabels, warns) { - // We hit a non-distributive node, so we can't push through it. - // No ancestor can help us distribute. - return false - } - } - // All ancestors are distributive, so the root (or the point where we - // stop traversing) can handle distribution. - return parents[current] != nil -} - func maxTime(a, b time.Time) time.Time { if a.After(b) { return a diff --git a/vendor/github.com/thanos-io/promql-engine/logicalplan/exprutil.go b/vendor/github.com/thanos-io/promql-engine/logicalplan/exprutil.go index 7cb4a8d4674..93ccd979245 100644 --- a/vendor/github.com/thanos-io/promql-engine/logicalplan/exprutil.go +++ b/vendor/github.com/thanos-io/promql-engine/logicalplan/exprutil.go @@ -63,6 +63,8 @@ func IsConstantExpr(expr Node) bool { return IsConstantExpr(texpr.Expr) case *Parens: return IsConstantExpr(texpr.Expr) + case *Unary: + return IsConstantExpr(texpr.Expr) case *FunctionCall: constArgs := true for _, arg := range texpr.Args { diff --git a/vendor/github.com/thanos-io/promql-engine/storage/prometheus/vector_selector.go b/vendor/github.com/thanos-io/promql-engine/storage/prometheus/vector_selector.go index 915c06c3b4a..2f621a6d073 100644 --- a/vendor/github.com/thanos-io/promql-engine/storage/prometheus/vector_selector.go +++ b/vendor/github.com/thanos-io/promql-engine/storage/prometheus/vector_selector.go @@ -231,14 +231,16 @@ func (o *vectorSelector) loadSeries(ctx context.Context) error { } func (o *vectorSelector) updateSampleTracker(totalSamples int) error { - if o.lastTrackedSamples > 0 { - o.opts.SampleTracker.Remove(o.lastTrackedSamples) - } - if totalSamples > 0 { - o.opts.SampleTracker.Add(totalSamples) + delta := totalSamples - o.lastTrackedSamples + if delta > 0 { + o.opts.SampleTracker.Add(delta) + o.lastTrackedSamples = totalSamples + return o.opts.SampleTracker.CheckLimit() + } else if delta < 0 { + o.opts.SampleTracker.Remove(-delta) } o.lastTrackedSamples = totalSamples - return o.opts.SampleTracker.CheckLimit() + return nil } func (o *vectorSelector) shouldCheckSampleLimit(fromSeries int64) bool { diff --git a/vendor/modules.txt b/vendor/modules.txt index 2b7c63658bb..ef00cbd5292 100644 --- a/vendor/modules.txt +++ b/vendor/modules.txt @@ -1291,7 +1291,7 @@ github.com/thanos-io/objstore/providers/oss github.com/thanos-io/objstore/providers/s3 github.com/thanos-io/objstore/providers/swift github.com/thanos-io/objstore/tracing/opentracing -# github.com/thanos-io/promql-engine v0.0.0-20260817205454-68fb2c2c5d9b +# github.com/thanos-io/promql-engine v0.0.0-20260915065525-1b359e941bde ## explicit; go 1.24.0 github.com/thanos-io/promql-engine/api github.com/thanos-io/promql-engine/compute