diff --git a/charts/telemetry-api/values-prod.yaml b/charts/telemetry-api/values-prod.yaml index 30eb312..24e8242 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 0547caf..9d018a1 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/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 743ec3b..4599d82 100644 --- a/go.mod +++ b/go.mod @@ -6,11 +6,11 @@ require ( github.com/99designs/gqlgen v0.17.89 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/cloudevent v1.0.4 + github.com/DIMO-Network/clickhouse-infra v0.0.8 + github.com/DIMO-Network/cloudevent v0.2.7 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.1.1 github.com/DIMO-Network/shared v1.1.5 github.com/DIMO-Network/token-exchange-api v0.4.0 diff --git a/go.sum b/go.sum index c1109d0..d909812 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/cloudevent v1.0.4 h1:bZ7tRPwCbiRgAlG1gfN7qpOOUI1OEXMMoMMyS9r4omk= -github.com/DIMO-Network/cloudevent v1.0.4/go.mod h1:I/9NcpMozV5Fw194WimhbkAsJtKVZf5UKYJ9hgc8Cdg= +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.7 h1:/cgFhUcWcliZYrmITkB8oIZb+zDhZvYNxWVGS2D3894= +github.com/DIMO-Network/cloudevent v0.2.7/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.1.1 h1:EYmyy+Fgi2BNW0Bufn04BViDtb8BCWaN7C7BbEuoI5s= github.com/DIMO-Network/server-garage v0.1.1/go.mod h1:Z3A1KDUsXey+XhrPhmw/wyCidfrQvmEdWp7nShno7ZM= github.com/DIMO-Network/shared v1.1.5 h1:cRI3BbeYOgolMkeBOSqbDVGxymnDfeP/q7xgIXJ5MkY= diff --git a/internal/config/settings.go b/internal/config/settings.go index d4c8f87..9584bff 100644 --- a/internal/config/settings.go +++ b/internal/config/settings.go @@ -16,7 +16,6 @@ type Settings struct { 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"` diff --git a/internal/service/ch/ch.go b/internal/service/ch/ch.go index 67cef10..34d4667 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,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) @@ -322,7 +313,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/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 6008344..5f7d692 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 = "(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" ) // 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...) } @@ -376,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), @@ -387,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...) } @@ -408,7 +436,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 } @@ -421,9 +449,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...) } @@ -751,19 +776,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), @@ -776,8 +807,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. @@ -808,7 +838,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