Skip to content

Commit

Permalink
MQE: Handle min/max NaN's same as promql (#9212)
Browse files Browse the repository at this point in the history
* MQE: Handle min/max NaN's same as promql

* Update CHANGELOG

* Add comment
  • Loading branch information
jhesketh authored Sep 6, 2024
1 parent 1a4cb60 commit 36b193c
Show file tree
Hide file tree
Showing 3 changed files with 26 additions and 6 deletions.
2 changes: 1 addition & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@
* [CHANGE] Querier: allow wrapping errors with context errors only when the former actually correspond to `context.Canceled` and `context.DeadlineExceeded`. #9175
* [FEATURE] Alertmanager: Added `-alertmanager.log-parsing-label-matchers` to control logging when parsing label matchers. This flag is intended to be used with `-alertmanager.utf8-strict-mode-enabled` to validate UTF-8 strict mode is working as intended. The default value is `false`. #9173
* [FEATURE] Alertmanager: Added `-alertmanager.utf8-migration-logging-enabled` to enable logging of tenant configurations that are incompatible with UTF-8 strict mode. The default value is `false`. #9174
* [FEATURE] Querier: add experimental streaming PromQL engine, enabled with `-querier.query-engine=mimir`. #8422 #8430 #8454 #8455 #8360 #8490 #8508 #8577 #8660 #8671 #8677 #8747 #8850 #8872 #8838 #8911 #8909 #8923 #8924 #8925 #8932 #8933 #8934 #8962 #8986 #8993 #8995 #9008 #9017 #9018 #9019 #9120 #9121 #9136 #9139 #9140 #9145 #9191 #9194 #9196
* [FEATURE] Querier: add experimental streaming PromQL engine, enabled with `-querier.query-engine=mimir`. #8422 #8430 #8454 #8455 #8360 #8490 #8508 #8577 #8660 #8671 #8677 #8747 #8850 #8872 #8838 #8911 #8909 #8923 #8924 #8925 #8932 #8933 #8934 #8962 #8986 #8993 #8995 #9008 #9017 #9018 #9019 #9120 #9121 #9136 #9139 #9140 #9145 #9191 #9194 #9196 #9212
* [FEATURE] Experimental Kafka-based ingest storage. #6888 #6894 #6929 #6940 #6951 #6974 #6982 #7029 #7030 #7091 #7142 #7147 #7148 #7153 #7160 #7193 #7349 #7376 #7388 #7391 #7393 #7394 #7402 #7404 #7423 #7424 #7437 #7486 #7503 #7508 #7540 #7621 #7682 #7685 #7694 #7695 #7696 #7697 #7701 #7733 #7734 #7741 #7752 #7838 #7851 #7871 #7877 #7880 #7882 #7887 #7891 #7925 #7955 #7967 #8031 #8063 #8077 #8088 #8135 #8176 #8184 #8194 #8216 #8217 #8222 #8233 #8503 #8542 #8579 #8657 #8686 #8688 #8703 #8706 #8708 #8738 #8750 #8778 #8808 #8809 #8841 #8842 #8845 #8853 #8886 #8988
* What it is:
* When the new ingest storage architecture is enabled, distributors write incoming write requests to a Kafka-compatible backend, and the ingesters asynchronously replay ingested data from Kafka. In this architecture, the write and read path are de-coupled through a Kafka-compatible backend. The write path and Kafka load is a function of the incoming write traffic, the read path load is a function of received queries. Whatever the load on the read path, it doesn't affect the write path.
Expand Down
9 changes: 4 additions & 5 deletions pkg/streamingpromql/aggregations/min_max.go
Original file line number Diff line number Diff line change
Expand Up @@ -34,14 +34,16 @@ func NewMinMaxAggregationGroup(max bool) *MinMaxAggregationGroup {
}

func (g *MinMaxAggregationGroup) maxAccumulatePoint(idx int64, f float64) {
if !g.floatPresent[idx] || g.floatPresent[idx] && f > g.floatValues[idx] {
// We return a NaN only if there are no other values to return
if !g.floatPresent[idx] || f > g.floatValues[idx] || math.IsNaN(g.floatValues[idx]) {
g.floatValues[idx] = f
g.floatPresent[idx] = true
}
}

func (g *MinMaxAggregationGroup) minAccumulatePoint(idx int64, f float64) {
if !g.floatPresent[idx] || g.floatPresent[idx] && f < g.floatValues[idx] {
// We return a NaN only if there are no other values to return
if !g.floatPresent[idx] || f < g.floatValues[idx] || math.IsNaN(g.floatValues[idx]) {
g.floatValues[idx] = f
g.floatPresent[idx] = true
}
Expand All @@ -68,9 +70,6 @@ func (g *MinMaxAggregationGroup) AccumulateSeries(data types.InstantVectorSeries
}

for _, p := range data.Floats {
if math.IsNaN(p.F) {
continue
}
idx := (p.T - start) / interval
g.accumulatePoint(idx, p.F)
}
Expand Down
21 changes: 21 additions & 0 deletions pkg/streamingpromql/testdata/ours/aggregators.test
Original file line number Diff line number Diff line change
Expand Up @@ -191,6 +191,27 @@ eval range from 0 to 2m step 1m min(histogram_only_series)

clear

# Test min/max with NaN
load 1m
mixed_series{type="float", label="value"} NaN
mixed_series{type="float", label="value2"} -5
mixed_series{type="float", label="value3"} 10
mixed_series{type="histogram", label="value"} {{schema:1 sum:5 count:5 buckets:[1 3 1]}}
mixed_series{type="histogram", label="value2"} NaN
mixed_series{type="onlyNaN"} NaN

eval instant at 1m min by (type) (mixed_series)
{type="float"} -5
{type="histogram"} 0
{type="onlyNaN"} NaN

eval instant at 1m max by (type) (mixed_series)
{type="float"} 10
{type="histogram"} 0
{type="onlyNaN"} NaN

clear

# Test native histogram compaction
load 1m
single_histogram{label="value"} {{schema:1 sum:5 count:5 buckets:[1 3 1 4 0 2]}}
Expand Down

0 comments on commit 36b193c

Please sign in to comment.