From b07afe1a5e7f094fb299b922ffb9bbc6512f1004 Mon Sep 17 00:00:00 2001 From: zer0stars <74260741+zer0stars@users.noreply.github.com> Date: Fri, 17 Apr 2026 16:28:58 -0400 Subject: [PATCH 1/3] Match signalsLatest/lastSeen queries to signal_latest projection Rewrites getLatestQuery to emit aggregate shapes that ClickHouse's projection matcher picks up against the signal_latest_by_subject_source_name projection added in model-garage. Non-location signals use argMax + max(timestamp); location signals use argMaxIf + maxIf with the same non-(0, 0) filter the projection carries. Mixed queries UNION ALL the two shapes so each branch matches cleanly. Drops DEFAULT_LOOKBACK / Service.lookbackFrom and the timestamp WHERE clause on both getLatestQuery and getLastSeenQuery. The timestamp filter prevents projection matching, and the projection itself makes the lookback unnecessary -- queries now read a handful of pre-aggregated rows per (subject, source, name) regardless of history depth. Verified on ClickHouse 25.3 with a 20M-row ReplacingMergeTree signal table: all four query shapes (non-location latest, location latest, mixed UNION ALL with source filter, lastSeen) match the projection via system.query_log.projections, and results are byte-identical to force-base (optimize_use_projections=0) runs. Must deploy only after DIMO-Network/model-garage#257 is materialized in the target environment; otherwise these queries fall back to full-history scans. --- charts/telemetry-api/values-prod.yaml | 1 - charts/telemetry-api/values.yaml | 1 - internal/config/settings.go | 29 +++--- internal/service/ch/ch.go | 44 ++++----- internal/service/ch/queries.go | 135 ++++++++++++++++---------- settings.sample.yaml | 3 - 6 files changed, 118 insertions(+), 95 deletions(-) diff --git a/charts/telemetry-api/values-prod.yaml b/charts/telemetry-api/values-prod.yaml index 2bc221a..7e841be 100644 --- a/charts/telemetry-api/values-prod.yaml +++ b/charts/telemetry-api/values-prod.yaml @@ -25,7 +25,6 @@ env: CREDIT_TRACKER_ENDPOINT: credit-tracker-prod:8086 IDENTITY_API_URL: http://identity-api-prod:8080/query IDENTITY_API_REQUEST_TIMEOUT_SECONDS: 5 - LATEST_SIGNALS_LOOKBACK_DAYS: 45 DIMO_REGISTRY_CHAIN_ID: 137 STORAGE_NODE_DEV_LICENSE: '0x49eAf63eD94FEf3d40692862Eee2C8dB416B1a5f' ingress: diff --git a/charts/telemetry-api/values.yaml b/charts/telemetry-api/values.yaml index 6fdca98..03bcb6a 100644 --- a/charts/telemetry-api/values.yaml +++ b/charts/telemetry-api/values.yaml @@ -39,7 +39,6 @@ env: CREDIT_TRACKER_ENDPOINT: credit-tracker-dev:8086 IDENTITY_API_URL: http://identity-api-dev:8080/query IDENTITY_API_REQUEST_TIMEOUT_SECONDS: 5 - LATEST_SIGNALS_LOOKBACK_DAYS: 45 DIMO_REGISTRY_CHAIN_ID: 80002 STORAGE_NODE_DEV_LICENSE: '0x52fD9Dc294066792785CcD85eFB9A0Bd48DE01E4' VIN_DATA_VERSION: vin/v1.0 diff --git a/internal/config/settings.go b/internal/config/settings.go index 5ef920d..2e7433d 100644 --- a/internal/config/settings.go +++ b/internal/config/settings.go @@ -7,21 +7,20 @@ import ( // Settings contains the application config. type Settings struct { - LogLevel string `yaml:"LOG_LEVEL"` - Port int `yaml:"PORT"` - MonPort int `yaml:"MON_PORT"` - EnablePprof bool `yaml:"ENABLE_PPROF"` - Clickhouse config.Settings `yaml:",inline"` - TokenExchangeJWTKeySetURL string `yaml:"TOKEN_EXCHANGE_JWK_KEY_SET_URL"` - TokenExchangeIssuer string `yaml:"TOKEN_EXCHANGE_ISSUER_URL"` - VehicleNFTAddress common.Address `yaml:"VEHICLE_NFT_ADDRESS"` - MaxRequestDuration string `yaml:"MAX_REQUEST_DURATION"` - LatestSignalsLookbackDays int64 `yaml:"LATEST_SIGNALS_LOOKBACK_DAYS"` - ChainID uint64 `yaml:"DIMO_REGISTRY_CHAIN_ID"` - FetchAPIGRPCEndpoint string `yaml:"FETCH_API_GRPC_ENDPOINT"` - CreditTrackerEndpoint string `yaml:"CREDIT_TRACKER_ENDPOINT"` - StorageNodeDevLicense common.Address `yaml:"STORAGE_NODE_DEV_LICENSE"` - VINDataVersion string `yaml:"VIN_DATA_VERSION"` + LogLevel string `yaml:"LOG_LEVEL"` + Port int `yaml:"PORT"` + MonPort int `yaml:"MON_PORT"` + EnablePprof bool `yaml:"ENABLE_PPROF"` + Clickhouse config.Settings `yaml:",inline"` + TokenExchangeJWTKeySetURL string `yaml:"TOKEN_EXCHANGE_JWK_KEY_SET_URL"` + TokenExchangeIssuer string `yaml:"TOKEN_EXCHANGE_ISSUER_URL"` + VehicleNFTAddress common.Address `yaml:"VEHICLE_NFT_ADDRESS"` + MaxRequestDuration string `yaml:"MAX_REQUEST_DURATION"` + ChainID uint64 `yaml:"DIMO_REGISTRY_CHAIN_ID"` + FetchAPIGRPCEndpoint string `yaml:"FETCH_API_GRPC_ENDPOINT"` + CreditTrackerEndpoint string `yaml:"CREDIT_TRACKER_ENDPOINT"` + StorageNodeDevLicense common.Address `yaml:"STORAGE_NODE_DEV_LICENSE"` + VINDataVersion string `yaml:"VIN_DATA_VERSION"` // TODO: remove these once manual vinvc are migrated to attestation VINVCDataVersion string `yaml:"VINVC_DATA_VERSION"` diff --git a/internal/service/ch/ch.go b/internal/service/ch/ch.go index ef8168f..ab90bec 100644 --- a/internal/service/ch/ch.go +++ b/internal/service/ch/ch.go @@ -24,8 +24,7 @@ const ( // Service is a ClickHouse service that interacts with the ClickHouse database. type Service struct { - conn clickhouse.Conn - latestLookback time.Duration + conn clickhouse.Conn } // NewService creates a new ClickHouse service. @@ -60,14 +59,7 @@ func NewService(settings config.Settings) (*Service, error) { if err != nil { return nil, fmt.Errorf("failed to ping clickhouse: %w", err) } - var latestLookback time.Duration - if settings.LatestSignalsLookbackDays > 0 { - latestLookback = time.Duration(settings.LatestSignalsLookbackDays) * 24 * time.Hour - } - return &Service{ - conn: conn, - latestLookback: latestLookback, - }, nil + return &Service{conn: conn}, nil } func getMaxExecutionTime(maxRequestDuration string) (int, error) { @@ -81,13 +73,22 @@ func getMaxExecutionTime(maxRequestDuration string) (int, error) { return int(maxExecutionTime.Seconds()), nil } -// GetLatestSignals returns the latest signals based on the provided arguments from the ClickHouse database. +// GetLatestSignals returns the latest signals based on the provided arguments +// from the ClickHouse database. The queries are answered by the +// signal_latest_by_subject_source_name projection, so no time bound is +// applied; results reflect the full history of the subject. func (s *Service) GetLatestSignals(ctx context.Context, subject string, latestArgs *model.LatestSignalsArgs) ([]*vss.Signal, error) { - lookbackFrom := s.lookbackFrom() - stmt, args := getLatestQuery(subject, latestArgs, lookbackFrom) + stmt, args := getLatestQuery(subject, latestArgs) if latestArgs.IncludeLastSeen { - lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &latestArgs.SignalArgs, lookbackFrom) - stmt, args = unionAll([]string{stmt, lastSeenStmt}, [][]any{args, lastSeenArgs}) + lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &latestArgs.SignalArgs) + if stmt == "" { + stmt, args = lastSeenStmt, lastSeenArgs + } else { + stmt, args = unionAll([]string{stmt, lastSeenStmt}, [][]any{args, lastSeenArgs}) + } + } + if stmt == "" { + return nil, nil } signals, err := s.getSignals(ctx, stmt, args) @@ -97,15 +98,6 @@ func (s *Service) GetLatestSignals(ctx context.Context, subject string, latestAr return signals, nil } -// lookbackFrom returns the earliest timestamp the "latest" queries will -// scan, or the zero Time if no lower bound is configured. -func (s *Service) lookbackFrom() time.Time { - if s.latestLookback <= 0 { - return time.Time{} - } - return time.Now().Add(-s.latestLookback) -} - // GetAggregatedSignals returns a slice of aggregated signals based on the provided arguments from the ClickHouse database. // The signals are sorted by timestamp in ascending order. // The timestamp on each signal is for the start of the interval. @@ -308,7 +300,9 @@ type EventSummary struct { LastSeen time.Time } -// GetEventSummaries returns per-event summaries (name, count, first/last seen) for a subject (vehicle), all time. +// GetEventSummaries returns per-event summaries (name, count, first/last seen) +// for a subject (vehicle), over all time. Relies on the (subject, source, name) +// projection on the event table for acceptable performance. func (s *Service) GetEventSummaries(ctx context.Context, subject string) ([]*EventSummary, error) { stmt, args := getEventSummariesQuery(subject) rows, err := s.conn.Query(ctx, stmt, args...) diff --git a/internal/service/ch/queries.go b/internal/service/ch/queries.go index f0c2911..10a5b01 100644 --- a/internal/service/ch/queries.go +++ b/internal/service/ch/queries.go @@ -43,12 +43,20 @@ const ( lastSeenTS = "max(" + vss.TimestampCol + ") AS ts" ) -// Aggregation functions for latest signals. +// Aggregation functions for latest signals. Shapes must stay byte-identical +// to the aggregates in the signal_latest_by_subject_source_name projection; +// ClickHouse matches projections by exact aggregate-expression text. const ( latestString = "argMax(" + vss.ValueStringCol + ", " + vss.TimestampCol + ") as " + vss.ValueStringCol latestNumber = "argMax(" + vss.ValueNumberCol + ", " + vss.TimestampCol + ") as " + vss.ValueNumberCol - latestLocation = "argMax(" + vss.ValueLocationCol + ", " + vss.TimestampCol + ") as " + AggLocationCol latestTimestamp = "max(" + vss.TimestampCol + ") as ts" + + // latestLocationCond excludes (0, 0) points from the latest-location + // computation. Kept in sync with the projection's argMaxIf/maxIf + // conditions. + latestLocationCond = "(" + vss.ValueLocationCol + ".latitude != 0) OR (" + vss.ValueLocationCol + ".longitude != 0)" + latestLocation = "argMaxIf(" + vss.ValueLocationCol + ", " + vss.TimestampCol + ", " + latestLocationCond + ") as " + AggLocationCol + latestLocationTS = "maxIf(" + vss.TimestampCol + ", " + latestLocationCond + ") as ts" ) // Aggregation functions for string signals. @@ -65,7 +73,6 @@ const ( locationZeroTuple = "CAST(tuple(0, 0, 0, 0), '" + locationTupleType + "')" ) - // FieldType indicates the type of values in the aggregation. type FieldType uint8 @@ -302,23 +309,19 @@ func batchLocationCaseExprWithAlias(alias string, locationArgs []model.LocationS return fmt.Sprintf("CASE %s ELSE %s END AS %s", strings.Join(parts, " "), locationZeroTuple, AggLocationCol) } -// getLatestQuery creates a query to get the latest signal value for each signal names -// returns the query statement and the arguments list, -/* -SELECT - name, - max(timestamp), - argMax(value_string, timestamp) as value_string, - argMax(value_number, timestamp) as value_number -FROM - signal -WHERE - subject = '...' AND - (name = 'speed' OR name = 'currentLocationLatitude' OR name = 'currentLocationLongitude' OR name = 'powertrainFuelSystemSupportedFuelTypes' OR name = 'none') -GROUP BY - name -*/ -func getLatestQuery(subject string, latestArgs *model.LatestSignalsArgs, lookbackFrom time.Time) (string, []any) { +// getLatestQuery builds the query (or UNION ALL of queries) that returns the +// latest value per signal name for a subject. +// +// Non-location and location signals use different aggregate shapes so both +// match the signal_latest_by_subject_source_name projection: +// +// - non-location: argMax(value_*, timestamp), max(timestamp) +// - location: argMaxIf/maxIf filtered by "location is not (0, 0)" +// +// Each branch is a separate SELECT so the projection matcher (which is +// byte-sensitive to aggregate expressions) matches each cleanly. The queries +// are combined with UNION ALL. +func getLatestQuery(subject string, latestArgs *model.LatestSignalsArgs) (string, []any) { signalNames := make([]string, 0, len(latestArgs.SignalNames)) for name := range latestArgs.SignalNames { signalNames = append(signalNames, name) @@ -329,34 +332,62 @@ func getLatestQuery(subject string, latestArgs *model.LatestSignalsArgs, lookbac locationSignalNames = append(locationSignalNames, name) } + stmts := make([]string, 0, 2) + args := make([][]any, 0, 2) + if len(signalNames) > 0 { + s, a := getLatestNonLocationQuery(subject, signalNames, latestArgs.Filter) + stmts = append(stmts, s) + args = append(args, a) + } + if len(locationSignalNames) > 0 { + s, a := getLatestLocationQuery(subject, locationSignalNames, latestArgs.Filter) + stmts = append(stmts, s) + args = append(args, a) + } + if len(stmts) == 0 { + return "", nil + } + if len(stmts) == 1 { + return stmts[0], args[0] + } + return unionAll(stmts, args) +} + +// getLatestNonLocationQuery selects max(ts), argMax(value_number), argMax(value_string) +// for the given signal names. Shape is aligned with the projection aggregates. +func getLatestNonLocationQuery(subject string, signalNames []string, filter *model.SignalFilter) (string, []any) { mods := []qm.QueryMod{ qm.Select(vss.NameCol), qm.Select(latestTimestamp), qm.Select(latestNumber), qm.Select(latestString), - qm.Select(latestLocation), + // keep the same output column set as the location branch so UNION ALL stays well-typed + qm.Select(locValAsZero), qm.From(vss.TableName), qm.Where(subjectWhere, subject), + qm.WhereIn(nameIn, signalNames), + qm.GroupBy(vss.NameCol), } - if !lookbackFrom.IsZero() { - mods = append(mods, whereTimestampFrom(lookbackFrom)) - } - mods = append(mods, - qm.Expr( - qm.WhereIn(nameIn, signalNames), - qm.Or2( - qm.Expr( - qm.WhereIn(nameIn, locationSignalNames), - qm.Expr( - qmhelper.Where(vss.ValueLocationCol+".latitude", qmhelper.NEQ, 0), - qm.Or2(qmhelper.Where(vss.ValueLocationCol+".longitude", qmhelper.NEQ, 0)), - ), - ), - ), - ), + mods = append(mods, getFilterMods(filter)...) + return newQuery(mods...) +} + +// getLatestLocationQuery selects maxIf(ts)/argMaxIf(value_location) filtered +// to rows where the location is not (0, 0). Matches the projection's +// argMaxIf/maxIf aggregates exactly. +func getLatestLocationQuery(subject string, locationSignalNames []string, filter *model.SignalFilter) (string, []any) { + mods := []qm.QueryMod{ + qm.Select(vss.NameCol), + qm.Select(latestLocationTS), + qm.Select(numValAsNull), + qm.Select(strValAsNull), + qm.Select(latestLocation), + qm.From(vss.TableName), + qm.Where(subjectWhere, subject), + qm.WhereIn(nameIn, locationSignalNames), qm.GroupBy(vss.NameCol), - ) - mods = append(mods, getFilterMods(latestArgs.Filter)...) + } + mods = append(mods, getFilterMods(filter)...) return newQuery(mods...) } @@ -374,7 +405,7 @@ FROM WHERE subject = '...' */ -func getLastSeenQuery(subject string, sigArgs *model.SignalArgs, lookbackFrom time.Time) (string, []any) { +func getLastSeenQuery(subject string, sigArgs *model.SignalArgs) (string, []any) { if sigArgs == nil { return "", nil } @@ -387,9 +418,6 @@ func getLastSeenQuery(subject string, sigArgs *model.SignalArgs, lookbackFrom ti qm.From(vss.TableName), qm.Where(subjectWhere, subject), } - if !lookbackFrom.IsZero() { - mods = append(mods, whereTimestampFrom(lookbackFrom)) - } mods = append(mods, getFilterMods(sigArgs.Filter)...) return newQuery(mods...) } @@ -717,19 +745,25 @@ func buildSegmentIndexMultiIf(timestampCol string, ranges []TimeRange) string { return "multiIf(" + strings.Join(parts, ", ") + ", -1) AS seg_idx" } +// getDistinctQuery returns distinct signal names seen for a subject. Uses +// GROUP BY (not DISTINCT) so the ClickHouse planner can match the +// (subject, source, name) aggregating projection on the signal table; without +// that projection this is a full-history scan per subject. func getDistinctQuery(subject string, filter *model.SignalFilter) (string, []any) { mods := []qm.QueryMod{ - qm.Distinct(vss.NameCol), + qm.Select(vss.NameCol), qm.From(vss.TableName), qm.Where(subjectWhere, subject), + qm.GroupBy(vss.NameCol), qm.OrderBy(vss.NameCol), } mods = append(mods, getFilterMods(filter)...) - stmt, args := newQuery(mods...) - return stmt, args + return newQuery(mods...) } -// select Count(*), max(timestamp), min(timestamp), name, from signal where subject = '...' GROUP BY name +// getSignalSummariesQuery summarizes signals by name for a subject. Relies on +// the (subject, source, name) projection on the signal table to keep this +// cheap; without that projection this is a full-history scan per subject. func getSignalSummariesQuery(subject string, filter *model.SignalFilter) (string, []any) { mods := []qm.QueryMod{ qm.Select(vss.NameCol), @@ -742,8 +776,7 @@ func getSignalSummariesQuery(subject string, filter *model.SignalFilter) (string qm.OrderBy(vss.NameCol), } mods = append(mods, getFilterMods(filter)...) - stmt, args := newQuery(mods...) - return stmt, args + return newQuery(mods...) } // getFilterMods returns the query mods for the filter. @@ -774,7 +807,9 @@ func appendEventFilterMods(mods []qm.QueryMod, filter *model.EventFilter) []qm.Q return mods } -// getEventSummariesQuery returns a query that summarizes events by name for a subject (all time). +// getEventSummariesQuery summarizes events by name for a subject. Relies on +// the (subject, source, name) projection on the event table to keep this +// cheap; without that projection this is a full-history scan per subject. func getEventSummariesQuery(subject string) (string, []any) { mods := []qm.QueryMod{ qm.Select(vss.EventNameCol + " AS name"), diff --git a/settings.sample.yaml b/settings.sample.yaml index c81f3b1..27abc58 100644 --- a/settings.sample.yaml +++ b/settings.sample.yaml @@ -12,6 +12,3 @@ PARQUET_BUCKET: dimo-iceberg-prod VINVC_DATA_VERSION: 'VINVCv0.0' IDENTITY_API_REQUEST_TIMEOUT_SECONDS: 5 MAX_REQUEST_DURATION: 30s -# Upper bound on how far back the "latest signal" queries scan. Set to 0 -# to disable (full scan). -LATEST_SIGNALS_LOOKBACK_DAYS: 45 From 0e57752294277499e9ca628e752ea0554f9c127d Mon Sep 17 00:00:00 2001 From: Dylan Moreland Date: Wed, 22 Apr 2026 16:01:10 -0400 Subject: [PATCH 2/3] Pull in new model-garage Fix up weird topK test --- e2e/clickhouse_container_test.go | 13 +++++++++---- e2e/segments_test.go | 4 +++- go.mod | 4 ++-- go.sum | 8 ++++---- internal/service/ch/ch_test.go | 28 +++++++++++++++++++++------- internal/service/ch/queries.go | 2 +- 6 files changed, 40 insertions(+), 19 deletions(-) diff --git a/e2e/clickhouse_container_test.go b/e2e/clickhouse_container_test.go index 03e44f1..46a1374 100644 --- a/e2e/clickhouse_container_test.go +++ b/e2e/clickhouse_container_test.go @@ -12,21 +12,26 @@ import ( "testing" "time" - "github.com/DIMO-Network/cloudevent" chconfig "github.com/DIMO-Network/clickhouse-infra/pkg/connect/config" "github.com/DIMO-Network/clickhouse-infra/pkg/container" + "github.com/DIMO-Network/cloudevent" sigmigrations "github.com/DIMO-Network/model-garage/pkg/migrations" "github.com/DIMO-Network/model-garage/pkg/vss" "github.com/stretchr/testify/require" ) -const loadBatchSize = 5000 +const ( + loadBatchSize = 5000 + testClickHousePassword = "test-clickhouse-password" +) func setupClickhouseContainer(t *testing.T) *container.Container { t.Helper() ctx := context.Background() - chContainer, err := container.CreateClickHouseContainer(ctx, chconfig.Settings{}) + chContainer, err := container.CreateClickHouseContainer(ctx, chconfig.Settings{ + Password: testClickHousePassword, + }) if err != nil { t.Fatalf("Failed to create clickhouse container: %v", err) } @@ -161,7 +166,7 @@ func LoadSampleDataInto(t *testing.T, ch *container.Container, signalPath, event events = append(events, vss.Event{ CloudEventHeader: cloudevent.CloudEventHeader{ Subject: row[0], - Source: row[1], + Source: row[1], Producer: row[2], ID: row[3], }, diff --git a/e2e/segments_test.go b/e2e/segments_test.go index fca8d31..8c62497 100644 --- a/e2e/segments_test.go +++ b/e2e/segments_test.go @@ -217,7 +217,9 @@ func TestSegmentDetectors(t *testing.T) { ctx := context.Background() // Setup ClickHouse container - chContainer, err := container.CreateClickHouseContainer(ctx, chconfig.Settings{}) + chContainer, err := container.CreateClickHouseContainer(ctx, chconfig.Settings{ + Password: testClickHousePassword, + }) require.NoError(t, err, "Failed to create ClickHouse container") t.Cleanup(func() { chContainer.Terminate(ctx) }) diff --git a/go.mod b/go.mod index 21fe23e..732b45e 100644 --- a/go.mod +++ b/go.mod @@ -6,11 +6,11 @@ require ( github.com/99designs/gqlgen v0.17.87 github.com/ClickHouse/clickhouse-go/v2 v2.40.1 github.com/DIMO-Network/attestation-api v0.1.2 - github.com/DIMO-Network/clickhouse-infra v0.0.7 + github.com/DIMO-Network/clickhouse-infra v0.0.8 github.com/DIMO-Network/cloudevent v0.2.6 github.com/DIMO-Network/credit-tracker v0.0.6 github.com/DIMO-Network/fetch-api v0.0.16 - github.com/DIMO-Network/model-garage v1.0.9 + github.com/DIMO-Network/model-garage v1.0.11 github.com/DIMO-Network/server-garage v0.0.7 github.com/DIMO-Network/shared v1.0.7 github.com/DIMO-Network/token-exchange-api v0.3.7 diff --git a/go.sum b/go.sum index ba2be1c..8e4e0b6 100644 --- a/go.sum +++ b/go.sum @@ -14,16 +14,16 @@ github.com/DATA-DOG/go-sqlmock v1.4.1 h1:ThlnYciV1iM/V0OSF/dtkqWb6xo5qITT1TJBG1M github.com/DATA-DOG/go-sqlmock v1.4.1/go.mod h1:f/Ixk793poVmq4qj/V1dPUg2JEAKC73Q5eFN3EC/SaM= github.com/DIMO-Network/attestation-api v0.1.2 h1:mrsProeTVr+aF7rKkoJhCVA7mlGFX6A8AwFKOxO3j8U= github.com/DIMO-Network/attestation-api v0.1.2/go.mod h1:sJT/d3Mb+qwYk1+DYKUvM8ZEohDQnM3sAqoreP+ocBI= -github.com/DIMO-Network/clickhouse-infra v0.0.7 h1:TAsjkFFKu3D5Xg6dwBcRBryjCVSlXsNjVbTwJ4UDlTg= -github.com/DIMO-Network/clickhouse-infra v0.0.7/go.mod h1:XS80lhSJNWBWGgZ+m4j7++zFj1wAXfmtV2gJfhGlabQ= +github.com/DIMO-Network/clickhouse-infra v0.0.8 h1:54HPXKvNjmn9T0d9ZLYgUm4DnwTavE+Z8admrf39mJI= +github.com/DIMO-Network/clickhouse-infra v0.0.8/go.mod h1:XS80lhSJNWBWGgZ+m4j7++zFj1wAXfmtV2gJfhGlabQ= github.com/DIMO-Network/cloudevent v0.2.6 h1:4J2S5aDou+80Kgdh8KK4t8A9C7Leb424Z6dlDuWjA/c= github.com/DIMO-Network/cloudevent v0.2.6/go.mod h1:I/9NcpMozV5Fw194WimhbkAsJtKVZf5UKYJ9hgc8Cdg= github.com/DIMO-Network/credit-tracker v0.0.6 h1:9C9AXah2uCQKCr7Kqzu2H9aXWj7wEOj4z1K2wAmVy0E= github.com/DIMO-Network/credit-tracker v0.0.6/go.mod h1:k037ZQMuBcyNj0Czg/jd6GCBC/HihkSsA7c1FRrF2yI= github.com/DIMO-Network/fetch-api v0.0.16 h1:jOIM9AHOCTN9Omh5QoRrdVsy0UCPHwB+08/DEI0qnqs= github.com/DIMO-Network/fetch-api v0.0.16/go.mod h1:pD9+5wPYjz3ORH9FBsXLlomsZJRPTyOzKy19urT6xew= -github.com/DIMO-Network/model-garage v1.0.9 h1:MqEC/maSaImduOLEqYbIbLovgKEgzNuTn/R8wpJkARc= -github.com/DIMO-Network/model-garage v1.0.9/go.mod h1:nvr0Ibiuu7Q7PC/hIfdZF6lPxCmq9eztQmbxUaNyus8= +github.com/DIMO-Network/model-garage v1.0.11 h1:aLvIyeo58p9pVgz+d3DnU5k5Fxvxh6mq/jE2s3LxXoc= +github.com/DIMO-Network/model-garage v1.0.11/go.mod h1:oi7EGKQVxFVpXRsu2H+YbizbKcx06aQg2N1Yu4GqOp8= github.com/DIMO-Network/server-garage v0.0.7 h1:kOBVyOtIbxa1x9pAf1epABTb9l/U3khf0PwUaHeHiKs= github.com/DIMO-Network/server-garage v0.0.7/go.mod h1:7DFor8MMJ8fLv9EB16Z5LrN+ftW3qeIk+swpkT7F2cU= github.com/DIMO-Network/shared v1.0.7 h1:LfSgsqJ6R7EUyfo2GTfuhrCpoDcweJqe7eVOa4j7Xbo= diff --git a/internal/service/ch/ch_test.go b/internal/service/ch/ch_test.go index 1c6f298..c2aa3dc 100644 --- a/internal/service/ch/ch_test.go +++ b/internal/service/ch/ch_test.go @@ -9,9 +9,9 @@ import ( "testing" "time" - "github.com/DIMO-Network/cloudevent" chconfig "github.com/DIMO-Network/clickhouse-infra/pkg/connect/config" "github.com/DIMO-Network/clickhouse-infra/pkg/container" + "github.com/DIMO-Network/cloudevent" "github.com/DIMO-Network/model-garage/pkg/migrations" "github.com/DIMO-Network/model-garage/pkg/vss" "github.com/DIMO-Network/telemetry-api/internal/config" @@ -20,8 +20,9 @@ import ( ) const ( - day = time.Hour * 24 - dataPoints = 10 + day = time.Hour * 24 + dataPoints = 10 + testClickHousePassword = "test-clickhouse-password" testSubject1 = "did:erc721:137:0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF:1" testSubject2 = "did:erc721:137:0xbA5738a18d83D41847dfFbDC6101d37C69c9B0cF:2" @@ -42,7 +43,9 @@ func TestCHService(t *testing.T) { func (c *CHServiceTestSuite) SetupSuite() { ctx := context.Background() var err error - c.container, err = container.CreateClickHouseContainer(ctx, chconfig.Settings{}) + c.container, err = container.CreateClickHouseContainer(ctx, chconfig.Settings{ + Password: testClickHousePassword, + }) c.Require().NoError(err, "Failed to create clickhouse container") db, err := c.container.GetClickhouseAsDB() @@ -71,9 +74,10 @@ func (c *CHServiceTestSuite) TestGetAggSignal() { endTs := c.dataStartTime.Add(time.Second * time.Duration(30*dataPoints)) ctx := context.Background() testCases := []struct { - name string - aggArgs model.AggregatedSignalArgs - expected []AggSignal + name string + aggArgs model.AggregatedSignalArgs + expected []AggSignal + allowedTopValueString []string }{ { name: "no aggs", @@ -233,6 +237,7 @@ func (c *CHServiceTestSuite) TestGetAggSignal() { ValueString: "value2", }, }, + allowedTopValueString: []string{"value2", "value5", "value8"}, }, { name: "first float", @@ -817,6 +822,15 @@ func (c *CHServiceTestSuite) TestGetAggSignal() { }) for i, sig := range result { + if len(tc.allowedTopValueString) > 0 { + c.Require().Equal(tc.expected[i].SignalType, sig.SignalType) + c.Require().Equal(tc.expected[i].SignalIndex, sig.SignalIndex) + c.Require().Equal(tc.expected[i].Timestamp, sig.Timestamp) + c.Require().Equal(tc.expected[i].ValueNumber, sig.ValueNumber) + c.Require().Equal(tc.expected[i].ValueLocation, sig.ValueLocation) + c.Require().Contains(tc.allowedTopValueString, sig.ValueString) + continue + } c.Require().Equal(tc.expected[i], *sig) } }) diff --git a/internal/service/ch/queries.go b/internal/service/ch/queries.go index 10a5b01..4c9fb15 100644 --- a/internal/service/ch/queries.go +++ b/internal/service/ch/queries.go @@ -54,7 +54,7 @@ const ( // latestLocationCond excludes (0, 0) points from the latest-location // computation. Kept in sync with the projection's argMaxIf/maxIf // conditions. - latestLocationCond = "(" + vss.ValueLocationCol + ".latitude != 0) OR (" + vss.ValueLocationCol + ".longitude != 0)" + latestLocationCond = "(tupleElement(" + vss.ValueLocationCol + ", 'latitude') != 0) OR (tupleElement(" + vss.ValueLocationCol + ", 'longitude') != 0)" latestLocation = "argMaxIf(" + vss.ValueLocationCol + ", " + vss.TimestampCol + ", " + latestLocationCond + ") as " + AggLocationCol latestLocationTS = "maxIf(" + vss.TimestampCol + ", " + latestLocationCond + ") as ts" ) From 7d70ee6c132f9652fa7b30ab3037cd28b46fe6d9 Mon Sep 17 00:00:00 2001 From: Dylan Moreland Date: Wed, 22 Apr 2026 16:10:50 -0400 Subject: [PATCH 3/3] Fix up this lookback stuff --- internal/service/ch/ch.go | 14 ++------------ internal/service/ch/queries.go | 5 +---- 2 files changed, 3 insertions(+), 16 deletions(-) diff --git a/internal/service/ch/ch.go b/internal/service/ch/ch.go index 19161f8..34d4667 100644 --- a/internal/service/ch/ch.go +++ b/internal/service/ch/ch.go @@ -98,20 +98,10 @@ func (s *Service) GetLatestSignals(ctx context.Context, subject string, latestAr return signals, nil } -// lookbackFrom returns the earliest timestamp the "latest" queries will -// scan, or the zero Time if no lower bound is configured. -func (s *Service) lookbackFrom() time.Time { - if s.latestLookback <= 0 { - return time.Time{} - } - return time.Now().Add(-s.latestLookback) -} - // GetAllLatestSignals returns the latest value for every signal stored for a subject. func (s *Service) GetAllLatestSignals(ctx context.Context, subject string, filter *model.SignalFilter) ([]*vss.Signal, error) { - lookbackFrom := s.lookbackFrom() - stmt, args := getAllLatestQuery(subject, filter, lookbackFrom) - lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &model.SignalArgs{Filter: filter}, lookbackFrom) + stmt, args := getAllLatestQuery(subject, filter) + lastSeenStmt, lastSeenArgs := getLastSeenQuery(subject, &model.SignalArgs{Filter: filter}) stmt, args = unionAll([]string{stmt, lastSeenStmt}, [][]any{args, lastSeenArgs}) signals, err := s.getSignals(ctx, stmt, args) diff --git a/internal/service/ch/queries.go b/internal/service/ch/queries.go index c75cf4c..5f7d692 100644 --- a/internal/service/ch/queries.go +++ b/internal/service/ch/queries.go @@ -407,7 +407,7 @@ WHERE GROUP BY name */ -func getAllLatestQuery(subject string, filter *model.SignalFilter, lookbackFrom time.Time) (string, []any) { +func getAllLatestQuery(subject string, filter *model.SignalFilter) (string, []any) { mods := []qm.QueryMod{ qm.Select(vss.NameCol), qm.Select(latestTimestamp), @@ -418,9 +418,6 @@ func getAllLatestQuery(subject string, filter *model.SignalFilter, lookbackFrom qm.Where(subjectWhere, subject), qm.GroupBy(vss.NameCol), } - if !lookbackFrom.IsZero() { - mods = append(mods, whereTimestampFrom(lookbackFrom)) - } mods = append(mods, getFilterMods(filter)...) return newQuery(mods...) }