diff --git a/Makefile b/Makefile index 62c280f..110c514 100644 --- a/Makefile +++ b/Makefile @@ -19,7 +19,7 @@ VERSION := $(shell git describe --tags || echo "v0.0.0") VER_CUT := $(shell echo $(VERSION) | cut -c2-) # Dependency versions -GOLANGCI_VERSION = latest +GOLANGCI_VERSION = v2.12.1 MODEL_GARAGE_VERSION = $(shell go list -m -f '{{.Version}}' github.com/DIMO-Network/model-garage) help: @echo "\nSpecify a subcommand:\n" @@ -89,7 +89,7 @@ generate-go: ## Generate go code. tools-golangci-lint: ## install golangci-lint tool @mkdir -p $(PATHINSTBIN) - curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s -- -b $(PATHINSTBIN) $(GOLANGCI_VERSION) + curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/$(GOLANGCI_VERSION)/install.sh | sh -s -- -b $(PATHINSTBIN) $(GOLANGCI_VERSION) tools: tools-golangci-lint ## Install all tools required for development. \ No newline at end of file diff --git a/internal/service/ch/ch_test.go b/internal/service/ch/ch_test.go index c2aa3dc..3044007 100644 --- a/internal/service/ch/ch_test.go +++ b/internal/service/ch/ch_test.go @@ -938,6 +938,36 @@ func (c *CHServiceTestSuite) TestGetLatestSignal() { } } +// TestLatestQueriesUseProjection asserts that every branch of the latest-signals +// query family can be served by the signal_latest_by_subject_source_name +// projection. A branch that falls back to the base table scans the subject's +// entire signal history, so its cost grows with history depth instead of with +// the request. force_optimize_projection makes ClickHouse reject any query the +// projection matcher cannot serve. +func (c *CHServiceTestSuite) TestLatestQueriesUseProjection() { + lastSeenStmt, lastSeenArgs := getLastSeenQuery(testSubject1, &model.SignalArgs{}) + nonLocStmt, nonLocArgs := getLatestNonLocationQuery(testSubject1, []string{vss.FieldSpeed}, nil) + locStmt, locArgs := getLatestLocationQuery(testSubject1, []string{vss.FieldCurrentLocationCoordinates}, nil) + + testCases := []struct { + name string + stmt string + args []any + }{ + {name: "lastSeen", stmt: lastSeenStmt, args: lastSeenArgs}, + {name: "nonLocation", stmt: nonLocStmt, args: nonLocArgs}, + {name: "location", stmt: locStmt, args: locArgs}, + } + for _, tc := range testCases { + c.Run(tc.name, func() { + stmt := strings.TrimSuffix(tc.stmt, ";") + " SETTINGS force_optimize_projection = 1" + rows, err := c.chService.conn.Query(context.Background(), stmt, tc.args...) + c.Require().NoError(err, "query is not served by a projection:\n%s", tc.stmt) + c.Require().NoError(rows.Close()) + }) + } +} + func (c *CHServiceTestSuite) TestGetAvailableSignals() { ctx := context.Background() c.Run("has signals", func() { diff --git a/internal/service/ch/queries.go b/internal/service/ch/queries.go index 5f7d692..51f87b2 100644 --- a/internal/service/ch/queries.go +++ b/internal/service/ch/queries.go @@ -422,35 +422,39 @@ func getAllLatestQuery(subject string, filter *model.SignalFilter) (string, []an return newQuery(mods...) } -// GetLastSeenQuery creates a query to get the last seen timestamp of any signal. +// getLastSeenQuery creates a query to get the last seen timestamp of any signal. // returns the query statement and the arguments list, +// +// The inner query aggregates at the same grain as the +// signal_latest_by_subject_source_name projection (GROUP BY name with subject +// pinned) so the projection matcher serves it from pre-aggregated rows. A flat +// max(timestamp) with no GROUP BY is not reliably matched and falls back to +// scanning the subject's full history. /* SELECT 'lastSeen' AS name, - max(timestamp) AS ts, + max(ts) AS ts, NULL AS value_number, NULL AS value_string, CAST(tuple(0, 0, 0, 0), 'Tuple(latitude Float64, longitude Float64, hdop Float64, heading Float64)') AS value_location FROM - signal -WHERE - subject = '...' + (SELECT max(timestamp) AS ts FROM signal WHERE subject = '...' GROUP BY name) */ func getLastSeenQuery(subject string, sigArgs *model.SignalArgs) (string, []any) { if sigArgs == nil { return "", nil } - mods := []qm.QueryMod{ - qm.Select(lastSeenName), + innerMods := []qm.QueryMod{ qm.Select(lastSeenTS), - qm.Select(numValAsNull), - qm.Select(strValAsNull), - qm.Select(locValAsZero), qm.From(vss.TableName), qm.Where(subjectWhere, subject), + qm.GroupBy(vss.NameCol), } - mods = append(mods, getFilterMods(sigArgs.Filter)...) - return newQuery(mods...) + innerMods = append(innerMods, getFilterMods(sigArgs.Filter)...) + innerStmt, args := newQuery(innerMods...) + stmt := "SELECT " + lastSeenName + ", max(ts) AS ts, " + numValAsNull + ", " + strValAsNull + ", " + locValAsZero + + " FROM (" + strings.TrimSuffix(innerStmt, ";") + ");" + return stmt, args } // unionAll creates a UNION ALL statement from the given statements and arguments. diff --git a/internal/service/ch/queries_test.go b/internal/service/ch/queries_test.go index f86e078..fa3c334 100644 --- a/internal/service/ch/queries_test.go +++ b/internal/service/ch/queries_test.go @@ -3,6 +3,7 @@ package ch import ( "testing" + "github.com/DIMO-Network/telemetry-api/internal/graph/model" "github.com/aarondl/sqlboiler/v4/queries/qm" "github.com/stretchr/testify/assert" ) @@ -32,3 +33,31 @@ func TestWithSource(t *testing.T) { }) } } + +func TestGetLastSeenQuery(t *testing.T) { + t.Run("nil args returns empty", func(t *testing.T) { + stmt, args := getLastSeenQuery("subj", nil) + assert.Empty(t, stmt) + assert.Nil(t, args) + }) + + t.Run("aggregates at projection grain", func(t *testing.T) { + stmt, args := getLastSeenQuery("subj", &model.SignalArgs{}) + want := "SELECT 'lastSeen' AS name, max(ts) AS ts, NULL AS value_number, NULL AS value_string, " + + "CAST(tuple(0, 0, 0, 0), 'Tuple(latitude Float64, longitude Float64, hdop Float64, heading Float64)') AS value_location " + + "FROM (SELECT max(timestamp) AS ts FROM `signal` WHERE (subject = ?) GROUP BY name);" + assert.Equal(t, want, stmt) + assert.Equal(t, []any{"subj"}, args) + }) + + t.Run("source filter stays in inner query", func(t *testing.T) { + stmt, args := getLastSeenQuery("subj", &model.SignalArgs{ + Filter: &model.SignalFilter{Source: ref("0xcd445F4c6bDAD32b68a2939b912150Fe3C88803E")}, + }) + want := "SELECT 'lastSeen' AS name, max(ts) AS ts, NULL AS value_number, NULL AS value_string, " + + "CAST(tuple(0, 0, 0, 0), 'Tuple(latitude Float64, longitude Float64, hdop Float64, heading Float64)') AS value_location " + + "FROM (SELECT max(timestamp) AS ts FROM `signal` WHERE (subject = ?) AND (source = ?) GROUP BY name);" + assert.Equal(t, want, stmt) + assert.Equal(t, []any{"subj", "0xcd445F4c6bDAD32b68a2939b912150Fe3C88803E"}, args) + }) +}