WHERE fields must be decoded during aggregates

This change ensures that if there are any fields in the WHERE clause of
an aggregate that are different from the fields in the SELECT clause,
that the cursors also decode those fields. Otherwise WHERE clauses of
the form 'SELECT f(w) FROM x WHERE y=z' will return incorrect results

Fixes issue #4701.

CHANGELOG.md

@@ -54,6 +54,7 @@ There are breaking changes in this release:
 - [#4685](https://github.com/influxdb/influxdb/pull/4685): Automatically promote node to raft peer if drop server results in removing a raft peer.
 
 ### Bugfixes
+- [#4789](https://github.com/influxdb/influxdb/pull/4789): Decode WHERE fields during aggregates. Fix [issue #4701](https://github.com/influxdb/influxdb/issues/4701).
 - [#4778](https://github.com/influxdb/influxdb/pull/4778): If there are no points to count, count is 0. Fixes [#4701](https://github.com/influxdb/influxdb/issues/4701)
 - [#4715](https://github.com/influxdb/influxdb/pull/4715): Fix panic during Raft-close. Fix [issue #4707](https://github.com/influxdb/influxdb/issues/4707). Thanks @oiooj
 - [#4643](https://github.com/influxdb/influxdb/pull/4643): Fix panic during backup restoration. Thanks @oiooj

cmd/influxd/run/server_test.go

@@ -1131,7 +1131,12 @@ func TestServer_Query_Count(t *testing.T) {
 	now := now()
 
 	test := NewTest("db0", "rp0")
-	test.write = `cpu,host=server01 value=1.0 ` + strconv.FormatInt(now.UnixNano(), 10)
+	writes := []string{
+		`cpu,host=server01 value=1.0 ` + strconv.FormatInt(now.UnixNano(), 10),
+		`ram value1=1.0,value2=2.0 ` + strconv.FormatInt(now.UnixNano(), 10),
+	}
+
+	test.write = strings.Join(writes, "\n")
 
 	hour_ago := now.Add(-time.Hour).UTC()
 
@@ -1151,6 +1156,16 @@ func TestServer_Query_Count(t *testing.T) {
 			command: fmt.Sprintf(`SELECT count(value) FROM db0.rp0.cpu WHERE value=100 AND time >= '%s'`, hour_ago.Format(time.RFC3339Nano)),
 			exp:     fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["%s",0]]}]}]}`, hour_ago.Format(time.RFC3339Nano)),
 		},
+		&Query{
+			name:    "selecting count(value1) with matching filter against value2 should return correct result",
+			command: fmt.Sprintf(`SELECT count(value1) FROM db0.rp0.ram WHERE value2=2 AND time >= '%s'`, hour_ago.Format(time.RFC3339Nano)),
+			exp:     fmt.Sprintf(`{"results":[{"series":[{"name":"ram","columns":["time","count"],"values":[["%s",1]]}]}]}`, hour_ago.Format(time.RFC3339Nano)),
+		},
+		&Query{
+			name:    "selecting count(value1) with non-matching filter against value2 should return correct result",
+			command: fmt.Sprintf(`SELECT count(value1) FROM db0.rp0.ram WHERE value2=3 AND time >= '%s'`, hour_ago.Format(time.RFC3339Nano)),
+			exp:     fmt.Sprintf(`{"results":[{"series":[{"name":"ram","columns":["time","count"],"values":[["%s",0]]}]}]}`, hour_ago.Format(time.RFC3339Nano)),
+		},
 		&Query{
 			name:    "selecting count(*) should error",
 			command: `SELECT count(*) FROM db0.rp0.cpu`,

tsdb/aggregate.go

@@ -649,7 +649,7 @@ func (m *AggregateMapper) openMeasurement(mm *Measurement) error {
 		}
 
 		for i, key := range t.SeriesKeys {
-			fields := slices.Union(selectFields, m.fieldNames, false)
+			fields := slices.Union(slices.Union(selectFields, m.fieldNames, false), m.whereFields, false)
 			c := m.tx.Cursor(key, fields, m.shard.FieldCodec(mm.Name), true)
 			if c == nil {
 				continue